iflytek/astron-agent · error · OutboundPolicyError

Tool path must not change the endpoint origin

Error message

Tool path must not change the endpoint origin

What it means

OutboundPolicyError raised by ensure_same_origin when the candidate URL (typically a URL built from a tool result or redirect path) has a different origin — scheme, normalized host, or effective port — than the base URL of the tool call. This guards against tool responses rewriting the request target to a different host/scheme (e.g. open-redirect style escalation). Relative paths are allowed; cross-origin changes are not.

Solutions

  1. Inspect the candidate URL: ensure its scheme, host (lowercased/normalized) and effective port exactly equal the base URL's origin.
  2. Strip absolute URLs returned by the tool down to their path and re-attach them to the base origin before building the follow-up request.
  3. If the tool legitimately needs a different endpoint, configure a separate tool/endpoint for it instead of rewriting the base URL.
  4. Fix scheme mismatches (http vs https) and port mismatches (e.g. :443 with http, or :80 with https) in the tool's configured base URL.

Example fix

// before
next_url = response_json["callback"]  # "http://evil.example/api"
ensure_same_origin(base_url, next_url)
// after
from urllib.parse import urlsplit
path = urlsplit(response_json["callback"]).path
next_url = base_url.rstrip("/") + path
Defensive patterns

Strategy: validation

Validate before calling

from urllib.parse import urlsplit

def same_origin(base_url: str, candidate_url: str) -> bool:
    b, c = urlsplit(base_url), urlsplit(candidate_url)
    def origin(p):
        return (p.scheme.lower(), (p.hostname or "").lower().rstrip("."),
                p.port if p.port is not None else (443 if p.scheme.lower() == "https" else 80))
    return origin(b) == origin(c)

# call ensure_same_origin only after confirming same_origin(base_url, candidate)

Type guard

def is_relative_path(candidate: str) -> bool:
    return not urlsplit(candidate).scheme and not urlsplit(candidate).netloc

Try / catch

try:
    ensure_same_origin(base_url, candidate_url)
except OutboundPolicyError as exc:
    logger.warning("tool attempted cross-origin redirect %s -> %s", base_url, candidate_url)
    return None  # or fall back to base_url + path of candidate

Prevention

When it happens

Trigger: Calling ensure_same_origin(base_url, candidate_url) where candidate changes scheme (https→http or vice versa), host (including trivially different spellings), or port; base or candidate URL lacking http/https scheme or hostname also raises earlier 'origin is invalid' from _origin. Called by _build_url and same-origin tests.

Common situations: A tool/API response returns an absolute URL on another host or plain-http URL that the client then tries to follow; building a follow-up request from a Location header that points off-origin; a base configured with an implicit port while the candidate spells it out inconsistently with scheme defaults (normalized, so mismatch means real change); credentials in the candidate URL (rejected separately as user info).

Understand the failure class

Background: "Invalid URL" / "URL cannot be empty": fix the malformed or missing URL behind request-construction failures — this error's family across 50 libraries.

Related errors


AI-assisted analysis of iflytek/astron-agent@5e758547a8 (2026-09-12). Data as JSON: /api/errors/03f6ab7f0ff4b409. Report an issue: GitHub.

Appendix: source

Thrown at core/plugin/link/infra/tool_exector/ssrf_guard.py:137

        if _is_never_connect_address(address):
            raise OutboundPolicyError("Outbound address is unsafe")
        if allow_private_endpoint:
            return
        if not _canonical_address(address).is_global:
            raise OutboundPolicyError("Outbound address is not globally routable")

    def is_domain_blocked(self, hostname: str) -> bool:
        """Match configured domains on label boundaries, including subdomains."""
        for rule in self.blocked_domains:
            if hostname == rule or hostname.endswith("." + rule):
                return True
        return False


def ensure_same_origin(base_url: str, candidate_url: str) -> None:
    """Reject path or authentication data that changes scheme, host, or port."""
    if _origin(base_url) != _origin(candidate_url):
        raise OutboundPolicyError("Tool path must not change the endpoint origin")


def create_socket_factory(
    policy: OutboundPolicy,
    target_url: str,
) -> Callable[[aiohttp.AddrInfoType], socket.socket]:
    """Create an aiohttp socket factory that checks the actual target sockaddr."""
    parsed = policy.validate_url(target_url)
    hostname = _normalize_hostname(parsed.hostname or "")
    literal_host = _parse_ip(hostname) is not None
    allow_private_endpoint = policy.is_private_endpoint_allowed(parsed)

    def socket_factory(addr_info: aiohttp.AddrInfoType) -> socket.socket:
        family, type_, proto, _, sockaddr = addr_info
        try:
            address = ipaddress.ip_address(sockaddr[0])
        except ValueError as exc:
            raise OutboundPolicyError("Resolved outbound address is invalid") from exc

View on GitHub (pinned to 5e758547a8)