4b1012056c
A delete removes the leaf's metadata, so the KV number restarts at 1. A login made right after a delete, before the next poll, was version 1 again and looked no newer than the recorded 1, so it never installed. The version file now records the store's write time next to the number and a later write installs whatever its number. A record from before the time was kept falls back to the number and gets its time when the same number is read again.
561 lines
20 KiB
Rust
561 lines
20 KiB
Rust
//! The claude credentials the swarm's login minter stored for this agent, at
|
|
//! `swarm/agents/<agent>/claude-credentials`, read under the agent's own
|
|
//! certificate while it waits for a login ([`crate::login::wait_for_login`]),
|
|
//! and every few minutes while it runs ([`crate::login_watch`]).
|
|
//!
|
|
//! ⚠️ **Installed only when the store's write is newer than the last one
|
|
//! installed here**, never because the bytes differ. `claude` rotates the
|
|
//! refresh token in `~/.claude/.credentials.json`, so after its first refresh
|
|
//! the stored copy is stale; putting it back would end the rotated session.
|
|
//! The last installed version is kept in [`VERSION_FILE`] under the harness
|
|
//! dir. A credentials file `wait_for_login` sees land resumes the agent as a
|
|
//! web-UI login does.
|
|
//!
|
|
//! A store that answers "nothing stored" after a version was installed here
|
|
//! means the operator deleted the stored login ([`Outcome::Deleted`]); a store
|
|
//! that does not answer is an error and means nothing.
|
|
|
|
use std::future::Future;
|
|
use std::path::{Path, PathBuf};
|
|
use std::time::Duration;
|
|
|
|
use anyhow::Context as _;
|
|
use swarm_secret_client::{
|
|
SecretStore,
|
|
claude_login::{Credentials, credentials_path},
|
|
client::DEFAULT_CERT_MOUNT,
|
|
};
|
|
|
|
/// The file under the harness dir holding the last installed version.
|
|
pub const VERSION_FILE: &str = "claude-credentials-version";
|
|
|
|
/// How often a waiting agent asks the store.
|
|
pub const POLL_INTERVAL: Duration = Duration::from_secs(30);
|
|
|
|
/// How long one read of the store may take.
|
|
const STORE_READ_TIMEOUT: Duration = Duration::from_secs(10);
|
|
|
|
/// The file `claude` reads its session from.
|
|
const CREDENTIALS_FILE: &str = ".credentials.json";
|
|
|
|
/// A stored version's identity: its KV number, and when the store wrote it, in
|
|
/// ms since the epoch. The number restarts when the operator deletes the stored
|
|
/// login, so the time is what orders a new login after the one before the
|
|
/// delete. A record from before the time was kept has none.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub struct Stamp {
|
|
pub version: u64,
|
|
pub written_ms: Option<i64>,
|
|
}
|
|
|
|
impl Stamp {
|
|
/// What the version file holds: `N`, or `N T` once the time is known.
|
|
fn encode(self) -> String {
|
|
match self.written_ms {
|
|
Some(ms) => format!("{} {ms}", self.version),
|
|
None => self.version.to_string(),
|
|
}
|
|
}
|
|
|
|
fn parse(s: &str) -> Option<Self> {
|
|
let mut parts = s.split_whitespace();
|
|
let version = parts.next()?.parse().ok()?;
|
|
let written_ms = match parts.next() {
|
|
Some(ms) => Some(ms.parse().ok()?),
|
|
None => None,
|
|
};
|
|
parts.next().is_none().then_some(Self {
|
|
version,
|
|
written_ms,
|
|
})
|
|
}
|
|
|
|
/// From the store's version: its number and RFC 3339 time. A time that
|
|
/// does not parse is no time.
|
|
fn of(version: &swarm_secret_client::client::Version) -> Self {
|
|
Self {
|
|
version: version.number,
|
|
written_ms: chrono::DateTime::parse_from_rfc3339(&version.created_time)
|
|
.ok()
|
|
.map(|t| t.timestamp_millis()),
|
|
}
|
|
}
|
|
}
|
|
|
|
/// What a stored version is to what is installed.
|
|
#[derive(Debug, PartialEq, Eq)]
|
|
pub enum Verdict {
|
|
/// Replace the installed login with it.
|
|
Install,
|
|
/// Leave the installed login alone.
|
|
Leave,
|
|
/// The same login the record names, which has no time yet: leave the login
|
|
/// and record the time.
|
|
Upgrade,
|
|
}
|
|
|
|
/// Whether a stored version replaces what is installed. Ordered by the time
|
|
/// the store wrote it when both have one, so a login made after a delete
|
|
/// installs though its number restarted; by number otherwise, where only a
|
|
/// larger one is newer. A record without a time next to the same number is
|
|
/// ambiguous (the same login, or the first after a delete), so it is left and
|
|
/// gets its time.
|
|
#[must_use]
|
|
pub fn judge(stored: Stamp, installed: Option<Stamp>) -> Verdict {
|
|
let Some(last) = installed else {
|
|
return Verdict::Install;
|
|
};
|
|
match (stored.written_ms, last.written_ms) {
|
|
(Some(s), Some(l)) if s > l => Verdict::Install,
|
|
(Some(_), None) if stored.version == last.version => Verdict::Upgrade,
|
|
(Some(_), None) => Verdict::Install,
|
|
(None, _) if stored.version > last.version => Verdict::Install,
|
|
_ => Verdict::Leave,
|
|
}
|
|
}
|
|
|
|
/// What one [`Pickup::check`] did.
|
|
#[derive(Debug, PartialEq, Eq)]
|
|
pub enum Outcome {
|
|
/// Nothing is stored for this agent, and no version was ever installed
|
|
/// here: no swarm login has reached this agent.
|
|
NothingStored,
|
|
/// The store answered that nothing is stored, though version `installed`
|
|
/// was installed here: the operator deleted the stored login.
|
|
Deleted { installed: u64 },
|
|
/// The stored version is not newer than the installed one.
|
|
NotNewer { stored: u64, installed: u64 },
|
|
/// The stored version is installed now.
|
|
Installed { version: u64 },
|
|
}
|
|
|
|
/// Where [`Pickup`] reads from. A trait so the guard can be exercised without
|
|
/// a store.
|
|
pub trait Source: Send + Sync {
|
|
/// The stored credentials and their version, or `None` when nothing is
|
|
/// stored.
|
|
fn fetch(&self) -> impl Future<Output = anyhow::Result<Option<(Credentials, Stamp)>>> + Send;
|
|
}
|
|
|
|
/// The agent's own leaf, read under its own certificate.
|
|
pub struct Store {
|
|
settings: swarm_secret_client::client::Settings,
|
|
role: String,
|
|
path: String,
|
|
}
|
|
|
|
impl Source for Store {
|
|
async fn fetch(&self) -> anyhow::Result<Option<(Credentials, Stamp)>> {
|
|
let read = async {
|
|
let store = SecretStore::connect(&self.settings, &self.role, DEFAULT_CERT_MOUNT)
|
|
.await
|
|
.context("logging in to the swarm secret store as this agent")?;
|
|
let read = store
|
|
.read_versioned(&self.path)
|
|
.await
|
|
.with_context(|| format!("reading {}", self.path))?;
|
|
Ok::<_, anyhow::Error>(read.map(|(value, version)| (value, Stamp::of(&version))))
|
|
};
|
|
tokio::time::timeout(STORE_READ_TIMEOUT, read)
|
|
.await
|
|
.with_context(|| format!("no answer from the store within {STORE_READ_TIMEOUT:?}"))?
|
|
}
|
|
}
|
|
|
|
impl Store {
|
|
/// This agent's leaf, from the environment `crate::swarm_queue` reads.
|
|
///
|
|
/// # Errors
|
|
/// A store coordinate is missing.
|
|
pub fn from_env() -> anyhow::Result<Self> {
|
|
let (settings, role, agent) = crate::swarm_queue::agent_store(|k| std::env::var(k).ok())?;
|
|
Ok(Self {
|
|
settings,
|
|
role,
|
|
path: credentials_path(&agent)?,
|
|
})
|
|
}
|
|
}
|
|
|
|
/// Installs the stored credentials into `claude_dir` when they are newer.
|
|
pub struct Pickup<S> {
|
|
source: S,
|
|
claude_dir: PathBuf,
|
|
version_file: PathBuf,
|
|
}
|
|
|
|
impl<S: Source> Pickup<S> {
|
|
pub fn new(source: S, claude_dir: PathBuf, version_file: PathBuf) -> Self {
|
|
Self {
|
|
source,
|
|
claude_dir,
|
|
version_file,
|
|
}
|
|
}
|
|
|
|
/// Read the store once and install what it holds if it is newer.
|
|
///
|
|
/// # Errors
|
|
/// The store read failed, or writing the credentials or the version did.
|
|
pub async fn check(&self) -> anyhow::Result<Outcome> {
|
|
let Some((credentials, stored)) = self.source.fetch().await? else {
|
|
return Ok(match read_stamp(&self.version_file)? {
|
|
Some(installed) => Outcome::Deleted {
|
|
installed: installed.version,
|
|
},
|
|
None => Outcome::NothingStored,
|
|
});
|
|
};
|
|
let installed = read_stamp(&self.version_file)?;
|
|
match judge(stored, installed) {
|
|
Verdict::Install => {}
|
|
verdict => {
|
|
if verdict == Verdict::Upgrade {
|
|
write_private(&self.version_file, stored.encode().as_bytes())
|
|
.context("recording when the installed credentials were stored")?;
|
|
}
|
|
return Ok(Outcome::NotNewer {
|
|
stored: stored.version,
|
|
installed: installed.map_or(0, |s| s.version),
|
|
});
|
|
}
|
|
}
|
|
write_private(
|
|
&self.claude_dir.join(CREDENTIALS_FILE),
|
|
credentials.value.as_bytes(),
|
|
)
|
|
.context("installing the stored claude credentials")?;
|
|
write_private(&self.version_file, stored.encode().as_bytes())
|
|
.context("recording the installed credentials version")?;
|
|
Ok(Outcome::Installed {
|
|
version: stored.version,
|
|
})
|
|
}
|
|
|
|
/// Forget the installed version, once the deletion it showed has been
|
|
/// acted on, so the next login installs whatever version it gets.
|
|
///
|
|
/// # Errors
|
|
/// Removing the version file failed for a reason other than its absence.
|
|
pub fn forget_installed(&self) -> anyhow::Result<()> {
|
|
match std::fs::remove_file(&self.version_file) {
|
|
Ok(()) => Ok(()),
|
|
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
|
|
Err(e) => Err(e).with_context(|| format!("removing {}", self.version_file.display())),
|
|
}
|
|
}
|
|
}
|
|
|
|
/// The version recorded in `file`, or `None` when there is no file.
|
|
fn read_stamp(file: &Path) -> anyhow::Result<Option<Stamp>> {
|
|
match std::fs::read_to_string(file) {
|
|
Ok(s) => Stamp::parse(&s)
|
|
.map(Some)
|
|
.with_context(|| format!("{} does not hold a version", file.display())),
|
|
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
|
|
Err(e) => Err(e).with_context(|| format!("reading {}", file.display())),
|
|
}
|
|
}
|
|
|
|
/// Replace `path` with `bytes`, mode 0600, through a sibling temp file so a
|
|
/// reader never sees half of it.
|
|
fn write_private(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
|
|
use std::io::Write as _;
|
|
use std::os::unix::fs::OpenOptionsExt as _;
|
|
|
|
let mut tmp = path.as_os_str().to_owned();
|
|
tmp.push(".tmp");
|
|
let tmp = PathBuf::from(tmp);
|
|
let mut f = std::fs::OpenOptions::new()
|
|
.write(true)
|
|
.create(true)
|
|
.truncate(true)
|
|
.mode(0o600)
|
|
.open(&tmp)?;
|
|
f.write_all(bytes)?;
|
|
f.sync_all()?;
|
|
std::fs::rename(&tmp, path)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use std::sync::Mutex;
|
|
|
|
use super::*;
|
|
|
|
struct Fake(Mutex<Option<(String, Stamp)>>);
|
|
|
|
impl Source for Fake {
|
|
async fn fetch(&self) -> anyhow::Result<Option<(Credentials, Stamp)>> {
|
|
Ok(self
|
|
.0
|
|
.lock()
|
|
.expect("lock")
|
|
.clone()
|
|
.map(|(value, stamp)| (Credentials { value }, stamp)))
|
|
}
|
|
}
|
|
|
|
struct Dir(PathBuf);
|
|
|
|
impl Drop for Dir {
|
|
fn drop(&mut self) {
|
|
let _ = std::fs::remove_dir_all(&self.0);
|
|
}
|
|
}
|
|
|
|
fn dir(name: &str) -> Dir {
|
|
let d = std::env::temp_dir().join(format!(
|
|
"hive-agent-claude-credentials-{name}-{}",
|
|
std::process::id()
|
|
));
|
|
let _ = std::fs::remove_dir_all(&d);
|
|
std::fs::create_dir_all(d.join("claude")).expect("temp dir");
|
|
Dir(d)
|
|
}
|
|
|
|
fn stamp(version: u64, written_ms: Option<i64>) -> Stamp {
|
|
Stamp {
|
|
version,
|
|
written_ms,
|
|
}
|
|
}
|
|
|
|
/// A pickup over a stored login that carries no write time.
|
|
fn pickup(d: &Dir, stored: Option<(&str, u64)>) -> Pickup<Fake> {
|
|
pickup_stamped(d, stored.map(|(v, n)| (v, stamp(n, None))))
|
|
}
|
|
|
|
fn pickup_stamped(d: &Dir, stored: Option<(&str, Stamp)>) -> Pickup<Fake> {
|
|
Pickup::new(
|
|
Fake(Mutex::new(stored.map(|(v, s)| (v.to_owned(), s)))),
|
|
d.0.join("claude"),
|
|
d.0.join(VERSION_FILE),
|
|
)
|
|
}
|
|
|
|
fn set(p: &Pickup<Fake>, stored: Option<(&str, Stamp)>) {
|
|
*p.source.0.lock().expect("lock") = stored.map(|(v, s)| (v.to_owned(), s));
|
|
}
|
|
|
|
fn installed(d: &Dir) -> Option<String> {
|
|
std::fs::read_to_string(d.0.join("claude").join(CREDENTIALS_FILE)).ok()
|
|
}
|
|
|
|
fn recorded(d: &Dir) -> Option<Stamp> {
|
|
read_stamp(&d.0.join(VERSION_FILE)).expect("read")
|
|
}
|
|
|
|
#[test]
|
|
fn without_times_only_a_larger_number_replaces_the_installed_one() {
|
|
assert_eq!(judge(stamp(1, None), None), Verdict::Install);
|
|
assert_eq!(
|
|
judge(stamp(3, None), Some(stamp(2, None))),
|
|
Verdict::Install
|
|
);
|
|
assert_eq!(judge(stamp(2, None), Some(stamp(2, None))), Verdict::Leave);
|
|
assert_eq!(judge(stamp(1, None), Some(stamp(2, None))), Verdict::Leave);
|
|
}
|
|
|
|
#[test]
|
|
fn with_times_the_later_write_wins_whatever_its_number() {
|
|
let before = Some(stamp(1, Some(1_000)));
|
|
assert_eq!(
|
|
judge(stamp(1, Some(2_000)), before),
|
|
Verdict::Install,
|
|
"a login after a delete restarts at 1 and is still newer"
|
|
);
|
|
assert_eq!(judge(stamp(1, Some(1_000)), before), Verdict::Leave);
|
|
assert_eq!(
|
|
judge(stamp(5, Some(900)), before),
|
|
Verdict::Leave,
|
|
"an older write is not newer for its larger number"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn a_record_without_a_time_gets_one_when_the_number_matches() {
|
|
let legacy = Some(stamp(1, None));
|
|
assert_eq!(judge(stamp(1, Some(5)), legacy), Verdict::Upgrade);
|
|
assert_eq!(judge(stamp(2, Some(5)), legacy), Verdict::Install);
|
|
assert_eq!(
|
|
judge(stamp(1, Some(5)), Some(stamp(3, None))),
|
|
Verdict::Install,
|
|
"a smaller number than recorded can only be a restarted counter"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn the_version_file_round_trips_and_reads_the_old_form() {
|
|
for s in [stamp(4, None), stamp(1, Some(1_731_000_000_123))] {
|
|
assert_eq!(Stamp::parse(&s.encode()), Some(s));
|
|
}
|
|
assert_eq!(Stamp::parse("7\n"), Some(stamp(7, None)));
|
|
assert_eq!(Stamp::parse("7 x"), None);
|
|
assert_eq!(Stamp::parse("7 8 9"), None);
|
|
assert_eq!(Stamp::parse(""), None);
|
|
}
|
|
|
|
#[test]
|
|
fn the_stores_rfc3339_time_is_read_to_ms() {
|
|
let v = swarm_secret_client::client::Version {
|
|
number: 3,
|
|
created_time: "2026-11-09T20:00:00.123456789Z".to_owned(),
|
|
};
|
|
assert_eq!(
|
|
Stamp::of(&v),
|
|
stamp(3, Some(1_794_254_400_123)),
|
|
"nanoseconds are cut to ms"
|
|
);
|
|
let v = swarm_secret_client::client::Version {
|
|
number: 3,
|
|
created_time: "not a time".to_owned(),
|
|
};
|
|
assert_eq!(Stamp::of(&v), stamp(3, None));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn the_first_stored_version_is_installed_private_and_recorded() {
|
|
use std::os::unix::fs::PermissionsExt as _;
|
|
|
|
let d = dir("first");
|
|
let p = pickup(&d, Some(("{\"v\":1}", 1)));
|
|
assert_eq!(
|
|
p.check().await.expect("check"),
|
|
Outcome::Installed { version: 1 }
|
|
);
|
|
assert_eq!(installed(&d).as_deref(), Some("{\"v\":1}"));
|
|
let mode = std::fs::metadata(d.0.join("claude").join(CREDENTIALS_FILE))
|
|
.expect("installed")
|
|
.permissions()
|
|
.mode();
|
|
assert_eq!(mode & 0o777, 0o600);
|
|
assert_eq!(recorded(&d), Some(stamp(1, None)));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn an_equal_or_older_version_leaves_the_rotated_session_alone() {
|
|
let d = dir("stale");
|
|
let p = pickup_stamped(&d, Some(("{\"v\":2}", stamp(2, Some(2_000)))));
|
|
p.check().await.expect("first install");
|
|
// claude refreshes the session in place.
|
|
std::fs::write(d.0.join("claude").join(CREDENTIALS_FILE), "rotated").expect("refresh");
|
|
assert_eq!(
|
|
p.check().await.expect("check"),
|
|
Outcome::NotNewer {
|
|
stored: 2,
|
|
installed: 2
|
|
}
|
|
);
|
|
set(&p, Some(("{\"v\":1}", stamp(1, Some(1_000)))));
|
|
assert_eq!(
|
|
p.check().await.expect("check"),
|
|
Outcome::NotNewer {
|
|
stored: 1,
|
|
installed: 2
|
|
}
|
|
);
|
|
assert_eq!(installed(&d).as_deref(), Some("rotated"));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_relogin_is_installed_over_the_rotated_session() {
|
|
let d = dir("relogin");
|
|
let p = pickup_stamped(&d, Some(("{\"v\":1}", stamp(1, Some(1_000)))));
|
|
p.check().await.expect("first install");
|
|
std::fs::write(d.0.join("claude").join(CREDENTIALS_FILE), "rotated").expect("refresh");
|
|
set(&p, Some(("{\"v\":2}", stamp(2, Some(2_000)))));
|
|
assert_eq!(
|
|
p.check().await.expect("check"),
|
|
Outcome::Installed { version: 2 }
|
|
);
|
|
assert_eq!(installed(&d).as_deref(), Some("{\"v\":2}"));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn nothing_stored_installs_nothing() {
|
|
let d = dir("empty");
|
|
let p = pickup(&d, None);
|
|
assert_eq!(p.check().await.expect("check"), Outcome::NothingStored);
|
|
assert_eq!(installed(&d), None);
|
|
assert_eq!(recorded(&d), None);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_login_deleted_after_an_install_is_reported_and_forgotten_on_request() {
|
|
let d = dir("deleted");
|
|
let p = pickup_stamped(&d, Some(("{\"v\":1}", stamp(1, Some(1_000)))));
|
|
p.check().await.expect("install");
|
|
set(&p, None);
|
|
assert_eq!(
|
|
p.check().await.expect("check"),
|
|
Outcome::Deleted { installed: 1 }
|
|
);
|
|
// Reporting it changes nothing: the caller acts, then forgets.
|
|
assert_eq!(
|
|
p.check().await.expect("check again"),
|
|
Outcome::Deleted { installed: 1 }
|
|
);
|
|
p.forget_installed().expect("forget");
|
|
p.forget_installed().expect("forgetting twice is fine");
|
|
assert_eq!(p.check().await.expect("check"), Outcome::NothingStored);
|
|
set(&p, Some(("{\"v\":1}", stamp(1, Some(3_000)))));
|
|
assert_eq!(
|
|
p.check().await.expect("a new login"),
|
|
Outcome::Installed { version: 1 },
|
|
"the counter restarts after a delete, and its first login still installs"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_login_made_between_two_polls_after_a_delete_still_installs() {
|
|
// The agent installed version 1. The operator deletes it and logs in
|
|
// again before the next poll, so the store never answers "nothing
|
|
// stored": it answers version 1 again, written later.
|
|
let d = dir("between");
|
|
let p = pickup_stamped(&d, Some(("{\"old\":1}", stamp(1, Some(1_000)))));
|
|
p.check().await.expect("install");
|
|
std::fs::write(d.0.join("claude").join(CREDENTIALS_FILE), "rotated").expect("refresh");
|
|
set(&p, Some(("{\"new\":1}", stamp(1, Some(9_000)))));
|
|
assert_eq!(
|
|
p.check().await.expect("check"),
|
|
Outcome::Installed { version: 1 }
|
|
);
|
|
assert_eq!(installed(&d).as_deref(), Some("{\"new\":1}"));
|
|
assert_eq!(recorded(&d), Some(stamp(1, Some(9_000))));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_record_from_before_the_time_was_kept_gets_its_time_and_keeps_the_session() {
|
|
let d = dir("upgrade");
|
|
std::fs::write(d.0.join(VERSION_FILE), "1").expect("an old record");
|
|
std::fs::write(d.0.join("claude").join(CREDENTIALS_FILE), "rotated").expect("session");
|
|
let p = pickup_stamped(&d, Some(("{\"v\":1}", stamp(1, Some(1_000)))));
|
|
assert_eq!(
|
|
p.check().await.expect("check"),
|
|
Outcome::NotNewer {
|
|
stored: 1,
|
|
installed: 1
|
|
}
|
|
);
|
|
assert_eq!(installed(&d).as_deref(), Some("rotated"));
|
|
assert_eq!(recorded(&d), Some(stamp(1, Some(1_000))));
|
|
}
|
|
|
|
struct Failing;
|
|
|
|
impl Source for Failing {
|
|
async fn fetch(&self) -> anyhow::Result<Option<(Credentials, Stamp)>> {
|
|
anyhow::bail!("the store did not answer")
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_failed_read_is_an_error_never_a_deletion() {
|
|
let d = dir("failing");
|
|
std::fs::write(d.0.join(VERSION_FILE), "4").expect("an installed version");
|
|
let p = Pickup::new(Failing, d.0.join("claude"), d.0.join(VERSION_FILE));
|
|
assert!(p.check().await.is_err());
|
|
assert_eq!(recorded(&d), Some(stamp(4, None)));
|
|
}
|
|
}
|