bobseares-2/agent-compliance-auditor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

a2a-source-edit: write agent.py

Browse Source
This commit is contained in:
a2a-cloud
2026-07-13 03:18:53 +00:00
parent d11f4f444c
commit af52d87ce9
1 changed files with 181 additions and 477 deletions
Download Patch File Download Diff File Expand all files Collapse all files

658
agent.py
View File

@@ -1,4 +1,3 @@
"""Deterministic read-only compliance auditor for A2A agents."""
from __future__ import annotations from __future__ import annotations
import hashlib import hashlib
@@ -15,18 +14,7 @@ import httpx
from pydantic import AnyHttpUrl, BaseModel, ConfigDict, Field, field_validator from pydantic import AnyHttpUrl, BaseModel, ConfigDict, Field, field_validator
import a2a_pack as a2a import a2a_pack as a2a
from a2a_pack import ( from a2a_pack import A2AAgent, ConsumerSetup, ConsumerSetupField, EgressPolicy, NoAuth, Pricing, Resources, RunContext, WorkspaceAccess, WorkspaceMode
A2AAgent,
ConsumerSetup,
ConsumerSetupField,
EgressPolicy,
NoAuth,
Pricing,
Resources,
RunContext,
WorkspaceAccess,
WorkspaceMode,
)
from a2a_pack.context import AgentEvent from a2a_pack.context import AgentEvent
from a2a_pack.workspace import FileType from a2a_pack.workspace import FileType
@@ -35,39 +23,20 @@ MAX_EVIDENCE_BYTES = 8_000
MAX_FILES_PER_AUDIT = 80 MAX_FILES_PER_AUDIT = 80
MAX_FINDINGS = 200 MAX_FINDINGS = 200
HTTP_TIMEOUT_SECONDS = 8.0 HTTP_TIMEOUT_SECONDS = 8.0
DENIED_REPO_GLOBS = (".git/**", "**/.env", "**/.env.*", "**/*secret*", "**/*credential*", "**/id_rsa", "**/*.pem", "**/*.key") DENIED_REPO_GLOBS = (".git/**", "**/.env", "**/.env.*", "**/*secret*", "**/*credential*", "**/*.pem", "**/*.key")
ALLOWED_REPO_GLOBS = ("agent.py", "a2a.yaml", "requirements.txt", "README.md", "pyproject.toml", "Dockerfile", "*.py", "*.yaml", "*.yml", "*.json", "*.md", "tests/*.py", "tests/**/*.py", "skills/**/SKILL.md") 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 = ( 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,}"))
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") 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): class StrictModel(BaseModel):
model_config = ConfigDict(extra="forbid") model_config = ConfigDict(extra="forbid")
class Severity(str, Enum): class Severity(str, Enum):
CRITICAL = "critical" CRITICAL="critical"; HIGH="high"; MEDIUM="medium"; LOW="low"; INFO="info"
HIGH = "high"
MEDIUM = "medium"
LOW = "low"
INFO = "info"
class Confidence(str, Enum): class Confidence(str, Enum):
HIGH = "high" HIGH="high"; MEDIUM="medium"; LOW="low"
MEDIUM = "medium"
LOW = "low"
class ProbeMode(str, Enum): class ProbeMode(str, Enum):
NONE = "none" NONE="none"; HEALTH_ONLY="health_only"; SYNTHETIC="synthetic"
HEALTH_ONLY = "health_only"
SYNTHETIC = "synthetic"
class AuditTarget(StrictModel): class AuditTarget(StrictModel):
agent_name: str = Field(min_length=1, max_length=128, pattern=r"^[a-z0-9][a-z0-9-]{0,127}$") agent_name: str = Field(min_length=1, max_length=128, pattern=r"^[a-z0-9][a-z0-9-]{0,127}$")
@@ -75,14 +44,10 @@ class AuditTarget(StrictModel):
repository: str | None = Field(default=None, max_length=256) repository: str | None = Field(default=None, max_length=256)
namespace: str | None = Field(default=None, max_length=128) namespace: str | None = Field(default=None, max_length=128)
deployment: str | None = Field(default=None, max_length=128) deployment: str | None = Field(default=None, max_length=128)
@field_validator("repository", "namespace", "deployment") @field_validator("repository", "namespace", "deployment")
@classmethod @classmethod
def clean_optional(cls, value: str | None) -> str | None: def clean_optional(cls, value: str | None) -> str | None:
if value is None: return value.strip() or None if value is not None else None
return None
return value.strip() or None
class AuditOptions(StrictModel): class AuditOptions(StrictModel):
audit_id: str | None = Field(default=None, max_length=80, pattern=r"^[A-Za-z0-9_.-]+$") audit_id: str | None = Field(default=None, max_length=80, pattern=r"^[A-Za-z0-9_.-]+$")
@@ -94,7 +59,6 @@ class AuditOptions(StrictModel):
allowed_hosts: list[str] = Field(default_factory=list, max_length=20) allowed_hosts: list[str] = Field(default_factory=list, max_length=20)
allowed_namespaces: list[str] = Field(default_factory=list, max_length=20) allowed_namespaces: list[str] = Field(default_factory=list, max_length=20)
class Finding(StrictModel): class Finding(StrictModel):
id: str id: str
control_id: str control_id: str
@@ -107,7 +71,6 @@ class Finding(StrictModel):
remediation: str remediation: str
verification_procedure: str verification_procedure: str
class Evidence(StrictModel): class Evidence(StrictModel):
id: str id: str
source: str source: str
@@ -116,503 +79,244 @@ class Evidence(StrictModel):
excerpt: str = Field(default="", max_length=MAX_EVIDENCE_BYTES) excerpt: str = Field(default="", max_length=MAX_EVIDENCE_BYTES)
sha256: str | None = None sha256: str | None = None
class OutputFile(StrictModel): class OutputFile(StrictModel):
path: str path: str
artifact_uri: str | None = None artifact_uri: str | None = None
mime_type: str mime_type: str
size_bytes: int = Field(ge=0) size_bytes: int = Field(ge=0)
class InventoryRequest(StrictModel): class InventoryRequest(StrictModel):
target: AuditTarget target: AuditTarget
options: AuditOptions = Field(default_factory=AuditOptions) options: AuditOptions = Field(default_factory=AuditOptions)
class AuditConfigurationRequest(StrictModel): class AuditConfigurationRequest(StrictModel):
target: AuditTarget target: AuditTarget
options: AuditOptions = Field(default_factory=AuditOptions) options: AuditOptions = Field(default_factory=AuditOptions)
inventory: dict[str, Any] | None = None inventory: dict[str, Any] | None = None
class AuditRuntimeRequest(StrictModel): class AuditRuntimeRequest(StrictModel):
target: AuditTarget target: AuditTarget
options: AuditOptions = Field(default_factory=AuditOptions) options: AuditOptions = Field(default_factory=AuditOptions)
inventory: dict[str, Any] | None = None inventory: dict[str, Any] | None = None
class AssessRiskRequest(StrictModel): class AssessRiskRequest(StrictModel):
target: AuditTarget target: AuditTarget
options: AuditOptions = Field(default_factory=AuditOptions) options: AuditOptions = Field(default_factory=AuditOptions)
inventory: dict[str, Any] | None = None inventory: dict[str, Any] | None = None
configuration_findings: list[Finding] = Field(default_factory=list, max_length=MAX_FINDINGS) configuration_findings: list[Finding] = Field(default_factory=list, max_length=MAX_FINDINGS)
runtime_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): class GenerateRemediationRequest(StrictModel):
target: AuditTarget target: AuditTarget
options: AuditOptions = Field(default_factory=AuditOptions) options: AuditOptions = Field(default_factory=AuditOptions)
findings: list[Finding] = Field(default_factory=list, max_length=MAX_FINDINGS) findings: list[Finding] = Field(default_factory=list, max_length=MAX_FINDINGS)
risk_summary: dict[str, Any] | None = None risk_summary: dict[str, Any] | None = None
class InventoryResult(StrictModel): class InventoryResult(StrictModel):
audit_id: str audit_id: str; target: AuditTarget; status: Literal["ok", "partial", "unreachable", "setup_required"]; output_dir: str
target: AuditTarget 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
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): class FindingResult(StrictModel):
audit_id: str audit_id: str; target: AuditTarget; status: Literal["ok", "partial", "setup_required"]; output_dir: str
target: AuditTarget 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
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): class RiskSummaryResult(StrictModel):
audit_id: str audit_id: str; target: AuditTarget; status: Literal["ok", "partial"]; output_dir: str
target: AuditTarget 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
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): class RemediationResult(StrictModel):
audit_id: str audit_id: str; target: AuditTarget; status: Literal["ok", "partial"]; output_dir: str
target: AuditTarget 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
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): class AgentComplianceAuditorConfig(StrictModel):
default_allowed_hosts: list[str] = Field(default_factory=list, max_length=50) default_allowed_hosts: list[str] = Field(default_factory=list, max_length=50)
default_allowed_namespaces: 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]): class AgentComplianceAuditor(A2AAgent[AgentComplianceAuditorConfig, NoAuth]):
name = "agent-compliance-auditor" name="agent-compliance-auditor"; description="Read-only compliance auditor for deployed A2A agents and repositories."; version="0.1.0"
description = "Read-only compliance auditor for deployed A2A agents and repositories." config_model=AgentComplianceAuditorConfig; auth_model=NoAuth
version = "0.1.0" pricing=Pricing(price_per_call_usd=0.0, caller_pays_llm=False, notes="Deterministic read-only auditing; no LLM calls and no production mutations.")
config_model = AgentComplianceAuditorConfig resources=Resources(cpu="500m", memory="512Mi", max_runtime_seconds=900)
auth_model = NoAuth workspace_access=WorkspaceAccess.dynamic(max_files=MAX_FILES_PER_AUDIT+10, allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), require_reason=False, deny_patterns=DENIED_REPO_GLOBS, max_total_size_bytes=12*1024*1024)
pricing = Pricing(price_per_call_usd=0.0, caller_pays_llm=False, notes="Deterministic read-only auditing; no LLM calls and no production mutations.") egress=EgressPolicy(allow_hosts=("gitea.a2a.svc.cluster.local","argocd-server.argocd.svc.cluster.local","kubernetes.default.svc","registry-1.docker.io","ghcr.io"), deny_internet_by_default=True)
resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=900) tools_used=("httpx","pydantic","workspace")
workspace_access = WorkspaceAccess.dynamic(max_files=MAX_FILES_PER_AUDIT + 10, allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), require_reason=False, deny_patterns=DENIED_REPO_GLOBS, max_total_size_bytes=12 * 1024 * 1024) consumer_setup=ConsumerSetup.from_fields(
egress = EgressPolicy(allow_hosts=("gitea.a2a.svc.cluster.local", "argocd-server.argocd.svc.cluster.local", "kubernetes.default.svc", "registry-1.docker.io", "ghcr.io"), deny_internet_by_default=True) ConsumerSetupField.config("GITEA_BASE_URL", label="Gitea base URL", required=False, input_type="url"), ConsumerSetupField.secret("GITEA_TOKEN", label="Gitea read-only token", required=False),
tools_used = ("httpx", "pydantic", "workspace") ConsumerSetupField.config("KUBERNETES_BASE_URL", label="Kubernetes API base URL", required=False, input_type="url"), ConsumerSetupField.secret("KUBERNETES_TOKEN", label="Kubernetes read-only bearer token", required=False),
consumer_setup = ConsumerSetup.from_fields( ConsumerSetupField.config("ARGOCD_BASE_URL", label="Argo CD base URL", required=False, input_type="url"), ConsumerSetupField.secret("ARGOCD_TOKEN", label="Argo CD read-only token", required=False),
ConsumerSetupField.config("GITEA_BASE_URL", label="Gitea base URL", required=False, input_type="url"), ConsumerSetupField.config("REGISTRY_BASE_URL", label="Registry metadata base URL", required=False, input_type="url"), ConsumerSetupField.secret("REGISTRY_TOKEN", label="Registry read-only token", required=False))
ConsumerSetupField.secret("GITEA_TOKEN", label="Gitea read-only token", required=False),
ConsumerSetupField.config("KUBERNETES_BASE_URL", label="Kubernetes API base URL", required=False, input_type="url"),
ConsumerSetupField.secret("KUBERNETES_TOKEN", label="Kubernetes read-only bearer token", required=False),
ConsumerSetupField.config("ARGOCD_BASE_URL", label="Argo CD base URL", required=False, input_type="url"),
ConsumerSetupField.secret("ARGOCD_TOKEN", label="Argo CD read-only token", required=False),
ConsumerSetupField.config("REGISTRY_BASE_URL", label="Registry metadata base URL", required=False, input_type="url"),
ConsumerSetupField.secret("REGISTRY_TOKEN", label="Registry read-only token", required=False),
)
@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") @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: async def inventory_agent(self, ctx: RunContext[NoAuth], request: InventoryRequest) -> InventoryResult:
audit_id = _audit_id(request.target, request.options.audit_id) 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)
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)
inventory, evidence, warnings, status = await _collect_inventory(self, ctx, request.target, request.options, 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)
output_dir = _output_dir(audit_id) return InventoryResult(audit_id=audit_id,target=request.target,status=status,output_dir=output_dir,generated_files=files,inventory=inventory,evidence=evidence,warnings=warnings)
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") @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: async def audit_configuration(self, ctx: RunContext[NoAuth], request: AuditConfigurationRequest) -> FindingResult:
audit_id = _audit_id(request.target, request.options.audit_id) 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)
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)
inventory = request.inventory or (await _collect_inventory(self, ctx, request.target, request.options, audit_id))[0] 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))
findings, evidence, warnings = _configuration_findings(request.target, inventory) return FindingResult(audit_id=audit_id,target=request.target,status="ok",output_dir=output_dir,generated_files=files,findings=findings,evidence=evidence,warnings=warnings)
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") @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: async def audit_runtime(self, ctx: RunContext[NoAuth], request: AuditRuntimeRequest) -> FindingResult:
audit_id = _audit_id(request.target, request.options.audit_id) 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)
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)
inventory = request.inventory or (await _collect_inventory(self, ctx, request.target, request.options, audit_id))[0] 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))
findings, evidence, warnings = _runtime_findings(request.target, inventory) return FindingResult(audit_id=audit_id,target=request.target,status="ok",output_dir=output_dir,generated_files=files,findings=findings,evidence=evidence,warnings=warnings)
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") @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: async def assess_risk(self, ctx: RunContext[NoAuth], request: AssessRiskRequest) -> RiskSummaryResult:
audit_id = _audit_id(request.target, request.options.audit_id) 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)
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)
all_findings = _dedupe([*request.configuration_findings, *request.runtime_findings]) 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"])
summary = _risk_summary(all_findings) return RiskSummaryResult(audit_id=audit_id,target=request.target,status="ok",output_dir=output_dir,generated_files=files,risk_summary=summary,findings=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 all_findings]}}, [])
await _event(ctx, "audit_completed", audit_id=audit_id, skill="assess_risk", findings=len(all_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=all_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") @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: async def generate_remediation(self, ctx: RunContext[NoAuth], request: GenerateRemediationRequest) -> RemediationResult:
audit_id = _audit_id(request.target, request.options.audit_id) 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)
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)
findings = _dedupe(request.findings) 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))
risk_summary = request.risk_summary or _risk_summary(findings) return RemediationResult(audit_id=audit_id,target=request.target,status="ok",output_dir=output_dir,generated_files=files,remediation_markdown=md,sarif=sarif)
markdown = _remediation_md(request.target, audit_id, findings, risk_summary)
sarif = _sarif(request.target, findings)
output_dir = _output_dir(audit_id)
files = await _write_outputs(ctx, output_dir, {"remediation.md": markdown, "sarif.json": sarif, "risk-summary.json": risk_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=markdown, 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 _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"]]: async def _fetch_card(agent_url:str|None,allowed_hosts:set[str],warnings:list[str])->dict[str,Any]:
warnings: list[str] = [] if not agent_url: return {"status":"unknown","reason":"agent_url not provided"}
evidence: list[Evidence] = [] 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"}
allowed_hosts = {h.lower().strip() for h in [*agent.config.default_allowed_hosts, *options.allowed_hosts] if h.strip()}
allowed_namespaces = {n.strip() for n in [*agent.config.default_allowed_namespaces, *options.allowed_namespaces] if n.strip()}
authorized = not allowed_namespaces or bool(target.namespace and target.namespace in allowed_namespaces)
status: Literal["ok", "partial", "unreachable", "setup_required"] = "ok" if authorized else "partial"
if not authorized:
warnings.append("Target namespace is outside the configured allowlist; runtime probes skipped.")
await _event(ctx, "audit_progress", audit_id=audit_id, phase="agent_card")
card = await _fetch_card(str(target.agent_url) if target.agent_url else None, allowed_hosts, warnings)
if card.get("status") == "unreachable" and status == "ok":
status = "unreachable"
evidence.append(_evidence("card", "agent_card", "observed" if card.get("card") else "unknown", card))
await _event(ctx, "audit_progress", audit_id=audit_id, phase="repository_static_inspection")
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:
await _event(ctx, "audit_progress", audit_id=audit_id, phase="bounded_runtime_probe")
health = await _health(str(target.agent_url) if target.agent_url else None, allowed_hosts, warnings)
inventory = {
"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 inventory, evidence, warnings, status
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"}
parsed = urlparse(agent_url)
if not _safe_url(parsed, 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: try:
async with httpx.AsyncClient(timeout=httpx.Timeout(HTTP_TIMEOUT_SECONDS), follow_redirects=False) as client: async with httpx.AsyncClient(timeout=httpx.Timeout(HTTP_TIMEOUT_SECONDS),follow_redirects=False) as client: resp=await client.get(agent_url.rstrip("/")+"/.well-known/agent-card",headers={"accept":"application/json"})
resp = await client.get(agent_url.rstrip("/") + "/.well-known/agent-card", headers={"accept": "application/json"}) return {"status":"unreachable","http_status":resp.status_code,"body_excerpt":_redact(resp.text[:500])} if resp.status_code>=400 else {"status":"ok","card":_bounded_json(resp.text,warnings)}
if resp.status_code >= 400: except Exception as exc: warnings.append(f"Agent card fetch failed: {type(exc).__name__}"); return {"status":"unreachable","error_type":type(exc).__name__}
return {"status": "unreachable", "http_status": resp.status_code, "body_excerpt": _redact(resp.text[:500])} async def _health(agent_url:str|None,allowed_hosts:set[str],warnings:list[str])->dict[str,Any]:
return {"status": "ok", "card": _bounded_json(resp.text, warnings)} if not agent_url: return {"status":"unknown","reason":"agent_url not provided"}
except Exception as exc: # noqa: BLE001 if not _safe_url(urlparse(agent_url),allowed_hosts): return {"status":"not_tested","reason":"host_not_allowlisted"}
warnings.append(f"Agent card fetch failed: {type(exc).__name__}") started=time.perf_counter()
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: try:
async with httpx.AsyncClient(timeout=httpx.Timeout(HTTP_TIMEOUT_SECONDS), follow_redirects=False) as client: async with httpx.AsyncClient(timeout=httpx.Timeout(HTTP_TIMEOUT_SECONDS),follow_redirects=False) as client: resp=await client.get(agent_url.rstrip("/")+"/healthz")
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)}
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__}
except Exception as exc: # noqa: BLE001 async def _read_repo(ctx:RunContext[NoAuth],max_files:int,warnings:list[str])->dict[str,str]:
warnings.append(f"Health probe failed: {type(exc).__name__}") if max_files<=0: return {}
return {"status": "unreachable", "error_type": type(exc).__name__} 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={}
async def _read_repo(ctx: RunContext[NoAuth], max_files: int, warnings: list[str]) -> dict[str, str]: for m in view.files[:max_files]:
if max_files <= 0: path=m.path.replace("\\","/").lstrip("/")
return {} if _path_allowed(path):
try: try: files[path]=(await view.read(path))[:MAX_TEXT_BYTES].decode("utf-8",errors="replace")
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"Could not read {path}: {type(exc).__name__}")
except Exception as exc: # noqa: BLE001
warnings.append(f"Repository workspace unavailable: {type(exc).__name__}")
return {}
files: dict[str, str] = {}
for match in view.files[:max_files]:
path = match.path.replace("\\", "/").lstrip("/")
if not _path_allowed(path):
continue
try:
files[path] = (await view.read(path))[:MAX_TEXT_BYTES].decode("utf-8", errors="replace")
except Exception as exc: # noqa: BLE001
warnings.append(f"Could not read {path}: {type(exc).__name__}")
return files return files
async def _write_outputs(ctx:RunContext[NoAuth],output_dir:str,files:dict[str,Any],warnings:list[str])->list[OutputFile]:
out=[]
async def _write_outputs(ctx: RunContext[NoAuth], output_dir: str, files: dict[str, Any], warnings: list[str]) -> list[OutputFile]: for name,payload in files.items():
out: list[OutputFile] = [] 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
for name, payload in files.items(): try: ref=await ctx.write_artifact(rel,data,mime); await ctx.emit_artifact(ref); uri=ref.uri
rel = f"{output_dir}/{name}" except Exception as exc: warnings.append(f"Artifact write failed for {rel}: {type(exc).__name__}")
data = payload.encode("utf-8") if isinstance(payload, str) else json.dumps(payload, indent=2, sort_keys=True, default=str).encode("utf-8") out.append(OutputFile(path=rel,artifact_uri=uri,mime_type=mime,size_bytes=len(data)))
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: # noqa: BLE001
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 return out
def _configuration_findings(target:AuditTarget,inv:dict[str,Any])->tuple[list[Finding],list[Evidence],list[str]]:
def _configuration_findings(target: AuditTarget, inventory: 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 []
findings: list[Finding] = [] 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."))
card = (inventory.get("agent_card") or {}).get("card") or {} 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."))
skills = inventory.get("skills") or [] for s in skills:
if not card: schema=s.get("input_schema") or s.get("inputSchema") or {}; out=s.get("output_schema") or s.get("outputSchema")
findings.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 over an approved host and returns a valid A2A Card.", "Fetch /.well-known/agent-card and validate metadata.")) 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','<unknown>')} has weak schema",Severity.MEDIUM,Confidence.HIGH,"schemas","Weak schemas increase integration errors.","Use strict Pydantic schemas.","Inspect Agent Card schemas."))
if not skills: 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."))
findings.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 with input and output schemas.", "Reload the live Agent Card and confirm skills[].input_schema and output_schema.")) 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."))
for skill in skills: 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."))
schema = skill.get("input_schema") or skill.get("inputSchema") or {} 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."))
out_schema = skill.get("output_schema") or skill.get("outputSchema") or {} return fs[:MAX_FINDINGS],[_evidence("configuration","configuration_rules","observed",{"finding_count":len(fs)})],[]
if not (isinstance(schema, dict) and schema.get("type") == "object" and schema.get("additionalProperties") is False) or not out_schema: def _runtime_findings(target:AuditTarget,inv:dict[str,Any])->tuple[list[Finding],list[Evidence],list[str]]:
findings.append(_finding(target, "A2A-SCHEMA-002", f"Skill {skill.get('name', '<unknown>')} has weak schema", Severity.MEDIUM, Confidence.HIGH, "schemas", "Weak schemas increase integration errors and injection surface.", "Use strict Pydantic models or explicit JSON Schema with additionalProperties=false.", "Inspect the Agent Card schemas and run negative validation tests.")) fs=[]; health=inv.get("deployment_health") or {"status":"unknown"}
if (inventory.get("network_exposure") or {}).get("public") is True: 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."))
findings.append(_finding(target, "A2A-NET-001", "Agent is publicly exposed", Severity.MEDIUM, Confidence.MEDIUM, "network_exposure", "Unexpected public exposure can broaden abuse and exfiltration paths.", "Set expose.public=false unless public access is required and documented.", "Review a2a.yaml and ingress policy.")) 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 (inventory.get("resource_limits") or {}).get("max_runtime_seconds") in (None, "unknown"): 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."))
findings.append(_finding(target, "A2A-OPS-001", "Runtime limit is unknown", Severity.LOW, Confidence.MEDIUM, "resource_limits", "Unknown runtime budgets can cause reliability and cost issues.", "Declare Resources and mirror runtime.resources in a2a.yaml.", "Load Agent Card runtime.resources and deployment manifest.")) return fs[:MAX_FINDINGS],[_evidence("runtime","runtime_rules","observed",{"finding_count":len(fs),"health":health})],[]
if inventory.get("secrets_references"): def _finding(t:AuditTarget,cid:str,title:str,sev:Severity,conf:Confidence,eref:str,impact:str,rem:str,ver:str)->Finding:
findings.append(_finding(target, "A2A-SECRET-001", "Potential secret or provider-key reference found in source", Severity.HIGH, Confidence.HIGH, "secrets_references", "Secrets in source or logs can be exfiltrated.", "Remove literal secrets and broad provider-key reads; use declared consumer setup or runtime secrets.", "Run secret scanning and verify no artifacts contain secret values.")) 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)
if (inventory.get("repository") or {}).get("suspicious_instruction_count", 0) > 0: def _dedupe(findings:list[Finding])->list[Finding]:
findings.append(_finding(target, "A2A-UNTRUSTED-001", "Repository contains malicious instruction-like text", Severity.MEDIUM, Confidence.HIGH, "repository", "Auditors or LLM-backed tools could be manipulated if repo text is treated as instructions.", "Keep repository content in evidence fields only; never treat it as instructions.", "Confirm malicious text appears only as redacted evidence.")) rank={Severity.CRITICAL:5,Severity.HIGH:4,Severity.MEDIUM:3,Severity.LOW:2,Severity.INFO:1}; seen={}
return findings[:MAX_FINDINGS], [_evidence("configuration", "configuration_rules", "observed", {"finding_count": len(findings)})], []
def _runtime_findings(target: AuditTarget, inventory: dict[str, Any]) -> tuple[list[Finding], list[Evidence], list[str]]:
findings: list[Finding] = []
health = inventory.get("deployment_health") or {"status": "unknown"}
if (inventory.get("identity") or {}).get("authorized") is False:
findings.append(_finding(target, "A2A-TENANT-001", "Runtime audit skipped by namespace allowlist", Severity.HIGH, Confidence.HIGH, "identity", "Auditing outside the authorized tenant boundary could violate isolation policy.", "Configure allowed_namespaces for the target tenant or run from the owning tenant.", "Confirm namespace allowlist contains only tenant-owned namespaces and retry."))
if health.get("status") in {"unreachable", "degraded"}:
findings.append(_finding(target, "A2A-HEALTH-001", "Agent health probe failed or degraded", Severity.MEDIUM, Confidence.MEDIUM, "deployment_health", "Consumers may experience failed calls or stale Agent Cards.", "Inspect deployment status, readiness probes, service routing, and rollout events.", "Repeat health-only probe and verify readiness."))
if health.get("status") in {"unknown", "not_tested"}:
findings.append(_finding(target, "A2A-HEALTH-002", "Runtime health state is unknown", Severity.LOW, Confidence.MEDIUM, "deployment_health", "Unknown runtime state leaves operational risk unmeasured.", "Provide an allowlisted agent_url and enable health_only probes.", "Run audit_runtime with probe_mode=health_only."))
return findings[:MAX_FINDINGS], [_evidence("runtime", "runtime_rules", "observed", {"finding_count": len(findings), "health": health})], []
def _finding(target: AuditTarget, control_id: str, title: str, severity: Severity, confidence: Confidence, evidence_ref: str, impact: str, remediation: str, verification: str) -> Finding:
suffix = hashlib.sha256(f"{target.agent_name}|{control_id}|{title}".encode()).hexdigest()[:10]
return Finding(id=f"{control_id.lower()}-{suffix}", control_id=control_id, title=title, severity=severity, confidence=confidence, evidence_refs=[evidence_ref], impact=impact, remediation=remediation, verification_procedure=verification)
def _dedupe(findings: list[Finding]) -> list[Finding]:
rank = {Severity.CRITICAL: 5, Severity.HIGH: 4, Severity.MEDIUM: 3, Severity.LOW: 2, Severity.INFO: 1}
seen: dict[str, Finding] = {}
for f in findings: for f in findings:
if f.id not in seen or rank[f.severity] > rank[seen[f.id].severity]: if f.id not in seen or rank[f.severity]>rank[seen[f.id].severity]: seen[f.id]=f
seen[f.id] = f return sorted(seen.values(),key=lambda f:(-rank[f.severity],f.control_id,f.id))[:MAX_FINDINGS]
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]
def _risk_summary(findings: list[Finding]) -> dict[str, Any]: 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"
weights = {Severity.CRITICAL: 100, Severity.HIGH: 40, Severity.MEDIUM: 15, Severity.LOW: 5, Severity.INFO: 1} 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}
counts = {s.value: 0 for s in Severity} def _remediation_md(t:AuditTarget,audit_id:str,findings:list[Finding],summary:dict[str,Any])->str:
score = 0 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.",""]
for f in findings: 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."]
counts[f.severity.value] += 1 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}",""]
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(target: AuditTarget, audit_id: str, findings: list[Finding], summary: dict[str, Any]) -> str:
lines = [f"# Remediation plan for {target.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) 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]
def _sarif(target: AuditTarget, findings: list[Finding]) -> dict[str, Any]: 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]
rules = [{"id": f.control_id, "name": f.title, "shortDescription": {"text": f.title}, "help": {"text": f.remediation}} 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}]}
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] def _repo_summary(files:dict[str,str])->dict[str,Any]:
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": target.agent_name, "mutations_performed": False}}], "results": results}]} 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]]:
def _repo_summary(files: dict[str, str]) -> dict[str, Any]: if isinstance(card,dict) and isinstance(card.get("skills"),list): return [s for s in card["skills"] if isinstance(s,dict)]
suspicious_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 [{"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",""))]
return {"files_inspected": len(files), "paths": sorted(files), "suspicious_instruction_count": len(suspicious_paths), "suspicious_instruction_paths": suspicious_paths[:20], "bounded_bytes_per_file": MAX_TEXT_BYTES} def _schemas(card:Any)->dict[str,Any]:
out={}
if isinstance(card,dict):
def _manifests(files: dict[str, str]) -> dict[str, Any]: for s in card.get("skills") or []:
return {path: ({"present": True, "sha256": _sha(files[path]), "excerpt": _redact(files[path][:1200])} if path in files else {"present": False}) for path in ("a2a.yaml", "agent.py", "requirements.txt", "Dockerfile")} 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")}
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: dict[str, Any] = {}
if isinstance(card, dict):
for skill in card.get("skills") or []:
if isinstance(skill, dict):
out[str(skill.get("name") or skill.get("id") or "unknown")] = {"input_schema": skill.get("input_schema") or skill.get("inputSchema"), "output_schema": skill.get("output_schema") or skill.get("outputSchema")}
return out return out
def _secret_refs(files:dict[str,str])->list[dict[str,str]]:
refs=[]
def _secret_refs(files: dict[str, str]) -> list[dict[str, str]]: for p,t in files.items():
refs: list[dict[str, str]] = [] for pat in SECRET_PATTERNS: refs += [{"path":p,"match":_redact(m.group(0))} for m in pat.finditer(t)]
for path, text in files.items(): refs += [{"path":p,"match":_redact(i)} for i in WRITE_INDICATORS if i in t]
for pattern in SECRET_PATTERNS:
refs += [{"path": path, "match": _redact(m.group(0))} for m in pattern.finditer(text)]
refs += [{"path": path, "match": _redact(ind)} for ind in WRITE_INDICATORS if ind in text]
return refs[:50] 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
def _exposure(manifests: dict[str, Any]) -> dict[str, Any]: return {"public":public,"documented_public_reason":"public_reason" in text or "exposure_reason" in text,"unknown_when_null":public is None}
text = json.dumps(manifests, default=str).lower() def _resources(card:Any,manifests:dict[str,Any])->dict[str,Any]:
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 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 {}
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 {}
resources = runtime.get("resources") if isinstance(runtime, dict) else {}
out = dict(resources or {}) if isinstance(resources, dict) else {}
if "max_runtime_seconds" not in out: if "max_runtime_seconds" not in out:
match = re.search(r"max_runtime_seconds[\s:=]+([0-9]+)", json.dumps(manifests, default=str)) m=re.search(r"max_runtime_seconds[\s:=]+([0-9]+)",json.dumps(manifests,default=str)); out["max_runtime_seconds"]=int(m.group(1)) if m else "unknown"
out["max_runtime_seconds"] = int(match.group(1)) if match else "unknown"
return out 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 _provenance(manifests: dict[str, Any], files: dict[str, str]) -> dict[str, Any]: def _rbac(manifests:dict[str,Any])->dict[str,Any]:
dockerfile = files.get("Dockerfile", "") 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"}
return {"dockerfile_present": bool(dockerfile), "digest_pinned_base_image": ("@sha256:" in dockerfile) if dockerfile else "unknown", "source_hashes": {k: v.get("sha256") for k, v in manifests.items() if isinstance(v, dict) and v.get("sha256")}} 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")
def _rbac(manifests: dict[str, Any]) -> dict[str, Any]: if not files: out.append("repository_files")
text = json.dumps(manifests, default=str).lower() if health.get("status") in {"unknown","not_tested"}: out.append("deployment_health")
return {"service_account_mentions": text.count("serviceaccount"), "cluster_role_mentions": text.count("clusterrole"), "status": "observed" if "role" in text else "unknown"} if not any(v.get("present") for v in manifests.values() if isinstance(v,dict)): out.append("source_manifests")
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 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 _controls() -> list[dict[str, str]]: def _evidence(eid:str,source:str,status:Literal["observed","missing","unknown","not_tested","redacted"],payload:Any,path:str|None=None)->Evidence:
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")]] 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
def _repo_evidence(files: dict[str, str]) -> list[Evidence]: seed=f"{target.agent_name}|{target.repository or ''}|{int(time.time())}"; return f"audit-{target.agent_name}-{hashlib.sha256(seed.encode()).hexdigest()[:12]}"
return [_evidence(f"repo:{path}", "repository", "observed", text[:1000], path=path) for path, text in list(files.items())[:MAX_FILES_PER_AUDIT]] 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 _evidence(eid: str, source: str, status: Literal["observed", "missing", "unknown", "not_tested", "redacted"], payload: Any, path: str | None = None) -> Evidence: def _safe_url(parsed:Any,allowed_hosts:set[str])->bool:
text = payload if isinstance(payload, str) else json.dumps(payload, sort_keys=True, default=str) 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
return Evidence(id=eid, source=source, path=path, status=status, excerpt=_redact(text)[:MAX_EVIDENCE_BYTES], sha256=_sha(text)) 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)
def _audit_id(target: AuditTarget, supplied: str | None) -> str: except json.JSONDecodeError: warnings.append("HTTP response was not valid JSON."); return {"invalid_json_excerpt":_redact(text[:1000])}
return supplied or f"audit-{target.agent_name}-{hashlib.sha256(f'{target.agent_name}|{target.repository or ""}|{int(time.time())}'.encode()).hexdigest()[:12]}" def _redact(text:str)->str:
out=text
for pat in SECRET_PATTERNS: out=pat.sub(lambda m:m.group(0)[:8]+"…REDACTED",out)
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("utf-8")) > 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(text: str) -> str:
out = text
for pattern in SECRET_PATTERNS:
out = pattern.sub(lambda m: m.group(0)[:8] + "…REDACTED", out)
return out return out
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
def _sha(text: str) -> str: async def _event(ctx:RunContext[NoAuth],kind:str,**payload:Any)->None:
return hashlib.sha256(text.encode("utf-8", errors="replace")).hexdigest() 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}")
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}")