Coherent Ramblings

#!/usr/bin/env python
 
import os
import re
 
from sys import argv, exit
from subprocess import Popen, PIPE
 
 
def blank(line):
    regex = re.compile('^\s*$')
    if regex.search(line):
        return True
    else:
        return False
 
 
def uncomment(line):
    return line.partition('#')[0]
 
 
def cleanup(lines):
    # Remove full-line comments.
    lines = [line for line in lines if not line.startswith('#')]
    # Remove blank lines.
    lines = [line for line in lines if not blank(line)]
    # Remove comments from line ends.
    lines = [uncomment(line) for line in lines]
 
    return lines
 
def parse_acl():
    try:
        fh = open('acl')
    except IOError:
        print "Could not open ACL file. Exiting."
        exit(1)
 
    # Read and close ACL file.
    lines = fh.readlines()
    fh.close()
    lines = cleanup(lines)
 
    # The ACL lines are whitespace separated. The first field is the
    # username, the second field is a regex describing the refs the
    # user is allowed to update.
    access = {}
    for line in lines:
        record = line.split()
        r_user = record[0]
        r_ref = record[1]
        if r_user in access:
            access[r_user] = access[r_user] + [r_ref]
        else:
            access[r_user] = [r_ref]
 
    return access
 
 
def authorize(refname='', user='', access={}):
    if not user in access:
        return False
 
    allowed_refs = access[user]
    for a_ref in allowed_refs:
        regex = re.compile("^%s$" % a_ref)
        if regex.search(refname):
            return True
 
    return False
 
 
def parse_git_config(config_entry):
    output = Popen(["git", "config", "--bool", config_entry], stdout=PIPE).communicate()[0]
    if output.strip() == 'true':
        return True
    else:
        return False
 
 
def parse_branch_tracking():
    try:
        fh = open('branch_tracking')
    except IOError:
        print "Could not open branch tracking file. Exiting."
        exit(1)
 
    # Read and close branch tracking file.
    lines = fh.readlines()
    fh.close()
    lines = cleanup(lines)
 
    # Branch tracking lines are whitespace separated. The first field
    # is the branch that must track the branch specified in the second
    # field.
    tracking = {}
    for line in lines:
        record = line.split()
        r_branch = record[0]
        r_track = record[1]
        tracking[r_branch] = r_track
 
    return tracking
 
 
def get_tracked_branch(refname):
    track = parse_branch_tracking()
    try:
        return track[refname]
    except KeyError:
        return None
 
 
def get_missing_refs(ref, base_ref):
    # git rev-list gives us the commits reachable from base_ref that
    # are NOT reachable from ref.
    output = Popen(["git", "rev-list", '%s..%s' % (ref, base_ref)], stdout=PIPE).communicate()[0]
    return output.splitlines()
 
 
def main():
    # The args are passed in by git.
    refname = argv[1]
    old_rev = argv[2]
    new_rev = argv[3]
    user = os.environ['USER']
 
    # Check if 'push acl' is enabled.
    push_acl = parse_git_config('hooks.pushacl')
 
    if push_acl:
        # Check the user's authorization to update these git refs.
        access = parse_acl()
        if not authorize(refname=refname, user=user, access=access):
            print "Could not update %s, permission denied by ACL." % refname
            exit(1)
 
    # Check if 'forced branch tracking' is enabled.
    force_tracking = parse_git_config('hooks.forcebranchtracking')
 
    if force_tracking:
        tbranch = get_tracked_branch(refname)
        if tbranch:
            # Get the refs that are reachable from our tracked branch
            # but NOT reachable from our new revision.
            missed_refs = get_missing_refs(new_rev, tbranch)
 
            if len(missed_refs) > 0:
                print "Could not update %s, you need to merge %s." % (refname, tbranch)
                exit(1)
 
 
if __name__ == '__main__':
    main()