housecarl/internal/server/h_traffic.go

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
	}
	if err := s.refuseInShadow("close connections"); 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)
		if p.WouldDrop != nil {
			// A copy: the tracer still holds the step.
			wd := *p.WouldDrop
			wd.Label = markLabel(st, wd)
			p.WouldDrop = &wd
		}
	}
	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
}