Per-user JWT stuffs

.gitignore

@@ -1,6 +1,6 @@
 /target
-src_old
 .sqlx
 
 .env
 
+reference-pds/

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]]

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"

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).
+

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

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

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<AppState>,
     Path(method): Path<String>,
     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(&params);
 
+    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<u8> = 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);
         }
     }

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::<String, _>("did"),
+    let (did, key_bytes) = match session {
+        Some(row) => (row.get::<String, _>("did"), row.get::<Vec<u8>, _>("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();
     }

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<u8> = 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<u8> = 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<u8> = 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);

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<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub lxm: Option<String>,
     pub jti: String,
 }
 
-pub fn create_access_token(did: &str) -> Result<String, jsonwebtoken::errors::Error> {
-    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<String>,
+}
+
+// fancy boy TokenData equivalent for compatibility/structure
+pub struct TokenData<T> {
+    pub claims: T,
+}
+
+pub fn get_did_from_token(token: &str) -> Result<String, String> {
+    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<String, anyhow::Error> {
+    create_signed_token(did, "access", key_bytes, Duration::minutes(15))
+}
+
+pub fn create_refresh_token(did: &str, key_bytes: &[u8]) -> Result<String, anyhow::Error> {
+    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<String, anyhow::Error> {
+    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<String, jsonwebtoken::errors::Error> {
-    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<String, anyhow::Error> {
+    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<TokenData<Claims>, jsonwebtoken::errors::Error> {
-    let secret = env::var("JWT_SECRET").unwrap_or_else(|_| "secret".to_string());
-    decode::<Claims>(
-        token,
-        &DecodingKey::from_secret(secret.as_ref()),
-        &Validation::default(),
-    )
+fn sign_claims(claims: Claims, key: &SigningKey) -> Result<String, anyhow::Error> {
+    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<TokenData<Claims>, 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 })
 }

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());
+}

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!({

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<String>)>) {
     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");
+}