assay-cli 6.5.0

//! `assay.enforcement_health.v1` — the enforcement-truth carrier for the Landlock TCP-connect
//! port-allowlist domain.
//!
//! This is a SEPARATE, explicit version bump from `assay.enforcement_health.v0` (the connect4/eBPF
//! egress carrier in `cli::commands::monitor_next::enforcement_health`). v0 is left untouched and
//! remains valid; a consumer reads v0 and v1 additively. The two carriers are different shapes
//! because they report different enforcement domains: v0 carries connect4 attach/count truth, v1
//! carries the Landlock ruleset/`restrict_self`/real-block truth.
//!
//! Scope: the carrier, its committed fixtures, and the Landlock sandbox producer. The fixtures pin
//! the exact bytes a consumer (e.g. an external review tool) reconstructs against, the same
//! producer-agnostic discipline the rest of the project uses, and the producer in
//! `cli::commands::sandbox::child` emits it on both the active and failed paths.
//!
//! This header read "No producer wires it up yet" until 2026-08-06, long after `child.rs` began
//! emitting it. It was believed on sight while compiling a spec-ahead-of-implementation list and the
//! row it produced was false, which is why that list warns that a stale entry is worse than none.
//!
//! Honesty rules baked into the shape:
//! - `status` is `active` or `failed` only. There is no `not_applicable`, and no `absent`.
//!   Presence of a v1 artifact is producer-reported observation for that episode. Absence of a
//!   v1 artifact is no artifact and no claim: it cannot prove that enforcement was not requested
//!   (no output flag, a non-Linux build, or a clap-rejected invocation all produce absence).
//!   A requested-but-unwritable artifact must be a hard error at the producer. This is not v0
//!   parity: v0 serializes `absent` inside the artifact; v1 has no such status.
//! - `probe` is always present and is `null` when no real-block probe was run, so a consumer can
//!   distinguish "schema knows about probe, none happened" from an older shape that lacked the field.
//!   `active` without a probe means the ruleset was applied (`no_new_privs` + `restrict_self`
//!   confirmed); `active` with a probe additionally means a denied connect was really blocked before
//!   the listener was reached.
//! - `failure.reason_code` is a machine-readable enum, never prose only.

use serde::{Deserialize, Serialize};

pub const SCHEMA_V1: &str = "assay.enforcement_health.v1";

/// The enforcement domain. Mechanism-accurate (not IPv4-specific): Landlock is not fundamentally
/// IPv4-only, so the transport lives under `probe`, not in the scope string.
pub const SCOPE_TCP_CONNECT_LANDLOCK_PORT: &str = "tcp_connect_landlock_port";

/// Whether enforcement was active or failed to install. No `not_applicable` and no `absent`
/// variant. Artifact absence is no artifact and no claim; it does not prove that enforcement
/// was not requested.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Status {
    /// The Landlock ruleset was applied: `no_new_privs` set and `restrict_self` confirmed.
    Active,
    /// Enforcement was requested but could not be installed (carries `failure.reason_code`).
    Failed,
}

/// How hard the enforcement point is to route around. Same axis and same value names as the v0
/// carrier's `EnforcementClass`, which is deliberate: a consumer comparing a v0 and a v1 artifact is
/// comparing two enforcement domains and must not have to learn two vocabularies to do it.
///
/// Adopted from `draft-schrock-ep-authorization-receipts-07` §9, with its rule: a record MUST NOT
/// state a stronger class than deployed.
///
/// **Landlock earns `Strong` for a reason worth stating.** `restrict_self` is applied by the process
/// to itself under `no_new_privs` and cannot afterwards be removed — not by the process, not by
/// regaining privilege. So unlike an interception layer, there is no "call it another way" for a
/// subject already inside the restriction. What remains is never entering it, which is a producer
/// question rather than a bypass.
///
/// **Deliberately not keyed to the probe.** `active` with a probe additionally demonstrates a real
/// block, and that is a stronger *evidentiary* statement — but this axis grades the architecture,
/// not the run. The probe carries demonstration on its own field, and folding two questions into one
/// value is how a field starts meaning two things.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum EnforcementClass {
    /// No enforcement claim. Nothing was installed, so nothing was refused.
    Basic = 0,
    /// An interception layer a subject can reach past by taking another route.
    Standard = 1,
    /// An enforcement point the subject cannot route around from inside it.
    Strong = 2,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Mechanism {
    Landlock,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PolicySemantics {
    /// Landlock network rules are allowlist-only: a handled right is denied by default, and rules
    /// grant exceptions for specific TCP ports. There is no endpoint/CIDR/negative-deny.
    Allowlist,
}

/// Machine-readable failure reasons. Only states that can actually occur on the Landlock TCP-connect
/// path (mirrors the PR2 usability-smoke failure taxonomy plus the ABI gate).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ReasonCode {
    /// Landlock ABI < 4: `LANDLOCK_ACCESS_NET_CONNECT_TCP` is unavailable.
    LandlockAbiTooOld,
    /// Landlock is built in but disabled at boot (`EOPNOTSUPP`).
    LandlockDisabled,
    /// `PR_SET_NO_NEW_PRIVS` could not be set, so an unprivileged `restrict_self` cannot apply.
    NoNewPrivsFailed,
    /// `landlock_restrict_self` returned an error or did not enforce.
    RestrictSelfFailed,
    /// The ruleset could not be built (handle-access or add-rule failed).
    RulesetBuildFailed,
    /// The network policy is not expressible as a Landlock TCP-connect allowlist (IP/CIDR, host,
    /// non-TCP protocol, port range, deny rule, or port 0). The specific entries are in `detail`.
    PolicyNotExpressible,
}

/// The Landlock-specific evidence block. Fields that do not apply to a given status are omitted
/// (`active` carries the applied-ruleset shape; `failed` carries the partial-capability shape), so
/// the serialized bytes match the domain exactly.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct LandlockBlock {
    /// The runtime Landlock ABI observed (may be < 4 on the failed-too-old path).
    pub abi: u32,
    /// Present on the failed path: whether ABI >= 4 (`CONNECT_TCP`) was available.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub net_connect_tcp_supported: Option<bool>,
    /// Present on the active path: the handled network access rights (e.g. `["connect_tcp"]`).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub handled_access_net: Option<Vec<String>>,
    /// Present on the active path: the explicit allowed TCP-connect ports (allowlist, never a range).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub allowed_connect_tcp_ports: Option<Vec<u16>>,
    /// Whether `PR_SET_NO_NEW_PRIVS` was confirmed set before `restrict_self`. A
    /// `restrict_self_confirmed = true` without this is too weak for this route.
    pub no_new_privs_confirmed: bool,
    /// Whether `landlock_restrict_self` was confirmed (by the enforcing child, not assumed by parent).
    pub restrict_self_confirmed: bool,
    /// Whether this kernel lets a restricted process shed its own Landlock restrictions, measured.
    ///
    /// `restrictions_held` | `restrictions_shed` | `inconclusive`, absent when the probe was not run.
    ///
    /// This exists because ADR-045 permits `aeeStillArmed` to rest on Landlock's irrevocability, and
    /// that invariant has a documented exception: CVE-2024-42318 let a process with `fork()` and
    /// `keyctl()` drop every Landlock restriction on itself, from 5.13 until the fix in v6.11-rc1.
    /// `abi` cannot stand in for it -- ABI 4 arrives at kernel 6.7, four releases before the fix --
    /// and a kernel-version comparison cannot see a distribution backport, so the property is
    /// measured on the running kernel instead of inferred. See `backend::landlock_shedding_probe`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub restriction_shedding: Option<String>,
}

/// The optional real-block probe. Its presence upgrades the claim from "ruleset applied" to "a denied
/// connect was really blocked before the listener was reached". `blocked_errno` is the
/// mechanism-specific signal (Landlock denies connect with `EACCES`); weak signals such as a timeout
/// or `ECONNREFUSED` never count as a block.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Probe {
    pub kind: String,
    pub transport: String,
    pub blocked_action: String,
    pub blocked_port: u16,
    pub blocked_errno: String,
    pub listener_reached: bool,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Failure {
    pub reason_code: ReasonCode,
    pub detail: String,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct EnforcementHealthV1 {
    pub schema: String,
    pub status: Status,
    pub mechanism: Mechanism,
    pub scope: String,
    pub policy_semantics: PolicySemantics,
    /// What the enforcement is worth, derived from whether the ruleset actually applied.
    pub enforcement_class: EnforcementClass,
    /// Present only on the failed path.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub failure: Option<Failure>,
    pub landlock: LandlockBlock,
    /// Always serialized; `null` when no real-block probe ran (null-over-absent, by design).
    pub probe: Option<Probe>,
    pub non_claims: Vec<String>,
}

/// The standing non-claims for the Landlock TCP-connect port-allowlist domain.
fn landlock_non_claims() -> Vec<String> {
    [
        "no ip or cidr enforcement",
        "no hostname enforcement",
        "no destination identity enforcement",
        "no udp or quic enforcement",
        "no http or tls route policy",
        "not a replacement for cgroup/connect4 endpoint enforcement",
    ]
    .iter()
    .map(|s| s.to_string())
    .collect()
}

impl EnforcementHealthV1 {
    /// Active Landlock TCP-connect enforcement: the ruleset was applied (`no_new_privs` set and
    /// `restrict_self` confirmed). `probe` is `Some` only when a real-block probe was actually run;
    /// `None` records "ruleset applied, no real-block claim".
    #[must_use]
    pub fn landlock_active(
        abi: u32,
        allowed_ports: Vec<u16>,
        probe: Option<Probe>,
        restriction_shedding: Option<String>,
    ) -> Self {
        Self {
            schema: SCHEMA_V1.to_string(),
            status: Status::Active,
            mechanism: Mechanism::Landlock,
            scope: SCOPE_TCP_CONNECT_LANDLOCK_PORT.to_string(),
            policy_semantics: PolicySemantics::Allowlist,
            // Derived from the constructor that recorded the install, never passed in, so a caller
            // cannot state a stronger class than deployed.
            enforcement_class: EnforcementClass::Strong,
            failure: None,
            landlock: LandlockBlock {
                abi,
                net_connect_tcp_supported: None,
                handled_access_net: Some(vec!["connect_tcp".to_string()]),
                allowed_connect_tcp_ports: Some(allowed_ports),
                no_new_privs_confirmed: true,
                restrict_self_confirmed: true,
                restriction_shedding,
            },
            probe,
            non_claims: landlock_non_claims(),
        }
    }

    /// Failed Landlock TCP-connect enforcement: requested but not installed. Carries the
    /// machine-readable reason and the partial-capability truth.
    #[must_use]
    pub fn landlock_failed(
        abi: u32,
        reason_code: ReasonCode,
        detail: impl Into<String>,
        net_connect_tcp_supported: bool,
        no_new_privs_confirmed: bool,
    ) -> Self {
        Self {
            schema: SCHEMA_V1.to_string(),
            status: Status::Failed,
            mechanism: Mechanism::Landlock,
            scope: SCOPE_TCP_CONNECT_LANDLOCK_PORT.to_string(),
            policy_semantics: PolicySemantics::Allowlist,
            // Requested and not installed. On this axis that is indistinguishable from never asking:
            // nothing attached, so nothing was refused. The distinction lives in `status` and
            // `failure.reason_code`, which is where a consumer should read it.
            enforcement_class: EnforcementClass::Basic,
            failure: Some(Failure {
                reason_code,
                detail: detail.into(),
            }),
            landlock: LandlockBlock {
                abi,
                net_connect_tcp_supported: Some(net_connect_tcp_supported),
                handled_access_net: None,
                allowed_connect_tcp_ports: None,
                no_new_privs_confirmed,
                restrict_self_confirmed: false,
                // Nothing was armed, so there is nothing to shed. Absent rather than
                // "inconclusive": the probe was not attempted, and a record that says
                // "could not tell" implies it tried.
                restriction_shedding: None,
            },
            probe: None,
            non_claims: landlock_non_claims(),
        }
    }

    pub fn write_to(&self, path: &std::path::Path) -> std::io::Result<()> {
        let json = serde_json::to_string_pretty(self)
            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
        std::fs::write(path, json)
    }
}

#[cfg(test)]
mod fixture_values {
    //! The canonical example values the committed fixtures are generated from. Kept next to the
    //! carrier so the fixtures and the typed shape can never drift: `emit_fixtures` writes these to
    //! disk, and the round-trip tests read them back and assert byte-stability.
    use super::*;

    fn base_non_claims() -> Vec<String> {
        [
            "no ip or cidr enforcement",
            "no hostname enforcement",
            "no destination identity enforcement",
            "no udp or quic enforcement",
            "no http or tls route policy",
            "not a replacement for cgroup/connect4 endpoint enforcement",
        ]
        .iter()
        .map(|s| s.to_string())
        .collect()
    }

    pub fn active_with_probe() -> EnforcementHealthV1 {
        EnforcementHealthV1 {
            schema: SCHEMA_V1.to_string(),
            status: Status::Active,
            mechanism: Mechanism::Landlock,
            scope: SCOPE_TCP_CONNECT_LANDLOCK_PORT.to_string(),
            policy_semantics: PolicySemantics::Allowlist,
            // Derived from the constructor that recorded the install, never passed in, so a caller
            // cannot state a stronger class than deployed.
            enforcement_class: EnforcementClass::Strong,
            failure: None,
            landlock: LandlockBlock {
                abi: 4,
                net_connect_tcp_supported: None,
                handled_access_net: Some(vec!["connect_tcp".to_string()]),
                allowed_connect_tcp_ports: Some(vec![443]),
                no_new_privs_confirmed: true,
                restrict_self_confirmed: true,
                restriction_shedding: Some("restrictions_held".to_string()),
            },
            probe: Some(Probe {
                kind: "real_block".to_string(),
                transport: "ipv4".to_string(),
                blocked_action: "tcp_connect".to_string(),
                blocked_port: 4444,
                blocked_errno: "EACCES".to_string(),
                listener_reached: false,
            }),
            non_claims: base_non_claims(),
        }
    }

    pub fn active_no_probe() -> EnforcementHealthV1 {
        EnforcementHealthV1 {
            probe: None,
            ..active_with_probe()
        }
    }

    pub fn failed() -> EnforcementHealthV1 {
        EnforcementHealthV1 {
            schema: SCHEMA_V1.to_string(),
            status: Status::Failed,
            mechanism: Mechanism::Landlock,
            scope: SCOPE_TCP_CONNECT_LANDLOCK_PORT.to_string(),
            policy_semantics: PolicySemantics::Allowlist,
            // Requested and not installed. On this axis that is indistinguishable from never asking:
            // nothing attached, so nothing was refused. The distinction lives in `status` and
            // `failure.reason_code`, which is where a consumer should read it.
            enforcement_class: EnforcementClass::Basic,
            failure: Some(Failure {
                reason_code: ReasonCode::LandlockAbiTooOld,
                detail: "Landlock ABI 4 is required for TCP connect port allowlists".to_string(),
            }),
            landlock: LandlockBlock {
                abi: 3,
                net_connect_tcp_supported: Some(false),
                handled_access_net: None,
                allowed_connect_tcp_ports: None,
                no_new_privs_confirmed: false,
                restrict_self_confirmed: false,
                restriction_shedding: None,
            },
            probe: None,
            non_claims: base_non_claims(),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::path::Path;

    const FIXTURE_DIR: &str = "tests/fixtures/enforcement_health/v1";

    fn fixture_path(name: &str) -> std::path::PathBuf {
        Path::new(env!("CARGO_MANIFEST_DIR"))
            .join(FIXTURE_DIR)
            .join(name)
    }

    /// Regenerates the committed fixtures from the canonical typed values. Run explicitly:
    /// `cargo test -p assay-cli --bin assay enforcement_health_v1::tests::emit_fixtures -- --ignored`.
    /// Not part of the normal run; the round-trip tests below are the guard that the committed bytes
    /// match the typed shape.
    #[test]
    #[ignore = "generator; run with --ignored to regenerate fixtures"]
    fn emit_fixtures() {
        for (name, value) in [
            (
                "active_with_probe.json",
                fixture_values::active_with_probe(),
            ),
            ("active_no_probe.json", fixture_values::active_no_probe()),
            ("failed.json", fixture_values::failed()),
        ] {
            let mut json = serde_json::to_string_pretty(&value).unwrap();
            json.push('\n');
            std::fs::write(fixture_path(name), json).unwrap();
        }
    }

    fn load_fixture(name: &str) -> (String, EnforcementHealthV1) {
        let path = Path::new(env!("CARGO_MANIFEST_DIR"))
            .join(FIXTURE_DIR)
            .join(name);
        let bytes = std::fs::read_to_string(&path)
            .unwrap_or_else(|e| panic!("read fixture {}: {e}", path.display()));
        let value: EnforcementHealthV1 = serde_json::from_str(&bytes)
            .unwrap_or_else(|e| panic!("parse fixture {}: {e}", path.display()));
        (bytes, value)
    }

    /// Each committed fixture parses into the typed carrier and re-serializes to byte-identical JSON.
    /// This pins the exact contract bytes a second implementation reconstructs against.
    fn assert_round_trip(name: &str) -> EnforcementHealthV1 {
        let (bytes, value) = load_fixture(name);
        let reserialized = serde_json::to_string_pretty(&value).unwrap();
        assert_eq!(
            reserialized.trim_end(),
            bytes.trim_end(),
            "fixture {name} is not byte-stable through the typed carrier"
        );
        value
    }

    #[test]
    fn an_applied_ruleset_earns_strong_and_a_failed_one_earns_nothing() {
        // Schrock's rule: a record MUST NOT state a stronger class than deployed. Derived inside the
        // constructor from the install truth, so there is no argument to overstate it through.
        assert_eq!(
            fixture_values::active_no_probe().enforcement_class,
            EnforcementClass::Strong
        );
        assert_eq!(
            fixture_values::failed().enforcement_class,
            EnforcementClass::Basic,
            "requested and not installed refuses nothing, so it claims nothing"
        );
    }

    #[test]
    fn the_probe_does_not_change_the_class() {
        // The decision this axis needed. `active` with a probe additionally demonstrates a real
        // block, which is a stronger EVIDENTIARY statement — but the class grades the architecture,
        // not the run. Landlock is equally unroutable-around whether or not this run happened to
        // prove it. Folding demonstration into the class is how one field starts meaning two things,
        // and `probe` already carries it.
        assert_eq!(
            fixture_values::active_with_probe().enforcement_class,
            fixture_values::active_no_probe().enforcement_class
        );
        assert_ne!(
            fixture_values::active_with_probe().probe,
            fixture_values::active_no_probe().probe,
            "the demonstration distinction must still be visible, on its own field"
        );
    }

    #[test]
    fn v0_and_v1_grade_the_same_axis_with_the_same_names() {
        // Two enforcement domains, one vocabulary. A consumer comparing a connect4 artifact with a
        // Landlock artifact should not have to learn two spellings of the same idea. This asserts the
        // wire values, which is what a consumer actually reads, rather than the Rust types, which are
        // deliberately separate per carrier.
        let v1 = serde_json::to_value(fixture_values::active_no_probe()).unwrap();
        assert_eq!(v1["enforcement_class"], serde_json::json!("strong"));
        let v0 = serde_json::to_value(
            crate::cli::commands::monitor::monitor_next::enforcement_health::EnforcementHealth::active(
                crate::cli::commands::monitor::monitor_next::enforcement_health::SCOPE_IPV4_TCP_CONNECT,
                0,
                0,
            ),
        )
        .unwrap();
        assert_eq!(v0["enforcement_class"], v1["enforcement_class"]);
    }

    #[test]
    fn active_with_probe_round_trips_and_claims_real_block() {
        let v = assert_round_trip("active_with_probe.json");
        assert_eq!(v.schema, SCHEMA_V1);
        assert_eq!(v.status, Status::Active);
        assert!(v.failure.is_none());
        let probe = v.probe.expect("active_with_probe carries a probe");
        assert_eq!(probe.blocked_errno, "EACCES");
        assert!(
            !probe.listener_reached,
            "a real block must not reach the listener"
        );
        assert!(v.landlock.no_new_privs_confirmed);
        assert!(v.landlock.restrict_self_confirmed);
    }

    #[test]
    fn active_no_probe_round_trips_with_null_probe() {
        let v = assert_round_trip("active_no_probe.json");
        assert_eq!(v.status, Status::Active);
        assert!(
            v.probe.is_none(),
            "active without a probe carries probe: null (ruleset applied, no real-block claim)"
        );
        assert!(v.landlock.restrict_self_confirmed);
    }

    #[test]
    fn failed_round_trips_with_machine_readable_reason() {
        let v = assert_round_trip("failed.json");
        assert_eq!(v.status, Status::Failed);
        let failure = v.failure.expect("failed carries a failure block");
        assert_eq!(failure.reason_code, ReasonCode::LandlockAbiTooOld);
        assert!(!v.landlock.restrict_self_confirmed);
        assert_eq!(v.landlock.net_connect_tcp_supported, Some(false));
    }

    #[test]
    fn v1_status_has_no_not_applicable_or_absent() {
        // The carrier vocabulary is only active/failed. Missing artifact is no claim, not a
        // v0-style `absent` / not-requested status. not_applicable does not exist in v1.
        assert!(serde_json::from_str::<Status>("\"active\"").is_ok());
        assert!(serde_json::from_str::<Status>("\"failed\"").is_ok());
        assert!(serde_json::from_str::<Status>("\"absent\"").is_err());
        assert!(serde_json::from_str::<Status>("\"not_applicable\"").is_err());
    }

    #[test]
    fn schema_id_is_an_explicit_version_bump_from_v0() {
        assert_eq!(SCHEMA_V1, "assay.enforcement_health.v1");
        assert_ne!(
            SCHEMA_V1, "assay.enforcement_health.v0",
            "v1 must be a distinct schema id from v0 (explicit version bump, not a silent reshape)"
        );
    }
}