feat: 支持漏洞关联多条流量证据

db/finding_traffic.go

@@ -0,0 +1,404 @@
+package db
+
+import (
+	"context"
+	"crypto/sha256"
+	"database/sql"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+)
+
+var (
+	ErrEvidenceConflict = errors.New("流量证据已变更，请刷新后重试")
+	ErrFindingNotFound  = errors.New("漏洞不存在")
+	ErrEvidenceNotFound = errors.New("流量证据不存在")
+)
+
+// This lock covers the evidence filesystem as well as its SQL references. All
+// processes sharing the database use it, including readers, exports and GC.
+const findingEvidenceLockKey int64 = 7337741004
+
+func (d *DB) WithEvidenceTx(ctx context.Context, fn func(*sql.Tx) error) error {
+	tx, err := d.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+	if _, err = tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock($1)`, findingEvidenceLockKey); err != nil {
+		return err
+	}
+	if err = fn(tx); err != nil {
+		return err
+	}
+	return tx.Commit()
+}
+
+type TrafficRef struct {
+	TrafficID string `json:"traffic_id"`
+	Role      string `json:"role"`
+	Note      string `json:"note"`
+}
+
+func NormalizeTrafficRefs(refs []TrafficRef) ([]TrafficRef, error) {
+	out := make([]TrafficRef, 0, len(refs))
+	seen := map[string]bool{}
+	for _, ref := range refs {
+		ref.TrafficID = strings.TrimSpace(ref.TrafficID)
+		if ref.TrafficID == "" {
+			return nil, errors.New("traffic_id 不能为空")
+		}
+		if ref.Role == "" {
+			ref.Role = "supporting"
+		}
+		if !ValidTrafficRole(ref.Role) {
+			return nil, fmt.Errorf("无效的流量用途 %q", ref.Role)
+		}
+		if !seen[ref.TrafficID] {
+			out = append(out, ref)
+			seen[ref.TrafficID] = true
+		}
+	}
+	return out, nil
+}
+
+func ValidTrafficRole(role string) bool {
+	return role == "baseline" || role == "proof" || role == "verification" || role == "supporting"
+}
+
+type TrafficEvidenceSnapshot struct {
+	ID              string    `json:"id"`
+	SourceTrafficID string    `json:"source_traffic_id"`
+	CapturedAt      int64     `json:"captured_at"`
+	URL             string    `json:"url"`
+	Method          string    `json:"method"`
+	Status          int       `json:"status"`
+	ContentType     string    `json:"content_type"`
+	ReqHead         string    `json:"req_head,omitempty"`
+	RespHead        string    `json:"resp_head,omitempty"`
+	ReqHash         string    `json:"req_hash"`
+	RespHash        string    `json:"resp_hash"`
+	ReqLen          int64     `json:"req_len"`
+	RespLen         int64     `json:"resp_len"`
+	CreatedAt       time.Time `json:"created_at"`
+}
+
+type FindingTrafficBinding struct {
+	ID         int64                   `json:"id,string"`
+	FindingID  int64                   `json:"finding_id,string"`
+	SnapshotID string                  `json:"snapshot_id"`
+	Role       string                  `json:"role"`
+	Note       string                  `json:"note"`
+	Position   int                     `json:"position"`
+	CreatedAt  time.Time               `json:"created_at"`
+	Snapshot   TrafficEvidenceSnapshot `json:"snapshot"`
+}
+
+type FindingTraffic struct {
+	FindingID     int64                   `json:"finding_id,string"`
+	Version       int64                   `json:"version"`
+	ReportVersion int64                   `json:"report_version"`
+	Bindings      []FindingTrafficBinding `json:"bindings"`
+}
+
+type PreparedTrafficEvidence struct {
+	Ref      TrafficRef
+	Snapshot TrafficEvidenceSnapshot
+}
+
+func TrafficSnapshotID(snapshot TrafficEvidenceSnapshot) string {
+	snapshot.ID = ""
+	snapshot.CreatedAt = time.Time{}
+	raw, _ := json.Marshal(snapshot)
+	sum := sha256.Sum256(raw)
+	return hex.EncodeToString(sum[:])
+}
+
+// LockTaskEvidenceTx serializes writes with archive queueing (which locks the
+// same task row). Once queued, its snapshot must not acquire new evidence.
+func LockTaskEvidenceTx(tx *sql.Tx, taskID int64) error {
+	if taskID == 0 {
+		return nil
+	}
+	var deleted sql.NullTime
+	if err := tx.QueryRow(`SELECT deleted_at FROM tasks WHERE id=$1 FOR UPDATE`, taskID).Scan(&deleted); err != nil {
+		return err
+	}
+	if deleted.Valid {
+		return ErrTaskArchiveState
+	}
+	var state string
+	err := tx.QueryRow(`SELECT state FROM task_archives WHERE task_id=$1`, taskID).Scan(&state)
+	if err != nil && !errors.Is(err, sql.ErrNoRows) {
+		return err
+	}
+	if err == nil && state != ArchiveFailed {
+		return ErrTaskArchiveState
+	}
+	return nil
+}
+
+func LockFindingEvidenceTx(tx *sql.Tx, findingID int64, version *int64) error {
+	var taskID sql.NullInt64
+	if err := tx.QueryRow(`SELECT task_id FROM findings WHERE id=$1`, findingID).Scan(&taskID); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return ErrFindingNotFound
+		}
+		return err
+	}
+	if err := LockTaskEvidenceTx(tx, taskID.Int64); err != nil {
+		return err
+	}
+	var current int64
+	if err := tx.QueryRow(`SELECT evidence_version FROM findings WHERE id=$1 FOR UPDATE`, findingID).Scan(&current); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return ErrFindingNotFound
+		}
+		return err
+	}
+	if version != nil && *version != current {
+		return ErrEvidenceConflict
+	}
+	return nil
+}
+
+func InsertEvidenceSnapshotTx(tx *sql.Tx, s TrafficEvidenceSnapshot) error {
+	if s.ID != TrafficSnapshotID(s) {
+		return errors.New("证据快照元数据哈希不匹配")
+	}
+	if s.CreatedAt.IsZero() {
+		s.CreatedAt = time.Now().UTC()
+	}
+	_, err := tx.Exec(`INSERT INTO traffic_evidence_snapshots
+(id,source_traffic_id,captured_at,url,method,status,content_type,req_head,resp_head,req_hash,resp_hash,req_len,resp_len,created_at)
+VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14) ON CONFLICT(id) DO NOTHING`,
+		s.ID, s.SourceTrafficID, s.CapturedAt, s.URL, s.Method, s.Status, s.ContentType, s.ReqHead, s.RespHead, s.ReqHash, s.RespHash, s.ReqLen, s.RespLen, s.CreatedAt)
+	return err
+}
+
+func AddFindingTrafficTx(tx *sql.Tx, findingID int64, prepared []PreparedTrafficEvidence) error {
+	var position int
+	if err := tx.QueryRow(`SELECT COALESCE(MAX(position)+1,0) FROM finding_traffic_bindings WHERE finding_id=$1`, findingID).Scan(&position); err != nil {
+		return err
+	}
+	changed := false
+	for _, item := range prepared {
+		if err := InsertEvidenceSnapshotTx(tx, item.Snapshot); err != nil {
+			return err
+		}
+		if _, err := tx.Exec(`UPDATE traffic_evidence_snapshots SET unreferenced_at=NULL WHERE id=$1`, item.Snapshot.ID); err != nil {
+			return err
+		}
+		res, err := tx.Exec(`INSERT INTO finding_traffic_bindings(finding_id,snapshot_id,role,note,position)
+VALUES($1,$2,$3,$4,$5) ON CONFLICT(finding_id,snapshot_id) DO NOTHING`, findingID, item.Snapshot.ID, item.Ref.Role, item.Ref.Note, position)
+		if err != nil {
+			return err
+		}
+		n, err := res.RowsAffected()
+		if err != nil {
+			return err
+		}
+		if n > 0 {
+			changed = true
+			position++
+		}
+	}
+	if changed {
+		return bumpEvidenceVersionTx(tx, findingID)
+	}
+	return nil
+}
+
+func bumpEvidenceVersionTx(tx *sql.Tx, findingID int64) error {
+	_, err := tx.Exec(`UPDATE findings SET evidence_version=evidence_version+1 WHERE id=$1`, findingID)
+	return err
+}
+
+func FindingTrafficTx(tx *sql.Tx, findingID int64) (*FindingTraffic, error) {
+	out := &FindingTraffic{FindingID: findingID, Bindings: []FindingTrafficBinding{}}
+	if err := tx.QueryRow(`SELECT evidence_version,report_evidence_version FROM findings WHERE id=$1`, findingID).Scan(&out.Version, &out.ReportVersion); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return nil, ErrFindingNotFound
+		}
+		return nil, err
+	}
+	rows, err := tx.Query(`SELECT b.id,b.finding_id,b.snapshot_id,b.role,b.note,b.position,b.created_at,to_jsonb(s)
+FROM finding_traffic_bindings b JOIN traffic_evidence_snapshots s ON s.id=b.snapshot_id WHERE b.finding_id=$1 ORDER BY b.position,b.id`, findingID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var b FindingTrafficBinding
+		var raw []byte
+		if err := rows.Scan(&b.ID, &b.FindingID, &b.SnapshotID, &b.Role, &b.Note, &b.Position, &b.CreatedAt, &raw); err != nil {
+			return nil, err
+		}
+		if err := json.Unmarshal(raw, &b.Snapshot); err != nil {
+			return nil, err
+		}
+		out.Bindings = append(out.Bindings, b)
+	}
+	return out, rows.Err()
+}
+
+func (d *DB) GetFindingTraffic(ctx context.Context, findingID int64) (out *FindingTraffic, err error) {
+	err = d.WithEvidenceTx(ctx, func(tx *sql.Tx) error { var e error; out, e = FindingTrafficTx(tx, findingID); return e })
+	return
+}
+
+func (d *DB) EditFindingTraffic(ctx context.Context, findingID, bindingID, version int64, role, note *string, remove bool, order []int64) error {
+	if role != nil && !ValidTrafficRole(*role) {
+		return errors.New("无效的流量用途")
+	}
+	return d.WithEvidenceTx(ctx, func(tx *sql.Tx) error {
+		if err := LockFindingEvidenceTx(tx, findingID, &version); err != nil {
+			return err
+		}
+		if order != nil {
+			current, err := FindingTrafficTx(tx, findingID)
+			if err != nil {
+				return err
+			}
+			if len(order) != len(current.Bindings) {
+				return ErrEvidenceConflict
+			}
+			ids := map[int64]bool{}
+			for _, b := range current.Bindings {
+				ids[b.ID] = true
+			}
+			for position, id := range order {
+				if !ids[id] {
+					return ErrEvidenceConflict
+				}
+				delete(ids, id)
+				if _, err := tx.Exec(`UPDATE finding_traffic_bindings SET position=$1 WHERE finding_id=$2 AND id=$3`, position, findingID, id); err != nil {
+					return err
+				}
+			}
+		} else {
+			var res sql.Result
+			var err error
+			if remove {
+				res, err = tx.Exec(`DELETE FROM finding_traffic_bindings WHERE finding_id=$1 AND id=$2`, findingID, bindingID)
+			} else {
+				res, err = tx.Exec(`UPDATE finding_traffic_bindings SET role=COALESCE($3,role),note=COALESCE($4,note) WHERE finding_id=$1 AND id=$2`, findingID, bindingID, role, note)
+			}
+			if err != nil {
+				return err
+			}
+			n, err := res.RowsAffected()
+			if err != nil {
+				return err
+			}
+			if n == 0 {
+				return ErrEvidenceNotFound
+			}
+		}
+		return bumpEvidenceVersionTx(tx, findingID)
+	})
+}
+
+type RecordFindingInput struct {
+	TaskID, ExplorationID, IntentID                      int64
+	VulnClass, Name, Severity, Summary, Evidence, Worker string
+	AssetIDs                                             []int64
+}
+
+type RecordedFinding struct {
+	FindingID int64           `json:"finding_id,string"`
+	NodeID    int64           `json:"finding_node_id,string"`
+	Traffic   *FindingTraffic `json:"traffic"`
+}
+
+func RecordFindingTx(tx *sql.Tx, in RecordFindingInput, prepared []PreparedTrafficEvidence) (*RecordedFinding, error) {
+	if err := LockTaskEvidenceTx(tx, in.TaskID); err != nil {
+		return nil, err
+	}
+	if in.TaskID > 0 {
+		var expID int64
+		if err := tx.QueryRow(`SELECT exploration_id FROM tasks WHERE id=$1`, in.TaskID).Scan(&expID); err != nil {
+			return nil, err
+		}
+		if expID != in.ExplorationID {
+			return nil, errors.New("漏洞所属任务与探索记录不匹配")
+		}
+	}
+	if in.IntentID > 0 {
+		var ok bool
+		if err := tx.QueryRow(`SELECT EXISTS(SELECT 1 FROM exploration_nodes WHERE id=$1 AND exploration_id=$2 AND kind='intent')`, in.IntentID, in.ExplorationID).Scan(&ok); err != nil {
+			return nil, err
+		}
+		if !ok {
+			return nil, errors.New("intent_id 必须是本任务的意图（关联任务意图只读）")
+		}
+	}
+	payload, _ := json.Marshal(map[string]any{"vulnclass": in.VulnClass, "name": in.Name, "severity": in.Severity, "summary": in.Summary, "evidence": map[string]any{"by": in.Worker, "poc": in.Evidence}})
+	out := &RecordedFinding{}
+	if err := tx.QueryRow(`INSERT INTO exploration_nodes(exploration_id,kind,payload,priority,state,origin)
+VALUES($1,'finding',$2,9,'confirmed',$3) RETURNING id`, in.ExplorationID, string(payload), in.Worker).Scan(&out.NodeID); err != nil {
+		return nil, err
+	}
+	for _, asset := range in.AssetIDs {
+		if _, err := tx.Exec(`INSERT INTO exploration_anchors(node_id,asset_id) VALUES($1,$2) ON CONFLICT DO NOTHING`, out.NodeID, asset); err != nil {
+			return nil, err
+		}
+	}
+	if in.IntentID > 0 {
+		if _, err := tx.Exec(`INSERT INTO exploration_edges(exploration_id,src_id,rel,dst_id) VALUES($1,$2,$3,$4)`, in.ExplorationID, in.IntentID, RelYields, out.NodeID); err != nil {
+			return nil, err
+		}
+	}
+	assets := in.AssetIDs
+	if assets == nil {
+		assets = []int64{}
+	}
+	raw, _ := json.Marshal(assets)
+	if err := tx.QueryRow(`INSERT INTO findings(task_id,node_id,vulnclass,name,severity,summary,evidence,worker,asset_ids)
+VALUES(NULLIF($1,0),$2,$3,$4,$5,$6,$7,$8,$9) RETURNING id`, in.TaskID, out.NodeID, in.VulnClass, in.Name, in.Severity, in.Summary, in.Evidence, in.Worker, string(raw)).Scan(&out.FindingID); err != nil {
+		return nil, err
+	}
+	if err := AddFindingTrafficTx(tx, out.FindingID, prepared); err != nil {
+		return nil, err
+	}
+	var err error
+	out.Traffic, err = FindingTrafficTx(tx, out.FindingID)
+	return out, err
+}
+
+// RecordFinding is the atomic legacy/no-recorder path used by tools and tests.
+func (s *ExplorationStore) RecordFinding(ctx context.Context, in RecordFindingInput) (out *RecordedFinding, err error) {
+	in.ExplorationID = s.expID
+	err = s.db.WithEvidenceTx(ctx, func(tx *sql.Tx) error { var e error; out, e = RecordFindingTx(tx, in, nil); return e })
+	return
+}
+
+func (d *DB) FindingIDByNodeID(nodeID int64) (id int64, err error) {
+	err = d.QueryRow(`SELECT id FROM findings WHERE node_id=$1`, nodeID).Scan(&id)
+	return
+}
+
+func (d *DB) SetFindingReportVersionByNodeID(ctx context.Context, nodeID int64, report string, version *int64) (n int64, err error) {
+	err = d.WithEvidenceTx(ctx, func(tx *sql.Tx) error {
+		var id int64
+		if err := tx.QueryRow(`SELECT id FROM findings WHERE node_id=$1`, nodeID).Scan(&id); err != nil {
+			if errors.Is(err, sql.ErrNoRows) {
+				return nil
+			}
+			return err
+		}
+		if err := LockFindingEvidenceTx(tx, id, version); err != nil {
+			return err
+		}
+		res, err := tx.Exec(`UPDATE findings SET report=$2,report_evidence_version=COALESCE($3::bigint,CASE WHEN evidence_version=0 THEN 0 ELSE -1 END) WHERE id=$1`, id, report, version)
+		if err != nil {
+			return err
+		}
+		n, err = res.RowsAffected()
+		return err
+	})
+	return
+}

db/finding_traffic_archive.go

@@ -0,0 +1,85 @@
+package db
+
+import (
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+)
+
+func ArchiveEvidenceSnapshots(snapshot *TaskArchiveSnapshot) ([]TrafficEvidenceSnapshot, error) {
+	var out []TrafficEvidenceSnapshot
+	if rawRowCount(snapshot.Tables["traffic_evidence_snapshots"]) == 0 {
+		return out, nil
+	}
+	if snapshot.FormatVersion < 3 {
+		return nil, ErrTaskArchiveFormatMismatch
+	}
+	err := json.Unmarshal(snapshot.Tables["traffic_evidence_snapshots"], &out)
+	return out, err
+}
+
+func restoreFindingTrafficTx(tx *sql.Tx, snapshot *TaskArchiveSnapshot) error {
+	snapshots, err := ArchiveEvidenceSnapshots(snapshot)
+	if err != nil {
+		return err
+	}
+	allowed := map[string]bool{}
+	for _, v := range snapshots {
+		if allowed[v.ID] {
+			return errors.New("duplicate archived evidence snapshot")
+		}
+		allowed[v.ID] = true
+		if err = InsertEvidenceSnapshotTx(tx, v); err != nil {
+			return err
+		}
+	}
+	rows, err := decodeArchiveRows(snapshot.Tables["finding_traffic_bindings"])
+	if err != nil {
+		return err
+	}
+	if len(rows) > 0 && snapshot.FormatVersion < 3 {
+		return ErrTaskArchiveFormatMismatch
+	}
+	for _, row := range rows {
+		fid, ok := jsonInt64(row["finding_id"])
+		if !ok {
+			return errors.New("invalid archived evidence finding id")
+		}
+		sid, _ := row["snapshot_id"].(string)
+		role, _ := row["role"].(string)
+		if !allowed[sid] || !ValidTrafficRole(role) {
+			return errors.New("invalid archived evidence binding")
+		}
+		var owned bool
+		if err = tx.QueryRow(`SELECT EXISTS(SELECT 1 FROM findings WHERE id=$1 AND task_id=$2)`, fid, snapshot.TaskID).Scan(&owned); err != nil {
+			return err
+		}
+		if !owned {
+			return fmt.Errorf("evidence references finding outside archived task: %d", fid)
+		}
+	}
+	if len(rows) > 0 {
+		raw, _ := json.Marshal(rows)
+		if _, err = tx.Exec(`INSERT INTO finding_traffic_bindings SELECT * FROM json_populate_recordset(NULL::finding_traffic_bindings,$1::json)`, string(raw)); err != nil {
+			return err
+		}
+		_, err = tx.Exec(`UPDATE traffic_evidence_snapshots s SET unreferenced_at=NULL WHERE EXISTS(SELECT 1 FROM finding_traffic_bindings b WHERE b.snapshot_id=s.id)`)
+	}
+	return err
+}
+
+func normalizeArchivedFindingVersions(raw json.RawMessage) (json.RawMessage, error) {
+	rows, err := decodeArchiveRows(raw)
+	if err != nil {
+		return nil, err
+	}
+	for _, row := range rows {
+		for _, key := range []string{"evidence_version", "report_evidence_version"} {
+			if row[key] == nil {
+				row[key] = 0
+			}
+		}
+	}
+	return json.Marshal(rows)
+}

db/finding_traffic_archive_test.go

@@ -0,0 +1,87 @@
+package db
+
+import (
+	"encoding/json"
+	"fmt"
+	"testing"
+)
+
+func TestFindingEvidenceLockNamespace(t *testing.T) {
+	for _, reserved := range []int64{7337741001, 7337741002, 7337741003} {
+		if findingEvidenceLockKey == reserved {
+			t.Fatal("evidence lock collides with migration/test/company lock")
+		}
+	}
+}
+
+func TestFindingTrafficLegacyArchiveDefaults(t *testing.T) {
+	d, err := Open(testDSN(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer d.Close()
+	if err = d.EnsureLLMRecordsTable(); err != nil {
+		t.Fatal(err)
+	}
+	if err = d.EnsureLLMUsageTable(); err != nil {
+		t.Fatal(err)
+	}
+	for _, version := range []int{1, 2} {
+		t.Run(fmt.Sprint(version), func(t *testing.T) {
+			task, err := d.CreateTask("legacy archive evidence defaults", "fixture", nil, 0, 0)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer func() { d.Exec(`DELETE FROM task_archives WHERE task_id=$1`, task.ID); d.DeleteTask(task.ID) }()
+			f, err := d.Exploration(task.ExplorationID).RecordFinding(t.Context(), RecordFindingInput{TaskID: task.ID, ExplorationID: task.ExplorationID, Summary: "legacy", Severity: "low"})
+			if err != nil {
+				t.Fatal(err)
+			}
+			if err = d.SetPaused(task.ID, true); err != nil {
+				t.Fatal(err)
+			}
+			archive, err := d.QueueTaskArchive(task.ID)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if _, err = d.ClaimTaskArchiveJob(t.Context()); err != nil {
+				t.Fatal(err)
+			}
+			snapshot, err := d.SnapshotTaskArchive(task.ID)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if err = d.CompleteTaskArchive(archive.ID, snapshot, "/tmp/legacy-test.tar.zst", "fixture", 1, 1); err != nil {
+				t.Fatal(err)
+			}
+			snapshot.FormatVersion = version
+			delete(snapshot.Tables, "finding_traffic_bindings")
+			delete(snapshot.Tables, "traffic_evidence_snapshots")
+			var rows []map[string]any
+			if err = json.Unmarshal(snapshot.Tables["findings"], &rows); err != nil {
+				t.Fatal(err)
+			}
+			for _, row := range rows {
+				delete(row, "evidence_version")
+				delete(row, "report_evidence_version")
+			}
+			snapshot.Tables["findings"], _ = json.Marshal(rows)
+			if _, err = d.QueueTaskArchiveRestore(archive.ID); err != nil {
+				t.Fatal(err)
+			}
+			if _, err = d.ClaimTaskArchiveJob(t.Context()); err != nil {
+				t.Fatal(err)
+			}
+			if _, err = d.RestoreTaskArchive(archive.ID, snapshot, 0); err != nil {
+				t.Fatal(err)
+			}
+			got, err := d.GetFindingTraffic(t.Context(), f.FindingID)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if got.Version != 0 || got.ReportVersion != 0 || len(got.Bindings) != 0 {
+				t.Fatal(got)
+			}
+		})
+	}
+}

db/findings.go

@@ -1,6 +1,7 @@
 package db
 
 import (
+	"context"
 	"database/sql"
 	"encoding/json"
 	"errors"
@@ -14,6 +15,11 @@ import (
 // DBFinding is a row in the standalone findings table. It persists across task
 // deletion unless the caller explicitly requests related finding cleanup.
 type DBFinding struct {
+	TrafficCount          int
+	EvidenceVersion       int64
+	ReportEvidenceVersion int64
+	TrafficBindings       []FindingTrafficBinding // populated only for export
+
 	ID              int64
 	TaskID          *int64
 	NodeID          *int64
@@ -97,7 +103,8 @@ func (d *DB) AddFinding(taskID, nodeID int64, vulnclass, name, severity, summary
 // list query selects, so scanFinding stays in sync across callers.
 const findingSelectCols = `f.id, f.task_id, f.node_id, f.vulnclass, COALESCE(f.name, ''), f.severity, f.summary,
 	       f.evidence, f.worker, f.asset_ids, COALESCE(f.status, 'pending'), f.created_at,
-	       COALESCE(t.description, '') AS task_description`
+	       COALESCE(t.description, '') AS task_description, f.evidence_version, f.report_evidence_version,
+ (SELECT count(*) FROM finding_traffic_bindings b WHERE b.finding_id=f.id)`
 
 // scanFindings materializes rows selected via findingSelectCols.
 func scanFindings(rows interface {
@@ -110,7 +117,7 @@ func scanFindings(rows interface {
 		f := &DBFinding{}
 		var aidsJSON string
 		if err := rows.Scan(&f.ID, &f.TaskID, &f.NodeID, &f.VulnClass, &f.Name, &f.Severity,
-			&f.Summary, &f.Evidence, &f.Worker, &aidsJSON, &f.Status, &f.CreatedAt, &f.TaskDescription); err != nil {
+			&f.Summary, &f.Evidence, &f.Worker, &aidsJSON, &f.Status, &f.CreatedAt, &f.TaskDescription, &f.EvidenceVersion, &f.ReportEvidenceVersion, &f.TrafficCount); err != nil {
 			return nil, err
 		}
 		_ = json.Unmarshal([]byte(aidsJSON), &f.AssetIDs)
@@ -503,7 +510,7 @@ func (d *DB) ListFindingsForExport(f FindingFilter, ids []int64) ([]*DBFinding,
 		var aidsJSON string
 		if err := rows.Scan(&f.ID, &f.TaskID, &f.NodeID, &f.VulnClass, &f.Name, &f.Severity,
 			&f.Summary, &f.Evidence, &f.Worker, &aidsJSON, &f.Status, &f.CreatedAt,
-			&f.TaskDescription, &f.Report); err != nil {
+			&f.TaskDescription, &f.EvidenceVersion, &f.ReportEvidenceVersion, &f.TrafficCount, &f.Report); err != nil {
 			return nil, err
 		}
 		_ = json.Unmarshal([]byte(aidsJSON), &f.AssetIDs)
@@ -631,7 +638,7 @@ func (d *DB) GetFinding(id int64) (*DBFinding, error) {
 		WHERE f.id = $1`, id).Scan(
 		&f.ID, &f.TaskID, &f.NodeID, &f.VulnClass, &f.Name, &f.Severity,
 		&f.Summary, &f.Evidence, &f.Worker, &aidsJSON, &f.Status, &f.CreatedAt,
-		&f.TaskDescription, &f.Report)
+		&f.TaskDescription, &f.EvidenceVersion, &f.ReportEvidenceVersion, &f.TrafficCount, &f.Report)
 	if err == sql.ErrNoRows {
 		return nil, nil
 	}
@@ -647,20 +654,27 @@ func (d *DB) GetFinding(id int64) (*DBFinding, error) {
 // list, the per-task 发现 Tab, and the exploration graph alike. Deleting the node
 // cascades its edges + node_assets and nulls any activity referencing it. Returns
 // rows affected (0 = no finding with that id).
-func (d *DB) DeleteFinding(id int64) (int64, error) {
-	var nodeID *int64
-	err := d.QueryRow(`DELETE FROM findings WHERE id=$1 RETURNING node_id`, id).Scan(&nodeID)
-	if err == sql.ErrNoRows {
-		return 0, nil
-	}
-	if err != nil {
-		return 0, err
-	}
-	if nodeID != nil {
-		// best-effort: the finding is already gone; a stray node must not fail the op.
-		_, _ = d.Exec(`DELETE FROM exploration_nodes WHERE id=$1 AND kind='finding'`, *nodeID)
-	}
-	return 1, nil
+func (d *DB) DeleteFinding(id int64) (n int64, err error) {
+	err = d.WithEvidenceTx(context.Background(), func(tx *sql.Tx) error {
+		if err := LockFindingEvidenceTx(tx, id, nil); err != nil {
+			if errors.Is(err, ErrFindingNotFound) {
+				return nil
+			}
+			return err
+		}
+		var nodeID sql.NullInt64
+		if err := tx.QueryRow(`DELETE FROM findings WHERE id=$1 RETURNING node_id`, id).Scan(&nodeID); err != nil {
+			return err
+		}
+		if nodeID.Valid {
+			if _, err := tx.Exec(`DELETE FROM exploration_nodes WHERE id=$1 AND kind='finding'`, nodeID.Int64); err != nil {
+				return err
+			}
+		}
+		n = 1
+		return nil
+	})
+	return
 }
 
 // DeleteFindingsByTask removes all findings rows of a task. The originating
@@ -687,11 +701,7 @@ func (d *DB) SetFindingStatus(id int64, status string) (int64, error) {
 // whose node_id matches — report_finding returns that node id, so an agent tool
 // can address the finding it just created. Returns rows affected (0 when no row).
 func (d *DB) SetFindingReportByNodeID(nodeID int64, report string) (int64, error) {
-	res, err := d.Exec(`UPDATE findings SET report=$1 WHERE node_id=$2`, report, nodeID)
-	if err != nil {
-		return 0, err
-	}
-	return res.RowsAffected()
+	return d.SetFindingReportVersionByNodeID(context.Background(), nodeID, report, nil)
 }
 
 // setFindingCol updates one text column on the standalone finding row AND mirrors
@@ -737,6 +747,8 @@ func (d *DB) SetFindingVulnClass(id int64, vulnclass string) (int64, error) {
 // FindingMeta is the standalone-row data (id, triage state, anchored assets) the
 // per-task view grafts onto its exploration-node findings.
 type FindingMeta struct {
+	TrafficCount int
+
 	ID       int64
 	Status   string
 	AssetIDs []int64
@@ -757,7 +769,7 @@ func (d *DB) FindingMetaByNodeID(taskID int64) (map[int64]FindingMeta, error) {
 	if taskID <= 0 {
 		return out, nil
 	}
-	rows, err := d.Query(`SELECT node_id, id, COALESCE(status,'pending'), asset_ids FROM findings
+	rows, err := d.Query(`SELECT node_id, id, COALESCE(status,'pending'), asset_ids, (SELECT count(*) FROM finding_traffic_bindings b WHERE b.finding_id=findings.id) FROM findings
 		WHERE task_id=$1 AND node_id IS NOT NULL`, taskID)
 	if err != nil {
 		return out, err
@@ -767,7 +779,7 @@ func (d *DB) FindingMetaByNodeID(taskID int64) (map[int64]FindingMeta, error) {
 		var nid int64
 		var m FindingMeta
 		var aidsJSON string
-		if err := rows.Scan(&nid, &m.ID, &m.Status, &aidsJSON); err != nil {
+		if err := rows.Scan(&nid, &m.ID, &m.Status, &aidsJSON, &m.TrafficCount); err != nil {
 			return out, err
 		}
 		_ = json.Unmarshal([]byte(aidsJSON), &m.AssetIDs)

db/schema.sql

@@ -1058,6 +1058,39 @@ CREATE INDEX IF NOT EXISTS idx_findings_status ON findings(status, created_at DE
 -- 「按资产」视图靠 asset_ids @> '[<id>]' 反查发现,没有这个 GIN 索引就是全表扫。
 CREATE INDEX IF NOT EXISTS idx_findings_asset_ids ON findings USING GIN(asset_ids jsonb_path_ops);
 
+ALTER TABLE findings ADD COLUMN IF NOT EXISTS evidence_version BIGINT NOT NULL DEFAULT 0;
+ALTER TABLE findings ADD COLUMN IF NOT EXISTS report_evidence_version BIGINT NOT NULL DEFAULT 0;
+
+CREATE TABLE IF NOT EXISTS traffic_evidence_snapshots (
+    id TEXT PRIMARY KEY,
+    source_traffic_id TEXT NOT NULL,
+    captured_at BIGINT NOT NULL,
+    url TEXT NOT NULL,
+    method TEXT NOT NULL,
+    status INTEGER NOT NULL,
+    content_type TEXT NOT NULL DEFAULT '',
+    req_head TEXT NOT NULL,
+    resp_head TEXT NOT NULL,
+    req_hash TEXT NOT NULL,
+    resp_hash TEXT NOT NULL,
+    req_len BIGINT NOT NULL,
+    resp_len BIGINT NOT NULL,
+    unreferenced_at TIMESTAMPTZ,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+CREATE TABLE IF NOT EXISTS finding_traffic_bindings (
+    id BIGSERIAL PRIMARY KEY,
+    finding_id BIGINT NOT NULL REFERENCES findings(id) ON DELETE CASCADE,
+    snapshot_id TEXT NOT NULL REFERENCES traffic_evidence_snapshots(id),
+    role TEXT NOT NULL DEFAULT 'supporting',
+    note TEXT NOT NULL DEFAULT '',
+    position INTEGER NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    UNIQUE(finding_id, snapshot_id)
+);
+CREATE INDEX IF NOT EXISTS idx_finding_traffic_order ON finding_traffic_bindings(finding_id, position, id);
+CREATE INDEX IF NOT EXISTS idx_finding_traffic_snapshot ON finding_traffic_bindings(snapshot_id);
+
 -- =====================================================================
 -- M. 后端日志持久化
 -- =====================================================================

db/task_archives.go

@@ -16,7 +16,7 @@ import (
 )
 
 const (
-	TaskArchiveFormatVersion       = 2
+	TaskArchiveFormatVersion       = 3
 	TaskArchiveLegacyFormatVersion = 1
 	TaskArchiveLLMRecordsPath      = "database/llm_records.ndjson"
 )
@@ -587,6 +587,8 @@ func (d *DB) snapshotTaskArchive(taskID int64, llmRecords io.Writer) (*TaskArchi
 		{"task_llm_profiles", `SELECT * FROM task_llm_profiles WHERE task_id=$1 ORDER BY position`, []any{taskID}},
 		{"task_scope", `SELECT * FROM task_scope WHERE task_id=$1 ORDER BY id`, []any{taskID}},
 		{"findings", `SELECT * FROM findings WHERE task_id=$1 ORDER BY id`, []any{taskID}},
+		{"finding_traffic_bindings", `SELECT b.* FROM finding_traffic_bindings b JOIN findings f ON f.id=b.finding_id WHERE f.task_id=$1 ORDER BY b.finding_id,b.position,b.id`, []any{taskID}},
+		{"traffic_evidence_snapshots", `SELECT s.* FROM traffic_evidence_snapshots s WHERE EXISTS(SELECT 1 FROM finding_traffic_bindings b JOIN findings f ON f.id=b.finding_id WHERE b.snapshot_id=s.id AND f.task_id=$1) ORDER BY s.id`, []any{taskID}},
 		{"llm_records", `SELECT * FROM llm_records WHERE COALESCE(task_id,'')=$1 ORDER BY id`, []any{strconv.FormatInt(taskID, 10)}},
 		{"llm_usage", `SELECT * FROM llm_usage WHERE COALESCE(task_id,'')=$1 OR exploration_id=$2 ORDER BY id`, []any{strconv.FormatInt(taskID, 10), expID}},
 		{"skill_usage", `SELECT * FROM skill_usage WHERE task_id=$1 OR exploration_id=$2 ORDER BY id`, []any{taskID, expID}},

db/task_archives_restore.go

@@ -233,6 +233,10 @@ WHERE archive.id=$1 FOR UPDATE OF archive,task`, archiveID).Scan(&taskID, &expID
 	if err != nil {
 		return nil, err
 	}
+	remappedTables["findings"], err = normalizeArchivedFindingVersions(remappedTables["findings"])
+	if err != nil {
+		return nil, err
+	}
 	// Insert graph rows in foreign-key order. The archived stub has no graph rows,
 	// so an ID conflict signals external corruption and must stop the restore.
 	for _, table := range []string{"exploration_nodes", "exploration_edges", "exploration_anchors", "task_constraints", "activity"} {
@@ -260,6 +264,9 @@ WHERE archive.id=$1 FOR UPDATE OF archive,task`, archiveID).Scan(&taskID, &expID
 			return nil, fmt.Errorf("restore %s: %w", table, err)
 		}
 	}
+	if err := restoreFindingTrafficTx(tx, snapshot); err != nil {
+		return nil, fmt.Errorf("restore finding traffic: %w", err)
+	}
 	if streamedLLMRecords != "" {
 		count, err := insertArchiveJSONSequenceRows(tx, "llm_records", llmRecords)
 		if err != nil {