添加用户鉴权

mini-program/app/security.py

@@ -0,0 +1,64 @@
+from datetime import UTC, datetime, timedelta
+
+import jwt
+from fastapi import HTTPException, Request, status
+from starlette.responses import JSONResponse
+
+try:
+    # For module mode: `uvicorn app.main:app`
+    from app.settings import settings
+except ModuleNotFoundError:
+    # For script mode: `python app/main.py` or VS Code "Run Python File"
+    from settings import settings
+
+
+def create_access_token(user_id: int) -> tuple[str, int]:
+    now = datetime.now(UTC)
+    expires_delta = timedelta(minutes=settings.jwt_access_token_expire_minutes)
+    expires_at = now + expires_delta
+    payload = {
+        "sub": str(user_id),
+        "token_type": "access",
+        "iat": int(now.timestamp()),
+        "exp": int(expires_at.timestamp()),
+    }
+    token = jwt.encode(
+        payload=payload,
+        key=settings.jwt_secret,
+        algorithm=settings.jwt_algorithm,
+    )
+    return token, int(expires_delta.total_seconds())
+
+
+def decode_access_token(token: str) -> int:
+    try:
+        payload = jwt.decode(
+            jwt=token,
+            key=settings.jwt_secret,
+            algorithms=[settings.jwt_algorithm],
+        )
+    except jwt.PyJWTError as exc:
+        raise HTTPException(status_code=401, detail="invalid or expired access token") from exc
+
+    if payload.get("token_type") != "access":
+        raise HTTPException(status_code=401, detail="invalid token type")
+
+    sub = payload.get("sub")
+    if not isinstance(sub, str) or not sub.isdigit():
+        raise HTTPException(status_code=401, detail="invalid token subject")
+    return int(sub)
+
+
+def auth_error_response(detail: str = "unauthorized") -> JSONResponse:
+    return JSONResponse(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        content={"detail": detail},
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+
+
+def get_current_user_id(request: Request) -> int:
+    user_id = getattr(request.state, "user_id", None)
+    if not isinstance(user_id, int):
+        raise HTTPException(status_code=401, detail="unauthorized")
+    return user_id