diff --git a/Cargo.lock b/Cargo.lock index bb1d35c..0ede77b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -914,6 +914,7 @@ dependencies = [ "bytes", "chrono", "cid", + "ctor", "dotenvy", "jacquard", "jacquard-axum", @@ -1372,6 +1373,22 @@ dependencies = [ "typenum", ] +[[package]] +name = "ctor" +version = "0.6.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "424e0138278faeb2b401f174ad17e715c829512d74f3d1e81eb43365c2e0590e" +dependencies = [ + "ctor-proc-macro", + "dtor", +] + +[[package]] +name = "ctor-proc-macro" +version = "0.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52560adf09603e58c9a7ee1fe1dcb95a16927b17c127f0ac02d6e768a0e25bc1" + [[package]] name = "curve25519-dalek" version = "4.1.3" @@ -1613,6 +1630,21 @@ version = "0.15.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b" +[[package]] +name = "dtor" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "404d02eeb088a82cfd873006cb713fe411306c7d182c344905e101fb1167d301" +dependencies = [ + "dtor-proc-macro", +] + +[[package]] +name = "dtor-proc-macro" +version = "0.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f678cf4a922c215c63e0de95eb1ff08a958a81d47e485cf9da1e27bf6305cfa5" + [[package]] name = "dunce" version = "1.0.5" diff --git a/Cargo.toml b/Cargo.toml index 0ff1d40..5b92d1d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -36,6 +36,7 @@ tracing-subscriber = "0.3.22" uuid = { version = "1.19.0", features = ["v4", "fast-rng"] } [dev-dependencies] +ctor = "0.6.3" testcontainers = "0.26.0" testcontainers-modules = { version = "0.14.0", features = ["postgres"] } wiremock = "0.6.5" diff --git a/migrations/202512211600_moderation_and_status.sql b/migrations/202512211600_moderation_and_status.sql new file mode 100644 index 0000000..f0ce2eb --- /dev/null +++ b/migrations/202512211600_moderation_and_status.sql @@ -0,0 +1,11 @@ +ALTER TABLE users ADD COLUMN deactivated_at TIMESTAMPTZ; + +-- * reports u * +CREATE TABLE reports ( + id BIGINT PRIMARY KEY, + reason_type TEXT NOT NULL, + reason TEXT, + subject_json JSONB NOT NULL, + reported_by_did TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); diff --git a/src/api/moderation/mod.rs b/src/api/moderation/mod.rs index 5877b3b..199fe33 100644 --- a/src/api/moderation/mod.rs +++ b/src/api/moderation/mod.rs @@ -110,8 +110,29 @@ pub async fn create_report( .into_response(); } - let created_at = chrono::Utc::now().to_rfc3339(); - let report_id = chrono::Utc::now().timestamp_millis(); + let created_at = chrono::Utc::now(); + let report_id = created_at.timestamp_millis(); + + 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)" + ) + .bind(report_id) + .bind(&input.reason_type) + .bind(&input.reason) + .bind(json!(input.subject)) + .bind(&did) + .bind(created_at) + .execute(&state.db) + .await; + + if let Err(e) = insert { + error!("Failed to insert report: {:?}", e); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": "InternalError"})), + ) + .into_response(); + } ( StatusCode::OK, @@ -121,7 +142,7 @@ pub async fn create_report( reason: input.reason, subject: input.subject, reported_by: did, - created_at, + created_at: created_at.to_rfc3339(), }), ) .into_response() diff --git a/src/api/repo/blob.rs b/src/api/repo/blob.rs index 2fe1e30..62506e2 100644 --- a/src/api/repo/blob.rs +++ b/src/api/repo/blob.rs @@ -7,11 +7,13 @@ use axum::{ response::{IntoResponse, Response}, }; use cid::Cid; +use jacquard_repo::storage::BlockStore; use multihash::Multihash; use serde::{Deserialize, Serialize}; use serde_json::json; use sha2::{Digest, Sha256}; use sqlx::Row; +use std::str::FromStr; use tracing::error; pub async fn upload_blob( @@ -157,10 +159,33 @@ pub struct ListMissingBlobsOutput { pub blobs: Vec, } +fn find_blobs(val: &serde_json::Value, blobs: &mut Vec) { + if let Some(obj) = val.as_object() { + if let Some(type_val) = obj.get("$type") { + if type_val == "blob" { + if let Some(r) = obj.get("ref") { + if let Some(link) = r.get("$link") { + if let Some(s) = link.as_str() { + blobs.push(s.to_string()); + } + } + } + } + } + for (_, v) in obj { + find_blobs(v, blobs); + } + } else if let Some(arr) = val.as_array() { + for v in arr { + find_blobs(v, blobs); + } + } +} + pub async fn list_missing_blobs( - State(_state): State, + State(state): State, headers: axum::http::HeaderMap, - Query(_params): Query, + Query(params): Query, ) -> Response { let auth_header = headers.get("Authorization"); if auth_header.is_none() { @@ -171,11 +196,153 @@ pub async fn list_missing_blobs( .into_response(); } + let token = auth_header + .unwrap() + .to_str() + .unwrap_or("") + .replace("Bearer ", ""); + + 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, 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(); + } + + let user_query = sqlx::query("SELECT id FROM users WHERE did = $1") + .bind(&did) + .fetch_optional(&state.db) + .await; + + let user_id: uuid::Uuid = match user_query { + Ok(Some(row)) => row.get("id"), + _ => { + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": "InternalError"})), + ) + .into_response(); + } + }; + + let limit = params.limit.unwrap_or(500).min(1000); + let cursor_str = params.cursor.unwrap_or_default(); + let (cursor_collection, cursor_rkey) = if cursor_str.contains('|') { + let parts: Vec<&str> = cursor_str.split('|').collect(); + (parts[0].to_string(), parts[1].to_string()) + } else { + (String::new(), String::new()) + }; + + let records_query = sqlx::query( + "SELECT collection, rkey, record_cid FROM records WHERE repo_id = $1 AND (collection, rkey) > ($2, $3) ORDER BY collection, rkey LIMIT $4" + ) + .bind(user_id) + .bind(cursor_collection) + .bind(cursor_rkey) + .bind(limit) + .fetch_all(&state.db) + .await; + + let records = match records_query { + Ok(r) => r, + Err(e) => { + error!("DB error fetching records: {:?}", e); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": "InternalError"})), + ) + .into_response(); + } + }; + + let mut missing_blobs = Vec::new(); + let mut last_cursor = None; + + for row in &records { + let collection: String = row.get("collection"); + let rkey: String = row.get("rkey"); + let record_cid_str: String = row.get("record_cid"); + + last_cursor = Some(format!("{}|{}", collection, rkey)); + + let record_cid = match Cid::from_str(&record_cid_str) { + Ok(c) => c, + Err(_) => continue, + }; + + let block_bytes = match state.block_store.get(&record_cid).await { + Ok(Some(b)) => b, + _ => continue, + }; + + let record_val: serde_json::Value = match serde_ipld_dagcbor::from_slice(&block_bytes) { + Ok(v) => v, + Err(_) => continue, + }; + + let mut blobs = Vec::new(); + find_blobs(&record_val, &mut blobs); + + for blob_cid_str in blobs { + let exists = sqlx::query("SELECT 1 FROM blobs WHERE cid = $1 AND created_by_user = $2") + .bind(&blob_cid_str) + .bind(user_id) + .fetch_optional(&state.db) + .await; + + match exists { + Ok(None) => { + missing_blobs.push(RecordBlob { + cid: blob_cid_str, + record_uri: format!("at://{}/{}/{}", did, collection, rkey), + }); + } + Err(e) => { + error!("DB error checking blob existence: {:?}", e); + } + _ => {} + } + } + } + + // if we fetched fewer records than limit, we are done, so cursor is None. + // otherwise, cursor is the last one we saw. + // ...right? + let next_cursor = if records.len() < limit as usize { + None + } else { + last_cursor + }; + ( StatusCode::OK, Json(ListMissingBlobsOutput { - cursor: None, - blobs: vec![], + cursor: next_cursor, + blobs: missing_blobs, }), ) .into_response() diff --git a/src/api/repo/record/write.rs b/src/api/repo/record/write.rs index 4e85968..93ada7f 100644 --- a/src/api/repo/record/write.rs +++ b/src/api/repo/record/write.rs @@ -182,6 +182,18 @@ pub async fn create_record( .rkey .unwrap_or_else(|| Utc::now().format("%Y%m%d%H%M%S%f").to_string()); + if input.validate.unwrap_or(true) { + if input.collection == "app.bsky.feed.post" { + if input.record.get("text").is_none() || input.record.get("createdAt").is_none() { + return ( + StatusCode::BAD_REQUEST, + Json(json!({"error": "InvalidRecord", "message": "Record validation failed"})), + ) + .into_response(); + } + } + } + let mut record_bytes = Vec::new(); if let Err(e) = serde_ipld_dagcbor::to_writer(&mut record_bytes, &input.record) { error!("Error serializing record: {:?}", e); @@ -472,6 +484,18 @@ pub async fn put_record( let rkey = input.rkey.clone(); + if input.validate.unwrap_or(true) { + if input.collection == "app.bsky.feed.post" { + if input.record.get("text").is_none() || input.record.get("createdAt").is_none() { + return ( + StatusCode::BAD_REQUEST, + Json(json!({"error": "InvalidRecord", "message": "Record validation failed"})), + ) + .into_response(); + } + } + } + let mut record_bytes = Vec::new(); if let Err(e) = serde_ipld_dagcbor::to_writer(&mut record_bytes, &input.record) { error!("Error serializing record: {:?}", e); diff --git a/src/api/server/session.rs b/src/api/server/session.rs index e09203d..5159faf 100644 --- a/src/api/server/session.rs +++ b/src/api/server/session.rs @@ -561,6 +561,16 @@ pub async fn check_account_status( .into_response(); } + let user_status = sqlx::query("SELECT deactivated_at FROM users WHERE did = $1") + .bind(&did) + .fetch_optional(&state.db) + .await; + + let deactivated_at: Option> = match user_status { + Ok(Some(row)) => row.get("deactivated_at"), + _ => None, + }; + let repo_result = sqlx::query("SELECT repo_root_cid FROM repos WHERE user_id = $1") .bind(user_id) .fetch_optional(&state.db) @@ -589,7 +599,7 @@ pub async fn check_account_status( ( StatusCode::OK, Json(CheckAccountStatusOutput { - activated: true, + activated: deactivated_at.is_none(), valid_did, repo_commit: repo_commit.clone(), repo_rev: chrono::Utc::now().timestamp_millis().to_string(), @@ -604,7 +614,7 @@ pub async fn check_account_status( } pub async fn activate_account( - State(_state): State, + State(state): State, headers: axum::http::HeaderMap, ) -> Response { let auth_header = headers.get("Authorization"); @@ -616,7 +626,71 @@ pub async fn activate_account( .into_response(); } - (StatusCode::OK, Json(json!({}))).into_response() + let token = auth_header + .unwrap() + .to_str() + .unwrap_or("") + .replace("Bearer ", ""); + + let session = sqlx::query( + r#" + 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; + + let (did, key_bytes) = match session { + Ok(Some(row)) => ( + row.get::("did"), + row.get::, _>("key_bytes"), + ), + Ok(None) => { + return ( + StatusCode::UNAUTHORIZED, + Json(json!({"error": "AuthenticationFailed"})), + ) + .into_response(); + } + Err(e) => { + error!("DB error in activate_account: {:?}", e); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": "InternalError"})), + ) + .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(); + } + + let result = sqlx::query("UPDATE users SET deactivated_at = NULL WHERE did = $1") + .bind(&did) + .execute(&state.db) + .await; + + match result { + Ok(_) => (StatusCode::OK, Json(json!({}))).into_response(), + Err(e) => { + error!("DB error activating account: {:?}", e); + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": "InternalError"})), + ) + .into_response() + } + } } #[derive(Deserialize)] @@ -626,7 +700,7 @@ pub struct DeactivateAccountInput { } pub async fn deactivate_account( - State(_state): State, + State(state): State, headers: axum::http::HeaderMap, Json(_input): Json, ) -> Response { @@ -639,7 +713,71 @@ pub async fn deactivate_account( .into_response(); } - (StatusCode::OK, Json(json!({}))).into_response() + let token = auth_header + .unwrap() + .to_str() + .unwrap_or("") + .replace("Bearer ", ""); + + let session = sqlx::query( + r#" + 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; + + let (did, key_bytes) = match session { + Ok(Some(row)) => ( + row.get::("did"), + row.get::, _>("key_bytes"), + ), + Ok(None) => { + return ( + StatusCode::UNAUTHORIZED, + Json(json!({"error": "AuthenticationFailed"})), + ) + .into_response(); + } + Err(e) => { + error!("DB error in deactivate_account: {:?}", e); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": "InternalError"})), + ) + .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(); + } + + let result = sqlx::query("UPDATE users SET deactivated_at = NOW() WHERE did = $1") + .bind(&did) + .execute(&state.db) + .await; + + match result { + Ok(_) => (StatusCode::OK, Json(json!({}))).into_response(), + Err(e) => { + error!("DB error deactivating account: {:?}", e); + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": "InternalError"})), + ) + .into_response() + } + } } #[derive(Serialize)] diff --git a/src/sync/mod.rs b/src/sync/mod.rs index c0cd9e2..6750a54 100644 --- a/src/sync/mod.rs +++ b/src/sync/mod.rs @@ -461,6 +461,8 @@ pub async fn notify_of_update( Query(params): Query, ) -> Response { info!("Received notifyOfUpdate from hostname: {}", params.hostname); + // TODO: Queue job for crawler interaction or relay notification + info!("TODO: Queue job for notifyOfUpdate (not implemented)"); (StatusCode::OK, Json(json!({}))).into_response() } @@ -475,6 +477,8 @@ pub async fn request_crawl( Json(input): Json, ) -> Response { info!("Received requestCrawl for hostname: {}", input.hostname); + // TODO: Queue job for crawling + info!("TODO: Queue job for requestCrawl (not implemented)"); (StatusCode::OK, Json(json!({}))).into_response() } diff --git a/tests/common/mod.rs b/tests/common/mod.rs index dbf3388..0b935fb 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -32,6 +32,22 @@ pub const AUTH_DID: &str = "did:plc:fake"; #[allow(dead_code)] pub const TARGET_DID: &str = "did:plc:target"; +#[cfg(test)] +#[ctor::dtor] +fn cleanup() { + // my attempt to force clean up containers created by this test binary. + // this is a fallback in case ryuk fails or is not supported + if std::env::var("XDG_RUNTIME_DIR").is_ok() { + let _ = std::process::Command::new("podman") + .args(&["rm", "-f", "--filter", "label=bspds_test=true"]) + .output(); + } + + let _ = std::process::Command::new("docker") + .args(&["container", "prune", "-f", "--filter", "label=bspds_test=true"]) + .output(); +} + #[allow(dead_code)] pub fn client() -> Client { Client::new() @@ -63,6 +79,7 @@ pub async fn base_url() -> &'static str { .with_env_var("MINIO_ROOT_USER", "minioadmin") .with_env_var("MINIO_ROOT_PASSWORD", "minioadmin") .with_cmd(vec!["server".to_string(), "/data".to_string()]) + .with_label("bspds_test", "true") .start() .await .expect("Failed to start MinIO"); @@ -131,6 +148,7 @@ pub async fn base_url() -> &'static str { let container = Postgres::default() .with_tag("18-alpine") + .with_label("bspds_test", "true") .start() .await .expect("Failed to start Postgres"); diff --git a/tests/repo.rs b/tests/repo.rs index 1841f2f..81c2aff 100644 --- a/tests/repo.rs +++ b/tests/repo.rs @@ -204,7 +204,6 @@ async fn test_put_record_mismatched_repo() { } #[tokio::test] -#[ignore] async fn test_put_record_invalid_schema() { let client = client(); let (token, did) = create_account_and_login(&client).await;