authentik.api.tests.test_auth

  1"""Test API Authentication"""
  2
  3import json
  4from base64 import b64encode
  5from unittest.mock import patch
  6
  7from django.conf import settings
  8from django.test import TestCase
  9from django.utils import timezone
 10from rest_framework.exceptions import AuthenticationFailed
 11
 12from authentik.api.authentication import IPCUser, TokenAuthentication
 13from authentik.blueprints.tests import reconcile_app
 14from authentik.common.oauth.constants import SCOPE_AUTHENTIK_API
 15from authentik.core.models import Token, TokenIntents, UserTypes
 16from authentik.core.tests.utils import create_test_admin_user, create_test_flow
 17from authentik.lib.generators import generate_id
 18from authentik.outposts.apps import MANAGED_OUTPOST
 19from authentik.outposts.models import Outpost
 20from authentik.providers.oauth2.models import AccessToken, OAuth2Provider
 21
 22
 23class TestAPIAuth(TestCase):
 24    """Test API Authentication"""
 25
 26    def test_invalid_type(self):
 27        """Test invalid type"""
 28        self.assertIsNone(TokenAuthentication().bearer_auth(b"foo bar"))
 29
 30    def test_invalid_empty(self):
 31        """Test invalid type"""
 32        self.assertIsNone(TokenAuthentication().bearer_auth(b"Bearer "))
 33        self.assertIsNone(TokenAuthentication().bearer_auth(b""))
 34
 35    def test_invalid_no_token(self):
 36        """Test invalid with no token"""
 37        auth = b64encode(b":abc").decode()
 38        self.assertIsNone(TokenAuthentication().bearer_auth(f"Basic :{auth}".encode()))
 39
 40    def test_bearer_valid(self):
 41        """Test valid token"""
 42        token = Token.objects.create(intent=TokenIntents.INTENT_API, user=create_test_admin_user())
 43        user, tk = TokenAuthentication().bearer_auth(f"Bearer {token.key}".encode())
 44        self.assertEqual(user, token.user)
 45        self.assertEqual(token, token)
 46
 47    def test_bearer_valid_deactivated(self):
 48        """Test valid token"""
 49        user = create_test_admin_user()
 50        user.is_active = False
 51        user.save()
 52        token = Token.objects.create(intent=TokenIntents.INTENT_API, user=user)
 53        with self.assertRaises(AuthenticationFailed):
 54            TokenAuthentication().bearer_auth(f"Bearer {token.key}".encode())
 55
 56    @reconcile_app("authentik_outposts")
 57    def test_managed_outpost_fail(self):
 58        """Test managed outpost"""
 59        outpost = Outpost.objects.filter(managed=MANAGED_OUTPOST).first()
 60        outpost.user.delete()
 61        outpost.delete()
 62        with self.assertRaises(AuthenticationFailed):
 63            TokenAuthentication().bearer_auth(f"Bearer {settings.SECRET_KEY}".encode())
 64
 65    @reconcile_app("authentik_outposts")
 66    def test_managed_outpost_success(self):
 67        """Test managed outpost"""
 68        user, outpost = TokenAuthentication().bearer_auth(f"Bearer {settings.SECRET_KEY}".encode())
 69        self.assertEqual(user.type, UserTypes.INTERNAL_SERVICE_ACCOUNT)
 70        self.assertEqual(outpost, Outpost.objects.filter(managed=MANAGED_OUTPOST).first())
 71
 72    def test_jwt_valid(self):
 73        """Test valid JWT"""
 74        provider = OAuth2Provider.objects.create(
 75            name=generate_id(), client_id=generate_id(), authorization_flow=create_test_flow()
 76        )
 77        access = AccessToken.objects.create(
 78            user=create_test_admin_user(),
 79            provider=provider,
 80            token=generate_id(),
 81            auth_time=timezone.now(),
 82            _scope=SCOPE_AUTHENTIK_API,
 83            _id_token=json.dumps({}),
 84        )
 85        user, token = TokenAuthentication().bearer_auth(f"Bearer {access.token}".encode())
 86        self.assertEqual(user, access.user)
 87        self.assertEqual(token, access)
 88
 89    def test_jwt_missing_scope(self):
 90        """Test valid JWT"""
 91        provider = OAuth2Provider.objects.create(
 92            name=generate_id(), client_id=generate_id(), authorization_flow=create_test_flow()
 93        )
 94        access = AccessToken.objects.create(
 95            user=create_test_admin_user(),
 96            provider=provider,
 97            token=generate_id(),
 98            auth_time=timezone.now(),
 99            _scope="",
100            _id_token=json.dumps({}),
101        )
102        with self.assertRaises(AuthenticationFailed):
103            TokenAuthentication().bearer_auth(f"Bearer {access.token}".encode())
104
105    def test_ipc(self):
106        """Test IPC auth (mock key)"""
107        key = generate_id()
108        with patch("authentik.api.authentication.ipc_key", key):
109            user, ctx = TokenAuthentication().bearer_auth(f"Bearer {key}".encode())
110        self.assertEqual(user, IPCUser())
111        self.assertEqual(ctx, None)