developer/galaxy-game
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

phase 4: connectrpc on the gateway authenticated edge

Browse Source 
Replace the native-gRPC server bootstrap with a single
`connectrpc.com/connect` HTTP/h2c listener. Connect-Go natively
serves Connect, gRPC, and gRPC-Web on the same port, so browsers can
now reach the authenticated surface without giving up the gRPC
framing native and desktop clients may use later. The decorator
stack (envelope → session → payload-hash → signature →
freshness/replay → rate-limit → routing/push) is reused unchanged
behind a small Connect → gRPC adapter and a `grpc.ServerStream`
shim around `*connect.ServerStream`.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Ilia Denisov
2026-05-07 11:49:28 +02:00
parent 39b7b2ef29
commit 118f7c17a2
30 changed files with 1009 additions and 772 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/internal/grpcapi/server_test.go
+82 -88
View File
@@ -2,6 +2,10 @@ package grpcapi
import (
"context"
"crypto/tls"
"errors"
"net"
"net/http"
"testing"
"time"
@@ -9,13 +13,12 @@ import (
"galaxy/gateway/internal/config"
"galaxy/gateway/internal/session"
gatewayv1 "galaxy/gateway/proto/galaxy/gateway/v1"
"galaxy/gateway/proto/galaxy/gateway/v1/gatewayv1connect"
"connectrpc.com/connect"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/status"
"golang.org/x/net/http2"
)
func TestExecuteCommandRejectsMalformedEnvelope(t *testing.T) {
@@ -25,15 +28,11 @@ func TestExecuteCommandRejectsMalformedEnvelope(t *testing.T) {
defer runGateway.stop(t)
addr := waitForListenAddr(t, server)
conn := dialGatewayClient(t, addr)
defer func() {
require.NoError(t, conn.Close())
}()
client := newEdgeClient(t, addr)
client := gatewayv1.NewEdgeGatewayClient(conn)
_, err := client.ExecuteCommand(context.Background(), &gatewayv1.ExecuteCommandRequest{})
_, err := client.ExecuteCommand(context.Background(), connect.NewRequest(&gatewayv1.ExecuteCommandRequest{}))
require.Error(t, err)
assert.Equal(t, codes.InvalidArgument, status.Code(err))
assert.Equal(t, connect.CodeInvalidArgument, connect.CodeOf(err))
}
func TestSubscribeEventsRejectsMalformedEnvelope(t *testing.T) {
@@ -43,15 +42,11 @@ func TestSubscribeEventsRejectsMalformedEnvelope(t *testing.T) {
defer runGateway.stop(t)
addr := waitForListenAddr(t, server)
conn := dialGatewayClient(t, addr)
defer func() {
require.NoError(t, conn.Close())
}()
client := newEdgeClient(t, addr)
client := gatewayv1.NewEdgeGatewayClient(conn)
err := subscribeEventsError(t, context.Background(), client, &gatewayv1.SubscribeEventsRequest{})
require.Error(t, err)
assert.Equal(t, codes.InvalidArgument, status.Code(err))
assert.Equal(t, connect.CodeInvalidArgument, connect.CodeOf(err))
}
func TestExecuteCommandRejectsUnsupportedProtocolVersion(t *testing.T) {
@@ -61,13 +56,9 @@ func TestExecuteCommandRejectsUnsupportedProtocolVersion(t *testing.T) {
defer runGateway.stop(t)
addr := waitForListenAddr(t, server)
conn := dialGatewayClient(t, addr)
defer func() {
require.NoError(t, conn.Close())
}()
client := newEdgeClient(t, addr)
client := gatewayv1.NewEdgeGatewayClient(conn)
_, err := client.ExecuteCommand(context.Background(), &gatewayv1.ExecuteCommandRequest{
_, err := client.ExecuteCommand(context.Background(), connect.NewRequest(&gatewayv1.ExecuteCommandRequest{
ProtocolVersion: "v2",
DeviceSessionId: "device-session-123",
MessageType: "fleet.move",
@@ -76,10 +67,10 @@ func TestExecuteCommandRejectsUnsupportedProtocolVersion(t *testing.T) {
PayloadBytes: []byte("payload"),
PayloadHash: []byte("hash"),
Signature: []byte("signature"),
})
}))
require.Error(t, err)
assert.Equal(t, codes.FailedPrecondition, status.Code(err))
assert.Equal(t, `unsupported protocol_version "v2"`, status.Convert(err).Message())
assert.Equal(t, connect.CodeFailedPrecondition, connect.CodeOf(err))
assert.Equal(t, `unsupported protocol_version "v2"`, connectErrorMessage(t, err))
}
func TestExecuteCommandValidEnvelopeStillReturnsUnimplemented(t *testing.T) {
@@ -96,15 +87,11 @@ func TestExecuteCommandValidEnvelopeStillReturnsUnimplemented(t *testing.T) {
defer runGateway.stop(t)
addr := waitForListenAddr(t, server)
conn := dialGatewayClient(t, addr)
defer func() {
require.NoError(t, conn.Close())
}()
client := newEdgeClient(t, addr)
client := gatewayv1.NewEdgeGatewayClient(conn)
_, err := client.ExecuteCommand(context.Background(), newValidExecuteCommandRequest())
_, err := client.ExecuteCommand(context.Background(), connect.NewRequest(newValidExecuteCommandRequest()))
require.Error(t, err)
assert.Equal(t, codes.Unimplemented, status.Code(err))
assert.Equal(t, connect.CodeUnimplemented, connect.CodeOf(err))
}
func TestExecuteCommandMissingReplayStoreFailsClosed(t *testing.T) {
@@ -120,16 +107,12 @@ func TestExecuteCommandMissingReplayStoreFailsClosed(t *testing.T) {
defer runGateway.stop(t)
addr := waitForListenAddr(t, server)
conn := dialGatewayClient(t, addr)
defer func() {
require.NoError(t, conn.Close())
}()
client := newEdgeClient(t, addr)
client := gatewayv1.NewEdgeGatewayClient(conn)
_, err := client.ExecuteCommand(context.Background(), newValidExecuteCommandRequest())
_, err := client.ExecuteCommand(context.Background(), connect.NewRequest(newValidExecuteCommandRequest()))
require.Error(t, err)
assert.Equal(t, codes.Unavailable, status.Code(err))
assert.Equal(t, "replay store is unavailable", status.Convert(err).Message())
assert.Equal(t, connect.CodeUnavailable, connect.CodeOf(err))
assert.Equal(t, "replay store is unavailable", connectErrorMessage(t, err))
}
func TestSubscribeEventsValidEnvelopeSendsBootstrapEventAndWaitsForCancellation(t *testing.T) {
@@ -149,22 +132,22 @@ func TestSubscribeEventsValidEnvelopeSendsBootstrapEventAndWaitsForCancellation(
defer runGateway.stop(t)
addr := waitForListenAddr(t, server)
conn := dialGatewayClient(t, addr)
defer func() {
require.NoError(t, conn.Close())
}()
client := newEdgeClient(t, addr)
client := gatewayv1.NewEdgeGatewayClient(conn)
stream, err := client.SubscribeEvents(ctx, newValidSubscribeEventsRequest())
stream, err := client.SubscribeEvents(ctx, connect.NewRequest(newValidSubscribeEventsRequest()))
require.NoError(t, err)
t.Cleanup(func() { _ = stream.Close() })
event := recvBootstrapEvent(t, stream)
assertServerTimeBootstrapEvent(t, event, newTestResponseSignerPublicKey(), "request-123", "trace-123", testCurrentTime.UnixMilli())
recvResult := make(chan error, 1)
go func() {
_, recvErr := stream.Recv()
recvResult <- recvErr
if stream.Receive() {
recvResult <- errors.New("stream produced unexpected event")
return
}
recvResult <- stream.Err()
}()
require.Never(t, func() bool {
@@ -188,7 +171,7 @@ func TestSubscribeEventsValidEnvelopeSendsBootstrapEventAndWaitsForCancellation(
}
}, time.Second, 10*time.Millisecond, "stream did not stop after client cancellation")
require.Error(t, recvErr)
assert.Equal(t, codes.Canceled, status.Code(recvErr))
assert.Equal(t, connect.CodeCanceled, connect.CodeOf(recvErr))
}
func TestSubscribeEventsMissingReplayStoreFailsClosed(t *testing.T) {
@@ -204,16 +187,12 @@ func TestSubscribeEventsMissingReplayStoreFailsClosed(t *testing.T) {
defer runGateway.stop(t)
addr := waitForListenAddr(t, server)
conn := dialGatewayClient(t, addr)
defer func() {
require.NoError(t, conn.Close())
}()
client := newEdgeClient(t, addr)
client := gatewayv1.NewEdgeGatewayClient(conn)
err := subscribeEventsError(t, context.Background(), client, newValidSubscribeEventsRequest())
require.Error(t, err)
assert.Equal(t, codes.Unavailable, status.Code(err))
assert.Equal(t, "replay store is unavailable", status.Convert(err).Message())
assert.Equal(t, connect.CodeUnavailable, connect.CodeOf(err))
assert.Equal(t, "replay store is unavailable", connectErrorMessage(t, err))
}
func TestSubscribeEventsFailsClosedWhenResponseSignerUnavailable(t *testing.T) {
@@ -231,16 +210,12 @@ func TestSubscribeEventsFailsClosedWhenResponseSignerUnavailable(t *testing.T) {
defer runGateway.stop(t)
addr := waitForListenAddr(t, server)
conn := dialGatewayClient(t, addr)
defer func() {
require.NoError(t, conn.Close())
}()
client := newEdgeClient(t, addr)
client := gatewayv1.NewEdgeGatewayClient(conn)
err := subscribeEventsError(t, context.Background(), client, newValidSubscribeEventsRequest())
require.Error(t, err)
assert.Equal(t, codes.Unavailable, status.Code(err))
assert.Equal(t, "response signer is unavailable", status.Convert(err).Message())
assert.Equal(t, connect.CodeUnavailable, connect.CodeOf(err))
assert.Equal(t, "response signer is unavailable", connectErrorMessage(t, err))
}
func TestServerLifecycle(t *testing.T) {
@@ -248,21 +223,23 @@ func TestServerLifecycle(t *testing.T) {
server, runGateway := newTestGateway(t, ServerDependencies{})
addr := waitForListenAddr(t, server)
conn := dialGatewayClient(t, addr)
require.NoError(t, conn.Close())
// Probe the listener before shutdown so we know it accepted at
// least one TCP connection.
probe, err := net.DialTimeout("tcp", addr, time.Second)
require.NoError(t, err)
require.NoError(t, probe.Close())
runGateway.stop(t)
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
// After shutdown the listener must refuse new TCP connections.
dialCtx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
defer cancel()
_, err := grpc.DialContext(
ctx,
addr,
grpc.WithTransportCredentials(insecure.NewCredentials()),
grpc.WithBlock(),
)
require.Error(t, err)
dialer := &net.Dialer{}
closedConn, err := dialer.DialContext(dialCtx, "tcp", addr)
if err == nil {
_ = closedConn.Close()
t.Fatalf("expected dial to %s to fail after shutdown", addr)
}
}
type runningGateway struct {
@@ -341,19 +318,36 @@ func waitForListenAddr(t *testing.T, server *Server) string {
return addr
}
func dialGatewayClient(t *testing.T, addr string) *grpc.ClientConn {
// newEdgeClient returns a Connect client speaking HTTP/2 cleartext to the
// authenticated edge listener. AllowHTTP forces the client to issue plain
// HTTP/2 requests (h2c) instead of attempting TLS, which the gateway's
// in-process test bootstrap does not configure.
func newEdgeClient(t *testing.T, addr string) gatewayv1connect.EdgeGatewayClient {
t.Helper()
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
conn, err := grpc.DialContext(
ctx,
addr,
grpc.WithTransportCredentials(insecure.NewCredentials()),
grpc.WithBlock(),
)
require.NoError(t, err)
return conn
httpClient := &http.Client{
Transport: &http2.Transport{
AllowHTTP: true,
DialTLSContext: func(ctx context.Context, network, target string, _ *tls.Config) (net.Conn, error) {
return (&net.Dialer{}).DialContext(ctx, network, target)
},
},
}
return gatewayv1connect.NewEdgeGatewayClient(httpClient, "http://"+addr)
}
// connectErrorMessage extracts the *connect.Error message from err. It
// fails the test if err is not a *connect.Error so the caller's expected
// message comparison doesn't accidentally match the wrapped Go error
// string instead of the protocol-level message.
func connectErrorMessage(t require.TestingT, err error) string {
if helper, ok := t.(interface{ Helper() }); ok {
helper.Helper()
}
var connectErr *connect.Error
if !errors.As(err, &connectErr) {
require.FailNowf(t, "expected *connect.Error", "got %T: %v", err, err)
}
return connectErr.Message()
}