[PATCH proxmox-ve-rs v3 12/45] ve-config: sdn: microseg: add the desired state

[Hannes Laimer <h.laimer@proxmox.com>]

The data plane needs the policy as the id each assigned guest NIC
stamps and, for every pair of ids, the verdict and the narrowed rules
to check. That form comes from the running config and the guest
inventory alone and does not depend on how a host enforces it. So it
is built here, next to the model.

The pairs cover every allocated id, whether a NIC uses it or not. So
a NIC that comes up later needs only its own id, which resolves
without folding the pairs.

Denies alone need no pair, a missing pair denies. The exception is a
narrowed deny of ICMPv6 where the reverse pair allows or the source is
untagged, since the data plane passes neighbor discovery there.

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 .../src/sdn/microseg/identity.rs              |  15 +-
 proxmox-ve-config/src/sdn/microseg/mod.rs     |  44 +-
 proxmox-ve-config/src/sdn/microseg/state.rs   | 612 ++++++++++++++++++
 3 files changed, 657 insertions(+), 14 deletions(-)
 create mode 100644 proxmox-ve-config/src/sdn/microseg/state.rs

diff --git a/proxmox-ve-config/src/sdn/microseg/identity.rs b/proxmox-ve-config/src/sdn/microseg/identity.rs
index eb8317076b44..ef1ea10f4260 100644
--- a/proxmox-ve-config/src/sdn/microseg/identity.rs
+++ b/proxmox-ve-config/src/sdn/microseg/identity.rs
@@ -179,6 +179,13 @@ impl Registry {
             total: u16::MAX as usize,
         }
     }
+
+    pub(crate) fn ids_by_signature(&self, policies: &[Policy]) -> HashMap<Signature, u16> {
+        self.classes
+            .iter()
+            .map(|(&id, rep)| (signature(rep, policies), id))
+            .collect()
+    }
 }
 
 /// Recompute the registry from the sets the assignments name and the policies, fresh-first against
@@ -295,11 +302,9 @@ mod tests {
     }
 
     fn id_of(reg: &Registry, set: &GroupSet, policies: &[Policy]) -> Option<u16> {
-        let sig = signature(set, policies);
-        reg.classes
-            .iter()
-            .find(|(_, rep)| signature(rep, policies) == sig)
-            .map(|(&id, _)| id)
+        reg.ids_by_signature(policies)
+            .get(&signature(set, policies))
+            .copied()
     }
 
     fn sets(xs: &[&[&str]]) -> BTreeSet<GroupSet> {
diff --git a/proxmox-ve-config/src/sdn/microseg/mod.rs b/proxmox-ve-config/src/sdn/microseg/mod.rs
index 1744e3e5fc02..be48b890206f 100644
--- a/proxmox-ve-config/src/sdn/microseg/mod.rs
+++ b/proxmox-ve-config/src/sdn/microseg/mod.rs
@@ -10,6 +10,7 @@
 
 pub mod api;
 pub mod identity;
+pub mod state;
 mod types;
 
 pub use identity::PredicateMatch;
@@ -383,6 +384,19 @@ fn assignments(entries: &BTreeMap<String, MicrosegEntry>) -> Vec<Assignment<'_>>
     assignments
 }
 
+/// The NIC does not have to be among the guest's `nics`, so a NIC the inventory does not list yet
+/// resolves too.
+fn assigned_groups<'a>(
+    entries: &'a BTreeMap<String, MicrosegEntry>,
+    guest: &GuestInfo,
+    iface: u32,
+) -> Option<&'a [MicrosegId]> {
+    assignments(entries)
+        .into_iter()
+        .find(|assignment| assignment.covers(guest, iface))
+        .map(|assignment| assignment.groups)
+}
+
 /// One NIC of the inventory with the assignment that covers it first and the groups that
 /// assignment binds.
 #[derive(Debug)]
@@ -568,7 +582,7 @@ pub fn assignment_order(entries: &BTreeMap<String, MicrosegEntry>) -> HashMap<St
 }
 
 #[derive(Debug)]
-pub struct ResolvedRule {
+struct ResolvedRule {
     id: MicrosegId,
     src: identity::Predicate,
     dst: identity::Predicate,
@@ -585,6 +599,21 @@ impl ResolvedRule {
         !self.protocols.is_empty()
     }
 
+    /// A narrowed deny that can match neighbor discovery.
+    fn can_deny_nd(&self) -> bool {
+        !self.allow
+            && self.protocols.iter().any(|protocol| match protocol {
+                Protocol::Icmpv6(icmp) => matches!(
+                    icmp.ty(),
+                    None | Some(Icmpv6Type::Any | Icmpv6Type::Numeric(135 | 136))
+                        | Some(Icmpv6Type::Named(
+                            "nd-neighbor-solicit" | "nd-neighbor-advert"
+                        ))
+                ),
+                _ => false,
+            })
+    }
+
     fn specificity(&self) -> usize {
         specificity(&self.src, &self.dst)
     }
@@ -604,9 +633,7 @@ fn reversed_protocol(protocol: &Protocol) -> Protocol {
     }
 }
 
-pub fn resolved_rules(
-    entries: &BTreeMap<String, MicrosegEntry>,
-) -> Result<Vec<ResolvedRule>, Error> {
+fn resolved_rules(entries: &BTreeMap<String, MicrosegEntry>) -> Result<Vec<ResolvedRule>, Error> {
     let mut resolved = Vec::new();
     for rule in rule_entries(entries) {
         let context = || format!("rule '{}'", rule.id);
@@ -635,10 +662,9 @@ pub fn resolved_rules(
 }
 
 #[derive(Debug)]
-pub struct Folded<'a> {
-    /// Narrowed rules in evaluation order. The first one whose protocol match fires decides.
-    pub checks: Vec<&'a ResolvedRule>,
-    pub allow: bool,
+struct Folded<'a> {
+    checks: Vec<&'a ResolvedRule>,
+    allow: bool,
 }
 
 /// Fold the rules that fire on one `(src_id, dst_id)` pair into its verdict and checks. The
@@ -646,7 +672,7 @@ pub struct Folded<'a> {
 /// specific of those rules decide, and a single deny among them wins. Without such a rule the pair
 /// denies. The checks are the narrowed rules from the top priority down to that one, the more
 /// specific first and denies ahead of allows within a priority.
-pub fn fold<'a>(firing: impl IntoIterator<Item = &'a ResolvedRule>) -> Folded<'a> {
+fn fold<'a>(firing: impl IntoIterator<Item = &'a ResolvedRule>) -> Folded<'a> {
     let mut firing: Vec<&ResolvedRule> = firing.into_iter().collect();
     firing.sort_by(|a, b| evaluation_order(a, b));
 
diff --git a/proxmox-ve-config/src/sdn/microseg/state.rs b/proxmox-ve-config/src/sdn/microseg/state.rs
new file mode 100644
index 000000000000..1392860f5062
--- /dev/null
+++ b/proxmox-ve-config/src/sdn/microseg/state.rs
@@ -0,0 +1,612 @@
+//! The mechanism-independent form of the microseg policy, built from the running config.
+//!
+//! It holds the wire id each managed guest NIC stamps and the folded `(src_id, dst_id)` pairs.
+
+use std::collections::{BTreeMap, BTreeSet, HashMap};
+
+use anyhow::{Context, bail};
+
+use super::identity;
+use super::identity::UNTAGGED_GROUP;
+use super::types::{GuestInfo, MicrosegEntry, MicrosegId, MicrosegRunningConfig};
+use super::{
+    ResolvedRule, assigned_groups, assigned_nics, build_policies, closed_groups, fold,
+    resolved_rules,
+};
+use crate::firewall::types::rule_match::Protocol;
+
+#[derive(Debug)]
+pub struct DesiredState {
+    /// The policy has an assignment, so there is something to enforce even on a node without an
+    /// assigned NIC.
+    pub configured: bool,
+    /// The pairs an allow fires on or where a deny of neighbor discovery has to hold, left out
+    /// when their fold neither admits nor checks anything. A missing pair is a deny, the data
+    /// plane's default.
+    pub pairs: Vec<Pair>,
+    /// The narrowed rules the pairs refer to by index.
+    pub checks: Vec<Check>,
+    /// The wire id each managed guest NIC should stamp.
+    pub nics: Vec<ResolvedNic>,
+}
+
+#[derive(Debug)]
+pub struct Pair {
+    pub src_id: u16,
+    pub dst_id: u16,
+    /// The indexes of the narrowed rules to run in order. The first whose match fires decides.
+    pub checks: Vec<usize>,
+    pub allow: bool,
+}
+
+/// A narrowed rule.
+#[derive(Debug)]
+pub struct Check {
+    pub protocols: Vec<Protocol>,
+    pub allow: bool,
+}
+
+#[derive(Debug)]
+pub struct ResolvedNic {
+    pub vmid: u32,
+    pub iface: u32,
+    /// The interface's wire id (the identity its group set resolves to).
+    pub id: u16,
+}
+
+/// Building the policies and the class signatures walks the whole config, so do it once per build
+/// instead of once per NIC.
+struct Resolver<'a> {
+    entries: &'a BTreeMap<String, MicrosegEntry>,
+    policies: Vec<identity::Policy>,
+    class_ids: HashMap<identity::Signature, u16>,
+}
+
+impl<'a> Resolver<'a> {
+    fn new(cfg: &'a MicrosegRunningConfig) -> anyhow::Result<Self> {
+        let entries = cfg.ids();
+        let policies = build_policies(entries).context("build policies")?;
+        let class_ids = cfg.identities().ids_by_signature(&policies);
+        Ok(Self {
+            entries,
+            policies,
+            class_ids,
+        })
+    }
+
+    /// The render allocates a class for every set an assignment names, so a set no class carries is
+    /// a broken registry rather than a NIC to leave untagged.
+    fn id_of(&self, vmid: u32, iface: u32, groups: &[MicrosegId]) -> anyhow::Result<u16> {
+        let set = closed_groups(self.entries, groups)
+            .with_context(|| format!("NIC vm{vmid} net{iface}"))?;
+        let Some(&id) = self
+            .class_ids
+            .get(&identity::signature(&set, &self.policies))
+        else {
+            bail!("NIC vm{vmid} net{iface} has no allocated identity");
+        };
+        Ok(id)
+    }
+
+    fn nics(&self, inventory: &[GuestInfo]) -> anyhow::Result<Vec<ResolvedNic>> {
+        assigned_nics(self.entries, inventory)
+            .into_iter()
+            .map(|nic| {
+                Ok(ResolvedNic {
+                    vmid: nic.vmid,
+                    iface: nic.iface,
+                    id: self.id_of(nic.vmid, nic.iface, nic.groups)?,
+                })
+            })
+            .collect()
+    }
+}
+
+/// The wire id a guest NIC stamps, `None` when no assignment covers it. Skips the pair fold, which
+/// one NIC does not need.
+pub fn resolve_iface(
+    cfg: &MicrosegRunningConfig,
+    guest: &GuestInfo,
+    iface: u32,
+) -> anyhow::Result<Option<u16>> {
+    let Some(groups) = assigned_groups(cfg.ids(), guest, iface) else {
+        return Ok(None);
+    };
+    Resolver::new(cfg)?
+        .id_of(guest.vmid, iface, groups)
+        .map(Some)
+}
+
+impl DesiredState {
+    /// `pairs_max` is the size of the data plane's pairs map. A policy that needs more pairs
+    /// returns Err before any pair is folded.
+    pub fn build(
+        cfg: &MicrosegRunningConfig,
+        inventory: &[GuestInfo],
+        pairs_max: usize,
+    ) -> anyhow::Result<Self> {
+        let resolver = Resolver::new(cfg)?;
+        let rules = resolved_rules(resolver.entries).context("resolve rules")?;
+        let nics = resolver.nics(inventory)?;
+        let firing = firing_pairs(cfg.identities().classes(), &rules, pairs_max)?;
+        let (pairs, checks) = fold_pairs(&firing);
+        Ok(Self {
+            // a class exists for every group set an assignment names
+            configured: !cfg.identities().classes().is_empty(),
+            pairs,
+            checks,
+            nics,
+        })
+    }
+}
+
+/// The classes a rule fires on, ascending for the binary search.
+struct Reach {
+    sources: Vec<u16>,
+    destinations: Vec<u16>,
+}
+
+fn rule_reaches(classes: &BTreeMap<u16, identity::GroupSet>, rules: &[ResolvedRule]) -> Vec<Reach> {
+    // Every allocated class is folded, whether or not a NIC carries it right now, since a NIC
+    // that comes up later can only land in one of them. Only the sources include the untagged
+    // id, so an `untagged -> X` rule admits unstamped traffic. Enforcement runs at assigned
+    // NICs, whose sets never carry the untagged group, so no untagged destination is folded.
+    let untagged_rep = identity::GroupSet::from([UNTAGGED_GROUP.to_string()]);
+    let sources: Vec<(u16, &identity::GroupSet)> =
+        std::iter::once((identity::UNTAGGED_ID, &untagged_rep))
+            .chain(classes.iter().map(|(&id, rep)| (id, rep)))
+            .collect();
+    rules
+        .iter()
+        .map(|rule| Reach {
+            sources: sources
+                .iter()
+                .filter(|(_, rep)| rule.src.fires(rep))
+                .map(|&(id, _)| id)
+                .collect(),
+            destinations: classes
+                .iter()
+                .filter(|(_, rep)| rule.dst.fires(rep))
+                .map(|(&id, _)| id)
+                .collect(),
+        })
+        .collect()
+}
+
+/// Each candidate pair with the rules that fire on it, as `(src_id, dst_id, rule)`, grouped by
+/// pair in pair order. A candidate is a pair an allow fires on, or one where a narrowed deny
+/// of neighbor discovery has to hold. More candidates than `pairs_max` fail the build.
+fn firing_pairs<'r>(
+    classes: &BTreeMap<u16, identity::GroupSet>,
+    rules: &'r [ResolvedRule],
+    pairs_max: usize,
+) -> anyhow::Result<Vec<(u16, u16, &'r ResolvedRule)>> {
+    let reaches = rule_reaches(classes, rules);
+    let mut candidates = BTreeSet::new();
+    for (rule, reach) in rules.iter().zip(&reaches) {
+        if !rule.allow {
+            continue;
+        }
+        for &src_id in &reach.sources {
+            for &dst_id in &reach.destinations {
+                add_candidate(&mut candidates, (src_id, dst_id), pairs_max)?;
+            }
+        }
+    }
+    // Denies alone need no pair, since a missing pair denies. The exception is a narrowed deny
+    // that can match neighbor discovery. The data plane passes neighbor discovery when the
+    // reverse pair admits anything, and from an unstamped sender when the host sent it or the
+    // NIC sits behind a firewall bridge. So such a deny gets a pair wherever that pass applies.
+    for (rule, reach) in rules.iter().zip(&reaches) {
+        if !rule.can_deny_nd() {
+            continue;
+        }
+        for &src_id in &reach.sources {
+            for &dst_id in &reach.destinations {
+                // the reverse test can read the set while it grows, an added exception only
+                // matches the test of the allow candidate it was added for, which is in the
+                // set already
+                if src_id == identity::UNTAGGED_ID || candidates.contains(&(dst_id, src_id)) {
+                    add_candidate(&mut candidates, (src_id, dst_id), pairs_max)?;
+                }
+            }
+        }
+    }
+    // a rule reaching more pairs than there are candidates is tested per candidate instead of
+    // expanded, so a deny over everything costs one pass
+    let mut firing = Vec::new();
+    for (rule, reach) in rules.iter().zip(&reaches) {
+        if reach.sources.len() * reach.destinations.len() <= candidates.len() {
+            for &src_id in &reach.sources {
+                for &dst_id in &reach.destinations {
+                    if candidates.contains(&(src_id, dst_id)) {
+                        firing.push((src_id, dst_id, rule));
+                    }
+                }
+            }
+        } else {
+            for &(src_id, dst_id) in &candidates {
+                if reach.sources.binary_search(&src_id).is_ok()
+                    && reach.destinations.binary_search(&dst_id).is_ok()
+                {
+                    firing.push((src_id, dst_id, rule));
+                }
+            }
+        }
+    }
+    firing.sort_by_key(|&(src_id, dst_id, _)| (src_id, dst_id));
+    Ok(firing)
+}
+
+fn add_candidate(
+    candidates: &mut BTreeSet<(u16, u16)>,
+    pair: (u16, u16),
+    pairs_max: usize,
+) -> anyhow::Result<()> {
+    candidates.insert(pair);
+    if candidates.len() > pairs_max {
+        bail!(
+            "the policy needs more than {pairs_max} pairs, the pairs map holds at most {pairs_max}"
+        );
+    }
+    Ok(())
+}
+
+fn fold_pairs(firing: &[(u16, u16, &ResolvedRule)]) -> (Vec<Pair>, Vec<Check>) {
+    let mut pairs = Vec::new();
+    let mut checks = Vec::new();
+    // The two orientations of a bidirectional rule carry the rule's id and match opposite ports,
+    // so each of them is a check of its own.
+    let mut check_index: HashMap<(&MicrosegId, bool), usize> = HashMap::new();
+    for pair in firing.chunk_by(|a, b| (a.0, a.1) == (b.0, b.1)) {
+        let (src_id, dst_id, _) = pair[0];
+        let folded = fold(pair.iter().map(|&(_, _, rule)| rule));
+        if !folded.allow && folded.checks.is_empty() {
+            continue;
+        }
+        let cell_checks = folded
+            .checks
+            .iter()
+            .map(|rule| {
+                *check_index
+                    .entry((&rule.id, rule.reverse))
+                    .or_insert_with(|| {
+                        checks.push(Check {
+                            protocols: rule.protocols.clone(),
+                            allow: rule.allow,
+                        });
+                        checks.len() - 1
+                    })
+            })
+            .collect();
+        pairs.push(Pair {
+            src_id,
+            dst_id,
+            checks: cell_checks,
+            allow: folded.allow,
+        });
+    }
+    (pairs, checks)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::tests::firing_rules;
+    use super::super::{Folded, render};
+    use super::*;
+
+    const WEB_AND_DB: &str = r#"{"next":3,"classes":{"1":["web"],"2":["db"]}}"#;
+
+    fn web_to_db(assignments: &str, identities: &str) -> MicrosegRunningConfig {
+        let json = format!(
+            r#"{{
+            "ids": {{
+                "web": {{"type":"group","id":"web"}},
+                "db": {{"type":"group","id":"db"}},
+                "r1": {{"type":"rule","id":"r1","src":["web"],"dst":["db"],"allow":1}},
+                {assignments}
+            }},
+            "identities": {identities}
+        }}"#
+        );
+        serde_json::from_str(&json).expect("parse running config")
+    }
+
+    fn vm100_in(group: &str) -> String {
+        format!(
+            r#""vm100i0": {{"type":"guest-assignment","id":"vm100i0","vmid":100,"iface":0,"groups":["{group}"]}}"#
+        )
+    }
+
+    fn guest(vmid: u32, tags: &[&str], nics: &[u32]) -> GuestInfo {
+        GuestInfo {
+            vmid,
+            tags: tags
+                .iter()
+                .map(|tag| tag.parse().expect("valid tag"))
+                .collect(),
+            name: None,
+            nics: nics.to_vec(),
+        }
+    }
+
+    fn stamped(state: &DesiredState) -> Vec<(u32, u32, u16)> {
+        state.nics.iter().map(|a| (a.vmid, a.iface, a.id)).collect()
+    }
+
+    /// Two tag assignments that both match a guest tagged `a` and `b`. By id `tdb` sorts ahead of
+    /// `tweb`.
+    fn two_tags(tweb: &str, tdb: &str) -> MicrosegRunningConfig {
+        let assignments = format!(
+            r#""tdb": {{"type":"tag-assignment","id":"tdb","tags":["b"],"groups":["db"]{tdb}}},
+                "tweb": {{"type":"tag-assignment","id":"tweb","tags":["a"],"groups":["web"]{tweb}}}"#
+        );
+        web_to_db(&assignments, WEB_AND_DB)
+    }
+
+    #[test]
+    fn resolve_iface_answers_one_nic() {
+        let cfg = web_to_db(&vm100_in("web"), WEB_AND_DB);
+        let vm100 = guest(100, &[], &[0]);
+        assert_eq!(resolve_iface(&cfg, &vm100, 0).expect("resolve"), Some(1));
+
+        assert_eq!(resolve_iface(&cfg, &vm100, 1).expect("resolve"), None);
+        assert_eq!(
+            resolve_iface(&cfg, &guest(404, &[], &[0]), 0).expect("resolve"),
+            None
+        );
+
+        let cfg = web_to_db(&vm100_in("db"), r#"{"next":2,"classes":{"1":["web"]}}"#);
+        assert!(resolve_iface(&cfg, &vm100, 0).is_err());
+        assert!(
+            DesiredState::build(&cfg, &[vm100], usize::MAX).is_err(),
+            "the full build agrees"
+        );
+    }
+
+    #[test]
+    fn first_match_takes_one_assignment_not_a_union() {
+        let cfg = two_tags(r#","order":1"#, r#","order":2"#);
+        let tagged = guest(100, &["a", "b"], &[0]);
+        let state = DesiredState::build(&cfg, std::slice::from_ref(&tagged), usize::MAX)
+            .expect("build state");
+
+        // the earlier tweb decides alone, a union {web, db} would have no class at all
+        assert_eq!(stamped(&state), [(100, 0, 1)]);
+        assert_eq!(resolve_iface(&cfg, &tagged, 0).expect("resolve"), Some(1));
+    }
+
+    #[test]
+    fn unmatched_assignment_still_gets_a_class() {
+        let cfg = two_tags("", "");
+        let rendered = render(&identity::Registry::new(), cfg.ids().clone(), 0).expect("render");
+        let state = DesiredState::build(&rendered, &[], usize::MAX).expect("build state");
+
+        assert!(stamped(&state).is_empty());
+        assert_eq!(rendered.describe_class(1), "1 (db)");
+        assert_eq!(rendered.describe_class(2), "2 (web)");
+        assert_eq!(
+            state.pairs.len(),
+            1,
+            "web -> db is folded for the empty classes"
+        );
+        assert_eq!(
+            resolve_iface(&rendered, &guest(100, &["b"], &[]), 0).expect("resolve"),
+            Some(1)
+        );
+    }
+
+    /// Renders a config of the groups, each with its parent, the rules, each given as its JSON
+    /// fields without the id, and one guest assignment per group list.
+    fn policy(
+        groups: &[(String, Option<String>)],
+        rules: &[String],
+        assigned: &[Vec<String>],
+    ) -> MicrosegRunningConfig {
+        let mut ids = Vec::new();
+        for (name, parent) in groups {
+            let parent = parent
+                .as_ref()
+                .map(|parent| format!(r#","parent":"{parent}""#))
+                .unwrap_or_default();
+            ids.push(format!(
+                r#""{name}":{{"type":"group","id":"{name}"{parent}}}"#
+            ));
+        }
+        for (n, fields) in rules.iter().enumerate() {
+            ids.push(format!(r#""r{n}":{{"type":"rule","id":"r{n}",{fields}}}"#));
+        }
+        for (n, groups) in assigned.iter().enumerate() {
+            let vmid = 100 + n;
+            ids.push(format!(
+                r#""vm{vmid}":{{"type":"guest-assignment","id":"vm{vmid}","vmid":{vmid},"groups":{groups:?}}}"#
+            ));
+        }
+        let json = format!(r#"{{"ids":{{{}}}}}"#, ids.join(","));
+        let cfg: MicrosegRunningConfig = serde_json::from_str(&json).expect("parse");
+        render(&identity::Registry::new(), cfg.ids().clone(), 0).expect("render")
+    }
+
+    fn tenants(count: usize) -> MicrosegRunningConfig {
+        let mut groups = vec![("mon".to_string(), None)];
+        let mut rules = Vec::new();
+        let mut assigned = vec![vec!["mon".to_string()]];
+        for t in 0..count {
+            groups.push((format!("t{t}"), None));
+            for tier in ["web", "app", "db"] {
+                groups.push((format!("t{t}-{tier}"), Some(format!("t{t}"))));
+                assigned.push(vec![format!("t{t}-{tier}")]);
+            }
+            rules.push(format!(
+                r#""src":["t{t}-web"],"dst":["t{t}-app"],"allow":1,"bidirectional":1,"prio":10"#
+            ));
+            rules.push(format!(
+                r#""src":["t{t}-app"],"dst":["t{t}-db"],"allow":1,"bidirectional":1,"prio":10"#
+            ));
+            rules.push(format!(
+                r#""src":["mon"],"dst":["t{t}"],"allow":1,"bidirectional":1,"prio":20,"proto":"tcp","dport":"9100""#
+            ));
+        }
+        policy(&groups, &rules, &assigned)
+    }
+
+    fn mesh(count: usize) -> MicrosegRunningConfig {
+        mesh_under(count, None)
+    }
+
+    /// The mesh with a deny over all of it at `deny_prio`, between the allow and the narrowed
+    /// denies, so every pair stays a candidate and only the diagonal is carried.
+    fn mesh_under(count: usize, deny_prio: Option<u16>) -> MicrosegRunningConfig {
+        let mut groups = vec![("mesh".to_string(), None)];
+        let mut rules = vec![r#""src":["mesh"],"dst":["mesh"],"allow":1,"prio":10"#.to_string()];
+        if let Some(prio) = deny_prio {
+            rules.push(format!(
+                r#""src":["mesh"],"dst":["mesh"],"allow":0,"prio":{prio}"#
+            ));
+        }
+        let mut assigned = Vec::new();
+        for k in 0..count {
+            groups.push((format!("g{k}"), None));
+            rules.push(format!(
+                r#""src":["g{k}"],"dst":["g{k}"],"allow":0,"prio":20,"proto":"tcp","dport":"22""#
+            ));
+            assigned.push(vec![format!("g{k}"), "mesh".to_string()]);
+        }
+        policy(&groups, &rules, &assigned)
+    }
+
+    fn mixed() -> MicrosegRunningConfig {
+        let groups: Vec<_> = ["prod", "staging", "web", "app", "db", "mon"]
+            .map(|name| (name.to_string(), None))
+            .into();
+        let rules = [
+            r#""src":["web","prod"],"src-match":"all","dst":["app","prod"],"dst-match":"all","allow":1,"prio":10"#,
+            r#""src":["web"],"dst":["app"],"allow":0,"prio":20,"proto":"tcp","dport":"8080""#,
+            r#""src":["web"],"dst":["db"],"allow":0,"prio":5"#,
+            r#""src":["web","staging"],"src-match":"all","dst":["db"],"allow":1,"prio":5"#,
+            r#""src":["db"],"src-match":"exact","dst":["app"],"allow":1,"prio":7"#,
+            r#""src":["mon"],"dst":["web","app","db"],"allow":1,"bidirectional":1,"prio":3,"proto":"icmp""#,
+            r#""src":["untagged"],"dst":["web"],"allow":1,"prio":1"#,
+            r#""src":["app"],"dst":["web"],"allow":1,"prio":30"#,
+            r#""src":["app"],"dst":["web"],"allow":0,"prio":30"#,
+        ]
+        .map(str::to_string);
+        let assigned = [
+            vec!["web", "prod"],
+            vec!["app", "prod"],
+            vec!["db", "prod"],
+            vec!["web", "staging"],
+            vec!["app", "staging"],
+            vec!["db"],
+            vec!["mon"],
+        ]
+        .map(|groups| groups.into_iter().map(str::to_string).collect());
+        policy(&groups, &rules, &assigned)
+    }
+
+    /// A pair as `(src_id, dst_id, checks, allow)`, each check by its protocols and verdict.
+    type CellContent = (u16, u16, Vec<(Vec<Protocol>, bool)>, bool);
+
+    fn admits(folded: &Folded) -> bool {
+        folded.allow || folded.checks.iter().any(|rule| rule.allow)
+    }
+
+    /// What `build` has to carry, folded pair by pair instead of rule by rule.
+    fn every_pair(cfg: &MicrosegRunningConfig) -> Vec<CellContent> {
+        let rules = resolved_rules(cfg.ids()).expect("resolve rules");
+        let classes = cfg.identities().classes();
+        let untagged_rep = identity::GroupSet::from([UNTAGGED_GROUP.to_string()]);
+        let sources = std::iter::once((identity::UNTAGGED_ID, &untagged_rep))
+            .chain(classes.iter().map(|(&id, rep)| (id, rep)));
+        let mut pairs = Vec::new();
+        for (src_id, src_rep) in sources {
+            for (&dst_id, dst_rep) in classes {
+                let folded = fold(firing_rules(&rules, src_rep, dst_rep));
+                let rescued = || {
+                    src_id == identity::UNTAGGED_ID
+                        || admits(&fold(firing_rules(&rules, dst_rep, src_rep)))
+                };
+                let denies_nd = folded.checks.iter().any(|rule| rule.can_deny_nd());
+                if admits(&folded) || (denies_nd && rescued()) {
+                    let checks = folded
+                        .checks
+                        .iter()
+                        .map(|rule| (rule.protocols.clone(), rule.allow))
+                        .collect();
+                    pairs.push((src_id, dst_id, checks, folded.allow));
+                }
+            }
+        }
+        pairs
+    }
+
+    #[test]
+    fn build_folds_the_cells_every_pair_would() {
+        for (name, cfg, count) in [
+            ("tenants", tenants(6), 6 * 10),
+            ("mesh", mesh(9), 9 * 9),
+            ("mixed", mixed(), 1 + 2 + 2 + 6 + 6 + 2),
+        ] {
+            let state = DesiredState::build(&cfg, &[], usize::MAX).expect("build state");
+            let built: Vec<_> = state
+                .pairs
+                .iter()
+                .map(|pair| {
+                    let checks = pair
+                        .checks
+                        .iter()
+                        .map(|&index| {
+                            let check = &state.checks[index];
+                            (check.protocols.clone(), check.allow)
+                        })
+                        .collect();
+                    (pair.src_id, pair.dst_id, checks, pair.allow)
+                })
+                .collect();
+            assert_eq!(built, every_pair(&cfg), "{name}");
+            assert_eq!(built.len(), count, "{name}");
+        }
+
+        // the bound counts the pairs an allow fires on, shadowed or not
+        let shadowed = mesh_under(9, Some(15));
+        let state = DesiredState::build(&shadowed, &[], usize::MAX).expect("build state");
+        assert_eq!(state.pairs.len(), 9, "only the diagonal is carried");
+        assert!(
+            DesiredState::build(&shadowed, &[], 80).is_err(),
+            "81 candidates are refused at a ceiling of 80"
+        );
+    }
+
+    #[test]
+    fn denies_alone_need_no_pair() {
+        let groups = [("web".to_string(), None), ("db".to_string(), None)];
+        let assigned = [vec!["web".to_string()], vec!["db".to_string()]];
+        let allow = r#""src":["web"],"dst":["db"],"allow":1,"prio":10"#.to_string();
+        let every =
+            r#""src":[],"src-match":"every","dst":[],"dst-match":"every","allow":0,"prio":20"#;
+        let pairs = |rules: &[String], pairs_max| {
+            let state = DesiredState::build(&policy(&groups, rules, &assigned), &[], pairs_max)
+                .expect("build state");
+            state
+                .pairs
+                .iter()
+                .map(|pair| (pair.src_id, pair.dst_id))
+                .collect::<Vec<_>>()
+        };
+
+        // a deny over every pair makes none of them a candidate on its own
+        let ssh = format!(r#"{every},"proto":"tcp","dport":"22""#);
+        assert_eq!(pairs(&[allow.clone(), ssh], 1), [(2, 1)]);
+
+        // a deny of ICMPv6 is kept where the reverse pair allows, db -> web next to web -> db
+        let nd = format!(r#"{every},"proto":"ipv6-icmp""#);
+        assert_eq!(pairs(&[allow.clone(), nd], 2), [(1, 2), (2, 1)]);
+        let alias = format!(r#"{every},"proto":"IPv6-ICMP""#);
+        assert_eq!(pairs(&[allow, alias], 2), [(1, 2), (2, 1)]);
+
+        // and from untagged, whose neighbor discovery passes without a reverse pair
+        let untagged =
+            r#""src":["untagged"],"dst":["web"],"allow":0,"prio":20,"proto":"ipv6-icmp""#;
+        assert_eq!(pairs(&[untagged.to_string()], 1), [(0, 2)]);
+    }
+}
-- 
2.47.3