From 4d907bc49da94251f46b365c08e743c5063f1312 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=96=E7=95=8C?= Date: Sat, 14 Mar 2026 18:06:40 +0800 Subject: [PATCH] ccm,ocm: allow URL-based credentials to accept reverse connections Previously, findReceiverCredential required baseURL == reverseProxyBaseURL, so only credentials with no URL could accept incoming reverse connections. Now credentials with a normal URL also accept reverse connections, preferring the reverse session when active and falling back to the direct URL when not. --- service/ccm/credential_external.go | 41 +++++++++++++++++++--- service/ccm/reverse.go | 4 +-- service/ocm/credential_external.go | 55 +++++++++++++++++++++++++++--- service/ocm/reverse.go | 4 +-- 4 files changed, 90 insertions(+), 14 deletions(-) diff --git a/service/ccm/credential_external.go b/service/ccm/credential_external.go index ade2d8361..8d1c1a08d 100644 --- a/service/ccm/credential_external.go +++ b/service/ccm/credential_external.go @@ -47,8 +47,9 @@ type externalCredential struct { requestAccess sync.Mutex // Reverse proxy fields - reverse bool - reverseSession *yamux.Session + reverse bool + reverseHttpClient *http.Client + reverseSession *yamux.Session reverseAccess sync.RWMutex closed bool reverseContext context.Context @@ -194,6 +195,14 @@ func newExternalCredential(ctx context.Context, tag string, options option.CCMEx } else { // Normal mode: standard HTTP client for proxying cred.httpClient = &http.Client{Transport: transport} + cred.reverseHttpClient = &http.Client{ + Transport: &http.Transport{ + ForceAttemptHTTP2: false, + DialContext: func(ctx context.Context, network, address string) (net.Conn, error) { + return cred.openReverseConnection(ctx) + }, + }, + } } } @@ -341,7 +350,14 @@ func (c *externalCredential) getAccessToken() (string, error) { } func (c *externalCredential) buildProxyRequest(ctx context.Context, original *http.Request, bodyBytes []byte, _ http.Header) (*http.Request, error) { - proxyURL := c.baseURL + original.URL.RequestURI() + baseURL := c.baseURL + if c.reverseHttpClient != nil { + session := c.getReverseSession() + if session != nil && !session.IsClosed() { + baseURL = reverseProxyBaseURL + } + } + proxyURL := baseURL + original.URL.RequestURI() var body io.Reader if bodyBytes != nil { body = bytes.NewReader(bodyBytes) @@ -489,9 +505,18 @@ func (c *externalCredential) pollUsage(ctx context.Context) { defer c.pollAccess.Unlock() defer c.markUsagePollAttempted() - statusURL := c.baseURL + "/ccm/v1/status" + activeBaseURL := c.baseURL + activeTransport := c.httpClient.Transport + if c.reverseHttpClient != nil { + session := c.getReverseSession() + if session != nil && !session.IsClosed() { + activeBaseURL = reverseProxyBaseURL + activeTransport = c.reverseHttpClient.Transport + } + } + statusURL := activeBaseURL + "/ccm/v1/status" httpClient := &http.Client{ - Transport: c.httpClient.Transport, + Transport: activeTransport, Timeout: 5 * time.Second, } @@ -602,6 +627,12 @@ func (c *externalCredential) usageTrackerOrNil() *AggregatedUsage { } func (c *externalCredential) httpTransport() *http.Client { + if c.reverseHttpClient != nil { + session := c.getReverseSession() + if session != nil && !session.IsClosed() { + return c.reverseHttpClient + } + } return c.httpClient } diff --git a/service/ccm/reverse.go b/service/ccm/reverse.go index 62a101117..6ecc224f9 100644 --- a/service/ccm/reverse.go +++ b/service/ccm/reverse.go @@ -126,10 +126,10 @@ func (s *Service) handleReverseConnect(ctx context.Context, w http.ResponseWrite func (s *Service) findReceiverCredential(token string) *externalCredential { for _, cred := range s.allCredentials { extCred, ok := cred.(*externalCredential) - if !ok { + if !ok || extCred.connectorURL != nil { continue } - if extCred.baseURL == reverseProxyBaseURL && extCred.token == token { + if extCred.token == token { return extCred } } diff --git a/service/ocm/credential_external.go b/service/ocm/credential_external.go index 3d46f8fa8..f09716db3 100644 --- a/service/ocm/credential_external.go +++ b/service/ocm/credential_external.go @@ -49,8 +49,10 @@ type externalCredential struct { requestAccess sync.Mutex // Reverse proxy fields - reverse bool - reverseSession *yamux.Session + reverse bool + reverseHttpClient *http.Client + reverseCredDialer N.Dialer + reverseSession *yamux.Session reverseAccess sync.RWMutex closed bool reverseContext context.Context @@ -213,6 +215,15 @@ func newExternalCredential(ctx context.Context, tag string, options option.OCMEx // Normal mode: standard HTTP client for proxying cred.credDialer = credentialDialer cred.httpClient = &http.Client{Transport: transport} + cred.reverseCredDialer = reverseSessionDialer{credential: cred} + cred.reverseHttpClient = &http.Client{ + Transport: &http.Transport{ + ForceAttemptHTTP2: false, + DialContext: func(ctx context.Context, network, address string) (net.Conn, error) { + return cred.openReverseConnection(ctx) + }, + }, + } } } @@ -363,7 +374,14 @@ func (c *externalCredential) getAccessToken() (string, error) { } func (c *externalCredential) buildProxyRequest(ctx context.Context, original *http.Request, bodyBytes []byte, _ http.Header) (*http.Request, error) { - proxyURL := c.baseURL + original.URL.RequestURI() + baseURL := c.baseURL + if c.reverseHttpClient != nil { + session := c.getReverseSession() + if session != nil && !session.IsClosed() { + baseURL = reverseProxyBaseURL + } + } + proxyURL := baseURL + original.URL.RequestURI() var body io.Reader if bodyBytes != nil { body = bytes.NewReader(bodyBytes) @@ -526,9 +544,18 @@ func (c *externalCredential) pollUsage(ctx context.Context) { defer c.pollAccess.Unlock() defer c.markUsagePollAttempted() - statusURL := c.baseURL + "/ocm/v1/status" + activeBaseURL := c.baseURL + activeTransport := c.httpClient.Transport + if c.reverseHttpClient != nil { + session := c.getReverseSession() + if session != nil && !session.IsClosed() { + activeBaseURL = reverseProxyBaseURL + activeTransport = c.reverseHttpClient.Transport + } + } + statusURL := activeBaseURL + "/ocm/v1/status" httpClient := &http.Client{ - Transport: c.httpClient.Transport, + Transport: activeTransport, Timeout: 5 * time.Second, } @@ -639,10 +666,22 @@ func (c *externalCredential) usageTrackerOrNil() *AggregatedUsage { } func (c *externalCredential) httpTransport() *http.Client { + if c.reverseHttpClient != nil { + session := c.getReverseSession() + if session != nil && !session.IsClosed() { + return c.reverseHttpClient + } + } return c.httpClient } func (c *externalCredential) ocmDialer() N.Dialer { + if c.reverseCredDialer != nil { + session := c.getReverseSession() + if session != nil && !session.IsClosed() { + return c.reverseCredDialer + } + } return c.credDialer } @@ -655,6 +694,12 @@ func (c *externalCredential) ocmGetAccountID() string { } func (c *externalCredential) ocmGetBaseURL() string { + if c.reverseHttpClient != nil { + session := c.getReverseSession() + if session != nil && !session.IsClosed() { + return reverseProxyBaseURL + } + } return c.baseURL } diff --git a/service/ocm/reverse.go b/service/ocm/reverse.go index 1ed274f6d..ab99c77a6 100644 --- a/service/ocm/reverse.go +++ b/service/ocm/reverse.go @@ -126,10 +126,10 @@ func (s *Service) handleReverseConnect(ctx context.Context, w http.ResponseWrite func (s *Service) findReceiverCredential(token string) *externalCredential { for _, cred := range s.allCredentials { extCred, ok := cred.(*externalCredential) - if !ok { + if !ok || extCred.connectorURL != nil { continue } - if extCred.baseURL == reverseProxyBaseURL && extCred.token == token { + if extCred.token == token { return extCred } }