@@ -3,25 +3,25 @@ package local
33import (
44 "context"
55 "fmt"
6+ "io"
67 "net"
78 "net/http"
8- "net/http/httputil"
9- "net/url"
109 "time"
1110
1211 "github.com/loft-sh/log"
12+ "google.golang.org/grpc"
13+ "google.golang.org/grpc/metadata"
14+ "google.golang.org/grpc/status"
1315 "tailscale.com/tsnet"
1416)
1517
1618const (
1719 // Listen on this port via tsnet.
1820 LocalCredentialsServerPort = 9999 // FIXME - use random prot
1921 // Target server: local gRPC server running on port 5555.
20- TargetServer = "http://localhost:5555 " // FIXME - get port from request
22+ TargetServer = "http://localhost:4795 " // FIXME - get port from request
2123)
2224
23- // LocalCredentialsServerProxy acts as a reverse proxy that blindly forwards
24- // all incoming traffic to the local gRPC server on port 5555.
2525type LocalCredentialsServerProxy struct {
2626 log log.Logger
2727 tsServer * tsnet.Server
@@ -30,16 +30,13 @@ type LocalCredentialsServerProxy struct {
3030 srv * http.Server
3131}
3232
33- // NewLocalCredentialsServerProxy initializes a new LocalCredentialsServerProxy.
3433func NewLocalCredentialsServerProxy (tsServer * tsnet.Server , log log.Logger ) (* LocalCredentialsServerProxy , error ) {
3534 return & LocalCredentialsServerProxy {
3635 log : log ,
3736 tsServer : tsServer ,
3837 }, nil
3938}
4039
41- // Listen creates the tsnet listener and HTTP server,
42- // and registers a catch-all handler that acts as the reverse proxy.
4340func (s * LocalCredentialsServerProxy ) Listen (ctx context.Context ) error {
4441 s .log .Info ("Starting reverse proxy for local gRPC server" )
4542
@@ -51,67 +48,132 @@ func (s *LocalCredentialsServerProxy) Listen(ctx context.Context) error {
5148 }
5249 s .ln = ln
5350
54- mux := http .NewServeMux ()
55- mux .HandleFunc ("/" , s .handleReverseProxy )
51+ serverOpts := []grpc.ServerOption {
52+ grpc .UnaryInterceptor (s .unaryProxyInterceptor ),
53+ grpc .StreamInterceptor (s .streamProxyInterceptor ),
54+ }
5655
57- // Create the HTTP server.
58- s .srv = & http.Server {
59- Handler : mux ,
56+ grpcServer := grpc .NewServer (serverOpts ... )
57+ s .log .Infof ("gRPC reverse proxy listening on %s" , fmt .Sprintf (":%d" , LocalCredentialsServerPort ))
58+ if err := grpcServer .Serve (ln ); err != nil {
59+ s .log .Fatalf ("failed to serve: %v" , err )
6060 }
61+ return nil
62+ }
6163
62- go func () {
63- <- ctx .Done ()
64- s .log .Info ("Context canceled, shutting down reverse proxy" )
65- shutdownCtx , cancel := context .WithTimeout (context .Background (), 5 * time .Second )
66- defer cancel ()
67- if err := s .srv .Shutdown (shutdownCtx ); err != nil {
68- s .log .Errorf ("Error shutting down reverse proxy: %v" , err )
69- }
70- }()
64+ func (s * LocalCredentialsServerProxy ) unaryProxyInterceptor (
65+ ctx context.Context ,
66+ req interface {},
67+ info * grpc.UnaryServerInfo ,
68+ handler grpc.UnaryHandler ,
69+ ) (interface {}, error ) {
70+ s .log .Infof ("Start unary proxy interceptor - %v \n " , info )
71+ md , ok := metadata .FromIncomingContext (ctx )
72+ if ! ok {
73+ return nil , status .Errorf (400 , "missing metadata" )
74+ }
7175
72- s . log . Infof ( "Reverse proxy listening on tsnet port %d" , LocalCredentialsServerPort )
73- err = s . srv . Serve ( ln )
74- if err != nil && err != http . ErrServerClosed {
75- s . log . Errorf ( "Reverse proxy error: %v" , err )
76- return err
76+ // Retrieve target headers.
77+ targetHosts := md . Get ( "x-target-host" )
78+ targetPorts := md . Get ( "x-target-port" )
79+ if len ( targetHosts ) == 0 || len ( targetPorts ) == 0 {
80+ return nil , status . Errorf ( 400 , "missing x-target-host or x-target-port metadata" )
7781 }
82+ targetAddr := fmt .Sprintf ("%s:%s" , targetHosts [0 ], targetPorts [0 ])
83+ s .log .Infof ("Proxying unary call %q to target %s" , info .FullMethod , targetAddr )
7884
79- return nil
85+ // Establish connection to the target server.
86+ conn , err := grpc .Dial (targetAddr , grpc .WithInsecure ())
87+ if err != nil {
88+ return nil , status .Errorf (503 , "failed to dial target %s: %v" , targetAddr , err )
89+ }
90+ defer conn .Close ()
91+
92+ var resp interface {}
93+ // Forward the call.
94+ err = conn .Invoke (ctx , info .FullMethod , req , & resp )
95+ if err != nil {
96+ return nil , status .Errorf (500 , "error invoking target: %v" , err )
97+ }
98+ return resp , nil
8099}
81100
82- // handleReverseProxy forwards every request to the target gRPC server.
83- func (s * LocalCredentialsServerProxy ) handleReverseProxy (w http.ResponseWriter , r * http.Request ) {
84- s .log .Infof ("Forwarding request %s %s to target server" , r .Method , r .URL .String ())
101+ func (s * LocalCredentialsServerProxy ) streamProxyInterceptor (
102+ srv interface {},
103+ ss grpc.ServerStream ,
104+ info * grpc.StreamServerInfo ,
105+ handler grpc.StreamHandler ,
106+ ) error {
107+ s .log .Infof ("Start stream proxy interceptor - %v \n " , info )
108+ // Extract incoming metadata.
109+ md , ok := metadata .FromIncomingContext (ss .Context ())
110+ if ! ok {
111+ return status .Errorf (400 , "missing metadata" )
112+ }
113+
114+ // Retrieve target metadata.
115+ targetHosts := md .Get ("x-target-host" )
116+ targetPorts := md .Get ("x-target-port" )
117+ if len (targetHosts ) == 0 || len (targetPorts ) == 0 {
118+ return status .Errorf (400 , "missing x-target-host or x-target-port metadata" )
119+ }
120+ targetAddr := fmt .Sprintf ("%s:%s" , targetHosts [0 ], targetPorts [0 ])
121+ s .log .Infof ("Proxying streaming call %q to target %s" , info .FullMethod , targetAddr )
85122
86- // Parse the target URL .
87- targetURL , err := url . Parse ( TargetServer )
123+ // Dial the target server .
124+ conn , err := grpc . Dial ( targetAddr , grpc . WithInsecure () )
88125 if err != nil {
89- s .log .Errorf ("Error parsing target URL %s: %v" , TargetServer , err )
90- http .Error (w , "Bad Gateway" , http .StatusBadGateway )
91- return
126+ return status .Errorf (503 , "failed to dial target %s: %v" , targetAddr , err )
92127 }
128+ defer conn .Close ()
93129
94- // Create the reverse proxy .
95- proxy := httputil . NewSingleHostReverseProxy ( targetURL )
130+ // Create a new context for the client stream and attach metadata .
131+ clientCtx := metadata . NewOutgoingContext ( ss . Context (), md )
96132
97- // Customize the director to forward the Host header to the target.
98- originalDirector := proxy .Director
99- proxy .Director = func (req * http.Request ) {
100- originalDirector (req )
101- req .Host = targetURL .Host
133+ clientStream , err := conn .NewStream (clientCtx , & grpc.StreamDesc {
134+ ServerStreams : info .IsServerStream ,
135+ ClientStreams : info .IsClientStream ,
136+ }, info .FullMethod )
137+ if err != nil {
138+ return status .Errorf (500 , "failed to create stream to target: %v" , err )
102139 }
103140
104- // Use an error handler to log any errors that occur.
105- proxy .ErrorHandler = func (w http.ResponseWriter , r * http.Request , err error ) {
106- s .log .Errorf ("Reverse proxy error: %v" , err )
107- http .Error (w , "Bad Gateway" , http .StatusBadGateway )
108- }
141+ errChan := make (chan error , 2 )
142+ go func () {
143+ for {
144+ var msg interface {}
145+ if err := ss .RecvMsg (& msg ); err != nil {
146+ errChan <- err
147+ return
148+ }
149+ if err := clientStream .SendMsg (msg ); err != nil {
150+ errChan <- err
151+ return
152+ }
153+ }
154+ }()
109155
110- // Forward the request.
111- proxy .ServeHTTP (w , r )
156+ go func () {
157+ for {
158+ var msg interface {}
159+ if err := clientStream .RecvMsg (& msg ); err != nil {
160+ errChan <- err
161+ return
162+ }
163+ if err := ss .SendMsg (msg ); err != nil {
164+ errChan <- err
165+ return
166+ }
167+ }
168+ }()
169+
170+ err = <- errChan
171+ if err == io .EOF {
172+ return nil
173+ }
174+ return err
112175}
113176
114- // Close gracefully shuts down the reverse proxy.
115177func (s * LocalCredentialsServerProxy ) Close () error {
116178 s .log .Info ("Closing reverse proxy" )
117179 if s .srv != nil {
0 commit comments