harden federation: sanitize incoming AP content, fix error handling, tighten rate limits

crates/adapters/activitypub/src/composite_handler.rs

@@ -45,7 +45,7 @@ impl ApObjectHandler for CompositeObjectHandler {
         } else if is_watchlist {
             self.watchlist.on_create(ap_id, actor_url, object).await
         } else {
-            tracing::debug!(ap_id = %ap_id, "ignoring Create for unknown object type");
+            tracing::warn!(ap_id = %ap_id, "ignoring Create for unknown object type");
             Ok(())
         }
     }

crates/adapters/activitypub/src/federation_event_bridge.rs

@@ -45,8 +45,8 @@ impl k_ap::EventPublisher for FederationEventBridge {
                     .await
                     .map_err(|e| anyhow::anyhow!(e.to_string()))
             }
-            other => {
-                tracing::debug!("ignoring federation event: {:?}", other);
+            FederationEvent::DeliveryFailed { inbox, error, .. } => {
+                tracing::warn!(inbox = %inbox, error = %error, "federation delivery failed permanently");
                 Ok(())
             }
         }

crates/adapters/activitypub/src/review_handler.rs

@@ -77,13 +77,15 @@ impl ApObjectHandler for ReviewObjectHandler {
         _actor_url: &Url,
         object: serde_json::Value,
     ) -> anyhow::Result<()> {
-        let obj: ReviewObject = match serde_json::from_value(object) {
+        let mut obj: ReviewObject = match serde_json::from_value(object) {
             Ok(o) => o,
             Err(e) => {
-                tracing::debug!("ignoring unrecognized Create object: {}", e);
+                tracing::warn!("ignoring unrecognized Create object: {}", e);
                 return Ok(());
             }
         };
+        obj.movie_title = ammonia::clean(&obj.movie_title);
+        obj.comment = obj.comment.map(|c| ammonia::clean(&c));
 
         let actor_url_str = obj.attributed_to.to_string();
         let review_id = ReviewId::generate();
@@ -130,13 +132,15 @@ impl ApObjectHandler for ReviewObjectHandler {
         actor_url: &Url,
         object: serde_json::Value,
     ) -> anyhow::Result<()> {
-        let obj: ReviewObject = match serde_json::from_value(object) {
+        let mut obj: ReviewObject = match serde_json::from_value(object) {
             Ok(o) => o,
             Err(_) => {
-                tracing::debug!(actor = %actor_url, "ignoring non-review Update activity");
+                tracing::warn!(actor = %actor_url, "ignoring non-review Update activity");
                 return Ok(());
             }
         };
+        obj.movie_title = ammonia::clean(&obj.movie_title);
+        obj.comment = obj.comment.map(|c| ammonia::clean(&c));
 
         if obj.attributed_to != *actor_url {
             anyhow::bail!("update actor does not match object attributed_to");

crates/adapters/activitypub/src/watchlist_handler.rs

@@ -24,7 +24,14 @@ impl ApObjectHandler for WatchlistObjectHandler {
         actor_url: &Url,
         object: serde_json::Value,
     ) -> anyhow::Result<()> {
-        let obj: WatchlistObject = serde_json::from_value(object)?;
+        let mut obj: WatchlistObject = match serde_json::from_value(object) {
+            Ok(o) => o,
+            Err(e) => {
+                tracing::warn!(ap_id = %ap_id, "ignoring malformed watchlist Create: {}", e);
+                return Ok(());
+            }
+        };
+        obj.movie_title = ammonia::clean(&obj.movie_title);
         let added_at = obj.published;
         let entry = RemoteWatchlistEntry {
             ap_id: ap_id.as_str().to_string(),