From 8be9f37656734d126871007821e4c3809772c9ae Mon Sep 17 00:00:00 2001 From: Vinicius Fortuna Date: Tue, 15 Sep 2026 19:02:11 -0400 Subject: [PATCH] refactor(intra): migrate UDP path from PacketProxy to packetrelay network.PacketProxy and NewPacketProxyFromPacketListener are deprecated in favor of the flow-based network/packetrelay API. This migrates Intra's UDP handling, mirroring OutlineFoundation/outline-apps#2777. - intraPacketProxy becomes intraPacketRelay, a packetrelay.PacketRelay whose associations share a dohAssociation between their sender and receiver halves. - lwip2transport.ConfigureDevice -> ConfigureDeviceWithRelay. - packetrelay.NewPacketRelayFromPacketListener replaces network.NewPacketProxyFromPacketListener, keeping the same 5 minute write-idle timeout (now a constructor argument). - Add unit tests for DNS interception, packet relaying, byte accounting and sender close semantics. Behavior is unchanged, except that SendPacket on a closed association now returns ErrClosed instead of issuing a DoH query whose response can no longer be delivered. --- Android/app/src/go/intra/packet_proxy.go | 168 ------------ Android/app/src/go/intra/packet_relay.go | 243 ++++++++++++++++++ Android/app/src/go/intra/packet_relay_test.go | 215 ++++++++++++++++ Android/app/src/go/intra/tunnel.go | 10 +- 4 files changed, 463 insertions(+), 173 deletions(-) delete mode 100644 Android/app/src/go/intra/packet_proxy.go create mode 100644 Android/app/src/go/intra/packet_relay.go create mode 100644 Android/app/src/go/intra/packet_relay_test.go diff --git a/Android/app/src/go/intra/packet_proxy.go b/Android/app/src/go/intra/packet_proxy.go deleted file mode 100644 index e55e10ac..00000000 --- a/Android/app/src/go/intra/packet_proxy.go +++ /dev/null @@ -1,168 +0,0 @@ -// Copyright 2023 Jigsaw Operations LLC -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package intra - -import ( - "context" - "errors" - "fmt" - "net" - "net/netip" - "sync/atomic" - "time" - - "localhost/Intra/Android/app/src/go/doh" - "localhost/Intra/Android/app/src/go/intra/protect" - - "golang.getoutline.org/sdk/network" - "golang.getoutline.org/sdk/transport" -) - -type intraPacketProxy struct { - fakeDNSAddr netip.AddrPort - dns atomic.Pointer[doh.Resolver] - proxy network.PacketProxy - listener UDPListener - ctx context.Context -} - -var _ network.PacketProxy = (*intraPacketProxy)(nil) - -func newIntraPacketProxy( - ctx context.Context, fakeDNS netip.AddrPort, dns doh.Resolver, protector protect.Protector, listener UDPListener, -) (*intraPacketProxy, error) { - if dns == nil { - return nil, errors.New("dns is required") - } - - pl := &transport.UDPListener{ - ListenConfig: *protect.MakeListenConfig(protector), - } - - // RFC 4787 REQ-5 requires a timeout no shorter than 5 minutes. - pp, err := network.NewPacketProxyFromPacketListener(pl, network.WithPacketListenerWriteIdleTimeout(5*time.Minute)) - if err != nil { - return nil, fmt.Errorf("failed to create packet proxy from listener: %w", err) - } - - dohpp := &intraPacketProxy{ - fakeDNSAddr: fakeDNS, - proxy: pp, - listener: listener, - ctx: ctx, - } - dohpp.dns.Store(&dns) - - return dohpp, nil -} - -// NewSession implements PacketProxy.NewSession. -func (p *intraPacketProxy) NewSession(resp network.PacketResponseReceiver) (network.PacketRequestSender, error) { - dohResp := &dohPacketRespReceiver{ - PacketResponseReceiver: resp, - stats: makeTracker(), - listener: p.listener, - } - req, err := p.proxy.NewSession(dohResp) - if err != nil { - return nil, fmt.Errorf("failed to create new session: %w", err) - } - - return &dohPacketReqSender{ - PacketRequestSender: req, - proxy: p, - response: dohResp, - stats: dohResp.stats, - }, nil -} - -func (p *intraPacketProxy) SetDNS(dns doh.Resolver) error { - if dns == nil { - return errors.New("dns is required") - } - p.dns.Store(&dns) - return nil -} - -// DoH PacketRequestSender wrapper -type dohPacketReqSender struct { - network.PacketRequestSender - - response *dohPacketRespReceiver - proxy *intraPacketProxy - stats *tracker -} - -// DoH PacketResponseReceiver wrapper -type dohPacketRespReceiver struct { - network.PacketResponseReceiver - - stats *tracker - listener UDPListener -} - -var _ network.PacketRequestSender = (*dohPacketReqSender)(nil) -var _ network.PacketResponseReceiver = (*dohPacketRespReceiver)(nil) - -// WriteTo implements PacketRequestSender.WriteTo. It will query the DoH server if the packet a DNS packet. -func (req *dohPacketReqSender) WriteTo(p []byte, destination netip.AddrPort) (int, error) { - if isEquivalentAddrPort(destination, req.proxy.fakeDNSAddr) { - defer func() { - // conn was only used for this DNS query, so it's unlikely to be used again - if req.stats.download.Load() == 0 && req.stats.upload.Load() == 0 { - req.Close() - } - }() - - resp, err := (*req.proxy.dns.Load()).Query(req.proxy.ctx, p) - if err != nil { - return 0, fmt.Errorf("DoH request error: %w", err) - } - if len(resp) == 0 { - return 0, errors.New("empty DoH response") - } - - return req.response.writeFrom(resp, net.UDPAddrFromAddrPort(req.proxy.fakeDNSAddr), false) - } - - req.stats.upload.Add(int64(len(p))) - return req.PacketRequestSender.WriteTo(p, destination) -} - -// Close terminates the UDP session, and reports session stats to the listener. -func (resp *dohPacketRespReceiver) Close() error { - if resp.listener != nil { - resp.listener.OnUDPSocketClosed(&UDPSocketSummary{ - Duration: int32(time.Since(resp.stats.start)), - UploadBytes: resp.stats.upload.Load(), - DownloadBytes: resp.stats.download.Load(), - }) - } - return resp.PacketResponseReceiver.Close() -} - -// WriteFrom implements PacketResponseReceiver.WriteFrom. -func (resp *dohPacketRespReceiver) WriteFrom(p []byte, source net.Addr) (int, error) { - return resp.writeFrom(p, source, true) -} - -// writeFrom writes to the underlying PacketResponseReceiver. -// It will also add len(p) to downloadBytes if doStat is true. -func (resp *dohPacketRespReceiver) writeFrom(p []byte, source net.Addr, doStat bool) (int, error) { - if doStat { - resp.stats.download.Add(int64(len(p))) - } - return resp.PacketResponseReceiver.WriteFrom(p, source) -} diff --git a/Android/app/src/go/intra/packet_relay.go b/Android/app/src/go/intra/packet_relay.go new file mode 100644 index 00000000..d53d3d3a --- /dev/null +++ b/Android/app/src/go/intra/packet_relay.go @@ -0,0 +1,243 @@ +// Copyright 2023 Jigsaw Operations LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package intra + +import ( + "context" + "errors" + "fmt" + "net/netip" + "sync" + "sync/atomic" + "time" + + "localhost/Intra/Android/app/src/go/doh" + "localhost/Intra/Android/app/src/go/intra/protect" + + "golang.getoutline.org/sdk/network/packetrelay" + "golang.getoutline.org/sdk/transport" +) + +// intraPacketRelay is a [packetrelay.PacketRelay] that answers DNS queries addressed to the +// fake DNS server with the configured DoH resolver, and relays all other packets to their +// destination. +type intraPacketRelay struct { + fakeDNSAddr netip.AddrPort + dns atomic.Pointer[doh.Resolver] + base packetrelay.PacketRelay + listener UDPListener + ctx context.Context +} + +var _ packetrelay.PacketRelay = (*intraPacketRelay)(nil) + +func newIntraPacketRelay( + ctx context.Context, fakeDNS netip.AddrPort, dns doh.Resolver, protector protect.Protector, listener UDPListener, +) (*intraPacketRelay, error) { + if dns == nil { + return nil, errors.New("dns is required") + } + + pl := &transport.UDPListener{ + ListenConfig: *protect.MakeListenConfig(protector), + } + + // RFC 4787 REQ-5 requires a timeout no shorter than 5 minutes. The timeout is reset by + // outgoing packets only, matching the previous behavior. + base, err := packetrelay.NewPacketRelayFromPacketListener(pl, 5*time.Minute) + if err != nil { + return nil, fmt.Errorf("failed to create packet relay from listener: %w", err) + } + + r := &intraPacketRelay{ + fakeDNSAddr: fakeDNS, + base: base, + listener: listener, + ctx: ctx, + } + r.dns.Store(&dns) + + return r, nil +} + +// NewAssociation implements [packetrelay.PacketRelay].NewAssociation. +func (r *intraPacketRelay) NewAssociation() (packetrelay.PacketSender, packetrelay.PacketReceiver, error) { + sender, receiver, err := r.base.NewAssociation() + if err != nil { + return nil, nil, fmt.Errorf("failed to create new association: %w", err) + } + + a := &dohAssociation{ + relay: r, + sender: sender, + receiver: receiver, + stats: makeTracker(), + handlerReady: make(chan struct{}), + closed: make(chan struct{}), + } + return &dohPacketSender{a}, &dohPacketReceiver{a}, nil +} + +func (r *intraPacketRelay) SetDNS(dns doh.Resolver) error { + if dns == nil { + return errors.New("dns is required") + } + r.dns.Store(&dns) + return nil +} + +// dohAssociation is the shared state of a UDP association. Its sender half intercepts DNS +// queries, and its receiver half reports the association stats once it concludes. +type dohAssociation struct { + relay *intraPacketRelay + sender packetrelay.PacketSender + receiver packetrelay.PacketReceiver + stats *tracker + + // handlerReady is closed once handler is set by ReceivePackets. handler must not be read + // before then. + handlerReady chan struct{} + handler packetrelay.PacketHandler + handlerSet atomic.Bool + + mu sync.Mutex // Protects isClosed. + isClosed bool + closed chan struct{} // Closed when the association is closed. + + reportOnce sync.Once +} + +// close terminates the association. Closing the underlying sender also unblocks +// ReceivePackets, which performs the remaining cleanup. +func (a *dohAssociation) close() error { + a.mu.Lock() + if a.isClosed { + a.mu.Unlock() + return packetrelay.ErrClosed + } + a.isClosed = true + a.mu.Unlock() + + close(a.closed) + return a.sender.Close() +} + +// report notifies the listener that the association has concluded. +func (a *dohAssociation) report() { + a.reportOnce.Do(func() { + if a.relay.listener == nil { + return + } + a.relay.listener.OnUDPSocketClosed(&UDPSocketSummary{ + Duration: int32(time.Since(a.stats.start)), + UploadBytes: a.stats.upload.Load(), + DownloadBytes: a.stats.download.Load(), + }) + }) +} + +// handleDNSQuery resolves a DNS query with the DoH resolver and delivers the response to the +// network stack, as if it came from the fake DNS server. +func (a *dohAssociation) handleDNSQuery(p []byte) error { + defer func() { + // The association was only used for this DNS query, so it's unlikely to be used again. + if a.stats.download.Load() == 0 && a.stats.upload.Load() == 0 { + a.close() + } + }() + + resp, err := (*a.relay.dns.Load()).Query(a.relay.ctx, p) + if err != nil { + return fmt.Errorf("DoH request error: %w", err) + } + if len(resp) == 0 { + return errors.New("empty DoH response") + } + + // DNS responses are not counted in the association stats. + select { + case <-a.handlerReady: + return a.handler.HandlePacket(resp, a.relay.fakeDNSAddr) + case <-a.closed: + return packetrelay.ErrClosed + } +} + +// dohPacketSender is the sender half of a [dohAssociation]. +type dohPacketSender struct { + a *dohAssociation +} + +var _ packetrelay.PacketSender = (*dohPacketSender)(nil) + +// SendPacket implements [packetrelay.PacketSender].SendPacket. It queries the DoH server if +// the packet is a DNS packet. +func (s *dohPacketSender) SendPacket(p []byte, destination netip.AddrPort) error { + select { + case <-s.a.closed: + return packetrelay.ErrClosed + default: + } + + if isEquivalentAddrPort(destination, s.a.relay.fakeDNSAddr) { + return s.a.handleDNSQuery(p) + } + + s.a.stats.upload.Add(int64(len(p))) + return s.a.sender.SendPacket(p, destination) +} + +// Close implements [packetrelay.PacketSender].Close. +func (s *dohPacketSender) Close() error { + return s.a.close() +} + +// dohPacketReceiver is the receiver half of a [dohAssociation]. +type dohPacketReceiver struct { + a *dohAssociation +} + +var _ packetrelay.PacketReceiver = (*dohPacketReceiver)(nil) + +// ReceivePackets implements [packetrelay.PacketReceiver].ReceivePackets. It returns when the +// association is closed, and reports the association stats to the listener. +func (r *dohPacketReceiver) ReceivePackets(handler packetrelay.PacketHandler) error { + a := r.a + if handler == nil { + return errors.New("handler is required") + } + if !a.handlerSet.CompareAndSwap(false, true) { + return errors.New("ReceivePackets called multiple times") + } + a.handler = handler + close(a.handlerReady) + + defer a.report() + return a.receiver.ReceivePackets(&statsPacketHandler{stats: a.stats, handler: handler}) +} + +// statsPacketHandler counts the bytes received by the association before handing the packet +// over to the network stack. +type statsPacketHandler struct { + stats *tracker + handler packetrelay.PacketHandler +} + +var _ packetrelay.PacketHandler = (*statsPacketHandler)(nil) + +func (h *statsPacketHandler) HandlePacket(p []byte, source netip.AddrPort) error { + h.stats.download.Add(int64(len(p))) + return h.handler.HandlePacket(p, source) +} diff --git a/Android/app/src/go/intra/packet_relay_test.go b/Android/app/src/go/intra/packet_relay_test.go new file mode 100644 index 00000000..cfeb415b --- /dev/null +++ b/Android/app/src/go/intra/packet_relay_test.go @@ -0,0 +1,215 @@ +// Copyright 2026 Jigsaw Operations LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package intra + +import ( + "context" + "errors" + "net" + "net/netip" + "testing" + "time" + + "golang.getoutline.org/sdk/network/packetrelay" + + "github.com/stretchr/testify/require" +) + +const testTimeout = 10 * time.Second + +var testFakeDNSAddr = netip.MustParseAddrPort("10.111.222.3:53") + +// capturedPacket is a packet delivered to the network stack. +type capturedPacket struct { + payload []byte + source netip.AddrPort +} + +// captureHandler is a [packetrelay.PacketHandler] that records the packets it receives. +type captureHandler struct { + packets chan capturedPacket +} + +var _ packetrelay.PacketHandler = (*captureHandler)(nil) + +func newCaptureHandler() *captureHandler { + return &captureHandler{packets: make(chan capturedPacket, 16)} +} + +func (h *captureHandler) HandlePacket(p []byte, source netip.AddrPort) error { + // p must not be retained, so make a copy. + h.packets <- capturedPacket{payload: append([]byte(nil), p...), source: source} + return nil +} + +func (h *captureHandler) next(t *testing.T) capturedPacket { + t.Helper() + select { + case p := <-h.packets: + return p + case <-time.After(testTimeout): + t.Fatal("timed out waiting for a packet") + return capturedPacket{} + } +} + +// udpListenerFunc adapts a function to the [UDPListener] interface. +type udpListenerFunc func(*UDPSocketSummary) + +func (f udpListenerFunc) OnUDPSocketClosed(s *UDPSocketSummary) { f(s) } + +// startAssociation creates a relay and one association, running the receive loop in the +// background. It returns the association halves, the captured packets, and the summary +// reported when the association concludes. +func startAssociation(t *testing.T, query qfunc) ( + packetrelay.PacketSender, *captureHandler, <-chan *UDPSocketSummary, <-chan error, +) { + t.Helper() + + summaries := make(chan *UDPSocketSummary, 1) + relay, err := newIntraPacketRelay( + context.Background(), testFakeDNSAddr, newFakeTransport(query), nil, + udpListenerFunc(func(s *UDPSocketSummary) { summaries <- s }), + ) + require.NoError(t, err) + + sender, receiver, err := relay.NewAssociation() + require.NoError(t, err) + t.Cleanup(func() { sender.Close() }) + + handler := newCaptureHandler() + received := make(chan error, 1) + go func() { received <- receiver.ReceivePackets(handler) }() + + return sender, handler, summaries, received +} + +// TestAssociationAnswersDNSLocally verifies that queries to the fake DNS server are answered +// by the resolver, that the response appears to come from the fake DNS server, and that the +// DNS-only association is torn down immediately afterwards without reporting any traffic. +func TestAssociationAnswersDNSLocally(t *testing.T) { + response := []byte("fake-dns-response") + queries := make(chan []byte, 1) + sender, handler, summaries, received := startAssociation(t, func(q []byte) ([]byte, error) { + queries <- q + return response, nil + }) + + require.NoError(t, sender.SendPacket([]byte("dns-query"), testFakeDNSAddr)) + require.Equal(t, []byte("dns-query"), <-queries) + + packet := handler.next(t) + require.Equal(t, response, packet.payload) + require.Equal(t, testFakeDNSAddr, packet.source) + + // The association was only used for this DNS query, so it is closed right away. + select { + case <-received: + case <-time.After(testTimeout): + t.Fatal("the DNS-only association was not closed") + } + + summary := <-summaries + require.Zero(t, summary.UploadBytes, "DNS queries must not be counted") + require.Zero(t, summary.DownloadBytes, "DNS responses must not be counted") +} + +// TestAssociationReportsDNSErrors verifies that resolution failures are reported to the caller. +func TestAssociationReportsDNSErrors(t *testing.T) { + t.Run("query error", func(t *testing.T) { + sender, _, _, _ := startAssociation(t, func([]byte) ([]byte, error) { + return nil, errors.New("no service") + }) + require.Error(t, sender.SendPacket([]byte("dns-query"), testFakeDNSAddr)) + }) + + t.Run("empty response", func(t *testing.T) { + sender, _, _, _ := startAssociation(t, func([]byte) ([]byte, error) { + return nil, nil + }) + require.Error(t, sender.SendPacket([]byte("dns-query"), testFakeDNSAddr)) + }) +} + +// TestAssociationRelaysPackets verifies that non-DNS packets reach their destination, that +// responses are delivered back to the network stack, and that both directions are counted. +func TestAssociationRelaysPackets(t *testing.T) { + echo := startEchoServer(t) + sender, handler, summaries, _ := startAssociation(t, failingQuery(t)) + + require.NoError(t, sender.SendPacket([]byte("ping"), echo)) + + packet := handler.next(t) + require.Equal(t, []byte("ping"), packet.payload) + // The source may be reported as a 4-in-6 address. + require.True(t, isEquivalentAddrPort(packet.source, echo), "got %v, want %v", packet.source, echo) + + require.NoError(t, sender.Close()) + summary := <-summaries + require.EqualValues(t, 4, summary.UploadBytes) + require.EqualValues(t, 4, summary.DownloadBytes) +} + +// TestAssociationCloseIsIdempotent verifies the [packetrelay.PacketSender] close semantics. +func TestAssociationCloseIsIdempotent(t *testing.T) { + echo := startEchoServer(t) + sender, _, _, received := startAssociation(t, failingQuery(t)) + + require.NoError(t, sender.Close()) + require.ErrorIs(t, sender.Close(), packetrelay.ErrClosed) + require.ErrorIs(t, sender.SendPacket([]byte("ping"), echo), packetrelay.ErrClosed) + // A closed association must not issue DNS queries either. + require.ErrorIs(t, sender.SendPacket([]byte("dns-query"), testFakeDNSAddr), packetrelay.ErrClosed) + + // Closing the sender must terminate ReceivePackets. + select { + case <-received: + case <-time.After(testTimeout): + t.Fatal("ReceivePackets did not return after Close") + } +} + +func failingQuery(t *testing.T) qfunc { + return func([]byte) ([]byte, error) { + t.Error("unexpected DNS query") + return nil, errors.New("unexpected DNS query") + } +} + +// startEchoServer runs a UDP server that echoes back everything it receives. +func startEchoServer(t *testing.T) netip.AddrPort { + t.Helper() + + conn, err := net.ListenPacket("udp", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { conn.Close() }) + + go func() { + buf := make([]byte, 2048) + for { + n, addr, err := conn.ReadFrom(buf) + if err != nil { + return + } + if _, err := conn.WriteTo(buf[:n], addr); err != nil { + return + } + } + }() + + addr, err := netip.ParseAddrPort(conn.LocalAddr().String()) + require.NoError(t, err) + return addr +} diff --git a/Android/app/src/go/intra/tunnel.go b/Android/app/src/go/intra/tunnel.go index 25677171..ab4ed624 100644 --- a/Android/app/src/go/intra/tunnel.go +++ b/Android/app/src/go/intra/tunnel.go @@ -44,7 +44,7 @@ type Tunnel struct { ctx context.Context cancel context.CancelFunc sd *intraStreamDialer - pp *intraPacketProxy + pr *intraPacketRelay sni *tcpSNIReporter tun io.Closer } @@ -87,12 +87,12 @@ func NewTunnel( return nil, fmt.Errorf("failed to create stream dialer: %w", err) } - t.pp, err = newIntraPacketProxy(t.ctx, fakeDNSAddr.AddrPort(), dohdns, protector, eventListener) + t.pr, err = newIntraPacketRelay(t.ctx, fakeDNSAddr.AddrPort(), dohdns, protector, eventListener) if err != nil { - return nil, fmt.Errorf("failed to create packet proxy: %w", err) + return nil, fmt.Errorf("failed to create packet relay: %w", err) } - if t.IPDevice, err = lwip2transport.ConfigureDevice(t.sd, t.pp); err != nil { + if t.IPDevice, err = lwip2transport.ConfigureDeviceWithRelay(t.sd, t.pr); err != nil { return nil, fmt.Errorf("failed to configure lwIP stack: %w", err) } @@ -105,7 +105,7 @@ func NewTunnel( // must not be nil. func (t *Tunnel) SetDNS(dns doh.Resolver) { t.sd.SetDNS(dns) - t.pp.SetDNS(dns) + t.pr.SetDNS(dns) t.sni.SetDNS(dns) }