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 11:10:12 +00:00
parent e1d81f80a3
commit 9c5903aba9
1 changed files with 5 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
agent.py
View File

@@ -128,7 +128,7 @@ 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)
class AgentComplianceAuditor(A2AAgent[AgentComplianceAuditorConfig, NoAuth]): class AgentComplianceAuditor(A2AAgent[AgentComplianceAuditorConfig, NoAuth]):
name="agent-compliance-auditor"; description="Read-only compliance auditor for deployed A2A agents and repositories."; version="0.1.1" name="agent-compliance-auditor"; description="Read-only compliance auditor for deployed A2A agents and repositories."; version="0.1.2"
config_model=AgentComplianceAuditorConfig; auth_model=NoAuth config_model=AgentComplianceAuditorConfig; auth_model=NoAuth
pricing=Pricing(price_per_call_usd=0.0, caller_pays_llm=False, notes="Deterministic read-only auditing; no LLM calls and no production mutations.") pricing=Pricing(price_per_call_usd=0.0, caller_pays_llm=False, notes="Deterministic read-only auditing; no LLM calls and no production mutations.")
resources=Resources(cpu="500m", memory="512Mi", max_runtime_seconds=900) resources=Resources(cpu="500m", memory="512Mi", max_runtime_seconds=900)
@@ -276,8 +276,8 @@ def _schemas(card:Any)->dict[str,Any]:
def _secret_refs(files:dict[str,str])->list[dict[str,str]]: def _secret_refs(files:dict[str,str])->list[dict[str,str]]:
refs=[] refs=[]
for p,t in files.items(): for p,t in files.items():
for pat in SECRET_PATTERNS: refs += [{"path":p,"match":_redact(m.group(0))} for m in pat.finditer(t)] for pat in SECRET_PATTERNS: refs += [{"path":p,"match":_redact(match.group(0))} for match in pat.finditer(t)]
refs += [{"path":p,"match":_redact(i)} for i in WRITE_INDICATORS if i in t] refs += [{"path":p,"match":_redact(indicator)} for indicator in WRITE_INDICATORS if indicator in t]
return refs[:50] return refs[:50]
def _exposure(manifests:dict[str,Any])->dict[str,Any]: 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 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
@@ -285,7 +285,7 @@ def _exposure(manifests:dict[str,Any])->dict[str,Any]:
def _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 {}; res=runtime.get("resources") if isinstance(runtime,dict) else {}; out=dict(res or {}) if isinstance(res,dict) else {} runtime=card.get("runtime") if isinstance(card,dict) else {}; res=runtime.get("resources") if isinstance(runtime,dict) else {}; out=dict(res or {}) if isinstance(res,dict) else {}
if "max_runtime_seconds" not in out: if "max_runtime_seconds" not in out:
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" match=re.search(r"max_runtime_seconds[\s:=]+([0-9]+)",json.dumps(manifests,default=str)); 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]: 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")}} 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")}}
@@ -316,7 +316,7 @@ def _bounded_json(text:str,warnings:list[str])->Any:
except json.JSONDecodeError: warnings.append("HTTP response was not valid JSON."); return {"invalid_json_excerpt":_redact(text[:1000])} except json.JSONDecodeError: warnings.append("HTTP response was not valid JSON."); return {"invalid_json_excerpt":_redact(text[:1000])}
def _redact(text:str)->str: def _redact(text:str)->str:
out=text out=text
for pat in SECRET_PATTERNS: out=pat.sub(lambda m:m.group(0)[:8]+"…REDACTED",out) for pat in SECRET_PATTERNS: out=pat.sub(lambda match:match.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 _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 _dump(value:Any)->Any: return value.model_dump(mode="json") if isinstance(value,BaseModel) else value