diff --git a/.gitignore b/.gitignore index 23c0a89..3f482b3 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,6 @@ /target -src_old .sqlx .env +reference-pds/ diff --git a/Cargo.lock b/Cargo.lock index a5564a6..1b80585 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -450,6 +450,7 @@ version = "0.1.0" dependencies = [ "anyhow", "axum", + "base64 0.22.1", "bcrypt", "bytes", "chrono", @@ -459,7 +460,9 @@ dependencies = [ "jacquard-axum", "jacquard-repo", "jsonwebtoken", + "k256", "multihash", + "rand 0.8.5", "reqwest", "serde", "serde_ipld_dagcbor", @@ -973,7 +976,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8d162beedaa69905488a8da94f5ac3edb4dd4788b732fadb7bd120b2625c1976" dependencies = [ "data-encoding", - "syn 2.0.111", + "syn 1.0.109", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index aad6195..af1e018 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,6 +6,7 @@ edition = "2024" [dependencies] anyhow = "1.0.100" axum = "0.8.7" +base64 = "0.22.1" bcrypt = "0.17.1" bytes = "1.11.0" chrono = { version = "0.4.42", features = ["serde"] } @@ -15,7 +16,9 @@ jacquard = "0.9.3" jacquard-axum = "0.9.2" jacquard-repo = "0.9.2" jsonwebtoken = { version = "10.2.0", features = ["rust_crypto"] } +k256 = { version = "0.13.3", features = ["ecdsa", "pem", "pkcs8"] } multihash = "0.19.3" +rand = "0.8.5" reqwest = { version = "0.12.24", features = ["json"] } serde = { version = "1.0.228", features = ["derive"] } serde_ipld_dagcbor = "0.6.4" diff --git a/TODO.md b/TODO.md index 430aa5f..7291a8c 100644 --- a/TODO.md +++ b/TODO.md @@ -2,7 +2,7 @@ Lewis' corrected big boy todofile -## 1. Server Infrastructure & Proxying +## Server Infrastructure & Proxying - [x] Health Check - [x] Implement `GET /health` endpoint (returns "OK"). - [x] Server Description @@ -11,8 +11,9 @@ Lewis' corrected big boy todofile - [x] Implement strict forwarding for all `app.bsky.*` and `chat.bsky.*` requests to an appview. - [x] Forward Auth headers correctly. - [x] Handle AppView errors/timeouts gracefully. + - [ ] Implement Read-After-Write (RAW) consistency (Local Overlay) for proxied requests (merge local unindexed records). -## 2. Authentication & Account Management (`com.atproto.server`) +## Authentication & Account Management (`com.atproto.server`) - [x] Account Creation - [x] Implement `com.atproto.server.createAccount`. - [x] Validate handle format (reject invalid characters). @@ -25,8 +26,24 @@ Lewis' corrected big boy todofile - [x] Implement `com.atproto.server.getSession`. - [x] Implement `com.atproto.server.refreshSession`. - [x] Implement `com.atproto.server.deleteSession` (Logout). + - [ ] Implement `com.atproto.server.activateAccount`. + - [ ] Implement `com.atproto.server.checkAccountStatus`. + - [ ] Implement `com.atproto.server.confirmEmail`. + - [ ] Implement `com.atproto.server.createAppPassword`. + - [ ] Implement `com.atproto.server.createInviteCode`. + - [ ] Implement `com.atproto.server.createInviteCodes`. + - [ ] Implement `com.atproto.server.deactivateAccount` / `deleteAccount`. + - [ ] Implement `com.atproto.server.getAccountInviteCodes`. + - [ ] Implement `com.atproto.server.getServiceAuth` (Cross-service auth). + - [ ] Implement `com.atproto.server.listAppPasswords`. + - [ ] Implement `com.atproto.server.requestAccountDelete`. + - [ ] Implement `com.atproto.server.requestEmailConfirmation` / `requestEmailUpdate`. + - [ ] Implement `com.atproto.server.requestPasswordReset` / `resetPassword`. + - [ ] Implement `com.atproto.server.reserveSigningKey`. + - [ ] Implement `com.atproto.server.revokeAppPassword`. + - [ ] Implement `com.atproto.server.updateEmail`. -## 3. Repository Operations (`com.atproto.repo`) +## Repository Operations (`com.atproto.repo`) - [ ] Record CRUD - [ ] Implement `com.atproto.repo.createRecord`. - [ ] Validate schema against Lexicon (just structure, not complex logic). @@ -38,12 +55,15 @@ Lewis' corrected big boy todofile - [ ] Implement `com.atproto.repo.deleteRecord`. - [ ] Implement `com.atproto.repo.listRecords`. - [ ] Implement `com.atproto.repo.describeRepo`. + - [ ] Implement `com.atproto.repo.applyWrites` (Batch writes). + - [ ] Implement `com.atproto.repo.importRepo` (Migration). + - [ ] Implement `com.atproto.repo.listMissingBlobs`. - [ ] Blob Management - [ ] Implement `com.atproto.repo.uploadBlob`. - [ ] Store blob (S3). - [ ] return `blob` ref (CID + MimeType). -## 4. Sync & Federation (`com.atproto.sync`) +## Sync & Federation (`com.atproto.sync`) - [ ] The Firehose (WebSocket) - [ ] Implement `com.atproto.sync.subscribeRepos`. - [ ] Broadcast real-time commit events. @@ -53,18 +73,41 @@ Lewis' corrected big boy todofile - [ ] Implement `com.atproto.sync.getBlocks` (Return specific blocks via CIDs). - [ ] Implement `com.atproto.sync.getLatestCommit`. - [ ] Implement `com.atproto.sync.getRecord` (Sync version, distinct from repo.getRecord). + - [ ] Implement `com.atproto.sync.getRepoStatus`. + - [ ] Implement `com.atproto.sync.listRepos`. + - [ ] Implement `com.atproto.sync.notifyOfUpdate`. - [ ] Blob Sync - [ ] Implement `com.atproto.sync.getBlob`. - [ ] Implement `com.atproto.sync.listBlobs`. - [ ] Crawler Interaction - [ ] Implement `com.atproto.sync.requestCrawl` (Notify relays to index us). -## 5. Identity (`com.atproto.identity`) +## Identity (`com.atproto.identity`) - [ ] Resolution - [ ] Implement `com.atproto.identity.resolveHandle` (Can be internal or proxy to PLC). + - [ ] Implement `com.atproto.identity.updateHandle`. + - [ ] Implement `com.atproto.identity.submitPlcOperation` / `signPlcOperation` / `requestPlcOperationSignature`. + - [ ] Implement `com.atproto.identity.getRecommendedDidCredentials`. - [ ] Implement `/.well-known/did.json` (Depends on supporting did:web). -## 6. Record Schema Validation +## Admin Management (`com.atproto.admin`) +- [ ] Implement `com.atproto.admin.deleteAccount`. +- [ ] Implement `com.atproto.admin.disableAccountInvites`. +- [ ] Implement `com.atproto.admin.disableInviteCodes`. +- [ ] Implement `com.atproto.admin.enableAccountInvites`. +- [ ] Implement `com.atproto.admin.getAccountInfo` / `getAccountInfos`. +- [ ] Implement `com.atproto.admin.getInviteCodes`. +- [ ] Implement `com.atproto.admin.getSubjectStatus`. +- [ ] Implement `com.atproto.admin.sendEmail`. +- [ ] Implement `com.atproto.admin.updateAccountEmail`. +- [ ] Implement `com.atproto.admin.updateAccountHandle`. +- [ ] Implement `com.atproto.admin.updateAccountPassword`. +- [ ] Implement `com.atproto.admin.updateSubjectStatus`. + +## Moderation (`com.atproto.moderation`) +- [ ] Implement `com.atproto.moderation.createReport`. + +## Record Schema Validation - [ ] `app.bsky.feed.post` - [ ] `app.bsky.feed.like` - [ ] `app.bsky.feed.repost` @@ -73,9 +116,32 @@ Lewis' corrected big boy todofile - [ ] `app.bsky.actor.profile` - [ ] Other app(view) validation too!!! -## 7. General Requirements +## Infrastructure & Core Components +- [ ] Sequencer (Event Log) + - [ ] Implement a `Sequencer` (backed by `repo_seq` table? Like in ref impl). + - [ ] Implement event formatting (`commit`, `handle`, `identity`, `account`). + - [ ] Implement database polling / event emission mechanism. + - [ ] Implement cursor-based event replay (`requestSeqRange`). +- [ ] Repo Storage & Consistency (in postgres) + - [ ] Implement `RepoStorage` for postgres (replaces per-user SQLite). + - [ ] Read/Write IPLD blocks to `blocks` table (global deduplication). + - [ ] Manage Repo Root in `repos` table. + - [ ] Implement Atomic Repo Transactions. + - [ ] Ensure `blocks` write, `repo_root` update, `records` index update, and `sequencer` event are committed in a single transaction. + - [ ] Implement concurrency control (row-level locking on `repos` table) to prevent concurrent writes to the same repo. +- [ ] DID Cache + - [ ] Implement caching layer for DID resolution (Redis or in-memory). + - [ ] Handle cache invalidation/expiry. +- [ ] Background Jobs + - [ ] Implement background queue for async tasks (crawler notifications, discord/telegram 2FA sending instead of email). + - [ ] Implement `Crawlers` service (debounce notifications to relays). +- [ ] Mailer equivalent + - [ ] Implement code/notification sending service as a replacement for the mailer because there's no way I'm starting with email. :D +- [ ] Image Processing + - [ ] Implement image resize/formatting pipeline (for blob uploads). - [ ] IPLD & MST - - [ ] Implement Merkle Search Tree (MST) logic for repo signing. - - [ ] Implement CAR (Content Addressable Archives) encoding/decoding. + - [ ] Implement Merkle Search Tree logic for repo signing. + - [ ] Implement CAR (Content Addressable Archive) encoding/decoding. - [ ] Validation - [ ] DID PLC Operations (Sign rotation keys). + diff --git a/justfile b/justfile index 43276c3..4696d48 100644 --- a/justfile +++ b/justfile @@ -14,6 +14,7 @@ test-lifecycle: test-others: cargo test --lib cargo test --test actor + cargo test --test auth cargo test --test feed cargo test --test graph cargo test --test identity diff --git a/ref_pds_downloader.sh b/ref_pds_downloader.sh new file mode 100755 index 0000000..4e2964e --- /dev/null +++ b/ref_pds_downloader.sh @@ -0,0 +1,11 @@ +git clone --depth 1 --filter=blob:none --sparse https://github.com/bluesky-social/atproto.git reference-pds + +cd reference-pds + +git sparse-checkout set packages/pds + +git checkout main + +mv packages/pds/* . +mv packages/pds/.[!.]* . 2>/dev/null +rm -rf .git diff --git a/src/api/proxy.rs b/src/api/proxy.rs index 01b8ccb..e26937c 100644 --- a/src/api/proxy.rs +++ b/src/api/proxy.rs @@ -1,5 +1,5 @@ use axum::{ - extract::{Path, Query}, + extract::{Path, Query, State}, http::{HeaderMap, Method, StatusCode}, response::{IntoResponse, Response}, body::Bytes, @@ -7,8 +7,11 @@ use axum::{ use reqwest::Client; use tracing::{info, error}; use std::collections::HashMap; +use crate::state::AppState; +use sqlx::Row; pub async fn proxy_handler( + State(state): State, Path(method): Path, method_verb: Method, headers: HeaderMap, @@ -20,8 +23,8 @@ pub async fn proxy_handler( .and_then(|h| h.to_str().ok()) .map(|s| s.to_string()); - let appview_url = match proxy_header { - Some(url) => url, + let appview_url = match &proxy_header { + Some(url) => url.clone(), None => match std::env::var("APPVIEW_URL") { Ok(url) => url, Err(_) => return (StatusCode::BAD_GATEWAY, "No upstream AppView configured").into_response(), @@ -38,8 +41,37 @@ pub async fn proxy_handler( .request(method_verb, &target_url) .query(¶ms); + let mut auth_header_val = headers.get("Authorization").map(|h| h.clone()); + + if let Some(aud) = &proxy_header { + if let Some(auth_val) = &auth_header_val { + if let Ok(token) = auth_val.to_str() { + let token = token.replace("Bearer ", ""); + if let Ok(did) = crate::auth::get_did_from_token(&token) { + let key_row = sqlx::query("SELECT k.key_bytes FROM user_keys k JOIN users u ON k.user_id = u.id WHERE u.did = $1") + .bind(&did) + .fetch_optional(&state.db) + .await; + + if let Ok(Some(row)) = key_row { + let key_bytes: Vec = row.get("key_bytes"); + if let Ok(new_token) = crate::auth::create_service_token(&did, aud, &method, &key_bytes) { + if let Ok(val) = axum::http::HeaderValue::from_str(&format!("Bearer {}", new_token)) { + auth_header_val = Some(val); + } + } + } + } + } + } + } + + if let Some(val) = auth_header_val { + request_builder = request_builder.header("Authorization", val); + } + for (key, value) in headers.iter() { - if key != "host" && key != "content-length" { + if key != "host" && key != "content-length" && key != "authorization" { request_builder = request_builder.header(key, value); } } diff --git a/src/api/repo.rs b/src/api/repo.rs index 0242f03..2f6c715 100644 --- a/src/api/repo.rs +++ b/src/api/repo.rs @@ -46,21 +46,23 @@ pub async fn create_record( } let token = auth_header.unwrap().to_str().unwrap_or("").replace("Bearer ", ""); - if let Err(_) = crate::auth::verify_token(&token) { - return (StatusCode::UNAUTHORIZED, Json(json!({"error": "AuthenticationFailed", "message": "Invalid token"}))).into_response(); - } - - let session = sqlx::query("SELECT did FROM sessions WHERE access_jwt = $1") + let session = sqlx::query( + "SELECT s.did, k.key_bytes FROM sessions s JOIN users u ON s.did = u.did JOIN user_keys k ON u.id = k.user_id WHERE s.access_jwt = $1" + ) .bind(&token) .fetch_optional(&state.db) .await .unwrap_or(None); - let did = match session { - Some(row) => row.get::("did"), + let (did, key_bytes) = match session { + Some(row) => (row.get::("did"), row.get::, _>("key_bytes")), None => return (StatusCode::UNAUTHORIZED, Json(json!({"error": "AuthenticationFailed"}))).into_response(), }; + if let Err(_) = crate::auth::verify_token(&token, &key_bytes) { + return (StatusCode::UNAUTHORIZED, Json(json!({"error": "AuthenticationFailed", "message": "Invalid token signature"}))).into_response(); + } + if input.repo != did { return (StatusCode::FORBIDDEN, Json(json!({"error": "InvalidRepo", "message": "Repo does not match authenticated user"}))).into_response(); } diff --git a/src/api/server.rs b/src/api/server.rs index b9d8a00..aef6e10 100644 --- a/src/api/server.rs +++ b/src/api/server.rs @@ -13,6 +13,8 @@ use tracing::{info, error, warn}; use jacquard_repo::{mst::Mst, commit::Commit, storage::BlockStore}; use jacquard::types::{string::Tid, did::Did, integer::LimitedU32}; use std::sync::Arc; +use k256::SecretKey; +use rand::rngs::OsRng; pub async fn describe_server() -> impl IntoResponse { let domains_str = std::env::var("AVAILABLE_USER_DOMAINS").unwrap_or_else(|_| "example.com".to_string()); @@ -139,6 +141,20 @@ pub async fn create_account( } }; + let secret_key = SecretKey::random(&mut OsRng); + let secret_key_bytes = secret_key.to_bytes(); + + let key_insert = sqlx::query("INSERT INTO user_keys (user_id, key_bytes) VALUES ($1, $2)") + .bind(user_id) + .bind(&secret_key_bytes[..]) + .execute(&mut *tx) + .await; + + if let Err(e) = key_insert { + error!("Error inserting user key: {:?}", e); + return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response(); + } + let store = Arc::new(state.block_store.clone()); let mst = Mst::new(store.clone()); let mst_root = match mst.root().await { @@ -203,7 +219,7 @@ pub async fn create_account( } } - let access_jwt = crate::auth::create_access_token(&did).map_err(|e| { + let access_jwt = crate::auth::create_access_token(&did, &secret_key_bytes[..]).map_err(|e| { error!("Error creating access token: {:?}", e); (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response() }); @@ -212,7 +228,7 @@ pub async fn create_account( Err(r) => return r, }; - let refresh_jwt = crate::auth::create_refresh_token(&did).map_err(|e| { + let refresh_jwt = crate::auth::create_refresh_token(&did, &secret_key_bytes[..]).map_err(|e| { error!("Error creating refresh token: {:?}", e); (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response() }); @@ -267,7 +283,7 @@ pub async fn create_session( ) -> Response { info!("create_session: identifier='{}'", input.identifier); - let user_row = sqlx::query("SELECT did, handle, password_hash FROM users WHERE handle = $1 OR email = $1") + let user_row = sqlx::query("SELECT u.did, u.handle, u.password_hash, k.key_bytes FROM users u JOIN user_keys k ON u.id = k.user_id WHERE u.handle = $1 OR u.email = $1") .bind(&input.identifier) .fetch_optional(&state.db) .await; @@ -279,8 +295,9 @@ pub async fn create_session( if verify(&input.password, &stored_hash).unwrap_or(false) { let did: String = row.get("did"); let handle: String = row.get("handle"); + let key_bytes: Vec = row.get("key_bytes"); - let access_jwt = match crate::auth::create_access_token(&did) { + let access_jwt = match crate::auth::create_access_token(&did, &key_bytes) { Ok(t) => t, Err(e) => { error!("Failed to create access token: {:?}", e); @@ -288,7 +305,7 @@ pub async fn create_session( } }; - let refresh_jwt = match crate::auth::create_refresh_token(&did) { + let refresh_jwt = match crate::auth::create_refresh_token(&did, &key_bytes) { Ok(t) => t, Err(e) => { error!("Failed to create refresh token: {:?}", e); @@ -344,19 +361,16 @@ pub async fn get_session( let token = auth_header.unwrap().to_str().unwrap_or("").replace("Bearer ", ""); - if let Err(_) = crate::auth::verify_token(&token) { - return (StatusCode::UNAUTHORIZED, Json(json!({"error": "AuthenticationFailed", "message": "Invalid token"}))).into_response(); - } - let result = sqlx::query( r#" - SELECT u.handle, u.did, u.email + SELECT u.handle, u.did, u.email, k.key_bytes FROM sessions s JOIN users u ON s.did = u.did + JOIN user_keys k ON u.id = k.user_id WHERE s.access_jwt = $1 "# ) - .bind(token) + .bind(&token) .fetch_optional(&state.db) .await; @@ -365,6 +379,11 @@ pub async fn get_session( let handle: String = row.get("handle"); let did: String = row.get("did"); let email: String = row.get("email"); + let key_bytes: Vec = row.get("key_bytes"); + + if let Err(_) = crate::auth::verify_token(&token, &key_bytes) { + return (StatusCode::UNAUTHORIZED, Json(json!({"error": "AuthenticationFailed", "message": "Invalid token signature"}))).into_response(); + } return (StatusCode::OK, Json(json!({ "handle": handle, @@ -424,11 +443,9 @@ pub async fn refresh_session( let refresh_token = auth_header.unwrap().to_str().unwrap_or("").replace("Bearer ", ""); - if let Err(_) = crate::auth::verify_token(&refresh_token) { - return (StatusCode::UNAUTHORIZED, Json(json!({"error": "AuthenticationFailed", "message": "Invalid refresh token"}))).into_response(); - } - - let session = sqlx::query("SELECT did FROM sessions WHERE refresh_jwt = $1") + let session = sqlx::query( + "SELECT s.did, k.key_bytes FROM sessions s JOIN users u ON s.did = u.did JOIN user_keys k ON u.id = k.user_id WHERE s.refresh_jwt = $1" + ) .bind(&refresh_token) .fetch_optional(&state.db) .await; @@ -436,14 +453,20 @@ pub async fn refresh_session( match session { Ok(Some(session_row)) => { let did: String = session_row.get("did"); - let new_access_jwt = match crate::auth::create_access_token(&did) { + let key_bytes: Vec = session_row.get("key_bytes"); + + if let Err(_) = crate::auth::verify_token(&refresh_token, &key_bytes) { + return (StatusCode::UNAUTHORIZED, Json(json!({"error": "AuthenticationFailed", "message": "Invalid refresh token signature"}))).into_response(); + } + + let new_access_jwt = match crate::auth::create_access_token(&did, &key_bytes) { Ok(t) => t, Err(e) => { error!("Failed to create access token: {:?}", e); return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response(); } }; - let new_refresh_jwt = match crate::auth::create_refresh_token(&did) { + let new_refresh_jwt = match crate::auth::create_refresh_token(&did, &key_bytes) { Ok(t) => t, Err(e) => { error!("Failed to create refresh token: {:?}", e); diff --git a/src/auth.rs b/src/auth.rs index b6442c8..be9f8e4 100644 --- a/src/auth.rs +++ b/src/auth.rs @@ -1,59 +1,157 @@ -use jsonwebtoken::{encode, decode, Header, Validation, EncodingKey, DecodingKey, TokenData}; use serde::{Deserialize, Serialize}; use chrono::{Utc, Duration}; -use std::env; +use k256::ecdsa::{SigningKey, VerifyingKey, signature::Signer, signature::Verifier, Signature}; +use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; +use anyhow::{Context, Result, anyhow}; #[derive(Debug, Serialize, Deserialize)] pub struct Claims { - // DID type shit + pub iss: String, pub sub: String, + pub aud: String, pub exp: usize, pub iat: usize, - pub scope: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub scope: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub lxm: Option, pub jti: String, } -pub fn create_access_token(did: &str) -> Result { - let secret = env::var("JWT_SECRET").unwrap_or_else(|_| "secret".to_string()); +#[derive(Debug, Serialize, Deserialize)] +struct Header { + alg: String, + typ: String, +} + +#[derive(Debug, Serialize, Deserialize)] +struct UnsafeClaims { + iss: String, + sub: Option, +} + +// fancy boy TokenData equivalent for compatibility/structure +pub struct TokenData { + pub claims: T, +} + +pub fn get_did_from_token(token: &str) -> Result { + let parts: Vec<&str> = token.split('.').collect(); + if parts.len() != 3 { + return Err("Invalid token format".to_string()); + } + + let payload_bytes = URL_SAFE_NO_PAD.decode(parts[1]) + .map_err(|e| format!("Base64 decode failed: {}", e))?; + + let claims: UnsafeClaims = serde_json::from_slice(&payload_bytes) + .map_err(|e| format!("JSON decode failed: {}", e))?; + + Ok(claims.sub.unwrap_or(claims.iss)) +} + +pub fn create_access_token(did: &str, key_bytes: &[u8]) -> Result { + create_signed_token(did, "access", key_bytes, Duration::minutes(15)) +} + +pub fn create_refresh_token(did: &str, key_bytes: &[u8]) -> Result { + create_signed_token(did, "refresh", key_bytes, Duration::days(7)) +} + +pub fn create_service_token(did: &str, aud: &str, lxm: &str, key_bytes: &[u8]) -> Result { + let signing_key = SigningKey::from_slice(key_bytes)?; + let expiration = Utc::now() - .checked_add_signed(Duration::minutes(15)) + .checked_add_signed(Duration::seconds(60)) .expect("valid timestamp") .timestamp(); let claims = Claims { + iss: did.to_owned(), sub: did.to_owned(), + aud: aud.to_owned(), exp: expiration as usize, iat: Utc::now().timestamp() as usize, - scope: "access".to_string(), + scope: None, + lxm: Some(lxm.to_string()), jti: uuid::Uuid::new_v4().to_string(), }; - encode(&Header::default(), &claims, &EncodingKey::from_secret(secret.as_ref())) + sign_claims(claims, &signing_key) } -pub fn create_refresh_token(did: &str) -> Result { - let secret = env::var("JWT_SECRET").unwrap_or_else(|_| "secret".to_string()); +fn create_signed_token(did: &str, scope: &str, key_bytes: &[u8], duration: Duration) -> Result { + let signing_key = SigningKey::from_slice(key_bytes)?; + let expiration = Utc::now() - .checked_add_signed(Duration::days(7)) + .checked_add_signed(duration) .expect("valid timestamp") .timestamp(); let claims = Claims { + iss: did.to_owned(), sub: did.to_owned(), + aud: format!("did:web:{}", std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string())), exp: expiration as usize, iat: Utc::now().timestamp() as usize, - scope: "refresh".to_string(), + scope: Some(scope.to_string()), + lxm: None, jti: uuid::Uuid::new_v4().to_string(), }; - encode(&Header::default(), &claims, &EncodingKey::from_secret(secret.as_ref())) + sign_claims(claims, &signing_key) } -pub fn verify_token(token: &str) -> Result, jsonwebtoken::errors::Error> { - let secret = env::var("JWT_SECRET").unwrap_or_else(|_| "secret".to_string()); - decode::( - token, - &DecodingKey::from_secret(secret.as_ref()), - &Validation::default(), - ) +fn sign_claims(claims: Claims, key: &SigningKey) -> Result { + let header = Header { + alg: "ES256K".to_string(), + typ: "JWT".to_string(), + }; + + let header_json = serde_json::to_string(&header)?; + let claims_json = serde_json::to_string(&claims)?; + + let header_b64 = URL_SAFE_NO_PAD.encode(header_json); + let claims_b64 = URL_SAFE_NO_PAD.encode(claims_json); + + let message = format!("{}.{}", header_b64, claims_b64); + let signature: Signature = key.sign(message.as_bytes()); + let signature_b64 = URL_SAFE_NO_PAD.encode(signature.to_bytes()); + + Ok(format!("{}.{}", message, signature_b64)) +} + +pub fn verify_token(token: &str, key_bytes: &[u8]) -> Result, anyhow::Error> { + let parts: Vec<&str> = token.split('.').collect(); + if parts.len() != 3 { + return Err(anyhow!("Invalid token format")); + } + + let header_b64 = parts[0]; + let claims_b64 = parts[1]; + let signature_b64 = parts[2]; + + let signature_bytes = URL_SAFE_NO_PAD.decode(signature_b64) + .context("Base64 decode of signature failed")?; + let signature = Signature::from_slice(&signature_bytes) + .map_err(|e| anyhow!("Invalid signature format: {}", e))?; + + let signing_key = SigningKey::from_slice(key_bytes)?; + let verifying_key = VerifyingKey::from(&signing_key); + + let message = format!("{}.{}", header_b64, claims_b64); + verifying_key.verify(message.as_bytes(), &signature) + .map_err(|e| anyhow!("Signature verification failed: {}", e))?; + + let claims_bytes = URL_SAFE_NO_PAD.decode(claims_b64) + .context("Base64 decode of claims failed")?; + let claims: Claims = serde_json::from_slice(&claims_bytes) + .context("JSON decode of claims failed")?; + + let now = Utc::now().timestamp() as usize; + if claims.exp < now { + return Err(anyhow!("Token expired")); + } + + Ok(TokenData { claims }) } diff --git a/tests/auth.rs b/tests/auth.rs new file mode 100644 index 0000000..250436c --- /dev/null +++ b/tests/auth.rs @@ -0,0 +1,122 @@ +use bspds::auth; +use k256::SecretKey; +use rand::rngs::OsRng; +use chrono::{Utc, Duration}; +use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; +use serde_json::json; +use k256::ecdsa::{SigningKey, signature::Signer}; + +#[test] +fn test_jwt_flow() { + let secret_key = SecretKey::random(&mut OsRng); + let key_bytes = secret_key.to_bytes(); + let did = "did:plc:test"; + + let token = auth::create_access_token(did, &key_bytes).expect("create token"); + let data = auth::verify_token(&token, &key_bytes).expect("verify token"); + assert_eq!(data.claims.sub, did); + assert_eq!(data.claims.iss, did); + assert_eq!(data.claims.scope, Some("access".to_string())); + + let r_token = auth::create_refresh_token(did, &key_bytes).expect("create refresh token"); + let r_data = auth::verify_token(&r_token, &key_bytes).expect("verify refresh token"); + assert_eq!(r_data.claims.scope, Some("refresh".to_string())); + + let aud = "did:web:service"; + let lxm = "com.example.test"; + let s_token = auth::create_service_token(did, aud, lxm, &key_bytes).expect("create service token"); + let s_data = auth::verify_token(&s_token, &key_bytes).expect("verify service token"); + assert_eq!(s_data.claims.aud, aud); + assert_eq!(s_data.claims.lxm, Some(lxm.to_string())); +} + +#[test] +fn test_verify_fails_with_wrong_key() { + let secret_key1 = SecretKey::random(&mut OsRng); + let key_bytes1 = secret_key1.to_bytes(); + + let secret_key2 = SecretKey::random(&mut OsRng); + let key_bytes2 = secret_key2.to_bytes(); + + let did = "did:plc:test"; + let token = auth::create_access_token(did, &key_bytes1).expect("create token"); + + let result = auth::verify_token(&token, &key_bytes2); + assert!(result.is_err()); +} + +#[test] +fn test_token_expiration() { + let secret_key = SecretKey::random(&mut OsRng); + let key_bytes = secret_key.to_bytes(); + let signing_key = SigningKey::from_slice(&key_bytes).expect("key"); + + let header = json!({ + "alg": "ES256K", + "typ": "JWT" + }); + let claims = json!({ + "iss": "did:plc:test", + "sub": "did:plc:test", + "aud": "did:web:test", + "exp": (Utc::now() - Duration::seconds(10)).timestamp(), + "iat": (Utc::now() - Duration::minutes(1)).timestamp(), + "jti": "unique", + }); + + let header_b64 = URL_SAFE_NO_PAD.encode(serde_json::to_string(&header).unwrap()); + let claims_b64 = URL_SAFE_NO_PAD.encode(serde_json::to_string(&claims).unwrap()); + let message = format!("{}.{}", header_b64, claims_b64); + let signature: k256::ecdsa::Signature = signing_key.sign(message.as_bytes()); + let signature_b64 = URL_SAFE_NO_PAD.encode(signature.to_bytes()); + let token = format!("{}.{}", message, signature_b64); + + let result = auth::verify_token(&token, &key_bytes); + match result { + Ok(_) => panic!("Token should be expired"), + Err(e) => assert_eq!(e.to_string(), "Token expired"), + } +} + +#[test] +fn test_invalid_token_format() { + let secret_key = SecretKey::random(&mut OsRng); + let key_bytes = secret_key.to_bytes(); + + assert!(auth::verify_token("invalid.token", &key_bytes).is_err()); + assert!(auth::verify_token("too.many.parts.here", &key_bytes).is_err()); + assert!(auth::verify_token("bad_base64.payload.sig", &key_bytes).is_err()); +} + +#[test] +fn test_tampered_token() { + let secret_key = SecretKey::random(&mut OsRng); + let key_bytes = secret_key.to_bytes(); + let did = "did:plc:test"; + + let token = auth::create_access_token(did, &key_bytes).expect("create token"); + let parts: Vec<&str> = token.split('.').collect(); + + let claims_json = String::from_utf8(URL_SAFE_NO_PAD.decode(parts[1]).unwrap()).unwrap(); + let mut claims: serde_json::Value = serde_json::from_str(&claims_json).unwrap(); + claims["sub"] = json!("did:plc:hacker"); + let tampered_claims_b64 = URL_SAFE_NO_PAD.encode(serde_json::to_string(&claims).unwrap()); + + let tampered_token = format!("{}.{}.{}", parts[0], tampered_claims_b64, parts[2]); + + let result = auth::verify_token(&tampered_token, &key_bytes); + assert!(result.is_err()); +} + +#[test] +fn test_get_did_from_token() { + let secret_key = SecretKey::random(&mut OsRng); + let key_bytes = secret_key.to_bytes(); + let did = "did:plc:test"; + + let token = auth::create_access_token(did, &key_bytes).expect("create token"); + let extracted_did = auth::get_did_from_token(&token).expect("get did"); + assert_eq!(extracted_did, did); + + assert!(auth::get_did_from_token("bad.token").is_err()); +} diff --git a/tests/common/mod.rs b/tests/common/mod.rs index df171bc..b1e4709 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -24,6 +24,7 @@ pub const AUTH_DID: &str = "did:plc:fake"; #[allow(dead_code)] pub const TARGET_DID: &str = "did:plc:target"; +#[allow(dead_code)] pub fn client() -> Client { Client::new() } @@ -142,6 +143,7 @@ pub async fn create_test_post( (uri, cid, rkey) } +#[allow(dead_code)] pub async fn create_account_and_login(client: &Client) -> (String, String) { let handle = format!("user_{}", uuid::Uuid::new_v4()); let payload = json!({ diff --git a/tests/proxy.rs b/tests/proxy.rs index 554c45f..8b66051 100644 --- a/tests/proxy.rs +++ b/tests/proxy.rs @@ -9,6 +9,7 @@ use axum::{ use tokio::net::TcpListener; use reqwest::Client; use std::sync::Arc; +use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; async fn spawn_mock_upstream() -> (String, tokio::sync::mpsc::Receiver<(String, String, Option)>) { let (tx, rx) = tokio::sync::mpsc::channel(10); @@ -94,3 +95,39 @@ async fn test_proxy_missing_config() { assert_eq!(res.status(), StatusCode::BAD_GATEWAY); } + +#[tokio::test] +async fn test_proxy_auth_signing() { + let app_url = common::base_url().await; + let (upstream_url, mut rx) = spawn_mock_upstream().await; + let client = Client::new(); + + let (access_jwt, did) = common::create_account_and_login(&client).await; + + let res = client.get(format!("{}/xrpc/com.example.signed", app_url)) + .header("atproto-proxy", &upstream_url) + .header("Authorization", format!("Bearer {}", access_jwt)) + .send() + .await + .unwrap(); + + assert_eq!(res.status(), StatusCode::OK); + + let (method, uri, auth) = rx.recv().await.expect("Upstream receive"); + assert_eq!(method, "GET"); + assert_eq!(uri, "/xrpc/com.example.signed"); + + let received_token = auth.expect("No auth header").replace("Bearer ", ""); + assert_ne!(received_token, access_jwt, "Token should be replaced"); + + let parts: Vec<&str> = received_token.split('.').collect(); + assert_eq!(parts.len(), 3); + + let payload_bytes = URL_SAFE_NO_PAD.decode(parts[1]).expect("payload b64"); + let claims: serde_json::Value = serde_json::from_slice(&payload_bytes).expect("payload json"); + + assert_eq!(claims["iss"], did); + assert_eq!(claims["sub"], did); + assert_eq!(claims["aud"], upstream_url); + assert_eq!(claims["lxm"], "com.example.signed"); +}