"""Client for the relay API (docs/V4-FEDERATION.md §4) — the brain's only way to
reach the relay. Push-only doctrine holds here too: every call is an OUTBOUND HTTPS
request (register, deposit, poll, ack, pair); the brain never listens.

Two transports behind one surface (§10.3): by default the client speaks HTTP with
stdlib ``urllib`` (no new dependency); when constructed with ``core=<relay.Relay>``
it calls that in-process relay directly, so the P4/P5 two-brain tests drive the real
relay logic with no network and no Starlette. Every method returns ``(status,
payload)`` — the same shape both transports produce — and callers interpret the
status (202/403/409/404 are expected control flow, not exceptions). Only a genuine
transport failure (connection refused, timeout) raises :class:`RelayUnavailable`.
"""
import base64
import ipaddress
import json
import socket
import urllib.error
import urllib.request
from urllib.parse import urlsplit

DEFAULT_TIMEOUT = 15


def _ip_literal(host):
    """The ``ipaddress`` object if ``host`` is an IP literal (IPv4, or IPv6 whose
    surrounding brackets ``urlsplit`` has already stripped), else ``None`` — a
    registered DNS name."""
    if not host:
        return None
    try:
        return ipaddress.ip_address(host)
    except ValueError:
        return None


def _ip_is_disallowed(ip):
    """True if ``ip`` is one we must never make an outbound request to: loopback,
    RFC1918 private, link-local (169.254/16, fe80::/10), unique-local (fc00::/7),
    reserved, unspecified, or multicast — IPv4 AND IPv6. Any IPv4 embedded in a v6
    address (``::ffff:0:0/96`` mapped, 6to4, Teredo) is re-checked against the same
    policy, since embedding an internal v4 in a v6 literal is a classic way to slip
    past a v6-only classifier."""
    embedded = getattr(ip, "ipv4_mapped", None) or getattr(ip, "sixtofour", None)
    if embedded is not None and _ip_is_disallowed(embedded):
        return True
    teredo = getattr(ip, "teredo", None)
    if teredo and any(_ip_is_disallowed(part) for part in teredo):
        return True
    return (ip.is_loopback or ip.is_private or ip.is_link_local
            or ip.is_reserved or ip.is_unspecified or ip.is_multicast)


def is_safe_relay_url(url):
    """Is ``url`` a relay endpoint we may make an OUTBOUND request to?

    A peer's ``relay`` is attacker-controlled data — it rides in verbatim from the
    sealed pairing blob (``pairing._peer_entry`` -> ``config.peers[].relay``), and
    the verification words are derived from PUBKEYS, so the owner's out-of-band word
    comparison cannot vet it. This is the SSRF gate on that value: a False here and
    the brain never connects. As a pure, I/O-free predicate it enforces:

    * scheme is ``http`` or ``https`` (never ``file``/``ftp``/``gopher``/…);
    * NO embedded credentials — ``user[:pass]@host`` is refused outright (never
      legitimate for a relay, and a classic way to disguise the real host, e.g.
      ``https://relay.example.com@169.254.169.254/``);
    * an IP-literal host is range-checked and REJECTED if loopback / private /
      link-local / unique-local / reserved / unspecified / multicast, v4 AND v6 —
      under https too, not only http. So ``https://169.254.169.254/``,
      ``https://10.0.0.5:9200/``, ``https://[::1]/``, ``https://[fe80::1]/``,
      ``https://[::ffff:10.0.0.5]/`` … all fail here (the old "https anywhere"
      blind spot);
    * plaintext ``http`` is admitted ONLY to real loopback ``127.0.0.1[:port]``
      (the deliberate dev / local-relay loophole; nothing else may use http).

    A host that is a DNS NAME (not an IP literal) passes iff https — this predicate
    cannot see what it resolves to. The resolve-and-range-check that closes the
    DNS-name and DNS-rebinding gap (e.g. ``metadata.google.internal``), plus the
    refusal to follow redirects to an unvalidated host, live at connect time in
    :meth:`RelayClient._guard_target` and the no-redirect opener below."""
    try:
        p = urlsplit(str(url))
    except ValueError:
        return False
    if p.scheme not in ("http", "https"):
        return False
    if p.username is not None or p.password is not None:   # any userinfo -> refuse
        return False
    host = p.hostname
    if not host:
        return False
    if p.scheme == "http":
        return host == "127.0.0.1"                         # the one plaintext exception
    ip = _ip_literal(host)                                 # https from here
    if ip is not None:
        return not _ip_is_disallowed(ip)                   # literal must be public
    return True                                            # DNS name -> checked at connect


class RelayError(RuntimeError):
    """Base class for relay-client failures."""


class RelayUnavailable(RelayError):
    """The relay could not be reached (connection/timeout) — worth retrying."""


class RelayRefused(RelayUnavailable):
    """Refused to SEND: the target failed a security check — an unsafe relay URL, a
    host that resolved to an internal address, or a 3xx redirect to an unvalidated
    host. A subclass of :class:`RelayUnavailable` so callers that already tolerate
    an unreachable relay tolerate a refused one; but the outbox worker treats it as
    TERMINAL (reject, never retry) — a security refusal will not heal on its own."""


class _RefuseRedirects(urllib.request.HTTPRedirectHandler):
    """Refuse every 3xx. A relay has no legitimate reason to redirect a register /
    deposit / poll call, and following one would resend the request — including the
    ``X-Deposit-Token`` header and the sealed envelope — to a host that passed no
    SSRF check (the redirect-to-internal vector). Raise, rather than silently
    return the 3xx, so the caller sees a hard refusal instead of a mystery status."""

    def redirect_request(self, req, fp, code, msg, headers, newurl):
        raise RelayRefused(f"relay attempted a {code} redirect to {newurl!r} — refused")


# One opener, built once: it follows NO redirects (the handler above raises).
_OPENER = urllib.request.build_opener(_RefuseRedirects())


class RelayClient:
    def __init__(self, relay_url, *, core=None, timeout=DEFAULT_TIMEOUT):
        self.base = (relay_url or "").rstrip("/")
        self._core = core          # an in-process relay.Relay for tests; None -> HTTP
        self.timeout = timeout

    def for_relay(self, relay_url):
        """A client for depositing into a peer's mailbox on the PEER's relay.

        In-proc (test) mode has one shared relay, so return self; over HTTP, reuse
        self when the URL matches our own relay, else make a client for the peer's
        relay — so a peer on a different relay actually receives the deposit."""
        if self._core is not None or not relay_url or (relay_url or "").rstrip("/") == self.base:
            return self
        return RelayClient(relay_url, timeout=self.timeout)

    # ---- HTTP transport -----------------------------------------------------
    def _guard_target(self):
        """SSRF gate applied to ``self.base`` before every outbound request.

        Belt-and-braces: a ``RelayClient`` can be constructed directly with any
        base (bypassing the worker's per-peer :func:`is_safe_relay_url` check), so
        we re-check it here — and, for an https DNS name (which is_safe_relay_url
        cannot resolve), we resolve it NOW and refuse if any address it maps to is
        internal. That closes the DNS-name SSRF (e.g. ``metadata.google.internal``
        -> 169.254.169.254) and, because the check sits right before the request,
        collapses most of the DNS-rebinding window too. Raises :class:`RelayRefused`
        (unsafe URL, or an internal resolution) or :class:`RelayUnavailable` (the
        name does not resolve at all)."""
        if not is_safe_relay_url(self.base):
            raise RelayRefused(f"refusing outbound request to unsafe relay URL {self.base!r}")
        p = urlsplit(self.base)
        host = p.hostname
        # An IP literal was already range-checked by is_safe_relay_url; http is
        # loopback-only and safe. Only an https DNS name still needs resolving.
        if p.scheme != "https" or _ip_literal(host) is not None:
            return
        try:
            infos = socket.getaddrinfo(host, p.port or 443, type=socket.SOCK_STREAM)
        except OSError as exc:
            raise RelayUnavailable(f"relay {self.base} unresolvable: {exc}") from exc
        for info in infos:
            addr = info[4][0]
            ip = _ip_literal(addr)
            if ip is not None and _ip_is_disallowed(ip):
                raise RelayRefused(
                    f"relay {host} resolved to internal address {addr} — refused (SSRF guard)")

    def _http(self, method, path, *, bearer=None, deposit_token=None,
              json_body=None, raw_body=None):
        self._guard_target()
        headers = {"Accept": "application/json"}
        data = None
        if bearer:
            headers["Authorization"] = "Bearer " + bearer
        if deposit_token:
            headers["X-Deposit-Token"] = deposit_token
        if json_body is not None:
            data = json.dumps(json_body).encode("utf-8")
            headers["Content-Type"] = "application/json"
        elif raw_body is not None:
            data = raw_body
            headers["Content-Type"] = "application/octet-stream"
        req = urllib.request.Request(self.base + path, data=data, method=method,
                                     headers=headers)
        try:
            # _OPENER refuses redirects (a 3xx raises RelayRefused, not followed).
            with _OPENER.open(req, timeout=self.timeout) as resp:  # noqa: S310
                body, status = resp.read(), resp.status
        except urllib.error.HTTPError as exc:
            body, status = exc.read(), exc.code
        except (urllib.error.URLError, OSError) as exc:
            raise RelayUnavailable(f"relay {self.base} unreachable: {exc}") from exc
        payload = None
        if body:
            try:
                payload = json.loads(body)
            except ValueError:
                payload = None
        return status, payload

    # ---- the API (dispatches to the in-proc core, or to HTTP) ---------------
    def create_mailbox(self, fingerprint, pubkey_b64):
        if self._core is not None:
            return self._core.create_mailbox(fingerprint, pubkey_b64, client_ip="127.0.0.1")
        return self._http("POST", "/v1/mbx",
                          json_body={"fingerprint": fingerprint, "pubkey": pubkey_b64})

    def add_token(self, fp, bearer, token_sha256):
        if self._core is not None:
            return self._core.add_token(fp, bearer, token_sha256)
        return self._http("POST", f"/v1/mbx/{fp}/tokens", bearer=bearer,
                          json_body={"token_sha256": token_sha256})

    def delete_token(self, fp, bearer, token_sha):
        if self._core is not None:
            return self._core.delete_token(fp, bearer, token_sha)
        return self._http("DELETE", f"/v1/mbx/{fp}/tokens/{token_sha}", bearer=bearer)

    def deposit(self, fp, token, env_bytes):
        if self._core is not None:
            return self._core.deposit(fp, token, env_bytes, client_ip="127.0.0.1")
        return self._http("POST", f"/v1/mbx/{fp}", deposit_token=token, raw_body=env_bytes)

    def poll(self, fp, bearer, limit=50, after=None):
        """``after`` = a blob id: page past it (relay 1.2.0; older relays ignore it and
        answer from the head — the receiver notices and stops) — how the receiver reaches
        what sits behind a blob it has chosen not to ack yet."""
        if self._core is not None:
            return self._core.poll(fp, bearer, limit, after)
        path = f"/v1/mbx/{fp}?limit={int(limit)}"
        if after:
            path += f"&after={after}"
        return self._http("GET", path, bearer=bearer)

    def delete_blob(self, fp, bearer, blob_id):
        if self._core is not None:
            return self._core.delete_blob(fp, bearer, blob_id)
        return self._http("DELETE", f"/v1/mbx/{fp}/{blob_id}", bearer=bearer)

    def birth(self, fp, bearer, record):
        """The birth record (docs/RELAY-CENSUS.md §4.2): bearer-bound PUT; 201 on
        the first write, 200 on a refresh. Outbound, like everything here."""
        if self._core is not None:
            return self._core.birth(fp, bearer, record)
        return self._http("PUT", f"/v1/mbx/{fp}/birth", bearer=bearer, json_body=record)

    def pair_put(self, rid, role, blob_bytes):
        blob_b64 = base64.b64encode(blob_bytes).decode("ascii")
        if self._core is not None:
            return self._core.pair_put(rid, role, blob_b64, client_ip="127.0.0.1")
        return self._http("POST", f"/v1/pair/{rid}",
                          json_body={"role": role, "blob_b64": blob_b64})

    def pair_get(self, rid, role):
        if self._core is not None:
            return self._core.pair_get(rid, role, client_ip="127.0.0.1")
        return self._http("GET", f"/v1/pair/{rid}?role={role}")

    # ---- convenience: the blob bytes out of a pair_get / poll payload -------
    @staticmethod
    def blob_of(payload):
        """Decode the base64 blob a pair_get returned (or None)."""
        if not payload or "blob_b64" not in payload:
            return None
        return base64.b64decode(payload["blob_b64"])
