a2a-source-edit: write agent.py

agent.py

@@ -128,7 +128,7 @@ class AgentComplianceAuditorConfig(StrictModel):
     default_allowed_namespaces: list[str] = Field(default_factory=list, max_length=50)
 
 class AgentComplianceAuditor(A2AAgent[AgentComplianceAuditorConfig, NoAuth]):
-    name="agent-compliance-auditor"; description="Read-only compliance auditor for deployed A2A agents and repositories."; version="0.1.2"
+    name="agent-compliance-auditor"; description="Read-only compliance auditor for deployed A2A agents and repositories."; version="0.1.3"
     config_model=AgentComplianceAuditorConfig; auth_model=NoAuth
     pricing=Pricing(price_per_call_usd=0.0, caller_pays_llm=False, notes="Deterministic read-only auditing; no LLM calls and no production mutations.")
     resources=Resources(cpu="500m", memory="512Mi", max_runtime_seconds=900)
@@ -275,9 +275,13 @@ def _schemas(card:Any)->dict[str,Any]:
     return out
 def _secret_refs(files:dict[str,str])->list[dict[str,str]]:
     refs=[]
-    for p,t in files.items():
-        for pat in SECRET_PATTERNS: refs += [{"path":p,"match":_redact(match.group(0))} for match in pat.finditer(t)]
-        refs += [{"path":p,"match":_redact(indicator)} for indicator in WRITE_INDICATORS if indicator in t]
+    for path,text in files.items():
+        for pattern in SECRET_PATTERNS:
+            for found in pattern.finditer(text):
+                refs.append({"path": path, "match": _redact(found.group(0))})
+        for indicator in WRITE_INDICATORS:
+            if indicator in text:
+                refs.append({"path": path, "match": _redact(indicator)})
     return refs[:50]
 def _exposure(manifests:dict[str,Any])->dict[str,Any]:
     text=json.dumps(manifests,default=str).lower(); public=True if "public: true" in text or "public=true" in text else False if "public: false" in text or "public=false" in text else None
@@ -285,7 +289,8 @@ def _exposure(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 {}
     if "max_runtime_seconds" not in out:
-        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"
+        resource_match = re.search(r"max_runtime_seconds[\s:=]+([0-9]+)",json.dumps(manifests,default=str))
+        out["max_runtime_seconds"] = int(resource_match.group(1)) if resource_match else "unknown"
     return out
 def _provenance(manifests:dict[str,Any],files:dict[str,str])->dict[str,Any]:
     docker=files.get("Dockerfile",""); return {"dockerfile_present":bool(docker),"digest_pinned_base_image":("@sha256:" in docker) if docker else "unknown","source_hashes":{k:v.get("sha256") for k,v in manifests.items() if isinstance(v,dict) and v.get("sha256")}}
@@ -314,9 +319,11 @@ def _bounded_json(text:str,warnings:list[str])->Any:
     if len(text.encode())>MAX_TEXT_BYTES: warnings.append("HTTP JSON response exceeded evidence bound and was truncated before parsing."); text=text[:MAX_TEXT_BYTES]
     try: return json.loads(text)
     except json.JSONDecodeError: warnings.append("HTTP response was not valid JSON."); return {"invalid_json_excerpt":_redact(text[:1000])}
+def _redact_match(redact_match: re.Match[str]) -> str:
+    return redact_match.group(0)[:8]+"…REDACTED"
 def _redact(text:str)->str:
     out=text
-    for pat in SECRET_PATTERNS: out=pat.sub(lambda match:match.group(0)[:8]+"…REDACTED",out)
+    for pat in SECRET_PATTERNS: out=pat.sub(_redact_match,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