OpenAI Agent Access to Australia's Medicare

from typing import Mapping
from urllib.parse import urlsplit

import requests

STOP_STATUSES = {401, 403, 407, 429, 451}
MAX_RESPONSE_BYTES = 1_000_000


class AccessRefused(Exception):
    """Ends the task and alerts a human; never gives the model a retry hint."""


def _ascii_host(host: str) -> str:
    return host.encode("idna").decode("ascii").lower().rstrip(".")


def _prepare_allowed_request(raw_url: str, allowed_hosts: frozenset[str]):
    # Reject ambiguous authority syntax before a URL parser can normalize it differently.
    if "\\" in raw_url or any(ord(char) < 0x21 or ord(char) == 0x7f for char in raw_url):
        raise AccessRefused("ambiguous URL syntax")
    try:
        prepared = requests.Request("GET", raw_url).prepare()
        parts = urlsplit(prepared.url)
        port = parts.port
        host = parts.hostname or ""
        allowed = {_ascii_host(item) for item in allowed_hosts}
        host = _ascii_host(host)
    except (requests.exceptions.RequestException, UnicodeError, ValueError) as exc:
        raise AccessRefused("invalid URL or host policy") from exc

    if (parts.scheme.lower() != "https" or parts.username is not None
            or parts.password is not None or port not in (None, 443)
            or host not in allowed):
        raise AccessRefused("destination is not approved for this task")
    return prepared


def agent_fetch(target_id: str, *, allowed_targets: Mapping[str, str], allowed_hosts: frozenset[str], egress_proxy: str) -> str:
    raw_url = allowed_targets.get(target_id)
    if raw_url is None:
        raise AccessRefused("destination is not approved for this task")
    # Load this set from server-owned, per-task policy, never from the request or model.
    prepared = _prepare_allowed_request(raw_url, allowed_hosts)
    # The model supplies an ID; only the server-owned target map may provide a URL.
    with requests.Session() as session:
        session.trust_env = False  # Ignore ambient proxy settings; use only the controlled proxy.
        response = session.send(
            prepared,
            timeout=10,
            allow_redirects=False,
            stream=True,
            proxies={"https": egress_proxy},
        )
        with response:
            if response.is_redirect:
                raise AccessRefused("redirects require a separately approved destination")
            if response.status_code in STOP_STATUSES:
                raise AccessRefused(f"approved host refused access (HTTP {response.status_code})")
            body = bytearray()
            for chunk in response.iter_content(chunk_size=16_384):
                if not chunk:
                    continue
                if len(body) + len(chunk) > MAX_RESPONSE_BYTES:
                    raise AccessRefused("response exceeds the configured size limit")
                body.extend(chunk)
            return bytes(body).decode(response.encoding or "utf-8", errors="replace")