zlayer-agent 0.13.0

//! Per-container credential minting.
//!
//! Every container can be handed a JWT so its workload can call the daemon API
//! back without external credentials. Historically that token was an unscoped,
//! 365-day, cluster-wide reader (`roles: ["container"]`) AND the host admin
//! Unix socket was bind-mounted into every container — so a container was, in
//! effect, daemon-admin. Both holes are closed here:
//!
//! - The token is now a **scoped** access token (`scopes: [deployment:<own>:read]`
//!   by default), bounded by a short TTL, carrying no privileged role.
//! - The admin socket is **not** mounted by default; a deployment must opt in
//!   explicitly via the [`LABEL_DAEMON_SOCKET`] label (documented as full-admin).
//!
//! All three knobs are overridable per-service via labels so a CI-runner
//! service can request broader scopes / a longer TTL / the socket when it
//! genuinely needs them.

use std::collections::HashMap;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use async_trait::async_trait;
use jsonwebtoken::{encode, EncodingKey, Header};
use zlayer_types::jwt::Claims;
use zlayer_types::storage::{PermissionLevel, StoredAccessToken, TokenScope};

/// Sink the runtime uses to persist + revoke per-container scoped access
/// tokens. Implemented in the bin over the daemon's `AccessTokenStorage`
/// (which is SecretsStore-backed, so it replicates cluster-wide).
#[async_trait]
pub trait ContainerTokenSink: Send + Sync + std::fmt::Debug {
    /// Persist a freshly-minted container token record (so its `jti` is
    /// accepted by the auth layer's fail-closed revocation check). Returns
    /// `true` when the record was persisted (the token is now revocable and its
    /// `jti` may safely be embedded); `false` on failure (the caller must mint
    /// without a `jti` so the fail-closed check doesn't reject it).
    async fn persist(&self, record: StoredAccessToken) -> bool;
    /// Revoke a container token by its `jti` (called on teardown).
    async fn revoke(&self, jti: &str);
}

/// Spawns and tears down a per-container Docker Engine API socket.
///
/// Implemented in the `zlayer` bin (which can see both `zlayer-agent` and the
/// higher-level `zlayer-docker` crate — the agent itself must NOT depend on
/// `zlayer-docker`, that would be a dependency cycle). The runtime calls
/// [`spawn`](Self::spawn) when a container opts into `zlayer.io/docker-socket`
/// and [`teardown`](Self::teardown) on stop/remove.
#[async_trait]
pub trait DockerSocketSpawner: Send + Sync + std::fmt::Debug {
    /// Provision a per-container Docker Engine API socket authenticated as the
    /// container's scoped `token`, and return the HOST path of the socket to
    /// bind-mount into the container at `/var/run/docker.sock`. The path is
    /// derived deterministically from `container_key`, so the server task may
    /// still be binding asynchronously when this returns (it connects to the
    /// daemon first, then binds — near-instant since the daemon is already up).
    /// Returns `None` on failure (the container still starts, just without a
    /// docker socket). Idempotent per `container_key`.
    async fn spawn(&self, container_key: &str, token: String) -> Option<String>;
    /// Abort the per-container socket server task and unlink its socket file.
    /// Safe to call for an unknown key (no-op).
    async fn teardown(&self, container_key: &str);
}

/// Sink the runtime uses to persist a service's most-recently-resolved image
/// digest into the deployment store, so a daemon restart can recreate the
/// service from the already-local image (by digest) with zero remote/S3 traffic.
///
/// Implemented in the `zlayer` bin over the daemon's `DeploymentStorage`
/// (`SqlxStorage` in `ZLayer`, the ZQL store in `ZLayerZQL`). The agent itself
/// must NOT depend on the storage layer — that would be a dependency cycle — so
/// the concrete write lives in the bin and is threaded in as this trait object.
///
/// `record` is best-effort: a failed store update is logged by the implementor
/// and MUST NOT fail the deploy/scale that triggered the pull. Breaking the
/// boot-time circular dependency (`ZLayer` using its own `ZataStorage` S3
/// backend as the blob cache) is the whole point — see
/// [`crate::service::ServiceInstance::set_restore_pin`].
#[async_trait]
pub trait DeploymentDigestSink: Send + Sync + std::fmt::Debug {
    /// Persist `digest` (e.g. `"sha256:abc…"`) as the resolved image digest for
    /// `service` within `deployment`, keyed by service name in
    /// `StoredDeployment.resolved_image_digests`. Best-effort; never panics.
    async fn record(&self, deployment: &str, service: &str, digest: &str);
}

/// Label opting a service's containers into having the host daemon Unix socket
/// bind-mounted (Docker-in-Docker / container-management workloads). **This
/// grants the container full daemon admin** via the socket auto-auth path, so
/// it is off by default and must be set deliberately.
pub const LABEL_DAEMON_SOCKET: &str = "zlayer.io/daemon-socket";

/// Label overriding the container token's TTL, in seconds. Defaults to
/// [`DEFAULT_CONTAINER_TOKEN_TTL_SECS`].
pub const LABEL_API_TOKEN_TTL: &str = "zlayer.io/api-token-ttl";

/// Label overriding the container token's scopes. Comma-separated
/// `kind:id:level` (or `kind:level` for a wildcard id); `*` id also means
/// wildcard. Replaces the default `deployment:<own>:read` scope entirely.
pub const LABEL_API_SCOPES: &str = "zlayer.io/api-scopes";

/// Default container-token lifetime: 24h. Short relative to the old 365-day
/// token — bounds a leaked token's window — while comfortably covering a
/// container's startup-time API calls. Long-running workloads that call the API
/// past this can raise it via [`LABEL_API_TOKEN_TTL`].
pub const DEFAULT_CONTAINER_TOKEN_TTL_SECS: u64 = 24 * 60 * 60;

/// The resolved API-access policy for a container, derived from its deployment
/// + service labels.
#[derive(Debug, Clone)]
pub struct ContainerApiAccess {
    /// Scopes baked into the minted token.
    pub scopes: Vec<TokenScope>,
    /// Token lifetime.
    pub ttl: Duration,
    /// Whether to bind-mount the host admin Unix socket (full-admin opt-in).
    pub mount_socket: bool,
}

/// Parse a single `kind:id:level` / `kind:level` scope string into a
/// [`TokenScope`]. An `id` of `*` (or the 2-segment form) is a wildcard. Returns
/// `None` for a malformed scope (caller logs + skips).
fn parse_scope(s: &str) -> Option<TokenScope> {
    let parts: Vec<&str> = s.split(':').collect();
    let (kind, id, level) = match parts.as_slice() {
        [kind, level] => (*kind, None, *level),
        [kind, id, level] => (
            *kind,
            if *id == "*" || id.is_empty() {
                None
            } else {
                Some((*id).to_string())
            },
            *level,
        ),
        _ => return None,
    };
    if kind.is_empty() {
        return None;
    }
    let level = match level.to_ascii_lowercase().as_str() {
        "none" => PermissionLevel::None,
        "read" => PermissionLevel::Read,
        "execute" => PermissionLevel::Execute,
        "write" => PermissionLevel::Write,
        _ => return None,
    };
    Some(TokenScope::new(kind, id, level))
}

/// Resolve a container's API-access policy from its deployment name and service
/// labels. Default: read + `container:*:write` + `image:*:write` on its own
/// deployment (so the default per-container Docker socket can drive
/// build/run/push), [`DEFAULT_CONTAINER_TOKEN_TTL_SECS`] TTL, no admin socket.
/// Labels override each knob.
#[must_use]
pub fn resolve_container_api_access<S: std::hash::BuildHasher>(
    deployment: &str,
    labels: &HashMap<String, String, S>,
) -> ContainerApiAccess {
    let scopes = match labels.get(LABEL_API_SCOPES) {
        Some(raw) if !raw.trim().is_empty() => {
            let parsed: Vec<TokenScope> = raw
                .split(',')
                .filter_map(|s| parse_scope(s.trim()))
                .collect();
            if parsed.is_empty() {
                default_scopes_for(deployment)
            } else {
                parsed
            }
        }
        _ => default_scopes_for(deployment),
    };

    let ttl = labels
        .get(LABEL_API_TOKEN_TTL)
        .and_then(|v| v.trim().parse::<u64>().ok())
        .filter(|secs| *secs > 0)
        .unwrap_or(DEFAULT_CONTAINER_TOKEN_TTL_SECS);

    let mount_socket = labels
        .get(LABEL_DAEMON_SOCKET)
        .is_some_and(|v| matches!(v.trim().to_ascii_lowercase().as_str(), "1" | "true" | "yes"));

    ContainerApiAccess {
        scopes,
        ttl: Duration::from_secs(ttl),
        mount_socket,
    }
}

/// Default scopes for a container. Every container gets a per-container Docker
/// Engine API socket by default, and `docker` needs to create containers and
/// pull/build/push images, so grant `container:*:write` + `image:*:write` (the
/// kinds the Docker-compat endpoints authorize against) in addition to the
/// read-only own-deployment default. A deployment can replace these wholesale
/// via [`LABEL_API_SCOPES`].
#[must_use]
pub fn default_scopes_for(deployment: &str) -> Vec<TokenScope> {
    let mut scopes = default_container_scopes(deployment);
    scopes.push(TokenScope::new("container", None, PermissionLevel::Write));
    scopes.push(TokenScope::new("image", None, PermissionLevel::Write));
    scopes
}

/// The default least-privilege container scope: read-only on its own deployment.
#[must_use]
pub fn default_container_scopes(deployment: &str) -> Vec<TokenScope> {
    vec![TokenScope::new(
        "deployment",
        Some(deployment.to_string()),
        PermissionLevel::Read,
    )]
}

/// Mint a scoped JWT for a container.
///
/// The token carries `scopes` (and the marker role `container`, which grants
/// nothing on its own — authority comes from the scopes) and the supplied
/// `jti`. When `jti` is `Some`, a matching [`StoredAccessToken`] record MUST
/// already be persisted (the auth layer is fail-closed and rejects a `jti` with
/// no record); pass `None` to mint an un-revocable token bounded only by the
/// short TTL. It is signed with the daemon's JWT secret so the API accepts it.
///
/// # Errors
///
/// Returns an error string if the system clock is unavailable or JWT encoding fails.
pub fn mint_container_token(
    secret: &str,
    service_name: &str,
    container_id: &str,
    scopes: Vec<TokenScope>,
    ttl: Duration,
    jti: Option<String>,
) -> Result<String, String> {
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_err(|e| e.to_string())?;
    let claims = Claims {
        sub: format!("container:{service_name}:{container_id}"),
        iat: now.as_secs(),
        exp: (now + ttl).as_secs(),
        iss: "zlayer".to_string(),
        roles: vec!["container".to_string()],
        email: None,
        node_id: None,
        scopes,
        jti,
    };
    encode(
        &Header::default(),
        &claims,
        &EncodingKey::from_secret(secret.as_bytes()),
    )
    .map_err(|e| e.to_string())
}

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

    #[test]
    fn default_access_includes_docker_scopes() {
        let access = resolve_container_api_access("myapp", &HashMap::new());
        // Read-only own-deployment base.
        assert!(access.scopes.iter().any(|s| s.resource_kind == "deployment"
            && s.resource_id.as_deref() == Some("myapp")
            && s.level == PermissionLevel::Read));
        // Default per-container Docker socket needs container + image write.
        assert!(access.scopes.iter().any(|s| s.resource_kind == "container"
            && s.resource_id.is_none()
            && s.level == PermissionLevel::Write));
        assert!(access.scopes.iter().any(|s| s.resource_kind == "image"
            && s.resource_id.is_none()
            && s.level == PermissionLevel::Write));
        assert_eq!(
            access.ttl,
            Duration::from_secs(DEFAULT_CONTAINER_TOKEN_TTL_SECS)
        );
        assert!(!access.mount_socket);
    }

    #[test]
    fn explicit_scopes_replace_defaults() {
        let mut labels = HashMap::new();
        labels.insert(
            LABEL_API_SCOPES.to_string(),
            "deployment:foo:read".to_string(),
        );
        let access = resolve_container_api_access("foo", &labels);
        // Explicit scopes replace the defaults wholesale — no widening.
        assert_eq!(access.scopes.len(), 1);
        assert_eq!(access.scopes[0].resource_kind, "deployment");
    }

    #[test]
    fn labels_override_scopes_ttl_and_socket() {
        let mut labels = HashMap::new();
        labels.insert(
            LABEL_API_SCOPES.to_string(),
            "deployment:foo:write, job:build:execute, container:*:read".to_string(),
        );
        labels.insert(LABEL_API_TOKEN_TTL.to_string(), "3600".to_string());
        labels.insert(LABEL_DAEMON_SOCKET.to_string(), "true".to_string());
        let access = resolve_container_api_access("foo", &labels);
        assert_eq!(access.scopes.len(), 3);
        assert_eq!(access.scopes[0].level, PermissionLevel::Write);
        assert_eq!(access.scopes[1].resource_kind, "job");
        assert_eq!(access.scopes[1].resource_id.as_deref(), Some("build"));
        assert_eq!(access.scopes[2].resource_id, None); // wildcard
        assert_eq!(access.ttl, Duration::from_secs(3600));
        assert!(access.mount_socket);
    }

    #[test]
    fn malformed_scopes_fall_back_to_default() {
        let mut labels = HashMap::new();
        labels.insert(LABEL_API_SCOPES.to_string(), "garbage::::".to_string());
        let access = resolve_container_api_access("d", &labels);
        assert_eq!(access.scopes, default_scopes_for("d"));
    }

    #[test]
    fn invalid_ttl_falls_back_to_default() {
        let mut labels = HashMap::new();
        labels.insert(LABEL_API_TOKEN_TTL.to_string(), "notanumber".to_string());
        let access = resolve_container_api_access("d", &labels);
        assert_eq!(
            access.ttl,
            Duration::from_secs(DEFAULT_CONTAINER_TOKEN_TTL_SECS)
        );
    }
}