authentik.outposts.apps

authentik outposts app config

 1"""authentik outposts app config"""
 2
 3from prometheus_client import Gauge
 4from structlog.stdlib import get_logger
 5
 6from authentik.blueprints.apps import ManagedAppConfig
 7from authentik.lib.config import CONFIG
 8from authentik.lib.utils.time import fqdn_rand
 9from authentik.tasks.schedules.common import ScheduleSpec
10
11LOGGER = get_logger()
12
13GAUGE_OUTPOSTS_CONNECTED = Gauge(
14    "authentik_outposts_connected",
15    "Currently connected outposts",
16    ["tenant", "outpost", "uid", "expected"],
17)
18GAUGE_OUTPOSTS_LAST_UPDATE = Gauge(
19    "authentik_outposts_last_update",
20    "Last update from any outpost",
21    ["tenant", "outpost", "uid", "version"],
22)
23MANAGED_OUTPOST = "goauthentik.io/outposts/embedded"
24MANAGED_OUTPOST_NAME = "authentik Embedded Outpost"
25
26
27class AuthentikOutpostConfig(ManagedAppConfig):
28    """authentik outposts app config"""
29
30    name = "authentik.outposts"
31    label = "authentik_outposts"
32    verbose_name = "authentik Outpost"
33    default = True
34
35    @ManagedAppConfig.reconcile_tenant
36    def embedded_outpost(self):
37        """Ensure embedded outpost"""
38        from authentik.outposts.models import (
39            DockerServiceConnection,
40            KubernetesServiceConnection,
41            Outpost,
42            OutpostType,
43        )
44
45        if not CONFIG.get_bool("outposts.disable_embedded_outpost", False):
46            if outpost := Outpost.objects.filter(name=MANAGED_OUTPOST_NAME, managed="").first():
47                outpost.managed = MANAGED_OUTPOST
48                outpost.save()
49                return
50            outpost, created = Outpost.objects.update_or_create(
51                defaults={
52                    "type": OutpostType.PROXY,
53                    "name": MANAGED_OUTPOST_NAME,
54                },
55                managed=MANAGED_OUTPOST,
56            )
57            if created:
58                if KubernetesServiceConnection.objects.exists():
59                    outpost.service_connection = KubernetesServiceConnection.objects.first()
60                elif DockerServiceConnection.objects.exists():
61                    outpost.service_connection = DockerServiceConnection.objects.first()
62                outpost.save()
63        else:
64            Outpost.objects.filter(managed=MANAGED_OUTPOST).delete()
65
66    @property
67    def tenant_schedule_specs(self) -> list[ScheduleSpec]:
68        from authentik.outposts.tasks import outpost_token_ensurer
69
70        return [
71            ScheduleSpec(
72                actor=outpost_token_ensurer,
73                crontab=f"{fqdn_rand('outpost_token_ensurer')} */8 * * *",
74            ),
75        ]
76
77    @property
78    def global_schedule_specs(self) -> list[ScheduleSpec]:
79        from authentik.outposts.tasks import outpost_connection_discovery
80
81        return [
82            ScheduleSpec(
83                actor=outpost_connection_discovery,
84                crontab=f"{fqdn_rand('outpost_connection_discovery')} */8 * * *",
85                send_on_startup=True,
86                paused=not CONFIG.get_bool("outposts.discover"),
87            ),
88        ]
LOGGER = <BoundLoggerLazyProxy(logger=None, wrapper_class=None, processors=None, context_class=None, initial_values={}, logger_factory_args=())>
GAUGE_OUTPOSTS_CONNECTED = prometheus_client.metrics.Gauge(authentik_outposts_connected)
GAUGE_OUTPOSTS_LAST_UPDATE = prometheus_client.metrics.Gauge(authentik_outposts_last_update)
MANAGED_OUTPOST = 'goauthentik.io/outposts/embedded'
MANAGED_OUTPOST_NAME = 'authentik Embedded Outpost'
class AuthentikOutpostConfig(authentik.blueprints.apps.ManagedAppConfig): 
28class AuthentikOutpostConfig(ManagedAppConfig):
29    """authentik outposts app config"""
30
31    name = "authentik.outposts"
32    label = "authentik_outposts"
33    verbose_name = "authentik Outpost"
34    default = True
35
36    @ManagedAppConfig.reconcile_tenant
37    def embedded_outpost(self):
38        """Ensure embedded outpost"""
39        from authentik.outposts.models import (
40            DockerServiceConnection,
41            KubernetesServiceConnection,
42            Outpost,
43            OutpostType,
44        )
45
46        if not CONFIG.get_bool("outposts.disable_embedded_outpost", False):
47            if outpost := Outpost.objects.filter(name=MANAGED_OUTPOST_NAME, managed="").first():
48                outpost.managed = MANAGED_OUTPOST
49                outpost.save()
50                return
51            outpost, created = Outpost.objects.update_or_create(
52                defaults={
53                    "type": OutpostType.PROXY,
54                    "name": MANAGED_OUTPOST_NAME,
55                },
56                managed=MANAGED_OUTPOST,
57            )
58            if created:
59                if KubernetesServiceConnection.objects.exists():
60                    outpost.service_connection = KubernetesServiceConnection.objects.first()
61                elif DockerServiceConnection.objects.exists():
62                    outpost.service_connection = DockerServiceConnection.objects.first()
63                outpost.save()
64        else:
65            Outpost.objects.filter(managed=MANAGED_OUTPOST).delete()
66
67    @property
68    def tenant_schedule_specs(self) -> list[ScheduleSpec]:
69        from authentik.outposts.tasks import outpost_token_ensurer
70
71        return [
72            ScheduleSpec(
73                actor=outpost_token_ensurer,
74                crontab=f"{fqdn_rand('outpost_token_ensurer')} */8 * * *",
75            ),
76        ]
77
78    @property
79    def global_schedule_specs(self) -> list[ScheduleSpec]:
80        from authentik.outposts.tasks import outpost_connection_discovery
81
82        return [
83            ScheduleSpec(
84                actor=outpost_connection_discovery,
85                crontab=f"{fqdn_rand('outpost_connection_discovery')} */8 * * *",
86                send_on_startup=True,
87                paused=not CONFIG.get_bool("outposts.discover"),
88            ),
89        ]

authentik outposts app config

label = 'authentik_outposts'
verbose_name = 'authentik Outpost'
default = True
@ManagedAppConfig.reconcile_tenant
def embedded_outpost(self): 
36    @ManagedAppConfig.reconcile_tenant
37    def embedded_outpost(self):
38        """Ensure embedded outpost"""
39        from authentik.outposts.models import (
40            DockerServiceConnection,
41            KubernetesServiceConnection,
42            Outpost,
43            OutpostType,
44        )
45
46        if not CONFIG.get_bool("outposts.disable_embedded_outpost", False):
47            if outpost := Outpost.objects.filter(name=MANAGED_OUTPOST_NAME, managed="").first():
48                outpost.managed = MANAGED_OUTPOST
49                outpost.save()
50                return
51            outpost, created = Outpost.objects.update_or_create(
52                defaults={
53                    "type": OutpostType.PROXY,
54                    "name": MANAGED_OUTPOST_NAME,
55                },
56                managed=MANAGED_OUTPOST,
57            )
58            if created:
59                if KubernetesServiceConnection.objects.exists():
60                    outpost.service_connection = KubernetesServiceConnection.objects.first()
61                elif DockerServiceConnection.objects.exists():
62                    outpost.service_connection = DockerServiceConnection.objects.first()
63                outpost.save()
64        else:
65            Outpost.objects.filter(managed=MANAGED_OUTPOST).delete()

Ensure embedded outpost

tenant_schedule_specs: list[authentik.tasks.schedules.common.ScheduleSpec] 
67    @property
68    def tenant_schedule_specs(self) -> list[ScheduleSpec]:
69        from authentik.outposts.tasks import outpost_token_ensurer
70
71        return [
72            ScheduleSpec(
73                actor=outpost_token_ensurer,
74                crontab=f"{fqdn_rand('outpost_token_ensurer')} */8 * * *",
75            ),
76        ]

Get a list of schedule specs that must exist in each tenant

global_schedule_specs: list[authentik.tasks.schedules.common.ScheduleSpec] 
78    @property
79    def global_schedule_specs(self) -> list[ScheduleSpec]:
80        from authentik.outposts.tasks import outpost_connection_discovery
81
82        return [
83            ScheduleSpec(
84                actor=outpost_connection_discovery,
85                crontab=f"{fqdn_rand('outpost_connection_discovery')} */8 * * *",
86                send_on_startup=True,
87                paused=not CONFIG.get_bool("outposts.discover"),
88            ),
89        ]

Get a list of schedule specs that must exist in the default tenant

Inherited Members
authentik.blueprints.apps.ManagedAppConfig
ManagedAppConfig
logger
RECONCILE_GLOBAL_CATEGORY
RECONCILE_TENANT_CATEGORY
ready
import_module
reconcile_tenant
reconcile_global