MegaProxy is an open-source Android VPN client for reliable and secure connections through proxy servers you control or trust.
-It contains no advertising, analytics SDKs, tracking identifiers, or remote telemetry. Connection statistics, profiles, and diagnostic logs remain on the device unless you explicitly choose to share them.
+It contains no advertising, analytics SDKs, tracking identifiers, or remote telemetry. Connection statistics and diagnostic logs remain on the device unless you explicitly choose to share them. Profiles are stored locally; connection credentials are used to authenticate to the configured servers.
Features include:
MegaProxy does not provide a proxy service. You need an HTTPS, SSH, or SSH-with-jump server that you operate or trust.
+MegaProxy does not provide a proxy service. You need HTTPS or SSH servers that you operate or trust; both transports support a jump server.
Only TCP application traffic is forwarded. General UDP and QUIC forwarding are not supported.
diff --git a/fastlane/metadata/android/ru-RU/full_description.txt b/fastlane/metadata/android/ru-RU/full_description.txt index f7c939d..e489177 100644 --- a/fastlane/metadata/android/ru-RU/full_description.txt +++ b/fastlane/metadata/android/ru-RU/full_description.txt @@ -1,12 +1,13 @@MegaProxy — открытый VPN-клиент для Android, предназначенный для надёжных и безопасных подключений через прокси-серверы, которыми вы управляете или которым доверяете.
-В приложении нет рекламы, аналитических SDK, идентификаторов отслеживания и удалённой телеметрии. Статистика соединения, профили и диагностические журналы остаются на устройстве, пока вы сами не решите ими поделиться.
+В приложении нет рекламы, аналитических SDK, идентификаторов отслеживания и удалённой телеметрии. Статистика соединения и диагностические журналы остаются на устройстве, пока вы сами не решите ими поделиться. Профили хранятся локально; данные авторизации используются для подключения к настроенным серверам.
Возможности:
MegaProxy не предоставляет прокси-сервис. Для работы нужен HTTPS-, SSH- или SSH-with-jump-сервер, которым вы управляете или которому доверяете.
+MegaProxy не предоставляет прокси-сервис. Для работы нужны HTTPS- или SSH-серверы, которыми вы управляете или которым доверяете; оба транспорта поддерживают промежуточный сервер.
Перенаправляется только TCP-трафик приложений. Произвольный UDP- и QUIC-трафик не поддерживается.
diff --git a/native/README.md b/native/README.md index 9767634..872d0e6 100644 --- a/native/README.md +++ b/native/README.md @@ -8,15 +8,24 @@ verification, authentication and HTTP/2 sessions. Only the jump is dialed direct the destination proxy hostname. Application traffic counters exclude the intermediate tunnel. SSH with Jump creates a nested SSH client through the jump session. SSH transports support TCP; DNS is carried over DoH, while arbitrary UDP (including QUIC) is intentionally blocked. -`Start` always takes ownership of the passed duplicate TUN descriptor, including error paths. +`Start` borrows the JVM-owned TUN descriptor for the call and duplicates it with CLOEXEC on +entry. Go closes only its own duplicate; Java retains responsibility for the descriptor it passed. +The bridge passes the Android TUN MTU explicitly (currently 1400). Generated gomobile types are +compile-time JVM dependencies, so signature changes must compile on both sides. -Build prerequisites: Go 1.26+, Android SDK/NDK and `gomobile`. +Use the supported Fastlane commands from the repository root: ```shell -go install golang.org/x/mobile/cmd/gomobile@latest -gomobile init -gomobile bind -target=android -androidapi 26 -o ../app/libs/megaproxy.aar ./mobile +bundle exec fastlane android native_tests +bundle exec fastlane android native_fuzz +bundle exec fastlane android debug_artifact ``` -Run `go test ./...` before producing the AAR. The dependency versions are pinned in `go.mod`; -commit the generated `go.sum` after the first successful dependency download. +`native_tests` includes the Go race detector; `native_fuzz` runs a bounded 20-second parser campaign. +Android build lanes prepare `app/libs/megaproxy.aar` through `scripts/build-fdroid-native.sh`, which +installs the pinned gomobile/gobind version and uses the reproducible binding flags. Do not replace +that build path with `gomobile@latest`. Go dependencies are pinned in `go.mod` and `go.sum`. + +Requirements and environment setup are in the root [README](../README.md#building-from-source) +and the [Fastlane reference](../docs/en/fastlane.md). Native tests use local servers and synthetic +file descriptors; they do not certify real Android TUN/JNI lifecycle behavior. diff --git a/native/mobile/bootstrap.go b/native/mobile/bootstrap.go index f3d8ec1..5e3263b 100644 --- a/native/mobile/bootstrap.go +++ b/native/mobile/bootstrap.go @@ -10,6 +10,7 @@ import ( "io" "net" "net/http" + "strings" "syscall" "time" ) @@ -33,7 +34,14 @@ var bootstrapResolvers = []bootstrapResolver{ // ResolveProxy bootstraps the proxy address through protected encrypted DNS. func ResolveProxy(host string, protector Protector, reporter Reporter) (string, error) { - query, id, err := buildAQuery(host) + host = strings.TrimSpace(host) + if ip := net.ParseIP(host); ip != nil { + return ip.String(), nil + } + if protector == nil { + return "", errors.New("Android socket protector is required") + } + query, id, err := buildAQuery(strings.TrimSuffix(host, ".")) if err != nil { return "", err } @@ -102,6 +110,9 @@ func resolveProxyWithResolver(query []byte, id uint16, resolver bootstrapResolve } func buildAQuery(host string) ([]byte, uint16, error) { + if len(host) == 0 || len(host) > 253 { + return nil, 0, errors.New("invalid proxy hostname length") + } var idBytes [2]byte if _, err := rand.Read(idBytes[:]); err != nil { return nil, 0, err @@ -163,7 +174,7 @@ func parseAResponse(message []byte, id uint16) (string, error) { func skipDNSName(message []byte, offset int) (int, error) { for { - if offset >= len(message) { + if offset < 0 || offset >= len(message) { return 0, io.ErrUnexpectedEOF } length := int(message[offset]) @@ -172,7 +183,7 @@ func skipDNSName(message []byte, offset int) (int, error) { return offset, nil } if length&0xc0 == 0xc0 { - if offset >= len(message) { + if offset < 0 || offset >= len(message) { return 0, io.ErrUnexpectedEOF } return offset + 1, nil diff --git a/native/mobile/bootstrap_test.go b/native/mobile/bootstrap_test.go index 3fb44ad..78b5deb 100644 --- a/native/mobile/bootstrap_test.go +++ b/native/mobile/bootstrap_test.go @@ -33,3 +33,12 @@ func TestBootstrapIncludesYandexRedundancy(t *testing.T) { t.Fatalf("Yandex resolver count = %d, want 2", count) } } + +func TestResolveProxyLiteralDoesNotRequireDNS(t *testing.T) { + for _, host := range []string{"203.0.113.7", " 203.0.113.7 "} { + ip, err := ResolveProxy(host, nil, nil) + if err != nil || ip != "203.0.113.7" { + t.Fatalf("literal: %q %v", ip, err) + } + } +} diff --git a/native/mobile/bridge_test.go b/native/mobile/bridge_test.go new file mode 100644 index 0000000..4459e1e --- /dev/null +++ b/native/mobile/bridge_test.go @@ -0,0 +1,98 @@ +package mobile + +import ( + "os" + "strings" + "testing" + "time" + + "github.com/xjasonlyu/tun2socks/v2/proxy/reject" + "github.com/xjasonlyu/tun2socks/v2/tunnel" + "golang.org/x/sys/unix" +) + +func TestBridgeOwnsOnlyDuplicatedDescriptor(t *testing.T) { + original, err := os.CreateTemp(t.TempDir(), "tun") + if err != nil { + t.Fatal(err) + } + defer original.Close() + fd, err := duplicateTunFD(int(original.Fd())) + if err != nil { + t.Fatal(err) + } + if fd == int(original.Fd()) { + t.Fatal("descriptor was not duplicated") + } + flags, err := unix.FcntlInt(uintptr(fd), unix.F_GETFD, 0) + if err != nil { + t.Fatal(err) + } + if flags&unix.FD_CLOEXEC == 0 { + t.Error("duplicate must not survive exec") + } + if err := unix.Close(fd); err != nil { + t.Fatal(err) + } + if _, err := original.WriteString("still owned by caller"); err != nil { + t.Fatal(err) + } + for i := 0; i < 20; i++ { + if err := Start(int(original.Fd()), 1400, "invalid json", nil, nil); err == nil { + t.Fatal("invalid config accepted") + } + if _, err := original.Stat(); err != nil { + t.Fatalf("Start closed borrowed descriptor: %v", err) + } + } + if _, err := duplicateTunFD(-1); err == nil { + t.Fatal("negative descriptor accepted") + } +} + +type blockingBridgeCloser struct{ entered, release chan struct{} } + +func (c *blockingBridgeCloser) Close() error { close(c.entered); <-c.release; return nil } + +func TestBridgeStartCannotOverlapStopCleanup(t *testing.T) { + closer := &blockingBridgeCloser{make(chan struct{}), make(chan struct{})} + state.Lock() + state.running = true + state.proxyCloser = closer + tunnel.T().SetProxy(&httpsConnectDialer{}) + state.Unlock() + done := make(chan struct{}) + go func() { Stop(); close(done) }() + defer func() { close(closer.release); <-done }() + select { + case <-closer.entered: + case <-time.After(time.Second): + t.Fatal("Stop did not close upstream") + } + original, err := os.CreateTemp(t.TempDir(), "tun") + if err != nil { + t.Fatal(err) + } + defer original.Close() + raw := `{"host":"proxy.example","dialHost":"192.0.2.1","port":443,"username":"u","password":"p","profile":"CHROME_ANDROID","dohUrl":"https://dns.google/dns-query"}` + if _, err := parseConfig(raw); err != nil { + t.Fatal(err) + } + err = Start(int(original.Fd()), 1400, raw, &jumpTestProtector{}, nil) + if err == nil || !strings.Contains(err.Error(), "already running") { + t.Fatalf("Start during Stop: %v", err) + } + if _, ok := tunnel.T().Proxy().(*reject.Reject); !ok { + t.Fatal("Stop retained the global dialer and its JVM callbacks") + } + Stop() // A concurrent second Stop must not clear the first Stop's guard. + state.Lock() + stopping := state.stopping + state.Unlock() + if !stopping { + t.Fatal("second Stop cleared cleanup guard") + } + if _, err := original.Stat(); err != nil { + t.Fatalf("rejected Start closed borrowed FD: %v", err) + } +} diff --git a/native/mobile/config.go b/native/mobile/config.go index c333fc2..c2d8a41 100644 --- a/native/mobile/config.go +++ b/native/mobile/config.go @@ -48,9 +48,20 @@ type config struct { func parseConfig(raw string) (config, error) { var c config + if len(raw) > 1024*1024 { + return c, errors.New("native config exceeds 1 MiB") + } if err := json.Unmarshal([]byte(raw), &c); err != nil { return c, fmt.Errorf("decode config: %w", err) } + if len(c.DoHFallbackURLs) > 16 { + return c, errors.New("too many fallback DoH providers") + } + if c.SSHKeepaliveSeconds < 0 || c.SSHKeepaliveSeconds > 3600 || + c.SSHRotationMinutes < 0 || c.SSHRotationMinutes > 1440 || + c.SSHRotationMB < 0 || c.SSHRotationMB > 10240 { + return c, errors.New("invalid SSH keepalive or rotation limit") + } c.Host = strings.TrimSpace(c.Host) c.DialHost = strings.TrimSpace(c.DialHost) if c.Type == "" { diff --git a/native/mobile/diagnostics.go b/native/mobile/diagnostics.go index 06fe3f7..5dc5cd1 100644 --- a/native/mobile/diagnostics.go +++ b/native/mobile/diagnostics.go @@ -25,7 +25,8 @@ func errorClass(err error) string { if errors.Is(err, io.EOF) { return "eof" } - if netErr, ok := err.(net.Error); ok && netErr.Timeout() { + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { return "timeout" } message := strings.ToLower(err.Error()) diff --git a/native/mobile/dialer.go b/native/mobile/dialer.go index 79e96b9..9612b55 100644 --- a/native/mobile/dialer.go +++ b/native/mobile/dialer.go @@ -37,11 +37,13 @@ type httpsConnectDialer struct { connections chan struct{} h2Session *http2ConnectSession h2Disabled bool + closed bool } func (d *httpsConnectDialer) Close() error { d.cacheMu.Lock() defer d.cacheMu.Unlock() + d.closed = true if d.jump != nil { _ = d.jump.Close() } @@ -77,6 +79,12 @@ func (d *httpsConnectDialer) DialContext(ctx context.Context, metadata *M.Metada } func (d *httpsConnectDialer) connectTarget(ctx context.Context, target string) (net.Conn, error) { + d.cacheMu.Lock() + closed := d.closed + d.cacheMu.Unlock() + if closed { + return nil, net.ErrClosed + } connectionID := nextDiagnosticConnectionID() totalStarted := time.Now() if !d.config.AllowIPv6 { @@ -111,6 +119,10 @@ func (d *httpsConnectDialer) connectTarget(ctx context.Context, target string) ( report(d.reporter, "event=http2_session result=unsupported action=fallback_http1") return d.connectTarget(ctx, target) } + // Rejection and cancellation belong to one stream, not the shared session. + if ctx.Err() != nil || session.canTakeRequest() { + return nil, err + } d.invalidateHTTP2Session(session) report(d.reporter, "event=http2_session result=stale action=reconnect reason=%s", errorClass(err)) } @@ -180,6 +192,9 @@ func (d *httpsConnectDialer) connectTarget(ctx context.Context, target string) ( } closeOnError = false // The HTTP/2 session now owns the outer TLS connection. session = d.installHTTP2Session(session) + if session == nil { + return nil, net.ErrClosed + } connection, connectErr := d.openHTTP2Tunnel(ctx, session, target, connectionID, totalStarted, false) if shouldFallbackToHTTP1(connectErr) { d.disableHTTP2(session) @@ -241,6 +256,10 @@ func (d *httpsConnectDialer) dialProxy(ctx context.Context) (net.Conn, error) { return d.protectedDialer().DialContext(ctx, "tcp", d.config.address()) } d.cacheMu.Lock() + if d.closed { + d.cacheMu.Unlock() + return nil, net.ErrClosed + } if d.jump == nil { c := d.config c.Type = "HTTPS" @@ -311,6 +330,10 @@ func (d *httpsConnectDialer) currentHTTP2Session() *http2ConnectSession { func (d *httpsConnectDialer) installHTTP2Session(candidate *http2ConnectSession) *http2ConnectSession { d.cacheMu.Lock() defer d.cacheMu.Unlock() + if d.closed { + _ = candidate.close() + return nil + } if d.h2Session != nil && d.h2Session.canTakeRequest() { _ = candidate.close() return d.h2Session @@ -432,6 +455,10 @@ func (d *httpsConnectDialer) DialUDP(metadata *M.Metadata) (net.PacketConn, erro return nil, errUDPBlocked } d.cacheMu.Lock() + if d.closed { + d.cacheMu.Unlock() + return nil, net.ErrClosed + } if d.dohClient == nil { d.dohClient = newDoHHTTPClient(d.connectTarget) } diff --git a/native/mobile/doh.go b/native/mobile/doh.go index b642b9c..1105de3 100644 --- a/native/mobile/doh.go +++ b/native/mobile/doh.go @@ -50,8 +50,8 @@ type dohPacketConn struct { inFlight chan struct{} closed chan struct{} closeOnce sync.Once - deadlineMu sync.Mutex - readDeadline time.Time + readDeadline packetDeadline + writeDeadline packetDeadline client *http.Client context context.Context cancel context.CancelFunc @@ -95,9 +95,15 @@ func newDoHPacketConnWithClient(c config, reporter Reporter, connect func(contex } func (c *dohPacketConn) WriteTo(payload []byte, addr net.Addr) (int, error) { + if len(payload) < 12 || len(payload) > 65535 { + return 0, errors.New("invalid DNS packet size") + } + deadline := c.writeDeadline.wait() select { case <-c.closed: return 0, net.ErrClosed + case <-deadline: + return 0, c.deadlineError() default: } if !c.config.AllowIPv6 { @@ -111,14 +117,17 @@ func (c *dohPacketConn) WriteTo(payload []byte, addr net.Addr) (int, error) { return len(payload), nil } } - query := append([]byte(nil), payload...) select { case c.inFlight <- struct{}{}: + case <-deadline: + return 0, c.deadlineError() case <-c.closed: return 0, net.ErrClosed case <-c.context.Done(): return 0, c.context.Err() } + // Allocate only after admission, so waiting writers cannot retain extra packet copies. + query := append([]byte(nil), payload...) go func() { defer func() { <-c.inFlight }() var lastErr error @@ -215,22 +224,20 @@ func (c *dohPacketConn) deliver(reply dnsReply) { select { case c.replies <- reply: case <-c.closed: + default: + // UDP may drop packets. A stalled reader must not consume every shared query slot. + report(c.reporter, "event=doh result=dropped reason=reply_queue_full") } } func (c *dohPacketConn) ReadFrom(buffer []byte) (int, net.Addr, error) { - c.deadlineMu.Lock() - deadline := c.readDeadline - c.deadlineMu.Unlock() - var timer <-chan time.Time - if !deadline.IsZero() { - duration := time.Until(deadline) - if duration <= 0 { - return 0, nil, timeoutError{} - } - t := time.NewTimer(duration) - defer t.Stop() - timer = t.C + deadline := c.readDeadline.wait() + select { + case <-deadline: + return 0, nil, c.deadlineError() + case <-c.closed: + return 0, nil, net.ErrClosed + default: } select { case reply := <-c.replies: @@ -241,29 +248,57 @@ func (c *dohPacketConn) ReadFrom(buffer []byte) (int, net.Addr, error) { return 0, reply.addr, io.ErrShortBuffer } return copy(buffer, reply.payload), reply.addr, nil - case <-timer: - return 0, nil, timeoutError{} + case <-deadline: + return 0, nil, c.deadlineError() case <-c.closed: return 0, nil, net.ErrClosed } } +// Close also wakes deadline waiters; report closure instead of a random timeout. +func (c *dohPacketConn) deadlineError() error { + select { + case <-c.closed: + return net.ErrClosed + default: + return timeoutError{} + } +} + func (c *dohPacketConn) Close() error { c.closeOnce.Do(func() { close(c.closed) c.cancel() + c.readDeadline.stop() + c.writeDeadline.stop() }) return nil } func (c *dohPacketConn) LocalAddr() net.Addr { return dnsAddr("megaproxy-doh") } func (c *dohPacketConn) SetDeadline(t time.Time) error { - c.deadlineMu.Lock() - c.readDeadline = t - c.deadlineMu.Unlock() + if err := c.SetReadDeadline(t); err != nil { + return err + } + return c.SetWriteDeadline(t) +} +func (c *dohPacketConn) SetReadDeadline(t time.Time) error { + select { + case <-c.closed: + return net.ErrClosed + default: + } + c.readDeadline.set(t) + return nil +} +func (c *dohPacketConn) SetWriteDeadline(t time.Time) error { + select { + case <-c.closed: + return net.ErrClosed + default: + } + c.writeDeadline.set(t) return nil } -func (c *dohPacketConn) SetReadDeadline(t time.Time) error { return c.SetDeadline(t) } -func (c *dohPacketConn) SetWriteDeadline(time.Time) error { return nil } type dnsAddr string @@ -272,6 +307,6 @@ func (a dnsAddr) String() string { return string(a) } type timeoutError struct{} -func (timeoutError) Error() string { return "DNS read deadline exceeded" } +func (timeoutError) Error() string { return "DNS operation deadline exceeded" } func (timeoutError) Timeout() bool { return true } func (timeoutError) Temporary() bool { return true } diff --git a/native/mobile/http2_connect.go b/native/mobile/http2_connect.go index 2cff3c5..8d2da83 100644 --- a/native/mobile/http2_connect.go +++ b/native/mobile/http2_connect.go @@ -27,8 +27,9 @@ type http2ConnectSession struct { func newHTTP2ConnectSession(raw net.Conn) (*http2ConnectSession, error) { transport := &http2.Transport{ - ReadIdleTimeout: 45 * time.Second, - PingTimeout: 10 * time.Second, + ReadIdleTimeout: 45 * time.Second, + MaxHeaderListSize: 64 * 1024, + PingTimeout: 10 * time.Second, } client, err := transport.NewClientConn(raw) if err != nil { @@ -72,7 +73,7 @@ func (s *http2ConnectSession) openTunnel(ctx context.Context, target, authorizat response *http.Response err error } - resultChannel := make(chan result, 1) + resultChannel := make(chan result) go func() { response, err := s.client.RoundTrip(request) select { @@ -112,16 +113,19 @@ func (s *http2ConnectSession) openTunnel(ctx context.Context, target, authorizat // http2StreamConn exposes one HTTP/2 CONNECT stream as a net.Conn. A deadline // closes only this stream, never the shared outer TLS connection. type http2StreamConn struct { - raw net.Conn - reader io.ReadCloser - writer *io.PipeWriter - cancel context.CancelFunc - closeOnce sync.Once - deadlineMu sync.Mutex - readTimer *time.Timer - writeTimer *time.Timer - readExpired bool - writeExpired bool + raw net.Conn + reader io.ReadCloser + writer *io.PipeWriter + cancel context.CancelFunc + closeOnce sync.Once + deadlineMu sync.Mutex + readTimer *time.Timer + writeTimer *time.Timer + readGeneration uint64 + writeGeneration uint64 + closed bool + readExpired bool + writeExpired bool } func newHTTP2StreamConn(raw net.Conn, reader io.ReadCloser, writer *io.PipeWriter, cancel context.CancelFunc) *http2StreamConn { @@ -154,6 +158,7 @@ func (c *http2StreamConn) Close() error { var closeErr error c.closeOnce.Do(func() { c.deadlineMu.Lock() + c.closed = true if c.readTimer != nil { c.readTimer.Stop() } @@ -190,6 +195,11 @@ func (c *http2StreamConn) SetWriteDeadline(deadline time.Time) error { func (c *http2StreamConn) setReadDeadline(deadline time.Time) { c.deadlineMu.Lock() defer c.deadlineMu.Unlock() + if c.closed { + return + } + c.readGeneration++ + generation := c.readGeneration c.readExpired = false if c.readTimer != nil { c.readTimer.Stop() @@ -197,10 +207,7 @@ func (c *http2StreamConn) setReadDeadline(deadline time.Time) { } if !deadline.IsZero() { c.readTimer = time.AfterFunc(time.Until(deadline), func() { - c.deadlineMu.Lock() - c.readExpired = true - c.deadlineMu.Unlock() - _ = c.Close() + c.expireDeadline(true, generation) }) } } @@ -208,6 +215,11 @@ func (c *http2StreamConn) setReadDeadline(deadline time.Time) { func (c *http2StreamConn) setWriteDeadline(deadline time.Time) { c.deadlineMu.Lock() defer c.deadlineMu.Unlock() + if c.closed { + return + } + c.writeGeneration++ + generation := c.writeGeneration c.writeExpired = false if c.writeTimer != nil { c.writeTimer.Stop() @@ -215,10 +227,28 @@ func (c *http2StreamConn) setWriteDeadline(deadline time.Time) { } if !deadline.IsZero() { c.writeTimer = time.AfterFunc(time.Until(deadline), func() { - c.deadlineMu.Lock() - c.writeExpired = true - c.deadlineMu.Unlock() - _ = c.Close() + c.expireDeadline(false, generation) }) } } + +func (c *http2StreamConn) expireDeadline(read bool, generation uint64) { + c.deadlineMu.Lock() + current := c.writeGeneration + if read { + current = c.readGeneration + } + if c.closed || generation != current { + c.deadlineMu.Unlock() + return + } + if read { + c.readExpired = true + } else { + c.writeExpired = true + } + // Commit expiration under the same lock as SetDeadline, before closing I/O. + c.closed = true + c.deadlineMu.Unlock() + _ = c.Close() +} diff --git a/native/mobile/http2_connect_test.go b/native/mobile/http2_connect_test.go index fde231c..5610916 100644 --- a/native/mobile/http2_connect_test.go +++ b/native/mobile/http2_connect_test.go @@ -3,11 +3,13 @@ package mobile import ( "context" "crypto/tls" + "errors" "fmt" "io" "net" "net/http" "net/http/httptest" + "strings" "sync" "testing" "time" @@ -123,3 +125,62 @@ func isTimeout(err error) bool { value, ok := err.(interface{ Timeout() bool }) return ok && value.Timeout() } + +type rejectedHTTP2Client struct{ closed bool } + +func (c *rejectedHTTP2Client) CanTakeNewRequest() bool { return !c.closed } +func (c *rejectedHTTP2Client) Close() error { c.closed = true; return nil } +func (c *rejectedHTTP2Client) RoundTrip(r *http.Request) (*http.Response, error) { + _ = r.Body.Close() + return &http.Response{StatusCode: http.StatusBadGateway, Body: io.NopCloser(strings.NewReader(""))}, nil +} +func TestHTTP2RejectedTargetPreservesSession(t *testing.T) { + client := &rejectedHTTP2Client{} + session := &http2ConnectSession{client: client} + d := &httpsConnectDialer{h2Session: session} + _, err := d.connectTarget(context.Background(), "example.com:443") + var rejected *http2ConnectStatusError + if !errors.As(err, &rejected) || rejected.status != http.StatusBadGateway { + t.Fatalf("error = %v", err) + } + if client.closed || d.currentHTTP2Session() != session { + t.Fatal("target failure closed shared session") + } +} + +func TestSupersededHTTP2DeadlineCannotCloseStream(t *testing.T) { + left, right := net.Pipe() + defer left.Close() + defer right.Close() + reader, writer := io.Pipe() + stream := newHTTP2StreamConn(left, reader, writer, func() {}) + defer stream.Close() + _ = stream.SetReadDeadline(time.Now().Add(time.Hour)) + old := stream.readGeneration + _ = stream.SetReadDeadline(time.Time{}) + stream.expireDeadline(true, old) // Timer callback already queued before Stop. + if stream.closed || stream.readExpired { + t.Fatal("old read timer expired a cleared deadline") + } + _ = stream.SetWriteDeadline(time.Now().Add(time.Hour)) + old = stream.writeGeneration + _ = stream.SetWriteDeadline(time.Time{}) + stream.expireDeadline(false, old) + if stream.closed || stream.writeExpired { + t.Fatal("old write timer expired a cleared deadline") + } + _ = stream.Close() + _ = stream.SetReadDeadline(time.Now().Add(time.Hour)) + if stream.readTimer != nil && !stream.readTimer.Stop() { + t.Fatal("closed stream installed an active timer") + } +} + +func TestClosedDialerRejectsLateHTTP2Session(t *testing.T) { + d := &httpsConnectDialer{} + _ = d.Close() + client := &rejectedHTTP2Client{} + if session := d.installHTTP2Session(&http2ConnectSession{client: client}); session != nil || !client.closed { + t.Fatal("late handshake resurrected a closed dialer") + } +} diff --git a/native/mobile/mobile.go b/native/mobile/mobile.go index 959cf82..74a3339 100644 --- a/native/mobile/mobile.go +++ b/native/mobile/mobile.go @@ -10,9 +10,12 @@ import ( "syscall" "time" + "golang.org/x/sys/unix" + "github.com/xjasonlyu/tun2socks/v2/core" "github.com/xjasonlyu/tun2socks/v2/core/device" "github.com/xjasonlyu/tun2socks/v2/core/device/fdbased" + "github.com/xjasonlyu/tun2socks/v2/proxy/reject" "github.com/xjasonlyu/tun2socks/v2/tunnel" "gvisor.dev/gvisor/pkg/tcpip/stack" ) @@ -22,13 +25,19 @@ var state struct { generation uint64 starting bool running bool + stopping bool device device.Device stack *stack.Stack proxyCloser io.Closer } -// Start takes ownership of a duplicate of tunFD held by Android's ParcelFileDescriptor. -func Start(tunFD int, rawConfig string, protector Protector, reporter Reporter) error { +// Start borrows tunFD for this call. Android keeps its ParcelFileDescriptor open; +// Go owns only the duplicate made here, including every failure path. +func Start(tunFD int, mtu int, rawConfig string, protector Protector, reporter Reporter) error { + tunFD, err := duplicateTunFD(tunFD) + if err != nil { + return err + } c, err := parseConfig(rawConfig) if err != nil { if tunFD >= 0 { @@ -36,14 +45,14 @@ func Start(tunFD int, rawConfig string, protector Protector, reporter Reporter) } return err } - if tunFD < 0 || protector == nil { + if protector == nil || mtu < 1280 || mtu > 65535 { if tunFD >= 0 { _ = syscall.Close(tunFD) } return errors.New("invalid Android VPN bridge") } state.Lock() - if state.running || state.starting { + if state.running || state.starting || state.stopping { state.Unlock() _ = syscall.Close(tunFD) return errors.New("proxy core is already running") @@ -58,11 +67,13 @@ func Start(tunFD int, rawConfig string, protector Protector, reporter Reporter) return } state.Lock() + // The global tunnel must not retain a failed dialer and its Java callbacks. + tunnel.T().SetProxy(&reject.Reject{}) state.starting = false state.Unlock() }() resetStats() - dev, err := fdbased.Open(strconv.Itoa(tunFD), 1500, 0) + dev, err := fdbased.Open(strconv.Itoa(tunFD), uint32(mtu), 0) if err != nil { _ = syscall.Close(tunFD) return err @@ -89,11 +100,15 @@ func Start(tunFD int, rawConfig string, protector Protector, reporter Reporter) netstack, err := core.CreateStack(&core.Config{LinkEndpoint: dev, TransportHandler: t}) if err != nil { dev.Close() + if proxyCloser != nil { + _ = proxyCloser.Close() + } return err } state.Lock() if state.generation != generation || !state.starting { state.Unlock() + dev.Close() netstack.Close() netstack.Wait() if proxyCloser != nil { @@ -109,13 +124,37 @@ func Start(tunFD int, rawConfig string, protector Protector, reporter Reporter) return nil } +func duplicateTunFD(fd int) (int, error) { + if fd < 0 { + return -1, errors.New("invalid Android TUN descriptor") + } + return unix.FcntlInt(uintptr(fd), unix.F_DUPFD_CLOEXEC, 0) +} + func Stop() { state.Lock() state.generation++ + if state.stopping { + state.Unlock() + return + } + state.stopping = true + // Drop the global strong reference to the dialer, config and JVM service callbacks. + // Late queued packets must fail closed rather than use a replacement connection. + tunnel.T().SetProxy(&reject.Reject{}) state.running = false dev, netstack, proxyCloser := state.device, state.stack, state.proxyCloser state.device, state.stack, state.proxyCloser = nil, nil, nil state.Unlock() + defer func() { + state.Lock() + state.stopping = false + state.Unlock() + }() + // Wake blocked upstream operations before waiting for the stack to finish. + if proxyCloser != nil { + _ = proxyCloser.Close() + } if dev != nil { dev.Close() } @@ -123,7 +162,4 @@ func Stop() { netstack.Close() netstack.Wait() } - if proxyCloser != nil { - _ = proxyCloser.Close() - } } diff --git a/native/mobile/packet_deadline.go b/native/mobile/packet_deadline.go new file mode 100644 index 0000000..597068c --- /dev/null +++ b/native/mobile/packet_deadline.go @@ -0,0 +1,82 @@ +package mobile + +import ( + "sync" + "time" +) + +// A deadline channel is shared by already-blocked and future packet operations. +// Resetting a timer cannot allow its stale callback to expire a newer deadline. +type packetDeadline struct { + mu sync.Mutex + timer *time.Timer + generation uint64 + expired bool + closed bool + signal chan struct{} +} + +func (d *packetDeadline) wait() <-chan struct{} { + d.mu.Lock() + defer d.mu.Unlock() + if d.signal == nil { + d.signal = make(chan struct{}) + } + return d.signal +} + +func (d *packetDeadline) set(t time.Time) { + d.mu.Lock() + defer d.mu.Unlock() + if d.closed { + return + } + d.generation++ + generation := d.generation + if d.timer != nil { + d.timer.Stop() + d.timer = nil + } + if d.signal == nil || d.expired { + d.signal = make(chan struct{}) + } + d.expired = false + if t.IsZero() { + return + } + if !t.After(time.Now()) { + d.expired = true + close(d.signal) + return + } + d.timer = time.AfterFunc(time.Until(t), func() { + d.mu.Lock() + defer d.mu.Unlock() + if d.generation != generation || d.expired { + return + } + d.expired = true + close(d.signal) + }) +} + +func (d *packetDeadline) stop() { + d.mu.Lock() + defer d.mu.Unlock() + if d.closed { + return + } + d.closed = true + d.generation++ + if d.timer != nil { + d.timer.Stop() + d.timer = nil + } + if d.signal == nil { + d.signal = make(chan struct{}) + } + if !d.expired { + close(d.signal) + d.expired = true + } +} diff --git a/native/mobile/robustness_test.go b/native/mobile/robustness_test.go new file mode 100644 index 0000000..a6138e6 --- /dev/null +++ b/native/mobile/robustness_test.go @@ -0,0 +1,178 @@ +package mobile + +import ( + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "net" + "strings" + "testing" + "time" +) + +const validNativeConfig = `{"host":"proxy.example","dialHost":"192.0.2.1","port":443,"username":"u","password":"p","profile":"CHROME_ANDROID","dohUrl":"https://dns.google/dns-query"}` + +func TestNativeConfigLimits(t *testing.T) { + for _, field := range []string{"sshKeepaliveSeconds", "sshRotationMinutes", "sshRotationMb"} { + for _, value := range []int64{-1, 9223372036854775807} { + var fields map[string]any + if err := json.Unmarshal([]byte(validNativeConfig), &fields); err != nil { + t.Fatal(err) + } + fields[field] = value + raw, err := json.Marshal(fields) + if err != nil { + t.Fatal(err) + } + if _, err := parseConfig(string(raw)); err == nil { + t.Fatalf("accepted %s=%d", field, value) + } + } + } + raw := strings.TrimSuffix(validNativeConfig, "}") + `,"unused":"` + strings.Repeat("x", 1024*1024) + `"}` + if _, err := parseConfig(raw); err == nil { + t.Fatal("accepted oversized JSON") + } + if _, err := TestConnection(validNativeConfig, nil, nil); err == nil || !strings.Contains(err.Error(), "protector") { + t.Fatalf("missing protector: %v", err) + } + if _, _, err := buildAQuery(strings.Repeat("a.", 200)); err == nil { + t.Fatal("accepted oversized DNS name") + } +} + +func TestDoHReplyQueueCannotBlockWriters(t *testing.T) { + c := newDoHPacketConnWithClient(config{}, nil, nil, "https://dns.example/query", nil, nil) + defer c.Close() + query, _, err := buildAQuery("example.com") + if err != nil { + t.Fatal(err) + } + binary.BigEndian.PutUint16(query[len(query)-4:], 28) + done := make(chan error, 1) + go func() { + for i := 0; i < 100; i++ { + if _, err := c.WriteTo(query, dnsAddr("dns")); err != nil { + done <- err + return + } + } + done <- nil + }() + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(2 * time.Second): + t.Fatal("full reply queue blocked DNS writers") + } + if len(c.replies) != cap(c.replies) { + t.Fatal("reply queue not bounded at capacity") + } + if _, err := c.WriteTo(make([]byte, 65536), dnsAddr("dns")); err == nil { + t.Fatal("accepted oversized DNS packet") + } +} + +func TestDoHDeadlinesWakePendingOperations(t *testing.T) { + c := newDoHPacketConnWithClient(config{AllowIPv6: true}, nil, nil, "https://dns.example/query", nil, make(chan struct{}, 1)) + defer c.Close() + read := make(chan error, 1) + go func() { _, _, err := c.ReadFrom(make([]byte, 512)); read <- err }() + if err := c.SetReadDeadline(time.Now().Add(20 * time.Millisecond)); err != nil { + t.Fatal(err) + } + select { + case err := <-read: + if !isTimeout(err) { + t.Fatalf("read: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("pending read ignored updated deadline") + } + if err := c.SetReadDeadline(time.Time{}); err != nil { + t.Fatal(err) + } + c.deliver(dnsReply{payload: []byte{1}, addr: dnsAddr("dns")}) + if n, _, err := c.ReadFrom(make([]byte, 512)); err != nil || n != 1 { + t.Fatalf("cleared deadline: %d %v", n, err) + } + c.inFlight <- struct{}{} // Hold every provider slot, without making a network request. + query, _, err := buildAQuery("example.com") + if err != nil { + t.Fatal(err) + } + write := make(chan error, 1) + go func() { _, err := c.WriteTo(query, dnsAddr("dns")); write <- err }() + if err := c.SetWriteDeadline(time.Now().Add(20 * time.Millisecond)); err != nil { + t.Fatal(err) + } + select { + case err := <-write: + if !isTimeout(err) { + t.Fatalf("write: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("pending write ignored deadline") + } +} + +func TestClosedPacketDeadlineCannotRetainNewTimers(t *testing.T) { + var d packetDeadline + d.stop() + d.set(time.Now().Add(time.Hour)) + if d.timer != nil { + t.Fatal("closed deadline installed a timer") + } + select { + case <-d.wait(): + default: + t.Fatal("closed deadline did not signal") + } +} + +func FuzzNativeParsers(f *testing.F) { + f.Add([]byte(validNativeConfig)) + f.Add([]byte("771,4865,0,29,0")) + f.Add([]byte{0, 1, 128, 0, 0, 1, 0, 0, 0, 0, 0, 0}) + f.Fuzz(func(t *testing.T, data []byte) { + if len(data) > 1024*1024 { + t.Skip() + } + _, _ = parseConfig(string(data)) + _, _ = parseJA3(string(data)) + _, _, _ = buildAQuery(string(data)) + _, _ = parseAResponse(data, 1) + _, _, _ = emptyAAAAResponse(data) + _, _ = skipDNSName(data, -1) + _, _ = skipDNSName(data, 0) + }) +} + +// Keep this fixture compile-checked against the production packet interface. +var _ net.PacketConn = (*dohPacketConn)(nil) + +func TestWrappedTimeoutClassification(t *testing.T) { + if got := errorClass(fmt.Errorf("dial proxy: %w", timeoutError{})); got != "timeout" { + t.Fatalf("wrapped timeout classified as %s", got) + } +} + +func TestClosedDoHDoesNotRandomlyReportDeadlineExceeded(t *testing.T) { + c := newDoHPacketConnWithClient(config{}, nil, nil, "https://dns.example/query", nil, nil) + _ = c.Close() + query, _, err := buildAQuery("example.com") + if err != nil { + t.Fatal(err) + } + for i := 0; i < 100; i++ { + if _, _, err := c.ReadFrom(make([]byte, 512)); !errors.Is(err, net.ErrClosed) { + t.Fatalf("read after close: %v", err) + } + if _, err := c.WriteTo(query, dnsAddr("dns")); !errors.Is(err, net.ErrClosed) { + t.Fatalf("write after close: %v", err) + } + } +} diff --git a/native/mobile/ssh_dialer.go b/native/mobile/ssh_dialer.go index c37451b..748e8fc 100644 --- a/native/mobile/ssh_dialer.go +++ b/native/mobile/ssh_dialer.go @@ -25,6 +25,7 @@ type sshDialer struct { reporter Reporter mu sync.Mutex client *ssh.Client + closed bool jumpClient *ssh.Client channels chan struct{} sessionCreated time.Time @@ -39,6 +40,12 @@ func (d *sshDialer) DialContext(ctx context.Context, metadata *M.Metadata) (net. } func (d *sshDialer) connectTarget(ctx context.Context, target string) (net.Conn, error) { + d.mu.Lock() + closed := d.closed + d.mu.Unlock() + if closed { + return nil, net.ErrClosed + } if !d.config.AllowIPv6 { host, _, _ := net.SplitHostPort(target) if ip := net.ParseIP(host); ip != nil && ip.To4() == nil { @@ -76,10 +83,19 @@ func (d *sshDialer) connectTarget(ctx context.Context, target string) (net.Conn, } started := time.Now() dialContext, cancelDial := context.WithTimeout(ctx, 20*time.Second) - conn, err := client.DialContext(dialContext, "tcp", target) + conn, err, abandoned := dialSSHChannel(dialContext, client, target, func() { <-channels }, func() { + report(d.reporter, "event=ssh_session result=stalled reason=abandoned_channel_open") + d.invalidateClient(client) + }, 30*time.Second) + if abandoned { + release = false + } // The blocked worker keeps its slot until it actually ends. cancelDial() if err != nil { - d.invalidate() + var rejected *ssh.OpenChannelError + if !errors.As(err, &rejected) && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) { + d.invalidateClient(client) + } report(d.reporter, "event=connection mode=ssh stage=direct_tcpip result=failed reason=%s", errorClass(err)) recordConnectionOutcome(false) return nil, fmt.Errorf("SSH direct-tcpip: %w", err) @@ -94,6 +110,9 @@ func (d *sshDialer) connectTarget(ctx context.Context, target string) (net.Conn, func (d *sshDialer) session(ctx context.Context) (*ssh.Client, error) { d.mu.Lock() defer d.mu.Unlock() + if d.closed { + return nil, net.ErrClosed + } if d.client != nil { return d.client, nil } @@ -376,9 +395,15 @@ func (d *sshDialer) sharedDoHResources() (*http.Client, chan struct{}) { return d.dohClient, d.dohInFlight } -func (d *sshDialer) invalidate() { +func (d *sshDialer) invalidate() { d.invalidateClient(nil) } + +func (d *sshDialer) invalidateClient(expected *ssh.Client) { d.mu.Lock() defer d.mu.Unlock() + // An error from an old channel must not tear down a replacement session. + if expected != nil && d.client != expected { + return + } if d.client != nil { d.client.Close() } @@ -393,6 +418,9 @@ func (d *sshDialer) invalidate() { } func (d *sshDialer) Close() error { + d.mu.Lock() + d.closed = true + d.mu.Unlock() d.invalidate() d.mu.Lock() if d.dohClient != nil { @@ -433,6 +461,8 @@ func (d *sshDialer) startKeepalive() { select { case <-ticker.C: if _, _, err := client.SendRequest("keepalive@openssh.com", true, nil); err != nil { + report(d.reporter, "event=ssh_keepalive result=failed reason=%s", errorClass(err)) + d.invalidateClient(client) return } case <-stop: @@ -460,3 +490,51 @@ func (c *sshTrackedConn) Write(p []byte) (int, error) { return n, err } func (c *sshTrackedConn) Close() error { err := c.Conn.Close(); c.once.Do(c.release); return err } + +// x/crypto's DialContext returns on cancellation while its internal Dial may still +// wait for CHANNEL_OPEN confirmation. Keep admission charged to that worker. +func dialSSHChannel(ctx context.Context, client *ssh.Client, target string, releaseAbandoned, abortStalled func(), grace time.Duration) (net.Conn, error, bool) { + if err := ctx.Err(); err != nil { + return nil, err, false + } + type result struct { + conn net.Conn + err error + } + results := make(chan result) + workerDone := make(chan struct{}) + go func() { + defer close(workerDone) + conn, err := client.Dial("tcp", target) + select { + case results <- result{conn, err}: + case <-ctx.Done(): + if conn != nil { + _ = conn.Close() + } + releaseAbandoned() + } + }() + select { + case outcome := <-results: + return outcome.conn, outcome.err, false + case <-ctx.Done(): + // Cancellation must not release admission early, but a peer that never answers + // CHANNEL_OPEN must not consume the entire pool forever either. + go func() { + timer := time.NewTimer(grace) + defer timer.Stop() + select { + case <-workerDone: + case <-timer.C: + select { + case <-workerDone: + return + default: + abortStalled() + } + } + }() + return nil, ctx.Err(), true + } +} diff --git a/native/mobile/ssh_dialer_test.go b/native/mobile/ssh_dialer_test.go index 9280d54..2afaa31 100644 --- a/native/mobile/ssh_dialer_test.go +++ b/native/mobile/ssh_dialer_test.go @@ -1,10 +1,14 @@ package mobile import ( + "context" "crypto/ed25519" "crypto/rand" + "io" "strings" + "sync" "testing" + "time" "golang.org/x/crypto/ssh" ) @@ -65,3 +69,137 @@ func TestSSHAuthenticationModes(t *testing.T) { t.Fatalf("password-only got %d methods", len(methods)) } } + +type rejectedSSHConn struct { + ssh.Conn + closed bool + failure error +} + +func (c *rejectedSSHConn) OpenChannel(string, []byte) (ssh.Channel, <-chan *ssh.Request, error) { + return nil, nil, c.failure +} +func (c *rejectedSSHConn) Close() error { c.closed = true; return nil } + +func TestSSHChannelFailureIsolation(t *testing.T) { + for _, failure := range []error{&ssh.OpenChannelError{Reason: ssh.ConnectionFailed}, context.Canceled, context.DeadlineExceeded, io.EOF} { + t.Run(failure.Error(), func(t *testing.T) { + conn := &rejectedSSHConn{failure: failure} + client := &ssh.Client{Conn: conn} + d := &sshDialer{client: client, config: config{SSHMaxChannels: 2}} + _, err := d.connectTarget(context.Background(), "example.com:443") + if err == nil { + t.Fatal("expected failure") + } + wantClosed := failure == io.EOF + if conn.closed != wantClosed { + t.Fatalf("session closed=%t want %t", conn.closed, wantClosed) + } + if len(d.channels) != 0 { + t.Fatal("channel slot leaked") + } + }) + } +} + +func TestSSHOldFailureDoesNotCloseReplacement(t *testing.T) { + current := &rejectedSSHConn{} + d := &sshDialer{client: &ssh.Client{Conn: current}} + d.invalidateClient(&ssh.Client{}) + if current.closed || d.client == nil { + t.Fatal("old session invalidated its replacement") + } +} + +type blockedSSHConn struct { + ssh.Conn + entered chan struct{} + unblock chan struct{} + once sync.Once +} + +func (c *blockedSSHConn) OpenChannel(string, []byte) (ssh.Channel, <-chan *ssh.Request, error) { + close(c.entered) + <-c.unblock + return nil, nil, io.EOF +} +func (c *blockedSSHConn) Close() error { c.once.Do(func() { close(c.unblock) }); return nil } + +func TestCancelledSSHOpenKeepsAdmissionUntilWorkerEnds(t *testing.T) { + raw := &blockedSSHConn{entered: make(chan struct{}), unblock: make(chan struct{})} + defer raw.Close() + client := &ssh.Client{Conn: raw} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + released := make(chan struct{}) + done := make(chan bool, 1) + go func() { + _, _, abandoned := dialSSHChannel(ctx, client, "example.com:443", func() { close(released) }, func() { t.Error("healthy worker aborted") }, time.Hour) + done <- abandoned + }() + <-raw.entered + cancel() + select { + case abandoned := <-done: + if !abandoned { + t.Fatal("worker not tracked") + } + case <-time.After(time.Second): + t.Fatal("cancel blocked") + } + select { + case <-released: + t.Fatal("released slot with worker still blocked") + default: + } + _ = raw.Close() + select { + case <-released: + case <-time.After(time.Second): + t.Fatal("worker did not release slot after transport closed") + } +} + +func TestClosedSSHDialerCannotReconnect(t *testing.T) { + d := &sshDialer{config: config{BypassLocalNetworks: true}} + _ = d.Close() + if _, err := d.session(context.Background()); err == nil { + t.Fatal("closed SSH dialer attempted a session") + } + if _, err := d.connectTarget(context.Background(), "127.0.0.1:443"); err == nil { + t.Fatal("closed SSH dialer used direct bypass") + } +} + +func TestAbandonedSSHOpenEventuallyReleasesPool(t *testing.T) { + raw := &blockedSSHConn{entered: make(chan struct{}), unblock: make(chan struct{})} + defer raw.Close() + client := &ssh.Client{Conn: raw} + d := &sshDialer{client: client} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + released := make(chan struct{}) + done := make(chan bool, 1) + go func() { + _, _, abandoned := dialSSHChannel(ctx, client, "example.com:443", func() { close(released) }, func() { d.invalidateClient(client) }, 20*time.Millisecond) + done <- abandoned + }() + <-raw.entered + cancel() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("caller did not cancel") + } + select { + case <-released: + case <-time.After(time.Second): + t.Fatal("unresponsive peer permanently occupied admission") + } + d.mu.Lock() + current := d.client + d.mu.Unlock() + if current != nil { + t.Fatal("stalled session not cleared for next connection") + } +} diff --git a/native/mobile/test_connection.go b/native/mobile/test_connection.go index 909d3d5..18c62b8 100644 --- a/native/mobile/test_connection.go +++ b/native/mobile/test_connection.go @@ -5,6 +5,7 @@ import ( "context" stdtls "crypto/tls" "encoding/json" + "errors" "fmt" "io" "net" @@ -38,6 +39,9 @@ type connectionTestResult struct { // TestConnection verifies the configured proxy path without starting a TUN device. func TestConnection(rawConfig string, protector Protector, reporter Reporter) (string, error) { + if protector == nil { + return "", errors.New("Android socket protector is required") + } c, err := parseConfig(rawConfig) if err != nil { return "", err @@ -143,6 +147,9 @@ func testHTTPSGet(ctx context.Context, connect func(context.Context, string) (ne return "", err } defer tunnel.Close() + // Close the raw tunnel directly: TLS Close can wait to send close_notify. + stopCancellation := context.AfterFunc(ctx, func() { _ = tunnel.Close() }) + defer stopCancellation() connection := stdtls.Client(tunnel, &stdtls.Config{ServerName: host, MinVersion: stdtls.VersionTLS12}) if err := connection.HandshakeContext(ctx); err != nil { @@ -151,6 +158,13 @@ func testHTTPSGet(ctx context.Context, connect func(context.Context, string) (ne tlsState := connection.ConnectionState() report(reporter, "event=connection_test stage=destination_tls result=success certificate=verified version=0x%04x cipher=0x%04x alpn=%s h2_negotiated=%t session_resumed=%t", tlsState.Version, tlsState.CipherSuite, normalizedALPN(tlsState.NegotiatedProtocol), tlsState.NegotiatedProtocol == "h2", tlsState.DidResume) + return testHTTPExchange(ctx, connection, host, path, readBody) +} + +func testHTTPExchange(ctx context.Context, connection net.Conn, host, path string, readBody bool) (string, error) { + // Request.Write/ReadResponse use raw I/O and do not observe Request.Context. + stopCancellation := context.AfterFunc(ctx, func() { _ = connection.Close() }) + defer stopCancellation() request, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://"+host+path, nil) if err != nil { return "", err @@ -160,7 +174,7 @@ func testHTTPSGet(ctx context.Context, connect func(context.Context, string) (ne if err := request.Write(connection); err != nil { return "", fmt.Errorf("write HTTPS request: %w", err) } - response, err := http.ReadResponse(bufio.NewReader(connection), request) + response, err := http.ReadResponse(bufio.NewReader(&limitedHeaderReader{reader: connection, remaining: 64 * 1024}), request) if err != nil { return "", fmt.Errorf("read HTTPS response: %w", err) } diff --git a/native/mobile/test_connection_test.go b/native/mobile/test_connection_test.go index 92e0d75..796fdd3 100644 --- a/native/mobile/test_connection_test.go +++ b/native/mobile/test_connection_test.go @@ -1,9 +1,13 @@ package mobile import ( + "bufio" "context" "errors" + "net" + "net/http" "testing" + "time" ) func TestParseIPAddress(t *testing.T) { @@ -75,3 +79,25 @@ func TestLookupEndpointValueFailsAfterEveryProvider(t *testing.T) { t.Fatalf("lookupEndpointValue returned value=%q err=%v attempts=%d", value, err, attempts) } } + +func TestHTTPExchangeCancellationInterruptsResponse(t *testing.T) { + client, server := net.Pipe() + defer client.Close() + defer server.Close() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { _, err := testHTTPExchange(ctx, client, "example.com", "/", true); done <- err }() + if _, err := http.ReadRequest(bufio.NewReader(server)); err != nil { + t.Fatal(err) + } + cancel() // Peer has accepted the request but never sends response headers. + select { + case err := <-done: + if err == nil { + t.Fatal("cancellation succeeded without an error") + } + case <-time.After(2 * time.Second): + t.Fatal("HTTP read ignored cancellation") + } +} diff --git a/scripts/build-fdroid-native.sh b/scripts/build-fdroid-native.sh index 50a6eb8..31b572c 100755 --- a/scripts/build-fdroid-native.sh +++ b/scripts/build-fdroid-native.sh @@ -4,6 +4,8 @@ set -euo pipefail script_dir="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" project_dir="$(cd -- "$script_dir/.." && pwd)" +source "$project_dir/scripts/java-toolchain.sh" + gomobile_version="v0.0.0-20260821190718-4776eadac327" : "${ANDROID_HOME:?ANDROID_HOME must point to the Android SDK}" diff --git a/scripts/build-release-apks.sh b/scripts/build-release-apks.sh index ea1a307..44e22c0 100755 --- a/scripts/build-release-apks.sh +++ b/scripts/build-release-apks.sh @@ -8,7 +8,7 @@ if [[ -f "$HOME/.zshrc.extra" ]]; then source "$HOME/.zshrc.extra" fi -: "${JAVA_HOME:=/opt/homebrew/opt/openjdk@21/libexec/openjdk.jdk/Contents/Home}" +source "$project_dir/scripts/java-toolchain.sh" : "${ANDROID_HOME:=$HOME/Library/Android/sdk}" : "${ANDROID_NDK_HOME:=$ANDROID_HOME/ndk/29.0.14206865}" : "${MEGAPROXY_KEYSTORE_PATH:=$HOME/AndroidApkKey}" diff --git a/scripts/build-release-bundle.sh b/scripts/build-release-bundle.sh index 570f67e..cc9aacd 100755 --- a/scripts/build-release-bundle.sh +++ b/scripts/build-release-bundle.sh @@ -7,7 +7,7 @@ if [[ -f "$HOME/.zshrc.extra" ]]; then source "$HOME/.zshrc.extra" fi -: "${JAVA_HOME:=/opt/homebrew/opt/openjdk@21/libexec/openjdk.jdk/Contents/Home}" +source "$project_dir/scripts/java-toolchain.sh" : "${ANDROID_HOME:=$HOME/Library/Android/sdk}" : "${ANDROID_NDK_HOME:=$ANDROID_HOME/ndk/29.0.14206865}" : "${MEGAPROXY_KEYSTORE_PATH:=$HOME/AndroidApkKey}" diff --git a/scripts/ci_changes.py b/scripts/ci_changes.py index a67c11a..1aa4310 100644 --- a/scripts/ci_changes.py +++ b/scripts/ci_changes.py @@ -81,7 +81,12 @@ def changed_files(base, head, pull_request=True): ] -def select_suites(base, head, pull_request=True, baselines=None): +def select_suites(base, head, pull_request=True, baselines=None, force_all=False): + if force_all: + return {suite: True for suite in SUITES}, { + suite: {"base": base, "run_id": None, "changed_files": None} + for suite in SUITES + } baselines = baselines or {} result = {} details = {} @@ -105,19 +110,28 @@ def main(): parser.add_argument( "--push", action="store_true", - help="Compare push endpoints instead of the PR merge base", + help="Run all suites for push CI (the workflow only accepts pushes to main)", ) parser.add_argument( "--history", action="store_true", help="Reuse successful ancestor checks for this PR", ) + parser.add_argument( + "--run-attempt", + type=int, + default=1, + help="GitHub run attempt; rerunning change scope forces all suites", + ) parser.add_argument("--github-output", type=Path) parser.add_argument("--summary", type=Path) args = parser.parse_args() + if args.run_attempt < 1: + parser.error("--run-attempt must be positive") + force_all = args.push or args.run_attempt > 1 try: baselines = {} - if args.history and not args.push: + if args.history and not args.push and not force_all: try: baselines = successful_baselines( os.environ["GITHUB_REPOSITORY"], @@ -138,8 +152,10 @@ def main(): print( "Check history unavailable; using the full PR diff", file=sys.stderr ) - result, details = select_suites(args.base, args.head, not args.push, baselines) - print(json.dumps({**result, "comparisons": details})) + result, details = select_suites( + args.base, args.head, not args.push, baselines, force_all + ) + print(json.dumps({**result, "forced": force_all, "comparisons": details})) if args.github_output: with args.github_output.open("a") as output: for suite, enabled in result.items(): @@ -150,12 +166,20 @@ def main(): for suite, enabled in result.items(): detail = details[suite] origin = ( - f"successful run {detail['run_id']}" - if detail["run_id"] + ( + "push CI (change filtering disabled)" + if args.push + else "full CI rerun (change filtering disabled)" + ) + if force_all else ( - "full PR diff (no reusable success)" - if not args.push - else "push endpoints" + f"successful run {detail['run_id']}" + if detail["run_id"] + else ( + "full PR diff (no reusable success)" + if not args.push + else "push endpoints" + ) ) ) summary.write( diff --git a/scripts/github_actions.py b/scripts/github_actions.py index cc93c6c..4df0a46 100755 --- a/scripts/github_actions.py +++ b/scripts/github_actions.py @@ -11,7 +11,7 @@ REPOSITORY = "andre487/AndroidMegaProxy" MODES = [ - ("ci", "Re-run all CI jobs"), + ("ci", "Re-run all CI jobs, including skipped checks"), ("failed", "Re-run failed CI jobs only"), ] PR_FIELDS = "number,title,state,headRefName,headRefOid,isCrossRepository" diff --git a/scripts/java-toolchain.sh b/scripts/java-toolchain.sh new file mode 100644 index 0000000..acc7d66 --- /dev/null +++ b/scripts/java-toolchain.sh @@ -0,0 +1,28 @@ +#!/usr/bin/env bash +# Source this file to select JDK 21 without installation-specific paths. +megaproxy_java_toolchain() { + local candidate="${JAVA_HOME:-}" properties version + if [[ -z "$candidate" && -x /usr/libexec/java_home ]]; then + candidate="$(/usr/libexec/java_home -v 21 2>/dev/null || true)" + fi + if [[ -n "$candidate" ]]; then + properties="$("$candidate/bin/java" -XshowSettings:properties -version 2>&1)" || { + echo "JAVA_HOME must point to a working JDK 21." >&2 + return 1 + } + else + properties="$(java -XshowSettings:properties -version 2>&1)" || { + echo "JDK 21 is required; configure JAVA_HOME or add it to PATH." >&2 + return 1 + } + candidate="$(sed -n 's/^[[:space:]]*java.home = //p' <<< "$properties")" + fi + version="$(sed -n 's/^[[:space:]]*java.specification.version = //p' <<< "$properties")" + if [[ "$version" != 21 || ! -x "$candidate/bin/javac" ]]; then + echo "JDK 21 is required; configure JAVA_HOME or add it to PATH." >&2 + return 1 + fi + export JAVA_HOME="$candidate" + export PATH="$JAVA_HOME/bin:$PATH" +} +megaproxy_java_toolchain diff --git a/scripts/tests/test_ci_changes.py b/scripts/tests/test_ci_changes.py index 7bc607e..842ea11 100644 --- a/scripts/tests/test_ci_changes.py +++ b/scripts/tests/test_ci_changes.py @@ -19,6 +19,99 @@ class ChangeScopeTest(unittest.TestCase): + def test_main_push_runs_every_suite_even_without_code_changes(self): + with tempfile.TemporaryDirectory() as root: + output = Path(root) / "output" + summary = Path(root) / "summary" + with ( + patch.object( + sys, + "argv", + [ + "ci_changes.py", + "--base", + "a" * 40, + "--head", + "a" * 40, + "--push", + "--run-attempt", + "1", + "--history", + "--github-output", + str(output), + "--summary", + str(summary), + ], + ), + patch.object(m, "successful_baselines", side_effect=AssertionError), + patch.object(m, "changed_files", side_effect=AssertionError), + patch.object(sys, "stdout", io.StringIO()) as stdout, + ): + self.assertEqual(0, m.main()) + self.assertTrue(json.loads(stdout.getvalue())["forced"]) + self.assertEqual( + "android=true\nnative=true\npython=true\n", output.read_text() + ) + self.assertIn("push CI (change filtering disabled)", summary.read_text()) + + def test_full_rerun_enables_skipped_suites_without_diff_or_history(self): + with tempfile.TemporaryDirectory() as root: + output = Path(root) / "output" + summary = Path(root) / "summary" + with ( + patch.object( + sys, + "argv", + [ + "ci_changes.py", + "--base", + "a" * 40, + "--head", + "b" * 40, + "--history", + "--run-attempt", + "2", + "--github-output", + str(output), + "--summary", + str(summary), + ], + ), + patch.object(m, "successful_baselines", side_effect=AssertionError), + patch.object(m, "changed_files", side_effect=AssertionError), + patch.object(sys, "stdout", io.StringIO()) as stdout, + ): + self.assertEqual(0, m.main()) + result = json.loads(stdout.getvalue()) + self.assertTrue(result["forced"]) + self.assertEqual( + "android=true\nnative=true\npython=true\n", output.read_text() + ) + self.assertIn("change filtering disabled", summary.read_text()) + + def test_initial_attempt_still_filters_docs_only_changes(self): + with ( + patch.object( + sys, + "argv", + [ + "ci_changes.py", + "--base", + "a" * 40, + "--head", + "b" * 40, + "--run-attempt", + "1", + ], + ), + patch.object(m, "changed_files", return_value=["README.md"]), + patch.object(sys, "stdout", io.StringIO()) as stdout, + ): + self.assertEqual(0, m.main()) + result = json.loads(stdout.getvalue()) + self.assertFalse(result["forced"]) + self.assertFalse(any(result[suite] for suite in m.SUITES)) + def test_python_and_markdown_do_not_run_android(self): self.assertEqual( {"android": False, "native": False, "python": True}, diff --git a/scripts/tests/test_github_actions.py b/scripts/tests/test_github_actions.py index 5098d2d..c83b360 100644 --- a/scripts/tests/test_github_actions.py +++ b/scripts/tests/test_github_actions.py @@ -54,6 +54,12 @@ def test_failed_mode_rejects_successful_run(self): with self.assertRaisesRegex(RuntimeError, "no failed conclusion"): m.plan_run(self.client, self.pr, "failed") + def test_full_rerun_accepts_successful_run_with_skipped_checks(self): + self.discovery.return_value[0]["conclusion"] = "success" + self.assertEqual( + ["run", "rerun", "20"], m.plan_run(self.client, self.pr, "ci")[0] + ) + def test_dry_run_never_launches_or_requests_confirmation(self): with ( patch.object(self.client, "run", side_effect=AssertionError), diff --git a/scripts/tests/test_java_toolchain.py b/scripts/tests/test_java_toolchain.py new file mode 100644 index 0000000..e4183c2 --- /dev/null +++ b/scripts/tests/test_java_toolchain.py @@ -0,0 +1,49 @@ +"""Exercise JDK validation without building or accessing signing material.""" + +import os +import subprocess +import tempfile +import unittest +from pathlib import Path + +SCRIPT = Path(__file__).resolve().parents[1] / "java-toolchain.sh" + + +class JavaToolchainTests(unittest.TestCase): + def run_toolchain(self, version, compiler=True): + with tempfile.TemporaryDirectory(prefix="megaproxy jdk ") as directory: + jdk = Path(directory) + (jdk / "bin").mkdir() + java = jdk / "bin/java" + java.write_text( + "#!/bin/sh\n" f"echo ' java.specification.version = {version}' >&2\n" + ) + java.chmod(0o755) + if compiler: + javac = jdk / "bin/javac" + javac.write_text("#!/bin/sh\nexit 0\n") + javac.chmod(0o755) + env = dict(os.environ, JAVA_HOME=str(jdk)) + return subprocess.run( + [ + "bash", + "-c", + 'set -e; source "$1"; test -x "$JAVA_HOME/bin/javac"', + "bash", + str(SCRIPT), + ], + env=env, + capture_output=True, + text=True, + timeout=10, + ) + + def test_accepts_jdk21_in_path_with_spaces(self): + result = self.run_toolchain(21) + self.assertEqual(0, result.returncode, result.stderr) + + def test_rejects_wrong_version(self): + self.assertNotEqual(0, self.run_toolchain(17).returncode) + + def test_rejects_runtime_without_compiler(self): + self.assertNotEqual(0, self.run_toolchain(21, compiler=False).returncode)