Switch auth password hashing to Argon2
This commit is contained in:
@@ -175,7 +175,7 @@ uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
|
||||
|
||||
安全实现的当前边界:
|
||||
|
||||
- 密码使用 scrypt 做哈希存储
|
||||
- 密码使用 Argon2 做哈希存储
|
||||
- session cookie 使用 `HttpOnly`
|
||||
- `Secure` 默认随 `APP_ENV` 切换:非 development 时默认开启
|
||||
- `SameSite=Lax`
|
||||
|
||||
+11
-45
@@ -1,12 +1,13 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import logging
|
||||
import secrets
|
||||
from dataclasses import dataclass
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
from argon2 import PasswordHasher
|
||||
from argon2.exceptions import InvalidHashError, VerificationError, VerifyMismatchError
|
||||
from sqlalchemy import Select, select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
@@ -14,11 +15,7 @@ from app.config import Settings
|
||||
from app.models.auth import AuthSession, AuthUser
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
SCRYPT_N = 2**14
|
||||
SCRYPT_R = 8
|
||||
SCRYPT_P = 1
|
||||
SCRYPT_DKLEN = 64
|
||||
password_hasher = PasswordHasher()
|
||||
|
||||
|
||||
class AuthBootstrapError(RuntimeError):
|
||||
@@ -62,52 +59,17 @@ def initialize_auth_schema(session: Session, settings: Settings) -> None:
|
||||
|
||||
|
||||
def hash_password(password: str) -> str:
|
||||
salt = secrets.token_bytes(16)
|
||||
derived_key = hashlib.scrypt(
|
||||
password.encode("utf-8"),
|
||||
salt=salt,
|
||||
n=SCRYPT_N,
|
||||
r=SCRYPT_R,
|
||||
p=SCRYPT_P,
|
||||
dklen=SCRYPT_DKLEN,
|
||||
)
|
||||
return "$".join(
|
||||
[
|
||||
"scrypt",
|
||||
str(SCRYPT_N),
|
||||
str(SCRYPT_R),
|
||||
str(SCRYPT_P),
|
||||
base64.b64encode(salt).decode("ascii"),
|
||||
base64.b64encode(derived_key).decode("ascii"),
|
||||
]
|
||||
)
|
||||
return password_hasher.hash(password)
|
||||
|
||||
|
||||
def verify_password(password: str, stored_hash: str) -> bool:
|
||||
try:
|
||||
algorithm, n, r, p, encoded_salt, encoded_key = stored_hash.split("$")
|
||||
except ValueError:
|
||||
return password_hasher.verify(stored_hash, password)
|
||||
except VerifyMismatchError:
|
||||
return False
|
||||
|
||||
if algorithm != "scrypt":
|
||||
except (InvalidHashError, VerificationError):
|
||||
return False
|
||||
|
||||
try:
|
||||
salt = base64.b64decode(encoded_salt.encode("ascii"))
|
||||
expected_key = base64.b64decode(encoded_key.encode("ascii"))
|
||||
derived_key = hashlib.scrypt(
|
||||
password.encode("utf-8"),
|
||||
salt=salt,
|
||||
n=int(n),
|
||||
r=int(r),
|
||||
p=int(p),
|
||||
dklen=len(expected_key),
|
||||
)
|
||||
except (ValueError, TypeError):
|
||||
return False
|
||||
|
||||
return secrets.compare_digest(derived_key, expected_key)
|
||||
|
||||
|
||||
def authenticate_user(session: Session, *, username: str, password: str) -> AuthUser | None:
|
||||
user = session.scalar(select(AuthUser).where(AuthUser.username == username).limit(1))
|
||||
@@ -156,6 +118,10 @@ def get_authenticated_session(session: Session, *, raw_token: str | None) -> Aut
|
||||
now = _utc_now()
|
||||
expires_at = _as_utc(auth_session.expires_at)
|
||||
revoked_at = _as_utc(auth_session.revoked_at)
|
||||
if expires_at is None:
|
||||
logger.warning("Auth session %s has no expires_at; treating it as invalid", auth_session.id)
|
||||
return None
|
||||
|
||||
if revoked_at is not None or expires_at <= now or not user.is_active:
|
||||
if revoked_at is None and expires_at <= now:
|
||||
auth_session.revoked_at = now
|
||||
|
||||
@@ -8,6 +8,10 @@ alembic==1.18.4
|
||||
# via -r requirements.in
|
||||
annotated-types==0.7.0
|
||||
# via pydantic
|
||||
argon2-cffi==25.1.0
|
||||
# via -r requirements.in
|
||||
argon2-cffi-bindings==25.1.0
|
||||
# via argon2-cffi
|
||||
anyio==4.13.0
|
||||
# via
|
||||
# httpx
|
||||
@@ -19,6 +23,8 @@ certifi==2026.2.25
|
||||
# via
|
||||
# httpcore
|
||||
# httpx
|
||||
cffi==2.0.0
|
||||
# via argon2-cffi-bindings
|
||||
click==8.3.2
|
||||
# via
|
||||
# pip-tools
|
||||
@@ -82,6 +88,8 @@ python-dotenv==1.2.2
|
||||
# uvicorn
|
||||
python-multipart==0.0.26
|
||||
# via -r requirements.in
|
||||
pycparser==2.23
|
||||
# via cffi
|
||||
pyyaml==6.0.3
|
||||
# via
|
||||
# -r requirements.in
|
||||
|
||||
+1
-1
@@ -77,7 +77,7 @@
|
||||
|
||||
当前这版已经落实的基础安全点:
|
||||
|
||||
- 密码不明文存储,使用 scrypt 哈希
|
||||
- 密码不明文存储,使用 Argon2 哈希
|
||||
- session cookie 为 `HttpOnly`
|
||||
- cookie 使用 `SameSite=Lax`
|
||||
- `Secure` cookie 在非 `development` 环境默认开启
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
alembic>=1.14,<2.0
|
||||
argon2-cffi>=25.1,<26.0
|
||||
fastapi>=0.115,<0.116
|
||||
jinja2>=3.1,<4.0
|
||||
pydantic-settings>=2.6,<3.0
|
||||
|
||||
@@ -8,10 +8,16 @@ alembic==1.18.4
|
||||
# via -r requirements.in
|
||||
annotated-types==0.7.0
|
||||
# via pydantic
|
||||
argon2-cffi==25.1.0
|
||||
# via -r requirements.in
|
||||
argon2-cffi-bindings==25.1.0
|
||||
# via argon2-cffi
|
||||
anyio==4.13.0
|
||||
# via
|
||||
# starlette
|
||||
# watchfiles
|
||||
cffi==2.0.0
|
||||
# via argon2-cffi-bindings
|
||||
click==8.3.2
|
||||
# via uvicorn
|
||||
fastapi==0.115.14
|
||||
@@ -46,6 +52,8 @@ python-dotenv==1.2.2
|
||||
# uvicorn
|
||||
python-multipart==0.0.26
|
||||
# via -r requirements.in
|
||||
pycparser==2.23
|
||||
# via cffi
|
||||
pyyaml==6.0.3
|
||||
# via
|
||||
# -r requirements.in
|
||||
|
||||
+4
-7
@@ -1,12 +1,7 @@
|
||||
import re
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
pytestmark = pytest.mark.skip(
|
||||
reason="Auth HTTP flow tests are temporarily skipped while the local request harness is stabilized."
|
||||
)
|
||||
|
||||
|
||||
def _extract_csrf_token(html: str) -> str:
|
||||
match = re.search(r'name="csrf_token" value="([^"]+)"', html)
|
||||
@@ -44,8 +39,10 @@ def test_login_success_sets_session_cookie_and_allows_admin_access(client: TestC
|
||||
|
||||
admin_response = client.get("/admin")
|
||||
assert admin_response.status_code == 200
|
||||
assert "当前用户" in admin_response.text
|
||||
assert "admin" in admin_response.text
|
||||
assert "首次登录后需要先修改密码" in admin_response.text
|
||||
assert "Current Password" in admin_response.text
|
||||
assert "New Password" in admin_response.text
|
||||
assert "当前用户" not in admin_response.text
|
||||
|
||||
|
||||
def test_login_failure_returns_generic_error(client: TestClient) -> None:
|
||||
|
||||
Reference in New Issue
Block a user