feat: add embedded web UI

Adds an optional web dashboard served by OpenGFW itself, enabled with a new `web` section in the config file. Backend (web package, decoupled from engine/io so it builds on any OS): - hub.go collects statistics off the engine logger callbacks: atomic counters, two ring-buffered time series (10s and 1min buckets), top N hosts/blocked destinations/rules/analyzers, and a 512 entry event buffer fanned out to connected clients over SSE. Slow clients drop frames instead of blocking the engine. - api.go exposes /api/v1 for info, meta, metrics, events, the SSE stream and ruleset read/validate/replace. - auth.go implements password login with in-memory session tokens and login rate limiting. Mutating endpoints require the bearer token (the session cookie is only accepted for GET), which makes them CSRF-safe. - cmd/web.go implements the rule manager: rules are compiled before anything is written, the file is replaced atomically and the engine is hot reloaded. The SIGHUP handler now shares that same path. - web/devserver serves the UI with synthetic traffic for frontend work on machines where the engine itself cannot be built. Frontend (web/frontend, Vue 3 + Vite + Tailwind CSS v4 + Reka UI): dashboard, live event feed with analyzer property inspection, visual and YAML rule editors, analyzer overview and settings. Responsive down to phone sizes with a bottom tab bar and bottom-sheet dialogs, plus light/dark themes and English/Chinese translations. The built UI in web/dist is committed and embedded with go:embed so that `go build` works without Node; CI builds the frontend and checks that the committed output is up to date. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-07-27 08:09:11 +08:00
parent 393c29bd2d
commit 5a7722d1d2
94 changed files with 8534 additions and 23 deletions
@@ -0,0 +1,513 @@
+package web
+
+import (
+	"reflect"
+	"sort"
+	"strconv"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+// PropMap is a set of properties produced by a single analyzer.
+type PropMap = map[string]interface{}
+
+// Props is the combined property map of all analyzers of a stream.
+type Props = map[string]PropMap
+
+// Event kinds.
+const (
+	EventKindAction = "action"
+	EventKindLog    = "log"
+	EventKindError  = "error"
+)
+
+// Event is a single entry of the live event feed shown by the web UI.
+type Event struct {
+	Seq     uint64 `json:"seq"`
+	Time    int64  `json:"time"` // Unix milliseconds
+	Kind    string `json:"kind"`
+	ID      int64  `json:"id"`
+	Proto   string `json:"proto"`
+	SrcIP   string `json:"srcIP"`
+	SrcPort uint16 `json:"srcPort"`
+	DstIP   string `json:"dstIP"`
+	DstPort uint16 `json:"dstPort"`
+	Action  string `json:"action,omitempty"`
+	Rule    string `json:"rule,omitempty"`
+	Host    string `json:"host,omitempty"`
+	Message string `json:"message,omitempty"`
+	Props   Props  `json:"props,omitempty"`
+}
+
+// Bucket is a single time slot of the traffic time series.
+type Bucket struct {
+	Time    int64  `json:"time"` // Unix milliseconds, start of the bucket
+	TCP     uint64 `json:"tcp"`
+	UDP     uint64 `json:"udp"`
+	Blocked uint64 `json:"blocked"`
+	Allowed uint64 `json:"allowed"`
+}
+
+// NameCount is a name/count pair used by the various "top N" lists.
+type NameCount struct {
+	Name  string `json:"name"`
+	Count uint64 `json:"count"`
+}
+
+const (
+	defaultEventBufferSize = 512
+	subscriberQueueSize    = 256
+	maxCounterEntries      = 4096
+	topNLimit              = 10
+)
+
+// Hub collects statistics from the engine and fans live events out to the
+// connected web UI clients. All methods are safe for concurrent use, and the
+// ones called from the engine hot path are kept as cheap as possible.
+type Hub struct {
+	startTime time.Time
+
+	// Hot path counters. Updated with atomics, no locking involved.
+	tcpStreams   atomic.Uint64
+	udpStreams   atomic.Uint64
+	actAllow     atomic.Uint64
+	actBlock     atomic.Uint64
+	actDrop      atomic.Uint64
+	actModify    atomic.Uint64
+	actMaybe     atomic.Uint64
+	errors       atomic.Uint64
+	workersAlive atomic.Int64
+
+	mu       sync.Mutex
+	seq      uint64
+	events   []*Event // ring buffer, oldest first once wrapped
+	evHead   int
+	evFilled bool
+
+	short *series // 10 seconds per bucket, 10 minutes total
+	long  *series // 1 minute per bucket, 1 hour total
+
+	hosts     *counterMap
+	dstIPs    *counterMap
+	rules     *counterMap
+	analyzers *counterMap
+
+	subs map[int]chan *Event
+	next int
+}
+
+// NewHub creates a new statistics hub.
+func NewHub() *Hub {
+	return &Hub{
+		startTime: time.Now(),
+		events:    make([]*Event, defaultEventBufferSize),
+		short:     newSeries(10*time.Second, 60),
+		long:      newSeries(time.Minute, 60),
+		hosts:     newCounterMap(),
+		dstIPs:    newCounterMap(),
+		rules:     newCounterMap(),
+		analyzers: newCounterMap(),
+		subs:      make(map[int]chan *Event),
+	}
+}
+
+// WorkerStarted / WorkerStopped track the number of running workers.
+func (h *Hub) WorkerStarted() { h.workersAlive.Add(1) }
+func (h *Hub) WorkerStopped() { h.workersAlive.Add(-1) }
+
+// StreamNew records a newly seen stream.
+func (h *Hub) StreamNew(proto string) {
+	if proto == "udp" {
+		h.udpStreams.Add(1)
+	} else {
+		h.tcpStreams.Add(1)
+	}
+	h.mu.Lock()
+	now := time.Now()
+	h.short.add(now, proto, "")
+	h.long.add(now, proto, "")
+	h.mu.Unlock()
+}
+
+// PropUpdate records the analyzer properties of a stream. It is used to build
+// the "top hosts" list and the per-analyzer hit counters.
+func (h *Hub) PropUpdate(props Props) {
+	if len(props) == 0 {
+		return
+	}
+	host := HostOf(props)
+	h.mu.Lock()
+	for name := range props {
+		h.analyzers.inc(name)
+	}
+	if host != "" {
+		h.hosts.inc(host)
+	}
+	h.mu.Unlock()
+}
+
+// StreamAction records the verdict of a stream and, unless the verdict is the
+// neutral "maybe", publishes an event to the live feed.
+func (h *Hub) StreamAction(info StreamInfo, action string) {
+	switch action {
+	case "allow":
+		h.actAllow.Add(1)
+	case "block":
+		h.actBlock.Add(1)
+	case "drop":
+		h.actDrop.Add(1)
+	case "modify":
+		h.actModify.Add(1)
+	default:
+		h.actMaybe.Add(1)
+		return
+	}
+	now := time.Now()
+	h.mu.Lock()
+	h.short.add(now, "", action)
+	h.long.add(now, "", action)
+	if action == "block" || action == "drop" {
+		if info.DstIP != "" {
+			h.dstIPs.inc(info.DstIP)
+		}
+	}
+	h.mu.Unlock()
+	h.publish(now, EventKindAction, info, action, "", "")
+}
+
+// RuleLog records a `log: true` hit of a ruleset rule.
+func (h *Hub) RuleLog(info StreamInfo, rule string) {
+	h.mu.Lock()
+	h.rules.inc(rule)
+	h.mu.Unlock()
+	h.publish(time.Now(), EventKindLog, info, "", rule, "")
+}
+
+// Error records an engine or ruleset error.
+func (h *Hub) Error(info StreamInfo, rule, message string) {
+	h.errors.Add(1)
+	h.publish(time.Now(), EventKindError, info, "", rule, message)
+}
+
+// StreamInfo is the subset of ruleset.StreamInfo the hub cares about.
+type StreamInfo struct {
+	ID      int64
+	Proto   string
+	SrcIP   string
+	SrcPort uint16
+	DstIP   string
+	DstPort uint16
+	Props   Props
+}
+
+func (h *Hub) publish(now time.Time, kind string, info StreamInfo, action, rule, message string) {
+	h.mu.Lock()
+	h.seq++
+	ev := &Event{
+		Seq:     h.seq,
+		Time:    now.UnixMilli(),
+		Kind:    kind,
+		ID:      info.ID,
+		Proto:   info.Proto,
+		SrcIP:   info.SrcIP,
+		SrcPort: info.SrcPort,
+		DstIP:   info.DstIP,
+		DstPort: info.DstPort,
+		Action:  action,
+		Rule:    rule,
+		Message: message,
+		Host:    HostOf(info.Props),
+		Props:   info.Props,
+	}
+	h.events[h.evHead] = ev
+	h.evHead = (h.evHead + 1) % len(h.events)
+	if h.evHead == 0 {
+		h.evFilled = true
+	}
+	for _, ch := range h.subs {
+		select {
+		case ch <- ev:
+		default: // Slow client, drop the event rather than blocking the engine
+		}
+	}
+	h.mu.Unlock()
+}
+
+// Subscribe returns a channel of live events and a function to unsubscribe.
+func (h *Hub) Subscribe() (<-chan *Event, func()) {
+	ch := make(chan *Event, subscriberQueueSize)
+	h.mu.Lock()
+	id := h.next
+	h.next++
+	h.subs[id] = ch
+	h.mu.Unlock()
+	return ch, func() {
+		h.mu.Lock()
+		if c, ok := h.subs[id]; ok {
+			delete(h.subs, id)
+			close(c)
+		}
+		h.mu.Unlock()
+	}
+}
+
+// Events returns up to limit most recent events, newest first.
+func (h *Hub) Events(limit int) []*Event {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	size := len(h.events)
+	n := h.evHead
+	if h.evFilled {
+		n = size
+	}
+	if limit <= 0 || limit > n {
+		limit = n
+	}
+	out := make([]*Event, 0, limit)
+	for i := 0; i < limit; i++ {
+		idx := (h.evHead - 1 - i + size*2) % size
+		if h.events[idx] == nil {
+			break
+		}
+		out = append(out, h.events[idx])
+	}
+	return out
+}
+
+// Metrics is the snapshot of everything the dashboard needs.
+type Metrics struct {
+	UptimeSeconds int64       `json:"uptimeSeconds"`
+	Workers       int64       `json:"workers"`
+	Streams       StreamStats `json:"streams"`
+	Actions       ActionStats `json:"actions"`
+	Errors        uint64      `json:"errors"`
+	Short         []Bucket    `json:"short"`
+	Long          []Bucket    `json:"long"`
+	TopHosts      []NameCount `json:"topHosts"`
+	TopBlockedIPs []NameCount `json:"topBlockedIPs"`
+	TopRules      []NameCount `json:"topRules"`
+	Analyzers     []NameCount `json:"analyzers"`
+}
+
+type StreamStats struct {
+	TCP   uint64 `json:"tcp"`
+	UDP   uint64 `json:"udp"`
+	Total uint64 `json:"total"`
+}
+
+type ActionStats struct {
+	Allow  uint64 `json:"allow"`
+	Block  uint64 `json:"block"`
+	Drop   uint64 `json:"drop"`
+	Modify uint64 `json:"modify"`
+	Maybe  uint64 `json:"maybe"`
+}
+
+// Metrics returns a consistent snapshot of the collected statistics.
+func (h *Hub) Metrics() Metrics {
+	tcp, udp := h.tcpStreams.Load(), h.udpStreams.Load()
+	m := Metrics{
+		UptimeSeconds: int64(time.Since(h.startTime).Seconds()),
+		Workers:       h.workersAlive.Load(),
+		Streams:       StreamStats{TCP: tcp, UDP: udp, Total: tcp + udp},
+		Actions: ActionStats{
+			Allow:  h.actAllow.Load(),
+			Block:  h.actBlock.Load(),
+			Drop:   h.actDrop.Load(),
+			Modify: h.actModify.Load(),
+			Maybe:  h.actMaybe.Load(),
+		},
+		Errors: h.errors.Load(),
+	}
+	now := time.Now()
+	h.mu.Lock()
+	m.Short = h.short.snapshot(now)
+	m.Long = h.long.snapshot(now)
+	m.TopHosts = h.hosts.top(topNLimit)
+	m.TopBlockedIPs = h.dstIPs.top(topNLimit)
+	m.TopRules = h.rules.top(topNLimit)
+	m.Analyzers = h.analyzers.top(maxCounterEntries)
+	h.mu.Unlock()
+	return m
+}
+
+// HostOf extracts the most meaningful hostname out of a property map, so that
+// the UI can show something more useful than a bare IP address.
+func HostOf(props Props) string {
+	if props == nil {
+		return ""
+	}
+	for _, path := range [][]string{
+		{"tls", "req", "sni"},
+		{"quic", "req", "sni"},
+		{"http", "req", "headers", "host"},
+		{"dns", "questions", "0", "name"},
+	} {
+		if s := nestedString(props, path...); s != "" {
+			return s
+		}
+	}
+	return ""
+}
+
+// nestedString walks a path of keys (or numeric slice indices) and returns the
+// value at the end of it if it is a string. Analyzers use named map types, so
+// the lookups go through reflection instead of plain type assertions.
+func nestedString(root interface{}, path ...string) string {
+	cur := root
+	for _, key := range path {
+		next, ok := index(cur, key)
+		if !ok {
+			return ""
+		}
+		cur = next
+	}
+	s, _ := cur.(string)
+	return s
+}
+
+func index(container interface{}, key string) (interface{}, bool) {
+	if container == nil {
+		return nil, false
+	}
+	if m, ok := container.(map[string]interface{}); ok {
+		v, ok := m[key]
+		return v, ok
+	}
+	v := reflect.ValueOf(container)
+	switch v.Kind() {
+	case reflect.Map:
+		if v.Type().Key().Kind() != reflect.String {
+			return nil, false
+		}
+		mv := v.MapIndex(reflect.ValueOf(key).Convert(v.Type().Key()))
+		if !mv.IsValid() {
+			return nil, false
+		}
+		return mv.Interface(), true
+	case reflect.Slice, reflect.Array:
+		i, err := strconv.Atoi(key)
+		if err != nil || i < 0 || i >= v.Len() {
+			return nil, false
+		}
+		return v.Index(i).Interface(), true
+	default:
+		return nil, false
+	}
+}
+
+// series is a fixed size ring of time buckets.
+type series struct {
+	dur     time.Duration
+	buckets []Bucket
+	head    int // index of the newest bucket
+}
+
+func newSeries(dur time.Duration, size int) *series {
+	s := &series{dur: dur, buckets: make([]Bucket, size)}
+	now := time.Now()
+	for i := range s.buckets {
+		s.buckets[i].Time = s.align(now.Add(-time.Duration(len(s.buckets)-1-i) * dur))
+	}
+	s.head = len(s.buckets) - 1
+	return s
+}
+
+func (s *series) align(t time.Time) int64 {
+	return t.Truncate(s.dur).UnixMilli()
+}
+
+// rotate advances the ring so that the newest bucket matches now.
+func (s *series) rotate(now time.Time) *Bucket {
+	ts := s.align(now)
+	head := &s.buckets[s.head]
+	if head.Time == ts {
+		return head
+	}
+	if ts < head.Time {
+		// Clock went backwards, keep using the current bucket.
+		return head
+	}
+	steps := int((ts - head.Time) / s.dur.Milliseconds())
+	if steps > len(s.buckets) {
+		steps = len(s.buckets)
+	}
+	for i := 0; i < steps; i++ {
+		s.head = (s.head + 1) % len(s.buckets)
+		s.buckets[s.head] = Bucket{Time: head.Time + int64(i+1)*s.dur.Milliseconds()}
+	}
+	s.buckets[s.head].Time = ts
+	return &s.buckets[s.head]
+}
+
+func (s *series) add(now time.Time, proto, action string) {
+	b := s.rotate(now)
+	switch proto {
+	case "tcp":
+		b.TCP++
+	case "udp":
+		b.UDP++
+	}
+	switch action {
+	case "block", "drop":
+		b.Blocked++
+	case "allow":
+		b.Allowed++
+	}
+}
+
+// snapshot returns the buckets in chronological order.
+func (s *series) snapshot(now time.Time) []Bucket {
+	s.rotate(now)
+	out := make([]Bucket, 0, len(s.buckets))
+	for i := 1; i <= len(s.buckets); i++ {
+		out = append(out, s.buckets[(s.head+i)%len(s.buckets)])
+	}
+	return out
+}
+
+// counterMap is a bounded string counter used for the "top N" lists.
+type counterMap struct {
+	m map[string]uint64
+}
+
+func newCounterMap() *counterMap {
+	return &counterMap{m: make(map[string]uint64)}
+}
+
+func (c *counterMap) inc(key string) {
+	if len(c.m) >= maxCounterEntries {
+		if _, ok := c.m[key]; !ok {
+			c.prune()
+		}
+	}
+	c.m[key]++
+}
+
+// prune keeps the most frequent half of the entries.
+func (c *counterMap) prune() {
+	keep := c.top(maxCounterEntries / 2)
+	m := make(map[string]uint64, len(keep))
+	for _, kv := range keep {
+		m[kv.Name] = kv.Count
+	}
+	c.m = m
+}
+
+func (c *counterMap) top(n int) []NameCount {
+	out := make([]NameCount, 0, len(c.m))
+	for k, v := range c.m {
+		out = append(out, NameCount{Name: k, Count: v})
+	}
+	sort.Slice(out, func(i, j int) bool {
+		if out[i].Count != out[j].Count {
+			return out[i].Count > out[j].Count
+		}
+		return out[i].Name < out[j].Name
+	})
+	if len(out) > n {
+		out = out[:n]
+	}
+	return out
+}