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

use std::sync::Arc;

use harmony::inventory::Inventory;
use harmony::modules::openbao::{
    OpenbaoInstance, OpenbaoPolicy, OpenbaoScore, OpenbaoSetupScore, OpenbaoUser, cached_root_token,
};
use harmony::score::Score;
use harmony::topology::{K8sAnywhereTopology, K8sclient};
use harmony_fleet_e2e::{StackOptions, shared_stack};

const E2E_ENV: &str = "HARMONY_FLEET_E2E";
const RELEASE: &str = "openbao-policy-test";
const USER: &str = "policy-device";
const PASSWORD: &str = "policy-device-pass";
const POLICY: &str = "deployment-policy-test";
const SECRET_PATH: &str = "fleet-staging/deployment-a";

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

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn entity_policy_grants_existing_openbao_token_access() -> anyhow::Result<()> {
    if !e2e_enabled() {
        skip_e2e();
        return Ok(());
    }

    let stack = openbao_stack().await?;
    stack.print_debug_info();

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

    let topology = K8sAnywhereTopology::from_env();
    let k8s = topology.k8s_client().await.map_err(anyhow::Error::msg)?;
    let root_token = cached_root_token(&instance).map_err(anyhow::Error::msg)?;
    let user_token = login_user(&k8s, &instance).await?;
    let entity_id = token_entity_id(&k8s, &instance, &user_token).await?;

    set_entity_policies(&k8s, &instance, &root_token, &entity_id, "").await?;
    assert_secret_denied(&k8s, &instance, &user_token).await?;

    set_entity_policies(&k8s, &instance, &root_token, &entity_id, POLICY).await?;
    assert_secret_allowed(&k8s, &instance, &user_token).await?;

    set_entity_policies(&k8s, &instance, &root_token, &entity_id, "").await?;
    Ok(())
}

async fn openbao_stack() -> anyhow::Result<Arc<harmony_fleet_e2e::Stack>> {
    let _ = tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
        )
        .try_init();

    Ok(shared_stack(StackOptions {
        deploy_agent: false,
        deploy_operator: false,
        ..StackOptions::infra_only()
    })
    .await?)
}

async fn deploy_openbao(instance: &OpenbaoInstance) -> anyhow::Result<()> {
    let topology = K8sAnywhereTopology::from_env();
    let policy_hcl = format!(
        r#"path "secret/data/{SECRET_PATH}" {{ capabilities = ["read"] }}
path "secret/metadata/{SECRET_PATH}" {{ capabilities = ["read"] }}"#
    );
    let scores: Vec<Box<dyn Score<K8sAnywhereTopology>>> = vec![
        Box::new(OpenbaoScore {
            instance: instance.clone(),
            host: "openbao-policy-test.local".to_string(),
            openshift: false,
            tls_issuer: None,
        }),
        Box::new(OpenbaoSetupScore {
            instance: instance.clone(),
            kv_mount: "secret".to_string(),
            policies: vec![OpenbaoPolicy {
                name: POLICY.to_string(),
                hcl: policy_hcl,
            }],
            users: vec![OpenbaoUser {
                username: USER.to_string(),
                password: PASSWORD.to_string(),
                policies: vec![],
            }],
            jwt_auth: None,
        }),
    ];

    for score in scores {
        score.interpret(&Inventory::autoload(), &topology).await?;
    }
    let k8s = topology.k8s_client().await.map_err(anyhow::Error::msg)?;
    let root_token = cached_root_token(instance).map_err(anyhow::Error::msg)?;
    bao(
        &k8s,
        instance,
        &root_token,
        "bao kv put secret/fleet-staging/deployment-a value=e2e",
    )
    .await?;
    Ok(())
}

async fn login_user(
    k8s: &harmony_k8s::K8sClient,
    instance: &OpenbaoInstance,
) -> anyhow::Result<String> {
    let out = exec(
        k8s,
        instance,
        &format!("bao login -method=userpass -format=json username={USER} password={PASSWORD}"),
    )
    .await?;
    Ok(
        serde_json::from_str::<serde_json::Value>(&out)?["auth"]["client_token"]
            .as_str()
            .ok_or_else(|| anyhow::anyhow!("login response missing auth.client_token: {out}"))?
            .to_string(),
    )
}

async fn token_entity_id(
    k8s: &harmony_k8s::K8sClient,
    instance: &OpenbaoInstance,
    token: &str,
) -> anyhow::Result<String> {
    let out = bao(k8s, instance, token, "bao token lookup -format=json").await?;
    Ok(
        serde_json::from_str::<serde_json::Value>(&out)?["data"]["entity_id"]
            .as_str()
            .ok_or_else(|| anyhow::anyhow!("token lookup response missing data.entity_id: {out}"))?
            .to_string(),
    )
}

async fn set_entity_policies(
    k8s: &harmony_k8s::K8sClient,
    instance: &OpenbaoInstance,
    root_token: &str,
    entity_id: &str,
    policies: &str,
) -> anyhow::Result<()> {
    bao(
        k8s,
        instance,
        root_token,
        &format!("bao write identity/entity/id/{entity_id} policies={policies}"),
    )
    .await?;
    Ok(())
}

async fn assert_secret_denied(
    k8s: &harmony_k8s::K8sClient,
    instance: &OpenbaoInstance,
    token: &str,
) -> anyhow::Result<()> {
    let result = bao(
        k8s,
        instance,
        token,
        &format!("bao kv get secret/{SECRET_PATH}"),
    )
    .await;
    assert!(result.is_err(), "secret read unexpectedly succeeded");
    Ok(())
}

async fn assert_secret_allowed(
    k8s: &harmony_k8s::K8sClient,
    instance: &OpenbaoInstance,
    token: &str,
) -> anyhow::Result<()> {
    bao(
        k8s,
        instance,
        token,
        &format!("bao kv get secret/{SECRET_PATH}"),
    )
    .await?;
    Ok(())
}

async fn bao(
    k8s: &harmony_k8s::K8sClient,
    instance: &OpenbaoInstance,
    token: &str,
    command: &str,
) -> anyhow::Result<String> {
    exec(
        k8s,
        instance,
        &format!("export VAULT_TOKEN={token} && {command}"),
    )
    .await
}

async fn exec(
    k8s: &harmony_k8s::K8sClient,
    instance: &OpenbaoInstance,
    command: &str,
) -> anyhow::Result<String> {
    k8s.exec_pod_capture_output(
        &instance.pod(),
        Some(&instance.namespace),
        vec!["sh", "-c", command],
    )
    .await
    .map_err(|e| anyhow::anyhow!(e))
}

fn skip_e2e() {
    eprintln!(
        "skipping {E2E_ENV}-gated OpenBao e2e test (set {E2E_ENV}=1 to run; requires k3d + helm)"
    );
}