311 lines
12 KiB
Python
311 lines
12 KiB
Python
import hashlib
|
|
import io
|
|
import secrets
|
|
|
|
from django.contrib import messages
|
|
from django.contrib.auth import authenticate, get_user_model, login, logout, update_session_auth_hash
|
|
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.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, ImpersonationLog, UserProfile
|
|
|
|
|
|
def register(request):
|
|
if request.method == 'POST':
|
|
form = UserCreationForm(request.POST)
|
|
if form.is_valid():
|
|
user = form.save()
|
|
UserProfile.objects.get_or_create(user=user)
|
|
login(request, user)
|
|
messages.success(request, 'Registration successful.')
|
|
return redirect('landing:home')
|
|
else:
|
|
form = UserCreationForm()
|
|
return render(request, 'profiles/register.html', {'form': form})
|
|
|
|
|
|
def user_login(request):
|
|
if request.method == 'POST':
|
|
form = AuthenticationForm(request, data=request.POST)
|
|
if form.is_valid():
|
|
user = form.get_user()
|
|
login(request, user)
|
|
return redirect('landing:home')
|
|
else:
|
|
form = AuthenticationForm(request)
|
|
return render(request, 'profiles/login.html', {'form': form})
|
|
|
|
|
|
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.
|
|
if was_authorized:
|
|
request.session['authorized'] = True
|
|
return redirect('landing:home')
|
|
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()
|
|
context = {'tokens': tokens}
|
|
return render(request, 'profiles/settings.html', context)
|
|
|
|
|
|
@login_required
|
|
def token_create(request):
|
|
if request.method == 'POST':
|
|
label = request.POST.get('label', '').strip() or 'API token'
|
|
token = secrets.token_urlsafe(48)
|
|
ApiToken.objects.create(
|
|
user=request.user,
|
|
token=token,
|
|
key_prefix=token[:8],
|
|
label=label,
|
|
)
|
|
messages.success(request, 'Token created.')
|
|
return redirect('profiles:settings')
|
|
|
|
|
|
@login_required
|
|
def token_delete(request, token_id):
|
|
ApiToken.objects.filter(id=token_id, user=request.user).delete()
|
|
return redirect('profiles:settings')
|
|
|
|
|
|
def _jpeg_bytes(uploaded):
|
|
"""Return JPEG bytes for an uploaded image.
|
|
|
|
Already-JPEG uploads are stored as-is; anything else is re-encoded
|
|
to JPEG via Pillow.
|
|
"""
|
|
import PIL.Image
|
|
|
|
image = PIL.Image.open(uploaded)
|
|
if image.format == 'JPEG':
|
|
uploaded.seek(0)
|
|
return uploaded.read()
|
|
|
|
if image.mode not in ('RGB', 'L'):
|
|
image = image.convert('RGB')
|
|
buffer = io.BytesIO()
|
|
image.save(buffer, format='JPEG', quality=88)
|
|
return buffer.getvalue()
|
|
|
|
|
|
def _store_avatar(user, uploaded):
|
|
"""Persist an avatar under /avatars/user_<id>/picture.jpeg, index it in
|
|
FileIndex, and return the FileIndex row (replacing any previous avatar)."""
|
|
profile, _ = UserProfile.objects.get_or_create(user=user)
|
|
|
|
old_index = profile.avatar
|
|
if old_index is not None:
|
|
if old_index.stored_path and default_storage.exists(old_index.stored_path):
|
|
default_storage.delete(old_index.stored_path)
|
|
FileIndex.objects.filter(pk=old_index.pk).delete()
|
|
profile.avatar = None
|
|
|
|
data = _jpeg_bytes(uploaded)
|
|
stored_path = f'avatars/user_{user.pk}/picture.jpeg'
|
|
actual_path = default_storage.save(stored_path, ContentFile(data))
|
|
file_index = FileIndex.objects.create(
|
|
owner=user,
|
|
kind='avatar',
|
|
stored_path=actual_path,
|
|
original_filename=getattr(uploaded, 'name', '') or 'picture.jpeg',
|
|
content_type='image/jpeg',
|
|
size=len(data),
|
|
md5=hashlib.md5(data).hexdigest(),
|
|
)
|
|
profile.avatar = file_index
|
|
profile.save(update_fields=['avatar'])
|
|
return file_index
|
|
|
|
|
|
@login_required
|
|
def account_page(request):
|
|
username_form = UsernameForm(instance=request.user)
|
|
email_form = EmailForm(instance=request.user)
|
|
password_form = PasswordChangeForm(request.user)
|
|
avatar_form = AvatarForm()
|
|
|
|
if request.method == 'POST':
|
|
action = request.POST.get('action')
|
|
if action == 'username':
|
|
username_form = UsernameForm(request.POST, instance=request.user)
|
|
if username_form.is_valid():
|
|
username_form.save()
|
|
messages.success(request, 'Username updated.')
|
|
return redirect('profiles:account')
|
|
messages.error(request, 'Could not update username.')
|
|
elif action == 'email':
|
|
email_form = EmailForm(request.POST, instance=request.user)
|
|
if email_form.is_valid():
|
|
email_form.save()
|
|
messages.success(request, 'Email updated.')
|
|
return redirect('profiles:account')
|
|
messages.error(request, 'Could not update email.')
|
|
elif action == 'password':
|
|
password_form = PasswordChangeForm(request.user, request.POST)
|
|
if password_form.is_valid():
|
|
user = password_form.save()
|
|
update_session_auth_hash(request, user)
|
|
messages.success(request, 'Password updated.')
|
|
return redirect('profiles:account')
|
|
messages.error(request, 'Could not update password.')
|
|
elif action == 'avatar':
|
|
avatar_form = AvatarForm(request.POST, request.FILES)
|
|
if avatar_form.is_valid():
|
|
_store_avatar(request.user, avatar_form.cleaned_data['avatar'])
|
|
messages.success(request, 'Profile picture updated.')
|
|
return redirect('profiles:account')
|
|
messages.error(request, 'Could not update profile picture.')
|
|
|
|
context = {
|
|
'username_form': username_form,
|
|
'email_form': email_form,
|
|
'password_form': password_form,
|
|
'avatar_form': avatar_form,
|
|
}
|
|
return render(request, 'profiles/account.html', context)
|
|
|
|
|
|
def user_profile(request, username):
|
|
User = get_user_model()
|
|
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':
|
|
bio_form = BioForm(request.POST, instance=profile)
|
|
if bio_form.is_valid():
|
|
bio_form.save()
|
|
messages.success(request, 'Bio updated.')
|
|
return redirect('profiles:user_profile', username=user.username)
|
|
messages.error(request, 'Could not update bio.')
|
|
else:
|
|
bio_form = BioForm(instance=profile)
|
|
return render(request, 'profiles/profile.html', {
|
|
'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',
|
|
).all(),
|
|
})
|
|
|
|
|
|
@login_required
|
|
def bio_preview(request):
|
|
"""Return the markdown-rendered HTML for a bio draft (editor live preview)."""
|
|
if request.method != 'POST':
|
|
return JsonResponse({'detail': 'Method not allowed'}, status=405)
|
|
text = request.POST.get('bio', '')
|
|
return JsonResponse({'html': render_markdown(text)}) |