48 lines
2.1 KiB
Python
48 lines
2.1 KiB
Python
from fastapi import Request
|
|
|
|
import db
|
|
from app.core.security import token_digest
|
|
|
|
|
|
# 管理后台写操作统一审计；不保存令牌明文和请求正文。
|
|
async def admin_audit_middleware(request: Request, call_next):
|
|
should_audit = (
|
|
request.url.path.startswith("/admin/")
|
|
and request.method not in ("GET", "HEAD", "OPTIONS")
|
|
and not request.url.path.startswith("/admin/audit-log/")
|
|
)
|
|
status_code = 500
|
|
try:
|
|
response = await call_next(request)
|
|
status_code = response.status_code
|
|
return response
|
|
finally:
|
|
if should_audit:
|
|
try:
|
|
principal = getattr(request.state, "admin_principal", None)
|
|
token = request.headers.get("authorization", "") or request.headers.get("X-Admin-Token", "")
|
|
conn = db.get_conn()
|
|
conn.execute(
|
|
"""INSERT INTO admin_audit_logs
|
|
(actor_hash,actor_username,actor_roles,auth_type,action,method,path,target,result,status_code,ip,user_agent)
|
|
VALUES(?,?,?,?,?,?,?,?,?,?,?,?)""",
|
|
(
|
|
principal.actor_hash if principal else (token_digest(token)[:16] if token else "anonymous"),
|
|
principal.username if principal else "anonymous",
|
|
",".join(principal.roles) if principal else "",
|
|
principal.auth_type if principal else "",
|
|
request.url.path.removeprefix("/admin/"),
|
|
request.method,
|
|
request.url.path,
|
|
request.url.query[:500],
|
|
"success" if 200 <= status_code < 400 else "fail",
|
|
status_code,
|
|
request.client.host if request.client else "",
|
|
request.headers.get("user-agent", "")[:300],
|
|
),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
except Exception as audit_error:
|
|
print("管理员审计日志写入失败:", audit_error)
|