Impersonation system implementation
This commit is contained in:
@@ -1,6 +1,6 @@
|
||||
from django.contrib import admin
|
||||
|
||||
from .models import ApiToken, UserProfile
|
||||
from .models import ApiToken, ImpersonationLog, UserProfile
|
||||
|
||||
|
||||
@admin.register(UserProfile)
|
||||
@@ -12,4 +12,21 @@ class UserProfileAdmin(admin.ModelAdmin):
|
||||
@admin.register(ApiToken)
|
||||
class ApiTokenAdmin(admin.ModelAdmin):
|
||||
list_display = ('user', 'label', 'key_prefix', 'created_at', 'last_used')
|
||||
search_fields = ('user__username', 'label')
|
||||
search_fields = ('user__username', 'label')
|
||||
|
||||
|
||||
@admin.register(ImpersonationLog)
|
||||
class ImpersonationLogAdmin(admin.ModelAdmin):
|
||||
list_display = ('impersonator', 'target', 'reason', 'started_at', 'ended_at')
|
||||
list_filter = ('started_at', 'ended_at')
|
||||
search_fields = ('impersonator__username', 'target__username', 'reason')
|
||||
readonly_fields = ('impersonator', 'target', 'reason', 'started_at', 'ended_at')
|
||||
|
||||
def has_add_permission(self, request):
|
||||
return False
|
||||
|
||||
def has_change_permission(self, request, obj=None):
|
||||
return False
|
||||
|
||||
def has_delete_permission(self, request, obj=None):
|
||||
return request.user.is_superuser
|
||||
@@ -0,0 +1,30 @@
|
||||
# Generated by Django 6.0.3 on 2026-08-04 02:06
|
||||
|
||||
import django.db.models.deletion
|
||||
from django.conf import settings
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('profiles', '0004_alter_userprofile_bio'),
|
||||
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='ImpersonationLog',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('reason', models.CharField(blank=True, default='', max_length=255)),
|
||||
('started_at', models.DateTimeField(auto_now_add=True)),
|
||||
('ended_at', models.DateTimeField(blank=True, null=True)),
|
||||
('impersonator', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='impersonation_sessions', to=settings.AUTH_USER_MODEL)),
|
||||
('target', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='impersonated_sessions', to=settings.AUTH_USER_MODEL)),
|
||||
],
|
||||
options={
|
||||
'ordering': ['-started_at'],
|
||||
},
|
||||
),
|
||||
]
|
||||
@@ -53,4 +53,27 @@ class ApiToken(models.Model):
|
||||
]
|
||||
|
||||
def __str__(self):
|
||||
return f'{self.user.username} - {self.label} ({self.key_prefix}...)'
|
||||
return f'{self.user.username} - {self.label} ({self.key_prefix}...)'
|
||||
|
||||
|
||||
class ImpersonationLog(models.Model):
|
||||
"""Audit trail of staff impersonation sessions (reviewed in Django admin)."""
|
||||
impersonator = models.ForeignKey(
|
||||
settings.AUTH_USER_MODEL,
|
||||
on_delete=models.CASCADE,
|
||||
related_name='impersonation_sessions',
|
||||
)
|
||||
target = models.ForeignKey(
|
||||
settings.AUTH_USER_MODEL,
|
||||
on_delete=models.CASCADE,
|
||||
related_name='impersonated_sessions',
|
||||
)
|
||||
reason = models.CharField(max_length=255, blank=True, default='')
|
||||
started_at = models.DateTimeField(auto_now_add=True)
|
||||
ended_at = models.DateTimeField(null=True, blank=True)
|
||||
|
||||
class Meta:
|
||||
ordering = ['-started_at']
|
||||
|
||||
def __str__(self):
|
||||
return f'{self.impersonator.username} → {self.target.username}'
|
||||
+151
-1
@@ -10,7 +10,7 @@ from django.urls import reverse
|
||||
|
||||
from library.models import FileIndex
|
||||
|
||||
from .models import BIO_MAX_LENGTH, ApiToken, UserProfile
|
||||
from .models import BIO_MAX_LENGTH, ApiToken, ImpersonationLog, UserProfile
|
||||
|
||||
|
||||
def _read_media(stored_path):
|
||||
@@ -423,3 +423,153 @@ class FileRequestTests(MediaTestCase, GatedTestCase):
|
||||
self.gate()
|
||||
resp = self.client.get(self.url)
|
||||
self.assertEqual(resp.status_code, 404)
|
||||
|
||||
|
||||
class ImpersonationTests(GatedTestCase):
|
||||
"""Staff impersonation: switch users without logging out, PK-based
|
||||
staff-to-staff rule, superusers off-limits, gate state preserved."""
|
||||
|
||||
def setUp(self):
|
||||
User = get_user_model()
|
||||
# Created in this order → increasing PKs: boss(1) < junior(2) < normal(3) < top(4).
|
||||
self.boss = User.objects.create_user(username='Boss', password='pw', is_staff=True)
|
||||
UserProfile.objects.get_or_create(user=self.boss)
|
||||
self.junior = User.objects.create_user(username='JuniorStaff', password='pw', is_staff=True)
|
||||
UserProfile.objects.get_or_create(user=self.junior)
|
||||
self.normal = User.objects.create_user(username='NormalUser', password='pw')
|
||||
UserProfile.objects.get_or_create(user=self.normal)
|
||||
self.top = User.objects.create_superuser(username='TopSuper', password='pw')
|
||||
UserProfile.objects.get_or_create(user=self.top)
|
||||
|
||||
def _login(self, user):
|
||||
self.gate()
|
||||
self.client.login(username=user.username, password='pw')
|
||||
|
||||
def _start(self, user, **extra):
|
||||
return self.client.post(
|
||||
reverse('profiles:impersonate_start', args=[user.pk]), data=extra,
|
||||
)
|
||||
|
||||
@property
|
||||
def active_user_id(self):
|
||||
value = self.client.session.get('_auth_user_id')
|
||||
return int(value) if value else None
|
||||
|
||||
def test_users_page_gated(self):
|
||||
resp = self.client.get(reverse('profiles:user_list'))
|
||||
self.assertEqual(resp.status_code, 302)
|
||||
|
||||
def test_users_page_lists_all(self):
|
||||
self.gate()
|
||||
resp = self.client.get(reverse('profiles:user_list'))
|
||||
self.assertEqual(resp.status_code, 200)
|
||||
for name in ('Boss', 'JuniorStaff', 'NormalUser', 'TopSuper'):
|
||||
self.assertContains(resp, name)
|
||||
|
||||
def test_users_page_search(self):
|
||||
self.gate()
|
||||
resp = self.client.get(reverse('profiles:user_list'), {'q': 'Normal'})
|
||||
self.assertContains(resp, 'NormalUser')
|
||||
self.assertNotContains(resp, 'Boss')
|
||||
|
||||
def test_impersonate_requires_post(self):
|
||||
self._login(self.boss)
|
||||
resp = self.client.get(reverse('profiles:impersonate_start', args=[self.normal.pk]))
|
||||
self.assertEqual(resp.status_code, 405)
|
||||
|
||||
def test_non_staff_cannot_impersonate(self):
|
||||
self._login(self.normal)
|
||||
resp = self._start(self.normal)
|
||||
self.assertEqual(resp.status_code, 403)
|
||||
self.assertEqual(ImpersonationLog.objects.count(), 0)
|
||||
|
||||
def test_staff_impersonates_normal_user(self):
|
||||
self._login(self.boss)
|
||||
resp = self._start(self.normal, reason='moderation review')
|
||||
self.assertRedirects(resp, reverse('profiles:user_profile', args=['NormalUser']))
|
||||
self.assertEqual(self.active_user_id, self.normal.pk)
|
||||
self.assertEqual(self.client.session.get('impersonator_id'), self.boss.pk)
|
||||
self.assertTrue(self.client.session.get('authorized'))
|
||||
log = ImpersonationLog.objects.get()
|
||||
self.assertEqual(log.impersonator, self.boss)
|
||||
self.assertEqual(log.target, self.normal)
|
||||
self.assertEqual(log.reason, 'moderation review')
|
||||
self.assertIsNone(log.ended_at)
|
||||
|
||||
def test_gate_preserved_while_impersonating(self):
|
||||
self._login(self.boss)
|
||||
self._start(self.normal)
|
||||
# Follow-up page loads render (200) instead of bouncing to the gate.
|
||||
resp = self.client.get(reverse('profiles:user_profile', args=['NormalUser']))
|
||||
self.assertEqual(resp.status_code, 200)
|
||||
|
||||
def test_cannot_impersonate_self(self):
|
||||
self._login(self.boss)
|
||||
resp = self._start(self.boss)
|
||||
self.assertEqual(resp.status_code, 302)
|
||||
self.assertEqual(self.active_user_id, self.boss.pk)
|
||||
self.assertNotIn('impersonator_id', self.client.session)
|
||||
self.assertEqual(ImpersonationLog.objects.count(), 0)
|
||||
|
||||
def test_older_staff_can_impersonate_newer_staff(self):
|
||||
self._login(self.boss)
|
||||
self._start(self.junior)
|
||||
self.assertEqual(self.active_user_id, self.junior.pk)
|
||||
|
||||
def test_newer_staff_cannot_impersonate_older_staff(self):
|
||||
self._login(self.junior)
|
||||
self._start(self.boss)
|
||||
self.assertEqual(self.active_user_id, self.junior.pk)
|
||||
self.assertNotIn('impersonator_id', self.client.session)
|
||||
self.assertEqual(ImpersonationLog.objects.count(), 0)
|
||||
|
||||
def test_cannot_impersonate_superuser(self):
|
||||
self._login(self.boss)
|
||||
self._start(self.top)
|
||||
self.assertEqual(self.active_user_id, self.boss.pk)
|
||||
self.assertEqual(ImpersonationLog.objects.count(), 0)
|
||||
|
||||
def test_stop_impersonation_restores_original(self):
|
||||
self._login(self.boss)
|
||||
self._start(self.normal)
|
||||
resp = self.client.post(reverse('profiles:impersonate_stop'))
|
||||
self.assertRedirects(resp, reverse('landing:home'))
|
||||
self.assertEqual(self.active_user_id, self.boss.pk)
|
||||
self.assertNotIn('impersonator_id', self.client.session)
|
||||
self.assertTrue(self.client.session.get('authorized'))
|
||||
log = ImpersonationLog.objects.get()
|
||||
self.assertIsNotNone(log.ended_at)
|
||||
|
||||
def test_logout_while_impersonating_stops_instead(self):
|
||||
self._login(self.boss)
|
||||
self._start(self.normal)
|
||||
resp = self.client.post(reverse('profiles:logout'))
|
||||
self.assertRedirects(resp, reverse('landing:home'))
|
||||
# Back on the real account, still logged in, gate still open.
|
||||
self.assertEqual(self.active_user_id, self.boss.pk)
|
||||
self.assertNotIn('impersonator_id', self.client.session)
|
||||
self.assertTrue(self.client.session.get('authorized'))
|
||||
self.assertTrue(ImpersonationLog.objects.get().ended_at)
|
||||
# A second logout is a real logout.
|
||||
self.client.post(reverse('profiles:logout'))
|
||||
self.assertIsNone(self.active_user_id)
|
||||
|
||||
def test_impersonate_button_visibility(self):
|
||||
self._login(self.boss)
|
||||
resp = self.client.get(reverse('profiles:user_profile', args=['NormalUser']))
|
||||
self.assertContains(resp, 'Impersonate')
|
||||
# Self → no button.
|
||||
resp = self.client.get(reverse('profiles:user_profile', args=['Boss']))
|
||||
self.assertNotContains(resp, 'Impersonate')
|
||||
|
||||
def test_impersonate_button_hidden_for_non_staff(self):
|
||||
self._login(self.normal)
|
||||
resp = self.client.get(reverse('profiles:user_profile', args=['NormalUser']))
|
||||
self.assertNotContains(resp, 'Impersonate')
|
||||
|
||||
def test_impersonate_button_hidden_while_impersonating(self):
|
||||
self._login(self.boss)
|
||||
self._start(self.junior) # acting as a staff account now
|
||||
# Even though the impersonated account is staff, no impersonate button.
|
||||
resp = self.client.get(reverse('profiles:user_profile', args=['Boss']))
|
||||
self.assertNotContains(resp, 'Impersonate')
|
||||
|
||||
@@ -14,4 +14,7 @@ urlpatterns = [
|
||||
path('settings/tokens/<int:token_id>/delete/', views.token_delete, name='token_delete'),
|
||||
path('profiles/<str:username>/', views.user_profile, name='user_profile'),
|
||||
path('settings/account/bio/preview/', views.bio_preview, name='bio_preview'),
|
||||
path('users/', views.user_list, name='user_list'),
|
||||
path('impersonate/stop/', views.impersonate_stop, name='impersonate_stop'),
|
||||
path('impersonate/<int:user_id>/', views.impersonate_start, name='impersonate_start'),
|
||||
]
|
||||
@@ -8,14 +8,16 @@ from django.contrib.auth.decorators import login_required
|
||||
from django.contrib.auth.forms import AuthenticationForm, PasswordChangeForm, UserCreationForm
|
||||
from django.core.files.base import ContentFile
|
||||
from django.core.files.storage import default_storage
|
||||
from django.http import JsonResponse
|
||||
from django.core.paginator import Paginator
|
||||
from django.http import HttpResponse, HttpResponseForbidden, JsonResponse
|
||||
from django.shortcuts import get_object_or_404, redirect, render
|
||||
from django.utils import timezone
|
||||
|
||||
from common.markdown import render_markdown
|
||||
from library.models import FileIndex
|
||||
|
||||
from .forms import AvatarForm, BioForm, EmailForm, UsernameForm
|
||||
from .models import ApiToken, UserProfile
|
||||
from .models import ApiToken, ImpersonationLog, UserProfile
|
||||
|
||||
|
||||
def register(request):
|
||||
@@ -47,6 +49,10 @@ def user_login(request):
|
||||
def user_logout(request):
|
||||
if request.method == 'POST':
|
||||
was_authorized = request.session.get('authorized', False)
|
||||
# While impersonating, "logout" returns to the real account instead of
|
||||
# ending the session entirely.
|
||||
if _stop_impersonation(request):
|
||||
return redirect('landing:home')
|
||||
logout(request)
|
||||
# logout() flushes the session (wiping the gate flag). The gate is a
|
||||
# site-wide master password, separate from user accounts — preserve it.
|
||||
@@ -56,6 +62,85 @@ def user_logout(request):
|
||||
return redirect('landing:home')
|
||||
|
||||
|
||||
# ---------- Impersonation (staff only) ----------
|
||||
|
||||
def _stop_impersonation(request):
|
||||
"""Return the session to the original (impersonator) account. Shared by
|
||||
impersonate_stop and user_logout. login() flushes the session, so the gate
|
||||
flag must be restored afterwards."""
|
||||
impersonator_id = request.session.pop('impersonator_id', None)
|
||||
if not impersonator_id:
|
||||
return False
|
||||
original = get_user_model().objects.filter(pk=impersonator_id).first()
|
||||
if original is None:
|
||||
return False
|
||||
ImpersonationLog.objects.filter(
|
||||
impersonator=original, ended_at__isnull=True,
|
||||
).update(ended_at=timezone.now())
|
||||
login(request, original, backend='django.contrib.auth.backends.ModelBackend')
|
||||
request.session['authorized'] = True
|
||||
return True
|
||||
|
||||
|
||||
def user_list(request):
|
||||
"""Browse-style directory of every account on the site."""
|
||||
q = request.GET.get('q', '').strip()
|
||||
users = get_user_model().objects.select_related('userprofile').order_by('username')
|
||||
if q:
|
||||
users = users.filter(username__icontains=q)
|
||||
paginator = Paginator(users, 24)
|
||||
page_obj = paginator.get_page(request.GET.get('page'))
|
||||
return render(request, 'profiles/users.html', {
|
||||
'users': page_obj,
|
||||
'q': q,
|
||||
})
|
||||
|
||||
|
||||
@login_required
|
||||
def impersonate_start(request, user_id):
|
||||
if request.method != 'POST':
|
||||
return HttpResponse(status=405)
|
||||
if not request.user.is_staff:
|
||||
return HttpResponseForbidden('Only staff can impersonate users.')
|
||||
if request.session.get('impersonator_id'):
|
||||
messages.error(request, 'Stop the current impersonation first.')
|
||||
return redirect('profiles:user_profile', username=request.user.username)
|
||||
|
||||
actor = request.user
|
||||
target = get_object_or_404(get_user_model(), pk=user_id)
|
||||
|
||||
if target.pk == actor.pk:
|
||||
messages.error(request, 'You cannot impersonate yourself.')
|
||||
return redirect('profiles:user_profile', username=target.username)
|
||||
|
||||
if target.is_superuser:
|
||||
messages.error(request, 'Superusers cannot be impersonated.')
|
||||
return redirect('profiles:user_profile', username=target.username)
|
||||
|
||||
if target.is_staff and actor.pk >= target.pk:
|
||||
# Staff-to-staff: only an older (lower PK) account may impersonate.
|
||||
messages.error(request, 'You cannot impersonate a more senior staff member.')
|
||||
return redirect('profiles:user_profile', username=target.username)
|
||||
|
||||
reason = request.POST.get('reason', '')[:255]
|
||||
ImpersonationLog.objects.create(impersonator=actor, target=target, reason=reason)
|
||||
|
||||
login(request, target, backend='django.contrib.auth.backends.ModelBackend')
|
||||
request.session['authorized'] = True
|
||||
request.session['impersonator_id'] = actor.pk
|
||||
messages.success(request, f'Now acting as {target.username}.')
|
||||
return redirect('profiles:user_profile', username=target.username)
|
||||
|
||||
|
||||
@login_required
|
||||
def impersonate_stop(request):
|
||||
if request.method != 'POST':
|
||||
return HttpResponse(status=405)
|
||||
_stop_impersonation(request)
|
||||
messages.success(request, 'Impersonation stopped.')
|
||||
return redirect('landing:home')
|
||||
|
||||
|
||||
@login_required
|
||||
def settings_page(request):
|
||||
tokens = request.user.api_tokens.all()
|
||||
@@ -186,6 +271,14 @@ def user_profile(request, username):
|
||||
user = get_object_or_404(User, username__iexact=username)
|
||||
profile, created = UserProfile.objects.get_or_create(user=user)
|
||||
is_owner = request.user.is_authenticated and request.user.pk == user.pk
|
||||
can_impersonate = (
|
||||
request.user.is_authenticated
|
||||
and request.user.is_staff
|
||||
and not request.session.get('impersonator_id')
|
||||
and request.user.pk != user.pk
|
||||
and not user.is_superuser
|
||||
and (not user.is_staff or request.user.pk < user.pk)
|
||||
)
|
||||
bio_form = None
|
||||
if is_owner:
|
||||
if request.method == 'POST' and request.POST.get('action') == 'bio':
|
||||
@@ -201,6 +294,7 @@ def user_profile(request, username):
|
||||
'profile': profile,
|
||||
'profile_user': user,
|
||||
'is_owner': is_owner,
|
||||
'can_impersonate': can_impersonate,
|
||||
'bio_form': bio_form,
|
||||
'projects': user.projects.select_related('owner', 'thumbnail').prefetch_related(
|
||||
'tag_links__tag__category', 'versions',
|
||||
|
||||
Reference in New Issue
Block a user