From 897f50e4306ae892f0d19c6aa935ab7795bd2f50 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Fri, 2 Oct 2026 16:23:14 +0800 Subject: [PATCH] test(e2e): live coverage for the standard Docker node kernel datapath Two real Alpine containers on an isolated instance, addressed the way a GNS3 user does it (the node's persistent /etc/network/interfaces, applied by init.sh) and driven through the Docker daemon socket the server itself uses. Four scenarios: * fast path: the per-link kernel bridge enslaves exactly the two veth host ends, real ICMP crosses, the L2-anchor spec E.1/E.2 assertions hold (idle silence measured with the guests' links down -- a container's own boot-time IGMP report re-fires ~10 s later, so silencing is not just IPv6), netem delay, suspend/resume, capture, link delete/re-create, container stop/start (the veth host ends die with the adapter and are rebuilt from the NIO), zero residue on project delete; * filter matrix, one type at a time, each asserted through real traffic and its tc state: netem core (delay, jitter+distribution, loss, corrupt, duplicate), the extensions (reorder+correl+gap, seed+limit, gemodel, rate), cls_bpf match-drop including a byte-size discriminator, the eBPF classifier (frequency_drop every-Nth and -1, quota byte-cap cutoff, window_drop single outage with per-seq pass/drop/pass evidence), plus available_filters offering the kernel-only types; * relay control: no per-link bridge, the kernel-only types hidden; * relay->kernel reopen upgrade: the datapath choice is flipped across a server restart and the project reopened -- the link is rebuilt on the kernel datapath and forwards. The harness gains a Docker guest driver (a stdlib Engine API client over the daemon socket: exec with stream demux, container id by deterministic name -- the controller's node payload does not carry it, image inspect), a digest-pinned image helper (cache hit = no network, miss = pull through the server's own route, neither = skip) and Server.restart() for the reopen scenario (which also needs the local compute's capabilities to be up before a project open is accepted). Live: 4/4 scenarios, 142 s for the file; full e2e suite 10/10 (1503 s); unit suite 2149 passed; ruff clean. The matrix measures loss with 20 packets at 0.2 s spacing (143 s -> 58 s) and the jitter RTT assertion keeps statistical margin (per-direction N(50,20); 10 packets, >= 65 ms). --- tests/e2e/README.md | 27 +- tests/e2e/harness.py | 322 +++++++++++++- tests/e2e/test_docker_kernel_datapath.py | 544 +++++++++++++++++++++++ 3 files changed, 885 insertions(+), 8 deletions(-) create mode 100644 tests/e2e/test_docker_kernel_datapath.py diff --git a/tests/e2e/README.md b/tests/e2e/README.md index c5fcfc4b6..fd5a6f6ff 100644 --- a/tests/e2e/README.md +++ b/tests/e2e/README.md @@ -27,7 +27,8 @@ Two run modes, chosen per test through `harness.live_server(kernel=...)`: process on a free port with its own config directory (the controller DB and every path follow the config), so nothing on the machine is touched. It needs `ubridge`, `dynamips` and an images directory - (`GNS3_E2E_IMAGES`, default `~/GNS3/images`) available locally. + (`GNS3_E2E_IMAGES`, default `~/GNS3/images`) available locally; the + Docker scenarios additionally need the Docker daemon the server talks to. Either way each test creates its own project and deletes it again. Set `GNS3_E2E_KEEP=1` to keep the project (and a started instance) on failure @@ -43,9 +44,11 @@ to skip the detection and use the given value (detection failing is never fatal: the routers simply run without one). `harness.py` holds the shared pieces — REST client, server lifecycle, the -IOS console driver, host-side kernel inspections (bridge membership, tap -admin state, tc qdiscs, the L2-anchor spec §E assertions) — so a new node -type adds only its own scenario module. +IOS console driver, the Docker guest driver (a tiny Docker Engine API +client over the daemon socket, plus the digest-pinned image helper), +host-side kernel inspections (bridge membership, tap admin state, tc +qdiscs, the L2-anchor spec §E assertions) — so a new node type adds only +its own scenario module. Current scenarios: @@ -62,6 +65,22 @@ Current scenarios: TAP leg (anchors at start, netem delay, suspend, capture, link delete/re-create swap, node stop/start rewiring) and a relay negative control. +* `test_docker_kernel_datapath.py` — two real Alpine containers (the plain + veth datapath): the per-link kernel bridge enslaves exactly the two veth + host ends, real ICMP crosses, the filter matrix runs one type at a time + (netem core and extensions, cls_bpf match-drop, the eBPF classifier — + each measured through real traffic), plus a relay negative control and a + relay→kernel reopen upgrade (the server's datapath choice is flipped + across a restart). + +The Docker scenarios run a digest-pinned `alpine:3`, so every developer +tests the same bytes: a cached copy in the local Docker daemon costs no +network, a cache miss pulls through the server's own image-pull route, and +with neither cache nor registry the scenario skips (`GNS3_E2E_DOCKER_IMAGE` +overrides the reference). Their guests are addressed the way a GNS3 user +does it — the node's persistent `/etc/network/interfaces`, which the +container's `init.sh` applies at boot — and driven through the Docker +daemon socket the server itself uses. Every kernel scenario also asserts the L2-anchor spec's server-side guarantees (`docs/design/ubridge-l2-anchor-spec.md` §E): anchors and bridges diff --git a/tests/e2e/harness.py b/tests/e2e/harness.py index 07391a45a..b3872f0f5 100644 --- a/tests/e2e/harness.py +++ b/tests/e2e/harness.py @@ -42,11 +42,18 @@ Environment knobs: * ``GNS3_E2E_IDLEPC`` — use this idle-PC value for the dynamips routers instead of detecting one (see ``dynamips_idlepc``; without an idle-PC each router burns a full CPU core for the whole scenario). +* ``GNS3_E2E_DOCKER_IMAGE`` — the image the Docker scenarios run (default + ``alpine:3``, pinned by digest and pulled on a cache miss; see + ``ensure_docker_image``). +* ``GNS3_E2E_DOCKER_SOCKET`` — the Docker daemon socket the Docker guest + driver talks to (default ``/var/run/docker.sock`` — the same daemon the + server uses). * ``GNS3_E2E_KEEP=1`` — keep the project (and the isolated instance) when a test fails, and print their ids, for inspection. """ import contextlib +import http.client import json import os import re @@ -57,6 +64,7 @@ import sys import tempfile import time import urllib.error +import urllib.parse import urllib.request import pytest @@ -135,6 +143,20 @@ class Compute: {"compute_id": "local", "name": name, "node_type": "dynamips", "properties": properties}, ) + def create_docker_node(self, project_id, name, image, adapters=1): + """A standard Docker node (the veth datapath): the scenario addresses + its eth0 from the host and pings through it.""" + return self.call( + "POST", + f"/projects/{project_id}/nodes", + { + "compute_id": "local", + "name": name, + "node_type": "docker", + "properties": {"image": image, "adapters": adapters, "console_type": "telnet"}, + }, + ) + def create_iol_router(self, project_id, name, image, adapters=2): """An IOL runner container (iol-runner images), the appliance's template properties.""" @@ -244,7 +266,7 @@ class Server: with open(path) as f: return "".join(f.readlines()[-lines:]) - def stop(self): + def _stop_process(self): if self.process and self.process.poll() is None: self.process.terminate() try: @@ -252,11 +274,30 @@ class Server: except subprocess.TimeoutExpired: self.process.kill() self.process.wait(timeout=10) + + def stop(self): + self._stop_process() if self.tmpdir: shutil.rmtree(self.tmpdir, ignore_errors=True) + def restart(self, kernel=True): + """Restart an isolated instance in place: stop the process — the + temporary directory (config, controller DB, projects) survives — and + start a new one on the same paths with the given kernel-datapath + setting and a fresh port. This is how the reopen-upgrade scenario + flips the datapath choice and reopens a project created on the + relay.""" + if not self.isolated: + raise AssertionError("restart() needs an isolated instance (the harness did not start one)") + self._stop_process() + self.process, self.compute = _spawn_isolated(self.tmpdir, kernel) + return self -def _start_isolated(kernel): + +def _spawn_isolated(tmpdir, kernel): + """Start one isolated instance inside *tmpdir* (its config is written + there; a fresh free port each boot) and return ``(process, compute)``, + logged in. Shared by the first boot and by ``Server.restart``.""" ubridge = shutil.which("ubridge") dynamips = shutil.which("dynamips") if not ubridge: @@ -267,7 +308,6 @@ def _start_isolated(kernel): if not os.path.isdir(images): pytest.skip(f"images directory not found: {images} (set GNS3_E2E_IMAGES)") - tmpdir = tempfile.mkdtemp(prefix="gns3-e2e-") os.makedirs(os.path.join(tmpdir, "projects"), exist_ok=True) port = _free_port() config = ISOLATED_CONFIG.format( @@ -277,7 +317,7 @@ def _start_isolated(kernel): with open(config_path, "w") as f: f.write(config) - log = open(os.path.join(tmpdir, "server.log"), "w") + log = open(os.path.join(tmpdir, "server.log"), "a") # S603: fixed argv, this interpreter, no shell involvement argv = [sys.executable, "-m", "gns3server", "--config", config_path] if os.environ.get("GNS3_E2E_DEBUG"): @@ -307,6 +347,17 @@ def _start_isolated(kernel): raise AssertionError("isolated server did not come up within 60s") compute.login() + # The controller's connection to its own local compute comes up + # asynchronously; until it reports capabilities the compute counts as + # disconnected and a project open is refused. Same wait as live_server. + if not wait_until(lambda: (compute.capabilities() or {}).get("version"), timeout=45): + raise AssertionError(f"{compute.url}: the controller's local compute did not report capabilities within 45s") + return process, compute + + +def _start_isolated(kernel): + tmpdir = tempfile.mkdtemp(prefix="gns3-e2e-") + process, compute = _spawn_isolated(tmpdir, kernel) return Server(compute, process, tmpdir) @@ -566,6 +617,262 @@ def wait_ping(console, target, attempts=4, repeat=3, timeout=1): return result +# --------------------------------------------------------------------------- +# Docker guest driver (Docker Engine API over the daemon socket) +# --------------------------------------------------------------------------- + +# The e2e needs two things the server's REST surface does not expose: the +# identity of the image a container runs (to verify a cached image is the +# pinned build) and a way to run commands inside a booted container (address +# the guest, drive real ICMP from it). Both are plain Engine API calls over +# the same socket the server itself uses. + +DOCKER_SOCKET = os.environ.get("GNS3_E2E_DOCKER_SOCKET", "/var/run/docker.sock") + +# The image the Docker scenarios run. Both the registry digest (what to pull +# on a cache miss; a rolling tag resolved once and then pinned forever) and +# the repo-digest it must resolve to locally are pinned, so every developer +# tests the same bytes. The Docker daemon is the cache: a hit costs no +# network, a miss pulls through the server's own route (the WebUI's path), +# and no cache + no registry skips the scenario. +DOCKER_IMAGE = "alpine:3" +DOCKER_IMAGE_DIGEST = "sha256:28bd5fe8b56d1bd048e5babf5b10710ebe0bae67db86916198a6eec434943f8b" + + +class _UnixHTTPConnection(http.client.HTTPConnection): + """An HTTP connection speaking over a Unix socket (stdlib only).""" + + def __init__(self, socket_path, timeout): + super().__init__("localhost", timeout=timeout) + self._socket_path = socket_path + + def connect(self): + self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self.sock.settimeout(self.timeout) + self.sock.connect(self._socket_path) + + +class DockerDaemon: + """Minimal Docker Engine API client over the daemon socket.""" + + def __init__(self, socket_path=DOCKER_SOCKET, timeout=60): + self.socket_path = socket_path + self.timeout = timeout + + def _request(self, method, path, body=None, timeout=None): + conn = _UnixHTTPConnection(self.socket_path, timeout or self.timeout) + try: + data = None + headers = {} + if body is not None: + data = json.dumps(body).encode() + headers["Content-Type"] = "application/json" + conn.request(method, path, body=data, headers=headers) + response = conn.getresponse() + return response.status, response.read() + except (OSError, http.client.HTTPException) as e: + raise AssertionError(f"docker {method} {path} failed: {e}") from None + finally: + conn.close() + + def container_id(self, name): + """The id of the container called *name* (any state), or None. The + controller's node payload does not carry a Docker node's container + id (a compute-side read-only field); the daemon knows the container + by the deterministic name the server gives it (DockerVM.docker_name: + ``GNS3..``).""" + filters = urllib.parse.quote(json.dumps({"name": [name]}), safe="") + status, payload = self._request("GET", f"/containers/json?all=1&filters={filters}") + assert status == 200, f"docker container list -> HTTP {status}: {payload[:300]!r}" + for container in json.loads(payload): + if name in [entry.lstrip("/") for entry in container.get("Names") or []]: + return container["Id"] + return None + + def image_inspect(self, reference): + """The image record stored under *reference*, or None.""" + status, payload = self._request("GET", "/images/" + urllib.parse.quote(reference, safe="") + "/json") + if status == 404: + return None + assert status == 200, f"docker image inspect {reference} -> HTTP {status}: {payload[:300]!r}" + return json.loads(payload) + + def pull(self, reference, timeout=600): + """Pull *reference* (digest-pinned when the caller wants + reproducibility). Raises AssertionError on the daemon's error + stream — offline, unknown manifest, ...""" + status, payload = self._request( + "POST", "/images/create?fromImage=" + urllib.parse.quote(reference, safe=""), timeout=timeout + ) + if status != 200: + raise AssertionError(f"docker pull {reference} -> HTTP {status}: {payload[:300]!r}") + # the body is a stream of JSON progress objects; any object carrying + # an error key (in either shape) is a failure + text = payload.decode(errors="replace").strip() + decoder = json.JSONDecoder() + while text: + try: + obj, index = decoder.raw_decode(text) + except ValueError: + break + error = obj.get("error") or (obj.get("errorDetail") or {}).get("message") + if error: + raise AssertionError(f"docker pull {reference} failed: {error}") + text = text[index:].strip() + + def exec(self, container_id, cmd, timeout=60): + """Run *cmd* inside a container; returns (exit_code, combined + output).""" + status, payload = self._request( + "POST", + f"/containers/{container_id}/exec", + {"AttachStdout": True, "AttachStderr": True, "Tty": False, "Cmd": list(cmd)}, + ) + assert status == 201, f"docker exec create in {container_id[:12]} -> HTTP {status}: {payload[:300]!r}" + exec_id = json.loads(payload)["Id"] + try: + status, payload = self._request( + "POST", f"/exec/{exec_id}/start", {"Detach": False, "Tty": False}, timeout=timeout + ) + except AssertionError as e: + # A stalled stream is diagnosable: was the process still running + # (stream open, nothing flushed) or already gone? + raise AssertionError(f"{e} (exec {exec_id[:12]} still running: {self._exec_running(exec_id)})") from None + assert status == 200, f"docker exec start {exec_id[:12]} -> HTTP {status}: {payload[:300]!r}" + # the hijacked stream multiplexes stdout/stderr: 8-byte frame headers + # (stream id + big-endian length) — demultiplex into one buffer + output = bytearray() + index = 0 + while index + 8 <= len(payload): + length = int.from_bytes(payload[index + 4 : index + 8], "big") + output += payload[index + 8 : index + 8 + length] + index += 8 + length + status, payload = self._request("GET", f"/exec/{exec_id}/json") + assert status == 200, f"docker exec inspect {exec_id[:12]} -> HTTP {status}: {payload[:300]!r}" + return json.loads(payload)["ExitCode"], output.decode(errors="replace") + + def _exec_running(self, exec_id): + """Whether an exec is still running (diagnostic only, never raises).""" + try: + status, payload = self._request("GET", f"/exec/{exec_id}/json", timeout=10) + return json.loads(payload).get("Running") if status == 200 else f"HTTP {status}" + except (AssertionError, ValueError): + return "unknown" + + +def ensure_docker_image(server): + """The Docker image the scenarios run, ensured present locally. + + The pinned reference (``DOCKER_IMAGE``, digest ``DOCKER_IMAGE_DIGEST``) + is pulled through the server's own pull route on a cache miss and + verified against the pinned repo digest afterwards. A cached image that + resolves to another digest is used with a warning if the pull fails + (offline developer), but with no cached image and no registry the + scenario skips. ``GNS3_E2E_DOCKER_IMAGE`` overrides the reference + entirely (the operator then owns its content).""" + + override = os.environ.get("GNS3_E2E_DOCKER_IMAGE") + daemon = DockerDaemon() + if override: + info = daemon.image_inspect(override) + if info is None: + pytest.skip(f"GNS3_E2E_DOCKER_IMAGE={override} is not present on the Docker daemon") + print(f".. using GNS3_E2E_DOCKER_IMAGE={override} ({info['Id'][:19]}…)") + return override + + reference = f"{DOCKER_IMAGE}@{DOCKER_IMAGE_DIGEST}" + info = daemon.image_inspect(DOCKER_IMAGE) + if info is not None and any(DOCKER_IMAGE_DIGEST in d for d in (info.get("RepoDigests") or [])): + print(f".. Docker image {DOCKER_IMAGE} is cached at the pinned digest ({info['Id'][:19]}…)") + return DOCKER_IMAGE + if info is None: + print(f".. Docker image {DOCKER_IMAGE} is not cached — pulling the pinned digest") + else: + print( + f".. cached {DOCKER_IMAGE} is {info.get('RepoDigests') or info['Id'][:19]}…, not the pinned digest — pulling" + ) + try: + server.compute.call("POST", "/computes/local/docker/images/pull", {"image": reference}, timeout=600) + except AssertionError as e: + if info is None: + pytest.skip(f"Docker image {DOCKER_IMAGE} is not cached and could not be pulled: {e}") + print(f".. pull failed ({e}); using the cached {DOCKER_IMAGE} — not the pinned build") + return DOCKER_IMAGE + print(f".. pulled {reference}") + return DOCKER_IMAGE + + +def configure_docker_interfaces(node, address, netmask="255.255.255.0", adapter=0): + """Give the container's eth{adapter} a static address the way a GNS3 + user does: write the persistent ``/etc/network/interfaces`` the server + bind-mounts into the container and init.sh applies with busybox ifup at + boot (CIDR is not understood — separate address/netmask lines). Call + before starting the node.""" + directory = node.get("node_directory") + if not directory or not os.path.isdir(directory): + pytest.skip(f"cannot reach the node directory {directory!r} — this scenario needs the compute on this host") + path = os.path.join(directory, "etc", "network", "interfaces") + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w") as f: + f.write(f"auto eth{adapter}\niface eth{adapter} inet static\n\taddress {address}\n\tnetmask {netmask}\n") + + +def docker_wait_address(daemon, container_id, address, timeout=30): + """Wait until the container's eth0 carries *address* — init.sh applies + the interfaces file (busybox ifup) a moment after container start.""" + + def present(): + _code, output = daemon.exec(container_id, ["ip", "-4", "-o", "addr", "show", "dev", "eth0"]) + return address in output + + if not wait_until(present, timeout=timeout, interval=0.5): + raise AssertionError(f"container {container_id[:12]} did not configure {address} within {timeout}s") + + +_BUSYBOX_PING_LOSS_RE = re.compile(r"(\d+)% packet loss") +_BUSYBOX_PING_RTT_RE = re.compile(r"round-trip min/avg/max = [\d.]+/([\d.]+)/[\d.]+ ms") + + +def docker_ping(daemon, container_id, target, count=3, timeout=1, interval=0.2, size=None): + """Run a real busybox ping inside a container; returns + ``{loss, avg, raw, exit}`` (loss in percent, avg in ms). The default + 0.2 s interval packs more samples into loss measurements; loss-based + checks pass a higher count. + + The ping is wrapped in busybox `timeout` with a generous cap so a + misbehaving guest can never hold the exec stream open past it — the + harness then reports a normal failed ping instead of a socket timeout. + """ + cap = int(count * interval + timeout + 15) + cmd = ["timeout", str(cap), "ping", "-c", str(count), "-W", str(timeout)] + if interval is not None: + cmd += ["-i", str(interval)] + if size is not None: + cmd += ["-s", str(size)] + cmd.append(target) + code, output = daemon.exec(container_id, cmd, timeout=cap + 10) + loss = _BUSYBOX_PING_LOSS_RE.search(output) + rtt = _BUSYBOX_PING_RTT_RE.search(output) + return { + "loss": int(loss.group(1)) if loss else 100, + "avg": float(rtt.group(1)) if rtt else None, + "raw": output, + "exit": code, + } + + +def docker_wait_ping(daemon, container_id, target, attempts=4, **kwargs): + """ping until it fully succeeds (a fresh veth/bridge needs a beat for + carrier and ARP); returns the last result either way.""" + result = None + for _ in range(attempts): + result = docker_ping(daemon, container_id, target, **kwargs) + if result["loss"] == 0: + return result + time.sleep(1.5) + return result + + # --------------------------------------------------------------------------- # Host-side inspection (same machine as the compute) # --------------------------------------------------------------------------- @@ -737,6 +1044,13 @@ def anchor_name(node_id, adapter, port): return "gd" + node_id.replace("-", "")[:8] + f"e{adapter}p{port}" +def docker_anchor_name(node_id, adapter, port=0): + """The standard Docker node's anchor: the veth host end the server keeps + in the root namespace (see DockerKernelDatapathMixin._veth_names and + utils.kernel_anchor).""" + return "gv" + node_id.replace("-", "")[:8] + f"e{adapter}p{port}" + + def iol_anchor_name(node_id, adapter, port): """The IOL runner container's anchor TAP name (see IOLDockerVM._tap_name).""" return "gx" + node_id.replace("-", "")[:8] + f"e{adapter}p{port}" diff --git a/tests/e2e/test_docker_kernel_datapath.py b/tests/e2e/test_docker_kernel_datapath.py new file mode 100644 index 000000000..a56101a91 --- /dev/null +++ b/tests/e2e/test_docker_kernel_datapath.py @@ -0,0 +1,544 @@ +# +# Copyright (C) 2026 GNS3 Technologies Inc. +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +""" +Live end-to-end tests for the standard Docker node's kernel datapath (pytest +marker ``e2e``): two real Alpine containers, addressed from the host the way +a GNS3 user does (the node's persistent ``/etc/network/interfaces``, applied +by the container's init.sh) and driven through the Docker daemon socket the +server itself uses. The scenarios cover: + +* ``test_docker_kernel_fast_path`` — the veth datapath itself: the link + reports ``kernel_datapath``, the per-link kernel bridge enslaves exactly + the two veth host ends, real ICMP crosses kernel-only, the L2-anchor spec + §E assertions hold (no L3 identity, FORWARDING ports, idle silence), a + ``delay 100`` filter lands as netem on both host ends, suspend kills the + traffic at the carrier, capture writes a real pcap, link delete/re-create + and container stop/start leave the wiring consistent, and deleting the + project leaves no veth or bridge behind. +* ``test_docker_kernel_filter_matrix`` — one filter type at a time on a live + kernel link, each asserted by its effect on real traffic (RTT, loss, + per-seq outage evidence) and its tc state, from netem core and the netem + extensions to the cls_bpf and eBPF classifiers; the kernel-only types are + offered by ``available_filters`` and everything restores cleanly. +* ``test_docker_relay_control`` — the negative control on an isolated + relay-configured instance: no per-link kernel bridge, the wire is the + uBridge relay, and the kernel-only filter types are hidden. +* ``test_docker_relay_to_kernel_reopen_upgrade`` — the server's datapath + choice is flipped (relay → kernel) across a restart and the project is + reopened: the link is rebuilt on the kernel datapath and forwards. + +The image is pinned by digest (``harness.ensure_docker_image``): a cached +copy costs no network, a miss pulls through the server's own pull route, +and no cache + no registry skips the scenario. +""" + +import os +import re +import time + +import pytest + +from tests.e2e import harness + +pytestmark = pytest.mark.e2e + +# every standard Docker adapter has a single port +ETH = (0, 0) +R1_IP = "10.1.1.1" +R2_IP = "10.1.1.2" + + +def _resolve_container(daemon, node): + """Fill in the node's container id from the daemon (the controller's + node payload does not carry the compute-side id), by the deterministic + name the server gives the container.""" + container_id = daemon.container_id(f"GNS3.{node['name']}.{node['project_id']}") + assert container_id, f"container for {node['name']} not found on the Docker daemon" + node["container_id"] = container_id + return node + + +def _create_topology(server, pid, image, name): + """Two containers with addressed eth0s. The harness reaches their guests + through the Docker daemon socket — the same socket the server uses.""" + compute = server.compute + daemon = harness.DockerDaemon() + nodes = [] + for suffix, address in (("1", R1_IP), ("2", R2_IP)): + node = compute.create_docker_node(pid, f"{name}-{suffix}", image) + harness.configure_docker_interfaces(node, address) + nodes.append(_resolve_container(daemon, node)) + return nodes[0], nodes[1], daemon + + +def _start_addresses(compute, pid, daemon, nodes, addresses): + """Start the containers and wait for init.sh to have applied their + addresses (busybox ifup runs a moment after container start).""" + for node in nodes: + compute.call("POST", f"/projects/{pid}/nodes/{node['node_id']}/start") + for node, address in zip(nodes, addresses, strict=False): + harness.docker_wait_address(daemon, node["container_id"], address) + + +def _put_filters(compute, pid, lid, filters): + compute.call("PUT", f"/projects/{pid}/links/{lid}", {"filters": filters}) + + +def _clear_filters(compute, pid, lid, *anchors): + """Clear the link's filters and assert the teardown reached the kernel: + no netem qdisc and no clsact left on the anchors.""" + _put_filters(compute, pid, lid, {}) + for anchor in anchors: + assert "netem" not in harness.qdiscs(anchor), harness.qdiscs(anchor) + assert "clsact" not in harness.qdiscs(anchor), harness.qdiscs(anchor) + + +def _filter_types(compute, pid, lid): + return {entry["type"] for entry in compute.call("GET", f"/projects/{pid}/links/{lid}/available_filters")} + + +def _silence_guest(daemon, container_id): + """Silence the guest for the idle-silence measurement — the docker + analogue of shutting the IOS interfaces in the dynamips scenario. The + link goes down first (a guest's own chatter is not just IPv6: Linux + re-sends its boot-time IGMP membership report seconds later), then its + IPv6 stack is switched off so the re-up does not re-fire DAD/MLD. The + IPv4 address stays configured and answers once the link is back.""" + code, output = daemon.exec( + container_id, + ["sh", "-c", "ip link set eth0 down && echo 1 > /proc/sys/net/ipv6/conf/eth0/disable_ipv6"], + ) + assert code == 0, output + + +def _unsilence_guest(daemon, container_id): + code, output = daemon.exec(container_id, ["ip", "link", "set", "eth0", "up"]) + assert code == 0, output + + +def test_docker_kernel_fast_path(): + server = harness.live_server(kernel=True) + compute = server.compute + image = harness.ensure_docker_image(server) + + project = compute.create_project("docker-e2e") + pid = project["project_id"] + a1 = a2 = bridge = None + try: + n1, n2, daemon = _create_topology(server, pid, image, "E2E") + n1_id, n2_id = n1["node_id"], n2["node_id"] + c1, c2 = n1["container_id"], n2["container_id"] + a1 = harness.docker_anchor_name(n1_id, *ETH) + a2 = harness.docker_anchor_name(n2_id, *ETH) + + # The link is created while both containers are stopped: the kernel + # NIO is bound with no anchors in existence yet and attached by + # container start (the deferred-wiring path). + link = compute.create_link(pid, (n1_id, *ETH), (n2_id, *ETH)) + lid = link["link_id"] + bridge = harness.link_bridge_name(lid) + assert link["kernel_datapath"] is True, link + assert not harness.tap_exists(a1), "the veth pair is born at container start" + + print(".. starting containers") + _start_addresses(compute, pid, daemon, (n1, n2), (R1_IP, R2_IP)) + + # Kernel objects: the veth host ends exist and the per-link bridge + # enslaves exactly the two of them. + assert harness.tap_exists(a1) and harness.tap_exists(a2), (a1, a2) + assert harness.bridge_members(bridge) == sorted([a1, a2]), harness.bridge_members(bridge) + + # L2-anchor spec §E.1 (anchors and bridge carry no L3 identity; + # skipped on a uBridge without `link l2only`) and the forwarding + # guard — a DOWN bridge silently forwards nothing. + if harness.l2only_supported(): + harness.assert_pure_l2(a1) + harness.assert_pure_l2(a2) + harness.assert_pure_l2(bridge) + else: + print(".. uBridge without link l2only: skipping the §E.1 assertions") + harness.assert_forwarding(a1) + harness.assert_forwarding(a2) + + print(".. pinging between the containers over the kernel link") + baseline = harness.docker_wait_ping(daemon, c1, R2_IP) + assert baseline["loss"] == 0, baseline["raw"] + assert baseline["avg"] is not None and baseline["avg"] < 50, baseline + + # §E.2: with the guests silenced the segment is silent for 5 s — + # the host anchors do not generate the MLD/DAD flood an unhardened + # device produces (baseline: 6 frames / 2 s). + if harness.l2only_supported(): + _silence_guest(daemon, c1) + _silence_guest(daemon, c2) + harness.assert_idle_silence(a1, a2, bridge) + _unsilence_guest(daemon, c1) + _unsilence_guest(daemon, c2) + assert harness.docker_wait_ping(daemon, c1, R2_IP, attempts=5)["loss"] == 0 + + # delay 100: netem on both host ends (each impairs one direction), + # one-way ~100 ms => RTT grows by ~200 ms + _put_filters(compute, pid, lid, {"delay": [100]}) + assert "netem" in harness.qdiscs(a1) and "netem" in harness.qdiscs(a2), ( + harness.qdiscs(a1), + harness.qdiscs(a2), + ) + delayed = harness.docker_wait_ping(daemon, c1, R2_IP) + assert delayed["loss"] == 0 and delayed["avg"] >= 150, (baseline, delayed) + _put_filters(compute, pid, lid, {}) + assert "netem" not in harness.qdiscs(a1), harness.qdiscs(a1) + fast = harness.docker_wait_ping(daemon, c1, R2_IP) + assert fast["loss"] == 0 and fast["avg"] < 50, fast + + # suspend: the host end admin-downs (carrier loss into the + # container), the link is dead; resume restores + compute.call("PUT", f"/projects/{pid}/links/{lid}", {"suspend": True}) + assert not harness.tap_up(a1) + dead = harness.docker_ping(daemon, c1, R2_IP, count=3, timeout=1) + assert dead["loss"] == 100, dead["raw"] + compute.call("PUT", f"/projects/{pid}/links/{lid}", {"suspend": False}) + assert harness.tap_up(a1) + assert harness.docker_wait_ping(daemon, c1, R2_IP)["loss"] == 0 + + # capture: AF_PACKET on the anchor writes a real pcap + capture = compute.call("POST", f"/projects/{pid}/links/{lid}/capture/start", {"data_link_type": "DLT_EN10MB"}) + harness.docker_ping(daemon, c1, R2_IP) + time.sleep(1) + compute.call("POST", f"/projects/{pid}/links/{lid}/capture/stop") + path = capture["capture_file_path"] + count, ethertypes = harness.pcap_records(path) + assert count >= 4, f"{path}: {count} records" + assert "0800" in ethertypes, (path, ethertypes) + + # link delete detaches both ends (the veths survive — they are the + # adapters); re-creating enslaves them again and the traffic returns + compute.call("DELETE", f"/projects/{pid}/links/{lid}") + assert harness.bridge_members(bridge) is None + assert harness.tap_exists(a1) and harness.tap_exists(a2) + link = compute.create_link(pid, (n1_id, *ETH), (n2_id, *ETH)) + lid = link["link_id"] + bridge = harness.link_bridge_name(lid) + assert link["kernel_datapath"] is True, link + assert harness.docker_wait_ping(daemon, c1, R2_IP, attempts=5)["loss"] == 0 + + # container stop: the veth host ends are deleted explicitly (a veth + # outlives its container's netns) and the bridge goes with them; a + # restart rebuilds the whole wiring from the NIO + print(".. stopping and restarting container 1") + compute.call("POST", f"/projects/{pid}/nodes/{n1_id}/stop") + assert harness.wait_until(lambda: not harness.tap_exists(a1), timeout=15), a1 + assert harness.bridge_members(bridge) in (None, [a2]), harness.bridge_members(bridge) + compute.call("POST", f"/projects/{pid}/nodes/{n1_id}/start") + harness.docker_wait_address(daemon, c1, R1_IP) + assert harness.tap_exists(a1) + assert harness.wait_until(lambda: harness.bridge_members(bridge) == sorted([a1, a2]), timeout=15) + assert harness.docker_wait_ping(daemon, c1, R2_IP, attempts=5)["loss"] == 0 + except BaseException: + harness.release(server, pid, failed=True) + raise + harness.release(server, pid, failed=False) + + # deleting the project closed the containers: veths and bridges go + assert harness.wait_until(lambda: not harness.tap_exists(a1) and not harness.tap_exists(a2), timeout=15), ( + a1, + a2, + ) + assert harness.wait_until(lambda: harness.bridge_members(bridge) is None, timeout=10), bridge + + +def test_docker_kernel_filter_matrix(): + """One filter type at a time on a live kernel link, each asserted by its + effect on real traffic and its tc state. Types the uBridge on this + compute cannot serve (bpf without cbpf, the eBPF modes without ebpf) are + skipped rather than failed — the capability chain is ``available_filters``' + contract, and a compute reporting nothing keeps the full list.""" + server = harness.live_server(kernel=True) + compute = server.compute + caps = (compute.capabilities() or {}).get("ubridge_tc") or {} + image = harness.ensure_docker_image(server) + + project = compute.create_project("docker-filter-e2e") + pid = project["project_id"] + try: + n1, n2, daemon = _create_topology(server, pid, image, "E2E") + n1_id, n2_id = n1["node_id"], n2["node_id"] + c1 = n1["container_id"] + a1 = harness.docker_anchor_name(n1_id, *ETH) + a2 = harness.docker_anchor_name(n2_id, *ETH) + + link = compute.create_link(pid, (n1_id, *ETH), (n2_id, *ETH)) + lid = link["link_id"] + assert link["kernel_datapath"] is True, link + + _start_addresses(compute, pid, daemon, (n1, n2), (R1_IP, R2_IP)) + baseline = harness.docker_wait_ping(daemon, c1, R2_IP) + assert baseline["loss"] == 0, baseline["raw"] + + # The kernel link offers the kernel-only types the compute's uBridge + # reports it can run (all of them on this uBridge). + offered = _filter_types(compute, pid, lid) + assert {"delay", "packet_loss", "corrupt", "duplicate", "bpf"} <= offered, offered + if not {"rate", "reorder", "gemodel", "seed", "limit"} & offered: + pytest.skip(f"this compute's uBridge reports no netem extensions: {sorted(offered)}") + + # -- netem core ---------------------------------------------------- + print(".. delay 100") + _put_filters(compute, pid, lid, {"delay": [100]}) + assert "netem" in harness.qdiscs(a1) and "netem" in harness.qdiscs(a2), ( + harness.qdiscs(a1), + harness.qdiscs(a2), + ) + delayed = harness.docker_ping(daemon, c1, R2_IP) + print(f".. {delayed['avg']} ms RTT") + assert delayed["loss"] == 0 and delayed["avg"] >= 150, (baseline, delayed) + + print(".. delay 50 jitter 20 distribution normal") + _put_filters(compute, pid, lid, {"delay": [50, 20, "normal"]}) + # the jitter parameter lands verbatim...; the sampled RTT mean is a + # random variable (per-direction ~N(50, 20)), so only the elevation + # is asserted, with margin — 10 packets put the sample mean's sigma + # around 9 ms + assert re.search(r"delay 50ms\s+20ms", harness.qdiscs(a1)), harness.qdiscs(a1) + jittered = harness.docker_ping(daemon, c1, R2_IP, count=10) + print(f".. {jittered['avg']} ms RTT") + assert jittered["loss"] == 0 and jittered["avg"] >= 65, jittered + _clear_filters(compute, pid, lid, a1, a2) + + print(".. packet_loss 30") + _put_filters(compute, pid, lid, {"packet_loss": [30]}) + lossy = harness.docker_ping(daemon, c1, R2_IP, count=20) + print(f".. {lossy['loss']} % round-trip loss") + # round trip = 1 - 0.7^2 ~ 51 %; the stochastic draw must land + # somewhere between "clearly lossy" and "not a black hole" + assert 20 <= lossy["loss"] <= 80, lossy["raw"] + _clear_filters(compute, pid, lid, a1, a2) + + print(".. corrupt 30") + _put_filters(compute, pid, lid, {"corrupt": [30]}) + assert "corrupt 30%" in harness.qdiscs(a1), harness.qdiscs(a1) + # a corrupted frame fails its checksum and is dropped: the same + # round-trip arithmetic as packet_loss (1 - 0.7^2) + corrupt = harness.docker_ping(daemon, c1, R2_IP, count=20) + print(f".. {corrupt['loss']} % round-trip loss") + assert 20 <= corrupt["loss"] <= 80, corrupt["raw"] + _clear_filters(compute, pid, lid, a1, a2) + + print(".. duplicate 50") + _put_filters(compute, pid, lid, {"duplicate": [50]}) + assert "dup" in harness.qdiscs(a1), harness.qdiscs(a1) + # duplicated frames are absorbed by the stack: no loss, traffic flows + dups = harness.docker_ping(daemon, c1, R2_IP, count=5) + assert dups["loss"] == 0, dups["raw"] + _clear_filters(compute, pid, lid, a1, a2) + + # -- netem extensions (kernel-only) -------------------------------- + print(".. delay 50 + reorder 25 correl 50 gap 5") + _put_filters(compute, pid, lid, {"delay": [50], "reorder": [25, 50, 5]}) + assert "reorder" in harness.qdiscs(a1), harness.qdiscs(a1) + reordered = harness.docker_ping(daemon, c1, R2_IP, count=10) + assert reordered["loss"] <= 60, reordered["raw"] + _clear_filters(compute, pid, lid, a1, a2) + + print(".. delay 50 + seed 42 + limit 5000") + _put_filters(compute, pid, lid, {"delay": [50], "seed": [42], "limit": [5000]}) + assert "limit 5000" in harness.qdiscs(a1), harness.qdiscs(a1) + seeded = harness.docker_ping(daemon, c1, R2_IP) + assert seeded["loss"] == 0 and seeded["avg"] >= 80, seeded + _clear_filters(compute, pid, lid, a1, a2) + + print(".. gemodel 100/0/30 (steady-state bad, 30 % loss)") + _put_filters(compute, pid, lid, {"gemodel": [100, 0, 30]}) + assert "gemodel" in harness.qdiscs(a1), harness.qdiscs(a1) + gemodel = harness.docker_ping(daemon, c1, R2_IP, count=20) + print(f".. {gemodel['loss']} % round-trip loss") + assert 20 <= gemodel["loss"] <= 90, gemodel["raw"] + _clear_filters(compute, pid, lid, a1, a2) + + print(".. rate 512kbit (1400-byte pings)") + _put_filters(compute, pid, lid, {"rate": ["512kbit"]}) + assert "rate" in harness.qdiscs(a1).lower(), harness.qdiscs(a1) + # 1400+28 bytes at 512 kbit ≈ 22 ms serialization per direction + rated = harness.docker_ping(daemon, c1, R2_IP, size=1400) + print(f".. {rated['avg']} ms RTT") + assert rated["loss"] == 0 and rated["avg"] >= 30, (baseline, rated) + _clear_filters(compute, pid, lid, a1, a2) + + # -- cls_bpf match-drop -------------------------------------------- + if caps.get("cbpf"): + print(".. bpf 'icmp' drops everything") + _put_filters(compute, pid, lid, {"bpf": ["icmp"]}) + assert "clsact" in harness.qdiscs(a1), harness.qdiscs(a1) + dropped = harness.docker_ping(daemon, c1, R2_IP, count=4) + assert dropped["loss"] == 100, dropped["raw"] + + print(".. bpf 'greater 150' discriminates by frame size") + _put_filters(compute, pid, lid, {"bpf": ["greater 150"]}) + small = harness.docker_ping(daemon, c1, R2_IP, count=4) + assert small["loss"] == 0, small["raw"] + big = harness.docker_ping(daemon, c1, R2_IP, count=4, size=300) + assert big["loss"] == 100, big["raw"] + _clear_filters(compute, pid, lid, a1, a2) + else: + print(".. uBridge reports no cbpf: skipping the bpf filter checks") + + # -- eBPF stateful classifier -------------------------------------- + modes = caps.get("ebpf_modes") or [] + if "nth" in modes: + print(".. frequency_drop 3 (every 3rd packet per direction)") + _put_filters(compute, pid, lid, {"frequency_drop": [3]}) + nth = harness.docker_ping(daemon, c1, R2_IP, count=20) + print(f".. {nth['loss']} % round-trip loss") + # kernel counts per direction: 1 - (2/3)^2 = 55.6 % round trip + assert 30 <= nth["loss"] <= 80, nth["raw"] + _put_filters(compute, pid, lid, {"frequency_drop": [-1]}) + all_dropped = harness.docker_ping(daemon, c1, R2_IP, count=4) + assert all_dropped["loss"] == 100, all_dropped["raw"] + _clear_filters(compute, pid, lid, a1, a2) + if "quota" in modes: + print(".. quota 3000 bytes hard cutoff") + _put_filters(compute, pid, lid, {"quota": [3000, 100]}) + # 1400-byte pings: the per-direction byte caps run out after + # ~2 packets, then nothing passes until the quota is reset + cut = harness.docker_ping(daemon, c1, R2_IP, count=5, size=1400) + assert cut["loss"] >= 40, cut["raw"] + after = harness.docker_ping(daemon, c1, R2_IP, count=4) + assert after["loss"] == 100, after["raw"] + _clear_filters(compute, pid, lid, a1, a2) + assert harness.docker_ping(daemon, c1, R2_IP, count=4)["loss"] == 0 + if "window" in modes: + print(".. window_drop [1000, 2000, 100] — one outage with passes on both sides") + _put_filters(compute, pid, lid, {"window_drop": [1000, 2000, 100]}) + # 25 pings at 5/s: seqs before t=1 s and after t=3 s answer, the + # middle is gone — per-sequence evidence of a bounded outage + code, output = daemon.exec(c1, ["ping", "-c", "25", "-i", "0.2", "-W", "1", R2_IP], timeout=30) + replied = {int(m.group(1)) for m in re.finditer(r"seq=(\d+)", output)} + missing = set(range(25)) - replied + assert 0 in replied and 24 in replied, (sorted(replied), output) + assert missing and max(missing) - min(missing) >= 5, (sorted(replied), output) + _clear_filters(compute, pid, lid, a1, a2) + assert harness.docker_ping(daemon, c1, R2_IP, count=4)["loss"] == 0 + except BaseException: + harness.release(server, pid, failed=True) + raise + harness.release(server, pid, failed=False) + + assert harness.wait_until(lambda: not harness.tap_exists(a1) and not harness.tap_exists(a2), timeout=15), ( + a1, + a2, + ) + + +def test_docker_relay_control(): + """Negative control: the same topology on an isolated relay-configured + instance wires nothing into the kernel — no per-link bridge, no + enslavement — and still pings over the uBridge UDP relay.""" + server = harness.live_server(kernel=False) + compute = server.compute + image = harness.ensure_docker_image(server) + + project = compute.create_project("docker-relay-e2e") + pid = project["project_id"] + a1 = None + try: + n1, n2, daemon = _create_topology(server, pid, image, "E2E") + n1_id, n2_id = n1["node_id"], n2["node_id"] + a1 = harness.docker_anchor_name(n1_id, *ETH) + + link = compute.create_link(pid, (n1_id, *ETH), (n2_id, *ETH)) + assert link["kernel_datapath"] is False, link + + _start_addresses(compute, pid, daemon, (n1, n2), (R1_IP, R2_IP)) + + # No kernel wiring: no per-link bridge, the veth host end is just a + # relay endpoint (up, pure L2 — the hardening is a creation-time + # property, not a datapath one) + assert harness.bridge_members(harness.link_bridge_name(link["link_id"])) is None + if harness.tap_exists(a1): + assert harness.tap_up(a1), a1 + if harness.l2only_supported(): + harness.assert_pure_l2(a1) + + relay_ping = harness.docker_wait_ping(daemon, n1["container_id"], R2_IP) + assert relay_ping["loss"] == 0, relay_ping["raw"] + + # available_filters hides the kernel-only types on the relay + offered = _filter_types(compute, pid, link["link_id"]) + assert "delay" in offered and "frequency_drop" in offered, offered + assert not (offered & {"rate", "reorder", "gemodel", "duplicate", "seed", "limit", "quota", "window_drop"}), ( + offered + ) + except BaseException: + harness.release(server, pid, failed=True) + raise + harness.release(server, pid, failed=False) + + assert harness.wait_until(lambda: not harness.tap_exists(a1), timeout=15), a1 + + +def test_docker_relay_to_kernel_reopen_upgrade(): + """The server's datapath choice is a live decision: with the project + created on a relay instance, flipping ``enable_kernel_datapath`` across + a server restart and reopening the project rebuilds the link on the + kernel datapath (the upgrade path a deployment takes) and it forwards.""" + if os.environ.get("GNS3_E2E_URL"): + pytest.skip("this scenario restarts the server — it needs an isolated instance") + server = harness.live_server(kernel=False) + compute = server.compute + image = harness.ensure_docker_image(server) + + project = compute.create_project("docker-upgrade-e2e") + pid = project["project_id"] + a1 = a2 = bridge = None + try: + n1, n2, daemon = _create_topology(server, pid, image, "E2E") + n1_id, n2_id = n1["node_id"], n2["node_id"] + a1 = harness.docker_anchor_name(n1_id, *ETH) + a2 = harness.docker_anchor_name(n2_id, *ETH) + + link = compute.create_link(pid, (n1_id, *ETH), (n2_id, *ETH)) + assert link["kernel_datapath"] is False, link + bridge = harness.link_bridge_name(link["link_id"]) + + _start_addresses(compute, pid, daemon, (n1, n2), (R1_IP, R2_IP)) + assert harness.bridge_members(bridge) is None + assert harness.docker_wait_ping(daemon, n1["container_id"], R2_IP)["loss"] == 0 + + # Flip the datapath choice and reopen the project: every link + # re-runs _prepare, which re-evaluates eligibility live. + print(".. restarting the server with enable_kernel_datapath=True") + server.restart(kernel=True) + compute = server.compute + compute.call("POST", f"/projects/{pid}/open") + n1 = _resolve_container(daemon, compute.node(pid, n1_id)) + n2 = _resolve_container(daemon, compute.node(pid, n2_id)) + link = compute.call("GET", f"/projects/{pid}/links/{link['link_id']}") + assert link["kernel_datapath"] is True, link + assert not harness.tap_exists(a1), "containers are stopped after the reopen" + + _start_addresses(compute, pid, daemon, (n1, n2), (R1_IP, R2_IP)) + assert harness.wait_until(lambda: harness.bridge_members(bridge) == sorted([a1, a2]), timeout=15) + assert harness.docker_wait_ping(daemon, n1["container_id"], R2_IP, attempts=5)["loss"] == 0 + except BaseException: + harness.release(server, pid, failed=True) + raise + harness.release(server, pid, failed=False) + + assert harness.wait_until(lambda: not harness.tap_exists(a1) and not harness.tap_exists(a2), timeout=15), ( + a1, + a2, + ) + assert harness.wait_until(lambda: harness.bridge_members(bridge) is None, timeout=10), bridge