| #!/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$ | |
| # $LastChangedDate$ | |
| # $LastChangedBy$ | |
| # $LastChangedRevision$ | |
| 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 |