"""Shaped network pipes: Linux netns + veth + tc/netem, validated like an instrument.

netshape makes cross-subject network comparisons legitimate by serving both

subjects from local disk through the *same* shaped pipe (a network namespace

connected to the host by a veth pair, with netem delay/jitter/loss/rate on

both directions). Profiles come from config/netshapes.json.

The pipe is an instrument; test it like one: self_test() measures achieved

RTT via TCP connect round trips through the shaped pipe and asserts it is

within +/-15% of the configured profile RTT before any campaign row is

trusted.

Skip-row honesty (ADR-0002): on non-Linux hosts (this includes the macOS dev

host), without CAP_NET_ADMIN, or with ip/tc missing, netshape_available()

returns (False, reason) and lanes emit returncode-77 skip rows; nothing in

this module raises merely because the host cannot shape packets.

Everything that decides *what* to run is a pure function over command lists

(netns_commands / teardown_commands / netem_args) so it is unit-testable

without root; only NetnsShaper executes, and only when available.

"""

from __future__ import annotations

import json

import os

import re

import shutil

import socket

import statistics

import subprocess

import sys

from dataclasses import dataclass

from pathlib import Path

from typing import Any, Callable, Optional

DEFAULT_NETSHAPES = Path(__file__).resolve().parents[2] / "config" / "netshapes.json"

SCHEMA_VERSION = 1

# Host side and namespace side of the veth pair. The subnet is deliberately

# obscure to avoid colliding with common LAN/docker ranges.

HOST_ADDR = "10.231.47.1"

NS_ADDR = "10.231.47.2"

SUBNET_PREFIX = 24

DEFAULT_NETNS_NAME = "oakshape"

_NAME_RE = re.compile(r"^[A-Za-z0-9_-]+$")

_IFNAMSIZ = 15  # Linux interface-name limit (IFNAMSIZ - 1)

RTT_TOLERANCE_PCT = 15.0

# A pure percentage tolerance is meaningless at rtt_ms=0 (loopback profile):

# any real syscall takes nonzero time. The floor keeps the assertion honest

# without making the zero-delay profile unfalsifiable-failing.

RTT_TOLERANCE_FLOOR_MS = 1.0

_PROFILE_KEYS = frozenset({"rtt_ms", "bandwidth_mbit", "loss_pct", "jitter_ms"})

class NetshapeConfigError(ValueError):

    """config/netshapes.json is missing, malformed, or out of contract."""

@dataclass(frozen=True)

class NetshapeProfile:

    """One shaped-pipe profile. rtt_ms/jitter_ms are round-trip figures;

    netem splits them half per direction. loss_pct and bandwidth_mbit are

    per-direction (each veth egress gets the full configured value)."""

    name: str

    rtt_ms: float

    bandwidth_mbit: Optional[float]  # null means unshaped bandwidth

    loss_pct: float

    jitter_ms: float

def _number(profile: str, key: str, value: Any) -> float:

    if isinstance(value, bool) or not isinstance(value, (int, float)):

        raise NetshapeConfigError(f"profile {profile!r}: {key} must be a number, got {value!r}")

    if value < 0:

        raise NetshapeConfigError(f"profile {profile!r}: {key} must be non-negative, got {value!r}")

    return float(value)

def _parse_profile(name: str, raw: Any) -> NetshapeProfile:

    if not isinstance(raw, dict):

        raise NetshapeConfigError(f"profile {name!r} must be an object, got {type(raw).__name__}")

    missing = sorted(_PROFILE_KEYS - raw.keys())

    if missing:

        raise NetshapeConfigError(f"profile {name!r} missing keys: {', '.join(missing)}")

    unknown = sorted(set(raw.keys()) - _PROFILE_KEYS)

    if unknown:

        raise NetshapeConfigError(f"profile {name!r} has unknown keys: {', '.join(unknown)}")

    loss_pct = _number(name, "loss_pct", raw["loss_pct"])

    if loss_pct > 100:

        raise NetshapeConfigError(f"profile {name!r}: loss_pct must be <= 100, got {loss_pct!r}")

    bandwidth = raw["bandwidth_mbit"]

    if bandwidth is not None:

        bandwidth = _number(name, "bandwidth_mbit", bandwidth)

        if bandwidth == 0:

            raise NetshapeConfigError(f"profile {name!r}: bandwidth_mbit must be positive or null")

    return NetshapeProfile(

        name=name,

        rtt_ms=_number(name, "rtt_ms", raw["rtt_ms"]),

        bandwidth_mbit=bandwidth,

        loss_pct=loss_pct,

        jitter_ms=_number(name, "jitter_ms", raw["jitter_ms"]),

    )

def load_netshapes(path: Path = DEFAULT_NETSHAPES) -> dict[str, NetshapeProfile]:

    """Load and validate the netshape profiles; raise NetshapeConfigError on

    any contract violation (a misconfigured instrument must never half-run)."""

    try:

        data = json.loads(Path(path).read_text())

    except (OSError, json.JSONDecodeError) as exc:

        raise NetshapeConfigError(f"cannot load netshapes config {path}: {exc}") from exc

    if not isinstance(data, dict):

        raise NetshapeConfigError("netshapes config root must be an object")

    if data.get("schema_version") != SCHEMA_VERSION:

        raise NetshapeConfigError(

            f"unsupported schema_version {data.get('schema_version')!r}; expected {SCHEMA_VERSION}"

        )

    profiles_raw = data.get("profiles")

    if not isinstance(profiles_raw, dict) or not profiles_raw:

        raise NetshapeConfigError("netshapes config needs a non-empty 'profiles' object")

    return {str(name): _parse_profile(str(name), raw) for name, raw in profiles_raw.items()}

def netshape_available(

    *,

    platform: Optional[str] = None,

    which: Optional[Callable[[str], Optional[str]]] = None,

    geteuid: Optional[Callable[[], int]] = None,

) -> tuple[bool, Optional[str]]:

    """(True, None) when this host can build shaped pipes, else (False, reason).

    The reason string is stable and machine-keyed (lanes embed it in

    skip_reason as "netshape_unavailable:<reason>"). Keyword overrides exist

    only for tests; the defaults are the real host probes. The privilege check

    is deliberately simple (euid == 0); CAP_NET_ADMIN-without-root setups can

    still run the lane under sudo.

    """

    platform = sys.platform if platform is None else platform

    which = shutil.which if which is None else which

    geteuid = os.geteuid if geteuid is None else geteuid

    if platform != "linux":

        return False, f"requires_linux:{platform}"

    for binary in ("ip", "tc"):

        if which(binary) is None:

            return False, f"missing_binary:{binary}"

    if geteuid() != 0:

        return False, "requires_cap_net_admin:euid_not_root"

    return True, None

def _fmt(value: float) -> str:

    """Compact numeric formatting for tc arguments: 50.0 -> '50', 0.5 -> '0.5'."""

    return f"{value:g}"

def netem_args(profile: NetshapeProfile) -> list[str]:

    """tc netem arguments for ONE direction of the pipe (pure).

    The configured rtt_ms/jitter_ms are round-trip figures, so each direction

    gets half; loss and rate apply per direction at their full configured

    values (that is what a lossy/limited duplex link does).

    """

    args = ["delay", f"{_fmt(profile.rtt_ms / 2)}ms"]

    if profile.jitter_ms:

        args.append(f"{_fmt(profile.jitter_ms / 2)}ms")

    if profile.loss_pct:

        args += ["loss", f"{_fmt(profile.loss_pct)}%"]

    if profile.bandwidth_mbit:

        args += ["rate", f"{_fmt(profile.bandwidth_mbit)}mbit"]

    return args

def _interface_names(name: str) -> tuple[str, str]:

    if not _NAME_RE.match(name):

        raise ValueError(f"netns name must match {_NAME_RE.pattern}, got {name!r}")

    host_if = f"ve-{name}-h"

    ns_if = f"ve-{name}-n"

    if len(host_if) > _IFNAMSIZ or len(ns_if) > _IFNAMSIZ:

        raise ValueError(f"netns name {name!r} too long for veth interface names (IFNAMSIZ)")

    return host_if, ns_if

def netns_commands(name: str, profile: NetshapeProfile) -> list[list[str]]:

    """The exact command lists that build the shaped pipe (pure, unit-testable

    without root). Order matters: namespace, veth pair, addressing, then netem

    on both directions."""

    host_if, ns_if = _interface_names(name)

    shaping = netem_args(profile)

    return [

        ["ip", "netns", "add", name],

        ["ip", "link", "add", host_if, "type", "veth", "peer", "name", ns_if],

        ["ip", "link", "set", ns_if, "netns", name],

        ["ip", "addr", "add", f"{HOST_ADDR}/{SUBNET_PREFIX}", "dev", host_if],

        ["ip", "link", "set", host_if, "up"],

        ["ip", "netns", "exec", name, "ip", "addr", "add", f"{NS_ADDR}/{SUBNET_PREFIX}", "dev", ns_if],

        ["ip", "netns", "exec", name, "ip", "link", "set", ns_if, "up"],

        ["ip", "netns", "exec", name, "ip", "link", "set", "lo", "up"],

        ["tc", "qdisc", "add", "dev", host_if, "root", "netem", *shaping],

        ["ip", "netns", "exec", name, "tc", "qdisc", "add", "dev", ns_if, "root", "netem", *shaping],

    ]

def teardown_commands(name: str) -> list[list[str]]:

    """Exact teardown command lists (pure). Deleting the host veth end removes

    its peer; deleting the namespace removes everything it still holds."""

    host_if, _ = _interface_names(name)

    return [

        ["ip", "link", "del", host_if],

        ["ip", "netns", "del", name],

    ]

def netns_exec(name: str, command: list[str]) -> list[str]:

    """Wrap a command so it runs inside the shaped namespace (pure)."""

    return ["ip", "netns", "exec", name, *command]

class NetnsShaper:

    """Builds and tears down one shaped pipe.

    Command construction is delegated to the pure functions above; this class

    only sequences execution. `run` and `available_fn` are injectable for

    tests — the default runner shells out and raises on failure (a half-built

    pipe must never be measured through).

    """

    def __init__(

        self,

        name: str = DEFAULT_NETNS_NAME,

        *,

        run: Optional[Callable[[list[str]], Any]] = None,

        available_fn: Callable[[], tuple[bool, Optional[str]]] = netshape_available,

    ) -> None:

        _interface_names(name)  # validate eagerly

        self.name = name

        self._run = run if run is not None else self._default_run

        self._available_fn = available_fn

        self.profile: Optional[NetshapeProfile] = None

        self._created = False

    @staticmethod

    def _default_run(command: list[str]) -> None:

        subprocess.run(

            command,

            check=True,

            stdout=subprocess.DEVNULL,

            stderr=subprocess.PIPE,

            text=True,

        )

    def create(self, profile: NetshapeProfile) -> "NetnsShaper":

        ok, reason = self._available_fn()

        if not ok:

            raise RuntimeError(f"netshape_unavailable:{reason}")

        # Mark created BEFORE the first command runs: a mid-create failure

        # leaves a partially built namespace/veth on the host, and teardown()

        # (guarded by _created) must still clean it up. The except below makes

        # cleanup happen even when the caller never reaches its own teardown;

        # teardown is per-command best-effort, so the original error is what

        # propagates.

        self._created = True

        try:

            for command in netns_commands(self.name, profile):

                self._run(command)

        except BaseException:

            self.teardown()

            raise

        self.profile = profile

        return self

    def teardown(self) -> None:

        if not self._created:

            return

        for command in teardown_commands(self.name):

            try:

                self._run(command)

            except (OSError, subprocess.CalledProcessError):

                # Best-effort: a vanished link/namespace is already torn down.

                pass

        self._created = False

        self.profile = None

    def __enter__(self) -> "NetnsShaper":

        return self

    def __exit__(self, exc_type, exc, tb) -> None:  # type: ignore[no-untyped-def]

        self.teardown()

    def tcp_ping(self, *, timeout_s: float = 30.0) -> float:

        """One TCP connect round trip through the shaped pipe, in milliseconds.

        A listener on the host end accepts; a tiny interpreter inside the

        namespace times the connect (the 3-way handshake costs one RTT).

        Process startup happens outside the timed window because the timing

        runs inside the child.

        """

        if not self._created:

            raise RuntimeError("tcp_ping requires a created shaper")

        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener:

            listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

            listener.bind((HOST_ADDR, 0))

            listener.listen(1)

            listener.settimeout(timeout_s)

            port = listener.getsockname()[1]

            script = (

                "import socket, time\n"

                "s = socket.socket()\n"

                "t = time.perf_counter()\n"

                f"s.connect(({HOST_ADDR!r}, {port}))\n"

                "print((time.perf_counter() - t) * 1000)\n"

                "s.close()\n"

            )

            proc = subprocess.Popen(

                netns_exec(self.name, [sys.executable, "-c", script]),

                stdout=subprocess.PIPE,

                stderr=subprocess.PIPE,

                text=True,

            )

            try:

                conn, _addr = listener.accept()

                conn.close()

            finally:

                out, err = proc.communicate(timeout=timeout_s)

            if proc.returncode != 0:

                raise RuntimeError(f"tcp_ping failed inside netns: {err.strip()}")

            return float(out.strip())

    def self_test(

        self,

        profile: Optional[NetshapeProfile] = None,

        *,

        ping_fn: Optional[Callable[[], float]] = None,

        samples: int = 5,

    ) -> dict[str, Any]:

        target = profile if profile is not None else self.profile

        if target is None:

            raise ValueError("self_test needs a profile (none configured on this shaper)")

        return self_test(target, ping_fn=ping_fn, samples=samples, shaper=self)

def self_test(

    profile: NetshapeProfile,

    *,

    ping_fn: Optional[Callable[[], float]] = None,

    samples: int = 5,

    shaper: Optional[NetnsShaper] = None,

) -> dict[str, Any]:

    """Measure achieved RTT through the shaped pipe and compare to configured.

    Returns a self-test row dict; the lane must refuse to run a campaign when

    within_tolerance is false. ping_fn (one RTT sample in ms per call) is

    injectable for tests; the default is the shaper's real TCP connect ping.

    The median of `samples` round trips smooths scheduler noise without hiding

    a misconfigured qdisc.

    """

    if ping_fn is None:

        if shaper is None:

            raise ValueError("self_test needs ping_fn or a shaper to ping through")

        ping_fn = shaper.tcp_ping

    samples_ms = [float(ping_fn()) for _ in range(max(1, samples))]

    measured = statistics.median(samples_ms)

    tolerance_ms = max(profile.rtt_ms * RTT_TOLERANCE_PCT / 100.0, RTT_TOLERANCE_FLOOR_MS)

    within = abs(measured - profile.rtt_ms) <= tolerance_ms

    return {

        "operation": "netshape.selftest",

        "netshape_profile": profile.name,

        "configured_rtt_ms": profile.rtt_ms,

        "measured_rtt_ms": round(measured, 3),

        "rtt_samples_ms": [round(sample, 3) for sample in samples_ms],

        "tolerance_pct": RTT_TOLERANCE_PCT,

        "tolerance_floor_ms": RTT_TOLERANCE_FLOOR_MS,

        "tolerance_ms": round(tolerance_ms, 3),

        "within_tolerance": within,

    }