svn: use shared configurable storage for svn_txn_id interception logic.

rhodecode/lib/hooks_daemon.py

@@ -1,324 +1,330 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2018 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 os
 import time
 import logging
 import tempfile
 import traceback
 import threading
 
 from BaseHTTPServer import BaseHTTPRequestHandler
 from SocketServer import TCPServer
 
 import rhodecode
 from rhodecode.lib.exceptions import HTTPLockedRC, HTTPBranchProtected
 from rhodecode.model import meta
 from rhodecode.lib.base import bootstrap_request, bootstrap_config
 from rhodecode.lib import hooks_base
 from rhodecode.lib.utils2 import AttributeDict
 from rhodecode.lib.ext_json import json
 from rhodecode.lib import rc_cache
 
 log = logging.getLogger(__name__)
 
 
 class HooksHttpHandler(BaseHTTPRequestHandler):
 
     def do_POST(self):
         method, extras = self._read_request()
         txn_id = getattr(self.server, 'txn_id', None)
         if txn_id:
             log.debug('Computing TXN_ID based on `%s`:`%s`',
                       extras['repository'], extras['txn_id'])
             computed_txn_id = rc_cache.utils.compute_key_from_params(
                 extras['repository'], extras['txn_id'])
             if txn_id != computed_txn_id:
                 raise Exception(
                     'TXN ID fail: expected {} got {} instead'.format(
                         txn_id, computed_txn_id))
 
         try:
             result = self._call_hook(method, extras)
         except Exception as e:
             exc_tb = traceback.format_exc()
             result = {
                 'exception': e.__class__.__name__,
                 'exception_traceback': exc_tb,
                 'exception_args': e.args
             }
         self._write_response(result)
 
     def _read_request(self):
         length = int(self.headers['Content-Length'])
         body = self.rfile.read(length).decode('utf-8')
         data = json.loads(body)
         return data['method'], data['extras']
 
     def _write_response(self, result):
         self.send_response(200)
         self.send_header("Content-type", "text/json")
         self.end_headers()
         self.wfile.write(json.dumps(result))
 
     def _call_hook(self, method, extras):
         hooks = Hooks()
         try:
             result = getattr(hooks, method)(extras)
         finally:
             meta.Session.remove()
         return result
 
     def log_message(self, format, *args):
         """
         This is an overridden method of BaseHTTPRequestHandler which logs using
         logging library instead of writing directly to stderr.
         """
 
         message = format % args
 
         log.debug(
             "%s - - [%s] %s", self.client_address[0],
             self.log_date_time_string(), message)
 
 
 class DummyHooksCallbackDaemon(object):
     hooks_uri = ''
 
     def __init__(self):
         self.hooks_module = Hooks.__module__
 
     def __enter__(self):
         log.debug('Running dummy hooks callback daemon')
         return self
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         log.debug('Exiting dummy hooks callback daemon')
 
 
 class ThreadedHookCallbackDaemon(object):
 
     _callback_thread = None
     _daemon = None
     _done = False
 
     def __init__(self, txn_id=None, host=None, port=None):
         self._prepare(txn_id=txn_id, host=None, port=port)
 
     def __enter__(self):
         self._run()
         return self
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         log.debug('Callback daemon exiting now...')
         self._stop()
 
     def _prepare(self, txn_id=None, host=None, port=None):
         raise NotImplementedError()
 
     def _run(self):
         raise NotImplementedError()
 
     def _stop(self):
         raise NotImplementedError()
 
 
 class HttpHooksCallbackDaemon(ThreadedHookCallbackDaemon):
     """
     Context manager which will run a callback daemon in a background thread.
     """
 
     hooks_uri = None
 
     # From Python docs: Polling reduces our responsiveness to a shutdown
     # request and wastes cpu at all other times.
     POLL_INTERVAL = 0.01
 
     def _prepare(self, txn_id=None, host=None, port=None):
         host = host or '127.0.0.1'
         self._done = False
         self._daemon = TCPServer((host, port or 0), HooksHttpHandler)
         _, port = self._daemon.server_address
         self.hooks_uri = '{}:{}'.format(host, port)
         self.txn_id = txn_id
         # inject transaction_id for later verification
         self._daemon.txn_id = self.txn_id
 
         log.debug(
             "Preparing HTTP callback daemon at `%s` and registering hook object",
             self.hooks_uri)
 
     def _run(self):
         log.debug("Running event loop of callback daemon in background thread")
         callback_thread = threading.Thread(
             target=self._daemon.serve_forever,
             kwargs={'poll_interval': self.POLL_INTERVAL})
         callback_thread.daemon = True
         callback_thread.start()
         self._callback_thread = callback_thread
 
     def _stop(self):
         log.debug("Waiting for background thread to finish.")
         self._daemon.shutdown()
         self._callback_thread.join()
         self._daemon = None
         self._callback_thread = None
         if self.txn_id:
             txn_id_file = get_txn_id_data_path(self.txn_id)
             log.debug('Cleaning up TXN ID %s', txn_id_file)
             if os.path.isfile(txn_id_file):
                 os.remove(txn_id_file)
 
         log.debug("Background thread done.")
 
 
 def get_txn_id_data_path(txn_id):
-    root = tempfile.gettempdir()
-    return os.path.join(root, 'rc_txn_id_{}'.format(txn_id))
+    import rhodecode
+
+    root = rhodecode.CONFIG.get('cache_dir') or tempfile.gettempdir()
+    final_dir = os.path.join(root, 'svn_txn_id')
+
+    if not os.path.isdir(final_dir):
+        os.makedirs(final_dir)
+    return os.path.join(final_dir, 'rc_txn_id_{}'.format(txn_id))
 
 
 def store_txn_id_data(txn_id, data_dict):
     if not txn_id:
         log.warning('Cannot store txn_id because it is empty')
         return
 
     path = get_txn_id_data_path(txn_id)
     try:
         with open(path, 'wb') as f:
             f.write(json.dumps(data_dict))
     except Exception:
         log.exception('Failed to write txn_id metadata')
 
 
 def get_txn_id_from_store(txn_id):
     """
     Reads txn_id from store and if present returns the data for callback manager
     """
     path = get_txn_id_data_path(txn_id)
     try:
         with open(path, 'rb') as f:
             return json.loads(f.read())
     except Exception:
         return {}
 
 
 def prepare_callback_daemon(extras, protocol, host, use_direct_calls, txn_id=None):
     txn_details = get_txn_id_from_store(txn_id)
     port = txn_details.get('port', 0)
     if use_direct_calls:
         callback_daemon = DummyHooksCallbackDaemon()
         extras['hooks_module'] = callback_daemon.hooks_module
     else:
         if protocol == 'http':
             callback_daemon = HttpHooksCallbackDaemon(
                 txn_id=txn_id, host=host, port=port)
         else:
             log.error('Unsupported callback daemon protocol "%s"', protocol)
             raise Exception('Unsupported callback daemon protocol.')
 
     extras['hooks_uri'] = callback_daemon.hooks_uri
     extras['hooks_protocol'] = protocol
     extras['time'] = time.time()
 
     # register txn_id
     extras['txn_id'] = txn_id
 
     log.debug('Prepared a callback daemon: %s at url `%s`',
               callback_daemon.__class__.__name__, callback_daemon.hooks_uri)
     return callback_daemon, extras
 
 
 class Hooks(object):
     """
     Exposes the hooks for remote call backs
     """
 
     def repo_size(self, extras):
         log.debug("Called repo_size of %s object", self)
         return self._call_hook(hooks_base.repo_size, extras)
 
     def pre_pull(self, extras):
         log.debug("Called pre_pull of %s object", self)
         return self._call_hook(hooks_base.pre_pull, extras)
 
     def post_pull(self, extras):
         log.debug("Called post_pull of %s object", self)
         return self._call_hook(hooks_base.post_pull, extras)
 
     def pre_push(self, extras):
         log.debug("Called pre_push of %s object", self)
         return self._call_hook(hooks_base.pre_push, extras)
 
     def post_push(self, extras):
         log.debug("Called post_push of %s object", self)
         return self._call_hook(hooks_base.post_push, extras)
 
     def _call_hook(self, hook, extras):
         extras = AttributeDict(extras)
         server_url = extras['server_url']
         request = bootstrap_request(application_url=server_url)
 
         bootstrap_config(request)  # inject routes and other interfaces
 
         # inject the user for usage in hooks
         request.user = AttributeDict({'username': extras.username,
                                       'ip_addr': extras.ip,
                                       'user_id': extras.user_id})
 
         extras.request = request
 
         try:
             result = hook(extras)
         except HTTPBranchProtected as handled_error:
             # Those special cases doesn't need error reporting. It's a case of
             # locked repo or protected branch
             result = AttributeDict({
                 'status': handled_error.code,
                 'output': handled_error.explanation
             })
         except (HTTPLockedRC, Exception) as error:
             # locked needs different handling since we need to also
             # handle PULL operations
             exc_tb = ''
             if not isinstance(error, HTTPLockedRC):
                 exc_tb = traceback.format_exc()
                 log.exception('Exception when handling hook %s', hook)
             error_args = error.args
             return {
                 'status': 128,
                 'output': '',
                 'exception': type(error).__name__,
                 'exception_traceback': exc_tb,
                 'exception_args': error_args,
             }
         finally:
             meta.Session.remove()
 
         log.debug('Got hook call response %s', result)
         return {
             'status': result.status,
             'output': result.output,
         }
 
     def __enter__(self):
         return self
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         pass

rhodecode/lib/middleware/simplesvn.py

@@ -1,204 +1,204 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2018 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 base64
 import logging
 import urllib
 from urlparse import urljoin
 
 import requests
 from pyramid.httpexceptions import HTTPNotAcceptable
 
 from rhodecode.lib import rc_cache
 from rhodecode.lib.middleware import simplevcs
 from rhodecode.lib.utils import is_valid_repo
 from rhodecode.lib.utils2 import str2bool, safe_int
 from rhodecode.lib.ext_json import json
 from rhodecode.lib.hooks_daemon import store_txn_id_data
 
 
 log = logging.getLogger(__name__)
 
 
 class SimpleSvnApp(object):
     IGNORED_HEADERS = [
         'connection', 'keep-alive', 'content-encoding',
         'transfer-encoding', 'content-length']
     rc_extras = {}
 
     def __init__(self, config):
         self.config = config
 
     def __call__(self, environ, start_response):
         request_headers = self._get_request_headers(environ)
 
         data = environ['wsgi.input']
         # johbo: Avoid that we end up with sending the request in chunked
         # transfer encoding (mainly on Gunicorn). If we know the content
         # length, then we should transfer the payload in one request.
         if environ['REQUEST_METHOD'] == 'MKCOL' or 'CONTENT_LENGTH' in environ:
             data = data.read()
             if data.startswith('(create-txn-with-props'):
                 # store on-the-fly our rc_extra using svn revision properties
                 # those can be read later on in hooks executed so we have a way
                 # to pass in the data into svn hooks
                 rc_data = base64.urlsafe_b64encode(json.dumps(self.rc_extras))
                 rc_data_len = len(rc_data)
                 # header defines data lenght, and serialized data
                 skel = ' rc-scm-extras {} {}'.format(rc_data_len, rc_data)
                 data = data[:-2] + skel + '))'
 
         log.debug('Calling: %s method via `%s`', environ['REQUEST_METHOD'],
                   self._get_url(environ['PATH_INFO']))
 
         response = requests.request(
             environ['REQUEST_METHOD'], self._get_url(environ['PATH_INFO']),
             data=data, headers=request_headers)
 
         if response.status_code not in [200, 401]:
             if response.status_code >= 500:
-                log.error('Got SVN response:%s with text:`%s`',
+                log.error('Got SVN response:%s with text:\n`%s`',
                           response, response.text)
             else:
-                log.debug('Got SVN response:%s with text:`%s`',
+                log.debug('Got SVN response:%s with text:\n`%s`',
                           response, response.text)
         else:
             log.debug('got response code: %s', response.status_code)
 
         response_headers = self._get_response_headers(response.headers)
 
         if response.headers.get('SVN-Txn-name'):
             svn_tx_id = response.headers.get('SVN-Txn-name')
             txn_id = rc_cache.utils.compute_key_from_params(
                 self.config['repository'], svn_tx_id)
             port = safe_int(self.rc_extras['hooks_uri'].split(':')[-1])
             store_txn_id_data(txn_id, {'port': port})
 
         start_response(
             '{} {}'.format(response.status_code, response.reason),
             response_headers)
         return response.iter_content(chunk_size=1024)
 
     def _get_url(self, path):
         url_path = urljoin(
             self.config.get('subversion_http_server_url', ''), path)
         url_path = urllib.quote(url_path, safe="/:=~+!$,;'")
         return url_path
 
     def _get_request_headers(self, environ):
         headers = {}
 
         for key in environ:
             if not key.startswith('HTTP_'):
                 continue
             new_key = key.split('_')
             new_key = [k.capitalize() for k in new_key[1:]]
             new_key = '-'.join(new_key)
             headers[new_key] = environ[key]
 
         if 'CONTENT_TYPE' in environ:
             headers['Content-Type'] = environ['CONTENT_TYPE']
 
         if 'CONTENT_LENGTH' in environ:
             headers['Content-Length'] = environ['CONTENT_LENGTH']
 
         return headers
 
     def _get_response_headers(self, headers):
         headers = [
             (h, headers[h])
             for h in headers
             if h.lower() not in self.IGNORED_HEADERS
         ]
 
         return headers
 
 
 class DisabledSimpleSvnApp(object):
     def __init__(self, config):
         self.config = config
 
     def __call__(self, environ, start_response):
         reason = 'Cannot handle SVN call because: SVN HTTP Proxy is not enabled'
         log.warning(reason)
         return HTTPNotAcceptable(reason)(environ, start_response)
 
 
 class SimpleSvn(simplevcs.SimpleVCS):
 
     SCM = 'svn'
     READ_ONLY_COMMANDS = ('OPTIONS', 'PROPFIND', 'GET', 'REPORT')
     DEFAULT_HTTP_SERVER = 'http://localhost:8090'
 
     def _get_repository_name(self, environ):
         """
         Gets repository name out of PATH_INFO header
 
         :param environ: environ where PATH_INFO is stored
         """
         path = environ['PATH_INFO'].split('!')
         repo_name = path[0].strip('/')
 
         # SVN includes the whole path in it's requests, including
         # subdirectories inside the repo. Therefore we have to search for
         # the repo root directory.
         if not is_valid_repo(
                 repo_name, self.base_path, explicit_scm=self.SCM):
             current_path = ''
             for component in repo_name.split('/'):
                 current_path += component
                 if is_valid_repo(
                         current_path, self.base_path, explicit_scm=self.SCM):
                     return current_path
                 current_path += '/'
 
         return repo_name
 
     def _get_action(self, environ):
         return (
             'pull'
             if environ['REQUEST_METHOD'] in self.READ_ONLY_COMMANDS
             else 'push')
 
     def _should_use_callback_daemon(self, extras, environ, action):
         # only MERGE command triggers hooks, so we don't want to start
         # hooks server too many times. POST however starts the svn transaction
         # so we also need to run the init of callback daemon of POST
         if environ['REQUEST_METHOD'] in ['MERGE', 'POST']:
             return True
         return False
 
     def _create_wsgi_app(self, repo_path, repo_name, config):
         if self._is_svn_enabled():
             return SimpleSvnApp(config)
         # we don't have http proxy enabled return dummy request handler
         return DisabledSimpleSvnApp(config)
 
     def _is_svn_enabled(self):
         conf = self.repo_vcs_config
         return str2bool(conf.get('vcs_svn_proxy', 'http_requests_enabled'))
 
     def _create_config(self, extras, repo_name):
         conf = self.repo_vcs_config
         server_url = conf.get('vcs_svn_proxy', 'http_server_url')
         server_url = server_url or self.DEFAULT_HTTP_SERVER
 
         extras['subversion_http_server_url'] = server_url
         return extras