92 lines
2.7 KiB
Python
92 lines
2.7 KiB
Python
import logging
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.core.config import get_settings
|
|
|
|
logger = logging.getLogger("gsfdc.audit")
|
|
|
|
|
|
def write_audit(
|
|
*,
|
|
user_id: int | None,
|
|
username: str | None,
|
|
method: str,
|
|
path: str,
|
|
status_code: int,
|
|
ip: str,
|
|
) -> None:
|
|
clean_path = (path or "").split("?", 1)[0][:255]
|
|
if not (get_settings().database_url or "").strip():
|
|
return
|
|
try:
|
|
from app.db import SessionLocal
|
|
from app.models.security import AdminAuditLog
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
db.add(
|
|
AdminAuditLog(
|
|
user_id=user_id,
|
|
username=None if username is None else username[:64],
|
|
method=(method or "")[:16],
|
|
path=clean_path,
|
|
status_code=status_code,
|
|
ip=(ip or "")[:64],
|
|
)
|
|
)
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
except Exception as exc:
|
|
logger.warning("admin audit log write failed: %s", type(exc).__name__)
|
|
|
|
|
|
def _staff_from_header(request) -> tuple[int | None, str | None]:
|
|
hinted = getattr(request.state, "audit_username", None)
|
|
username = hinted.strip()[:64] if isinstance(hinted, str) and hinted.strip() else None
|
|
user_id = None
|
|
header = request.headers.get("authorization", "")
|
|
if header.lower().startswith("bearer "):
|
|
token = header.split(" ", 1)[1].strip()
|
|
try:
|
|
from app.core.security import decode_access_token
|
|
|
|
payload = decode_access_token(token, "staff")
|
|
subject = str(payload.get("sub") or "")
|
|
if subject.isdigit():
|
|
user_id = int(subject)
|
|
except Exception:
|
|
user_id = None
|
|
if user_id is not None and username is None and (get_settings().database_url or "").strip():
|
|
try:
|
|
from app.db import SessionLocal
|
|
from app.models.rbac import AdminUser
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
found = db.scalar(select(AdminUser.username).where(AdminUser.id == user_id))
|
|
finally:
|
|
db.close()
|
|
if isinstance(found, str) and found.strip():
|
|
username = found.strip()[:64]
|
|
except Exception:
|
|
pass
|
|
return user_id, username
|
|
|
|
|
|
def record_admin_call(request, status_code: int) -> None:
|
|
path = request.url.path
|
|
if not path.startswith("/api/admin"):
|
|
return
|
|
user_id, username = _staff_from_header(request)
|
|
client = request.client.host if request.client else ""
|
|
write_audit(
|
|
user_id=user_id,
|
|
username=username,
|
|
method=request.method,
|
|
path=path,
|
|
status_code=status_code,
|
|
ip=client,
|
|
)
|