path: root/tools/hook-scripts/svnperms.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
#
#

# $HeadURL: https://svn.apache.org/repos/asf/subversion/branches/1.13.x/tools/hook-scripts/svnperms.py $
# $LastChangedDate: 2016-04-30 08:16:53 +0000 (Sat, 30 Apr 2016) $
# $LastChangedBy: stefan2 $
# $LastChangedRevision: 1741723 $

import sys, os
import getopt
import shlex

try:
  # Python >=3.0
  from subprocess import getstatusoutput as subprocess_getstatusoutput
except ImportError:
  # Python <3.0
  from commands import getstatusoutput as subprocess_getstatusoutput
try:
    my_getopt = getopt.gnu_getopt
except AttributeError:
    my_getopt = getopt.getopt
import re

__author__ = "Gustavo Niemeyer <gustavo@niemeyer.net>"

class Error(Exception): pass

SECTION = re.compile(r'\[([^]]+?)(?:\s+extends\s+([^]]+))?\]')
OPTION = re.compile(r'(\S+)\s*=\s*(.*)$')

class Config:
    def __init__(self, filename):
        # Options are stored in __sections_list like this:
        # [(sectname, [(optname, optval), ...]), ...]
        self._sections_list = []
        self._sections_dict = {}
        self._read(filename)

    def _read(self, filename):
        # Use the same logic as in ConfigParser.__read()
        file = open(filename)
        cursectdict = None
        optname = None
        lineno = 0
        for line in file:
            lineno = lineno + 1
            if line.isspace() or line[0] == '#':
                continue
            if line[0].isspace() and cursectdict is not None and optname:
                value = line.strip()
                cursectdict[optname] = "%s %s" % (cursectdict[optname], value)
                cursectlist[-1][1] = "%s %s" % (cursectlist[-1][1], value)
            else:
                m = SECTION.match(line)
                if m:
                    sectname = m.group(1)
                    parentsectname = m.group(2)
                    if parentsectname is None:
                        # No parent section defined, so start a new section
                        cursectdict = self._sections_dict.setdefault \
                            (sectname, {})
                        cursectlist = []
                    else:
                        # Copy the parent section into the new section
                        parentsectdict = self._sections_dict.get \
                            (parentsectname, {})
                        cursectdict = self._sections_dict.setdefault \
                            (sectname, parentsectdict.copy())
                        cursectlist = self.walk(parentsectname)
                    self._sections_list.append((sectname, cursectlist))
                    optname = None
                elif cursectdict is None:
                    raise Error("%s:%d: no section header" % \
                                 (filename, lineno))
                else:
                    m = OPTION.match(line)
                    if m:
                        optname, optval = m.groups()
                        optval = optval.strip()
                        cursectdict[optname] = optval
                        cursectlist.append([optname, optval])
                    else:
                        raise Error("%s:%d: parsing error" % \
                                     (filename, lineno))

    def sections(self):
        return list(self._sections_dict.keys())

    def options(self, section):
        return list(self._sections_dict.get(section, {}).keys())

    def get(self, section, option, default=None):
        return self._sections_dict.get(option, default)

    def walk(self, section, option=None):
        ret = []
        for sectname, options in self._sections_list:
            if sectname == section:
                for optname, value in options:
                    if not option or optname == option:
                        ret.append((optname, value))
        return ret


class Permission:
    def __init__(self):
        self._group = {}
        self._permlist = []

    def parse_groups(self, groupsiter):
        for option, value in groupsiter:
            groupusers = []
            for token in shlex.split(value):
                # expand nested groups in place; no forward decls
                if token[0] == "@":
                    try:
                        groupusers.extend(self._group[token[1:]])
                    except KeyError:
                        raise Error, "group '%s' not found" % token[1:]
                else:
                    groupusers.append(token)
            self._group[option] = groupusers

    def parse_perms(self, permsiter):
        for option, value in permsiter:
            # Paths never start with /, so remove it if provided
            if option[0] == "/":
                option = option[1:]
            pattern = re.compile("^%s$" % option)
            for entry in value.split():
                openpar, closepar = entry.find("("), entry.find(")")
                groupsusers = entry[:openpar].split(",")
                perms = entry[openpar+1:closepar].split(",")
                users = []
                for groupuser in groupsusers:
                    if groupuser[0] == "@":
                        try:
                            users.extend(self._group[groupuser[1:]])
                        except KeyError:
                            raise Error("group '%s' not found" % \
                                         groupuser[1:])
                    else:
                        users.append(groupuser)
                self._permlist.append((pattern, users, perms))

    def get(self, user, path):
        ret = []
        for pattern, users, perms in self._permlist:
            if pattern.match(path) and (user in users or "*" in users):
                ret = perms
        return ret

class SVNLook:
    def __init__(self, repospath, txn=None, rev=None):
        self.repospath = repospath
        self.txn = txn
        self.rev = rev

    def _execcmd(self, *cmd, **kwargs):
        cmdstr = " ".join(cmd)
        status, output = subprocess_getstatusoutput(cmdstr)
        if status != 0:
            sys.stderr.write(cmdstr)
            sys.stderr.write("\n")
            sys.stderr.write(output)
            raise Error("command failed: %s\n%s" % (cmdstr, output))
        return status, output

    def _execsvnlook(self, cmd, *args, **kwargs):
        execcmd_args = ["svnlook", cmd, self.repospath]
        self._add_txnrev(execcmd_args, kwargs)
        execcmd_args += args
        execcmd_kwargs = {}
        keywords = ["show", "noerror"]
        for key in keywords:
            if key in kwargs:
                execcmd_kwargs[key] = kwargs[key]
        return self._execcmd(*execcmd_args, **execcmd_kwargs)

    def _add_txnrev(self, cmd_args, received_kwargs):
        if "txn" in received_kwargs:
            txn = received_kwargs.get("txn")
            if txn is not None:
                cmd_args += ["-t", txn]
        elif self.txn is not None:
            cmd_args += ["-t", self.txn]
        if "rev" in received_kwargs:
            rev = received_kwargs.get("rev")
            if rev is not None:
                cmd_args += ["-r", rev]
        elif self.rev is not None:
            cmd_args += ["-r", self.rev]

    def changed(self, **kwargs):
        status, output = self._execsvnlook("changed", **kwargs)
        if status != 0:
            return None
        changes = []
        for line in output.splitlines():
            line = line.rstrip()
            if not line: continue
            entry = [None, None, None]
            changedata, changeprop, path = None, None, None
            if line[0] != "_":
                changedata = line[0]
            if line[1] != " ":
                changeprop = line[1]
            path = line[4:]
            changes.append((changedata, changeprop, path))
        return changes

    def author(self, **kwargs):
        status, output = self._execsvnlook("author", **kwargs)
        if status != 0:
            return None
        return output.strip()


def check_perms(filename, section, repos, txn=None, rev=None, author=None):
    svnlook = SVNLook(repos, txn=txn, rev=rev)
    if author is None:
        author = svnlook.author()
    changes = svnlook.changed()
    try:
        config = Config(filename)
    except IOError:
        raise Error("can't read config file "+filename)
    if not section in config.sections():
        raise Error("section '%s' not found in config file" % section)
    perm = Permission()
    perm.parse_groups(config.walk("groups"))
    perm.parse_groups(config.walk(section+" groups"))
    perm.parse_perms(config.walk(section))
    permerrors = []
    for changedata, changeprop, path in changes:
        pathperms = perm.get(author, path)
        if changedata == "A" and "add" not in pathperms:
            permerrors.append("you can't add "+path)
        elif changedata == "U" and "update" not in pathperms:
            permerrors.append("you can't update "+path)
        elif changedata == "D" and "remove" not in pathperms:
            permerrors.append("you can't remove "+path)
        elif changeprop == "U" and "update" not in pathperms:
            permerrors.append("you can't update properties of "+path)
        #else:
        #    print "cdata=%s cprop=%s path=%s perms=%s" % \
        #          (str(changedata), str(changeprop), path, str(pathperms))
    if permerrors:
        permerrors.insert(0, "you don't have enough permissions for "
                             "this transaction:")
        raise Error("\n".join(permerrors))


# Command:

USAGE = """\
Usage: svnperms.py OPTIONS

Options:
    -r PATH    Use repository at PATH to check transactions
    -t TXN     Query transaction TXN for commit information
    -f PATH    Use PATH as configuration file (default is repository
               path + /conf/svnperms.conf)
    -s NAME    Use section NAME as permission section (default is
               repository name, extracted from repository path)
    -R REV     Query revision REV for commit information (for tests)
    -A AUTHOR  Check commit as if AUTHOR had committed it (for tests)
    -h         Show this message
"""

class MissingArgumentsException(Exception):
    "Thrown when required arguments are missing."
    pass

def parse_options():
    try:
        opts, args = my_getopt(sys.argv[1:], "f:s:r:t:R:A:h", ["help"])
    except getopt.GetoptError as e:
        raise Error(e.msg)
    class Options: pass
    obj = Options()
    obj.filename = None
    obj.section = None
    obj.repository = None
    obj.transaction = None
    obj.revision = None
    obj.author = None
    for opt, val in opts:
        if opt == "-f":
            obj.filename = val
        elif opt == "-s":
            obj.section = val
        elif opt == "-r":
            obj.repository = val
        elif opt == "-t":
            obj.transaction = val
        elif opt == "-R":
            obj.revision = val
        elif opt == "-A":
            obj.author = val
        elif opt in ["-h", "--help"]:
            sys.stdout.write(USAGE)
            sys.exit(0)
    missingopts = []
    if not obj.repository:
        missingopts.append("repository")
    if not (obj.transaction or obj.revision):
        missingopts.append("either transaction or a revision")
    if missingopts:
        raise MissingArgumentsException("missing required option(s): " + ", ".join(missingopts))
    obj.repository = os.path.abspath(obj.repository)
    if obj.filename is None:
        obj.filename = os.path.join(obj.repository, "conf", "svnperms.conf")
    if obj.section is None:
        obj.section = os.path.basename(obj.repository)
    if not (os.path.isdir(obj.repository) and
            os.path.isdir(os.path.join(obj.repository, "db")) and
            os.path.isdir(os.path.join(obj.repository, "hooks")) and
            os.path.isfile(os.path.join(obj.repository, "format"))):
        raise Error("path '%s' doesn't look like a repository" % \
                     obj.repository)

    return obj

def main():
    try:
        opts = parse_options()
        check_perms(opts.filename, opts.section,
                    opts.repository, opts.transaction, opts.revision,
                    opts.author)
    except MissingArgumentsException as e:
        sys.stderr.write("%s\n" % str(e))
        sys.stderr.write(USAGE)
        sys.exit(1)
    except Error as e:
        sys.stderr.write("error: %s\n" % str(e))
        sys.exit(1)

if __name__ == "__main__":
    main()

# vim:et:ts=4:sw=4