139 lines
4.3 KiB
Python
139 lines
4.3 KiB
Python
from django.contrib import admin, messages
|
|
from django.contrib.auth import admin as auth_admin, get_user_model
|
|
from django.utils.translation import gettext_lazy as _
|
|
|
|
from vbv_lernwelt.course.models import CourseSessionUser
|
|
from vbv_lernwelt.importer.services import update_user_json_data
|
|
from vbv_lernwelt.sso.exceptions import MyVbvKeycloakDeleteError, MyVbvKeycloakPostError
|
|
from vbv_lernwelt.sso.models import SsoSyncError, SsoUser
|
|
from vbv_lernwelt.sso.role_sync.services import create_user, sync_roles_for_user
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
def create_sso_user_from_admin(user: User, request):
|
|
try:
|
|
sso_data = {"intermediate_sso_id": create_user(user)}
|
|
update_user_json_data(user, sso_data)
|
|
user.save()
|
|
messages.add_message(
|
|
request, messages.SUCCESS, f"Der Bentuzer wurde in Keycloak erstellt."
|
|
)
|
|
except MyVbvKeycloakPostError as e:
|
|
additional_data = getattr(e, "additional_data", {})
|
|
SsoSyncError.objects.create(
|
|
user=user, action=SsoSyncError.Action.CREATE, data=additional_data
|
|
)
|
|
messages.add_message(
|
|
request,
|
|
messages.WARNING,
|
|
f"Der Benutzer ({e}) konnte nicht in Keycloak erstellt werden.",
|
|
)
|
|
|
|
|
|
def sync_sso_roles_from_admin(user: User, request):
|
|
course_roles = [
|
|
(csu.course_session.course.slug, csu.role)
|
|
for csu in CourseSessionUser.objects.filter(user=user)
|
|
]
|
|
try:
|
|
sync_roles_for_user(user, course_roles)
|
|
messages.add_message(
|
|
request, messages.SUCCESS, f"Die Daten wurden mit Keycloak synchronisiert."
|
|
)
|
|
except MyVbvKeycloakDeleteError as e:
|
|
additional_data = getattr(e, "additional_data", {})
|
|
SsoSyncError.objects.create(
|
|
user=user, action=SsoSyncError.Action.REMOVE, data=additional_data
|
|
)
|
|
messages.add_message(
|
|
request,
|
|
messages.WARNING,
|
|
f"Die bestehenden Rollen für Benutzer ({e}) konnten in Keycloak nicht gelöscht werden.",
|
|
)
|
|
except MyVbvKeycloakPostError as e:
|
|
additional_data = getattr(e, "additional_data", {})
|
|
SsoSyncError.objects.create(
|
|
user=user, action=SsoSyncError.Action.ADD, data=additional_data
|
|
)
|
|
messages.add_message(
|
|
request,
|
|
messages.WARNING,
|
|
f"Die neuen Rollen für Benutzer ({e}) konnten in Keycloak nicht erstellt werden.",
|
|
)
|
|
|
|
|
|
@admin.action(description="KEYCLOAK: Sync SSO Roles")
|
|
def sync_sso_roles(modeladmin, request, queryset):
|
|
for user in queryset:
|
|
sync_sso_roles_from_admin(user, request)
|
|
|
|
|
|
@admin.action(description="KEYCLOAK: Create User")
|
|
def create_sso_user(modeladmin, request, queryset):
|
|
for user in queryset:
|
|
create_sso_user_from_admin(user, request)
|
|
|
|
|
|
@admin.register(SsoUser)
|
|
class SsoUserAdmin(auth_admin.UserAdmin):
|
|
fieldsets = (
|
|
(
|
|
_("Personal info"),
|
|
{"fields": ("first_name", "last_name", "email", "sso_id")},
|
|
),
|
|
(_("Additional data"), {"fields": ("additional_json_data",)}),
|
|
)
|
|
list_display = [
|
|
"username",
|
|
"first_name",
|
|
"last_name",
|
|
"sso_id",
|
|
"intermedia_sso_id",
|
|
]
|
|
search_fields = ["first_name", "last_name", "email", "username", "sso_id"]
|
|
actions = [sync_sso_roles, create_sso_user]
|
|
|
|
# Make fields read-only
|
|
readonly_fields = (
|
|
"username",
|
|
"password",
|
|
"first_name",
|
|
"last_name",
|
|
"email",
|
|
"additional_json_data",
|
|
)
|
|
|
|
# Disable delete action
|
|
def has_delete_permission(self, request, obj=None):
|
|
return False
|
|
|
|
def get_actions(self, request):
|
|
actions = super().get_actions(request)
|
|
if "delete_selected" in actions:
|
|
del actions["delete_selected"]
|
|
return actions
|
|
|
|
def intermedia_sso_id(self, obj):
|
|
return obj.additional_json_data.get("intermediate_sso_id", "")
|
|
|
|
intermedia_sso_id.short_description = "Keycloak SSO ID"
|
|
|
|
|
|
@admin.register(SsoSyncError)
|
|
class SsoSyncErrorAdmin(admin.ModelAdmin):
|
|
list_display = [
|
|
"created_at",
|
|
"user",
|
|
"action",
|
|
"data",
|
|
]
|
|
raw_id_fields = [
|
|
"user",
|
|
]
|
|
search_fields = [
|
|
"user.email",
|
|
"user.username",
|
|
]
|
|
list_filter = ("action",)
|