Foundry Security Spec

# orchestrator.pyimport asynciofrom typing import List, Dictfrom datastore import FindingStore, ClaimStorefrom agents import Planner, Detector, Explorer, Validator
class Orchestrator:    def __init__(self,                  llm_client,                 finding_store: FindingStore,                 claim_store: ClaimStore,                 budget_manager):        self.llm = llm_client        self.findings = finding_store        self.claims = claim_store        self.budget = budget_manager                # Initialize agent roles        self.planner = Planner(llm_client)        self.detector = Detector(llm_client, rules_corpus)        self.explorer = Explorer(llm_client)        self.validator = Validator(llm_client)        async def evaluate(self, target_repo: str) -> Dict:        """Run complete evaluation coordinating all agents."""                # FR-001: Orchestrator creates evaluation record        eval_id = await self.findings.create_evaluation(            target=target_repo,            status="running"        )                try:            # FR-010: Planner creates work plan            plan = await self.planner.create_plan(target_repo)            await self.claims.store_plan(eval_id, plan)                        # FR-020: Distribute work to detection and exploration            detection_task = asyncio.create_task(                self.run_detection(eval_id, plan)            )            exploration_task = asyncio.create_task(                self.run_exploration(eval_id, plan)            )                        # FR-005: Monitor heartbeats and budgets            monitor_task = asyncio.create_task(                self.monitor_health(eval_id)            )                        # Wait for completion            await asyncio.gather(                detection_task,                exploration_task,                monitor_task            )                        # FR-006: Check coverage gate before completion            coverage = await self.calculate_coverage(eval_id)            if coverage < plan.required_coverage:                raise InsufficientCoverageError(                    f"Coverage {coverage}% < required {plan.required_coverage}%"                )                        # Mark evaluation complete            await self.findings.update_evaluation(                eval_id,                status="complete",                coverage=coverage            )                        return {                "eval_id": eval_id,                "status": "complete",                "findings": await self.findings.count(eval_id),                "coverage": coverage            }                    except Exception as e:            # FR-008: Fail-safe: mark evaluation failed            await self.findings.update_evaluation(                eval_id,                status="failed",                error=str(e)            )            raise        async def monitor_health(self, eval_id: str):        """Monitor agent heartbeats and budgets."""        while True:            await asyncio.sleep(30)                        # FR-007: Check heartbeats            stalled = await self.claims.find_stalled_claims(                eval_id,                heartbeat_threshold=300  # 5 minutes            )                        for claim in stalled:                # FR-007: Auto-block stalled claims                await self.claims.block_claim(                    claim.id,                    reason="heartbeat_timeout"                )                        # FR-009: Check budget exhaustion            if await self.budget.is_exhausted(eval_id):                await self.findings.update_evaluation(                    eval_id,                    status="budget_exhausted"                )                break

# detector.pyfrom typing import Listfrom codeguard import RuleEngine, Finding as CodeGuardFindingfrom models import Claim, Finding
class Detector:    def __init__(self, llm_client, rules_corpus_path: str):        self.llm = llm_client        # FR-030: Load CodeGuard rules        self.rule_engine = RuleEngine.load(rules_corpus_path)        async def process_claim(self, claim: Claim) -> List[Finding]:        """Apply detection rules to a code claim."""                # FR-031: Extract relevant code from claim        code_units = await self.extract_code_units(claim)                findings = []                for unit in code_units:            # FR-032: Run rule engine            rule_hits = await self.rule_engine.evaluate(                code=unit.content,                context=unit.context,                language=unit.language            )                        for hit in rule_hits:                # FR-033: Convert rule hit to finding                finding = Finding(                    claim_id=claim.id,                    rule_id=hit.rule_id,                    severity=hit.severity,                    weakness_id=hit.cwe_id,                    location=hit.location,                    evidence={                        "rule_match": hit.matched_pattern,                        "code_snippet": unit.content,                        "line_range": hit.line_range                    },                    verdict="confirmed",  # Rules are deterministic                    status="validated"                )                findings.append(finding)                                # FR-034: Record coverage                await self.record_coverage(claim.id, unit.path)                return findings        async def extract_code_units(self, claim: Claim):        """Use LLM to identify relevant code units in claim scope."""                prompt = f"""        Claim: {claim.description}        Scope: {claim.scope}                Identify all code units (functions, methods, classes) that should be        evaluated for security issues related to this claim.                Return as JSON array with: path, name, start_line, end_line        """                response = await self.llm.complete(prompt)        return parse_code_units(response)

# explorer.pyimport asynciofrom typing import List, Optionalfrom models import Claim, Finding, RuleGap
class Explorer:    def __init__(self, llm_client, sandbox_runtime):        self.llm = llm_client        self.sandbox = sandbox_runtime        async def investigate_claim(self, claim: Claim) -> List[Finding]:        """Creative exploration beyond static rules."""                findings = []                # FR-040: Generate investigation hypotheses        hypotheses = await self.generate_hypotheses(claim)                for hypothesis in hypotheses:            # FR-041: Execute in isolated sandbox            async with self.sandbox.session() as session:                result = await self.test_hypothesis(                    session,                    hypothesis,                    claim                )                                if result.is_vulnerability:                    # FR-042: Evidence-gated finding                    if not result.has_reproduction:                        # Don't create finding without evidence                        continue                                        finding = Finding(                        claim_id=claim.id,                        severity=result.severity,                        weakness_id=result.weakness_id,                        description=result.description,                        evidence=result.evidence,                        verdict="needs-validation",                        status="pending"                    )                    findings.append(finding)                                        # FR-043: Check if rules missed this                    if await self.should_have_detected(finding):                        await self.record_rule_gap(finding)                return findings        async def generate_hypotheses(self, claim: Claim) -> List[Dict]:        """Use LLM to generate creative test hypotheses."""                prompt = f"""        You are exploring code for security issues that static rules may miss.                Claim: {claim.description}        Code scope: {claim.scope}                Generate 3-5 security hypotheses to test:        - Focus on logic bugs, state confusion, race conditions        - Consider what rules can't express (context-dependent issues)        - Prioritize high-impact scenarios                For each hypothesis provide:        - What to test        - Why it might be vulnerable        - How to reproduce if vulnerable                Return as JSON array.        """                response = await self.llm.complete(            prompt,            temperature=0.7  # Higher for creative exploration        )        return parse_hypotheses(response)        async def record_rule_gap(self, finding: Finding):        """Record that rules failed to detect this issue."""                gap = RuleGap(            finding_id=finding.id,            weakness_id=finding.weakness_id,            pattern=finding.evidence.get("vulnerable_pattern"),            reason="explorer_found_missed_by_detector",            suggested_rule=await self.draft_rule(finding)        )                # FR-044: Feed into rule corpus improvement        await self.rule_gaps.store(gap)

# validator.pyfrom models import Finding, ValidationResult
class Validator:    def __init__(self, llm_client, sandbox_runtime):        self.llm = llm_client        self.sandbox = sandbox_runtime        async def validate_finding(self, finding: Finding) -> ValidationResult:        """Reproduce and confirm finding from evidence."""                # FR-050: Check evidence completeness        if not self.has_sufficient_evidence(finding):            return ValidationResult(                verdict="rejected",                reason="insufficient_evidence"            )                # FR-051: Attempt reproduction        async with self.sandbox.session() as session:            reproduced = await self.reproduce_issue(                session,                finding.evidence            )                        if not reproduced:                return ValidationResult(                    verdict="rejected",                    reason="not_reproducible"                )                        # FR-052: Verify severity assessment            actual_severity = await self.assess_severity(                session,                finding            )                        if actual_severity != finding.severity:                finding.severity = actual_severity                finding.evidence["severity_adjustment"] = {                    "original": finding.severity,                    "validated": actual_severity                }                        # FR-053: Generate fingerprint for deduplication            fingerprint = await self.generate_fingerprint(finding)                        return ValidationResult(                verdict="confirmed",                fingerprint=fingerprint,                severity=actual_severity,                reproduction_evidence=session.get_transcript()            )        def has_sufficient_evidence(self, finding: Finding) -> bool:        """Check if finding has required evidence."""        required = ["location", "description"]                if finding.severity in ["critical", "high"]:            required.extend(["reproduction_steps", "impact"])                return all(k in finding.evidence for k in required)        async def generate_fingerprint(self, finding: Finding) -> str:        """Create stable fingerprint for deduplication."""                # FR-054: Fingerprint combines weakness + location + root cause        components = [            finding.weakness_id,            finding.location.get("file_path"),            finding.location.get("function_name"),            finding.evidence.get("root_cause_pattern")        ]                fingerprint_input = "|".join(str(c) for c in components if c)        return hashlib.sha256(fingerprint_input.encode()).hexdigest()[:16]

# publisher.pyimport aiohttpfrom typing import Dictfrom models import Finding
class Publisher:    def __init__(self, issue_tracker_config: Dict):        self.tracker_url = issue_tracker_config["url"]        self.project_key = issue_tracker_config["project"]        self.api_token = issue_tracker_config["token"]  # From env: ISSUE_TRACKER_TOKEN        async def publish_finding(self, finding: Finding) -> str:        """Create issue in tracker for confirmed finding."""                # FR-060: Only publish confirmed findings        if finding.verdict != "confirmed":            raise ValueError(f"Cannot publish {finding.verdict} finding")                # FR-061: Check for existing issue via fingerprint        existing = await self.find_existing_issue(finding.fingerprint)        if existing:            return existing.issue_id                # FR-062: Format issue according to tracker schema        issue_body = self.format_issue(finding)                # FR-063: Create issue        async with aiohttp.ClientSession() as session:            async with session.post(                f"{self.tracker_url}/rest/api/2/issue",                headers={                    "Authorization": f"Bearer {self.api_token}",                    "Content-Type": "application/json"                },                json=issue_body            ) as resp:                resp.raise_for_status()                result = await resp.json()                issue_id = result["key"]                # FR-064: Update finding with issue reference        finding.issue_id = issue_id        finding.status = "published"        await finding.save()                return issue_id        def format_issue(self, finding: Finding) -> Dict:        """Format finding as issue tracker ticket."""                description = f"""        *Security Finding from Foundry Evaluation*                *Severity:* {finding.severity.upper()}        *Weakness:* {finding.weakness_id}        *Location:* {finding.location.get('file_path')}:{finding.location.get('line_number')}                h3. Description        {finding.description}                h3. Evidence        {self.format_evidence(finding.evidence)}                h3. Reproduction        {finding.evidence.get('reproduction_steps', 'See evidence above')}                ---        Fingerprint: {finding.fingerprint}        Evaluation ID: {finding.eval_id}        """                return {            "fields": {                "project": {"key": self.project_key},                "summary": f"[{finding.severity.upper()}] {finding.weakness_id}: {finding.get_short_description()}",                "description": description,                "issuetype": {"name": "Security Vulnerability"},                "priority": {"name": self.map_severity_to_priority(finding.severity)},                "labels": [                    f"foundry:eval:{finding.eval_id}",                    f"foundry:weakness:{finding.weakness_id}",                    f"foundry:fingerprint:{finding.fingerprint}"                ]            }        }

# run_evaluation.pyimport asynciofrom orchestrator import Orchestratorfrom config import load_config
async def main():    # Load configuration    config = load_config("config/evaluation.yaml")        # Initialize orchestrator    orchestrator = Orchestrator(        llm_client=create_llm_client(config.integrations.llm),        finding_store=FindingStore(config.integrations.datastore),        claim_store=ClaimStore(config.integrations.datastore),        budget_manager=BudgetManager(config.budget)    )        # Run evaluation    result = await orchestrator.evaluate(        target_repo="https://gitlab.internal/acme/webapp"    )        print(f"Evaluation {result['eval_id']} complete")    print(f"Findings: {result['findings']}")    print(f"Coverage: {result['coverage']}%")
if __name__ == "__main__":    asyncio.run(main())

# Every agent processes claims atomicallyasync def process_claim(self, claim: Claim):    # Claim the work atomically    if not await self.claims.try_claim(claim.id, self.agent_id):        return  # Another agent got it        try:        # Send heartbeat while working        heartbeat_task = asyncio.create_task(            self.send_heartbeats(claim.id)        )                # Do the work        result = await self.do_work(claim)                # Mark complete        await self.claims.complete(claim.id, result)            except Exception as e:        # Fail claim, don't retry (constitution principle)        await self.claims.fail(claim.id, str(e))    finally:        heartbeat_task.cancel()

# Never create findings without evidencedef create_finding(self, claim: Claim, issue: Dict) -> Optional[Finding]:    # Check evidence requirements    if not issue.get("location"):        logger.warning(f"No location for issue in {claim.id}, skipping")        return None        if not issue.get("reproduction"):        logger.warning(f"No reproduction for issue in {claim.id}, skipping")        return None        # Evidence is sufficient    return Finding(        claim_id=claim.id,        location=issue["location"],        evidence={            "reproduction": issue["reproduction"],            "impact": issue["impact"],            "code_snippet": issue["code"]        },        verdict="needs-validation"    )

class BudgetManager:    async def check_budget(self, eval_id: str, tokens_requested: int) -> bool:        """Check if budget allows operation."""        used = await self.get_tokens_used(eval_id)        limit = self.config.max_tokens                if used + tokens_requested > limit:            await self.notify_budget_exhausted(eval_id)            return False                return True        async def record_usage(self, eval_id: str, tokens: int):        """Record token usage."""        await self.db.execute(            "INSERT INTO token_usage (eval_id, tokens, timestamp) VALUES ($1, $2, NOW())",            eval_id, tokens        )

# Increase temperature for hypothesis generationhypotheses = await self.llm.complete(    prompt,    temperature=0.8  # Higher = more creative, less reliable)

# Ensure heartbeats are sent frequently enoughHEARTBEAT_INTERVAL = 60  # secondsSTALL_THRESHOLD = 300     # 5 minutes
# Heartbeats must be faster than thresholdassert HEARTBEAT_INTERVAL < STALL_THRESHOLD / 2

# Only confirmed findings publishif finding.verdict != "confirmed":    logger.error(f"Cannot publish {finding.id}, verdict={finding.verdict}")    # Finding needs validation first

# Verify fingerprints are stablef1 = generate_fingerprint(finding)f2 = generate_fingerprint(finding)assert f1 == f2, "Fingerprints must be deterministic"

# Verify planner creates sufficient claimsplan = await planner.create_plan(target)print(f"Claims created: {len(plan.claims)}")print(f"Surface area: {plan.surface_area}")# Claims should cover all in-scope files

# Load CodeGuard rules as Detector corpusfrom codeguard import RuleEngine
class Detector:    def __init__(self, rules_path: str):        # Load portable CodeGuard rules        self.rules = RuleEngine.load(rules_path)            async def scan(self, code: str, language: str):        # Rules execute deterministically        return await self.rules.evaluate(code, language)
# Export rule gaps back to CodeGuard corpusasync def export_rule_gap(gap: RuleGap):    """Convert discovered gap into CodeGuard rule."""    rule = {        "id": f"CWE-{gap.weakness_id}-{gap.pattern_hash}",        "name": gap.suggested_name,        "description": gap.description,        "pattern": gap.pattern,        "severity": gap.severity,        "languages": gap.applicable_languages    }        # Write to CodeGuard format    await write_codeguard_rule(rule, "rules/corpus/custom/")