[PATCH proxmox-ve-rs v2 02/27] ve-config: sdn: add microseg config types

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

The microseg config as Rust types, shared between the cluster-side
render and the host agent so both sides interpret the same model:
groups with a numeric mark, rules as a src/dst predicate pair
(any/all/exact over groups) with a priority and verdict, and
assignments binding a guest NIC to a set of groups. Group sets are
additive across assignments. Resolution is highest-priority-wins,
conflicting ties deny, no match denies.

Rendering validates the whole config, expands assignments into concrete
per-NIC group sets, allocates wire identities against the previous
registry and stores all of it in the running config, which is the only
thing the agent reads. Rule and assignment ids are derived from their
content, so identical objects collapse and re-creating one is
idempotent.

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 proxmox-ve-config/src/sdn/config.rs       |    9 +-
 proxmox-ve-config/src/sdn/microseg/mod.rs | 1599 +++++++++++++++++++++
 2 files changed, 1607 insertions(+), 1 deletion(-)

diff --git a/proxmox-ve-config/src/sdn/config.rs b/proxmox-ve-config/src/sdn/config.rs
index 2f30cf2..fa6f376 100644
--- a/proxmox-ve-config/src/sdn/config.rs
+++ b/proxmox-ve-config/src/sdn/config.rs
@@ -16,7 +16,7 @@ use crate::{
         Ipset,
         ipset::{IpsetEntry, IpsetName, IpsetScope},
     },
-    sdn::{SdnNameError, SubnetName, VnetName, ZoneName},
+    sdn::{SdnNameError, SubnetName, VnetName, ZoneName, microseg::MicrosegRunningConfig},
 };
 
 #[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
@@ -214,6 +214,13 @@ pub struct RunningConfig {
     zones: Option<ZonesRunningConfig>,
     subnets: Option<SubnetsRunningConfig>,
     vnets: Option<VnetsRunningConfig>,
+    microseg: Option<MicrosegRunningConfig>,
+}
+
+impl RunningConfig {
+    pub fn microseg(&self) -> Option<&MicrosegRunningConfig> {
+        self.microseg.as_ref()
+    }
 }
 
 /// A struct containing the configuration for an SDN subnet
diff --git a/proxmox-ve-config/src/sdn/microseg/mod.rs b/proxmox-ve-config/src/sdn/microseg/mod.rs
index 5217cb5..555effa 100644
--- a/proxmox-ve-config/src/sdn/microseg/mod.rs
+++ b/proxmox-ve-config/src/sdn/microseg/mod.rs
@@ -6,4 +6,1603 @@
 //! traffic by a `src` and a `dst` *predicate*, each a match kind (`any`/`all`/`exact`) over a list
 //! of groups.
 
+use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
+
+use anyhow::{anyhow, bail};
+use const_format::concatcp;
+use serde::{Deserialize, Serialize};
+
+use proxmox_schema::{ApiStringFormat, UpdaterType, api, api_string_type, const_regex};
+
 pub mod identity;
+
+pub const MICROSEG_ID_REGEX_STR: &str = r"(?:[a-zA-Z0-9](?:[a-zA-Z0-9\-_]){0,30}[a-zA-Z0-9])";
+
+const_regex! {
+    pub MICROSEG_ID_REGEX = concatcp!(r"^", MICROSEG_ID_REGEX_STR, r"$");
+}
+
+pub const MICROSEG_ID_FORMAT: ApiStringFormat = ApiStringFormat::Pattern(&MICROSEG_ID_REGEX);
+
+api_string_type! {
+    /// Identifier of a microseg object, used as the section id and to reference groups.
+    #[api(format: &MICROSEG_ID_FORMAT)]
+    #[derive(Debug, Deserialize, Serialize, Clone, Hash, PartialEq, Eq, PartialOrd, Ord, UpdaterType)]
+    pub struct MicrosegId(String);
+}
+
+#[api()]
+#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "lowercase")]
+/// How a rule predicate matches an interface's group set.
+pub enum PredicateMatch {
+    /// Fires when the interface shares at least one of the listed groups (intersection non-empty).
+    #[default]
+    Any,
+    /// Fires when the interface has all of the listed groups (the list is a subset of its groups).
+    All,
+    /// Fires only when the interface's groups equal the listed set exactly.
+    Exact,
+}
+
+impl From<PredicateMatch> for identity::PredicateKind {
+    fn from(value: PredicateMatch) -> Self {
+        match value {
+            PredicateMatch::Any => identity::PredicateKind::Any,
+            PredicateMatch::All => identity::PredicateKind::All,
+            PredicateMatch::Exact => identity::PredicateKind::Exact,
+        }
+    }
+}
+
+#[api(
+    properties: {
+        mark: {
+            type: Integer,
+            minimum: 1,
+            maximum: 65535,
+            description: "Numeric group mark, the building block of the wire identity (1-65535).",
+        },
+        comment: {
+            type: String,
+            optional: true,
+            max_length: 256,
+            description: "Free-form comment.",
+        },
+    },
+)]
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+/// A microseg group. Its numeric [`mark`](Self::mark) is one element of the group set an interface
+/// is assigned. The agent maps each distinct set to a wire identity. Mark 0 is reserved.
+pub struct GroupSection {
+    pub(crate) id: MicrosegId,
+    pub(crate) mark: u16,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(crate) comment: Option<String>,
+}
+
+impl GroupSection {
+    pub fn mark(&self) -> u16 {
+        self.mark
+    }
+    pub fn comment(&self) -> Option<&str> {
+        self.comment.as_deref()
+    }
+}
+
+#[api(
+    properties: {
+        src: {
+            type: Array,
+            items: { type: String, description: "A group id.", format: &MICROSEG_ID_FORMAT },
+        },
+        dst: {
+            type: Array,
+            items: { type: String, description: "A group id.", format: &MICROSEG_ID_FORMAT },
+        },
+        comment: {
+            type: String,
+            optional: true,
+            max_length: 256,
+            description: "Free-form comment.",
+        },
+    },
+)]
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+/// A policy rule. Traffic is permitted when `allow` is true and both predicates fire: the source
+/// interface's group set matches `(src_match, src)` and the destination's matches `(dst_match,
+/// dst)`. With no matching rule the data plane denies.
+pub struct RuleSection {
+    pub(crate) id: MicrosegId,
+    /// Groups the source predicate is evaluated against.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub(crate) src: Vec<MicrosegId>,
+    /// How the source predicate matches its groups.
+    #[serde(default)]
+    pub(crate) src_match: PredicateMatch,
+    /// Groups the destination predicate is evaluated against.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub(crate) dst: Vec<MicrosegId>,
+    /// How the destination predicate matches its groups.
+    #[serde(default)]
+    pub(crate) dst_match: PredicateMatch,
+    /// Priority. The highest-priority matching rule wins. At a tie, a deny overrides an allow.
+    #[serde(default)]
+    pub(crate) prio: u16,
+    /// Whether traffic matching both predicates is allowed.
+    #[serde(deserialize_with = "proxmox_serde::perl::deserialize_bool")]
+    pub(crate) allow: bool,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(crate) comment: Option<String>,
+}
+
+impl RuleSection {
+    pub fn src(&self) -> &[MicrosegId] {
+        &self.src
+    }
+    pub fn src_match(&self) -> PredicateMatch {
+        self.src_match
+    }
+    pub fn dst(&self) -> &[MicrosegId] {
+        &self.dst
+    }
+    pub fn dst_match(&self) -> PredicateMatch {
+        self.dst_match
+    }
+    pub fn prio(&self) -> u16 {
+        self.prio
+    }
+    pub fn allow(&self) -> bool {
+        self.allow
+    }
+    pub fn comment(&self) -> Option<&str> {
+        self.comment.as_deref()
+    }
+}
+
+#[api(
+    properties: {
+        vmid: {
+            type: Integer,
+            minimum: 1,
+            maximum: 999999999,
+            description: "Guest (VM or CT) id whose NIC(s) are bound.",
+        },
+        iface: {
+            type: Integer,
+            optional: true,
+            minimum: 0,
+            maximum: 31,
+            description: "The guest's netN interface. If unset, all of the guest's NICs.",
+        },
+        groups: {
+            type: Array,
+            description: "Groups the matching NICs belong to.",
+            items: { type: String, description: "A group id.", format: &MICROSEG_ID_FORMAT },
+        },
+        comment: {
+            type: String,
+            optional: true,
+            max_length: 256,
+            description: "Free-form comment.",
+        },
+    },
+)]
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+/// Binds a specific guest's NIC(s) to a set of groups: all of guest `vmid`'s NICs, or just the one
+/// named by `iface` when set. Stored as the `guestassignment` section type. A new matcher kind is a
+/// new sibling section type discriminated by `type`, not a new field here.
+pub struct GuestAssignmentSection {
+    pub(crate) id: MicrosegId,
+    pub(crate) vmid: u32,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(crate) iface: Option<u32>,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub(crate) groups: Vec<MicrosegId>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(crate) comment: Option<String>,
+}
+
+#[api(
+    "id-property": "id",
+    "id-schema": {
+        type: String,
+        description: "Microseg object identifier.",
+        format: &MICROSEG_ID_FORMAT,
+    },
+    "type-key": "type",
+)]
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "lowercase", tag = "type")]
+/// One entry in the microseg section config, discriminated by the section `type`.
+pub enum MicrosegEntry {
+    /// A group.
+    Group(GroupSection),
+    /// A policy rule.
+    Rule(RuleSection),
+    /// A guest-matcher NIC-to-groups assignment.
+    GuestAssignment(GuestAssignmentSection),
+}
+
+impl MicrosegEntry {
+    /// The section id this entry is keyed by.
+    pub fn id(&self) -> &MicrosegId {
+        match self {
+            Self::Group(g) => &g.id,
+            Self::Rule(r) => &r.id,
+            Self::GuestAssignment(a) => &a.id,
+        }
+    }
+}
+
+/// A concrete per-NIC group assignment produced by [`render`], stored in the running config and
+/// resolved to a wire id by the agent.
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
+pub struct RealizedAssignment {
+    pub vmid: u32,
+    pub iface: u32,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub groups: Vec<MicrosegId>,
+}
+
+/// The guests/NICs a single assignment expands to. Unlike [`RealizedAssignment`] this is not merged
+/// across assignments and carries no groups: it answers "which NICs does *this* matcher cover".
+#[derive(Clone, Debug, Serialize)]
+pub struct AssignmentMember {
+    pub vmid: u32,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub nics: Vec<u32>,
+}
+
+/// Deserialize a `Vec<u32>` whose elements arrive as perl's untyped (often string) scalars, by
+/// running each through the [`deserialize_u32`](proxmox_serde::perl::deserialize_u32) helper.
+fn deserialize_perl_nics<'de, D>(deserializer: D) -> Result<Vec<u32>, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    #[derive(Deserialize)]
+    struct Nic(#[serde(deserialize_with = "proxmox_serde::perl::deserialize_u32")] u32);
+    Ok(Vec::<Nic>::deserialize(deserializer)?
+        .into_iter()
+        .map(|Nic(index)| index)
+        .collect())
+}
+
+/// One guest in the inventory that [`render`] matches assignments against. Gathered cluster-side
+/// (the only layer with guest data) and passed in, so the engine needs no cluster access.
+#[derive(Clone, Debug, Default, Deserialize)]
+pub struct GuestInfo {
+    #[serde(deserialize_with = "proxmox_serde::perl::deserialize_u32")]
+    pub vmid: u32,
+    /// The present netN interface indices. A matcher with no `iface` covers all of them.
+    #[serde(default, deserialize_with = "deserialize_perl_nics")]
+    pub nics: Vec<u32>,
+}
+
+/// The microseg block of the SDN running config (`running.microseg`), an `ids` map of entries
+/// keyed by section id.
+#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
+pub struct MicrosegRunningConfig {
+    #[serde(default)]
+    ids: HashMap<String, MicrosegEntry>,
+    /// The compiled identity registry (id to representative set), keeping ids stable across renders.
+    #[serde(default)]
+    identities: identity::Registry,
+    /// Concrete per-NIC assignments after expanding selectors against the guest inventory and
+    /// merging with the static assignments.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    realized: Vec<RealizedAssignment>,
+}
+
+impl MicrosegRunningConfig {
+    pub fn ids(&self) -> &HashMap<String, MicrosegEntry> {
+        &self.ids
+    }
+
+    pub fn identities(&self) -> &identity::Registry {
+        &self.identities
+    }
+
+    pub fn groups(&self) -> impl Iterator<Item = (&str, &GroupSection)> + '_ {
+        self.ids.iter().filter_map(|(k, v)| match v {
+            MicrosegEntry::Group(g) => Some((k.as_str(), g)),
+            _ => None,
+        })
+    }
+
+    pub fn rules(&self) -> impl Iterator<Item = (&str, &RuleSection)> + '_ {
+        self.ids.iter().filter_map(|(k, v)| match v {
+            MicrosegEntry::Rule(r) => Some((k.as_str(), r)),
+            _ => None,
+        })
+    }
+
+    /// The rendered concrete per-NIC assignments (static plus selector-expanded) the agent enforces.
+    pub fn realized(&self) -> &[RealizedAssignment] {
+        &self.realized
+    }
+}
+
+/// The reserved name of the built-in *untagged* group (mark 0), the no-group identity. A rule
+/// predicate may name it to match unstamped traffic (gateway / DHCP / ARP replies, cross-node
+/// frames without a GBP tag), and it composes with real groups (e.g. `any{untagged, web}`). It is
+/// not a storable group and cannot be used in an assignment.
+pub const UNTAGGED_GROUP: &str = "untagged";
+
+/// The mark of the [`UNTAGGED_GROUP`], the same number as the untagged wire id.
+pub const UNTAGGED_MARK: u16 = 0;
+
+/// Validation of a microseg config. Group marks are unique and the reserved `untagged` name is not
+/// taken by a real group. Every group named by a rule predicate exists or is the built-in
+/// `untagged`. Every assignment group exists and is not `untagged`. Predicates and assignments name
+/// at least one group. Exact-duplicate rules and assignments cannot arise, because their ids are
+/// derived from their contents, so they collapse to one entry.
+pub fn validate(entries: &HashMap<String, MicrosegEntry>) -> Result<(), anyhow::Error> {
+    let mut group_marks: HashMap<u16, &str> = HashMap::new();
+    let mut group_names: HashSet<&str> = HashSet::new();
+    for (name, entry) in entries {
+        if let MicrosegEntry::Group(group) = entry {
+            if name == UNTAGGED_GROUP {
+                bail!("'{UNTAGGED_GROUP}' is a reserved group name");
+            }
+            group_names.insert(name.as_str());
+            if let Some(other) = group_marks.insert(group.mark, name.as_str()) {
+                bail!(
+                    "group mark {} used by both '{other}' and '{name}'",
+                    group.mark
+                );
+            }
+        }
+    }
+
+    let check_group = |kind: &str, name: &str, group: &str| -> Result<(), anyhow::Error> {
+        if !group_names.contains(group) {
+            bail!("{kind} '{name}' references unknown group '{group}'");
+        }
+        Ok(())
+    };
+
+    let check_assignment_groups =
+        |name: &str, groups: &[MicrosegId]| -> Result<(), anyhow::Error> {
+            if groups.is_empty() {
+                bail!("assignment '{name}' is not assigned to any group");
+            }
+            for group in groups {
+                if group.as_ref() == UNTAGGED_GROUP {
+                    bail!("assignment '{name}' cannot use the reserved '{UNTAGGED_GROUP}' group");
+                }
+                check_group("assignment", name, group)?;
+            }
+            Ok(())
+        };
+
+    for (name, entry) in entries {
+        match entry {
+            MicrosegEntry::Rule(rule) => {
+                if rule.src.is_empty() {
+                    bail!("rule '{name}' has an empty source predicate");
+                }
+                if rule.dst.is_empty() {
+                    bail!("rule '{name}' has an empty destination predicate");
+                }
+                for group in rule.src.iter().chain(&rule.dst) {
+                    if group.as_ref() == UNTAGGED_GROUP {
+                        continue;
+                    }
+                    check_group("rule", name, group)?;
+                }
+            }
+            MicrosegEntry::GuestAssignment(assignment) => {
+                check_assignment_groups(name, &assignment.groups)?;
+            }
+            MicrosegEntry::Group(_) => {}
+        }
+    }
+
+    Ok(())
+}
+
+/// Map each group name to its numeric mark, including the built-in `untagged` group (mark 0).
+fn group_marks(entries: &HashMap<String, MicrosegEntry>) -> HashMap<&str, u16> {
+    let mut marks: HashMap<&str, u16> = entries
+        .iter()
+        .filter_map(|(name, entry)| match entry {
+            MicrosegEntry::Group(group) => Some((name.as_str(), group.mark)),
+            _ => None,
+        })
+        .collect();
+    marks.insert(UNTAGGED_GROUP, UNTAGGED_MARK);
+    marks
+}
+
+/// Build an engine predicate from a config predicate, resolving group names to marks.
+fn predicate(
+    name_marks: &HashMap<&str, u16>,
+    match_kind: PredicateMatch,
+    groups: &[MicrosegId],
+) -> anyhow::Result<identity::Predicate> {
+    let marks = groups
+        .iter()
+        .map(|group| {
+            name_marks
+                .get(group.as_ref())
+                .copied()
+                .ok_or_else(|| anyhow!("references unknown group '{group}'"))
+        })
+        .collect::<anyhow::Result<Vec<u16>>>()?;
+    Ok(identity::Predicate::new(match_kind.into(), marks))
+}
+
+/// The policy list in canonical order (sorted by rule id), predicates resolved to group marks. The
+/// agent must rebuild this in the same order, so the signatures it computes match the rendered
+/// ones.
+pub fn build_policies(
+    entries: &HashMap<String, MicrosegEntry>,
+) -> anyhow::Result<Vec<identity::Policy>> {
+    let name_marks = group_marks(entries);
+    let mut rules: Vec<(&str, &RuleSection)> = entries
+        .iter()
+        .filter_map(|(id, entry)| match entry {
+            MicrosegEntry::Rule(rule) => Some((id.as_str(), rule)),
+            _ => None,
+        })
+        .collect();
+    rules.sort_by(|a, b| a.0.cmp(b.0));
+    rules
+        .into_iter()
+        .map(|(_, rule)| {
+            Ok(identity::Policy {
+                src: predicate(&name_marks, rule.src_match, &rule.src)?,
+                dst: predicate(&name_marks, rule.dst_match, &rule.dst)?,
+            })
+        })
+        .collect()
+}
+
+fn covered_nics(iface: Option<u32>, nics: &[u32]) -> Vec<u32> {
+    match iface {
+        Some(iface) => nics.iter().copied().filter(|&nic| nic == iface).collect(),
+        None => nics.to_vec(),
+    }
+}
+
+/// Resolve every assignment into the concrete per-NIC group assignments, unioning the groups of
+/// every matcher that names the same NIC. A matcher expands against `inventory` to the guests it
+/// covers, binding its `iface` or, when unset, all of a guest's present NICs. Sorted by (vmid,
+/// iface) for a stable running config.
+pub fn realized_assignments(
+    entries: &HashMap<String, MicrosegEntry>,
+    inventory: &[GuestInfo],
+) -> Vec<RealizedAssignment> {
+    let mut merged: BTreeMap<(u32, u32), BTreeSet<MicrosegId>> = BTreeMap::new();
+    // vmid -> present NICs, so a guest matcher with no iface can cover all of the guest's NICs.
+    let nics_of: HashMap<u32, &[u32]> = inventory
+        .iter()
+        .map(|guest| (guest.vmid, guest.nics.as_slice()))
+        .collect();
+
+    for entry in entries.values() {
+        match entry {
+            MicrosegEntry::GuestAssignment(assignment) => {
+                let nics = nics_of.get(&assignment.vmid).copied().unwrap_or_default();
+                for nic in covered_nics(assignment.iface, nics) {
+                    merged
+                        .entry((assignment.vmid, nic))
+                        .or_default()
+                        .extend(assignment.groups.iter().cloned());
+                }
+            }
+            MicrosegEntry::Group(_) | MicrosegEntry::Rule(_) => {}
+        }
+    }
+
+    merged
+        .into_iter()
+        .map(|((vmid, iface), groups)| RealizedAssignment {
+            vmid,
+            iface,
+            groups: groups.into_iter().collect(),
+        })
+        .collect()
+}
+
+/// Expand each assignment *independently* against `inventory` into the guests/NICs it covers, keyed
+/// by assignment id. Unlike [`realized_assignments`] nothing is merged. Each assignment yields its
+/// own members, one per guest it matches.
+pub fn realized_per_assignment(
+    entries: &HashMap<String, MicrosegEntry>,
+    inventory: &[GuestInfo],
+) -> HashMap<String, Vec<AssignmentMember>> {
+    let nics_of: HashMap<u32, &[u32]> = inventory
+        .iter()
+        .map(|guest| (guest.vmid, guest.nics.as_slice()))
+        .collect();
+
+    entries
+        .iter()
+        .filter_map(|(id, entry)| {
+            let members = match entry {
+                MicrosegEntry::GuestAssignment(assignment) => {
+                    let nics = covered_nics(
+                        assignment.iface,
+                        nics_of.get(&assignment.vmid).copied().unwrap_or_default(),
+                    );
+                    if nics.is_empty() {
+                        Vec::new()
+                    } else {
+                        vec![AssignmentMember {
+                            vmid: assignment.vmid,
+                            nics,
+                        }]
+                    }
+                }
+                MicrosegEntry::Group(_) | MicrosegEntry::Rule(_) => {
+                    return None;
+                }
+            };
+            Some((id.clone(), members))
+        })
+        .collect()
+}
+
+/// A realized identity class, a distinct group-set carried by some realized NIC, paired with the
+/// rules that fire on it. `outbound` where it is a valid source, `inbound` where a valid
+/// destination. Matchability is the engine's own [`identity::Predicate::fires`], so this stays the
+/// single source of truth.
+#[derive(Debug, Clone, Serialize)]
+pub struct IdentityRules {
+    /// The class, its groups sorted.
+    pub groups: Vec<MicrosegId>,
+    /// Ids of the rules whose source predicate fires on this identity.
+    pub outbound: Vec<MicrosegId>,
+    /// Ids of the rules whose destination predicate fires on this identity.
+    pub inbound: Vec<MicrosegId>,
+}
+
+/// For every realized identity class, the rules it fires as source and as destination.
+pub fn rules_by_identity(
+    entries: &HashMap<String, MicrosegEntry>,
+    inventory: &[GuestInfo],
+) -> anyhow::Result<Vec<IdentityRules>> {
+    let name_marks = group_marks(entries);
+
+    let mut rules: Vec<(&MicrosegId, identity::Predicate, identity::Predicate)> = Vec::new();
+    for entry in entries.values() {
+        if let MicrosegEntry::Rule(rule) = entry {
+            let src = predicate(&name_marks, rule.src_match, &rule.src)?;
+            let dst = predicate(&name_marks, rule.dst_match, &rule.dst)?;
+            rules.push((&rule.id, src, dst));
+        }
+    }
+    rules.sort_by(|a, b| a.0.cmp(b.0));
+
+    let mut classes: BTreeSet<BTreeSet<MicrosegId>> = BTreeSet::new();
+    for assignment in realized_assignments(entries, inventory) {
+        classes.insert(assignment.groups.into_iter().collect());
+    }
+
+    Ok(classes
+        .into_iter()
+        .map(|class| {
+            let marks: identity::GroupSet = class
+                .iter()
+                .filter_map(|group| name_marks.get(group.as_ref()).copied())
+                .collect();
+            let mut outbound = Vec::new();
+            let mut inbound = Vec::new();
+            for (id, src, dst) in &rules {
+                if src.fires(&marks) {
+                    outbound.push((*id).clone());
+                }
+                if dst.fires(&marks) {
+                    inbound.push((*id).clone());
+                }
+            }
+            IdentityRules {
+                groups: class.into_iter().collect(),
+                outbound,
+                inbound,
+            }
+        })
+        .collect())
+}
+
+/// The distinct group-sets (in marks) of the realized assignments, the input the identity allocator
+/// partitions.
+fn realized_group_sets(
+    entries: &HashMap<String, MicrosegEntry>,
+    realized: &[RealizedAssignment],
+) -> anyhow::Result<BTreeSet<identity::GroupSet>> {
+    let name_marks = group_marks(entries);
+    realized
+        .iter()
+        .map(|assignment| {
+            assignment
+                .groups
+                .iter()
+                .map(|group| {
+                    name_marks.get(group.as_ref()).copied().ok_or_else(|| {
+                        anyhow!("realized assignment references unknown group '{group}'")
+                    })
+                })
+                .collect::<anyhow::Result<identity::GroupSet>>()
+        })
+        .collect()
+}
+
+/// Resolve a list of group names to their canonical mark set (sorted, deduped).
+pub fn group_set(
+    entries: &HashMap<String, MicrosegEntry>,
+    groups: &[MicrosegId],
+) -> anyhow::Result<identity::GroupSet> {
+    let name_marks = group_marks(entries);
+    groups
+        .iter()
+        .map(|group| {
+            name_marks
+                .get(group.as_ref())
+                .copied()
+                .ok_or_else(|| anyhow!("references unknown group '{group}'"))
+        })
+        .collect()
+}
+
+/// Render the running config: validate `entries`, expand selectors against `inventory` into the
+/// concrete realized assignments, then allocate wire identities fresh-first against `prev`. Must be
+/// fed the *previous* registry (from the prior running config), never a fresh one, or every host
+/// renumbers and the wire tags stop agreeing. `now` (unix seconds, the render node's clock) drives
+/// the retirement quarantine that gates id reclamation.
+pub fn render(
+    prev: &identity::Registry,
+    entries: HashMap<String, MicrosegEntry>,
+    inventory: &[GuestInfo],
+    now: u64,
+) -> anyhow::Result<MicrosegRunningConfig> {
+    validate(&entries)?;
+    let policies = build_policies(&entries)?;
+    let realized = realized_assignments(&entries, inventory);
+    let sets = realized_group_sets(&entries, &realized)?;
+    let identities = identity::allocate(prev, &sets, &policies, now)?;
+    Ok(MicrosegRunningConfig {
+        ids: entries,
+        identities,
+        realized,
+    })
+}
+
+/// A rule resolved for verdict folding: its predicates in group marks, plus its priority and
+/// verdict. Built once per apply from the running config.
+pub struct ResolvedRule {
+    pub src: identity::Predicate,
+    pub dst: identity::Predicate,
+    pub prio: u16,
+    pub allow: bool,
+}
+
+/// Resolve every rule's predicates to marks for folding.
+pub fn resolved_rules(
+    entries: &HashMap<String, MicrosegEntry>,
+) -> anyhow::Result<Vec<ResolvedRule>> {
+    let name_marks = group_marks(entries);
+    entries
+        .values()
+        .filter_map(|entry| match entry {
+            MicrosegEntry::Rule(rule) => Some(rule),
+            _ => None,
+        })
+        .map(|rule| {
+            Ok(ResolvedRule {
+                src: predicate(&name_marks, rule.src_match, &rule.src)?,
+                dst: predicate(&name_marks, rule.dst_match, &rule.dst)?,
+                prio: rule.prio,
+                allow: rule.allow,
+            })
+        })
+        .collect()
+}
+
+/// Fold the rules into the verdict for one `(src_class, dst_class)` cell, identified by the two
+/// representative sets. Among rules whose `src` predicate fires on `src_rep` and `dst` predicate
+/// fires on `dst_rep`, the highest priority wins. Verdicts tied at the top priority are ANDed, so
+/// a single deny overrides the allows. No matching rule means default-deny.
+pub fn verdict(
+    rules: &[ResolvedRule],
+    src_rep: &identity::GroupSet,
+    dst_rep: &identity::GroupSet,
+) -> bool {
+    let mut best: Option<(u16, bool)> = None;
+    for rule in rules {
+        if rule.src.fires(src_rep) && rule.dst.fires(dst_rep) {
+            best = Some(match best {
+                Some((prio, _)) if rule.prio > prio => (rule.prio, rule.allow),
+                Some((prio, allow)) if rule.prio == prio => (prio, allow && rule.allow),
+                Some(current) => current,
+                None => (rule.prio, rule.allow),
+            });
+        }
+    }
+    best.map_or(false, |(_, allow)| allow)
+}
+
+/// API helper types. The perl API passes its raw parameter hash here and perlmod deserializes it
+/// into [`MicrosegCreate`] or [`MicrosegUpdate`].
+pub mod api {
+    use std::collections::{HashMap, HashSet};
+
+    use anyhow::{Error, anyhow, bail};
+    use serde::Deserialize;
+
+    use super::{
+        GroupSection, GuestAssignmentSection, MicrosegEntry, MicrosegId,
+        PredicateMatch, RuleSection,
+    };
+
+    #[derive(Debug, Clone, Deserialize)]
+    #[serde(tag = "type", rename_all = "lowercase")]
+    pub enum MicrosegCreate {
+        Group(GroupCreate),
+        Rule(RuleCreate),
+        GuestAssignment(GuestAssignmentCreate),
+    }
+
+    #[derive(Debug, Clone, Deserialize)]
+    pub struct GroupCreate {
+        pub id: MicrosegId,
+        #[serde(default, deserialize_with = "proxmox_serde::perl::deserialize_u16")]
+        pub mark: Option<u16>,
+        #[serde(default)]
+        pub comment: Option<String>,
+    }
+
+    #[derive(Debug, Clone, Deserialize)]
+    pub struct RuleCreate {
+        #[serde(default)]
+        pub src: Vec<MicrosegId>,
+        #[serde(default)]
+        pub src_match: PredicateMatch,
+        #[serde(default)]
+        pub dst: Vec<MicrosegId>,
+        #[serde(default)]
+        pub dst_match: PredicateMatch,
+        #[serde(default, deserialize_with = "proxmox_serde::perl::deserialize_u16")]
+        pub prio: Option<u16>,
+        #[serde(deserialize_with = "proxmox_serde::perl::deserialize_bool")]
+        pub allow: bool,
+        #[serde(default)]
+        pub comment: Option<String>,
+    }
+
+    #[derive(Debug, Clone, Deserialize)]
+    pub struct GuestAssignmentCreate {
+        #[serde(deserialize_with = "proxmox_serde::perl::deserialize_u32")]
+        pub vmid: u32,
+        #[serde(default, deserialize_with = "proxmox_serde::perl::deserialize_u32")]
+        pub iface: Option<u32>,
+        #[serde(default)]
+        pub groups: Vec<MicrosegId>,
+        #[serde(default)]
+        pub comment: Option<String>,
+    }
+
+    #[derive(Debug, Clone, Deserialize)]
+    #[serde(tag = "type", rename_all = "lowercase")]
+    pub enum MicrosegUpdate {
+        Group(GroupUpdate),
+        Rule(RuleUpdate),
+        GuestAssignment(AssignmentUpdate),
+    }
+
+    #[derive(Debug, Clone, Default, Deserialize)]
+    pub struct GroupUpdate {
+        #[serde(default)]
+        pub comment: Option<String>,
+        #[serde(default)]
+        pub delete: Vec<GroupDeletableProperty>,
+    }
+
+    #[derive(Debug, Clone, Copy, Deserialize)]
+    #[serde(rename_all = "lowercase")]
+    pub enum GroupDeletableProperty {
+        Comment,
+    }
+
+    #[derive(Debug, Clone, Default, Deserialize)]
+    pub struct RuleUpdate {
+        #[serde(default, deserialize_with = "proxmox_serde::perl::deserialize_u16")]
+        pub prio: Option<u16>,
+        #[serde(default, deserialize_with = "proxmox_serde::perl::deserialize_bool")]
+        pub allow: Option<bool>,
+        #[serde(default)]
+        pub comment: Option<String>,
+        #[serde(default)]
+        pub delete: Vec<RuleDeletableProperty>,
+    }
+
+    #[derive(Debug, Clone, Copy, Deserialize)]
+    #[serde(rename_all = "lowercase")]
+    pub enum RuleDeletableProperty {
+        Comment,
+    }
+
+    #[derive(Debug, Clone, Default, Deserialize)]
+    pub struct AssignmentUpdate {
+        #[serde(default)]
+        pub groups: Option<Vec<MicrosegId>>,
+        #[serde(default)]
+        pub comment: Option<String>,
+        #[serde(default)]
+        pub delete: Vec<AssignmentDeletableProperty>,
+    }
+
+    #[derive(Debug, Clone, Copy, Deserialize)]
+    #[serde(rename_all = "lowercase")]
+    pub enum AssignmentDeletableProperty {
+        Comment,
+    }
+
+    /// Lowest unused group mark in `1..=65535`.
+    pub fn next_free_mark(entries: &HashMap<String, MicrosegEntry>) -> Option<u16> {
+        let used: HashSet<u16> = entries
+            .values()
+            .filter_map(|entry| match entry {
+                MicrosegEntry::Group(group) => Some(group.mark),
+                _ => None,
+            })
+            .collect();
+        (1..=u16::MAX).find(|mark| !used.contains(mark))
+    }
+
+    /// Resolve group names to their marks, sorted and deduped (the canonical set).
+    fn marks_of(
+        names: &[MicrosegId],
+        entries: &HashMap<String, MicrosegEntry>,
+    ) -> Result<Vec<u16>, Error> {
+        let mut marks: Vec<u16> = names
+            .iter()
+            .map(|name| {
+                if name.as_ref() == super::UNTAGGED_GROUP {
+                    return Ok(super::UNTAGGED_MARK);
+                }
+                match entries.get(name.as_ref()) {
+                    Some(MicrosegEntry::Group(group)) => Ok(group.mark),
+                    _ => Err(anyhow!("references group '{name}' which does not exist")),
+                }
+            })
+            .collect::<Result<_, _>>()?;
+        marks.sort_unstable();
+        marks.dedup();
+        Ok(marks)
+    }
+
+    /// A stable, content-derived rule id (FNV-1a over the canonical predicate pair), so identical
+    /// rules collapse to one entry and re-creating one is idempotent.
+    fn rule_id(
+        src_match: PredicateMatch,
+        src_marks: &[u16],
+        dst_match: PredicateMatch,
+        dst_marks: &[u16],
+    ) -> Result<MicrosegId, Error> {
+        let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
+        let mut feed = |byte: u8| {
+            hash ^= byte as u64;
+            hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
+        };
+        feed(src_match as u8);
+        for mark in src_marks {
+            feed((mark >> 8) as u8);
+            feed(*mark as u8);
+        }
+        feed(0xff);
+        feed(dst_match as u8);
+        for mark in dst_marks {
+            feed((mark >> 8) as u8);
+            feed(*mark as u8);
+        }
+        format!("r{hash:016x}").parse().map_err(Error::from)
+    }
+
+    /// Build an entry from a create payload, auto-assigning a free group mark when none is given and
+    /// deriving rule and assignment ids from their contents.
+    pub fn build_entry(
+        payload: MicrosegCreate,
+        entries: &HashMap<String, MicrosegEntry>,
+    ) -> Result<MicrosegEntry, Error> {
+        Ok(match payload {
+            MicrosegCreate::Group(group) => {
+                let mark = match group.mark {
+                    Some(mark) => mark,
+                    None => next_free_mark(entries)
+                        .ok_or_else(|| anyhow!("no free group mark available"))?,
+                };
+                MicrosegEntry::Group(GroupSection {
+                    id: group.id,
+                    mark,
+                    comment: group.comment,
+                })
+            }
+            MicrosegCreate::Rule(rule) => {
+                let src_marks = marks_of(&rule.src, entries).map_err(|e| anyhow!("rule {e}"))?;
+                let dst_marks = marks_of(&rule.dst, entries).map_err(|e| anyhow!("rule {e}"))?;
+                let id = rule_id(rule.src_match, &src_marks, rule.dst_match, &dst_marks)?;
+                MicrosegEntry::Rule(RuleSection {
+                    id,
+                    src: rule.src,
+                    src_match: rule.src_match,
+                    dst: rule.dst,
+                    dst_match: rule.dst_match,
+                    prio: rule.prio.unwrap_or(0),
+                    allow: rule.allow,
+                    comment: rule.comment,
+                })
+            }
+            MicrosegCreate::GuestAssignment(assignment) => {
+                let id = match assignment.iface {
+                    Some(iface) => format!("vm{}i{}", assignment.vmid, iface).parse()?,
+                    None => format!("vm{}", assignment.vmid).parse()?,
+                };
+                MicrosegEntry::GuestAssignment(GuestAssignmentSection {
+                    id,
+                    vmid: assignment.vmid,
+                    iface: assignment.iface,
+                    groups: assignment.groups,
+                    comment: assignment.comment,
+                })
+            }
+        })
+    }
+
+    /// Apply a partial update to an existing entry. The update is tagged by type and carries its
+    /// own property deletions, so it must match the existing object's type.
+    pub fn apply_update(entry: &mut MicrosegEntry, update: MicrosegUpdate) -> Result<(), Error> {
+        match (entry, update) {
+            (MicrosegEntry::Group(group), MicrosegUpdate::Group(update)) => {
+                if update.comment.is_some() {
+                    group.comment = update.comment;
+                }
+                for property in update.delete {
+                    match property {
+                        GroupDeletableProperty::Comment => group.comment = None,
+                    }
+                }
+            }
+            (MicrosegEntry::Rule(rule), MicrosegUpdate::Rule(update)) => {
+                if let Some(prio) = update.prio {
+                    rule.prio = prio;
+                }
+                if let Some(allow) = update.allow {
+                    rule.allow = allow;
+                }
+                if update.comment.is_some() {
+                    rule.comment = update.comment;
+                }
+                for property in update.delete {
+                    match property {
+                        RuleDeletableProperty::Comment => rule.comment = None,
+                    }
+                }
+            }
+            (
+                MicrosegEntry::GuestAssignment(assignment),
+                MicrosegUpdate::GuestAssignment(update),
+            ) => {
+                if let Some(groups) = update.groups {
+                    assignment.groups = groups;
+                }
+                if update.comment.is_some() {
+                    assignment.comment = update.comment;
+                }
+                for property in update.delete {
+                    match property {
+                        AssignmentDeletableProperty::Comment => assignment.comment = None,
+                    }
+                }
+            }
+            _ => bail!("update type does not match the existing object's type"),
+        }
+        Ok(())
+    }
+
+    /// Format a predicate (a match kind over a group list) as `any{web, db}`, for diagnostics.
+    fn predicate_str(match_kind: PredicateMatch, groups: &[MicrosegId]) -> String {
+        let kind = match match_kind {
+            PredicateMatch::Any => "any",
+            PredicateMatch::All => "all",
+            PredicateMatch::Exact => "exact",
+        };
+        let list = groups
+            .iter()
+            .map(|g| g.as_ref())
+            .collect::<Vec<_>>()
+            .join(", ");
+        format!("{kind}{{{list}}}")
+    }
+
+    /// A human-readable description of an entry, for diagnostics where its content-derived id is
+    /// meaningless. The object's comment, when set, is appended.
+    fn describe_referrer(entry: &MicrosegEntry) -> String {
+        let (desc, comment) = match entry {
+            MicrosegEntry::Rule(rule) => (
+                format!(
+                    "rule {} -> {}",
+                    predicate_str(rule.src_match, &rule.src),
+                    predicate_str(rule.dst_match, &rule.dst),
+                ),
+                rule.comment.as_deref(),
+            ),
+            MicrosegEntry::GuestAssignment(assignment) => (
+                match assignment.iface {
+                    Some(iface) => format!("guest assignment vm{} net{iface}", assignment.vmid),
+                    None => format!("guest assignment vm{}", assignment.vmid),
+                },
+                assignment.comment.as_deref(),
+            ),
+            MicrosegEntry::Group(_) => ("object".to_string(), None),
+        };
+        match comment {
+            Some(comment) => format!("{desc} (comment: \"{comment}\")"),
+            None => desc,
+        }
+    }
+
+    /// Human-readable descriptions of every rule or assignment that references group `name`, sorted
+    /// for a stable message, since the referrers' content-derived ids would be meaningless to a user.
+    pub fn group_referenced_by(
+        entries: &HashMap<String, MicrosegEntry>,
+        name: &str,
+    ) -> Vec<String> {
+        let mut referrers: Vec<String> = entries
+            .values()
+            .filter(|entry| match entry {
+                MicrosegEntry::Rule(rule) => {
+                    rule.src.iter().any(|g| g.as_ref() == name)
+                        || rule.dst.iter().any(|g| g.as_ref() == name)
+                }
+                MicrosegEntry::GuestAssignment(assignment) => {
+                    assignment.groups.iter().any(|g| g.as_ref() == name)
+                }
+                MicrosegEntry::Group(_) => false,
+            })
+            .map(describe_referrer)
+            .collect();
+        referrers.sort();
+        referrers
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use proxmox_section_config::typed::{ApiSectionDataEntry, SectionConfigData};
+
+    use super::*;
+
+    fn id(s: &str) -> MicrosegId {
+        s.parse().expect("valid microseg id")
+    }
+
+    fn ids(names: &[&str]) -> Vec<MicrosegId> {
+        names.iter().map(|n| id(n)).collect()
+    }
+
+    fn group(name: &str, mark: u16) -> (String, MicrosegEntry) {
+        (
+            name.to_string(),
+            MicrosegEntry::Group(GroupSection {
+                id: id(name),
+                mark,
+                comment: None,
+            }),
+        )
+    }
+
+    fn rule(name: &str, src: &[&str], dst: &[&str], allow: bool) -> (String, MicrosegEntry) {
+        (
+            name.to_string(),
+            MicrosegEntry::Rule(RuleSection {
+                id: id(name),
+                src: ids(src),
+                src_match: PredicateMatch::Any,
+                dst: ids(dst),
+                dst_match: PredicateMatch::Any,
+                prio: 0,
+                allow,
+                comment: None,
+            }),
+        )
+    }
+
+    #[test]
+    fn rules_by_identity_honours_match_kinds() {
+        use PredicateMatch::{All, Any, Exact};
+        let mk = |name: &str, sm, src: &[&str], dm, dst: &[&str]| {
+            (
+                name.to_string(),
+                MicrosegEntry::Rule(RuleSection {
+                    id: id(name),
+                    src: ids(src),
+                    src_match: sm,
+                    dst: ids(dst),
+                    dst_match: dm,
+                    prio: 0,
+                    allow: true,
+                    comment: None,
+                }),
+            )
+        };
+        let entries: HashMap<String, MicrosegEntry> = HashMap::from([
+            group("web", 2),
+            group("db", 3),
+            group("mon", 4),
+            mk("r_any", Any, &["web"], Any, &["db"]),
+            mk("r_all", All, &["web", "db"], Any, &["mon"]),
+            mk("r_exact", Exact, &["web"], Any, &["db"]),
+            assignment("vm100i0", 100, 0, &["web"]),
+            assignment("vm200i0", 200, 0, &["web", "db"]),
+        ]);
+
+        let inv = vec![guest(100, &[], &[0]), guest(200, &[], &[0])];
+        let by_class = rules_by_identity(&entries, &inv).expect("rules_by_identity");
+        let names = |v: &[MicrosegId]| {
+            let mut s: Vec<String> = v.iter().map(|i| i.to_string()).collect();
+            s.sort();
+            s
+        };
+        let find = |groups: &[&str]| {
+            let want: BTreeSet<MicrosegId> = ids(groups).into_iter().collect();
+            by_class
+                .iter()
+                .find(|c| c.groups.iter().cloned().collect::<BTreeSet<_>>() == want)
+                .unwrap_or_else(|| panic!("no class {groups:?}"))
+        };
+
+        // {web}: only the any and exact source predicates fire, nothing names it as a destination.
+        let web = find(&["web"]);
+        assert_eq!(names(&web.outbound), vec!["r_any", "r_exact"]);
+        assert!(web.inbound.is_empty());
+
+        // {web,db}: the all predicate now fires as source, db-destinations fire as inbound.
+        let webdb = find(&["web", "db"]);
+        assert_eq!(names(&webdb.outbound), vec!["r_all", "r_any"]);
+        assert_eq!(names(&webdb.inbound), vec!["r_any", "r_exact"]);
+    }
+
+    #[test]
+    fn section_config_round_trip() {
+        // Build entries, render to the section-config text, parse it back, and assert equality.
+        // This exercises the array (src/dst/groups) and enum (match) serialization without pinning
+        // the exact on-disk syntax.
+        let entries: HashMap<String, MicrosegEntry> = HashMap::from([
+            group("web", 5),
+            group("db", 7),
+            (
+                "r1".to_string(),
+                MicrosegEntry::Rule(RuleSection {
+                    id: id("r1"),
+                    src: ids(&["web"]),
+                    src_match: PredicateMatch::Any,
+                    dst: ids(&["db"]),
+                    dst_match: PredicateMatch::Exact,
+                    prio: 5,
+                    allow: true,
+                    comment: None,
+                }),
+            ),
+            (
+                "vm100i0".to_string(),
+                MicrosegEntry::GuestAssignment(GuestAssignmentSection {
+                    id: id("vm100i0"),
+                    vmid: 100,
+                    iface: Some(0),
+                    groups: ids(&["web", "db"]),
+                    comment: None,
+                }),
+            ),
+        ]);
+
+        validate(&entries).expect("config is valid");
+
+        let data: SectionConfigData<MicrosegEntry> = SectionConfigData::from_iter(entries.clone());
+        let raw = MicrosegEntry::write_section_config("microseg.cfg", &data).expect("write");
+        let parsed: HashMap<String, MicrosegEntry> =
+            MicrosegEntry::parse_section_config("microseg.cfg", &raw)
+                .expect("parse")
+                .into_iter()
+                .collect();
+
+        assert_eq!(parsed, entries, "round trip preserves every entry");
+    }
+
+    #[test]
+    fn agent_reads_numeric_allow_from_running_config() {
+        // perl to_json renders the rule `allow` flag as 0/1, so the agent's serde_json read of the
+        // raw running config must accept the numeric form (and still accept a real bool).
+        let json = r#"{"ids":{
+            "r1":{"type":"rule","id":"r1","src":["web"],"dst":["db"],"allow":1},
+            "r2":{"type":"rule","id":"r2","src":["web"],"dst":["db"],"src_match":"all","allow":0},
+            "r3":{"type":"rule","id":"r3","src":["web"],"dst":["db"],"allow":true},
+            "web":{"type":"group","id":"web","mark":5},
+            "db":{"type":"group","id":"db","mark":7}
+        }}"#;
+        let cfg: MicrosegRunningConfig = serde_json::from_str(json).expect("parse running config");
+        let by_id: HashMap<&str, &RuleSection> = cfg.rules().collect();
+        assert!(by_id["r1"].allow());
+        assert!(!by_id["r2"].allow());
+        assert!(by_id["r3"].allow());
+        // src_match defaults to `any` when absent, and parses the explicit form
+        assert_eq!(by_id["r1"].src_match(), PredicateMatch::Any);
+        assert_eq!(by_id["r2"].src_match(), PredicateMatch::All);
+    }
+
+    #[test]
+    fn create_accepts_perl_string_scalars() {
+        // perlmod hands scalars to serde as strings, so create payloads must accept "0"/"1" for the
+        // allow flag and string forms of numeric fields.
+        let rule: api::MicrosegCreate = serde_json::from_str(
+            r#"{"type":"rule","src":["web"],"src_match":"exact","dst":["db"],"allow":"0"}"#,
+        )
+        .expect("rule create");
+        match rule {
+            api::MicrosegCreate::Rule(rule) => {
+                assert!(!rule.allow);
+                assert_eq!(rule.src_match, PredicateMatch::Exact);
+            }
+            other => panic!("expected rule, got {other:?}"),
+        }
+
+        let assignment: api::MicrosegCreate = serde_json::from_str(
+            r#"{"type":"guestassignment","vmid":"100","iface":"0","groups":["web","db"]}"#,
+        )
+        .expect("assignment create");
+        match assignment {
+            api::MicrosegCreate::GuestAssignment(assignment) => {
+                assert_eq!(assignment.vmid, 100);
+                assert_eq!(assignment.groups.len(), 2);
+            }
+            other => panic!("expected guest assignment, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn validate_rejects_unknown_groups_empty_predicates_and_dup_marks() {
+        let base: HashMap<String, MicrosegEntry> = HashMap::from([group("web", 1), group("db", 2)]);
+
+        let mut ok = base.clone();
+        ok.extend([rule("ok", &["web"], &["db"], true)]);
+        validate(&ok).expect("valid rule");
+
+        let mut unknown = base.clone();
+        unknown.extend([rule("bad", &["web"], &["ghost"], true)]);
+        assert!(validate(&unknown).is_err(), "unknown dst group must fail");
+
+        let mut empty_src = base.clone();
+        empty_src.extend([rule("nosrc", &[], &["db"], true)]);
+        assert!(
+            validate(&empty_src).is_err(),
+            "empty source predicate must fail"
+        );
+
+        let dup_mark: HashMap<String, MicrosegEntry> =
+            HashMap::from([group("ga", 1), group("gb", 1)]);
+        assert!(validate(&dup_mark).is_err(), "duplicate mark must fail");
+    }
+
+    #[test]
+    fn build_entry_derives_stable_ids() {
+        let entries: HashMap<String, MicrosegEntry> =
+            HashMap::from([group("web", 11), group("db", 22)]);
+
+        let make_rule = || {
+            api::build_entry(
+                api::MicrosegCreate::Rule(api::RuleCreate {
+                    src: ids(&["web"]),
+                    src_match: PredicateMatch::Any,
+                    dst: ids(&["db"]),
+                    dst_match: PredicateMatch::Exact,
+                    prio: None,
+                    allow: true,
+                    comment: None,
+                }),
+                &entries,
+            )
+            .expect("build rule")
+        };
+        let r1 = make_rule();
+        let r2 = make_rule();
+        assert!(r1.id().as_ref().starts_with('r'));
+        assert_eq!(r1.id(), r2.id(), "same predicate pair derives the same id");
+
+        // a different predicate derives a different id
+        let other = api::build_entry(
+            api::MicrosegCreate::Rule(api::RuleCreate {
+                src: ids(&["db"]),
+                src_match: PredicateMatch::Any,
+                dst: ids(&["web"]),
+                dst_match: PredicateMatch::Exact,
+                prio: None,
+                allow: true,
+                comment: None,
+            }),
+            &entries,
+        )
+        .expect("build other rule");
+        assert_ne!(r1.id(), other.id());
+
+        let assignment = api::build_entry(
+            api::MicrosegCreate::GuestAssignment(api::GuestAssignmentCreate {
+                vmid: 100,
+                iface: Some(0),
+                groups: ids(&["web"]),
+                comment: None,
+            }),
+            &entries,
+        )
+        .expect("build assignment");
+        assert_eq!(assignment.id().to_string(), "vm100i0");
+    }
+
+    fn assignment(id_str: &str, vmid: u32, iface: u32, groups: &[&str]) -> (String, MicrosegEntry) {
+        (
+            id_str.to_string(),
+            MicrosegEntry::GuestAssignment(GuestAssignmentSection {
+                id: id(id_str),
+                vmid,
+                iface: Some(iface),
+                groups: ids(groups),
+                comment: None,
+            }),
+        )
+    }
+
+    fn guest(vmid: u32, _tag_names: &[&str], nics: &[u32]) -> GuestInfo {
+        GuestInfo {
+            vmid,
+            nics: nics.to_vec(),
+        }
+    }
+
+    #[test]
+    fn render_allocates_identities_and_round_trips() {
+        let entries: HashMap<String, MicrosegEntry> = HashMap::from([
+            group("web", 2),
+            group("db", 3),
+            rule("r1", &["web"], &["db"], true),
+            assignment("vm100i0", 100, 0, &["web"]),
+            assignment("vm200i0", 200, 0, &["db"]),
+        ]);
+
+        let inv = vec![guest(100, &[], &[0]), guest(200, &[], &[0])];
+        let running = render(&identity::Registry::new(), entries, &inv, 0).expect("render");
+        // {2} (web) matches the src predicate, {3} (db) the dst predicate -> two classes.
+        assert_eq!(running.identities().classes().len(), 2);
+        assert_eq!(running.identities().next(), 3);
+
+        // the running config (entries + registry) survives a JSON round trip
+        let json = serde_json::to_string(&running).expect("to json");
+        let back: MicrosegRunningConfig = serde_json::from_str(&json).expect("from json");
+        assert_eq!(back, running);
+    }
+
+    #[test]
+    fn unmatched_assignment_is_denied_not_treated_as_untagged() {
+        use PredicateMatch::Any;
+        // An allow for untagged -> web, and a NIC assigned only to `iso`, a group no rule names.
+        // The iso set matches no policy, it must not alias onto the untagged identity and inherit
+        // that allow.
+        let entries: HashMap<String, MicrosegEntry> = HashMap::from([
+            group("web", 2),
+            group("iso", 5),
+            rule_full("ru", &["untagged"], Any, &["web"], Any, 0, true),
+            assignment("vm100i0", 100, 0, &["iso"]),
+        ]);
+        let inv = vec![guest(100, &[], &[0])];
+        let running = render(&identity::Registry::new(), entries.clone(), &inv, 0).expect("render");
+        let policies = build_policies(&entries).expect("policies");
+
+        // the iso NIC resolves to a real id, not the untagged id that carries the untagged allows
+        let iso_id = running
+            .identities()
+            .id_of(&marks(&[5]), &policies)
+            .expect("the unmatched iso set still has a class");
+        assert_ne!(iso_id, identity::UNTAGGED_ID);
+
+        // iso is denied to web, while genuinely untagged traffic keeps its allow
+        let rules = resolved_rules(&entries).expect("resolve");
+        assert!(
+            !verdict(&rules, &marks(&[5]), &marks(&[2])),
+            "iso must not reach web"
+        );
+        assert!(
+            verdict(&rules, &marks(&[0]), &marks(&[2])),
+            "untagged still reaches web"
+        );
+    }
+
+    #[test]
+    fn render_is_add_only_across_reruns() {
+        let entries: HashMap<String, MicrosegEntry> = HashMap::from([
+            group("web", 2),
+            group("db", 3),
+            rule("r1", &["web"], &["db"], true),
+            assignment("vm100i0", 100, 0, &["web"]),
+        ]);
+        let inv = vec![guest(100, &[], &[0])];
+        let first = render(&identity::Registry::new(), entries, &inv, 0).expect("first render");
+        let second =
+            render(first.identities(), first.ids().clone(), &inv, 0).expect("second render");
+        assert_eq!(
+            second.identities(),
+            first.identities(),
+            "re-render keeps ids stable"
+        );
+    }
+
+    #[test]
+    fn guest_assignment_to_absent_nic_covers_nothing() {
+        let entries = HashMap::from([group("web", 2), assignment("vm100i3", 100, 3, &["web"])]);
+        // an explicit binding to net3 realizes nothing when the guest has only net0, so the
+        // realized set only ever names NICs that exist
+        let inv = vec![guest(100, &[], &[0])];
+        assert!(realized_assignments(&entries, &inv).is_empty());
+        assert!(realized_per_assignment(&entries, &inv)["vm100i3"].is_empty());
+    }
+
+    fn marks(xs: &[u16]) -> identity::GroupSet {
+        xs.iter().copied().collect()
+    }
+
+    fn rule_full(
+        name: &str,
+        src: &[&str],
+        src_match: PredicateMatch,
+        dst: &[&str],
+        dst_match: PredicateMatch,
+        prio: u16,
+        allow: bool,
+    ) -> (String, MicrosegEntry) {
+        (
+            name.to_string(),
+            MicrosegEntry::Rule(RuleSection {
+                id: id(name),
+                src: ids(src),
+                src_match,
+                dst: ids(dst),
+                dst_match,
+                prio,
+                allow,
+                comment: None,
+            }),
+        )
+    }
+
+    #[test]
+    fn fold_priority_overrides_and_ties_deny() {
+        use PredicateMatch::{Any, Exact};
+        // A broad deny over web/app -> db at prio 0, and a specific allow web -> db at prio 10.
+        let entries: HashMap<String, MicrosegEntry> = HashMap::from([
+            group("web", 2),
+            group("db", 3),
+            group("app", 4),
+            rule_full("broad", &["web", "app"], Any, &["db"], Any, 0, false),
+            rule_full("spec", &["web"], Exact, &["db"], Any, 10, true),
+        ]);
+        let rules = resolved_rules(&entries).expect("resolve");
+
+        // web -> db: both fire, the prio-10 allow wins over the prio-0 deny.
+        assert!(verdict(&rules, &marks(&[2]), &marks(&[3])));
+        // app -> db: only the broad deny fires.
+        assert!(!verdict(&rules, &marks(&[4]), &marks(&[3])));
+        // web -> web: no rule fires -> default deny.
+        assert!(!verdict(&rules, &marks(&[2]), &marks(&[2])));
+
+        // A deny overrides an allow at the same priority: two rules both fire on a web+app interface
+        // going to db at the same priority, one allow one deny.
+        let tied: HashMap<String, MicrosegEntry> = HashMap::from([
+            group("web", 2),
+            group("db", 3),
+            group("app", 4),
+            rule_full("ra", &["web"], Any, &["db"], Any, 5, true),
+            rule_full("rb", &["app"], Any, &["db"], Any, 5, false),
+        ]);
+        let tied_rules = resolved_rules(&tied).expect("resolve");
+        assert!(
+            !verdict(&tied_rules, &marks(&[2, 4]), &marks(&[3])),
+            "allow and deny tied at the top priority deny"
+        );
+        // an interface in only web, to db: only the allow fires.
+        assert!(verdict(&tied_rules, &marks(&[2]), &marks(&[3])));
+    }
+
+    #[test]
+    fn untagged_group_matches_the_untagged_id_and_composes() {
+        use PredicateMatch::Any;
+        let entries: HashMap<String, MicrosegEntry> = HashMap::from([
+            group("web", 2),
+            group("db", 3),
+            group("app", 4),
+            // any{untagged, web} -> db : the untagged source and web both reach db
+            rule_full("ru", &["untagged", "web"], Any, &["db"], Any, 0, true),
+        ]);
+        validate(&entries).expect("a predicate naming 'untagged' is valid");
+        let rules = resolved_rules(&entries).expect("resolve");
+
+        // the untagged identity is the set containing the reserved mark 0
+        assert!(
+            verdict(&rules, &marks(&[0]), &marks(&[3])),
+            "untagged source reaches db"
+        );
+        assert!(
+            verdict(&rules, &marks(&[2]), &marks(&[3])),
+            "web reaches db (same 'any')"
+        );
+        assert!(
+            !verdict(&rules, &marks(&[4]), &marks(&[3])),
+            "app is not in the source predicate"
+        );
+    }
+
+    #[test]
+    fn exact_untagged_matches_only_the_no_group_identity() {
+        use PredicateMatch::{Any, Exact};
+        let entries: HashMap<String, MicrosegEntry> = HashMap::from([
+            group("web", 2),
+            rule_full("ru", &["untagged"], Exact, &["web"], Any, 0, true),
+        ]);
+        let rules = resolved_rules(&entries).expect("resolve");
+        assert!(
+            verdict(&rules, &marks(&[0]), &marks(&[2])),
+            "untagged -> web allowed"
+        );
+        assert!(
+            !verdict(&rules, &marks(&[2]), &marks(&[2])),
+            "web -> web denied"
+        );
+        assert!(
+            !verdict(&rules, &marks(&[0, 2]), &marks(&[2])),
+            "exact untagged is strict"
+        );
+    }
+
+    #[test]
+    fn validate_reserves_the_untagged_group() {
+        // a real group cannot take the reserved name
+        let reserved: HashMap<String, MicrosegEntry> = HashMap::from([group("untagged", 5)]);
+        assert!(validate(&reserved).is_err(), "'untagged' is reserved");
+
+        // an assignment cannot use the untagged group
+        let bad_assign: HashMap<String, MicrosegEntry> =
+            HashMap::from([group("web", 2), assignment("vm1i0", 1, 0, &["untagged"])]);
+        assert!(
+            validate(&bad_assign).is_err(),
+            "assignments cannot use 'untagged'"
+        );
+
+        // a rule predicate may name it
+        let ok: HashMap<String, MicrosegEntry> = HashMap::from([
+            group("web", 2),
+            rule_full(
+                "ru",
+                &["untagged"],
+                PredicateMatch::Any,
+                &["web"],
+                PredicateMatch::Any,
+                0,
+                true,
+            ),
+        ]);
+        validate(&ok).expect("a rule may name 'untagged'");
+    }
+
+    #[test]
+    fn guest_info_reads_perl_string_scalar_nics() {
+        // perlmod hands array elements as untyped (string) scalars, the nics deserializer must read
+        // those, and real JSON numbers, into a Vec<u32>.
+        let from_strings: GuestInfo =
+            serde_json::from_str(r#"{"vmid":"100","tags":["db"],"nics":["0","1","2"]}"#)
+                .expect("parse string-scalar guest info");
+        assert_eq!(from_strings.vmid, 100);
+        assert_eq!(from_strings.nics, vec![0, 1, 2]);
+
+        let from_numbers: GuestInfo =
+            serde_json::from_str(r#"{"vmid":100,"nics":[3,4]}"#).expect("parse numeric guest info");
+        assert_eq!(from_numbers.nics, vec![3, 4]);
+    }
+}
-- 
2.47.3