diff --git a/.a2a/agent.dsl.json b/.a2a/agent.dsl.json index a9b3f7a..186f22e 100644 --- a/.a2a/agent.dsl.json +++ b/.a2a/agent.dsl.json @@ -132,19 +132,6 @@ "captcha_policy": { "type": "string" }, - "allowed_captcha_domains": { - "anyOf": [ - { - "items": { - "type": "string" - }, - "type": "array" - }, - { - "type": "null" - } - ] - }, "ai_explore": { "type": "boolean" }, @@ -293,7 +280,8 @@ ], "noise_filtering": "Drops analytics, static assets, service workers, and obvious bot-defense plumbing by default.", "schema_inference": "Infers request/response JSON schemas, query params, path params, GraphQL dispatches, and auth-shaped headers.", - "captcha_policy": "SeleniumBase solve_captcha is available only for a2acloud.io and subdomains.", + "target_scope": "Public browser capture is restricted to a2acloud.io and subdomains.", + "captcha_policy": "SeleniumBase solve_captcha is fixed to a2acloud.io and subdomains.", "llm_exploration": "Uses ctx.llm to choose bounded, non-destructive page actions after the page loads." } }, @@ -322,10 +310,14 @@ "concurrency": 1, "egress": { "allow_hosts": [ - "*" + "a2acloud.io", + "www.a2acloud.io", + "blog.a2acloud.io", + "docs.a2acloud.io", + "api.a2acloud.io" ], "allow_internal_services": [], - "deny_internet_by_default": false + "deny_internet_by_default": true }, "tools_used": [ "seleniumbase", diff --git a/a2a.yaml b/a2a.yaml index 764caed..c0633e1 100644 --- a/a2a.yaml +++ b/a2a.yaml @@ -11,8 +11,12 @@ runtime: max_runtime_seconds: 900 egress: allow_hosts: - - "*" - deny_internet_by_default: false + - a2acloud.io + - www.a2acloud.io + - blog.a2acloud.io + - docs.a2acloud.io + - api.a2acloud.io + deny_internet_by_default: true apt_packages: - ca-certificates - chromium diff --git a/agent.py b/agent.py index 891ae33..609ab6a 100644 --- a/agent.py +++ b/agent.py @@ -24,7 +24,7 @@ from pathlib import Path from typing import Any from urllib.parse import parse_qsl, urljoin, urlparse -from a2a_pack import A2AAgent, LLMProvisioning, NoAuth, Pricing, RunContext, skill +from a2a_pack import A2AAgent, EgressPolicy, LLMProvisioning, NoAuth, Pricing, RunContext, skill from pydantic import BaseModel, Field @@ -84,6 +84,14 @@ NOISE_URL_PATTERNS = [ r"/favicon\.ico(\?|$)", ] DEFAULT_CAPTCHA_SOLVE_DOMAINS = ("a2acloud.io", ".a2acloud.io") +ALLOWED_TARGET_DOMAINS = ("a2acloud.io", ".a2acloud.io") +ALLOWED_EGRESS_HOSTS = ( + "a2acloud.io", + "www.a2acloud.io", + "blog.a2acloud.io", + "docs.a2acloud.io", + "api.a2acloud.io", +) CAPTCHA_POLICIES = {"detect_only", "solve_if_allowed", "never"} BROWSER_MODES = {"seleniumbase_stealth", "playwright"} @@ -135,6 +143,7 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]): auth_model = NoAuth llm_provisioning = LLMProvisioning.PLATFORM pricing = Pricing(price_per_call_usd=0.0, caller_pays_llm=False) + egress = EgressPolicy(allow_hosts=ALLOWED_EGRESS_HOSTS, deny_internet_by_default=True) tools_used = ("seleniumbase", "playwright", "browser-to-api", "openapi") capabilities = { "browser_to_api": { @@ -151,7 +160,8 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]): ], "noise_filtering": "Drops analytics, static assets, service workers, and obvious bot-defense plumbing by default.", "schema_inference": "Infers request/response JSON schemas, query params, path params, GraphQL dispatches, and auth-shaped headers.", - "captcha_policy": "SeleniumBase solve_captcha is available only for a2acloud.io and subdomains.", + "target_scope": "Public browser capture is restricted to a2acloud.io and subdomains.", + "captcha_policy": "SeleniumBase solve_captcha is fixed to a2acloud.io and subdomains.", "llm_exploration": "Uses ctx.llm to choose bounded, non-destructive page actions after the page loads.", } } @@ -180,13 +190,18 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]): redact: list[str] | None = None, browser_mode: str = "seleniumbase_stealth", captcha_policy: str = "solve_if_allowed", - allowed_captcha_domains: list[str] | None = None, ai_explore: bool = True, ai_steps: int = 10, ai_goal: str | None = None, ) -> dict[str, Any]: - if not url or not str(url).strip().startswith(("http://", "https://")): + clean_url = str(url).strip() + if not clean_url or not clean_url.startswith(("http://", "https://")): return {"error": "url must start with http:// or https://"} + if not _host_allowed_for_target(clean_url): + return { + "error": "url host is outside this public agent's allowed target scope", + "allowed_domains": list(ALLOWED_TARGET_DOMAINS), + } if browser_mode not in BROWSER_MODES: return {"error": f"browser_mode must be one of: {', '.join(sorted(BROWSER_MODES))}"} if captcha_policy not in CAPTCHA_POLICIES: @@ -196,13 +211,12 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]): await ctx.emit_progress("launching browser and capturing network traffic") samples, capture_log = await asyncio.to_thread( capture_browser_traffic, - url=str(url).strip(), + url=clean_url, wait_seconds=max(1, min(int(wait_seconds), 60)), max_clicks=max(0, min(int(max_clicks), 20)), search_term=search_term, browser_mode=browser_mode, captcha_policy=captcha_policy, - allowed_captcha_domains=allowed_captcha_domains, ai_explore=ai_explore, ai_steps=max(0, min(int(ai_steps), 20)), ai_goal=ai_goal, @@ -223,14 +237,14 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]): await ctx.emit_progress(f"wrote OpenAPI bundle to {output_dir}") return _result_payload( - url=url, + url=clean_url, samples=samples, bundle=bundle, output_dir=output_dir, capture_log=capture_log, browser_mode=browser_mode, captcha_policy=captcha_policy, - allowed_captcha_domains=_effective_captcha_domains(allowed_captcha_domains), + allowed_captcha_domains=_effective_captcha_domains(None), ai_explore=ai_explore, ai_steps=max(0, min(int(ai_steps), 20)), ) @@ -994,14 +1008,18 @@ def _seleniumbase_sleep(sb: Any, seconds: int) -> None: def _effective_captcha_domains(allowed_captcha_domains: list[str] | None) -> list[str]: - domains = [str(item).strip().lower() for item in (allowed_captcha_domains or []) if str(item).strip()] - for domain in DEFAULT_CAPTCHA_SOLVE_DOMAINS: - if domain not in domains: - domains.append(domain) - return domains + return list(DEFAULT_CAPTCHA_SOLVE_DOMAINS) + + +def _host_allowed_for_target(url: str) -> bool: + return _host_allowed_for_domains(url, list(ALLOWED_TARGET_DOMAINS)) def _host_allowed_for_captcha(url: str, allowed_domains: list[str]) -> bool: + return _host_allowed_for_domains(url, allowed_domains) + + +def _host_allowed_for_domains(url: str, allowed_domains: list[str]) -> bool: host = urlparse(url).hostname or "" host = host.lower().rstrip(".") for raw_domain in allowed_domains: