scriptoriumadmin/strix
1
0
Fork 0
mirror of https://github.com/usestrix/strix.git synced 2026-10-16 04:28:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

feat(agent): add Target Complexity Index (TCI) + adaptive ScanPlanner for dynamic vulnerability scanning (#46)

Browse source 
- Implement Target Complexity Index (TCI) calculation (0..100) in strix.core.tci based on target fingerprint with configurable weights.
- Implement ScanModule and ModuleRegistry in strix.modules.registry supporting priority metadata, quotas, timeouts, and safe_mode hints.
- Implement adaptive ScanPlanner in strix.agents.planner mapping TCI and fingerprints to prioritized, context-aware scan plans.
- Add planner agent tools in strix.tools.planner.tools and register in factory._BASE_TOOLS.
- Update root_agent.md coordination documentation for dynamic scan planning.
- Add comprehensive unit test suites in tests/test_tci.py, tests/test_module_registry.py, tests/test_scan_planner.py, and tests/test_planner_tools.py.

Resolves usestrix/strix#46
This commit is contained in:
MriBoyd 2026-10-14 11:12:00 +03:00
parent 956a26715b
commit 64622e8d7e
14 changed files with 2384 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
strix/agents/__init__.py
View file

@ -0,0 +1,16 @@
"""Strix agent definitions and planning."""
from __future__ import annotations
from strix.agents.planner import (
ScanPlan,
ScanPlanner,
ScanPlanStep,
)
__all__ = [
"ScanPlan",
"ScanPlanStep",
"ScanPlanner",
]

8
strix/agents/factory.py
View file

@ -45,6 +45,11 @@ from strix.tools.notes.tools import (
)
from strix.tools.nullish import is_nullish
from strix.tools.output_store import bound_and_store, bound_text
from strix.tools.planner.tools import (
calculate_target_complexity,
create_scan_plan,
list_scan_modules,
)
from strix.tools.proxy.tools import (
list_requests,
list_sitemap,
@ -591,6 +596,9 @@ _BASE_TOOLS: tuple[Tool, ...] = (
get_threat_model,
save_threat_model,
amend_threat_model,
calculate_target_complexity,
create_scan_plan,
list_scan_modules,
web_search,
web_get_contents,
create_vulnerability_report,

382
strix/agents/planner.py Normal file
View file

@ -0,0 +1,382 @@
"""Adaptive ScanPlanner: Maps TCI and fingerprints to dynamic scan plans."""
from __future__ import annotations
import json
import logging
from typing import Any
from pydantic import BaseModel, ConfigDict, Field
from strix.core.tci import TargetFingerprint, calculate_tci
from strix.modules.registry import ModuleRegistry, PriorityTier, ScanModule, default_registry
logger = logging.getLogger(__name__)
# Phase weights for tactical ordering
_PHASE_ORDER: dict[str, int] = {
"recon": 0,
"authentication": 1,
"authorization": 2,
"injection": 3,
"api": 3,
"logic": 4,
"infrastructure": 5,
}
class ScanPlanStep(BaseModel):
"""An individual step in a dynamic scan plan."""
model_config = ConfigDict(extra="allow")
step: int = Field(ge=1)
module: str
priority: str
safe_mode: bool
timeout_s: float = 300.0
quota: int = 50
skills: list[str] = Field(default_factory=list)
reason: str = ""
def to_dict(self, *, compact: bool = True) -> dict[str, Any]:
"""Convert step to dictionary.
When compact=True (default), outputs the standard wire format:
{"step": 1, "module": "auth-enum", "priority": "high", "safe_mode": true}
"""
payload: dict[str, Any] = {
"step": self.step,
"module": self.module,
"priority": self.priority,
"safe_mode": self.safe_mode,
}
if not compact:
payload.update(
{
"timeout_s": self.timeout_s,
"quota": self.quota,
"skills": self.skills,
"reason": self.reason,
}
)
return payload
class ScanPlan(BaseModel):
"""Context-aware scan plan generated from target fingerprint and TCI."""
model_config = ConfigDict(extra="allow")
target: str
fingerprint: dict[str, Any]
tci: float
tci_level: str
plan: list[ScanPlanStep]
metadata: dict[str, Any] = Field(default_factory=dict)
def to_dict(self, *, compact: bool = True) -> dict[str, Any]:
"""Convert plan to serializable dictionary."""
return {
"target": self.target,
"fingerprint": self.fingerprint,
"tci": self.tci,
"plan": [step.to_dict(compact=compact) for step in self.plan],
}
def to_json(self, *, compact: bool = True, indent: int | None = 2) -> str:
"""Convert plan to JSON string."""
return json.dumps(self.to_dict(compact=compact), indent=indent, default=str)
def get_step(self, step_number: int) -> ScanPlanStep | None:
"""Retrieve step by 1-based step index."""
for s in self.plan:
if s.step == step_number:
return s
return None
def get_step_by_module(self, module_name: str) -> ScanPlanStep | None:
"""Retrieve step by module name."""
clean = module_name.lower().strip()
for s in self.plan:
if s.module.lower() == clean:
return s
return None
def get_steps_by_priority(self, priority: str) -> list[ScanPlanStep]:
"""Retrieve steps matching a priority tier."""
clean = priority.lower().strip()
return [s for s in self.plan if s.priority.lower() == clean]
def module_names(self) -> list[str]:
"""Return ordered list of module names."""
return [s.module for s in self.plan]
def total_quota(self) -> int:
"""Sum total quota allocated across all steps."""
return sum(s.quota for s in self.plan)
def estimated_duration_s(self) -> float:
"""Sum total duration estimate across all steps."""
return sum(s.timeout_s for s in self.plan)
class ScanPlanner:
"""Generates context-aware, tactical scan plans from Target Complexity Index."""
def __init__(
self,
registry: ModuleRegistry | None = None,
) -> None:
self.registry = registry or default_registry
def generate_plan(
self,
target: str,
fingerprint: TargetFingerprint | dict[str, Any],
*,
scan_mode: str = "deep",
safe_mode_override: bool | None = None,
priority_filter: list[str] | None = None,
max_steps: int | None = None,
module_overrides: dict[str, dict[str, Any]] | None = None,
) -> ScanPlan:
"""Generate an adaptive scan plan.
Args:
target: Target URL, hostname, or checkout path.
fingerprint: Target observables (ports, auth, api, waf, etc.).
scan_mode: Scan mode ('quick', 'standard', 'deep', 'diff').
safe_mode_override: Explicit safe_mode flag for all steps.
priority_filter: Allowed priority tiers (e.g. ['critical', 'high']).
max_steps: Maximum number of steps to schedule.
module_overrides: Per-module overrides for priority, safe_mode, quota, timeout_s.
Returns:
Structured ScanPlan with ordered, prioritized steps.
"""
fp = (
fingerprint
if isinstance(fingerprint, TargetFingerprint)
else TargetFingerprint(**fingerprint)
)
if not fp.target and target:
fp.target = target
# 1. Compute Target Complexity Index
tci_res = calculate_tci(fp)
tci_score = tci_res.score
tci_level = tci_res.level.value
# 2. Select applicable modules based on target context
candidates = self.registry.find_applicable(fp)
# 3. Filter candidates by scan_mode
candidates = self._filter_by_scan_mode(candidates, scan_mode)
# 4. Filter by priority if requested
if priority_filter:
allowed_p = {p.lower().strip() for p in priority_filter}
candidates = [m for m in candidates if m.default_priority.value.lower() in allowed_p]
# 5. Build and adapt plan steps
steps = self._build_steps(
candidates=candidates,
fp=fp,
tci_score=tci_score,
safe_mode_override=safe_mode_override,
module_overrides=module_overrides or {},
)
# 6. Apply max_steps limit if specified
if max_steps is not None and max_steps > 0:
steps = steps[:max_steps]
# Renumber sequentially
for idx, step in enumerate(steps, start=1):
step.step = idx
fp_dict = fp.model_dump(exclude={"metadata"})
return ScanPlan(
target=target or fp.target,
fingerprint=fp_dict,
tci=tci_score,
tci_level=tci_level,
plan=steps,
metadata={
"scan_mode": scan_mode,
"total_steps": len(steps),
"total_quota": sum(s.quota for s in steps),
"estimated_duration_s": sum(s.timeout_s for s in steps),
"tci_breakdown": tci_res.breakdown,
},
)
def _filter_by_scan_mode(
self,
candidates: list[ScanModule],
scan_mode: str,
) -> list[ScanModule]:
mode = scan_mode.lower().strip()
if mode == "quick":
return [
m
for m in candidates
if m.default_priority in (PriorityTier.CRITICAL, PriorityTier.HIGH)
]
if mode == "standard":
return [
m
for m in candidates
if m.default_priority
in (PriorityTier.CRITICAL, PriorityTier.HIGH, PriorityTier.MEDIUM)
]
return candidates
def _resolve_safe_mode(
self,
mod: ScanModule,
fp: TargetFingerprint,
tci_score: float,
safe_mode_override: bool | None,
overrides: dict[str, Any],
) -> bool:
if safe_mode_override is not None:
return safe_mode_override
if "safe_mode" in overrides:
return bool(overrides["safe_mode"])
if bool(fp.waf_cdn) or tci_score >= 65.0:
return True
return mod.safe_mode_default
def _resolve_quota(
self,
mod: ScanModule,
tci_score: float,
overrides: dict[str, Any],
) -> int:
quota_val = overrides.get("quota")
if quota_val is not None:
return int(quota_val)
if tci_score >= 65.0:
return round(mod.default_quota * 1.5)
if tci_score <= 30.0:
return max(15, round(mod.default_quota * 0.6))
return mod.default_quota
def _resolve_timeout(
self,
mod: ScanModule,
tci_score: float,
overrides: dict[str, Any],
) -> float:
timeout_val = overrides.get("timeout_s")
if timeout_val is not None:
return float(timeout_val)
if tci_score >= 65.0:
return round(mod.default_timeout_s * 1.25, 1)
if tci_score <= 30.0:
return max(60.0, round(mod.default_timeout_s * 0.7, 1))
return mod.default_timeout_s
def _build_steps(
self,
candidates: list[ScanModule],
fp: TargetFingerprint,
tci_score: float,
safe_mode_override: bool | None,
module_overrides: dict[str, dict[str, Any]],
) -> list[ScanPlanStep]:
ordered_modules = self._order_modules(candidates)
steps: list[ScanPlanStep] = []
for idx, mod in enumerate(ordered_modules, start=1):
overrides = module_overrides.get(mod.name, {})
priority_val = overrides.get("priority")
priority = (
str(priority_val).lower()
if priority_val is not None
else self._elevate_priority(mod, fp, tci_score)
)
safe_mode = self._resolve_safe_mode(mod, fp, tci_score, safe_mode_override, overrides)
quota = self._resolve_quota(mod, tci_score, overrides)
timeout_s = self._resolve_timeout(mod, tci_score, overrides)
reason = self._build_step_reason(mod, fp, tci_score, safe_mode)
steps.append(
ScanPlanStep(
step=idx,
module=mod.name,
priority=priority,
safe_mode=safe_mode,
timeout_s=timeout_s,
quota=quota,
skills=list(mod.skills),
reason=reason,
)
)
return steps
def _elevate_priority(
self,
mod: ScanModule,
fp: TargetFingerprint,
tci_score: float,
) -> str:
base = mod.default_priority.value.lower()
if tci_score >= 70.0 and mod.name in ("id_or", "business-logic"):
return "high" if base == "medium" else base
has_db = any("sql" in b or "db" in b for b in fp.backend_technologies)
if has_db and mod.name == "sqli-fuzz":
return "high"
return base
def _build_step_reason(
self,
mod: ScanModule,
fp: TargetFingerprint,
tci_score: float,
safe_mode: bool,
) -> str:
notes: list[str] = [f"Phase: {mod.category}"]
if safe_mode and bool(fp.waf_cdn):
notes.append("safe_mode enabled due to WAF/CDN presence")
elif safe_mode and tci_score >= 65.0:
notes.append("safe_mode enabled due to high target complexity")
if mod.prerequisites:
notes.append(f"requires {', '.join(mod.prerequisites)}")
return "; ".join(notes)
def _order_modules(self, modules: list[ScanModule]) -> list[ScanModule]:
"""Order modules tactically by phase and prerequisite dependencies."""
by_name = {m.name: m for m in modules}
def sort_key(m: ScanModule) -> tuple[int, int, int, str]:
phase = _PHASE_ORDER.get(m.category.lower(), 99)
p_rank = -m.default_priority.rank
prereq_depth = len(m.prerequisites)
return (phase, prereq_depth, p_rank, m.name)
sorted_mods = sorted(modules, key=sort_key)
ordered: list[ScanModule] = []
placed: set[str] = set()
def place(m: ScanModule) -> None:
if m.name in placed:
return
for req in m.prerequisites:
if req in by_name and req not in placed:
place(by_name[req])
ordered.append(m)
placed.add(m.name)
for mod in sorted_mods:
place(mod)
return ordered

23
strix/core/__init__.py
View file

@ -1 +1,24 @@
"""Strix scan runtime core."""
from __future__ import annotations
from strix.core.tci import (
ApiSurface,
TargetComplexityIndex,
TargetFingerprint,
TCILevel,
TCIResult,
TCIWeights,
calculate_tci,
)
__all__ = [
"ApiSurface",
"TCILevel",
"TCIResult",
"TCIWeights",
"TargetComplexityIndex",
"TargetFingerprint",
"calculate_tci",
]

576
strix/core/tci.py Normal file
View file

@ -0,0 +1,576 @@
"""Target Complexity Index (TCI) calculation for dynamic vulnerability scanning.
TCI is a normalized score (0.0 .. 100.0) derived from target fingerprint data
(open ports, protocols, auth patterns, API surface, tech stack, WAF/CDN presence,
data sensitivity, and patch posture). Configurable weights allow customizing
the relative importance of each dimension.
"""
from __future__ import annotations
import logging
from enum import StrEnum
from typing import Any
from pydantic import BaseModel, ConfigDict, Field, field_validator
logger = logging.getLogger(__name__)
class TCILevel(StrEnum):
"""Complexity tiers based on TCI score."""
LOW = "low"
MODERATE = "moderate"
COMPLEX = "complex"
CRITICAL = "critical"
class ApiSurface(BaseModel):
"""API surface metrics for a target."""
model_config = ConfigDict(extra="allow")
endpoints_count: int = Field(default=0, ge=0)
parameters_count: int = Field(default=0, ge=0)
api_types: list[str] = Field(default_factory=list)
has_spec: bool = False
dynamic_routes: bool = False
class TargetFingerprint(BaseModel):
"""Structured fingerprint of a target's observable characteristics."""
model_config = ConfigDict(extra="allow")
target: str = ""
target_type: str = "web_app"
open_ports: list[int] = Field(default_factory=list)
protocols: list[str] = Field(default_factory=list)
auth_patterns: list[str] = Field(default_factory=list)
api_surface: ApiSurface = Field(default_factory=ApiSurface)
frontend_technologies: list[str] = Field(default_factory=list)
backend_technologies: list[str] = Field(default_factory=list)
waf_cdn: list[str] = Field(default_factory=list)
data_sensitivity: str = "standard"
patch_posture: str = "unknown"
metadata: dict[str, Any] = Field(default_factory=dict)
@field_validator("api_surface", mode="before")
@classmethod
def _coerce_api_surface(cls, v: Any) -> ApiSurface:
if isinstance(v, dict):
return ApiSurface(**v)
if isinstance(v, ApiSurface):
return v
return ApiSurface()
class TCIWeights(BaseModel):
"""Configurable weights for TCI scoring dimensions.
Weights are non-negative and automatically normalized so they sum to 1.0.
"""
model_config = ConfigDict(extra="forbid")
ports_protocols: float = Field(default=0.15, ge=0.0)
auth: float = Field(default=0.20, ge=0.0)
api_surface: float = Field(default=0.20, ge=0.0)
tech_stack: float = Field(default=0.15, ge=0.0)
waf_defense: float = Field(default=0.10, ge=0.0)
data_sensitivity: float = Field(default=0.10, ge=0.0)
patch_posture: float = Field(default=0.10, ge=0.0)
def normalized_weights(self) -> dict[str, float]:
"""Return dimension weights normalized to sum to 1.0."""
raw = {
"ports_protocols": self.ports_protocols,
"auth": self.auth,
"api_surface": self.api_surface,
"tech_stack": self.tech_stack,
"waf_defense": self.waf_defense,
"data_sensitivity": self.data_sensitivity,
"patch_posture": self.patch_posture,
}
total = sum(raw.values())
if total <= 0:
count = len(raw)
return dict.fromkeys(raw, 1.0 / count)
return {k: v / total for k, v in raw.items()}
class TCIResult(BaseModel):
"""Result of Target Complexity Index calculation."""
model_config = ConfigDict(extra="allow")
score: float = Field(ge=0.0, le=100.0)
level: TCILevel
sub_scores: dict[str, float]
weights: dict[str, float]
breakdown: dict[str, str]
def to_dict(self) -> dict[str, Any]:
"""Convert result to a plain serializable dict."""
return {
"score": self.score,
"level": self.level.value,
"sub_scores": self.sub_scores,
"weights": self.weights,
"breakdown": self.breakdown,
}
# Standard web and infrastructure port definitions for scoring
_STANDARD_WEB_PORTS = frozenset({80, 443})
_NON_STANDARD_WEB_PORTS = frozenset(
{
8080,
8443,
3000,
5000,
8000,
8008,
8888,
9000,
9443,
}
)
_INFRA_PORTS = frozenset(
{
21,
22,
23,
25,
53,
110,
143,
445,
1433,
1521,
3306,
3389,
5432,
5900,
6379,
8081,
9200,
11211,
27017,
}
)
def calc_ports_protocols_score(ports: list[int], protocols: list[str]) -> tuple[float, str]:
"""Score ports and protocols attack surface (0..100)."""
score = 0.0
notes: list[str] = []
clean_ports = list(dict.fromkeys(p for p in ports if isinstance(p, int) and p > 0))
clean_protocols = [p.lower().strip() for p in protocols if isinstance(p, str) and p.strip()]
if not clean_ports and not clean_protocols:
return 10.0, "No open ports or protocols reported (baseline web assumption)"
web_ports = [p for p in clean_ports if p in _STANDARD_WEB_PORTS]
non_std_web = [p for p in clean_ports if p in _NON_STANDARD_WEB_PORTS]
infra_ports = [p for p in clean_ports if p in _INFRA_PORTS]
other_ports = [
p
for p in clean_ports
if p not in _STANDARD_WEB_PORTS
and p not in _NON_STANDARD_WEB_PORTS
and p not in _INFRA_PORTS
]
if web_ports:
score += 20.0
notes.append(f"Standard web ports: {web_ports}")
if non_std_web:
added = min(35.0, len(non_std_web) * 15.0)
score += added
notes.append(f"Non-standard web ports (+{added:.0f}): {non_std_web}")
if infra_ports:
added = min(40.0, len(infra_ports) * 20.0)
score += added
notes.append(f"Infra/database ports (+{added:.0f}): {infra_ports}")
if other_ports:
added = min(25.0, len(other_ports) * 10.0)
score += added
notes.append(f"Other ports (+{added:.0f}): {other_ports}")
# Protocol diversity
proto_bonus = 0.0
for proto in clean_protocols:
if proto in ("grpc", "grpc-web"):
proto_bonus += 20.0
elif proto in ("ws", "wss", "websocket"):
proto_bonus += 15.0
elif proto not in ("http", "https"):
proto_bonus += 10.0
if proto_bonus > 0:
proto_bonus = min(30.0, proto_bonus)
score += proto_bonus
notes.append(f"Protocols (+{proto_bonus:.0f}): {clean_protocols}")
final_score = max(10.0, min(100.0, score))
return final_score, "; ".join(notes) or "Standard port exposure"
_AUTH_BASE_SCORES: dict[str, float] = {
"basic": 25.0,
"basic_auth": 25.0,
"api_key": 35.0,
"apikey": 35.0,
"token": 35.0, # nosec B105
"session": 45.0,
"cookie": 45.0,
"session_cookie": 45.0,
"jwt": 60.0,
"bearer": 60.0,
"oauth": 75.0,
"oauth2": 75.0,
"oidc": 75.0,
"sso": 75.0,
"saml": 85.0,
"mtls": 85.0,
"client_cert": 85.0,
}
def calc_auth_score(auth_patterns: list[str]) -> tuple[float, str]:
"""Score authentication complexity (0..100)."""
clean_patterns = [p.lower().strip() for p in auth_patterns if isinstance(p, str) and p.strip()]
if not clean_patterns:
return 0.0, "No authentication mechanisms detected"
max_base = 0.0
matched: list[str] = []
bonus = 0.0
for pattern in clean_patterns:
score = _AUTH_BASE_SCORES.get(pattern)
if score is not None:
max_base = max(max_base, score)
matched.append(pattern)
elif any(k in pattern for k in ("mfa", "2fa", "totp")):
bonus += 15.0
matched.append("mfa")
elif any(k in pattern for k in ("rbac", "abac", "multi_role", "tenant", "multi_tenant")):
bonus += 15.0
matched.append("rbac/multi-tenant")
else:
max_base = max(max_base, 30.0)
matched.append(pattern)
# Multi-pattern complexity
distinct_mechanisms = len(set(matched))
if distinct_mechanisms > 1:
bonus += min(20.0, (distinct_mechanisms - 1) * 10.0)
final_score = min(100.0, max_base + bonus)
desc = f"Auth patterns ({', '.join(clean_patterns)}): base={max_base:.0f}, bonus={bonus:.0f}"
return final_score, desc
def _score_endpoint_count(count: int, has_api_types: bool) -> tuple[float, str | None]:
if count <= 0:
desc = (
"API detected with uncounted endpoints"
if has_api_types
else "Minimal/static API surface"
)
return (15.0 if has_api_types else 10.0), desc
thresholds = (
(5, 25.0, f"{count} endpoints (small)"),
(20, 45.0, f"{count} endpoints (moderate)"),
(50, 65.0, f"{count} endpoints (broad)"),
(150, 85.0, f"{count} endpoints (large)"),
)
for limit, score, desc in thresholds:
if count <= limit:
return score, desc
return 95.0, f"{count} endpoints (enterprise-scale)"
def calc_api_surface_score(api_surface: ApiSurface | dict[str, Any]) -> tuple[float, str]:
"""Score API surface magnitude and diversity (0..100)."""
surface = ApiSurface(**api_surface) if isinstance(api_surface, dict) else api_surface
notes: list[str] = []
base_score, count_note = _score_endpoint_count(surface.endpoints_count, bool(surface.api_types))
if count_note:
notes.append(count_note)
score = base_score
# API Types diversity
for api_type in surface.api_types:
atype = api_type.lower().strip()
if atype in ("graphql", "grpc"):
score += 20.0
notes.append(atype.upper())
elif atype in ("ws", "websocket"):
score += 15.0
notes.append("WebSocket")
elif atype == "soap":
score += 15.0
notes.append("SOAP")
# Parameters & Schema
if surface.parameters_count > 50:
score += 15.0
notes.append(f"{surface.parameters_count} parameters")
elif surface.parameters_count > 15:
score += 10.0
notes.append(f"{surface.parameters_count} parameters")
if surface.has_spec:
score += 10.0
notes.append("Formal API specification discovered")
if surface.dynamic_routes:
score += 10.0
notes.append("Dynamic/parameterized routing")
final_score = min(100.0, max(10.0, score))
return final_score, "; ".join(notes)
_SPA_FRAMEWORKS = frozenset(
{
"react",
"vue",
"angular",
"nextjs",
"nuxt",
"svelte",
"spa",
"solid",
}
)
_BACKEND_FRAMEWORKS = frozenset(
{
"django",
"fastapi",
"flask",
"express",
"rails",
"spring",
"springboot",
"nestjs",
"laravel",
"aspnet",
"gin",
"echo",
"fiber",
}
)
_DB_ENGINES = frozenset(
{
"postgres",
"postgresql",
"mysql",
"mariadb",
"mongodb",
"redis",
"elasticsearch",
"cassandra",
"oracle",
"mssql",
"sqlite",
}
)
def calc_tech_stack_score(
frontend: list[str],
backend: list[str],
) -> tuple[float, str]:
"""Score technology stack heterogeneity and modern complexity (0..100)."""
clean_fe = [f.lower().strip() for f in frontend if isinstance(f, str) and f.strip()]
clean_be = [b.lower().strip() for b in backend if isinstance(b, str) and b.strip()]
score = 15.0 # baseline
notes: list[str] = []
has_spa = any(f in _SPA_FRAMEWORKS or "react" in f or "vue" in f for f in clean_fe)
if has_spa:
score += 30.0
notes.append(f"Modern SPA frontend: {clean_fe}")
elif clean_fe:
score += 15.0
notes.append(f"Frontend: {clean_fe}")
matched_be = [
b for b in clean_be if b in _BACKEND_FRAMEWORKS or any(k in b for k in _BACKEND_FRAMEWORKS)
]
if len(matched_be) > 1:
score += 35.0
notes.append(f"Polyglot/microservice backend: {clean_be}")
elif matched_be:
score += 25.0
notes.append(f"Backend framework: {matched_be}")
elif clean_be:
score += 15.0
notes.append(f"Backend: {clean_be}")
matched_db = [b for b in clean_be if b in _DB_ENGINES]
if matched_db:
db_pts = min(20.0, len(matched_db) * 10.0)
score += db_pts
notes.append(f"Databases: {matched_db}")
final_score = min(100.0, score)
return final_score, "; ".join(notes) or "Minimal tech stack"
_WAF_PROVIDERS = frozenset(
{
"cloudflare",
"aws_waf",
"akamai",
"imperva",
"fastly",
"incapsula",
"f5",
"modsecurity",
"barracuda",
"sucuri",
}
)
def calc_waf_defense_score(waf_cdn: list[str]) -> tuple[float, str]:
"""Score defensive posture and WAF presence (0..100)."""
clean_defenses = [d.lower().strip() for d in waf_cdn if isinstance(d, str) and d.strip()]
if not clean_defenses:
return 0.0, "No WAF or edge security detected"
has_waf = any(d in _WAF_PROVIDERS or "waf" in d for d in clean_defenses)
count = len(clean_defenses)
if has_waf:
score = 75.0 + min(25.0, (count - 1) * 15.0)
return min(100.0, score), f"Active WAF/defense: {', '.join(clean_defenses)}"
score = 40.0 + min(20.0, (count - 1) * 10.0)
return min(100.0, score), f"CDN/edge proxy present: {', '.join(clean_defenses)}"
def calc_data_sensitivity_score(sensitivity: str) -> tuple[float, str]:
"""Score target data and workflow sensitivity (0..100)."""
clean = (sensitivity or "").lower().strip()
if clean in ("low", "public", "static", "marketing"):
return 15.0, "Public or marketing content"
if clean in ("standard", "medium", "internal"):
return 45.0, "Standard application workflow / internal service"
if clean in ("high", "confidential", "admin", "admin_panel"):
return 75.0, "High-privilege admin panel or confidential business workflow"
if clean in ("critical", "pii", "financial", "payment", "pci", "healthcare", "hipaa"):
return 95.0, "Critical compliance scope (PII, financial, or healthcare data)"
return 45.0, f"Unspecified sensitivity ({sensitivity}), defaulting to standard"
def calc_patch_posture_score(posture: str) -> tuple[float, str]:
"""Score target freshness and legacy vulnerability likelihood (0..100)."""
clean = (posture or "").lower().strip()
if clean in ("modern", "hardened", "latest"):
return 20.0, "Modern, actively maintained and hardened posture"
if clean in ("current", "stable"):
return 35.0, "Current stable release cycle"
if clean in ("unknown", ""):
return 50.0, "Patch posture unknown / uninspected"
if clean in ("patch_lag", "outdated", "deprecated"):
return 75.0, "Known patch lag or deprecated components detected"
if clean in ("legacy", "vulnerable", "eol", "end_of_life"):
return 95.0, "Legacy stack or end-of-life framework version"
return 50.0, f"Unrecognized posture ({posture}), defaulting to moderate"
def calculate_tci(
fingerprint: TargetFingerprint | dict[str, Any],
weights: TCIWeights | None = None,
) -> TCIResult:
"""Calculate the Target Complexity Index (0..100) from a target fingerprint.
Args:
fingerprint: Target fingerprint model or dictionary of observables.
weights: Configurable weights. If None, default balanced weights are used.
Returns:
TCIResult containing score (0.0..100.0), tier level, sub-scores,
weights, and descriptive breakdown.
"""
if isinstance(fingerprint, TargetFingerprint):
fp = fingerprint
else:
fp = TargetFingerprint(**fingerprint)
w = weights or TCIWeights()
norm_w = w.normalized_weights()
s_ports, b_ports = calc_ports_protocols_score(fp.open_ports, fp.protocols)
s_auth, b_auth = calc_auth_score(fp.auth_patterns)
s_api, b_api = calc_api_surface_score(fp.api_surface)
s_tech, b_tech = calc_tech_stack_score(fp.frontend_technologies, fp.backend_technologies)
s_waf, b_waf = calc_waf_defense_score(fp.waf_cdn)
s_sens, b_sens = calc_data_sensitivity_score(fp.data_sensitivity)
s_patch, b_patch = calc_patch_posture_score(fp.patch_posture)
sub_scores = {
"ports_protocols": round(s_ports, 1),
"auth": round(s_auth, 1),
"api_surface": round(s_api, 1),
"tech_stack": round(s_tech, 1),
"waf_defense": round(s_waf, 1),
"data_sensitivity": round(s_sens, 1),
"patch_posture": round(s_patch, 1),
}
breakdown = {
"ports_protocols": b_ports,
"auth": b_auth,
"api_surface": b_api,
"tech_stack": b_tech,
"waf_defense": b_waf,
"data_sensitivity": b_sens,
"patch_posture": b_patch,
}
composite = sum(sub_scores[k] * norm_w[k] for k in sub_scores)
final_score = round(max(0.0, min(100.0, composite)), 1)
if final_score <= 30.0:
level = TCILevel.LOW
elif final_score <= 60.0:
level = TCILevel.MODERATE
elif final_score <= 80.0:
level = TCILevel.COMPLEX
else:
level = TCILevel.CRITICAL
return TCIResult(
score=final_score,
level=level,
sub_scores=sub_scores,
weights={k: round(v, 4) for k, v in norm_w.items()},
breakdown=breakdown,
)
class TargetComplexityIndex:
"""Namespace and helper class for Target Complexity Index (TCI) operations."""
@staticmethod
def calculate(
fingerprint: TargetFingerprint | dict[str, Any],
weights: TCIWeights | None = None,
) -> TCIResult:
"""Calculate TCI from fingerprint with optional weights."""
return calculate_tci(fingerprint, weights)

24
strix/modules/__init__.py Normal file
View file

@ -0,0 +1,24 @@
"""Scanning modules and registry for Strix."""
from __future__ import annotations
from strix.modules.registry import (
ModuleRegistry,
PriorityTier,
ScanModule,
default_registry,
get_module,
list_modules,
register_module,
)
__all__ = [
"ModuleRegistry",
"PriorityTier",
"ScanModule",
"default_registry",
"get_module",
"list_modules",
"register_module",
]

489
strix/modules/registry.py Normal file
View file

@ -0,0 +1,489 @@
"""Registry for vulnerability scanning modules with priority, quotas, and safe_mode hints."""
from __future__ import annotations
import logging
import threading
from enum import StrEnum
from typing import Any
from pydantic import BaseModel, ConfigDict, Field
from strix.core.tci import TargetFingerprint
logger = logging.getLogger(__name__)
class PriorityTier(StrEnum):
"""Scan module priority tiers."""
CRITICAL = "critical"
HIGH = "high"
MEDIUM = "medium"
LOW = "low"
@property
def rank(self) -> int:
"""Numeric rank for sorting (higher is higher priority)."""
ranks = {
PriorityTier.CRITICAL: 4,
PriorityTier.HIGH: 3,
PriorityTier.MEDIUM: 2,
PriorityTier.LOW: 1,
}
return ranks.get(self, 0)
class ScanModule(BaseModel):
"""Specification of a security testing module."""
model_config = ConfigDict(extra="allow")
name: str
description: str
category: str
target_types: list[str] = Field(default_factory=lambda: ["web_app", "api"])
default_priority: PriorityTier = PriorityTier.MEDIUM
safe_mode_default: bool = True
default_timeout_s: float = 300.0
default_quota: int = 50
skills: list[str] = Field(default_factory=list)
prerequisites: list[str] = Field(default_factory=list)
tags: list[str] = Field(default_factory=list)
intrusive: bool = False
min_tci: float = 0.0
# Applicability conditions
requires_auth: bool = False
requires_api: bool = False
requires_graphql: bool = False
requires_jwt: bool = False
class ModuleRegistry:
"""Registry managing available scan modules and metadata overrides."""
def __init__(self, populate_defaults: bool = True) -> None:
self._lock = threading.RLock()
self._modules: dict[str, ScanModule] = {}
if populate_defaults:
self._load_builtins()
def register(self, module: ScanModule) -> None:
"""Register or replace a scan module."""
with self._lock:
self._modules[module.name] = module
logger.debug("Registered scan module: %s (%s)", module.name, module.category)
def unregister(self, name: str) -> bool:
"""Unregister a scan module by name. Returns True if removed."""
with self._lock:
return self._modules.pop(name, None) is not None
def get(self, name: str) -> ScanModule | None:
"""Retrieve a scan module by name."""
with self._lock:
return self._modules.get(name)
def get_or_raise(self, name: str) -> ScanModule:
"""Retrieve a scan module by name or raise KeyError."""
with self._lock:
mod = self._modules.get(name)
if mod is None:
msg = f"Unknown scan module: '{name}'"
raise KeyError(msg)
return mod
def list_all(self) -> list[ScanModule]:
"""Return all registered modules."""
with self._lock:
return list(self._modules.values())
def list_by_category(self, category: str) -> list[ScanModule]:
"""Return modules matching the given category."""
clean = category.lower().strip()
with self._lock:
return [m for m in self._modules.values() if m.category.lower() == clean]
def list_by_tag(self, tag: str) -> list[ScanModule]:
"""Return modules carrying the given tag."""
clean = tag.lower().strip()
with self._lock:
return [m for m in self._modules.values() if clean in (t.lower() for t in m.tags)]
def update_metadata(
self,
name: str,
*,
priority: PriorityTier | str | None = None,
quota: int | None = None,
safe_mode: bool | None = None,
timeout_s: float | None = None,
description: str | None = None,
) -> ScanModule:
"""Update priority, quotas, safe_mode hints or timeouts on a registered module.
Args:
name: Name of the module to update.
priority: Updated priority tier.
quota: Updated execution quota.
safe_mode: Updated safe_mode default.
timeout_s: Updated timeout in seconds.
description: Updated description.
Returns:
The updated ScanModule instance.
"""
with self._lock:
current = self.get_or_raise(name)
data = current.model_dump()
if priority is not None:
p_tier = PriorityTier(priority) if isinstance(priority, str) else priority
data["default_priority"] = p_tier
if quota is not None:
data["default_quota"] = quota
if safe_mode is not None:
data["safe_mode_default"] = safe_mode
if timeout_s is not None:
data["default_timeout_s"] = timeout_s
if description is not None:
data["description"] = description
updated = ScanModule(**data)
self._modules[name] = updated
logger.info("Updated metadata for scan module: %s", name)
return updated
def find_applicable(
self,
fingerprint: TargetFingerprint | dict[str, Any],
) -> list[ScanModule]:
"""Find modules applicable to a given target fingerprint.
Filters out irrelevant modules based on tech stack, target type,
and missing prerequisite capabilities (e.g. omitting JWT checks when
no JWT authentication pattern was observed).
"""
fp = (
fingerprint
if isinstance(fingerprint, TargetFingerprint)
else TargetFingerprint(**fingerprint)
)
with self._lock:
candidates = list(self._modules.values())
applicable: list[ScanModule] = []
has_auth = bool(fp.auth_patterns)
has_jwt = any("jwt" in p.lower() or "bearer" in p.lower() for p in fp.auth_patterns)
has_api = (
fp.target_type == "api"
or fp.api_surface.endpoints_count > 0
or bool(fp.api_surface.api_types)
)
has_graphql = any("graphql" in t.lower() for t in fp.api_surface.api_types) or any(
"graphql" in p.lower() for p in fp.protocols
)
for mod in candidates:
# Target type check
if fp.target_type and mod.target_types and fp.target_type not in mod.target_types:
continue
# Feature conditions
if mod.requires_auth and not has_auth:
continue
if mod.requires_jwt and not has_jwt:
continue
if mod.requires_api and not has_api:
continue
if mod.requires_graphql and not has_graphql:
continue
applicable.append(mod)
return applicable
def _load_builtins(self) -> None:
"""Register the built-in vulnerability scanning modules."""
builtins: list[ScanModule] = [
ScanModule(
name="auth-enum",
description=(
"Authentication surface enumeration, credential exposure, "
"and brute-force resistance"
),
category="authentication",
target_types=["web_app", "api"],
default_priority=PriorityTier.HIGH,
safe_mode_default=True,
default_timeout_s=300.0,
default_quota=60,
skills=[
"vulnerabilities/authentication_jwt",
"vulnerabilities/weak_password_detection",
],
tags=["auth", "credentials", "owasp-a07"],
requires_auth=True,
),
ScanModule(
name="id_or",
description=(
"Insecure Direct Object Reference (IDOR) and broken object-level "
"authorization testing"
),
category="authorization",
target_types=["web_app", "api"],
default_priority=PriorityTier.HIGH,
safe_mode_default=True,
default_timeout_s=360.0,
default_quota=80,
skills=[
"vulnerabilities/idor",
"vulnerabilities/broken_function_level_authorization",
],
prerequisites=["auth-enum"],
tags=["authz", "idor", "owasp-a01"],
requires_auth=True,
),
ScanModule(
name="sqli-fuzz",
description=(
"SQL injection detection, parameter boundary fuzzing, and blind "
"SQLi verification"
),
category="injection",
target_types=["web_app", "api", "whitebox"],
default_priority=PriorityTier.HIGH,
safe_mode_default=False,
default_timeout_s=300.0,
default_quota=50,
skills=["vulnerabilities/sql_injection", "tooling/sqlmap"],
tags=["injection", "database", "owasp-a03"],
intrusive=True,
),
ScanModule(
name="xss-test",
description=(
"Cross-site scripting (reflected, stored, and DOM-based) payload evaluation"
),
category="injection",
target_types=["web_app"],
default_priority=PriorityTier.MEDIUM,
safe_mode_default=True,
default_timeout_s=240.0,
default_quota=40,
skills=["vulnerabilities/xss", "vulnerabilities/browser_security"],
tags=["xss", "client-side", "owasp-a03"],
),
ScanModule(
name="ssrf-probe",
description=(
"Server-Side Request Forgery probing against internal metadata and "
"private network endpoints"
),
category="injection",
target_types=["web_app", "api", "cloud"],
default_priority=PriorityTier.HIGH,
safe_mode_default=True,
default_timeout_s=300.0,
default_quota=50,
skills=["vulnerabilities/ssrf"],
tags=["ssrf", "network", "owasp-a10"],
),
ScanModule(
name="rce-eval",
description="Remote code execution and command injection verification",
category="injection",
target_types=["web_app", "api", "infra", "whitebox"],
default_priority=PriorityTier.CRITICAL,
safe_mode_default=True,
default_timeout_s=360.0,
default_quota=40,
skills=["vulnerabilities/rce", "vulnerabilities/argument_injection"],
tags=["rce", "command-injection", "critical"],
intrusive=True,
),
ScanModule(
name="api-route-fuzz",
description=(
"API route discovery, undocumented endpoint enumeration, and schema "
"mismatch testing"
),
category="recon",
target_types=["web_app", "api"],
default_priority=PriorityTier.HIGH,
safe_mode_default=False,
default_timeout_s=300.0,
default_quota=100,
skills=["custom/api_spec_testing", "tooling/ffuf", "tooling/katana"],
tags=["recon", "api", "discovery"],
),
ScanModule(
name="jwt-analyze",
description=(
"JWT signature verification, algorithm confusion (none/HS256), "
"and claim tampering"
),
category="authentication",
target_types=["web_app", "api"],
default_priority=PriorityTier.HIGH,
safe_mode_default=True,
default_timeout_s=240.0,
default_quota=40,
skills=["vulnerabilities/authentication_jwt"],
prerequisites=["auth-enum"],
tags=["jwt", "tokens", "auth"],
requires_auth=True,
requires_jwt=True,
),
ScanModule(
name="oauth-audit",
description=(
"OAuth2/OIDC flow tampering, redirect URI validation, and state "
"parameter verification"
),
category="authentication",
target_types=["web_app", "api"],
default_priority=PriorityTier.MEDIUM,
safe_mode_default=True,
default_timeout_s=240.0,
default_quota=40,
skills=["protocols/oauth"],
tags=["oauth", "oidc", "auth"],
requires_auth=True,
),
ScanModule(
name="graphql-audit",
description=(
"GraphQL schema introspection, batch query abuse, and nested "
"circular complexity checks"
),
category="api",
target_types=["web_app", "api"],
default_priority=PriorityTier.MEDIUM,
safe_mode_default=True,
default_timeout_s=300.0,
default_quota=50,
skills=["protocols/graphql"],
tags=["graphql", "api"],
requires_graphql=True,
),
ScanModule(
name="traversal-test",
description=(
"Directory/path traversal, local and remote file inclusion (LFI/RFI) validation"
),
category="injection",
target_types=["web_app", "api", "whitebox"],
default_priority=PriorityTier.HIGH,
safe_mode_default=True,
default_timeout_s=240.0,
default_quota=50,
skills=["vulnerabilities/path_traversal_lfi_rfi"],
tags=["lfi", "rfi", "traversal"],
intrusive=True,
),
ScanModule(
name="business-logic",
description=(
"Business workflow flaws, race conditions, price/quantity tampering, "
"and state bypasses"
),
category="logic",
target_types=["web_app", "api"],
default_priority=PriorityTier.MEDIUM,
safe_mode_default=True,
default_timeout_s=360.0,
default_quota=60,
skills=[
"vulnerabilities/business_logic",
"vulnerabilities/race_conditions",
],
prerequisites=["auth-enum"],
tags=["logic", "workflow"],
requires_auth=True,
),
ScanModule(
name="mass-assignment",
description=(
"Object property injection and mass assignment on REST update/create endpoints"
),
category="api",
target_types=["web_app", "api"],
default_priority=PriorityTier.MEDIUM,
safe_mode_default=True,
default_timeout_s=240.0,
default_quota=40,
skills=["vulnerabilities/mass_assignment"],
tags=["api", "mass-assignment"],
requires_api=True,
),
ScanModule(
name="cors-misconfig",
description=(
"Cross-Origin Resource Sharing (CORS) origin reflection and "
"null-origin validation"
),
category="recon",
target_types=["web_app", "api"],
default_priority=PriorityTier.LOW,
safe_mode_default=True,
default_timeout_s=180.0,
default_quota=30,
skills=["vulnerabilities/browser_security"],
tags=["cors", "headers"],
),
ScanModule(
name="info-disclosure",
description=(
"Information disclosure via verbose stack traces, debug headers, "
"and exposed files"
),
category="recon",
target_types=["web_app", "api", "infra"],
default_priority=PriorityTier.LOW,
safe_mode_default=True,
default_timeout_s=180.0,
default_quota=40,
skills=["vulnerabilities/information_disclosure"],
tags=["recon", "disclosure"],
),
ScanModule(
name="csrf-audit",
description="Cross-Site Request Forgery and anti-CSRF token defense verification",
category="authorization",
target_types=["web_app"],
default_priority=PriorityTier.LOW,
safe_mode_default=True,
default_timeout_s=180.0,
default_quota=30,
skills=["vulnerabilities/csrf"],
tags=["csrf", "authz"],
requires_auth=True,
),
]
for mod in builtins:
self._modules[mod.name] = mod
# Default singleton registry
default_registry = ModuleRegistry(populate_defaults=True)
def get_module(name: str) -> ScanModule | None:
"""Retrieve a module from the default registry."""
return default_registry.get(name)
def register_module(module: ScanModule) -> None:
"""Register a module in the default registry."""
default_registry.register(module)
def list_modules() -> list[ScanModule]:
"""List all modules in the default registry."""
return default_registry.list_all()

8
strix/skills/coordination/root_agent.md
View file

@ -25,6 +25,14 @@ Before spawning agents, analyze the target from the scan config/scope and any pr
3. **Determine approach** - blackbox, greybox, or whitebox assessment
4. **Prioritize by risk** - critical assets and high-value targets first
## Dynamic Scan Planning & Target Complexity Index (TCI)
After reconnaissance or when whitebox repository context is analyzed:
1. **Fingerprint the target** — summarize open ports, protocols, authentication schemes (JWT, OAuth, session cookies), API surface magnitude, frameworks, WAF/CDN presence, and data sensitivity.
2. **Compute TCI** — call `calculate_target_complexity` to calculate the Target Complexity Index (0..100) and identify primary complexity drivers.
3. **Generate an adaptive scan plan** — call `create_scan_plan` to produce a context-aware sequence of prioritized testing modules (e.g. `auth-enum` -> `id_or` -> `sqli-fuzz`). The plan automatically weeds out irrelevant tests (e.g. skipping JWT tests if JWT is unused), configures `safe_mode` hints when WAFs or high-complexity targets are detected, and orders prerequisites.
4. **Execute adaptively** — spawn specialized subagents according to the plan's priority tiers, assigning relevant skills and respecting quotas.
## Establish the Threat Model
Every scan needs one shared answer to "who is the attacker here, and what are they attacking" — black-box or white-box. Without it, five agents derive five different answers and their findings cannot be reconciled. Call `get_threat_model` on the target (a host, a URL, or a repository path) before you spawn hunters; if no model exists yet, derive one and share it with `save_threat_model`. It lives for this scan only — nothing carries over from an earlier run, so every scan derives its own — but within the run every agent reads the same document, and a model written from source is read back by an agent testing the deployment.

16
strix/tools/planner/__init__.py Normal file
View file

@ -0,0 +1,16 @@
"""Target Complexity Index and Scan Planner agent tools."""
from __future__ import annotations
from strix.tools.planner.tools import (
calculate_target_complexity,
create_scan_plan,
list_scan_modules,
)
__all__ = [
"calculate_target_complexity",
"create_scan_plan",
"list_scan_modules",
]

177
strix/tools/planner/tools.py Normal file
View file

@ -0,0 +1,177 @@
"""Agent tools for Target Complexity Index (TCI) and dynamic scan planning."""
from __future__ import annotations
import json
import logging
from typing import Any
from agents import RunContextWrapper, function_tool
from strix.agents.planner import ScanPlanner
from strix.core.tci import TargetFingerprint, calculate_tci
from strix.modules.registry import default_registry
logger = logging.getLogger(__name__)
def _parse_fingerprint_arg(raw: str | dict[str, Any]) -> dict[str, Any]:
"""Parse JSON string or dict into fingerprint dictionary."""
if isinstance(raw, dict):
return raw
if isinstance(raw, str):
stripped = raw.strip()
if not stripped:
return {}
try:
parsed = json.loads(stripped)
if isinstance(parsed, dict):
return parsed
except json.JSONDecodeError:
logger.warning("Failed to parse fingerprint JSON argument: %s", raw)
return {}
@function_tool(timeout=30)
async def calculate_target_complexity(
ctx: RunContextWrapper,
target: str,
fingerprint: str = "{}",
) -> str:
"""Calculate the Target Complexity Index (TCI: 0.0 .. 100.0) from fingerprint observables.
TCI measures the complexity and attack surface of a target based on:
- Open ports and protocols
- Authentication mechanisms (JWT, OAuth, SAML, session cookies, MFA, RBAC)
- API surface size (endpoint count, GraphQL vs REST, parameter count)
- Technology stack (front-end SPA, back-end polyglot/frameworks, databases)
- WAF and CDN presence (Cloudflare, AWS WAF, etc.)
- Data sensitivity (PII, financial, admin panels)
- Patch posture / legacy indicators
Args:
target: The target URL, domain, or identifier.
fingerprint: JSON object string containing fingerprint metadata:
- ``open_ports`` (list of int): e.g. ``[80, 443, 8080]``
- ``protocols`` (list of str): e.g. ``["https", "grpc"]``
- ``auth_patterns`` (list of str): e.g. ``["jwt", "oauth2"]``
- ``api_surface`` (dict): e.g. ``{"endpoints_count": 25, "api_types": ["rest"]}``
- ``frontend_technologies`` (list of str): e.g. ``["react"]``
- ``backend_technologies`` (list of str): e.g. ``["fastapi", "postgres"]``
- ``waf_cdn`` (list of str): e.g. ``["cloudflare"]``
- ``data_sensitivity`` (str): e.g. ``"high"`` or ``"standard"``
- ``patch_posture`` (str): e.g. ``"current"`` or ``"outdated"``
Returns:
JSON object containing ``target``, ``tci`` (0..100), ``level``,
``sub_scores``, and descriptive ``breakdown``.
"""
del ctx # Unused by calculation
fp_dict = _parse_fingerprint_arg(fingerprint)
if target and "target" not in fp_dict:
fp_dict["target"] = target
fp = TargetFingerprint(**fp_dict)
res = calculate_tci(fp)
payload = {
"target": target or fp.target,
"tci": res.score,
"level": res.level.value,
"sub_scores": res.sub_scores,
"breakdown": res.breakdown,
}
return json.dumps(payload, ensure_ascii=False, indent=2, default=str)
@function_tool(timeout=30)
async def create_scan_plan(
ctx: RunContextWrapper,
target: str,
fingerprint: str = "{}",
scan_mode: str = "deep",
safe_mode: bool | None = None,
) -> str:
"""Generate a dynamic, context-aware scan plan from fingerprint and TCI.
Maps target fingerprint to a prioritized sequence of tactical testing modules,
eliminating irrelevant tests and optimizing resource allocation.
Args:
target: Target URL, API endpoint, or hostname to assess.
fingerprint: JSON string describing target observables.
scan_mode: Scan mode ('quick', 'standard', 'deep', 'diff').
- ``quick``: critical/high priority tests only with constrained quotas.
- ``standard``: balanced coverage of critical, high, and medium vectors.
- ``deep``: thorough assessment across all applicable modules.
safe_mode: Optional boolean override for safe mode (cautious, evasive rate-limiting).
If omitted, safe_mode is automatically decided based on WAF presence and TCI.
Returns:
JSON payload containing ``target``, ``fingerprint``, ``tci``, and ``plan``:
```json
{
"target": "https://api.example.local",
"fingerprint": { ... },
"tci": 78.3,
"plan": [
{"step": 1, "module": "auth-enum", "priority": "high", "safe_mode": true},
{"step": 2, "module": "id_or", "priority": "high", "safe_mode": true},
{"step": 3, "module": "sqli-fuzz", "priority": "medium", "safe_mode": true}
]
}
```
"""
del ctx
fp_dict = _parse_fingerprint_arg(fingerprint)
if target and "target" not in fp_dict:
fp_dict["target"] = target
planner = ScanPlanner()
plan = planner.generate_plan(
target=target,
fingerprint=fp_dict,
scan_mode=scan_mode,
safe_mode_override=safe_mode,
)
return plan.to_json(compact=True, indent=2)
@function_tool(timeout=30)
async def list_scan_modules(
ctx: RunContextWrapper,
category: str = "",
) -> str:
"""List available vulnerability scanning modules in the registry.
Args:
category: Optional category filter (e.g. 'authentication', 'injection',
'authorization', 'api', 'recon', 'logic').
Returns:
JSON list of registered modules with description, priority, category,
safe_mode hints, and associated skills.
"""
del ctx
clean_cat = category.strip().lower()
if clean_cat:
modules = default_registry.list_by_category(clean_cat)
else:
modules = default_registry.list_all()
items = [
{
"name": m.name,
"category": m.category,
"description": m.description,
"default_priority": m.default_priority.value,
"safe_mode_default": m.safe_mode_default,
"skills": m.skills,
"prerequisites": m.prerequisites,
"tags": m.tags,
"intrusive": m.intrusive,
}
for m in modules
]
return json.dumps(items, ensure_ascii=False, indent=2, default=str)

152
tests/test_module_registry.py Normal file
View file

@ -0,0 +1,152 @@
"""Tests for ScanModule and ModuleRegistry."""
from __future__ import annotations
import pytest
from strix.core.tci import ApiSurface, TargetFingerprint
from strix.modules.registry import (
ModuleRegistry,
PriorityTier,
ScanModule,
default_registry,
get_module,
list_modules,
register_module,
)
def test_default_registry_has_builtins() -> None:
modules = list_modules()
assert len(modules) >= 10
auth_enum = get_module("auth-enum")
assert auth_enum is not None
assert auth_enum.category == "authentication"
assert auth_enum.default_priority == PriorityTier.HIGH
assert auth_enum.requires_auth is True
id_or = get_module("id_or")
assert id_or is not None
assert id_or.category == "authorization"
assert "auth-enum" in id_or.prerequisites
sqli = get_module("sqli-fuzz")
assert sqli is not None
assert sqli.intrusive is True
def test_custom_module_lifecycle() -> None:
reg = ModuleRegistry(populate_defaults=False)
assert len(reg.list_all()) == 0
custom = ScanModule(
name="custom-cors-audit",
description="Custom CORS validator",
category="recon",
target_types=["web_app", "api"],
default_priority=PriorityTier.LOW,
safe_mode_default=True,
default_quota=25,
default_timeout_s=120.0,
tags=["custom", "cors"],
)
reg.register(custom)
assert reg.get("custom-cors-audit") == custom
assert reg.get_or_raise("custom-cors-audit") == custom
assert len(reg.list_all()) == 1
assert len(reg.list_by_category("recon")) == 1
assert len(reg.list_by_tag("cors")) == 1
assert reg.unregister("custom-cors-audit") is True
assert reg.get("custom-cors-audit") is None
assert reg.unregister("custom-cors-audit") is False
with pytest.raises(KeyError, match="Unknown scan module"):
reg.get_or_raise("custom-cors-audit")
def test_update_metadata_priority_quota_safe_mode() -> None:
reg = ModuleRegistry(populate_defaults=True)
# Initial state
orig = reg.get_or_raise("sqli-fuzz")
assert orig.safe_mode_default is False
assert orig.default_quota == 50
# Accept priority metadata, quotas, safe_mode hints, and timeouts
updated = reg.update_metadata(
"sqli-fuzz",
priority=PriorityTier.CRITICAL,
quota=120,
safe_mode=True,
timeout_s=600.0,
description="Deep aggressive SQLi testing with strict rate limits",
)
assert updated.default_priority == PriorityTier.CRITICAL
assert updated.default_quota == 120
assert updated.safe_mode_default is True
assert updated.default_timeout_s == 600.0
assert updated.description == "Deep aggressive SQLi testing with strict rate limits"
# String priority conversion
updated2 = reg.update_metadata("sqli-fuzz", priority="low")
assert updated2.default_priority == PriorityTier.LOW
def test_find_applicable_filters_irrelevant_tests() -> None:
reg = ModuleRegistry(populate_defaults=True)
# 1. Unauthenticated target: auth-dependent modules must be excluded
fp_no_auth = TargetFingerprint(
target="https://public.example.com",
target_type="web_app",
auth_patterns=[],
)
applicable_no_auth = {m.name for m in reg.find_applicable(fp_no_auth)}
assert "auth-enum" not in applicable_no_auth
assert "id_or" not in applicable_no_auth
assert "jwt-analyze" not in applicable_no_auth
assert "xss-test" in applicable_no_auth
# 2. Authenticated target with session cookie (no JWT, no GraphQL)
fp_cookie = TargetFingerprint(
target="https://app.example.com",
target_type="web_app",
auth_patterns=["session_cookie"],
)
applicable_cookie = {m.name for m in reg.find_applicable(fp_cookie)}
assert "auth-enum" in applicable_cookie
assert "id_or" in applicable_cookie
assert "jwt-analyze" not in applicable_cookie
assert "graphql-audit" not in applicable_cookie
# 3. Authenticated API with JWT and GraphQL
fp_jwt_graphql = TargetFingerprint(
target="https://api.example.com",
target_type="api",
auth_patterns=["jwt"],
api_surface=ApiSurface(
endpoints_count=10,
api_types=["rest", "graphql"],
),
)
applicable_jwt_graphql = {m.name for m in reg.find_applicable(fp_jwt_graphql)}
assert "jwt-analyze" in applicable_jwt_graphql
assert "graphql-audit" in applicable_jwt_graphql
assert "id_or" in applicable_jwt_graphql
def test_register_module_helper() -> None:
test_mod = ScanModule(
name="test-singleton-mod",
description="Helper test module",
category="recon",
)
register_module(test_mod)
try:
assert default_registry.get("test-singleton-mod") == test_mod
finally:
default_registry.unregister("test-singleton-mod")

107
tests/test_planner_tools.py Normal file
View file

@ -0,0 +1,107 @@
"""Tests for planner agent tools."""
from __future__ import annotations
import json
from agents.tool_context import ToolContext
from strix.tools.planner.tools import (
calculate_target_complexity,
create_scan_plan,
list_scan_modules,
)
def _make_ctx(tool_name: str) -> ToolContext:
return ToolContext(
context={"agent_id": "test-agent"},
tool_name=tool_name,
tool_call_id="call-1",
tool_arguments="{}",
)
async def test_calculate_target_complexity_tool() -> None:
fp_json = json.dumps(
{
"open_ports": [443, 8080],
"auth_patterns": ["jwt"],
"api_surface": {"endpoints_count": 20},
"waf_cdn": ["cloudflare"],
}
)
result_raw = await calculate_target_complexity.on_invoke_tool(
_make_ctx("calculate_target_complexity"),
json.dumps({"target": "https://test.local", "fingerprint": fp_json}),
)
data = json.loads(result_raw)
assert data["target"] == "https://test.local"
assert "tci" in data
assert 0.0 <= data["tci"] <= 100.0
assert "sub_scores" in data
assert "breakdown" in data
assert "level" in data
async def test_create_scan_plan_tool() -> None:
fp_json = json.dumps(
{
"target_type": "api",
"open_ports": [443],
"auth_patterns": ["jwt", "oauth2"],
"api_surface": {"endpoints_count": 40, "api_types": ["rest"]},
"data_sensitivity": "high",
}
)
result_raw = await create_scan_plan.on_invoke_tool(
_make_ctx("create_scan_plan"),
json.dumps(
{
"target": "https://api.test.local",
"fingerprint": fp_json,
"scan_mode": "quick",
}
),
)
data = json.loads(result_raw)
assert data["target"] == "https://api.test.local"
assert "tci" in data
assert "plan" in data
assert isinstance(data["plan"], list)
assert len(data["plan"]) > 0
step1 = data["plan"][0]
assert "step" in step1
assert "module" in step1
assert "priority" in step1
assert "safe_mode" in step1
async def test_list_scan_modules_tool() -> None:
result_raw = await list_scan_modules.on_invoke_tool(
_make_ctx("list_scan_modules"),
json.dumps({}),
)
data = json.loads(result_raw)
assert isinstance(data, list)
assert len(data) >= 10
names = [m["name"] for m in data]
assert "auth-enum" in names
assert "id_or" in names
assert "sqli-fuzz" in names
# Filtered by category
filtered_raw = await list_scan_modules.on_invoke_tool(
_make_ctx("list_scan_modules"),
json.dumps({"category": "authentication"}),
)
filtered = json.loads(filtered_raw)
for m in filtered:
assert m["category"] == "authentication"

200
tests/test_scan_planner.py Normal file
View file

@ -0,0 +1,200 @@
"""Tests for adaptive ScanPlanner and plan generation from TCI."""
from __future__ import annotations
import json
from strix.agents.planner import ScanPlan, ScanPlanner
from strix.core.tci import ApiSurface, TargetFingerprint
def test_planner_generates_exact_payload_schema() -> None:
planner = ScanPlanner()
target = "https://api.example.local"
fp = {
"target_type": "api",
"open_ports": [443],
"protocols": ["https"],
"auth_patterns": ["jwt"],
"api_surface": {
"endpoints_count": 30,
"api_types": ["rest"],
},
"waf_cdn": ["cloudflare"],
"data_sensitivity": "high",
}
plan = planner.generate_plan(target, fp, scan_mode="deep")
assert isinstance(plan, ScanPlan)
assert plan.target == target
assert isinstance(plan.tci, float)
assert plan.tci > 0.0
assert len(plan.plan) > 0
wire_dict = plan.to_dict(compact=True)
assert wire_dict["target"] == target
assert wire_dict["tci"] == plan.tci
assert isinstance(wire_dict["fingerprint"], dict)
assert isinstance(wire_dict["plan"], list)
for item in wire_dict["plan"]:
assert "step" in item
assert "module" in item
assert "priority" in item
assert "safe_mode" in item
assert isinstance(item["step"], int)
assert isinstance(item["module"], str)
assert isinstance(item["safe_mode"], bool)
# Validate JSON serialization round-trip
serialized = plan.to_json(compact=True)
parsed = json.loads(serialized)
assert parsed["target"] == target
assert parsed["tci"] == plan.tci
assert len(parsed["plan"]) == len(plan.plan)
def test_tactical_ordering_and_prerequisites() -> None:
planner = ScanPlanner()
target = "https://app.local"
fp = TargetFingerprint(
target=target,
auth_patterns=["session_cookie"],
api_surface=ApiSurface(endpoints_count=15),
backend_technologies=["django", "postgres"],
)
plan = planner.generate_plan(target, fp, scan_mode="deep")
mod_order = plan.module_names()
# auth-enum must be scheduled before id_or and business-logic
if "auth-enum" in mod_order and "id_or" in mod_order:
auth_idx = mod_order.index("auth-enum")
idor_idx = mod_order.index("id_or")
assert auth_idx < idor_idx
if "auth-enum" in mod_order and "business-logic" in mod_order:
auth_idx = mod_order.index("auth-enum")
logic_idx = mod_order.index("business-logic")
assert auth_idx < logic_idx
def test_waf_activates_safe_mode() -> None:
planner = ScanPlanner()
target = "https://protected.local"
# Target behind Cloudflare WAF
fp_waf = TargetFingerprint(
target=target,
auth_patterns=["basic"],
waf_cdn=["cloudflare_waf"],
)
plan_waf = planner.generate_plan(target, fp_waf)
# Intrusive modules should have safe_mode=True
sqli_step = plan_waf.get_step_by_module("sqli-fuzz")
if sqli_step:
assert sqli_step.safe_mode is True
# Target without WAF and low complexity
fp_clean = TargetFingerprint(
target=target,
auth_patterns=["basic"],
waf_cdn=[],
data_sensitivity="low",
)
plan_clean = planner.generate_plan(target, fp_clean)
sqli_clean_step = plan_clean.get_step_by_module("sqli-fuzz")
if sqli_clean_step:
# sqli-fuzz default is False when no WAF and low complexity
assert sqli_clean_step.safe_mode is False
def test_quick_mode_filters_priorities() -> None:
planner = ScanPlanner()
target = "https://quick.local"
fp = TargetFingerprint(
target=target,
auth_patterns=["jwt"],
api_surface=ApiSurface(endpoints_count=10),
)
plan_quick = planner.generate_plan(target, fp, scan_mode="quick")
for step in plan_quick.plan:
assert step.priority in ("critical", "high")
def test_max_steps_constraint() -> None:
planner = ScanPlanner()
target = "https://limited.local"
fp = TargetFingerprint(
target=target,
auth_patterns=["jwt", "oauth2"],
api_surface=ApiSurface(endpoints_count=20, api_types=["rest", "graphql"]),
)
plan = planner.generate_plan(target, fp, scan_mode="deep", max_steps=3)
assert len(plan.plan) == 3
assert [s.step for s in plan.plan] == [1, 2, 3]
def test_module_overrides_application() -> None:
planner = ScanPlanner()
target = "https://overrides.local"
fp = TargetFingerprint(
target=target,
auth_patterns=["jwt"],
)
overrides = {
"auth-enum": {
"priority": "critical",
"safe_mode": False,
"quota": 99,
"timeout_s": 555.0,
}
}
plan = planner.generate_plan(
target,
fp,
module_overrides=overrides,
)
step = plan.get_step_by_module("auth-enum")
assert step is not None
assert step.priority == "critical"
assert step.safe_mode is False
assert step.quota == 99
assert step.timeout_s == 555.0
def test_plan_helpers_and_aggregation() -> None:
planner = ScanPlanner()
target = "https://helpers.local"
fp = TargetFingerprint(
target=target,
auth_patterns=["jwt"],
api_surface=ApiSurface(endpoints_count=10),
)
plan = planner.generate_plan(target, fp)
step1 = plan.get_step(1)
assert step1 is not None
assert step1.step == 1
assert plan.get_step(999) is None
high_steps = plan.get_steps_by_priority("high")
assert isinstance(high_steps, list)
assert plan.total_quota() > 0
assert plan.estimated_duration_s() > 0.0
# Non-compact serialization contains extended fields
full_dict = step1.to_dict(compact=False)
assert "timeout_s" in full_dict
assert "quota" in full_dict
assert "skills" in full_dict
assert "reason" in full_dict

206
tests/test_tci.py Normal file
View file

@ -0,0 +1,206 @@
"""Tests for Target Complexity Index (TCI) calculation and weighting."""
from __future__ import annotations
import pytest
from strix.core.tci import (
ApiSurface,
TargetComplexityIndex,
TargetFingerprint,
TCILevel,
TCIResult,
TCIWeights,
calc_api_surface_score,
calc_auth_score,
calc_data_sensitivity_score,
calc_patch_posture_score,
calc_ports_protocols_score,
calc_tech_stack_score,
calc_waf_defense_score,
calculate_tci,
)
def test_empty_fingerprint_baseline() -> None:
fp = TargetFingerprint()
res = calculate_tci(fp)
assert isinstance(res, TCIResult)
assert 0.0 <= res.score <= 30.0
assert res.level == TCILevel.LOW
assert "ports_protocols" in res.sub_scores
assert "auth" in res.sub_scores
assert res.sub_scores["auth"] == 0.0
def test_complex_enterprise_fingerprint() -> None:
fp = TargetFingerprint(
target="https://api.enterprise.local",
target_type="api",
open_ports=[80, 443, 8080, 8443, 5432],
protocols=["https", "grpc", "wss"],
auth_patterns=["jwt", "oauth2", "mfa", "rbac"],
api_surface=ApiSurface(
endpoints_count=65,
parameters_count=120,
api_types=["rest", "graphql", "grpc"],
has_spec=True,
dynamic_routes=True,
),
frontend_technologies=["react", "nextjs"],
backend_technologies=["fastapi", "postgres", "redis"],
waf_cdn=["cloudflare", "aws_waf"],
data_sensitivity="critical",
patch_posture="patch_lag",
)
res = calculate_tci(fp)
assert res.score >= 70.0
assert res.level in (TCILevel.COMPLEX, TCILevel.CRITICAL)
assert res.sub_scores["auth"] >= 80.0
assert res.sub_scores["api_surface"] >= 80.0
assert res.sub_scores["waf_defense"] >= 80.0
assert res.sub_scores["data_sensitivity"] >= 90.0
# Serialization
d = res.to_dict()
assert d["score"] == res.score
assert d["level"] == res.level.value
assert isinstance(d["sub_scores"], dict)
assert isinstance(d["breakdown"], dict)
def test_dict_coercion_and_target_complexity_index_helper() -> None:
payload = {
"target": "https://service.local",
"open_ports": [443],
"protocols": ["https"],
"auth_patterns": ["session_cookie"],
"api_surface": {
"endpoints_count": 10,
"api_types": ["rest"],
},
}
res1 = calculate_tci(payload)
res2 = TargetComplexityIndex.calculate(payload)
assert res1.score == res2.score
assert res1.level == res2.level
assert res1.sub_scores == res2.sub_scores
def test_weights_normalization_and_customization() -> None:
w = TCIWeights(
ports_protocols=0.0,
auth=1.0,
api_surface=0.0,
tech_stack=0.0,
waf_defense=0.0,
data_sensitivity=0.0,
patch_posture=0.0,
)
norm = w.normalized_weights()
assert norm["auth"] == 1.0
assert norm["ports_protocols"] == 0.0
fp_no_auth = TargetFingerprint(open_ports=[443, 8080, 9000])
res_no_auth = calculate_tci(fp_no_auth, weights=w)
assert res_no_auth.score == 0.0
fp_jwt = TargetFingerprint(auth_patterns=["jwt"])
res_jwt = calculate_tci(fp_jwt, weights=w)
assert res_jwt.score == res_jwt.sub_scores["auth"]
def test_zero_weights_fallback() -> None:
w = TCIWeights(
ports_protocols=0.0,
auth=0.0,
api_surface=0.0,
tech_stack=0.0,
waf_defense=0.0,
data_sensitivity=0.0,
patch_posture=0.0,
)
norm = w.normalized_weights()
assert len(norm) == 7
assert sum(norm.values()) == pytest.approx(1.0)
def test_ports_protocols_scoring() -> None:
s0, _ = calc_ports_protocols_score([], [])
assert s0 == 10.0
s1, _ = calc_ports_protocols_score([80, 443], ["https"])
assert s1 == 20.0
s2, _ = calc_ports_protocols_score([80, 443, 8080, 8443, 3306], ["https", "grpc"])
assert s2 > s1
assert s2 <= 100.0
def test_auth_scoring() -> None:
s_none, _ = calc_auth_score([])
assert s_none == 0.0
s_basic, _ = calc_auth_score(["basic"])
s_jwt, _ = calc_auth_score(["jwt"])
s_oauth, _ = calc_auth_score(["oauth2", "mfa"])
assert s_basic < s_jwt < s_oauth
assert s_oauth <= 100.0
def test_api_surface_scoring() -> None:
s_minimal, _ = calc_api_surface_score(ApiSurface(endpoints_count=0))
s_small, _ = calc_api_surface_score(ApiSurface(endpoints_count=4))
s_large, _ = calc_api_surface_score(
ApiSurface(endpoints_count=60, api_types=["graphql"], has_spec=True)
)
assert s_minimal < s_small < s_large
assert s_large <= 100.0
def test_tech_stack_scoring() -> None:
s_static, _ = calc_tech_stack_score([], [])
s_spa, _ = calc_tech_stack_score(["react", "nextjs"], [])
s_polyglot, _ = calc_tech_stack_score(["react"], ["fastapi", "spring", "postgres", "redis"])
assert s_static < s_spa < s_polyglot
assert s_polyglot <= 100.0
def test_waf_defense_scoring() -> None:
s_none, _ = calc_waf_defense_score([])
assert s_none == 0.0
s_cdn, _ = calc_waf_defense_score(["cloudfront"])
s_waf, _ = calc_waf_defense_score(["cloudflare_waf", "aws_waf"])
assert s_cdn < s_waf
assert s_waf <= 100.0
def test_data_sensitivity_scoring() -> None:
s_pub, _ = calc_data_sensitivity_score("public")
s_std, _ = calc_data_sensitivity_score("standard")
s_admin, _ = calc_data_sensitivity_score("admin_panel")
s_crit, _ = calc_data_sensitivity_score("financial")
assert s_pub < s_std < s_admin < s_crit
assert s_crit == 95.0
def test_patch_posture_scoring() -> None:
s_mod, _ = calc_patch_posture_score("modern")
s_cur, _ = calc_patch_posture_score("current")
s_unk, _ = calc_patch_posture_score("unknown")
s_lag, _ = calc_patch_posture_score("patch_lag")
s_eol, _ = calc_patch_posture_score("legacy")
assert s_mod < s_cur < s_unk < s_lag < s_eol
assert s_eol == 95.0