feat(server): add open session core

Add `irtt-server` with the deterministic protocol and session engine the
upcoming Tokio runtime will call. `ServerCore::handle_datagram` takes a source
endpoint and the bytes of one datagram and returns the bytes to send back, or
nothing. It performs no I/O and keeps no clock, so every decision is a pure
function of its state and that input apart from drawing session tokens.

Rejection is silence. The protocol defines no error reply, no reset and no
NACK, so malformed, unauthenticated and policy-refused datagrams are discarded
without a response and without disturbing any live session; a client tells
"rejected" from "lost" only by timing out. `ServerError` is reserved for
internal failures on the server's own side, so an inbound `ProtoError` never
surfaces as a server failure.

Open handling authenticates before decoding parameters, so a packet that fails
authentication is discarded without parsing untrusted data and no reply
behavior reveals which stage rejected it. Duration and Interval are accepted
when absent and rejected when explicitly present and not positive, which is
what the presence-aware parameter decode is for. Negotiation currently rewrites
only the protocol version; the restriction knobs the specification describes
are deliberately left for their own slices, with one obvious place to add them.

Sessions are bounded to a finite default. A single unauthenticated datagram
creates one and opens are never deduplicated, so an unbounded table is a
liability rather than a compatibility feature; upstream's absence of session
bounds is explicitly not a target. A full table drops new opens silently and
evicts nothing. No-test opens keep working at capacity because they create no
state.

Session tokens come from the operating system's random source, since a token is
the only thing standing between an off-path attacker and a usable session on an
unauthenticated server. A drawn value that is zero or collides with a live
session is redrawn within a finite budget; there is no predictable fallback,
and a colliding token never overwrites the session that holds it. The reply is
encoded before the session is inserted, so an internal failure cannot leave a
half-created session behind.

Echo and close requests pass admission and are then deliberately ignored: this
slice creates sessions, and lifecycle and receive semantics belong to the
slices that implement them. Tests pin that so a half-implemented lifecycle
cannot appear by accident.

crates/irtt-server/src/tests/open.rs

@@ -0,0 +1,197 @@
+use irtt_proto::{Clock, Params, ReceivedStats, ServerFill, StampAt};
+
+use super::support::{
+    client_params, core_with_tokens, expect_normal_open_reply, open_request,
+    open_request_with_raw_params, param_int, peer, ScriptedTokens,
+};
+use crate::ServerConfig;
+
+const TOKEN_A: u64 = 0x7896_b6ab_8771_5213;
+
+#[test]
+fn an_ordinary_open_is_answered_and_creates_one_session() {
+    let mut core = core_with_tokens(ServerConfig::default(), ScriptedTokens::new([TOKEN_A]));
+
+    let packet = core
+        .handle_datagram(peer(), &open_request(&client_params(), None))
+        .unwrap()
+        .expect("an ordinary open must be answered");
+
+    let reply = expect_normal_open_reply(&packet, None);
+    assert_eq!(reply.token, TOKEN_A);
+    assert_eq!(core.session_count(), 1);
+
+    let session = core.session(TOKEN_A).expect("the session must be live");
+    assert_eq!(session.params(), &reply.params);
+}
+
+#[test]
+fn a_session_records_the_exact_source_endpoint() {
+    // Address family, address, port and — where present — the IPv6 scope all
+    // form part of session identity, so the endpoint is stored verbatim rather
+    // than normalized to an address.
+    for endpoint in [
+        "198.51.100.7:41234",
+        "198.51.100.7:41235",
+        "[2001:db8::1]:2112",
+        "[fe80::1%3]:2112",
+    ] {
+        let endpoint = endpoint.parse().unwrap();
+        let mut core = core_with_tokens(ServerConfig::default(), ScriptedTokens::new([TOKEN_A]));
+
+        core.handle_datagram(endpoint, &open_request(&client_params(), None))
+            .unwrap()
+            .expect("open must be answered");
+
+        assert_eq!(core.session(TOKEN_A).unwrap().peer(), endpoint);
+    }
+}
+
+#[test]
+fn any_requested_protocol_version_is_accepted_and_answered_with_one() {
+    // Version mismatch is detected client-side only: the server rewrites the
+    // version and creates a normal session whatever was asked for.
+    for requested in [Some(1), Some(0), Some(2), Some(-1), None] {
+        let mut core = core_with_tokens(ServerConfig::default(), ScriptedTokens::new([TOKEN_A]));
+        let params = Params {
+            protocol_version: requested.unwrap_or(0),
+            ..client_params()
+        };
+        let request = match requested {
+            // An explicit zero cannot be encoded by `Params::encode`, which
+            // omits zero-valued integers, so encode that tag by hand.
+            Some(0) => open_request_with_raw_params(&param_int(1, 0), None),
+            Some(_) => open_request(&params, None),
+            None => open_request(
+                &Params {
+                    protocol_version: 0,
+                    ..params
+                },
+                None,
+            ),
+        };
+
+        let packet = core
+            .handle_datagram(peer(), &request)
+            .unwrap()
+            .unwrap_or_else(|| panic!("protocol version {requested:?} must be answered"));
+
+        let reply = expect_normal_open_reply(&packet, None);
+        assert_eq!(reply.params.protocol_version, 1);
+        assert_eq!(core.session_count(), 1);
+    }
+}
+
+#[test]
+fn an_empty_parameter_payload_is_answered_with_only_a_version() {
+    let mut core = core_with_tokens(ServerConfig::default(), ScriptedTokens::new([TOKEN_A]));
+
+    let packet = core
+        .handle_datagram(peer(), &open_request_with_raw_params(&[], None))
+        .unwrap()
+        .expect("an empty parameter payload is a valid open");
+
+    // header (4) + token (8) + ProtocolVersion=1 (2). Every other parameter
+    // negotiates to its wire-default zero and is omitted entirely, which is
+    // only reachable if absent parameters really did decode as zero.
+    assert_eq!(packet.len(), 14);
+
+    let reply = expect_normal_open_reply(&packet, None);
+    assert_eq!(
+        reply.params,
+        Params {
+            protocol_version: 1,
+            ..Params::default()
+        }
+    );
+    assert_eq!(reply.params.clock, Clock::Unspecified);
+    assert_eq!(core.session_count(), 1);
+}
+
+#[test]
+fn unknown_parameter_tags_are_ignored_and_not_reflected_in_the_reply() {
+    let mut core = core_with_tokens(ServerConfig::default(), ScriptedTokens::new([TOKEN_A]));
+
+    let mut payload = param_int(1, 1);
+    payload.extend_from_slice(&param_int(42, 7));
+    payload.extend_from_slice(&param_int(200, 9));
+
+    let packet = core
+        .handle_datagram(peer(), &open_request_with_raw_params(&payload, None))
+        .unwrap()
+        .expect("unknown tags must not prevent an open");
+
+    let reply = expect_normal_open_reply(&packet, None);
+    assert_eq!(
+        reply.params,
+        Params {
+            protocol_version: 1,
+            ..Params::default()
+        }
+    );
+    // The reply is the same 14 bytes an empty payload produces: no unknown tag
+    // was copied through.
+    assert_eq!(packet.len(), 14);
+    assert_eq!(core.session_count(), 1);
+}
+
+#[test]
+fn this_slice_restricts_nothing_but_the_protocol_version() {
+    // These are the values a later restriction-policy slice will deliberately
+    // start changing. Pinning them now makes that a visible decision rather
+    // than an accident.
+    let unrestricted = Params {
+        protocol_version: 2,
+        duration_ns: 3_600_000_000_000,
+        interval_ns: 1,
+        length: 65_535,
+        received_stats: ReceivedStats::Window,
+        stamp_at: StampAt::Midpoint,
+        clock: Clock::Monotonic,
+        dscp: 184,
+        server_fill: Some(ServerFill {
+            value: "pattern:abc".to_owned(),
+        }),
+    };
+
+    let mut core = core_with_tokens(ServerConfig::default(), ScriptedTokens::new([TOKEN_A]));
+    let packet = core
+        .handle_datagram(peer(), &open_request(&unrestricted, None))
+        .unwrap()
+        .expect("open must be answered");
+
+    let reply = expect_normal_open_reply(&packet, None);
+    assert_eq!(
+        reply.params,
+        Params {
+            protocol_version: 1,
+            ..unrestricted
+        }
+    );
+}
+
+#[test]
+fn negative_and_out_of_range_length_and_dscp_survive_negotiation() {
+    // Length and DSCP are accepted as decoded in this slice, including values a
+    // socket could never carry. Restricting a DSCP the server cannot apply is a
+    // later policy decision, not an admission rule.
+    for (length, dscp) in [(-1, -1), (0, 0), (1472, 184), (-4096, 512)] {
+        let requested = Params {
+            protocol_version: 1,
+            length,
+            dscp,
+            ..Params::default()
+        };
+        let mut core = core_with_tokens(ServerConfig::default(), ScriptedTokens::new([TOKEN_A]));
+
+        let packet = core
+            .handle_datagram(peer(), &open_request(&requested, None))
+            .unwrap()
+            .unwrap_or_else(|| panic!("length {length} dscp {dscp} must be answered"));
+
+        let reply = expect_normal_open_reply(&packet, None);
+        assert_eq!(reply.params.length, length);
+        assert_eq!(reply.params.dscp, dscp);
+        assert_eq!(core.session_count(), 1);
+    }
+}