| 1 | """Shaped network pipes: Linux netns + veth + tc/netem, validated like an instrument.
|
| 2 |
|
| 3 | netshape makes cross-subject network comparisons legitimate by serving both
|
| 4 | subjects from local disk through the *same* shaped pipe (a network namespace
|
| 5 | connected to the host by a veth pair, with netem delay/jitter/loss/rate on
|
| 6 | both directions). Profiles come from config/netshapes.json.
|
| 7 |
|
| 8 | The pipe is an instrument; test it like one: self_test() measures achieved
|
| 9 | RTT via TCP connect round trips through the shaped pipe and asserts it is
|
| 10 | within +/-15% of the configured profile RTT before any campaign row is
|
| 11 | trusted.
|
| 12 |
|
| 13 | Skip-row honesty (ADR-0002): on non-Linux hosts (this includes the macOS dev
|
| 14 | host), without CAP_NET_ADMIN, or with ip/tc missing, netshape_available()
|
| 15 | returns (False, reason) and lanes emit returncode-77 skip rows; nothing in
|
| 16 | this module raises merely because the host cannot shape packets.
|
| 17 |
|
| 18 | Everything that decides *what* to run is a pure function over command lists
|
| 19 | (netns_commands / teardown_commands / netem_args) so it is unit-testable
|
| 20 | without root; only NetnsShaper executes, and only when available.
|
| 21 | """
|
| 22 |
|
| 23 | from __future__ import annotations
|
| 24 |
|
| 25 | import json
|
| 26 | import os
|
| 27 | import re
|
| 28 | import shutil
|
| 29 | import socket
|
| 30 | import statistics
|
| 31 | import subprocess
|
| 32 | import sys
|
| 33 | from dataclasses import dataclass
|
| 34 | from pathlib import Path
|
| 35 | from typing import Any, Callable, Optional
|
| 36 |
|
| 37 | DEFAULT_NETSHAPES = Path(__file__).resolve().parents[2] / "config" / "netshapes.json"
|
| 38 | SCHEMA_VERSION = 1
|
| 39 |
|
| 40 | # Host side and namespace side of the veth pair. The subnet is deliberately
|
| 41 | # obscure to avoid colliding with common LAN/docker ranges.
|
| 42 | HOST_ADDR = "10.231.47.1"
|
| 43 | NS_ADDR = "10.231.47.2"
|
| 44 | SUBNET_PREFIX = 24
|
| 45 |
|
| 46 | DEFAULT_NETNS_NAME = "oakshape"
|
| 47 | _NAME_RE = re.compile(r"^[A-Za-z0-9_-]+$")
|
| 48 | _IFNAMSIZ = 15 # Linux interface-name limit (IFNAMSIZ - 1)
|
| 49 |
|
| 50 | RTT_TOLERANCE_PCT = 15.0
|
| 51 | # A pure percentage tolerance is meaningless at rtt_ms=0 (loopback profile):
|
| 52 | # any real syscall takes nonzero time. The floor keeps the assertion honest
|
| 53 | # without making the zero-delay profile unfalsifiable-failing.
|
| 54 | RTT_TOLERANCE_FLOOR_MS = 1.0
|
| 55 |
|
| 56 | _PROFILE_KEYS = frozenset({"rtt_ms", "bandwidth_mbit", "loss_pct", "jitter_ms"})
|
| 57 |
|
| 58 |
|
| 59 | class NetshapeConfigError(ValueError):
|
| 60 | """config/netshapes.json is missing, malformed, or out of contract."""
|
| 61 |
|
| 62 |
|
| 63 | @dataclass(frozen=True)
|
| 64 | class NetshapeProfile:
|
| 65 | """One shaped-pipe profile. rtt_ms/jitter_ms are round-trip figures;
|
| 66 | netem splits them half per direction. loss_pct and bandwidth_mbit are
|
| 67 | per-direction (each veth egress gets the full configured value)."""
|
| 68 |
|
| 69 | name: str
|
| 70 | rtt_ms: float
|
| 71 | bandwidth_mbit: Optional[float] # null means unshaped bandwidth
|
| 72 | loss_pct: float
|
| 73 | jitter_ms: float
|
| 74 |
|
| 75 |
|
| 76 | def _number(profile: str, key: str, value: Any) -> float:
|
| 77 | if isinstance(value, bool) or not isinstance(value, (int, float)):
|
| 78 | raise NetshapeConfigError(f"profile {profile!r}: {key} must be a number, got {value!r}")
|
| 79 | if value < 0:
|
| 80 | raise NetshapeConfigError(f"profile {profile!r}: {key} must be non-negative, got {value!r}")
|
| 81 | return float(value)
|
| 82 |
|
| 83 |
|
| 84 | def _parse_profile(name: str, raw: Any) -> NetshapeProfile:
|
| 85 | if not isinstance(raw, dict):
|
| 86 | raise NetshapeConfigError(f"profile {name!r} must be an object, got {type(raw).__name__}")
|
| 87 | missing = sorted(_PROFILE_KEYS - raw.keys())
|
| 88 | if missing:
|
| 89 | raise NetshapeConfigError(f"profile {name!r} missing keys: {', '.join(missing)}")
|
| 90 | unknown = sorted(set(raw.keys()) - _PROFILE_KEYS)
|
| 91 | if unknown:
|
| 92 | raise NetshapeConfigError(f"profile {name!r} has unknown keys: {', '.join(unknown)}")
|
| 93 | loss_pct = _number(name, "loss_pct", raw["loss_pct"])
|
| 94 | if loss_pct > 100:
|
| 95 | raise NetshapeConfigError(f"profile {name!r}: loss_pct must be <= 100, got {loss_pct!r}")
|
| 96 | bandwidth = raw["bandwidth_mbit"]
|
| 97 | if bandwidth is not None:
|
| 98 | bandwidth = _number(name, "bandwidth_mbit", bandwidth)
|
| 99 | if bandwidth == 0:
|
| 100 | raise NetshapeConfigError(f"profile {name!r}: bandwidth_mbit must be positive or null")
|
| 101 | return NetshapeProfile(
|
| 102 | name=name,
|
| 103 | rtt_ms=_number(name, "rtt_ms", raw["rtt_ms"]),
|
| 104 | bandwidth_mbit=bandwidth,
|
| 105 | loss_pct=loss_pct,
|
| 106 | jitter_ms=_number(name, "jitter_ms", raw["jitter_ms"]),
|
| 107 | )
|
| 108 |
|
| 109 |
|
| 110 | def load_netshapes(path: Path = DEFAULT_NETSHAPES) -> dict[str, NetshapeProfile]:
|
| 111 | """Load and validate the netshape profiles; raise NetshapeConfigError on
|
| 112 | any contract violation (a misconfigured instrument must never half-run)."""
|
| 113 | try:
|
| 114 | data = json.loads(Path(path).read_text())
|
| 115 | except (OSError, json.JSONDecodeError) as exc:
|
| 116 | raise NetshapeConfigError(f"cannot load netshapes config {path}: {exc}") from exc
|
| 117 | if not isinstance(data, dict):
|
| 118 | raise NetshapeConfigError("netshapes config root must be an object")
|
| 119 | if data.get("schema_version") != SCHEMA_VERSION:
|
| 120 | raise NetshapeConfigError(
|
| 121 | f"unsupported schema_version {data.get('schema_version')!r}; expected {SCHEMA_VERSION}"
|
| 122 | )
|
| 123 | profiles_raw = data.get("profiles")
|
| 124 | if not isinstance(profiles_raw, dict) or not profiles_raw:
|
| 125 | raise NetshapeConfigError("netshapes config needs a non-empty 'profiles' object")
|
| 126 | return {str(name): _parse_profile(str(name), raw) for name, raw in profiles_raw.items()}
|
| 127 |
|
| 128 |
|
| 129 | def netshape_available(
|
| 130 | *,
|
| 131 | platform: Optional[str] = None,
|
| 132 | which: Optional[Callable[[str], Optional[str]]] = None,
|
| 133 | geteuid: Optional[Callable[[], int]] = None,
|
| 134 | ) -> tuple[bool, Optional[str]]:
|
| 135 | """(True, None) when this host can build shaped pipes, else (False, reason).
|
| 136 |
|
| 137 | The reason string is stable and machine-keyed (lanes embed it in
|
| 138 | skip_reason as "netshape_unavailable:<reason>"). Keyword overrides exist
|
| 139 | only for tests; the defaults are the real host probes. The privilege check
|
| 140 | is deliberately simple (euid == 0); CAP_NET_ADMIN-without-root setups can
|
| 141 | still run the lane under sudo.
|
| 142 | """
|
| 143 | platform = sys.platform if platform is None else platform
|
| 144 | which = shutil.which if which is None else which
|
| 145 | geteuid = os.geteuid if geteuid is None else geteuid
|
| 146 | if platform != "linux":
|
| 147 | return False, f"requires_linux:{platform}"
|
| 148 | for binary in ("ip", "tc"):
|
| 149 | if which(binary) is None:
|
| 150 | return False, f"missing_binary:{binary}"
|
| 151 | if geteuid() != 0:
|
| 152 | return False, "requires_cap_net_admin:euid_not_root"
|
| 153 | return True, None
|
| 154 |
|
| 155 |
|
| 156 | def _fmt(value: float) -> str:
|
| 157 | """Compact numeric formatting for tc arguments: 50.0 -> '50', 0.5 -> '0.5'."""
|
| 158 | return f"{value:g}"
|
| 159 |
|
| 160 |
|
| 161 | def netem_args(profile: NetshapeProfile) -> list[str]:
|
| 162 | """tc netem arguments for ONE direction of the pipe (pure).
|
| 163 |
|
| 164 | The configured rtt_ms/jitter_ms are round-trip figures, so each direction
|
| 165 | gets half; loss and rate apply per direction at their full configured
|
| 166 | values (that is what a lossy/limited duplex link does).
|
| 167 | """
|
| 168 | args = ["delay", f"{_fmt(profile.rtt_ms / 2)}ms"]
|
| 169 | if profile.jitter_ms:
|
| 170 | args.append(f"{_fmt(profile.jitter_ms / 2)}ms")
|
| 171 | if profile.loss_pct:
|
| 172 | args += ["loss", f"{_fmt(profile.loss_pct)}%"]
|
| 173 | if profile.bandwidth_mbit:
|
| 174 | args += ["rate", f"{_fmt(profile.bandwidth_mbit)}mbit"]
|
| 175 | return args
|
| 176 |
|
| 177 |
|
| 178 | def _interface_names(name: str) -> tuple[str, str]:
|
| 179 | if not _NAME_RE.match(name):
|
| 180 | raise ValueError(f"netns name must match {_NAME_RE.pattern}, got {name!r}")
|
| 181 | host_if = f"ve-{name}-h"
|
| 182 | ns_if = f"ve-{name}-n"
|
| 183 | if len(host_if) > _IFNAMSIZ or len(ns_if) > _IFNAMSIZ:
|
| 184 | raise ValueError(f"netns name {name!r} too long for veth interface names (IFNAMSIZ)")
|
| 185 | return host_if, ns_if
|
| 186 |
|
| 187 |
|
| 188 | def netns_commands(name: str, profile: NetshapeProfile) -> list[list[str]]:
|
| 189 | """The exact command lists that build the shaped pipe (pure, unit-testable
|
| 190 | without root). Order matters: namespace, veth pair, addressing, then netem
|
| 191 | on both directions."""
|
| 192 | host_if, ns_if = _interface_names(name)
|
| 193 | shaping = netem_args(profile)
|
| 194 | return [
|
| 195 | ["ip", "netns", "add", name],
|
| 196 | ["ip", "link", "add", host_if, "type", "veth", "peer", "name", ns_if],
|
| 197 | ["ip", "link", "set", ns_if, "netns", name],
|
| 198 | ["ip", "addr", "add", f"{HOST_ADDR}/{SUBNET_PREFIX}", "dev", host_if],
|
| 199 | ["ip", "link", "set", host_if, "up"],
|
| 200 | ["ip", "netns", "exec", name, "ip", "addr", "add", f"{NS_ADDR}/{SUBNET_PREFIX}", "dev", ns_if],
|
| 201 | ["ip", "netns", "exec", name, "ip", "link", "set", ns_if, "up"],
|
| 202 | ["ip", "netns", "exec", name, "ip", "link", "set", "lo", "up"],
|
| 203 | ["tc", "qdisc", "add", "dev", host_if, "root", "netem", *shaping],
|
| 204 | ["ip", "netns", "exec", name, "tc", "qdisc", "add", "dev", ns_if, "root", "netem", *shaping],
|
| 205 | ]
|
| 206 |
|
| 207 |
|
| 208 | def teardown_commands(name: str) -> list[list[str]]:
|
| 209 | """Exact teardown command lists (pure). Deleting the host veth end removes
|
| 210 | its peer; deleting the namespace removes everything it still holds."""
|
| 211 | host_if, _ = _interface_names(name)
|
| 212 | return [
|
| 213 | ["ip", "link", "del", host_if],
|
| 214 | ["ip", "netns", "del", name],
|
| 215 | ]
|
| 216 |
|
| 217 |
|
| 218 | def netns_exec(name: str, command: list[str]) -> list[str]:
|
| 219 | """Wrap a command so it runs inside the shaped namespace (pure)."""
|
| 220 | return ["ip", "netns", "exec", name, *command]
|
| 221 |
|
| 222 |
|
| 223 | class NetnsShaper:
|
| 224 | """Builds and tears down one shaped pipe.
|
| 225 |
|
| 226 | Command construction is delegated to the pure functions above; this class
|
| 227 | only sequences execution. `run` and `available_fn` are injectable for
|
| 228 | tests — the default runner shells out and raises on failure (a half-built
|
| 229 | pipe must never be measured through).
|
| 230 | """
|
| 231 |
|
| 232 | def __init__(
|
| 233 | self,
|
| 234 | name: str = DEFAULT_NETNS_NAME,
|
| 235 | *,
|
| 236 | run: Optional[Callable[[list[str]], Any]] = None,
|
| 237 | available_fn: Callable[[], tuple[bool, Optional[str]]] = netshape_available,
|
| 238 | ) -> None:
|
| 239 | _interface_names(name) # validate eagerly
|
| 240 | self.name = name
|
| 241 | self._run = run if run is not None else self._default_run
|
| 242 | self._available_fn = available_fn
|
| 243 | self.profile: Optional[NetshapeProfile] = None
|
| 244 | self._created = False
|
| 245 |
|
| 246 | @staticmethod
|
| 247 | def _default_run(command: list[str]) -> None:
|
| 248 | subprocess.run(
|
| 249 | command,
|
| 250 | check=True,
|
| 251 | stdout=subprocess.DEVNULL,
|
| 252 | stderr=subprocess.PIPE,
|
| 253 | text=True,
|
| 254 | )
|
| 255 |
|
| 256 | def create(self, profile: NetshapeProfile) -> "NetnsShaper":
|
| 257 | ok, reason = self._available_fn()
|
| 258 | if not ok:
|
| 259 | raise RuntimeError(f"netshape_unavailable:{reason}")
|
| 260 | # Mark created BEFORE the first command runs: a mid-create failure
|
| 261 | # leaves a partially built namespace/veth on the host, and teardown()
|
| 262 | # (guarded by _created) must still clean it up. The except below makes
|
| 263 | # cleanup happen even when the caller never reaches its own teardown;
|
| 264 | # teardown is per-command best-effort, so the original error is what
|
| 265 | # propagates.
|
| 266 | self._created = True
|
| 267 | try:
|
| 268 | for command in netns_commands(self.name, profile):
|
| 269 | self._run(command)
|
| 270 | except BaseException:
|
| 271 | self.teardown()
|
| 272 | raise
|
| 273 | self.profile = profile
|
| 274 | return self
|
| 275 |
|
| 276 | def teardown(self) -> None:
|
| 277 | if not self._created:
|
| 278 | return
|
| 279 | for command in teardown_commands(self.name):
|
| 280 | try:
|
| 281 | self._run(command)
|
| 282 | except (OSError, subprocess.CalledProcessError):
|
| 283 | # Best-effort: a vanished link/namespace is already torn down.
|
| 284 | pass
|
| 285 | self._created = False
|
| 286 | self.profile = None
|
| 287 |
|
| 288 | def __enter__(self) -> "NetnsShaper":
|
| 289 | return self
|
| 290 |
|
| 291 | def __exit__(self, exc_type, exc, tb) -> None: # type: ignore[no-untyped-def]
|
| 292 | self.teardown()
|
| 293 |
|
| 294 | def tcp_ping(self, *, timeout_s: float = 30.0) -> float:
|
| 295 | """One TCP connect round trip through the shaped pipe, in milliseconds.
|
| 296 |
|
| 297 | A listener on the host end accepts; a tiny interpreter inside the
|
| 298 | namespace times the connect (the 3-way handshake costs one RTT).
|
| 299 | Process startup happens outside the timed window because the timing
|
| 300 | runs inside the child.
|
| 301 | """
|
| 302 | if not self._created:
|
| 303 | raise RuntimeError("tcp_ping requires a created shaper")
|
| 304 | with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener:
|
| 305 | listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
| 306 | listener.bind((HOST_ADDR, 0))
|
| 307 | listener.listen(1)
|
| 308 | listener.settimeout(timeout_s)
|
| 309 | port = listener.getsockname()[1]
|
| 310 | script = (
|
| 311 | "import socket, time\n"
|
| 312 | "s = socket.socket()\n"
|
| 313 | "t = time.perf_counter()\n"
|
| 314 | f"s.connect(({HOST_ADDR!r}, {port}))\n"
|
| 315 | "print((time.perf_counter() - t) * 1000)\n"
|
| 316 | "s.close()\n"
|
| 317 | )
|
| 318 | proc = subprocess.Popen(
|
| 319 | netns_exec(self.name, [sys.executable, "-c", script]),
|
| 320 | stdout=subprocess.PIPE,
|
| 321 | stderr=subprocess.PIPE,
|
| 322 | text=True,
|
| 323 | )
|
| 324 | try:
|
| 325 | conn, _addr = listener.accept()
|
| 326 | conn.close()
|
| 327 | finally:
|
| 328 | out, err = proc.communicate(timeout=timeout_s)
|
| 329 | if proc.returncode != 0:
|
| 330 | raise RuntimeError(f"tcp_ping failed inside netns: {err.strip()}")
|
| 331 | return float(out.strip())
|
| 332 |
|
| 333 | def self_test(
|
| 334 | self,
|
| 335 | profile: Optional[NetshapeProfile] = None,
|
| 336 | *,
|
| 337 | ping_fn: Optional[Callable[[], float]] = None,
|
| 338 | samples: int = 5,
|
| 339 | ) -> dict[str, Any]:
|
| 340 | target = profile if profile is not None else self.profile
|
| 341 | if target is None:
|
| 342 | raise ValueError("self_test needs a profile (none configured on this shaper)")
|
| 343 | return self_test(target, ping_fn=ping_fn, samples=samples, shaper=self)
|
| 344 |
|
| 345 |
|
| 346 | def self_test(
|
| 347 | profile: NetshapeProfile,
|
| 348 | *,
|
| 349 | ping_fn: Optional[Callable[[], float]] = None,
|
| 350 | samples: int = 5,
|
| 351 | shaper: Optional[NetnsShaper] = None,
|
| 352 | ) -> dict[str, Any]:
|
| 353 | """Measure achieved RTT through the shaped pipe and compare to configured.
|
| 354 |
|
| 355 | Returns a self-test row dict; the lane must refuse to run a campaign when
|
| 356 | within_tolerance is false. ping_fn (one RTT sample in ms per call) is
|
| 357 | injectable for tests; the default is the shaper's real TCP connect ping.
|
| 358 | The median of `samples` round trips smooths scheduler noise without hiding
|
| 359 | a misconfigured qdisc.
|
| 360 | """
|
| 361 | if ping_fn is None:
|
| 362 | if shaper is None:
|
| 363 | raise ValueError("self_test needs ping_fn or a shaper to ping through")
|
| 364 | ping_fn = shaper.tcp_ping
|
| 365 | samples_ms = [float(ping_fn()) for _ in range(max(1, samples))]
|
| 366 | measured = statistics.median(samples_ms)
|
| 367 | tolerance_ms = max(profile.rtt_ms * RTT_TOLERANCE_PCT / 100.0, RTT_TOLERANCE_FLOOR_MS)
|
| 368 | within = abs(measured - profile.rtt_ms) <= tolerance_ms
|
| 369 | return {
|
| 370 | "operation": "netshape.selftest",
|
| 371 | "netshape_profile": profile.name,
|
| 372 | "configured_rtt_ms": profile.rtt_ms,
|
| 373 | "measured_rtt_ms": round(measured, 3),
|
| 374 | "rtt_samples_ms": [round(sample, 3) for sample in samples_ms],
|
| 375 | "tolerance_pct": RTT_TOLERANCE_PCT,
|
| 376 | "tolerance_floor_ms": RTT_TOLERANCE_FLOOR_MS,
|
| 377 | "tolerance_ms": round(tolerance_ms, 3),
|
| 378 | "within_tolerance": within,
|
| 379 | }
|