subversion: Detect requests also based on magic path.

rhodecode/lib/middleware/vcs.py

@@ -1,156 +1,160 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import gzip
 import shutil
 import logging
 import tempfile
 import urlparse
 
 import rhodecode
 from rhodecode.lib.middleware.appenlight import wrap_in_appenlight_if_enabled
 from rhodecode.lib.middleware.simplegit import SimpleGit, GIT_PROTO_PAT
 from rhodecode.lib.middleware.simplehg import SimpleHg
 from rhodecode.lib.middleware.simplesvn import SimpleSvn
 
 
 log = logging.getLogger(__name__)
 
 
 def is_git(environ):
     """
     Returns True if requests should be handled by GIT wsgi middleware
     """
     is_git_path = GIT_PROTO_PAT.match(environ['PATH_INFO'])
     log.debug(
         'request path: `%s` detected as GIT PROTOCOL %s', environ['PATH_INFO'],
         is_git_path is not None)
 
     return is_git_path
 
 
 def is_hg(environ):
     """
     Returns True if requests target is mercurial server - header
     ``HTTP_ACCEPT`` of such request would start with ``application/mercurial``.
     """
     is_hg_path = False
 
     http_accept = environ.get('HTTP_ACCEPT')
 
     if http_accept and http_accept.startswith('application/mercurial'):
         query = urlparse.parse_qs(environ['QUERY_STRING'])
         if 'cmd' in query:
             is_hg_path = True
 
     log.debug(
         'request path: `%s` detected as HG PROTOCOL %s', environ['PATH_INFO'],
         is_hg_path)
 
     return is_hg_path
 
 
 def is_svn(environ):
     """
     Returns True if requests target is Subversion server
     """
     http_dav = environ.get('HTTP_DAV', '')
-    is_svn_path = 'subversion' in http_dav
+    magic_path_segment = rhodecode.CONFIG.get(
+        'rhodecode_subversion_magic_path', '/!svn')
+    is_svn_path = (
+        'subversion' in http_dav or
+        magic_path_segment in environ['PATH_INFO'])
     log.debug(
         'request path: `%s` detected as SVN PROTOCOL %s', environ['PATH_INFO'],
         is_svn_path)
 
     return is_svn_path
 
 
 class GunzipMiddleware(object):
     """
     WSGI middleware that unzips gzip-encoded requests before
     passing on to the underlying application.
     """
 
     def __init__(self, application):
         self.app = application
 
     def __call__(self, environ, start_response):
         accepts_encoding_header = environ.get('HTTP_CONTENT_ENCODING', b'')
 
         if b'gzip' in accepts_encoding_header:
             log.debug('gzip detected, now running gunzip wrapper')
             wsgi_input = environ['wsgi.input']
 
             if not hasattr(environ['wsgi.input'], 'seek'):
                 # The gzip implementation in the standard library of Python 2.x
                 # requires the '.seek()' and '.tell()' methods to be available
                 # on the input stream.  Read the data into a temporary file to
                 # work around this limitation.
 
                 wsgi_input = tempfile.SpooledTemporaryFile(64 * 1024 * 1024)
                 shutil.copyfileobj(environ['wsgi.input'], wsgi_input)
                 wsgi_input.seek(0)
 
             environ['wsgi.input'] = gzip.GzipFile(fileobj=wsgi_input, mode='r')
             # since we "Ungzipped" the content we say now it's no longer gzip
             # content encoding
             del environ['HTTP_CONTENT_ENCODING']
 
             # content length has changes ? or i'm not sure
             if 'CONTENT_LENGTH' in environ:
                 del environ['CONTENT_LENGTH']
         else:
             log.debug('content not gzipped, gzipMiddleware passing '
                       'request further')
         return self.app(environ, start_response)
 
 
 class VCSMiddleware(object):
 
     def __init__(self, app, config, appenlight_client):
         self.application = app
         self.config = config
         self.appenlight_client = appenlight_client
 
     def _get_handler_app(self, environ):
         app = None
         if is_hg(environ):
             app = SimpleHg(self.application, self.config)
 
         if is_git(environ):
             app = SimpleGit(self.application, self.config)
 
         proxy_svn = rhodecode.CONFIG.get(
             'rhodecode_proxy_subversion_http_requests', False)
         if proxy_svn and is_svn(environ):
             app = SimpleSvn(self.application, self.config)
 
         if app:
             app = GunzipMiddleware(app)
             app, _ = wrap_in_appenlight_if_enabled(
                 app, self.config, self.appenlight_client)
 
         return app
 
     def __call__(self, environ, start_response):
         # check if we handle one of interesting protocols ?
         vcs_handler = self._get_handler_app(environ)
         if vcs_handler:
             return vcs_handler(environ, start_response)
 
         return self.application(environ, start_response)

rhodecode/tests/lib/middleware/test_vcs.py

@@ -1,114 +1,134 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 from mock import patch, Mock
 
 import rhodecode
 from rhodecode.lib.middleware import vcs
 
 
 def test_is_hg():
     environ = {
         'PATH_INFO': 'rhodecode-dev',
         'QUERY_STRING': 'cmd=changegroup',
         'HTTP_ACCEPT': 'application/mercurial'
     }
     assert vcs.is_hg(environ)
 
 
 def test_is_hg_no_cmd():
     environ = {
         'PATH_INFO': 'rhodecode-dev',
         'QUERY_STRING': '',
         'HTTP_ACCEPT': 'application/mercurial'
     }
     assert not vcs.is_hg(environ)
 
 
 def test_is_hg_empty_cmd():
     environ = {
         'PATH_INFO': 'rhodecode-dev',
         'QUERY_STRING': 'cmd=',
         'HTTP_ACCEPT': 'application/mercurial'
     }
     assert not vcs.is_hg(environ)
 
 
 def test_is_svn_returns_true_if_subversion_is_in_a_dav_header():
     environ = {
         'PATH_INFO': 'rhodecode-dev',
         'HTTP_DAV': 'http://subversion.tigris.org/xmlns/dav/svn/log-revprops'
     }
     assert vcs.is_svn(environ) is True
 
 
 def test_is_svn_returns_false_if_subversion_is_not_in_a_dav_header():
     environ = {
         'PATH_INFO': 'rhodecode-dev',
         'HTTP_DAV': 'http://stuff.tigris.org/xmlns/dav/svn/log-revprops'
     }
     assert vcs.is_svn(environ) is False
 
 
 def test_is_svn_returns_false_if_no_dav_header():
     environ = {
         'PATH_INFO': 'rhodecode-dev',
     }
     assert vcs.is_svn(environ) is False
 
 
+def test_is_svn_returns_true_if_magic_path_segment():
+    environ = {
+        'PATH_INFO': '/stub-repository/!svn/rev/4',
+    }
+    assert vcs.is_svn(environ)
+
+
+def test_is_svn_allows_to_configure_the_magic_path(monkeypatch):
+    """
+    This is intended as a fallback in case someone has configured his
+    Subversion server with a different magic path segment.
+    """
+    monkeypatch.setitem(
+        rhodecode.CONFIG, 'rhodecode_subversion_magic_path', '/!my-magic')
+    environ = {
+        'PATH_INFO': '/stub-repository/!my-magic/rev/4',
+    }
+    assert vcs.is_svn(environ)
+
+
 class TestVCSMiddleware(object):
     def test_get_handler_app_retuns_svn_app_when_proxy_enabled(self):
         environ = {
             'PATH_INFO': 'rhodecode-dev',
             'HTTP_DAV': 'http://subversion.tigris.org/xmlns/dav/svn/log'
         }
         app = Mock()
         config = Mock()
         middleware = vcs.VCSMiddleware(
             app, config=config, appenlight_client=None)
         snv_patch = patch('rhodecode.lib.middleware.vcs.SimpleSvn')
         settings_patch = patch.dict(
             rhodecode.CONFIG,
             {'rhodecode_proxy_subversion_http_requests': True})
         with snv_patch as svn_mock, settings_patch:
             svn_mock.return_value = None
             middleware._get_handler_app(environ)
 
         svn_mock.assert_called_once_with(app, config)
 
     def test_get_handler_app_retuns_no_svn_app_when_proxy_disabled(self):
         environ = {
             'PATH_INFO': 'rhodecode-dev',
             'HTTP_DAV': 'http://subversion.tigris.org/xmlns/dav/svn/log'
         }
         app = Mock()
         config = Mock()
         middleware = vcs.VCSMiddleware(
             app, config=config, appenlight_client=None)
         snv_patch = patch('rhodecode.lib.middleware.vcs.SimpleSvn')
         settings_patch = patch.dict(
             rhodecode.CONFIG,
             {'rhodecode_proxy_subversion_http_requests': False})
         with snv_patch as svn_mock, settings_patch:
             app = middleware._get_handler_app(environ)
 
         assert svn_mock.call_count == 0
         assert app is None