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

ai_crew_sync/
admin_cli.rs
