[16.01] retroactively disable Mercurial push API in TS

config/tool_shed.ini.sample

@@ -35,6 +35,9 @@ database_file = database/community.sqlite
 # The default is the Galaxy installation directory.
 #hgweb_config_dir = None
 
+# Disable Mercurial pushing to repositories.
+#disable_push = True
+
 # Where tool shed repositories are stored.
 file_path = database/community_files
 # Temporary storage for additional datasets,

lib/galaxy/webapps/tool_shed/config.py

@@ -136,6 +136,7 @@ def __init__( self, **kwargs ):
         self.sentry_dsn = kwargs.get( 'sentry_dsn', None )
         # Where the tool shed hgweb.config file is stored - the default is the Galaxy installation directory.
         self.hgweb_config_dir = resolve_path( kwargs.get( 'hgweb_config_dir', '' ), self.root )
+        self.disable_push = string_as_bool( kwargs.get( "disable_push", "True" ) )
         # Proxy features
         self.apache_xsendfile = kwargs.get( 'apache_xsendfile', False )
         self.nginx_x_accel_redirect_base = kwargs.get( 'nginx_x_accel_redirect_base', False )

lib/galaxy/webapps/tool_shed/framework/middleware/hg.py

@@ -74,6 +74,9 @@ def __call__( self, environ, start_response ):
                     connection.execute( sql_cmd )
                     connection.close()
         elif cmd in [ 'unbundle', 'pushkey' ]:
+            if self.config.get('disable_push', True):
+                msg = 'Pushing to Tool Shed is disabled.'
+                return self.__display_exception_remotely( start_response, msg )
             # This is an hg push from the command line.  When doing this, the following commands, in order,
             # will be retrieved from environ (see the docs at http://mercurial.selenic.com/wiki/WireProtocol):
             # # If mercurial version >= '2.2.3': capabilities -> batch -> branchmap -> unbundle -> listkeys -> pushkey -> listkeys

test/tool_shed/base/twilltestcase.py

@@ -258,12 +258,17 @@ def commit_and_push( self, repository, hgrepo, options, username, password ):
         commands.commit( ui.ui(), hgrepo, **options )
         try:
             commands.push( ui.ui(), hgrepo, dest=url )
-        except Abort, e:
-            message = e
+        except Abort as a:
+            message = a
             if 'authorization failed' in message:
                 return False
             else:
                 raise
+        except Exception as e:
+            if str(e).find('Pushing to Tool Shed is disabled') != -1:
+                return False
+            else:
+                raise
         return True
 
     def create_category( self, **kwd ):

test/tool_shed/functional/test_0310_hg_push_from_api.py → test/tool_shed/functional/test_0310_hg_api_features.py

@@ -12,15 +12,10 @@
 category_name = 'Test 0310 - HTTP Repo features'
 category_description = 'Test 0310 for verifying the tool shed http interface to mercurial.'
 
-# Declare clone_path here so multiple tests can access it.
-clone_path = None
-
 '''
 1. Create a repository.
 2. Clone the repository to a local path.
-3. Change a file and try to push as non-owner.
-4. Change another file and push as owner.
-5. Verify that the changesets have been applied.
+3. Check Mercurial push is disabled.
 '''
 
 
@@ -79,10 +74,10 @@ def test_0005_create_filtering_repository( self ):
                           strings_displayed=[],
                           strings_not_displayed=[] )
 
-    def test_0010_edit_and_commit( self ):
-        '''Edit a file and attempt a push as a user that does not have write access.'''
+    def test_0010_clone( self ):
+        '''Clone the repository to a local path.'''
         '''
-        We are at step 3 - Change a file and try to push as non-owner.
+        We are at step 2 - Clone the repository to a local path.
         The repository should have the following files:
 
         filtering.py
@@ -96,43 +91,19 @@ def test_0010_edit_and_commit( self ):
         test-data/filter1_test2.bed
         test-data/filter1_test3.sam
         test-data/filter1_test4.bed
-
-        We will be prepending a comment to filtering.py.
         '''
         repository = self.test_db_util.get_repository_by_name_and_owner( repository_name, common.test_user_1_name )
         clone_path = self.generate_temp_path( 'test_0310', additional_paths=[ 'filtering_0310', 'user2' ] )
         self.clone_repository( repository, clone_path )
-        hgrepo = self.get_hg_repo( clone_path )
         files_in_repository = os.listdir( clone_path )
         assert 'filtering.py' in files_in_repository, 'File not found in repository: filtering.py'
-        filepath = os.path.join( clone_path, 'filtering.py' )
-        file_contents = [ '# This is a dummy comment to generate a new changeset.' ]
-        file_contents.extend( file( filepath, 'r' ).readlines() )
-        file( filepath, 'w' ).write( '\n'.join( file_contents ) )
-        commit_options = dict( user=common.test_user_2_name, message='Added a line to filtering.py' )
-        # The repository is owned by test_user_1, so this operation should fail.
-        authorized = self.commit_and_push( repository, hgrepo, commit_options, username=common.test_user_2_name, password='testuser' )
-        assert authorized is False, 'Test user 2 was able to commit and push to the remote repository.'
 
-    def test_0015_edit_and_commit( self ):
-        '''Edit a file again and attempt a push as a user that does have write access.'''
+    def test_0015_commit_and_push( self ):
         '''
-        We are at step 4 - Change another file and try to push as non-owner.
-        The repository should have the following files:
-
-        filtering.py
-        filtering.xml
-        test-data/
-        test-data/1.bed
-        test-data/7.bed
-        test-data/filter1_in3.sam
-        test-data/filter1_inbad.bed
-        test-data/filter1_test1.bed
-        test-data/filter1_test2.bed
-        test-data/filter1_test3.sam
-        test-data/filter1_test4.bed
+        Edit a file and attempt a push as a user that does have write access.
+        We expect this to fail as pushing is disabled by default via disable_push config option.
 
-        We will be prepending a second comment to filtering.py.
+        We are at step 3 - Check Mercurial push is disabled.
         '''
         repository = self.test_db_util.get_repository_by_name_and_owner( repository_name, common.test_user_1_name )
         clone_path = self.generate_temp_path( 'test_0310', additional_paths=[ 'filtering_0310', 'user1' ] )
@@ -145,25 +116,5 @@ def test_0015_edit_and_commit( self ):
         file_contents.extend( file( filepath, 'r' ).readlines() )
         file( filepath, 'w' ).write( '\n'.join( file_contents ) )
         commit_options = dict( user=common.test_user_1_name, message='Added another line to filtering.py.' )
-        # The repository is owned by test_user_1, so this operation should succeed.
-        authorized = self.commit_and_push( repository, hgrepo, commit_options, username=common.test_user_1_name, password='testuser' )
-        assert authorized is True, 'Test user 1 was not able to commit and push to the remote repository.'
-
-    def test_0020_verify_new_changelog( self ):
-        '''Verify that the authorized commit was applied, and the unauthorized commit was not..'''
-        '''
-        We are at step 5 - Verify that the changeset has been applied.
-
-        The repository changelog should now look like:
-
-        0:nnnnnnnnnnnn: Uploaded filtering 1.1.0.
-        1:nnnnnnnnnnnn: Uploaded filtering test data.
-        2:nnnnnnnnnnnn: Added another line to filtering.py.
-
-        The commit from test_user_2 should not be present in the changelog, since the repositories were cloned to separate locations.
-        '''
-        repository = self.test_db_util.get_repository_by_name_and_owner( repository_name, common.test_user_1_name )
-        strings_displayed = [ 'Uploaded filtering 1.1.0.', 'Uploaded filtering test data.',
-                              'Added another line to filtering.py.' ]
-        strings_not_displayed = [ 'Added a line to filtering.py' ]
-        self.check_repository_changelog( repository, strings_displayed=strings_displayed, strings_not_displayed=strings_not_displayed )
+        success = self.commit_and_push( repository, hgrepo, commit_options, username=common.test_user_1_name, password='testuser' )
+        assert success is False, 'Test user 1 (repo owner) was able to commit and push to the remote repository.'