swarm-controller grants internal/docs read to agents; hives stop doing it
Agent creation gains a GrantDocsRead node (after CreateForgeUser) that adds the agent as a read collaborator on internal/docs, and forge::docs_grant runs a pass over the store's hive-agent-* roster at start and every five minutes that queues the same job for any agent that is not one. The pass only adds; an unreadable roster or forge read changes nothing. hive-c0re drops shared_docs_access, its call in sync_agent, and the SHARED_ORG / SHARED_DOCS_REPO constants only it used.
This commit is contained in:
atlas
parent
4e7142ae53
commit
1ca758daa7
7 changed files with 643 additions and 59 deletions
427
swarm-controller/src/forge/docs_grant.rs
Normal file
427
swarm-controller/src/forge/docs_grant.rs
Normal file
|
|
@ -0,0 +1,427 @@
|
|||
//! Each agent's read access to `internal/docs`: a collaborator grant at
|
||||
//! **read**, made here with the admin API so it does not depend on which hive
|
||||
//! the agent runs on or whether that hive has a forge container.
|
||||
//!
|
||||
//! Two callers insert the same `GrantDocsRead` job node: agent creation, and
|
||||
//! [`spawn`]'s pass at start and every [`RECONCILE_INTERVAL`] over every
|
||||
//! agent that holds a store identity. The pass reaches agents created before
|
||||
//! this node existed and agents whose creation-time grant failed. It only ever
|
||||
//! adds: nobody is removed, whatever the roster or the forge says. The
|
||||
//! decision is pure ([`plan`]), so the tests pin it.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use forgejo_api::structs::{AddCollaboratorOptionPermission, RepoCollaboratorPermission};
|
||||
use swarm_secret_client::{client::DEFAULT_CERT_MOUNT, policy};
|
||||
|
||||
use super::Client;
|
||||
use super::agent_token::is_not_found;
|
||||
use super::objects::{SHARED_DOCS_REPO, SHARED_ORG};
|
||||
|
||||
/// How often [`spawn`] re-checks every agent.
|
||||
const RECONCILE_INTERVAL: std::time::Duration = std::time::Duration::from_mins(5);
|
||||
|
||||
/// What one pass found for one agent.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum Observed {
|
||||
/// The agent is already a collaborator on `internal/docs`, at any level.
|
||||
Granted,
|
||||
/// The forge says the agent is not a collaborator, or has no such user —
|
||||
/// the queued job creates the user before it grants.
|
||||
Missing,
|
||||
/// The read failed. Nothing is known, so nothing is done.
|
||||
Unknown,
|
||||
}
|
||||
|
||||
/// The agents a pass queues a grant for: those the forge says are not
|
||||
/// collaborators. [`Observed::Unknown`] is never queued.
|
||||
pub fn plan(observed: &[(String, Observed)]) -> Vec<String> {
|
||||
observed
|
||||
.iter()
|
||||
.filter(|(_, o)| *o == Observed::Missing)
|
||||
.map(|(agent, _)| agent.clone())
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Whether a permission read says the user already has read access or more.
|
||||
fn holds_read(p: &RepoCollaboratorPermission) -> bool {
|
||||
matches!(
|
||||
p.permission.as_deref(),
|
||||
Some("read" | "write" | "admin" | "owner")
|
||||
)
|
||||
}
|
||||
|
||||
impl Client {
|
||||
/// Make `agent` a **read** collaborator on `internal/docs` — the whole job
|
||||
/// of the `GrantDocsRead` node. Idempotent, and never lowers: an agent that
|
||||
/// already holds read or more (write, admin, owner) is left as it is, since
|
||||
/// the add endpoint would overwrite the permission. Needs the forge user to
|
||||
/// exist.
|
||||
pub async fn grant_docs_read(&self, agent: &str) -> Result<()> {
|
||||
match self
|
||||
.api
|
||||
.repo_get_repo_permissions(SHARED_ORG, SHARED_DOCS_REPO, agent)
|
||||
.await
|
||||
{
|
||||
Ok(p) if holds_read(&p) => return Ok(()),
|
||||
Ok(_) => {}
|
||||
Err(e) if is_not_found(&e) => {}
|
||||
Err(e) => {
|
||||
return Err(e).with_context(|| {
|
||||
format!("read {agent}'s permission on {SHARED_ORG}/{SHARED_DOCS_REPO}")
|
||||
});
|
||||
}
|
||||
}
|
||||
self.add_collaborator_to(
|
||||
SHARED_ORG,
|
||||
SHARED_DOCS_REPO,
|
||||
agent,
|
||||
AddCollaboratorOptionPermission::Read,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Whether `agent` is a collaborator on `internal/docs` now.
|
||||
async fn observe_grant(&self, agent: &str) -> Observed {
|
||||
match self
|
||||
.api
|
||||
.repo_check_collaborator(SHARED_ORG, SHARED_DOCS_REPO, agent)
|
||||
.await
|
||||
{
|
||||
Ok(()) => Observed::Granted,
|
||||
Err(e) if is_not_found(&e) => Observed::Missing,
|
||||
Err(e) => {
|
||||
tracing::warn!(agent, error = %e, "docs grant: forge read failed");
|
||||
Observed::Unknown
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Whether `internal/docs` exists now: `None` when the forge could not
|
||||
/// say. A missing repo makes every collaborator check answer 404 too, so
|
||||
/// it has to be told apart from "not a collaborator" before they run.
|
||||
async fn docs_repo_exists(&self) -> Option<bool> {
|
||||
match self.api.repo_get(SHARED_ORG, SHARED_DOCS_REPO).await {
|
||||
Ok(_) => Some(true),
|
||||
Err(e) if is_not_found(&e) => Some(false),
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %e, "docs grants: forge read of the repo failed");
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// One pass over `roster`.
|
||||
async fn observe_grants(&self, roster: &[String]) -> Vec<(String, Observed)> {
|
||||
let mut observed = Vec::with_capacity(roster.len());
|
||||
for agent in roster {
|
||||
observed.push((agent.clone(), self.observe_grant(agent).await));
|
||||
}
|
||||
observed
|
||||
}
|
||||
}
|
||||
|
||||
/// The store's `hive-agent-*` cert-auth roles: exactly the agents that hold a
|
||||
/// store identity.
|
||||
async fn read_roster() -> Result<Vec<String>> {
|
||||
let store = crate::store::connect()
|
||||
.await
|
||||
.context("logging in to the swarm secret store")?;
|
||||
let roles = store
|
||||
.list_cert_roles(DEFAULT_CERT_MOUNT)
|
||||
.await
|
||||
.context("listing the store's cert-auth roles")?;
|
||||
Ok(policy::agents_from_role_names(&roles))
|
||||
}
|
||||
|
||||
/// One pass: observe every agent of `roster` and return the ones without a
|
||||
/// grant. A roster that could not be read yields none and touches nothing; so
|
||||
/// does a pass whose `internal/docs` is missing or unreadable, which is checked
|
||||
/// once before any agent.
|
||||
async fn run_pass(client: &Client, roster: Result<Vec<String>>) -> Vec<String> {
|
||||
let roster = match roster {
|
||||
Ok(roster) => roster,
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
error = %format!("{e:#}"),
|
||||
retry_in_s = RECONCILE_INTERVAL.as_secs(),
|
||||
"docs grants: no roster; retrying next tick"
|
||||
);
|
||||
return Vec::new();
|
||||
}
|
||||
};
|
||||
match client.docs_repo_exists().await {
|
||||
Some(true) => {}
|
||||
Some(false) => {
|
||||
tracing::warn!(
|
||||
repo = format!("{SHARED_ORG}/{SHARED_DOCS_REPO}"),
|
||||
"docs grants: repo missing; no grants queued"
|
||||
);
|
||||
return Vec::new();
|
||||
}
|
||||
None => return Vec::new(),
|
||||
}
|
||||
let observed = client.observe_grants(&roster).await;
|
||||
let agents = plan(&observed);
|
||||
if agents.is_empty() {
|
||||
tracing::debug!(checked = observed.len(), "docs grants: all current");
|
||||
} else {
|
||||
tracing::info!(
|
||||
checked = observed.len(),
|
||||
queueing = agents.len(),
|
||||
"docs grants: queueing grants"
|
||||
);
|
||||
}
|
||||
agents
|
||||
}
|
||||
|
||||
/// Check every agent now and every [`RECONCILE_INTERVAL`] after, and hand the
|
||||
/// agents without a grant to `enqueue`, which inserts a `GrantDocsRead` node
|
||||
/// for each. A pass that fails is logged and retried on the next tick; it
|
||||
/// never stops the daemon.
|
||||
pub fn spawn(client: Arc<Client>, enqueue: impl Fn(Vec<String>) + Send + 'static) {
|
||||
tokio::spawn(async move {
|
||||
let mut ticker = tokio::time::interval(RECONCILE_INTERVAL);
|
||||
loop {
|
||||
ticker.tick().await;
|
||||
let agents = run_pass(&client, read_roster().await).await;
|
||||
if !agents.is_empty() {
|
||||
enqueue(agents);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::collections::BTreeMap;
|
||||
use std::sync::Mutex;
|
||||
|
||||
use axum::extract::{Path, State};
|
||||
use axum::http::{StatusCode, header::CONTENT_TYPE};
|
||||
use forgejo_api::{Auth, Forgejo};
|
||||
|
||||
use super::*;
|
||||
|
||||
/// What a stand-in `internal/docs` holds: who is a collaborator, and the
|
||||
/// request log.
|
||||
#[derive(Default)]
|
||||
struct Docs {
|
||||
/// The repo does not exist: its own lookup and every collaborator
|
||||
/// check answer 404.
|
||||
repo_missing: bool,
|
||||
collaborators: BTreeMap<String, String>,
|
||||
requests: Vec<(String, String)>,
|
||||
/// Answer every collaborator check with a 500.
|
||||
broken: bool,
|
||||
}
|
||||
|
||||
type Shared = Arc<Mutex<Docs>>;
|
||||
|
||||
/// A stand-in forge on a loopback port — never the real one — serving
|
||||
/// the collaborator check (`GET`) and add (`PUT`) on `internal/docs`.
|
||||
async fn stub_forge(collaborators: &[&str], broken: bool) -> (Client, Shared) {
|
||||
serve(Docs {
|
||||
collaborators: collaborators
|
||||
.iter()
|
||||
.map(|c| ((*c).to_owned(), "read".to_owned()))
|
||||
.collect(),
|
||||
broken,
|
||||
..Docs::default()
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
async fn serve(docs: Docs) -> (Client, Shared) {
|
||||
let docs: Shared = Arc::new(Mutex::new(docs));
|
||||
let app = axum::Router::new()
|
||||
.route("/api/v1/repos/internal/docs", axum::routing::get(repo))
|
||||
.route(
|
||||
"/api/v1/repos/internal/docs/collaborators/{user}",
|
||||
axum::routing::get(check).put(add),
|
||||
)
|
||||
.route(
|
||||
"/api/v1/repos/internal/docs/collaborators/{user}/permission",
|
||||
axum::routing::get(permission),
|
||||
)
|
||||
.with_state(Arc::clone(&docs));
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let url = url::Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap();
|
||||
tokio::spawn(async move { axum::serve(listener, app).await });
|
||||
let api = Forgejo::new(Auth::Token("stub-token"), url.clone()).unwrap();
|
||||
(Client { api, url }, docs)
|
||||
}
|
||||
|
||||
async fn check(
|
||||
State(docs): State<Shared>,
|
||||
Path(user): Path<String>,
|
||||
) -> (
|
||||
StatusCode,
|
||||
[(axum::http::HeaderName, &'static str); 1],
|
||||
&'static str,
|
||||
) {
|
||||
let mut docs = docs.lock().unwrap();
|
||||
docs.requests.push(("GET".to_owned(), user.clone()));
|
||||
let status = if docs.repo_missing {
|
||||
StatusCode::NOT_FOUND
|
||||
} else if docs.broken {
|
||||
StatusCode::INTERNAL_SERVER_ERROR
|
||||
} else if docs.collaborators.contains_key(&user) {
|
||||
StatusCode::NO_CONTENT
|
||||
} else {
|
||||
StatusCode::NOT_FOUND
|
||||
};
|
||||
(status, [(CONTENT_TYPE, "application/json")], "{}")
|
||||
}
|
||||
|
||||
async fn repo(
|
||||
State(docs): State<Shared>,
|
||||
) -> (
|
||||
StatusCode,
|
||||
[(axum::http::HeaderName, &'static str); 1],
|
||||
&'static str,
|
||||
) {
|
||||
let mut docs = docs.lock().unwrap();
|
||||
docs.requests.push(("GET".to_owned(), "repo".to_owned()));
|
||||
let status = if docs.repo_missing {
|
||||
StatusCode::NOT_FOUND
|
||||
} else {
|
||||
StatusCode::OK
|
||||
};
|
||||
(status, [(CONTENT_TYPE, "application/json")], "{}")
|
||||
}
|
||||
|
||||
async fn permission(State(docs): State<Shared>, Path(user): Path<String>) -> String {
|
||||
let mut docs = docs.lock().unwrap();
|
||||
docs.requests
|
||||
.push(("GET".to_owned(), format!("{user} permission")));
|
||||
let level = docs.collaborators.get(&user).map_or("none", String::as_str);
|
||||
format!(r#"{{"permission":"{level}"}}"#)
|
||||
}
|
||||
|
||||
async fn add(State(docs): State<Shared>, Path(user): Path<String>, body: String) -> StatusCode {
|
||||
let mut docs = docs.lock().unwrap();
|
||||
docs.requests
|
||||
.push(("PUT".to_owned(), format!("{user} {body}")));
|
||||
docs.collaborators.insert(user, "read".to_owned());
|
||||
StatusCode::NO_CONTENT
|
||||
}
|
||||
|
||||
fn roster(agents: &[&str]) -> Vec<String> {
|
||||
agents.iter().map(|a| (*a).to_owned()).collect()
|
||||
}
|
||||
|
||||
fn puts(docs: &Shared) -> Vec<String> {
|
||||
docs.lock()
|
||||
.unwrap()
|
||||
.requests
|
||||
.iter()
|
||||
.filter(|(m, _)| m == "PUT")
|
||||
.map(|(_, what)| what.clone())
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn only_a_missing_grant_is_planned() {
|
||||
let observed = [
|
||||
("a".to_owned(), Observed::Granted),
|
||||
("b".to_owned(), Observed::Missing),
|
||||
("c".to_owned(), Observed::Unknown),
|
||||
];
|
||||
assert_eq!(plan(&observed), ["b"]);
|
||||
}
|
||||
|
||||
/// The `GrantDocsRead` node's work: a **read** collaborator on
|
||||
/// `internal/docs`, not write.
|
||||
#[tokio::test]
|
||||
async fn the_grant_adds_the_agent_as_a_read_collaborator() {
|
||||
let (client, docs) = stub_forge(&[], false).await;
|
||||
|
||||
client.grant_docs_read("alice").await.unwrap();
|
||||
|
||||
let sent = puts(&docs);
|
||||
assert_eq!(sent.len(), 1, "{sent:?}");
|
||||
let (user, body) = sent[0].split_once(' ').unwrap();
|
||||
assert_eq!(user, "alice");
|
||||
let body: serde_json::Value = serde_json::from_str(body).unwrap();
|
||||
assert_eq!(body["permission"], "read");
|
||||
}
|
||||
|
||||
/// The backfill queues the agents that lack the grant, and only those.
|
||||
#[tokio::test]
|
||||
async fn the_pass_queues_the_agents_missing_the_grant() {
|
||||
let (client, docs) = stub_forge(&["bob"], false).await;
|
||||
let queued = run_pass(&client, Ok(roster(&["alice", "bob", "carol"]))).await;
|
||||
|
||||
assert_eq!(queued, ["alice", "carol"]);
|
||||
assert!(puts(&docs).is_empty(), "the pass itself only reads");
|
||||
}
|
||||
|
||||
/// Once every agent holds the grant a pass queues nothing, and writes
|
||||
/// nothing — in particular it does not re-add a collaborator that holds a
|
||||
/// higher permission.
|
||||
#[tokio::test]
|
||||
async fn the_pass_is_a_no_op_for_agents_already_granted() {
|
||||
let (client, docs) = stub_forge(&["alice", "bob"], false).await;
|
||||
let queued = run_pass(&client, Ok(roster(&["alice", "bob"]))).await;
|
||||
|
||||
assert!(queued.is_empty());
|
||||
assert!(puts(&docs).is_empty());
|
||||
}
|
||||
|
||||
/// A roster that could not be read changes nothing: no grant queued, no
|
||||
/// request to the forge at all.
|
||||
#[tokio::test]
|
||||
async fn an_unreadable_roster_changes_nothing() {
|
||||
let (client, docs) = stub_forge(&["alice"], false).await;
|
||||
let queued = run_pass(&client, Err(anyhow::anyhow!("store is down"))).await;
|
||||
|
||||
assert!(queued.is_empty());
|
||||
assert!(docs.lock().unwrap().requests.is_empty());
|
||||
}
|
||||
|
||||
/// A forge that cannot answer the check is not a forge that says "not a
|
||||
/// collaborator": nothing is queued.
|
||||
#[tokio::test]
|
||||
async fn an_unreadable_forge_queues_nothing() {
|
||||
let (client, docs) = stub_forge(&[], true).await;
|
||||
let queued = run_pass(&client, Ok(roster(&["alice"]))).await;
|
||||
|
||||
assert!(queued.is_empty());
|
||||
assert!(puts(&docs).is_empty());
|
||||
}
|
||||
|
||||
/// A missing `internal/docs` answers 404 to every collaborator check, which
|
||||
/// would read as "not a collaborator" for every agent: nothing is queued,
|
||||
/// and the agents are not asked about one by one.
|
||||
#[tokio::test]
|
||||
async fn a_missing_repo_queues_nothing() {
|
||||
let (client, docs) = serve(Docs {
|
||||
repo_missing: true,
|
||||
..Docs::default()
|
||||
})
|
||||
.await;
|
||||
let queued = run_pass(&client, Ok(roster(&["alice", "bob"]))).await;
|
||||
|
||||
assert!(queued.is_empty());
|
||||
let requests = docs.lock().unwrap().requests.clone();
|
||||
assert_eq!(requests, [("GET".to_owned(), "repo".to_owned())]);
|
||||
}
|
||||
|
||||
/// An agent an operator already raised above read keeps it: the node reads
|
||||
/// the permission first and sends no write.
|
||||
#[tokio::test]
|
||||
async fn the_grant_leaves_a_higher_permission_alone() {
|
||||
let (client, docs) = serve(Docs {
|
||||
collaborators: BTreeMap::from([("alice".to_owned(), "write".to_owned())]),
|
||||
..Docs::default()
|
||||
})
|
||||
.await;
|
||||
|
||||
client.grant_docs_read("alice").await.unwrap();
|
||||
|
||||
assert!(puts(&docs).is_empty(), "{:?}", puts(&docs));
|
||||
assert_eq!(docs.lock().unwrap().collaborators["alice"], "write");
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue