authentik/authentik/blueprints/migrations/0001_initial.py

# Generated by Django 4.0.6 on 2022-07-31 17:35
import uuid
from glob import glob
from pathlib import Path

import django.contrib.postgres.fields
from dacite.core import from_dict
from django.apps.registry import Apps
from django.conf import settings
from django.db import migrations, models
from django.db.backends.base.schema import BaseDatabaseSchemaEditor
from yaml import load

from authentik.blueprints.v1.labels import LABEL_AUTHENTIK_SYSTEM
from authentik.lib.config import CONFIG


def check_blueprint_v1_file(BlueprintInstance: type["BlueprintInstance"], path: Path):
    """Check if blueprint should be imported"""
    from authentik.blueprints.models import BlueprintInstanceStatus
    from authentik.blueprints.v1.common import BlueprintLoader, BlueprintMetadata
    from authentik.blueprints.v1.labels import LABEL_AUTHENTIK_INSTANTIATE

    with open(path, "r", encoding="utf-8") as blueprint_file:
        raw_blueprint = load(blueprint_file.read(), BlueprintLoader)
        if not raw_blueprint:
            return
        metadata = raw_blueprint.get("metadata", None)
        version = raw_blueprint.get("version", 1)
        if version != 1:
            return
        blueprint_file.seek(0)
    instance: BlueprintInstance = BlueprintInstance.objects.filter(path=path).first()
    rel_path = path.relative_to(Path(CONFIG.y("blueprints_dir")))
    meta = None
    if metadata:
        meta = from_dict(BlueprintMetadata, metadata)
        if meta.labels.get(LABEL_AUTHENTIK_INSTANTIATE, "").lower() == "false":
            return
    if not instance:
        instance = BlueprintInstance(
            name=meta.name if meta else str(rel_path),
            path=str(rel_path),
            context={},
            status=BlueprintInstanceStatus.UNKNOWN,
            enabled=True,
            managed_models=[],
            last_applied_hash="",
            metadata=metadata,
        )
        instance.save()


def migration_blueprint_import(apps: Apps, schema_editor: BaseDatabaseSchemaEditor):
    BlueprintInstance = apps.get_model("authentik_blueprints", "BlueprintInstance")
    Flow = apps.get_model("authentik_flows", "Flow")

    db_alias = schema_editor.connection.alias
    for file in glob(f"{CONFIG.y('blueprints_dir')}/**/*.yaml", recursive=True):
        check_blueprint_v1_file(BlueprintInstance, Path(file))

    for blueprint in BlueprintInstance.objects.using(db_alias).all():
        # If we already have flows (and we should always run before flow migrations)
        # then this is an existing install and we want to disable all blueprints
        if Flow.objects.using(db_alias).all().exists():
            blueprint.enabled = False
        # System blueprints are always enabled
        if blueprint.metadata.get("labels", {}).get(LABEL_AUTHENTIK_SYSTEM, "").lower() == "true":
            blueprint.enabled = True
        blueprint.save()


class Migration(migrations.Migration):
    initial = True

    dependencies = [("authentik_flows", "0001_initial")]

    operations = [
        migrations.CreateModel(
            name="BlueprintInstance",
            fields=[
                ("created", models.DateTimeField(auto_now_add=True)),
                ("last_updated", models.DateTimeField(auto_now=True)),
                (
                    "managed",
                    models.TextField(
                        default=None,
                        help_text=(
                            "Objects which are managed by authentik. These objects are created and"
                            " updated automatically. This is flag only indicates that an object can"
                            " be overwritten by migrations. You can still modify the objects via"
                            " the API, but expect changes to be overwritten in a later update."
                        ),
                        null=True,
                        unique=True,
                        verbose_name="Managed by authentik",
                    ),
                ),
                (
                    "instance_uuid",
                    models.UUIDField(
                        default=uuid.uuid4, editable=False, primary_key=True, serialize=False
                    ),
                ),
                ("name", models.TextField()),
                ("metadata", models.JSONField(default=dict)),
                ("path", models.TextField()),
                ("context", models.JSONField(default=dict)),
                ("last_applied", models.DateTimeField(auto_now=True)),
                ("last_applied_hash", models.TextField()),
                (
                    "status",
                    models.TextField(
                        choices=[
                            ("successful", "Successful"),
                            ("warning", "Warning"),
                            ("error", "Error"),
                            ("orphaned", "Orphaned"),
                            ("unknown", "Unknown"),
                        ],
                        default="unknown",
                    ),
                ),
                ("enabled", models.BooleanField(default=True)),
                (
                    "managed_models",
                    django.contrib.postgres.fields.ArrayField(
                        base_field=models.TextField(), default=list, size=None
                    ),
                ),
            ],
            options={
                "verbose_name": "Blueprint Instance",
                "verbose_name_plural": "Blueprint Instances",
                "unique_together": {("name", "path")},
            },
        ),
        migrations.RunPython(migration_blueprint_import),
    ]