CVE-2020-25722 pytest: test sAMAccountName/userPrincipalName over ldap

Because the sam account name + the dns host name is used as the
default user principal name, we need to check for collisions between
these. Fixes are coming in upcoming patches.

BUG: https://bugzilla.samba.org/show_bug.cgi?id=14564

Signed-off-by: Douglas Bagnall <douglas.bagnall@catalyst.net.nz>
Reviewed-by: Andrew Bartlett <abartlet@samba.org>

1 files changed, 510 insertions, 0 deletions

diff --git a/python/samba/tests/ldap_upn_sam_account.py b/python/samba/tests/ldap_upn_sam_account.py
new file mode 100644
index 00000000000..cc1cce9b6c3
--- /dev/null
+++ b/python/samba/tests/ldap_upn_sam_account.py
@@ -0,0 +1,510 @@
+# Unix SMB/CIFS implementation.
+#
+# Copyright 2021 (C) Catalyst IT Ltd
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+
+import os
+import sys
+from samba.samdb import SamDB
+from samba.auth import system_session
+import ldb
+from samba.tests.subunitrun import SubunitOptions, TestProgram
+from samba.tests import TestCase, ldb_err
+from samba.tests import DynamicTestCase
+import samba.getopt as options
+import optparse
+from samba.colour import c_RED, c_GREEN, c_DARK_YELLOW
+import re
+import pprint
+from samba.dsdb import (
+    UF_SERVER_TRUST_ACCOUNT,
+    UF_TRUSTED_FOR_DELEGATION,
+)
+
+
+# bad sAMAccountName characters from [MS-SAMR]
+# "3.1.1.6 Attribute Constraints for Originating Updates"
+BAD_SAM_CHARS = (''.join(chr(x) for x in range(0, 32)) +
+                 '"/\\[]:|<>+=;?,*')
+
+# 0x7f is *said* to be bad, but turns out to be fine.
+ALLEGED_BAD_SAM_CHARS = chr(127)
+
+LATIN1_BAD_CHARS = set([chr(x) for x in range(129, 160)] +
+                       list("ªºÿ") +
+                       [chr(x) for x in range(0xc0, 0xc6)] +
+                       [chr(x) for x in range(0xc7, 0xd7)] +
+                       [chr(x) for x in range(0xd8, 0xde)] +
+                       [chr(x) for x in range(0xe0, 0xe6)] +
+                       [chr(x) for x in range(0xe7, 0xf7)] +
+                       [chr(x) for x in range(0xf8, 0xfe)])
+
+
+LATIN_EXTENDED_A_NO_CLASH = {306, 307, 330, 331, 338, 339, 358, 359, 383}
+
+#XXX does '\x00' just truncate the string though?
+#XXX elsewhere we see "[\\\"|,/:<>+=;?*']" with "'"
+
+
+## UPN limits
+# max length 1024 UTF-8 bytes, following "rfc822"
+# for o365 sync https://docs.microsoft.com/en-us/microsoft-365/enterprise/prepare-for-directory-synchronization?view=o365-worldwide
+# max length is 113 [64 before @] "@" [48 after @]
+# invalid chars: '\\%&*+/=?{}|<>();:,[]"'
+# allowed chars: A – Z, a - z, 0 – 9, ' . - _ ! # ^ ~
+# "Letters with diacritical marks, such as umlauts, accents, and tildes, are invalid characters."
+#
+# "@" can't be first
+# "The username cannot end with a period (.), an ampersand (&), a space, or an at sign (@)."
+#
+
+# per RFC 822, «"a b" @ example.org» is
+
+
+ok = True
+bad = False
+report = 'report'
+exists = ldb.ERR_ENTRY_ALREADY_EXISTS
+
+
+if sys.stdout.isatty():
+    c_doc = c_DARK_YELLOW
+else:
+    c_doc = lambda x: x
+
+
+def get_samdb():
+    return SamDB(url=f"ldap://{SERVER}",
+                 lp=LP,
+                 session_info=system_session(),
+                 credentials=CREDS)
+
+
+def format(s):
+    if type(s) is str:
+        s = s.format(realm=REALM.upper(),
+                     lrealm=REALM.lower(),
+                     other_realm=(REALM + ".another.example.net"))
+    return s
+
+
+class LdapUpnSamTestBase(TestCase):
+    """Make sure we can't add userPrincipalNames or sAMAccountNames that
+    implicitly collide.
+    """
+    _disabled = False
+
+    @classmethod
+    def setUpDynamicTestCases(cls):
+        if getattr(cls, '_disabled', False):
+            return
+        for doc, *rows in cls.cases:
+            name = re.sub(r'\W+', '_', doc)
+            cls.generate_dynamic_test("test_upn_sam", name, rows, doc)
+
+    def setup_objects(self, rows):
+        objects = set(r[0] for r in rows)
+        for name in objects:
+            if ':' in name:
+                objtype, name = name.split(':', 1)
+            else:
+                objtype = 'user'
+            getattr(self, f'add_{objtype}')(name)
+            self.addCleanup(self.remove_object, name)
+
+    def _test_upn_sam_with_args(self, rows, doc):
+        self.setup_objects(rows)
+        cdoc = c_doc(doc)
+
+        for i, row in enumerate(rows):
+            if len(row) == 4:
+                obj, data, expected, op = row
+            else:
+                obj, data, expected = row
+                op = ldb.FLAG_MOD_REPLACE
+
+            dn, dnsname = self.objects[obj]
+            sam, upn = None, None
+            if isinstance(data, dict):
+                sam = data.get('sam')
+                upn = data.get('upn')
+            elif isinstance(data, str):
+                if '@' in data:
+                    upn = data
+                else:
+                    sam = data
+            else:  # bytes
+                if b'@' in data:
+                    upn = data
+                else:
+                    sam = data
+
+            m = {"dn": dn}
+
+            if upn is not None:
+                m["userPrincipalName"] = format(upn)
+
+            if sam is not None:
+                m["sAMAccountName"] = format(sam)
+
+            msg = ldb.Message.from_dict(self.samdb, m, op)
+
+            if expected is bad:
+                try:
+                    self.samdb.modify(msg)
+                except ldb.LdbError as e:
+                    print(f"row {i+1} of '{cdoc}' failed as expected with "
+                          f"{ldb_err(e)}\n")
+                    continue
+                self.fail(f"row {i+1} of '{cdoc}' should have failed:\n"
+                          f"{pprint.pformat(m)} on {obj}")
+            elif expected is ok:
+                try:
+                    self.samdb.modify(msg)
+                except ldb.LdbError as e:
+                    raise AssertionError(
+                        f"row {i+1} of '{cdoc}' failed with {ldb_err(e)}:\n"
+                        f"{pprint.pformat(m)} on {obj}") from None
+            elif expected is report:
+                try:
+                    self.samdb.modify(msg)
+                    print(f"row {i+1} of '{cdoc}' SUCCEEDED:\n"
+                          f"{pprint.pformat(m)} on {obj}")
+                except ldb.LdbError as e:
+                    print(f"row {i+1} of '{cdoc}' FAILED "
+                          f"with {ldb_err(e)}:\n"
+                          f"{pprint.pformat(m)} on {obj}")
+
+            else:
+                try:
+                    self.samdb.modify(msg)
+                except ldb.LdbError as e:
+                    if hasattr(expected, '__contains__'):
+                        if e.args[0] in expected:
+                            continue
+
+                    if e.args[0] == expected:
+                        continue
+
+                    self.fail(f"row {i+1} of '{cdoc}' "
+                              f"should have failed with {ldb_err(expected)} "
+                              f"but instead failed with {ldb_err(e)}:\n"
+                              f"{pprint.pformat(m)} on {obj}")
+                self.fail(f"row {i+1} of '{cdoc}' "
+                          f"should have failed with {ldb_err(expected)}:\n"
+                          f"{pprint.pformat(m)} on {obj}")
+
+    def add_dc(self, name):
+        dn = f"CN={name},OU=Domain Controllers,{self.base_dn}"
+        dnsname = f"{name}.{REALM}".lower()
+        self.samdb.add({
+            "dn": dn,
+            "objectclass": "computer",
+            "userAccountControl": str(UF_SERVER_TRUST_ACCOUNT |
+                                      UF_TRUSTED_FOR_DELEGATION),
+            "dnsHostName": dnsname,
+            "carLicense": self.id()
+        })
+        self.objects[name] = (dn, dnsname)
+
+    def add_user(self, name):
+        dn = f"CN={name},{self.ou}"
+        self.samdb.add({
+            "dn": dn,
+            "name": name,
+            "objectclass": "user",
+            "carLicense": self.id()
+        })
+        self.objects[name] = (dn, None)
+
+    def remove_object(self, name):
+        dn, dnsname = self.objects.pop(name)
+        self.samdb.delete(dn)
+
+    def setUp(self):
+        super().setUp()
+        self.samdb = get_samdb()
+        self.base_dn = self.samdb.get_default_basedn()
+        self.short_id = self.id().rsplit('.', 1)[1][:63]
+        self.objects = {}
+        self.ou = f"OU={ self.short_id },{ self.base_dn }"
+        self.addCleanup(self.samdb.delete, self.ou, ["tree_delete:1"])
+        self.samdb.add({"dn": self.ou, "objectclass": "organizationalUnit"})
+
+
+@DynamicTestCase
+class LdapUpnSamTest(LdapUpnSamTestBase):
+    cases = [
+        # The structure is
+        # ( «documentation/message that becomes test name»,
+        #    («short object id», «upn or sam or mapping», «expected»),
+        #    («short object id», «upn or sam or mapping», «expected»),
+        #    ...,
+        # )
+        #
+        # where the first item is a one line string explaining the
+        # test, and subsequent items describe database modifications,
+        # to be applied in series.
+        #
+        # First is a short ID, which maps to an object DN. Second is
+        # either a string or a dictionary.
+        #
+        # * If a string, if it contains '@', it is a UPN, otherwise a
+        #   samaccountname.
+        #
+        # * If a dictionary, it is a mapping of some of ['sam', 'upn']
+        #   to strings (in this way, you can add two attributes in one
+        #   mesage, or attempt a samaccountname with '@').
+        #
+        # expected can be «ok», «bad» (mapped to True and False,
+        # respectively), or a specific LDB error code, if that exact
+        # exception is wanted.
+        ("add good UPN",
+         ('A', 'a@{realm}', ok),
+        ),
+        ("add the same upn to different objects",
+         ('A', 'a@{realm}', ok),
+         ('B', 'a@{realm}', ldb.ERR_CONSTRAINT_VIOLATION),
+         ('B', 'a@{lrealm}', ldb.ERR_CONSTRAINT_VIOLATION),  # lowercase realm
+        ),
+        ("replace UPN with itself",
+         ('A', 'a@{realm}', ok),
+         ('A', 'a@{realm}', ok),
+         ('A', 'a@{lrealm}', ok),
+        ),
+        ("replace SAM with itself",
+         ('A', 'a', ok),
+         ('A', 'a', ok),
+        ),
+        ("replace UPN realm",
+         ('A', 'a@{realm}', ok),
+         ('A', 'a@{other_realm}', ok),
+        ),
+        ("matching SAM and UPN",
+         ('A', 'a', ok),
+         ('A', 'a@{realm}', ok),
+        ),
+        ("matching SAM and UPN, other realm",
+         ('A', 'a', ok),
+         ('A', 'a@{other_realm}', ok),
+        ),
+        ("matching SAM and UPN, single message",
+         ('A', {'sam': 'a', 'upn': 'a@{realm}'}, ok),
+         ('A', {'sam': 'a', 'upn': 'a@{other_realm}'}, ok),
+        ),
+        ("different objects, different realms",
+         ('A', 'a@{realm}', ok),
+         ('B', 'a@{other_realm}', ok),
+        ),
+        ("different objects, same UPN, different case",
+         ('A', 'a@{realm}', ok),
+         ('B', 'A@{realm}', ldb.ERR_CONSTRAINT_VIOLATION),
+        ),
+        ("different objects, SAM after UPN",
+         ('A', 'a@{realm}', ok),
+         ('B', 'a', ldb.ERR_CONSTRAINT_VIOLATION),
+        ),
+        ("different objects, SAM before UPN",
+         ('A', 'a', ok),
+         ('B', 'a@{realm}', exists),
+        ),
+        ("different objects, SAM account clash",
+         ('A', 'a', ok),
+         ('B', 'a', exists),
+        ),
+        ("different objects, SAM account clash, different case",
+         ('A', 'a', ok),
+         ('B', 'A', exists),
+        ),
+        ("two way clash",
+         ('A', {'sam': 'x', 'upn': 'y@{realm}'}, ok),
+         # The sam account raises EXISTS while the UPN raises
+         # CONSTRAINT_VIOLATION. We don't really care in which order
+         # they are checked, so either error is ok.
+         ('B', {'sam': 'y', 'upn': 'x@{realm}'},
+          (exists, ldb.ERR_CONSTRAINT_VIOLATION)),
+        ),
+        ("two way clash, other realm",
+         ('A', {'sam': 'x', 'upn': 'y@{other_realm}'}, ok),
+         ('B', {'sam': 'y', 'upn': 'x@{other_realm}'}, ok),
+        ),
+        # UPN versions of bad sam account names
+        ("UPN clash on other realm",
+         ('A', 'a@x.x', ok),
+         ('B', 'a@x.x', ldb.ERR_CONSTRAINT_VIOLATION),
+        ),
+        ("UPN same but for trailing spaces",
+         ('A', 'a@{realm}', ok),
+         ('B', 'a @{realm}', ok),
+        ),
+        # UPN has no at
+        ("UPN has no at",
+         ('A', {'upn': 'noat'}, ok),
+         ('B', {'upn': 'noat'}, ldb.ERR_CONSTRAINT_VIOLATION),
+         ('C', {'upn': 'NOAT'}, ldb.ERR_CONSTRAINT_VIOLATION),
+        ),
+        # UPN has non-ascii at, followed by real at.
+        ("UPN with non-ascii at vs real at",
+         ('A', {'upn': 'smallat﹫{realm}'}, ok),
+         ('B', {'upn': 'smallat@{realm}'}, ok),
+         ('C', {'upn': 'tagat\U000e0040{realm}'}, ok),
+         ('D', {'upn': 'tagat@{realm}'}, ok),
+        ),
+        ("UPN with unicode at vs real at, real at first",
+         ('B', {'upn': 'smallat@{realm}'}, ok),
+         ('A', {'upn': 'smallat﹫{realm}'}, ok),
+         ('D', {'upn': 'tagat@{realm}'}, ok),
+         ('C', {'upn': 'tagat\U000e0040{realm}'}, ok),
+        ),
+        ("UPN username too long",
+         #  SPN soft limit 20; hard limit 256, overall UPN 1024
+         ('A', 'a' * 25 + '@b.c', ok),
+         ('A', 'a' * 65 + '@b.c', ok),   # Azure AD limit is 64
+         ('A', 'a' * 257 + '@b.c', ok),  # 256 is sam account name limit
+        ),
+        ("sam account name 20 long",
+         #  SPN soft limit 20
+         ('A', 'a' * 20, ok),
+        ),
+        ("UPN has two at signs",
+         ('A', 'a@{realm}', ok),
+         ('A', 'a@{realm}@{realm}', ok),
+         ('A', 'a@a.b', ok),
+         ('A', 'a@a@a.b', ok),
+        ),
+        ("SAM has at signs clashing upn second, non-realm",
+         ('A', {'sam': 'a@a.b'}, ok),
+         ('B', 'a@a.b@a.b', ok),  # UPN won't clash with SAM, because realm
+        ),
+        ("SAM has at signs clashing upn second",
+         ('A', {'sam': 'a@{realm}'}, ok),
+         ('B', 'a@{realm}@{realm}', bad),  # UPN would clashes with SAM
+        ),
+        ("SAM has at signs clashing upn first",
+         ('B', 'a@{realm}@{realm}', ok),
+         ('A', {'sam': 'a@{realm}'}, bad),
+        ),
+        ("spaces around at",
+         ('A', 'a name @ {realm}', ok),
+         ('B', 'a name @ {realm}', ldb.ERR_CONSTRAINT_VIOLATION),
+         ('B', 'a name @{realm}', ok),  # because realm looks different
+         ('C', 'a name@{realm}', ok),
+         ('D', 'a name', ldb.ERR_CONSTRAINT_VIOLATION),
+         ('D', 'a name ', (exists, ldb.ERR_CONSTRAINT_VIOLATION)),  # matches B
+        ),
+        ("SAM starts with at",
+         ('A', {'sam': '@{realm}'}, ok),
+         ('B', {'sam': '@a'}, ok),
+         ('C', {'sam': '@{realm}'}, exists),
+         ('C', {'sam': '@a'}, exists),
+         ('C', {'upn': '@{realm}@{realm}'}, bad),
+         ('C', {'upn': '@a@{realm}'}, bad),
+        ),
+        ("UPN starts with at",
+         ('A', {'upn': '@{realm}'}, ok),
+         ('B', {'upn': '@a@{realm}'}, ok),
+         ('C', {'upn': '@{realm}'}, bad),
+         ('C', {'sam': '@a'}, bad),
+        ),
+        ("SAM ends with at",
+         ('A', {'sam': '{realm}@'}, ok),
+         ('B', {'sam': 'a@'}, ok),
+         ('C', {'sam': '{realm}@'}, exists),
+         ('C', {'sam': 'a@'}, exists),
+         ('C', {'upn': 'a@@{realm}'}, bad),
+         ('C', {'upn': '{realm}@@{realm}'}, bad),
+        ),
+        ("UPN ends with at",
+         ('A', {'upn': '{realm}@'}, ok),
+         ('B', {'upn': '@a@{realm}@'}, ok),
+         ('C', {'upn': '{realm}@'}, bad),
+         ('C', {'sam': '@a@{realm}'}, ok),  # not like B, because other realm
+        ),
+    ]
+
+
+@DynamicTestCase
+class LdapUpnSamSambaOnlyTest(LdapUpnSamTestBase):
+    # We don't run these ones outside of selftest, where we are
+    # probably testing against Windows and these are known failures.
+    _disabled = 'SAMBA_SELFTEST' not in os.environ
+    cases = [
+        ("sam account name too long",
+         # SPN soft limit 20
+         ('A', 'a' * 19, ok),
+         ('A', 'a' * 20, ok),
+         ('A', 'a' * 65, ok),
+         ('A', 'a' * 255, ok),
+         ('A', 'a' * 256, ok),
+         ('A', 'a' * 257, ldb.ERR_INVALID_ATTRIBUTE_SYNTAX),
+        ),
+        ("UPN username too long",
+         ('A', 'a' * 254 + '@' + 'b.c' * 257,
+          ldb.ERR_INVALID_ATTRIBUTE_SYNTAX),  # 1024 is alleged UPN limit
+        ),
+        ("UPN same but for internal spaces",
+         ('A', 'a b@x.x', ok),
+         ('B', 'a  b@x.x', ldb.ERR_CONSTRAINT_VIOLATION),
+        ),
+        ("SAM contains delete",
+         # forbidden according to documentation, but works in practice on Windows
+         ('A', 'a\x7f', ldb.ERR_CONSTRAINT_VIOLATION),
+         ('A', 'a\x7f'.encode(), ldb.ERR_CONSTRAINT_VIOLATION),
+         ('A', 'a\x7fb', ldb.ERR_CONSTRAINT_VIOLATION),
+         ('A', 'a\x7fb'.encode(), ldb.ERR_CONSTRAINT_VIOLATION),
+         ('A', '\x7fb', ldb.ERR_CONSTRAINT_VIOLATION),
+         ('A', '\x7fb'.encode(), ldb.ERR_CONSTRAINT_VIOLATION),
+         ),
+        # The wide at symbol ('＠' U+FF20) does not count as '@' for Samba
+        # so it will look like a string with no @s.
+        ("UPN with unicode wide at vs real at",
+         ('A', {'upn': 'wideat＠{realm}'}, ok),
+         ('B', {'upn': 'wideat@{realm}'}, ok),
+        ),
+        ("UPN with real at vs wide at",
+         ('B', {'upn': 'wideat@{realm}'}, ok),
+         ('A', {'upn': 'wideat＠{realm}'}, ok)
+        ),
+    ]
+
+
+def main():
+    global LP, CREDS, SERVER, REALM
+
+    parser = optparse.OptionParser(
+        "python3 ldap_upn_sam_account_name.py <server> [options]")
+    sambaopts = options.SambaOptions(parser)
+    parser.add_option_group(sambaopts)
+
+    # use command line creds if available
+    credopts = options.CredentialsOptions(parser)
+    parser.add_option_group(credopts)
+    subunitopts = SubunitOptions(parser)
+    parser.add_option_group(subunitopts)
+
+    opts, args = parser.parse_args()
+    if len(args) != 1:
+        parser.print_usage()
+        sys.exit(1)
+
+    LP = sambaopts.get_loadparm()
+    CREDS = credopts.get_credentials(LP)
+    SERVER = args[0]
+    REALM = CREDS.get_realm()
+
+    TestProgram(module=__name__, opts=subunitopts)
+
+main()