authentik.admin.api.version

authentik administration overview

 1"""authentik administration overview"""
 2
 3from django.core.cache import cache
 4from django_tenants.utils import get_public_schema_name
 5from drf_spectacular.utils import extend_schema
 6from packaging.version import parse
 7from rest_framework.fields import SerializerMethodField
 8from rest_framework.permissions import IsAuthenticated
 9from rest_framework.request import Request
10from rest_framework.response import Response
11from rest_framework.views import APIView
12
13from authentik import authentik_build_hash, authentik_version
14from authentik.admin.tasks import VERSION_CACHE_KEY, VERSION_NULL, update_latest_version
15from authentik.core.api.utils import PassiveSerializer
16from authentik.outposts.models import Outpost
17from authentik.tenants.utils import get_current_tenant
18
19
20class VersionSerializer(PassiveSerializer):
21    """Get running and latest version."""
22
23    version_current = SerializerMethodField()
24    version_latest = SerializerMethodField()
25    version_latest_valid = SerializerMethodField()
26    build_hash = SerializerMethodField()
27    outdated = SerializerMethodField()
28    outpost_outdated = SerializerMethodField()
29
30    def get_build_hash(self, _) -> str:
31        """Get build hash, if version is not latest or released"""
32        return authentik_build_hash()
33
34    def get_version_current(self, _) -> str:
35        """Get current version"""
36        return authentik_version()
37
38    def get_version_latest(self, _) -> str:
39        """Get latest version from cache"""
40        if get_current_tenant().schema_name != get_public_schema_name():
41            return authentik_version()
42        version_in_cache = cache.get(VERSION_CACHE_KEY)
43        if not version_in_cache:  # pragma: no cover
44            update_latest_version.send()
45            return authentik_version()
46        return version_in_cache
47
48    def get_version_latest_valid(self, _) -> bool:
49        """Check if latest version is valid"""
50        return cache.get(VERSION_CACHE_KEY) != VERSION_NULL
51
52    def get_outdated(self, instance) -> bool:
53        """Check if we're running the latest version"""
54        return parse(self.get_version_current(instance)) < parse(self.get_version_latest(instance))
55
56    def get_outpost_outdated(self, _) -> bool:
57        """Check if any outpost is outdated/has a version mismatch"""
58        any_outdated = False
59        for outpost in Outpost.objects.all():
60            for state in outpost.state:
61                if state.version_outdated:
62                    any_outdated = True
63        return any_outdated
64
65
66class VersionView(APIView):
67    """Get running and latest version."""
68
69    permission_classes = [IsAuthenticated]
70    pagination_class = None
71    filter_backends = []
72
73    @extend_schema(responses={200: VersionSerializer(many=False)})
74    def get(self, request: Request) -> Response:
75        """Get running and latest version."""
76        return Response(VersionSerializer(True).data)
class VersionSerializer(authentik.core.api.utils.PassiveSerializer): 
21class VersionSerializer(PassiveSerializer):
22    """Get running and latest version."""
23
24    version_current = SerializerMethodField()
25    version_latest = SerializerMethodField()
26    version_latest_valid = SerializerMethodField()
27    build_hash = SerializerMethodField()
28    outdated = SerializerMethodField()
29    outpost_outdated = SerializerMethodField()
30
31    def get_build_hash(self, _) -> str:
32        """Get build hash, if version is not latest or released"""
33        return authentik_build_hash()
34
35    def get_version_current(self, _) -> str:
36        """Get current version"""
37        return authentik_version()
38
39    def get_version_latest(self, _) -> str:
40        """Get latest version from cache"""
41        if get_current_tenant().schema_name != get_public_schema_name():
42            return authentik_version()
43        version_in_cache = cache.get(VERSION_CACHE_KEY)
44        if not version_in_cache:  # pragma: no cover
45            update_latest_version.send()
46            return authentik_version()
47        return version_in_cache
48
49    def get_version_latest_valid(self, _) -> bool:
50        """Check if latest version is valid"""
51        return cache.get(VERSION_CACHE_KEY) != VERSION_NULL
52
53    def get_outdated(self, instance) -> bool:
54        """Check if we're running the latest version"""
55        return parse(self.get_version_current(instance)) < parse(self.get_version_latest(instance))
56
57    def get_outpost_outdated(self, _) -> bool:
58        """Check if any outpost is outdated/has a version mismatch"""
59        any_outdated = False
60        for outpost in Outpost.objects.all():
61            for state in outpost.state:
62                if state.version_outdated:
63                    any_outdated = True
64        return any_outdated

Get running and latest version.

version_current
version_latest
version_latest_valid
build_hash
outdated
outpost_outdated
def get_build_hash(self, _) -> str: 
31    def get_build_hash(self, _) -> str:
32        """Get build hash, if version is not latest or released"""
33        return authentik_build_hash()

Get build hash, if version is not latest or released

def get_version_current(self, _) -> str: 
35    def get_version_current(self, _) -> str:
36        """Get current version"""
37        return authentik_version()

Get current version

def get_version_latest(self, _) -> str: 
39    def get_version_latest(self, _) -> str:
40        """Get latest version from cache"""
41        if get_current_tenant().schema_name != get_public_schema_name():
42            return authentik_version()
43        version_in_cache = cache.get(VERSION_CACHE_KEY)
44        if not version_in_cache:  # pragma: no cover
45            update_latest_version.send()
46            return authentik_version()
47        return version_in_cache

Get latest version from cache

def get_version_latest_valid(self, _) -> bool: 
49    def get_version_latest_valid(self, _) -> bool:
50        """Check if latest version is valid"""
51        return cache.get(VERSION_CACHE_KEY) != VERSION_NULL

Check if latest version is valid

def get_outdated(self, instance) -> bool: 
53    def get_outdated(self, instance) -> bool:
54        """Check if we're running the latest version"""
55        return parse(self.get_version_current(instance)) < parse(self.get_version_latest(instance))

Check if we're running the latest version

def get_outpost_outdated(self, _) -> bool: 
57    def get_outpost_outdated(self, _) -> bool:
58        """Check if any outpost is outdated/has a version mismatch"""
59        any_outdated = False
60        for outpost in Outpost.objects.all():
61            for state in outpost.state:
62                if state.version_outdated:
63                    any_outdated = True
64        return any_outdated

Check if any outpost is outdated/has a version mismatch

Inherited Members
authentik.core.api.utils.PassiveSerializer
create
update
class VersionView(rest_framework.views.APIView): 
67class VersionView(APIView):
68    """Get running and latest version."""
69
70    permission_classes = [IsAuthenticated]
71    pagination_class = None
72    filter_backends = []
73
74    @extend_schema(responses={200: VersionSerializer(many=False)})
75    def get(self, request: Request) -> Response:
76        """Get running and latest version."""
77        return Response(VersionSerializer(True).data)

Get running and latest version.

permission_classes = [<class 'rest_framework.permissions.IsAuthenticated'>]
pagination_class = None
filter_backends = []
@extend_schema(responses={200: VersionSerializer(many=False)})
def get( self, request: rest_framework.request.Request) -> rest_framework.response.Response: 
74    @extend_schema(responses={200: VersionSerializer(many=False)})
75    def get(self, request: Request) -> Response:
76        """Get running and latest version."""
77        return Response(VersionSerializer(True).data)

Get running and latest version.