Implement Cisco's Foundry specification for agentic AI security evaluation systems with multi-agent architecture
Implement Cisco's Foundry specification for agentic AI security evaluation systems with multi-agent architecture
npx skills add https://github.com/Aradotso/security-skills --skill foundry-security-spec이 명령을 Claude Code에 복사하여 붙여넣어 스킬을 설치하세요
Security testing framework for automotive vehicle networks including CAN, LIN, and FlexRay protocols
Audit email account security, validate credentials, and manage email lists using this C++ email security testing utility
Self-hosted pentest management workspace for tracking engagements, running tools, auto-importing findings, and generating reports
Advanced Python web security scanner with 49 modules, evasion engine, CVE database, and WAF bypass for penetration testing
AI-driven autonomous penetration testing with Kali Linux tools, multi-phase attack planning, and human approval gates for high-risk actions
Execute deterministic, event-sourced security audits using ESAA-Security's LLM-based agent architecture with 95 checks across 16 security domains
| name | foundry-security-spec |
| description | Implement Cisco's Foundry specification for agentic AI security evaluation systems with multi-agent architecture |
| triggers | ["implement foundry security evaluation","set up agentic security testing","create foundry spec system","build security evaluation agents","configure foundry security roles","design ai vulnerability discovery","implement foundry detector agents","create security finding workflow"] |
Skill by ara.so — Security Skills collection.
Foundry is an open specification from Cisco for building agentic AI security evaluation systems. It defines a multi-agent architecture with 8 core roles and 5 extension roles that coordinate to discover, validate, and report security findings. This is NOT a tool to install—it's a blueprint for building your own security evaluation system.
Foundry provides:
Works with CodeGuard rule format for portable detection rules that transfer between evaluation and prevention.
foundry-security-spec/
├── spec.md # Main specification (~130 functional requirements)
├── constitution.md # 11 inviolable principles
├── GLOSSARY.md # Terminology reference
└── README.md # Implementation guide
# The constitution contains 11 principles that constrain all implementations
cat constitution.md
Key principles to understand:
# In your project directory
npm install -g @github/spec-kit
# or follow spec-kit installation for your coding agent
# Initialize in your project
cd your-security-eval-project
speckit init
This creates .specify/ directory for spec-driven development.
# Copy constitution into spec-kit memory
cp path/to/foundry-security-spec/constitution.md .specify/memory/constitution.md
# Register it with your agent
/speckit.constitution
# Select "adopt existing constitution" when prompted
# Create specs directory
mkdir -p specs/001-foundry
# Copy seed specification
cp path/to/foundry-security-spec/spec.md specs/001-foundry/spec.md
# Run clarification workflow
/speckit.clarify
Answer questions in these categories:
Identity & Scope:
Q: What is your system name?
A: acme-security-eval
Q: Does "authorized evaluation with source access" hold?
A: yes
Q: Merge, split, or keep the 8 core roles as-is?
A: keep as-is for first implementation
Integration Choices:
Q: Version control system?
A: GitLab self-hosted at https://gitlab.internal
Q: Issue tracker?
A: Jira at https://jira.acme.com
Q: LLM provider?
A: OpenAI via internal gateway at https://llm.internal/v1
Q: Datastore?
A: PostgreSQL
Q: Isolation runtime?
A: Docker containers with network isolation
Q: Deployment target?
A: Kubernetes cluster
Policy Choices:
Q: Severity taxonomy?
A: Critical/High/Medium/Low matching our existing CVE scale
Q: Surface needs-review findings?
A: No, validator rejects inconclusive findings
Q: Label naming convention?
A: foundry:role/name format
Extension Scope (recommend NO for first build):
Q: Include Attack-Mapper role?
A: no
Q: Include Regression-Tracker role?
A: no
Q: Include Compliance-Mapper role?
A: no
Q: Include Impact-Assessor role?
A: no
Q: Include Remediation-Drafter role?
A: no
# Harden clarified spec
/speckit.specify
# Check for remaining clarifications
/speckit.clarify
# Repeat until no markers remain
Your specs/001-foundry/spec.md now contains YOUR specification with decisions filled in.
# Generate technical design
/speckit.plan
# Generate task backlog
/speckit.tasks
# Start implementation
/speckit.implement
# orchestrator.py
import asyncio
from typing import List, Dict
from datastore import FindingStore, ClaimStore
from 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.py
from typing import List
from codeguard import RuleEngine, Finding as CodeGuardFinding
from 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.py
import asyncio
from typing import List, Optional
from 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.py
from 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.py
import aiohttp
from typing import Dict
from 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}"
]
}
}
# config/evaluation.yaml
evaluation:
name: "acme-security-eval"
# Orchestrator settings
orchestrator:
max_concurrent_claims: 10
heartbeat_interval: 60
stall_threshold: 300
# Budget limits
budget:
max_tokens: 10000000
max_duration_hours: 48
per_agent_token_limit: 1000000
# Coverage requirements
coverage:
required_percentage: 80
scope:
- src/**/*.py
- lib/**/*.js
exclude:
- tests/**
- docs/**
# Agent roles enabled
agents:
core:
- orchestrator
- planner
- navigator
- detector
- explorer
- validator
- investigator
- publisher
extensions: [] # Start with core only
# Integration endpoints
integrations:
llm:
provider: "openai"
endpoint: "${LLM_GATEWAY_URL}"
model: "gpt-4"
api_key: "${LLM_API_KEY}"
vcs:
type: "gitlab"
url: "${GITLAB_URL}"
token: "${GITLAB_TOKEN}"
issue_tracker:
type: "jira"
url: "${JIRA_URL}"
project: "SEC"
token: "${JIRA_TOKEN}"
datastore:
type: "postgresql"
connection_string: "${DATABASE_URL}"
# Security controls
sandbox:
runtime: "docker"
network_isolation: true
timeout_seconds: 300
resource_limits:
cpu: "1"
memory: "2Gi"
# config/detection.yaml
detection:
# CodeGuard rules corpus
rules_corpus: "/etc/foundry/rules/codeguard"
# Rule categories to enable
enabled_categories:
- injection
- authentication
- authorization
- cryptography
- data-exposure
- configuration
# Severity mapping
severity_mapping:
critical: ["CWE-89", "CWE-78", "CWE-79"] # SQLi, Command Injection, XSS
high: ["CWE-306", "CWE-862"] # Missing Auth, Authorization
medium: ["CWE-327", "CWE-338"] # Weak Crypto, Weak PRNG
low: ["CWE-209", "CWE-532"] # Info Leak, Log Injection
# Rule gap tracking
rule_gaps:
enabled: true
auto_draft_rules: true
review_queue: "rule-improvements"
# run_evaluation.py
import asyncio
from orchestrator import Orchestrator
from 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 atomically
async 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 evidence
def 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
)
Check rule corpus:
# Verify rules loaded
foundry-ctl list-rules --corpus /etc/foundry/rules/codeguard
# Test rule against sample code
foundry-ctl test-rule CWE-89 --file sample.py
Check explorer creativity:
# Increase temperature for hypothesis generation
hypotheses = await self.llm.complete(
prompt,
temperature=0.8 # Higher = more creative, less reliable
)
Check heartbeat configuration:
# Ensure heartbeats are sent frequently enough
HEARTBEAT_INTERVAL = 60 # seconds
STALL_THRESHOLD = 300 # 5 minutes
# Heartbeats must be faster than threshold
assert HEARTBEAT_INTERVAL < STALL_THRESHOLD / 2
Check for deadlocks:
-- Find stalled claims
SELECT claim_id, agent_id, last_heartbeat, status
FROM claims
WHERE status = 'in_progress'
AND last_heartbeat < NOW() - INTERVAL '5 minutes';
Check verdict state:
# Only confirmed findings publish
if finding.verdict != "confirmed":
logger.error(f"Cannot publish {finding.id}, verdict={finding.verdict}")
# Finding needs validation first
Check deduplication:
# Verify fingerprints are stable
f1 = generate_fingerprint(finding)
f2 = generate_fingerprint(finding)
assert f1 == f2, "Fingerprints must be deterministic"
Check scope configuration:
# Ensure scope includes all target code
coverage:
scope:
- src/**/*.py # Include all source
- lib/**/*.{js,ts} # Multiple extensions
exclude:
- tests/** # Don't count tests
- vendor/** # Don't count dependencies
Check claim distribution:
# Verify planner creates sufficient claims
plan = 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 corpus
from 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 corpus
async 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/")
spec.md — 130 functional requirements with rationaleconstitution.md — 11 inviolable principlesGLOSSARY.md — Foundry terminology