zhanghongyu_0626
3995cb44d8
- Added `auth.py` for mock login and JWT issuance. - Introduced `chat.py` for handling chat requests with role-based access control. - Enhanced `main.py` to include new routers and middleware for tracing. - Implemented input validation in `input_guard.py` to prevent SQL injection. - Created repositories for managing agent sessions and audit logs. - Added exception handling for authorization errors. - Updated settings to include JWT configuration. - Introduced tests for authentication and input validation.
53 lines
1.6 KiB
Python
53 lines
1.6 KiB
Python
"""审计总账写入（只 INSERT）。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
from sqlalchemy import text
|
|
from sqlalchemy.engine import Engine
|
|
|
|
from app.config.database import get_agent_engine
|
|
from app.model.schemas import AgentType
|
|
|
|
|
|
class AuditRepository:
|
|
def __init__(self, engine: Engine | None = None) -> None:
|
|
self._engine = engine or get_agent_engine()
|
|
|
|
def insert(
|
|
self,
|
|
*,
|
|
trace_id: str,
|
|
event_type: str,
|
|
agent_type: AgentType | str,
|
|
actor_id: str,
|
|
customer_id: str | None = None,
|
|
rule_id: str | None = None,
|
|
input_summary: dict[str, Any] | None = None,
|
|
decision: str | None = None,
|
|
) -> None:
|
|
sql = text(
|
|
"""
|
|
INSERT INTO audit_log
|
|
(trace_id, event_type, agent_type, actor_id, customer_id,
|
|
rule_id, input_summary, decision)
|
|
VALUES
|
|
(:trace_id, :event_type, :agent_type, :actor_id, :customer_id,
|
|
:rule_id, :input_summary, :decision)
|
|
"""
|
|
)
|
|
payload = {
|
|
"trace_id": trace_id,
|
|
"event_type": event_type,
|
|
"agent_type": agent_type if agent_type != "platform" else "platform",
|
|
"actor_id": actor_id,
|
|
"customer_id": customer_id,
|
|
"rule_id": rule_id,
|
|
"input_summary": json.dumps(input_summary, ensure_ascii=False) if input_summary else None,
|
|
"decision": decision,
|
|
}
|
|
with self._engine.begin() as conn:
|
|
conn.execute(sql, payload)
|