code editor: Agent Studio improvement iteration 1 for agent-compliance-auditor. Goa
This commit is contained in:
25
agent.py
25
agent.py
@@ -23,7 +23,7 @@ MAX_EVIDENCE_BYTES = 8_000
|
||||
MAX_FILES_PER_AUDIT = 80
|
||||
MAX_FINDINGS = 200
|
||||
HTTP_TIMEOUT_SECONDS = 8.0
|
||||
DENIED_REPO_GLOBS = (".git/**", "**/.env", "**/.env.*", "**/*secret*", "**/*credential*", "**/*.pem", "**/*.key")
|
||||
DENIED_REPO_GLOBS = (".git/**", ".env", ".env.*", "*.pem", "*.key", "*secret*", "*credential*", "**/.env", "**/.env.*", "**/*secret*", "**/*credential*", "**/*.pem", "**/*.key")
|
||||
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 = (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")
|
||||
@@ -88,15 +88,18 @@ class OutputFile(StrictModel):
|
||||
mime_type: str
|
||||
size_bytes: int = Field(ge=0)
|
||||
|
||||
def _default_target() -> AuditTarget:
|
||||
return AuditTarget(agent_name="agent-compliance-auditor")
|
||||
|
||||
class InventoryRequest(StrictModel):
|
||||
target: AuditTarget
|
||||
target: AuditTarget = Field(default_factory=_default_target)
|
||||
options: AuditOptions = Field(default_factory=AuditOptions)
|
||||
class AuditConfigurationRequest(StrictModel):
|
||||
target: AuditTarget
|
||||
target: AuditTarget = Field(default_factory=_default_target)
|
||||
options: AuditOptions = Field(default_factory=AuditOptions)
|
||||
inventory: dict[str, Any] | None = None
|
||||
class AuditRuntimeRequest(StrictModel):
|
||||
target: AuditTarget
|
||||
target: AuditTarget = Field(default_factory=_default_target)
|
||||
options: AuditOptions = Field(default_factory=AuditOptions)
|
||||
inventory: dict[str, Any] | None = None
|
||||
class AssessRiskRequest(StrictModel):
|
||||
@@ -128,18 +131,18 @@ 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.3"
|
||||
name="agent-compliance-auditor"; description="Read-only compliance auditor for deployed A2A agents and repositories."; version="0.1.4"
|
||||
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)
|
||||
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)
|
||||
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)
|
||||
workspace_access=WorkspaceAccess.dynamic(max_files=MAX_FILES_PER_AUDIT+10, allowed_modes=(WorkspaceMode.READ_ONLY,), require_reason=False, deny_patterns=DENIED_REPO_GLOBS, max_total_size_bytes=12*1024*1024)
|
||||
egress=EgressPolicy(allow_hosts=("gitea.a2a.svc.cluster.local",), deny_internet_by_default=True)
|
||||
tools_used=("httpx","pydantic","workspace")
|
||||
consumer_setup=ConsumerSetup.from_fields(
|
||||
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),
|
||||
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))
|
||||
ConsumerSetupField.config("GITEA_BASE_URL", label="Gitea base URL", required=False, input_type="url"),
|
||||
ConsumerSetupField.config("KUBERNETES_BASE_URL", label="Kubernetes API base URL", required=False, input_type="url"),
|
||||
ConsumerSetupField.config("ARGOCD_BASE_URL", label="Argo CD base URL", required=False, input_type="url"),
|
||||
ConsumerSetupField.config("REGISTRY_BASE_URL", label="Registry metadata base URL", required=False, input_type="url"))
|
||||
|
||||
@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:
|
||||
|
||||
Reference in New Issue
Block a user