diff --git a/intra/common.go b/intra/common.go index 827e5cf3..1254e89b 100644 --- a/intra/common.go +++ b/intra/common.go @@ -9,6 +9,7 @@ package intra import ( "context" "fmt" + "io" "math/rand" "net" "net/netip" @@ -312,12 +313,20 @@ func (h *baseHandler) forward(local, remote net.Conn, smm *FlowSummary) { isrwext := false didSet := false timeoutsecs := 0 - // enable core.Pipe (sendfile/zero-copy) optimizations on TCP if - // read & write deadlines are not set (as in rwext is effectively - // a no-op) by unwrapping the underlying remote conn from rwext. + // enable core.Pipe (sendfile/zero-copy) optimizations on TCP only + // when no read/write deadline is actually configured (timeoutsecs + // <= 0), in which case rwext is a no-op wrapper and unwrapping is + // safe. Do NOT unwrap merely because didSet is true: SetTimeout + // (via core.SetTimeoutSockOpt) only sets TCP_USER_TIMEOUT, which + // bounds unacknowledged *writes*, not idle *reads*. If remote is + // unwrapped here while a positive timeoutsecs is configured, the + // only mechanism that can bound a stalled Read() (rwext's + // extendr/extendw, which set a real per-call deadline) is lost, + // and a peer that silently stops sending (no RST/FIN) causes + // Read() -- and thus this whole forward() -- to block forever. if r, ok := remote.(rwext); ok { isrwext = true - if timeoutsecs, didSet = r.SetTimeout(); didSet || timeoutsecs <= 0 { + if timeoutsecs, didSet = r.SetTimeout(); timeoutsecs <= 0 { remote = r.Unwrap() // c may be *net.TCPConn or *demuxconn or *dialers.retrier|splitter } } @@ -568,8 +577,41 @@ func (h *baseHandler) Reset() { log.I("com: %s: handler reset", h.proto) } +// aborter is implemented by conns (currently just *netstack.GTCPConn) that +// can forcibly terminate with a TCP RST, as opposed to a graceful FIN. +type aborter interface { + Abort() +} + +// resetOrClose closes op on c, unless err is a genuine abnormal error (not a +// plain io.EOF, which indicates the peer closed gracefully) and c supports +// aborting -- in which case a TCP RST is sent instead of a graceful FIN. +// +// This addresses a previously-known gap (see the old "TODO: Propagate TCP +// RST using local.Abort(), on appropriate errors" note this replaces): +// download() closing the app-facing conn's write-half gracefully (FIN) even +// when the actual cause was an abnormal upstream failure (eg: our own +// idle-read timeout firing because the remote peer silently stopped +// sending) meant the app could observe what looks like a clean, complete +// close -- rather than an unambiguous connection-reset -- for a connection +// that, in fact, terminated abnormally. Some HTTP clients treat a graceful +// close of an in-flight response as ambiguous/retryable at best, or a +// silently-truncated "complete" response at worst; a RST is unambiguous. +func resetOrClose(c io.Closer, op core.CloserOp, err error) { + if err != nil && err != io.EOF { + if a, ok := c.(aborter); ok && a != nil && core.IsNotNil(a) { + // DEBUG-INSTRUMENTATION (temporary, remove once Zee5/PMTUD + // investigation is closed): confirms this RST-on-error path + // actually engages (vs falling through to a graceful close). + log.I("com: dbg: resetOrClose: aborting (RST) %T on err: %v", c, err) + a.Abort() + return + } + } + core.CloseOp(c, op) +} + // upload copies data from remote to local, and returns the number of bytes copied and error, if any. -// TODO: Propagate TCP RST using local.Abort(), on appropriate errors. func upload(id string, local, remote net.Conn, ioch chan<- ioinfo) { defer core.Recover(core.Exit11, "c.upload."+id) defer core.CloseOp(local, core.CopR) @@ -587,7 +629,7 @@ func upload(id string, local, remote net.Conn, ioch chan<- ioinfo) { // download copies data from local to remote, and returns the number of bytes copied and error, if any. func download(id string, local, remote net.Conn) (n int64, err error) { - defer core.CloseOp(local, core.CopW) + defer func() { resetOrClose(local, core.CopW, err) }() defer core.CloseOp(remote, core.CopR) n, err = core.Pipe(local, remote) diff --git a/intra/dialers/retrier.go b/intra/dialers/retrier.go index e5e0f464..bb293538 100644 --- a/intra/dialers/retrier.go +++ b/intra/dialers/retrier.go @@ -40,6 +40,7 @@ import ( "github.com/celzero/firestack/intra/log" "github.com/celzero/firestack/intra/protect" "github.com/celzero/firestack/intra/settings" + "golang.org/x/sys/unix" ) type zeroNetAddr struct{} @@ -67,6 +68,38 @@ const ( // TODO: with context.TODO, expmap's reaper goroutine will leak. var ippPins = core.NewSieve[netip.AddrPort, string](context.TODO(), "d.ippPins", desync_cache_ttl) +// dbgTCPInfo is a DEBUG-INSTRUMENTATION helper (temporary, remove once +// Zee5/PMTUD investigation is closed). It fetches kernel-level TCP_INFO +// stats (retransmits, rtt, last-data-received) for c, if c is backed by a +// real *net.TCPConn (best-effort; returns "" if unavailable). This lets us +// see, precisely at the moment a Read() stalls or times out, whether the +// kernel itself ever observed a retransmit from the peer during the stall +// -- distinguishing a genuine network-path blackhole (no retransmits seen, +// peer/path truly silent) from data arriving-but-undelivered (retransmits +// seen, rx_queue would be non-zero) which would point back at our code. +func dbgTCPInfo(c protect.Conn) string { + sc, ok := c.(syscall.Conn) + if !ok || sc == nil { + return "" + } + raw, err := sc.SyscallConn() + if err != nil || raw == nil { + return "" + } + var info *unix.TCPInfo + var operr error + cerr := raw.Control(func(fd uintptr) { + info, operr = unix.GetsockoptTCPInfo(int(fd), unix.IPPROTO_TCP, unix.TCP_INFO) + }) + if cerr != nil || operr != nil || info == nil { + return fmt.Sprintf("tcpinfo-err(ctl=%v,op=%v)", cerr, operr) + } + return fmt.Sprintf("tcpi[state=%d rtt=%dus rttvar=%dus retx=%d total_retx=%d last_data_recv=%dms last_data_sent=%dms unacked=%d snd_mss=%d rcv_mss=%d pmtu=%d]", + info.State, info.Rtt, info.Rttvar, info.Retransmits, info.Total_retrans, + info.Last_data_recv, info.Last_data_sent, info.Unacked, + info.Snd_mss, info.Rcv_mss, info.Pmtu) +} + // retrier implements the DuplexConn interface and must // be typecastable to *net.TCPConn (see: xdial.DialTCP) // inheritance: go.dev/play/p/mMiQgXsPM7Y @@ -308,6 +341,13 @@ func (r *retrier) dialStratLocked() (strat int32, err error) { strat = r.dialerOpts.Strat } + // DEBUG-INSTRUMENTATION (temporary, remove once Zee5/PMTUD investigation + // is closed): confirms at the dial layer exactly which split-strategy is + // actually being used per attempt, independent of what the UI/settings + // claim is configured. + log.I("retrier: dbg: %s: dialStrat: %s: strat=%d auto=%t retryStrat=%d split=%t retryCount=%d/%d", + r.dialerID(), r.raddr, strat, auto, retryStrat, split, r.retryCount, r.maxRetries) + return } @@ -476,9 +516,25 @@ func (r *retrier) Read(buf []byte) (n int, err error) { r.mu.Lock() c := r.conn // r.conn may be provisional or final connection + rdeadline := r.readDeadline r.mu.Unlock() if c != nil { + // always (re)apply the caller's current read deadline (as set via + // SetReadDeadline, eg: rwext.extendr) to the underlying conn before + // reading from it; otherwise, once the retry sequence completes + // (see below), this deadline is applied to c just once and never + // again, so subsequent idle reads on c can block indefinitely even + // as callers keep extending r.readDeadline on every call. + _ = c.SetReadDeadline(rdeadline) + + // DEBUG-INSTRUMENTATION (temporary, remove once Zee5/PMTUD + // investigation is closed): kernel TCP_INFO snapshot right before + // issuing the read, so we can diff against the post-read snapshot + // below to see exactly what changed (or didn't) at the kernel level + // during this call, especially across a timeout/stall. + preTCPInfo := dbgTCPInfo(c) + for reads := range maxEmptyReads { n, err = c.Read(buf) if n == 0 && err == nil { // no data and no error @@ -488,6 +544,19 @@ func (r *retrier) Read(buf []byte) (n int, err error) { } // else: check if retry is needed (c == nil or err != nil) break } + + // DEBUG-INSTRUMENTATION (temporary, remove once Zee5/PMTUD + // investigation is closed): post-read snapshot. If Total_retrans + // increased between pre/post while err is a timeout, the kernel + // DID see retransmits from the peer during the stall (data was + // attempted but never fully arrived/ack'd) -- a genuine network + // issue. If Total_retrans is unchanged, the peer never even tried + // to resend, consistent with a silent path blackhole (eg PMTUD) + // upstream of this device entirely. + postTCPInfo := dbgTCPInfo(c) + log.I("retrier: dbg: %s: read-tcpinfo: [%s<=%s]; pre: %s; post: %s; n=%d err=%v", + r.dialerID(), laddr(c), r.raddr, preTCPInfo, postTCPInfo, n, err) + if n == 0 && err == nil { err = io.ErrNoProgress } @@ -670,6 +739,7 @@ func (r *retrier) Write(b []byte) (int, error) { r.mu.Lock() c := r.conn // retry has completed, so r.conn is final and may not need locking? + wdeadline := r.writeDeadline r.mu.Unlock() if c == nil { cerr := log.EE("retrier: write: %s: [] => %s (b: %d, tee: %d), not retrying, but no conn; after: %s", @@ -677,6 +747,8 @@ func (r *retrier) Write(b []byte) (int, error) { return 0, core.JoinErr(cerr, errNilConn) } + // always (re)apply the caller's current write deadline; see Read() for why. + _ = c.SetWriteDeadline(wdeadline) n, err := c.Write(b) if err != nil { err = log.EE("retrier: write: %s: [%s=>%s]; b: %d/%d (retried? %t); after: %s; err? %v", @@ -746,8 +818,11 @@ func (r *retrier) WriteTo(w io.Writer) (bytes int64, err error) { } if !optimizedWriteTo { - // write to w from c until EOF - b, err = core.Stream(w, c) + // write to w from r (not raw c) until EOF, so that r.Read's + // per-call deadline refresh (see Read()) stays in effect; reading + // from c directly bypasses that refresh and can hang indefinitely + // once the retry sequence above has completed. + b, err = core.Stream(w, r) bytes += b } @@ -842,8 +917,8 @@ func (r *retrier) ReadFrom(reader io.Reader) (bytes int64, err error) { } if !optimizedReadFrom { - // read from reader into c until EOF - b, err = core.Stream(c, reader) + // read from reader into r (not raw c) until EOF; see WriteTo for why. + b, err = core.Stream(r, reader) bytes += b } diff --git a/intra/dnsx/alg.go b/intra/dnsx/alg.go index 1e66c021..cbe6b49f 100644 --- a/intra/dnsx/alg.go +++ b/intra/dnsx/alg.go @@ -1366,8 +1366,10 @@ func (t *dnsgateway) querySecondary(t2 Transport, uid, fid, network string, msg result.smm.Blocklists = blocklistnames result.smm.BlockedTarget = blockedtarget } - if xdns.AQuadAUnspecified(r) { - // A/AAAA must be 0.0.0.0/::, set UpstreamBlocks to true + if xdns.AQuadAUnspecified(r) || xdns.AQuadALoopback(r) { + // A/AAAA must be 0.0.0.0/:: (or, for some upstream resolvers, + // sinkholed to loopback 127.0.0.1/::1); either way, set + // UpstreamBlocks to true result.smm.UpstreamBlocks = true // discard all other answers result.ips = append(result.ips, anyaddr4, anyaddr6) @@ -1518,7 +1520,7 @@ func (t *dnsgateway) q(t1, t2 Transport, preset []netip.Addr, origin, network, s // for t1, ansin's already evaluated for ans0000 in querySecondary // (secans.pri is set to true). ansin may be from t2 (if t2 != nil), // ans64, which is a modified ansin, depending on settings.PtMode - ans0000 := xdns.AQuadAUnspecified(ansin) // ansin is not nil; ans64 may be nil + ans0000 := xdns.AQuadAUnspecified(ansin) || xdns.AQuadALoopback(ansin) // ansin is not nil; ans64 may be nil if ans0000 { smm.BlockedTarget = qname diff --git a/intra/dnsx/plus.go b/intra/dnsx/plus.go index 5890f0d9..11d5de09 100644 --- a/intra/dnsx/plus.go +++ b/intra/dnsx/plus.go @@ -384,7 +384,7 @@ func (t *plus) forward(network string, q *dns.Msg, outSmm *x.DNSSummary, all ... failed := xdns.IsServFailOrInvalid(ans) noans := !failed && !xdns.HasAnyAnswer(ans) - ipblock := xdns.HasAQuadAQuestion(q) && xdns.AQuadAUnspecified(ans) + ipblock := xdns.HasAQuadAQuestion(q) && (xdns.AQuadAUnspecified(ans) || xdns.AQuadALoopback(ans)) // HTTPS/SVCB blocks have 0 answer records when blocked svcbblock := (xdns.HasHTTPQuestion(q) || xdns.HasSVCBQuestion(q)) && noans diff --git a/intra/dnsx/transport.go b/intra/dnsx/transport.go index 35fcad22..f9b920e7 100644 --- a/intra/dnsx/transport.go +++ b/intra/dnsx/transport.go @@ -836,8 +836,10 @@ runagain: } // else: discard ans2 (which is always exclusively a blocked ans from rdns blocklists) realips := Netip2Csv(xdns.IPs(nonalg)) - // ans1 is upstream answer... does upstream block? - ansblocked := xdns.AQuadAUnspecified(ans1) + // ans1 is upstream answer... does upstream block? some public + // ad/tracker-blocking resolvers sinkhole to loopback (127.0.0.1/::1) + // instead of the unspecified address; treat both as upstream blocks. + ansblocked := xdns.AQuadAUnspecified(ans1) || xdns.AQuadALoopback(ans1) if log.Verbose { log.V("dns: fwd: 7 for %s[%s] (fid: %s); query %s:%d, r%d, onQueryTime: %s / onAnswerTime: %s, ips: %s; smm[data: %s, status: %d]; new-ans? %t, blocklists? %t, blocked? %t", diff --git a/intra/ipn/auto.go b/intra/ipn/auto.go index 9696c655..d81fb0fa 100644 --- a/intra/ipn/auto.go +++ b/intra/ipn/auto.go @@ -23,6 +23,21 @@ const ( ttl30s = 30 * time.Second shortdelay = 100 * time.Millisecond delayForUnhealthyProxies = 2 * time.Second + + // STREAMSHIELD HOOK: short-interval TCP keepalive tuning for Exit-proxied + // connections. Some video-CDN/load-balancer stacks (observed: Zee5's + // Akamai-fronted CDN) close idle-but-alive connections at ~120s via a + // clean FIN, and the app's HTTP client fails to transparently retry on + // the dead pooled connection (permanent hang). sockopt.go's global + // defaults (600s idle) are far too slow to matter here. These constants + // are used only when the user enables the existing "TCP keep alive" + // toggle (settings.DialerOpts.LowerKeepAlive / PersistentState.tcpKeepAlive) + // -- best-effort experiment: only helps if the peer's idle-timeout tracks + // raw TCP socket activity (keepalive probes are pure ACKs) rather than + // HTTP-level request/response activity. + exitKeepAliveIdleSec = 20 // secs of inactivity before first probe + exitKeepAliveIntervalSec = 5 // secs between probes + exitKeepAliveCount = 4 // unacked probes before conn is declared dead ) // auto is a proxy that dials multiple, preset outbounds. @@ -624,8 +639,9 @@ func maybeKeepAlive2(c net.Conn) (keepingalive, ok bool) { } if opts := settings.GetDialerOpts(); opts.LowerKeepAlive { - // adjust socket's keepalive config - lowered := core.SetKeepAliveConfigSockOpt(c) + // STREAMSHIELD HOOK: use the short-interval tuning above instead of + // sockopt.go's slow global defaults (600s idle) -- see const block. + lowered := core.SetKeepAliveConfigSockOpt(c, exitKeepAliveIdleSec, exitKeepAliveIntervalSec, exitKeepAliveCount) keepingalive = lowered ok = lowered return diff --git a/intra/ipn/proxies.go b/intra/ipn/proxies.go index 797941b4..f752fd3b 100644 --- a/intra/ipn/proxies.go +++ b/intra/ipn/proxies.go @@ -929,7 +929,26 @@ func (px *proxifier) proxyFor(id string) (Proxy, error) { // Ingress (dummy): no fast path, fall through to general lookup } - timeout := time.Duration(minWaitPeriodSec/2) * time.Second + // Regression fix: this used to be getproxytimeout (5s) and was + // inadvertently shortened to minWaitPeriodSec/2 (1s) in 8677a52c + // ("core/volatile: cr by muse spark" era commit chain). proxyFor is + // called for every proxy id, including non-wellknown, app-registered + // ids (see isWellknown/ProxyFor above) for which there is NO retry/ + // wait fallback -- ProxyFor returns immediately with errProxyNotFound + // for those ids, so this is the *only* window a caller gets to find + // a just-registered proxy. The lookup itself is a cheap RLock'd map + // read (see below), but on loaded/low-RAM devices the paired Lock() + // in AddProxy/RemoveProxy can legitimately hold the mutex for longer + // than 1s during proxy setup/teardown, especially for proxies that do + // real I/O in their constructor. Shortening this guard to 1s turns a + // rare, recoverable stall into a hard, unretried lookup failure for + // any non-wellknown proxy id registered right around this window -- + // observed in production as a permanently-failing custom local proxy + // route until the next reconnect. Restoring getproxytimeout (5s) + // keeps this a deadlock-recovery guard (its original documented + // purpose, see the ProxyFor doc-comment above) rather than a + // register-race timeout. + timeout := getproxytimeout // go.dev/play/p/xCug1W3OcMH p, completed := core.Grx("pxr.ProxyFor: "+id, func(_ context.Context) (Proxy, error) { px.RLock() diff --git a/intra/netstack/tcp.go b/intra/netstack/tcp.go index 640ddfe8..e5b58fc1 100644 --- a/intra/netstack/tcp.go +++ b/intra/netstack/tcp.go @@ -326,14 +326,26 @@ func (g *GTCPConn) RemoteAddr() net.Addr { func (g *GTCPConn) Write(data []byte) (int, error) { if c := g.conn(); c != nil { - return c.Write(data) + n, err := c.Write(data) + // DEBUG-INSTRUMENTATION (temporary, remove once Zee5/PMTUD + // investigation is closed): app-facing (netstack) side write + // visibility, to check whether the app itself ever stops writing + // (eg: waiting on a stalled read) vs our exit-side relay stalling + // independently. + log.VV("netstack: dbg: gconn(%s): write: b=%d/%d; err=%v", g.o, n, len(data), err) + return n, err } return 0, netError(g, "tcp", g.o+":write", io.ErrClosedPipe) } func (g *GTCPConn) Read(data []byte) (int, error) { if c := g.conn(); c != nil { - return c.Read(data) + n, err := c.Read(data) + // DEBUG-INSTRUMENTATION (temporary, remove once Zee5/PMTUD + // investigation is closed): app-facing (netstack) side read + // visibility, paired with the Write() log above. + log.VV("netstack: dbg: gconn(%s): read: b=%d/%d; err=%v", g.o, n, len(data), err) + return n, err } return 0, netError(g, "tcp", g.o+":read", io.ErrNoProgress) } diff --git a/intra/protect/protect.go b/intra/protect/protect.go index b6ecde06..54a6c840 100644 --- a/intra/protect/protect.go +++ b/intra/protect/protect.go @@ -36,6 +36,7 @@ import ( b "github.com/celzero/firestack/intra/backend" "github.com/celzero/firestack/intra/core" "github.com/celzero/firestack/intra/log" + "golang.org/x/sys/unix" ) // See: ipmap.LookupNetIP; Selfhost -> dnsx.Default; Systemhost -> dnsx.System @@ -55,6 +56,22 @@ const ( // if true, only protects the socket from routing loops & binds to active network. onlyProtectWildcardAddrs = false + + // clampedMSS is a conservative TCP MSS ceiling advertised (pre-connect, + // via TCP_MAXSEG) on all outbound TCP sockets. Mitigates a path-MTU + // discovery blackhole class of bug: a remote peer sends a full-MTU-sized + // segment that needs in-flight fragmentation on some hop of the path; + // the fragmentation-needed ICMP reply required for real PMTUD is + // filtered/lost (common middlebox misconfiguration), so the peer's + // retransmits of the oversized segment never make it through and the + // connection goes silent forever. Clamping our advertised MSS forces + // the remote peer to never send us a segment large enough to trigger + // this, at the cost of marginally smaller segments on high-throughput + // transfers. 1400 leaves headroom under the standard 1500-byte Ethernet + // MTU for the worst-case combined IPv4/IPv6 + TCP header overhead plus + // any additional encapsulation (eg: PPPoE, WireGuard, GRE, VPN) that may + // exist on the path. + clampedMSS = 1400 ) var MyUid = strconv.Itoa(os.Getuid()) @@ -86,6 +103,7 @@ func ifbind(who string, ctl Controller) func(string, string, syscall.RawConn) er log.VV("control: netbinder: %s: %s(%s); err? %v", who, network, addr, err) return c.Control(func(fd uintptr) { sock := int(fd) + clampMSS(who, network, sock) if onlyProtectWildcardAddrs && !maybeGlobalUnicast(addr, true) { ctl.Protect(who, sock) return @@ -104,6 +122,25 @@ func ifbind(who string, ctl Controller) func(string, string, syscall.RawConn) er } } +// clampMSS sets a conservative TCP_MAXSEG ceiling (best-effort; errors are +// logged, never fatal) on newly-created, not-yet-connected TCP sockets, to +// guard against a path-MTU-discovery blackhole (see clampedMSS doc above). +// No-op for non-TCP networks. +// +// DEBUG-INSTRUMENTATION (temporary, remove once Zee5/PMTUD investigation is +// closed): logs a getsockopt readback of TCP_MAXSEG immediately after +// setting it, since some kernels silently cap/ignore the requested value. +func clampMSS(who, network string, sock int) { + switch network { + case "tcp", "tcp4", "tcp6": + serr := unix.SetsockoptInt(sock, unix.IPPROTO_TCP, unix.TCP_MAXSEG, clampedMSS) + got, gerr := unix.GetsockoptInt(sock, unix.IPPROTO_TCP, unix.TCP_MAXSEG) + log.I("protect: dbg: %s: mss-clamp: set(%d)? err=%v; readback=%d; err=%v; %s sock", + who, clampedMSS, serr, got, gerr, network) + default: // udp, unix, etc: no-op + } +} + // unused: Binds a socket to a local ip. func ipbind(p Protector) func(string, string, syscall.RawConn) error { return func(network, addr string, c syscall.RawConn) (err error) { diff --git a/intra/rwconn.go b/intra/rwconn.go index 812471de..e40bd2f7 100644 --- a/intra/rwconn.go +++ b/intra/rwconn.go @@ -9,6 +9,7 @@ package intra import ( "io" "net" + "sync/atomic" "syscall" "time" @@ -16,11 +17,44 @@ import ( "github.com/celzero/firestack/intra/settings" ) +// A TCP conn's idle-read deadline is governed by +// settings.DialerOpts.ReadTimeoutSec (default 10s, see: +// PersistentState.kt:dialTimeoutSec). That deadline exists to catch conns +// that are PMTUD/censorship blackholed -- silently dropped with no RST/FIN -- +// which would otherwise hang forever with no timeout at all (the original +// failure mode dialTimeoutSec's 0->10 change was introduced to fix). +// +// But persistent HTTP/1.1|2 keep-alive conns routinely sit fully idle between +// a completed response and the app's next logical request, and that idle gap +// is not bounded by any protocol invariant -- it's purely a function of how +// long the app takes to decide it wants more data. Eg: Zee5's cold-cache +// DRM/token provisioning after receiving its DASH/HLS manifest response can +// legitimately, and variably, exceed even a generous fixed grace period (45s +// was tried and empirically still insufficient on some runs), causing +// Firestack to RST an already-healthy, still-in-use conn out from under the +// app before it's reused -- manifesting as a silent, permanent player hang +// (no retry, no app-visible error). Root caused via 3-state (vpn-off / +// vpn-on+warm-app-cache / vpn-on+cold-app-cache) logcat capture+diff, +// 2026-09-13; 45s-grace attempt empirically insufficient, re-tested same day. +// +// Since no finite grace period can be guaranteed sufficient, once a conn has +// successfully read >=1 byte from the remote it is proven NOT to be a +// blackhole candidate (a genuinely blackholed conn could never have done +// so) -- at that point the dopt.ReadTimeoutSec-derived deadline is dropped +// entirely, falling back to just the conn-type's own floor (rwext.minidle; +// 0 for tcp => extendr/extend treat that as "no deadline", see common.go). +// A real mid-stream death of a warm conn is still caught: either the peer +// eventually sends a real RST/FIN (no artificial timer needed), or the OS +// TCP stack's own keepalive eventually detects the dead link. Never-yet- +// successful (cold) conns are unaffected and keep the short, aggressive base +// deadline (settings.DialerOpts.ReadTimeoutSec). + // rwext wraps MinConn and extends deadline to minimum(min, settings.DialerOpts) // on every read and write. type rwext struct { - net.Conn // underlying conn - minidle uint32 // min idle timeout in secs + net.Conn // underlying conn + minidle uint32 // min idle timeout in secs + warm *atomic.Bool // set once a read succeeds on this conn; never nil } // TODO? var _ core.DuplexCloser = (*rwext)(nil) @@ -51,7 +85,11 @@ func (rw rwext) Unwrap() net.Conn { func (rw rwext) Read(b []byte) (n int, err error) { rw.extendr() - return rw.Conn.Read(b) + n, err = rw.Conn.Read(b) + if n > 0 && rw.warm != nil { + rw.warm.Store(true) // conn proven alive; grant a longer idle grace hereon + } + return } func (rw rwext) Write(b []byte) (n int, err error) { @@ -62,25 +100,28 @@ func (rw rwext) Write(b []byte) (n int, err error) { // ReadFrom implements core.RetrierConn. func (rw rwext) ReadFrom(r io.Reader) (n int64, err error) { switch c := rw.Unwrap().(type) { - case io.ReaderFrom: - // disable read and write deadlines for rw.Conn as - // io.ReaderFrom does not support io.Reader+io.Writer - // semantics which rwext relies on to extend deadlines. + case *net.TCPConn: + // disable read and write deadlines for rw.Conn as io.ReaderFrom + // (splice/sendfile) does not support io.Reader+io.Writer semantics + // which rwext relies on to extend deadlines; safe only for a true + // os-level conn where the syscall itself is a single operation, not + // an unbounded higher-level relay loop that can go idle forever. rw.extendForever() return c.ReadFrom(r) default: } - // nb: stream rw (which extends deadlines) not rw.Conn + // nb: stream rw (which extends deadlines) not rw.Conn; this also covers + // wrapper types (eg: *dialers.retrier) that implement io.ReaderFrom but + // may internally fall back to a plain, idle-able copy loop of their own + // -- such types must not have their deadlines disabled outright. return core.Stream(rw, r) } // WriteTo implements core.RetrierConn. func (rw rwext) WriteTo(w io.Writer) (n int64, err error) { switch c := rw.Unwrap().(type) { - case io.WriterTo: - // disable read and write deadlines for rw.Conn as - // io.WriterTo does not support io.Reader+io.Writer - // semantics which rwext relies on to extend deadlines. + case *net.TCPConn: + // see ReadFrom for why this is scoped to a genuine os-level conn. rw.extendForever() return c.WriteTo(w) default: @@ -99,9 +140,22 @@ func (rw rwext) SyscallConn() (syscall.RawConn, error) { func (rw rwext) deadlines() (r, w uint32) { dopt := settings.GetDialerOpts() - // -ve ints go higher than 2^31 w/ uint: go.dev/play/p/Rrqk_V8a7W0 - return max(rw.minidle, uint32(dopt.ReadTimeoutSec)), - max(rw.minidle, uint32(dopt.WriteTimeoutSec)) + if rw.warm != nil && rw.warm.Load() { + // already exchanged >=1 byte on this conn: not a blackhole candidate; + // no finite idle-read grace is guaranteed sufficient (see rationale + // above), so drop the dopt.ReadTimeoutSec-derived deadline entirely + // and fall back to just rw.minidle (the conn-type's own floor, if + // any: 0 for tcp => extendr/extend see <=0 and call + // SetDeadline(zero-value), ie: no deadline at all, see common.go; + // 120s for udp => unchanged from before, still bounded so idle udp + // "conns" -- which have no real end-of-stream signal like tcp's + // RST/FIN -- don't leak NAT/socket state forever). + r = rw.minidle + } else { + // -ve ints go higher than 2^31 w/ uint: go.dev/play/p/Rrqk_V8a7W0 + r = max(rw.minidle, uint32(dopt.ReadTimeoutSec)) + } + return r, max(rw.minidle, uint32(dopt.WriteTimeoutSec)) } func (rw rwext) extendForever() { diff --git a/intra/tcp.go b/intra/tcp.go index 4fcb86d6..8d8e2f99 100644 --- a/intra/tcp.go +++ b/intra/tcp.go @@ -31,6 +31,7 @@ import ( "net" "net/netip" "sync" + "sync/atomic" "time" "github.com/celzero/firestack/intra/core" @@ -193,7 +194,7 @@ func (h *tcpHandler) ReverseProxy(gconn *netstack.GTCPConn, in net.Conn, to, fro } core.Go("tcp.reverse:"+cid, func() { - h.forward(gconn, rwext{in, tcptimeout}, smm) + h.forward(gconn, rwext{Conn: in, minidle: tcptimeout, warm: new(atomic.Bool)}, smm) }) return true } @@ -483,7 +484,7 @@ func (h *tcpHandler) handle(px ipn.Proxy, gconn *netstack.GTCPConn, src, target core.Go("tcp.forward."+smm.ID, func() { defer h.loopUnassoc(smm) h.flowing(smm) - h.forward(gconn, rwext{dst, tcptimeout}, smm) // src always *gonet.TCPConn + h.forward(gconn, rwext{Conn: dst, minidle: tcptimeout, warm: new(atomic.Bool)}, smm) // src always *gonet.TCPConn // TODO: assoc if forward was successful if eim { h.natAssoc(smm.PID, src, dstlocal) diff --git a/intra/udp.go b/intra/udp.go index 59c4391a..081b566c 100644 --- a/intra/udp.go +++ b/intra/udp.go @@ -30,6 +30,7 @@ import ( "errors" "net" "net/netip" + "sync/atomic" "time" "github.com/celzero/firestack/intra/dnsx" @@ -124,7 +125,7 @@ func (h *udpHandler) ReverseProxy(gconn *netstack.GUDPConn, in net.Conn, to, fro } core.Go("udp.reverse:"+cid, func() { - h.forward(gconn, rwext{in, udptimeout}, smm) + h.forward(gconn, rwext{Conn: in, minidle: udptimeout, warm: new(atomic.Bool)}, smm) }) return true } @@ -201,7 +202,7 @@ func (h *udpHandler) proxy(gconn *netstack.GUDPConn, src, dst netip.AddrPort, dm core.Go("udp.forward."+cid, func() { defer h.loopUnassoc(smm) h.flowing(smm) - h.forward(gconn, rwext{remote, udptimeout}, smm) + h.forward(gconn, rwext{Conn: remote, minidle: udptimeout, warm: new(atomic.Bool)}, smm) }) return true // ok } diff --git a/intra/x64/dns64.go b/intra/x64/dns64.go index 67d23f55..2fedf097 100644 --- a/intra/x64/dns64.go +++ b/intra/x64/dns64.go @@ -262,7 +262,7 @@ func (d *dns64) eval(network string, force64 bool, ansin *dns.Msg, r, uid string // hasaaaq(true) hasans(true) rgood(true) ans0000(false) hasq6 := xdns.HasAAAAQuestion(ansin) hasans6 := xdns.HasAAAAAnswer(ansin) - ans00006 := xdns.AQuadAUnspecified(ansin) + ans00006 := xdns.AQuadAUnspecified(ansin) || xdns.AQuadALoopback(ansin) hasauth := xdns.IsDNSSECAnswerAuthenticated(ansin) // treat as if v6 answer missing if enforcing 6to4 if !hasq6 || ((hasauth || hasans6) && !force64) || ans00006 { @@ -289,7 +289,7 @@ func (d *dns64) eval(network string, force64 bool, ansin *dns.Msg, r, uid string ans4, err := d.query64(network, ansin, r, uid) rgood := xdns.HasRcodeSuccess(ans4) hasans := xdns.HasAnyAnswer(ans4) - ans0000 := xdns.AQuadAUnspecified(ans4) + ans0000 := xdns.AQuadAUnspecified(ans4) || xdns.AQuadALoopback(ans4) if err != nil || ans4 == nil || !hasans || ans0000 { log.W("dns64: skip: for %s, query(n:%s / a? %t) on resolver(%s[%s]/%s), code(good? %t / blocked? %t), err(%v)", uid, qname, hasans, r, id, network, rgood, ans0000, err) diff --git a/intra/xdns/dnsutil.go b/intra/xdns/dnsutil.go index e669ba96..7c0d9cb6 100644 --- a/intra/xdns/dnsutil.go +++ b/intra/xdns/dnsutil.go @@ -1433,6 +1433,36 @@ func AQuadAUnspecified(msg *dns.Msg) bool { return false } +// AQuadALoopback returns true if any A/AAAA record in msg resolves to a +// loopback address (127.0.0.0/8 or ::1). Some upstream resolvers (public +// ad/tracker-blocking DNS providers) signal a block by sinkholing the +// answer to loopback instead of the more common 0.0.0.0/:: unspecified +// address that AQuadAUnspecified detects. Without this check, such +// upstream-blocked answers are silently treated as valid, real answers: +// the resulting connection attempt never reaches the VPN's tun interface +// (loopback-destined traffic is routed by the OS outside any VPN tunnel), +// so it is invisible to on-device firewall/connection logs, and is never +// flagged as blocked even though nothing will ever answer on it. +func AQuadALoopback(msg *dns.Msg) bool { + if msg == nil { + return false + } + ans := msg.Answer + for _, rr := range ans { + switch v := rr.(type) { + case *dns.AAAA: + if v.AAAA.IsLoopback() { + return true + } + case *dns.A: + if v.A.IsLoopback() { + return true + } + } + } + return false +} + func Len(msg *dns.Msg) int { if msg == nil { return 0