a2a-source-edit: write agent.py
This commit is contained in:
a2a-cloud
478
agent.py
478
agent.py
@@ -1,189 +1,383 @@
|
||||
"""customer-integration-engineer agent.
|
||||
"""Deterministic release validation agent for customer-integration-engineer.
|
||||
|
||||
Starter stack:
|
||||
- DeepAgents for tool-calling orchestration
|
||||
- Caller-provided LLM credentials via ctx.llm
|
||||
- A tiny model-call middleware hook you can replace with tracing,
|
||||
routing, rate limits, or policy checks
|
||||
This agent validates a customer-integration-engineer release without editing the
|
||||
validated source. It is intentionally local-logic only: no LLM credentials,
|
||||
provider keys, code-editor-agent delegation, or cosmetic mutation paths.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
import re
|
||||
from typing import Any
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from pydantic import BaseModel
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
import a2a_pack as a2a
|
||||
from a2a_pack import (
|
||||
A2AAgent,
|
||||
LLMProvisioning,
|
||||
{{ auth_type }},
|
||||
NoAuth,
|
||||
Pricing,
|
||||
Resources,
|
||||
RunContext,
|
||||
WorkspaceAccess,
|
||||
WorkspaceMode,
|
||||
)
|
||||
from a2a_pack.context import LLMCreds
|
||||
|
||||
|
||||
EXPECTED_AGENT_NAME = "customer-integration-engineer"
|
||||
EXPECTED_LIVE_VERSION = "0.1.5"
|
||||
EXPECTED_PUBLIC = False
|
||||
EXPECTED_WORKSPACE_MODES = {"read_only", "read_write_overlay"}
|
||||
EXPECTED_SKILL_COUNT = 5
|
||||
MAX_ALLOWED_IMPROVEMENT_ITERATIONS = 0
|
||||
SECRET_KEY_RE = re.compile(r"(api[_-]?key|token|secret|password|authorization|bearer)", re.I)
|
||||
SECRET_VALUE_RE = re.compile(
|
||||
r"(?i)(sk-[A-Za-z0-9_-]{12,}|xox[baprs]-[A-Za-z0-9-]{12,}|gh[pousr]_[A-Za-z0-9_]{20,}|bearer\s+[A-Za-z0-9._-]{16,})"
|
||||
)
|
||||
|
||||
|
||||
class CustomerIntegrationEngineerConfig(BaseModel):
|
||||
pass
|
||||
"""Static release-validation policy."""
|
||||
|
||||
target_agent_name: str = EXPECTED_AGENT_NAME
|
||||
target_version: str = EXPECTED_LIVE_VERSION
|
||||
required_skill_count: int = EXPECTED_SKILL_COUNT
|
||||
|
||||
|
||||
SYSTEM_PROMPT = """\
|
||||
You are a compact tool-calling agent.
|
||||
class CardValidationInput(BaseModel):
|
||||
"""Bounded card details copied from the live Agent Card or registry view."""
|
||||
|
||||
Use the text_stats tool when the user asks about text, counts, summaries,
|
||||
or anything where exact length/word numbers would help. Mention tool results
|
||||
briefly instead of dumping raw JSON.
|
||||
"""
|
||||
|
||||
RUNTIME_SKILLS_DIR = "customer-integration-engineer/.deepagents/skills/"
|
||||
DEEPAGENTS_RECURSION_LIMIT = 500
|
||||
card: dict[str, Any] = Field(description="Live Agent Card JSON for the target release.")
|
||||
expected_version: str = Field(default=EXPECTED_LIVE_VERSION, max_length=32)
|
||||
expected_skill_count: int = Field(default=EXPECTED_SKILL_COUNT, ge=1, le=20)
|
||||
require_private: bool = True
|
||||
|
||||
|
||||
class CustomerIntegrationEngineer(A2AAgent[CustomerIntegrationEngineerConfig, {{ auth_type }}]):
|
||||
name = "customer-integration-engineer"
|
||||
description = "Validates the customer-integration-engineer release end to end, preserving existing source unless critical defects remain."
|
||||
class SmokeCallInput(BaseModel):
|
||||
"""Structured result from an Agent Studio delegated live smoke invocation."""
|
||||
|
||||
skill_name: str = Field(min_length=1, max_length=80)
|
||||
input_args: dict[str, Any] = Field(default_factory=dict)
|
||||
result: dict[str, Any] = Field(default_factory=dict)
|
||||
status_code: int = Field(default=200, ge=100, le=599)
|
||||
used_workspace_delegation: bool = True
|
||||
delegated_workspace_mode: str = Field(default="read_only", max_length=64)
|
||||
|
||||
|
||||
class ReviewerFinding(BaseModel):
|
||||
severity: str = Field(description="critical, warning, info, or pass", max_length=32)
|
||||
area: str = Field(max_length=120)
|
||||
message: str = Field(max_length=1000)
|
||||
|
||||
|
||||
class CustomerIntegrationEngineer(A2AAgent[CustomerIntegrationEngineerConfig, NoAuth]):
|
||||
name = EXPECTED_AGENT_NAME
|
||||
description = (
|
||||
"Validates the customer-integration-engineer release end to end while "
|
||||
"preserving source unless a critical defect remains."
|
||||
)
|
||||
version = "0.1.0"
|
||||
|
||||
config_model = CustomerIntegrationEngineerConfig
|
||||
auth_model = {{ auth_type }}
|
||||
auth_model = NoAuth
|
||||
|
||||
# Hosted generated agents read the caller's saved LLM credential through
|
||||
# ctx.llm. The platform may proxy that credential through LiteLLM, but agent
|
||||
# code never reads provider keys, LiteLLM master keys, or OPENAI_API_KEY
|
||||
# directly.
|
||||
llm_provisioning = LLMProvisioning.PLATFORM
|
||||
pricing = Pricing(
|
||||
price_per_call_usd=0.0,
|
||||
caller_pays_llm=True,
|
||||
notes="Starter agent uses the caller's saved LLM credential via ctx.llm.",
|
||||
caller_pays_llm=False,
|
||||
notes="Deterministic validation checks only; no LLM credential required.",
|
||||
)
|
||||
resources = Resources(cpu="250m", memory="512Mi", max_runtime_seconds=300)
|
||||
workspace_access = WorkspaceAccess.dynamic(
|
||||
max_files=64,
|
||||
allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
|
||||
require_reason=False,
|
||||
)
|
||||
tools_used = ("deepagents", "langchain")
|
||||
tools_used = ("a2a-pack", "workspace-delegation-review")
|
||||
|
||||
@a2a.tool(description="Ask the starter DeepAgent to answer with tool calls when useful")
|
||||
async def ask(self, ctx: RunContext[{{ auth_type }}], prompt: str) -> str:
|
||||
creds = ctx.llm
|
||||
await ctx.emit_progress(f"llm: {creds.model} via {creds.source}")
|
||||
if not creds.api_key:
|
||||
return (
|
||||
"LLM key required. Add an LLM credential in Settings > LLM "
|
||||
"credentials before running this agent; for local --invoke "
|
||||
"runs set AGENT_LLM_KEY."
|
||||
)
|
||||
graph = self._build_deep_agent(ctx=ctx, creds=creds)
|
||||
state = await graph.ainvoke(
|
||||
{"messages": [{"role": "user", "content": prompt}]},
|
||||
config={"recursion_limit": DEEPAGENTS_RECURSION_LIMIT},
|
||||
)
|
||||
await ctx.emit_progress("deepagent finished")
|
||||
return _last_message_text(state)
|
||||
|
||||
def _build_deep_agent(
|
||||
@a2a.tool(
|
||||
description="Validate the live Agent Card version, privacy, five public skills, and workspace safety controls.",
|
||||
timeout_seconds=60,
|
||||
idempotent=True,
|
||||
cost_class="cheap",
|
||||
)
|
||||
async def validate_live_card(
|
||||
self,
|
||||
*,
|
||||
ctx: RunContext[{{ auth_type }}],
|
||||
creds: LLMCreds,
|
||||
) -> Any:
|
||||
# Lazy imports keep `a2a card` usable before local dependencies are
|
||||
# installed. `a2a deploy` installs requirements.txt during the build.
|
||||
from a2a_pack.deepagents import create_a2a_deep_agent
|
||||
from langchain.agents.middleware import wrap_model_call
|
||||
from langchain_core.tools import tool
|
||||
ctx: RunContext[NoAuth],
|
||||
payload: CardValidationInput,
|
||||
) -> dict[str, Any]:
|
||||
await ctx.emit_progress("validating live card contract")
|
||||
card = payload.card
|
||||
skills = _skills(card)
|
||||
workspace = _workspace_access(card)
|
||||
runtime = _runtime(card)
|
||||
checks = [
|
||||
_check("agent_name", card.get("name") == self.config.target_agent_name, f"expected {self.config.target_agent_name}"),
|
||||
_check("version", str(card.get("version")) == payload.expected_version, f"expected live version {payload.expected_version}"),
|
||||
_check("skill_count", len(skills) == payload.expected_skill_count, f"expected {payload.expected_skill_count} skills, got {len(skills)}"),
|
||||
_check("private_exposure", _is_private(card) if payload.require_private else True, "agent must remain private"),
|
||||
_check("workspace_enabled", bool(workspace.get("enabled")), "WorkspaceAccess.dynamic must be enabled"),
|
||||
_check("workspace_modes", EXPECTED_WORKSPACE_MODES <= set(_mode_values(workspace.get("allowed_modes"))), "requires read_only and read_write_overlay"),
|
||||
_check("no_llm_required", not _runtime_requires_caller_llm(runtime), "validator/smoke path must not require caller LLM"),
|
||||
]
|
||||
return _result("validate_live_card", checks, extra={"skills": [_skill_name(s) for s in skills]})
|
||||
|
||||
@tool
|
||||
def text_stats(text: str) -> str:
|
||||
"""Return exact word, character, and line counts for text."""
|
||||
words = [part for part in text.split() if part.strip()]
|
||||
return json.dumps(
|
||||
{
|
||||
"characters": len(text),
|
||||
"words": len(words),
|
||||
"lines": len(text.splitlines()) or 1,
|
||||
}
|
||||
)
|
||||
|
||||
@wrap_model_call
|
||||
async def log_model_call(request: Any, handler: Any) -> Any:
|
||||
messages = request.state.get("messages", [])
|
||||
print(
|
||||
"[middleware] model_call "
|
||||
f"model={creds.model} source={creds.source} messages={len(messages)}"
|
||||
)
|
||||
return await handler(request)
|
||||
|
||||
backend = ctx.workspace_backend()
|
||||
skill_sources = _seed_runtime_skills(backend, ctx)
|
||||
# create_a2a_deep_agent resolves provider:model strings with
|
||||
# langchain.init_chat_model from ctx.llm, preserving LiteLLM routing,
|
||||
# provider-specific extra body, and runtime model overrides.
|
||||
return create_a2a_deep_agent(
|
||||
ctx,
|
||||
creds=creds,
|
||||
backend=backend,
|
||||
skills=skill_sources or None,
|
||||
tools=[text_stats],
|
||||
middleware=[log_model_call],
|
||||
system_prompt=SYSTEM_PROMPT,
|
||||
@a2a.tool(
|
||||
description="Confirm the primary public skill has bounded typed inputs and no credential-shaped parameters.",
|
||||
timeout_seconds=60,
|
||||
idempotent=True,
|
||||
cost_class="cheap",
|
||||
)
|
||||
async def validate_primary_skill_schema(
|
||||
self,
|
||||
ctx: RunContext[NoAuth],
|
||||
payload: CardValidationInput,
|
||||
primary_skill_name: str = "",
|
||||
) -> dict[str, Any]:
|
||||
await ctx.emit_progress("validating primary skill schema")
|
||||
skills = _skills(payload.card)
|
||||
primary = _select_skill(skills, primary_skill_name)
|
||||
schema = _input_schema(primary) if primary else {}
|
||||
properties = schema.get("properties") if isinstance(schema, dict) else {}
|
||||
required = schema.get("required") if isinstance(schema, dict) else []
|
||||
property_names = sorted(properties) if isinstance(properties, dict) else []
|
||||
checks = [
|
||||
_check("primary_skill_present", primary is not None, "primary skill must exist"),
|
||||
_check("schema_is_object", isinstance(schema, dict) and schema.get("type") == "object", "input_schema must be a JSON object"),
|
||||
_check("bounded_property_count", isinstance(properties, dict) and 1 <= len(properties) <= 12, "primary skill should expose 1-12 typed inputs"),
|
||||
_check("additional_properties_false", schema.get("additionalProperties") is False, "schema should reject unknown inputs"),
|
||||
_check("all_inputs_typed", _all_properties_typed(properties), "each input property must declare a type or structured schema"),
|
||||
_check("no_credential_inputs", not any(SECRET_KEY_RE.search(name) for name in property_names), "public inputs must not request secrets"),
|
||||
_check("required_is_bounded", isinstance(required, list) and len(required) <= len(property_names), "required list must be bounded"),
|
||||
]
|
||||
return _result(
|
||||
"validate_primary_skill_schema",
|
||||
checks,
|
||||
extra={"primary_skill": _skill_name(primary) if primary else None, "input_names": property_names},
|
||||
)
|
||||
|
||||
@a2a.tool(
|
||||
description="Review source-change policy for critical-only edits, no code-editor-agent use, and version bump discipline.",
|
||||
timeout_seconds=60,
|
||||
idempotent=True,
|
||||
cost_class="cheap",
|
||||
)
|
||||
async def review_source_change_policy(
|
||||
self,
|
||||
ctx: RunContext[NoAuth],
|
||||
mutated_source: bool,
|
||||
critical_defect_remaining: bool,
|
||||
used_code_editor_agent: bool,
|
||||
improvement_iterations: int = 0,
|
||||
previous_version: str = EXPECTED_LIVE_VERSION,
|
||||
resulting_version: str = EXPECTED_LIVE_VERSION,
|
||||
changed_files: list[str] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
await ctx.emit_progress("reviewing source mutation policy")
|
||||
files = changed_files or []
|
||||
checks = [
|
||||
_check("no_code_editor_agent", not used_code_editor_agent, "code-editor-agent must not be used"),
|
||||
_check("iteration_limit", 0 <= improvement_iterations <= MAX_ALLOWED_IMPROVEMENT_ITERATIONS, "at most 0 improvement iterations allowed"),
|
||||
_check("critical_only_mutation", (not mutated_source) or critical_defect_remaining, "source may change only when a critical defect remains"),
|
||||
_check("version_bumped_on_mutation", (not mutated_source) or previous_version != resulting_version, "version must change after real code-editor mutation"),
|
||||
_check("no_cosmetic_files", not _looks_cosmetic_only(files), "avoid cosmetic-only source changes"),
|
||||
]
|
||||
return _result("review_source_change_policy", checks, extra={"changed_files": files})
|
||||
|
||||
def _runtime_skills_root(ctx: RunContext[Any]) -> str:
|
||||
workspace = getattr(ctx, "_workspace", None)
|
||||
prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ())
|
||||
if not prefixes:
|
||||
outputs_prefix = getattr(workspace, "outputs_prefix", None)
|
||||
prefixes = (outputs_prefix or "outputs/",)
|
||||
prefix = str(prefixes[0]).strip("/")
|
||||
return f"/{prefix}/{RUNTIME_SKILLS_DIR}" if prefix else f"/{RUNTIME_SKILLS_DIR}"
|
||||
@a2a.tool(
|
||||
description="Validate a live smoke-call result from Agent Studio workspace delegation and redact any secret-like values.",
|
||||
timeout_seconds=60,
|
||||
idempotent=True,
|
||||
cost_class="cheap",
|
||||
)
|
||||
async def validate_smoke_call(
|
||||
self,
|
||||
ctx: RunContext[NoAuth],
|
||||
smoke: SmokeCallInput,
|
||||
) -> dict[str, Any]:
|
||||
await ctx.emit_progress("validating smoke-call result")
|
||||
secret_paths = _secret_paths(smoke.result)
|
||||
checks = [
|
||||
_check("http_success", 200 <= smoke.status_code < 300, f"status_code={smoke.status_code}"),
|
||||
_check("structured_result", isinstance(smoke.result, dict) and bool(smoke.result), "smoke result must be a non-empty JSON object"),
|
||||
_check("workspace_delegated", smoke.used_workspace_delegation, "smoke must use Agent Studio workspace delegation"),
|
||||
_check("delegation_mode_safe", smoke.delegated_workspace_mode in EXPECTED_WORKSPACE_MODES, "delegated mode must be read_only or read_write_overlay"),
|
||||
_check("no_secrets_returned", not secret_paths, "result must not contain secret-shaped keys or values"),
|
||||
]
|
||||
return _result(
|
||||
"validate_smoke_call",
|
||||
checks,
|
||||
extra={"skill_name": smoke.skill_name, "secret_like_paths": secret_paths},
|
||||
)
|
||||
|
||||
@a2a.tool(
|
||||
description="Combine validation outputs and reviewer findings into a final go/no-go release report.",
|
||||
timeout_seconds=60,
|
||||
idempotent=True,
|
||||
cost_class="cheap",
|
||||
)
|
||||
async def final_release_report(
|
||||
self,
|
||||
ctx: RunContext[NoAuth],
|
||||
validation_results: list[dict[str, Any]],
|
||||
reviewer_findings: list[ReviewerFinding] | None = None,
|
||||
warnings_to_accept: list[str] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
await ctx.emit_progress("building final release report")
|
||||
findings = reviewer_findings or []
|
||||
warnings = warnings_to_accept or []
|
||||
criticals = [f.model_dump() for f in findings if f.severity.lower() == "critical"]
|
||||
failed_checks = [
|
||||
{"result_index": i, "check": check}
|
||||
for i, result in enumerate(validation_results)
|
||||
for check in result.get("checks", [])
|
||||
if isinstance(check, dict) and not check.get("ok")
|
||||
]
|
||||
residual_warnings = [f.model_dump() for f in findings if f.severity.lower() == "warning"]
|
||||
all_warnings_accounted = len(residual_warnings) == 0 or bool(warnings)
|
||||
checks = [
|
||||
_check("zero_critical_findings", not criticals, "reviewer must report zero critical findings"),
|
||||
_check("all_validation_checks_pass", not failed_checks, "all prior validation checks must pass"),
|
||||
_check("warnings_patched_or_listed", all_warnings_accounted, "warnings must be patched or listed as residual risks"),
|
||||
]
|
||||
status = "pass" if all(c["ok"] for c in checks) else "fail"
|
||||
return {
|
||||
"ok": status == "pass",
|
||||
"status": status,
|
||||
"target_agent": self.config.target_agent_name,
|
||||
"target_version": self.config.target_version,
|
||||
"checks": checks,
|
||||
"critical_findings": criticals,
|
||||
"failed_checks": failed_checks,
|
||||
"residual_risks": warnings or [f["message"] for f in residual_warnings],
|
||||
"secrets_included": False,
|
||||
}
|
||||
|
||||
|
||||
def _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]:
|
||||
"""Copy packaged DeepAgents skills into the invocation workspace.
|
||||
|
||||
DeepAgents loads skills from its backend, while source-controlled
|
||||
``skills/`` folders live in the image. This bridge lets generated agents
|
||||
ship reusable SKILL.md bundles without giving up durable A2A workspace
|
||||
files.
|
||||
"""
|
||||
root = Path(__file__).parent / "skills"
|
||||
if not root.exists():
|
||||
return []
|
||||
runtime_skills_root = _runtime_skills_root(ctx)
|
||||
uploads: list[tuple[str, bytes]] = []
|
||||
for path in root.rglob("*"):
|
||||
if path.is_file():
|
||||
rel = path.relative_to(root).as_posix()
|
||||
uploads.append((runtime_skills_root + rel, path.read_bytes()))
|
||||
if uploads:
|
||||
backend.upload_files(uploads)
|
||||
return [runtime_skills_root]
|
||||
return []
|
||||
def _runtime(card: dict[str, Any]) -> dict[str, Any]:
|
||||
value = card.get("runtime")
|
||||
return value if isinstance(value, dict) else {}
|
||||
|
||||
|
||||
def _last_message_text(state: dict[str, Any]) -> str:
|
||||
messages = state.get("messages") or []
|
||||
if not messages:
|
||||
return json.dumps(state, default=str)
|
||||
def _workspace_access(card: dict[str, Any]) -> dict[str, Any]:
|
||||
for key in ("workspace_access", "workspaceAccess", "workspace"):
|
||||
value = card.get(key)
|
||||
if isinstance(value, dict):
|
||||
return value
|
||||
runtime = _runtime(card)
|
||||
for key in ("workspace_access", "workspaceAccess", "workspace"):
|
||||
value = runtime.get(key)
|
||||
if isinstance(value, dict):
|
||||
return value
|
||||
return {}
|
||||
|
||||
content = getattr(messages[-1], "content", None)
|
||||
if isinstance(content, str):
|
||||
return content
|
||||
if isinstance(content, list):
|
||||
parts: list[str] = []
|
||||
for item in content:
|
||||
if isinstance(item, dict):
|
||||
text = item.get("text") or item.get("content")
|
||||
if text:
|
||||
parts.append(str(text))
|
||||
elif item:
|
||||
parts.append(str(item))
|
||||
return "\n".join(parts) if parts else json.dumps(content, default=str)
|
||||
return str(content or messages[-1])
|
||||
|
||||
def _skills(card: dict[str, Any]) -> list[dict[str, Any]]:
|
||||
value = card.get("skills") or card.get("tools") or []
|
||||
return [item for item in value if isinstance(item, dict)] if isinstance(value, list) else []
|
||||
|
||||
|
||||
def _skill_name(skill: dict[str, Any] | None) -> str:
|
||||
if not isinstance(skill, dict):
|
||||
return ""
|
||||
return str(skill.get("name") or skill.get("id") or "")
|
||||
|
||||
|
||||
def _select_skill(skills: list[dict[str, Any]], name: str) -> dict[str, Any] | None:
|
||||
if name:
|
||||
return next((skill for skill in skills if _skill_name(skill) == name), None)
|
||||
return skills[0] if skills else None
|
||||
|
||||
|
||||
def _input_schema(skill: dict[str, Any] | None) -> dict[str, Any]:
|
||||
if not isinstance(skill, dict):
|
||||
return {}
|
||||
for key in ("input_schema", "inputSchema", "parameters"):
|
||||
value = skill.get(key)
|
||||
if isinstance(value, dict):
|
||||
return value
|
||||
return {}
|
||||
|
||||
|
||||
def _mode_values(value: Any) -> list[str]:
|
||||
if isinstance(value, list):
|
||||
raw = value
|
||||
elif isinstance(value, tuple):
|
||||
raw = list(value)
|
||||
elif isinstance(value, str):
|
||||
raw = [value]
|
||||
else:
|
||||
raw = []
|
||||
return [str(item.get("value") if isinstance(item, dict) else item).lower() for item in raw]
|
||||
|
||||
|
||||
def _is_private(card: dict[str, Any]) -> bool:
|
||||
expose = card.get("expose")
|
||||
if isinstance(expose, dict) and "public" in expose:
|
||||
return expose.get("public") is False
|
||||
visibility = str(card.get("visibility") or card.get("access") or "").lower()
|
||||
if visibility in {"private", "unlisted"}:
|
||||
return True
|
||||
public = card.get("public")
|
||||
return public is False if isinstance(public, bool) else True
|
||||
|
||||
|
||||
def _runtime_requires_caller_llm(runtime: dict[str, Any]) -> bool:
|
||||
provisioning = str(runtime.get("llm_provisioning") or runtime.get("llmProvisioning") or "").lower()
|
||||
pricing = runtime.get("pricing") if isinstance(runtime.get("pricing"), dict) else {}
|
||||
return provisioning in {"platform", "caller_provided", "platform_or_caller_provided"} and bool(pricing.get("caller_pays_llm"))
|
||||
|
||||
|
||||
def _all_properties_typed(properties: Any) -> bool:
|
||||
if not isinstance(properties, dict) or not properties:
|
||||
return False
|
||||
for schema in properties.values():
|
||||
if not isinstance(schema, dict):
|
||||
return False
|
||||
if not any(key in schema for key in ("type", "anyOf", "oneOf", "allOf", "$ref")):
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def _looks_cosmetic_only(paths: list[str]) -> bool:
|
||||
if not paths:
|
||||
return False
|
||||
non_cosmetic_exts = {".py", ".yaml", ".yml", ".toml", ".json", ".txt"}
|
||||
cosmetic_markers = {"readme", "docs/", ".md"}
|
||||
lowered = [p.lower() for p in paths]
|
||||
return all(any(marker in p for marker in cosmetic_markers) for p in lowered) and not any(
|
||||
any(p.endswith(ext) for ext in non_cosmetic_exts - {".txt"}) for p in lowered
|
||||
)
|
||||
|
||||
|
||||
def _secret_paths(value: Any, prefix: str = "$") -> list[str]:
|
||||
found: list[str] = []
|
||||
if isinstance(value, dict):
|
||||
for key, item in value.items():
|
||||
path = f"{prefix}.{key}"
|
||||
if SECRET_KEY_RE.search(str(key)):
|
||||
found.append(path)
|
||||
found.extend(_secret_paths(item, path))
|
||||
elif isinstance(value, list):
|
||||
for i, item in enumerate(value[:100]):
|
||||
found.extend(_secret_paths(item, f"{prefix}[{i}]"))
|
||||
elif isinstance(value, str) and SECRET_VALUE_RE.search(value):
|
||||
found.append(prefix)
|
||||
return sorted(set(found))
|
||||
|
||||
|
||||
def _check(name: str, ok: bool, detail: str) -> dict[str, Any]:
|
||||
return {"name": name, "ok": bool(ok), "detail": detail}
|
||||
|
||||
|
||||
def _result(name: str, checks: list[dict[str, Any]], *, extra: dict[str, Any] | None = None) -> dict[str, Any]:
|
||||
ok = all(item["ok"] for item in checks)
|
||||
payload = {"ok": ok, "status": "pass" if ok else "fail", "checkset": name, "checks": checks, "secrets_included": False}
|
||||
if extra:
|
||||
payload.update(extra)
|
||||
return json.loads(json.dumps(payload, default=str))
|
||||
|
||||
|
||||
def _is_safe_url(url: str) -> bool:
|
||||
parsed = urlparse(url)
|
||||
return parsed.scheme in {"https", "http"} and bool(parsed.netloc)
|
||||
|
||||
Reference in New Issue
Block a user