diff --git a/cruciverb-server/src/activitypub/inbox.rs b/cruciverb-server/src/activitypub/inbox.rs new file mode 100644 index 0000000..31054fc --- /dev/null +++ b/cruciverb-server/src/activitypub/inbox.rs @@ -0,0 +1,1015 @@ +use std::sync::Arc; +use axum::extract::State; +use axum::http::HeaderMap; +use axum::response::{IntoResponse, Response}; + +use crate::state::AppState; +use super::types::*; + +fn status_err(code: u16, msg: &str) -> Response { + (axum::http::StatusCode::from_u16(code).unwrap(), msg.to_string()).into_response() +} + +fn accepted() -> Response { + (axum::http::StatusCode::ACCEPTED, "").into_response() +} + +fn strip_html(s: &str) -> String { + let mut out = String::with_capacity(s.len()); + let mut in_tag = false; + for c in s.chars() { + if c == '<' { in_tag = true; continue; } + if c == '>' { in_tag = false; continue; } + if !in_tag { out.push(c); } + } + out.trim().to_string() +} + +const MAX_INBOX_BODY: usize = 256 * 1024; // 256KB max +const INBOX_RATE_MAX: u32 = 60; // 60 activities per minute per domain + +pub async fn inbox( + State(state): State>, + headers: HeaderMap, + body: String, +) -> Result { + let ap = state.ap.as_ref() + .ok_or_else(|| status_err(404, ""))?; + + // #16: federation freeze - return 503 when frozen + if super::is_frozen() { + return Err(status_err(503, "federation temporarily frozen")); + } + + // accept ActivityPub content types + let content_type = headers.get("content-type") + .and_then(|v| v.to_str().ok()) + .unwrap_or(""); + if !content_type.is_empty() + && !content_type.contains("application/json") + && !content_type.contains("application/activity+json") + && !content_type.contains("application/ld+json") { + return Err(status_err(415, "unsupported content type")); + } + + // payload size check + if body.len() > MAX_INBOX_BODY { + return Err(status_err(413, "payload too large")); + } + + let activity: serde_json::Value = serde_json::from_str(&body) + .map_err(|_| status_err(400, "invalid JSON"))?; + + let activity_type = activity.get("type").and_then(|v| v.as_str()).unwrap_or(""); + let actor_uri = activity.get("actor").and_then(|v| v.as_str()).unwrap_or(""); + + if actor_uri.is_empty() { + return Err(status_err(400, "missing actor")); + } + + // #7: enforce HTTPS on all actor URIs + if !actor_uri.starts_with("https://") { + return Err(status_err(400, "actor URI must use HTTPS")); + } + + // require HTTP signature on all inbox activities + let has_signature = headers.iter().any(|(k, _)| k.as_str().to_lowercase() == "signature"); + if !has_signature { + return Err(status_err(401, "missing HTTP signature")); + } + + let sender_domain = actor_uri.split("//").nth(1).and_then(|s| s.split('/').next()).unwrap_or(""); + + // #3: activity domain verification - actor domain must match request origin + { + // verify that the actor URI domain matches the signing key domain + let sig_header = headers.iter() + .find(|(k, _)| k.as_str().to_lowercase() == "signature") + .and_then(|(_, v)| v.to_str().ok()) + .unwrap_or(""); + if !sig_header.is_empty() { + // extract keyId from signature header + let key_id_domain = sig_header.split("keyId=\"") + .nth(1) + .and_then(|s| s.split('"').next()) + .and_then(|uri| uri.split("//").nth(1)) + .and_then(|s| s.split('/').next()) + .unwrap_or(""); + if !key_id_domain.is_empty() && key_id_domain != sender_domain { + tracing::warn!("domain mismatch: actor {} but keyId domain {}", sender_domain, key_id_domain); + return Err(status_err(403, "actor domain does not match signing key domain")); + } + } + } + + // replay protection - reject requests with missing or stale Date header + let date_header = headers.get("date").and_then(|v| v.to_str().ok()).unwrap_or(""); + if date_header.is_empty() { + return Err(status_err(401, "missing Date header")); + } + if let Ok(parsed) = chrono::DateTime::parse_from_rfc2822(date_header) { + let age = chrono::Utc::now().signed_duration_since(parsed); + if age.num_seconds().abs() > 300 { + tracing::warn!("stale Date header from {}: {} seconds old", actor_uri, age.num_seconds()); + return Err(status_err(401, "request too old or too far in future")); + } + } else { + return Err(status_err(401, "invalid Date header format")); + } + + // check blocklist, allowlist, and load config in one DB lock + let allowlist_mode; + { + let dbi = state.db.lock().await; + let blocked: bool = dbi.db.query_row( + "SELECT COUNT(*) > 0 FROM ap_blocked WHERE domain = ?1 OR ?1 LIKE '%.' || domain", + [sender_domain], |r| r.get(0), + ).unwrap_or(false); + if blocked { + return Err(status_err(403, "blocked")); + } + + allowlist_mode = crate::db::get_config_bool(&dbi.db, "AP_ALLOWLIST_MODE"); + if allowlist_mode { + let is_peer: bool = dbi.db.query_row( + "SELECT COUNT(*) > 0 FROM ap_peers WHERE domain = ?1", + [sender_domain], |r| r.get(0), + ).unwrap_or(false); + if !is_peer { + tracing::info!("allowlist mode: rejected activity from non-peer {}", sender_domain); + return Err(status_err(403, "instance not on allowlist")); + } + } + } + + // per-domain rate limiting + if let Err(_) = crate::security::check_rate_limit(sender_domain, "ap_inbox", INBOX_RATE_MAX, 60) { + tracing::warn!("rate limited inbox from {}", sender_domain); + return Err(status_err(429, "too many requests")); + } + + // track federation stats + { + let dbi = state.db.lock().await; + crate::db::track_federation_activity(&dbi.db, sender_domain, "inbound"); + } + + // verify HTTP signature + let header_pairs: Vec<(String, String)> = headers.iter() + .filter_map(|(k, v)| Some((k.as_str().to_string(), v.to_str().ok()?.to_string()))) + .collect(); + + let sender_key = fetch_actor_key(actor_uri, &state).await + .map_err(|e| status_err(401, &format!("failed to fetch actor key: {}", e)))?; + + let actor_path = format!("/@{}/inbox", ap.actor_name); + let shared_path = "/inbox"; + let sig_result_1 = super::signatures::verify_request(&sender_key, "POST", &actor_path, &header_pairs, Some(body.as_bytes())); + let sig_result_2 = super::signatures::verify_request(&sender_key, "POST", shared_path, &header_pairs, Some(body.as_bytes())); + let sig_ok = sig_result_1.is_ok() || sig_result_2.is_ok(); + if !sig_ok { + tracing::warn!("sig verify actor path ({}): {:?}", actor_path, sig_result_1.err()); + tracing::warn!("sig verify shared path ({}): {:?}", shared_path, sig_result_2.err()); + // key might have rotated - try refetching + let mut retry_ok = false; + if let Ok(fresh_key) = refetch_actor_key(actor_uri, &state).await { + retry_ok = super::signatures::verify_request(&fresh_key, "POST", &actor_path, &header_pairs, Some(body.as_bytes())).is_ok() + || super::signatures::verify_request(&fresh_key, "POST", shared_path, &header_pairs, Some(body.as_bytes())).is_ok(); + } + if !retry_ok { + tracing::warn!("signature verification failed from {}", actor_uri); + return Err(status_err(401, "signature error")); + } + } + + match activity_type { + "Follow" => handle_follow(&state, &activity, actor_uri).await, + "Undo" => handle_undo(&state, &activity, actor_uri).await, + "Create" => handle_create(&state, &activity).await, + "Like" => handle_vote(&state, &activity, 1).await, + "Dislike" => handle_vote(&state, &activity, -1).await, + "Delete" => handle_delete(&state, &activity, actor_uri).await, + "Move" => handle_move(&state, &activity, actor_uri).await, + "Announce" => handle_announce(&state, &activity).await, + _ => { + tracing::debug!("ignoring activity type: {}", activity_type); + Ok(accepted()) + } + } +} + +async fn handle_follow(state: &Arc, activity: &serde_json::Value, actor_uri: &str) -> Result { + let ap = state.ap.as_ref().unwrap(); + + // #7: reject http:// follower URIs + if !actor_uri.starts_with("https://") { + return Err(status_err(400, "follower must use HTTPS")); + } + + // extract inbox URI from actor + let actor_inbox = fetch_actor_inbox(actor_uri, state).await.unwrap_or_default(); + if actor_inbox.is_empty() { + return Ok(accepted()); + } + + // #5: follower approval mode + let approve_followers = { let dbi = state.db.lock().await; crate::db::get_config_bool(&dbi.db, "AP_APPROVE_FOLLOWERS") }; + let accepted_val = if approve_followers { 0 } else { 1 }; + + let dbi = state.db.lock().await; + let now = chrono::Utc::now().to_rfc3339(); + dbi.db.execute( + "INSERT OR REPLACE INTO ap_followers (actor_uri, inbox_uri, accepted, created_at) VALUES (?1, ?2, ?3, ?4)", + rusqlite::params![actor_uri, actor_inbox, accepted_val, now], + ).ok(); + + if approve_followers { + tracing::info!("follow request from {} held for approval", actor_uri); + return Ok(accepted()); + } + + // send Accept + let accept = serde_json::json!({ + "@context": [CONTEXT, SECURITY_CONTEXT], + "id": format!("{}/accept/{}", ap.actor_uri, uuid::Uuid::new_v4()), + "type": "Accept", + "actor": ap.actor_uri, + "object": activity, + }); + + let payload = serde_json::to_string(&accept).unwrap(); + let retry = chrono::Utc::now().to_rfc3339(); + dbi.db.execute( + "INSERT INTO ap_delivery_queue (inbox_uri, payload, next_retry, created_at) VALUES (?1, ?2, ?3, ?3)", + rusqlite::params![actor_inbox, payload, retry], + ).ok(); + + tracing::info!("accepted follow from {}", actor_uri); + + // auto-peer detection: check if the follower is another cruciverb instance + let sender_domain = actor_uri.split("//").nth(1).and_then(|s| s.split('/').next()).unwrap_or(""); + let is_already_peer: bool = dbi.db.query_row( + "SELECT COUNT(*) > 0 FROM ap_peers WHERE domain = ?1", [sender_domain], |r| r.get(0) + ).unwrap_or(false); + + if !is_already_peer && !sender_domain.is_empty() { + // try to detect if this is a cruciverb instance via nodeinfo + drop(dbi); + let is_cruciverb = check_is_cruciverb(sender_domain).await; + if is_cruciverb { + let dbi = state.db.lock().await; + // fetch and cache their key (already done by signature verification, should be in ap_peers) + let already = dbi.db.query_row("SELECT COUNT(*) > 0 FROM ap_peers WHERE domain = ?1", [sender_domain], |r| r.get::<_, bool>(0)).unwrap_or(false); + if !already { + dbi.db.execute( + "INSERT OR REPLACE INTO ap_peers (domain, actor_uri, inbox_uri, last_seen, created_at) + VALUES (?1, ?2, ?3, ?4, ?4)", + rusqlite::params![sender_domain, actor_uri, actor_inbox, chrono::Utc::now().to_rfc3339()], + ).ok(); + tracing::info!("auto-peered with cruciverb instance {}", sender_domain); + } + } + } + + Ok(super::inbox::accepted()) +} + +async fn handle_undo(state: &Arc, activity: &serde_json::Value, actor_uri: &str) -> Result { + let object_type = activity.get("object") + .and_then(|o| o.get("type")).and_then(|t| t.as_str()).unwrap_or(""); + + if object_type == "Follow" { + let dbi = state.db.lock().await; + dbi.db.execute("DELETE FROM ap_followers WHERE actor_uri = ?1", [actor_uri]).ok(); + tracing::info!("unfollowed by {}", actor_uri); + } + + Ok(accepted()) +} + +async fn handle_create(state: &Arc, activity: &serde_json::Value) -> Result { + let object = activity.get("object").ok_or_else(|| status_err(400, ""))?; + + // check if this is a reply to one of our outbox posts + if let Some(in_reply_to) = object.get("inReplyTo").and_then(|v| v.as_str()) { + if let Some(ap) = state.ap.as_ref() { + let is_ours = in_reply_to.starts_with(&ap.actor_uri) + || in_reply_to.starts_with(&format!("https://{}/ap/", ap.domain)); + if is_ours { + let dbi = state.db.lock().await; + // verify the outbox entry exists + let outbox_activity: Option = dbi.db.query_row( + "SELECT activity FROM ap_outbox WHERE id = ?1 OR activity LIKE '%' || ?1 || '%'", + [in_reply_to], |r| r.get(0), + ).ok(); + + if outbox_activity.is_some() { + let actor_uri = activity.get("actor").and_then(|v| v.as_str()).unwrap_or(""); + let content_raw = object.get("content").and_then(|v| v.as_str()).unwrap_or(""); + let content = strip_html(content_raw); + let published = object.get("published").and_then(|v| v.as_str()); + let actor_name = object.get("attributedTo") + .and_then(|v| v.as_str()) + .or_else(|| activity.get("actor").and_then(|v| v.as_str())); + + // try to extract puzzle_id from the outbox activity's url + let puzzle_id = outbox_activity.as_ref().and_then(|act_json| { + let act: serde_json::Value = serde_json::from_str(act_json).ok()?; + let url = act.get("object")?.get("url")?.as_str()?; + // url looks like https://domain/p/{short_id} + let short_id = url.rsplit('/').next()?; + let (pid, _) = crate::db::get_puzzle_by_short_id(&dbi.db, short_id)?; + Some(pid) + }); + + crate::db::add_ap_reply( + &dbi.db, in_reply_to, + puzzle_id.as_deref(), + actor_uri, + actor_name, + &content, + published, + ); + tracing::info!("reply from {} on {}", actor_uri, in_reply_to); + return Ok(accepted()); + } + } + } + } + + let tags = object.get("tag").and_then(|t| t.as_array()); + let has_tag = |name: &str| -> bool { + tags.map(|t| t.iter().any(|t| t.get("name").and_then(|n| n.as_str()) == Some(name))).unwrap_or(false) + }; + + // handle leaderboard data from peers + if has_tag("#cruciverb-leaderboard") { + return handle_leaderboard(state, activity, object).await; + } + + // handle full puzzle data from peers + if has_tag("#cruciverb-puzzle") { + handle_puzzle_data(state, activity, object).await; + // don't return - fall through to also check for clue if tagged + } + + // check for cruciverb-clue tag + let is_clue = has_tag("#cruciverb-clue"); + + if !is_clue { + return Ok(accepted()); + } + + // extract word and clue from attachments + let attachments = object.get("attachment").and_then(|a| a.as_array()); + let word = attachments.and_then(|a| a.iter() + .find(|att| att.get("name").and_then(|n| n.as_str()) == Some("cruciverb:word")) + .and_then(|att| att.get("value").and_then(|v| v.as_str()))); + let clue = attachments.and_then(|a| a.iter() + .find(|att| att.get("name").and_then(|n| n.as_str()) == Some("cruciverb:clue")) + .and_then(|att| att.get("value").and_then(|v| v.as_str()))); + + if let (Some(word), Some(clue)) = (word, clue) { + let word_upper = word.to_uppercase(); + + // validate: word must be alphabetic, 3-25 chars + if word_upper.len() < 3 || word_upper.len() > 25 || !word_upper.chars().all(|c| c.is_alphabetic()) { + tracing::debug!("rejected federated clue: invalid word '{}'", word); + return Ok(accepted()); + } + + // validate: clue must be reasonable length + if clue.len() > 500 || clue.is_empty() { + tracing::debug!("rejected federated clue: clue too long or empty"); + return Ok(accepted()); + } + + // reject clues containing URLs + if crate::db::text_contains_url(clue) { + tracing::debug!("rejected federated clue: contains URL"); + return Ok(accepted()); + } + + // sanitize content + let clean_clue = crate::security::sanitize_html(clue); + + // check against the instance blocklist (loaded from DATA_DIR, not a + // hardcoded relative path that misses non-default deployments) + if state.blocklist.read().unwrap().contains_blocked(&word_upper) { + tracing::debug!("rejected federated clue: banned word '{}'", word); + return Ok(accepted()); + } + + let origin_id = object.get("id").and_then(|v| v.as_str()).unwrap_or(""); + let actor = activity.get("actor").and_then(|v| v.as_str()).unwrap_or(""); + let origin_domain = actor.split("//").nth(1).and_then(|s| s.split('/').next()).unwrap_or(""); + + let dbi = state.db.lock().await; + + // check instance reputation + let rep_score = crate::db::get_instance_reputation(&dbi.db, origin_domain); + if rep_score < 20.0 { + tracing::warn!("auto-rejecting clue from {} (reputation {:.0})", origin_domain, rep_score); + // auto-block + let now = chrono::Utc::now().to_rfc3339(); + dbi.db.execute( + "INSERT OR IGNORE INTO ap_blocked (domain, reason, blocked_at) VALUES (?1, ?2, ?3)", + rusqlite::params![origin_domain, "auto-blocked: reputation below 20", now], + ).ok(); + crate::db::log_admin_action(&dbi.db, "auto_block_instance", Some(origin_domain), + Some(&format!("reputation {:.0}", rep_score))); + return Ok(accepted()); + } + + // suspicious activity: >50 clues in last hour + let recent = crate::db::count_recent_clues_from_domain(&dbi.db, origin_domain); + if recent > 50 { + crate::db::log_admin_action(&dbi.db, "suspicious_activity", Some(origin_domain), + Some(&format!("{} clues in last hour", recent))); + } + + // repeat offender: 5+ consecutive rejections + let consecutive = crate::db::count_consecutive_rejections(&dbi.db, origin_domain); + if consecutive >= 5 { + crate::db::log_admin_action(&dbi.db, "repeat_offender", Some(origin_domain), + Some(&format!("{} consecutive rejections", consecutive))); + } + + // per-domain clue limit: max 100 pending clues from one domain + let domain_count: i64 = dbi.db.query_row( + "SELECT COUNT(*) FROM community_clues WHERE origin_domain = ?1 AND status = 'pending'", + [origin_domain], |r| r.get(0), + ).unwrap_or(0); + if domain_count >= 100 { + tracing::warn!("rejected federated clue from {}: too many pending clues", origin_domain); + return Ok(accepted()); + } + + // skip duplicates + let exists: bool = dbi.db.query_row( + "SELECT COUNT(*) > 0 FROM community_clues WHERE origin_id = ?1", + [origin_id], + |row| row.get(0), + ).unwrap_or(false); + + if !exists { + // set status based on reputation + let mut status = if rep_score < 50.0 { "held" } else { "pending" }; + + // #2: new peer review - hold clues from domains with fewer than 10 accepted clues + if crate::db::get_config_bool(&dbi.db, "AP_NEW_PEER_REVIEW") { + let accepted_count: i64 = dbi.db.query_row( + "SELECT COUNT(*) FROM community_clues WHERE origin_domain = ?1 AND status = 'approved'", + [origin_domain], |r| r.get(0), + ).unwrap_or(0); + if accepted_count < 10 { + status = "held"; + } + } + + let now = chrono::Utc::now().to_rfc3339(); + dbi.db.execute( + "INSERT INTO community_clues (word, clue_text, submitted_by, status, created_at, origin_domain, origin_id) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", + rusqlite::params![word_upper, clean_clue, actor, status, now, origin_domain, origin_id], + ).ok(); + if status == "held" { + tracing::info!("held federated clue for {} from {} (reputation {:.0})", word_upper, origin_domain, rep_score); + } else { + tracing::info!("received federated clue for {} from {}", word_upper, origin_domain); + } + } + } + + Ok(accepted()) +} + +async fn handle_vote(state: &Arc, activity: &serde_json::Value, vote: i32) -> Result { + let object_uri = activity.get("object").and_then(|v| v.as_str()).unwrap_or(""); + let actor = activity.get("actor").and_then(|v| v.as_str()).unwrap_or(""); + let origin_domain = actor.split("//").nth(1).and_then(|s| s.split('/').next()).unwrap_or(""); + + if object_uri.is_empty() { + return Ok(accepted()); + } + + // find the clue by origin_id + let dbi = state.db.lock().await; + let clue_id: Option = dbi.db.query_row( + "SELECT id FROM community_clues WHERE origin_id = ?1", + [object_uri], + |row| row.get(0), + ).ok(); + + if let Some(id) = clue_id { + if vote > 0 { + dbi.db.execute("UPDATE community_clues SET upvotes = upvotes + 1 WHERE id = ?1", [id]).ok(); + } else { + dbi.db.execute("UPDATE community_clues SET downvotes = downvotes + 1 WHERE id = ?1", [id]).ok(); + } + + // auto-approve at +3 net + let (up, down): (i64, i64) = dbi.db.query_row( + "SELECT upvotes, downvotes FROM community_clues WHERE id = ?1", + [id], |row| Ok((row.get(0)?, row.get(1)?)), + ).unwrap_or((0, 0)); + if up - down >= 3 { + dbi.db.execute("UPDATE community_clues SET status = 'approved' WHERE id = ?1 AND status = 'pending'", [id]).ok(); + } + + tracing::debug!("federated vote on clue {} from {}", id, origin_domain); + } + + Ok(accepted()) +} + +async fn handle_delete(state: &Arc, activity: &serde_json::Value, actor_uri: &str) -> Result { + let object_id = match activity.get("object") { + Some(serde_json::Value::String(s)) => s.clone(), + Some(obj) => obj.get("id").and_then(|v| v.as_str()).unwrap_or("").to_string(), + None => return Ok(accepted()), + }; + if object_id.is_empty() { return Ok(accepted()); } + + let dbi = state.db.lock().await; + // only delete if the actor domain matches the object domain + let actor_domain = actor_uri.split("//").nth(1).and_then(|s| s.split('/').next()).unwrap_or(""); + let object_domain = object_id.split("//").nth(1).and_then(|s| s.split('/').next()).unwrap_or(""); + if actor_domain == object_domain { + dbi.db.execute("DELETE FROM community_clues WHERE origin_id = ?1", [&object_id]).ok(); + // also delete federated puzzles matching this object + let fed_id = format!("fed-{}-{}", actor_domain, + object_id.rsplit('/').next().unwrap_or("")); + dbi.db.execute("DELETE FROM user_puzzles WHERE id = ?1 AND session_token = 'federation'", [&fed_id]).ok(); + // also try matching by origin_domain + the puzzle ID from the URI + if let Some(puzzle_part) = object_id.rsplit('/').next() { + dbi.db.execute( + "DELETE FROM user_puzzles WHERE session_token = 'federation' AND origin_domain = ?1 AND id LIKE '%' || ?2", + rusqlite::params![actor_domain, puzzle_part], + ).ok(); + } + tracing::info!("deleted federated object {} from {}", object_id, actor_domain); + } + Ok(accepted()) +} + +// #6: sign outbound GET requests +fn build_signed_client(state: &Arc) -> (reqwest::Client, Option<(String, rsa::RsaPrivateKey)>) { + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(10)) + .redirect(reqwest::redirect::Policy::limited(3)) + .build() + .unwrap_or_else(|_| reqwest::Client::new()); + let signing_info = state.ap.as_ref().map(|ap| { + let key_id = format!("{}#main-key", ap.actor_uri); + (key_id, ap.keys.private_key.clone()) + }); + (client, signing_info) +} + +async fn signed_get(url: &str, state: &Arc) -> Result { + // #8: DNS rebinding protection + if crate::security::is_internal_url(url) { + return Err("refusing to fetch internal URL".into()); + } + + let (client, signing_info) = build_signed_client(state); + + let host = url.split("//").nth(1) + .and_then(|s| s.split('/').next()) + .unwrap_or(""); + let path = url.split("//").nth(1) + .and_then(|s| s.find('/').map(|i| &s[i..])) + .unwrap_or("/"); + + let mut req = client.get(url) + .header("Accept", "application/activity+json"); + + // sign GET requests if we have keys + if let Some((key_id, private_key)) = signing_info { + let sig_headers = super::signatures::sign_request( + &private_key, &key_id, "GET", path, host, None, + ); + for (name, value) in &sig_headers { + req = req.header(name.as_str(), value.as_str()); + } + } + + let resp = req.send().await.map_err(|e| format!("fetch failed: {}", e))?; + let final_url = resp.url().as_str(); + if crate::security::is_internal_url(final_url) { + return Err("redirect to internal URL blocked".into()); + } + Ok(resp) +} + +async fn fetch_actor_key(actor_uri: &str, state: &Arc) -> Result { + // #7: enforce HTTPS + if !actor_uri.starts_with("https://") { + return Err("actor URI must use HTTPS".into()); + } + + // check cache first + let domain = actor_uri.split("//").nth(1).and_then(|s| s.split('/').next()).unwrap_or(""); + { + let dbi = state.db.lock().await; + if let Ok(pem) = dbi.db.query_row( + "SELECT public_pem FROM ap_peers WHERE domain = ?1", [domain], |row| row.get::<_, String>(0), + ) { + return Ok(pem); + } + } + + // #6: signed GET request + let resp = signed_get(actor_uri, state).await?; + let content_type = resp.headers().get("content-type").and_then(|v| v.to_str().ok()).unwrap_or(""); + if !content_type.contains("json") { + return Err(format!("unexpected content type from actor: {}", content_type)); + } + let actor: serde_json::Value = resp.json().await + .map_err(|e| format!("invalid actor JSON: {}", e))?; + + let public_key_pem = actor.get("publicKey") + .and_then(|pk| pk.get("publicKeyPem")) + .and_then(|v| v.as_str()) + .ok_or("no publicKeyPem in actor")? + .to_string(); + let inbox_uri = actor.get("inbox").and_then(|v| v.as_str()).unwrap_or("").to_string(); + + // key change detection + { + let dbi = state.db.lock().await; + if let Ok(old_pem) = dbi.db.query_row( + "SELECT public_pem FROM ap_peers WHERE domain = ?1", [domain], |r| r.get::<_, String>(0), + ) { + if !old_pem.is_empty() && old_pem != public_key_pem { + tracing::warn!("PUBLIC KEY CHANGED for {} - old key replaced. This could indicate compromise.", domain); + crate::db::log_admin_action(&dbi.db, "key_change_detected", Some(domain), + Some("public key changed - verify this is legitimate")); + } + } + } + + // cache the peer + let dbi = state.db.lock().await; + let now = chrono::Utc::now().to_rfc3339(); + dbi.db.execute( + "INSERT OR REPLACE INTO ap_peers (domain, actor_uri, inbox_uri, public_pem, last_seen, created_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?5)", + rusqlite::params![domain, actor_uri, inbox_uri, public_key_pem, now], + ).ok(); + + Ok(public_key_pem) +} + +async fn refetch_actor_key(actor_uri: &str, state: &Arc) -> Result { + if !actor_uri.starts_with("https://") { + return Err("actor URI must use HTTPS".into()); + } + + let resp = signed_get(actor_uri, state).await?; + let content_type = resp.headers().get("content-type").and_then(|v| v.to_str().ok()).unwrap_or(""); + if !content_type.contains("json") { + return Err(format!("unexpected content type from actor: {}", content_type)); + } + let actor: serde_json::Value = resp.json().await + .map_err(|e| format!("invalid actor JSON: {}", e))?; + + let public_key_pem = actor.get("publicKey") + .and_then(|pk| pk.get("publicKeyPem")) + .and_then(|v| v.as_str()) + .ok_or("no publicKeyPem in actor")? + .to_string(); + let inbox_uri = actor.get("inbox").and_then(|v| v.as_str()).unwrap_or("").to_string(); + let domain = actor_uri.split("//").nth(1).and_then(|s| s.split('/').next()).unwrap_or(""); + + let dbi = state.db.lock().await; + let now = chrono::Utc::now().to_rfc3339(); + dbi.db.execute( + "INSERT OR REPLACE INTO ap_peers (domain, actor_uri, inbox_uri, public_pem, last_seen, created_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?5)", + rusqlite::params![domain, actor_uri, inbox_uri, public_key_pem, now], + ).ok(); + + Ok(public_key_pem) +} + +async fn fetch_actor_inbox(actor_uri: &str, state: &Arc) -> Result { + // check cache + let domain = actor_uri.split("//").nth(1).and_then(|s| s.split('/').next()).unwrap_or(""); + { + let dbi = state.db.lock().await; + if let Ok(inbox) = dbi.db.query_row( + "SELECT inbox_uri FROM ap_peers WHERE domain = ?1", [domain], |row| row.get::<_, String>(0), + ) { + return Ok(inbox); + } + } + + // fetch and cache + let _ = fetch_actor_key(actor_uri, state).await?; + let dbi = state.db.lock().await; + dbi.db.query_row( + "SELECT inbox_uri FROM ap_peers WHERE domain = ?1", [domain], |row| row.get::<_, String>(0), + ).map_err(|e| format!("no inbox: {}", e)) +} + +async fn handle_move(state: &Arc, activity: &serde_json::Value, actor_uri: &str) -> Result { + let target = activity.get("target").and_then(|v| v.as_str()).unwrap_or(""); + if target.is_empty() { + return Ok(accepted()); + } + + let old_domain = actor_uri.split("//").nth(1).and_then(|s| s.split('/').next()).unwrap_or(""); + let new_domain = target.split("//").nth(1).and_then(|s| s.split('/').next()).unwrap_or(""); + + if old_domain.is_empty() || new_domain.is_empty() || old_domain == new_domain { + return Ok(accepted()); + } + + let dbi = state.db.lock().await; + crate::db::log_admin_action(&dbi.db, "move_received", + Some(old_domain), Some(&format!("moving to {}", new_domain))); + + // if the actor was a peer, update to the new domain + let is_peer: bool = dbi.db.query_row( + "SELECT COUNT(*) > 0 FROM ap_peers WHERE domain = ?1", + [old_domain], |r| r.get(0), + ).unwrap_or(false); + + if is_peer { + let new_inbox = format!("https://{}/@cruciverb/inbox", new_domain); + crate::db::update_peer_domain(&dbi.db, old_domain, new_domain, target, &new_inbox); + tracing::info!("peer {} moved to {}", old_domain, new_domain); + } + + Ok(accepted()) +} + +async fn handle_announce(state: &Arc, activity: &serde_json::Value) -> Result { + // check for blocklist tag + let tags = activity.get("tag").and_then(|t| t.as_array()); + let is_blocklist = tags.map(|t| t.iter().any(|t| + t.get("name").and_then(|n| n.as_str()) == Some("#cruciverb-blocklist") + )).unwrap_or(false); + + if !is_blocklist { + // regular boost - check if it's boosting one of our outbox entries + let object_uri = match activity.get("object").and_then(|v| v.as_str()) { + Some(u) => u, + None => return Ok(accepted()), + }; + let actor_uri = activity.get("actor").and_then(|v| v.as_str()).unwrap_or(""); + if actor_uri.is_empty() { return Ok(accepted()); } + + let ap = match state.ap.as_ref() { + Some(ap) => ap, + None => return Ok(accepted()), + }; + + // check if the boosted URI belongs to our outbox + if object_uri.starts_with(&ap.actor_uri) || object_uri.starts_with(&format!("https://{}/ap/", ap.domain)) { + let dbi = state.db.lock().await; + // verify it exists in our outbox + let exists: bool = dbi.db.query_row( + "SELECT COUNT(*) > 0 FROM ap_outbox WHERE id = ?1 OR activity LIKE '%' || ?1 || '%'", + [object_uri], |r| r.get(0), + ).unwrap_or(false); + if exists { + crate::db::add_ap_boost(&dbi.db, object_uri, actor_uri); + tracing::info!("boost from {} on {}", actor_uri, object_uri); + } + } + + return Ok(accepted()); + } + + let attachments = activity.get("attachment").and_then(|a| a.as_array()); + let get_att = |name: &str| -> Option<&str> { + attachments.and_then(|a| a.iter() + .find(|att| att.get("name").and_then(|n| n.as_str()) == Some(name)) + .and_then(|att| att.get("value").and_then(|v| v.as_str()))) + }; + + let domain = match get_att("cruciverb:blocklist-domain") { Some(d) => d, None => return Ok(accepted()) }; + let action = match get_att("cruciverb:blocklist-action") { Some(a) => a, None => return Ok(accepted()) }; + + let actor = activity.get("actor").and_then(|v| v.as_str()).unwrap_or(""); + let origin = actor.split("//").nth(1).and_then(|s| s.split('/').next()).unwrap_or(""); + + let auto_apply = std::env::var("AP_SHARED_BLOCKLIST").as_deref() == Ok("true"); + + let dbi = state.db.lock().await; + if auto_apply { + let now = chrono::Utc::now().to_rfc3339(); + match action { + "block" => { + dbi.db.execute( + "INSERT OR IGNORE INTO ap_blocked (domain, reason, blocked_at) VALUES (?1, ?2, ?3)", + rusqlite::params![domain, format!("shared blocklist from {}", origin), now], + ).ok(); + tracing::info!("auto-blocked {} via shared blocklist from {}", domain, origin); + } + "unblock" => { + dbi.db.execute("DELETE FROM ap_blocked WHERE domain = ?1", [domain]).ok(); + tracing::info!("auto-unblocked {} via shared blocklist from {}", domain, origin); + } + _ => {} + } + crate::db::log_admin_action(&dbi.db, "shared_blocklist_applied", + Some(domain), Some(&format!("{} (from {})", action, origin))); + } else { + crate::db::log_admin_action(&dbi.db, "shared_blocklist_suggestion", + Some(domain), Some(&format!("{} (from {})", action, origin))); + } + + Ok(accepted()) +} + +async fn handle_leaderboard(state: &Arc, activity: &serde_json::Value, object: &serde_json::Value) -> Result { + let actor = activity.get("actor").and_then(|v| v.as_str()).unwrap_or(""); + let origin = actor.split("//").nth(1).and_then(|s| s.split('/').next()).unwrap_or(""); + if origin.is_empty() { return Ok(accepted()); } + + let attachments = object.get("attachment").and_then(|a| a.as_array()); + let get_att = |name: &str| -> Option<&str> { + attachments.and_then(|a| a.iter() + .find(|att| att.get("name").and_then(|n| n.as_str()) == Some(name)) + .and_then(|att| att.get("value").and_then(|v| v.as_str()))) + }; + + let date = match get_att("cruciverb:date") { Some(d) => d, None => return Ok(accepted()) }; + let preset = match get_att("cruciverb:preset") { Some(p) => p, None => return Ok(accepted()) }; + let entries_json = match get_att("cruciverb:entries") { Some(e) => e, None => return Ok(accepted()) }; + + // validate preset is one we recognize + let valid_presets = ["glyph", "rune", "scroll", "codex", "grimoire"]; + if !valid_presets.contains(&preset) { + tracing::warn!("invalid leaderboard preset from {}: {}", origin, preset); + return Ok(accepted()); + } + + // validate date format (YYYY-MM-DD) + if date.len() != 10 || date.chars().nth(4) != Some('-') || date.chars().nth(7) != Some('-') { + tracing::warn!("invalid leaderboard date from {}: {}", origin, date); + return Ok(accepted()); + } + + // reject dates more than 2 days old or in the future + if let Ok(parsed) = chrono::NaiveDate::parse_from_str(date, "%Y-%m-%d") { + let today = chrono::Utc::now().date_naive(); + let age = today.signed_duration_since(parsed).num_days(); + if age > 2 || age < 0 { + tracing::warn!("stale/future leaderboard date from {}: {} ({}d old)", origin, date, age); + return Ok(accepted()); + } + } else { + return Ok(accepted()); + } + + let entries: Vec = match serde_json::from_str(entries_json) { + Ok(e) => e, + Err(_) => return Ok(accepted()), + }; + + // minimum solve times by preset (seconds) - anything faster is physically impossible + let min_time: i64 = match preset { + "glyph" => 3, + "rune" => 5, + "scroll" => 15, + "codex" => 30, + "grimoire" => 60, + _ => 3, + }; + + // max 86400 seconds (24 hours) - anything longer is garbage + let max_time: i64 = 86400; + + // per-instance limit: max 10 entries per submission, 5 unique players per instance per puzzle + let dbi = state.db.lock().await; + + // check instance reputation + let reputation: f64 = dbi.db.query_row( + "SELECT reputation FROM ap_instance_reputation WHERE domain = ?1", + [origin], |row| row.get(0), + ).unwrap_or(100.0); + if reputation < 30.0 { + tracing::warn!("rejecting leaderboard from low-reputation instance {}: {}", origin, reputation); + return Ok(accepted()); + } + + // count existing entries from this instance for this puzzle + let existing_count: i64 = dbi.db.query_row( + "SELECT COUNT(*) FROM federated_leaderboard WHERE origin_domain = ?1 AND puzzle_date = ?2 AND preset = ?3", + rusqlite::params![origin, date, preset], |row| row.get(0), + ).unwrap_or(0); + if existing_count >= 10 { + tracing::debug!("instance {} already has {} entries for {} {}, skipping", origin, existing_count, date, preset); + return Ok(accepted()); + } + + let remaining_slots = (10 - existing_count).max(0) as usize; + let mut inserted = 0; + + for entry in entries.iter().take(10) { + if inserted >= remaining_slots { break; } + + let name = entry.get("name").and_then(|n| n.as_str()).unwrap_or("Anonymous"); + let time = entry.get("time").and_then(|t| t.as_i64()).unwrap_or(0); + + // validate time range + if time < min_time || time > max_time { + tracing::debug!("rejecting invalid time {}s from {} for {} {}", time, origin, preset, date); + continue; + } + + // sanitize name + let clean_name: String = name.chars().take(50).filter(|c| !c.is_control()).collect(); + if clean_name.is_empty() { continue; } + + // dedup: skip if same player from same instance already has an entry + let already: bool = dbi.db.query_row( + "SELECT COUNT(*) > 0 FROM federated_leaderboard WHERE origin_domain = ?1 AND puzzle_date = ?2 AND preset = ?3 AND player_name = ?4", + rusqlite::params![origin, date, preset, clean_name], |row| row.get(0), + ).unwrap_or(false); + if already { continue; } + + crate::db::insert_federated_leaderboard(&dbi.db, origin, date, preset, &clean_name, time); + inserted += 1; + } + + tracing::info!("accepted {}/{} leaderboard entries for {} {} from {} (rep: {})", inserted, entries.len(), preset, date, origin, reputation); + Ok(accepted()) +} + +async fn handle_puzzle_data(state: &Arc, activity: &serde_json::Value, object: &serde_json::Value) { + let actor = activity.get("actor").and_then(|v| v.as_str()).unwrap_or(""); + let origin = actor.split("//").nth(1).and_then(|s| s.split('/').next()).unwrap_or(""); + if origin.is_empty() { return; } + + let attachments = object.get("attachment").and_then(|a| a.as_array()); + let get_att = |name: &str| -> Option<&str> { + attachments.and_then(|a| a.iter() + .find(|att| att.get("name").and_then(|n| n.as_str()) == Some(name)) + .and_then(|att| att.get("value").and_then(|v| v.as_str()))) + }; + + let puzzle_id = match get_att("cruciverb:puzzle-id") { Some(id) => id, None => return }; + let title = get_att("cruciverb:title").unwrap_or("Untitled"); + let width: usize = get_att("cruciverb:width").and_then(|w| w.parse().ok()).unwrap_or(0); + let height: usize = get_att("cruciverb:height").and_then(|h| h.parse().ok()).unwrap_or(0); + + // decode puzzle data if present + let puzzle_data_b64 = get_att("cruciverb:puzzle-data"); + if let Some(encoded) = puzzle_data_b64 { + use base64::Engine; + if let Ok(decoded) = base64::engine::general_purpose::STANDARD.decode(encoded) { + if let Ok(json_str) = String::from_utf8(decoded) { + // validate it's real JSON + if serde_json::from_str::(&json_str).is_ok() { + let dbi = state.db.lock().await; + // use puzzle_id with origin prefix to avoid collisions + let local_id = format!("fed-{}-{}", origin, puzzle_id); + crate::db::insert_federated_puzzle( + &dbi.db, &local_id, title, width, height, + &json_str, &json_str, &json_str, origin, + ); + tracing::info!("stored federated puzzle '{}' from {}", title, origin); + } + } + } + } +} + +pub async fn check_is_cruciverb(domain: &str) -> bool { + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(5)) + .build() + .unwrap_or_default(); + + // check nodeinfo + let url = format!("https://{}/.well-known/nodeinfo", domain); + if let Ok(resp) = client.get(&url).send().await { + if let Ok(body) = resp.json::().await { + if let Some(links) = body.get("links").and_then(|l| l.as_array()) { + for link in links { + if let Some(href) = link.get("href").and_then(|h| h.as_str()) { + if let Ok(ni_resp) = client.get(href).send().await { + if let Ok(ni) = ni_resp.json::().await { + let name = ni.get("software") + .and_then(|s| s.get("name")) + .and_then(|n| n.as_str()) + .unwrap_or(""); + if name == "cruciverb" { + return true; + } + } + } + } + } + } + } + } + + false +}