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

//! The connection side of the egress allow-list KV bucket: opening it for
//! writing or reading, and the hive's direct read. The bucket's names, subjects
//! and stored type are `swarm_wire::egress_allow`.

/// Open the allow-list bucket for writing, 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::egress_allow::BUCKET,
        "egress-allow",
        "The destinations the swarm lets agents reach through their hive's proxy".to_owned(),
    )
    .await
}

/// Open the allow-list bucket for reading, or `None` when nothing has created
/// it yet, which reads the same as no list.
///
/// **Controller-side**, for reads that must not create the bucket.
///
/// # Errors
///
/// [`crate::Error::OpenBucket`] for every failure other than the bucket not
/// existing.
#[cfg(feature = "kv")]
pub async fn open(
    client: &async_nats::Client,
) -> Result<Option<async_nats::jetstream::kv::Store>, crate::Error> {
    let js = async_nats::jetstream::new(client.clone());
    match js.get_key_value(swarm_wire::egress_allow::BUCKET).await {
        Ok(store) => Ok(Some(store)),
        Err(e) if crate::wanted::is_bucket_not_found(&e) => Ok(None),
        Err(source) => Err(crate::Error::OpenBucket {
            bucket: swarm_wire::egress_allow::BUCKET.to_owned(),
            source,
        }),
    }
}

/// What reading the list found.
#[cfg(feature = "egress-allow")]
#[derive(Debug)]
pub enum Read {
    /// The controller has written a list, which may be empty.
    Current(swarm_wire::egress_allow::AllowList),
    /// There is no list to read: no bucket answers, or the key was never
    /// written or has been deleted. A controller that predates the bucket reads
    /// this way, and so does a queue that is not serving the stream right now,
    /// so it says nothing about what the swarm wants.
    NotSet,
}

/// What reading the deny list found.
#[cfg(feature = "egress-allow")]
#[derive(Debug)]
pub enum DenyRead {
    /// The controller has written a deny list, which may be empty, at this
    /// revision of its key: the stream sequence of the write, which is what the
    /// controller's own read of the key reports as its revision.
    Current {
        list: swarm_wire::egress_allow::DenyList,
        revision: u64,
    },
    /// There is no deny list to read, for the reasons [`Read::NotSet`] gives.
    NotSet,
}

/// Read the current allow-list with one direct get.
///
/// # 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 = "egress-allow")]
pub async fn read(
    client: &async_nats::Client,
) -> Result<Read, Box<dyn std::error::Error + Send + Sync>> {
    let Some(message) = last_value(client, &swarm_wire::egress_allow::key_subject()).await? else {
        return Ok(Read::NotSet);
    };
    Ok(Read::Current(serde_json::from_slice(&message.payload)?))
}

/// Read the current deny list with one direct get.
///
/// # Errors
///
/// As [`read`].
#[cfg(feature = "egress-allow")]
pub async fn read_deny(
    client: &async_nats::Client,
) -> Result<DenyRead, Box<dyn std::error::Error + Send + Sync>> {
    let Some(message) = last_value(client, &swarm_wire::egress_allow::deny_key_subject()).await?
    else {
        return Ok(DenyRead::NotSet);
    };
    Ok(DenyRead::Current {
        list: serde_json::from_slice(&message.payload)?,
        revision: message.sequence,
    })
}

/// The last message under `subject` of the bucket's stream, or `None` when
/// there is none or it is a KV delete marker.
#[cfg(feature = "egress-allow")]
async fn last_value(
    client: &async_nats::Client,
    subject: &str,
) -> Result<
    Option<async_nats::jetstream::message::StreamMessage>,
    Box<dyn std::error::Error + Send + Sync>,
> {
    use async_nats::jetstream::stream::DirectGetErrorKind;
    use swarm_wire::egress_allow::STREAM;

    let js = async_nats::jetstream::new(client.clone());
    let stream = js.get_stream_no_info(STREAM).await?;
    let message = match stream.direct_get_last_for_subject(subject).await {
        Ok(message) => message,
        Err(e) if matches!(e.kind(), DirectGetErrorKind::NotFound) => return Ok(None),
        Err(e) if is_no_responders(&e) => return Ok(None),
        Err(e) => return Err(e.into()),
    };
    // A KV delete or purge is stored as an empty marker carrying this header.
    let removed = message
        .headers
        .get("KV-Operation")
        .is_some_and(|op| matches!(op.as_str(), "DEL" | "PURGE"));
    Ok((!removed).then_some(message))
}

#[cfg(feature = "egress-allow")]
fn is_no_responders(e: &async_nats::jetstream::stream::DirectGetError) -> bool {
    matches!(
        e.kind(),
        async_nats::jetstream::stream::DirectGetErrorKind::ErrorResponse(status, _)
            if status == async_nats::StatusCode::NO_RESPONDERS
    )
}

#[cfg(all(test, feature = "kv", feature = "egress-allow"))]
mod tests {
    use swarm_wire::egress_allow::{DENY_KEY, DenyEntry, DenyList, KEY};

    use super::{DenyRead, Read, open_or_create, read, read_deny};
    use crate::test_nats::TestNats;

    /// The deny list is read from its own key with the revision of the write,
    /// the allow-list's key is not mistaken for it, and a deleted or never
    /// written key reads as unset.
    #[tokio::test]
    async fn the_deny_list_is_read_from_its_own_key_with_the_revision_it_was_written_at() {
        let nats = TestNats::start();
        let client = nats.client().await;

        assert!(matches!(
            read_deny(&client).await.expect("no bucket"),
            DenyRead::NotSet
        ));
        let store = open_or_create(&client).await.expect("bucket");
        assert!(matches!(
            read_deny(&client).await.expect("no key"),
            DenyRead::NotSet
        ));

        let list = DenyList {
            entries: vec![DenyEntry {
                domain: ".evil.example".to_owned(),
                agent: None,
            }],
        };
        store
            .put(
                KEY,
                br#"{"entries":[{"domain":"allowed.example"}]}"#.to_vec().into(),
            )
            .await
            .expect("an allow-list write");
        assert!(
            matches!(
                read_deny(&client).await.expect("still unset"),
                DenyRead::NotSet
            ),
            "the allow-list's key is not the deny list"
        );
        let revision = store
            .put(DENY_KEY, serde_json::to_vec(&list).expect("encodes").into())
            .await
            .expect("a deny-list write");
        match read_deny(&client).await.expect("read") {
            DenyRead::Current {
                list: got,
                revision: at,
            } => {
                assert_eq!(got, list);
                assert_eq!(at, revision, "the revision the controller's write answered");
            }
            DenyRead::NotSet => panic!("the list was written"),
        }
        assert!(
            matches!(read(&client).await.expect("allow read"), Read::Current(l) if l.entries.len() == 1),
            "control: the allow-list read still finds its own key"
        );

        store.delete(DENY_KEY).await.expect("deleted");
        assert!(matches!(
            read_deny(&client).await.expect("deleted"),
            DenyRead::NotSet
        ));
    }
}