# -*- coding: utf-8 -*- # Copyright (C) 2010-2020 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 . # # 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 import socket import msgpack from http.server 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): hooks_proto, method, extras = self._read_request() log.debug('Handling HooksHttpHandler %s with %s proto', method, hooks_proto) 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)) request = getattr(self.server, 'request', None) try: hooks = Hooks(request=request, log_prefix='HOOKS: {} '.format(self.server.server_address)) result = self._call_hook_method(hooks, 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(hooks_proto, result) def _read_request(self): length = int(self.headers['Content-Length']) hooks_proto = self.headers.get('rc-hooks-protocol') or 'json.v1' if hooks_proto == 'msgpack.v1': # support for new vcsserver msgpack based protocol hooks data = msgpack.unpackb(self.rfile.read(length), raw=False) else: body = self.rfile.read(length) data = json.loads(body) return hooks_proto, data['method'], data['extras'] def _write_response(self, hooks_proto, result): self.send_response(200) if hooks_proto == 'msgpack.v1': self.send_header("Content-type", "application/msgpack") self.end_headers() self.wfile.write(msgpack.packb(result)) else: self.send_header("Content-type", "text/json") self.end_headers() self.wfile.write(json.dumps(result)) def _call_hook_method(self, hooks, method, extras): 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( "HOOKS: %s - - [%s] %s", self.client_address, 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 `%s` callback daemon', self.__class__.__name__) return self def __exit__(self, exc_type, exc_val, exc_tb): log.debug('Exiting `%s` callback daemon', self.__class__.__name__) 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=host, port=port) def __enter__(self): log.debug('Running `%s` callback daemon', self.__class__.__name__) self._run() return self def __exit__(self, exc_type, exc_val, exc_tb): log.debug('Exiting `%s` callback daemon', self.__class__.__name__) 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 @property def _hook_prefix(self): return 'HOOKS: {} '.format(self.hooks_uri) def get_hostname(self): return socket.gethostname() or '127.0.0.1' def get_available_port(self, min_port=20000, max_port=65535): from rhodecode.lib.utils2 import get_available_port as _get_port return _get_port(min_port, max_port) def _prepare(self, txn_id=None, host=None, port=None): from pyramid.threadlocal import get_current_request if not host or host == "*": host = self.get_hostname() if not port: port = self.get_available_port() server_address = (host, port) self.hooks_uri = '{}:{}'.format(host, port) self.txn_id = txn_id self._done = False log.debug( "%s Preparing HTTP callback daemon registering hook object: %s", self._hook_prefix, HooksHttpHandler) self._daemon = TCPServer(server_address, HooksHttpHandler) # inject transaction_id for later verification self._daemon.txn_id = self.txn_id # pass the WEB app request into daemon self._daemon.request = get_current_request() 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): 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 __init__(self, request=None, log_prefix=''): self.log_prefix = log_prefix self.request = request def repo_size(self, extras): log.debug("%sCalled repo_size of %s object", self.log_prefix, self) return self._call_hook(hooks_base.repo_size, extras) def pre_pull(self, extras): log.debug("%sCalled pre_pull of %s object", self.log_prefix, self) return self._call_hook(hooks_base.pre_pull, extras) def post_pull(self, extras): log.debug("%sCalled post_pull of %s object", self.log_prefix, self) return self._call_hook(hooks_base.post_pull, extras) def pre_push(self, extras): log.debug("%sCalled pre_push of %s object", self.log_prefix, self) return self._call_hook(hooks_base.pre_push, extras) def post_push(self, extras): log.debug("%sCalled post_push of %s object", self.log_prefix, self) return self._call_hook(hooks_base.post_push, extras) def _call_hook(self, hook, extras): extras = AttributeDict(extras) server_url = extras['server_url'] extras.request = self.request try: result = hook(extras) if result is None: raise Exception( 'Failed to obtain hook result from func: {}'.format(hook)) 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('%sException when handling hook %s', self.log_prefix, 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('%sGot hook call response %s', self.log_prefix, result) return { 'status': result.status, 'output': result.output, } def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): pass