Ticket #1366: authz_groups.diff

trac/util.py

@@ -95 +95 @@
             break
     return text
 
+def strip(text, skip):
+    """Python2.1 doesn't support custom skip characters"""
+    return lstrip(rstrip(text, skip), skip)
+
 def to_utf8(text, charset='iso-8859-15'):
     """Convert a string to utf-8, assume the encoding is either utf-8 or latin1"""
     try:

trac/versioncontrol/cache.py

@@ -22 +22 @@
 from __future__ import generators
 
 from trac.util import TracError
-from trac.versioncontrol import Changeset, Node, Repository
+from trac.versioncontrol import Changeset, Node, Repository, Authorizer
 
 
 _kindmap = {'D': Node.DIRECTORY, 'F': Node.FILE}
@@ -52 +52 @@
         self.log.debug("Checking whether sync with repository is needed")
         youngest_stored = self.repos.get_youngest_rev_in_cache(self.db)
         if youngest_stored != str(self.repos.youngest_rev):
+            authz = self.repos.authz
+            self.repos.authz = Authorizer() # remove permission checking
+
             kindmap = dict(zip(_kindmap.values(), _kindmap.keys()))
             actionmap = dict(zip(_actionmap.values(), _actionmap.keys()))
             self.log.info("Syncing with repository (%s to %s)"
@@ -80 +83 @@
                                    base_path, base_rev))
                 current_rev = self.repos.next_rev(current_rev)
             self.db.commit()
+            self.repos.authz = authz # restore permission checking
 
     def get_node(self, path, rev=None):
         return self.repos.get_node(path, rev)

trac/versioncontrol/svn_authz.py

@@ -19 +19 @@
 #
 # Author: Francois Harvey <fharvey@securiweb.net>
 
+from __future__ import generators
 from trac.versioncontrol import Authorizer
 
+def SubversionAuthorizer(env, authname):
+    authz_file = env.get_config('trac','authz_file')    
+    if not authz_file:
+        return Authorizer()
 
-class SubversionAuthorizer(Authorizer):
+    module_name = env.get_config('trac','authz_module_name','')
+    db = env.get_db_cnx()
+    return RealSubversionAuthorizer(db, authname, module_name, authz_file)
 
+def parent_iter(path):
+    from trac.util import strip
+    path = strip(path, '/')
+    if path:
+        path = '/' + path + '/'
+    else:
+        path = '/'
+
+    while 1:
+        yield path
+        if path == '/':
+            raise StopIteration()
+        path = path[:-1]
+        yield path
+        idx = path.rfind('/')
+        path = path[:idx + 1]
+
+class RealSubversionAuthorizer(Authorizer):
+
     auth_name = ''
     module_name = ''
     conf_authz = None
-    authz_file = ''
 
-    def __init__(self, env, authname):
-        self.auth_name = authname
-        
-        if env.get_config('trac','authz_module_name','') == '':
-            self.module_name = ''
+    def __init__(self, db, auth_name, module_name, cfg_file, cfg_fp=None):
+        self.db = db
+        self.auth_name = auth_name
+        self.module_name = module_name
+                                
+        from ConfigParser import ConfigParser
+        self.conf_authz = ConfigParser()
+        if cfg_fp:
+            self.conf_authz.readfp(cfg_fp, cfg_file)
+        elif cfg_file:
+            self.conf_authz.read(cfg_file)
+
+        self.groups = self.get_groups()
+
+    def get_groups(self):
+        if not self.conf_authz.has_section('groups'):
+            return []
         else:
-            self.module_name = env.get_config('trac','authz_module_name') + ':'
-                                 
-        self.autz_file = env.get_config('trac','authz_file')    
-        if env.get_config('trac','authz_file'):
-            from ConfigParser import ConfigParser
-            self.conf_authz = ConfigParser()
-            self.conf_authz.read(self.autz_file)
+            return [group for group in self.conf_authz.options('groups') \
+                          if self.in_group(group)]
 
-        self.db = env.get_db_cnx()
-
-    def group_contains_user(self, group_name, user_name):
-        if self.conf_authz.has_section('groups'):
-            if self.conf_authz.has_option('groups', group_name):
-                users_list = self.conf_authz.get('groups', group_name).split(',')
-                return users_list.has_key(user_name)
+    def in_group(self, group):
+        for user in self.conf_authz.get('groups', group).split(','):
+            if self.auth_name == user.strip():
+                return 1
         return 0
 
     def has_permission(self, path):
-        acc = ''
+        if path is None:
+            return 1
 
-        if path != None and self.conf_authz != None:
-            if self.conf_authz.has_section(self.module_name + '/') and \
-                   self.conf_authz.has_option(self.module_name  + '/',
-                                              self.auth_name):
-                acc = self.conf_authz.get(self.module_name + '/',self.auth_name)
-            elif self.conf_authz.has_section(self.module_name + '/') and \
-                 self.conf_authz.has_option(self.module_name  + '/', '*'):
-                     acc = self.conf_authz.get(self.module_name + '/','*')
-            path_comb = ''
-            for path_elem in path.split('/'):
-                if path_elem != '':
-                    path_comb = self.module_name + path_comb + '/' + path_elem
-                    if self.conf_authz.has_section(path_comb) and \
-                           self.conf_authz.has_option(path_comb, self.auth_name):
-                        acc =  self.conf_authz.get(path_comb, self.auth_name)
-                    elif self.conf_authz.has_section(path_comb) and \
-                            self.conf_authz.has_option(path_comb, '*'):
-                        acc =  self.conf_authz.get(path_comb, '*')
-        else:
-            acc = 'r'
-        return acc
+        for p in parent_iter(path):
+            if self.module_name:
+                for perm in self.get_section(self.module_name + ':' + p):
+                    if perm is not None:
+                        return perm
+            for perm in self.get_section(p):
+                if perm is not None:
+                    return perm
 
+        return 0
+
+    def get_section(self, section):
+        if not self.conf_authz.has_section(section):
+            return
+
+        yield self.get_permission(section, self.auth_name)
+
+        group_perm = None
+        for g in self.groups:
+            p = self.get_permission(section, '@' + g)
+            if p is not None:
+                group_perm = p
+
+            if group_perm:
+                yield 1
+
+        yield group_perm
+
+        yield self.get_permission(section, '*')
+
+    def get_permission(self, section, subject):
+        if self.conf_authz.has_option(section, subject):
+            return 'r' in self.conf_authz.get(section, subject)
+        return None
+
     def has_permission_for_changeset(self, rev):
         cursor = self.db.cursor()
         cursor.execute("SELECT path FROM node_change WHERE rev=%s", (rev,))

trac/versioncontrol/tests/__init__.py

@@ -1 +1 @@
 import unittest
 
-from trac.versioncontrol.tests import cache, diff
+from trac.versioncontrol.tests import cache, diff, svn_authz
 
 def suite():
 
     suite = unittest.TestSuite()
     suite.addTest(cache.suite())
     suite.addTest(diff.suite())
+    suite.addTest(svn_authz.suite())
     return suite
 
 if __name__ == '__main__':

trac/versioncontrol/tests/svn_authz.py

@@ +1 @@
+from trac.versioncontrol import svn_authz
+
+import unittest
+
+def suite():
+    try:
+        from doctest import DocFileSuite
+        return DocFileSuite('svn_authz.txt')
+    except ImportError:
+        import sys
+        print>>sys.stderr, "WARNING: DocTestSuite required to run these tests"
+    return unittest.TestSuite()
+
+if __name__ == '__main__':
+    runner = unittest.TextTestRunner()
+    runner.run(suite())
+

trac/versioncontrol/tests/svn_authz.txt

@@ +1 @@
+Subversion Authz File Permissions
+=================================
+
+Setup code
+----------
+We'll use the ``make_auth`` method to create Authorizer objects
+for testing the use of authz files.  ``make_auth`` takes a module name
+and a string for the authz configuration contents.
+
+>>> from trac.versioncontrol.svn_authz import RealSubversionAuthorizer
+>>> from StringIO import StringIO
+>>> make_auth = lambda mod, cfg: RealSubversionAuthorizer(None,
+...                   'user', mod, None, StringIO(cfg))
+
+
+Simple operation
+----------------
+Returns 1 if no path is given:
+    >>> int(make_auth('', '').has_permission(None))
+    1
+
+By default read permission is not enabled:
+    >>> int(make_auth('', '').has_permission('/'))
+    0
+
+Read and Write Permissions
+----------------------
+Trac is only concerned about read permissions.
+    >>> a = make_auth('', '''
+    ... [/readonly]
+    ... user = r
+    ... [/writeonly]
+    ... user = w
+    ... [/readwrite]
+    ... user = rw
+    ... [/empty]
+    ... user = 
+    ... ''')
+
+Permissions of 'r' or 'rw' will allow access:
+    >>> int(a.has_permission('/readonly'))
+    1
+    >>> int(a.has_permission('/readwrite'))
+    1
+
+If only 'w' permission is given, Trac does not allow access:
+    >>> int(a.has_permission('/writeonly'))
+    0
+
+And an empty permission does not give access:
+    >>> int(a.has_permission('/empty'))
+    0
+
+Trailing Slashes
+----------------
+Checks all combinations of trailing slashes in the configuration
+or in the path parameter:
+    >>> a = make_auth('', '''
+    ... [/a]
+    ... user = r
+    ... [/b/]
+    ... user = r
+    ... ''')
+    >>> int(a.has_permission('/a'))
+    1
+    >>> int(a.has_permission('/a/'))
+    1
+    >>> int(a.has_permission('/b'))
+    1
+    >>> int(a.has_permission('/b/'))
+    1
+
+
+Module Usage
+------------
+If a module name is specified, the rules used are specific to the module.
+    >>> a = make_auth('module', '''
+    ... [module:/a]
+    ... user = r
+    ... [other:/b]
+    ... user = r
+    ... ''')
+    >>> int(a.has_permission('/a'))
+    1
+    >>> int(a.has_permission('/b'))
+    0
+
+If a module is specified, but the configuration contains a non-module
+path, the non-module path can still apply:
+    >>> int(make_auth('module', '''
+    ... [/a]
+    ... user = r
+    ... ''').has_permission('/a'))
+    1
+
+However, the module-specific rule will take precedence if both exist:
+    >>> int(make_auth('module', '''
+    ... [module:/a]
+    ... user = 
+    ... [/a]
+    ... user = r
+    ... ''').has_permission('/a'))
+    0
+
+
+Groups and Wildcards
+--------------------
+Authz provides a * wildcard for matching any user:
+    >>> int(make_auth('', '''
+    ... [/a]
+    ... * = r
+    ... ''').has_permission('/a'))
+    1
+
+Groups are specified in a separate section and used with an @ prefix:
+    >>> int(make_auth('', '''
+    ... [groups]
+    ... grp = user
+    ... [/a]
+    ... @grp = r
+    ... ''').has_permission('/a'))
+    1
+
+If more than one group matches at the specific path, access is granted
+if any of the group rules allow access.
+    >>> a = make_auth('', '''
+    ... [groups]
+    ... grp1 = user
+    ... grp2 = user
+    ... [/a]
+    ... @grp1 = r
+    ... @grp2 = 
+    ... [/b]
+    ... @grp1 = 
+    ... @grp2 = r
+    ... ''')
+    >>> int(a.has_permission('/a'))
+    1
+    >>> int(a.has_permission('/b'))
+    1
+
+
+Precedence
+----------
+Precedence is user, group, then *:
+    >>> a = make_auth('', '''
+    ... [groups]
+    ... grp = user
+    ... [/a]
+    ... @grp = r
+    ... user = 
+    ... [/b]
+    ... * = r
+    ... @grp = 
+    ... ''')
+
+User specific permission overrides the group permission:
+    >>> int(a.has_permission('/a'))
+    0
+
+And group permission overrides the * permission:
+    >>> int(a.has_permission('/b'))
+    0
+
+The most specific matching path takes precedence:
+    >>> a = make_auth('', '''
+    ... [/]
+    ... * = r
+    ... [/b]
+    ... user = 
+    ... ''')
+    >>> int(a.has_permission('/'))
+    1
+    >>> int(a.has_permission('/a'))
+    1
+    >>> int(a.has_permission('/b'))
+    0
+
+Changeset Permissions
+---------------------
+A test should go here for the changeset permissions.