307 lines
16 KiB
Go
307 lines
16 KiB
Go
package server
|
||
|
||
import (
|
||
"archive/zip"
|
||
"bytes"
|
||
"database/sql"
|
||
"encoding/json"
|
||
"fmt"
|
||
"net/http"
|
||
"regexp"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/gin-gonic/gin"
|
||
"github.com/jmoiron/sqlx"
|
||
"vdi-cloud/platform/internal/store"
|
||
)
|
||
|
||
const diagnosticExportLimit = 2000
|
||
|
||
type diagnosticSubject struct {
|
||
Kind string `json:"kind" db:"kind"`
|
||
ID string `json:"id" db:"id"`
|
||
Name string `json:"name" db:"name"`
|
||
State string `json:"state" db:"state"`
|
||
Node string `json:"node" db:"node"`
|
||
VMID int `json:"vmid" db:"vmid"`
|
||
AgentVersion string `json:"agentVersion" db:"agent_version"`
|
||
LastSeen *time.Time `json:"lastSeen" db:"last_seen"`
|
||
}
|
||
type diagnosticEvent struct {
|
||
ID string `json:"id" db:"id"`
|
||
Source string `json:"source" db:"source"`
|
||
Time time.Time `json:"time" db:"time"`
|
||
Action string `json:"action" db:"action"`
|
||
Status string `json:"status" db:"status"`
|
||
TargetID string `json:"targetId" db:"target_id"`
|
||
TargetName string `json:"targetName" db:"target_name"`
|
||
Actor string `json:"actor" db:"actor"`
|
||
Message string `json:"message" db:"message"`
|
||
Details string `json:"details" db:"details"`
|
||
}
|
||
type diagnosticReport struct {
|
||
Subject diagnosticSubject `json:"subject"`
|
||
From time.Time `json:"from"`
|
||
To time.Time `json:"to"`
|
||
ObservedAt time.Time `json:"observedAt"`
|
||
Items []diagnosticEvent `json:"items"`
|
||
Counts map[string]int `json:"counts"`
|
||
Total int `json:"total"`
|
||
Page int `json:"page"`
|
||
PageSize int `json:"pageSize"`
|
||
Source string `json:"source"`
|
||
Notes []string `json:"notes"`
|
||
}
|
||
type terminalLogArchive struct {
|
||
ID string `json:"id" db:"id"`
|
||
TaskID string `json:"taskId" db:"task_id"`
|
||
TerminalID string `json:"terminalId" db:"terminal_id"`
|
||
TerminalName string `json:"terminalName" db:"terminal_name"`
|
||
CapturedAt time.Time `json:"capturedAt" db:"captured_at"`
|
||
BootID string `json:"bootId" db:"boot_id"`
|
||
AgentVersion string `json:"agentVersion" db:"agent_version"`
|
||
KioskState string `json:"kioskState" db:"kiosk_state"`
|
||
LogText string `json:"logText" db:"log_text"`
|
||
}
|
||
type terminalLogArchivePage struct {
|
||
Items []terminalLogArchive `json:"items"`
|
||
Page int `json:"page"`
|
||
PageSize int `json:"pageSize"`
|
||
Total int `json:"total"`
|
||
ObservedAt time.Time `json:"observedAt"`
|
||
RetentionDays int `json:"retentionDays"`
|
||
}
|
||
|
||
func (s *Server) terminalLogArchives(c *gin.Context) {
|
||
terminalID, keyword := strings.TrimSpace(c.Query("terminalId")), strings.TrimSpace(c.Query("keyword"))
|
||
if len(terminalID) > 100 || len(keyword) > 200 || strings.IndexByte(terminalID, 0) >= 0 || strings.IndexByte(keyword, 0) >= 0 {
|
||
s.fail(c, 400, "invalid_request", "日志筛选条件无效")
|
||
return
|
||
}
|
||
from, to, ok := timeRange(c, 30)
|
||
if !ok {
|
||
s.fail(c, 400, "invalid_range", "请选择有效时间范围，最长 30 天，时间需包含时区")
|
||
return
|
||
}
|
||
page, size := pageParams(c)
|
||
args := []any{from, to, terminalID, keyword}
|
||
var total int
|
||
if err := s.DB.GetContext(c.Request.Context(), &total, `SELECT count(*) FROM terminal_log_archives WHERE captured_at>=$1 AND captured_at<=$2 AND ($3='' OR terminal_id=$3) AND ($4='' OR log_text ILIKE '%'||$4||'%')`, args...); err != nil {
|
||
s.dbError(c, err)
|
||
return
|
||
}
|
||
items := []terminalLogArchive{}
|
||
args = append(args, size, (page-1)*size)
|
||
if err := s.DB.SelectContext(c.Request.Context(), &items, `SELECT a.id,a.task_id,a.terminal_id,t.name terminal_name,a.captured_at,a.boot_id,a.agent_version,a.kiosk_state,a.log_text FROM terminal_log_archives a JOIN terminals t ON t.id=a.terminal_id WHERE a.captured_at>=$1 AND a.captured_at<=$2 AND ($3='' OR a.terminal_id=$3) AND ($4='' OR a.log_text ILIKE '%'||$4||'%') ORDER BY a.captured_at DESC,a.id DESC LIMIT $5 OFFSET $6`, args...); err != nil {
|
||
s.dbError(c, err)
|
||
return
|
||
}
|
||
for i := range items {
|
||
items[i].BootID = redactDiagnostic(items[i].BootID)
|
||
items[i].AgentVersion = redactDiagnostic(items[i].AgentVersion)
|
||
items[i].KioskState = redactDiagnostic(items[i].KioskState)
|
||
items[i].TerminalName = redactDiagnostic(items[i].TerminalName)
|
||
items[i].LogText = redactDiagnostic(items[i].LogText)
|
||
}
|
||
c.JSON(200, terminalLogArchivePage{Items: items, Page: page, PageSize: size, Total: total, ObservedAt: time.Now().UTC(), RetentionDays: 30})
|
||
}
|
||
|
||
// Correlation uses stable IDs, never a display name. Desktop logs are not
|
||
// inferred from a device's current binding: that device may serve other VMs.
|
||
const diagnosticEventsSQL = `WITH scoped_tasks AS (
|
||
SELECT * FROM tasks WHERE
|
||
($4='terminal' AND ((kind='terminal' AND target_id=$1) OR (kind='gpu-prepare' AND actor='device:'||$1)))
|
||
OR ($4='desktop' AND target_id=$1 AND kind!='terminal')
|
||
), scoped_sessions AS (
|
||
SELECT * FROM session_history WHERE ($4='terminal' AND terminal_id=$1) OR ($4='desktop' AND desktop_id=$1)
|
||
), scoped_alerts AS (
|
||
SELECT * FROM alerts WHERE ($4='terminal' AND target_id=$1 AND metric LIKE 'terminal_%')
|
||
OR ($4='desktop' AND (target_id=$1 OR (target_id=$5 AND metric LIKE 'node_%')))
|
||
), events AS (
|
||
SELECT 'task:'||id id,'tasks' source,created_at time,action,status,target_id,target_name,actor,
|
||
'任务提交；状态为查询时当前状态' message,jsonb_build_object('taskId',id,'kind',kind)::text details FROM scoped_tasks
|
||
UNION ALL
|
||
SELECT 'task-update:'||id,'tasks',updated_at,action,status,target_id,target_name,actor,message,
|
||
jsonb_build_object('taskId',id,'kind',kind)::text FROM scoped_tasks WHERE updated_at>created_at
|
||
UNION ALL
|
||
SELECT 'audit:'||a.id,'audits',a.created_at,a.action,a.result,a.target,a.target,a.actor,'配置或操作审计',
|
||
jsonb_build_object('auditId',a.id,'requestId',a.request_id)::text FROM audit_logs a
|
||
WHERE a.target=$1 OR a.target IN (SELECT id FROM scoped_tasks) OR a.request_id IN (SELECT id FROM scoped_tasks)
|
||
UNION ALL
|
||
SELECT 'session-start:'||id,'sessions',reserved_at,'session-reserved','connecting',desktop_id,desktop_name,username,
|
||
terminal_name||' → '||desktop_name,jsonb_build_object('historyId',id,'terminalId',terminal_id,'terminalName',terminal_name,'desktopId',desktop_id,'source',source)::text FROM scoped_sessions
|
||
UNION ALL
|
||
SELECT 'session-connected:'||id,'sessions',connected_at,'session-connected','connected',desktop_id,desktop_name,username,
|
||
terminal_name||' → '||desktop_name,jsonb_build_object('historyId',id,'terminalId',terminal_id,'terminalName',terminal_name,'source',source)::text FROM scoped_sessions WHERE connected_at IS NOT NULL
|
||
UNION ALL
|
||
SELECT 'session-ended:'||id,'sessions',ended_at,'session-ended',state,desktop_id,desktop_name,username,end_reason,
|
||
jsonb_build_object('historyId',id,'terminalId',terminal_id,'terminalName',terminal_name)::text FROM scoped_sessions WHERE ended_at IS NOT NULL
|
||
UNION ALL
|
||
SELECT 'alert-open:'||id,'alerts',opened_at,'alert-opened','firing',target_id,target_name,'',rule_name,
|
||
jsonb_build_object('alertId',id,'metric',metric,'value',value,'threshold',threshold)::text FROM scoped_alerts
|
||
UNION ALL
|
||
SELECT 'alert-ack:'||id,'alerts',acknowledged_at,'acknowledge_alert','acknowledged',target_id,target_name,acknowledged_by,rule_name,
|
||
jsonb_build_object('alertId',id)::text FROM scoped_alerts WHERE acknowledged_at IS NOT NULL
|
||
UNION ALL
|
||
SELECT 'alert-resolved:'||id,'alerts',resolved_at,'alert-resolved','resolved',target_id,target_name,'',rule_name,
|
||
jsonb_build_object('alertId',id,'reason',resolve_reason)::text FROM scoped_alerts WHERE resolved_at IS NOT NULL
|
||
UNION ALL
|
||
SELECT 'log:'||t.id,'logs',d.created_at,'collect-logs','success',t.target_id,t.target_name,t.actor,'终端已回传的日志片段',
|
||
jsonb_build_object('taskId',t.id,'agentVersion',left(d.content->>'agentVersion',80),'bootId',left(d.content->>'bootId',80),
|
||
'uptimeSeconds',d.content->'uptimeSeconds','kioskState',left(d.content->>'kioskState',80),'logTail',left(d.content->>'logTail',16384))::text
|
||
FROM task_diagnostics d JOIN scoped_tasks t ON t.id=d.task_id WHERE $4='terminal' AND t.kind='terminal' AND t.action='collect-logs' AND t.status='success'
|
||
), filtered AS (SELECT * FROM events WHERE time>=$2 AND time<=$3)
|
||
`
|
||
|
||
// Redact common credential formats in free text before it reaches either the
|
||
// browser or the archive. No provider payload, ticket or activation data is read.
|
||
var diagnosticRedactions = []*regexp.Regexp{
|
||
regexp.MustCompile(`(?is)-----BEGIN [^-]*PRIVATE KEY-----.*?-----END [^-]*PRIVATE KEY-----`),
|
||
regexp.MustCompile(`(?i)\b(?:Bearer|Basic)\s+[A-Za-z0-9._~+/=-]+`),
|
||
regexp.MustCompile(`(?i)\bPVEAPIToken=[^\s"'<>]+`),
|
||
regexp.MustCompile(`(?i)\b(?:device|activate|ticket)_[a-f0-9]{16,}\b`),
|
||
regexp.MustCompile(`\beyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\b`),
|
||
regexp.MustCompile(`(?i)(?:"|')?(?:password|passwd|token|access[_-]?token|refresh[_-]?token|activation[_-]?secret|secret|api[_-]?key|authorization|cookie|ticket)(?:"|')?\s*[:=]\s*(?:"[^"\r\n]*"|'[^'\r\n]*'|[^\s&,;\r\n]+)`),
|
||
regexp.MustCompile(`(?i)\b[a-z][a-z0-9+.-]*://[^\s/@]+:[^\s/@]+@`),
|
||
}
|
||
|
||
func redactDiagnostic(value string) string {
|
||
for _, pattern := range diagnosticRedactions {
|
||
value = pattern.ReplaceAllString(value, "[REDACTED]")
|
||
}
|
||
return value
|
||
}
|
||
func sanitizeDiagnosticEvent(e *diagnosticEvent) {
|
||
e.TargetName = redactDiagnostic(e.TargetName)
|
||
e.Actor = redactDiagnostic(e.Actor)
|
||
e.Message = redactDiagnostic(e.Message)
|
||
var fields map[string]any
|
||
if json.Unmarshal([]byte(e.Details), &fields) != nil {
|
||
e.Details = "{}"
|
||
return
|
||
}
|
||
for key, value := range fields {
|
||
if text, ok := value.(string); ok {
|
||
fields[key] = redactDiagnostic(text)
|
||
}
|
||
}
|
||
data, _ := json.Marshal(fields)
|
||
e.Details = string(data)
|
||
}
|
||
|
||
func (s *Server) diagnosticReport(c *gin.Context, export bool) (*diagnosticReport, bool) {
|
||
kind, id, source := c.Query("targetType"), c.Query("targetId"), c.Query("source")
|
||
if (kind != "terminal" && kind != "desktop") || id == "" || len(id) > 100 || strings.IndexByte(id, 0) >= 0 || (source != "" && source != "tasks" && source != "sessions" && source != "alerts" && source != "audits" && source != "logs") {
|
||
s.fail(c, 400, "invalid_request", "请选择设备或桌面，以及有效的记录来源")
|
||
return nil, false
|
||
}
|
||
from, to, ok := timeRange(c, 7)
|
||
if !ok {
|
||
s.fail(c, 400, "invalid_range", "请选择有效时间范围，最长 7 天，时间需包含时区")
|
||
return nil, false
|
||
}
|
||
tx, err := s.DB.BeginTxx(c.Request.Context(), &sql.TxOptions{Isolation: sql.LevelRepeatableRead, ReadOnly: true})
|
||
if err != nil {
|
||
s.dbError(c, err)
|
||
return nil, false
|
||
}
|
||
defer tx.Rollback()
|
||
report := &diagnosticReport{From: from, To: to, ObservedAt: time.Now().UTC(), Items: []diagnosticEvent{}, Counts: map[string]int{"tasks": 0, "audits": 0, "sessions": 0, "alerts": 0, "logs": 0}, Source: source,
|
||
Notes: []string{"按唯一标识关联；按名称记录且缺少稳定关联标识的旧审计不会推测关联。", "当前对象快照独立于历史时间范围；任务提交条目的状态为查询时当前状态，观察到的会话起点不代表真实首次连接时间。", "桌面视角包含所属节点告警；终端日志只在设备视角展示，不根据当前绑定推断历史日志属于哪台桌面。", "日志为设备已回传片段，不是持续实时日志；未收集、未上报和超出留存期均可能没有记录。", "已遮盖常见凭据格式；诊断包仍含设备标识、人员名称及运行信息，请按实际需要共享。"}}
|
||
if err = loadDiagnosticSubject(c, tx, kind, id, &report.Subject); err != nil {
|
||
s.dbError(c, err)
|
||
return nil, false
|
||
}
|
||
args := []any{id, from, to, kind, report.Subject.Node}
|
||
var counts []struct {
|
||
Source string `db:"source"`
|
||
Total int `db:"total"`
|
||
}
|
||
if err = tx.SelectContext(c.Request.Context(), &counts, diagnosticEventsSQL+"SELECT source,count(*) total FROM filtered GROUP BY source", args...); err != nil {
|
||
s.dbError(c, err)
|
||
return nil, false
|
||
}
|
||
for _, count := range counts {
|
||
report.Counts[count.Source] = count.Total
|
||
if source == "" || source == count.Source {
|
||
report.Total += count.Total
|
||
}
|
||
}
|
||
if export && report.Total > diagnosticExportLimit {
|
||
s.fail(c, 413, "export_too_large", "诊断事件超过 2000 条，请缩小时间范围或选择记录来源后导出")
|
||
return nil, false
|
||
}
|
||
report.Page, report.PageSize = pageParams(c)
|
||
if export {
|
||
report.Page = 1
|
||
report.PageSize = diagnosticExportLimit
|
||
}
|
||
args = append(args, source, report.PageSize, (report.Page-1)*report.PageSize)
|
||
if err = tx.SelectContext(c.Request.Context(), &report.Items, diagnosticEventsSQL+"SELECT * FROM filtered WHERE ($6='' OR source=$6) ORDER BY time DESC,source,id LIMIT $7 OFFSET $8", args...); err != nil {
|
||
s.dbError(c, err)
|
||
return nil, false
|
||
}
|
||
for i := range report.Items {
|
||
sanitizeDiagnosticEvent(&report.Items[i])
|
||
}
|
||
report.Subject.Name = redactDiagnostic(report.Subject.Name)
|
||
report.Subject.AgentVersion = redactDiagnostic(report.Subject.AgentVersion)
|
||
return report, true
|
||
}
|
||
func loadDiagnosticSubject(c *gin.Context, tx *sqlx.Tx, kind, id string, subject *diagnosticSubject) error {
|
||
if kind == "terminal" {
|
||
return tx.GetContext(c.Request.Context(), subject, `SELECT 'terminal' kind,id,name,CASE WHEN NOT enabled THEN 'disabled' WHEN status='online' AND last_seen<now()-interval '90 seconds' THEN 'offline' ELSE status END state,'' node,0 vmid,agent_version,last_seen FROM terminals WHERE id=$1`, id)
|
||
}
|
||
return tx.GetContext(c.Request.Context(), subject, `SELECT 'desktop' kind,id,name,power_state state,node,vmid,'' agent_version,NULL::timestamptz last_seen FROM desktops WHERE id=$1`, id)
|
||
}
|
||
func (s *Server) diagnostics(c *gin.Context) {
|
||
report, ok := s.diagnosticReport(c, false)
|
||
if !ok {
|
||
return
|
||
}
|
||
c.Header("Cache-Control", "no-store")
|
||
c.JSON(http.StatusOK, report)
|
||
}
|
||
func (s *Server) exportDiagnostics(c *gin.Context) {
|
||
report, ok := s.diagnosticReport(c, true)
|
||
if !ok {
|
||
return
|
||
}
|
||
content, err := json.MarshalIndent(report, "", " ")
|
||
if err != nil {
|
||
s.fail(c, 500, "export_failed", "诊断包生成失败")
|
||
return
|
||
}
|
||
var output bytes.Buffer
|
||
archive := zip.NewWriter(&output)
|
||
for _, entry := range []struct {
|
||
name string
|
||
data []byte
|
||
}{
|
||
{"diagnostics.json", content},
|
||
{"README.txt", []byte("VDI Cloud 诊断包\n包含所选唯一对象、时间范围和记录来源的全部匹配事件，最多 2000 条。\n当前对象信息为生成时快照。事件按 UTC 排序；任务当前结果不代表历史状态变更日志。\n日志为设备已回传片段；没有日志不代表设备没有故障。桌面视角不推断终端日志归属。\n遮盖常见凭据格式不保证消除所有个人信息或自由文本中的敏感信息。\n生成时不主动读取平台密钥、设备令牌、激活密钥或 PVE 原始任务载荷。自由日志中的自定义敏感格式仍需人工检查。\n")},
|
||
} {
|
||
writer, e := archive.Create(entry.name)
|
||
if e == nil {
|
||
_, e = writer.Write(entry.data)
|
||
}
|
||
if e != nil {
|
||
s.fail(c, 500, "export_failed", "诊断包生成失败")
|
||
return
|
||
}
|
||
}
|
||
if err = archive.Close(); err != nil {
|
||
s.fail(c, 500, "export_failed", "诊断包生成失败")
|
||
return
|
||
}
|
||
if _, err = s.DB.ExecContext(c.Request.Context(), `INSERT INTO audit_logs(id,actor,action,target,result,request_id) VALUES($1,$2,'export_diagnostics',$3,$4,$5)`, store.ID("audit_"), s.admin(c).Username, report.Subject.ID, fmt.Sprintf("success;events=%d", report.Total), c.GetString("requestId")); err != nil {
|
||
s.dbError(c, err)
|
||
return
|
||
}
|
||
c.Header("Cache-Control", "no-store")
|
||
c.Header("X-Content-Type-Options", "nosniff")
|
||
c.Header("Content-Disposition", `attachment; filename="vdi-diagnostics-`+time.Now().UTC().Format("20060102-150405")+`.zip"`)
|
||
c.Data(200, "application/zip", output.Bytes())
|
||
}
|