From 5eab3b577d486a2166afd5207b7fb5cde4b19780 Mon Sep 17 00:00:00 2001
From: Mark Reynolds <mreynolds@redhat.com>
Date: Dec 04 2018 00:49:39 +0000
Subject: Ticket 50065 - lib389 aci parsing is too strict


Bug Description:  ACI parsing is very strict around parsing "version 3.0;".
                  If there are any spaces around the semicolon parsing fails.

Fix Description:  Add a normalization function that removes duplicate
                  consecutive spaces, and handles spaces around the version
                  string.

https://pagure.io/389-ds-base/issue/50065

Reviewed by: spichugi(Thanks!)

---

diff --git a/src/lib389/lib389/_entry.py b/src/lib389/lib389/_entry.py
index fb72fca..1f039ff 100644
--- a/src/lib389/lib389/_entry.py
+++ b/src/lib389/lib389/_entry.py
@@ -235,7 +235,7 @@ class Entry(object):
             # This converts the dict to a list of tuples,
             lt = []
             for k in self.data.keys():
-                # l here is the 
+                # l here is the
                 vals = None
                 if isinstance(self.data[k], list) or isinstance(self.data[k], tuple):
                     vals = ensure_list_bytes(self.data[k])
@@ -503,6 +503,21 @@ class EntryAci(object):
         rawaci += ")"
         return ensure_bytes(rawaci)
 
+    def _normalize_term(self, term):
+        """Normalize the term, remove duplicate spaces around sensitive keywords,
+        and remove spaces around "version 3.0 ;"
+        :retuns: normalized term string
+        """
+
+        # First reduce multiple consecutive spaces
+        new_term = ' '.join(term.split())
+
+        # Our parsing of version 3.0; is strict, need to take special care
+        if new_term.lower().startswith('version 3.0'):
+            parts = new_term.split(';', 1)
+            new_term = parts[0].rstrip() + ';' + parts[1]
+        return new_term
+
     def _find_terms(self, aci):
         if self.verbose:
             print("_find_terms aci: %s" % aci)
@@ -525,7 +540,7 @@ class EntryAci(object):
             print("_find_terms lbr_list" % lbr_list)
             print("_find_terms rbr_list" % rbr_list)
         for lb, rb in zip(lbr_list, rbr_list):
-            terms.append(aci[lb + 1:rb])
+            terms.append(self._normalize_term(aci[lb + 1:rb]))
         if self.verbose:
             print("_find_terms terms: %s" % terms)
         return terms
diff --git a/src/lib389/lib389/tests/aci_parse_test.py b/src/lib389/lib389/tests/aci_parse_test.py
new file mode 100644
index 0000000..55a8ed6
--- /dev/null
+++ b/src/lib389/lib389/tests/aci_parse_test.py
@@ -0,0 +1,57 @@
+from lib389._entry import EntryAci
+from lib389.utils import *
+import pytest
+import os
+
+DEBUGGING = os.getenv("DEBUGGING", default=False)
+if DEBUGGING:
+    logging.getLogger(__name__).setLevel(logging.DEBUG)
+else:
+    logging.getLogger(__name__).setLevel(logging.INFO)
+log = logging.getLogger(__name__)
+
+
+rawaci0 = ('(  targetattr =  "objectclass")( target = "ldap:///cn=retrieve '
+           'certificate,cn=virtual operations,cn=etc,dc=ipa,dc=example"   )('
+           '   version 3.0    ;    acl "permission:Retrieve   Certificates '
+           'from the CA";allow (write) groupdn = "ldap:///cn=Retrieve Certif'
+           'icates from the CA,cn=permissions,cn=pbac,dc=ipa,dc=example;)')
+rawaci2 = ('(targetattr = "objectclass")(target = "ldap:///cn=retrieve certi'
+           'ficate,cn=virtual operations,cn=etc,dc=ipa,dc=example" )(version'
+           ' 3.0; acl "permission:Retrieve Certificates from the CA" ; allow '
+           '(write) groupdn = "ldap:///cn=Retrieve Certificates from the CA,'
+           'cn=permissions,cn=pbac,dc=ipa,dc=example;)')
+rawaci3 = ('(targetfilter ="(ou=groups)")(targetattr ="uniqueMember || member")'
+           '(version 3.0; acl "Allow test aci";allow (read, search, write)'
+           '(userdn="ldap:///dc=example,dc=com??sub?(ou=engineering)" '
+           'and userdn="ldap:///dc=example,dc=com??sub?(manager=uid='
+           'wbrown,ou=managers,dc=example,dc=com) || ldap:///dc=examp'
+           'le,dc=com??subrawaci?(manager=uid=tbrown,ou=managers,dc=exampl'
+           'e,dc=com)" );)')
+rawaci4 = ('(  targetattr =  "objectclass")( target = "ldap:///cn=retrieve '
+           'certificate,cn=virtual operations,cn=etc,dc=ipa,dc=example"   )('
+           '   version   3.0  ;  acl "permission:Retrieve Certificates '
+           'from the CA";allow (write) groupdn =   "ldap:///cn=Retrieve Certif'
+           'icates from the CA,cn=permissions,cn=pbac,dc=ipa,dc=example;)')
+rawaci5 = ('(targetfilter ="(ou=groups)")( targetattr =    "uniqueMember  ||   member")'
+           '(version 3.0; acl "Allow test aci";allow (read, search, write)'
+           '(userdn="ldap:///dc=example,dc=com??sub?(ou=engineering)" '
+           'and userdn="ldap:///dc=example,dc=com??sub?(manager=uid='
+           'wbrown,ou=managers,dc=example,dc=com)   ||  ldap:///dc=examp'
+           'le,dc=com??subrawaci?(manager=uid=tbrown,ou=managers,dc=exampl'
+           'e,dc=com)" );)')
+
+acis = [rawaci0, rawaci2, rawaci3, rawaci4, rawaci5]
+
+
+def test_parse_aci():
+    for rawaci in acis:
+        # parse the aci - there should be no exceptions
+        aci_obj = EntryAci(None, rawaci)
+        aci_obj.getRawAci()
+
+
+if __name__ == "__main__":
+    CURRENT_FILE = os.path.realpath(__file__)
+    pytest.main("-s -vv %s" % CURRENT_FILE)
+