From f80b89786495e4364132f2099397dd67dd1ac9fe Mon Sep 17 00:00:00 2001 From: Matt <47545907+SoundMatt@users.noreply.github.com> Date: Fri, 31 Jul 2026 17:52:01 -0700 Subject: [PATCH] =?UTF-8?q?feat:=20v3.2.0=20=E2=80=94=20real=20UDP=20socke?= =?UTF-8?q?t=20+=20new=20L2=20raw-Ethernet=20transport?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This crate's transport layer had two real gaps, confirmed by direct inspection rather than assumption: no raw-Ethernet/L2 transport existed at all (the same gap every other RCP-family repo — go-RCP, cpp-RCP, c-RCP — has), and src/udp.rs's UdpSocket trait had no implementation over a real OS socket either — only the in-process EchoUdp/QueuedUdpSocket test doubles, and src/bin/rcp.rs's own prior doc comment admitted this plainly. TC18 §10.1 names both a layer-2 EtherType (0x22F0) and UDP/IP encapsulation ("described in Annex J", of the base IEEE 1722-2016 standard) as legal transports; this builds both as permanent, first-class, equally-supported options, closing all three real gaps at once. This is the first real network I/O this crate has ever shipped for RCP. - src/udp.rs gains StdUdpSocket, a real UdpSocket implementation over a bound std::net::UdpSocket, corrected to IEEE 1722-2016 Annex J framing from the start. Every send_to prepends, and every recv_from strips, a 4-byte big-endian encapsulation sequence number (encode_annex_j_udp_payload/decode_annex_j_udp_payload) — a monotonic per-socket counter with no invented receiver-side semantics beyond that. New ANNEX_J_CONTROL_PORT (17221, the default) and ANNEX_J_CONTINUOUS_PORT (17220) constants. Provenance: this crate has no access to the paywalled IEEE 1722-2016 standard text — the port numbers and sequence-number field are taken from two independent public secondary sources (a Wireshark issue tracker discussion, and the COVESA Open1722 reference implementation's Avtp_Udp_t header struct), flagged as such rather than presented with false certainty. - New src/l2.rs — a raw-Ethernet (layer 2) transport, Linux only, mirroring udp.rs's own UdpSocket/UdpTransport abstraction one wire layer down: encode_ethernet_frame/decode_ethernet_frame (dest MAC + src MAC + EtherType 0x22F0 + AVTPDU directly, no encapsulation sequence number — that field is Annex J/UDP-specific), an L2Socket trait, an L2Transport client, and — target_os = "linux" only — RawEthernetSocket, a real AF_PACKET/SOCK_RAW production L2Socket that reads its own interface's MAC via getifaddrs rather than requiring the caller to supply one. Every other target gets a same-named stub whose bind always returns a clear Err rather than silently no-op-ing. - This crate is #![forbid(unsafe_code)] crate-wide, which rules out a direct libc socket()/bind()/sendto()/recvfrom() implementation (would need unsafe extern "C" calls; forbid cannot be locally overridden). RawEthernetSocket is instead built on the nix crate (new target_os = "linux"-only dependency), whose socket/bind/sendto/recvfrom/ setsockopt/getifaddrs are all safe Rust fns — unsafe lives inside nix's own crate, never this one's. Flagged in src/l2.rs's own module doc comment as a deliberate judgment call. - src/bin/rcp.rs gains a new `serve --udp [--port ] [--stream ] [--max-requests ]` command — the first rust-rcp command backed by a real OS socket instead of an in-process RcServer invoked directly: it binds a real StdUdpSocket and runs UdpRcServer (previously only ever exercised against mock sockets) against it. The module doc comment's prior "no concrete UdpSocket implementation over a real OS socket" note is corrected accordingly. - New tests: pure byte-manipulation round trips for both Annex J encapsulation and Ethernet framing (no socket/privileges), mock-backed L2Transport/UdpTransport request/response tests, real loopback StdUdpSocket round trips (including a real end-to-end StdUdpSocket+UdpRcServer discovery request), and a new Linux-only CI job (l2-veth) that creates a real veth0/veth1 pair and runs a real RawEthernetSocket frame round trip under sudo. MINOR release: StdUdpSocket, the Annex J constants/functions, and the entire new l2 module are new pub items only. docs/PUBLIC_API.txt regenerated (purely additive diff); .fusa-reqs.json gains REQ-UDP-012..014, REQ-L2-001..008, REQ-CLI-010 (564/564 traced). Signed-off-by: Matt <47545907+SoundMatt@users.noreply.github.com> --- .fusa-reqs.json | 108 ++++++ .github/workflows/ci.yml | 30 ++ CHANGELOG.md | 119 ++++++ Cargo.lock | 31 +- Cargo.toml | 15 +- docs/PUBLIC_API.txt | 65 ++++ docs/SEMVER.md | 2 +- src/bin/rcp.rs | 135 ++++++- src/l2.rs | 804 +++++++++++++++++++++++++++++++++++++++ src/lib.rs | 1 + src/udp.rs | 345 +++++++++++++++++ 11 files changed, 1646 insertions(+), 9 deletions(-) create mode 100644 src/l2.rs diff --git a/.fusa-reqs.json b/.fusa-reqs.json index fdb7f7c..cce023c 100644 --- a/.fusa-reqs.json +++ b/.fusa-reqs.json @@ -4932,6 +4932,114 @@ "level": "HLR", "asil": "ASIL-B", "verificationMethod": "test" + }, + { + "id": "REQ-UDP-012", + "title": "Annex J UDP encapsulation sequence number is prepended/stripped correctly", + "text": "encode_annex_j_udp_payload(seq, avtpdu) prepends seq as 4 big-endian octets before avtpdu with no other framing; decode_annex_j_udp_payload(buf) is its exact inverse, returning (seq, avtpdu_bytes) for any buf of at least 4 bytes and Err(RcpError::ShortFrame) for fewer than 4 bytes; ANNEX_J_CONTROL_PORT (17221) and ANNEX_J_CONTINUOUS_PORT (17220) are distinct constants documented as taken from public secondary sources (a Wireshark issue tracker discussion and the COVESA Open1722 reference implementation), not the paywalled IEEE 1722-2016 primary standard text", + "standard": "iso26262", + "level": "HLR", + "asil": "ASIL-B", + "verificationMethod": "test" + }, + { + "id": "REQ-UDP-013", + "title": "StdUdpSocket::send_to encapsulates outgoing datagrams with a monotonically increasing sequence number", + "text": "StdUdpSocket is a real UdpSocket implementation over a bound std::net::UdpSocket; StdUdpSocket::bind/new_default_port construct it against a real OS socket, and every StdUdpSocket::send_to call prepends the current value of a per-instance monotonically increasing u32 counter (starting at 0) via encode_annex_j_udp_payload before writing to the real socket, with no receiver-side semantics (e.g. loss detection) invented or implied for that counter beyond monotonic increase", + "standard": "iso26262", + "level": "HLR", + "asil": "ASIL-B", + "verificationMethod": "test" + }, + { + "id": "REQ-UDP-014", + "title": "StdUdpSocket::recv_from strips the Annex J sequence number and maps a real socket timeout to RcpError::Timeout", + "text": "StdUdpSocket::recv_from applies timeout via the real socket's SO_RCVTIMEO (None blocks indefinitely), strips the leading 4-byte Annex J sequence number from whatever datagram is received via decode_annex_j_udp_payload before returning the remaining AVTPDU bytes to the caller, and maps a real OS-level receive timeout (WouldBlock/TimedOut) to Err(RcpError::Timeout) rather than propagating the raw std::io::Error", + "standard": "iso26262", + "level": "HLR", + "asil": "ASIL-B", + "verificationMethod": "test" + }, + { + "id": "REQ-L2-001", + "title": "ETHERTYPE_AVTP names the real IEEE 1722 EtherType and is placed big-endian in the Ethernet header", + "text": "ETHERTYPE_AVTP is 0x22F0, matching TC18 \u00a710.1 (\u201can AVTPDU is marked by an EtherType value of 0x22F0\u201d); encode_ethernet_frame writes it as bytes 12-13 of the frame in big-endian order, and decode_ethernet_frame reads bytes 12-13 the same way when validating a frame's EtherType", + "standard": "iso26262", + "level": "HLR", + "asil": "ASIL-B", + "verificationMethod": "test" + }, + { + "id": "REQ-L2-002", + "title": "encode_ethernet_frame/decode_ethernet_frame round-trip a raw Ethernet II frame with no encapsulation sequence number", + "text": "encode_ethernet_frame(dest_mac, src_mac, avtpdu) produces dest_mac (6 bytes) || src_mac (6 bytes) || ETHERTYPE_AVTP (2 bytes, big-endian) || avtpdu directly, with no additional framing (unlike crate::udp::encode_annex_j_udp_payload's 4-byte sequence number, which has no L2 counterpart); decode_ethernet_frame is its exact inverse for any well-formed input, returns Err(RcpError::ShortFrame) for fewer than 14 bytes, and returns Err(RcpError::Other(_)) (not a panic) when the EtherType field is not ETHERTYPE_AVTP", + "standard": "iso26262", + "level": "HLR", + "asil": "ASIL-B", + "verificationMethod": "test" + }, + { + "id": "REQ-L2-003", + "title": "L2Socket trait mirrors UdpSocket's testable-abstraction shape for MAC-addressed raw Ethernet frames", + "text": "L2Socket::send(frame) and L2Socket::recv(timeout) operate on already-framed encode_ethernet_frame/decode_ethernet_frame bytes, the same already-framed-bytes-in-bytes-out contract crate::udp::UdpSocket documents, letting L2Transport be tested against a mock L2Socket with no real socket involved", + "standard": "iso26262", + "level": "HLR", + "asil": "ASIL-B", + "verificationMethod": "test" + }, + { + "id": "REQ-L2-004", + "title": "L2Transport is constructed with a local StreamId, an L2Socket, a caller-supplied destination MAC, and this transport's own source MAC", + "text": "L2Transport::new(local_stream, socket, dest_mac, src_mac) stores all four; L2Transport::local_stream/dest_mac/src_mac each return exactly the value passed to the constructor; dest_mac may be unicast or multicast and is always a caller input, since this crate does not derive or allocate a multicast MAC of its own", + "standard": "iso26262", + "level": "HLR", + "asil": "ASIL-B", + "verificationMethod": "test" + }, + { + "id": "REQ-L2-005", + "title": "L2Transport::send_acf_abb/send_acf_gbb reject a zero timeout immediately, matching UdpTransport's own discipline", + "text": "L2Transport::send_acf_abb and L2Transport::send_acf_gbb both return Err(RcpError::Timeout) immediately, without ever calling the underlying L2Socket, when passed Some(Duration::ZERO), mirroring crate::udp::UdpTransport::send_acf_abb's own zero-timeout short-circuit", + "standard": "iso26262", + "level": "HLR", + "asil": "ASIL-B", + "verificationMethod": "test" + }, + { + "id": "REQ-L2-006", + "title": "L2Transport::send_acf_abb/send_acf_gbb build an NTSCF/AVTPDU frame, wrap it as a raw Ethernet frame, and verify the response echoes byte_bus_id", + "text": "L2Transport::send_acf_abb/send_acf_gbb encode the request via crate::acf::encode_acf_abb/encode_acf_gbb, wrap it in an NTSCF frame via crate::avtp::encode_ntscf_frame under this transport's local_stream, wrap that in a raw Ethernet frame via encode_ethernet_frame addressed to dest_mac from src_mac, send it, decode the received response through decode_ethernet_frame/crate::avtp::decode_ntscf_frame/crate::acf::decode_acf_abb (or decode_acf_gbb), and reject it via crate::acf::verify_echo_back if the response's byte_bus_id does not match the request's", + "standard": "iso26262", + "level": "HLR", + "asil": "ASIL-B", + "verificationMethod": "test" + }, + { + "id": "REQ-L2-007", + "title": "RawEthernetSocket (Linux) binds an AF_PACKET/SOCK_RAW socket to a named interface, reading that interface's own MAC rather than requiring the caller to supply one", + "text": "On target_os = \"linux\", RawEthernetSocket::bind(interface_name) opens a real AF_PACKET/SOCK_RAW socket (requiring CAP_NET_RAW or root) and binds it to interface_name, reading that interface's own link-layer address via getifaddrs for RawEthernetSocket::mac() rather than accepting a caller-supplied MAC; RawEthernetSocket::send/recv implement L2Socket over that real socket and round-trip a real Ethernet frame byte-for-byte over a veth pair", + "standard": "iso26262", + "level": "HLR", + "asil": "ASIL-B", + "verificationMethod": "test" + }, + { + "id": "REQ-L2-008", + "title": "RawEthernetSocket exists and fails explicitly, never silently, on every non-Linux target", + "text": "On any target where target_os != \"linux\", crate::l2::RawEthernetSocket still exists as a type (so downstream code can reference it unconditionally) but RawEthernetSocket::bind always returns Err(RcpError::Other(_)) explaining that AF_PACKET raw sockets are a Linux-specific facility, rather than silently no-op-ing or panicking", + "standard": "iso26262", + "level": "HLR", + "asil": "ASIL-B", + "verificationMethod": "test" + }, + { + "id": "REQ-CLI-010", + "title": "serve --udp [--port ] [--stream ] [--max-requests ] runs UdpRcServer over a real StdUdpSocket", + "text": "The CLI must, when invoked with serve --udp , bind a real rcp::udp::StdUdpSocket to :--port (default rcp::udp::ANNEX_J_CONTROL_PORT) and dispatch inbound requests through rcp::udp::UdpRcServer::serve_one against a fresh RcServer, stopping after --max-requests requests have been served if given (default: unbounded, until a fatal socket error), exiting 3 on a bind or fatal serve error", + "standard": "iso26262", + "level": "HLR", + "asil": "ASIL-B", + "verificationMethod": "test" } ] } diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d337550..52e4223 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -47,6 +47,36 @@ jobs: - name: test release run: cargo test --all-targets --release --locked + # ── L2 (raw Ethernet) real veth round trip (Linux only) ─────────────────── + # `src/l2.rs`'s RawEthernetSocket needs CAP_NET_RAW (or root) to open an + # AF_PACKET/SOCK_RAW socket at all — not exercisable by the normal `test` + # job above, which is why the round-trip test itself is #[ignore]d by + # default. This job creates a real veth0/veth1 pair and runs that one + # test under sudo against it, proving a real frame round-trips + # byte-for-byte over a real (virtual) Ethernet link — not just that the + # framing/trait logic type-checks. + l2-veth: + name: L2 raw-Ethernet veth round trip (Linux) + needs: lint + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: dtolnay/rust-toolchain@stable + - uses: Swatinem/rust-cache@v2 + - name: create veth0/veth1 pair + run: | + sudo ip link add veth0 type veth peer name veth1 + sudo ip link set veth0 up + sudo ip link set veth1 up + - name: build l2 tests + run: cargo test --lib --no-run --locked --message-format=json > build.json + - name: run real_raw_ethernet_socket_round_trips_a_frame_over_a_veth_pair under sudo + run: | + bin=$(jq -r 'select(.profile.test == true and (.target.name == "rcp")) | .filenames[]' build.json | head -n1) + test -n "$bin" + sudo "$bin" l2::tests::real_raw_ethernet_socket_round_trips_a_frame_over_a_veth_pair \ + --exact --ignored --nocapture + # ── Coverage ────────────────────────────────────────────────────────────── coverage: name: Coverage (llvm-cov) diff --git a/CHANGELOG.md b/CHANGELOG.md index ad79855..ea14524 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,125 @@ the roadmap milestone that produced them (see `ROADMAP.md`), since this crate's `Cargo.toml` version does not move until the OPEN Alliance TC18 core replacement reaches `v1.0.0`. +## v3.2.0 (2026-07-31 real UDP socket + new L2 raw-Ethernet transport) — closed + +This crate's transport layer had two real gaps, confirmed by direct +inspection rather than assumption: no raw-Ethernet/L2 transport existed at +all (the same gap every other RCP-family repo — `go-RCP`, `cpp-RCP`, +`c-RCP` — has), and `src/udp.rs`'s `UdpSocket` trait had no implementation +over a real OS socket either — only the in-process `EchoUdp`/ +`QueuedUdpSocket` test doubles, and `src/bin/rcp.rs`'s own prior doc +comment admitted this plainly. TC18 §10.1 names both a layer-2 EtherType +(`0x22F0`) and UDP/IP encapsulation ("described in Annex J", of the base +IEEE 1722-2016 standard) as legal transports; this item builds both as +permanent, first-class, equally-supported options, closing all three real +gaps rather than just one — this is the first real network I/O this crate +has ever shipped for RCP. + +- **rust-RCP-NET-01 (feature):** `src/udp.rs` gains + [`StdUdpSocket`], a real `UdpSocket` implementation over a bound + `std::net::UdpSocket`, corrected to IEEE 1722-2016 Annex J framing from + the start (there was no legacy UDP wire format to preserve). Every + `send_to` prepends, and every `recv_from` strips, a 4-byte big-endian + "encapsulation sequence number" ([`encode_annex_j_udp_payload`]/ + [`decode_annex_j_udp_payload`]) — a per-`StdUdpSocket` monotonically + increasing counter with no invented receiver-side semantics (e.g. loss + detection) beyond that. New constants `ANNEX_J_CONTROL_PORT` (17221, + the applicable port for RCP's control-plane request/response/ + acknowledgement traffic, and `StdUdpSocket::new_default_port`'s + default) and `ANNEX_J_CONTINUOUS_PORT` (17220, streaming traffic, named + but unused). **Provenance note**, stated once here and referenced from + every touchpoint in code: this crate has no access to the paywalled + IEEE 1722-2016 standard text: the port numbers and the sequence-number + field are taken from two independent public secondary sources instead + — a Wireshark issue tracker discussion of the real Annex J framing, and + the COVESA Open1722 open-source reference implementation's `Avtp_Udp_t` + header struct (`include/avtp/Udp.h`, BSD-3-Clause, + ) — and are flagged as such rather + than presented with false certainty. New `REQ-UDP-012`/`REQ-UDP-013`/ + `REQ-UDP-014`. +- **rust-RCP-NET-02 (feature):** new `src/l2.rs` — a raw-Ethernet (layer + 2) transport, Linux only, mirroring `src/udp.rs`'s own + `UdpSocket`/`UdpTransport` abstraction one wire layer down: + [`encode_ethernet_frame`]/[`decode_ethernet_frame`] (destination MAC + + source MAC + EtherType `0x22F0` big-endian + the AVTPDU bytes directly + — no encapsulation sequence number; that field is Annex J/UDP-specific + and has no L2 counterpart), an [`L2Socket`] trait mirroring `UdpSocket` + (`SocketAddr` replaced by a raw `[u8; 6]` MAC), [`L2Transport`] + mirroring `UdpTransport`'s `send_acf_abb`/`send_acf_gbb` client shape, + and — `target_os = "linux"` only — [`RawEthernetSocket`], a real + `AF_PACKET`/`SOCK_RAW` production `L2Socket` that reads its own + interface's MAC via `getifaddrs` rather than requiring the caller to + supply one (a caller-supplied destination MAC is still required — + multicast-MAC derivation is a base-IEEE-1722 algorithm this crate does + not have). Every other target gets a same-named stub whose `bind` + always returns a clear `Err` rather than silently no-op-ing, so the + type can be referenced unconditionally. Server-side L2 dispatch (an + `L2RcServer` mirroring `UdpRcServer`) is out of scope for this item — + flagged as a deliberate follow-up, not bundled in silently; this item's + server-facing wiring is `UdpRcServer` run over `StdUdpSocket` (see + rust-RCP-NET-03 below). New `REQ-L2-001` through `REQ-L2-008`. +- **A flagged judgment call — `nix`, not raw `libc` `unsafe` syscalls:** + this crate is `#![forbid(unsafe_code)]` crate-wide, and `forbid` cannot + be locally overridden (E0453) — `src/capi.rs`'s own doc comment already + named this rule as the reason this crate has never built a raw-pointer + FFI boundary. A direct `libc` `socket()`/`bind()`/`sendto()`/ + `recvfrom()` implementation would require `unsafe extern "C"` calls in + this crate's own source, which is not available at all here, not a + style choice. `RawEthernetSocket` is instead built on the `nix` crate + (`target_os = "linux"`-only dependency, new to `Cargo.toml`), whose + `socket`/`bind`/`sendto`/`recvfrom`/`setsockopt`/`getifaddrs` functions + are all safe Rust `fn`s — `unsafe` lives inside `nix`'s own crate, + never this one's — confirmed against `nix` 0.31's published API before + writing the module, not assumed. `nix` is a narrowly-scoped + POSIX-bindings crate, not a heavyweight packet-crafting framework like + `pnet`, matching this item's own minimal-footprint intent. +- **rust-RCP-NET-03 (feature):** `src/bin/rcp.rs` gains a new `serve --udp + [--port ] [--stream ] [--max-requests ]` command — + the first `rust-rcp` command backed by a real OS socket instead of an + in-process `RcServer` invoked directly. It binds a real `StdUdpSocket` + and runs `UdpRcServer` (previously only ever exercised against mock + sockets in this crate's own unit tests) against it. `discover`/ + `register`/`endpoint` remain deliberately ephemeral/in-process, per + this file's own pre-existing "Provenance note" (unchanged by this + item); `serve` is a new, additive, real-network-facing command, not a + replacement for them. The module doc comment's prior "no concrete + `rcp::udp::UdpSocket` implementation over a real OS socket" note is + updated accordingly. New `REQ-CLI-010`. +- **Tests, no privileges/Linux required:** pure byte-manipulation round + trips for both the Annex J encapsulation + (`annex_j_encode_decode_round_trips`, short-buffer rejection) and the + Ethernet frame encode/decode (`ethernet_frame_encode_decode_round_trips`, + short-frame/wrong-EtherType rejection), plus mock-socket-backed + `L2Transport`/`UdpTransport` request/response tests (`EchoL2`/`QueuedL2`, + the `L2Socket` analogs of `udp`'s own `EchoUdp`/`QueuedUdpSocket`) — all + run everywhere, no real socket involved. +- **Tests, real sockets:** a real loopback `StdUdpSocket` round trip + (`std_udp_socket_round_trips_over_real_loopback_socket`), a test + proving the encapsulation sequence number actually increments on the + wire by inspecting raw bytes with a bypass `std::net::UdpSocket`, a + real receive-timeout test, and a new end-to-end test composing a real + `StdUdpSocket` client against a real `StdUdpSocket` + `UdpRcServer` + server over real loopback sockets + (`std_udp_socket_and_udp_rc_server_serve_a_real_discovery_request_end_to_end`) + — all run in the normal cross-platform `test` CI job (ubuntu/macos/ + windows), no privileges required. +- **New Linux-only CI job (`l2-veth`):** creates a real `veth0`/`veth1` + pair under `sudo`, then runs a `#[cfg(target_os = "linux")]`, + `#[ignore]`d-by-default test + (`real_raw_ethernet_socket_round_trips_a_frame_over_a_veth_pair`) with + `-- --ignored`, proving a real `RawEthernetSocket` frame round-trips + byte-for-byte over a real (virtual) Ethernet link — not just that the + framing/trait logic type-checks. +- This is a MINOR (additive, non-breaking) release: `StdUdpSocket`, + `ANNEX_J_CONTROL_PORT`/`ANNEX_J_CONTINUOUS_PORT`, + `encode_annex_j_udp_payload`/`decode_annex_j_udp_payload`, and the + entire new `l2` module are new `pub` items only — no existing item + changed shape. `docs/PUBLIC_API.txt` is regenerated accordingly (purely + additive diff) per `docs/SEMVER.md`; `.fusa-reqs.json` gains + `REQ-UDP-012`-`REQ-UDP-014`, `REQ-L2-001`-`REQ-L2-008`, and + `REQ-CLI-010` (564/564 traced). + ## v3.1.0 (2026-07-31 E2E CRC trailer wire-order fix) — closed While independently verifying `v3.0.0`'s `acf` wire-format rework byte-for- diff --git a/Cargo.lock b/Cargo.lock index 80caf6d..d3e0c32 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -62,6 +62,12 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "cfg_aliases" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527" + [[package]] name = "chrono" version = "0.4.45" @@ -235,6 +241,28 @@ version = "2.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" +[[package]] +name = "memoffset" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "488016bfae457b036d996092f6cb448677611ce4449e970ceaf42695203f218a" +dependencies = [ + "autocfg", +] + +[[package]] +name = "nix" +version = "0.31.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf20d2fde8ff38632c426f1165ed7436270b44f199fc55284c38276f9db47c3d" +dependencies = [ + "bitflags", + "cfg-if", + "cfg_aliases", + "libc", + "memoffset", +] + [[package]] name = "num-traits" version = "0.2.19" @@ -305,11 +333,12 @@ checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" [[package]] name = "rcp" -version = "3.1.0" +version = "3.2.0" dependencies = [ "async-trait", "base64", "chrono", + "nix", "parking_lot", "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index 60757d7..f75293e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "rcp" -version = "3.1.0" +version = "3.2.0" edition = "2021" rust-version = "1.75" license = "MPL-2.0" @@ -44,3 +44,16 @@ strip = true opt-level = "z" lto = true codegen-units = 1 + +# `src/l2.rs`'s raw AF_PACKET/SOCK_RAW transport: this crate is +# `#![forbid(unsafe_code)]` crate-wide (`src/lib.rs`), which rules out a +# direct `libc` `socket()`/`bind()`/`sendto()`/`recvfrom()` implementation +# (that would require `unsafe extern "C"` calls in this crate's own +# source, and `forbid` cannot be locally overridden). `nix` wraps those +# same syscalls in a fully safe Rust API (`unsafe` lives inside `nix`'s +# own crate, not this one) and is gated to `target_os = "linux"` here so +# no other build target's dependency graph grows because of it — see +# `src/l2.rs`'s own module doc comment, "Why `nix`, not raw `libc` +# `unsafe` syscalls — a flagged judgment call", for the full reasoning. +[target.'cfg(target_os = "linux")'.dependencies] +nix = { version = "0.31.3", features = ["socket", "net"] } diff --git a/docs/PUBLIC_API.txt b/docs/PUBLIC_API.txt index e114abe..7232491 100644 --- a/docs/PUBLIC_API.txt +++ b/docs/PUBLIC_API.txt @@ -1765,6 +1765,49 @@ impl core::panic::unwind_safe::RefUnwindSafe for rcp::iso21434::Threat impl core::panic::unwind_safe::UnwindSafe for rcp::iso21434::Threat pub fn rcp::iso21434::filter_by_risk(&[rcp::iso21434::Threat], rcp::iso21434::RiskLevel) -> alloc::vec::Vec<&rcp::iso21434::Threat> pub fn rcp::iso21434::risk_level(rcp::iso21434::Feasibility, rcp::iso21434::Impact) -> rcp::iso21434::RiskLevel +pub mod rcp::l2 +pub struct rcp::l2::L2Transport +impl rcp::l2::L2Transport +pub fn rcp::l2::L2Transport::close(&self) -> core::result::Result<(), rcp::RcpError> +pub fn rcp::l2::L2Transport::dest_mac(&self) -> [u8; 6] +pub fn rcp::l2::L2Transport::local_stream(&self) -> rcp::avtp::StreamId +pub fn rcp::l2::L2Transport::new(rcp::avtp::StreamId, alloc::sync::Arc, [u8; 6], [u8; 6]) -> Self +pub fn rcp::l2::L2Transport::send_acf_abb(&self, &rcp::acf::AcfAbbMessage, u8, core::option::Option) -> core::result::Result +pub fn rcp::l2::L2Transport::send_acf_gbb(&self, &rcp::acf::AcfGbbMessage, u8, core::option::Option) -> core::result::Result +pub fn rcp::l2::L2Transport::src_mac(&self) -> [u8; 6] +impl core::marker::Freeze for rcp::l2::L2Transport +impl core::marker::Send for rcp::l2::L2Transport +impl core::marker::Sync for rcp::l2::L2Transport +impl core::marker::Unpin for rcp::l2::L2Transport +impl core::marker::UnsafeUnpin for rcp::l2::L2Transport +impl !core::panic::unwind_safe::RefUnwindSafe for rcp::l2::L2Transport +impl !core::panic::unwind_safe::UnwindSafe for rcp::l2::L2Transport +pub struct rcp::l2::RawEthernetSocket +impl rcp::l2::RawEthernetSocket +pub fn rcp::l2::RawEthernetSocket::bind(&str) -> core::result::Result +pub fn rcp::l2::RawEthernetSocket::mac(&self) -> [u8; 6] +impl core::fmt::Debug for rcp::l2::RawEthernetSocket +pub fn rcp::l2::RawEthernetSocket::fmt(&self, &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl rcp::l2::L2Socket for rcp::l2::RawEthernetSocket +pub fn rcp::l2::RawEthernetSocket::recv(&self, core::option::Option) -> core::result::Result, rcp::RcpError> +pub fn rcp::l2::RawEthernetSocket::send(&self, &[u8]) -> core::result::Result +impl core::marker::Freeze for rcp::l2::RawEthernetSocket +impl core::marker::Send for rcp::l2::RawEthernetSocket +impl core::marker::Sync for rcp::l2::RawEthernetSocket +impl core::marker::Unpin for rcp::l2::RawEthernetSocket +impl core::marker::UnsafeUnpin for rcp::l2::RawEthernetSocket +impl core::panic::unwind_safe::RefUnwindSafe for rcp::l2::RawEthernetSocket +impl core::panic::unwind_safe::UnwindSafe for rcp::l2::RawEthernetSocket +pub const rcp::l2::ETHERTYPE_AVTP: u16 +pub trait rcp::l2::L2Socket: core::marker::Send + core::marker::Sync +pub fn rcp::l2::L2Socket::recv(&self, core::option::Option) -> core::result::Result, rcp::RcpError> +pub fn rcp::l2::L2Socket::send(&self, &[u8]) -> core::result::Result +impl rcp::l2::L2Socket for rcp::l2::RawEthernetSocket +pub fn rcp::l2::RawEthernetSocket::recv(&self, core::option::Option) -> core::result::Result, rcp::RcpError> +pub fn rcp::l2::RawEthernetSocket::send(&self, &[u8]) -> core::result::Result +pub fn rcp::l2::decode_ethernet_frame(&[u8]) -> core::result::Result, rcp::RcpError> +pub fn rcp::l2::encode_ethernet_frame([u8; 6], [u8; 6], &[u8]) -> alloc::vec::Vec +pub type rcp::l2::DecodedEthernetFrame<'a> = ([u8; 6], [u8; 6], &'a [u8]) pub mod rcp::lifecycle pub enum rcp::lifecycle::LockPolicy pub rcp::lifecycle::LockPolicy::W @@ -4014,6 +4057,21 @@ impl core::marker::Unpin for rcp::udp::ResolvedEndpoint impl core::marker::UnsafeUnpin for rcp::udp::ResolvedEndpoint impl core::panic::unwind_safe::RefUnwindSafe for rcp::udp::ResolvedEndpoint impl core::panic::unwind_safe::UnwindSafe for rcp::udp::ResolvedEndpoint +pub struct rcp::udp::StdUdpSocket +impl rcp::udp::StdUdpSocket +pub fn rcp::udp::StdUdpSocket::bind(core::net::socket_addr::SocketAddr) -> core::result::Result +pub fn rcp::udp::StdUdpSocket::local_addr(&self) -> core::result::Result +pub fn rcp::udp::StdUdpSocket::new_default_port(core::net::ip_addr::IpAddr) -> core::result::Result +impl rcp::udp::UdpSocket for rcp::udp::StdUdpSocket +pub fn rcp::udp::StdUdpSocket::recv_from(&self, core::option::Option) -> core::result::Result<(alloc::vec::Vec, core::net::socket_addr::SocketAddr), rcp::RcpError> +pub fn rcp::udp::StdUdpSocket::send_to(&self, &[u8], core::net::socket_addr::SocketAddr) -> core::result::Result +impl !core::marker::Freeze for rcp::udp::StdUdpSocket +impl core::marker::Send for rcp::udp::StdUdpSocket +impl core::marker::Sync for rcp::udp::StdUdpSocket +impl core::marker::Unpin for rcp::udp::StdUdpSocket +impl core::marker::UnsafeUnpin for rcp::udp::StdUdpSocket +impl core::panic::unwind_safe::RefUnwindSafe for rcp::udp::StdUdpSocket +impl core::panic::unwind_safe::UnwindSafe for rcp::udp::StdUdpSocket pub struct rcp::udp::UdpRcServer impl rcp::udp::UdpRcServer pub fn rcp::udp::UdpRcServer::discovery_claim(&self) -> core::option::Option @@ -4042,9 +4100,16 @@ impl core::marker::Unpin for rcp::udp::UdpTransport impl core::marker::UnsafeUnpin for rcp::udp::UdpTransport impl !core::panic::unwind_safe::RefUnwindSafe for rcp::udp::UdpTransport impl !core::panic::unwind_safe::UnwindSafe for rcp::udp::UdpTransport +pub const rcp::udp::ANNEX_J_CONTINUOUS_PORT: u16 +pub const rcp::udp::ANNEX_J_CONTROL_PORT: u16 pub trait rcp::udp::UdpSocket: core::marker::Send + core::marker::Sync pub fn rcp::udp::UdpSocket::recv_from(&self, core::option::Option) -> core::result::Result<(alloc::vec::Vec, core::net::socket_addr::SocketAddr), rcp::RcpError> pub fn rcp::udp::UdpSocket::send_to(&self, &[u8], core::net::socket_addr::SocketAddr) -> core::result::Result +impl rcp::udp::UdpSocket for rcp::udp::StdUdpSocket +pub fn rcp::udp::StdUdpSocket::recv_from(&self, core::option::Option) -> core::result::Result<(alloc::vec::Vec, core::net::socket_addr::SocketAddr), rcp::RcpError> +pub fn rcp::udp::StdUdpSocket::send_to(&self, &[u8], core::net::socket_addr::SocketAddr) -> core::result::Result +pub fn rcp::udp::decode_annex_j_udp_payload(&[u8]) -> core::result::Result<(u32, &[u8]), rcp::RcpError> +pub fn rcp::udp::encode_annex_j_udp_payload(u32, &[u8]) -> alloc::vec::Vec pub fn rcp::udp::resolve_endpoint(&rcp::addressing::EndpointTable, rcp::avtp::StreamId, u16) -> core::result::Result pub mod rcp::wakeup pub struct rcp::wakeup::SleepCmdRequest diff --git a/docs/SEMVER.md b/docs/SEMVER.md index ebea37a..3fc6726 100644 --- a/docs/SEMVER.md +++ b/docs/SEMVER.md @@ -88,7 +88,7 @@ wire-codec modules (`can`, `lin`, `gpio`, `i2c`, `spi`, `uart`, `adc`, (`ratelimit`, `deadline`, `faultinject`, `proxy`, `redundancy`, `observe`, `authz`, `record`, `loan`, `admin`, `powerstate`, `watchdog`, `evtgroup`, `federation`, `dyndata`, `config`), and the transport bridges (`udp`, -`tlstransport`, `shmem`, `mdns`). These get the same semver *mechanics* as +`l2`, `tlstransport`, `shmem`, `mdns`). These get the same semver *mechanics* as Tier 1 (a breaking change here is still a MAJOR bump post-`v1.0.0`) but are individually newer and less exercised end-to-end than Tier 1, so expect more of them to gain `#[non_exhaustive]`/other stability annotations as diff --git a/src/bin/rcp.rs b/src/bin/rcp.rs index e7af3bf..f6d6c20 100644 --- a/src/bin/rcp.rs +++ b/src/bin/rcp.rs @@ -45,14 +45,13 @@ //! [--initial ] [--read-size ] [--format json] //! rust-rcp endpoint write --bus-id --payload [--stream ] //! [--ep-type ] [--initial ] +//! rust-rcp serve --udp [--port ] [--stream ] +//! [--max-requests ] //! //! ## Provenance note //! -//! This crate has no concrete `rcp::udp::UdpSocket` implementation over a -//! real OS socket — only the in-process [`rcp::mock::RcServer`] test -//! double and `rcp::udp`'s own unit-test fakes exist. `discover`/ -//! `register`/`endpoint` therefore each construct and address a fresh -//! in-process `RcServer` for the lifetime of one invocation, the same +//! `discover`/`register`/`endpoint` each construct and address a fresh +//! in-process [`RcServer`] for the lifetime of one invocation, the same //! ephemeral-server discipline this file's pre-Milestone-10 `send`/ //! `status --zone` commands already used against a fresh //! `rcp::mock::MockRegistry` each invocation — not a regression this item @@ -61,9 +60,25 @@ //! [`GeneralRegisters::default`] and an empty endpoint table; there is no //! state carried between separate `rust-rcp` invocations. This is flagged //! here per Guiding Principle 5 rather than left an unstated limitation. +//! +//! `serve` (added alongside [`rcp::udp::StdUdpSocket`]) is this binary's +//! first command backed by a real OS socket rather than an in-process +//! `RcServer` invoked directly: it runs [`rcp::udp::UdpRcServer`] — +//! previously only ever exercised in this crate's own unit tests, against +//! mock sockets — over a real, bound [`rcp::udp::StdUdpSocket`], serving +//! real inbound UDP datagrams until `--max-requests` is reached (default: +//! unbounded, run until a fatal socket error). Before this item, this +//! crate had no concrete `rcp::udp::UdpSocket` implementation over a real +//! OS socket at all — only the in-process [`rcp::mock::RcServer`] test +//! double and `rcp::udp`'s own unit-test fakes existed, and every command +//! above predates that gap closing. `discover`/`register`/`endpoint` +//! remain deliberately ephemeral/in-process per the note above; `serve` is +//! the new, real-network-facing counterpart, not a replacement for them. use std::io::Read; use std::process; +use std::sync::Arc; +use std::time::Instant; use base64::engine::general_purpose::STANDARD as BASE64; use base64::Engine; @@ -73,6 +88,7 @@ use rcp::discovery; use rcp::ep0::EP0_BYTE_BUS_ID; use rcp::mock::{MockEndpoint, RcServer}; use rcp::regmap::{EndpointType, GeneralRegisters}; +use rcp::udp::{StdUdpSocket, UdpRcServer}; const TOOL: &str = "rust-rcp"; const PROTOCOL: &str = "RCP"; @@ -83,7 +99,9 @@ fn main() { if args.len() < 2 { eprintln!("Usage: rust-rcp [options]"); - eprintln!("Commands: version, capabilities, status, convert, discover, register, endpoint"); + eprintln!( + "Commands: version, capabilities, status, convert, discover, register, endpoint, serve" + ); process::exit(1); } @@ -253,6 +271,10 @@ fn main() { // fusa:req REQ-CLI-005 "endpoint" => cmd_endpoint(&args), + // ── serve ───────────────────────────────────────────────────────────── + // fusa:req REQ-CLI-010 + "serve" => cmd_serve(&args), + cmd => { eprintln!("unknown command: {}", cmd); process::exit(1); @@ -589,6 +611,89 @@ fn cmd_endpoint_write(args: &[String]) { } } +// ── serve ──────────────────────────────────────────────────────────────────── + +/// `rust-rcp serve --udp [--port ] [--stream ] +/// [--max-requests ]`. +/// +/// Binds a real [`StdUdpSocket`] to `--udp:--port` (default +/// [`rcp::udp::ANNEX_J_CONTROL_PORT`]) and runs [`UdpRcServer`] against it +/// — a fresh [`RcServer`] with [`GeneralRegisters::default`] and an empty +/// endpoint table, the same starting state `discover`/`register`/ +/// `endpoint` already use (see this file's own "Provenance note") — until +/// `--max-requests` requests have been served (default: unbounded; runs +/// until a fatal socket error). Each served request is logged to stdout. +/// +/// This is the first `rust-rcp` command to talk to a real OS socket rather +/// than dispatching directly against an in-process `RcServer` — see this +/// file's own module doc comment. +// fusa:req REQ-CLI-010 +fn cmd_serve(args: &[String]) { + let bind_ip_str = match flag_value(args, "--udp") { + Some(ip) => ip, + None => { + eprintln!("error: --udp required (e.g. --udp 0.0.0.0)"); + process::exit(1); + } + }; + let bind_ip: std::net::IpAddr = match bind_ip_str.parse() { + Ok(ip) => ip, + Err(e) => { + eprintln!("error: invalid --udp address {bind_ip_str:?}: {e}"); + process::exit(1); + } + }; + let port = parse_u16_arg(args, "--port").unwrap_or(rcp::udp::ANNEX_J_CONTROL_PORT); + let local_stream = parse_stream_arg(args, "--stream").unwrap_or_else(|| StreamId::from_u64(0)); + let max_requests = parse_u32_arg(args, "--max-requests"); + + let socket = match StdUdpSocket::bind(std::net::SocketAddr::new(bind_ip, port)) { + Ok(s) => s, + Err(e) => { + eprintln!("error: {}", e); + process::exit(3); + } + }; + let bound_addr = socket + .local_addr() + .map(|a| a.to_string()) + .unwrap_or_else(|_| format!("{bind_ip}:{port}")); + + let rc_server = RcServer::new(GeneralRegisters::default()); + let server = UdpRcServer::new(local_stream, Arc::new(socket), rc_server); + + println!( + "serve: listening on udp {} (stream={})", + bound_addr, + format_stream_hex(local_stream) + ); + + let mut served: u32 = 0; + loop { + if let Some(max) = max_requests { + if served >= max { + println!("serve: reached --max-requests={max}, stopping"); + break; + } + } + match server.serve_one( + None, + served as u8, + Instant::now(), + discovery::DISCOVERY_TIME_OUT, + ) { + Ok(()) => { + served += 1; + println!("serve: dispatched request #{served}"); + } + Err(e) => { + eprintln!("serve: fatal error after {served} request(s): {e}"); + process::exit(3); + } + } + } +} + // ── §11.2 / §15.5 rcp.Message → relay.Message conversion ──────────────────── // // RELAY spec §11.2's `convert` driver reads one canonical-type value for the @@ -756,6 +861,10 @@ fn parse_u8_arg(args: &[String], flag: &str) -> Option { flag_value(args, flag).and_then(|v| v.parse().ok()) } +fn parse_u32_arg(args: &[String], flag: &str) -> Option { + flag_value(args, flag).and_then(|v| v.parse().ok()) +} + fn parse_hex_arg(args: &[String], flag: &str) -> Option> { flag_value(args, flag).map(|v| { (0..v.len()) @@ -823,6 +932,20 @@ mod tests { assert_eq!(parse_u16_arg(&args, "--bus-id"), Some(42u16)); } + #[test] + // fusa:test REQ-CLI-010 + fn parse_u32_arg_parses_decimal() { + let args: Vec = vec!["rcp".into(), "--max-requests".into(), "5".into()]; + assert_eq!(parse_u32_arg(&args, "--max-requests"), Some(5u32)); + } + + #[test] + // fusa:test REQ-CLI-010 + fn parse_u32_arg_absent_returns_none() { + let args: Vec = vec!["rcp".into(), "serve".into()]; + assert!(parse_u32_arg(&args, "--max-requests").is_none()); + } + #[test] // fusa:test REQ-CLI-002 fn parse_u8_arg_parses_transaction() { diff --git a/src/l2.rs b/src/l2.rs new file mode 100644 index 0000000..204b750 --- /dev/null +++ b/src/l2.rs @@ -0,0 +1,804 @@ +// fusa:req REQ-L2-001 +// fusa:req REQ-L2-002 +// fusa:req REQ-L2-003 +// fusa:req REQ-L2-004 +// fusa:req REQ-L2-005 +// fusa:req REQ-L2-006 +// fusa:req REQ-L2-007 +// fusa:req REQ-L2-008 + +//! Layer-2 (raw Ethernet) transport for the TC18 AVTPDU/ACF wire format. +//! +//! TC18 §10.1: "[IEEE1722] can be used as a layer-2 protocol, which is +//! independent from the physical layer below... an AVTPDU is marked by an +//! EtherType value of 0x22F0." This is a second, wire-incompatible +//! transport option alongside [`crate::udp`]'s IEEE 1722-2016 Annex J +//! UDP/IP encapsulation — not an alternative socket API over the same +//! bytes. Frame layout: destination MAC (6 bytes) + source MAC (6 bytes) + +//! EtherType (2 bytes, big-endian, [`ETHERTYPE_AVTP`]) + the AVTPDU bytes +//! directly, with **no** 4-byte encapsulation sequence number — +//! [`crate::udp::encode_annex_j_udp_payload`]'s own field exists only for +//! the Annex J UDP/IP encapsulation and has no L2 counterpart. See +//! [`encode_ethernet_frame`]/[`decode_ethernet_frame`]. +//! +//! Before this item, this crate had no layer-2 transport of any kind — the +//! same gap every other RCP-family repo (`go-RCP`, `cpp-RCP`, `c-RCP`) had +//! at the time this item was scoped. +//! +//! Mirrors [`crate::udp`]'s own `UdpSocket`/`UdpTransport` abstraction one +//! wire layer down: [`L2Socket`] is the `UdpSocket` analog (`SocketAddr` +//! replaced by a raw `[u8; 6]` MAC address), and [`L2Transport`] is the +//! `UdpTransport` analog (`send_acf_abb`/`send_acf_gbb`, the same +//! echo-back-verified request/response client shape). On `target_os = +//! "linux"`, [`RawEthernetSocket`] is a real, production [`L2Socket`] over +//! an `AF_PACKET`/`SOCK_RAW` socket; every other target gets a same-named +//! stub whose constructor always errors (see "Non-Linux platforms" below). +//! +//! Server-side dispatch — an `L2RcServer` mirroring +//! [`crate::udp::UdpRcServer`]'s register-map-driven request dispatch — is +//! intentionally out of scope for this item. Flagged here per Guiding +//! Principle 5 as a real, deliberate scope limit rather than an oversight: +//! this item's server-facing wiring is [`crate::udp::UdpRcServer`] run +//! over [`crate::udp::StdUdpSocket`] (see `src/bin/rcp.rs`'s `serve` +//! command) — building a second, independent copy of `UdpRcServer`'s +//! discovery/dispatch logic against `L2Socket` instead of duplicating it +//! is a follow-up, not bundled silently into this one. +//! +//! # Why `nix`, not raw `libc` `unsafe` syscalls — a flagged judgment call +//! +//! Per Guiding Principle 5: this crate is `#![forbid(unsafe_code)]` +//! crate-wide (`src/lib.rs`) — `src/capi.rs`'s own doc comment already +//! flags that this rule is the reason this crate has never built a real +//! raw-pointer FFI boundary. `forbid` cannot be locally overridden by an +//! inner `#[allow(unsafe_code)]` (attempting to do so is itself a compile +//! error, E0453), so a raw `socket()`/`bind()`/`sendto()`/`recvfrom()` +//! implementation directly against `libc` — which would require `unsafe +//! extern "C"` calls written in this crate's own source — is not an +//! option here at all, not merely a style preference this item chose +//! against. +//! +//! [`RawEthernetSocket`] is instead built on the [`nix`](https://docs.rs/nix) +//! crate (a new dependency — not already present in `Cargo.toml`, gated to +//! `target_os = "linux"` only), whose `socket`/`bind`/`sendto`/`recvfrom`/ +//! `setsockopt`/`getifaddrs` functions are all safe Rust `fn`s, not +//! `unsafe fn`s — the `unsafe` needed to actually call into `libc` lives +//! inside `nix`'s own crate, never in this one. This was confirmed against +//! `nix` 0.31's own published API (docs.rs) before writing this module, +//! not assumed. `nix` is a narrowly-scoped POSIX-bindings crate, not a +//! heavyweight packet-crafting framework like `pnet`, matching this item's +//! own minimal-footprint intent for a Linux-only transport. +//! +//! # Runtime requirement +//! +//! [`RawEthernetSocket::bind`] opens an `AF_PACKET`/`SOCK_RAW` socket, +//! which the Linux kernel only permits to a process holding `CAP_NET_RAW` +//! (or running as root). This is a real operational requirement of raw +//! packet sockets themselves, not an artifact of this module's design — +//! every caller needs one of those two. The CI job that exercises this +//! module against a real interface (`.github/workflows/ci.yml`, the +//! `l2-veth` job) runs under `sudo` accordingly. +//! +//! # Non-Linux platforms +//! +//! [`RawEthernetSocket`] does not exist as a raw-socket implementation +//! outside `target_os = "linux"` — `AF_PACKET` is a Linux-specific +//! facility. A stub of the same name is compiled in for every other +//! target instead, whose `bind` always returns a clear `Err` explaining +//! why, rather than silently no-op-ing — so the rest of this crate (and +//! any downstream caller) can reference `crate::l2::RawEthernetSocket` +//! unconditionally, without its own `#[cfg(...)]` gate. + +use std::sync::Arc; +use std::time::Duration; + +use crate::acf::{self, AcfAbbMessage, AcfGbbMessage}; +use crate::avtp::{self, StreamId}; +use crate::RcpError; + +// ── Ethernet II framing (pure functions — no socket) ─────────────────────────── + +/// IEEE 802 EtherType assigned to IEEE 1722 (AVTP) — TC18 §10.1: "an +/// AVTPDU is marked by an EtherType value of 0x22F0." Sent in place of, +/// never alongside, [`crate::udp::encode_annex_j_udp_payload`]'s 4-byte +/// encapsulation sequence number — see this module's own doc comment. +// fusa:req REQ-L2-001 +pub const ETHERTYPE_AVTP: u16 = 0x22F0; + +/// Ethernet II header length: 6-byte destination MAC + 6-byte source MAC + +/// 2-byte EtherType. +const ETHERNET_HEADER_LEN: usize = 14; + +/// Encode a raw Ethernet II frame carrying `avtpdu`: `dest_mac` + +/// `src_mac` + [`ETHERTYPE_AVTP`] (big-endian) + `avtpdu` directly, with +/// no encapsulation sequence number — see this module's own doc comment. +// fusa:req REQ-L2-001 +// fusa:req REQ-L2-002 +pub fn encode_ethernet_frame(dest_mac: [u8; 6], src_mac: [u8; 6], avtpdu: &[u8]) -> Vec { + let mut frame = Vec::with_capacity(ETHERNET_HEADER_LEN + avtpdu.len()); + frame.extend_from_slice(&dest_mac); + frame.extend_from_slice(&src_mac); + frame.extend_from_slice(ÐERTYPE_AVTP.to_be_bytes()); + frame.extend_from_slice(avtpdu); + frame +} + +/// `(dest_mac, src_mac, avtpdu_bytes)` — [`decode_ethernet_frame`]'s +/// return shape, named so its signature stays readable. +pub type DecodedEthernetFrame<'a> = ([u8; 6], [u8; 6], &'a [u8]); + +/// The inverse of [`encode_ethernet_frame`]: `(dest_mac, src_mac, +/// avtpdu_bytes)`. +/// +/// `Err(RcpError::ShortFrame)` for fewer than 14 bytes. `Err(RcpError:: +/// Other(_))` if the EtherType field is not [`ETHERTYPE_AVTP`] — such a +/// frame is real Ethernet traffic this transport is simply not addressed +/// to decode (e.g. ARP, IPv4/IPv6), not a malformed AVTPDU. +// fusa:req REQ-L2-001 +// fusa:req REQ-L2-002 +pub fn decode_ethernet_frame(frame: &[u8]) -> Result, RcpError> { + if frame.len() < ETHERNET_HEADER_LEN { + return Err(RcpError::ShortFrame); + } + let mut dest_mac = [0u8; 6]; + let mut src_mac = [0u8; 6]; + dest_mac.copy_from_slice(&frame[0..6]); + src_mac.copy_from_slice(&frame[6..12]); + let ethertype = u16::from_be_bytes([frame[12], frame[13]]); + if ethertype != ETHERTYPE_AVTP { + return Err(RcpError::Other(format!( + "l2: unexpected EtherType 0x{ethertype:04x}, expected 0x{ETHERTYPE_AVTP:04x}" + ))); + } + Ok((dest_mac, src_mac, &frame[ETHERNET_HEADER_LEN..])) +} + +// ── L2Socket trait ─────────────────────────────────────────────────────────── + +/// Abstract raw-Ethernet socket for testability — the L2 analog of +/// [`crate::udp::UdpSocket`], addressed by a `[u8; 6]` MAC rather than a +/// `SocketAddr`. `send`/`recv` operate on already-framed +/// ([`encode_ethernet_frame`]/[`decode_ethernet_frame`]) wire bytes, the +/// same "callers see only already-framed bytes" contract +/// [`crate::udp::UdpSocket`] documents. +/// +/// `Some(Duration::ZERO)` passed to [`Self::recv`] is not guaranteed to +/// mean "return immediately without blocking" — [`RawEthernetSocket`]'s +/// own implementation uses `SO_RCVTIMEO`, whose own POSIX-defined +/// zero-value means "block indefinitely," not "poll." A caller wanting a +/// true immediate-return poll should not rely on this method for that; +/// [`L2Transport::send_acf_abb`]/[`L2Transport::send_acf_gbb`] avoid the +/// ambiguity entirely by special-casing `Duration::ZERO` themselves before +/// ever reaching this method, the same discipline +/// [`crate::udp::UdpTransport::send_acf_abb`] uses. +// fusa:req REQ-L2-003 +pub trait L2Socket: Send + Sync { + /// Send `frame` (a full Ethernet frame — see [`encode_ethernet_frame`]) + /// out this socket's bound interface. + fn send(&self, frame: &[u8]) -> Result; + + /// Receive one Ethernet frame, waiting up to `timeout` (`None` blocks + /// indefinitely) — see this trait's own doc comment for + /// `Some(Duration::ZERO)`'s caveat. + fn recv(&self, timeout: Option) -> Result, RcpError>; +} + +// ── L2Transport ────────────────────────────────────────────────────────────── + +/// RCP-over-L2 transport, mirroring [`crate::udp::UdpTransport`] one wire +/// layer down: addressed by `local_stream` ([`StreamId`]) plus a +/// caller-supplied `dest_mac` (unicast or multicast — this crate does not +/// derive/allocate a multicast MAC of its own; that algorithm lives in the +/// base IEEE 1722 standard, not available to this crate, so it is always a +/// caller input, never computed here) and `src_mac` (this transport's own +/// address, used to build every outgoing frame's Ethernet header — see +/// [`RawEthernetSocket::bind`] for how a real caller obtains its +/// interface's own MAC without supplying one itself). +// fusa:req REQ-L2-004 +pub struct L2Transport { + local_stream: StreamId, + socket: Arc, + dest_mac: [u8; 6], + src_mac: [u8; 6], +} + +impl L2Transport { + /// Construct a transport bound to `local_stream`, sending to + /// `dest_mac` from `src_mac` over `socket`. + pub fn new( + local_stream: StreamId, + socket: Arc, + dest_mac: [u8; 6], + src_mac: [u8; 6], + ) -> Self { + L2Transport { + local_stream, + socket, + dest_mac, + src_mac, + } + } + + /// This transport's local [`StreamId`]. + pub fn local_stream(&self) -> StreamId { + self.local_stream + } + + /// The destination MAC address every outgoing frame is addressed to. + pub fn dest_mac(&self) -> [u8; 6] { + self.dest_mac + } + + /// The source MAC address every outgoing frame is sent from. + pub fn src_mac(&self) -> [u8; 6] { + self.src_mac + } + + /// Send an ACF_ABB request wrapped in an NTSCF frame addressed under + /// `local_stream`, framed as a raw Ethernet II frame + /// ([`encode_ethernet_frame`]), and decode the ACF_ABB response, + /// verifying it echoes the request's `byte_bus_id` + /// ([`crate::acf::verify_echo_back`]) — the same request/response + /// shape as [`crate::udp::UdpTransport::send_acf_abb`], one wire layer + /// down. + /// + /// Returns `Err(RcpError::Timeout)` immediately for a zero `timeout`, + /// matching [`crate::udp::UdpTransport::send_acf_abb`]'s own + /// discipline. + // fusa:req REQ-L2-005 + // fusa:req REQ-L2-006 + pub fn send_acf_abb( + &self, + msg: &AcfAbbMessage, + sequence_num: u8, + timeout: Option, + ) -> Result { + if timeout == Some(Duration::ZERO) { + return Err(RcpError::Timeout); + } + let payload = acf::encode_acf_abb(msg)?; + let ntscf = avtp::encode_ntscf_frame(self.local_stream, sequence_num, &payload)?; + let frame = encode_ethernet_frame(self.dest_mac, self.src_mac, &ntscf); + self.socket.send(&frame)?; + let resp_frame = self.socket.recv(timeout)?; + let (_dest, _src, resp_ntscf) = decode_ethernet_frame(&resp_frame)?; + let (_, resp_payload) = avtp::decode_ntscf_frame(resp_ntscf)?; + let resp = acf::decode_acf_abb(resp_payload)?; + acf::verify_echo_back(&msg.info, &resp.info)?; + Ok(resp) + } + + /// Same as [`Self::send_acf_abb`], for an ACF_GBB request/response + /// pair. + // fusa:req REQ-L2-005 + // fusa:req REQ-L2-006 + pub fn send_acf_gbb( + &self, + msg: &AcfGbbMessage, + sequence_num: u8, + timeout: Option, + ) -> Result { + if timeout == Some(Duration::ZERO) { + return Err(RcpError::Timeout); + } + let payload = acf::encode_acf_gbb(msg)?; + let ntscf = avtp::encode_ntscf_frame(self.local_stream, sequence_num, &payload)?; + let frame = encode_ethernet_frame(self.dest_mac, self.src_mac, &ntscf); + self.socket.send(&frame)?; + let resp_frame = self.socket.recv(timeout)?; + let (_dest, _src, resp_ntscf) = decode_ethernet_frame(&resp_frame)?; + let (_, resp_payload) = avtp::decode_ntscf_frame(resp_ntscf)?; + let resp = acf::decode_acf_gbb(resp_payload)?; + acf::verify_echo_back(&msg.info, &resp.info)?; + Ok(resp) + } + + /// No-op, matching [`crate::udp::UdpTransport::close`]. + pub fn close(&self) -> Result<(), RcpError> { + Ok(()) + } +} + +// ── RawEthernetSocket — real production L2Socket ──────────────────────────── + +#[cfg(target_os = "linux")] +mod raw_socket { + use std::os::fd::{AsRawFd, OwnedFd}; + use std::time::Duration; + + use nix::ifaddrs::getifaddrs; + use nix::sys::socket::sockopt::ReceiveTimeout; + use nix::sys::socket::{ + bind, recvfrom, sendto, setsockopt, socket, AddressFamily, LinkAddr, MsgFlags, SockFlag, + SockProtocol, SockType, + }; + use nix::sys::time::TimeVal; + + use super::L2Socket; + use crate::RcpError; + + /// Real, production [`L2Socket`] over a Linux `AF_PACKET`/`SOCK_RAW` + /// socket bound to one named network interface. See `l2`'s own module + /// doc comment ("Why `nix`, not raw `libc` `unsafe` syscalls" and + /// "Runtime requirement") for the design rationale and privilege + /// requirement. + // fusa:req REQ-L2-007 + // fusa:req REQ-L2-008 + #[derive(Debug)] + pub struct RawEthernetSocket { + fd: OwnedFd, + bind_addr: LinkAddr, + mac: [u8; 6], + } + + impl RawEthernetSocket { + /// Open a raw `AF_PACKET`/`SOCK_RAW` socket and bind it to + /// `interface_name` (e.g. `"eth0"`). Requires `CAP_NET_RAW` (or + /// root) — see this module's "Runtime requirement" doc note; + /// `Err(RcpError::Other(_))` (not a panic) if that fails, or if + /// `interface_name` does not exist or has no link-layer address. + /// + /// The interface's own MAC address ([`Self::mac`]) is read from + /// the interface itself via `getifaddrs`, never supplied by the + /// caller — mirroring how [`crate::udp::StdUdpSocket::bind`] never + /// asks a caller for its own local IP address. + // fusa:req REQ-L2-007 + pub fn bind(interface_name: &str) -> Result { + let addrs = + getifaddrs().map_err(|e| RcpError::Other(format!("l2: getifaddrs: {e}")))?; + let link_addr = addrs + .filter(|ifa| ifa.interface_name == interface_name) + .find_map(|ifa| ifa.address.and_then(|a| a.as_link_addr().copied())) + .ok_or_else(|| { + RcpError::Other(format!( + "l2: interface {interface_name:?} not found, or has no AF_PACKET \ + link-layer address" + )) + })?; + + let mac = link_addr.addr().ok_or_else(|| { + RcpError::Other(format!( + "l2: interface {interface_name:?} has no MAC address (halen != 6)" + )) + })?; + + // SockProtocol::EthAll (ETH_P_ALL, htons(0x0003)) registers + // this socket for every EtherType, not just ETHERTYPE_AVTP — + // AF_PACKET sockets otherwise receive nothing at all (the + // `protocol` argument to the real socket(2) syscall is itself + // a packet filter, not merely descriptive). This module's own + // `recv` relies on `decode_ethernet_frame`'s EtherType check + // to reject anything that isn't ours, rather than filtering + // at the socket layer. + let fd = socket( + AddressFamily::Packet, + SockType::Raw, + SockFlag::empty(), + SockProtocol::EthAll, + ) + .map_err(|e| { + RcpError::Other(format!( + "l2: socket(AF_PACKET, SOCK_RAW): {e} (needs CAP_NET_RAW or root)" + )) + })?; + + bind(fd.as_raw_fd(), &link_addr) + .map_err(|e| RcpError::Other(format!("l2: bind {interface_name:?}: {e}")))?; + + Ok(RawEthernetSocket { + fd, + bind_addr: link_addr, + mac, + }) + } + + /// This interface's own MAC address, read from the OS at + /// [`Self::bind`] time. + pub fn mac(&self) -> [u8; 6] { + self.mac + } + + fn set_recv_timeout(&self, timeout: Option) -> Result<(), RcpError> { + // A zero TimeVal means "block indefinitely" (POSIX SO_RCVTIMEO + // semantics) — see L2Socket::recv's own doc comment for this + // caveat on `Some(Duration::ZERO)`. + let tv = match timeout { + Some(d) => TimeVal::new(d.as_secs() as i64, d.subsec_micros() as i64), + None => TimeVal::new(0, 0), + }; + setsockopt(&self.fd, ReceiveTimeout, &tv) + .map_err(|e| RcpError::Other(format!("l2: setsockopt(SO_RCVTIMEO): {e}"))) + } + } + + impl L2Socket for RawEthernetSocket { + /// Sends `frame` out this socket's bound interface. For an + /// `AF_PACKET`/`SOCK_RAW` socket, the destination address the + /// kernel actually transmits to is read from `frame`'s own + /// Ethernet header (already built by [`super::encode_ethernet_frame`]), + /// not from `sendto`'s own destination-address argument — only + /// that argument's interface index matters for a raw send, so + /// this reuses [`Self::bind_addr`], which already carries the + /// correct one. + // fusa:req REQ-L2-007 + fn send(&self, frame: &[u8]) -> Result { + sendto( + self.fd.as_raw_fd(), + frame, + &self.bind_addr, + MsgFlags::empty(), + ) + .map_err(|e| RcpError::Other(format!("l2: sendto: {e}"))) + } + + // fusa:req REQ-L2-007 + fn recv(&self, timeout: Option) -> Result, RcpError> { + self.set_recv_timeout(timeout)?; + let mut buf = [0u8; 65535]; + let (n, _peer) = recvfrom::(self.fd.as_raw_fd(), &mut buf).map_err(|e| { + if e == nix::errno::Errno::EAGAIN { + RcpError::Timeout + } else { + RcpError::Other(format!("l2: recvfrom: {e}")) + } + })?; + Ok(buf[..n].to_vec()) + } + } +} + +#[cfg(not(target_os = "linux"))] +mod raw_socket { + use std::time::Duration; + + use super::L2Socket; + use crate::RcpError; + + /// Non-Linux stub — see `l2`'s own module doc comment, "Non-Linux + /// platforms". [`Self::bind`] always fails explicitly; this type + /// exists at all only so `crate::l2::RawEthernetSocket` resolves on + /// every target. + // fusa:req REQ-L2-008 + #[derive(Debug)] + pub struct RawEthernetSocket { + _unconstructible: (), + } + + impl RawEthernetSocket { + /// Always returns `Err(RcpError::Other(_))` — `AF_PACKET` raw + /// sockets are a Linux-specific facility this crate has no + /// implementation of on this target. + // fusa:req REQ-L2-008 + pub fn bind(_interface_name: &str) -> Result { + Err(RcpError::Other( + "l2::RawEthernetSocket is only implemented on target_os = \"linux\" \ + (AF_PACKET/SOCK_RAW raw sockets are a Linux-specific facility); this \ + platform has no real L2Socket implementation" + .to_string(), + )) + } + + /// Never actually callable: [`Self::bind`] always errors on this + /// platform, so no value of this type can exist to call it on. + pub fn mac(&self) -> [u8; 6] { + unreachable!("RawEthernetSocket::bind always errors on this platform") + } + } + + impl L2Socket for RawEthernetSocket { + fn send(&self, _frame: &[u8]) -> Result { + unreachable!("RawEthernetSocket::bind always errors on this platform") + } + + fn recv(&self, _timeout: Option) -> Result, RcpError> { + unreachable!("RawEthernetSocket::bind always errors on this platform") + } + } +} + +pub use raw_socket::RawEthernetSocket; + +// ───────────────────────────────────────────────────────────────────────────── +// Tests +// ───────────────────────────────────────────────────────────────────────────── +#[cfg(test)] +mod tests { + use super::*; + use crate::acf::ByteMessageInfo; + use std::sync::Mutex; + + fn local_stream() -> StreamId { + StreamId::new([0x02, 0x11, 0x22, 0x33, 0x44, 0x55], 0x0001) + } + + // ── Ethernet II framing (pure byte manipulation, no socket) ─────────── + + #[test] + // fusa:test REQ-L2-001 + // fusa:test REQ-L2-002 + fn ethernet_frame_encode_decode_round_trips() { + let dest = [0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF]; + let src = [0x02, 0x11, 0x22, 0x33, 0x44, 0x55]; + let avtpdu = vec![0x01, 0x02, 0x03]; + let frame = encode_ethernet_frame(dest, src, &avtpdu); + assert_eq!(frame.len(), 14 + avtpdu.len()); + // Byte layout: dest || src || ethertype (BE) || payload. + assert_eq!(&frame[0..6], &dest); + assert_eq!(&frame[6..12], &src); + assert_eq!(&frame[12..14], &[0x22, 0xF0]); + assert_eq!(&frame[14..], avtpdu.as_slice()); + + let (d, s, payload) = decode_ethernet_frame(&frame).unwrap(); + assert_eq!(d, dest); + assert_eq!(s, src); + assert_eq!(payload, avtpdu.as_slice()); + } + + #[test] + // fusa:test REQ-L2-002 + fn ethernet_frame_encode_handles_empty_avtpdu() { + let frame = encode_ethernet_frame([0; 6], [0; 6], &[]); + assert_eq!(frame.len(), 14); + let (_, _, payload) = decode_ethernet_frame(&frame).unwrap(); + assert!(payload.is_empty()); + } + + #[test] + // fusa:test REQ-L2-002 + fn ethernet_frame_decode_rejects_short_frames() { + for len in 0..14 { + let buf = vec![0u8; len]; + let err = decode_ethernet_frame(&buf).unwrap_err(); + assert_eq!(err, RcpError::ShortFrame); + } + } + + #[test] + // fusa:test REQ-L2-002 + fn ethernet_frame_decode_rejects_wrong_ethertype() { + let mut frame = encode_ethernet_frame([0; 6], [0; 6], &[0xAA]); + // Corrupt the EtherType field to something real but not AVTP + // (0x0800 = IPv4). + frame[12] = 0x08; + frame[13] = 0x00; + let err = decode_ethernet_frame(&frame).unwrap_err(); + assert!(matches!(err, RcpError::Other(_))); + } + + // ── L2Transport (mocked L2Socket — no real socket, no privileges) ───── + + /// A mock socket that echoes back a well-formed ACF_ABB response, + /// copying `byte_bus_id` from whatever request it received unless + /// `mismatch` is set — the L2 analog of `udp`'s own `EchoUdp`. + struct EchoL2 { + mismatch: bool, + } + + impl L2Socket for EchoL2 { + fn send(&self, _frame: &[u8]) -> Result { + Ok(0) + } + + fn recv(&self, _timeout: Option) -> Result, RcpError> { + let byte_bus_id = if self.mismatch { 99 } else { 7 }; + let resp = AcfAbbMessage { + info: ByteMessageInfo { + byte_bus_id, + rsp: true, + ..Default::default() + }, + payload: vec![0xAA], + }; + let payload = acf::encode_acf_abb(&resp).unwrap(); + let ntscf = avtp::encode_ntscf_frame(local_stream(), 1, &payload).unwrap(); + Ok(encode_ethernet_frame( + [0x02, 0x11, 0x22, 0x33, 0x44, 0x55], + [0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF], + &ntscf, + )) + } + } + + /// A test double recording every frame handed to `send`, replaying + /// queued frames from `recv` — the L2 analog of `udp`'s own + /// `QueuedUdpSocket`. + struct QueuedL2 { + inbound: Mutex>>, + outbound: Mutex>>, + } + + impl QueuedL2 { + fn with_inbound(frames: Vec>) -> Arc { + Arc::new(Self { + inbound: Mutex::new(frames), + outbound: Mutex::new(Vec::new()), + }) + } + } + + impl L2Socket for QueuedL2 { + fn send(&self, frame: &[u8]) -> Result { + self.outbound.lock().unwrap().push(frame.to_vec()); + Ok(frame.len()) + } + + fn recv(&self, _timeout: Option) -> Result, RcpError> { + let mut inbound = self.inbound.lock().unwrap(); + if inbound.is_empty() { + Err(RcpError::Timeout) + } else { + Ok(inbound.remove(0)) + } + } + } + + fn request(byte_bus_id: u16) -> AcfAbbMessage { + AcfAbbMessage { + info: ByteMessageInfo { + byte_bus_id, + op: true, + ..Default::default() + }, + payload: vec![0x01, 0x02], + } + } + + const DEST_MAC: [u8; 6] = [0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF]; + const SRC_MAC: [u8; 6] = [0x02, 0x11, 0x22, 0x33, 0x44, 0x55]; + + #[test] + // fusa:test REQ-L2-004 + // fusa:test REQ-L2-005 + // fusa:test REQ-L2-006 + fn l2_send_acf_abb_round_trips_over_socket() { + let socket = Arc::new(EchoL2 { mismatch: false }); + let transport = L2Transport::new(local_stream(), socket, DEST_MAC, SRC_MAC); + let resp = transport.send_acf_abb(&request(7), 0, None).unwrap(); + assert_eq!(resp.info.byte_bus_id, 7); + assert!(resp.info.rsp); + } + + #[test] + // fusa:test REQ-L2-006 + fn l2_send_acf_abb_rejects_echo_back_mismatch() { + let socket = Arc::new(EchoL2 { mismatch: true }); + let transport = L2Transport::new(local_stream(), socket, DEST_MAC, SRC_MAC); + let err = transport.send_acf_abb(&request(7), 0, None).unwrap_err(); + assert_eq!(err, RcpError::EpError); + } + + #[test] + // fusa:test REQ-L2-005 + fn l2_send_acf_abb_rejects_zero_timeout() { + let socket = Arc::new(EchoL2 { mismatch: false }); + let transport = L2Transport::new(local_stream(), socket, DEST_MAC, SRC_MAC); + let err = transport + .send_acf_abb(&request(7), 0, Some(Duration::ZERO)) + .unwrap_err(); + assert_eq!(err, RcpError::Timeout); + } + + #[test] + // fusa:test REQ-L2-004 + fn l2_transport_getters_match_constructor() { + let socket = Arc::new(EchoL2 { mismatch: false }); + let sid = local_stream(); + let transport = L2Transport::new(sid, socket, DEST_MAC, SRC_MAC); + assert_eq!(transport.local_stream(), sid); + assert_eq!(transport.dest_mac(), DEST_MAC); + assert_eq!(transport.src_mac(), SRC_MAC); + assert!(transport.close().is_ok()); + } + + #[test] + // fusa:test REQ-L2-004 + // fusa:test REQ-L2-005 + fn l2_send_acf_gbb_round_trips_over_socket() { + struct EchoGbb; + impl L2Socket for EchoGbb { + fn send(&self, _frame: &[u8]) -> Result { + Ok(0) + } + fn recv(&self, _timeout: Option) -> Result, RcpError> { + let resp = AcfGbbMessage { + info: ByteMessageInfo { + byte_bus_id: 3, + rsp: true, + ..Default::default() + }, + message_timestamp: 0, + payload: vec![0x55], + }; + let payload = acf::encode_acf_gbb(&resp).unwrap(); + let ntscf = avtp::encode_ntscf_frame(local_stream(), 1, &payload).unwrap(); + Ok(encode_ethernet_frame(DEST_MAC, SRC_MAC, &ntscf)) + } + } + let socket = Arc::new(EchoGbb); + let transport = L2Transport::new(local_stream(), socket, DEST_MAC, SRC_MAC); + let msg = AcfGbbMessage { + info: ByteMessageInfo { + byte_bus_id: 3, + op: true, + ..Default::default() + }, + message_timestamp: 0, + payload: vec![0x01], + }; + let resp = transport.send_acf_gbb(&msg, 0, None).unwrap(); + assert_eq!(resp.info.byte_bus_id, 3); + assert_eq!(resp.payload, vec![0x55]); + } + + #[test] + // fusa:test REQ-L2-003 + // fusa:test REQ-L2-004 + fn l2_transport_send_records_the_real_ethernet_frame() { + let socket = QueuedL2::with_inbound(Vec::new()); + let transport = L2Transport::new(local_stream(), socket.clone(), DEST_MAC, SRC_MAC); + // Uses recv's Timeout error (empty queue) just to exercise send(); + // don't care about the response here. + let _ = transport.send_acf_abb(&request(7), 0, None); + + let sent = socket.outbound.lock().unwrap(); + assert_eq!(sent.len(), 1); + let (dest, src, _payload) = decode_ethernet_frame(&sent[0]).unwrap(); + assert_eq!(dest, DEST_MAC); + assert_eq!(src, SRC_MAC); + } + + // ── Non-Linux RawEthernetSocket stub ─────────────────────────────────── + + #[cfg(not(target_os = "linux"))] + #[test] + // fusa:test REQ-L2-008 + fn raw_ethernet_socket_bind_fails_explicitly_off_linux() { + let err = RawEthernetSocket::bind("eth0").unwrap_err(); + assert!(matches!(err, RcpError::Other(_))); + } + + // ── Real raw socket over a veth pair (Linux only, requires root/ + // CAP_NET_RAW; #[ignore]d by default — see .github/workflows/ + // ci.yml's `l2-veth` job, which sets up veth0/veth1 and runs this + // with `-- --ignored`) ────────────────────────────────────────────── + + #[cfg(target_os = "linux")] + #[test] + #[ignore = "requires root/CAP_NET_RAW and a pre-existing veth0/veth1 pair; see ci.yml's l2-veth job"] + // fusa:test REQ-L2-007 + fn real_raw_ethernet_socket_round_trips_a_frame_over_a_veth_pair() { + let tx = RawEthernetSocket::bind("veth0").expect("bind veth0 (needs sudo/CAP_NET_RAW)"); + let rx = RawEthernetSocket::bind("veth1").expect("bind veth1 (needs sudo/CAP_NET_RAW)"); + + let src_mac = tx.mac(); + let dest_mac = rx.mac(); + let avtpdu = vec![0xDE, 0xAD, 0xBE, 0xEF, 0x00, 0x01, 0x02, 0x03]; + let frame = encode_ethernet_frame(dest_mac, src_mac, &avtpdu); + + tx.send(&frame).expect("send over veth0"); + + // veth1 may also see other link-local traffic (e.g. NDP/ARP) on a + // freshly created interface; loop past anything that doesn't + // decode as our own EtherType/AVTPDU rather than assuming the very + // first received frame is ours. + let deadline = std::time::Instant::now() + Duration::from_secs(5); + loop { + assert!( + std::time::Instant::now() < deadline, + "did not observe our frame on veth1 within 5s" + ); + let received = rx + .recv(Some(Duration::from_secs(5))) + .expect("recv on veth1"); + if let Ok((d, s, payload)) = decode_ethernet_frame(&received) { + if d == dest_mac && s == src_mac { + assert_eq!( + payload, + avtpdu.as_slice(), + "frame must round-trip byte-for-byte" + ); + break; + } + } + } + } +} diff --git a/src/lib.rs b/src/lib.rs index 3542045..38dd80b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -96,6 +96,7 @@ pub mod gpio; pub mod i2c; pub mod iseled; pub mod iso21434; +pub mod l2; pub mod lifecycle; pub mod lin; pub mod loan; diff --git a/src/udp.rs b/src/udp.rs index f65303e..bf68181 100644 --- a/src/udp.rs +++ b/src/udp.rs @@ -9,6 +9,9 @@ // fusa:req REQ-UDP-009 // fusa:req REQ-UDP-010 // fusa:req REQ-UDP-011 +// fusa:req REQ-UDP-012 +// fusa:req REQ-UDP-013 +// fusa:req REQ-UDP-014 //! UDP unicast transport for the TC18 AVTPDU/ACF wire format. //! @@ -48,8 +51,47 @@ //! [`UdpTransport::send_acf_abb`]/[`UdpTransport::send_acf_gbb`] and //! [`resolve_endpoint`] are unchanged by this item — see their own doc //! comments. +//! +//! # Real OS-socket transport and IEEE 1722-2016 Annex J encapsulation +//! +//! Before this item, this module's only [`UdpSocket`] implementations were +//! in-process test doubles (`EchoUdp`/`QueuedUdpSocket`, both in this +//! module's own `#[cfg(test)]` block) — there was no concrete +//! implementation over a real OS socket anywhere in this crate; +//! `src/bin/rcp.rs`'s own pre-this-item doc comment said so explicitly. +//! [`StdUdpSocket`] closes that gap: this is the first real network I/O +//! this crate has ever shipped for RCP. +//! +//! TC18 §10.1 states AVTPDUs can be carried over UDP/IP, "Encapsulation of +//! 1722 frames in IP/UDP and port usage is described in Annex J" (of the +//! base IEEE 1722-2016 standard, not TC18 itself). This crate does not +//! have access to the paywalled IEEE 1722-2016 standard text; the framing +//! [`StdUdpSocket`]/[`encode_annex_j_udp_payload`]/ +//! [`decode_annex_j_udp_payload`] implement — a 4-byte big-endian +//! "encapsulation sequence number" prepended to every UDP payload before +//! the AVTPDU itself, and control-plane traffic (RCP requests/responses, +//! which this crate is exclusively concerned with) using destination port +//! [`ANNEX_J_CONTROL_PORT`] (17221), distinct from port 17220 for +//! "Continuous" streaming traffic ([`ANNEX_J_CONTINUOUS_PORT`]) — is taken +//! from two independent public secondary sources instead: a Wireshark +//! issue tracker discussion of the real Annex J framing, and the COVESA +//! Open1722 open-source reference implementation's `Avtp_Udp_t` header +//! struct (`include/avtp/Udp.h`, BSD-3-Clause, +//! ). This is flagged here per +//! Guiding Principle 5 as *not* independently verified against the +//! primary standard, rather than presented with false certainty. +//! +//! The encapsulation sequence number's exact intended receiver-side +//! semantics (e.g. loss detection) are not specified by either secondary +//! source consulted, and this crate does not invent any — [`StdUdpSocket`] +//! only guarantees it is monotonically increasing per sender, nothing +//! more. This field exists only for UDP/IP encapsulation (Annex J); it has +//! no counterpart when an AVTPDU is instead carried directly at layer 2 +//! with EtherType `0x22F0` — see [`crate::l2`], added alongside this item +//! as the other, equally-supported transport option TC18 §10.1 names. use std::net::SocketAddr; +use std::sync::atomic::{AtomicU32, Ordering}; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; @@ -72,6 +114,151 @@ pub trait UdpSocket: Send + Sync { fn recv_from(&self, timeout: Option) -> Result<(Vec, SocketAddr), RcpError>; } +// ── Annex J UDP encapsulation ──────────────────────────────────────────────── + +/// Standard destination UDP port for IEEE 1722-2016 Annex J "Discrete" +/// (control-plane) traffic — RCP requests/responses/acknowledgements are +/// control-plane traffic, so this is the applicable port for RCP-over-UDP +/// and [`StdUdpSocket::new_default_port`]'s default. See this module's own +/// doc comment, "Real OS-socket transport and IEEE 1722-2016 Annex J +/// encapsulation", for this constant's provenance (public secondary +/// sources, not the paywalled primary standard). +// fusa:req REQ-UDP-012 +pub const ANNEX_J_CONTROL_PORT: u16 = 17221; + +/// Standard destination UDP port for IEEE 1722-2016 Annex J "Continuous" +/// (streaming/periodic) traffic — not RCP's traffic class, and not used by +/// any constructor in this module; named here only so both of Annex J's +/// two standard ports are documented rather than one left unstated. Same +/// provenance note as [`ANNEX_J_CONTROL_PORT`]. +pub const ANNEX_J_CONTINUOUS_PORT: u16 = 17220; + +/// Prepend a 4-byte big-endian encapsulation sequence number to `avtpdu` +/// — TC18 §10.1's IEEE 1722-2016 Annex J UDP/IP encapsulation, per this +/// module's own doc comment provenance note. Byte order matches this +/// crate's existing big-endian wire convention (e.g. +/// [`crate::avtp::encode_ntscf_frame`]'s `stream_id` field, +/// [`crate::acf`]'s `message_timestamp`). +// fusa:req REQ-UDP-012 +pub fn encode_annex_j_udp_payload(seq: u32, avtpdu: &[u8]) -> Vec { + let mut buf = Vec::with_capacity(4 + avtpdu.len()); + buf.extend_from_slice(&seq.to_be_bytes()); + buf.extend_from_slice(avtpdu); + buf +} + +/// The inverse of [`encode_annex_j_udp_payload`]: split a raw UDP payload +/// into its 4-byte encapsulation sequence number and the AVTPDU bytes that +/// follow it. `Err(RcpError::ShortFrame)` for fewer than 4 bytes — never +/// panics on truncated or empty input. +// fusa:req REQ-UDP-012 +pub fn decode_annex_j_udp_payload(buf: &[u8]) -> Result<(u32, &[u8]), RcpError> { + if buf.len() < 4 { + return Err(RcpError::ShortFrame); + } + let mut seq_bytes = [0u8; 4]; + seq_bytes.copy_from_slice(&buf[..4]); + Ok((u32::from_be_bytes(seq_bytes), &buf[4..])) +} + +// ── StdUdpSocket ────────────────────────────────────────────────────────────── + +/// Real, production [`UdpSocket`] implementation over a bound +/// `std::net::UdpSocket`. See this module's own doc comment, "Real +/// OS-socket transport and IEEE 1722-2016 Annex J encapsulation", for full +/// context — this is the first concrete implementation of [`UdpSocket`] +/// over a real OS socket this crate has ever shipped. +/// +/// `send_to` prepends, and `recv_from` strips, the 4-byte encapsulation +/// sequence number [`encode_annex_j_udp_payload`]/ +/// [`decode_annex_j_udp_payload`] implement — entirely transparent to +/// [`UdpSocket`] trait callers ([`UdpTransport`], [`UdpRcServer`]), which +/// see only already-framed NTSCF/AVTPDU bytes, the same contract the +/// mock `EchoUdp`/`QueuedUdpSocket` test doubles already provide. +/// `send_to`'s sequence number is a per-`StdUdpSocket` monotonically +/// increasing counter, starting at 0 on construction; it is not exposed to +/// callers (see this module's own doc comment for why no receiver-side +/// semantics are attached to it). +// fusa:req REQ-UDP-013 +// fusa:req REQ-UDP-014 +pub struct StdUdpSocket { + socket: std::net::UdpSocket, + send_seq: AtomicU32, +} + +impl StdUdpSocket { + /// Bind a real UDP socket to `local_addr`. + // fusa:req REQ-UDP-013 + pub fn bind(local_addr: SocketAddr) -> Result { + let socket = std::net::UdpSocket::bind(local_addr) + .map_err(|e| RcpError::Other(format!("udp: bind {local_addr}: {e}")))?; + Ok(StdUdpSocket { + socket, + send_seq: AtomicU32::new(0), + }) + } + + /// Convenience constructor: bind to `bind_ip` on + /// [`ANNEX_J_CONTROL_PORT`] — the sensible default for RCP's + /// control-plane traffic. [`Self::bind`] remains available directly + /// for an explicit port (testing, or a deployment that cannot use the + /// standard port). + // fusa:req REQ-UDP-013 + pub fn new_default_port(bind_ip: std::net::IpAddr) -> Result { + Self::bind(SocketAddr::new(bind_ip, ANNEX_J_CONTROL_PORT)) + } + + /// The local address this socket is actually bound to — useful when + /// [`Self::bind`]'s `local_addr` used an ephemeral (`:0`) port. + pub fn local_addr(&self) -> Result { + self.socket + .local_addr() + .map_err(|e| RcpError::Other(format!("udp: local_addr: {e}"))) + } +} + +impl UdpSocket for StdUdpSocket { + /// Returns the number of bytes of `buf` (the caller-supplied + /// NTSCF/AVTPDU frame) sent — not the larger on-wire byte count + /// including the prepended encapsulation sequence number — matching + /// this trait's existing mock-implementation convention of echoing + /// `buf.len()` back rather than any wire-framing overhead. + // fusa:req REQ-UDP-013 + fn send_to(&self, buf: &[u8], addr: SocketAddr) -> Result { + let seq = self.send_seq.fetch_add(1, Ordering::Relaxed); + let framed = encode_annex_j_udp_payload(seq, buf); + let sent = self + .socket + .send_to(&framed, addr) + .map_err(|e| RcpError::Other(format!("udp: send_to {addr}: {e}")))?; + Ok(sent.saturating_sub(4)) + } + + /// `timeout` is applied via `SO_RCVTIMEO` on every call. `None` blocks + /// indefinitely. A real OS-level timeout is mapped to + /// `Err(RcpError::Timeout)`, matching every other timeout path in this + /// crate. + // fusa:req REQ-UDP-014 + fn recv_from(&self, timeout: Option) -> Result<(Vec, SocketAddr), RcpError> { + self.socket + .set_read_timeout(timeout) + .map_err(|e| RcpError::Other(format!("udp: set_read_timeout: {e}")))?; + let mut buf = [0u8; 65535]; + let (n, addr) = self.socket.recv_from(&mut buf).map_err(|e| { + if matches!( + e.kind(), + std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut + ) { + RcpError::Timeout + } else { + RcpError::Other(format!("udp: recv_from: {e}")) + } + })?; + let (_seq, avtpdu) = decode_annex_j_udp_payload(&buf[..n])?; + Ok((avtpdu.to_vec(), addr)) + } +} + // ── UdpTransport ───────────────────────────────────────────────────────────── /// RCP-over-UDP transport, addressed by `local_stream` @@ -486,6 +673,164 @@ mod tests { use super::*; use crate::acf::ByteMessageInfo; + // ── Annex J encapsulation (pure byte manipulation, no socket) ───────── + + #[test] + // fusa:test REQ-UDP-012 + fn annex_j_encode_decode_round_trips() { + let avtpdu = vec![0xDE, 0xAD, 0xBE, 0xEF, 0x01]; + let encoded = encode_annex_j_udp_payload(7, &avtpdu); + assert_eq!(encoded.len(), 4 + avtpdu.len()); + // Big-endian, matching this crate's existing wire convention. + assert_eq!(&encoded[..4], &[0x00, 0x00, 0x00, 0x07]); + let (seq, decoded) = decode_annex_j_udp_payload(&encoded).unwrap(); + assert_eq!(seq, 7); + assert_eq!(decoded, avtpdu.as_slice()); + } + + #[test] + // fusa:test REQ-UDP-012 + fn annex_j_encode_handles_empty_avtpdu() { + let encoded = encode_annex_j_udp_payload(0xFFFF_FFFF, &[]); + assert_eq!(encoded, vec![0xFF, 0xFF, 0xFF, 0xFF]); + let (seq, decoded) = decode_annex_j_udp_payload(&encoded).unwrap(); + assert_eq!(seq, 0xFFFF_FFFF); + assert!(decoded.is_empty()); + } + + #[test] + // fusa:test REQ-UDP-012 + fn annex_j_decode_rejects_short_buffers() { + for len in 0..4 { + let buf = vec![0u8; len]; + let err = decode_annex_j_udp_payload(&buf).unwrap_err(); + assert_eq!(err, RcpError::ShortFrame); + } + } + + #[test] + fn annex_j_control_and_continuous_ports_are_distinct_and_documented() { + assert_eq!(ANNEX_J_CONTROL_PORT, 17221); + assert_eq!(ANNEX_J_CONTINUOUS_PORT, 17220); + assert_ne!(ANNEX_J_CONTROL_PORT, ANNEX_J_CONTINUOUS_PORT); + } + + // ── StdUdpSocket (real loopback sockets — no privileges required) ───── + + #[test] + // fusa:test REQ-UDP-013 + // fusa:test REQ-UDP-014 + fn std_udp_socket_round_trips_over_real_loopback_socket() { + let a = StdUdpSocket::bind("127.0.0.1:0".parse().unwrap()).unwrap(); + let b = StdUdpSocket::bind("127.0.0.1:0".parse().unwrap()).unwrap(); + let b_addr = b.local_addr().unwrap(); + + let payload = vec![0x01, 0x02, 0x03, 0x04, 0x05]; + UdpSocket::send_to(&a, &payload, b_addr).unwrap(); + + let (received, _from) = UdpSocket::recv_from(&b, Some(Duration::from_secs(5))).unwrap(); + assert_eq!(received, payload); + } + + #[test] + // fusa:test REQ-UDP-013 + // fusa:test REQ-UDP-014 + fn std_udp_socket_and_udp_rc_server_serve_a_real_discovery_request_end_to_end() { + // The same composition `src/bin/rcp.rs`'s `serve` command builds + // (StdUdpSocket + UdpRcServer), but with both a real client and a + // real server talking over real loopback sockets — proving + // StdUdpSocket works end-to-end through UdpRcServer's own request + // dispatch, not just as a bare send/recv byte pipe. + use crate::regmap::GeneralRegisters; + + let server_stream = StreamId::new([0x02, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA], 0x00AA); + let server_socket = StdUdpSocket::bind("127.0.0.1:0".parse().unwrap()).unwrap(); + let server_addr = server_socket.local_addr().unwrap(); + let general = GeneralRegisters { + svr_vendor_id: 0x4242, + ..Default::default() + }; + let rc_server = RcServer::new(general); + let server = UdpRcServer::new(server_stream, Arc::new(server_socket), rc_server); + + let client_stream = local_stream(); + let client_socket = StdUdpSocket::bind("127.0.0.1:0".parse().unwrap()).unwrap(); + let request = discovery::build_discovery_request(0x11); + let payload = acf::encode_acf_abb(&request).unwrap(); + let frame = avtp::encode_ntscf_frame(client_stream, 0, &payload).unwrap(); + UdpSocket::send_to(&client_socket, &frame, server_addr).unwrap(); + + server + .serve_one( + Some(Duration::from_secs(5)), + 0, + Instant::now(), + discovery::DISCOVERY_TIME_OUT, + ) + .unwrap(); + + let (resp_frame, _from) = + UdpSocket::recv_from(&client_socket, Some(Duration::from_secs(5))).unwrap(); + let (hdr, acf_bytes) = avtp::decode_ntscf_frame(&resp_frame).unwrap(); + assert_eq!(StreamId::from_u64(hdr.stream_id), server_stream); + let resp = acf::decode_acf_abb(acf_bytes).unwrap(); + let regs = GeneralRegisters::decode(&resp.payload).unwrap(); + assert_eq!(regs.svr_vendor_id, 0x4242); + } + + #[test] + // fusa:test REQ-UDP-013 + fn std_udp_socket_send_seq_is_monotonically_increasing_on_the_wire() { + // Inspect the real encapsulated bytes with a plain std socket + // (bypassing StdUdpSocket's own recv_from, which strips the + // sequence number) to prove send_to's sequence counter actually + // increments on the wire, not just that decode_annex_j_udp_payload + // can parse whatever value is there. + let sender = StdUdpSocket::bind("127.0.0.1:0".parse().unwrap()).unwrap(); + let raw_receiver = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + let receiver_addr = raw_receiver.local_addr().unwrap(); + + for _ in 0..3 { + UdpSocket::send_to(&sender, &[0xAA], receiver_addr).unwrap(); + } + + let mut seqs = Vec::new(); + let mut buf = [0u8; 64]; + for _ in 0..3 { + let (n, _) = raw_receiver.recv_from(&mut buf).unwrap(); + let (seq, avtpdu) = decode_annex_j_udp_payload(&buf[..n]).unwrap(); + assert_eq!(avtpdu, &[0xAA]); + seqs.push(seq); + } + assert_eq!(seqs, vec![0, 1, 2]); + } + + #[test] + fn std_udp_socket_recv_from_times_out_on_a_real_socket() { + let socket = StdUdpSocket::bind("127.0.0.1:0".parse().unwrap()).unwrap(); + let err = socket + .recv_from(Some(Duration::from_millis(50))) + .unwrap_err(); + assert_eq!(err, RcpError::Timeout); + } + + #[test] + fn std_udp_socket_new_default_port_binds_annex_j_control_port() { + // Bind to an ephemeral port instead of the real 17221 so this test + // doesn't require exclusive access to a well-known port / root on + // some platforms; the constructor logic under test is the address + // construction itself, exercised via `bind` with an explicit + // ANNEX_J_CONTROL_PORT. + let addr: SocketAddr = format!("127.0.0.1:{ANNEX_J_CONTROL_PORT}").parse().unwrap(); + // A CI runner or developer machine may already have something + // bound to the real control port, or lack permission; either is + // an environment fact, not a bug in this constructor, so only the + // success case is asserted on. + if let Ok(socket) = StdUdpSocket::bind(addr) { + assert_eq!(socket.local_addr().unwrap().port(), ANNEX_J_CONTROL_PORT); + } + } + fn local_stream() -> StreamId { StreamId::new([0x02, 0x11, 0x22, 0x33, 0x44, 0x55], 0x0001) }