hyperhive/swarm-queue-client/src/agent_permissions.rs

//! Opening and reading the per-agent permissions KV bucket, and the copy
//! hive-c0re keeps of a record for the agent's own processes. The names, the
//! stored type and the import messages are `swarm_wire::agent_permissions`;
//! reading needs the `agent-permissions` feature and writing needs `kv`.

#[cfg(feature = "agent-permissions")]
use swarm_wire::agent_permissions::AgentPermissions;

/// Open the permissions bucket, creating it if nothing has yet.
///
/// **Controller-side only.** `history: 1`: a hive wants the current value.
#[cfg(feature = "kv")]
pub async fn open_or_create(
    client: &async_nats::Client,
) -> Result<async_nats::jetstream::kv::Store, crate::Error> {
    let js = async_nats::jetstream::new(client.clone());
    crate::kv::open_or_create(
        &js,
        swarm_wire::agent_permissions::BUCKET,
        "agent-permissions",
        "Permissions the swarm controller sets for each agent".to_owned(),
    )
    .await
}

/// What reading one agent's key found.
#[cfg(feature = "agent-permissions")]
#[derive(Debug)]
pub enum Read {
    /// The bucket answered; an absent or deleted key has every field unset.
    Current(AgentPermissions),
    /// Nothing answers for the bucket's stream. Not an unset value: it says
    /// nothing about the agent's permissions.
    NoBucket,
}

/// The file hive-c0re writes an agent's [`HiveCopy`] to in that agent's
/// runtime dir.
#[cfg(feature = "agent-permissions")]
pub const HIVE_COPY_FILE: &str = "swarm-permissions.json";

/// Where the agent's own processes read [`HIVE_COPY_FILE`]: the runtime dir
/// is mounted at `/run/hive` in the container.
#[cfg(feature = "agent-permissions")]
pub const HIVE_COPY_PATH: &str = "/run/hive/swarm-permissions.json";

/// What hive-c0re holds for one agent: its record as the hive last stored it
/// (read from this bucket, or taken from the meta repo before the import),
/// and whether the meta repo's import is still pending. The agent's harness
/// and subagent daemon hold [`HiveCopy::before_read`] until their own first
/// read of the key succeeds, and judge every read with
/// [`HiveCopy::after_read`].
#[cfg(feature = "agent-permissions")]
#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct HiveCopy {
    #[serde(flatten)]
    pub stored: AgentPermissions,
    #[serde(default)]
    pub import_pending: bool,
}

#[cfg(feature = "agent-permissions")]
impl HiveCopy {
    /// The copy at `path`. A missing or unreadable file is the empty copy,
    /// which holds the default and has no import pending.
    #[must_use]
    pub fn read_at(path: &std::path::Path) -> Self {
        let bytes = match std::fs::read(path) {
            Ok(bytes) => bytes,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Self::default(),
            Err(e) => {
                tracing::warn!(path = %path.display(), error = %e, "permissions: the hive's copy is unreadable");
                return Self::default();
            }
        };
        serde_json::from_slice(&bytes).unwrap_or_else(|e| {
            tracing::warn!(path = %path.display(), error = %e, "permissions: the hive's copy does not decode");
            Self::default()
        })
    }

    /// What the agent holds before any read of its key has succeeded: the
    /// stored list, or the default when the hive stored none.
    #[must_use]
    pub fn before_read(&self) -> Vec<hive_types::Permission> {
        self.stored
            .permissions
            .clone()
            .unwrap_or_else(|| hive_types::Permission::DEFAULT.to_vec())
    }

    /// What the agent holds after reading `read` from its key. An unset key
    /// while the import is pending may only mean the grant is not imported
    /// yet, so the stored list stands; otherwise unset is the default.
    #[must_use]
    pub fn after_read(&self, read: &AgentPermissions) -> Vec<hive_types::Permission> {
        match &read.permissions {
            Some(list) => list.clone(),
            None if self.import_pending => self.before_read(),
            None => hive_types::Permission::DEFAULT.to_vec(),
        }
    }
}

/// Read `agent`'s current permissions with one direct get: the one subject
/// an agent's own grant covers, and one of those a hive's covers.
///
/// # Errors
///
/// A failed request or an undecodable value. A request the queue refused for
/// permissions reaches the client as a timeout, not as a refusal.
#[cfg(feature = "agent-permissions")]
pub async fn read(
    client: &async_nats::Client,
    agent: &str,
) -> Result<Read, Box<dyn std::error::Error + Send + Sync>> {
    use async_nats::jetstream::stream::DirectGetErrorKind;

    let js = async_nats::jetstream::new(client.clone());
    let stream = js
        .get_stream_no_info(swarm_wire::agent_permissions::STREAM)
        .await?;
    let message = match stream
        .direct_get_last_for_subject(swarm_wire::agent_permissions::key_subject(agent))
        .await
    {
        Ok(message) => message,
        Err(e) if matches!(e.kind(), DirectGetErrorKind::NotFound) => {
            return Ok(Read::Current(AgentPermissions::default()));
        }
        Err(e)
            if matches!(
                e.kind(),
                DirectGetErrorKind::ErrorResponse(status, _)
                    if status == async_nats::StatusCode::NO_RESPONDERS
            ) =>
        {
            return Ok(Read::NoBucket);
        }
        Err(e) => return Err(e.into()),
    };
    Ok(Read::Current(decode(
        message
            .headers
            .get("KV-Operation")
            .map(async_nats::HeaderValue::as_str),
        &message.payload,
    )?))
}

/// One stored message as a value: a KV delete or purge is stored as an empty
/// marker carrying the `KV-Operation` header, and reads as every field unset.
#[cfg(feature = "agent-permissions")]
fn decode(operation: Option<&str>, payload: &[u8]) -> serde_json::Result<AgentPermissions> {
    if matches!(operation, Some("DEL" | "PURGE")) {
        return Ok(AgentPermissions::default());
    }
    serde_json::from_slice(payload)
}

#[cfg(all(test, feature = "agent-permissions"))]
mod tests {
    use super::*;

    /// Before a read the stored list stands; after one, a set key wins and an
    /// unset key is the default unless the import is pending.
    #[cfg(feature = "agent-permissions")]
    #[test]
    fn the_hive_copy_stands_until_read_and_while_the_import_is_pending() {
        use hive_types::Permission;
        let record = |json: &str| -> AgentPermissions { serde_json::from_str(json).unwrap() };
        let copy = |json: &str| -> HiveCopy { serde_json::from_str(json).unwrap() };
        let web = vec![Permission::WebTools];
        let default = Permission::DEFAULT.to_vec();

        assert_eq!(copy(r#"{"permissions":["web_tools"]}"#).before_read(), web);
        assert_eq!(copy("{}").before_read(), default);

        let pending = copy(r#"{"permissions":["web_tools"],"import_pending":true}"#);
        assert_eq!(pending.after_read(&record("{}")), web);
        assert_eq!(
            pending.after_read(&record(r#"{"permissions":[]}"#)),
            Vec::new()
        );
        let settled = copy(r#"{"permissions":["web_tools"]}"#);
        assert_eq!(settled.after_read(&record("{}")), default);
    }

    /// Fields a newer controller adds must not cost an older hive the ones it
    /// knows; a deleted key is unset.
    #[cfg(feature = "agent-permissions")]
    #[test]
    fn decoding_tolerates_unknown_fields_and_reads_a_delete_as_unset() {
        let decoded = decode(
            None,
            br#"{"permissions":["manage_root_agent"],"later_field":["meta"]}"#,
        )
        .unwrap();
        assert_eq!(decoded.permissions.map(|c| c.len()), Some(1));
        assert_eq!(
            decode(Some("DEL"), b"").unwrap(),
            AgentPermissions::default()
        );
        assert_eq!(
            decode(Some("PURGE"), b"").unwrap(),
            AgentPermissions::default()
        );
    }
}