ai_crew_sync/
admin_cli.rs

1//! `ai-crew-sync admin …` from the operator's own machine.
2//!
3//! Talks to `/admin/*` with an administrative credential stored by `admin
4//! login`, verifies every token it mints by presenting it to `/mcp` and
5//! checking `whoami` answers with the agent and team that were asked for,
6//! and can write the result straight into the per-team token file.
7//!
8//! Every function takes the configuration directory explicitly so the
9//! integration tests run the real flow against a temporary directory; the
10//! binary resolves it once with [`config_dir`].
11//!
12//! Secrets: the credential is read from a hidden prompt or stdin, never from
13//! an argument; a minted token is printed exactly once, or not at all when it
14//! is saved to a file. Nothing here logs.
15
16use std::{
17    io::{Read, Write},
18    path::{Path, PathBuf},
19};
20
21use anyhow::{Context, bail};
22use rmcp::{
23    ServiceExt,
24    model::{CallToolRequestParams, ClientConfig},
25    transport::{
26        StreamableHttpClientTransport, streamable_http_client::StreamableHttpClientTransportConfig,
27    },
28};
29use serde_json::{Value, json};
30use uuid::Uuid;
31
32use crate::auth::ADMIN_TOKEN_PREFIX;
33
34/// Name of the file holding the endpoint and credential inside the
35/// configuration directory.
36pub const CONFIG_FILE: &str = "admin";
37
38/// Where `admin login` keeps its state and `--save` writes token files.
39/// `BUS_CONFIG_DIR`, else `$XDG_CONFIG_HOME/ai-crew-sync`, else
40/// `$HOME/.config/ai-crew-sync`.
41pub fn config_dir() -> anyhow::Result<PathBuf> {
42    if let Some(dir) = std::env::var_os("BUS_CONFIG_DIR").filter(|v| !v.is_empty()) {
43        return Ok(PathBuf::from(dir));
44    }
45    if let Some(xdg) = std::env::var_os("XDG_CONFIG_HOME").filter(|v| !v.is_empty()) {
46        return Ok(PathBuf::from(xdg).join("ai-crew-sync"));
47    }
48    let home = std::env::var_os("HOME")
49        .filter(|v| !v.is_empty())
50        .context("neither BUS_CONFIG_DIR, XDG_CONFIG_HOME nor HOME is set")?;
51    Ok(PathBuf::from(home).join(".config").join("ai-crew-sync"))
52}
53
54/// Endpoint and credential of a logged-in administrator.
55#[derive(Clone, Debug)]
56pub struct AdminConfig {
57    /// Base URL of the bus, without a path: `https://bus.example.com:8443`.
58    pub url: String,
59    pub token: String,
60}
61
62impl AdminConfig {
63    pub fn mcp_url(&self) -> String {
64        format!("{}/mcp", self.url)
65    }
66}
67
68/// `https://host:8443`, `https://host:8443/`, `https://host:8443/mcp` and
69/// `https://host:8443/admin` all mean the same bus.
70pub fn normalize_base_url(raw: &str) -> anyhow::Result<String> {
71    let mut url = raw.trim().trim_end_matches('/').to_owned();
72    for suffix in ["/mcp", "/admin"] {
73        if let Some(stripped) = url.strip_suffix(suffix) {
74            url = stripped.trim_end_matches('/').to_owned();
75        }
76    }
77    if !(url.starts_with("http://") || url.starts_with("https://")) {
78        bail!("--url must start with http:// or https:// (got '{raw}')");
79    }
80    if url.len() <= "https://".len() {
81        bail!("--url has no host (got '{raw}')");
82    }
83    Ok(url)
84}
85
86fn config_path(dir: &Path) -> PathBuf {
87    dir.join(CONFIG_FILE)
88}
89
90/// Load the stored configuration. `BUS_ADMIN_URL` and `BUS_ADMIN_TOKEN`
91/// override the file, for scripts and CI that never run `admin login`.
92pub fn load_config(dir: &Path) -> anyhow::Result<AdminConfig> {
93    let env_url = std::env::var("BUS_ADMIN_URL")
94        .ok()
95        .filter(|v| !v.trim().is_empty());
96    let env_token = std::env::var("BUS_ADMIN_TOKEN")
97        .ok()
98        .filter(|v| !v.trim().is_empty());
99    if let (Some(url), Some(token)) = (&env_url, &env_token) {
100        return Ok(AdminConfig {
101            url: normalize_base_url(url)?,
102            token: token.trim().to_owned(),
103        });
104    }
105
106    let path = config_path(dir);
107    let text = match std::fs::read_to_string(&path) {
108        Ok(t) => t,
109        Err(e) if e.kind() == std::io::ErrorKind::NotFound => bail!(
110            "not logged in: run `ai-crew-sync admin login --url <bus>` first \
111             (or set BUS_ADMIN_URL and BUS_ADMIN_TOKEN)"
112        ),
113        Err(e) => return Err(e).with_context(|| format!("reading {}", path.display())),
114    };
115    let mut url = None;
116    let mut token = None;
117    for line in text.lines() {
118        let line = line.trim();
119        if line.is_empty() || line.starts_with('#') {
120            continue;
121        }
122        match line.split_once('=') {
123            Some(("url", v)) => url = Some(v.trim().to_owned()),
124            Some(("token", v)) => token = Some(v.trim().to_owned()),
125            _ => {}
126        }
127    }
128    let url = env_url
129        .or(url)
130        .with_context(|| format!("{} has no url= line; log in again", path.display()))?;
131    let token = env_token
132        .or(token)
133        .with_context(|| format!("{} has no token= line; log in again", path.display()))?;
134    Ok(AdminConfig {
135        url: normalize_base_url(&url)?,
136        token,
137    })
138}
139
140/// Write `content` to `path` atomically (temp file + rename) with mode 0600.
141/// A crash mid-write leaves the previous file intact, never a truncated one.
142pub fn write_private(path: &Path, content: &str) -> anyhow::Result<()> {
143    let dir = path.parent().context("path has no parent directory")?;
144    std::fs::create_dir_all(dir).with_context(|| format!("creating {}", dir.display()))?;
145    let tmp = dir.join(format!(
146        ".{}.{}.tmp",
147        path.file_name().and_then(|n| n.to_str()).unwrap_or("file"),
148        std::process::id()
149    ));
150    let mut options = std::fs::OpenOptions::new();
151    options.write(true).create(true).truncate(true);
152    #[cfg(unix)]
153    {
154        use std::os::unix::fs::OpenOptionsExt;
155        options.mode(0o600);
156    }
157    let result = (|| -> anyhow::Result<()> {
158        let mut f = options
159            .open(&tmp)
160            .with_context(|| format!("creating {}", tmp.display()))?;
161        f.write_all(content.as_bytes())?;
162        f.sync_all()?;
163        #[cfg(unix)]
164        {
165            use std::os::unix::fs::PermissionsExt;
166            // `mode` only applies at creation; a pre-existing temp file from a
167            // crashed run keeps its bits otherwise.
168            std::fs::set_permissions(&tmp, std::fs::Permissions::from_mode(0o600))?;
169        }
170        std::fs::rename(&tmp, path).with_context(|| format!("replacing {}", path.display()))?;
171        Ok(())
172    })();
173    if result.is_err() {
174        let _ = std::fs::remove_file(&tmp);
175    }
176    result
177}
178
179pub fn save_config(dir: &Path, cfg: &AdminConfig) -> anyhow::Result<PathBuf> {
180    let path = config_path(dir);
181    write_private(
182        &path,
183        &format!(
184            "# ai-crew-sync administrative credential — written by `admin login`\nurl={}\ntoken={}\n",
185            cfg.url, cfg.token
186        ),
187    )?;
188    Ok(path)
189}
190
191pub fn remove_config(dir: &Path) -> anyhow::Result<bool> {
192    match std::fs::remove_file(config_path(dir)) {
193        Ok(()) => Ok(true),
194        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
195        Err(e) => Err(e.into()),
196    }
197}
198
199// ------------------------------------------------------------ HTTP client --
200
201/// How many times a throttled (429) call is retried before giving up, and the
202/// longest single wait honoured from `Retry-After`.
203const MAX_THROTTLE_RETRIES: u32 = 5;
204const MAX_THROTTLE_WAIT_SECS: u64 = 5;
205
206/// A logged-in client for `/admin/*`.
207pub struct Api {
208    http: reqwest::Client,
209    cfg: AdminConfig,
210}
211
212impl Api {
213    pub fn new(cfg: AdminConfig) -> Self {
214        Self {
215            http: reqwest::Client::new(),
216            cfg,
217        }
218    }
219
220    pub fn config(&self) -> &AdminConfig {
221        &self.cfg
222    }
223
224    async fn call(
225        &self,
226        method: reqwest::Method,
227        path: &str,
228        body: Option<Value>,
229    ) -> anyhow::Result<Value> {
230        let url = format!("{}/admin{path}", self.cfg.url);
231        // A 429 is answered before the server does anything, so retrying it
232        // is safe for every verb, minting included. Bounded: a script that
233        // onboards twenty repositories waits a few seconds, a runaway loop
234        // still fails.
235        for attempt in 0..MAX_THROTTLE_RETRIES {
236            let mut req = self
237                .http
238                .request(method.clone(), &url)
239                .header("Authorization", format!("Bearer {}", self.cfg.token));
240            if let Some(body) = &body {
241                req = req.json(body);
242            }
243            let resp = req
244                .send()
245                .await
246                .with_context(|| format!("could not reach {url}"))?;
247            let status = resp.status();
248            if status == reqwest::StatusCode::TOO_MANY_REQUESTS
249                && attempt + 1 < MAX_THROTTLE_RETRIES
250            {
251                let wait = resp
252                    .headers()
253                    .get(reqwest::header::RETRY_AFTER)
254                    .and_then(|v| v.to_str().ok())
255                    .and_then(|v| v.parse::<u64>().ok())
256                    .unwrap_or(1)
257                    .clamp(1, MAX_THROTTLE_WAIT_SECS);
258                tokio::time::sleep(std::time::Duration::from_secs(wait)).await;
259                continue;
260            }
261            let text = resp.text().await.unwrap_or_default();
262            let value: Value = serde_json::from_str(&text).unwrap_or(Value::Null);
263            if status.is_success() {
264                return Ok(value);
265            }
266            let msg = value["error"]
267                .as_str()
268                .map(str::to_owned)
269                .unwrap_or_else(|| text.trim().to_owned());
270            bail!("{method} {path} failed ({status}): {msg}");
271        }
272        unreachable!("the retry loop returns on its last attempt")
273    }
274
275    pub async fn whoami(&self) -> anyhow::Result<Value> {
276        self.call(reqwest::Method::GET, "/whoami", None).await
277    }
278    pub async fn list_teams(&self) -> anyhow::Result<Value> {
279        self.call(reqwest::Method::GET, "/teams", None).await
280    }
281    pub async fn create_team(&self, slug: &str, name: Option<&str>) -> anyhow::Result<Value> {
282        self.call(
283            reqwest::Method::POST,
284            "/teams",
285            Some(json!({ "slug": slug, "name": name })),
286        )
287        .await
288    }
289    pub async fn list_agents(&self, team: &str) -> anyhow::Result<Value> {
290        self.call(reqwest::Method::GET, &format!("/teams/{team}/agents"), None)
291            .await
292    }
293    pub async fn create_agent(
294        &self,
295        team: &str,
296        name: &str,
297        display_name: Option<&str>,
298    ) -> anyhow::Result<Value> {
299        self.call(
300            reqwest::Method::POST,
301            &format!("/teams/{team}/agents"),
302            Some(json!({ "name": name, "display_name": display_name })),
303        )
304        .await
305    }
306    pub async fn list_tokens(&self, team: &str) -> anyhow::Result<Value> {
307        self.call(reqwest::Method::GET, &format!("/teams/{team}/tokens"), None)
308            .await
309    }
310    pub async fn issue_token(
311        &self,
312        team: &str,
313        agent: &str,
314        label: Option<&str>,
315    ) -> anyhow::Result<Issued> {
316        let v = self
317            .call(
318                reqwest::Method::POST,
319                &format!("/teams/{team}/tokens"),
320                Some(json!({ "agent": agent, "label": label })),
321            )
322            .await?;
323        let t = &v["token"];
324        let field = |k: &str| {
325            t[k].as_str()
326                .map(str::to_owned)
327                .with_context(|| format!("server response has no token.{k}"))
328        };
329        Ok(Issued {
330            id: field("id")?.parse().context("token.id is not a UUID")?,
331            token: field("token")?,
332            agent: field("agent")?,
333            team: field("team")?,
334        })
335    }
336    pub async fn revoke_token(&self, team: &str, id: Uuid) -> anyhow::Result<Value> {
337        self.call(
338            reqwest::Method::DELETE,
339            &format!("/teams/{team}/tokens/{id}"),
340            None,
341        )
342        .await
343    }
344    pub async fn list_credentials(&self) -> anyhow::Result<Value> {
345        self.call(reqwest::Method::GET, "/credentials", None).await
346    }
347    pub async fn grant_credential(
348        &self,
349        team: Option<&str>,
350        label: Option<&str>,
351    ) -> anyhow::Result<Value> {
352        self.call(
353            reqwest::Method::POST,
354            "/credentials",
355            Some(json!({ "team": team, "label": label })),
356        )
357        .await
358    }
359    pub async fn revoke_credential(&self, id: Uuid) -> anyhow::Result<Value> {
360        self.call(reqwest::Method::DELETE, &format!("/credentials/{id}"), None)
361            .await
362    }
363}
364
365/// What `/admin` minted: the secret plus the identity the server says it has.
366#[derive(Clone, Debug)]
367pub struct Issued {
368    pub id: Uuid,
369    pub token: String,
370    pub agent: String,
371    pub team: String,
372}
373
374// ----------------------------------------------------------------- login --
375
376/// Read the credential without it touching argv or shell history: from stdin
377/// when asked, else from a hidden prompt.
378pub fn read_credential(from_stdin: bool) -> anyhow::Result<String> {
379    let raw = if from_stdin {
380        let mut s = String::new();
381        std::io::stdin()
382            .read_to_string(&mut s)
383            .context("reading the credential from stdin")?;
384        s
385    } else {
386        rpassword::prompt_password("Administrative credential (acsa_…): ")
387            .context("reading the credential from the terminal")?
388    };
389    let token = raw.trim().to_owned();
390    if token.is_empty() {
391        bail!("no credential given");
392    }
393    if !token.starts_with(ADMIN_TOKEN_PREFIX) {
394        bail!(
395            "that is not an administrative credential (expected the {ADMIN_TOKEN_PREFIX} \
396             prefix). Agent tokens cannot administer the bus; mint a credential with \
397             `ai-crew-sync admin bootstrap` next to Postgres, or ask a global administrator \
398             for `admin grant`"
399        );
400    }
401    Ok(token)
402}
403
404/// Verify the credential against the bus and, only then, persist it.
405/// Returns the scope the server reported.
406pub async fn login(dir: &Path, url: &str, token: String) -> anyhow::Result<(Value, PathBuf)> {
407    let cfg = AdminConfig {
408        url: normalize_base_url(url)?,
409        token,
410    };
411    let me = Api::new(cfg.clone())
412        .whoami()
413        .await
414        .context("the bus did not accept this credential; nothing was saved")?;
415    let path = save_config(dir, &cfg)?;
416    Ok((me, path))
417}
418
419// ------------------------------------------------------- verify and save --
420
421/// Present a freshly minted token to `/mcp` and return the agent and team it
422/// authenticates as. This is the server's word, not the request's.
423pub async fn whoami_on_mcp(mcp_url: &str, token: &str) -> anyhow::Result<(String, String)> {
424    let mut config = StreamableHttpClientTransportConfig::with_uri(mcp_url.to_owned());
425    config.auth_header = Some(token.to_owned());
426    config.allow_stateless = true;
427    let transport = StreamableHttpClientTransport::from_config(config);
428    let client = ClientConfig::default()
429        .serve(transport)
430        .await
431        .context("the new token could not open an MCP session on the bus")?;
432    let outcome = async {
433        let result = client
434            .call_tool(CallToolRequestParams::new("whoami"))
435            .await
436            .context("whoami failed with the new token")?;
437        let v = result
438            .structured_content
439            .context("whoami returned no structured content")?;
440        let agent = v["agent"]
441            .as_str()
442            .context("whoami has no agent")?
443            .to_owned();
444        let team = v["team"].as_str().context("whoami has no team")?.to_owned();
445        Ok::<_, anyhow::Error>((agent, team))
446    }
447    .await;
448    let _ = client.cancel().await;
449    outcome
450}
451
452/// Where a saved token goes: `tokens-<team>` in the config directory, as the
453/// line `<repo>=<token>`.
454#[derive(Clone, Debug)]
455pub struct SaveTarget {
456    pub dir: PathBuf,
457    pub repo: String,
458}
459
460/// `--repo` names a line in a file people edit by hand: one word, no `=`,
461/// no whitespace, and never a path.
462pub fn validate_repo_name(raw: &str) -> anyhow::Result<String> {
463    let name = raw.trim();
464    if name == "_base" {
465        return Ok(name.to_owned());
466    }
467    let ok = !name.is_empty()
468        && name.len() <= 128
469        && name
470            .chars()
471            .next()
472            .is_some_and(|c| c.is_ascii_alphanumeric())
473        && name
474            .chars()
475            .all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.'));
476    if !ok {
477        bail!(
478            "--repo '{raw}' is not a valid entry name: use letters, digits, '-', '_' and '.', \
479             starting with a letter or digit (or exactly `_base` for the fallback entry)"
480        );
481    }
482    Ok(name.to_owned())
483}
484
485pub fn tokens_file(dir: &Path, team: &str) -> PathBuf {
486    dir.join(format!("tokens-{team}"))
487}
488
489/// Set `<name>=<token>` in the file, keeping every other line as it is:
490/// comments, blank lines, order, and above all `_base`. A duplicate of
491/// `name` left by a hand edit collapses to the one updated line. Atomic and
492/// 0600, and nothing else in the file is touched — a previous token for the
493/// same name is replaced in the file but never revoked on the bus.
494pub fn upsert_token_entry(path: &Path, name: &str, token: &str) -> anyhow::Result<()> {
495    let existing = match std::fs::read_to_string(path) {
496        Ok(t) => t,
497        Err(e) if e.kind() == std::io::ErrorKind::NotFound => String::new(),
498        Err(e) => return Err(e).with_context(|| format!("reading {}", path.display())),
499    };
500    let mut out = Vec::new();
501    let mut replaced = false;
502    for line in existing.lines() {
503        let is_entry = line.split_once('=').is_some_and(|(k, _)| k.trim() == name);
504        if is_entry {
505            if !replaced {
506                out.push(format!("{name}={token}"));
507                replaced = true;
508            }
509            continue;
510        }
511        out.push(line.to_owned());
512    }
513    if !replaced {
514        out.push(format!("{name}={token}"));
515    }
516    let mut content = out.join("\n");
517    content.push('\n');
518    write_private(path, &content)
519}
520
521/// Everything after the mint: verify the token on `/mcp` as exactly the
522/// requested agent and team, then either save it (and say nothing of the
523/// secret) or print it once. On a mismatch the token is revoked, nothing is
524/// written, and the error says what the server answered.
525pub async fn finish_issue(
526    api: &Api,
527    issued: &Issued,
528    expected_agent: &str,
529    expected_team: &str,
530    save: Option<&SaveTarget>,
531) -> anyhow::Result<Option<PathBuf>> {
532    let expected_agent = expected_agent.trim().to_lowercase();
533    let expected_team = expected_team.trim().to_lowercase();
534    // Best effort: the token must not stay usable if we cannot vouch for it.
535    // A failed revoke is reported alongside, never hidden. Revoked through
536    // the team that was asked for: that is the scope the credential has.
537    let mismatch = |what: String| {
538        let revoke_in = expected_team.clone();
539        async move {
540            let cleanup = match api.revoke_token(&revoke_in, issued.id).await {
541                Ok(_) => "the token has been revoked".to_owned(),
542                Err(e) => format!(
543                    "and revoking it FAILED ({e}); revoke token {} by hand",
544                    issued.id
545                ),
546            };
547            anyhow::anyhow!("{what}; {cleanup}. Nothing was saved or printed.")
548        }
549    };
550
551    let (agent, team) = match whoami_on_mcp(&api.config().mcp_url(), &issued.token).await {
552        Ok(identity) => identity,
553        Err(e) => return Err(mismatch(format!("could not verify the new token: {e:#}")).await),
554    };
555    if agent != expected_agent || team != expected_team {
556        return Err(mismatch(format!(
557            "the new token authenticates as {agent}@{team}, not {expected_agent}@{expected_team}"
558        ))
559        .await);
560    }
561    if issued.agent != agent || issued.team != team {
562        return Err(mismatch(format!(
563            "the server reported the token as {}@{} but it authenticates as {agent}@{team}",
564            issued.agent, issued.team
565        ))
566        .await);
567    }
568
569    match save {
570        Some(target) => {
571            let path = tokens_file(&target.dir, &team);
572            if let Err(e) = upsert_token_entry(&path, &target.repo, &issued.token) {
573                // A token that was never printed and could not be saved is
574                // one nobody can use: do not leave it active.
575                return Err(mismatch(format!(
576                    "the token was verified but could not be saved to {}: {e:#}",
577                    path.display()
578                ))
579                .await);
580            }
581            Ok(Some(path))
582        }
583        None => Ok(None),
584    }
585}
586
587#[cfg(test)]
588mod tests {
589    use super::*;
590
591    #[test]
592    fn base_url_is_normalised_whatever_path_was_pasted() {
593        for raw in [
594            "https://crew.example.com:8443",
595            "https://crew.example.com:8443/",
596            "https://crew.example.com:8443/mcp",
597            "https://crew.example.com:8443/admin/",
598            "  https://crew.example.com:8443/mcp/ ",
599        ] {
600            assert_eq!(
601                normalize_base_url(raw).unwrap(),
602                "https://crew.example.com:8443",
603                "{raw}"
604            );
605        }
606        assert!(normalize_base_url("crew.example.com").is_err());
607        assert!(normalize_base_url("https://").is_err());
608    }
609
610    #[test]
611    fn repo_names_are_one_safe_word() {
612        assert_eq!(validate_repo_name(" backend ").unwrap(), "backend");
613        assert_eq!(validate_repo_name("_base").unwrap(), "_base");
614        assert_eq!(validate_repo_name("0dte-api.v2").unwrap(), "0dte-api.v2");
615        for bad in ["", "_other", "-x", "a b", "a=b", "../etc", "a/b", "ünicode"] {
616            assert!(validate_repo_name(bad).is_err(), "{bad:?}");
617        }
618    }
619
620    #[test]
621    fn upsert_keeps_every_other_line_and_collapses_duplicates() {
622        let dir = std::env::temp_dir().join(format!("acs-upsert-{}", Uuid::new_v4()));
623        let path = dir.join("tokens-acme");
624        upsert_token_entry(&path, "backend", "acs_1").unwrap();
625        assert_eq!(std::fs::read_to_string(&path).unwrap(), "backend=acs_1\n");
626
627        std::fs::write(
628            &path,
629            "# hand-written\n_base=acs_base\nbackend=acs_old\n\nweb=acs_web\nbackend=acs_dup\n",
630        )
631        .unwrap();
632        upsert_token_entry(&path, "backend", "acs_new").unwrap();
633        assert_eq!(
634            std::fs::read_to_string(&path).unwrap(),
635            "# hand-written\n_base=acs_base\nbackend=acs_new\n\nweb=acs_web\n"
636        );
637        upsert_token_entry(&path, "docs", "acs_docs").unwrap();
638        assert!(
639            std::fs::read_to_string(&path)
640                .unwrap()
641                .ends_with("web=acs_web\ndocs=acs_docs\n")
642        );
643        #[cfg(unix)]
644        {
645            use std::os::unix::fs::PermissionsExt;
646            let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777;
647            assert_eq!(mode, 0o600);
648            assert!(
649                std::fs::read_dir(&dir).unwrap().count() == 1,
650                "no temp file left"
651            );
652        }
653        let _ = std::fs::remove_dir_all(&dir);
654    }
655}
ai_crew_sync/admin_cli.rs

ai_crew_sync/
admin_cli.rs
