1225 lines
55 KiB
Python
1225 lines
55 KiB
Python
"""Production-safe deterministic customer integration engineering agent.
|
|
|
|
This agent guides an integration from requirements through generated workspace
|
|
artifacts and sandbox-style acceptance evidence. It deliberately does not call
|
|
an LLM, read provider credentials, execute arbitrary shell, deploy, merge, or
|
|
mutate customer systems. All customer credentials stay behind caller-managed
|
|
secret references.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import ipaddress
|
|
import json
|
|
import re
|
|
import socket
|
|
from datetime import UTC, datetime
|
|
from typing import Any, Literal
|
|
from urllib.parse import urlparse
|
|
|
|
import a2a_pack as a2a
|
|
from a2a_pack import (
|
|
A2AAgent,
|
|
ConsumerSetup,
|
|
ConsumerSetupField,
|
|
EgressPolicy,
|
|
NoAuth,
|
|
Pricing,
|
|
Resources,
|
|
RunContext,
|
|
WorkspaceAccess,
|
|
WorkspaceMode,
|
|
)
|
|
from a2a_pack.context import AgentEvent
|
|
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
|
|
|
|
|
|
MAX_JSON_BYTES = 256_000
|
|
MAX_TEXT_BYTES = 128_000
|
|
APPROVAL_PHRASE = "I_APPROVE_THE_LISTED_STEPS"
|
|
OUTPUT_ROOT = "outputs/integrations"
|
|
SECRET_WORDS = ("secret", "token", "password", "passwd", "apikey", "api_key", "authorization", "bearer")
|
|
|
|
|
|
def _tool_input_schema(request_model: type[BaseModel]) -> dict[str, Any]:
|
|
return {
|
|
"type": "object",
|
|
"required": ["request"],
|
|
"additionalProperties": False,
|
|
"properties": {"request": request_model.model_json_schema()},
|
|
}
|
|
|
|
|
|
class StrictModel(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
|
|
|
|
|
|
class AuditEntry(StrictModel):
|
|
ts: str
|
|
action: str
|
|
integration_id: str
|
|
details: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class ArtifactRecord(StrictModel):
|
|
path: str
|
|
sha256: str
|
|
size_bytes: int = Field(ge=0)
|
|
mime_type: str
|
|
|
|
|
|
class SkillStatus(StrictModel):
|
|
status: Literal["ok", "needs_setup", "blocked", "failed"]
|
|
integration_id: str
|
|
dry_run: bool
|
|
audit: list[AuditEntry] = Field(default_factory=list)
|
|
warnings: list[str] = Field(default_factory=list)
|
|
residual_risks: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class SystemSpec(StrictModel):
|
|
name: str = Field(min_length=1, max_length=80)
|
|
kind: Literal[
|
|
"rest_api",
|
|
"webhook",
|
|
"database",
|
|
"object_storage",
|
|
"observability",
|
|
"gitea",
|
|
"kubernetes",
|
|
"argo_cd",
|
|
"queue",
|
|
"other",
|
|
]
|
|
base_url: str | None = Field(default=None, max_length=2048)
|
|
environment: str = Field(default="sandbox", max_length=64)
|
|
owner: str | None = Field(default=None, max_length=120)
|
|
notes: str | None = Field(default=None, max_length=2000)
|
|
|
|
@field_validator("base_url")
|
|
@classmethod
|
|
def _safe_url_text(cls, value: str | None) -> str | None:
|
|
if value is None or not value:
|
|
return None
|
|
parsed = urlparse(value)
|
|
if parsed.scheme not in {"https", "http"} or not parsed.netloc:
|
|
raise ValueError("base_url must be an absolute http(s) URL")
|
|
return value
|
|
|
|
|
|
class AuthMethod(StrictModel):
|
|
system_name: str = Field(min_length=1, max_length=80)
|
|
auth_type: Literal["none", "api_key", "bearer_token", "oauth2_client_credentials", "hmac_signature", "mtls", "basic", "oidc"]
|
|
secret_ref: str | None = Field(
|
|
default=None,
|
|
pattern=r"^[A-Z][A-Z0-9_]{2,127}$",
|
|
description="Caller-managed secret reference name only; never a raw secret value.",
|
|
)
|
|
header_name: str | None = Field(default=None, max_length=80)
|
|
token_url: str | None = Field(default=None, max_length=2048)
|
|
scopes: list[str] = Field(default_factory=list, max_length=20)
|
|
|
|
@model_validator(mode="after")
|
|
def _require_secret_ref_for_secret_auth(self) -> "AuthMethod":
|
|
if self.auth_type != "none" and not self.secret_ref:
|
|
raise ValueError("secret_ref is required for non-none auth methods")
|
|
_reject_raw_secret(self.secret_ref or "")
|
|
return self
|
|
|
|
|
|
class FieldMapping(StrictModel):
|
|
source: str = Field(min_length=1, max_length=160)
|
|
target: str = Field(min_length=1, max_length=160)
|
|
transform: str = Field(default="copy", max_length=500)
|
|
required: bool = True
|
|
pii: bool = False
|
|
|
|
|
|
class DataContract(StrictModel):
|
|
name: str = Field(min_length=1, max_length=120)
|
|
direction: Literal["inbound", "outbound", "bidirectional"]
|
|
content_type: str = Field(default="application/json", max_length=120)
|
|
schema_ref: str | None = Field(default=None, max_length=500)
|
|
fields: list[FieldMapping] = Field(default_factory=list, max_length=200)
|
|
example_payload: dict[str, Any] | None = None
|
|
|
|
@field_validator("example_payload")
|
|
@classmethod
|
|
def _bounded_example(cls, value: dict[str, Any] | None) -> dict[str, Any] | None:
|
|
if value is not None:
|
|
_ensure_bounded_json(value, MAX_JSON_BYTES)
|
|
return value
|
|
|
|
|
|
class RateLimitSpec(StrictModel):
|
|
system_name: str = Field(min_length=1, max_length=80)
|
|
max_requests: int = Field(ge=1, le=1_000_000)
|
|
per_seconds: int = Field(ge=1, le=86_400)
|
|
burst: int | None = Field(default=None, ge=1, le=1_000_000)
|
|
retry_after_header: str | None = Field(default="Retry-After", max_length=80)
|
|
|
|
|
|
class EnvironmentSpec(StrictModel):
|
|
name: str = Field(min_length=1, max_length=64)
|
|
purpose: Literal["dev", "test", "sandbox", "staging", "prod"]
|
|
endpoint_refs: list[str] = Field(default_factory=list, max_length=20)
|
|
object_storage_ref: str | None = Field(default=None, max_length=128)
|
|
observability_ref: str | None = Field(default=None, max_length=128)
|
|
mutable: bool = False
|
|
|
|
|
|
class ComplianceConstraint(StrictModel):
|
|
name: str = Field(min_length=1, max_length=120)
|
|
category: Literal["pii", "phi", "pci", "soc2", "gdpr", "hipaa", "data_residency", "retention", "audit", "other"]
|
|
requirement: str = Field(min_length=1, max_length=2000)
|
|
|
|
|
|
class SuccessCriterion(StrictModel):
|
|
name: str = Field(min_length=1, max_length=120)
|
|
measurement: str = Field(min_length=1, max_length=500)
|
|
target: str = Field(min_length=1, max_length=200)
|
|
|
|
|
|
class IntegrationRequirements(StrictModel):
|
|
integration_id: str = Field(pattern=r"^[a-z][a-z0-9-]{2,63}$")
|
|
title: str = Field(min_length=1, max_length=160)
|
|
systems: list[SystemSpec] = Field(min_length=1, max_length=12)
|
|
auth_methods: list[AuthMethod] = Field(default_factory=list, max_length=12)
|
|
data_contracts: list[DataContract] = Field(default_factory=list, max_length=20)
|
|
rate_limits: list[RateLimitSpec] = Field(default_factory=list, max_length=20)
|
|
environments: list[EnvironmentSpec] = Field(default_factory=list, max_length=12)
|
|
compliance_constraints: list[ComplianceConstraint] = Field(default_factory=list, max_length=30)
|
|
success_criteria: list[SuccessCriterion] = Field(min_length=1, max_length=20)
|
|
external_host_allowlist: list[str] = Field(default_factory=list, max_length=50)
|
|
repository_write_allowlist: list[str] = Field(default_factory=list, max_length=20)
|
|
notes: str | None = Field(default=None, max_length=4000)
|
|
dry_run: bool = True
|
|
|
|
@field_validator("external_host_allowlist")
|
|
@classmethod
|
|
def _clean_hosts(cls, value: list[str]) -> list[str]:
|
|
return [_validate_hostname(host) for host in value]
|
|
|
|
@field_validator("repository_write_allowlist")
|
|
@classmethod
|
|
def _clean_repos(cls, value: list[str]) -> list[str]:
|
|
out: list[str] = []
|
|
for repo in value:
|
|
clean = repo.strip()
|
|
if not re.fullmatch(r"[A-Za-z0-9_.-]{1,80}/[A-Za-z0-9_.-]{1,120}", clean):
|
|
raise ValueError("repository allowlist entries must look like owner/repo")
|
|
out.append(clean)
|
|
return out
|
|
|
|
|
|
class DesignIntegrationRequest(StrictModel):
|
|
requirements: IntegrationRequirements
|
|
create_artifacts: bool = True
|
|
dry_run: bool = True
|
|
|
|
|
|
class DesignIntegrationResult(SkillStatus):
|
|
architecture_path: str | None = None
|
|
mapping_path: str | None = None
|
|
plan_digest: str
|
|
artifacts: list[ArtifactRecord] = Field(default_factory=list)
|
|
architecture_summary: str
|
|
field_mapping_count: int = Field(ge=0)
|
|
required_setup: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class GenerateConnectorRequest(StrictModel):
|
|
requirements: IntegrationRequirements
|
|
connector_language: Literal["python", "typescript"] = "python"
|
|
include_tests: bool = True
|
|
dry_run: bool = True
|
|
external_host_allowlist: list[str] = Field(default_factory=list, max_length=50)
|
|
repository_write_allowlist: list[str] = Field(default_factory=list, max_length=20)
|
|
|
|
@field_validator("external_host_allowlist")
|
|
@classmethod
|
|
def _clean_hosts(cls, value: list[str]) -> list[str]:
|
|
return [_validate_hostname(host) for host in value]
|
|
|
|
|
|
class GenerateConnectorResult(SkillStatus):
|
|
connector_path: str | None = None
|
|
test_path: str | None = None
|
|
plan_digest: str
|
|
artifacts: list[ArtifactRecord] = Field(default_factory=list)
|
|
generated_files: list[str] = Field(default_factory=list)
|
|
blocked_actions: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class ContractFixture(StrictModel):
|
|
name: str = Field(min_length=1, max_length=120)
|
|
payload: dict[str, Any]
|
|
should_pass: bool = True
|
|
|
|
@field_validator("payload")
|
|
@classmethod
|
|
def _bounded_payload(cls, value: dict[str, Any]) -> Any:
|
|
_ensure_bounded_json(value, MAX_JSON_BYTES)
|
|
return value
|
|
|
|
|
|
class ContractArtifact(StrictModel):
|
|
name: str = Field(min_length=1, max_length=120)
|
|
kind: Literal["json_schema", "openapi_3", "webhook"]
|
|
schema_doc: dict[str, Any]
|
|
fixtures: list[ContractFixture] = Field(default_factory=list, max_length=50)
|
|
signature_header: str | None = Field(default=None, max_length=120)
|
|
secret_ref: str | None = Field(default=None, pattern=r"^[A-Z][A-Z0-9_]{2,127}$")
|
|
|
|
@field_validator("schema_doc")
|
|
@classmethod
|
|
def _bounded_schema(cls, value: dict[str, Any]) -> Any:
|
|
_ensure_bounded_json(value, MAX_JSON_BYTES)
|
|
return value
|
|
|
|
|
|
class ValidateContractRequest(StrictModel):
|
|
integration_id: str = Field(pattern=r"^[a-z][a-z0-9-]{2,63}$")
|
|
contracts: list[ContractArtifact] = Field(min_length=1, max_length=20)
|
|
fail_on_drift: bool = True
|
|
dry_run: bool = True
|
|
|
|
|
|
class ContractCheck(StrictModel):
|
|
name: str
|
|
kind: str
|
|
status: Literal["pass", "fail", "warn"]
|
|
findings: list[str] = Field(default_factory=list)
|
|
fixture_results: list[dict[str, Any]] = Field(default_factory=list)
|
|
|
|
|
|
class ValidateContractResult(SkillStatus):
|
|
contract_report_path: str | None = None
|
|
artifacts: list[ArtifactRecord] = Field(default_factory=list)
|
|
checks: list[ContractCheck] = Field(default_factory=list)
|
|
drift_detected: bool = False
|
|
|
|
|
|
class EndpointTest(StrictModel):
|
|
name: str = Field(min_length=1, max_length=120)
|
|
method: Literal["GET", "POST", "PUT", "PATCH", "DELETE"] = "POST"
|
|
url: str = Field(max_length=2048)
|
|
expected_status: int = Field(default=200, ge=100, le=599)
|
|
request_json: dict[str, Any] | None = None
|
|
required_secret_refs: list[str] = Field(default_factory=list, max_length=10)
|
|
idempotency_key: str | None = Field(default=None, max_length=160)
|
|
|
|
@field_validator("request_json")
|
|
@classmethod
|
|
def _bounded_request(cls, value: dict[str, Any] | None) -> dict[str, Any] | None:
|
|
if value is not None:
|
|
_ensure_bounded_json(value, MAX_JSON_BYTES)
|
|
return value
|
|
|
|
@field_validator("required_secret_refs")
|
|
@classmethod
|
|
def _secret_refs_only(cls, value: list[str]) -> list[str]:
|
|
for ref in value:
|
|
if not re.fullmatch(r"[A-Z][A-Z0-9_]{2,127}", ref):
|
|
raise ValueError("secret refs must be uppercase reference names")
|
|
_reject_raw_secret(ref)
|
|
return value
|
|
|
|
|
|
class RunAcceptanceRequest(StrictModel):
|
|
integration_id: str = Field(pattern=r"^[a-z][a-z0-9-]{2,63}$")
|
|
endpoint_tests: list[EndpointTest] = Field(default_factory=list, max_length=20)
|
|
contract_fixtures: list[ContractArtifact] = Field(default_factory=list, max_length=20)
|
|
external_host_allowlist: list[str] = Field(default_factory=list, max_length=50)
|
|
allow_network: bool = False
|
|
max_retries: int = Field(default=1, ge=0, le=3)
|
|
timeout_seconds: float = Field(default=8.0, ge=1.0, le=30.0)
|
|
dry_run: bool = True
|
|
|
|
@field_validator("external_host_allowlist")
|
|
@classmethod
|
|
def _clean_hosts(cls, value: list[str]) -> list[str]:
|
|
return [_validate_hostname(host) for host in value]
|
|
|
|
|
|
class AcceptanceStepResult(StrictModel):
|
|
name: str
|
|
status: Literal["pass", "fail", "skipped", "blocked"]
|
|
attempts: int = Field(ge=0)
|
|
detail: str
|
|
idempotency_key: str | None = None
|
|
|
|
|
|
class RunAcceptanceResult(SkillStatus):
|
|
acceptance_report_path: str | None = None
|
|
artifacts: list[ArtifactRecord] = Field(default_factory=list)
|
|
step_results: list[AcceptanceStepResult] = Field(default_factory=list)
|
|
contract_results: list[ContractCheck] = Field(default_factory=list)
|
|
network_calls_made: int = Field(ge=0)
|
|
|
|
|
|
class HandoffSection(StrictModel):
|
|
heading: str = Field(min_length=1, max_length=120)
|
|
body: str = Field(min_length=1, max_length=8000)
|
|
|
|
@field_validator("body")
|
|
@classmethod
|
|
def _redact_body(cls, value: str) -> str:
|
|
return redact_secrets(value)
|
|
|
|
|
|
class PrepareHandoffRequest(StrictModel):
|
|
integration_id: str = Field(pattern=r"^[a-z][a-z0-9-]{2,63}$")
|
|
title: str = Field(min_length=1, max_length=160)
|
|
rollout_steps: list[str] = Field(min_length=1, max_length=30)
|
|
rollback_steps: list[str] = Field(min_length=1, max_length=30)
|
|
monitoring_checks: list[str] = Field(min_length=1, max_length=30)
|
|
operator_notes: list[HandoffSection] = Field(default_factory=list, max_length=20)
|
|
delivery_actions: list[str] = Field(default_factory=list, max_length=20)
|
|
plan_digest: str | None = Field(default=None, max_length=128)
|
|
approval_phrase: str | None = Field(default=None, max_length=64)
|
|
approve_delivery: bool = False
|
|
dry_run: bool = True
|
|
|
|
|
|
class PrepareHandoffResult(SkillStatus):
|
|
handoff_path: str | None = None
|
|
plan_digest: str
|
|
artifacts: list[ArtifactRecord] = Field(default_factory=list)
|
|
approval_required: bool
|
|
approval_valid: bool
|
|
delivery_executed: bool = False
|
|
|
|
|
|
class CustomerIntegrationEngineerConfig(StrictModel):
|
|
default_dry_run: bool = True
|
|
|
|
|
|
_DESIGN_OUTPUT_SCHEMA = DesignIntegrationResult.model_json_schema()
|
|
_GENERATE_OUTPUT_SCHEMA = GenerateConnectorResult.model_json_schema()
|
|
_VALIDATE_OUTPUT_SCHEMA = ValidateContractResult.model_json_schema()
|
|
_ACCEPTANCE_OUTPUT_SCHEMA = RunAcceptanceResult.model_json_schema()
|
|
_HANDOFF_OUTPUT_SCHEMA = PrepareHandoffResult.model_json_schema()
|
|
|
|
|
|
class CustomerIntegrationEngineer(A2AAgent[CustomerIntegrationEngineerConfig, NoAuth]):
|
|
name = "customer-integration-engineer"
|
|
description = (
|
|
"Guides and validates customer integrations from requirements through a tested handoff while "
|
|
"keeping credentials and production changes under customer control."
|
|
)
|
|
version = "0.1.6"
|
|
|
|
config_model = CustomerIntegrationEngineerConfig
|
|
auth_model = NoAuth
|
|
pricing = Pricing(
|
|
price_per_call_usd=0.0,
|
|
caller_pays_llm=False,
|
|
notes="Deterministic no-LLM integration engineering; callers keep credentials and production changes under their control.",
|
|
)
|
|
resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=600)
|
|
egress = EgressPolicy(deny_internet_by_default=True)
|
|
workspace_access = WorkspaceAccess.dynamic(
|
|
max_files=256,
|
|
allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
|
|
require_reason=True,
|
|
deny_patterns=("**/.env", "**/*secret*", "**/*token*", "**/id_rsa", "**/.git/**"),
|
|
max_total_size_bytes=20 * 1024 * 1024,
|
|
)
|
|
tools_used = ("pydantic", "jsonschema", "httpx")
|
|
consumer_setup = ConsumerSetup.from_fields(
|
|
ConsumerSetupField.config("GITEA_REPOSITORY", label="Gitea repository", description="Optional owner/repo target controlled by the caller.", required=False),
|
|
ConsumerSetupField.secret("GITEA_TOKEN", label="Gitea token", description="Optional caller-managed token reference; never returned or stored.", required=False),
|
|
ConsumerSetupField.config("API_BASE_URL", label="API base URL", description="Optional endpoint used only when allowlisted and dry_run=false.", required=False, input_type="url"),
|
|
ConsumerSetupField.secret("API_AUTH_TOKEN", label="API auth token", description="Optional caller-managed API credential reference.", required=False),
|
|
ConsumerSetupField.config("OBJECT_STORAGE_REF", label="Object storage reference", description="Caller-owned object storage config reference.", required=False),
|
|
ConsumerSetupField.config("OBSERVABILITY_REF", label="Observability reference", description="Caller-owned logs/metrics/traces reference.", required=False),
|
|
ConsumerSetupField.config("KUBERNETES_CONTEXT_REF", label="Kubernetes context reference", description="Optional caller-owned Kubernetes context reference.", required=False),
|
|
ConsumerSetupField.config("ARGOCD_APP_REF", label="Argo CD app reference", description="Optional caller-owned Argo CD app reference.", required=False),
|
|
)
|
|
|
|
@a2a.tool(
|
|
description="Capture requirements, produce architecture and field mapping artifacts",
|
|
input_schema=_tool_input_schema(DesignIntegrationRequest),
|
|
timeout_seconds=120,
|
|
idempotent=True,
|
|
max_retries=0,
|
|
cost_class="deterministic",
|
|
grant_mode="read_write_overlay",
|
|
grant_allow_patterns=("outputs/integrations/**",),
|
|
grant_outputs_prefix="outputs/integrations/",
|
|
grant_write_prefixes=("outputs/integrations/",),
|
|
)
|
|
async def design_integration(self, ctx: RunContext[NoAuth], request: DesignIntegrationRequest) -> DesignIntegrationResult:
|
|
req = request.requirements.model_copy(
|
|
update={"dry_run": _effective_dry_run(self, request.dry_run, request.requirements.dry_run)}
|
|
)
|
|
audit = [_audit("design_started", req.integration_id, {"dry_run": req.dry_run})]
|
|
await _emit(ctx, audit[-1])
|
|
warnings = _requirements_warnings(req)
|
|
required_setup = _required_setup(req)
|
|
digest = _plan_digest({"phase": "design", "requirements": req.model_dump(mode="json")})
|
|
architecture = _render_architecture(req, digest, warnings, required_setup)
|
|
mapping = _build_mapping(req)
|
|
artifacts: list[ArtifactRecord] = []
|
|
arch_path = mapping_path = None
|
|
if request.create_artifacts:
|
|
base = _integration_base(req.integration_id)
|
|
arch_path = f"{base}/architecture.md"
|
|
mapping_path = f"{base}/mapping.json"
|
|
artifacts.append(await _write_workspace_text(ctx, arch_path, architecture, "text/markdown"))
|
|
artifacts.append(await _write_workspace_json(ctx, mapping_path, mapping))
|
|
audit.append(_audit("design_completed", req.integration_id, {"artifacts": [a.path for a in artifacts]}))
|
|
await _emit(ctx, audit[-1])
|
|
return DesignIntegrationResult(
|
|
status="ok",
|
|
integration_id=req.integration_id,
|
|
dry_run=req.dry_run,
|
|
audit=audit,
|
|
warnings=warnings,
|
|
residual_risks=_residual_risks(req),
|
|
architecture_path=arch_path,
|
|
mapping_path=mapping_path,
|
|
plan_digest=digest,
|
|
artifacts=artifacts,
|
|
architecture_summary=f"{req.title}: {len(req.systems)} systems, {len(req.data_contracts)} contracts, {len(req.success_criteria)} success criteria.",
|
|
field_mapping_count=sum(len(c.fields) for c in req.data_contracts),
|
|
required_setup=required_setup,
|
|
)
|
|
|
|
@a2a.tool(
|
|
description="Generate a minimal workspace-contained connector scaffold with no deployment or repository mutation",
|
|
input_schema=_tool_input_schema(GenerateConnectorRequest),
|
|
timeout_seconds=120,
|
|
idempotent=True,
|
|
max_retries=0,
|
|
cost_class="deterministic",
|
|
grant_mode="read_write_overlay",
|
|
grant_allow_patterns=("outputs/integrations/**",),
|
|
grant_outputs_prefix="outputs/integrations/",
|
|
grant_write_prefixes=("outputs/integrations/",),
|
|
)
|
|
async def generate_connector(self, ctx: RunContext[NoAuth], request: GenerateConnectorRequest) -> GenerateConnectorResult:
|
|
req = request.requirements.model_copy(
|
|
update={"dry_run": _effective_dry_run(self, request.dry_run, request.requirements.dry_run)}
|
|
)
|
|
audit = [_audit("connector_generation_started", req.integration_id, {"language": request.connector_language, "dry_run": req.dry_run})]
|
|
await _emit(ctx, audit[-1])
|
|
warnings = _requirements_warnings(req)
|
|
blocked_actions: list[str] = []
|
|
if request.repository_write_allowlist or req.repository_write_allowlist:
|
|
blocked_actions.append("repository writes are never performed; generated source is written only under outputs/integrations/{integration_id}/")
|
|
allowlist = sorted(set(req.external_host_allowlist + request.external_host_allowlist))
|
|
digest = _plan_digest({"phase": "generate_connector", "requirements": req.model_dump(mode="json"), "language": request.connector_language})
|
|
base = _integration_base(req.integration_id)
|
|
artifacts: list[ArtifactRecord] = []
|
|
generated_files: list[str] = []
|
|
if request.connector_language == "python":
|
|
connector_path = f"{base}/connector/python/connector.py"
|
|
test_path = f"{base}/connector/python/test_connector.py" if request.include_tests else None
|
|
source = _python_connector(req, allowlist, digest)
|
|
artifacts.append(await _write_workspace_text(ctx, connector_path, source, "text/x-python"))
|
|
generated_files.append(connector_path)
|
|
if test_path:
|
|
test_source = _python_connector_test(req)
|
|
artifacts.append(await _write_workspace_text(ctx, test_path, test_source, "text/x-python"))
|
|
generated_files.append(test_path)
|
|
else:
|
|
connector_path = f"{base}/connector/typescript/connector.ts"
|
|
test_path = f"{base}/connector/typescript/connector.test.ts" if request.include_tests else None
|
|
source = _typescript_connector(req, allowlist, digest)
|
|
artifacts.append(await _write_workspace_text(ctx, connector_path, source, "text/typescript"))
|
|
generated_files.append(connector_path)
|
|
if test_path:
|
|
test_source = _typescript_connector_test(req)
|
|
artifacts.append(await _write_workspace_text(ctx, test_path, test_source, "text/typescript"))
|
|
generated_files.append(test_path)
|
|
audit.append(_audit("connector_generation_completed", req.integration_id, {"files": generated_files, "blocked_actions": blocked_actions}))
|
|
await _emit(ctx, audit[-1])
|
|
status: Literal["ok", "needs_setup", "blocked", "failed"] = "blocked" if blocked_actions and not req.dry_run else "ok"
|
|
return GenerateConnectorResult(
|
|
status=status,
|
|
integration_id=req.integration_id,
|
|
dry_run=req.dry_run,
|
|
audit=audit,
|
|
warnings=warnings,
|
|
residual_risks=_residual_risks(req),
|
|
connector_path=connector_path,
|
|
test_path=test_path,
|
|
plan_digest=digest,
|
|
artifacts=artifacts,
|
|
generated_files=generated_files,
|
|
blocked_actions=blocked_actions,
|
|
)
|
|
|
|
@a2a.tool(
|
|
description="Validate OpenAPI, JSON Schema, and webhook contracts with deterministic fixtures",
|
|
input_schema=_tool_input_schema(ValidateContractRequest),
|
|
timeout_seconds=120,
|
|
idempotent=True,
|
|
max_retries=0,
|
|
cost_class="deterministic",
|
|
grant_mode="read_write_overlay",
|
|
grant_allow_patterns=("outputs/integrations/**",),
|
|
grant_outputs_prefix="outputs/integrations/",
|
|
grant_write_prefixes=("outputs/integrations/",),
|
|
)
|
|
async def validate_contract(self, ctx: RunContext[NoAuth], request: ValidateContractRequest) -> ValidateContractResult:
|
|
dry_run = _effective_dry_run(self, request.dry_run)
|
|
audit = [_audit("contract_validation_started", request.integration_id, {"contracts": len(request.contracts), "dry_run": dry_run})]
|
|
await _emit(ctx, audit[-1])
|
|
checks = [_validate_one_contract(contract) for contract in request.contracts]
|
|
drift = any(check.status == "fail" for check in checks)
|
|
report = {
|
|
"integration_id": request.integration_id,
|
|
"status": "fail" if drift and request.fail_on_drift else "pass",
|
|
"dry_run": dry_run,
|
|
"generated_at": _now(),
|
|
"checks": [c.model_dump(mode="json") for c in checks],
|
|
"secret_handling": "secret_ref values only; raw credentials are rejected and redacted",
|
|
}
|
|
path = f"{_integration_base(request.integration_id)}/contract-report.json"
|
|
artifact = await _write_workspace_json(ctx, path, report)
|
|
audit.append(_audit("contract_validation_completed", request.integration_id, {"drift_detected": drift, "report": path}))
|
|
await _emit(ctx, audit[-1])
|
|
return ValidateContractResult(
|
|
status="failed" if drift and request.fail_on_drift else "ok",
|
|
integration_id=request.integration_id,
|
|
dry_run=dry_run,
|
|
audit=audit,
|
|
warnings=[] if not drift else ["contract drift detected; inspect contract-report.json before handoff"],
|
|
residual_risks=[] if not drift else ["fixtures are deterministic and bounded; live provider behavior still requires caller-controlled staging verification"],
|
|
contract_report_path=path,
|
|
artifacts=[artifact],
|
|
checks=checks,
|
|
drift_detected=drift,
|
|
)
|
|
|
|
@a2a.tool(
|
|
description="Run dry-run or allowlisted acceptance checks using endpoint metadata and secret references",
|
|
input_schema=_tool_input_schema(RunAcceptanceRequest),
|
|
timeout_seconds=180,
|
|
idempotent=True,
|
|
max_retries=0,
|
|
cost_class="deterministic",
|
|
grant_mode="read_write_overlay",
|
|
grant_allow_patterns=("outputs/integrations/**",),
|
|
grant_outputs_prefix="outputs/integrations/",
|
|
grant_write_prefixes=("outputs/integrations/",),
|
|
)
|
|
async def run_acceptance(self, ctx: RunContext[NoAuth], request: RunAcceptanceRequest) -> RunAcceptanceResult:
|
|
dry_run = _effective_dry_run(self, request.dry_run)
|
|
live_network_block = _live_network_unavailable_reason(self)
|
|
audit = [_audit("acceptance_started", request.integration_id, {"endpoint_tests": len(request.endpoint_tests), "dry_run": dry_run, "allow_network": request.allow_network})]
|
|
await _emit(ctx, audit[-1])
|
|
contract_results = [_validate_one_contract(contract) for contract in request.contract_fixtures]
|
|
step_results: list[AcceptanceStepResult] = []
|
|
network_calls = 0
|
|
warnings: list[str] = []
|
|
for test in request.endpoint_tests:
|
|
idem = test.idempotency_key or _plan_digest({"integration_id": request.integration_id, "test": test.model_dump(mode="json")})[:32]
|
|
if dry_run or not request.allow_network:
|
|
validation = _validate_endpoint_metadata(test.url, request.external_host_allowlist)
|
|
if validation:
|
|
step_results.append(AcceptanceStepResult(name=test.name, status="blocked", attempts=0, detail=validation, idempotency_key=idem))
|
|
warnings.append(validation)
|
|
else:
|
|
step_results.append(AcceptanceStepResult(name=test.name, status="skipped", attempts=0, detail="dry_run/default safe mode: endpoint metadata was validated but no DNS lookup or external request was sent", idempotency_key=idem))
|
|
continue
|
|
metadata_validation = _validate_endpoint_metadata(test.url, request.external_host_allowlist)
|
|
if metadata_validation:
|
|
step_results.append(AcceptanceStepResult(name=test.name, status="blocked", attempts=0, detail=metadata_validation, idempotency_key=idem))
|
|
warnings.append(metadata_validation)
|
|
continue
|
|
if live_network_block:
|
|
step_results.append(AcceptanceStepResult(name=test.name, status="blocked", attempts=0, detail=live_network_block, idempotency_key=idem))
|
|
warnings.append(live_network_block)
|
|
continue
|
|
validation = await _validate_endpoint_target(test.url, request.external_host_allowlist)
|
|
if validation:
|
|
step_results.append(AcceptanceStepResult(name=test.name, status="blocked", attempts=0, detail=validation, idempotency_key=idem))
|
|
warnings.append(validation)
|
|
continue
|
|
result = await _call_endpoint_with_retries(test, request.max_retries, request.timeout_seconds, idem)
|
|
network_calls += result.attempts
|
|
step_results.append(result)
|
|
report = {
|
|
"integration_id": request.integration_id,
|
|
"dry_run": dry_run,
|
|
"allow_network": request.allow_network,
|
|
"network_calls_made": network_calls,
|
|
"generated_at": _now(),
|
|
"endpoint_results": [s.model_dump(mode="json") for s in step_results],
|
|
"contract_results": [c.model_dump(mode="json") for c in contract_results],
|
|
"safety": {
|
|
"ssrf_protection": "scheme, host allowlist, DNS/IP public-routability, and redirect revalidation enforced",
|
|
"secrets": "only secret references accepted; no raw values stored",
|
|
"payload_limits_bytes": MAX_JSON_BYTES,
|
|
"timeouts_seconds": request.timeout_seconds,
|
|
},
|
|
}
|
|
path = f"{_integration_base(request.integration_id)}/acceptance-report.json"
|
|
artifact = await _write_workspace_json(ctx, path, report)
|
|
failed = any(s.status in {"fail", "blocked"} for s in step_results) or any(c.status == "fail" for c in contract_results)
|
|
audit.append(_audit("acceptance_completed", request.integration_id, {"report": path, "network_calls_made": network_calls, "failed_or_blocked": failed}))
|
|
await _emit(ctx, audit[-1])
|
|
return RunAcceptanceResult(
|
|
status="blocked" if any(s.status == "blocked" for s in step_results) else "failed" if failed else "ok",
|
|
integration_id=request.integration_id,
|
|
dry_run=dry_run,
|
|
audit=audit,
|
|
warnings=sorted(set(warnings)),
|
|
residual_risks=["No production deployment, merge, or environment mutation is performed by this agent."],
|
|
acceptance_report_path=path,
|
|
artifacts=[artifact],
|
|
step_results=step_results,
|
|
contract_results=contract_results,
|
|
network_calls_made=network_calls,
|
|
)
|
|
|
|
@a2a.tool(
|
|
description="Prepare rollout, rollback, monitoring, and operator handoff documentation with approval-gated delivery planning",
|
|
input_schema=_tool_input_schema(PrepareHandoffRequest),
|
|
timeout_seconds=120,
|
|
idempotent=True,
|
|
max_retries=0,
|
|
cost_class="deterministic",
|
|
grant_mode="read_write_overlay",
|
|
grant_allow_patterns=("outputs/integrations/**",),
|
|
grant_outputs_prefix="outputs/integrations/",
|
|
grant_write_prefixes=("outputs/integrations/",),
|
|
)
|
|
async def prepare_handoff(self, ctx: RunContext[NoAuth], request: PrepareHandoffRequest) -> PrepareHandoffResult:
|
|
dry_run = _effective_dry_run(self, request.dry_run)
|
|
audit = [_audit("handoff_started", request.integration_id, {"dry_run": dry_run, "delivery_actions": len(request.delivery_actions)})]
|
|
await _emit(ctx, audit[-1])
|
|
plan_material = {
|
|
"integration_id": request.integration_id,
|
|
"rollout_steps": request.rollout_steps,
|
|
"rollback_steps": request.rollback_steps,
|
|
"monitoring_checks": request.monitoring_checks,
|
|
"delivery_actions": request.delivery_actions,
|
|
}
|
|
digest = _plan_digest(plan_material)
|
|
approval_required = bool(request.delivery_actions)
|
|
approval_valid = (not approval_required) or (
|
|
request.approve_delivery
|
|
and request.approval_phrase == APPROVAL_PHRASE
|
|
and request.plan_digest == digest
|
|
)
|
|
warnings: list[str] = []
|
|
if approval_required and not approval_valid:
|
|
warnings.append(f"delivery actions require approval_phrase={APPROVAL_PHRASE!r} and plan_digest={digest}")
|
|
if approval_required and approval_valid:
|
|
warnings.append("approval validated for the listed plan, but this agent still does not deploy, merge, or modify customer systems")
|
|
handoff_request = request.model_copy(update={"dry_run": dry_run})
|
|
handoff = _render_handoff(handoff_request, digest, approval_required, approval_valid)
|
|
path = f"{_integration_base(request.integration_id)}/handoff.md"
|
|
artifact = await _write_workspace_text(ctx, path, handoff, "text/markdown")
|
|
audit.append(_audit("handoff_completed", request.integration_id, {"handoff": path, "approval_required": approval_required, "approval_valid": approval_valid}))
|
|
await _emit(ctx, audit[-1])
|
|
return PrepareHandoffResult(
|
|
status="blocked" if approval_required and not approval_valid else "ok",
|
|
integration_id=request.integration_id,
|
|
dry_run=dry_run,
|
|
audit=audit,
|
|
warnings=warnings,
|
|
residual_risks=["Operators must execute rollout or rollback in caller-controlled systems outside this agent."],
|
|
handoff_path=path,
|
|
plan_digest=digest,
|
|
artifacts=[artifact],
|
|
approval_required=approval_required,
|
|
approval_valid=approval_valid,
|
|
delivery_executed=False,
|
|
)
|
|
|
|
|
|
def _now() -> str:
|
|
return datetime.now(UTC).isoformat()
|
|
|
|
|
|
def _effective_dry_run(agent: CustomerIntegrationEngineer, *values: bool) -> bool:
|
|
return bool(agent.config.default_dry_run or any(values))
|
|
|
|
|
|
def _live_network_unavailable_reason(agent: CustomerIntegrationEngineer) -> str | None:
|
|
if getattr(agent.egress, "deny_internet_by_default", False):
|
|
return (
|
|
"live network checks are disabled by this runtime because egress denies internet by default; "
|
|
"acceptance can validate metadata in dry_run mode but cannot contact external endpoints"
|
|
)
|
|
return None
|
|
|
|
|
|
def _audit(action: str, integration_id: str, details: dict[str, Any] | None = None) -> AuditEntry:
|
|
return AuditEntry(ts=_now(), action=action, integration_id=integration_id, details=redact_obj(details or {}))
|
|
|
|
|
|
async def _emit(ctx: RunContext[NoAuth], entry: AuditEntry) -> None:
|
|
await ctx.emit_event(AgentEvent(kind="audit", payload=entry.model_dump(mode="json")))
|
|
await ctx.emit_progress(f"{entry.action}: {entry.integration_id}")
|
|
|
|
|
|
def _integration_base(integration_id: str) -> str:
|
|
if not re.fullmatch(r"[a-z][a-z0-9-]{2,63}", integration_id):
|
|
raise ValueError("invalid integration_id")
|
|
return f"{OUTPUT_ROOT}/{integration_id}"
|
|
|
|
|
|
def _safe_output_path(path: str) -> str:
|
|
clean = path.replace("\\", "/").strip("/")
|
|
if not clean.startswith(f"{OUTPUT_ROOT}/"):
|
|
raise ValueError("writes must stay under outputs/integrations/{integration_id}/")
|
|
if ".." in clean.split("/") or clean.startswith("/"):
|
|
raise ValueError("unsafe output path")
|
|
return clean
|
|
|
|
|
|
async def _write_workspace_text(ctx: RunContext[NoAuth], path: str, text: str, mime_type: str) -> ArtifactRecord:
|
|
clean = _safe_output_path(path)
|
|
data = redact_secrets(text).encode("utf-8")
|
|
if len(data) > MAX_TEXT_BYTES * 4:
|
|
raise ValueError("artifact exceeds bounded text size")
|
|
ws = ctx.workspace
|
|
if hasattr(ws, "is_writable_output") and ws.is_writable_output(clean): # type: ignore[attr-defined]
|
|
if hasattr(ws, "write_bytes"):
|
|
ws.write_bytes(clean, data) # type: ignore[attr-defined]
|
|
else:
|
|
view = await ws.open_view(purpose=f"write {clean}", hints=[clean], max_files=1, mode=WorkspaceMode.READ_WRITE_OVERLAY, reason="Persist integration engineering output under outputs/integrations")
|
|
await view.write(clean, data)
|
|
else:
|
|
raise PermissionError("workspace grant does not allow writes under outputs/integrations")
|
|
artifact_name = clean.replace("/", "__")
|
|
ref = await ctx.write_artifact(artifact_name, data, mime_type)
|
|
await ctx.emit_artifact(ref)
|
|
return ArtifactRecord(path=clean, sha256=hashlib.sha256(data).hexdigest(), size_bytes=len(data), mime_type=mime_type)
|
|
|
|
|
|
async def _write_workspace_json(ctx: RunContext[NoAuth], path: str, payload: dict[str, Any]) -> ArtifactRecord:
|
|
safe_payload = redact_obj(payload)
|
|
_ensure_bounded_json(safe_payload, MAX_JSON_BYTES)
|
|
return await _write_workspace_text(ctx, path, json.dumps(safe_payload, indent=2, sort_keys=True) + "\n", "application/json")
|
|
|
|
|
|
def _ensure_bounded_json(value: Any, max_bytes: int) -> None:
|
|
data = json.dumps(value, default=str, sort_keys=True).encode("utf-8")
|
|
if len(data) > max_bytes:
|
|
raise ValueError(f"JSON payload exceeds {max_bytes} bytes")
|
|
|
|
|
|
def _reject_raw_secret(value: str) -> None:
|
|
if not value:
|
|
return
|
|
if any(marker in value.lower() for marker in ("sk-", "ghp_", "xoxb-", "-----begin", "bearer ")):
|
|
raise ValueError("raw credential-looking values are not allowed; use a secret_ref")
|
|
|
|
|
|
def redact_secrets(text: str) -> str:
|
|
redacted = text
|
|
patterns = [
|
|
r"(?i)(authorization\s*[:=]\s*bearer\s+)[A-Za-z0-9._~+/=-]+",
|
|
r"(?i)(api[_-]?key\s*[:=]\s*)[A-Za-z0-9._~+/=-]{8,}",
|
|
r"(?i)(token\s*[:=]\s*)[A-Za-z0-9._~+/=-]{8,}",
|
|
r"(?i)(password\s*[:=]\s*)[^\s,;]+",
|
|
r"sk-[A-Za-z0-9]{12,}",
|
|
r"ghp_[A-Za-z0-9]{12,}",
|
|
]
|
|
for pattern in patterns:
|
|
redacted = re.sub(pattern, _redact_match, redacted)
|
|
return redacted
|
|
|
|
|
|
def _redact_match(match: re.Match[str]) -> str:
|
|
return (match.group(1) if match.lastindex else "") + "[REDACTED]"
|
|
|
|
|
|
def _jsonschema_error_path(error: Any) -> Any:
|
|
return error.path
|
|
|
|
|
|
def redact_obj(value: Any) -> Any:
|
|
if isinstance(value, dict):
|
|
out: dict[str, Any] = {}
|
|
for key, item in value.items():
|
|
if any(word in str(key).lower() for word in SECRET_WORDS):
|
|
out[key] = "[REDACTED_REF]" if isinstance(item, str) else "[REDACTED]"
|
|
else:
|
|
out[key] = redact_obj(item)
|
|
return out
|
|
if isinstance(value, list):
|
|
return [redact_obj(item) for item in value]
|
|
if isinstance(value, str):
|
|
return redact_secrets(value)
|
|
return value
|
|
|
|
|
|
def _plan_digest(payload: dict[str, Any]) -> str:
|
|
safe = redact_obj(payload)
|
|
encoded = json.dumps(safe, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")
|
|
return hashlib.sha256(encoded).hexdigest()
|
|
|
|
|
|
def _requirements_warnings(req: IntegrationRequirements) -> list[str]:
|
|
warnings: list[str] = []
|
|
if not req.auth_methods:
|
|
warnings.append("no auth methods supplied; confirm this is intentional")
|
|
if not req.data_contracts:
|
|
warnings.append("no data contracts supplied; mapping and validation will be limited")
|
|
if not req.rate_limits:
|
|
warnings.append("no rate limits supplied; connector will use conservative defaults")
|
|
if any(env.purpose == "prod" and env.mutable for env in req.environments):
|
|
warnings.append("production environment marked mutable; this agent will still not mutate it")
|
|
return warnings
|
|
|
|
|
|
def _residual_risks(req: IntegrationRequirements) -> list[str]:
|
|
risks = [
|
|
"Generated connector is minimal and must be reviewed by customer operators before production use.",
|
|
"Provider-specific authentication and observability wiring stay under caller-controlled setup references.",
|
|
]
|
|
if req.dry_run:
|
|
risks.append("dry_run=true: acceptance evidence is deterministic and non-mutating, not proof of live provider behavior.")
|
|
return risks
|
|
|
|
|
|
def _required_setup(req: IntegrationRequirements) -> list[str]:
|
|
setup = {auth.secret_ref for auth in req.auth_methods if auth.secret_ref}
|
|
for env in req.environments:
|
|
for ref in (env.object_storage_ref, env.observability_ref):
|
|
if ref:
|
|
setup.add(ref)
|
|
return sorted(setup)
|
|
|
|
|
|
def _build_mapping(req: IntegrationRequirements) -> Any:
|
|
return {
|
|
"integration_id": req.integration_id,
|
|
"title": req.title,
|
|
"contracts": [
|
|
{
|
|
"name": c.name,
|
|
"direction": c.direction,
|
|
"content_type": c.content_type,
|
|
"schema_ref": c.schema_ref,
|
|
"fields": [f.model_dump(mode="json") for f in c.fields],
|
|
}
|
|
for c in req.data_contracts
|
|
],
|
|
"pii_fields": [f"{c.name}.{field.target}" for c in req.data_contracts for field in c.fields if field.pii],
|
|
}
|
|
|
|
|
|
def _render_architecture(req: IntegrationRequirements, digest: str, warnings: list[str], setup: list[str]) -> str:
|
|
systems = "\n".join(f"- **{s.name}** ({s.kind}, {s.environment}) {s.base_url or '(endpoint configured by caller)'}" for s in req.systems)
|
|
auth = "\n".join(f"- {a.system_name}: {a.auth_type}, secret_ref={a.secret_ref or 'none'}" for a in req.auth_methods) or "- No auth methods supplied."
|
|
contracts = "\n".join(f"- {c.name}: {c.direction}, {len(c.fields)} mapped fields, schema_ref={c.schema_ref or 'inline/fixture'}" for c in req.data_contracts) or "- No contracts supplied."
|
|
rates = "\n".join(f"- {r.system_name}: {r.max_requests}/{r.per_seconds}s burst={r.burst or r.max_requests}" for r in req.rate_limits) or "- Conservative default: bounded retries with idempotency keys."
|
|
compliance = "\n".join(f"- {c.category}: {c.name} — {c.requirement}" for c in req.compliance_constraints) or "- No explicit compliance constraints supplied."
|
|
success = "\n".join(f"- {s.name}: {s.measurement} target={s.target}" for s in req.success_criteria)
|
|
warning_text = "\n".join(f"- {w}" for w in warnings) or "- None"
|
|
setup_text = "\n".join(f"- {s}" for s in setup) or "- None"
|
|
return f"""# Integration Architecture: {req.title}
|
|
|
|
Integration ID: `{req.integration_id}`
|
|
Plan digest: `{digest}`
|
|
Dry run: `{req.dry_run}`
|
|
|
|
## Systems
|
|
{systems}
|
|
|
|
## Authentication and Secret References
|
|
{auth}
|
|
|
|
Raw credentials are neither accepted nor stored. Operators configure referenced secrets outside this agent.
|
|
|
|
## Data Contracts and Field Mapping
|
|
{contracts}
|
|
|
|
See `mapping.json` for exact field-level mapping.
|
|
|
|
## Rate Limits and Retries
|
|
{rates}
|
|
|
|
All mutating calls must carry idempotency keys. Retries are bounded and honor provider retry metadata.
|
|
|
|
## Environments
|
|
{chr(10).join(f"- {e.name}: {e.purpose}, mutable={e.mutable}" for e in req.environments) or '- No environments supplied.'}
|
|
|
|
## Compliance Constraints
|
|
{compliance}
|
|
|
|
## Success Criteria
|
|
{success}
|
|
|
|
## Required Caller Setup
|
|
{setup_text}
|
|
|
|
## Safety Controls
|
|
- Default dry_run=true.
|
|
- External requests require explicit host allowlist, DNS/IP validation, redirect revalidation, payload limits, and timeouts.
|
|
- Repository writes beyond outputs, deployments, merges, and environment mutations are not performed.
|
|
- Delivery actions require `{APPROVAL_PHRASE}` bound to a plan digest and still remain operator-executed outside this agent.
|
|
|
|
## Warnings
|
|
{warning_text}
|
|
"""
|
|
|
|
|
|
def _python_connector(req: IntegrationRequirements, allowlist: list[str], digest: str) -> str:
|
|
return f'''"""Minimal connector scaffold for {req.integration_id}.
|
|
Generated by customer-integration-engineer. Review before use.
|
|
Plan digest: {digest}
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import ipaddress
|
|
import json
|
|
import socket
|
|
from urllib.parse import urlparse
|
|
|
|
import httpx
|
|
|
|
ALLOWED_HOSTS = {allowlist!r}
|
|
MAX_BYTES = {MAX_JSON_BYTES}
|
|
TIMEOUT_SECONDS = 8.0
|
|
|
|
|
|
def validate_url(url: str) -> str:
|
|
parsed = urlparse(url)
|
|
if parsed.scheme != "https" or not parsed.hostname:
|
|
raise ValueError("connector only permits absolute https URLs")
|
|
host = parsed.hostname.lower()
|
|
if host not in ALLOWED_HOSTS:
|
|
raise ValueError(f"host {{host!r}} is not allowlisted")
|
|
for info in socket.getaddrinfo(host, parsed.port or 443, type=socket.SOCK_STREAM):
|
|
ip = ipaddress.ip_address(info[4][0])
|
|
if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved or ip.is_unspecified:
|
|
raise ValueError(f"host {{host!r}} resolved to unsafe address")
|
|
return url
|
|
|
|
|
|
def idempotency_key(operation: str, payload: dict) -> str:
|
|
material = json.dumps({{"operation": operation, "payload": payload}}, sort_keys=True).encode("utf-8")
|
|
return hashlib.sha256(material).hexdigest()
|
|
|
|
|
|
async def send_json(url: str, payload: dict, *, secret_ref: str | None = None, dry_run: bool = True) -> dict:
|
|
body = json.dumps(payload).encode("utf-8")
|
|
if len(body) > MAX_BYTES:
|
|
raise ValueError("payload too large")
|
|
safe_url = validate_url(url)
|
|
key = idempotency_key("send_json", payload)
|
|
if dry_run:
|
|
return {{"status": "skipped", "reason": "dry_run", "idempotency_key": key, "url": safe_url}}
|
|
# Resolve secret_ref outside this scaffold using your approved secret manager.
|
|
headers = {{"Idempotency-Key": key, "Content-Type": "application/json"}}
|
|
async with httpx.AsyncClient(timeout=TIMEOUT_SECONDS, follow_redirects=False) as client:
|
|
response = await client.post(safe_url, content=body, headers=headers)
|
|
return {{"status_code": response.status_code, "idempotency_key": key}}
|
|
'''
|
|
|
|
|
|
def _python_connector_test(req: IntegrationRequirements) -> str:
|
|
return f'''from connector import idempotency_key
|
|
|
|
|
|
def test_idempotency_key_stable():
|
|
payload = {{"integration_id": "{req.integration_id}", "event": "synthetic"}}
|
|
assert idempotency_key("send_json", payload) == idempotency_key("send_json", payload)
|
|
'''
|
|
|
|
|
|
def _typescript_connector(req: IntegrationRequirements, allowlist: list[str], digest: str) -> str:
|
|
hosts = json.dumps(allowlist)
|
|
return f'''// Minimal connector scaffold for {req.integration_id}. Plan digest: {digest}
|
|
const ALLOWED_HOSTS = new Set<string>({hosts});
|
|
const MAX_BYTES = {MAX_JSON_BYTES};
|
|
|
|
export function validateUrl(raw: string): URL {{
|
|
const url = new URL(raw);
|
|
if (url.protocol !== "https:") throw new Error("connector only permits https URLs");
|
|
if (!ALLOWED_HOSTS.has(url.hostname.toLowerCase())) throw new Error("host is not allowlisted");
|
|
return url;
|
|
}}
|
|
|
|
export async function sendJson(rawUrl: string, payload: unknown, options: {{ dryRun?: boolean, secretRef?: string }} = {{}}) {{
|
|
const url = validateUrl(rawUrl);
|
|
const body = JSON.stringify(payload);
|
|
if (new TextEncoder().encode(body).length > MAX_BYTES) throw new Error("payload too large");
|
|
const idem = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(body));
|
|
const idempotencyKey = Array.from(new Uint8Array(idem)).map(b => b.toString(16).padStart(2, "0")).join("");
|
|
if (options.dryRun !== false) return {{ status: "skipped", reason: "dry_run", idempotencyKey, url: url.toString() }};
|
|
const response = await fetch(url, {{ method: "POST", body, headers: {{ "Content-Type": "application/json", "Idempotency-Key": idempotencyKey }} }});
|
|
return {{ statusCode: response.status, idempotencyKey }};
|
|
}}
|
|
'''
|
|
|
|
|
|
def _typescript_connector_test(req: IntegrationRequirements) -> str:
|
|
return f'''import {{ validateUrl }} from "./connector";
|
|
|
|
test("allowlist rejects unknown host for {req.integration_id}", () => {{
|
|
expect(() => validateUrl("https://example.invalid/webhook")).toThrow();
|
|
}});
|
|
'''
|
|
|
|
|
|
def _validate_one_contract(contract: ContractArtifact) -> ContractCheck:
|
|
findings: list[str] = []
|
|
fixture_results: list[dict[str, Any]] = []
|
|
try:
|
|
if contract.kind == "json_schema":
|
|
from jsonschema import Draft202012Validator
|
|
Draft202012Validator.check_schema(contract.schema_doc)
|
|
validator = Draft202012Validator(contract.schema_doc)
|
|
for fixture in contract.fixtures:
|
|
errors = sorted(validator.iter_errors(fixture.payload), key=_jsonschema_error_path)
|
|
passed = not errors
|
|
expected = fixture.should_pass
|
|
fixture_results.append({"fixture": fixture.name, "passed": passed, "expected": expected, "errors": [e.message for e in errors[:5]]})
|
|
if passed != expected:
|
|
findings.append(f"fixture {fixture.name!r} expected should_pass={expected} but got {passed}")
|
|
elif contract.kind == "openapi_3":
|
|
version = str(contract.schema_doc.get("openapi", ""))
|
|
if not version.startswith("3."):
|
|
findings.append("OpenAPI document must declare openapi: 3.x")
|
|
if not isinstance(contract.schema_doc.get("paths"), dict) or not contract.schema_doc.get("paths"):
|
|
findings.append("OpenAPI document must include non-empty paths")
|
|
if "servers" in contract.schema_doc and not isinstance(contract.schema_doc["servers"], list):
|
|
findings.append("OpenAPI servers must be a list")
|
|
elif contract.kind == "webhook":
|
|
if not contract.signature_header:
|
|
findings.append("webhook contract requires signature_header")
|
|
if not contract.secret_ref:
|
|
findings.append("webhook contract requires secret_ref, not a raw secret")
|
|
event_type = contract.schema_doc.get("event_type") or contract.schema_doc.get("type")
|
|
if not event_type:
|
|
findings.append("webhook schema_doc should include event_type or type")
|
|
for fixture in contract.fixtures:
|
|
has_event = "event" in fixture.payload or "type" in fixture.payload
|
|
passed = bool(has_event)
|
|
fixture_results.append({"fixture": fixture.name, "passed": passed, "expected": fixture.should_pass, "errors": [] if passed else ["missing event/type"]})
|
|
if passed != fixture.should_pass:
|
|
findings.append(f"fixture {fixture.name!r} webhook expectation mismatch")
|
|
except Exception as exc: # noqa: BLE001
|
|
findings.append(f"contract validation error: {type(exc).__name__}: {exc}")
|
|
status: Literal["pass", "fail", "warn"] = "fail" if findings else "pass"
|
|
return ContractCheck(name=contract.name, kind=contract.kind, status=status, findings=findings, fixture_results=fixture_results)
|
|
|
|
|
|
def _validate_hostname(host: str) -> str:
|
|
clean = host.strip().lower().rstrip(".")
|
|
if not clean or len(clean) > 253:
|
|
raise ValueError("invalid hostname")
|
|
if clean in {"localhost", "metadata.google.internal"} or clean.endswith(".local") or clean.endswith(".internal"):
|
|
raise ValueError("internal hostnames are not allowed")
|
|
try:
|
|
ip = ipaddress.ip_address(clean)
|
|
except ValueError:
|
|
if not re.fullmatch(r"[a-z0-9.-]+", clean) or ".." in clean:
|
|
raise ValueError("hostname contains invalid characters")
|
|
return clean
|
|
if _unsafe_ip(ip):
|
|
raise ValueError("private, loopback, link-local, multicast, reserved, or unspecified IPs are not allowed")
|
|
return clean
|
|
|
|
|
|
def _unsafe_ip(ip: ipaddress._BaseAddress) -> bool:
|
|
return bool(ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved or ip.is_unspecified)
|
|
|
|
|
|
def _validate_endpoint_metadata(url: str, allowlist: list[str]) -> str | None:
|
|
parsed = urlparse(url)
|
|
if parsed.scheme != "https" or not parsed.hostname:
|
|
return "endpoint URL must be absolute https with a hostname"
|
|
host = parsed.hostname.lower().rstrip(".")
|
|
if host not in set(allowlist):
|
|
return f"host {host!r} is not in external_host_allowlist"
|
|
try:
|
|
_validate_hostname(host)
|
|
except ValueError as exc:
|
|
return str(exc)
|
|
return None
|
|
|
|
|
|
async def _validate_endpoint_target(url: str, allowlist: list[str]) -> str | None:
|
|
metadata_error = _validate_endpoint_metadata(url, allowlist)
|
|
if metadata_error:
|
|
return metadata_error
|
|
parsed = urlparse(url)
|
|
host = parsed.hostname.lower().rstrip(".") # type: ignore[union-attr]
|
|
try:
|
|
infos = await asyncio.to_thread(socket.getaddrinfo, host, parsed.port or 443, type=socket.SOCK_STREAM)
|
|
except socket.gaierror as exc:
|
|
return f"DNS resolution failed for {host!r}: {exc}"
|
|
for info in infos:
|
|
ip = ipaddress.ip_address(info[4][0])
|
|
if _unsafe_ip(ip):
|
|
return f"host {host!r} resolves to unsafe IP {ip}"
|
|
return None
|
|
|
|
|
|
async def _call_endpoint_with_retries(test: EndpointTest, max_retries: int, timeout_seconds: float, idem: str) -> AcceptanceStepResult:
|
|
import httpx
|
|
|
|
attempts = 0
|
|
last_detail = "not attempted"
|
|
headers = {"Idempotency-Key": idem, "Content-Type": "application/json"}
|
|
for attempt in range(max_retries + 1):
|
|
attempts = attempt + 1
|
|
try:
|
|
async with httpx.AsyncClient(timeout=timeout_seconds, follow_redirects=False) as client:
|
|
response = await client.request(test.method, test.url, json=test.request_json, headers=headers)
|
|
if 300 <= response.status_code < 400:
|
|
location = response.headers.get("location", "")
|
|
return AcceptanceStepResult(name=test.name, status="blocked", attempts=attempts, detail=f"redirect blocked pending revalidation: {location[:200]}", idempotency_key=idem)
|
|
if response.status_code == test.expected_status:
|
|
return AcceptanceStepResult(name=test.name, status="pass", attempts=attempts, detail=f"received expected status {response.status_code}", idempotency_key=idem)
|
|
last_detail = f"expected {test.expected_status}, got {response.status_code}"
|
|
if response.status_code not in {408, 409, 425, 429, 500, 502, 503, 504}:
|
|
break
|
|
await asyncio.sleep(min(2.0, 0.25 * (attempt + 1)))
|
|
except Exception as exc: # noqa: BLE001
|
|
last_detail = f"{type(exc).__name__}: {exc}"
|
|
await asyncio.sleep(min(2.0, 0.25 * (attempt + 1)))
|
|
return AcceptanceStepResult(name=test.name, status="fail", attempts=attempts, detail=redact_secrets(last_detail), idempotency_key=idem)
|
|
|
|
|
|
def _render_handoff(request: PrepareHandoffRequest, digest: str, approval_required: bool, approval_valid: bool) -> str:
|
|
operator_sections = "\n\n".join(f"## {s.heading}\n{s.body}" for s in request.operator_notes) or "## Operator Notes\nNo additional operator notes supplied."
|
|
delivery = "\n".join(f"- {item}" for item in request.delivery_actions) or "- No delivery actions requested."
|
|
return f"""# Integration Handoff: {request.title}
|
|
|
|
Integration ID: `{request.integration_id}`
|
|
Plan digest: `{digest}`
|
|
Dry run: `{request.dry_run}`
|
|
|
|
## Rollout Plan
|
|
{chr(10).join(f"{i+1}. {redact_secrets(step)}" for i, step in enumerate(request.rollout_steps))}
|
|
|
|
## Rollback Plan
|
|
{chr(10).join(f"{i+1}. {redact_secrets(step)}" for i, step in enumerate(request.rollback_steps))}
|
|
|
|
## Monitoring and Alerting Checks
|
|
{chr(10).join(f"- {redact_secrets(check)}" for check in request.monitoring_checks)}
|
|
|
|
{operator_sections}
|
|
|
|
## Delivery Actions Requiring Operator Control
|
|
{delivery}
|
|
|
|
Approval required: `{approval_required}`
|
|
Approval valid for this exact digest: `{approval_valid}`
|
|
|
|
This agent does not deploy, merge, modify customer systems, or execute delivery actions. Operators must perform approved delivery steps in their own controlled systems.
|
|
|
|
## Evidence Checklist
|
|
- Review `architecture.md` and `mapping.json`.
|
|
- Review connector source under `connector/`.
|
|
- Review `contract-report.json`.
|
|
- Review `acceptance-report.json`.
|
|
- Confirm all secret references are configured in caller-owned systems.
|
|
"""
|