261 lines
11 KiB
Python
261 lines
11 KiB
Python
"""Optional browser form authentication for the independently deployed web app.
|
|
|
|
Authentication is disabled for the original local deployment. Cloud/preview
|
|
deployments opt in explicitly; invalid enabled configuration aborts startup.
|
|
There is deliberately no HTTP Basic challenge, including on API failures.
|
|
"""
|
|
import asyncio
|
|
import base64
|
|
from collections import OrderedDict, deque
|
|
from dataclasses import dataclass
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import math
|
|
import os
|
|
from pathlib import Path
|
|
import threading
|
|
import time
|
|
from time import monotonic
|
|
from urllib.parse import urlsplit
|
|
|
|
from fastapi import Request
|
|
from starlette.responses import FileResponse, JSONResponse, RedirectResponse
|
|
|
|
|
|
COOKIE_NAME = "wolin_session"
|
|
SESSION_SECONDS = 8 * 60 * 60
|
|
MAX_LOGIN_BYTES = 4096
|
|
LOGIN_HTML = Path(__file__).resolve().parents[1] / "frontend" / "login.html"
|
|
|
|
|
|
def _flag(name, default):
|
|
value = os.getenv(name, default).strip().lower()
|
|
if value in {"true", "1", "yes", "on"}:
|
|
return True
|
|
if value in {"false", "0", "no", "off"}:
|
|
return False
|
|
raise RuntimeError(f"{name} must be true or false")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _Config:
|
|
username: str
|
|
salt: bytes
|
|
password_key: bytes
|
|
secret: bytes
|
|
secure: bool
|
|
|
|
@classmethod
|
|
def from_environment(cls):
|
|
username = os.getenv("WOLIN_AUTH_USER", "admin")
|
|
secret = os.getenv("WOLIN_SESSION_SECRET", "")
|
|
if not username or len(username) > 128 or len(secret) < 32:
|
|
raise RuntimeError("Web authentication requires a valid user and a session secret of at least 32 characters")
|
|
try:
|
|
algorithm, n, r, p, salt_hex, key_hex = os.getenv("WOLIN_AUTH_PASSWORD_HASH", "").split("$")
|
|
if (algorithm, n, r, p) != ("scrypt", "16384", "8", "1"):
|
|
raise ValueError()
|
|
if len(salt_hex) != 32 or len(key_hex) != 64:
|
|
raise ValueError()
|
|
salt, key = bytes.fromhex(salt_hex), bytes.fromhex(key_hex)
|
|
if len(salt) != 16 or len(key) != 32:
|
|
raise ValueError()
|
|
except (ValueError, TypeError):
|
|
raise RuntimeError("WOLIN_AUTH_PASSWORD_HASH must be a valid scrypt password hash") from None
|
|
return cls(username, salt, key, secret.encode("utf-8"), _flag("AUTH_COOKIE_SECURE", "true"))
|
|
|
|
|
|
def _base_path(scope):
|
|
return scope.get("root_path", "").rstrip("/")
|
|
|
|
|
|
def _route_path(scope):
|
|
path = scope.get("path", "/")
|
|
root = _base_path(scope)
|
|
if root and (path == root or path.startswith(root + "/")):
|
|
return path[len(root):] or "/"
|
|
return path
|
|
|
|
|
|
def _json(content, status=200, headers=None):
|
|
return JSONResponse(content, status_code=status, headers={"Cache-Control": "no-store", **(headers or {})})
|
|
|
|
|
|
def _encode(data):
|
|
return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
|
|
|
|
|
|
def _decode(value):
|
|
return base64.b64decode(value + "=" * (-len(value) % 4), altchars=b"-_", validate=True)
|
|
|
|
|
|
def _session(config):
|
|
now = int(time.time())
|
|
payload = _encode(json.dumps({"u": config.username, "iat": now, "exp": now + SESSION_SECONDS}, separators=(",", ":")).encode("utf-8"))
|
|
signature = _encode(hmac.digest(config.secret, payload.encode("ascii"), "sha256"))
|
|
return payload + "." + signature
|
|
|
|
|
|
def _valid_session(value, config):
|
|
if not value or len(value) > 1024:
|
|
return False
|
|
try:
|
|
payload, signature = value.split(".")
|
|
expected = hmac.digest(config.secret, payload.encode("ascii"), "sha256")
|
|
if not hmac.compare_digest(expected, _decode(signature)):
|
|
return False
|
|
claims = json.loads(_decode(payload))
|
|
issued, expires = claims["iat"], claims["exp"]
|
|
now = time.time()
|
|
return (claims["u"] == config.username
|
|
and type(issued) is int and type(expires) is int
|
|
and expires - issued == SESSION_SECONDS
|
|
and issued <= now + 60 and now < expires)
|
|
except (ValueError, TypeError, KeyError, UnicodeError):
|
|
return False
|
|
|
|
|
|
def _cross_site(request):
|
|
if request.headers.get("sec-fetch-site", "").lower() == "cross-site":
|
|
return True
|
|
origin = request.headers.get("origin")
|
|
if origin is None:
|
|
return False
|
|
try:
|
|
incoming = urlsplit(origin)
|
|
target = request.url
|
|
incoming_port = incoming.port or (443 if incoming.scheme == "https" else 80)
|
|
target_port = target.port or (443 if target.scheme == "https" else 80)
|
|
return (incoming.scheme not in {"http", "https"}
|
|
or incoming.username is not None or incoming.password is not None
|
|
or incoming.path not in {"", "/"} or bool(incoming.query or incoming.fragment)
|
|
or incoming.scheme != target.scheme
|
|
or incoming.hostname != target.hostname
|
|
or incoming_port != target_port)
|
|
except (ValueError, TypeError):
|
|
return True
|
|
|
|
|
|
class _LoginLimiter:
|
|
"""Bounded per-process attempts; successful authentication clears failures."""
|
|
def __init__(self):
|
|
self.entries = OrderedDict()
|
|
self.lock = threading.Lock()
|
|
|
|
def reserve(self, address):
|
|
now = monotonic()
|
|
with self.lock:
|
|
for key in list(self.entries):
|
|
if self.entries[key][-1] <= now - 60:
|
|
del self.entries[key]
|
|
attempts = self.entries.setdefault(address, deque())
|
|
while attempts and attempts[0] <= now - 60:
|
|
attempts.popleft()
|
|
self.entries.move_to_end(address)
|
|
if len(attempts) >= 5:
|
|
return max(1, math.ceil(60 - (now - attempts[0])))
|
|
attempts.append(now)
|
|
while len(self.entries) > 4096:
|
|
self.entries.popitem(last=False)
|
|
return 0
|
|
|
|
def clear(self, address):
|
|
with self.lock:
|
|
self.entries.pop(address, None)
|
|
|
|
|
|
class _AuthMiddleware:
|
|
def __init__(self, app, config):
|
|
self.app = app
|
|
self.config = config
|
|
|
|
async def __call__(self, scope, receive, send):
|
|
if scope["type"] != "http":
|
|
if scope["type"] == "websocket":
|
|
# No unauthenticated websocket path may bypass the web gate.
|
|
await send({"type": "websocket.close", "code": 1008})
|
|
return
|
|
await self.app(scope, receive, send)
|
|
return
|
|
request = Request(scope, receive)
|
|
path = _route_path(scope)
|
|
method = request.method
|
|
public = ((method in {"GET", "HEAD"} and path == "/login")
|
|
or (method == "GET" and path == "/healthz")
|
|
or (method == "POST" and path in {"/auth/login", "/auth/logout"}))
|
|
authenticated = _valid_session(request.cookies.get(COOKIE_NAME), self.config)
|
|
check_origin = ((method == "POST" and path in {"/auth/login", "/auth/logout"})
|
|
or (authenticated and method in {"POST", "PUT", "PATCH", "DELETE"}))
|
|
if check_origin and _cross_site(request):
|
|
response = _json({"detail": "不允许跨站请求。"}, 403)
|
|
elif public or authenticated:
|
|
await self.app(scope, receive, send)
|
|
return
|
|
else:
|
|
api_path = (path == "/api" or path.startswith("/api/")
|
|
or path in {"/docs", "/redoc", "/openapi.json", "/static", "/auth/login", "/auth/logout"}
|
|
or path.startswith(("/static/", "/docs/", "/redoc/")))
|
|
if method in {"GET", "HEAD"} and "text/html" in request.headers.get("accept", "") and not api_path:
|
|
response = RedirectResponse(_base_path(scope) + "/login", status_code=303, headers={"Cache-Control": "no-store"})
|
|
else:
|
|
response = _json({"detail": "请先登录沃林学生管理系统。"}, 401)
|
|
await response(scope, receive, send)
|
|
|
|
|
|
def install_web_auth(app):
|
|
"""Install once, after other middleware, so authentication is outermost."""
|
|
if not _flag("AUTH_ENABLED", "false"):
|
|
return
|
|
config = _Config.from_environment()
|
|
limiter = _LoginLimiter()
|
|
|
|
async def login_page(request: Request):
|
|
return FileResponse(LOGIN_HTML, media_type="text/html", headers={"Cache-Control": "no-store"})
|
|
|
|
async def login(request: Request):
|
|
address = request.client.host if request.client else "unknown"
|
|
retry_after = limiter.reserve(address)
|
|
if retry_after:
|
|
return _json({"detail": "登录尝试过多，请稍后重试。"}, 429, {"Retry-After": str(retry_after)})
|
|
if request.headers.get("content-type", "").split(";")[0].strip().lower() != "application/json":
|
|
return _json({"detail": "请使用 JSON 提交登录信息。"}, 415)
|
|
body = bytearray()
|
|
try:
|
|
async for chunk in request.stream():
|
|
if len(body) + len(chunk) > MAX_LOGIN_BYTES:
|
|
return _json({"detail": "登录信息过长。"}, 413)
|
|
body.extend(chunk)
|
|
data = json.loads(body)
|
|
username, password = data.get("username"), data.get("password")
|
|
if (not isinstance(username, str) or not isinstance(password, str)
|
|
or not 1 <= len(username) <= 128 or not 1 <= len(password) <= 1024):
|
|
return _json({"detail": "登录信息格式不正确。"}, 400)
|
|
except (ValueError, UnicodeError, AttributeError, TypeError):
|
|
return _json({"detail": "登录信息格式不正确。"}, 400)
|
|
try:
|
|
candidate = await asyncio.to_thread(hashlib.scrypt, password.encode("utf-8"), salt=config.salt, n=16384, r=8, p=1, dklen=32, maxmem=64 * 1024 * 1024)
|
|
valid_password = hmac.compare_digest(candidate, config.password_key)
|
|
valid_user = hmac.compare_digest(username.encode("utf-8"), config.username.encode("utf-8"))
|
|
except (ValueError, UnicodeError):
|
|
return _json({"detail": "登录信息格式不正确。"}, 400)
|
|
if not (valid_password and valid_user):
|
|
return _json({"detail": "账号或密码错误。"}, 401)
|
|
limiter.clear(address)
|
|
response = _json({"success": True})
|
|
response.set_cookie(COOKIE_NAME, _session(config), max_age=SESSION_SECONDS,
|
|
path=_base_path(request.scope) + "/", secure=config.secure,
|
|
httponly=True, samesite="strict")
|
|
return response
|
|
|
|
async def logout(request: Request):
|
|
response = _json({"success": True})
|
|
response.delete_cookie(COOKIE_NAME, path=_base_path(request.scope) + "/",
|
|
secure=config.secure, httponly=True, samesite="strict")
|
|
return response
|
|
|
|
app.add_api_route("/login", login_page, methods=["GET", "HEAD"], include_in_schema=False)
|
|
app.add_api_route("/auth/login", login, methods=["POST"], include_in_schema=False)
|
|
app.add_api_route("/auth/logout", logout, methods=["POST"], include_in_schema=False)
|
|
app.add_middleware(_AuthMiddleware, config=config)
|