Framework Guide

Vulnerability Check Framework Guide

Overview

Two concepts: DataSources check what data is available, Checks run queries and analyse results.

• A DataSource registers a platform (e.g. mssql, sccm) and answers one question: is this data present in the graph?
• A Check declares what data it needs, runs its own Cypher query, and returns findings keyed by SID.
• The framework skips checks whose DataSource isn't available — no boilerplate in the check itself.

AD checks don't need a DataSource. They use preloaded entity data (get_users(), get_computers()) which is always available.

Writing an AD Check

The simplest pattern. Preloaded data, entity filtering, SID mapping — all handled by the base class.

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 and domain controllers, maps SIDs, and calls your function for each entity.

@check() parameters

Parameter	Required	Description
risk	Yes	"Critical", "High", "Medium", "Low"
category	Yes	Display name for the finding
entity	No	"computer" or "user"
data	No	List of preloaded data types needed: "computers", "users", "enterprise_cas", "escalation_paths", "full_escalation_paths", "weak_password", "admin_privileges", "bad_successor_ou_privileges"
display	No	"simple" (default), "escalation_paths", "grouped_escalation_paths", "group_analysis"
requires	No	List of DataSource names: ["mssql"], ["sccm"]
section	No	Groups checks under a platform heading in output

Writing a Platform Check (MSSQL, SCCM)

Platform checks run their own Cypher queries via self.query(). Domain filtering comes from mixins. The requires= parameter gates the check on a DataSource.

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):
        server_filter = ""
        if self._domain_filter:
            server_filter = '\nWHERE true' + self._server_domain_condition("source.name")

        rows = self.query("""
        MATCH (source:MSSQL_Server)-[r:MSSQL_LinkedTo]->(target)""" + server_filter + """
        RETURN source.name as sourceServer,
               target.name as targetServer,
               r.localLogin as localLogin,
               r.remoteCurrentLogin as remoteLogin
        """, name="mssql_linked_servers")

        if not rows:
            return {}

        results = {}
        for row in rows:
            # ... process rows, map to SIDs, build findings
            pass
        return results

Key differences from AD checks:

• Inherits MSSQLDomainMixin for _server_domain_condition()
• data=[] — no preloaded entity data needed
• requires=["mssql"] — skipped entirely if no MSSQL data in graph
• Uses self.query() to run Cypher directly
• Handles its own result processing (no process_entity_results)

Domain Filter Mixins

Mixin	Method	Use case
(base class)	_domain_condition(var_name)	AD entity nodes — filters by node.domain
MSSQLDomainMixin	_server_domain_condition(server_field)	MSSQL servers — extracts domain from FQDN
SCCMDomainMixin	_site_domain_condition(site_var)	SCCM sites — scoped by edge endpoints

All read from self._domain_filter which is set automatically from the Neo4j connection's domain filter.

Writing an ADCS Check

ADCS checks extend ADCSCheck, which provides shared Cypher helpers and 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):
        if not self.neo4j_data:
            return {}
        target_groups = self._get_target_groups()
        if not target_groups:
            return {}
        rows = self._query_templates("""
            MATCH (g:Group)-[:Enroll]->(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 provides:

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

No entity or data in @check() — defaults come from ADCSCheck base class.

Writing an OpenGraph Check

BloodHound OpenGraph supports arbitrary node labels and edge types defined by each collector. The pattern is identical to platform checks — the check runs its own query against whatever labels the collector creates.

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")

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

No special base class needed. If the graph has GitHubRepo nodes, the query returns data. If not, the check returns empty.

For a new platform with many checks, register a DataSource so the framework can skip all of them at once when the data isn't present (see below).

Adding a New Platform (DataSource)

A DataSource tells the framework whether a platform's data exists in the graph. This avoids running queries against nodes that don't exist.

# datasources/github.py
from checks.core import DataSource, datasource

@datasource("github")
class GitHubAvailability(DataSource):
    def available(self):
        result = self.query("MATCH (n:GitHubRepo) RETURN count(n) as c")
        return result[0]['c'] > 0 if result else False

Then checks declare the dependency:

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

DataSource files in datasources/ are auto-discovered at startup.

When to create a DataSource

• Multiple checks for the same platform — one availability check, not one per check
• Expensive-to-discover data — cache the availability answer for the run
• Single check — don't bother, just handle empty results 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.query(cypher, parameters=None, name=None)	Run Cypher directly, returns list of rows (never None)

Domain Filtering

Property/Method	Description
self._domain_filter	Current domain string, or None
self._domain_condition(var_name)	AD entity filter: AND var.domain IN [...]

Platform mixins add _server_domain_condition() and _site_domain_condition().

Entity Checking

Method	Description
self.has_weak_password(entity)	Requires "weak_password" in data
self.is_admin(entity)	Requires "admin_privileges" in data
self.has_escalation_path(entity)	Requires "escalation_paths" in data
self.has_shared_password(entity)	Requires shared password data

Result Methods

Method	Use case
self.finding(description)	Standard finding
self.finding(description, details=dict)	Finding with structured details

Processing

Method	Description
self.process_entity_results(entities, check_fn)	Filters entities, applies check function, maps SIDs

Testing

Checks are auto-discovered via /checks/__init__.py. After creating a check file, verify it registers:

python3 -c "from checks.core.registry import CheckRegistry; print(len(CheckRegistry.get_all_checks()))"

Run the full tool to test against a lab environment:

./adpathfinder.py --ad --diagnostics

Compare against baselines:

python3 tests/compare_baseline.py