angee.integrate.http

Shared SSRF-pinned outbound HTTP for integration backends.

The single owner of "make one outbound HTTP request." The transport is httpx over a custom httpcore network backend (:class:_PinnedBackend) that resolves the host once and dials a validated IP — closing the resolve-then-connect (DNS-rebind) gap — while httpcore's start_tls(server_hostname=…) keeps TLS verification on the original hostname. The address judgement is owned by net.is_unsafe_address; this module only pins and dials.

One outbound-network policy lives here and everything composes it: integration backends reach it as self.http (:class:HttpClientMixin), and the OAuth client hands the same :class:PinnedTransport to Authlib. TLS trusts the system store (ssl.create_default_context) per docs/backend/guidelines.md, not httpx's bundled certifi default.

By default only public addresses are dialled. allow_private=True is the operator-configured-connection policy — a self-hosted host on a private network: it permits RFC-1918 / loopback so those connections work, but still rejects the SSRF escapes that have no legitimate target either way — cloud metadata (the well-known IPs, link-local 169.254/16, and the RFC 6598 shared range that front metadata services), multicast, and unspecified. Redirects are not followed unless follow_redirects=True; each hop re-enters the pinned backend, so following stays safe.

HTTP_TIMEOUT_SECONDS

Default timeout (seconds) for one outbound request.

HttpResponse

@dataclass(frozen=True, slots=True)
class HttpResponse()

One outbound HTTP response: the status code, raw body bytes, and headers.

ok

@property
def ok() -> bool

Return whether the status is a 2xx success.

json

def json() -> Any

Return the body parsed as JSON (None for an empty body).

header

def header(name: str) -> str

Return one response header by case-insensitive name, or "".

_PinnedBackend

class _PinnedBackend(httpcore.SyncBackend)

httpcore backend that resolves once, rejects SSRF-unsafe addresses, and dials a validated IP — so a DNS rebind between check and connect cannot move the request. net.is_unsafe_address owns the judgement; this only pins and dials.

__init__

def __init__(*, allow_private: bool) -> None

Bind the address policy for every connection this backend dials.

connect_tcp

def connect_tcp(
        host: str,
        port: int,
        timeout: float | None = None,
        local_address: str | None = None,
        socket_options: Iterable[Any] | None = None) -> httpcore.NetworkStream

Resolve host, reject unsafe addresses, and dial a validated IP.

host is the origin hostname; httpcore later calls start_tls(server_hostname=host), so dialing a validated IP here leaves SNI and certificate verification on the real hostname.

PinnedTransport

class PinnedTransport(httpx.HTTPTransport)

An httpx transport whose connections are SSRF-pinned and whose TLS trusts the system store. The shared pinned-httpx primitive: :class:HttpClient issues requests over it, and the OAuth client hands it to Authlib's OAuth2Client.

__init__

def __init__(*, allow_private: bool = False) -> None

Build a pinned transport; allow_private permits self-hosted RFC-1918 hosts.

HttpClient

class HttpClient()

A reusable SSRF-pinned outbound HTTP client over httpx.

Stateless to the caller — one instance per backend is fine. Each call gates the URL, pins via :class:PinnedTransport, and dials the validated IP; a DNS rebind between check and connect cannot redirect it. A caller-supplied Host header cannot displace the URL's real host. Redirects are followed only when follow_redirects=True (each hop re-validates).

get

def get(url: str,
        *,
        headers: dict[str, str] | None = None,
        allow_private: bool = False,
        follow_redirects: bool = False,
        timeout: int = HTTP_TIMEOUT_SECONDS) -> HttpResponse

GET url and return the response.

post

def post(url: str,
         *,
         headers: dict[str, str] | None = None,
         body: bytes | None = None,
         allow_private: bool = False,
         follow_redirects: bool = False,
         timeout: int = HTTP_TIMEOUT_SECONDS) -> HttpResponse

POST body to url and return the response.

request

def request(method: str,
            url: str,
            *,
            headers: dict[str, str] | None = None,
            body: bytes | None = None,
            allow_private: bool = False,
            follow_redirects: bool = False,
            timeout: int = HTTP_TIMEOUT_SECONDS) -> HttpResponse

Send one pinned request to url and return the response.

Raises ValidationError when the URL or a resolved address is rejected by the SSRF gate, and OSError when every validated address is unreachable.

HttpClientMixin

class HttpClientMixin()

Gives an integration backend the shared SSRF-pinned client as self.http.

Compose it into a backend that makes outbound calls (alongside its BridgeImpl / Client base) so it calls self.http.get(url, headers=…) rather than opening its own connection. HTTP stays opt-in this way — an implementation that does no I/O carries no client.

http

@cached_property
def http() -> HttpClient

Return this backend's shared outbound HTTP client.