from django.contrib import auth from django.contrib.auth.models import AbstractUser from django.db import models, transaction from django.db.utils import IntegrityError from nacl.encoding import HexEncoder from nacl.exceptions import BadSignatureError from nacl.signing import SigningKey, VerifyKey class KnownIdentity(models.Model): username = models.CharField(max_length=255) domain = models.CharField(max_length=255) public_key = models.CharField(max_length=255) friends = models.ManyToManyField('self', symmetrical=True) class Meta: unique_together = ('username', 'domain') indexes = [ models.Index(fields=['username', 'domain'], name='identity_idx'), ] def __str__(self): return f"{self.username}@{self.domain}" def is_authenticated(self): return True def friends_or_self(self): return ToolshedUser.objects.filter(public_identity__friends=self) | ToolshedUser.objects.filter( public_identity=self) def verify(self, message, signature): if len(signature) != 128 or type(signature) != str: raise TypeError('Signature must be 128 characters long and a string') if type(message) != str: raise TypeError('Message must be a string') try: VerifyKey(bytes.fromhex(self.public_key)).verify(message.encode('utf-8'), bytes.fromhex(signature)) return True except BadSignatureError: return False class ToolshedUserManager(auth.models.BaseUserManager): def create_user(self, username, email, password, **extra_fields): domain = extra_fields.pop('domain', 'localhost') private_key_hex = extra_fields.pop('private_key', None) if private_key_hex and type(private_key_hex) != str: raise TypeError('Private key must be a string or no private key must be provided') if private_key_hex and len(private_key_hex) != 64: raise ValueError('Private key must be 64 characters long or no private key must be provided') if private_key_hex and not all(c in '0123456789abcdef' for c in private_key_hex): raise ValueError('Private key must be a hexadecimal string or no private key must be provided') private_key = SigningKey(bytes.fromhex(private_key_hex)) if private_key_hex else SigningKey.generate() public_key = SigningKey(private_key.encode()).verify_key extra_fields['private_key'] = private_key.encode(encoder=HexEncoder).decode('utf-8') try: with transaction.atomic(): extra_fields['public_identity'] = identity = KnownIdentity.objects.get_or_create( username=username, domain=domain, public_key=public_key.encode(encoder=HexEncoder).decode('utf-8'))[0] try: with transaction.atomic(): user = super().create(username=username, email=email, password=password, domain=domain, **extra_fields) user.set_password(password) user.save() except IntegrityError: identity.delete() raise ValueError('Username or email already exists') else: return user except IntegrityError: raise ValueError('Username already exists') def create_superuser(self, username, email, password, **extra_fields): user = self.create_user(username, email, password, **extra_fields) user.is_staff = True user.is_superuser = True user.save() return user class ToolshedUser(AbstractUser): email = models.EmailField(unique=True) domain = models.CharField(max_length=255, default='localhost') private_key = models.CharField(max_length=255) public_identity = models.ForeignKey(KnownIdentity, on_delete=models.CASCADE, related_name='user') profile_picture = models.ForeignKey('files.File', on_delete=models.SET_NULL, null=True, blank=True, related_name='profile_picture_users') objects = ToolshedUserManager() class Meta: unique_together = [('username', 'domain'), ['email']] indexes = [ models.Index(fields=['username', 'domain'], name='user_idx'), ] def __str__(self): return f"{self.username}@{self.domain}" @property def friends(self): return self.public_identity.friends def sign(self, message): if type(message) != str: raise TypeError('Message must be a string') private_key = SigningKey(self.private_key.encode(), encoder=HexEncoder) return private_key.sign(message.encode('utf-8'), encoder=HexEncoder).signature.decode('utf-8') def public_key(self): return self.public_identity.public_key class AccountPreference(models.Model): """A single account-level (server-synced, cross-device) user preference, stored as a key/value pair. Device-level preferences are intentionally *not* stored here - they stay in the browser's local storage since they describe the device, not the account. """ user = models.ForeignKey(ToolshedUser, on_delete=models.CASCADE, related_name='preferences') key = models.CharField(max_length=255) value = models.JSONField() class Meta: unique_together = ('user', 'key') def __str__(self): return f"{self.user}: {self.key}" class FriendRequestOutgoing(models.Model): secret = models.CharField(max_length=255) befriender_user = models.ForeignKey(ToolshedUser, on_delete=models.CASCADE, related_name='friend_requests_outgoing') befriendee_username = models.CharField(max_length=255) befriendee_domain = models.CharField(max_length=255) class FriendRequestIncoming(models.Model): secret = models.CharField(max_length=255) befriender_username = models.CharField(max_length=255) befriender_domain = models.CharField(max_length=255) befriender_public_key = models.CharField(max_length=255) befriendee_user = models.ForeignKey(ToolshedUser, on_delete=models.CASCADE, related_name='friend_requests_incoming') created_at = models.DateTimeField(auto_now_add=True) class Group(models.Model): name = models.CharField(max_length=255) domain = models.CharField(max_length=255) members = models.ManyToManyField(KnownIdentity, related_name='member_of_groups') class Meta: unique_together = ('name', 'domain') indexes = [ models.Index(fields=['name', 'domain'], name='group_idx'), ] def __str__(self): return f"#{self.name}@{self.domain}" def is_member(self, identity): return self.members.filter(pk=identity.pk).exists() class GroupInvite(models.Model): """A pending invite tracked on the group's own home backend, checked when the invitee's accept request arrives (see GroupInviteIncoming for the mirror record on the invitee's own backend, and docs/design-in-progress/groups-mvp.md for the full invite/accept dance).""" secret = models.CharField(max_length=255) group = models.ForeignKey(Group, on_delete=models.CASCADE, related_name='invites') invitee_username = models.CharField(max_length=255) invitee_domain = models.CharField(max_length=255) created_at = models.DateTimeField(auto_now_add=True) class GroupInviteIncoming(models.Model): secret = models.CharField(max_length=255) group_name = models.CharField(max_length=255) group_domain = models.CharField(max_length=255) inviter_username = models.CharField(max_length=255) inviter_domain = models.CharField(max_length=255) invitee_user = models.ForeignKey(ToolshedUser, on_delete=models.CASCADE, related_name='group_invites_incoming') created_at = models.DateTimeField(auto_now_add=True)