АРХ - МЕЖПРОЕКТНАЯ КОММУНИКАЦИЯ: Plane OIDC mapping для существующего пользователя

plane-src/apps/api/plane/db/management/commands/link_authentik_user.py

@@ -0,0 +1,48 @@
+from django.core.management import BaseCommand, CommandError
+
+from plane.db.models import ExternalIdentityLink, User
+
+
+class Command(BaseCommand):
+    help = "Link an existing Plane user to an Authentik OIDC subject"
+
+    def add_arguments(self, parser):
+        parser.add_argument("--email", required=True, help="Existing Plane user email")
+        parser.add_argument("--sub", required=True, help="Authentik OIDC subject")
+        parser.add_argument("--dry-run", action="store_true", help="Validate without writing")
+
+    def handle(self, *args, **options):
+        email = options["email"].strip().lower()
+        subject = options["sub"].strip()
+        dry_run = options["dry_run"]
+
+        if not email or not subject:
+            raise CommandError("--email and --sub are required")
+
+        user = User.objects.filter(email__iexact=email).first()
+        if user is None:
+            raise CommandError(f"Plane user not found: {email}")
+
+        existing_subject_link = ExternalIdentityLink.objects.filter(provider="authentik", subject=subject).first()
+        if existing_subject_link and existing_subject_link.user_id != user.id:
+            raise CommandError(f"Subject is already linked to another Plane user: {existing_subject_link.user.email}")
+
+        existing_user_link = ExternalIdentityLink.objects.filter(provider="authentik", user=user).exclude(subject=subject).first()
+        if existing_user_link:
+            raise CommandError(f"Plane user is already linked to another Authentik subject: {existing_user_link.subject}")
+
+        if dry_run:
+            self.stdout.write(self.style.SUCCESS(f"Dry run OK: {email} can be linked to {subject}"))
+            return
+
+        link, created = ExternalIdentityLink.objects.update_or_create(
+            provider="authentik",
+            subject=subject,
+            defaults={
+                "user": user,
+                "email": email,
+                "status": ExternalIdentityLink.Status.ACTIVE,
+            },
+        )
+        action = "created" if created else "updated"
+        self.stdout.write(self.style.SUCCESS(f"Authentik link {action}: {user.email} -> {link.subject}"))