rhodecode-enterprise-ce Commit - r4878:c5087cb0

celery: update how reqquest object is passed arround....

super-admin -

r4878:c5087cb0 default

parent child

rhodecode/integrations/types/email.py

0 +2 -3

             # -*- coding: utf-8 -*-
             # Copyright (C) 2012-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 <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 __future__ import unicode_literals
             import logging
             import colander
             import deform.widget
             from mako.template import Template
             from rhodecode import events
             from rhodecode.model.validation_schema.widgets import CheckboxChoiceWidgetDesc
             from rhodecode.translation import _
             from rhodecode.lib.celerylib import run_task
             from rhodecode.lib.celerylib import tasks
             from rhodecode.integrations.types.base import (
                 IntegrationTypeBase, render_with_traceback)
             log = logging.getLogger(__name__)
             REPO_PUSH_TEMPLATE_PLAINTEXT = Template('''
             Commits:
             % for commit in data['push']['commits']:
             ${commit['url']} by ${commit['author']} at ${commit['date']}
             ${commit['message']}
             ----
             % endfor
             ''')
             REPO_PUSH_TEMPLATE_HTML = Template('''
             <!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Strict//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1-strict.dtd">
             <html xmlns="http://www.w3.org/1999/xhtml">
             <head>
                 <meta http-equiv="Content-Type" content="text/html; charset=utf-8" />
                 <meta name="viewport" content="width=device-width, initial-scale=1.0"/>
                 <title>${subject}</title>
                 <style type="text/css">
                     /* Based on The MailChimp Reset INLINE: Yes. */
                     #outlook a {padding:0;} /* Force Outlook to provide a "view in browser" menu link. */
                     body{width:100% !important; -webkit-text-size-adjust:100%; -ms-text-size-adjust:100%; margin:0; padding:0;}
                     /* Prevent Webkit and Windows Mobile platforms from changing default font sizes.*/
                     .ExternalClass {width:100%;} /* Force Hotmail to display emails at full width */
                     .ExternalClass, .ExternalClass p, .ExternalClass span, .ExternalClass font, .ExternalClass td, .ExternalClass div {line-height: 100%;}
                     /* Forces Hotmail to display normal line spacing.  More on that: http://www.emailonacid.com/forum/viewthread/43/ */
                     #backgroundTable {margin:0; padding:0; line-height: 100% !important;}
                     /* End reset */
                     /* defaults for images*/
                     img {outline:none; text-decoration:none; -ms-interpolation-mode: bicubic;}
                     a img {border:none;}
                     .image_fix {display:block;}
                     body {line-height:1.2em;}
                     p {margin: 0 0 20px;}
                     h1, h2, h3, h4, h5, h6 {color:#323232!important;}
                     a {color:#427cc9;text-decoration:none;outline:none;cursor:pointer;}
                     a:focus {outline:none;}
                     a:hover {color: #305b91;}
                     h1 a, h2 a, h3 a, h4 a, h5 a, h6 a {color:#427cc9!important;text-decoration:none!important;}
                     h1 a:active, h2 a:active,  h3 a:active, h4 a:active, h5 a:active, h6 a:active {color: #305b91!important;}
                     h1 a:visited, h2 a:visited,  h3 a:visited, h4 a:visited, h5 a:visited, h6 a:visited {color: #305b91!important;}
                     table {font-size:13px;border-collapse:collapse;mso-table-lspace:0pt;mso-table-rspace:0pt;}
                     table td {padding:.65em 1em .65em 0;border-collapse:collapse;vertical-align:top;text-align:left;}
                     input {display:inline;border-radius:2px;border-style:solid;border: 1px solid #dbd9da;padding:.5em;}
                     input:focus {outline: 1px solid #979797}
                     @media only screen and (-webkit-min-device-pixel-ratio: 2) {
                     /* Put your iPhone 4g styles in here */
                     }
                     /* Android targeting */
                     @media only screen and (-webkit-device-pixel-ratio:.75){
                     /* Put CSS for low density (ldpi) Android layouts in here */
                     }
                     @media only screen and (-webkit-device-pixel-ratio:1){
                     /* Put CSS for medium density (mdpi) Android layouts in here */
                     }
                     @media only screen and (-webkit-device-pixel-ratio:1.5){
                     /* Put CSS for high density (hdpi) Android layouts in here */
                     }
                     /* end Android targeting */
                 </style>
                 <!-- Targeting Windows Mobile -->
                 <!--[if IEMobile 7]>
                 <style type="text/css">
                 </style>
                 <![endif]-->
                 <!--[if gte mso 9]>
                     <style>
                     /* Target Outlook 2007 and 2010 */
                     </style>
                 <![endif]-->
             </head>
             <body>
             <!-- Wrapper/Container Table: Use a wrapper table to control the width and the background color consistently of your email. Use this approach instead of setting attributes on the body tag. -->
             <table cellpadding="0" cellspacing="0" border="0" id="backgroundTable" align="left" style="margin:1%;width:97%;padding:0;font-family:sans-serif;font-weight:100;border:1px solid #dbd9da">
                 <tr>
                     <td valign="top" style="padding:0;">
                         <table cellpadding="0" cellspacing="0" border="0" align="left" width="100%">
                             <tr><td style="width:100%;padding:7px;background-color:#202020" valign="top">
                                 <a style="color:#eeeeee;text-decoration:none;" href="${instance_url}">
                                     ${'RhodeCode'}
                                 </a>
                             </td></tr>
                             <tr>
                                 <td style="padding:15px;" valign="top">
                                     % if data['push']['commits']:
                                         % for commit in data['push']['commits']:
                                         <a href="${commit['url']}">${commit['short_id']}</a> by ${commit['author']} at ${commit['date']} <br/>
                                         ${commit['message_html']} <br/>
                                         <br/>
                                         % endfor
                                     % else:
                                         No commit data
                                     % endif
                                 </td>
                             </tr>
                         </table>
                     </td>
                 </tr>
             </table>
             <!-- End of wrapper table -->
             <p><a style="margin-top:15px;margin-left:1%;font-family:sans-serif;font-weight:100;font-size:11px;color:#666666;text-decoration:none;" href="${instance_url}">
                 ${'This is a notification from RhodeCode. %(instance_url)s' % {'instance_url': instance_url}}
             </a></p>
             </body>
             </html>
             ''')
             class EmailSettingsSchema(colander.Schema):
                 @colander.instantiate(validator=colander.Length(min=1))
                 class recipients(colander.SequenceSchema):
                     title = _('Recipients')
                     description = _('Email addresses to send push events to')
                     widget = deform.widget.SequenceWidget(min_len=1)
                     recipient = colander.SchemaNode(
                         colander.String(),
                         title=_('Email address'),
                         description=_('Email address'),
                         default='',
                         validator=colander.Email(),
                         widget=deform.widget.TextInputWidget(
                             placeholder='user@domain.com',
                         ),
                     )
             class EmailIntegrationType(IntegrationTypeBase):
                 key = 'email'
                 display_name = _('Email')
                 description = _('Send repo push summaries to a list of recipients via email')
                 valid_events = [
                     events.RepoPushEvent
                 ]
                 @classmethod
                 def icon(cls):
                     return '''
             <?xml version="1.0" encoding="UTF-8" standalone="no"?>
             <svg
                xmlns:dc="http://purl.org/dc/elements/1.1/"
                xmlns:cc="http://creativecommons.org/ns#"
                xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"
                xmlns:svg="http://www.w3.org/2000/svg"
                xmlns="http://www.w3.org/2000/svg"
                xmlns:sodipodi="http://sodipodi.sourceforge.net/DTD/sodipodi-0.dtd"
                xmlns:inkscape="http://www.inkscape.org/namespaces/inkscape"
                viewBox="0 -256 1850 1850"
                id="svg2989"
                version="1.1"
                inkscape:version="0.48.3.1 r9886"
                width="100%"
                height="100%"
                sodipodi:docname="envelope_font_awesome.svg">
               <metadata
                  id="metadata2999">
                 <rdf:RDF>
                   <cc:Work
                      rdf:about="">
                     <dc:format>image/svg+xml</dc:format>
                     <dc:type
                        rdf:resource="http://purl.org/dc/dcmitype/StillImage" />
                   </cc:Work>
                 </rdf:RDF>
               </metadata>
               <defs
                  id="defs2997" />
               <sodipodi:namedview
                  pagecolor="#ffffff"
                  bordercolor="#666666"
                  borderopacity="1"
                  objecttolerance="10"
                  gridtolerance="10"
                  guidetolerance="10"
                  inkscape:pageopacity="0"
                  inkscape:pageshadow="2"
                  inkscape:window-width="640"
                  inkscape:window-height="480"
                  id="namedview2995"
                  showgrid="false"
                  inkscape:zoom="0.13169643"
                  inkscape:cx="896"
                  inkscape:cy="896"
                  inkscape:window-x="0"
                  inkscape:window-y="25"
                  inkscape:window-maximized="0"
                  inkscape:current-layer="svg2989" />
               <g
                  transform="matrix(1,0,0,-1,37.966102,1282.678)"
                  id="g2991">
                 <path
                    d="m 1664,32 v 768 q -32,-36 -69,-66 -268,-206 -426,-338 -51,-43 -83,-67 -32,-24 -86.5,-48.5 Q 945,256 897,256 h -1 -1 Q 847,256 792.5,280.5 738,305 706,329 674,353 623,396 465,528 197,734 160,764 128,800 V 32 Q 128,19 137.5,9.5 147,0 160,0 h 1472 q 13,0 22.5,9.5 9.5,9.5 9.5,22.5 z m 0,1051 v 11 13.5 q 0,0 -0.5,13 -0.5,13 -3,12.5 -2.5,-0.5 -5.5,9 -3,9.5 -9,7.5 -6,-2 -14,2.5 H 160 q -13,0 -22.5,-9.5 Q 128,1133 128,1120 128,952 275,836 468,684 676,519 682,514 711,489.5 740,465 757,452 774,439 801.5,420.5 829,402 852,393 q 23,-9 43,-9 h 1 1 q 20,0 43,9 23,9 50.5,27.5 27.5,18.5 44.5,31.5 17,13 46,37.5 29,24.5 35,29.5 208,165 401,317 54,43 100.5,115.5 46.5,72.5 46.5,131.5 z m 128,37 V 32 q 0,-66 -47,-113 -47,-47 -113,-47 H 160 Q 94,-128 47,-81 0,-34 0,32 v 1088 q 0,66 47,113 47,47 113,47 h 1472 q 66,0 113,-47 47,-47 47,-113 z"
                    id="path2993"
                    inkscape:connector-curvature="0"
                    style="fill:currentColor" />
               </g>
             </svg>
             '''
                 def settings_schema(self):
                     schema = EmailSettingsSchema()
                     schema.add(colander.SchemaNode(
                         colander.Set(),
                         widget=CheckboxChoiceWidgetDesc(
                             values=sorted(
                                 [(e.name, e.display_name, e.description) for e in self.valid_events]
                             ),
                         ),
                         description="List of events activated for this integration",
                         name='events'
                     ))
                     return schema
                 def send_event(self, event):
                     log.debug('handling event %s with integration %s', event.name, self)
                     if event.__class__ not in self.valid_events:
                         log.debug('event %r not present in valid event list (%s)', event, self.valid_events)
                         return
                     if not self.event_enabled(event):
                         # NOTE(marcink): for legacy reasons we're skipping this check...
                         # since the email event haven't had any settings...
                         pass
                     handler = EmailEventHandler(self.settings)
                     handler(event, event_data=event.as_dict())
             class EmailEventHandler(object):
                 def __init__(self, integration_settings):
                     self.integration_settings = integration_settings
                 def __call__(self, event, event_data):
                     if isinstance(event, events.RepoPushEvent):
                         self.repo_push_handler(event, event_data)
                     else:
                         log.debug('ignoring event: %r', event)
                 def repo_push_handler(self, event, data):
                     commit_num = len(data['push']['commits'])
                     server_url = data['server_url']
                     if commit_num == 1:
                         if data['push']['branches']:
                             _subject = '[{repo_name}] {author} pushed {commit_num} commit on branches: {branches}'
                         else:
                             _subject = '[{repo_name}] {author} pushed {commit_num} commit'
                         subject = _subject.format(
                             author=data['actor']['username'],
                             repo_name=data['repo']['repo_name'],
                             commit_num=commit_num,
                             branches=', '.join(
                                 branch['name'] for branch in data['push']['branches'])
                         )
                     else:
                         if data['push']['branches']:
                             _subject = '[{repo_name}] {author} pushed {commit_num} commits on branches: {branches}'
                         else:
                             _subject = '[{repo_name}] {author} pushed {commit_num} commits'
                         subject = _subject.format(
                             author=data['actor']['username'],
                             repo_name=data['repo']['repo_name'],
                             commit_num=commit_num,
                             branches=', '.join(
                                 branch['name'] for branch in data['push']['branches']))
                     email_body_plaintext = render_with_traceback(
                         REPO_PUSH_TEMPLATE_PLAINTEXT,
                         data=data,
                         subject=subject,
                         instance_url=server_url)
                     email_body_html = render_with_traceback(
                         REPO_PUSH_TEMPLATE_HTML,
                         data=data,
                         subject=subject,
                         instance_url=server_url)
                     recipients = self.integration_settings['recipients']
                     for email_address in recipients:
-                        run_task(
+                        run_task(tasks.send_email, email_address, subject,
-                            tasks.send_email, email_address, subject,
+                                 email_body_plaintext, email_body_html)
-                            email_body_plaintext, email_body_html)

rhodecode/lib/celerylib/loader.py

0 +31 -34

             # -*- 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 <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/
             """
             Celery loader, run with::
                 celery worker \
                     --task-events \
                     --beat \
                     --autoscale=20,2 \
                     --max-tasks-per-child 1 \
                     --app rhodecode.lib.celerylib.loader \
                     --scheduler rhodecode.lib.celerylib.scheduler.RcScheduler \
                     --loglevel DEBUG --ini=.dev/dev.ini
             """
             import os
             import logging
             import importlib
             from celery import Celery
             from celery import signals
             from celery import Task
             from celery import exceptions  # pragma: no cover
             from kombu.serialization import register
-            from pyramid.threadlocal import get_current_request
             import rhodecode
-            from rhodecode.lib.auth import AuthUser
             from rhodecode.lib.celerylib.utils import parse_ini_vars, ping_db
             from rhodecode.lib.ext_json import json
             from rhodecode.lib.pyramid_utils import bootstrap, setup_logging
             from rhodecode.lib.utils2 import str2bool
             from rhodecode.model import meta
             register('json_ext', json.dumps, json.loads,
                      content_type='application/x-json-ext',
                      content_encoding='utf-8')
             log = logging.getLogger('celery.rhodecode.loader')
             def add_preload_arguments(parser):
                 parser.add_argument(
                     '--ini', default=None,
                     help='Path to ini configuration file.'
                 )
                 parser.add_argument(
                     '--ini-var', default=None,
                     help='Comma separated list of key=value to pass to ini.'
                 )
             def get_logger(obj):
                 custom_log = logging.getLogger(
                     'rhodecode.task.{}'.format(obj.__class__.__name__))
                 if rhodecode.CELERY_ENABLED:
                     try:
                         custom_log = obj.get_logger()
                     except Exception:
                         pass
                 return custom_log
             imports = ['rhodecode.lib.celerylib.tasks']
             try:
                 # try if we have EE tasks available
                 importlib.import_module('rc_ee')
                 imports.append('rc_ee.lib.celerylib.tasks')
             except ImportError:
                 pass
             base_celery_config = {
                 'result_backend': 'rpc://',
                 'result_expires': 60 * 60 * 24,
                 'result_persistent': True,
                 'imports': imports,
                 'worker_max_tasks_per_child': 100,
                 'accept_content': ['json_ext'],
                 'task_serializer': 'json_ext',
                 'result_serializer': 'json_ext',
                 'worker_hijack_root_logger': False,
                 'database_table_names': {
                     'task': 'beat_taskmeta',
                     'group': 'beat_groupmeta',
                 }
             }
             # init main celery app
             celery_app = Celery()
             celery_app.user_options['preload'].add(add_preload_arguments)
             ini_file_glob = None
             @signals.setup_logging.connect
             def setup_logging_callback(**kwargs):
                 setup_logging(ini_file_glob)
             @signals.user_preload_options.connect
             def on_preload_parsed(options, **kwargs):
                 from rhodecode.config.middleware import get_celery_config
                 ini_location = options['ini']
                 ini_vars = options['ini_var']
                 celery_app.conf['INI_PYRAMID'] = options['ini']
                 if ini_location is None:
                     print('You must provide the paste --ini argument')
                     exit(-1)
                 options = None
                 if ini_vars is not None:
                     options = parse_ini_vars(ini_vars)
                 global ini_file_glob
                 ini_file_glob = ini_location
                 log.debug('Bootstrapping RhodeCode application...')
                 try:
                     env = bootstrap(ini_location, options=options)
                 except Exception:
                     log.exception('Failed to bootstrap RhodeCode APP')
                     raise
                 log.debug('Got Pyramid ENV: %s', env)
                 celery_settings = get_celery_config(env['registry'].settings)
                 setup_celery_app(
                     app=env['app'], root=env['root'], request=env['request'],
                     registry=env['registry'], closer=env['closer'],
                     celery_settings=celery_settings)
                 # fix the global flag even if it's disabled via .ini file because this
                 # is a worker code that doesn't need this to be disabled.
                 rhodecode.CELERY_ENABLED = True
             @signals.task_prerun.connect
             def task_prerun_signal(task_id, task, args, **kwargs):
                 ping_db()
             @signals.task_success.connect
             def task_success_signal(result, **kwargs):
                 meta.Session.commit()
                 closer = celery_app.conf['PYRAMID_CLOSER']
                 if closer:
                     closer()
             @signals.task_retry.connect
             def task_retry_signal(
                     request, reason, einfo, **kwargs):
                 meta.Session.remove()
                 closer = celery_app.conf['PYRAMID_CLOSER']
                 if closer:
                     closer()
             @signals.task_failure.connect
             def task_failure_signal(
                     task_id, exception, args, kwargs, traceback, einfo, **kargs):
                 log.error('Task: %s failed !! exc_info: %s', task_id, einfo)
                 from rhodecode.lib.exc_tracking import store_exception
                 from rhodecode.lib.statsd_client import StatsdClient
                 meta.Session.remove()
                 # simulate sys.exc_info()
                 exc_info = (einfo.type, einfo.exception, einfo.tb)
                 store_exception(id(exc_info), exc_info, prefix='rhodecode-celery')
                 statsd = StatsdClient.statsd
                 if statsd:
                     exc_type = "{}.{}".format(einfo.__class__.__module__, einfo.__class__.__name__)
                     statsd.incr('rhodecode_exception_total',
                                 tags=["exc_source:celery", "type:{}".format(exc_type)])
                 closer = celery_app.conf['PYRAMID_CLOSER']
                 if closer:
                     closer()
             @signals.task_revoked.connect
             def task_revoked_signal(
                     request, terminated, signum, expired, **kwargs):
                 closer = celery_app.conf['PYRAMID_CLOSER']
                 if closer:
                     closer()
+            class UNSET(object):
+                pass
+            def set_celery_conf(app=UNSET(), root=UNSET(), request=UNSET(), registry=UNSET(), closer=UNSET()):
+                if request is not UNSET:
+                    celery_app.conf.update({'PYRAMID_REQUEST': request})
+                if registry is not UNSET:
+                    celery_app.conf.update({'PYRAMID_REGISTRY': registry})
             def setup_celery_app(app, root, request, registry, closer, celery_settings):
                 log.debug('Got custom celery conf: %s', celery_settings)
                 celery_config = base_celery_config
                 celery_config.update({
                     # store celerybeat scheduler db where the .ini file is
                     'beat_schedule_filename': registry.settings['celerybeat-schedule.path'],
                 })
                 celery_config.update(celery_settings)
                 celery_app.config_from_object(celery_config)
                 celery_app.conf.update({'PYRAMID_APP': app})
                 celery_app.conf.update({'PYRAMID_ROOT': root})
                 celery_app.conf.update({'PYRAMID_REQUEST': request})
                 celery_app.conf.update({'PYRAMID_REGISTRY': registry})
                 celery_app.conf.update({'PYRAMID_CLOSER': closer})
             def configure_celery(config, celery_settings):
                 """
                 Helper that is called from our application creation logic. It gives
                 connection info into running webapp and allows execution of tasks from
                 RhodeCode itself
                 """
                 # store some globals into rhodecode
                 rhodecode.CELERY_ENABLED = str2bool(
                     config.registry.settings.get('use_celery'))
                 if rhodecode.CELERY_ENABLED:
                     log.info('Configuring celery based on `%s` settings', celery_settings)
                     setup_celery_app(
                         app=None, root=None, request=None, registry=config.registry,
                         closer=None, celery_settings=celery_settings)
             def maybe_prepare_env(req):
                 environ = {}
                 try:
                     environ.update({
                         'PATH_INFO': req.environ['PATH_INFO'],
                         'SCRIPT_NAME': req.environ['SCRIPT_NAME'],
                         'HTTP_HOST': req.environ.get('HTTP_HOST', req.environ['SERVER_NAME']),
                         'SERVER_NAME': req.environ['SERVER_NAME'],
                         'SERVER_PORT': req.environ['SERVER_PORT'],
                         'wsgi.url_scheme': req.environ['wsgi.url_scheme'],
                     })
                 except Exception:
                     pass
                 return environ
             class RequestContextTask(Task):
                 """
                 This is a celery task which will create a rhodecode app instance context
                 for the task, patch pyramid with the original request
                 that created the task and also add the user to the context.
                 """
                 def apply_async(self, args=None, kwargs=None, task_id=None, producer=None,
                                 link=None, link_error=None, shadow=None, **options):
                     """ queue the job to run (we are in web request context here) """
+                    from rhodecode.lib.base import get_ip_addr
-                    req = self.app.conf['PYRAMID_REQUEST'] or get_current_request()
+                    req = self.app.conf['PYRAMID_REQUEST']
+                    if not req:
+                        raise ValueError('celery_app.conf is having empty PYRAMID_REQUEST key')
                     log.debug('Running Task with class: %s. Request Class: %s',
                               self.__class__, req.__class__)
-                    user_id = None
+                    user_id = 0
-                    ip_addr = None
                     # web case
                     if hasattr(req, 'user'):
-                        ip_addr = req.user.ip_addr
                         user_id = req.user.user_id
                     # api case
                     elif hasattr(req, 'rpc_user'):
-                        ip_addr = req.rpc_user.ip_addr
                         user_id = req.rpc_user.user_id
-                    else:
-                        if user_id and ip_addr:
-                            log.debug('Using data from celery proxy user')
-                        else:
+                    # we hook into kwargs since it is the only way to pass our data to
-                            raise Exception(
+                    # the celery worker
-                                'Unable to fetch required data from request: {}. \n'
+                    environ = maybe_prepare_env(req)
-                                'This task is required to be executed from context of '
+                    options['headers'] = options.get('headers', {})
-                                'request in a webapp. Task: {}'.format(
+                    options['headers'].update({
-                                    repr(req),
+                        'rhodecode_proxy_data': {
-                                    self.__class__
+                            'environ': environ,
+                            'auth_user': {
+                                'ip_addr': get_ip_addr(req.environ),
+                                'user_id': user_id
-                    if req:
+                            },
-                        # we hook into kwargs since it is the only way to pass our data to
-                        # the celery worker
+                    })
-                        environ = maybe_prepare_env(req)
-                        options['headers'] = options.get('headers', {})
-                        options['headers'].update({
-                            'rhodecode_proxy_data': {
-                                'environ': environ,
-                                'auth_user': {
-                                    'ip_addr': ip_addr,
-                                    'user_id': user_id
-                                },
-                        })
                     return super(RequestContextTask, self).apply_async(
                         args, kwargs, task_id, producer, link, link_error, shadow, **options)

rhodecode/lib/hooks_daemon.py

0 +27 -22

             # -*- 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 <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
             import socket
             import random
             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))
+                    request = getattr(self.server, 'request', None)
                     try:
-                        result = self._call_hook(method, extras)
+                        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(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):
+                def _call_hook_method(self, hooks, 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],
+                        "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(
-                        "Preparing HTTP callback daemon at `%s` and registering hook object: %s",
+                        "%s Preparing HTTP callback daemon registering hook object: %s",
-                        self.hooks_uri, HooksHttpHandler)
+                        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("Called repo_size of %s object", self)
+                    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("Called pre_pull of %s object", self)
+                    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("Called post_pull of %s object", self)
+                    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("Called pre_push of %s object", self)
+                    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("Called post_push of %s object", self)
+                    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']
-                    request = bootstrap_request(application_url=server_url)
-                    bootstrap_config(request)  # inject routes and other interfaces
+                    extras.request = self.request
-                    # 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)
                         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('Exception when handling hook %s', hook)
+                            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('Got hook call response %s', result)
+                    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

rhodecode/model/notification.py

0 +2 -3

             # -*- coding: utf-8 -*-
             # Copyright (C) 2011-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 <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/
             """
             Model for notifications
             """
             import logging
             import traceback
             import premailer
             from pyramid.threadlocal import get_current_request
             from sqlalchemy.sql.expression import false, true
             import rhodecode
             from rhodecode.lib import helpers as h
             from rhodecode.model import BaseModel
             from rhodecode.model.db import Notification, User, UserNotification
             from rhodecode.model.meta import Session
             from rhodecode.translation import TranslationString
             log = logging.getLogger(__name__)
             class NotificationModel(BaseModel):
                 cls = Notification
                 def __get_notification(self, notification):
                     if isinstance(notification, Notification):
                         return notification
                     elif isinstance(notification, (int, long)):
                         return Notification.get(notification)
                     else:
                         if notification:
                             raise Exception('notification must be int, long or Instance'
                                             ' of Notification got %s' % type(notification))
                 def create(
                         self, created_by, notification_subject='', notification_body='',
                         notification_type=Notification.TYPE_MESSAGE, recipients=None,
                         mention_recipients=None, with_email=True, email_kwargs=None):
                     """
                     Creates notification of given type
                     :param created_by: int, str or User instance. User who created this
                         notification
                     :param notification_subject: subject of notification itself,
                         it will be generated automatically from notification_type if not specified
                     :param notification_body: body of notification text
                         it will be generated automatically from notification_type if not specified
                     :param notification_type: type of notification, based on that we
                         pick templates
                     :param recipients: list of int, str or User objects, when None
                         is given send to all admins
                     :param mention_recipients: list of int, str or User objects,
                         that were mentioned
                     :param with_email: send email with this notification
                     :param email_kwargs: dict with arguments to generate email
                     """
                     from rhodecode.lib.celerylib import tasks, run_task
                     if recipients and not getattr(recipients, '__iter__', False):
                         raise Exception('recipients must be an iterable object')
                     if not (notification_subject and notification_body) and not notification_type:
                         raise ValueError('notification_subject, and notification_body '
                                          'cannot be empty when notification_type is not specified')
                     created_by_obj = self._get_user(created_by)
                     if not created_by_obj:
                         raise Exception('unknown user %s' % created_by)
                     # default MAIN body if not given
                     email_kwargs = email_kwargs or {'body': notification_body}
                     mention_recipients = mention_recipients or set()
                     if recipients is None:
                         # recipients is None means to all admins
                         recipients_objs = User.query().filter(User.admin == true()).all()
                         log.debug('sending notifications %s to admins: %s',
                                   notification_type, recipients_objs)
                     else:
                         recipients_objs = set()
                         for u in recipients:
                             obj = self._get_user(u)
                             if obj:
                                 recipients_objs.add(obj)
                             else:  # we didn't find this user, log the error and carry on
                                 log.error('cannot notify unknown user %r', u)
                         if not recipients_objs:
                             raise Exception('no valid recipients specified')
                         log.debug('sending notifications %s to %s',
                                   notification_type, recipients_objs)
                     # add mentioned users into recipients
                     final_recipients = set(recipients_objs).union(mention_recipients)
                     (subject, email_body, email_body_plaintext) = \
                         EmailNotificationModel().render_email(notification_type, **email_kwargs)
                     if not notification_subject:
                         notification_subject = subject
                     if not notification_body:
                         notification_body = email_body_plaintext
                     notification = Notification.create(
                         created_by=created_by_obj, subject=notification_subject,
                         body=notification_body, recipients=final_recipients,
                         type_=notification_type
                     )
                     if not with_email:  # skip sending email, and just create notification
                         return notification
                     # don't send email to person who created this comment
                     rec_objs = set(recipients_objs).difference({created_by_obj})
                     # now notify all recipients in question
                     for recipient in rec_objs.union(mention_recipients):
                         # inject current recipient
                         email_kwargs['recipient'] = recipient
                         email_kwargs['mention'] = recipient in mention_recipients
                         (subject, email_body, email_body_plaintext) = EmailNotificationModel().render_email(
                             notification_type, **email_kwargs)
                         extra_headers = None
                         if 'thread_ids' in email_kwargs:
                             extra_headers = {'thread_ids': email_kwargs.pop('thread_ids')}
                         log.debug('Creating notification email task for user:`%s`', recipient)
-                        task = run_task(
+                        task = run_task(tasks.send_email, recipient.email, subject,
-                            tasks.send_email, recipient.email, subject,
+                                        email_body_plaintext, email_body, extra_headers=extra_headers)
-                            email_body_plaintext, email_body, extra_headers=extra_headers)
                         log.debug('Created email task: %s', task)
                     return notification
                 def delete(self, user, notification):
                     # we don't want to remove actual notification just the assignment
                     try:
                         notification = self.__get_notification(notification)
                         user = self._get_user(user)
                         if notification and user:
                             obj = UserNotification.query()\
                                 .filter(UserNotification.user == user)\
                                 .filter(UserNotification.notification == notification)\
                                 .one()
                             Session().delete(obj)
                             return True
                     except Exception:
                         log.error(traceback.format_exc())
                         raise
                 def get_for_user(self, user, filter_=None):
                     """
                     Get mentions for given user, filter them if filter dict is given
                     """
                     user = self._get_user(user)
                     q = UserNotification.query()\
                         .filter(UserNotification.user == user)\
                         .join((
                             Notification, UserNotification.notification_id ==
                             Notification.notification_id))
                     if filter_ == ['all']:
                         q = q  # no filter
                     elif filter_ == ['unread']:
                         q = q.filter(UserNotification.read == false())
                     elif filter_:
                         q = q.filter(Notification.type_.in_(filter_))
                     return q
                 def mark_read(self, user, notification):
                     try:
                         notification = self.__get_notification(notification)
                         user = self._get_user(user)
                         if notification and user:
                             obj = UserNotification.query()\
                                 .filter(UserNotification.user == user)\
                                 .filter(UserNotification.notification == notification)\
                                 .one()
                             obj.read = True
                             Session().add(obj)
                             return True
                     except Exception:
                         log.error(traceback.format_exc())
                         raise
                 def mark_all_read_for_user(self, user, filter_=None):
                     user = self._get_user(user)
                     q = UserNotification.query()\
                         .filter(UserNotification.user == user)\
                         .filter(UserNotification.read == false())\
                         .join((
                             Notification, UserNotification.notification_id ==
                             Notification.notification_id))
                     if filter_ == ['unread']:
                         q = q.filter(UserNotification.read == false())
                     elif filter_:
                         q = q.filter(Notification.type_.in_(filter_))
                     # this is a little inefficient but sqlalchemy doesn't support
                     # update on joined tables :(
                     for obj in q.all():
                         obj.read = True
                         Session().add(obj)
                 def get_unread_cnt_for_user(self, user):
                     user = self._get_user(user)
                     return UserNotification.query()\
                         .filter(UserNotification.read == false())\
                         .filter(UserNotification.user == user).count()
                 def get_unread_for_user(self, user):
                     user = self._get_user(user)
                     return [x.notification for x in UserNotification.query()
                             .filter(UserNotification.read == false())
                             .filter(UserNotification.user == user).all()]
                 def get_user_notification(self, user, notification):
                     user = self._get_user(user)
                     notification = self.__get_notification(notification)
                     return UserNotification.query()\
                         .filter(UserNotification.notification == notification)\
                         .filter(UserNotification.user == user).scalar()
                 def make_description(self, notification, translate, show_age=True):
                     """
                     Creates a human readable description based on properties
                     of notification object
                     """
                     _ = translate
                     _map = {
                         notification.TYPE_CHANGESET_COMMENT: [
                             _('%(user)s commented on commit %(date_or_age)s'),
                             _('%(user)s commented on commit at %(date_or_age)s'),
                             ],
                         notification.TYPE_MESSAGE: [
                             _('%(user)s sent message %(date_or_age)s'),
                             _('%(user)s sent message at %(date_or_age)s'),
                             ],
                         notification.TYPE_MENTION: [
                             _('%(user)s mentioned you %(date_or_age)s'),
                             _('%(user)s mentioned you at %(date_or_age)s'),
                             ],
                         notification.TYPE_REGISTRATION: [
                             _('%(user)s registered in RhodeCode %(date_or_age)s'),
                             _('%(user)s registered in RhodeCode at %(date_or_age)s'),
                             ],
                         notification.TYPE_PULL_REQUEST: [
                             _('%(user)s opened new pull request %(date_or_age)s'),
                             _('%(user)s opened new pull request at %(date_or_age)s'),
                             ],
                         notification.TYPE_PULL_REQUEST_UPDATE: [
                             _('%(user)s updated pull request %(date_or_age)s'),
                             _('%(user)s updated pull request at %(date_or_age)s'),
                         ],
                         notification.TYPE_PULL_REQUEST_COMMENT: [
                             _('%(user)s commented on pull request %(date_or_age)s'),
                             _('%(user)s commented on pull request at %(date_or_age)s'),
                             ],
                     }
                     templates = _map[notification.type_]
                     if show_age:
                         template = templates[0]
                         date_or_age = h.age(notification.created_on)
                         if translate:
                             date_or_age = translate(date_or_age)
                         if isinstance(date_or_age, TranslationString):
                             date_or_age = date_or_age.interpolate()
                     else:
                         template = templates[1]
                         date_or_age = h.format_date(notification.created_on)
                     return template % {
                         'user': notification.created_by_user.username,
                         'date_or_age': date_or_age,
                     }
             # Templates for Titles, that could be overwritten by rcextensions
             # Title of email for pull-request update
             EMAIL_PR_UPDATE_SUBJECT_TEMPLATE = ''
             # Title of email for request for pull request review
             EMAIL_PR_REVIEW_SUBJECT_TEMPLATE = ''
             # Title of email for general comment on pull request
             EMAIL_PR_COMMENT_SUBJECT_TEMPLATE = ''
             # Title of email for general comment which includes status change on pull request
             EMAIL_PR_COMMENT_STATUS_CHANGE_SUBJECT_TEMPLATE = ''
             # Title of email for inline comment on a file in pull request
             EMAIL_PR_COMMENT_FILE_SUBJECT_TEMPLATE = ''
             # Title of email for general comment on commit
             EMAIL_COMMENT_SUBJECT_TEMPLATE = ''
             # Title of email for general comment which includes status change on commit
             EMAIL_COMMENT_STATUS_CHANGE_SUBJECT_TEMPLATE = ''
             # Title of email for inline comment on a file in commit
             EMAIL_COMMENT_FILE_SUBJECT_TEMPLATE = ''
             import cssutils
             # hijack css utils logger and replace with ours
             log = logging.getLogger('rhodecode.cssutils.premailer')
             cssutils.log.setLog(log)
             class EmailNotificationModel(BaseModel):
                 TYPE_COMMIT_COMMENT = Notification.TYPE_CHANGESET_COMMENT
                 TYPE_REGISTRATION = Notification.TYPE_REGISTRATION
                 TYPE_PULL_REQUEST = Notification.TYPE_PULL_REQUEST
                 TYPE_PULL_REQUEST_COMMENT = Notification.TYPE_PULL_REQUEST_COMMENT
                 TYPE_PULL_REQUEST_UPDATE = Notification.TYPE_PULL_REQUEST_UPDATE
                 TYPE_MAIN = Notification.TYPE_MESSAGE
                 TYPE_PASSWORD_RESET = 'password_reset'
                 TYPE_PASSWORD_RESET_CONFIRMATION = 'password_reset_confirmation'
                 TYPE_EMAIL_TEST = 'email_test'
                 TYPE_EMAIL_EXCEPTION = 'exception'
                 TYPE_UPDATE_AVAILABLE = 'update_available'
                 TYPE_TEST = 'test'
                 email_types = {
                     TYPE_MAIN:
                         'rhodecode:templates/email_templates/main.mako',
                     TYPE_TEST:
                         'rhodecode:templates/email_templates/test.mako',
                     TYPE_EMAIL_EXCEPTION:
                         'rhodecode:templates/email_templates/exception_tracker.mako',
                     TYPE_UPDATE_AVAILABLE:
                         'rhodecode:templates/email_templates/update_available.mako',
                     TYPE_EMAIL_TEST:
                         'rhodecode:templates/email_templates/email_test.mako',
                     TYPE_REGISTRATION:
                         'rhodecode:templates/email_templates/user_registration.mako',
                     TYPE_PASSWORD_RESET:
                         'rhodecode:templates/email_templates/password_reset.mako',
                     TYPE_PASSWORD_RESET_CONFIRMATION:
                         'rhodecode:templates/email_templates/password_reset_confirmation.mako',
                     TYPE_COMMIT_COMMENT:
                         'rhodecode:templates/email_templates/commit_comment.mako',
                     TYPE_PULL_REQUEST:
                         'rhodecode:templates/email_templates/pull_request_review.mako',
                     TYPE_PULL_REQUEST_COMMENT:
                         'rhodecode:templates/email_templates/pull_request_comment.mako',
                     TYPE_PULL_REQUEST_UPDATE:
                         'rhodecode:templates/email_templates/pull_request_update.mako',
                 }
                 premailer_instance = premailer.Premailer()
                 def __init__(self):
                     """
                     Example usage::
                         (subject, email_body, email_body_plaintext) = EmailNotificationModel().render_email(
                             EmailNotificationModel.TYPE_TEST, **email_kwargs)
                     """
                     super(EmailNotificationModel, self).__init__()
                     self.rhodecode_instance_name = rhodecode.CONFIG.get('rhodecode_title')
                 def _update_kwargs_for_render(self, kwargs):
                     """
                     Inject params required for Mako rendering
                     :param kwargs:
                     """
                     kwargs['rhodecode_instance_name'] = self.rhodecode_instance_name
                     kwargs['rhodecode_version'] = rhodecode.__version__
                     instance_url = h.route_url('home')
                     _kwargs = {
                         'instance_url': instance_url,
                         'whitespace_filter': self.whitespace_filter,
                         'email_pr_update_subject_template': EMAIL_PR_UPDATE_SUBJECT_TEMPLATE,
                         'email_pr_review_subject_template': EMAIL_PR_REVIEW_SUBJECT_TEMPLATE,
                         'email_pr_comment_subject_template': EMAIL_PR_COMMENT_SUBJECT_TEMPLATE,
                         'email_pr_comment_status_change_subject_template': EMAIL_PR_COMMENT_STATUS_CHANGE_SUBJECT_TEMPLATE,
                         'email_pr_comment_file_subject_template': EMAIL_PR_COMMENT_FILE_SUBJECT_TEMPLATE,
                         'email_comment_subject_template': EMAIL_COMMENT_SUBJECT_TEMPLATE,
                         'email_comment_status_change_subject_template': EMAIL_COMMENT_STATUS_CHANGE_SUBJECT_TEMPLATE,
                         'email_comment_file_subject_template': EMAIL_COMMENT_FILE_SUBJECT_TEMPLATE,
                     }
                     _kwargs.update(kwargs)
                     return _kwargs
                 def whitespace_filter(self, text):
                     return text.replace('\n', '').replace('\t', '')
                 def get_renderer(self, type_, request):
                     template_name = self.email_types[type_]
                     return request.get_partial_renderer(template_name)
                 def render_email(self, type_, **kwargs):
                     """
                     renders template for email, and returns a tuple of
                     (subject, email_headers, email_html_body, email_plaintext_body)
                     """
                     request = get_current_request()
                     # translator and helpers inject
                     _kwargs = self._update_kwargs_for_render(kwargs)
                     email_template = self.get_renderer(type_, request=request)
                     subject = email_template.render('subject', **_kwargs)
                     try:
                         body_plaintext = email_template.render('body_plaintext', **_kwargs)
                     except AttributeError:
                         # it's not defined in template, ok we can skip it
                         body_plaintext = ''
                     # render WHOLE template
                     body = email_template.render(None, **_kwargs)
                     try:
                         # Inline CSS styles and conversion
                         body = self.premailer_instance.transform(body)
                     except Exception:
                         log.exception('Failed to parse body with premailer')
                         pass
                     return subject, body, body_plaintext

rhodecode/subscribers.py

0 +6 0

             # -*- 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 <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 io
             import shlex
             import math
             import re
             import os
             import datetime
             import logging
             import Queue
             import subprocess32
             from dateutil.parser import parse
             from pyramid.threadlocal import get_current_request
             from pyramid.interfaces import IRoutesMapper
             from pyramid.settings import asbool
             from pyramid.path import AssetResolver
             from threading import Thread
             from rhodecode.config.jsroutes import generate_jsroutes_content
             from rhodecode.lib.base import get_auth_user
             import rhodecode
             log = logging.getLogger(__name__)
             def add_renderer_globals(event):
                 from rhodecode.lib import helpers
                 # TODO: When executed in pyramid view context the request is not available
                 # in the event. Find a better solution to get the request.
                 request = event['request'] or get_current_request()
                 # Add Pyramid translation as '_' to context
                 event['_'] = request.translate
                 event['_ungettext'] = request.plularize
                 event['h'] = helpers
             def set_user_lang(event):
                 request = event.request
                 cur_user = getattr(request, 'user', None)
                 if cur_user:
                     user_lang = cur_user.get_instance().user_data.get('language')
                     if user_lang:
                         log.debug('lang: setting current user:%s language to: %s', cur_user, user_lang)
                         event.request._LOCALE_ = user_lang
+            def update_celery_conf(event):
+                from rhodecode.lib.celerylib.loader import set_celery_conf
+                log.debug('Setting celery config from new request')
+                set_celery_conf(request=event.request, registry=event.request.registry)
             def add_request_user_context(event):
                 """
                 Adds auth user into request context
                 """
                 request = event.request
                 # access req_id as soon as possible
                 req_id = request.req_id
                 if hasattr(request, 'vcs_call'):
                     # skip vcs calls
                     return
                 if hasattr(request, 'rpc_method'):
                     # skip api calls
                     return
                 auth_user, auth_token = get_auth_user(request)
                 request.user = auth_user
                 request.user_auth_token = auth_token
                 request.environ['rc_auth_user'] = auth_user
                 request.environ['rc_auth_user_id'] = auth_user.user_id
                 request.environ['rc_req_id'] = req_id
             def reset_log_bucket(event):
                 """
                 reset the log bucket on new request
                 """
                 request = event.request
                 request.req_id_records_init()
             def scan_repositories_if_enabled(event):
                 """
                 This is subscribed to the `pyramid.events.ApplicationCreated` event. It
                 does a repository scan if enabled in the settings.
                 """
                 settings = event.app.registry.settings
                 vcs_server_enabled = settings['vcs.server.enable']
                 import_on_startup = settings['startup.import_repos']
                 if vcs_server_enabled and import_on_startup:
                     from rhodecode.model.scm import ScmModel
                     from rhodecode.lib.utils import repo2db_mapper, get_rhodecode_base_path
                     repositories = ScmModel().repo_scan(get_rhodecode_base_path())
                     repo2db_mapper(repositories, remove_obsolete=False)
             def write_metadata_if_needed(event):
                 """
                 Writes upgrade metadata
                 """
                 import rhodecode
                 from rhodecode.lib import system_info
                 from rhodecode.lib import ext_json
                 fname = '.rcmetadata.json'
                 ini_loc = os.path.dirname(rhodecode.CONFIG.get('__file__'))
                 metadata_destination = os.path.join(ini_loc, fname)
                 def get_update_age():
                     now = datetime.datetime.utcnow()
                     with open(metadata_destination, 'rb') as f:
                         data = ext_json.json.loads(f.read())
                         if 'created_on' in data:
                             update_date = parse(data['created_on'])
                             diff = now - update_date
                             return diff.total_seconds() / 60.0
                     return 0
                 def write():
                     configuration = system_info.SysInfo(
                         system_info.rhodecode_config)()['value']
                     license_token = configuration['config']['license_token']
                     setup = dict(
                         workers=configuration['config']['server:main'].get(
                             'workers', '?'),
                         worker_type=configuration['config']['server:main'].get(
                             'worker_class', 'sync'),
                     )
                     dbinfo = system_info.SysInfo(system_info.database_info)()['value']
                     del dbinfo['url']
                     metadata = dict(
                         desc='upgrade metadata info',
                         license_token=license_token,
                         created_on=datetime.datetime.utcnow().isoformat(),
                         usage=system_info.SysInfo(system_info.usage_info)()['value'],
                         platform=system_info.SysInfo(system_info.platform_type)()['value'],
                         database=dbinfo,
                         cpu=system_info.SysInfo(system_info.cpu)()['value'],
                         memory=system_info.SysInfo(system_info.memory)()['value'],
                         setup=setup
                     )
                     with open(metadata_destination, 'wb') as f:
                         f.write(ext_json.json.dumps(metadata))
                 settings = event.app.registry.settings
                 if settings.get('metadata.skip'):
                     return
                 # only write this every 24h, workers restart caused unwanted delays
                 try:
                     age_in_min = get_update_age()
                 except Exception:
                     age_in_min = 0
                 if age_in_min > 60 * 60 * 24:
                     return
                 try:
                     write()
                 except Exception:
                     pass
             def write_usage_data(event):
                 import rhodecode
                 from rhodecode.lib import system_info
                 from rhodecode.lib import ext_json
                 settings = event.app.registry.settings
                 instance_tag = settings.get('metadata.write_usage_tag')
                 if not settings.get('metadata.write_usage'):
                     return
                 def get_update_age(dest_file):
                     now = datetime.datetime.utcnow()
                     with open(dest_file, 'rb') as f:
                         data = ext_json.json.loads(f.read())
                         if 'created_on' in data:
                             update_date = parse(data['created_on'])
                             diff = now - update_date
                             return math.ceil(diff.total_seconds() / 60.0)
                     return 0
                 utc_date = datetime.datetime.utcnow()
                 hour_quarter = int(math.ceil((utc_date.hour + utc_date.minute/60.0) / 6.))
                 fname = '.rc_usage_{date.year}{date.month:02d}{date.day:02d}_{hour}.json'.format(
                     date=utc_date, hour=hour_quarter)
                 ini_loc = os.path.dirname(rhodecode.CONFIG.get('__file__'))
                 usage_dir = os.path.join(ini_loc, '.rcusage')
                 if not os.path.isdir(usage_dir):
                     os.makedirs(usage_dir)
                 usage_metadata_destination = os.path.join(usage_dir, fname)
                 try:
                     age_in_min = get_update_age(usage_metadata_destination)
                 except Exception:
                     age_in_min = 0
                 # write every 6th hour
                 if age_in_min and age_in_min < 60 * 6:
                     log.debug('Usage file created %s minutes ago, skipping (threshold: %s minutes)...',
                               age_in_min, 60 * 6)
                     return
                 def write(dest_file):
                     configuration = system_info.SysInfo(system_info.rhodecode_config)()['value']
                     license_token = configuration['config']['license_token']
                     metadata = dict(
                         desc='Usage data',
                         instance_tag=instance_tag,
                         license_token=license_token,
                         created_on=datetime.datetime.utcnow().isoformat(),
                         usage=system_info.SysInfo(system_info.usage_info)()['value'],
                     )
                     with open(dest_file, 'wb') as f:
                         f.write(ext_json.json.dumps(metadata, indent=2, sort_keys=True))
                 try:
                     log.debug('Writing usage file at: %s', usage_metadata_destination)
                     write(usage_metadata_destination)
                 except Exception:
                     pass
             def write_js_routes_if_enabled(event):
                 registry = event.app.registry
                 mapper = registry.queryUtility(IRoutesMapper)
                 _argument_prog = re.compile('\{(.*?)\}|:\((.*)\)')
                 def _extract_route_information(route):
                     """
                     Convert a route into tuple(name, path, args), eg:
                         ('show_user', '/profile/%(username)s', ['username'])
                     """
                     routepath = route.pattern
                     pattern = route.pattern
                     def replace(matchobj):
                         if matchobj.group(1):
                             return "%%(%s)s" % matchobj.group(1).split(':')[0]
                         else:
                             return "%%(%s)s" % matchobj.group(2)
                     routepath = _argument_prog.sub(replace, routepath)
                     if not routepath.startswith('/'):
                         routepath = '/'+routepath
                     return (
                         route.name,
                         routepath,
                         [(arg[0].split(':')[0] if arg[0] != '' else arg[1])
                           for arg in _argument_prog.findall(pattern)]
                     )
                 def get_routes():
                     # pyramid routes
                     for route in mapper.get_routes():
                         if not route.name.startswith('__'):
                             yield _extract_route_information(route)
                 if asbool(registry.settings.get('generate_js_files', 'false')):
                     static_path = AssetResolver().resolve('rhodecode:public').abspath()
                     jsroutes = get_routes()
                     jsroutes_file_content = generate_jsroutes_content(jsroutes)
                     jsroutes_file_path = os.path.join(
                         static_path, 'js', 'rhodecode', 'routes.js')
                     try:
                         with io.open(jsroutes_file_path, 'w', encoding='utf-8') as f:
                             f.write(jsroutes_file_content)
                     except Exception:
                         log.exception('Failed to write routes.js into %s', jsroutes_file_path)
             class Subscriber(object):
                 """
                 Base class for subscribers to the pyramid event system.
                 """
                 def __call__(self, event):
                     self.run(event)
                 def run(self, event):
                     raise NotImplementedError('Subclass has to implement this.')
             class AsyncSubscriber(Subscriber):
                 """
                 Subscriber that handles the execution of events in a separate task to not
                 block the execution of the code which triggers the event. It puts the
                 received events into a queue from which the worker process takes them in
                 order.
                 """
                 def __init__(self):
                     self._stop = False
                     self._eventq = Queue.Queue()
                     self._worker = self.create_worker()
                     self._worker.start()
                 def __call__(self, event):
                     self._eventq.put(event)
                 def create_worker(self):
                     worker = Thread(target=self.do_work)
                     worker.daemon = True
                     return worker
                 def stop_worker(self):
                     self._stop = False
                     self._eventq.put(None)
                     self._worker.join()
                 def do_work(self):
                     while not self._stop:
                         event = self._eventq.get()
                         if event is not None:
                             self.run(event)
             class AsyncSubprocessSubscriber(AsyncSubscriber):
                 """
                 Subscriber that uses the subprocess32 module to execute a command if an
                 event is received. Events are handled asynchronously::
                     subscriber = AsyncSubprocessSubscriber('ls -la', timeout=10)
                     subscriber(dummyEvent) # running __call__(event)
                 """
                 def __init__(self, cmd, timeout=None):
                     if not isinstance(cmd, (list, tuple)):
                         cmd = shlex.split(cmd)
                     super(AsyncSubprocessSubscriber, self).__init__()
                     self._cmd = cmd
                     self._timeout = timeout
                 def run(self, event):
                     cmd = self._cmd
                     timeout = self._timeout
                     log.debug('Executing command %s.', cmd)
                     try:
                         output = subprocess32.check_output(
                             cmd, timeout=timeout, stderr=subprocess32.STDOUT)
                         log.debug('Command finished %s', cmd)
                         if output:
                             log.debug('Command output: %s', output)
                     except subprocess32.TimeoutExpired as e:
                         log.exception('Timeout while executing command.')
                         if e.output:
                             log.error('Command output: %s', e.output)
                     except subprocess32.CalledProcessError as e:
                         log.exception('Error while executing command.')
                         if e.output:
                             log.error('Command output: %s', e.output)
                     except Exception:
                         log.exception(
                             'Exception while executing command %s.', cmd)

rhodecode/tweens.py

0 +2 0

             # -*- 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 <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 logging
             from pyramid.httpexceptions import HTTPException, HTTPBadRequest
             from rhodecode.lib.middleware.vcs import (
                 detect_vcs_request, VCS_TYPE_KEY, VCS_TYPE_SKIP)
             log = logging.getLogger(__name__)
             def vcs_detection_tween_factory(handler, registry):
                 def vcs_detection_tween(request):
                     """
                     Do detection of vcs type, and save results for other layers to re-use
                     this information
                     """
                     vcs_server_enabled = request.registry.settings.get('vcs.server.enable')
                     vcs_handler = vcs_server_enabled and detect_vcs_request(
                         request.environ, request.registry.settings.get('vcs.backends'))
                     if vcs_handler:
                         # save detected VCS type for later re-use
                         request.environ[VCS_TYPE_KEY] = vcs_handler.SCM
                         request.vcs_call = vcs_handler.SCM
                         log.debug('Processing request with `%s` handler', handler)
                         return handler(request)
                     # mark that we didn't detect an VCS, and we can skip detection later on
                     request.environ[VCS_TYPE_KEY] = VCS_TYPE_SKIP
                     log.debug('Processing request with `%s` handler', handler)
                     return handler(request)
                 return vcs_detection_tween
             def junk_encoding_detector(request):
                 """
                 Detect bad encoded GET params, and fail immediately with BadRequest
                 """
                 try:
                     request.GET.get("", None)
                 except UnicodeDecodeError:
                     raise HTTPBadRequest("Invalid bytes in query string.")
             def bad_url_data_detector(request):
                 """
                 Detect invalid bytes in a path.
                 """
                 try:
                     request.path_info
                 except UnicodeDecodeError:
                     raise HTTPBadRequest("Invalid bytes in URL.")
             def junk_form_data_detector(request):
                 """
                 Detect bad encoded POST params, and fail immediately with BadRequest
                 """
                 if request.method == "POST":
                     try:
                         request.POST.get("", None)
                     except ValueError:
                         raise HTTPBadRequest("Invalid bytes in form data.")
             def sanity_check_factory(handler, registry):
                 def sanity_check(request):
                     log.debug('Checking current URL sanity for bad data')
                     try:
                         junk_encoding_detector(request)
                         bad_url_data_detector(request)
                         junk_form_data_detector(request)
                     except HTTPException as exc:
                         return exc
                     return handler(request)
                 return sanity_check
             def includeme(config):
                 config.add_subscriber('rhodecode.subscribers.add_renderer_globals',
                                       'pyramid.events.BeforeRender')
+                config.add_subscriber('rhodecode.subscribers.update_celery_conf',
+                                      'pyramid.events.NewRequest')
                 config.add_subscriber('rhodecode.subscribers.set_user_lang',
                                       'pyramid.events.NewRequest')
                 config.add_subscriber('rhodecode.subscribers.reset_log_bucket',
                                       'pyramid.events.NewRequest')
                 config.add_subscriber('rhodecode.subscribers.add_request_user_context',
                                       'pyramid.events.ContextFound')
                 config.add_tween('rhodecode.tweens.vcs_detection_tween_factory')
                 config.add_tween('rhodecode.tweens.sanity_check_factory')
                 # This needs to be the LAST item
                 config.add_tween('rhodecode.lib.middleware.request_wrapper.RequestWrapperTween')
                 log.debug('configured all tweens')

General Comments 0

Write
Preview

You need to be logged in to leave comments. Login now

No TODOs yet

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages