from os import environ from django.db import transaction from authentik.brands.models import Brand from authentik.common.oauth.constants import SubModes from authentik.core.models import Application, Group, User from authentik.crypto.models import CertificateKeyPair from authentik.flows.models import Flow, FlowStageBinding from authentik.policies.models import PolicyBinding from authentik.providers.oauth2.models import ( ClientTypes, IssuerMode, OAuth2LogoutMethod, OAuth2Provider, RedirectURI, RedirectURIMatchingMode, ScopeMapping, ) from authentik.stages.identification.models import IdentificationStage from authentik.stages.password.models import PasswordStage GROUP_SPECS = [ ("dc-cms:owner", False), ("dc-cms:admin", False), ("dc-cms:editor", False), ("dc-cms:publisher", False), ("dc-cms:site:nodedc:admin", False), ] CMS_APP_SPEC = { "slug": "dc-cms", "name": "DC CMS", "provider_name": "DC CMS OIDC", "client_id_env": "CMS_OIDC_CLIENT_ID", "client_secret_env": "CMS_OIDC_CLIENT_SECRET", "redirect_uri_env": "CMS_OIDC_REDIRECT_URI", "logged_out_redirect_uri_env": "CMS_OIDC_LOGGED_OUT_REDIRECT_URI", "launch_url_env": "CMS_BASE_URL", "launch_url": "http://cms.local.nodedc:8210", "logout_uri_env": "CMS_LOGOUT_URI", "logout_uri": "http://cms.local.nodedc:8210/auth/logout", "groups": [ "dc-cms:owner", "dc-cms:admin", "dc-cms:editor", "dc-cms:publisher", "dc-cms:site:nodedc:admin", ], "description": "Standalone CMS for NODE.DC websites and SEO knowledge bases.", } def required_env(name): value = environ.get(name, "").strip() if not value: raise RuntimeError(f"{name} is required") return value def optional_env(name, default=""): return environ.get(name, default).strip() def ensure_group(name, is_superuser=False): group, _ = Group.objects.get_or_create(name=name) group.is_superuser = is_superuser group.save() return group def ensure_groups(): groups = {} for name, is_superuser in GROUP_SPECS: groups[name] = ensure_group(name, is_superuser) return groups def ensure_admin_user(groups): admin_email = ( environ.get("CMS_BOOTSTRAP_ADMIN_EMAIL", "").strip() or environ.get("AUTHENTIK_BOOTSTRAP_EMAIL", "").strip() ) if not admin_email: return None user = User.objects.filter(email__iexact=admin_email).first() or User.objects.filter( username=admin_email ).first() if user is None: user = User(username=admin_email, email=admin_email, name=admin_email, type="internal") user.username = admin_email user.email = admin_email user.name = environ.get("CMS_BOOTSTRAP_ADMIN_NAME", "DCTOUCH CMS Admin").strip() or admin_email user.is_active = True user.type = "internal" admin_password = ( environ.get("CMS_BOOTSTRAP_ADMIN_PASSWORD", "") or environ.get("AUTHENTIK_BOOTSTRAP_PASSWORD", "") ) if admin_password: user.set_password(admin_password) user.save() authentik_admins, _ = Group.objects.get_or_create(name="authentik Admins") if not authentik_admins.is_superuser: authentik_admins.is_superuser = True authentik_admins.save(update_fields=["is_superuser"]) user.groups.add(authentik_admins) for group in groups.values(): user.groups.add(group) return user def ensure_groups_scope_mapping(): mapping, _ = ScopeMapping.objects.get_or_create( name="DC CMS OAuth Mapping: groups", defaults={ "scope_name": "groups", "description": "Adds Authentik group names to DC CMS OIDC tokens.", "expression": 'return {"groups": [group.name for group in request.user.groups.all()]}', }, ) mapping.scope_name = "groups" mapping.description = "Adds Authentik group names to DC CMS OIDC tokens." mapping.expression = 'return {"groups": [group.name for group in request.user.groups.all()]}' mapping.save() return mapping def ensure_profile_scope_mapping(): expression = """ attributes = request.user.attributes or {} display_name = request.user.name or request.user.username name_parts = display_name.split(" ", 1) avatar_url = attributes.get("picture") or attributes.get("avatar_url") or attributes.get("avatar") return { "name": display_name, "given_name": name_parts[0] if name_parts else display_name, "family_name": name_parts[1] if len(name_parts) > 1 else "", "preferred_username": request.user.username, "nickname": request.user.username, "picture": avatar_url, "avatar_url": avatar_url, } """.strip() mapping, _ = ScopeMapping.objects.get_or_create( name="DC CMS OAuth Mapping: profile context", defaults={ "scope_name": "profile", "description": "Adds normalized DC CMS profile claims to OIDC tokens.", "expression": expression, }, ) mapping.scope_name = "profile" mapping.description = "Adds normalized DC CMS profile claims to OIDC tokens." mapping.expression = expression mapping.save() return mapping def default_scope_mappings(): scope_names = ["openid", "email", "offline_access"] mappings = list(ScopeMapping.objects.filter(scope_name__in=scope_names)) mappings.append(ensure_profile_scope_mapping()) mappings.append(ensure_groups_scope_mapping()) return mappings def ensure_cms_brand(): auth_domain = environ.get("CMS_AUTH_DOMAIN", "cms-auth.local.nodedc").strip() or "cms-auth.local.nodedc" authentication_flow = Flow.objects.get(slug="default-authentication-flow") invalidation_flow = Flow.objects.get(slug="default-invalidation-flow") authentication_flow.name = "DC CMS authentication" authentication_flow.title = "Управление сайтами." authentication_flow.layout = "stacked" authentication_flow.background = "" authentication_flow.save() identification_stage = IdentificationStage.objects.get( name="default-authentication-identification" ) password_stage = PasswordStage.objects.get(name="default-authentication-password") password_stage.allow_show_password = True password_stage.save() identification_stage.user_fields = ["email"] identification_stage.password_stage = password_stage identification_stage.show_matched_user = False identification_stage.enable_remember_me = False identification_stage.save() FlowStageBinding.objects.filter(target=authentication_flow, stage=password_stage).delete() brand = Brand.objects.filter(domain=auth_domain).first() if brand is None: brand = Brand(domain=auth_domain) Brand.objects.exclude(brand_uuid=brand.brand_uuid).update(default=False) brand.default = True brand.domain = auth_domain brand.branding_title = "DC CMS" brand.branding_logo = "" brand.branding_favicon = "" brand.branding_custom_css = "" brand.flow_authentication = authentication_flow brand.flow_invalidation = invalidation_flow brand.attributes = { **(brand.attributes or {}), "settings": { **((brand.attributes or {}).get("settings") or {}), "locale": "ru", "theme": { **(((brand.attributes or {}).get("settings") or {}).get("theme") or {}), "base": "dark", }, }, } brand.save() return brand def ensure_provider(spec, mappings): authorization_flow = Flow.objects.get(slug="default-provider-authorization-implicit-consent") invalidation_flow = Flow.objects.get(slug="default-invalidation-flow") signing_key = ( CertificateKeyPair.objects.filter(name="authentik Self-signed Certificate").first() or CertificateKeyPair.objects.first() ) if signing_key is None: raise RuntimeError("No Authentik CertificateKeyPair exists for OIDC signing") provider = OAuth2Provider.objects.filter(name=spec["provider_name"]).first() if provider is None: provider = OAuth2Provider(name=spec["provider_name"]) provider.name = spec["provider_name"] provider.client_type = ClientTypes.CONFIDENTIAL provider.client_id = required_env(spec["client_id_env"]) provider.client_secret = required_env(spec["client_secret_env"]) redirect_uri_values = [required_env(spec["redirect_uri_env"])] logged_out_redirect_uri = optional_env(spec.get("logged_out_redirect_uri_env", ""), "") if logged_out_redirect_uri: redirect_uri_values.append(logged_out_redirect_uri) provider.redirect_uris = [ RedirectURI(RedirectURIMatchingMode.STRICT, redirect_uri) for redirect_uri in dict.fromkeys(redirect_uri_values) ] provider.logout_uri = optional_env(spec.get("logout_uri_env", ""), spec["logout_uri"]) provider.logout_method = OAuth2LogoutMethod.FRONTCHANNEL provider.include_claims_in_id_token = True provider.sub_mode = SubModes.USER_UUID provider.issuer_mode = IssuerMode.PER_PROVIDER provider.authorization_flow = authorization_flow provider.invalidation_flow = invalidation_flow provider.signing_key = signing_key provider.save() provider.property_mappings.set(mappings) return provider def ensure_application(spec, provider, groups): application = Application.objects.filter(slug=spec["slug"]).first() if application is None: application = Application(slug=spec["slug"], name=spec["name"]) application.name = spec["name"] application.slug = spec["slug"] application.group = "DC CMS" application.provider = provider application.meta_launch_url = optional_env(spec.get("launch_url_env", ""), spec["launch_url"]) application.meta_description = spec["description"] application.meta_publisher = "DCTOUCH" application.open_in_new_tab = False application.policy_engine_mode = "any" application.save() PolicyBinding.objects.filter(target=application).exclude( group__name__in=spec["groups"] ).delete() for order, group_name in enumerate(spec["groups"]): binding = PolicyBinding.objects.filter(target=application, group=groups[group_name]).first() if binding is None: binding = PolicyBinding(target=application, group=groups[group_name]) binding.enabled = True binding.negate = False binding.timeout = 30 binding.failure_result = False binding.order = order binding.save() return application @transaction.atomic def main(): brand = ensure_cms_brand() groups = ensure_groups() user = ensure_admin_user(groups) mappings = default_scope_mappings() provider = ensure_provider(CMS_APP_SPEC, mappings) application = ensure_application(CMS_APP_SPEC, provider, groups) summary = { "groups": list(groups), "admin_user": user.email if user else None, "brand": brand.domain, "application": application.slug, "provider": provider.name, } print(summary) main()