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.
This commit is contained in:
Andy Dixon
@@ -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<?`, id, s.Base, s.Base+s.Count)
|
||||
}
|
||||
e.dropFilterCache(id)
|
||||
}
|
||||
Reference in New Issue
Block a user