test(e2e): assert the L2-anchor spec's server-side guarantees

The spec's gns3-server alignment section promised three assertions the
suites never carried; they land in harness.py and run in every kernel
scenario (plus the relay controls' anchor checks):

- E.1 assert_pure_l2: anchors and bridges carry no L3 identity (addrgenmode
  none, no address in /proc/net/if_inet6, no ip -o -4 addr) -- the kernel
  scenarios' anchors and per-link/switch bridges, relay-mode anchors and
  the switch's relay TAPs alike (hardening is a creation-time property,
  not a datapath one);
- E.2 assert_idle_silence (Dynamips suite): guest interfaces shut to
  silence IOS's own CDP/keepalives, settle 2.5 s, then a 5 s window of
  zero traffic over both anchors and the per-link bridge;
- assert_forwarding: every attached port reports brport/state == 3 -- a
  bridge left DOWN keeps its ports disabled and forwards nothing, with no
  error anywhere; the assertion is the only signal.

harness.l2only_supported() probes a throwaway uBridge once (link l2only
against a nonexistent device: 208 = command present, 202-Unknown command
= old build); without the command the assertions skip -- the spec's
degradation contract, mirroring the tc-capability skip.

The first live run caught a fifth-creator gap: the switch's relay port
TAPs (minted implicitly by bridge add_nio_tap) carried a live fe80::.
Delivered uBridge-side per the spec (fb72758: the transient TAP is
hardened at creation, attaches untouched); the server needed no change,
and the relay-control E.1 check is that creator's end-to-end coverage.

tests/e2e/README.md

@@ -35,8 +35,8 @@ and print where to look at it.
 
 `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) — so a new node type adds only its own scenario
+admin state, tc qdiscs, the L2-anchor spec §E assertions) — so a new node
-module.
+type adds only its own scenario module.
 
 Current scenarios:
 
@@ -54,5 +54,12 @@ Current scenarios:
   delete/re-create swap, node stop/start rewiring) and a relay negative
   control.
 
+Every kernel scenario also asserts the L2-anchor spec's server-side
+guarantees (`docs/design/ubridge-l2-anchor-spec.md` §E): anchors and bridges
+carry no L3 identity (`assert_pure_l2`), joined ports are FORWARDING
+(`assert_forwarding`), and — in the Dynamips suite — an idle link with the
+guests silenced stays silent for 5 s (`assert_idle_silence`). These skip on
+a uBridge without `link l2only` (probed once by `harness.l2only_supported()`).
+
 Set `GNS3_E2E_DEBUG=1` to run the isolated instance with debug logging (the
 uBridge command stream lands in its `server.log`).

tests/e2e/harness.py

@@ -43,6 +43,7 @@ Environment knobs:
   test fails, and print their ids, for inspection.
 """
 
+import contextlib
 import json
 import os
 import re
@@ -62,6 +63,9 @@ REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__fi
 # tc lives in /usr/sbin on most distributions; subprocesses do not always
 # inherit a PATH that has it.
 TC = shutil.which("tc", path=os.environ.get("PATH", "") + ":/usr/sbin:/sbin")
+# Reading addresses with iproute2 needs no privileges; the L2 assertions
+# below refuse to pass silently when it is missing.
+IP = shutil.which("ip", path=os.environ.get("PATH", "") + ":/usr/sbin:/sbin")
 
 # IOS prompt as the last thing on the stream: any hostname/mode shape.
 PROMPT_RE = re.compile(r"[A-Za-z0-9\-_.()/]+[>#]\s*$")
@@ -512,6 +516,138 @@ def bridge_members(bridge):
     return sorted(os.listdir(f"/sys/class/net/{bridge}/brif"))
 
 
+# ---------------------------------------------------------------------------
+# L2-anchor spec assertions (docs/design/ubridge-l2-anchor-spec.md §E)
+# ---------------------------------------------------------------------------
+
+# uBridge's creators harden every host-side datapath device with
+# ``link l2only``; these assertions verify the result from the server side.
+# They need the probe below to confirm the command exists — the spec's
+# degradation contract is a skip on old builds, where anchors keep the
+# kernel-default L3 behaviour (measured: 6 frames of MLD/DAD noise per 2 s).
+
+
+def l2only_supported():
+    """Whether this host's uBridge knows ``link l2only`` (probed once).
+
+    Sends the command for a nonexistent device to a throwaway uBridge: a
+    208 reply means the command exists, ``202-Unknown command`` an old
+    build. Anything unreadable also degrades to False — the L2 assertions
+    skip rather than fail, mirroring the tc-capability degradation."""
+    global _l2only_support
+    if _l2only_support is None:
+        _l2only_support = _probe_l2only()
+    return _l2only_support
+
+
+_l2only_support = None
+
+
+def _probe_l2only():
+    ubridge = shutil.which("ubridge")
+    if not ubridge:
+        return False
+    port = _free_port()
+    with tempfile.TemporaryDirectory(prefix="gns3-e2e-l2probe-") as workdir:
+        # S603: resolved binary, fixed argv, no shell
+        proc = subprocess.Popen(  # noqa: S603
+            [ubridge, "-H", f"127.0.0.1:{port}"],
+            cwd=workdir,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+        try:
+            sock = None
+            deadline = time.time() + 10
+            while time.time() < deadline and sock is None:
+                if proc.poll() is not None:
+                    return False
+                try:
+                    sock = socket.create_connection(("127.0.0.1", port), timeout=1)
+                except OSError:
+                    time.sleep(0.1)
+            if sock is None:
+                return False
+            with sock:
+                sock.settimeout(3)
+                sock.sendall(b"link l2only e2el2probe0\n")
+                buf = b""
+                deadline = time.time() + 5
+                while time.time() < deadline:
+                    try:
+                        chunk = sock.recv(4096)
+                    except TimeoutError:
+                        break
+                    if not chunk:
+                        break
+                    buf += chunk
+                    if re.match(rb"\d{3}-", buf.split(b"\n")[0]):
+                        break
+            reply = buf.decode(errors="replace").splitlines()[0] if buf else ""
+            # 208 = the command ran and refused the nonexistent device
+            return reply.startswith("208-")
+        finally:
+            proc.terminate()
+            with contextlib.suppress(Exception):
+                proc.wait(timeout=5)
+
+
+def assert_pure_l2(name):
+    """§E.1: a host-side datapath device (anchor or bridge) carries no L3
+    identity — no IPv4/IPv6 address, and addrgenmode none so none can
+    self-provision (no DAD, MLD or RS is ever generated from it)."""
+    mode_path = f"/sys/class/net/{name}/addr_gen_mode"
+    mode = sysfs(mode_path) if os.path.exists(mode_path) else None
+    assert mode in (None, "1"), f"{name}: addr_gen_mode is {mode!r}, expected none ('1')"
+    if os.path.exists("/proc/net/if_inet6"):
+        with open("/proc/net/if_inet6") as f:
+            assigned = [line.strip() for line in f if line.split()[-1] == name]
+        assert not assigned, f"{name}: unexpected IPv6 address: {assigned}"
+    if not IP:
+        raise AssertionError("ip binary not found — cannot verify the anchor has no IPv4 address")
+    # S603: resolved binary, fixed argv, no shell
+    v4 = subprocess.run(  # noqa: S603
+        [IP, "-o", "-4", "addr", "show", "dev", name], capture_output=True, text=True, check=False
+    ).stdout.strip()
+    assert not v4, f"{name}: unexpected IPv4 address: {v4}"
+
+
+def brport_state(name):
+    """The interface's bridge-port STP state (None when not enslaved); the
+    kernel reports 3 for forwarding."""
+    path = f"/sys/class/net/{name}/brport/state"
+    return sysfs(path) if os.path.exists(path) else None
+
+
+def assert_forwarding(name):
+    """Silent-failure guard found while validating the L2 spec: an attached
+    bridge port must be FORWARDING. A bridge device left DOWN keeps its
+    ports disabled and forwards nothing — with no error anywhere, this
+    assertion is the only signal."""
+    state = brport_state(name)
+    assert state == "3", f"{name}: bridge port state {state!r}, expected forwarding (3)"
+
+
+def packet_counters(name):
+    """(rx, tx) packet counters of a host interface."""
+    return (
+        int(sysfs(f"/sys/class/net/{name}/statistics/rx_packets")),
+        int(sysfs(f"/sys/class/net/{name}/statistics/tx_packets")),
+    )
+
+
+def assert_idle_silence(*names, settle=2.5, window=5.0):
+    """§E.2: after the one-shot enslavement burst settles (the IGMP/MLD
+    membership reports of the bridge are accepted residual, see §0), an
+    idle device in its normal role shows no traffic for *window* seconds.
+    The baseline without the hardening: 6 frames / 2 s, continuous."""
+    time.sleep(settle)
+    before = {name: packet_counters(name) for name in names}
+    time.sleep(window)
+    noisy = {name: (before[name], packet_counters(name)) for name in names if packet_counters(name) != before[name]}
+    assert not noisy, f"idle devices saw traffic (before -> after): {noisy}"
+
+
 def link_bridge_name(link_id):
     """The per-link kernel bridge name the controller derives from a link
     id (see UDPLink._prepare)."""

tests/e2e/test_dynamips_kernel_datapath.py

@@ -71,6 +71,13 @@ def _boot(server, project_id, node_id, timeout=360):
     return console
 
 
+def _eth_shutdown(console, shut):
+    console.run("conf t")
+    console.run("interface f1/0")
+    console.run("shutdown" if shut else "no shutdown")
+    console.run("end")
+
+
 def test_dynamips_kernel_datapath():
     server = harness.live_server(kernel=True)
     compute = server.compute
@@ -108,6 +115,19 @@ def test_dynamips_kernel_datapath():
         assert harness.tap_exists(t1) and harness.tap_exists(t2), (t1, t2)
         assert harness.bridge_members(bridge) == sorted([t1, t2]), harness.bridge_members(bridge)
 
+        # L2-anchor spec §E.1: anchors and the per-link bridge carry no L3
+        # identity (skipped on a uBridge without `link l2only` — the spec's
+        # degradation), and both ports are FORWARDING — the silent-failure
+        # guard against a bridge device left DOWN.
+        if harness.l2only_supported():
+            harness.assert_pure_l2(t1)
+            harness.assert_pure_l2(t2)
+            harness.assert_pure_l2(bridge)
+        else:
+            print(".. uBridge without link l2only: skipping the §E.1 assertions")
+        harness.assert_forwarding(t1)
+        harness.assert_forwarding(t2)
+
         harness.configure_ios(c1, "R1", eth_ip="10.1.1.1")
         harness.configure_ios(c2, "R2", eth_ip="10.1.1.2")
         out = c1.run("show ip int brief")
@@ -116,6 +136,18 @@ def test_dynamips_kernel_datapath():
         baseline = harness.wait_ping(c1, "10.1.1.2")
         assert baseline["success"] == 100, baseline["raw"]
 
+        # L2-anchor spec §E.2: shut the guest interfaces (silencing IOS's own
+        # CDP/keepalives — the only legitimate speakers on the segment) and
+        # demand the host side stays silent for 5 s: without the hardening
+        # the host's own MLD/DAD noise floods the link (6 frames / 2 s).
+        if harness.l2only_supported():
+            for console in (c1, c2):
+                _eth_shutdown(console, shut=True)
+            harness.assert_idle_silence(t1, t2, bridge)
+            for console in (c1, c2):
+                _eth_shutdown(console, shut=False)
+            assert harness.wait_ping(c1, "10.1.1.2", attempts=5)["success"] == 100
+
         # delay 100: netem on both anchors (each impairs one direction),
         # one-way ~100 ms => RTT grows by ~200 ms
         compute.call("PUT", f"/projects/{pid}/links/{lid}", {"filters": {"delay": [100]}})
@@ -237,6 +269,10 @@ def test_dynamips_relay_control():
         assert harness.bridge_members(harness.link_bridge_name(lid)) is None
         if harness.tap_exists(t1):
             assert not harness.tap_up(t1), "an unattached relay anchor should sit down"
+            if harness.l2only_supported():
+                # the L2 hardening is a creation-time property, not a
+                # datapath one: relay anchors are pure L2 too
+                harness.assert_pure_l2(t1)
 
         harness.configure_ios(c1, "R1", eth_ip="10.1.1.1")
         harness.configure_ios(c2, "R2", eth_ip="10.1.1.2")

tests/e2e/test_ethernet_switch_kernel_datapath.py

@@ -139,6 +139,18 @@ def test_ethernet_switch_kernel_fast_path():
         baseline = harness.wait_ping(c1, "10.1.1.2")
         assert baseline["success"] == 100, baseline["raw"]
 
+        # L2-anchor spec §E.1: the absorbed anchors and the switch bridge
+        # carry no L3 identity (skipped on a uBridge without `link l2only`),
+        # and the joined ports are FORWARDING — the silent-failure guard.
+        if harness.l2only_supported():
+            harness.assert_pure_l2(a1)
+            harness.assert_pure_l2(a2)
+            harness.assert_pure_l2(sw_bridge)
+        else:
+            print(".. uBridge without link l2only: skipping the §E.1 assertions")
+        harness.assert_forwarding(a1)
+        harness.assert_forwarding(a2)
+
         # VLAN programming on absorbed anchors: different access VLANs
         # isolate, the same VLAN bridges again.
         compute.call(
@@ -274,6 +286,12 @@ def test_ethernet_switch_relay_control():
         members = harness.bridge_members(sw_bridge)
         assert members == sorted([f"{sw_bridge}-0", f"{sw_bridge}-1"]), members
         assert a1 not in members
+        if harness.l2only_supported():
+            # creation-time hardening applies on the relay too: the switch's
+            # own port TAPs and its bridge are pure L2
+            for member in members:
+                harness.assert_pure_l2(member)
+            harness.assert_pure_l2(sw_bridge)
 
         harness.configure_ios(c1, "R1", eth_ip="10.1.1.1")
         harness.configure_ios(c2, "R2", eth_ip="10.1.1.2")

tests/e2e/test_iol_docker_kernel_datapath.py

@@ -110,6 +110,18 @@ def test_iol_docker_kernel_datapath():
         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 the per-link bridge carry no L3
+        # identity (skipped on a uBridge without `link l2only`), and both
+        # ports are FORWARDING — the silent-failure guard.
+        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)
+
         harness.configure_ios(c1, "R1", eth_ip="10.1.1.1", eth_if="Ethernet0/0")
         harness.configure_ios(c2, "R2", eth_ip="10.1.1.2", eth_if="Ethernet0/0")
         out = c1.run("show ip int brief")
@@ -236,6 +248,10 @@ def test_iol_docker_relay_control():
         assert harness.bridge_members(harness.link_bridge_name(link["link_id"])) is None
         if harness.tap_exists(a1):
             assert not harness.tap_up(a1), a1
+            if harness.l2only_supported():
+                # the L2 hardening is a creation-time property, not a
+                # datapath one: relay anchors are pure L2 too
+                harness.assert_pure_l2(a1)
         harness.configure_ios(c1, "R1", eth_ip="10.1.1.1", eth_if="Ethernet0/0")
         harness.configure_ios(c2, "R2", eth_ip="10.1.1.2", eth_if="Ethernet0/0")
         relay_ping = harness.wait_ping(c1, "10.1.1.2")