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:15:39 +00:00
parent 132cfd6620
commit 095ec4a88e
1 changed files with 211 additions and 455 deletions
Download Patch File Download Diff File Expand all files Collapse all files

666
agent.py
View File

@@ -1,14 +1,6 @@
"""Read-only A2A agent compliance auditor. """Deterministic read-only compliance auditor for A2A agents."""
This agent is deterministic by design: it does not call an LLM, does not execute
arbitrary commands, does not mutate production systems, and treats all audited
content as untrusted evidence. External integrations are optional and
caller-configured; when they are absent the corresponding evidence is marked as
``unknown`` / ``not_tested`` rather than guessed.
"""
from __future__ import annotations from __future__ import annotations
import asyncio
import hashlib import hashlib
import json import json
import re import re
@@ -38,55 +30,19 @@ from a2a_pack import (
from a2a_pack.context import AgentEvent from a2a_pack.context import AgentEvent
from a2a_pack.workspace import FileType from a2a_pack.workspace import FileType
MAX_TEXT_BYTES = 256_000 MAX_TEXT_BYTES = 256_000
MAX_EVIDENCE_BYTES = 8_000 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
ALLOWED_REPO_GLOBS = ( DENIED_REPO_GLOBS = (".git/**", "**/.env", "**/.env.*", "**/*secret*", "**/*credential*", "**/id_rsa", "**/*.pem", "**/*.key")
"agent.py", 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")
"a2a.yaml",
"requirements.txt",
"README.md",
"pyproject.toml",
"Dockerfile",
"*.py",
"*.yaml",
"*.yml",
"*.json",
"*.md",
"tests/*.py",
"tests/**/*.py",
"skills/**/SKILL.md",
)
DENIED_REPO_GLOBS = (
".git/**",
"**/.env",
"**/.env.*",
"**/*secret*",
"**/*credential*",
"**/id_rsa",
"**/*.pem",
"**/*.key",
)
SECRET_PATTERNS = ( SECRET_PATTERNS = (
re.compile(r"(?i)(api[_-]?key|token|secret|password)\s*[:=]\s*['\"]?([A-Za-z0-9_./+=-]{12,})"), 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"sk-[A-Za-z0-9]{20,}"),
re.compile(r"(?i)bearer\s+[A-Za-z0-9_./+=-]{16,}"), re.compile(r"(?i)bearer\s+[A-Za-z0-9_./+=-]{16,}"),
) )
WRITE_INDICATORS = ( WRITE_INDICATORS = ("kubectl apply", "kubectl delete", "kubectl patch", "kubectl scale", "helm upgrade", "git push", "ctx.secret(", "OPENAI_API_KEY", "A2A_LITELLM_KEY")
"kubectl apply",
"kubectl delete",
"kubectl patch",
"kubectl scale",
"helm upgrade",
"git push",
"write_file(",
"ctx.secret(",
"OPENAI_API_KEY",
"A2A_LITELLM_KEY",
)
class StrictModel(BaseModel): class StrictModel(BaseModel):
@@ -113,22 +69,30 @@ class ProbeMode(str, Enum):
SYNTHETIC = "synthetic" SYNTHETIC = "synthetic"
class AuditScope(str, Enum): class AuditTarget(StrictModel):
CARD = "card" agent_name: str = Field(min_length=1, max_length=128, pattern=r"^[a-z0-9][a-z0-9-]{0,127}$")
CONFIGURATION = "configuration" agent_url: AnyHttpUrl | None = None
RUNTIME = "runtime" repository: str | None = Field(default=None, max_length=256)
REPOSITORY = "repository" namespace: str | None = Field(default=None, max_length=128)
RISK = "risk" deployment: str | None = Field(default=None, max_length=128)
REMEDIATION = "remediation"
@field_validator("repository", "namespace", "deployment")
@classmethod
def clean_optional(cls, value: str | None) -> str | None:
if value is None:
return None
return value.strip() or None
class Evidence(StrictModel): class AuditOptions(StrictModel):
id: str audit_id: str | None = Field(default=None, max_length=80, pattern=r"^[A-Za-z0-9_.-]+$")
source: str max_files: int = Field(default=40, ge=0, le=MAX_FILES_PER_AUDIT)
path: str | None = None probe_mode: ProbeMode = ProbeMode.HEALTH_ONLY
status: Literal["observed", "missing", "unknown", "not_tested", "redacted"] include_repository: bool = True
excerpt: str = Field(default="", max_length=MAX_EVIDENCE_BYTES) include_runtime: bool = True
sha256: str | None = None 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): class Finding(StrictModel):
@@ -144,6 +108,15 @@ class Finding(StrictModel):
verification_procedure: 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): class OutputFile(StrictModel):
path: str path: str
artifact_uri: str | None = None artifact_uri: str | None = None
@@ -151,33 +124,6 @@ class OutputFile(StrictModel):
size_bytes: int = Field(ge=0) size_bytes: int = Field(ge=0)
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:
if value is None:
return None
clean = value.strip()
return clean or 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 InventoryRequest(StrictModel): class InventoryRequest(StrictModel):
target: AuditTarget target: AuditTarget
options: AuditOptions = Field(default_factory=AuditOptions) options: AuditOptions = Field(default_factory=AuditOptions)
@@ -210,13 +156,52 @@ class GenerateRemediationRequest(StrictModel):
risk_summary: dict[str, Any] | None = None risk_summary: dict[str, Any] | None = None
AgentStatus = Literal["ok", "partial", "unreachable", "setup_required"] 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 AgentComplianceAuditorResult(dict[str, Any]): class FindingResult(StrictModel):
"""Compact JSON-shaped result alias for a smaller Agent Card output schema.""" 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): class AgentComplianceAuditorConfig(StrictModel):
@@ -224,101 +209,16 @@ class AgentComplianceAuditorConfig(StrictModel):
default_allowed_namespaces: list[str] = Field(default_factory=list, max_length=50) default_allowed_namespaces: list[str] = Field(default_factory=list, max_length=50)
def _tool_schema(kind: str) -> dict[str, Any]:
request_schema: dict[str, Any] = {
"type": "object",
"additionalProperties": False,
"properties": {
"target": {
"type": "object",
"additionalProperties": False,
"properties": {
"agent_name": {"type": "string", "minLength": 1, "maxLength": 128, "pattern": "^[a-z0-9][a-z0-9-]{0,127}$"},
"agent_url": {"type": ["string", "null"], "format": "uri"},
"repository": {"type": ["string", "null"], "maxLength": 256},
"namespace": {"type": ["string", "null"], "maxLength": 128},
"deployment": {"type": ["string", "null"], "maxLength": 128},
},
"required": ["agent_name"],
},
"options": {
"type": "object",
"additionalProperties": False,
"properties": {
"audit_id": {"type": ["string", "null"], "maxLength": 80, "pattern": "^[A-Za-z0-9_.-]+$"},
"max_files": {"type": "integer", "minimum": 0, "maximum": MAX_FILES_PER_AUDIT, "default": 40},
"probe_mode": {"type": "string", "enum": ["none", "health_only", "synthetic"], "default": "health_only"},
"include_repository": {"type": "boolean", "default": True},
"include_runtime": {"type": "boolean", "default": True},
"include_remediation": {"type": "boolean", "default": True},
"allowed_hosts": {"type": "array", "items": {"type": "string"}, "maxItems": 20, "default": []},
"allowed_namespaces": {"type": "array", "items": {"type": "string"}, "maxItems": 20, "default": []},
},
},
},
"required": ["target"],
}
if kind in {"configuration", "runtime", "risk"}:
request_schema["properties"]["inventory"] = {"type": ["object", "null"], "additionalProperties": True}
finding_schema = {
"type": "object",
"additionalProperties": False,
"properties": {
"id": {"type": "string"},
"control_id": {"type": "string"},
"title": {"type": "string"},
"severity": {"type": "string", "enum": ["critical", "high", "medium", "low", "info"]},
"confidence": {"type": "string", "enum": ["high", "medium", "low"]},
"status": {"type": "string", "enum": ["open", "passed", "unknown", "not_tested"], "default": "open"},
"evidence_refs": {"type": "array", "items": {"type": "string"}, "maxItems": 20, "default": []},
"impact": {"type": "string"},
"remediation": {"type": "string"},
"verification_procedure": {"type": "string"},
},
"required": ["id", "control_id", "title", "severity", "confidence", "impact", "remediation", "verification_procedure"],
}
if kind == "risk":
request_schema["properties"]["configuration_findings"] = {"type": "array", "items": finding_schema, "maxItems": MAX_FINDINGS, "default": []}
request_schema["properties"]["runtime_findings"] = {"type": "array", "items": finding_schema, "maxItems": MAX_FINDINGS, "default": []}
if kind == "remediation":
request_schema["properties"]["findings"] = {"type": "array", "items": finding_schema, "maxItems": MAX_FINDINGS, "default": []}
request_schema["properties"]["risk_summary"] = {"type": ["object", "null"], "additionalProperties": True}
return {"type": "object", "additionalProperties": False, "properties": {"request": request_schema}, "required": ["request"]}
class AgentComplianceAuditor(A2AAgent[AgentComplianceAuditorConfig, NoAuth]): class AgentComplianceAuditor(A2AAgent[AgentComplianceAuditorConfig, NoAuth]):
name = "agent-compliance-auditor" name = "agent-compliance-auditor"
description = ( description = "Read-only compliance auditor for deployed A2A agents and repositories."
"Read-only compliance auditor for deployed A2A agents and repositories. "
"Produces bounded evidence, stable findings, risk summaries, and remediation artifacts."
)
version = "0.1.0" version = "0.1.0"
config_model = AgentComplianceAuditorConfig config_model = AgentComplianceAuditorConfig
auth_model = NoAuth auth_model = NoAuth
pricing = Pricing( pricing = Pricing(price_per_call_usd=0.0, caller_pays_llm=False, notes="Deterministic read-only auditing; no LLM calls and no production mutations.")
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) resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=900)
workspace_access = WorkspaceAccess.dynamic( 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)
max_files=MAX_FILES_PER_AUDIT + 10, 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)
allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
require_reason=False,
deny_patterns=DENIED_REPO_GLOBS,
max_total_size_bytes=12 * 1024 * 1024,
)
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,
)
tools_used = ("httpx", "pydantic", "workspace") tools_used = ("httpx", "pydantic", "workspace")
consumer_setup = ConsumerSetup.from_fields( consumer_setup = ConsumerSetup.from_fields(
ConsumerSetupField.config("GITEA_BASE_URL", label="Gitea base URL", required=False, input_type="url"), ConsumerSetupField.config("GITEA_BASE_URL", label="Gitea base URL", required=False, input_type="url"),
@@ -331,13 +231,13 @@ class AgentComplianceAuditor(A2AAgent[AgentComplianceAuditorConfig, NoAuth]):
ConsumerSetupField.secret("REGISTRY_TOKEN", label="Registry read-only token", required=False), 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", input_schema=_tool_schema("inventory")) @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) -> Any: 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) inventory, evidence, warnings, status = await _collect_inventory(self, ctx, request.target, request.options, audit_id)
output_dir = _output_dir(audit_id) output_dir = _output_dir(audit_id)
files = await _write_json_bundle(ctx, output_dir, {"inventory.json": inventory}, 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) 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) return InventoryResult(audit_id=audit_id, target=request.target, status=status, output_dir=output_dir, generated_files=files, inventory=inventory, evidence=evidence, warnings=warnings)
@@ -347,45 +247,45 @@ class AgentComplianceAuditor(A2AAgent[AgentComplianceAuditorConfig, NoAuth]):
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] 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, evidence, warnings = _configuration_findings(request.target, inventory)
findings = _dedupe_findings(findings) findings = _dedupe(findings)
output_dir = _output_dir(audit_id) output_dir = _output_dir(audit_id)
files = await _write_json_bundle(ctx, output_dir, {"findings.json": {"findings": [_dump(f) for f in findings]}}, warnings) 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)) await _event(ctx, "audit_completed", audit_id=audit_id, skill="audit_configuration", findings=len(findings))
return {"audit_id": audit_id, "target": _dump(request.target), "status": "ok", "output_dir": output_dir, "generated_files": [_dump(f) for f in files], "findings": [_dump(f) for f in findings], "evidence": [_dump(e) for e in evidence], "warnings": warnings, "mutations_performed": False} 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] inventory = request.inventory or (await _collect_inventory(self, ctx, request.target, request.options, audit_id))[0]
findings, evidence, warnings = await _runtime_findings(self, ctx, request.target, request.options, inventory) findings, evidence, warnings = _runtime_findings(request.target, inventory)
findings = _dedupe_findings(findings) findings = _dedupe(findings)
output_dir = _output_dir(audit_id) output_dir = _output_dir(audit_id)
files = await _write_json_bundle(ctx, output_dir, {"findings.json": {"findings": [_dump(f) for f in findings]}}, warnings) 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)) await _event(ctx, "audit_completed", audit_id=audit_id, skill="audit_runtime", findings=len(findings))
return {"audit_id": audit_id, "target": _dump(request.target), "status": "ok", "output_dir": output_dir, "generated_files": [_dump(f) for f in files], "findings": [_dump(f) for f in findings], "evidence": [_dump(e) for e in evidence], "warnings": warnings, "mutations_performed": False} 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)
all_findings = _dedupe_findings([*request.configuration_findings, *request.runtime_findings]) all_findings = _dedupe([*request.configuration_findings, *request.runtime_findings])
summary = _risk_summary(all_findings) summary = _risk_summary(all_findings)
output_dir = _output_dir(audit_id) output_dir = _output_dir(audit_id)
files = await _write_json_bundle(ctx, output_dir, {"risk-summary.json": summary, "findings.json": {"findings": [_dump(f) for f in all_findings]}}, []) 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"]) await _event(ctx, "audit_completed", audit_id=audit_id, skill="assess_risk", findings=len(all_findings), score=summary["risk_score"])
return {"audit_id": audit_id, "target": _dump(request.target), "status": "ok", "output_dir": output_dir, "generated_files": [_dump(f) for f in files], "risk_summary": summary, "findings": [_dump(f) for f in all_findings], "warnings": [], "mutations_performed": False} 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", input_schema=_tool_schema("remediation")) @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) -> Any: 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_findings(request.findings) findings = _dedupe(request.findings)
risk_summary = request.risk_summary or _risk_summary(findings) risk_summary = request.risk_summary or _risk_summary(findings)
markdown = _remediation_markdown(request.target, audit_id, findings, risk_summary) markdown = _remediation_md(request.target, audit_id, findings, risk_summary)
sarif = _sarif(request.target, findings) sarif = _sarif(request.target, findings)
output_dir = _output_dir(audit_id) output_dir = _output_dir(audit_id)
files = await _write_json_bundle(ctx, output_dir, {"remediation.md": markdown, "sarif.json": sarif, "risk-summary.json": risk_summary}, []) 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)) 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) return RemediationResult(audit_id=audit_id, target=request.target, status="ok", output_dir=output_dir, generated_files=files, remediation_markdown=markdown, sarif=sarif)
@@ -393,222 +293,164 @@ class AgentComplianceAuditor(A2AAgent[AgentComplianceAuditorConfig, NoAuth]):
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 _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: list[str] = [] warnings: list[str] = []
evidence: list[Evidence] = [] evidence: list[Evidence] = []
status: Literal["ok", "partial", "unreachable", "setup_required"] = "ok" allowed_hosts = {h.lower().strip() for h in [*agent.config.default_allowed_hosts, *options.allowed_hosts] if h.strip()}
allowed_hosts = _allowed_hosts(agent, options) allowed_namespaces = {n.strip() for n in [*agent.config.default_allowed_namespaces, *options.allowed_namespaces] if n.strip()}
allowed_namespaces = _allowed_namespaces(agent, options) authorized = not allowed_namespaces or bool(target.namespace and target.namespace in allowed_namespaces)
await _event(ctx, "audit_progress", audit_id=audit_id, phase="resolve_identity") status: Literal["ok", "partial", "unreachable", "setup_required"] = "ok" if authorized else "partial"
identity = { if not authorized:
"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": _namespace_allowed(target.namespace, allowed_namespaces),
}
if not identity["authorized"]:
status = "partial"
warnings.append("Target namespace is outside the configured allowlist; runtime probes skipped.") warnings.append("Target namespace is outside the configured allowlist; runtime probes skipped.")
await _event(ctx, "audit_progress", audit_id=audit_id, phase="agent_card") await _event(ctx, "audit_progress", audit_id=audit_id, phase="agent_card")
card = await _fetch_agent_card(ctx, str(target.agent_url) if target.agent_url else None, allowed_hosts, warnings) card = await _fetch_card(str(target.agent_url) if target.agent_url else None, allowed_hosts, warnings)
if card.get("status") == "unreachable": if card.get("status") == "unreachable" and status == "ok":
status = "unreachable" if status == "ok" else status status = "unreachable"
evidence.append(_evidence("card", "agent_card", "observed" if card.get("card") else "unknown", card)) 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") await _event(ctx, "audit_progress", audit_id=audit_id, phase="repository_static_inspection")
repo_files = await _read_workspace_repo(ctx, options.max_files if options.include_repository else 0, warnings) repo_files = await _read_repo(ctx, options.max_files if options.include_repository else 0, warnings)
repo_summary = _summarize_repo(repo_files)
evidence.extend(_repo_evidence(repo_files)) evidence.extend(_repo_evidence(repo_files))
manifests = _manifests(repo_files)
manifests = _extract_manifests(repo_files)
skills = _extract_skills(card.get("card"), repo_files)
secrets = _detect_secret_references(repo_files)
exposure = _extract_exposure(card.get("card"), manifests)
resources = _extract_resources(card.get("card"), manifests)
provenance = _extract_provenance(manifests, repo_files)
rbac = _extract_rbac(manifests)
health = {"status": "not_tested", "reason": "probe_mode is none"} health = {"status": "not_tested", "reason": "probe_mode is none"}
if options.probe_mode is not ProbeMode.NONE and identity["authorized"]: if options.probe_mode is not ProbeMode.NONE and authorized:
await _event(ctx, "audit_progress", audit_id=audit_id, phase="bounded_runtime_probe") await _event(ctx, "audit_progress", audit_id=audit_id, phase="bounded_runtime_probe")
health = await _health_probe(str(target.agent_url) if target.agent_url else None, allowed_hosts, warnings) health = await _health(str(target.agent_url) if target.agent_url else None, allowed_hosts, warnings)
inventory = { inventory = {
"schema_version": "2026-07-13", "schema_version": "2026-07-13",
"audit_id": audit_id, "audit_id": audit_id,
"generated_at": datetime.now(UTC).isoformat(), "generated_at": datetime.now(UTC).isoformat(),
"identity": identity, "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, "agent_card": card,
"skills": skills, "skills": _skills(card.get("card"), repo_files),
"schemas": _extract_schemas(card.get("card")), "schemas": _schemas(card.get("card")),
"manifests": manifests, "manifests": manifests,
"rbac": rbac, "rbac": _rbac(manifests),
"secrets_references": secrets, "secrets_references": _secret_refs(repo_files),
"network_exposure": exposure, "network_exposure": _exposure(manifests),
"image_provenance": provenance, "image_provenance": _provenance(manifests, repo_files),
"resource_limits": resources, "resource_limits": _resources(card.get("card"), manifests),
"deployment_health": health, "deployment_health": health,
"repository": repo_summary, "repository": _repo_summary(repo_files),
"controls": _controls_catalog(), "controls": _controls(),
"unknowns": _unknowns(card, manifests, health, repo_files), "unknowns": _unknowns(card, manifests, health, repo_files),
"mutations_performed": False, "mutations_performed": False,
} }
return inventory, evidence, warnings, status return inventory, evidence, warnings, status
async def _fetch_agent_card(ctx: RunContext[NoAuth], agent_url: str | None, allowed_hosts: set[str], warnings: list[str]) -> dict[str, Any]: async def _fetch_card(agent_url: str | None, allowed_hosts: set[str], warnings: list[str]) -> dict[str, Any]:
if not agent_url: if not agent_url:
return {"status": "unknown", "reason": "agent_url not provided"} return {"status": "unknown", "reason": "agent_url not provided"}
parsed = urlparse(agent_url) parsed = urlparse(agent_url)
if not _safe_http_url(parsed, allowed_hosts): if not _safe_url(parsed, allowed_hosts):
warnings.append("Agent URL host is not allowlisted or is unsafe; card fetch skipped.") warnings.append("Agent URL host is not allowlisted or is unsafe; card fetch skipped.")
return {"status": "not_tested", "reason": "host_not_allowlisted"} return {"status": "not_tested", "reason": "host_not_allowlisted"}
url = agent_url.rstrip("/") + "/.well-known/agent-card"
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(url, headers={"accept": "application/json"}) resp = await client.get(agent_url.rstrip("/") + "/.well-known/agent-card", headers={"accept": "application/json"})
if resp.status_code >= 400: if resp.status_code >= 400:
return {"status": "unreachable", "http_status": resp.status_code, "body_excerpt": _redact(resp.text[:500])} return {"status": "unreachable", "http_status": resp.status_code, "body_excerpt": _redact(resp.text[:500])}
data = _bounded_json(resp.text, warnings) return {"status": "ok", "card": _bounded_json(resp.text, warnings)}
return {"status": "ok", "url": url, "card": data}
except Exception as exc: # noqa: BLE001 except Exception as exc: # noqa: BLE001
warnings.append(f"Agent card fetch failed: {type(exc).__name__}") warnings.append(f"Agent card fetch failed: {type(exc).__name__}")
return {"status": "unreachable", "error_type": type(exc).__name__} return {"status": "unreachable", "error_type": type(exc).__name__}
async def _health_probe(agent_url: str | None, allowed_hosts: set[str], warnings: list[str]) -> dict[str, Any]: async def _health(agent_url: str | None, allowed_hosts: set[str], warnings: list[str]) -> dict[str, Any]:
if not agent_url: if not agent_url:
return {"status": "unknown", "reason": "agent_url not provided"} return {"status": "unknown", "reason": "agent_url not provided"}
parsed = urlparse(agent_url) if not _safe_url(urlparse(agent_url), allowed_hosts):
if not _safe_http_url(parsed, allowed_hosts):
return {"status": "not_tested", "reason": "host_not_allowlisted"} return {"status": "not_tested", "reason": "host_not_allowlisted"}
url = agent_url.rstrip("/") + "/healthz"
started = time.perf_counter() 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(url) 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: # noqa: BLE001 except Exception as exc: # noqa: BLE001
warnings.append(f"Health probe failed: {type(exc).__name__}") warnings.append(f"Health probe failed: {type(exc).__name__}")
return {"status": "unreachable", "error_type": type(exc).__name__} return {"status": "unreachable", "error_type": type(exc).__name__}
async def _read_workspace_repo(ctx: RunContext[NoAuth], max_files: int, warnings: list[str]) -> dict[str, str]: async def _read_repo(ctx: RunContext[NoAuth], max_files: int, warnings: list[str]) -> dict[str, str]:
if max_files <= 0: if max_files <= 0:
return {} return {}
try: try:
view = await ctx.workspace.open_view( 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.")
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: # noqa: BLE001 except Exception as exc: # noqa: BLE001
warnings.append(f"Repository workspace unavailable: {type(exc).__name__}") warnings.append(f"Repository workspace unavailable: {type(exc).__name__}")
return {} return {}
files: dict[str, str] = {} files: dict[str, str] = {}
for match in view.files[:max_files]: for match in view.files[:max_files]:
path = match.path.replace("\\", "/") path = match.path.replace("\\", "/").lstrip("/")
if not _repo_path_allowed(path): if not _path_allowed(path):
continue continue
try: try:
raw = await view.read(path) files[path] = (await view.read(path))[:MAX_TEXT_BYTES].decode("utf-8", errors="replace")
except Exception as exc: # noqa: BLE001 except Exception as exc: # noqa: BLE001
warnings.append(f"Could not read {path}: {type(exc).__name__}") warnings.append(f"Could not read {path}: {type(exc).__name__}")
continue
files[path] = raw[:MAX_TEXT_BYTES].decode("utf-8", errors="replace")
return files return files
async def _write_json_bundle(ctx: RunContext[NoAuth], output_dir: str, files: dict[str, Any], warnings: list[str]) -> list[OutputFile]: async def _write_outputs(ctx: RunContext[NoAuth], output_dir: str, files: dict[str, Any], warnings: list[str]) -> list[OutputFile]:
out: list[OutputFile] = [] out: list[OutputFile] = []
for name, payload in files.items(): for name, payload in files.items():
rel = f"{output_dir}/{name}" rel = f"{output_dir}/{name}"
if name.endswith(".md") and isinstance(payload, str): data = payload.encode("utf-8") if isinstance(payload, str) else json.dumps(payload, indent=2, sort_keys=True, default=str).encode("utf-8")
data = payload.encode("utf-8") mime = "text/markdown" if name.endswith(".md") else "application/json"
mime = "text/markdown" uri = None
else:
data = json.dumps(payload, indent=2, sort_keys=True, default=str).encode("utf-8")
mime = "application/json"
artifact_uri: str | None = None
try: try:
ref = await ctx.write_artifact(rel, data, mime) ref = await ctx.write_artifact(rel, data, mime)
await ctx.emit_artifact(ref) await ctx.emit_artifact(ref)
artifact_uri = ref.uri uri = ref.uri
except Exception as exc: # noqa: BLE001 except Exception as exc: # noqa: BLE001
warnings.append(f"Artifact write failed for {rel}: {type(exc).__name__}") warnings.append(f"Artifact write failed for {rel}: {type(exc).__name__}")
try: out.append(OutputFile(path=rel, artifact_uri=uri, mime_type=mime, size_bytes=len(data)))
ws = ctx.workspace
if getattr(ws, "is_writable_output", lambda _p: False)(rel) and hasattr(ws, "write_bytes"):
ws.write_bytes(rel, data)
except Exception:
pass
out.append(OutputFile(path=rel, artifact_uri=artifact_uri, mime_type=mime, size_bytes=len(data)))
return out return out
def _configuration_findings(target: AuditTarget, inventory: 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]]:
findings: list[Finding] = [] findings: list[Finding] = []
evidence: list[Evidence] = []
warnings: list[str] = []
card = (inventory.get("agent_card") or {}).get("card") or {} card = (inventory.get("agent_card") or {}).get("card") or {}
repo = inventory.get("repository") or {}
resources = inventory.get("resource_limits") or {}
secrets = inventory.get("secrets_references") or []
exposure = inventory.get("network_exposure") or {}
skills = inventory.get("skills") or [] skills = inventory.get("skills") or []
if not card: if not card:
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 that all public skills and runtime metadata are present.")) 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 len(skills) == 0: if not skills:
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 Pydantic-compatible input and output schemas.", "Reload the live Agent Card and confirm skills[].input_schema and output_schema are non-null.")) 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."))
for skill in skills: for skill in skills:
schema = skill.get("input_schema") or skill.get("inputSchema") or {} schema = skill.get("input_schema") or skill.get("inputSchema") or {}
out_schema = skill.get("output_schema") or skill.get("outputSchema") or {} out_schema = skill.get("output_schema") or skill.get("outputSchema") or {}
if not _schema_strict(schema) or not out_schema: if not (isinstance(schema, dict) and schema.get("type") == "object" and schema.get("additionalProperties") is False) or not out_schema:
findings.append(_finding(target, "A2A-SCHEMA-002", f"Skill {skill.get('name', '<unknown>')} has weak schema", Severity.MEDIUM, Confidence.HIGH, "schemas", "Weak or nullable schemas increase integration errors and prompt-injection attack surface.", "Use strict Pydantic models or explicit JSON Schema with type=object and additionalProperties=false.", "Inspect the Agent Card skill schemas and run negative validation tests for extra fields.")) 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."))
if exposure.get("public") is True and not exposure.get("documented_public_reason"): if (inventory.get("network_exposure") or {}).get("public") is True:
findings.append(_finding(target, "A2A-NET-001", "Agent is publicly exposed without documented reason", Severity.MEDIUM, Confidence.MEDIUM, "network_exposure", "Unexpected public exposure can broaden abuse and data-exfiltration paths.", "Set expose.public=false unless public access is required; document the reason and authentication boundary.", "Review a2a.yaml and ingress policy; confirm exposure matches intended audience.")) 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 resources.get("max_runtime_seconds") in (None, "unknown"): if (inventory.get("resource_limits") or {}).get("max_runtime_seconds") in (None, "unknown"):
findings.append(_finding(target, "A2A-OPS-001", "Runtime limit is unknown", Severity.LOW, Confidence.MEDIUM, "resource_limits", "Unbounded or unknown runtime budgets can cause reliability and cost issues.", "Declare Resources(... max_runtime_seconds=...) in agent.py and mirror runtime.resources in a2a.yaml for larger pods.", "Load Agent Card runtime.resources and deployment manifest to confirm limits.")) 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."))
if secrets: if inventory.get("secrets_references"):
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 by callers or dependencies.", "Remove literal secrets, provider-key environment reads, and broad secret access. Use ctx.consumer_secret or declared runtime secrets only.", "Run secret scanning and verify no returned artifacts contain secret values.")) 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."))
if repo.get("suspicious_instruction_count", 0) > 0: if (inventory.get("repository") or {}).get("suspicious_instruction_count", 0) > 0:
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 concatenate it into privileged prompts without quoting and policy guards.", "Re-run audit and confirm malicious text appears only as redacted evidence, not as executed instructions.")) 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."))
evidence.append(_evidence("configuration", "configuration_rules", "observed", {"finding_count": len(findings)})) return findings[:MAX_FINDINGS], [_evidence("configuration", "configuration_rules", "observed", {"finding_count": len(findings)})], []
return findings[:MAX_FINDINGS], evidence, warnings
async def _runtime_findings(agent: AgentComplianceAuditor, ctx: RunContext[NoAuth], target: AuditTarget, options: AuditOptions, inventory: dict[str, Any]) -> tuple[list[Finding], list[Evidence], list[str]]: def _runtime_findings(target: AuditTarget, inventory: dict[str, Any]) -> tuple[list[Finding], list[Evidence], list[str]]:
findings: list[Finding] = [] findings: list[Finding] = []
evidence: list[Evidence] = []
warnings: list[str] = []
health = inventory.get("deployment_health") or {"status": "unknown"} health = inventory.get("deployment_health") or {"status": "unknown"}
identity = inventory.get("identity") or {} if (inventory.get("identity") or {}).get("authorized") is False:
if identity.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."))
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 the audit from the owning tenant.", "Confirm namespace allowlist contains only tenant-owned namespaces and retry."))
if health.get("status") in {"unreachable", "degraded"}: 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 recent rollout events.", "Repeat health-only probe and verify HTTP 2xx/3xx or documented readiness semantics.")) 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"}: 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; configure read-only Kubernetes/Argo metadata if needed.", "Run audit_runtime with probe_mode=health_only and verify deployment_health.status.")) 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."))
evidence.append(_evidence("runtime", "runtime_rules", "observed", {"finding_count": len(findings), "health": health})) return findings[:MAX_FINDINGS], [_evidence("runtime", "runtime_rules", "observed", {"finding_count": len(findings), "health": health})], []
return findings[:MAX_FINDINGS], evidence, warnings
def _finding(target: AuditTarget, control_id: str, title: str, severity: Severity, confidence: Confidence, evidence_ref: str, impact: str, remediation: str, verification: str) -> Finding: def _finding(target: AuditTarget, control_id: str, title: str, severity: Severity, confidence: Confidence, evidence_ref: str, impact: str, remediation: str, verification: str) -> Finding:
base = f"{target.agent_name}|{control_id}|{title}" suffix = hashlib.sha256(f"{target.agent_name}|{control_id}|{title}".encode()).hexdigest()[:10]
suffix = hashlib.sha256(base.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) 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(findings: list[Finding]) -> list[Finding]: def _dedupe(findings: list[Finding]) -> list[Finding]:
seen: dict[str, Finding] = {}
rank = {Severity.CRITICAL: 5, Severity.HIGH: 4, Severity.MEDIUM: 3, Severity.LOW: 2, Severity.INFO: 1} 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:
prev = seen.get(f.id) if f.id not in seen or rank[f.severity] > rank[seen[f.id].severity]:
if prev is None or rank[f.severity] > rank[prev.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]
@@ -621,128 +463,86 @@ def _risk_summary(findings: list[Finding]) -> dict[str, Any]:
counts[f.severity.value] += 1 counts[f.severity.value] += 1
score += weights[f.severity] score += weights[f.severity]
score = min(score, 100) score = min(score, 100)
if counts["critical"]: rating = "critical" if counts["critical"] else "high" if score >= 70 else "medium" if score >= 30 else "low" if score else "clean"
rating = "critical"
elif score >= 70:
rating = "high"
elif score >= 30:
rating = "medium"
elif score > 0:
rating = "low"
else:
rating = "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} 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_markdown(target: AuditTarget, audit_id: str, findings: list[Finding], risk_summary: dict[str, Any]) -> str: 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: **{risk_summary.get('risk_rating', 'unknown')}** ({risk_summary.get('risk_score', 0)}/100)", "", "This plan is read-only output. No production systems or repositories were changed.", ""] 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: if not findings:
lines.extend(["## No open findings", "", "The synthetic audit found no policy violations in the supplied evidence. Unknown/not-tested areas should still be reviewed."]) 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: for f in findings:
lines.extend([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}", ""]) 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(target: AuditTarget, findings: list[Finding]) -> dict[str, Any]: def _sarif(target: AuditTarget, findings: list[Finding]) -> dict[str, Any]:
rules = [] rules = [{"id": f.control_id, "name": f.title, "shortDescription": {"text": f.title}, "help": {"text": f.remediation}} for f in findings]
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]
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": target.agent_name, "mutations_performed": False}}], "results": results}]}
rules.append({"id": f.control_id, "name": f.title, "shortDescription": {"text": f.title}, "help": {"text": f.remediation}})
results.append({"ruleId": f.control_id, "level": _sarif_level(f.severity), "message": {"text": f"{f.title}: {f.impact}"}, "partialFingerprints": {"findingId": f.id}, "properties": {"severity": f.severity.value, "confidence": f.confidence.value, "verification": f.verification_procedure}})
return {"version": "2.1.0", "$schema": "https://json.schemastore.org/sarif-2.1.0.json", "runs": [{"tool": {"driver": {"name": "agent-compliance-auditor", "informationUri": "https://docs.a2acloud.io/", "rules": rules}}, "invocations": [{"executionSuccessful": True, "properties": {"target": target.agent_name, "mutations_performed": False}}], "results": results}]}
def _sarif_level(severity: Severity) -> str: def _repo_summary(files: dict[str, str]) -> dict[str, Any]:
return {Severity.CRITICAL: "error", Severity.HIGH: "error", Severity.MEDIUM: "warning", Severity.LOW: "note", Severity.INFO: "none"}[severity] 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 {"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 _summarize_repo(files: dict[str, str]) -> dict[str, Any]: def _manifests(files: dict[str, str]) -> dict[str, Any]:
suspicious = 0 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")}
indicators: list[str] = []
for path, text in files.items():
lower = text.lower()
if any(token in lower for token in ("ignore previous instructions", "exfiltrate", "send secrets", "system prompt")):
suspicious += 1
indicators.append(path)
return {"files_inspected": len(files), "paths": sorted(files), "suspicious_instruction_count": suspicious, "suspicious_instruction_paths": indicators[:20], "bounded_bytes_per_file": MAX_TEXT_BYTES}
def _extract_manifests(files: dict[str, str]) -> dict[str, Any]: def _skills(card: Any, files: dict[str, str]) -> list[dict[str, Any]]:
manifests: dict[str, Any] = {} if isinstance(card, dict) and isinstance(card.get("skills"), list):
for path in ("a2a.yaml", "agent.py", "requirements.txt", "Dockerfile"): return [s for s in card["skills"] if isinstance(s, dict)]
if path in files: 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", ""))]
manifests[path] = {"present": True, "sha256": _sha(files[path]), "excerpt": _redact(files[path][:1200])}
else:
manifests[path] = {"present": False}
return manifests
def _extract_skills(card: Any, files: dict[str, str]) -> list[dict[str, Any]]: def _schemas(card: Any) -> dict[str, Any]:
if isinstance(card, dict): out: dict[str, Any] = {}
skills = card.get("skills") or []
if isinstance(skills, list):
return [s for s in skills if isinstance(s, dict)]
text = files.get("agent.py", "")
names = re.findall(r"async\s+def\s+([a-zA-Z_][a-zA-Z0-9_]*)\s*\(", text)
return [{"name": n, "source": "agent.py"} for n in names]
def _extract_schemas(card: Any) -> dict[str, Any]:
schemas: dict[str, Any] = {}
if isinstance(card, dict): if isinstance(card, dict):
for skill in card.get("skills") or []: for skill in card.get("skills") or []:
if isinstance(skill, dict): if isinstance(skill, dict):
name = str(skill.get("name") or skill.get("id") or "unknown") 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")}
schemas[name] = {"input_schema": skill.get("input_schema") or skill.get("inputSchema"), "output_schema": skill.get("output_schema") or skill.get("outputSchema")} return out
return schemas
def _detect_secret_references(files: dict[str, str]) -> list[dict[str, str]]: def _secret_refs(files: dict[str, str]) -> list[dict[str, str]]:
out: list[dict[str, str]] = [] refs: list[dict[str, str]] = []
for path, text in files.items(): for path, text in files.items():
for pattern in SECRET_PATTERNS: for pattern in SECRET_PATTERNS:
for match in pattern.finditer(text): refs += [{"path": path, "match": _redact(m.group(0))} for m in pattern.finditer(text)]
out.append({"path": path, "match": _redact(match.group(0))}) refs += [{"path": path, "match": _redact(ind)} for ind in WRITE_INDICATORS if ind in text]
for indicator in WRITE_INDICATORS: return refs[:50]
if indicator in text:
out.append({"path": path, "match": _redact(indicator)})
return out[:50]
def _extract_exposure(card: Any, manifests: dict[str, Any]) -> dict[str, Any]: def _exposure(manifests: dict[str, Any]) -> dict[str, Any]:
public = None
text = json.dumps(manifests, default=str).lower() text = json.dumps(manifests, default=str).lower()
if "public: true" in text or "public=true" in text: 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
public = True
if "public: false" in text or "public=false" in text:
public = False
return {"public": public, "documented_public_reason": "public_reason" in text or "exposure_reason" in text, "unknown_when_null": public is None} return {"public": public, "documented_public_reason": "public_reason" in text or "exposure_reason" in text, "unknown_when_null": public is None}
def _extract_resources(card: Any, manifests: dict[str, Any]) -> dict[str, Any]: def _resources(card: Any, manifests: dict[str, Any]) -> dict[str, Any]:
runtime = card.get("runtime") if isinstance(card, dict) else {} runtime = card.get("runtime") if isinstance(card, dict) else {}
resources = runtime.get("resources") if isinstance(runtime, dict) else {} resources = runtime.get("resources") if isinstance(runtime, dict) else {}
out = dict(resources or {}) if isinstance(resources, 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:
text = json.dumps(manifests, default=str) match = re.search(r"max_runtime_seconds[\s:=]+([0-9]+)", json.dumps(manifests, default=str))
match = re.search(r"max_runtime_seconds[\s:=]+([0-9]+)", text)
out["max_runtime_seconds"] = int(match.group(1)) if match else "unknown" out["max_runtime_seconds"] = int(match.group(1)) if match else "unknown"
return out return out
def _extract_provenance(manifests: dict[str, Any], files: dict[str, str]) -> dict[str, Any]: def _provenance(manifests: dict[str, Any], files: dict[str, str]) -> dict[str, Any]:
dockerfile = files.get("Dockerfile", "") dockerfile = files.get("Dockerfile", "")
digest_pinned = "@sha256:" in dockerfile 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")}}
return {"dockerfile_present": bool(dockerfile), "digest_pinned_base_image": digest_pinned 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 _extract_rbac(manifests: dict[str, Any]) -> dict[str, Any]: def _rbac(manifests: dict[str, Any]) -> dict[str, Any]:
text = json.dumps(manifests, default=str).lower() 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 {"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]: def _unknowns(card: dict[str, Any], manifests: dict[str, Any], health: dict[str, Any], files: dict[str, str]) -> list[str]:
out: list[str] = [] out = []
if not card.get("card"): if not card.get("card"):
out.append("agent_card") out.append("agent_card")
if not files: if not files:
@@ -754,22 +554,8 @@ def _unknowns(card: dict[str, Any], manifests: dict[str, Any], health: dict[str,
return out return out
def _controls_catalog() -> list[dict[str, str]]: def _controls() -> list[dict[str, str]]:
return [ 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")]]
{"control_id": "A2A-CARD-001", "title": "Agent Card availability and integrity"},
{"control_id": "A2A-SCHEMA-001", "title": "Public skill inventory"},
{"control_id": "A2A-SCHEMA-002", "title": "Strict non-null schemas"},
{"control_id": "A2A-NET-001", "title": "Network exposure minimization"},
{"control_id": "A2A-OPS-001", "title": "Runtime resource limits"},
{"control_id": "A2A-SECRET-001", "title": "Secret handling and redaction"},
{"control_id": "A2A-UNTRUSTED-001", "title": "Untrusted content isolation"},
{"control_id": "A2A-TENANT-001", "title": "Tenant and namespace isolation"},
{"control_id": "A2A-HEALTH-001", "title": "Deployment health"},
]
def _schema_strict(schema: Any) -> bool:
return isinstance(schema, dict) and schema.get("type") == "object" and schema.get("additionalProperties") is False and bool(schema.get("properties"))
def _repo_evidence(files: dict[str, str]) -> list[Evidence]: def _repo_evidence(files: dict[str, str]) -> list[Evidence]:
@@ -778,52 +564,25 @@ def _repo_evidence(files: dict[str, str]) -> list[Evidence]:
def _evidence(eid: str, source: str, status: Literal["observed", "missing", "unknown", "not_tested", "redacted"], payload: Any, path: str | None = None) -> Evidence: 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) text = payload if isinstance(payload, str) else json.dumps(payload, sort_keys=True, default=str)
redacted = _redact(text)[:MAX_EVIDENCE_BYTES] return Evidence(id=eid, source=source, path=path, status=status, excerpt=_redact(text)[:MAX_EVIDENCE_BYTES], sha256=_sha(text))
return Evidence(id=eid, source=source, path=path, status=status, excerpt=redacted, sha256=_sha(text))
def _audit_id(target: AuditTarget, supplied: str | None) -> str: def _audit_id(target: AuditTarget, supplied: str | None) -> str:
if supplied: return supplied or f"audit-{target.agent_name}-{hashlib.sha256(f'{target.agent_name}|{target.repository or ""}|{int(time.time())}'.encode()).hexdigest()[:12]}"
return supplied
digest = hashlib.sha256(f"{target.agent_name}|{target.repository or ''}|{int(time.time())}".encode()).hexdigest()[:12]
return f"audit-{target.agent_name}-{digest}"
def _output_dir(audit_id: str) -> str: def _output_dir(audit_id: str) -> str:
clean = re.sub(r"[^A-Za-z0-9_.-]", "-", audit_id)[:80] return f"outputs/compliance/{re.sub(r'[^A-Za-z0-9_.-]', '-', audit_id)[:80]}"
return f"outputs/compliance/{clean}"
def _repo_path_allowed(path: str) -> bool: def _path_allowed(path: str) -> bool:
clean = path.replace("\\", "/").lstrip("/") clean = path.replace("\\", "/").lstrip("/")
if ".." in clean.split("/"): 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)
return False
if any(fnmatch(clean, pat) for pat in DENIED_REPO_GLOBS):
return False
return any(fnmatch(clean, pat) for pat in ALLOWED_REPO_GLOBS)
def _allowed_hosts(agent: AgentComplianceAuditor, options: AuditOptions) -> set[str]: def _safe_url(parsed: Any, allowed_hosts: set[str]) -> bool:
return {h.lower().strip() for h in [*agent.config.default_allowed_hosts, *options.allowed_hosts] if h.strip()}
def _allowed_namespaces(agent: AgentComplianceAuditor, options: AuditOptions) -> set[str]:
return {n.strip() for n in [*agent.config.default_allowed_namespaces, *options.allowed_namespaces] if n.strip()}
def _namespace_allowed(namespace: str | None, allowed_namespaces: set[str]) -> bool:
if not allowed_namespaces:
return True
return bool(namespace and namespace in allowed_namespaces)
def _safe_http_url(parsed: Any, allowed_hosts: set[str]) -> bool:
if parsed.scheme not in {"https", "http"}:
return False
host = (parsed.hostname or "").lower() host = (parsed.hostname or "").lower()
if not host or host in {"localhost", "127.0.0.1", "0.0.0.0"} or host.startswith("169.254.") or host.endswith(".local"): 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 False
return host in allowed_hosts
def _bounded_json(text: str, warnings: list[str]) -> Any: def _bounded_json(text: str, warnings: list[str]) -> Any:
@@ -848,15 +607,12 @@ def _sha(text: str) -> str:
return hashlib.sha256(text.encode("utf-8", errors="replace")).hexdigest() return hashlib.sha256(text.encode("utf-8", errors="replace")).hexdigest()
def _dump(model_or_value: Any) -> Any: def _dump(value: Any) -> Any:
if isinstance(model_or_value, BaseModel): return value.model_dump(mode="json") if isinstance(value, BaseModel) else value
return model_or_value.model_dump(mode="json")
return model_or_value
async def _event(ctx: RunContext[NoAuth], kind: str, **payload: Any) -> None: async def _event(ctx: RunContext[NoAuth], kind: str, **payload: Any) -> None:
safe_payload = json.loads(json.dumps(payload, default=str)) safe = json.loads(json.dumps(payload, default=str))
await ctx.emit_event(AgentEvent(kind=kind, payload=safe_payload)) await ctx.emit_event(AgentEvent(kind=kind, payload=safe))
if kind in {"audit_started", "audit_progress", "audit_completed"}: if kind in {"audit_started", "audit_progress", "audit_completed"}:
message = safe_payload.get("phase") or safe_payload.get("skill") or kind await ctx.emit_progress(f"{kind}: {safe.get('phase') or safe.get('skill') or kind}")
await ctx.emit_progress(f"{kind}: {message}")