authentik.enterprise.signals

 1"""Enterprise signals"""
 2
 3from datetime import UTC, datetime
 4
 5from django.core.cache import cache
 6from django.db.models.signals import post_delete, post_save, pre_save
 7from django.dispatch import receiver
 8from django.utils.timezone import get_current_timezone, now
 9
10from authentik.enterprise.apps import GAUGE_LICENSE_EXPIRY, GAUGE_LICENSE_USAGE
11from authentik.enterprise.license import CACHE_KEY_ENTERPRISE_LICENSE, LicenseKey
12from authentik.enterprise.models import License, LicenseUsageStatus
13from authentik.enterprise.tasks import enterprise_update_usage
14from authentik.root.monitoring import monitoring_set
15from authentik.tasks.schedules.models import Schedule
16
17
18@receiver(monitoring_set)
19def monitoring_set_enterprise(sender, **kwargs):
20    """set enterprise gauges"""
21    summary = LicenseKey.cached_summary()
22    if summary.status == LicenseUsageStatus.UNLICENSED:
23        return
24    percentage_internal = (
25        0
26        if summary.internal_users <= 0
27        else LicenseKey.get_internal_user_count() / (summary.internal_users / 100)
28    )
29    percentage_external = (
30        0
31        if summary.external_users <= 0
32        else LicenseKey.get_external_user_count() / (summary.external_users / 100)
33    )
34    GAUGE_LICENSE_USAGE.labels(user_type="internal").set(percentage_internal)
35    GAUGE_LICENSE_USAGE.labels(user_type="external").set(percentage_external)
36    GAUGE_LICENSE_EXPIRY.set((summary.latest_valid.replace(tzinfo=UTC) - now()).total_seconds())
37
38
39@receiver(pre_save, sender=License)
40def pre_save_license(sender: type[License], instance: License, **_):
41    """Extract data from license jwt and save it into model"""
42    status = instance.status
43    instance.name = status.name
44    instance.internal_users = status.internal_users
45    instance.external_users = status.external_users
46    instance.expiry = datetime.fromtimestamp(status.exp, tz=get_current_timezone())
47
48
49@receiver(post_save, sender=License)
50def post_save_license(sender: type[License], instance: License, **_):
51    """Trigger license usage calculation when license is saved"""
52    cache.delete(CACHE_KEY_ENTERPRISE_LICENSE)
53    Schedule.dispatch_by_actor(enterprise_update_usage)
54
55
56@receiver(post_delete, sender=License)
57def post_delete_license(sender: type[License], instance: License, **_):
58    """Clear license cache when license is deleted"""
59    cache.delete(CACHE_KEY_ENTERPRISE_LICENSE)