Shadow API Detector

Build a CLI tool that detects undocumented shadow APIs from traffic logs.

Introduction

30 Second Summary

Every company has APIs that nobody officially documented. These "shadow" APIs slip through security controls because defenders don't even know they exist.

In this project, you will build a Python command-line tool that discovers undocumented API endpoints by analyzing gateway traffic logs and comparing them against an official API specification.

What You'll Build

A Shadow API Detector that parses NGINX access logs, normalizes dynamic path segments into route patterns, diffs discovered endpoints against an OpenAPI inventory, and flags undocumented APIs with risk scores.

By the end of this project, you'll have:

  • A traffic log generator and parser that produces realistic NGINX combined-format logs and extracts normalized endpoint patterns from them.
  • A shadow detection engine that diffs discovered endpoints against an OpenAPI spec and classifies findings as shadow, zombie, or method-mismatch.
  • A risk-scored SQLite database of findings with a prioritized terminal report showing authentication gaps, PII exposure, and deprecated version risks.
  • Secret Mission: Add temporal anomaly detection that flags endpoints active only during off-hours as potential backdoors.

Are there any prerequisites?

You should be comfortable writing Python scripts, including functions, regex, file I/O, and dictionaries. Familiarity with HTTP methods and REST APIs is also helpful.

Before We Start

Shadow APIs are undocumented endpoints hiding in your network traffic. They bypass security controls like rate limiting and authentication, creating attack surface that security teams cannot monitor.

In this step, you'll set up the tools and project structure for your Shadow API Detector. By the end, you'll have a Python project directory ready to start parsing traffic logs.

In this step, get ready to:
  • Confirm Python 3.13+ is installed on your machine.
  • Create the shadow_detector/ project directory with all placeholder files.
  • Set up a virtual environment and install PyYAML.
Confirm Python 3.13+ is installed

This project uses Python 3.13+ features and libraries. Let's make sure the right version is available on your machine.

  • Press Cmd+Space to open Spotlight Search.
  • Type Terminal and press Enter to open it.
  • Check your Python version by running this command:
python3 --version

✔️ I see version 3.13 or higher

You're all set. Your output looks something like Python 3.13.14 or higher. Move on to the next substep.

ⓧ I see an older version

If you see a version lower than 3.13 (for example Python 3.11.5), you need to install a newer version. macOS sometimes ships with or retains an older Python.

  • Visit python.org/downloads/macos and download the latest Python 3.13 macOS installer.
  • Run the downloaded .pkg file and follow the installer prompts.
  • Once installed, close and reopen Terminal.
  • Run python3 --version again to confirm it now shows 3.13 or higher.

Still seeing the old version?

The python.org installer places the new version at /Library/Frameworks/Python.framework/Versions/3.13/bin/python3. If python3 --version still shows the old version after reopening Terminal, try running python3.13 --version instead. Use python3.13 in place of python3 for the rest of this project.

ⓧ Command not found

Python is not installed on your machine yet. Let's install it.

  • Visit python.org/downloads/macos and download the latest Python 3.13 macOS installer.
  • Run the downloaded .pkg file and follow the installer prompts.
  • Once installed, close and reopen Terminal.
  • Run python3 --version to confirm you see 3.13 or higher.
Create the project directory and files

Your Shadow API Detector lives in a single folder with dedicated Python files for each component of the pipeline.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder, move into it, and create the placeholder files by running these commands:
mkdir shadow_detector
cd shadow_detector
touch generate_traffic.py parse_logs.py detect_shadows.py openapi_spec.yaml main.py
  • Confirm the files exist by running:
ls

You should see detect_shadows.py, generate_traffic.py, main.py, openapi_spec.yaml, and parse_logs.py in the output.

Don't see all five files?

Make sure you ran the touch command with all five filenames on the same line, separated by spaces. If a file is missing, run touch again with just the missing filename.

help me create my project files for shadow_detector

Set up a virtual environment and install PyYAML

A virtual environment keeps this project's dependencies isolated from your system Python. This prevents version conflicts with other projects on your machine.

  • Create a virtual environment inside your shadow_detector/ folder by running:
python3 -m venv venv
  • Activate the virtual environment by running:
source venv/bin/activate

You should see (venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.

No (venv) in your prompt?

Double-check you're inside the shadow_detector/ directory. Run pwd to confirm you see Desktop/shadow_detector at the end of the path. If the venv folder does not exist, re-run the python3 -m venv venv command.

help me activate my Python virtual environment

  • Install the project's only external dependency by running:
pip install PyYAML==6.0.3

You should see output ending with Successfully installed PyYAML-6.0.3.

Why PyYAML?

Your detector compares discovered traffic against an OpenAPI spec file written in YAML. PyYAML lets Python read and parse that spec into a dictionary you can search through.

Everything else in this project (regex, SQLite, file I/O, datetime) comes from Python's standard library. No other installs needed.

Install failed?

Make sure (venv) is visible in your prompt before running pip install. If you see a permissions error, you may have forgotten to activate the virtual environment. Run source venv/bin/activate first, then try the install again.

help me install PyYAML in my virtual environment

Your project skeleton is ready. Next up, you'll fill in the OpenAPI spec that defines your "official" API inventory, then build a traffic generator that simulates real NGINX access logs.

Scaffold the Project and Generate Traffic

Your environment is set up and ready to go. Now it is time to define what your detector considers "official" and generate the traffic data it will analyze.

Every shadow API detector needs two inputs: a source of truth listing known endpoints, and a stream of real traffic to compare against it. You will create both from scratch so the project is fully self-contained and reproducible.

In this step, get ready to:
  • Define the official API inventory in an OpenAPI spec.
  • Build a traffic generator that outputs realistic NGINX access logs with planted shadow APIs.
Create the OpenAPI spec

An OpenAPI spec is a machine-readable document that lists every endpoint an API officially supports. Your detector will treat this file as the single source of truth for "documented" endpoints.

  • Open openapi_spec.yaml in your text editor and paste the following YAML:
openapi: "3.0.3"
info:
  title: "Acme E-Commerce API"
  version: "2.0.0"
paths:
  /v2/users:
    get:
      summary: "List users"
    post:
      summary: "Create user"
  /v2/users/{id}:
    get:
      summary: "Get user by ID"
    put:
      summary: "Update user"
    delete:
      summary: "Delete user"
  /v2/products:
    get:
      summary: "List products"
    post:
      summary: "Create product"
  /v2/products/{id}:
    get:
      summary: "Get product by ID"
    put:
      summary: "Update product"
  /v2/orders:
    get:
      summary: "List orders"
    post:
      summary: "Create order"
  /v2/orders/{id}:
    get:
      summary: "Get order by ID"
  /v2/auth/login:
    post:
      summary: "Login"
  /v2/auth/logout:
    post:
      summary: "Logout"
  /v2/auth/refresh:
    post:
      summary: "Refresh token"

What does this spec define?

This spec describes a fictional Acme E-Commerce API at version 2.0.0. It documents four resource groups: users, products, orders, and auth.

Every path under /v2/ with its allowed HTTP methods (GET, POST, PUT, DELETE) is considered "known" by your detector. Anything in traffic that does NOT match these paths is a potential shadow API.

  • Save openapi_spec.yaml.
  • Verify the YAML is valid by running this command in your terminal:
python3 -c "import yaml; print(yaml.safe_load(open('openapi_spec.yaml'))['info']['title'])"
  • You should see Acme E-Commerce API printed in your terminal.

Seeing a YAML error?

  • YAML is indentation-sensitive. Make sure every nested key uses exactly 2 spaces (not tabs).
  • Check that your virtual environment is still active (you should see (venv) in your prompt). If not, reactivate it with source venv/bin/activate.

Why is my YAML file failing to parse?

Build the traffic generator

Your detector needs traffic to analyze. Instead of requiring a live production gateway, you will build a generator that produces realistic NGINX combined-format access logs with planted shadow APIs mixed in.

Why simulated traffic?

Real NGINX logs require a production environment. Simulated data makes the project self-contained and reproducible.

Planting known shadow endpoints in the generated data lets you verify detection works correctly because you know the ground truth.

  • Open generate_traffic.py in your text editor and paste the following imports and endpoint configuration:
import random
import datetime


# Simulated browser and tool user agents
USER_AGENTS = [
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)",
    "curl/8.4.0",
    "python-requests/2.31.0",
    "PostmanRuntime/7.36.0",
    "Go-http-client/2.0",
]

# Endpoints that exist in the OpenAPI spec (legitimate traffic)
DOCUMENTED_ENDPOINTS = [
    ("GET", "/v2/users"),
    ("POST", "/v2/users"),
    ("GET", "/v2/users/{}"),
    ("PUT", "/v2/users/{}"),
    ("DELETE", "/v2/users/{}"),
    ("GET", "/v2/products"),
    ("POST", "/v2/products"),
    ("GET", "/v2/products/{}"),
    ("PUT", "/v2/products/{}"),
    ("GET", "/v2/orders"),
    ("POST", "/v2/orders"),
    ("GET", "/v2/orders/{}"),
    ("POST", "/v2/auth/login"),
    ("POST", "/v2/auth/logout"),
    ("POST", "/v2/auth/refresh"),
]

# Shadow APIs: undocumented endpoints the detector should catch
SHADOW_ENDPOINTS = [
    ("GET", "/v1/users"),
    ("GET", "/v1/users/{}"),
    ("POST", "/v1/orders"),
    ("GET", "/admin/dashboard"),
    ("POST", "/admin/users/export"),
    ("GET", "/internal/metrics"),
    ("GET", "/internal/health"),
    ("GET", "/debug/vars"),
    ("POST", "/debug/reset"),
    ("GET", "/v2/users/{}/ssn"),
    ("GET", "/v2/users/{}/email-raw"),
    ("DELETE", "/v2/products"),
    ("PATCH", "/v2/users/{}"),
    ("GET", "/api/v1/tokens"),
    ("POST", "/internal/admin/password-reset"),
]

# Potential backdoors: only appear during off-hours
BACKDOOR_ENDPOINTS = [
    ("POST", "/internal/sync"),
    ("GET", "/debug/dump"),
    ("POST", "/admin/shell"),
]

What are these endpoint lists?

  • DOCUMENTED_ENDPOINTS mirrors the paths in your OpenAPI spec. These represent normal, expected traffic.
  • SHADOW_ENDPOINTS contains undocumented paths like admin panels, deprecated v1 routes, debug endpoints, and PII-leaking paths. These are what your detector should flag.
  • BACKDOOR_ENDPOINTS are suspicious endpoints that only appear during off-hours (00:00-06:00). These will be relevant in the Secret Mission.
  • You won't see a change yet. The next chunk adds the functions that use these lists to generate log lines.
  • Below the BACKDOOR_ENDPOINTS list, add the helper functions that generate randomized log components:

def random_ip():
    return f"{random.randint(10, 192)}.{random.randint(0, 255)}.{random.randint(0, 255)}.{random.randint(1, 254)}"


def random_id():
    # Generate a random identifier (numeric, UUID, or hex hash)
    kind = random.choice(["numeric", "uuid", "hex"])
    if kind == "numeric":
        return str(random.randint(1, 99999))
    elif kind == "uuid":
        return f"{random.randbytes(4).hex()}-{random.randbytes(2).hex()}-{random.randbytes(2).hex()}-{random.randbytes(2).hex()}-{random.randbytes(6).hex()}"
    else:
        return random.randbytes(8).hex()


def format_log_line(timestamp, method, path, status, user_agent):
    # Format a single line in NGINX combined log format
    ip = random_ip()
    size = random.randint(50, 15000)
    time_str = timestamp.strftime("%d/%b/%Y:%H:%M:%S +0000")
    return f'{ip} - - [{time_str}] "{method} {path} HTTP/1.1" {status} {size} "-" "{user_agent}"'

What do these helpers do?

  • random_ip() creates fake client IP addresses.
  • random_id() produces dynamic path segments (numeric IDs, UUIDs, or hex hashes) that simulate real resource identifiers. Your parser will need to normalize these back into patterns later.
  • format_log_line() assembles a single line matching the NGINX combined log format: $remote_addr - $remote_user [$time_local] "$request" $status $body_bytes_sent "$http_referer" "$http_user_agent".
  • You won't see a change yet. The next chunk adds the main generation logic that ties everything together.
  • Below format_log_line, add the main log generation and file-writing functions:

def generate_logs(num_lines=600):
    lines = []
    base_time = datetime.datetime(2026, 6, 25, 8, 0, 0)

    for i in range(num_lines):
        timestamp = base_time + datetime.timedelta(seconds=random.randint(0, 86400))
        user_agent = random.choice(USER_AGENTS)

        # 70% of traffic goes to documented endpoints
        if random.random() < 0.70:
            method, path_template = random.choice(DOCUMENTED_ENDPOINTS)
            path = path_template.format(random_id()) if "{}" in path_template else path_template
            status = random.choices([200, 201, 204, 301, 400, 401, 404, 500], weights=[50, 10, 5, 2, 5, 3, 3, 2])[0]
        # ~15% goes to shadow endpoints
        elif random.random() < 0.85:
            method, path_template = random.choice(SHADOW_ENDPOINTS)
            path = path_template.format(random_id()) if "{}" in path_template else path_template
            status = random.choices([200, 403, 401, 500], weights=[40, 20, 20, 20])[0]
        # Remaining traffic goes to backdoor endpoints (off-hours only)
        else:
            method, path_template = random.choice(BACKDOOR_ENDPOINTS)
            path = path_template.format(random_id()) if "{}" in path_template else path_template
            hour = random.randint(0, 5)
            timestamp = base_time.replace(hour=hour, minute=random.randint(0, 59))
            status = random.choices([200, 201], weights=[70, 30])[0]
            user_agent = "curl/8.4.0"

        lines.append((timestamp, format_log_line(timestamp, method, path, status, user_agent)))

    # Sort all log lines chronologically
    lines.sort(key=lambda x: x[0])
    return [line for _, line in lines]


def write_logs(filepath="access.log"):
    logs = generate_logs()
    with open(filepath, "w") as f:
        for line in logs:
            f.write(line + "\n")
    print(f"Generated {len(logs)} log lines -> {filepath}")
    return filepath


if __name__ == "__main__":
    write_logs()

What does this code do?

  • generate_logs() creates 600 log lines spread across a 24-hour window. It distributes traffic with 70% going to documented endpoints, about 15% to shadow endpoints, and the rest to backdoor endpoints during off-hours only.
  • write_logs() calls the generator and writes all lines to access.log, sorted chronologically.
  • The if __name__ == "__main__" block lets you run this file directly to generate the log file.
  • Save generate_traffic.py.
  • Run the traffic generator by executing this command:
python3 generate_traffic.py
  • You should see Generated 600 log lines -> access.log in your terminal.

Not seeing the expected output?

  • Make sure your virtual environment is active ((venv) in your prompt). If not, run source venv/bin/activate first.
  • If you see a SyntaxError, double-check that you pasted all three code sections in order with no missing lines between them.
  • If you see NameError: name 'random' is not defined, confirm the import random line is at the very top of the file.

Why is my traffic generator throwing an error?

  • Confirm the log file was created by running:
head -5 access.log
  • You should see 5 lines of NGINX-formatted log entries, each containing an IP address, timestamp, HTTP method, path, status code, and user agent.

✔️ Awesome, I've got everything!

Great. Double check you have saved both openapi_spec.yaml and generate_traffic.py.

ⓧ I'd like to double check the full code

Here is the complete generate_traffic.py for comparison:

import random
import datetime


USER_AGENTS = [
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)",
    "curl/8.4.0",
    "python-requests/2.31.0",
    "PostmanRuntime/7.36.0",
    "Go-http-client/2.0",
]

DOCUMENTED_ENDPOINTS = [
    ("GET", "/v2/users"),
    ("POST", "/v2/users"),
    ("GET", "/v2/users/{}"),
    ("PUT", "/v2/users/{}"),
    ("DELETE", "/v2/users/{}"),
    ("GET", "/v2/products"),
    ("POST", "/v2/products"),
    ("GET", "/v2/products/{}"),
    ("PUT", "/v2/products/{}"),
    ("GET", "/v2/orders"),
    ("POST", "/v2/orders"),
    ("GET", "/v2/orders/{}"),
    ("POST", "/v2/auth/login"),
    ("POST", "/v2/auth/logout"),
    ("POST", "/v2/auth/refresh"),
]

SHADOW_ENDPOINTS = [
    ("GET", "/v1/users"),
    ("GET", "/v1/users/{}"),
    ("POST", "/v1/orders"),
    ("GET", "/admin/dashboard"),
    ("POST", "/admin/users/export"),
    ("GET", "/internal/metrics"),
    ("GET", "/internal/health"),
    ("GET", "/debug/vars"),
    ("POST", "/debug/reset"),
    ("GET", "/v2/users/{}/ssn"),
    ("GET", "/v2/users/{}/email-raw"),
    ("DELETE", "/v2/products"),
    ("PATCH", "/v2/users/{}"),
    ("GET", "/api/v1/tokens"),
    ("POST", "/internal/admin/password-reset"),
]

BACKDOOR_ENDPOINTS = [
    ("POST", "/internal/sync"),
    ("GET", "/debug/dump"),
    ("POST", "/admin/shell"),
]


def random_ip():
    return f"{random.randint(10, 192)}.{random.randint(0, 255)}.{random.randint(0, 255)}.{random.randint(1, 254)}"


def random_id():
    kind = random.choice(["numeric", "uuid", "hex"])
    if kind == "numeric":
        return str(random.randint(1, 99999))
    elif kind == "uuid":
        return f"{random.randbytes(4).hex()}-{random.randbytes(2).hex()}-{random.randbytes(2).hex()}-{random.randbytes(2).hex()}-{random.randbytes(6).hex()}"
    else:
        return random.randbytes(8).hex()


def format_log_line(timestamp, method, path, status, user_agent):
    ip = random_ip()
    size = random.randint(50, 15000)
    time_str = timestamp.strftime("%d/%b/%Y:%H:%M:%S +0000")
    return f'{ip} - - [{time_str}] "{method} {path} HTTP/1.1" {status} {size} "-" "{user_agent}"'


def generate_logs(num_lines=600):
    lines = []
    base_time = datetime.datetime(2026, 6, 25, 8, 0, 0)

    for i in range(num_lines):
        timestamp = base_time + datetime.timedelta(seconds=random.randint(0, 86400))
        user_agent = random.choice(USER_AGENTS)

        if random.random() < 0.70:
            method, path_template = random.choice(DOCUMENTED_ENDPOINTS)
            path = path_template.format(random_id()) if "{}" in path_template else path_template
            status = random.choices([200, 201, 204, 301, 400, 401, 404, 500], weights=[50, 10, 5, 2, 5, 3, 3, 2])[0]
        elif random.random() < 0.85:
            method, path_template = random.choice(SHADOW_ENDPOINTS)
            path = path_template.format(random_id()) if "{}" in path_template else path_template
            status = random.choices([200, 403, 401, 500], weights=[40, 20, 20, 20])[0]
        else:
            method, path_template = random.choice(BACKDOOR_ENDPOINTS)
            path = path_template.format(random_id()) if "{}" in path_template else path_template
            hour = random.randint(0, 5)
            timestamp = base_time.replace(hour=hour, minute=random.randint(0, 59))
            status = random.choices([200, 201], weights=[70, 30])[0]
            user_agent = "curl/8.4.0"

        lines.append((timestamp, format_log_line(timestamp, method, path, status, user_agent)))

    lines.sort(key=lambda x: x[0])
    return [line for _, line in lines]


def write_logs(filepath="access.log"):
    logs = generate_logs()
    with open(filepath, "w") as f:
        for line in logs:
            f.write(line + "\n")
    print(f"Generated {len(logs)} log lines -> {filepath}")
    return filepath


if __name__ == "__main__":
    write_logs()

You now have a documented API inventory and 600 lines of realistic traffic data sitting in access.log. Next up, you will build the parser that extracts and normalizes endpoints from these raw log lines.

Parse Logs and Normalize Endpoints

You have a traffic generator producing realistic NGINX access logs. Now you need a way to read those logs and extract meaningful data from each line.

Raw logs contain thousands of lines with dynamic path segments like user IDs and UUIDs. To detect shadow APIs, you need to collapse those dynamic segments into route patterns so that /users/12345 and /users/67890 both map to the same endpoint. In this step, you will build the log parser and endpoint normalizer.

In this step, get ready to:
  • Build a regex-based log parser that extracts fields from NGINX combined-format logs.
  • Implement endpoint normalization that collapses dynamic path segments into parameterized patterns.
  • Aggregate discovered endpoints into a unique inventory with hit counts and metadata.
Build the log parser regex

Every line in an NGINX access log follows a predictable structure. The combined log format is: $remote_addr - $remote_user [$time_local] "$request" $status $body_bytes_sent "$http_referer" "$http_user_agent".

You will write a regex that captures each of these fields from a raw log line, then build two functions that apply it line by line across the entire file.

  • Open parse_logs.py in your editor and add the imports and regex pattern by copying the following code:
import re
from collections import defaultdict


# Regex pattern matching the NGINX combined log format
LOG_PATTERN = re.compile(
    r'(?P<ip>\S+) - \S+ \[(?P<timestamp>[^\]]+)\] '
    r'"(?P<method>\S+) (?P<path>\S+) \S+" (?P<status>\d+) (?P<size>\d+) '
    r'"[^"]*" "(?P<user_agent>[^"]*)"'
)

What does this code do?

The LOG_PATTERN regex uses named capture groups (?P<ip>, ?P<timestamp>, etc.) to extract specific fields from each log line.

It matches the NGINX combined format: an IP address, a dash, a user field, a bracketed timestamp, a quoted request line (method + path + protocol), the status code, body size, a quoted referer, and a quoted user agent.

  • Below LOG_PATTERN, add the parse_log_line() function that applies the regex to a single line:
def parse_log_line(line):
    match = LOG_PATTERN.match(line)
    if not match:
        return None
    return {
        "ip": match.group("ip"),
        "timestamp": match.group("timestamp"),
        "method": match.group("method"),
        "path": match.group("path"),
        "status": int(match.group("status")),
        "user_agent": match.group("user_agent"),
    }

What does this function do?

The parse_log_line() function applies LOG_PATTERN to a single log line. If the regex matches, it returns a dictionary with the extracted fields. If not, it returns None so malformed lines are silently skipped.

  • Below parse_log_line(), add the parse_logs() function that reads an entire log file:
def parse_logs(filepath):
    entries = []
    with open(filepath, "r") as f:
        for line in f:
            parsed = parse_log_line(line.strip())
            if parsed:
                entries.append(parsed)
    return entries

What does this function do?

The parse_logs() function opens the log file, iterates through every line, calls parse_log_line() on each, and collects all successfully parsed entries into a list.

  • Save parse_logs.py.

Seeing a syntax error?

Check that the regex string uses raw strings (r'...'). Without the r prefix, Python interprets backslashes as escape characters, breaking the pattern.

Also confirm the two r'...' strings inside re.compile() are on separate lines with no comma between them. Python concatenates adjacent string literals automatically.

help me fix a regex syntax error in my NGINX log parser

Implement endpoint normalization

A real API receives traffic to paths like /v2/users/12345, /v2/users/67890, and /v2/orders/abc-def-123. These all represent the same route pattern, just with different IDs.

To compare discovered traffic against an API spec, you need to collapse those dynamic segments into a canonical form like /v2/users/{id}.

  • In parse_logs.py, add the ID_PATTERNS list directly below LOG_PATTERN (above the parse_log_line function):
# Heuristics to detect dynamic path segments (IDs, UUIDs, hashes)
ID_PATTERNS = [
    re.compile(r"^[0-9]+$"),
    re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$"),
    re.compile(r"^[0-9a-f]{16}$"),
    re.compile(r"^[0-9a-f]{8,}$"),
]

What do these patterns match?

  • The first pattern matches purely numeric segments like 12345.
  • The second matches standard UUIDs like a1b2c3d4-e5f6-7890-abcd-ef1234567890.
  • The third matches 16-character hex strings (short hashes).
  • The fourth catches any hex string 8+ characters long.
  • Directly below ID_PATTERNS, add the normalize_path() function:
def normalize_path(path):
    segments = path.strip("/").split("/")
    normalized = []
    for segment in segments:
        if any(pattern.match(segment) for pattern in ID_PATTERNS):
            normalized.append("{id}")
        else:
            normalized.append(segment)
    return "/" + "/".join(normalized)

What does normalize_path do?

It splits a path into segments, checks each segment against the ID_PATTERNS heuristics, and replaces any match with {id}. The result is a canonical route pattern.

For example, /v2/users/99381 becomes /v2/users/{id}, and /v2/orders/a1b2c3d4-e5f6-7890-abcd-ef1234567890 becomes /v2/orders/{id}.

  • Save parse_logs.py.

Unsure where to place the new code?

Your file should now have this order from top to bottom: imports, LOG_PATTERN, ID_PATTERNS, normalize_path(), parse_log_line(), parse_logs(). If anything is out of order, rearrange it now.

help me organize the functions in my parse_logs.py file

Aggregate endpoints into an inventory

Parsing individual lines gives you raw data. To detect shadows, you need an aggregated view: which unique route patterns appear in traffic, how often they are hit, and which HTTP methods are used.

  • Below parse_logs(), add the build_endpoint_inventory() function:
def build_endpoint_inventory(entries):
    inventory = defaultdict(lambda: {
        "methods": set(),
        "hit_count": 0,
        "status_codes": defaultdict(int),
        "first_seen": None,
        "last_seen": None,
        "timestamps": [],
    })

    for entry in entries:
        normalized = normalize_path(entry["path"])
        key = normalized
        record = inventory[key]
        record["methods"].add(entry["method"])
        record["hit_count"] += 1
        record["status_codes"][entry["status"]] += 1
        record["timestamps"].append(entry["timestamp"])

        if record["first_seen"] is None:
            record["first_seen"] = entry["timestamp"]
        record["last_seen"] = entry["timestamp"]

    return dict(inventory)

What does build_endpoint_inventory do?

It loops through every parsed log entry, normalizes the path, and groups entries by their canonical route pattern. For each pattern it tracks the set of HTTP methods seen, total hit count, a breakdown of status codes, and the first and last timestamps.

The defaultdict with a lambda factory creates a fresh record automatically the first time a new path is encountered.

  • At the very bottom of parse_logs.py, add a __main__ block so you can test the parser independently:
if __name__ == "__main__":
    entries = parse_logs("access.log")
    inventory = build_endpoint_inventory(entries)
    print(f"Parsed {len(entries)} log entries")
    print(f"Discovered {len(inventory)} unique endpoints")
    for path, data in sorted(inventory.items()):
        methods = ",".join(sorted(data["methods"]))
        print(f"  [{methods}] {path} ({data['hit_count']} hits)")

What does the __main__ block do?

It lets you run parse_logs.py directly to test parsing against an existing access.log file. It prints how many entries were parsed, how many unique endpoints were discovered, and lists each with its observed methods and hit count.

  • Save parse_logs.py.
  • First, generate a log file to test against by running:
python3 generate_traffic.py
  • Now run the parser against the generated logs:
python3 parse_logs.py
  • You should see output showing approximately 600 parsed log entries and around 15-25 unique endpoints, each listed with their HTTP methods and hit counts.

Seeing 'FileNotFoundError: access.log'?

Make sure you ran python3 generate_traffic.py first. The parser expects an access.log file in the same directory.

Also confirm your terminal is inside the shadow_detector/ folder. Run ls and check that you see generate_traffic.py and parse_logs.py listed.

help me fix a FileNotFoundError when running parse_logs.py

✔️ Awesome, I've got everything!

Great. Double check you have saved parse_logs.py and that the output shows both parsed entries and discovered endpoints.

ⓧ I'd like to double check the full code

Here is the complete parse_logs.py file:

import re
from collections import defaultdict


LOG_PATTERN = re.compile(
    r'(?P<ip>\S+) - \S+ \[(?P<timestamp>[^\]]+)\] '
    r'"(?P<method>\S+) (?P<path>\S+) \S+" (?P<status>\d+) (?P<size>\d+) '
    r'"[^"]*" "(?P<user_agent>[^"]*)"'
)

ID_PATTERNS = [
    re.compile(r"^[0-9]+$"),
    re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$"),
    re.compile(r"^[0-9a-f]{16}$"),
    re.compile(r"^[0-9a-f]{8,}$"),
]


def normalize_path(path):
    segments = path.strip("/").split("/")
    normalized = []
    for segment in segments:
        if any(pattern.match(segment) for pattern in ID_PATTERNS):
            normalized.append("{id}")
        else:
            normalized.append(segment)
    return "/" + "/".join(normalized)


def parse_log_line(line):
    match = LOG_PATTERN.match(line)
    if not match:
        return None
    return {
        "ip": match.group("ip"),
        "timestamp": match.group("timestamp"),
        "method": match.group("method"),
        "path": match.group("path"),
        "status": int(match.group("status")),
        "user_agent": match.group("user_agent"),
    }


def parse_logs(filepath):
    entries = []
    with open(filepath, "r") as f:
        for line in f:
            parsed = parse_log_line(line.strip())
            if parsed:
                entries.append(parsed)
    return entries


def build_endpoint_inventory(entries):
    inventory = defaultdict(lambda: {
        "methods": set(),
        "hit_count": 0,
        "status_codes": defaultdict(int),
        "first_seen": None,
        "last_seen": None,
        "timestamps": [],
    })

    for entry in entries:
        normalized = normalize_path(entry["path"])
        key = normalized
        record = inventory[key]
        record["methods"].add(entry["method"])
        record["hit_count"] += 1
        record["status_codes"][entry["status"]] += 1
        record["timestamps"].append(entry["timestamp"])

        if record["first_seen"] is None:
            record["first_seen"] = entry["timestamp"]
        record["last_seen"] = entry["timestamp"]

    return dict(inventory)


if __name__ == "__main__":
    entries = parse_logs("access.log")
    inventory = build_endpoint_inventory(entries)
    print(f"Parsed {len(entries)} log entries")
    print(f"Discovered {len(inventory)} unique endpoints")
    for path, data in sorted(inventory.items()):
        methods = ",".join(sorted(data["methods"]))
        print(f"  [{methods}] {path} ({data['hit_count']} hits)")

Your log parser is ready. It can read hundreds of NGINX access log lines, extract structured data, and collapse them into a clean endpoint inventory. Next up, you will load the OpenAPI spec and diff it against this inventory to flag the shadow APIs.

Detect Shadow APIs and Store Findings

You now have a log parser that discovers every endpoint your API gateway sees. The next challenge is figuring out which of those endpoints are legitimate and which are undocumented shadows lurking outside your official inventory.

In this step, you will load your OpenAPI spec as the "known good" list, diff the discovered traffic against it, score each shadow endpoint for risk, and store the results in a SQLite database.

In this step, get ready to:
  • Build the OpenAPI spec loader and shadow detection engine.
  • Implement risk scoring that evaluates authentication gaps, PII exposure, and version signals.
  • Store findings in SQLite and generate a prioritized terminal report.
Build the OpenAPI spec loader and shadow detection engine

The detection logic has two parts. First, you need to read your OpenAPI spec to build a dictionary of every documented path and its allowed HTTP methods. Second, you compare the traffic inventory against that dictionary to find endpoints that should not exist.

  • Open detect_shadows.py in your editor.
  • Add the imports and the spec loader function by pasting this code:
import yaml
import sqlite3
import datetime


def load_openapi_spec(filepath):
    with open(filepath, "r") as f:
        spec = yaml.safe_load(f)

    documented = {}
    paths = spec.get("paths", {})
    for path, methods in paths.items():
        normalized = path.replace("{id}", "{id}")
        documented[normalized] = set()
        for method in methods:
            if method in ("get", "post", "put", "delete", "patch", "head", "options"):
                documented[normalized].add(method.upper())

    return documented

What does this code do?

The load_openapi_spec() function uses PyYAML to parse your spec file into a Python dictionary.

  • It walks the paths section and collects each route along with its allowed HTTP methods (uppercased to match the traffic data).
  • Only standard HTTP verbs are included. OpenAPI metadata keys like summary or parameters are ignored.
  • The result is a dict like {"/v2/users": {"GET", "POST"}, ...} representing your official API inventory.

Now you have the "known good" list. The next function compares it against what you discovered in traffic.

  • Add the detection engine below load_openapi_spec() by pasting this code:

def detect_shadows(inventory, documented):
    findings = []

    for path, data in inventory.items():
        if path in documented:
            undocumented_methods = data["methods"] - documented[path]
            if undocumented_methods:
                for method in undocumented_methods:
                    findings.append({
                        "path": path,
                        "method": method,
                        "category": "method_mismatch",
                        "hit_count": data["hit_count"],
                        "first_seen": data["first_seen"],
                        "last_seen": data["last_seen"],
                        "timestamps": data["timestamps"],
                        "status_codes": dict(data["status_codes"]),
                    })
        else:
            if "/v1/" in path:
                category = "zombie"
            else:
                category = "shadow"

            for method in data["methods"]:
                findings.append({
                    "path": path,
                    "method": method,
                    "category": category,
                    "hit_count": data["hit_count"],
                    "first_seen": data["first_seen"],
                    "last_seen": data["last_seen"],
                    "timestamps": data["timestamps"],
                    "status_codes": dict(data["status_codes"]),
                })

    return findings

What does the detection logic do?

For each endpoint discovered in traffic, the function checks whether the path exists in the documented spec.

  • If the path IS documented but the HTTP method is not (e.g. a DELETE on a read-only route), it is classified as method_mismatch.
  • If the path is NOT in the spec and contains /v1/, it is a zombie. These are deprecated API versions that were never fully decommissioned.
  • Everything else not in the spec is a shadow. These endpoints were never documented at all.
  • Save detect_shadows.py.

You will not see output yet. The detection engine needs the rest of the pipeline (risk scoring and storage) before you can run a full test.

Seeing a syntax error on save?

  • Make sure the detect_shadows() function starts after the closing return documented line of load_openapi_spec(). There should be two blank lines between the functions.
  • Check that you pasted both chunks in order. The imports must be at the very top of the file.

Why is my detect_shadows.py showing an error?

Implement risk scoring

Not all shadow APIs are equally dangerous. An undocumented debug endpoint that returns 200s without authentication is far more urgent than a zombie route that consistently 403s. The risk scorer evaluates three dimensions to produce a 0-10 score.

  • Add the score_risk() function below detect_shadows() by pasting this code:

def score_risk(finding):
    score = 0
    details = []

    status_codes = finding.get("status_codes", {})
    success_codes = sum(v for k, v in status_codes.items() if 200 <= k < 300)
    total = sum(status_codes.values())
    if total > 0 and success_codes / total > 0.5:
        score += 4
        details.append("unauthenticated_success")
    elif total > 0 and success_codes / total > 0.2:
        score += 2
        details.append("partial_auth_bypass")

    pii_patterns = ["ssn", "email-raw", "password", "token", "admin", "internal", "secret"]
    path_lower = finding["path"].lower()
    for pattern in pii_patterns:
        if pattern in path_lower:
            score += 3
            details.append(f"pii_pattern:{pattern}")
            break

    if "/v1/" in finding["path"]:
        score += 2
        details.append("deprecated_version")
    if "/debug/" in finding["path"]:
        score += 3
        details.append("debug_endpoint")
    elif "/test/" in finding["path"]:
        score += 1
        details.append("test_endpoint")

    finding["risk_score"] = min(score, 10)
    finding["details"] = ", ".join(details)
    return finding

How does risk scoring work?

The scorer evaluates each shadow endpoint on three dimensions, each contributing points up to a maximum of 10:

  • Authentication gap (0-4 points): If more than 50% of responses are 2xx successes, the endpoint likely allows unauthenticated access. That is a critical signal.
  • PII exposure (0-3 points): Path segments like ssn, password, or admin suggest sensitive data flows.
  • Version/debug risk (0-3 points): Deprecated /v1/ routes and /debug/ paths are high-value targets for attackers.
  • Save detect_shadows.py.

You will not see a change yet. The next chunk adds the database and report so you can run a full test.

Store findings in SQLite and generate the report

Findings need a permanent home so you can query them later. SQLite is built into Python's standard library, so no extra installation is needed. You will also add a report function that prints a prioritized table to the terminal.

  • Add the database functions below score_risk() by pasting this code:

def init_database(db_path="findings.db"):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS findings (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            path TEXT NOT NULL,
            method TEXT NOT NULL,
            category TEXT NOT NULL,
            hit_count INTEGER,
            first_seen TEXT,
            last_seen TEXT,
            risk_score INTEGER,
            details TEXT
        )
    """)
    cursor.execute("DELETE FROM findings")
    conn.commit()
    return conn


def store_findings(conn, findings):
    cursor = conn.cursor()
    for f in findings:
        cursor.execute(
            "INSERT INTO findings (path, method, category, hit_count, first_seen, last_seen, risk_score, details) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
            (f["path"], f["method"], f["category"], f["hit_count"], f["first_seen"], f["last_seen"], f["risk_score"], f["details"]),
        )
    conn.commit()

What do these database functions do?

  • init_database() creates (or connects to) a findings.db file, sets up the findings table schema, and clears any old data so each run starts fresh.
  • store_findings() inserts each scored finding as a row. The ? placeholders prevent SQL injection. Even in a local tool, parameterized queries are a good habit.

The last piece is the report function that queries the database and prints a prioritized table.

  • Add the report generator below store_findings() by pasting this code:

def print_report(conn):
    cursor = conn.cursor()
    cursor.execute("SELECT COUNT(*) FROM findings")
    total = cursor.fetchone()[0]

    cursor.execute("SELECT COUNT(*) FROM findings WHERE category = 'shadow'")
    shadows = cursor.fetchone()[0]

    cursor.execute("SELECT COUNT(*) FROM findings WHERE category = 'zombie'")
    zombies = cursor.fetchone()[0]

    cursor.execute("SELECT COUNT(*) FROM findings WHERE category = 'method_mismatch'")
    mismatches = cursor.fetchone()[0]

    print("\n" + "=" * 70)
    print("  SHADOW API DETECTION REPORT")
    print("=" * 70)
    print(f"\n  Total findings: {total}")
    print(f"  Shadow APIs:    {shadows}")
    print(f"  Zombie APIs:    {zombies}")
    print(f"  Method issues:  {mismatches}")
    print("\n" + "-" * 70)
    print(f"  {'RISK':<6}{'METHOD':<8}{'CATEGORY':<18}{'HITS':<8}{'PATH'}")
    print("-" * 70)

    cursor.execute("SELECT risk_score, method, category, hit_count, path, details FROM findings ORDER BY risk_score DESC")
    for row in cursor.fetchall():
        risk, method, category, hits, path, details = row
        risk_label = f"[{risk}/10]"
        print(f"  {risk_label:<6}{method:<8}{category:<18}{hits:<8}{path}")
        if details:
            print(f"  {'':6}{'':8}-> {details}")

    print("\n" + "=" * 70)

What does the report show?

The report queries the findings table and prints a summary of total findings broken down by category. Below the summary, it lists every finding sorted by risk score (highest first) with the method, category, hit count, path, and the scoring details that explain why the score is what it is.

  • Save detect_shadows.py.
  • Verify the file has no syntax errors by running:
python3 -c "import detect_shadows; print('All functions loaded successfully')"

You should see All functions loaded successfully printed to your terminal.

Seeing a ModuleNotFoundError for yaml?

  • Make sure your virtual environment is active. You should see (venv) at the start of your terminal prompt.
  • If the virtual environment is active and you still see the error, re-run pip install PyYAML==6.0.3 to reinstall the package.

Help me fix my PyYAML import error in detect_shadows.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved detect_shadows.py and the import check printed successfully.

ⓧ I'd like to double check the full code

Here is the complete detect_shadows.py file. Compare it against yours:

import yaml
import sqlite3
import datetime


def load_openapi_spec(filepath):
    with open(filepath, "r") as f:
        spec = yaml.safe_load(f)

    documented = {}
    paths = spec.get("paths", {})
    for path, methods in paths.items():
        normalized = path.replace("{id}", "{id}")
        documented[normalized] = set()
        for method in methods:
            if method in ("get", "post", "put", "delete", "patch", "head", "options"):
                documented[normalized].add(method.upper())

    return documented


def detect_shadows(inventory, documented):
    findings = []

    for path, data in inventory.items():
        if path in documented:
            undocumented_methods = data["methods"] - documented[path]
            if undocumented_methods:
                for method in undocumented_methods:
                    findings.append({
                        "path": path,
                        "method": method,
                        "category": "method_mismatch",
                        "hit_count": data["hit_count"],
                        "first_seen": data["first_seen"],
                        "last_seen": data["last_seen"],
                        "timestamps": data["timestamps"],
                        "status_codes": dict(data["status_codes"]),
                    })
        else:
            if "/v1/" in path:
                category = "zombie"
            else:
                category = "shadow"

            for method in data["methods"]:
                findings.append({
                    "path": path,
                    "method": method,
                    "category": category,
                    "hit_count": data["hit_count"],
                    "first_seen": data["first_seen"],
                    "last_seen": data["last_seen"],
                    "timestamps": data["timestamps"],
                    "status_codes": dict(data["status_codes"]),
                })

    return findings


def score_risk(finding):
    score = 0
    details = []

    status_codes = finding.get("status_codes", {})
    success_codes = sum(v for k, v in status_codes.items() if 200 <= k < 300)
    total = sum(status_codes.values())
    if total > 0 and success_codes / total > 0.5:
        score += 4
        details.append("unauthenticated_success")
    elif total > 0 and success_codes / total > 0.2:
        score += 2
        details.append("partial_auth_bypass")

    pii_patterns = ["ssn", "email-raw", "password", "token", "admin", "internal", "secret"]
    path_lower = finding["path"].lower()
    for pattern in pii_patterns:
        if pattern in path_lower:
            score += 3
            details.append(f"pii_pattern:{pattern}")
            break

    if "/v1/" in finding["path"]:
        score += 2
        details.append("deprecated_version")
    if "/debug/" in finding["path"]:
        score += 3
        details.append("debug_endpoint")
    elif "/test/" in finding["path"]:
        score += 1
        details.append("test_endpoint")

    finding["risk_score"] = min(score, 10)
    finding["details"] = ", ".join(details)
    return finding


def init_database(db_path="findings.db"):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS findings (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            path TEXT NOT NULL,
            method TEXT NOT NULL,
            category TEXT NOT NULL,
            hit_count INTEGER,
            first_seen TEXT,
            last_seen TEXT,
            risk_score INTEGER,
            details TEXT
        )
    """)
    cursor.execute("DELETE FROM findings")
    conn.commit()
    return conn


def store_findings(conn, findings):
    cursor = conn.cursor()
    for f in findings:
        cursor.execute(
            "INSERT INTO findings (path, method, category, hit_count, first_seen, last_seen, risk_score, details) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
            (f["path"], f["method"], f["category"], f["hit_count"], f["first_seen"], f["last_seen"], f["risk_score"], f["details"]),
        )
    conn.commit()


def print_report(conn):
    cursor = conn.cursor()
    cursor.execute("SELECT COUNT(*) FROM findings")
    total = cursor.fetchone()[0]

    cursor.execute("SELECT COUNT(*) FROM findings WHERE category = 'shadow'")
    shadows = cursor.fetchone()[0]

    cursor.execute("SELECT COUNT(*) FROM findings WHERE category = 'zombie'")
    zombies = cursor.fetchone()[0]

    cursor.execute("SELECT COUNT(*) FROM findings WHERE category = 'method_mismatch'")
    mismatches = cursor.fetchone()[0]

    print("\n" + "=" * 70)
    print("  SHADOW API DETECTION REPORT")
    print("=" * 70)
    print(f"\n  Total findings: {total}")
    print(f"  Shadow APIs:    {shadows}")
    print(f"  Zombie APIs:    {zombies}")
    print(f"  Method issues:  {mismatches}")
    print("\n" + "-" * 70)
    print(f"  {'RISK':<6}{'METHOD':<8}{'CATEGORY':<18}{'HITS':<8}{'PATH'}")
    print("-" * 70)

    cursor.execute("SELECT risk_score, method, category, hit_count, path, details FROM findings ORDER BY risk_score DESC")
    for row in cursor.fetchall():
        risk, method, category, hits, path, details = row
        risk_label = f"[{risk}/10]"
        print(f"  {risk_label:<6}{method:<8}{category:<18}{hits:<8}{path}")
        if details:
            print(f"  {'':6}{'':8}-> {details}")

    print("\n" + "=" * 70)

Your detection engine is complete. It can load a spec, find shadows, score risk, persist findings, and print a report. Next up, you will wire everything together in main.py to run the full pipeline end to end.

Wire the Pipeline and Run the Detector

You have built every module of the detection pipeline as a standalone piece. The traffic generator creates realistic logs, the parser normalizes endpoints, the spec loader builds an inventory, and the detection engine diffs and scores findings.

Now you need an orchestrator that calls each module in sequence and feeds data from one stage into the next. In this step, you will write main.py as a CLI entry point that runs the full 5-stage pipeline and prints a prioritized report.

In this step, get ready to:
  • Wire all modules into a single CLI entry point.
  • Run the full shadow API detection pipeline.
  • Review the prioritized findings report and query the SQLite database.
Write the pipeline imports

The main.py file needs access to every function you have built across the project. Python's os.path module lets you resolve file paths relative to the script itself, so the detector works regardless of where you run it from.

  • Open main.py in your editor and add the following imports:
import os
import sys

from generate_traffic import write_logs
from parse_logs import parse_logs, build_endpoint_inventory
from detect_shadows import (
    load_openapi_spec,
    detect_shadows,
    score_risk,
    init_database,
    store_findings,
    print_report,
)

What does this code do?

The os and sys imports give you path resolution utilities. The three from ... import lines pull in every function you built across the project's modules.

Each import maps to a pipeline stage: write_logs generates traffic, parse_logs and build_endpoint_inventory extract endpoints, and the detect_shadows group handles spec loading, diffing, scoring, storage, and reporting.

  • Save main.py.
Build the pipeline orchestrator function

The main() function runs five stages in sequence: generate traffic, parse logs, load the spec, detect shadows, then score and store findings. Each stage prints progress so you can see the pipeline advancing.

  • Add the following main() function below your imports in main.py:

def main():
    print("Shadow API Detector v1.0")
    print("=" * 40)

    # Resolve file paths relative to this script's directory
    script_dir = os.path.dirname(os.path.abspath(__file__))
    log_path = os.path.join(script_dir, "access.log")
    spec_path = os.path.join(script_dir, "openapi_spec.yaml")
    db_path = os.path.join(script_dir, "findings.db")

    # Stage 1: Generate simulated gateway traffic
    print("\n[1/5] Generating simulated traffic...")
    write_logs(log_path)

    # Stage 2: Parse logs and build endpoint inventory
    print("\n[2/5] Parsing access logs...")
    entries = parse_logs(log_path)
    inventory = build_endpoint_inventory(entries)
    print(f"  Parsed {len(entries)} log entries")
    print(f"  Discovered {len(inventory)} unique endpoints")

    # Stage 3: Load the documented API spec
    print("\n[3/5] Loading OpenAPI spec...")
    documented = load_openapi_spec(spec_path)
    print(f"  Documented endpoints: {sum(len(m) for m in documented.values())}")

    # Stage 4: Diff discovered endpoints against the spec
    print("\n[4/5] Detecting shadow APIs...")
    findings = detect_shadows(inventory, documented)
    print(f"  Found {len(findings)} undocumented endpoint/method combinations")

    # Stage 5: Score risks, store in SQLite, print report
    print("\n[5/5] Scoring risks and storing findings...")
    scored = [score_risk(f) for f in findings]

    conn = init_database(db_path)
    store_findings(conn, scored)

    print_report(conn)
    conn.close()

    print(f"\n  Results saved to: {db_path}")
    print("  Run `sqlite3 findings.db 'SELECT * FROM findings ORDER BY risk_score DESC'` to query directly.\n")


if __name__ == "__main__":
    main()

What does this code do?

  • os.path.dirname(os.path.abspath(__file__)) finds the folder where main.py lives. All file paths are built relative to that folder so the script works no matter which directory you run it from.
  • Stages 1-5 call the functions you already built in order, passing output from one stage as input to the next.
  • The progress messages ([1/5], [2/5], etc.) let you track which stage is running.
  • The if __name__ == "__main__" guard ensures main() only runs when you execute the file directly, not when it is imported.
  • Save main.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved your file before running the pipeline.

ⓧ I'd like to double check the full code

import os
import sys

from generate_traffic import write_logs
from parse_logs import parse_logs, build_endpoint_inventory
from detect_shadows import (
    load_openapi_spec,
    detect_shadows,
    score_risk,
    init_database,
    store_findings,
    print_report,
)


def main():
    print("Shadow API Detector v1.0")
    print("=" * 40)

    script_dir = os.path.dirname(os.path.abspath(__file__))
    log_path = os.path.join(script_dir, "access.log")
    spec_path = os.path.join(script_dir, "openapi_spec.yaml")
    db_path = os.path.join(script_dir, "findings.db")

    print("\n[1/5] Generating simulated traffic...")
    write_logs(log_path)

    print("\n[2/5] Parsing access logs...")
    entries = parse_logs(log_path)
    inventory = build_endpoint_inventory(entries)
    print(f"  Parsed {len(entries)} log entries")
    print(f"  Discovered {len(inventory)} unique endpoints")

    print("\n[3/5] Loading OpenAPI spec...")
    documented = load_openapi_spec(spec_path)
    print(f"  Documented endpoints: {sum(len(m) for m in documented.values())}")

    print("\n[4/5] Detecting shadow APIs...")
    findings = detect_shadows(inventory, documented)
    print(f"  Found {len(findings)} undocumented endpoint/method combinations")

    print("\n[5/5] Scoring risks and storing findings...")
    scored = [score_risk(f) for f in findings]

    conn = init_database(db_path)
    store_findings(conn, scored)

    print_report(conn)
    conn.close()

    print(f"\n  Results saved to: {db_path}")
    print("  Run `sqlite3 findings.db 'SELECT * FROM findings ORDER BY risk_score DESC'` to query directly.\n")


if __name__ == "__main__":
    main()
Run the full detection pipeline

With all five modules wired together, a single command now generates traffic, parses it, loads the spec, detects shadows, scores risks, and prints the report.

  • Run the detector from inside your shadow_detector folder by running this command:
python main.py

You should see progress messages for each of the five pipeline stages, followed by the full Shadow API Detection Report table sorted by risk score.

What should I see?

The output starts with a banner (Shadow API Detector v1.0) and progress for each stage. After stage 5, the full report prints with a table showing each finding's risk score, HTTP method, category, hit count, and path.

Look for findings with [10/10] or [7/10] risk scores at the top. These are the most dangerous shadow endpoints. Paths like /debug/vars and /internal/admin/password-reset should score highest because they combine multiple risk dimensions.

Seeing a ModuleNotFoundError?

  • Make sure you are running the command from inside the shadow_detector folder. Python resolves imports relative to the current working directory.
  • Confirm your virtual environment is active. You should see (venv) at the start of your terminal prompt.
  • Double check that all five .py files are saved in the same folder.

Help me debug a ModuleNotFoundError when running main.py

Query the findings database directly

The pipeline saved every finding to a SQLite database called findings.db. You can query it directly to slice the data however you want.

  • Query the top 5 riskiest findings by running this command:
sqlite3 findings.db 'SELECT risk_score, method, category, path FROM findings ORDER BY risk_score DESC LIMIT 5'

You should see a table of the five highest-risk shadow APIs with their scores, methods, categories, and paths.

Why store findings in SQLite?

A database makes findings queryable after the scan completes. You can filter by category, sort by risk, or join with other data sources. Commercial API security tools use the same approach to let analysts investigate findings long after the initial detection run.

Command not found for sqlite3?

  • On macOS, sqlite3 is pre-installed. If it is missing, install it with brew install sqlite.
  • Make sure findings.db exists in your current folder. Running ls findings.db should show the file.

Help me query my SQLite findings database

Your shadow API detector is fully operational. You have a working security tool that discovers undocumented endpoints from gateway traffic and prioritizes them by risk. Next up, the Secret Mission challenges you to add temporal anomaly detection for potential backdoors.

Secret mission

Add Temporal Anomaly Detection

Your shadow API detector flags undocumented endpoints. But what about endpoints that only appear at 3 AM? Attackers often deploy backdoors that activate during off-hours when no one is watching. Add temporal analysis to catch them.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your project files for future reference, or delete them to free up disk space.

Resources you used:

  • The shadow_detector/ project folder on your Desktop (Python scripts, YAML spec, virtual environment, generated logs, and SQLite database).

Keep everything running

No action needed. Choose this if you want to keep experimenting with the shadow API detector, tweak the risk scoring, or add more detection heuristics.

  • Your project folder stays on your Desktop at shadow_detector/.
  • The virtual environment, generated access.log, and findings.db remain inside the folder.
  • To reactivate the virtual environment later, navigate into the folder and run source venv/bin/activate.

Pause - I'll come back to this later

Deactivate the virtual environment to free up your terminal session. Your files stay on disk for when you return.

  • Deactivate the virtual environment by running:
deactivate
  • Your project folder and all files remain on your Desktop. When you come back, navigate into shadow_detector/ and reactivate the environment with source venv/bin/activate.

Delete - I don't want to use this again

Remove all project files, the virtual environment, and generated data. This frees up disk space completely.

  • Deactivate the virtual environment if it is still active:
deactivate
  • Navigate to your Desktop:
cd ~/Desktop
  • Delete the entire project folder:
rm -rf shadow_detector
  • Confirm the folder is gone by running:
ls
  • You should no longer see shadow_detector in the output.

Nice Work!

Nice Work!

You just built a Shadow API Detector that discovers undocumented attack surface from network telemetry. This is the same passive-traffic-analysis approach used by commercial API security platforms.

You've learned how to:

  • Parse NGINX combined-format access logs and normalize dynamic path segments into parameterized route patterns for endpoint discovery.
  • Diff discovered traffic against an OpenAPI spec to classify undocumented endpoints as shadow, zombie, or method-mismatch findings.
  • Score risk across authentication gaps, PII exposure patterns, and deprecated version prefixes, then store structured findings in SQLite with a prioritized terminal report.
  • Secret Mission: Added temporal anomaly detection to flag potential backdoor endpoints that appear only during off-hours.

Ready to quiz yourself?