Public release

checks/bad_successor.py

@@ -1,66 +1,19 @@
 from checks.core import Check, check
 
 
-@check(risk="High", category="BadSuccessor - Default Groups with OU Privileges",
+@check(risk="High", category="BadSuccessor - Privileged OU/Container Control",
        entity="user", data=["bad_successor_ou_privileges"])
 class BadSuccessorCheck(Check):
-
-    CRITICAL_GROUPS_BY_SID = {
-        'S-1-5-11': 'AUTHENTICATED USERS',
-        '-513': 'DOMAIN USERS',
-        '-515': 'DOMAIN COMPUTERS',
-        'S-1-1-0': 'EVERYONE',
-        'S-1-5-32-545': 'USERS',
-    }
 
-    DANGEROUS_PERMISSIONS = {
-        'GenericAll', 'WriteOwner', 'WriteDacl', 'Owns',
-        'GenericWrite', 'WriteOwnerRaw', 'OwnsRaw',
-        'WriteGPLink', 'GPLink', 'CreateChild',
-    }
-
     def execute(self):
-        return self.check_critical_groups_ou_privileges()
-
-    def check_critical_groups_ou_privileges(self):
         results = {}
-
-        try:
-            ou_privileges = self.get_bad_successor_ou_privileges()
-
-            for record in ou_privileges:
-                if not record:
-                    continue
-
-                entity_sid = record.get('entity_sid', '')
-                rel_type = record.get('relationship_type', '')
-                target_name = record.get('target_name', record.get('ou_name', ''))
-
-                for sid_pattern, group_display in self.CRITICAL_GROUPS_BY_SID.items():
-                    if entity_sid and self._matches_sid_pattern(entity_sid, sid_pattern):
-                        if rel_type in self.DANGEROUS_PERMISSIONS:
-                            description = f"{group_display} HAS {rel_type.upper()} ON OU: {target_name}"
-                            results[description] = self.finding("", inline=True)
-                        break  # Only match one group per entity
-
-        except Exception:
-            pass
-
+        for record in self.get_bad_successor_ou_privileges():
+            if not record:
+                continue
+            entity_name = record.get('entity_name') or record.get('entity_id') or 'Unknown'
+            rel_type = record.get('relationship_type') or ''
+            target_type = record.get('target_type', 'OU')
+            target_name = record.get('target_name') or record.get('target_id') or 'Unknown'
+            description = f"{entity_name} HAS {rel_type.upper()} ON {target_type.upper()}: {target_name}"
+            results[description] = self.finding("", inline=True)
         return results
-
-    def _matches_sid_pattern(self, entity_sid, pattern):
-        if not entity_sid or not pattern:
-            return False
-
-        if pattern.startswith('S-1-'):
-            return entity_sid.endswith(pattern)
-        elif pattern.startswith('-'):
-            if not entity_sid.endswith(pattern):
-                return False
-            prefix_pos = len(entity_sid) - len(pattern)
-            if prefix_pos <= 0:
-                return False
-            preceding_char = entity_sid[prefix_pos - 1]
-            return preceding_char.isdigit() or preceding_char == '-'
-        else:
-            return pattern in entity_sid

checks/core/mssql_common.py

@@ -68,10 +68,10 @@ def linked_server_target_resolution(real_var, stub_var):
     )
 
 
-def host_sid_resolves_to_single_server(sid_expr):
-    # Avoid bridging ambiguous host SID prefixes shared by sibling instances.
+def host_sid_resolves_to_single_server(sid_expr, source_var):
     return (
-        f"COUNT {{ MATCH (sidServer:MSSQL_Server) "
-        f"WHERE sidServer.objectid CONTAINS ':' "
-        f"AND split(sidServer.objectid, ':')[0] = {sid_expr} }} = 1"
+        f"NOT EXISTS {{ MATCH (sidDup:MSSQL_Server) "
+        f"WHERE sidDup <> {source_var} "
+        f"AND sidDup.objectid CONTAINS ':' "
+        f"AND split(sidDup.objectid, ':')[0] = {sid_expr} }}"
     )

checks/mssql_privilege_escalation.py

@@ -292,7 +292,7 @@ def _fetch_linked_server_paths(self):
                OR (stub.objectid IS NOT NULL AND srvA.objectid IS NOT NULL
                    AND stub.objectid CONTAINS ':' AND srvA.objectid CONTAINS ':'
                    AND split(stub.objectid, ':')[0] = split(srvA.objectid, ':')[0]
-                   AND {host_sid_resolves_to_single_server("split(srvA.objectid, ':')[0]")})
+                   AND {host_sid_resolves_to_single_server("split(srvA.objectid, ':')[0]", "srvA")})
 
             MATCH (srvB:MSSQL_Server)-[:MSSQL_Contains]->(target:MSSQL_ServerRole)
             WHERE {linked_server_target_resolution('srvB', 'srvB_stub')}

checks/sccm_elevate1.py

@@ -22,6 +22,7 @@ def execute(self):
         rows = self.query(f"""
             MATCH (g)-[:CoerceAndRelayToSMB]->(target:Computer)
             WHERE target.SMBSigningRequired = false
+            AND coalesce(target.enabled, true) = true
             AND target.SCCMSiteSystemRoles IS NOT NULL {cdf}
 
             UNWIND target.SCCMSiteSystemRoles as role
@@ -30,6 +31,7 @@ def execute(self):
 
             OPTIONAL MATCH (ss:Computer)
             WHERE ss.SCCMSiteSystemRoles IS NOT NULL
+            AND coalesce(ss.enabled, true) = true
             AND any(r IN ss.SCCMSiteSystemRoles WHERE r = 'SMS Site Server@' + site_code)
             AND ss.objectid <> target.objectid
 

checks/sccm_privilege_escalation.py

@@ -380,7 +380,7 @@ def _fetch_linked_sccm_paths(self):
                OR (stub.objectid IS NOT NULL AND srvA.objectid IS NOT NULL
                    AND stub.objectid CONTAINS ':' AND srvA.objectid CONTAINS ':'
                    AND split(stub.objectid, ':')[0] = split(srvA.objectid, ':')[0]
-                   AND {host_sid_resolves_to_single_server("split(srvA.objectid, ':')[0]")})
+                   AND {host_sid_resolves_to_single_server("split(srvA.objectid, ':')[0]", "srvA")})
 
             MATCH (srvB:MSSQL_Server)
             WHERE {linked_server_target_resolution('srvB', 'srvB_stub')}
@@ -506,7 +506,7 @@ def _fetch_linked_execute_host_sccm_paths(self):
                   AND stub.objectid CONTAINS ':'
                   AND src.objectid CONTAINS ':'
                   AND split(stub.objectid, ':')[0] = split(src.objectid, ':')[0]
-                  AND {host_sid_resolves_to_single_server("split(src.objectid, ':')[0]")}
+                  AND {host_sid_resolves_to_single_server("split(src.objectid, ':')[0]", "src")}
                 MATCH (linked:MSSQL_Server)
                 WHERE linked = linkedStub
                    OR {linked_server_target_resolution('linked', 'linkedStub')}

checks/sccm_takeover1.py

@@ -23,17 +23,24 @@ class SCCMTakeover1Check(SCCMDomainMixin, Check):
     )
 
     def execute(self):
+        source_domain_cond = self._domain_condition("g")
         sdf = self._site_domain_condition()
         rows = self.query(f"""
             MATCH (g)-[relay:CoerceAndRelayToMSSQL]->(login)
                   -[:MSSQL_MemberOf|MSSQL_IsMappedTo|MSSQL_ControlServer
                     |MSSQL_ControlDB|MSSQL_Contains
                     |SCCM_AssignAllPermissions*1..6]->(site:SCCM_Site)
-            WHERE true {sdf}
+            WHERE (
+                ((g:User OR g:Computer OR g:Group){source_domain_cond})
+                OR NOT (g:User OR g:Computer OR g:Group)
+              )
+              AND (NOT g:Computer OR coalesce(g.enabled, true) = true)
+              AND true{sdf}
 
             OPTIONAL MATCH (login)-[:MSSQL_MemberOf]->(:MSSQL_ServerRole)
                            -[:MSSQL_ControlServer]->(server:MSSQL_Server)
             OPTIONAL MATCH (target:Computer)-[:MSSQL_HostFor]->(server)
+            WHERE coalesce(target.enabled, true) = true
 
             RETURN DISTINCT
                 site.siteCode as site_code,

checks/sccm_takeover2.py

@@ -27,10 +27,12 @@ def execute(self):
             MATCH (g)-[:CoerceAndRelayToSMB]->(target:Computer)
                   -[:MSSQL_HostFor]->(:MSSQL_Server)-[:MSSQL_Contains]->(db:MSSQL_Database)
                   -[:SCCM_AssignAllPermissions]->(site:SCCM_Site)
-            WHERE target.SMBSigningRequired = false {cdf}
+            WHERE target.SMBSigningRequired = false
+            AND coalesce(target.enabled, true) = true {cdf}
 
             OPTIONAL MATCH (ss:Computer)
             WHERE ss.SCCMSiteSystemRoles IS NOT NULL
+            AND coalesce(ss.enabled, true) = true
             AND any(role IN ss.SCCMSiteSystemRoles WHERE role = 'SMS Site Server@' + site.siteCode)
             AND ss.objectid <> target.objectid
 

checks/sccm_takeover4.py

@@ -44,13 +44,15 @@ def _fetch_paths(self):
 
             OPTIONAL MATCH (cas_comp:Computer)
             WHERE cas_comp.SCCMSiteSystemRoles IS NOT NULL
+            AND coalesce(cas_comp.enabled, true) = true
             AND any(role IN cas_comp.SCCMSiteSystemRoles
                     WHERE role = 'SMS Site Server@' + parent_site.siteCode)
             WITH parent_site, child_site,
                  collect(DISTINCT {self.host_name_expr("cas_comp")}) as cas_servers
 
             MATCH (child_comp:Computer)
             WHERE child_comp.SCCMSiteSystemRoles IS NOT NULL
+            AND coalesce(child_comp.enabled, true) = true
             AND any(role IN child_comp.SCCMSiteSystemRoles
                     WHERE role = 'SMS Site Server@' + child_site.siteCode)
             AND child_comp.SMBSigningRequired = false

checks/sccm_takeover5.py

@@ -41,6 +41,7 @@ def _fetch_paths(self):
 
             MATCH (ss_comp:Computer)
             WHERE ss_comp.SCCMSiteSystemRoles IS NOT NULL
+            AND coalesce(ss_comp.enabled, true) = true
             AND any(role IN ss_comp.SCCMSiteSystemRoles
                     WHERE role = 'SMS Site Server@' + site.siteCode)
             WITH site,
@@ -49,6 +50,7 @@ def _fetch_paths(self):
 
             MATCH (sp_comp:Computer)
             WHERE sp_comp.SCCMSiteSystemRoles IS NOT NULL
+            AND coalesce(sp_comp.enabled, true) = true
             AND any(role IN sp_comp.SCCMSiteSystemRoles
                     WHERE role = 'SMS Provider@' + site.siteCode)
             WITH site, site_server_sids, site_servers,

checks/sccm_takeover6.py

@@ -21,11 +21,13 @@ def execute(self):
 
         rows = self.query(f"""
             MATCH (g)-[:CoerceAndRelayToSMB]->(target:Computer)-[:SCCM_AssignAllPermissions]->(site:SCCM_Site)
-            WHERE target.SMBSigningRequired = false {cdf}
+            WHERE target.SMBSigningRequired = false
+            AND coalesce(target.enabled, true) = true {cdf}
             AND any(role IN target.SCCMSiteSystemRoles WHERE role = 'SMS Provider@' + site.siteCode)
 
             OPTIONAL MATCH (ss:Computer)
             WHERE ss.SCCMSiteSystemRoles IS NOT NULL
+            AND coalesce(ss.enabled, true) = true
             AND any(role IN ss.SCCMSiteSystemRoles WHERE role = 'SMS Site Server@' + site.siteCode)
             AND ss.objectid <> target.objectid
 

checks/sccm_takeover7.py

@@ -41,7 +41,8 @@ def _fetch_paths(self):
 
         rows = self.query(f"""
             MATCH (role_comp:Computer)
-            WHERE role_comp.SCCMSiteSystemRoles IS NOT NULL{comp_filter}
+            WHERE role_comp.SCCMSiteSystemRoles IS NOT NULL
+            AND coalesce(role_comp.enabled, true) = true{comp_filter}
             UNWIND role_comp.SCCMSiteSystemRoles as role
             WITH role_comp,
                  trim(split(role, '@')[0]) as role_name,

checks/sccm_takeover8.py

@@ -42,6 +42,7 @@ def execute(self):
             WHERE
                 ((dc.ldapavailable = true AND dc.ldapsigning = false)
                  OR (dc.ldapsavailable = true AND dc.ldapsepa = false))
+                AND coalesce(dc.enabled, true) = true
                 {cdf}
             WITH dc,
                  (dc.ldapavailable = true AND dc.ldapsigning = false) AS ldap_relay,
@@ -51,6 +52,7 @@ def execute(self):
             WHERE coalesce(src.webclientrunning, src.WebClientRunning, false) = true
               AND src.domain = dc.domain
               AND src.SCCMSiteSystemRoles IS NOT NULL
+              AND coalesce(src.enabled, true) = true
               AND (
                 any(role IN src.SCCMSiteSystemRoles WHERE role STARTS WITH 'SMS Site Server@')
                 OR any(role IN src.SCCMSiteSystemRoles WHERE role STARTS WITH 'SMS Provider@')

checks/tier0_session_exposure.py

@@ -11,26 +11,23 @@ def execute(self):
         if not self.neo4j_data:
             return {}
 
-        try:
-            rows = self.neo4j_data.conn.query("""
-                MATCH (c:Computer)-[:HasSession]->(u:User)
-                WHERE u.system_tags CONTAINS 'admin_tier_0'
-                  AND (c.system_tags IS NULL OR NOT c.system_tags CONTAINS 'admin_tier_0')
-                  AND u.enabled = true
-                  AND c.enabled = true
-                RETURN u.objectid AS sid,
-                       collect(DISTINCT c.name) AS hosts
-            """, name="tier0_session_exposure")
+        rows = self.neo4j_data.conn.query("""
+            MATCH (c:Computer)-[:HasSession]->(u:User)
+            WHERE u.system_tags CONTAINS 'admin_tier_0'
+              AND (c.system_tags IS NULL OR NOT c.system_tags CONTAINS 'admin_tier_0')
+              AND u.enabled = true
+              AND c.enabled = true
+            RETURN u.objectid AS sid,
+                   collect(DISTINCT c.name) AS hosts
+        """, name="tier0_session_exposure")
 
-            findings = {}
-            for row in rows:
-                sid = row.get('sid')
-                hosts = sorted(row.get('hosts') or [])
-                if sid and hosts:
-                    findings[sid] = self.finding(
-                        f"Active session on: {', '.join(hosts)}",
-                        details={"hosts": hosts},
-                    )
-            return findings
-        except Exception:
-            return {}
+        findings = {}
+        for row in rows:
+            sid = row.get('sid')
+            hosts = sorted(row.get('hosts') or [])
+            if sid and hosts:
+                findings[sid] = self.finding(
+                    f"Active session on: {', '.join(hosts)}",
+                    details={"hosts": hosts},
+                )
+        return findings

modules/neo4j_data.py

@@ -1020,40 +1020,28 @@ def get_bad_successor_ou_privileges(self, force_refresh=False):
         if not force_refresh and hasattr(self, 'bad_successor_cache') and self.bad_successor_cache is not None:
             return self.bad_successor_cache
 
-        # no 2025 DC = no BadSuccessor; skip the OU scan
-        dc_check_query = """
-        MATCH (dc:Computer)-[:MemberOf*1..]->(g:Group)
-        WHERE dc.operatingsystem =~ '(?i).*WINDOWS SERVER 2025.*'
-          AND g.name =~ '(?i).*DOMAIN CONTROLLERS.*'
-          AND dc.enabled = true
-        RETURN count(DISTINCT dc) as dc_count
-        """
-
-        dc_result = self.conn.query(dc_check_query, name="get_bad_successor_dc_check")
-
-        if not dc_result or not dc_result[0] or dc_result[0].get('dc_count', 0) == 0:
-            self.bad_successor_cache = []
-            return self.bad_successor_cache
+        dc_domain = self._domain_condition("dc")
+        target_domain = self._domain_condition("target")
 
-        badsuccessor_query = """
+        badsuccessor_query = f"""
         MATCH (dc:Computer)-[:MemberOf*1..]->(g:Group)
         WHERE dc.operatingsystem =~ '(?i).*WINDOWS SERVER 2025.*'
           AND g.name =~ '(?i).*DOMAIN CONTROLLERS.*'
-          AND dc.enabled = true
+          AND dc.enabled = true{dc_domain}
         WITH count(DISTINCT dc) > 0 AS has2025DC
         WHERE has2025DC = true
 
-        // Query for OU privileges only - BadSuccessor focuses on OU control
-        MATCH p = (ou:OU)<-[r:WriteDacl|Owns|GenericAll|WriteOwner]-(n:Base)
-        WHERE NOT ((n:Tag_Tier_Zero) OR COALESCE(n.system_tags, '') CONTAINS 'admin_tier_0')
-        RETURN 'OU' AS target_type,
-               ou.name AS target_name,
-               n.name AS entity_name,
-               n.objectid AS entity_sid,
-               labels(n) AS entity_labels,
-               type(r) AS relationship_type,
-               null AS gpo_linked_domain,
-               ou.name AS ou_name
+        MATCH (target)<-[r:WriteDacl|Owns|GenericAll|WriteOwner]-(n)
+        WHERE (target:OU OR target:Container){target_domain}
+          AND (n:User OR n:Computer OR n:Group)
+          AND NOT ((n:Tag_Tier_Zero) OR COALESCE(n.system_tags, '') CONTAINS 'admin_tier_0')
+          AND COALESCE(n.enabled, true) = true
+        RETURN CASE WHEN target:OU THEN 'OU' ELSE 'Container' END AS target_type,
+               coalesce(target.name, target.objectid, 'Unknown') AS target_name,
+               coalesce(n.name, n.objectid, 'Unknown') AS entity_name,
+               target.objectid AS target_id,
+               n.objectid AS entity_id,
+               type(r) AS relationship_type
         """
 
         result = self.conn.query(badsuccessor_query, name="get_bad_successor_ou_privileges")

tests/checks/test_bad_successor.py

@@ -6,13 +6,42 @@
 pytestmark = pytest.mark.neo4j
 
 
-def test_fires_on_default_group_with_ou_privilege(clean_neo4j):
+def test_flags_non_tier0_control_of_ous_and_containers(clean_neo4j):
     load_fixture(clean_neo4j, "bad_successor.cypher")
 
     findings = run_check(BadSuccessorCheck, clean_neo4j, domain_filter="TEST.LOCAL")
     assert BadSuccessorCheck.RISK_LEVEL == "High"
-    assert len(findings) == 1
+    assert len(findings) == 2
 
-    key = list(findings.keys())[0]
-    assert "DOMAIN USERS" in key and "GENERICALL" in key
-    assert isinstance(findings[key], dict) and "inline_description" in findings[key]
+    assert "DOMAIN USERS@TEST.LOCAL HAS GENERICALL ON OU: WORKSTATIONS@TEST.LOCAL" in findings
+    assert ("HELPDESK@TEST.LOCAL HAS WRITEDACL ON CONTAINER: "
+            "MANAGED SERVICE ACCOUNTS@TEST.LOCAL") in findings
+
+    keys = "\n".join(findings)
+    assert "STALE" not in keys          # disabled principal excluded
+    assert "DOMAIN ADMINS" not in keys  # Tier-0 principal excluded
+    assert "OTHER.LOCAL" not in keys    # different-domain targets excluded
+    assert "LinkedServer" not in keys    # non-principal Base/OpenGraph nodes excluded
+
+    sample = findings["DOMAIN USERS@TEST.LOCAL HAS GENERICALL ON OU: WORKSTATIONS@TEST.LOCAL"]
+    assert isinstance(sample, dict) and "inline_description" in sample
+
+
+def test_falls_back_to_objectids_for_missing_names(clean_neo4j):
+    load_fixture(clean_neo4j, "bad_successor_fallback_names.cypher")
+
+    findings = run_check(BadSuccessorCheck, clean_neo4j, domain_filter="TEST.LOCAL")
+    keys = "\n".join(findings)
+
+    assert len(findings) == 3
+    assert "None" not in keys
+    assert "S-1-5-21-TEST-2001 HAS GENERICALL ON OU: WORKSTATIONS@TEST.LOCAL" in findings
+    assert "S-1-5-21-TEST-2002 HAS GENERICALL ON OU: WORKSTATIONS@TEST.LOCAL" in findings
+    assert "NAMED@TEST.LOCAL HAS GENERICALL ON OU: TEST-OU-NONAME" in findings
+
+
+def test_requires_2025_dc_in_filtered_domain(clean_neo4j):
+    load_fixture(clean_neo4j, "bad_successor_foreign_dc.cypher")
+
+    findings = run_check(BadSuccessorCheck, clean_neo4j, domain_filter="TEST.LOCAL")
+    assert findings == {}