From c42da4d05978beb80358efd813e453c40dedd3ec Mon Sep 17 00:00:00 2001
From: William Brown <firstyear@redhat.com>
Date: Fri, 1 Apr 2016 13:12:51 +1000
Subject: [PATCH] Ticket 48433 - Aci linting tools

Bug Description:  Directory Server acis are very complex. We should provide
linting tools that can detect and provide guidance around potential aci errors
that are common or easily made.

Fix Description:  Provide an aci linting test framework, and some initial lint
tests.

https://fedorahosted.org/389/ticket/48433

Author: wibrown

Review by: ???
---
 lib389/_entry.py               |  12 +++
 lib389/aci.py                  | 185 +++++++++++++++++++++++++++++++++++++++++
 lib389/clitools/ds_aci_lint.py |  41 +++++++++
 lib389/tests/aci_lint_test.py  |  90 ++++++++++++++++++++
 4 files changed, 328 insertions(+)
 create mode 100644 lib389/clitools/ds_aci_lint.py
 create mode 100644 lib389/tests/aci_lint_test.py

diff --git a/lib389/_entry.py b/lib389/_entry.py
index b5d5988..9de7cf7 100644
--- a/lib389/_entry.py
+++ b/lib389/_entry.py
@@ -385,6 +385,18 @@ class EntryAci(object):
         self._rawaci = rawaci
         self.acidata = self._parse_aci(self._rawaci)
 
+    def __eq__(self, other):
+
+        # If we are from different entries, we should fail
+        # Can this fail if we have different filters to get the acis though?
+        if self.entry.dn != other.entry.dn:
+            return False
+        # Should we be checking the generated Aci incase it has changed?
+        return self.getRawAci() == other.getRawAci()
+
+    def __ne__(self, other):
+        return not self.__eq__(other)
+
     def _format_term(self, key, value_dict):
         rawaci = ''
         if value_dict['equal']:
diff --git a/lib389/aci.py b/lib389/aci.py
index c068a10..6fdc0de 100644
--- a/lib389/aci.py
+++ b/lib389/aci.py
@@ -15,6 +15,20 @@ import ldap
 
 from lib389._constants import *
 
+# Helpers to detect common patterns in aci
+
+def _aci_any_targetattr_ne(aci):
+    """
+    Returns if any of the targetattr types is a != type
+    """
+    potential = False
+    if 'targetattr' in aci.acidata:
+        for ta in aci.acidata['targetattr']:
+            if ta['equal'] is False:
+                # Got, it lets do this
+                potential = True
+    return potential
+
 
 class Aci(object):
     def __init__(self, conn):
@@ -35,3 +49,174 @@ class Aci(object):
         for rawacientry in rawacientries:
             acis += rawacientry.getAcis()
         return acis
+
+    def lint(self, basedn, scope=ldap.SCOPE_SUBTREE):
+        """
+        Validate and check for potential aci issues.
+
+        Given a scope and basedn, this will retrieve all the aci's below.
+        A number of checks are then run on the aci in isolation, and
+        in groups.
+
+        returns a tuple of (bool, list( dict ))
+
+        bool represents if the acis pass or fail as a whole.
+        the list contains a list of warnings about your acis.
+        the dict is structured as:
+        {
+          name: "" # DSALEXXXX
+          severity: "" # LOW MEDIUM HIGH
+          detail: "" # explination
+        }
+        """
+        result = True
+        # Not thread safe!!!
+        self.warnings = []
+
+        acis = self.list(basedn, scope)
+        # Checks again "all acis" go here.
+        self._lint_dsale_0001_ne_internal(acis)
+        self._lint_dsale_0002_ne_mult_subtree(acis)
+        # checks again individual here
+
+        if len(self.warnings) > 0:
+            result = False
+
+        return (result, self.warnings)
+
+    def format_lint(self, warnings):
+        """
+        Takes the array of warnings and returns a formatted string.
+        """
+        buf = "-------------------------------------------------------------------------------"
+
+        for warning in warnings:
+            buf += """
+Directory Server Aci Lint Error: {DSALE}
+Severity: {SEVERITY}
+
+Affected Acis:
+{ACIS}
+
+Details: {DETAIL}
+
+Advice: {FIX}
+-------------------------------------------------------------------------------""".format(
+                DSALE=warning['dsale'],
+                SEVERITY=warning['severity'],
+                ACIS=warning['acis'],
+                DETAIL=warning['detail'],
+                FIX=warning['fix'],
+            )
+        return buf
+
+    ### These are the aci lint checks.
+
+    def _lint_dsale_0001_ne_internal(self, acis):
+        """
+        Check for the presence of "not equals" attributes that will inadvertantly
+        allow the return / modification of internal attributes.
+        """
+
+        affected = []
+        for aci in acis:
+            if 'targetattr' in aci.acidata:
+                for ta in aci.acidata['targetattr']:
+                    if ta['equal'] is False:
+                        affected.append(aci.acidata['rawaci'])
+
+        if len(affected) > 0:
+            self.warnings.append(
+                {
+                    'dsale' : 'DSALE0001',
+                    'severity': 'HIGH',
+                    'acis' : "\n".join(affected),
+                    'detail': """
+An aci of the form "(targetAttr!="attr")" exists on your system. This aci
+will internally be expanded to mean "all possible attributes including system,
+excluding the listed attributes".
+
+This may allow access to a bound user or anonymous to read more data about
+directory internals, including aci state or user limits. In the case of write 
+acis it may allow a dn to set their own resource limits, unlock passwords or
+their own aci.
+
+The ability to change the aci on the object may lead to privilege escalation in
+some cases.
+                    """,
+                    'fix' : """
+Convert the aci to the form "(targetAttr="x || y || z")".
+                    """
+                }
+            )
+
+    def _lint_dsale_0002_ne_mult_subtree(self, acis):
+        """
+        This check will show pairs or more of aci that match the same subtree
+        with a != rute. These can cause the other rule to be invalidated!
+        """
+
+        affected = []
+        for aci in acis:
+            # The aci has to be a NE, else don't bother checking
+            if not _aci_any_targetattr_ne(aci):
+                continue
+
+            affect = False
+            buf = "%s %s\n" % (aci.entry.dn, aci.acidata['rawaci'])
+            for aci_inner in acis:
+
+                if aci_inner == aci:
+                    # Don't compare to self!
+                    continue
+                # Check the inner is a not equal also
+                if not _aci_any_targetattr_ne(aci_inner):
+                    continue
+
+                # Check if the dn is a substring, ie child, or equal.
+                if aci.entry.dn.endswith(aci_inner.entry.dn):
+                    # alias the allow rules
+                    allow_inner = set(aci_inner.acidata['allow'][0]['values'])
+                    allow_outer = set(aci.acidata['allow'][0]['values'])
+                    if len(allow_inner & allow_outer):
+                        buf += "|- %s %s\n" % (aci_inner.entry.dn, aci_inner.acidata['rawaci'])
+                        affect = True
+
+            if affect:
+                affected.append(buf)
+
+        if len(affected) > 0:
+            self.warnings.append(
+                {
+                    'dsale' : 'DSALE0002',
+                    'severity': 'HIGH',
+                    'acis' : "\n".join(affected),
+                    'detail': """
+Acis on your system exist which are both not equals targetattr, and overlap in
+scope.
+
+The way that directory server processes these, is to invert them to to white
+lists, then union the results.
+
+As a result, these acis *may* allow access to the attributes you want them to
+exclude.
+
+Consider:
+
+aci: (targetattr !="cn")(version 3.0;acl "Self write all but cn";allow (write)
+    (userdn = "ldap:///self");)
+aci: (targetattr !="sn")(version 3.0;acl "Self write all but sn";allow (write)
+    (userdn = "ldap:///self");)
+
+This combination allows self write to *all* attributes within the subtree.
+
+In cases where the target is members of a group, it may allow a member who is
+within two groups to have elevated privilege.
+                    """,
+                    'fix' : """
+Convert the aci to the form "(targetAttr="x || y || z")".
+
+Prevent the acis from overlapping, and have them on unique subtrees.
+                    """
+                }
+            )
diff --git a/lib389/clitools/ds_aci_lint.py b/lib389/clitools/ds_aci_lint.py
new file mode 100644
index 0000000..c967cdb
--- /dev/null
+++ b/lib389/clitools/ds_aci_lint.py
@@ -0,0 +1,41 @@
+#!/usr/bin/python
+
+# --- BEGIN COPYRIGHT BLOCK ---
+# Copyright (C) 2015 Red Hat, Inc.
+# All rights reserved.
+#
+# License: GPL (version 3 or any later version).
+# See LICENSE for details.
+# --- END COPYRIGHT BLOCK ---
+
+# from clitools import clitools_parser, get_instance_dict, get_rootdn_pass
+from lib389.clitools import CliTool, clitools_parser
+# from lib389 import DirSrv
+from lib389._constants import *
+from argparse import ArgumentParser
+
+
+class AciLintTool(CliTool):
+    def aci_lint(self):
+        try:
+            self.populate_instance_dict(self.args.instance)
+            self.connect()
+            # This is pretty rough, it just dumps the objects
+            (result, details) = status = self.ds.aci.lint(self.args.suffix)
+            print(self.ds.aci.format_lint(details))
+            if result:
+                print("PASS")
+            else:
+                print("FAIL")
+        finally:
+            self.disconnect()
+
+if __name__ == '__main__':
+    # Do some arg parse stuff
+    # You can always add a child parser here too ...
+    parser = clitools_parser.add_argument_group('aci', 'aci linting options')
+    parser.add_argument('-b', '--suffix', help='The name of the suffix to ' +
+                        ' validate acis.', required=True)
+    args = clitools_parser.parse_args()
+    acilinttool = AciLintTool(args)
+    acilinttool.aci_lint()
diff --git a/lib389/tests/aci_lint_test.py b/lib389/tests/aci_lint_test.py
new file mode 100644
index 0000000..1963d00
--- /dev/null
+++ b/lib389/tests/aci_lint_test.py
@@ -0,0 +1,90 @@
+# --- BEGIN COPYRIGHT BLOCK ---
+# Copyright (C) 2015 Red Hat, Inc.
+# All rights reserved.
+#
+# License: GPL (version 3 or any later version).
+# See LICENSE for details.
+# --- END COPYRIGHT BLOCK ---
+#
+from lib389._constants import *
+from lib389 import DirSrv, Entry
+import pytest
+
+INSTANCE_PORT = 54321
+INSTANCE_SERVERID = 'aciparseds'
+# INSTANCE_PREFIX   = None
+
+
+class TopologyStandalone(object):
+    def __init__(self, standalone):
+        standalone.open()
+        self.standalone = standalone
+
+
+@pytest.fixture(scope="module")
+def topology(request):
+    standalone = DirSrv(verbose=True)
+    standalone.log.debug("Instance allocated")
+    args = {SER_HOST: LOCALHOST,
+            SER_PORT: INSTANCE_PORT,
+            # SER_DEPLOYED_DIR:  INSTANCE_PREFIX,
+            SER_SERVERID_PROP: INSTANCE_SERVERID}
+    standalone.allocate(args)
+    if standalone.exists():
+        standalone.delete()
+    standalone.create()
+    standalone.open()
+
+    def fin():
+        standalone.delete()
+    request.addfinalizer(fin)
+
+    return TopologyStandalone(standalone)
+
+
+@pytest.fixture(scope="module")
+def complex_aci(topology):
+    ## Add aci's that will be "invalid"
+    ACI_TARGET = ('(targetfilter ="(ou=groups)")(targetattr ="uniqueMember '
+                  '|| member")')
+    ACI_ALLOW = ('(version 3.0; acl "Allow test aci";allow (read, search, '
+                 'write)')
+    ACI_SUBJECT = ('(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??sub?(manager=uid=tbrown,ou=managers,dc=exampl'
+                   'e,dc=com)" );)')
+    ACI_BODY = ACI_TARGET + ACI_ALLOW + ACI_SUBJECT
+
+    gentry = Entry('cn=testgroup,%s' % DEFAULT_SUFFIX)
+    gentry.setValues('objectclass', 'top', 'extensibleobject')
+    gentry.setValues('cn', 'testgroup')
+    gentry.setValues('aci', ACI_BODY)
+
+    topology.standalone.add_s(gentry)
+
+    return ACI_BODY
+
+
+def test_aci_lint(topology, complex_aci):
+    """Checks content of previously added aci"""
+
+    # Make sure we actually have acis to check!
+    acis = topology.standalone.aci.list('cn=testgroup,%s' % DEFAULT_SUFFIX)
+    #print(acis)
+    assert len(acis) == 1
+
+    # Actually test them
+    (result, detail) = topology.standalone.aci.lint(DEFAULT_SUFFIX)
+    # What is the point of testing perfect acis! We need this to fail!
+    assert (result is False)
+    #print(topology.standalone.aci.format_lint(detail))
+
+    assert(len(detail) == 2)
+    assert (detail[0]['dsale'] == 'DSALE0001')
+    assert (detail[1]['dsale'] == 'DSALE0002')
+    # Check all the results
+
+if __name__ == "__main__":
+    CURRENT_FILE = os.path.realpath(__file__)
+    pytest.main("-s -vv %s" % CURRENT_FILE)
-- 
2.5.5

