fix(import): send PCAP INSERT through node /exec (11.0.337)

src/coordinator/handlers/pcap_sip_import.go

@@ -168,7 +168,7 @@ func importPcapSIP(
 		if err != nil {
 			return err
 		}
-		if _, err := flight.QueryFirstConnected(ctx, sql); err != nil {
+		if err := flight.ExecFirstConnected(ctx, sql); err != nil {
 			return err
 		}
 		inserted += len(rows)

src/coordinator/handlers/pcap_sip_import_test.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/google/gopacket"
+	"github.com/sipcapture/homer-core/src/coordinator/sqlvalidator"
 	"github.com/sipcapture/homer-core/src/decoder"
 	"github.com/sipcapture/homer-core/src/pcapwriter"
 	"github.com/sipcapture/homer-core/src/storage/ducklake"
@@ -79,6 +80,9 @@ func TestPcapSyntheticInvitePipeline(t *testing.T) {
 	if !strings.Contains(sql, "314159") {
 		t.Fatalf("cseq not in sql: %s", sql)
 	}
+	if err := sqlvalidator.ValidateWriteSQL(sql); err != nil {
+		t.Fatalf("generated INSERT must pass write validator: %v\n%s", err, sql)
+	}
 }
 
 func TestPcapForceInviteIntoRegistrationTable(t *testing.T) {

src/coordinator/services/flight_service.go

@@ -301,6 +301,27 @@ func (s *FlightService) QueryFirstConnected(ctx context.Context, sql string) ([]
 	return nil, fmt.Errorf("no connected storage nodes")
 }
 
+// ExecFirstConnected runs write SQL (INSERT) on the first connected node via POST /exec.
+// /query stays read-only; this is the path for PCAP import and other coordinator writes.
+func (s *FlightService) ExecFirstConnected(ctx context.Context, sql string) error {
+	s.mu.RLock()
+	nodes := make([]config.NodeEndpoint, len(s.nodes))
+	copy(nodes, s.nodes)
+	connected := make(map[string]bool)
+	for k, v := range s.connected {
+		connected[k] = v
+	}
+	s.mu.RUnlock()
+
+	for _, node := range nodes {
+		if !connected[node.Name] {
+			continue
+		}
+		return s.execNode(ctx, node, sql)
+	}
+	return fmt.Errorf("no connected storage nodes")
+}
+
 // nodesForRange filters connected nodes to those whose cached timestamp range
 // may overlap the query window [fromNs, toNs]. A node is skipped only when its
 // cached range (with flush/health slack on max) provably does not overlap the
@@ -441,16 +462,25 @@ func (s *FlightService) QueryNode(ctx context.Context, nodeName string, sql stri
 	return s.queryNode(ctx, node, sql)
 }
 
-// queryNode executes a query on a single node via HTTP
+// queryNode executes a query on a single node via HTTP POST /query
 func (s *FlightService) queryNode(ctx context.Context, node config.NodeEndpoint, sql string) ([]map[string]interface{}, error) {
+	return s.postNodeSQL(ctx, node, "/query", sql)
+}
+
+func (s *FlightService) execNode(ctx context.Context, node config.NodeEndpoint, sql string) error {
+	_, err := s.postNodeSQL(ctx, node, "/exec", sql)
+	return err
+}
+
+func (s *FlightService) postNodeSQL(ctx context.Context, node config.NodeEndpoint, path, sql string) ([]map[string]interface{}, error) {
 	// Per-query timeout; context.WithTimeout keeps the parent deadline when
 	// the caller's is sooner.
 	ctx, cancel := context.WithTimeout(ctx, s.queryTimeout)
 	defer cancel()
 
 	// Node HTTP API runs on FlightServer.Port + 1
 	httpPort := node.Port + 1
-	url := fmt.Sprintf("http://%s:%d/query", node.Host, httpPort)
+	url := fmt.Sprintf("http://%s:%d%s", node.Host, httpPort, path)
 
 	reqBody, err := json.Marshal(queryRequest{SQL: sql})
 	if err != nil {

src/coordinator/services/flight_service_test.go

@@ -74,6 +74,45 @@ func TestFlightServiceQuerySendsBearerToken(t *testing.T) {
 	}
 }
 
+func TestFlightServiceExecFirstConnectedPostsExec(t *testing.T) {
+	var gotPath, gotAuth, gotSQL string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		gotPath = r.URL.Path
+		gotAuth = r.Header.Get("Authorization")
+		var req queryRequest
+		_ = json.NewDecoder(r.Body).Decode(&req)
+		gotSQL = req.SQL
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write([]byte(`{"success":true,"data":[],"count":0}`))
+	}))
+	defer srv.Close()
+
+	u, err := url.Parse(srv.URL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	host := u.Hostname()
+	port, _ := strconv.Atoi(u.Port())
+	svc := NewFlightService([]config.NodeEndpoint{
+		{Name: "local", Host: host, Port: port - 1, Token: "node-secret"},
+	}, time.Second, false)
+	svc.connected["local"] = true
+
+	sql := "INSERT INTO homer_lake.main.hep_proto_1_call (id) VALUES (1)"
+	if err := svc.ExecFirstConnected(context.Background(), sql); err != nil {
+		t.Fatal(err)
+	}
+	if gotPath != "/exec" {
+		t.Fatalf("path=%q want /exec", gotPath)
+	}
+	if gotAuth != "Bearer node-secret" {
+		t.Fatalf("Authorization=%q", gotAuth)
+	}
+	if gotSQL != sql {
+		t.Fatalf("sql=%q", gotSQL)
+	}
+}
+
 func TestFetchNodeRangeParsesStats(t *testing.T) {
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		if r.URL.Path != "/metadata/stats" {

src/coordinator/sqlvalidator/validator.go

@@ -4,6 +4,7 @@ package sqlvalidator
 
 import (
 	"fmt"
+	"regexp"
 	"strings"
 	"unicode"
 )
@@ -210,8 +211,8 @@ var blockedDML = map[string]bool{
 	"ATTACH":   true,
 	"DETACH":   true,
 	"LOAD":     true,
-	"INSTALL": true,
-	"SET":     true,
+	"INSTALL":  true,
+	"SET":      true,
 }
 
 // blockedFunctions are DuckDB functions that access the filesystem or network.
@@ -260,54 +261,164 @@ var allowedStatementStarts = map[string]bool{
 // ValidateRawSQL validates a full SQL statement for safety.
 // Returns nil if the SQL is safe to execute, or an error describing the violation.
 func ValidateRawSQL(sql string) error {
-	trimmed := strings.TrimSpace(sql)
-	if trimmed == "" {
-		return fmt.Errorf("empty SQL query")
-	}
-
-	tokens := tokenize(trimmed)
-	if len(tokens) == 0 {
-		return fmt.Errorf("empty SQL query after parsing")
+	tokens, err := tokensForValidation(sql)
+	if err != nil {
+		return err
 	}
 
-	// 1. Must start with an allowed statement type
 	first := tokens[0]
 	if first.kind != tkIdent || !allowedStatementStarts[first.upper] {
 		return fmt.Errorf("query must start with SELECT, WITH, SHOW, DESCRIBE, EXPLAIN, or PRAGMA (got %q)", first.value)
 	}
 
-	// 2. No semicolons (prevents statement stacking)
+	return rejectSQLHazards(tokens, nil)
+}
+
+// hepProtoTableName is the only table name allowed on POST /exec (PCAP import).
+// Matches LakeTableFQN: <catalog>.main.hep_proto_<proto>_<suffix>.
+var hepProtoTableName = regexp.MustCompile(`(?i)^hep_proto_[0-9]+_[a-z][a-z0-9_]*$`)
+
+var blockedWriteCatalogs = map[string]bool{
+	"MEMORY":             true,
+	"TEMP":               true,
+	"SYSTEM":             true,
+	"INFORMATION_SCHEMA": true,
+	"PG_CATALOG":         true,
+}
+
+// ValidateWriteSQL validates coordinator-generated writes (PCAP import INSERT).
+// Only INSERT INTO <catalog>.main.hep_proto_* (...) VALUES ... is allowed.
+// SELECT/FROM, system catalogs, quoted identifiers, and DML/filesystem hazards stay blocked.
+func ValidateWriteSQL(sql string) error {
+	tokens, err := tokensForValidation(sql)
+	if err != nil {
+		return err
+	}
+
+	if err := requireInsertIntoHepProtoValues(tokens); err != nil {
+		return err
+	}
+
+	return rejectSQLHazards(tokens, map[string]bool{"INSERT": true})
+}
+
+func requireInsertIntoHepProtoValues(tokens []token) error {
 	for _, tok := range tokens {
 		if tok.kind == tkSemi {
 			return fmt.Errorf("semicolons are not allowed (prevents multi-statement injection)")
 		}
+		if tok.kind == tkIdent && strings.HasPrefix(tok.value, "\"") {
+			return fmt.Errorf("quoted identifiers are not allowed in write SQL")
+		}
+		if tok.kind == tkOp && (tok.value == "[" || tok.value == "`") {
+			return fmt.Errorf("quoted identifiers are not allowed in write SQL")
+		}
+	}
+
+	if len(tokens) < 8 {
+		return fmt.Errorf("write SQL must be INSERT INTO <catalog>.main.hep_proto_* VALUES ...")
+	}
+	if tokens[0].kind != tkIdent || tokens[0].upper != "INSERT" ||
+		tokens[1].kind != tkIdent || tokens[1].upper != "INTO" {
+		return fmt.Errorf("write SQL must start with INSERT INTO (got %q)", tokens[0].value)
+	}
+	if tokens[2].kind != tkIdent {
+		return fmt.Errorf("write target catalog must be an identifier")
+	}
+	if blockedWriteCatalogs[tokens[2].upper] {
+		return fmt.Errorf("system catalog %q is not allowed as write target", tokens[2].value)
+	}
+	if tokens[3].kind != tkDot ||
+		tokens[4].kind != tkIdent || tokens[4].upper != "MAIN" ||
+		tokens[5].kind != tkDot ||
+		tokens[6].kind != tkIdent {
+		return fmt.Errorf("write target must be <catalog>.main.hep_proto_<proto>_<name>")
+	}
+	if !hepProtoTableName.MatchString(tokens[6].value) {
+		return fmt.Errorf("write target table must match hep_proto_<proto>_<name> (got %q)", tokens[6].value)
+	}
+
+	i := 7
+	if i < len(tokens) && tokens[i].kind == tkLParen {
+		depth := 0
+		for i < len(tokens) {
+			switch tokens[i].kind {
+			case tkLParen:
+				depth++
+			case tkRParen:
+				depth--
+				i++
+				if depth == 0 {
+					goto afterCols
+				}
+				continue
+			}
+			i++
+		}
+		return fmt.Errorf("write SQL has an unclosed column list")
+	}
+afterCols:
+	if i >= len(tokens) || tokens[i].kind != tkIdent || tokens[i].upper != "VALUES" {
+		return fmt.Errorf("write SQL must use INSERT ... VALUES (SELECT/FROM is not allowed)")
 	}
 
-	// 3. Check for blocked DML/DDL keywords
 	for _, tok := range tokens {
 		if tok.kind != tkIdent {
 			continue
 		}
+		switch tok.upper {
+		case "SELECT", "FROM", "JOIN", "UNION", "EXCEPT", "INTERSECT",
+			"TABLE", "PIVOT", "UNPIVOT", "REPLACE", "OVERWRITE", "RETURNING":
+			return fmt.Errorf("blocked keyword %q in write SQL", tok.value)
+		}
+	}
+	return nil
+}
+
+func tokensForValidation(sql string) ([]token, error) {
+	trimmed := strings.TrimSpace(sql)
+	if trimmed == "" {
+		return nil, fmt.Errorf("empty SQL query")
+	}
+	tokens := tokenize(trimmed)
+	if len(tokens) == 0 {
+		return nil, fmt.Errorf("empty SQL query after parsing")
+	}
+	return tokens, nil
+}
+
+// rejectSQLHazards applies the GHSA-rm5w-rqr7-2h54 checks shared by read and write SQL.
+// skipDML lists keywords that may appear (the write path allows INSERT itself).
+func rejectSQLHazards(tokens []token, skipDML map[string]bool) error {
+	for _, tok := range tokens {
+		if tok.kind == tkSemi {
+			return fmt.Errorf("semicolons are not allowed (prevents multi-statement injection)")
+		}
+	}
+
+	for _, tok := range tokens {
+		if tok.kind != tkIdent {
+			continue
+		}
+		if skipDML[tok.upper] {
+			continue
+		}
 		if blockedDML[tok.upper] {
 			return fmt.Errorf("blocked keyword %q (DML/DDL operations are not allowed)", tok.value)
 		}
 	}
 
-	// 4. Check for blocked filesystem/network functions
 	for i, tok := range tokens {
 		if tok.kind != tkIdent {
 			continue
 		}
 		if blockedFunctions[tok.upper] {
-			// Only flag if it looks like a function call (followed by parenthesis)
 			if i+1 < len(tokens) && tokens[i+1].kind == tkLParen {
 				return fmt.Errorf("blocked function %q (filesystem/network access is not allowed)", tok.value)
 			}
-			// Also block as keyword for COPY, EXPORT, IMPORT (already in blockedDML)
 		}
 	}
 
-	// 5. Check for SELECT ... INTO (prevents writing to tables/files)
 	selectSeen := false
 	fromSeen := false
 	for _, tok := range tokens {
@@ -326,7 +437,6 @@ func ValidateRawSQL(sql string) error {
 		}
 	}
 
-	// 6. Check for CALL (except within string literals, already handled by tokenizer)
 	for _, tok := range tokens {
 		if tok.kind == tkIdent && tok.upper == "CALL" {
 			return fmt.Errorf("CALL statements are not allowed")
@@ -368,8 +478,8 @@ var blockedExprKeywords = map[string]bool{
 	"ATTACH":    true,
 	"DETACH":    true,
 	"LOAD":      true,
-	"INSTALL":  true,
-	"SET":      true,
+	"INSTALL":   true,
+	"SET":       true,
 	"INTO":      true,
 	"CALL":      true,
 	"UNION":     true,