From 90fd37eebbb15abbb527732f2d72023b7bdd335a Mon Sep 17 00:00:00 2001 From: Daniel Lavrushin Date: Sat, 6 Jun 2026 02:45:23 +0200 Subject: [PATCH] fix: resolve device duplication and missing names on Traffic page (#240) --- changelog.md | 2 + src/config/bind.go | 173 +++++++++++++++--- src/config/methods.go | 2 +- src/config/methods_test.go | 2 +- src/config/migration.go | 32 ++-- src/config/migration_test.go | 21 ++- src/config/types.go | 2 + .../ui/src/components/connections/Page.tsx | 24 ++- .../connections/views/AggregatedView.tsx | 4 +- src/http/ui/src/hooks/useConnectionGroups.ts | 21 +++ .../workers/connections-aggregator.worker.ts | 12 +- src/http/ui/src/workers/connections-types.ts | 1 + src/main.go | 10 +- src/mtproto/logtag.go | 21 +++ src/mtproto/obfuscated2.go | 67 +++---- src/mtproto/server.go | 31 ++-- src/mtproto/transparent.go | 62 ++++++- src/tproxy/listener.go | 4 + 18 files changed, 362 insertions(+), 129 deletions(-) create mode 100644 src/mtproto/logtag.go diff --git a/changelog.md b/changelog.md index 4ac9babd..b24a0d5c 100644 --- a/changelog.md +++ b/changelog.md @@ -4,6 +4,8 @@ - ADDED: **Block (blackhole) routing mode** - a new "Block" option in a set's Routing tab that blocks all matched traffic (ad/tracker domains, IPs, or a GeoSite category like `category-ads-all`) across the whole network - every LAN device and the router itself - with no output interface needed. It blocks by name and not just by IP, so it keeps working even with encrypted DNS and won't break unrelated sites sharing the same servers. See [Blocking](https://daniellavrushin.github.io/b4/docs/sets/blocking) for setup and details. - ADDED: **Upstream SOCKS5 proxy now routes UDP too** - sets that send traffic to an upstream SOCKS5 proxy used to forward only regular (TCP) connections; UDP traffic - such as QUIC (used by YouTube and many Google and video services) and DNS - bypassed the proxy and went out directly. UDP from LAN devices now goes through the upstream proxy as well, so a set can route a device's full traffic through it. The proxy must accept UDP. +- FIXED: **A device could appear twice on the Traffic page, and hand-added names were missing** - in the Aggregated view's Devices sidebar, one device sometimes showed up as two: once under its name and once under its bare IP address, because some of its connections arrived without a recognizable hardware address. Those connections are now matched back to the right device by its IP, so each device appears once. Devices you add by hand (Settings -> Devices), including ones behind another router, now show their assigned name here too instead of a raw IP. +- FIXED: **Command-line options like `--skip-tables` got saved into the config and stuck** - starting b4 with an extra command-line option used to write that option into the configuration file, so it kept applying on later starts even when you ran b4 without it. Command-line options now affect only the run you pass them to and are never saved into the config. ## [1.64.0] - 2026-06-01 diff --git a/src/config/bind.go b/src/config/bind.go index af995275..3a225631 100644 --- a/src/config/bind.go +++ b/src/config/bind.go @@ -2,28 +2,157 @@ package config import "github.com/spf13/cobra" -func (c *Config) BindFlags(cmd *cobra.Command) { - // Config path +type CLIOverrides struct { + QueueNum int + Threads int + Mark uint + IPv4Enabled bool + IPv6Enabled bool + MonitorInterval int + SkipSetup bool + Masquerade bool + MasqueradeInterface string + Instaflush bool + Syslog bool + ErrorFile string + WebPort int +} + +func (c *Config) BindFlags(cmd *cobra.Command, o *CLIOverrides) { cmd.Flags().StringVar(&c.ConfigPath, "config", c.ConfigPath, "Path to config file") - // Queue configuration - cmd.Flags().IntVar(&c.Queue.StartNum, "queue-num", c.Queue.StartNum, "Netfilter queue number") - cmd.Flags().IntVar(&c.Queue.Threads, "threads", c.Queue.Threads, "Number of worker threads") - cmd.Flags().UintVar(&c.Queue.Mark, "mark", c.Queue.Mark, "Packet mark value (default 32768)") - cmd.Flags().BoolVar(&c.Queue.IPv4Enabled, "ipv4", c.Queue.IPv4Enabled, "Enable IPv4 processing") - cmd.Flags().BoolVar(&c.Queue.IPv6Enabled, "ipv6", c.Queue.IPv6Enabled, "Enable IPv6 processing") - - // System configuration - cmd.Flags().IntVar(&c.System.Tables.MonitorInterval, "tables-monitor-interval", c.System.Tables.MonitorInterval, "Tables monitor interval in seconds (default 10, 0 to disable)") - cmd.Flags().BoolVar(&c.System.Tables.SkipSetup, "skip-tables", c.System.Tables.SkipSetup, "Skip iptables/nftables setup on startup") - cmd.Flags().BoolVar(&c.System.Tables.Masquerade, "masquerade", c.System.Tables.Masquerade, "Enable NAT masquerade (useful for containers)") - cmd.Flags().StringVar(&c.System.Tables.MasqueradeInterface, "masquerade-interface", c.System.Tables.MasqueradeInterface, "Restrict masquerade to this output interface (empty = all)") - - // Logging configuration - cmd.Flags().BoolVarP(&c.System.Logging.Instaflush, "instaflush", "i", c.System.Logging.Instaflush, "Flush logs immediately") - cmd.Flags().BoolVar(&c.System.Logging.Syslog, "syslog", c.System.Logging.Syslog, "Enable syslog output") - cmd.Flags().StringVar(&c.System.Logging.ErrorFile, "error-file", c.System.Logging.ErrorFile, "Path to error log file (empty disables)") - - // Web Server configuration - cmd.Flags().IntVar(&c.System.WebServer.Port, "web-port", c.System.WebServer.Port, "Port for internal web server (0 disables)") + d := DefaultConfig + cmd.Flags().IntVar(&o.QueueNum, "queue-num", d.Queue.StartNum, "Netfilter queue number") + cmd.Flags().IntVar(&o.Threads, "threads", d.Queue.Threads, "Number of worker threads") + cmd.Flags().UintVar(&o.Mark, "mark", d.Queue.Mark, "Packet mark value (default 32768)") + cmd.Flags().BoolVar(&o.IPv4Enabled, "ipv4", d.Queue.IPv4Enabled, "Enable IPv4 processing") + cmd.Flags().BoolVar(&o.IPv6Enabled, "ipv6", d.Queue.IPv6Enabled, "Enable IPv6 processing") + cmd.Flags().IntVar(&o.MonitorInterval, "tables-monitor-interval", d.System.Tables.MonitorInterval, "Tables monitor interval in seconds (default 10, 0 to disable)") + cmd.Flags().BoolVar(&o.SkipSetup, "skip-tables", d.System.Tables.SkipSetup, "Skip iptables/nftables setup on startup") + cmd.Flags().BoolVar(&o.Masquerade, "masquerade", d.System.Tables.Masquerade, "Enable NAT masquerade (useful for containers)") + cmd.Flags().StringVar(&o.MasqueradeInterface, "masquerade-interface", d.System.Tables.MasqueradeInterface, "Restrict masquerade to this output interface (empty = all)") + cmd.Flags().BoolVarP(&o.Instaflush, "instaflush", "i", d.System.Logging.Instaflush, "Flush logs immediately") + cmd.Flags().BoolVar(&o.Syslog, "syslog", d.System.Logging.Syslog, "Enable syslog output") + cmd.Flags().StringVar(&o.ErrorFile, "error-file", d.System.Logging.ErrorFile, "Path to error log file (empty disables)") + cmd.Flags().IntVar(&o.WebPort, "web-port", d.System.WebServer.Port, "Port for internal web server (0 disables)") +} + +var persistedOverrides struct { + snapshot *Config + fields map[string]bool + overrides CLIOverrides +} + +func (c *Config) ApplyCLIOverrides(cmd *cobra.Command, o *CLIOverrides) { + fields := make(map[string]bool) + changed := func(name string) bool { + if cmd.Flags().Changed(name) { + fields[name] = true + return true + } + return false + } + + snapshot := &Config{Queue: c.Queue, System: c.System} + + if changed("queue-num") { + c.Queue.StartNum = o.QueueNum + } + if changed("threads") { + c.Queue.Threads = o.Threads + } + if changed("mark") { + c.Queue.Mark = o.Mark + } + if changed("ipv4") { + c.Queue.IPv4Enabled = o.IPv4Enabled + } + if changed("ipv6") { + c.Queue.IPv6Enabled = o.IPv6Enabled + } + if changed("tables-monitor-interval") { + c.System.Tables.MonitorInterval = o.MonitorInterval + } + if changed("skip-tables") { + c.System.Tables.SkipSetup = o.SkipSetup + } + if changed("masquerade") { + c.System.Tables.Masquerade = o.Masquerade + } + if changed("masquerade-interface") { + c.System.Tables.MasqueradeInterface = o.MasqueradeInterface + } + if changed("instaflush") { + c.System.Logging.Instaflush = o.Instaflush + } + if changed("syslog") { + c.System.Logging.Syslog = o.Syslog + } + if changed("error-file") { + c.System.Logging.ErrorFile = o.ErrorFile + } + if changed("web-port") { + c.System.WebServer.Port = o.WebPort + } + + if len(fields) > 0 { + persistedOverrides.snapshot = snapshot + persistedOverrides.fields = fields + persistedOverrides.overrides = *o + } else { + persistedOverrides.snapshot = nil + persistedOverrides.fields = nil + persistedOverrides.overrides = CLIOverrides{} + } +} + +func stripCLIOverrides(c *Config) *Config { + snap := persistedOverrides.snapshot + f := persistedOverrides.fields + if snap == nil || len(f) == 0 { + return c + } + ov := persistedOverrides.overrides + + clone := *c + if f["queue-num"] && c.Queue.StartNum == ov.QueueNum { + clone.Queue.StartNum = snap.Queue.StartNum + } + if f["threads"] && c.Queue.Threads == ov.Threads { + clone.Queue.Threads = snap.Queue.Threads + } + if f["mark"] && c.Queue.Mark == ov.Mark { + clone.Queue.Mark = snap.Queue.Mark + } + if f["ipv4"] && c.Queue.IPv4Enabled == ov.IPv4Enabled { + clone.Queue.IPv4Enabled = snap.Queue.IPv4Enabled + } + if f["ipv6"] && c.Queue.IPv6Enabled == ov.IPv6Enabled { + clone.Queue.IPv6Enabled = snap.Queue.IPv6Enabled + } + if f["tables-monitor-interval"] && c.System.Tables.MonitorInterval == ov.MonitorInterval { + clone.System.Tables.MonitorInterval = snap.System.Tables.MonitorInterval + } + if f["skip-tables"] && c.System.Tables.SkipSetup == ov.SkipSetup { + clone.System.Tables.SkipSetup = snap.System.Tables.SkipSetup + } + if f["masquerade"] && c.System.Tables.Masquerade == ov.Masquerade { + clone.System.Tables.Masquerade = snap.System.Tables.Masquerade + } + if f["masquerade-interface"] && c.System.Tables.MasqueradeInterface == ov.MasqueradeInterface { + clone.System.Tables.MasqueradeInterface = snap.System.Tables.MasqueradeInterface + } + if f["instaflush"] && c.System.Logging.Instaflush == ov.Instaflush { + clone.System.Logging.Instaflush = snap.System.Logging.Instaflush + } + if f["syslog"] && c.System.Logging.Syslog == ov.Syslog { + clone.System.Logging.Syslog = snap.System.Logging.Syslog + } + if f["error-file"] && c.System.Logging.ErrorFile == ov.ErrorFile { + clone.System.Logging.ErrorFile = snap.System.Logging.ErrorFile + } + if f["web-port"] && c.System.WebServer.Port == ov.WebPort { + clone.System.WebServer.Port = snap.System.WebServer.Port + } + return &clone } diff --git a/src/config/methods.go b/src/config/methods.go index e43ef7aa..4a0dd07e 100644 --- a/src/config/methods.go +++ b/src/config/methods.go @@ -22,7 +22,7 @@ func (c *Config) SaveToFile(path string) error { c.Version = CurrentConfigVersion - data, err := MarshalSparse(c) + data, err := MarshalSparse(stripCLIOverrides(c)) if err != nil { return log.Errorf("failed to marshal config: %v", err) } diff --git a/src/config/methods_test.go b/src/config/methods_test.go index 05125ac3..3769c56a 100644 --- a/src/config/methods_test.go +++ b/src/config/methods_test.go @@ -1087,7 +1087,7 @@ func TestEscalateTo_RoundtripAndDefault(t *testing.T) { } loaded := NewConfig() - if err := loaded.LoadWithMigration(path); err != nil { + if _, err := loaded.LoadWithMigration(path); err != nil { t.Fatalf("LoadWithMigration: %v", err) } if got := loaded.GetSetById("a"); got == nil || got.Escalate.To != "b" { diff --git a/src/config/migration.go b/src/config/migration.go index 8969fb38..34ba74aa 100644 --- a/src/config/migration.go +++ b/src/config/migration.go @@ -641,40 +641,38 @@ func discoverConfigPath() string { return "/etc/b4/b4.json" } -func (c *Config) LoadWithMigration(path string) error { - discovered := false +func (c *Config) LoadWithMigration(path string) (bool, error) { if path == "" { path = discoverConfigPath() - c.ConfigPath = path - discovered = true log.Infof("Using config path: %s", path) } + c.ConfigPath = path info, err := os.Stat(path) if err != nil { - if discovered && os.IsNotExist(err) { + if os.IsNotExist(err) { log.Infof("Config file does not exist yet, using defaults: %s", path) - return nil + return true, nil } - return log.Errorf("failed to stat config file: %v", err) + return false, log.Errorf("failed to stat config file: %v", err) } if info.IsDir() { - return log.Errorf("config path is a directory, not a file: %s", path) + return false, log.Errorf("config path is a directory, not a file: %s", path) } data, err := os.ReadFile(path) if err != nil { - return log.Errorf("failed to read config file: %v", err) + return false, log.Errorf("failed to read config file: %v", err) } var rawJSON map[string]interface{} if err := json.Unmarshal(data, &rawJSON); err != nil { - return log.Errorf("failed to parse config file: %v", err) + return false, log.Errorf("failed to parse config file: %v", err) } var raw map[string]json.RawMessage if err := json.Unmarshal(data, &raw); err != nil { - return log.Errorf("failed to parse config file: %v", err) + return false, log.Errorf("failed to parse config file: %v", err) } rawSets := raw["sets"] @@ -682,35 +680,37 @@ func (c *Config) LoadWithMigration(path string) error { withoutSets, _ := json.Marshal(raw) if err := json.Unmarshal(withoutSets, c); err != nil { - return log.Errorf("failed to parse config file: %v", err) + return false, log.Errorf("failed to parse config file: %v", err) } if rawSets != nil { var setArray []json.RawMessage if err := json.Unmarshal(rawSets, &setArray); err != nil { - return log.Errorf("failed to parse sets: %v", err) + return false, log.Errorf("failed to parse sets: %v", err) } c.Sets = make([]*SetConfig, 0, len(setArray)) for _, rs := range setArray { set := NewSetConfig() if err := json.Unmarshal(rs, &set); err != nil { - return log.Errorf("failed to parse set: %v", err) + return false, log.Errorf("failed to parse set: %v", err) } c.Sets = append(c.Sets, &set) } } + migrated := false if c.Version < CurrentConfigVersion { + migrated = true log.Infof("Config version %d is older than current version %d, migrating", c.Version, CurrentConfigVersion) if err := c.applyMigrations(c.Version, rawJSON); err != nil { - return err + return false, err } } c.System.Geo.SanitizePaths(filepath.Dir(c.ConfigPath)) - return nil + return migrated, nil } func (c *Config) applyMigrations(startVersion int, rawJSON map[string]interface{}) error { diff --git a/src/config/migration_test.go b/src/config/migration_test.go index 1060c6af..a3ec6daf 100644 --- a/src/config/migration_test.go +++ b/src/config/migration_test.go @@ -46,22 +46,25 @@ func TestLoadWithMigration(t *testing.T) { t.Run("empty path with no default config returns nil", func(t *testing.T) { cfg := NewConfig() - if err := cfg.LoadWithMigration(""); err != nil { + if _, err := cfg.LoadWithMigration(""); err != nil { t.Errorf("expected nil for empty path with no discoverable config: %v", err) } }) - t.Run("nonexistent file errors", func(t *testing.T) { + t.Run("nonexistent file signals creation", func(t *testing.T) { cfg := NewConfig() - err := cfg.LoadWithMigration(filepath.Join(tmpDir, "nope.json")) - if err == nil { - t.Error("expected error for nonexistent file") + needsSave, err := cfg.LoadWithMigration(filepath.Join(tmpDir, "nope.json")) + if err != nil { + t.Errorf("expected no error for nonexistent file, got %v", err) + } + if !needsSave { + t.Error("expected needsSave=true for nonexistent file") } }) t.Run("directory path errors", func(t *testing.T) { cfg := NewConfig() - err := cfg.LoadWithMigration(tmpDir) + _, err := cfg.LoadWithMigration(tmpDir) if err == nil { t.Error("expected error for directory path") } @@ -78,7 +81,7 @@ func TestLoadWithMigration(t *testing.T) { os.WriteFile(path, []byte(v0Json), 0644) cfg := NewConfig() - if err := cfg.LoadWithMigration(path); err != nil { + if _, err := cfg.LoadWithMigration(path); err != nil { t.Fatalf("LoadWithMigration failed: %v", err) } @@ -97,7 +100,7 @@ func TestLoadWithMigration(t *testing.T) { cfg.SaveToFile(path) loaded := NewConfig() - if err := loaded.LoadWithMigration(path); err != nil { + if _, err := loaded.LoadWithMigration(path); err != nil { t.Fatalf("LoadWithMigration failed: %v", err) } if loaded.Version != CurrentConfigVersion { @@ -115,7 +118,7 @@ func TestLoadWithMigration(t *testing.T) { cfg.SaveToFile(path) loaded := NewConfig() - if err := loaded.LoadWithMigration(path); err != nil { + if _, err := loaded.LoadWithMigration(path); err != nil { t.Fatalf("LoadWithMigration failed: %v", err) } diff --git a/src/config/types.go b/src/config/types.go index cb0c78b7..0b867800 100644 --- a/src/config/types.go +++ b/src/config/types.go @@ -253,6 +253,8 @@ type MTProtoConfig struct { CFProxyEnabled bool `json:"cfproxy_enabled"` // enable Cloudflare-proxied fallback WS domains (rescues DCs the network blocks) CFProxyURL string `json:"cfproxy_url"` // URL to refresh CF-proxy domain list; empty = built-in default CFWorkerDomain string `json:"cfworker_domain"` // user's Cloudflare Worker domain(s) (workers.dev), comma-separated; free per-user WS relay tried before the shared CF pool + + BridgeSkipNativeEdge bool `json:"-"` } type Socks5Config struct { diff --git a/src/http/ui/src/components/connections/Page.tsx b/src/http/ui/src/components/connections/Page.tsx index 4250ee11..a9a85dfe 100644 --- a/src/http/ui/src/components/connections/Page.tsx +++ b/src/http/ui/src/components/connections/Page.tsx @@ -105,6 +105,10 @@ export function ConnectionsPage() { const [ipInfoToken, setIpInfoToken] = useState(""); const [devicesEnabled, setDevicesEnabled] = useState(false); const [deviceMap, setDeviceMap] = useState>({}); + const [ipToMac, setIpToMac] = useState>({}); + const [configIpToMac, setConfigIpToMac] = useState>( + {}, + ); const [configDeviceNames, setConfigDeviceNames] = useState< Record >({}); @@ -122,23 +126,30 @@ export function ConnectionsPage() { useEffect(() => { if (!devicesEnabled) { setDeviceMap({ ...configDeviceNames }); + setIpToMac({ ...configIpToMac }); return; } devicesApi .list() .then((data) => { const map: Record = {}; + const ipMap: Record = {}; for (const d of data.devices || []) { const normalized = d.mac.toUpperCase().replaceAll("-", ":"); map[normalized] = d.alias || d.vendor || ""; + if (d.ip) ipMap[d.ip] = normalized; } for (const [mac, name] of Object.entries(configDeviceNames)) { map[mac] = name; } + for (const [ip, mac] of Object.entries(configIpToMac)) { + ipMap[ip] = mac; + } setDeviceMap(map); + setIpToMac(ipMap); }) .catch(() => {}); - }, [devicesEnabled, configDeviceNames]); + }, [devicesEnabled, configDeviceNames, configIpToMac]); const fetchSets = useCallback(async (signal?: AbortSignal) => { try { @@ -157,12 +168,18 @@ export function ConnectionsPage() { false, ); const names: Record = {}; + const configIps: Record = {}; for (const d of data.queue?.devices?.devices || []) { - if (d.mac && d.name) { - names[d.mac.toUpperCase().replaceAll("-", ":")] = d.name; + const normalized = d.mac?.toUpperCase().replaceAll("-", ":"); + if (normalized && d.name) { + names[normalized] = d.name; + } + if (normalized && d.ip) { + configIps[d.ip] = normalized; } } setConfigDeviceNames(names); + setConfigIpToMac(configIps); } } catch (error) { if ((error as Error).name !== "AbortError") { @@ -359,6 +376,7 @@ export function ConnectionsPage() { setPauseDomains(!pauseDomains)} showAll={showAll} diff --git a/src/http/ui/src/components/connections/views/AggregatedView.tsx b/src/http/ui/src/components/connections/views/AggregatedView.tsx index afa0625e..3b53841f 100644 --- a/src/http/ui/src/components/connections/views/AggregatedView.tsx +++ b/src/http/ui/src/components/connections/views/AggregatedView.tsx @@ -13,6 +13,7 @@ import { useTranslation } from "react-i18next"; interface Props { lines: string[]; deviceMap: Record; + ipToMac: Record; paused: boolean; onTogglePause: () => void; showAll: boolean; @@ -67,6 +68,7 @@ const getGroupSearchableValues = (g: EnrichedGroup): (string | null)[] => [ export const AggregatedView = ({ lines, deviceMap, + ipToMac, paused, onTogglePause, showAll, @@ -94,7 +96,7 @@ export const AggregatedView = ({ localStorage.setItem("b4_connections_sidebar_collapsed", sidebarCollapsed ? "1" : "0"); }, [sidebarCollapsed]); - const state = useConnectionGroups(lines, deviceMap, paused); + const state = useConnectionGroups(lines, deviceMap, paused, ipToMac); useEffect(() => { const id = globalThis.setInterval(() => setNowTick(Date.now()), 1000); diff --git a/src/http/ui/src/hooks/useConnectionGroups.ts b/src/http/ui/src/hooks/useConnectionGroups.ts index e7022c9e..61b2ce08 100644 --- a/src/http/ui/src/hooks/useConnectionGroups.ts +++ b/src/http/ui/src/hooks/useConnectionGroups.ts @@ -38,9 +38,14 @@ export function useConnectionGroups( lines: string[], deviceMap: Record, paused: boolean, + ipToMac: Record = {}, ): GroupsState { const workerRef = useRef(null); const lastSentinelRef = useRef(null); + const linesRef = useRef(lines); + const pausedRef = useRef(paused); + linesRef.current = lines; + pausedRef.current = paused; const [snapshot, setSnapshot] = useState(null); useEffect(() => { @@ -58,6 +63,22 @@ export function useConnectionGroups( }; }, []); + useEffect(() => { + const w = workerRef.current; + if (!w) return; + const setMsg: AggregatorInput = { type: "setIpToMac", map: ipToMac }; + w.postMessage(setMsg); + if (pausedRef.current) return; + const clearMsg: AggregatorInput = { type: "clear" }; + w.postMessage(clearMsg); + const all = linesRef.current; + if (all.length > 0) { + const ingestMsg: AggregatorInput = { type: "ingest", lines: all }; + w.postMessage(ingestMsg); + } + lastSentinelRef.current = all.at(-1) ?? null; + }, [ipToMac]); + useEffect(() => { const w = workerRef.current; if (!w) return; diff --git a/src/http/ui/src/workers/connections-aggregator.worker.ts b/src/http/ui/src/workers/connections-aggregator.worker.ts index 6154bc41..b8270f81 100644 --- a/src/http/ui/src/workers/connections-aggregator.worker.ts +++ b/src/http/ui/src/workers/connections-aggregator.worker.ts @@ -18,6 +18,7 @@ let snapshotInterval = 500; const groups = new Map(); const devices = new Map(); +let ipToMac: Record = {}; let totalPackets = 0; let unmatchedCount = 0; let dirty = false; @@ -118,9 +119,11 @@ function ingest(lines: string[]): void { const p = parseLine(line); if (!p) continue; - const mac = p.sourceAlias - ? normalizeMac(p.sourceAlias) - : stripPort(p.source); + let mac = p.sourceAlias ? normalizeMac(p.sourceAlias) : ""; + if (!mac) { + const ip = stripPort(p.source); + mac = ipToMac[ip] || ip; + } const groupIdent = p.domain || p.destination || "?"; const key = `${mac}|${p.protocol}|${groupIdent}`; @@ -250,6 +253,9 @@ ctx.onmessage = (e: MessageEvent) => { bucketCount = Math.max(10, Math.min(300, msg.count)); clearState(); break; + case "setIpToMac": + ipToMac = msg.map; + break; case "flush": dirty = true; postSnapshot(); diff --git a/src/http/ui/src/workers/connections-types.ts b/src/http/ui/src/workers/connections-types.ts index 26a5746c..fb732cde 100644 --- a/src/http/ui/src/workers/connections-types.ts +++ b/src/http/ui/src/workers/connections-types.ts @@ -36,6 +36,7 @@ export type AggregatorInput = | { type: "clear" } | { type: "setSnapshotInterval"; ms: number } | { type: "setBucketCount"; count: number } + | { type: "setIpToMac"; map: Record } | { type: "flush" }; export type AggregatorOutput = { type: "snapshot"; payload: AggregatorSnapshot }; diff --git a/src/main.go b/src/main.go index 41b08fd4..601bd32c 100644 --- a/src/main.go +++ b/src/main.go @@ -37,6 +37,7 @@ import ( var ( cfg = config.NewConfig() + cliOverrides config.CLIOverrides verboseFlag string showVersion bool clearTables bool @@ -57,7 +58,7 @@ var rootCmd = &cobra.Command{ func init() { // Bind all configuration flags - cfg.BindFlags(rootCmd) + cfg.BindFlags(rootCmd, &cliOverrides) // Add verbosity flags separately since they need special handling rootCmd.Flags().StringVar(&verboseFlag, "verbose", "info", "Set verbosity level (debug, trace, info, silent), default: info") @@ -97,8 +98,11 @@ func runB4(cmd *cobra.Command, args []string) error { initTimezone() - cfg.LoadWithMigration(cfg.ConfigPath) - cfg.SaveToFile(cfg.ConfigPath) + needsSave, _ := cfg.LoadWithMigration(cfg.ConfigPath) + if needsSave { + cfg.SaveToFile(cfg.ConfigPath) + } + cfg.ApplyCLIOverrides(cmd, &cliOverrides) if cfg.System.Timezone != "" { config.ApplyTimezone(cfg.System.Timezone) diff --git a/src/mtproto/logtag.go b/src/mtproto/logtag.go new file mode 100644 index 00000000..d64711dd --- /dev/null +++ b/src/mtproto/logtag.go @@ -0,0 +1,21 @@ +package mtproto + +import ( + "strconv" + "sync/atomic" +) + +const logTag = "tg-bridge" + +var connSeq atomic.Uint64 + +func nextConnID() string { + return strconv.FormatUint(connSeq.Add(1), 36) +} + +func tg(id string) string { + if id == "" { + return "[" + logTag + "]" + } + return "[" + logTag + " c=" + id + "]" +} diff --git a/src/mtproto/obfuscated2.go b/src/mtproto/obfuscated2.go index 5cf15713..233b25ae 100644 --- a/src/mtproto/obfuscated2.go +++ b/src/mtproto/obfuscated2.go @@ -36,7 +36,6 @@ func wsEdgeServesDC(absDC int) bool { return wsEdgeServedDCs[absDC] } -// workerDomains parses the comma-separated CF Worker domain list from config. func workerDomains(cfg *config.MTProtoConfig) []string { raw := strings.TrimSpace(cfg.CFWorkerDomain) if raw == "" { @@ -51,8 +50,6 @@ func workerDomains(cfg *config.MTProtoConfig) []string { return out } -// workerDstIP returns the canonical public DC IP the worker should TCP-connect to -// (the worker's ?dst= target). Mirrors tg-ws-proxy DC_DEFAULT_IPS. func workerDstIP(absDC int) string { addr, ok := dcAddressesV4[absDC] if !ok { @@ -169,9 +166,9 @@ type transportPlan struct { sni string dialHost string dc int - cfBase string // CF-proxy base domain (without "kwsN."); set => pin in balancer on success - wsPath string // WS request path; "" defaults to /apiws. CF Worker uses /apiws?dst=&dc= - isWorker bool // CF Worker plan: dedicated per-user relay, reaches any DC via ?dst= + cfBase string + wsPath string + isWorker bool } func (p transportPlan) describe() string { @@ -227,7 +224,7 @@ func planTransports(cfg *config.MTProtoConfig, queueCfg config.QueueConfig, dc i if edgeIP == "" { edgeIP = telegramWSEdgeIP } - if wsEdgeServesDC(absDC) { + if wsEdgeServesDC(absDC) && !cfg.BridgeSkipNativeEdge { primary := transportPlan{kind: transportWS, dc: dc, sni: fmt.Sprintf("kws%d.web.telegram.org", absDC), dialHost: edgeIP} media := transportPlan{kind: transportWS, dc: dc, sni: fmt.Sprintf("kws%d-1.web.telegram.org", absDC), dialHost: edgeIP} if dc < 0 { @@ -236,11 +233,6 @@ func planTransports(cfg *config.MTProtoConfig, queueCfg config.QueueConfig, dc i plans = append(plans, primary, media) } } - // CF Worker (free per-user workers.dev relay). Tried after TG's own edge so - // the fast native path wins for DC2/4, but before the shared CF pool so DCs - // without a native edge (1/3/5) and throttled cases use the dedicated worker - // instead of the rate-limited shared domains. The worker reaches any DC via - // ?dst=&dc= (matches tg-ws-proxy CfWorker mode). if dst := workerDstIP(absDC); dst != "" { for _, wd := range workerDomains(cfg) { plans = append(plans, transportPlan{ @@ -261,8 +253,6 @@ func planTransports(cfg *config.MTProtoConfig, queueCfg config.QueueConfig, dc i cfBase: d, }) } - // CF-proxy fallback pool (matches tg-ws-proxy). Tried after TG's own edge so - // the fast path wins when it works; CF rescues DCs the network blocks (esp. DC 1). if cfg.CFProxyEnabled { for _, base := range cfBalancerInst.domainsForDC(dc) { plans = append(plans, transportPlan{ @@ -290,22 +280,23 @@ func planTransports(cfg *config.MTProtoConfig, queueCfg config.QueueConfig, dc i } func DialObfuscatedDC(cfg *config.MTProtoConfig, queueCfg config.QueueConfig, dc int, protoTag uint32) (*ObfuscatedConn, string, error) { - return DialObfuscatedDCWithPool(cfg, queueCfg, dc, protoTag, nil) + return DialObfuscatedDCWithPool(cfg, queueCfg, dc, protoTag, nil, "") } -func DialObfuscatedDCWithPool(cfg *config.MTProtoConfig, queueCfg config.QueueConfig, dc int, protoTag uint32, pool *wsPool) (*ObfuscatedConn, string, error) { +func DialObfuscatedDCWithPool(cfg *config.MTProtoConfig, queueCfg config.QueueConfig, dc int, protoTag uint32, pool *wsPool, logID string) (*ObfuscatedConn, string, error) { + tag := tg(logID) if pool != nil && !wsIsBlacklisted(dc) { if raw := pool.get(dc); raw != nil { obf, err := completeObfuscation(raw, dc, protoTag) if err == nil && raw.liveNow() { - log.Infof("MTProto DC %d connected via ws-pool", dc) + log.Infof("%s DC %d connected via ws-pool", tag, dc) wsRecordSuccess(dc) return obf, "ws-pool", nil } if err != nil { - log.Debugf("MTProto DC %d pool conn obf init failed: %v", dc, err) + log.Debugf("%s DC %d pool conn obf init failed: %v", tag, dc, err) } else { - log.Debugf("MTProto DC %d pool conn died before relay; re-dialing fresh", dc) + log.Debugf("%s DC %d pool conn died before relay; re-dialing fresh", tag, dc) } _ = raw.Close() } @@ -325,7 +316,7 @@ func DialObfuscatedDCWithPool(cfg *config.MTProtoConfig, queueCfg config.QueueCo wsTried := 0 wsRedirects := 0 for _, p := range plans { - log.Debugf("MTProto DC %d dialing %s", dc, p.describe()) + log.Debugf("%s DC %d dialing %s", tag, dc, p.describe()) start := time.Now() var conn net.Conn var derr error @@ -347,29 +338,27 @@ func DialObfuscatedDCWithPool(cfg *config.MTProtoConfig, queueCfg config.QueueCo } else if isDialTimeout(derr) { tcpRecordFailure(p.addr) } - log.Debugf("MTProto DC %d %s failed after %dms: %v", dc, p.describe(), time.Since(start).Milliseconds(), derr) + log.Debugf("%s DC %d %s failed after %dms: %v", tag, dc, p.describe(), time.Since(start).Milliseconds(), derr) continue } obfConn, oerr := completeObfuscation(conn, dc, protoTag) if oerr != nil { attempts = append(attempts, fmt.Sprintf("%s: %s", p.describe(), shortErr(oerr))) conn.Close() - log.Debugf("MTProto DC %d obf init failed on %s: %v", dc, p.describe(), oerr) + log.Debugf("%s DC %d obf init failed on %s: %v", tag, dc, p.describe(), oerr) continue } if p.kind == transportWS { wsRecordSuccess(dc) - // pin successful CF-proxy domain for this DC so subsequent connections - // try it first (mirrors tg-ws-proxy/proxy/balancer.py:update_domain_for_dc) if p.cfBase != "" { if cfBalancerInst.pin(dc, p.cfBase) { - log.Infof("MTProto DC %d switched active CF domain to %s", dc, p.cfBase) + log.Infof("%s DC %d switched active CF domain to %s", tag, dc, p.cfBase) } } } else { tcpRecordSuccess(p.addr) } - log.Infof("MTProto DC %d connected via %s in %dms", dc, p.describe(), time.Since(start).Milliseconds()) + log.Infof("%s DC %d connected via %s in %dms", tag, dc, p.describe(), time.Since(start).Milliseconds()) return obfConn, p.describe(), nil } @@ -543,10 +532,6 @@ func completeObfuscation(conn net.Conn, dc int, protoTag uint32) (*ObfuscatedCon encrypted := make([]byte, obfuscatedFrameLen) copy(encrypted, frame) encStream.XORKeyStream(encrypted, encrypted) - // Restore bytes 0:56 to plaintext - only bytes 56:64 (proto_tag||dc||rnd) are - // sent encrypted. tg-ws-proxy keeps the SKIP region (0:8) as plaintext too; - // b4 was only restoring 8:56, leaving 0:7 as ciphertext on the wire. Some TG - // endpoints inspect the SKIP bytes before decryption. copy(encrypted[0:56], frame[0:56]) if _, err := conn.Write(encrypted); err != nil { @@ -560,24 +545,16 @@ func completeObfuscation(conn net.Conn, dc int, protoTag uint32) (*ObfuscatedCon }, nil } -// reservedFirst4Words are the first-4-byte little-endian values an obfuscated -// frame must never start with: they collide with TLS/HTTP and the obfuscation -// transport tags, so TG treats a connection beginning with one of them as a -// different protocol. generateFrame avoids producing them; reservedFirst4 -// (transparent bridge) uses them to detect a non-obfuscated transport. Keep the -// two in sync via this single list. var reservedFirst4Words = []uint32{ - 0x44414548, // "HEAD" - 0x54534f50, // "POST" - 0x20544547, // "GET " - 0x4954504f, // "OPTI" - 0x02010316, // TLS record header - 0xdddddddd, // padded intermediate tag - 0xeeeeeeee, // intermediate tag + 0x44414548, + 0x54534f50, + 0x20544547, + 0x4954504f, + 0x02010316, + 0xdddddddd, + 0xeeeeeeee, } -// isReservedFirst4 reports whether the first 4 bytes are a reserved value -// (0xef abridged-tag first byte, or any reservedFirst4Words value). func isReservedFirst4(b []byte) bool { if b[0] == 0xef { return true diff --git a/src/mtproto/server.go b/src/mtproto/server.go index d1c81e6a..61e58f5e 100644 --- a/src/mtproto/server.go +++ b/src/mtproto/server.go @@ -233,11 +233,13 @@ func (s *Server) acceptLoop(ln net.Listener) { func (s *Server) handleConn(raw net.Conn) { clientAddr := raw.RemoteAddr().String() - log.Infof("MTProto new connection from %s", clientAddr) + id := nextConnID() + tag := tg(id) + log.Infof("%s proxy new connection from %s", tag, clientAddr) defer func() { if r := recover(); r != nil { - log.Errorf("MTProto panic from %s: %v", clientAddr, r) + log.Errorf("%s proxy panic from %s: %v", tag, clientAddr, r) } }() @@ -253,42 +255,41 @@ func (s *Server) handleConn(raw net.Conn) { tlsConn, err := AcceptFakeTLS(raw, secret) if err != nil { - log.Debugf("MTProto fake-TLS failed from %s: %v", clientAddr, err) + log.Debugf("%s proxy fake-TLS failed from %s: %v", tag, clientAddr, err) var vErr *FakeTLSVerifyError if errors.As(err, &vErr) && cfg.System.MTProto.FakeSNI != "" { proxyToMaskingDomain(raw, vErr.Initial, cfg.System.MTProto.FakeSNI, cfg.Queue.Mark) } return } - log.Debugf("MTProto fake-TLS handshake OK from %s", clientAddr) + log.Debugf("%s proxy fake-TLS handshake OK from %s", tag, clientAddr) result, err := AcceptObfuscated(tlsConn, secret) if err != nil { - log.Tracef("MTProto obfuscated2 failed from %s: %v", clientAddr, err) + log.Tracef("%s proxy obfuscated2 failed from %s: %v", tag, clientAddr, err) return } - log.Debugf("MTProto client from %s wants DC %d proto=0x%08x", clientAddr, result.DC, result.ProtoTag) + log.Debugf("%s proxy client from %s wants DC %d proto=0x%08x", tag, clientAddr, result.DC, result.ProtoTag) _ = raw.SetDeadline(time.Time{}) - dcConn, transport, err := DialObfuscatedDCWithPool(&cfg.System.MTProto, cfg.Queue, result.DC, result.ProtoTag, s.wsPool.Load()) + dcConn, transport, err := DialObfuscatedDCWithPool(&cfg.System.MTProto, cfg.Queue, result.DC, result.ProtoTag, s.wsPool.Load(), id) if err != nil { - // throttle ERROR-level spam from a permanently-broken DC; full detail still goes to Debug if shouldLogDialError(result.DC) { - log.Errorf("MTProto dial DC %d: %v", result.DC, err) + log.Errorf("%s proxy dial DC %d failed: %v", tag, result.DC, err) } else { - log.Debugf("MTProto dial DC %d (suppressed): %v", result.DC, err) + log.Debugf("%s proxy dial DC %d failed (suppressed): %v", tag, result.DC, err) } return } defer dcConn.Close() - log.Infof("MTProto relay: %s <-> DC%d (%s)", clientAddr, result.DC, transport) + log.Infof("%s proxy relay %s <-> DC%d via %s", tag, clientAddr, result.DC, transport) var splitter *msgSplitter if _, ok := dcConn.Conn.(*wsConn); ok { splitter = newMsgSplitter(result.ProtoTag) } - s.relay(result.Conn, dcConn, splitter, fmt.Sprintf("%s<->DC%d", clientAddr, result.DC)) + s.relay(result.Conn, dcConn, splitter, fmt.Sprintf("%s %s<->DC%d", tag, clientAddr, result.DC)) } func (s *Server) relay(client, dc io.ReadWriteCloser, splitter *msgSplitter, label string) { @@ -321,7 +322,7 @@ func relayConns(client, dc io.ReadWriteCloser, splitter *msgSplitter, label stri } } counter.Store(total) - log.Debugf("MTProto relay %s %s: %d bytes, err=%v", label, dir, total, err) + log.Debugf("%s %s: %d bytes, err=%v", label, dir, total, err) errCh <- err } @@ -351,7 +352,7 @@ func relayConns(client, dc io.ReadWriteCloser, splitter *msgSplitter, label stri } } counter.Store(total) - log.Debugf("MTProto relay %s %s: %d bytes, err=%v", label, dir, total, err) + log.Debugf("%s %s: %d bytes, err=%v", label, dir, total, err) errCh <- err } @@ -366,5 +367,5 @@ func relayConns(client, dc io.ReadWriteCloser, splitter *msgSplitter, label stri _ = client.Close() _ = dc.Close() <-errCh - log.Infof("MTProto session %s closed: up=%d down=%d in %dms", label, upBytes.Load(), downBytes.Load(), time.Since(start).Milliseconds()) + log.Infof("%s closed: up=%d down=%d in %dms", label, upBytes.Load(), downBytes.Load(), time.Since(start).Milliseconds()) } diff --git a/src/mtproto/transparent.go b/src/mtproto/transparent.go index 683a7425..6eaea129 100644 --- a/src/mtproto/transparent.go +++ b/src/mtproto/transparent.go @@ -91,64 +91,75 @@ func (b *TransparentBridge) getPool() *wsPool { } func (b *TransparentBridge) Handle(client net.Conn, origIP net.IP, origPort int) (bool, net.Conn) { + id := nextConnID() + tag := tg(id) + log.Tracef("%s bridge accept %s -> %s:%d", tag, client.RemoteAddr(), origIP, origPort) _ = client.SetReadDeadline(time.Now().Add(5 * time.Second)) init := make([]byte, obfuscatedFrameLen) head, herr := io.ReadFull(client, init[:4]) if herr != nil { _ = client.SetReadDeadline(time.Time{}) if head == 0 { + log.Tracef("%s bridge empty conn from %s -> drop", tag, origIP) return true, nil } + log.Debugf("%s bridge short head (%d B) from %s:%d -> fail open", tag, head, origIP, origPort) return false, &prefixConn{Conn: client, prefix: append([]byte(nil), init[:head]...)} } if reservedFirst4(init[:4]) { _ = client.SetReadDeadline(time.Time{}) - log.Debugf("MTProto transparent: non-obfuscated transport (% x) from %s -> fail open", init[:4], origIP) + log.Debugf("%s bridge non-obfuscated transport (% x) from %s:%d -> fail open", tag, init[:4], origIP, origPort) return false, &prefixConn{Conn: client, prefix: append([]byte(nil), init[:4]...)} } n, rerr := io.ReadFull(client, init[4:]) _ = client.SetReadDeadline(time.Time{}) if rerr != nil { + log.Debugf("%s bridge short handshake (%d/%d B) from %s:%d -> fail open", tag, 4+n, obfuscatedFrameLen, origIP, origPort) return false, &prefixConn{Conn: client, prefix: append([]byte(nil), init[:4+n]...)} } res, derr := decodeObfuscatedDirect(init, client) if derr != nil { - log.Debugf("MTProto transparent: obfuscated decode failed from %s: %v -> fail open", origIP, derr) + log.Debugf("%s bridge obfuscated decode failed from %s:%d: %v -> fail open", tag, origIP, origPort, derr) return false, &prefixConn{Conn: client, prefix: append([]byte(nil), init...)} } + log.Tracef("%s bridge handshake ok from %s:%d: proto=0x%08x handshake-dc=%d", tag, origIP, origPort, res.ProtoTag, res.DC) var dc int var dcSrc string if mapped, ok := dcForIP(origIP); ok { dc, dcSrc = mapped, "ip" - } else if mapped, ok := dcForIPRange(origIP); ok { - dc, dcSrc = mapped, "ip-range" } else if validTransparentDC(res.DC) { dc, dcSrc = res.DC, "handshake" + } else if mapped, ok := dcForIPRange(origIP); ok { + dc, dcSrc = mapped, "ip-range" } else { - log.Debugf("MTProto transparent: unresolved DC for %s (handshake dc=%d proto=0x%08x) -> fail open", origIP, res.DC, res.ProtoTag) + log.Debugf("%s bridge unresolved DC for %s:%d (handshake dc=%d proto=0x%08x) -> fail open", tag, origIP, origPort, res.DC, res.ProtoTag) return false, &prefixConn{Conn: client, prefix: append([]byte(nil), init...)} } + if rng, ok := dcForIPRange(origIP); ok && validTransparentDC(res.DC) && rng != res.DC { + log.Debugf("%s bridge DC ambiguity for %s: ip-range=DC%d handshake=DC%d -> using DC%d (src=%s)", tag, origIP, rng, res.DC, dc, dcSrc) + } cfg := b.cfg.Load() mtCfg := cfg.System.MTProto mtCfg.UpstreamMode = "auto" mtCfg.DCRelay = "" + mtCfg.BridgeSkipNativeEdge = true - dcConn, transport, err := DialObfuscatedDCWithPool(&mtCfg, cfg.Queue, dc, res.ProtoTag, b.getPool()) + dcConn, transport, err := DialObfuscatedDCWithPool(&mtCfg, cfg.Queue, dc, res.ProtoTag, nil, id) if err != nil { if shouldLogDialError(dc) { - log.Errorf("MTProto transparent dial DC %d: %v", dc, err) + log.Errorf("%s bridge dial DC %d failed: %v", tag, dc, err) } else { - log.Debugf("MTProto transparent dial DC %d (suppressed): %v", dc, err) + log.Debugf("%s bridge dial DC %d failed (suppressed): %v", tag, dc, err) } return true, nil } defer dcConn.Close() - label := fmt.Sprintf("%s<->DC%d(transparent)", client.RemoteAddr(), dc) - log.Infof("MTProto transparent relay: %s -> DC%d (%s) [dc-from=%s]", origIP, dc, transport, dcSrc) + label := fmt.Sprintf("%s %s<->DC%d(transparent)", tag, client.RemoteAddr(), dc) + log.Infof("%s bridge relay %s:%d -> DC%d via %s [dc-from=%s]", tag, origIP, origPort, dc, transport, dcSrc) var splitter *msgSplitter if _, isWS := dcConn.Conn.(*wsConn); isWS { @@ -158,6 +169,37 @@ func (b *TransparentBridge) Handle(client net.Conn, origIP net.IP, origPort int) return true, nil } +func (b *TransparentBridge) FailOpenViaWorker(client net.Conn, origIP net.IP, origPort int) bool { + cfg := b.cfg.Load() + mt := cfg.System.MTProto + domains := workerDomains(&mt) + if len(domains) == 0 { + return false + } + id := nextConnID() + tag := tg(id) + dst := origIP.String() + dc := 0 + if m, ok := dcForIP(origIP); ok { + dc = m + } else if m, ok := dcForIPRange(origIP); ok { + dc = m + } + for _, wd := range domains { + path := fmt.Sprintf("/apiws?dst=%s&dc=%d&port=%d", dst, dc, origPort) + wc, derr := dialWS(wd, wd, path, wsDialTimeout, cfg.Queue.Mark) + if derr != nil { + log.Debugf("%s failopen worker dial %s for %s:%d failed: %v", tag, wd, dst, origPort, derr) + continue + } + log.Infof("%s failopen relay %s:%d via wsworker://%s", tag, dst, origPort, wd) + label := fmt.Sprintf("%s %s<->%s:%d(failopen)", tag, client.RemoteAddr(), dst, origPort) + relayConns(client, wc, nil, label, &b.bufPool) + return true + } + return false +} + func validTransparentDC(dc int) bool { a := dc if a < 0 { diff --git a/src/tproxy/listener.go b/src/tproxy/listener.go index d416f0ee..06c7d351 100644 --- a/src/tproxy/listener.go +++ b/src/tproxy/listener.go @@ -28,6 +28,7 @@ type DomainResolver interface { type MTProtoBridge interface { Handle(client net.Conn, origIP net.IP, origPort int) (bool, net.Conn) + FailOpenViaWorker(client net.Conn, origIP net.IP, origPort int) bool } type Listener struct { @@ -201,6 +202,9 @@ func (l *Listener) handle(client net.Conn) { } else if failover != nil { client = failover } + if l.Bridge.FailOpenViaWorker(client, origIP, origPort) { + return + } } l.failOpenDirect(client, origIP, origPort) return