kopiur_api/common/

mover.rs

use super::*;
use k8s_openapi::api::core::v1::{
    Affinity, PodSecurityContext, ResourceRequirements, SecurityContext, Toleration,
};
use k8s_openapi::apimachinery::pkg::apis::meta::v1::LabelSelector;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;

/// Per-recipe mover overrides (resources, cache, security context).
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Default, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct MoverSpec {
    /// Resource requests/limits for the mover container.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub resources: Option<k8s_openapi::api::core::v1::ResourceRequirements>,
    /// Override the repository's [`CacheDefaults`] for this recipe's movers.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cache: Option<CacheDefaults>,
    /// Container security context for the mover; merged field-wise over the hardened base,
    /// `moverDefaults`, and any inherited context — this is the highest layer, so every field
    /// set here wins. Combines with `inheritSecurityContextFrom`: fields you set override the
    /// workload's, fields you omit are inherited, and this context stands in alone when
    /// inheritance cannot resolve a pod.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub security_context: Option<k8s_openapi::api::core::v1::SecurityContext>,
    /// Pod security context for the mover (notably `fsGroup` for group-writable restore
    /// volumes). Same layering as `securityContext`: highest layer, merged field-wise, and
    /// combinable with `inheritSecurityContextFrom`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub pod_security_context: Option<k8s_openapi::api::core::v1::PodSecurityContext>,
    /// Opt-in, namespace-gated privileged mode. Never changes the UID. When the mover runs
    /// as root (`runAsUser: 0`), adds the capabilities the operation needs on top of
    /// `drop: [ALL]`: `DAC_OVERRIDE` for a backup (read files owned by any uid/gid);
    /// `CHOWN`, `DAC_OVERRIDE`, `FOWNER`, `FSETID` for a restore (reproduce original
    /// ownership). Grants nothing for maintenance, or when the mover does not run as
    /// root (capabilities are inert for non-root); set `runAsUser: 0` yourself.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub privileged_mode: Option<bool>,
    /// Copy the UID/GID security context from a live workload rather than hard-coding it.
    ///
    /// Requires the workload to pin `runAsUser` (container or pod level): a UID that comes
    /// from the container image's `USER` line is invisible in the pod spec and cannot be
    /// inherited — the mover would silently run as its own image's UID instead.
    ///
    /// May be combined with `securityContext`/`podSecurityContext`, which override it
    /// field-wise and act as the fallback when no workload pod can be resolved.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub inherit_security_context_from: Option<InheritSecurityContextFrom>,
    /// Per-recipe override of `Job.spec.ttlSecondsAfterFinished` so finished Jobs self-GC.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub ttl_seconds_after_finished: Option<i64>,
}

impl MoverSpec {
    /// Whether this mover requests **elevated privileges** that the workload
    /// namespace must explicitly opt into (ADR §4.11/§G16). True when
    /// `privilegedMode` is set, or the `securityContext` runs as root / privileged
    /// / with escalation / with added Linux capabilities.
    ///
    /// The rationale is the same as VolSync's `privileged-movers` model: the
    /// controller mints a mover `ServiceAccount` in the workload namespace, and a
    /// tenant with access there could reuse it to run pods at the mover's privilege.
    /// Granting an elevated mover is therefore a per-namespace admin decision, gated
    /// by a namespace annotation rather than allowed implicitly. Pure + exhaustive
    /// so the definition of "privileged" lives in one tested place.
    pub fn requires_privilege(&self) -> bool {
        requires_privilege_resolved(
            self.security_context.as_ref(),
            self.pod_security_context.as_ref(),
            self.privileged_mode,
        )
    }

    /// Admission-time warning when `privilegedMode: true` would certainly grant nothing:
    /// this spec pins a non-zero `runAsUser` (capabilities are inert for non-root, and
    /// `privilegedMode` never picks a UID).
    ///
    /// Judged on `hardened ⊂ self` only. A spec with no `runAsUser` stays silent: the
    /// repository's `moverDefaults` may pin `runAsUser: 0` and is invisible here, so
    /// warning would be a false positive. The controller's `PrivilegedModeInert` Event,
    /// which sees the merged context, reports that case. An `inheritSecurityContextFrom`
    /// identity resolves at reconcile time, so it stays silent too.
    pub fn privileged_mode_warning(&self, op: MoverOperation) -> Option<String> {
        if self.inherit_security_context_from.is_some() {
            return None;
        }
        let resolved = resolve_mover(
            None,
            self.security_context.as_ref(),
            self.pod_security_context.as_ref(),
            None,
            None,
            None,
        );
        let effect = privileged_mode_effect(
            &resolved.security_context,
            resolved.pod_security_context.as_ref(),
            self.privileged_mode,
            op,
        );
        match effect {
            // A non-root uid pinned on this object stays non-root whatever the
            // repository's moverDefaults say, so it is certainly inert.
            PrivilegedModeEffect::InertNonRootUid { .. } => effect.warning(),
            // No uid here: moverDefaults (invisible at admission) may pin root. Leave
            // it to the controller's PrivilegedModeInert Event, which sees the merge.
            PrivilegedModeEffect::InertUnsetUid
            | PrivilegedModeEffect::NotRequested
            | PrivilegedModeEffect::NothingToGrant
            | PrivilegedModeEffect::Granted => None,
        }
    }

    /// Overlay `self` — the **HIGHER** layer — onto `base`, field by field, producing
    /// one `MoverSpec` that stands in for both. This is what lets an invocation carry
    /// its own mover overrides on top of a shared, GitOps-managed recipe:
    /// `Snapshot.spec.mover` merged over `SnapshotPolicy.spec.mover` (#464).
    ///
    /// The merge delegates to the SAME primitives the repository-layer merge uses, so a
    /// per-run override can never resolve differently from a per-recipe one:
    ///
    /// - `securityContext`/`podSecurityContext` go through [`merge_context_pair`] as one
    ///   `(container, pod)` layer pair — field-wise per dimension **plus** identity
    ///   promotion, so a policy's container-level `runAsUser` can never shadow a
    ///   snapshot's pod-level one. That merge is associative, so folding these two
    ///   layers here and then handing the result to [`resolve_mover`] is identical to a
    ///   flat `hardened ⊂ moverDefaults ⊂ inherited ⊂ policy ⊂ snapshot` merge.
    /// - `resources` via [`merge_resources_opt`] (per-key requests/limits).
    /// - `cache` via [`CacheDefaults::merge`] (per-field).
    /// - `privilegedMode`, `ttlSecondsAfterFinished` and `inheritSecurityContextFrom`
    ///   are whole-value knobs with no inner structure to merge, so the higher layer's
    ///   value wins outright and `None` falls through to `base`.
    ///
    /// `base: None` is the identity: the result is `self` unchanged. Because the layer
    /// order is fixed by this signature, a caller cannot accidentally invert it.
    pub fn merge_over(&self, base: Option<&MoverSpec>) -> MoverSpec {
        let Some(base) = base else {
            return self.clone();
        };
        // One layer pair, not two independent dimension merges — a lone
        // `merge_security_context` here would reintroduce cross-dimension identity
        // shadowing (see `merge_context_pair`'s docs).
        let (security_context, pod_security_context) = merge_context_pair(
            base.security_context.as_ref(),
            base.pod_security_context.as_ref(),
            self.security_context.as_ref(),
            self.pod_security_context.as_ref(),
        );
        MoverSpec {
            resources: merge_resources_opt(base.resources.as_ref(), self.resources.as_ref()),
            cache: CacheDefaults::merge(base.cache.as_ref(), self.cache.as_ref()),
            security_context,
            pod_security_context,
            privileged_mode: self.privileged_mode.or(base.privileged_mode),
            // Whole-value: a higher layer that names its own inherit source replaces the
            // lower layer's outright rather than blending two workload selectors into
            // something neither manifest asked for.
            inherit_security_context_from: self
                .inherit_security_context_from
                .clone()
                .or_else(|| base.inherit_security_context_from.clone()),
            ttl_seconds_after_finished: self
                .ttl_seconds_after_finished
                .or(base.ttl_seconds_after_finished),
        }
    }
}

/// How the mover co-locates with the node an RWO source/destination PVC is attached to.
#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq, Default, JsonSchema)]
pub enum SourceColocationMode {
    /// Pin to the attached node when discoverable, else schedule freely; the default.
    #[default]
    Auto,
    /// Like `Auto`, but fail the run when an RWO PVC's node cannot be determined.
    Required,
    /// Never compute a node pin; use only the explicit `nodeSelector`/`affinity`/`tolerations`.
    Disabled,
}

/// Controls mover/source-PVC node co-location (RWO Multi-Attach avoidance).
#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq, Default, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct SourceColocation {
    /// The co-location strategy. Defaults to [`SourceColocationMode::Auto`].
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub mode: Option<SourceColocationMode>,
}

/// Repository-wide mover defaults inherited by every mover, overridable per-recipe via `mover`.
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Default, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct MoverDefaults {
    /// Container security-context base for every mover.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub security_context: Option<SecurityContext>,
    /// Pod security-context base (notably `fsGroup`) for every mover.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub pod_security_context: Option<PodSecurityContext>,
    /// Resource requests/limits base for the mover container.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub resources: Option<ResourceRequirements>,
    /// kopia cache defaults for every mover.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cache: Option<CacheDefaults>,
    /// Defaults for the deep-verification scratch (restore-test) volume.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub scratch: Option<ScratchDefaults>,
    /// Pod `nodeSelector` for every mover.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub node_selector: Option<BTreeMap<String, String>>,
    /// Pod tolerations for every mover.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub tolerations: Option<Vec<Toleration>>,
    /// Pod affinity for every mover.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub affinity: Option<Affinity>,
    /// How a mover co-locates with its RWO PVC's node; defaults to [`SourceColocationMode::Auto`].
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub source_colocation: Option<SourceColocation>,
    /// `Job.spec.ttlSecondsAfterFinished` for every mover Job so finished Jobs self-GC.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub ttl_seconds_after_finished: Option<i64>,
    /// Repository throttle limits applied by every mover after it connects.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub throttle: Option<Throttle>,
    /// Extra labels for every mover POD (and the `Job` that owns it), merged
    /// UNDER kopiur's own labels — a key kopiur sets always wins, so a
    /// user-supplied value can never break the selectors the controller counts
    /// and reaps by.
    ///
    /// This is the hook for cluster machinery that keys off pod labels and that
    /// kopiur has no field of its own for: a Kueue `kueue.x-k8s.io/queue-name` to
    /// put movers under a cluster queue, a monitoring/`NetworkPolicy` selector, a
    /// service-mesh exclusion label.
    ///
    /// Keys under `kopiur.home-operations.com/` and the exact key
    /// `app.kubernetes.io/managed-by` are reserved and rejected at admission.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub pod_labels: Option<BTreeMap<String, String>>,
    /// Extra annotations for every mover pod. Unlike `podLabels`
    /// these are **pod-template-only** — they are not mirrored onto the `Job`
    /// object, because the common case is a sidecar-injection opt-out
    /// (`sidecar.istio.io/inject: "false"`, `linkerd.io/inject: disabled`,
    /// `vault.hashicorp.com/agent-inject: "false"`) that only means anything on
    /// the pod a mesh webhook actually sees. A mover is a short-lived batch pod;
    /// an injected sidecar that never exits keeps its Job running forever.
    ///
    /// Same reserved keys as `podLabels`, rejected at admission.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub pod_annotations: Option<BTreeMap<String, String>>,
}

/// Built-in default `Job.spec.ttlSecondsAfterFinished` (1h) applied to a mover Job
/// when neither `moverDefaults.ttlSecondsAfterFinished` nor the recipe's
/// `mover.ttlSecondsAfterFinished` sets one, so finished backup/restore Jobs and
/// their pods self-GC instead of lingering (ADR-0005 §12).
pub const DEFAULT_JOB_TTL_SECONDS: i64 = 3600;

/// Per-side throttle overrides for a `kopia snapshot migrate` run, which reads
/// from a SOURCE repository and writes into a DESTINATION repository under two
/// separate kopia connections.
///
/// `snapshot migrate` has **no speed flags of its own** — the only lever is
/// `kopia repository throttle set` on each side's connection (kopia persists the
/// limits in that connection's client config, and the migrate honors them when it
/// reopens it). So a cap here is expressed per side, and each side overrides
/// **that side's repository's** `moverDefaults.throttle` field by field: a field
/// set here wins, a field left unset falls back to the repository default, and
/// all four [`Throttle`] knobs are available on each side independently.
///
/// Not `Copy` (neither is [`Throttle`]).
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, Default, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct MigrateThrottle {
    /// Caps for the SOURCE (read) side, overriding the source repository's
    /// `moverDefaults.throttle` field by field. Applied with `repository
    /// throttle set` on the migrate's read-only source connection — accepted
    /// there, so a read-only source is throttled like any other.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub source: Option<Throttle>,
    /// Caps for the DESTINATION (write) side, overriding the destination
    /// repository's `moverDefaults.throttle` field by field. Applied with
    /// `repository throttle set` on the destination connection the migrate
    /// writes through.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub destination: Option<Throttle>,
}

/// Repository-wide throttling for a mover's kopia connection; each `None` leaves kopia's current limit.
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, Default, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct Throttle {
    /// Cap upload throughput in bytes/sec (`--upload-bytes-per-second`).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub upload_bytes_per_second: Option<i64>,
    /// Cap download throughput in bytes/sec (`--download-bytes-per-second`).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub download_bytes_per_second: Option<i64>,
    /// Cap read/list ops/sec (`--read-requests-per-second`).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub read_ops_per_second: Option<i64>,
    /// Cap write ops/sec (`--write-requests-per-second`).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub write_ops_per_second: Option<i64>,
}

/// The fully-resolved mover configuration for a single run, after the 3-layer
/// field-wise merge `hardened ⊂ repo.moverDefaults ⊂ recipe.mover` (ADR-0004 §1/§2).
/// `security_context` is ALWAYS present (the hardened base guarantees it); the rest are
/// `Some` only when some layer set them. The privileged-mover gate (§4.11/§G16) runs on
/// `security_context`/`pod_security_context` *here* — the merged result — not on the raw
/// recipe, so an elevation introduced by `moverDefaults` is still gated.
#[derive(Clone, Debug, PartialEq)]
pub struct ResolvedMover {
    /// Merged container security context — always present (hardened base).
    pub security_context: SecurityContext,
    /// Merged pod security context, if any layer set one.
    pub pod_security_context: Option<PodSecurityContext>,
    /// Merged resource requirements, if any layer set them.
    pub resources: Option<ResourceRequirements>,
    /// Merged cache config, if any layer set it.
    pub cache: Option<CacheDefaults>,
    /// Pod node selector from `moverDefaults` (no per-recipe override surface today).
    pub node_selector: Option<BTreeMap<String, String>>,
    /// Pod tolerations from `moverDefaults`.
    pub tolerations: Option<Vec<Toleration>>,
    /// Pod affinity from `moverDefaults`.
    pub affinity: Option<Affinity>,
    /// Resolved RWO source/destination co-location mode (`moverDefaults.sourceColocation.mode`),
    /// defaulting to [`SourceColocationMode::Auto`]. Always `Some` so the reconciler
    /// has a concrete strategy. RWO multi-attach fix.
    pub source_colocation: SourceColocationMode,
    /// Resolved Job TTL (recipe `mover.ttlSecondsAfterFinished` wins over
    /// `moverDefaults.ttlSecondsAfterFinished`, falling back to
    /// [`DEFAULT_JOB_TTL_SECONDS`]). Always `Some` so finished Jobs self-GC. §12.
    pub ttl_seconds_after_finished: Option<i64>,
    /// Resolved repository throttle (`moverDefaults.throttle`), if any. §13(e).
    pub throttle: Option<Throttle>,
    /// Extra pod (and Job) labels from `moverDefaults.podLabels`, merged UNDER
    /// kopiur's own labels by the caller — kopiur-managed keys always win.
    pub pod_labels: Option<BTreeMap<String, String>>,
    /// Extra pod-template annotations from `moverDefaults.podAnnotations`
    /// (pod-only; never mirrored onto the `Job`).
    pub pod_annotations: Option<BTreeMap<String, String>>,
}

/// Resolve the effective mover configuration via the layer merge
/// `hardened ⊂ moverDefaults ⊂ recipe` (ADR-0004 §1/§2).
///
/// - `defaults`: the repository's `moverDefaults` (None when the repo sets none).
/// - `recipe_sc`/`recipe_psc`: the recipe's **effective** container/pod context, which the
///   controller has already resolved — the explicit `mover.securityContext`/
///   `podSecurityContext` overlaid on top of any context inherited from a workload via
///   `inheritSecurityContextFrom` (they combine; explicit wins). The full ladder is
///   therefore `hardened ⊂ moverDefaults ⊂ inherited ⊂ explicit`. Layers merge as
///   `(container, pod)` **pairs** via [`merge_context_pair`]: field-wise per dimension,
///   plus identity promotion so the effective UID/GID belongs to the **highest layer that
///   pins one, regardless of which dimension it wrote** — a `moverDefaults` container-level
///   `runAsUser` can never shadow an inherited pod-level one. The pair merge is
///   associative, so folding the inner two layers before this call is identical to a flat
///   four-layer merge. The recipe layer enters here as a *layer*, NOT a whole-chain
///   replacement — the hardened base + `moverDefaults` still supply `drop:[ALL]`/seccomp
///   and a partial recipe context can only tighten.
/// - `recipe_resources`/`recipe_cache`: from `mover.resources` / `mover.cache`.
///
/// `node_selector`/`tolerations`/`affinity`/`pod_labels`/`pod_annotations`/`ttl` flow from
/// `moverDefaults` (no per-recipe surface for the first five today; TTL is overridable by
/// the caller post-resolve).
pub fn resolve_mover(
    defaults: Option<&MoverDefaults>,
    recipe_sc: Option<&SecurityContext>,
    recipe_psc: Option<&PodSecurityContext>,
    recipe_resources: Option<&ResourceRequirements>,
    recipe_cache: Option<&CacheDefaults>,
    recipe_ttl_seconds_after_finished: Option<i64>,
) -> ResolvedMover {
    // The hardened pair is the lowest layer: the container hardening plus the pod-level
    // fsGroup that makes the cache writable. Both are always present, so every mover pod
    // — bootstrap, backup, restore, maintenance, verification, replication — carries the
    // hardened defaults unless a higher layer overrides them.
    let hardened_sc = hardened_security_context();
    let hardened_psc = hardened_pod_security_context();
    // hardened ⊂ moverDefaults, as one (container, pod) layer pair.
    let (base_sc, base_psc) = merge_context_pair(
        Some(&hardened_sc),
        Some(&hardened_psc),
        defaults.and_then(|d| d.security_context.as_ref()),
        defaults.and_then(|d| d.pod_security_context.as_ref()),
    );
    // (hardened ⊂ moverDefaults) ⊂ recipe.
    let (security_context, pod_security_context) =
        merge_context_pair(base_sc.as_ref(), base_psc.as_ref(), recipe_sc, recipe_psc);
    let security_context =
        security_context.expect("the hardened container base layer is always present");
    // Normalize the merged result against every kubelet/apiserver security-context invariant
    // (see `crate::invariants`) so a contradiction the field-wise merge can assemble — most
    // importantly an inherited-root `runAsUser: 0` left under the hardened `runAsNonRoot:
    // true` — becomes a VALID (privileged-gated) mover rather than a pod wedged forever in
    // `CreateContainerConfigError`.
    let (security_context, pod_security_context) =
        crate::invariants::enforce_security_context_invariants(
            security_context,
            pod_security_context,
        );
    ResolvedMover {
        security_context,
        pod_security_context,
        resources: merge_resources_opt(
            defaults.and_then(|d| d.resources.as_ref()),
            recipe_resources,
        ),
        cache: CacheDefaults::merge(defaults.and_then(|d| d.cache.as_ref()), recipe_cache),
        node_selector: defaults.and_then(|d| d.node_selector.clone()),
        tolerations: defaults.and_then(|d| d.tolerations.clone()),
        affinity: defaults.and_then(|d| d.affinity.clone()),
        // `moverDefaults.sourceColocation.mode`, defaulting to `Auto` so RWO movers
        // co-locate with their source PVC's node out of the box (RWO multi-attach fix).
        source_colocation: defaults
            .and_then(|d| d.source_colocation.as_ref())
            .and_then(|c| c.mode)
            .unwrap_or_default(),
        // Recipe TTL wins over the repo default; a built-in default applies when
        // neither sets one so every finished Job self-GCs (ADR-0005 §12).
        ttl_seconds_after_finished: Some(
            recipe_ttl_seconds_after_finished
                .or_else(|| defaults.and_then(|d| d.ttl_seconds_after_finished))
                .unwrap_or(DEFAULT_JOB_TTL_SECONDS),
        ),
        throttle: defaults.and_then(|d| d.throttle.clone()),
        pod_labels: defaults.and_then(|d| d.pod_labels.clone()),
        pod_annotations: defaults.and_then(|d| d.pod_annotations.clone()),
    }
}

impl ResolvedMover {
    /// Apply `privilegedMode` for `op` to this resolved mover (#552): a root mover gets the
    /// operation's capabilities added on top of `drop: [ALL]`; the UID is never changed.
    /// See [`apply_privileged_mode`] for the contract. Re-runs the (idempotent)
    /// security-context invariants afterwards. A no-op unless `privileged_mode ==
    /// Some(true)` and the effective UID is `0`.
    pub fn with_privileged_mode(self, privileged_mode: Option<bool>, op: MoverOperation) -> Self {
        self.with_privileged_mode_effect(privileged_mode, op).0
    }

    /// [`Self::with_privileged_mode`], also returning what was done, so a caller can
    /// surface an inert grant ([`PrivilegedModeEffect::warning`]) to the user.
    pub fn with_privileged_mode_effect(
        self,
        privileged_mode: Option<bool>,
        op: MoverOperation,
    ) -> (Self, PrivilegedModeEffect) {
        let (sc, psc, effect) = apply_privileged_mode(
            self.security_context,
            self.pod_security_context,
            privileged_mode,
            op,
        );
        let (security_context, pod_security_context) =
            crate::invariants::enforce_security_context_invariants(sc, psc);
        (
            Self {
                security_context,
                pod_security_context,
                ..self
            },
            effect,
        )
    }
}

/// Selects workload pods by label.
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct PodSelector {
    /// Label selector matching the workload pod(s) to read context/hooks from.
    pub pod_selector: LabelSelector,
    /// Which container within the matched pod; absent uses the first/only container.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub container: Option<String>,
}

/// Where the mover copies its security context from instead of an explicit context.
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub enum InheritSecurityContextFrom {
    /// Inherit from workload pod(s) matched by an explicit label selector (backup or restore).
    WorkloadSelector(PodSelector),
    /// Backup sources only: auto-derive the workload from the PVC this snapshot backs up.
    PvcConsumer(PvcConsumerInherit),
    /// Restores only: inherit the identity RECORDED on the backup itself
    /// (`Snapshot.status.recorded`, decoded from the `kopiur-meta` kopia tag) —
    /// uid/gid/fsGroup the backup mover actually ran as. Needs no live workload
    /// pod, so it works on a rebuilt cluster and with `target.populator`.
    /// Rejected at admission on SnapshotPolicy/Maintenance (backups read the
    /// live workload; maintenance has no snapshot). Write it as `snapshot: {}`
    /// (an empty sub-object) — a bare `snapshot:` is null and rejected.
    Snapshot(SnapshotInherit),
}

/// Tuning for [`InheritSecurityContextFrom::Snapshot`]. Empty today; a
/// sub-object (like [`PopulatorTarget`](crate::restore::PopulatorTarget)) so
/// future knobs slot in without API breakage.
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, Default, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct SnapshotInherit {}

/// Tuning for [`InheritSecurityContextFrom::PvcConsumer`].
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Default, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct PvcConsumerInherit {
    /// Which container within the matched consumer pod to inherit from; absent uses the first/only.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub container: Option<String>,
}