authentik.outposts.controllers.k8s.deployment

  1"""Kubernetes Deployment Reconciler"""
  2
  3from typing import TYPE_CHECKING
  4
  5from django.utils.text import slugify
  6from kubernetes.client import (
  7    AppsV1Api,
  8    V1Capabilities,
  9    V1Container,
 10    V1ContainerPort,
 11    V1Deployment,
 12    V1DeploymentSpec,
 13    V1EnvVar,
 14    V1EnvVarSource,
 15    V1LabelSelector,
 16    V1ObjectMeta,
 17    V1ObjectReference,
 18    V1PodSecurityContext,
 19    V1PodSpec,
 20    V1PodTemplateSpec,
 21    V1SeccompProfile,
 22    V1SecretKeySelector,
 23    V1SecurityContext,
 24)
 25
 26from authentik import authentik_full_version
 27from authentik.outposts.controllers.base import FIELD_MANAGER
 28from authentik.outposts.controllers.k8s.base import KubernetesObjectReconciler
 29from authentik.outposts.controllers.k8s.triggers import NeedsUpdate
 30from authentik.outposts.controllers.k8s.utils import compare_ports
 31from authentik.outposts.models import Outpost
 32
 33if TYPE_CHECKING:
 34    from authentik.outposts.controllers.kubernetes import KubernetesController
 35
 36
 37class DeploymentReconciler(KubernetesObjectReconciler[V1Deployment]):
 38    """Kubernetes Deployment Reconciler"""
 39
 40    outpost: Outpost
 41
 42    def __init__(self, controller: KubernetesController) -> None:
 43        super().__init__(controller)
 44        self.api = AppsV1Api(controller.client)
 45        self.outpost = self.controller.outpost
 46
 47    @property
 48    def noop(self) -> bool:
 49        return self.is_embedded
 50
 51    @staticmethod
 52    def reconciler_name() -> str:
 53        return "deployment"
 54
 55    def reconcile(self, current: V1Deployment, reference: V1Deployment):
 56        compare_ports(
 57            current.spec.template.spec.containers[0].ports,
 58            reference.spec.template.spec.containers[0].ports,
 59        )
 60        if current.spec.replicas != reference.spec.replicas:
 61            raise NeedsUpdate()
 62        if (
 63            current.spec.template.spec.containers[0].image
 64            != reference.spec.template.spec.containers[0].image
 65        ):
 66            raise NeedsUpdate()
 67        super().reconcile(current, reference)
 68
 69    def get_pod_meta(self, **kwargs) -> dict[str, str]:
 70        """Get common object metadata"""
 71        kwargs.update(
 72            {
 73                "app.kubernetes.io/name": f"authentik-outpost-{self.outpost.type}",
 74                "app.kubernetes.io/managed-by": "goauthentik.io",
 75                "goauthentik.io/outpost-uuid": self.controller.outpost.uuid.hex,
 76                "goauthentik.io/outpost-name": slugify(self.controller.outpost.name),
 77                "goauthentik.io/outpost-type": str(self.controller.outpost.type),
 78            }
 79        )
 80        return kwargs
 81
 82    def get_reference_object(self) -> V1Deployment:
 83        """Get deployment object for outpost"""
 84        # Generate V1ContainerPort objects
 85        container_ports = []
 86        for port in self.controller.deployment_ports:
 87            container_ports.append(
 88                V1ContainerPort(
 89                    container_port=port.inner_port or port.port,
 90                    name=port.name,
 91                    protocol=port.protocol.upper(),
 92                )
 93            )
 94        meta = self.get_object_meta(name=self.name)
 95        image_name = self.controller.get_container_image()
 96        image_pull_secrets = self.outpost.config.kubernetes_image_pull_secrets
 97        version = authentik_full_version().replace("+", "-")
 98        return V1Deployment(
 99            metadata=meta,
100            spec=V1DeploymentSpec(
101                replicas=self.outpost.config.kubernetes_replicas,
102                selector=V1LabelSelector(match_labels=self.get_pod_meta()),
103                template=V1PodTemplateSpec(
104                    metadata=V1ObjectMeta(
105                        labels=self.get_pod_meta(
106                            **{
107                                # Support istio-specific labels, but also use the standard k8s
108                                # recommendations
109                                "app.kubernetes.io/version": version,
110                                "app": "authentik-outpost",
111                                "version": version,
112                            }
113                        )
114                    ),
115                    spec=V1PodSpec(
116                        image_pull_secrets=[
117                            V1ObjectReference(name=secret) for secret in image_pull_secrets
118                        ],
119                        security_context=V1PodSecurityContext(
120                            seccomp_profile=V1SeccompProfile(
121                                type="RuntimeDefault",
122                            ),
123                        ),
124                        containers=[
125                            V1Container(
126                                name=str(self.outpost.type),
127                                image=image_name,
128                                ports=container_ports,
129                                env=[
130                                    V1EnvVar(
131                                        name="AUTHENTIK_HOST",
132                                        value_from=V1EnvVarSource(
133                                            secret_key_ref=V1SecretKeySelector(
134                                                name=self.name,
135                                                key="authentik_host",
136                                            )
137                                        ),
138                                    ),
139                                    V1EnvVar(
140                                        name="AUTHENTIK_HOST_BROWSER",
141                                        value_from=V1EnvVarSource(
142                                            secret_key_ref=V1SecretKeySelector(
143                                                name=self.name,
144                                                key="authentik_host_browser",
145                                            )
146                                        ),
147                                    ),
148                                    V1EnvVar(
149                                        name="AUTHENTIK_TOKEN",
150                                        value_from=V1EnvVarSource(
151                                            secret_key_ref=V1SecretKeySelector(
152                                                name=self.name,
153                                                key="token",
154                                            )
155                                        ),
156                                    ),
157                                    V1EnvVar(
158                                        name="AUTHENTIK_INSECURE",
159                                        value_from=V1EnvVarSource(
160                                            secret_key_ref=V1SecretKeySelector(
161                                                name=self.name,
162                                                key="authentik_host_insecure",
163                                            )
164                                        ),
165                                    ),
166                                ],
167                                security_context=V1SecurityContext(
168                                    run_as_non_root=True,
169                                    allow_privilege_escalation=False,
170                                    capabilities=V1Capabilities(
171                                        drop=["ALL"],
172                                    ),
173                                ),
174                            )
175                        ],
176                    ),
177                ),
178            ),
179        )
180
181    def create(self, reference: V1Deployment):
182        return self.api.create_namespaced_deployment(
183            self.namespace, reference, field_manager=FIELD_MANAGER
184        )
185
186    def delete(self, reference: V1Deployment):
187        return self.api.delete_namespaced_deployment(reference.metadata.name, self.namespace)
188
189    def retrieve(self) -> V1Deployment:
190        return self.api.read_namespaced_deployment(self.name, self.namespace)
191
192    def update(self, current: V1Deployment, reference: V1Deployment):
193        return self.api.patch_namespaced_deployment(
194            current.metadata.name,
195            self.namespace,
196            reference,
197            field_manager=FIELD_MANAGER,
198        )