@samitouri / QOSAMI-X402Tunnel / commits / 9d8122d6

tunnel: support multi relays

Kim committed Mar 6, 2026 at 16:20 UTC 9d8122d67355404ceeea96c1174c88cd3d2a60c7
4 files changed +352 -198
cmd/portal-tunnel/README.md
+4 -3
@@ -6,7 +6,7 @@ Portal-tunnel connects a local service to a Portal relay with the legacy CLI sha
6
7 ```bash
8 ./portal-tunnel --host localhost:8080 \
9 - --relay https://portal.example.com \
9 + --relays https://portal.example.com \
10 --name myapp \
11 --description "Service description" \
12 --tags tag1,tag2 \
@@ -17,7 +17,7 @@ Portal-tunnel connects a local service to a Portal relay with the legacy CLI sha
17 ## Flags
18
19 ```text
20 ---relay Portal relay server API URLs (comma-separated, https only) [env: RELAYS]
20 +--relays Portal relay server API URLs (comma-separated, https only) [env: RELAYS]
21 --host Target host to proxy to (host:port or URL) [env: APP_HOST]
22 --name Service name [env: APP_NAME]
23 --description Service description metadata [env: APP_DESCRIPTION]
@@ -29,6 +29,7 @@ Portal-tunnel connects a local service to a Portal relay with the legacy CLI sha
29
30 ## Notes
31
32 -- The current runtime accepts multiple relay URLs but uses the first one.
32 +- Multiple relay URLs are registered independently. Each relay gets its own lease ID and public URLs.
33 +- Startup is fail-fast: if any configured relay cannot register, the tunnel exits instead of partially publishing.
34 - Tenant TLS is provisioned automatically through the relay keyless signer. The SDK fetches the relay certificate chain and uses `/v1/sign` for remote signing.
35 - When the local service is unreachable, the tunnel returns an HTTP 503 page.
cmd/portal-tunnel/main.go
+37 -192
@@ -5,12 +5,10 @@ import (
5 "errors"
6 "flag"
7 "fmt"
8 - "io"
9 - "net"
8 "os"
9 "os/signal"
12 - "strings"
10 "sync"
11 + "sync/atomic"
12 "syscall"
13 "time"
14
@@ -42,7 +40,7 @@ func main() {
40 defaultRelayURLs = "https://localhost:4017"
41 }
42
45 - flag.StringVar(&flagRelayURLs, "relay", defaultRelayURLs, "Portal relay server API URLs (comma-separated, https only) [env: RELAYS]")
43 + flag.StringVar(&flagRelayURLs, "relays", defaultRelayURLs, "Portal relay server API URLs (comma-separated, https only) [env: RELAYS]")
44 flag.StringVar(&flagHost, "host", os.Getenv("APP_HOST"), "Target host to proxy to (host:port or URL) [env: APP_HOST]")
45 flag.StringVar(&flagName, "name", os.Getenv("APP_NAME"), "Service name [env: APP_NAME]")
46 flag.StringVar(&flagDesc, "description", os.Getenv("APP_DESCRIPTION"), "Service description metadata [env: APP_DESCRIPTION]")
@@ -64,93 +62,63 @@ func runTunnel() error {
62 ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
63 defer stop()
64
67 - relayURLs := parseURLs(flagRelayURLs)
65 + relayURLs, err := normalizeRelayURLs(flagRelayURLs)
66 + if err != nil {
67 + return err
68 + }
69 if len(relayURLs) == 0 {
70 return errors.New("no relay URLs provided")
71 }
71 - relayURL := strings.TrimSpace(relayURLs[0])
72
73 logger.Info().
74 Str("local", flagHost).
75 + Int("relay_count", len(relayURLs)).
76 Strs("relays", relayURLs).
77 Msg("starting portal tunnel")
77 - if len(relayURLs) > 1 {
78 - logger.Warn().Str("selected_relay", relayURL).Msg("multiple relays configured; using first relay only")
79 - }
78
81 - sdkClient, err := sdk.NewClient(sdk.ClientConfig{RelayURL: relayURL})
82 - if err != nil {
83 - return fmt.Errorf("service %s: failed to create client: %w", flagName, err)
84 - }
85 - defer sdkClient.Close()
86 -
87 - listener, err := sdkClient.Listen(ctx, sdk.ListenRequest{
79 + listenReq := sdk.ListenRequest{
80 Name: flagName,
81 Metadata: types.LeaseMetadata{
82 Description: flagDesc,
91 - Tags: parseURLs(flagTags),
83 + Tags: splitCSV(flagTags),
84 Owner: flagOwner,
85 Thumbnail: flagThumbnail,
86 Hide: flagHide,
87 },
96 - })
97 - if err != nil {
98 - return fmt.Errorf("service %s: failed to register service: %w", flagName, err)
88 }
100 - defer listener.Close()
89
102 - go func() {
103 - <-ctx.Done()
104 - _ = listener.Close()
105 - }()
106 -
107 - logger.Info().
108 - Str("relay", relayURL).
109 - Str("lease_id", listener.LeaseID()).
110 - Strs("public_urls", listener.PublicURLs()).
111 - Msg("tunnel ready")
90 + runtimes, err := startRelayRuntimes(ctx, relayURLs, listenReq)
91 + if err != nil {
92 + return fmt.Errorf("service %s: failed to start relays: %w", flagName, err)
93 + }
94
113 - connCount := 0
95 var connWG sync.WaitGroup
96 + var connCount atomic.Int64
97 + relayDone := make(chan relayLoopResult, len(runtimes))
98
116 -loop:
117 - for {
118 - select {
119 - case <-ctx.Done():
120 - logger.Info().Msg("tunnel shutting down")
121 - break loop
122 - default:
123 - }
124 -
125 - relayConn, err := listener.Accept()
126 - if err != nil {
127 - if errors.Is(err, net.ErrClosed) {
128 - logger.Info().Msg("tunnel listener closed")
129 - break loop
130 - }
131 - select {
132 - case <-ctx.Done():
133 - break loop
134 - default:
135 - logger.Error().Err(err).Msg("failed to accept relay connection")
136 - continue
137 - }
138 - }
139 -
140 - connCount++
99 + for _, runtime := range runtimes {
100 logger.Info().
142 - Int("conn_id", connCount).
143 - Str("remote_addr", relayConn.RemoteAddr().String()).
144 - Msg("accepted relay connection")
101 + Str("relay", runtime.relayURL).
102 + Str("lease_id", runtime.listener.LeaseID()).
103 + Strs("public_urls", runtime.listener.PublicURLs()).
104 + Msg("relay tunnel ready")
105 + go runtime.run(ctx, flagHost, &connWG, &connCount, relayDone)
106 + }
107
146 - connWG.Add(1)
147 - go func(connID int, relayConn net.Conn) {
148 - defer connWG.Done()
149 - if err := proxyConnection(ctx, flagHost, relayConn); err != nil {
150 - logger.Error().Err(err).Int("conn_id", connID).Msg("proxy connection failed")
151 - }
152 - logger.Info().Int("conn_id", connID).Msg("proxy connection closed")
153 - }(connCount, relayConn)
108 + waitErr := waitForRelayLoops(ctx, relayDone, len(runtimes))
109 + if waitErr != nil {
110 + stop()
111 + }
112 + closeErr := closeRelayRuntimes(runtimes)
113 + if waitErr != nil {
114 + logger.Error().Err(waitErr).Msg("relay supervisor exited with error")
115 + }
116 + if closeErr != nil {
117 + logger.Error().Err(closeErr).Msg("relay shutdown failed")
118 + }
119 +
120 + if ctx.Err() != nil {
121 + logger.Info().Msg("tunnel shutting down")
122 }
123
124 done := make(chan struct{})
@@ -166,128 +134,5 @@ loop:
134 }
135
136 logger.Info().Msg("tunnel shutdown complete")
169 - return nil
170 -}
171 -
172 -var bufferPool = sync.Pool{
173 - New: func() any {
174 - b := make([]byte, 64*1024)
175 - return &b
176 - },
177 -}
178 -
179 -func proxyConnection(ctx context.Context, localAddr string, relayConn net.Conn) error {
180 - defer relayConn.Close()
181 -
182 - targetAddr, err := normalizeTargetAddr(localAddr)
183 - if err != nil {
184 - return fmt.Errorf("invalid --host value %q: %w", localAddr, err)
185 - }
186 -
187 - dialer := &net.Dialer{Timeout: 5 * time.Second}
188 - localConn, err := dialer.DialContext(ctx, "tcp", targetAddr)
189 - if err != nil {
190 - return writeEmptyHTTPResponse(relayConn)
191 - }
192 - defer localConn.Close()
193 -
194 - errCh := make(chan error, 2)
195 - stopCh := make(chan struct{})
196 -
197 - go func() {
198 - select {
199 - case <-ctx.Done():
200 - _ = relayConn.Close()
201 - _ = localConn.Close()
202 - case <-stopCh:
203 - }
204 - }()
205 -
206 - go func() {
207 - bufPtr := bufferPool.Get().(*[]byte)
208 - defer bufferPool.Put(bufPtr)
209 - _, err := io.CopyBuffer(localConn, relayConn, *bufPtr)
210 - if tcpConn, ok := localConn.(*net.TCPConn); ok {
211 - _ = tcpConn.CloseWrite()
212 - }
213 - errCh <- err
214 - }()
215 -
216 - go func() {
217 - bufPtr := bufferPool.Get().(*[]byte)
218 - defer bufferPool.Put(bufPtr)
219 - _, err := io.CopyBuffer(relayConn, localConn, *bufPtr)
220 - _ = relayConn.Close()
221 - errCh <- err
222 - }()
223 -
224 - var firstErr error
225 - for range 2 {
226 - if err := <-errCh; err != nil && firstErr == nil {
227 - firstErr = err
228 - }
229 - }
230 -
231 - close(stopCh)
232 - if errors.Is(firstErr, io.EOF) || errors.Is(firstErr, net.ErrClosed) {
233 - return nil
234 - }
235 - return firstErr
236 -}
237 -
238 -func writeEmptyHTTPResponse(conn net.Conn) error {
239 - htmlBody := `<!DOCTYPE html>
240 -<html>
241 -<head><title>Service Unavailable</title></head>
242 -<body style="font-family:sans-serif;text-align:center;padding:50px;">
243 -<h1>Service Unavailable</h1>
244 -<p>The local service is not currently running.</p>
245 -<p>Please start your local application and refresh this page.</p>
246 -</body>
247 -</html>`
248 - response := fmt.Sprintf("HTTP/1.1 503 Service Unavailable\r\n"+
249 - "Content-Type: text/html; charset=utf-8\r\n"+
250 - "Content-Length: %d\r\n"+
251 - "Connection: close\r\n"+
252 - "\r\n%s", len(htmlBody), htmlBody)
253 - _, err := conn.Write([]byte(response))
254 - return err
255 -}
256 -
257 -func parseURLs(raw string) []string {
258 - if strings.TrimSpace(raw) == "" {
259 - return nil
260 - }
261 - parts := strings.Split(raw, ",")
262 - out := make([]string, 0, len(parts))
263 - for _, part := range parts {
264 - part = strings.TrimSpace(part)
265 - if part != "" {
266 - out = append(out, part)
267 - }
268 - }
269 - return out
270 -}
271 -
272 -func normalizeTargetAddr(raw string) (string, error) {
273 - raw = strings.TrimSpace(raw)
274 - if raw == "" {
275 - return "", errors.New("target address is required")
276 - }
277 - if strings.Contains(raw, "://") {
278 - if strings.HasPrefix(strings.ToLower(raw), "http://") {
279 - raw = strings.TrimPrefix(raw, "http://")
280 - }
281 - if strings.HasPrefix(strings.ToLower(raw), "https://") {
282 - raw = strings.TrimPrefix(raw, "https://")
283 - }
284 - raw = strings.TrimSuffix(raw, "/")
285 - }
286 - if _, _, err := net.SplitHostPort(raw); err == nil {
287 - return raw, nil
288 - }
289 - if strings.Count(raw, ":") == 0 {
290 - return net.JoinHostPort(raw, "80"), nil
291 - }
292 - return "", fmt.Errorf("invalid target address %q", raw)
137 + return errors.Join(waitErr, closeErr)
138 }
cmd/portal-tunnel/relays.go new
+308
@@ -0,0 +1,308 @@
1 +package main
2 +
3 +import (
4 + "context"
5 + "errors"
6 + "fmt"
7 + "io"
8 + "net"
9 + "net/url"
10 + "strings"
11 + "sync"
12 + "sync/atomic"
13 + "time"
14 +
15 + "github.com/rs/zerolog/log"
16 +
17 + "gosuda.org/portal/sdk"
18 +)
19 +
20 +type relayRuntime struct {
21 + relayURL string
22 + client *sdk.Client
23 + listener *sdk.Listener
24 +}
25 +
26 +type relayLoopResult struct {
27 + leaseID string
28 + relayURL string
29 + err error
30 +}
31 +
32 +func (r *relayRuntime) run(ctx context.Context, localAddr string, connWG *sync.WaitGroup, connCount *atomic.Int64, done chan<- relayLoopResult) {
33 + logger := log.With().
34 + Str("component", "portal-tunnel").
35 + Str("relay", r.relayURL).
36 + Str("lease_id", r.listener.LeaseID()).
37 + Logger()
38 +
39 + var runErr error
40 + defer func() {
41 + done <- relayLoopResult{
42 + leaseID: r.listener.LeaseID(),
43 + relayURL: r.relayURL,
44 + err: runErr,
45 + }
46 + }()
47 +
48 + for {
49 + relayConn, err := r.listener.Accept()
50 + if err != nil {
51 + if errors.Is(err, net.ErrClosed) || errors.Is(err, context.Canceled) || ctx.Err() != nil {
52 + return
53 + }
54 + runErr = err
55 + return
56 + }
57 +
58 + connID := connCount.Add(1)
59 + logger.Info().
60 + Int64("conn_id", connID).
61 + Str("remote_addr", relayConn.RemoteAddr().String()).
62 + Msg("accepted relay connection")
63 +
64 + connWG.Add(1)
65 + go func(connID int64, relayConn net.Conn) {
66 + defer connWG.Done()
67 + if err := proxyConnection(ctx, localAddr, relayConn); err != nil {
68 + logger.Error().Err(err).Int64("conn_id", connID).Msg("proxy connection failed")
69 + }
70 + logger.Info().Int64("conn_id", connID).Msg("proxy connection closed")
71 + }(connID, relayConn)
72 + }
73 +}
74 +
75 +func startRelayRuntimes(ctx context.Context, relayURLs []string, req sdk.ListenRequest) ([]*relayRuntime, error) {
76 + runtimes := make([]*relayRuntime, 0, len(relayURLs))
77 + for _, relayURL := range relayURLs {
78 + client, err := sdk.NewClient(sdk.ClientConfig{RelayURL: relayURL})
79 + if err != nil {
80 + _ = closeRelayRuntimes(runtimes)
81 + return nil, fmt.Errorf("create relay client %s: %w", relayURL, err)
82 + }
83 +
84 + listener, err := client.Listen(ctx, req)
85 + if err != nil {
86 + client.Close()
87 + _ = closeRelayRuntimes(runtimes)
88 + return nil, fmt.Errorf("register relay lease %s: %w", relayURL, err)
89 + }
90 +
91 + runtimes = append(runtimes, &relayRuntime{
92 + relayURL: relayURL,
93 + client: client,
94 + listener: listener,
95 + })
96 + }
97 + return runtimes, nil
98 +}
99 +
100 +func closeRelayRuntimes(runtimes []*relayRuntime) error {
101 + var closeErr error
102 + for _, runtime := range runtimes {
103 + if runtime == nil {
104 + continue
105 + }
106 + if runtime.listener != nil {
107 + closeErr = errors.Join(closeErr, runtime.listener.Close())
108 + }
109 + if runtime.client != nil {
110 + runtime.client.Close()
111 + }
112 + }
113 + return closeErr
114 +}
115 +
116 +func waitForRelayLoops(ctx context.Context, done <-chan relayLoopResult, relayCount int) error {
117 + logger := log.With().Str("component", "portal-tunnel").Logger()
118 + active := relayCount
119 +
120 + for active > 0 {
121 + result := <-done
122 + active--
123 +
124 + switch {
125 + case result.err != nil:
126 + logger.Error().
127 + Err(result.err).
128 + Str("relay", result.relayURL).
129 + Str("lease_id", result.leaseID).
130 + Int("remaining_relays", active).
131 + Msg("relay accept loop stopped")
132 + case ctx.Err() != nil:
133 + logger.Info().
134 + Str("relay", result.relayURL).
135 + Str("lease_id", result.leaseID).
136 + Int("remaining_relays", active).
137 + Msg("relay accept loop stopped during shutdown")
138 + default:
139 + logger.Warn().
140 + Str("relay", result.relayURL).
141 + Str("lease_id", result.leaseID).
142 + Int("remaining_relays", active).
143 + Msg("relay accept loop stopped")
144 + }
145 + }
146 +
147 + if ctx.Err() != nil {
148 + return nil
149 + }
150 + return errors.New("all relay listeners stopped")
151 +}
152 +
153 +func normalizeRelayURLs(raw string) ([]string, error) {
154 + seen := make(map[string]struct{})
155 + var relayURLs []string
156 + for _, relayURL := range splitCSV(raw) {
157 + normalized, err := normalizeRelayURL(relayURL)
158 + if err != nil {
159 + return nil, err
160 + }
161 + if _, ok := seen[normalized]; ok {
162 + continue
163 + }
164 + seen[normalized] = struct{}{}
165 + relayURLs = append(relayURLs, normalized)
166 + }
167 + return relayURLs, nil
168 +}
169 +
170 +func normalizeRelayURL(raw string) (string, error) {
171 + u, err := url.Parse(strings.TrimSpace(raw))
172 + if err != nil {
173 + return "", fmt.Errorf("parse relay url: %w", err)
174 + }
175 + if !strings.EqualFold(u.Scheme, "https") {
176 + return "", fmt.Errorf("relay url must use https: %q", raw)
177 + }
178 + if u.Host == "" {
179 + return "", fmt.Errorf("relay url host is empty: %q", raw)
180 + }
181 + u.Path = strings.TrimRight(u.Path, "/")
182 + u.RawQuery = ""
183 + u.Fragment = ""
184 + return u.String(), nil
185 +}
186 +
187 +var bufferPool = sync.Pool{
188 + New: func() any {
189 + b := make([]byte, 64*1024)
190 + return &b
191 + },
192 +}
193 +
194 +func proxyConnection(ctx context.Context, localAddr string, relayConn net.Conn) error {
195 + defer relayConn.Close()
196 +
197 + targetAddr, err := normalizeTargetAddr(localAddr)
198 + if err != nil {
199 + return fmt.Errorf("invalid --host value %q: %w", localAddr, err)
200 + }
201 +
202 + dialer := &net.Dialer{Timeout: 5 * time.Second}
203 + localConn, err := dialer.DialContext(ctx, "tcp", targetAddr)
204 + if err != nil {
205 + return writeEmptyHTTPResponse(relayConn)
206 + }
207 + defer localConn.Close()
208 +
209 + errCh := make(chan error, 2)
210 + stopCh := make(chan struct{})
211 +
212 + go func() {
213 + select {
214 + case <-ctx.Done():
215 + _ = relayConn.Close()
216 + _ = localConn.Close()
217 + case <-stopCh:
218 + }
219 + }()
220 +
221 + go func() {
222 + bufPtr := bufferPool.Get().(*[]byte)
223 + defer bufferPool.Put(bufPtr)
224 + _, err := io.CopyBuffer(localConn, relayConn, *bufPtr)
225 + if tcpConn, ok := localConn.(*net.TCPConn); ok {
226 + _ = tcpConn.CloseWrite()
227 + }
228 + errCh <- err
229 + }()
230 +
231 + go func() {
232 + bufPtr := bufferPool.Get().(*[]byte)
233 + defer bufferPool.Put(bufPtr)
234 + _, err := io.CopyBuffer(relayConn, localConn, *bufPtr)
235 + _ = relayConn.Close()
236 + errCh <- err
237 + }()
238 +
239 + var firstErr error
240 + for range 2 {
241 + if err := <-errCh; err != nil && firstErr == nil {
242 + firstErr = err
243 + }
244 + }
245 +
246 + close(stopCh)
247 + if errors.Is(firstErr, io.EOF) || errors.Is(firstErr, net.ErrClosed) {
248 + return nil
249 + }
250 + return firstErr
251 +}
252 +
253 +func writeEmptyHTTPResponse(conn net.Conn) error {
254 + htmlBody := `<!DOCTYPE html>
255 +<html>
256 +<head><title>Service Unavailable</title></head>
257 +<body style="font-family:sans-serif;text-align:center;padding:50px;">
258 +<h1>Service Unavailable</h1>
259 +<p>The local service is not currently running.</p>
260 +<p>Please start your local application and refresh this page.</p>
261 +</body>
262 +</html>`
263 + response := fmt.Sprintf("HTTP/1.1 503 Service Unavailable\r\n"+
264 + "Content-Type: text/html; charset=utf-8\r\n"+
265 + "Content-Length: %d\r\n"+
266 + "Connection: close\r\n"+
267 + "\r\n%s", len(htmlBody), htmlBody)
268 + _, err := conn.Write([]byte(response))
269 + return err
270 +}
271 +
272 +func splitCSV(raw string) []string {
273 + if strings.TrimSpace(raw) == "" {
274 + return nil
275 + }
276 + parts := strings.Split(raw, ",")
277 + out := make([]string, 0, len(parts))
278 + for _, part := range parts {
279 + part = strings.TrimSpace(part)
280 + if part != "" {
281 + out = append(out, part)
282 + }
283 + }
284 + return out
285 +}
286 +
287 +func normalizeTargetAddr(raw string) (string, error) {
288 + raw = strings.TrimSpace(raw)
289 + if raw == "" {
290 + return "", errors.New("target address is required")
291 + }
292 + if strings.Contains(raw, "://") {
293 + if strings.HasPrefix(strings.ToLower(raw), "http://") {
294 + raw = strings.TrimPrefix(raw, "http://")
295 + }
296 + if strings.HasPrefix(strings.ToLower(raw), "https://") {
297 + raw = strings.TrimPrefix(raw, "https://")
298 + }
299 + raw = strings.TrimSuffix(raw, "/")
300 + }
301 + if _, _, err := net.SplitHostPort(raw); err == nil {
302 + return raw, nil
303 + }
304 + if strings.Count(raw, ":") == 0 {
305 + return net.JoinHostPort(raw, "80"), nil
306 + }
307 + return "", fmt.Errorf("invalid target address %q", raw)
308 +}
cmd/relay-server/tunnel.go
+3 -3
@@ -75,7 +75,7 @@ fi
75
76 chmod +x "$BIN_PATH"
77
78 -set -- "$BIN_PATH" --relay "$RELAYS" --host "${APP_HOST:-localhost:3000}"
78 +set -- "$BIN_PATH" --relays "$RELAYS" --host "${APP_HOST:-localhost:3000}"
79 [ -n "${APP_NAME:-}" ] && set -- "$@" --name "$APP_NAME"
80 [ -n "${APP_DESCRIPTION:-}" ] && set -- "$@" --description "$APP_DESCRIPTION"
81 [ -n "${APP_TAGS:-}" ] && set -- "$@" --tags "$APP_TAGS"
@@ -92,7 +92,7 @@ exec "$@"
92 const tunnelPowerShellScriptTemplate = `$ErrorActionPreference = "Stop"
93
94 $BaseUrl = if ($env:BASE_URL) { $env:BASE_URL } else { "%s" }
95 -$RelayUrl = if ($env:RELAY_URL) { $env:RELAY_URL } else { $BaseUrl }
95 +$RelayUrls = if ($env:RELAYS) { $env:RELAYS } else { $BaseUrl }
96 $OriginalSecurityProtocol = [System.Net.ServicePointManager]::SecurityProtocol
97 [System.Net.ServicePointManager]::SecurityProtocol = [System.Net.SecurityProtocolType]::Tls12
98
@@ -149,7 +149,7 @@ if ($ActualHash -ne $ExpectedHash) {
149 exit 1
150 }
151
152 -$ArgsList = @("--relay", $RelayUrl)
152 +$ArgsList = @("--relays", $RelayUrls)
153
154 if ($env:HOST) { $ArgsList += "--host", $env:HOST } else { $ArgsList += "--host", "localhost:3000" }
155 if ($env:NAME) { $ArgsList += "--name", $env:NAME }