From 52da3c13f6498d4503bf12cd6a0f69dc971187e0 Mon Sep 17 00:00:00 2001 From: Andy Dixon Date: Fri, 9 Oct 2026 17:34:32 +0100 Subject: [PATCH] Core packages: config validation, network renderers, root helper, auth, SQLite store, capture engine Capture engine streams dumpcap pcapng into verbatim, hashed segments with per-segment tshark summaries; display filters, PDML dissection with verified byte ranges, export and import. Tests run in a Debian trixie container. --- .gitignore | 4 + cmd/pilens-helper/main.go | 484 ++++++++++ docs/architecture.md | 172 ++++ go.mod | 20 + go.sum | 52 ++ internal/auth/auth.go | 173 ++++ internal/auth/auth_test.go | 41 + internal/capture/engine.go | 824 ++++++++++++++++++ internal/capture/engine_test.go | 198 +++++ internal/capture/hub.go | 53 ++ internal/capture/pdml.go | 107 +++ internal/capture/query.go | 503 +++++++++++ internal/capture/tshark.go | 273 ++++++ internal/config/config.go | 293 +++++++ internal/helper/helper.go | 101 +++ internal/helper/parse.go | 75 ++ internal/helper/parse_test.go | 25 + internal/netgen/netgen.go | 242 +++++ internal/netgen/netgen_test.go | 90 ++ internal/pcapfile/pcapfile.go | 336 +++++++ internal/pcapfile/pcapfile_test.go | 54 ++ internal/store/store.go | 88 ++ migrations/001_init.sql | 112 +++ migrations/embed.go | 7 + .../fixtures/sample-loopback-http-dns.pcapng | Bin 0 -> 5104 bytes tests/integration/Dockerfile | 8 + 26 files changed, 4335 insertions(+) create mode 100644 .gitignore create mode 100644 cmd/pilens-helper/main.go create mode 100644 docs/architecture.md create mode 100644 go.mod create mode 100644 go.sum create mode 100644 internal/auth/auth.go create mode 100644 internal/auth/auth_test.go create mode 100644 internal/capture/engine.go create mode 100644 internal/capture/engine_test.go create mode 100644 internal/capture/hub.go create mode 100644 internal/capture/pdml.go create mode 100644 internal/capture/query.go create mode 100644 internal/capture/tshark.go create mode 100644 internal/config/config.go create mode 100644 internal/helper/helper.go create mode 100644 internal/helper/parse.go create mode 100644 internal/helper/parse_test.go create mode 100644 internal/netgen/netgen.go create mode 100644 internal/netgen/netgen_test.go create mode 100644 internal/pcapfile/pcapfile.go create mode 100644 internal/pcapfile/pcapfile_test.go create mode 100644 internal/store/store.go create mode 100644 migrations/001_init.sql create mode 100644 migrations/embed.go create mode 100644 tests/fixtures/sample-loopback-http-dns.pcapng create mode 100644 tests/integration/Dockerfile diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ce9852a --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +/dist/ +/web/node_modules/ +/web/dist/ +*.test diff --git a/cmd/pilens-helper/main.go b/cmd/pilens-helper/main.go new file mode 100644 index 0000000..3641c40 --- /dev/null +++ b/cmd/pilens-helper/main.go @@ -0,0 +1,484 @@ +// pilens-helper is the only root component of PiLens. It accepts a fixed set of +// typed requests from the pilens service user over a unix socket and applies +// network configuration transactionally. It never executes caller-supplied commands. +package main + +import ( + "context" + "encoding/json" + "errors" + "flag" + "fmt" + "log/slog" + "net" + "os" + "os/exec" + "os/signal" + "os/user" + "path/filepath" + "strconv" + "strings" + "sync" + "syscall" + "time" + + "pilens.dev/pilens/internal/config" + "pilens.dev/pilens/internal/helper" + "pilens.dev/pilens/internal/netgen" +) + +var ( + stateDir = "/var/lib/pilens-helper" + keyfilePath = "/etc/NetworkManager/system-connections/" + netgen.ConnName + ".nmconnection" +) + +func statePath(n string) string { return filepath.Join(stateDir, n) } + +type server struct { + mu sync.Mutex // one network transaction at a time + allowUID uint32 + lastError string +} + +func main() { + socket := flag.String("socket", helper.DefaultSocket, "unix socket path") + group := flag.String("user", "pilens", "service user allowed to connect") + boot := flag.Bool("boot", false, "load the last good firewall ruleset and exit (ExecStartPre)") + render := flag.String("render", "", "render config JSON file to stdout sections and exit (diagnostics)") + flag.Parse() + log := slog.New(slog.NewJSONHandler(os.Stderr, nil)) + slog.SetDefault(log) + + if *render != "" { + os.Exit(renderCmd(*render)) + } + if *boot { + if err := bootLoad(); err != nil { + log.Error("boot load failed", "err", err) + os.Exit(1) + } + return + } + + u, err := user.Lookup(*group) + if err != nil { + log.Error("service user missing", "user", *group, "err", err) + os.Exit(1) + } + uid, _ := strconv.Atoi(u.Uid) + gid, _ := strconv.Atoi(u.Gid) + s := &server{allowUID: uint32(uid)} + + os.MkdirAll(filepath.Dir(*socket), 0o750) + os.Chown(filepath.Dir(*socket), 0, gid) + os.Remove(*socket) + ln, err := net.Listen("unix", *socket) + if err != nil { + log.Error("listen", "err", err) + os.Exit(1) + } + os.Chown(*socket, 0, gid) + os.Chmod(*socket, 0o660) + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT) + defer stop() + go func() { <-ctx.Done(); ln.Close() }() + log.Info("pilens-helper ready", "socket", *socket) + for { + c, err := ln.Accept() + if err != nil { + if ctx.Err() != nil { + return + } + continue + } + go s.handle(ctx, c.(*net.UnixConn)) + } +} + +func peerUID(c *net.UnixConn) (uint32, error) { + raw, err := c.SyscallConn() + if err != nil { + return 0, err + } + var cred *syscall.Ucred + var cerr error + raw.Control(func(fd uintptr) { cred, cerr = syscall.GetsockoptUcred(int(fd), syscall.SOL_SOCKET, syscall.SO_PEERCRED) }) + if cerr != nil { + return 0, cerr + } + return cred.Uid, nil +} + +func (s *server) handle(ctx context.Context, c *net.UnixConn) { + defer c.Close() + uid, err := peerUID(c) + if err != nil || (uid != s.allowUID && uid != 0) { + slog.Warn("rejected helper connection", "uid", uid, "err", err) + return + } + c.SetDeadline(time.Now().Add(3 * time.Minute)) + var req helper.Request + if err := json.NewDecoder(&limitedReader{c, 1 << 20}).Decode(&req); err != nil { + writeResp(c, nil, errors.New("bad request")) + return + } + slog.Info("helper request", "op", req.Op, "uid", uid) + var data any + switch req.Op { + case helper.OpApplyNetwork: + if req.Network == nil { + err = errors.New("network config required") + break + } + r := s.apply(ctx, *req.Network) + data = r + if r.Error != "" { + err = errors.New(r.Error) + } + case helper.OpStatus: + data = s.status(ctx) + case helper.OpStations: + data, err = stations(ctx) + case helper.OpLeases: + b, rerr := os.ReadFile(netgen.LeaseFile) + if rerr != nil && !os.IsNotExist(rerr) { + err = rerr + } + data = helper.ParseLeases(string(b)) + default: + err = fmt.Errorf("unknown op %q", req.Op) + } + writeResp(c, data, err) +} + +type limitedReader struct { + c *net.UnixConn + n int +} + +func (l *limitedReader) Read(p []byte) (int, error) { + if l.n <= 0 { + return 0, errors.New("request too large") + } + if len(p) > l.n { + p = p[:l.n] + } + n, err := l.c.Read(p) + l.n -= n + return n, err +} + +func writeResp(c net.Conn, data any, err error) { + r := helper.Response{OK: err == nil} + if err != nil { + r.Error = err.Error() + } + if data != nil { + r.Data, _ = json.Marshal(data) + } + json.NewEncoder(c).Encode(r) +} + +// run executes a fixed program with an argv list (never a shell). +func run(ctx context.Context, timeout time.Duration, name string, args ...string) (string, error) { + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + out, err := exec.CommandContext(ctx, name, args...).CombinedOutput() + if err != nil { + return string(out), fmt.Errorf("%s %s: %v: %s", name, strings.Join(args, " "), err, strings.TrimSpace(string(out))) + } + return string(out), nil +} + +func writeAtomic(path string, data []byte, mode os.FileMode) error { + os.MkdirAll(filepath.Dir(path), 0o755) + tmp := path + ".tmp" + if err := os.WriteFile(tmp, data, mode); err != nil { + return err + } + if err := os.Chmod(tmp, mode); err != nil { + return err + } + return os.Rename(tmp, path) +} + +// snapshot of the files a transaction touches; nil content = file absent. +type snapshot map[string][]byte + +func takeSnapshot(paths ...string) snapshot { + s := snapshot{} + for _, p := range paths { + b, err := os.ReadFile(p) + if err == nil { + s[p] = b + } else { + s[p] = nil + } + } + return s +} + +func (s *server) apply(ctx context.Context, n config.Network) helper.ApplyResult { + s.mu.Lock() + defer s.mu.Unlock() + var r helper.ApplyResult + step := func(f string, a ...any) { r.Steps = append(r.Steps, fmt.Sprintf(f, a...)) } + + if err := n.Validate(true); err != nil { + r.Error = "validation: " + err.Error() + return r + } + step("validated configuration") + ruleset := netgen.Nftables(n) + tmp := statePath("ruleset.check.nft") + if err := writeAtomic(tmp, []byte(ruleset), 0o600); err != nil { + r.Error = err.Error() + return r + } + if _, err := run(ctx, 20*time.Second, "nft", "-c", "-f", tmp); err != nil { + r.Error = "firewall ruleset rejected by nft -c: " + err.Error() + return r + } + step("nft dry-run passed") + + files := map[string][]byte{ + statePath("ruleset.nft"): []byte(ruleset), + statePath("dnsmasq.conf"): []byte(netgen.DnsmasqConf(n)), + keyfilePath: []byte(netgen.NMKeyfile(n)), + } + var paths []string + for p := range files { + paths = append(paths, p) + } + snap := takeSnapshot(paths...) + prevCfg, _ := os.ReadFile(statePath("last-good.json")) + step("saved previous configuration") + + err := activate(ctx, n, files, step) + if err == nil { + err = verify(ctx, n, step) + } + if err != nil { + r.Error = err.Error() + step("FAILED: %v", err) + step("rolling back") + if rbErr := rollback(ctx, snap, prevCfg); rbErr != nil { + r.Error += "; ROLLBACK ALSO FAILED: " + rbErr.Error() + step("rollback failed: %v", rbErr) + } else { + step("previous configuration restored") + } + r.RolledBack = true + s.lastError = r.Error + return r + } + b, _ := json.Marshal(n) + writeAtomic(statePath("last-good.json"), b, 0o600) + s.lastError = "" + r.Applied = true + step("configuration committed") + return r +} + +func activate(ctx context.Context, n config.Network, files map[string][]byte, step func(string, ...any)) error { + for p, b := range files { + if err := writeAtomic(p, b, 0o600); err != nil { + return err + } + } + if _, err := run(ctx, 20*time.Second, "nft", "-f", statePath("ruleset.nft")); err != nil { + return err + } + step("firewall table inet pilens loaded") + if n.Forwarding.Mode == "internet" { + // Only ever enabled here, after the ruleset is in place. We never disable it: + // other host services may rely on it and the pilens table drops AP traffic in deny mode. + if err := os.WriteFile("/proc/sys/net/ipv4/ip_forward", []byte("1"), 0o644); err != nil { + return err + } + step("IPv4 forwarding enabled") + } + if _, err := run(ctx, 10*time.Second, "iw", "reg", "set", n.AP.Country); err != nil { + step("warning: could not set regulatory domain: %v", err) + } + if _, err := run(ctx, 20*time.Second, "nmcli", "connection", "load", keyfilePath); err != nil { + return err + } + if n.AP.Enabled { + if _, err := run(ctx, 60*time.Second, "nmcli", "--wait", "45", "connection", "up", netgen.ConnName); err != nil { + return err + } + step("access point %q activated", n.AP.SSID) + if _, err := run(ctx, 30*time.Second, "systemctl", "restart", "pilens-dnsmasq.service"); err != nil { + return err + } + step("DHCP/DNS restarted") + } else { + run(ctx, 30*time.Second, "nmcli", "connection", "down", netgen.ConnName) + run(ctx, 30*time.Second, "systemctl", "stop", "pilens-dnsmasq.service") + step("access point disabled") + } + return nil +} + +func verify(ctx context.Context, n config.Network, step func(string, ...any)) error { + if _, err := run(ctx, 10*time.Second, "nft", "list", "table", "inet", netgen.TableName); err != nil { + return errors.New("firewall table missing after apply") + } + if !n.AP.Enabled { + return nil + } + deadline := time.Now().Add(15 * time.Second) + for { + st := apState(ctx) + dn := unitActive(ctx, "pilens-dnsmasq.service") + if st == "activated" && dn { + step("verified: AP activated, DHCP/DNS active") + return nil + } + if time.Now().After(deadline) { + return fmt.Errorf("health check failed: AP state %q, dnsmasq active=%v", st, dn) + } + time.Sleep(time.Second) + } +} + +func rollback(ctx context.Context, snap snapshot, prevCfg []byte) error { + var errs []error + for p, b := range snap { + if b == nil { + os.Remove(p) + } else if err := writeAtomic(p, b, 0o600); err != nil { + errs = append(errs, err) + } + } + rs := statePath("ruleset.nft") + if snap[rs] != nil { + if _, err := run(ctx, 20*time.Second, "nft", "-f", rs); err != nil { + errs = append(errs, err) + } + } else { + // No previous policy: fail closed on the AP side. + var n config.Network + ap := "wlan0" + if json.Unmarshal(prevCfg, &n) == nil && n.AP.Interface != "" { + ap = n.AP.Interface + } + writeAtomic(rs, []byte(netgen.DenyAll(ap)), 0o600) + if _, err := run(ctx, 20*time.Second, "nft", "-f", rs); err != nil { + errs = append(errs, err) + } + } + if snap[keyfilePath] != nil { + run(ctx, 20*time.Second, "nmcli", "connection", "load", keyfilePath) + var n config.Network + if json.Unmarshal(prevCfg, &n) == nil && n.AP.Enabled { + if _, err := run(ctx, 60*time.Second, "nmcli", "--wait", "45", "connection", "up", netgen.ConnName); err != nil { + errs = append(errs, err) + } + run(ctx, 30*time.Second, "systemctl", "restart", "pilens-dnsmasq.service") + } + } else { + run(ctx, 20*time.Second, "nmcli", "connection", "delete", netgen.ConnName) + run(ctx, 30*time.Second, "systemctl", "stop", "pilens-dnsmasq.service") + } + return errors.Join(errs...) +} + +func apState(ctx context.Context) string { + out, err := run(ctx, 10*time.Second, "nmcli", "-g", "GENERAL.STATE", "connection", "show", netgen.ConnName) + if err != nil { + return "absent" + } + if s := strings.TrimSpace(out); s != "" { + return s + } + return "inactive" +} + +func unitActive(ctx context.Context, unit string) bool { + _, err := run(ctx, 10*time.Second, "systemctl", "is-active", "--quiet", unit) + return err == nil +} + +func (s *server) status(ctx context.Context) helper.Status { + st := helper.Status{LastError: s.lastError} + _, err := run(ctx, 10*time.Second, "nft", "list", "table", "inet", netgen.TableName) + st.TableLoaded = err == nil + st.APState = apState(ctx) + st.DnsmasqActive = unitActive(ctx, "pilens-dnsmasq.service") + b, _ := os.ReadFile("/proc/sys/net/ipv4/ip_forward") + st.IPForward = strings.TrimSpace(string(b)) == "1" + if out, err := run(ctx, 5*time.Second, "iw", "reg", "get"); err == nil { + for _, l := range strings.Split(out, "\n") { + if strings.HasPrefix(l, "country ") { + st.Regdomain = strings.TrimSuffix(strings.Fields(l)[1], ":") + break + } + } + } + _, err = os.Stat(statePath("last-good.json")) + st.Configured = err == nil + return st +} + +func stations(ctx context.Context) ([]helper.Station, error) { + b, err := os.ReadFile(statePath("last-good.json")) + if err != nil { + return nil, nil + } + var n config.Network + if err := json.Unmarshal(b, &n); err != nil || !config.ValidInterface(n.AP.Interface) { + return nil, errors.New("stored configuration unreadable") + } + out, err := run(ctx, 10*time.Second, "iw", "dev", n.AP.Interface, "station", "dump") + if err != nil { + return nil, err + } + return helper.ParseStationDump(out), nil +} + +// bootLoad restores the last good ruleset (or a fail-closed one) before networking starts. +func bootLoad() error { + ctx := context.Background() + rs := statePath("ruleset.nft") + if _, err := os.Stat(rs); err != nil { + slog.Info("no ruleset yet; nothing to load") + return nil + } + if _, err := run(ctx, 20*time.Second, "nft", "-f", rs); err != nil { + return err + } + b, err := os.ReadFile(statePath("last-good.json")) + var n config.Network + if err == nil && json.Unmarshal(b, &n) == nil { + if n.Forwarding.Mode == "internet" { + os.WriteFile("/proc/sys/net/ipv4/ip_forward", []byte("1"), 0o644) + } + if n.AP.Country != "" { + run(ctx, 10*time.Second, "iw", "reg", "set", n.AP.Country) + } + } + slog.Info("firewall restored at boot") + return nil +} + +func renderCmd(path string) int { + b, err := os.ReadFile(path) + if err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + var n config.Network + if err := json.Unmarshal(b, &n); err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + if err := n.Validate(true); err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + fmt.Printf("### nftables\n%s\n### dnsmasq.conf\n%s\n### %s (psk redacted)\n%s", netgen.Nftables(n), netgen.DnsmasqConf(n), keyfilePath, netgen.NMKeyfile(n.Redacted())) + return 0 +} diff --git a/docs/architecture.md b/docs/architecture.md new file mode 100644 index 0000000..fe3d93b --- /dev/null +++ b/docs/architecture.md @@ -0,0 +1,172 @@ +# PiLens architecture + +PiLens turns a Raspberry Pi into a routed Wi-Fi test network with packet capture +and analysis. This document records the decisions and the reasons for them. +Anything here that has not been checked on real hardware says so. + +## Processes and privilege + +``` + management LAN (TLS :8443, allow-listed subnets only) + │ +┌───────────────────────────────▼────────────────────────────────┐ +│ pilens (user "pilens", no root) │ +│ HTTP API + embedded React UI, auth, SQLite, SSE │ +│ capture engine: dumpcap ─stdout─▶ pcapng splitter ─▶ segments │ +│ └─▶ tshark (per segment) │ +│ dissection on demand: tshark -T pdml (rlimited child) │ +└───────────────┬─────────────────────────────────────────────────┘ + │ unix socket /run/pilens/helper.sock (SO_PEERCRED) +┌───────────────▼─────────────────────────────────────────────────┐ +│ pilens-helper (root, small, fixed operations only) │ +│ apply_network: NM keyfile + dnsmasq conf + nft table, verify, │ +│ roll back on failure │ +│ ap_stations, leases, status │ +└─────────────────────────────────────────────────────────────────┘ +pilens-dnsmasq (DHCP + DNS for the AP interface only) +NetworkManager (owns the AP connection "pilens-ap") +``` + +* **The core never runs as root.** It captures by running `dumpcap`, which + Debian ships with file capabilities (`cap_net_raw,cap_net_admin`) for members + of the `wireshark` group. Because file capabilities are lost under + `NoNewPrivileges`, the core unit does not set it. Instead it restricts its + `CapabilityBoundingSet` to those two capabilities, so no child can gain more. +* **tshark parses hostile input** (live traffic, uploaded files). Every tshark + child goes through `pilens rlimit-exec`, which sets `RLIMIT_AS`, `RLIMIT_CPU` + and `PR_SET_NO_NEW_PRIVS` and then execs tshark. Every call also has a context + timeout. +* **The helper is the only root component.** It takes typed JSON requests. It + checks the peer uid, validates every request again (it does not trust the + core), renders the configuration itself, and runs fixed argv lists. It has no + generic command API. + +## Networking + +The Pi is a **router**, not a bridge. The AP interface has a static gateway +address; clients get DHCP and DNS from a dedicated dnsmasq instance bound only +to that interface. + +| Concern | Owner | Why | +|---|---|---| +| AP radio | NetworkManager keyfile `pilens-ap` (`mode=ap`, `ipv4.method=manual`, `ipv6.method=disabled`) | NM is the supported network manager on Raspberry Pi OS Bookworm and later. We write a keyfile (0600) and `nmcli connection load` it. The Wi-Fi passphrase never appears in a process argv. | +| DHCP + DNS | `pilens-dnsmasq.service` | NM "shared" mode gives no reservations, query logging or overrides, and it adds its own NAT rules. We use `ipv4.method=manual` so NM runs no DHCP server, which means there are never two competing DHCP servers. Only `dnsmasq-base` is installed, so the distro's `dnsmasq.service` does not exist. | +| Firewall / NAT | nft table `inet pilens` | This is the only table we create or delete. Other tables are never flushed. | +| Forwarding sysctl | helper | It is enabled only after the PiLens table has loaded. It is not persisted in sysctl.d, so after a reboot nothing forwards until the rules are back. | + +### Firewall semantics + +In nftables a packet must survive every base chain on a hook: `accept` only +ends the current chain, but `drop` is final. Our chains use `policy accept` for +traffic that is not ours, so SSH and other host services are untouched. For +anything that enters through the AP interface we decide explicitly and end with +a `drop`. A permissive table added by someone else (Docker, NM) therefore cannot +open a path that we drop. + +* AP → Pi: DHCP, DNS and ICMP echo only. Everything else is dropped, including + the management port and SSH. +* AP → forward: only out of the uplink interface. Never to RFC 1918, link-local, + CGNAT, multicast or the management subnets, unless the administrator adds an + explicit allow CIDR. New connections are permitted only when forwarding mode is + `internet`. +* IPv6 from AP clients is dropped in `forward`, and IPv6 is disabled on the AP + interface. That closes the uninspected IPv6 escape path. IPv6 forwarding is + not implemented. +* Management port: accepted only on the configured management interfaces from + the configured source subnets, and dropped everywhere else. The core also + checks the source address in middleware, as defence in depth. +* `pilens-helper.service` loads the last good ruleset in `ExecStartPre`, ordered + `Before=network-pre.target`, so the rules are in place before NM brings up the + AP. + +### Transactional apply + +`apply_network` is one helper operation: + +1. Validate. +2. Render the NM keyfile, dnsmasq.conf and nft ruleset. +3. `nft -c` dry run. +4. Snapshot the current files. +5. Write the files atomically, load the ruleset, `nmcli connection load` + `up`, + and restart dnsmasq. +6. Verify: the connection is `activated`, dnsmasq is active, the table is + present, and forwarding matches the configured policy. +7. On any failure, restore the snapshot and re-apply it, then report both errors. + +Management and uplink NM profiles are never modified. + +## Capture engine + +``` +dumpcap -i IF [-f BPF] -w - ──pcapng──▶ splitter ──▶ seg-000001.pcapng (sha256 while writing) + │ seg-000002.pcapng … + ├──▶ tshark -r - -T fields (summary, one process per segment) + │ └─▶ SQLite packets(session,num,seg,off,…) ─▶ SSE batches + └──▶ ISB drop counters +``` + +* **Original bytes are never modified.** Every packet block is written verbatim. + On rotation, the SHB and IDBs are copied verbatim into the new segment, so + each segment is a self-contained, valid pcapng file. +* **One tshark per segment** keeps tshark's memory bounded. tshark's + conversation state otherwise grows without limit on long captures. The cost: + TCP stream numbers and reassembly restart at segment boundaries. Summary rows, + filters and the detail view all use the same per-segment dissection, so they + agree with each other. Segments default to 64 MiB. +* **SQLite holds one summary row per packet** (number, segment, file offset, + time, addresses, protocol, length, info). It does not hold dissection trees. + The protocol tree comes from `tshark -T pdml` on demand. +* **Display filters are evaluated by tshark** (`-Y`), segment by segment. The + results are cached per session and filter. PiLens does not implement its own + filter language. +* **Hex ↔ field mapping.** The core reads frame bytes from the segment file at + the indexed offset. A PDML field is highlighted only when the frame bytes at + `pos..pos+size` equal its `value`. Fields from reassembled or decompressed + data sources fail that check and are not highlighted. +* **Backpressure.** Each SSE subscriber has a bounded queue. When it is full, + batches for that subscriber are dropped, counted as *UI drops*, and the client + is told to resync from SQLite. Capture never waits for a browser. Capture + drops (from dumpcap/ISB) and ingestion lag (packets written but not yet + summarised) are reported separately. +* **Storage.** There is a global byte quota and a free-space reserve. When the + quota is reached, the oldest *unpreserved* sessions are deleted. If nothing + can be deleted, the capture stops with a visible error. Preserved sessions are + never deleted automatically. Per-session ring mode (`ring_segments`) deletes + that session's oldest segments and records which ones were rotated out. +* **Imports** keep the uploaded file unchanged as the session's only segment + (pcap or pcapng), with its SHA-256. Magic number and size are checked before + any parsing. + +## Web + +* TLS only, on `:8443`, with a self-signed management certificate generated on + first start (ECDSA P-256). It is separate from any interception CA. The + installer prints its fingerprint. +* Auth: Argon2id (64 MiB, t=2, p=2). Sessions are 256-bit random tokens; only + their SHA-256 is stored. Cookies are `Secure; HttpOnly; SameSite=Strict`. + CSRF uses a per-session token in the `X-CSRF-Token` header on every + state-changing request. Optional TOTP (RFC 6238). Logins are rate-limited per + IP and per user. +* First run needs a **setup token** that the installer generates and prints. + Without it, anyone on the management LAN could race to create the admin. +* Live updates use SSE: stdlib only, automatic reconnect, and `Last-Event-ID` + resync. +* React + TypeScript + Vite. The build is embedded in the Go binary with + `go:embed`. There are no CDNs or external fonts. + +## Decisions taken without a spec answer + +| Decision | Default | Reason | +|---|---|---| +| Live transport | SSE | One-way server push is all that is needed. It's stdlib, and reconnect is built in. | +| Config storage | JSON document in SQLite `settings`, with a `last_good` copy | One validated struct, simple rollback. | +| WPA2+WPA3 transition mode | Not offered | NM AP-mode support depends on the driver. We offer WPA2 or WPA3, and WPA3 failure triggers rollback. | +| SSID characters | Printable ASCII without `\` or `;` | Avoids keyfile escaping ambiguity. | +| AP client SSH to Pi | Blocked | The AP network is the untrusted side. | +| Max clients | Not offered | NM has no setting for it. The UI says so. | + +## Not yet implemented (phase 3+) + +These pages are not in the navigation until they work: conversations / follow +TCP stream, DNS/TLS/HTTP explorers, statistics, findings, TLS interception +(modes B and C), and scheduling. diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..56531c7 --- /dev/null +++ b/go.mod @@ -0,0 +1,20 @@ +module pilens.dev/pilens + +go 1.27.1 + +require ( + golang.org/x/crypto v0.57.0 + modernc.org/sqlite v1.60.1 +) + +require ( + github.com/dustin/go-humanize v1.0.1 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/mattn/go-isatty v0.0.24 // indirect + github.com/ncruces/go-strftime v1.0.0 // indirect + github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect + golang.org/x/sys v0.48.0 // indirect + modernc.org/libc v1.77.1 // indirect + modernc.org/mathutil v1.7.1 // indirect + modernc.org/memory v1.12.1 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..84f9bec --- /dev/null +++ b/go.sum @@ -0,0 +1,52 @@ +github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= +github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= +github.com/google/pprof v0.0.0-20260802141513-ef3492d7dac3 h1:LMLX+LgTNWpfvCBdFebv6EsYotImrt/Ppc5cXIriCSo= +github.com/google/pprof v0.0.0-20260802141513-ef3492d7dac3/go.mod h1:jl5iWTm0/hd5PjEYEOuwAJ57L/CibdZfrqZ5XA5GrCk= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= +github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= +github.com/mattn/go-isatty v0.0.24 h1:tGZZoVgT/KiqK1c8ocVLeDS8BSWMRd47J3Lbz7vsReI= +github.com/mattn/go-isatty v0.0.24/go.mod h1:nMCL3Zebbrt45jsMDgnfIwz6ydEQApk5oEI3HqDio6A= +github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w= +github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/mod v0.41.0 h1:qJmnOUb4YB+FsEuM3HcWucdZASCPGhsX6uljO6pog0c= +golang.org/x/mod v0.41.0/go.mod h1:Ek9pY8RKWXwsWvd3rQiHYtMqkjSUV+s1Rj7j4H5Ur6o= +golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk= +golang.org/x/sync v0.23.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/tools v0.50.0 h1:c2ifzfcuY7L90lZ2aKd8S4K2NpASF08SZx9ZuJkHmSU= +golang.org/x/tools v0.50.0/go.mod h1:7ulVMw3831Mwi5EZD6RomGyffr4VFjuNYXf2BbCEAV0= +modernc.org/cc/v4 v4.29.7 h1:q+NXGJ0bK3b4TXFYQQVr9pYETGnmwFWkrUzJnMya/Tg= +modernc.org/cc/v4 v4.29.7/go.mod h1:OnovgIhbbMXMu1aISnJ0wvVD1KnW+cAUJkIrAWh+kVI= +modernc.org/ccgo/v4 v4.36.1 h1:ZNIUZAryN0UgnJwtyxrdEzcFc3yD4Cu4AzjfPXsLsIE= +modernc.org/ccgo/v4 v4.36.1/go.mod h1:rrtGc2QkS239nYb/mQNuBMyjq3/y3ZXWbBjPoV3wqzA= +modernc.org/fileutil v1.4.0 h1:j6ZzNTftVS054gi281TyLjHPp6CPHr2KCxEXjEbD6SM= +modernc.org/fileutil v1.4.0/go.mod h1:EqdKFDxiByqxLk8ozOxObDSfcVOv/54xDs/DUHdvCUU= +modernc.org/gc/v2 v2.6.5 h1:nyqdV8q46KvTpZlsw66kWqwXRHdjIlJOhG6kxiV/9xI= +modernc.org/gc/v2 v2.6.5/go.mod h1:YgIahr1ypgfe7chRuJi2gD7DBQiKSLMPgBQe9oIiito= +modernc.org/gc/v3 v3.1.5 h1:21ldfPfRYE31Tb7B3mwAK8gy1AxP4+dKjrOQPfqakoc= +modernc.org/gc/v3 v3.1.5/go.mod h1:HFK/6AGESC7Ex+EZJhJ2Gni6cTaYpSMmU/cT9RmlfYY= +modernc.org/goabi0 v0.2.0 h1:HvEowk7LxcPd0eq6mVOAEMai46V+i7Jrj13t4AzuNks= +modernc.org/goabi0 v0.2.0/go.mod h1:CEFRnnJhKvWT1c1JTI3Avm+tgOWbkOu5oPA8eH8LnMI= +modernc.org/libc v1.77.1 h1:Ct8j47QtiZ1Enj2DtFXQtUqrPCAjdCmPjtCuvrYQ0Hs= +modernc.org/libc v1.77.1/go.mod h1:87/pZ4L6nD1zqW4nItuS12YO7hN1igAah34xjnQo/W0= +modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU= +modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg= +modernc.org/memory v1.12.1 h1:nFMiWrpStgZczNl6XI9GnIk/rWhYIyHGUaR04pGbp9g= +modernc.org/memory v1.12.1/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw= +modernc.org/opt v0.2.0 h1:tGyef5ApycA7FSEOMraay9SaTk5zmbx7Tu+cJs4QKZg= +modernc.org/opt v0.2.0/go.mod h1:03fq9lsNfvkYSfxrfUhZCWPk1lm4cq4N+Bh//bEtgns= +modernc.org/sortutil v1.2.1 h1:+xyoGf15mM3NMlPDnFqrteY07klSFxLElE2PVuWIJ7w= +modernc.org/sortutil v1.2.1/go.mod h1:7ZI3a3REbai7gzCLcotuw9AC4VZVpYMjDzETGsSMqJE= +modernc.org/sqlite v1.60.1 h1:/blz53O951KWFOso4QQvEs/Fq6cDBKLtMVrYNSeJVKw= +modernc.org/sqlite v1.60.1/go.mod h1:1dIoEagfDE72QytD5scH1lxARtaUgKgHC/NuApA27r0= +modernc.org/strutil v1.2.1 h1:UneZBkQA+DX2Rp35KcM69cSsNES9ly8mQWD71HKlOA0= +modernc.org/strutil v1.2.1/go.mod h1:EHkiggD70koQxjVdSBM3JKM7k6L0FbGE5eymy9i3B9A= +modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y= +modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM= diff --git a/internal/auth/auth.go b/internal/auth/auth.go new file mode 100644 index 0000000..d3b9a5a --- /dev/null +++ b/internal/auth/auth.go @@ -0,0 +1,173 @@ +// Package auth provides password hashing, TOTP, session tokens and login throttling. +package auth + +import ( + "crypto/hmac" + "crypto/rand" + "crypto/sha1" + "crypto/sha256" + "crypto/subtle" + "encoding/base32" + "encoding/base64" + "encoding/binary" + "encoding/hex" + "errors" + "fmt" + "strings" + "sync" + "time" + + "golang.org/x/crypto/argon2" +) + +// Argon2id parameters: ~64 MiB, sized for a Raspberry Pi 4/5. +const ( + argonTime = 2 + argonMemory = 64 * 1024 + argonThreads = 2 + argonKeyLen = 32 +) + +var b64 = base64.RawStdEncoding + +func HashPassword(pw string) (string, error) { + salt := make([]byte, 16) + if _, err := rand.Read(salt); err != nil { + return "", err + } + k := argon2.IDKey([]byte(pw), salt, argonTime, argonMemory, argonThreads, argonKeyLen) + return fmt.Sprintf("$argon2id$v=19$m=%d,t=%d,p=%d$%s$%s", argonMemory, argonTime, argonThreads, b64.EncodeToString(salt), b64.EncodeToString(k)), nil +} + +func CheckPassword(pw, encoded string) bool { + p := strings.Split(encoded, "$") + if len(p) != 6 || p[1] != "argon2id" { + return false + } + var m uint32 + var t uint32 + var th uint8 + if _, err := fmt.Sscanf(p[3], "m=%d,t=%d,p=%d", &m, &t, &th); err != nil || m > 1<<20 || t > 10 { + return false + } + salt, err1 := b64.DecodeString(p[4]) + want, err2 := b64.DecodeString(p[5]) + if err1 != nil || err2 != nil { + return false + } + got := argon2.IDKey([]byte(pw), salt, t, m, th, uint32(len(want))) + return subtle.ConstantTimeCompare(got, want) == 1 +} + +// PasswordPolicy: length is what matters; no composition rules. +func PasswordPolicy(pw string) error { + if len(pw) < 12 { + return errors.New("password must be at least 12 characters") + } + if len(pw) > 256 { + return errors.New("password too long") + } + return nil +} + +// RandomToken returns n random bytes hex-encoded. +func RandomToken(n int) string { + b := make([]byte, n) + rand.Read(b) + return hex.EncodeToString(b) +} + +// TokenHash is what we store for session tokens. +func TokenHash(tok string) string { + h := sha256.Sum256([]byte(tok)) + return hex.EncodeToString(h[:]) +} + +// --- TOTP (RFC 6238, SHA-1, 30 s, 6 digits) --- + +func NewTOTPSecret() string { + b := make([]byte, 20) + rand.Read(b) + return base32.StdEncoding.WithPadding(base32.NoPadding).EncodeToString(b) +} + +func totpAt(secret string, counter uint64) (string, error) { + key, err := base32.StdEncoding.WithPadding(base32.NoPadding).DecodeString(strings.ToUpper(secret)) + if err != nil { + return "", err + } + var msg [8]byte + binary.BigEndian.PutUint64(msg[:], counter) + m := hmac.New(sha1.New, key) + m.Write(msg[:]) + sum := m.Sum(nil) + off := sum[len(sum)-1] & 0xf + v := binary.BigEndian.Uint32(sum[off:]) & 0x7fffffff + return fmt.Sprintf("%06d", v%1_000_000), nil +} + +// CheckTOTP accepts the current code and one step either side for clock skew. +func CheckTOTP(secret, code string, now time.Time) bool { + c := uint64(now.Unix() / 30) + for _, d := range []int64{0, -1, 1} { + want, err := totpAt(secret, uint64(int64(c)+d)) + if err == nil && subtle.ConstantTimeCompare([]byte(want), []byte(code)) == 1 { + return true + } + } + return false +} + +func TOTPURI(issuer, user, secret string) string { + return fmt.Sprintf("otpauth://totp/%s:%s?secret=%s&issuer=%s&algorithm=SHA1&digits=6&period=30", issuer, user, secret, issuer) +} + +// --- login throttling --- + +// Limiter applies exponential lockout per key (IP and username tracked separately). +type Limiter struct { + mu sync.Mutex + m map[string]*entry +} + +type entry struct { + fails int + until time.Time +} + +func NewLimiter() *Limiter { return &Limiter{m: map[string]*entry{}} } + +// Allowed reports whether key may attempt now, and the wait otherwise. +func (l *Limiter) Allowed(key string, now time.Time) (bool, time.Duration) { + l.mu.Lock() + defer l.mu.Unlock() + e := l.m[key] + if e == nil || now.After(e.until) { + return true, 0 + } + return false, e.until.Sub(now) +} + +func (l *Limiter) Fail(key string, now time.Time) { + l.mu.Lock() + defer l.mu.Unlock() + if len(l.m) > 10000 { // ponytail: crude bound on memory; LRU if this ever matters + l.m = map[string]*entry{} + } + e := l.m[key] + if e == nil { + e = &entry{} + l.m[key] = e + } + e.fails++ + if e.fails >= 5 { + d := time.Duration(1< 8-digit 94287082 at T=59; 6-digit = 287082. + secret := "GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ" + if !CheckTOTP(secret, "287082", time.Unix(59, 0)) { + t.Fatal("RFC vector failed") + } + if CheckTOTP(secret, "287082", time.Unix(59+300, 0)) { + t.Fatal("stale code accepted") + } +} + +func TestLimiter(t *testing.T) { + l := NewLimiter() + now := time.Unix(1000, 0) + for i := 0; i < 5; i++ { + l.Fail("ip", now) + } + if ok, _ := l.Allowed("ip", now); ok { + t.Fatal("should be locked") + } + if ok, _ := l.Allowed("ip", now.Add(2*time.Second)); !ok { + t.Fatal("lock should expire") + } +} diff --git a/internal/capture/engine.go b/internal/capture/engine.go new file mode 100644 index 0000000..719f6ad --- /dev/null +++ b/internal/capture/engine.go @@ -0,0 +1,824 @@ +// Package capture runs dumpcap, stores original packets as pcapng segments, +// indexes per-packet summaries in SQLite and serves on-demand dissection. +package capture + +import ( + "bufio" + "bytes" + "context" + "crypto/sha256" + "database/sql" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "hash" + "io" + "log/slog" + "net" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "syscall" + "time" + + "pilens.dev/pilens/internal/auth" + "pilens.dev/pilens/internal/config" + "pilens.dev/pilens/internal/pcapfile" + "pilens.dev/pilens/internal/store" +) + +type Limits struct { + MaxBytes int64 `json:"max_bytes"` // total bytes of capture files + ReserveBytes int64 `json:"reserve_bytes"` // free space that must remain on the filesystem + SegmentBytes int64 `json:"segment_bytes"` // rotate segment files at this size + ImportBytes int64 `json:"import_bytes"` // largest accepted upload +} + +func DefaultLimits() Limits { + return Limits{MaxBytes: 8 << 30, ReserveBytes: 1 << 30, SegmentBytes: 64 << 20, ImportBytes: 1 << 30} +} + +type Engine struct { + DB *store.Store + Dir string // root of capture storage + Tools *Tools + Hub *Hub + Version string + + limMu sync.Mutex + limits Limits + + mu sync.Mutex + live *live + + fcMu sync.Mutex + fcache map[string]*filterEntry +} + +func (e *Engine) Limits() Limits { e.limMu.Lock(); defer e.limMu.Unlock(); return e.limits } +func (e *Engine) SetLimits(l Limits) { e.limMu.Lock(); e.limits = l; e.limMu.Unlock() } + +// Session is the API view of a capture session. +type Session struct { + ID string `json:"id"` + Name string `json:"name"` + Description string `json:"description"` + Kind string `json:"kind"` + State string `json:"state"` + Interfaces []string `json:"interfaces"` + CaptureFilter string `json:"capture_filter"` + StartedAt int64 `json:"started_at"` + EndedAt int64 `json:"ended_at"` + Packets int64 `json:"packets"` + Bytes int64 `json:"bytes"` + Drops string `json:"drops"` + Error string `json:"error"` + Preserved bool `json:"preserved"` + Tags string `json:"tags"` + Notes string `json:"notes"` + RingSegments int `json:"ring_segments"` + Meta json.RawMessage `json:"meta"` + Paused bool `json:"paused"` + Lag int64 `json:"ingest_lag"` // packets written but not yet summarised +} + +type Segment struct { + Seq int `json:"seq"` + File string `json:"file"` + Format string `json:"format"` + Base int64 `json:"base"` + Count int64 `json:"count"` + Bytes int64 `json:"bytes"` + SHA256 string `json:"sha256"` + Complete bool `json:"complete"` + Deleted bool `json:"deleted"` +} + +func newID() string { + return time.Now().UTC().Format("20060102T150405Z") + "-" + auth.RandomToken(3) +} + +const sessCols = `id,name,description,kind,state,interfaces,capture_filter,started_at,ended_at,packets,bytes,drops,error,preserved,tags,notes,ring_segments,meta` + +func scanSession(sc interface{ Scan(...any) error }) (Session, error) { + var s Session + var ifs, meta string + err := sc.Scan(&s.ID, &s.Name, &s.Description, &s.Kind, &s.State, &ifs, &s.CaptureFilter, &s.StartedAt, &s.EndedAt, &s.Packets, &s.Bytes, &s.Drops, &s.Error, &s.Preserved, &s.Tags, &s.Notes, &s.RingSegments, &meta) + json.Unmarshal([]byte(ifs), &s.Interfaces) + s.Meta = json.RawMessage(meta) + return s, err +} + +func (e *Engine) withLive(s Session) Session { + e.mu.Lock() + l := e.live + e.mu.Unlock() + if l != nil && l.id == s.ID { + s.Packets = l.packets.Load() + s.Bytes = l.bytes.Load() + s.Paused = l.paused.Load() + s.Lag = s.Packets - l.summarised.Load() + } + return s +} + +func (e *Engine) Sessions() ([]Session, error) { + rows, err := e.DB.Query(`SELECT ` + sessCols + ` FROM capture_sessions ORDER BY started_at DESC`) + if err != nil { + return nil, err + } + defer rows.Close() + out := []Session{} + for rows.Next() { + s, err := scanSession(rows) + if err != nil { + return nil, err + } + out = append(out, e.withLive(s)) + } + return out, rows.Err() +} + +func (e *Engine) Get(id string) (Session, error) { + s, err := scanSession(e.DB.QueryRow(`SELECT `+sessCols+` FROM capture_sessions WHERE id=?`, id)) + if err == sql.ErrNoRows { + return s, ErrNotFound + } + return e.withLive(s), err +} + +var ErrNotFound = errors.New("not found") + +func (e *Engine) Segments(id string) ([]Segment, error) { + rows, err := e.DB.Query(`SELECT seq,file,format,base,count,bytes,sha256,complete,deleted FROM segments WHERE session_id=? ORDER BY seq`, id) + if err != nil { + return nil, err + } + defer rows.Close() + var out []Segment + for rows.Next() { + var s Segment + if err := rows.Scan(&s.Seq, &s.File, &s.Format, &s.Base, &s.Count, &s.Bytes, &s.SHA256, &s.Complete, &s.Deleted); err != nil { + return nil, err + } + out = append(out, s) + } + return out, rows.Err() +} + +func (e *Engine) sessionDir(id string) string { return filepath.Join(e.Dir, id) } + +// Update edits analyst metadata. +func (e *Engine) Update(id, name, desc, tags, notes string, preserved bool) error { + r, err := e.DB.Exec(`UPDATE capture_sessions SET name=?,description=?,tags=?,notes=?,preserved=? WHERE id=?`, name, desc, tags, notes, preserved, id) + if err != nil { + return err + } + if n, _ := r.RowsAffected(); n == 0 { + return ErrNotFound + } + return nil +} + +func (e *Engine) Delete(id string) error { + e.mu.Lock() + running := e.live != nil && e.live.id == id + e.mu.Unlock() + if running { + return errors.New("stop the capture before deleting it") + } + if _, err := e.Get(id); err != nil { + return err + } + if err := os.RemoveAll(e.sessionDir(id)); err != nil { + return err + } + e.DB.Exec(`DELETE FROM packets WHERE session_id=?`, id) + _, err := e.DB.Exec(`DELETE FROM capture_sessions WHERE id=?`, id) + e.dropFilterCache(id) + return err +} + +// ---------------------------------------------------------------- live capture + +type StartOptions struct { + Name string `json:"name"` + Interfaces []string `json:"interfaces"` + CaptureFilter string `json:"capture_filter"` + RingSegments int `json:"ring_segments"` +} + +type live struct { + id string + cmd *exec.Cmd + done chan struct{} + stopping atomic.Bool + paused atomic.Bool + discarded atomic.Int64 // packets discarded while paused + packets atomic.Int64 + bytes atomic.Int64 + summarised atomic.Int64 + stderr limitedBuffer + + mu sync.Mutex // guards seg + seg *segWriter + wg sync.WaitGroup // summarisers +} + +type segWriter struct { + seq int + base int64 + path string + f *os.File + bw *bufio.Writer + h hash.Hash + size int64 + count int64 + offs []int64 + + feed chan []byte + feedLost bool +} + +type limitedBuffer struct { + mu sync.Mutex + b bytes.Buffer +} + +func (l *limitedBuffer) Write(p []byte) (int, error) { + l.mu.Lock() + defer l.mu.Unlock() + if l.b.Len() < 64<<10 { + l.b.Write(p) + } + return len(p), nil +} + +func (l *limitedBuffer) String() string { l.mu.Lock(); defer l.mu.Unlock(); return l.b.String() } + +func clockSynced() string { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + out, err := exec.CommandContext(ctx, "timedatectl", "show", "-p", "NTPSynchronized", "--value").Output() + if err != nil { + return "unknown" + } + return strings.TrimSpace(string(out)) +} + +func (e *Engine) meta(extra map[string]any) string { + host, _ := os.Hostname() + m := map[string]any{ + "pilens_version": e.Version, + "tshark_version": e.Tools.Version(), + "clock_synchronised": clockSynced(), + "hostname": host, + "recorded_utc": time.Now().UTC().Format(time.RFC3339Nano), + } + for k, v := range extra { + m[k] = v + } + b, _ := json.Marshal(m) + return string(b) +} + +func (e *Engine) Live() (Session, bool) { + e.mu.Lock() + l := e.live + e.mu.Unlock() + if l == nil { + return Session{}, false + } + s, err := e.Get(l.id) + return s, err == nil +} + +func (e *Engine) Start(ctx context.Context, o StartOptions) (Session, error) { + if len(o.Interfaces) == 0 || len(o.Interfaces) > 8 { + return Session{}, errors.New("choose 1-8 interfaces") + } + for _, i := range o.Interfaces { + if i == "any" { + continue + } + if !config.ValidInterface(i) { + return Session{}, fmt.Errorf("invalid interface %q", i) + } + if _, err := net.InterfaceByName(i); err != nil { + return Session{}, fmt.Errorf("interface %q not found", i) + } + } + if o.CaptureFilter != "" { + if err := e.Tools.ValidateCaptureFilter(ctx, o.Interfaces[0], o.CaptureFilter); err != nil { + return Session{}, fmt.Errorf("capture filter: %v", err) + } + } + if o.RingSegments < 0 || o.RingSegments > 100000 { + return Session{}, errors.New("invalid ring size") + } + if strings.TrimSpace(o.Name) == "" { + o.Name = "Capture " + time.Now().Format("2006-01-02 15:04") + } + if len(o.Name) > 200 { + return Session{}, errors.New("name too long") + } + + e.mu.Lock() + defer e.mu.Unlock() + if e.live != nil { + return Session{}, errors.New("a capture is already running") + } + if err := e.enforceStorage(""); err != nil { + return Session{}, err + } + id := newID() + dir := e.sessionDir(id) + if err := os.MkdirAll(dir, 0o750); err != nil { + return Session{}, err + } + ifs, _ := json.Marshal(o.Interfaces) + lim := e.Limits() + meta := e.meta(map[string]any{"segment_bytes": lim.SegmentBytes, "snaplen": "dumpcap default (full)"}) + _, err := e.DB.Exec(`INSERT INTO capture_sessions(id,name,kind,state,interfaces,capture_filter,started_at,dir,meta,ring_segments) VALUES(?,?,?,?,?,?,?,?,?,?)`, + id, o.Name, "live", "running", string(ifs), o.CaptureFilter, time.Now().UnixMilli(), dir, meta, o.RingSegments) + if err != nil { + return Session{}, err + } + + args := []string{"-q", "-n", "-w", "-"} + for _, i := range o.Interfaces { + args = append(args, "-i", i) + if o.CaptureFilter != "" { + args = append(args, "-f", o.CaptureFilter) // applies to the preceding -i + } + } + cmd := exec.Command(e.Tools.Dumpcap, args...) + l := &live{id: id, cmd: cmd, done: make(chan struct{})} + cmd.Stderr = &l.stderr + stdout, err := cmd.StdoutPipe() + if err != nil { + return Session{}, err + } + if err := cmd.Start(); err != nil { + e.finishSession(id, "failed", "dumpcap failed to start: "+err.Error(), "") + return Session{}, err + } + e.live = l + go e.runLive(l, stdout) + slog.Info("capture started", "session", id, "interfaces", o.Interfaces) + s, err := scanSession(e.DB.QueryRow(`SELECT `+sessCols+` FROM capture_sessions WHERE id=?`, id)) + return s, err // e.mu is held: must not call e.Get (withLive locks it) +} + +func (e *Engine) SetPaused(p bool) error { + e.mu.Lock() + l := e.live + e.mu.Unlock() + if l == nil { + return errors.New("no capture running") + } + if l.paused.Swap(p) != p { + ev := "resumed" + if p { + ev = "paused" + } + e.DB.Exec(`UPDATE capture_sessions SET meta=json_insert(meta,'$.pause_log[#]',json_object('event',?,'utc',?)) WHERE id=?`, ev, time.Now().UTC().Format(time.RFC3339Nano), l.id) + } + return nil +} + +func (e *Engine) Stop() error { + e.mu.Lock() + l := e.live + e.mu.Unlock() + if l == nil { + return errors.New("no capture running") + } + l.stopping.Store(true) + l.cmd.Process.Signal(syscall.SIGINT) // dumpcap flushes and reports statistics + select { + case <-l.done: + case <-time.After(30 * time.Second): + l.cmd.Process.Kill() + <-l.done + } + return nil +} + +func (e *Engine) runLive(l *live, stdout io.Reader) { + defer close(l.done) + rowsC := make(chan []Row, 64) + insDone := make(chan struct{}) + go func() { e.inserter(l, rowsC); close(insDone) }() + + // Periodic flush so on-demand reads of the active segment see recent packets, + // and status events for the UI. + stopTick := make(chan struct{}) + go func() { + t := time.NewTicker(time.Second) + defer t.Stop() + for { + select { + case <-stopTick: + return + case <-t.C: + l.mu.Lock() + if l.seg != nil { + l.seg.bw.Flush() + } + l.mu.Unlock() + e.publishStatus(l) + } + } + }() + + var failure string + br := pcapfile.NewBlockReader(bufio.NewReaderSize(stdout, 1<<20)) + var hdrs [][]byte + for { + b, err := br.Next() + if err != nil { + if err != io.EOF && !l.stopping.Load() { + failure = "capture stream error: " + err.Error() + } + break + } + switch b.Type { + case pcapfile.BlockSHB: + hdrs = [][]byte{b.Raw} + case pcapfile.BlockIDB: + hdrs = append(hdrs, b.Raw) + } + if b.IsPacket() && l.paused.Load() { + l.discarded.Add(1) + continue + } + if err := e.writeBlock(l, b, hdrs, rowsC); err != nil { + failure = err.Error() + l.stopping.Store(true) + l.cmd.Process.Signal(syscall.SIGINT) + io.Copy(io.Discard, stdout) + break + } + } + waitErr := l.cmd.Wait() + l.mu.Lock() + e.closeSegment(l, rowsC) + l.mu.Unlock() + l.wg.Wait() + close(rowsC) + <-insDone + close(stopTick) + + state := "stopped" + if failure == "" && !l.stopping.Load() { + failure = "dumpcap exited unexpectedly" + if waitErr != nil { + failure += ": " + waitErr.Error() + } + failure += ": " + strings.TrimSpace(l.stderr.String()) + } + if failure != "" { + state = "failed" + } + e.DB.Exec(`UPDATE capture_sessions SET packets=?,bytes=?,meta=json_set(meta,'$.discarded_while_paused',?) WHERE id=?`, l.packets.Load(), l.bytes.Load(), l.discarded.Load(), l.id) + e.finishSession(l.id, state, failure, dropLines(l.stderr.String())) + e.mu.Lock() + e.live = nil + e.mu.Unlock() + e.publishStatus(l) + slog.Info("capture finished", "session", l.id, "state", state, "packets", l.packets.Load(), "error", failure) +} + +// dropLines keeps dumpcap's per-interface "Packets received/dropped" report verbatim. +func dropLines(s string) string { + var out []string + for _, l := range strings.Split(s, "\n") { + if strings.Contains(l, "received/dropped") { + out = append(out, strings.TrimSpace(l)) + } + } + return strings.Join(out, "\n") +} + +func (e *Engine) finishSession(id, state, errMsg, drops string) { + e.DB.Exec(`UPDATE capture_sessions SET state=?,error=?,drops=?,ended_at=? WHERE id=?`, state, errMsg, drops, time.Now().UnixMilli(), id) + e.dropFilterCache(id) +} + +// writeBlock is called only from the splitter goroutine. +func (e *Engine) writeBlock(l *live, b pcapfile.Block, hdrs [][]byte, rowsC chan []Row) error { + l.mu.Lock() + defer l.mu.Unlock() + lim := e.Limits() + if b.IsPacket() && l.seg != nil && l.seg.size >= lim.SegmentBytes { + e.closeSegment(l, rowsC) + if err := e.enforceStorage(l.id); err != nil { + return err + } + } + if l.seg == nil { + if !b.IsPacket() { + return nil // headers are written when the first packet opens a segment + } + if err := e.openSegment(l, hdrs, rowsC); err != nil { + return err + } + } + s := l.seg + if b.IsPacket() { + s.offs = append(s.offs, s.size) + s.count++ + l.packets.Add(1) + if d, err := pcapfile.PacketData(b); err == nil { + l.bytes.Add(int64(len(d))) + } + } + if _, err := s.bw.Write(b.Raw); err != nil { + return fmt.Errorf("writing capture file: %w", err) + } + s.size += int64(len(b.Raw)) + s.sendFeed(b.Raw) + return nil +} + +func (s *segWriter) sendFeed(raw []byte) { + if s.feedLost { + return + } + select { + case s.feed <- raw: + default: + // tshark cannot keep up. Never block capture: stop live summaries for this + // segment; they are regenerated from the file when the segment closes. + s.feedLost = true + close(s.feed) + } +} + +func (e *Engine) openSegment(l *live, hdrs [][]byte, rowsC chan []Row) error { + var seq int + var base int64 + e.DB.QueryRow(`SELECT COALESCE(MAX(seq),0)+1 FROM segments WHERE session_id=?`, l.id).Scan(&seq) + base = l.packets.Load() + 1 + name := fmt.Sprintf("seg-%06d.pcapng", seq) + path := filepath.Join(e.sessionDir(l.id), name) + f, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o640) + if err != nil { + return err + } + h := sha256.New() + s := &segWriter{seq: seq, base: base, path: path, f: f, h: h, bw: bufio.NewWriterSize(io.MultiWriter(f, h), 1<<20), feed: make(chan []byte, 2048)} + if _, err := e.DB.Exec(`INSERT INTO segments(session_id,seq,file,format,base) VALUES(?,?,?,?,?)`, l.id, seq, name, "pcapng", base); err != nil { + f.Close() + return err + } + for _, hb := range hdrs { + s.bw.Write(hb) + s.size += int64(len(hb)) + s.sendFeed(hb) + } + l.seg = s + l.wg.Add(1) + go e.summarise(l, s, rowsC) + return nil +} + +// closeSegment finalises the current segment. Caller holds l.mu. +func (e *Engine) closeSegment(l *live, rowsC chan []Row) { + s := l.seg + if s == nil { + return + } + l.seg = nil + s.bw.Flush() + s.f.Sync() + s.f.Close() + if !s.feedLost { + close(s.feed) + } + sum := hex.EncodeToString(s.h.Sum(nil)) + e.DB.Exec(`UPDATE segments SET count=?,bytes=?,sha256=?,complete=1 WHERE session_id=? AND seq=?`, s.count, s.size, sum, l.id, s.seq) + e.DB.Exec(`UPDATE capture_sessions SET packets=?,bytes=? WHERE id=?`, l.packets.Load(), l.bytes.Load(), l.id) + e.applyRing(l.id) +} + +// summarise runs one tshark per segment fed from the splitter, then backfills +// from the completed file if the live feed was cut short. +func (e *Engine) summarise(l *live, s *segWriter, rowsC chan []Row) { + defer l.wg.Done() + offAt := func(local int64) int64 { + l.mu.Lock() + defer l.mu.Unlock() + if local < 1 || local > int64(len(s.offs)) { + return -1 + } + return s.offs[local-1] + } + var got int64 + emit := func(batch *[]Row, r Row) { + r.Off = offAt(r.Num) + if r.Off < 0 { + return + } + r.Num = s.base + r.Num - 1 + r.Seg = s.seq + got++ + *batch = append(*batch, r) + if len(*batch) >= 500 { + rowsC <- *batch + *batch = nil + } + } + + ctx := context.Background() + cmd := e.Tools.Command(ctx, 0, e.Tools.SummaryArgs("-")...) + stdin, _ := cmd.StdinPipe() + stdout, _ := cmd.StdoutPipe() + if err := cmd.Start(); err != nil { + slog.Error("tshark start", "err", err) + for range s.feed { + } + } else { + go func() { + for b := range s.feed { + if _, err := stdin.Write(b); err != nil { + break + } + } + stdin.Close() + for range s.feed { // drain if tshark died + } + }() + var batch []Row + lines := make(chan Row, 1024) + go func() { ReadSummaries(stdout, func(r Row) { lines <- r }); close(lines) }() + tick := time.NewTicker(300 * time.Millisecond) + loop: + for { + select { + case r, ok := <-lines: + if !ok { + break loop + } + emit(&batch, r) + case <-tick.C: + if len(batch) > 0 { + rowsC <- batch + batch = nil + } + } + } + tick.Stop() + if len(batch) > 0 { + rowsC <- batch + } + cmd.Wait() + } + + // Backfill from the finished file if live summaries are incomplete. + l.mu.Lock() + closed := l.seg != s + count := s.count + l.mu.Unlock() + if closed && got < count { + slog.Info("backfilling summaries", "session", l.id, "segment", s.seq, "have", got, "want", count) + cmd := e.Tools.Command(ctx, 3600, e.Tools.SummaryArgs(s.path)...) + out, _ := cmd.StdoutPipe() + if cmd.Start() == nil { + var batch []Row + ReadSummaries(out, func(r Row) { emit(&batch, r) }) + if len(batch) > 0 { + rowsC <- batch + } + cmd.Wait() + } + e.Hub.Publish(Event{Type: "resync", Data: []byte(`{"session":"` + l.id + `"}`)}) + } +} + +func insertRows(tx *sql.Tx, session string, rows []Row) error { + st, err := tx.Prepare(`INSERT OR IGNORE INTO packets(session_id,num,seg,off,ts,src,dst,proto,len,info,sport,dport,stream,tcpflags,ttl,vlan,color) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`) + if err != nil { + return err + } + defer st.Close() + for _, r := range rows { + if _, err := st.Exec(session, r.Num, r.Seg, r.Off, r.TS, r.Src, r.Dst, r.Proto, r.Len, r.Info, r.SPort, r.DPort, r.Stream, r.TCPFlags, r.TTL, r.VLAN, r.Color); err != nil { + return err + } + } + return nil +} + +func (e *Engine) inserter(l *live, rowsC chan []Row) { + for batch := range rowsC { + tx, err := e.DB.Begin() + if err != nil { + slog.Error("index insert", "err", err) + continue + } + if err := insertRows(tx, l.id, batch); err != nil { + tx.Rollback() + slog.Error("index insert", "err", err) + continue + } + tx.Commit() + l.summarised.Add(int64(len(batch))) + data, _ := json.Marshal(map[string]any{"session": l.id, "rows": batch}) + e.Hub.Publish(Event{Type: "packets", Data: data}) + } +} + +func (e *Engine) publishStatus(l *live) { + s, err := e.Get(l.id) + if err != nil { + return + } + data, _ := json.Marshal(s) + e.Hub.Publish(Event{Type: "capture", Data: data}) +} + +// ---------------------------------------------------------------- storage + +func (e *Engine) usedBytes() int64 { + var n int64 + e.DB.QueryRow(`SELECT COALESCE(SUM(bytes),0) FROM segments WHERE deleted=0`).Scan(&n) + return n +} + +func freeBytes(dir string) int64 { + var st syscall.Statfs_t + if syscall.Statfs(dir, &st) != nil { + return 0 + } + return int64(st.Bavail) * int64(st.Bsize) +} + +type StorageInfo struct { + Used int64 `json:"used"` + Free int64 `json:"free"` + Limits Limits `json:"limits"` + Dir string `json:"dir"` + LowWarning bool `json:"low_warning"` +} + +func (e *Engine) Storage() StorageInfo { + lim := e.Limits() + i := StorageInfo{Used: e.usedBytes(), Free: freeBytes(e.Dir), Limits: lim, Dir: e.Dir} + i.LowWarning = i.Free < 2*lim.ReserveBytes || i.Used > lim.MaxBytes*9/10 + return i +} + +// enforceStorage deletes the oldest unpreserved, finished sessions until within +// limits. If that is not possible it returns an error and nothing is overwritten. +func (e *Engine) enforceStorage(activeID string) error { + lim := e.Limits() + for { + used, free := e.usedBytes(), freeBytes(e.Dir) + if used+lim.SegmentBytes <= lim.MaxBytes && free-lim.SegmentBytes >= lim.ReserveBytes { + return nil + } + var victim string + e.DB.QueryRow(`SELECT id FROM capture_sessions WHERE preserved=0 AND id!=? AND state NOT IN ('running','importing') ORDER BY started_at LIMIT 1`, activeID).Scan(&victim) + if victim == "" { + return fmt.Errorf("storage limit reached (used %d of %d bytes, %d free, reserve %d) and no unpreserved sessions remain to delete", used, lim.MaxBytes, free, lim.ReserveBytes) + } + slog.Warn("storage limit: deleting oldest unpreserved session", "session", victim) + e.DB.Audit("system", "capture.retention_delete", victim, "") + if err := os.RemoveAll(e.sessionDir(victim)); err != nil { + return err + } + e.DB.Exec(`DELETE FROM packets WHERE session_id=?`, victim) + e.DB.Exec(`DELETE FROM capture_sessions WHERE id=?`, victim) + e.dropFilterCache(victim) + } +} + +func (e *Engine) applyRing(id string) { + var ring int + var preserved bool + e.DB.QueryRow(`SELECT ring_segments,preserved FROM capture_sessions WHERE id=?`, id).Scan(&ring, &preserved) + if ring <= 0 || preserved { + return + } + segs, _ := e.Segments(id) + var live []Segment + for _, s := range segs { + if s.Complete && !s.Deleted { + live = append(live, s) + } + } + for len(live) > ring { + s := live[0] + live = live[1:] + os.Remove(filepath.Join(e.sessionDir(id), s.File)) + e.DB.Exec(`UPDATE segments SET deleted=1 WHERE session_id=? AND seq=?`, id, s.Seq) + e.DB.Exec(`DELETE FROM packets WHERE session_id=? AND num>=? AND num 0 { + if ev := <-sub.C; ev.Type == "packets" { + got++ + } + } + if got == 0 && sub.Dropped.Load() == 0 { + t.Fatal("no live packet events") + } +} diff --git a/internal/capture/hub.go b/internal/capture/hub.go new file mode 100644 index 0000000..7e6654e --- /dev/null +++ b/internal/capture/hub.go @@ -0,0 +1,53 @@ +package capture + +import ( + "sync" + "sync/atomic" +) + +// Hub fans out live events to browser subscribers. A subscriber that cannot +// keep up loses events (counted) and is told to resync; capture never waits. +type Hub struct { + mu sync.Mutex + subs map[*Sub]struct{} +} + +type Event struct { + Type string // SSE event name + Data []byte // JSON +} + +type Sub struct { + C chan Event + Dropped atomic.Int64 + Resync atomic.Bool +} + +func NewHub() *Hub { return &Hub{subs: map[*Sub]struct{}{}} } + +func (h *Hub) Subscribe() *Sub { + s := &Sub{C: make(chan Event, 64)} + h.mu.Lock() + h.subs[s] = struct{}{} + h.mu.Unlock() + return s +} + +func (h *Hub) Unsubscribe(s *Sub) { + h.mu.Lock() + delete(h.subs, s) + h.mu.Unlock() +} + +func (h *Hub) Publish(e Event) { + h.mu.Lock() + defer h.mu.Unlock() + for s := range h.subs { + select { + case s.C <- e: + default: + s.Dropped.Add(1) + s.Resync.Store(true) + } + } +} diff --git a/internal/capture/pdml.go b/internal/capture/pdml.go new file mode 100644 index 0000000..76c9c26 --- /dev/null +++ b/internal/capture/pdml.go @@ -0,0 +1,107 @@ +package capture + +import ( + "bytes" + "encoding/hex" + "encoding/xml" + "errors" + "strconv" +) + +// Node is one line of the protocol tree sent to the browser. +type Node struct { + Label string `json:"label"` + Field string `json:"field,omitempty"` + Pos int `json:"pos"` + Size int `json:"size"` + Verified bool `json:"verified"` // bytes at pos..pos+size are proven to belong to this field + Children []Node `json:"children,omitempty"` +} + +type pdmlNode struct { + XMLName xml.Name + Name string `xml:"name,attr"` + Show string `xml:"show,attr"` + Showname string `xml:"showname,attr"` + Value string `xml:"value,attr"` + Unmasked string `xml:"unmaskedvalue,attr"` + Pos string `xml:"pos,attr"` + Size string `xml:"size,attr"` + Hide string `xml:"hide,attr"` + Children []pdmlNode `xml:",any"` +} + +// ParsePDML converts tshark PDML for a single packet into a tree. frame holds +// the original captured bytes, used to verify each field's byte range. +func ParsePDML(b []byte, frame []byte) ([]Node, error) { + var doc struct { + Packets []struct { + Protos []pdmlNode `xml:",any"` + } `xml:"packet"` + } + d := xml.NewDecoder(bytes.NewReader(b)) + d.Strict = false + if err := d.Decode(&doc); err != nil { + return nil, err + } + if len(doc.Packets) == 0 { + return nil, errors.New("packet not found") + } + var out []Node + for _, p := range doc.Packets[0].Protos { + if p.Name == "geninfo" || p.Hide == "yes" { + continue + } + n, _ := convert(p, frame) + out = append(out, n) + } + return out, nil +} + +// convert returns the node and whether any field in its subtree contradicts the frame bytes. +func convert(p pdmlNode, frame []byte) (Node, bool) { + n := Node{Field: p.Name} + n.Label = p.Showname + if n.Label == "" { + n.Label = p.Show + } + if n.Label == "" { + n.Label = p.Name + } + n.Pos, _ = strconv.Atoi(p.Pos) + n.Size, _ = strconv.Atoi(p.Size) + inRange := n.Size > 0 && n.Pos >= 0 && n.Pos+n.Size <= len(frame) + raw := p.Value + if p.Unmasked != "" { + raw = p.Unmasked // bitfields: value is the masked number, unmaskedvalue the bytes + } + contradicts := false + if v, err := hex.DecodeString(raw); err == nil && len(v) > 0 && len(v) == n.Size { + if inRange && bytes.Equal(v, frame[n.Pos:n.Pos+n.Size]) { + n.Verified = true + } else { + contradicts = true + } + } + matched, mismatched := false, contradicts + for _, c := range p.Children { + if c.Hide == "yes" { + continue + } + cn, cm := convert(c, frame) + if c.XMLName.Local != "proto" { // a nested (e.g. reassembled) protocol is judged on its own + matched = matched || cn.Verified + mismatched = mismatched || cm + } + n.Children = append(n.Children, cn) + } + // Protocols carry no value: trust their range only if their own fields verify + // against the frame bytes and none contradict it (rules out reassembled sources). + if raw == "" && inRange && matched && !mismatched { + n.Verified = true + } + if p.Name == "frame" && inRange { + n.Verified = true + } + return n, mismatched +} diff --git a/internal/capture/query.go b/internal/capture/query.go new file mode 100644 index 0000000..9b9444f --- /dev/null +++ b/internal/capture/query.go @@ -0,0 +1,503 @@ +package capture + +import ( + "bufio" + "context" + "crypto/sha256" + "database/sql" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "sort" + "strings" + "time" + + "pilens.dev/pilens/internal/pcapfile" +) + +const rowCols = `num,seg,off,ts,src,dst,proto,len,info,sport,dport,stream,tcpflags,ttl,vlan,color` + +func scanRows(rows *sql.Rows) ([]Row, error) { + defer rows.Close() + out := []Row{} + for rows.Next() { + var r Row + var flags, color sql.NullString + if err := rows.Scan(&r.Num, &r.Seg, &r.Off, &r.TS, &r.Src, &r.Dst, &r.Proto, &r.Len, &r.Info, &r.SPort, &r.DPort, &r.Stream, &flags, &r.TTL, &r.VLAN, &color); err != nil { + return nil, err + } + r.TCPFlags, r.Color = flags.String, color.String + out = append(out, r) + } + return out, rows.Err() +} + +// Page is a window of packet rows. Position i in the (filtered) list is +// Offset+i; rows may be missing where summaries are still being generated. +type Page struct { + Total int64 `json:"total"` + First int64 `json:"first"` // packet number at position 0 (unfiltered views) + Offset int64 `json:"offset"` + Rows []Row `json:"rows"` + Nums []int64 `json:"nums,omitempty"` // packet numbers for each position (filtered views) + Filtered bool `json:"filtered"` + Partial bool `json:"partial"` // filter results cover only part of the capture +} + +func (e *Engine) Packets(ctx context.Context, id, filter string, offset, limit int64) (Page, error) { + if limit <= 0 || limit > 2000 { + limit = 500 + } + if offset < 0 { + offset = 0 + } + p := Page{Offset: offset} + if strings.TrimSpace(filter) == "" { + var first, last sql.NullInt64 + e.DB.QueryRow(`SELECT MIN(num),MAX(num) FROM packets WHERE session_id=?`, id).Scan(&first, &last) + if !first.Valid { + return p, nil + } + p.First = first.Int64 + p.Total = last.Int64 - first.Int64 + 1 + from := first.Int64 + offset + rows, err := e.DB.QueryContext(ctx, `SELECT `+rowCols+` FROM packets WHERE session_id=? AND num>=? AND num= p.Total { + return p, nil + } + end := min(offset+limit, p.Total) + p.Nums = nums[offset:end] + if len(p.Nums) == 0 { + return p, nil + } + rows, err := e.DB.QueryContext(ctx, `SELECT `+rowCols+` FROM packets WHERE session_id=? AND num>=? AND num<=? ORDER BY num`, id, p.Nums[0], p.Nums[len(p.Nums)-1]) + if err != nil { + return p, err + } + all, err := scanRows(rows) + if err != nil { + return p, err + } + want := map[int64]bool{} + for _, n := range p.Nums { + want[n] = true + } + for _, r := range all { + if want[r.Num] { + p.Rows = append(p.Rows, r) + } + } + return p, nil +} + +// ---------------------------------------------------------------- display filter cache + +type filterEntry struct { + perSeg map[int]segResult + used time.Time +} + +type segResult struct { + count int64 // segment packet count when evaluated + nums []int64 +} + +func (e *Engine) dropFilterCache(id string) { + e.fcMu.Lock() + defer e.fcMu.Unlock() + for k := range e.fcache { + if strings.HasPrefix(k, id+"\x00") { + delete(e.fcache, k) + } + } +} + +// filterNums evaluates a display filter with tshark segment by segment. Complete +// segments are cached; the segment being written is re-evaluated when it grows. +func (e *Engine) filterNums(ctx context.Context, id, filter string) ([]int64, error) { + if err := sanityFilter(filter); err != nil { + return nil, err + } + segs, err := e.Segments(id) + if err != nil { + return nil, err + } + key := id + "\x00" + filter + e.fcMu.Lock() + if e.fcache == nil { + e.fcache = map[string]*filterEntry{} + } + ent := e.fcache[key] + if ent == nil { + if len(e.fcache) >= 16 { // ponytail: evict oldest of a tiny cache + var oldK string + var oldT time.Time + for k, v := range e.fcache { + if oldK == "" || v.used.Before(oldT) { + oldK, oldT = k, v.used + } + } + delete(e.fcache, oldK) + } + ent = &filterEntry{perSeg: map[int]segResult{}} + e.fcache[key] = ent + } + ent.used = time.Now() + e.fcMu.Unlock() + + var out []int64 + for _, s := range segs { + if s.Deleted { + continue + } + count := s.Count + if !s.Complete { + count = -1 // live: always re-evaluate + e.flushLive(id) + } + e.fcMu.Lock() + r, ok := ent.perSeg[s.Seq] + e.fcMu.Unlock() + if !ok || r.count != count || count == -1 { + local, err := e.Tools.MatchFrames(ctx, filepath.Join(e.sessionDir(id), s.File), filter) + if err != nil { + return nil, err + } + r = segResult{count: count} + for _, n := range local { + r.nums = append(r.nums, s.Base+n-1) + } + e.fcMu.Lock() + ent.perSeg[s.Seq] = r + e.fcMu.Unlock() + } + out = append(out, r.nums...) + } + return out, nil +} + +func (e *Engine) flushLive(id string) { + e.mu.Lock() + l := e.live + e.mu.Unlock() + if l != nil && l.id == id { + l.mu.Lock() + if l.seg != nil { + l.seg.bw.Flush() + } + l.mu.Unlock() + } +} + +// ---------------------------------------------------------------- packet detail + +type Detail struct { + Row Row `json:"row"` + Bytes string `json:"bytes"` // hex of the original captured bytes + Tree []Node `json:"tree"` + Segment int `json:"segment"` + Note string `json:"note,omitempty"` +} + +func (e *Engine) segmentFor(id string, seq int) (Segment, error) { + segs, err := e.Segments(id) + if err != nil { + return Segment{}, err + } + for _, s := range segs { + if s.Seq == seq && !s.Deleted { + return s, nil + } + } + return Segment{}, errors.New("segment no longer stored (ring buffer or retention)") +} + +func (e *Engine) Packet(ctx context.Context, id string, num int64) (Detail, error) { + rows, err := e.DB.QueryContext(ctx, `SELECT `+rowCols+` FROM packets WHERE session_id=? AND num=?`, id, num) + if err != nil { + return Detail{}, err + } + rs, err := scanRows(rows) + if err != nil { + return Detail{}, err + } + if len(rs) == 0 { + return Detail{}, ErrNotFound + } + d := Detail{Row: rs[0], Segment: rs[0].Seg} + seg, err := e.segmentFor(id, d.Row.Seg) + if err != nil { + return d, err + } + path := filepath.Join(e.sessionDir(id), seg.File) + e.flushLive(id) + data, err := pcapfile.ReadPacketAt(path, seg.Format, d.Row.Off) + if err != nil { + return d, fmt.Errorf("reading packet bytes: %w", err) + } + d.Bytes = hex.EncodeToString(data) + pdml, err := e.Tools.PDML(ctx, path, num-seg.Base+1) + if err != nil { + return d, fmt.Errorf("dissection failed: %w", err) + } + d.Tree, err = ParsePDML(pdml, data) + if err != nil { + return d, fmt.Errorf("dissection output: %w", err) + } + if seg.Seq > 1 || !seg.Complete { + d.Note = "Dissected within segment " + fmt.Sprint(seg.Seq) + ": TCP stream numbers and reassembly restart at segment boundaries." + } + return d, nil +} + +// ---------------------------------------------------------------- export + +// ExportFormat reports the container format an export of id will use. +func (e *Engine) ExportFormat(id string) string { + segs, _ := e.Segments(id) + for _, s := range segs { + if s.Format == "pcap" { + return "pcap" + } + } + return "pcapng" +} + +// Export writes original packet records (all, or those matching filter) to w. +// Packet data is copied byte-for-byte from the stored segments. +func (e *Engine) Export(ctx context.Context, w io.Writer, id, filter string) (int, error) { + segs, err := e.Segments(id) + if err != nil { + return 0, err + } + var keep map[int64]bool + if strings.TrimSpace(filter) != "" { + nums, err := e.filterNums(ctx, id, filter) + if err != nil { + return 0, err + } + keep = make(map[int64]bool, len(nums)) + for _, n := range nums { + keep[n] = true + } + } + e.flushLive(id) + bw := bufio.NewWriterSize(w, 1<<20) + total, first := 0, true + for _, s := range segs { + if s.Deleted { + continue + } + base := s.Base + n, err := pcapfile.CopySelected(bw, filepath.Join(e.sessionDir(id), s.File), s.Format, first, func(idx int) bool { + return keep == nil || keep[base+int64(idx)] + }) + total += n + first = false + if err != nil && !s.Complete && errors.Is(err, io.ErrUnexpectedEOF) { + continue // live segment: the last block is still being written + } + if err != nil { + bw.Flush() + return total, err + } + } + return total, bw.Flush() +} + +// ---------------------------------------------------------------- integrity + +type Manifest struct { + Session Session `json:"session"` + Segments []Segment `json:"segments"` + Generated string `json:"generated_utc"` + Note string `json:"note"` +} + +func (e *Engine) Manifest(id string) (Manifest, error) { + s, err := e.Get(id) + if err != nil { + return Manifest{}, err + } + segs, err := e.Segments(id) + return Manifest{Session: s, Segments: segs, Generated: time.Now().UTC().Format(time.RFC3339Nano), + Note: "Segment files are original capture data written verbatim from dumpcap output (live) or the uploaded file (import). SHA-256 covers each complete file. Summary rows and dissections are derived data and are not part of the evidence."}, err +} + +type VerifyResult struct { + Seq int `json:"seq"` + File string `json:"file"` + Status string `json:"status"` // ok | mismatch | missing | incomplete | deleted + Actual string `json:"actual,omitempty"` +} + +func (e *Engine) Verify(id string) ([]VerifyResult, error) { + segs, err := e.Segments(id) + if err != nil { + return nil, err + } + var out []VerifyResult + for _, s := range segs { + r := VerifyResult{Seq: s.Seq, File: s.File} + switch { + case s.Deleted: + r.Status = "deleted" + case !s.Complete: + r.Status = "incomplete" + default: + f, err := os.Open(filepath.Join(e.sessionDir(id), s.File)) + if err != nil { + r.Status = "missing" + break + } + h := sha256.New() + io.Copy(h, f) + f.Close() + r.Actual = hex.EncodeToString(h.Sum(nil)) + r.Status = "ok" + if r.Actual != s.SHA256 { + r.Status = "mismatch" + } + } + out = append(out, r) + } + sort.Slice(out, func(i, j int) bool { return out[i].Seq < out[j].Seq }) + return out, nil +} + +// ---------------------------------------------------------------- import + +// Import stores an uploaded capture unchanged and indexes it in the background. +// The upload is hostile input: size-limited, magic-checked, and parsed only by +// bounds-checked code and resource-limited tshark. +func (e *Engine) Import(ctx context.Context, name string, r io.Reader) (Session, error) { + lim := e.Limits() + name = strings.TrimSpace(name) + if name == "" || len(name) > 200 { + name = "Import " + time.Now().Format("2006-01-02 15:04") + } + if err := e.enforceStorage(""); err != nil { + return Session{}, err + } + id := newID() + dir := e.sessionDir(id) + if err := os.MkdirAll(dir, 0o750); err != nil { + return Session{}, err + } + br := bufio.NewReader(io.LimitReader(r, lim.ImportBytes+1)) + head, _ := br.Peek(4) + format, err := pcapfile.Sniff(head) + if err != nil { + os.RemoveAll(dir) + return Session{}, errors.New("file is not pcap or pcapng (compressed captures are not accepted)") + } + file := "original." + format + f, err := os.OpenFile(filepath.Join(dir, file), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o640) + if err != nil { + os.RemoveAll(dir) + return Session{}, err + } + h := sha256.New() + n, err := io.Copy(io.MultiWriter(f, h), br) + f.Sync() + f.Close() + if err != nil || n > lim.ImportBytes { + os.RemoveAll(dir) + if err == nil { + err = fmt.Errorf("file exceeds import limit of %d bytes", lim.ImportBytes) + } + return Session{}, err + } + sum := hex.EncodeToString(h.Sum(nil)) + meta := e.meta(map[string]any{"import_sha256": sum, "import_bytes": n}) + tx, err := e.DB.Begin() + if err != nil { + return Session{}, err + } + tx.Exec(`INSERT INTO capture_sessions(id,name,kind,state,started_at,dir,meta,bytes) VALUES(?,?,?,?,?,?,?,?)`, id, name, "import", "importing", time.Now().UnixMilli(), dir, meta, n) + tx.Exec(`INSERT INTO segments(session_id,seq,file,format,base,bytes,sha256,complete) VALUES(?,?,?,?,?,?,?,1)`, id, 1, file, format, 1, n, sum) + if err := tx.Commit(); err != nil { + return Session{}, err + } + go e.indexImport(id, filepath.Join(dir, file), format) + return e.Get(id) +} + +func (e *Engine) indexImport(id, path, format string) { + fail := func(err error) { e.finishSession(id, "failed", err.Error(), "") } + var offs []int64 + if err := pcapfile.Scan(path, format, func(idx int, off int64, _ []byte) error { + if len(offs) >= 50_000_000 { + return errors.New("too many packets") + } + offs = append(offs, off) + return nil + }); err != nil { + fail(fmt.Errorf("malformed capture after packet %d: %v", len(offs), err)) + return + } + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Hour) + defer cancel() + cmd := e.Tools.Command(ctx, 3600, e.Tools.SummaryArgs(path)...) + out, err := cmd.StdoutPipe() + var stderr limitedBuffer + cmd.Stderr = &stderr + if err == nil { + err = cmd.Start() + } + if err != nil { + fail(err) + return + } + var batch []Row + var count int64 + flush := func() { + tx, err := e.DB.Begin() + if err == nil { + insertRows(tx, id, batch) + tx.Commit() + } + batch = batch[:0] + } + ReadSummaries(out, func(r Row) { + if r.Num < 1 || r.Num > int64(len(offs)) { + return + } + r.Off, r.Seg = offs[r.Num-1], 1 + batch = append(batch, r) + count++ + if len(batch) >= 2000 { + flush() + } + }) + flush() + werr := cmd.Wait() + e.DB.Exec(`UPDATE segments SET count=? WHERE session_id=? AND seq=1`, len(offs), id) + e.DB.Exec(`UPDATE capture_sessions SET packets=? WHERE id=?`, len(offs), id) + msg := "" + if werr != nil { + msg = "tshark: " + strings.TrimSpace(stderr.String()) + } + if count < int64(len(offs)) { + msg = strings.TrimSpace(fmt.Sprintf("%s; summarised %d of %d packets", msg, count, len(offs))) + } + e.finishSession(id, "ready", msg, "") + data, _ := json.Marshal(map[string]string{"session": id}) + e.Hub.Publish(Event{Type: "resync", Data: data}) +} diff --git a/internal/capture/tshark.go b/internal/capture/tshark.go new file mode 100644 index 0000000..46e4253 --- /dev/null +++ b/internal/capture/tshark.go @@ -0,0 +1,273 @@ +package capture + +import ( + "bufio" + "bytes" + "context" + "encoding/binary" + "errors" + "fmt" + "io" + "os" + "os/exec" + "strconv" + "strings" + "time" +) + +// Tools locates and runs Wireshark command-line tools. All tshark children run +// through `pilens rlimit-exec` so hostile input cannot exhaust the appliance. +type Tools struct { + Self string // path to the pilens binary (for rlimit-exec) + TShark string + Dumpcap string + MemMB int // RLIMIT_AS for tshark children + + cols []string // summary -e fields, version dependent + emptyNG string // path to an empty pcapng used for filter validation +} + +// summaryFields; the column fields are filled in by Init (names changed in Wireshark 4.2). +var summaryTail = []string{ + "frame.len", "tcp.srcport", "udp.srcport", "tcp.dstport", "udp.dstport", + "tcp.stream", "tcp.flags.str", "ip.ttl", "ipv6.hlim", "vlan.id", "frame.coloring_rule.name", +} + +// EmptyPcapng is a valid pcapng with one Ethernet interface and no packets. +func EmptyPcapng() []byte { + var b bytes.Buffer + le := binary.LittleEndian + w32 := func(v uint32) { binary.Write(&b, le, v) } + // SHB + w32(0x0A0D0D0A) + w32(28) + w32(0x1A2B3C4D) + binary.Write(&b, le, uint16(1)) + binary.Write(&b, le, uint16(0)) + binary.Write(&b, le, int64(-1)) + w32(28) + // IDB: linktype 1 (Ethernet), snaplen 262144 + w32(1) + w32(20) + binary.Write(&b, le, uint16(1)) + binary.Write(&b, le, uint16(0)) + w32(262144) + w32(20) + return b.Bytes() +} + +func (t *Tools) Init(stateDir string) error { + if t.MemMB == 0 { + t.MemMB = 1536 + } + p := stateDir + "/empty.pcapng" + if err := os.WriteFile(p, EmptyPcapng(), 0o640); err != nil { + return err + } + t.emptyNG = p + newCols := []string{"_ws.col.def_src", "_ws.col.def_dst", "_ws.col.protocol", "_ws.col.info"} + oldCols := []string{"_ws.col.Source", "_ws.col.Destination", "_ws.col.Protocol", "_ws.col.Info"} + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if _, err := t.run(ctx, "-n", "-r", p, "-T", "fields", "-e", newCols[3]); err == nil { + t.cols = newCols + } else if _, err2 := t.run(ctx, "-n", "-r", p, "-T", "fields", "-e", oldCols[3]); err2 == nil { + t.cols = oldCols + } else { + return fmt.Errorf("tshark unusable: %v", err) + } + return nil +} + +// Command builds a resource-limited tshark command. cpuSec 0 = no CPU limit. +func (t *Tools) Command(ctx context.Context, cpuSec int, args ...string) *exec.Cmd { + var cmd *exec.Cmd + if t.Self == "" { // tests only + cmd = exec.CommandContext(ctx, t.TShark, args...) + } else { + a := append([]string{"rlimit-exec", strconv.Itoa(t.MemMB), strconv.Itoa(cpuSec), "--", t.TShark}, args...) + cmd = exec.CommandContext(ctx, t.Self, a...) + } + cmd.Env = []string{"PATH=/usr/bin:/bin", "HOME=/nonexistent", "LANG=C.UTF-8"} + cmd.WaitDelay = 5 * time.Second + return cmd +} + +func (t *Tools) run(ctx context.Context, args ...string) ([]byte, error) { + cmd := t.Command(ctx, 300, args...) + var stderr bytes.Buffer + cmd.Stderr = &stderr + out, err := cmd.Output() + if err != nil { + msg := strings.TrimSpace(stderr.String()) + if msg == "" { + msg = err.Error() + } + return out, errors.New(msg) + } + return out, nil +} + +// SummaryArgs returns tshark arguments that print one tab-separated summary row per packet. +func (t *Tools) SummaryArgs(input string) []string { + a := []string{"-n", "-l", "--color", "-r", input, "-T", "fields", "-E", "separator=/t", "-E", "occurrence=f", "-E", "quote=n", + "-e", "frame.number", "-e", "frame.time_epoch", "-e", t.cols[0], "-e", t.cols[1], "-e", t.cols[2]} + for _, f := range summaryTail { + a = append(a, "-e", f) + } + return append(a, "-e", t.cols[3]) // info last: it may contain anything +} + +// Row is one packet summary. +type Row struct { + Num int64 `json:"n"` + TS float64 `json:"t"` + Src string `json:"s"` + Dst string `json:"d"` + Proto string `json:"p"` + Len int `json:"l"` + Info string `json:"i"` + SPort *int `json:"sp,omitempty"` + DPort *int `json:"dp,omitempty"` + Stream *int `json:"st,omitempty"` + TCPFlags string `json:"tf,omitempty"` + TTL *int `json:"ttl,omitempty"` + VLAN *int `json:"vl,omitempty"` + Color string `json:"c,omitempty"` + Seg int `json:"-"` + Off int64 `json:"-"` +} + +func atoiPtr(s string) *int { + if s == "" { + return nil + } + v, err := strconv.Atoi(s) + if err != nil { + return nil + } + return &v +} + +func firstNonEmpty(a, b string) string { + if a != "" { + return a + } + return b +} + +// ParseSummaryLine parses one line of SummaryArgs output. Frame numbers are local to the input. +func ParseSummaryLine(line string) (Row, error) { + f := strings.SplitN(line, "\t", 17) + if len(f) != 17 { + return Row{}, fmt.Errorf("unexpected tshark summary line (%d fields)", len(f)) + } + var r Row + var err error + if r.Num, err = strconv.ParseInt(f[0], 10, 64); err != nil { + return r, err + } + r.TS, _ = strconv.ParseFloat(f[1], 64) + r.Src, r.Dst, r.Proto = f[2], f[3], f[4] + r.Len, _ = strconv.Atoi(f[5]) + r.SPort = atoiPtr(firstNonEmpty(f[6], f[7])) + r.DPort = atoiPtr(firstNonEmpty(f[8], f[9])) + r.Stream = atoiPtr(f[10]) + r.TCPFlags = f[11] + r.TTL = atoiPtr(firstNonEmpty(f[12], f[13])) + r.VLAN = atoiPtr(f[14]) + r.Color = f[15] + r.Info = strings.TrimSpace(f[16]) + return r, nil +} + +// ReadSummaries parses tshark summary output, calling fn per row. +func ReadSummaries(r io.Reader, fn func(Row)) error { + sc := bufio.NewScanner(r) + sc.Buffer(make([]byte, 64*1024), 4<<20) + for sc.Scan() { + row, err := ParseSummaryLine(sc.Text()) + if err != nil { + continue // ponytail: skip malformed lines rather than abort a capture; count if it ever matters + } + fn(row) + } + return sc.Err() +} + +// ValidateDisplayFilter asks tshark to compile the filter against an empty file. +func (t *Tools) ValidateDisplayFilter(ctx context.Context, f string) error { + if err := sanityFilter(f); err != nil { + return err + } + _, err := t.run(ctx, "-n", "-r", t.emptyNG, "-Y", f) + return err +} + +// ValidateCaptureFilter asks dumpcap to compile a BPF filter for iface. +func (t *Tools) ValidateCaptureFilter(ctx context.Context, iface, f string) error { + if err := sanityFilter(f); err != nil { + return err + } + ctx, cancel := context.WithTimeout(ctx, 15*time.Second) + defer cancel() + out, err := exec.CommandContext(ctx, t.Dumpcap, "-d", "-i", iface, "-f", f).CombinedOutput() + if err != nil { + return errors.New(strings.TrimSpace(string(out))) + } + return nil +} + +func sanityFilter(f string) error { + if len(f) > 2048 { + return errors.New("filter too long") + } + for _, r := range f { + if r < 0x20 && r != '\t' { + return errors.New("filter contains control characters") + } + } + return nil +} + +// MatchFrames returns the local frame numbers in file that match a display filter. +// Truncated trailing data (a segment still being written) is tolerated. +func (t *Tools) MatchFrames(ctx context.Context, file, filter string) ([]int64, error) { + cmd := t.Command(ctx, 600, "-n", "-r", file, "-Y", filter, "-T", "fields", "-e", "frame.number") + var stderr bytes.Buffer + cmd.Stderr = &stderr + out, err := cmd.Output() + if err != nil && !strings.Contains(stderr.String(), "cut short") { + return nil, errors.New(strings.TrimSpace(stderr.String() + " " + err.Error())) + } + var nums []int64 + for _, l := range strings.Split(string(out), "\n") { + if n, err := strconv.ParseInt(strings.TrimSpace(l), 10, 64); err == nil { + nums = append(nums, n) + } + } + return nums, nil +} + +// PDML returns the PDML dissection for one local frame of file. +func (t *Tools) PDML(ctx context.Context, file string, local int64) ([]byte, error) { + cmd := t.Command(ctx, 120, "-n", "-r", file, "-Y", "frame.number=="+strconv.FormatInt(local, 10), "-T", "pdml") + var stderr bytes.Buffer + cmd.Stderr = &stderr + out, err := cmd.Output() + if err != nil && !strings.Contains(stderr.String(), "cut short") { + return nil, errors.New(strings.TrimSpace(stderr.String() + " " + err.Error())) + } + return out, nil +} + +func (t *Tools) Version() string { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + out, err := exec.CommandContext(ctx, t.TShark, "--version").Output() + if err != nil { + return "unknown" + } + l, _, _ := strings.Cut(string(out), "\n") + return l +} diff --git a/internal/config/config.go b/internal/config/config.go new file mode 100644 index 0000000..324b8f2 --- /dev/null +++ b/internal/config/config.go @@ -0,0 +1,293 @@ +// Package config holds the appliance configuration and its validation. +// The same validation runs in the unprivileged core and again in the root helper. +package config + +import ( + "errors" + "fmt" + "net" + "net/netip" + "regexp" + "strings" +) + +// Network is everything the root helper needs to render AP, DHCP/DNS and firewall state. +type Network struct { + Management Management `json:"management" yaml:"management"` + Uplink string `json:"uplink" yaml:"uplink"` // interface name, e.g. eth0 + AP AP `json:"ap" yaml:"ap"` + LAN LAN `json:"lan" yaml:"lan"` + Forwarding Forwarding `json:"forwarding" yaml:"forwarding"` +} + +type Management struct { + Interfaces []string `json:"interfaces" yaml:"interfaces"` + AllowedSubnets []string `json:"allowed_subnets" yaml:"allowed_subnets"` + Port int `json:"port" yaml:"port"` +} + +type AP struct { + Enabled bool `json:"enabled" yaml:"enabled"` + Interface string `json:"interface" yaml:"interface"` + SSID string `json:"ssid" yaml:"ssid"` + Passphrase string `json:"passphrase,omitempty" yaml:"passphrase"` + Security string `json:"security" yaml:"security"` // wpa2 | wpa3 + Band string `json:"band" yaml:"band"` // bg (2.4 GHz) | a (5 GHz) + Channel int `json:"channel" yaml:"channel"` // 0 = automatic + Hidden bool `json:"hidden" yaml:"hidden"` + Country string `json:"country" yaml:"country"` + ClientIsolation bool `json:"client_isolation" yaml:"client_isolation"` +} + +type LAN struct { + Address string `json:"address" yaml:"address"` // gateway CIDR, e.g. 192.168.50.1/24 + DHCPStart string `json:"dhcp_start" yaml:"dhcp_start"` + DHCPEnd string `json:"dhcp_end" yaml:"dhcp_end"` + LeaseMinutes int `json:"lease_minutes" yaml:"lease_minutes"` + UpstreamDNS []string `json:"upstream_dns" yaml:"upstream_dns"` // empty = system resolvers + Reservations []Reservation `json:"reservations" yaml:"reservations"` + DNSOverrides []DNSOverride `json:"dns_overrides" yaml:"dns_overrides"` +} + +type Reservation struct { + MAC string `json:"mac" yaml:"mac"` + IP string `json:"ip" yaml:"ip"` + Hostname string `json:"hostname,omitempty" yaml:"hostname"` +} + +type DNSOverride struct { + Name string `json:"name" yaml:"name"` + IP string `json:"ip" yaml:"ip"` +} + +type Forwarding struct { + // Mode: "deny" forwards nothing; "internet" forwards to non-private destinations via the uplink. + Mode string `json:"mode" yaml:"mode"` + // AllowCIDRs are private destinations explicitly opened to AP clients. + AllowCIDRs []string `json:"allow_cidrs" yaml:"allow_cidrs"` +} + +func DefaultNetwork() Network { + return Network{ + Management: Management{Port: 8443}, + AP: AP{Enabled: true, SSID: "PiLens-Lab", Security: "wpa2", Band: "bg", Channel: 0, Country: "GB"}, + LAN: LAN{Address: "192.168.50.1/24", DHCPStart: "192.168.50.100", DHCPEnd: "192.168.50.199", LeaseMinutes: 720}, + Forwarding: Forwarding{Mode: "internet"}, + } +} + +var ( + ifaceRe = regexp.MustCompile(`^[A-Za-z0-9_.-]{1,15}$`) + countryRe = regexp.MustCompile(`^[A-Z]{2}$`) + hostnameRe = regexp.MustCompile(`^(?i)[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?(\.[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?)*$`) +) + +// PrivateRanges are never reachable from AP clients unless explicitly allowed. +var PrivateRanges = []string{"10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "169.254.0.0/16", "100.64.0.0/10", "0.0.0.0/8", "224.0.0.0/4", "240.0.0.0/4"} + +func ValidInterface(s string) bool { return ifaceRe.MatchString(s) && s != "." && s != ".." } + +// ValidSSID: 1-32 printable ASCII bytes, no backslash/semicolon, no edge spaces (keeps NM keyfile unambiguous). +func ValidSSID(s string) error { + if len(s) < 1 || len(s) > 32 { + return errors.New("SSID must be 1-32 characters") + } + if strings.TrimSpace(s) != s { + return errors.New("SSID must not start or end with a space") + } + for _, r := range s { + if r < 0x20 || r > 0x7e || r == '\\' || r == ';' { + return errors.New("SSID may contain printable ASCII except \\ and ;") + } + } + return nil +} + +func ValidPassphrase(s string) error { + if len(s) < 8 || len(s) > 63 { + return errors.New("passphrase must be 8-63 characters") + } + for _, r := range s { + if r < 0x20 || r > 0x7e { + return errors.New("passphrase must be printable ASCII") + } + } + return nil +} + +func ValidChannel(band string, ch int) error { + if ch == 0 { + return nil + } + switch band { + case "bg": + if ch >= 1 && ch <= 13 { + return nil + } + case "a": + for _, c := range []int{36, 40, 44, 48, 52, 56, 60, 64, 100, 104, 108, 112, 116, 120, 124, 128, 132, 136, 140, 144, 149, 153, 157, 161, 165} { + if c == ch { + return nil + } + } + } + return fmt.Errorf("channel %d is not valid for band %q", ch, band) +} + +// Validate checks n for internal consistency. requirePassphrase is false when the +// caller keeps the stored passphrase. +func (n *Network) Validate(requirePassphrase bool) error { + var errs []error + add := func(f string, a ...any) { errs = append(errs, fmt.Errorf(f, a...)) } + + if n.Management.Port < 1 || n.Management.Port > 65535 { + add("management.port out of range") + } + if len(n.Management.Interfaces) == 0 { + add("at least one management interface is required") + } + for _, i := range n.Management.Interfaces { + if !ValidInterface(i) { + add("invalid management interface %q", i) + } + } + if len(n.Management.AllowedSubnets) == 0 { + add("at least one management source subnet is required") + } + var mgmtNets []netip.Prefix + for _, s := range n.Management.AllowedSubnets { + p, err := netip.ParsePrefix(s) + if err != nil { + add("invalid management subnet %q", s) + continue + } + mgmtNets = append(mgmtNets, p.Masked()) + } + if !ValidInterface(n.Uplink) { + add("invalid uplink interface %q", n.Uplink) + } + + a := n.AP + if !ValidInterface(a.Interface) { + add("invalid AP interface %q", a.Interface) + } + if a.Interface == n.Uplink { + add("AP and uplink must be different interfaces") + } + for _, m := range n.Management.Interfaces { + if m == a.Interface { + add("the AP interface cannot be a management interface") + } + } + if err := ValidSSID(a.SSID); err != nil { + add("%v", err) + } + if requirePassphrase || a.Passphrase != "" { + if err := ValidPassphrase(a.Passphrase); err != nil { + add("%v", err) + } + } + if a.Security != "wpa2" && a.Security != "wpa3" { + add("security must be wpa2 or wpa3") + } + if a.Band != "bg" && a.Band != "a" { + add("band must be bg (2.4 GHz) or a (5 GHz)") + } + if err := ValidChannel(a.Band, a.Channel); err != nil { + add("%v", err) + } + if !countryRe.MatchString(a.Country) { + add("country must be a two-letter ISO 3166 code") + } + + l := n.LAN + gw, err := netip.ParsePrefix(l.Address) + if err != nil || !gw.Addr().Is4() || gw.Bits() < 16 || gw.Bits() > 30 { + add("lan.address must be an IPv4 CIDR between /16 and /30, e.g. 192.168.50.1/24") + } else { + subnet := gw.Masked() + if gw.Addr() == subnet.Addr() || gw.Addr() == broadcast(subnet) { + add("gateway address cannot be the network or broadcast address") + } + for _, m := range mgmtNets { + if m.Overlaps(subnet) { + add("AP subnet %s overlaps management subnet %s", subnet, m) + } + } + start, e1 := netip.ParseAddr(l.DHCPStart) + end, e2 := netip.ParseAddr(l.DHCPEnd) + switch { + case e1 != nil || e2 != nil: + add("invalid DHCP range") + case !subnet.Contains(start) || !subnet.Contains(end) || end.Less(start): + add("DHCP range must be inside %s and start <= end", subnet) + case !gw.Addr().Less(start) && !end.Less(gw.Addr()): + add("DHCP range must not include the gateway address") + case start == subnet.Addr() || end == broadcast(subnet): + add("DHCP range must not include network or broadcast addresses") + } + seen := map[string]bool{} + for _, r := range l.Reservations { + hw, err := net.ParseMAC(r.MAC) + ip, err2 := netip.ParseAddr(r.IP) + if err != nil || len(hw) != 6 { + add("invalid reservation MAC %q", r.MAC) + } else if seen[hw.String()] { + add("duplicate reservation for %s", hw) + } else { + seen[hw.String()] = true + } + if err2 != nil || !subnet.Contains(ip) || ip == gw.Addr() { + add("reservation IP %q must be inside %s and not the gateway", r.IP, subnet) + } + if r.Hostname != "" && !hostnameRe.MatchString(r.Hostname) { + add("invalid reservation hostname %q", r.Hostname) + } + } + } + if l.LeaseMinutes < 2 || l.LeaseMinutes > 60*24*30 { + add("lease duration must be 2 minutes to 30 days") + } + for _, s := range l.UpstreamDNS { + if _, err := netip.ParseAddr(s); err != nil { + add("invalid upstream DNS server %q", s) + } + } + for _, o := range l.DNSOverrides { + if !hostnameRe.MatchString(o.Name) || len(o.Name) > 253 { + add("invalid DNS override name %q", o.Name) + } + if _, err := netip.ParseAddr(o.IP); err != nil { + add("invalid DNS override address %q", o.IP) + } + } + + if n.Forwarding.Mode != "deny" && n.Forwarding.Mode != "internet" { + add("forwarding.mode must be deny or internet") + } + for _, c := range n.Forwarding.AllowCIDRs { + if p, err := netip.ParsePrefix(c); err != nil || !p.Addr().Is4() { + add("invalid IPv4 allow CIDR %q", c) + } + } + return errors.Join(errs...) +} + +func broadcast(p netip.Prefix) netip.Addr { + a := p.Masked().Addr().As4() + host := uint32(1)<<(32-p.Bits()) - 1 + v := uint32(a[0])<<24 | uint32(a[1])<<16 | uint32(a[2])<<8 | uint32(a[3]) | host + return netip.AddrFrom4([4]byte{byte(v >> 24), byte(v >> 16), byte(v >> 8), byte(v)}) +} + +// Subnet returns the AP subnet (masked) for a validated config. +func (l LAN) Subnet() netip.Prefix { return netip.MustParsePrefix(l.Address).Masked() } + +// Gateway returns the AP gateway address for a validated config. +func (l LAN) Gateway() netip.Addr { return netip.MustParsePrefix(l.Address).Addr() } + +// Redacted returns a copy safe to send to the browser or logs. +func (n Network) Redacted() Network { + n.AP.Passphrase = "" + return n +} diff --git a/internal/helper/helper.go b/internal/helper/helper.go new file mode 100644 index 0000000..07157b0 --- /dev/null +++ b/internal/helper/helper.go @@ -0,0 +1,101 @@ +// Package helper defines the request protocol between the unprivileged core and +// the root pilens-helper. Every operation is fixed and typed; there is no generic +// command execution. +package helper + +import ( + "context" + "encoding/json" + "errors" + "net" + "time" + + "pilens.dev/pilens/internal/config" +) + +const DefaultSocket = "/run/pilens/helper.sock" + +type Request struct { + Op string `json:"op"` + Network *config.Network `json:"network,omitempty"` +} + +type Response struct { + OK bool `json:"ok"` + Error string `json:"error,omitempty"` + Data json.RawMessage `json:"data,omitempty"` +} + +// Operation names. +const ( + OpApplyNetwork = "apply_network" + OpStatus = "status" + OpStations = "stations" + OpLeases = "leases" +) + +type ApplyResult struct { + Applied bool `json:"applied"` + RolledBack bool `json:"rolled_back"` + Steps []string `json:"steps"` + Error string `json:"error,omitempty"` +} + +type Status struct { + TableLoaded bool `json:"table_loaded"` + APState string `json:"ap_state"` // NM GENERAL.STATE or "absent" + DnsmasqActive bool `json:"dnsmasq_active"` + IPForward bool `json:"ip_forward"` + Regdomain string `json:"regdomain"` + Configured bool `json:"configured"` + LastError string `json:"last_error,omitempty"` +} + +type Station struct { + MAC string `json:"mac"` + SignalDBm *int `json:"signal_dbm,omitempty"` + RxBytes uint64 `json:"rx_bytes"` + TxBytes uint64 `json:"tx_bytes"` + ConnectedS int64 `json:"connected_s"` + InactiveMS int64 `json:"inactive_ms"` +} + +type Lease struct { + Expiry int64 `json:"expiry"` // unix seconds; 0 = infinite + MAC string `json:"mac"` + IP string `json:"ip"` + Hostname string `json:"hostname"` +} + +// Client talks to the helper socket. +type Client struct{ Socket string } + +func (c Client) Call(ctx context.Context, req Request, out any) error { + d := net.Dialer{} + conn, err := d.DialContext(ctx, "unix", c.Socket) + if err != nil { + return errors.New("network helper unavailable: " + err.Error()) + } + defer conn.Close() + dl, ok := ctx.Deadline() + if !ok { + dl = time.Now().Add(2 * time.Minute) + } + conn.SetDeadline(dl) + if err := json.NewEncoder(conn).Encode(req); err != nil { + return err + } + var resp Response + if err := json.NewDecoder(conn).Decode(&resp); err != nil { + return err + } + if out != nil && len(resp.Data) > 0 { + if err := json.Unmarshal(resp.Data, out); err != nil { + return err + } + } + if !resp.OK { + return errors.New(resp.Error) + } + return nil +} diff --git a/internal/helper/parse.go b/internal/helper/parse.go new file mode 100644 index 0000000..decd783 --- /dev/null +++ b/internal/helper/parse.go @@ -0,0 +1,75 @@ +package helper + +import ( + "bufio" + "strconv" + "strings" +) + +// ParseStationDump parses `iw dev X station dump` output. +func ParseStationDump(s string) []Station { + var out []Station + var cur *Station + sc := bufio.NewScanner(strings.NewReader(s)) + for sc.Scan() { + line := sc.Text() + if strings.HasPrefix(line, "Station ") { + f := strings.Fields(line) + if len(f) >= 2 { + out = append(out, Station{MAC: strings.ToLower(f[1])}) + cur = &out[len(out)-1] + } + continue + } + if cur == nil { + continue + } + k, v, ok := strings.Cut(strings.TrimSpace(line), ":") + if !ok { + continue + } + f := strings.Fields(v) + if len(f) == 0 { + continue + } + n, err := strconv.ParseInt(f[0], 10, 64) + if err != nil { + continue + } + switch k { + case "inactive time": + cur.InactiveMS = n + case "rx bytes": + cur.RxBytes = uint64(n) + case "tx bytes": + cur.TxBytes = uint64(n) + case "signal": + v := int(n) + cur.SignalDBm = &v + case "connected time": + cur.ConnectedS = n + } + } + return out +} + +// ParseLeases parses a dnsmasq lease file: "expiry mac ip hostname clientid". +func ParseLeases(s string) []Lease { + var out []Lease + for _, line := range strings.Split(s, "\n") { + f := strings.Fields(line) + if len(f) < 4 { + continue + } + exp, err := strconv.ParseInt(f[0], 10, 64) + if err != nil { + continue + } + h := f[3] + if h == "*" { + h = "" + } + out = append(out, Lease{Expiry: exp, MAC: strings.ToLower(f[1]), IP: f[2], Hostname: h}) + } + return out +} diff --git a/internal/helper/parse_test.go b/internal/helper/parse_test.go new file mode 100644 index 0000000..efa2d49 --- /dev/null +++ b/internal/helper/parse_test.go @@ -0,0 +1,25 @@ +package helper + +import "testing" + +func TestParseStationDump(t *testing.T) { + s := ParseStationDump(`Station AA:bb:cc:dd:ee:ff (on wlan0) + inactive time: 1200 ms + rx bytes: 5000 + tx bytes: 7000 + signal: -47 [-47, -49] dBm + connected time: 33 seconds +Station 11:22:33:44:55:66 (on wlan0) + rx bytes: 1 +`) + if len(s) != 2 || s[0].MAC != "aa:bb:cc:dd:ee:ff" || *s[0].SignalDBm != -47 || s[0].RxBytes != 5000 || s[0].ConnectedS != 33 || s[1].SignalDBm != nil { + t.Fatalf("%+v", s) + } +} + +func TestParseLeases(t *testing.T) { + l := ParseLeases("1700000000 AA:BB:CC:DD:EE:FF 192.168.50.101 phone 01:aa\n1700000001 11:22:33:44:55:66 192.168.50.102 * *\ngarbage\n") + if len(l) != 2 || l[0].Hostname != "phone" || l[1].Hostname != "" || l[0].MAC != "aa:bb:cc:dd:ee:ff" { + t.Fatalf("%+v", l) + } +} diff --git a/internal/netgen/netgen.go b/internal/netgen/netgen.go new file mode 100644 index 0000000..77e85ad --- /dev/null +++ b/internal/netgen/netgen.go @@ -0,0 +1,242 @@ +// Package netgen renders NetworkManager, dnsmasq and nftables configuration from +// a validated config.Network. Pure functions: no I/O, fully unit-tested. +package netgen + +import ( + "fmt" + "net" + "strings" + + "pilens.dev/pilens/internal/config" +) + +const ( + ConnName = "pilens-ap" + ConnUUID = "6f1c3a2e-9d4b-4e57-8a51-0c7e1b2d9f30" // stable so reloads replace, not duplicate + TableName = "pilens" +) + +// keyfileEscape escapes a GKeyFile string value (NM keyfile format). +func keyfileEscape(s string) string { + s = strings.ReplaceAll(s, `\`, `\\`) + if strings.HasPrefix(s, " ") { + s = `\s` + s[1:] + } + return s +} + +// NMKeyfile renders /etc/NetworkManager/system-connections/pilens-ap.nmconnection. +func NMKeyfile(n config.Network) string { + a := n.AP + var b strings.Builder + w := func(f string, v ...any) { fmt.Fprintf(&b, f+"\n", v...) } + w("# Managed by PiLens. Changes are overwritten; use the PiLens UI.") + w("[connection]") + w("id=%s", ConnName) + w("uuid=%s", ConnUUID) + w("type=wifi") + w("interface-name=%s", a.Interface) + w("autoconnect=%t", a.Enabled) + w("autoconnect-priority=100") + w("") + w("[wifi]") + w("mode=ap") + w("ssid=%s", keyfileEscape(a.SSID)) + w("band=%s", a.Band) + if a.Channel != 0 { + w("channel=%d", a.Channel) + } + w("hidden=%t", a.Hidden) + if a.ClientIsolation { + w("ap-isolation=1") + } else { + w("ap-isolation=0") + } + w("") + w("[wifi-security]") + if a.Security == "wpa3" { + w("key-mgmt=sae") + w("pmf=3") // required for SAE + } else { + w("key-mgmt=wpa-psk") + w("pmf=1") // disabled: broadest client compatibility for WPA2-only + } + w("proto=rsn") + w("pairwise=ccmp") + w("group=ccmp") + w("psk=%s", keyfileEscape(a.Passphrase)) + w("") + w("[ipv4]") + w("method=manual") + w("address1=%s", n.LAN.Address) + w("never-default=true") + w("") + w("[ipv6]") + w("method=disabled") + return b.String() +} + +// Paths used by the rendered dnsmasq config. +const ( + LeaseFile = "/var/lib/pilens-helper/dnsmasq.leases" +) + +// DnsmasqConf renders the dedicated dnsmasq instance for the AP interface. +func DnsmasqConf(n config.Network) string { + l := n.LAN + sub := l.Subnet() + mask := net.CIDRMask(sub.Bits(), 32) + gw := l.Gateway().String() + var b strings.Builder + w := func(f string, v ...any) { fmt.Fprintf(&b, f+"\n", v...) } + w("# Managed by PiLens. Changes are overwritten; use the PiLens UI.") + w("interface=%s", n.AP.Interface) + w("bind-dynamic") // only the AP interface; survives the address appearing late + w("no-hosts") + w("domain-needed") + w("bogus-priv") + w("cache-size=1000") + if len(l.UpstreamDNS) == 0 { + w("resolv-file=/etc/resolv.conf") + } else { + w("no-resolv") + for _, s := range l.UpstreamDNS { + w("server=%s", s) + } + } + w("dhcp-authoritative") + w("dhcp-range=%s,%s,%s,%dm", l.DHCPStart, l.DHCPEnd, net.IP(mask).String(), l.LeaseMinutes) + w("dhcp-option=option:router,%s", gw) + w("dhcp-option=option:dns-server,%s", gw) + w("dhcp-leasefile=%s", LeaseFile) + for _, r := range l.Reservations { + hw, _ := net.ParseMAC(r.MAC) + if r.Hostname != "" { + w("dhcp-host=%s,%s,%s", hw, r.IP, r.Hostname) + } else { + w("dhcp-host=%s,%s", hw, r.IP) + } + } + for _, o := range l.DNSOverrides { + w("host-record=%s,%s", strings.ToLower(o.Name), o.IP) // exact name only, unlike address=/x/ + } + w("log-queries=extra") + w("log-dhcp") + w("log-facility=-") + return b.String() +} + +func q(s string) string { return `"` + s + `"` } // inputs are validated interface names + +// Nftables renders the complete inet pilens table. The output begins by +// (re)creating and deleting the table so `nft -f` replaces it atomically while +// leaving every other table untouched. +func Nftables(n config.Network) string { + ap, up := q(n.AP.Interface), q(n.Uplink) + sub := n.LAN.Subnet().String() + var mgmtIf []string + for _, i := range n.Management.Interfaces { + mgmtIf = append(mgmtIf, q(i)) + } + var v4, v6 []string + for _, s := range n.Management.AllowedSubnets { + if strings.Contains(s, ":") { + v6 = append(v6, s) + } else { + v4 = append(v4, s) + } + } + blocked := append(append([]string{}, config.PrivateRanges...), v4...) + blocked = append(blocked, "255.255.255.255/32") + + var b strings.Builder + w := func(f string, v ...any) { fmt.Fprintf(&b, f+"\n", v...) } + w("# Managed by PiLens. Only this table is ever created or deleted by PiLens.") + w("table inet %s", TableName) + w("delete table inet %s", TableName) + w("table inet %s {", TableName) + set := func(name, typ string, elems []string) { + w("\tset %s {", name) + w("\t\ttype %s; flags interval; auto-merge", typ) + if len(elems) > 0 { + w("\t\telements = { %s }", strings.Join(elems, ", ")) + } + w("\t}") + } + set("mgmt_v4", "ipv4_addr", v4) + set("mgmt_v6", "ipv6_addr", v6) + set("blocked_v4", "ipv4_addr", blocked) + set("allow_v4", "ipv4_addr", n.Forwarding.AllowCIDRs) + + // Input: decide everything from the AP side; restrict only the management port elsewhere. + w("\tchain input {") + w("\t\ttype filter hook input priority filter - 10; policy accept;") + w("\t\tiifname \"lo\" accept") + w("\t\tiifname %s jump ap_input", ap) + w("\t\ttcp dport %d iifname { %s } ip saddr @mgmt_v4 accept", n.Management.Port, strings.Join(mgmtIf, ", ")) + w("\t\ttcp dport %d iifname { %s } ip6 saddr @mgmt_v6 accept", n.Management.Port, strings.Join(mgmtIf, ", ")) + w("\t\ttcp dport %d counter drop comment \"management not allowed from here\"", n.Management.Port) + w("\t}") + w("\tchain ap_input {") + w("\t\tmeta nfproto ipv6 counter drop") + w("\t\tct state established,related accept") + w("\t\tct state invalid drop") + w("\t\tudp dport 67 accept") + w("\t\tip daddr %s udp dport 53 accept", n.LAN.Gateway()) + w("\t\tip daddr %s tcp dport 53 accept", n.LAN.Gateway()) + w("\t\ticmp type { echo-request, destination-unreachable, time-exceeded } accept") + w("\t\tcounter drop comment \"AP clients cannot reach the appliance\"") + w("\t}") + + w("\tchain forward {") + w("\t\ttype filter hook forward priority filter - 10; policy accept;") + w("\t\tiifname %s jump ap_forward", ap) + w("\t\toifname %s jump ap_return", ap) + w("\t}") + w("\tchain ap_forward {") + w("\t\tmeta nfproto ipv6 counter drop comment \"IPv6 forwarding not implemented: blocked\"") + w("\t\tct state invalid counter drop") + w("\t\tip saddr != %s counter drop comment \"spoofed source\"", sub) + w("\t\toifname != %s counter drop comment \"only the uplink\"", up) + if n.Forwarding.Mode != "internet" { + w("\t\tcounter drop comment \"forwarding disabled\"") + } else { + w("\t\tip daddr @allow_v4 ct state { new, established, related } accept") + w("\t\tip daddr @blocked_v4 counter drop comment \"private/management destinations\"") + w("\t\tct state { new, established, related } accept") + } + w("\t\tcounter drop") + w("\t}") + w("\tchain ap_return {") + w("\t\tiifname %s ct state established,related accept", up) + w("\t\tcounter drop") + w("\t}") + + w("\tchain postrouting {") + w("\t\ttype nat hook postrouting priority srcnat; policy accept;") + w("\t\tip saddr %s oifname %s masquerade", sub, up) + w("\t}") + w("}") + return b.String() +} + +// DenyAll renders the fail-closed table used before any configuration exists: +// AP-side traffic is dropped entirely, nothing else is touched. +func DenyAll(apIface string) string { + ap := q(apIface) + return fmt.Sprintf(`# Managed by PiLens: fail-closed placeholder. +table inet %[1]s +delete table inet %[1]s +table inet %[1]s { + chain input { + type filter hook input priority filter - 10; policy accept; + iifname %[2]s counter drop + } + chain forward { + type filter hook forward priority filter - 10; policy accept; + iifname %[2]s counter drop + oifname %[2]s counter drop + } +} +`, TableName, ap) +} diff --git a/internal/netgen/netgen_test.go b/internal/netgen/netgen_test.go new file mode 100644 index 0000000..45b59ba --- /dev/null +++ b/internal/netgen/netgen_test.go @@ -0,0 +1,90 @@ +package netgen + +import ( + "strings" + "testing" + + "pilens.dev/pilens/internal/config" +) + +func testNet() config.Network { + n := config.DefaultNetwork() + n.Management.Interfaces = []string{"eth0"} + n.Management.AllowedSubnets = []string{"10.1.0.0/24"} + n.Uplink = "eth0" + n.AP.Interface = "wlan0" + n.AP.Passphrase = `pass\word 123` + n.LAN.Reservations = []config.Reservation{{MAC: "AA:BB:CC:00:11:22", IP: "192.168.50.20", Hostname: "phone"}} + n.LAN.DNSOverrides = []config.DNSOverride{{Name: "Api.Test.Local", IP: "192.168.50.1"}} + return n +} + +func TestValidate(t *testing.T) { + n := testNet() + if err := n.Validate(true); err != nil { + t.Fatal(err) + } + bad := []func(*config.Network){ + func(n *config.Network) { n.AP.Interface = "eth0" }, + func(n *config.Network) { n.AP.SSID = "a;b" }, + func(n *config.Network) { n.AP.SSID = "bad\nname" }, + func(n *config.Network) { n.AP.Passphrase = "short" }, + func(n *config.Network) { n.AP.Channel = 36 }, + func(n *config.Network) { n.LAN.DHCPStart = "192.168.50.1" }, + func(n *config.Network) { n.LAN.DHCPEnd = "192.168.51.10" }, + func(n *config.Network) { n.Management.AllowedSubnets = []string{"192.168.50.0/25"} }, + func(n *config.Network) { n.LAN.DNSOverrides[0].Name = "x\nserver=1.2.3.4" }, + func(n *config.Network) { n.Uplink = "eth0; rm -rf" }, + func(n *config.Network) { n.Forwarding.Mode = "open" }, + func(n *config.Network) { n.LAN.Address = "192.168.50.0/24" }, + } + for i, f := range bad { + c := testNet() + f(&c) + if c.Validate(true) == nil { + t.Errorf("case %d: expected validation error", i) + } + } +} + +func TestKeyfile(t *testing.T) { + k := NMKeyfile(testNet()) + for _, want := range []string{"mode=ap", "interface-name=wlan0", `psk=pass\\word 123`, "key-mgmt=wpa-psk", "method=disabled", "address1=192.168.50.1/24", "ssid=PiLens-Lab"} { + if !strings.Contains(k, want) { + t.Errorf("keyfile missing %q", want) + } + } +} + +func TestDnsmasq(t *testing.T) { + d := DnsmasqConf(testNet()) + for _, want := range []string{"interface=wlan0", "dhcp-range=192.168.50.100,192.168.50.199,255.255.255.0,720m", "dhcp-host=aa:bb:cc:00:11:22,192.168.50.20,phone", "host-record=api.test.local,192.168.50.1", "option:router,192.168.50.1"} { + if !strings.Contains(d, want) { + t.Errorf("dnsmasq missing %q\n%s", want, d) + } + } +} + +func TestNftables(t *testing.T) { + s := Nftables(testNet()) + for _, want := range []string{ + "delete table inet pilens", + `iifname "wlan0" jump ap_forward`, + "meta nfproto ipv6 counter drop", + `ip saddr 192.168.50.0/24 oifname "eth0" masquerade`, + "10.1.0.0/24", + `tcp dport 8443 counter drop`, + } { + if !strings.Contains(s, want) { + t.Errorf("ruleset missing %q", want) + } + } + if strings.Contains(s, "flush ruleset") { + t.Error("must never flush the whole ruleset") + } + n := testNet() + n.Forwarding.Mode = "deny" + if !strings.Contains(Nftables(n), `"forwarding disabled"`) { + t.Error("deny mode must drop forwarded traffic") + } +} diff --git a/internal/pcapfile/pcapfile.go b/internal/pcapfile/pcapfile.go new file mode 100644 index 0000000..55f69aa --- /dev/null +++ b/internal/pcapfile/pcapfile.go @@ -0,0 +1,336 @@ +// Package pcapfile reads and copies pcap/pcapng files at the block/record level. +// It never re-encodes packet data: blocks are copied byte-for-byte. +// All input is treated as hostile: every length is bounds-checked. +package pcapfile + +import ( + "bufio" + "encoding/binary" + "errors" + "fmt" + "io" + "os" +) + +const ( + BlockSHB = 0x0A0D0D0A + BlockIDB = 1 + BlockPB = 2 // obsolete Packet Block + BlockSPB = 3 + BlockEPB = 6 + BlockISB = 5 + + MaxBlock = 16 << 20 // refuse blocks larger than this +) + +var ErrFormat = errors.New("not a pcap or pcapng file") + +// Sniff identifies the format from the first 4 bytes. +func Sniff(b []byte) (string, error) { + if len(b) < 4 { + return "", ErrFormat + } + switch binary.LittleEndian.Uint32(b) { + case BlockSHB: + return "pcapng", nil + case 0xa1b2c3d4, 0xd4c3b2a1, 0xa1b23c4d, 0x4d3cb2a1: + return "pcap", nil + } + return "", ErrFormat +} + +// Block is one raw pcapng block. +type Block struct { + Type uint32 + Raw []byte // the complete block, verbatim + Order binary.ByteOrder +} + +func (b Block) IsPacket() bool { + return b.Type == BlockEPB || b.Type == BlockSPB || b.Type == BlockPB +} + +// BlockReader reads pcapng blocks from a stream, tracking section byte order. +type BlockReader struct { + r io.Reader + order binary.ByteOrder +} + +func NewBlockReader(r io.Reader) *BlockReader { return &BlockReader{r: r} } + +func (br *BlockReader) Next() (Block, error) { + var hdr [12]byte + if _, err := io.ReadFull(br.r, hdr[:8]); err != nil { + return Block{}, err + } + typ := binary.LittleEndian.Uint32(hdr[:4]) // SHB type is a palindrome + if typ == BlockSHB { + if _, err := io.ReadFull(br.r, hdr[8:12]); err != nil { + return Block{}, io.ErrUnexpectedEOF + } + switch binary.LittleEndian.Uint32(hdr[8:12]) { + case 0x1A2B3C4D: + br.order = binary.LittleEndian + case 0x4D3C2B1A: + br.order = binary.BigEndian + default: + return Block{}, errors.New("pcapng: bad byte-order magic") + } + } else if br.order == nil { + return Block{}, errors.New("pcapng: data before section header") + } else { + typ = br.order.Uint32(hdr[:4]) + } + n := br.order.Uint32(hdr[4:8]) + if n < 12 || n%4 != 0 || n > MaxBlock { + return Block{}, fmt.Errorf("pcapng: invalid block length %d", n) + } + raw := make([]byte, n) + got := 8 + copy(raw, hdr[:8]) + if typ == BlockSHB { + copy(raw[8:], hdr[8:12]) + got = 12 + } + if _, err := io.ReadFull(br.r, raw[got:]); err != nil { + return Block{}, io.ErrUnexpectedEOF + } + if br.order.Uint32(raw[n-4:]) != n { + return Block{}, errors.New("pcapng: trailing length mismatch") + } + return Block{Type: typ, Raw: raw, Order: br.order}, nil +} + +// PacketData extracts captured bytes from a packet block. +func PacketData(b Block) ([]byte, error) { + r := b.Raw + o := b.Order + switch b.Type { + case BlockEPB, BlockPB: + if len(r) < 32 { + return nil, errors.New("short packet block") + } + cl := o.Uint32(r[20:24]) + if int(cl) > len(r)-32 { + return nil, errors.New("captured length exceeds block") + } + return r[28 : 28+cl], nil + case BlockSPB: + if len(r) < 16 { + return nil, errors.New("short simple packet block") + } + cl := min(int(o.Uint32(r[8:12])), len(r)-16) + return r[12 : 12+cl], nil + } + return nil, errors.New("not a packet block") +} + +// pcap (classic) -------------------------------------------------------------- + +type pcapHeader struct { + order binary.ByteOrder + raw [24]byte +} + +func readPcapHeader(r io.Reader) (pcapHeader, error) { + var h pcapHeader + if _, err := io.ReadFull(r, h.raw[:]); err != nil { + return h, ErrFormat + } + switch binary.LittleEndian.Uint32(h.raw[:4]) { + case 0xa1b2c3d4, 0xa1b23c4d: + h.order = binary.LittleEndian + case 0xd4c3b2a1, 0x4d3cb2a1: + h.order = binary.BigEndian + default: + return h, ErrFormat + } + return h, nil +} + +func readPcapRecord(r io.Reader, o binary.ByteOrder) ([]byte, error) { + var rh [16]byte + if _, err := io.ReadFull(r, rh[:]); err != nil { + return nil, err + } + n := o.Uint32(rh[8:12]) + if n > MaxBlock { + return nil, fmt.Errorf("pcap: record length %d too large", n) + } + rec := make([]byte, 16+n) + copy(rec, rh[:]) + if _, err := io.ReadFull(r, rec[16:]); err != nil { + return nil, io.ErrUnexpectedEOF + } + return rec, nil +} + +// Scan calls fn for each packet in a file with its 0-based index, byte offset and data. +func Scan(path, format string, fn func(idx int, off int64, data []byte) error) error { + f, err := os.Open(path) + if err != nil { + return err + } + defer f.Close() + cr := &countingReader{r: bufio.NewReaderSize(f, 1<<16)} + idx := 0 + if format == "pcap" { + h, err := readPcapHeader(cr) + if err != nil { + return err + } + for { + off := cr.n + rec, err := readPcapRecord(cr, h.order) + if err == io.EOF { + return nil + } + if err != nil { + return err + } + if err := fn(idx, off, rec[16:]); err != nil { + return err + } + idx++ + } + } + br := NewBlockReader(cr) + for { + off := cr.n + b, err := br.Next() + if err == io.EOF { + return nil + } + if err != nil { + return err + } + if b.IsPacket() { + d, err := PacketData(b) + if err != nil { + return err + } + if err := fn(idx, off, d); err != nil { + return err + } + idx++ + } + } +} + +// ReadPacketAt returns the captured bytes of the packet stored at off. +func ReadPacketAt(path, format string, off int64) ([]byte, error) { + f, err := os.Open(path) + if err != nil { + return nil, err + } + defer f.Close() + if format == "pcap" { + h, err := readPcapHeader(f) + if err != nil { + return nil, err + } + if _, err := f.Seek(off, io.SeekStart); err != nil { + return nil, err + } + rec, err := readPcapRecord(f, h.order) + if err != nil { + return nil, err + } + return rec[16:], nil + } + var hdr [8]byte + if _, err := f.ReadAt(hdr[:], off); err != nil { + return nil, err + } + var o binary.ByteOrder = binary.LittleEndian + if t := binary.BigEndian.Uint32(hdr[:4]); t == BlockEPB || t == BlockSPB || t == BlockPB { + o = binary.BigEndian + } + t, n := o.Uint32(hdr[:4]), o.Uint32(hdr[4:8]) + if (t != BlockEPB && t != BlockSPB && t != BlockPB) || n < 12 || n > MaxBlock { + return nil, errors.New("no packet block at offset") + } + raw := make([]byte, n) + if _, err := f.ReadAt(raw, off); err != nil { + return nil, err + } + return PacketData(Block{Type: t, Raw: raw, Order: o}) +} + +// CopySelected writes the packets of src whose 0-based index satisfies keep to w, +// together with the section and interface headers they depend on. Packet records +// are copied verbatim. If writeHeader is false, a pcap global header is not +// repeated (used when appending further pcap segments to the same output). +func CopySelected(w io.Writer, path, format string, writeHeader bool, keep func(idx int) bool) (int, error) { + f, err := os.Open(path) + if err != nil { + return 0, err + } + defer f.Close() + r := bufio.NewReaderSize(f, 1<<16) + n, idx := 0, 0 + if format == "pcap" { + h, err := readPcapHeader(r) + if err != nil { + return 0, err + } + if writeHeader { + if _, err := w.Write(h.raw[:]); err != nil { + return 0, err + } + } + for { + rec, err := readPcapRecord(r, h.order) + if err == io.EOF { + return n, nil + } + if err != nil { + return n, err + } + if keep(idx) { + if _, err := w.Write(rec); err != nil { + return n, err + } + n++ + } + idx++ + } + } + br := NewBlockReader(r) + for { + b, err := br.Next() + if err == io.EOF { + return n, nil + } + if err != nil { + return n, err + } + switch { + case b.Type == BlockSHB, b.Type == BlockIDB: + // Always emitted, even when appending: repeated sections are valid + // pcapng and keep each segment's interface IDs resolvable. + if _, err := w.Write(b.Raw); err != nil { + return n, err + } + case b.IsPacket(): + if keep(idx) { + if _, err := w.Write(b.Raw); err != nil { + return n, err + } + n++ + } + idx++ + } + } +} + +type countingReader struct { + r io.Reader + n int64 +} + +func (c *countingReader) Read(p []byte) (int, error) { + n, err := c.r.Read(p) + c.n += int64(n) + return n, err +} diff --git a/internal/pcapfile/pcapfile_test.go b/internal/pcapfile/pcapfile_test.go new file mode 100644 index 0000000..1d39e56 --- /dev/null +++ b/internal/pcapfile/pcapfile_test.go @@ -0,0 +1,54 @@ +package pcapfile + +import ( + "bytes" + "os" + "path/filepath" + "testing" +) + +const fixture = "../../tests/fixtures/sample-loopback-http-dns.pcapng" + +func TestScanAndCopy(t *testing.T) { + var offs []int64 + var first []byte + if err := Scan(fixture, "pcapng", func(i int, off int64, d []byte) error { + offs = append(offs, off) + if i == 0 { + first = append([]byte(nil), d...) + } + return nil + }); err != nil || len(offs) < 20 { + t.Fatalf("scan: %v %d", err, len(offs)) + } + d, err := ReadPacketAt(fixture, "pcapng", offs[0]) + if err != nil || !bytes.Equal(d, first) { + t.Fatalf("ReadPacketAt: %v", err) + } + var buf bytes.Buffer + n, err := CopySelected(&buf, fixture, "pcapng", true, func(i int) bool { return i%2 == 0 }) + if err != nil || n != (len(offs)+1)/2 { + t.Fatalf("copy: %d %v", n, err) + } + p := filepath.Join(t.TempDir(), "o.pcapng") + os.WriteFile(p, buf.Bytes(), 0o600) + c := 0 + if err := Scan(p, "pcapng", func(int, int64, []byte) error { c++; return nil }); err != nil || c != n { + t.Fatalf("rescan %d %v", c, err) + } +} + +func TestMalformed(t *testing.T) { + good, _ := os.ReadFile(fixture) + p := filepath.Join(t.TempDir(), "bad.pcapng") + // truncated, and a block with an absurd length + for _, b := range [][]byte{good[:len(good)-7], append(append([]byte{}, good[:28]...), 6, 0, 0, 0, 0xff, 0xff, 0xff, 0x7f)} { + os.WriteFile(p, b, 0o600) + if err := Scan(p, "pcapng", func(int, int64, []byte) error { return nil }); err == nil { + t.Fatal("malformed file accepted") + } + } + if _, err := Sniff([]byte("PK\x03\x04")); err == nil { + t.Fatal("zip accepted") + } +} diff --git a/internal/store/store.go b/internal/store/store.go new file mode 100644 index 0000000..1bcf43c --- /dev/null +++ b/internal/store/store.go @@ -0,0 +1,88 @@ +// Package store opens the SQLite database and applies migrations. +package store + +import ( + "database/sql" + "encoding/json" + "fmt" + "io/fs" + "sort" + "time" + + _ "modernc.org/sqlite" + + "pilens.dev/pilens/migrations" +) + +type Store struct{ *sql.DB } + +func Open(path string) (*Store, error) { + dsn := "file:" + path + "?_pragma=journal_mode(WAL)&_pragma=foreign_keys(1)&_pragma=busy_timeout(5000)&_pragma=synchronous(NORMAL)" + db, err := sql.Open("sqlite", dsn) + if err != nil { + return nil, err + } + db.SetMaxOpenConns(4) + s := &Store{db} + if err := s.migrate(); err != nil { + db.Close() + return nil, err + } + return s, nil +} + +func (s *Store) migrate() error { + if _, err := s.Exec(`CREATE TABLE IF NOT EXISTS schema_migrations (name TEXT PRIMARY KEY, applied_at INTEGER NOT NULL)`); err != nil { + return err + } + names, err := fs.Glob(migrations.FS, "*.sql") + if err != nil { + return err + } + sort.Strings(names) + for _, n := range names { + var x int + if s.QueryRow(`SELECT 1 FROM schema_migrations WHERE name=?`, n).Scan(&x) == nil { + continue + } + b, _ := migrations.FS.ReadFile(n) + tx, err := s.Begin() + if err != nil { + return err + } + if _, err := tx.Exec(string(b)); err != nil { + tx.Rollback() + return fmt.Errorf("migration %s: %w", n, err) + } + tx.Exec(`INSERT INTO schema_migrations VALUES (?,?)`, n, time.Now().Unix()) + if err := tx.Commit(); err != nil { + return err + } + } + return nil +} + +func (s *Store) GetJSON(key string, v any) (bool, error) { + var raw string + err := s.QueryRow(`SELECT value FROM settings WHERE key=?`, key).Scan(&raw) + if err == sql.ErrNoRows { + return false, nil + } + if err != nil { + return false, err + } + return true, json.Unmarshal([]byte(raw), v) +} + +func (s *Store) PutJSON(key string, v any) error { + b, err := json.Marshal(v) + if err != nil { + return err + } + _, err = s.Exec(`INSERT INTO settings(key,value,updated_at) VALUES(?,?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at`, key, string(b), time.Now().Unix()) + return err +} + +func (s *Store) Audit(actor, action, detail, remote string) { + s.Exec(`INSERT INTO audit(ts,actor,action,detail,remote) VALUES(?,?,?,?,?)`, time.Now().UnixMilli(), actor, action, detail, remote) +} diff --git a/migrations/001_init.sql b/migrations/001_init.sql new file mode 100644 index 0000000..cdb3e1b --- /dev/null +++ b/migrations/001_init.sql @@ -0,0 +1,112 @@ +CREATE TABLE settings ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL, + updated_at INTEGER NOT NULL +); + +CREATE TABLE users ( + id INTEGER PRIMARY KEY, + username TEXT NOT NULL UNIQUE, + password_hash TEXT NOT NULL, + totp_secret TEXT NOT NULL DEFAULT '', + totp_enabled INTEGER NOT NULL DEFAULT 0, + created_at INTEGER NOT NULL +); + +CREATE TABLE sessions ( + token_hash TEXT PRIMARY KEY, + user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, + csrf TEXT NOT NULL, + created_at INTEGER NOT NULL, + last_seen INTEGER NOT NULL, + expires_at INTEGER NOT NULL, + remote TEXT NOT NULL DEFAULT '' +); + +CREATE TABLE audit ( + id INTEGER PRIMARY KEY, + ts INTEGER NOT NULL, + actor TEXT NOT NULL, + action TEXT NOT NULL, + detail TEXT NOT NULL DEFAULT '', + remote TEXT NOT NULL DEFAULT '' +); +CREATE INDEX audit_ts ON audit(ts); + +-- A device is keyed by observed MAC address. Randomised MACs mean a physical +-- device may appear as several rows; nothing here merges them automatically. +CREATE TABLE devices ( + mac TEXT PRIMARY KEY, + label TEXT NOT NULL DEFAULT '', + notes TEXT NOT NULL DEFAULT '', + hostname TEXT NOT NULL DEFAULT '', + last_ip TEXT NOT NULL DEFAULT '', + first_seen INTEGER NOT NULL, + last_seen INTEGER NOT NULL +); + +CREATE TABLE capture_sessions ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL, + description TEXT NOT NULL DEFAULT '', + kind TEXT NOT NULL, -- live | import + state TEXT NOT NULL, -- running | stopped | failed | ready + interfaces TEXT NOT NULL DEFAULT '[]', + capture_filter TEXT NOT NULL DEFAULT '', + started_at INTEGER NOT NULL, -- unix ms UTC + ended_at INTEGER NOT NULL DEFAULT 0, + packets INTEGER NOT NULL DEFAULT 0, + bytes INTEGER NOT NULL DEFAULT 0, + drops TEXT NOT NULL DEFAULT '', -- capture-engine drop report, verbatim + error TEXT NOT NULL DEFAULT '', + preserved INTEGER NOT NULL DEFAULT 0, + tags TEXT NOT NULL DEFAULT '', + notes TEXT NOT NULL DEFAULT '', + dir TEXT NOT NULL, + meta TEXT NOT NULL DEFAULT '{}', -- versions, clock sync, parameters + ring_segments INTEGER NOT NULL DEFAULT 0 +); + +CREATE TABLE segments ( + session_id TEXT NOT NULL REFERENCES capture_sessions(id) ON DELETE CASCADE, + seq INTEGER NOT NULL, + file TEXT NOT NULL, + format TEXT NOT NULL, -- pcapng | pcap + base INTEGER NOT NULL, -- global number of the first packet + count INTEGER NOT NULL DEFAULT 0, + bytes INTEGER NOT NULL DEFAULT 0, + sha256 TEXT NOT NULL DEFAULT '', + complete INTEGER NOT NULL DEFAULT 0, + deleted INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (session_id, seq) +); + +-- One summary row per packet. Full dissection is produced on demand by tshark. +CREATE TABLE packets ( + session_id TEXT NOT NULL, + num INTEGER NOT NULL, + seg INTEGER NOT NULL, + off INTEGER NOT NULL, + ts REAL NOT NULL, + src TEXT NOT NULL, + dst TEXT NOT NULL, + proto TEXT NOT NULL, + len INTEGER NOT NULL, + info TEXT NOT NULL, + sport INTEGER, + dport INTEGER, + stream INTEGER, + tcpflags TEXT, + ttl INTEGER, + vlan INTEGER, + color TEXT, + PRIMARY KEY (session_id, num) +) WITHOUT ROWID; + +CREATE TABLE saved_filters ( + id INTEGER PRIMARY KEY, + name TEXT NOT NULL, + kind TEXT NOT NULL, -- display | capture + expr TEXT NOT NULL, + created_at INTEGER NOT NULL +); diff --git a/migrations/embed.go b/migrations/embed.go new file mode 100644 index 0000000..ae7c975 --- /dev/null +++ b/migrations/embed.go @@ -0,0 +1,7 @@ +// Package migrations embeds the ordered SQL schema migrations. +package migrations + +import "embed" + +//go:embed *.sql +var FS embed.FS diff --git a/tests/fixtures/sample-loopback-http-dns.pcapng b/tests/fixtures/sample-loopback-http-dns.pcapng new file mode 100644 index 0000000000000000000000000000000000000000..ee0baf81ab5640456b83cf609814e1dd36693528 GIT binary patch literal 5104 zcmcJTe{56N700hlaDx?Zq)RBHpHtpopxqr zp&=T!Q?bR-(&Bd0&d4C`piNGndw2|su zOtIZk0S~&_JQnzzz!Nan+@Cf^UifM66L5@;w@T7YNPl8CAsy8l zho1&VuuWgAnImnBk;z#TWv;=s`-DO)lS>N~ijY^}9Ix9eK46no5i6-Y>hqra(ap0@ zF)s2aH4yUO+xK=a$U(62#`uH5Uv&Wg)2lkaW^Vc-*}E8>oSm9bNq6GETc!1ZV0dTI zAC+CM!R*yk;BkOsyH}WQGQ2RIChZ7^9!6SrRh|A% z5P796(485^^|io_z@u=6jW-sTHyLJeDLPAw>kRNL6YELGVgtlsQe#3EAvib2^SU+} zGA>@b5qJ*HvGK+j&m<2p#`ZfhHUeX;l(D}%N?o)k8jZL-PLIvnlg%rw)YHiJ4X3Ad zx7+QuSqJli?AQxi2=v8A<+N+J(*wJO&AK-p7jkfMr)#IpYO|uB=r6B{{E?9+3zO(? zPn+(qroRRBzr|~!YhTnv%RrXLH?Q3e{292|cw?GeU6?|~{_}TapeDaiH33El*H=@+ zO&ejO?F-th1A;sz$gOlh#O;Q4yPZB78COzS$<^faG&!4X)~=W$w94{ zl1i$THu`{IgA@l>S8CjM^R`jm^|73g_4u{Cy9If7p4S*+mr#SE#%XnzVH$6s9|wYP zj*U0e*wnUglxf_kGqiO)&(;w%Uhs$S8tI1I3y0dRepG>`V^XrMUXbeX zf)Goyc0{Rl&o*pkGO#@ zHYc0irTN0vz1>Ps;ZkqWw@=95}b`!5gf~(i7`O_?}hO)TEfOnGD4jITQ=D^z+ zA{f@I;qCpw7vUHiZ%n)6^FKgFxHmtg?i)IT)vJt{&*Q^>ugVrdf6Jf2U%UDZ zv>BQ~zpwAo{f4~4bvOy@aGBReUxT7Hb=;on*jxyLYy=x`Oq<(7FCgQi<$b$^f!h34 z)dm=gsLyiU^18irpO_Y?lvQ9y9F-FHx`s)x?TT3m-UiayyihDx(Xz63m#Ws(q*gW5 zq+I{coCO2@=U-;IxkJ}qd*&`5n+RvBp%;oTVt$ghfDN7b=R6k!L+)DhiI$RH?^kCiQhqlnI^^mRc<#4*(p2j9`v#+K2J(~C7wE6o+KDhOAt z?n5P6`xDT?d{oonM$F8_``>h^_?Dp$In?1(pY8)8wtM6Q|Nb(NIedKeccsn`KnKge zme=BkPO|Q;n%}|iN-N)$9-UOyBb`t?6+zC^oSYpK6GDOxkJE(u/dev/null && rm -rf /var/lib/apt/lists/*