From a839c7d7b3234231a9ba888d4885a6f9c8dcaf54 Mon Sep 17 00:00:00 2001 From: lewis Date: Mon, 29 Dec 2025 21:30:23 +0200 Subject: [PATCH] Moderation conf. vs ref --- .env.example | 9 ++ src/api/identity/did.rs | 2 - src/api/moderation/mod.rs | 199 ++++++++++++++++++++++++++++++- src/api/server/service_auth.rs | 1 + src/auth/mod.rs | 17 ++- tests/moderation.rs | 212 +++++++++++++++++++++++++++++++++ 6 files changed, 431 insertions(+), 9 deletions(-) diff --git a/.env.example b/.env.example index 05651a7..5b0126b 100644 --- a/.env.example +++ b/.env.example @@ -119,6 +119,15 @@ AWS_SECRET_ACCESS_KEY=minioadmin # How often to check for scheduled account deletions (default: 3600 = 1 hour) # SCHEDULED_DELETE_CHECK_INTERVAL_SECS=3600 # ============================================================================= +# Moderation / Report Service +# ============================================================================= +# If configured, moderation reports will be proxied to this service +# instead of being stored locally. The service should implement the +# com.atproto.moderation.createReport endpoint (e.g., Bluesky's Ozone). +# Both URL and DID must be set for proxying to be enabled. +# REPORT_SERVICE_URL=https://mod.bsky.app +# REPORT_SERVICE_DID=did:plc:ar7c4by46qjdydhdevvrndac +# ============================================================================= # Miscellaneous # ============================================================================= # Allow HTTP for proxy requests (development only) diff --git a/src/api/identity/did.rs b/src/api/identity/did.rs index 7d12505..7cfde31 100644 --- a/src/api/identity/did.rs +++ b/src/api/identity/did.rs @@ -447,9 +447,7 @@ pub struct VerificationMethods { } #[derive(serde::Serialize)] -#[serde(rename_all = "camelCase")] pub struct Services { - #[serde(rename = "atproto_pds")] pub atproto_pds: AtprotoPds, } diff --git a/src/api/moderation/mod.rs b/src/api/moderation/mod.rs index 793abb3..170c016 100644 --- a/src/api/moderation/mod.rs +++ b/src/api/moderation/mod.rs @@ -1,4 +1,5 @@ use crate::api::ApiError; +use crate::api::proxy_client::{is_ssrf_safe, proxy_client}; use crate::state::AppState; use axum::{ Json, @@ -8,7 +9,7 @@ use axum::{ }; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; -use tracing::error; +use tracing::{error, info}; #[derive(Deserialize)] #[serde(rename_all = "camelCase")] @@ -29,6 +30,15 @@ pub struct CreateReportOutput { pub created_at: String, } +fn get_report_service_config() -> Option<(String, String)> { + let url = std::env::var("REPORT_SERVICE_URL").ok()?; + let did = std::env::var("REPORT_SERVICE_DID").ok()?; + if url.is_empty() || did.is_empty() { + return None; + } + Some((url, did)) +} + pub async fn create_report( State(state): State, headers: axum::http::HeaderMap, @@ -40,10 +50,177 @@ pub async fn create_report( Some(t) => t, None => return ApiError::AuthenticationRequired.into_response(), }; - let did = match crate::auth::validate_bearer_token(&state.db, &token).await { - Ok(user) => user.did, + + let auth_user = match crate::auth::validate_bearer_token_allow_takendown(&state.db, &token).await + { + Ok(user) => user, Err(e) => return ApiError::from(e).into_response(), }; + + let did = &auth_user.did; + + if let Some((service_url, service_did)) = get_report_service_config() { + return proxy_to_report_service( + &state, + &auth_user, + &service_url, + &service_did, + &input, + ) + .await; + } + + create_report_locally(&state, did, auth_user.is_takendown, input).await +} + +async fn proxy_to_report_service( + state: &AppState, + auth_user: &crate::auth::AuthenticatedUser, + service_url: &str, + service_did: &str, + input: &CreateReportInput, +) -> Response { + if let Err(e) = is_ssrf_safe(service_url) { + error!("Report service URL failed SSRF check: {:?}", e); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": "InternalError", "message": "Invalid report service configuration"})), + ) + .into_response(); + } + + let key_bytes = match &auth_user.key_bytes { + Some(kb) => kb.clone(), + None => { + match sqlx::query_as::<_, (Vec, Option)>( + "SELECT k.key_bytes, k.encryption_version + FROM users u + JOIN user_keys k ON u.id = k.user_id + WHERE u.did = $1", + ) + .bind(&auth_user.did) + .fetch_optional(&state.db) + .await + { + Ok(Some((key_bytes_enc, encryption_version))) => { + match crate::config::decrypt_key(&key_bytes_enc, encryption_version) { + Ok(key) => key, + Err(e) => { + error!(error = ?e, "Failed to decrypt user key for report service auth"); + return ApiError::AuthenticationFailedMsg( + "Failed to get signing key".into(), + ) + .into_response(); + } + } + } + Ok(None) => { + return ApiError::AuthenticationFailedMsg("User has no signing key".into()) + .into_response(); + } + Err(e) => { + error!(error = ?e, "DB error fetching user key for report"); + return ApiError::AuthenticationFailedMsg("Failed to get signing key".into()) + .into_response(); + } + } + } + }; + + let service_token = match crate::auth::create_service_token( + &auth_user.did, + service_did, + "com.atproto.moderation.createReport", + &key_bytes, + ) { + Ok(t) => t, + Err(e) => { + error!("Failed to create service token for report: {:?}", e); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": "InternalError"})), + ) + .into_response(); + } + }; + + let target_url = format!("{}/xrpc/com.atproto.moderation.createReport", service_url); + info!( + did = %auth_user.did, + service_did = %service_did, + "Proxying createReport to report service" + ); + + let request_body = json!({ + "reasonType": input.reason_type, + "reason": input.reason, + "subject": input.subject + }); + + let client = proxy_client(); + let result = client + .post(&target_url) + .header("Authorization", format!("Bearer {}", service_token)) + .header("Content-Type", "application/json") + .json(&request_body) + .send() + .await; + + match result { + Ok(resp) => { + let status = resp.status(); + let headers = resp.headers().clone(); + + let body = match resp.bytes().await { + Ok(b) => b, + Err(e) => { + error!("Error reading report service response: {:?}", e); + return (StatusCode::BAD_GATEWAY, "Error reading upstream response") + .into_response(); + } + }; + + let mut response_builder = Response::builder().status(status); + + if let Some(ct) = headers.get("content-type") { + response_builder = response_builder.header("content-type", ct); + } + + match response_builder.body(axum::body::Body::from(body)) { + Ok(r) => r, + Err(e) => { + error!("Error building proxy response: {:?}", e); + (StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error").into_response() + } + } + } + Err(e) => { + error!("Error sending report to service: {:?}", e); + if e.is_timeout() { + (StatusCode::GATEWAY_TIMEOUT, "Report service timeout").into_response() + } else { + (StatusCode::BAD_GATEWAY, "Report service error").into_response() + } + } + } +} + +async fn create_report_locally( + state: &AppState, + did: &str, + is_takendown: bool, + input: CreateReportInput, +) -> Response { + const REASON_APPEAL: &str = "com.atproto.moderation.defs#reasonAppeal"; + + if is_takendown && input.reason_type != REASON_APPEAL { + return ( + StatusCode::BAD_REQUEST, + Json(json!({"error": "InvalidRequest", "message": "Report not accepted from takendown account"})), + ) + .into_response(); + } + let valid_reason_types = [ "com.atproto.moderation.defs#reasonSpam", "com.atproto.moderation.defs#reasonViolation", @@ -51,8 +228,9 @@ pub async fn create_report( "com.atproto.moderation.defs#reasonSexual", "com.atproto.moderation.defs#reasonRude", "com.atproto.moderation.defs#reasonOther", - "com.atproto.moderation.defs#reasonAppeal", + REASON_APPEAL, ]; + if !valid_reason_types.contains(&input.reason_type.as_str()) { return ( StatusCode::BAD_REQUEST, @@ -60,9 +238,11 @@ pub async fn create_report( ) .into_response(); } + let created_at = chrono::Utc::now(); let report_id = created_at.timestamp_millis(); let subject_json = json!(input.subject); + let insert = sqlx::query!( "INSERT INTO reports (id, reason_type, reason, subject_json, reported_by_did, created_at) VALUES ($1, $2, $3, $4, $5, $6)", report_id, @@ -74,6 +254,7 @@ pub async fn create_report( ) .execute(&state.db) .await; + if let Err(e) = insert { error!("Failed to insert report: {:?}", e); return ( @@ -82,6 +263,14 @@ pub async fn create_report( ) .into_response(); } + + info!( + report_id = %report_id, + reported_by = %did, + reason_type = %input.reason_type, + "Report created locally (no report service configured)" + ); + ( StatusCode::OK, Json(CreateReportOutput { @@ -89,7 +278,7 @@ pub async fn create_report( reason_type: input.reason_type, reason: input.reason, subject: input.subject, - reported_by: did, + reported_by: did.to_string(), created_at: created_at.to_rfc3339(), }), ) diff --git a/src/api/server/service_auth.rs b/src/api/server/service_auth.rs index ffd606c..28fdd9f 100644 --- a/src/api/server/service_auth.rs +++ b/src/api/server/service_auth.rs @@ -95,6 +95,7 @@ pub async fn get_service_auth( did: result.did, is_oauth: true, is_admin: false, + is_takendown: false, scope: result.scope, key_bytes: None, controller_did: None, diff --git a/src/auth/mod.rs b/src/auth/mod.rs index 7c0c644..b403ee0 100644 --- a/src/auth/mod.rs +++ b/src/auth/mod.rs @@ -62,6 +62,7 @@ pub struct AuthenticatedUser { pub key_bytes: Option>, pub is_oauth: bool, pub is_admin: bool, + pub is_takendown: bool, pub scope: Option, pub controller_did: Option, } @@ -117,6 +118,13 @@ pub async fn validate_bearer_token_for_service_auth( validate_bearer_token_with_options_internal(db, None, token, true, true).await } +pub async fn validate_bearer_token_allow_takendown( + db: &PgPool, + token: &str, +) -> Result { + validate_bearer_token_with_options_internal(db, None, token, false, true).await +} + async fn validate_bearer_token_with_options_internal( db: &PgPool, cache: Option<&Arc>, @@ -254,6 +262,7 @@ async fn validate_bearer_token_with_options_internal( key_bytes: Some(decrypted_key), is_oauth: false, is_admin, + is_takendown: takedown_ref.is_some(), scope: token_data.claims.scope.clone(), controller_did, }); @@ -286,7 +295,8 @@ async fn validate_bearer_token_with_options_internal( return Err(TokenValidationError::AccountDeactivated); } - if oauth_token.takedown_ref.is_some() { + let is_takendown = oauth_token.takedown_ref.is_some(); + if !allow_takendown && is_takendown { return Err(TokenValidationError::AccountTakedown); } @@ -304,6 +314,7 @@ async fn validate_bearer_token_with_options_internal( key_bytes, is_oauth: true, is_admin: oauth_token.is_admin, + is_takendown, scope: oauth_info.scope, controller_did: oauth_info.controller_did, }); @@ -364,7 +375,8 @@ pub async fn validate_token_with_dpop( if !allow_deactivated && user_info.deactivated_at.is_some() { return Err(TokenValidationError::AccountDeactivated); } - if user_info.takedown_ref.is_some() { + let is_takendown = user_info.takedown_ref.is_some(); + if is_takendown { return Err(TokenValidationError::AccountTakedown); } let key_bytes = if let (Some(kb), Some(ev)) = @@ -379,6 +391,7 @@ pub async fn validate_token_with_dpop( key_bytes, is_oauth: true, is_admin: user_info.is_admin, + is_takendown, scope: result.scope, controller_did: None, }) diff --git a/tests/moderation.rs b/tests/moderation.rs index ffbc54b..cf393e4 100644 --- a/tests/moderation.rs +++ b/tests/moderation.rs @@ -59,3 +59,215 @@ async fn test_moderation_report_lifecycle() { .expect("Failed to create account report"); assert_eq!(account_report_res.status(), StatusCode::OK); } + +#[tokio::test] +async fn test_moderation_report_invalid_reason_type() { + let client = client(); + let (alice_did, alice_jwt) = setup_new_user("alice-invalid-reason").await; + let report_payload = json!({ + "reasonType": "invalid.reason.type", + "reason": "Testing invalid reason", + "subject": { + "$type": "com.atproto.admin.defs#repoRef", + "did": alice_did + } + }); + let res = client + .post(format!( + "{}/xrpc/com.atproto.moderation.createReport", + base_url().await + )) + .bearer_auth(&alice_jwt) + .json(&report_payload) + .send() + .await + .expect("Failed to send request"); + assert_eq!(res.status(), StatusCode::BAD_REQUEST); + let body: Value = res.json().await.unwrap(); + assert_eq!(body["error"], "InvalidRequest"); + assert!(body["message"] + .as_str() + .unwrap() + .contains("reasonType")); +} + +#[tokio::test] +async fn test_moderation_report_unauthenticated() { + let client = client(); + let report_payload = json!({ + "reasonType": "com.atproto.moderation.defs#reasonSpam", + "reason": "Spam report", + "subject": { + "$type": "com.atproto.admin.defs#repoRef", + "did": "did:plc:test" + } + }); + let res = client + .post(format!( + "{}/xrpc/com.atproto.moderation.createReport", + base_url().await + )) + .json(&report_payload) + .send() + .await + .expect("Failed to send request"); + assert_eq!(res.status(), StatusCode::UNAUTHORIZED); +} + +#[tokio::test] +async fn test_moderation_report_all_reason_types() { + let client = client(); + let (alice_did, alice_jwt) = setup_new_user("alice-all-reasons").await; + let (bob_did, _) = setup_new_user("bob-all-reasons").await; + let reason_types = [ + "com.atproto.moderation.defs#reasonSpam", + "com.atproto.moderation.defs#reasonViolation", + "com.atproto.moderation.defs#reasonMisleading", + "com.atproto.moderation.defs#reasonSexual", + "com.atproto.moderation.defs#reasonRude", + "com.atproto.moderation.defs#reasonOther", + "com.atproto.moderation.defs#reasonAppeal", + ]; + for reason_type in reason_types { + let report_payload = json!({ + "reasonType": reason_type, + "subject": { + "$type": "com.atproto.admin.defs#repoRef", + "did": bob_did + } + }); + let res = client + .post(format!( + "{}/xrpc/com.atproto.moderation.createReport", + base_url().await + )) + .bearer_auth(&alice_jwt) + .json(&report_payload) + .send() + .await + .expect("Failed to send request"); + assert_eq!( + res.status(), + StatusCode::OK, + "Failed for reason type: {}", + reason_type + ); + let body: Value = res.json().await.unwrap(); + assert_eq!(body["reasonType"], reason_type); + assert_eq!(body["reportedBy"], alice_did); + } +} + +#[tokio::test] +async fn test_moderation_report_takendown_user_can_appeal() { + let client = client(); + let (admin_jwt, _) = create_admin_account_and_login(&client).await; + let (target_jwt, target_did) = create_account_and_login(&client).await; + let takedown_payload = json!({ + "subject": { + "$type": "com.atproto.admin.defs#repoRef", + "did": target_did + }, + "takedown": { + "applied": true, + "ref": "mod-action-test" + } + }); + let takedown_res = client + .post(format!( + "{}/xrpc/com.atproto.admin.updateSubjectStatus", + base_url().await + )) + .bearer_auth(&admin_jwt) + .json(&takedown_payload) + .send() + .await + .expect("Failed to takedown"); + assert_eq!(takedown_res.status(), StatusCode::OK); + let appeal_payload = json!({ + "reasonType": "com.atproto.moderation.defs#reasonAppeal", + "reason": "I believe this takedown was a mistake", + "subject": { + "$type": "com.atproto.admin.defs#repoRef", + "did": target_did + } + }); + let appeal_res = client + .post(format!( + "{}/xrpc/com.atproto.moderation.createReport", + base_url().await + )) + .bearer_auth(&target_jwt) + .json(&appeal_payload) + .send() + .await + .expect("Failed to send appeal"); + assert_eq!( + appeal_res.status(), + StatusCode::OK, + "Takendown user should be able to file appeal reports" + ); + let appeal_body: Value = appeal_res.json().await.unwrap(); + assert_eq!( + appeal_body["reasonType"], + "com.atproto.moderation.defs#reasonAppeal" + ); + assert_eq!(appeal_body["reportedBy"], target_did); +} + +#[tokio::test] +async fn test_moderation_report_takendown_user_cannot_file_non_appeal() { + let client = client(); + let (admin_jwt, _) = create_admin_account_and_login(&client).await; + let (target_jwt, target_did) = create_account_and_login(&client).await; + let takedown_payload = json!({ + "subject": { + "$type": "com.atproto.admin.defs#repoRef", + "did": target_did + }, + "takedown": { + "applied": true, + "ref": "mod-action-test-non-appeal" + } + }); + let takedown_res = client + .post(format!( + "{}/xrpc/com.atproto.admin.updateSubjectStatus", + base_url().await + )) + .bearer_auth(&admin_jwt) + .json(&takedown_payload) + .send() + .await + .expect("Failed to takedown"); + assert_eq!(takedown_res.status(), StatusCode::OK); + let report_payload = json!({ + "reasonType": "com.atproto.moderation.defs#reasonSpam", + "reason": "Trying to report spam", + "subject": { + "$type": "com.atproto.admin.defs#repoRef", + "did": "did:plc:test" + } + }); + let report_res = client + .post(format!( + "{}/xrpc/com.atproto.moderation.createReport", + base_url().await + )) + .bearer_auth(&target_jwt) + .json(&report_payload) + .send() + .await + .expect("Failed to send report"); + assert_eq!( + report_res.status(), + StatusCode::BAD_REQUEST, + "Takendown user should not be able to file non-appeal reports" + ); + let body: Value = report_res.json().await.unwrap(); + assert_eq!(body["error"], "InvalidRequest"); + assert!(body["message"] + .as_str() + .unwrap() + .contains("takendown")); +}