Repository navigation
fix(distributed): harden cluster pre-auth admission and rejections - #2705
Conversation
Five follow-ups from the #2674 security review on the authenticated cluster transport. None changes what a party without the secret can read or inject; they close DoS and robustness gaps on the accept path and the router demux. - Pending-handshake per-source limit keys IPv6 sources per /64 (`pending_source_key`), so a host that owns a /64 cannot take one pending slot per address; nodes inside one /64 share the budget (documented). - The unread `active_conns` counter and its manual decrements are gone; every per-connection resource is RAII, guarded by `panicking_rpc_handler_releases_its_connection_slot`. - Handshake warnings move to `distributed::handshake_warn` with separate `Noise` and `Diagnostic` budgets (5 per 60 s each) and a rollover summary, so junk connections cannot push the secret-mismatch line down to debug. - The disaggregated router refuses an unauthenticated transport at build time and its demux delivers result frames only from configured `--prefill-peers` / `--decode-peers` identities; other holders of the secret can no longer inject tokens through the sequential request ids. - Admission runs inside the handshake after the client authenticated: `server_handshake_admitting` answers a full cap with the new 0x02 verdict, the client gets the retryable `ServerAtCapacity`, and `PROTOCOL_VERSION` goes 3 to 4 (no released version carries 3). Verified on GB10: fmt, clippy, the distributed:: and server::router_front lib modules, zero_config_cluster_init, the real-model remote pipeline suites, both disaggregated e2e suites, and a manual router + prefill + decode loopback (HTTP 200, no dropped result frame). Closes #2692
Implementation Review SummaryIntent
Findings AddressedNone needed fixing at CRITICAL or HIGH. Checked specifically: Remaining Items
Verification
|
|
Security and performance review of 699d208: no CRITICAL or HIGH findings, so no fixes were pushed. Points already raised in the implementation review are not repeated. Checked and sound:
LOW, not fixed:
Tests run (test-fast, cuda, |
Correct the residual-risk text: IdentityNotOwned and FrameSuiteMismatch are decided from the hello before the client authenticates, so a party without the secret can trigger them and saturate the Diagnostic window, not only ClientAbortedAfterServerProof; the Noise class definition no longer claims to be the only stranger-producible class. Describe ServerAtCapacity precisely: it is not fatal like HandshakeRejected, connect_with_retry returns it without backoff, the next acquire reconnects, and Transport::connect warns. The "failed handshake is terminal" comment in connect_with_retry now covers this case. State that the router sender check authenticates the source IP only; the port is the advertised listener port and is not verified. Note in startup.rs that RouterState::build also enforces the cluster secret. Merge a stray re-wrapped line in docs/distributed.md. spawn_result_demux now logs the drop through tracing fields in both branches instead of building a String with format!. No behavior change. Validation: cargo fmt --check, cargo clippy --profile test-fast --features cuda -p mlxcel --lib --tests -- -D warnings, and the server::router_front:: and distributed::handshake_warn:: lib tests (38 passed). Refs #2692
… (#2708) Closes #2694. The seven residuals of #2671, #2680 and #2684 in one PR, each with its own test. Reconciled against current main: the handshake `PROTOCOL_VERSION` stays 4 (#2705), the `FrameRejected` call sites keep their `handshake_warn` budget unchanged, and `ProtectedStream::peer_closed` (#2700) is untouched. ## Per item 1. `read_frame` interprets the kind byte after `FrameOpener::open` on a protected stream: a tampered invalid kind is `FrameRejected` through the existing budgeted call sites (no call-site change); an authenticated unknown kind is `unknown message kind {n} from an authenticated peer`. Plain streams and the header-only oversize check are unchanged. 2. `encode_message_frame` reserves the header and `MAX_TAG_LEN` (32, const-asserted against `FrameSuite::ALL`); `finish_frame` is the one sealing path, used by `write_prepared_frame` (`TcpTransport::send`) and by `write_frame(kind, &[u8])`. A buffer shorter than the header is an error, not a panic, and `write_prepared_frame` is `pub(crate)`. `serialize_message` is test-only. 3. `ActivationMessage::reply_nonce` (`serde(default)`) carries the `RunEntry` nonce from the entry stage through middle stages; the final stage sends `RunResult` to the coordinator's authenticated identity (the source of its most recent command, so a coordinator configured on `0.0.0.0` still gets the reply across hosts). A reverse activation, or a nonce-less forward activation on the final stage, is dropped with a warn. `PendingEntryReply`, its map and the `pending_entry_replies` fields are gone. `ExchangeMatch::result_peer` is `stage_peers.last()` for `RunEntry` and the command's peer otherwise; `Ack`, `State`, `Error` keep `peer`. Frames per decoded token: 3 (`RunEntry`, forward activation, `RunResult`) instead of 4. 4. `render_peer_assignment_toml` (shared `render_toml_body`) is what a `--pp-peer` writes; the coordinator renderer and its callers are unchanged. 5. `Activate` / `Abort { detail }` after `Committed`, `ASSIGNMENT_PROTOCOL_VERSION` 2. The peer delivers only on `Activate`; `Abort`, another frame, close or the commit timeout reset `committed` and return it to listening. The coordinator activates all peers after every commit (a failed `Activate` write is an `AssignmentPushFailed` entry whose message says the other peers are running) or aborts every committed peer (detail cut to 1 KiB) when any commit failed. The peer logs a received detail bounded to 1 KiB and escaped to one line. The residual window is in the module docs. 6. `text_prefill_masks` returns `(None, None)` when `sliding_attn` is `None`. 7. `two_stage_split_matches_dense_after_front_trim_bitwise` (12-token prefill, `trim_front(4)` on every `Cache::Standard`, 3-token continuation, one decode step, bitwise against dense). Docs: `docs/distributed.md` (assignment protocol, peer TOML header, reply authorization, 3-frame per-token flow); no Korean counterpart exists for these sections. ## Failing-first evidence Run against the unchanged behaviour before each fix, green after: - Item 1 `protected_stream_rejects_tampered_invalid_kind_as_frame_rejected`: `a tampered invalid kind must surface as FrameRejected, got: unknown message kind: 255`; `protected_stream_reports_authenticated_unknown_kind_distinctly`: `unexpected error: unknown message kind: 255`. - Item 3 `final_stage_sends_run_result_to_coordinator_not_upstream`: `stage must answer within 5s (it may have died): Elapsed(())`; `run_result_from_last_stage_completes_forward`: `remote stage 10.1.1.1:1 timed out after 5s`; `run_result_from_entry_stage_is_dropped_in_multi_stage_pipeline`: `a RunResult from a stage other than the last must not complete the exchange`; `final_stage_sends_run_result_to_authenticated_coordinator_identity` (coordinator configured as `[::]:port`): `Elapsed(())` when the reply went to the configured address. - Item 5 `peer_discards_committed_assignment_on_abort_and_activates_on_the_next`: `the peer delivered an assignment that was never activated`; `coordinator_aborts_committed_peers_when_another_commit_fails`: `the real peer delivered an assignment the coordinator aborted`. - Item 6 `text_prefill_masks_without_sliding_layer_builds_no_mask`: `a stage without a sliding layer must not allocate a sliding mask`. ## Item 7 sensitivity A temporary mutation (never committed, file restored byte for byte, sha256 `de5f8b65…c457036` before and after) gave the stage path an offset-sized global prefill mask (`Some(create_causal_mask(seq_len, first_cache_offset(caches, "full_attention")))` instead of `None`). The trim test then fails with `3-token continuation prefill after trim_front(4): stage logits differ from the dense logits (max abs diff 5.46875e-2)`. It is the only gemma4 stage test whose scenario has `offset != live_len` (12 vs 8). ## Decode throughput `pipeline_remote_decode_bench` (qwen3-0.6b-4bit, 2 stages, loopback, 128 tokens, 7 timed repeats after a warm-up per run), main 433560c (A) and this branch (B) interleaved A B over 5 rounds on a quiet GB10, each run under `gpu-lock`. Both binaries were built with the bench's fixed port moved to 19990 (local only, not committed). Greedy tokens are identical in all 80 repeats. | | per-round medians (tok/s) | median of medians | range | |---|---|---|---| | main | 120.67, 140.70, 131.38, 140.29, 136.73 | 136.73 | 120.67 to 140.70 | | branch | 149.33, 142.75, 159.04, 156.94, 162.01 | 156.94 | 142.75 to 162.01 | The branch is about 15% faster, and its slowest round beats main's fastest. ## Verification (GB10 spark-102, `--profile test-fast --features cuda,test-utils`, test binaries built outside the lock and run under `gpu-lock`, `--test-threads=1`, `MLXCEL_REQUIRE_MODELS=1 MLXCEL_SDPA_DETERMINISTIC=1`, at 7c8d340) - `cargo fmt --check` and `cargo clippy ... --lib --tests -- -D warnings`: clean. - `--lib distributed::`: 1494 passed, 4 ignored. `--lib models::gemma4`: 127 passed, 15 ignored. - `zero_config_cluster_init --include-ignored`: 7/7 (the `zero_config_cli_*` rehearsals drive the real binaries through Activate). - `pipeline_remote_real_models --include-ignored --skip glm_moe_dsa`: 10/10. `pipeline_server_remote_real_models --include-ignored --skip glm_moe_dsa`: 7/7. The skipped `glm_moe_dsa` tests need the 418.6 GB `glm5-4bit` checkpoint, which cannot load on a 128 GB GB10. - `pipeline_ci_multi_stage_real_models --include-ignored`: 4/4, including `pipeline_multi_stage_three_host_real_model_parity` (logits now skip the middle stage) and the two-host smoke. - `disaggregated_serving_e2e` 2/2, `disaggregated_router_e2e` 7/7, `disaggregated_e2e` 51/51 (all `--include-ignored`). ## Compatibility The stage RPC messages change shape (`pending_entry_replies` is gone from `Ack` and `State`, `RunResult` comes from the final stage) while the handshake `PROTOCOL_VERSION` stays 4. A node built from main between #2705 and this PR therefore completes the handshake with a node built from this PR and then fails with exchange timeouts instead of a version error. No tagged release is affected (the latest is v0.8.0-beta.1, before the v3 and v4 handshakes); upgrade all nodes together.
Summary
Five hardening items from the #2674 security review on the authenticated cluster transport (
src/distributed/) and the disaggregated router demux, in one PR with one verification pass. None changes what a party without the secret can read or inject.pending_source_key(ip)canonicalizes, keeps IPv4 as is and masks IPv6 to its /64;try_acquireandpending_fromuse it,PendingSlotRefusal::PerIpcarries the key and prints.../64for IPv6. Module doc, themax_pending_handshakes_per_ipfield doc anddocs/distributed.mdsay that nodes inside one /64 share the per-source budget.active_connsdeleted with its three manual decrements and theAtomicUsizeimport; no replacement counter.cluster_auth::panicking_rpc_handler_releases_its_connection_slotis the regression guard (see the note below on its observable).src/distributed/handshake_warn.rs(HandshakeWarnClass {Noise, Diagnostic},handshake_warn_class(&anyhow::Error),WarnBudget::allow_at(class, now) -> WarnDecision { warn, suppressed_in_previous_window }, process-widehandshake_warn_allowed(class)with the rollover summaryN cluster handshake warnings (<class>) were logged at debug in the last 60s). Pending refusals drawNoise, handshake failureshandshake_warn_class(&err), rejected framesDiagnostic;tcp_transport.rsandstage_assignment.rscall sites updated, the old limiter removed. Residual risk noted indocs/distributed.md.TcpTransport::is_authenticated();RouterState::buildfails withthe disaggregated router requires a role transport bound with the cluster secret;result_senders: HashSet<SocketAddr>(prefill ∪ decode peers, canonical IPs) andaccepts_result_from(&str); the demux drops other senders with therouter: dropping result frame from ...warn through theDiagnosticbudget. Both router test fixtures bind with auth.docs/distributed.md,docs/CONTINUOUS_BATCHING.mdand the--prefill-peers/--decode-peershelp now say a peer must be listed under the address it connects from (its--serving-bind, with an unspecified IP replaced by the source IP).docs/distributed.mdstates that the check authenticates the source IP (handshake-verified) while the port is the sender's advertised listener port, so it tells members apart by source IP plus claimed port, not by port-level proof. The demux logs drops through tracing fields without building a message string.VERDICT_AT_CAPACITY = 0x02,PROTOCOL_VERSION3 to 4 (cluster_handshake.rs, constant unchanged in name and place),pub(crate) async fn server_handshake_admitting<P>(stream, auth, local_identity, suite, channel, admit: impl FnOnce() -> Option<P>) -> Result<(String, SessionKeys, P)>callingadmit()only after the client authenticator verified;server_handshake/server_handshake_onwrap it with|| Some(()). New errorsConnectionLimitReached { peer }(server) andServerAtCapacity(client; not treated as fatal likeHandshakeRejected:connect_with_retryreturns it without backoff, the next poolacquirereconnects, andTransport::connectwarns), the latter re-exported fromdistributed/mod.rs.accept_looppasses|| conn_semaphore.try_acquire_owned().ok()and keeps themax connections (N) reachedwarn. Wire diagram and docs updated.Pre-existing limitation found while writing the item 2 test:
TcpTransportreadsRpcResponseframes only on accepted connections, never on the pooled outbound connection anrpc_callwent out on, so no RPC echo can reach a TCP client (the only in-repo RPC-over-TCP caller,bench.rs, ignores the result). The guard therefore keeps the panicking RPC handler as the trigger (client A'srpc_call(b"boom")fails) and uses client B'ssendreachingserver.recv()within 2 s as the observable.Changes during review
The implementation review and the security review found no CRITICAL or HIGH issues. The follow-up commit tightens wording only: the residual-risk notes now name
IdentityNotOwnedandFrameSuiteMismatch(decided from the hello alone) next toClientAbortedAfterServerProofas refusals a party without the secret can push into theDiagnosticwindow; theServerAtCapacitydocs and theconnection_pool.rs"failed handshake is terminal" comment describe the no-backoff reconnect precisely; thestartup.rsrouter comment creditsRouterState::buildfor enforcingcluster_auth; and the sender-check docs say only the source IP is verified.Failing-first evidence (main 84e20ce)
Checked by temporarily reverting the fix hunks on this branch (key on the full address, one shared warn window, no
is_authenticatedcheck, no sender check), rebuilding the lib tests and running the tests by name, then restoring the files byte for byte (sha256 verified):ipv6_sources_in_one_64_share_the_per_source_limit: FAILED (third acquire from2001:db8:1:2:ffff::3succeeded,unwrap_errpanicked atpending_handshakes_tests.rs:83).ipv6_sources_in_different_64s_are_counted_separately: FAILED (second address in the /64 acquired a slot,pending_handshakes_tests.rs:110).noise_flood_does_not_suppress_diagnostic_warnings: FAILED (left: WarnDecision { warn: false, .. }).build_refuses_an_unauthenticated_transport: FAILED (an unauthenticated transport must be refused).result_demux_drops_frames_from_unconfigured_senders: FAILED (unconfigured sender's frame was delivered: ... tokens: ["injected"]).rollover_reports_suppressed_countpasses with a shared window too (only one class is exercised);classifies_handshake_errorsandaccepts_result_from_only_configured_peerscall functions that do not exist on main, so they do not compile there.server_at_capacity_sends_capacity_verdict,admission_is_not_consulted_before_the_client_authenticatesandcluster_auth::full_connection_cap_reports_capacity_to_the_clientdo not compile on main (server_handshake_admitting,ServerAtCapacity,ConnectionLimitReachedare new).cluster_auth::panicking_rpc_handler_releases_its_connection_slotpasses with and without the counter, by design.Verification (GB10 spark-102,
--profile test-fast, builds outsidegpu-lock, runs undergpu-lock run --tag issue-2692)cargo fmt --check: clean.cargo clippy --profile test-fast --features cuda -p mlxcel --lib --tests -- -D warnings: clean (rerun after the review commit: clean;server::router_front::+distributed::handshake_warn::38 passed, and the touched modulescluster_handshake,pending_handshakes,tcp_transport53 passed at 699d208).--lib -- --test-threads=1): 13 passed.--lib distributed:: -- --test-threads=1: 1473 passed, 0 failed, 4 ignored.--lib server::router_front -- --test-threads=1: 35 passed.--test zero_config_cluster_init: 4 passed, 3 ignored.tests/disaggregated_serving_e2e.rs(--features cuda,test-utils -- --include-ignored --test-threads=1): 2 passed.tests/disaggregated_router_e2e.rs(same flags): 7 passed (49.7 s).MLXCEL_REQUIRE_MODELS=1 MLXCEL_SDPA_DETERMINISTIC=1,--features cuda,test-utils -- --ignored --test-threads=1:pipeline_remote_real_models10 passed, 1 failed;pipeline_server_remote_real_models7 passed, 1 failed;pipeline_ci_multi_stage_real_models2 passed. The two failures are theglm_moe_dsatests panicking inrepo_model_dirbecauseglm5-4bitis absent:mlx-community/GLM-5-4bitis 418.6 GB and cannot be loaded on this 128 GB box, and the tests predate fix(distributed): cap pre-auth slots, type rejections, auth DI roles #2674 (added 2026-04-13), so this is a missing checkpoint, not a regression. Every llama, gemma3, gemma4, qwen3, qwen3.5 and glm-4.7 parity test passed.--cluster-secret-filemode 0600,qwen3-0.6b-4bit, 127.0.0.1 ports 19801-19806,target/test-fast/mlxcel-serverbuilt from this branch):POST /v1/chat/completionsreturned HTTP 200 with"2 + 2 = 4.", and the router log contains nodropping result frameline. Processes killed by PID afterwards.Closes #2692