authentik.blueprints.models

  1"""blueprint models"""
  2
  3from pathlib import Path
  4from uuid import uuid4
  5
  6from django.contrib.contenttypes.fields import GenericRelation
  7from django.contrib.postgres.fields import ArrayField
  8from django.db import models
  9from django.utils.translation import gettext_lazy as _
 10from rest_framework.serializers import Serializer
 11from structlog import get_logger
 12
 13from authentik.blueprints.v1.oci import OCI_PREFIX, BlueprintOCIClient, OCIException
 14from authentik.lib.config import CONFIG
 15from authentik.lib.models import CreatedUpdatedModel, SerializerModel
 16from authentik.lib.sentry import SentryIgnoredException
 17
 18LOGGER = get_logger()
 19
 20
 21class BlueprintRetrievalFailed(SentryIgnoredException):
 22    """Error raised when we are unable to fetch the blueprint contents, whether it be HTTP files
 23    not being accessible or local files not being readable"""
 24
 25
 26class ManagedModel(models.Model):
 27    """Model that can be managed by authentik exclusively"""
 28
 29    managed = models.TextField(
 30        default=None,
 31        null=True,
 32        verbose_name=_("Managed by authentik"),
 33        help_text=_(
 34            "Objects that are managed by authentik. These objects are created and updated "
 35            "automatically. This flag only indicates that an object can be overwritten by "
 36            "migrations. You can still modify the objects via the API, but expect changes "
 37            "to be overwritten in a later update."
 38        ),
 39        unique=True,
 40    )
 41
 42    class Meta:
 43        abstract = True
 44
 45
 46class BlueprintInstanceStatus(models.TextChoices):
 47    """Instance status"""
 48
 49    SUCCESSFUL = "successful"
 50    WARNING = "warning"
 51    ERROR = "error"
 52    ORPHANED = "orphaned"
 53    UNKNOWN = "unknown"
 54
 55
 56class BlueprintInstance(SerializerModel, ManagedModel, CreatedUpdatedModel):
 57    """Instance of a single blueprint. Can be parameterized via context attribute when
 58    blueprint in `path` has inputs."""
 59
 60    instance_uuid = models.UUIDField(primary_key=True, editable=False, default=uuid4)
 61
 62    name = models.TextField(unique=True)
 63    metadata = models.JSONField(default=dict)
 64    path = models.TextField(default="", blank=True)
 65    content = models.TextField(default="", blank=True)
 66    context = models.JSONField(default=dict)
 67    last_applied = models.DateTimeField(auto_now=True)
 68    last_applied_hash = models.TextField()
 69    status = models.TextField(
 70        choices=BlueprintInstanceStatus.choices, default=BlueprintInstanceStatus.UNKNOWN
 71    )
 72    enabled = models.BooleanField(default=True)
 73    managed_models = ArrayField(models.TextField(), default=list)
 74
 75    # Manual link to tasks instead of using TasksModel because of loop imports
 76    tasks = GenericRelation(
 77        "authentik_tasks.Task",
 78        content_type_field="rel_obj_content_type",
 79        object_id_field="rel_obj_id",
 80    )
 81
 82    class Meta:
 83        verbose_name = _("Blueprint Instance")
 84        verbose_name_plural = _("Blueprint Instances")
 85        unique_together = (
 86            (
 87                "name",
 88                "path",
 89            ),
 90        )
 91
 92    def __str__(self) -> str:
 93        return f"Blueprint Instance {self.name}"
 94
 95    def retrieve_oci(self) -> str:
 96        """Get blueprint from an OCI registry"""
 97        client = BlueprintOCIClient(self.path.replace(OCI_PREFIX, "https://"))
 98        try:
 99            manifests = client.fetch_manifests()
100            return client.fetch_blobs(manifests)
101        except OCIException as exc:
102            raise BlueprintRetrievalFailed(exc) from exc
103
104    def retrieve_file(self) -> str:
105        """Get blueprint from path"""
106        try:
107            base = Path(CONFIG.get("blueprints_dir"))
108            full_path = base.joinpath(Path(self.path)).resolve()
109            if not str(full_path).startswith(str(base.resolve())):
110                raise BlueprintRetrievalFailed("Invalid blueprint path")
111            with full_path.open("r", encoding="utf-8") as _file:
112                return _file.read()
113        except OSError as exc:
114            raise BlueprintRetrievalFailed(exc) from exc
115
116    def retrieve(self) -> str:
117        """Retrieve blueprint contents"""
118        if self.path.startswith(OCI_PREFIX):
119            return self.retrieve_oci()
120        if self.path != "":
121            return self.retrieve_file()
122        return self.content
123
124    @property
125    def serializer(self) -> Serializer:
126        from authentik.blueprints.api import BlueprintInstanceSerializer
127
128        return BlueprintInstanceSerializer