From 64622e8d7e3a5595d64d1afbc05e402603db7d2b Mon Sep 17 00:00:00 2001 From: MriBoyd Date: Wed, 14 Oct 2026 11:12:00 +0300 Subject: [PATCH] feat(agent): add Target Complexity Index (TCI) + adaptive ScanPlanner for dynamic vulnerability scanning (#46) - 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 --- strix/agents/__init__.py | 16 + strix/agents/factory.py | 8 + strix/agents/planner.py | 382 ++++++++++++++++ strix/core/__init__.py | 23 + strix/core/tci.py | 576 ++++++++++++++++++++++++ strix/modules/__init__.py | 24 + strix/modules/registry.py | 489 ++++++++++++++++++++ strix/skills/coordination/root_agent.md | 8 + strix/tools/planner/__init__.py | 16 + strix/tools/planner/tools.py | 177 ++++++++ tests/test_module_registry.py | 152 +++++++ tests/test_planner_tools.py | 107 +++++ tests/test_scan_planner.py | 200 ++++++++ tests/test_tci.py | 206 +++++++++ 14 files changed, 2384 insertions(+) create mode 100644 strix/agents/planner.py create mode 100644 strix/core/tci.py create mode 100644 strix/modules/__init__.py create mode 100644 strix/modules/registry.py create mode 100644 strix/tools/planner/__init__.py create mode 100644 strix/tools/planner/tools.py create mode 100644 tests/test_module_registry.py create mode 100644 tests/test_planner_tools.py create mode 100644 tests/test_scan_planner.py create mode 100644 tests/test_tci.py diff --git a/strix/agents/__init__.py b/strix/agents/__init__.py index e69de29b..b8cdf582 100644 --- a/strix/agents/__init__.py +++ b/strix/agents/__init__.py @@ -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", +] diff --git a/strix/agents/factory.py b/strix/agents/factory.py index 4cd9e037..9e3c9d9c 100644 --- a/strix/agents/factory.py +++ b/strix/agents/factory.py @@ -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, diff --git a/strix/agents/planner.py b/strix/agents/planner.py new file mode 100644 index 00000000..ff0d5833 --- /dev/null +++ b/strix/agents/planner.py @@ -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 diff --git a/strix/core/__init__.py b/strix/core/__init__.py index 8e07f9d9..62a5b201 100644 --- a/strix/core/__init__.py +++ b/strix/core/__init__.py @@ -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", +] diff --git a/strix/core/tci.py b/strix/core/tci.py new file mode 100644 index 00000000..25c378f4 --- /dev/null +++ b/strix/core/tci.py @@ -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) diff --git a/strix/modules/__init__.py b/strix/modules/__init__.py new file mode 100644 index 00000000..a275159a --- /dev/null +++ b/strix/modules/__init__.py @@ -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", +] diff --git a/strix/modules/registry.py b/strix/modules/registry.py new file mode 100644 index 00000000..fabfb1c0 --- /dev/null +++ b/strix/modules/registry.py @@ -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() diff --git a/strix/skills/coordination/root_agent.md b/strix/skills/coordination/root_agent.md index fa88aaa5..9b31809b 100644 --- a/strix/skills/coordination/root_agent.md +++ b/strix/skills/coordination/root_agent.md @@ -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. diff --git a/strix/tools/planner/__init__.py b/strix/tools/planner/__init__.py new file mode 100644 index 00000000..fe039b12 --- /dev/null +++ b/strix/tools/planner/__init__.py @@ -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", +] diff --git a/strix/tools/planner/tools.py b/strix/tools/planner/tools.py new file mode 100644 index 00000000..5fb9a6df --- /dev/null +++ b/strix/tools/planner/tools.py @@ -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) diff --git a/tests/test_module_registry.py b/tests/test_module_registry.py new file mode 100644 index 00000000..fe8544f8 --- /dev/null +++ b/tests/test_module_registry.py @@ -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") diff --git a/tests/test_planner_tools.py b/tests/test_planner_tools.py new file mode 100644 index 00000000..aca7115b --- /dev/null +++ b/tests/test_planner_tools.py @@ -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" diff --git a/tests/test_scan_planner.py b/tests/test_scan_planner.py new file mode 100644 index 00000000..0236c3c5 --- /dev/null +++ b/tests/test_scan_planner.py @@ -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 diff --git a/tests/test_tci.py b/tests/test_tci.py new file mode 100644 index 00000000..d877fd51 --- /dev/null +++ b/tests/test_tci.py @@ -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