big update

backend/app/core/security.py

@@ -8,9 +8,15 @@ import time
 from typing import Tuple
 
 
-PASSWORD_HASH_ITERATIONS = int(os.getenv("PASSWORD_HASH_ITERATIONS", "120000"))
+DEFAULT_PASSWORD_HASH_ITERATIONS = 120000
+PASSWORD_HASH_ITERATIONS = int(
+    os.getenv("PASSWORD_HASH_ITERATIONS", str(DEFAULT_PASSWORD_HASH_ITERATIONS))
+)
 AUTH_SECRET_KEY = os.getenv("AUTH_SECRET_KEY", "change-this-secret-in-env")
-AUTH_TOKEN_EXPIRE_MINUTES = int(os.getenv("AUTH_TOKEN_EXPIRE_MINUTES", "10080"))
+DEFAULT_AUTH_TOKEN_EXPIRE_MINUTES = 10080
+AUTH_TOKEN_EXPIRE_MINUTES = int(
+    os.getenv("AUTH_TOKEN_EXPIRE_MINUTES", str(DEFAULT_AUTH_TOKEN_EXPIRE_MINUTES))
+)
 
 
 def hash_password(password: str) -> str:
@@ -61,6 +67,11 @@ def _urlsafe_b64encode(raw: bytes) -> str:
     return base64.urlsafe_b64encode(raw).decode("utf-8").rstrip("=")
 
 
+def _urlsafe_b64decode(raw: str) -> bytes:
+    padding = "=" * (-len(raw) % 4)
+    return base64.urlsafe_b64decode(raw + padding)
+
+
 def create_access_token(user_id: int, email: str) -> Tuple[str, int]:
     expires_in = AUTH_TOKEN_EXPIRE_MINUTES * 60
     payload = {
@@ -77,3 +88,51 @@ def create_access_token(user_id: int, email: str) -> Tuple[str, int]:
     ).digest()
     token = f"{encoded_payload}.{_urlsafe_b64encode(signature)}"
     return token, expires_in
+
+
+def decode_access_token(token: str) -> Tuple[int, str]:
+    """
+    解码并校验访问令牌，返回 (user_id, email)。
+    校验项包括：结构、签名、过期时间、字段完整性。
+    """
+    try:
+        encoded_payload, encoded_signature = token.split(".", 1)
+    except ValueError as exc:
+        raise ValueError("Invalid token format") from exc
+
+    try:
+        provided_signature = _urlsafe_b64decode(encoded_signature)
+    except Exception as exc:
+        raise ValueError("Invalid token signature encoding") from exc
+
+    expected_signature = hmac.new(
+        AUTH_SECRET_KEY.encode("utf-8"),
+        encoded_payload.encode("utf-8"),
+        hashlib.sha256,
+    ).digest()
+    if not hmac.compare_digest(provided_signature, expected_signature):
+        raise ValueError("Invalid token signature")
+
+    try:
+        payload_bytes = _urlsafe_b64decode(encoded_payload)
+        payload = json.loads(payload_bytes.decode("utf-8"))
+    except Exception as exc:
+        raise ValueError("Invalid token payload") from exc
+
+    sub = payload.get("sub")
+    email = payload.get("email")
+    exp = payload.get("exp")
+
+    if not sub or not email or exp is None:
+        raise ValueError("Token payload missing required fields")
+
+    try:
+        user_id = int(sub)
+        exp_ts = int(exp)
+    except (TypeError, ValueError) as exc:
+        raise ValueError("Invalid token payload types") from exc
+
+    if time.time() >= exp_ts:
+        raise ValueError("Token expired")
+
+    return user_id, str(email)