feat(api): JWT 单用户认证(auth.py)+ config 扩展
This commit is contained in:
@@ -7,3 +7,12 @@ backtest:
|
||||
api:
|
||||
host: 0.0.0.0
|
||||
port: 8000
|
||||
|
||||
auth:
|
||||
username: admin
|
||||
password_hash: "$2b$12$SGYJW1GKsCTSOAcnjxxV4.rs57OYnPni3YRGKUOqOPGTFHnqO1xdC" # default: admin — change on deploy
|
||||
jwt_secret: "change-me-in-production"
|
||||
token_expire_minutes: 60
|
||||
|
||||
pool:
|
||||
max_workers: 2
|
||||
|
||||
@@ -0,0 +1,38 @@
|
||||
# sanguo_api/auth.py
|
||||
"""JWT 单用户认证。secret/用户名/密码 hash 来自 config/backtest.yaml。"""
|
||||
import os
|
||||
from datetime import datetime, timedelta, timezone
|
||||
import jwt
|
||||
from passlib.context import CryptContext
|
||||
from fastapi import HTTPException, status
|
||||
|
||||
_CONFIG = {"secret": "change-me", "expire_minutes": 60, "algorithm": "HS256"}
|
||||
_pwd = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
||||
|
||||
|
||||
def set_jwt_config(secret: str, expire_minutes: int, algorithm: str = "HS256"):
|
||||
_CONFIG.update(secret=secret, expire_minutes=expire_minutes, algorithm=algorithm)
|
||||
|
||||
|
||||
def hash_password(password: str) -> str:
|
||||
return _pwd.hash(password)
|
||||
|
||||
|
||||
def verify_password(password: str, password_hash: str) -> bool:
|
||||
return _pwd.verify(password, password_hash)
|
||||
|
||||
|
||||
def create_token(username: str) -> str:
|
||||
payload = {
|
||||
"sub": username,
|
||||
"exp": datetime.now(timezone.utc) + timedelta(minutes=_CONFIG["expire_minutes"]),
|
||||
}
|
||||
return jwt.encode(payload, _CONFIG["secret"], algorithm=_CONFIG["algorithm"])
|
||||
|
||||
|
||||
def verify_token(token: str) -> str:
|
||||
try:
|
||||
payload = jwt.decode(token, _CONFIG["secret"], algorithms=[_CONFIG["algorithm"]])
|
||||
return payload["sub"]
|
||||
except jwt.PyJWTError:
|
||||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="无效 token")
|
||||
@@ -0,0 +1,24 @@
|
||||
# tests/api/test_auth.py
|
||||
import pytest
|
||||
|
||||
def test_create_and_verify_token():
|
||||
from sanguo_api.auth import create_token, verify_token, set_jwt_config
|
||||
set_jwt_config(secret="test_secret", expire_minutes=60)
|
||||
token = create_token("admin")
|
||||
assert isinstance(token, str)
|
||||
assert verify_token(token) == "admin"
|
||||
|
||||
def test_verify_token_invalid_raises_401():
|
||||
from fastapi import HTTPException
|
||||
from sanguo_api.auth import verify_token, set_jwt_config
|
||||
set_jwt_config(secret="test_secret", expire_minutes=60)
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
verify_token("invalid.token.here")
|
||||
assert exc.value.status_code == 401
|
||||
|
||||
def test_hash_and_verify_password():
|
||||
from sanguo_api.auth import hash_password, verify_password
|
||||
h = hash_password("mypass")
|
||||
assert h != "mypass"
|
||||
assert verify_password("mypass", h) is True
|
||||
assert verify_password("wrong", h) is False
|
||||
Reference in New Issue
Block a user