[PATCH datacenter-manager v3 03/12] subscription: pool: add data model and config layer

[Thomas Lamprecht <t.lamprecht@proxmox.com>]

Introduce the on-disk data model and locked config helpers that the
following commits build on, mirroring the pdm-config::remotes
pattern. The shadow file holds the signed SubscriptionInfo blob a
future shop-bundle import will provide, kept apart from the main
config so the bare keys list stays human-readable.

The source field is an enum so other origins (shop-bundle import,
remote adoption) can be added later without a wire-format break.

Signed-off-by: Thomas Lamprecht <t.lamprecht@proxmox.com>
---

Changes v2 -> 3:
* Move the on-disk files into a dedicated subscriptions/ subdir
  (keys.cfg + keys.shadow).

 lib/pdm-api-types/Cargo.toml           |   1 +
 lib/pdm-api-types/src/subscription.rs  | 399 ++++++++++++++++++++++++-
 lib/pdm-api-types/tests/test_import.rs | 338 +++++++++++++++++++++
 lib/pdm-config/src/lib.rs              |   1 +
 lib/pdm-config/src/setup.rs            |   7 +
 lib/pdm-config/src/subscriptions.rs    | 116 +++++++
 6 files changed, 861 insertions(+), 1 deletion(-)
 create mode 100644 lib/pdm-api-types/tests/test_import.rs
 create mode 100644 lib/pdm-config/src/subscriptions.rs

diff --git a/lib/pdm-api-types/Cargo.toml b/lib/pdm-api-types/Cargo.toml
index cb8b5054..f9e3d07e 100644
--- a/lib/pdm-api-types/Cargo.toml
+++ b/lib/pdm-api-types/Cargo.toml
@@ -15,6 +15,7 @@ serde_plain.workspace = true
 serde_json.workspace = true
 
 proxmox-acme-api.workspace = true
+proxmox-base64.workspace = true
 proxmox-access-control = { workspace = true, features = ["acl"] }
 proxmox-auth-api = { workspace = true, features = ["api-types"] }
 proxmox-apt-api-types.workspace = true
diff --git a/lib/pdm-api-types/src/subscription.rs b/lib/pdm-api-types/src/subscription.rs
index f0eb525b..811bce4c 100644
--- a/lib/pdm-api-types/src/subscription.rs
+++ b/lib/pdm-api-types/src/subscription.rs
@@ -1,11 +1,16 @@
+use std::sync::OnceLock;
 use std::{collections::HashMap, str::FromStr};
 
 use anyhow::Error;
 use serde::{Deserialize, Serialize};
 
-use proxmox_schema::api;
+use proxmox_schema::{api, const_regex, ApiStringFormat, ApiType, Schema, StringSchema};
+use proxmox_section_config::typed::ApiSectionDataEntry;
+use proxmox_section_config::{SectionConfig, SectionConfigPlugin};
 use proxmox_subscription::{SubscriptionInfo, SubscriptionStatus};
 
+use crate::remotes::RemoteType;
+
 #[api]
 // order is important here, since we use that for determining if a node has a valid subscription
 #[derive(Default, Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
@@ -174,3 +179,395 @@ pub struct PdmSubscriptionInfo {
     /// PDM subscription statistics
     pub statistics: SubscriptionStatistics,
 }
+
+const_regex! {
+    /// Subscription key pattern, restricted to the products PDM can drive.
+    ///
+    /// All keys follow `<prefix>-<10 hex>`. PVE encodes the maximum CPU socket count between
+    /// the product letters and the level letter, for example `pve4b-1234567890`. PBS has no
+    /// socket count, so its keys look like `pbsc-1234567890`. Level letters are c/b/s/p
+    /// (Community/Basic/Standard/Premium).
+    ///
+    /// PMG and POM keys are not accepted yet: PDM has no remote-side handler for them. Widen
+    /// this regex and `ProductType::from_key` in lockstep when PDM grows support for them.
+    pub PRODUCT_KEY_REGEX = r"^(?:pve[0-9]+|pbs)[cbsp]-[0-9a-f]{10}$";
+}
+
+pub const PRODUCT_KEY_FORMAT: ApiStringFormat = ApiStringFormat::Pattern(&PRODUCT_KEY_REGEX);
+
+pub const SUBSCRIPTION_KEY_SCHEMA: Schema = StringSchema::new("Subscription key.")
+    .format(&PRODUCT_KEY_FORMAT)
+    .min_length(15)
+    .max_length(18)
+    .schema();
+
+#[api]
+#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Hash, Deserialize, Serialize)]
+#[serde(rename_all = "lowercase")]
+/// Proxmox product line a subscription key belongs to.
+pub enum ProductType {
+    /// Proxmox Virtual Environment (PVE).
+    #[default]
+    Pve,
+    /// Proxmox Backup Server (PBS).
+    Pbs,
+    /// Proxmox Mail Gateway (PMG).
+    Pmg,
+    /// Proxmox Offline Mirror (POM).
+    Pom,
+}
+
+impl ProductType {
+    /// Static string used as the section-config type marker on disk.
+    pub const fn as_section_type(self) -> &'static str {
+        match self {
+            ProductType::Pve => "pve",
+            ProductType::Pbs => "pbs",
+            ProductType::Pmg => "pmg",
+            ProductType::Pom => "pom",
+        }
+    }
+
+    /// Classify a key by its prefix.
+    ///
+    /// Returns None when the prefix does not match any product PDM currently knows about;
+    /// callers should log that case so a new product line gets noticed instead of silently
+    /// sorted into a default bucket.
+    pub fn from_key(key: &str) -> Option<Self> {
+        let (prefix, _) = key.split_once('-')?;
+        if prefix.starts_with("pve") {
+            Some(ProductType::Pve)
+        } else if prefix.starts_with("pbs") {
+            Some(ProductType::Pbs)
+        } else if prefix.starts_with("pmg") {
+            Some(ProductType::Pmg)
+        } else if prefix.starts_with("pom") {
+            Some(ProductType::Pom)
+        } else {
+            None
+        }
+    }
+
+    /// Whether PDM currently knows how to drive a remote of this product type.
+    ///
+    /// PDM only manages PVE and PBS remotes today, and the schema regex rejects everything else
+    /// at insert time. This method covers in-memory paths for forward-compat, for example
+    /// existing pool entries loaded after the regex is widened in a future release.
+    pub fn matches_remote_type(self, remote_type: RemoteType) -> bool {
+        matches!(
+            (self, remote_type),
+            (ProductType::Pve, RemoteType::Pve) | (ProductType::Pbs, RemoteType::Pbs)
+        )
+    }
+}
+
+impl std::fmt::Display for ProductType {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(self.as_section_type())
+    }
+}
+
+/// Extract the socket count a PVE key covers (for example, 4 from "pve4b-...").
+///
+/// Returns None for non-PVE keys or unparseable prefixes.
+pub fn socket_count_from_key(key: &str) -> Option<u32> {
+    let (prefix, _) = key.split_once('-')?;
+    if !prefix.starts_with("pve") {
+        return None;
+    }
+    let after_pve = &prefix[3..];
+    let digits: String = after_pve
+        .chars()
+        .take_while(|c| c.is_ascii_digit())
+        .collect();
+    digits.parse().ok()
+}
+
+/// Pick the candidate PVE key with the smallest socket count that still covers `node_sockets`.
+///
+/// `candidates` yields `(id, key_string)` pairs. Keys without a parseable PVE socket count are
+/// skipped, and keys covering fewer sockets than the node needs are filtered out. Returns the
+/// id of the best fit, or None when no candidate covers the node.
+pub fn pick_best_pve_socket_key<'a, I, K>(node_sockets: u32, candidates: I) -> Option<K>
+where
+    I: IntoIterator<Item = (K, &'a str)>,
+{
+    candidates
+        .into_iter()
+        .filter_map(|(id, key)| socket_count_from_key(key).map(|s| (id, s)))
+        .filter(|(_, s)| *s >= node_sockets)
+        .min_by_key(|(_, s)| *s)
+        .map(|(id, _)| id)
+}
+
+#[api]
+#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Deserialize, Serialize)]
+#[serde(rename_all = "kebab-case")]
+/// Origin of a subscription key entry.
+pub enum SubscriptionKeySource {
+    /// Hand-entered into the pool by an admin. Used for any key added through the manual-entry
+    /// UI or CLI, and as the `serde(default)` for entries that predate this field.
+    #[default]
+    Manual,
+}
+
+#[api(
+    properties: {
+        "key": { schema: SUBSCRIPTION_KEY_SCHEMA },
+        "level": { optional: true },
+        "status": { optional: true },
+        "source": { optional: true },
+        "pending-clear": { optional: true },
+    },
+)]
+#[derive(Clone, Debug, Default, Deserialize, Serialize, PartialEq)]
+#[serde(rename_all = "kebab-case")]
+/// An entry in the subscription key pool.
+pub struct SubscriptionKeyEntry {
+    /// The subscription key (for example, pve4b-1234567890).
+    pub key: String,
+
+    /// Product type derived from the key prefix.
+    #[serde(rename = "product-type")]
+    pub product_type: ProductType,
+
+    /// Subscription level, derived from the key suffix.
+    #[serde(default)]
+    pub level: SubscriptionLevel,
+
+    /// Where the key entry came from. Defaults to manual entry.
+    #[serde(default)]
+    pub source: SubscriptionKeySource,
+
+    /// Remote this key is assigned to (if any).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub remote: Option<String>,
+
+    /// Node within the remote this key is assigned to (if any).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub node: Option<String>,
+
+    /// True when the operator queued a clear for this entry's bound node, that is, a request
+    /// to free the key from `remote`/`node` so it can be reassigned to a different node.
+    ///
+    /// Apply Pending issues a DELETE on the remote and then clears `remote`/`node` on success.
+    /// Clear Pending only resets this flag and leaves the binding untouched so the operator can
+    /// retry. A bare flag is enough since the (remote, node) binding lives next to it.
+    ///
+    /// Omitted from the serialised representation when false so the on-disk section and the
+    /// API response do not carry `pending-clear false` lines for every entry.
+    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
+    pub pending_clear: bool,
+
+    /// Server ID this key is bound to (from signed info, if available).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub serverid: Option<String>,
+
+    /// Subscription status from last check.
+    #[serde(default)]
+    pub status: SubscriptionStatus,
+
+    /// Next due date.
+    ///
+    /// Accepts the upstream `nextduedate` spelling on deserialisation so a future shop-bundle
+    /// import path can hand a raw `SubscriptionInfo` blob through without a field-name
+    /// translation step; canonical (and on-disk) form is `next-due-date` per the struct's
+    /// kebab-case rename.
+    #[serde(alias = "nextduedate", skip_serializing_if = "Option::is_none")]
+    pub next_due_date: Option<String>,
+
+    /// Product name.
+    ///
+    /// Accepts the upstream `productname` spelling on deserialisation; canonical form is
+    /// `product-name` to stay self-consistent with the sibling `product-type` field.
+    #[serde(alias = "productname", skip_serializing_if = "Option::is_none")]
+    pub product_name: Option<String>,
+
+    /// Epoch of last import or refresh of this key's data.
+    ///
+    /// Accepts the upstream `checktime` spelling on deserialisation; canonical form is
+    /// `check-time`.
+    #[serde(alias = "checktime", skip_serializing_if = "Option::is_none")]
+    pub check_time: Option<i64>,
+}
+
+impl ApiSectionDataEntry for SubscriptionKeyEntry {
+    const INTERNALLY_TAGGED: Option<&'static str> = Some("product-type");
+    const SECION_CONFIG_USES_TYPE_KEY: bool = true;
+
+    fn section_config() -> &'static SectionConfig {
+        static CONFIG: OnceLock<SectionConfig> = OnceLock::new();
+
+        CONFIG.get_or_init(|| {
+            let mut this =
+                SectionConfig::new(&SUBSCRIPTION_KEY_SCHEMA).with_type_key("product-type");
+            for ty in [
+                ProductType::Pve,
+                ProductType::Pbs,
+                ProductType::Pmg,
+                ProductType::Pom,
+            ] {
+                this.register_plugin(SectionConfigPlugin::new(
+                    ty.as_section_type().to_string(),
+                    Some("key".to_string()),
+                    SubscriptionKeyEntry::API_SCHEMA.unwrap_object_schema(),
+                ));
+            }
+            this
+        })
+    }
+
+    fn section_type(&self) -> &'static str {
+        self.product_type.as_section_type()
+    }
+}
+
+#[api(
+    properties: {
+        "key": { schema: SUBSCRIPTION_KEY_SCHEMA },
+    },
+)]
+#[derive(Clone, Debug, Default, Deserialize, Serialize)]
+#[serde(rename_all = "kebab-case")]
+/// Shadow entry storing the signed subscription info blob for a key.
+///
+/// Currently only populated by the future shop-bundle import flow; manually-added keys leave
+/// this table empty. The data layer is in place so that adding the import path later does not
+/// require reshaping the on-disk config.
+pub struct SubscriptionKeyShadow {
+    /// The subscription key.
+    pub key: String,
+
+    /// Product type (section type marker).
+    #[serde(rename = "product-type")]
+    pub product_type: ProductType,
+
+    /// Base64-encoded signed SubscriptionInfo JSON.
+    #[serde(default)]
+    pub info: String,
+}
+
+impl ApiSectionDataEntry for SubscriptionKeyShadow {
+    const INTERNALLY_TAGGED: Option<&'static str> = Some("product-type");
+    const SECION_CONFIG_USES_TYPE_KEY: bool = true;
+
+    fn section_config() -> &'static SectionConfig {
+        static CONFIG: OnceLock<SectionConfig> = OnceLock::new();
+
+        CONFIG.get_or_init(|| {
+            let mut this =
+                SectionConfig::new(&SUBSCRIPTION_KEY_SCHEMA).with_type_key("product-type");
+            for ty in [
+                ProductType::Pve,
+                ProductType::Pbs,
+                ProductType::Pmg,
+                ProductType::Pom,
+            ] {
+                this.register_plugin(SectionConfigPlugin::new(
+                    ty.as_section_type().to_string(),
+                    Some("key".to_string()),
+                    SubscriptionKeyShadow::API_SCHEMA.unwrap_object_schema(),
+                ));
+            }
+            this
+        })
+    }
+
+    fn section_type(&self) -> &'static str {
+        self.product_type.as_section_type()
+    }
+}
+
+/// Decode a base64-encoded `SubscriptionInfo` JSON blob from the shadow file.
+///
+/// Forward-compat helper for the future shop-bundle import path. Returns the parsed
+/// `SubscriptionInfo`; the caller is responsible for verifying the signature against the shop's
+/// signing key.
+pub fn parse_signed_info_blob(b64: &str) -> Result<SubscriptionInfo, Error> {
+    let bytes = proxmox_base64::decode(b64)?;
+    let info = serde_json::from_slice(&bytes)?;
+    Ok(info)
+}
+
+/// Cross-check the `serverid` of a shadowed entry against what the remote reports.
+///
+/// Forward-compat helper for the future bundle-import and push flow: when the shadow has a
+/// signed serverid binding, the operator should be warned if the remote it is being pushed to
+/// has a different hardware id. Returns Ok(None) when there is nothing to compare.
+pub fn verify_serverid(
+    entry: &SubscriptionKeyEntry,
+    remote_info: &SubscriptionInfo,
+) -> Result<Option<ServeridMismatch>, Error> {
+    let Some(expected) = entry.serverid.as_deref() else {
+        return Ok(None);
+    };
+    let Some(actual) = remote_info.serverid.as_deref() else {
+        return Ok(None);
+    };
+    if expected == actual {
+        Ok(None)
+    } else {
+        Ok(Some(ServeridMismatch {
+            key: entry.key.clone(),
+            expected: expected.to_string(),
+            actual: actual.to_string(),
+        }))
+    }
+}
+
+/// Result of [`verify_serverid`] when the bound and observed server-ids disagree.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct ServeridMismatch {
+    pub key: String,
+    pub expected: String,
+    pub actual: String,
+}
+
+#[api]
+#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
+#[serde(rename_all = "kebab-case")]
+/// Subscription status of a single remote node, combining remote query data with key pool
+/// assignment information.
+pub struct RemoteNodeStatus {
+    /// Remote name.
+    pub remote: String,
+    /// Remote type (pve or pbs).
+    #[serde(rename = "type")]
+    pub ty: RemoteType,
+    /// Node name.
+    pub node: String,
+    /// Number of CPU sockets (PVE only).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub sockets: Option<i64>,
+    /// Current subscription status.
+    #[serde(default)]
+    pub status: SubscriptionStatus,
+    /// Subscription level.
+    #[serde(default)]
+    pub level: SubscriptionLevel,
+    /// Currently assigned key from the pool (if any).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub assigned_key: Option<String>,
+    /// Current key on the node (from remote query).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub current_key: Option<String>,
+}
+
+#[api]
+#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "kebab-case")]
+/// A proposed key-to-node assignment from the auto-assign algorithm.
+pub struct ProposedAssignment {
+    /// The subscription key to assign.
+    pub key: String,
+    /// Target remote.
+    pub remote: String,
+    /// Target node.
+    pub node: String,
+    /// Socket count of the key (PVE only).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub key_sockets: Option<u32>,
+    /// Socket count of the node (PVE only).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub node_sockets: Option<i64>,
+}
diff --git a/lib/pdm-api-types/tests/test_import.rs b/lib/pdm-api-types/tests/test_import.rs
new file mode 100644
index 00000000..33601620
--- /dev/null
+++ b/lib/pdm-api-types/tests/test_import.rs
@@ -0,0 +1,338 @@
+//! SectionConfig round-trip and helper tests for the subscription key pool.
+//!
+//! Run with: cargo test -p pdm-api-types --test test_import
+
+use pdm_api_types::subscription::*;
+use proxmox_section_config::typed::{ApiSectionDataEntry, SectionConfigData};
+use proxmox_subscription::SubscriptionStatus;
+
+#[test]
+fn entry_roundtrip() {
+    let mut config = SectionConfigData::<SubscriptionKeyEntry>::default();
+
+    let entry = SubscriptionKeyEntry {
+        key: "pve4b-aa11bb2233".to_string(),
+        product_type: ProductType::Pve,
+        level: SubscriptionLevel::Basic,
+        source: SubscriptionKeySource::Manual,
+        remote: Some("my-cluster".to_string()),
+        node: Some("node1".to_string()),
+        pending_clear: false,
+        serverid: Some("AABBCCDD".to_string()),
+        status: SubscriptionStatus::Active,
+        next_due_date: Some("2027-06-01".to_string()),
+        product_name: Some("Proxmox VE Basic".to_string()),
+        check_time: Some(1700000000),
+    };
+
+    config.insert("pve4b-aa11bb2233".to_string(), entry);
+
+    let raw = SubscriptionKeyEntry::write_section_config("test", &config).expect("write failed");
+    let parsed = SubscriptionKeyEntry::parse_section_config("test", &raw).expect("parse failed");
+
+    let back = parsed.get("pve4b-aa11bb2233").expect("key not found");
+    assert_eq!(back.key, "pve4b-aa11bb2233");
+    assert_eq!(back.product_type, ProductType::Pve);
+    assert_eq!(back.source, SubscriptionKeySource::Manual);
+    assert_eq!(back.remote.as_deref(), Some("my-cluster"));
+    assert_eq!(back.node.as_deref(), Some("node1"));
+    assert_eq!(back.status, SubscriptionStatus::Active);
+    assert_eq!(back.next_due_date.as_deref(), Some("2027-06-01"));
+}
+
+#[test]
+fn shadow_roundtrip() {
+    let mut shadow = SectionConfigData::<SubscriptionKeyShadow>::default();
+
+    shadow.insert(
+        "pve4b-aa11bb2233".to_string(),
+        SubscriptionKeyShadow {
+            key: "pve4b-aa11bb2233".to_string(),
+            product_type: ProductType::Pve,
+            info: "dGVzdA==".to_string(),
+        },
+    );
+
+    let raw = SubscriptionKeyShadow::write_section_config("test", &shadow).expect("write failed");
+    let parsed = SubscriptionKeyShadow::parse_section_config("test", &raw).expect("parse failed");
+
+    let back = parsed.get("pve4b-aa11bb2233").expect("key not found");
+    assert_eq!(back.info, "dGVzdA==");
+}
+
+#[test]
+fn deserialize_api_response_json() {
+    // The legacy `nextduedate` / `productname` / `checktime` spellings are the shop's wire
+    // format (mirrored from `proxmox_subscription::SubscriptionInfo`); a future shop-bundle
+    // import path will feed exactly these into the pool. Keep the alias coverage explicit so a
+    // serde rename without an accompanying alias gets caught at test time.
+    let json = serde_json::json!({
+        "key": "pve4b-aa11bb2233",
+        "nextduedate": "2027-06-01",
+        "product-type": "pve",
+        "productname": "Proxmox VE Basic",
+        "checktime": 1700000000,
+        "serverid": "AABBCCDD",
+        "status": "active"
+    });
+
+    let entry: SubscriptionKeyEntry = serde_json::from_value(json).unwrap();
+    assert_eq!(entry.key, "pve4b-aa11bb2233");
+    assert_eq!(entry.product_type, ProductType::Pve);
+    assert_eq!(entry.status, SubscriptionStatus::Active);
+    assert_eq!(entry.source, SubscriptionKeySource::Manual);
+    assert_eq!(entry.next_due_date.as_deref(), Some("2027-06-01"));
+    assert_eq!(entry.product_name.as_deref(), Some("Proxmox VE Basic"));
+    assert_eq!(entry.check_time, Some(1700000000));
+}
+
+#[test]
+fn deserialize_canonical_kebab_case_json() {
+    // The canonical wire form for these fields uses the struct's `kebab-case` rename; verify
+    // the renamed spelling round-trips through serde even though the field shapes share the
+    // alias with the legacy form above.
+    let json = serde_json::json!({
+        "key": "pve4b-aa11bb2233",
+        "next-due-date": "2027-06-01",
+        "product-type": "pve",
+        "product-name": "Proxmox VE Basic",
+        "check-time": 1700000000,
+        "status": "active"
+    });
+
+    let entry: SubscriptionKeyEntry = serde_json::from_value(json).unwrap();
+    assert_eq!(entry.next_due_date.as_deref(), Some("2027-06-01"));
+    assert_eq!(entry.product_name.as_deref(), Some("Proxmox VE Basic"));
+    assert_eq!(entry.check_time, Some(1700000000));
+}
+
+#[test]
+fn deserialize_without_optional_fields() {
+    let json = serde_json::json!({
+        "key": "pbsb-ee77ff8899",
+        "product-type": "pbs",
+    });
+
+    let entry: SubscriptionKeyEntry = serde_json::from_value(json).unwrap();
+    assert_eq!(entry.key, "pbsb-ee77ff8899");
+    assert_eq!(entry.product_type, ProductType::Pbs);
+    assert!(entry.remote.is_none());
+    assert!(entry.next_due_date.is_none());
+}
+
+#[test]
+fn product_type_classification() {
+    let cases = [
+        ("pve4b-1234567890", Some(ProductType::Pve), "pve"),
+        ("pbss-abcdef0123", Some(ProductType::Pbs), "pbs"),
+        ("pmgb-1234567890", Some(ProductType::Pmg), "pmg"),
+        ("pomb-1234567890", Some(ProductType::Pom), "pom"),
+        ("xxx-1234567890", None, ""),
+        ("no-dash", None, ""),
+    ];
+    for (key, expected, marker) in cases {
+        assert_eq!(ProductType::from_key(key), expected, "from_key({key})");
+        if let Some(pt) = expected {
+            assert_eq!(pt.as_section_type(), marker, "section_type for {key}");
+        }
+    }
+}
+
+#[test]
+fn socket_count_extraction() {
+    assert_eq!(socket_count_from_key("pve1c-1234567890"), Some(1));
+    assert_eq!(socket_count_from_key("pve2b-1234567890"), Some(2));
+    assert_eq!(socket_count_from_key("pve4s-1234567890"), Some(4));
+    assert_eq!(socket_count_from_key("pve8p-1234567890"), Some(8));
+    assert_eq!(socket_count_from_key("pbss-1234567890"), None);
+    assert_eq!(socket_count_from_key("pvexb-1234567890"), None);
+}
+
+#[test]
+fn remote_type_matching() {
+    use pdm_api_types::remotes::RemoteType;
+
+    assert!(ProductType::Pve.matches_remote_type(RemoteType::Pve));
+    assert!(!ProductType::Pve.matches_remote_type(RemoteType::Pbs));
+    assert!(ProductType::Pbs.matches_remote_type(RemoteType::Pbs));
+    assert!(!ProductType::Pbs.matches_remote_type(RemoteType::Pve));
+    // PMG and POM are reserved product types but PDM cannot manage those remotes yet.
+    assert!(!ProductType::Pmg.matches_remote_type(RemoteType::Pve));
+    assert!(!ProductType::Pmg.matches_remote_type(RemoteType::Pbs));
+    assert!(!ProductType::Pom.matches_remote_type(RemoteType::Pbs));
+}
+
+#[test]
+fn subscription_level_from_key_suffix() {
+    assert_eq!(
+        SubscriptionLevel::from_key(Some("pve4c-123")),
+        SubscriptionLevel::Community
+    );
+    assert_eq!(
+        SubscriptionLevel::from_key(Some("pve4b-123")),
+        SubscriptionLevel::Basic
+    );
+    assert_eq!(
+        SubscriptionLevel::from_key(Some("pve4s-123")),
+        SubscriptionLevel::Standard
+    );
+    assert_eq!(
+        SubscriptionLevel::from_key(Some("pve2p-123")),
+        SubscriptionLevel::Premium
+    );
+    assert_eq!(
+        SubscriptionLevel::from_key(Some("pbsb-123")),
+        SubscriptionLevel::Basic
+    );
+    assert_eq!(SubscriptionLevel::from_key(None), SubscriptionLevel::None);
+    assert_eq!(
+        SubscriptionLevel::from_key(Some("")),
+        SubscriptionLevel::None
+    );
+}
+
+#[test]
+fn subscription_level_display_fromstr_roundtrip() {
+    for level in [
+        SubscriptionLevel::None,
+        SubscriptionLevel::Community,
+        SubscriptionLevel::Basic,
+        SubscriptionLevel::Standard,
+        SubscriptionLevel::Premium,
+        SubscriptionLevel::Unknown,
+    ] {
+        let s = format!("{level}");
+        let parsed: SubscriptionLevel = s.parse().unwrap();
+        assert_eq!(parsed, level, "roundtrip failed for {s}");
+    }
+
+    // Backward compatibility: legacy single-letter wire format still parses.
+    for (letter, level) in [
+        ("c", SubscriptionLevel::Community),
+        ("b", SubscriptionLevel::Basic),
+        ("s", SubscriptionLevel::Standard),
+        ("p", SubscriptionLevel::Premium),
+    ] {
+        assert_eq!(letter.parse::<SubscriptionLevel>().unwrap(), level);
+    }
+}
+
+#[test]
+fn multiple_keys_different_types() {
+    let mut config = SectionConfigData::<SubscriptionKeyEntry>::default();
+
+    config.insert(
+        "pve4b-aaaa111111".to_string(),
+        SubscriptionKeyEntry {
+            key: "pve4b-aaaa111111".to_string(),
+            product_type: ProductType::Pve,
+            status: SubscriptionStatus::Active,
+            ..Default::default()
+        },
+    );
+    config.insert(
+        "pbss-bbbb222222".to_string(),
+        SubscriptionKeyEntry {
+            key: "pbss-bbbb222222".to_string(),
+            product_type: ProductType::Pbs,
+            status: SubscriptionStatus::Active,
+            ..Default::default()
+        },
+    );
+
+    let raw = SubscriptionKeyEntry::write_section_config("test", &config).unwrap();
+    let parsed = SubscriptionKeyEntry::parse_section_config("test", &raw).unwrap();
+
+    assert_eq!(
+        parsed.get("pve4b-aaaa111111").unwrap().product_type,
+        ProductType::Pve
+    );
+    assert_eq!(
+        parsed.get("pbss-bbbb222222").unwrap().product_type,
+        ProductType::Pbs
+    );
+}
+
+#[test]
+fn pick_best_pve_socket_key_edge_cases() {
+    let pool = [
+        ("pve1c-aaa", "pve1c-aaa"),
+        ("pve2b-bbb", "pve2b-bbb"),
+        ("pve4s-ccc", "pve4s-ccc"),
+        ("pve8p-ddd", "pve8p-ddd"),
+    ];
+    let pick =
+        |sockets: u32| pick_best_pve_socket_key(sockets, pool.iter().map(|(id, k)| (*id, *k)));
+
+    // Exact match prefers the equally-sized key over a larger one.
+    assert_eq!(pick(2), Some("pve2b-bbb"));
+
+    // No exact match: fall through to the smallest key that still covers the node.
+    assert_eq!(pick(3), Some("pve4s-ccc"));
+    assert_eq!(pick(5), Some("pve8p-ddd"));
+
+    // Single-socket node still picks the single-socket key (does not overprovision).
+    assert_eq!(pick(1), Some("pve1c-aaa"));
+
+    // Node larger than every key has no fit.
+    assert_eq!(pick(16), None);
+
+    // Empty candidate list is None.
+    let empty: [(&str, &str); 0] = [];
+    assert_eq!(
+        pick_best_pve_socket_key(2, empty.iter().map(|(id, k)| (*id, *k))),
+        None,
+    );
+
+    // Non-PVE keys are skipped silently.
+    let mixed = [("a", "pbsc-aaaa111111"), ("b", "pve2b-bbbb222222")];
+    assert_eq!(
+        pick_best_pve_socket_key(1, mixed.iter().map(|(id, k)| (*id, *k))),
+        Some("b"),
+    );
+}
+
+#[test]
+fn schema_accepts_pve_pbs_only() {
+    use proxmox_schema::ApiType;
+    let schema = SubscriptionKeyEntry::API_SCHEMA.unwrap_object_schema();
+    let key_schema = schema
+        .lookup("key")
+        .expect("key property in object schema")
+        .1;
+    assert!(key_schema.parse_simple_value("garbage").is_err());
+    assert!(key_schema.parse_simple_value("xxx-yyyyyyyyyy").is_err());
+    assert!(key_schema.parse_simple_value("pve4b-1234567890").is_ok());
+    assert!(key_schema.parse_simple_value("pbss-abcdef0123").is_ok());
+    // PMG and POM are not driven by PDM today, so the schema rejects them; widen the regex
+    // when remote-side support lands.
+    assert!(key_schema.parse_simple_value("pmgb-deadbeef00").is_err());
+    assert!(key_schema.parse_simple_value("pomb-deadbeef00").is_err());
+}
+
+#[test]
+fn verify_serverid_helper() {
+    let entry = SubscriptionKeyEntry {
+        key: "pve4b-aa11bb2233".to_string(),
+        product_type: ProductType::Pve,
+        serverid: Some("AABBCCDD".to_string()),
+        ..Default::default()
+    };
+
+    let mut info = proxmox_subscription::SubscriptionInfo::default();
+    info.serverid = Some("AABBCCDD".to_string());
+    assert_eq!(verify_serverid(&entry, &info).unwrap(), None);
+
+    info.serverid = Some("DEADBEEF".to_string());
+    let mismatch = verify_serverid(&entry, &info).unwrap().unwrap();
+    assert_eq!(mismatch.expected, "AABBCCDD");
+    assert_eq!(mismatch.actual, "DEADBEEF");
+
+    // entry without serverid -> nothing to verify
+    let entry = SubscriptionKeyEntry {
+        key: "pve4b-aa11bb2233".to_string(),
+        product_type: ProductType::Pve,
+        ..Default::default()
+    };
+    assert_eq!(verify_serverid(&entry, &info).unwrap(), None);
+}
diff --git a/lib/pdm-config/src/lib.rs b/lib/pdm-config/src/lib.rs
index 5b9bcca3..46ad1a2b 100644
--- a/lib/pdm-config/src/lib.rs
+++ b/lib/pdm-config/src/lib.rs
@@ -8,6 +8,7 @@ pub mod domains;
 pub mod node;
 pub mod remotes;
 pub mod setup;
+pub mod subscriptions;
 pub mod views;
 
 mod config_version_cache;
diff --git a/lib/pdm-config/src/setup.rs b/lib/pdm-config/src/setup.rs
index 5adb05f8..77941fc4 100644
--- a/lib/pdm-config/src/setup.rs
+++ b/lib/pdm-config/src/setup.rs
@@ -31,6 +31,13 @@ pub fn create_configdir() -> Result<(), Error> {
         0o750,
     )?;
 
+    mkdir_perms(
+        crate::subscriptions::CONFIG_PATH,
+        api_user.uid,
+        api_user.gid,
+        0o750,
+    )?;
+
     Ok(())
 }
 
diff --git a/lib/pdm-config/src/subscriptions.rs b/lib/pdm-config/src/subscriptions.rs
new file mode 100644
index 00000000..9e6eeeca
--- /dev/null
+++ b/lib/pdm-config/src/subscriptions.rs
@@ -0,0 +1,116 @@
+//! Read/write subscription key pool configuration.
+//!
+//! Call [`init`] to inject a concrete `SubscriptionKeyConfig` instance before using the
+//! module-level functions.
+//!
+//! The shadow-config functions stash signed `SubscriptionInfo` blobs alongside the plain key
+//! entries, which is intended as future proofing for a more automated (shop) import without having
+//! to adapt the data layer.
+
+use std::sync::OnceLock;
+
+use anyhow::Error;
+
+use proxmox_config_digest::ConfigDigest;
+use proxmox_product_config::{open_api_lockfile, replace_config, replace_secret_config, ApiLockGuard};
+use proxmox_section_config::typed::{ApiSectionDataEntry, SectionConfigData};
+
+use pdm_api_types::subscription::{SubscriptionKeyEntry, SubscriptionKeyShadow};
+use pdm_buildcfg::configdir;
+
+pub const CONFIG_PATH: &str = configdir!("/subscriptions");
+pub const SUBSCRIPTIONS_CFG_FILENAME: &str = configdir!("/subscriptions/keys.cfg");
+const SUBSCRIPTIONS_SHADOW_FILENAME: &str = configdir!("/subscriptions/keys.shadow");
+pub const SUBSCRIPTIONS_CFG_LOCKFILE: &str = configdir!("/subscriptions/.keys.lock");
+
+static INSTANCE: OnceLock<Box<dyn SubscriptionKeyConfig + Send + Sync>> = OnceLock::new();
+
+fn instance() -> &'static (dyn SubscriptionKeyConfig + Send + Sync) {
+    INSTANCE
+        .get()
+        .expect("subscription key config not initialized")
+        .as_ref()
+}
+
+pub fn lock_config() -> Result<ApiLockGuard, Error> {
+    instance().lock_config()
+}
+
+pub fn config() -> Result<(SectionConfigData<SubscriptionKeyEntry>, ConfigDigest), Error> {
+    instance().config()
+}
+
+pub fn shadow_config() -> Result<SectionConfigData<SubscriptionKeyShadow>, Error> {
+    instance().shadow_config()
+}
+
+pub fn save_config(
+    config: &SectionConfigData<SubscriptionKeyEntry>,
+) -> Result<ConfigDigest, Error> {
+    instance().save_config(config)
+}
+
+pub fn save_shadow(shadow: &SectionConfigData<SubscriptionKeyShadow>) -> Result<(), Error> {
+    instance().save_shadow(shadow)
+}
+
+pub trait SubscriptionKeyConfig {
+    fn config(&self) -> Result<(SectionConfigData<SubscriptionKeyEntry>, ConfigDigest), Error>;
+    fn shadow_config(&self) -> Result<SectionConfigData<SubscriptionKeyShadow>, Error>;
+    fn lock_config(&self) -> Result<ApiLockGuard, Error>;
+    fn save_config(
+        &self,
+        config: &SectionConfigData<SubscriptionKeyEntry>,
+    ) -> Result<ConfigDigest, Error>;
+    fn save_shadow(&self, shadow: &SectionConfigData<SubscriptionKeyShadow>) -> Result<(), Error>;
+}
+
+pub struct DefaultSubscriptionKeyConfig;
+
+impl SubscriptionKeyConfig for DefaultSubscriptionKeyConfig {
+    fn lock_config(&self) -> Result<ApiLockGuard, Error> {
+        open_api_lockfile(SUBSCRIPTIONS_CFG_LOCKFILE, None, true)
+    }
+
+    fn config(&self) -> Result<(SectionConfigData<SubscriptionKeyEntry>, ConfigDigest), Error> {
+        let content = proxmox_sys::fs::file_read_optional_string(SUBSCRIPTIONS_CFG_FILENAME)?
+            .unwrap_or_default();
+
+        let digest = openssl::sha::sha256(content.as_bytes());
+        let data =
+            SubscriptionKeyEntry::parse_section_config(SUBSCRIPTIONS_CFG_FILENAME, &content)?;
+
+        Ok((data, digest.into()))
+    }
+
+    fn shadow_config(&self) -> Result<SectionConfigData<SubscriptionKeyShadow>, Error> {
+        let content = proxmox_sys::fs::file_read_optional_string(SUBSCRIPTIONS_SHADOW_FILENAME)?
+            .unwrap_or_default();
+        SubscriptionKeyShadow::parse_section_config(SUBSCRIPTIONS_SHADOW_FILENAME, &content)
+    }
+
+    fn save_config(
+        &self,
+        config: &SectionConfigData<SubscriptionKeyEntry>,
+    ) -> Result<ConfigDigest, Error> {
+        let raw = SubscriptionKeyEntry::write_section_config(SUBSCRIPTIONS_CFG_FILENAME, config)?;
+        let digest: ConfigDigest = openssl::sha::sha256(raw.as_bytes()).into();
+        replace_config(SUBSCRIPTIONS_CFG_FILENAME, raw.as_bytes())?;
+        Ok(digest)
+    }
+
+    fn save_shadow(&self, shadow: &SectionConfigData<SubscriptionKeyShadow>) -> Result<(), Error> {
+        let raw =
+            SubscriptionKeyShadow::write_section_config(SUBSCRIPTIONS_SHADOW_FILENAME, shadow)?;
+        // Signed `SubscriptionInfo` blobs are secrets - mode 0600, priv:priv, so the
+        // unprivileged API user cannot read them. The main keys.cfg keeps 0640 since the API
+        // process still needs to read the key strings.
+        replace_secret_config(SUBSCRIPTIONS_SHADOW_FILENAME, raw.as_bytes())
+    }
+}
+
+pub fn init(instance: Box<dyn SubscriptionKeyConfig + Send + Sync>) {
+    if INSTANCE.set(instance).is_err() {
+        panic!("subscription key config instance already set");
+    }
+}
-- 
2.47.3