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

//! ADR-025 group enforcement, end to end against a real OpenBao: a
//! device's secret access is decided by the groups in its signed JWT
//! and the deployment policies attached to those groups. Self-contained
//! — the JWT auth mount validates against a test RSA key instead of a
//! live Zitadel, and tests mint their own device tokens.

use harmony::inventory::Inventory;
use harmony::modules::openbao::{
    OpenbaoInstance, OpenbaoJwtAuth, OpenbaoScore, OpenbaoSetupScore, cached_root_token,
};
use harmony::score::Score;
use harmony::topology::{K8sAnywhereTopology, K8sclient};
use harmony_fleet_e2e::{StackOptions, shared_stack};
use harmony_reconciler_contracts::{DeploymentName, DeploymentSecretGrants};
use harmony_secret::OpenBaoDeploymentSecretGrants;
use serde_json::json;

const E2E_ENV: &str = "HARMONY_FLEET_E2E";
const RELEASE: &str = "openbao-groups-test";
const KV_MOUNT: &str = "secret";
const FLEET_NS: &str = "fleet-e2e";
const ISSUER: &str = "https://fleet-e2e.test";
const AUDIENCE: &str = "fleet-e2e-project";
const ROLE: &str = "fleet-device";
const JWT_MOUNT: &str = "jwt";

// Test-only RSA keypair: lets the suite mint "Zitadel" tokens without a
// live IdP. Never used outside this test.
const TEST_JWT_PRIVATE_PEM: &str = r#"-----BEGIN PRIVATE KEY-----
MIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQC0VwVrePTYWsym
J+MIOToktME/60/nYFhLi0awgpF5T6McF2TJ/v+Bo1KQQ5Fg53XRGZuFnSKr5WZ+
KzJ7d9YpUSCc/9UMcd3ujhrp35X4rXOlCOqIklRXb/TTmxrl5LMr91zlHFOSxa+/
N5yNYKQVsy9bHeJvOpX5ktWELVS2JoAbvvuVhsuWrFyAS9s3i4a2q1jkg9f+vyPc
BtExIw/Oe5L03x6ZQlPNPLh/c3uQlK7ikOB1877IAWK7/Nnj+L8FPmG9FmLp/CeZ
n6gsqeDq6KloyffCx0ZiyK3ijz2wIHIYo3SVe3jyDjRJLiBIJrcxynq4aid3OGDP
Tw13tshPAgMBAAECggEAFpEhK366XOyX2V76aQgIP5mi6SXAHPsQcXufP+1G2QBf
L9+5jBFt1cIjTcw2wAKW2rCypniwYwRU3Ar5yTuIwCjXxO26slD5baW4tnV1JeZG
p2tDNg3hogNvhp9NMWF6DlFo/X6a8q3sPShvXb2vKsv+bA+ES1jwdYerUlv90sEu
PzW4GcBPCQ19udZEIM1Y2bObKi6bAT3QAANkzQC7kRhV104l7rC9eHhfbaW1Ce9N
rPY2Ekq2ck3kuEz+YrZ0Lh3yvozslK8aPn05ZM1sYv+9Mxb0rwv4LqGdQfNvONbA
ZMLL++chHqTSr5tZ26POLjZ5ZTm/jlOK0ZJ5RE6ZQQKBgQD5JW3U2vtntGl6V2yj
zvDNpXeKNh2CBMB5cuub+G0J2XtAJXW4W3zsGytwNdmrc/IVVJI9CeJ/g+2PvI7d
fiUvSSDZyrvib6+kGj5qDLHEiLJ0LhqEP4/jzHcKoL1lID63C65Sw0ZhKgy/lPXn
elBcuzDIaI+9sre6A9TdnL1vrwKBgQC5TQkIh0aI9PFq184Ugd/Qaj8ckjAbu5ms
4c4cZIVvENFIjSwdk3/p88AP8UmRMTIUUViZ2rvZYJkVUzi15AFbti4sYB5IXTc6
D00Ev+SYW/b1bf/Gu8o/c2UWrNY1823xwXjW0bRT0O9nJJhSDs7GCDOCL8pIDWNd
3xJWWea5YQKBgQDo6sjm+WGW7d699QUnjssoNcyvdS/Y4UBfScRiJ7koQB0tNtf6
5ey/fhb57B32X4qtzkZ4sckBlQB+I7Ub9FYnAqKneRR0+BynubEny35VUcYjPwaj
JBkMyZVwZh0pcR+wWEc8MDqXk4wffs70/DAswQM2GCppGa8RQ4u5MlMXGQKBgC2J
2xB3fkQmPjZNtt2fVUu48K2q15vZLNkp3SdpJMm4zlYQK6lRGHx8ZNUGLVg7SN1p
aGbXNJclgDmXbDZ3REYvrYVGUerizbG9FOZ36kvvPe/vCo7ik7+oMk2ySCqWGozx
SSsx+iknDc7BVZXFAdf1KAN0jzfJ4oOtUKpcCbHBAoGBAJGyuQUswc7EQeYLn/H5
Zl3kRQv/VU8+lW0tro//osD2Cl5sW1YRaU0+QsOHy/GDPRqcWEqGfyUmFvTK09vc
vSYCbIcS2U18wo7I/j/vicqou0vnNeZCOQHrYvHurGLNUQKqVmVuDArsHx9aDIH0
lJT0SeBwt1iqJi1lIPtEQbOy
-----END PRIVATE KEY-----"#;

const TEST_JWT_PUBLIC_PEM: &str = r#"-----BEGIN PUBLIC KEY-----
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAtFcFa3j02FrMpifjCDk6
JLTBP+tP52BYS4tGsIKReU+jHBdkyf7/gaNSkEORYOd10RmbhZ0iq+Vmfisye3fW
KVEgnP/VDHHd7o4a6d+V+K1zpQjqiJJUV2/005sa5eSzK/dc5RxTksWvvzecjWCk
FbMvWx3ibzqV+ZLVhC1UtiaAG777lYbLlqxcgEvbN4uGtqtY5IPX/r8j3AbRMSMP
znuS9N8emUJTzTy4f3N7kJSu4pDgdfO+yAFiu/zZ4/i/BT5hvRZi6fwnmZ+oLKng
6uipaMn3wsdGYsit4o89sCByGKN0lXt48g40SS4gSCa3Mcp6uGondzhgz08Nd7bI
TwIDAQAB
-----END PUBLIC KEY-----"#;

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

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

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn group_grants_govern_device_secret_access() -> anyhow::Result<()> {
    if !e2e_enabled() {
        eprintln!("skipping {E2E_ENV}-gated OpenBao e2e (set {E2E_ENV}=1; requires k3d + helm)");
        return Ok(());
    }

    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_stack(StackOptions {
        deploy_agent: false,
        deploy_operator: false,
        ..StackOptions::infra_only()
    })
    .await?;
    stack.print_debug_info();

    let instance = OpenbaoInstance {
        namespace: stack.namespace.clone(),
        release: RELEASE.to_string(),
    };
    let bao = deploy_openbao(&instance).await?;

    // The unit under test: deployment "web" readable by group edge-a.
    let grants = OpenBaoDeploymentSecretGrants::new(
        bao.base_url.clone(),
        bao.root_token.clone(),
        KV_MOUNT.to_string(),
        FLEET_NS.to_string(),
    );
    grants
        .set_deployment_groups(&[(dn("web"), vec!["edge-a".to_string()])])
        .await?;

    // Device in edge-a: login binds the group, secret readable, batch token.
    let login_a = bao.jwt_login(&mint_jwt("pi-01", &["edge-a"])).await?;
    assert_eq!(
        login_a.token_type, "batch",
        "fleet role must issue batch tokens"
    );
    bao.assert_can_read(&login_a.token, "web/config").await?;

    // Device in another group: valid identity, no grant.
    let login_b = bao.jwt_login(&mint_jwt("pi-02", &["edge-b"])).await?;
    bao.assert_denied(&login_b.token, "web/config").await?;

    // Detach binds at request time: the in-hand token loses access with
    // no re-login involved.
    grants.set_deployment_groups(&[(dn("web"), vec![])]).await?;
    bao.assert_denied(&login_a.token, "web/config").await?;

    // Re-attach binds at request time too.
    grants
        .set_deployment_groups(&[(dn("web"), vec!["edge-a".to_string()])])
        .await?;
    bao.assert_can_read(&login_a.token, "web/config").await?;

    // Group membership is read from the claim only at login: pi-02's old
    // token keeps its old groups; a fresh login with edge-a membership
    // gets access.
    bao.assert_denied(&login_b.token, "web/config").await?;
    let login_b2 = bao.jwt_login(&mint_jwt("pi-02", &["edge-a"])).await?;
    bao.assert_can_read(&login_b2.token, "web/config").await?;

    // Per-device kill switch: disabling the entity blocks its in-flight
    // batch token at request time (ADR-025-3 open item 1).
    bao.set_entity_disabled("pi-01", true).await?;
    bao.assert_denied(&login_a.token, "web/config").await?;
    bao.set_entity_disabled("pi-01", false).await?;
    bao.assert_can_read(&login_a.token, "web/config").await?;

    Ok(())
}

/// Deploys OpenBao + JWT auth configured for the test keypair, seeds the
/// deployment secret, and port-forwards 8200 for host-side HTTP.
async fn deploy_openbao(instance: &OpenbaoInstance) -> anyhow::Result<Bao> {
    let topology = K8sAnywhereTopology::from_env();
    // Raw `Score::interpret` skips topology preparation; without this the
    // k8s client is never initialized (`ensure_ready` is what the CLI /
    // stack bring-up normally run).
    harmony::topology::Topology::ensure_ready(&topology)
        .await
        .map_err(|e| anyhow::anyhow!("topology not ready: {e:?}"))?;
    let scores: Vec<Box<dyn Score<K8sAnywhereTopology>>> = vec![
        Box::new(OpenbaoScore {
            instance: instance.clone(),
            host: "openbao-groups-test.local".to_string(),
            openshift: false,
            tls_issuer: None,
            node_port: None,
            create_namespace: true,
        }),
        Box::new(OpenbaoSetupScore {
            instance: instance.clone(),
            kv_mount: KV_MOUNT.to_string(),
            policies: vec![],
            users: vec![],
            jwt_auth: Some(OpenbaoJwtAuth {
                oidc_discovery_url: String::new(),
                jwt_validation_pubkeys: TEST_JWT_PUBLIC_PEM.to_string(),
                bound_issuer: ISSUER.to_string(),
                role_name: ROLE.to_string(),
                bound_audiences: AUDIENCE.to_string(),
                user_claim: "sub".to_string(),
                groups_claim: "groups".to_string(),
                token_type: "batch".to_string(),
                ttl: "1h".to_string(),
                max_ttl: "1h".to_string(),
            }),
            oidc_application: None,
            endpoint: None,
        }),
    ];
    for score in scores {
        score.interpret(&Inventory::autoload(), &topology).await?;
    }

    let k8s = topology.k8s_client().await.map_err(anyhow::Error::msg)?;
    let forward = k8s
        .port_forward(&instance.pod(), &instance.namespace, 0, 8200)
        .await?;
    let bao = Bao {
        base_url: format!("http://127.0.0.1:{}", forward.port()),
        root_token: cached_root_token(instance).map_err(anyhow::Error::msg)?,
        http: reqwest::Client::new(),
        _forward: forward,
    };

    bao.http
        .post(format!(
            "{}/v1/{KV_MOUNT}/data/{FLEET_NS}/web/config",
            bao.base_url
        ))
        .header("X-Vault-Token", &bao.root_token)
        .json(&json!({ "data": { "value": "e2e" } }))
        .send()
        .await?
        .error_for_status()?;
    Ok(bao)
}

struct Bao {
    base_url: String,
    root_token: String,
    http: reqwest::Client,
    _forward: harmony_k8s::PortForwardHandle,
}

struct Login {
    token: String,
    token_type: String,
}

impl Bao {
    async fn jwt_login(&self, jwt: &str) -> anyhow::Result<Login> {
        let body: serde_json::Value = self
            .http
            .post(format!("{}/v1/auth/{JWT_MOUNT}/login", self.base_url))
            .json(&json!({ "role": ROLE, "jwt": jwt }))
            .send()
            .await?
            .error_for_status()?
            .json()
            .await?;
        let auth = &body["auth"];
        Ok(Login {
            token: auth["client_token"]
                .as_str()
                .ok_or_else(|| anyhow::anyhow!("login response missing client_token: {body}"))?
                .to_string(),
            token_type: auth["token_type"].as_str().unwrap_or_default().to_string(),
        })
    }

    async fn read_secret(&self, token: &str, rel: &str) -> anyhow::Result<reqwest::StatusCode> {
        Ok(self
            .http
            .get(format!(
                "{}/v1/{KV_MOUNT}/data/{FLEET_NS}/{rel}",
                self.base_url
            ))
            .header("X-Vault-Token", token)
            .send()
            .await?
            .status())
    }

    async fn assert_can_read(&self, token: &str, rel: &str) -> anyhow::Result<()> {
        let status = self.read_secret(token, rel).await?;
        anyhow::ensure!(
            status.is_success(),
            "expected read of {rel} to succeed, got {status}"
        );
        Ok(())
    }

    async fn assert_denied(&self, token: &str, rel: &str) -> anyhow::Result<()> {
        let status = self.read_secret(token, rel).await?;
        anyhow::ensure!(
            status == reqwest::StatusCode::FORBIDDEN,
            "expected read of {rel} to be denied, got {status}"
        );
        Ok(())
    }

    /// Resolve the device's entity through its JWT-mount alias and flip
    /// `disabled` — the per-device kill switch.
    async fn set_entity_disabled(&self, device: &str, disabled: bool) -> anyhow::Result<()> {
        let accessor: serde_json::Value = self
            .http
            .get(format!("{}/v1/sys/auth", self.base_url))
            .header("X-Vault-Token", &self.root_token)
            .send()
            .await?
            .error_for_status()?
            .json()
            .await?;
        let accessor = accessor["data"][&format!("{JWT_MOUNT}/")]["accessor"]
            .as_str()
            .ok_or_else(|| anyhow::anyhow!("jwt mount accessor not found"))?
            .to_string();

        let entity: serde_json::Value = self
            .http
            .post(format!("{}/v1/identity/lookup/entity", self.base_url))
            .header("X-Vault-Token", &self.root_token)
            .json(&json!({ "alias_name": device, "alias_mount_accessor": accessor }))
            .send()
            .await?
            .error_for_status()?
            .json()
            .await?;
        let id = entity["data"]["id"]
            .as_str()
            .ok_or_else(|| anyhow::anyhow!("entity for {device} not found: {entity}"))?;

        self.http
            .post(format!("{}/v1/identity/entity/id/{id}", self.base_url))
            .header("X-Vault-Token", &self.root_token)
            .json(&json!({ "disabled": disabled }))
            .send()
            .await?
            .error_for_status()?;
        Ok(())
    }
}

fn mint_jwt(sub: &str, groups: &[&str]) -> String {
    use jsonwebtoken::{Algorithm, EncodingKey, Header, encode};
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_secs();
    let claims = json!({
        "iss": ISSUER,
        "aud": AUDIENCE,
        "sub": sub,
        "groups": groups,
        "iat": now,
        "exp": now + 300,
    });
    encode(
        &Header::new(Algorithm::RS256),
        &claims,
        &EncodingKey::from_rsa_pem(TEST_JWT_PRIVATE_PEM.as_bytes()).expect("test key parses"),
    )
    .expect("test jwt signs")
}