diff --git a/agent.py b/agent.py index 563709d..3127308 100644 --- a/agent.py +++ b/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