feat(create_report): PdsXrpcClient trait + Real/Fake impls

+439 -1

2 changed files

expand all

src

commands

test

labeler

create_report.rs

tests

common

mod.rs

+292

src/commands/test/labeler/create_report.rs

··· 114 114 ) -> Result<RawCreateReportResponse, CreateReportStageError>; 115 115 } 116 116 117 + /// Raw HTTP response from XRPC calls to the PDS. 118 + /// 119 + /// Similar to `RawCreateReportResponse` but used for PDS-specific calls 120 + /// (createSession, getServiceAuth) where the response needs to be parsed 121 + /// as JSON by the caller. 122 + #[derive(Debug)] 123 + pub struct RawPdsXrpcResponse { 124 + /// HTTP status code. 125 + pub status: StatusCode, 126 + /// Raw response body bytes. 127 + pub raw_body: Arc<[u8]>, 128 + /// Content-Type header value, if present. Lowercased for matching. 129 + pub content_type: Option<String>, 130 + /// The URL that was requested (for diagnostics). 131 + pub source_url: String, 132 + } 133 + 134 + /// Narrow seam for POSTing/GETting against the user's PDS. 135 + /// 136 + /// The existing `HttpClient` in `src/common/identity.rs` is GET-only and 137 + /// does not support bearer headers or request bodies. This trait exists 138 + /// to keep those capabilities out of the identity-resolution seam. 139 + #[async_trait] 140 + pub trait PdsXrpcClient: Send + Sync { 141 + /// POST `body` (JSON-serialized) to the PDS endpoint at the given path 142 + /// (e.g., `"xrpc/com.atproto.server.createSession"`). Optional bearer 143 + /// and `atproto-proxy` headers. 144 + async fn post( 145 + &self, 146 + path: &str, 147 + bearer: Option<&str>, 148 + atproto_proxy: Option<&str>, 149 + body: &serde_json::Value, 150 + ) -> Result<RawPdsXrpcResponse, CreateReportStageError>; 151 + 152 + /// GET the PDS endpoint at the given path with optional bearer and 153 + /// URL-encoded query pairs. 154 + async fn get( 155 + &self, 156 + path: &str, 157 + bearer: Option<&str>, 158 + query: &[(&str, &str)], 159 + ) -> Result<RawPdsXrpcResponse, CreateReportStageError>; 160 + } 161 + 162 + /// Real `PdsXrpcClient` implementation using reqwest. 163 + pub struct RealPdsXrpcClient { 164 + client: reqwest::Client, 165 + base: Url, 166 + } 167 + 168 + impl RealPdsXrpcClient { 169 + /// Create a new `RealPdsXrpcClient` using the given shared reqwest 170 + /// client and PDS base URL. 171 + pub fn new(client: reqwest::Client, base: Url) -> Self { 172 + Self { client, base } 173 + } 174 + } 175 + 176 + #[async_trait] 177 + impl PdsXrpcClient for RealPdsXrpcClient { 178 + async fn post( 179 + &self, 180 + path: &str, 181 + bearer: Option<&str>, 182 + atproto_proxy: Option<&str>, 183 + body: &serde_json::Value, 184 + ) -> Result<RawPdsXrpcResponse, CreateReportStageError> { 185 + let mut url = self.base.clone(); 186 + url.set_path(path); 187 + let source_url = url.to_string(); 188 + let mut req = self 189 + .client 190 + .post(url.as_str()) 191 + .header("Content-Type", "application/json") 192 + .body(serde_json::to_vec(body).expect("serde_json::Value always serializes")); 193 + if let Some(b) = bearer { 194 + req = req.header("Authorization", format!("Bearer {b}")); 195 + } 196 + if let Some(p) = atproto_proxy { 197 + req = req.header("atproto-proxy", p); 198 + } 199 + let resp = req 200 + .send() 201 + .await 202 + .map_err(|e| CreateReportStageError::Transport { 203 + message: e.to_string(), 204 + source: Some(Box::new(e)), 205 + })?; 206 + let status = resp.status(); 207 + let content_type = resp 208 + .headers() 209 + .get(reqwest::header::CONTENT_TYPE) 210 + .and_then(|h| h.to_str().ok()) 211 + .map(|s| s.to_ascii_lowercase()); 212 + let body = resp 213 + .bytes() 214 + .await 215 + .map_err(|e| CreateReportStageError::Transport { 216 + message: e.to_string(), 217 + source: Some(Box::new(e)), 218 + })?; 219 + Ok(RawPdsXrpcResponse { 220 + status, 221 + raw_body: Arc::from(body.as_ref()), 222 + content_type, 223 + source_url, 224 + }) 225 + } 226 + 227 + async fn get( 228 + &self, 229 + path: &str, 230 + bearer: Option<&str>, 231 + query: &[(&str, &str)], 232 + ) -> Result<RawPdsXrpcResponse, CreateReportStageError> { 233 + let mut url = self.base.clone(); 234 + url.set_path(path); 235 + { 236 + let mut pairs = url.query_pairs_mut(); 237 + for (k, v) in query { 238 + pairs.append_pair(k, v); 239 + } 240 + } 241 + let source_url = url.to_string(); 242 + let mut req = self.client.get(url.as_str()); 243 + if let Some(b) = bearer { 244 + req = req.header("Authorization", format!("Bearer {b}")); 245 + } 246 + let resp = req 247 + .send() 248 + .await 249 + .map_err(|e| CreateReportStageError::Transport { 250 + message: e.to_string(), 251 + source: Some(Box::new(e)), 252 + })?; 253 + let status = resp.status(); 254 + let content_type = resp 255 + .headers() 256 + .get(reqwest::header::CONTENT_TYPE) 257 + .and_then(|h| h.to_str().ok()) 258 + .map(|s| s.to_ascii_lowercase()); 259 + let body = resp 260 + .bytes() 261 + .await 262 + .map_err(|e| CreateReportStageError::Transport { 263 + message: e.to_string(), 264 + source: Some(Box::new(e)), 265 + })?; 266 + Ok(RawPdsXrpcResponse { 267 + status, 268 + raw_body: Arc::from(body.as_ref()), 269 + content_type, 270 + source_url, 271 + }) 272 + } 273 + } 274 + 117 275 /// Real `CreateReportTee` implementation using reqwest. 118 276 pub struct RealCreateReportTee { 119 277 client: reqwest::Client, ··· 196 354 raw_body: Arc::from(body_bytes.as_ref()), 197 355 source_url, 198 356 }) 357 + } 358 + } 359 + 360 + /// Error type for `PdsJwtFetcher` operations. 361 + /// 362 + /// Carries a human-readable message; every PDS-side failure is treated as 363 + /// a `NetworkError` by the report stage (per AC5.3 / AC6.3). 364 + #[derive(Debug)] 365 + pub enum PdsJwtFetchError { 366 + /// Any failure at the PDS boundary — unreachable, auth rejected, etc. 367 + Pds { message: String }, 368 + } 369 + 370 + impl std::fmt::Display for PdsJwtFetchError { 371 + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { 372 + match self { 373 + PdsJwtFetchError::Pds { message } => write!(f, "{message}"), 374 + } 375 + } 376 + } 377 + 378 + /// Fetches a service-auth JWT from a PDS by first creating a session and 379 + /// then calling `getServiceAuth`. Used in mode-2 (`pds_service_auth_accepted`). 380 + pub struct PdsJwtFetcher<'a> { 381 + client: &'a dyn PdsXrpcClient, 382 + } 383 + 384 + impl<'a> PdsJwtFetcher<'a> { 385 + /// Create a new `PdsJwtFetcher` using the given PDS client. 386 + pub fn new(client: &'a dyn PdsXrpcClient) -> Self { 387 + Self { client } 388 + } 389 + 390 + /// Run `createSession` then `getServiceAuth`, returning the minted 391 + /// service-auth JWT. 392 + pub async fn fetch( 393 + &self, 394 + handle: &str, 395 + app_password: &str, 396 + aud: &str, 397 + lxm: &str, 398 + exp_absolute_unix: i64, 399 + ) -> Result<String, PdsJwtFetchError> { 400 + // 1. createSession. 401 + let body = serde_json::json!({ 402 + "identifier": handle, 403 + "password": app_password, 404 + }); 405 + let resp = self 406 + .client 407 + .post("xrpc/com.atproto.server.createSession", None, None, &body) 408 + .await 409 + .map_err(|e| PdsJwtFetchError::Pds { 410 + message: format!("createSession transport: {e}"), 411 + })?; 412 + if !resp.status.is_success() { 413 + return Err(PdsJwtFetchError::Pds { 414 + message: format!("createSession returned status {}", resp.status), 415 + }); 416 + } 417 + let session: serde_json::Value = 418 + serde_json::from_slice(&resp.raw_body).map_err(|e| PdsJwtFetchError::Pds { 419 + message: format!("createSession body not JSON: {e}"), 420 + })?; 421 + let access_jwt = session["accessJwt"] 422 + .as_str() 423 + .ok_or_else(|| PdsJwtFetchError::Pds { 424 + message: "createSession response missing accessJwt".to_string(), 425 + })? 426 + .to_string(); 427 + 428 + // 2. getServiceAuth (GET with query params). 429 + let exp_s = exp_absolute_unix.to_string(); 430 + let resp = self 431 + .client 432 + .get( 433 + "xrpc/com.atproto.server.getServiceAuth", 434 + Some(&access_jwt), 435 + &[("aud", aud), ("lxm", lxm), ("exp", &exp_s)], 436 + ) 437 + .await 438 + .map_err(|e| PdsJwtFetchError::Pds { 439 + message: format!("getServiceAuth transport: {e}"), 440 + })?; 441 + if !resp.status.is_success() { 442 + return Err(PdsJwtFetchError::Pds { 443 + message: format!("getServiceAuth returned status {}", resp.status), 444 + }); 445 + } 446 + let auth: serde_json::Value = 447 + serde_json::from_slice(&resp.raw_body).map_err(|e| PdsJwtFetchError::Pds { 448 + message: format!("getServiceAuth body not JSON: {e}"), 449 + })?; 450 + let token = auth["token"] 451 + .as_str() 452 + .ok_or_else(|| PdsJwtFetchError::Pds { 453 + message: "getServiceAuth response missing token".to_string(), 454 + })? 455 + .to_string(); 456 + 457 + Ok(token) 458 + } 459 + } 460 + 461 + /// Posts `com.atproto.moderation.createReport` to the PDS (not the 462 + /// labeler) with the `atproto-proxy` header, letting the PDS mint and 463 + /// forward the JWT itself. 464 + pub struct PdsProxiedPoster<'a> { 465 + client: &'a dyn PdsXrpcClient, 466 + } 467 + 468 + impl<'a> PdsProxiedPoster<'a> { 469 + /// Create a new `PdsProxiedPoster` using the given PDS client. 470 + pub fn new(client: &'a dyn PdsXrpcClient) -> Self { 471 + Self { client } 472 + } 473 + 474 + /// Post the createReport body through the PDS with the given user 475 + /// access JWT. Returns the `RawPdsXrpcResponse` so the caller can 476 + /// classify success / labeler-side rejection / PDS-side rejection. 477 + pub async fn post( 478 + &self, 479 + labeler_did: &str, 480 + access_jwt: &str, 481 + body: &serde_json::Value, 482 + ) -> Result<RawPdsXrpcResponse, CreateReportStageError> { 483 + self.client 484 + .post( 485 + "xrpc/com.atproto.moderation.createReport", 486 + Some(access_jwt), 487 + Some(&format!("{labeler_did}#atproto_labeler")), 488 + body, 489 + ) 490 + .await 199 491 } 200 492 } 201 493

+147 -1

tests/common/mod.rs

··· 9 9 10 10 use async_trait::async_trait; 11 11 use atproto_devtool::commands::test::labeler::create_report::{ 12 - CreateReportStageError, CreateReportTee, RawCreateReportResponse, 12 + CreateReportStageError, CreateReportTee, PdsXrpcClient, RawCreateReportResponse, 13 + RawPdsXrpcResponse, 13 14 }; 14 15 use atproto_devtool::commands::test::labeler::http::{HttpStageError, RawHttpTee, RawXrpcResponse}; 15 16 use atproto_devtool::commands::test::labeler::subscription::{ ··· 268 269 }) 269 270 } 270 271 } 272 + } 273 + } 274 + 275 + /// Scripted response for a single `FakePdsXrpcClient` call. A `Transport` 276 + /// variant short-circuits with an error; a `Response` variant returns a 277 + /// `RawPdsXrpcResponse` built from the supplied parts. 278 + #[derive(Debug, Clone)] 279 + pub enum FakePdsXrpcResponse { 280 + /// Simulate a transport-level failure (no HTTP exchange took place). 281 + Transport { message: String }, 282 + /// Simulate a well-formed HTTP response. 283 + Response { status: u16, body: Vec<u8> }, 284 + } 285 + 286 + /// A recorded request observed by `FakePdsXrpcClient`. 287 + #[derive(Debug, Clone)] 288 + pub struct RecordedPdsRequest { 289 + /// HTTP method: "POST" or "GET". 290 + pub method: &'static str, 291 + /// Request path (e.g., "xrpc/com.atproto.server.createSession"). 292 + pub path: String, 293 + /// Bearer token, if any (stripped of "Bearer " prefix). 294 + pub bearer: Option<String>, 295 + /// atproto-proxy header, if any. 296 + pub atproto_proxy: Option<String>, 297 + /// JSON body for POST requests; None for GET. 298 + pub body: Option<serde_json::Value>, 299 + /// Query parameters for GET requests; empty for POST. 300 + pub query: Vec<(String, String)>, 301 + } 302 + 303 + /// Fake `PdsXrpcClient` for integration tests. 304 + /// 305 + /// Scripted per-call-index responses: first call gets `responses[0]`, 306 + /// second gets `responses[1]`, etc. Panics if a call is made with no 307 + /// script queued — tests must declare every call the stage is expected to make. 308 + pub struct FakePdsXrpcClient { 309 + /// Queued responses. 310 + scripts: Arc<Mutex<Vec<FakePdsXrpcResponse>>>, 311 + /// Every request observed (in order). 312 + recorded: Arc<Mutex<Vec<RecordedPdsRequest>>>, 313 + } 314 + 315 + impl FakePdsXrpcClient { 316 + /// Create a fake with no scripted responses. 317 + pub fn new() -> Self { 318 + Self { 319 + scripts: Arc::new(Mutex::new(Vec::new())), 320 + recorded: Arc::new(Mutex::new(Vec::new())), 321 + } 322 + } 323 + 324 + /// Queue a scripted response for the next call. 325 + pub fn enqueue(&self, response: FakePdsXrpcResponse) { 326 + self.scripts.lock().unwrap().push(response); 327 + } 328 + 329 + /// Return the recorded request history (cloned). 330 + pub fn recorded_requests(&self) -> Vec<RecordedPdsRequest> { 331 + self.recorded.lock().unwrap().clone() 332 + } 333 + 334 + /// Get the last recorded request, panicking if none. 335 + pub fn last_request(&self) -> RecordedPdsRequest { 336 + self.recorded 337 + .lock() 338 + .unwrap() 339 + .last() 340 + .cloned() 341 + .expect("FakePdsXrpcClient: no requests recorded yet") 342 + } 343 + 344 + /// Pop the next script and return the result. Panics if no script queued. 345 + fn dispatch_next(&self) -> Result<RawPdsXrpcResponse, CreateReportStageError> { 346 + let mut scripts = self.scripts.lock().unwrap(); 347 + if scripts.is_empty() { 348 + panic!( 349 + "FakePdsXrpcClient: call made with no script queued. \ 350 + Each test must enqueue() exactly the responses it expects." 351 + ); 352 + } 353 + let script = scripts.remove(0); 354 + 355 + match script { 356 + FakePdsXrpcResponse::Transport { message } => Err(CreateReportStageError::Transport { 357 + message, 358 + source: None, 359 + }), 360 + FakePdsXrpcResponse::Response { status, body } => { 361 + let raw_body: Arc<[u8]> = Arc::from(body.as_slice()); 362 + Ok(RawPdsXrpcResponse { 363 + status: StatusCode::from_u16(status).expect("test must use valid HTTP status"), 364 + raw_body, 365 + content_type: Some("application/json".to_string()), 366 + source_url: "https://pds.test/xrpc".to_string(), 367 + }) 368 + } 369 + } 370 + } 371 + } 372 + 373 + impl Default for FakePdsXrpcClient { 374 + fn default() -> Self { 375 + Self::new() 376 + } 377 + } 378 + 379 + #[async_trait] 380 + impl PdsXrpcClient for FakePdsXrpcClient { 381 + async fn post( 382 + &self, 383 + path: &str, 384 + bearer: Option<&str>, 385 + atproto_proxy: Option<&str>, 386 + body: &serde_json::Value, 387 + ) -> Result<RawPdsXrpcResponse, CreateReportStageError> { 388 + self.recorded.lock().unwrap().push(RecordedPdsRequest { 389 + method: "POST", 390 + path: path.to_string(), 391 + bearer: bearer.map(String::from), 392 + atproto_proxy: atproto_proxy.map(String::from), 393 + body: Some(body.clone()), 394 + query: Vec::new(), 395 + }); 396 + self.dispatch_next() 397 + } 398 + 399 + async fn get( 400 + &self, 401 + path: &str, 402 + bearer: Option<&str>, 403 + query: &[(&str, &str)], 404 + ) -> Result<RawPdsXrpcResponse, CreateReportStageError> { 405 + self.recorded.lock().unwrap().push(RecordedPdsRequest { 406 + method: "GET", 407 + path: path.to_string(), 408 + bearer: bearer.map(String::from), 409 + atproto_proxy: None, 410 + body: None, 411 + query: query 412 + .iter() 413 + .map(|(k, v)| (k.to_string(), v.to_string())) 414 + .collect(), 415 + }); 416 + self.dispatch_next() 271 417 } 272 418 } 273 419

Configure Feed

Configure Feed