Add admin login API and remove hardcoded secrets
- Add token-based admin login/logout/me endpoints - Bootstrap admin from ADMIN_USERNAME/ADMIN_PASSWORD in Docker entrypoint - Read ALLOWED_HOSTS and CORS from env only (no hardcoded server IPs) - Keep docs/Postman/tests free of real credentials - Cover login and auth flows with 31 local API tests Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -1,7 +1,34 @@
|
||||
from rest_framework import serializers
|
||||
from django.contrib.auth import authenticate
|
||||
from .models import ContactUs, Composition, Campaign, CompositionImage
|
||||
|
||||
|
||||
class AdminLoginSerializer(serializers.Serializer):
|
||||
"""Serializer for admin username/password login."""
|
||||
username = serializers.CharField(required=True)
|
||||
password = serializers.CharField(required=True, write_only=True)
|
||||
|
||||
def validate(self, data):
|
||||
username = data.get('username', '').strip()
|
||||
password = data.get('password', '')
|
||||
|
||||
if not username or not password:
|
||||
raise serializers.ValidationError('Username and password are required.')
|
||||
|
||||
user = authenticate(username=username, password=password)
|
||||
if user is None:
|
||||
raise serializers.ValidationError('Invalid username or password.')
|
||||
if not user.is_active:
|
||||
raise serializers.ValidationError('This account is disabled.')
|
||||
if not (user.is_staff or user.is_superuser):
|
||||
raise serializers.ValidationError(
|
||||
'This account does not have admin access.'
|
||||
)
|
||||
|
||||
data['user'] = user
|
||||
return data
|
||||
|
||||
|
||||
class ContactUsSerializer(serializers.ModelSerializer):
|
||||
"""
|
||||
Serializer for ContactUs model.
|
||||
|
||||
113
api/tests.py
113
api/tests.py
@@ -64,7 +64,10 @@ class ContactUsAPITests(APITestCase):
|
||||
{'name': 'Updated'},
|
||||
format='json',
|
||||
)
|
||||
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
||||
self.assertIn(response.status_code, (
|
||||
status.HTTP_401_UNAUTHORIZED,
|
||||
status.HTTP_403_FORBIDDEN,
|
||||
))
|
||||
|
||||
|
||||
class CompositionAPITests(APITestCase):
|
||||
@@ -200,7 +203,10 @@ class CompositionAPITests(APITestCase):
|
||||
{'name': 'Updated'},
|
||||
format='json',
|
||||
)
|
||||
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
||||
self.assertIn(response.status_code, (
|
||||
status.HTTP_401_UNAUTHORIZED,
|
||||
status.HTTP_403_FORBIDDEN,
|
||||
))
|
||||
|
||||
|
||||
class CampaignAPITests(APITestCase):
|
||||
@@ -277,3 +283,106 @@ class CampaignAPITests(APITestCase):
|
||||
}
|
||||
response = self.client.post('/api/campaigns/', payload, format='json')
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
|
||||
class AdminLoginAPITests(APITestCase):
|
||||
def setUp(self):
|
||||
User = get_user_model()
|
||||
self.admin_username = 'admin_test'
|
||||
self.admin_password = 'test_admin_pass_123'
|
||||
self.admin = User.objects.create_user(
|
||||
username=self.admin_username,
|
||||
password=self.admin_password,
|
||||
is_staff=True,
|
||||
is_superuser=True,
|
||||
)
|
||||
self.regular = User.objects.create_user(
|
||||
username='normaluser',
|
||||
password='normalpass123',
|
||||
is_staff=False,
|
||||
)
|
||||
|
||||
def test_admin_login_success(self):
|
||||
response = self.client.post(
|
||||
'/api/admin/login/',
|
||||
{'username': self.admin_username, 'password': self.admin_password},
|
||||
format='json',
|
||||
)
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertIn('token', response.data)
|
||||
self.assertEqual(response.data['user']['username'], self.admin_username)
|
||||
self.assertTrue(response.data['user']['is_staff'])
|
||||
|
||||
def test_admin_login_wrong_password(self):
|
||||
response = self.client.post(
|
||||
'/api/admin/login/',
|
||||
{'username': self.admin_username, 'password': 'wrong'},
|
||||
format='json',
|
||||
)
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
def test_non_staff_cannot_login(self):
|
||||
response = self.client.post(
|
||||
'/api/admin/login/',
|
||||
{'username': 'normaluser', 'password': 'normalpass123'},
|
||||
format='json',
|
||||
)
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
def test_admin_me_with_token(self):
|
||||
login = self.client.post(
|
||||
'/api/admin/login/',
|
||||
{'username': self.admin_username, 'password': self.admin_password},
|
||||
format='json',
|
||||
)
|
||||
token = login.data['token']
|
||||
self.client.credentials(HTTP_AUTHORIZATION=f'Token {token}')
|
||||
response = self.client.get('/api/admin/me/')
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertEqual(response.data['username'], self.admin_username)
|
||||
|
||||
def test_admin_me_requires_auth(self):
|
||||
response = self.client.get('/api/admin/me/')
|
||||
self.assertIn(response.status_code, (
|
||||
status.HTTP_401_UNAUTHORIZED,
|
||||
status.HTTP_403_FORBIDDEN,
|
||||
))
|
||||
|
||||
def test_token_allows_protected_update(self):
|
||||
contact = ContactUs.objects.create(
|
||||
name='Old Name',
|
||||
email_or_phone='a@b.com',
|
||||
description='msg',
|
||||
category='سایر',
|
||||
)
|
||||
login = self.client.post(
|
||||
'/api/admin/login/',
|
||||
{'username': self.admin_username, 'password': self.admin_password},
|
||||
format='json',
|
||||
)
|
||||
self.client.credentials(HTTP_AUTHORIZATION=f'Token {login.data["token"]}')
|
||||
response = self.client.patch(
|
||||
f'/api/contact-us/{contact.id}/',
|
||||
{'name': 'New Name'},
|
||||
format='json',
|
||||
)
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertEqual(response.data['name'], 'New Name')
|
||||
|
||||
def test_admin_logout(self):
|
||||
login = self.client.post(
|
||||
'/api/admin/login/',
|
||||
{'username': self.admin_username, 'password': self.admin_password},
|
||||
format='json',
|
||||
)
|
||||
token = login.data['token']
|
||||
self.client.credentials(HTTP_AUTHORIZATION=f'Token {token}')
|
||||
response = self.client.post('/api/admin/logout/')
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
# Token should no longer work
|
||||
response = self.client.get('/api/admin/me/')
|
||||
self.assertIn(response.status_code, (
|
||||
status.HTTP_401_UNAUTHORIZED,
|
||||
status.HTTP_403_FORBIDDEN,
|
||||
))
|
||||
|
||||
13
api/urls.py
13
api/urls.py
@@ -1,6 +1,13 @@
|
||||
from django.urls import path, include
|
||||
from rest_framework.routers import DefaultRouter
|
||||
from .views import ContactUsViewSet, CompositionViewSet, CampaignViewSet
|
||||
from .views import (
|
||||
ContactUsViewSet,
|
||||
CompositionViewSet,
|
||||
CampaignViewSet,
|
||||
admin_login,
|
||||
admin_logout,
|
||||
admin_me,
|
||||
)
|
||||
|
||||
router = DefaultRouter()
|
||||
router.register(r'contact-us', ContactUsViewSet, basename='contact-us')
|
||||
@@ -8,6 +15,8 @@ router.register(r'compositions', CompositionViewSet, basename='composition')
|
||||
router.register(r'campaigns', CampaignViewSet, basename='campaign')
|
||||
|
||||
urlpatterns = [
|
||||
path('admin/login/', admin_login, name='admin-login'),
|
||||
path('admin/logout/', admin_logout, name='admin-logout'),
|
||||
path('admin/me/', admin_me, name='admin-me'),
|
||||
path('', include(router.urls)),
|
||||
]
|
||||
|
||||
|
||||
50
api/views.py
50
api/views.py
@@ -1,8 +1,9 @@
|
||||
from rest_framework import viewsets, status
|
||||
from rest_framework.decorators import action
|
||||
from rest_framework.decorators import action, api_view, permission_classes
|
||||
from rest_framework.response import Response
|
||||
from rest_framework.permissions import AllowAny, IsAuthenticated
|
||||
from rest_framework.parsers import MultiPartParser, FormParser, JSONParser
|
||||
from rest_framework.authtoken.models import Token
|
||||
from django.shortcuts import get_object_or_404
|
||||
from django.db.models import Q
|
||||
from django.utils import timezone
|
||||
@@ -13,9 +14,56 @@ from .serializers import (
|
||||
CompositionSerializer,
|
||||
CampaignSerializer,
|
||||
CompositionImageSerializer,
|
||||
AdminLoginSerializer,
|
||||
)
|
||||
|
||||
|
||||
@api_view(['POST'])
|
||||
@permission_classes([AllowAny])
|
||||
def admin_login(request):
|
||||
"""
|
||||
Admin login with username and password.
|
||||
|
||||
Returns an auth token to use as:
|
||||
Authorization: Token <token>
|
||||
"""
|
||||
serializer = AdminLoginSerializer(data=request.data)
|
||||
serializer.is_valid(raise_exception=True)
|
||||
user = serializer.validated_data['user']
|
||||
token, _ = Token.objects.get_or_create(user=user)
|
||||
return Response({
|
||||
'token': token.key,
|
||||
'user': {
|
||||
'id': user.id,
|
||||
'username': user.username,
|
||||
'is_staff': user.is_staff,
|
||||
'is_superuser': user.is_superuser,
|
||||
},
|
||||
})
|
||||
|
||||
|
||||
@api_view(['POST'])
|
||||
@permission_classes([IsAuthenticated])
|
||||
def admin_logout(request):
|
||||
"""Delete the current admin auth token (logout)."""
|
||||
Token.objects.filter(user=request.user).delete()
|
||||
return Response({'detail': 'Logged out successfully.'})
|
||||
|
||||
|
||||
@api_view(['GET'])
|
||||
@permission_classes([IsAuthenticated])
|
||||
def admin_me(request):
|
||||
"""Return the currently authenticated admin user."""
|
||||
user = request.user
|
||||
return Response({
|
||||
'id': user.id,
|
||||
'username': user.username,
|
||||
'is_staff': user.is_staff,
|
||||
'is_superuser': user.is_superuser,
|
||||
'is_active': user.is_active,
|
||||
})
|
||||
|
||||
|
||||
class ContactUsViewSet(viewsets.ModelViewSet):
|
||||
"""
|
||||
ViewSet for ContactUs model.
|
||||
|
||||
Reference in New Issue
Block a user