Connections, "every packet" rules, and a packet tracer

- Connections: the kernel's tracked connections, read and closed over
  ctnetlink (internal/conntrack, raw netlink like the drop log), with
  their NAT and TCP state; closing one makes the rules check its next
  packet again (GET /connections, POST /connections/close; hc
  connections; Traffic → Connections).
- Rules can check every packet (every_packet, --every-packet): before
  established connections are accepted, so a drop or reject also stops
  connections already open. They can't accept.
- A packet tracer (internal/trace): a table of its own, inet
  housecarl-trace, marks the packets a filter picks, `nft monitor trace`
  reports every rule and policy they meet in every table, and the tracer
  names what accepted or dropped them, Housecarl's rules by name (POST,
  GET, DELETE /trace; hc trace; Traffic → Trace). One trace at a time,
  at most 200 packets and 20 a second; the table goes when it ends.

e2e: 157 checks, among them an upload a normal block doesn't stop and an
every-packet one does, and a trace naming the blocking rule.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

internal/server/h_traffic.go

@@ -0,0 +1,309 @@
+package server
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"sort"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"housecarl/internal/api"
+	"housecarl/internal/auth"
+	"housecarl/internal/model"
+	"housecarl/internal/trace"
+)
+
+// Traffic as it happens: the connections the kernel tracks (and closing
+// them, so the rules check them again), and tracing packets through the
+// whole ruleset.
+
+// Conntrack lists and closes tracked connections (internal/conntrack).
+type Conntrack interface {
+	List(ctx context.Context) ([]api.Connection, error)
+	Close(ctx context.Context, c api.Connection) error
+}
+
+// SetConntrack gives the server the kernel's connection tracking.
+func (s *Server) SetConntrack(c Conntrack) { s.conns = c }
+
+func (s *Server) conntrack() (Conntrack, error) {
+	if s.conns == nil {
+		return nil, &httpError{http.StatusServiceUnavailable, "unavailable", "connection tracking isn't available here"}
+	}
+	return s.conns, nil
+}
+
+// connMatches: a search in a connection's addresses, ports (":443"),
+// protocol, state and NAT addresses.
+func connMatches(c api.Connection, q string) bool {
+	if q == "" {
+		return true
+	}
+	hay := strings.ToLower(strings.Join([]string{c.Protocol, c.Source, ":" + strconv.Itoa(c.SourcePort), c.Destination, ":" + strconv.Itoa(c.DestinationPort),
+		c.DNATTo, c.SNATTo, c.State}, " ") + " ")
+	for _, w := range strings.Fields(strings.ToLower(q)) {
+		if strings.HasPrefix(w, ":") {
+			w += " " // a whole port
+		}
+		if !strings.Contains(hay, w) {
+			return false
+		}
+	}
+	return true
+}
+
+func (s *Server) listConnections(w http.ResponseWriter, r *http.Request, p *auth.Principal) error {
+	if err := require(p, auth.ResLog, false); err != nil {
+		return err
+	}
+	ct, err := s.conntrack()
+	if err != nil {
+		return err
+	}
+	all, err := ct.List(r.Context())
+	if err != nil {
+		return fmt.Errorf("listing connections: %w", err)
+	}
+	q := r.URL.Query().Get("q")
+	out := []api.Connection{}
+	for _, c := range all {
+		if connMatches(c, q) {
+			out = append(out, c)
+		}
+	}
+	sort.Slice(out, func(i, j int) bool {
+		if out[i].Source != out[j].Source {
+			return out[i].Source < out[j].Source
+		}
+		return out[i].Destination < out[j].Destination
+	})
+	if n := queryInt(r, "limit", 1000); n > 0 && len(out) > n {
+		out = out[:n]
+	}
+	w.Header().Set("X-Total-Count", strconv.Itoa(len(all)))
+	writeJSON(w, http.StatusOK, out)
+	return nil
+}
+
+// closeConnections deletes tracking entries: the rules check those
+// connections again with their next packet.
+func (s *Server) closeConnections(w http.ResponseWriter, r *http.Request, p *auth.Principal) error {
+	if err := require(p, auth.ResConfig, true); err != nil {
+		return err
+	}
+	ct, err := s.conntrack()
+	if err != nil {
+		return err
+	}
+	var in api.CloseConnections
+	if err := decodeJSON(r, &in); err != nil {
+		return err
+	}
+	list := in.Connections
+	if strings.TrimSpace(in.Search) != "" {
+		all, err := ct.List(r.Context())
+		if err != nil {
+			return fmt.Errorf("listing connections: %w", err)
+		}
+		for _, c := range all {
+			if connMatches(c, in.Search) {
+				list = append(list, c)
+			}
+		}
+	} else if len(list) == 0 {
+		return badRequest("say which connections: a list, or a search")
+	}
+	closed := 0
+	for _, c := range list {
+		if err := ct.Close(r.Context(), c); err != nil {
+			return fmt.Errorf("closing %s %s -> %s: %w", c.Protocol, c.Source, c.Destination, err)
+		}
+		closed++
+	}
+	what := in.Search
+	if what == "" && len(list) == 1 {
+		what = fmt.Sprintf("%s %s:%d -> %s:%d", list[0].Protocol, list[0].Source, list[0].SourcePort, list[0].Destination, list[0].DestinationPort)
+	}
+	s.eng.Audit(r.Context(), p, "connections.close", what, "success", fmt.Sprintf("%d closed", closed))
+	writeJSON(w, http.StatusOK, map[string]int{"closed": closed})
+	return nil
+}
+
+// ---- tracing
+
+type tracing struct {
+	mu     sync.Mutex
+	status api.TraceStatus
+	tracer *trace.Tracer
+	cancel context.CancelFunc
+}
+
+func (s *Server) startTrace(w http.ResponseWriter, r *http.Request, p *auth.Principal) error {
+	if err := require(p, auth.ResConfig, true); err != nil {
+		return err
+	}
+	var in api.TraceInput
+	if err := decodeJSON(r, &in); err != nil {
+		return err
+	}
+	if in.Seconds == 0 {
+		in.Seconds = 60
+	}
+	if in.Seconds < 5 || in.Seconds > 600 {
+		return badRequest("trace for 5 to 600 seconds")
+	}
+	script, err := trace.Script(in)
+	if err != nil {
+		return badRequest("%v", err)
+	}
+	t := &s.tracing
+	t.mu.Lock()
+	if t.status.Running {
+		t.mu.Unlock()
+		return conflict("a trace is running already (started %s); stop it first", t.status.Started.Format("15:04:05"))
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(in.Seconds)*time.Second)
+	now := s.nowFn()
+	t.status = api.TraceStatus{Running: true, Input: in, Started: now, Ends: now.Add(time.Duration(in.Seconds) * time.Second), Packets: []api.TracePacket{}}
+	t.tracer = &trace.Tracer{Now: s.nowFn}
+	t.cancel = cancel
+	t.mu.Unlock()
+	s.eng.Audit(r.Context(), p, "trace.start", traceWhat(in), "success", fmt.Sprintf("%d seconds", in.Seconds))
+	go func() {
+		defer cancel()
+		err := s.fw.Trace(ctx, script, func(line string) {
+			t.mu.Lock()
+			defer t.mu.Unlock()
+			t.tracer.Line(line)
+			if t.tracer.Full() {
+				cancel()
+			}
+		})
+		t.mu.Lock()
+		defer t.mu.Unlock()
+		t.status.Running = false
+		if err != nil {
+			t.status.Error = err.Error()
+			s.log.Warn("tracing failed", "err", err)
+		}
+	}()
+	writeJSON(w, http.StatusOK, s.traceStatus())
+	return nil
+}
+
+func traceWhat(in api.TraceInput) string {
+	var parts []string
+	if in.Address != "" {
+		parts = append(parts, in.Address)
+	}
+	if in.Protocol != "" {
+		parts = append(parts, in.Protocol)
+	}
+	if in.Port != 0 {
+		parts = append(parts, "port "+strconv.Itoa(in.Port))
+	}
+	return strings.Join(parts, " ")
+}
+
+// traceStatus is the trace with each step named after Housecarl's object.
+func (s *Server) traceStatus() api.TraceStatus {
+	t := &s.tracing
+	t.mu.Lock()
+	out := t.status
+	if t.tracer != nil {
+		out.Packets = t.tracer.Packets()
+	}
+	t.mu.Unlock()
+	if out.Packets == nil {
+		out.Packets = []api.TracePacket{}
+	}
+	st := s.eng.State()
+	for i := range out.Packets {
+		p := &out.Packets[i]
+		for j := range p.Steps {
+			p.Steps[j].Label = markLabel(st, p.Steps[j])
+		}
+		p.DecidedBy.Label = markLabel(st, p.DecidedBy)
+	}
+	return out
+}
+
+// markLabel names the Housecarl object behind a step: "rule input #12
+// (ssh)", "built-in: established", "port forward web", "input policy".
+func markLabel(st *model.State, step api.TraceStep) string {
+	if step.Kind == "policy" {
+		if step.Table == "inet housecarl" {
+			return step.Chain + " policy"
+		}
+		return ""
+	}
+	m := step.Mark
+	kind, id := "", int64(0)
+	switch {
+	case m == "":
+		return ""
+	case strings.HasPrefix(m, "b-"):
+		return "built-in: " + strings.TrimPrefix(m, "b-")
+	case strings.HasPrefix(m, "r"):
+		kind = model.KindRule
+	case strings.HasPrefix(m, "f"):
+		kind = model.KindForward
+	case strings.HasPrefix(m, "s"):
+		kind = model.KindSNAT
+	case strings.HasPrefix(m, "d"):
+		kind = model.KindRedirect
+	default:
+		return ""
+	}
+	num := strings.TrimLeft(m[1:], "0123456789")
+	id, _ = strconv.ParseInt(strings.TrimSuffix(m[1:], num), 10, 64)
+	if o := st.ByID(kind, id); o != nil {
+		label := o.Label() // "source NAT #3 (…)", "redirect #4", a forward's name…
+		switch {
+		case kind == model.KindRule:
+			label = "rule " + label
+		case kind == model.KindForward && !strings.HasPrefix(label, "port forward"):
+			label = "port forward " + label
+		}
+		if num != "" {
+			label += " (" + strings.TrimPrefix(num, "-") + ")"
+		}
+		return label
+	}
+	return ""
+}
+
+func (s *Server) getTrace(w http.ResponseWriter, r *http.Request, p *auth.Principal) error {
+	if err := require(p, auth.ResLog, false); err != nil {
+		return err
+	}
+	writeJSON(w, http.StatusOK, s.traceStatus())
+	return nil
+}
+
+func (s *Server) stopTrace(w http.ResponseWriter, r *http.Request, p *auth.Principal) error {
+	if err := require(p, auth.ResConfig, true); err != nil {
+		return err
+	}
+	t := &s.tracing
+	t.mu.Lock()
+	if t.cancel != nil && t.status.Running {
+		t.cancel()
+	}
+	t.mu.Unlock()
+	// Let the trace end, so the answer shows it stopped.
+	for i := 0; i < 50; i++ {
+		t.mu.Lock()
+		running := t.status.Running
+		t.mu.Unlock()
+		if !running {
+			break
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	writeJSON(w, http.StatusOK, s.traceStatus())
+	return nil
+}

internal/server/routes.go

@@ -42,6 +42,11 @@ func (s *Server) routes() {
 	m.HandleFunc("POST "+p+"/sets/{ref}/elements", s.authed(s.addElement))
 	m.HandleFunc("DELETE "+p+"/sets/{ref}/elements/{addr}", s.authed(s.deleteElement))
 	m.HandleFunc("GET "+p+"/quotas", s.authed(s.listQuotas))
+	m.HandleFunc("GET "+p+"/connections", s.authed(s.listConnections))
+	m.HandleFunc("POST "+p+"/connections/close", s.authed(s.closeConnections))
+	m.HandleFunc("GET "+p+"/trace", s.authed(s.getTrace))
+	m.HandleFunc("POST "+p+"/trace", s.authed(s.startTrace))
+	m.HandleFunc("DELETE "+p+"/trace", s.authed(s.stopTrace))
 	m.HandleFunc("POST "+p+"/rules/{ref}/reset-quota", s.authed(s.resetQuota))
 	m.HandleFunc("GET "+p+"/changesets", s.authed(s.listChangesets))
 	m.HandleFunc("POST "+p+"/changesets", s.authed(s.createChangeset))

internal/server/routes_cfg.go

@@ -1,6 +1,10 @@
 package server
 
-import "encoding/json"
+import (
+	"encoding/json"
+
+	"housecarl/internal/trace"
+)
 
 // otherTables lists the tables in an `nft -j list ruleset` other than
 // Housecarl's ("inet main").
@@ -18,7 +22,7 @@ func otherTables(data []byte) []string {
 			continue
 		}
 		var t struct{ Family, Name string }
-		if json.Unmarshal(raw, &t) == nil && !(t.Family == "inet" && t.Name == "housecarl") {
+		if json.Unmarshal(raw, &t) == nil && !(t.Family == "inet" && (t.Name == "housecarl" || t.Name == trace.Table)) {
 			out = append(out, t.Family+" "+t.Name)
 		}
 	}

internal/server/server.go

@@ -67,6 +67,8 @@ type Firewall interface {
 	SaveBoot(st *model.State) error
 	// ResetQuota starts a rule's quota again (perAddress: its sets).
 	ResetQuota(ctx context.Context, ruleID int64, perAddress bool) error
+	// Trace follows packets (internal/trace) until ctx ends.
+	Trace(ctx context.Context, script string, line func(string)) error
 }
 
 // Server is the HTTP server.
@@ -91,6 +93,10 @@ type Server struct {
 	// unitLinks are where an enabled nftables.service shows (tests change
 	// them).
 	unitLinks []string
+	// conns is the kernel's connection tracking (nil: unavailable).
+	conns Conntrack
+	// tracing is the packet trace running (or the last one).
+	tracing tracing
 	// csMu orders changes to changesets (staging, applying, discarding).
 	csMu  sync.Mutex
 	nowFn func() time.Time

internal/server/server_test.go

@@ -29,14 +29,16 @@ import (
 // fakeFW stands in for nftables: it records the applied states and keeps
 // dynamic sets' entries.
 type fakeFW struct {
-	mu       sync.Mutex
-	applied  []*model.State
-	boot     *model.State
-	elements map[string][]render.Element
-	applyErr error
-	deleted  []string
-	restored []string
-	resets   []string
+	mu         sync.Mutex
+	applied    []*model.State
+	boot       *model.State
+	elements   map[string][]render.Element
+	applyErr   error
+	deleted    []string
+	restored   []string
+	resets     []string
+	traced     string
+	traceLines []string
 }
 
 func (f *fakeFW) Check(_ context.Context, st *model.State) error { return nil }
@@ -96,6 +98,17 @@ func (f *fakeFW) ResetQuota(_ context.Context, id int64, perAddress bool) error
 	f.resets = append(f.resets, fmt.Sprintf("r%d:%v", id, perAddress))
 	return nil
 }
+func (f *fakeFW) Trace(ctx context.Context, script string, line func(string)) error {
+	f.mu.Lock()
+	f.traced = script
+	lines := f.traceLines
+	f.mu.Unlock()
+	for _, l := range lines {
+		line(l)
+	}
+	<-ctx.Done()
+	return nil
+}
 func (f *fakeFW) TableText(_ context.Context, fam, name string) (string, error) {
 	return "table " + fam + " " + name + " {}", nil
 }
@@ -865,3 +878,91 @@ func TestHostProblems(t *testing.T) {
 		t.Errorf("warned about a table that is gone: %s", c)
 	}
 }
+
+type fakeConns struct {
+	list   []api.Connection
+	closed []api.Connection
+}
+
+func (f *fakeConns) List(context.Context) ([]api.Connection, error) { return f.list, nil }
+func (f *fakeConns) Close(_ context.Context, c api.Connection) error {
+	f.closed = append(f.closed, c)
+	return nil
+}
+
+// TestConnections: connections listed and searched; closed one by one, or
+// all a search finds; closing needs config:write, and is audited.
+func TestConnections(t *testing.T) {
+	e := newEnv(t)
+	var code int
+	if code = e.call(e.token, "GET", "/connections", nil, nil); code != 503 {
+		t.Errorf("without conntrack: %d", code)
+	}
+	fc := &fakeConns{list: []api.Connection{
+		{Family: "ipv4", Protocol: "tcp", Source: "192.168.1.50", SourcePort: 51432, Destination: "198.51.100.9", DestinationPort: 443, SNATTo: "203.0.113.10", State: "established"},
+		{Family: "ipv4", Protocol: "udp", Source: "192.168.1.51", SourcePort: 5353, Destination: "192.168.1.1", DestinationPort: 53},
+		{Family: "ipv6", Protocol: "tcp", Source: "2001:db8:1::50", SourcePort: 40000, Destination: "2001:db8:2::9", DestinationPort: 4430},
+	}}
+	e.srv.SetConntrack(fc)
+	var cs []api.Connection
+	e.call(e.token, "GET", "/connections?q=192.168.1.50", nil, &cs)
+	if len(cs) != 1 || cs[0].DestinationPort != 443 {
+		t.Errorf("search by address: %+v", cs)
+	}
+	e.call(e.token, "GET", "/connections?q=:443", nil, &cs)
+	if len(cs) != 1 || cs[0].Source != "192.168.1.50" {
+		t.Errorf("search by port (not :4430): %+v", cs)
+	}
+	var out map[string]int
+	if code := e.call(e.token, "POST", "/connections/close", api.CloseConnections{Search: "tcp"}, &out); code != 200 || out["closed"] != 2 || len(fc.closed) != 2 {
+		t.Errorf("close a search: %d %+v %+v", code, out, fc.closed)
+	}
+	if code := e.call(e.token, "POST", "/connections/close", api.CloseConnections{}, nil); code != 400 {
+		t.Errorf("close nothing said: %d", code)
+	}
+	var tok api.Token
+	e.call(e.token, "POST", "/me/tokens", api.TokenInput{Name: "monitor", Role: "log"}, &tok)
+	if code := e.call(tok.Secret, "GET", "/connections", nil, nil); code != 200 {
+		t.Errorf("listing with log:read: %d", code)
+	}
+	if code := e.call(tok.Secret, "POST", "/connections/close", api.CloseConnections{Search: "udp"}, nil); code != 403 {
+		t.Errorf("closing with log:read: %d", code)
+	}
+}
+
+// TestTrace: a trace runs with the filter's table, collects packets named
+// after Housecarl's objects, and stops when asked; one at a time.
+func TestTrace(t *testing.T) {
+	e := newEnv(t)
+	var mr api.MutationResponse
+	e.call(e.token, "POST", "/rules", model.Rule{Chain: "input", Match: model.Match{Services: []string{"tcp/9999"}}, Action: "drop", Comment: "block 9999"}, &mr)
+	var rule model.Rule
+	json.Unmarshal(mr.Object, &rule)
+	e.fw.traceLines = []string{
+		`trace id 1 inet housecarl input packet: iif "wan" ip saddr 198.51.100.7 ip daddr 203.0.113.10 tcp sport 4000 tcp dport 9999 tcp flags == syn`,
+		`trace id 1 inet housecarl input rule ct state established,related accept comment "hc:b-established" (verdict continue)`,
+		fmt.Sprintf(`trace id 1 inet housecarl input rule tcp dport 9999 counter drop comment "hc:r%d block 9999" (verdict drop)`, rule.ID),
+	}
+	var st api.TraceStatus
+	if code := e.call(e.token, "POST", "/trace", api.TraceInput{Port: 9999, Seconds: 30}, &st); code != 200 || !st.Running {
+		t.Fatalf("start: %d %+v", code, st)
+	}
+	if code := e.call(e.token, "POST", "/trace", api.TraceInput{Port: 22}, nil); code != 409 {
+		t.Errorf("a second trace: %d", code)
+	}
+	time.Sleep(100 * time.Millisecond)
+	e.call(e.token, "GET", "/trace", nil, &st)
+	if len(st.Packets) != 1 || st.Packets[0].Verdict != "dropped" || st.Packets[0].DecidedBy.Label != "rule input #"+itoa(rule.ID)+" (block 9999)" ||
+		st.Packets[0].Steps[0].Label != "built-in: established" {
+		t.Errorf("packets: %+v", st.Packets)
+	}
+	if !strings.Contains(e.fw.traced, "th dport 9999") {
+		t.Errorf("the trace table: %s", e.fw.traced)
+	}
+	if code := e.call(e.token, "DELETE", "/trace", nil, &st); code != 200 || st.Running || len(st.Packets) != 1 {
+		t.Errorf("stop: %d %+v", code, st)
+	}
+	if code := e.call(e.token, "POST", "/trace", api.TraceInput{}, nil); code != 400 {
+		t.Errorf("tracing everything: %d", code)
+	}
+}