vulnerability_validator.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Maxwell VOSS - Strict Vulnerability Validator.

3-Phase validation pipeline before any finding reaches the sales pipeline:

  Phase A: Slither JSON -> real dataflow analysis
  Phase B: Context check -> CEI pattern, access control, modifiers
  Phase C: Confidence scoring -> only findings >= 70/100 pass

This module is the QUALITY GATE. A finding that fails this gate
is NEVER sold, NEVER reported as valid, and logged as false positive.
"""

import json
import os
import re
import subprocess
import sys
import time
from datetime import datetime, timezone
from typing import Any

_TOOLS_DIR = os.path.dirname(os.path.abspath(__file__))
_WS = os.environ.get("OPENCLAW_WORKSPACE") or os.path.expanduser("~/.openclaw/workspace")
if _WS not in sys.path:
    sys.path.insert(0, _WS)

# Prepend venv_vectorb/bin to PATH so Slither and solc are found
_venv_bin = os.path.join(_WS, "venv_vectorb", "bin")
if os.path.exists(_venv_bin) and _venv_bin not in os.environ.get("PATH", ""):
    os.environ["PATH"] = _venv_bin + os.pathsep + os.environ.get("PATH", "")

try:
    from lib.anti_hallucination import _log
except ImportError:
    def _log(tag, msg): print(f"[{tag}] {msg}")


# ════════════════════════════════════════════════════
# Slither Detector → Confidence Mapping
# ════════════════════════════════════════════════════

# Maps Slither detector impact/confidence to our confidence score
SLITHER_CONFIDENCE_MAP = {
    # (impact, confidence) → our_score
    ("High",   "High"):   95,
    ("High",   "Medium"): 80,
    ("High",   "Low"):    55,
    ("Medium", "High"):   75,
    ("Medium", "Medium"): 60,
    ("Medium", "Low"):    40,
    ("Low",    "High"):   50,
    ("Low",    "Medium"): 35,
    ("Low",    "Low"):    20,
}

# Slither detectors that are HIGH CONFIDENCE (dataflow-verified)
HIGH_SIGNAL_DETECTORS = {
    "reentrancy-eth",
    "reentrancy-no-eth",
    "reentrancy-unlimited-gas",
    "unprotected-upgrade",
    "arbitrary-send-eth",
    "arbitrary-send-erc20",
    "controlled-delegatecall",
    "delegatecall-loop",
    "suicidal",
    "tx-origin",
    "unchecked-transfer",
    "write-after-write",
    "tautology",
    "boolean-equality",
    "divide-before-multiply",
    "msg-value-loop",
    "uninitialized-local",
    "uninitialized-state",
    "storage-array",
    "array-by-reference",
}

# Detectors that REQUIRE manual review (medium confidence)
MEDIUM_SIGNAL_DETECTORS = {
    "timestamp",
    "weak-prng",
    "controlled-array-length",
    "incorrect-equality",
    "locked-ether",
    "low-level-calls",
    "calls-loop",
    "events-maths",
    "events-access",
    "missing-zero-check",
    "shadowing-state",
    "shadowing-local",
}

# Detectors that are LOW SIGNAL (often false positives in audited code)
LOW_SIGNAL_DETECTORS = {
    "unused-return",
    "dead-code",
    "constable-states",
    "constant-function-asm",
    "constant-function-state",
    "naming-convention",
    "too-many-digits",
    "encode-packed-collision",
    "assembly",
    "missing-inheritance",
}

# Known audited contracts — findings here require extra scrutiny
KNOWN_AUDITED_CONTRACTS = {
    "0x111111125421ca6dc452d289314280a0f8842a65",  # 1inch v6
    "0x1111111254eeb25477b68fb85ed929f73a960582",  # 1inch v5
    "0x68b3465833fb72a70ecdf485e0e4c7bd8665fc45",  # Uniswap Universal Router v1
    "0x66a9893cc07d91d95644aedd05d03f95e1dba8af",  # Uniswap Universal Router v2
    "0xe592427a0aece92de3edee1f18e0157c05861564",  # Uniswap V3 SwapRouter
    "0x0000000071727de22e5e9d8baf0edac6f37da032",  # ERC-4337 EntryPoint
    "0x1231deb6f5749ef6ce6943a275a1d3e7486f4eae",  # LiFi Diamond
    "0x663dc15d3c1ac63ff12e45ab68fea3f0a883c251",  # CrosschainForwarderProxy
}


class VulnerabilityValidator:
    """3-phase strict validator for smart contract vulnerabilities."""

    def __init__(self, workspace: str = None):
        self.workspace = workspace or _WS
        self.contracts_dir = os.path.join(self.workspace, "contracts_to_audit")
        self.reports_dir = os.path.join(self.workspace, "security_reports")
        self.validated_dir = os.path.join(self.workspace, "validated_findings")
        self.rejected_dir = os.path.join(self.workspace, "rejected_findings")
        os.makedirs(self.validated_dir, exist_ok=True)
        os.makedirs(self.rejected_dir, exist_ok=True)

    def _find_contract_sol(self, address: str) -> str | None:
        """Find the .sol file for a given contract address."""
        addr_lower = address.lower()
        candidates = [
            os.path.join(self.contracts_dir, f"{addr_lower}.sol"),
            os.path.join(self.contracts_dir, f"{address}.sol"),
        ]
        for c in candidates:
            if os.path.exists(c):
                return c
        # Search recursively
        if os.path.isdir(self.contracts_dir):
            for f in os.listdir(self.contracts_dir):
                if addr_lower in f.lower() and f.endswith(".sol"):
                    return os.path.join(self.contracts_dir, f)
        return None

    def _run_slither_json(self, sol_path: str) -> dict | None:
        """Run Slither with JSON output. Returns parsed JSON or None on failure."""
        venv_slither = os.path.join(self.workspace, "venv_vectorb", "bin", "slither")
        slither_bin = venv_slither if os.path.exists(venv_slither) else "slither"

        cmd = [
            slither_bin, sol_path,
            "--json", "-",
            "--disable-color",
            "--no-fail-pedantic",
            "--filter-paths", "node_modules,@openzeppelin,@solady,lib/,test/,forge-std",
        ]

        # Dynamically build solc remappings
        remaps = []
        base_dir = os.path.dirname(sol_path)
        cand_dir = base_dir
        for _ in range(3):
            if not cand_dir or cand_dir == "/":
                break
            try:
                for item in os.listdir(cand_dir):
                    item_path = os.path.join(cand_dir, item)
                    if os.path.isdir(item_path) and item.startswith("@"):
                        remaps.append(f"{item}/={item_path}/")
            except Exception:
                pass
            cand_dir = os.path.dirname(cand_dir)

        if remaps:
            cmd.extend(["--solc-remaps", ",".join(remaps)])

        try:
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=120,
                cwd=os.path.dirname(sol_path),
            )
            # Slither writes JSON to stdout
            stdout = result.stdout.strip()
            if stdout and stdout.startswith("{"):
                return json.loads(stdout)
            # Sometimes it's in stderr
            stderr = result.stderr.strip()
            # Try to find JSON in stderr
            for line in stderr.splitlines():
                if line.strip().startswith("{"):
                    try:
                        return json.loads(line.strip())
                    except Exception:
                        pass
            return None
        except subprocess.TimeoutExpired:
            _log("VALIDATOR", f"Slither timeout for {sol_path}")
            return None
        except Exception as e:
            _log("VALIDATOR", f"Slither error: {type(e).__name__}: {e}")
            return None

    def _parse_slither_findings(self, slither_json: dict, address: str) -> list[dict]:
        """Extract structured findings from Slither JSON output."""
        findings = []
        if not slither_json:
            return findings

        results = slither_json.get("results", {})
        detectors = results.get("detectors", [])

        addr_lower = address.lower()

        for det in detectors:
            detector_id = det.get("check", "unknown")
            impact = det.get("impact", "Low")
            confidence = det.get("confidence", "Low")
            description = det.get("description", "")

            # Skip low signal detectors entirely
            if detector_id in LOW_SIGNAL_DETECTORS:
                continue

            # Extract elements (files, functions, lines)
            elements = det.get("elements", [])
            locations = []
            for el in elements:
                src = el.get("source_mapping", {})
                fname = src.get("filename_short", "")
                lines = src.get("lines", [])
                func = el.get("name", "")
                if lines:
                    locations.append({
                        "file": fname,
                        "lines": lines,
                        "function": func,
                    })

            # Calculate base confidence score
            score = SLITHER_CONFIDENCE_MAP.get((impact, confidence), 30)

            # Boost for high-signal detectors
            if detector_id in HIGH_SIGNAL_DETECTORS:
                score = min(score + 15, 100)

            # Penalize for known audited contracts (they've been reviewed)
            if addr_lower in KNOWN_AUDITED_CONTRACTS:
                score = max(score - 20, 10)
                # But if it's a VERY high signal, keep it
                if detector_id in HIGH_SIGNAL_DETECTORS and impact == "High":
                    score = max(score, 60)

            findings.append({
                "detector": detector_id,
                "impact": impact,
                "confidence": confidence,
                "score": score,
                "description": description[:500],
                "locations": locations[:3],  # Top 3 locations
                "first_markdown": det.get("markdown", "")[:300],
            })

        return findings

    def _phase_b_context_check(self, finding: dict, sol_content: str) -> dict:
        """Phase B: Context-aware false positive reduction.

        Checks CEI pattern, access modifiers, and known safe patterns.
        Returns finding with updated score and fp_flags.
        """
        fp_flags = []
        score = finding["score"]
        detector = finding["detector"]

        # ── Reentrancy checks ──
        if "reentrancy" in detector:
            # Check if function has nonReentrant modifier
            for loc in finding["locations"]:
                func_name = loc.get("function", "")
                if func_name and func_name in sol_content:
                    # Find function block
                    func_pattern = rf"function\s+{re.escape(func_name)}[^{{]*\{{[^}}]{{0,500}}"
                    m = re.search(func_pattern, sol_content, re.DOTALL)
                    if m:
                        func_block = m.group(0)
                        if "nonReentrant" in func_block or "ReentrancyGuard" in func_block:
                            fp_flags.append("nonReentrant_modifier_present")
                            score -= 40
                        # CEI check: state updates BEFORE .call
                        call_pos = func_block.find(".call{")
                        if call_pos == -1:
                            call_pos = func_block.find(".call(")
                        if call_pos > 0:
                            before_call = func_block[:call_pos]
                            after_call = func_block[call_pos:]
                            # State updates after call = potential reentrancy
                            state_after = bool(re.search(r"\b\w+\s*=\s*", after_call))
                            state_before = bool(re.search(r"\b\w+\s*=\s*", before_call))
                            if not state_after and state_before:
                                fp_flags.append("CEI_pattern_followed")
                                score -= 25

        # ── Timestamp manipulation ──
        if detector == "timestamp":
            # In DEX/order systems, timestamp is often acceptable
            if any(k in sol_content for k in ["isExpired", "deadline", "expiration"]):
                fp_flags.append("timestamp_usage_is_by_design")
                score -= 30

        # ── Delegatecall checks ──
        if "delegatecall" in detector:
            # Check if it's in a simulate() function (intentional pattern)
            for loc in finding["locations"]:
                func_name = loc.get("function", "")
                if func_name in ("simulate", "_simulate"):
                    fp_flags.append("simulate_function_intentional_delegatecall")
                    score -= 35
            # Check if onlyOwner or access control wraps the delegatecall
            if "onlyOwner" in sol_content or "onlyProxy" in sol_content:
                fp_flags.append("access_control_present")
                score -= 20

        # ── tx.origin checks ──
        if detector == "tx-origin":
            # Some contracts use tx.origin for specific EIP patterns
            if "isContract" in sol_content or "EOA" in sol_content:
                fp_flags.append("tx_origin_used_for_eoa_check")
                score -= 20

        finding["score"] = max(score, 0)
        finding["fp_flags"] = fp_flags
        return finding

    def validate_contract(self, address: str) -> dict:
        """Run full 3-phase validation on a contract."""
        start = time.monotonic()
        
        result = {
            "address": address,
            "validated": False,
            "confidence_score": 0,
            "validated_findings": [],
            "rejected_findings": [],
            "slither_available": False,
            "false_positive_rate": 0.0,
            "validation_time": 0,
            "timestamp": datetime.now(timezone.utc).isoformat(),
        }

        # Find contract source
        sol_path = self._find_contract_sol(address)
        if not sol_path:
            _log("VALIDATOR", f"  No .sol source — using report fallback for {address}")
            return self._fallback_from_report(address, "", start)

        with open(sol_path, encoding="utf-8", errors="replace") as f:
            sol_content = f.read()

        # ── Phase A: Slither JSON Analysis ──
        _log("VALIDATOR", f"  Phase A: Running Slither on {os.path.basename(sol_path)}...")
        slither_json = self._run_slither_json(sol_path)

        if slither_json is None:
            _log("VALIDATOR", f"  Slither failed. Using heuristic fallback.")
            return self._fallback_from_report(address, sol_content, start)

        result["slither_available"] = True
        raw_findings = self._parse_slither_findings(slither_json, address)
        _log("VALIDATOR", f"  Phase A: {len(raw_findings)} raw Slither detections")

        # ── Phase B: Context Check ──
        _log("VALIDATOR", "  Phase B: Context-aware false positive reduction...")
        checked_findings = []
        for f in raw_findings:
            checked = self._phase_b_context_check(f.copy(), sol_content)
            checked_findings.append(checked)

        # ── Phase C: Confidence Scoring & Filtering ──
    def _fallback_from_report(self, address: str, sol_content: str, start: float) -> dict:
        """When Slither fails or .sol missing, parse existing markdown report.

        Uses Phase B context checks on content extracted from report.
        Threshold: 65 (slightly lower than Slither path since we can't do full dataflow).
        """
        report_path = os.path.join(self.reports_dir, f"report_{address.lower()}.md")

        result = {
            "address": address,
            "validated": False,
            "confidence_score": 0,
            "validated_findings": [],
            "rejected_findings": [],
            "slither_available": False,
            "false_positive_rate": 0.0,
            "validation_time": 0,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "source": "heuristic_fallback",
        }

        if not os.path.exists(report_path):
            result["error"] = "no_report"
            result["validation_time"] = time.monotonic() - start
            return result

        with open(report_path, encoding="utf-8") as f:
            content = f.read()

        # Check if CLEAN
        if "**Overall Risk / الخطورة** | **CLEAN**" in content:
            result["error"] = "clean_contract"
            result["validation_time"] = time.monotonic() - start
            return result

        findings = []
        addr_lower = address.lower()
        is_audited = addr_lower in KNOWN_AUDITED_CONTRACTS

        # ── Parse ### [Severity] heading blocks ──
        # Matches: ### 🔴 [Critical] Title text  OR  ### 🟡 [Medium] Title
        # Uses [^\[]+ to skip any emoji without encoding issues
        heading_pattern = re.compile(
            r"###\s+[^\[]+\[(\w+)\]\s+(.+?)\n"
            r"(.*?)(?=###|\Z)",
            re.DOTALL,
        )

        for m in heading_pattern.finditer(content):
            severity_raw = m.group(1).strip().upper()
            title = m.group(2).strip()
            block = m.group(3)

            # Extract Rule ID and line number
            rule_id = "heuristic"
            rule_m = re.search(r"`([A-Z]+-\d+)`", block)
            if rule_m:
                rule_id = rule_m.group(1).lower()

            line_num = 0
            line_m = re.search(r"\*\*Line\*\*.*?(\d+)", block)
            if line_m:
                line_num = int(line_m.group(1))

            func_name = ""
            func_m = re.search(r"\*\*Function\*\*.*?`?([\w_]+)`?", block)
            if func_m:
                func_name = func_m.group(1)

            # Base score from severity
            base_scores = {"CRITICAL": 82, "HIGH": 72, "MEDIUM": 58, "LOW": 38}
            score = base_scores.get(severity_raw, 35)

            # Audited contract penalty
            if is_audited:
                penalty = 18 if severity_raw in ("CRITICAL", "HIGH") else 12
                score -= penalty

            # Build finding for Phase B
            finding = {
                "detector": rule_id,
                "impact": severity_raw.capitalize(),
                "confidence": "Medium",
                "score": score,
                "description": title,
                "locations": [{"lines": [line_num], "function": func_name}],
                "fp_flags": [],
                "source": "heuristic",
                "vulnerable_code": re.search(r"```solidity(.+?)```", block, re.DOTALL).group(1)[:300] if re.search(r"```solidity(.+?)```", block, re.DOTALL) else "",
            }

            # Phase B: use embedded code snippet as context
            code_context = finding["vulnerable_code"] + "\n" + content
            finding = self._phase_b_context_check(finding, code_context)
            findings.append(finding)

        # Deduplicate by (rule_id, line_num)
        seen = set()
        deduped = []
        for f in findings:
            key = (f["detector"], tuple(f["locations"][0].get("lines", []) if f["locations"] else []))
            if key not in seen:
                seen.add(key)
                deduped.append(f)
        findings = deduped

        THRESHOLD = 65  # Slightly lower threshold for heuristic path
        validated = [f for f in findings if f["score"] >= THRESHOLD]
        rejected  = [f for f in findings if f["score"] < THRESHOLD]

        total = len(findings)
        fp_rate = (len(rejected) / total * 100) if total > 0 else 0.0

        result["validated_findings"] = validated
        result["rejected_findings"] = rejected
        result["validated"] = len(validated) > 0
        result["confidence_score"] = max((f["score"] for f in validated), default=0)
        result["false_positive_rate"] = round(fp_rate, 1)
        result["validation_time"] = round(time.monotonic() - start, 2)

        self._save_validation(address, result)
        return result

    def _save_validation(self, address: str, result: dict) -> None:
        """Save validation result to disk."""
        addr_lower = address.lower()
        if result.get("validated"):
            out_path = os.path.join(self.validated_dir, f"validated_{addr_lower}.json")
        else:
            out_path = os.path.join(self.rejected_dir, f"rejected_{addr_lower}.json")

        with open(out_path, "w", encoding="utf-8") as f:
            json.dump(result, f, indent=2, ensure_ascii=False)

        status = "VALIDATED" if result.get("validated") else "REJECTED"
        score = result.get("confidence_score", 0)
        fp = result.get("false_positive_rate", 0)
        _log("VALIDATOR", f"  [{status}] {address[:12]}... | Score: {score} | FP Rate: {fp}%")

    def validate_all_reports(self) -> dict:
        """Validate all contracts in security_reports directory."""
        if not os.path.isdir(self.reports_dir):
            _log("VALIDATOR", "No security_reports directory found.")
            return {}

        all_results = {}
        for fname in sorted(os.listdir(self.reports_dir)):
            if not fname.endswith(".md"):
                continue
            address = fname.replace("report_", "").replace(".md", "")
            try:
                result = self.validate_contract(address)
                if result is not None:
                    all_results[address] = result
            except Exception as e:
                _log("VALIDATOR", f"  ERROR validating {address}: {e}")
                all_results[address] = {
                    "address": address, "validated": False,
                    "confidence_score": 0, "validated_findings": [],
                    "rejected_findings": [], "error": str(e),
                }

        # Summary
        total = len(all_results)
        valid = sum(1 for r in all_results.values() if r and r.get("validated"))
        _log("VALIDATOR", f"\n{'='*50}")
        _log("VALIDATOR", f"VALIDATION COMPLETE: {valid}/{total} contracts validated")
        _log("VALIDATOR", f"{'='*50}")

        return all_results


    def get_sale_ready_contracts(self) -> list[dict]:
        """Return contracts with validated findings, ready for sales pipeline."""
        ready = []
        if not os.path.isdir(self.validated_dir):
            return ready

        for fname in os.listdir(self.validated_dir):
            if not fname.endswith(".json"):
                continue
            path = os.path.join(self.validated_dir, fname)
            try:
                with open(path, encoding="utf-8") as f:
                    data = json.load(f)
                if data.get("validated") and data.get("validated_findings"):
                    ready.append(data)
            except Exception:
                pass

        # Sort by confidence score descending
        ready.sort(key=lambda x: x.get("confidence_score", 0), reverse=True)
        return ready


def validate_contract_cli(address: str) -> None:
    """CLI entry point for validating a single contract."""
    validator = VulnerabilityValidator()
    result = validator.validate_contract(address)
    sys.stdout.write(json.dumps(result, indent=2) + "\n")


def validate_all_cli() -> None:
    """CLI entry point for validating all reports."""
    validator = VulnerabilityValidator()
    results = validator.validate_all_reports()

    # Print summary table
    print("\n" + "="*65)
    print("  MAXWELL VOSS — VALIDATED VULNERABILITY INTELLIGENCE")
    print("="*65)
    for addr, r in sorted(results.items(), key=lambda x: -x[1].get("confidence_score", 0)):
        if r.get("validated"):
            n_valid = len(r.get("validated_findings", []))
            score = r.get("confidence_score", 0)
            fp = r.get("false_positive_rate", 0)
            print(f"  ✅ {addr[:12]}... | Score:{score:>3} | Valid:{n_valid} | FP:{fp:.0f}%")

    print()
    for addr, r in sorted(results.items(), key=lambda x: x[1].get("confidence_score", 0)):
        if not r.get("validated") and r.get("rejected_findings"):
            n_rej = len(r.get("rejected_findings", []))
            print(f"  ❌ {addr[:12]}... | Rejected: {n_rej} findings (below threshold)")
    print("="*65 + "\n")

    sale_ready = VulnerabilityValidator().get_sale_ready_contracts()
    print(f"  SALE-READY CONTRACTS: {len(sale_ready)}")
    for c in sale_ready:
        print(f"    → {c['address'][:16]}... | Confidence: {c['confidence_score']}")
    print()

    sys.stdout.write(json.dumps({"validated": len(results)}, indent=2) + "\n")


if __name__ == "__main__":
    if len(sys.argv) > 1:
        validate_contract_cli(sys.argv[1])
    else:
        validate_all_cli()