tools/hook-scripts/svnperms.py

   1 #!/usr/bin/env python
   2
   3 # $HeadURL$
   4 # $LastChangedDate$
   5 # $LastChangedBy$
   6 # $LastChangedRevision$
   7
   8 import commands
   9 import sys, os
  10 import getopt
  11 try:
  12     my_getopt = getopt.gnu_getopt
  13 except AttributeError:
  14     my_getopt = getopt.getopt
  15 import re
  16
  17 __author__ = "Gustavo Niemeyer <gustavo@niemeyer.net>"
  18
  19 class Error(Exception): pass
  20
  21 SECTION = re.compile(r'\[([^]]+?)(?:\s+extends\s+([^]]+))?\]')
  22 OPTION = re.compile(r'(\S+)\s*=\s*(.*)$')
  23
  24 class Config:
  25     def __init__(self, filename):
  26         # Options are stored in __sections_list like this:
  27         # [(sectname, [(optname, optval), ...]), ...]
  28         self._sections_list = []
  29         self._sections_dict = {}
  30         self._read(filename)
  31
  32     def _read(self, filename):
  33         # Use the same logic as in ConfigParser.__read()
  34         file = open(filename)
  35         cursectdict = None
  36         optname = None
  37         lineno = 0
  38         for line in file.xreadlines():
  39             lineno = lineno + 1
  40             if line.isspace() or line[0] == '#':
  41                 continue
  42             if line[0].isspace() and cursectdict is not None and optname:
  43                 value = line.strip()
  44                 cursectdict[optname] = "%s %s" % (cursectdict[optname], value)
  45                 cursectlist[-1][1] = "%s %s" % (cursectlist[-1][1], value)
  46             else:
  47                 m = SECTION.match(line)
  48                 if m:
  49                     sectname = m.group(1)
  50                     parentsectname = m.group(2)
  51                     if parentsectname is None:
  52                         # No parent section defined, so start a new section
  53                         cursectdict = self._sections_dict.setdefault \
  54                             (sectname, {})
  55                         cursectlist = []
  56                     else:
  57                         # Copy the parent section into the new section
  58                         parentsectdict = self._sections_dict.get \
  59                             (parentsectname, {})
  60                         cursectdict = self._sections_dict.setdefault \
  61                             (sectname, parentsectdict.copy())
  62                         cursectlist = self.walk(parentsectname)
  63                     self._sections_list.append((sectname, cursectlist))
  64                     optname = None
  65                 elif cursectdict is None:
  66                     raise Error, "%s:%d: no section header" % \
  67                                  (filename, lineno)
  68                 else:
  69                     m = OPTION.match(line)
  70                     if m:
  71                         optname, optval = m.groups()
  72                         optval = optval.strip()
  73                         cursectdict[optname] = optval
  74                         cursectlist.append([optname, optval])
  75                     else:
  76                         raise Error, "%s:%d: parsing error" % \
  77                                      (filename, lineno)
  78
  79     def sections(self):
  80         return self._sections_dict.keys()
  81
  82     def options(self, section):
  83         return self._sections_dict.get(section, {}).keys()
  84
  85     def get(self, section, option, default=None):
  86         return self._sections_dict.get(option, default)
  87
  88     def walk(self, section, option=None):
  89         ret = []
  90         for sectname, options in self._sections_list:
  91             if sectname == section:
  92                 for optname, value in options:
  93                     if not option or optname == option:
  94                         ret.append((optname, value))
  95         return ret
  96
  97
  98 class Permission:
  99     def __init__(self):
 100         self._group = {}
 101         self._permlist = []
 102
 103     def parse_groups(self, groupsiter):
 104         for option, value in groupsiter:
 105             self._group[option] = value.split()
 106
 107     def parse_perms(self, permsiter):
 108         for option, value in permsiter:
 109             # Paths never start with /, so remove it if provided
 110             if option[0] == "/":
 111                 option = option[1:]
 112             pattern = re.compile("^%s$" % option)
 113             for entry in value.split():
 114                 openpar, closepar = entry.find("("), entry.find(")")
 115                 groupsusers = entry[:openpar].split(",")
 116                 perms = entry[openpar+1:closepar].split(",")
 117                 users = []
 118                 for groupuser in groupsusers:
 119                     if groupuser[0] == "@":
 120                         try:
 121                             users.extend(self._group[groupuser[1:]])
 122                         except KeyError:
 123                             raise Error, "group '%s' not found" % \
 124                                          groupuser[1:]
 125                     else:
 126                         users.append(groupuser)
 127                 self._permlist.append((pattern, users, perms))
 128
 129     def get(self, user, path):
 130         ret = []
 131         for pattern, users, perms in self._permlist:
 132             if pattern.match(path) and (user in users or "*" in users):
 133                 ret = perms
 134         return ret
 135
 136 class SVNLook:
 137     def __init__(self, repospath, txn=None, rev=None):
 138         self.repospath = repospath
 139         self.txn = txn
 140         self.rev = rev
 141
 142     def _execcmd(self, *cmd, **kwargs):
 143         cmdstr = " ".join(cmd)
 144         status, output = commands.getstatusoutput(cmdstr)
 145         if status != 0:
 146             sys.stderr.write(cmdstr)
 147             sys.stderr.write("\n")
 148             sys.stderr.write(output)
 149             raise Error, "command failed: %s\n%s" % (cmdstr, output)
 150         return status, output
 151
 152     def _execsvnlook(self, cmd, *args, **kwargs):
 153         execcmd_args = ["svnlook", cmd, self.repospath]
 154         self._add_txnrev(execcmd_args, kwargs)
 155         execcmd_args += args
 156         execcmd_kwargs = {}
 157         keywords = ["show", "noerror"]
 158         for key in keywords:
 159             if kwargs.has_key(key):
 160                 execcmd_kwargs[key] = kwargs[key]
 161         return self._execcmd(*execcmd_args, **execcmd_kwargs)
 162
 163     def _add_txnrev(self, cmd_args, received_kwargs):
 164         if received_kwargs.has_key("txn"):
 165             txn = received_kwargs.get("txn")
 166             if txn is not None:
 167                 cmd_args += ["-t", txn]
 168         elif self.txn is not None:
 169             cmd_args += ["-t", self.txn]
 170         if received_kwargs.has_key("rev"):
 171             rev = received_kwargs.get("rev")
 172             if rev is not None:
 173                 cmd_args += ["-r", rev]
 174         elif self.rev is not None:
 175             cmd_args += ["-r", self.rev]
 176
 177     def changed(self, **kwargs):
 178         status, output = self._execsvnlook("changed", **kwargs)
 179         if status != 0:
 180             return None
 181         changes = []
 182         for line in output.splitlines():
 183             line = line.rstrip()
 184             if not line: continue
 185             entry = [None, None, None]
 186             changedata, changeprop, path = None, None, None
 187             if line[0] != "_":
 188                 changedata = line[0]
 189             if line[1] != " ":
 190                 changeprop = line[1]
 191             path = line[4:]
 192             changes.append((changedata, changeprop, path))
 193         return changes
 194
 195     def author(self, **kwargs):
 196         status, output = self._execsvnlook("author", **kwargs)
 197         if status != 0:
 198             return None
 199         return output.strip()
 200
 201
 202 def check_perms(filename, section, repos, txn=None, rev=None, author=None):
 203     svnlook = SVNLook(repos, txn=txn, rev=rev)
 204     if author is None:
 205         author = svnlook.author()
 206     changes = svnlook.changed()
 207     try:
 208         config = Config(filename)
 209     except IOError:
 210         raise Error, "can't read config file "+filename
 211     if not section in config.sections():
 212         raise Error, "section '%s' not found in config file" % section
 213     perm = Permission()
 214     perm.parse_groups(config.walk("groups"))
 215     perm.parse_groups(config.walk(section+" groups"))
 216     perm.parse_perms(config.walk(section))
 217     permerrors = []
 218     for changedata, changeprop, path in changes:
 219         pathperms = perm.get(author, path)
 220         if changedata == "A" and "add" not in pathperms:
 221             permerrors.append("you can't add "+path)
 222         elif changedata == "U" and "update" not in pathperms:
 223             permerrors.append("you can't update "+path)
 224         elif changedata == "D" and "remove" not in pathperms:
 225             permerrors.append("you can't remove "+path)
 226         elif changeprop == "U" and "update" not in pathperms:
 227             permerrors.append("you can't update properties of "+path)
 228         #else:
 229         #    print "cdata=%s cprop=%s path=%s perms=%s" % \
 230         #          (str(changedata), str(changeprop), path, str(pathperms))
 231     if permerrors:
 232         permerrors.insert(0, "you don't have enough permissions for "
 233                              "this transaction:")
 234         raise Error, "\n".join(permerrors)
 235
 236
 237 # Command:
 238
 239 USAGE = """\
 240 Usage: svnperms.py OPTIONS
 241
 242 Options:
 243     -r PATH    Use repository at PATH to check transactions
 244     -t TXN     Query transaction TXN for commit information
 245     -f PATH    Use PATH as configuration file (default is repository
 246                path + /conf/svnperms.conf)
 247     -s NAME    Use section NAME as permission section (default is
 248                repository name, extracted from repository path)
 249     -R REV     Query revision REV for commit information (for tests)
 250     -A AUTHOR  Check commit as if AUTHOR had commited it (for tests)
 251     -h         Show this message
 252 """
 253
 254 class MissingArgumentsException(Exception):
 255     "Thrown when required arguments are missing."
 256     pass
 257
 258 def parse_options():
 259     try:
 260         opts, args = my_getopt(sys.argv[1:], "f:s:r:t:R:A:h", ["help"])
 261     except getopt.GetoptError, e:
 262         raise Error, e.msg
 263     class Options: pass
 264     obj = Options()
 265     obj.filename = None
 266     obj.section = None
 267     obj.repository = None
 268     obj.transaction = None
 269     obj.revision = None
 270     obj.author = None
 271     for opt, val in opts:
 272         if opt == "-f":
 273             obj.filename = val
 274         elif opt == "-s":
 275             obj.section = val
 276         elif opt == "-r":
 277             obj.repository = val
 278         elif opt == "-t":
 279             obj.transaction = val
 280         elif opt == "-R":
 281             obj.revision = val
 282         elif opt == "-A":
 283             obj.author = val
 284         elif opt in ["-h", "--help"]:
 285             sys.stdout.write(USAGE)
 286             sys.exit(0)
 287     missingopts = []
 288     if not obj.repository:
 289         missingopts.append("repository")
 290     if not (obj.transaction or obj.revision):
 291         missingopts.append("either transaction or a revision")
 292     if missingopts:
 293         raise MissingArgumentsException, \
 294               "missing required option(s): " + ", ".join(missingopts)
 295     obj.repository = os.path.abspath(obj.repository)
 296     if obj.filename is None:
 297         obj.filename = os.path.join(obj.repository, "conf", "svnperms.conf")
 298     if obj.section is None:
 299         obj.section = os.path.basename(obj.repository)
 300     if not (os.path.isdir(obj.repository) and
 301             os.path.isdir(os.path.join(obj.repository, "db")) and
 302             os.path.isdir(os.path.join(obj.repository, "hooks")) and
 303             os.path.isfile(os.path.join(obj.repository, "format"))):
 304         raise Error, "path '%s' doesn't look like a repository" % \
 305                      obj.repository
 306
 307     return obj
 308
 309 def main():
 310     try:
 311         opts = parse_options()
 312         check_perms(opts.filename, opts.section,
 313                     opts.repository, opts.transaction, opts.revision,
 314                     opts.author)
 315     except MissingArgumentsException, e:
 316         sys.stderr.write("%s\n" % str(e))
 317         sys.stderr.write(USAGE)
 318         sys.exit(1)
 319     except Error, e:
 320         sys.stderr.write("error: %s\n" % str(e))
 321         sys.exit(1)
 322
 323 if __name__ == "__main__":
 324     main()
 325
 326 # vim:et:ts=4:sw=4