first commit
This commit is contained in:
commit
367533c02b
171 files changed
+101175
No files matched your search
Whitespace-only changes.
@@ -0,0 +1,33 @@
|
||||
from django.contrib import admin
|
||||
from django.contrib.auth.admin import UserAdmin as DjangoUserAdmin
|
||||
|
||||
from apps.accounts.models import ApiKey, User
|
||||
|
||||
|
||||
@admin.register(User)
|
||||
class UserAdmin(DjangoUserAdmin):
|
||||
ordering = ("user_name",)
|
||||
list_display = ("user_name", "status", "created_at")
|
||||
search_fields = ("user_name",)
|
||||
fieldsets = (
|
||||
(None, {"fields": ("user_name", "password")}),
|
||||
("Permissions", {"fields": ("status", "groups", "user_permissions")}),
|
||||
)
|
||||
add_fieldsets = (
|
||||
(
|
||||
None,
|
||||
{
|
||||
"classes": ("wide",),
|
||||
"fields": ("user_name", "password1", "password2", "status"),
|
||||
},
|
||||
),
|
||||
)
|
||||
filter_horizontal = ("groups", "user_permissions")
|
||||
|
||||
|
||||
@admin.register(ApiKey)
|
||||
class ApiKeyAdmin(admin.ModelAdmin):
|
||||
list_display = ("name", "prefix", "user", "is_active", "last_used_at", "created_at")
|
||||
list_filter = ("is_active",)
|
||||
search_fields = ("name", "prefix", "user__user_name")
|
||||
readonly_fields = ("prefix", "key_hash", "last_used_at", "created_at", "updated_at")
|
||||
@@ -0,0 +1,8 @@
|
||||
from django.apps import AppConfig
|
||||
|
||||
|
||||
class AccountsConfig(AppConfig):
|
||||
default_auto_field = "django.db.models.BigAutoField"
|
||||
name = "apps.accounts"
|
||||
label = "accounts"
|
||||
verbose_name = "Accounts"
|
||||
@@ -0,0 +1,32 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
from rest_framework import authentication, exceptions
|
||||
|
||||
from apps.accounts.models import ApiKey
|
||||
|
||||
|
||||
class ApiKeyAuthentication(authentication.BaseAuthentication):
|
||||
"""Authenticate via X-API-Key header (hashed lookup)."""
|
||||
|
||||
keyword = "X-API-Key"
|
||||
|
||||
def authenticate(self, request):
|
||||
raw_key = request.headers.get(self.keyword) or request.META.get("HTTP_X_API_KEY")
|
||||
if not raw_key:
|
||||
return None
|
||||
if len(raw_key) < 8:
|
||||
raise exceptions.AuthenticationFailed(_("Invalid API key"))
|
||||
|
||||
prefix = raw_key[:8]
|
||||
candidates = ApiKey.objects.select_related("user").filter(
|
||||
prefix=prefix, is_active=True, user__is_active=True
|
||||
)
|
||||
for api_key in candidates:
|
||||
if api_key.verify(raw_key):
|
||||
api_key.touch()
|
||||
return (api_key.user, api_key)
|
||||
raise exceptions.AuthenticationFailed(_("Invalid API key"))
|
||||
|
||||
def authenticate_header(self, request):
|
||||
return self.keyword
|
||||
@@ -0,0 +1,55 @@
|
||||
# Generated by Django 5.2.17 on 2026-08-10 08:31
|
||||
|
||||
import django.db.models.deletion
|
||||
from django.conf import settings
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
initial = True
|
||||
|
||||
dependencies = [
|
||||
('auth', '0012_alter_user_first_name_max_length'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='User',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('password', models.CharField(max_length=128, verbose_name='password')),
|
||||
('last_login', models.DateTimeField(blank=True, null=True, verbose_name='last login')),
|
||||
('is_superuser', models.BooleanField(default=False, help_text='Designates that this user has all permissions without explicitly assigning them.', verbose_name='superuser status')),
|
||||
('user_name', models.CharField(max_length=30, unique=True)),
|
||||
('is_staff', models.BooleanField(default=False)),
|
||||
('is_active', models.BooleanField(default=True)),
|
||||
('created_at', models.DateTimeField(auto_now_add=True)),
|
||||
('updated_at', models.DateTimeField(auto_now=True)),
|
||||
('groups', models.ManyToManyField(blank=True, help_text='The groups this user belongs to. A user will get all permissions granted to each of their groups.', related_name='user_set', related_query_name='user', to='auth.group', verbose_name='groups')),
|
||||
('user_permissions', models.ManyToManyField(blank=True, help_text='Specific permissions for this user.', related_name='user_set', related_query_name='user', to='auth.permission', verbose_name='user permissions')),
|
||||
],
|
||||
options={
|
||||
'db_table': 'user_access',
|
||||
'ordering': ['user_name'],
|
||||
},
|
||||
),
|
||||
migrations.CreateModel(
|
||||
name='ApiKey',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('name', models.CharField(max_length=100)),
|
||||
('prefix', models.CharField(db_index=True, max_length=12)),
|
||||
('key_hash', models.CharField(max_length=64)),
|
||||
('is_active', models.BooleanField(default=True)),
|
||||
('last_used_at', models.DateTimeField(blank=True, null=True)),
|
||||
('created_at', models.DateTimeField(auto_now_add=True)),
|
||||
('updated_at', models.DateTimeField(auto_now=True)),
|
||||
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='api_keys', to=settings.AUTH_USER_MODEL)),
|
||||
],
|
||||
options={
|
||||
'db_table': 'api_keys',
|
||||
'ordering': ['-created_at'],
|
||||
},
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,16 @@
|
||||
from django.db import migrations
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
("accounts", "0001_initial"),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.RenameField(
|
||||
model_name="user",
|
||||
old_name="id",
|
||||
new_name="user_id",
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,39 @@
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
def backfill_user_status(apps, schema_editor):
|
||||
User = apps.get_model("accounts", "User")
|
||||
for user in User.objects.all():
|
||||
if user.is_superuser:
|
||||
status = "superadmin"
|
||||
flags = {"is_active": True, "is_staff": True, "is_superuser": True}
|
||||
elif not user.is_active:
|
||||
status = "inactive"
|
||||
flags = {"is_active": False, "is_staff": False, "is_superuser": False}
|
||||
else:
|
||||
status = "active"
|
||||
flags = {"is_active": True, "is_staff": False, "is_superuser": False}
|
||||
User.objects.filter(pk=user.pk).update(status=status, **flags)
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
dependencies = [
|
||||
("accounts", "0002_erd_named_primary_keys"),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AddField(
|
||||
model_name="user",
|
||||
name="status",
|
||||
field=models.CharField(
|
||||
choices=[
|
||||
("active", "Active"),
|
||||
("inactive", "Inactive"),
|
||||
("superadmin", "Super Admin"),
|
||||
],
|
||||
default="active",
|
||||
max_length=30,
|
||||
),
|
||||
),
|
||||
migrations.RunPython(backfill_user_status, migrations.RunPython.noop),
|
||||
]
|
||||
@@ -0,0 +1,29 @@
|
||||
from django.db import migrations
|
||||
|
||||
|
||||
def sync_status_flags(apps, schema_editor):
|
||||
User = apps.get_model("accounts", "User")
|
||||
for user in User.objects.all():
|
||||
if user.status == "inactive":
|
||||
user.is_active = False
|
||||
user.is_staff = False
|
||||
user.is_superuser = False
|
||||
elif user.status == "superadmin":
|
||||
user.is_active = True
|
||||
user.is_staff = False
|
||||
user.is_superuser = True
|
||||
else:
|
||||
user.is_active = True
|
||||
user.is_superuser = False
|
||||
user.save(update_fields=["is_active", "is_staff", "is_superuser", "updated_at"])
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
("accounts", "0003_user_status"),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.RunPython(sync_status_flags, migrations.RunPython.noop),
|
||||
]
|
||||
Whitespace-only changes.
@@ -0,0 +1,117 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import secrets
|
||||
|
||||
from django.contrib.auth.models import AbstractBaseUser, BaseUserManager, PermissionsMixin
|
||||
from django.db import models
|
||||
from django.utils import timezone
|
||||
|
||||
|
||||
class UserManager(BaseUserManager):
|
||||
def create_user(self, user_name: str, password: str | None = None, **extra_fields):
|
||||
if not user_name:
|
||||
raise ValueError("user_name is required")
|
||||
extra_fields.setdefault("status", User.STATUS_ACTIVE)
|
||||
user = self.model(user_name=user_name, **extra_fields)
|
||||
user.set_password(password)
|
||||
user.save(using=self._db)
|
||||
return user
|
||||
|
||||
def create_superuser(self, user_name: str, password: str | None = None, **extra_fields):
|
||||
extra_fields["status"] = User.STATUS_SUPERADMIN
|
||||
extra_fields.setdefault("is_superuser", True)
|
||||
extra_fields.setdefault("is_active", True)
|
||||
return self.create_user(user_name, password, **extra_fields)
|
||||
|
||||
|
||||
class User(AbstractBaseUser, PermissionsMixin):
|
||||
"""ERD User Access — username field is user_name."""
|
||||
|
||||
STATUS_ACTIVE = "active"
|
||||
STATUS_INACTIVE = "inactive"
|
||||
STATUS_SUPERADMIN = "superadmin"
|
||||
STATUS_CHOICES = [
|
||||
(STATUS_ACTIVE, "Active"),
|
||||
(STATUS_INACTIVE, "Inactive"),
|
||||
(STATUS_SUPERADMIN, "Super Admin"),
|
||||
]
|
||||
|
||||
user_id = models.BigAutoField(primary_key=True)
|
||||
user_name = models.CharField(max_length=30, unique=True)
|
||||
status = models.CharField(max_length=30, choices=STATUS_CHOICES, default=STATUS_ACTIVE)
|
||||
is_staff = models.BooleanField(default=False)
|
||||
is_active = models.BooleanField(default=True)
|
||||
created_at = models.DateTimeField(auto_now_add=True)
|
||||
updated_at = models.DateTimeField(auto_now=True)
|
||||
|
||||
objects = UserManager()
|
||||
|
||||
USERNAME_FIELD = "user_name"
|
||||
REQUIRED_FIELDS: list[str] = []
|
||||
|
||||
class Meta:
|
||||
db_table = "user_access"
|
||||
ordering = ["user_name"]
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.user_name
|
||||
|
||||
def save(self, *args, **kwargs):
|
||||
if self.status == self.STATUS_INACTIVE:
|
||||
self.is_active = False
|
||||
self.is_staff = False
|
||||
self.is_superuser = False
|
||||
elif self.status == self.STATUS_SUPERADMIN:
|
||||
self.is_active = True
|
||||
self.is_staff = False
|
||||
self.is_superuser = True
|
||||
else:
|
||||
self.is_active = True
|
||||
self.is_superuser = False
|
||||
# is_staff is left untouched for active users, so an operator can be
|
||||
# explicitly flagged as staff without being a superadmin.
|
||||
super().save(*args, **kwargs)
|
||||
|
||||
|
||||
class ApiKey(models.Model):
|
||||
"""Hashed API keys for X-API-Key authentication."""
|
||||
|
||||
user = models.ForeignKey(User, on_delete=models.CASCADE, related_name="api_keys")
|
||||
name = models.CharField(max_length=100)
|
||||
prefix = models.CharField(max_length=12, db_index=True)
|
||||
key_hash = models.CharField(max_length=64)
|
||||
is_active = models.BooleanField(default=True)
|
||||
last_used_at = models.DateTimeField(null=True, blank=True)
|
||||
created_at = models.DateTimeField(auto_now_add=True)
|
||||
updated_at = models.DateTimeField(auto_now=True)
|
||||
|
||||
class Meta:
|
||||
db_table = "api_keys"
|
||||
ordering = ["-created_at"]
|
||||
|
||||
def __str__(self) -> str:
|
||||
return f"{self.name} ({self.prefix}…)"
|
||||
|
||||
@staticmethod
|
||||
def hash_key(raw_key: str) -> str:
|
||||
return hashlib.sha256(raw_key.encode("utf-8")).hexdigest()
|
||||
|
||||
@classmethod
|
||||
def generate(cls, user: User, name: str) -> tuple[ApiKey, str]:
|
||||
raw = secrets.token_urlsafe(32)
|
||||
prefix = raw[:8]
|
||||
instance = cls.objects.create(
|
||||
user=user,
|
||||
name=name,
|
||||
prefix=prefix,
|
||||
key_hash=cls.hash_key(raw),
|
||||
)
|
||||
return instance, raw
|
||||
|
||||
def verify(self, raw_key: str) -> bool:
|
||||
return secrets.compare_digest(self.key_hash, self.hash_key(raw_key))
|
||||
|
||||
def touch(self) -> None:
|
||||
self.last_used_at = timezone.now()
|
||||
self.save(update_fields=["last_used_at", "updated_at"])
|
||||
@@ -0,0 +1,45 @@
|
||||
from django.contrib.auth import get_user_model
|
||||
from rest_framework import serializers
|
||||
|
||||
from apps.accounts.models import ApiKey
|
||||
|
||||
User = get_user_model()
|
||||
|
||||
|
||||
class PkAsIdMixin(serializers.Serializer):
|
||||
"""Keep JSON `id` stable after ERD-named primary keys."""
|
||||
|
||||
id = serializers.IntegerField(source="pk", read_only=True)
|
||||
|
||||
|
||||
class UserSerializer(PkAsIdMixin, serializers.ModelSerializer):
|
||||
class Meta:
|
||||
model = User
|
||||
fields = ["id", "user_name", "status", "created_at", "updated_at"]
|
||||
read_only_fields = ["id", "created_at", "updated_at"]
|
||||
|
||||
|
||||
class LoginSerializer(serializers.Serializer):
|
||||
user_name = serializers.CharField(max_length=30)
|
||||
password = serializers.CharField(write_only=True)
|
||||
|
||||
|
||||
class ApiKeySerializer(serializers.ModelSerializer):
|
||||
class Meta:
|
||||
model = ApiKey
|
||||
fields = [
|
||||
"id",
|
||||
"name",
|
||||
"prefix",
|
||||
"is_active",
|
||||
"last_used_at",
|
||||
"created_at",
|
||||
"updated_at",
|
||||
"user",
|
||||
]
|
||||
read_only_fields = fields
|
||||
|
||||
|
||||
class ApiKeyCreateSerializer(serializers.Serializer):
|
||||
name = serializers.CharField(max_length=100)
|
||||
user_id = serializers.IntegerField(required=False)
|
||||
@@ -0,0 +1,23 @@
|
||||
from django.urls import path
|
||||
from rest_framework.routers import DefaultRouter
|
||||
|
||||
from apps.accounts.views import (
|
||||
ApiKeyViewSet,
|
||||
CsrfView,
|
||||
LoginView,
|
||||
LogoutView,
|
||||
MeView,
|
||||
UserViewSet,
|
||||
)
|
||||
|
||||
router = DefaultRouter()
|
||||
router.register("users", UserViewSet, basename="user")
|
||||
router.register("api-keys", ApiKeyViewSet, basename="api-key")
|
||||
|
||||
urlpatterns = [
|
||||
path("auth/csrf/", CsrfView.as_view(), name="auth-csrf"),
|
||||
path("auth/login/", LoginView.as_view(), name="auth-login"),
|
||||
path("auth/logout/", LogoutView.as_view(), name="auth-logout"),
|
||||
path("auth/me/", MeView.as_view(), name="auth-me"),
|
||||
*router.urls,
|
||||
]
|
||||
@@ -0,0 +1,117 @@
|
||||
from django.contrib.auth import authenticate, login, logout
|
||||
from django.middleware.csrf import get_token
|
||||
from django.views.decorators.csrf import ensure_csrf_cookie
|
||||
from django.utils.decorators import method_decorator
|
||||
from rest_framework import permissions, status, viewsets
|
||||
from rest_framework.response import Response
|
||||
from rest_framework.views import APIView
|
||||
|
||||
from apps.accounts.models import ApiKey, User
|
||||
from apps.accounts.serializers import (
|
||||
ApiKeyCreateSerializer,
|
||||
ApiKeySerializer,
|
||||
LoginSerializer,
|
||||
UserSerializer,
|
||||
)
|
||||
|
||||
|
||||
class IsOwnerOrStaff(permissions.BasePermission):
|
||||
def has_object_permission(self, request, view, obj):
|
||||
if request.user.is_staff or request.user.is_superuser:
|
||||
return True
|
||||
return getattr(obj, "user_id", None) == request.user.pk or obj == request.user
|
||||
|
||||
|
||||
class IsAdminUser(permissions.IsAdminUser):
|
||||
"""Admin access for staff OR superadmin (superadmins may have is_staff=False)."""
|
||||
|
||||
def has_permission(self, request, view):
|
||||
return bool(request.user and (request.user.is_staff or request.user.is_superuser))
|
||||
|
||||
|
||||
class UserViewSet(viewsets.ModelViewSet):
|
||||
queryset = User.objects.all().order_by("user_name")
|
||||
serializer_class = UserSerializer
|
||||
permission_classes = [IsAdminUser]
|
||||
|
||||
|
||||
class ApiKeyViewSet(viewsets.ModelViewSet):
|
||||
http_method_names = ["get", "post", "delete", "head", "options"]
|
||||
|
||||
def get_queryset(self):
|
||||
qs = ApiKey.objects.select_related("user").order_by("-created_at")
|
||||
if self.request.user.is_staff or self.request.user.is_superuser:
|
||||
return qs
|
||||
return qs.filter(user=self.request.user)
|
||||
|
||||
def get_serializer_class(self):
|
||||
if self.action == "create":
|
||||
return ApiKeyCreateSerializer
|
||||
return ApiKeySerializer
|
||||
|
||||
def get_permissions(self):
|
||||
return [permissions.IsAuthenticated()]
|
||||
|
||||
def create(self, request, *args, **kwargs):
|
||||
serializer = self.get_serializer(data=request.data)
|
||||
serializer.is_valid(raise_exception=True)
|
||||
name = serializer.validated_data["name"]
|
||||
user = request.user
|
||||
if (request.user.is_staff or request.user.is_superuser) and serializer.validated_data.get(
|
||||
"user_id"
|
||||
):
|
||||
user = User.objects.get(pk=serializer.validated_data["user_id"])
|
||||
api_key, raw = ApiKey.generate(user=user, name=name)
|
||||
data = ApiKeySerializer(api_key).data
|
||||
data["raw_key"] = raw
|
||||
return Response(data, status=status.HTTP_201_CREATED)
|
||||
|
||||
def perform_destroy(self, instance):
|
||||
if (
|
||||
not self.request.user.is_staff
|
||||
and not self.request.user.is_superuser
|
||||
and instance.user_id != self.request.user.pk
|
||||
):
|
||||
raise permissions.PermissionDenied()
|
||||
instance.delete()
|
||||
|
||||
|
||||
@method_decorator(ensure_csrf_cookie, name="dispatch")
|
||||
class CsrfView(APIView):
|
||||
permission_classes = [permissions.AllowAny]
|
||||
authentication_classes = []
|
||||
|
||||
def get(self, request):
|
||||
return Response({"csrfToken": get_token(request)})
|
||||
|
||||
|
||||
class LoginView(APIView):
|
||||
permission_classes = [permissions.AllowAny]
|
||||
authentication_classes = []
|
||||
|
||||
def post(self, request):
|
||||
serializer = LoginSerializer(data=request.data)
|
||||
serializer.is_valid(raise_exception=True)
|
||||
user = authenticate(
|
||||
request,
|
||||
username=serializer.validated_data["user_name"],
|
||||
password=serializer.validated_data["password"],
|
||||
)
|
||||
if user is None:
|
||||
return Response(
|
||||
{"detail": "Invalid credentials"},
|
||||
status=status.HTTP_401_UNAUTHORIZED,
|
||||
)
|
||||
login(request, user)
|
||||
return Response(UserSerializer(user).data)
|
||||
|
||||
|
||||
class LogoutView(APIView):
|
||||
def post(self, request):
|
||||
logout(request)
|
||||
return Response(status=status.HTTP_204_NO_CONTENT)
|
||||
|
||||
|
||||
class MeView(APIView):
|
||||
def get(self, request):
|
||||
return Response(UserSerializer(request.user).data)
|
||||
Reference in new issue
Block a user