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


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)
    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):
        self.plain = secrets.token_urlsafe(21)
        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):
        # 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()

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