Refactoring inbox processing to smaller tasks

activities/models/post.py

@@ -884,7 +884,7 @@ def by_ap(cls, data, create=False, update=False, fetch_author=False) -> "Post":
  except IntegrityError:
  # despite previous checks, a parallel thread managed
  # to create the same object already
- post = cls.by_object_uri(object_uri=data["id"])
+ raise TryAgainLater()
  else:
  raise cls.DoesNotExist(f"No post with ID {data['id']}", data)
  if update or created:
@@ -1014,7 +1014,7 @@ def by_object_uri(cls, object_uri, fetch=False) -> "Post":
  response = SystemActor().signed_request(
  method="get", uri=object_uri
  )
- except (httpx.HTTPError, ssl.SSLCertVerificationError):
+ except (httpx.HTTPError, ssl.SSLCertVerificationError, ValueError):
  raise cls.DoesNotExist(f"Could not fetch {object_uri}")
  if response.status_code in [404, 410]:
  raise cls.DoesNotExist(f"No post at {object_uri}")
@@ -1072,7 +1072,7 @@ def handle_create_ap(cls, data):
  if data["actor"] != data["object"]["attributedTo"]:
  raise ValueError("Create actor does not match its Post object", data)
  # Create it, stator will fan it out locally
- cls.by_ap(data["object"], create=True, update=True)
+ cls.by_ap(data["object"], create=True, update=True, fetch_author=True)
 
  @classmethod
  def handle_update_ap(cls, data):

core/signatures.py

@@ -295,6 +295,8 @@ def verify_signature(cls, document: dict, public_key: str) -> None:
  Verifies a document
  """
  try:
+ # causing side effects to the original document is bad form
+ document = document.copy()
  # Strip out the signature from the incoming document
  signature = document.pop("signature")
  # Create the options document
@@ -322,7 +324,7 @@ def verify_signature(cls, document: dict, public_key: str) -> None:
  hashes.SHA256(),
  )
  except InvalidSignature:
- raise VerificationError("Signature mismatch")
+ raise VerificationError("LDSignature mismatch")
 
  @classmethod
  def create_signature(

tests/core/test_signatures.py

@@ -12,6 +12,7 @@ def test_sign_ld(keypair):
  """
  # Create the signature
  document = {
+ "@context": ["https://www.w3.org/ns/activitystreams"],
  "id": "https://example.com/test-create",
  "type": "Create",
  "actor": "https://example.com/test-actor",
@@ -38,20 +39,23 @@ def test_verifying_ld(keypair):
  Tests verifying JSON-LD signatures from a known-good document
  """
  document = {
+ "@context": ["https://www.w3.org/ns/activitystreams"],
  "id": "https://example.com/test-create",
  "type": "Create",
  "actor": "https://example.com/test-actor",
  "object": {"id": "https://example.com/test-object", "type": "Note"},
  "signature": {
  "@context": "https://w3id.org/identity/v1",
  "creator": "https://example.com/test-actor#test-key",
- "created": "2022-11-12T21:41:47Z",
- "signatureValue": "nTHfkHqG4hegfnjpHucXtXDLDaIKi2Duk+NeCzqTtkjf4NneXsofbZY2tGew4uAooEe1UeM23PIyjWYnR16KwcD4YY8nMj8L3xY2czwQPScMM9n+KhSHzkWfX+iI4FWKbjpPI8M53EtTRJU+1qEjjmGUx03Ip0vfvT5821etIgvY4wLNhg3y7R8fevnNux+BeytcEV6gM4awJJ6RK0xrWGLyTgDNon5V5aNUjwcV/UVPy9UAQi1KYWtA74/F0Y4oPzL5CTudPpyiViyVHZQaal4r+ExzgSvGztqKxQeT1ya6gLXxbm1YQ+8UiGVSS8zoGhMFDEZWVsRPv7e0jm5wfA==",
+ "created": "2023-10-25T08:08:47.702Z",
+ "signatureValue": "ajg4ukZzCtBWjflO1u6MlTc4tBVO6MsqzBr/L+kO5VI2ucutFaUdDa/Kx4W12ZCm9oYvTyMQMnoeELx5BifslRWEeMmo1wWMPXmg2/BMKgm8Spt+Zanq68uTlYGyKvuw1Q0FyNq84N2PbRZRXu2Yhlj2KnAVTRtKrsfEiCg3yNfVQ7lbUpDtlXvXLAq2yBN8H/BnZDoynjaDlafFW9Noq8025q1K/lz5jNzBEL22CSrKsD2qYWq1TK3s3h6SJ+j3J+5s0Ni3F/TH7W/5VeGBpzx4z6MSjmn7aHAS3JNCnAWDW9Rf6yKLg2y5htj6FpexiGcoEjO3VqjLoIP4f/115Q==",
  "type": "RsaSignature2017",
  },
  }
  # Ensure it verifies with correct data
  LDSignature.verify_signature(document, keypair["public_key"])
+ # signature should remain in document if it was valid
+ assert "signature" in document
  # Mutate it slightly and ensure it does not verify
  with pytest.raises(VerificationError):
  document["actor"] = "https://example.com/evil-actor"

tests/users/models/test_identity.py

@@ -199,6 +199,7 @@ def test_fetch_actor(httpx_mock, config_system):
  identity.featured_collection_uri
  == "https://example.com/test-actor/collections/featured/"
  )
+ identity.fetch_pinned_post_uris(identity.featured_collection_uri)
  assert identity.icon_uri == "https://example.com/icon.jpg"
  assert identity.image_uri == "https://example.com/image.jpg"
  assert identity.summary == "<p>A test user</p>"

users/models/identity.py

@@ -34,6 +34,7 @@
 from stator.exceptions import TryAgainLater
 from stator.models import State, StateField, StateGraph, StatorModel
 from users.models.domain import Domain
+from users.models.inbox_message import InboxMessage
 from users.models.system_actor import SystemActor
 
 
@@ -743,7 +744,7 @@ def fetch_webfinger(cls, handle: str) -> tuple[str | None, str | None]:
  except (httpx.HTTPError, ssl.SSLCertVerificationError) as ex:
  response = getattr(ex, "response", None)
  if isinstance(ex, httpx.TimeoutException) or (
- response and response.status_code in [408, 504]
+ response and response.status_code in [408, 429, 504]
  ):
  raise TryAgainLater() from ex
  elif (
@@ -800,7 +801,7 @@ def fetch_pinned_post_uris(cls, uri: str) -> list[str]:
  except (httpx.HTTPError, ssl.SSLCertVerificationError) as ex:
  response = getattr(ex, "response", None)
  if isinstance(ex, httpx.TimeoutException) or (
- response and response.status_code in [408, 504]
+ response and response.status_code in [408, 429, 504]
  ):
  raise TryAgainLater() from ex
  elif (
@@ -847,7 +848,6 @@ def fetch_actor(self) -> bool:
  webfinger if it's available.
  """
  from activities.models import Emoji
- from users.services import IdentityService
 
  if self.local:
  raise ValueError("Cannot fetch local identities")
@@ -866,30 +866,24 @@ def fetch_actor(self) -> bool:
  return False
  status_code = response.status_code
  if status_code >= 400:
- if status_code in [408, 504]:
+ if status_code in [408, 429, 504]:
  raise TryAgainLater()
  if status_code == 410 and self.pk:
  # Their account got deleted, so let's do the same.
  Identity.objects.filter(pk=self.pk).delete()
  if status_code < 500 and status_code not in [401, 403, 404, 406, 410]:
  logging.info(
- f"Client error fetching actor at {self.actor_uri}: {status_code}",
- extra={
- "identity": self.pk,
- "domain": self.domain_id,
- "content": response.content,
- },
+ "Client error fetching actor: %d %s", status_code, self.actor_uri
  )
  return False
  try:
  document = canonicalise(response.json(), include_security=True)
  except ValueError:
  # servers with empty or invalid responses are inevitable
  logging.info(
- f"Invalid response fetching actor at {self.actor_uri}",
+ "Invalid response fetching actor %s",
+ self.actor_uri,
  extra={
- "identity": self.pk,
- "domain": self.domain_id,
  "content": response.content,
  },
  )
@@ -948,7 +942,16 @@ def fetch_actor(self) -> bool:
  self.domain = Domain.get_remote_domain(webfinger_domain)
  except TryAgainLater:
  # continue with original domain when webfinger times out
+ logging.info("WebFinger timed out: %s", self.actor_uri)
  pass
+ except ValueError as exc:
+ logging.info(
+ "Can't parse WebFinger: %s %s",
+ exc.args[0],
+ self.actor_uri,
+ exc_info=exc,
+ )
+ return False
  # Emojis (we need the domain so we do them here)
  for tag in get_list(document, "tag"):
  if tag["type"].lower() in ["toot:emoji", "emoji"]:
@@ -973,11 +976,14 @@ def fetch_actor(self) -> bool:
  with transaction.atomic():
  self.save()
 
- # Fetch pinned posts after identity has been fetched and saved
+ # Fetch pinned posts in a followup task
  if self.featured_collection_uri:
- featured = self.fetch_pinned_post_uris(self.featured_collection_uri)
- service = IdentityService(self)
- service.sync_pins(featured)
+ InboxMessage.create_internal(
+ {
+ "type": "SyncPins",
+ "identity": self.pk,
+ }
+ )
 
  return True
 

users/models/inbox_message.py

@@ -141,6 +141,10 @@ def handle_received(cls, instance: "InboxMessage"):
  IdentityService.handle_internal_add_follow(
  instance.message["object"]
  )
+ case "syncpins":
+ IdentityService.handle_internal_sync_pins(
+ instance.message["object"]
+ )
  case unknown:
  return cls.errored
  case unknown:

users/services/identity.py

@@ -1,3 +1,6 @@
+import logging
+
+from django.core.exceptions import MultipleObjectsReturned
 from django.db import models, transaction
 from django.template.defaultfilters import linebreaks_filter
 
@@ -222,12 +225,14 @@ def sync_pins(self, object_uris):
  post=post,
  state__in=PostInteractionStates.group_active(),
  )
+ except MultipleObjectsReturned as exc:
+ logging.exception("%s on %s", exc, object_uri)
+ pass
  except Post.DoesNotExist:
  # ignore 404s...
  pass
  except TryAgainLater:
- # when fetching a post -> author -> post we can
- # get into a state. Ignore this round.
+ # don't wait for it now, it'll be synced on next refresh
  pass
  for removed in PostInteraction.objects.filter(
  type=PostInteraction.Types.pin,
@@ -319,3 +324,22 @@ def handle_internal_add_follow(cls, payload):
  raise ValueError(f"Cannot find identity to follow: {target_identity}")
  # Follow!
  self.follow(target_identity=target_identity, boosts=payload.get("boosts", True))
+
+ @classmethod
+ def handle_internal_sync_pins(cls, payload):
+ """
+ Handles an inbox message saying we need to sync featured posts
+
+ Message format:
+ {
+ "type": "SyncPins",
+ "identity": "90310938129083",
+ }
+ """
+ # Retrieve ourselves
+ actor = Identity.objects.get(pk=payload["identity"])
+ self = cls(actor)
+ # Get the remote end (may need a fetch)
+ if actor.featured_collection_uri:
+ featured = actor.fetch_pinned_post_uris(actor.featured_collection_uri)
+ self.sync_pins(featured)

users/views/activitypub.py

@@ -1,5 +1,6 @@
 import json
 import logging
+from urllib.parse import urldefrag, urlparse
 
 from django.conf import settings
 from django.http import Http404, HttpResponse, HttpResponseBadRequest, JsonResponse
@@ -20,9 +21,9 @@
  VerificationFormatError,
 )
 from core.views import StaticContentView
-from stator.exceptions import TryAgainLater
 from takahe import __version__
 from users.models import Identity, InboxMessage, SystemActor
+from users.models.domain import Domain
 from users.shortcuts import by_handle_or_404
 
 
@@ -153,50 +154,91 @@ def post(self, request, handle=None):
  # We don't have an Identity record for the user. No-op
  return HttpResponse(status=202)
 
- if not identity.public_key:
- # See if we can fetch it right now
- try:
- identity.fetch_actor()
- except TryAgainLater:
- logging.warning(
- f"Inbox error: timed out fetching actor {document['actor']}"
- )
- return HttpResponse(status=504)
-
- if not identity.public_key:
- logging.warning(f"Inbox error: cannot fetch actor {document['actor']}")
- return HttpResponseBadRequest("Cannot retrieve actor")
-
- # See if it's from a blocked user or domain
- if identity.blocked or identity.domain.recursively_blocked():
+ # See if it's from a blocked user or domain - without calling
+ # fetch_actor, which would fetch data from potentially bad actor
+ domain = identity.domain
+ if not domain:
+ actor_url_parts = urlparse(document["actor"])
+ domain = Domain.get_remote_domain(actor_url_parts.hostname)
+ if identity.blocked or domain.recursively_blocked():
  # I love to lie! Throw it away!
- logging.warning(f"Inbox: Discarded message from {identity.actor_uri}")
+ logging.info(
+ "Inbox: Discarded message from blocked %s %s",
+ "domain" if domain.recursively_blocked() else "user",
+ identity.actor_uri,
+ )
  return HttpResponse(status=202)
 
- # If there's a "signature" payload, verify against that
- if "signature" in document:
+ # authenticate HTTP signature first, if one is present and the actor
+ # is already known to us. An invalid signature is an error and message
+ # should be discarded. NOTE: for previously unknown actors, we
+ # don't have their public key yet!
+ if "signature" in request:
  try:
- LDSignature.verify_signature(document, identity.public_key)
+ if identity.public_key:
+ HttpSignature.verify_request(
+ request,
+ identity.public_key,
+ )
+ logging.debug(
+ "Inbox: %s from %s has good HTTP signature",
+ document["type"],
+ identity,
+ )
+ else:
+ logging.info(
+ "Inbox: New actor, no key available: %s",
+ document["actor"],
+ )
  except VerificationFormatError as e:
- logging.warning(f"Inbox error: Bad LD signature format: {e.args[0]}")
+ logging.warning("Inbox error: Bad HTTP signature format: %s", e.args[0])
  return HttpResponseBadRequest(e.args[0])
  except VerificationError:
- logging.warning("Inbox error: Bad LD signature")
+ logging.warning("Inbox error: Bad HTTP signature from %s", identity)
  return HttpResponseUnauthorized("Bad signature")
 
- # Otherwise, verify against the header (assuming it's the same actor)
- else:
+ # Mastodon advices not implementing LD Signatures, but
+ # they're widely deployed today. Validate it if one exists.
+ # https://docs.joinmastodon.org/spec/security/#ld
+ if "signature" in document:
  try:
- HttpSignature.verify_request(
- request,
- identity.public_key,
+ # signatures are identified by the signature block
+ creator = urldefrag(document["signature"]["creator"]).url
+ creator_identity = Identity.by_actor_uri(
+ creator, create=True, transient=True
  )
+ if not creator_identity.public_key:
+ logging.info("Inbox: New actor, no key available: %s", creator)
+ # if we can't verify it, we don't keep it
+ document.pop("signature")
+ else:
+ LDSignature.verify_signature(document, creator_identity.public_key)
+ logging.debug(
+ "Inbox: %s from %s has good LD signature",
+ document["type"],
+ creator_identity,
+ )
  except VerificationFormatError as e:
- logging.warning(f"Inbox error: Bad HTTP signature format: {e.args[0]}")
+ logging.warning("Inbox error: Bad LD signature format: %s", e.args[0])
  return HttpResponseBadRequest(e.args[0])
  except VerificationError:
- logging.warning("Inbox error: Bad HTTP signature")
- return HttpResponseUnauthorized("Bad signature")
+ # An invalid LD Signature might also indicate nothing but
+ # a syntactical difference between implementations.
+ # Strip it out if we can't verify it.
+ if "signature" in document:
+ document.pop("signature")
+ logging.info(
+ "Inbox: Stripping invalid LD signature from %s %s",
+ creator_identity,
+ document["id"],
+ )
+
+ if not ("signature" in request or "signature" in document):
+ logging.debug(
+ "Inbox: %s from %s is unauthenticated. That's OK.",
+ document["type"],
+ identity,
+ )
 
  # Don't allow injection of internal messages
  if document["type"].startswith("__"):