net/adapter/net/channel/config.rs
1//! Channel configuration and visibility.
2//!
3//! Channel policy uses the existing capability system (`CapabilityFilter`)
4//! for access rules, combined with L1 permission tokens. This avoids
5//! building a separate rule engine.
6
7use super::name::{ChannelHash, ChannelId, ChannelName};
8use crate::adapter::net::behavior::capability::{CapabilityFilter, CapabilitySet};
9use crate::adapter::net::identity::{EntityId, RevocationRegistry, TokenChain, TokenScope};
10use dashmap::DashMap;
11
12/// How a channel binds its dynamic name suffix to the subscribing
13/// peer's authenticated identity.
14///
15/// Set on a **prefix**-registered [`ChannelConfig`], this turns a
16/// family of dynamically-named channels from "anyone may subscribe to
17/// any name under the prefix" into "a peer may subscribe only to the
18/// one name that encodes its own identity".
19///
20/// The motivating case is nRPC's per-caller reply channels
21/// (`<service>.replies.<caller_origin>`). Those resolve through a
22/// permissive prefix entry, so pre-fix any mesh peer could hold a live
23/// subscription to *another* caller's reply channel and receive that
24/// caller's response bodies whenever the server's direct route missed
25/// and the response fell back to roster fan-out.
26///
27/// Evaluated against the **pinned** peer identity (the TOFU binding
28/// installed from a signature-verified direct capability announcement),
29/// never a wire-claimed value. A peer whose identity is not yet pinned
30/// is rejected: admitting it would hand an attacker the bypass of
31/// simply never announcing.
32#[derive(Debug, Clone, Copy, PartialEq, Eq)]
33pub enum OriginBinding {
34 /// The requested name's suffix — everything after the matched
35 /// prefix — must equal the subscriber's
36 /// [`EntityId::origin_hash`](crate::adapter::net::identity::EntityId::origin_hash)
37 /// rendered as exactly 16 lowercase hex digits, which is the
38 /// format nRPC uses to build the channel name.
39 OriginHashHex16,
40}
41
42impl OriginBinding {
43 /// The complete subscribe decision for a bound channel family.
44 ///
45 /// `pinned_origin` is the subscriber's TOFU-pinned
46 /// `EntityId::origin_hash()`, or `None` when the publisher has not
47 /// pinned that peer yet (no signature-verified direct capability
48 /// announcement has arrived from it).
49 ///
50 /// **`None` rejects.** This is the rule the whole finding turns on:
51 /// admitting a peer whose identity we do not know would let an
52 /// attacker bypass the binding entirely by simply never announcing,
53 /// which is not a fix. It is stated here, as one branch of a pure
54 /// function, rather than left implicit at the call site — that
55 /// makes it directly testable and hard to "simplify" away.
56 ///
57 /// The cost is an ordering requirement: a peer must be pinned
58 /// before its first subscribe to a bound family. In practice a node
59 /// announces as part of coming up, and the publisher pushes/learns
60 /// identities at session establishment, so this is satisfied well
61 /// before any application traffic; the nRPC client additionally
62 /// re-announces and retries on rejection.
63 pub fn authorizes(
64 self,
65 name: &str,
66 matched_prefix: Option<&str>,
67 pinned_origin: Option<u64>,
68 ) -> bool {
69 let Some(origin_hash) = pinned_origin else {
70 return false;
71 };
72 self.matches(name, matched_prefix, origin_hash)
73 }
74
75 /// Does `name` bind to `origin_hash` under `matched_prefix`?
76 ///
77 /// `matched_prefix` is `None` when the config was resolved by exact
78 /// name rather than through the prefix table. That combination has
79 /// no coherent meaning — there is no dynamic suffix to check — so
80 /// it fails closed rather than silently admitting.
81 pub fn matches(self, name: &str, matched_prefix: Option<&str>, origin_hash: u64) -> bool {
82 let Some(prefix) = matched_prefix else {
83 return false;
84 };
85 let Some(suffix) = name.strip_prefix(prefix) else {
86 // Defensive: the registry only hands us a prefix it
87 // matched, so this is unreachable — but a mismatch must
88 // never read as "bound".
89 return false;
90 };
91 match self {
92 Self::OriginHashHex16 => suffix == format!("{origin_hash:016x}"),
93 }
94 }
95}
96
97/// Who may join a **queue group** on a channel.
98///
99/// Queue groups are work distribution: every published event is
100/// delivered to exactly one member of each group. So joining a group is
101/// not a routing preference, it is a claim on other members' work — an
102/// attacker who joins a production group receives a share of its events
103/// and, by not processing them, destroys that share. With `L` honest
104/// members and `A` attacker identities, attackers collectively take
105/// `A/(L+A)` of selections and each identity takes `1/(L+A)`; the
106/// attacker scales its share simply by joining under more identities.
107///
108/// Note this is an **integrity and availability** boundary, not a
109/// confidentiality one: a peer that can subscribe at all can already
110/// take every event by subscribing in `Broadcast` mode, so joining a
111/// group exposes nothing new. What it does is take work away from the
112/// members meant to do it.
113#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
114pub enum QueueGroupPolicy {
115 /// Any peer that clears the channel's subscribe gate may join any
116 /// group. Historical behaviour, and the default so existing
117 /// deployments are unaffected.
118 #[default]
119 Unrestricted,
120 /// Refuse queue-group subscriptions entirely — broadcast only.
121 Deny,
122 /// A peer may join group `G` only by presenting a chain that
123 /// authorizes `SUBSCRIBE` on [`queue_group_hash(channel, G)`],
124 /// i.e. a grant that names the specific group.
125 ///
126 /// Under this policy the group grant **is** the subscribe
127 /// authority for that request — a worker does not additionally
128 /// need a channel-scoped token. It cannot: the `Subscribe` wire
129 /// message carries exactly one chain, so requiring both would make
130 /// worker subscription unrepresentable. The model an operator gets
131 /// is the intended one: channel-scoped tokens for readers,
132 /// group-scoped tokens for workers, and a reader's token is
133 /// explicitly not a worker grant. Capability filters still apply.
134 ///
135 /// An allowlist of group *names* would not do: group names are
136 /// operational constants, not secrets, so an attacker simply joins
137 /// an allowed one. Nor would a generic "may join queue groups"
138 /// scope bit — that separates readers from workers but still lets
139 /// any worker join any group on the channel. The authority has to
140 /// bind the peer to the group.
141 ///
142 /// [`queue_group_hash(channel, G)`]: super::queue_group_hash
143 TokenBound,
144}
145
146/// Channel visibility scope.
147///
148/// # Visibility is a propagation filter, not an access boundary
149///
150/// Visibility is evaluated against *topology state* — the local
151/// node's configured subnet and a per-peer subnet derived from each
152/// peer's own self-declared capability tags. A peer that misdeclares
153/// its tags can place itself wherever the local `SubnetPolicy` maps
154/// those tags; visibility narrows where traffic propagates, it does
155/// not decide who is allowed in. A protected channel must pair a
156/// scoped visibility with token enforcement (`token_roots`): the
157/// lying peer may pass the visibility test, but it cannot forge the
158/// channel token. Purely soft channels (scoped visibility, no token
159/// gate) are valid — they are a routing decision, and registration
160/// logs one info-level note so that decision is recorded
161/// (SUBNET_AUTH_PLAN.md Q1).
162#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
163pub enum Visibility {
164 /// Propagation limited to the same subnet. A routing filter —
165 /// pair with `token_roots` if the channel must exclude anyone.
166 SubnetLocal,
167 /// Propagates to the parent subnet but not siblings. A routing
168 /// filter — pair with `token_roots` if the channel must exclude
169 /// anyone.
170 ParentVisible,
171 /// Explicitly exported to specific target subnets.
172 Exported,
173 /// Visible everywhere, no subnet restriction.
174 #[default]
175 Global,
176}
177
178/// Channel configuration with capability-based access control.
179///
180/// Authorization flow:
181/// 1. Node announces capabilities via `CapabilityAd`
182/// 2. If `publish_caps` is set, node's `CapabilitySet` must match the filter
183/// 3. If `require_token` is true, node must also have a valid `PermissionToken`
184/// 4. On success, `(origin_hash, channel_hash)` is inserted into the `AuthGuard`
185///
186/// # Capability filters are advisory, not an access boundary
187///
188/// `publish_caps` / `subscribe_caps` match against a node's
189/// *self-advertised* `CapabilitySet`: a peer declares its own
190/// capabilities in its own signed announcement, so any peer can
191/// satisfy a cap-filter simply by advertising the required tag
192/// (e.g. self-asserting `role:admin`). Treat cap-filters as
193/// matchmaking / intent-routing, **not** as a security boundary.
194///
195/// The actual access boundary is `require_token` + `token_roots`:
196/// a root-anchored [`TokenChain`] cannot be forged because each link
197/// is signature-verified up to a root the channel explicitly trusts.
198/// Any channel that must restrict who can publish or subscribe must
199/// use token enforcement; a cap-filter alone restricts nothing.
200#[derive(Debug, Clone)]
201pub struct ChannelConfig {
202 /// Channel identity (name + hash).
203 pub channel_id: ChannelId,
204 /// Visibility scope for subnet routing.
205 pub visibility: Visibility,
206 /// Capability requirements for publishing. `None` = any node can
207 /// publish. Advisory only — matched against the node's
208 /// self-advertised caps; use `require_token` for a real boundary.
209 pub publish_caps: Option<CapabilityFilter>,
210 /// Capability requirements for subscribing. `None` = any node can
211 /// subscribe. Advisory only — matched against the node's
212 /// self-advertised caps; use `require_token` for a real boundary.
213 pub subscribe_caps: Option<CapabilityFilter>,
214 /// Whether a valid `PermissionToken` is required (in addition to capabilities).
215 pub require_token: bool,
216 /// Entities whose signature roots a valid token chain for this
217 /// channel — the channel's root(s) of trust.
218 ///
219 /// When `require_token` is set, a presented [`TokenChain`] is only
220 /// honored if its root link (`tokens[0].issuer`) is one of these
221 /// entities. This is the anchor the bare-token path lacked: without
222 /// it `check`/`can_subscribe` only verified a token was internally
223 /// self-consistent (the named issuer signed it), so any peer could
224 /// self-issue `issuer = subject = self` and pass. An empty
225 /// `token_roots` combined with `require_token = true` **fails
226 /// closed** — there is no authority a chain could anchor to, so
227 /// nothing is authorized.
228 pub token_roots: Vec<EntityId>,
229 /// Bind the dynamic name suffix to the subscriber's own pinned
230 /// identity. `None` (default) = any peer that clears the other
231 /// gates may subscribe to any name this config covers.
232 ///
233 /// Only meaningful on a prefix-registered config; see
234 /// [`OriginBinding`]. Unlike `publish_caps` / `subscribe_caps`,
235 /// this **is** an access boundary — it is evaluated against the
236 /// TOFU-pinned peer identity, which a peer cannot self-assert.
237 pub subscriber_origin_binding: Option<OriginBinding>,
238 /// Who may join a queue group on this channel. See
239 /// [`QueueGroupPolicy`]; defaults to `Unrestricted`, which is the
240 /// historical behaviour.
241 pub queue_group_policy: QueueGroupPolicy,
242 /// Default priority level for this channel's packets (0 = lowest).
243 pub priority: u8,
244 /// Default reliability mode for streams on this channel.
245 pub reliable: bool,
246 /// Optional rate limit in packets per second.
247 pub max_rate_pps: Option<u32>,
248}
249
250impl ChannelConfig {
251 /// Create a new channel config with defaults (open access, global visibility).
252 pub fn new(channel_id: ChannelId) -> Self {
253 Self {
254 channel_id,
255 visibility: Visibility::default(),
256 publish_caps: None,
257 subscribe_caps: None,
258 require_token: false,
259 token_roots: Vec::new(),
260 subscriber_origin_binding: None,
261 queue_group_policy: QueueGroupPolicy::default(),
262 priority: 0,
263 reliable: false,
264 max_rate_pps: None,
265 }
266 }
267
268 /// Set visibility.
269 pub fn with_visibility(mut self, visibility: Visibility) -> Self {
270 self.visibility = visibility;
271 self
272 }
273
274 /// Set capability requirements for publishing.
275 ///
276 /// Advisory matchmaking, not access control: caps are
277 /// self-advertised, so any peer can satisfy the filter by
278 /// declaring the tag. Combine with [`Self::with_token_roots`] to
279 /// actually restrict publishers.
280 pub fn with_publish_caps(mut self, filter: CapabilityFilter) -> Self {
281 self.publish_caps = Some(filter);
282 self
283 }
284
285 /// Set capability requirements for subscribing.
286 ///
287 /// Advisory matchmaking, not access control: caps are
288 /// self-advertised, so any peer can satisfy the filter by
289 /// declaring the tag. Combine with [`Self::with_token_roots`] to
290 /// actually restrict subscribers.
291 pub fn with_subscribe_caps(mut self, filter: CapabilityFilter) -> Self {
292 self.subscribe_caps = Some(filter);
293 self
294 }
295
296 /// Require a valid permission token.
297 pub fn with_require_token(mut self, require: bool) -> Self {
298 self.require_token = require;
299 self
300 }
301
302 /// Require a token chain rooted at one of `roots`. Sets
303 /// `require_token = true` and installs the channel's authorizing
304 /// root(s). This is the safe way to turn on token enforcement —
305 /// `with_require_token(true)` alone (no roots) fails every
306 /// authorization closed, since a chain has no authority to anchor
307 /// to.
308 pub fn with_token_roots(mut self, roots: Vec<EntityId>) -> Self {
309 self.require_token = true;
310 self.token_roots = roots;
311 self
312 }
313
314 /// Whether this channel enforces token authorization.
315 ///
316 /// Enforcement is on when `require_token` is set **or** any
317 /// `token_roots` are configured. Coupling the two means a config
318 /// that names roots but forgot to flip `require_token` (e.g. built
319 /// by struct literal or direct field assignment rather than
320 /// [`Self::with_token_roots`]) still enforces, instead of silently
321 /// admitting every peer — the fields are both public, so the
322 /// invariant can't be guaranteed at construction. All token gates
323 /// (subscribe, publish, the periodic sweep, the publish re-check)
324 /// consult this rather than `require_token` directly.
325 pub fn token_required(&self) -> bool {
326 self.require_token || !self.token_roots.is_empty()
327 }
328
329 /// Bind this (prefix-registered) channel family's dynamic suffix to
330 /// the subscribing peer's own pinned identity — see
331 /// [`OriginBinding`].
332 ///
333 /// Callers subscribing to a bound family must have had their
334 /// identity pinned on the publisher first, which happens when their
335 /// signature-verified direct capability announcement arrives. A peer
336 /// that has not announced is rejected (fail closed).
337 pub fn with_subscriber_origin_binding(mut self, binding: OriginBinding) -> Self {
338 self.subscriber_origin_binding = Some(binding);
339 self
340 }
341
342 /// Do this node's advertised capabilities satisfy the channel's
343 /// `subscribe_caps` filter?
344 ///
345 /// Split out of [`Self::can_subscribe`] for the `TokenBound`
346 /// queue-group path, which supplies its own token authority (the
347 /// group grant) but must still apply the capability filter.
348 /// Advisory, like every cap filter — see the type docs.
349 pub fn caps_allow_subscribe(&self, node_caps: &CapabilitySet) -> bool {
350 match self.subscribe_caps {
351 Some(ref filter) => filter.matches(node_caps),
352 None => true,
353 }
354 }
355
356 /// Restrict who may join a queue group on this channel — see
357 /// [`QueueGroupPolicy`].
358 pub fn with_queue_group_policy(mut self, policy: QueueGroupPolicy) -> Self {
359 self.queue_group_policy = policy;
360 self
361 }
362
363 /// Does `chain` authorize this peer to join queue group `group` on
364 /// `channel`?
365 ///
366 /// Returns `true` when the channel places no restriction. Under
367 /// [`QueueGroupPolicy::TokenBound`] the chain must root at one of
368 /// this channel's `token_roots`, bind at its leaf to `entity_id`,
369 /// and authorize `SUBSCRIBE` on the derived group-grant hash — a
370 /// grant naming the specific group, not the channel.
371 ///
372 /// Fails closed: `Deny` refuses, and `TokenBound` with no chain (or
373 /// no roots) refuses.
374 pub fn can_join_queue_group(
375 &self,
376 entity_id: &EntityId,
377 channel: &str,
378 group: &str,
379 chain: Option<&TokenChain>,
380 revocation: &RevocationRegistry,
381 skew_secs: u64,
382 ) -> bool {
383 match self.queue_group_policy {
384 QueueGroupPolicy::Unrestricted => true,
385 QueueGroupPolicy::Deny => false,
386 QueueGroupPolicy::TokenBound => {
387 if self.token_roots.is_empty() {
388 return false;
389 }
390 let Some(chain) = chain else {
391 return false;
392 };
393 chain
394 .verify_authorizes(
395 TokenScope::SUBSCRIBE,
396 super::name::queue_group_hash(channel, group),
397 entity_id,
398 &self.token_roots,
399 revocation,
400 skew_secs,
401 )
402 .is_ok()
403 }
404 }
405 }
406
407 /// Set default priority.
408 pub fn with_priority(mut self, priority: u8) -> Self {
409 self.priority = priority;
410 self
411 }
412
413 /// Set default reliability.
414 pub fn with_reliable(mut self, reliable: bool) -> Self {
415 self.reliable = reliable;
416 self
417 }
418
419 /// Set rate limit.
420 pub fn with_rate_limit(mut self, pps: u32) -> Self {
421 self.max_rate_pps = Some(pps);
422 self
423 }
424
425 /// Check if `entity_id` is authorized to publish on `channel_hash`,
426 /// presenting `chain`.
427 ///
428 /// See [`Self::can_subscribe`] for the chain-verification contract
429 /// and for why `channel_hash` is a parameter; this is the
430 /// `PUBLISH`-scope counterpart.
431 pub fn can_publish(
432 &self,
433 node_caps: &CapabilitySet,
434 entity_id: &EntityId,
435 channel_hash: ChannelHash,
436 chain: Option<&TokenChain>,
437 revocation: &RevocationRegistry,
438 skew_secs: u64,
439 ) -> bool {
440 if let Some(ref filter) = self.publish_caps {
441 if !filter.matches(node_caps) {
442 return false;
443 }
444 }
445 self.token_gate(
446 TokenScope::PUBLISH,
447 entity_id,
448 channel_hash,
449 chain,
450 revocation,
451 skew_secs,
452 )
453 }
454
455 /// Check if `entity_id` is authorized to subscribe to
456 /// `channel_hash`, presenting `chain`.
457 ///
458 /// When `require_token` is set, `chain` must be a [`TokenChain`]
459 /// that (a) roots at one of [`Self::token_roots`], (b) is bound at
460 /// its leaf to `entity_id` (the AEAD-verified presenter), and (c)
461 /// authorizes `SUBSCRIBE` on `channel_hash` at every link with no
462 /// link revoked. A missing chain, an empty `token_roots`, or a
463 /// chain that fails verification all reject — fail closed.
464 ///
465 /// # Why `channel_hash` is a parameter
466 ///
467 /// It is the hash of the channel the caller actually asked for, NOT
468 /// `self.channel_id.hash()`. Those coincide for an exact-match
469 /// config, but a **prefix**-registered config's `channel_id` is a
470 /// sentinel that `insert_prefix` itself documents as "not used for
471 /// hash lookups" — and verifying against it meant a token minted
472 /// for the sentinel authorized *every* channel under the prefix,
473 /// silently degrading a per-channel binding to a per-prefix one.
474 /// Taking the channel explicitly also removes the standing
475 /// temptation to reuse one config across many channels and get a
476 /// gate that answers about the wrong one.
477 pub fn can_subscribe(
478 &self,
479 node_caps: &CapabilitySet,
480 entity_id: &EntityId,
481 channel_hash: ChannelHash,
482 chain: Option<&TokenChain>,
483 revocation: &RevocationRegistry,
484 skew_secs: u64,
485 ) -> bool {
486 if let Some(ref filter) = self.subscribe_caps {
487 if !filter.matches(node_caps) {
488 return false;
489 }
490 }
491 self.token_gate(
492 TokenScope::SUBSCRIBE,
493 entity_id,
494 channel_hash,
495 chain,
496 revocation,
497 skew_secs,
498 )
499 }
500
501 /// Shared token-chain gate for the publish / subscribe checks.
502 /// Returns `true` when token enforcement is off (capability filters
503 /// already applied by the caller), else verifies the presented
504 /// chain roots at one of `token_roots` and authorizes
505 /// `channel_hash`. Fails closed when tokens are required but no
506 /// roots are configured or no chain is presented.
507 fn token_gate(
508 &self,
509 action: TokenScope,
510 entity_id: &EntityId,
511 channel_hash: ChannelHash,
512 chain: Option<&TokenChain>,
513 revocation: &RevocationRegistry,
514 skew_secs: u64,
515 ) -> bool {
516 if !self.token_required() {
517 return true;
518 }
519 // No authorizing root → nothing can satisfy the gate. Fail
520 // closed rather than (pre-fix) honoring any self-consistent
521 // token.
522 if self.token_roots.is_empty() {
523 return false;
524 }
525 let Some(chain) = chain else {
526 return false;
527 };
528 chain
529 .verify_authorizes(
530 action,
531 channel_hash,
532 entity_id,
533 &self.token_roots,
534 revocation,
535 skew_secs,
536 )
537 .is_ok()
538 }
539
540 /// Re-verify a previously-presented `SUBSCRIBE` chain for
541 /// `channel_hash` against the current clock + revocation floors,
542 /// anchored to this channel's roots. Shared by the periodic expiry
543 /// sweep and the publish-time re-check so the root-anchoring
544 /// contract (which roots, which action, which channel hash) lives
545 /// in exactly one place instead of being re-threaded at each call
546 /// site — where it had already started to diverge (`token_roots`
547 /// vs. an `unwrap_or(&[])` fallback).
548 ///
549 /// `channel_hash` is the requested channel's — see
550 /// [`Self::can_subscribe`]. Passing the config's own hash here is
551 /// what made prefix-registered channels retain a chain under the
552 /// sentinel key that the publish path (keyed on the real channel)
553 /// could never find, so every such subscriber was accepted and then
554 /// revoked before its first delivery.
555 pub fn reverify_subscribe(
556 &self,
557 chain: &TokenChain,
558 entity_id: &EntityId,
559 channel_hash: ChannelHash,
560 revocation: &RevocationRegistry,
561 skew_secs: u64,
562 ) -> bool {
563 chain
564 .verify_authorizes(
565 TokenScope::SUBSCRIBE,
566 channel_hash,
567 entity_id,
568 &self.token_roots,
569 revocation,
570 skew_secs,
571 )
572 .is_ok()
573 }
574
575 /// Like [`Self::reverify_subscribe`] but skips the per-link ed25519
576 /// signature verification — for callers re-checking a chain whose
577 /// signatures already verified once (immutable tokens). Time
578 /// bounds, revocation, anchoring, and scope are still re-checked.
579 /// See [`TokenChain::verify_authorizes_presigned`].
580 pub fn reverify_subscribe_presigned(
581 &self,
582 chain: &TokenChain,
583 entity_id: &EntityId,
584 channel_hash: ChannelHash,
585 revocation: &RevocationRegistry,
586 skew_secs: u64,
587 ) -> bool {
588 chain
589 .verify_authorizes_presigned(
590 TokenScope::SUBSCRIBE,
591 channel_hash,
592 entity_id,
593 &self.token_roots,
594 revocation,
595 skew_secs,
596 )
597 .is_ok()
598 }
599}
600
601/// A channel config plus how it was resolved, from
602/// [`ChannelConfigRegistry::resolve_by_name`].
603#[derive(Debug, Clone)]
604pub struct ResolvedConfig {
605 /// The resolved configuration.
606 pub config: ChannelConfig,
607 /// `Some(prefix)` when resolution fell through to the prefix table,
608 /// `None` for an exact-name match. [`OriginBinding`] uses this to
609 /// split the requested name into prefix + dynamic suffix.
610 pub matched_prefix: Option<String>,
611}
612
613/// Registry of channel configurations.
614///
615/// Keyed by channel name (not hash) to prevent hash collisions from silently
616/// overwriting security policies. The canonical [`ChannelHash`] (`u64`) is
617/// collision-resistant at realistic scale (~65 K channels), and `by_hash`
618/// gives O(1) canonical-hash lookup; `by_wire_hash` resolves the wire
619/// `u16` fast-path hint into a list of canonical channels for receive-side
620/// dispatch (routine collisions at scale).
621///
622/// Surface the deny-all misconfigurations loudly at registration time.
623///
624/// Each of these is a legitimate fail-closed state that the
625/// authorization path handles correctly, and each is far more often a
626/// mistake. Fail-closed is the right behaviour and stays; what it lacks
627/// on its own is any way for the operator to find out, because a channel
628/// that denies everyone looks exactly like a channel nobody happens to
629/// be using. Logging at insert turns that into an actionable warning.
630///
631/// `is_prefix` distinguishes [`ChannelConfigRegistry::insert`] and
632/// [`ChannelConfigRegistry::insert_if_absent`] from their
633/// `insert_prefix*` counterparts — one of the two checks below depends
634/// on how the config is being registered, not just on its contents.
635/// Record the "visibility without access control" decision at
636/// registration time (SUBNET_AUTH_PLAN.md S0).
637///
638/// A `SubnetLocal` / `ParentVisible` channel with no token gate and
639/// no origin binding is a *soft* channel: its visibility narrows
640/// propagation against peer-declared topology, and nothing
641/// cryptographic keeps a misdeclaring peer out. That is a legitimate
642/// routing configuration — hence `info`, not `warn`, and no behavior
643/// change — but it looks identical to an operator who believed
644/// `SubnetLocal` was an access boundary, so the decision gets one
645/// recorded line. Capability filters don't count as access control
646/// here: they match self-advertised tags (advisory by their own
647/// rustdoc above).
648fn note_if_visibility_only(config: &ChannelConfig) {
649 let scoped = matches!(
650 config.visibility,
651 Visibility::SubnetLocal | Visibility::ParentVisible
652 );
653 if scoped && !config.token_required() && config.subscriber_origin_binding.is_none() {
654 tracing::info!(
655 channel = config.channel_id.name().as_str(),
656 visibility = ?config.visibility,
657 "channel has subnet-scoped visibility but no token gate or \
658 origin binding: visibility is a propagation filter over \
659 peer-declared topology, not an access boundary. If this \
660 channel must exclude anyone, add `with_token_roots(...)`."
661 );
662 }
663}
664
665fn warn_if_fail_closed(config: &ChannelConfig, is_prefix: bool) {
666 // `with_require_token(true)` called instead of
667 // `with_token_roots(...)`: there is no authority a chain could
668 // anchor to, so nothing is ever authorized.
669 if config.require_token && config.token_roots.is_empty() {
670 tracing::warn!(
671 channel = config.channel_id.name().as_str(),
672 "channel requires a token but has no token_roots: all publish \
673 and subscribe will be denied (fail closed). Use \
674 `with_token_roots(...)` to anchor a root of trust."
675 );
676 }
677 // An origin binding on an EXACT registration denies every
678 // subscriber. The binding's whole job is to split a requested name
679 // into "the prefix that matched" + "the dynamic suffix that must
680 // encode the subscriber", and an exact-match resolution reports no
681 // matched prefix — there is no suffix to check, so
682 // `OriginBinding::matches` fails closed on every request.
683 //
684 // The builder's rustdoc says it is "only meaningful on a
685 // prefix-registered config", but a channel that silently accepts
686 // nobody is not a documentation-sized failure: it is
687 // indistinguishable from an idle channel, and the peers being
688 // refused see a generic `Unauthorized`.
689 if is_prefix {
690 return;
691 }
692 if config.subscriber_origin_binding.is_some() {
693 tracing::warn!(
694 channel = config.channel_id.name().as_str(),
695 "channel has a subscriber_origin_binding but is registered by \
696 EXACT name: every subscribe will be denied (fail closed). The \
697 binding matches a dynamic suffix against the subscriber's own \
698 pinned origin, which only exists for a prefix registration — \
699 register it with `insert_prefix(...)` / \
700 `Mesh::register_channel_prefix(...)`."
701 );
702 }
703}
704
705/// Consulted at subscription/channel-creation time (slow path).
706/// The fast path uses the `AuthGuard` bloom filter.
707pub struct ChannelConfigRegistry {
708 /// Primary storage: name → config (collision-safe)
709 configs: DashMap<String, ChannelConfig>,
710 /// Reverse index: canonical hash → names (collision-resistant at u32).
711 by_hash: DashMap<ChannelHash, Vec<String>>,
712 /// Wire-hash reverse index: u16 wire-hash → names (routine collisions).
713 /// Used by receive-side dispatch to disambiguate the `NetHeader`
714 /// fast-path hint into canonical channels.
715 by_wire_hash: DashMap<u16, Vec<String>>,
716 /// Prefix registry: prefix → config. Consulted by
717 /// `get_by_name` when no exact match exists; the first prefix
718 /// that the queried name starts with wins. Used by nRPC's
719 /// SDK glue to register `<service>.replies.` once and admit
720 /// every `<service>.replies.<caller_origin>` subscribe that
721 /// follows.
722 ///
723 /// Prefix lookups are O(num_prefixes) — a small constant in
724 /// practice (one prefix per nRPC service). The exact-match
725 /// hot path is unaffected.
726 prefix_configs: DashMap<String, ChannelConfig>,
727 /// Serializes every mutation of `configs` + the two reverse
728 /// indices, so the three maps are only ever observed in a
729 /// consistent state.
730 ///
731 /// `configs` and the indices are separate DashMaps, so no per-entry
732 /// guard can span them. Under concurrent insert/remove that showed
733 /// up as index corruption in both directions: a re-registration
734 /// racing a removal could have its fresh index entry deleted (the
735 /// channel present in `configs` but invisible to `get(hash)`), and
736 /// the repair for THAT could re-add a name a second removal had
737 /// just taken out (a phantom name in a bucket, which `get` and
738 /// `remove` read as a hash collision and answer `None` to — taking
739 /// out the *real* channel's lookup as collateral).
740 ///
741 /// Writes only. Readers stay lock-free on the DashMaps: they are
742 /// the hot path, and a reader that observes a mid-write state
743 /// resolves through `configs` and simply misses, which is the
744 /// pre-existing behaviour for an unregistered channel. Mutations
745 /// are control-plane — registration, `net channel rm` — so
746 /// serializing them costs nothing measurable.
747 ///
748 /// NOT reentrant (`parking_lot::Mutex`). Methods that hold it call
749 /// the `_locked` inner helpers, never each other.
750 write_lock: parking_lot::Mutex<()>,
751}
752
753impl ChannelConfigRegistry {
754 /// Create an empty registry.
755 pub fn new() -> Self {
756 Self {
757 configs: DashMap::new(),
758 by_hash: DashMap::new(),
759 by_wire_hash: DashMap::new(),
760 prefix_configs: DashMap::new(),
761 write_lock: parking_lot::Mutex::new(()),
762 }
763 }
764
765 /// Register a prefix-matched channel configuration. Any
766 /// channel name starting with `prefix` that has no exact-match
767 /// entry will resolve to `config` via [`Self::get_by_name`].
768 ///
769 /// **Use sparingly.** Prefix lookups bypass the `by_hash`
770 /// fast path and walk the prefix list on the slow path; one
771 /// prefix per service is fine, hundreds is not. nRPC uses
772 /// this for its dynamic per-caller reply channels
773 /// (`<service>.replies.<caller_origin>`) — one prefix per
774 /// `serve_rpc` registration.
775 ///
776 /// `config.channel_id` should carry the prefix as a sentinel
777 /// name (e.g. `<svc>.replies.`); it isn't used for hash
778 /// lookups, so the channel-name validation rules don't apply
779 /// strictly. Prefix entries are collision-safe with respect
780 /// to each other (DashMap on the prefix string). When multiple
781 /// prefixes match a queried name, [`Self::get_by_name`] returns
782 /// the LONGEST one — so a more specific entry safely overrides
783 /// a more general one. Resolution is deterministic across
784 /// processes (the longest-length tiebreaker can never tie since
785 /// DashMap deduplicates keys).
786 pub fn insert_prefix(&self, prefix: impl Into<String>, config: ChannelConfig) {
787 warn_if_fail_closed(&config, true);
788 note_if_visibility_only(&config);
789 self.prefix_configs.insert(prefix.into(), config);
790 }
791
792 /// Register a prefix-matched config **only if that prefix has no
793 /// entry yet**. Returns `true` if this call installed the config,
794 /// `false` if an entry already existed (which is left untouched).
795 ///
796 /// The prefix counterpart of [`Self::insert_if_absent`], and the
797 /// operation auto-registration must use so it cannot silently
798 /// discard an operator's ACL. See that method for the rationale.
799 pub fn insert_prefix_if_absent(
800 &self,
801 prefix: impl Into<String>,
802 config: ChannelConfig,
803 ) -> bool {
804 match self.prefix_configs.entry(prefix.into()) {
805 dashmap::mapref::entry::Entry::Occupied(_) => false,
806 dashmap::mapref::entry::Entry::Vacant(slot) => {
807 warn_if_fail_closed(&config, true);
808 note_if_visibility_only(&config);
809 slot.insert(config);
810 true
811 }
812 }
813 }
814
815 /// Remove a prefix-matched config. Returns the removed config
816 /// if it existed.
817 pub fn remove_prefix(&self, prefix: &str) -> Option<ChannelConfig> {
818 self.prefix_configs.remove(prefix).map(|(_, v)| v)
819 }
820
821 /// Install the standard channel policy for an RPC-style service:
822 /// the exact `<service>.requests` channel, and the
823 /// `<service>.replies.` prefix bound to each caller's own origin.
824 ///
825 /// **Install-if-absent, never replace** (H2) — an ACL the operator
826 /// registered before serving survives untouched. **Origin-bound
827 /// reply prefix** (H3) — a peer may subscribe only to the one reply
828 /// channel that encodes its own pinned identity, not to another
829 /// caller's.
830 ///
831 /// Lives on the registry, not on an SDK type, because there are
832 /// several serve paths and they do not share a receiver: the SDK's
833 /// `Mesh::serve_rpc*`, the `aggregator` module, and the org facade's
834 /// `serve_org_bytes_node` (which holds an `Arc<MeshNode>` for the
835 /// language bindings). What they DO share is this registry.
836 ///
837 /// That sharing is the whole point. This policy has now drifted
838 /// twice, both times the same way — a serve path carrying its own
839 /// copy of the registration and not receiving a later fix. The
840 /// aggregator kept a replacing insert and never gained the origin
841 /// binding, so aggregator reply channels stayed world-subscribable
842 /// after H2 and H3 were fixed for `serve_rpc`; the org path did the
843 /// same and was still doing it after the aggregator was folded in.
844 /// A copy per receiver type is not a shared implementation. One
845 /// implementation on the object all of them already hold is.
846 ///
847 /// A caller with no registry (possible via the bare `MeshNode::new`
848 /// path) simply has no channel ACLs in play and does not call this.
849 ///
850 /// **All or nothing**, and validated against the names callers will
851 /// actually use.
852 ///
853 /// Three names are in play and none of them is the same length:
854 ///
855 /// | name | suffix bytes |
856 /// |---|---|
857 /// | `<service>.requests` | 9 |
858 /// | `<service>.replies.prefix` (sentinel) | 15 |
859 /// | `<service>.replies.<16 hex>` (real) | 25 |
860 ///
861 /// So there are two bands near the channel-name length limit where a
862 /// naive implementation half-succeeds, and they fail differently:
863 ///
864 /// - Request fits, sentinel does not. Installing the half that fits
865 /// leaves the request channel looking deliberately configured
866 /// while replies fall through to the unregistered-channel policy,
867 /// unbound — the H3 posture, reached by accident.
868 /// - Both fit, but no REAL reply channel does. Everything looks
869 /// installed, and then no caller can ever name a reply channel
870 /// that validates, so calls hang until they time out. Checking the
871 /// sentinel does not catch this: it is 10 bytes shorter than the
872 /// thing it stands for.
873 ///
874 /// Hence the concrete probe below. The sentinel is still what gets
875 /// STORED — it must stay unroutable — but what gets VALIDATED is a
876 /// real per-caller name.
877 pub fn install_rpc_service_defaults(&self, service: &str) {
878 let Ok(req_channel) = ChannelName::new(&format!("{service}.requests")) else {
879 return;
880 };
881 // Probe, not a channel: every origin hash renders as exactly 16
882 // hex digits, so any value answers "does a per-caller reply
883 // channel fit under this service name?" for all of them.
884 if ChannelName::new(&format!("{service}.replies.{:016x}", 0u64)).is_err() {
885 return;
886 }
887 // The sentinel name is never routed — it exists so the prefix
888 // entry has a `ChannelId` to carry. Token gates on a prefix
889 // entry evaluate against the requested CONCRETE channel (M1),
890 // not this. Kept deliberately unroutable rather than reusing the
891 // probe: `<service>.replies.0000000000000000` is a name a real
892 // caller could hold, and a sentinel should not collide with one.
893 let Ok(sentinel) = ChannelName::new(&format!("{service}.replies.prefix")) else {
894 return;
895 };
896
897 // Return values ignored on purpose: "already registered" is the
898 // operator-configured case, which is exactly what this protects.
899 let _ = self.insert_if_absent(ChannelConfig::new(ChannelId::new(req_channel)));
900 let cfg = ChannelConfig::new(ChannelId::new(sentinel))
901 .with_subscriber_origin_binding(OriginBinding::OriginHashHex16);
902 let _ = self.insert_prefix_if_absent(format!("{service}.replies."), cfg);
903 }
904
905 /// Register a channel configuration, **replacing** any existing
906 /// entry for the same canonical name.
907 ///
908 /// Callers that must not clobber an existing policy — notably
909 /// anything auto-registering a default on behalf of a subsystem —
910 /// want [`Self::insert_if_absent`] instead.
911 pub fn insert(&self, config: ChannelConfig) {
912 warn_if_fail_closed(&config, false);
913 note_if_visibility_only(&config);
914 let name = config.channel_id.name().to_string();
915 let hash = config.channel_id.hash();
916 let wire_hash = config.channel_id.wire_hash();
917 let _w = self.write_lock.lock();
918 self.configs.insert(name.clone(), config);
919 self.index_name(hash, wire_hash, name);
920 }
921
922 /// Register a channel configuration **only if that canonical name
923 /// has no entry yet**. Returns `true` if this call installed the
924 /// config, `false` if an entry already existed (which is left
925 /// untouched).
926 ///
927 /// This exists because [`Self::insert`] replaces, and a subsystem
928 /// that auto-registers a permissive default for a channel it owns
929 /// (nRPC's `<service>.requests` / `<service>.replies.`) would
930 /// otherwise silently destroy an ACL the operator installed first
931 /// — with no error and no log, leaving a posture identical to the
932 /// default. Auto-registration must be "install a default if the
933 /// operator expressed no opinion," which is exactly this
934 /// operation.
935 ///
936 /// Atomic against concurrent callers: exactly one of N racing
937 /// callers observes `true`, and its index update is not visible
938 /// before its `configs` entry.
939 pub fn insert_if_absent(&self, config: ChannelConfig) -> bool {
940 let name = config.channel_id.name().to_string();
941 let hash = config.channel_id.hash();
942 let wire_hash = config.channel_id.wire_hash();
943 let _w = self.write_lock.lock();
944 let installed = match self.configs.entry(name.clone()) {
945 dashmap::mapref::entry::Entry::Occupied(_) => false,
946 dashmap::mapref::entry::Entry::Vacant(slot) => {
947 warn_if_fail_closed(&config, false);
948 note_if_visibility_only(&config);
949 slot.insert(config);
950 true
951 }
952 };
953 if installed {
954 self.index_name(hash, wire_hash, name);
955 }
956 installed
957 }
958
959 /// Add `name` to the canonical- and wire-hash reverse indices,
960 /// skipping a name already present in either bucket.
961 ///
962 /// The de-dup is load-bearing, not hygiene. [`Self::get`] and
963 /// [`Self::remove`] treat a bucket holding more than one name as a
964 /// hash collision and return `None` to avoid applying the wrong
965 /// channel's policy. Pre-fix, `insert` pushed unconditionally, so
966 /// re-registering the *same* channel (which the SDK documents as
967 /// idempotent, and which `serve_rpc` does on every call) grew the
968 /// bucket to `[name, name]` and made `get(hash)` start returning
969 /// `None` for a channel that plainly exists — a self-inflicted
970 /// collision that disabled canonical-hash lookup for that channel.
971 fn index_name(&self, hash: ChannelHash, wire_hash: u16, name: String) {
972 let mut by_hash = self.by_hash.entry(hash).or_default();
973 if !by_hash.iter().any(|n| n == &name) {
974 by_hash.push(name.clone());
975 }
976 drop(by_hash);
977 let mut by_wire = self.by_wire_hash.entry(wire_hash).or_default();
978 if !by_wire.iter().any(|n| n == &name) {
979 by_wire.push(name);
980 }
981 }
982
983 /// Look up a channel config by canonical [`ChannelHash`] (`u64`).
984 ///
985 /// Returns `None` if the hash is unknown **or** if multiple channels
986 /// share the same canonical hash (rare at u64 — ~65 K channels before
987 /// birthday-collision threshold). Callers that need collision-safe
988 /// lookups should use [`Self::get_by_name`] with the full channel name.
989 ///
990 /// Returning `None` on collision forces callers to fall back to safe
991 /// defaults rather than silently applying the wrong channel's policy.
992 pub fn get(
993 &self,
994 channel_hash: ChannelHash,
995 ) -> Option<dashmap::mapref::one::Ref<'_, String, ChannelConfig>> {
996 let names = self.by_hash.get(&channel_hash)?;
997 // Refuse to return an arbitrary config when hashes collide.
998 if names.len() != 1 {
999 return None;
1000 }
1001 let name = names.first()?;
1002 self.configs.get(name)
1003 }
1004
1005 /// Look up a channel config by the wire `u16` fast-path hint.
1006 ///
1007 /// Returns `None` if the wire bucket is empty **or** if multiple
1008 /// channels share the same `u16` bucket (routine at scale).
1009 /// On wire-bucket collision, receive-side dispatch must fall through
1010 /// to a name-aware path; the wire hash is only a fast-path hint.
1011 pub fn get_by_wire_hash(
1012 &self,
1013 wire_hash: u16,
1014 ) -> Option<dashmap::mapref::one::Ref<'_, String, ChannelConfig>> {
1015 let names = self.by_wire_hash.get(&wire_hash)?;
1016 if names.len() != 1 {
1017 return None;
1018 }
1019 let name = names.first()?;
1020 self.configs.get(name)
1021 }
1022
1023 /// Look up a channel config by exact name (collision-safe).
1024 ///
1025 /// Falls back to the prefix registry if no exact match exists.
1026 /// Resolution is **longest-prefix-match** (the standard semantic
1027 /// for prefix tables): if both `foo.` and `foo.bar.` are
1028 /// registered and the queried name is `foo.bar.baz`, the
1029 /// `foo.bar.` config wins because it's the more specific match.
1030 /// Length ties are impossible (DashMap deduplicates keys), so
1031 /// resolution is fully deterministic across processes.
1032 ///
1033 /// Used by nRPC's dynamic reply channels — one
1034 /// `<service>.replies.` prefix admits every per-caller
1035 /// `<service>.replies.<caller_origin>` subscribe.
1036 pub fn get_by_name(
1037 &self,
1038 name: &str,
1039 ) -> Option<dashmap::mapref::one::Ref<'_, String, ChannelConfig>> {
1040 if let Some(exact) = self.configs.get(name) {
1041 return Some(exact);
1042 }
1043 self.prefix_configs
1044 .get(&self.longest_matching_prefix(name)?)
1045 }
1046
1047 /// The longest registered prefix that `name` starts with, if any.
1048 ///
1049 /// Single source of truth for prefix resolution, shared by
1050 /// [`Self::get_by_name`] and [`Self::resolve_by_name`]. Those two
1051 /// answer the same authorization question and previously each
1052 /// carried their own copy of this loop — two places to keep the
1053 /// longest-match rule correct, on the path that decides which ACL
1054 /// applies.
1055 ///
1056 /// Longest match means a more specific entry overrides a more
1057 /// general one, and makes resolution deterministic across runs: an
1058 /// earlier "first match wins" was DashMap-shard-order dependent and
1059 /// could silently flip between builds.
1060 fn longest_matching_prefix(&self, name: &str) -> Option<String> {
1061 let mut best_len = 0usize;
1062 let mut best_key: Option<String> = None;
1063 for entry in self.prefix_configs.iter() {
1064 let prefix = entry.key();
1065 if name.starts_with(prefix) && prefix.len() >= best_len {
1066 best_len = prefix.len();
1067 best_key = Some(prefix.clone());
1068 }
1069 }
1070 best_key
1071 }
1072
1073 /// Resolve `name` the same way [`Self::get_by_name`] does, but also
1074 /// report **which prefix matched** when resolution came from the
1075 /// prefix table.
1076 ///
1077 /// [`OriginBinding`] needs that prefix to locate the dynamic suffix
1078 /// inside the requested name; `get_by_name` alone discards it, and
1079 /// re-deriving it at the call site would duplicate the
1080 /// longest-match rule (and drift from it). Returns owned values
1081 /// because every caller on the authorization path clones the config
1082 /// immediately anyway, to drop the registry guard before doing
1083 /// signature work.
1084 pub fn resolve_by_name(&self, name: &str) -> Option<ResolvedConfig> {
1085 if let Some(exact) = self.configs.get(name) {
1086 return Some(ResolvedConfig {
1087 config: exact.clone(),
1088 matched_prefix: None,
1089 });
1090 }
1091 let key = self.longest_matching_prefix(name)?;
1092 let config = self.prefix_configs.get(&key)?.clone();
1093 Some(ResolvedConfig {
1094 config,
1095 matched_prefix: Some(key),
1096 })
1097 }
1098
1099 /// Remove a channel config by canonical [`ChannelHash`].
1100 ///
1101 /// Returns `None` if the hash is unknown **or** if multiple channels
1102 /// share the same canonical hash — mirroring the collision-safe
1103 /// semantics of `get()`. Removing an arbitrary config on collision
1104 /// would silently delete the wrong channel's policy (e.g. dropping a
1105 /// `SubnetLocal` entry and leaving a `Global` sibling in place).
1106 ///
1107 /// Callers that need to remove a specific channel should use
1108 /// [`remove_by_name`](Self::remove_by_name).
1109 pub fn remove(&self, channel_hash: ChannelHash) -> Option<ChannelConfig> {
1110 // One critical section covering the index read AND the removal
1111 // it selects, so the name cannot be replaced by a different
1112 // channel in between and get removed in its place.
1113 let _w = self.write_lock.lock();
1114 let name = {
1115 let names = self.by_hash.get(&channel_hash)?;
1116 if names.len() != 1 {
1117 return None;
1118 }
1119 names.first()?.clone()
1120 };
1121 self.remove_by_name_locked(&name)
1122 }
1123
1124 /// Remove a channel config by exact name (collision-safe).
1125 ///
1126 /// Returns the removed config if it existed.
1127 pub fn remove_by_name(&self, name: &str) -> Option<ChannelConfig> {
1128 let _w = self.write_lock.lock();
1129 self.remove_by_name_locked(name)
1130 }
1131
1132 /// [`Self::remove_by_name`] for callers already holding
1133 /// `write_lock`. Split out because `parking_lot::Mutex` is not
1134 /// reentrant and [`Self::remove`] must hold the lock across its
1135 /// index lookup.
1136 ///
1137 /// Under the lock, `configs.remove` and the index cleanup are one
1138 /// step, which is what makes the pair sound. Previously they were
1139 /// not, and the repair each defect needed reintroduced the other:
1140 ///
1141 /// - Without a repair pass, a re-registration landing between the
1142 /// `configs.remove` and the `retain` had its fresh index entry
1143 /// deleted — the channel present in `configs`, invisible to
1144 /// `get(hash)`.
1145 /// - With one (`if configs.contains_key(name) { index_name(..) }`),
1146 /// a second removal completing between that test and the re-index
1147 /// put a name back into the bucket with nothing behind it. `get`
1148 /// and `remove` treat a bucket holding more than one name as a
1149 /// hash collision and answer `None`, so a phantom entry disables
1150 /// lookup for whatever real channel shares the bucket.
1151 ///
1152 /// Serializing removes the interleaving both were patching around,
1153 /// so neither the repair nor its own failure mode remains.
1154 fn remove_by_name_locked(&self, name: &str) -> Option<ChannelConfig> {
1155 let (_, removed) = self.configs.remove(name)?;
1156 let hash = removed.channel_id.hash();
1157 let wire_hash = removed.channel_id.wire_hash();
1158 if let Some(mut hash_names) = self.by_hash.get_mut(&hash) {
1159 hash_names.retain(|n| n != name);
1160 }
1161 if let Some(mut wire_names) = self.by_wire_hash.get_mut(&wire_hash) {
1162 wire_names.retain(|n| n != name);
1163 }
1164 Some(removed)
1165 }
1166
1167 /// Number of registered channels.
1168 pub fn len(&self) -> usize {
1169 self.configs.len()
1170 }
1171
1172 /// Check if empty.
1173 pub fn is_empty(&self) -> bool {
1174 self.configs.is_empty()
1175 }
1176
1177 /// Snapshot every registered channel as `(name, config)` pairs,
1178 /// sorted by name for stable operator-tool output. Walks the
1179 /// exact-match table only — prefix entries are excluded
1180 /// because their `channel_id.name()` is a sentinel rather
1181 /// than a routable channel.
1182 ///
1183 /// O(N) clone — N is the registry size (typically tens to a
1184 /// few hundred). Suitable for `net channel ls` / Deck-panel
1185 /// renders, not for hot-path use.
1186 pub fn snapshot(&self) -> Vec<(String, ChannelConfig)> {
1187 let mut out: Vec<(String, ChannelConfig)> = self
1188 .configs
1189 .iter()
1190 .map(|e| (e.key().clone(), e.value().clone()))
1191 .collect();
1192 out.sort_by(|a, b| a.0.cmp(&b.0));
1193 out
1194 }
1195
1196 /// Same as [`Self::snapshot`] but for prefix entries — emits
1197 /// `(prefix, config)` pairs for every prefix registered via
1198 /// [`Self::insert_prefix`]. Sorted by prefix for stable
1199 /// output.
1200 pub fn snapshot_prefixes(&self) -> Vec<(String, ChannelConfig)> {
1201 let mut out: Vec<(String, ChannelConfig)> = self
1202 .prefix_configs
1203 .iter()
1204 .map(|e| (e.key().clone(), e.value().clone()))
1205 .collect();
1206 out.sort_by(|a, b| a.0.cmp(&b.0));
1207 out
1208 }
1209
1210 /// Get the priority for a channel (0 if not configured).
1211 #[inline]
1212 pub fn priority(&self, channel_hash: ChannelHash) -> u8 {
1213 self.get(channel_hash).map(|c| c.priority).unwrap_or(0)
1214 }
1215}
1216
1217impl Default for ChannelConfigRegistry {
1218 fn default() -> Self {
1219 Self::new()
1220 }
1221}
1222
1223impl std::fmt::Debug for ChannelConfigRegistry {
1224 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1225 f.debug_struct("ChannelConfigRegistry")
1226 .field("channels", &self.configs.len())
1227 .finish()
1228 }
1229}
1230
1231#[cfg(test)]
1232mod tests {
1233 use super::*;
1234 use crate::adapter::net::behavior::capability::{GpuInfo, GpuVendor, HardwareCapabilities};
1235 use crate::adapter::net::channel::{channel_hash, queue_group_hash, ChannelName};
1236 use crate::adapter::net::identity::{EntityKeypair, PermissionToken};
1237
1238 fn make_caps(gpu: bool) -> CapabilitySet {
1239 if gpu {
1240 let gpu_info = GpuInfo {
1241 vendor: GpuVendor::Nvidia,
1242 model: "test".to_string(),
1243 vram_gb: 8,
1244 compute_units: 0,
1245 tensor_cores: 0,
1246 fp16_tflops_x10: 0,
1247 };
1248 CapabilitySet::new().with_hardware(HardwareCapabilities::new().with_gpu(gpu_info))
1249 } else {
1250 CapabilitySet::new()
1251 }
1252 }
1253
1254 /// One-link chain wrapping a token directly issued by `issuer` to
1255 /// `subject`.
1256 fn direct_chain(
1257 issuer: &EntityKeypair,
1258 subject: &EntityKeypair,
1259 scope: TokenScope,
1260 channel_hash: ChannelHash,
1261 ) -> TokenChain {
1262 TokenChain::single(PermissionToken::issue(
1263 issuer,
1264 subject.entity_id().clone(),
1265 scope,
1266 channel_hash,
1267 3600,
1268 0,
1269 ))
1270 }
1271
1272 #[test]
1273 fn test_open_channel() {
1274 let id = ChannelId::parse("sensors/lidar").unwrap();
1275 let config = ChannelConfig::new(id);
1276 let caps = make_caps(false);
1277 let entity = EntityKeypair::generate();
1278 let rev = RevocationRegistry::new();
1279
1280 assert!(config.can_publish(
1281 &caps,
1282 entity.entity_id(),
1283 config.channel_id.hash(),
1284 None,
1285 &rev,
1286 0
1287 ));
1288 assert!(config.can_subscribe(
1289 &caps,
1290 entity.entity_id(),
1291 config.channel_id.hash(),
1292 None,
1293 &rev,
1294 0
1295 ));
1296 }
1297
1298 #[test]
1299 fn test_capability_restricted_channel() {
1300 let id = ChannelId::parse("compute/gpu-tasks").unwrap();
1301 let config =
1302 ChannelConfig::new(id).with_publish_caps(CapabilityFilter::new().require_gpu());
1303
1304 let entity = EntityKeypair::generate();
1305 let rev = RevocationRegistry::new();
1306
1307 let no_gpu = make_caps(false);
1308 assert!(!config.can_publish(
1309 &no_gpu,
1310 entity.entity_id(),
1311 config.channel_id.hash(),
1312 None,
1313 &rev,
1314 0
1315 ));
1316
1317 let with_gpu = make_caps(true);
1318 assert!(config.can_publish(
1319 &with_gpu,
1320 entity.entity_id(),
1321 config.channel_id.hash(),
1322 None,
1323 &rev,
1324 0
1325 ));
1326 }
1327
1328 /// The C1 fix: a `require_token` channel anchored to an owner must
1329 /// reject a self-issued token and accept an owner-issued one.
1330 #[test]
1331 fn token_channel_rejects_self_issued_accepts_owner_issued() {
1332 let id = ChannelId::parse("control/estop").unwrap();
1333 let owner = EntityKeypair::generate();
1334 let subject = EntityKeypair::generate();
1335 let config =
1336 ChannelConfig::new(id.clone()).with_token_roots(vec![owner.entity_id().clone()]);
1337 let caps = make_caps(false);
1338 let rev = RevocationRegistry::new();
1339
1340 // No chain -> denied.
1341 assert!(!config.can_publish(
1342 &caps,
1343 subject.entity_id(),
1344 config.channel_id.hash(),
1345 None,
1346 &rev,
1347 0
1348 ));
1349
1350 // Self-issued (issuer == subject, NOT the channel owner) ->
1351 // denied. Pre-fix this was the privilege-escalation hole:
1352 // `verify()` + `TokenCache::check` accepted any self-consistent
1353 // token regardless of issuer.
1354 let self_chain = direct_chain(&subject, &subject, TokenScope::PUBLISH, id.hash());
1355 assert!(
1356 !config.can_publish(
1357 &caps,
1358 subject.entity_id(),
1359 config.channel_id.hash(),
1360 Some(&self_chain),
1361 &rev,
1362 0
1363 ),
1364 "self-issued token must be rejected: its issuer is not a channel root"
1365 );
1366
1367 // Owner-issued -> allowed.
1368 let owner_chain = direct_chain(&owner, &subject, TokenScope::PUBLISH, id.hash());
1369 assert!(config.can_publish(
1370 &caps,
1371 subject.entity_id(),
1372 config.channel_id.hash(),
1373 Some(&owner_chain),
1374 &rev,
1375 0
1376 ));
1377 }
1378
1379 /// `with_require_token(true)` without any roots fails closed — there
1380 /// is no authority a chain could anchor to.
1381 #[test]
1382 fn require_token_with_no_roots_fails_closed() {
1383 let id = ChannelId::parse("control/locked").unwrap();
1384 let config = ChannelConfig::new(id.clone()).with_require_token(true);
1385 let caps = make_caps(false);
1386 let rev = RevocationRegistry::new();
1387 let anyone = EntityKeypair::generate();
1388
1389 // Even an otherwise-well-formed token can't anchor to nothing.
1390 let chain = direct_chain(&anyone, &anyone, TokenScope::SUBSCRIBE, id.hash());
1391 assert!(!config.can_subscribe(
1392 &caps,
1393 anyone.entity_id(),
1394 config.channel_id.hash(),
1395 Some(&chain),
1396 &rev,
1397 0
1398 ));
1399 assert!(!config.can_subscribe(
1400 &caps,
1401 anyone.entity_id(),
1402 config.channel_id.hash(),
1403 None,
1404 &rev,
1405 0
1406 ));
1407 }
1408
1409 /// A config that names roots but never set the `require_token`
1410 /// flag (e.g. built field-by-field rather than via
1411 /// `with_token_roots`) must still enforce. Pre-fix the gate keyed
1412 /// only off `require_token`, so this drifted-open config silently
1413 /// admitted every peer.
1414 #[test]
1415 fn roots_without_require_token_flag_still_enforces() {
1416 let id = ChannelId::parse("control/estop").unwrap();
1417 let owner = EntityKeypair::generate();
1418 let subject = EntityKeypair::generate();
1419 let caps = make_caps(false);
1420 let rev = RevocationRegistry::new();
1421
1422 let mut config = ChannelConfig::new(id.clone());
1423 config.token_roots = vec![owner.entity_id().clone()];
1424 // Deliberately leave `require_token` false — the two fields are
1425 // both public and can drift out of sync.
1426 assert!(!config.require_token);
1427 assert!(
1428 config.token_required(),
1429 "named roots must imply enforcement"
1430 );
1431
1432 // No chain -> denied (would have been silently admitted pre-fix).
1433 assert!(!config.can_subscribe(
1434 &caps,
1435 subject.entity_id(),
1436 config.channel_id.hash(),
1437 None,
1438 &rev,
1439 0
1440 ));
1441 // Owner-issued chain -> allowed.
1442 let owner_chain = direct_chain(&owner, &subject, TokenScope::SUBSCRIBE, id.hash());
1443 assert!(config.can_subscribe(
1444 &caps,
1445 subject.entity_id(),
1446 config.channel_id.hash(),
1447 Some(&owner_chain),
1448 &rev,
1449 0
1450 ));
1451 }
1452
1453 /// The chain's leaf must be bound to the presenting entity — a peer
1454 /// can't replay a chain minted for someone else.
1455 #[test]
1456 fn leaf_subject_must_match_presenter() {
1457 let id = ChannelId::parse("control/estop").unwrap();
1458 let owner = EntityKeypair::generate();
1459 let intended = EntityKeypair::generate();
1460 let attacker = EntityKeypair::generate();
1461 let config =
1462 ChannelConfig::new(id.clone()).with_token_roots(vec![owner.entity_id().clone()]);
1463 let caps = make_caps(false);
1464 let rev = RevocationRegistry::new();
1465
1466 // Owner issued this to `intended`; `attacker` presents it.
1467 let chain = direct_chain(&owner, &intended, TokenScope::SUBSCRIBE, id.hash());
1468 assert!(!config.can_subscribe(
1469 &caps,
1470 attacker.entity_id(),
1471 config.channel_id.hash(),
1472 Some(&chain),
1473 &rev,
1474 0
1475 ));
1476 // The intended subject is accepted.
1477 assert!(config.can_subscribe(
1478 &caps,
1479 intended.entity_id(),
1480 config.channel_id.hash(),
1481 Some(&chain),
1482 &rev,
1483 0
1484 ));
1485 }
1486
1487 /// A valid owner → intermediate → leaf delegation chain is accepted;
1488 /// scope narrows correctly down the chain.
1489 #[test]
1490 fn delegation_chain_accepted() {
1491 let id = ChannelId::parse("fleet/telemetry").unwrap();
1492 let owner = EntityKeypair::generate();
1493 let mid = EntityKeypair::generate();
1494 let leaf = EntityKeypair::generate();
1495 let config =
1496 ChannelConfig::new(id.clone()).with_token_roots(vec![owner.entity_id().clone()]);
1497 let caps = make_caps(false);
1498 let rev = RevocationRegistry::new();
1499
1500 // Owner grants `mid` SUBSCRIBE + DELEGATE, depth 2.
1501 let root = PermissionToken::issue(
1502 &owner,
1503 mid.entity_id().clone(),
1504 TokenScope::SUBSCRIBE.union(TokenScope::DELEGATE),
1505 id.hash(),
1506 3600,
1507 2,
1508 );
1509 // `mid` delegates SUBSCRIBE to `leaf` (drops DELEGATE).
1510 let child = root
1511 .delegate(&mid, leaf.entity_id().clone(), TokenScope::SUBSCRIBE)
1512 .expect("delegation should succeed");
1513 let chain = TokenChain {
1514 tokens: vec![root, child],
1515 };
1516 assert!(config.can_subscribe(
1517 &caps,
1518 leaf.entity_id(),
1519 config.channel_id.hash(),
1520 Some(&chain),
1521 &rev,
1522 0
1523 ));
1524 }
1525
1526 /// A chain whose links don't connect (`child.issuer != parent.subject`)
1527 /// is rejected — no splicing an unrelated token onto a real root.
1528 #[test]
1529 fn delegation_broken_continuity_rejected() {
1530 let id = ChannelId::parse("fleet/telemetry").unwrap();
1531 let owner = EntityKeypair::generate();
1532 let mid = EntityKeypair::generate();
1533 let rogue = EntityKeypair::generate();
1534 let leaf = EntityKeypair::generate();
1535 let config =
1536 ChannelConfig::new(id.clone()).with_token_roots(vec![owner.entity_id().clone()]);
1537 let caps = make_caps(false);
1538 let rev = RevocationRegistry::new();
1539
1540 // Real owner→mid root link.
1541 let root = PermissionToken::issue(
1542 &owner,
1543 mid.entity_id().clone(),
1544 TokenScope::SUBSCRIBE.union(TokenScope::DELEGATE),
1545 id.hash(),
1546 3600,
1547 2,
1548 );
1549 // Spliced second link issued by `rogue` (NOT `mid`), so
1550 // child.issuer (rogue) != root.subject (mid).
1551 let spliced = PermissionToken::issue(
1552 &rogue,
1553 leaf.entity_id().clone(),
1554 TokenScope::SUBSCRIBE,
1555 id.hash(),
1556 3600,
1557 0,
1558 );
1559 let chain = TokenChain {
1560 tokens: vec![root, spliced],
1561 };
1562 assert!(!config.can_subscribe(
1563 &caps,
1564 leaf.entity_id(),
1565 config.channel_id.hash(),
1566 Some(&chain),
1567 &rev,
1568 0
1569 ));
1570 }
1571
1572 /// A delegated child can't authorize a scope its parent lacked —
1573 /// chain authority is the intersection of all links.
1574 #[test]
1575 fn delegation_cannot_broaden_scope() {
1576 let id = ChannelId::parse("fleet/telemetry").unwrap();
1577 let owner = EntityKeypair::generate();
1578 let mid = EntityKeypair::generate();
1579 let leaf = EntityKeypair::generate();
1580 let config =
1581 ChannelConfig::new(id.clone()).with_token_roots(vec![owner.entity_id().clone()]);
1582 let caps = make_caps(false);
1583 let rev = RevocationRegistry::new();
1584
1585 // Owner grants `mid` only SUBSCRIBE + DELEGATE — no PUBLISH.
1586 let root = PermissionToken::issue(
1587 &owner,
1588 mid.entity_id().clone(),
1589 TokenScope::SUBSCRIBE.union(TokenScope::DELEGATE),
1590 id.hash(),
1591 3600,
1592 2,
1593 );
1594 // `mid` forges a child claiming PUBLISH (which it never held).
1595 // `delegate` would intersect it away, so mint the child by hand
1596 // to simulate a malicious intermediate.
1597 let forged_child = PermissionToken::issue(
1598 &mid,
1599 leaf.entity_id().clone(),
1600 TokenScope::PUBLISH,
1601 id.hash(),
1602 3600,
1603 0,
1604 );
1605 let chain = TokenChain {
1606 tokens: vec![root, forged_child],
1607 };
1608 // The root link doesn't authorize PUBLISH, so the chain can't.
1609 assert!(!config.can_publish(
1610 &caps,
1611 leaf.entity_id(),
1612 config.channel_id.hash(),
1613 Some(&chain),
1614 &rev,
1615 0
1616 ));
1617 }
1618
1619 /// The H1 fix: revoking the root issuer invalidates the whole chain,
1620 /// including offline-delegated descendants, because the root grant
1621 /// is itself a verified link.
1622 #[test]
1623 fn root_revocation_kills_delegated_chain() {
1624 let id = ChannelId::parse("fleet/telemetry").unwrap();
1625 let owner = EntityKeypair::generate();
1626 let mid = EntityKeypair::generate();
1627 let leaf = EntityKeypair::generate();
1628 let config =
1629 ChannelConfig::new(id.clone()).with_token_roots(vec![owner.entity_id().clone()]);
1630 let caps = make_caps(false);
1631 let rev = RevocationRegistry::new();
1632
1633 let root = PermissionToken::issue(
1634 &owner,
1635 mid.entity_id().clone(),
1636 TokenScope::SUBSCRIBE.union(TokenScope::DELEGATE),
1637 id.hash(),
1638 3600,
1639 2,
1640 );
1641 let child = root
1642 .delegate(&mid, leaf.entity_id().clone(), TokenScope::SUBSCRIBE)
1643 .expect("delegation should succeed");
1644 let chain = TokenChain {
1645 tokens: vec![root, child],
1646 };
1647
1648 // Accepted before revocation.
1649 assert!(config.can_subscribe(
1650 &caps,
1651 leaf.entity_id(),
1652 config.channel_id.hash(),
1653 Some(&chain),
1654 &rev,
1655 0
1656 ));
1657
1658 // Owner bumps its revocation floor above the chain's generation
1659 // (0). The root link falls below the floor → whole chain dies,
1660 // even though the delegated child's issuer is `mid`, not `owner`.
1661 rev.revoke_below(owner.entity_id(), 1);
1662 assert!(
1663 !config.can_subscribe(
1664 &caps,
1665 leaf.entity_id(),
1666 config.channel_id.hash(),
1667 Some(&chain),
1668 &rev,
1669 0
1670 ),
1671 "revoking the root must kill the delegated descendant"
1672 );
1673 }
1674
1675 #[test]
1676 fn test_caps_and_token_combined() {
1677 let id = ChannelId::parse("compute/secure").unwrap();
1678 let owner = EntityKeypair::generate();
1679 let subject = EntityKeypair::generate();
1680 let config = ChannelConfig::new(id.clone())
1681 .with_publish_caps(CapabilityFilter::new().require_gpu())
1682 .with_token_roots(vec![owner.entity_id().clone()]);
1683 let rev = RevocationRegistry::new();
1684
1685 let owner_chain = direct_chain(&owner, &subject, TokenScope::PUBLISH, id.hash());
1686
1687 // Has GPU but no token -> denied.
1688 let with_gpu = make_caps(true);
1689 assert!(!config.can_publish(
1690 &with_gpu,
1691 subject.entity_id(),
1692 config.channel_id.hash(),
1693 None,
1694 &rev,
1695 0
1696 ));
1697
1698 // Has token but no GPU -> denied.
1699 let no_gpu = make_caps(false);
1700 assert!(!config.can_publish(
1701 &no_gpu,
1702 subject.entity_id(),
1703 config.channel_id.hash(),
1704 Some(&owner_chain),
1705 &rev,
1706 0
1707 ));
1708
1709 // Has both -> allowed.
1710 assert!(config.can_publish(
1711 &with_gpu,
1712 subject.entity_id(),
1713 config.channel_id.hash(),
1714 Some(&owner_chain),
1715 &rev,
1716 0
1717 ));
1718 }
1719
1720 #[test]
1721 fn test_config_registry() {
1722 let reg = ChannelConfigRegistry::new();
1723 let id = ChannelId::parse("sensors/lidar").unwrap();
1724 let config = ChannelConfig::new(id.clone()).with_priority(5);
1725
1726 reg.insert(config);
1727 assert_eq!(reg.len(), 1);
1728 assert_eq!(reg.priority(id.hash()), 5);
1729
1730 let retrieved = reg.get(id.hash()).unwrap();
1731 assert_eq!(retrieved.priority, 5);
1732 }
1733
1734 #[test]
1735 fn test_visibility_default() {
1736 let id = ChannelId::parse("test").unwrap();
1737 let config = ChannelConfig::new(id);
1738 assert_eq!(config.visibility, Visibility::Global);
1739 }
1740
1741 #[test]
1742 fn snapshot_returns_sorted_exact_matches_excludes_prefixes() {
1743 // Pin the operator-tool surface: `snapshot` yields every
1744 // exact-match channel in lex order, and `snapshot_prefixes`
1745 // is a sibling for the prefix table — exact-matches and
1746 // prefixes don't mix.
1747 let reg = ChannelConfigRegistry::new();
1748 let zeta = ChannelConfig::new(ChannelId::parse("zeta/c").unwrap())
1749 .with_visibility(Visibility::SubnetLocal);
1750 let alpha = ChannelConfig::new(ChannelId::parse("alpha/a").unwrap())
1751 .with_visibility(Visibility::Global);
1752 let middle = ChannelConfig::new(ChannelId::parse("middle/b").unwrap());
1753 reg.insert(zeta);
1754 reg.insert(alpha);
1755 reg.insert(middle);
1756 reg.insert_prefix(
1757 "rpc.replies.",
1758 ChannelConfig::new(ChannelId::parse("rpc.replies.").unwrap()),
1759 );
1760
1761 let snap = reg.snapshot();
1762 let names: Vec<&str> = snap.iter().map(|(n, _)| n.as_str()).collect();
1763 assert_eq!(names, vec!["alpha/a", "middle/b", "zeta/c"]);
1764 // Prefix entries excluded from `snapshot`.
1765 assert!(!names.contains(&"rpc.replies."));
1766 // Per-entry visibility round-trips.
1767 let alpha_cfg = snap.iter().find(|(n, _)| n == "alpha/a").unwrap();
1768 assert_eq!(alpha_cfg.1.visibility, Visibility::Global);
1769 let zeta_cfg = snap.iter().find(|(n, _)| n == "zeta/c").unwrap();
1770 assert_eq!(zeta_cfg.1.visibility, Visibility::SubnetLocal);
1771
1772 let prefixes = reg.snapshot_prefixes();
1773 let prefix_names: Vec<&str> = prefixes.iter().map(|(p, _)| p.as_str()).collect();
1774 assert_eq!(prefix_names, vec!["rpc.replies."]);
1775 }
1776
1777 #[test]
1778 fn test_regression_config_registry_hash_collision_no_overwrite() {
1779 // Regression: ChannelConfigRegistry used u16 hash as the key,
1780 // so two channels with the same hash silently overwrote each
1781 // other's configs — including visibility and security policies.
1782 // With only 65536 hashes, the birthday paradox makes collisions
1783 // likely at ~300 channels.
1784 //
1785 // Fix: keyed by channel name with a hash→names reverse index.
1786 let reg = ChannelConfigRegistry::new();
1787
1788 let id1 = ChannelId::parse("channel/alpha").unwrap();
1789 let id2 = ChannelId::parse("channel/beta").unwrap();
1790
1791 let config1 = ChannelConfig::new(id1.clone()).with_priority(1);
1792 let config2 = ChannelConfig::new(id2.clone()).with_priority(2);
1793
1794 reg.insert(config1);
1795 reg.insert(config2);
1796
1797 // Both configs should be present regardless of hash collision
1798 assert_eq!(reg.len(), 2, "both channels should exist in registry");
1799
1800 // Each should retain its own priority
1801 let c1 = reg.get_by_name("channel/alpha").unwrap();
1802 assert_eq!(c1.priority, 1, "channel/alpha priority should be 1");
1803 let c2 = reg.get_by_name("channel/beta").unwrap();
1804 assert_eq!(c2.priority, 2, "channel/beta priority should be 2");
1805 }
1806
1807 #[test]
1808 fn test_regression_config_registry_get_returns_none_on_collision() {
1809 // Regression: get() returned an arbitrary config when multiple
1810 // channels shared the same u16 hash. A SubnetLocal channel
1811 // colliding with a Global channel could silently receive the
1812 // wrong visibility policy, leaking traffic across subnet
1813 // boundaries.
1814 //
1815 // Fix: get() returns None when the hash maps to more than one
1816 // channel name. Callers fall back to safe defaults or use
1817 // get_by_name() for collision-safe lookups.
1818 use crate::adapter::net::channel::name::wire_channel_hash;
1819
1820 // Find two valid channel names that produce the same wire `u16`
1821 // hash. With 65 536 possible values, birthday paradox gives a
1822 // collision within ~300 names on average. (Canonical `u32`
1823 // collisions are rare enough — ~65 K names — that exercising
1824 // them in tests would be slow; the wire-hash bucket is the
1825 // observable collision surface here.)
1826 let mut seen = std::collections::HashMap::<u16, String>::new();
1827 let (name1, name2) = loop {
1828 let name = format!("ch-{}", seen.len());
1829 let wire = wire_channel_hash(&name);
1830 if let Some(existing) = seen.get(&wire) {
1831 break (existing.clone(), name);
1832 }
1833 seen.insert(wire, name);
1834 };
1835
1836 let reg = ChannelConfigRegistry::new();
1837 let id1 = ChannelId::parse(&name1).unwrap();
1838 let id2 = ChannelId::parse(&name2).unwrap();
1839 assert_eq!(
1840 id1.wire_hash(),
1841 id2.wire_hash(),
1842 "precondition: wire hashes must collide"
1843 );
1844
1845 // Insert a SubnetLocal channel and a Global channel that collide
1846 let config1 = ChannelConfig::new(id1.clone()).with_visibility(Visibility::SubnetLocal);
1847 let config2 = ChannelConfig::new(id2.clone()).with_visibility(Visibility::Global);
1848 reg.insert(config1);
1849 reg.insert(config2);
1850
1851 // get_by_wire_hash() must return None — not an arbitrary
1852 // config — on a wire-bucket collision.
1853 assert!(
1854 reg.get_by_wire_hash(id1.wire_hash()).is_none(),
1855 "get_by_wire_hash() must return None when wire hashes collide between channels"
1856 );
1857
1858 // The canonical-hash path stays unaffected: each name has a
1859 // distinct canonical [`ChannelHash`] (collision-resistant at
1860 // u32), so `get(canonical)` resolves uniquely.
1861 assert_eq!(
1862 reg.get(id1.hash()).unwrap().visibility,
1863 Visibility::SubnetLocal
1864 );
1865 assert_eq!(reg.get(id2.hash()).unwrap().visibility, Visibility::Global);
1866
1867 // get_by_name() must still work for each channel individually
1868 let c1 = reg.get_by_name(&name1).unwrap();
1869 assert_eq!(c1.visibility, Visibility::SubnetLocal);
1870 let c2 = reg.get_by_name(&name2).unwrap();
1871 assert_eq!(c2.visibility, Visibility::Global);
1872 }
1873
1874 #[test]
1875 fn test_regression_remove_by_wire_hash_safe_on_wire_collision() {
1876 // Regression: the wire-keyed remove path used to silently
1877 // delete the first name bucketed under a colliding `u16` wire
1878 // hash, swapping policies between unrelated channels. With
1879 // the substrate-wide widening to canonical [`ChannelHash`]
1880 // (`u32`), the primary `remove(hash)` keys on the canonical
1881 // value (unique per name); the wire-bucket collision space
1882 // is exercised below via two names that share a `u16` bucket
1883 // and asserts each name is independently addressable through
1884 // both `remove(canonical)` and `remove_by_name`.
1885 use crate::adapter::net::channel::name::wire_channel_hash;
1886
1887 let mut seen = std::collections::HashMap::<u16, String>::new();
1888 let (name1, name2) = loop {
1889 let name = format!("rm-{}", seen.len());
1890 let wire = wire_channel_hash(&name);
1891 if let Some(existing) = seen.get(&wire) {
1892 break (existing.clone(), name);
1893 }
1894 seen.insert(wire, name);
1895 };
1896
1897 let reg = ChannelConfigRegistry::new();
1898 let id1 = ChannelId::parse(&name1).unwrap();
1899 let id2 = ChannelId::parse(&name2).unwrap();
1900 assert_eq!(
1901 id1.wire_hash(),
1902 id2.wire_hash(),
1903 "precondition: wire hashes must collide"
1904 );
1905
1906 reg.insert(ChannelConfig::new(id1.clone()).with_visibility(Visibility::SubnetLocal));
1907 reg.insert(ChannelConfig::new(id2.clone()).with_visibility(Visibility::Global));
1908
1909 // Canonical `remove(hash)` keys on the u32 canonical hash,
1910 // which is unique per name, so each config is removable
1911 // individually even under a wire-bucket collision.
1912 let removed1 = reg.remove(id1.hash()).expect("remove canonical1");
1913 assert_eq!(removed1.visibility, Visibility::SubnetLocal);
1914 assert_eq!(reg.len(), 1, "the other config must still be present");
1915 assert_eq!(
1916 reg.get_by_name(&name2).unwrap().visibility,
1917 Visibility::Global,
1918 "name2 must be untouched by the canonical remove of name1"
1919 );
1920
1921 // `remove_by_name` is the explicit-collision-safe path used
1922 // by callers that already hold the name string; it must
1923 // continue to work alongside the canonical-hash path.
1924 let removed2 = reg.remove_by_name(&name2).unwrap();
1925 assert_eq!(removed2.visibility, Visibility::Global);
1926 assert_eq!(reg.len(), 0);
1927 }
1928
1929 #[test]
1930 fn prefix_resolution_picks_longest_match_deterministically() {
1931 // Regression: prior `get_by_name` used DashMap iteration
1932 // order to pick "first matching prefix wins", which is shard-
1933 // order dependent and non-deterministic across processes.
1934 // With both `foo.` and `foo.bar.` registered against
1935 // `foo.bar.baz`, the longer (more specific) prefix must win.
1936 let reg = ChannelConfigRegistry::new();
1937 reg.insert_prefix(
1938 "foo.",
1939 ChannelConfig::new(ChannelId::parse("foo.sentinel").unwrap()).with_priority(1),
1940 );
1941 reg.insert_prefix(
1942 "foo.bar.",
1943 ChannelConfig::new(ChannelId::parse("foo.bar.sentinel").unwrap()).with_priority(2),
1944 );
1945 reg.insert_prefix(
1946 "foo.bar.baz.",
1947 ChannelConfig::new(ChannelId::parse("foo.bar.baz.sentinel").unwrap()).with_priority(3),
1948 );
1949
1950 // Most-specific match wins regardless of insertion order.
1951 let c = reg.get_by_name("foo.bar.baz.qux").unwrap();
1952 assert_eq!(c.priority, 3, "longest matching prefix must win");
1953
1954 // Slightly shorter target — `foo.bar.baz.` no longer matches
1955 // (target doesn't start with the trailing dot), so `foo.bar.`
1956 // wins.
1957 let c = reg.get_by_name("foo.bar.something").unwrap();
1958 assert_eq!(c.priority, 2);
1959
1960 // Shortest matching prefix wins when no others apply.
1961 let c = reg.get_by_name("foo.something").unwrap();
1962 assert_eq!(c.priority, 1);
1963
1964 // No match.
1965 assert!(reg.get_by_name("other.thing").is_none());
1966
1967 // Run the lookup many times; result must be stable.
1968 for _ in 0..100 {
1969 assert_eq!(reg.get_by_name("foo.bar.baz.x").unwrap().priority, 3);
1970 }
1971 }
1972
1973 #[test]
1974 fn test_remove_by_hash_works_when_unique() {
1975 // Baseline: `remove(hash)` still works for the common non-collision
1976 // case — only refuses when ambiguous.
1977 let reg = ChannelConfigRegistry::new();
1978 let id = ChannelId::parse("sensors/only").unwrap();
1979 let hash = id.hash();
1980 reg.insert(ChannelConfig::new(id).with_priority(7));
1981
1982 let removed = reg.remove(hash).unwrap();
1983 assert_eq!(removed.priority, 7);
1984 assert_eq!(reg.len(), 0);
1985 assert!(reg.get(hash).is_none());
1986 }
1987
1988 // ---- Review follow-ups: registry index consistency ----
1989
1990 /// A remove that interleaves with a re-registration must not leave
1991 /// the NEW config stranded — present in `configs` but invisible to
1992 /// the reverse indices that `get(hash)` / `get_by_wire_hash`
1993 /// resolve through.
1994 ///
1995 /// Sequenced deterministically rather than raced, so it pins the
1996 /// outcome rather than an interleaving: remove-then-reregister and
1997 /// reregister-then-remove must both leave the index agreeing with
1998 /// `configs`. (The interleaving itself can no longer occur —
1999 /// `remove_by_name_locked` holds the registry write lock across both
2000 /// steps — but the property is what callers depend on, and it should
2001 /// keep being asserted independently of how it is achieved.)
2002 #[test]
2003 fn remove_racing_reregistration_leaves_the_index_consistent() {
2004 let reg = ChannelConfigRegistry::new();
2005 let id = ChannelId::parse("svc.requests").unwrap();
2006 let hash = id.hash();
2007 let wire = id.wire_hash();
2008
2009 reg.insert(ChannelConfig::new(id.clone()).with_priority(1));
2010 // Re-register, then remove: the remove's cleanup targets a name
2011 // that is legitimately present again.
2012 reg.insert(ChannelConfig::new(id.clone()).with_priority(2));
2013 reg.remove_by_name("svc.requests");
2014 assert!(
2015 reg.get(hash).is_none(),
2016 "after a completed remove the channel is gone"
2017 );
2018
2019 // Now the interleaved shape: removed, then re-registered.
2020 reg.insert(ChannelConfig::new(id).with_priority(3));
2021 assert_eq!(
2022 reg.get(hash).map(|c| c.priority),
2023 Some(3),
2024 "a channel re-registered after removal must be reachable by \
2025 canonical hash"
2026 );
2027 assert_eq!(
2028 reg.get_by_wire_hash(wire).map(|c| c.priority),
2029 Some(3),
2030 "…and by wire hash"
2031 );
2032 }
2033
2034 /// The concurrent form of the same property. Whatever the
2035 /// interleaving, the registry must not end with a config that
2036 /// `get_by_name` finds but `get(hash)` cannot.
2037 #[test]
2038 fn concurrent_remove_and_reregister_never_strands_the_index() {
2039 use std::sync::Arc as StdArc;
2040
2041 for _ in 0..64 {
2042 let reg = StdArc::new(ChannelConfigRegistry::new());
2043 let id = ChannelId::parse("svc.requests").unwrap();
2044 let hash = id.hash();
2045 reg.insert(ChannelConfig::new(id.clone()));
2046
2047 std::thread::scope(|s| {
2048 let r1 = reg.clone();
2049 s.spawn(move || {
2050 r1.remove_by_name("svc.requests");
2051 });
2052 let r2 = reg.clone();
2053 let id2 = id.clone();
2054 s.spawn(move || {
2055 r2.insert(ChannelConfig::new(id2).with_priority(9));
2056 });
2057 });
2058
2059 // The invariant: `configs` and the reverse index agree.
2060 if reg.get_by_name("svc.requests").is_some() {
2061 assert!(
2062 reg.get(hash).is_some(),
2063 "config present by name but unreachable by canonical \
2064 hash — the reverse index was stranded"
2065 );
2066 }
2067 }
2068 }
2069
2070 /// Both resolution entry points must agree, since they answer the
2071 /// same authorization question. Pinned because they used to carry
2072 /// separate copies of the longest-match loop.
2073 #[test]
2074 fn get_by_name_and_resolve_by_name_agree_on_prefix_resolution() {
2075 let reg = ChannelConfigRegistry::new();
2076 reg.insert_prefix(
2077 "svc.",
2078 ChannelConfig::new(ChannelId::parse("svc.general").unwrap()).with_priority(1),
2079 );
2080 reg.insert_prefix(
2081 "svc.replies.",
2082 ChannelConfig::new(ChannelId::parse("svc.replies.prefix").unwrap()).with_priority(2),
2083 );
2084 reg.insert(ChannelConfig::new(ChannelId::parse("svc.exact").unwrap()).with_priority(3));
2085
2086 for name in ["svc.replies.aa", "svc.other", "svc.exact", "nomatch"] {
2087 let via_get = reg.get_by_name(name).map(|c| c.priority);
2088 let via_resolve = reg.resolve_by_name(name).map(|r| r.config.priority);
2089 assert_eq!(
2090 via_get, via_resolve,
2091 "get_by_name and resolve_by_name disagreed for {name:?}"
2092 );
2093 }
2094
2095 // And the longest prefix wins, not merely any match.
2096 assert_eq!(
2097 reg.resolve_by_name("svc.replies.aa")
2098 .unwrap()
2099 .matched_prefix
2100 .as_deref(),
2101 Some("svc.replies.")
2102 );
2103 }
2104
2105 // ---- M2 (2026-07-31 audit): queue-group membership authority ----
2106
2107 /// Default is unchanged: any subscriber may join any group.
2108 #[test]
2109 fn queue_group_unrestricted_by_default() {
2110 let peer = EntityKeypair::generate();
2111 let rev = RevocationRegistry::new();
2112 let config = ChannelConfig::new(ChannelId::parse("work/queue").unwrap());
2113 assert_eq!(config.queue_group_policy, QueueGroupPolicy::Unrestricted);
2114 assert!(config.can_join_queue_group(
2115 peer.entity_id(),
2116 "work/queue",
2117 "workers",
2118 None,
2119 &rev,
2120 0
2121 ));
2122 }
2123
2124 /// `Deny` refuses every group, chain or not.
2125 #[test]
2126 fn queue_group_deny_refuses_even_with_a_valid_chain() {
2127 let owner = EntityKeypair::generate();
2128 let peer = EntityKeypair::generate();
2129 let rev = RevocationRegistry::new();
2130 let config = ChannelConfig::new(ChannelId::parse("work/queue").unwrap())
2131 .with_token_roots(vec![owner.entity_id().clone()])
2132 .with_queue_group_policy(QueueGroupPolicy::Deny);
2133
2134 let chain = direct_chain(
2135 &owner,
2136 &peer,
2137 TokenScope::SUBSCRIBE,
2138 queue_group_hash("work/queue", "workers"),
2139 );
2140 assert!(!config.can_join_queue_group(
2141 peer.entity_id(),
2142 "work/queue",
2143 "workers",
2144 Some(&chain),
2145 &rev,
2146 0
2147 ));
2148 }
2149
2150 /// The core M2 property: a grant for one group must not admit the
2151 /// holder to a DIFFERENT group. An allowlist of group names could
2152 /// not express this — names are operational constants, not secrets.
2153 #[test]
2154 fn queue_group_grant_binds_to_one_specific_group() {
2155 let owner = EntityKeypair::generate();
2156 let worker = EntityKeypair::generate();
2157 let rev = RevocationRegistry::new();
2158 let channel = "work/queue";
2159 let config = ChannelConfig::new(ChannelId::parse(channel).unwrap())
2160 .with_token_roots(vec![owner.entity_id().clone()])
2161 .with_queue_group_policy(QueueGroupPolicy::TokenBound);
2162
2163 let chain = direct_chain(
2164 &owner,
2165 &worker,
2166 TokenScope::SUBSCRIBE,
2167 queue_group_hash(channel, "batch"),
2168 );
2169
2170 assert!(
2171 config.can_join_queue_group(
2172 worker.entity_id(),
2173 channel,
2174 "batch",
2175 Some(&chain),
2176 &rev,
2177 0
2178 ),
2179 "the granted group must be joinable"
2180 );
2181 assert!(
2182 !config.can_join_queue_group(
2183 worker.entity_id(),
2184 channel,
2185 "realtime",
2186 Some(&chain),
2187 &rev,
2188 0
2189 ),
2190 "a grant for one group must not admit the holder to another — \
2191 that is the work-stealing this policy exists to stop"
2192 );
2193 }
2194
2195 /// A plain channel-scoped SUBSCRIBE token is NOT a worker grant.
2196 /// Otherwise every legitimate subscriber would silently keep the
2197 /// ability to join any group and the policy would be a no-op.
2198 #[test]
2199 fn channel_subscribe_token_is_not_a_queue_group_grant() {
2200 let owner = EntityKeypair::generate();
2201 let reader = EntityKeypair::generate();
2202 let rev = RevocationRegistry::new();
2203 let channel = "work/queue";
2204 let id = ChannelId::parse(channel).unwrap();
2205 let config = ChannelConfig::new(id.clone())
2206 .with_token_roots(vec![owner.entity_id().clone()])
2207 .with_queue_group_policy(QueueGroupPolicy::TokenBound);
2208
2209 // Scoped to the CHANNEL, which is what an ordinary
2210 // read-only subscriber (e.g. an auditor) would hold.
2211 let chain = direct_chain(&owner, &reader, TokenScope::SUBSCRIBE, id.hash());
2212
2213 assert!(
2214 config.can_subscribe(
2215 &make_caps(false),
2216 reader.entity_id(),
2217 id.hash(),
2218 Some(&chain),
2219 &rev,
2220 0
2221 ),
2222 "precondition: it is a valid subscribe credential"
2223 );
2224 assert!(
2225 !config.can_join_queue_group(
2226 reader.entity_id(),
2227 channel,
2228 "workers",
2229 Some(&chain),
2230 &rev,
2231 0
2232 ),
2233 "a read-only subscriber must not be able to steal worker traffic"
2234 );
2235 }
2236
2237 /// TokenBound fails closed with no chain and with no roots.
2238 #[test]
2239 fn queue_group_token_bound_fails_closed() {
2240 let owner = EntityKeypair::generate();
2241 let peer = EntityKeypair::generate();
2242 let rev = RevocationRegistry::new();
2243 let channel = "work/queue";
2244
2245 let rooted = ChannelConfig::new(ChannelId::parse(channel).unwrap())
2246 .with_token_roots(vec![owner.entity_id().clone()])
2247 .with_queue_group_policy(QueueGroupPolicy::TokenBound);
2248 assert!(
2249 !rooted.can_join_queue_group(peer.entity_id(), channel, "w", None, &rev, 0),
2250 "no chain presented → refuse"
2251 );
2252
2253 let rootless = ChannelConfig::new(ChannelId::parse(channel).unwrap())
2254 .with_queue_group_policy(QueueGroupPolicy::TokenBound);
2255 let chain = direct_chain(
2256 &owner,
2257 &peer,
2258 TokenScope::SUBSCRIBE,
2259 queue_group_hash(channel, "w"),
2260 );
2261 assert!(
2262 !rootless.can_join_queue_group(peer.entity_id(), channel, "w", Some(&chain), &rev, 0),
2263 "no roots to anchor against → refuse"
2264 );
2265 }
2266
2267 /// The `#` separator keeps group grants and channel grants in
2268 /// disjoint hash spaces: `#` is outside the channel-name charset,
2269 /// so no legitimate channel name can ever hash to a group grant.
2270 #[test]
2271 fn queue_group_hash_cannot_collide_with_a_channel_name() {
2272 let h = queue_group_hash("work/queue", "workers");
2273 // The only string that would produce it is not a legal name.
2274 assert!(ChannelName::new("work/queue#workers").is_err());
2275 assert_ne!(h, channel_hash("work/queue"));
2276 assert_ne!(h, channel_hash("work/queueworkers"));
2277 // Distinct groups on one channel are distinct grants.
2278 assert_ne!(h, queue_group_hash("work/queue", "other"));
2279 // Same group name on distinct channels are distinct grants.
2280 assert_ne!(h, queue_group_hash("work/other", "workers"));
2281 }
2282
2283 /// …and they stay disjoint only while they stay in ONE hash space.
2284 ///
2285 /// The disjointness argument above is entirely about `#` being
2286 /// outside the channel-name charset — which proves nothing unless
2287 /// both sides are hashed the same way. `queue_group_hash` delegates
2288 /// to `channel_hash` for that reason, and this pins the delegation:
2289 /// a seed, a domain-separation prefix, or an algorithm change
2290 /// applied to one and not the other would leave the test above
2291 /// passing (the values would still differ) while the documented
2292 /// derivation quietly became false, and grants minted before the
2293 /// change would stop matching the ones checked after it.
2294 #[test]
2295 fn queue_group_hash_is_the_channel_hash_of_the_joined_name() {
2296 for (channel, group) in [
2297 ("work/queue", "workers"),
2298 ("a", "b"),
2299 ("svc.replies.deadbeefdeadbeef", "shard-3"),
2300 ] {
2301 assert_eq!(
2302 queue_group_hash(channel, group),
2303 channel_hash(&format!("{channel}#{group}")),
2304 "queue_group_hash({channel:?}, {group:?}) no longer equals the \
2305 canonical hash of \"{channel}#{group}\" — the two have drifted \
2306 into separate hash spaces"
2307 );
2308 }
2309 }
2310
2311 // ---- M1 (2026-07-31 audit): gates key on the REQUESTED channel ----
2312
2313 /// A token minted for one channel under a prefix must not authorize
2314 /// a sibling under the same prefix.
2315 ///
2316 /// Pre-fix the gate verified against `self.channel_id.hash()`, and
2317 /// for a prefix-registered config that is a sentinel standing for
2318 /// the whole family — so one token minted for the sentinel
2319 /// authorized every channel beneath it, silently degrading a
2320 /// per-channel binding to a per-prefix one.
2321 #[test]
2322 fn prefix_config_gate_binds_to_the_requested_channel_not_the_sentinel() {
2323 let owner = EntityKeypair::generate();
2324 let subject = EntityKeypair::generate();
2325 let caps = make_caps(false);
2326 let rev = RevocationRegistry::new();
2327
2328 let sentinel = ChannelId::parse("svc.replies.prefix").unwrap();
2329 let config =
2330 ChannelConfig::new(sentinel.clone()).with_token_roots(vec![owner.entity_id().clone()]);
2331
2332 let mine = channel_hash("svc.replies.aaaa");
2333 let theirs = channel_hash("svc.replies.bbbb");
2334
2335 // A token for MY channel authorizes my channel...
2336 let chain = direct_chain(&owner, &subject, TokenScope::SUBSCRIBE, mine);
2337 assert!(config.can_subscribe(&caps, subject.entity_id(), mine, Some(&chain), &rev, 0));
2338 // ...and not a sibling under the same prefix.
2339 assert!(
2340 !config.can_subscribe(&caps, subject.entity_id(), theirs, Some(&chain), &rev, 0),
2341 "a token for one channel must not authorize a sibling under the \
2342 same prefix"
2343 );
2344
2345 // A token minted for the SENTINEL authorizes nothing real —
2346 // that was the per-prefix skeleton key.
2347 let sentinel_chain = direct_chain(&owner, &subject, TokenScope::SUBSCRIBE, sentinel.hash());
2348 assert!(
2349 !config.can_subscribe(
2350 &caps,
2351 subject.entity_id(),
2352 mine,
2353 Some(&sentinel_chain),
2354 &rev,
2355 0
2356 ),
2357 "a sentinel-scoped token must not authorize a real channel"
2358 );
2359 }
2360
2361 /// The publish counterpart, and the reason `set_publish_chain` was
2362 /// unreachable for token-gated prefix channels: it stores under the
2363 /// real channel hash while the gate asked about the sentinel.
2364 #[test]
2365 fn prefix_config_publish_gate_binds_to_the_requested_channel() {
2366 let owner = EntityKeypair::generate();
2367 let subject = EntityKeypair::generate();
2368 let caps = make_caps(false);
2369 let rev = RevocationRegistry::new();
2370
2371 let sentinel = ChannelId::parse("svc.requests.prefix").unwrap();
2372 let config = ChannelConfig::new(sentinel).with_token_roots(vec![owner.entity_id().clone()]);
2373
2374 let real = channel_hash("svc.requests.aaaa");
2375 let chain = direct_chain(&owner, &subject, TokenScope::PUBLISH, real);
2376
2377 assert!(config.can_publish(&caps, subject.entity_id(), real, Some(&chain), &rev, 0));
2378 assert!(
2379 !config.can_publish(
2380 &caps,
2381 subject.entity_id(),
2382 channel_hash("svc.requests.bbbb"),
2383 Some(&chain),
2384 &rev,
2385 0
2386 ),
2387 "a publish token for one channel must not authorize a sibling"
2388 );
2389 }
2390
2391 /// `reverify_subscribe*` must ask about the same channel the
2392 /// subscribe gate did, or the publish-time re-check and the sweep
2393 /// disagree with the decision that admitted the peer.
2394 #[test]
2395 fn reverify_paths_bind_to_the_requested_channel() {
2396 let owner = EntityKeypair::generate();
2397 let subject = EntityKeypair::generate();
2398 let rev = RevocationRegistry::new();
2399
2400 let sentinel = ChannelId::parse("svc.replies.prefix").unwrap();
2401 let config = ChannelConfig::new(sentinel).with_token_roots(vec![owner.entity_id().clone()]);
2402
2403 let mine = channel_hash("svc.replies.aaaa");
2404 let chain = direct_chain(&owner, &subject, TokenScope::SUBSCRIBE, mine);
2405
2406 for reverify in [
2407 ChannelConfig::reverify_subscribe as fn(&_, &_, &_, u64, &_, u64) -> bool,
2408 ChannelConfig::reverify_subscribe_presigned,
2409 ] {
2410 assert!(reverify(
2411 &config,
2412 &chain,
2413 subject.entity_id(),
2414 mine,
2415 &rev,
2416 0
2417 ));
2418 assert!(
2419 !reverify(
2420 &config,
2421 &chain,
2422 subject.entity_id(),
2423 channel_hash("svc.replies.bbbb"),
2424 &rev,
2425 0
2426 ),
2427 "re-verify must reject a chain that does not authorize the \
2428 channel being published to"
2429 );
2430 }
2431 }
2432
2433 // ---- H3 (2026-07-31 audit): origin-bound channel families ----
2434
2435 const OB: OriginBinding = OriginBinding::OriginHashHex16;
2436 const OB_PREFIX: &str = "svc.replies.";
2437
2438 /// The rule the whole finding turns on: a peer whose identity is
2439 /// not pinned is REJECTED. Admitting it would let an attacker
2440 /// bypass the binding by simply never announcing.
2441 #[test]
2442 fn origin_binding_rejects_unpinned_peer() {
2443 let name = format!("{OB_PREFIX}{:016x}", 0xABCD_1234_5678_9ABCu64);
2444 assert!(
2445 !OB.authorizes(&name, Some(OB_PREFIX), None),
2446 "an unpinned peer must never be authorized, even for a \
2447 well-formed name"
2448 );
2449 }
2450
2451 #[test]
2452 fn origin_binding_admits_matching_origin() {
2453 let origin = 0xABCD_1234_5678_9ABCu64;
2454 let name = format!("{OB_PREFIX}{origin:016x}");
2455 assert!(OB.authorizes(&name, Some(OB_PREFIX), Some(origin)));
2456 }
2457
2458 /// The attack: a pinned peer asking for a name that encodes some
2459 /// OTHER peer's origin.
2460 #[test]
2461 fn origin_binding_rejects_other_peers_origin() {
2462 let victim = 0xABCD_1234_5678_9ABCu64;
2463 let attacker = 0x0011_2233_4455_6677u64;
2464 let name = format!("{OB_PREFIX}{victim:016x}");
2465 assert!(
2466 !OB.authorizes(&name, Some(OB_PREFIX), Some(attacker)),
2467 "a peer must not claim a channel naming another peer's origin"
2468 );
2469 }
2470
2471 /// A binding on an exact-match config has no dynamic suffix to
2472 /// check, so it fails closed rather than admitting.
2473 #[test]
2474 fn origin_binding_without_a_matched_prefix_fails_closed() {
2475 let origin = 0xABCD_1234_5678_9ABCu64;
2476 let name = format!("{OB_PREFIX}{origin:016x}");
2477 assert!(!OB.authorizes(&name, None, Some(origin)));
2478 }
2479
2480 /// The same fail-closed rule reached the way an operator actually
2481 /// reaches it: registering an origin-bound config by EXACT name.
2482 ///
2483 /// `resolve_by_name` reports no matched prefix for an exact hit, so
2484 /// the binding has nothing to split and denies every subscriber —
2485 /// including the one peer whose origin the name encodes. Correct,
2486 /// and completely invisible: a channel that accepts nobody looks
2487 /// exactly like a channel nobody is using, and the refused peers see
2488 /// a generic `Unauthorized`. `warn_if_fail_closed` logs this at
2489 /// registration for that reason; this pins the behaviour the warning
2490 /// is about.
2491 #[test]
2492 fn origin_bound_config_registered_by_exact_name_denies_everyone() {
2493 let reg = ChannelConfigRegistry::new();
2494 let origin = 0xABCD_1234_5678_9ABCu64;
2495 let name = format!("{OB_PREFIX}{origin:016x}");
2496
2497 // The misregistration: `insert`, not `insert_prefix`.
2498 reg.insert(
2499 ChannelConfig::new(ChannelId::parse(&name).unwrap()).with_subscriber_origin_binding(OB),
2500 );
2501
2502 let resolved = reg.resolve_by_name(&name).expect("exact entry resolves");
2503 assert_eq!(
2504 resolved.matched_prefix, None,
2505 "an exact hit reports no matched prefix — this is the input that \
2506 makes the binding fail closed"
2507 );
2508 assert!(
2509 !OB.authorizes(&name, resolved.matched_prefix.as_deref(), Some(origin)),
2510 "an exact-registered origin binding denies even the peer the name \
2511 encodes; registering it as a prefix is the only working shape"
2512 );
2513
2514 // Registered as a prefix instead, the same peer is admitted —
2515 // so the denial above is about the registration, not the name.
2516 let reg = ChannelConfigRegistry::new();
2517 reg.insert_prefix(
2518 OB_PREFIX,
2519 ChannelConfig::new(ChannelId::parse("svc.replies.prefix").unwrap())
2520 .with_subscriber_origin_binding(OB),
2521 );
2522 let resolved = reg.resolve_by_name(&name).expect("prefix entry resolves");
2523 assert_eq!(resolved.matched_prefix.as_deref(), Some(OB_PREFIX));
2524 assert!(OB.authorizes(&name, resolved.matched_prefix.as_deref(), Some(origin)));
2525 }
2526
2527 /// Formatting is exact: no truncation, no case folding, no
2528 /// suffix-prefix matching.
2529 #[test]
2530 fn origin_binding_requires_exact_16_hex_suffix() {
2531 let origin = 0x0000_0000_0000_00ABu64;
2532 for bad in [
2533 "ab", // unpadded
2534 "AB", // uppercase (also unpadded)
2535 "00000000000000ab0", // trailing garbage
2536 "00000000000000a", // short
2537 "00000000000000AB", // uppercase, padded
2538 ] {
2539 let name = format!("{OB_PREFIX}{bad}");
2540 assert!(
2541 !OB.authorizes(&name, Some(OB_PREFIX), Some(origin)),
2542 "suffix {bad:?} must not authorize origin {origin:#x}"
2543 );
2544 }
2545 // The canonical rendering does authorize.
2546 let good = format!("{OB_PREFIX}{origin:016x}");
2547 assert!(OB.authorizes(&good, Some(OB_PREFIX), Some(origin)));
2548 }
2549
2550 /// A config carrying no binding is unaffected — the gate is opt-in.
2551 #[test]
2552 fn config_without_binding_is_unconstrained() {
2553 let cfg = ChannelConfig::new(ChannelId::parse("svc.replies.prefix").unwrap());
2554 assert!(cfg.subscriber_origin_binding.is_none());
2555 }
2556
2557 /// `resolve_by_name` must report the prefix it matched, or the
2558 /// binding has no way to locate the dynamic suffix.
2559 #[test]
2560 fn resolve_by_name_reports_the_matched_prefix() {
2561 let reg = ChannelConfigRegistry::new();
2562 let sentinel = ChannelId::parse("svc.replies.prefix").unwrap();
2563 reg.insert_prefix(
2564 OB_PREFIX,
2565 ChannelConfig::new(sentinel).with_subscriber_origin_binding(OB),
2566 );
2567
2568 let resolved = reg
2569 .resolve_by_name("svc.replies.00112233445566aa")
2570 .expect("prefix must resolve");
2571 assert_eq!(resolved.matched_prefix.as_deref(), Some(OB_PREFIX));
2572 assert_eq!(resolved.config.subscriber_origin_binding, Some(OB));
2573
2574 // An exact-match resolution reports no prefix.
2575 let exact = ChannelId::parse("plain.channel").unwrap();
2576 reg.insert(ChannelConfig::new(exact));
2577 let resolved = reg.resolve_by_name("plain.channel").expect("exact");
2578 assert!(resolved.matched_prefix.is_none());
2579 }
2580
2581 // ---- R9: the one shared RPC service-channel registration ----
2582
2583 /// The H2 + H3 content of `install_rpc_service_defaults`, asserted
2584 /// behaviourally rather than by scanning for method names.
2585 ///
2586 /// This is the policy every serve path now shares — `serve_rpc*`,
2587 /// the aggregator, and the org facade. It has drifted twice, each
2588 /// time because a serve path carried its own copy and a later fix
2589 /// landed on only one of them, so it is worth pinning what the
2590 /// policy DOES and not just where it lives.
2591 #[test]
2592 fn rpc_service_defaults_are_install_if_absent_and_origin_bound() {
2593 let reg = ChannelConfigRegistry::new();
2594 let root = EntityKeypair::generate();
2595
2596 // H2: an operator's strict ACL, registered before serving.
2597 reg.insert(
2598 ChannelConfig::new(ChannelId::parse("svc.requests").unwrap())
2599 .with_token_roots(vec![root.entity_id().clone()]),
2600 );
2601 reg.insert_prefix(
2602 "svc.replies.",
2603 ChannelConfig::new(ChannelId::parse("svc.replies.prefix").unwrap())
2604 .with_token_roots(vec![root.entity_id().clone()]),
2605 );
2606
2607 reg.install_rpc_service_defaults("svc");
2608
2609 assert!(
2610 reg.get_by_name("svc.requests").unwrap().token_required(),
2611 "H2: serving must not replace an ACL the operator registered \
2612 first — a replacing insert destroys it silently, leaving a \
2613 posture identical to the default"
2614 );
2615 assert!(
2616 reg.get_by_name("svc.replies.abcdef0123456789")
2617 .unwrap()
2618 .token_required(),
2619 "H2 applies to the reply PREFIX too"
2620 );
2621
2622 // …and on a clean registry it installs both, with the binding.
2623 let fresh = ChannelConfigRegistry::new();
2624 fresh.install_rpc_service_defaults("svc");
2625
2626 assert!(
2627 fresh.get_by_name("svc.requests").is_some(),
2628 "the request channel must be installed when unclaimed"
2629 );
2630 let replies = fresh
2631 .get_by_name("svc.replies.abcdef0123456789")
2632 .expect("the reply prefix must admit a per-caller channel");
2633 assert_eq!(
2634 replies.subscriber_origin_binding,
2635 Some(OriginBinding::OriginHashHex16),
2636 "H3: the reply prefix must be origin-bound. Unbound, any mesh peer \
2637 can hold a live subscription to another caller's reply channel and \
2638 receive that caller's response bodies whenever the server's direct \
2639 route misses and the response falls back to roster fan-out."
2640 );
2641 }
2642
2643 /// A service name that cannot form a valid channel name installs
2644 /// NOTHING — not a half-configured pair where the requests channel
2645 /// exists and the reply prefix does not.
2646 ///
2647 /// The LENGTH cases are the ones that matter and the ones an
2648 /// invalid-character name does not reach. There are TWO of them,
2649 /// because the three names involved are three different lengths —
2650 /// `.requests` is 9 bytes, the `.replies.prefix` sentinel is 15, and
2651 /// a real `.replies.<16 hex>` is 25:
2652 ///
2653 /// - request fits, sentinel does not;
2654 /// - both fit, but no real per-caller reply channel does.
2655 ///
2656 /// The second is the one a sentinel-based check misses, and it fails
2657 /// worse than a visible refusal: everything looks installed, and
2658 /// then every call hangs until it times out because no caller can
2659 /// name a reply channel that validates.
2660 ///
2661 /// A character-invalid name fails all three and would pass this test
2662 /// against an implementation that got either band wrong, which is
2663 /// why the bands are enumerated explicitly with preconditions.
2664 #[test]
2665 fn rpc_service_defaults_install_nothing_for_an_unrepresentable_name() {
2666 use super::super::name::MAX_NAME_LEN;
2667
2668 // Band 1: request fits, sentinel does not.
2669 let no_sentinel = "s".repeat(MAX_NAME_LEN - ".requests".len());
2670 assert!(ChannelName::new(&format!("{no_sentinel}.requests")).is_ok());
2671 assert!(
2672 ChannelName::new(&format!("{no_sentinel}.replies.prefix")).is_err(),
2673 "precondition: band 1 must have an unrepresentable sentinel"
2674 );
2675
2676 // Band 2: request AND sentinel fit; a real reply channel does not.
2677 let no_real_reply = "s".repeat(MAX_NAME_LEN - ".replies.prefix".len());
2678 assert!(ChannelName::new(&format!("{no_real_reply}.requests")).is_ok());
2679 assert!(
2680 ChannelName::new(&format!("{no_real_reply}.replies.prefix")).is_ok(),
2681 "precondition: band 2's sentinel must VALIDATE — that is what \
2682 makes checking the sentinel insufficient"
2683 );
2684 assert!(
2685 ChannelName::new(&format!("{no_real_reply}.replies.{:016x}", 0u64)).is_err(),
2686 "precondition: …while no real per-caller reply channel fits"
2687 );
2688
2689 for (band, service) in [
2690 ("no sentinel", no_sentinel.as_str()),
2691 ("no real reply channel", no_real_reply.as_str()),
2692 ("invalid characters", "bad name#with/invalid chars"),
2693 ] {
2694 let reg = ChannelConfigRegistry::new();
2695 reg.install_rpc_service_defaults(service);
2696 assert_eq!(
2697 reg.len(),
2698 0,
2699 "[{band}] installed a request channel the reply side cannot \
2700 match (service len {})",
2701 service.len()
2702 );
2703 assert_eq!(
2704 reg.snapshot_prefixes().len(),
2705 0,
2706 "[{band}] installed a reply prefix no caller can ever use \
2707 (service len {})",
2708 service.len()
2709 );
2710 }
2711 }
2712
2713 /// The largest service name that IS fully usable must still install.
2714 ///
2715 /// Paired with the refusal test above so the boundary is pinned from
2716 /// both sides — a validator that is too strict silently stops
2717 /// configuring services that work, which no test asserting "installs
2718 /// nothing" would ever catch.
2719 #[test]
2720 fn rpc_service_defaults_install_at_the_longest_usable_service_name() {
2721 use super::super::name::MAX_NAME_LEN;
2722
2723 let longest = "s".repeat(MAX_NAME_LEN - ".replies.0123456789abcdef".len());
2724 let reg = ChannelConfigRegistry::new();
2725 reg.install_rpc_service_defaults(&longest);
2726
2727 assert!(
2728 reg.get_by_name(&format!("{longest}.requests")).is_some(),
2729 "the longest fully-usable service name must still get its request \
2730 channel"
2731 );
2732 let reply = format!("{longest}.replies.{:016x}", u64::MAX);
2733 assert_eq!(
2734 ChannelName::new(&reply).map(|_| ()),
2735 Ok(()),
2736 "precondition: this is the longest name where a real reply \
2737 channel still fits"
2738 );
2739 assert_eq!(
2740 reg.get_by_name(&reply)
2741 .expect("the reply prefix must resolve it")
2742 .subscriber_origin_binding,
2743 Some(OriginBinding::OriginHashHex16)
2744 );
2745 }
2746
2747 // ---- H2 (2026-07-31 audit): install-if-absent must not clobber ----
2748
2749 /// `insert_if_absent` installs into an empty slot and reports it.
2750 #[test]
2751 fn insert_if_absent_installs_when_vacant() {
2752 let reg = ChannelConfigRegistry::new();
2753 let id = ChannelId::parse("svc.requests").unwrap();
2754 assert!(reg.insert_if_absent(ChannelConfig::new(id.clone()).with_priority(4)));
2755 assert_eq!(reg.get_by_name("svc.requests").unwrap().priority, 4);
2756 assert_eq!(reg.get(id.hash()).unwrap().priority, 4);
2757 }
2758
2759 /// The H2 regression at the registry level: an operator's strict
2760 /// config must survive a later auto-registered permissive default.
2761 /// Pre-fix `insert` replaced unconditionally and the ACL vanished.
2762 #[test]
2763 fn insert_if_absent_preserves_existing_strict_config() {
2764 let reg = ChannelConfigRegistry::new();
2765 let id = ChannelId::parse("svc.requests").unwrap();
2766 let root = EntityKeypair::generate();
2767 reg.insert(ChannelConfig::new(id.clone()).with_token_roots(vec![root.entity_id().clone()]));
2768
2769 // Auto-registration's permissive default arrives second.
2770 let installed = reg.insert_if_absent(ChannelConfig::new(id.clone()));
2771
2772 assert!(!installed, "must report that it did not install");
2773 let cfg = reg.get_by_name("svc.requests").unwrap();
2774 assert!(
2775 cfg.token_required(),
2776 "operator's token gate must survive auto-registration"
2777 );
2778 assert_eq!(cfg.token_roots.len(), 1);
2779 }
2780
2781 /// Same guarantee on the prefix table, which is where nRPC's
2782 /// reply-channel ACL would live.
2783 #[test]
2784 fn insert_prefix_if_absent_preserves_existing_strict_prefix() {
2785 let reg = ChannelConfigRegistry::new();
2786 let sentinel = ChannelId::parse("svc.replies.prefix").unwrap();
2787 let root = EntityKeypair::generate();
2788 reg.insert_prefix(
2789 "svc.replies.",
2790 ChannelConfig::new(sentinel.clone()).with_token_roots(vec![root.entity_id().clone()]),
2791 );
2792
2793 let installed = reg.insert_prefix_if_absent("svc.replies.", ChannelConfig::new(sentinel));
2794
2795 assert!(!installed);
2796 let cfg = reg.get_by_name("svc.replies.abcdef0123456789").unwrap();
2797 assert!(
2798 cfg.token_required(),
2799 "operator's reply-prefix gate must survive auto-registration"
2800 );
2801 }
2802
2803 #[test]
2804 fn insert_prefix_if_absent_installs_when_vacant() {
2805 let reg = ChannelConfigRegistry::new();
2806 let sentinel = ChannelId::parse("svc.replies.prefix").unwrap();
2807 assert!(reg.insert_prefix_if_absent(
2808 "svc.replies.",
2809 ChannelConfig::new(sentinel).with_priority(2)
2810 ));
2811 assert_eq!(reg.get_by_name("svc.replies.deadbeef").unwrap().priority, 2);
2812 }
2813
2814 /// Re-registering the same channel must not corrupt the canonical-
2815 /// hash index. Pre-fix `insert` pushed the name unconditionally, so
2816 /// the second registration grew the bucket to `[name, name]`,
2817 /// `get()` read that as a hash collision, and canonical-hash lookup
2818 /// started returning `None` for a channel that plainly exists.
2819 #[test]
2820 fn repeated_insert_does_not_self_collide_the_hash_index() {
2821 let reg = ChannelConfigRegistry::new();
2822 let id = ChannelId::parse("svc.requests").unwrap();
2823 let hash = id.hash();
2824
2825 reg.insert(ChannelConfig::new(id.clone()).with_priority(1));
2826 reg.insert(ChannelConfig::new(id.clone()).with_priority(2));
2827 reg.insert(ChannelConfig::new(id).with_priority(3));
2828
2829 assert_eq!(reg.len(), 1, "one channel, not three");
2830 let cfg = reg
2831 .get(hash)
2832 .expect("canonical-hash lookup must survive re-registration");
2833 assert_eq!(cfg.priority, 3, "latest config wins");
2834 }
2835
2836 /// The mixed path auto-registration actually takes: a replacing
2837 /// `insert` followed by install-if-absent attempts.
2838 #[test]
2839 fn insert_then_if_absent_leaves_hash_index_unambiguous() {
2840 let reg = ChannelConfigRegistry::new();
2841 let id = ChannelId::parse("svc.requests").unwrap();
2842 let hash = id.hash();
2843
2844 reg.insert(ChannelConfig::new(id.clone()).with_priority(9));
2845 assert!(!reg.insert_if_absent(ChannelConfig::new(id.clone())));
2846 assert!(!reg.insert_if_absent(ChannelConfig::new(id)));
2847
2848 assert_eq!(
2849 reg.get(hash).expect("must stay resolvable").priority,
2850 9,
2851 "the operator's config must remain, and the index unambiguous"
2852 );
2853 }
2854
2855 /// Exactly one of N concurrent `insert_if_absent` callers may win,
2856 /// and the reverse index must not end up ambiguous afterwards.
2857 #[test]
2858 fn concurrent_insert_if_absent_elects_exactly_one_winner() {
2859 use std::sync::atomic::{AtomicUsize, Ordering};
2860 use std::sync::Arc as StdArc;
2861
2862 let reg = StdArc::new(ChannelConfigRegistry::new());
2863 let wins = StdArc::new(AtomicUsize::new(0));
2864 let id = ChannelId::parse("svc.requests").unwrap();
2865
2866 std::thread::scope(|s| {
2867 for i in 0..8 {
2868 let reg = reg.clone();
2869 let wins = wins.clone();
2870 let id = id.clone();
2871 s.spawn(move || {
2872 if reg.insert_if_absent(ChannelConfig::new(id).with_priority(i)) {
2873 wins.fetch_add(1, Ordering::Relaxed);
2874 }
2875 });
2876 }
2877 });
2878
2879 assert_eq!(wins.load(Ordering::Relaxed), 1, "exactly one installer");
2880 assert_eq!(reg.len(), 1);
2881 assert!(
2882 reg.get(id.hash()).is_some(),
2883 "hash index must stay unambiguous under concurrent installs"
2884 );
2885 }
2886
2887 /// The reverse indices must agree with `configs` after arbitrary
2888 /// concurrent registration and removal — no name indexed that
2889 /// `configs` does not hold, and none held that is not indexed.
2890 ///
2891 /// Both directions matter and they used to fail in turn:
2892 ///
2893 /// - Un-indexed-but-present: a re-registration landing between
2894 /// `remove`'s `configs.remove` and its `retain` had its fresh
2895 /// index entry deleted. The channel is registered and `get_by_name`
2896 /// finds it, but `get(hash)` — the path publish and subscribe
2897 /// authorization take — answers `None`, so its ACL stops being
2898 /// enforced.
2899 /// - Indexed-but-absent: the repair pass for the above re-added a
2900 /// name a second concurrent removal had just taken out. `get` and
2901 /// `remove` read a bucket of more than one name as a hash
2902 /// collision and refuse it, so a phantom name disables lookup for
2903 /// whatever real channel shares the bucket.
2904 ///
2905 /// Interleaving-dependent, so a green run is evidence rather than
2906 /// proof. It fails reliably against the unsynchronized version
2907 /// (typically within a few hundred iterations), which is what makes
2908 /// it worth keeping: the assertion states the invariant exactly, and
2909 /// a future change that drops the serialization has a real chance of
2910 /// being caught here.
2911 #[test]
2912 fn concurrent_registration_and_removal_keep_the_indices_consistent() {
2913 use std::sync::Arc as StdArc;
2914
2915 // Distinct names, so threads contend on the registry rather
2916 // than on one key — the churn that produced both defects.
2917 let names: Vec<String> = (0..4).map(|i| format!("svc.chan{i}")).collect();
2918
2919 for _round in 0..200 {
2920 let reg = StdArc::new(ChannelConfigRegistry::new());
2921 std::thread::scope(|s| {
2922 for name in &names {
2923 for _ in 0..2 {
2924 let reg = reg.clone();
2925 let id = ChannelId::parse(name).unwrap();
2926 s.spawn(move || {
2927 reg.insert(ChannelConfig::new(id.clone()));
2928 reg.remove_by_name(id.name().as_str());
2929 reg.insert(ChannelConfig::new(id));
2930 });
2931 }
2932 }
2933 for name in &names {
2934 let reg = reg.clone();
2935 let name = name.clone();
2936 s.spawn(move || {
2937 reg.remove_by_name(&name);
2938 });
2939 }
2940 });
2941
2942 for name in &names {
2943 let present = reg.get_by_name(name).is_some();
2944 // Straight at `by_hash`: the invariant is about the
2945 // index itself, and the collision-safe public accessors
2946 // hide exactly the corruption being asserted on.
2947 let hash = ChannelId::parse(name).unwrap().hash();
2948 let indexed = reg
2949 .by_hash
2950 .get(&hash)
2951 .is_some_and(|names| names.iter().any(|n| n == name));
2952 assert_eq!(
2953 present, indexed,
2954 "index and `configs` disagree about {name:?}: present={present}, \
2955 indexed={indexed}. Registered-but-unindexed silently stops \
2956 enforcing that channel's ACL on the `get(hash)` path; \
2957 indexed-but-absent poisons the bucket for every channel \
2958 sharing it."
2959 );
2960 }
2961 }
2962 }
2963}