rcextensions: added example for validation of commit messages for git.

rhodecode/config/rcextensions/examples/validate_commit_message_author.py

@@ -0,0 +1,91 b''
+# Example to validate commit message or author using some sort of rules
+
+
+@has_kwargs({
+    'server_url': 'url of instance that triggered this hook',
+    'config': 'path to .ini config used',
+    'scm': 'type of version control "git", "hg", "svn"',
+    'username': 'username of actor who triggered this event',
+    'ip': 'ip address of actor who triggered this hook',
+    'action': '',
+    'repository': 'repository name',
+    'repo_store_path': 'full path to where repositories are stored',
+    'commit_ids': 'pre transaction metadata for commit ids',
+    'hook_type': '',
+    'user_agent': 'Client user agent, e.g git or mercurial CLI version',
+})
+@has_kwargs({
+    'server_url': 'url of instance that triggered this hook',
+    'config': 'path to .ini config used',
+    'scm': 'type of version control "git", "hg", "svn"',
+    'username': 'username of actor who triggered this event',
+    'ip': 'ip address of actor who triggered this hook',
+    'action': '',
+    'repository': 'repository name',
+    'repo_store_path': 'full path to where repositories are stored',
+    'commit_ids': 'pre transaction metadata for commit ids',
+    'hook_type': '',
+    'user_agent': 'Client user agent, e.g git or mercurial CLI version',
+})
+def _pre_push_hook(*args, **kwargs):
+    """
+    Post push hook
+    To stop version control from storing the transaction and send a message to user
+    use non-zero HookResponse with a message, e.g return HookResponse(1, 'Not allowed')
+
+    This message will be shown back to client during PUSH operation
+
+    Commit ids might look like that::
+
+    [{u'hg_env|git_env': ...,
+      u'multiple_heads': [],
+      u'name': u'default',
+      u'new_rev': u'd0befe0692e722e01d5677f27a104631cf798b69',
+      u'old_rev': u'd0befe0692e722e01d5677f27a104631cf798b69',
+      u'ref': u'',
+      u'total_commits': 2,
+      u'type': u'branch'}]
+    """
+    import re
+    from .helpers import extra_fields, extract_pre_commits
+    from .utils import str2bool
+
+    # returns list of dicts with key-val fetched from extra fields
+    repo_extra_fields = extra_fields.run(**kwargs)
+
+    # optionally use 'extra fields' to control the logic per repo
+    should_validate = str2bool(repo_extra_fields.get('validate_author', True))
+
+    # optionally store validation regex into extra fields
+    validation_regex = repo_extra_fields.get('validation_regex', '')
+
+    def validate_commit_message(commit_message, message_regex=None):
+        """
+        This function validates commit_message against some sort of rules.
+        It should return a valid boolean, and a reason for failure
+        """
+
+        if "secret_string" in commit_message:
+            msg = "!!Push forbidden: secret string found in commit messages"
+            return False, msg
+
+        if validation_regex:
+            regexp = re.compile(validation_regex)
+            if not regexp.match(message):
+                msg = "!!Push forbidden: commit message does not match regexp"
+                return False, msg
+
+        return True, ''
+
+    if should_validate:
+        # returns list of dicts with key-val fetched from extra fields
+        commit_list = extract_pre_commits.run(**kwargs)
+
+        for commit_data in commit_list:
+            message = commit_data['message']
+
+            message_valid, reason = validate_commit_message(message, validation_regex)
+            if not message_valid:
+                return HookResponse(1, reason)
+
+    return HookResponse(0, '')

rhodecode/config/rcextensions/helpers/extract_pre_commits.py

@@ -1,63 +1,90 b''
 # -*- coding: utf-8 -*-
 # Copyright (C) 2016-2018 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 us in hooks::
 
     from .helpers import extract_pre_commits
     # returns list of dicts with key-val fetched from extra fields
     commit_list = extract_pre_commits.run(**kwargs)
 
 """
 import re
 import collections
+import json
 
 
 def get_hg_commits(repo, refs):
     commits = []
     return commits
 
 
 def get_git_commits(repo, refs):
     commits = []
+
+    for data in refs:
+        # we should now extract commit data
+        old_rev = data['old_rev']
+        new_rev = data['new_rev']
+
+        if '00000000' in old_rev:
+            # new branch, we don't need to extract nothing
+            return commits
+
+        git_env = dict(data['git_env'])
+        cmd = [
+            'log',
+            '--pretty=format:{"commit_id": "%H",  "author": "%aN <%aE>",  "date": "%ad",  "message": "%f"}',
+            '{}...{}'.format(old_rev, new_rev)
+        ]
+
+        stdout, stderr = repo.run_git_command(cmd, env=git_env)
+        for line in stdout.splitlines():
+            try:
+                data = json.loads(line)
+                commits.append(data)
+            except Exception:
+                print('Failed to load data from GIT line')
+
     return commits
 
 
 def run(*args, **kwargs):
     from rhodecode.model.db import Repository
 
     vcs_type = kwargs['scm']
     # use temp name then the main one propagated
     repo_name = kwargs.pop('REPOSITORY', None) or kwargs['repository']
 
     repo = Repository.get_by_repo_name(repo_name)
     vcs_repo = repo.scm_instance(cache=False)
 
     commits = []
 
+    if vcs_type == 'git':
     for rev_data in kwargs['commit_ids']:
-        new_environ = dict((k, v) for k, v in rev_data['hg_env'])
-
-    if vcs_type == 'git':
+            new_environ = dict((k, v) for k, v in rev_data['git_env'])
         commits = get_git_commits(vcs_repo, kwargs['commit_ids'])
 
     if vcs_type == 'hg':
+        for rev_data in kwargs['commit_ids']:
+            new_environ = dict((k, v) for k, v in rev_data['hg_env'])
         commits = get_hg_commits(vcs_repo, kwargs['commit_ids'])
 
     return commits

rhodecode/config/rcextensions/utils.py

@@ -1,147 +1,185 b''
 # Copyright (C) 2016-2018 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import os
 import functools
 import collections
 
 
 class HookResponse(object):
     def __init__(self, status, output):
         self.status = status
         self.output = output
 
     def __add__(self, other):
         other_status = getattr(other, 'status', 0)
         new_status = max(self.status, other_status)
         other_output = getattr(other, 'output', '')
         new_output = self.output + other_output
 
         return HookResponse(new_status, new_output)
 
     def __bool__(self):
         return self.status == 0
 
 
 class DotDict(dict):
 
     def __contains__(self, k):
         try:
             return dict.__contains__(self, k) or hasattr(self, k)
         except:
             return False
 
     # only called if k not found in normal places
     def __getattr__(self, k):
         try:
             return object.__getattribute__(self, k)
         except AttributeError:
             try:
                 return self[k]
             except KeyError:
                 raise AttributeError(k)
 
     def __setattr__(self, k, v):
         try:
             object.__getattribute__(self, k)
         except AttributeError:
             try:
                 self[k] = v
             except:
                 raise AttributeError(k)
         else:
             object.__setattr__(self, k, v)
 
     def __delattr__(self, k):
         try:
             object.__getattribute__(self, k)
         except AttributeError:
             try:
                 del self[k]
             except KeyError:
                 raise AttributeError(k)
         else:
             object.__delattr__(self, k)
 
     def toDict(self):
         return unserialize(self)
 
     def __repr__(self):
         keys = list(self.keys())
         keys.sort()
         args = ', '.join(['%s=%r' % (key, self[key]) for key in keys])
         return '%s(%s)' % (self.__class__.__name__, args)
 
     @staticmethod
     def fromDict(d):
         return serialize(d)
 
 
 def serialize(x):
     if isinstance(x, dict):
         return DotDict((k, serialize(v)) for k, v in x.items())
     elif isinstance(x, (list, tuple)):
         return type(x)(serialize(v) for v in x)
     else:
         return x
 
 
 def unserialize(x):
     if isinstance(x, dict):
         return dict((k, unserialize(v)) for k, v in x.items())
     elif isinstance(x, (list, tuple)):
         return type(x)(unserialize(v) for v in x)
     else:
         return x
 
 
 def _verify_kwargs(func_name, expected_parameters, kwargs):
     """
     Verify that exactly `expected_parameters` are passed in as `kwargs`.
     """
     expected_parameters = set(expected_parameters)
     kwargs_keys = set(kwargs.keys())
     if kwargs_keys != expected_parameters:
         missing_kwargs = expected_parameters - kwargs_keys
         unexpected_kwargs = kwargs_keys - expected_parameters
         raise AssertionError(
             "func:%s: missing parameters: %r, unexpected parameters: %s" %
             (func_name, missing_kwargs, unexpected_kwargs))
 
 
 def has_kwargs(required_args):
     """
     decorator to verify extension calls arguments.
 
     :param required_args:
     """
     def wrap(func):
         def wrapper(*args, **kwargs):
             _verify_kwargs(func.func_name, required_args.keys(), kwargs)
             # in case there's `calls` defined on module we store the data
             maybe_log_call(func.func_name, args, kwargs)
             return func(*args, **kwargs)
         return wrapper
     return wrap
 
 
 def maybe_log_call(name, args, kwargs):
     from rhodecode.config import rcextensions
     if hasattr(rcextensions, 'calls'):
         calls = rcextensions.calls
         calls[name].append((args, kwargs))
+
+
+def str2bool(_str):
+    """
+    returns True/False value from given string, it tries to translate the
+    string into boolean
+
+    :param _str: string value to translate into boolean
+    :rtype: boolean
+    :returns: boolean from given string
+    """
+    if _str is None:
+        return False
+    if _str in (True, False):
+        return _str
+    _str = str(_str).strip().lower()
+    return _str in ('t', 'true', 'y', 'yes', 'on', '1')
+
+
+def aslist(obj, sep=None, strip=True):
+    """
+    Returns given string separated by sep as list
+
+    :param obj:
+    :param sep:
+    :param strip:
+    """
+    if isinstance(obj, (basestring,)):
+        lst = obj.split(sep)
+        if strip:
+            lst = [v.strip() for v in lst]
+        return lst
+    elif isinstance(obj, (list, tuple)):
+        return obj
+    elif obj is None:
+        return []
+    else:
+        return [obj]
  No newline at end of file