import re from django.contrib.auth import get_user_model from rest_framework.views import APIView from rest_framework.response import Response from rest_framework.permissions import AllowAny, IsAdminUser from rest_framework.throttling import ScopedRateThrottle from rest_framework_simplejwt.views import TokenObtainPairView, TokenRefreshView from tracker.api.envelope import EnvelopeMixin, ok # 对齐 Django UnicodeUsernameValidator(\w 默认 Unicode)——create_user 不跑校验器,入口自防 _USERNAME_RE = re.compile(r"[\w.@+-]+") class LoginView(EnvelopeMixin, TokenObtainPairView): """登录。JWT 取舍(M1 决策):access 12h / refresh 24h、无黑名单轮换—— token 泄露唯一应急是换 SECRET_KEY(全体登出),README-deploy 写明。""" throttle_classes = [ScopedRateThrottle] throttle_scope = "login" # 10/min/IP 防爆破(settings 已配) def post(self, request, *args, **kwargs): resp = super().post(request, *args, **kwargs) # 失败由 simplejwt 返回 401(信封由 mixin 补) if resp.status_code == 200: return ok({"access": resp.data["access"], "refresh": resp.data["refresh"]}) return resp class RefreshView(EnvelopeMixin, TokenRefreshView): permission_classes = [AllowAny] # TokenRefreshView 自带 AllowAny,显式声明表意 def post(self, request, *args, **kwargs): resp = super().post(request, *args, **kwargs) if resp.status_code == 200: # 成功同样走信封——与 LoginView 对齐,前端只需一种解析路径 return ok({"access": resp.data["access"]}) return resp class MeView(EnvelopeMixin, APIView): def get(self, request): return ok({"username": request.user.username, "is_admin": request.user.is_staff}) class UserCreateView(EnvelopeMixin, APIView): permission_classes = [IsAdminUser] # is_staff def post(self, request): data = request.data if isinstance(request.data, dict) else {} # JSON 数组等畸形 body → 空表(.get 会 500) username, password = data.get("username"), data.get("password") if not isinstance(username, str) or not isinstance(password, str): # 123.strip() / len(None) 会 500 return Response({"code": 400, "data": None, "message": "invalid_input"}, status=400) username = username.strip() # 字符集对齐 UnicodeUsernameValidator(create_user 不跑校验器);长度上限防 MySQL DataError(sqlite 不报错) if (not username or len(username) > 150 or not _USERNAME_RE.fullmatch(username) or len(password) < 8): return Response({"code": 400, "data": None, "message": "invalid_input"}, status=400) User = get_user_model() if User.objects.filter(username=username).exists(): return Response({"code": 409, "data": None, "message": "username_exists"}, status=409) User.objects.create_user(username=username, password=password, is_staff=False) return ok({"username": username}, "created")