Skip to main content

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}