desec-stack/api/desecapi/views/dyndns.py

import base64
import binascii
from functools import cached_property

from rest_framework import generics
from rest_framework.authentication import get_authorization_header
from rest_framework.exceptions import NotFound, ValidationError
from rest_framework.response import Response

from desecapi import metrics
from desecapi.authentication import BasicTokenAuthentication, TokenAuthentication, URLParamAuthentication
from desecapi.exceptions import ConcurrencyException
from desecapi.models import Domain
from desecapi.pdns_change_tracker import PDNSChangeTracker
from desecapi.permissions import TokenHasDomainDynDNSPermission
from desecapi.renderers import PlainTextRenderer
from desecapi.serializers import RRsetSerializer


class DynDNS12UpdateView(generics.GenericAPIView):
    authentication_classes = (TokenAuthentication, BasicTokenAuthentication, URLParamAuthentication,)
    permission_classes = (TokenHasDomainDynDNSPermission,)
    renderer_classes = [PlainTextRenderer]
    serializer_class = RRsetSerializer
    throttle_scope = 'dyndns'

    @property
    def throttle_scope_bucket(self):
        return self.domain.name

    def _find_ip(self, params, version):
        if version == 4:
            look_for = '.'
        elif version == 6:
            look_for = ':'
        else:
            raise Exception

        # Check URL parameters
        for p in params:
            if p in self.request.query_params:
                if not len(self.request.query_params[p]):
                    return None
                if look_for in self.request.query_params[p]:
                    return self.request.query_params[p]

        # Check remote IP address
        client_ip = self.request.META.get('REMOTE_ADDR')
        if look_for in client_ip:
            return client_ip

        # give up
        return None

    @cached_property
    def qname(self):
        # hostname parameter
        try:
            if self.request.query_params['hostname'] != 'YES':
                return self.request.query_params['hostname'].lower()
        except KeyError:
            pass

        # host_id parameter
        try:
            return self.request.query_params['host_id'].lower()
        except KeyError:
            pass

        # http basic auth username
        try:
            domain_name = base64.b64decode(
                get_authorization_header(self.request).decode().split(' ')[1].encode()).decode().split(':')[0]
            if domain_name and '@' not in domain_name:
                return domain_name.lower()
        except (binascii.Error, IndexError, UnicodeDecodeError):
            pass

        # username parameter
        try:
            return self.request.query_params['username'].lower()
        except KeyError:
            pass

        # only domain associated with this user account
        try:
            return self.request.user.domains.get().name
        except Domain.MultipleObjectsReturned:
            raise ValidationError(detail={
                "detail": "Request does not properly specify domain for update.",
                "code": "domain-unspecified"
            })
        except Domain.DoesNotExist:
            metrics.get('desecapi_dynDNS12_domain_not_found').inc()
            raise NotFound('nohost')

    @cached_property
    def domain(self):
        try:
            return Domain.objects.filter_qname(self.qname, owner=self.request.user).order_by('-name_length')[0]
        except (IndexError, ValueError):
            raise NotFound('nohost')

    @property
    def subname(self):
        return self.qname.rpartition(f'.{self.domain.name}')[0]

    def get_serializer_context(self):
        return {**super().get_serializer_context(), 'domain': self.domain, 'minimum_ttl': 60}

    def get_queryset(self):
        return self.domain.rrset_set.filter(subname=self.subname, type__in=['A', 'AAAA'])

    def get(self, request, *args, **kwargs):
        instances = self.get_queryset().all()

        ipv4 = self._find_ip(['myip', 'myipv4', 'ip'], version=4)
        ipv6 = self._find_ip(['myipv6', 'ipv6', 'myip', 'ip'], version=6)

        data = [
            {'type': 'A', 'subname': self.subname, 'ttl': 60, 'records': [ipv4] if ipv4 else []},
            {'type': 'AAAA', 'subname': self.subname, 'ttl': 60, 'records': [ipv6] if ipv6 else []},
        ]

        serializer = self.get_serializer(instances, data=data, many=True, partial=True)
        try:
            serializer.is_valid(raise_exception=True)
        except ValidationError as e:
            if any(
                    any(
                        getattr(non_field_error, 'code', '') == 'unique'
                        for non_field_error
                        in err.get('non_field_errors', [])
                    )
                    for err in e.detail
            ):
                raise ConcurrencyException from e
            raise e

        with PDNSChangeTracker():
            serializer.save()

        return Response('good', content_type='text/plain')