From bc949d7bfe6bd5b991275c3b15bf3e0675dc4616 Mon Sep 17 00:00:00 2001 From: MriBoyd Date: Wed, 14 Oct 2026 12:00:24 +0300 Subject: [PATCH] fix(planner): resolve prerequisite handling, target aliases, cycle detection, and TCI scoring - Issue 2: Automatically include missing prerequisites from applicable modules or drop dependent modules whose requirements cannot be met - Issue 3: Map target type aliases (web_application, api_spec, local_code) and explicitly reject unknown target types with ValueError - Issue 4: Return explicit input error on malformed or non-object fingerprint JSON - Issue 5: Calculate final effective priorities before scan-mode and priority filtering - Issue 6: Add requires_oauth condition to oauth-audit so non-OAuth targets skip it - Issue 8: Deduplicate observations in TCI scoring to prevent artificial inflation - Issue 9: Detect circular prerequisites during topological ordering and raise ValueError --- strix/agents/planner.py | 164 +++++++++++++++++++++++++++------- strix/core/tci.py | 139 ++++++++++++++++++++++++---- strix/modules/registry.py | 19 +++- strix/tools/planner/tools.py | 63 ++++++++----- tests/test_module_registry.py | 78 ++++++++++++++++ tests/test_planner_tools.py | 41 +++++++++ tests/test_scan_planner.py | 122 ++++++++++++++++++++++++- tests/test_tci.py | 52 +++++++++++ 8 files changed, 605 insertions(+), 73 deletions(-) diff --git a/strix/agents/planner.py b/strix/agents/planner.py index 1a723923..d6e6a75b 100644 --- a/strix/agents/planner.py +++ b/strix/agents/planner.py @@ -142,6 +142,7 @@ class ScanPlanner: 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. @@ -153,6 +154,9 @@ class ScanPlanner: 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. @@ -160,7 +164,7 @@ class ScanPlanner: fp = ( fingerprint if isinstance(fingerprint, TargetFingerprint) - else TargetFingerprint(**fingerprint) + else TargetFingerprint(**(fingerprint or {})) ) if not fp.target and target: fp.target = target @@ -171,26 +175,45 @@ class ScanPlanner: tci_level = tci_res.level.value # 2. Select applicable modules based on target context - candidates = self.registry.find_applicable(fp) + applicable = self.registry.find_applicable(fp) + applicable_by_name = {m.name: m for m in applicable} + overrides_dict = module_overrides or {} - # 3. Filter candidates by scan_mode - candidates = self._filter_by_scan_mode(candidates, scan_mode) + # 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 by priority if requested + # 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} - candidates = [m for m in candidates if m.default_priority.value.lower() in allowed_p] + selected = [ + m + for m in selected + if effective_priorities.get(m.name, m.default_priority).value.lower() in allowed_p + ] - # 5. Build and adapt plan steps + # 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=module_overrides or {}, + module_overrides=overrides_dict, + effective_priorities=effective_priorities, ) - # 6. Apply max_steps limit if specified + # 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") @@ -216,24 +239,82 @@ class ScanPlanner: }, ) + 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 m.default_priority in (PriorityTier.CRITICAL, PriorityTier.HIGH) - ] + 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 m.default_priority - in (PriorityTier.CRITICAL, PriorityTier.HIGH, PriorityTier.MEDIUM) + if get_p(m) in (PriorityTier.CRITICAL, PriorityTier.HIGH, PriorityTier.MEDIUM) ] return candidates @@ -290,6 +371,7 @@ class ScanPlanner: 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] = [] @@ -297,12 +379,15 @@ class ScanPlanner: for idx, mod in enumerate(ordered_modules, start=1): overrides = module_overrides.get(mod.name, {}) - priority_val = overrides.get("priority") - priority = ( - str(priority_val).lower() - if priority_val is not None - else self._elevate_priority(mod, fp, tci_score) - ) + 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) @@ -355,7 +440,10 @@ class ScanPlanner: return "; ".join(notes) def _order_modules(self, modules: list[ScanModule]) -> list[ScanModule]: - """Order modules tactically by phase and prerequisite dependencies.""" + """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]: @@ -368,15 +456,31 @@ class ScanPlanner: ordered: list[ScanModule] = [] placed: set[str] = set() + visiting: set[str] = set() - def place(m: ScanModule) -> None: + def place(m: ScanModule) -> bool: if m.name in placed: - return - for req in m.prerequisites: - if req in by_name and req not in placed: - place(by_name[req]) - ordered.append(m) - placed.add(m.name) + 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) diff --git a/strix/core/tci.py b/strix/core/tci.py index 25c378f4..34069e3c 100644 --- a/strix/core/tci.py +++ b/strix/core/tci.py @@ -18,6 +18,75 @@ 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.""" @@ -38,6 +107,11 @@ class ApiSurface(BaseModel): 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.""" @@ -57,6 +131,28 @@ class TargetFingerprint(BaseModel): 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: @@ -168,8 +264,8 @@ def calc_ports_protocols_score(ports: list[int], protocols: list[str]) -> tuple[ score = 0.0 notes: list[str] = [] - clean_ports = list(dict.fromkeys(p for p in ports if isinstance(p, int) and p > 0)) - clean_protocols = [p.lower().strip() for p in protocols if isinstance(p, str) and p.strip()] + 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)" @@ -245,13 +341,15 @@ _AUTH_BASE_SCORES: dict[str, float] = { def calc_auth_score(auth_patterns: list[str]) -> tuple[float, str]: """Score authentication complexity (0..100).""" - clean_patterns = [p.lower().strip() for p in auth_patterns if isinstance(p, str) and p.strip()] + 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) @@ -259,15 +357,20 @@ def calc_auth_score(auth_patterns: list[str]) -> tuple[float, str]: max_base = max(max_base, score) matched.append(pattern) elif any(k in pattern for k in ("mfa", "2fa", "totp")): - bonus += 15.0 + has_mfa = True matched.append("mfa") elif any(k in pattern for k in ("rbac", "abac", "multi_role", "tenant", "multi_tenant")): - bonus += 15.0 + 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: @@ -303,15 +406,15 @@ def calc_api_surface_score(api_surface: ApiSurface | dict[str, Any]) -> tuple[fl surface = ApiSurface(**api_surface) if isinstance(api_surface, dict) else api_surface notes: list[str] = [] - base_score, count_note = _score_endpoint_count(surface.endpoints_count, bool(surface.api_types)) + 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 - for api_type in surface.api_types: - atype = api_type.lower().strip() + # API Types diversity (deduplicated) + for atype in clean_api_types: if atype in ("graphql", "grpc"): score += 20.0 notes.append(atype.upper()) @@ -393,8 +496,8 @@ def calc_tech_stack_score( backend: list[str], ) -> tuple[float, str]: """Score technology stack heterogeneity and modern complexity (0..100).""" - clean_fe = [f.lower().strip() for f in frontend if isinstance(f, str) and f.strip()] - clean_be = [b.lower().strip() for b in backend if isinstance(b, str) and b.strip()] + clean_fe = _dedup_clean_strings(frontend) + clean_be = _dedup_clean_strings(backend) score = 15.0 # baseline notes: list[str] = [] @@ -407,9 +510,13 @@ def calc_tech_stack_score( score += 15.0 notes.append(f"Frontend: {clean_fe}") - matched_be = [ - b for b in clean_be if b in _BACKEND_FRAMEWORKS or any(k in b for k in _BACKEND_FRAMEWORKS) - ] + 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}") @@ -420,7 +527,7 @@ def calc_tech_stack_score( score += 15.0 notes.append(f"Backend: {clean_be}") - matched_db = [b for b in clean_be if b in _DB_ENGINES] + 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 @@ -448,7 +555,7 @@ _WAF_PROVIDERS = frozenset( def calc_waf_defense_score(waf_cdn: list[str]) -> tuple[float, str]: """Score defensive posture and WAF presence (0..100).""" - clean_defenses = [d.lower().strip() for d in waf_cdn if isinstance(d, str) and d.strip()] + clean_defenses = _dedup_clean_strings(waf_cdn) if not clean_defenses: return 0.0, "No WAF or edge security detected" diff --git a/strix/modules/registry.py b/strix/modules/registry.py index 35ad8cb6..6f8e08aa 100644 --- a/strix/modules/registry.py +++ b/strix/modules/registry.py @@ -9,7 +9,7 @@ from typing import Any from pydantic import BaseModel, ConfigDict, Field -from strix.core.tci import TargetFingerprint +from strix.core.tci import TargetFingerprint, normalize_target_type logger = logging.getLogger(__name__) @@ -59,6 +59,7 @@ class ScanModule(BaseModel): requires_api: bool = False requires_graphql: bool = False requires_jwt: bool = False + requires_oauth: bool = False class ModuleRegistry: @@ -171,6 +172,7 @@ class ModuleRegistry: if isinstance(fingerprint, TargetFingerprint) else TargetFingerprint(**fingerprint) ) + canonical_target_type = normalize_target_type(fp.target_type) with self._lock: candidates = list(self._modules.values()) @@ -178,8 +180,12 @@ class ModuleRegistry: 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 = ( - fp.target_type == "api" + canonical_target_type == "api" or fp.api_surface.endpoints_count > 0 or bool(fp.api_surface.api_types) ) @@ -189,7 +195,11 @@ class ModuleRegistry: for mod in candidates: # Target type check - if fp.target_type and mod.target_types and fp.target_type not in mod.target_types: + if ( + canonical_target_type + and mod.target_types + and canonical_target_type not in mod.target_types + ): continue # Feature conditions @@ -197,6 +207,8 @@ class ModuleRegistry: 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: @@ -354,6 +366,7 @@ class ModuleRegistry: skills=["protocols/oauth"], tags=["oauth", "oidc", "auth"], requires_auth=True, + requires_oauth=True, ), ScanModule( name="graphql-audit", diff --git a/strix/tools/planner/tools.py b/strix/tools/planner/tools.py index cc1307c6..4d6335ed 100644 --- a/strix/tools/planner/tools.py +++ b/strix/tools/planner/tools.py @@ -16,8 +16,13 @@ from strix.modules.registry import default_registry logger = logging.getLogger(__name__) -def _parse_fingerprint_arg(raw: str | dict[str, Any]) -> dict[str, Any]: - """Parse JSON string or dict into fingerprint dictionary.""" +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): @@ -26,11 +31,15 @@ def _parse_fingerprint_arg(raw: str | dict[str, Any]) -> dict[str, Any]: return {} try: parsed = json.loads(stripped) - if isinstance(parsed, dict): - return parsed - except json.JSONDecodeError: - logger.warning("Failed to parse fingerprint JSON argument: %s", raw) - return {} + 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) @@ -68,12 +77,16 @@ async def calculate_target_complexity( ``sub_scores``, and descriptive ``breakdown``. """ del ctx # Unused by calculation - fp_dict = _parse_fingerprint_arg(fingerprint) - if target and "target" not in fp_dict: - fp_dict["target"] = target + 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) + 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, @@ -124,18 +137,22 @@ async def create_scan_plan( ``` """ del ctx - fp_dict = _parse_fingerprint_arg(fingerprint) - if target and "target" not in fp_dict: - fp_dict["target"] = target + 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) + 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) diff --git a/tests/test_module_registry.py b/tests/test_module_registry.py index fe8544f8..9e015c8f 100644 --- a/tests/test_module_registry.py +++ b/tests/test_module_registry.py @@ -150,3 +150,81 @@ def test_register_module_helper() -> None: 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"}) diff --git a/tests/test_planner_tools.py b/tests/test_planner_tools.py index aca7115b..66d948b5 100644 --- a/tests/test_planner_tools.py +++ b/tests/test_planner_tools.py @@ -105,3 +105,44 @@ async def test_list_scan_modules_tool() -> None: 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"] diff --git a/tests/test_scan_planner.py b/tests/test_scan_planner.py index 0236c3c5..9c77f037 100644 --- a/tests/test_scan_planner.py +++ b/tests/test_scan_planner.py @@ -4,8 +4,11 @@ from __future__ import annotations import json +import pytest + from strix.agents.planner import ScanPlan, ScanPlanner -from strix.core.tci import ApiSurface, TargetFingerprint +from strix.core.tci import ApiSurface, TargetFingerprint, calculate_tci +from strix.modules.registry import PriorityTier, ScanModule def test_planner_generates_exact_payload_schema() -> None: @@ -198,3 +201,120 @@ def test_plan_helpers_and_aggregation() -> None: 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]) diff --git a/tests/test_tci.py b/tests/test_tci.py index d877fd51..7220a92d 100644 --- a/tests/test_tci.py +++ b/tests/test_tci.py @@ -19,6 +19,7 @@ from strix.core.tci import ( calc_tech_stack_score, calc_waf_defense_score, calculate_tci, + normalize_target_type, ) @@ -204,3 +205,54 @@ def test_patch_posture_scoring() -> None: 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")