feat(audit):审计表单独存储到其他数据库中
This commit is contained in:
1 parent
879d94a957
commit
479340d4ea
12 files changed
+470
-107
No files matched your search
@@ -88,11 +88,11 @@ func (s *sAdmin) UpdateAuditConfig(ctx context.Context, req *v1.AuditConfigUpdat
|
||||
}
|
||||
|
||||
// queryPage 使用 GoFrame ORM 分页查询,返回 []map[string]any。
|
||||
// 注意:该函数使用 g.DB().Model(table) 直接操作动态表名而非 DAO,
|
||||
// 适用于通用分页工具场景(多表共用)。固定表查询应优先使用对应的 DAO。
|
||||
// 注意:该函数使用审计数据库连接(common.GetAuditDB())查询审计表。
|
||||
// 分开调用 Count 和 All 避免 ScanAndCount 对 map[string]any 的 bug。
|
||||
func queryPage(ctx context.Context, table, fields, where, orderBy string, page, pageSize int, args ...any) ([]map[string]any, int, error) {
|
||||
m := g.DB().Ctx(ctx).Model(table).Safe()
|
||||
auditDB := common.GetAuditDB()
|
||||
m := auditDB.Ctx(ctx).Model(table).Safe()
|
||||
if where != "" {
|
||||
m = m.Where(where, args...)
|
||||
}
|
||||
@@ -106,7 +106,7 @@ func queryPage(ctx context.Context, table, fields, where, orderBy string, page,
|
||||
}
|
||||
|
||||
// 重新构建查询(Count 会消耗 Model 状态)
|
||||
q := g.DB().Ctx(ctx).Model(table).Safe()
|
||||
q := auditDB.Ctx(ctx).Model(table).Safe()
|
||||
if fields != "" {
|
||||
q = q.Fields(fields)
|
||||
}
|
||||
@@ -228,7 +228,7 @@ func (s *sAdmin) ListSensitiveAccessLogs(ctx context.Context, req *v1.SensitiveL
|
||||
|
||||
// LogSensitiveAccess inserts a sensitive data access log entry.
|
||||
func LogSensitiveAccess(ctx context.Context, userID int64, userType, resourceType string, resourceID int64, action, reason, ip, userAgent string) error {
|
||||
_, err := dao.AudSensitiveAccessLogs.Ctx(ctx).Insert(do.AudSensitiveAccessLogs{
|
||||
_, err := common.AuditModelCtx(ctx, "aud_sensitive_access_logs").Data(do.AudSensitiveAccessLogs{
|
||||
UserId: userID,
|
||||
UserType: userType,
|
||||
ResourceType: resourceType,
|
||||
@@ -237,7 +237,7 @@ func LogSensitiveAccess(ctx context.Context, userID int64, userType, resourceTyp
|
||||
Reason: reason,
|
||||
IpAddress: ip,
|
||||
UserAgent: userAgent,
|
||||
})
|
||||
}).Insert()
|
||||
if err != nil {
|
||||
return gerror.Wrapf(err, "记录敏感数据访问日志失败")
|
||||
}
|
||||
@@ -250,6 +250,7 @@ func MaskSensitiveData(data string) string {
|
||||
}
|
||||
|
||||
// ListRequestAuditLogs 分页查询请求审计日志(不返回 request_body/response_body 以优化性能)
|
||||
// 审计数据从审计库查询,关联信息(用户名、租户名等)从主库批量查询后在应用层合并。
|
||||
func (s *sAdmin) ListRequestAuditLogs(ctx context.Context, req *v1.RequestAuditLogListReq) (*v1.RequestAuditLogListRes, error) {
|
||||
if err := common.ValidateDateParam(req.StartDate, "开始日期"); err != nil {
|
||||
return nil, err
|
||||
@@ -264,72 +265,94 @@ func (s *sAdmin) ListRequestAuditLogs(ctx context.Context, req *v1.RequestAuditL
|
||||
var args []any
|
||||
|
||||
if req.TenantID > 0 {
|
||||
conditions = append(conditions, "a.tenant_id = ?")
|
||||
conditions = append(conditions, "tenant_id = ?")
|
||||
args = append(args, int64(req.TenantID))
|
||||
}
|
||||
if req.ApiKeyID > 0 {
|
||||
conditions = append(conditions, "a.api_key_id = ?")
|
||||
conditions = append(conditions, "api_key_id = ?")
|
||||
args = append(args, int64(req.ApiKeyID))
|
||||
}
|
||||
// 用户名筛选:先从主库查匹配的 user_id 集合,转为 IN 条件
|
||||
if req.Username != "" {
|
||||
conditions = append(conditions, "t.username LIKE ?")
|
||||
args = append(args, "%"+req.Username+"%")
|
||||
type userRow struct {
|
||||
Id int64 `json:"id"`
|
||||
TenantId int64 `json:"tenant_id"`
|
||||
}
|
||||
var users []userRow
|
||||
err := g.DB().Ctx(ctx).Model("tnt_users").
|
||||
Where("username LIKE ?", "%"+req.Username+"%").
|
||||
Fields("id, tenant_id").
|
||||
Scan(&users)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(users) == 0 {
|
||||
return &v1.RequestAuditLogListRes{
|
||||
List: []map[string]any{}, Total: 0, Page: page, PageSize: pageSize,
|
||||
}, nil
|
||||
}
|
||||
// 构建 (tenant_id, user_id) IN 条件
|
||||
userConditions := make([]string, 0, len(users))
|
||||
for _, u := range users {
|
||||
userConditions = append(userConditions, fmt.Sprintf("(tenant_id = %d AND user_id = %d)", u.TenantId, u.Id))
|
||||
}
|
||||
conditions = append(conditions, "("+strings.Join(userConditions, " OR ")+")")
|
||||
}
|
||||
if req.RequestId != "" {
|
||||
conditions = append(conditions, "a.request_id = ?")
|
||||
conditions = append(conditions, "request_id = ?")
|
||||
args = append(args, req.RequestId)
|
||||
}
|
||||
if req.TaskId != "" {
|
||||
conditions = append(conditions, "a.task_id = ?")
|
||||
conditions = append(conditions, "task_id = ?")
|
||||
args = append(args, req.TaskId)
|
||||
}
|
||||
if req.Method != "" {
|
||||
conditions = append(conditions, "a.method = ?")
|
||||
conditions = append(conditions, "method = ?")
|
||||
args = append(args, req.Method)
|
||||
}
|
||||
if req.Path != "" {
|
||||
conditions = append(conditions, "a.path LIKE ?")
|
||||
conditions = append(conditions, "path LIKE ?")
|
||||
args = append(args, "%"+req.Path+"%")
|
||||
}
|
||||
if req.StatusCode > 0 {
|
||||
conditions = append(conditions, "a.status_code = ?")
|
||||
conditions = append(conditions, "status_code = ?")
|
||||
args = append(args, req.StatusCode)
|
||||
}
|
||||
if req.StartDate != "" {
|
||||
conditions = append(conditions, "a.created_at >= ?")
|
||||
conditions = append(conditions, "created_at >= ?")
|
||||
args = append(args, req.StartDate+" 00:00:00")
|
||||
}
|
||||
if req.EndDate != "" {
|
||||
conditions = append(conditions, "a.created_at <= ?")
|
||||
conditions = append(conditions, "created_at <= ?")
|
||||
args = append(args, req.EndDate+" 23:59:59")
|
||||
}
|
||||
|
||||
where := strings.Join(conditions, " AND ")
|
||||
fromClause := "aud_request_logs a LEFT JOIN tnt_users t ON a.user_id = t.id AND a.tenant_id = t.tenant_id LEFT JOIN tnt_projects p ON a.project_id = p.id LEFT JOIN tnt_tenants tn ON a.tenant_id = tn.id LEFT JOIN api_keys ak ON a.api_key_id = ak.id"
|
||||
whereClause := ""
|
||||
if where != "" {
|
||||
whereClause = " WHERE " + where
|
||||
}
|
||||
|
||||
countSQL := "SELECT COUNT(*) AS total FROM " + fromClause + whereClause
|
||||
countResult, err := g.DB().Ctx(ctx).Query(ctx, countSQL, args...)
|
||||
// Step 1: 从审计库查询审计记录(不带 JOIN)
|
||||
auditDB := common.GetAuditDB()
|
||||
countM := auditDB.Ctx(ctx).Model("aud_request_logs").Safe()
|
||||
if where != "" {
|
||||
countM = countM.Where(where, args...)
|
||||
}
|
||||
total, err := countM.Count()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
total := 0
|
||||
if len(countResult) > 0 {
|
||||
total = countResult[0]["total"].Int()
|
||||
if total == 0 {
|
||||
return &v1.RequestAuditLogListRes{
|
||||
List: []map[string]any{}, Total: 0, Page: page, PageSize: pageSize,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// NOTE: fromClause and whereClause are constructed from hardcoded strings
|
||||
// and parameterized conditions (using ? placeholders). No user input is
|
||||
// directly interpolated into the SQL template.
|
||||
dataSQL := fmt.Sprintf(
|
||||
`SELECT a.id, a.tenant_id, COALESCE(tn.name, '') AS tenant_name, a.user_id, COALESCE(t.username, '') AS username, a.project_id, COALESCE(p.name, '') AS project_name, a.api_key_id, COALESCE(ak.name, '') AS api_key_name, a.request_id, a.method, a.path, a.query_params, a.status_code, a.client_ip, a.user_agent, a.latency_ms, a.first_token_ms, a.created_at, a.updated_at, a.audit_level
|
||||
FROM %s%s ORDER BY a.created_at DESC LIMIT ? OFFSET ?`,
|
||||
fromClause, whereClause,
|
||||
)
|
||||
result, err := g.DB().Ctx(ctx).Query(ctx, dataSQL, append(args, pageSize, (page-1)*pageSize)...)
|
||||
dataM := auditDB.Ctx(ctx).Model("aud_request_logs").Safe().
|
||||
Fields("id, tenant_id, user_id, api_key_id, project_id, request_id, method, path, query_params, status_code, client_ip, user_agent, latency_ms, first_token_ms, audit_level, created_at").
|
||||
OrderDesc("created_at").
|
||||
Page(page, pageSize)
|
||||
if where != "" {
|
||||
dataM = dataM.Where(where, args...)
|
||||
}
|
||||
result, err := dataM.All()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -343,6 +366,9 @@ func (s *sAdmin) ListRequestAuditLogs(ctx context.Context, req *v1.RequestAuditL
|
||||
items = append(items, m)
|
||||
}
|
||||
|
||||
// Step 2: 从主库批量查询关联信息并合并
|
||||
common.EnrichAuditRecords(ctx, items)
|
||||
|
||||
return &v1.RequestAuditLogListRes{
|
||||
List: items,
|
||||
Total: total,
|
||||
@@ -354,7 +380,7 @@ func (s *sAdmin) ListRequestAuditLogs(ctx context.Context, req *v1.RequestAuditL
|
||||
// GetRequestAuditLogDetail 查询单条请求审计日志详情(含完整 request_body 和 response_body)
|
||||
func (s *sAdmin) GetRequestAuditLogDetail(ctx context.Context, req *v1.RequestAuditLogDetailReq) (*v1.RequestAuditLogDetailRes, error) {
|
||||
var record *entity.AudRequestLogs
|
||||
err := dao.AudRequestLogs.Ctx(ctx).
|
||||
err := common.AuditModelCtx(ctx, "aud_request_logs").
|
||||
Where("id", req.Id).
|
||||
Scan(&record)
|
||||
if err != nil {
|
||||
@@ -430,7 +456,7 @@ func (s *sAdmin) ExportOperationLogs(ctx context.Context, req *v1.OperationLogEx
|
||||
offset := 0
|
||||
for {
|
||||
where, args := buildOpLogWhere()
|
||||
q := g.DB().Ctx(ctx).Model("aud_operation_logs").Fields(selectFields).OrderDesc("created_at").Offset(offset).Limit(1000)
|
||||
q := common.GetAuditDB().Ctx(ctx).Model("aud_operation_logs").Fields(selectFields).OrderDesc("created_at").Offset(offset).Limit(1000)
|
||||
if where != "" {
|
||||
q = q.Where(where, args...)
|
||||
}
|
||||
@@ -477,6 +503,7 @@ func (s *sAdmin) ExportOperationLogs(ctx context.Context, req *v1.OperationLogEx
|
||||
}
|
||||
|
||||
// ContentFilterLogList returns a paginated list of content filter interception logs.
|
||||
// 审计数据从审计库查询,关联信息从主库批量查询后在应用层合并。
|
||||
func (s *sAdmin) ContentFilterLogList(ctx context.Context, req *v1.ContentFilterLogListReq) (*v1.ContentFilterLogListRes, error) {
|
||||
if err := common.ValidateDateParam(req.StartDate, "开始日期"); err != nil {
|
||||
return nil, err
|
||||
@@ -491,61 +518,60 @@ func (s *sAdmin) ContentFilterLogList(ctx context.Context, req *v1.ContentFilter
|
||||
var args []any
|
||||
|
||||
if req.TenantID > 0 {
|
||||
conditions = append(conditions, "l.tenant_id = ?")
|
||||
conditions = append(conditions, "tenant_id = ?")
|
||||
args = append(args, int64(req.TenantID))
|
||||
}
|
||||
if req.Mode != "" {
|
||||
conditions = append(conditions, "l.filter_mode = ?")
|
||||
conditions = append(conditions, "filter_mode = ?")
|
||||
args = append(args, req.Mode)
|
||||
}
|
||||
if req.Blocked == "true" {
|
||||
conditions = append(conditions, "l.blocked = true")
|
||||
conditions = append(conditions, "blocked = true")
|
||||
} else if req.Blocked == "false" {
|
||||
conditions = append(conditions, "l.blocked = false")
|
||||
conditions = append(conditions, "blocked = false")
|
||||
}
|
||||
if req.StartDate != "" {
|
||||
conditions = append(conditions, "l.created_at >= ?")
|
||||
conditions = append(conditions, "created_at >= ?")
|
||||
args = append(args, req.StartDate+" 00:00:00")
|
||||
}
|
||||
if req.EndDate != "" {
|
||||
conditions = append(conditions, "l.created_at <= ?")
|
||||
conditions = append(conditions, "created_at <= ?")
|
||||
args = append(args, req.EndDate+" 23:59:59")
|
||||
}
|
||||
if req.Keyword != "" {
|
||||
conditions = append(conditions, "(l.matched_words::text ILIKE ? OR l.path ILIKE ?)")
|
||||
conditions = append(conditions, "(matched_words::text ILIKE ? OR path ILIKE ?)")
|
||||
kw := "%" + req.Keyword + "%"
|
||||
args = append(args, kw, kw)
|
||||
}
|
||||
|
||||
where := ""
|
||||
if len(conditions) > 0 {
|
||||
where = " WHERE " + strings.Join(conditions, " AND ")
|
||||
where := strings.Join(conditions, " AND ")
|
||||
|
||||
// Step 1: 从审计库查询审计记录(不带 JOIN)
|
||||
auditDB := common.GetAuditDB()
|
||||
fields := "id, tenant_id, user_id, api_key_id, project_id, request_id, method, path, client_ip, filter_mode, matched_words, original_snippet, blocked, created_at"
|
||||
|
||||
countM := auditDB.Ctx(ctx).Model("aud_content_filter_logs").Safe()
|
||||
if where != "" {
|
||||
countM = countM.Where(where, args...)
|
||||
}
|
||||
|
||||
fields := `l.id, l.tenant_id, l.user_id, l.api_key_id, l.project_id, l.request_id, l.method, l.path, l.client_ip,
|
||||
l.filter_mode, l.matched_words, l.original_snippet, l.blocked, l.created_at,
|
||||
t.name AS tenant_name, u.username AS user_name, k.name AS api_key_name, p.name AS project_name`
|
||||
|
||||
// Count
|
||||
countSQL := "SELECT COUNT(*) AS total FROM aud_content_filter_logs l" + where
|
||||
countResult, err := g.DB().Ctx(ctx).Query(ctx, countSQL, args...)
|
||||
total, err := countM.Count()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
total := 0
|
||||
if len(countResult) > 0 {
|
||||
total = countResult[0]["total"].Int()
|
||||
if total == 0 {
|
||||
return &v1.ContentFilterLogListRes{
|
||||
List: []map[string]any{}, Total: 0, Page: page, PageSize: pageSize,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Query with JOINs
|
||||
dataSQL := fmt.Sprintf(`SELECT %s FROM aud_content_filter_logs l
|
||||
LEFT JOIN tnt_tenants t ON l.tenant_id = t.id
|
||||
LEFT JOIN tnt_users u ON l.user_id = u.id
|
||||
LEFT JOIN api_keys k ON l.api_key_id = k.id
|
||||
LEFT JOIN tnt_projects p ON l.project_id = p.id
|
||||
%s ORDER BY l.created_at DESC LIMIT ? OFFSET ?`, fields, where)
|
||||
|
||||
result, err := g.DB().Ctx(ctx).Query(ctx, dataSQL, append(args, pageSize, (page-1)*pageSize)...)
|
||||
dataM := auditDB.Ctx(ctx).Model("aud_content_filter_logs").Safe().
|
||||
Fields(fields).
|
||||
OrderDesc("created_at").
|
||||
Page(page, pageSize)
|
||||
if where != "" {
|
||||
dataM = dataM.Where(where, args...)
|
||||
}
|
||||
result, err := dataM.All()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -559,6 +585,9 @@ func (s *sAdmin) ContentFilterLogList(ctx context.Context, req *v1.ContentFilter
|
||||
items = append(items, m)
|
||||
}
|
||||
|
||||
// Step 2: 从主库批量查询关联信息并合并
|
||||
common.EnrichAuditRecords(ctx, items)
|
||||
|
||||
return &v1.ContentFilterLogListRes{
|
||||
List: items,
|
||||
Total: total,
|
||||
|
||||
@@ -210,7 +210,7 @@ func cleanupTableByDate(ctx context.Context, table, dateColumn string, days int)
|
||||
}
|
||||
|
||||
for {
|
||||
result, err := g.DB().Ctx(ctx).Exec(ctx, deleteSQL, cutoff, batchSize)
|
||||
result, err := common.GetAuditDB().Ctx(ctx).Exec(ctx, deleteSQL, cutoff, batchSize)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -207,7 +207,7 @@ func (s *sAdmin) LoginHistory(ctx context.Context, req *v1.AdminLoginHistoryReq)
|
||||
pageSize = 20
|
||||
}
|
||||
|
||||
q := dao.AudLoginHistory.Ctx(ctx).Where("user_type", "admin")
|
||||
q := common.AuditModelCtx(ctx, "aud_login_history").Where("user_type", "admin")
|
||||
|
||||
if req.Username != "" {
|
||||
userIds, err := adminUserIdsByUsername(ctx, req.Username)
|
||||
@@ -327,7 +327,7 @@ func buildAdminUserMap(ctx context.Context, records []entity.AudLoginHistory) ma
|
||||
func (s *sAdmin) TenantLoginHistory(ctx context.Context, req *v1.AdminTenantLoginHistoryReq) (*v1.AdminTenantLoginHistoryRes, error) {
|
||||
page, pageSize := common.NormalizePagination(req.Page, req.PageSize)
|
||||
|
||||
q := dao.AudLoginHistory.Ctx(ctx).Where("user_type", "tenant")
|
||||
q := common.AuditModelCtx(ctx, "aud_login_history").Where("user_type", "tenant")
|
||||
|
||||
if req.TenantID > 0 {
|
||||
q = q.Where("tenant_id", req.TenantID)
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
package common
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
|
||||
"github.com/gogf/gf/v2/database/gdb"
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
)
|
||||
|
||||
const auditDBGroup = "audit"
|
||||
|
||||
var (
|
||||
auditDBConfigured bool
|
||||
auditDBConfiguredOnce sync.Once
|
||||
)
|
||||
|
||||
// GetAuditDB 返回审计数据库连接。
|
||||
// 如果配置了 database.audit 分组则使用独立审计库,否则回退到主库。
|
||||
func GetAuditDB() gdb.DB {
|
||||
if !IsAuditDBConfigured() {
|
||||
return g.DB()
|
||||
}
|
||||
return g.DB(auditDBGroup)
|
||||
}
|
||||
|
||||
// IsAuditDBConfigured 检查是否配置了独立的审计数据库。
|
||||
// 通过检查配置文件中是否存在 database.audit.link 来判断。
|
||||
func IsAuditDBConfigured() bool {
|
||||
auditDBConfiguredOnce.Do(func() {
|
||||
cfg := g.Cfg()
|
||||
if cfg == nil {
|
||||
return
|
||||
}
|
||||
auditLink := cfg.MustGet(context.Background(), "database.audit.link")
|
||||
auditDBConfigured = !auditLink.IsEmpty()
|
||||
})
|
||||
return auditDBConfigured
|
||||
}
|
||||
|
||||
// AuditModelCtx 返回指定审计表的 Model(带 context,绑定到审计库连接)。
|
||||
// 使用示例:
|
||||
//
|
||||
// common.AuditModelCtx(ctx, "aud_request_logs").Data(data).Insert()
|
||||
// common.AuditModelCtx(ctx, "aud_operation_logs").Where("id", id).Scan(&record)
|
||||
func AuditModelCtx(ctx context.Context, table string) *gdb.Model {
|
||||
return GetAuditDB().Model(table).Safe().Ctx(ctx)
|
||||
}
|
||||
@@ -0,0 +1,254 @@
|
||||
package common
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
)
|
||||
|
||||
// EnrichAuditRecords 为审计记录批量补充关联信息(用户名、租户名、API Key 名、项目名)。
|
||||
// records 中的每条记录应包含 tenant_id、user_id、api_key_id、project_id 字段,
|
||||
// 函数会从主库批量查询关联信息并回填 tenant_name、username、api_key_name、project_name 字段。
|
||||
func EnrichAuditRecords(ctx context.Context, records []map[string]any) {
|
||||
if len(records) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
// 收集所有需要查询的 ID
|
||||
tenantIDs := collectInt64Fields(records, "tenant_id")
|
||||
userKeys := collectUserKeys(records)
|
||||
apiKeyIDs := collectInt64Fields(records, "api_key_id")
|
||||
projectIDs := collectInt64Fields(records, "project_id")
|
||||
|
||||
// 批量查询关联信息(从主库)
|
||||
tenantMap := BatchQueryTenantNames(ctx, tenantIDs)
|
||||
userMap := BatchQueryUserNames(ctx, userKeys)
|
||||
apiKeyMap := BatchQueryApiKeyNames(ctx, apiKeyIDs)
|
||||
projectMap := BatchQueryProjectNames(ctx, projectIDs)
|
||||
|
||||
// 回填关联信息
|
||||
for _, record := range records {
|
||||
if tenantID, ok := getInt64Field(record, "tenant_id"); ok && tenantID > 0 {
|
||||
record["tenant_name"] = tenantMap[tenantID]
|
||||
}
|
||||
if userKey, ok := getUserKey(record); ok {
|
||||
record["username"] = userMap[userKey]
|
||||
}
|
||||
if apiKeyID, ok := getInt64Field(record, "api_key_id"); ok && apiKeyID > 0 {
|
||||
record["api_key_name"] = apiKeyMap[apiKeyID]
|
||||
}
|
||||
if projectID, ok := getInt64Field(record, "project_id"); ok && projectID > 0 {
|
||||
record["project_name"] = projectMap[projectID]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// BatchQueryUserNames 批量查询用户名。
|
||||
// key 格式为 "tenantID:userID",返回 map[key]username。
|
||||
func BatchQueryUserNames(ctx context.Context, userKeys []string) map[string]string {
|
||||
if len(userKeys) == 0 {
|
||||
return nil
|
||||
}
|
||||
result := make(map[string]string, len(userKeys))
|
||||
// 去重
|
||||
seen := make(map[string]bool, len(userKeys))
|
||||
unique := make([]string, 0, len(userKeys))
|
||||
for _, k := range userKeys {
|
||||
if !seen[k] {
|
||||
seen[k] = true
|
||||
unique = append(unique, k)
|
||||
}
|
||||
}
|
||||
|
||||
// 拆分 tenantID 和 userID,按 tenantID 分组批量查询
|
||||
tenantUsers := make(map[int64][]int64) // tenantID -> []userID
|
||||
userKeyToID := make(map[string]int64) // "tenantID:userID" -> userID
|
||||
for _, key := range unique {
|
||||
var tenantID, userID int64
|
||||
fmt.Sscanf(key, "%d:%d", &tenantID, &userID)
|
||||
if tenantID > 0 && userID > 0 {
|
||||
tenantUsers[tenantID] = append(tenantUsers[tenantID], userID)
|
||||
userKeyToID[key] = userID
|
||||
}
|
||||
}
|
||||
|
||||
// 按 tenantID 分组查询
|
||||
for tenantID, userIDs := range tenantUsers {
|
||||
type row struct {
|
||||
Id int64 `json:"id"`
|
||||
Username string `json:"username"`
|
||||
}
|
||||
var rows []row
|
||||
err := g.DB().Ctx(ctx).Model("tnt_users").
|
||||
Where("tenant_id", tenantID).
|
||||
WhereIn("id", userIDs).
|
||||
Fields("id, username").
|
||||
Scan(&rows)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
for _, r := range rows {
|
||||
mapKey := fmt.Sprintf("%d:%d", tenantID, r.Id)
|
||||
result[mapKey] = r.Username
|
||||
}
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
// BatchQueryTenantNames 批量查询租户名。
|
||||
func BatchQueryTenantNames(ctx context.Context, tenantIDs []int64) map[int64]string {
|
||||
if len(tenantIDs) == 0 {
|
||||
return nil
|
||||
}
|
||||
result := make(map[int64]string, len(tenantIDs))
|
||||
// 去重
|
||||
seen := make(map[int64]bool, len(tenantIDs))
|
||||
unique := make([]int64, 0, len(tenantIDs))
|
||||
for _, id := range tenantIDs {
|
||||
if !seen[id] {
|
||||
seen[id] = true
|
||||
unique = append(unique, id)
|
||||
}
|
||||
}
|
||||
|
||||
type row struct {
|
||||
Id int64 `json:"id"`
|
||||
Name string `json:"name"`
|
||||
}
|
||||
var rows []row
|
||||
err := g.DB().Ctx(ctx).Model("tnt_tenants").
|
||||
WhereIn("id", unique).
|
||||
Fields("id, name").
|
||||
Scan(&rows)
|
||||
if err != nil {
|
||||
return result
|
||||
}
|
||||
for _, r := range rows {
|
||||
result[r.Id] = r.Name
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
// BatchQueryApiKeyNames 批量查询 API Key 名。
|
||||
func BatchQueryApiKeyNames(ctx context.Context, apiKeyIDs []int64) map[int64]string {
|
||||
if len(apiKeyIDs) == 0 {
|
||||
return nil
|
||||
}
|
||||
result := make(map[int64]string, len(apiKeyIDs))
|
||||
// 去重
|
||||
seen := make(map[int64]bool, len(apiKeyIDs))
|
||||
unique := make([]int64, 0, len(apiKeyIDs))
|
||||
for _, id := range apiKeyIDs {
|
||||
if !seen[id] {
|
||||
seen[id] = true
|
||||
unique = append(unique, id)
|
||||
}
|
||||
}
|
||||
|
||||
type row struct {
|
||||
Id int64 `json:"id"`
|
||||
Name string `json:"name"`
|
||||
}
|
||||
var rows []row
|
||||
err := g.DB().Ctx(ctx).Model("api_keys").
|
||||
WhereIn("id", unique).
|
||||
Fields("id, name").
|
||||
Scan(&rows)
|
||||
if err != nil {
|
||||
return result
|
||||
}
|
||||
for _, r := range rows {
|
||||
result[r.Id] = r.Name
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
// BatchQueryProjectNames 批量查询项目名。
|
||||
func BatchQueryProjectNames(ctx context.Context, projectIDs []int64) map[int64]string {
|
||||
if len(projectIDs) == 0 {
|
||||
return nil
|
||||
}
|
||||
result := make(map[int64]string, len(projectIDs))
|
||||
// 去重
|
||||
seen := make(map[int64]bool, len(projectIDs))
|
||||
unique := make([]int64, 0, len(projectIDs))
|
||||
for _, id := range projectIDs {
|
||||
if !seen[id] {
|
||||
seen[id] = true
|
||||
unique = append(unique, id)
|
||||
}
|
||||
}
|
||||
|
||||
type row struct {
|
||||
Id int64 `json:"id"`
|
||||
Name string `json:"name"`
|
||||
}
|
||||
var rows []row
|
||||
err := g.DB().Ctx(ctx).Model("tnt_projects").
|
||||
WhereIn("id", unique).
|
||||
Fields("id, name").
|
||||
Scan(&rows)
|
||||
if err != nil {
|
||||
return result
|
||||
}
|
||||
for _, r := range rows {
|
||||
result[r.Id] = r.Name
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
// collectInt64Fields 从记录集合中收集指定字段的所有非零 int64 值(去重)。
|
||||
func collectInt64Fields(records []map[string]any, field string) []int64 {
|
||||
seen := make(map[int64]bool)
|
||||
result := make([]int64, 0)
|
||||
for _, record := range records {
|
||||
if v, ok := getInt64Field(record, field); ok && v > 0 && !seen[v] {
|
||||
seen[v] = true
|
||||
result = append(result, v)
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
// collectUserKeys 收集 "tenantID:userID" 格式的 key(去重)。
|
||||
func collectUserKeys(records []map[string]any) []string {
|
||||
seen := make(map[string]bool)
|
||||
result := make([]string, 0)
|
||||
for _, record := range records {
|
||||
if key, ok := getUserKey(record); ok && !seen[key] {
|
||||
seen[key] = true
|
||||
result = append(result, key)
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
// getInt64Field 从 map 中安全获取 int64 字段值。
|
||||
func getInt64Field(record map[string]any, field string) (int64, bool) {
|
||||
v, ok := record[field]
|
||||
if !ok || v == nil {
|
||||
return 0, false
|
||||
}
|
||||
switch val := v.(type) {
|
||||
case int64:
|
||||
return val, true
|
||||
case int:
|
||||
return int64(val), true
|
||||
case float64:
|
||||
return int64(val), true
|
||||
default:
|
||||
return 0, false
|
||||
}
|
||||
}
|
||||
|
||||
// getUserKey 从记录中构建 "tenantID:userID" 格式的 key。
|
||||
func getUserKey(record map[string]any) (string, bool) {
|
||||
tenantID, ok1 := getInt64Field(record, "tenant_id")
|
||||
userID, ok2 := getInt64Field(record, "user_id")
|
||||
if !ok1 || !ok2 || tenantID == 0 || userID == 0 {
|
||||
return "", false
|
||||
}
|
||||
return fmt.Sprintf("%d:%d", tenantID, userID), true
|
||||
}
|
||||
@@ -405,7 +405,7 @@ func RecordLoginHistory(ctx context.Context, userType string, userID, tenantID i
|
||||
|
||||
// Check if this device has been seen before
|
||||
if deviceFP != "" && success {
|
||||
count, err := dao.AudLoginHistory.Ctx(ctx).
|
||||
count, err := AuditModelCtx(ctx, "aud_login_history").
|
||||
Where("user_type", userType).
|
||||
Where("user_id", userID).
|
||||
Where("device_fingerprint", deviceFP).
|
||||
@@ -417,7 +417,7 @@ func RecordLoginHistory(ctx context.Context, userType string, userID, tenantID i
|
||||
}
|
||||
|
||||
if r != nil {
|
||||
_, err := dao.AudLoginHistory.Ctx(ctx).Data(do.AudLoginHistory{
|
||||
_, err := AuditModelCtx(ctx, "aud_login_history").Data(do.AudLoginHistory{
|
||||
UserType: userType,
|
||||
UserId: userID,
|
||||
TenantId: tenantID,
|
||||
|
||||
@@ -561,7 +561,7 @@ func (p *DataProviderImpl) RecordAudit(ctx context.Context, record *common.Audit
|
||||
insertData.TaskStatus = record.TaskStatus
|
||||
}
|
||||
|
||||
_, insertErr := dao.AudRequestLogs.Ctx(bgCtx).Data(insertData).Insert()
|
||||
_, insertErr := lcommon.AuditModelCtx(bgCtx, "aud_request_logs").Data(insertData).Insert()
|
||||
if insertErr != nil {
|
||||
g.Log().Errorf(bgCtx,
|
||||
"record audit log failed: request_id=%s tenant_id=%d api_key_id=%d path=%s status=%d err=%v",
|
||||
@@ -614,7 +614,7 @@ func (p *DataProviderImpl) UpdateTaskAudit(ctx context.Context, record *common.A
|
||||
updateData.TaskCompletedAt = gtime.NewFromTime(*record.TaskCompletedAt)
|
||||
}
|
||||
|
||||
_, err := dao.AudRequestLogs.Ctx(bgCtx).
|
||||
_, err := lcommon.AuditModelCtx(bgCtx, "aud_request_logs").
|
||||
Where("task_id", record.TaskID).
|
||||
Data(updateData).
|
||||
Update()
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
|
||||
"github.com/qianfree/team-api/internal/dao"
|
||||
"github.com/qianfree/team-api/internal/logic/common"
|
||||
do "github.com/qianfree/team-api/internal/model/do"
|
||||
)
|
||||
|
||||
@@ -69,7 +70,7 @@ func handleDataExport(ctx context.Context, payload json.RawMessage) (any, error)
|
||||
exportData["billing_records"] = records
|
||||
case "logs":
|
||||
var logs []map[string]any
|
||||
if err := dao.AudOperationLogs.Ctx(ctx).Where("tenant_id", p.TenantID).
|
||||
if err := common.AuditModelCtx(ctx, "aud_operation_logs").Where("tenant_id", p.TenantID).
|
||||
OrderDesc("created_at").Limit(10000).
|
||||
Fields("id, action, resource_type, resource_id, ip_address, user_agent, created_at").Scan(&logs); err != nil {
|
||||
g.Log().Warningf(ctx, "export logs for tenant %d: %v", p.TenantID, err)
|
||||
@@ -132,7 +133,7 @@ func handleDeletionRequest(ctx context.Context, payload json.RawMessage) (any, e
|
||||
return nil, fmt.Errorf("disable api keys: %w", err)
|
||||
}
|
||||
|
||||
if _, err := dao.AudSensitiveAccessLogs.Ctx(ctx).Where("tenant_id", p.TenantID).Delete(); err != nil {
|
||||
if _, err := common.AuditModelCtx(ctx, "aud_sensitive_access_logs").Where("tenant_id", p.TenantID).Delete(); err != nil {
|
||||
return nil, fmt.Errorf("delete sensitive access logs: %w", err)
|
||||
}
|
||||
|
||||
|
||||
@@ -106,12 +106,12 @@ func (s *sTenant) AuditLogs(ctx context.Context, req *v1.TenantAuditLogsReq) (*v
|
||||
// 使用原生 SQL 查询,绕过 GoFrame ScanAndCount 对 map[string]any 的 bug
|
||||
dataSQL := `SELECT id, tenant_id, user_id, user_type, action, resource_type, resource_id, detail, changes_json, ip_address, created_at
|
||||
FROM aud_operation_logs WHERE tenant_id = ? ORDER BY created_at DESC LIMIT ? OFFSET ?`
|
||||
result, err := g.DB().Query(ctx, dataSQL, tenantID, pageSize, (page-1)*pageSize)
|
||||
result, err := common.GetAuditDB().Query(ctx, dataSQL, tenantID, pageSize, (page-1)*pageSize)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
countResult, err := g.DB().Query(ctx, "SELECT COUNT(*) AS total FROM aud_operation_logs WHERE tenant_id = ?", tenantID)
|
||||
countResult, err := common.GetAuditDB().Query(ctx, "SELECT COUNT(*) AS total FROM aud_operation_logs WHERE tenant_id = ?", tenantID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -167,55 +167,86 @@ func (s *sTenant) TenantRequestAuditLogs(ctx context.Context, req *v1.TenantRequ
|
||||
var conditions []string
|
||||
var args []any
|
||||
|
||||
conditions = append(conditions, "a.tenant_id = ?")
|
||||
conditions = append(conditions, "tenant_id = ?")
|
||||
args = append(args, tenantID)
|
||||
|
||||
// 用户名筛选:先从主库查匹配的 user_id 集合,转为 IN 条件
|
||||
if req.Username != "" {
|
||||
conditions = append(conditions, "t.username LIKE ?")
|
||||
args = append(args, "%"+req.Username+"%")
|
||||
type userRow struct {
|
||||
Id int64 `json:"id"`
|
||||
}
|
||||
var users []userRow
|
||||
err := g.DB().Ctx(ctx).Model("tnt_users").
|
||||
Where("tenant_id", tenantID).
|
||||
Where("username LIKE ?", "%"+req.Username+"%").
|
||||
Fields("id").
|
||||
Scan(&users)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(users) == 0 {
|
||||
return &v1.TenantRequestAuditLogsRes{
|
||||
List: []map[string]any{}, Total: 0, Page: page, PageSize: pageSize,
|
||||
}, nil
|
||||
}
|
||||
userIDs := make([]int64, 0, len(users))
|
||||
for _, u := range users {
|
||||
userIDs = append(userIDs, u.Id)
|
||||
}
|
||||
conditions = append(conditions, "user_id IN (?)")
|
||||
args = append(args, userIDs)
|
||||
}
|
||||
if req.RequestId != "" {
|
||||
conditions = append(conditions, "a.request_id = ?")
|
||||
conditions = append(conditions, "request_id = ?")
|
||||
args = append(args, req.RequestId)
|
||||
}
|
||||
if req.TaskId != "" {
|
||||
conditions = append(conditions, "a.task_id = ?")
|
||||
conditions = append(conditions, "task_id = ?")
|
||||
args = append(args, req.TaskId)
|
||||
}
|
||||
if req.Path != "" {
|
||||
conditions = append(conditions, "a.path LIKE ?")
|
||||
conditions = append(conditions, "path LIKE ?")
|
||||
args = append(args, "%"+req.Path+"%")
|
||||
}
|
||||
if req.StatusCode > 0 {
|
||||
conditions = append(conditions, "a.status_code = ?")
|
||||
conditions = append(conditions, "status_code = ?")
|
||||
args = append(args, req.StatusCode)
|
||||
}
|
||||
if req.StartDate != "" {
|
||||
conditions = append(conditions, "a.created_at >= ?")
|
||||
conditions = append(conditions, "created_at >= ?")
|
||||
args = append(args, req.StartDate+" 00:00:00")
|
||||
}
|
||||
if req.EndDate != "" {
|
||||
conditions = append(conditions, "a.created_at <= ?")
|
||||
conditions = append(conditions, "created_at <= ?")
|
||||
args = append(args, req.EndDate+" 23:59:59")
|
||||
}
|
||||
|
||||
where := strings.Join(conditions, " AND ")
|
||||
fromClause := "aud_request_logs a LEFT JOIN tnt_users t ON a.user_id = t.id AND a.tenant_id = t.tenant_id LEFT JOIN tnt_projects p ON a.project_id = p.id"
|
||||
|
||||
countSQL := "SELECT COUNT(*) AS total FROM " + fromClause + " WHERE " + where
|
||||
countResult, err := g.DB().Query(ctx, countSQL, args...)
|
||||
// Step 1: 从审计库查询审计记录(不带 JOIN)
|
||||
auditDB := common.GetAuditDB()
|
||||
countM := auditDB.Ctx(ctx).Model("aud_request_logs").Safe()
|
||||
if where != "" {
|
||||
countM = countM.Where(where, args...)
|
||||
}
|
||||
total, err := countM.Count()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
total := 0
|
||||
if len(countResult) > 0 {
|
||||
total = countResult[0]["total"].Int()
|
||||
if total == 0 {
|
||||
return &v1.TenantRequestAuditLogsRes{
|
||||
List: []map[string]any{}, Total: 0, Page: page, PageSize: pageSize,
|
||||
}, nil
|
||||
}
|
||||
|
||||
dataSQL := `SELECT a.id, a.request_id, COALESCE(t.username, '') AS username, a.project_id, COALESCE(p.name, '') AS project_name, a.user_id, a.method, a.path, a.query_params, a.status_code, a.client_ip, a.user_agent, a.latency_ms, a.first_token_ms, a.tenant_audit_level AS audit_level, a.task_id, a.task_status, a.task_completed_at, a.created_at
|
||||
FROM ` + fromClause + ` WHERE ` + where + ` ORDER BY a.created_at DESC LIMIT ? OFFSET ?`
|
||||
dataArgs := append(args, pageSize, (page-1)*pageSize)
|
||||
result, err := g.DB().Query(ctx, dataSQL, dataArgs...)
|
||||
dataM := auditDB.Ctx(ctx).Model("aud_request_logs").Safe().
|
||||
Fields("id, request_id, user_id, project_id, method, path, query_params, status_code, client_ip, user_agent, latency_ms, first_token_ms, tenant_audit_level as audit_level, task_id, task_status, task_completed_at, created_at").
|
||||
OrderDesc("created_at").
|
||||
Page(page, pageSize)
|
||||
if where != "" {
|
||||
dataM = dataM.Where(where, args...)
|
||||
}
|
||||
result, err := dataM.All()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -229,15 +260,16 @@ func (s *sTenant) TenantRequestAuditLogs(ctx context.Context, req *v1.TenantRequ
|
||||
items = append(items, m)
|
||||
}
|
||||
|
||||
// Step 2: 从主库批量查询关联信息并合并
|
||||
common.EnrichAuditRecords(ctx, items)
|
||||
|
||||
return &v1.TenantRequestAuditLogsRes{
|
||||
List: items,
|
||||
Total: total,
|
||||
Page: page,
|
||||
PageSize: pageSize,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// TenantRequestAuditLogDetail 查询单条请求审计日志详情(含 request_body 和 response_body)
|
||||
} // TenantRequestAuditLogDetail 查询单条请求审计日志详情(含 request_body 和 response_body)
|
||||
func (s *sTenant) TenantRequestAuditLogDetail(ctx context.Context, req *v1.TenantRequestAuditLogDetailReq) (*v1.TenantRequestAuditLogDetailRes, error) {
|
||||
role := middleware.GetUserRole(ctx)
|
||||
if role != "owner" && role != "admin" {
|
||||
@@ -246,7 +278,7 @@ func (s *sTenant) TenantRequestAuditLogDetail(ctx context.Context, req *v1.Tenan
|
||||
tenantID := middleware.GetTenantID(ctx)
|
||||
|
||||
var record *entity.AudRequestLogs
|
||||
err := dao.AudRequestLogs.Ctx(ctx).
|
||||
err := common.AuditModelCtx(ctx, "aud_request_logs").
|
||||
Where("id", req.Id).
|
||||
Where("tenant_id", tenantID).
|
||||
Scan(&record)
|
||||
|
||||
@@ -201,7 +201,7 @@ func (s *sTenant) LoginHistory(ctx context.Context, req *v1.TenantLoginHistoryRe
|
||||
role := middleware.GetUserRole(ctx)
|
||||
page, pageSize := common.NormalizePagination(req.Page, req.PageSize)
|
||||
|
||||
q := dao.AudLoginHistory.Ctx(ctx).
|
||||
q := common.AuditModelCtx(ctx, "aud_login_history").
|
||||
Where("user_type", "tenant").
|
||||
Where("tenant_id", tenantID)
|
||||
|
||||
|
||||
@@ -8,7 +8,6 @@ import (
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
"github.com/gogf/gf/v2/net/ghttp"
|
||||
|
||||
"github.com/qianfree/team-api/internal/dao"
|
||||
"github.com/qianfree/team-api/internal/logic/common"
|
||||
do "github.com/qianfree/team-api/internal/model/do"
|
||||
)
|
||||
@@ -55,7 +54,7 @@ func init() {
|
||||
func contentFilterLogWorker() {
|
||||
ctx := context.Background()
|
||||
for entry := range contentFilterLogCh {
|
||||
_, err := dao.AudContentFilterLogs.Ctx(ctx).Data(do.AudContentFilterLogs{
|
||||
_, err := common.AuditModelCtx(ctx, "aud_content_filter_logs").Data(do.AudContentFilterLogs{
|
||||
TenantId: entry.TenantId,
|
||||
UserId: entry.UserId,
|
||||
ApiKeyId: entry.ApiKeyId,
|
||||
|
||||
@@ -9,7 +9,7 @@ import (
|
||||
"github.com/gogf/gf/v2/net/ghttp"
|
||||
"github.com/gogf/gf/v2/util/gconv"
|
||||
|
||||
"github.com/qianfree/team-api/internal/dao"
|
||||
lcommon "github.com/qianfree/team-api/internal/logic/common"
|
||||
do "github.com/qianfree/team-api/internal/model/do"
|
||||
)
|
||||
|
||||
@@ -150,7 +150,7 @@ func OperationLog(r *ghttp.Request) {
|
||||
|
||||
go func() {
|
||||
ctx := context.Background()
|
||||
_, err := dao.AudOperationLogs.Ctx(ctx).Data(logData).Insert()
|
||||
_, err := lcommon.AuditModelCtx(ctx, "aud_operation_logs").Data(logData).Insert()
|
||||
if err != nil {
|
||||
g.Log().Errorf(ctx, "write operation log: %v", err)
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user