import base64 import uuid 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.utils import timezone from core.ld import canonicalise 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" def upload_namer(prefix, instance, filename): """ Names uploaded images etc. """ now = timezone.now() filename = base64.b32encode(uuid.uuid4().bytes).decode("ascii") return f"{prefix}/{now.year}/{now.month}/{now.day}/{filename}" 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}/" ### 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 @") 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 KeyError(f"No identity found matching {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 (boutbound) ### 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": self.urls.view_nice, } if self.name: response["name"] = self.name if self.summary: response["summary"] = self.summary return response ### 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"] 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.save()