from __future__ import annotations
import hmac
import html
from urllib.parse import parse_qs, quote
from fastapi import APIRouter, Depends, HTTPException, Request, status
from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse, Response
from fastapi.security import HTTPBasicCredentials
from ..auth import (
configured_password,
is_admin_authenticated,
is_records_authenticated,
security,
session_token,
verify_any_auth,
)
from ..config import (
ACCOUNTS_SESSION_COOKIE,
ADMIN_AUTH_PASSWORD,
ADMIN_SESSION_COOKIE,
BASIC_AUTH_PASSWORD,
RECORDS_SESSION_COOKIE,
SESSION_MAX_AGE,
STATIC_DIR,
)
router = APIRouter()
def safe_next_path(value: str | None) -> str:
if not value or not value.startswith("/") or value.startswith("//"):
return "/"
return value
def render_login_page(title: str, action: str, next_path: str = "/", has_error: bool = False) -> str:
escaped_title = html.escape(title)
escaped_action = html.escape(action, quote=True)
escaped_next = html.escape(next_path, quote=True)
error_html = '密码不正确，请重新输入。
' if has_error else '请输入访问密码。
'
return f"""
{escaped_title}
{escaped_title}
{error_html}
"""
@router.get("/")
def index(
request: Request,
credentials: HTTPBasicCredentials | None = Depends(security),
):
if not is_records_authenticated(request, credentials):
return RedirectResponse(url="/login", status_code=status.HTTP_303_SEE_OTHER)
return FileResponse(STATIC_DIR / "index.html")
@router.head("/")
def index_head(
request: Request,
credentials: HTTPBasicCredentials | None = Depends(security),
):
if not is_records_authenticated(request, credentials):
return RedirectResponse(url="/login", status_code=status.HTTP_303_SEE_OTHER)
return Response(status_code=status.HTTP_200_OK)
@router.get("/login")
def login_page(
request: Request,
credentials: HTTPBasicCredentials | None = Depends(security),
):
next_path = safe_next_path(request.query_params.get("next"))
has_error = request.query_params.get("error") == "1"
if is_records_authenticated(request, credentials):
return RedirectResponse(url=next_path, status_code=status.HTTP_303_SEE_OTHER)
return HTMLResponse(
render_login_page(
title="新时空教务管家",
action="/login",
next_path=next_path,
has_error=has_error,
)
)
@router.post("/login")
async def login_submit(request: Request):
body = (await request.body()).decode("utf-8")
form = parse_qs(body, keep_blank_values=True)
password = form.get("password", [""])[0]
next_path = safe_next_path(form.get("next", ["/"])[0])
records_password = configured_password("课程记录", BASIC_AUTH_PASSWORD)
if hmac.compare_digest(password, records_password):
response = RedirectResponse(url=next_path, status_code=status.HTTP_303_SEE_OTHER)
response.set_cookie(
RECORDS_SESSION_COOKIE,
session_token(records_password, b"xsk-records-web-session-v1"),
max_age=SESSION_MAX_AGE,
httponly=True,
samesite="lax",
)
return response
error_url = f"/login?error=1&next={quote(next_path)}"
return RedirectResponse(url=error_url, status_code=status.HTTP_303_SEE_OTHER)
@router.get("/logout")
def logout():
response = RedirectResponse(url="/login", status_code=status.HTTP_303_SEE_OTHER)
response.delete_cookie(RECORDS_SESSION_COOKIE)
return response
@router.get("/accounts")
def accounts_index():
return RedirectResponse(url="/admin", status_code=status.HTTP_303_SEE_OTHER)
@router.get("/admin")
def admin_index(
request: Request,
credentials: HTTPBasicCredentials | None = Depends(security),
):
if not is_admin_authenticated(request, credentials):
return RedirectResponse(url="/admin/login?next=/admin", status_code=status.HTTP_303_SEE_OTHER)
return FileResponse(STATIC_DIR / "admin.html")
@router.get("/admin/login")
def admin_login_page(
request: Request,
credentials: HTTPBasicCredentials | None = Depends(security),
):
next_path = safe_next_path(request.query_params.get("next") or "/admin")
has_error = request.query_params.get("error") == "1"
if is_admin_authenticated(request, credentials):
return RedirectResponse(url=next_path, status_code=status.HTTP_303_SEE_OTHER)
return HTMLResponse(render_login_page("管理后台", "/admin/login", next_path, has_error))
@router.post("/admin/login")
async def admin_login_submit(request: Request):
body = (await request.body()).decode("utf-8")
form = parse_qs(body, keep_blank_values=True)
password = form.get("password", [""])[0]
next_path = safe_next_path(form.get("next", ["/admin"])[0])
admin_password = configured_password("管理后台", ADMIN_AUTH_PASSWORD)
if hmac.compare_digest(password, admin_password):
response = RedirectResponse(url=next_path, status_code=status.HTTP_303_SEE_OTHER)
response.set_cookie(
ADMIN_SESSION_COOKIE,
session_token(admin_password, b"xsk-admin-web-session-v1"),
max_age=SESSION_MAX_AGE,
httponly=True,
samesite="lax",
)
return response
error_url = f"/admin/login?error=1&next={quote(next_path)}"
return RedirectResponse(url=error_url, status_code=status.HTTP_303_SEE_OTHER)
@router.get("/admin/logout")
def admin_logout():
response = RedirectResponse(url="/admin/login", status_code=status.HTTP_303_SEE_OTHER)
response.delete_cookie(ADMIN_SESSION_COOKIE)
response.delete_cookie(ACCOUNTS_SESSION_COOKIE)
return response
@router.get("/accounts/login")
def accounts_login_page():
return RedirectResponse(url="/admin/login?next=/admin", status_code=status.HTTP_303_SEE_OTHER)
@router.post("/accounts/login")
async def accounts_login_submit(request: Request):
return await admin_login_submit(request)
@router.get("/accounts/logout")
def accounts_logout():
response = RedirectResponse(url="/admin/login", status_code=status.HTTP_303_SEE_OTHER)
response.delete_cookie(ACCOUNTS_SESSION_COOKIE)
response.delete_cookie(ADMIN_SESSION_COOKIE)
return response
@router.get("/static/{asset_path:path}")
def static_asset(asset_path: str, _user: str = Depends(verify_any_auth)):
target = (STATIC_DIR / asset_path).resolve()
static_root = STATIC_DIR.resolve()
if not target.is_file() or static_root not in target.parents:
raise HTTPException(status_code=404, detail="静态资源不存在")
return FileResponse(target)