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 干净。
189 lines
8.0 KiB
Go
189 lines
8.0 KiB
Go
package apikey
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
platformid "aigateway.local/core/internal/platform/id"
|
|
"aigateway.local/core/internal/platform/oplog"
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
type Repository struct{ pool *pgxpool.Pool }
|
|
|
|
func NewRepository(pool *pgxpool.Pool) *Repository { return &Repository{pool: pool} }
|
|
|
|
func (r *Repository) List(ctx context.Context) ([]Record, error) {
|
|
if r.pool == nil {
|
|
return nil, ErrStore
|
|
}
|
|
rows, err := r.pool.Query(ctx, `
|
|
SELECT id::text, tenant_id::text, name, key_prefix, scopes, enabled,
|
|
requests_per_minute, monthly_request_quota, monthly_token_quota, expires_at, last_used_at, created_at
|
|
FROM gateway.api_keys ORDER BY created_at DESC LIMIT 200`)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("%w: %v", ErrStore, err)
|
|
}
|
|
defer rows.Close()
|
|
var records []Record
|
|
for rows.Next() {
|
|
var record Record
|
|
if err := rows.Scan(&record.ID, &record.TenantID, &record.Name, &record.KeyPrefix, &record.Scopes, &record.Enabled, &record.RequestsPerMinute, &record.MonthlyRequestQuota, &record.MonthlyTokenQuota, &record.ExpiresAt, &record.LastUsedAt, &record.CreatedAt); err != nil {
|
|
return nil, fmt.Errorf("%w: %v", ErrStore, err)
|
|
}
|
|
records = append(records, record)
|
|
}
|
|
return records, rows.Err()
|
|
}
|
|
|
|
func (r *Repository) Create(ctx context.Context, name string, scopes []string, requestsPerMinute int, monthlyRequestQuota, monthlyTokenQuota int64, expiresAt *time.Time, actorID string) (Record, string, error) {
|
|
if r.pool == nil {
|
|
return Record{}, "", ErrStore
|
|
}
|
|
secret, prefix, hash, err := Generate()
|
|
if err != nil {
|
|
return Record{}, "", err
|
|
}
|
|
id, err := platformid.NewUUID()
|
|
if err != nil {
|
|
return Record{}, "", err
|
|
}
|
|
eventID, err := platformid.NewUUID()
|
|
if err != nil {
|
|
return Record{}, "", err
|
|
}
|
|
tx, err := r.pool.Begin(ctx)
|
|
if err != nil {
|
|
return Record{}, "", fmt.Errorf("%w: %v", ErrStore, err)
|
|
}
|
|
defer func() { _ = tx.Rollback(ctx) }()
|
|
record := Record{ID: id, Name: name, KeyPrefix: prefix, KeyHash: hash, Scopes: scopes, Enabled: true, RequestsPerMinute: requestsPerMinute, MonthlyRequestQuota: monthlyRequestQuota, MonthlyTokenQuota: monthlyTokenQuota, ExpiresAt: expiresAt}
|
|
err = tx.QueryRow(ctx, `
|
|
INSERT INTO gateway.api_keys (id, name, key_prefix, key_hash, scopes, requests_per_minute, monthly_request_quota, monthly_token_quota, expires_at, created_by)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, nullif($10,'')::uuid)
|
|
RETURNING created_at`, id, name, prefix, hash, scopes, requestsPerMinute, monthlyRequestQuota, monthlyTokenQuota, expiresAt, actorID).Scan(&record.CreatedAt)
|
|
if err != nil {
|
|
return Record{}, "", fmt.Errorf("%w: %v", ErrStore, err)
|
|
}
|
|
payload, _ := json.Marshal(map[string]any{"api_key_id": id, "name": name, "key_prefix": prefix, "requests_per_minute": requestsPerMinute, "monthly_request_quota": monthlyRequestQuota, "monthly_token_quota": monthlyTokenQuota})
|
|
if _, err := tx.Exec(ctx, `
|
|
INSERT INTO gateway.outbox_events (event_id, event_type, event_version, aggregate_type, aggregate_id, payload)
|
|
VALUES ($1, 'api_key.created', 1, 'api_key', $2, $3)`, eventID, id, payload); err != nil {
|
|
return Record{}, "", fmt.Errorf("%w: %v", ErrStore, err)
|
|
}
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return Record{}, "", fmt.Errorf("%w: %v", ErrStore, err)
|
|
}
|
|
oplog.Record(ctx, r.pool, nil, actorID, "", "api_key.create", "api_key", id, map[string]any{
|
|
"name": name, "key_prefix": prefix, "requests_per_minute": requestsPerMinute,
|
|
"monthly_request_quota": monthlyRequestQuota, "monthly_token_quota": monthlyTokenQuota,
|
|
})
|
|
return record, secret, nil
|
|
}
|
|
|
|
func (r *Repository) Validate(ctx context.Context, hash []byte) (Principal, error) {
|
|
if r.pool == nil {
|
|
return Principal{}, ErrStore
|
|
}
|
|
var principal Principal
|
|
err := r.pool.QueryRow(ctx, `
|
|
UPDATE gateway.api_keys
|
|
SET last_used_at = CASE WHEN last_used_at IS NULL OR last_used_at < clock_timestamp() - interval '5 minutes' THEN clock_timestamp() ELSE last_used_at END
|
|
WHERE key_hash = $1 AND enabled AND (expires_at IS NULL OR expires_at > clock_timestamp())
|
|
RETURNING id::text, tenant_id::text, scopes, requests_per_minute, monthly_request_quota, monthly_token_quota`, hash).Scan(&principal.APIKeyID, &principal.TenantID, &principal.Scopes, &principal.RequestsPerMinute, &principal.MonthlyRequestQuota, &principal.MonthlyTokenQuota)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return Principal{}, ErrInvalid
|
|
}
|
|
if err != nil {
|
|
return Principal{}, fmt.Errorf("%w: %v", ErrStore, err)
|
|
}
|
|
return principal, nil
|
|
}
|
|
|
|
func (r *Repository) Revoke(ctx context.Context, id, actorID string) ([]byte, error) {
|
|
if r.pool == nil {
|
|
return nil, ErrStore
|
|
}
|
|
eventID, err := platformid.NewUUID()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
tx, err := r.pool.Begin(ctx)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("%w: %v", ErrStore, err)
|
|
}
|
|
defer func() { _ = tx.Rollback(ctx) }()
|
|
var hash []byte
|
|
err = tx.QueryRow(ctx, `
|
|
UPDATE gateway.api_keys SET enabled = false, updated_at = clock_timestamp()
|
|
WHERE id = $1 AND enabled RETURNING key_hash`, id).Scan(&hash)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, ErrInvalid
|
|
}
|
|
if err != nil {
|
|
return nil, fmt.Errorf("%w: %v", ErrStore, err)
|
|
}
|
|
payload, _ := json.Marshal(map[string]any{"api_key_id": id, "actor_id": actorID})
|
|
if _, err := tx.Exec(ctx, `
|
|
INSERT INTO gateway.outbox_events (event_id, event_type, event_version, aggregate_type, aggregate_id, payload)
|
|
VALUES ($1, 'api_key.revoked', 1, 'api_key', $2, $3)`, eventID, id, payload); err != nil {
|
|
return nil, fmt.Errorf("%w: %v", ErrStore, err)
|
|
}
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return nil, fmt.Errorf("%w: %v", ErrStore, err)
|
|
}
|
|
oplog.Record(ctx, r.pool, nil, actorID, "", "api_key.revoke", "api_key", id, nil)
|
|
return hash, nil
|
|
}
|
|
|
|
func (r *Repository) UpdateLimits(ctx context.Context, id string, requestsPerMinute int, monthlyRequestQuota, monthlyTokenQuota int64, actorID string) (Record, []byte, error) {
|
|
if r.pool == nil {
|
|
return Record{}, nil, ErrStore
|
|
}
|
|
eventID, err := platformid.NewUUID()
|
|
if err != nil {
|
|
return Record{}, nil, err
|
|
}
|
|
tx, err := r.pool.Begin(ctx)
|
|
if err != nil {
|
|
return Record{}, nil, fmt.Errorf("%w: %v", ErrStore, err)
|
|
}
|
|
defer func() { _ = tx.Rollback(ctx) }()
|
|
var record Record
|
|
err = tx.QueryRow(ctx, `
|
|
UPDATE gateway.api_keys
|
|
SET requests_per_minute = $2, monthly_request_quota = $3, monthly_token_quota = $4, updated_at = clock_timestamp()
|
|
WHERE id = $1 AND enabled
|
|
RETURNING id::text, tenant_id::text, name, key_prefix, key_hash, scopes, enabled,
|
|
requests_per_minute, monthly_request_quota, monthly_token_quota, expires_at, last_used_at, created_at`,
|
|
id, requestsPerMinute, monthlyRequestQuota, monthlyTokenQuota,
|
|
).Scan(&record.ID, &record.TenantID, &record.Name, &record.KeyPrefix, &record.KeyHash, &record.Scopes, &record.Enabled,
|
|
&record.RequestsPerMinute, &record.MonthlyRequestQuota, &record.MonthlyTokenQuota, &record.ExpiresAt, &record.LastUsedAt, &record.CreatedAt)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return Record{}, nil, ErrInvalid
|
|
}
|
|
if err != nil {
|
|
return Record{}, nil, fmt.Errorf("%w: %v", ErrStore, err)
|
|
}
|
|
payload, _ := json.Marshal(map[string]any{
|
|
"api_key_id": id, "actor_id": actorID, "requests_per_minute": requestsPerMinute,
|
|
"monthly_request_quota": monthlyRequestQuota, "monthly_token_quota": monthlyTokenQuota,
|
|
})
|
|
if _, err := tx.Exec(ctx, `
|
|
INSERT INTO gateway.outbox_events (event_id, event_type, event_version, aggregate_type, aggregate_id, payload)
|
|
VALUES ($1, 'api_key.limits_updated', 1, 'api_key', $2, $3)`, eventID, id, payload); err != nil {
|
|
return Record{}, nil, fmt.Errorf("%w: %v", ErrStore, err)
|
|
}
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return Record{}, nil, fmt.Errorf("%w: %v", ErrStore, err)
|
|
}
|
|
oplog.Record(ctx, r.pool, nil, actorID, "", "api_key.limits_update", "api_key", id, map[string]any{
|
|
"requests_per_minute": requestsPerMinute, "monthly_request_quota": monthlyRequestQuota, "monthly_token_quota": monthlyTokenQuota,
|
|
})
|
|
return record, record.KeyHash, nil
|
|
}
|