diff --git a/internal/api/analysis.go b/internal/api/analysis.go new file mode 100644 index 0000000..276a49a --- /dev/null +++ b/internal/api/analysis.go @@ -0,0 +1,87 @@ +package api + +import ( + "net/http" + "strconv" +) + +func (s *Server) analysisRoutes(priv func(string, http.HandlerFunc)) { + priv("GET /api/captures/{id}/conversations", s.withSession(s.conversations)) + priv("GET /api/captures/{id}/endpoints", s.withSession(s.endpoints)) + priv("GET /api/captures/{id}/statistics", s.withSession(s.statistics)) + priv("GET /api/captures/{id}/follow", s.withSession(s.follow)) + priv("GET /api/captures/{id}/dns", s.withSession(s.dns)) + priv("GET /api/captures/{id}/tls", s.withSession(s.tls)) +} + +func qint(r *http.Request, k string, def int) int { + if v, err := strconv.Atoi(r.URL.Query().Get(k)); err == nil { + return v + } + return def +} + +func (s *Server) conversations(w http.ResponseWriter, r *http.Request, id string) { + t := r.URL.Query().Get("transport") + if t == "" { + t = "tcp" + } + c, err := s.Engine.Conversations(r.Context(), id, t) + if err != nil { + engineErr(w, err) + return + } + writeJSON(w, c) +} + +func (s *Server) endpoints(w http.ResponseWriter, r *http.Request, id string) { + c, err := s.Engine.Endpoints(r.Context(), id, qint(r, "limit", 100)) + if err != nil { + engineErr(w, err) + return + } + writeJSON(w, c) +} + +func (s *Server) statistics(w http.ResponseWriter, r *http.Request, id string) { + st, err := s.Engine.Statistics(r.Context(), id, qint(r, "buckets", 300)) + if err != nil { + engineErr(w, err) + return + } + writeJSON(w, st) +} + +func (s *Server) follow(w http.ResponseWriter, r *http.Request, id string) { + seg := qint(r, "seg", 0) + stream := qint(r, "stream", -1) + if stream < 0 { + httpErr(w, http.StatusBadRequest, "seg and stream are required") + return + } + res, err := s.Engine.FollowTCP(r.Context(), id, seg, stream) + if err != nil { + engineErr(w, err) + return + } + writeJSON(w, res) +} + +func (s *Server) dns(w http.ResponseWriter, r *http.Request, id string) { + recs, err := s.Engine.DNS(r.Context(), id, qint(r, "limit", 5000)) + if err != nil { + engineErr(w, err) + return + } + writeJSON(w, recs) +} + +func (s *Server) tls(w http.ResponseWriter, r *http.Request, id string) { + handshakes, certs, err := s.Engine.TLS(r.Context(), id, qint(r, "limit", 5000)) + if err != nil { + engineErr(w, err) + return + } + writeJSON(w, map[string]any{"handshakes": handshakes, "certificates": certs, + "note": "Observed metadata only. In capture-only mode TLS payloads are not decrypted; interception status is 'encrypted pass-through' for every session."}) +} diff --git a/internal/api/server.go b/internal/api/server.go index d5eb196..1d11799 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -81,6 +81,7 @@ func (s *Server) Handler() http.Handler { s.networkRoutes(priv) s.captureRoutes(priv) + s.analysisRoutes(priv) if s.Web != nil { files := http.FileServerFS(s.Web) diff --git a/internal/capture/analysis.go b/internal/capture/analysis.go new file mode 100644 index 0000000..5ab4791 --- /dev/null +++ b/internal/capture/analysis.go @@ -0,0 +1,570 @@ +package capture + +import ( + "bytes" + "context" + "encoding/hex" + "errors" + "fmt" + "path/filepath" + "sort" + "strconv" + "strings" +) + +// This file derives higher-level views (conversations, streams, DNS, TLS, +// statistics) from data already captured. Conversations and statistics come +// from the SQLite summary index, so they work across segments. Stream payloads +// and DNS/TLS field detail come from tshark over the relevant segment files. + +// ---------------------------------------------------------------- conversations + +type Conversation struct { + Transport string `json:"transport"` // tcp | udp + AAddr string `json:"a_addr"` + APort int `json:"a_port"` + BAddr string `json:"b_addr"` + BPort int `json:"b_port"` + Packets int64 `json:"packets"` + Bytes int64 `json:"bytes"` + BytesAtoB int64 `json:"bytes_a_to_b"` + BytesBtoA int64 `json:"bytes_b_to_a"` + Start float64 `json:"start"` + End float64 `json:"end"` + Protos string `json:"protocols"` + Seg int `json:"seg"` // segment of the first packet (for Follow) + Stream *int `json:"stream"` // tcp.stream within that segment + FirstNum int64 `json:"first_num"` +} + +// Conversations aggregates TCP or UDP conversations from the summary index. +// Endpoints are ordered canonically (A is the lexicographically smaller +// address, or the lower port when addresses match) so each conversation +// appears once regardless of who initiated it. +func (e *Engine) Conversations(ctx context.Context, id, transport string) ([]Conversation, error) { + var where string + switch transport { + case "tcp": + where = "stream IS NOT NULL" + case "udp": + where = "stream IS NULL AND sport IS NOT NULL" + default: + return nil, errors.New("transport must be tcp or udp") + } + // fwd = row is already in canonical (A->B) orientation. + const fwd = "(src TLS, etc. + for _, pre := range []string{"TLSv", "SSLv"} { + if strings.HasPrefix(p, pre) { + base = "TLS" + } + } + if base != "" && !seen[base] { + seen[base] = true + keep = append(keep, base) + } + } + sort.Strings(keep) + return strings.Join(keep, ",") +} + +// ---------------------------------------------------------------- top talkers + +type Endpoint struct { + Addr string `json:"addr"` + Packets int64 `json:"packets"` + Bytes int64 `json:"bytes"` + TxBytes int64 `json:"tx_bytes"` + RxBytes int64 `json:"rx_bytes"` + Start float64 `json:"start"` + End float64 `json:"end"` +} + +func (e *Engine) Endpoints(ctx context.Context, id string, limit int) ([]Endpoint, error) { + if limit <= 0 || limit > 1000 { + limit = 100 + } + // Union of per-direction aggregates keyed by address. + q := ` +SELECT addr, SUM(pk), SUM(by), SUM(tx), SUM(rx), MIN(st), MAX(en) FROM ( + SELECT src AS addr, COUNT(*) pk, SUM(len) by, SUM(len) tx, 0 rx, MIN(ts) st, MAX(ts) en FROM packets WHERE session_id=? GROUP BY src + UNION ALL + SELECT dst AS addr, COUNT(*) pk, SUM(len) by, 0 tx, SUM(len) rx, MIN(ts) st, MAX(ts) en FROM packets WHERE session_id=? GROUP BY dst +) GROUP BY addr ORDER BY SUM(by) DESC LIMIT ?` + rows, err := e.DB.QueryContext(ctx, q, id, id, limit) + if err != nil { + return nil, err + } + defer rows.Close() + out := []Endpoint{} + for rows.Next() { + var p Endpoint + if err := rows.Scan(&p.Addr, &p.Packets, &p.Bytes, &p.TxBytes, &p.RxBytes, &p.Start, &p.End); err != nil { + return nil, err + } + out = append(out, p) + } + return out, rows.Err() +} + +// ---------------------------------------------------------------- statistics + +type Stats struct { + Packets int64 `json:"packets"` + Bytes int64 `json:"bytes"` + Start float64 `json:"start"` + End float64 `json:"end"` + Protocols []NameCount `json:"protocols"` + SizeBuckets []NameCount `json:"size_buckets"` + Timeline []TimeBucket `json:"timeline"` + TCPStreams int64 `json:"tcp_streams"` + Counters map[string]int64 `json:"counters"` +} + +type NameCount struct { + Name string `json:"name"` + Packets int64 `json:"packets"` + Bytes int64 `json:"bytes"` +} + +type TimeBucket struct { + T float64 `json:"t"` + Packets int64 `json:"packets"` + Bytes int64 `json:"bytes"` +} + +func (e *Engine) Statistics(ctx context.Context, id string, buckets int) (Stats, error) { + if buckets <= 0 || buckets > 2000 { + buckets = 300 + } + var s Stats + s.Counters = map[string]int64{} + err := e.DB.QueryRowContext(ctx, `SELECT COUNT(*), COALESCE(SUM(len),0), COALESCE(MIN(ts),0), COALESCE(MAX(ts),0) FROM packets WHERE session_id=?`, id). + Scan(&s.Packets, &s.Bytes, &s.Start, &s.End) + if err != nil || s.Packets == 0 { + return s, err + } + e.DB.QueryRowContext(ctx, `SELECT COUNT(DISTINCT seg||'/'||stream) FROM packets WHERE session_id=? AND stream IS NOT NULL`, id).Scan(&s.TCPStreams) + + rows, _ := e.DB.QueryContext(ctx, `SELECT proto, COUNT(*), SUM(len) FROM packets WHERE session_id=? GROUP BY proto ORDER BY 2 DESC LIMIT 40`, id) + for rows.Next() { + var n NameCount + rows.Scan(&n.Name, &n.Packets, &n.Bytes) + s.Protocols = append(s.Protocols, n) + } + rows.Close() + + // Packet-size distribution (Wireshark's standard buckets). + sizeCase := `CASE + WHEN len<=63 THEN '0-63' WHEN len<=127 THEN '64-127' WHEN len<=255 THEN '128-255' + WHEN len<=511 THEN '256-511' WHEN len<=1023 THEN '512-1023' WHEN len<=1513 THEN '1024-1513' + ELSE '1514+' END` + rows, _ = e.DB.QueryContext(ctx, `SELECT `+sizeCase+` b, COUNT(*), SUM(len) FROM packets WHERE session_id=? GROUP BY b`, id) + order := map[string]int{"0-63": 0, "64-127": 1, "128-255": 2, "256-511": 3, "512-1023": 4, "1024-1513": 5, "1514+": 6} + for rows.Next() { + var n NameCount + rows.Scan(&n.Name, &n.Packets, &n.Bytes) + s.SizeBuckets = append(s.SizeBuckets, n) + } + rows.Close() + sort.Slice(s.SizeBuckets, func(i, j int) bool { return order[s.SizeBuckets[i].Name] < order[s.SizeBuckets[j].Name] }) + + // Timeline: fixed number of equal time buckets. + span := s.End - s.Start + if span <= 0 { + span = 1 + } + width := span / float64(buckets) + rows, _ = e.DB.QueryContext(ctx, `SELECT CAST((ts-?)/? AS INTEGER) b, COUNT(*), SUM(len) FROM packets WHERE session_id=? GROUP BY b ORDER BY b`, s.Start, width, id) + for rows.Next() { + var b int + var n TimeBucket + rows.Scan(&b, &n.Packets, &n.Bytes) + if b >= buckets { + b = buckets - 1 + } + n.T = s.Start + (float64(b)+0.5)*width + s.Timeline = append(s.Timeline, n) + } + rows.Close() + s.Counters["bucket_seconds"] = int64(width) + return s, nil +} + +// ---------------------------------------------------------------- follow TCP stream + +type StreamChunk struct { + Dir string `json:"dir"` // "ab" client→server, "ba" server→client + Hex string `json:"hex"` + Bytes int `json:"bytes"` +} + +type FollowResult struct { + Seg int `json:"seg"` + Stream int `json:"stream"` + Chunks []StreamChunk `json:"chunks"` + AtoB int `json:"a_to_b"` + BtoA int `json:"b_to_a"` + Note string `json:"note,omitempty"` + Endpoint string `json:"endpoint,omitempty"` +} + +// FollowTCP reconstructs one TCP stream's payload using tshark's own reassembly +// (`-z follow,tcp,raw`). Stream numbers are per-segment, so seg selects the file. +func (e *Engine) FollowTCP(ctx context.Context, id string, seg, stream int) (FollowResult, error) { + s, err := e.segmentFor(id, seg) + if err != nil { + return FollowResult{}, err + } + e.flushLive(id) + path := filepath.Join(e.sessionDir(id), s.File) + cmd := e.Tools.Command(ctx, 300, "-n", "-q", "-r", path, "-z", "follow,tcp,raw,"+strconv.Itoa(stream)) + var errb bytes.Buffer + cmd.Stderr = &errb + out, err := cmd.Output() + if err != nil && !strings.Contains(errb.String(), "cut short") { + return FollowResult{}, errors.New(strings.TrimSpace(errb.String() + " " + err.Error())) + } + r := parseFollow(string(out)) + r.Seg, r.Stream = seg, stream + if seg > 1 || !s.Complete { + r.Note = "Reconstructed within segment " + strconv.Itoa(seg) + "; a stream spanning segment boundaries is shown only in part." + } + return r, nil +} + +// parseFollow parses `-z follow,tcp,raw` output. Lines of hex are payload; +// server→client lines are indented with a tab. +func parseFollow(s string) FollowResult { + var r FollowResult + for _, line := range strings.Split(s, "\n") { + if line == "" { + continue + } + if strings.HasPrefix(line, "=") || strings.HasPrefix(line, "Follow:") || strings.HasPrefix(line, "Filter:") || strings.HasPrefix(line, "Node ") { + if strings.HasPrefix(line, "Node ") { + if r.Endpoint != "" { + r.Endpoint += " ↔ " + } + r.Endpoint += strings.TrimSpace(strings.TrimPrefix(line, "Node"))[2:] + } + continue + } + dir := "ab" + h := line + if strings.HasPrefix(line, "\t") { + dir = "ba" + h = strings.TrimPrefix(line, "\t") + } + h = strings.TrimSpace(h) + if _, err := hex.DecodeString(h); err != nil || h == "" { + continue + } + n := len(h) / 2 + if dir == "ab" { + r.AtoB += n + } else { + r.BtoA += n + } + // merge consecutive same-direction chunks for a cleaner view + if len(r.Chunks) > 0 && r.Chunks[len(r.Chunks)-1].Dir == dir { + last := &r.Chunks[len(r.Chunks)-1] + last.Hex += h + last.Bytes += n + } else { + r.Chunks = append(r.Chunks, StreamChunk{Dir: dir, Hex: h, Bytes: n}) + } + } + return r +} + +// ---------------------------------------------------------------- field extraction (DNS, TLS) + +// extractFields runs tshark over every live segment with a display filter and a +// set of fields, returning rows of field values keyed with the global packet +// number. Used by the DNS and TLS explorers. Results are not cached (these are +// opened on demand and the captures are bounded); callers may paginate in the UI. +func (e *Engine) extractFields(ctx context.Context, id, filter string, fields []string, limit int) ([][]string, error) { + segs, err := e.Segments(id) + if err != nil { + return nil, err + } + e.flushLive(id) + args := []string{"-n", "-r", "", "-Y", filter, "-T", "fields", "-E", "separator=/t", "-E", "occurrence=a", "-E", "aggregator=,", "-E", "quote=n", "-e", "frame.number", "-e", "frame.time_epoch"} + for _, f := range fields { + args = append(args, "-e", f) + } + var out [][]string + for _, s := range segs { + if s.Deleted { + continue + } + a := append([]string{}, args...) + a[2] = filepath.Join(e.sessionDir(id), s.File) + cmd := e.Tools.Command(ctx, 600, a...) + var errb bytes.Buffer + cmd.Stderr = &errb + data, err := cmd.Output() + if err != nil && !strings.Contains(errb.String(), "cut short") { + return nil, fmt.Errorf("%s: %s", s.File, strings.TrimSpace(errb.String())) + } + for _, line := range strings.Split(string(data), "\n") { + if line == "" { + continue + } + cells := strings.Split(line, "\t") + // global packet number = segment base + local - 1 + if local, err := strconv.ParseInt(cells[0], 10, 64); err == nil { + cells[0] = strconv.FormatInt(s.Base+local-1, 10) + } + cells[0] = cells[0] + "\x00" + strconv.Itoa(s.Seq) // keep seg for linking + out = append(out, cells) + if len(out) >= limit { + return out, nil + } + } + } + return out, nil +} + +// ---------------------------------------------------------------- DNS explorer + +type DNSRecord struct { + Num int64 `json:"num"` + Seg int `json:"seg"` + Time float64 `json:"time"` + Client string `json:"client"` + Server string `json:"server"` + Query string `json:"query"` + Type string `json:"type"` + Response bool `json:"response"` + RCode string `json:"rcode"` + Answers string `json:"answers"` +} + +func splitNumSeg(s string) (int64, int) { + id, seg, _ := strings.Cut(s, "\x00") + n, _ := strconv.ParseInt(id, 10, 64) + sq, _ := strconv.Atoi(seg) + return n, sq +} + +var dnsTypes = map[string]string{"1": "A", "28": "AAAA", "5": "CNAME", "15": "MX", "16": "TXT", "33": "SRV", "12": "PTR", "65": "HTTPS", "64": "SVCB", "2": "NS", "6": "SOA", "255": "ANY", "257": "CAA", "35": "NAPTR", "43": "DS", "48": "DNSKEY"} +var dnsRcodes = map[string]string{"0": "NOERROR", "1": "FORMERR", "2": "SERVFAIL", "3": "NXDOMAIN", "4": "NOTIMP", "5": "REFUSED"} + +func (e *Engine) DNS(ctx context.Context, id string, limit int) ([]DNSRecord, error) { + if limit <= 0 || limit > 20000 { + limit = 5000 + } + fields := []string{"ip.src", "ipv6.src", "ip.dst", "ipv6.dst", "dns.flags.response", "dns.qry.name", "dns.qry.type", "dns.flags.rcode", "dns.a", "dns.aaaa", "dns.cname", "dns.resp.name", "dns.ptr.domain_name"} + rowsF, err := e.extractFields(ctx, id, "dns && !mdns", fields, limit) + if err != nil { + return nil, err + } + out := []DNSRecord{} + for _, c := range rowsF { + if len(c) < 2+len(fields) { + continue + } + r := DNSRecord{} + r.Num, r.Seg = splitNumSeg(c[0]) + r.Time, _ = strconv.ParseFloat(c[1], 64) + r.Client = firstNonEmpty(c[2], c[3]) + r.Server = firstNonEmpty(c[4], c[5]) + r.Response = c[6] == "1" + r.Query = c[7] + if t, ok := dnsTypes[c[8]]; ok { + r.Type = t + } else { + r.Type = c[8] + } + if rc, ok := dnsRcodes[c[9]]; ok { + r.RCode = rc + } else if c[9] != "" { + r.RCode = "rcode " + c[9] + } + // For a query the "client" is src and "server" is dst; for a response swap for readability. + if r.Response { + r.Client, r.Server = r.Server, r.Client + r.Answers = strings.Trim(strings.Join([]string{c[10], c[11], c[12], c[16]}, " "), " ") + } + out = append(out, r) + } + return out, nil +} + +// ---------------------------------------------------------------- TLS explorer (metadata only) + +type TLSRecord struct { + Num int64 `json:"num"` + Seg int `json:"seg"` + Time float64 `json:"time"` + Client string `json:"client"` + Server string `json:"server"` + ServerPort int `json:"server_port"` + SNI string `json:"sni"` + Version string `json:"version"` + Cipher string `json:"cipher"` + ALPN string `json:"alpn"` + Stream *int `json:"stream"` +} + +type CertRecord struct { + Num int64 `json:"num"` + Seg int `json:"seg"` + Time float64 `json:"time"` + Server string `json:"server"` + Subject string `json:"subject"` + Issuer string `json:"issuer"` + SANs string `json:"sans"` + NotBefore string `json:"not_before"` + NotAfter string `json:"not_after"` +} + +var tlsVersions = map[string]string{"0x0300": "SSL 3.0", "0x0301": "TLS 1.0", "0x0302": "TLS 1.1", "0x0303": "TLS 1.2", "0x0304": "TLS 1.3"} + +// TLS returns observed TLS handshakes: ClientHello (SNI/ALPN) joined with the +// server's chosen version and cipher. This is metadata only; no payload is +// decrypted. The interception status is always "encrypted pass-through" in this +// version, which the UI states plainly. +func (e *Engine) TLS(ctx context.Context, id string, limit int) ([]TLSRecord, []CertRecord, error) { + if limit <= 0 || limit > 20000 { + limit = 5000 + } + // ClientHello: SNI, ALPN, offered; ServerHello: negotiated version + cipher. + chFields := []string{"ip.src", "ipv6.src", "ip.dst", "ipv6.dst", "tcp.dstport", "tcp.srcport", "tcp.stream", "tls.handshake.type", "tls.handshake.extensions_server_name", "tls.handshake.version", "tls.record.version", "tls.handshake.ciphersuite", "tls.handshake.extensions_alpn_str"} + rowsF, err := e.extractFields(ctx, id, "tls.handshake.type==1 || tls.handshake.type==2", chFields, limit) + if err != nil { + return nil, nil, err + } + // Index ServerHello (type 2) by stream to join onto the ClientHello. + type sh struct{ version, cipher string } + byStream := map[string]sh{} + for _, c := range rowsF { + if len(c) < 2+len(chFields) || c[9] != "2" { + continue + } + byStream[c[8]] = sh{version: firstNonEmpty(c[11], c[12]), cipher: c[13]} + } + recs := []TLSRecord{} + for _, c := range rowsF { + if len(c) < 2+len(chFields) || c[9] != "1" { + continue + } + r := TLSRecord{} + r.Num, r.Seg = splitNumSeg(c[0]) + r.Time, _ = strconv.ParseFloat(c[1], 64) + r.Client = firstNonEmpty(c[2], c[3]) + r.Server = firstNonEmpty(c[4], c[5]) + r.ServerPort, _ = strconv.Atoi(c[6]) + r.SNI = c[10] + r.ALPN = c[14] + if st := c[8]; st != "" { + if v, err := strconv.Atoi(st); err == nil { + r.Stream = &v + } + if s, ok := byStream[st]; ok { + r.Version = tlsVersion(s.version) + r.Cipher = cipherName(s.cipher) + } + } + if r.Version == "" { + r.Version = tlsVersion(firstNonEmpty(c[11], c[12])) + } + recs = append(recs, r) + } + + // Certificates (visible in TLS 1.2 and earlier; encrypted in TLS 1.3). + certFields := []string{"ip.src", "ipv6.src", "x509if.printableString", "x509sat.uTF8String", "x509ce.dNSName", "x509af.notBefore", "x509af.notAfter"} + certRows, err := e.extractFields(ctx, id, "tls.handshake.type==11", certFields, limit) + if err != nil { + return recs, nil, nil // handshake records are still useful + } + certs := []CertRecord{} + for _, c := range certRows { + if len(c) < 2+len(certFields) { + continue + } + cr := CertRecord{} + cr.Num, cr.Seg = splitNumSeg(c[0]) + cr.Time, _ = strconv.ParseFloat(c[1], 64) + cr.Server = firstNonEmpty(c[2], c[3]) + subs := strings.Split(firstNonEmpty(c[4], c[5]), ",") + if len(subs) > 0 { + cr.Subject = subs[0] + } + cr.SANs = c[6] + cr.NotBefore, cr.NotAfter = c[7], c[8] + certs = append(certs, cr) + } + return recs, certs, nil +} + +func tlsVersion(code string) string { + if v, ok := tlsVersions[strings.ToLower(code)]; ok { + return v + } + return code +} + +// cipherName renders a cipher-suite code. tshark prints the IANA code number; +// we keep the hex for precision rather than ship a full lookup table. +func cipherName(code string) string { + if code == "" { + return "" + } + if n, err := strconv.Atoi(code); err == nil { + return fmt.Sprintf("0x%04x", n) + } + return code +} diff --git a/internal/capture/engine_test.go b/internal/capture/engine_test.go index d4ed65c..bf6420f 100644 --- a/internal/capture/engine_test.go +++ b/internal/capture/engine_test.go @@ -3,6 +3,7 @@ package capture import ( "bytes" "context" + hexpkg "encoding/hex" "fmt" "net" "os" @@ -196,3 +197,55 @@ func TestLiveCaptureRotation(t *testing.T) { t.Fatal("no live packet events") } } + +func TestAnalysisFromFixture(t *testing.T) { + e := newEngine(t) + ctx := context.Background() + f, _ := os.Open(fixture) + defer f.Close() + s, err := e.Import(ctx, "fixture", f) + if err != nil { + t.Fatal(err) + } + for i := 0; i < 100 && func() bool { x, _ := e.Get(s.ID); return x.State == "importing" }(); i++ { + time.Sleep(50 * time.Millisecond) + } + conv, err := e.Conversations(ctx, s.ID, "tcp") + if err != nil || len(conv) == 0 { + t.Fatalf("tcp conversations: %v %d", err, len(conv)) + } + c := conv[0] + if c.Packets == 0 || c.Bytes == 0 || c.Stream == nil || (c.BytesAtoB+c.BytesBtoA) != c.Bytes { + t.Fatalf("conversation sums wrong: %+v", c) + } + // Follow the first TCP stream and check both directions carry payload. + fr, err := e.FollowTCP(ctx, s.ID, c.Seg, *c.Stream) + if err != nil || fr.AtoB == 0 || fr.BtoA == 0 { + t.Fatalf("follow: %v %+v", err, fr) + } + // The HTTP request bytes must appear in the client->server direction. + var ab string + for _, ch := range fr.Chunks { + if ch.Dir == "ab" { + ab += ch.Hex + } + } + raw, _ := hexpkg.DecodeString(ab) + if !strings.Contains(string(raw), "GET /") { + t.Fatalf("stream payload missing HTTP request") + } + st, err := e.Statistics(ctx, s.ID, 50) + if err != nil || st.Packets == 0 || len(st.Protocols) == 0 || len(st.Timeline) == 0 || st.TCPStreams == 0 { + t.Fatalf("stats: %v %+v", err, st) + } + eps, _ := e.Endpoints(ctx, s.ID, 10) + if len(eps) == 0 || eps[0].Bytes == 0 { + t.Fatalf("endpoints: %+v", eps) + } + if _, err := e.DNS(ctx, s.ID, 100); err != nil { // fixture uses mDNS; real DNS checked on hardware + t.Fatalf("dns: %v", err) + } + if _, _, err := e.TLS(ctx, s.ID, 100); err != nil { + t.Fatalf("tls: %v", err) + } +} diff --git a/web/dist/.gitkeep b/web/dist/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/web/src/App.tsx b/web/src/App.tsx index c86ad92..af6d235 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -9,6 +9,10 @@ import Library from "./pages/Library"; import NetworkPage from "./pages/Network"; import Devices from "./pages/Devices"; import Health from "./pages/Health"; +import Conversations from "./pages/Conversations"; +import DnsExplorer from "./pages/DnsExplorer"; +import TlsExplorer from "./pages/TlsExplorer"; +import Statistics from "./pages/Statistics"; import Settings from "./pages/Settings"; export interface Prefs { theme: "dark" | "light" | "system"; density: "compact" | "comfortable"; scale: number; motion: "system" | "reduce" } @@ -20,6 +24,10 @@ interface Sess { user: string; csrf: string; totp_enabled: boolean; setup_comple const NAV: [string, string, string][] = [ ["/", "Overview", "◉"], ["/analyse", "Live Capture", "≋"], + ["/conversations", "Conversations", "⇌"], + ["/dns", "DNS Explorer", "◇"], + ["/tls", "TLS Explorer", "⛨"], + ["/statistics", "Statistics", "▦"], ["/library", "Capture Library", "▤"], ["/devices", "Devices", "⌬"], ["", "", ""], @@ -85,6 +93,10 @@ export default function App() { const page = (() => { if (path === "/" ) return ; if (path.startsWith("/analyse")) return ; + if (path.startsWith("/conversations")) return ; + if (path.startsWith("/dns")) return ; + if (path.startsWith("/tls")) return ; + if (path.startsWith("/statistics")) return ; if (path === "/library") return ; if (path === "/devices") return ; if (path === "/network") return ; diff --git a/web/src/api.ts b/web/src/api.ts index a3e2c5e..690e1c5 100644 --- a/web/src/api.ts +++ b/web/src/api.ts @@ -84,3 +84,15 @@ export const fmtDur = (s: number) => { const h = Math.floor(s / 3600), m = Math.floor((s % 3600) / 60), x = Math.floor(s % 60); return h ? `${h}h ${m}m` : m ? `${m}m ${x}s` : `${x}s`; }; + +// ---- Phase 3 analysis types ---- +export interface Conversation { transport: string; a_addr: string; a_port: number; b_addr: string; b_port: number; packets: number; bytes: number; bytes_a_to_b: number; bytes_b_to_a: number; start: number; end: number; protocols: string; seg: number; stream: number | null; first_num: number } +export interface Endpoint { addr: string; packets: number; bytes: number; tx_bytes: number; rx_bytes: number; start: number; end: number } +export interface NameCount { name: string; packets: number; bytes: number } +export interface TimeBucket { t: number; packets: number; bytes: number } +export interface Stats { packets: number; bytes: number; start: number; end: number; protocols: NameCount[]; size_buckets: NameCount[]; timeline: TimeBucket[]; tcp_streams: number; counters: Record } +export interface StreamChunk { dir: string; hex: string; bytes: number } +export interface FollowResult { seg: number; stream: number; chunks: StreamChunk[]; a_to_b: number; b_to_a: number; note?: string; endpoint?: string } +export interface DNSRecord { num: number; seg: number; time: number; client: string; server: string; query: string; type: string; response: boolean; rcode: string; answers: string } +export interface TLSRecord { num: number; seg: number; time: number; client: string; server: string; server_port: number; sni: string; version: string; cipher: string; alpn: string; stream: number | null } +export interface CertRecord { num: number; seg: number; time: number; server: string; subject: string; issuer: string; sans: string; not_before: string; not_after: string } diff --git a/web/src/components/chrome.tsx b/web/src/components/chrome.tsx new file mode 100644 index 0000000..179c31d --- /dev/null +++ b/web/src/components/chrome.tsx @@ -0,0 +1,37 @@ +// Shared chrome for the analysis pages: a session picker header, and a tiny +// decode-vs-encrypted banner used so the UI is honest about what is being shown. +import { fmtBytes, type Session } from "../api"; +import { Badge, go } from "../ui"; + +export function SessionPicker({ sessions, id, base, children }: { sessions?: Session[]; id: string; base: string; children?: React.ReactNode }) { + return ( +
+ + {children} +
+ ); +} + +export function PacketsLink({ id, filter, children }: { id: string; filter: string; children: React.ReactNode }) { + return { e.preventDefault(); go(`/analyse/${id}?filter=${encodeURIComponent(filter)}`); }} title={filter}>{children}; +} + +export function Bar({ value, max, label, kind }: { value: number; max: number; label?: string; kind?: string }) { + const pct = max > 0 ? Math.max(1.5, (value / max) * 100) : 0; + return ( +
+
+ {label ?? fmtBytes(value)} +
+ ); +} + +export function EncryptedNote() { + return
+ Encrypted traffic is shown as metadata only. + In capture-only mode PiLens does not decrypt TLS/QUIC. You see endpoints, SNI, versions, ciphers, ALPN and certificates where they are sent in the clear — not application content. interception: encrypted pass-through +
; +} diff --git a/web/src/pages/Conversations.tsx b/web/src/pages/Conversations.tsx new file mode 100644 index 0000000..fc22f80 --- /dev/null +++ b/web/src/pages/Conversations.tsx @@ -0,0 +1,139 @@ +import { useEffect, useMemo, useState } from "react"; +import { api, fmtBytes, fmtDur, type Conversation, type FollowResult, type Session, type StreamChunk } from "../api"; +import { Badge, Modal, copyText, go, usePoll, usePersist } from "../ui"; +import { SessionPicker, Bar, PacketsLink } from "../components/chrome"; + +export default function Conversations({ sessionId }: { sessionId: string }) { + const [sessions] = usePoll("/api/captures", 5000); + const id = sessionId || sessions?.[0]?.id || ""; + const [transport, setTransport] = usePersist<"tcp" | "udp">("conv.transport", "tcp"); + const [convs, err] = usePoll(id ? `/api/captures/${id}/conversations?transport=${transport}` : null); + const [q, setQ] = useState(""); + const [sort, setSort] = useState("bytes"); + const [follow, setFollow] = useState(null); + + const rows = useMemo(() => { + let r = (convs ?? []).filter((c) => !q || `${c.a_addr}:${c.a_port} ${c.b_addr}:${c.b_port} ${c.protocols}`.toLowerCase().includes(q.toLowerCase())); + r = [...r].sort((a, b) => (b[sort] as number) - (a[sort] as number) || 0); + return r; + }, [convs, q, sort, transport]); + const maxBytes = rows[0]?.bytes ?? 1; + + const convFilter = (c: Conversation) => { + const ipA = c.a_addr.includes(":") ? "ipv6.addr" : "ip.addr"; + const p = c.transport; + return `${ipA}==${c.a_addr} && ${ipA}==${c.b_addr} && ${p}.port==${c.a_port} && ${p}.port==${c.b_port}`; + }; + + return ( +
+
+

Conversations

Traffic grouped by endpoint pair, from the packet index. Counts agree with the packet list (subject to capture drops).

+ +
+
+
+ + +
+ setQ(e.target.value)} /> + {rows.length.toLocaleString()} conversations +
+ {err &&
{err}
} +
+ + + + {([["packets", "Packets"], ["bytes", "Bytes"], ["bytes_a_to_b", "A→B"], ["bytes_b_to_a", "B→A"]] as [keyof Conversation, string][]).map(([k, l]) => + )} + + + + {rows.map((c, i) => ( + + + + + + + + + + + + ))} + {rows.length === 0 && } + +
Endpoint AEndpoint BProtocols setSort(k)}>{l}{sort === k ? " ▾" : ""}Duration
{c.a_addr}:{c.a_port}{c.b_addr}:{c.b_port}{c.protocols.split(",").map((p) => {p})}{c.packets.toLocaleString()}{fmtBytes(c.bytes_a_to_b)}{fmtBytes(c.bytes_b_to_a)}{fmtDur(c.end - c.start)}
+ {transport === "tcp" && c.stream != null && } + +
No {transport.toUpperCase()} conversations.
+
+ {follow && setFollow(null)} onPackets={(f) => go(`/analyse/${id}?filter=${encodeURIComponent(f)}`)} />} +
+ ); +} + +const printable = (b: number) => (b >= 0x20 && b < 0x7f) || b === 0x0a || b === 0x09; +function hexToText(hex: string) { + let s = ""; + for (let i = 0; i < hex.length; i += 2) { const b = parseInt(hex.substr(i, 2), 16); s += printable(b) ? String.fromCharCode(b) : "."; } + return s; +} +function hexDump(hex: string) { + const out: string[] = []; + for (let off = 0; off < hex.length; off += 32) { + const slice = hex.substr(off, 32); + const pairs = slice.match(/../g) ?? []; + out.push((off / 2).toString(16).padStart(6, "0") + " " + pairs.join(" ").padEnd(47) + " " + hexToText(slice)); + } + return out.join("\n"); +} + +function FollowModal({ id, conv, onClose, onPackets }: { id: string; conv: Conversation; onClose: () => void; onPackets: (f: string) => void }) { + const [res, setRes] = useState(null); + const [err, setErr] = useState(""); + const [view, setView] = useState<"ascii" | "hex">("ascii"); + const [dir, setDir] = useState<"both" | "ab" | "ba">("both"); + + useEffect(() => { + setRes(null); setErr(""); + api(`/api/captures/${id}/follow?seg=${conv.seg}&stream=${conv.stream}`) + .then(setRes).catch((e) => setErr(e.message)); + }, [id, conv]); + + const shown = (res?.chunks ?? []).filter((c) => dir === "both" || c.dir === dir); + const copyAll = () => copyText(shown.map((c) => view === "hex" ? c.hex : hexToText(c.hex)).join("")); + + return ( + + + + }> +
+ {conv.a_addr}:{conv.a_port} ↔ {conv.b_addr}:{conv.b_port} +
+ + + +
+
+ {res &&
{fmtBytes(res.a_to_b)} client→server · {fmtBytes(res.b_to_a)} server→client{res.note ? " · " + res.note : ""}
} + {err &&
{err}
} + {!res && !err &&
Reassembling…
} + {res && ( + 
+          {shown.map((c, i) => )}
+          {shown.length === 0 && No payload in this direction (handshake/empty).}
+
+ )} +

Reconstructed from the captured TCP payload by tshark. This is the transport stream, not decrypted application content: for TLS it is ciphertext.

+ + ); +} + +function ChunkView({ chunk, view }: { chunk: StreamChunk; view: "ascii" | "hex" }) { + const col = chunk.dir === "ab" ? "var(--cyan)" : "var(--accent)"; + const text = view === "hex" ? hexDump(chunk.hex) : hexToText(chunk.hex); + return {text}{view === "ascii" ? "" : "\n"}; +} diff --git a/web/src/pages/DnsExplorer.tsx b/web/src/pages/DnsExplorer.tsx new file mode 100644 index 0000000..1e8586c --- /dev/null +++ b/web/src/pages/DnsExplorer.tsx @@ -0,0 +1,83 @@ +import { useMemo, useState } from "react"; +import { fmtTime, type DNSRecord, type Session } from "../api"; +import { Badge, usePoll } from "../ui"; +import { SessionPicker, PacketsLink } from "../components/chrome"; + +export default function DnsExplorer({ sessionId }: { sessionId: string }) { + const [sessions] = usePoll("/api/captures", 5000); + const id = sessionId || sessions?.[0]?.id || ""; + const [recs, err] = usePoll(id ? `/api/captures/${id}/dns?limit=5000` : null); + const [q, setQ] = useState(""); + const [onlyErr, setOnlyErr] = useState(false); + const [grouped, setGrouped] = useState(true); + + const rows = useMemo(() => (recs ?? []).filter((r) => + (!q || `${r.query} ${r.client} ${r.answers}`.toLowerCase().includes(q.toLowerCase())) && + (!onlyErr || (r.response && r.rcode && r.rcode !== "NOERROR"))), [recs, q, onlyErr]); + + // Group queries with their most useful response per (client, name, type). + const names = useMemo(() => { + const m = new Map; answers: Set; rcodes: Set; last: number }>(); + for (const r of rows) { + const k = r.query + "|" + r.type; + let g = m.get(k); + if (!g) { g = { name: r.query, type: r.type, count: 0, clients: new Set(), answers: new Set(), rcodes: new Set(), last: 0 }; m.set(k, g); } + g.count++; if (r.client) g.clients.add(r.client); g.last = Math.max(g.last, r.time); + if (r.answers) r.answers.split(/\s+/).filter(Boolean).forEach((a) => g!.answers.add(a)); + if (r.response && r.rcode) g.rcodes.add(r.rcode); + } + return [...m.values()].sort((a, b) => b.count - a.count); + }, [rows]); + + return ( +
+
+

DNS Explorer

Cleartext DNS observed through the appliance. Encrypted DNS (DoH/DoT) is not shown here; it appears as TLS/HTTPS to the resolver.

+ +
+
+ setQ(e.target.value)} /> + + + {rows.length.toLocaleString()} messages +
+ {err &&
{err}
} +
+ {grouped ? ( + + + {names.map((g, i) => ( + + + + + + + + + + ))} + {names.length === 0 && } + +
NameTypeQueriesClientsAnswersResultLast seen
{g.name || "(root)"}{g.type}{g.count}{[...g.clients].join(", ")}{[...g.answers].slice(0, 6).join(", ")}{g.answers.size > 6 ? ` +${g.answers.size - 6}` : ""}{[...g.rcodes].map((rc) => {rc})}{fmtTime(g.last * 1000)}
No cleartext DNS observed.
+ ) : ( + + + {rows.map((r, i) => ( + + + + + + + + + ))} + {rows.length === 0 && } + +
TimeClientResolverNameTypeResult / answers
{fmtTime(r.time * 1000)}{r.response ? resp : query}{r.client}{r.server}{r.query}{r.type}{r.response ? (r.rcode && r.rcode !== "NOERROR" ? r.rcode : r.answers) : ""}
No cleartext DNS observed.
+ )} +
+
+ ); +} diff --git a/web/src/pages/Statistics.tsx b/web/src/pages/Statistics.tsx new file mode 100644 index 0000000..052fff9 --- /dev/null +++ b/web/src/pages/Statistics.tsx @@ -0,0 +1,113 @@ +import { useMemo, useState } from "react"; +import { fmtBytes, type Endpoint, type Session, type Stats, type TimeBucket } from "../api"; +import { usePoll } from "../ui"; +import { SessionPicker, Bar, PacketsLink } from "../components/chrome"; + +export default function Statistics({ sessionId }: { sessionId: string }) { + const [sessions] = usePoll("/api/captures", 5000); + const id = sessionId || sessions?.[0]?.id || ""; + const [stats, err] = usePoll(id ? `/api/captures/${id}/statistics?buckets=300` : null, 0); + const [eps] = usePoll(id ? `/api/captures/${id}/endpoints?limit=15` : null); + const [metric, setMetric] = useState<"bytes" | "packets">("bytes"); + + const dur = stats ? stats.end - stats.start : 0; + const bucketS = stats?.counters?.bucket_seconds || 1; + const maxProto = stats?.protocols[0]?.[metric] ?? 1; + const maxSize = useMemo(() => Math.max(1, ...(stats?.size_buckets ?? []).map((s) => s[metric])), [stats, metric]); + const maxEp = eps?.[0]?.bytes ?? 1; + + return ( +
+
+

Statistics

Computed from the packet index. Every value is a count of captured packets or bytes, not an estimate.

+ +
+ + +
+ +
+ {err &&
{err}
} + {!stats ?
Loading…
: stats.packets === 0 ?
No packets captured yet.
: ( + <> +
+

Packets

{stats.packets.toLocaleString()}
+

Bytes

{fmtBytes(stats.bytes)}
+

Duration

{dur > 0 ? dur.toFixed(1) + " s" : "–"}
{dur > 0 ? `${Math.round(stats.packets / dur)} pkt/s avg · ${fmtBytes(stats.bytes / dur)}/s` : ""}
+

TCP streams

{stats.tcp_streams.toLocaleString()}
+
+ +
+

Traffic over time

{metric} per {bucketS >= 1 ? bucketS + " s" : "bucket"}
+ +
+ +
+

Protocol hierarchy

+ + {stats.protocols.slice(0, 15).map((p) => ( + + + + ))} +
{p.name}{((p[metric] / (metric === "bytes" ? stats.bytes : stats.packets)) * 100).toFixed(1)}%
+
+

Top talkers

+ + {eps?.map((e) => ( + + + + ))} +
AddressTrafficTx / Rx
{e.addr}{fmtBytes(e.tx_bytes)} / {fmtBytes(e.rx_bytes)}
+
+

Packet size distribution

+ + {stats.size_buckets.map((b) => ( + + + ))} +
{b.name}
+
+

Definitions

+
+
Protocol
The highest-layer protocol TShark assigned to each packet (one packet counts once).
+
TCP streams
Distinct tshark stream indices, counted per segment, so a connection crossing a segment boundary counts more than once.
+
Tx / Rx
Bytes with this address as source / destination. A packet contributes to both its endpoints.
+
Bytes
On-the-wire frame length as captured.
+
+
+
+ + )} +
+ ); +} + +function Timeline({ buckets, metric, t0 }: { buckets: TimeBucket[]; metric: "bytes" | "packets"; t0: number }) { + const [hover, setHover] = useState(null); + const W = 1000, H = 160, pad = 4; + if (!buckets.length) return
No data
; + const max = Math.max(1, ...buckets.map((b) => b[metric])); + const n = buckets.length; + const bw = (W - pad * 2) / n; + const bars = buckets.map((b, i) => { + const h = (b[metric] / max) * (H - 20); + return setHover(i)} onMouseLeave={() => setHover(null)} />; + }); + const hb = hover != null ? buckets[hover] : null; + return ( +
+ + {bars} + + +
+ 0 s + {hb && +{(hb.t - t0).toFixed(1)} s: {metric === "bytes" ? fmtBytes(hb.bytes) : hb.packets + " pkts"}} + {(buckets[n - 1].t - t0).toFixed(0)} s +
+
+ ); +} diff --git a/web/src/pages/TlsExplorer.tsx b/web/src/pages/TlsExplorer.tsx new file mode 100644 index 0000000..c651a3b --- /dev/null +++ b/web/src/pages/TlsExplorer.tsx @@ -0,0 +1,75 @@ +import { useMemo, useState } from "react"; +import { fmtTime, type CertRecord, type Session, type TLSRecord } from "../api"; +import { Badge, usePoll } from "../ui"; +import { SessionPicker, PacketsLink, EncryptedNote } from "../components/chrome"; + +interface TLSResp { handshakes: TLSRecord[]; certificates: CertRecord[]; note: string } + +export default function TlsExplorer({ sessionId }: { sessionId: string }) { + const [sessions] = usePoll("/api/captures", 5000); + const id = sessionId || sessions?.[0]?.id || ""; + const [data, err] = usePoll(id ? `/api/captures/${id}/tls?limit=5000` : null); + const [q, setQ] = useState(""); + const [tab, setTab] = useState<"handshakes" | "certs">("handshakes"); + + const hs = useMemo(() => (data?.handshakes ?? []).filter((r) => !q || `${r.sni} ${r.server} ${r.client} ${r.version} ${r.alpn}`.toLowerCase().includes(q.toLowerCase())), [data, q]); + const certs = useMemo(() => (data?.certificates ?? []).filter((r) => !q || `${r.subject} ${r.sans} ${r.server}`.toLowerCase().includes(q.toLowerCase())), [data, q]); + const verCount = useMemo(() => { const m: Record = {}; for (const r of data?.handshakes ?? []) if (r.version) m[r.version] = (m[r.version] || 0) + 1; return m; }, [data]); + + return ( +
+
+

TLS Explorer

Observed TLS handshakes and any certificates sent in the clear.

+ +
+ + {Object.keys(verCount).length > 0 &&
{Object.entries(verCount).map(([v, n]) => {v}: {n})}
} +
+
+ + +
+ setQ(e.target.value)} /> +
+ {err &&
{err}
} +
+ {tab === "handshakes" ? ( + + + {hs.map((r, i) => ( + + + + + + + + + + + + ))} + {hs.length === 0 && } + +
TimeClientServerSNIVersionCipherALPNStatus
{fmtTime(r.time * 1000)}{r.client}{r.server}:{r.server_port}{r.sni || — (none / ECH)}{r.version ? {r.version} : ?}{r.cipher}{r.alpn}encrypted pass-through{r.stream != null ? : ↗}
No TLS handshakes observed.
+ ) : ( + + + {certs.map((r, i) => ( + + + + + + + + + ))} + {certs.length === 0 && } + +
TimeServerSubjectSubject alt namesValid fromValid to
{fmtTime(r.time * 1000)}{r.server}{r.subject}{r.sans}{r.not_before}{r.not_after}↗
No certificates seen in the clear. TLS 1.3 encrypts the certificate, so this is expected for modern servers.
+ )} +
+
+ ); +}