On Thu, 2015-08-20 at 13:29 +0930, William Brown wrote:
> On Wed, 2015-08-19 at 11:35 -0400, Mark Reynolds wrote:
> > Hi William,
> >
> > Still reviewing your patch, but I noticed that you are missing
> > "authmethod" from your ACI class keys, as well as "ip", "dns",
> > "userattr", and perhaps others.
>
>
> Quite likely I have missed some. I'll add these and review the documents you
> referenced to see if there are more.
Here is an updated patch for review. It's still not complete, but I welcome
comments and suggestions.
It can now parse all the bind rule types as you previously mentioned.
It can also parse an aci such as:
(targetattr !="cn || sn || uid")(targetfilter ="(ou=Product
Development)")(version 3.0;acl "Engineering Group Permissions";allow
(write)(groupdn = "ldap:///cn=PD Managers,ou=groups,dc=example,dc=com");)
Turn it into a set of dictionaries:
---------
dayofweek : []
allow : [{'values': ['write']}]
dns : []
target : []
targetattr : [{'values': ['cn', 'sn', 'uid'], 'equal': False}]
targattrfilters : []
authmethod : []
userattr : []
userdn : []
acl : [{'values': ['Engineering Group Permissions']}]
targetattrfilters : []
version 3.0; : []
targetfilter : [{'values': ['(ou=Product Development)'], 'equal': True}]
targetscope : []
ip : []
roledn : []
timeofday : []
groupdn : [{'values': ['cn=PD Managers,ou=groups,dc=example,dc=com'], 'equal':
True}]
Then use this to re-create acis:
(targetfilter ="(ou=Product Development)")(targetattr !="cn || sn ||
uid")(version 3.0; acl "Engineering Group Permissions";allow (write)(groupdn
="ldap:///cn=PD Managers,ou=groups,dc=example,dc=com");)
This means we can edit the contents of the EntryAci, then potentially save this
back to LDAP.
Right now, complex bind rules won't work (IE anything with and, or, not) and
certain ldap urls in userdn, roledn, and groupdn with filters in them may not
parse correctly.
I still need to add more complex tests also.
Finally, there are likely some code clean ups and clarity improvements to make.
I can see this being useful for:
* Programmatic updates of acis in systems.
* Ability to automatically invert acis (turn != into = targetattr acis).
* Be able to write better unit tests regarding aci functionality.
* Easily able to discover acis.
* Autogenerated tests can take advantage of this, so we could make randomised /
complex acis easily.
Sincerely,
--
William Brown <william@xxxxxxxxxxxxxxxx>
From c55ba77e1301c9c1883201c3c702d0221a01cfa9 Mon Sep 17 00:00:00 2001
From: William Brown <wililam@xxxxxxxxxxxxxxx>
Date: Tue, 4 Aug 2015 13:30:24 +0930
Subject: [PATCH] Add aci parsing utilities, which will return an EntryAci. Not
ready for merge, review only. Can work with simple bind rules, but not
complex bind rules yet (IE and, or). Can rebuild acis from dictionary data so
that EntryAci objects can be edited and then saved back to ldap.
---
lib389/__init__.py | 23 ++++--
lib389/_entry.py | 198 ++++++++++++++++++++++++++++++++++++++++++++++++
lib389/aci.py | 29 +++++++
tests/aci_parse_test.py | 54 +++++++++++++
4 files changed, 297 insertions(+), 7 deletions(-)
create mode 100644 lib389/aci.py
create mode 100644 tests/aci_parse_test.py
diff --git a/lib389/__init__.py b/lib389/__init__.py
index 519f04a..8c34fd8 100644
--- a/lib389/__init__.py
+++ b/lib389/__init__.py
@@ -137,16 +137,22 @@ def wrapper(f, name):
# print data
if data:
if isinstance(data, tuple):
+ # Probably should do the same aci vs entry check here
return objtype, Entry(data)
elif isinstance(data, list):
# AD sends back these search references
-# if objtype == ldap.RES_SEARCH_RESULT and \
-# isinstance(data[-1],tuple) and \
-# not data[-1][0]:
-# print "Received search reference: "
-# pprint.pprint(data[-1][1])
-# data.pop() # remove the last non-entry element
-
+ # if objtype == ldap.RES_SEARCH_RESULT and \
+ # isinstance(data[-1],tuple) and \
+ # not data[-1][0]:
+ # print "Received search reference: "
+ # pprint.pprint(data[-1][1])
+ # data.pop() # remove the last non-entry element
+
+ for x in data:
+ print(x)
+ # So here we need to check if the data contains aci, if so
+ # we return an AciEntry instead.
+ # There may be a better way to achieve this however.....
return objtype, [Entry(x) for x in data]
else:
raise TypeError("unknown data type %s returned by result" %
@@ -159,6 +165,7 @@ def wrapper(f, name):
# We need to convert the Entry into the format used by
# python-ldap
ent = args[0]
+ # Make this an "or" AciEntry
if isinstance(ent, Entry):
return f(ent.dn, ent.toTupleList(), *args[2:])
else:
@@ -343,6 +350,7 @@ class DirSrv(SimpleLDAPObject):
from lib389.plugins import Plugins
from lib389.tasks import Tasks
from lib389.index import Index
+ from lib389.aci import Aci
self.agreement = Agreement(self)
self.replica = Replica(self)
@@ -355,6 +363,7 @@ class DirSrv(SimpleLDAPObject):
self.schema = Schema(self)
self.plugins = Plugins(self)
self.tasks = Tasks(self)
+ self.aci = Aci(self)
def __init__(self, verbose=False, timeout=10):
"""
diff --git a/lib389/_entry.py b/lib389/_entry.py
index b97b6e7..39cd5e0 100644
--- a/lib389/_entry.py
+++ b/lib389/_entry.py
@@ -212,3 +212,201 @@ class Entry(object):
except MissingEntryError:
log.exception("This entry should exist!")
raise
+
+ def getAcis(self):
+ if not self.hasAttr('aci'):
+ # There should be a better way to do this? Perhaps
+ # self search for the aci attr?
+ return []
+ self.acis = map(lambda a: EntryAci(self, a), self.getValues('aci'))
+ return self.acis
+
+class EntryAci(object):
+
+ # See https://access.redhat.com/documentation/en-US/Red_Hat_Directory_Server/10/html/Administration_Guide/Managing_Access_Control-Bind_Rules.html
+ # https://access.redhat.com/documentation/en-US/Red_Hat_Directory_Server/10/html/Administration_Guide/Managing_Access_Control-Creating_ACIs_Manually.html
+ # We seperate the keys into 3 groups, and one group that has overlap.
+ # This is so we can not only split the aci, but rebuild it from the dictionary
+ # at a later point in time.
+ # These are top level aci comoponent keys
+ _keys = [
+ 'targetscope',
+ 'targetattrfilters',
+ 'targattrfilters',
+ 'targetfilter',
+ 'targetattr',
+ 'target',
+ 'version 3.0;',
+ ]
+ # These are the keys which are seperated by ; in the version 3.0 stanza.
+ _v3keys = [
+ 'allow',
+ 'acl',
+ ]
+ # These are the keys which are used on the inside of a v3 allow statement
+ _v3innerkeys = [
+ 'roledn',
+ 'userattr',
+ 'ip',
+ 'dns',
+ 'dayofweek',
+ 'timeofday',
+ 'authmethod',
+ 'userdn',
+ 'groupdn',
+ ]
+ # These keys values are prefixed with ldap:///, so we need to know to re-prefix
+ # ldap:/// onto the value when we rebuild the aci
+ _urlkeys = ['target',
+ 'userdn',
+ 'groupdn',
+ 'roledn',
+ ]
+
+ def __init__(self, entry, rawaci):
+ self.entry = entry
+ self._rawaci = rawaci
+ self.acidata = self._parse_aci(self._rawaci)
+
+ def _format_term(self, key, value_dict):
+ rawaci = ''
+ if value_dict['equal']:
+ rawaci += '="'
+ else:
+ rawaci += '!="'
+ if key in self._urlkeys:
+ values = map(lambda x: 'ldap:///%s' % x, value_dict['values'])
+ else:
+ values = value_dict['values']
+ for value in values[:-1]:
+ rawaci += "%s || " % value
+ rawaci += values[-1]
+ rawaci += '"'
+ return rawaci
+
+ def getRawAci(self):
+ # Rebuild the aci from the .acidata.
+ rawaci = ''
+ # For each key in the outer segment
+ ## Add a (key = val);. Depending on key format val:
+ for key in self._keys:
+ for value_dict in self.acidata[key]:
+ rawaci += '(%s %s)' % (key, self._format_term(key, value_dict))
+ # Now create the v3.0 aci part
+ rawaci += "(version 3.0; "
+ # This could be neater ...
+ rawaci += 'acl "%s";' % self.acidata['acl'][0]['values'][0]
+ rawaci += 'allow ('
+ for value in self.acidata['allow'][0]['values'][:-1]:
+ rawaci += '%s, ' % value
+ rawaci += '%s)' % self.acidata['allow'][0]['values'][-1]
+ # There may be a neater way to do this to handle the "and" case
+ for key in self._v3innerkeys:
+ if len(self.acidata[key]) == 0:
+ continue
+ rawaci += '(%s ' % key
+ for value_dict in self.acidata[key][:-1]:
+ rawaci += '%s and ' % self._format_term(key, value_dict)
+ rawaci += '%s)' % self._format_term(key, self.acidata[key][-1])
+ rawaci += ";)"
+ return rawaci
+
+ def _find_terms(self, aci):
+ lbr_list = []
+ rbr_list = []
+ depth = 0
+ for i, char in enumerate(aci):
+ if char == '(' and depth == 0:
+ lbr_list.append(i)
+ if char == '(':
+ depth += 1
+ if char == ')' and depth == 1:
+ rbr_list.append(i)
+ if char == ')':
+ depth -= 1
+ # Now build a set of terms.
+ terms = []
+ for lb, rb in zip(lbr_list, rbr_list):
+ terms.append(aci[lb + 1:rb])
+ return terms
+
+ def _parse_version_3_0(self, rawacipart, data):
+ terms = []
+ interms = rawacipart.split(';')
+ interms = map(lambda x: x.strip(), interms)
+ # We have to do this because it's not the same as other term formats.
+ for iwork in interms:
+ for j in self._v3keys + self._v3innerkeys:
+ if iwork.startswith(j) and j == 'acl':
+ t = iwork.split(' ', 1)[1]
+ t = t.replace('"', '')
+ data[j].append({ 'values' : [t]})
+ if iwork.startswith(j) and j == 'allow':
+ #First we need to get the bits.
+ first = iwork.index('(') + 1
+ second = iwork.index(')', first)
+ # This could likely be neater ...
+ data[j].append({
+ 'values' : map( lambda x: x.strip(),
+ iwork[first:second].split(',')
+ )
+ })
+ # This wouldn't parse properly:
+ ### (version 3.0; acl "Enable anonymous access"; allow (read, search, compare) userdn="ldap:///anyone";;)
+ # it's because _find_terms is expecting userdn wrapped in braces.
+ # but the term that follows an allow, the () is optional.
+ # so as a result, it upsets the _find_terms parser.
+
+ # How do we handle "userdn=foo and userdn=bar"
+ # This likely won't work with "userdn=ldap:///(filter)"
+ # How can we do this better? We likely need to make sure the () are not between ""
+ # Find the locations of the " chars.
+ subterm = iwork[second + 1:]
+ if '(' not in subterm and ')' not in subterm:
+ terms += map(lambda x: x.strip(), subterm.split('and'))
+ else:
+ terms += self._find_terms(subterm)
+ return terms
+
+ def _parse_aci(self, rawaci):
+ #hostdn, aci = entry
+ aci = rawaci
+ depth = 0
+ data = {
+ 'rawaci': rawaci,
+ }
+ #data['dn'] = entry.dn
+ for k in self._keys + self._v3keys + self._v3innerkeys:
+ data[k] = []
+ # We need to get a list of all the depth 0 ( and )
+ terms = self._find_terms(aci)
+
+ while len(terms) > 0:
+ work = terms.pop()
+ for k in self._keys + self._v3keys + self._v3innerkeys:
+ if work.startswith(k):
+ aci = work.replace(k, '', 1)
+ if k == 'version 3.0;':
+ #We pop more inner terms out, but we don't need to parse them "now"
+ # they get added to the queue
+ terms += self._parse_version_3_0(aci, data)
+ continue
+ # Nearly all terms are = seperated
+ ## We make a dict that holds "equal" and an array of values
+ wdict = { 'values': [] , 'equal': True}
+ pre, val = aci.split('=', 1)
+ val = val.replace('"', '')
+ #elif k == 'targetattr':
+ if pre.strip() == '!':
+ wdict['equal'] = False
+ else:
+ wdict['equal'] = True
+ wdict['values'] = val.split('||')
+ if k in self._urlkeys:
+ ### / We could replace ldap:/// in some attrs?
+ wdict['values'] = map(lambda x: x.replace('ldap:///',''), wdict['values'])
+ wdict['values'] = map(lambda x: x.strip(), wdict['values'])
+ data[k].append(wdict)
+ break # So we don't double trigger the check
+ return data
+
diff --git a/lib389/aci.py b/lib389/aci.py
new file mode 100644
index 0000000..40b0a7a
--- /dev/null
+++ b/lib389/aci.py
@@ -0,0 +1,29 @@
+"""Aci class to help parse and create ACIs.
+
+You will access this via the Entry Class.
+"""
+
+import ldap
+
+from lib389._constants import *
+from lib389 import Entry, InvalidArgumentError
+
+# Add a helper for aci listing.
+
+class Aci(object):
+
+
+ def __init__(self, conn):
+ """
+ """
+ self.conn = conn
+ self.log = conn.log
+
+ def list(self, basedn):
+ acis = []
+ rawacientries = self.conn.search_s(basedn, ldap.SCOPE_SUBTREE, 'objectClass=*', ['aci'])
+ for rawacientry in rawacientries:
+ acis += rawacientry.getAcis()
+ return acis
+
+
diff --git a/tests/aci_parse_test.py b/tests/aci_parse_test.py
new file mode 100644
index 0000000..56c7869
--- /dev/null
+++ b/tests/aci_parse_test.py
@@ -0,0 +1,54 @@
+'''
+Created on Aug 3, 2015
+
+@author: William Brown
+'''
+from lib389._constants import *
+from lib389._aci import Aci
+from lib389 import DirSrv,Entry
+import ldap
+
+INSTANCE_PORT = 54321
+INSTANCE_SERVERID = 'aciparseds'
+#INSTANCE_PREFIX = None
+
+class Test_schema():
+ def setUp(self):
+ instance = DirSrv(verbose=False)
+ instance.log.debug("Instance allocated")
+ args = {SER_HOST: LOCALHOST,
+ SER_PORT: INSTANCE_PORT,
+ #SER_DEPLOYED_DIR: INSTANCE_PREFIX,
+ SER_SERVERID_PROP: INSTANCE_SERVERID
+ }
+ instance.allocate(args)
+ if instance.exists():
+ instance.delete()
+ instance.create()
+ instance.open()
+ self.instance = instance
+
+ def tearDown(self):
+ if self.instance.exists():
+ #self.instance.db2ldif(bename='userRoot', suffixes=[DEFAULT_SUFFIX], excludeSuffixes=[], encrypt=False, repl_data=False, outputfile='%s/ldif/%s.ldif' % (self.instance.dbdir,INSTANCE_SERVERID ))
+ #self.instance.clearBackupFS()
+ #self.instance.backupFS()
+ self.instance.delete()
+
+ def test_aci(self):
+ acis = self.instance.aci.list(DEFAULT_SUFFIX)
+ # Return the list
+ for aci in acis:
+ print('---------')
+ for key in aci.acidata:
+ print('%s : %s' % (key, aci.acidata[key]))
+ print(aci.getRawAci())
+ # Add some better tests that actually check we got the
+ # default aci's back.
+
+if __name__ == "__main__":
+ test = Test_schema()
+ test.setUp()
+ test.test_aci()
+ test.tearDown()
+
--
2.4.3
--
389-devel mailing list
389-devel@xxxxxxxxxxxxxxxxxxxxxxx
https://admin.fedoraproject.org/mailman/listinfo/389-devel
