diff --git a/server/config/urls.py b/server/config/urls.py index fe5268d..bfb01c5 100644 --- a/server/config/urls.py +++ b/server/config/urls.py @@ -1,6 +1,14 @@ from django.urls import path from tracker.api.collect import CollectBatchView +from tracker.api.auth import LoginView, RefreshView, MeView, UserCreateView urlpatterns = [ path("api/collect/batch", CollectBatchView.as_view()), ] + +urlpatterns += [ + path("api/admin/auth/token", LoginView.as_view()), + path("api/admin/auth/refresh", RefreshView.as_view()), + path("api/admin/auth/me", MeView.as_view()), + path("api/admin/users", UserCreateView.as_view()), +] diff --git a/server/tracker/api/auth.py b/server/tracker/api/auth.py new file mode 100644 index 0000000..e242dda --- /dev/null +++ b/server/tracker/api/auth.py @@ -0,0 +1,72 @@ +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 + + +def _ok(data, message="ok"): + return Response({"code": 200, "data": data, "message": message}) + + +class _EnvelopeMixin: + """错误信封统一收口(Task7 同款裁剪,无 ACAO——管理台与 API 同源,跨域由 nginx 层面不存在)。 + + simplejwt 认证失败 401 / 权限 403 / 限流 429 等由 DRF 异常处理器产出的响应 + body 是 {detail},管理台前端(Task17)按 {code,data,message} 信封对接—— + 在此统一转换,视图自身的 _ok/4xx 响应不受影响(只动含 detail 的 dict)。 + """ + + def finalize_response(self, request, response, *args, **kwargs): + response = super().finalize_response(request, response, *args, **kwargs) + if response.status_code >= 400 and isinstance(response.data, dict) and "detail" in response.data: + response.data = {"code": response.status_code, "data": None, + "message": response.data["detail"]} + return response + + +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): + username = (request.data.get("username") or "").strip() + password = request.data.get("password") or "" + # 用户名长度上限对齐 Django User.username(max_length=150)——create_user 不跑 + # full_clean,超长会直奔 MySQL DataError 500,故入口拦截 + if not username or len(username) > 150 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") diff --git a/server/tracker/tests/test_auth.py b/server/tracker/tests/test_auth.py new file mode 100644 index 0000000..32e98a6 --- /dev/null +++ b/server/tracker/tests/test_auth.py @@ -0,0 +1,87 @@ +import pytest +from django.contrib.auth import get_user_model +from django.core.cache import cache +from rest_framework.test import APIClient + +pytestmark = pytest.mark.django_db +User = get_user_model() + + +@pytest.fixture(autouse=True) +def _clear_cache(): + cache.clear() # locmem 限流历史跨测试残留,会让 429 用例假红(同 test_collect 模式) + + +@pytest.fixture +def client(): # 覆盖 pytest-django 的 Django Client——登录用例需要 credentials()/format="json" + return APIClient() + + +@pytest.fixture +def admin_user(): + return User.objects.create_user(username="admin", password="adminpass123", is_staff=True) + + +def _auth(client, user): + from rest_framework_simplejwt.tokens import RefreshToken + token = RefreshToken.for_user(user) + client.credentials(HTTP_AUTHORIZATION=f"Bearer {token.access_token}") + + +def test_login_ok_returns_tokens(client, admin_user): + r = client.post("/api/admin/auth/token", + {"username": "admin", "password": "adminpass123"}, format="json") + assert r.status_code == 200 + body = r.json() + assert body["code"] == 200 and body["data"]["access"] and body["data"]["refresh"] + + +def test_login_wrong_password_401_envelope(client, admin_user): + """simplejwt 失败默认 {detail} 无信封——同 Task7 经验,finalize_response 统一转 + {code,data,message},管理台前端(Task17)按信封对接。""" + r = client.post("/api/admin/auth/token", + {"username": "admin", "password": "wrongpass"}, format="json") + assert r.status_code == 401 + body = r.json() + assert body["code"] == 401 and body["data"] is None and body["message"] + + +def test_refresh_returns_new_access(client, admin_user): + from rest_framework_simplejwt.tokens import RefreshToken + refresh = RefreshToken.for_user(admin_user) + r = client.post("/api/admin/auth/refresh", {"refresh": str(refresh)}, format="json") + assert r.status_code == 200 + body = r.json() + assert body["code"] == 200 and body["data"]["access"] + + +def test_me_requires_auth_envelope(client): + r = client.get("/api/admin/auth/me") + assert r.status_code == 401 + assert r.json()["code"] == 401 # 无 token 的 401 同样走信封 + + +def test_me_returns_username_and_flag(client, admin_user): + _auth(client, admin_user) + r = client.get("/api/admin/auth/me") + assert r.json()["data"] == {"username": "admin", "is_admin": True} + + +def test_user_create_admin_only(client, admin_user): + member = User.objects.create_user(username="member", password="memberpass123") + _auth(client, member) + r = client.post("/api/admin/users", + {"username": "newbie", "password": "longenough1"}, format="json") + assert r.status_code == 403 + _auth(client, admin_user) + r = client.post("/api/admin/users", + {"username": "newbie", "password": "longenough1"}, format="json") + assert r.status_code == 200 and User.objects.filter(username="newbie", is_staff=False).exists() + + +def test_user_create_validates_input(client, admin_user): + _auth(client, admin_user) + assert client.post("/api/admin/users", + {"username": "x", "password": "short"}, format="json").status_code == 400 + assert client.post("/api/admin/users", + {"username": "admin", "password": "longenough1"}, format="json").status_code == 409