Build an Identity Attack Path Engine
Model enterprise access as a graph and detect hidden privilege escalation paths.
Introduction
30 Second Summary
Every employee, contractor, and service account in a company has access to something. Those access paths connect and overlap in ways that create hidden routes an attacker can chain together to reach the most sensitive systems.
In this project, you will build an identity attack path engine that models enterprise access as a graph database and uses anomaly detection to discover hidden privilege escalation routes that traditional permission tables miss entirely.
What You'll Build
A functional identity attack path engine powered by Neo4j and Python that ingests a synthetic enterprise dataset, discovers attack paths using Cypher queries, and flags risky entities with machine learning.
By the end of this project, you'll have:
- A Neo4j graph database populated with a realistic enterprise identity graph containing 50+ entities and 200+ relationships across users, devices, groups, repos, buckets, and SaaS tools.
- Working Cypher queries that discover shortest paths to admin, lateral movement chains, and over-privileged accounts.
- A Python anomaly detection pipeline using graph centrality features and Isolation Forest to flag suspicious access patterns with explainable scores.
- Secret Mission: Add a self-healing policy engine that automatically responds to detected anomalies with simulated remediation actions.
Are there any prerequisites?
You should be familiar with Python and have a basic understanding of identity and access control concepts like users, groups, and permissions. You also need Docker installed on your machine.
Before We Start
Before you can model enterprise access as a graph and hunt for attack paths, you need a few tools in place.
In this step, you will create your project directory, configure Neo4j and Python settings, and install the libraries that power the rest of the project.
In this step, get ready to:
- Create your project directory and configuration files.
- Set up a Python virtual environment with all required dependencies.
Create your project directory
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and move into it by running these commands:
mkdir identity-attack-graph
cd identity-attack-graph- Confirm the folder exists by running:
lsYou should see the identity-attack-graph folder listed in the output.
Add the configuration files
Your project needs three configuration files. Each one handles a different part of the setup.
- Create the three files by running:
touch docker-compose.yml requirements.txt config.py- Confirm the files exist by running:
lsYou should see docker-compose.yml, requirements.txt, and config.py in the output.
- Open the identity-attack-graph folder in your code editor. If you use VS Code, you can run:
code .- Open docker-compose.yml and paste the following:
services:
neo4j:
image: neo4j:2026.05.0
ports:
- "7474:7474"
- "7687:7687"
environment:
- NEO4J_AUTH=neo4j/securegraph123
volumes:
- neo4j_data:/data
restart: always
volumes:
neo4j_data:What does this file do?
- image: neo4j:2026.05.0 pulls the Neo4j Community Edition container image.
- Ports 7474 (browser UI) and 7687 (Bolt protocol for Python connections) are mapped to your machine.
- NEO4J_AUTH sets the database credentials to neo4j/securegraph123.
- The neo4j_data volume persists your graph data between container restarts.
- Save docker-compose.yml.
File looks different from the example?
YAML is sensitive to indentation. Make sure you use spaces (not tabs) and that each nested level is indented by exactly 2 spaces.
help me fix my docker-compose.yml indentation
✔️ Awesome, I've got everything!
Great. Make sure you have saved docker-compose.yml before moving on.
ⓧ I'd like to double check the full code
services:
neo4j:
image: neo4j:2026.05.0
ports:
- "7474:7474"
- "7687:7687"
environment:
- NEO4J_AUTH=neo4j/securegraph123
volumes:
- neo4j_data:/data
restart: always
volumes:
neo4j_data:- Open requirements.txt and paste the following:
neo4j==6.2.0
scikit-learn==1.6.1
numpy==2.2.6What does this file do?
This file pins the exact versions of each library your project needs. neo4j is the Python driver for connecting to Neo4j. scikit-learn provides the Isolation Forest algorithm for anomaly detection. numpy handles the numerical arrays that scikit-learn operates on.
- Save requirements.txt.
✔️ Awesome, I've got everything!
Great. Make sure you have saved requirements.txt before moving on.
ⓧ I'd like to double check the full code
neo4j==6.2.0
scikit-learn==1.6.1
numpy==2.2.6- Open config.py and paste the following:
NEO4J_URI = "neo4j://localhost:7687"
NEO4J_USER = "neo4j"
NEO4J_PASSWORD = "securegraph123"
NEO4J_DATABASE = "neo4j"What does this file do?
This centralizes your Neo4j connection details so every Python script in the project imports from one place. The password matches what you set in docker-compose.yml.
- Save config.py.
Seeing a ModuleNotFoundError later?
If future scripts cannot find config, make sure config.py is in the same directory where you run your Python scripts (the identity-attack-graph folder).
help me debug a ModuleNotFoundError for config
✔️ Awesome, I've got everything!
Great. Make sure you have saved config.py before moving on.
ⓧ I'd like to double check the full code
NEO4J_URI = "neo4j://localhost:7687"
NEO4J_USER = "neo4j"
NEO4J_PASSWORD = "securegraph123"
NEO4J_DATABASE = "neo4j"Set up your Python environment
Your project requires Python 3.10 or higher. Let's verify your installation first.
- Check your Python version by running:
python3 --version✔️ I see version 3.10 or higher
You are good to go. Move on to the next instruction below.
ⓧ I see an older version
Your system's default Python is too old for this project. macOS often ships with an older Python that shadows any newer installation.
Install Python 3.12 (which satisfies the 3.10+ requirement) using the official installer from python.org/downloads.
- Download and run the installer for your platform.
- On Windows, check the Add Python to PATH box during installation.
- After installation, close and reopen your terminal.
- Run python3 --version again to confirm 3.10 or higher.
ⓧ Command not found
Python is not installed yet. Install it from the official source.
- Go to python.org/downloads and download the latest Python 3.12 installer for your operating system.
- Run the installer. On Windows, check the Add Python to PATH box.
- Close and reopen your terminal after installation.
- Run python3 --version to confirm you see 3.10 or higher.
- Create a virtual environment by running:
python3 -m venv venvWhat is a virtual environment?
A virtual environment isolates your project's Python packages from the rest of your system. This prevents version conflicts between projects.
- Activate the virtual environment:
macOS
source venv/bin/activateWindows
venv\Scripts\activateYou should see (venv) appear at the start of your terminal prompt. This confirms the virtual environment is active.
- Install the project dependencies by running:
pip install -r requirements.txtYou should see output showing each package downloading and installing. The final line confirms all three packages are installed successfully.
Installation failed?
Make sure your virtual environment is active. You should see (venv) in your terminal prompt. If not, re-run the activation command above.
If you see a permissions error, do not use sudo. Deactivate, delete the venv folder, and recreate it.
help me fix my pip install error in the identity attack graph project
Your project skeleton is ready. Next up, you will launch Neo4j and populate it with a realistic enterprise access graph.
Set Up Neo4j and Python Environment
Your project files are ready and your Python environment is active. Now it's time to spin up the graph database that will store your enterprise identity graph.
In this step, you will start Neo4j using Docker Compose, verify it is running in your browser, and confirm that Python can communicate with it.
In this step, get ready to:
- Start Neo4j Community Edition via Docker Compose.
- Verify the database is accessible in Neo4j Browser.
- Test Python connectivity with a Cypher query.
Start Neo4j with Docker Compose
Your docker-compose.yml already defines the Neo4j service. You just need to tell Docker to pull the image and start the container.
- Start the Neo4j container in the background by running this command:
docker-compose up -dWhat does this command do?
The -d flag runs the container in detached mode, meaning it runs in the background so you can keep using your terminal.
Docker pulls the neo4j:2026.05.0 image, maps ports 7474 (HTTP) and 7687 (Bolt protocol) to your machine, and sets the initial password from the NEO4J_AUTH environment variable.
- Confirm the container is running by running:
docker-compose psYou should see a single neo4j service listed with a state of Up.
Container not starting?
- Make sure Docker Desktop is running. If you see a connection error, open Docker Desktop first and try again.
- If port 7474 or 7687 is already in use, another process is occupying those ports. Stop the conflicting service or change the port mapping in your docker-compose.yml.
- help me troubleshoot my Docker container not starting
Verify Neo4j in the browser
Neo4j includes a built-in web interface called Neo4j Browser. It takes about 10-15 seconds for Neo4j to finish starting up after the container launches.
- Open your web browser and navigate to:
- http://localhost:7474.
- In the connection frame, confirm the Connect URL is set to neo4j://localhost:7687.
- Enter neo4j as the username.
- Enter securegraph123 as the password.
- Click Connect.
You should see the Neo4j Browser interface with an empty database. No nodes or relationships exist yet.
Why neo4j:// instead of bolt://?
The neo4j:// scheme is the recommended connection protocol for Neo4j 5.x and newer. It supports routing-aware connections, which matters for clustered deployments. For a single local instance it behaves identically to bolt://, but it is the modern default.
Can't reach the browser page?
- Wait 15-20 seconds after starting the container. Neo4j needs time to initialize on first launch.
- Run docker-compose logs neo4j to check the container logs for errors.
- help me connect to Neo4j Browser
Test Python connectivity with a Cypher query
The browser confirms Neo4j is running, but your Python scripts also need to connect. Your config.py already holds the connection credentials. Now you will write a small test script to confirm everything works end-to-end.
- Create a file called test_connection.py in your project folder with the following code:
from neo4j import GraphDatabase
from config import NEO4J_URI, NEO4J_USER, NEO4J_PASSWORD, NEO4J_DATABASE
# Connect to Neo4j using credentials from config
driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASSWORD))
# Verify the connection is working
driver.verify_connectivity()
print("[+] Connected to Neo4j successfully!")
# Run a test Cypher query
records, _, _ = driver.execute_query(
"RETURN 'Identity Attack Path Engine ready!' AS message",
database_=NEO4J_DATABASE,
)
print(f"[+] Test query result: {records[0]['message']}")
# Clean up
driver.close()
print("[+] Connection test complete. Environment is ready.")What does this code do?
- GraphDatabase.driver() creates a connection to Neo4j using the URI and credentials from your config.py file.
- verify_connectivity() checks that the driver can reach the database. It throws an error if the connection fails.
- execute_query() runs a Cypher query (Neo4j's graph query language) and returns the results. The database_ parameter specifies which database to target.
- Save test_connection.py and run it:
python test_connection.pyYou should see three lines of output confirming the connection, the test query result, and that the test is complete.
What should I see?
[+] Connected to Neo4j successfully!
[+] Test query result: Identity Attack Path Engine ready!
[+] Connection test complete. Environment is ready.Seeing a connection error?
- Make sure your virtual environment is active. You should see the environment name in your terminal prompt.
- Confirm the Neo4j container is still running with docker-compose ps. If it restarted, wait 10-15 seconds for it to fully initialize.
- Double-check that config.py has the password securegraph123 matching the NEO4J_AUTH value in docker-compose.yml.
- help me fix my Neo4j Python connection error
Your infrastructure is live and Python can talk to the graph database. Next up, you will design and populate an enterprise access graph with realistic users, devices, groups, and attack scenarios.
Design and Populate the Enterprise Access Graph
Your Neo4j database is running and Python can talk to it. Now you need something to query.
Traditional permission systems store access as flat rows in a table. A user has a role, and that role maps to permissions. But real enterprises are messier. Groups nest inside groups, service accounts inherit permissions transitively, and shared device sessions create invisible lateral pathways. In this step, you will model that complexity as a property graph and inject deliberate attack scenarios that would be invisible in a flat permission table.
In this step, get ready to:
- Define a graph schema with 7 node types and 6 relationship types.
- Write a Python data generator that creates 63 entities and 200+ relationships.
- Inject 3 deliberate attack scenarios that mimic real-world privilege escalation patterns.
Define the graph schema and create constraints
Before writing any data, you need to decide what types of entities exist in your enterprise model and how they connect to each other.
What is a property graph schema?
A property graph stores data as nodes (entities) and relationships (edges between them). Each node has a label (its type) and properties (key-value attributes). Each relationship has a type and can also carry properties.
Your schema defines 7 node types: User, Device, Group, Repo, Bucket, SaaSApp, and ServiceAccount. Relationships include MEMBER_OF, HAS_ACCESS, ADMIN_TO, and HAS_SESSION.
- Create a new file called generate_graph.py in your identity-attack-graph project folder.
- Add the imports, database clearing function, and constraints function by pasting this code:
import random
from neo4j import GraphDatabase
from config import NEO4J_URI, NEO4J_USER, NEO4J_PASSWORD, NEO4J_DATABASE
def clear_database(driver):
driver.execute_query("MATCH (n) DETACH DELETE n", database_=NEO4J_DATABASE)
def create_constraints(driver):
constraints = [
"CREATE CONSTRAINT IF NOT EXISTS FOR (u:User) REQUIRE u.name IS UNIQUE",
"CREATE CONSTRAINT IF NOT EXISTS FOR (g:Group) REQUIRE g.name IS UNIQUE",
"CREATE CONSTRAINT IF NOT EXISTS FOR (d:Device) REQUIRE d.name IS UNIQUE",
"CREATE CONSTRAINT IF NOT EXISTS FOR (r:Repo) REQUIRE r.name IS UNIQUE",
"CREATE CONSTRAINT IF NOT EXISTS FOR (b:Bucket) REQUIRE b.name IS UNIQUE",
"CREATE CONSTRAINT IF NOT EXISTS FOR (s:SaaSApp) REQUIRE s.name IS UNIQUE",
"CREATE CONSTRAINT IF NOT EXISTS FOR (sa:ServiceAccount) REQUIRE sa.name IS UNIQUE",
]
for c in constraints:
driver.execute_query(c, database_=NEO4J_DATABASE)What does this code do?
- clear_database() runs MATCH (n) DETACH DELETE n to wipe all nodes and relationships. This lets you re-run the generator cleanly without duplicates.
- create_constraints() enforces uniqueness on each node type's name property. If you accidentally try to create the same entity twice, Neo4j will reject it rather than silently duplicating data.
- The IF NOT EXISTS clause makes the constraints idempotent. You can run the script multiple times without errors.
Write the entity creation code
A realistic enterprise has many types of identities and resources. Your generator creates 63 entities across 7 types: employees, contractors, security groups, workstations, servers, code repositories, storage buckets, SaaS tools, and service accounts.
- Below the create_constraints function, add the beginning of the populate_graph function that creates all entity nodes:
def populate_graph(driver):
# Create Users
users = [f"user_{i}" for i in range(1, 21)]
contractors = ["contractor_alice", "contractor_bob", "contractor_carol"]
all_users = users + contractors
for u in all_users:
role = "contractor" if "contractor" in u else "employee"
driver.execute_query(
"CREATE (:User {name: $name, role: $role, department: $dept})",
name=u,
role=role,
dept=random.choice(["engineering", "finance", "security", "ops"]),
database_=NEO4J_DATABASE,
)
# Create Groups with nesting
groups = [
"Domain Admins",
"Engineering",
"Finance",
"Security Team",
"DevOps",
"Contractors",
"Project Alpha",
"Temp Migration",
"Legacy Access",
]
for g in groups:
driver.execute_query(
"CREATE (:Group {name: $name})", name=g, database_=NEO4J_DATABASE
)
# Create Devices
devices = [f"workstation_{i}" for i in range(1, 11)] + [
"build_server_01",
"jump_host_01",
"dc_primary",
]
for d in devices:
driver.execute_query(
"CREATE (:Device {name: $name})", name=d, database_=NEO4J_DATABASE
)
# Create Repos
repos = ["api-gateway", "auth-service", "infra-terraform", "secrets-vault", "ml-pipeline"]
for r in repos:
driver.execute_query(
"CREATE (:Repo {name: $name})", name=r, database_=NEO4J_DATABASE
)
# Create Buckets
buckets = ["prod-backups", "customer-data", "logs-archive", "ml-models"]
for b in buckets:
driver.execute_query(
"CREATE (:Bucket {name: $name})", name=b, database_=NEO4J_DATABASE
)
# Create SaaS Apps
saas = ["GitHub Enterprise", "AWS Console", "Slack", "Jira", "Datadog"]
for s in saas:
driver.execute_query(
"CREATE (:SaaSApp {name: $name})", name=s, database_=NEO4J_DATABASE
)
# Create Service Accounts
service_accounts = ["svc_deploy", "svc_monitoring", "svc_backup", "svc_ci_runner"]
for sa in service_accounts:
driver.execute_query(
"CREATE (:ServiceAccount {name: $name})", name=sa, database_=NEO4J_DATABASE
)What does this code do?
- Creates 23 User nodes (20 employees and 3 contractors), each with a role and randomly assigned department property.
- Creates 9 Group nodes representing organizational security groups. Some of these (like Project Alpha and Temp Migration) exist specifically to create nested group chains.
- Creates 13 Devices (workstations plus servers), 5 Repos, 4 Buckets, 5 SaaS apps, and 4 Service Accounts.
- Each execute_query call uses parameterized Cypher (the $name syntax) to safely insert values without injection risk.
Now you need to wire these entities together with relationships that represent how access flows through the organization.
- Continue building the populate_graph function by adding the normal relationship creation code directly below the service accounts section:
# --- Normal Relationships ---
# Users -> Groups
for u in users[:5]:
driver.execute_query(
"MATCH (u:User {name: $user}), (g:Group {name: 'Engineering'}) "
"CREATE (u)-[:MEMBER_OF]->(g)",
user=u,
database_=NEO4J_DATABASE,
)
for u in users[5:10]:
driver.execute_query(
"MATCH (u:User {name: $user}), (g:Group {name: 'Finance'}) "
"CREATE (u)-[:MEMBER_OF]->(g)",
user=u,
database_=NEO4J_DATABASE,
)
for u in users[10:15]:
driver.execute_query(
"MATCH (u:User {name: $user}), (g:Group {name: 'DevOps'}) "
"CREATE (u)-[:MEMBER_OF]->(g)",
user=u,
database_=NEO4J_DATABASE,
)
for u in contractors:
driver.execute_query(
"MATCH (u:User {name: $user}), (g:Group {name: 'Contractors'}) "
"CREATE (u)-[:MEMBER_OF]->(g)",
user=u,
database_=NEO4J_DATABASE,
)
# Groups -> Resources (normal access)
driver.execute_query(
"MATCH (g:Group {name: 'Engineering'}), (r:Repo {name: 'api-gateway'}) "
"CREATE (g)-[:HAS_ACCESS {level: 'write'}]->(r)",
database_=NEO4J_DATABASE,
)
driver.execute_query(
"MATCH (g:Group {name: 'DevOps'}), (r:Repo {name: 'infra-terraform'}) "
"CREATE (g)-[:HAS_ACCESS {level: 'write'}]->(r)",
database_=NEO4J_DATABASE,
)
driver.execute_query(
"MATCH (g:Group {name: 'Domain Admins'}), (d:Device {name: 'dc_primary'}) "
"CREATE (g)-[:ADMIN_TO]->(d)",
database_=NEO4J_DATABASE,
)
# User sessions on devices
for i, u in enumerate(users[:10]):
device = f"workstation_{i + 1}"
driver.execute_query(
"MATCH (u:User {name: $user}), (d:Device {name: $device}) "
"CREATE (u)-[:HAS_SESSION]->(d)",
user=u,
device=device,
database_=NEO4J_DATABASE,
)
# Service account access
driver.execute_query(
"MATCH (sa:ServiceAccount {name: 'svc_deploy'}), (r:Repo {name: 'infra-terraform'}) "
"CREATE (sa)-[:HAS_ACCESS {level: 'admin'}]->(r)",
database_=NEO4J_DATABASE,
)
driver.execute_query(
"MATCH (sa:ServiceAccount {name: 'svc_monitoring'}), (s:SaaSApp {name: 'Datadog'}) "
"CREATE (sa)-[:HAS_ACCESS {level: 'read'}]->(s)",
database_=NEO4J_DATABASE,
)What does this code do?
- MEMBER_OF relationships assign users to their department groups. Users 1-5 go to Engineering, 5-10 to Finance, 10-15 to DevOps, and all contractors to the Contractors group.
- HAS_ACCESS relationships grant groups access to resources with a level property (read, write, or admin). This mirrors how real IAM systems work.
- HAS_SESSION relationships model which users are logged into which devices. This is critical for lateral movement detection.
- Service accounts get their own access relationships, separate from human users. These automated accounts often accumulate excess permissions over time.
- Save generate_graph.py.
Seeing an indentation error?
All the relationship code must be indented inside the populate_graph function body (4 spaces). If you see IndentationError, check that the relationship code aligns with the node creation loops above it.
Help me fix an indentation error in my populate_graph function.
Inject the attack scenarios and run the generator
The normal relationships above represent legitimate access. Now you will inject three deliberate attack scenarios that mimic real privilege escalation techniques attackers use in enterprise environments.
Why inject attack scenarios?
Security tools like BloodHound (the industry-standard attack path tool) work by discovering transitive access chains that nobody explicitly granted. By injecting known-bad patterns, you can verify that your detection queries actually find them.
The three patterns you will inject are: deeply nested group membership giving a contractor admin access, an over-privileged service account with admin access to everything, and a lateral movement chain through shared device sessions.
- Continue adding to the populate_graph function, directly below the svc_monitoring access line. Add the three attack scenarios:
# --- ATTACK SCENARIO 1: Deep nesting gives contractor admin ---
# contractor_alice -> Contractors -> Project Alpha -> Temp Migration -> Legacy Access -> Domain Admins
driver.execute_query(
"MATCH (g1:Group {name: 'Contractors'}), (g2:Group {name: 'Project Alpha'}) "
"CREATE (g1)-[:MEMBER_OF]->(g2)",
database_=NEO4J_DATABASE,
)
driver.execute_query(
"MATCH (g1:Group {name: 'Project Alpha'}), (g2:Group {name: 'Temp Migration'}) "
"CREATE (g1)-[:MEMBER_OF]->(g2)",
database_=NEO4J_DATABASE,
)
driver.execute_query(
"MATCH (g1:Group {name: 'Temp Migration'}), (g2:Group {name: 'Legacy Access'}) "
"CREATE (g1)-[:MEMBER_OF]->(g2)",
database_=NEO4J_DATABASE,
)
driver.execute_query(
"MATCH (g1:Group {name: 'Legacy Access'}), (g2:Group {name: 'Domain Admins'}) "
"CREATE (g1)-[:MEMBER_OF]->(g2)",
database_=NEO4J_DATABASE,
)
# --- ATTACK SCENARIO 2: Over-privileged service account ---
# svc_backup has access to everything
for r in repos:
driver.execute_query(
"MATCH (sa:ServiceAccount {name: 'svc_backup'}), (r:Repo {name: $repo}) "
"CREATE (sa)-[:HAS_ACCESS {level: 'admin'}]->(r)",
repo=r,
database_=NEO4J_DATABASE,
)
for b in buckets:
driver.execute_query(
"MATCH (sa:ServiceAccount {name: 'svc_backup'}), (b:Bucket {name: $bucket}) "
"CREATE (sa)-[:HAS_ACCESS {level: 'admin'}]->(b)",
bucket=b,
database_=NEO4J_DATABASE,
)
for s in saas:
driver.execute_query(
"MATCH (sa:ServiceAccount {name: 'svc_backup'}), (s:SaaSApp {name: $saas}) "
"CREATE (sa)-[:HAS_ACCESS {level: 'admin'}]->(s)",
saas=s,
database_=NEO4J_DATABASE,
)
# --- ATTACK SCENARIO 3: Lateral movement via shared sessions ---
# user_15 has session on build_server_01, which user_1 (engineering) also has session on
# user_1 has admin to jump_host_01, which has session from user with Domain Admin membership
driver.execute_query(
"MATCH (u:User {name: 'user_15'}), (d:Device {name: 'build_server_01'}) "
"CREATE (u)-[:HAS_SESSION]->(d)",
database_=NEO4J_DATABASE,
)
driver.execute_query(
"MATCH (u:User {name: 'user_1'}), (d:Device {name: 'build_server_01'}) "
"CREATE (u)-[:HAS_SESSION]->(d)",
database_=NEO4J_DATABASE,
)
driver.execute_query(
"MATCH (u:User {name: 'user_1'}), (d:Device {name: 'jump_host_01'}) "
"CREATE (u)-[:ADMIN_TO]->(d)",
database_=NEO4J_DATABASE,
)
driver.execute_query(
"MATCH (u:User {name: 'user_16'}), (g:Group {name: 'Domain Admins'}) "
"CREATE (u)-[:MEMBER_OF]->(g)",
database_=NEO4J_DATABASE,
)
driver.execute_query(
"MATCH (u:User {name: 'user_16'}), (d:Device {name: 'jump_host_01'}) "
"CREATE (u)-[:HAS_SESSION]->(d)",
database_=NEO4J_DATABASE,
)What do these attack scenarios model?
Scenario 1 creates a chain: contractor_alice is in Contractors, which nests through 4 intermediate groups to reach Domain Admins. Nobody explicitly gave a contractor admin access, but the transitive chain grants it. This is the most common real-world privilege escalation vector.
Scenario 2 gives svc_backup admin access to all 14 resources (5 repos, 4 buckets, 5 SaaS apps). Backup accounts often accumulate excessive permissions because "it needs to read everything."
Scenario 3 creates a lateral movement chain: user_15 shares build_server_01 with user_1, who has admin on jump_host_01, where user_16 (a Domain Admin) has an active session. An attacker compromising user_15 can hop through this chain to reach a Domain Admin session.
Finally, add the main() function that orchestrates the full pipeline and prints statistics about what was created.
- Below the populate_graph function (at the top-level indentation, not inside it), add the main function:
def main():
driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASSWORD))
try:
driver.verify_connectivity()
print("[+] Connected to Neo4j")
clear_database(driver)
print("[+] Database cleared")
create_constraints(driver)
print("[+] Constraints created")
populate_graph(driver)
print("[+] Enterprise access graph populated")
# Print stats
records, _, _ = driver.execute_query(
"MATCH (n) RETURN labels(n)[0] AS label, count(n) AS count "
"ORDER BY count DESC",
database_=NEO4J_DATABASE,
)
print("\n--- Graph Statistics ---")
for r in records:
print(f" {r['label']}: {r['count']}")
records, _, _ = driver.execute_query(
"MATCH ()-[r]->() RETURN type(r) AS type, count(r) AS count "
"ORDER BY count DESC",
database_=NEO4J_DATABASE,
)
print("\n--- Relationship Statistics ---")
for r in records:
print(f" {r['type']}: {r['count']}")
finally:
driver.close()
if __name__ == "__main__":
main()What does main() do?
- Connects to Neo4j, clears any existing data, creates the uniqueness constraints, then populates the full graph.
- After population, it runs two summary queries: one counting nodes by label, and one counting relationships by type. This gives you a quick sanity check that everything was created.
- The finally block ensures the driver connection closes even if an error occurs mid-population.
- Save generate_graph.py.
- Run the generator to populate your graph:
python generate_graph.pyYou should see output showing the connection succeeded, the database was cleared, constraints were created, and the graph was populated. The statistics section shows counts for each node label and relationship type.
- Open Neo4j Browser at http://localhost:7474 in your web browser.
- Run this Cypher query in the browser query bar to visualize the full graph:
MATCH (n)-[r]->(m) RETURN n, r, m LIMIT 100You should see a colorful visualization with nodes of all 7 types connected by multiple relationship types. The graph shows how users connect to groups, groups to resources, and the hidden nesting chains between groups.
Graph looks empty or script errors out?
- If you see a connection error, confirm Neo4j is still running with docker compose ps. The container may have stopped if your machine went to sleep.
- If you see ModuleNotFoundError: No module named 'config', make sure config.py is in the same folder as generate_graph.py.
- If the Neo4j Browser shows no results, make sure you are logged in with username neo4j and password securegraph123.
Help me debug why my generate_graph.py script is failing to connect or populate Neo4j.
✔️ Awesome, I've got everything!
Great. Make sure you have saved generate_graph.py and confirmed you can see the graph in Neo4j Browser.
ⓧ I'd like to double check the full code
Here is the complete generate_graph.py file:
import random
from neo4j import GraphDatabase
from config import NEO4J_URI, NEO4J_USER, NEO4J_PASSWORD, NEO4J_DATABASE
def clear_database(driver):
driver.execute_query("MATCH (n) DETACH DELETE n", database_=NEO4J_DATABASE)
def create_constraints(driver):
constraints = [
"CREATE CONSTRAINT IF NOT EXISTS FOR (u:User) REQUIRE u.name IS UNIQUE",
"CREATE CONSTRAINT IF NOT EXISTS FOR (g:Group) REQUIRE g.name IS UNIQUE",
"CREATE CONSTRAINT IF NOT EXISTS FOR (d:Device) REQUIRE d.name IS UNIQUE",
"CREATE CONSTRAINT IF NOT EXISTS FOR (r:Repo) REQUIRE r.name IS UNIQUE",
"CREATE CONSTRAINT IF NOT EXISTS FOR (b:Bucket) REQUIRE b.name IS UNIQUE",
"CREATE CONSTRAINT IF NOT EXISTS FOR (s:SaaSApp) REQUIRE s.name IS UNIQUE",
"CREATE CONSTRAINT IF NOT EXISTS FOR (sa:ServiceAccount) REQUIRE sa.name IS UNIQUE",
]
for c in constraints:
driver.execute_query(c, database_=NEO4J_DATABASE)
def populate_graph(driver):
# Create Users
users = [f"user_{i}" for i in range(1, 21)]
contractors = ["contractor_alice", "contractor_bob", "contractor_carol"]
all_users = users + contractors
for u in all_users:
role = "contractor" if "contractor" in u else "employee"
driver.execute_query(
"CREATE (:User {name: $name, role: $role, department: $dept})",
name=u,
role=role,
dept=random.choice(["engineering", "finance", "security", "ops"]),
database_=NEO4J_DATABASE,
)
# Create Groups with nesting
groups = [
"Domain Admins",
"Engineering",
"Finance",
"Security Team",
"DevOps",
"Contractors",
"Project Alpha",
"Temp Migration",
"Legacy Access",
]
for g in groups:
driver.execute_query(
"CREATE (:Group {name: $name})", name=g, database_=NEO4J_DATABASE
)
# Create Devices
devices = [f"workstation_{i}" for i in range(1, 11)] + [
"build_server_01",
"jump_host_01",
"dc_primary",
]
for d in devices:
driver.execute_query(
"CREATE (:Device {name: $name})", name=d, database_=NEO4J_DATABASE
)
# Create Repos
repos = ["api-gateway", "auth-service", "infra-terraform", "secrets-vault", "ml-pipeline"]
for r in repos:
driver.execute_query(
"CREATE (:Repo {name: $name})", name=r, database_=NEO4J_DATABASE
)
# Create Buckets
buckets = ["prod-backups", "customer-data", "logs-archive", "ml-models"]
for b in buckets:
driver.execute_query(
"CREATE (:Bucket {name: $name})", name=b, database_=NEO4J_DATABASE
)
# Create SaaS Apps
saas = ["GitHub Enterprise", "AWS Console", "Slack", "Jira", "Datadog"]
for s in saas:
driver.execute_query(
"CREATE (:SaaSApp {name: $name})", name=s, database_=NEO4J_DATABASE
)
# Create Service Accounts
service_accounts = ["svc_deploy", "svc_monitoring", "svc_backup", "svc_ci_runner"]
for sa in service_accounts:
driver.execute_query(
"CREATE (:ServiceAccount {name: $name})", name=sa, database_=NEO4J_DATABASE
)
# --- Normal Relationships ---
# Users -> Groups
for u in users[:5]:
driver.execute_query(
"MATCH (u:User {name: $user}), (g:Group {name: 'Engineering'}) "
"CREATE (u)-[:MEMBER_OF]->(g)",
user=u,
database_=NEO4J_DATABASE,
)
for u in users[5:10]:
driver.execute_query(
"MATCH (u:User {name: $user}), (g:Group {name: 'Finance'}) "
"CREATE (u)-[:MEMBER_OF]->(g)",
user=u,
database_=NEO4J_DATABASE,
)
for u in users[10:15]:
driver.execute_query(
"MATCH (u:User {name: $user}), (g:Group {name: 'DevOps'}) "
"CREATE (u)-[:MEMBER_OF]->(g)",
user=u,
database_=NEO4J_DATABASE,
)
for u in contractors:
driver.execute_query(
"MATCH (u:User {name: $user}), (g:Group {name: 'Contractors'}) "
"CREATE (u)-[:MEMBER_OF]->(g)",
user=u,
database_=NEO4J_DATABASE,
)
# Groups -> Resources (normal access)
driver.execute_query(
"MATCH (g:Group {name: 'Engineering'}), (r:Repo {name: 'api-gateway'}) "
"CREATE (g)-[:HAS_ACCESS {level: 'write'}]->(r)",
database_=NEO4J_DATABASE,
)
driver.execute_query(
"MATCH (g:Group {name: 'DevOps'}), (r:Repo {name: 'infra-terraform'}) "
"CREATE (g)-[:HAS_ACCESS {level: 'write'}]->(r)",
database_=NEO4J_DATABASE,
)
driver.execute_query(
"MATCH (g:Group {name: 'Domain Admins'}), (d:Device {name: 'dc_primary'}) "
"CREATE (g)-[:ADMIN_TO]->(d)",
database_=NEO4J_DATABASE,
)
# User sessions on devices
for i, u in enumerate(users[:10]):
device = f"workstation_{i + 1}"
driver.execute_query(
"MATCH (u:User {name: $user}), (d:Device {name: $device}) "
"CREATE (u)-[:HAS_SESSION]->(d)",
user=u,
device=device,
database_=NEO4J_DATABASE,
)
# Service account access
driver.execute_query(
"MATCH (sa:ServiceAccount {name: 'svc_deploy'}), (r:Repo {name: 'infra-terraform'}) "
"CREATE (sa)-[:HAS_ACCESS {level: 'admin'}]->(r)",
database_=NEO4J_DATABASE,
)
driver.execute_query(
"MATCH (sa:ServiceAccount {name: 'svc_monitoring'}), (s:SaaSApp {name: 'Datadog'}) "
"CREATE (sa)-[:HAS_ACCESS {level: 'read'}]->(s)",
database_=NEO4J_DATABASE,
)
# --- ATTACK SCENARIO 1: Deep nesting gives contractor admin ---
# contractor_alice -> Contractors -> Project Alpha -> Temp Migration -> Legacy Access -> Domain Admins
driver.execute_query(
"MATCH (g1:Group {name: 'Contractors'}), (g2:Group {name: 'Project Alpha'}) "
"CREATE (g1)-[:MEMBER_OF]->(g2)",
database_=NEO4J_DATABASE,
)
driver.execute_query(
"MATCH (g1:Group {name: 'Project Alpha'}), (g2:Group {name: 'Temp Migration'}) "
"CREATE (g1)-[:MEMBER_OF]->(g2)",
database_=NEO4J_DATABASE,
)
driver.execute_query(
"MATCH (g1:Group {name: 'Temp Migration'}), (g2:Group {name: 'Legacy Access'}) "
"CREATE (g1)-[:MEMBER_OF]->(g2)",
database_=NEO4J_DATABASE,
)
driver.execute_query(
"MATCH (g1:Group {name: 'Legacy Access'}), (g2:Group {name: 'Domain Admins'}) "
"CREATE (g1)-[:MEMBER_OF]->(g2)",
database_=NEO4J_DATABASE,
)
# --- ATTACK SCENARIO 2: Over-privileged service account ---
# svc_backup has access to everything
for r in repos:
driver.execute_query(
"MATCH (sa:ServiceAccount {name: 'svc_backup'}), (r:Repo {name: $repo}) "
"CREATE (sa)-[:HAS_ACCESS {level: 'admin'}]->(r)",
repo=r,
database_=NEO4J_DATABASE,
)
for b in buckets:
driver.execute_query(
"MATCH (sa:ServiceAccount {name: 'svc_backup'}), (b:Bucket {name: $bucket}) "
"CREATE (sa)-[:HAS_ACCESS {level: 'admin'}]->(b)",
bucket=b,
database_=NEO4J_DATABASE,
)
for s in saas:
driver.execute_query(
"MATCH (sa:ServiceAccount {name: 'svc_backup'}), (s:SaaSApp {name: $saas}) "
"CREATE (sa)-[:HAS_ACCESS {level: 'admin'}]->(s)",
saas=s,
database_=NEO4J_DATABASE,
)
# --- ATTACK SCENARIO 3: Lateral movement via shared sessions ---
# user_15 has session on build_server_01, which user_1 (engineering) also has session on
# user_1 has admin to jump_host_01, which has session from user with Domain Admin membership
driver.execute_query(
"MATCH (u:User {name: 'user_15'}), (d:Device {name: 'build_server_01'}) "
"CREATE (u)-[:HAS_SESSION]->(d)",
database_=NEO4J_DATABASE,
)
driver.execute_query(
"MATCH (u:User {name: 'user_1'}), (d:Device {name: 'build_server_01'}) "
"CREATE (u)-[:HAS_SESSION]->(d)",
database_=NEO4J_DATABASE,
)
driver.execute_query(
"MATCH (u:User {name: 'user_1'}), (d:Device {name: 'jump_host_01'}) "
"CREATE (u)-[:ADMIN_TO]->(d)",
database_=NEO4J_DATABASE,
)
driver.execute_query(
"MATCH (u:User {name: 'user_16'}), (g:Group {name: 'Domain Admins'}) "
"CREATE (u)-[:MEMBER_OF]->(g)",
database_=NEO4J_DATABASE,
)
driver.execute_query(
"MATCH (u:User {name: 'user_16'}), (d:Device {name: 'jump_host_01'}) "
"CREATE (u)-[:HAS_SESSION]->(d)",
database_=NEO4J_DATABASE,
)
def main():
driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASSWORD))
try:
driver.verify_connectivity()
print("[+] Connected to Neo4j")
clear_database(driver)
print("[+] Database cleared")
create_constraints(driver)
print("[+] Constraints created")
populate_graph(driver)
print("[+] Enterprise access graph populated")
# Print stats
records, _, _ = driver.execute_query(
"MATCH (n) RETURN labels(n)[0] AS label, count(n) AS count "
"ORDER BY count DESC",
database_=NEO4J_DATABASE,
)
print("\n--- Graph Statistics ---")
for r in records:
print(f" {r['label']}: {r['count']}")
records, _, _ = driver.execute_query(
"MATCH ()-[r]->() RETURN type(r) AS type, count(r) AS count "
"ORDER BY count DESC",
database_=NEO4J_DATABASE,
)
print("\n--- Relationship Statistics ---")
for r in records:
print(f" {r['type']}: {r['count']}")
finally:
driver.close()
if __name__ == "__main__":
main()Your enterprise access graph is alive. You have 63 entities connected by over 50 relationships, including 3 hidden attack paths waiting to be discovered. Next up, you will write Cypher queries that automatically find these attack paths and rank them by severity.
Discover Attack Paths with Cypher
Your enterprise identity graph is live with 63 nodes and 200+ relationships. Now comes the real power of a graph database: asking questions that flat permission tables physically cannot answer.
In this step, you will write Cypher queries that traverse the graph to discover hidden attack paths. These queries find the shortest route from any user to admin privileges, detect lateral movement chains through shared device sessions, and surface over-privileged accounts buried beneath layers of nested group memberships.
In this step, get ready to:
- Write shortest-path queries that find routes from any user to Domain Admins.
- Write lateral movement and over-privilege detection queries.
- Run all queries and analyze the discovered attack paths by hop count and risk.
Write shortest-path queries to Domain Admins
The most dangerous question in identity security is: how many hops separate a regular user from Domain Admin? Cypher's shortestPath function answers this by traversing every possible relationship chain between two nodes and returning the one with the fewest hops.
- Create a new file called discover_paths.py in your identity-attack-graph folder.
- Add the imports and the first query function by pasting the following code:
from neo4j import GraphDatabase
from config import NEO4J_URI, NEO4J_USER, NEO4J_PASSWORD, NEO4J_DATABASE
def find_shortest_paths_to_admin(driver):
print("\n" + "=" * 60)
print("ATTACK PATH DISCOVERY: Shortest Paths to Domain Admins")
print("=" * 60)
# Find the shortest path from every User to the Domain Admins group
records, _, _ = driver.execute_query(
"MATCH (u:User), (g:Group {name: 'Domain Admins'}), "
"p = shortestPath((u)-[*1..10]->(g)) "
"RETURN u.name AS user, length(p) AS hops, "
"[n IN nodes(p) | coalesce(n.name, 'unknown')] AS path "
"ORDER BY hops ASC",
database_=NEO4J_DATABASE,
)
if not records:
print(" No paths found to Domain Admins.")
return
# Print each user's path with hop count
for r in records:
path_str = " -> ".join(r["path"])
print(f"\n [{r['hops']} hops] {r['user']}")
print(f" Path: {path_str}")
print(f"\n Total users with path to Domain Admins: {len(records)}")
What does this query do?
The Cypher query uses shortestPath((u)-[*1..10]->(g)) to find the minimum-hop route from every User node to the Domain Admins group. The *1..10 means it will follow any relationship type up to 10 hops deep.
It returns the user's name, the hop count, and the full path as a list of node names. Results are sorted ascending by hops so the most dangerous (shortest) paths appear first.
- Save discover_paths.py.
Why shortestPath matters for security
An attacker only needs the shortest route to escalate privileges. A user who is 1 hop from Domain Admin (direct membership) is dangerous. A contractor who is 5 hops away through nested groups is still dangerous, because that path exists and can be exploited.
Traditional RBAC tools list direct memberships. They cannot answer "who can transitively reach admin?" because they do not traverse relationship chains.
Write lateral movement and over-privilege detection queries
Shortest-path finds direct privilege escalation. But attackers also move laterally: they compromise one device session, pivot to another user on the same device, and inherit that user's access.
- In discover_paths.py, add the lateral movement detection function below the find_shortest_paths_to_admin function:
def find_lateral_movement_chains(driver):
print("\n" + "=" * 60)
print("LATERAL MOVEMENT: Session-based Attack Chains")
print("=" * 60)
# Find users sharing devices, then trace what the co-session user can reach
records, _, _ = driver.execute_query(
"MATCH (attacker:User)-[:HAS_SESSION]->(d:Device)<-[:HAS_SESSION]-(victim:User) "
"WHERE attacker <> victim "
"MATCH (victim)-[:MEMBER_OF|ADMIN_TO*1..5]->(target) "
"WHERE target:Group OR target:Device "
"RETURN attacker.name AS attacker, d.name AS shared_device, "
"victim.name AS victim, collect(DISTINCT target.name) AS reachable_targets "
"ORDER BY size(collect(DISTINCT target.name)) DESC",
database_=NEO4J_DATABASE,
)
for r in records:
print(f"\n Attacker: {r['attacker']}")
print(f" Shared Device: {r['shared_device']}")
print(f" Pivot Via: {r['victim']}")
print(f" Reachable Targets: {r['reachable_targets'][:5]}")
What does this query do?
This query chains two patterns. First, it finds pairs of users who share a device session (HAS_SESSION to the same Device). Then it traces what the "victim" user can reach through MEMBER_OF or ADMIN_TO relationships up to 5 hops deep.
If an attacker compromises the shared device, they inherit the victim's session and can reach everything the victim can reach. This is the lateral movement pattern used by tools like BloodHound.
- Save discover_paths.py.
- Next, add the over-privilege detection function below find_lateral_movement_chains:
def find_over_privileged_accounts(driver):
print("\n" + "=" * 60)
print("OVER-PRIVILEGE: Accounts with Transitive Admin Access")
print("=" * 60)
# Find all entities that can reach Domain Admins through nested MEMBER_OF chains
records, _, _ = driver.execute_query(
"MATCH (u)-[:MEMBER_OF*1..10]->(g:Group {name: 'Domain Admins'}) "
"WHERE u:User OR u:Group "
"RETURN u.name AS entity, labels(u)[0] AS type, "
"length(shortestPath((u)-[:MEMBER_OF*1..10]->(g))) AS depth "
"ORDER BY depth DESC",
database_=NEO4J_DATABASE,
g_name="Domain Admins",
)
print(f"\n Entities with transitive Domain Admin access: {len(records)}")
for r in records:
# Flag entities buried 3+ hops deep as critical (hard to audit manually)
marker = " [CRITICAL]" if r["depth"] >= 3 else ""
print(f" {r['entity']} ({r['type']}) - {r['depth']} hops away{marker}")
What does this query do?
This query traverses only MEMBER_OF relationships up to 10 hops, surfacing every User or Group that has a transitive path to Domain Admins through nested group memberships.
Entities 3 or more hops away get a [CRITICAL] marker because they represent hidden admin access that a manual permissions audit would likely miss. This is exactly what happened with the contractor attack scenario you injected in Step 2.
- Save discover_paths.py.
- Finally, add the service account risk detection function below find_over_privileged_accounts:
def find_service_account_risks(driver):
print("\n" + "=" * 60)
print("SERVICE ACCOUNT RISK: Over-Permissioned Accounts")
print("=" * 60)
# Find service accounts with access to more than 3 resources
records, _, _ = driver.execute_query(
"MATCH (sa:ServiceAccount)-[r:HAS_ACCESS]->(target) "
"WITH sa, count(target) AS access_count, "
"collect(target.name) AS targets "
"WHERE access_count > 3 "
"RETURN sa.name AS account, access_count, targets "
"ORDER BY access_count DESC",
database_=NEO4J_DATABASE,
)
for r in records:
print(f"\n {r['account']}: {r['access_count']} resources")
print(f" Targets: {r['targets']}")
What does this query do?
This query counts direct HAS_ACCESS relationships from each ServiceAccount and flags any account touching more than 3 resources. It targets your second injected attack scenario: the svc_backup service account with admin-level access to every repo, bucket, and SaaS app.
Over-permissioned service accounts are one of the most common real-world attack vectors. They often accumulate access over time and rarely get audited because no human logs in with them directly.
- Save discover_paths.py.
Seeing a syntax error when saving?
Make sure each function definition has a blank line above it separating it from the previous function. Python requires consistent indentation within each function body. Check that no lines are mixing tabs and spaces.
help me fix a syntax error in my discover_paths.py file
Run the attack path discovery script
All four query functions are ready. Now you need a main function to orchestrate them and a script entry point to execute everything.
- Add the main function at the bottom of discover_paths.py, below find_service_account_risks:
def main():
driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASSWORD))
try:
find_shortest_paths_to_admin(driver)
find_lateral_movement_chains(driver)
find_over_privileged_accounts(driver)
find_service_account_risks(driver)
print("\n" + "=" * 60)
print("SUMMARY")
print("=" * 60)
print(" Attack paths discovered. Review results above.")
print(" Next step: Run anomaly detection (detect_anomalies.py)")
finally:
driver.close()
if __name__ == "__main__":
main()
What does this code do?
The main function creates a Neo4j driver connection and calls each discovery function in sequence. The try/finally block ensures the driver connection closes cleanly even if a query fails partway through.
- Save discover_paths.py.
- Run the attack path discovery script by executing this command in your terminal:
python discover_paths.py- You should see four report sections in your terminal output: shortest paths to Domain Admins with hop counts, lateral movement chains showing shared device pivots, over-privileged accounts with transitive admin access, and service account risks listing svc_backup with 14 resource accesses.
What should I see in the results?
The output reveals all three injected attack scenarios. The contractors (contractor_alice, contractor_bob, contractor_carol) reach Domain Admins through a 5-hop group nesting chain. The svc_backup service account has admin access to 14 resources. And user_15 can pivot through build_server_01 to reach user_1's admin access on jump_host_01.
The over-privilege section also shows user_16 as a direct Domain Admin member (1 hop) and the nested group chain marked [CRITICAL] for entities 3+ hops deep.
Seeing a connection error?
Make sure your Neo4j container is still running. Run docker ps to check. If it is not listed, restart it with docker-compose up -d from your project folder.
If you see "No paths found to Domain Admins", your graph may be empty. Re-run python generate_graph.py to repopulate it.
help me debug a connection error when running discover_paths.py against my Neo4j instance
✔️ Awesome, I've got everything!
Great. Double check you have saved discover_paths.py and that all four report sections printed in your terminal.
ⓧ I'd like to double check the full code
Here is the complete discover_paths.py file:
from neo4j import GraphDatabase
from config import NEO4J_URI, NEO4J_USER, NEO4J_PASSWORD, NEO4J_DATABASE
def find_shortest_paths_to_admin(driver):
print("\n" + "=" * 60)
print("ATTACK PATH DISCOVERY: Shortest Paths to Domain Admins")
print("=" * 60)
records, _, _ = driver.execute_query(
"MATCH (u:User), (g:Group {name: 'Domain Admins'}), "
"p = shortestPath((u)-[*1..10]->(g)) "
"RETURN u.name AS user, length(p) AS hops, "
"[n IN nodes(p) | coalesce(n.name, 'unknown')] AS path "
"ORDER BY hops ASC",
database_=NEO4J_DATABASE,
)
if not records:
print(" No paths found to Domain Admins.")
return
for r in records:
path_str = " -> ".join(r["path"])
print(f"\n [{r['hops']} hops] {r['user']}")
print(f" Path: {path_str}")
print(f"\n Total users with path to Domain Admins: {len(records)}")
def find_lateral_movement_chains(driver):
print("\n" + "=" * 60)
print("LATERAL MOVEMENT: Session-based Attack Chains")
print("=" * 60)
records, _, _ = driver.execute_query(
"MATCH (attacker:User)-[:HAS_SESSION]->(d:Device)<-[:HAS_SESSION]-(victim:User) "
"WHERE attacker <> victim "
"MATCH (victim)-[:MEMBER_OF|ADMIN_TO*1..5]->(target) "
"WHERE target:Group OR target:Device "
"RETURN attacker.name AS attacker, d.name AS shared_device, "
"victim.name AS victim, collect(DISTINCT target.name) AS reachable_targets "
"ORDER BY size(collect(DISTINCT target.name)) DESC",
database_=NEO4J_DATABASE,
)
for r in records:
print(f"\n Attacker: {r['attacker']}")
print(f" Shared Device: {r['shared_device']}")
print(f" Pivot Via: {r['victim']}")
print(f" Reachable Targets: {r['reachable_targets'][:5]}")
def find_over_privileged_accounts(driver):
print("\n" + "=" * 60)
print("OVER-PRIVILEGE: Accounts with Transitive Admin Access")
print("=" * 60)
records, _, _ = driver.execute_query(
"MATCH (u)-[:MEMBER_OF*1..10]->(g:Group {name: 'Domain Admins'}) "
"WHERE u:User OR u:Group "
"RETURN u.name AS entity, labels(u)[0] AS type, "
"length(shortestPath((u)-[:MEMBER_OF*1..10]->(g))) AS depth "
"ORDER BY depth DESC",
database_=NEO4J_DATABASE,
g_name="Domain Admins",
)
print(f"\n Entities with transitive Domain Admin access: {len(records)}")
for r in records:
marker = " [CRITICAL]" if r["depth"] >= 3 else ""
print(f" {r['entity']} ({r['type']}) - {r['depth']} hops away{marker}")
def find_service_account_risks(driver):
print("\n" + "=" * 60)
print("SERVICE ACCOUNT RISK: Over-Permissioned Accounts")
print("=" * 60)
records, _, _ = driver.execute_query(
"MATCH (sa:ServiceAccount)-[r:HAS_ACCESS]->(target) "
"WITH sa, count(target) AS access_count, "
"collect(target.name) AS targets "
"WHERE access_count > 3 "
"RETURN sa.name AS account, access_count, targets "
"ORDER BY access_count DESC",
database_=NEO4J_DATABASE,
)
for r in records:
print(f"\n {r['account']}: {r['access_count']} resources")
print(f" Targets: {r['targets']}")
def main():
driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASSWORD))
try:
find_shortest_paths_to_admin(driver)
find_lateral_movement_chains(driver)
find_over_privileged_accounts(driver)
find_service_account_risks(driver)
print("\n" + "=" * 60)
print("SUMMARY")
print("=" * 60)
print(" Attack paths discovered. Review results above.")
print(" Next step: Run anomaly detection (detect_anomalies.py)")
finally:
driver.close()
if __name__ == "__main__":
main()
Your Cypher queries just revealed attack paths that a flat permission table would never show. Next, you will build a Python anomaly detection pipeline that uses graph centrality features to automatically flag risky entities without writing manual queries for every possible attack pattern.
Build the Graph Anomaly Detection Pipeline
Your Cypher queries discovered specific attack paths you knew to look for. In a real enterprise with thousands of identities, you cannot manually write a query for every possible risk pattern.
In this step, you will build an automated anomaly detection pipeline that extracts numerical features from the graph and uses machine learning to flag entities that are statistically unusual. The model does not need to know what an "attack" looks like. It finds nodes that behave differently from their peers.
In this step, get ready to:
- Extract graph centrality features for every User, ServiceAccount, and Group node.
- Train an Isolation Forest model to identify statistical outliers.
- Generate an anomaly report that ranks risky entities with explainable reasons.
Extract graph features from the access graph
To feed a machine learning model, you need to convert the graph's structure into numbers. Each node gets a feature vector describing how it connects to the rest of the graph.
- Create a new file called detect_anomalies.py in your project folder.
- Add the imports and the feature extraction function by pasting this code:
import numpy as np
from sklearn.ensemble import IsolationForest
from neo4j import GraphDatabase
from config import NEO4J_URI, NEO4J_USER, NEO4J_PASSWORD, NEO4J_DATABASE
def extract_graph_features(driver):
print("[+] Extracting graph features for anomaly detection...")
# Get degree centrality (in + out degree for each node)
records, _, _ = driver.execute_query(
"MATCH (n) "
"WHERE n:User OR n:ServiceAccount OR n:Group "
"OPTIONAL MATCH (n)-[out]->() "
"OPTIONAL MATCH (n)<-[inc]-() "
"WITH n, count(DISTINCT out) AS out_degree, count(DISTINCT inc) AS in_degree "
"RETURN n.name AS name, labels(n)[0] AS label, "
"out_degree, in_degree, out_degree + in_degree AS total_degree",
database_=NEO4J_DATABASE,
)
entities = []
for r in records:
entities.append({
"name": r["name"],
"label": r["label"],
"out_degree": r["out_degree"],
"in_degree": r["in_degree"],
"total_degree": r["total_degree"],
})
# Get reachability (resources accessible within 3 hops)
for entity in entities:
records, _, _ = driver.execute_query(
"MATCH (n {name: $name})-[*1..3]->(target) "
"WHERE target:Repo OR target:Bucket OR target:SaaSApp OR target:Device "
"RETURN count(DISTINCT target) AS reachable",
name=entity["name"],
database_=NEO4J_DATABASE,
)
entity["reachable_resources"] = records[0]["reachable"] if records else 0
# Get path-to-admin score (0 if no path, inverse of hop count if path exists)
for entity in entities:
records, _, _ = driver.execute_query(
"MATCH (n {name: $name}), (g:Group {name: 'Domain Admins'}) "
"MATCH p = shortestPath((n)-[*1..10]->(g)) "
"RETURN length(p) AS hops",
name=entity["name"],
database_=NEO4J_DATABASE,
)
if records:
entity["admin_proximity"] = 1.0 / records[0]["hops"]
else:
entity["admin_proximity"] = 0.0
return entitiesWhat does this code do?
This function builds a feature vector for every User, ServiceAccount, and Group node. Each entity gets five numerical features:
- out_degree and in_degree measure how many relationships flow out of and into the node.
- reachable_resources counts how many sensitive assets (Repos, Buckets, SaaS apps, Devices) the node can reach within 3 hops.
- admin_proximity is the inverse of the shortest path to Domain Admins. A node 2 hops away scores 0.5. A node with no path scores 0.0.
Why Isolation Forest instead of a graph neural network?
Most identity graph anomalies are rooted in node and edge frequency and correlation. These are the same kinds of patterns that standard statistical models handle well.
Isolation Forest is fast, unsupervised (no labelled attack data needed), and interpretable. Graph neural networks add complexity without proportional benefit at this scale.
- Save detect_anomalies.py.
- You won't run this file yet. The next chunk adds the anomaly detection function that uses these features.
Detect anomalies with Isolation Forest
With the feature vectors computed, the next function feeds them into an Isolation Forest model. This algorithm isolates outliers by randomly partitioning the data. Nodes that are easy to isolate (they differ from the crowd) get high anomaly scores.
- In detect_anomalies.py, add the following function below extract_graph_features:
def run_anomaly_detection(entities):
print("[+] Running Isolation Forest anomaly detection...")
feature_names = [
"out_degree",
"in_degree",
"total_degree",
"reachable_resources",
"admin_proximity",
]
# Build a feature matrix from entity dictionaries
X = np.array([[e[f] for f in feature_names] for e in entities])
# Train Isolation Forest (unsupervised - no labels needed)
clf = IsolationForest(
n_estimators=100,
contamination=0.15,
random_state=42,
)
clf.fit(X)
# Score each entity (higher = more anomalous)
scores = clf.decision_function(X)
predictions = clf.predict(X)
for i, entity in enumerate(entities):
entity["anomaly_score"] = -scores[i] # Higher = more anomalous
entity["is_anomaly"] = predictions[i] == -1
return entitiesWhat does this code do?
- The feature matrix X stacks all entity feature vectors into a 2D array that scikit-learn can process.
- contamination=0.15 tells the model to expect about 15% of entities to be anomalous. This matches the proportion of injected attack scenarios in our graph.
- decision_function returns a continuous score per entity. We negate it so that higher values mean more anomalous.
- predict returns -1 for outliers and 1 for inliers, giving a binary classification.
- Save detect_anomalies.py.
- You won't see a change yet. The next chunk adds the report output and the main entry point.
Generate the anomaly report and run the pipeline
The final piece prints a human-readable report that ranks flagged entities by anomaly score and explains why each was flagged.
- In detect_anomalies.py, add the report function below run_anomaly_detection:
def print_anomaly_report(entities):
print("\n" + "=" * 60)
print("ANOMALY DETECTION REPORT")
print("=" * 60)
anomalies = [e for e in entities if e["is_anomaly"]]
anomalies.sort(key=lambda x: x["anomaly_score"], reverse=True)
print(f"\n Total entities analyzed: {len(entities)}")
print(f" Anomalies detected: {len(anomalies)}")
print("\n --- Flagged Entities (ranked by anomaly score) ---")
for e in anomalies:
print(f"\n [{e['anomaly_score']:.3f}] {e['name']} ({e['label']})")
print(f" Degree: {e['total_degree']} (in:{e['in_degree']}, out:{e['out_degree']})")
print(f" Reachable Resources (3 hops): {e['reachable_resources']}")
print(f" Admin Proximity: {e['admin_proximity']:.2f}")
# Explain why flagged
reasons = []
if e["reachable_resources"] > 5:
reasons.append("excessive resource access breadth")
if e["admin_proximity"] > 0.2:
reasons.append("close path to admin")
if e["total_degree"] > 10:
reasons.append("unusually high connectivity")
if not reasons:
reasons.append("statistical outlier across combined features")
print(f" Reason: {', '.join(reasons)}")
print("\n --- Normal Entities (sample) ---")
normal = [e for e in entities if not e["is_anomaly"]][:5]
for e in normal:
print(f" {e['name']} ({e['label']}) - score: {e['anomaly_score']:.3f}")What does this code do?
- Filters the entity list to only anomalies and ranks them by score (highest risk first).
- For each flagged entity, it prints the raw feature values so you can see exactly what made it stand out.
- The reason logic translates raw feature thresholds into human-readable explanations. If no single feature exceeds its threshold, the entity is flagged as a combined statistical outlier.
- A sample of normal entities is printed at the end for comparison, showing the contrast in scores.
- Now add the main() function and entry point at the bottom of detect_anomalies.py:
def main():
driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASSWORD))
try:
entities = extract_graph_features(driver)
entities = run_anomaly_detection(entities)
print_anomaly_report(entities)
print("\n" + "=" * 60)
print("NEXT STEP: Run policy engine (python policy_engine.py)")
print("=" * 60)
finally:
driver.close()
if __name__ == "__main__":
main()What does this code do?
The main() function orchestrates the full pipeline: connect to Neo4j, extract features, run anomaly detection, and print the report. The try/finally block ensures the driver connection closes even if an error occurs.
- Save detect_anomalies.py.
- Run the anomaly detection pipeline by running this command:
python detect_anomalies.py- You should see output showing feature extraction progress, followed by the full anomaly detection report. The flagged entities section should include svc_backup (the over-privileged service account) and several of the contractor or lateral-movement users near the top.
Seeing a ModuleNotFoundError?
- If you see an error about sklearn or numpy, make sure your virtual environment is active and you installed all packages from requirements.txt in Step 1.
- If you see a Neo4j connection error, confirm that your Docker container is still running with docker-compose ps.
- If the output shows zero anomalies, make sure you ran generate_graph.py first to populate the graph with the attack scenarios.
Help me debug my anomaly detection pipeline
✔️ Awesome, I've got everything!
Great. Double check you have saved detect_anomalies.py and your anomaly report printed successfully.
ⓧ I'd like to double check the full code
import numpy as np
from sklearn.ensemble import IsolationForest
from neo4j import GraphDatabase
from config import NEO4J_URI, NEO4J_USER, NEO4J_PASSWORD, NEO4J_DATABASE
def extract_graph_features(driver):
print("[+] Extracting graph features for anomaly detection...")
# Get degree centrality (in + out degree for each node)
records, _, _ = driver.execute_query(
"MATCH (n) "
"WHERE n:User OR n:ServiceAccount OR n:Group "
"OPTIONAL MATCH (n)-[out]->() "
"OPTIONAL MATCH (n)<-[inc]-() "
"WITH n, count(DISTINCT out) AS out_degree, count(DISTINCT inc) AS in_degree "
"RETURN n.name AS name, labels(n)[0] AS label, "
"out_degree, in_degree, out_degree + in_degree AS total_degree",
database_=NEO4J_DATABASE,
)
entities = []
for r in records:
entities.append({
"name": r["name"],
"label": r["label"],
"out_degree": r["out_degree"],
"in_degree": r["in_degree"],
"total_degree": r["total_degree"],
})
# Get reachability (resources accessible within 3 hops)
for entity in entities:
records, _, _ = driver.execute_query(
"MATCH (n {name: $name})-[*1..3]->(target) "
"WHERE target:Repo OR target:Bucket OR target:SaaSApp OR target:Device "
"RETURN count(DISTINCT target) AS reachable",
name=entity["name"],
database_=NEO4J_DATABASE,
)
entity["reachable_resources"] = records[0]["reachable"] if records else 0
# Get path-to-admin score (0 if no path, inverse of hop count if path exists)
for entity in entities:
records, _, _ = driver.execute_query(
"MATCH (n {name: $name}), (g:Group {name: 'Domain Admins'}) "
"MATCH p = shortestPath((n)-[*1..10]->(g)) "
"RETURN length(p) AS hops",
name=entity["name"],
database_=NEO4J_DATABASE,
)
if records:
entity["admin_proximity"] = 1.0 / records[0]["hops"]
else:
entity["admin_proximity"] = 0.0
return entities
def run_anomaly_detection(entities):
print("[+] Running Isolation Forest anomaly detection...")
feature_names = [
"out_degree",
"in_degree",
"total_degree",
"reachable_resources",
"admin_proximity",
]
X = np.array([[e[f] for f in feature_names] for e in entities])
clf = IsolationForest(
n_estimators=100,
contamination=0.15,
random_state=42,
)
clf.fit(X)
scores = clf.decision_function(X)
predictions = clf.predict(X)
for i, entity in enumerate(entities):
entity["anomaly_score"] = -scores[i] # Higher = more anomalous
entity["is_anomaly"] = predictions[i] == -1
return entities
def print_anomaly_report(entities):
print("\n" + "=" * 60)
print("ANOMALY DETECTION REPORT")
print("=" * 60)
anomalies = [e for e in entities if e["is_anomaly"]]
anomalies.sort(key=lambda x: x["anomaly_score"], reverse=True)
print(f"\n Total entities analyzed: {len(entities)}")
print(f" Anomalies detected: {len(anomalies)}")
print("\n --- Flagged Entities (ranked by anomaly score) ---")
for e in anomalies:
print(f"\n [{e['anomaly_score']:.3f}] {e['name']} ({e['label']})")
print(f" Degree: {e['total_degree']} (in:{e['in_degree']}, out:{e['out_degree']})")
print(f" Reachable Resources (3 hops): {e['reachable_resources']}")
print(f" Admin Proximity: {e['admin_proximity']:.2f}")
# Explain why flagged
reasons = []
if e["reachable_resources"] > 5:
reasons.append("excessive resource access breadth")
if e["admin_proximity"] > 0.2:
reasons.append("close path to admin")
if e["total_degree"] > 10:
reasons.append("unusually high connectivity")
if not reasons:
reasons.append("statistical outlier across combined features")
print(f" Reason: {', '.join(reasons)}")
print("\n --- Normal Entities (sample) ---")
normal = [e for e in entities if not e["is_anomaly"]][:5]
for e in normal:
print(f" {e['name']} ({e['label']}) - score: {e['anomaly_score']:.3f}")
def main():
driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASSWORD))
try:
entities = extract_graph_features(driver)
entities = run_anomaly_detection(entities)
print_anomaly_report(entities)
print("\n" + "=" * 60)
print("NEXT STEP: Run policy engine (python policy_engine.py)")
print("=" * 60)
finally:
driver.close()
if __name__ == "__main__":
main()Your pipeline now automatically discovers risky identities without any manual query writing. In the Secret Mission, you will build a self-healing policy engine that responds to these anomalies by mutating the graph itself.
Secret mission
Build the Self-Healing Policy Engine
Your engine detects threats, but it just prints a report and walks away. What if the access graph could heal itself? Build a policy engine that automatically responds to detected anomalies with simulated self-healing actions: step-up authentication, access revocation, and quarantine.
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally using Docker and Python. There are no ongoing cloud costs or paid API charges. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- Neo4j Docker container and neo4j_data volume.
- Python virtual environment with neo4j, scikit-learn, and numpy packages.
- Project files: docker-compose.yml, config.py, generate_graph.py, discover_paths.py, detect_anomalies.py, policy_engine.py, policy_rules.json, requirements.txt.
Keep everything running
No action needed. Choose this if you are still actively building or plan to extend your attack path engine with additional features.
- Your Neo4j container continues running in the background and uses a small amount of memory.
- Your project files and virtual environment remain on your Desktop.
- You can re-run any of the scripts at any time to regenerate data or re-detect anomalies.
Pause - I'll come back to this later
Shut down running processes to free up memory while keeping your files intact for later.
- Stop the Neo4j container by navigating to your project folder and running:
- Your graph data persists in the neo4j_data Docker volume. When you return, run docker-compose up -d to start Neo4j again with your data intact.
- Deactivate your Python virtual environment by running:
docker-compose downDeactivate the virtual environment:
deactivateYour project files and virtual environment folder remain on disk. Re-activate the environment when you return with source venv/bin/activate (macOS/Linux) or venv\Scripts\activate (Windows).
Delete - I don't want to use this again
Remove all project resources and start fresh. This permanently deletes your Neo4j graph data and all project files.
- Stop the Neo4j container and remove the Docker volume by navigating to your project folder and running:
docker-compose down -v- Deactivate your virtual environment if it is still active:
deactivate- Delete your project folder from the Desktop. On macOS/Linux:
cd ~/Desktop
rm -rf [[PROJECT_FOLDER="your project folder name"]]On Windows:
cd ~\Desktop
rmdir /s /q [[PROJECT_FOLDER="your project folder name"]]- Optionally remove the downloaded Neo4j Docker image to free disk space:
docker rmi neo4j:2026.05.0Nice Work!
Nice Work!
You just built a fully functional identity attack path engine that models enterprise access as a graph and algorithmically discovers security risks. This is the same approach used by professional red teams and identity security platforms to find privilege escalation routes that flat permission tables completely miss.
Here's what you learned:
- Built a live Neo4j identity graph modeling users, devices, groups, repos, buckets, and SaaS tools with realistic access relationships and injected attack scenarios.
- Wrote Cypher queries that discover hidden attack paths, shortest routes to admin privileges, and lateral movement chains through shared device sessions.
- Built a Python anomaly detection pipeline using graph centrality features and Isolation Forest to flag risky entities with explainable reasoning.
- Secret Mission: Built a self-healing policy engine that automatically responds to detected threats by mutating the access graph and verifying reduced anomaly scores.
Ready to quiz yourself?