+
+ {group.label}
+ {#if group.capabilities.length}
+ {on}/{group.capabilities.length} on
{/if}
- {phaseLabel(group.phase)}
-
- {group.stages.filter((s) => stageStates[s.name]).length}/{group.stages.length} on
-
-
- {#each levels as level, li (level[0].level)}
- {#if li > 0}
-
-
- {#snippet child({ props })}
-
-
-
- {/snippet}
-
-
- Waits for the stages above to finish
-
-
- {/if}
-
- {#each level as stage, si (stage.name)}
- {@const rails = railKinds(si, level, li, levels, gi, visibleGroups.length)}
- {
- openStages = { ...openStages, [stage.name]: !openStages[stage.name] };
- flashStage(stage.name);
- }}
- onChange={(field, value) => setStageField(stage.name, field, value)}
- onReset={() => resetStage(stage.name)}
- />
- {/each}
-
+
+ {#each group.capabilities as stage (stage.name)}
+ {@render stageRow(stage, false)}
{/each}
+ {#each group.active as stage (stage.name)}
+ {@render stageRow(stage, true)}
+ {/each}
+ {#if group.automatic.length}
+
+ Runs automatically: {group.automatic.map((s) => s.title).join(', ')}
+
+ {/if}
{/each}
@@ -998,67 +968,40 @@
padding: 8px 2px;
}
- .dotted {
- display: block;
- width: 3px;
- background: radial-gradient(
- circle,
- color-mix(in oklch, var(--muted-foreground) 65%, transparent) 1px,
- transparent 1.6px
- )
- center / 3px 5px repeat-y;
+ .group + .group {
+ margin-top: 14px;
}
- .phase.linked {
- margin-top: 12px;
- }
- .phase-head {
- position: relative;
+ .group-head {
display: flex;
align-items: baseline;
justify-content: space-between;
gap: 10px;
- padding: 2px 2px 8px 32px;
+ padding: 2px 2px 8px 12px;
}
- .phase-head .phase-link {
- position: absolute;
- left: 15px;
- top: -12px;
- bottom: 0;
- }
- .phase-head h2 {
+ .group-head h2 {
font-size: 11px;
font-weight: 600;
text-transform: uppercase;
letter-spacing: 0.06em;
color: var(--muted-foreground);
}
- .phase-count {
+ .group-count {
font-size: 11px;
color: var(--muted-foreground);
font-variant-numeric: tabular-nums;
}
- .phase-body {
+ .group-body {
position: relative;
border: 1px solid var(--border);
border-radius: 0.7rem;
background: var(--card);
overflow: hidden;
}
- .level {
- position: relative;
- }
- .level-gap {
- position: relative;
- height: 14px;
- border-top: 1px solid var(--border);
- border-bottom: 1px solid var(--border);
- background: color-mix(in oklch, var(--muted) 35%, transparent);
- }
- .level-gap .dotted {
- position: absolute;
- left: 15px;
- top: 0;
- bottom: 0;
+ .automatic {
+ padding: 8px 14px 9px 32px;
+ font-size: 11px;
+ color: var(--muted-foreground);
+ background: color-mix(in oklch, var(--muted) 30%, transparent);
}
.side {
diff --git a/shared/definitions/default_engine.py b/shared/definitions/default_engine.py
new file mode 100644
index 000000000..c786a82dd
--- /dev/null
+++ b/shared/definitions/default_engine.py
@@ -0,0 +1,36 @@
+"""The built-in engine every project starts with."""
+
+from __future__ import annotations
+
+from shared.enums.scan import Intensity
+
+DEFAULT_ENGINE_NAME = "Default"
+DEFAULT_ENGINE_DESCRIPTION = (
+ "Subdomain discovery on subfinder, port scan, HTTP probe, screenshots, "
+ "URL discovery and a nuclei vulnerability scan at Normal intensity. Built in."
+)
+DEFAULT_ENGINE_INTENSITY = Intensity.NORMAL.value
+SUBDOMAIN_STAGE = "subdomain_discovery"
+VULNERABILITY_STAGE = "vulnerability_scan"
+DEFAULT_PASSIVE_SOURCES: tuple[str, ...] = ("subfinder",)
+DEFAULT_VULN_SCANNERS: tuple[str, ...] = ("nuclei",)
+
+
+def default_engine_stages() -> dict[str, dict]:
+ """Only the settings that differ from the stage defaults."""
+ return {
+ SUBDOMAIN_STAGE: {"passive_tools": list(DEFAULT_PASSIVE_SOURCES)},
+ VULNERABILITY_STAGE: {"enabled": True, "scanners": list(DEFAULT_VULN_SCANNERS)},
+ }
+
+
+__all__ = [
+ "DEFAULT_ENGINE_DESCRIPTION",
+ "DEFAULT_ENGINE_INTENSITY",
+ "DEFAULT_ENGINE_NAME",
+ "DEFAULT_PASSIVE_SOURCES",
+ "DEFAULT_VULN_SCANNERS",
+ "SUBDOMAIN_STAGE",
+ "VULNERABILITY_STAGE",
+ "default_engine_stages",
+]
diff --git a/shared/definitions/intensity.py b/shared/definitions/intensity.py
new file mode 100644
index 000000000..ab97a0952
--- /dev/null
+++ b/shared/definitions/intensity.py
@@ -0,0 +1,155 @@
+"""Per-tool transport (rate, concurrency, timeout) for each intensity."""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+from enum import StrEnum
+
+from shared.definitions.constants import MAX_RATE, MAX_THREADS, MAX_TIMEOUT
+from shared.enums.scan import Intensity
+
+
+class TransportTool(StrEnum):
+ HTTPX = "httpx"
+ NAABU = "naabu"
+ NUCLEI = "nuclei"
+ KATANA = "katana"
+ FFUF = "ffuf"
+ DNSX = "dnsx"
+ BANNER = "banner"
+
+
+@dataclass(frozen=True)
+class Transport:
+ tool: str
+ rate: int | None
+ threads: int
+ timeout: int
+ retries: int
+
+ def as_dict(self) -> dict:
+ return asdict(self)
+
+
+# (rate per second or None for unlimited, concurrency)
+_NORMAL: dict[str, tuple[int | None, int]] = {
+ TransportTool.HTTPX.value: (150, 150),
+ TransportTool.NAABU.value: (1000, 100),
+ TransportTool.NUCLEI.value: (150, 25),
+ TransportTool.KATANA.value: (150, 50),
+ TransportTool.FFUF.value: (150, 40),
+ TransportTool.DNSX.value: (None, 30),
+ TransportTool.BANNER.value: (None, 32),
+}
+_AGGRESSIVE: dict[str, tuple[int | None, int]] = {
+ TransportTool.HTTPX.value: (400, 300),
+ TransportTool.NAABU.value: (3000, 200),
+ TransportTool.NUCLEI.value: (400, 50),
+ TransportTool.KATANA.value: (300, 100),
+ TransportTool.FFUF.value: (400, 80),
+ TransportTool.DNSX.value: (None, 50),
+ TransportTool.BANNER.value: (None, 64),
+}
+PROFILES: dict[str, dict[str, tuple[int | None, int]]] = {
+ Intensity.PASSIVE.value: _NORMAL,
+ Intensity.NORMAL.value: _NORMAL,
+ Intensity.AGGRESSIVE.value: _AGGRESSIVE,
+}
+
+TOOL_TIMEOUT: dict[str, int] = {
+ TransportTool.HTTPX.value: 10,
+ TransportTool.NAABU.value: 3,
+ TransportTool.NUCLEI.value: 10,
+ TransportTool.KATANA.value: 15,
+ TransportTool.FFUF.value: 8,
+ TransportTool.DNSX.value: 5,
+ TransportTool.BANNER.value: 4,
+}
+TOOL_RETRIES: dict[str, int] = {
+ TransportTool.NAABU.value: 1,
+ TransportTool.NUCLEI.value: 1,
+}
+
+RATE_TOOLS: tuple[str, ...] = tuple(
+ tool for tool, (rate, _) in _NORMAL.items() if rate is not None
+)
+
+INTENSITY_LABELS: dict[str, str] = {
+ Intensity.PASSIVE.value: "Passive",
+ Intensity.NORMAL.value: "Normal",
+ Intensity.AGGRESSIVE.value: "Aggressive",
+}
+INTENSITY_HELP: dict[str, str] = {
+ Intensity.PASSIVE.value: "Public sources only. No traffic is sent to the target.",
+ Intensity.NORMAL.value: "150 requests per second per tool, 1,000 packets per second for the port scan.",
+ Intensity.AGGRESSIVE.value: "400 requests per second per tool, 3,000 packets per second for the port scan. Concurrency doubled.",
+}
+
+
+def _clamp(value: int, lo: int, hi: int) -> int:
+ return max(lo, min(hi, value))
+
+
+def tool_rate(
+ tool: str,
+ intensity: str,
+ *,
+ rate_override: int | None = None,
+ ceiling: int | None = None,
+) -> int | None:
+ """The rate one tool runs at under this intensity, after the context's say."""
+ base = PROFILES.get(intensity, _NORMAL).get(tool, (None, 1))[0]
+ value = rate_override if rate_override is not None else base
+ if value is None:
+ return None
+ if ceiling is not None:
+ value = min(value, ceiling)
+ return _clamp(int(value), 1, MAX_RATE)
+
+
+def transport_for(
+ tool: str,
+ intensity: str,
+ *,
+ rate_weight: float = 1.0,
+ thread_weight: float = 1.0,
+ timeout: int | None = None,
+ thread_multiplier: float = 1.0,
+ timeout_multiplier: float = 1.0,
+ rate_override: int | None = None,
+ ceiling: int | None = None,
+) -> Transport:
+ profile = PROFILES.get(intensity, _NORMAL)
+ if tool not in profile:
+ msg = f"{tool!r} has no transport profile."
+ raise KeyError(msg)
+ _, threads = profile[tool]
+ rate = tool_rate(tool, intensity, rate_override=rate_override, ceiling=ceiling)
+ if rate is not None:
+ rate = _clamp(round(rate * rate_weight) or 1, 1, MAX_RATE)
+ threads = _clamp(
+ round(threads * thread_weight * thread_multiplier) or 1, 1, MAX_THREADS
+ )
+ base_timeout = TOOL_TIMEOUT[tool] if timeout is None else timeout
+ seconds = _clamp(round(base_timeout * timeout_multiplier) or 1, 1, MAX_TIMEOUT)
+ return Transport(
+ tool=tool,
+ rate=rate,
+ threads=threads,
+ timeout=seconds,
+ retries=TOOL_RETRIES.get(tool, 0),
+ )
+
+
+__all__ = [
+ "INTENSITY_HELP",
+ "INTENSITY_LABELS",
+ "PROFILES",
+ "RATE_TOOLS",
+ "TOOL_RETRIES",
+ "TOOL_TIMEOUT",
+ "Transport",
+ "TransportTool",
+ "tool_rate",
+ "transport_for",
+]
diff --git a/shared/models/scan_engine.py b/shared/models/scan_engine.py
index 863c5c439..879aa9512 100644
--- a/shared/models/scan_engine.py
+++ b/shared/models/scan_engine.py
@@ -31,6 +31,7 @@ class ScanEngine(SQLModel, table=True):
tool_options: dict = Field(
default_factory=dict, sa_column=Column(JSON, nullable=False)
)
+ builtin: bool = Field(default=False)
created_at: datetime = Field(default_factory=utc_now)
updated_at: datetime = Field(default_factory=utc_now)
last_used_at: datetime | None = Field(default=None)
@@ -77,6 +78,7 @@ class ScanEngineRead(BaseModel):
yaml_source: str | None
tool_options: dict[str, str] = PydanticField(default_factory=dict)
usage: EngineUsage = PydanticField(default_factory=EngineUsage)
+ builtin: bool = False
created_at: datetime
updated_at: datetime
last_used_at: datetime | None
@@ -92,12 +94,19 @@ class StageField(BaseModel):
option_labels: dict[str, str] | None = None
minimum: int | None = None
maximum: int | None = None
- scale: str | None = None
+ tier: str = "basic"
widget: str | None = None
kind: str | None = None
launch: bool = False
+class StageTransport(BaseModel):
+ tool: str
+ rates: dict[str, int | None]
+ threads: dict[str, int]
+ timeout: int
+
+
class StageCatalogEntry(BaseModel):
name: str
title: str
@@ -115,6 +124,7 @@ class StageCatalogEntry(BaseModel):
role: str
consumes: list[str] = PydanticField(default_factory=list)
produces: list[str] = PydanticField(default_factory=list)
+ transport: StageTransport | None = None
defaults: dict
fields: list[StageField] = PydanticField(default_factory=list)
diff --git a/shared/services/scan_resolve.py b/shared/services/scan_resolve.py
index ff329c665..1667aaf22 100644
--- a/shared/services/scan_resolve.py
+++ b/shared/services/scan_resolve.py
@@ -7,11 +7,8 @@ from typing import TYPE_CHECKING, ClassVar
from pydantic import BaseModel, Field, PrivateAttr, ValidationError
-from shared.definitions.constants import (
- MAX_RATE,
- MAX_THREADS,
- MAX_TIMEOUT,
-)
+from shared.definitions.constants import MAX_RATE
+from shared.definitions.intensity import tool_rate
from shared.definitions.tools import parse_tool_args
if TYPE_CHECKING:
@@ -246,6 +243,7 @@ class ResolvedScanConfig(BaseModel):
thread_multiplier: float = 1.0
timeout_multiplier: float = 1.0
stages: dict[str, dict] = Field(default_factory=dict)
+ transports: dict[str, dict] = Field(default_factory=dict)
excluded_subdomains: list[str] = Field(default_factory=list)
excluded_paths: list[str] = Field(default_factory=list)
excluded_ips: list[str] = Field(default_factory=list)
@@ -326,22 +324,6 @@ def _build_headers(engine, ctx) -> tuple[dict[str, str], list[str]]:
return headers, auth_header_names
-def _resolve_rate(
- base: int, tool: str | None, overrides: dict, ceiling: int | None
-) -> int:
- value = overrides.get(tool, base) if tool else base
- if ceiling is not None:
- value = min(value, ceiling)
- return _clamp(int(value), 1, MAX_RATE)
-
-
-def _assert_flags_preserved(stage: str, original: dict, scaled: dict) -> None:
- for key, value in original.items():
- if isinstance(value, bool) and scaled[key] != value:
- msg = f"Resolved enable flag {stage}.{key} diverged from engine."
- raise RuntimeError(msg)
-
-
def validate_overrides(overrides: dict | None) -> dict[str, dict]:
"""Keep only stage keys that exist and values their own config model accepts."""
from stages.registry import stage_by_name # noqa: PLC0415
@@ -394,7 +376,7 @@ def merge_engine_context(
intensity: str | None = None,
) -> ResolvedScanConfig:
from shared.enums.scan import INTENSITIES, Intensity # noqa: PLC0415
- from stages.config import Scale # noqa: PLC0415
+ from stages.registry import rate_tools # noqa: PLC0415
from stages.registry import stages as stage_specs # noqa: PLC0415
run_intensity = intensity or engine.intensity
@@ -415,7 +397,18 @@ def merge_engine_context(
stored = engine.stages or {}
run_overrides = validate_overrides(overrides)
stages: dict[str, dict] = {}
+ transports: dict[str, dict] = {}
per_tool_rate_limits: dict[str, int] = {}
+ for tool in rate_tools():
+ override = rate_overrides.get(tool)
+ limit = tool_rate(
+ tool,
+ run_intensity,
+ rate_override=_clamp(int(override), 1, MAX_RATE) if override else None,
+ ceiling=global_rate_limit_ceiling,
+ )
+ if limit is not None:
+ per_tool_rate_limits[tool] = limit
for spec in stage_specs():
authored = (
@@ -430,18 +423,15 @@ def merge_engine_context(
authored["enabled"] = True
config = spec.config_model(**authored)
values = config.model_dump()
- for name, (scale, tool) in spec.config_model.scaled_fields().items():
- base = values[name]
- if scale is Scale.THREADS:
- values[name] = _clamp(int(base * thread_mult) or 1, 1, MAX_THREADS)
- elif scale is Scale.TIMEOUT:
- values[name] = _clamp(int(base * timeout_mult) or 1, 1, MAX_TIMEOUT)
- elif scale is Scale.RATE:
- values[name] = _resolve_rate(
- base, tool, rate_overrides, global_rate_limit_ceiling
- )
- per_tool_rate_limits[tool] = values[name]
- _assert_flags_preserved(spec.name, config.model_dump(), values)
+ if spec.transport_tool is not None:
+ override = rate_overrides.get(spec.transport_tool)
+ transports[spec.name] = spec.transport(
+ run_intensity,
+ thread_multiplier=thread_mult,
+ timeout_multiplier=timeout_mult,
+ rate_override=_clamp(int(override), 1, MAX_RATE) if override else None,
+ ceiling=global_rate_limit_ceiling,
+ ).as_dict()
if passive and spec.touches_target:
values["enabled"] = False
stages[spec.name] = values
@@ -463,6 +453,7 @@ def merge_engine_context(
thread_multiplier=thread_mult,
timeout_multiplier=timeout_mult,
stages=stages,
+ transports=transports,
excluded_subdomains=excluded_subdomains,
excluded_paths=excluded_paths,
excluded_ips=excluded_ips,
diff --git a/stages/base.py b/stages/base.py
index bcab81f6b..6ad726e54 100644
--- a/stages/base.py
+++ b/stages/base.py
@@ -9,6 +9,7 @@ from typing import TYPE_CHECKING, ClassVar
from sqlalchemy import update
+from shared.definitions.intensity import Transport, transport_for
from shared.definitions.surface import SURFACE_COUNT_COLUMNS, SurfaceDimension
from shared.enums.scan import Phase
from shared.enums.target import TargetType
@@ -105,6 +106,10 @@ class Stage(ABC):
produces: ClassVar[frozenset[str]] = frozenset()
group: ClassVar[str] = ""
role: ClassVar[str] = ""
+ transport_tool: ClassVar[str | None] = None
+ rate_weight: ClassVar[float] = 1.0
+ thread_weight: ClassVar[float] = 1.0
+ transport_timeout: ClassVar[int | None] = None
config_model: ClassVar[type[StageConfig]] = StageConfig
def __init__(self, session: Session, context: StageContext) -> None:
@@ -118,6 +123,28 @@ class Stage(ABC):
def should_run(self) -> bool:
return self.cfg.enabled
+ @cached_property
+ def transport(self) -> Transport:
+ """Rate, concurrency and timeout for this stage's tool under the run's intensity."""
+ stored = (self.ctx.resolved.transports or {}).get(self.name)
+ if stored:
+ return Transport(**stored)
+ return self.default_transport(self.ctx.resolved.intensity)
+
+ @classmethod
+ def default_transport(cls, intensity: str, **scaling) -> Transport:
+ if cls.transport_tool is None:
+ msg = f"{cls.name} declares no transport tool."
+ raise RuntimeError(msg)
+ return transport_for(
+ cls.transport_tool,
+ intensity,
+ rate_weight=cls.rate_weight,
+ thread_weight=cls.thread_weight,
+ timeout=cls.transport_timeout,
+ **scaling,
+ )
+
@abstractmethod
def run(self) -> StageResult: ...
diff --git a/stages/config.py b/stages/config.py
index 8ca257a04..f7579fdeb 100644
--- a/stages/config.py
+++ b/stages/config.py
@@ -5,36 +5,24 @@ from typing import Any
from pydantic import BaseModel, ConfigDict, Field
-from shared.definitions.constants import MAX_RATE, MAX_THREADS, MAX_TIMEOUT
from shared.definitions.wordlists import DEFAULT_WORDLIST, MAX_SLUG_LENGTH
-class Scale(StrEnum):
- THREADS = "threads"
- TIMEOUT = "timeout"
- RATE = "rate"
+class FieldTier(StrEnum):
+ BASIC = "basic"
+ ADVANCED = "advanced"
-def threads(default: int, *, title: str, description: str = "") -> Any:
- return Field(
- default,
- ge=1,
- le=MAX_THREADS,
- title=title,
- description=description,
- json_schema_extra={"scale": Scale.THREADS.value},
- )
+ADVANCED: dict[str, str] = {"tier": FieldTier.ADVANCED.value}
-def timeout(default: int, *, title: str, description: str = "") -> Any:
- return Field(
- default,
- ge=1,
- le=MAX_TIMEOUT,
- title=title,
- description=description,
- json_schema_extra={"scale": Scale.TIMEOUT.value},
- )
+def advanced(default: Any = None, **kwargs: Any) -> Any:
+ """A field behind the stage's Advanced disclosure."""
+ extra = dict(kwargs.pop("json_schema_extra", None) or {})
+ extra.update(ADVANCED)
+ if "default_factory" in kwargs:
+ return Field(json_schema_extra=extra, **kwargs)
+ return Field(default, json_schema_extra=extra, **kwargs)
def wordlist(kind: str, *, title: str, description: str = "") -> Any:
@@ -48,17 +36,6 @@ def wordlist(kind: str, *, title: str, description: str = "") -> Any:
)
-def rate(default: int, *, tool: str, title: str, description: str = "") -> Any:
- return Field(
- default,
- ge=1,
- le=MAX_RATE,
- title=title,
- description=description,
- json_schema_extra={"scale": Scale.RATE.value, "tool": tool},
- )
-
-
def share_rate(rate: int, processes: int) -> int:
"""One process's share of the stage's rate budget."""
return max(1, rate // max(1, processes))
@@ -72,19 +49,10 @@ class StageConfig(BaseModel):
)
@classmethod
- def scaled_fields(cls) -> dict[str, tuple[Scale, str | None]]:
- out: dict[str, tuple[Scale, str | None]] = {}
+ def tiers(cls) -> dict[str, str]:
+ out: dict[str, str] = {}
for name, field in cls.model_fields.items():
extra = field.json_schema_extra
- if not isinstance(extra, dict) or "scale" not in extra:
- continue
- out[name] = (Scale(extra["scale"]), extra.get("tool"))
+ tier = extra.get("tier") if isinstance(extra, dict) else None
+ out[name] = tier or FieldTier.BASIC.value
return out
-
- @classmethod
- def rate_tools(cls) -> set[str]:
- return {
- tool
- for scale, tool in cls.scaled_fields().values()
- if scale is Scale.RATE and tool
- }
diff --git a/stages/content_discovery/config.py b/stages/content_discovery/config.py
index 4b0ea95d7..f6a2aea7f 100644
--- a/stages/content_discovery/config.py
+++ b/stages/content_discovery/config.py
@@ -3,7 +3,7 @@ from __future__ import annotations
from pydantic import Field
from shared.definitions.wordlists import WordlistKind
-from stages.config import StageConfig, rate, threads, timeout, wordlist
+from stages.config import StageConfig, advanced, wordlist
class ContentDiscoveryConfig(StageConfig):
@@ -24,13 +24,6 @@ class ContentDiscoveryConfig(StageConfig):
title="Words to try",
description="Words tried per site, from the top of the list.",
)
- max_hosts: int = Field(
- default=25,
- ge=1,
- le=1_000,
- title="Sites to guess against",
- description="Sites tried. Sites that answered are picked first.",
- )
max_minutes: int = Field(
default=20,
ge=1,
@@ -38,6 +31,10 @@ class ContentDiscoveryConfig(StageConfig):
title="Time budget (minutes)",
description="The run stops after this many minutes and reports partial coverage.",
)
- threads: int = threads(40, title="Threads")
- rate: int = rate(50, tool="ffuf", title="Requests/s")
- timeout: int = timeout(8, title="Request timeout (s)")
+ max_hosts: int = advanced(
+ 25,
+ ge=1,
+ le=1_000,
+ title="Sites to guess against",
+ description="Sites tried. Sites that answered are picked first.",
+ )
diff --git a/stages/content_discovery/stage.py b/stages/content_discovery/stage.py
index bf0cc74f9..308bf643a 100644
--- a/stages/content_discovery/stage.py
+++ b/stages/content_discovery/stage.py
@@ -10,6 +10,7 @@ from pathlib import Path
from sqlalchemy import select
from shared.definitions.endpoints import EndpointSource
+from shared.definitions.intensity import TransportTool
from shared.definitions.surface import SurfaceDimension
from shared.enums.scan import AssetKind, Phase, StageGroup, StageRole
from shared.logging import get_logger
@@ -17,6 +18,7 @@ from shared.models.http_asset import HttpAsset
from shared.services import endpoint_inventory
from shared.services.endpoint_inventory import EndpointObservation
from shared.services.endpoint_noise import NoisePolicy
+from shared.services.scope_filter import matches_any
from shared.services.wordlists import WordlistError, read_words
from stages.base import ALL_TARGETS, Stage, StageResult
from stages.config import share_rate
@@ -49,6 +51,8 @@ class ContentDiscoveryStage(Stage):
produces = frozenset({AssetKind.ENDPOINTS.value})
applies_to = ALL_TARGETS
tools = ("ffuf",)
+ transport_tool = TransportTool.FFUF.value
+ rate_weight = 1 / 3
touches_target = True
config_model = ContentDiscoveryConfig
launch_fields = (
@@ -86,9 +90,9 @@ class ContentDiscoveryStage(Stage):
net = self.net_options()
args = {
"wordlist": word_file,
- "threads": cfg.threads,
- "rate": share_rate(cfg.rate, workers),
- "request_timeout": cfg.timeout,
+ "threads": self.transport.threads,
+ "rate": share_rate(self.transport.rate or 1, workers),
+ "request_timeout": self.transport.timeout,
"proxy_url": net.proxy_url,
"headers": net.headers,
"recorder": self.ctx.recorder,
@@ -202,6 +206,9 @@ class ContentDiscoveryStage(Stage):
except WordlistError as exc:
yield None, str(exc)
return
+ excluded = self.ctx.resolved.excluded_paths or []
+ if excluded:
+ words = [w for w in words if not matches_any(f"/{w.lstrip('/')}", excluded)]
if not words:
yield None, f"{label} has no usable words"
return
diff --git a/stages/dast_scan/config.py b/stages/dast_scan/config.py
index ba187dbee..3a618695b 100644
--- a/stages/dast_scan/config.py
+++ b/stages/dast_scan/config.py
@@ -2,20 +2,14 @@ from __future__ import annotations
from pydantic import Field, field_validator
-from shared.definitions.scan_surface import (
- BASE_MAX_DEPTH,
- MAX_BASES_PER_ORIGIN,
- MAX_REQUESTS,
- MAX_REQUESTS_PER_ORIGIN,
-)
+from shared.definitions.scan_surface import MAX_REQUESTS, MAX_REQUESTS_PER_ORIGIN
from shared.definitions.vulnerabilities import (
DEFAULT_SEVERITIES,
- MAX_FINDINGS_PER_SCAN,
SEVERITY_LABELS,
SEVERITY_ORDER,
reject_unknown,
)
-from stages.config import StageConfig, rate, threads, timeout
+from stages.config import StageConfig, advanced
class DastScanConfig(StageConfig):
@@ -43,58 +37,15 @@ class DastScanConfig(StageConfig):
"widget": "severity",
},
)
- max_requests_per_origin: int = Field(
- default=MAX_REQUESTS_PER_ORIGIN,
- ge=1,
- le=1000,
- title="Requests per origin",
- description="Parameter sets fuzzed per origin at most.",
- )
- max_requests: int = Field(
- default=MAX_REQUESTS,
- ge=1,
- le=20000,
- title="Request budget",
- description="Parameter sets fuzzed in one scan at most.",
- )
directories: bool = Field(
default=False,
title="Directory exposure checks",
description="Run the exposure checks under each discovered directory.",
)
- max_dirs_per_origin: int = Field(
- default=MAX_BASES_PER_ORIGIN,
- ge=1,
- le=200,
- title="Directories per origin",
- )
- dir_depth: int = Field(
- default=BASE_MAX_DEPTH,
- ge=1,
- le=3,
- title="Directory depth",
- description="Deepest directory level checked.",
- )
- fuzz_param_frequency: int = Field(
- default=10,
- ge=1,
- le=100,
- title="Parameter patience",
- description="Sightings of one parameter name without a match before it is skipped on that host.",
- )
- rate: int = rate(50, tool="nuclei", title="Request rate (rps)")
- threads: int = threads(
- 25, title="Concurrency", description="Checks running in parallel."
- )
- bulk_size: int = Field(default=25, ge=1, le=500, title="Requests per check")
- timeout: int = timeout(10, title="Request timeout (s)")
- retries: int = Field(default=1, ge=0, le=5, title="Retries")
- max_host_error: int = Field(
- default=30,
- ge=1,
- le=500,
- title="Host error budget",
- description="Errors a host may return before it is dropped.",
+ headless: bool = Field(
+ default=False,
+ title="Browser fuzzing",
+ description="Run the fuzzing checks that need a rendered page. Most DAST checks do.",
)
max_minutes: int = Field(
default=30,
@@ -103,43 +54,24 @@ class DastScanConfig(StageConfig):
title="Time budget (min)",
description="Stop starting batches after this many minutes. 0 means no limit.",
)
- waf_rate_divisor: int = Field(
- default=3,
+ max_requests_per_origin: int = advanced(
+ MAX_REQUESTS_PER_ORIGIN,
ge=1,
- le=20,
- title="WAF rate divisor",
- description="Hosts behind a WAF or CDN are fuzzed at the request rate divided by this. 1 disables.",
+ le=1000,
+ title="Requests per origin",
+ description="Parameter sets fuzzed per origin at most.",
)
- honeypot_threshold: int = Field(default=0, ge=0, le=200, title="Honeypot threshold")
- headless: bool = Field(
- default=False,
- title="Browser fuzzing",
- description="Run the fuzzing checks that need a rendered page. Most DAST checks do.",
+ max_requests: int = advanced(
+ MAX_REQUESTS,
+ ge=1,
+ le=20000,
+ title="Request budget",
+ description="Parameter sets fuzzed in one scan at most.",
)
- interactsh: bool = Field(
- default=False,
+ interactsh: bool = advanced(
+ False,
title="Out-of-band testing",
- description="Detect blind injection through callbacks to an external server.",
- )
- interactsh_server: str = Field(
- default="", max_length=200, title="Self-hosted OAST server"
- )
- oast_wait_minutes: int = Field(
- default=5, ge=0, le=120, title="Keep listening (minutes)"
- )
- store_evidence: bool = Field(
- default=True,
- title="Store request and response",
- description="Keep the request and response that produced each finding.",
- )
- max_evidence_recovery: int = Field(
- default=0, ge=0, le=2000, title="Evidence recovery budget"
- )
- max_findings: int = Field(
- default=MAX_FINDINGS_PER_SCAN,
- ge=1,
- le=MAX_FINDINGS_PER_SCAN,
- title="Finding budget",
+ description="Detect blind injection through callbacks. The server is set on the vulnerability scan.",
)
@field_validator("severities")
diff --git a/stages/dast_scan/scanners/dalfox.py b/stages/dast_scan/scanners/dalfox.py
index b15d9396a..af9560e51 100644
--- a/stages/dast_scan/scanners/dalfox.py
+++ b/stages/dast_scan/scanners/dalfox.py
@@ -62,8 +62,8 @@ class DalfoxScanner(VulnScanner):
severities=list(ctx.cfg.severities),
hosts_total=len(plan.requests),
hosts_covered=len(reflecting),
- rate_limit=ctx.cfg.rate,
- concurrency=ctx.cfg.threads,
+ rate_limit=ctx.transport.rate,
+ concurrency=ctx.transport.threads,
)
if not reflecting:
coverage.status = CoverageStatus.COMPLETED.value
@@ -93,8 +93,8 @@ class DalfoxScanner(VulnScanner):
return []
try:
client = HttpxClient(
- rate_limit=ctx.cfg.rate,
- threads=ctx.cfg.threads,
+ rate_limit=ctx.transport.rate,
+ threads=ctx.transport.threads,
timeout=_PROBE_TIMEOUT,
proxy_url=ctx.net.proxy_url,
headers=ctx.net.headers,
@@ -121,10 +121,10 @@ class DalfoxScanner(VulnScanner):
ctx = self.ctx
secrets_list = [v for v in (ctx.net.headers or {}).values() if v]
options = DalfoxOptions(
- workers=ctx.cfg.threads,
- rate=ctx.cfg.rate,
- timeout=ctx.cfg.timeout,
- retries=ctx.cfg.retries,
+ workers=ctx.transport.threads,
+ rate=ctx.transport.rate or 1,
+ timeout=ctx.transport.timeout,
+ retries=ctx.transport.retries,
proxy_url=ctx.net.proxy_url,
headers=dict(ctx.net.headers or {}),
follow_redirects=bool(ctx.resolved.follow_redirects),
@@ -142,7 +142,7 @@ class DalfoxScanner(VulnScanner):
return coverage
def _on_finding(finding) -> None:
- if ctx.cfg.store_evidence:
+ if ctx.keep_evidence():
finding.request = redact_secrets(finding.request, secrets_list)
finding.response = redact_secrets(finding.response, secrets_list)
else:
diff --git a/stages/dast_scan/scanners/nuclei.py b/stages/dast_scan/scanners/nuclei.py
index 372da143e..6c04ad9bb 100644
--- a/stages/dast_scan/scanners/nuclei.py
+++ b/stages/dast_scan/scanners/nuclei.py
@@ -26,6 +26,7 @@ from shared.services.scan_surface import (
from shared.services.vuln_templates import selection_predicate
from stages.vulnerability_scan.scanners.base import ScannerResult
from stages.vulnerability_scan.scanners.nuclei import (
+ WAF_RATE_DIVISOR,
Job,
NucleiScanner,
_by_lane,
@@ -36,6 +37,7 @@ from tools.nuclei.parser import Finding
_EXPOSURE_SET = "exposure"
_REQUESTS_PER_INVOCATION = 100
+FUZZ_PARAM_FREQUENCY = 10
_BASE_URL = "{{BaseURL}}"
@@ -72,6 +74,10 @@ def exposure_templates(session, cfg) -> list[VulnTemplate]:
class NucleiDastScanner(NucleiScanner):
"""The same runner, handed request and directory items instead of roots."""
+ honeypot_threshold = 0
+ evidence_recovery = 0
+ fuzz_param_frequency = FUZZ_PARAM_FREQUENCY
+
def run(self) -> ScannerResult:
ctx = self.ctx
ok, reason = self.availability()
@@ -106,7 +112,6 @@ class NucleiDastScanner(NucleiScanner):
def _schedule_requests(
self, plan: SurfacePlan, fuzz, exposure
) -> dict[str, list[Job]]:
- cfg = self.ctx.cfg
rates = self._rates(plan)
lanes: dict[str, list[Job]] = {}
counter: dict[str, int] = {}
@@ -118,7 +123,7 @@ class NucleiDastScanner(NucleiScanner):
if fuzz.paths and plan.requests:
files = fuzz.files(fuzz.rows)
- for lane, wanted in _by_lane(plan.requests, cfg.waf_rate_divisor).items():
+ for lane, wanted in _by_lane(plan.requests, WAF_RATE_DIVISOR).items():
grouped: list[SurfaceItem] = [i for g in by_origin(wanted) for i in g]
for items in chunk(grouped, _REQUESTS_PER_INVOCATION):
_push(
@@ -136,7 +141,7 @@ class NucleiDastScanner(NucleiScanner):
if exposure.paths and plan.bases:
files = exposure.files(exposure.rows)
per_host = cost(exposure.rows)
- for lane, wanted in _by_lane(plan.bases, cfg.waf_rate_divisor).items():
+ for lane, wanted in _by_lane(plan.bases, WAF_RATE_DIVISOR).items():
for items in batches(wanted, per_host, rates[lane]):
_push(
Job(
diff --git a/stages/dast_scan/stage.py b/stages/dast_scan/stage.py
index f436d0fea..2bd0903fa 100644
--- a/stages/dast_scan/stage.py
+++ b/stages/dast_scan/stage.py
@@ -2,9 +2,15 @@ from __future__ import annotations
import uuid
-from shared.definitions.scan_surface import SurfaceClass
+from shared.definitions.intensity import TransportTool
+from shared.definitions.scan_surface import (
+ BASE_MAX_DEPTH,
+ MAX_BASES_PER_ORIGIN,
+ SurfaceClass,
+)
from shared.definitions.surface import SurfaceDimension
from shared.definitions.vulnerabilities import (
+ MAX_FINDINGS_PER_SCAN,
SEVERITY_LABELS,
CoverageStatus,
Severity,
@@ -55,6 +61,8 @@ class DastScanStage(Stage):
produces = frozenset({AssetKind.VULNERABILITIES.value})
applies_to = ALL_TARGETS
tools = ("nuclei",)
+ transport_tool = TransportTool.NUCLEI.value
+ rate_weight = 1 / 3
touches_target = True
config_model = DastScanConfig
launch_fields = ("enabled", "severities", "directories", "headless", "max_minutes")
@@ -76,8 +84,8 @@ class DastScanStage(Stage):
max_per_origin=cfg.max_requests_per_origin,
max_total=cfg.max_requests,
bases=cfg.directories,
- max_bases_per_origin=cfg.max_dirs_per_origin,
- base_depth=cfg.dir_depth,
+ max_bases_per_origin=MAX_BASES_PER_ORIGIN,
+ base_depth=BASE_MAX_DEPTH,
)
scan_surface.write(
self.session,
@@ -99,7 +107,7 @@ class DastScanStage(Stage):
def _store(findings: list) -> int:
self._check_abort()
- room = cfg.max_findings - len(stored)
+ room = MAX_FINDINGS_PER_SCAN - len(stored)
if room <= 0:
return 0
written = vuln_inventory.upsert(
@@ -137,6 +145,7 @@ class DastScanStage(Stage):
target_id=self.ctx.target_id,
project_id=self.ctx.project_id,
cfg=cfg,
+ transport=self.transport,
resolved=self.ctx.resolved,
net=self.net_options(),
surface=plan,
diff --git a/stages/endpoint_probe/config.py b/stages/endpoint_probe/config.py
index 32588f5d8..faee3dbd0 100644
--- a/stages/endpoint_probe/config.py
+++ b/stages/endpoint_probe/config.py
@@ -3,7 +3,7 @@ from __future__ import annotations
from pydantic import Field
from shared.definitions.endpoints import DEFAULT_PROBE_CAP
-from stages.config import StageConfig, rate, threads, timeout
+from stages.config import StageConfig, advanced
class EndpointProbeConfig(StageConfig):
@@ -12,23 +12,18 @@ class EndpointProbeConfig(StageConfig):
title="Verify endpoints",
description="Request the discovered URLs and record each status.",
)
- threads: int = threads(40, title="Threads")
- timeout: int = timeout(10, title="Timeout (s)")
- rate: int = rate(150, tool="httpx", title="Requests/s")
- max_urls: int = Field(
- default=DEFAULT_PROBE_CAP,
+ skip_static: bool = advanced(
+ True,
+ title="Skip images and media",
+ description="Skip images, stylesheets, fonts and other static files.",
+ )
+ max_urls: int = advanced(
+ DEFAULT_PROBE_CAP,
ge=0,
le=100000,
title="URLs to verify",
description="Endpoints requested. The rest stay unverified.",
)
- skip_static: bool = Field(
- default=True,
- title="Skip images and media",
- description="Skip images, stylesheets, fonts and other static files.",
- )
- follow_redirects: bool = Field(
- default=False,
- title="Follow redirects",
- description="Follow 3xx redirects.",
- )
+
+
+FOLLOW_REDIRECTS = False
diff --git a/stages/endpoint_probe/stage.py b/stages/endpoint_probe/stage.py
index 282cd5d01..091773474 100644
--- a/stages/endpoint_probe/stage.py
+++ b/stages/endpoint_probe/stage.py
@@ -7,6 +7,7 @@ from sqlalchemy import cast, func, select
from sqlalchemy.dialects.postgresql import JSONB
from shared.definitions.endpoints import PROBE_COVERAGE_SOURCE, STATIC_CLASSES
+from shared.definitions.intensity import TransportTool
from shared.definitions.surface import SurfaceDimension
from shared.definitions.vulnerabilities import CoverageStatus
from shared.enums.scan import AssetKind, Phase, StageGroup, StageRole
@@ -19,7 +20,7 @@ from shared.services.endpoint_judge import Fingerprint
from shared.services.scope_filter import matches_any
from shared.utils.datetime import utc_now
from stages.base import ALL_TARGETS, Stage, StageResult
-from stages.endpoint_probe.config import EndpointProbeConfig
+from stages.endpoint_probe.config import FOLLOW_REDIRECTS, EndpointProbeConfig
from tools.httpx.client import HttpxClient, HttpxError
from tools.httpx.parser import parse_httpx_record
@@ -42,6 +43,8 @@ class EndpointProbeStage(Stage):
consumes = frozenset({AssetKind.ENDPOINTS.value})
applies_to = ALL_TARGETS
tools = ("httpx",)
+ transport_tool = TransportTool.HTTPX.value
+ thread_weight = 4 / 15
config_model = EndpointProbeConfig
launch_fields = ("enabled", "max_urls")
@@ -63,13 +66,13 @@ class EndpointProbeStage(Stage):
try:
client = HttpxClient(
- rate_limit=cfg.rate,
- threads=cfg.threads,
- timeout=cfg.timeout,
+ rate_limit=self.transport.rate,
+ threads=self.transport.threads,
+ timeout=self.transport.timeout,
proxy_url=net.proxy_url,
headers=net.headers,
probe_scheme=net.probe_scheme,
- follow_redirects=self.follow_redirects(cfg.follow_redirects),
+ follow_redirects=self.follow_redirects(FOLLOW_REDIRECTS),
recorder=self.ctx.recorder,
extra_args=self.ctx.resolved.tool_args("httpx"),
)
diff --git a/stages/host_discovery/config.py b/stages/host_discovery/config.py
index 4c343e8b8..31eb8d597 100644
--- a/stages/host_discovery/config.py
+++ b/stages/host_discovery/config.py
@@ -2,7 +2,7 @@ from __future__ import annotations
from pydantic import Field
-from stages.config import StageConfig, rate, threads, timeout
+from stages.config import StageConfig
class HostDiscoveryConfig(StageConfig):
@@ -11,6 +11,3 @@ class HostDiscoveryConfig(StageConfig):
title="Find live hosts",
description="Sweep a netblock for responsive hosts before port scanning.",
)
- rate: int = rate(150, tool="naabu", title="Packet rate (pps)")
- threads: int = threads(30, title="Concurrency")
- timeout: int = timeout(5, title="Timeout (s)")
diff --git a/stages/host_discovery/stage.py b/stages/host_discovery/stage.py
index 1fead67d6..dc4031797 100644
--- a/stages/host_discovery/stage.py
+++ b/stages/host_discovery/stage.py
@@ -2,9 +2,11 @@ from __future__ import annotations
from sqlalchemy import select
+from shared.definitions.intensity import TransportTool
from shared.enums.scan import AssetKind, Phase, StageGroup, StageRole
from shared.logging import get_logger
from shared.models.ip_address import IpAddress
+from shared.services.scope_filter import ip_excluded
from stages.base import RANGE_TARGETS, Stage, StageResult
from stages.host_discovery.config import HostDiscoveryConfig
from tools.naabu.client import NaabuClient, NaabuError, NaabuOptions
@@ -25,11 +27,14 @@ class HostDiscoveryStage(Stage):
produces = frozenset({AssetKind.ADDRESSES.value})
applies_to = RANGE_TARGETS
tools = ("naabu",)
+ transport_tool = TransportTool.NAABU.value
+ rate_weight = 0.15
+ thread_weight = 0.3
+ transport_timeout = 5
config_model = HostDiscoveryConfig
def run(self) -> StageResult:
self._check_abort()
- cfg = self.cfg
net = self.net_options()
rows = list(
self.session.execute(
@@ -38,15 +43,19 @@ class HostDiscoveryStage(Stage):
.scalars()
.all()
)
+ excluded = self.ctx.resolved.excluded_ips or []
+ if excluded:
+ rows = [row for row in rows if not ip_excluded(row.ip, excluded)]
if not rows:
return StageResult(counts={"alive": 0})
try:
client = NaabuClient(
options=NaabuOptions(
- rate=cfg.rate,
- concurrency=cfg.threads,
- timeout=cfg.timeout,
+ rate=self.transport.rate or 1,
+ concurrency=self.transport.threads,
+ timeout=self.transport.timeout,
+ retries=self.transport.retries,
proxy_url=net.proxy_url,
extra_args=self.ctx.resolved.tool_args("naabu"),
),
diff --git a/stages/http_probe/config.py b/stages/http_probe/config.py
index 57d162f3b..bbcf4c980 100644
--- a/stages/http_probe/config.py
+++ b/stages/http_probe/config.py
@@ -2,7 +2,7 @@ from __future__ import annotations
from pydantic import Field
-from stages.config import StageConfig, rate, threads, timeout
+from stages.config import StageConfig, advanced
class HttpProbeConfig(StageConfig):
@@ -11,28 +11,12 @@ class HttpProbeConfig(StageConfig):
title="Probe HTTP services",
description="Fingerprint every host and port for live HTTP, technologies and titles.",
)
- threads: int = threads(150, title="Threads")
- timeout: int = timeout(10, title="Timeout (s)")
- rate: int = rate(150, tool="httpx", title="Requests/s")
- follow_redirects: bool = Field(
- default=True,
- title="Follow redirects",
- description="Follow 3xx to the final URL.",
- )
- probe_open_ports: bool = Field(
- default=True,
- title="Probe open ports",
- description="Probe the ports the port scan found, not only 80 and 443.",
- )
- probe_all_ports: bool = Field(
- default=False,
+ probe_all_ports: bool = advanced(
+ False,
title="Include non-web ports",
description="Also probe ports classed as non-HTTP services.",
)
- max_ports_per_host: int = Field(
- default=25,
- ge=1,
- le=1000,
- title="Ports per host",
- description="Cap the ports probed on any one host.",
- )
+
+
+FOLLOW_REDIRECTS = True
+MAX_PORTS_PER_HOST = 25
diff --git a/stages/http_probe/stage.py b/stages/http_probe/stage.py
index ca4ee49b2..356f1caa7 100644
--- a/stages/http_probe/stage.py
+++ b/stages/http_probe/stage.py
@@ -5,6 +5,7 @@ from collections.abc import Iterable
from sqlalchemy import bindparam, delete, select, update
from sqlalchemy.orm import defer
+from shared.definitions.intensity import TransportTool
from shared.definitions.ports import (
DEFAULT_WEB_PORTS,
PortSource,
@@ -25,7 +26,11 @@ from shared.utils.datetime import utc_now
from shared.utils.net import host_port
from shared.utils.software import parse_banner
from stages.base import Stage, StageResult
-from stages.http_probe.config import HttpProbeConfig
+from stages.http_probe.config import (
+ FOLLOW_REDIRECTS,
+ MAX_PORTS_PER_HOST,
+ HttpProbeConfig,
+)
from tools.httpx.client import HttpxClient, HttpxError
from tools.httpx.parser import parse_httpx_record
@@ -93,11 +98,11 @@ class HttpProbeStage(Stage):
consumes = frozenset({AssetKind.HOSTS.value, AssetKind.ADDRESSES.value})
produces = frozenset({AssetKind.HTTP_ASSETS.value})
tools = ("httpx",)
+ transport_tool = TransportTool.HTTPX.value
config_model = HttpProbeConfig
def run(self) -> StageResult:
self._check_abort()
- cfg = self.cfg
net = self.net_options()
targets = self._build_targets()
if not targets:
@@ -105,13 +110,13 @@ class HttpProbeStage(Stage):
try:
client = HttpxClient(
- rate_limit=cfg.rate,
- threads=cfg.threads,
- timeout=cfg.timeout,
+ rate_limit=self.transport.rate,
+ threads=self.transport.threads,
+ timeout=self.transport.timeout,
proxy_url=net.proxy_url,
headers=net.headers,
probe_scheme=net.probe_scheme,
- follow_redirects=self.follow_redirects(cfg.follow_redirects),
+ follow_redirects=self.follow_redirects(FOLLOW_REDIRECTS),
recorder=self.ctx.recorder,
extra_args=self.ctx.resolved.tool_args("httpx"),
)
@@ -146,8 +151,6 @@ class HttpProbeStage(Stage):
def _port_map(self) -> dict[str, set[int]]:
"""Open ports per address, filtered to what can plausibly answer HTTP."""
- if not self.cfg.probe_open_ports:
- return {}
rows = self.session.execute(
select(Port.ip, Port.number, Port.service_name, Port.service_class).where(
Port.scan_id == self.ctx.scan_id
@@ -167,7 +170,7 @@ class HttpProbeStage(Stage):
for ip in ips or []:
ports |= port_map.get(ip, set())
ordered = sorted(ports, key=lambda p: (p not in DEFAULT_WEB_PORTS, p))
- return ordered[: self.cfg.max_ports_per_host]
+ return ordered[:MAX_PORTS_PER_HOST]
def _build_targets(self) -> list[str]:
target_type = self.ctx.target_type
diff --git a/stages/netblock_sweep/config.py b/stages/netblock_sweep/config.py
index 6222fbbcc..a5f2a7f83 100644
--- a/stages/netblock_sweep/config.py
+++ b/stages/netblock_sweep/config.py
@@ -2,7 +2,7 @@ from __future__ import annotations
from pydantic import Field
-from stages.config import StageConfig, threads, timeout
+from stages.config import StageConfig, advanced
class NetblockSweepConfig(StageConfig):
@@ -11,33 +11,15 @@ class NetblockSweepConfig(StageConfig):
title="Netblock sweep",
description="Resolve reverse DNS across the network ranges the organisation announces.",
)
- min_addresses: int = Field(
- default=2,
- ge=1,
- le=100,
- title="Addresses before a network counts",
- description="Minimum number of the scan's addresses a network must hold to count as owned.",
- )
- max_asn_addresses: int = Field(
- default=65536,
- ge=256,
- le=1048576,
- title="Largest network an organisation may own",
- description="Networks announcing more addresses than this are excluded.",
- )
- min_share: int = Field(
- default=10,
- ge=1,
- le=100,
- title="Share of the estate (%)",
- description="Minimum share of the scan's addresses a network must hold.",
- )
- max_sweep: int = Field(
- default=8192,
+ max_sweep: int = advanced(
+ 8192,
ge=256,
le=65536,
title="Address budget",
description="Addresses resolved at most. A larger range is reported as partial.",
)
- dns_threads: int = threads(50, title="DNS threads")
- dns_timeout: int = timeout(5, title="DNS timeout (s)")
+
+
+MIN_ADDRESSES = 2
+MIN_SHARE = 10
+MAX_ASN_ADDRESSES = 65536
diff --git a/stages/netblock_sweep/stage.py b/stages/netblock_sweep/stage.py
index 0bba3f054..be0baebcf 100644
--- a/stages/netblock_sweep/stage.py
+++ b/stages/netblock_sweep/stage.py
@@ -5,16 +5,22 @@ from dataclasses import dataclass, field
from sqlalchemy import select, text
+from shared.definitions.intensity import TransportTool
from shared.enums.scan import AssetKind, Phase, StageGroup, StageRole
from shared.enums.subdomain import SubdomainSource
from shared.logging import get_logger
from shared.models.subdomain import Subdomain
from shared.services.ip_asn import ranges_ready
from shared.services.ip_inventory import collect_ips
-from shared.services.scope_filter import ip_excluded
+from shared.services.scope_filter import ip_excluded, matches_any
from shared.utils.datetime import utc_now
from stages.base import DOMAIN_TARGETS, Stage, StageResult
-from stages.netblock_sweep.config import NetblockSweepConfig
+from stages.netblock_sweep.config import (
+ MAX_ASN_ADDRESSES,
+ MIN_ADDRESSES,
+ MIN_SHARE,
+ NetblockSweepConfig,
+)
from stages.subdomain.parser import in_scope, normalize_host, passes_included
from tools.dnsx.client import DnsxClient, DnsxError
from tools.dnsx.parser import parse_dnsx_jsonl
@@ -65,6 +71,8 @@ class NetblockSweepStage(Stage):
produces = frozenset({AssetKind.HOSTS.value})
applies_to = DOMAIN_TARGETS
tools = ("dnsx",)
+ transport_tool = TransportTool.DNSX.value
+ thread_weight = 5 / 3
touches_target = False
config_model = NetblockSweepConfig
@@ -83,7 +91,7 @@ class NetblockSweepStage(Stage):
if not ips:
return StageResult(counts={"hosts": 0})
- owned, rejected = self._owned_networks(ips, cfg)
+ owned, rejected = self._owned_networks(ips)
if not owned:
note = (
f"No network attributed. {rejected} candidate networks were too "
@@ -100,7 +108,7 @@ class NetblockSweepStage(Stage):
label = ", ".join(f"AS{n.asn}" for n in owned)
self.emit_progress(f"sweeping {len(addresses):,} addresses in {label}")
- names, note = self._sweep(addresses, cfg)
+ names, note = self._sweep(addresses)
in_scope_names, foreign = self._scope(names)
added = self._persist(in_scope_names)
@@ -131,16 +139,14 @@ class NetblockSweepStage(Stage):
partial=bool(note) or truncated,
)
- def _owned_networks(
- self, ips: list[str], cfg: NetblockSweepConfig
- ) -> tuple[list[Network], int]:
+ def _owned_networks(self, ips: list[str]) -> tuple[list[Network], int]:
"""ASNs meeting the share and size thresholds."""
rows = self.session.execute(_ASN_SQL, {"ips": ips}).all()
client = RIPEStatClient()
owned: list[Network] = []
rejected = 0
for asn, as_name, hosts in rows:
- if hosts < cfg.min_addresses or hosts * 100 < len(ips) * cfg.min_share:
+ if hosts < MIN_ADDRESSES or hosts * 100 < len(ips) * MIN_SHARE:
continue
self._check_abort()
net = Network(asn=int(asn), name=as_name or "", hosts=int(hosts))
@@ -159,7 +165,7 @@ class NetblockSweepStage(Stage):
continue
net.prefixes.append(str(block))
net.addresses += block.num_addresses
- if not net.prefixes or net.addresses > cfg.max_asn_addresses:
+ if not net.prefixes or net.addresses > MAX_ASN_ADDRESSES:
rejected += 1
continue
owned.append(net)
@@ -184,11 +190,11 @@ class NetblockSweepStage(Stage):
return out, True
return out, False
- def _client(self, cfg: NetblockSweepConfig, count: int) -> DnsxClient | None:
+ def _client(self, count: int) -> DnsxClient | None:
try:
return DnsxClient(
timeout=max(_MIN_BUDGET, count // _FLOOR_RATE),
- threads=cfg.dns_threads,
+ threads=self.transport.threads,
recorder=self.ctx.recorder,
extra_args=self.ctx.resolved.tool_args("dnsx"),
)
@@ -196,11 +202,9 @@ class NetblockSweepStage(Stage):
logger.warning("dnsx unavailable, skipping netblock sweep")
return None
- def _sweep(
- self, addresses: list[str], cfg: NetblockSweepConfig
- ) -> tuple[dict[str, str], str | None]:
+ def _sweep(self, addresses: list[str]) -> tuple[dict[str, str], str | None]:
"""PTR across the range, then forward-confirm each name resolves back into it."""
- client = self._client(cfg, len(addresses))
+ client = self._client(len(addresses))
if client is None:
return {}, f"dnsx unavailable. {len(addresses):,} addresses were not swept."
@@ -218,15 +222,13 @@ class NetblockSweepStage(Stage):
note = f"dnsx {kind}. {len(addresses):,} addresses were only partly swept."
if found:
- confirmed = self._forward_confirm(list(found), set(addresses), cfg)
+ confirmed = self._forward_confirm(list(found), set(addresses))
found = {name: ip for name, ip in found.items() if name in confirmed}
return found, note
- def _forward_confirm(
- self, names: list[str], swept: set[str], cfg: NetblockSweepConfig
- ) -> set[str]:
+ def _forward_confirm(self, names: list[str], swept: set[str]) -> set[str]:
"""Names whose forward record points back into the swept range."""
- client = self._client(cfg, len(names))
+ client = self._client(len(names))
if client is None:
return set(names)
result = client.query(names, record_types=["a"])
@@ -268,6 +270,7 @@ class NetblockSweepStage(Stage):
source = SubdomainSource.NETBLOCK.value
now = utc_now()
added = 0
+ excluded = self.ctx.resolved.excluded_subdomains or []
for name, ip in sorted(names.items()):
row = existing.get(name)
if row is not None:
@@ -284,6 +287,7 @@ class NetblockSweepStage(Stage):
sources=[source],
resolved_ips=[ip],
is_active=True,
+ is_excluded=matches_any(name, excluded),
discovered_at=now,
)
)
diff --git a/stages/origin_probe/config.py b/stages/origin_probe/config.py
index 17c181527..40540c1c2 100644
--- a/stages/origin_probe/config.py
+++ b/stages/origin_probe/config.py
@@ -2,7 +2,7 @@ from __future__ import annotations
from pydantic import Field
-from stages.config import StageConfig, rate, threads, timeout
+from stages.config import StageConfig, advanced
class OriginProbeConfig(StageConfig):
@@ -11,20 +11,13 @@ class OriginProbeConfig(StageConfig):
title="Probe addresses directly",
description="Request each address by IP and record what it serves without a hostname.",
)
- threads: int = threads(30, title="Threads")
- timeout: int = timeout(10, title="Timeout (s)")
- rate: int = rate(150, tool="httpx", title="Requests/s")
- max_addresses: int = Field(
- default=2048,
+ max_addresses: int = advanced(
+ 2048,
ge=1,
le=65536,
title="Address budget",
description="Stop after this many addresses.",
)
- max_ports_per_address: int = Field(
- default=6,
- ge=1,
- le=100,
- title="Ports per address",
- description="Cap the web ports requested on any one address.",
- )
+
+
+MAX_PORTS_PER_ADDRESS = 6
diff --git a/stages/origin_probe/stage.py b/stages/origin_probe/stage.py
index 34f1e1fa3..a78a26924 100644
--- a/stages/origin_probe/stage.py
+++ b/stages/origin_probe/stage.py
@@ -2,6 +2,7 @@ from __future__ import annotations
from sqlalchemy import select
+from shared.definitions.intensity import TransportTool
from shared.definitions.ports import DEFAULT_WEB_PORTS, ServiceClass
from shared.definitions.surface import SurfaceDimension
from shared.enums.scan import AssetKind, Phase, StageGroup, StageRole
@@ -14,7 +15,7 @@ from shared.services.scope_filter import ip_excluded
from shared.utils.datetime import utc_now
from shared.utils.net import host_port
from stages.base import ALL_TARGETS, Stage, StageResult
-from stages.origin_probe.config import OriginProbeConfig
+from stages.origin_probe.config import MAX_PORTS_PER_ADDRESS, OriginProbeConfig
from stages.origin_probe.finding import origin_finding
from tools.httpx.client import HttpxClient, HttpxError
from tools.httpx.parser import parse_httpx_record
@@ -39,6 +40,8 @@ class OriginProbeStage(Stage):
consumes = frozenset({AssetKind.PORTS.value})
applies_to = ALL_TARGETS
tools = ("httpx",)
+ transport_tool = TransportTool.HTTPX.value
+ thread_weight = 0.2
config_model = OriginProbeConfig
def run(self) -> StageResult:
@@ -51,9 +54,9 @@ class OriginProbeStage(Stage):
try:
client = HttpxClient(
- rate_limit=cfg.rate,
- threads=cfg.threads,
- timeout=cfg.timeout,
+ rate_limit=self.transport.rate,
+ threads=self.transport.threads,
+ timeout=self.transport.timeout,
proxy_url=net.proxy_url,
headers=net.headers,
probe_scheme=net.probe_scheme,
@@ -129,7 +132,7 @@ class OriginProbeStage(Stage):
targets: list[str] = []
for ip in list(by_ip)[: cfg.max_addresses]:
ports = sorted(by_ip[ip], key=lambda p: (p not in DEFAULT_WEB_PORTS, p))[
- : cfg.max_ports_per_address
+ :MAX_PORTS_PER_ADDRESS
]
targets.extend(host_port(ip, port) for port in ports)
return targets
diff --git a/stages/passive_ports/config.py b/stages/passive_ports/config.py
index 29890b5c4..df2b61bb6 100644
--- a/stages/passive_ports/config.py
+++ b/stages/passive_ports/config.py
@@ -11,10 +11,6 @@ class PassivePortsConfig(StageConfig):
title="Look up indexed ports",
description="Read ports indexed for each address by internet-wide scanners.",
)
- max_addresses: int = Field(
- default=1024,
- ge=1,
- le=65536,
- title="Address budget",
- description="Stop after this many addresses.",
- )
+
+
+MAX_ADDRESSES = 1024
diff --git a/stages/passive_ports/stage.py b/stages/passive_ports/stage.py
index 821a31cdd..e0380e7f8 100644
--- a/stages/passive_ports/stage.py
+++ b/stages/passive_ports/stage.py
@@ -13,7 +13,7 @@ from shared.services import ip_inventory, port_inventory
from shared.services.port_inventory import ServiceObservation
from shared.services.scope_filter import ip_excluded
from stages.base import ALL_TARGETS, Stage, StageResult
-from stages.passive_ports.config import PassivePortsConfig
+from stages.passive_ports.config import MAX_ADDRESSES, PassivePortsConfig
from tools.naabu.client import NaabuClient, NaabuError, NaabuOptions
logger = get_logger(__name__)
@@ -36,7 +36,6 @@ class PassivePortsStage(Stage):
def run(self) -> StageResult:
self._check_abort()
- cfg = self.cfg
ip_inventory.ensure(
self.session,
scan_id=self.ctx.scan_id,
@@ -44,7 +43,7 @@ class PassivePortsStage(Stage):
project_id=self.ctx.project_id,
)
self.publish_results(SurfaceDimension.IPS.value)
- addresses = self._candidates(cfg.max_addresses)
+ addresses = self._candidates(MAX_ADDRESSES)
if not addresses:
return StageResult(counts={"known_ports": 0, "addresses": 0})
diff --git a/stages/port_scan/config.py b/stages/port_scan/config.py
index 0ef7e0597..89342271d 100644
--- a/stages/port_scan/config.py
+++ b/stages/port_scan/config.py
@@ -8,7 +8,7 @@ from shared.definitions.ports import (
PortProfile,
ScanPolicy,
)
-from stages.config import StageConfig, rate, threads, timeout
+from stages.config import StageConfig, advanced
_PROFILE_LABELS = {spec.key: spec.label for spec in PORT_PROFILES}
_SCAN_TYPES = {"connect": "Connect", "syn": "SYN"}
@@ -32,59 +32,41 @@ class PortScanConfig(StageConfig):
title="Custom ports",
description="Used only when the port set is Custom. A list or range like 80,443,8000-8100.",
)
- exclude_ports: str = Field(
- default="",
+ exclude_ports: str = advanced(
+ "",
max_length=2000,
title="Exclude ports",
description="Ports excluded from every scan, as a list or range.",
)
- scan_type: str = Field(
- default="connect",
+ scan_type: str = advanced(
+ "connect",
title="Scan type",
- description="Connect completes the TCP handshake. SYN needs raw sockets.",
+ description="Connect completes the TCP handshake. SYN needs raw sockets on the worker.",
json_schema_extra={"options": list(_SCAN_TYPES), "option_labels": _SCAN_TYPES},
)
- rate: int = rate(1000, tool="naabu", title="Packet rate (pps)")
- threads: int = threads(
- 100,
- title="Concurrency",
- description="Sockets in flight.",
- )
- timeout: int = timeout(3, title="Timeout (s)")
- retries: int = Field(
- default=1,
- ge=0,
- le=5,
- title="Retries",
- description="Extra attempts per port.",
- )
- cdn_policy: ScanPolicy = Field(
- default=ScanPolicy.WEB,
+ cdn_policy: ScanPolicy = advanced(
+ ScanPolicy.WEB,
title="CDN-fronted addresses",
description="Port set for addresses attributed to a CDN or WAF.",
json_schema_extra={"option_labels": SCAN_POLICY_LABELS},
)
- scan_cloud: bool = Field(
- default=True,
+ scan_cloud: bool = advanced(
+ True,
title="Scan cloud addresses in full",
description="Scan addresses attributed to a cloud provider with the full port set.",
)
- skip_private: bool = Field(
- default=True,
+ skip_private: bool = advanced(
+ True,
title="Skip private addresses",
description="Skip loopback, link-local and RFC1918 addresses. An address or netblock named as the target is scanned.",
)
- port_threshold: int = Field(
- default=500,
- ge=0,
- le=65535,
- title="Open-port threshold",
- description="Discard a host reporting more open ports than this. 0 disables.",
- )
- max_addresses: int = Field(
- default=8192,
+ max_addresses: int = advanced(
+ 8192,
ge=1,
le=100000,
title="Address budget",
description="Stop after this many addresses.",
)
+
+
+PORT_THRESHOLD = 500
diff --git a/stages/port_scan/stage.py b/stages/port_scan/stage.py
index ead54eb84..a5f0f2665 100644
--- a/stages/port_scan/stage.py
+++ b/stages/port_scan/stage.py
@@ -5,6 +5,7 @@ from dataclasses import dataclass
from sqlalchemy import select
+from shared.definitions.intensity import TransportTool
from shared.definitions.ports import (
CDN_EDGE_PORTS,
PortProfile,
@@ -22,7 +23,7 @@ from shared.services.port_inventory import ServiceObservation
from shared.services.scope_filter import ip_excluded
from shared.utils.net import is_registry_routable
from stages.base import ALL_TARGETS, Stage, StageResult
-from stages.port_scan.config import PortScanConfig
+from stages.port_scan.config import PORT_THRESHOLD, PortScanConfig
from tools.naabu.client import NaabuClient, NaabuError, NaabuOptions, port_args
logger = get_logger(__name__)
@@ -51,6 +52,7 @@ class PortScanStage(Stage):
produces = frozenset({AssetKind.PORTS.value})
applies_to = ALL_TARGETS
tools = ("naabu",)
+ transport_tool = TransportTool.NAABU.value
config_model = PortScanConfig
def run(self) -> StageResult:
@@ -80,12 +82,12 @@ class PortScanStage(Stage):
try:
client = NaabuClient(
options=NaabuOptions(
- rate=cfg.rate,
- concurrency=cfg.threads,
- timeout=cfg.timeout,
- retries=cfg.retries,
+ rate=self.transport.rate or 1,
+ concurrency=self.transport.threads,
+ timeout=self.transport.timeout,
+ retries=self.transport.retries,
scan_type=cfg.scan_type,
- port_threshold=cfg.port_threshold,
+ port_threshold=PORT_THRESHOLD,
exclude_ports=cfg.exclude_ports,
proxy_url=self.net_options().proxy_url,
extra_args=self.ctx.resolved.tool_args("naabu"),
diff --git a/stages/registry.py b/stages/registry.py
index c56fc40c9..e808309c6 100644
--- a/stages/registry.py
+++ b/stages/registry.py
@@ -9,7 +9,8 @@ from functools import lru_cache
from pathlib import Path
import stages as stages_pkg
-from shared.enums.scan import PHASE_ORDER, AssetKind, StageGroup, StageRole
+from shared.definitions.intensity import PROFILES, RATE_TOOLS
+from shared.enums.scan import PHASE_ORDER, AssetKind, Intensity, StageGroup, StageRole
from stages.base import Stage
from stages.config import StageConfig
@@ -35,6 +36,10 @@ class StageSpec:
produces: frozenset[str]
group: str
role: str
+ transport_tool: str | None
+ rate_weight: float
+ thread_weight: float
+ transport_timeout: int | None
stage_cls: type[Stage]
config_model: type[StageConfig]
@@ -46,6 +51,12 @@ class StageSpec:
def schema(self) -> dict:
return self.config_model.model_json_schema()
+ def transport(self, intensity: str, **scaling):
+ """The stage's transport under this intensity, or None when it runs no tool."""
+ if self.transport_tool is None:
+ return None
+ return self.stage_cls.default_transport(intensity, **scaling)
+
class StageRegistrationError(RuntimeError):
"""A stage module declares an invalid or duplicate stage."""
@@ -111,6 +122,10 @@ def _spec(stage_cls: type[Stage], level: int) -> StageSpec:
if unknown:
msg = f"{stage_cls.name}: unknown asset kind {sorted(unknown)[0]!r}."
raise StageRegistrationError(msg)
+ tool = stage_cls.transport_tool
+ if tool is not None and tool not in PROFILES[Intensity.NORMAL.value]:
+ msg = f"{stage_cls.name}: no transport profile for {tool!r}."
+ raise StageRegistrationError(msg)
return StageSpec(
name=stage_cls.name,
title=getattr(stage_cls, "title", None)
@@ -132,6 +147,10 @@ def _spec(stage_cls: type[Stage], level: int) -> StageSpec:
produces=frozenset(stage_cls.produces),
group=stage_cls.group,
role=stage_cls.role,
+ transport_tool=stage_cls.transport_tool,
+ rate_weight=stage_cls.rate_weight,
+ thread_weight=stage_cls.thread_weight,
+ transport_timeout=stage_cls.transport_timeout,
stage_cls=stage_cls,
config_model=stage_cls.config_model,
)
@@ -267,12 +286,8 @@ def phases() -> list[tuple[str, list[StageSpec]]]:
def rate_tools() -> tuple[str, ...]:
- tools: list[str] = []
- for spec in stages():
- for tool in sorted(spec.config_model.rate_tools()):
- if tool not in tools:
- tools.append(tool)
- return tuple(tools)
+ used = {spec.transport_tool for spec in stages()}
+ return tuple(tool for tool in RATE_TOOLS if tool in used)
__all__ = [
diff --git a/stages/reverse_dns/config.py b/stages/reverse_dns/config.py
index 34898b586..1cb75b8b4 100644
--- a/stages/reverse_dns/config.py
+++ b/stages/reverse_dns/config.py
@@ -2,7 +2,7 @@ from __future__ import annotations
from pydantic import Field
-from stages.config import StageConfig, threads, timeout
+from stages.config import StageConfig
class ReverseDnsConfig(StageConfig):
@@ -11,5 +11,3 @@ class ReverseDnsConfig(StageConfig):
title="Reverse DNS",
description="Resolve PTR records for every discovered IP.",
)
- dns_threads: int = threads(30, title="DNS threads")
- dns_timeout: int = timeout(5, title="DNS timeout (s)")
diff --git a/stages/reverse_dns/stage.py b/stages/reverse_dns/stage.py
index e59863c5b..0844bdc7c 100644
--- a/stages/reverse_dns/stage.py
+++ b/stages/reverse_dns/stage.py
@@ -2,9 +2,11 @@ from __future__ import annotations
from sqlalchemy import select
+from shared.definitions.intensity import TransportTool
from shared.enums.scan import AssetKind, Phase, StageGroup, StageRole
from shared.logging import get_logger
from shared.models.ip_address import IpAddress
+from shared.services.scope_filter import ip_excluded
from stages.base import ALL_TARGETS, Stage, StageResult
from stages.reverse_dns.config import ReverseDnsConfig
from tools.dnsx.client import DnsxClient, DnsxError
@@ -27,12 +29,12 @@ class ReverseDnsStage(Stage):
produces = frozenset({AssetKind.HOSTS.value})
applies_to = ALL_TARGETS
tools = ("dnsx",)
+ transport_tool = TransportTool.DNSX.value
touches_target = False
config_model = ReverseDnsConfig
def run(self) -> StageResult:
self._check_abort()
- cfg = self.cfg
rows = list(
self.session.execute(
select(IpAddress).where(IpAddress.scan_id == self.ctx.scan_id)
@@ -40,12 +42,15 @@ class ReverseDnsStage(Stage):
.scalars()
.all()
)
+ excluded = self.ctx.resolved.excluded_ips or []
+ if excluded:
+ rows = [row for row in rows if not ip_excluded(row.ip, excluded)]
if not rows:
return StageResult(counts={"ptr": 0})
by_ip = {row.ip: row for row in rows}
ips = list(by_ip.keys())[:_MAX_PTR]
- ptr_map, note = self._lookup_ptr(ips, cfg)
+ ptr_map, note = self._lookup_ptr(ips)
resolved = 0
for ip, names in ptr_map.items():
@@ -62,15 +67,13 @@ class ReverseDnsStage(Stage):
partial=bool(note),
)
- def _lookup_ptr(
- self, ips: list[str], cfg: ReverseDnsConfig
- ) -> tuple[dict[str, list], str | None]:
+ def _lookup_ptr(self, ips: list[str]) -> tuple[dict[str, list], str | None]:
if not ips:
return {}, None
try:
client = DnsxClient(
- timeout=max(120, cfg.dns_timeout),
- threads=cfg.dns_threads,
+ timeout=max(120, self.transport.timeout),
+ threads=self.transport.threads,
recorder=self.ctx.recorder,
extra_args=self.ctx.resolved.tool_args("dnsx"),
)
diff --git a/stages/screenshot/config.py b/stages/screenshot/config.py
index ed5f704d8..1c211a88e 100644
--- a/stages/screenshot/config.py
+++ b/stages/screenshot/config.py
@@ -2,7 +2,7 @@ from __future__ import annotations
from pydantic import Field
-from stages.config import StageConfig, rate, threads, timeout
+from stages.config import StageConfig
class ScreenshotConfig(StageConfig):
@@ -11,8 +11,3 @@ class ScreenshotConfig(StageConfig):
title="Capture screenshots",
description="Render every live HTTP service to an image.",
)
- threads: int = threads(40, title="Threads")
- rate: int = rate(150, tool="httpx", title="Renders/s")
- timeout: int = timeout(
- 10, title="Timeout (s)", description="Per-page render budget."
- )
diff --git a/stages/screenshot/stage.py b/stages/screenshot/stage.py
index b2c376178..55874f7c7 100644
--- a/stages/screenshot/stage.py
+++ b/stages/screenshot/stage.py
@@ -4,6 +4,7 @@ from pathlib import Path
from sqlalchemy import select, update
+from shared.definitions.intensity import TransportTool
from shared.definitions.surface import SurfaceDimension
from shared.enums.scan import AssetKind, Phase, StageGroup, StageRole
from shared.enums.target import TargetType
@@ -42,11 +43,12 @@ class ScreenshotStage(Stage):
role = StageRole.CAPABILITY.value
consumes = frozenset({AssetKind.HTTP_ASSETS.value})
tools = ("httpx",)
+ transport_tool = TransportTool.HTTPX.value
+ thread_weight = 4 / 15
config_model = ScreenshotConfig
def run(self) -> StageResult:
self._check_abort()
- cfg = self.cfg
net = self.net_options()
live = self.session.execute(
select(HttpAsset.id, HttpAsset.url).where(
@@ -60,9 +62,9 @@ class ScreenshotStage(Stage):
store_dir = str(Path(_MEDIA_ROOT) / str(self.ctx.scan_id))
try:
client = HttpxClient(
- rate_limit=cfg.rate,
- threads=cfg.threads,
- timeout=cfg.timeout,
+ rate_limit=self.transport.rate,
+ threads=self.transport.threads,
+ timeout=self.transport.timeout,
proxy_url=net.proxy_url,
headers=net.headers,
probe_scheme=net.probe_scheme,
diff --git a/stages/seed_resolution/config.py b/stages/seed_resolution/config.py
index 5b3bdcac1..b2e3bed0d 100644
--- a/stages/seed_resolution/config.py
+++ b/stages/seed_resolution/config.py
@@ -12,7 +12,7 @@ _MAX_HOSTS = {"smart": 4096, "full": 65536}
class SeedResolutionConfig(StageConfig):
enabled: bool = Field(
default=True,
- title="Enabled",
+ title="Seed resolution",
description="Expand an IP, netblock, ASN or URL seed into individual addresses.",
)
asn_scan_mode: Literal["smart", "full"] = Field(
@@ -20,11 +20,6 @@ class SeedResolutionConfig(StageConfig):
title="ASN / CIDR expansion",
description="Smart samples large netblocks. Full enumerates every host.",
)
- cidr_skip_rfc1918: bool = Field(
- default=True,
- title="Skip private ranges in ASN prefixes",
- description="Drop RFC1918 addresses when expanding an ASN's prefixes. A netblock target is expanded in full.",
- )
@property
def max_expansion_hosts(self) -> int:
diff --git a/stages/seed_resolution/stage.py b/stages/seed_resolution/stage.py
index 34bf55b52..68c28966a 100644
--- a/stages/seed_resolution/stage.py
+++ b/stages/seed_resolution/stage.py
@@ -9,6 +9,7 @@ from shared.enums.scan import AssetKind, Phase, StageGroup, StageRole
from shared.enums.target import TargetType
from shared.logging import get_logger
from shared.models.ip_address import IpAddress
+from shared.services.scope_filter import ip_excluded
from shared.utils.cidr import expand_network, parse_network
from shared.utils.datetime import utc_now
from shared.utils.validation import normalize_domain
@@ -166,7 +167,7 @@ class SeedResolutionStage(Stage):
if net is None:
continue
ips, was_truncated = expand_network(
- prefix, max_hosts=per_prefix, skip_private=cfg.cidr_skip_rfc1918
+ prefix, max_hosts=per_prefix, skip_private=True
)
truncated = truncated or was_truncated
records.extend(
@@ -202,9 +203,10 @@ class SeedResolutionStage(Stage):
)
now = utc_now()
seen: set[str] = set()
+ excluded = self.ctx.resolved.excluded_ips or []
for rec in records:
ip = rec["ip"]
- if ip in seen:
+ if ip in seen or (excluded and ip_excluded(ip, excluded)):
continue
seen.add(ip)
self.session.add(
diff --git a/stages/service_fingerprint/config.py b/stages/service_fingerprint/config.py
index b40e35ea7..06f976c14 100644
--- a/stages/service_fingerprint/config.py
+++ b/stages/service_fingerprint/config.py
@@ -2,7 +2,7 @@ from __future__ import annotations
from pydantic import Field
-from stages.config import StageConfig, threads, timeout
+from stages.config import StageConfig, advanced
class ServiceFingerprintConfig(StageConfig):
@@ -11,17 +11,10 @@ class ServiceFingerprintConfig(StageConfig):
title="Fingerprint services",
description="Identify the software behind every non-web port from its service banner.",
)
- threads: int = threads(32, title="Concurrency")
- timeout: int = timeout(4, title="Timeout (s)")
- max_services: int = Field(
- default=2000,
+ max_services: int = advanced(
+ 2000,
ge=1,
le=50000,
title="Service budget",
description="Stop after this many ports.",
)
- include_unknown: bool = Field(
- default=True,
- title="Include unrecognised ports",
- description="Also probe ports with no well-known service.",
- )
diff --git a/stages/service_fingerprint/stage.py b/stages/service_fingerprint/stage.py
index 5eb8d75ca..128efd95f 100644
--- a/stages/service_fingerprint/stage.py
+++ b/stages/service_fingerprint/stage.py
@@ -2,11 +2,11 @@ from __future__ import annotations
from sqlalchemy import select
+from shared.definitions.intensity import TransportTool
from shared.definitions.ports import (
CDN_EDGE_PORTS,
PortSource,
ScanPolicy,
- ServiceClass,
likely_tls,
)
from shared.enums.scan import AssetKind, Phase, StageGroup, StageRole
@@ -36,6 +36,7 @@ class ServiceFingerprintStage(Stage):
consumes = frozenset({AssetKind.PORTS.value})
applies_to = ALL_TARGETS
tools = ()
+ transport_tool = TransportTool.BANNER.value
config_model = ServiceFingerprintConfig
def run(self) -> StageResult:
@@ -46,8 +47,8 @@ class ServiceFingerprintStage(Stage):
return StageResult(counts={"fingerprinted": 0, "probed": 0})
client = BannerClient(
- timeout=float(cfg.timeout),
- concurrency=cfg.threads,
+ timeout=float(self.transport.timeout),
+ concurrency=self.transport.threads,
proxy_url=self.net_options().proxy_url,
)
if client.proxy_warning:
@@ -112,8 +113,6 @@ class ServiceFingerprintStage(Stage):
Port.protocol == "tcp",
)
)
- if not cfg.include_unknown:
- query = query.where(Port.service_class != ServiceClass.OTHER.value)
rows = self.session.execute(
query.order_by(Port.ip, Port.number).limit(cfg.max_services)
).all()
diff --git a/stages/session_check/config.py b/stages/session_check/config.py
index e478fe96b..094a499a2 100644
--- a/stages/session_check/config.py
+++ b/stages/session_check/config.py
@@ -2,7 +2,7 @@ from __future__ import annotations
from pydantic import Field
-from stages.config import StageConfig, timeout
+from stages.config import StageConfig
class SessionCheckConfig(StageConfig):
@@ -11,4 +11,3 @@ class SessionCheckConfig(StageConfig):
title="Check the session before scanning",
description="Request the target once with the scan context's credentials and once without. Runs only when the context carries credentials.",
)
- timeout: int = timeout(15, title="Timeout (s)")
diff --git a/stages/session_check/stage.py b/stages/session_check/stage.py
index dae15a822..b6e1d7652 100644
--- a/stages/session_check/stage.py
+++ b/stages/session_check/stage.py
@@ -10,6 +10,8 @@ from shared.logging import get_logger
from stages.base import ALL_TARGETS, Stage, StageResult
from stages.session_check.config import SessionCheckConfig
+REQUEST_TIMEOUT = 15
+
logger = get_logger(__name__)
_UNAUTHORISED = (401, 403)
@@ -73,16 +75,18 @@ class SessionCheckStage(Stage):
value = self.ctx.target_value.strip().rstrip("/")
if not value:
return []
+ scheme = self.net_options().probe_scheme
if self.ctx.target_type == TargetType.URL.value:
- return [value]
+ return [value] if scheme is None or value.startswith(f"{scheme}://") else []
if self.ctx.target_type != TargetType.DOMAIN.value:
return []
- return [f"https://{value}", f"http://{value}"]
+ schemes = [scheme] if scheme else ["https", "http"]
+ return [f"{s}://{value}" for s in schemes]
def _client(self) -> httpx.Client:
return httpx.Client(
- timeout=self.cfg.timeout,
- follow_redirects=True,
+ timeout=REQUEST_TIMEOUT,
+ follow_redirects=self.follow_redirects(True),
verify=False, # noqa: S501
proxy=self.ctx.resolved.proxy_url or None,
)
diff --git a/stages/subdomain/config.py b/stages/subdomain/config.py
index 29ec1171f..38baf005b 100644
--- a/stages/subdomain/config.py
+++ b/stages/subdomain/config.py
@@ -4,7 +4,7 @@ from pydantic import Field
from shared.definitions.wordlists import WordlistKind
from shared.enums.scan import Intensity
-from stages.config import StageConfig, threads, timeout, wordlist
+from stages.config import StageConfig, wordlist
from stages.subdomain.providers import PASSIVE_PROVIDERS
PASSIVE_TOOLS: tuple[str, ...] = tuple(
@@ -24,10 +24,15 @@ _TOOL_TIMEOUTS = {
}
+def tool_timeout(intensity: str) -> int:
+ """Seconds a passive source may run under this intensity."""
+ return _TOOL_TIMEOUTS.get(intensity, _TOOL_TIMEOUTS[Intensity.NORMAL.value])
+
+
class SubdomainConfig(StageConfig):
enabled: bool = Field(
default=True,
- title="Enabled",
+ title="Subdomain discovery",
description="Enumerate subdomains from passive sources, certificates, wordlists and permutations.",
)
passive_tools: list[str] = Field(
@@ -68,49 +73,14 @@ class SubdomainConfig(StageConfig):
title="Permute discovered names",
description="Resolve variants of discovered names, such as api-dev and api2.",
)
- permutation_seeds: int = Field(
- default=250,
- ge=1,
- le=10_000,
- title="Names to permute",
- description="Discovered names to build variants from.",
- )
- permutation_limit: int = Field(
- default=20_000,
- ge=100,
- le=500_000,
- title="Variants to resolve",
- description="Cap on generated variants.",
- )
- dns_threads: int = threads(30, title="Resolver threads")
- dns_batch_size: int = Field(
- default=1000,
- ge=100,
- le=20000,
- title="Resolver batch size",
- description="Names sent to the resolver per invocation.",
- )
- dns_batch_concurrency: int = Field(
- default=1,
- ge=1,
- le=8,
- title="Resolver batches in parallel",
- description="Resolver invocations in flight at once. Above 1 only with dedicated resolvers.",
- )
- dns_idle_timeout: int = timeout(
- 90,
- title="Resolver stall timeout",
- description="Abandon a resolver batch after this many seconds with no answer.",
- )
- dns_retry_silent: bool = Field(
- default=True,
- title="Re-query unanswered names",
- description="Ask the resolver a second time for names that returned nothing.",
- )
@property
def enabled_sources(self) -> list[str]:
return [tool for tool in self.passive_tools if tool in PASSIVE_PROVIDERS]
- def tool_timeout(self, intensity: str) -> int:
- return _TOOL_TIMEOUTS.get(intensity, _TOOL_TIMEOUTS[Intensity.NORMAL.value])
+
+PERMUTATION_SEEDS = 250
+PERMUTATION_LIMIT = 20_000
+DNS_BATCH_SIZE = 1000
+DNS_BATCH_CONCURRENCY = 1
+DNS_IDLE_TIMEOUT = 90
diff --git a/stages/subdomain/stage.py b/stages/subdomain/stage.py
index a1ffdd413..f05a06d3c 100644
--- a/stages/subdomain/stage.py
+++ b/stages/subdomain/stage.py
@@ -15,6 +15,7 @@ from pathlib import Path
from sqlalchemy import bindparam, cast, delete, not_, or_, update
from sqlalchemy.dialects.postgresql import JSONB, insert
+from shared.definitions.intensity import TransportTool
from shared.definitions.rescan import SEED_SOURCES
from shared.definitions.surface import SurfaceDimension
from shared.enums.activity import ActivityEvent, ActivityLevel
@@ -30,7 +31,16 @@ from shared.services.scope_filter import matches_any
from shared.services.wordlists import WordlistError, read_words
from shared.utils.datetime import utc_now
from stages.base import DOMAIN_TARGETS, Stage, StageResult
-from stages.subdomain.config import PASSIVE_TOOLS, SubdomainConfig
+from stages.subdomain.config import (
+ DNS_BATCH_CONCURRENCY,
+ DNS_BATCH_SIZE,
+ DNS_IDLE_TIMEOUT,
+ PASSIVE_TOOLS,
+ PERMUTATION_LIMIT,
+ PERMUTATION_SEEDS,
+ SubdomainConfig,
+ tool_timeout,
+)
from stages.subdomain.parser import in_scope, merge_and_filter
from stages.subdomain.providers import (
PASSIVE_PROVIDERS,
@@ -136,6 +146,7 @@ class SubdomainStage(Stage):
produces = frozenset({AssetKind.HOSTS.value, AssetKind.ADDRESSES.value})
applies_to = DOMAIN_TARGETS
tools = PASSIVE_TOOLS
+ transport_tool = TransportTool.DNSX.value
api_keys = tuple(p.value for p in _PREFETCH_KEYS)
touches_target = False
config_model = SubdomainConfig
@@ -145,6 +156,7 @@ class SubdomainStage(Stage):
return cfg.enabled and bool(
cfg.enabled_sources
or cfg.tls_discovery
+ or cfg.zone_transfer
or cfg.bruteforce
or cfg.permutations
)
@@ -162,8 +174,8 @@ class SubdomainStage(Stage):
api_keys = self._prefetch_keys()
pctx = ProviderContext(
domain=domain,
- timeout=cfg.tool_timeout(resolved.intensity),
- threads=cfg.dns_threads,
+ timeout=tool_timeout(resolved.intensity),
+ threads=self.transport.threads,
proxy_url=resolved.proxy_url,
api_keys=api_keys,
recorder=self.ctx.recorder,
@@ -182,7 +194,7 @@ class SubdomainStage(Stage):
self._check_abort()
merged = merge_and_filter(results, domain, resolved.included_subdomains)
- wildcard_ips = self._wildcard_ips(domain, cfg)
+ wildcard_ips = self._wildcard_ips(domain)
extra = self._expand(
domain, cfg, sorted(merged, key=_seed_rank), wildcard_ips, activity
)
@@ -205,7 +217,7 @@ class SubdomainStage(Stage):
self._write_names(merged)
to_resolve = [n for n in merged if n not in excluded]
- state = self._resolve(to_resolve, cfg, wildcard_ips)
+ state = self._resolve(to_resolve, wildcard_ips)
active, ips_seen = self._persist(merged, state.records, wildcard_ips, excluded)
return StageResult(
@@ -233,11 +245,11 @@ class SubdomainStage(Stage):
passive = self.ctx.resolved.intensity == Intensity.PASSIVE.value
out: list[ProviderResult] = []
if cfg.zone_transfer:
- out.append(self._zone_transfer(domain, cfg, passive=passive))
+ out.append(self._zone_transfer(domain, passive=passive))
if cfg.bruteforce:
out.append(self._bruteforce(domain, cfg, wildcard_ips, passive=passive))
if cfg.permutations:
- out.append(self._permute(cfg, seeds, wildcard_ips, passive=passive))
+ out.append(self._permute(seeds, wildcard_ips, passive=passive))
for result in out:
self._check_abort()
self._log_provider(activity, result)
@@ -270,7 +282,6 @@ class SubdomainStage(Stage):
def _zone_transfer(
self,
domain: str,
- cfg: SubdomainConfig,
*,
passive: bool,
) -> ProviderResult:
@@ -279,7 +290,7 @@ class SubdomainStage(Stage):
return self._skipped(
source, "a passive scan does not query the target's nameservers"
)
- client = self._client(cfg)
+ client = self._client()
if client is None:
return self._skipped(source, "dnsx is not installed on this instance")
@@ -335,7 +346,7 @@ class SubdomainStage(Stage):
return self._skipped(
source, "a passive scan does not query the target's nameservers"
)
- client = self._client(cfg)
+ client = self._client()
if client is None:
return self._skipped(source, "dnsx is not installed on this instance")
@@ -381,7 +392,6 @@ class SubdomainStage(Stage):
def _permute(
self,
- cfg: SubdomainConfig,
seeds: list[str],
wildcard_ips: set[str],
*,
@@ -398,7 +408,7 @@ class SubdomainStage(Stage):
)
try:
client = AlterxClient(
- limit=cfg.permutation_limit,
+ limit=PERMUTATION_LIMIT,
recorder=self.ctx.recorder,
extra_args=self.ctx.resolved.tool_args("alterx"),
)
@@ -406,11 +416,11 @@ class SubdomainStage(Stage):
return self._skipped(source, "alterx is not installed on this instance")
start = time.monotonic()
- candidates = client.permute(seeds[: cfg.permutation_seeds])
+ candidates = client.permute(seeds[:PERMUTATION_SEEDS])
if not candidates:
return ProviderResult(source=source, duration_seconds=0.0)
self.emit_progress(f"resolving {len(candidates):,} name variants")
- client = self._client(cfg)
+ client = self._client()
if client is None:
return self._skipped(source, "dnsx is not installed on this instance")
found: set[str] = set()
@@ -438,9 +448,9 @@ class SubdomainStage(Stage):
duration_seconds=round(time.monotonic() - start, 2),
)
- def _wildcard_ips(self, domain: str, cfg: SubdomainConfig) -> set[str]:
+ def _wildcard_ips(self, domain: str) -> set[str]:
probe = f"{uuid.uuid4().hex[:12]}.{domain}"
- info = self._resolve([probe], cfg).records.get(probe)
+ info = self._resolve([probe]).records.get(probe)
return set(info["ips"]) if info and info.get("ips") else set()
@staticmethod
@@ -530,11 +540,13 @@ class SubdomainStage(Stage):
self.session.commit()
self.emit_progress(message, source=source)
- def _client(self, cfg: SubdomainConfig) -> DnsxClient | None:
- threads = min(max(cfg.dns_threads, _MIN_RESOLVE_THREADS), _MAX_RESOLVE_THREADS)
+ def _client(self) -> DnsxClient | None:
+ threads = min(
+ max(self.transport.threads, _MIN_RESOLVE_THREADS), _MAX_RESOLVE_THREADS
+ )
try:
return DnsxClient(
- timeout=max(120, cfg.tool_timeout(self.ctx.resolved.intensity)),
+ timeout=max(120, tool_timeout(self.ctx.resolved.intensity)),
threads=threads,
recorder=self.ctx.recorder,
extra_args=self.ctx.resolved.tool_args("dnsx"),
@@ -558,14 +570,14 @@ class SubdomainStage(Stage):
}
def _run_batch(
- self, client: DnsxClient, names: list[str], cfg: SubdomainConfig
+ self, client: DnsxClient, names: list[str]
) -> tuple[dict[str, dict], bool]:
"""Resolve one batch."""
out: dict[str, dict] = {}
with client.stream_query(
names,
record_types=["a", "aaaa", "cname"],
- idle_timeout=cfg.dns_idle_timeout,
+ idle_timeout=DNS_IDLE_TIMEOUT,
) as stream:
for rec in stream.records:
parsed = self._record(rec)
@@ -577,28 +589,27 @@ class SubdomainStage(Stage):
def _resolve(
self,
names: list[str],
- cfg: SubdomainConfig,
wildcard_ips: set[str] | None = None,
) -> _Resolution:
"""wildcard_ips is None for the wildcard probe itself, whose name is not stored."""
state = _Resolution(submitted=len(names))
if not names:
return state
- client = self._client(cfg)
+ client = self._client()
if client is None:
state.unavailable = True
return state
shuffled = list(names)
random.Random(_SHUFFLE_SEED).shuffle(shuffled) # noqa: S311
- size = max(1, cfg.dns_batch_size)
+ size = DNS_BATCH_SIZE
batches = [
_Batch(names=shuffled[i : i + size]) for i in range(0, len(shuffled), size)
]
state.batches = len(batches)
for done, (batch, (records, stalled)) in enumerate(
- self._resolve_batches(client, batches, cfg), start=1
+ self._resolve_batches(client, batches), start=1
):
state.records.update(records)
batch.answered = len(records)
@@ -618,26 +629,24 @@ class SubdomainStage(Stage):
note += ", stopped on silence"
self.emit_progress(note)
- self._retry_degraded(client, batches, state, cfg)
- self._retry_silent(client, names, state, cfg, wildcard_ips)
+ self._retry_degraded(client, batches, state)
+ self._retry_silent(client, names, state, wildcard_ips)
return state
- def _resolve_batches(
- self, client: DnsxClient, batches: list[_Batch], cfg: SubdomainConfig
- ):
+ def _resolve_batches(self, client: DnsxClient, batches: list[_Batch]):
"""Batches are independent."""
- workers = min(max(1, cfg.dns_batch_concurrency), len(batches))
+ workers = min(DNS_BATCH_CONCURRENCY, len(batches))
if workers == 1:
for batch in batches:
started = time.monotonic()
- answered = self._run_batch(client, batch.names, cfg)
+ answered = self._run_batch(client, batch.names)
batch.seconds = time.monotonic() - started
yield batch, answered
return
with ThreadPoolExecutor(max_workers=workers) as pool:
started = time.monotonic()
futures = {
- pool.submit(self._run_batch, client, batch.names, cfg): batch
+ pool.submit(self._run_batch, client, batch.names): batch
for batch in batches
}
for future in as_completed(futures):
@@ -650,7 +659,6 @@ class SubdomainStage(Stage):
client: DnsxClient,
batches: list[_Batch],
state: _Resolution,
- cfg: SubdomainConfig,
) -> None:
"""Resolve again any batch far below its peers."""
healthy = [b.rate for b in batches if not b.stalled]
@@ -669,7 +677,7 @@ class SubdomainStage(Stage):
]
retries = [(batch, pending) for batch, pending in retries if pending]
state.retried += len(retries)
- for batch, (records, stalled) in self._retry_batches(client, retries, cfg):
+ for batch, (records, stalled) in self._retry_batches(client, retries):
batch.stalled = stalled
state.records.update(records)
batch.answered += len(records)
@@ -683,13 +691,10 @@ class SubdomainStage(Stage):
client: DnsxClient,
names: list[str],
state: _Resolution,
- cfg: SubdomainConfig,
wildcard_ips: set[str] | None,
) -> None:
"""dnsx says nothing for a name it dropped and for a name that does not exist."""
- if not cfg.dns_retry_silent:
- return
- size = max(1, cfg.dns_batch_size)
+ size = DNS_BATCH_SIZE
for _ in range(_SILENCE_PASSES):
pending = [n for n in names if n not in state.records]
if not pending:
@@ -699,7 +704,7 @@ class SubdomainStage(Stage):
for start in range(0, len(pending), size):
self._check_abort()
records, _stalled = self._run_batch(
- client, pending[start : start + size], cfg
+ client, pending[start : start + size]
)
state.records.update(records)
found += len(records)
@@ -714,16 +719,15 @@ class SubdomainStage(Stage):
self,
client: DnsxClient,
retries: list[tuple[_Batch, list[str]]],
- cfg: SubdomainConfig,
):
- workers = min(max(1, cfg.dns_batch_concurrency), len(retries) or 1)
+ workers = min(DNS_BATCH_CONCURRENCY, len(retries) or 1)
if workers == 1:
for batch, pending in retries:
- yield batch, self._run_batch(client, pending, cfg)
+ yield batch, self._run_batch(client, pending)
return
with ThreadPoolExecutor(max_workers=workers) as pool:
futures = {
- pool.submit(self._run_batch, client, pending, cfg): batch
+ pool.submit(self._run_batch, client, pending): batch
for batch, pending in retries
}
for future in as_completed(futures):
diff --git a/stages/target_enrichment/config.py b/stages/target_enrichment/config.py
index fb28e2712..53fedda9e 100644
--- a/stages/target_enrichment/config.py
+++ b/stages/target_enrichment/config.py
@@ -2,14 +2,12 @@ from __future__ import annotations
from pydantic import Field
-from stages.config import StageConfig, threads, timeout
+from stages.config import StageConfig
class TargetEnrichmentConfig(StageConfig):
enabled: bool = Field(
default=True,
- title="Enabled",
+ title="Target enrichment",
description="Resolve the target and attach DNS, WHOIS and BGP context.",
)
- dns_threads: int = threads(30, title="DNS threads")
- dns_timeout: int = timeout(5, title="DNS timeout (s)")
diff --git a/stages/target_enrichment/stage.py b/stages/target_enrichment/stage.py
index efdd71769..d154b87b2 100644
--- a/stages/target_enrichment/stage.py
+++ b/stages/target_enrichment/stage.py
@@ -4,6 +4,7 @@ from datetime import datetime, timedelta
from sqlalchemy import select
+from shared.definitions.intensity import TransportTool
from shared.enums.scan import Phase, StageGroup, StageRole
from shared.enums.target import TargetType
from shared.enums.task_status import TaskStatus
@@ -39,19 +40,19 @@ class TargetEnrichmentStage(Stage):
group = StageGroup.HOSTS.value
role = StageRole.SUPPORT.value
tools = ("dnsx", "whois")
+ transport_tool = TransportTool.DNSX.value
touches_target = False
config_model = TargetEnrichmentConfig
_notes: list[str]
def run(self) -> StageResult:
self._check_abort()
- cfg = self.cfg
target = self.session.get(Target, self.ctx.target_id)
if target is None:
return StageResult(counts={})
self._notes = []
- dns_records = self._ensure_dns(target, cfg)
+ dns_records = self._ensure_dns(target)
whois_present = self._ensure_whois(target)
bgp_present = self._read_bgp()
@@ -70,7 +71,7 @@ class TargetEnrichmentStage(Stage):
partial=bool(self._notes),
)
- def _ensure_dns(self, target: Target, cfg: TargetEnrichmentConfig) -> int:
+ def _ensure_dns(self, target: Target) -> int:
if self.ctx.target_type not in _DNS_TYPES:
return 0
lookup = (
@@ -82,8 +83,8 @@ class TargetEnrichmentStage(Stage):
host = normalize_domain(self.ctx.target_value)
try:
lookup = DnsxService(
- timeout=max(120, cfg.dns_timeout),
- threads=cfg.dns_threads,
+ timeout=max(120, self.transport.timeout),
+ threads=self.transport.threads,
recorder=self.ctx.recorder,
extra_args=self.ctx.resolved.tool_args("dnsx"),
).lookup_and_store(self.session, target.id, host)
diff --git a/stages/url_discovery/config.py b/stages/url_discovery/config.py
index af9a61a67..effa324db 100644
--- a/stages/url_discovery/config.py
+++ b/stages/url_discovery/config.py
@@ -13,7 +13,7 @@ from shared.definitions.endpoints import (
CrawlScope,
EndpointSource,
)
-from stages.config import StageConfig, rate, threads, timeout
+from stages.config import StageConfig, advanced
DEFAULT_PROVIDERS: list[str] = [
EndpointSource.RESPONSE_MINING.value,
@@ -44,9 +44,6 @@ class UrlDiscoveryConfig(StageConfig):
description="Sources URLs are collected from. Response mining reads stored response bodies and sends no request.",
json_schema_extra={"options": list(_PROVIDER_LABELS)},
)
- threads: int = threads(50, title="Threads")
- timeout: int = timeout(15, title="Timeout (s)")
- rate: int = rate(150, tool="katana", title="Requests/s")
crawl_depth: int = Field(
default=3,
ge=1,
@@ -54,17 +51,6 @@ class UrlDiscoveryConfig(StageConfig):
title="Crawl depth",
description="Link depth followed from each site root.",
)
- crawl_scope: str = Field(
- default=CrawlScope.RDN.value,
- title="Crawl scope",
- description="Which hosts the crawler may follow links to.",
- json_schema_extra={"options": list(CRAWL_SCOPES)},
- )
- crawl_javascript: bool = Field(
- default=True,
- title="Parse JavaScript",
- description="Read URLs out of the JavaScript each page loads.",
- )
headless: bool = Field(
default=False,
title="Use a browser",
@@ -77,68 +63,47 @@ class UrlDiscoveryConfig(StageConfig):
title="Crawl budget (min)",
description="Stop crawling after this long. 0 means no limit.",
)
- max_urls: int = Field(
- default=50000,
- ge=100,
- le=200000,
- title="URLs per source",
- description="Cap the URLs any one source may contribute.",
+ crawl_scope: str = advanced(
+ CrawlScope.RDN.value,
+ title="Crawl scope",
+ description="Which hosts the crawler may follow links to.",
+ json_schema_extra={"options": list(CRAWL_SCOPES)},
)
- max_known_file_hosts: int = Field(
- default=200,
- ge=1,
- le=5000,
- title="Hosts for robots and sitemaps",
- description="Cap the hosts whose robots.txt and sitemap are fetched.",
- )
- max_source_maps: int = Field(
- default=200,
- ge=1,
- le=5_000,
- title="Bundles to ask for a map",
- description="JavaScript bundles asked for the .map beside them.",
- )
- max_archive_domains: int = Field(
- default=10,
- ge=1,
- le=200,
- title="Domains to query archives for",
- description="Registrable domains queried in public archives.",
- )
- max_hosts: int = Field(
- default=500,
+ max_hosts: int = advanced(
+ 500,
ge=1,
le=10000,
title="Hosts to crawl",
description="Cap the live web assets handed to the crawler.",
)
-
- drop_noise: bool = Field(
- default=True,
+ drop_noise: bool = advanced(
+ True,
title="Drop noise",
description="Static files, crawler artifacts, platform noise and URLs past the family and sibling caps are not stored. Every drop is counted.",
)
- static_extensions: list[str] = Field(
+ static_extensions: list[str] = advanced(
+ None,
default_factory=lambda: list(NOISE_STATIC_EXTENSIONS),
max_length=200,
title="Static extensions",
description="File extensions dropped as static.",
)
- ignored_params: list[str] = Field(
+ ignored_params: list[str] = advanced(
+ None,
default_factory=lambda: list(NOISE_IGNORED_PARAMS),
max_length=400,
title="Ignored parameters",
description="Query parameters removed from every URL. Tracking and cache-busting names by default.",
)
- keep_per_family: int = Field(
- default=NOISE_KEEP_PER_FAMILY,
+ keep_per_family: int = advanced(
+ NOISE_KEEP_PER_FAMILY,
ge=1,
le=MAX_NOISE_KEEP_PER_FAMILY,
title="URLs kept per family",
description="URLs stored for one path shape and parameter set. The rest are dropped.",
)
- sibling_cap: int = Field(
- default=NOISE_SIBLING_CAP,
+ sibling_cap: int = advanced(
+ NOISE_SIBLING_CAP,
ge=1,
le=MAX_NOISE_SIBLING_CAP,
title="Sibling cap",
@@ -163,3 +128,9 @@ class UrlDiscoveryConfig(StageConfig):
@classmethod
def _known_scope(cls, value: str) -> str:
return value if value in CRAWL_SCOPES else CrawlScope.RDN.value
+
+
+MAX_URLS = 50_000
+MAX_KNOWN_FILE_HOSTS = 200
+MAX_SOURCE_MAPS = 200
+MAX_ARCHIVE_DOMAINS = 10
diff --git a/stages/url_discovery/providers/archive.py b/stages/url_discovery/providers/archive.py
index 3b02dd7f3..c960fd50f 100644
--- a/stages/url_discovery/providers/archive.py
+++ b/stages/url_discovery/providers/archive.py
@@ -2,6 +2,7 @@ from __future__ import annotations
from shared.definitions.endpoints import EndpointSource
from shared.services.endpoint_inventory import EndpointObservation
+from stages.url_discovery.config import MAX_ARCHIVE_DOMAINS, MAX_URLS
from stages.url_discovery.providers.base import ProviderResult, UrlProvider
from tools.urlfinder.client import UrlfinderClient, UrlfinderError
@@ -15,13 +16,12 @@ class ArchiveProvider(UrlProvider):
touches_target = False
def discover(self, result: ProviderResult) -> None:
- cfg = self.ctx.cfg
- domains = self.ctx.apex_domains[: cfg.max_archive_domains]
+ domains = self.ctx.apex_domains[:MAX_ARCHIVE_DOMAINS]
if not domains:
return
try:
client = UrlfinderClient(
- timeout=cfg.timeout,
+ timeout=self.ctx.transport.timeout,
proxy_url=self.ctx.net.proxy_url,
recorder=self.ctx.recorder,
extra_args=self.extra_args,
@@ -32,7 +32,7 @@ class ArchiveProvider(UrlProvider):
found = 0
seen: set[str] = set()
observations: list[EndpointObservation] = []
- cap = cfg.max_urls
+ cap = MAX_URLS
scanned = 0
for domain in domains:
@@ -69,7 +69,7 @@ class ArchiveProvider(UrlProvider):
result.capped = True
result.cap_reason = (
f"{dropped} registrable domains not queried. Limit is "
- f"{cfg.max_archive_domains} domains."
+ f"{MAX_ARCHIVE_DOMAINS} domains."
)
result.urls_found = found
result.hosts_scanned = scanned
diff --git a/stages/url_discovery/providers/base.py b/stages/url_discovery/providers/base.py
index 193308e13..478dfb28f 100644
--- a/stages/url_discovery/providers/base.py
+++ b/stages/url_discovery/providers/base.py
@@ -1,6 +1,7 @@
from __future__ import annotations
import shutil
+import threading
import time
from abc import ABC, abstractmethod
from collections.abc import Callable
@@ -9,9 +10,11 @@ from functools import cached_property
from typing import TYPE_CHECKING, ClassVar
from urllib.parse import urlsplit
+from shared.definitions.intensity import Transport
from shared.definitions.vulnerabilities import CoverageStatus
from shared.logging import get_logger
from shared.services.endpoint_inventory import EndpointObservation
+from shared.services.scope_filter import matches_any
from shared.utils.datetime import utc_now
from tools.runner import tool_path
@@ -50,6 +53,7 @@ class ProviderContext:
hosts: list[Host]
apex_domains: list[str]
cfg: object
+ transport: Transport
resolved: ResolvedScanConfig
net: NetOptions
recorder: CommandRecorder | None = None
@@ -143,6 +147,41 @@ class UrlProvider(ABC):
def aborted(self) -> bool:
return self.ctx.is_aborted is not None and self.ctx.is_aborted()
+ def follow_redirects(self, default: bool) -> bool:
+ override = self.ctx.resolved.follow_redirects
+ return default if override is None else override
+
+ def path_excluded(self, url: str) -> bool:
+ excluded = self.ctx.resolved.excluded_paths or []
+ if not excluded:
+ return False
+ try:
+ path = urlsplit(url).path or "/"
+ except ValueError:
+ return True
+ return matches_any(path, excluded)
+
+ def workers(self, cap: int) -> int:
+ return max(1, min(cap, self.ctx.transport.threads))
+
+ def throttle(self) -> None:
+ """One request per 1/rate seconds across the provider's workers."""
+ rate = self.ctx.transport.rate
+ if not rate:
+ return
+ with self._gate:
+ now = time.monotonic()
+ wait = self._next_slot - now
+ if wait > 0:
+ time.sleep(wait)
+ now = time.monotonic()
+ self._next_slot = max(now, self._next_slot) + 1.0 / rate
+
+ @cached_property
+ def _gate(self) -> threading.Lock:
+ self._next_slot = 0.0
+ return threading.Lock()
+
@abstractmethod
def discover(self, result: ProviderResult) -> None:
"""Fill result.observations, and every count the provider actually knows."""
diff --git a/stages/url_discovery/providers/katana.py b/stages/url_discovery/providers/katana.py
index 23d8fce55..18162ebab 100644
--- a/stages/url_discovery/providers/katana.py
+++ b/stages/url_discovery/providers/katana.py
@@ -5,6 +5,7 @@ from dataclasses import dataclass, field
from shared.definitions.endpoints import EndpointSource
from shared.services.endpoint_inventory import EndpointObservation
+from stages.url_discovery.config import MAX_URLS
from stages.url_discovery.providers.base import ProviderResult, UrlProvider
from tools.katana.client import KatanaClient, KatanaError
from tools.katana.parser import parse_katana_record
@@ -32,12 +33,12 @@ class KatanaProvider(UrlProvider):
try:
return KatanaClient(
depth=cfg.crawl_depth,
- threads=cfg.threads,
- timeout=cfg.timeout,
+ threads=self.ctx.transport.threads,
+ timeout=self.ctx.transport.timeout,
max_duration_minutes=cfg.max_crawl_minutes,
- rate_limit=cfg.rate,
+ rate_limit=self.ctx.transport.rate,
crawl_scope=cfg.crawl_scope,
- include_js=cfg.crawl_javascript,
+ include_js=True,
headless=cfg.headless,
exclude_extensions=list(cfg.static_extensions)
if cfg.drop_noise
@@ -73,7 +74,7 @@ class KatanaProvider(UrlProvider):
with client.stream_crawl(
targets, should_stop=self.ctx.is_aborted, stderr_sink=_stderr
) as records:
- self._ingest(records, state, result, cfg.max_urls)
+ self._ingest(records, state, result, MAX_URLS)
if fatal:
msg = f"katana could not run: {fatal[0]}"
diff --git a/stages/url_discovery/providers/known_files.py b/stages/url_discovery/providers/known_files.py
index c4c66b6e6..d9e83d26a 100644
--- a/stages/url_discovery/providers/known_files.py
+++ b/stages/url_discovery/providers/known_files.py
@@ -11,6 +11,7 @@ import httpx
from shared.definitions.endpoints import EndpointSource
from shared.services.endpoint_inventory import EndpointObservation
from shared.utils.net import host_port
+from stages.url_discovery.config import MAX_KNOWN_FILE_HOSTS
from stages.url_discovery.providers.base import ProviderResult, UrlProvider
_ROBOTS = "/robots.txt"
@@ -38,12 +39,12 @@ class KnownFilesProvider(UrlProvider):
roots = _roots(self.ctx.hosts)
if not roots:
return
- limit = self.ctx.cfg.max_known_file_hosts
+ limit = MAX_KNOWN_FILE_HOSTS
selected = roots[:limit]
state = _State()
client = self._client()
try:
- workers = min(_MAX_WORKERS, len(selected))
+ workers = min(self.workers(_MAX_WORKERS), len(selected))
with ThreadPoolExecutor(max_workers=workers) as pool:
for found in pool.map(lambda root: self._mine(client, root), selected):
if found is None:
@@ -71,8 +72,8 @@ class KnownFilesProvider(UrlProvider):
headers = dict(self.ctx.net.headers or {})
headers.setdefault("User-Agent", "reNgine/3.0 (+https://rengine.wiki)")
return httpx.Client(
- timeout=self.ctx.cfg.timeout,
- follow_redirects=True,
+ timeout=self.ctx.transport.timeout,
+ follow_redirects=self.follow_redirects(True),
verify=False, # noqa: S501
proxy=self.ctx.net.proxy_url or None,
headers=headers,
@@ -117,6 +118,10 @@ class KnownFilesProvider(UrlProvider):
return state
def _get(self, client: httpx.Client, url: str, state: _State) -> str | None:
+ if self.path_excluded(url):
+ state.refused += 1
+ return None
+ self.throttle()
state.fetched += 1
body = bytearray()
try:
diff --git a/stages/url_discovery/providers/source_maps.py b/stages/url_discovery/providers/source_maps.py
index d8bbba989..1581f0243 100644
--- a/stages/url_discovery/providers/source_maps.py
+++ b/stages/url_discovery/providers/source_maps.py
@@ -11,6 +11,7 @@ from sqlalchemy import select
from shared.definitions.endpoints import EndpointSource
from shared.models.endpoint import Endpoint
from shared.services.endpoint_inventory import EndpointObservation
+from stages.url_discovery.config import MAX_SOURCE_MAPS
from stages.url_discovery.providers import mine
from stages.url_discovery.providers.base import ProviderResult, UrlProvider
@@ -36,11 +37,11 @@ class SourceMapProvider(UrlProvider):
self.progress("no javascript bundle answered")
return
- limit = self.ctx.cfg.max_source_maps
+ limit = MAX_SOURCE_MAPS
selected = bundles[:limit]
client = self._client()
try:
- workers = min(_MAX_WORKERS, len(selected))
+ workers = min(self.workers(_MAX_WORKERS), len(selected))
with ThreadPoolExecutor(max_workers=workers) as pool:
found = list(pool.map(lambda url: self._fetch(client, url), selected))
finally:
@@ -78,10 +79,11 @@ class SourceMapProvider(UrlProvider):
)
.order_by(Endpoint.url)
)
+ hosts = {h.host for h in self.ctx.hosts}
seen: dict[str, None] = {}
for url in rows:
base = _strip(url)
- if base:
+ if base and (urlsplit(base).hostname or "").lower() in hosts:
seen.setdefault(base, None)
return list(seen)
@@ -89,8 +91,8 @@ class SourceMapProvider(UrlProvider):
headers = dict(self.ctx.net.headers or {})
headers.setdefault("User-Agent", "reNgine/3.0 (+https://rengine.wiki)")
return httpx.Client(
- timeout=self.ctx.cfg.timeout,
- follow_redirects=True,
+ timeout=self.ctx.transport.timeout,
+ follow_redirects=self.follow_redirects(True),
verify=False, # noqa: S501
proxy=self.ctx.net.proxy_url or None,
headers=headers,
@@ -101,6 +103,9 @@ class SourceMapProvider(UrlProvider):
return None
url = f"{bundle}.map"
outcome = _Outcome(bundle=bundle, url=url)
+ if self.path_excluded(url):
+ return outcome
+ self.throttle()
try:
with client.stream("GET", url) as response:
if response.status_code != _OK:
diff --git a/stages/url_discovery/stage.py b/stages/url_discovery/stage.py
index d2393fc70..b642963f3 100644
--- a/stages/url_discovery/stage.py
+++ b/stages/url_discovery/stage.py
@@ -7,6 +7,7 @@ from sqlalchemy import func, select
from shared.definitions.domains import registrable_domain
from shared.definitions.endpoints import parse_url
+from shared.definitions.intensity import TransportTool
from shared.definitions.surface import SurfaceDimension
from shared.definitions.vulnerabilities import CoverageStatus
from shared.enums.scan import AssetKind, Intensity, Phase, StageGroup, StageRole
@@ -60,6 +61,7 @@ class UrlDiscoveryStage(Stage):
produces = frozenset({AssetKind.ENDPOINTS.value})
applies_to = ALL_TARGETS
tools = ("katana", "urlfinder")
+ transport_tool = TransportTool.KATANA.value
touches_target = False
config_model = UrlDiscoveryConfig
launch_fields = ("enabled", "providers", "crawl_depth", "max_crawl_minutes")
@@ -97,6 +99,7 @@ class UrlDiscoveryStage(Stage):
hosts=hosts,
apex_domains=self._apex_domains(hosts),
cfg=cfg,
+ transport=self.transport,
resolved=self.ctx.resolved,
net=self.net_options(),
recorder=self.ctx.recorder,
diff --git a/stages/vhost/config.py b/stages/vhost/config.py
index 502f9a660..0c1b6df5e 100644
--- a/stages/vhost/config.py
+++ b/stages/vhost/config.py
@@ -3,7 +3,7 @@ from __future__ import annotations
from pydantic import Field
from shared.definitions.wordlists import WordlistKind
-from stages.config import StageConfig, rate, threads, wordlist
+from stages.config import StageConfig, wordlist
class VhostConfig(StageConfig):
@@ -17,5 +17,3 @@ class VhostConfig(StageConfig):
title="Wordlist",
description="List of host names to try. Custom lists are uploaded in the Arsenal.",
)
- threads: int = threads(30, title="Threads")
- rate: int = rate(150, tool="ffuf", title="Requests/s")
diff --git a/stages/vhost/stage.py b/stages/vhost/stage.py
index b277c95fd..725d17d93 100644
--- a/stages/vhost/stage.py
+++ b/stages/vhost/stage.py
@@ -2,10 +2,12 @@ from __future__ import annotations
from sqlalchemy import select
+from shared.definitions.intensity import TransportTool
from shared.enums.scan import AssetKind, Phase, StageGroup, StageRole
from shared.enums.subdomain import SubdomainSource
from shared.logging import get_logger
from shared.models.subdomain import Subdomain
+from shared.services.scope_filter import matches_any
from shared.services.wordlists import WordlistError, lookup, resolve_path
from shared.utils.datetime import utc_now
from stages.base import DOMAIN_TARGETS, Stage, StageResult
@@ -31,6 +33,7 @@ class VhostStage(Stage):
produces = frozenset({AssetKind.HOSTS.value})
applies_to = DOMAIN_TARGETS
tools = ("ffuf",)
+ transport_tool = TransportTool.FFUF.value
config_model = VhostConfig
def run(self) -> StageResult:
@@ -66,8 +69,8 @@ class VhostStage(Stage):
try:
client = FfufClient(
wordlist=str(wordlist),
- threads=cfg.threads,
- rate=cfg.rate,
+ threads=self.transport.threads,
+ rate=self.transport.rate or 1,
proxy_url=net.proxy_url,
headers=net.headers,
probe_scheme=net.probe_scheme,
@@ -120,6 +123,7 @@ class VhostStage(Stage):
)
now = utc_now()
added = 0
+ excluded = self.ctx.resolved.excluded_subdomains or []
for name, ips in found.items():
if name in existing:
continue
@@ -134,7 +138,7 @@ class VhostStage(Stage):
cname=None,
is_active=True,
is_wildcard=False,
- is_excluded=False,
+ is_excluded=matches_any(name, excluded),
discovered_at=now,
)
)
diff --git a/stages/vulnerability_scan/config.py b/stages/vulnerability_scan/config.py
index 6fa3e8bde..8adb794c2 100644
--- a/stages/vulnerability_scan/config.py
+++ b/stages/vulnerability_scan/config.py
@@ -10,7 +10,6 @@ from shared.definitions.vulnerabilities import (
DEFAULT_SCANNERS,
DEFAULT_SEVERITIES,
DEFAULT_TEMPLATE_SETS,
- MAX_FINDINGS_PER_SCAN,
SEVERITY_LABELS,
SEVERITY_ORDER,
SURFACE_LABELS,
@@ -20,7 +19,7 @@ from shared.definitions.vulnerabilities import (
reject_unknown,
)
from shared.models.vuln_template import TemplateSelection
-from stages.config import StageConfig, rate, threads, timeout
+from stages.config import StageConfig, advanced
_SET_LABELS = {spec.key: spec.label for spec in TEMPLATE_SETS}
@@ -31,15 +30,6 @@ class VulnerabilityScanConfig(StageConfig):
title="Scan for vulnerabilities",
description="Run the selected checks against every asset this scan found.",
)
- scanners: list[str] = Field(
- default_factory=lambda: list(DEFAULT_SCANNERS),
- title="Scanners",
- description="Tools that produce findings.",
- json_schema_extra={
- "options": list(CENSUS_SCANNERS),
- "option_labels": dict(CENSUS_SCANNER_LABELS),
- },
- )
severities: list[str] = Field(
default_factory=lambda: list(DEFAULT_SEVERITIES),
title="Severities",
@@ -66,69 +56,16 @@ class VulnerabilityScanConfig(StageConfig):
description="Selected custom templates run regardless of set or severity.",
json_schema_extra={"widget": "custom_templates"},
)
- include_tags: list[str] = Field(
- default_factory=list,
- max_length=40,
- title="Extra tags",
- description="Checks carrying any of these tags are added to the selection.",
- )
- exclude_tags: list[str] = Field(
- default_factory=list,
- max_length=40,
- title="Excluded tags",
- description="Checks carrying any of these tags are removed from the selection.",
- )
- exclude_templates: list[str] = Field(
- default_factory=list,
- max_length=200,
- title="Excluded checks",
- description="Check identifiers excluded from every run.",
- )
surface: Surface = Field(
default=Surface.WEB,
title="Scan surface",
description="Network services adds every non-web port. Everything adds hostnames for DNS checks.",
json_schema_extra={"option_labels": dict(SURFACE_LABELS)},
)
- blind_sweep: bool = Field(
- default=True,
- title="Remaining checks",
- description="Run software-specific checks on origins where that software was not detected. Runs last.",
- )
- max_targets: int = Field(
- default=5000,
- ge=1,
- le=100000,
- title="Origin budget",
- description="Origins tested at most, highest ranked first.",
- )
- rate: int = rate(150, tool="nuclei", title="Request rate (rps)")
- threads: int = threads(
- 25,
- title="Concurrency",
- description="Checks running in parallel.",
- )
- bulk_size: int = Field(
- default=25,
- ge=1,
- le=500,
- title="Hosts per check",
- description="Hosts a single check is run against at once.",
- )
- timeout: int = timeout(10, title="Request timeout (s)")
- retries: int = Field(
- default=1,
- ge=0,
- le=5,
- title="Retries",
- description="Extra attempts per request.",
- )
- max_host_error: int = Field(
- default=30,
- ge=1,
- le=500,
- title="Host error budget",
- description="Errors a host may return before it is dropped. Dropped hosts are listed in coverage.",
+ headless: bool = Field(
+ default=False,
+ title="Browser checks",
+ description="Run checks that need a rendered page.",
)
max_minutes: int = Field(
default=90,
@@ -137,61 +74,64 @@ class VulnerabilityScanConfig(StageConfig):
title="Time budget (min)",
description="Stop starting batches after this many minutes. Origins not reached are recorded as not scanned. 0 means no limit.",
)
- waf_rate_divisor: int = Field(
- default=3,
+ scanners: list[str] = advanced(
+ None,
+ default_factory=lambda: list(DEFAULT_SCANNERS),
+ title="Scanners",
+ description="Tools that produce findings.",
+ json_schema_extra={
+ "options": list(CENSUS_SCANNERS),
+ "option_labels": dict(CENSUS_SCANNER_LABELS),
+ },
+ )
+ include_tags: list[str] = advanced(
+ None,
+ default_factory=list,
+ max_length=40,
+ title="Extra tags",
+ description="Checks carrying any of these tags are added to the selection.",
+ )
+ exclude_tags: list[str] = advanced(
+ None,
+ default_factory=list,
+ max_length=40,
+ title="Excluded tags",
+ description="Checks carrying any of these tags are removed from the selection.",
+ )
+ exclude_templates: list[str] = advanced(
+ None,
+ default_factory=list,
+ max_length=200,
+ title="Excluded checks",
+ description="Check identifiers excluded from every run.",
+ )
+ blind_sweep: bool = advanced(
+ True,
+ title="Remaining checks",
+ description="Run software-specific checks on origins where that software was not detected. Runs last.",
+ )
+ max_targets: int = advanced(
+ 5000,
ge=1,
- le=20,
- title="WAF rate divisor",
- description="Hosts behind a WAF or CDN are scanned at the request rate divided by this. 1 disables.",
+ le=100000,
+ title="Origin budget",
+ description="Origins tested at most, highest ranked first.",
)
- honeypot_threshold: int = Field(
- default=15,
- ge=0,
- le=200,
- title="Honeypot threshold",
- description="Discard a host matching more distinct checks than this. 0 disables.",
- )
- headless: bool = Field(
- default=False,
- title="Browser checks",
- description="Run checks that need a rendered page.",
- )
- interactsh: bool = Field(
- default=False,
+ interactsh: bool = advanced(
+ False,
title="Out-of-band testing",
description="Detect blind injection through callbacks to an external server. Uses the public interact.sh unless a self-hosted server is set.",
)
- interactsh_server: str = Field(
- default="",
+ interactsh_server: str = advanced(
+ "",
max_length=200,
title="Self-hosted OAST server",
- description="Empty uses the public server. The token is set under API keys.",
+ description="Empty uses the public server. The token is set under API keys. Fuzzing uses the same server.",
)
- oast_wait_minutes: int = Field(
- default=5,
- ge=0,
- le=120,
- title="Keep listening (minutes)",
- description="Minutes to wait for a callback after the scan.",
- )
- store_evidence: bool = Field(
- default=True,
+ store_evidence: bool = advanced(
+ True,
title="Store request and response",
- description="Keep the request and response that produced each finding.",
- )
- max_evidence_recovery: int = Field(
- default=100,
- ge=0,
- le=2000,
- title="Evidence recovery budget",
- description="Findings re-checked at one request per second to retain the exchange they matched on. Network and TLS checks report a match without one. 0 disables.",
- )
- max_findings: int = Field(
- default=MAX_FINDINGS_PER_SCAN,
- ge=1,
- le=MAX_FINDINGS_PER_SCAN,
- title="Finding budget",
- description="Stop recording after this many findings in one scan.",
+ description="Keep the exchange that produced each finding. Fuzzing follows the same setting.",
)
@field_validator("severities")
diff --git a/stages/vulnerability_scan/scanners/base.py b/stages/vulnerability_scan/scanners/base.py
index 2191e53c5..1d0d2bdb5 100644
--- a/stages/vulnerability_scan/scanners/base.py
+++ b/stages/vulnerability_scan/scanners/base.py
@@ -8,6 +8,7 @@ from dataclasses import dataclass, field
from datetime import datetime
from typing import TYPE_CHECKING, ClassVar
+from shared.definitions.intensity import Transport
from shared.definitions.vulnerabilities import CoverageStatus
from shared.utils.datetime import utc_now
from tools.runner import tool_path
@@ -24,6 +25,9 @@ if TYPE_CHECKING:
from tools.runner.models import CommandRecorder
+VULN_STAGE = "vulnerability_scan"
+
+
@dataclass
class ScannerContext:
session: Session
@@ -31,6 +35,7 @@ class ScannerContext:
target_id: uuid.UUID
project_id: uuid.UUID
cfg: VulnerabilityScanConfig
+ transport: Transport
resolved: ResolvedScanConfig
net: NetOptions
surface: SurfacePlan
@@ -62,6 +67,16 @@ class ScannerContext:
def aborted(self) -> bool:
return bool(self.is_aborted is not None and self.is_aborted())
+ def keep_evidence(self) -> bool:
+ """The vulnerability scan's evidence setting, shared by every scanner stage."""
+ own = getattr(self.cfg, "store_evidence", None)
+ if own is not None:
+ return bool(own)
+ resolved = self.resolved
+ if resolved is None or not hasattr(resolved, "stage"):
+ return True
+ return bool(resolved.stage(VULN_STAGE).get("store_evidence", True))
+
@dataclass
class Coverage:
diff --git a/stages/vulnerability_scan/scanners/nuclei.py b/stages/vulnerability_scan/scanners/nuclei.py
index 35a2e682f..8c43b3d2b 100644
--- a/stages/vulnerability_scan/scanners/nuclei.py
+++ b/stages/vulnerability_scan/scanners/nuclei.py
@@ -9,6 +9,7 @@ from collections.abc import Iterable
from concurrent.futures import ThreadPoolExecutor, wait
from dataclasses import dataclass, field
from pathlib import Path
+from typing import ClassVar
from shared.definitions.scan_surface import (
BATCH_OVERRUN,
@@ -49,6 +50,7 @@ from stages.base import StageAbortedError
from stages.sink import ResultSink
from stages.vulnerability_scan import evidence
from stages.vulnerability_scan.scanners.base import (
+ VULN_STAGE,
Coverage,
ScannerResult,
VulnScanner,
@@ -64,6 +66,10 @@ _DRAIN_SECONDS = 2.0
_STANDARD = "Standard rate"
_REDUCED = "Reduced rate · WAF or CDN"
_RECOVERY_RATE = 1
+BULK_SIZE = 25
+MAX_HOST_ERROR = 30
+WAF_RATE_DIVISOR = 3
+OAST_WAIT_SECONDS = 300
# a job needs at least this long to be worth starting
_MIN_JOB_SECONDS = 60
_BUDGET = "budget"
@@ -137,6 +143,9 @@ class NucleiScanner(VulnScanner):
name = Scanner.NUCLEI.value
label = SCANNER_LABELS[Scanner.NUCLEI.value]
binary = "nuclei"
+ honeypot_threshold: ClassVar[int] = 15
+ evidence_recovery: ClassVar[int] = 100
+ fuzz_param_frequency: ClassVar[int | None] = None
def run(self) -> ScannerResult:
ctx = self.ctx
@@ -174,7 +183,6 @@ class NucleiScanner(VulnScanner):
self, plan: SurfacePlan, tiers: TierPlan, library: _Templates
) -> dict[str, list[Job]]:
"""Every invocation, in the order the budget is spent, per rate lane."""
- cfg = self.ctx.cfg
rates = self._rates(plan)
lanes: dict[str, list[Job]] = {_STANDARD: [], _REDUCED: []}
counter = {_STANDARD: 0, _REDUCED: 0}
@@ -184,7 +192,7 @@ class NucleiScanner(VulnScanner):
if not files or not items:
return
per_host = cost(rows)
- for lane, wanted in _by_lane(items, cfg.waf_rate_divisor).items():
+ for lane, wanted in _by_lane(items, WAF_RATE_DIVISOR).items():
rate = rates[lane]
for chunk in batches(wanted, per_host, rate):
counter[lane] += 1
@@ -207,16 +215,22 @@ class NucleiScanner(VulnScanner):
_add(Tier.NAMES.value, plan.names, tiers.names, NAME_TYPES)
for group in tech_groups(plan.roots, tiers):
_add(Tier.MATCHED.value, group.items, group.rows)
- _add(Tier.BLIND.value, plan.roots, tiers.product)
+ if getattr(self.ctx.cfg, "blind_sweep", True):
+ _add(Tier.BLIND.value, plan.roots, tiers.product)
return {lane: jobs for lane, jobs in lanes.items() if jobs}
def _rates(self, plan: SurfacePlan) -> dict[str, int]:
"""The lane rates: one ceiling, split only when a guarded lane exists."""
- cfg = self.ctx.cfg
- guarded = cfg.waf_rate_divisor > 1 and any(
- item.guarded for item in (*plan.every_root, *plan.services, *plan.names)
+ scheduled = (
+ *plan.every_root,
+ *plan.services,
+ *plan.names,
+ *plan.requests,
+ *plan.bases,
)
- standard, reduced = _rate_plan(cfg.rate, cfg.waf_rate_divisor, guarded)
+ guarded = WAF_RATE_DIVISOR > 1 and any(item.guarded for item in scheduled)
+ rate = self.ctx.transport.rate or 1
+ standard, reduced = _rate_plan(rate, WAF_RATE_DIVISOR, guarded)
return {_STANDARD: standard, _REDUCED: reduced}
# ---------- lanes ----------
@@ -310,28 +324,39 @@ class NucleiScanner(VulnScanner):
return NucleiOptions(
templates_file=str(listing),
rate=job.rate,
- concurrency=cfg.threads,
- bulk_size=cfg.bulk_size,
- timeout=cfg.timeout,
- retries=cfg.retries,
- max_host_error=cfg.max_host_error,
+ concurrency=ctx.transport.threads,
+ bulk_size=BULK_SIZE,
+ timeout=ctx.transport.timeout,
+ retries=ctx.transport.retries,
+ max_host_error=MAX_HOST_ERROR,
max_minutes=minutes,
headless=cfg.headless,
interactsh=cfg.interactsh,
- interactsh_server=_oast_server(cfg.interactsh_server),
+ interactsh_server=_oast_server(self._oast_server_value(ctx)),
interactsh_token=self._oast_token(ctx) if cfg.interactsh else None,
- oast_wait_seconds=cfg.oast_wait_minutes * 60,
- honeypot_threshold=cfg.honeypot_threshold,
+ oast_wait_seconds=OAST_WAIT_SECONDS,
+ honeypot_threshold=self.honeypot_threshold,
proxy_url=ctx.net.proxy_url,
headers=dict(ctx.net.headers or {}),
exclude_hosts=tuple(ctx.resolved.excluded_subdomains or ()),
follow_redirects=ctx.resolved.follow_redirects,
protocol_types=job.types,
dast=job.dast,
- fuzz_param_frequency=getattr(cfg, "fuzz_param_frequency", None),
+ fuzz_param_frequency=self.fuzz_param_frequency,
extra_args=ctx.resolved.tool_args("nuclei"),
)
+ @staticmethod
+ def _oast_server_value(ctx) -> str | None:
+ """The vulnerability scan's server setting, shared by every nuclei stage."""
+ own = getattr(ctx.cfg, "interactsh_server", None)
+ if own:
+ return own
+ resolved = getattr(ctx, "resolved", None)
+ if resolved is None or not hasattr(resolved, "stage"):
+ return None
+ return resolved.stage(VULN_STAGE).get("interactsh_server")
+
@staticmethod
def _oast_token(ctx) -> str | None:
"""The self-hosted server's token."""
@@ -357,7 +382,7 @@ class NucleiScanner(VulnScanner):
custom_templates=library.custom,
hosts_total=len(job.items),
rate_limit=job.rate,
- concurrency=cfg.threads,
+ concurrency=ctx.transport.threads,
)
listing = write_template_list(job.templates)
try:
@@ -384,7 +409,7 @@ class NucleiScanner(VulnScanner):
sink = ResultSink(_write, rows=_FLUSH_EVERY, seconds=_FLUSH_SECONDS)
def _on_finding(finding: Finding) -> None:
- if cfg.store_evidence:
+ if ctx.keep_evidence():
finding.request = redact_secrets(finding.request, secrets)
finding.response = redact_secrets(finding.response, secrets)
else:
@@ -516,9 +541,9 @@ class NucleiScanner(VulnScanner):
def _recover_evidence(self) -> None:
"""Re-run the checks whose match nuclei reported without an exchange."""
ctx = self.ctx
- if not ctx.cfg.store_evidence or ctx.aborted():
+ if not self.evidence_recovery or not ctx.keep_evidence() or ctx.aborted():
return
- gaps = evidence.pending(ctx.session, ctx.scan_id, ctx.cfg.max_evidence_recovery)
+ gaps = evidence.pending(ctx.session, ctx.scan_id, self.evidence_recovery)
if not gaps:
return
covered = sum(len(gap.fingerprints) for gap in gaps)
diff --git a/stages/vulnerability_scan/stage.py b/stages/vulnerability_scan/stage.py
index b8f19b2b6..9e37632f7 100644
--- a/stages/vulnerability_scan/stage.py
+++ b/stages/vulnerability_scan/stage.py
@@ -2,9 +2,11 @@ from __future__ import annotations
import uuid
+from shared.definitions.intensity import TransportTool
from shared.definitions.scan_surface import ROOT_TIERS
from shared.definitions.surface import SurfaceDimension
from shared.definitions.vulnerabilities import (
+ MAX_FINDINGS_PER_SCAN,
SEVERITY_LABELS,
CoverageStatus,
Severity,
@@ -61,6 +63,7 @@ class VulnerabilityScanStage(Stage):
produces = frozenset({AssetKind.VULNERABILITIES.value})
applies_to = ALL_TARGETS
tools = ("nuclei",)
+ transport_tool = TransportTool.NUCLEI.value
touches_target = True
config_model = VulnerabilityScanConfig
launch_fields = (
@@ -92,7 +95,7 @@ class VulnerabilityScanStage(Stage):
index = vuln_inventory.build_index(self.session, self.ctx.scan_id)
stored: list[str] = []
- budget = cfg.max_findings
+ budget = MAX_FINDINGS_PER_SCAN
def _store(findings: list) -> int:
self._check_abort()
@@ -134,6 +137,7 @@ class VulnerabilityScanStage(Stage):
target_id=self.ctx.target_id,
project_id=self.ctx.project_id,
cfg=cfg,
+ transport=self.transport,
resolved=self.ctx.resolved,
net=self.net_options(),
surface=plan,
diff --git a/tests/pipeline/test_dast_surface.py b/tests/pipeline/test_dast_surface.py
index 617c5d498..c53ce2e98 100644
--- a/tests/pipeline/test_dast_surface.py
+++ b/tests/pipeline/test_dast_surface.py
@@ -6,12 +6,13 @@ from types import SimpleNamespace
import pytest
+from shared.definitions.intensity import Transport
from shared.definitions.scan_surface import Tier
from shared.services.scan_surface import SurfaceItem, SurfacePlan, by_origin
from shared.services.scan_surface.requests import _better, _rewrite
from stages.dast_scan.config import DastScanConfig
from stages.dast_scan.scanners import scanners
-from stages.dast_scan.scanners.nuclei import NucleiDastScanner
+from stages.dast_scan.scanners.nuclei import FUZZ_PARAM_FREQUENCY, NucleiDastScanner
from stages.registry import execution_plan, stage_by_name
from stages.vulnerability_scan.scanners.base import ScannerContext
from stages.vulnerability_scan.scanners.nuclei import Job, _Templates
@@ -42,6 +43,7 @@ def _scanner(**cfg) -> NucleiDastScanner:
target_id=uuid.uuid4(),
project_id=uuid.uuid4(),
cfg=DastScanConfig(**cfg),
+ transport=Transport(tool="nuclei", rate=150, threads=25, timeout=10, retries=1),
resolved=None,
net=None,
surface=SurfacePlan(),
@@ -156,7 +158,7 @@ def test_a_fuzzing_finding_replays_the_same_request_on_each_equivalent():
def test_a_fuzzing_job_passes_dast_and_the_parameter_patience():
- scanner = _scanner(fuzz_param_frequency=7)
+ scanner = _scanner()
scanner.ctx.net = SimpleNamespace(proxy_url=None, headers={})
scanner.ctx.resolved = SimpleNamespace(
follow_redirects=None, tool_args=lambda _t: [], excluded_subdomains=[]
@@ -173,4 +175,4 @@ def test_a_fuzzing_job_passes_dast_and_the_parameter_patience():
options = scanner._options(job, Path("t.txt"), None)
assert options.dast is True
- assert options.fuzz_param_frequency == 7
+ assert options.fuzz_param_frequency == FUZZ_PARAM_FREQUENCY
diff --git a/tests/pipeline/test_default_engine.py b/tests/pipeline/test_default_engine.py
new file mode 100644
index 000000000..92f9cb1a2
--- /dev/null
+++ b/tests/pipeline/test_default_engine.py
@@ -0,0 +1,68 @@
+"""The built-in engine resolves and runs subdomain discovery on subfinder alone."""
+
+from __future__ import annotations
+
+from typing import ClassVar
+
+import pytest
+
+from shared.definitions.default_engine import (
+ DEFAULT_ENGINE_INTENSITY,
+ DEFAULT_ENGINE_NAME,
+ DEFAULT_PASSIVE_SOURCES,
+ DEFAULT_VULN_SCANNERS,
+ SUBDOMAIN_STAGE,
+ VULNERABILITY_STAGE,
+ default_engine_stages,
+)
+from shared.services.scan_resolve import merge_engine_context
+from stages.registry import stage_by_name
+from stages.subdomain.config import SubdomainConfig
+from stages.vulnerability_scan.config import VulnerabilityScanConfig
+
+pytestmark = pytest.mark.pipeline
+
+
+class _Builtin:
+ name = DEFAULT_ENGINE_NAME
+ intensity = DEFAULT_ENGINE_INTENSITY
+ global_headers: ClassVar[list] = []
+ stages: ClassVar[dict] = default_engine_stages()
+ tool_options: ClassVar[dict] = {}
+
+
+_DELTA = {
+ SUBDOMAIN_STAGE: {"passive_tools"},
+ VULNERABILITY_STAGE: {"enabled", "scanners"},
+}
+
+
+def test_the_delta_is_the_sources_and_the_scanner():
+ assert default_engine_stages() == {
+ SUBDOMAIN_STAGE: {"passive_tools": list(DEFAULT_PASSIVE_SOURCES)},
+ VULNERABILITY_STAGE: {"enabled": True, "scanners": list(DEFAULT_VULN_SCANNERS)},
+ }
+ assert SubdomainConfig(
+ **default_engine_stages()[SUBDOMAIN_STAGE]
+ ).enabled_sources == ["subfinder"]
+ assert VulnerabilityScanConfig(
+ **default_engine_stages()[VULNERABILITY_STAGE]
+ ).enabled
+
+
+def test_every_other_stage_runs_at_its_default():
+ resolved = merge_engine_context(_Builtin(), None, "example.com", "domain")
+ for spec in stage_by_name().values():
+ if spec.catalog_hidden:
+ continue
+ expected = spec.defaults
+ got = resolved.stage(spec.name)
+ for key, value in expected.items():
+ if key in _DELTA.get(spec.name, set()):
+ continue
+ assert got[key] == value, f"{spec.name}.{key}"
+ assert resolved.stage(SUBDOMAIN_STAGE)["passive_tools"] == ["subfinder"]
+ assert resolved.stage(VULNERABILITY_STAGE)["enabled"] is True
+ assert resolved.stage(VULNERABILITY_STAGE)["scanners"] == ["nuclei"]
+ assert resolved.stage("dast_scan")["enabled"] is False
+ assert resolved.intensity == DEFAULT_ENGINE_INTENSITY
diff --git a/tests/pipeline/test_nuclei_groups.py b/tests/pipeline/test_nuclei_groups.py
index 4e4371e12..147456de2 100644
--- a/tests/pipeline/test_nuclei_groups.py
+++ b/tests/pipeline/test_nuclei_groups.py
@@ -7,10 +7,12 @@ from types import SimpleNamespace
import pytest
+from shared.definitions.intensity import Transport
from shared.definitions.scan_surface import Tier
from shared.definitions.vulnerabilities import Scanner
from shared.services.scan_surface import SurfaceItem, SurfacePlan, split
from stages.vulnerability_scan.config import VulnerabilityScanConfig
+from stages.vulnerability_scan.scanners import nuclei as nuclei_scanner
from stages.vulnerability_scan.scanners.base import (
Coverage,
ScannerContext,
@@ -63,6 +65,7 @@ def _scanner(writes: _Writes, **cfg) -> NucleiScanner:
target_id=uuid.uuid4(),
project_id=uuid.uuid4(),
cfg=VulnerabilityScanConfig(**cfg),
+ transport=Transport(tool="nuclei", rate=150, threads=25, timeout=10, retries=1),
resolved=None,
net=None,
surface=SurfacePlan(),
@@ -202,9 +205,10 @@ def test_guarded_items_take_the_reduced_lane_only_with_a_divisor():
assert len(single[STANDARD]) == 2
-def test_the_schedule_spends_the_budget_in_tier_order():
+def test_the_schedule_spends_the_budget_in_tier_order(monkeypatch):
writes = _Writes()
- scanner = _scanner(writes, waf_rate_divisor=1)
+ monkeypatch.setattr(nuclei_scanner, "WAF_RATE_DIVISOR", 1)
+ scanner = _scanner(writes)
rep = _item("https://rep.example", members=2)
rep.tags = ["geoserver"]
plan = SurfacePlan(roots=[rep])
diff --git a/tests/pipeline/test_oast.py b/tests/pipeline/test_oast.py
index 5d9ae7d60..58df952ec 100644
--- a/tests/pipeline/test_oast.py
+++ b/tests/pipeline/test_oast.py
@@ -5,11 +5,16 @@ from types import SimpleNamespace
import pytest
+from shared.definitions.intensity import Transport
from shared.enums.api_key import APIProvider
from shared.models.api_key import API_PROVIDER_META
from shared.services.scan_resolve import MASK, redact_command
-from stages.vulnerability_scan.config import VulnerabilityScanConfig
-from stages.vulnerability_scan.scanners.nuclei import Job, NucleiScanner, _oast_server
+from stages.vulnerability_scan.scanners.nuclei import (
+ OAST_WAIT_SECONDS,
+ Job,
+ NucleiScanner,
+ _oast_server,
+)
from tools.nuclei.client import _EVICTION_SLACK, NucleiClient, NucleiOptions
pytestmark = pytest.mark.pipeline
@@ -49,7 +54,7 @@ def test_no_window_leaves_nuclei_on_its_own_defaults():
def test_the_default_waits_minutes_not_seconds():
- assert VulnerabilityScanConfig().oast_wait_minutes >= 1
+ assert OAST_WAIT_SECONDS >= 60
def test_the_token_reaches_nuclei():
@@ -77,18 +82,12 @@ def test_the_token_is_not_even_read_when_oast_is_off():
scanner = NucleiScanner.__new__(NucleiScanner)
scanner.ctx = SimpleNamespace(
cfg=SimpleNamespace(
- threads=1,
- bulk_size=1,
- timeout=1,
- retries=0,
- max_host_error=1,
max_minutes=0,
headless=False,
interactsh=False,
interactsh_server="",
- oast_wait_minutes=5,
- honeypot_threshold=0,
),
+ transport=Transport(tool="nuclei", rate=150, threads=25, timeout=10, retries=1),
net=SimpleNamespace(proxy_url=None, headers={}),
resolved=SimpleNamespace(
follow_redirects=None, tool_args=lambda _t: [], excluded_subdomains=[]
diff --git a/tests/pipeline/test_stage_knobs.py b/tests/pipeline/test_stage_knobs.py
new file mode 100644
index 000000000..9d55879ef
--- /dev/null
+++ b/tests/pipeline/test_stage_knobs.py
@@ -0,0 +1,191 @@
+"""Every engine field is a decision a stage reads; transport comes from intensity."""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+from typing import ClassVar
+
+import pytest
+
+from shared.definitions.intensity import (
+ PROFILES,
+ RATE_TOOLS,
+ TOOL_TIMEOUT,
+ Transport,
+ transport_for,
+)
+from shared.enums.scan import Intensity, StageRole
+from shared.services.scan_resolve import merge_engine_context
+from stages.config import FieldTier
+from stages.registry import rate_tools, stages
+
+pytestmark = pytest.mark.pipeline
+
+_ROOT = Path(__file__).resolve().parents[2]
+_NO_TRANSPORT = {"waf_detect"}
+_TRANSPORT_NAMES = {
+ "rate",
+ "threads",
+ "timeout",
+ "retries",
+ "dns_threads",
+ "dns_timeout",
+}
+_READ_ELSEWHERE = {
+ ("dast_scan", "interactsh"),
+ ("url_discovery", "static_extensions"),
+ ("url_discovery", "ignored_params"),
+ ("url_discovery", "keep_per_family"),
+ ("url_discovery", "sibling_cap"),
+ ("url_discovery", "drop_noise"),
+}
+
+
+def _package_source(spec) -> str:
+ package = Path(spec.stage_cls.__module__.replace(".", "/")).parent
+ files = (_ROOT / package).rglob("*.py")
+ return "\n".join(p.read_text() for p in files if p.name != "config.py")
+
+
+def _config_source(spec) -> str:
+ return (
+ _ROOT / Path(spec.config_model.__module__.replace(".", "/") + ".py")
+ ).read_text()
+
+
+@pytest.mark.parametrize("spec", stages(), ids=lambda s: s.name)
+def test_every_field_is_read_by_its_stage(spec):
+ body = _package_source(spec)
+ own = _config_source(spec)
+ for name in spec.config_model.model_fields:
+ if name == "enabled" or (spec.name, name) in _READ_ELSEWHERE:
+ continue
+ declared = len(re.findall(rf"^ {name}:", own, re.M))
+ used_in_config = len(re.findall(rf"\b{name}\b", own)) - declared
+ used = re.search(rf"\.{name}\b|\"{name}\"", body) or used_in_config > 0
+ assert used, f"{spec.name}.{name} is on the form and read by nothing"
+
+
+@pytest.mark.parametrize("spec", stages(), ids=lambda s: s.name)
+def test_no_stage_carries_a_transport_field(spec):
+ fields = set(spec.config_model.model_fields)
+ assert not (fields & _TRANSPORT_NAMES), (
+ f"{spec.name} declares {fields & _TRANSPORT_NAMES}"
+ )
+
+
+@pytest.mark.parametrize("spec", stages(), ids=lambda s: s.name)
+def test_every_stage_that_sends_traffic_declares_its_transport(spec):
+ if spec.touches_target and spec.tools and spec.name not in _NO_TRANSPORT:
+ assert spec.transport_tool is not None, spec.name
+
+
+@pytest.mark.parametrize("spec", stages(), ids=lambda s: s.name)
+def test_tiers_are_declared_once_per_field(spec):
+ tiers = spec.config_model.tiers()
+ assert set(tiers.values()) <= {FieldTier.BASIC.value, FieldTier.ADVANCED.value}
+ assert tiers["enabled"] == FieldTier.BASIC.value
+
+
+def test_rate_tools_are_the_tools_with_a_rate():
+ assert set(rate_tools()) <= set(RATE_TOOLS)
+ assert "dnsx" not in rate_tools()
+
+
+@pytest.mark.parametrize(
+ ("tool", "rate", "threads"),
+ [
+ ("httpx", 150, 150),
+ ("naabu", 1000, 100),
+ ("nuclei", 150, 25),
+ ("katana", 150, 50),
+ ("ffuf", 150, 40),
+ ("dnsx", None, 30),
+ ("banner", None, 32),
+ ],
+)
+def test_normal_keeps_the_numbers_every_stage_shipped_with(tool, rate, threads):
+ assert PROFILES[Intensity.NORMAL.value][tool] == (rate, threads)
+ assert TOOL_TIMEOUT[tool] >= 1
+
+
+def test_aggressive_is_louder_than_normal_for_every_rated_tool():
+ for tool in RATE_TOOLS:
+ normal = PROFILES[Intensity.NORMAL.value][tool]
+ loud = PROFILES[Intensity.AGGRESSIVE.value][tool]
+ assert loud[0] > normal[0], tool
+ assert loud[1] > normal[1], tool
+
+
+def test_a_weight_scales_the_profile_not_the_ceiling():
+ third = transport_for("nuclei", Intensity.NORMAL.value, rate_weight=1 / 3)
+ assert third.rate == 50
+ capped = transport_for(
+ "nuclei", Intensity.AGGRESSIVE.value, rate_weight=1 / 3, ceiling=30
+ )
+ assert capped.rate == 10
+
+
+def test_the_context_multipliers_and_ceiling_still_apply():
+ t = transport_for(
+ "httpx",
+ Intensity.NORMAL.value,
+ thread_multiplier=2.0,
+ timeout_multiplier=3.0,
+ rate_override=40,
+ )
+ assert t == Transport(tool="httpx", rate=40, threads=300, timeout=30, retries=0)
+
+
+class _Engine:
+ intensity = Intensity.NORMAL.value
+ global_headers: ClassVar[list] = []
+ stages: ClassVar[dict] = {}
+ tool_options: ClassVar[dict] = {}
+
+
+class _Context:
+ auth: ClassVar[dict] = {"auth_type": "none"}
+ extra_headers: ClassVar[list] = []
+ global_rate_limit_override = 20
+ per_tool_rate_overrides: ClassVar[dict] = {"naabu": 500}
+ thread_multiplier = 1.0
+ timeout_multiplier = 1.0
+ excluded_subdomains: ClassVar[list] = []
+ excluded_paths: ClassVar[list] = []
+ excluded_ips: ClassVar[list] = []
+ included_subdomains: ClassVar[list] = []
+ follow_redirects_override = None
+ http_protocol = "both"
+
+
+def test_the_resolver_hands_every_tool_stage_a_transport():
+ resolved = merge_engine_context(_Engine(), _Context(), "example.com", "domain")
+ for spec in stages():
+ if spec.transport_tool is None:
+ assert spec.name not in resolved.transports
+ continue
+ transport = Transport(**resolved.transports[spec.name])
+ assert transport.tool == spec.transport_tool
+ if transport.rate is not None:
+ assert transport.rate <= 20
+ assert resolved.per_tool_rate_limits["naabu"] == 20
+ assert resolved.per_tool_rate_limits["httpx"] == 20
+
+
+def test_aggressive_moves_the_footprint():
+ normal = merge_engine_context(_Engine(), None, "example.com", "domain")
+ loud = merge_engine_context(
+ _Engine(), None, "example.com", "domain", intensity=Intensity.AGGRESSIVE.value
+ )
+ assert (
+ loud.transports["http_probe"]["rate"] > normal.transports["http_probe"]["rate"]
+ )
+ assert loud.transports["port_scan"]["rate"] > normal.transports["port_scan"]["rate"]
+
+
+def test_support_stages_are_never_launch_decisions():
+ for spec in stages():
+ if spec.role == StageRole.SUPPORT.value:
+ assert "enabled" in spec.config_model.model_fields
diff --git a/tests/pipeline/test_tool_args_precedence.py b/tests/pipeline/test_tool_args_precedence.py
new file mode 100644
index 000000000..ceee7d188
--- /dev/null
+++ b/tests/pipeline/test_tool_args_precedence.py
@@ -0,0 +1,35 @@
+"""Tool args add flags; they never restate one the stage set."""
+
+from __future__ import annotations
+
+import pytest
+
+from tools.runner.executor import merge_extra_args
+
+pytestmark = pytest.mark.pipeline
+
+
+def test_a_restated_flag_and_its_value_are_dropped():
+ stage = ["-rate-limit", "5", "-threads", "20"]
+ extra = ["-rate-limit", "9999", "-retries", "3"]
+ assert merge_extra_args(stage, extra) == ["-retries", "3"]
+
+
+def test_the_equals_spelling_is_dropped_too():
+ assert merge_extra_args(["-timeout", "10"], ["-timeout=99", "-silent"]) == [
+ "-silent"
+ ]
+
+
+def test_reserved_flags_are_dropped_before_the_runner_adds_them():
+ extra = ["-l", "/etc/passwd", "-json", "-o", "elsewhere.txt", "-stats"]
+ assert merge_extra_args([], extra, ("-l", "-json", "-o")) == ["-stats"]
+
+
+def test_a_boolean_flag_does_not_swallow_the_next_flag():
+ assert merge_extra_args(["-silent"], ["-silent", "-stats"]) == ["-stats"]
+
+
+def test_unrelated_flags_pass_through_in_order():
+ extra = ["-stats", "-si", "5", "-tags", "cve"]
+ assert merge_extra_args(["-rate-limit", "5"], extra) == extra
diff --git a/tools/runner/executor.py b/tools/runner/executor.py
index edf29f116..bbb8d6257 100644
--- a/tools/runner/executor.py
+++ b/tools/runner/executor.py
@@ -47,6 +47,33 @@ def _terminate(proc: subprocess.Popen) -> None:
proc.wait(timeout=_KILL_GRACE_SECONDS)
+def _flag_name(token: str) -> str:
+ return token.split("=", 1)[0]
+
+
+def merge_extra_args(
+ args: list[str], extra: list[str], reserved: tuple[str, ...] = ()
+) -> list[str]:
+ """Tool args, minus any flag the stage already set."""
+ taken = {_flag_name(a) for a in args if a.startswith("-")} | set(reserved)
+ out: list[str] = []
+ skip = False
+ for index, token in enumerate(extra):
+ if skip:
+ skip = False
+ continue
+ if token.startswith("-") and _flag_name(token) in taken:
+ following = extra[index + 1] if index + 1 < len(extra) else None
+ skip = (
+ "=" not in token
+ and following is not None
+ and not following.startswith("-")
+ )
+ continue
+ out.append(token)
+ return out
+
+
def _drain(stream, into: list[str]) -> None:
with contextlib.suppress(Exception):
into.append(stream.read())
@@ -208,7 +235,13 @@ class CLIToolRunner:
args = list(args) if args else []
recorder = recorder if recorder is not None else self._recorder
tool = tool if tool is not None else self._tool
- args.extend(extra_args if extra_args is not None else self._extra_args)
+ args.extend(
+ merge_extra_args(
+ args,
+ extra_args if extra_args is not None else self._extra_args,
+ (input_flag, output_flag, json_flag, silent_flag),
+ )
+ )
input_file: Path | None = None
output_file: Path | None = None
@@ -366,7 +399,13 @@ class CLIToolRunner:
args = list(args) if args else []
recorder = recorder if recorder is not None else self._recorder
tool = tool if tool is not None else self._tool
- args.extend(extra_args if extra_args is not None else self._extra_args)
+ args.extend(
+ merge_extra_args(
+ args,
+ extra_args if extra_args is not None else self._extra_args,
+ (input_flag, json_flag, silent_flag),
+ )
+ )
input_file: Path | None = None
start_time = time.monotonic()
diff --git a/worker/app/tasks/endpoints.py b/worker/app/tasks/endpoints.py
index 99cdf6a03..3bf831a02 100644
--- a/worker/app/tasks/endpoints.py
+++ b/worker/app/tasks/endpoints.py
@@ -11,6 +11,7 @@ from sqlalchemy.dialects.postgresql import JSONB
from app.database import get_sync_session
from app.orchestrator.stage import load_resolved
from shared.definitions.endpoints import PROBE_COVERAGE_SOURCE, STATIC_CLASSES
+from shared.definitions.intensity import Transport
from shared.definitions.vulnerabilities import CoverageStatus
from shared.logging import get_logger
from shared.models.endpoint import Endpoint, EndpointCoverage
@@ -21,7 +22,8 @@ from shared.services.asset_query.lead_cache import bump_sync
from shared.services.endpoint_inventory import EndpointObservation
from shared.services.scope_filter import matches_any
from shared.utils.datetime import utc_now
-from stages.endpoint_probe.config import EndpointProbeConfig
+from stages.endpoint_probe.config import FOLLOW_REDIRECTS
+from stages.endpoint_probe.stage import EndpointProbeStage
from tools.httpx.client import HttpxClient, HttpxError
from tools.httpx.parser import parse_httpx_record
@@ -66,7 +68,12 @@ def verify_branch(
if scan is None:
return {"error": "scan not found"}
resolved = load_resolved(scan.execution_config)
- cfg = EndpointProbeConfig()
+ stored = (resolved.transports or {}).get(EndpointProbeStage.name)
+ transport = (
+ Transport(**stored)
+ if stored
+ else EndpointProbeStage.default_transport(resolved.intensity)
+ )
candidates = branch_candidates(scan.id, host, dir_path)
excluded = resolved.excluded_paths or []
if excluded:
@@ -84,13 +91,13 @@ def verify_branch(
try:
client = HttpxClient(
- rate_limit=cfg.rate,
- threads=cfg.threads,
- timeout=cfg.timeout,
+ rate_limit=transport.rate,
+ threads=transport.threads,
+ timeout=transport.timeout,
proxy_url=resolved.proxy_url,
headers=dict(resolved.headers or {}),
probe_scheme=PROBE_SCHEME.get(resolved.http_protocol),
- follow_redirects=cfg.follow_redirects,
+ follow_redirects=FOLLOW_REDIRECTS,
extra_args=resolved.tool_args("httpx"),
)
except HttpxError as e: