car_auth/

lib.rs

//! Shared Parslee OAuth2 PKCE + token/keychain logic.
//!
//! Used by `car-cli` (`car auth login parslee`, loopback flow) and by
//! `car-server` (the `auth.*` JSON-RPC surface that CAR Host.app's
//! signup GUI drives). The keychain keys + default service exactly
//! match what `car-inference` reads at request time
//! (`PARSLEE_ACCESS_TOKEN`, default `"car"` service) — see
//! `car-inference` `remote.rs::lease_key`.

use base64::Engine;
use serde::Deserialize;
use sha2::{Digest, Sha256};

use car_secrets::{SecretRef, SecretStore};

pub const PARSLEE_ACCESS_TOKEN_KEY: &str = "PARSLEE_ACCESS_TOKEN";
pub const PARSLEE_REFRESH_TOKEN_KEY: &str = "PARSLEE_REFRESH_TOKEN";
pub const PARSLEE_EXPIRES_AT_KEY: &str = "PARSLEE_ACCESS_TOKEN_EXPIRES_AT";
pub const PARSLEE_API_BASE_KEY: &str = "PARSLEE_API_BASE";
pub const DEFAULT_API_BASE: &str = "https://api.parslee.ai";

/// `/connect/token` success body.
#[derive(Debug, Clone, Deserialize)]
pub struct TokenSet {
    pub access_token: String,
    pub refresh_token: String,
    pub expires_in: u64,
    pub token_type: String,
}

fn epoch_seconds() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

/// PKCE code verifier (URL-safe, no padding).
pub fn pkce_verifier() -> String {
    let raw = format!(
        "{}{}",
        uuid::Uuid::new_v4().simple(),
        uuid::Uuid::new_v4().simple()
    );
    base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(raw.as_bytes())
}

/// Opaque OAuth `state` value (CSRF guard).
pub fn new_state() -> String {
    uuid::Uuid::new_v4().simple().to_string()
}

/// PKCE S256 challenge for a verifier.
pub fn pkce_challenge(verifier: &str) -> String {
    let digest = Sha256::digest(verifier.as_bytes());
    base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(digest)
}

/// Build the `/connect/authorize` URL the user opens in a browser.
pub fn authorize_url(
    api_base: &str,
    client_id: &str,
    redirect_uri: &str,
    state: &str,
    challenge: &str,
    provider: Option<&str>,
) -> Result<String, String> {
    let mut url = reqwest::Url::parse(&format!(
        "{}/connect/authorize",
        api_base.trim_end_matches('/')
    ))
    .map_err(|e| format!("build authorize URL: {e}"))?;
    url.query_pairs_mut()
        .append_pair("client_id", client_id)
        .append_pair("redirect_uri", redirect_uri)
        .append_pair("response_type", "code")
        .append_pair("scope", "openid profile email")
        .append_pair("state", state)
        .append_pair("code_challenge", challenge)
        .append_pair("code_challenge_method", "S256");
    if let Some(provider) = provider {
        url.query_pairs_mut().append_pair("provider", provider);
    }
    Ok(url.to_string())
}

fn form_body(pairs: &[(&str, &str)]) -> String {
    let mut s = String::new();
    for (i, (k, v)) in pairs.iter().enumerate() {
        if i > 0 {
            s.push('&');
        }
        s.push_str(&urlencode(k));
        s.push('=');
        s.push_str(&urlencode(v));
    }
    s
}

fn urlencode(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    for b in s.bytes() {
        match b {
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
                out.push(b as char)
            }
            _ => out.push_str(&format!("%{b:02X}")),
        }
    }
    out
}

/// Exchange an authorization code + PKCE verifier for tokens.
pub async fn exchange_code(
    api_base: &str,
    client_id: &str,
    redirect_uri: &str,
    code: &str,
    verifier: &str,
) -> Result<TokenSet, String> {
    let body = form_body(&[
        ("grant_type", "authorization_code"),
        ("client_id", client_id),
        ("redirect_uri", redirect_uri),
        ("code", code),
        ("code_verifier", verifier),
    ]);
    let token_url = format!("{}/connect/token", api_base.trim_end_matches('/'));
    let response = reqwest::Client::new()
        .post(token_url)
        .header("content-type", "application/x-www-form-urlencoded")
        .body(body)
        .send()
        .await
        .map_err(|e| format!("exchange Parslee authorization code: {e}"))?;
    let status = response.status();
    let text = response
        .text()
        .await
        .map_err(|e| format!("read token response: {e}"))?;
    if !status.is_success() {
        return Err(format!(
            "Parslee token exchange failed: HTTP {status}: {text}"
        ));
    }
    let token: TokenSet =
        serde_json::from_str(&text).map_err(|e| format!("parse token response: {e}"))?;
    if !token.token_type.eq_ignore_ascii_case("bearer") {
        return Err(format!(
            "unexpected Parslee token_type `{}`",
            token.token_type
        ));
    }
    Ok(token)
}

fn put(key: &str, value: &str) -> Result<(), String> {
    SecretStore::new()
        .put(&SecretRef::with_default_service(key), value)
        .map_err(|e| format!("store {key}: {e}"))
}

/// Persist a token set + the API base into the OS keychain (default
/// `"car"` service — the same place `car-inference` reads from).
pub fn store_tokens(api_base: &str, token: &TokenSet) -> Result<(), String> {
    put(PARSLEE_ACCESS_TOKEN_KEY, &token.access_token)?;
    put(PARSLEE_REFRESH_TOKEN_KEY, &token.refresh_token)?;
    put(PARSLEE_API_BASE_KEY, api_base.trim_end_matches('/'))?;
    put(
        PARSLEE_EXPIRES_AT_KEY,
        &(epoch_seconds() + token.expires_in).to_string(),
    )?;
    Ok(())
}

/// Remove all stored Parslee credentials. Idempotent.
pub fn clear_tokens() -> Result<(), String> {
    let store = SecretStore::new();
    for key in [
        PARSLEE_ACCESS_TOKEN_KEY,
        PARSLEE_REFRESH_TOKEN_KEY,
        PARSLEE_EXPIRES_AT_KEY,
        PARSLEE_API_BASE_KEY,
    ] {
        let _ = store.delete(&SecretRef::with_default_service(key));
    }
    Ok(())
}

/// Current access token (env override first, then keychain).
pub fn access_token() -> Option<String> {
    car_secrets::resolve_env_or_keychain(PARSLEE_ACCESS_TOKEN_KEY)
}

/// Seconds before the stored expiry at which [`access_token_refreshing`]
/// proactively refreshes — absorbs clock skew plus a slow request. Public so
/// the daemon's `load_or_refresh` shares the same threshold (#320).
pub const REFRESH_SKEW_SECS: u64 = 120;

/// Result of a [`refresh_grant`]. The gateway may omit a rotated refresh
/// token (reuse the prior one) and/or an expiry, so both are optional.
#[derive(Debug, Clone)]
pub struct RefreshedTokens {
    pub access_token: String,
    pub refresh_token: Option<String>,
    pub expires_in: Option<u64>,
}

/// `refresh_token` grant against `/connect/token`. Network-only — the
/// caller persists. Mirrors the Parslee gateway contract used by the
/// daemon's own refresh path (`car-server-core::parslee_auth`): the
/// gateway treats this as a public-client grant, so no `client_id` is
/// sent. This lives in `car-auth` (not `car-server-core`) so the
/// request-time inference path — which cannot depend on `car-server-core`
/// — shares one definition of "mint a fresh Parslee bearer" (#313).
pub async fn refresh_grant(api_base: &str, refresh_token: &str) -> Result<RefreshedTokens, String> {
    #[derive(Deserialize)]
    struct Resp {
        access_token: String,
        #[serde(default)]
        refresh_token: Option<String>,
        #[serde(default)]
        expires_in: Option<u64>,
    }
    let body = form_body(&[
        ("grant_type", "refresh_token"),
        ("refresh_token", refresh_token),
    ]);
    let token_url = format!("{}/connect/token", api_base.trim_end_matches('/'));
    let response = reqwest::Client::new()
        .post(token_url)
        .header("content-type", "application/x-www-form-urlencoded")
        .body(body)
        .send()
        .await
        .map_err(|e| format!("refresh Parslee token: {e}"))?;
    let status = response.status();
    let text = response
        .text()
        .await
        .map_err(|e| format!("read Parslee token response: {e}"))?;
    if !status.is_success() {
        return Err(format!("refresh Parslee token: HTTP {status}: {text}"));
    }
    let r: Resp =
        serde_json::from_str(&text).map_err(|e| format!("parse Parslee token response: {e}"))?;
    Ok(RefreshedTokens {
        access_token: r.access_token,
        refresh_token: r.refresh_token,
        expires_in: r.expires_in,
    })
}

/// Persist refreshed credentials to the keychain (the same keys
/// `car-inference` reads). Best-effort: a keychain write failure must not
/// fail the in-flight request — the returned access token still works.
fn persist_refreshed(api_base: &str, t: &RefreshedTokens) {
    let _ = put(PARSLEE_ACCESS_TOKEN_KEY, &t.access_token);
    if let Some(refresh) = &t.refresh_token {
        let _ = put(PARSLEE_REFRESH_TOKEN_KEY, refresh);
    }
    if let Some(expires_in) = t.expires_in {
        let _ = put(
            PARSLEE_EXPIRES_AT_KEY,
            &(epoch_seconds() + expires_in).to_string(),
        );
    }
    let _ = put(PARSLEE_API_BASE_KEY, api_base.trim_end_matches('/'));
}

/// Current access token, **proactively refreshed** when the stored token
/// is within [`REFRESH_SKEW_SECS`] of expiry (or already expired) and a
/// refresh token is available. The `PARSLEE_ACCESS_TOKEN` env override
/// always wins and is never refreshed — it's a deliberate injection for
/// tests/CI. Returns `None` only when no token is available at all.
///
/// Request-time consumers (notably `car-inference`) should call this
/// instead of [`access_token`]: it's the difference between a lapsed
/// token producing a 401 burst that poisons 30-day model health and a
/// transparent refresh-and-proceed (#313).
pub async fn access_token_refreshing() -> Option<String> {
    // Env override wins and is never refreshed.
    if let Ok(tok) = std::env::var(PARSLEE_ACCESS_TOKEN_KEY) {
        if !tok.is_empty() {
            return Some(tok);
        }
    }
    let current = car_secrets::resolve_env_or_keychain(PARSLEE_ACCESS_TOKEN_KEY)?;
    // Refresh only when we can *see* the token is (nearly) expired and we
    // have a refresh token. An unknown/missing expiry means "don't churn".
    let expiring = car_secrets::resolve_env_or_keychain(PARSLEE_EXPIRES_AT_KEY)
        .and_then(|s| s.trim().parse::<u64>().ok())
        .map(|exp| epoch_seconds() + REFRESH_SKEW_SECS >= exp)
        .unwrap_or(false);
    if !expiring {
        return Some(current);
    }
    let Some(refresh) = car_secrets::resolve_env_or_keychain(PARSLEE_REFRESH_TOKEN_KEY) else {
        return Some(current);
    };
    let base = api_base(None);
    match refresh_grant(&base, &refresh).await {
        Ok(tokens) => {
            let access = tokens.access_token.clone();
            persist_refreshed(&base, &tokens);
            Some(access)
        }
        // Refresh failed (expired refresh token / offline): fall back to the
        // stored access token and let the server decide. No worse than today
        // — a still-valid access token keeps working.
        Err(_) => Some(current),
    }
}

/// Unconditionally refresh the Parslee bearer, for the **reactive 401**
/// path. [`access_token_refreshing`] only refreshes inside a proactive
/// window keyed on the stored expiry — but a token can be revoked or
/// invalidated server-side *before* its advertised expiry, and a token
/// stored without an expiry never enters that window at all. When a live
/// request is rejected with 401/403, the caller invokes this to mint a
/// fresh bearer and retry once, instead of letting the failure poison
/// 30-day model health (#313).
///
/// Returns the new access token, or `None` when there is no refresh token
/// to use or the refresh itself fails. The `PARSLEE_ACCESS_TOKEN` env
/// override is authoritative and never refreshed (returns `None` so the
/// caller keeps using the injected token).
pub async fn force_refresh() -> Option<String> {
    if let Ok(tok) = std::env::var(PARSLEE_ACCESS_TOKEN_KEY) {
        if !tok.is_empty() {
            return None;
        }
    }
    let refresh = car_secrets::resolve_env_or_keychain(PARSLEE_REFRESH_TOKEN_KEY)?;
    let base = api_base(None);
    match refresh_grant(&base, &refresh).await {
        Ok(tokens) => {
            let access = tokens.access_token.clone();
            persist_refreshed(&base, &tokens);
            Some(access)
        }
        Err(_) => None,
    }
}

/// Resolve the API base: explicit override → stored → default.
pub fn api_base(override_: Option<&str>) -> String {
    override_
        .map(|s| s.trim_end_matches('/').to_string())
        .or_else(|| car_secrets::resolve_env_or_keychain(PARSLEE_API_BASE_KEY))
        .unwrap_or_else(|| DEFAULT_API_BASE.to_string())
}

/// Fetch the Parslee session JSON for the stored token. Returns the
/// raw response body (the caller renders it). `Ok(None)` = not signed in.
pub async fn fetch_status(api_base_override: Option<&str>) -> Result<Option<String>, String> {
    let Some(access) = access_token() else {
        return Ok(None);
    };
    let base = api_base(api_base_override);
    let response = reqwest::Client::new()
        .get(format!("{}/connect/session", base.trim_end_matches('/')))
        .bearer_auth(access)
        .send()
        .await
        .map_err(|e| format!("fetch Parslee session: {e}"))?;
    let status = response.status();
    let text = response
        .text()
        .await
        .map_err(|e| format!("read Parslee session response: {e}"))?;
    if !status.is_success() {
        return Err(format!(
            "Parslee session check failed: HTTP {status}: {text}"
        ));
    }
    Ok(Some(text))
}

// First-login onboarding is intentionally NOT here. Brand-new users
// are routed through Parslee's existing hosted web consent/org page
// during the `/connect/authorize` browser hand-off (see m365dotnet
// `specs/draft/car-inference-gateway-auth.md` B6), so the token CAR
// redeems already carries `active_org`. CAR is a pure OAuth client and
// never touches consent — there is no `ensure_org`, by design.

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn pkce_challenge_is_s256_urlsafe_nopad() {
        let v = pkce_verifier();
        let c = pkce_challenge(&v);
        assert!(!c.contains('=') && !c.contains('+') && !c.contains('/'));
        assert_eq!(c, pkce_challenge(&v)); // deterministic
    }

    #[test]
    fn authorize_url_has_pkce_and_provider() {
        let u = authorize_url(
            "https://api.parslee.ai/",
            "parslee-car",
            "http://localhost:8765/auth/callback",
            "st8",
            "chal",
            Some("microsoft"),
        )
        .unwrap();
        assert!(u.starts_with("https://api.parslee.ai/connect/authorize?"));
        assert!(u.contains("code_challenge=chal"));
        assert!(u.contains("code_challenge_method=S256"));
        assert!(u.contains("client_id=parslee-car"));
        assert!(u.contains("provider=microsoft"));
    }

    #[test]
    fn api_base_precedence() {
        assert_eq!(api_base(Some("https://x.test/")), "https://x.test");
    }

    /// Hand-rolled loopback HTTP mock — no extra prod dep, no feature
    /// flags. Serves exactly `expected` one-shot requests, records
    /// what came in, and replies with whatever `respond` returns.
    /// Lets the networked auth fns be exercised end-to-end in CI
    /// without the real Parslee backend (or the OS keychain — the
    /// token is injected via the `PARSLEE_ACCESS_TOKEN` env override).
    mod mock {
        use std::io::{Read, Write};
        use std::net::TcpListener;
        use std::sync::{Arc, Mutex};
        use std::thread;

        pub struct Recorded {
            pub method: String,
            pub path: String,
            pub authorization: Option<String>,
            pub content_type: Option<String>,
            pub body: String,
        }

        pub struct Mock {
            pub base: String,
            pub recorded: Arc<Mutex<Vec<Recorded>>>,
            handle: Option<thread::JoinHandle<()>>,
        }

        impl Drop for Mock {
            fn drop(&mut self) {
                if let Some(h) = self.handle.take() {
                    let _ = h.join();
                }
            }
        }

        fn find(hay: &[u8], needle: &[u8]) -> Option<usize> {
            hay.windows(needle.len()).position(|w| w == needle)
        }

        pub fn start(
            expected: usize,
            respond: impl Fn(&Recorded) -> (u16, String) + Send + 'static,
        ) -> Mock {
            let listener = TcpListener::bind("127.0.0.1:0").unwrap();
            let port = listener.local_addr().unwrap().port();
            let recorded = Arc::new(Mutex::new(Vec::new()));
            let rec = recorded.clone();
            let handle = thread::spawn(move || {
                for _ in 0..expected {
                    let (mut stream, _) = listener.accept().unwrap();
                    let mut buf = Vec::new();
                    let mut tmp = [0u8; 1024];
                    loop {
                        let n = stream.read(&mut tmp).unwrap();
                        if n == 0 {
                            break;
                        }
                        buf.extend_from_slice(&tmp[..n]);
                        let Some(hdr_end) = find(&buf, b"\r\n\r\n") else {
                            continue;
                        };
                        let headers = String::from_utf8_lossy(&buf[..hdr_end]).into_owned();
                        let content_length = headers
                            .lines()
                            .find_map(|l| {
                                let (k, v) = l.split_once(':')?;
                                if k.eq_ignore_ascii_case("content-length") {
                                    v.trim().parse::<usize>().ok()
                                } else {
                                    None
                                }
                            })
                            .unwrap_or(0);
                        let body_start = hdr_end + 4;
                        while buf.len() < body_start + content_length {
                            let n = stream.read(&mut tmp).unwrap();
                            if n == 0 {
                                break;
                            }
                            buf.extend_from_slice(&tmp[..n]);
                        }
                        let mut header_lines = headers.lines();
                        let req_line = header_lines.next().unwrap_or("");
                        let mut rl = req_line.split_whitespace();
                        let method = rl.next().unwrap_or("").to_string();
                        let path = rl.next().unwrap_or("").to_string();
                        let mut authorization = None;
                        let mut content_type = None;
                        for l in header_lines {
                            if let Some((k, v)) = l.split_once(':') {
                                if k.eq_ignore_ascii_case("authorization") {
                                    authorization = Some(v.trim().to_string());
                                } else if k.eq_ignore_ascii_case("content-type") {
                                    content_type = Some(v.trim().to_string());
                                }
                            }
                        }
                        let body = String::from_utf8_lossy(
                            &buf[body_start..(body_start + content_length).min(buf.len())],
                        )
                        .into_owned();
                        let r = Recorded {
                            method,
                            path,
                            authorization,
                            content_type,
                            body,
                        };
                        let (code, resp_body) = respond(&r);
                        rec.lock().unwrap().push(r);
                        let resp = format!(
                            "HTTP/1.1 {code} OK\r\ncontent-type: application/json\r\n\
                             content-length: {}\r\nconnection: close\r\n\r\n{}",
                            resp_body.len(),
                            resp_body
                        );
                        stream.write_all(resp.as_bytes()).unwrap();
                        let _ = stream.flush();
                        break;
                    }
                }
            });
            Mock {
                base: format!("http://127.0.0.1:{port}"),
                recorded,
                handle: Some(handle),
            }
        }
    }

    #[tokio::test]
    async fn exchange_code_round_trips_token() {
        let mock = mock::start(1, |_r| {
            (
                200,
                r#"{"access_token":"a","refresh_token":"r","expires_in":3600,"token_type":"Bearer"}"#
                    .to_string(),
            )
        });
        let token = exchange_code(
            &mock.base,
            "parslee-car",
            "http://localhost:1/cb",
            "thecode",
            "theverifier",
        )
        .await
        .unwrap();
        assert_eq!(token.access_token, "a");
        assert_eq!(token.refresh_token, "r");
        assert_eq!(token.expires_in, 3600);

        let reqs = mock.recorded.lock().unwrap();
        assert_eq!(reqs.len(), 1);
        assert_eq!(reqs[0].method, "POST");
        assert_eq!(reqs[0].path, "/connect/token");
        assert!(reqs[0].body.contains("grant_type=authorization_code"));
        assert!(reqs[0].body.contains("code=thecode"));
        assert!(reqs[0].body.contains("code_verifier=theverifier"));
    }

    #[tokio::test]
    async fn refresh_grant_round_trips_token() {
        // Gateway reuses the refresh token (omits it from the response) — the
        // `Option` fields must tolerate that.
        let mock = mock::start(1, |_r| {
            (
                200,
                r#"{"access_token":"a2","expires_in":3600,"token_type":"Bearer"}"#.to_string(),
            )
        });
        let tokens = refresh_grant(&mock.base, "the-refresh-token")
            .await
            .unwrap();
        assert_eq!(tokens.access_token, "a2");
        assert_eq!(tokens.refresh_token, None);
        assert_eq!(tokens.expires_in, Some(3600));

        let reqs = mock.recorded.lock().unwrap();
        assert_eq!(reqs.len(), 1);
        assert_eq!(reqs[0].method, "POST");
        assert_eq!(reqs[0].path, "/connect/token");
        assert!(reqs[0].body.contains("grant_type=refresh_token"));
        assert!(reqs[0].body.contains("refresh_token=the-refresh-token"));
        // Public-client refresh: no client_id is sent (matches the daemon).
        assert!(!reqs[0].body.contains("client_id"));
    }

    #[tokio::test]
    async fn fetch_status_sends_bearer() {
        // Inject the token via the env override so the keychain is
        // never touched. No other car-auth test reads this var.
        std::env::set_var(PARSLEE_ACCESS_TOKEN_KEY, "test-token");

        let mock = mock::start(1, |_r| (200, r#"{"authenticated":true}"#.to_string()));

        let session = fetch_status(Some(&mock.base)).await.unwrap();
        assert_eq!(session.as_deref(), Some(r#"{"authenticated":true}"#));

        let reqs = mock.recorded.lock().unwrap();
        assert_eq!(reqs.len(), 1);
        let sess = &reqs[0];
        assert_eq!(sess.method, "GET");
        assert_eq!(sess.path, "/connect/session");
        assert_eq!(sess.authorization.as_deref(), Some("Bearer test-token"));

        std::env::remove_var(PARSLEE_ACCESS_TOKEN_KEY);
    }
}