authentik.outposts.controllers.k8s.secret

Kubernetes Secret Reconciler

View Source

 1"""Kubernetes Secret Reconciler"""
 2
 3from base64 import b64encode
 4from typing import TYPE_CHECKING
 5
 6from kubernetes.client import CoreV1Api, V1Secret
 7
 8from authentik.outposts.controllers.base import FIELD_MANAGER
 9from authentik.outposts.controllers.k8s.base import KubernetesObjectReconciler
10from authentik.outposts.controllers.k8s.triggers import NeedsUpdate
11
12if TYPE_CHECKING:
13    from authentik.outposts.controllers.kubernetes import KubernetesController
14
15
16def b64string(source: str) -> str:
17    """Base64 Encode string"""
18    return b64encode(source.encode()).decode("utf-8")
19
20
21class SecretReconciler(KubernetesObjectReconciler[V1Secret]):
22    """Kubernetes Secret Reconciler"""
23
24    def __init__(self, controller: KubernetesController) -> None:
25        super().__init__(controller)
26        self.api = CoreV1Api(controller.client)
27
28    @property
29    def noop(self) -> bool:
30        return self.is_embedded
31
32    @staticmethod
33    def reconciler_name() -> str:
34        return "secret"
35
36    def reconcile(self, current: V1Secret, reference: V1Secret):
37        super().reconcile(current, reference)
38        for key in reference.data.keys():
39            if key not in current.data or current.data[key] != reference.data[key]:
40                raise NeedsUpdate()
41
42    def get_reference_object(self) -> V1Secret:
43        """Get deployment object for outpost"""
44        meta = self.get_object_meta(name=self.name)
45        return V1Secret(
46            metadata=meta,
47            data={
48                "authentik_host": b64string(self.controller.outpost.config.authentik_host),
49                "authentik_host_insecure": b64string(
50                    str(self.controller.outpost.config.authentik_host_insecure)
51                ),
52                "token": b64string(self.controller.outpost.token.key),
53                "authentik_host_browser": b64string(
54                    self.controller.outpost.config.authentik_host_browser
55                ),
56            },
57        )
58
59    def create(self, reference: V1Secret):
60        return self.api.create_namespaced_secret(
61            self.namespace, reference, field_manager=FIELD_MANAGER
62        )
63
64    def delete(self, reference: V1Secret):
65        return self.api.delete_namespaced_secret(reference.metadata.name, self.namespace)
66
67    def retrieve(self) -> V1Secret:
68        return self.api.read_namespaced_secret(self.name, self.namespace)
69
70    def update(self, current: V1Secret, reference: V1Secret):
71        return self.api.patch_namespaced_secret(
72            current.metadata.name,
73            self.namespace,
74            reference,
75            field_manager=FIELD_MANAGER,
76        )

def b64string(source: str) -> str: View Source

17def b64string(source: str) -> str:
18    """Base64 Encode string"""
19    return b64encode(source.encode()).decode("utf-8")

Base64 Encode string

class SecretReconciler(authentik.outposts.controllers.k8s.base.KubernetesObjectReconciler[kubernetes.client.models.v1_secret.V1Secret]): View Source

22class SecretReconciler(KubernetesObjectReconciler[V1Secret]):
23    """Kubernetes Secret Reconciler"""
24
25    def __init__(self, controller: KubernetesController) -> None:
26        super().__init__(controller)
27        self.api = CoreV1Api(controller.client)
28
29    @property
30    def noop(self) -> bool:
31        return self.is_embedded
32
33    @staticmethod
34    def reconciler_name() -> str:
35        return "secret"
36
37    def reconcile(self, current: V1Secret, reference: V1Secret):
38        super().reconcile(current, reference)
39        for key in reference.data.keys():
40            if key not in current.data or current.data[key] != reference.data[key]:
41                raise NeedsUpdate()
42
43    def get_reference_object(self) -> V1Secret:
44        """Get deployment object for outpost"""
45        meta = self.get_object_meta(name=self.name)
46        return V1Secret(
47            metadata=meta,
48            data={
49                "authentik_host": b64string(self.controller.outpost.config.authentik_host),
50                "authentik_host_insecure": b64string(
51                    str(self.controller.outpost.config.authentik_host_insecure)
52                ),
53                "token": b64string(self.controller.outpost.token.key),
54                "authentik_host_browser": b64string(
55                    self.controller.outpost.config.authentik_host_browser
56                ),
57            },
58        )
59
60    def create(self, reference: V1Secret):
61        return self.api.create_namespaced_secret(
62            self.namespace, reference, field_manager=FIELD_MANAGER
63        )
64
65    def delete(self, reference: V1Secret):
66        return self.api.delete_namespaced_secret(reference.metadata.name, self.namespace)
67
68    def retrieve(self) -> V1Secret:
69        return self.api.read_namespaced_secret(self.name, self.namespace)
70
71    def update(self, current: V1Secret, reference: V1Secret):
72        return self.api.patch_namespaced_secret(
73            current.metadata.name,
74            self.namespace,
75            reference,
76            field_manager=FIELD_MANAGER,
77        )

Kubernetes Secret Reconciler

api

noop: bool View Source

29    @property
30    def noop(self) -> bool:
31        return self.is_embedded

Return true if this object should not be created/updated/deleted in this cluster

@staticmethod

def reconciler_name() -> str: View Source

33    @staticmethod
34    def reconciler_name() -> str:
35        return "secret"

A name this reconciler is identified by in the configuration

def reconcile( self, current: kubernetes.client.models.v1_secret.V1Secret, reference: kubernetes.client.models.v1_secret.V1Secret): View Source

37    def reconcile(self, current: V1Secret, reference: V1Secret):
38        super().reconcile(current, reference)
39        for key in reference.data.keys():
40            if key not in current.data or current.data[key] != reference.data[key]:
41                raise NeedsUpdate()

Check what operations should be done, should be raised as ReconcileTrigger

def get_reference_object(self) -> kubernetes.client.models.v1_secret.V1Secret: View Source

43    def get_reference_object(self) -> V1Secret:
44        """Get deployment object for outpost"""
45        meta = self.get_object_meta(name=self.name)
46        return V1Secret(
47            metadata=meta,
48            data={
49                "authentik_host": b64string(self.controller.outpost.config.authentik_host),
50                "authentik_host_insecure": b64string(
51                    str(self.controller.outpost.config.authentik_host_insecure)
52                ),
53                "token": b64string(self.controller.outpost.token.key),
54                "authentik_host_browser": b64string(
55                    self.controller.outpost.config.authentik_host_browser
56                ),
57            },
58        )

Get deployment object for outpost

def create(self, reference: kubernetes.client.models.v1_secret.V1Secret): View Source

60    def create(self, reference: V1Secret):
61        return self.api.create_namespaced_secret(
62            self.namespace, reference, field_manager=FIELD_MANAGER
63        )

API Wrapper to create object

def delete(self, reference: kubernetes.client.models.v1_secret.V1Secret): View Source

65    def delete(self, reference: V1Secret):
66        return self.api.delete_namespaced_secret(reference.metadata.name, self.namespace)

API Wrapper to delete object

def retrieve(self) -> kubernetes.client.models.v1_secret.V1Secret: View Source

68    def retrieve(self) -> V1Secret:
69        return self.api.read_namespaced_secret(self.name, self.namespace)

API Wrapper to retrieve object

def update( self, current: kubernetes.client.models.v1_secret.V1Secret, reference: kubernetes.client.models.v1_secret.V1Secret): View Source

71    def update(self, current: V1Secret, reference: V1Secret):
72        return self.api.patch_namespaced_secret(
73            current.metadata.name,
74            self.namespace,
75            reference,
76            field_manager=FIELD_MANAGER,
77        )

API Wrapper to update object

Inherited Members

authentik.outposts.controllers.k8s.base.KubernetesObjectReconciler: controller; namespace; logger; get_patch; is_embedded; name; get_patched_reference_object; up; down; get_object_meta