LLMGuardX Dev
ea78ef5674
P1-聊天 SSE 流式响应:
- 新增 POST /chat/sessions/{id}/messages/stream:网关 text/event-stream 实时
透传,流结束整轮落库(哈希链),上游忽略 stream 返回普通 JSON 时自动转
SSE 事件,非 2xx 错误缓冲后走统一错误处理(不落 header);
- 前端 fetch+ReadableStream 解析 SSE,占位气泡实时填充,支持停止生成
(AbortController),切会话丢弃迟到增量防串扰。
P2-管理操作审计(admin_op_logs):
- 新表+oplog 包(同步写,失败不阻塞业务);管理端查询端点
GET /api/v1/admin/op-logs(操作者/类型过滤+分页,audit:read);
- 埋点:渠道 save/delete/grant(幂等重复不重复记)/revoke_grant、账号
create/update、角色 CRUD、API Key create/revoke/limits、工具
save/delete、审批决定(资源/工具)、模型配额;管理端「操作审计」菜单。
P2-列表分页与安全上限:
- 用户/管理员列表 q+limit+offset 分页(默认 50 上限 200),渠道授权弹窗
改远程搜索,不再全量拉取 portal-users;api_keys/channels List 加
LIMIT 200 防全表扫描。
健壮性:
- ChatModels/approvedModel 对 decided_at 为 NULL 的历史批准记录
COALESCE 兜底,修复 NULL scan 报错;
- docker-compose 补 ALLOW_PRIVATE_PROVIDER_URLS 透传(默认 false)。
测试:
- portal: 流式解析/错误提取/stream writer 模式单测,会话生命周期/哈希链
完整性/200 条上限/busy 租约回收集成测试;
- channel: CRUD+加解密+部门可见性+授权撤销+幂等+审计落库集成测试。
全部通过;全量 go vet 干净。
292 lines
12 KiB
Go
292 lines
12 KiB
Go
package operations
|
|
|
|
import (
|
|
"context"
|
|
"net/http"
|
|
"runtime"
|
|
"strconv"
|
|
"time"
|
|
|
|
"aigateway.local/core/internal/identity"
|
|
"aigateway.local/core/internal/platform/apiresponse"
|
|
"aigateway.local/core/internal/platform/oplog"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
type AdminHTTPHandler struct {
|
|
pool *pgxpool.Pool
|
|
identity *identity.Service
|
|
version string
|
|
startedAt time.Time
|
|
reload func(context.Context) error
|
|
mux *http.ServeMux
|
|
}
|
|
|
|
func NewAdminHTTPHandler(pool *pgxpool.Pool, identityService *identity.Service, version string, startedAt time.Time, reload func(context.Context) error) *AdminHTTPHandler {
|
|
h := &AdminHTTPHandler{pool: pool, identity: identityService, version: version, startedAt: startedAt, reload: reload, mux: http.NewServeMux()}
|
|
h.mux.HandleFunc("GET /api/v1/admin/system-info", h.systemInfo)
|
|
h.mux.HandleFunc("GET /api/v1/admin/monitoring/overview", h.overview)
|
|
h.mux.HandleFunc("GET /api/v1/admin/tenants/overview", h.tenantsOverview)
|
|
h.mux.HandleFunc("GET /api/v1/admin/reports/tools", h.reportTools)
|
|
h.mux.HandleFunc("GET /api/v1/admin/reports/approvals", h.reportApprovals)
|
|
h.mux.HandleFunc("GET /api/v1/admin/reports/security", h.reportSecurity)
|
|
h.mux.HandleFunc("GET /api/v1/admin/op-logs", h.opLogs)
|
|
h.mux.HandleFunc("POST /api/v1/admin/reload", h.reloadSnapshots)
|
|
return h
|
|
}
|
|
|
|
// opLogs 查询管理操作审计(操作者/操作类型过滤 + 分页)。
|
|
func (h *AdminHTTPHandler) opLogs(w http.ResponseWriter, r *http.Request) {
|
|
if _, ok := h.requirePermission(w, r, identity.PermissionAuditRead); !ok {
|
|
return
|
|
}
|
|
limit, _ := strconv.Atoi(r.URL.Query().Get("limit"))
|
|
if limit < 1 || limit > 200 {
|
|
limit = 50
|
|
}
|
|
offset, _ := strconv.Atoi(r.URL.Query().Get("offset"))
|
|
if offset < 0 {
|
|
offset = 0
|
|
}
|
|
items, total, err := oplog.List(r.Context(), h.pool, r.URL.Query().Get("actor"), r.URL.Query().Get("action"), limit, offset)
|
|
if err != nil {
|
|
apiresponse.Error(w, http.StatusServiceUnavailable, "操作审计查询失败")
|
|
return
|
|
}
|
|
apiresponse.OK(w, map[string]any{"items": items, "total": total, "limit": limit, "offset": offset})
|
|
}
|
|
|
|
func (h *AdminHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { h.mux.ServeHTTP(w, r) }
|
|
|
|
func (h *AdminHTTPHandler) account(w http.ResponseWriter, r *http.Request) (identity.Account, bool) {
|
|
account, err := h.identity.Authenticate(r.Context(), identity.KindAdmin, r.Header.Get("Authorization"))
|
|
if err != nil {
|
|
apiresponse.Error(w, http.StatusUnauthorized, "登录状态无效")
|
|
return identity.Account{}, false
|
|
}
|
|
return account, true
|
|
}
|
|
|
|
func (h *AdminHTTPHandler) requirePermission(w http.ResponseWriter, r *http.Request, permission string) (identity.Account, bool) {
|
|
account, ok := h.account(w, r)
|
|
if !ok {
|
|
return account, false
|
|
}
|
|
if !identity.HasPermission(account, permission) {
|
|
apiresponse.Error(w, http.StatusForbidden, "缺少权限")
|
|
return identity.Account{}, false
|
|
}
|
|
return account, true
|
|
}
|
|
|
|
func (h *AdminHTTPHandler) systemInfo(w http.ResponseWriter, r *http.Request) {
|
|
if _, ok := h.requirePermission(w, r, identity.PermissionSystemManage); !ok {
|
|
return
|
|
}
|
|
var providers, models, keys int64
|
|
if err := h.pool.QueryRow(r.Context(), `SELECT (SELECT count(*) FROM gateway.providers WHERE enabled),(SELECT count(*) FROM gateway.provider_models WHERE enabled),(SELECT count(*) FROM gateway.api_keys WHERE enabled)`).Scan(&providers, &models, &keys); err != nil {
|
|
apiresponse.Error(w, http.StatusServiceUnavailable, "系统信息查询失败")
|
|
return
|
|
}
|
|
apiresponse.OK(w, map[string]any{"version": h.version, "go_version": runtime.Version(), "uptime_seconds": int64(time.Since(h.startedAt).Seconds()), "database": "postgresql", "object_storage": true, "clickhouse": false, "enabled_providers": providers, "enabled_models": models, "enabled_api_keys": keys})
|
|
}
|
|
|
|
func (h *AdminHTTPHandler) overview(w http.ResponseWriter, r *http.Request) {
|
|
if _, ok := h.requirePermission(w, r, identity.PermissionUsageRead); !ok {
|
|
return
|
|
}
|
|
var requests, failures, promptTokens, completionTokens, cost int64
|
|
var latency *float64
|
|
err := h.pool.QueryRow(r.Context(), `SELECT count(*),count(*) FILTER(WHERE coalesce(status_code,500)>=400),coalesce(sum(prompt_tokens),0),coalesce(sum(completion_tokens),0),coalesce(sum(cost_microunits),0),avg(latency_ms) FROM gateway.audit_events WHERE recorded_at>=clock_timestamp()-interval '24 hours'`).Scan(&requests, &failures, &promptTokens, &completionTokens, &cost, &latency)
|
|
if err != nil {
|
|
apiresponse.Error(w, http.StatusServiceUnavailable, "监控汇总查询失败")
|
|
return
|
|
}
|
|
apiresponse.OK(w, map[string]any{"window": "24h", "requests": requests, "failed_requests": failures, "prompt_tokens": promptTokens, "completion_tokens": completionTokens, "cost_microunits": cost, "avg_latency_ms": latency})
|
|
}
|
|
|
|
func (h *AdminHTTPHandler) reloadSnapshots(w http.ResponseWriter, r *http.Request) {
|
|
account, ok := h.account(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
if !identity.HasPermission(account, identity.PermissionProviderManage) {
|
|
apiresponse.Error(w, http.StatusForbidden, "缺少运行时配置管理权限")
|
|
return
|
|
}
|
|
if h.reload != nil {
|
|
if err := h.reload(r.Context()); err != nil {
|
|
apiresponse.Error(w, http.StatusServiceUnavailable, "运行时快照刷新不完整，请查看服务端日志")
|
|
return
|
|
}
|
|
}
|
|
apiresponse.OK(w, map[string]bool{"reloaded": true})
|
|
}
|
|
|
|
// tenantsOverview 以部门为租户维度,汇总各租户的账号/Key/用量与配额。
|
|
func (h *AdminHTTPHandler) tenantsOverview(w http.ResponseWriter, r *http.Request) {
|
|
if _, ok := h.requirePermission(w, r, identity.PermissionUsageRead); !ok {
|
|
return
|
|
}
|
|
rows, err := h.pool.Query(r.Context(), `SELECT d.id::text,d.name,d.max_api_keys,d.max_monthly_tokens,
|
|
(SELECT count(*) FROM gateway.portal_users u WHERE u.department_id=d.id),
|
|
(SELECT count(*) FROM gateway.api_keys k WHERE k.tenant_id=d.id AND k.enabled),
|
|
(SELECT count(*) FROM gateway.audit_events a WHERE a.tenant_id=d.id AND a.recorded_at>=date_trunc('day',now())),
|
|
(SELECT COALESCE(sum(a.prompt_tokens+a.completion_tokens),0) FROM gateway.audit_events a WHERE a.tenant_id=d.id AND a.recorded_at>=date_trunc('day',now())),
|
|
(SELECT COALESCE(sum(a.prompt_tokens+a.completion_tokens),0) FROM gateway.audit_events a WHERE a.tenant_id=d.id AND a.recorded_at>=date_trunc('month',now()))
|
|
FROM gateway.departments d WHERE d.active ORDER BY d.name`)
|
|
if err != nil {
|
|
apiresponse.Error(w, http.StatusServiceUnavailable, "租户概览查询失败")
|
|
return
|
|
}
|
|
defer rows.Close()
|
|
type tenantRow struct {
|
|
ID string `json:"id"`
|
|
Name string `json:"name"`
|
|
MaxAPIKeys int64 `json:"max_api_keys"`
|
|
MaxMonthlyTokens int64 `json:"max_monthly_tokens"`
|
|
PortalUsers int64 `json:"portal_users"`
|
|
EnabledKeys int64 `json:"enabled_api_keys"`
|
|
TodayRequests int64 `json:"today_requests"`
|
|
TodayTokens int64 `json:"today_tokens"`
|
|
MonthTokens int64 `json:"month_tokens"`
|
|
}
|
|
items := []tenantRow{}
|
|
for rows.Next() {
|
|
var item tenantRow
|
|
if err := rows.Scan(&item.ID, &item.Name, &item.MaxAPIKeys, &item.MaxMonthlyTokens, &item.PortalUsers, &item.EnabledKeys, &item.TodayRequests, &item.TodayTokens, &item.MonthTokens); err != nil {
|
|
apiresponse.Error(w, http.StatusServiceUnavailable, "租户概览查询失败")
|
|
return
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
apiresponse.Error(w, http.StatusServiceUnavailable, "租户概览查询失败")
|
|
return
|
|
}
|
|
apiresponse.OK(w, map[string]any{"tenants": items})
|
|
}
|
|
|
|
// reportRange 解析 from/to 日期(YYYY-MM-DD),返回起止时间。
|
|
func (h *AdminHTTPHandler) reportRange(r *http.Request) (from, to time.Time) {
|
|
now := time.Now().UTC()
|
|
to = now
|
|
from = now.AddDate(0, 0, -6)
|
|
if value := r.URL.Query().Get("from"); value != "" {
|
|
if parsed, err := time.Parse("2006-01-02", value); err == nil {
|
|
from = parsed
|
|
}
|
|
}
|
|
if value := r.URL.Query().Get("to"); value != "" {
|
|
if parsed, err := time.Parse("2006-01-02", value); err == nil {
|
|
to = parsed.AddDate(0, 0, 1)
|
|
}
|
|
}
|
|
return from, to
|
|
}
|
|
|
|
// reportTools 工具维度统计:调用数/成功率/平均延迟。
|
|
func (h *AdminHTTPHandler) reportTools(w http.ResponseWriter, r *http.Request) {
|
|
if _, ok := h.requirePermission(w, r, identity.PermissionUsageRead); !ok {
|
|
return
|
|
}
|
|
from, to := h.reportRange(r)
|
|
rows, err := h.pool.Query(r.Context(), `SELECT t.code,t.name,
|
|
count(r.id),
|
|
count(r.id) FILTER (WHERE r.status='success'),
|
|
COALESCE(avg(r.latency_ms),0)::bigint
|
|
FROM gateway.tool_runs r JOIN gateway.tool_definitions t ON t.id=r.tool_id
|
|
WHERE r.created_at>=$1 AND r.created_at<$2
|
|
GROUP BY t.code,t.name ORDER BY count(r.id) DESC LIMIT 100`, from, to)
|
|
if err != nil {
|
|
apiresponse.Error(w, http.StatusServiceUnavailable, "工具报表查询失败")
|
|
return
|
|
}
|
|
defer rows.Close()
|
|
items := []map[string]any{}
|
|
for rows.Next() {
|
|
var code, name string
|
|
var total, success, avgLatency int64
|
|
if err := rows.Scan(&code, &name, &total, &success, &avgLatency); err != nil {
|
|
apiresponse.Error(w, http.StatusServiceUnavailable, "工具报表查询失败")
|
|
return
|
|
}
|
|
items = append(items, map[string]any{"code": code, "name": name, "requests": total, "success": success, "failed": total - success, "avg_latency_ms": avgLatency})
|
|
}
|
|
apiresponse.OK(w, items)
|
|
}
|
|
|
|
// reportApprovals 审批维度统计:模型/资源/工具申请的发起与审批结果。
|
|
func (h *AdminHTTPHandler) reportApprovals(w http.ResponseWriter, r *http.Request) {
|
|
if _, ok := h.requirePermission(w, r, identity.PermissionUsageRead); !ok {
|
|
return
|
|
}
|
|
from, to := h.reportRange(r)
|
|
rows, err := h.pool.Query(r.Context(), `SELECT 'model' AS kind,status,count(*) FROM gateway.model_access_requests WHERE created_at>=$1 AND created_at<$2 GROUP BY status
|
|
UNION ALL SELECT 'resource',status,count(*) FROM gateway.resource_access_requests WHERE created_at>=$1 AND created_at<$2 GROUP BY status
|
|
UNION ALL SELECT 'tool',status,count(*) FROM gateway.tool_approval_requests WHERE created_at>=$1 AND created_at<$2 GROUP BY status`, from, to)
|
|
if err != nil {
|
|
apiresponse.Error(w, http.StatusServiceUnavailable, "审批报表查询失败")
|
|
return
|
|
}
|
|
defer rows.Close()
|
|
items := []map[string]any{}
|
|
for rows.Next() {
|
|
var kind, status string
|
|
var count int64
|
|
if err := rows.Scan(&kind, &status, &count); err != nil {
|
|
apiresponse.Error(w, http.StatusServiceUnavailable, "审批报表查询失败")
|
|
return
|
|
}
|
|
items = append(items, map[string]any{"kind": kind, "status": status, "count": count})
|
|
}
|
|
apiresponse.OK(w, items)
|
|
}
|
|
|
|
// reportSecurity 安全维度统计:登录成功/失败、锁定与来源 IP 分布。
|
|
func (h *AdminHTTPHandler) reportSecurity(w http.ResponseWriter, r *http.Request) {
|
|
if _, ok := h.requirePermission(w, r, identity.PermissionAuditRead); !ok {
|
|
return
|
|
}
|
|
from, to := h.reportRange(r)
|
|
rows, err := h.pool.Query(r.Context(), `SELECT success,count(*),count(DISTINCT ip) FROM gateway.login_logs WHERE created_at>=$1 AND created_at<$2 GROUP BY success
|
|
UNION ALL SELECT NULL,0,0 WHERE false`, from, to)
|
|
if err != nil {
|
|
apiresponse.Error(w, http.StatusServiceUnavailable, "安全报表查询失败")
|
|
return
|
|
}
|
|
defer rows.Close()
|
|
type loginStat struct {
|
|
Success bool `json:"success"`
|
|
Count int64 `json:"count"`
|
|
DistinctIPs int64 `json:"distinct_ips"`
|
|
}
|
|
stats := []loginStat{}
|
|
for rows.Next() {
|
|
var item loginStat
|
|
if err := rows.Scan(&item.Success, &item.Count, &item.DistinctIPs); err != nil {
|
|
apiresponse.Error(w, http.StatusServiceUnavailable, "安全报表查询失败")
|
|
return
|
|
}
|
|
stats = append(stats, item)
|
|
}
|
|
rows.Close()
|
|
ips, err := h.pool.Query(r.Context(), `SELECT ip::text,count(*) FROM gateway.login_logs WHERE created_at>=$1 AND created_at<$2 GROUP BY ip ORDER BY count(*) DESC LIMIT 10`, from, to)
|
|
if err != nil {
|
|
apiresponse.Error(w, http.StatusServiceUnavailable, "安全报表查询失败")
|
|
return
|
|
}
|
|
defer ips.Close()
|
|
topIPs := []map[string]any{}
|
|
for ips.Next() {
|
|
var ip string
|
|
var count int64
|
|
if err := ips.Scan(&ip, &count); err != nil {
|
|
apiresponse.Error(w, http.StatusServiceUnavailable, "安全报表查询失败")
|
|
return
|
|
}
|
|
topIPs = append(topIPs, map[string]any{"ip": ip, "count": count})
|
|
}
|
|
apiresponse.OK(w, map[string]any{"login_stats": stats, "top_ips": topIPs})
|
|
}
|