Building a Custom Scanner
Create your own vulnerability scanner that integrates with PhantomYerra's AI orchestrator, evidence pipeline, and reporting system. Your scanner runs alongside the built-in engines and produces findings in the same format.
The BaseToolAdapter Interface
Every custom scanner must subclass BaseToolAdapter. This base class defines the contract that PhantomYerra's orchestrator uses to discover, configure, and run your scanner.
Required Attributes
| Attribute | Type | Description |
|---|---|---|
TOOL_NAME | str | A unique identifier for your scanner (e.g., "my_custom_scanner"). Must be lowercase with underscores. Used internally for routing and logging. |
DISPLAY_NAME | str | Human-readable name shown in the UI (e.g., "My Custom Scanner"). |
DESCRIPTION | str | One-line description of what the scanner tests for. |
SURFACES | list[str] | Attack surfaces this scanner handles. Must match registered surface names (e.g., ["web_app", "api", "graphql"]). |
SEVERITY_RANGE | tuple | Min and max severity this scanner can report: ("info", "critical"). |
VERSION | str | Scanner version string (e.g., "1.0.0"). |
Required Methods
| Method | Signature | Description |
|---|---|---|
scan() |
async def scan(self, target: str, config: dict) -> list[Finding] |
The main entry point. Receives the target URL/host and configuration. Returns a list of Finding objects. This is where all scanning logic lives. |
is_applicable() |
async def is_applicable(self, target: str, fingerprint: dict) -> bool |
Called by the orchestrator before scan(). Return True if your scanner is relevant for this target based on its technology fingerprint. Return False to skip. |
health_check() |
async def health_check(self) -> bool |
Verify that your scanner's dependencies are available. Called on startup. Return True if ready, False if not (with an error logged). |
Optional Methods
| Method | Description |
|---|---|
configure(config) | Receive user-defined configuration options (e.g., aggressiveness level, custom wordlists). |
cleanup() | Called after scanning completes. Clean up temporary files, close connections. |
get_config_schema() | Return a JSON Schema describing configuration options. Used to render settings UI. |
Minimal Scanner Example
Below is a complete, minimal custom scanner that checks for exposed configuration files:
"""
Custom scanner: Exposed Configuration File Finder
Checks for common configuration files left accessible on web servers.
"""
from phantomyerra.sdk.base import BaseToolAdapter
from phantomyerra.sdk.finding import Finding
from phantomyerra.sdk.evidence import Evidence
from phantomyerra.sdk.events import emit_activity
import httpx
class ConfigFileFinder(BaseToolAdapter):
TOOL_NAME = "config_file_finder"
DISPLAY_NAME = "Configuration File Finder"
DESCRIPTION = "Detects exposed configuration files on web servers"
SURFACES = ["web_app"]
SEVERITY_RANGE = ("info", "high")
VERSION = "1.0.0"
# Paths to check for exposed config files
CONFIG_PATHS = [
"/.env", "/config.yml", "/config.json", "/wp-config.php.bak",
"/application.properties", "/appsettings.json", "/.git/config",
"/web.config", "/.htaccess", "/phpinfo.php", "/server-status",
"/elmah.axd", "/.well-known/security.txt", "/crossdomain.xml",
"/sitemap.xml", "/robots.txt", "/.DS_Store", "/Thumbs.db",
"/backup.sql", "/dump.sql", "/database.sql",
]
async def is_applicable(self, target, fingerprint):
"""Applicable to any HTTP/HTTPS target."""
return target.startswith("http://") or target.startswith("https://")
async def health_check(self):
"""No external dependencies required."""
return True
async def scan(self, target, config):
findings = []
total = len(self.CONFIG_PATHS)
emit_activity(
tool=self.TOOL_NAME,
phase="scanning",
message=f"Checking {total} configuration file paths",
percent=0,
)
async with httpx.AsyncClient(
timeout=10, verify=False, follow_redirects=True
) as client:
for i, path in enumerate(self.CONFIG_PATHS):
url = target.rstrip("/") + path
try:
resp = await client.get(url)
if resp.status_code == 200 and len(resp.content) > 0:
# Determine severity based on content
severity = self._assess_severity(path, resp.text)
finding = Finding(
title=f"Exposed Configuration File: {path}",
severity=severity,
description=(
f"The file {path} is publicly accessible and "
f"returned a {resp.status_code} response with "
f"{len(resp.content)} bytes of content."
),
evidence=Evidence(
request=f"GET {url} HTTP/1.1",
response_status=resp.status_code,
response_body=resp.text[:2000],
response_headers=dict(resp.headers),
),
remediation=(
f"Remove or restrict access to {path}. "
"Configure the web server to deny access to "
"configuration files, backup files, and "
"version control directories."
),
poc_steps=[
f"1. Navigate to {url}",
"2. Observe the file contents are returned",
"3. Check for sensitive data (credentials, "
"API keys, database connection strings)",
],
cvss_vector=self._get_cvss(severity),
)
findings.append(finding)
except httpx.RequestError:
pass # Connection error, skip this path
# Emit progress
pct = int(((i + 1) / total) * 100)
emit_activity(
tool=self.TOOL_NAME,
phase="scanning",
message=f"Checked {i + 1}/{total} paths",
percent=pct,
)
emit_activity(
tool=self.TOOL_NAME,
phase="complete",
message=f"Found {len(findings)} exposed files",
percent=100,
)
return findings
def _assess_severity(self, path, content):
"""Assess severity based on file type and content indicators."""
high_indicators = ["password", "secret", "api_key", "private_key",
"AWS_ACCESS", "DATABASE_URL", "DB_PASSWORD"]
if any(ind.lower() in content.lower() for ind in high_indicators):
return "high"
sensitive_paths = ["/.env", "/.git/config", "/wp-config.php.bak",
"/backup.sql", "/dump.sql", "/database.sql"]
if path in sensitive_paths:
return "medium"
return "low"
def _get_cvss(self, severity):
cvss_map = {
"critical": "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:N",
"high": "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:N/A:N",
"medium": "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:L/I:N/A:N",
"low": "CVSS:3.1/AV:N/AC:L/PR:N/UI:R/S:U/C:L/I:N/A:N",
"info": "CVSS:3.1/AV:N/AC:H/PR:N/UI:R/S:U/C:N/I:N/A:N",
}
return cvss_map.get(severity, cvss_map["low"])
Registering Your Scanner
In your extension's __init__.py, register the scanner with the surface registry:
"""Extension entry point — register the custom scanner."""
from phantomyerra.sdk.registry import surface_registry
from .scanner import ConfigFileFinder
def register():
"""Called by PhantomYerra on startup to register this extension."""
scanner = ConfigFileFinder()
surface_registry.register(
tool=scanner,
surfaces=scanner.SURFACES,
priority=50, # 0-100, higher = runs earlier in scan order
)
Registration Parameters
| Parameter | Type | Description |
|---|---|---|
tool | BaseToolAdapter | An instance of your scanner class. |
surfaces | list[str] | Attack surfaces to register for. Valid values: web_app, api, graphql, network, cloud, mobile, firmware, source_code, container, active_directory. |
priority | int | Execution priority (0-100). Higher priority scanners run first. Built-in scanners use 70-90. Use 50 for default, higher if your scanner should run early (e.g., reconnaissance). |
The Finding Object
Every vulnerability your scanner detects must be wrapped in a Finding object. This ensures uniform presentation in the UI, reports, and API.
Finding Fields
| Field | Type | Required | Description |
|---|---|---|---|
title | str | Yes | Short, descriptive title for the vulnerability. |
severity | str | Yes | One of: critical, high, medium, low, info. |
description | str | Yes | Detailed description of the vulnerability: what it is, where it was found, why it matters. |
evidence | Evidence | Yes | Proof of the vulnerability. At minimum: the HTTP request and response that demonstrates the issue. |
remediation | str | Yes | Specific, actionable remediation steps. Not generic advice. |
poc_steps | list[str] | Yes | Numbered reproduction steps. Must be copy-paste ready. |
cvss_vector | str | Recommended | CVSS v3.1 vector string. PhantomYerra calculates the numeric score automatically. |
cwe_id | str | Recommended | CWE identifier (e.g., "CWE-200"). |
affected_url | str | Recommended | The specific URL or endpoint where the vulnerability was found. |
tags | list[str] | Optional | Tags for categorization (e.g., ["config", "exposure", "owasp-a05"]). |
references | list[str] | Optional | Links to relevant documentation, CVE entries, or OWASP pages. |
compliance_map | dict | Optional | Compliance framework mappings (e.g., {"PCI-DSS": "6.5.8", "NIST": "AC-3"}). |
Real-Time Progress: emit_activity()
Call emit_activity() throughout your scan to keep the user informed. These events appear in the PhantomYerra UI as live progress updates.
from phantomyerra.sdk.events import emit_activity
# Phase: initializing
emit_activity(tool="my_scanner", phase="init", message="Loading payloads", percent=0)
# Phase: scanning (update frequently)
emit_activity(tool="my_scanner", phase="scanning", message="Testing endpoint /api/users", percent=35)
# Phase: complete
emit_activity(tool="my_scanner", phase="complete", message="Scan finished, 3 findings", percent=100)
emit_activity Parameters
| Parameter | Type | Description |
|---|---|---|
tool | str | Your scanner's TOOL_NAME. |
phase | str | Current phase: "init", "scanning", "analyzing", "complete", "error". |
message | str | Human-readable status message. |
percent | int | Progress percentage (0-100). |
Testing Your Scanner
-
1
Unit Test
Write unit tests for your scanner logic. Mock HTTP responses using
httpx.MockTransportorrespx. Verify that findings are created correctly with all required fields. -
2
Integration Test
Run your scanner against a known-vulnerable test target (e.g., DVWA, Juice Shop, WebGoat). Verify that it correctly identifies and reports the expected vulnerabilities.
-
3
UI Verification
Launch PhantomYerra with your extension installed. Run a scan against the test target. Verify your scanner appears in the tool list, progress events show in the UI, and findings appear in the results with correct severity, evidence, and PoC steps.
-
4
Report Verification
Generate a report from the scan. Verify your findings are included with proper formatting, evidence, and remediation guidance.
Test target: PhantomYerra includes a built-in vulnerable test application (Settings → Test Target → Launch). Use this to test your scanner without needing an external target.