kranz-engine 0.4.1

//! Per-run Sgian client credential (optional coordination lane).
//!
//! When the operator runs a mission inside a repository that a Sgian daemon
//! also serves, every worker session the engine spawns identifies itself to
//! that daemon as its own principal: the engine asks the daemon for a
//! credential held by `kranz:<run-id>` with the `write` scope, hands the
//! token to the worker through `SGIAN_CLIENT_TOKEN`, and revokes the
//! credential when the run ends. Sgian then attributes every pane the worker
//! drives, every lease it takes and every ledger record it produces to the
//! run rather than to the operator. Revocation is best-effort, including on
//! future cancellation; engine death still needs operator reconciliation.
//!
//! The lane is best-effort and never a reason to fail a spawn:
//! - `KRANZ_SGIAN_BIN` set to a path uses that `sgian` binary; set but empty
//!   disables the lane; unset searches `PATH` for `sgian`.
//! - No binary, no daemon serving the repository root, a refused request, a
//!   malformed reply or a call that outlasts [`DEADLINE`] all degrade to a
//!   session without the variable, logged at `debug` (absent daemon is the
//!   common case) or `warn` (the daemon answered but the reply was unusable).
//!
//! Only the credential id and holder are logged; the token crosses into the
//! session env and nowhere else.

use std::collections::HashMap;
use std::ffi::OsString;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::time::Duration;

/// Environment variable the worker reads (Sgian's own client convention).
pub const TOKEN_ENV: &str = "SGIAN_CLIENT_TOKEN";
/// Operator override for the `sgian` binary; empty disables the lane.
pub const BIN_ENV: &str = "KRANZ_SGIAN_BIN";
/// Upper bound on one `sgian ctl` call. The daemon answers on a local socket
/// in milliseconds; anything longer is a wedged client and the run must not
/// wait on it.
pub const DEADLINE: Duration = Duration::from_secs(5);

/// A credential the engine issued for one run and owes a revocation for.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SgianCredential {
    /// Daemon-assigned credential id (`identity revoke` takes this).
    pub id: String,
    /// `kranz:<run-id>`.
    pub holder: String,
    bin: PathBuf,
    workspace: PathBuf,
}

/// Revokes an issued worker credential even if its running future is dropped.
pub(crate) struct RevocationGuard(pub SgianCredential);

impl Drop for RevocationGuard {
    fn drop(&mut self) {
        self.0.revoke();
    }
}

/// The holder name Sgian records for a run.
pub fn holder_for(run_id: &str) -> String {
    format!("kranz:{run_id}")
}

/// Issue a `write` credential for `run_id` against the daemon serving
/// `workspace`, returning the credential and its one-time token, or `None`
/// when the lane is disabled or unavailable.
pub fn issue(workspace: &Path, run_id: &str) -> Option<(SgianCredential, String)> {
    let bin = resolve_bin(std::env::var_os(BIN_ENV), std::env::var_os("PATH"))?;
    issue_with(&bin, workspace, run_id, DEADLINE)
}

/// [`issue`] against an explicit binary and deadline (the testable core).
pub fn issue_with(
    bin: &Path,
    workspace: &Path,
    run_id: &str,
    deadline: Duration,
) -> Option<(SgianCredential, String)> {
    let holder = holder_for(run_id);
    let args = [
        "ctl",
        "--workspace",
        &workspace.display().to_string(),
        "--json",
        "identity",
        "issue",
        "--holder",
        &holder,
        "--scope",
        "write",
    ];
    let output = match run_ctl(bin, &args, deadline) {
        Ok(output) => output,
        Err(reason) => {
            tracing::debug!(
                run_id,
                workspace = %workspace.display(),
                reason,
                "sgian credential not issued; the session runs without one"
            );
            return None;
        }
    };
    let record: serde_json::Value = match serde_json::from_slice(&output) {
        Ok(record) => record,
        Err(error) => {
            tracing::warn!(
                run_id,
                error = %error,
                "sgian identity issue replied with something other than a credential record"
            );
            return None;
        }
    };
    let id = record.get("id").and_then(serde_json::Value::as_str);
    let issued = record.get("token").and_then(serde_json::Value::as_str);
    let (Some(id), Some(token)) = (id, issued) else {
        tracing::warn!(
            run_id,
            "sgian identity issue record lacks an id or token; ignoring it"
        );
        return None;
    };
    if id.is_empty() || token.is_empty() {
        return None;
    }
    tracing::info!(
        run_id,
        id,
        holder = %holder,
        "sgian credential issued for the run (revoked when the run ends)"
    );
    Some((
        SgianCredential {
            id: id.to_string(),
            holder,
            bin: bin.to_path_buf(),
            workspace: workspace.to_path_buf(),
        },
        token.to_string(),
    ))
}

impl SgianCredential {
    /// Attempt revocation. Failure is logged; daemon state must be reconciled
    /// by the operator if cleanup could not be confirmed.
    pub fn revoke(&self) {
        self.revoke_with(DEADLINE);
    }

    /// [`revoke`](Self::revoke) with an explicit deadline.
    pub fn revoke_with(&self, deadline: Duration) {
        let args = [
            "ctl",
            "--workspace",
            &self.workspace.display().to_string(),
            "--json",
            "identity",
            "revoke",
            &self.id,
        ];
        match run_ctl(&self.bin, &args, deadline) {
            Ok(_) => tracing::info!(
                id = %self.id,
                holder = %self.holder,
                "sgian credential revoked"
            ),
            Err(reason) => tracing::warn!(
                id = %self.id,
                holder = %self.holder,
                reason,
                "sgian credential could not be revoked; revoke it by hand with \
                 `sgian ctl identity revoke` if the daemon is still running"
            ),
        }
    }
}

/// Pick the `sgian` binary: the override wins, an empty override disables,
/// otherwise the first `sgian` on `PATH`.
pub fn resolve_bin(override_var: Option<OsString>, path_var: Option<OsString>) -> Option<PathBuf> {
    match override_var {
        Some(value) if value.is_empty() => None,
        Some(value) => Some(PathBuf::from(value)),
        None => std::env::split_paths(&path_var?).find_map(|dir| {
            let candidate = dir.join(BIN_NAME);
            candidate.is_file().then_some(candidate)
        }),
    }
}

#[cfg(windows)]
const BIN_NAME: &str = "sgian.exe";
#[cfg(not(windows))]
const BIN_NAME: &str = "sgian";

/// Run the operator helper with bounded pipes and process-tree cleanup. The
/// discovery allowlist keeps HOME for daemon lookup, without forwarding provider
/// credentials or a worker's client token. Never log the helper's reply bytes.
fn run_ctl(bin: &Path, args: &[&str], deadline: Duration) -> Result<Vec<u8>, String> {
    let mut command = Command::new(bin);
    command
        .args(args)
        .env_clear()
        .envs(crate::agent_env::probe_child_env(&[]));
    let output = crate::git_ops::process::output(
        command,
        crate::git_ops::process::Limits::for_control(deadline),
    )
    .map_err(|error| format!("{} control call failed: {error}", bin.display()))?;
    if output.status.success() {
        Ok(output.stdout)
    } else {
        Err(format!("{} exited with {}", bin.display(), output.status))
    }
}

/// Apply an issued credential to a session env; a convenience for callers
/// that already hold the token.
pub fn seed_env(env: &mut HashMap<String, String>, token: String) {
    env.insert(TOKEN_ENV.to_string(), token);
}

#[cfg(all(test, unix))]
mod tests {
    use super::*;
    use std::os::unix::fs::PermissionsExt;
    use std::time::Instant;

    /// A fake `sgian` that appends its argv to `<dir>/calls`, then behaves
    /// per `body` (a shell snippet run with the args still in `$@`).
    fn fake_sgian(dir: &Path, body: &str) -> PathBuf {
        let bin = dir.join("sgian");
        let calls = dir.join("calls");
        std::fs::write(
            &bin,
            format!(
                "#!/bin/sh\nprintf '%s\\n' \"$*\" >> '{}'\n{body}\n",
                calls.display()
            ),
        )
        .unwrap();
        std::fs::set_permissions(&bin, std::fs::Permissions::from_mode(0o755)).unwrap();
        bin
    }

    fn calls(dir: &Path) -> Vec<String> {
        std::fs::read_to_string(dir.join("calls"))
            .unwrap_or_default()
            .lines()
            .map(str::to_string)
            .collect()
    }

    #[test]
    fn issue_parses_the_record_and_revoke_uses_its_id() {
        let dir = tempfile::tempdir().unwrap();
        let ws = dir.path().join("repo");
        std::fs::create_dir_all(&ws).unwrap();
        let bin = fake_sgian(
            dir.path(),
            r#"case "$*" in *"identity issue"*) printf '{"id":"cred-7","holder":"kranz:run-1","scopes":["write"],"token":"sgc_t"}\n';; *) printf '{"id":"cred-7","revoked":true}\n';; esac"#,
        );
        // The fixture value stays under eight characters so the repository's own
        // secret scanner, which runs with the base branch's allowlist, does not
        // read it as a credential assignment.
        let (cred, token) = issue_with(&bin, &ws, "run-1", DEADLINE).expect("credential issued");
        assert_eq!(cred.id, "cred-7");
        assert_eq!(cred.holder, "kranz:run-1");
        assert_eq!(token, "sgc_t");
        cred.revoke();
        let calls = calls(dir.path());
        assert_eq!(
            calls,
            vec![
                format!(
                    "ctl --workspace {} --json identity issue --holder kranz:run-1 --scope write",
                    ws.display()
                ),
                format!(
                    "ctl --workspace {} --json identity revoke cred-7",
                    ws.display()
                ),
            ]
        );
    }

    #[test]
    fn issue_is_none_when_ctl_fails_or_answers_nonsense() {
        let dir = tempfile::tempdir().unwrap();
        let ws = dir.path().to_path_buf();
        let failing = fake_sgian(
            dir.path(),
            "echo 'no daemon serves this workspace' >&2; exit 1",
        );
        assert!(issue_with(&failing, &ws, "run-2", DEADLINE).is_none());
        let garbage = fake_sgian(dir.path(), "echo 'not json'");
        assert!(issue_with(&garbage, &ws, "run-2", DEADLINE).is_none());
        let partial = fake_sgian(dir.path(), r#"printf '{"id":"x"}\n'"#);
        assert!(issue_with(&partial, &ws, "run-2", DEADLINE).is_none());
    }

    #[test]
    fn issue_gives_up_on_a_wedged_client() {
        let dir = tempfile::tempdir().unwrap();
        let bin = fake_sgian(dir.path(), "sleep 5");
        let started = Instant::now();
        assert!(issue_with(&bin, dir.path(), "run-3", Duration::from_millis(200)).is_none());
        assert!(
            started.elapsed() < Duration::from_secs(3),
            "the deadline must cut the wait short"
        );
    }

    #[test]
    fn sgian_control_deadline_covers_pipes_after_leader_exit() {
        let dir = tempfile::tempdir().unwrap();
        let marker = dir.path().join("descendant-ran");
        let bin = fake_sgian(
            dir.path(),
            &format!("(sleep 1; touch '{}') & exit 0", marker.display()),
        );
        // Prove the same descendant would act without cancellation.
        assert!(run_ctl(&bin, &[], Duration::from_secs(3)).is_ok());
        assert!(marker.exists());
        std::fs::remove_file(&marker).unwrap();
        let started = Instant::now();
        assert!(run_ctl(&bin, &[], Duration::from_millis(100)).is_err());
        assert!(started.elapsed() < Duration::from_secs(3));
        std::thread::sleep(Duration::from_millis(1200));
        assert!(!marker.exists(), "timed-out descendants must not continue");
    }

    #[test]
    fn sgian_control_refuses_oversize_output_and_does_not_log_reply_bytes() {
        let dir = tempfile::tempdir().unwrap();
        for stream in ["", " >&2"] {
            let bin = fake_sgian(dir.path(), &format!("head -c 65537 /dev/zero{stream}"));
            assert!(run_ctl(&bin, &[], DEADLINE).is_err());
        }
        let bin = fake_sgian(dir.path(), "echo 'private-reply' >&2; exit 1");
        let error = run_ctl(&bin, &[], DEADLINE).unwrap_err();
        assert!(!error.contains("private-reply"));
    }

    #[test]
    fn sgian_control_keeps_discovery_environment_without_ambient_secrets() {
        let name =
            "sgian::tests::sgian_control_keeps_discovery_environment_without_ambient_secrets";
        if crate::agent_env::isolated_global_home_test(name) {
            return;
        }
        std::env::set_var("KRANZ_SGIAN_PRIVATE_FIXTURE", "private-value");
        std::env::set_var(TOKEN_ENV, "private-value");
        let dir = tempfile::tempdir().unwrap();
        let bin = fake_sgian(dir.path(), "env");
        let output = run_ctl(&bin, &[], DEADLINE).unwrap();
        let env = String::from_utf8(output).unwrap();
        assert!(env.lines().any(|line| line.starts_with("HOME=")));
        assert!(env.lines().any(|line| line.starts_with("PATH=")));
        assert!(!env.contains("KRANZ_SGIAN_PRIVATE_FIXTURE="));
        assert!(!env.contains("SGIAN_CLIENT_TOKEN="));
    }

    #[test]
    fn issue_is_none_when_the_binary_is_missing() {
        let dir = tempfile::tempdir().unwrap();
        assert!(issue_with(&dir.path().join("absent"), dir.path(), "run-4", DEADLINE).is_none());
    }

    #[test]
    fn resolve_bin_honours_the_override_and_searches_path() {
        let dir = tempfile::tempdir().unwrap();
        let bin = fake_sgian(dir.path(), "true");
        assert_eq!(
            resolve_bin(Some(OsString::from("/opt/sgian")), None),
            Some(PathBuf::from("/opt/sgian"))
        );
        assert_eq!(
            resolve_bin(Some(OsString::new()), Some(dir.path().into())),
            None
        );
        let path =
            std::env::join_paths([dir.path().join("nowhere"), dir.path().to_path_buf()]).unwrap();
        assert_eq!(resolve_bin(None, Some(path)), Some(bin));
        assert_eq!(
            resolve_bin(None, Some(OsString::from(dir.path().join("nowhere")))),
            None
        );
        assert_eq!(resolve_bin(None, None), None);
    }

    #[test]
    fn holder_names_the_run() {
        assert_eq!(holder_for("abc"), "kranz:abc");
    }
}