authentik.outposts.controllers.k8s.base

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