35 lines
1.1 KiB
Python
35 lines
1.1 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Audit log for MCP tool invocations."""
|
|
|
|
import json
|
|
from typing import Optional
|
|
|
|
from .db import get_conn
|
|
|
|
|
|
def log_tool(username: str, tool_name: str, resource_id: Optional[str] = None, detail: Optional[dict] = None) -> None:
|
|
conn = get_conn()
|
|
conn.execute(
|
|
"INSERT INTO audit_log(username,tool_name,resource_id,detail) VALUES (?,?,?,?)",
|
|
(username, tool_name, resource_id, json.dumps(detail or {}, ensure_ascii=False)),
|
|
)
|
|
conn.execute(
|
|
"DELETE FROM audit_log WHERE id NOT IN "
|
|
"(SELECT id FROM audit_log ORDER BY id DESC LIMIT 5000)"
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def list_audit(limit: int = 100, username: Optional[str] = None) -> list:
|
|
conn = get_conn()
|
|
if username:
|
|
rows = conn.execute(
|
|
"SELECT * FROM audit_log WHERE username=? ORDER BY id DESC LIMIT ?",
|
|
(username, limit),
|
|
).fetchall()
|
|
else:
|
|
rows = conn.execute(
|
|
"SELECT * FROM audit_log ORDER BY id DESC LIMIT ?", (limit,)
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|