authentik.outposts.controllers.k8s.base

  1"""Base Kubernetes Reconciler"""
  2
  3import re
  4import ssl
  5from dataclasses import asdict
  6from json import dumps
  7from typing import TYPE_CHECKING, TypeVar
  8
  9import urllib3
 10from dacite.core import from_dict
 11from django.http import HttpResponseNotFound
 12from django.utils.text import slugify
 13from jsonpatch import JsonPatchConflict, JsonPatchException, JsonPatchTestFailed, apply_patch
 14from kubernetes.client import ApiClient, V1ObjectMeta
 15from kubernetes.client.exceptions import ApiException, OpenApiException
 16from kubernetes.client.models.v1_deployment import V1Deployment
 17from kubernetes.client.models.v1_pod import V1Pod
 18from requests import Response
 19from structlog.stdlib import get_logger
 20from urllib3.exceptions import HTTPError
 21
 22from authentik import authentik_version
 23from authentik.outposts.apps import MANAGED_OUTPOST
 24from authentik.outposts.controllers.base import ControllerException
 25from authentik.outposts.controllers.k8s.triggers import NeedsRecreate, NeedsUpdate
 26
 27if TYPE_CHECKING:
 28    from authentik.outposts.controllers.kubernetes import KubernetesController
 29
 30T = TypeVar("T", V1Pod, V1Deployment)
 31
 32
 33def get_version() -> str:
 34    """Wrapper for authentik_version() to make testing easier"""
 35    return authentik_version()
 36
 37
 38class KubernetesObjectReconciler[T]:
 39    """Base Kubernetes Reconciler, handles the basic logic."""
 40
 41    controller: KubernetesController
 42
 43    def __init__(self, controller: KubernetesController):
 44        self.controller = controller
 45        self.namespace = controller.outpost.config.kubernetes_namespace
 46        self.kubernetes_disable_x509_strict = (
 47            controller.outpost.config.kubernetes_disable_x509_strict
 48        )
 49        self.logger = get_logger().bind(type=self.__class__.__name__)
 50        self.api_client = self.k8s_client()
 51
 52    def get_patch(self):
 53        """Get any patches that apply to this CRD"""
 54        patches = self.controller.outpost.config.kubernetes_json_patches
 55        if not patches:
 56            return None
 57        return patches.get(self.reconciler_name(), None)
 58
 59    @property
 60    def is_embedded(self) -> bool:
 61        """Return true if the current outpost is embedded"""
 62        return self.controller.outpost.managed == MANAGED_OUTPOST
 63
 64    @staticmethod
 65    def reconciler_name() -> str:
 66        """A name this reconciler is identified by in the configuration"""
 67        raise NotImplementedError
 68
 69    @property
 70    def noop(self) -> bool:
 71        """Return true if this object should not be created/updated/deleted in this cluster"""
 72        return False
 73
 74    @property
 75    def name(self) -> str:
 76        """Get the name of the object this reconciler manages"""
 77
 78        base_name = (
 79            self.controller.outpost.config.object_naming_template
 80            % {
 81                "name": slugify(self.controller.outpost.name),
 82                "uuid": self.controller.outpost.uuid.hex,
 83            }
 84        ).lower()
 85
 86        formatted = slugify(base_name)
 87        formatted = re.sub(r"[^a-z0-9-]", "-", formatted)
 88        formatted = re.sub(r"-+", "-", formatted)
 89        formatted = formatted[:63]
 90
 91        if not formatted:
 92            formatted = f"outpost-{self.controller.outpost.uuid.hex}"[:63]
 93
 94        return formatted
 95
 96    def get_patched_reference_object(self) -> T:
 97        """Get patched reference object"""
 98        reference = self.get_reference_object()
 99        patch = self.get_patch()
100        try:
101            json = self.api_client.sanitize_for_serialization(reference)
102        # Custom objects will not be known to the clients openapi types
103        except AttributeError:
104            json = asdict(reference)
105        try:
106            ref = json
107            if patch is not None:
108                ref = apply_patch(json, patch)
109        except (JsonPatchException, JsonPatchConflict, JsonPatchTestFailed) as exc:
110            raise ControllerException(f"JSON Patch failed: {exc}") from exc
111        mock_response = Response()
112        mock_response.data = dumps(ref)
113
114        try:
115            result = self.api_client.deserialize(mock_response, reference.__class__.__name__)
116        # Custom objects will not be known to the clients openapi types
117        except AttributeError:
118            result = from_dict(reference.__class__, data=ref)
119
120        return result
121
122    def up(self):
123        """Create object if it doesn't exist, update if needed or recreate if needed."""
124        current = None
125        if self.noop:
126            self.logger.debug("Object is noop")
127            return
128        reference = self.get_patched_reference_object()
129        try:
130            try:
131                current = self.retrieve()
132            except (OpenApiException, HTTPError) as exc:
133                if isinstance(exc, ApiException) and exc.status == HttpResponseNotFound.status_code:
134                    self.logger.debug("Failed to get current, triggering recreate")
135                    raise NeedsRecreate from exc
136                self.logger.debug("Other unhandled error", exc=exc)
137                raise exc
138            self.reconcile(current, reference)
139        except NeedsUpdate:
140            try:
141                self.update(current, reference)
142                self.logger.debug("Updating")
143            except (OpenApiException, HTTPError) as exc:
144                if isinstance(exc, ApiException) and exc.status == 422:  # noqa: PLR2004
145                    self.logger.debug("Failed to update current, triggering re-create")
146                    self._recreate(current=current, reference=reference)
147                    return
148                self.logger.debug("Other unhandled error", exc=exc)
149                raise exc
150        except NeedsRecreate:
151            self._recreate(current=current, reference=reference)
152        else:
153            self.logger.debug("Object is up-to-date.")
154
155    def _recreate(self, reference: T, current: T | None = None):
156        """Recreate object"""
157        self.logger.debug("Recreate requested")
158        if current:
159            self.logger.debug("Deleted old")
160            self.delete(current)
161        else:
162            self.logger.debug("No old found, creating")
163        self.logger.debug("Creating")
164        self.create(reference)
165
166    def down(self):
167        """Delete object if found"""
168        if self.noop:
169            self.logger.debug("Object is noop")
170            return
171        try:
172            current = self.retrieve()
173            self.delete(current)
174            self.logger.debug("Removing")
175        except (OpenApiException, HTTPError) as exc:
176            if isinstance(exc, ApiException) and exc.status == HttpResponseNotFound.status_code:
177                self.logger.debug("Failed to get current, assuming non-existent")
178                return
179            self.logger.debug("Other unhandled error", exc=exc)
180            raise exc
181
182    def get_reference_object(self) -> T:
183        """Return object as it should be"""
184        raise NotImplementedError
185
186    def reconcile(self, current: T, reference: T):
187        """Check what operations should be done, should be raised as
188        ReconcileTrigger"""
189        if current.metadata.labels != reference.metadata.labels:
190            raise NeedsUpdate()
191
192        patch = self.get_patch()
193        if patch is not None:
194            try:
195                current_json = self.api_client.sanitize_for_serialization(current)
196            except AttributeError:
197                current_json = asdict(current)
198            try:
199                if apply_patch(current_json, patch) != current_json:
200                    raise NeedsUpdate()
201            except (JsonPatchException, JsonPatchConflict, JsonPatchTestFailed) as exc:
202                raise ControllerException(f"JSON Patch failed: {exc}") from exc
203
204    def create(self, reference: T):
205        """API Wrapper to create object"""
206        raise NotImplementedError
207
208    def retrieve(self) -> T:
209        """API Wrapper to retrieve object"""
210        raise NotImplementedError
211
212    def delete(self, reference: T):
213        """API Wrapper to delete object"""
214        raise NotImplementedError
215
216    def update(self, current: T, reference: T):
217        """API Wrapper to update object"""
218        raise NotImplementedError
219
220    def get_object_meta(self, **kwargs) -> V1ObjectMeta:
221        """Get common object metadata"""
222        return V1ObjectMeta(
223            namespace=self.namespace,
224            labels={
225                "app.kubernetes.io/instance": slugify(self.controller.outpost.name),
226                "app.kubernetes.io/managed-by": "goauthentik.io",
227                "app.kubernetes.io/name": f"authentik-{self.controller.outpost.type.lower()}",
228                "app.kubernetes.io/version": get_version().replace("+", "-"),
229                "goauthentik.io/outpost-name": slugify(self.controller.outpost.name),
230                "goauthentik.io/outpost-type": str(self.controller.outpost.type),
231                "goauthentik.io/outpost-uuid": self.controller.outpost.uuid.hex,
232            },
233            **kwargs,
234        )
235
236    def k8s_client(self) -> ApiClient:
237        """Get Kubernetes API client"""
238        api_client = ApiClient()
239        if self.kubernetes_disable_x509_strict:
240            self.logger.warning("Disabling strict X.509 certificate verification")
241            # Relax OpenSSL TLS validation to support legacy root CA certificates
242            # (e.g. from Kubernetes <= 1.16) which may not satisfy the stricter
243            # VERIFY_X509_STRICT flags enforced by default in Python 3.13+.
244            # See https://github.com/kubernetes-client/python/issues/2394
245            ctx = ssl.create_default_context()
246            ctx.verify_flags = ctx.verify_flags & ~ssl.VERIFY_X509_STRICT
247
248            # We need to recreate the pool manager with the new SSL context
249            # We try to preserve existing pool manager arguments
250            pool_args = api_client.rest_client.pool_manager.connection_pool_kw
251
252            api_client.rest_client.pool_manager = urllib3.PoolManager(
253                num_pools=4,
254                ssl_context=ctx,
255                **pool_args,
256            )
257        return api_client