egress_guard.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/egress_guard.py
 
"""
Network-egress policy (M8.0 WU-4; §5.1.3, F2/F2a).
 
Purpose
-------
A shared SSRF guard for every in-process network sink. The destination host is
resolved and every A/AAAA result classified; non-public classes are refused --
loopback, RFC1918 / IPv6 ULA (private), link-local (incl. the cloud-metadata
address 169.254.169.254 and IMDSv6), multicast, reserved, and unspecified.
URLs are ``https``-only by default and embedded credentials are refused. Callers
re-validate each redirect hop (``follow_redirects=False``).
 
Closed vector: a literal-internal-IP or DNS-resolves-to-internal target (the
common SSRF -- IMDS, loopback services, LAN hosts). Documented residual: a
DNS-rebinding flip between this validation and the client's own connect-time
re-resolution; true socket-pinning / the cross-process broker close that, and the
provider channel (§5.1.5) reuses :func:`is_blocked_ip` for its own classification.
 
Responsibilities
----------------
- :func:`is_blocked_ip` -- classify one address: ``True`` unless it is a
  globally-routable public address (an IPv4-mapped IPv6 form is unwrapped first).
- :func:`validate_url_egress` -- the per-hop URL check for ``fetch_url``: scheme,
  embedded credentials, host resolution, and refusal if any resolved address is
  blocked.
- :func:`is_forbidden_provider_host` -- the provider-endpoint check (F72): flag an
  IP-literal host in the link-local, multicast or reserved classes.
- :data:`MAX_REDIRECTS` -- the redirect-hop ceiling ``fetch_url`` follows.
- :class:`EgressDenied` -- the exception every refusal raises.
 
Contracts
---------
- Fail closed: an unparseable address string counts as blocked; a non-``https``
  scheme, embedded credentials, a missing host or an unresolvable host raises
  :class:`EgressDenied`.
- :func:`validate_url_egress` refuses when ANY resolved A/AAAA address is blocked,
  and returns the validated host otherwise. It checks one hop; the caller disables
  automatic redirects and calls it again for every redirect target.
- :func:`is_forbidden_provider_host` never resolves DNS: loopback, RFC1918 and
  public IP literals are not forbidden (the F72c operator-trusted endpoint
  posture), and a bare hostname is not flagged (a hostname resolving to an
  internal address is the M8.3 endpoint-authority residual).
 
Diagnostics
-----------
Domain: OC_CHAT.CORE
Levels:
    L2 -- a refused destination: the source label, host and blocked addresses
    L3 -- (none) allowed destinations are not logged per call
    L4 -- deep tracing
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg of refused destinations)
"""
from __future__ import annotations
 
import ipaddress
import socket
from urllib.parse import urlsplit
 
from oc_diagnostics import _dbg
 
#: Max redirect hops fetch_url follows (Appendix B).
MAX_REDIRECTS = 5
 
#: Allowed URL schemes for fetch_url (https-only by default, §5.1.3).
_ALLOWED_SCHEMES = frozenset({"https"})
 
 
class EgressDenied(Exception):
    """Raised when a network destination is refused by egress policy."""
 
 
def is_blocked_ip(ip: "str | ipaddress.IPv4Address | ipaddress.IPv6Address") -> bool:
    # OCT-LINT: disable=no_hardcoded_secrets reason="Function docstring prose listing the refused address classes; no credential material (operator-ruled 2026-09-28)"
    """Return ``True`` if *ip* is NOT a globally-routable public address.
 
    Refuses loopback / private (RFC1918 + IPv6 ULA) / link-local (incl. IMDS
    ``169.254.169.254`` and IMDSv6) / multicast / reserved / unspecified, and the
    IPv4-mapped IPv6 form (``::ffff:127.0.0.1``) of any of those. An unparseable
    value is refused (fail-closed).
    """
    try:
        addr = ipaddress.ip_address(ip) if isinstance(ip, str) else ip
    except ValueError:
        return True
    if isinstance(addr, ipaddress.IPv6Address) and addr.ipv4_mapped is not None:
        addr = addr.ipv4_mapped
    return (
        addr.is_loopback or addr.is_private or addr.is_link_local
        or addr.is_multicast or addr.is_reserved or addr.is_unspecified
        or not addr.is_global
    )
 
 
def _resolved_ips(host: str, port: int) -> list[str]:
    """Resolve *host* to all A/AAAA addresses (may raise :class:`socket.gaierror`)."""
    infos = socket.getaddrinfo(host, port, proto=socket.IPPROTO_TCP)
    return [info[4][0] for info in infos]
 
 
def validate_url_egress(url: str, *, source_label: str = "fetch_url") -> str:
    """Validate *url* for outbound egress; raise :class:`EgressDenied` if refused.
 
    Enforces the ``https``-only scheme, refuses embedded credentials, resolves the
    host, and refuses if ANY resolved IP is in a blocked class. Returns the
    validated host. Call once per redirect hop.
    """
    func = "validate_url_egress"
    parts = urlsplit(url)
    scheme = (parts.scheme or "").lower()
    if scheme not in _ALLOWED_SCHEMES:
        raise EgressDenied(f"scheme {scheme!r} not allowed (https only): {url}")
    if parts.username or parts.password:
        raise EgressDenied("URL credentials are not allowed")
    host = parts.hostname
    if not host:
        raise EgressDenied(f"no host in URL: {url}")
    port = parts.port or 443
    try:
        ips = _resolved_ips(host, port)
    except socket.gaierror as exc:
        raise EgressDenied(f"cannot resolve host {host!r}: {exc}") from exc
    blocked = sorted({ip for ip in ips if is_blocked_ip(ip)})
    if blocked:
        _dbg("OC_CHAT.CORE", func,
             f"{source_label}: blocked {host} -> {blocked}", 2)
        raise EgressDenied(
            f"host {host!r} resolves to a non-public address "
            f"(loopback/private/link-local/metadata); refused"
        )
    return host
 
 
def is_forbidden_provider_host(host_or_url: str) -> bool:
    """Return ``True`` if *host_or_url* is an IP literal that is never a
    legitimate provider endpoint (M8.0 WU-4c; F72).
 
    Forbidden = link-local (incl. cloud-metadata ``169.254.169.254`` and IMDSv6),
    multicast, or reserved. Loopback (an Ollama on localhost), RFC1918 (a LAN
    Ollama box), and public hosts (cloud APIs) are NOT forbidden -- the
    cloud-API-normal posture (F72c) allows operator-trusted endpoints of any of
    those classes. Bare hostnames are not flagged here: the closed vector is the
    literal-IP injection (``DEEPSEEK_BASE_URL=http://169.254.169.254``); a
    hostname that resolves to an internal address is the M8.3 endpoint-authority
    residual (the remote-requires-OOB-authorization gate).
    """
    host = urlsplit(host_or_url).hostname if "://" in host_or_url else host_or_url
    host = (host or "").strip().strip("[]")
    try:
        addr = ipaddress.ip_address(host)
    except ValueError:
        return False
    if isinstance(addr, ipaddress.IPv6Address) and addr.ipv4_mapped is not None:
        addr = addr.ipv4_mapped
    return addr.is_link_local or addr.is_multicast or addr.is_reserved
 
 
__all__ = [
    "EgressDenied", "MAX_REDIRECTS", "is_blocked_ip",
    "is_forbidden_provider_host", "validate_url_egress",
]