Skip to content

Framework Guide

Jump to bottom
danti1988 edited this page May 18, 2026 · 4 revisions

Vulnerability Check Framework Guide

ADPathfinder checks have two moving parts:

  • DataSources answer whether optional data is present, such as MSSQLHound or ConfigManBearPig OpenGraph data.
  • Checks run the analysis and return findings.

AD checks do not need a datasource. They use preloaded users, computers, enterprise CAs, path caches, or password analysis data. Platform checks usually run their own Cypher and use a datasource gate so they are skipped cleanly when the matching data is not in the graph.

Writing an AD Check

Use this pattern when the check reads properties already loaded from BloodHound user or computer nodes.

from checks.core import Check, check


@check(
    risk="High",
    category="Computer with Unconstrained Delegation",
    entity="computer",
    data=["computers"],
)
class ComputerUnconstrainedDelegationCheck(Check):

    def execute(self):
        def check_delegation(computer):
            if computer.get("unconstrainedDelegation") is True:
                return self.finding("Unconstrained delegation enabled")
            return None

        return self.process_entity_results(self.get_computers(), check_delegation)

process_entity_results(...) filters disabled accounts, skips domain controllers for computer checks, maps findings by SID, and applies your callback to each remaining entity.

@check() Parameters

Parameter Required Description
risk Yes One of "Critical", "High", "Medium", or "Low"
category Yes Display name for the finding
entity No "computer" or "user"
data No Preloaded data the check needs: "computers", "users", "enterprise_cas", "bad_successor_ou_privileges", "escalation_paths", "full_escalation_paths", "weak_password", or "admin_privileges"
display No "simple" by default, or "escalation_paths", "grouped_escalation_paths", "group_analysis"
requires No Datasource names such as ["mssql"] or ["sccm"]
section No Groups output under a platform section

The registry validates data=[...] against the supported data types. If a check needs password analysis helpers such as has_shared_password(...), make sure it is only useful in password-audit runs where account analysis is available.

Writing a Platform Check

Use this pattern for MSSQL, SCCM, or similar OpenGraph data. The check runs Cypher directly with self.query(...) and declares requires=[...] so the framework can skip it when the collector data is absent.

from checks.core import Check, check
from checks.core.platform_mixins import MSSQLDomainMixin


@check(
    risk="Medium",
    category="MSSQL Linked Servers",
    entity="computer",
    data=[],
    requires=["mssql"],
    section="mssql",
)
class MSSQLLinkedServersCheck(MSSQLDomainMixin, Check):

    def execute(self):
        domain_filter = self._server_domain_condition("source.name")
        where = f"\nWHERE true{domain_filter}" if domain_filter else ""

        rows = self.query(f"""
            MATCH (source:MSSQL_Server)-[r:MSSQL_LinkedTo|MSSQL_LinkedAsAdmin]->(target)
            MATCH (host:Computer)-[:MSSQL_HostFor]->(source){where}
            RETURN host.objectid AS hostSid,
                   target.name AS targetServer,
                   r.localLogin AS localLogin,
                   r.remoteCurrentLogin AS remoteLogin
        """, name="mssql_linked_servers")

        results = {}
        for row in rows:
            host_sid = row.get("hostSid")
            if not host_sid:
                continue
            results.setdefault(host_sid, []).append(
                f"{row.get('localLogin') or 'unknown'} -> {row.get('targetServer') or 'unknown'}"
            )

        return {
            sid: self.finding("\n".join(sorted(lines)))
            for sid, lines in results.items()
        }

The important differences from an AD check are:

  • data=[] because the query does its own graph lookup.
  • requires=["mssql"] or requires=["sccm"] so the check is gated by datasource availability.
  • Domain filtering is applied inside the Cypher query.
  • The check builds its own result dictionary rather than using process_entity_results(...).

Domain Filter Helpers

Helper Use case
_domain_condition(var_name) AD entity nodes with a domain property
MSSQLDomainMixin._server_domain_condition(server_field) MSSQL server names where the domain is derived from the FQDN
SCCMDomainMixin._site_domain_condition(site_var) SCCM sites scoped by sourceForest

All helpers read self._domain_filter, which is set from the current Neo4j data context.

Writing an ADCS Check

ADCS checks extend ADCSCheck, which provides the target group list, a template query wrapper, and shared result formatting.

from checks.core import check
from checks.adcs_base import ADCSCheck


@check(risk="High", category="ESC Example - Template Misconfiguration")
class ESCExampleCheck(ADCSCheck):

    def execute(self):
        target_groups = self._get_target_groups()
        if not target_groups:
            return {}

        rows = self._query_templates("""
            MATCH (g:Group)-[:Enroll|AutoEnroll]->(t:CertTemplate)-[:PublishedTo]->(ca:EnterpriseCA)
            WHERE t.some_flag = true
              AND g.name IN $target_groups
            RETURN t.name AS template,
                   ca.caname AS ca_name,
                   ca.dnshostname AS ca_host,
                   collect(DISTINCT g.name) AS abusers
            ORDER BY template
        """, {"target_groups": target_groups})

        return self._format_results(rows) if rows else {}

ADCSCheck supplies:

  • _get_target_groups() - low-privileged groups for the current domain.
  • _query_templates(cypher, parameters) - runs Cypher through self.query(...).
  • _format_results(rows) - turns template, CA, and abuser rows into findings.

You normally do not need entity= or data= on an ADCS check because the base class sets safe defaults.

Writing an OpenGraph Check

For a one-off OpenGraph check, you can query the collector's labels directly. If the graph does not contain those labels, the query returns no rows.

from checks.core import Check, check


@check(risk="High", category="GitHub Repo Public Access", entity="computer", data=[])
class GitHubPublicRepoCheck(Check):

    def execute(self):
        rows = self.query("""
            MATCH (r:GitHubRepo)
            WHERE r.visibility = 'public'
            RETURN r.objectid AS id, r.name AS name
        """, name="github_public_repos")

        return {
            row["id"]: self.finding(f"Public repository: {row['name']}")
            for row in rows
            if row.get("id")
        }

For a platform with several checks, add a datasource gate so all related checks are skipped together when the data is absent.

Adding a Datasource

A datasource lives under datasources/ and registers with @datasource("<name>").

from checks.core import DataSource, datasource


@datasource("github")
class GitHubAvailability(DataSource):

    def available(self):
        rows = self.query(
            "MATCH (n:GitHubRepo) RETURN count(n) AS c",
            name="github_availability",
        )
        return rows[0]["c"] > 0 if rows else False

Checks then declare the dependency:

@check(risk="High", category="GitHub Repo Public Access", entity="computer", data=[], requires=["github"])
class GitHubPublicRepoCheck(Check):
    ...

Datasource files in datasources/ are imported at startup.

Create a datasource when several checks share the same platform, when availability is expensive to detect, or when missing data would otherwise create noisy warnings. For a single cheap query, it is fine to handle an empty result in execute().

Check API Reference

Data Access

Method Returns
self.get_computers() Preloaded computer entities
self.get_users() Preloaded user entities
self.get_enterprise_cas() Preloaded enterprise CAs
self.get_bad_successor_ou_privileges() Preloaded BadSuccessor OU privilege rows
self.query(cypher, parameters=None, name=None) Cypher results as a list of dictionaries, never None

Helpers

Method Notes
self.finding(description="") Standard finding string
self.finding(description, details=dict) Finding with structured details
self.finding(description, inline=True) Returns {"inline_description": description} for compact rendering
self.process_entity_results(entities, check_fn) Filters entities, applies a callback, and maps results by SID
self.has_escalation_path(entity_or_sid) Reads the preloaded escalation path cache
self.has_weak_password(entity_or_sid) Used in password-audit checks; conventionally paired with data=["weak_password"]
self.is_admin(entity_or_sid) Used in password-audit checks; conventionally paired with data=["admin_privileges"] when the check depends on admin status
self.has_shared_password(entity_or_sid) Uses shared-password analysis context

Writing a Cross-Domain Check

Cross-domain checks compare data across multiple domains — shared passwords, trust relationships, or escalation paths that cross a domain boundary. They live under checks/cross_domain/ and use a separate registry.

from checks.cross_domain.base import CrossDomainCheck, CrossDomainRegistry


@CrossDomainRegistry.register
class ExampleCrossDomainCheck(CrossDomainCheck):
    RISK_LEVEL = "High"
    CATEGORY_NAME = "Example Cross-Domain Finding"
    REQUIRES_NTDS = False

    def execute(self):
        findings = []
        for domain in self.deps.all_domains:
            # Compare across domains using self.deps
            pass
        return findings

Key differences from a normal check:

  • Subclass CrossDomainCheck, not Check.
  • Register with @CrossDomainRegistry.register instead of @check(...).
  • Set RISK_LEVEL and CATEGORY_NAME as class attributes.
  • Set REQUIRES_NTDS = True if the check needs NTDS hash data — the framework skips it automatically when hashes are unavailable.
  • Access shared state through self.deps, which carries the Neo4j connection, domain list, per-domain hashes, cracked hashes, admin user lists, and the relationship pattern string.

See checks/cross_domain/admin_weak_crossdomain.py for a complete example.

Testing

Install the test extras first:

pip install -e ".[test]"

After adding a check file, make sure it imports and registers:

python -c "import modules.main, checks, datasources"
python -c "from checks.core.registry import CheckRegistry; print(len(CheckRegistry.get_all_checks()))"

Run the fast suite:

pytest tests/ -m "not neo4j and not integration"

Run fixture tests against a disposable Neo4j instance:

ADPF_TEST_ALLOW_WIPE=1 \
NEO4J_URI=bolt://localhost:17687 \
NEO4J_USER=neo4j NEO4J_PASSWORD=testpassword \
pytest tests/checks -m "neo4j and not integration" -v

The full container setup and teardown commands are in tests/README.md.

For a lab smoke test, run:

adpathfinder --ad --diagnostics

Clone this wiki locally