harmony/fleet/harmony-fleet-e2e/tests/vm_secret_access.rs

//! Full-stack device secret access (ADR-025), end to end with nothing
//! mocked:
//!
//! 1. the agent on a VM connects to NATS **through the auth callout**
//!    with its Zitadel machine identity (the readiness ping proves the
//!    whole SSO → callout → NATS chain);
//! 2. a deployment referencing a secret lands in desired state;
//! 3. the agent logs into OpenBao with the same Zitadel JWT — the
//!    `groups` claim comes from the flattening Action, the group grant
//!    from the same `OpenBaoDeploymentSecretGrants` the operator
//!    composes — fetches the secret and injects it into the container
//!    (SSH `podman inspect` is the ground truth);
//! 4. deleting the deployment's OpenBao policy makes the next fetch
//!    fail, surfaced as `Phase::Failed` naming the denied secret.
//!
//! Gating: skipped unless `HARMONY_FLEET_VM_E2E=1`.

use std::time::Duration;

use harmony::modules::fleet::secret_access::DeploymentSecretGrants;
use harmony::modules::openbao::OpenBaoDeploymentSecretGrants;
use harmony::modules::podman::{EnvVar, PodmanService, PodmanV0Score, SecretEnvVar};
use harmony::topology::{K8sAnywhereTopology, K8sclient, RestartPolicy};
use harmony_fleet_e2e::callout::{DEVICE_ROLE, KV_MOUNT, SECRET_PREFIX};
use harmony_fleet_e2e::{
    AdminKv, AuthMode, PhaseExpectation, VmStack, VmStackOptions, shared_vm_stack,
};
use harmony_reconciler_contracts::{DeploymentName, Phase};
use serde_json::json;

const ENV_GATE: &str = "HARMONY_FLEET_VM_E2E";

fn enabled() -> bool {
    matches!(std::env::var(ENV_GATE).as_deref(), Ok("1" | "true"))
}

fn dn(s: &str) -> DeploymentName {
    DeploymentName::try_new(s).expect("test-static valid deployment name")
}

/// First image pull on a TCG aarch64 guest can take minutes.
const RUN_BUDGET: Duration = Duration::from_secs(600);
const FAIL_BUDGET: Duration = Duration::from_secs(120);

/// Both tests drive the same device + OpenBao; serialize so policy
/// surgery in one can't race the other's apply.
static SERIAL: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());

fn callout_stack_options() -> VmStackOptions {
    VmStackOptions {
        auth_mode: AuthMode::Callout,
        // `device` drives NATS permissions in the callout; `edge-a`
        // is the OpenBao group the deployments below are granted to.
        device_roles: vec![DEVICE_ROLE.to_string(), "edge-a".to_string()],
        ..VmStackOptions::from_env()
    }
}

fn secret_score(service: &str, spec_marker: &str) -> PodmanV0Score {
    PodmanV0Score {
        services: vec![PodmanService {
            name: service.to_string(),
            image: "docker.io/library/nginx:alpine".to_string(),
            ports: vec![],
            // Changing the marker changes the serialized spec, which
            // is how the tests force a re-apply (and with it a fresh
            // secret resolution).
            env: vec![EnvVar::new("SPEC_MARKER", spec_marker)],
            secret_env: vec![SecretEnvVar {
                name: "DB_PASSWORD".to_string(),
                secret: "db_password".to_string(),
            }],
            volumes: vec![],
            restart_policy: RestartPolicy::default(),
        }],
    }
}

/// Host-side OpenBao admin surface: port-forward + root token.
struct BaoAdmin {
    base_url: String,
    root_token: String,
    http: reqwest::Client,
    _forward: harmony_k8s::PortForwardHandle,
}

impl BaoAdmin {
    async fn connect(stack: &VmStack) -> anyhow::Result<Self> {
        let handles = stack
            .infra
            .callout
            .as_ref()
            .expect("callout stack carries OpenBao handles");
        let topology = K8sAnywhereTopology::from_env();
        harmony::topology::Topology::ensure_ready(&topology)
            .await
            .map_err(|e| anyhow::anyhow!("topology not ready: {e:?}"))?;
        let k8s = topology.k8s_client().await.map_err(anyhow::Error::msg)?;
        let forward = k8s
            .port_forward(&handles.openbao.pod(), &handles.openbao.namespace, 0, 8200)
            .await?;
        Ok(Self {
            base_url: format!("http://127.0.0.1:{}", forward.port()),
            root_token: handles.openbao_root_token.clone(),
            http: reqwest::Client::new(),
            _forward: forward,
        })
    }

    async fn put_secret(
        &self,
        deployment: &DeploymentName,
        name: &str,
        value: &str,
    ) -> anyhow::Result<()> {
        self.http
            .post(format!(
                "{}/v1/{KV_MOUNT}/data/{SECRET_PREFIX}/{deployment}/{name}",
                self.base_url
            ))
            .header("X-Vault-Token", &self.root_token)
            .json(&json!({ "data": { "value": value } }))
            .send()
            .await?
            .error_for_status()?;
        Ok(())
    }

    fn grants(&self) -> OpenBaoDeploymentSecretGrants {
        OpenBaoDeploymentSecretGrants::new(
            self.base_url.clone(),
            self.root_token.clone(),
            KV_MOUNT.to_string(),
            SECRET_PREFIX.to_string(),
        )
    }

    /// The negative test's act of sabotage: drop the deployment's
    /// policy directly (operator drift / manual revocation), leaving
    /// group attachments in place.
    async fn delete_deployment_policy(&self, deployment: &DeploymentName) -> anyhow::Result<()> {
        self.http
            .delete(format!(
                "{}/v1/sys/policies/acl/deployment-{deployment}",
                self.base_url
            ))
            .header("X-Vault-Token", &self.root_token)
            .send()
            .await?
            .error_for_status()?;
        Ok(())
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn device_reads_deployment_secret_via_sso_groups() -> anyhow::Result<()> {
    if !enabled() {
        eprintln!("skipping {ENV_GATE}-gated VM e2e test (set {ENV_GATE}=1 to run)");
        return Ok(());
    }
    let _serial = SERIAL.lock().await;
    let _ = tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
        )
        .try_init();

    let stack = shared_vm_stack(callout_stack_options()).await?;
    stack.print_debug_info();
    // Ping over the callout-authenticated NATS connection — the agent
    // is online only if its Zitadel JWT made it through the callout.
    stack.wait_until_ready(Duration::from_secs(120)).await?;

    let device = stack.devices.first().expect("one VM device");
    let device_id = device.device_id.to_string();
    let deployment = dn("secret-web");
    let bao = BaoAdmin::connect(&stack).await?;
    // Fresh per-test connection — the stack's shared client dies
    // with the bring-up test's runtime (`Stack::connect_admin`).
    let nats = stack.infra.connect_admin().await?;
    let admin = AdminKv::connect(&nats).await?;

    let secret_value = format!("hunter2-{}", uuid::Uuid::new_v4().simple());
    bao.put_secret(&deployment, "db_password", &secret_value)
        .await?;
    bao.grants()
        .set_deployment_groups(&[(deployment.clone(), vec!["edge-a".to_string()])])
        .await?;

    admin
        .put_podman(
            &device_id,
            &deployment,
            &secret_score("secret-web-svc", "v1"),
        )
        .await?;
    let state = admin
        .wait_for_phase(
            &device_id,
            &deployment,
            PhaseExpectation::running(),
            RUN_BUDGET,
        )
        .await?
        .expect("Running phase implies a state entry");
    assert_eq!(state.phase, Phase::Running);

    // Ground truth: the secret VALUE made it into the container env.
    let inspect = device
        .ssh(
            "sudo -iu fleet-agent podman inspect secret-web-svc \
             --format '{{json .Config.Env}}'",
        )
        .await?
        .into_successful()
        .map_err(|e| anyhow::anyhow!("podman inspect failed: {e}"))?;
    assert!(
        inspect
            .stdout
            .contains(&format!("DB_PASSWORD={secret_value}")),
        "container env must carry the fetched secret, got:\n{}",
        inspect.stdout,
    );

    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn removed_policy_denies_secret_fetch() -> anyhow::Result<()> {
    if !enabled() {
        eprintln!("skipping {ENV_GATE}-gated VM e2e test (set {ENV_GATE}=1 to run)");
        return Ok(());
    }
    let _serial = SERIAL.lock().await;
    let _ = tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
        )
        .try_init();

    let stack = shared_vm_stack(callout_stack_options()).await?;
    stack.wait_until_ready(Duration::from_secs(120)).await?;

    let device = stack.devices.first().expect("one VM device");
    let device_id = device.device_id.to_string();
    let deployment = dn("secret-denied");
    let bao = BaoAdmin::connect(&stack).await?;
    // Fresh per-test connection — the stack's shared client dies
    // with the bring-up test's runtime (`Stack::connect_admin`).
    let nats = stack.infra.connect_admin().await?;
    let admin = AdminKv::connect(&nats).await?;

    bao.put_secret(&deployment, "db_password", "soon-revoked")
        .await?;
    bao.grants()
        .set_deployment_groups(&[(deployment.clone(), vec!["edge-a".to_string()])])
        .await?;

    // Sanity: with the grant in place the deployment converges.
    admin
        .put_podman(
            &device_id,
            &deployment,
            &secret_score("secret-denied-svc", "v1"),
        )
        .await?;
    admin
        .wait_for_phase(
            &device_id,
            &deployment,
            PhaseExpectation::running(),
            RUN_BUDGET,
        )
        .await?;

    // Remove the policy; the spec bump forces a re-apply and with it
    // a fresh secret resolution (resolution binds at apply time —
    // converged deployments never re-read, by design).
    bao.delete_deployment_policy(&deployment).await?;
    admin
        .put_podman(
            &device_id,
            &deployment,
            &secret_score("secret-denied-svc", "v2"),
        )
        .await?;
    let state = admin
        .wait_for_phase(
            &device_id,
            &deployment,
            PhaseExpectation::OneOf(vec![Phase::Failed]),
            FAIL_BUDGET,
        )
        .await?
        .expect("Failed phase implies a state entry");
    assert_eq!(state.phase, Phase::Failed);
    let err = state.last_error.unwrap_or_default();
    assert!(
        err.contains("fetching secret")
            && (err.contains("permission denied") || err.contains("403")),
        "failure must name the denied secret fetch, got: {err}",
    );

    Ok(())
}