import os, secrets, hashlib, hmac, time
from fastapi import FastAPI, Form, Request
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
from starlette.middleware.base import BaseHTTPMiddleware
PWD = os.environ.get("COPAW_PASSWORD", "")
SEC = os.environ.get("COPAW_SESSION_SECRET", "") or (secrets.token_hex(32) if PWD else "")
MAX = int(os.environ.get("COPAW_SESSION_MAX_AGE", "86400"))
ON = bool(PWD)
SKIP = {"/auth/login", "/auth/logout", "/health"}
EXTS = (".js", ".css", ".png", ".svg", ".ico", ".woff", ".woff2", ".ttf", ".jpg", ".gif", ".wasm")
def _mk():
t = str(int(time.time()))
k = secrets.token_urlsafe(32)
s = hmac.new(SEC.encode(), f"{k}:{t}".encode(), hashlib.sha256).hexdigest()
return f"{k}:{t}:{s}"
def _ok(v):
if not v:
return False
try:
k, t, s = v.split(":")
expected = hmac.new(SEC.encode(), f"{k}:{t}".encode(), hashlib.sha256).hexdigest()
return hmac.compare_digest(s, expected) and time.time() - int(t) <= MAX
except:
return False
LOGIN_HTML = """
Login
"""
class AuthMiddleware(BaseHTTPMiddleware):
async def dispatch(self, req, call_next):
path = req.url.path
# 直接在中间件处理 auth 路由（避免被 catch-all 拦截）
if path == "/auth/login":
if req.method == "GET":
return HTMLResponse(content=LOGIN_HTML)
elif req.method == "POST":
# 解析 form 数据
form = await req.form()
password = form.get("password", "")
if ON and secrets.compare_digest(password, PWD):
r = JSONResponse({"ok": True, "redirect": "/"})
r.set_cookie("copaw_session", _mk(), max_age=MAX, httponly=True, samesite="lax")
return r
return JSONResponse({"error": "Invalid password"}, status_code=401)
if path == "/auth/logout":
r = RedirectResponse(url="/auth/login", status_code=302)
r.delete_cookie("copaw_session")
return r
if path == "/auth/status":
return JSONResponse({"authenticated": _ok(req.cookies.get("copaw_session")), "auth_required": ON})
# 未启用鉴权则直接放行
if not ON:
return await call_next(req)
# 静态资源放行
if any(path.endswith(e) for e in EXTS):
return await call_next(req)
# 已登录则放行
if _ok(req.cookies.get("copaw_session")):
return await call_next(req)
# API 返回 401，其他跳转登录
if path.startswith("/api"):
return JSONResponse({"error": "Unauthorized"}, status_code=401)
return RedirectResponse(f"/auth/login?next={path}", status_code=302)
def setup_auth(app: FastAPI):
if ON:
print(f"[Auth] ENABLED - password protected")
else:
print(f"[Auth] DISABLED - no COPAW_PASSWORD set")
app.add_middleware(AuthMiddleware)