desec-stack/api/desecapi/tests/test_authentication.py

from datetime import timedelta
import json
from unittest import mock

from django.utils import timezone
from rest_framework.status import HTTP_200_OK, HTTP_401_UNAUTHORIZED

from desecapi.models import Token
from desecapi.tests.base import DynDomainOwnerTestCase


class DynUpdateAuthenticationTestCase(DynDomainOwnerTestCase):
    NUM_OWNED_DOMAINS = 1

    def _get_dyndns12(self):
        with self.assertPdnsNoRequestsBut(self.requests_desec_rr_sets_update()):
            return self.client.get(self.reverse('v1:dyndns12update'))

    def assertDynDNS12Status(self, code=HTTP_200_OK, authorization=None):
        if authorization:
            self.client.set_credentials_basic_auth(authorization)
        self.assertStatus(self._get_dyndns12(), code)

    def test_username_password(self):
        # noinspection PyPep8Naming
        def assertDynDNS12AuthenticationStatus(username, token, code):
            self.client.set_credentials_basic_auth(username, token)
            self.assertDynDNS12Status(code)

        assertDynDNS12AuthenticationStatus('', self.token.plain, HTTP_200_OK)
        assertDynDNS12AuthenticationStatus(self.owner.get_username(), self.token.plain, HTTP_200_OK)
        assertDynDNS12AuthenticationStatus(self.my_domain.name, self.token.plain, HTTP_200_OK)
        assertDynDNS12AuthenticationStatus(' ' + self.my_domain.name, self.token.plain, HTTP_401_UNAUTHORIZED)
        assertDynDNS12AuthenticationStatus('wrong', self.token.plain, HTTP_401_UNAUTHORIZED)
        assertDynDNS12AuthenticationStatus('', 'wrong', HTTP_401_UNAUTHORIZED)
        assertDynDNS12AuthenticationStatus(self.user.get_username(), 'wrong', HTTP_401_UNAUTHORIZED)

    def test_malformed_basic_auth(self):
        for authorization in [
            'asdf:asdf:sadf',
            'asdf',
            'bull[%]shit',
            '你好',
            '💩💩💩💩',
            '💩💩:💩💩',
        ]:
            self.assertDynDNS12Status(authorization=authorization, code=HTTP_401_UNAUTHORIZED)


class TokenAuthenticationTestCase(DynDomainOwnerTestCase):

    def setUp(self):
        super().setUp()
        # Refresh token from database, but keep plain value
        self.token, self.token.plain = Token.objects.get(pk=self.token.pk), self.token.plain

    def assertAuthenticationStatus(self, code, plain=None, expired=False ,**kwargs):
        plain = plain or self.token.plain
        self.client.set_credentials_token_auth(plain)

        # only forward REMOTE_ADDR if not None
        if kwargs.get('REMOTE_ADDR') is None:
            kwargs.pop('REMOTE_ADDR', None)

        response = self.client.get(self.reverse('v1:root'), **kwargs)
        body = json.dumps({'detail': 'Invalid token.'}) if code == HTTP_401_UNAUTHORIZED else None
        self.assertResponse(response, code, body)

        if expired:
            key = Token.make_hash(plain)
            self.assertFalse(Token.objects.get(key=key).is_valid)

    def test_token_case_sensitive(self):
        self.assertAuthenticationStatus(HTTP_200_OK)
        self.assertAuthenticationStatus(HTTP_401_UNAUTHORIZED, self.token.plain.upper())
        self.assertAuthenticationStatus(HTTP_401_UNAUTHORIZED, self.token.plain.lower())

    def test_token_subnets(self):
        datas = [  # Format: allowed_subnets, status, client_ip | None, [client_ip, ...]
            ([], HTTP_401_UNAUTHORIZED, None),
            (['127.0.0.1'], HTTP_200_OK, None),
            (['1.2.3.4'], HTTP_401_UNAUTHORIZED, None),
            (['1.2.3.4'], HTTP_200_OK, '1.2.3.4'),
            (['1.2.3.0/24'], HTTP_200_OK, '1.2.3.4'),
            (['1.2.3.0/24'], HTTP_401_UNAUTHORIZED, 'bade::affe'),
            (['bade::/64'], HTTP_200_OK, 'bade::affe'),
            (['bade::/64', '1.2.3.0/24'], HTTP_200_OK, 'bade::affe', '1.2.3.66'),
        ]

        for allowed_subnets, status, client_ips in ((*data[:2], data[2:]) for data in datas):
            self.token.allowed_subnets = allowed_subnets
            self.token.save()
            for client_ip in client_ips:
                self.assertAuthenticationStatus(status, REMOTE_ADDR=client_ip)

    def test_token_max_age(self):
        # No maximum age: can use now and in ten years
        self.token.max_age = None
        self.token.save()

        self.assertAuthenticationStatus(HTTP_200_OK)
        with mock.patch('desecapi.models.timezone.now', return_value=timezone.now() + timedelta(days=3650)):
            self.assertAuthenticationStatus(HTTP_200_OK)

        # Maximum age zero: token cannot be used
        self.token.max_age = timedelta(0)
        self.token.save()
        self.assertAuthenticationStatus(HTTP_401_UNAUTHORIZED, expired=True)

        # Maximum age 10 10:10:10: can use one second before, but not once second after
        period = timedelta(days=10, hours=10, minutes=10, seconds=10)
        self.token.max_age = period
        self.token.save()

        second = timedelta(seconds=1)
        with mock.patch('desecapi.models.timezone.now', return_value=self.token.created + period - second):
            self.assertAuthenticationStatus(HTTP_200_OK)
        with mock.patch('desecapi.models.timezone.now', return_value=self.token.created + period + second):
            self.assertAuthenticationStatus(HTTP_401_UNAUTHORIZED, expired=True)

    def test_token_max_unused_period(self):
        plain = self.token.plain
        second = timedelta(seconds=1)

        # Maximum unused period zero: token cannot be used
        self.token.max_unused_period = timedelta(0)
        self.token.save()
        self.assertAuthenticationStatus(HTTP_401_UNAUTHORIZED, expired=True)

        # Maximum unused period
        period = timedelta(days=10, hours=10, minutes=10, seconds=10)
        self.token.max_unused_period = period
        self.token.save()

        # Can't use after period if token was never used (last_used is None)
        self.assertIsNone(self.token.last_used)
        with mock.patch('desecapi.models.timezone.now', return_value=self.token.created + period + second):
            self.assertAuthenticationStatus(HTTP_401_UNAUTHORIZED, plain=plain, expired=True)
            self.assertIsNone(Token.objects.get(pk=self.token.pk).last_used)  # unchanged

        # Can use after half the period
        with mock.patch('desecapi.models.timezone.now', return_value=self.token.created + period/2):
            self.assertAuthenticationStatus(HTTP_200_OK, plain=plain)
        self.token = Token.objects.get(pk=self.token.pk)  # update last_used field

        # Can't use once another period is over
        with mock.patch('desecapi.models.timezone.now', return_value=self.token.last_used + period + second):
            self.assertAuthenticationStatus(HTTP_401_UNAUTHORIZED, plain=plain, expired=True)
            self.assertEqual(self.token.last_used, Token.objects.get(pk=self.token.pk).last_used)  # unchanged

        # ... but one second before, and also for one more period
        with mock.patch('desecapi.models.timezone.now', return_value=self.token.last_used + period - second):
            self.assertAuthenticationStatus(HTTP_200_OK, plain=plain)
        with mock.patch('desecapi.models.timezone.now', return_value=self.token.last_used + 2*period - 2*second):
            self.assertAuthenticationStatus(HTTP_200_OK, plain=plain)

        # No maximum age: can use now and in ten years
        self.token.max_unused_period = None
        self.token.save()

        self.assertAuthenticationStatus(HTTP_200_OK, plain=plain)
        with mock.patch('desecapi.models.timezone.now', return_value=timezone.now() + timedelta(days=3650)):
            self.assertAuthenticationStatus(HTTP_200_OK, plain=plain)

    def test_token_max_age_max_unused_period(self):
        hour = timedelta(hours=1)
        self.token.max_age = 3 * hour
        self.token.max_unused_period = hour
        self.token.save()

        # max_unused_period wins if tighter than max_age
        with mock.patch('desecapi.models.timezone.now', return_value=self.token.created + 1.25*hour):
            self.assertAuthenticationStatus(HTTP_401_UNAUTHORIZED, expired=True)

        # Can use immediately
        self.assertAuthenticationStatus(HTTP_200_OK)

        # Can use continuously within max_unused_period
        with mock.patch('desecapi.models.timezone.now', return_value=self.token.created + 0.75*hour):
            self.assertAuthenticationStatus(HTTP_200_OK)
        with mock.patch('desecapi.models.timezone.now', return_value=self.token.created + 1.5*hour):
            self.assertAuthenticationStatus(HTTP_200_OK)

        # max_unused_period wins again if tighter than max_age
        with mock.patch('desecapi.models.timezone.now', return_value=self.token.created + 2.75*hour):
            self.assertAuthenticationStatus(HTTP_401_UNAUTHORIZED, expired=True)

        # Can use continuously within max_unused_period
        with mock.patch('desecapi.models.timezone.now', return_value=self.token.created + 2.25*hour):
            self.assertAuthenticationStatus(HTTP_200_OK)
        with mock.patch('desecapi.models.timezone.now', return_value=self.token.created + 2.75*hour):
            self.assertAuthenticationStatus(HTTP_200_OK)

        # max_age wins again if tighter than max_unused_period
        with mock.patch('desecapi.models.timezone.now', return_value=self.token.created + 3.25*hour):
            self.assertAuthenticationStatus(HTTP_401_UNAUTHORIZED, expired=True)