kopiur_api/validate/

snapshot.rs

use super::*;
use crate::common::ScheduleDeletePolicy;
use crate::error::{ValidationError, ValidationResult};
use crate::snapshot::{Origin, SnapshotSpec};
use crate::snapshot_policy::{CopyMethod, Hook, HttpRequestHook, SnapshotPolicySpec};
use crate::snapshot_schedule::SnapshotScheduleSpec;

/// Validate the `spec.repository` / `spec.repositories` surface of a
/// `SnapshotPolicy`, accumulating every independent problem:
///
///   1. **Exactly one of** the two shapes (mirrors the spec-level CEL rule, so
///      the refusal also covers objects stored before the CRD carried it).
///   2. Per-ref shape validity ([`validate_repository_ref`] over whatever
///      refs exist — tolerant iteration, one error per bad ref).
///   3. **Duplicate rejection** over `spec.repositories`, normalized via
///      [`crate::common::repo_key`]. Validators are spec-only (no owner
///      namespace here), so the key uses a fixed `""` owner-namespace
///      sentinel: it is identical for every entry of ONE policy, which is all
///      duplicate detection needs. (A `namespace`-omitted and an explicit
///      same-namespace ref to one repository are not caught here; the webhook
///      guards that identity-level collision separately.)
///   4. `hooks` × `repositories` mutual exclusion
///      ([`ValidationError::PolicyHooksWithRepositories`]) — hooks quiesce a
///      workload around ONE capture; N concurrent fan-out children void that
///      contract (use a single-repo policy + `SnapshotReplication` instead).
///
/// A well-formed multi-repo spec is ACCEPTED — the M7 "not yet enabled"
/// feature gate was lifted once the fan-out data path (per-child pins,
/// per-repo retention/cache/verification) landed end-to-end.
fn validate_policy_repositories(spec: &SnapshotPolicySpec) -> Vec<ValidationError> {
    let mut errs = Vec::new();
    if let Err(e) = crate::snapshot_policy::policy_repositories(spec) {
        errs.push(e);
    }
    for r in crate::snapshot_policy::repository_refs(spec) {
        if let Err(e) = validate_repository_ref(r) {
            errs.push(e);
        }
    }
    let mut seen: std::collections::BTreeMap<String, usize> = std::collections::BTreeMap::new();
    for (i, r) in spec.repositories.iter().enumerate() {
        let key = crate::common::repo_key(r, "");
        if let Some(&first) = seen.get(&key) {
            errs.push(ValidationError::PolicyRepositoriesDuplicate {
                key,
                first,
                second: i,
            });
        } else {
            seen.insert(key, i);
        }
    }
    if crate::snapshot_policy::is_multi_repo(spec) && spec.hooks.is_some() {
        errs.push(ValidationError::PolicyHooksWithRepositories);
    }
    errs
}

/// Validate a `SnapshotPolicy` spec, accumulating all problems.
pub fn validate_backup_config(spec: &SnapshotPolicySpec) -> Vec<ValidationError> {
    let mut errs = Vec::new();
    errs.extend(validate_policy_repositories(spec));
    if spec.sources.is_empty() {
        errs.push(ValidationError::MissingRequiredField {
            field: "spec.sources (at least one source required)".to_string(),
        });
    }
    for source in &spec.sources {
        if let Err(e) = validate_source(source) {
            errs.push(e);
        }
    }
    // A CSI volume CLONE has no group counterpart — there is no
    // "VolumeGroupClone" — so `copyMethod: Clone` can only ever capture each PVC
    // independently. Since `groupBy` server-side-defaults to
    // `VolumeGroupSnapshot`, an untouched selector policy with `Clone` would
    // silently get N unrelated live-volume clones while every member CR still
    // named a VolumeGroupSnapshot that is never created. Refuse instead: a
    // consistency guarantee that is quietly not honored is worse than none.
    if spec.sources.iter().any(|s| s.pvc_selector.is_some())
        && spec.copy_method == crate::snapshot_policy::CopyMethod::Clone
        && spec.group_by != Some(crate::snapshot_policy::GroupBy::None)
    {
        errs.push(ValidationError::InvalidFieldValue {
            field: "spec.groupBy".to_string(),
            reason: "`copyMethod: Clone` cannot do `groupBy: VolumeGroupSnapshot` (the \
                     default). Fix: use `copyMethod: Snapshot` for a consistency group, or set \
                     `groupBy: None`"
                .to_string(),
        });
    }
    // Identity shape (kopia's username@hostname:path contract). The explicit
    // overrides are validated here — client-free, so this runs on every admission
    // even when the webhook has no kube client. CEL-resolved values and the
    // name/namespace defaults are validated where they are resolved
    // (`resolve_identity`), and again defensively at reconcile time.
    if let Some(id) = &spec.identity {
        if let Some(u) = &id.username
            && let Err(e) = validate_identity_component("spec.identity.username", u)
        {
            errs.push(e);
        }
        if let Some(h) = &id.hostname
            && let Err(e) = validate_identity_component("spec.identity.hostname", h)
        {
            errs.push(e);
        }
    }
    for (i, source) in spec.sources.iter().enumerate() {
        if let Some(p) = &source.source_path_override
            && let Err(e) =
                validate_source_path(&format!("spec.sources[{i}].sourcePathOverride"), p)
        {
            errs.push(e);
        }
    }
    // `copyMethod: Direct` + `readOnly: false` is the one combination that reaches the
    // workload's own volume. `readOnly: false` is only ever set to make `fsGroup` apply,
    // and the kubelet applies it by recursively chgrp-ing the mount and adding
    // group-write. Under Snapshot/Clone that walk rewrites a throwaway staged PVC and is
    // free; under Direct it permanently rewrites live production data while the workload
    // runs — and the mover ships `fsGroup: 65532` by DEFAULT, so a user who sets one bool
    // to fix a permissions error would have their data re-grouped with no other signal.
    // Databases that refuse an over-permissive data directory (postgres, redis) fail to
    // restart afterwards. Require the intent to be stated; it is not inferable.
    for (i, source) in spec.sources.iter().enumerate() {
        if crate::snapshot_policy::source_mutates_live_volume(spec.copy_method, source)
            && !source.acknowledge_live_mutation.unwrap_or(false)
        {
            errs.push(ValidationError::InvalidFieldValue {
                field: format!("spec.sources[{i}].readOnly"),
                reason: "copyMethod: Direct with readOnly: false lets the kubelet permanently \
                         change ownership of the live volume to the mover's fsGroup. Fix: use \
                         copyMethod: Snapshot/Clone, or set acknowledgeLiveMutation: true on \
                         this source if you mean it"
                    .to_string(),
            });
        }
    }
    // `volumeSnapshotClassName` only applies when a PVC source is CSI-snapshotted/cloned
    // (`copyMethod: Snapshot`/`Clone`). An NFS source has no PVC to snapshot, so pairing
    // it with an explicit class is a configuration mistake — reject it at admission with
    // an actionable message rather than silently ignoring the class. (`copyMethod` itself
    // can't be rejected for NFS: it defaults to `Snapshot` implicitly and an NFS source
    // is simply read directly.)
    if spec.volume_snapshot_class_name.is_some() && spec.sources.iter().any(|s| s.nfs.is_some()) {
        errs.push(ValidationError::InvalidFieldValue {
            field: "spec.volumeSnapshotClassName".to_string(),
            reason: "does not apply to an NFS source, which cannot be CSI-snapshotted. Fix: \
                     remove volumeSnapshotClassName"
                .to_string(),
        });
    }
    // Same reasoning as the NFS arm above: a stream source has no PVC to snapshot.
    if spec.volume_snapshot_class_name.is_some() && spec.sources.iter().any(|s| s.stream.is_some())
    {
        errs.push(ValidationError::InvalidFieldValue {
            field: "spec.volumeSnapshotClassName".to_string(),
            reason: "does not apply to a stream source, which mounts no volume. Fix: remove \
                     volumeSnapshotClassName"
                .to_string(),
        });
    }
    // A stream source must be its policy's ONLY source. Expansion fans out selector
    // sources only (`expand_sources` returns None when no `pvcSelector` is present), so
    // a non-selector policy runs `sources[0]` and nothing else — pairing a stream with
    // another source would silently back up just one of them. Refuse instead of
    // dropping a backup the user believes is configured.
    if spec.sources.len() > 1
        && let Some(i) = spec.sources.iter().position(|s| s.stream.is_some())
    {
        errs.push(ValidationError::InvalidFieldValue {
            field: format!("spec.sources[{i}].stream"),
            reason: format!(
                "a stream source must be the only source in its SnapshotPolicy, but this one \
                 has {}. Fix: move the stream source into its own SnapshotPolicy",
                spec.sources.len()
            ),
        });
    }
    if let Some(m) = &spec.mover {
        // `inheritSecurityContextFrom.snapshot` replays a backup's RECORDED identity;
        // a backup has no recorded identity to replay — it is the run that records one.
        if let Err(e) = forbid_snapshot_inherit(
            m,
            "snapshotPolicy",
            "is restore-only. Fix: use pvcConsumer or workloadSelector for a backup",
        ) {
            errs.push(e);
        }
        if let Err(e) = validate_mover(m, "SnapshotPolicy mover") {
            errs.push(e);
        }
    }
    // `snapshot create --upload-limit-mb` (M4 flag sweep, issue #216): a count
    // knob, must be at least 1 (0 or negative disables the flag's own purpose).
    if let Some(u) = &spec.upload
        && let Some(mb) = u.limit_mb
        && let Some(e) = require_min(
            "SnapshotPolicy spec.upload.limitMb",
            mb,
            NumericBound::Megabytes,
        )
    {
        errs.push(e);
    }
    // Data-loss guard: a retention that selects no snapshots prunes EVERY Snapshot the
    // moment it runs. `retention: None` means "don't prune" (safe) and is not flagged;
    // only an explicit but empty/all-zero retention is the trap.
    if let Some(r) = &spec.retention
        && retention_keeps_nothing(r)
    {
        errs.push(ValidationError::InvalidFieldValue {
            field: "spec.retention".to_string(),
            reason: "keeps no snapshots (every keep* is unset or 0), so every Snapshot would be \
                     pruned. Fix: set at least one (e.g. keepLatest: 1), or omit spec.retention \
                     to disable pruning"
                .to_string(),
        });
    }
    // Verification (ADR-0005 §4): override schedules must parse, and the optional
    // `successExpr` (ADR-0005 §15) must compile + trial-evaluate to a bool with no
    // out-of-scope variable — rejected at admission rather than at first verify run.
    if let Some(v) = &spec.verification {
        if let Some(q) = &v.quick {
            // The flat `verification.quick.cron` shape moved under `quick.schedule`
            // (GitHub #174) — a required schedule would break decode of old persisted
            // objects, so `schedule` is Option and this validator is the gate that
            // rejects a re-apply of the old shape with an actionable pointer.
            match &q.schedule {
                None => errs.push(ValidationError::InvalidFieldValue {
                    field: "spec.verification.quick.schedule".to_string(),
                    reason: "`verification.quick.cron` moved to \
                             `verification.quick.schedule.cron`. Fix: move your \
                             cron/jitter/timezone fields under `schedule:`"
                        .to_string(),
                }),
                Some(s) => {
                    if let Err(e) = validate_cron(&s.cron) {
                        errs.push(e);
                    }
                    if let Err(e) = validate_timezone(s.timezone.as_deref()) {
                        errs.push(e);
                    }
                    if let Err(e) = validate_jitter(
                        "spec.verification.quick.schedule.jitter",
                        s.jitter.as_deref(),
                    ) {
                        errs.push(e);
                    }
                }
            }
            // `kopia snapshot verify` tuning knobs: counts must be at least 1.
            // `maxErrors` is deliberately unconstrained — 0 is a valid, meaningful
            // value (kopia's own default, "stop at the first error").
            if let Some(p) = q.parallel
                && let Some(e) = require_min(
                    "SnapshotPolicy spec.verification.quick.parallel",
                    p.into(),
                    NumericBound::Count,
                )
            {
                errs.push(e);
            }
            if let Some(p) = q.file_parallelism
                && let Some(e) = require_min(
                    "SnapshotPolicy spec.verification.quick.fileParallelism",
                    p.into(),
                    NumericBound::Count,
                )
            {
                errs.push(e);
            }
            if let Some(p) = q.file_queue_length
                && let Some(e) = require_min(
                    "SnapshotPolicy spec.verification.quick.fileQueueLength",
                    p.into(),
                    NumericBound::Count,
                )
            {
                errs.push(e);
            }
        }
        if let Some(d) = &v.deep {
            if let Err(e) = validate_cron(&d.schedule.cron) {
                errs.push(e);
            }
            if let Err(e) = validate_timezone(d.schedule.timezone.as_deref()) {
                errs.push(e);
            }
            if let Err(e) = validate_jitter(
                "spec.verification.deep.schedule.jitter",
                d.schedule.jitter.as_deref(),
            ) {
                errs.push(e);
            }
            // `restore --parallel` under the hood (deep verify IS a restore).
            if let Some(p) = d.parallel
                && let Some(e) = require_min(
                    "SnapshotPolicy spec.verification.deep.parallel",
                    p.into(),
                    NumericBound::Count,
                )
            {
                errs.push(e);
            }
        }
        if let Some(expr) = &v.success_expr
            && let Err(e) = crate::success_expr::validate_success_expr(expr)
        {
            errs.push(e);
        }
    }
    errs.extend(validate_staging(spec));
    // Preflight: the timeout must parse, check names must be unique + non-blank, and
    // each check expression must compile + trial-evaluate to a bool with no
    // out-of-scope variable — rejected at admission rather than at the first run.
    if let Some(pf) = &spec.preflight {
        if let Some(t) = &pf.timeout
            && crate::duration::parse_go_duration(t).is_none()
        {
            errs.push(ValidationError::InvalidFieldValue {
                field: "spec.preflight.timeout".to_string(),
                reason: format!(
                    "{t:?} is not a valid duration. Fix: use a Go-style duration like 10m or 1h; \
                     omit for the default (10m), or 0 to wait forever"
                ),
            });
        }
        let mut seen = std::collections::BTreeSet::new();
        for (i, c) in pf.checks.iter().enumerate() {
            let name = c.name.trim();
            if name.is_empty() {
                errs.push(ValidationError::MissingRequiredField {
                    field: format!("spec.preflight.checks[{i}].name"),
                });
            } else if !seen.insert(name.to_string()) {
                errs.push(ValidationError::InvalidFieldValue {
                    field: format!("spec.preflight.checks[{i}].name"),
                    reason: format!(
                        "duplicate preflight check name {name:?}; names must be unique"
                    ),
                });
            }
            if let Err(e) = crate::preflight::validate_preflight_expr(&c.expr) {
                errs.push(e);
            }
        }
    }
    // Hooks (ADR §4.8): per-hook shape problems are caught at admission rather
    // than at the first backup run (where a quiesce hook failing on a typo would
    // abort the backup).
    if let Some(h) = &spec.hooks {
        for (list, hooks) in [
            ("beforeSnapshot", &h.before_snapshot),
            ("afterSnapshot", &h.after_snapshot),
        ] {
            for (i, hook) in hooks.iter().enumerate() {
                if let Err(e) = validate_hook(list, i, hook) {
                    errs.push(e);
                }
            }
        }
    }
    errs
}

/// Validate `spec.staging` (+ its interplay with `copyMethod` and the sources):
///
///   * `timeout` must parse — rejected at admission rather than silently falling
///     back to the default at the first backup run.
///   * `accessModes` entries must be canonical/unique, and `ReadWriteOncePod`
///     sole ([`validate_access_modes`]).
///   * The staged-PVC override fields (`storageClassName`/`accessModes`) must have
///     a staged PVC to act on — rejected for `copyMethod: Direct` (no staged PVC
///     at all), an NFS source (never staged), and `pvcSelector` sources (staging
///     is skipped for selector expansion). The pre-existing `timeout` and
///     `volumeSnapshotClassName` stay deliberately lenient in those combinations —
///     tightening them now would reject already-persisted objects on re-apply.
fn validate_staging(spec: &SnapshotPolicySpec) -> Vec<ValidationError> {
    let mut errs = Vec::new();
    let Some(st) = &spec.staging else {
        return errs;
    };
    if let Some(t) = &st.timeout
        && crate::duration::parse_go_duration(t).is_none()
    {
        errs.push(ValidationError::InvalidFieldValue {
            field: "spec.staging.timeout".to_string(),
            reason: format!(
                "{t:?} is not a valid duration. Fix: use a Go-style duration like 10m or 1h; \
                 omit for the default (10m), or 0 to wait forever"
            ),
        });
    }
    errs.extend(validate_access_modes(
        "spec.staging.accessModes",
        &st.access_modes,
    ));
    // A ReadOnlyMany staged PVC cannot be mounted read-write: the kubelet fails the
    // mount and the backup dies at run time with an opaque error. Catch it here.
    if st.access_modes.contains(&PvcAccessMode::ReadOnlyMany)
        && let Some(i) = spec
            .sources
            .iter()
            .position(|s| !crate::snapshot_policy::source_read_only(s))
    {
        errs.push(ValidationError::InvalidFieldValue {
            field: format!("spec.sources[{i}].readOnly"),
            reason: "readOnly: false cannot work when spec.staging.accessModes is \
                     [ReadOnlyMany], so the backup would never start. Fix: drop ReadOnlyMany, \
                     or drop readOnly: false"
                .to_string(),
        });
    }
    let overrides: Vec<&str> = [
        (
            "spec.staging.storageClassName",
            st.storage_class_name.is_some(),
        ),
        ("spec.staging.accessModes", !st.access_modes.is_empty()),
    ]
    .into_iter()
    .filter_map(|(name, present)| present.then_some(name))
    .collect();
    if overrides.is_empty() {
        return errs;
    }
    let overrides = overrides.join(" / ");
    match spec.copy_method {
        CopyMethod::Direct => errs.push(ValidationError::InvalidFieldValue {
            field: overrides.clone(),
            reason: "copyMethod: Direct has no staged PVC to override. Fix: remove the \
                     override(s), or use copyMethod: Snapshot/Clone"
                .to_string(),
        }),
        CopyMethod::Snapshot | CopyMethod::Clone => {}
    }
    if spec.sources.iter().any(|s| s.nfs.is_some()) {
        errs.push(ValidationError::InvalidFieldValue {
            field: overrides.clone(),
            reason: "an NFS source is never staged, so there is no staged PVC to override. \
                     Fix: remove the override(s)"
                .to_string(),
        });
    }
    if spec.sources.iter().any(|s| s.stream.is_some()) {
        errs.push(ValidationError::InvalidFieldValue {
            field: overrides.clone(),
            reason: "a stream source mounts no volume, so there is no staged PVC to override. \
                     Fix: remove the override(s)"
                .to_string(),
        });
    }
    // NOTE: `pvcSelector` sources used to be rejected here on the grounds that
    // they "are not CSI-staged". That was true only because the selector was
    // never implemented (#346). Each expanded member is now an ordinary
    // single-PVC Snapshot and stages exactly like one, so the override applies
    // to every member and the rejection is gone.
    let _ = overrides;
    errs
}

/// Validate one hook entry — the controller executes these with the SAME parsers
/// (Go-style `timeout`, URL/method for `httpRequest`), so a value admitted here
/// can never fail to parse at run time. Exhaustive over [`Hook`].
fn validate_hook(list: &str, index: usize, hook: &Hook) -> ValidationResult {
    let field = |leaf: &str| format!("spec.hooks.{list}[{index}].{leaf}");
    let check_timeout = |leaf: &str, t: Option<&str>| -> ValidationResult {
        if let Some(t) = t
            && crate::duration::parse_go_duration(t).is_none()
        {
            return Err(ValidationError::InvalidFieldValue {
                field: field(leaf),
                reason: format!(
                    "{t:?} is not a valid duration. Fix: use a Go-style duration like 90s or 2m"
                ),
            });
        }
        Ok(())
    };
    match hook {
        Hook::WorkloadExec(h) => {
            if h.command.is_empty() {
                return Err(ValidationError::MissingRequiredField {
                    field: field("workloadExec.command"),
                });
            }
            check_timeout("workloadExec.timeout", h.timeout.as_deref())
        }
        Hook::RunJob(h) => check_timeout("runJob.timeout", h.timeout.as_deref()),
        Hook::HttpRequest(h) => {
            if !(h.url.starts_with("http://") || h.url.starts_with("https://")) {
                return Err(ValidationError::InvalidFieldValue {
                    field: field("httpRequest.url"),
                    reason: format!(
                        "{:?} must be an absolute http:// or https:// URL the controller can \
                         reach (e.g. http://notifier.tools.svc:8080/fire)",
                        h.url
                    ),
                });
            }
            if let Some(m) = &h.method {
                const METHODS: [&str; 7] =
                    ["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"];
                if !METHODS.contains(&m.to_ascii_uppercase().as_str()) {
                    return Err(ValidationError::InvalidFieldValue {
                        field: field("httpRequest.method"),
                        reason: format!(
                            "{m:?} is not an HTTP method; use one of GET, POST (default), PUT, \
                             PATCH, DELETE, HEAD, OPTIONS"
                        ),
                    });
                }
            }
            if let Some(e) = validate_http_hook_headers(h, list, index) {
                return Err(e);
            }
            check_timeout("httpRequest.timeout", h.timeout.as_deref())
        }
    }
}

/// Largest header name `http::HeaderName::from_bytes` will accept: `http` 1.4.2
/// rejects anything longer via its `MAX_HEADER_NAME_LEN = 65535` guard. Mirrored
/// EXACTLY here so a name admitted at the webhook can never fail to parse at run
/// time (the branch's "anything admitted never fails at runtime" guarantee).
const MAX_HEADER_NAME_LEN: usize = 65_535;

/// RFC 7230 token — exactly the character set `http::HeaderName` accepts. This
/// is the token check only; `http` ALSO caps the byte length at
/// [`MAX_HEADER_NAME_LEN`], enforced separately in [`header_name_error`].
fn is_valid_header_name(name: &str) -> bool {
    !name.is_empty()
        && name.bytes().all(|b| {
            matches!(b,
                b'a'..=b'z' | b'A'..=b'Z' | b'0'..=b'9'
                | b'!' | b'#' | b'$' | b'%' | b'&' | b'\'' | b'*' | b'+'
                | b'-' | b'.' | b'^' | b'_' | b'`' | b'|' | b'~')
        })
}

/// Validate a single header name against everything `http::HeaderName` enforces
/// — the RFC 7230 token set AND the [`MAX_HEADER_NAME_LEN`] byte cap — returning
/// the first problem as a ready-to-return [`ValidationError`]. A length problem
/// is not a token problem, so each carries its own what/why/fix message. Split
/// out of [`validate_http_hook_headers`] to keep that function's branch count
/// (and cognitive complexity) unchanged.
fn header_name_error(name: &str, list: &str, i: usize, j: usize) -> Option<ValidationError> {
    if !is_valid_header_name(name) {
        return Some(ValidationError::InvalidFieldValue {
            field: format!("spec.hooks.{list}[{i}].httpRequest.headers[{j}].name"),
            reason: format!(
                "{name:?} is not a valid HTTP header name. Fix: use only letters, digits, and \
                 !#$%&'*+-.^_`|~ (no spaces)"
            ),
        });
    }
    if name.len() > MAX_HEADER_NAME_LEN {
        return Some(ValidationError::InvalidFieldValue {
            field: format!("spec.hooks.{list}[{i}].httpRequest.headers[{j}].name"),
            reason: format!(
                "header name is {} bytes; the maximum is {MAX_HEADER_NAME_LEN}",
                name.len()
            ),
        });
    }
    None
}

/// Field-content bytes `http::HeaderValue::from_str` accepts: HTAB, or any
/// byte >= 0x20 except DEL (0x7F). Blocks CR/LF header injection.
fn is_valid_header_value(value: &str) -> bool {
    value
        .bytes()
        .all(|b| b == b'\t' || (b >= 0x20 && b != 0x7f))
}

/// True when the URL's authority component carries `user[:pass]@` credentials.
fn url_has_userinfo(url: &str) -> bool {
    let rest = url.split_once("://").map_or(url, |(_, r)| r);
    let authority = rest.split(['/', '?', '#']).next().unwrap_or("");
    authority.contains('@')
}

/// Validate an `httpRequest` hook's headers so a value admitted here can never
/// fail `http::HeaderName`/`HeaderValue` parsing at run time. Extracted from
/// [`validate_hook`] to keep that match arm below the cognitive-complexity
/// ratchet. Returns the first problem found, matching the arm's early-return
/// style; the field paths mirror the CRD shape
/// (`spec.hooks.{list}[{i}].httpRequest.headers[{j}].{name|value}`).
fn validate_http_hook_headers(
    h: &HttpRequestHook,
    list: &str,
    i: usize,
) -> Option<ValidationError> {
    let mut seen: Vec<String> = Vec::new();
    for (j, header) in h.headers.iter().enumerate() {
        if let Some(e) = header_name_error(&header.name, list, i, j) {
            return Some(e);
        }
        if !is_valid_header_value(&header.value) {
            return Some(ValidationError::InvalidFieldValue {
                field: format!("spec.hooks.{list}[{i}].httpRequest.headers[{j}].value"),
                reason: "header values cannot contain control characters (including CR/LF). \
                         Fix: put multi-line data in `body`"
                    .into(),
            });
        }
        let lower = header.name.to_ascii_lowercase();
        if seen.contains(&lower) {
            return Some(ValidationError::InvalidFieldValue {
                field: format!("spec.hooks.{list}[{i}].httpRequest.headers[{j}].name"),
                reason: format!(
                    "duplicate header {:?}. Fix: set it once, with comma-separated values if \
                     needed",
                    header.name
                ),
            });
        }
        seen.push(lower);
    }
    if url_has_userinfo(&h.url)
        && h.headers
            .iter()
            .any(|hd| hd.name.eq_ignore_ascii_case("authorization"))
    {
        return Some(ValidationError::InvalidFieldValue {
            field: format!("spec.hooks.{list}[{i}].httpRequest.headers"),
            reason: "an Authorization header conflicts with credentials in the URL \
                     (user:pass@…). Fix: use one or the other"
                .into(),
        });
    }
    None
}

/// Validate a `Snapshot` spec for a given origin, accumulating all problems.
///
/// `origin` is supplied by the caller because the canonical value lives in
/// `status.origin` / the `kopiur.home-operations.com/origin` label, not in
/// `spec` (ADR §3.4). `None` means the row carries an origin marker the caller
/// could not parse (`Origin::parse` returned `None`): the origin-GATED rules
/// (deletionPolicy legality, onScheduleDelete) are **skipped, not failed** —
/// refusing would wedge metadata-only writes (finalizer removal above all) on
/// rows written under a newer operator during version skew, while skipping is
/// safe because the controller's conservative resolution of an unparseable
/// origin (forced `Retain`, never scheduled, never run) makes both gated
/// fields inert at reconcile time. The origin-independent rules still run.
pub fn validate_backup(spec: &SnapshotSpec, origin: Option<Origin>) -> Vec<ValidationError> {
    let mut errs = Vec::new();
    if let Some(origin) = origin {
        if let Err(e) = validate_backup_deletion_policy(origin, spec.deletion_policy) {
            errs.push(e);
        }
        if let Err(e) = validate_backup_on_schedule_delete(origin, spec.on_schedule_delete) {
            errs.push(e);
        }
    }
    if let Some(fp) = &spec.failure_policy
        && let Err(e) = validate_failure_policy(fp, "Snapshot")
    {
        errs.push(e);
    }
    errs.extend(validate_snapshot_tags(spec.tags.as_ref()));
    if let Some(m) = &spec.mover {
        // Same two rules the recipe layer gets, because this is the same layer one
        // step up (#464): a per-run `mover` is merged OVER the policy's, so anything
        // inadmissible there is inadmissible here.
        //
        // `inheritSecurityContextFrom.snapshot` replays the identity RECORDED on a
        // backup; a backup has no recorded identity to replay — it is the run that
        // records one.
        if let Err(e) = forbid_snapshot_inherit(
            m,
            "snapshot",
            "is restore-only. Fix: use pvcConsumer or workloadSelector for a backup",
        ) {
            errs.push(e);
        }
        if let Err(e) = validate_mover(m, "Snapshot mover") {
            errs.push(e);
        }
        // NOT rejected here: `inheritSecurityContextFrom.pvcConsumer`. Validators are
        // spec-only and client-free, so this one cannot see the referenced
        // `SnapshotPolicy`'s sources to know whether a source PVC exists to derive a
        // consumer from. Refusing on that guess would reject the common, correct case
        // (a one-shot against an ordinary `source.pvc` policy); admitting it lets the
        // controller resolve it and, when it cannot, park the run on a condition that
        // names the reason — which is strictly more informative than an admission
        // error built on missing information.
    }
    errs
}

/// At most this many user tags per `Snapshot` — unbounded user tags would
/// inflate every kopia manifest AND the catalog result wire.
pub const MAX_SNAPSHOT_TAGS: usize = 10;
/// Longest admissible user tag key, in bytes.
pub const MAX_SNAPSHOT_TAG_KEY_LEN: usize = 63;
/// Longest admissible user tag value, in bytes.
pub const MAX_SNAPSHOT_TAG_VALUE_LEN: usize = 256;

/// Why one `spec.tags` key/value pair is invalid, or `None` when it is clean.
///
/// One predicate, two callers (the shared-validator pattern): the admission
/// validator ([`validate_snapshot_tags`]) rejects NEW objects with these
/// reasons, and the controller's build path defensively SKIPS (warn, never
/// fail) the same keys on already-stored pre-feature objects.
pub fn snapshot_tag_error(key: &str, value: &str) -> Option<String> {
    if key.is_empty() {
        return Some("tag keys must be non-empty; remove the empty key".to_string());
    }
    if key.contains(':') {
        return Some(format!(
            "tag key {key:?} contains a colon. Fix: use a key without `:`"
        ));
    }
    if key.starts_with("kopiur") {
        return Some(format!(
            "tag key {key:?} uses the reserved `kopiur` prefix. Fix: use a key that does not \
             start with `kopiur`"
        ));
    }
    if key.len() > MAX_SNAPSHOT_TAG_KEY_LEN {
        return Some(format!(
            "tag key is {} bytes; the maximum is {MAX_SNAPSHOT_TAG_KEY_LEN}",
            key.len()
        ));
    }
    if value.len() > MAX_SNAPSHOT_TAG_VALUE_LEN {
        return Some(format!(
            "tag value is {} bytes; the maximum is {MAX_SNAPSHOT_TAG_VALUE_LEN}",
            value.len()
        ));
    }
    None
}

/// Validate `Snapshot.spec.tags` (admission): every key/value must pass
/// [`snapshot_tag_error`] and the map is bounded to [`MAX_SNAPSHOT_TAGS`]
/// entries. Accumulates every problem, one error per offending tag.
pub fn validate_snapshot_tags(
    tags: Option<&std::collections::BTreeMap<String, String>>,
) -> Vec<ValidationError> {
    let mut errs = Vec::new();
    let Some(tags) = tags else {
        return errs;
    };
    if tags.len() > MAX_SNAPSHOT_TAGS {
        errs.push(ValidationError::InvalidFieldValue {
            field: "spec.tags".to_string(),
            reason: format!(
                "{} tags; at most {MAX_SNAPSHOT_TAGS} are allowed per Snapshot",
                tags.len()
            ),
        });
    }
    for (key, value) in tags {
        if let Some(reason) = snapshot_tag_error(key, value) {
            errs.push(ValidationError::InvalidFieldValue {
                field: format!("spec.tags[{key:?}]"),
                reason,
            });
        }
    }
    errs
}

/// `origin: discovered` Snapshots carry an empty spec; a stamped cascade
/// policy on one is meaningless (their owner is a repository, not a schedule)
/// and forbidden, like a non-Retain deletionPolicy. `origin: adopted` is
/// forbidden for the same reason: an adopted row's owner is the
/// `SnapshotPolicy` it was re-attached to, never a `SnapshotSchedule`. So is
/// `origin: replicated`: a copy CR belongs to its `SnapshotReplication`, and
/// no `SnapshotSchedule` ever fires (or cascades onto) one.
pub fn validate_backup_on_schedule_delete(
    origin: Origin,
    value: Option<ScheduleDeletePolicy>,
) -> ValidationResult {
    match origin {
        Origin::Discovered | Origin::Adopted | Origin::Replicated => match value {
            None => Ok(()),
            Some(v) => Err(ValidationError::DiscoveredCannotSetOnScheduleDelete {
                origin: origin.label_value(),
                got: format!("{v:?}"),
            }),
        },
        Origin::Scheduled | Origin::Manual => Ok(()),
    }
}

/// Exactly one of `policyRef` / `policySelector` is set on a `SnapshotSchedule`
/// (ADR-0005 §10). Neither ⇒ `MissingRequiredField`; both ⇒ `MutuallyExclusive`.
/// Pure so the XOR decision is unit-tested directly.
pub fn validate_schedule_policy_target(spec: &SnapshotScheduleSpec) -> ValidationResult {
    match (spec.policy_ref.is_some(), spec.policy_selector.is_some()) {
        (true, true) => Err(ValidationError::MutuallyExclusive {
            a: "policyRef".to_string(),
            b: "policySelector".to_string(),
            context: "SnapshotSchedule".to_string(),
        }),
        (false, false) => Err(ValidationError::MissingRequiredField {
            field: "exactly one of spec.policyRef or spec.policySelector".to_string(),
        }),
        _ => Ok(()),
    }
}

/// Validate a `SnapshotSchedule` spec, accumulating all problems.
pub fn validate_backup_schedule(spec: &SnapshotScheduleSpec) -> Vec<ValidationError> {
    let mut errs = Vec::new();
    if let Err(e) = validate_schedule_policy_target(spec) {
        errs.push(e);
    }
    if let Err(e) = validate_cron(&spec.schedule.cron) {
        errs.push(e);
    }
    if let Err(e) = validate_timezone(spec.schedule.timezone.as_deref()) {
        errs.push(e);
    }
    if let Err(e) = validate_jitter("spec.schedule.jitter", spec.schedule.jitter.as_deref()) {
        errs.push(e);
    }
    errs
}