Add evidence-backed zero-day hypothesis engine

tests/test_zero_day.py

@@ -0,0 +1,86 @@
+"""Tests for evidence-backed zero-day hypothesis generation."""
+
+from wraith_cli.knowledge.zero_day import SEVERITY_WEIGHT, ZeroDayHypothesisEngine
+
+
+def _recon_data():
+    return {
+        "technologies": ["FastAPI"],
+        "languages": {"Python": 8},
+        "dependencies": [
+            {"name": "openai", "version": "1.0", "ecosystem": "pypi"},
+            {"name": "redis", "version": "5", "ecosystem": "pypi"},
+            {"name": "pyyaml", "version": "6", "ecosystem": "pypi"},
+        ],
+        "entry_points": [
+            {
+                "file": "app/api/chat.py",
+                "line": 12,
+                "pattern": "@router.post",
+                "code": "async def chat(message: str):",
+            },
+            {
+                "file": "app/api/upload.py",
+                "line": 28,
+                "pattern": "@router.post",
+                "code": "async def upload_yaml(file):",
+            },
+        ],
+        "sensitive_files": ["app/auth.py", "app/api/chat.py", "app/cache/session_cache.py"],
+        "trust_boundaries": ["user prompt to llm tool router", "tenant session cache"],
+    }
+
+
+def test_generates_ranked_evidence_backed_hypotheses():
+    engine = ZeroDayHypothesisEngine()
+    findings = [
+        {
+            "title": "Prompt injection can influence tool selection",
+            "vuln_class": "prompt_injection",
+            "severity": "high",
+            "confidence": 0.8,
+        }
+    ]
+
+    hypotheses = engine.generate(_recon_data(), findings, limit=5)
+
+    assert hypotheses
+    assert hypotheses == sorted(
+        hypotheses,
+        key=lambda item: (
+            item["rigor_score"],
+            item["novelty_score"],
+            SEVERITY_WEIGHT[item["severity"]],
+        ),
+        reverse=True,
+    )
+    top = hypotheses[0]
+    assert top["type"] == "zero_day_hypothesis"
+    assert top["hypothesis_status"] == "speculative_requires_validation"
+    assert top["evidence"]
+    assert 0.0 <= top["rigor_score"] <= 1.0
+    assert 0.0 <= top["novelty_score"] <= 1.0
+    assert top["validation_safety"]["environment"].startswith("authorised")
+    assert top["negative_controls"]
+
+
+def test_llm_hypotheses_are_normalized_and_safely_bounded():
+    engine = ZeroDayHypothesisEngine()
+    normalized = engine.normalize_llm_hypotheses([
+        {
+            "title": "Speculative Boundary Bypass",
+            "novel_class": "boundary_bypass",
+            "estimated_severity": "critical",
+            "confidence": 0.99,
+            "validation_experiment": "Use benign fixtures to compare policy decisions.",
+            "evidence": "GraphQL route and auth finding overlap.",
+        }
+    ])
+
+    assert len(normalized) == 1
+    hypothesis = normalized[0]
+    assert hypothesis["source"] == "zero_day_llm"
+    assert hypothesis["confidence"] == 0.65
+    assert hypothesis["severity"] == "critical"
+    assert hypothesis["evidence"] == ["GraphQL route and auth finding overlap."]
+    assert "benign" in hypothesis["validation_safety"]["payload_policy"]

wraith_cli/agents/zero_day.py

@@ -11,6 +11,7 @@
 from typing import Any
 
 from wraith_cli.agents.base import BaseAgent, AgentResult
+from wraith_cli.knowledge.zero_day import ZeroDayHypothesisEngine
 
 
 class ZeroDayAgent(BaseAgent):
@@ -19,6 +20,10 @@ class ZeroDayAgent(BaseAgent):
     name = "zero_day"
     description = "Generates zero-day vulnerability hypotheses from CVE evolutionary analysis"
 
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        self.hypothesis_engine = ZeroDayHypothesisEngine()
+
     async def execute(self, task: dict[str, Any], context: dict[str, Any]) -> AgentResult:
         chain = self.create_chain()
         recon_data = context.get("recon_data", {})
@@ -28,9 +33,15 @@ async def execute(self, task: dict[str, Any], context: dict[str, Any]) -> AgentR
         chain.observe(f"Technology stack: {recon_data.get('technologies', [])}")
         chain.observe(f"Existing findings: {len(existing_findings)}")
 
+        deterministic = self.hypothesis_engine.generate(recon_data, existing_findings, limit=8)
+        chain.infer(
+            f"Offline hypothesis engine generated {len(deterministic)} evidence-backed hypotheses"
+        )
+
+        llm_hypotheses: list[dict[str, Any]] = []
         try:
             result = await self._call_llm_json(
-                system=(
+                system_prompt=(
                     "You are a vulnerability researcher specialising in zero-day discovery. "
                     "Based on the target's technology stack and known vulnerability patterns, "
                     "generate hypotheses about NOVEL vulnerability classes that haven't been "
@@ -39,9 +50,11 @@ async def execute(self, task: dict[str, Any], context: dict[str, Any]) -> AgentR
                     "2. Project evolutionary trajectories\n"
                     "3. Imagine novel attack vectors at component boundaries\n"
                     "4. Consider emerging attack surfaces (LLM integration, supply chain, etc.)\n"
-                    "For each hypothesis, specify a validation experiment."
+                    "For each hypothesis, specify a validation experiment. Keep all validation "
+                    "experiments defensive: authorised lab tests, benign canaries, property tests, "
+                    "or local fixtures only."
                 ),
-                user=(
+                user_prompt=(
                     f"Technologies: {recon_data.get('technologies', [])}\n"
                     f"Languages: {recon_data.get('languages', {})}\n"
                     f"Dependencies: {json.dumps(recon_data.get('dependencies', [])[:15], indent=2)}\n"
@@ -53,41 +66,70 @@ async def execute(self, task: dict[str, Any], context: dict[str, Any]) -> AgentR
                     '- "affected_components": which components are at risk\n'
                     '- "evolutionary_basis": which existing CVE patterns led to this prediction\n'
                     '- "validation_experiment": how to test this hypothesis\n'
+                    '- "evidence": target-specific evidence signals, not generic warnings\n'
+                    '- "negative_controls": validation cases that should stay safe\n'
                     '- "estimated_severity": critical/high/medium\n'
                     '- "confidence": 0.0-1.0 (be honest — these are speculative)\n'
                 ),
             )
 
-            hypotheses = result.get("hypotheses", [])
-            chain.infer(f"Generated {len(hypotheses)} zero-day hypotheses")
-
-            findings = []
-            for h in hypotheses:
-                findings.append({
-                    "type": "zero_day_hypothesis",
-                    "vuln_class": h.get("novel_class", "unknown"),
-                    "title": f"[Hypothesis] {h.get('title', 'Unknown')}",
-                    "severity": h.get("estimated_severity", "medium"),
-                    "description": h.get("description", ""),
-                    "validation": h.get("validation_experiment", ""),
-                    "confidence": h.get("confidence", 0.2),
-                    "source": "zero_day",
-                })
-
-            chain.conclude(
-                f"Zero-day hypothesis generation complete: {len(findings)} hypotheses",
-                confidence=0.6,
+            raw_hypotheses = result.get("hypotheses", [])
+            existing_titles = {h.get("title", "") for h in deterministic}
+            llm_hypotheses = self.hypothesis_engine.normalize_llm_hypotheses(
+                raw_hypotheses,
+                existing_titles=existing_titles,
             )
+            chain.infer(f"LLM generated {len(llm_hypotheses)} normalized hypotheses")
+        except Exception as e:
+            chain.assume(f"LLM hypothesis expansion unavailable: {e}")
 
-            self.publish("zero_day.complete", {"hypotheses": findings})
-            return AgentResult(
-                agent_name=self.name, agent_id=self.agent_id,
-                success=True, findings=findings, reasoning_chain=chain,
-            )
+        findings = self._merge_and_rank(deterministic, llm_hypotheses)
 
-        except Exception as e:
-            chain.conclude(f"Hypothesis generation failed: {e}", confidence=0.2)
-            return AgentResult(
-                agent_name=self.name, agent_id=self.agent_id,
-                success=False, errors=[str(e)], reasoning_chain=chain,
+        chain.conclude(
+            f"Zero-day hypothesis generation complete: {len(findings)} hypotheses",
+            confidence=0.72 if deterministic else 0.35,
+        )
+
+        self.publish("zero_day.complete", {"hypotheses": findings})
+        return AgentResult(
+            agent_name=self.name,
+            agent_id=self.agent_id,
+            success=True,
+            findings=findings,
+            data={
+                "offline_hypotheses": len(deterministic),
+                "llm_hypotheses": len(llm_hypotheses),
+                "methodology": "evidence_weighted_hypothesis_generation",
+            },
+            reasoning_chain=chain,
+        )
+
+    def _merge_and_rank(
+        self,
+        deterministic: list[dict[str, Any]],
+        llm_hypotheses: list[dict[str, Any]],
+    ) -> list[dict[str, Any]]:
+        """Deduplicate and rank hypotheses by rigor, novelty, then confidence."""
+        merged: list[dict[str, Any]] = []
+        seen: set[tuple[str, str]] = set()
+        for hypothesis in deterministic + llm_hypotheses:
+            key = (
+                str(hypothesis.get("vuln_class", "")).lower(),
+                str(hypothesis.get("title", "")).lower(),
             )
+            if key in seen:
+                continue
+            seen.add(key)
+            merged.append(hypothesis)
+
+        merged.sort(
+            key=lambda item: (
+                item.get("rigor_score", 0.0),
+                item.get("novelty_score", 0.0),
+                item.get("confidence", 0.0),
+            ),
+            reverse=True,
+        )
+        for index, hypothesis in enumerate(merged, 1):
+            hypothesis.setdefault("id", f"ZD-HYP-{index:03d}")
+        return merged

wraith_cli/knowledge/__init__.py

@@ -1,4 +1,5 @@
 from wraith_cli.knowledge.cve_db import CVEDatabase
 from wraith_cli.knowledge.patterns import VulnPatterns
+from wraith_cli.knowledge.zero_day import ZeroDayHypothesisEngine
 
-__all__ = ["CVEDatabase", "VulnPatterns"]
+__all__ = ["CVEDatabase", "VulnPatterns", "ZeroDayHypothesisEngine"]