desec-stack/api/desecapi/models/tokens.py

from __future__ import annotations

import ipaddress
import secrets
import uuid
from datetime import timedelta

import pgtrigger
import rest_framework.authtoken.models
from django.contrib.auth.hashers import make_password
from django.contrib.postgres.fields import ArrayField
from django.core import validators
from django.core.exceptions import ValidationError
from django.db import models, transaction
from django.db.models import F, Q
from django.utils import timezone
from django_prometheus.models import ExportModelOperationsMixin
from netfields import CidrAddressField, NetManager


# No 0OIl characters, non-alphanumeric only (select by double-click no line-break)
# https://github.com/bitcoin/bitcoin/blob/master/src/base58.h
ALPHABET = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz"


class Token(ExportModelOperationsMixin("Token"), rest_framework.authtoken.models.Token):
    @staticmethod
    def _allowed_subnets_default():
        return [ipaddress.IPv4Network("0.0.0.0/0"), ipaddress.IPv6Network("::/0")]

    _validators = [
        validators.MinValueValidator(timedelta(0)),
        validators.MaxValueValidator(timedelta(days=365 * 1000)),
    ]

    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
    key = models.CharField("Key", max_length=128, db_index=True, unique=True)
    user = models.ForeignKey("User", on_delete=models.CASCADE)
    name = models.CharField("Name", blank=True, max_length=64)
    last_used = models.DateTimeField(null=True, blank=True)
    mfa = models.BooleanField(default=None, null=True)
    perm_manage_tokens = models.BooleanField(default=False)
    allowed_subnets = ArrayField(
        CidrAddressField(), default=_allowed_subnets_default.__func__
    )
    max_age = models.DurationField(null=True, default=None, validators=_validators)
    max_unused_period = models.DurationField(
        null=True, default=None, validators=_validators
    )
    domain_policies = models.ManyToManyField("Domain", through="TokenDomainPolicy")

    plain = None
    objects = NetManager()

    class Meta:
        constraints = [
            models.UniqueConstraint(fields=["id", "user"], name="unique_id_user")
        ]

    @property
    def is_valid(self):
        now = timezone.now()

        # Check max age
        try:
            if self.created + self.max_age < now:
                return False
        except TypeError:
            pass

        # Check regular usage requirement
        try:
            if (self.last_used or self.created) + self.max_unused_period < now:
                return False
        except TypeError:
            pass

        return True

    def generate_key(self):
        # Entropy: len(ALPHABET) == 58, log_2(58) * 28 = 164.02
        self.plain = "".join(secrets.choice(ALPHABET) for _ in range(28))
        self.key = Token.make_hash(self.plain)
        return self.key

    @staticmethod
    def make_hash(plain):
        return make_password(plain, salt="static", hasher="pbkdf2_sha256_iter1")

    def get_policy(self, *, domain=None):
        order_by = F("domain").asc(
            nulls_last=True
        )  # default Postgres sorting, but: explicit is better than implicit
        return (
            self.tokendomainpolicy_set.filter(Q(domain=domain) | Q(domain__isnull=True))
            .order_by(order_by)
            .first()
        )

    @transaction.atomic
    def delete(self):
        # This is needed because Model.delete() emulates cascade delete via django.db.models.deletion.Collector.delete()
        # which deletes related objects in pk order.  However, the default policy has to be deleted last.
        # Perhaps this will change with https://code.djangoproject.com/ticket/21961
        self.tokendomainpolicy_set.filter(domain__isnull=False).delete()
        self.tokendomainpolicy_set.filter(domain__isnull=True).delete()
        return super().delete()


@pgtrigger.register(
    # Ensure that token_user is consistent with token
    pgtrigger.Trigger(
        name="token_user",
        operation=pgtrigger.Update | pgtrigger.Insert,
        when=pgtrigger.Before,
        func="NEW.token_user_id = (SELECT user_id FROM desecapi_token WHERE id = NEW.token_id); RETURN NEW;",
    ),
    # Ensure that if there is *any* domain policy for a given token, there is always one with domain=None.
    pgtrigger.Trigger(
        name="default_policy_on_insert",
        operation=pgtrigger.Insert,
        when=pgtrigger.Before,
        # Trigger `condition` arguments (corresponding to WHEN clause) don't support subqueries, so we use `func`
        func="IF (NEW.domain_id IS NOT NULL and NOT EXISTS(SELECT * FROM desecapi_tokendomainpolicy WHERE domain_id IS NULL AND token_id = NEW.token_id)) THEN "
        "  RAISE EXCEPTION 'Cannot insert non-default policy into % table when default policy is not present', TG_TABLE_NAME; "
        "END IF; RETURN NEW;",
    ),
    pgtrigger.Protect(
        name="default_policy_on_update",
        operation=pgtrigger.Update,
        when=pgtrigger.Before,
        condition=pgtrigger.Q(old__domain__isnull=True, new__domain__isnull=False),
    ),
    # Ideally, a deferred trigger (https://github.com/Opus10/django-pgtrigger/issues/14). Available in 3.4.0.
    pgtrigger.Trigger(
        name="default_policy_on_delete",
        operation=pgtrigger.Delete,
        when=pgtrigger.Before,
        # Trigger `condition` arguments (corresponding to WHEN clause) don't support subqueries, so we use `func`
        func="IF (OLD.domain_id IS NULL and EXISTS(SELECT * FROM desecapi_tokendomainpolicy WHERE domain_id IS NOT NULL AND token_id = OLD.token_id)) THEN "
        "  RAISE EXCEPTION 'Cannot delete default policy from % table when non-default policy is present', TG_TABLE_NAME; "
        "END IF; RETURN OLD;",
    ),
)
class TokenDomainPolicy(ExportModelOperationsMixin("TokenDomainPolicy"), models.Model):
    token = models.ForeignKey(Token, on_delete=models.CASCADE)
    domain = models.ForeignKey("Domain", on_delete=models.CASCADE, null=True)
    perm_dyndns = models.BooleanField(default=False)
    perm_rrsets = models.BooleanField(default=False)
    # Token user, filled via trigger. Used by compound FK constraints to tie domain.owner to token.user (see migration).
    token_user = models.ForeignKey(
        "User", on_delete=models.CASCADE, db_constraint=False, related_name="+"
    )

    class Meta:
        constraints = [
            models.UniqueConstraint(fields=["token", "domain"], name="unique_entry"),
            models.UniqueConstraint(
                fields=["token"],
                condition=Q(domain__isnull=True),
                name="unique_entry_null_domain",
            ),
        ]

    def clean(self):
        default_policy = self.token.get_policy(domain=None)
        if self.pk:  # update
            # Can't change policy's default status ("domain NULLness") to maintain policy precedence
            if (self.domain is None) != (self.pk == default_policy.pk):
                raise ValidationError(
                    {
                        "domain": "Policy precedence: Cannot disable default policy when others exist."
                    }
                )
        else:  # create
            # Can't violate policy precedence (default policy has to be first)
            if (self.domain is not None) and (default_policy is None):
                raise ValidationError(
                    {
                        "domain": "Policy precedence: The first policy must be the default policy."
                    }
                )

    def delete(self, *args, **kwargs):
        # Can't delete default policy when others exist
        if (self.domain is None) and self.token.tokendomainpolicy_set.exclude(
            domain__isnull=True
        ).exists():
            raise ValidationError(
                {
                    "domain": "Policy precedence: Can't delete default policy when there exist others."
                }
            )
        return super().delete(*args, **kwargs)

    def save(self, *args, **kwargs):
        self.clean()
        super().save(*args, **kwargs)