from functools import partial from typing import Optional, Tuple from urllib.parse import urlparse import httpx import urlman from asgiref.sync import async_to_sync, sync_to_async from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import rsa from django.db import models from django.template.defaultfilters import linebreaks_filter from django.templatetags.static import static from django.utils import timezone from core.html import sanitize_post from core.ld import canonicalise, media_type_from_filename from core.uploads import upload_namer from stator.models import State, StateField, StateGraph, StatorModel from users.models.domain import Domain class IdentityStates(StateGraph): outdated = State(try_interval=3600) updated = State() outdated.transitions_to(updated) @classmethod async def handle_outdated(cls, identity: "Identity"): # Local identities never need fetching if identity.local: return "updated" # Run the actor fetch and progress to updated if it succeeds if await identity.fetch_actor(): return "updated" class Identity(StatorModel): """ Represents both local and remote Fediverse identities (actors) """ # The Actor URI is essentially also a PK - we keep the default numeric # one around as well for making nice URLs etc. actor_uri = models.CharField(max_length=500, unique=True) state = StateField(IdentityStates) local = models.BooleanField() users = models.ManyToManyField( "users.User", related_name="identities", blank=True, ) username = models.CharField(max_length=500, blank=True, null=True) # Must be a display domain if present domain = models.ForeignKey( "users.Domain", blank=True, null=True, on_delete=models.PROTECT, related_name="identities", ) name = models.CharField(max_length=500, blank=True, null=True) summary = models.TextField(blank=True, null=True) manually_approves_followers = models.BooleanField(blank=True, null=True) profile_uri = models.CharField(max_length=500, blank=True, null=True) inbox_uri = models.CharField(max_length=500, blank=True, null=True) outbox_uri = models.CharField(max_length=500, blank=True, null=True) icon_uri = models.CharField(max_length=500, blank=True, null=True) image_uri = models.CharField(max_length=500, blank=True, null=True) icon = models.ImageField( upload_to=partial(upload_namer, "profile_images"), blank=True, null=True ) image = models.ImageField( upload_to=partial(upload_namer, "background_images"), blank=True, null=True ) private_key = models.TextField(null=True, blank=True) public_key = models.TextField(null=True, blank=True) public_key_id = models.TextField(null=True, blank=True) created = models.DateTimeField(auto_now_add=True) updated = models.DateTimeField(auto_now=True) fetched = models.DateTimeField(null=True, blank=True) deleted = models.DateTimeField(null=True, blank=True) ### Model attributes ### class Meta: verbose_name_plural = "identities" unique_together = [("username", "domain")] class urls(urlman.Urls): view_nice = "{self._nice_view_url}" view = "/@{self.username}@{self.domain_id}/" action = "{view}action/" activate = "{view}activate/" def get_scheme(self, url): return "https" def get_hostname(self, url): return self.instance.domain.uri_domain def __str__(self): if self.username and self.domain_id: return self.handle return self.actor_uri def _nice_view_url(self): """ Returns the "nice" user URL if they're local, otherwise our general one """ if self.local: return f"https://{self.domain.uri_domain}/@{self.username}/" else: return f"/@{self.username}@{self.domain_id}/" def local_icon_url(self): """ Returns an icon for us, with fallbacks to a placeholder """ if self.icon: return self.icon.url elif self.icon_uri: return self.icon_uri else: return static("img/unknown-icon-128.png") def local_image_url(self): """ Returns a background image for us, returning None if there isn't one """ if self.image: return self.image.url elif self.image_uri: return self.image_uri @property def safe_summary(self): return sanitize_post(self.summary) ### Alternate constructors/fetchers ### @classmethod def by_username_and_domain(cls, username, domain, fetch=False, local=False): if username.startswith("@"): raise ValueError("Username must not start with @") username = username.lower() try: if local: return cls.objects.get(username=username, domain_id=domain, local=True) else: return cls.objects.get(username=username, domain_id=domain) except cls.DoesNotExist: if fetch and not local: actor_uri, handle = async_to_sync(cls.fetch_webfinger)( f"{username}@{domain}" ) username, domain = handle.split("@") domain = Domain.get_remote_domain(domain) return cls.objects.create( actor_uri=actor_uri, username=username, domain_id=domain, local=False, ) return None @classmethod def by_actor_uri(cls, uri, create=False) -> "Identity": try: return cls.objects.get(actor_uri=uri) except cls.DoesNotExist: if create: return cls.objects.create(actor_uri=uri, local=False) else: raise cls.DoesNotExist(f"No identity found with actor_uri {uri}") ### Dynamic properties ### @property def name_or_handle(self): return self.name or self.handle @property def handle(self): if self.domain_id: return f"{self.username}@{self.domain_id}" return f"{self.username}@UNKNOWN-DOMAIN" @property def data_age(self) -> float: """ How old our copy of this data is, in seconds """ if self.local: return 0 if self.fetched is None: return 10000000000 return (timezone.now() - self.fetched).total_seconds() @property def outdated(self) -> bool: # TODO: Setting return self.data_age > 60 * 24 * 24 ### ActivityPub (outbound) ### def to_ap(self): response = { "id": self.actor_uri, "type": "Person", "inbox": self.actor_uri + "inbox/", "preferredUsername": self.username, "publicKey": { "id": self.public_key_id, "owner": self.actor_uri, "publicKeyPem": self.public_key, }, "published": self.created.strftime("%Y-%m-%dT%H:%M:%SZ"), "url": str(self.urls.view_nice), } if self.name: response["name"] = self.name if self.summary: response["summary"] = str(linebreaks_filter(self.summary)) if self.icon: response["icon"] = { "type": "Image", "mediaType": media_type_from_filename(self.icon.name), "url": self.icon.url, } if self.image: response["image"] = { "type": "Image", "mediaType": media_type_from_filename(self.image.name), "url": self.image.url, } return response ### ActivityPub (inbound) ### @classmethod def handle_update_ap(cls, data): """ Takes an incoming update.person message and just forces us to add it to our fetch queue (don't want to bother with two load paths right now) """ # Find by actor try: actor = cls.by_actor_uri(data["actor"]) actor.transition_perform(IdentityStates.outdated) except cls.DoesNotExist: pass ### Actor/Webfinger fetching ### @classmethod async def fetch_webfinger(cls, handle: str) -> Tuple[Optional[str], Optional[str]]: """ Given a username@domain handle, returns a tuple of (actor uri, canonical handle) or None, None if it does not resolve. """ domain = handle.split("@")[1] async with httpx.AsyncClient() as client: response = await client.get( f"https://{domain}/.well-known/webfinger?resource=acct:{handle}", headers={"Accept": "application/json"}, follow_redirects=True, ) if response.status_code >= 400: return None, None data = response.json() if data["subject"].startswith("acct:"): data["subject"] = data["subject"][5:] for link in data["links"]: if ( link.get("type") == "application/activity+json" and link.get("rel") == "self" ): return link["href"], data["subject"] return None, None async def fetch_actor(self) -> bool: """ Fetches the user's actor information, as well as their domain from webfinger if it's available. """ if self.local: raise ValueError("Cannot fetch local identities") async with httpx.AsyncClient() as client: response = await client.get( self.actor_uri, headers={"Accept": "application/json"}, follow_redirects=True, ) if response.status_code >= 400: return False document = canonicalise(response.json(), include_security=True) self.name = document.get("name") self.profile_uri = document.get("url") self.inbox_uri = document.get("inbox") self.outbox_uri = document.get("outbox") self.summary = document.get("summary") self.username = document.get("preferredUsername") if self.username and "@value" in self.username: self.username = self.username["@value"] if self.username: self.username = self.username.lower() self.manually_approves_followers = document.get( "as:manuallyApprovesFollowers" ) self.public_key = document.get("publicKey", {}).get("publicKeyPem") self.public_key_id = document.get("publicKey", {}).get("id") self.icon_uri = document.get("icon", {}).get("url") self.image_uri = document.get("image", {}).get("url") # Now go do webfinger with that info to see if we can get a canonical domain actor_url_parts = urlparse(self.actor_uri) get_domain = sync_to_async(Domain.get_remote_domain) if self.username: webfinger_actor, webfinger_handle = await self.fetch_webfinger( f"{self.username}@{actor_url_parts.hostname}" ) if webfinger_handle: webfinger_username, webfinger_domain = webfinger_handle.split("@") self.username = webfinger_username self.domain = await get_domain(webfinger_domain) else: self.domain = await get_domain(actor_url_parts.hostname) else: self.domain = await get_domain(actor_url_parts.hostname) self.fetched = timezone.now() await sync_to_async(self.save)() return True ### Cryptography ### def generate_keypair(self): if not self.local: raise ValueError("Cannot generate keypair for remote user") private_key = rsa.generate_private_key( public_exponent=65537, key_size=2048, ) self.private_key = private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption(), ).decode("ascii") self.public_key = ( private_key.public_key() .public_bytes( encoding=serialization.Encoding.PEM, format=serialization.PublicFormat.SubjectPublicKeyInfo, ) .decode("ascii") ) self.public_key_id = self.actor_uri + "#main-key" self.save()