143 lines
7.2 KiB
Python
143 lines
7.2 KiB
Python
"""Independent local operator authentication adapted from the reference session approach."""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import re
|
|
import secrets
|
|
import sqlite3
|
|
import time
|
|
|
|
from fastapi import HTTPException, Request, Response
|
|
|
|
from .db import Store, StoreConflict, identifier, now
|
|
|
|
COOKIE = "assistant_maintainer"
|
|
TTL = 12 * 60 * 60
|
|
|
|
|
|
def password_hash(password: str) -> str:
|
|
salt = secrets.token_bytes(16)
|
|
result = hashlib.scrypt(password.encode(), salt=salt, n=16384, r=8, p=1, dklen=64)
|
|
return f"scrypt${salt.hex()}${result.hex()}"
|
|
|
|
|
|
def password_matches(password: str, stored: str) -> bool:
|
|
try:
|
|
_, salt, expected = stored.split("$")
|
|
result = hashlib.scrypt(password.encode(), salt=bytes.fromhex(salt), n=16384, r=8, p=1, dklen=64)
|
|
return hmac.compare_digest(result.hex(), expected)
|
|
except (ValueError, TypeError):
|
|
return False
|
|
|
|
|
|
def token_hash(token: str) -> str:
|
|
return hashlib.sha256(token.encode()).hexdigest()
|
|
|
|
|
|
class Auth:
|
|
def __init__(self, store: Store):
|
|
self.store = store
|
|
|
|
def bootstrap_admin(self) -> bool:
|
|
"""Create the requested local account only when no maintainer exists."""
|
|
with self.store.connect(immediate=True) as conn:
|
|
if conn.execute("SELECT 1 FROM operators LIMIT 1").fetchone():
|
|
return False
|
|
conn.execute("INSERT INTO operators(id,username,password_hash,created_at) VALUES($1,$2,$3,$4)", (identifier("operator"), "admin", password_hash("123456"), now()))
|
|
return True
|
|
|
|
def session(self, request: Request) -> dict | None:
|
|
raw = request.cookies.get(COOKIE, "")
|
|
if not raw:
|
|
return None
|
|
return self.store.one("SELECT s.*,o.username FROM auth_sessions s LEFT JOIN operators o ON o.id=s.operator_id WHERE token_hash=$1 AND expires_at>$2", (token_hash(raw), int(time.time())))
|
|
|
|
def issue(self, request: Request, response: Response, operator_id: str | None = None) -> dict:
|
|
previous = request.cookies.get(COOKIE)
|
|
raw, csrf = secrets.token_urlsafe(32), secrets.token_urlsafe(32)
|
|
with self.store.connect(immediate=True) as conn:
|
|
if previous:
|
|
conn.execute("DELETE FROM auth_sessions WHERE token_hash=$1", (token_hash(previous),))
|
|
conn.execute("DELETE FROM auth_sessions WHERE expires_at<$1", (int(time.time()),))
|
|
conn.execute("INSERT INTO auth_sessions(token_hash,operator_id,csrf,expires_at,created_at) VALUES($1,$2,$3,$4,$5)", (token_hash(raw), operator_id, csrf, int(time.time()) + TTL, now()))
|
|
response.set_cookie(COOKIE, raw, max_age=TTL, httponly=True, samesite="strict", secure=request.url.scheme == "https", path="/")
|
|
return {"operator_id": operator_id, "csrf": csrf}
|
|
|
|
def payload(self, session: dict) -> dict:
|
|
initialized = bool(self.store.one("SELECT id FROM operators LIMIT 1"))
|
|
authenticated = bool(session.get("operator_id"))
|
|
user = None
|
|
if authenticated:
|
|
user = self.store.one("SELECT id,username,created_at FROM operators WHERE id=$1", (session["operator_id"],))
|
|
return {"initialized": initialized, "authenticated": authenticated, "user": user, "csrf_token": session["csrf"], "mode": "local_maintainer"}
|
|
|
|
def status(self, request: Request, response: Response) -> dict:
|
|
return self.payload(self.session(request) or self.issue(request, response))
|
|
|
|
def csrf(self, request: Request) -> dict:
|
|
session = self.session(request)
|
|
header = request.headers.get("X-CSRF-Token", "")
|
|
if not session or not header or not hmac.compare_digest(header, session["csrf"]):
|
|
raise HTTPException(403, "会话校验已失效，请刷新页面后重试")
|
|
return session
|
|
|
|
def operator(self, request: Request, mutate: bool = False) -> str:
|
|
session = self.csrf(request) if mutate else self.session(request)
|
|
if not session or not session.get("operator_id"):
|
|
raise HTTPException(401, "请先登录维护后台")
|
|
return session["operator_id"]
|
|
|
|
@staticmethod
|
|
def validate_credentials(username: str, password: str) -> tuple[str, str]:
|
|
username = username.strip()
|
|
if not re.fullmatch(r"[\w.-]{3,64}", username):
|
|
raise HTTPException(422, "维护者用户名需为 3 至 64 位字母、数字、中文、点或下划线")
|
|
if len(password) < 10 or len(password) > 256:
|
|
raise HTTPException(422, "维护者密码需为 10 至 256 个字符")
|
|
return username, password
|
|
|
|
def setup(self, request: Request, response: Response, username: str, password: str) -> dict:
|
|
self.csrf(request)
|
|
username, password = self.validate_credentials(username, password)
|
|
encoded = password_hash(password)
|
|
operator_id = identifier("operator")
|
|
try:
|
|
with self.store.connect(immediate=True) as conn:
|
|
if conn.execute("SELECT 1 FROM operators LIMIT 1").fetchone():
|
|
raise HTTPException(409, "维护者已初始化，请使用登录入口")
|
|
conn.execute("INSERT INTO operators(id,username,password_hash,created_at) VALUES($1,$2,$3,$4)", (operator_id, username, encoded, now()))
|
|
except (sqlite3.IntegrityError, StoreConflict):
|
|
raise HTTPException(409, "维护者已初始化，请使用登录入口") from None
|
|
return self.payload(self.issue(request, response, operator_id))
|
|
|
|
def login(self, request: Request, response: Response, username: str, password: str) -> dict:
|
|
self.csrf(request)
|
|
key = token_hash(username.strip().casefold())
|
|
current_time = int(time.time())
|
|
with self.store.connect(immediate=True) as conn:
|
|
attempt = conn.execute("SELECT * FROM auth_attempts WHERE key=$1", (key,)).fetchone()
|
|
if attempt and current_time - attempt["window_start"] < 300 and attempt["failures"] >= 5:
|
|
raise HTTPException(429, "尝试次数较多，请稍后再试")
|
|
user = conn.execute("SELECT * FROM operators WHERE username=$1", (username.strip(),)).fetchone()
|
|
# The fixed dummy hash keeps a nonexistent username on the scrypt path.
|
|
dummy = "scrypt$" + "00" * 16 + "$" + "00" * 64
|
|
valid = password_matches(password, user["password_hash"] if user else dummy)
|
|
if not user or not valid:
|
|
if not attempt or current_time - attempt["window_start"] >= 300:
|
|
conn.execute("INSERT INTO auth_attempts(key,failures,window_start) VALUES($1,$2,$3) ON CONFLICT(key) DO UPDATE SET failures=excluded.failures,window_start=excluded.window_start", (key, 1, current_time))
|
|
else:
|
|
conn.execute("UPDATE auth_attempts SET failures=failures+1 WHERE key=$1", (key,))
|
|
error = True
|
|
else:
|
|
conn.execute("DELETE FROM auth_attempts WHERE key=$1", (key,))
|
|
operator_id = user["id"]
|
|
error = False
|
|
if error:
|
|
raise HTTPException(401, "用户名或密码不正确")
|
|
return self.payload(self.issue(request, response, operator_id))
|
|
|
|
def logout(self, request: Request, response: Response) -> dict:
|
|
self.csrf(request)
|
|
return self.payload(self.issue(request, response))
|