diff --git a/changelog.md b/changelog.md index 07229e40..18cc63c6 100644 --- a/changelog.md +++ b/changelog.md @@ -4,6 +4,7 @@ - IMPROVED: **Refreshed UI** — the whole web UI has been redesigned: cleaner typography, tighter spacing, calmer colour palette, larger and easier-to-read numbers, and better keyboard accessibility throughout. - ADDED: **Sequence overlap pattern in Combo fragmentation** — the Combo fragmentation strategy can again be tuned from the UI: choose a preset (TLS handshake, HTTP GET, zeros) or build a custom byte pattern for the overlap step. +- ADDED: **Upstream SOCKS5 routing** - per-set routing can now forward matched traffic to a SOCKS5 proxy (local or remote) instead of out a network interface. Use this to chain b4 with Xray, sing-box, or any SOCKS5-speaking proxy. Pick "Upstream SOCKS5 proxy" mode in the routing tab and set the host and port. ## [1.49.1] - 2026-04-20 diff --git a/src/config/config.go b/src/config/config.go index 7265e7c4..8fa3e8b8 100644 --- a/src/config/config.go +++ b/src/config/config.go @@ -93,7 +93,9 @@ var DefaultSetConfig = SetConfig{ Routing: RoutingConfig{ Enabled: false, + Mode: RoutingModeInterface, EgressInterface: "", + Upstream: UpstreamProxyConfig{UseDomain: true}, FWMark: 0, Table: 0, SourceInterfaces: []string{}, diff --git a/src/config/methods.go b/src/config/methods.go index 29cd6632..bf436f8a 100644 --- a/src/config/methods.go +++ b/src/config/methods.go @@ -171,11 +171,33 @@ func (c *Config) Validate() error { if set.Routing.IPTTLSeconds <= 0 { set.Routing.IPTTLSeconds = DefaultSetConfig.Routing.IPTTLSeconds } + switch set.Routing.Mode { + case "": + set.Routing.Mode = RoutingModeInterface + case RoutingModeProxy, RoutingModeInterface: + default: + return fmt.Errorf("set %q: unknown routing mode %q", set.Name, set.Routing.Mode) + } set.Routing.EgressInterface = sanitizeIfaceName(set.Routing.EgressInterface) for i, src := range set.Routing.SourceInterfaces { set.Routing.SourceInterfaces[i] = sanitizeIfaceName(src) } + if set.Routing.Enabled && set.Routing.Mode == RoutingModeProxy { + if set.Routing.Upstream.Port < 1 || set.Routing.Upstream.Port > 65535 { + return fmt.Errorf("set %q: upstream proxy port must be 1-65535", set.Name) + } + h := strings.ToLower(strings.TrimSpace(set.Routing.Upstream.Host)) + if h == "" { + h = "127.0.0.1" + } + if c.System.Socks5.Enabled && set.Routing.Upstream.Port == c.System.Socks5.Port { + if h == "127.0.0.1" || h == "::1" || h == "localhost" || h == "0.0.0.0" { + return fmt.Errorf("set %q: upstream proxy points to b4's own SOCKS5 server (loop)", set.Name) + } + } + } + if len(set.Fragmentation.SeqOverlapPattern) > 0 { set.Fragmentation.SeqOverlapBytes = make([]byte, len(set.Fragmentation.SeqOverlapPattern)) for i, s := range set.Fragmentation.SeqOverlapPattern { diff --git a/src/config/migration.go b/src/config/migration.go index e743043f..e71d581c 100644 --- a/src/config/migration.go +++ b/src/config/migration.go @@ -54,6 +54,18 @@ var migrationRegistry = map[int]MigrationFunc{ 31: migrateV31to32, // Add watchdog config 32: migrateV32to33, // Add TCP RST protection config 33: migrateV33to34, // Add manual devices to device config + 34: migrateV34to35, // Add routing mode and upstream proxy config +} + +func migrateV34to35(c *Config, _ map[string]interface{}) error { + log.Tracef("Migration v34->v35: Adding routing mode and upstream proxy config") + for _, set := range c.Sets { + if set.Routing.Mode == "" { + set.Routing.Mode = RoutingModeInterface + } + set.Routing.Upstream.UseDomain = true + } + return nil } func migrateV33to34(c *Config, raw map[string]interface{}) error { diff --git a/src/config/types.go b/src/config/types.go index 2264e410..74d3221f 100644 --- a/src/config/types.go +++ b/src/config/types.go @@ -11,6 +11,11 @@ const ( ConfigNone = "none" ) +const ( + RoutingModeInterface = "interface" + RoutingModeProxy = "proxy" +) + const ( FakePayloadRandom = iota FakePayloadCustom @@ -344,10 +349,21 @@ type MSSClampConfig struct { type RoutingConfig struct { - Enabled bool `json:"enabled"` - EgressInterface string `json:"egress_interface"` - FWMark uint32 `json:"fwmark"` - Table int `json:"table"` - SourceInterfaces []string `json:"source_interfaces"` - IPTTLSeconds int `json:"ip_ttl_seconds"` + Enabled bool `json:"enabled"` + Mode string `json:"mode"` + EgressInterface string `json:"egress_interface"` + Upstream UpstreamProxyConfig `json:"upstream"` + FWMark uint32 `json:"fwmark"` + Table int `json:"table"` + SourceInterfaces []string `json:"source_interfaces"` + IPTTLSeconds int `json:"ip_ttl_seconds"` +} + +type UpstreamProxyConfig struct { + Host string `json:"host"` + Port int `json:"port"` + Username string `json:"username,omitempty"` + Password string `json:"password,omitempty"` + FailOpen bool `json:"fail_open"` + UseDomain bool `json:"use_domain"` } diff --git a/src/http/ui/src/components/sets/routing/TrafficRouting.tsx b/src/http/ui/src/components/sets/routing/TrafficRouting.tsx index 4f105458..2fef3d10 100644 --- a/src/http/ui/src/components/sets/routing/TrafficRouting.tsx +++ b/src/http/ui/src/components/sets/routing/TrafficRouting.tsx @@ -1,6 +1,6 @@ import { Box, Grid, MenuItem, Typography } from "@mui/material"; import { B4Alert, B4Badge, B4Switch, B4TextField } from "@b4.elements"; -import { B4SetConfig } from "@models/config"; +import { B4SetConfig, RoutingMode } from "@models/config"; import { colors } from "@design"; import { useTranslation } from "react-i18next"; import ArrowForwardIcon from "@mui/icons-material/ArrowForward"; @@ -21,11 +21,14 @@ export const TrafficRouting = ({ }: TrafficRoutingProps) => { const { t } = useTranslation(); const routing = config.routing; + const mode: RoutingMode = routing.mode === "proxy" ? "proxy" : "interface"; + const isProxy = mode === "proxy"; + const selectedIfaceAvailable = availableIfaces.includes( routing.egress_interface, ); const shouldShowUnavailableSelected = Boolean( - routing.egress_interface && !selectedIfaceAvailable, + !isProxy && routing.egress_interface && !selectedIfaceAvailable, ); const toggleSourceIface = (iface: string) => { @@ -36,6 +39,26 @@ export const TrafficRouting = ({ onChange("routing.source_interfaces", updated); }; + const upstream = routing.upstream || { + host: "", + port: 0, + username: "", + password: "", + fail_open: false, + use_domain: true, + }; + + let flowDestination: string; + if (isProxy) { + flowDestination = + upstream.host && upstream.port + ? `${upstream.host}:${upstream.port}` + : t("sets.routing.flowNoUpstream"); + } else { + flowDestination = + routing.egress_interface || t("sets.routing.flowNoOutput"); + } + return ( @@ -44,13 +67,31 @@ export const TrafficRouting = ({ checked={routing.enabled} onChange={(checked: boolean) => onChange("routing.enabled", checked)} description={t("sets.routing.enableDesc")} - disabled={availableIfaces.length === 0 && !routing.enabled} + disabled={ + !isProxy && availableIfaces.length === 0 && !routing.enabled + } /> {routing.enabled && ( <> - {/* Traffic flow diagram */} + + onChange("routing.mode", e.target.value)} + helperText={t("sets.routing.modeHelper")} + > + + {t("sets.routing.modeInterface")} + + + {t("sets.routing.modeProxy")} + + + + - {routing.egress_interface || t("sets.routing.flowNoOutput")} + {flowDestination} - {t("sets.routing.flowCaption")} + {isProxy + ? t("sets.routing.flowProxyCaption") + : t("sets.routing.flowCaption")} - {t("sets.routing.howItWorks")} + + {isProxy + ? t("sets.routing.howItWorksProxy") + : t("sets.routing.howItWorks")} + @@ -190,7 +237,7 @@ export const TrafficRouting = ({ })} - {availableIfaces.length === 0 && ( + {availableIfaces.length === 0 && !isProxy && ( {t("sets.routing.noInterfaces")} @@ -205,38 +252,124 @@ export const TrafficRouting = ({ )} - {t("sets.routing.info")} + {isProxy ? t("sets.routing.infoProxy") : t("sets.routing.info")} - - - onChange("routing.egress_interface", e.target.value) - } - helperText={ - shouldShowUnavailableSelected - ? t("sets.routing.interfaceUnavailable") - : t("sets.routing.outputInterfaceHelper") - } - > - {shouldShowUnavailableSelected && ( - - {t("sets.routing.interfaceUnavailableOption", { - iface: routing.egress_interface, - })} - - )} - {availableIfaces.map((iface) => ( - - {iface} - - ))} - - + {!isProxy && ( + + + onChange("routing.egress_interface", e.target.value) + } + helperText={ + shouldShowUnavailableSelected + ? t("sets.routing.interfaceUnavailable") + : t("sets.routing.outputInterfaceHelper") + } + > + {shouldShowUnavailableSelected && ( + + {t("sets.routing.interfaceUnavailableOption", { + iface: routing.egress_interface, + })} + + )} + {availableIfaces.map((iface) => ( + + {iface} + + ))} + + + )} + + {isProxy && ( + <> + + + onChange("routing.upstream.host", e.target.value) + } + helperText={t("sets.routing.upstreamHostHelper")} + placeholder="127.0.0.1" + /> + + + + onChange( + "routing.upstream.port", + Number(e.target.value) || 0, + ) + } + helperText={t("sets.routing.upstreamPortHelper")} + placeholder="1080" + /> + + + + onChange("routing.upstream.username", e.target.value) + } + helperText={t("sets.routing.upstreamAuthHelper")} + autoComplete="new-password" + /> + + + + onChange("routing.upstream.password", e.target.value) + } + helperText={t("sets.routing.upstreamAuthHelper")} + autoComplete="new-password" + /> + + + + onChange("routing.upstream.use_domain", checked) + } + description={t("sets.routing.useDomainDesc")} + /> + + + + onChange("routing.upstream.fail_open", checked) + } + description={t("sets.routing.failOpenDesc")} + /> + {upstream.fail_open && ( + + {t("sets.routing.failOpenWarning")} + + )} + + + + {t("sets.routing.proxyManipulationNote")} + + + + )} 0 { @@ -249,6 +260,8 @@ func runB4(cmd *cobra.Command, args []string) error { return fmt.Errorf("failed to save config: %v", err) } cfgPtr.Store(c) + tproxyResolver.Set(pool.GetMatcher()) + tproxyMgr.SyncConfig(c) tables.RoutingSyncConfig(c) return nil }) @@ -267,6 +280,7 @@ func runB4(cmd *cobra.Command, args []string) error { metrics.RecordEvent("info", fmt.Sprintf("Shutdown initiated by signal: %v", sig)) wd.Stop() + tproxyMgr.Stop() // Perform graceful shutdown with timeout return gracefulShutdown(cfgPtr.Load(), pool, httpServer, socks5Server, mtprotoServer, metrics, discoveryRT) diff --git a/src/nfq/pool.go b/src/nfq/pool.go index 2fad2a5c..39b55fab 100644 --- a/src/nfq/pool.go +++ b/src/nfq/pool.go @@ -187,6 +187,13 @@ func (p *Pool) GetIPBlockCache() IPBlockCache { return p.state.ipBlocker } +func (p *Pool) GetMatcher() *sni.SuffixSet { + if len(p.Workers) == 0 { + return nil + } + return p.Workers[0].getMatcher() +} + func (p *Pool) GetFirstWorkerConfig() *config.Config { if len(p.Workers) == 0 { return nil diff --git a/src/socks5/client.go b/src/socks5/client.go new file mode 100644 index 00000000..53e11983 --- /dev/null +++ b/src/socks5/client.go @@ -0,0 +1,171 @@ +package socks5 + +import ( + "context" + "encoding/binary" + "fmt" + "io" + "net" + "strconv" + "time" +) + +type ClientConfig struct { + Host string + Port int + Username string + Password string + Timeout time.Duration +} + +func DialUpstream(ctx context.Context, cfg ClientConfig, targetHost string, targetPort int) (net.Conn, error) { + if cfg.Host == "" || cfg.Port < 1 || cfg.Port > 65535 { + return nil, fmt.Errorf("invalid upstream config") + } + if targetPort < 1 || targetPort > 65535 { + return nil, fmt.Errorf("invalid target port") + } + + timeout := cfg.Timeout + if timeout <= 0 { + timeout = dialTimeout + } + + d := net.Dialer{Timeout: timeout} + addr := net.JoinHostPort(cfg.Host, strconv.Itoa(cfg.Port)) + conn, err := d.DialContext(ctx, "tcp", addr) + if err != nil { + return nil, fmt.Errorf("dial upstream: %w", err) + } + + deadline := time.Now().Add(timeout) + _ = conn.SetDeadline(deadline) + + if err := clientGreet(conn, cfg.Username, cfg.Password); err != nil { + conn.Close() + return nil, err + } + if err := clientConnect(conn, targetHost, targetPort); err != nil { + conn.Close() + return nil, err + } + + _ = conn.SetDeadline(time.Time{}) + return conn, nil +} + +func clientGreet(conn net.Conn, user, pass string) error { + useAuth := user != "" || pass != "" + + var greet []byte + if useAuth { + greet = []byte{socks5Version, 2, authNone, authUserPass} + } else { + greet = []byte{socks5Version, 1, authNone} + } + if _, err := conn.Write(greet); err != nil { + return fmt.Errorf("greet write: %w", err) + } + + resp := make([]byte, 2) + if _, err := io.ReadFull(conn, resp); err != nil { + return fmt.Errorf("greet read: %w", err) + } + if resp[0] != socks5Version { + return fmt.Errorf("upstream bad version: %d", resp[0]) + } + switch resp[1] { + case authNone: + return nil + case authUserPass: + if !useAuth { + return fmt.Errorf("upstream requires auth but none configured") + } + return clientUserPass(conn, user, pass) + case authNoAccept: + return fmt.Errorf("upstream rejected all auth methods") + default: + return fmt.Errorf("upstream selected unsupported auth: %d", resp[1]) + } +} + +func clientUserPass(conn net.Conn, user, pass string) error { + if len(user) > 255 || len(pass) > 255 { + return fmt.Errorf("user/pass too long") + } + buf := make([]byte, 0, 3+len(user)+len(pass)) + buf = append(buf, authSubVersion, byte(len(user))) + buf = append(buf, user...) + buf = append(buf, byte(len(pass))) + buf = append(buf, pass...) + if _, err := conn.Write(buf); err != nil { + return fmt.Errorf("auth write: %w", err) + } + resp := make([]byte, 2) + if _, err := io.ReadFull(conn, resp); err != nil { + return fmt.Errorf("auth read: %w", err) + } + if resp[0] != authSubVersion || resp[1] != 0 { + return fmt.Errorf("upstream auth failed") + } + return nil +} + +func clientConnect(conn net.Conn, targetHost string, targetPort int) error { + req := []byte{socks5Version, cmdConnect, 0x00} + + if ip := net.ParseIP(targetHost); ip != nil { + if v4 := ip.To4(); v4 != nil { + req = append(req, atypIPv4) + req = append(req, v4...) + } else { + req = append(req, atypIPv6) + req = append(req, ip.To16()...) + } + } else { + if len(targetHost) > 255 { + return fmt.Errorf("target host too long") + } + req = append(req, atypDomain, byte(len(targetHost))) + req = append(req, targetHost...) + } + + var portBuf [2]byte + binary.BigEndian.PutUint16(portBuf[:], uint16(targetPort)) + req = append(req, portBuf[:]...) + + if _, err := conn.Write(req); err != nil { + return fmt.Errorf("connect write: %w", err) + } + + head := make([]byte, 4) + if _, err := io.ReadFull(conn, head); err != nil { + return fmt.Errorf("connect reply head: %w", err) + } + if head[0] != socks5Version { + return fmt.Errorf("upstream bad version in reply: %d", head[0]) + } + if head[1] != repSuccess { + return fmt.Errorf("upstream connect rejected: code=%d", head[1]) + } + + var skip int + switch head[3] { + case atypIPv4: + skip = 4 + case atypIPv6: + skip = 16 + case atypDomain: + l := make([]byte, 1) + if _, err := io.ReadFull(conn, l); err != nil { + return fmt.Errorf("connect reply addr len: %w", err) + } + skip = int(l[0]) + default: + return fmt.Errorf("upstream bad atyp in reply: %d", head[3]) + } + if _, err := io.ReadFull(conn, make([]byte, skip+2)); err != nil { + return fmt.Errorf("connect reply addr/port: %w", err) + } + return nil +} diff --git a/src/tables/routing.go b/src/tables/routing.go index 167b7274..e3d9701b 100644 --- a/src/tables/routing.go +++ b/src/tables/routing.go @@ -17,9 +17,12 @@ import ( const hostRouteCTMark = uint32(0x40000000) type routeState struct { + mode string mark uint32 table int iface string + tproxyPort int + upstreamKey string sourcesKey string setV4 string setV6 string @@ -84,7 +87,11 @@ func RoutingHandleDNS(cfg *config.Config, set *config.SetConfig, ips []net.IP) { if cfg == nil || set == nil || !set.Routing.Enabled || len(ips) == 0 { return } - if set.Routing.EgressInterface == "" { + mode := set.Routing.Mode + if mode == "" { + mode = config.RoutingModeInterface + } + if mode == config.RoutingModeInterface && set.Routing.EgressInterface == "" { return } if !hasBinary("ip") { @@ -106,33 +113,33 @@ func RoutingHandleDNS(cfg *config.Config, set *config.SetConfig, ips []net.IP) { return } + cur := buildRouteState(cfg, set) sources := routeNormalizedSources(set.Routing.SourceInterfaces) - sourcesKey := strings.Join(sources, ",") - setV4, setV6 := routeBuildSetNames(set.Id) - chainPre, chainOut, chainSNAT := routeBuildChainNames(set.Id) - mark, table := routeResolveIDs(cfg, set) - - cur := routeState{ - mark: mark, table: table, - iface: set.Routing.EgressInterface, sourcesKey: sourcesKey, - setV4: setV4, setV6: setV6, - chainPre: chainPre, chainOut: chainOut, chainSNAT: chainSNAT, - } if old, ok := routeRuleCache[set.Id]; ok { - if old.mark != cur.mark || old.table != cur.table || old.iface != cur.iface || old.sourcesKey != cur.sourcesKey { - routeCleanupRule(be, old) + if !routeStateEqual(old, cur) { + routeCleanupAny(be, old) delete(routeRuleCache, set.Id) } } if _, ok := routeRuleCache[set.Id]; !ok { - if err := routeEnsureRule(be, cfg, set, cur, sources); err != nil { + var err error + if cur.mode == config.RoutingModeProxy { + err = routeEnsureProxyRule(be, cfg, set, cur, sources) + } else { + err = routeEnsureRule(be, cfg, set, cur, sources) + } + if err != nil { log.Errorf("Routing: failed to ensure rule for set '%s': %v", set.Name, err) return } routeRuleCache[set.Id] = cur - log.Infof("Routing [%s]: enabled set '%s' -> iface=%s mark=0x%x table=%d", be.name(), set.Name, set.Routing.EgressInterface, mark, table) + if cur.mode == config.RoutingModeProxy { + log.Infof("Routing [%s]: enabled proxy set '%s' -> %s:%d mark=0x%x port=%d", be.name(), set.Name, set.Routing.Upstream.Host, set.Routing.Upstream.Port, cur.mark, cur.tproxyPort) + } else { + log.Infof("Routing [%s]: enabled set '%s' -> iface=%s mark=0x%x table=%d", be.name(), set.Name, set.Routing.EgressInterface, cur.mark, cur.table) + } } ttl := set.Routing.IPTTLSeconds @@ -143,6 +150,56 @@ func RoutingHandleDNS(cfg *config.Config, set *config.SetConfig, ips []net.IP) { routeAddIPsToSets(be, cur, ttl, ips, cfg.Queue.IPv4Enabled, cfg.Queue.IPv6Enabled) } +func buildRouteState(cfg *config.Config, set *config.SetConfig) routeState { + mode := set.Routing.Mode + if mode == "" { + mode = config.RoutingModeInterface + } + sources := routeNormalizedSources(set.Routing.SourceInterfaces) + sourcesKey := strings.Join(sources, ",") + setV4, setV6 := routeBuildSetNames(set.Id) + chainPre, chainOut, chainSNAT := routeBuildChainNames(set.Id) + + st := routeState{ + mode: mode, + sourcesKey: sourcesKey, + setV4: setV4, setV6: setV6, + chainPre: chainPre, chainOut: chainOut, chainSNAT: chainSNAT, + } + + if mode == config.RoutingModeProxy { + mark, port := proxyMarkAndPort(set) + st.mark = mark + st.table = proxyTable(mark) + st.tproxyPort = port + st.upstreamKey = fmt.Sprintf("%s:%d|%s", set.Routing.Upstream.Host, set.Routing.Upstream.Port, set.Routing.Upstream.Username) + } else { + mark, table := routeResolveIDs(cfg, set) + st.mark = mark + st.table = table + st.iface = set.Routing.EgressInterface + } + return st +} + +func routeStateEqual(a, b routeState) bool { + return a.mode == b.mode && + a.mark == b.mark && + a.table == b.table && + a.iface == b.iface && + a.tproxyPort == b.tproxyPort && + a.upstreamKey == b.upstreamKey && + a.sourcesKey == b.sourcesKey +} + +func routeCleanupAny(be routeBackend, st routeState) { + if st.mode == config.RoutingModeProxy { + routeCleanupProxyRule(be, st) + return + } + routeCleanupRule(be, st) +} + func routeAddIPsToSets(be routeBackend, st routeState, ttl int, ips []net.IP, ipv4Enabled, ipv6Enabled bool) { v4 := make([]string, 0, len(ips)) v6 := make([]string, 0, len(ips)) @@ -248,7 +305,7 @@ func RoutingClearAll() { } } else { for _, st := range routeRuleCache { - routeCleanupRule(be, st) + routeCleanupAny(be, st) } be.clearAll() } @@ -288,7 +345,17 @@ func RoutingSyncConfig(cfg *config.Config) { desired := make(map[string]*config.SetConfig, len(cfg.Sets)) for _, set := range cfg.Sets { - if set == nil || !set.Enabled || !set.Routing.Enabled || set.Routing.EgressInterface == "" { + if set == nil || !set.Enabled || !set.Routing.Enabled { + continue + } + mode := set.Routing.Mode + if mode == "" { + mode = config.RoutingModeInterface + } + if mode == config.RoutingModeInterface && set.Routing.EgressInterface == "" { + continue + } + if mode == config.RoutingModeProxy && set.Routing.Upstream.Port < 1 { continue } desired[set.Id] = set @@ -296,7 +363,7 @@ func RoutingSyncConfig(cfg *config.Config) { for setID, st := range routeRuleCache { if _, ok := desired[setID]; !ok { - routeCleanupRule(be, st) + routeCleanupAny(be, st) delete(routeRuleCache, setID) } } @@ -310,28 +377,24 @@ func RoutingSyncConfig(cfg *config.Config) { continue } + cur := buildRouteState(cfg, set) sources := routeNormalizedSources(set.Routing.SourceInterfaces) - sourcesKey := strings.Join(sources, ",") - setV4, setV6 := routeBuildSetNames(set.Id) - chainPre, chainOut, chainSNAT := routeBuildChainNames(set.Id) - mark, table := routeResolveIDs(cfg, set) - - cur := routeState{ - mark: mark, table: table, - iface: set.Routing.EgressInterface, sourcesKey: sourcesKey, - setV4: setV4, setV6: setV6, - chainPre: chainPre, chainOut: chainOut, chainSNAT: chainSNAT, - } if old, ok := routeRuleCache[set.Id]; ok { - if old.mark != cur.mark || old.table != cur.table || old.iface != cur.iface || old.sourcesKey != cur.sourcesKey { - routeCleanupRule(be, old) + if !routeStateEqual(old, cur) { + routeCleanupAny(be, old) delete(routeRuleCache, set.Id) } } if _, ok := routeRuleCache[set.Id]; !ok { - if err := routeEnsureRule(be, cfg, set, cur, sources); err != nil { + var err error + if cur.mode == config.RoutingModeProxy { + err = routeEnsureProxyRule(be, cfg, set, cur, sources) + } else { + err = routeEnsureRule(be, cfg, set, cur, sources) + } + if err != nil { log.Errorf("Routing: failed to ensure rule for set '%s' during sync: %v", set.Name, err) continue } @@ -350,6 +413,9 @@ func RoutingSyncConfig(cfg *config.Config) { routeIfaceAuto = make(map[string]routeState) for _, st := range routeRuleCache { + if st.mode == config.RoutingModeProxy || st.iface == "" { + continue + } if _, ok := routeIfaceAuto[st.iface]; !ok { routeIfaceAuto[st.iface] = routeState{mark: st.mark, table: st.table} } @@ -374,7 +440,14 @@ func RoutingPeriodicReResolve(cfg *config.Config) { var setsToResolve []*config.SetConfig for _, set := range cfg.Sets { - if set == nil || !set.Enabled || !set.Routing.Enabled || set.Routing.EgressInterface == "" { + if set == nil || !set.Enabled || !set.Routing.Enabled { + continue + } + mode := set.Routing.Mode + if mode == "" { + mode = config.RoutingModeInterface + } + if mode == config.RoutingModeInterface && set.Routing.EgressInterface == "" { continue } if _, ok := routeRuleCache[set.Id]; !ok { @@ -590,7 +663,7 @@ func RoutingReinstallForInterface(cfg *config.Config, iface string) { ipv6 := cfg.Queue.IPv6Enabled count := 0 for _, st := range routeRuleCache { - if st.iface != iface { + if st.mode == config.RoutingModeProxy || st.iface != iface { continue } routeEnsurePolicyRouting(st.iface, st.mark, st.table, ipv4, ipv6) diff --git a/src/tables/routing_proxy.go b/src/tables/routing_proxy.go new file mode 100644 index 00000000..0fe71cfa --- /dev/null +++ b/src/tables/routing_proxy.go @@ -0,0 +1,367 @@ +package tables + +import ( + "fmt" + "os" + "strings" + + "github.com/daniellavrushin/b4/config" + "github.com/daniellavrushin/b4/log" + "github.com/daniellavrushin/b4/tproxy" +) + +const proxyRulePriority = 5 + +func proxyMarkAndPort(set *config.SetConfig) (uint32, int) { + mark := tproxy.MarkForSet(set.Id, set.Routing.FWMark) + port := tproxy.PortFor(mark) + return mark, port +} + +func proxyTable(mark uint32) int { + return 1000 + int(mark) +} + +func routeEnsureProxyRule(be routeBackend, cfg *config.Config, set *config.SetConfig, st routeState, sources []string) error { + if cfg.Queue.IPv4Enabled { + if err := be.ensureIPSet(st.setV4, false); err != nil { + return err + } + } + if cfg.Queue.IPv6Enabled { + if err := be.ensureIPSet(st.setV6, true); err != nil { + return err + } + } + if err := be.ensureChain(st.chainPre, true); err != nil { + return err + } + be.flushChain(st.chainPre, true) + + queueMark := routeQueueBypassMark(cfg) + be.addBypassRule(st.chainPre, queueMark) + be.addBypassRule(st.chainPre, st.mark) + + port, _ := portFromState(st) + legacy := isLegacyIptBackend(be) + + switch be.name() { + case backendNFTables: + if cfg.Queue.IPv4Enabled { + addProxyDivertRuleNft(st.chainPre, false, st.setV4, st.mark) + addProxyTProxyRuleNft(st.chainPre, false, st.setV4, st.mark, port, sources) + } + if cfg.Queue.IPv6Enabled { + addProxyDivertRuleNft(st.chainPre, true, st.setV6, st.mark) + addProxyTProxyRuleNft(st.chainPre, true, st.setV6, st.mark, port, sources) + } + default: + if cfg.Queue.IPv4Enabled { + addProxyDivertRuleIpt(false, st.chainPre, st.setV4, st.mark, legacy) + addProxyTProxyRuleIpt(false, st.chainPre, st.setV4, st.mark, port, sources, legacy) + } + if cfg.Queue.IPv6Enabled { + addProxyDivertRuleIpt(true, st.chainPre, st.setV6, st.mark, legacy) + addProxyTProxyRuleIpt(true, st.chainPre, st.setV6, st.mark, port, sources, legacy) + } + } + + insertProxyJumpAtTop(be, st.chainPre) + addProxyInputAccept(be, st.mark) + + routeEnsureLocalDelivery(st.mark, st.table, cfg.Queue.IPv4Enabled, cfg.Queue.IPv6Enabled) + return nil +} + +func routeCleanupProxyRule(be routeBackend, st routeState) { + markStr := fmt.Sprintf("0x%x", st.mark) + markStrMask := fmt.Sprintf("0x%x/0x%x", st.mark, st.mark) + tableStr := fmt.Sprintf("%d", st.table) + + if hasBinary("ip") { + routeDelRuleLoop(false, markStr, tableStr) + routeDelRuleLoop(false, markStrMask, tableStr) + routeDelRuleLoop(true, markStr, tableStr) + routeDelRuleLoop(true, markStrMask, tableStr) + runLogged("routing: delete proxy local route v4", "ip", "route", "del", "local", "0.0.0.0/0", "dev", "lo", "table", tableStr) + runLogged("routing: delete proxy local route v6", "ip", "-6", "route", "del", "local", "::/0", "dev", "lo", "table", tableStr) + } + + removeProxyInputAccept(be, st.mark) + be.deleteJumpRules("PREROUTING", st.chainPre, true) + be.flushChain(st.chainPre, true) + be.deleteChain(st.chainPre, true) + be.flushIPSet(st.setV4) + be.destroyIPSet(st.setV4) + be.flushIPSet(st.setV6) + be.destroyIPSet(st.setV6) +} + +func routeEnsureLocalDelivery(mark uint32, table int, ipv4, ipv6 bool) { + markStrMask := fmt.Sprintf("0x%x/0x%x", mark, mark) + tableStr := fmt.Sprintf("%d", table) + prioStr := fmt.Sprintf("%d", proxyRulePriority) + + writeSysctl("/proc/sys/net/ipv4/conf/lo/rp_filter", "0") + writeSysctl("/proc/sys/net/ipv4/conf/all/rp_filter", "2") + + if ipv4 { + routeDelRuleLoop(false, fmt.Sprintf("0x%x", mark), tableStr) + routeDelRuleLoop(false, markStrMask, tableStr) + runLogged("routing: add ip rule v4 (proxy)", "ip", "rule", "add", "fwmark", markStrMask, "lookup", tableStr, "priority", prioStr) + runLogged("routing: add local route v4 (proxy)", "ip", "route", "replace", "local", "0.0.0.0/0", "dev", "lo", "table", tableStr) + } + if ipv6 { + routeDelRuleLoop(true, fmt.Sprintf("0x%x", mark), tableStr) + routeDelRuleLoop(true, markStrMask, tableStr) + runLogged("routing: add ip rule v6 (proxy)", "ip", "-6", "rule", "add", "fwmark", markStrMask, "lookup", tableStr, "priority", prioStr) + runLogged("routing: add local route v6 (proxy)", "ip", "-6", "route", "replace", "local", "::/0", "dev", "lo", "table", tableStr) + } +} + +func writeSysctl(path, value string) { + cur, err := os.ReadFile(path) + if err == nil && strings.TrimSpace(string(cur)) == value { + return + } + if err := os.WriteFile(path, []byte(value), 0644); err != nil { + log.Tracef("routing: sysctl %s=%s failed: %v", path, value, err) + } +} + +func deleteNftJumpRules(table, parentChain, targetChain string) { + out, err := run("nft", "-a", "list", "chain", "inet", table, parentChain) + if err != nil { + log.Tracef("routing: list nft chain inet %s %s failed: %v", table, parentChain, err) + return + } + for _, line := range strings.Split(out, "\n") { + handleIdx := strings.LastIndex(line, "# handle ") + if handleIdx == -1 { + continue + } + rule := strings.TrimSpace(line[:handleIdx]) + if !strings.Contains(rule, "jump "+targetChain) { + continue + } + handle := strings.TrimSpace(line[handleIdx+len("# handle "):]) + if handle == "" { + continue + } + runLogged("routing: delete leftover prerouting jump (proxy)", + "nft", "delete", "rule", "inet", table, parentChain, "handle", handle) + } +} + +func insertProxyJumpAtTop(be routeBackend, chain string) { + if be.name() == backendNFTables { + deleteNftJumpRules(routeNftTable, routeNftPrerouting, chain) + runLogged("routing: insert prerouting jump (proxy)", "nft", "insert", "rule", "inet", routeNftTable, routeNftPrerouting, "jump", chain) + return + } + for _, fam := range []string{backendIPTables, backendIP6Tables, backendIPTablesLegacy, backendIP6TablesLegacy} { + if !hasBinary(fam) { + continue + } + for i := 0; i < 100; i++ { + if _, err := run(fam, "-w", "-t", "mangle", "-D", "PREROUTING", "-j", chain); err != nil { + break + } + } + runLogged("routing: insert prerouting jump (proxy) "+fam, + fam, "-w", "-t", "mangle", "-I", "PREROUTING", "1", "-j", chain) + } +} + +func addProxyDivertRuleIpt(v6 bool, chain, setName string, mark uint32, legacy bool) { + cmd := backendIPTables + if v6 { + cmd = backendIP6Tables + } + if legacy { + if v6 { + cmd = backendIP6TablesLegacy + } else { + cmd = backendIPTablesLegacy + } + } + if !hasBinary(cmd) { + return + } + markHex := fmt.Sprintf("0x%x/0x%x", mark, mark) + runLogged("routing: add divert mark "+chain, + cmd, "-w", "-t", "mangle", "-A", chain, "-p", "tcp", + "-m", "socket", "--transparent", + "-m", "set", "--match-set", setName, "dst", + "-j", "MARK", "--set-mark", markHex) + runLogged("routing: add divert accept "+chain, + cmd, "-w", "-t", "mangle", "-A", chain, "-p", "tcp", + "-m", "socket", "--transparent", + "-m", "set", "--match-set", setName, "dst", + "-j", "ACCEPT") +} + +func addProxyDivertRuleNft(chain string, v6 bool, setName string, mark uint32) { + markHex := fmt.Sprintf("0x%x", mark) + args := []string{"add", "rule", "inet", routeNftTable, chain} + if v6 { + args = append(args, "ip6", "daddr", "@"+setName) + } else { + args = append(args, "ip", "daddr", "@"+setName) + } + args = append(args, "socket", "transparent", "1", "meta", "mark", "set", markHex, "accept") + runLogged("routing: add divert "+chain, append([]string{"nft"}, args...)...) +} + +func addProxyInputAccept(be routeBackend, mark uint32) { + markHex := fmt.Sprintf("0x%x/0x%x", mark, mark) + if be.name() == backendNFTables { + runLogged("routing: add input accept (proxy)", + "nft", "insert", "rule", "inet", "filter", "input", + "meta", "mark", "&", fmt.Sprintf("0x%x", mark), "==", fmt.Sprintf("0x%x", mark), "accept") + return + } + for _, fam := range []string{backendIPTables, backendIP6Tables, backendIPTablesLegacy, backendIP6TablesLegacy} { + if !hasBinary(fam) { + continue + } + for i := 0; i < 100; i++ { + if _, err := run(fam, "-w", "-D", "INPUT", "-m", "mark", "--mark", markHex, "-j", "ACCEPT"); err != nil { + break + } + } + runLogged("routing: add input accept (proxy) "+fam, + fam, "-w", "-I", "INPUT", "1", "-m", "mark", "--mark", markHex, "-j", "ACCEPT") + } +} + +func removeProxyInputAccept(be routeBackend, mark uint32) { + markHex := fmt.Sprintf("0x%x/0x%x", mark, mark) + if be.name() == backendNFTables { + markStr := fmt.Sprintf("0x%x", mark) + out, err := run("nft", "-a", "list", "chain", "inet", "filter", "input") + if err != nil { + log.Tracef("routing: list nft inet filter input failed: %v", err) + return + } + for _, line := range strings.Split(out, "\n") { + handleIdx := strings.LastIndex(line, "# handle ") + if handleIdx == -1 { + continue + } + rule := strings.TrimSpace(line[:handleIdx]) + if !strings.Contains(rule, markStr) || !strings.Contains(rule, "accept") { + continue + } + handle := strings.TrimSpace(line[handleIdx+len("# handle "):]) + if handle == "" { + continue + } + runLogged("routing: delete input accept (proxy)", + "nft", "delete", "rule", "inet", "filter", "input", "handle", handle) + } + return + } + for _, fam := range []string{backendIPTables, backendIP6Tables, backendIPTablesLegacy, backendIP6TablesLegacy} { + if !hasBinary(fam) { + continue + } + for i := 0; i < 100; i++ { + if _, err := run(fam, "-w", "-D", "INPUT", "-m", "mark", "--mark", markHex, "-j", "ACCEPT"); err != nil { + break + } + } + } +} + +func addProxyTProxyRuleNft(chain string, v6 bool, setName string, mark uint32, port int, sources []string) { + markHex := fmt.Sprintf("0x%x", mark) + portStr := fmt.Sprintf(":%d", port) + + emit := func(src string) { + args := []string{"add", "rule", "inet", routeNftTable, chain} + if src != "" { + args = append(args, "iifname", src) + } + if v6 { + args = append(args, + "meta", "l4proto", "tcp", + "ip6", "daddr", "@"+setName, + "meta", "mark", "set", markHex, + "tproxy", "ip6", "to", portStr, + "accept", + ) + } else { + args = append(args, + "ip", "protocol", "tcp", + "ip", "daddr", "@"+setName, + "meta", "mark", "set", markHex, + "tproxy", "ip", "to", portStr, + "accept", + ) + } + runLogged("routing: add tproxy rule "+chain, append([]string{"nft"}, args...)...) + } + + if len(sources) == 0 { + emit("") + return + } + for _, src := range sources { + emit(src) + } +} + +func addProxyTProxyRuleIpt(v6 bool, chain, setName string, mark uint32, port int, sources []string, legacy bool) { + cmd := backendIPTables + if v6 { + cmd = backendIP6Tables + } + if legacy { + if v6 { + cmd = backendIP6TablesLegacy + } else { + cmd = backendIPTablesLegacy + } + } + if !hasBinary(cmd) { + return + } + markHex := fmt.Sprintf("0x%x/0x%x", mark, mark) + + emit := func(src string) { + args := []string{cmd, "-w", "-t", "mangle", "-A", chain, "-p", "tcp"} + if src != "" { + args = append(args, "-i", src) + } + args = append(args, + "-m", "set", "--match-set", setName, "dst", + "-j", "TPROXY", + "--tproxy-mark", markHex, + "--on-port", fmt.Sprintf("%d", port), + ) + runLogged("routing: add tproxy rule "+chain, args...) + } + + if len(sources) == 0 { + emit("") + return + } + for _, src := range sources { + emit(src) + } +} + +func portFromState(st routeState) (int, bool) { + if st.tproxyPort > 0 { + return st.tproxyPort, true + } + return tproxy.PortFor(st.mark), false +} + +func isLegacyIptBackend(be routeBackend) bool { + if ipt, ok := be.(*routeIptBackend); ok { + return ipt.legacy + } + return false +} diff --git a/src/tproxy/listener.go b/src/tproxy/listener.go new file mode 100644 index 00000000..8d644e69 --- /dev/null +++ b/src/tproxy/listener.go @@ -0,0 +1,212 @@ +package tproxy + +import ( + "context" + "errors" + "fmt" + "io" + "net" + "sync" + "sync/atomic" + "syscall" + "time" + + "github.com/daniellavrushin/b4/log" + "github.com/daniellavrushin/b4/socks5" + "golang.org/x/sys/unix" +) + +type DomainResolver interface { + DomainFor(ip net.IP) string +} + +type Listener struct { + SetID string + SetName string + BindAddr string + BindAddr6 string + Port int + Upstream socks5.ClientConfig + UseDomain bool + FailOpen bool + Resolver DomainResolver + + ctx context.Context + cancel context.CancelFunc + lnV4 net.Listener + lnV6 net.Listener + + activeConns atomic.Int64 +} + +func (l *Listener) Start(parent context.Context) error { + if l.Port < 1 || l.Port > 65535 { + return fmt.Errorf("invalid tproxy port: %d", l.Port) + } + bind4 := l.BindAddr + if bind4 == "" { + bind4 = "0.0.0.0" + } + bind6 := l.BindAddr6 + if bind6 == "" { + bind6 = "::" + } + addr4 := net.JoinHostPort(bind4, fmt.Sprintf("%d", l.Port)) + addr6 := net.JoinHostPort(bind6, fmt.Sprintf("%d", l.Port)) + + l.ctx, l.cancel = context.WithCancel(parent) + + lnV4, err := listenTransparent(l.ctx, "tcp4", addr4, false) + if err != nil { + l.cancel() + return fmt.Errorf("tproxy v4 listen %s: %w", addr4, err) + } + l.lnV4 = lnV4 + go l.acceptLoop(lnV4, "v4") + log.Infof("tproxy: listening on %s (v4) for set %q -> %s:%d", addr4, l.SetName, l.Upstream.Host, l.Upstream.Port) + + lnV6, err := listenTransparent(l.ctx, "tcp6", addr6, true) + if err != nil { + log.Tracef("tproxy: v6 listener disabled for set %q: %v", l.SetName, err) + return nil + } + l.lnV6 = lnV6 + go l.acceptLoop(lnV6, "v6") + log.Infof("tproxy: listening on %s (v6) for set %q -> %s:%d", addr6, l.SetName, l.Upstream.Host, l.Upstream.Port) + + return nil +} + +func listenTransparent(ctx context.Context, network, addr string, v6 bool) (net.Listener, error) { + lc := net.ListenConfig{ + Control: func(network, address string, c syscall.RawConn) error { + var ctlErr error + err := c.Control(func(fd uintptr) { + if e := unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_REUSEADDR, 1); e != nil { + ctlErr = fmt.Errorf("set SO_REUSEADDR: %w", e) + return + } + if v6 { + if e := unix.SetsockoptInt(int(fd), unix.IPPROTO_IPV6, unix.IPV6_V6ONLY, 1); e != nil { + ctlErr = fmt.Errorf("set IPV6_V6ONLY: %w", e) + return + } + if e := unix.SetsockoptInt(int(fd), unix.SOL_IPV6, unix.IPV6_TRANSPARENT, 1); e != nil { + ctlErr = fmt.Errorf("set IPV6_TRANSPARENT: %w", e) + return + } + } else { + if e := unix.SetsockoptInt(int(fd), unix.SOL_IP, unix.IP_TRANSPARENT, 1); e != nil { + ctlErr = fmt.Errorf("set IP_TRANSPARENT: %w", e) + return + } + } + }) + if err != nil { + return err + } + return ctlErr + }, + } + return lc.Listen(ctx, network, addr) +} + +func (l *Listener) Stop() error { + if l.cancel != nil { + l.cancel() + } + var firstErr error + if l.lnV4 != nil { + if err := l.lnV4.Close(); err != nil && firstErr == nil { + firstErr = err + } + } + if l.lnV6 != nil { + if err := l.lnV6.Close(); err != nil && firstErr == nil { + firstErr = err + } + } + return firstErr +} + +func (l *Listener) Active() int64 { + return l.activeConns.Load() +} + +func (l *Listener) acceptLoop(ln net.Listener, family string) { + for { + conn, err := ln.Accept() + if err != nil { + if l.ctx.Err() != nil { + return + } + if errors.Is(err, net.ErrClosed) { + return + } + log.Tracef("tproxy: accept error on set %q (%s): %v", l.SetName, family, err) + time.Sleep(50 * time.Millisecond) + continue + } + go l.handle(conn) + } +} + +func (l *Listener) handle(client net.Conn) { + l.activeConns.Add(1) + defer l.activeConns.Add(-1) + defer client.Close() + + tcpAddr, ok := client.LocalAddr().(*net.TCPAddr) + if !ok || tcpAddr == nil || tcpAddr.IP == nil { + log.Tracef("tproxy: missing original dst on set %q", l.SetName) + return + } + origIP := tcpAddr.IP + origPort := tcpAddr.Port + + targetHost := origIP.String() + if l.UseDomain && l.Resolver != nil { + if d := l.Resolver.DomainFor(origIP); d != "" { + targetHost = d + } + } + + dialCtx, cancel := context.WithTimeout(l.ctx, 15*time.Second) + upstream, err := socks5.DialUpstream(dialCtx, l.Upstream, targetHost, origPort) + cancel() + if err != nil { + log.Tracef("tproxy: upstream dial failed for %s:%d on set %q: %v", targetHost, origPort, l.SetName, err) + if !l.FailOpen { + return + } + direct, derr := net.DialTimeout("tcp", net.JoinHostPort(origIP.String(), fmt.Sprintf("%d", origPort)), 10*time.Second) + if derr != nil { + log.Tracef("tproxy: fail-open direct dial failed: %v", derr) + return + } + upstream = direct + } + defer upstream.Close() + + pipe(client, upstream) +} + +func pipe(a, b net.Conn) { + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + _, _ = io.Copy(a, b) + if c, ok := a.(*net.TCPConn); ok { + _ = c.CloseWrite() + } + }() + go func() { + defer wg.Done() + _, _ = io.Copy(b, a) + if c, ok := b.(*net.TCPConn); ok { + _ = c.CloseWrite() + } + }() + wg.Wait() +} diff --git a/src/tproxy/manager.go b/src/tproxy/manager.go new file mode 100644 index 00000000..e3495055 --- /dev/null +++ b/src/tproxy/manager.go @@ -0,0 +1,143 @@ +package tproxy + +import ( + "context" + "sync" + "time" + + "github.com/daniellavrushin/b4/config" + "github.com/daniellavrushin/b4/log" + "github.com/daniellavrushin/b4/socks5" +) + +type Manager struct { + mu sync.Mutex + listeners map[string]*Listener + resolver DomainResolver + ctx context.Context + cancel context.CancelFunc +} + +func NewManager(resolver DomainResolver) *Manager { + ctx, cancel := context.WithCancel(context.Background()) + return &Manager{ + listeners: make(map[string]*Listener), + resolver: resolver, + ctx: ctx, + cancel: cancel, + } +} + +func (m *Manager) SetResolver(r DomainResolver) { + m.mu.Lock() + defer m.mu.Unlock() + m.resolver = r + for _, l := range m.listeners { + l.Resolver = r + } +} + +func (m *Manager) SyncConfig(cfg *config.Config) { + if cfg == nil { + return + } + + m.mu.Lock() + defer m.mu.Unlock() + + desired := make(map[string]*config.SetConfig, len(cfg.Sets)) + for _, set := range cfg.Sets { + if set == nil || !set.Enabled || !set.Routing.Enabled { + continue + } + if set.Routing.Mode != config.RoutingModeProxy { + continue + } + desired[set.Id] = set + } + + for id, l := range m.listeners { + set, keep := desired[id] + if !keep { + log.Infof("tproxy: stopping listener for removed set %q", l.SetName) + _ = l.Stop() + delete(m.listeners, id) + continue + } + mark := effectiveMark(set) + port := PortFor(mark) + desiredHost := set.Routing.Upstream.Host + if desiredHost == "" { + desiredHost = "127.0.0.1" + } + if l.Port != port || + l.Upstream.Host != desiredHost || + l.Upstream.Port != set.Routing.Upstream.Port || + l.Upstream.Username != set.Routing.Upstream.Username || + l.Upstream.Password != set.Routing.Upstream.Password || + l.UseDomain != set.Routing.Upstream.UseDomain || + l.FailOpen != set.Routing.Upstream.FailOpen { + log.Infof("tproxy: restarting listener for set %q (config changed)", set.Name) + _ = l.Stop() + delete(m.listeners, id) + } + } + + for id, set := range desired { + if _, ok := m.listeners[id]; ok { + continue + } + mark := effectiveMark(set) + port := PortFor(mark) + host := set.Routing.Upstream.Host + if host == "" { + host = "127.0.0.1" + } + l := &Listener{ + SetID: set.Id, + SetName: set.Name, + Port: port, + Upstream: socks5.ClientConfig{ + Host: host, + Port: set.Routing.Upstream.Port, + Username: set.Routing.Upstream.Username, + Password: set.Routing.Upstream.Password, + Timeout: 10 * time.Second, + }, + UseDomain: set.Routing.Upstream.UseDomain, + FailOpen: set.Routing.Upstream.FailOpen, + Resolver: m.resolver, + } + if err := l.Start(m.ctx); err != nil { + log.Errorf("tproxy: failed to start listener for set %q: %v", set.Name, err) + continue + } + m.listeners[id] = l + } +} + +func (m *Manager) Stop() { + m.mu.Lock() + defer m.mu.Unlock() + for id, l := range m.listeners { + _ = l.Stop() + delete(m.listeners, id) + } + if m.cancel != nil { + m.cancel() + } +} + +func (m *Manager) PortForSet(set *config.SetConfig) int { + if set == nil { + return 0 + } + return PortFor(effectiveMark(set)) +} + +func effectiveMark(set *config.SetConfig) uint32 { + if set == nil { + return 0 + } + return MarkForSet(set.Id, set.Routing.FWMark) +} diff --git a/src/tproxy/port.go b/src/tproxy/port.go new file mode 100644 index 00000000..569f26eb --- /dev/null +++ b/src/tproxy/port.go @@ -0,0 +1,27 @@ +package tproxy + +import "hash/fnv" + +const ( + DefaultPortBase = 13000 + PortRange = 50000 + MarkBase = 0x10000 + MarkRange = 0xFE00 +) + +func MarkForSet(setID string, pinned uint32) uint32 { + if pinned > 0 { + return pinned + } + h := fnv.New32a() + _, _ = h.Write([]byte(setID)) + return MarkBase + (h.Sum32() % MarkRange) +} + +func PortFor(mark uint32) int { + if mark == 0 { + return DefaultPortBase + } + return DefaultPortBase + int(mark%PortRange) +} + diff --git a/src/tproxy/resolver.go b/src/tproxy/resolver.go new file mode 100644 index 00000000..90f77800 --- /dev/null +++ b/src/tproxy/resolver.go @@ -0,0 +1,34 @@ +package tproxy + +import ( + "net" + "sync/atomic" + + "github.com/daniellavrushin/b4/sni" +) + +type LearnedIPResolver struct { + matcher atomic.Pointer[sni.SuffixSet] +} + +func NewLearnedIPResolver(m *sni.SuffixSet) *LearnedIPResolver { + r := &LearnedIPResolver{} + r.matcher.Store(m) + return r +} + +func (r *LearnedIPResolver) Set(m *sni.SuffixSet) { + r.matcher.Store(m) +} + +func (r *LearnedIPResolver) DomainFor(ip net.IP) string { + m := r.matcher.Load() + if m == nil || ip == nil { + return "" + } + matched, _, domain := m.MatchLearnedIP(ip) + if !matched { + return "" + } + return domain +}