Merge bc949d7bfe into ca20626f84

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",
+]

strix/agents/factory.py

@@ -46,6 +46,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,
@@ -596,6 +601,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,

strix/agents/planner.py

@@ -0,0 +1,488 @@
+"""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,
+        include_prerequisites: bool = True,
+    ) -> 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.
+            include_prerequisites: Whether to automatically include missing required
+                prerequisite modules from applicable pool (default: True). If False or
+                if requirements cannot be met on the target, dependent modules are dropped.
+
+        Returns:
+            Structured ScanPlan with ordered, prioritized steps.
+        """
+        fp = (
+            fingerprint
+            if isinstance(fingerprint, TargetFingerprint)
+            else TargetFingerprint(**(fingerprint or {}))
+        )
+        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
+        applicable = self.registry.find_applicable(fp)
+        applicable_by_name = {m.name: m for m in applicable}
+        overrides_dict = module_overrides or {}
+
+        # 3. Calculate final effective priorities BEFORE filtering (reflects overrides)
+        effective_priorities: dict[str, PriorityTier] = {}
+        for mod in applicable:
+            effective_priorities[mod.name] = self._resolve_module_priority(
+                mod, fp, tci_score, overrides_dict.get(mod.name, {})
+            )
+
+        # 4. Filter candidates by scan_mode using effective priority
+        selected = self._filter_by_scan_mode(applicable, scan_mode, effective_priorities)
+
+        # 5. Filter by priority if requested using effective priority
+        if priority_filter:
+            allowed_p = {p.lower().strip() for p in priority_filter}
+            selected = [
+                m
+                for m in selected
+                if effective_priorities.get(m.name, m.default_priority).value.lower() in allowed_p
+            ]
+
+        # 6. Resolve prerequisites: include required steps or drop unsatisfied modules
+        candidates = self._resolve_prerequisites(
+            selected, applicable_by_name, include_prerequisites=include_prerequisites
+        )
+
+        # 7. 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=overrides_dict,
+            effective_priorities=effective_priorities,
+        )
+
+        # 8. Apply max_steps limit if specified
+        if max_steps is not None:
+            if max_steps < 0:
+                raise ValueError("max_steps must be non-negative")
+            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 _resolve_prerequisites(
+        self,
+        selected: list[ScanModule],
+        applicable_by_name: dict[str, ScanModule],
+        include_prerequisites: bool,
+    ) -> list[ScanModule]:
+        """Include required prerequisites or drop modules whose requirements cannot be met."""
+        selected_by_name = {m.name: m for m in selected}
+
+        if include_prerequisites:
+            added_any = True
+            while added_any:
+                added_any = False
+                for m in list(selected_by_name.values()):
+                    for req in m.prerequisites:
+                        if req not in selected_by_name and req in applicable_by_name:
+                            selected_by_name[req] = applicable_by_name[req]
+                            added_any = True
+
+        removed_any = True
+        while removed_any:
+            removed_any = False
+            for name, m in list(selected_by_name.items()):
+                missing = [req for req in m.prerequisites if req not in selected_by_name]
+                if missing:
+                    logger.warning(
+                        "Dropping module '%s' from plan: prerequisite(s) %s cannot be met",
+                        name,
+                        missing,
+                    )
+                    del selected_by_name[name]
+                    removed_any = True
+
+        return list(selected_by_name.values())
+
+    def _resolve_module_priority(
+        self,
+        mod: ScanModule,
+        fp: TargetFingerprint,
+        tci_score: float,
+        overrides: dict[str, Any],
+    ) -> PriorityTier:
+        override = overrides.get("priority")
+        if override is not None:
+            if isinstance(override, PriorityTier):
+                return override
+            try:
+                return PriorityTier(str(override).lower())
+            except ValueError:
+                logger.warning("Invalid priority override '%s' for module '%s'", override, mod.name)
+        elevated_str = self._elevate_priority(mod, fp, tci_score)
+        try:
+            return PriorityTier(elevated_str.lower())
+        except ValueError:
+            return mod.default_priority
+
+    def _filter_by_scan_mode(
+        self,
+        candidates: list[ScanModule],
+        scan_mode: str,
+        effective_priorities: dict[str, PriorityTier] | None = None,
+    ) -> list[ScanModule]:
+        mode = scan_mode.lower().strip()
+
+        def get_p(m: ScanModule) -> PriorityTier:
+            if effective_priorities and m.name in effective_priorities:
+                return effective_priorities[m.name]
+            return m.default_priority
+
+        if mode == "quick":
+            return [m for m in candidates if get_p(m) in (PriorityTier.CRITICAL, PriorityTier.HIGH)]
+        if mode == "standard":
+            return [
+                m
+                for m in candidates
+                if get_p(m) 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]],
+        effective_priorities: dict[str, PriorityTier] | None = None,
+    ) -> 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, {})
+
+            if effective_priorities and mod.name in effective_priorities:
+                priority = effective_priorities[mod.name].value.lower()
+            else:
+                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.
+
+        Detects circular dependencies and reports them clearly with ValueError.
+        """
+        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()
+        visiting: set[str] = set()
+
+        def place(m: ScanModule) -> bool:
+            if m.name in placed:
+                return True
+            if m.name in visiting:
+                msg = f"Circular prerequisite dependency detected involving module '{m.name}'"
+                raise ValueError(msg)
+            visiting.add(m.name)
+            try:
+                for req in m.prerequisites:
+                    if req not in by_name:
+                        logger.warning(
+                            "Cannot place module '%s': prerequisite '%s' cannot be met",
+                            m.name,
+                            req,
+                        )
+                        return False
+                    if not place(by_name[req]):
+                        return False
+                ordered.append(m)
+                placed.add(m.name)
+                return True
+            finally:
+                visiting.remove(m.name)
+
+        for mod in sorted_mods:
+            place(mod)
+
+        return ordered

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",
+]

strix/core/tci.py

@@ -0,0 +1,683 @@
+"""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__)
+
+
+SUPPORTED_TARGET_TYPES: frozenset[str] = frozenset({"web_app", "api", "whitebox", "infra", "cloud"})
+
+TARGET_TYPE_ALIASES: dict[str, str] = {
+    # web
+    "web_app": "web_app",
+    "web_application": "web_app",
+    "web": "web_app",
+    "url": "web_app",
+    "domain": "web_app",
+    "hostname": "web_app",
+    # api
+    "api": "api",
+    "api_spec": "api",
+    "openapi": "api",
+    "swagger": "api",
+    "rest": "api",
+    "graphql": "api",
+    # whitebox / code
+    "whitebox": "whitebox",
+    "local_code": "whitebox",
+    "code": "whitebox",
+    "source_code": "whitebox",
+    "repository": "whitebox",
+    "repo": "whitebox",
+    # infra
+    "infra": "infra",
+    "infrastructure": "infra",
+    "network": "infra",
+    "ip": "infra",
+    "ip_address": "infra",
+    "host": "infra",
+    # cloud
+    "cloud": "cloud",
+    "aws": "cloud",
+    "azure": "cloud",
+    "gcp": "cloud",
+}
+
+
+def normalize_target_type(target_type: str | None) -> str:
+    """Normalize and validate a target type string to canonical form.
+
+    Raises:
+        ValueError: If target_type is unrecognized.
+    """
+    if not target_type:
+        return "web_app"
+    clean = target_type.strip().lower()
+    canonical = TARGET_TYPE_ALIASES.get(clean)
+    if canonical is None:
+        valid_options = ", ".join(sorted(SUPPORTED_TARGET_TYPES))
+        msg = f"Unknown target type '{target_type}'. Supported canonical types: {valid_options}."
+        raise ValueError(msg)
+    return canonical
+
+
+def _dedup_clean_strings(values: list[str]) -> list[str]:
+    """Clean and deduplicate list of strings preserving insertion order."""
+    seen: set[str] = set()
+    result: list[str] = []
+    for item in values:
+        if isinstance(item, str):
+            clean = item.strip().lower()
+            if clean and clean not in seen:
+                seen.add(clean)
+                result.append(clean)
+    return result
+
+
+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
+
+    @field_validator("api_types", mode="after")
+    @classmethod
+    def _dedup_api_types(cls, v: list[str]) -> list[str]:
+        return _dedup_clean_strings(v)
+
+
+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("target_type", mode="after")
+    @classmethod
+    def _validate_target_type(cls, v: str) -> str:
+        return normalize_target_type(v)
+
+    @field_validator("open_ports", mode="after")
+    @classmethod
+    def _dedup_ports(cls, v: list[int]) -> list[int]:
+        return sorted({p for p in v if isinstance(p, int) and p > 0})
+
+    @field_validator(
+        "protocols",
+        "auth_patterns",
+        "frontend_technologies",
+        "backend_technologies",
+        "waf_cdn",
+        mode="after",
+    )
+    @classmethod
+    def _dedup_string_lists(cls, v: list[str]) -> list[str]:
+        return _dedup_clean_strings(v)
+
+    @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 = sorted({p for p in ports if isinstance(p, int) and p > 0})
+    clean_protocols = _dedup_clean_strings(protocols)
+
+    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 = _dedup_clean_strings(auth_patterns)
+    if not clean_patterns:
+        return 0.0, "No authentication mechanisms detected"
+
+    max_base = 0.0
+    matched: list[str] = []
+    bonus = 0.0
+    has_mfa = False
+    has_rbac = False
+
+    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")):
+            has_mfa = True
+            matched.append("mfa")
+        elif any(k in pattern for k in ("rbac", "abac", "multi_role", "tenant", "multi_tenant")):
+            has_rbac = True
+            matched.append("rbac/multi-tenant")
+        else:
+            max_base = max(max_base, 30.0)
+            matched.append(pattern)
+
+    if has_mfa:
+        bonus += 15.0
+    if has_rbac:
+        bonus += 15.0
+
+    # 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] = []
+    clean_api_types = _dedup_clean_strings(surface.api_types)
+    base_score, count_note = _score_endpoint_count(surface.endpoints_count, bool(clean_api_types))
+    if count_note:
+        notes.append(count_note)
+
+    score = base_score
+
+    # API Types diversity (deduplicated)
+    for atype in clean_api_types:
+        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 = _dedup_clean_strings(frontend)
+    clean_be = _dedup_clean_strings(backend)
+
+    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 = list(
+        dict.fromkeys(
+            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 = list(dict.fromkeys(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 = _dedup_clean_strings(waf_cdn)
+    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)

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",
+]

strix/modules/registry.py

@@ -0,0 +1,502 @@
+"""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, normalize_target_type
+
+
+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
+    requires_oauth: 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)
+        )
+        canonical_target_type = normalize_target_type(fp.target_type)
+
+        with self._lock:
+            candidates = list(self._modules.values())
+
+        applicable: list[ScanModule] = []
+        has_auth = bool(fp.auth_patterns)
+        has_jwt = any("jwt" in p.lower() for p in fp.auth_patterns)
+        has_oauth = any(
+            any(tok in p.lower() for tok in ("oauth", "oauth2", "oidc", "sso", "saml"))
+            for p in fp.auth_patterns
+        )
+        has_api = (
+            canonical_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 (
+                canonical_target_type
+                and mod.target_types
+                and canonical_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_oauth and not has_oauth:
+                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,
+                requires_oauth=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()

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.

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",
+]

strix/tools/planner/tools.py

@@ -0,0 +1,194 @@
+"""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: Any) -> dict[str, Any]:
+    """Parse JSON string or dict into fingerprint dictionary.
+
+    Raises:
+        ValueError: If raw is invalid JSON.
+        TypeError: If raw is not a JSON string, dict, or parsed value is not a dict.
+    """
+    if isinstance(raw, dict):
+        return raw
+    if isinstance(raw, str):
+        stripped = raw.strip()
+        if not stripped:
+            return {}
+        try:
+            parsed = json.loads(stripped)
+        except json.JSONDecodeError as exc:
+            msg = f"Invalid fingerprint JSON: {exc}"
+            raise ValueError(msg) from exc
+        if not isinstance(parsed, dict):
+            msg = f"Fingerprint JSON must be an object/dict, got {type(parsed).__name__}"
+            raise TypeError(msg)
+        return parsed
+    msg = f"Fingerprint argument must be a JSON string or dict, got {type(raw).__name__}"
+    raise TypeError(msg)
+
+
+@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
+    try:
+        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)
+    except (ValueError, TypeError) as exc:
+        logger.warning("Error calculating target complexity: %s", exc)
+        return json.dumps({"error": str(exc)}, ensure_ascii=False, indent=2)
+
+    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
+    try:
+        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=False, indent=2)
+    except (ValueError, TypeError) as exc:
+        logger.warning("Error creating scan plan: %s", exc)
+        return json.dumps({"error": str(exc)}, ensure_ascii=False, 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)

tests/test_module_registry.py

@@ -0,0 +1,230 @@
+"""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")
+
+
+def test_oauth_audit_requires_oauth_not_just_any_auth() -> None:
+    """Issue 6: oauth-audit must require OAuth/OIDC rather than any generic auth."""
+    reg = ModuleRegistry(populate_defaults=True)
+
+    # Basic auth target: should NOT have oauth-audit
+    fp_basic = TargetFingerprint(
+        target="https://basic.example.com",
+        target_type="web_app",
+        auth_patterns=["basic"],
+    )
+    names_basic = {m.name for m in reg.find_applicable(fp_basic)}
+    assert "oauth-audit" not in names_basic
+    assert "auth-enum" in names_basic
+
+    # Session cookie target: should NOT have oauth-audit
+    fp_session = TargetFingerprint(
+        target="https://session.example.com",
+        target_type="web_app",
+        auth_patterns=["session_cookie"],
+    )
+    names_session = {m.name for m in reg.find_applicable(fp_session)}
+    assert "oauth-audit" not in names_session
+    assert "auth-enum" in names_session
+
+    # OAuth2 target: SHOULD have oauth-audit
+    fp_oauth = TargetFingerprint(
+        target="https://oauth.example.com",
+        target_type="web_app",
+        auth_patterns=["oauth2"],
+    )
+    names_oauth = {m.name for m in reg.find_applicable(fp_oauth)}
+    assert "oauth-audit" in names_oauth
+
+    # OIDC target: SHOULD have oauth-audit
+    fp_oidc = TargetFingerprint(
+        target="https://oidc.example.com",
+        target_type="web_app",
+        auth_patterns=["oidc"],
+    )
+    names_oidc = {m.name for m in reg.find_applicable(fp_oidc)}
+    assert "oauth-audit" in names_oidc
+
+
+def test_find_applicable_maps_target_type_aliases_and_rejects_unknown() -> None:
+    """Issue 3: find_applicable accepts existing target type aliases and rejects unknown types."""
+    reg = ModuleRegistry(populate_defaults=True)
+
+    # Existing configuration uses web_application
+    fp_web_app = {"target": "https://app.example.com", "target_type": "web_application"}
+    modules_web = reg.find_applicable(fp_web_app)
+    assert len(modules_web) > 0
+    web_names = {m.name for m in modules_web}
+    assert "sqli-fuzz" in web_names
+    assert "xss-test" in web_names
+
+    # Existing configuration uses api_spec
+    fp_api_spec = {
+        "target": "https://api.example.com",
+        "target_type": "api_spec",
+        "api_surface": {"endpoints_count": 5},
+    }
+    modules_api = reg.find_applicable(fp_api_spec)
+    assert len(modules_api) > 0
+    api_names = {m.name for m in modules_api}
+    assert "api-route-fuzz" in api_names
+
+    # Existing configuration uses local_code
+    fp_code = {"target": "/src/repo", "target_type": "local_code"}
+    modules_code = reg.find_applicable(fp_code)
+    assert len(modules_code) > 0
+    code_names = {m.name for m in modules_code}
+    assert "traversal-test" in code_names
+
+    # Reject unknown target types with explicit ValueError
+    with pytest.raises(ValueError, match="Unknown target type 'mainframe_cobol'"):
+        reg.find_applicable({"target": "host", "target_type": "mainframe_cobol"})

tests/test_planner_tools.py

@@ -0,0 +1,148 @@
+"""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"
+
+
+async def test_tools_reject_malformed_fingerprint_json() -> None:
+    """Issue 4: malformed JSON returns an explicit error payload."""
+    # 1. Malformed JSON to calculate_target_complexity
+    raw_calc = await calculate_target_complexity.on_invoke_tool(
+        _make_ctx("calculate_target_complexity"),
+        json.dumps({"target": "https://test.local", "fingerprint": "{malformed json"}),
+    )
+    res_calc = json.loads(raw_calc)
+    assert "error" in res_calc
+    assert "Invalid fingerprint JSON" in res_calc["error"]
+
+    # 2. Malformed JSON to create_scan_plan
+    raw_plan = await create_scan_plan.on_invoke_tool(
+        _make_ctx("create_scan_plan"),
+        json.dumps({"target": "https://test.local", "fingerprint": "{malformed json"}),
+    )
+    res_plan = json.loads(raw_plan)
+    assert "error" in res_plan
+    assert "Invalid fingerprint JSON" in res_plan["error"]
+
+    # 3. Non-dict JSON (e.g. array) to calculate_target_complexity
+    raw_arr = await calculate_target_complexity.on_invoke_tool(
+        _make_ctx("calculate_target_complexity"),
+        json.dumps({"target": "https://test.local", "fingerprint": "[1, 2, 3]"}),
+    )
+    res_arr = json.loads(raw_arr)
+    assert "error" in res_arr
+    assert "must be an object" in res_arr["error"]
+
+    # 4. Unknown target type returns explicit error
+    raw_unk = await calculate_target_complexity.on_invoke_tool(
+        _make_ctx("calculate_target_complexity"),
+        json.dumps(
+            {"target": "https://test.local", "fingerprint": json.dumps({"target_type": "quantum"})}
+        ),
+    )
+    res_unk = json.loads(raw_unk)
+    assert "error" in res_unk
+    assert "Unknown target type" in res_unk["error"]

tests/test_scan_planner.py

@@ -0,0 +1,320 @@
+"""Tests for adaptive ScanPlanner and plan generation from TCI."""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from strix.agents.planner import ScanPlan, ScanPlanner
+from strix.core.tci import ApiSurface, TargetFingerprint, calculate_tci
+from strix.modules.registry import PriorityTier, ScanModule
+
+
+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
+
+
+def test_priority_filter_prerequisite_resolution() -> None:
+    """Issue 2: priority_filter keeps business-logic and includes auth-enum prerequisite."""
+    planner = ScanPlanner()
+    target = "https://app.local"
+    # Target with low complexity so business-logic is medium priority
+    fp = TargetFingerprint(
+        target=target,
+        auth_patterns=["session_cookie"],
+        data_sensitivity="low",
+    )
+
+    # By default, include_prerequisites=True: auth-enum is included before business-logic
+    plan_with_prereq = planner.generate_plan(
+        target,
+        fp,
+        priority_filter=["medium"],
+        include_prerequisites=True,
+    )
+    names = plan_with_prereq.module_names()
+    assert "business-logic" in names
+    assert "auth-enum" in names
+    assert names.index("auth-enum") < names.index("business-logic")
+
+    # With include_prerequisites=False: business-logic dropped when requirement cannot be met
+    plan_without_prereq = planner.generate_plan(
+        target,
+        fp,
+        priority_filter=["medium"],
+        include_prerequisites=False,
+    )
+    assert "business-logic" not in plan_without_prereq.module_names()
+
+
+def test_priority_elevation_before_mode_filtering() -> None:
+    """Issue 5: high TCI targets elevate business-logic before quick mode filtering."""
+    planner = ScanPlanner()
+    target = "https://complex.example.com"
+    # High complexity target (TCI >= 70)
+    fp_complex = TargetFingerprint(
+        target=target,
+        open_ports=[443, 8080, 8443, 5432],
+        protocols=["https", "grpc"],
+        auth_patterns=["jwt", "oauth2", "mfa", "rbac"],
+        api_surface=ApiSurface(endpoints_count=60, api_types=["rest", "graphql"], has_spec=True),
+        backend_technologies=["fastapi", "spring", "postgres", "redis"],
+        waf_cdn=["cloudflare_waf", "aws_waf"],
+        data_sensitivity="financial",
+        patch_posture="legacy",
+    )
+    tci = calculate_tci(fp_complex).score
+    assert tci >= 70.0
+
+    # Quick mode keeps business-logic because its final effective priority was elevated to high
+    plan_quick = planner.generate_plan(target, fp_complex, scan_mode="quick")
+    names = plan_quick.module_names()
+    assert "business-logic" in names
+    step = plan_quick.get_step_by_module("business-logic")
+    assert step is not None
+    assert step.priority == "high"
+
+    # User override elevates a medium module so quick mode keeps it
+    plan_override = planner.generate_plan(
+        target,
+        fp_complex,
+        scan_mode="quick",
+        module_overrides={"cors-misconfig": {"priority": "high"}},
+    )
+    assert "cors-misconfig" in plan_override.module_names()
+
+
+def test_circular_prerequisite_detection() -> None:
+    """Issue 9: circular prerequisites report ValueError clearly instead of RecursionError."""
+    planner = ScanPlanner()
+
+    # 2-module cycle: A -> B -> A
+    mod_a = ScanModule(
+        name="cycle-a",
+        description="Module A",
+        category="injection",
+        default_priority=PriorityTier.MEDIUM,
+        prerequisites=["cycle-b"],
+    )
+    mod_b = ScanModule(
+        name="cycle-b",
+        description="Module B",
+        category="injection",
+        default_priority=PriorityTier.MEDIUM,
+        prerequisites=["cycle-a"],
+    )
+
+    with pytest.raises(ValueError, match="Circular prerequisite dependency detected"):
+        planner._order_modules([mod_a, mod_b])
+
+    # 3-module cycle: X -> Y -> Z -> X
+    mod_x = ScanModule(
+        name="cycle-x",
+        description="Module X",
+        category="api",
+        prerequisites=["cycle-y"],
+    )
+    mod_y = ScanModule(
+        name="cycle-y",
+        description="Module Y",
+        category="api",
+        prerequisites=["cycle-z"],
+    )
+    mod_z = ScanModule(
+        name="cycle-z",
+        description="Module Z",
+        category="api",
+        prerequisites=["cycle-x"],
+    )
+
+    with pytest.raises(ValueError, match="Circular prerequisite dependency detected"):
+        planner._order_modules([mod_x, mod_y, mod_z])

tests/test_tci.py

@@ -0,0 +1,258 @@
+"""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,
+    normalize_target_type,
+)
+
+
+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
+
+
+def test_deduplication_prevents_inflated_scores() -> None:
+    """Issue 8: repeated observations must not inflate TCI score."""
+    # API types deduplication
+    score_dupe, _ = calc_api_surface_score(
+        ApiSurface(endpoints_count=0, api_types=["graphql", "graphql"])
+    )
+    score_single, _ = calc_api_surface_score(ApiSurface(endpoints_count=0, api_types=["graphql"]))
+    assert score_dupe == 35.0
+    assert score_dupe == score_single
+
+    # Ports & protocols deduplication
+    s_proto_dupe, _ = calc_ports_protocols_score([443, 443], ["grpc", "grpc"])
+    s_proto_single, _ = calc_ports_protocols_score([443], ["grpc"])
+    assert s_proto_dupe == s_proto_single
+
+    # Auth pattern MFA & repeated deduplication
+    s_mfa_dupe, _ = calc_auth_score(["basic", "mfa", "totp", "mfa"])
+    s_mfa_single, _ = calc_auth_score(["basic", "mfa"])
+    assert s_mfa_dupe == s_mfa_single
+
+    # Tech stack deduplication (no false polyglot bonus)
+    s_stack_dupe, _ = calc_tech_stack_score(["react", "react"], ["django", "django"])
+    s_stack_single, _ = calc_tech_stack_score(["react"], ["django"])
+    assert s_stack_dupe == s_stack_single
+
+    # WAF deduplication (no multi-WAF bonus for repeated provider)
+    s_waf_dupe, _ = calc_waf_defense_score(["cloudflare", "cloudflare"])
+    s_waf_single, _ = calc_waf_defense_score(["cloudflare"])
+    assert s_waf_dupe == 75.0
+    assert s_waf_dupe == s_waf_single
+
+
+def test_target_type_normalization_and_validation() -> None:
+    """Issue 3: target types map aliases and reject unknown types."""
+    assert normalize_target_type("web_application") == "web_app"
+    assert normalize_target_type("api_spec") == "api"
+    assert normalize_target_type("local_code") == "whitebox"
+    assert normalize_target_type("ip_address") == "infra"
+
+    # Fingerprint model normalizes on construction
+    fp = TargetFingerprint(target_type="web_application")
+    assert fp.target_type == "web_app"
+
+    # Reject unknown target types
+    with pytest.raises(ValueError, match="Unknown target type 'quantum_satellite'"):
+        normalize_target_type("quantum_satellite")
+
+    with pytest.raises(ValueError, match="Unknown target type 'quantum_satellite'"):
+        TargetFingerprint(target_type="quantum_satellite")