Fix #954: Superuser detection; Feature #1004: User account disable/enable

Bug #954: The initialize_superuser endpoint only checked is_superuser=True, missing admin accounts created via the API (user_level=10). This caused users to intermittently see the "Create Super User" page instead of login.

Feature #1004: Admins can now disable/enable user accounts. Disabled users are blocked from JWT login, token refresh, XC API access, and all
authenticated endpoints. The last active admin cannot be disabled.

Files changed:

Backend:
- apps/accounts/api_views.py — Superuser check uses user_level>=10; token refresh blocks disabled users
- apps/accounts/models.py — CustomUserManager ensures create_superuser() always sets user_level=10
- apps/accounts/serializers.py — Last-admin protection guard; respect is_active on user creation
- apps/accounts/tests.py — new tests covering superuser detection, token
 refresh blocking, last-admin protection, disabled user login/access
- apps/backups/api_views.py — Switched from DRF's IsAdminUser (checks is_staff) to app's IsAdmin (checks user_level) on all 8 endpoints
- apps/backups/tests.py — new tests verifying user_level-based admin permission on backup endpoints
- apps/output/views.py — is_active check on xc_get_user(), xc_movie_stream(), xc_series_stream()
- apps/proxy/ts_proxy/views.py — is_active check on stream_xc()
- core/api_views.py — Admin notification filter uses user_level>=10
- core/developer_notifications.py — Admin checks use user_level>=10

Frontend:
- frontend/src/App.jsx — Null-safe superuser existence check
- frontend/src/components/forms/User.jsx — Account Enabled switch with
  self-disable prevention and tooltip
- frontend/src/components/tables/UsersTable.jsx — Status column (Active/Disabled badge)
This commit is contained in:
None 2026-02-22 18:05:02 -06:00
parent 8babb7d111
commit db318e4981
13 changed files with 487 additions and 21 deletions

View file

@ -113,13 +113,31 @@ class TokenRefreshView(TokenRefreshView):
)
return Response({"error": "Unauthorized"}, status=status.HTTP_403_FORBIDDEN)
# Check if user account is still active before issuing new access token
raw_token = request.data.get("refresh")
if raw_token:
try:
from rest_framework_simplejwt.tokens import RefreshToken as RefreshTokenClass
token = RefreshTokenClass(raw_token)
user_id = token.payload.get("user_id")
if user_id:
user = User.objects.filter(id=user_id).first()
if user and not user.is_active:
logger.info(f"Token refresh blocked for disabled user: user_id={user_id}")
return Response(
{"error": "Account is disabled"},
status=status.HTTP_403_FORBIDDEN
)
except Exception:
pass # Let parent handle invalid tokens
return super().post(request, *args, **kwargs)
@csrf_exempt # In production, consider CSRF protection strategies or ensure this endpoint is only accessible when no superuser exists.
def initialize_superuser(request):
# If a superuser already exists, always indicate that
if User.objects.filter(is_superuser=True).exists():
# If an admin-level user already exists, the system is configured
if User.objects.filter(user_level__gte=10).exists():
return JsonResponse({"superuser_exists": True})
if request.method == "POST":

View file

@ -1,9 +1,16 @@
# apps/accounts/models.py
from django.db import models
from django.contrib.auth.models import AbstractUser, Permission
from django.contrib.auth.models import AbstractUser, Permission, UserManager
class CustomUserManager(UserManager):
def create_superuser(self, username, email=None, password=None, **extra_fields):
extra_fields.setdefault('user_level', 10)
return super().create_superuser(username, email, password, **extra_fields)
class User(AbstractUser):
objects = CustomUserManager()
"""
Custom user model for Dispatcharr.
Inherits from Django's AbstractUser to add additional fields if needed.

View file

@ -52,9 +52,10 @@ class UserSerializer(serializers.ModelSerializer):
def create(self, validated_data):
channel_profiles = validated_data.pop("channel_profiles", [])
is_active = validated_data.pop("is_active", True)
user = User(**validated_data)
user.set_password(validated_data["password"])
user.is_active = True
user.is_active = is_active
user.save()
user.channel_profiles.set(channel_profiles)
@ -62,6 +63,17 @@ class UserSerializer(serializers.ModelSerializer):
return user
def update(self, instance, validated_data):
# Prevent disabling the last active admin account
if 'is_active' in validated_data and not validated_data['is_active']:
other_active_admins = User.objects.filter(
user_level__gte=10,
is_active=True
).exclude(id=instance.id).exists()
if not other_active_admins:
raise serializers.ValidationError(
{"is_active": "Cannot disable the last active admin account."}
)
password = validated_data.pop("password", None)
channel_profiles = validated_data.pop("channel_profiles", None)

283
apps/accounts/tests.py Normal file
View file

@ -0,0 +1,283 @@
from django.test import TestCase, RequestFactory
from django.contrib.auth import get_user_model
from rest_framework.test import APIClient
from rest_framework import status
User = get_user_model()
class InitializeSuperuserTests(TestCase):
"""Tests for the initialize_superuser endpoint"""
def setUp(self):
self.client = APIClient()
self.url = "/api/accounts/initialize-superuser/"
def test_returns_true_when_superuser_exists(self):
"""Superuser with is_superuser=True should be detected"""
User.objects.create_superuser(
username="admin", password="testpass123", user_level=10
)
response = self.client.get(self.url)
self.assertEqual(response.status_code, 200)
self.assertTrue(response.json()["superuser_exists"])
def test_returns_true_when_admin_level_user_exists(self):
"""User with user_level=10 but is_superuser=False should be detected"""
user = User.objects.create_user(username="admin", password="testpass123")
user.user_level = 10
user.is_superuser = False
user.save()
response = self.client.get(self.url)
self.assertEqual(response.status_code, 200)
self.assertTrue(response.json()["superuser_exists"])
def test_returns_false_when_no_admin_exists(self):
"""No admin or superuser should return false"""
# Create a non-admin user
User.objects.create_user(username="regular", password="testpass123")
response = self.client.get(self.url)
self.assertEqual(response.status_code, 200)
self.assertFalse(response.json()["superuser_exists"])
def test_returns_false_when_no_users_exist(self):
"""Empty database should return false"""
response = self.client.get(self.url)
self.assertEqual(response.status_code, 200)
self.assertFalse(response.json()["superuser_exists"])
def test_create_superuser_when_none_exists(self):
"""POST should create superuser when none exists"""
response = self.client.post(
self.url,
{"username": "newadmin", "password": "testpass123", "email": "admin@test.com"},
format="json",
)
self.assertEqual(response.status_code, 200)
self.assertTrue(response.json()["superuser_exists"])
self.assertTrue(User.objects.filter(username="newadmin", user_level=10).exists())
def test_cannot_create_superuser_when_admin_exists(self):
"""POST should fail when an admin-level user already exists"""
user = User.objects.create_user(username="existing", password="testpass123")
user.user_level = 10
user.save()
response = self.client.post(
self.url,
{"username": "newadmin", "password": "testpass123"},
format="json",
)
self.assertEqual(response.status_code, 200)
self.assertTrue(response.json()["superuser_exists"])
# Should NOT have created a new user
self.assertFalse(User.objects.filter(username="newadmin").exists())
class TokenRefreshDisabledUserTests(TestCase):
"""Tests for blocking token refresh on disabled users"""
def setUp(self):
self.client = APIClient()
self.token_url = "/api/accounts/token/"
self.refresh_url = "/api/accounts/token/refresh/"
def test_refresh_works_for_active_user(self):
"""Active user should be able to refresh their token"""
user = User.objects.create_user(username="active", password="testpass123")
user.user_level = 1
user.is_active = True
user.save()
# Get tokens
login_response = self.client.post(
self.token_url,
{"username": "active", "password": "testpass123"},
format="json",
)
self.assertEqual(login_response.status_code, 200)
refresh_token = login_response.data["refresh"]
# Refresh should succeed
refresh_response = self.client.post(
self.refresh_url,
{"refresh": refresh_token},
format="json",
)
self.assertEqual(refresh_response.status_code, 200)
self.assertIn("access", refresh_response.data)
def test_refresh_blocked_for_disabled_user(self):
"""Disabled user should not be able to refresh their token"""
user = User.objects.create_user(username="disabled", password="testpass123")
user.user_level = 1
user.is_active = True
user.save()
# Get tokens while user is active
login_response = self.client.post(
self.token_url,
{"username": "disabled", "password": "testpass123"},
format="json",
)
self.assertEqual(login_response.status_code, 200)
refresh_token = login_response.data["refresh"]
# Disable the user
user.is_active = False
user.save()
# Refresh should be blocked
refresh_response = self.client.post(
self.refresh_url,
{"refresh": refresh_token},
format="json",
)
self.assertEqual(refresh_response.status_code, 403)
class LastAdminProtectionTests(TestCase):
"""Tests for preventing disabling the last active admin"""
def setUp(self):
self.admin = User.objects.create_user(username="admin1", password="testpass123")
self.admin.user_level = 10
self.admin.save()
self.client = APIClient()
self.client.force_authenticate(user=self.admin)
def test_cannot_disable_last_admin(self):
"""Should reject disabling the only active admin"""
response = self.client.patch(
f"/api/accounts/users/{self.admin.id}/",
{"is_active": False},
format="json",
)
self.assertEqual(response.status_code, 400)
self.assertIn("is_active", response.data)
# Verify admin is still active
self.admin.refresh_from_db()
self.assertTrue(self.admin.is_active)
def test_can_disable_admin_when_another_exists(self):
"""Should allow disabling an admin when another active admin exists"""
admin2 = User.objects.create_user(username="admin2", password="testpass123")
admin2.user_level = 10
admin2.save()
response = self.client.patch(
f"/api/accounts/users/{admin2.id}/",
{"is_active": False},
format="json",
)
self.assertEqual(response.status_code, 200)
admin2.refresh_from_db()
self.assertFalse(admin2.is_active)
def test_can_disable_non_admin_user(self):
"""Should always allow disabling non-admin users"""
regular = User.objects.create_user(username="regular", password="testpass123")
regular.user_level = 1
regular.save()
response = self.client.patch(
f"/api/accounts/users/{regular.id}/",
{"is_active": False},
format="json",
)
self.assertEqual(response.status_code, 200)
regular.refresh_from_db()
self.assertFalse(regular.is_active)
def test_can_reenable_disabled_user(self):
"""Should allow re-enabling a disabled user"""
regular = User.objects.create_user(username="disabled", password="testpass123")
regular.user_level = 1
regular.is_active = False
regular.save()
response = self.client.patch(
f"/api/accounts/users/{regular.id}/",
{"is_active": True},
format="json",
)
self.assertEqual(response.status_code, 200)
regular.refresh_from_db()
self.assertTrue(regular.is_active)
class DisabledUserLoginTests(TestCase):
"""Tests that disabled users cannot log in"""
def setUp(self):
self.client = APIClient()
self.token_url = "/api/accounts/token/"
def test_disabled_user_cannot_login(self):
"""Disabled user should get rejected at login"""
user = User.objects.create_user(username="disabled", password="testpass123")
user.is_active = False
user.save()
response = self.client.post(
self.token_url,
{"username": "disabled", "password": "testpass123"},
format="json",
)
self.assertEqual(response.status_code, 401)
def test_active_user_can_login(self):
"""Active user should be able to log in"""
user = User.objects.create_user(username="active", password="testpass123")
user.is_active = True
user.save()
response = self.client.post(
self.token_url,
{"username": "active", "password": "testpass123"},
format="json",
)
self.assertEqual(response.status_code, 200)
self.assertIn("access", response.data)
class DisabledUserAccessTokenTests(TestCase):
"""Tests that a disabled user's existing access token is rejected on authenticated endpoints"""
def setUp(self):
self.client = APIClient()
self.token_url = "/api/accounts/token/"
self.users_url = "/api/accounts/users/"
def test_existing_token_rejected_after_disable(self):
"""Access token obtained while active should be rejected after user is disabled"""
user = User.objects.create_user(username="willdisable", password="testpass123")
user.user_level = 10 # Admin so they can access the users endpoint
user.is_active = True
user.save()
# Get tokens while user is active
login_response = self.client.post(
self.token_url,
{"username": "willdisable", "password": "testpass123"},
format="json",
)
self.assertEqual(login_response.status_code, 200)
access_token = login_response.data["access"]
# Verify token works while active
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {access_token}")
response = self.client.get(self.users_url)
self.assertEqual(response.status_code, 200)
# Disable the user
user.is_active = False
user.save()
# Same token should now be rejected
response = self.client.get(self.users_url)
self.assertIn(response.status_code, [401, 403])

View file

@ -9,7 +9,8 @@ from django.conf import settings
from django.http import HttpResponse, StreamingHttpResponse, Http404
from rest_framework import status
from rest_framework.decorators import api_view, permission_classes, parser_classes
from rest_framework.permissions import IsAdminUser, AllowAny
from rest_framework.permissions import AllowAny
from apps.accounts.permissions import IsAdmin
from rest_framework.parsers import MultiPartParser, FormParser
from rest_framework.response import Response
@ -33,7 +34,7 @@ def _verify_task_token(task_id: str, token: str) -> bool:
@api_view(["GET"])
@permission_classes([IsAdminUser])
@permission_classes([IsAdmin])
def list_backups(request):
"""List all available backup files."""
try:
@ -47,7 +48,7 @@ def list_backups(request):
@api_view(["POST"])
@permission_classes([IsAdminUser])
@permission_classes([IsAdmin])
def create_backup(request):
"""Create a new backup (async via Celery)."""
try:
@ -86,7 +87,7 @@ def backup_status(request, task_id):
)
else:
# Fall back to admin auth check
if not request.user.is_authenticated or not request.user.is_staff:
if not request.user.is_authenticated or getattr(request.user, 'user_level', 0) < 10:
return Response(
{"detail": "Authentication required"},
status=status.HTTP_401_UNAUTHORIZED,
@ -124,7 +125,7 @@ def backup_status(request, task_id):
@api_view(["GET"])
@permission_classes([IsAdminUser])
@permission_classes([IsAdmin])
def get_download_token(request, filename):
"""Get a signed token for downloading a backup file."""
try:
@ -168,7 +169,7 @@ def download_backup(request, filename):
)
else:
# Fall back to admin auth check
if not request.user.is_authenticated or not request.user.is_staff:
if not request.user.is_authenticated or getattr(request.user, 'user_level', 0) < 10:
return Response(
{"detail": "Authentication required"},
status=status.HTTP_401_UNAUTHORIZED,
@ -230,7 +231,7 @@ def download_backup(request, filename):
@api_view(["DELETE"])
@permission_classes([IsAdminUser])
@permission_classes([IsAdmin])
def delete_backup(request, filename):
"""Delete a backup file."""
try:
@ -253,7 +254,7 @@ def delete_backup(request, filename):
@api_view(["POST"])
@permission_classes([IsAdminUser])
@permission_classes([IsAdmin])
@parser_classes([MultiPartParser, FormParser])
def upload_backup(request):
"""Upload a backup file for restoration."""
@ -299,7 +300,7 @@ def upload_backup(request):
@api_view(["POST"])
@permission_classes([IsAdminUser])
@permission_classes([IsAdmin])
def restore_backup(request, filename):
"""Restore from a backup file (async via Celery). WARNING: This will flush the database!"""
try:
@ -332,7 +333,7 @@ def restore_backup(request, filename):
@api_view(["GET"])
@permission_classes([IsAdminUser])
@permission_classes([IsAdmin])
def get_schedule(request):
"""Get backup schedule settings."""
try:
@ -346,7 +347,7 @@ def get_schedule(request):
@api_view(["PUT"])
@permission_classes([IsAdminUser])
@permission_classes([IsAdmin])
def update_schedule(request):
"""Update backup schedule settings."""
try:

View file

@ -735,6 +735,94 @@ class BackupAPITestCase(TestCase):
self.assertIn('frequency', data['detail'])
class BackupAdminPermissionTestCase(TestCase):
"""Test that backup endpoints use user_level (not is_staff/is_superuser) for admin checks.
This validates the IsAdminUser -> IsAdmin permission change.
API-created admins have user_level=10 but is_staff=False and is_superuser=False.
"""
def setUp(self):
self.client = APIClient()
# API-created admin: user_level=10 but NOT is_staff or is_superuser
self.api_admin = User.objects.create_user(
username='api_admin',
email='apiadmin@example.com',
password='testpass123'
)
self.api_admin.user_level = 10
self.api_admin.is_staff = False
self.api_admin.is_superuser = False
self.api_admin.save()
# User with is_staff=True but low user_level (should NOT have access)
self.staff_user = User.objects.create_user(
username='staffuser',
email='staff@example.com',
password='testpass123'
)
self.staff_user.is_staff = True
self.staff_user.user_level = 1
self.staff_user.save()
self.temp_backup_dir = tempfile.mkdtemp()
def get_auth_header(self, user):
refresh = RefreshToken.for_user(user)
return f'Bearer {str(refresh.access_token)}'
def tearDown(self):
import shutil
if Path(self.temp_backup_dir).exists():
shutil.rmtree(self.temp_backup_dir)
@patch('apps.backups.services.list_backups')
def test_api_created_admin_can_list_backups(self, mock_list_backups):
"""API-created admin (user_level=10, is_staff=False) should access backup endpoints"""
mock_list_backups.return_value = []
auth_header = self.get_auth_header(self.api_admin)
response = self.client.get('/api/backups/', HTTP_AUTHORIZATION=auth_header)
self.assertEqual(response.status_code, 200)
def test_staff_user_without_user_level_cannot_list_backups(self):
"""User with is_staff=True but user_level < 10 should NOT access backup endpoints"""
auth_header = self.get_auth_header(self.staff_user)
response = self.client.get('/api/backups/', HTTP_AUTHORIZATION=auth_header)
self.assertIn(response.status_code, [401, 403])
@patch('apps.backups.tasks.create_backup_task.delay')
def test_api_created_admin_can_create_backup(self, mock_create_task):
"""API-created admin should be able to create backups"""
mock_task = MagicMock()
mock_task.id = 'test-task-id'
mock_create_task.return_value = mock_task
auth_header = self.get_auth_header(self.api_admin)
response = self.client.post('/api/backups/create/', HTTP_AUTHORIZATION=auth_header)
self.assertEqual(response.status_code, 202)
@patch('apps.backups.services.get_backup_dir')
def test_api_created_admin_can_delete_backup(self, mock_get_backup_dir):
"""API-created admin should be able to delete backups"""
backup_dir = Path(self.temp_backup_dir)
mock_get_backup_dir.return_value = backup_dir
backup_file = backup_dir / "test-backup.zip"
backup_file.write_text("test content")
auth_header = self.get_auth_header(self.api_admin)
response = self.client.delete(
'/api/backups/test-backup.zip/delete/',
HTTP_AUTHORIZATION=auth_header
)
self.assertEqual(response.status_code, 204)
class BackupSchedulerTestCase(TestCase):
"""Test cases for backup scheduler"""

View file

@ -1945,6 +1945,10 @@ def xc_get_user(request):
return None
user = get_object_or_404(User, username=username)
if not user.is_active:
return None
custom_properties = user.custom_properties or {}
if "xc_password" not in custom_properties:
@ -2924,6 +2928,9 @@ def xc_movie_stream(request, username, password, stream_id, extension):
user = get_object_or_404(User, username=username)
if not user.is_active:
return JsonResponse({"error": "Account is disabled"}, status=403)
custom_properties = user.custom_properties or {}
if "xc_password" not in custom_properties:
@ -2961,6 +2968,9 @@ def xc_series_stream(request, username, password, stream_id, extension):
user = get_object_or_404(User, username=username)
if not user.is_active:
return JsonResponse({"error": "Account is disabled"}, status=403)
custom_properties = user.custom_properties or {}
if "xc_password" not in custom_properties:

View file

@ -514,6 +514,9 @@ def stream_ts(request, channel_id):
def stream_xc(request, username, password, channel_id):
user = get_object_or_404(User, username=username)
if not user.is_active:
return JsonResponse({"error": "Account is disabled"}, status=403)
extension = pathlib.Path(channel_id).suffix
channel_id = pathlib.Path(channel_id).stem

View file

@ -506,7 +506,7 @@ class SystemNotificationViewSet(viewsets.ModelViewSet):
)
# Filter admin-only notifications for non-admins
if not getattr(user, 'is_superuser', False) and getattr(user, 'user_level', 0) < 10:
if getattr(user, 'user_level', 0) < 10:
queryset = queryset.filter(admin_only=False)
# For developer notifications, evaluate conditions

View file

@ -200,7 +200,7 @@ def should_show_notification(notification_data: dict, user) -> bool:
# Check user level
user_level = notification_data.get('user_level', 'all')
if user_level == 'admin' and not getattr(user, 'is_superuser', False):
if user_level == 'admin' and getattr(user, 'user_level', 0) < 10:
return False
# Check conditions
@ -396,7 +396,7 @@ def get_user_developer_notifications(user) -> list:
)
# Filter by admin_only based on user
if not getattr(user, 'is_superuser', False):
if getattr(user, 'user_level', 0) < 10:
notifications = notifications.filter(admin_only=False)
# Filter by conditions

View file

@ -63,7 +63,7 @@ const App = () => {
async function checkSuperuser() {
try {
const response = await API.fetchSuperUser();
if (!response.superuser_exists) {
if (response && response.superuser_exists === false) {
setSuperuserExists(false);
}
} catch (error) {

View file

@ -42,6 +42,7 @@ const User = ({ user = null, isOpen, onClose }) => {
xc_password: '',
channel_profiles: [],
hide_adult_content: false,
is_active: true,
},
validate: (values) => ({
@ -134,6 +135,7 @@ const User = ({ user = null, isOpen, onClose }) => {
: ['0'],
xc_password: customProps.xc_password || '',
hide_adult_content: customProps.hide_adult_content || false,
is_active: user.is_active !== false,
});
if (customProps.xc_password) {
@ -154,8 +156,9 @@ const User = ({ user = null, isOpen, onClose }) => {
return <></>;
}
const showPermissions =
authUser.user_level == USER_LEVELS.ADMIN && authUser.id !== user?.id;
const isAdmin = authUser.user_level == USER_LEVELS.ADMIN;
const isEditingSelf = authUser.id === user?.id;
const showPermissions = isAdmin && !isEditingSelf;
return (
<Modal opened={isOpen} onClose={onClose} title="User" size="xl">
@ -199,6 +202,31 @@ const User = ({ user = null, isOpen, onClose }) => {
key={form.key('user_level')}
/>
)}
{isAdmin && (
<Box>
<Tooltip
label={isEditingSelf
? "You cannot disable your own account"
: "Disabled accounts cannot log in or access any resources"
}
position="top"
withArrow
>
<div style={isEditingSelf ? { cursor: 'not-allowed' } : undefined}>
<Switch
label="Account Enabled"
{...form.getInputProps('is_active', {
type: 'checkbox',
})}
key={form.key('is_active')}
disabled={isEditingSelf}
styles={isEditingSelf ? { track: { pointerEvents: 'none' } } : undefined}
/>
</div>
</Tooltip>
</Box>
)}
</Stack>
<Stack gap="xs" style={{ flex: 1 }}>

View file

@ -8,6 +8,7 @@ import useWarningsStore from '../../store/warnings';
import { SquarePlus, SquareMinus, SquarePen, Eye, EyeOff } from 'lucide-react';
import {
ActionIcon,
Badge,
Box,
Text,
Paper,
@ -147,6 +148,20 @@ const UsersTable = () => {
<Text size="sm">{USER_LEVEL_LABELS[getValue()]}</Text>
),
},
{
header: 'Status',
accessorKey: 'is_active',
size: 90,
cell: ({ getValue }) => (
<Badge
size="sm"
color={getValue() !== false ? 'green' : 'red'}
variant="light"
>
{getValue() !== false ? 'Active' : 'Disabled'}
</Badge>
),
},
{
header: 'Username',
accessorKey: 'username',
@ -316,6 +331,7 @@ const UsersTable = () => {
name: renderHeaderCell,
email: renderHeaderCell,
user_level: renderHeaderCell,
is_active: renderHeaderCell,
last_login: renderHeaderCell,
date_joined: renderHeaderCell,
custom_properties: renderHeaderCell,