diff --git a/agent.py b/agent.py new file mode 100644 index 0000000..35ab075 --- /dev/null +++ b/agent.py @@ -0,0 +1,347 @@ +from __future__ import annotations + +import hashlib +import json +import re +import time +from datetime import UTC, datetime +from enum import Enum +from fnmatch import fnmatch +from ipaddress import ip_address +from typing import Any, Literal +from urllib.parse import urlparse + +import httpx +from pydantic import AnyHttpUrl, BaseModel, ConfigDict, Field, field_validator + +import a2a_pack as a2a +from a2a_pack import A2AAgent, ConsumerSetup, ConsumerSetupField, EgressPolicy, NoAuth, Pricing, Resources, RunContext, WorkspaceAccess, WorkspaceMode +from a2a_pack.context import AgentEvent +from a2a_pack.workspace import FileType + +MAX_TEXT_BYTES = 256_000 +MAX_EVIDENCE_BYTES = 8_000 +MAX_FILES_PER_AUDIT = 80 +MAX_FINDINGS = 200 +HTTP_TIMEOUT_SECONDS = 8.0 +DENIED_REPO_GLOBS = (".git/**", ".env", ".env.*", "*.pem", "*.key", "*secret*", "*credential*", "**/.env", "**/.env.*", "**/*secret*", "**/*credential*", "**/*.pem", "**/*.key") +ALLOWED_REPO_GLOBS = ("agent.py", "a2a.yaml", "requirements.txt", "README.md", "*.py", "*.yaml", "*.yml", "*.json", "*.md", "tests/*.py", "tests/**/*.py", "skills/**/SKILL.md") +SECRET_PATTERNS = (re.compile(r"(?i)(api[_-]?key|token|secret|password)\s*[:=]\s*['\"]?([A-Za-z0-9_./+=-]{12,})"), re.compile(r"sk-[A-Za-z0-9]{20,}"), re.compile(r"(?i)bearer\s+[A-Za-z0-9_./+=-]{16,}")) +WRITE_INDICATORS = ("kubectl apply", "kubectl delete", "kubectl patch", "kubectl scale", "helm upgrade", "git push", "ctx.secret(", "OPENAI_API_KEY", "A2A_LITELLM_KEY") + +class StrictModel(BaseModel): + model_config = ConfigDict(extra="forbid") + + def __getitem__(self, key: str) -> Any: + return getattr(self, key) + +class Severity(str, Enum): + CRITICAL="critical"; HIGH="high"; MEDIUM="medium"; LOW="low"; INFO="info" +class Confidence(str, Enum): + HIGH="high"; MEDIUM="medium"; LOW="low" +class ProbeMode(str, Enum): + NONE="none"; HEALTH_ONLY="health_only"; SYNTHETIC="synthetic" + +class AuditTarget(StrictModel): + agent_name: str = Field(min_length=1, max_length=128, pattern=r"^[a-z0-9][a-z0-9-]{0,127}$") + agent_url: AnyHttpUrl | None = None + repository: str | None = Field(default=None, max_length=256) + namespace: str | None = Field(default=None, max_length=128) + deployment: str | None = Field(default=None, max_length=128) + @field_validator("repository", "namespace", "deployment") + @classmethod + def clean_optional(cls, value: str | None) -> str | None: + return value.strip() or None if value is not None else None + +class AuditOptions(StrictModel): + audit_id: str | None = Field(default=None, max_length=80, pattern=r"^[A-Za-z0-9_.-]+$") + max_files: int = Field(default=40, ge=0, le=MAX_FILES_PER_AUDIT) + probe_mode: ProbeMode = ProbeMode.HEALTH_ONLY + include_repository: bool = True + include_runtime: bool = True + include_remediation: bool = True + allowed_hosts: list[str] = Field(default_factory=list, max_length=20) + allowed_namespaces: list[str] = Field(default_factory=list, max_length=20) + +class Finding(StrictModel): + id: str + control_id: str + title: str + severity: Severity + confidence: Confidence + status: Literal["open", "passed", "unknown", "not_tested"] = "open" + evidence_refs: list[str] = Field(default_factory=list, max_length=20) + impact: str + remediation: str + verification_procedure: str + +class Evidence(StrictModel): + id: str + source: str + path: str | None = None + status: Literal["observed", "missing", "unknown", "not_tested", "redacted"] + excerpt: str = Field(default="", max_length=MAX_EVIDENCE_BYTES) + sha256: str | None = None + +class OutputFile(StrictModel): + path: str + artifact_uri: str | None = None + mime_type: str + size_bytes: int = Field(ge=0) + +def _default_target() -> AuditTarget: + return AuditTarget(agent_name="agent-compliance-auditor") + +class InventoryRequest(StrictModel): + target: AuditTarget = Field(default_factory=_default_target) + options: AuditOptions = Field(default_factory=AuditOptions) +class AuditConfigurationRequest(StrictModel): + target: AuditTarget = Field(default_factory=_default_target) + options: AuditOptions = Field(default_factory=AuditOptions) + inventory: dict[str, Any] | None = None +class AuditRuntimeRequest(StrictModel): + target: AuditTarget = Field(default_factory=_default_target) + options: AuditOptions = Field(default_factory=AuditOptions) + inventory: dict[str, Any] | None = None +class AssessRiskRequest(StrictModel): + target: AuditTarget + options: AuditOptions = Field(default_factory=AuditOptions) + inventory: dict[str, Any] | None = None + configuration_findings: list[Finding] = Field(default_factory=list, max_length=MAX_FINDINGS) + runtime_findings: list[Finding] = Field(default_factory=list, max_length=MAX_FINDINGS) +class GenerateRemediationRequest(StrictModel): + target: AuditTarget + options: AuditOptions = Field(default_factory=AuditOptions) + findings: list[Finding] = Field(default_factory=list, max_length=MAX_FINDINGS) + risk_summary: dict[str, Any] | None = None + +class InventoryResult(StrictModel): + audit_id: str; target: AuditTarget; status: Literal["ok", "partial", "unreachable", "setup_required"]; output_dir: str + generated_files: list[OutputFile] = Field(default_factory=list); inventory: dict[str, Any]; evidence: list[Evidence] = Field(default_factory=list); warnings: list[str] = Field(default_factory=list); mutations_performed: bool = False +class FindingResult(StrictModel): + audit_id: str; target: AuditTarget; status: Literal["ok", "partial", "setup_required"]; output_dir: str + generated_files: list[OutputFile] = Field(default_factory=list); findings: list[Finding]; evidence: list[Evidence] = Field(default_factory=list); warnings: list[str] = Field(default_factory=list); mutations_performed: bool = False +class RiskSummaryResult(StrictModel): + audit_id: str; target: AuditTarget; status: Literal["ok", "partial"]; output_dir: str + generated_files: list[OutputFile] = Field(default_factory=list); risk_summary: dict[str, Any]; findings: list[Finding]; warnings: list[str] = Field(default_factory=list); mutations_performed: bool = False +class RemediationResult(StrictModel): + audit_id: str; target: AuditTarget; status: Literal["ok", "partial"]; output_dir: str + generated_files: list[OutputFile] = Field(default_factory=list); remediation_markdown: str; sarif: dict[str, Any]; warnings: list[str] = Field(default_factory=list); mutations_performed: bool = False +class AgentComplianceAuditorConfig(StrictModel): + default_allowed_hosts: list[str] = Field(default_factory=list, max_length=50) + default_allowed_namespaces: list[str] = Field(default_factory=list, max_length=50) + +class AgentComplianceAuditor(A2AAgent[AgentComplianceAuditorConfig, NoAuth]): + name="agent-compliance-auditor"; description="Read-only compliance auditor for deployed A2A agents and repositories."; version="0.1.6" + config_model=AgentComplianceAuditorConfig; auth_model=NoAuth + pricing=Pricing(price_per_call_usd=0.0, caller_pays_llm=False, notes="Deterministic read-only auditing; no LLM calls and no production mutations.") + resources=Resources(cpu="500m", memory="512Mi", max_runtime_seconds=900) + workspace_access=WorkspaceAccess.dynamic(max_files=MAX_FILES_PER_AUDIT+10, allowed_modes=(WorkspaceMode.READ_ONLY,), require_reason=False, deny_patterns=DENIED_REPO_GLOBS, max_total_size_bytes=12*1024*1024) + egress=EgressPolicy(allow_hosts=("gitea.a2a.svc.cluster.local",), deny_internet_by_default=True) + tools_used=("httpx","pydantic","workspace") + consumer_setup=ConsumerSetup.from_fields( + ConsumerSetupField.config("GITEA_BASE_URL", label="Gitea base URL", required=False, input_type="url"), + ConsumerSetupField.config("KUBERNETES_BASE_URL", label="Kubernetes API base URL", required=False, input_type="url"), + ConsumerSetupField.config("ARGOCD_BASE_URL", label="Argo CD base URL", required=False, input_type="url"), + ConsumerSetupField.config("REGISTRY_BASE_URL", label="Registry metadata base URL", required=False, input_type="url")) + + @a2a.tool(description="Inventory a deployed A2A agent card, repository evidence, manifests, exposure, provenance, resources, and health", timeout_seconds=180, idempotent=True, cost_class="read-only") + async def inventory_agent(self, ctx: RunContext[NoAuth], request: InventoryRequest) -> InventoryResult: + audit_id=_audit_id(request.target, request.options.audit_id); await _event(ctx,"audit_started",audit_id=audit_id,skill="inventory_agent",target=request.target.agent_name) + inventory,evidence,warnings,status=await _collect_inventory(self,ctx,request.target,request.options,audit_id); output_dir=_output_dir(audit_id) + files=await _write_outputs(ctx,output_dir,{"inventory.json":inventory},warnings); await _event(ctx,"audit_completed",audit_id=audit_id,skill="inventory_agent",findings=0) + return InventoryResult(audit_id=audit_id,target=request.target,status=status,output_dir=output_dir,generated_files=files,inventory=inventory,evidence=evidence,warnings=warnings) + + @a2a.tool(description="Audit static configuration and repository files against A2A security, privacy, and platform controls", timeout_seconds=180, idempotent=True, cost_class="read-only") + async def audit_configuration(self, ctx: RunContext[NoAuth], request: AuditConfigurationRequest) -> FindingResult: + audit_id=_audit_id(request.target,request.options.audit_id); await _event(ctx,"audit_started",audit_id=audit_id,skill="audit_configuration",target=request.target.agent_name) + inventory=request.inventory or (await _collect_inventory(self,ctx,request.target,request.options,audit_id))[0]; findings,evidence,warnings=_configuration_findings(request.target,inventory); findings=_dedupe(findings); output_dir=_output_dir(audit_id) + files=await _write_outputs(ctx,output_dir,{"findings.json":{"findings":[_dump(f) for f in findings]}},warnings); await _event(ctx,"audit_completed",audit_id=audit_id,skill="audit_configuration",findings=len(findings)) + return FindingResult(audit_id=audit_id,target=request.target,status="ok",output_dir=output_dir,generated_files=files,findings=findings,evidence=evidence,warnings=warnings) + + @a2a.tool(description="Run bounded non-destructive runtime probes and audit deployment health without mutations", timeout_seconds=180, idempotent=True, cost_class="read-only") + async def audit_runtime(self, ctx: RunContext[NoAuth], request: AuditRuntimeRequest) -> FindingResult: + audit_id=_audit_id(request.target,request.options.audit_id); await _event(ctx,"audit_started",audit_id=audit_id,skill="audit_runtime",target=request.target.agent_name) + inventory=request.inventory or (await _collect_inventory(self,ctx,request.target,request.options,audit_id))[0]; findings,evidence,warnings=_runtime_findings(request.target,inventory); findings=_dedupe(findings); output_dir=_output_dir(audit_id) + files=await _write_outputs(ctx,output_dir,{"findings.json":{"findings":[_dump(f) for f in findings]}},warnings); await _event(ctx,"audit_completed",audit_id=audit_id,skill="audit_runtime",findings=len(findings)) + return FindingResult(audit_id=audit_id,target=request.target,status="ok",output_dir=output_dir,generated_files=files,findings=findings,evidence=evidence,warnings=warnings) + + @a2a.tool(description="Deduplicate findings, map evidence to controls, and score severity, confidence, and aggregate risk", timeout_seconds=120, idempotent=True, cost_class="read-only") + async def assess_risk(self, ctx: RunContext[NoAuth], request: AssessRiskRequest) -> RiskSummaryResult: + audit_id=_audit_id(request.target,request.options.audit_id); await _event(ctx,"audit_started",audit_id=audit_id,skill="assess_risk",target=request.target.agent_name) + findings=_dedupe([*request.configuration_findings,*request.runtime_findings]); summary=_risk_summary(findings); output_dir=_output_dir(audit_id) + files=await _write_outputs(ctx,output_dir,{"risk-summary.json":summary,"findings.json":{"findings":[_dump(f) for f in findings]}},[]); await _event(ctx,"audit_completed",audit_id=audit_id,skill="assess_risk",findings=len(findings),score=summary["risk_score"]) + return RiskSummaryResult(audit_id=audit_id,target=request.target,status="ok",output_dir=output_dir,generated_files=files,risk_summary=summary,findings=findings) + + @a2a.tool(description="Generate actionable remediation and a machine-readable SARIF report for compliance findings", timeout_seconds=120, idempotent=True, cost_class="read-only") + async def generate_remediation(self, ctx: RunContext[NoAuth], request: GenerateRemediationRequest) -> RemediationResult: + audit_id=_audit_id(request.target,request.options.audit_id); await _event(ctx,"audit_started",audit_id=audit_id,skill="generate_remediation",target=request.target.agent_name) + findings=_dedupe(request.findings); summary=request.risk_summary or _risk_summary(findings); md=_remediation_md(request.target,audit_id,findings,summary); sarif=_sarif(request.target,findings); output_dir=_output_dir(audit_id) + files=await _write_outputs(ctx,output_dir,{"remediation.md":md,"sarif.json":sarif,"risk-summary.json":summary},[]); await _event(ctx,"audit_completed",audit_id=audit_id,skill="generate_remediation",findings=len(findings)) + return RemediationResult(audit_id=audit_id,target=request.target,status="ok",output_dir=output_dir,generated_files=files,remediation_markdown=md,sarif=sarif) + +async def _collect_inventory(agent:AgentComplianceAuditor,ctx:RunContext[NoAuth],target:AuditTarget,options:AuditOptions,audit_id:str)->tuple[dict[str,Any],list[Evidence],list[str],Literal["ok","partial","unreachable","setup_required"]]: + warnings=[]; evidence=[]; allowed_hosts={h.lower().strip() for h in [*agent.config.default_allowed_hosts,*options.allowed_hosts] if h.strip()}; allowed_ns={n.strip() for n in [*agent.config.default_allowed_namespaces,*options.allowed_namespaces] if n.strip()} + authorized=not allowed_ns or bool(target.namespace and target.namespace in allowed_ns); status="ok" if authorized else "partial" + if not authorized: warnings.append("Target namespace is outside the configured allowlist; runtime probes skipped.") + card=await _fetch_card(str(target.agent_url) if target.agent_url else None,allowed_hosts,warnings); status="unreachable" if status=="ok" and card.get("status")=="unreachable" else status; evidence.append(_evidence("card","agent_card","observed" if card.get("card") else "unknown",card)) + repo_files=await _read_repo(ctx,options.max_files if options.include_repository else 0,warnings); evidence.extend(_repo_evidence(repo_files)); manifests=_manifests(repo_files) + health={"status":"not_tested","reason":"probe_mode is none"} + if options.probe_mode is not ProbeMode.NONE and authorized: health=await _health(str(target.agent_url) if target.agent_url else None,allowed_hosts,warnings) + inv={"schema_version":"2026-07-13","audit_id":audit_id,"generated_at":datetime.now(UTC).isoformat(),"identity":{"agent_name":target.agent_name,"agent_url":str(target.agent_url) if target.agent_url else None,"repository":target.repository,"namespace":target.namespace,"deployment":target.deployment,"authorized":authorized},"agent_card":card,"skills":_skills(card.get("card"),repo_files),"schemas":_schemas(card.get("card")),"manifests":manifests,"rbac":_rbac(manifests),"secrets_references":_secret_refs(repo_files),"network_exposure":_exposure(manifests),"image_provenance":_provenance(manifests,repo_files),"resource_limits":_resources(card.get("card"),manifests),"deployment_health":health,"repository":_repo_summary(repo_files),"controls":_controls(),"unknowns":_unknowns(card,manifests,health,repo_files),"mutations_performed":False} + return inv,evidence,warnings,status # type: ignore[return-value] + +async def _fetch_card(agent_url:str|None,allowed_hosts:set[str],warnings:list[str])->dict[str,Any]: + if not agent_url: return {"status":"unknown","reason":"agent_url not provided"} + if not _safe_url(urlparse(agent_url),allowed_hosts): warnings.append("Agent URL host is not allowlisted or is unsafe; card fetch skipped."); return {"status":"not_tested","reason":"host_not_allowlisted"} + try: + async with httpx.AsyncClient(timeout=httpx.Timeout(HTTP_TIMEOUT_SECONDS),follow_redirects=False) as client: + async with client.stream("GET",agent_url.rstrip("/")+"/.well-known/agent-card",headers={"accept":"application/json"}) as resp: + chunks=[]; size=0 + async for chunk in resp.aiter_bytes(): + chunks.append(chunk); size+=len(chunk) + if size>MAX_TEXT_BYTES: break + body=b"".join(chunks)[:MAX_TEXT_BYTES+1].decode("utf-8",errors="replace") + return {"status":"unreachable","http_status":resp.status_code,"body_excerpt":_redact(body[:500])} if resp.status_code>=400 else {"status":"ok","card":_redact_obj(_bounded_json(body,warnings))} + except Exception as exc: warnings.append(f"Agent card fetch failed: {type(exc).__name__}"); return {"status":"unreachable","error_type":type(exc).__name__} +async def _health(agent_url:str|None,allowed_hosts:set[str],warnings:list[str])->dict[str,Any]: + if not agent_url: return {"status":"unknown","reason":"agent_url not provided"} + if not _safe_url(urlparse(agent_url),allowed_hosts): return {"status":"not_tested","reason":"host_not_allowlisted"} + started=time.perf_counter() + try: + async with httpx.AsyncClient(timeout=httpx.Timeout(HTTP_TIMEOUT_SECONDS),follow_redirects=False) as client: resp=await client.get(agent_url.rstrip("/")+"/healthz") + return {"status":"ok" if resp.status_code<500 else "degraded","http_status":resp.status_code,"latency_ms":round((time.perf_counter()-started)*1000,2)} + except Exception as exc: warnings.append(f"Health probe failed: {type(exc).__name__}"); return {"status":"unreachable","error_type":type(exc).__name__} +async def _read_repo(ctx:RunContext[NoAuth],max_files:int,warnings:list[str])->dict[str,str]: + if max_files<=0: return {} + try: view=await ctx.workspace.open_view(purpose="Read-only static inspection of A2A agent source files",hints=("agent.py a2a.yaml requirements README tests skills",),file_types=(FileType.PYTHON,FileType.YAML,FileType.JSON,FileType.MARKDOWN,FileType.TOML),max_files=max_files,mode=WorkspaceMode.READ_ONLY,reason="Compliance audit reads bounded source files and never writes repository paths.") + except Exception as exc: warnings.append(f"Repository workspace unavailable: {type(exc).__name__}"); return {} + files={} + for m in view.files[:max_files]: + path=m.path.replace("\\","/").lstrip("/") + if _path_allowed(path): + try: files[path]=(await view.read(path))[:MAX_TEXT_BYTES].decode("utf-8",errors="replace") + except Exception as exc: warnings.append(f"Could not read {path}: {type(exc).__name__}") + return files +async def _write_outputs(ctx:RunContext[NoAuth],output_dir:str,files:dict[str,Any],warnings:list[str])->list[OutputFile]: + out=[] + for name,payload in files.items(): + rel=f"{output_dir}/{name}"; data=payload.encode() if isinstance(payload,str) else json.dumps(payload,indent=2,sort_keys=True,default=str).encode(); mime="text/markdown" if name.endswith(".md") else "application/json"; uri=None + try: ref=await ctx.write_artifact(rel,data,mime); await ctx.emit_artifact(ref); uri=ref.uri + except Exception as exc: warnings.append(f"Artifact write failed for {rel}: {type(exc).__name__}") + out.append(OutputFile(path=rel,artifact_uri=uri,mime_type=mime,size_bytes=len(data))) + return out + +def _configuration_findings(target:AuditTarget,inv:dict[str,Any])->tuple[list[Finding],list[Evidence],list[str]]: + fs=[]; card=(inv.get("agent_card") or {}).get("card") or {}; skills=inv.get("skills") or [] + if not card: fs.append(_finding(target,"A2A-CARD-001","Agent Card could not be verified",Severity.HIGH,Confidence.MEDIUM,"card","Consumers cannot validate skills, schemas, runtime, or billing metadata.","Ensure /.well-known/agent-card is reachable and valid.","Fetch and validate /.well-known/agent-card.")) + if not skills: fs.append(_finding(target,"A2A-SCHEMA-001","No public skills discovered",Severity.HIGH,Confidence.MEDIUM,"skills","Callers cannot rely on a stable typed API contract.","Expose typed @a2a.tool methods.","Confirm skills in live Agent Card.")) + for s in skills: + schema=s.get("input_schema") or s.get("inputSchema") or {}; out=s.get("output_schema") or s.get("outputSchema") + if not (isinstance(schema,dict) and schema.get("type")=="object" and schema.get("additionalProperties") is False) or not out: fs.append(_finding(target,"A2A-SCHEMA-002",f"Skill {s.get('name','')} has weak schema",Severity.MEDIUM,Confidence.HIGH,"schemas","Weak schemas increase integration errors.","Use strict Pydantic schemas.","Inspect Agent Card schemas.")) + if (inv.get("network_exposure") or {}).get("public") is True: fs.append(_finding(target,"A2A-NET-001","Agent is publicly exposed",Severity.MEDIUM,Confidence.MEDIUM,"network_exposure","Unexpected public exposure broadens abuse paths.","Set expose.public=false unless required.","Review a2a.yaml.")) + if (inv.get("resource_limits") or {}).get("max_runtime_seconds") in (None,"unknown"): fs.append(_finding(target,"A2A-OPS-001","Runtime limit is unknown",Severity.LOW,Confidence.MEDIUM,"resource_limits","Unknown runtime budgets can cause reliability issues.","Declare Resources and manifest runtime resources.","Check Agent Card runtime.resources.")) + if inv.get("secrets_references"): fs.append(_finding(target,"A2A-SECRET-001","Potential secret or provider-key reference found in source",Severity.HIGH,Confidence.HIGH,"secrets_references","Secrets can be exfiltrated.","Remove literal secrets and provider-key reads.","Run secret scan.")) + if (inv.get("repository") or {}).get("suspicious_instruction_count",0)>0: fs.append(_finding(target,"A2A-UNTRUSTED-001","Repository contains malicious instruction-like text",Severity.MEDIUM,Confidence.HIGH,"repository","Untrusted text could manipulate auditors.","Treat repository content only as evidence.","Confirm text remains evidence only.")) + return fs[:MAX_FINDINGS],[_evidence("configuration","configuration_rules","observed",{"finding_count":len(fs)})],[] +def _runtime_findings(target:AuditTarget,inv:dict[str,Any])->tuple[list[Finding],list[Evidence],list[str]]: + fs=[]; health=inv.get("deployment_health") or {"status":"unknown"} + if (inv.get("identity") or {}).get("authorized") is False: fs.append(_finding(target,"A2A-TENANT-001","Runtime audit skipped by namespace allowlist",Severity.HIGH,Confidence.HIGH,"identity","Tenant isolation boundary prevented probes.","Run from owning tenant or allowlist namespace.","Verify namespace allowlist.")) + if health.get("status") in {"unreachable","degraded"}: fs.append(_finding(target,"A2A-HEALTH-001","Agent health probe failed or degraded",Severity.MEDIUM,Confidence.MEDIUM,"deployment_health","Consumers may experience failed calls.","Inspect readiness and routing.","Repeat health probe.")) + if health.get("status") in {"unknown","not_tested"}: fs.append(_finding(target,"A2A-HEALTH-002","Runtime health state is unknown",Severity.LOW,Confidence.MEDIUM,"deployment_health","Operational state is unmeasured.","Provide allowlisted agent_url.","Run health_only probe.")) + return fs[:MAX_FINDINGS],[_evidence("runtime","runtime_rules","observed",{"finding_count":len(fs),"health":health})],[] +def _finding(t:AuditTarget,cid:str,title:str,sev:Severity,conf:Confidence,eref:str,impact:str,rem:str,ver:str)->Finding: + suffix=hashlib.sha256(f"{t.agent_name}|{cid}|{title}".encode()).hexdigest()[:10]; return Finding(id=f"{cid.lower()}-{suffix}",control_id=cid,title=title,severity=sev,confidence=conf,evidence_refs=[eref],impact=impact,remediation=rem,verification_procedure=ver) +def _dedupe(findings:list[Finding])->list[Finding]: + rank={Severity.CRITICAL:5,Severity.HIGH:4,Severity.MEDIUM:3,Severity.LOW:2,Severity.INFO:1}; seen={} + for f in findings: + if f.id not in seen or rank[f.severity]>rank[seen[f.id].severity]: seen[f.id]=f + return sorted(seen.values(),key=lambda f:(-rank[f.severity],f.control_id,f.id))[:MAX_FINDINGS] +def _risk_summary(findings:list[Finding])->dict[str,Any]: + weights={Severity.CRITICAL:100,Severity.HIGH:40,Severity.MEDIUM:15,Severity.LOW:5,Severity.INFO:1}; counts={s.value:0 for s in Severity}; score=0 + for f in findings: counts[f.severity.value]+=1; score+=weights[f.severity] + score=min(score,100); rating="critical" if counts["critical"] else "high" if score>=70 else "medium" if score>=30 else "low" if score else "clean" + return {"schema_version":"2026-07-13","risk_score":score,"risk_rating":rating,"finding_counts":counts,"top_controls":sorted({f.control_id for f in findings})[:20],"unknown_or_not_tested":[f.control_id for f in findings if f.status in {"unknown","not_tested"}],"mutations_performed":False} +def _remediation_md(t:AuditTarget,audit_id:str,findings:list[Finding],summary:dict[str,Any])->str: + lines=[f"# Remediation plan for {t.agent_name}","",f"Audit ID: `{audit_id}`",f"Risk rating: **{summary.get('risk_rating','unknown')}** ({summary.get('risk_score',0)}/100)","","This plan is read-only output. No production systems or repositories were changed.",""] + if not findings: lines += ["## No open findings","","The synthetic audit found no policy violations in the supplied evidence. Unknown/not-tested areas should still be reviewed."] + for f in findings: lines += [f"## {f.severity.value.upper()} — {f.title}","",f"- Finding ID: `{f.id}`",f"- Control: `{f.control_id}`",f"- Confidence: `{f.confidence.value}`",f"- Evidence references: {', '.join(f.evidence_refs) or 'none'}",f"- Impact: {f.impact}",f"- Remediation: {f.remediation}",f"- Verification: {f.verification_procedure}",""] + return "\n".join(lines) +def _sarif(t:AuditTarget,findings:list[Finding])->dict[str,Any]: + rules=[{"id":f.control_id,"name":f.title,"shortDescription":{"text":f.title},"help":{"text":f.remediation}} for f in findings] + results=[{"ruleId":f.control_id,"level":"error" if f.severity in {Severity.CRITICAL,Severity.HIGH} else "warning" if f.severity is Severity.MEDIUM else "note","message":{"text":f"{f.title}: {f.impact}"},"partialFingerprints":{"findingId":f.id},"properties":{"severity":f.severity.value,"confidence":f.confidence.value,"verification":f.verification_procedure}} for f in findings] + return {"version":"2.1.0","$schema":"https://json.schemastore.org/sarif-2.1.0.json","runs":[{"tool":{"driver":{"name":"agent-compliance-auditor","rules":rules}},"invocations":[{"executionSuccessful":True,"properties":{"target":t.agent_name,"mutations_performed":False}}],"results":results}]} +def _repo_summary(files:dict[str,str])->dict[str,Any]: + paths=[p for p,t in files.items() if any(tok in t.lower() for tok in ("ignore previous instructions","exfiltrate","send secrets","system prompt"))]; return {"files_inspected":len(files),"paths":sorted(files),"suspicious_instruction_count":len(paths),"suspicious_instruction_paths":paths[:20],"bounded_bytes_per_file":MAX_TEXT_BYTES} +def _manifests(files:dict[str,str])->dict[str,Any]: return {p:({"present":True,"sha256":_sha(files[p]),"excerpt":_redact(files[p][:1200])} if p in files else {"present":False}) for p in ("a2a.yaml","agent.py","requirements.txt","Dockerfile")} +def _skills(card:Any,files:dict[str,str])->list[dict[str,Any]]: + if isinstance(card,dict) and isinstance(card.get("skills"),list): return [s for s in card["skills"] if isinstance(s,dict)] + return [{"name":n,"source":"agent.py"} for n in re.findall(r"async\s+def\s+([a-zA-Z_][a-zA-Z0-9_]*)\s*\(",files.get("agent.py",""))] +def _schemas(card:Any)->dict[str,Any]: + out={} + if isinstance(card,dict): + for s in card.get("skills") or []: + if isinstance(s,dict): out[str(s.get("name") or s.get("id") or "unknown")]={"input_schema":s.get("input_schema") or s.get("inputSchema"),"output_schema":s.get("output_schema") or s.get("outputSchema")} + return out +def _secret_refs(files:dict[str,str])->list[dict[str,str]]: + refs=[] + for path,text in files.items(): + for pattern in SECRET_PATTERNS: + for found in pattern.finditer(text): + refs.append({"path": path, "match": _redact(found.group(0))}) + for indicator in WRITE_INDICATORS: + if indicator in text: + refs.append({"path": path, "match": _redact(indicator)}) + return refs[:50] +def _exposure(manifests:dict[str,Any])->dict[str,Any]: + text=json.dumps(manifests,default=str).lower(); public=True if "public: true" in text or "public=true" in text else False if "public: false" in text or "public=false" in text else None + return {"public":public,"documented_public_reason":"public_reason" in text or "exposure_reason" in text,"unknown_when_null":public is None} +def _resources(card:Any,manifests:dict[str,Any])->dict[str,Any]: + runtime=card.get("runtime") if isinstance(card,dict) else {}; res=runtime.get("resources") if isinstance(runtime,dict) else {}; out=dict(res or {}) if isinstance(res,dict) else {} + if "max_runtime_seconds" not in out: + resource_match = re.search(r"max_runtime_seconds[\s:=]+([0-9]+)",json.dumps(manifests,default=str)) + out["max_runtime_seconds"] = int(resource_match.group(1)) if resource_match else "unknown" + return out +def _provenance(manifests:dict[str,Any],files:dict[str,str])->dict[str,Any]: + docker=files.get("Dockerfile",""); return {"dockerfile_present":bool(docker),"digest_pinned_base_image":("@sha256:" in docker) if docker else "unknown","source_hashes":{k:v.get("sha256") for k,v in manifests.items() if isinstance(v,dict) and v.get("sha256")}} +def _rbac(manifests:dict[str,Any])->dict[str,Any]: + text=json.dumps(manifests,default=str).lower(); return {"service_account_mentions":text.count("serviceaccount"),"cluster_role_mentions":text.count("clusterrole"),"status":"observed" if "role" in text else "unknown"} +def _unknowns(card:dict[str,Any],manifests:dict[str,Any],health:dict[str,Any],files:dict[str,str])->list[str]: + out=[] + if not card.get("card"): out.append("agent_card") + if not files: out.append("repository_files") + if health.get("status") in {"unknown","not_tested"}: out.append("deployment_health") + if not any(v.get("present") for v in manifests.values() if isinstance(v,dict)): out.append("source_manifests") + return out +def _controls()->list[dict[str,str]]: return [{"control_id":cid,"title":title} for cid,title in [("A2A-CARD-001","Agent Card availability and integrity"),("A2A-SCHEMA-001","Public skill inventory"),("A2A-SCHEMA-002","Strict non-null schemas"),("A2A-NET-001","Network exposure minimization"),("A2A-OPS-001","Runtime resource limits"),("A2A-SECRET-001","Secret handling and redaction"),("A2A-UNTRUSTED-001","Untrusted content isolation"),("A2A-TENANT-001","Tenant and namespace isolation"),("A2A-HEALTH-001","Deployment health")]] +def _repo_evidence(files:dict[str,str])->list[Evidence]: return [_evidence(f"repo:{p}","repository","observed",t[:1000],path=p) for p,t in list(files.items())[:MAX_FILES_PER_AUDIT]] +def _evidence(eid:str,source:str,status:Literal["observed","missing","unknown","not_tested","redacted"],payload:Any,path:str|None=None)->Evidence: + text=payload if isinstance(payload,str) else json.dumps(payload,sort_keys=True,default=str); return Evidence(id=eid,source=source,path=path,status=status,excerpt=_redact(text)[:MAX_EVIDENCE_BYTES],sha256=_sha(text)) +def _audit_id(target:AuditTarget,supplied:str|None)->str: + if supplied: return supplied + seed=f"{target.agent_name}|{target.repository or ''}|{int(time.time())}"; return f"audit-{target.agent_name}-{hashlib.sha256(seed.encode()).hexdigest()[:12]}" +def _output_dir(audit_id:str)->str: return f"outputs/compliance/{re.sub(r'[^A-Za-z0-9_.-]','-',audit_id)[:80]}" +def _path_allowed(path:str)->bool: + clean=path.replace("\\","/").lstrip("/"); return ".." not in clean.split("/") and not any(fnmatch(clean,p) for p in DENIED_REPO_GLOBS) and any(fnmatch(clean,p) for p in ALLOWED_REPO_GLOBS) +def _safe_url(parsed:Any,allowed_hosts:set[str])->bool: + host=(parsed.hostname or "").lower(); return parsed.scheme in {"https","http"} and bool(host) and host not in {"localhost","127.0.0.1","0.0.0.0"} and not host.startswith("169.254.") and not host.endswith(".local") and host in allowed_hosts +def _bounded_json(text:str,warnings:list[str])->Any: + if len(text.encode())>MAX_TEXT_BYTES: warnings.append("HTTP JSON response exceeded evidence bound and was truncated before parsing."); text=text[:MAX_TEXT_BYTES] + try: return json.loads(text) + except json.JSONDecodeError: warnings.append("HTTP response was not valid JSON."); return {"invalid_json_excerpt":_redact(text[:1000])} +def _redact_match(redact_match: re.Match[str]) -> str: + return redact_match.group(0)[:8]+"…REDACTED" +def _redact(text:str)->str: + out=text + for pat in SECRET_PATTERNS: out=pat.sub(_redact_match,out) + return out +def _redact_obj(value:Any)->Any: + if isinstance(value,str): return _redact(value) + if isinstance(value,list): return [_redact_obj(v) for v in value] + if isinstance(value,dict): return {str(k): _redact_obj(v) for k,v in value.items()} + return value +def _sha(text:str)->str: return hashlib.sha256(text.encode("utf-8",errors="replace")).hexdigest() +def _dump(value:Any)->Any: return value.model_dump(mode="json") if isinstance(value,BaseModel) else value +async def _event(ctx:RunContext[NoAuth],kind:str,**payload:Any)->None: + safe=json.loads(json.dumps(payload,default=str)); await ctx.emit_event(AgentEvent(kind=kind,payload=safe)) + if kind in {"audit_started","audit_progress","audit_completed"}: await ctx.emit_progress(f"{kind}: {safe.get('phase') or safe.get('skill') or kind}")