Refactor page, history, workflow display code to use same security code as get() methods.

This commit is contained in:
Jeremy Goecks
2010-03-04 13:49:09 -05:00
parent b5504272ac
commit fb88f43899
4 changed files with 10 additions and 25 deletions
+1 -2
View File
@@ -110,8 +110,7 @@ class SharableItemSecurity:
if check_accessible:
# Verify accessible.
if ( item.user != user ) and ( not item.importable ) and ( user not in item.users_shared_with_dot_users ):
raise "hi"
error( "%s is not accessible by current user" % item.__class__.__name__ )
error( "%s is not accessible to current user" % item.__class__.__name__ )
return item
class UsesHistoryDatasetAssociation:
+3 -7
View File
@@ -589,15 +589,11 @@ class HistoryController( BaseController, Sharable, UsesAnnotations, UsesHistory
# Get history.
session = trans.sa_session
user = session.query( model.User ).filter_by( username=username ).first()
history_query_base = trans.sa_session.query( model.History ).filter_by( user=user, slug=slug, deleted=False )
if user is not None:
# User can view history if it's importable or if it's shared with him/her.
history = history_query_base.filter( or_( model.History.importable==True, model.History.users_shared_with.any( model.HistoryUserShareAssociation.user==trans.get_user() ) ) ).first()
else:
# User not logged in, so only way to view history is if it's importable.
history = history_query_base.filter_by( importable=True ).first()
history = trans.sa_session.query( model.History ).filter_by( user=user, slug=slug, deleted=False ).first()
if history is None:
raise web.httpexceptions.HTTPNotFound()
# Security check raises error if user cannot access history.
self.security_check( trans.get_user(), history, False, True)
# Get datasets.
datasets = self.get_history_datasets( trans, history )
+3 -7
View File
@@ -555,15 +555,11 @@ class PageController( BaseController, Sharable, UsesAnnotations, UsesHistory, Us
# Get page.
session = trans.sa_session
user = session.query( model.User ).filter_by( username=username ).first()
page_query_base = trans.sa_session.query( model.Page ).filter_by( user=user, slug=slug, deleted=False )
if user is not None:
# User can view page if it's importable or if it's shared with him/her.
page = page_query_base.filter( or_( model.Page.user==trans.get_user(), model.Page.importable==True, model.Page.users_shared_with.any( model.PageUserShareAssociation.user==trans.get_user() ) ) ).first()
else:
# User not logged in, so only way to view page is if it's importable.
page = page_query_base.filter_by( importable=True ).first()
page = trans.sa_session.query( model.Page ).filter_by( user=user, slug=slug, deleted=False ).first()
if page is None:
raise web.httpexceptions.HTTPNotFound()
# Security check raises error if user cannot access page.
self.security_check( trans.get_user(), page, False, True)
# Process page content.
processor = _PageContentProcessor( trans, 'utf-8', 'text/html', self._get_embed_html )
+3 -9
View File
@@ -169,15 +169,9 @@ class WorkflowController( BaseController, Sharable, UsesStoredWorkflow, UsesAnno
# Get workflow.
session = trans.sa_session
user = session.query( model.User ).filter_by( username=username ).first()
workflow_query_base = trans.sa_session.query( model.StoredWorkflow ).filter_by( user=user, slug=slug, deleted=False )
if user is not None:
# User can view workflow if it's importable or if it's shared with him/her.
stored_workflow = workflow_query_base.filter( or_( model.StoredWorkflow.importable==True, model.StoredWorkflow.users_shared_with.any( model.StoredWorkflowUserShareAssociation.user==trans.get_user() ) ) ).first()
else:
# User not logged in, so only way to view workflow is if it's importable.
stored_workflow = workflow_query_base.filter_by( importable=True ).first()
if stored_workflow is None:
raise web.httpexceptions.HTTPNotFound()
stored_workflow = trans.sa_session.query( model.StoredWorkflow ).filter_by( user=user, slug=slug, deleted=False ).first()
# Security check raises error if user cannot access workflow.
self.security_check( trans.get_user(), stored_workflow, False, True)
# Get data for workflow's steps.
self.get_stored_workflow_steps( trans, stored_workflow )