+8 -8

Cargo.toml

··· 24 24 minijinja-embed = {version = "2.7"} 25 25 26 26 [dependencies] 27 27 - atproto-client = { version = "0.11.3" } 28 28 - atproto-identity = { version = "0.11.3", features = ["lru", "zeroize"] } 29 29 - atproto-oauth = { version = "0.11.3", features = ["lru", "zeroize"] } 30 30 - atproto-oauth-aip = { version = "0.11.3" } 31 31 - atproto-oauth-axum = { version = "0.11.3", features = ["zeroize"] } 32 32 - atproto-record = { version = "0.11.3" } 33 33 - atproto-jetstream = { version = "0.11.3" } 34 34 - atproto-xrpcs = { version = "0.11.3" } 27 27 + atproto-client = { version = "0.12.0" } 28 28 + atproto-identity = { version = "0.12.0", features = ["lru", "zeroize"] } 29 29 + atproto-oauth = { version = "0.12.0", features = ["lru", "zeroize"] } 30 30 + atproto-oauth-aip = { version = "0.12.0" } 31 31 + atproto-oauth-axum = { version = "0.12.0", features = ["zeroize"] } 32 32 + atproto-record = { version = "0.12.0" } 33 33 + atproto-jetstream = { version = "0.12.0" } 34 34 + atproto-xrpcs = { version = "0.12.0" } 35 35 36 36 anyhow = "1.0" 37 37 async-trait = "0.1"

+1 -1

Dockerfile

··· 1 1 # syntax=docker/dockerfile:1.4 2 2 - FROM rust:1.89-slim AS builder 2 2 + FROM rust:1.89-slim-bookworm AS builder 3 3 4 4 RUN apt-get update && apt-get install -y \ 5 5 pkg-config \

+5 -1

src/atproto/auth.rs

··· 28 28 let session_response = 29 29 session_exchange(http_client, &protected_resource.resource, access_token).await?; 30 30 31 31 - let dpop_private_key_data = identify_key(&session_response.dpop_key)?; 31 31 + let dpop_key = session_response 32 32 + .dpop_key 33 33 + .ok_or_else(|| anyhow::anyhow!("error-smokesignal-auth-1 DPoP key missing from session response"))?; 34 34 + 35 35 + let dpop_private_key_data = identify_key(&dpop_key)?; 32 36 33 37 Ok(DPoPAuth { 34 38 dpop_private_key_data,

-27

src/bin/smokesignal.rs

··· 29 29 use smokesignal::task_oauth_requests_cleanup::{ 30 30 OAuthRequestsCleanupTask, OAuthRequestsCleanupTaskConfig, 31 31 }; 32 32 - use smokesignal::task_refresh_tokens::{RefreshTokensTask, RefreshTokensTaskConfig}; 33 32 use sqlx::PgPool; 34 33 use std::{collections::HashMap, env, str::FromStr, sync::Arc}; 35 34 use tokio::net::TcpListener; ··· 249 248 } 250 249 251 250 tracker.close(); 252 252 - inner_token.cancel(); 253 253 - }); 254 254 - } 255 255 - 256 256 - // Only spawn refresh tokens task for AT Protocol OAuth backend 257 257 - if let OAuthBackendConfig::ATProtocol { signing_keys } = &config.oauth_backend { 258 258 - let task_config = RefreshTokensTaskConfig { 259 259 - sleep_interval: Duration::seconds(10), 260 260 - worker_id: "dev".to_string(), 261 261 - external_url_base: config.external_base.clone(), 262 262 - signing_keys: signing_keys.clone(), 263 263 - }; 264 264 - let task = RefreshTokensTask::new( 265 265 - task_config, 266 266 - http_client.clone(), 267 267 - pool.clone(), 268 268 - cache_pool.clone(), 269 269 - document_storage.clone(), 270 270 - token.clone(), 271 271 - ); 272 272 - 273 273 - let inner_token = token.clone(); 274 274 - tracker.spawn(async move { 275 275 - if let Err(err) = task.run().await { 276 276 - tracing::error!("Refresh tokens task failed: {}", err); 277 277 - } 278 251 inner_token.cancel(); 279 252 }); 280 253 }

+1 -1

src/http/handle_create_event.rs

··· 286 286 287 287 let create_record_result = create_record( 288 288 &web_context.http_client, 289 289 - &dpop_auth, 289 289 + &atproto_client::client::Auth::DPoP(dpop_auth), 290 290 &current_handle.pds, 291 291 event_record, 292 292 )

+1 -1

src/http/handle_create_rsvp.rs

··· 178 178 179 179 let put_record_result = put_record( 180 180 &web_context.http_client, 181 181 - &dpop_auth, 181 181 + &atproto_client::client::Auth::DPoP(dpop_auth), 182 182 &current_handle.pds, 183 183 rsvp_record, 184 184 )

+1 -1

src/http/handle_delete_event.rs

··· 153 153 154 154 match delete_record( 155 155 &ctx.web_context.http_client, 156 156 - &dpop_auth, 156 156 + &atproto_client::client::Auth::DPoP(dpop_auth), 157 157 &current_handle.pds, 158 158 delete_record_request, 159 159 )

+1 -1

src/http/handle_edit_event.rs

··· 582 582 583 583 let update_record_result = put_record( 584 584 &ctx.web_context.http_client, 585 585 - &dpop_auth, 585 585 + &atproto_client::client::Auth::DPoP(dpop_auth), 586 586 &current_handle.pds, 587 587 update_record_request, 588 588 )

+2 -2

src/http/import_utils.rs

··· 34 34 35 35 let results = list_records::<CommunityEvent>( 36 36 http_client, 37 37 - None, 37 37 + &atproto_client::client::Auth::None, 38 38 repository_endpoint, 39 39 did.to_string(), 40 40 COMMUNITY_EVENT_NSID.to_string(), ··· 111 111 112 112 let results = list_records::<CommunityRsvp>( 113 113 http_client, 114 114 - None, 114 114 + &atproto_client::client::Auth::None, 115 115 repository_endpoint, 116 116 did.to_string(), 117 117 COMMUNITY_RSVP_NSID.to_string(),

-1

src/lib.rs

··· 13 13 pub mod storage; 14 14 pub mod task_identity_refresh; 15 15 pub mod task_oauth_requests_cleanup; 16 16 - pub mod task_refresh_tokens; 17 16 pub mod task_search_indexer; 18 17 pub mod task_search_indexer_errors; 19 18 pub mod task_webhooks;

-244

src/task_refresh_tokens.rs

··· 1 1 - use anyhow::Result; 2 2 - use atproto_identity::key::identify_key; 3 3 - use atproto_oauth::workflow::{OAuthClient, oauth_refresh}; 4 4 - use chrono::{Duration, Utc}; 5 5 - use deadpool_redis::redis::{AsyncCommands, pipe}; 6 6 - use std::borrow::Cow; 7 7 - use tokio::time::{Instant, sleep}; 8 8 - use tokio_util::sync::CancellationToken; 9 9 - 10 10 - use crate::{ 11 11 - config::SigningKeys, 12 12 - refresh_tokens_errors::RefreshError, 13 13 - storage::{ 14 14 - CachePool, StoragePool, 15 15 - cache::{OAUTH_REFRESH_HEARTBEATS, OAUTH_REFRESH_QUEUE, build_worker_queue}, 16 16 - oauth::{oauth_session_delete, oauth_session_update, web_session_lookup}, 17 17 - }, 18 18 - }; 19 19 - 20 20 - pub struct RefreshTokensTaskConfig { 21 21 - pub sleep_interval: Duration, 22 22 - pub worker_id: String, 23 23 - pub external_url_base: String, 24 24 - pub signing_keys: SigningKeys, 25 25 - } 26 26 - 27 27 - pub struct RefreshTokensTask { 28 28 - pub config: RefreshTokensTaskConfig, 29 29 - pub http_client: reqwest::Client, 30 30 - pub storage_pool: StoragePool, 31 31 - pub cache_pool: CachePool, 32 32 - pub document_storage: std::sync::Arc<dyn atproto_identity::storage::DidDocumentStorage>, 33 33 - pub cancellation_token: CancellationToken, 34 34 - } 35 35 - 36 36 - impl RefreshTokensTask { 37 37 - #[must_use] 38 38 - pub fn new( 39 39 - config: RefreshTokensTaskConfig, 40 40 - http_client: reqwest::Client, 41 41 - storage_pool: StoragePool, 42 42 - cache_pool: CachePool, 43 43 - document_storage: std::sync::Arc<dyn atproto_identity::storage::DidDocumentStorage>, 44 44 - cancellation_token: CancellationToken, 45 45 - ) -> Self { 46 46 - Self { 47 47 - config, 48 48 - http_client, 49 49 - storage_pool, 50 50 - cache_pool, 51 51 - document_storage, 52 52 - cancellation_token, 53 53 - } 54 54 - } 55 55 - 56 56 - /// Runs the refresh tokens task as a long-running process 57 57 - /// 58 58 - /// # Errors 59 59 - /// Returns an error if the sleep interval cannot be converted, or if there's a problem 60 60 - /// processing the work items 61 61 - pub async fn run(&self) -> Result<()> { 62 62 - tracing::debug!("RefreshTokensTask started"); 63 63 - 64 64 - let interval = self.config.sleep_interval.to_std()?; 65 65 - 66 66 - let sleeper = sleep(interval); 67 67 - tokio::pin!(sleeper); 68 68 - 69 69 - loop { 70 70 - tokio::select! { 71 71 - () = self.cancellation_token.cancelled() => { 72 72 - break; 73 73 - }, 74 74 - () = &mut sleeper => { 75 75 - if let Err(err) = self.process_work().await { 76 76 - tracing::error!("RefreshTokensTask failed: {}", err); 77 77 - } 78 78 - sleeper.as_mut().reset(Instant::now() + interval); 79 79 - } 80 80 - } 81 81 - } 82 82 - 83 83 - tracing::info!("RefreshTokensTask stopped"); 84 84 - 85 85 - Ok(()) 86 86 - } 87 87 - 88 88 - async fn process_work(&self) -> Result<i32> { 89 89 - let worker_queue = build_worker_queue(&self.config.worker_id); 90 90 - 91 91 - let mut conn = self.cache_pool.get().await?; 92 92 - 93 93 - let now = chrono::Utc::now(); 94 94 - let epoch_millis = now.timestamp_millis(); 95 95 - 96 96 - let _: () = conn 97 97 - .hset( 98 98 - OAUTH_REFRESH_HEARTBEATS, 99 99 - &self.config.worker_id, 100 100 - now.to_string(), 101 101 - ) 102 102 - .await?; 103 103 - 104 104 - let global_queue_count: i32 = conn 105 105 - .zcount(OAUTH_REFRESH_QUEUE, 0, epoch_millis + 1) 106 106 - .await?; 107 107 - let worker_queue_count: i32 = conn.zcount(&worker_queue, 0, epoch_millis + 1).await?; 108 108 - 109 109 - tracing::trace!( 110 110 - global_queue_count = global_queue_count, 111 111 - worker_queue_count = worker_queue_count, 112 112 - "queue counts" 113 113 - ); 114 114 - 115 115 - let mut process_work = worker_queue_count > 0; 116 116 - 117 117 - if global_queue_count > 0 && worker_queue_count == 0 { 118 118 - let (moved, new_count): (i64, i64) = pipe() 119 119 - .atomic() 120 120 - // Take some work from the global queue and put it in the worker queue 121 121 - // ZRANGESTORE dst src min max [BYSCORE | BYLEX] [REV] [LIMIT offset count] 122 122 - .cmd("ZRANGESTORE") 123 123 - .arg(&worker_queue) 124 124 - .arg(OAUTH_REFRESH_QUEUE) 125 125 - .arg(0) 126 126 - .arg(epoch_millis) 127 127 - .arg("BYSCORE") 128 128 - .arg("LIMIT") 129 129 - .arg(0) 130 130 - .arg(5) 131 131 - // Update the global queue to remove the items that were moved 132 132 - .cmd("ZDIFFSTORE") 133 133 - .arg(OAUTH_REFRESH_QUEUE) 134 134 - .arg(2) 135 135 - .arg(OAUTH_REFRESH_QUEUE) 136 136 - .arg(&worker_queue) 137 137 - .query_async(&mut conn) 138 138 - .await?; 139 139 - process_work = true; 140 140 - 141 141 - tracing::debug!( 142 142 - moved = moved, 143 143 - new_count = new_count, 144 144 - "moved work from global queue to worker queue" 145 145 - ); 146 146 - } 147 147 - 148 148 - if !process_work { 149 149 - return Ok(0); 150 150 - } 151 151 - 152 152 - let count = 0; 153 153 - let results: Vec<(String, i64)> = conn 154 154 - .zrangebyscore_limit_withscores(&worker_queue, 0, epoch_millis, 0, 5) 155 155 - .await?; 156 156 - 157 157 - for (session_group, deadline) in results { 158 158 - let _: () = conn.zrem(&worker_queue, &session_group).await?; 159 159 - 160 160 - if let Err(err) = self 161 161 - .refresh_oauth_session(&mut conn, &session_group, deadline) 162 162 - .await 163 163 - { 164 164 - tracing::error!(session_group, deadline, err = ?err, "failed to refresh oauth session: {}", err); 165 165 - 166 166 - if let Err(err) = oauth_session_delete(&self.storage_pool, &session_group).await { 167 167 - tracing::error!(session_group, err = ?err, "failed to delete oauth session: {}", err); 168 168 - } 169 169 - } 170 170 - } 171 171 - 172 172 - Ok(count) 173 173 - } 174 174 - 175 175 - async fn refresh_oauth_session( 176 176 - &self, 177 177 - conn: &mut deadpool_redis::Connection, 178 178 - session_group: &str, 179 179 - _deadline: i64, 180 180 - ) -> Result<()> { 181 181 - let (handle, oauth_session) = 182 182 - web_session_lookup(&self.storage_pool, session_group, None).await?; 183 183 - 184 184 - let secret_signing_key_string = self 185 185 - .config 186 186 - .signing_keys 187 187 - .as_ref() 188 188 - .get(&oauth_session.secret_jwk_id) 189 189 - .cloned() 190 190 - .ok_or_else(|| anyhow::Error::from(RefreshError::SecretSigningKeyNotFound))?; 191 191 - 192 192 - let private_signing_key_data = identify_key(&secret_signing_key_string)?; 193 193 - 194 194 - let private_dpop_key_data = identify_key(&oauth_session.dpop_jwk)?; 195 195 - 196 196 - let document = match self 197 197 - .document_storage 198 198 - .get_document_by_did(&handle.did) 199 199 - .await? 200 200 - { 201 201 - Some(doc) => doc, 202 202 - None => return Err(RefreshError::IdentityDocumentNotFound.into()), 203 203 - }; 204 204 - 205 205 - let oauth_client = OAuthClient { 206 206 - redirect_uri: format!("https://{}/oauth/callback", self.config.external_url_base), 207 207 - client_id: format!( 208 208 - "https://{}/oauth/client-metadata.json", 209 209 - self.config.external_url_base 210 210 - ), 211 211 - private_signing_key_data, 212 212 - }; 213 213 - 214 214 - let token_response = oauth_refresh( 215 215 - &self.http_client, 216 216 - &oauth_client, 217 217 - &private_dpop_key_data, 218 218 - &oauth_session.refresh_token, 219 219 - &document, 220 220 - ) 221 221 - .await?; 222 222 - 223 223 - let now = Utc::now(); 224 224 - 225 225 - oauth_session_update( 226 226 - &self.storage_pool, 227 227 - Cow::Borrowed(session_group), 228 228 - Cow::Borrowed(&token_response.access_token), 229 229 - Cow::Borrowed(&token_response.refresh_token), 230 230 - now + chrono::Duration::seconds(i64::from(token_response.expires_in)), 231 231 - ) 232 232 - .await?; 233 233 - 234 234 - let modified_expires_at = ((f64::from(token_response.expires_in)) * 0.8).round() as i64; 235 235 - let refresh_at = (now + chrono::Duration::seconds(modified_expires_at)).timestamp_millis(); 236 236 - 237 237 - let _: () = conn 238 238 - .zadd(OAUTH_REFRESH_QUEUE, session_group, refresh_at) 239 239 - .await 240 240 - .map_err(RefreshError::PlaceInRefreshQueueFailed)?; 241 241 - 242 242 - Ok(()) 243 243 - } 244 244 - }