Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 7 additions & 2 deletions src/vizier/services/adaptive_export/cmd/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -363,6 +363,10 @@ func main() {
MaxInflightQueriesGlobal: intEnvOrZero(envMaxInflightQueriesGlobal),
EmptyResultSkipAfterN: intEnvOrZero(envEmptyResultSkipAfterN),
EmptyResultSkipTTL: durEnvOrZero(envEmptyResultSkipTTLSec, time.Second),
// Source the kubescape process-tree index from the same ClickHouse the
// trigger reads, so the operator can reattribute DNS rows pixie leaves
// unattributed (short-lived resolver pids). See pixie#80.
ProcessTreeIndex: trg.ProcessTreeIndex,
}
if streamingMode {
// Route through the non-blocking notifier — handle() returns
Expand Down Expand Up @@ -576,9 +580,10 @@ func main() {
// Wire the controller as the /query runner: dx OrderQuery → one-shot pixie
// capture written to forensic_db (write⊇read; entlein/dx#93). When the
// operator-side querier is disabled (no PushPixieTables), OrderQuery returns
// an error and /query 502s — start/stop + dx_attack_graph still work.
// an error and /query 502s — start/stop + dx_evidence_graph still work.
ctrlSrv := control.New(activeSet, ctl)
ctrlSrv.SetGraphWriter(applier) // dx_attack_graph ingest → ClickHouse
ctrlSrv.SetGraphWriter(applier) // dx_evidence_graph ingest → ClickHouse
ctrlSrv.SetManifestWriter(applier) // dx_evidence_manifest ingest → ClickHouse
// Bearer-JWT auth on the control surface (CodeRabbit: protect control
// endpoints). Same shared lib + signing key the broker/PEM use — dx
// attaches the service JWT it already mints. Default-OFF so this can
Expand Down
32 changes: 25 additions & 7 deletions src/vizier/services/adaptive_export/internal/clickhouse/apply.go
Original file line number Diff line number Diff line change
Expand Up @@ -67,9 +67,13 @@ var OperatorOwnedTables = []string{
// dx_evidence_graph UI (px.DataFrame clickhouse_dsn) has a real,
// globally-registered table to read. dx emits edges, AE persists.
// Not a pixie socket_tracer table → not in PixieTables().
"dx_attack_graph",
// rule-ins-only VIEW over dx_attack_graph; created AFTER it (depends on it).
"dx_attack_graph_malicious",
"dx_evidence_graph",
// rule-ins-only VIEW over dx_evidence_graph; created AFTER it (depends on it).
"dx_evidence_graph_malignant",
// dx §9 completeness manifest — one row per verdict naming the evidence rows
// dx consulted. Created on boot so POST /dx/evidence_manifest has a target.
// Independent of dx_evidence_graph. Not a pixie table → not in PixieTables().
"dx_evidence_manifest",
}

// Applier applies operator-owned DDL to a ClickHouse cluster over the
Expand Down Expand Up @@ -108,15 +112,29 @@ func (a *Applier) Apply(ctx context.Context) error {
return nil
}

// WriteAttackGraph inserts dx evidence-graph edges into
// forensic_db.dx_attack_graph. jsonEachRow is newline-delimited JSON objects
// WriteEvidenceGraph inserts dx evidence-graph edges into
// forensic_db.dx_evidence_graph. jsonEachRow is newline-delimited JSON objects
// whose keys are the column names (JSONEachRow; unknown keys are skipped,
// missing columns default). No-op on empty input.
func (a *Applier) WriteAttackGraph(ctx context.Context, jsonEachRow []byte) error {
func (a *Applier) WriteEvidenceGraph(ctx context.Context, jsonEachRow []byte) error {
if len(jsonEachRow) == 0 {
return nil
}
_, err := a.c.Insert(ctx, "INSERT INTO forensic_db.dx_attack_graph FORMAT JSONEachRow",
_, err := a.c.Insert(ctx, "INSERT INTO forensic_db.dx_evidence_graph FORMAT JSONEachRow",
jsonEachRow, chhttp.InsertOptions{})
return err
}

// WriteEvidenceManifest inserts one dx §9 completeness manifest (per verdict)
// into forensic_db.dx_evidence_manifest. jsonEachRow is a single JSONEachRow
// object whose keys are the column names; the control handler pre-renders the
// nested collections (case_window/findings/orders/seeds/chain) as JSON text so
// the insert is ClickHouse-version independent. No-op on empty input.
func (a *Applier) WriteEvidenceManifest(ctx context.Context, jsonEachRow []byte) error {
if len(jsonEachRow) == 0 {
return nil
}
_, err := a.c.Insert(ctx, "INSERT INTO forensic_db.dx_evidence_manifest FORMAT JSONEachRow",
jsonEachRow, chhttp.InsertOptions{})
return err
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,7 @@ func TestApply_ExecutesEveryOperatorOwnedTable(t *testing.T) {
}
// Spot-check that the SECOND call is for the first OperatorOwnedTables entry,
// and that the LAST call is for the last OperatorOwnedTables entry (robust to
// new operator-owned tables being appended, e.g. dx_attack_graph).
// new operator-owned tables being appended, e.g. dx_evidence_graph).
if !strings.Contains(bodies[1], "forensic_db."+OperatorOwnedTables[0]) {
t.Fatalf("second DDL not for %s; got: %s", OperatorOwnedTables[0], bodies[1])
}
Expand Down Expand Up @@ -228,7 +228,7 @@ func TestOperatorOwnedTables_DoesNotIncludeKubescape(t *testing.T) {
// plugin can auto-DDL them with the wrong schema), then the operator's
// own write targets in declared order.
func TestOperatorOwnedTables_TrailingOperatorTables(t *testing.T) {
want := []string{"adaptive_attribution", "trigger_watermark", "ae_reconcile", "dx_attack_graph", "dx_attack_graph_malicious"}
want := []string{"adaptive_attribution", "trigger_watermark", "ae_reconcile", "dx_evidence_graph", "dx_evidence_graph_malignant", "dx_evidence_manifest"}
got := OperatorOwnedTables[len(OperatorOwnedTables)-len(want):]
for i, w := range want {
if got[i] != w {
Expand Down
12 changes: 8 additions & 4 deletions src/vizier/services/adaptive_export/internal/clickhouse/ddl.go
Original file line number Diff line number Diff line change
Expand Up @@ -66,11 +66,15 @@ var KnownTables = []string{
"ae_reconcile",
// operator-owned dx evidence-graph edge list (read by the Pixie
// dx_evidence_graph UI via clickhouse_dsn). NOT a pixie table.
"dx_attack_graph",
// rule-ins-only VIEW over dx_attack_graph (condition != ''); the
"dx_evidence_graph",
// rule-ins-only VIEW over dx_evidence_graph (condition != ''); the
// dx_evidence_graph UI reads this by default so benign rows are filtered
// in ClickHouse, not pulled. Must follow dx_attack_graph (depends on it).
"dx_attack_graph_malicious",
// in ClickHouse, not pulled. Must follow dx_evidence_graph (depends on it).
"dx_evidence_graph_malignant",
// operator-owned dx §9 completeness manifest — one row per verdict naming
// the evidence rows dx consulted (POST /dx/evidence_manifest). NOT a pixie
// table. Independent of dx_evidence_graph.
"dx_evidence_manifest",
}

// ErrUnknownTable is returned by DDL / Columns when asked for a table
Expand Down
41 changes: 36 additions & 5 deletions src/vizier/services/adaptive_export/internal/clickhouse/schema.sql
Original file line number Diff line number Diff line change
Expand Up @@ -487,7 +487,7 @@ CREATE TABLE IF NOT EXISTS forensic_db.ae_reconcile (
-- unbounded storage (CodeRabbit). 30d matches the pixie observation tables.
TTL toDateTime(ts) + INTERVAL 30 DAY DELETE;

-- dx_attack_graph — dx evidence-graph edge list: one row per directed hop of an
-- dx_evidence_graph — dx evidence-graph edge list: one row per directed hop of an
-- investigation (delivery/egress/execution/exfil/pivot), read by the Pixie
-- dx_evidence_graph UI via px.DataFrame(clickhouse_dsn=...). Operator-owned
-- (dx emits the edges, AE persists them); NOT a pixie socket_tracer table.
Expand All @@ -498,7 +498,7 @@ CREATE TABLE IF NOT EXISTS forensic_db.ae_reconcile (
-- event_time". Same convention as kubescape_logs. event_time is nanos, so the
-- partition/TTL use fromUnixTimestamp64Nano (toDateTime would read ns as seconds
-- → year ~58e9 → broken partitions; see the soc#225 fix).
CREATE TABLE IF NOT EXISTS forensic_db.dx_attack_graph (
CREATE TABLE IF NOT EXISTS forensic_db.dx_evidence_graph (
investigation_id String,
event_time UInt64,
hostname String,
Expand Down Expand Up @@ -526,7 +526,38 @@ CREATE TABLE IF NOT EXISTS forensic_db.dx_attack_graph (
TTL toDateTime(fromUnixTimestamp64Nano(event_time)) + INTERVAL 30 DAY DELETE
SETTINGS index_granularity = 8192;

-- dx_attack_graph_malicious — rule-ins-only view (condition != '') the
-- dx_evidence_graph_malignant — rule-ins-only view (condition != '') the
-- dx_evidence_graph UI reads by default so benign rows stay in ClickHouse.
CREATE VIEW IF NOT EXISTS forensic_db.dx_attack_graph_malicious AS
SELECT * FROM forensic_db.dx_attack_graph WHERE `condition` != '';
CREATE VIEW IF NOT EXISTS forensic_db.dx_evidence_graph_malignant AS
SELECT * FROM forensic_db.dx_evidence_graph WHERE `condition` != '';

-- dx_evidence_manifest — the §9 completeness contract: one row per verdict
-- (ruled_in | metastasis), naming the evidence rows dx consulted so the
-- validator can join them against what AE persisted (write⊇read, checkable).
-- Operator-owned (dx emits the manifest via POST /dx/evidence_manifest, AE
-- persists it); NOT a pixie table. Column names are the manifest.Manifest
-- JSON tags (dx internal/manifest). Same event_time (unix NANOSECONDS) +
-- hostname read-path convention as dx_evidence_graph so it is px-readable.
-- The nested collections (case_window/findings/orders/seeds/chain) are stored
-- as JSON text in String columns; the control handler pre-renders them so the
-- JSONEachRow insert is ClickHouse-version independent.
CREATE TABLE IF NOT EXISTS forensic_db.dx_evidence_manifest (
investigation_id String,
event_time UInt64,
hostname String,
`condition` String,
verdict String,
confidence Float64,
posterior Float64,
catalog_version String,
case_window String,
findings String,
orders String,
seeds String,
chain String,
evidence_hash String
) ENGINE = MergeTree()
ORDER BY (event_time, hostname)
PARTITION BY toYYYYMM(fromUnixTimestamp64Nano(event_time))
TTL toDateTime(fromUnixTimestamp64Nano(event_time)) + INTERVAL 30 DAY DELETE
SETTINGS index_granularity = 8192;
114 changes: 100 additions & 14 deletions src/vizier/services/adaptive_export/internal/control/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -54,18 +54,25 @@ type queryRunner interface {
}

// graphWriter persists dx evidence-graph edges (newline-delimited JSON,
// JSONEachRow) to forensic_db.dx_attack_graph. nil → /dx/attack_graph 501s.
// JSONEachRow) to forensic_db.dx_evidence_graph. nil → /dx/evidence_graph 501s.
type graphWriter interface {
WriteAttackGraph(ctx context.Context, jsonEachRow []byte) error
WriteEvidenceGraph(ctx context.Context, jsonEachRow []byte) error
}

// manifestWriter persists one dx §9 completeness manifest per verdict
// (JSONEachRow) to forensic_db.dx_evidence_manifest. nil → /dx/evidence_manifest 501s.
type manifestWriter interface {
WriteEvidenceManifest(ctx context.Context, jsonEachRow []byte) error
}

// Server is the control HTTP surface.
type Server struct {
set exporter
runner queryRunner // may be nil; /query then returns 501
graph graphWriter // may be nil; /dx/attack_graph then returns 501
mux *http.ServeMux
verify func(bearer string) error // nil → auth disabled; set via SetAuth
set exporter
runner queryRunner // may be nil; /query then returns 501
graph graphWriter // may be nil; /dx/evidence_graph then returns 501
manifest manifestWriter // may be nil; /dx/evidence_manifest then returns 501
mux *http.ServeMux
verify func(bearer string) error // nil → auth disabled; set via SetAuth
}

// New builds the control server. runner may be nil for deployments that
Expand All @@ -76,13 +83,17 @@ func New(set exporter, runner queryRunner) *Server {
s.mux.HandleFunc("/export/start", s.handleStart)
s.mux.HandleFunc("/export/stop", s.handleStop)
s.mux.HandleFunc("/query", s.handleQuery)
s.mux.HandleFunc("/dx/attack_graph", s.handleDXAttackGraph)
s.mux.HandleFunc("/dx/evidence_graph", s.handleDXEvidenceGraph)
s.mux.HandleFunc("/dx/evidence_manifest", s.handleDXEvidenceManifest)
return s
}

// SetGraphWriter wires the dx_attack_graph sink.
// SetGraphWriter wires the dx_evidence_graph sink.
func (s *Server) SetGraphWriter(g graphWriter) { s.graph = g }

// SetManifestWriter wires the dx_evidence_manifest sink.
func (s *Server) SetManifestWriter(m manifestWriter) { s.manifest = m }

// SetAuth turns on bearer-JWT auth for the control surface, verified with the
// SAME shared lib + signing key the vizier broker/PEM use (px.dev/pixie/src/
// shared/services/utils). dx already mints a service JWT (GenerateJWTForService,
Expand Down Expand Up @@ -115,9 +126,9 @@ func (s *Server) Handler() http.Handler {
})
}

// handleDXAttackGraph ingests a JSON array of dx evidence-graph edges and writes
// them to forensic_db.dx_attack_graph (as JSONEachRow).
func (s *Server) handleDXAttackGraph(w http.ResponseWriter, r *http.Request) {
// handleDXEvidenceGraph ingests a JSON array of dx evidence-graph edges and writes
// them to forensic_db.dx_evidence_graph (as JSONEachRow).
func (s *Server) handleDXEvidenceGraph(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
w.WriteHeader(http.StatusMethodNotAllowed)
return
Expand All @@ -140,13 +151,88 @@ func (s *Server) handleDXAttackGraph(w http.ResponseWriter, r *http.Request) {
buf.Write(e)
buf.WriteByte('\n')
}
if err := s.graph.WriteAttackGraph(r.Context(), buf.Bytes()); err != nil {
if err := s.graph.WriteEvidenceGraph(r.Context(), buf.Bytes()); err != nil {
w.WriteHeader(http.StatusBadGateway)
return
}
w.WriteHeader(http.StatusAccepted)
}

// dxManifest mirrors the wire shape of dx's manifest.Manifest (internal/manifest).
// Scalars map to typed forensic_db.dx_evidence_manifest columns; the nested
// collections are held as raw JSON and persisted as JSON text in String columns
// so the JSONEachRow insert is ClickHouse-version independent.
type dxManifest struct {
InvestigationID string `json:"investigation_id"`
EventTime int64 `json:"event_time"`
Hostname string `json:"hostname"`
Condition string `json:"condition"`
Verdict string `json:"verdict"`
Confidence float64 `json:"confidence"`
Posterior float64 `json:"posterior"`
CatalogVersion string `json:"catalog_version"`
CaseWindow json.RawMessage `json:"case_window"`
Findings json.RawMessage `json:"findings"`
Orders json.RawMessage `json:"orders"`
Seeds json.RawMessage `json:"seeds"`
Chain json.RawMessage `json:"chain"`
EvidenceHash string `json:"evidence_hash"`
}

// handleDXEvidenceManifest ingests ONE dx completeness manifest (per verdict)
// and writes it to forensic_db.dx_evidence_manifest as a single JSONEachRow
// row, with the nested collections rendered as JSON text.
func (s *Server) handleDXEvidenceManifest(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
w.WriteHeader(http.StatusMethodNotAllowed)
return
}
if s.manifest == nil {
w.WriteHeader(http.StatusNotImplemented)
return
}
var m dxManifest
if !decode(w, r, &m) {
w.WriteHeader(http.StatusBadRequest)
return
}
row := map[string]any{
"investigation_id": m.InvestigationID,
"event_time": m.EventTime,
"hostname": m.Hostname,
"condition": m.Condition,
"verdict": m.Verdict,
"confidence": m.Confidence,
"posterior": m.Posterior,
"catalog_version": m.CatalogVersion,
"case_window": jsonText(m.CaseWindow),
"findings": jsonText(m.Findings),
"orders": jsonText(m.Orders),
"seeds": jsonText(m.Seeds),
"chain": jsonText(m.Chain),
"evidence_hash": m.EvidenceHash,
}
line, err := json.Marshal(row)
if err != nil {
w.WriteHeader(http.StatusBadRequest)
return
}
if err := s.manifest.WriteEvidenceManifest(r.Context(), append(line, '\n')); err != nil {
w.WriteHeader(http.StatusBadGateway)
return
}
w.WriteHeader(http.StatusAccepted)
}

// jsonText renders a nested JSON value as compact text for a String column;
// nil/absent/null → "" so the column holds an empty string rather than "null".
func jsonText(raw json.RawMessage) string {
if len(raw) == 0 || string(raw) == "null" {
return ""
}
return string(raw)
}

// ── wire types ────────────────────────────────────────────────────────
type targetReq struct {
Namespace string `json:"namespace"`
Expand Down Expand Up @@ -175,7 +261,7 @@ func (t targetReq) target() anomaly.Target {
}

// maxControlBodyBytes caps a single control-surface request body. The
// largest legitimate payload we accept is /dx/attack_graph which is a
// largest legitimate payload we accept is /dx/evidence_graph which is a
// JSON array of pre-marshalled JSONEachRow lines — measured live the
// hottest dx rule-in pass fits in ~256 KiB. 4 MiB is well above that
// and below the per-pod memory headroom an oversized POST could
Expand Down
Loading
Loading