techne-sdk 0.3.5

//! PDS password sessions (`net`): `com.atproto.server.createSession` and `refreshSession`.
//!
//! The client a service uses to sign in to an account it owns (a service account whose password is
//! [`crate::account::account_password`], or any app password). After the commons' `XrpcSessions`
//! (P20.7 G4). Sign in with the account's DID as the identifier, not its handle, and refuse (in the caller) a session
//! that answers another DID: an account is its DID.

use std::time::Duration;

use serde_json::{json, Value};

/// A signed-in session.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Session {
    /// The DID the PDS signed in as.
    pub did: String,
    pub access_jwt: String,
    /// The refresh JWT, when the PDS issued one.
    pub refresh_jwt: Option<String>,
}

/// The PDS failed, or answered something that is not a session.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
#[error("the PDS failed: {0}")]
pub struct SessionError(pub String);

/// A PDS's session endpoints, over HTTP.
#[derive(Debug, Clone)]
pub struct PasswordSessions {
    pds: String,
    http: reqwest::Client,
}

impl PasswordSessions {
    /// Against `pds` (an origin), with a 30-second timeout per call.
    pub fn new(pds: impl Into<String>) -> Result<Self, reqwest::Error> {
        Ok(Self::with_client(pds, reqwest::Client::builder().timeout(Duration::from_secs(30)).build()?))
    }

    /// Against `pds`, over the caller's client.
    pub fn with_client(pds: impl Into<String>, http: reqwest::Client) -> Self {
        Self { pds: pds.into().trim_end_matches('/').to_owned(), http }
    }

    /// `com.atproto.server.createSession`. `Ok(None)`: the PDS refused the password (401). The session
    /// is returned as the PDS answered it: a caller signing in by DID compares [`Session::did`] with
    /// that DID and refuses a session for any other (the caller owns that error, e.g. a pinned-account
    /// mismatch).
    pub async fn create_session(&self, identifier: &str, password: &str) -> Result<Option<Session>, SessionError> {
        let req = self
            .http
            .post(format!("{}/xrpc/com.atproto.server.createSession", self.pds))
            .json(&json!({"identifier": identifier, "password": password}));
        self.session(req, "createSession", &[]).await
    }

    /// `com.atproto.server.refreshSession` with `refresh_jwt`. `Ok(None)`: the PDS refused the
    /// refresh token (401, or `ExpiredToken` / `InvalidToken`); sign in again.
    pub async fn refresh_session(&self, refresh_jwt: &str) -> Result<Option<Session>, SessionError> {
        let req = self.http.post(format!("{}/xrpc/com.atproto.server.refreshSession", self.pds)).bearer_auth(refresh_jwt);
        self.session(req, "refreshSession", &["ExpiredToken", "InvalidToken"]).await
    }

    async fn session(&self, req: reqwest::RequestBuilder, method: &str, refused: &[&str]) -> Result<Option<Session>, SessionError> {
        let resp = req.send().await.map_err(|e| SessionError(e.without_url().to_string()))?;
        let status = resp.status();
        if status == reqwest::StatusCode::UNAUTHORIZED {
            return Ok(None);
        }
        let body: Value = resp.json().await.unwrap_or(Value::Null);
        if !status.is_success() {
            let error = body["error"].as_str().unwrap_or_default();
            if status == reqwest::StatusCode::BAD_REQUEST && refused.contains(&error) {
                return Ok(None);
            }
            return Err(SessionError(format!("{method} answered {status} {error}")));
        }
        match (body["did"].as_str(), body["accessJwt"].as_str()) {
            (Some(did), Some(jwt)) => Ok(Some(Session {
                did: did.to_owned(),
                access_jwt: jwt.to_owned(),
                refresh_jwt: body["refreshJwt"].as_str().map(str::to_owned),
            })),
            _ => Err(SessionError(format!("{method}: no did or accessJwt"))),
        }
    }
}

#[cfg(test)]
mod tests {
    use axum::http::{HeaderMap, StatusCode};
    use axum::response::{IntoResponse, Response};
    use axum::Json;

    use super::*;

    const DID: &str = "did:plc:sessionsessionsessionses";

    async fn fake_pds() -> String {
        async fn create(Json(body): Json<Value>) -> Response {
            match (body["identifier"].as_str(), body["password"].as_str()) {
                (Some(DID), Some("right")) => Json(json!({"did": DID, "handle": "a.test", "accessJwt": "access-1", "refreshJwt": "refresh-1"})).into_response(),
                (Some(DID), Some("no-refresh")) => Json(json!({"did": DID, "accessJwt": "access-1"})).into_response(),
                (Some(DID), Some("garbled")) => Json(json!({"handle": "a.test"})).into_response(),
                (Some("did:plc:swapped"), _) => Json(json!({"did": DID, "accessJwt": "access-1"})).into_response(),
                (Some("did:plc:down"), _) => (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalServerError"}))).into_response(),
                _ => (StatusCode::UNAUTHORIZED, Json(json!({"error": "AuthenticationRequired"}))).into_response(),
            }
        }
        async fn refresh(headers: HeaderMap) -> Response {
            match headers.get("authorization").and_then(|v| v.to_str().ok()) {
                Some("Bearer refresh-1") => Json(json!({"did": DID, "accessJwt": "access-2", "refreshJwt": "refresh-2"})).into_response(),
                Some("Bearer expired") => (StatusCode::BAD_REQUEST, Json(json!({"error": "ExpiredToken"}))).into_response(),
                Some("Bearer odd") => (StatusCode::BAD_REQUEST, Json(json!({"error": "InvalidRequest"}))).into_response(),
                _ => (StatusCode::UNAUTHORIZED, Json(json!({"error": "AuthenticationRequired"}))).into_response(),
            }
        }
        let app = axum::Router::new()
            .route("/xrpc/com.atproto.server.createSession", axum::routing::post(create))
            .route("/xrpc/com.atproto.server.refreshSession", axum::routing::post(refresh));
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
        format!("http://{addr}/")
    }

    #[tokio::test]
    async fn create_session_signs_in_or_answers_none_on_401() {
        let pds = PasswordSessions::new(fake_pds().await).unwrap();
        let s = pds.create_session(DID, "right").await.unwrap().expect("signed in");
        assert_eq!(s, Session { did: DID.into(), access_jwt: "access-1".into(), refresh_jwt: Some("refresh-1".into()) });
        assert_eq!(pds.create_session(DID, "no-refresh").await.unwrap().unwrap().refresh_jwt, None);
        assert_eq!(pds.create_session(DID, "wrong").await.unwrap(), None, "401 is a refused password");
        // Another DID's session comes back as answered, for the caller to refuse.
        assert_eq!(pds.create_session("did:plc:swapped", "right").await.unwrap().unwrap().did, DID);
        let err = pds.create_session("did:plc:down", "right").await.unwrap_err();
        assert!(err.0.contains("createSession answered 500"), "{err}");
        assert!(pds.create_session(DID, "garbled").await.unwrap_err().0.contains("no did or accessJwt"));
    }

    #[tokio::test]
    async fn refresh_session_rotates_or_answers_none_when_refused() {
        let pds = PasswordSessions::new(fake_pds().await).unwrap();
        let s = pds.refresh_session("refresh-1").await.unwrap().expect("refreshed");
        assert_eq!((s.access_jwt.as_str(), s.refresh_jwt.as_deref()), ("access-2", Some("refresh-2")));
        assert_eq!(pds.refresh_session("expired").await.unwrap(), None);
        assert_eq!(pds.refresh_session("revoked").await.unwrap(), None, "401");
        assert!(pds.refresh_session("odd").await.unwrap_err().0.contains("refreshSession answered 400"));
    }

    #[tokio::test]
    async fn an_unreachable_pds_is_an_error() {
        let pds = PasswordSessions::new("http://127.0.0.1:1").unwrap();
        assert!(pds.create_session(DID, "right").await.is_err());
    }
}