feat(auth): client-side token exchange for admin API calls
- commit
- 76ac5f425d9c6eb3cd6201e3d0d025c8ddefcf92
- author
- brice · · 2026-03-22 19:49 UTC
- parent
- 589c4c3
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
1 file · +165 -13
M internal/adapters/in/cli/remote/client.go +165 -13
| @@ -15,6 +15,8 @@ import ( | |||
| 15 | 15 | "strings" | |
| 16 | 16 | "time" | |
| 17 | 17 | ||
| 18 | + | "github.com/golang-jwt/jwt/v5" | |
| 19 | + | ||
| 18 | 20 | "github.com/bnema/gordon/internal/adapters/dto" | |
| 19 | 21 | "github.com/bnema/gordon/internal/domain" | |
| 20 | 22 | ) | |
| @@ -26,6 +28,11 @@ type Client struct { | |||
| 26 | 28 | httpClient *http.Client | |
| 27 | 29 | insecureTLS bool | |
| 28 | 30 | onTokenRefreshed func(newToken string) // optional callback to persist a refreshed token | |
| 31 | + | ||
| 32 | + | // Ephemeral admin token exchange fields. | |
| 33 | + | subject string // JWT subject extracted from long-lived token | |
| 34 | + | ephemeral string // cached ephemeral admin token | |
| 35 | + | ephemeralExp time.Time // expiry of cached ephemeral token | |
| 29 | 36 | } | |
| 30 | 37 | ||
| 31 | 38 | var ( | |
| @@ -60,10 +67,31 @@ func NewClient(baseURL string, opts ...ClientOption) *Client { | |||
| 60 | 67 | } | |
| 61 | 68 | ||
| 62 | 69 | // WithToken sets the authentication token. | |
| 70 | + | // If the token is a valid JWT, the subject claim is extracted for use | |
| 71 | + | // in ephemeral token exchange. If parsing fails the subject is left | |
| 72 | + | // empty and the token will be sent directly (backward compat). | |
| 63 | 73 | func WithToken(token string) ClientOption { | |
| 64 | 74 | return func(c *Client) { | |
| 65 | 75 | c.token = token | |
| 76 | + | c.subject = extractJWTSubject(token) | |
| 77 | + | } | |
| 78 | + | } | |
| 79 | + | ||
| 80 | + | // extractJWTSubject parses a JWT without verification and returns the | |
| 81 | + | // "sub" claim. Returns "" if the token is not a valid JWT or has no sub. | |
| 82 | + | func extractJWTSubject(tokenStr string) string { | |
| 83 | + | if tokenStr == "" { | |
| 84 | + | return "" | |
| 66 | 85 | } | |
| 86 | + | parser := jwt.NewParser() | |
| 87 | + | claims := jwt.MapClaims{} | |
| 88 | + | // We only need the claims; signature verification happens server-side. | |
| 89 | + | _, _, err := parser.ParseUnverified(tokenStr, claims) | |
| 90 | + | if err != nil { | |
| 91 | + | return "" | |
| 92 | + | } | |
| 93 | + | sub, _ := claims.GetSubject() | |
| 94 | + | return sub | |
| 67 | 95 | } | |
| 68 | 96 | ||
| 69 | 97 | // WithHTTPClient sets a custom HTTP client. | |
| @@ -118,6 +146,82 @@ func (c *Client) observeTokenRotation(resp *http.Response) { | |||
| 118 | 146 | } | |
| 119 | 147 | } | |
| 120 | 148 | ||
| 149 | + | // ephemeralMargin is the safety margin before expiry at which | |
| 150 | + | // the cached ephemeral token is considered stale. | |
| 151 | + | const ephemeralMargin = 30 * time.Second | |
| 152 | + | ||
| 153 | + | // ephemeralValid reports whether the cached ephemeral token is still usable. | |
| 154 | + | func (c *Client) ephemeralValid() bool { | |
| 155 | + | return c.ephemeral != "" && time.Now().Before(c.ephemeralExp.Add(-ephemeralMargin)) | |
| 156 | + | } | |
| 157 | + | ||
| 158 | + | // exchangeToken exchanges the long-lived token for a short-lived ephemeral | |
| 159 | + | // admin token via /auth/token. Follows the same pattern as ExchangeRegistryToken. | |
| 160 | + | func (c *Client) exchangeToken() error { | |
| 161 | + | url := c.baseURL + "/auth/token?scope=admin:*:*&service=gordon-registry" | |
| 162 | + | ||
| 163 | + | req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, url, nil) | |
| 164 | + | if err != nil { | |
| 165 | + | return fmt.Errorf("ephemeral token exchange: %w", err) | |
| 166 | + | } | |
| 167 | + | ||
| 168 | + | req.SetBasicAuth(c.subject, c.token) | |
| 169 | + | req.Header.Set("Accept", "application/json") | |
| 170 | + | ||
| 171 | + | resp, err := c.httpClient.Do(req) | |
| 172 | + | if err != nil { | |
| 173 | + | return fmt.Errorf("ephemeral token exchange: %w", err) | |
| 174 | + | } | |
| 175 | + | defer resp.Body.Close() | |
| 176 | + | ||
| 177 | + | if resp.StatusCode >= 400 { | |
| 178 | + | body, _ := io.ReadAll(io.LimitReader(resp.Body, maxErrorBodySize)) | |
| 179 | + | var errResp struct { | |
| 180 | + | Error string `json:"error"` | |
| 181 | + | } | |
| 182 | + | if err := json.Unmarshal(body, &errResp); err == nil && errResp.Error != "" { | |
| 183 | + | return fmt.Errorf("ephemeral token exchange: %s: %s", resp.Status, errResp.Error) | |
| 184 | + | } | |
| 185 | + | return fmt.Errorf("ephemeral token exchange: %s: %s", resp.Status, string(body)) | |
| 186 | + | } | |
| 187 | + | ||
| 188 | + | var result dto.TokenResponse | |
| 189 | + | if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { | |
| 190 | + | return fmt.Errorf("ephemeral token exchange: decode: %w", err) | |
| 191 | + | } | |
| 192 | + | ||
| 193 | + | token := result.Token | |
| 194 | + | if token == "" { | |
| 195 | + | token = result.AccessToken | |
| 196 | + | } | |
| 197 | + | if token == "" { | |
| 198 | + | return fmt.Errorf("ephemeral token exchange returned empty token") | |
| 199 | + | } | |
| 200 | + | ||
| 201 | + | c.ephemeral = token | |
| 202 | + | c.ephemeralExp = time.Now().Add(time.Duration(result.ExpiresIn) * time.Second) | |
| 203 | + | return nil | |
| 204 | + | } | |
| 205 | + | ||
| 206 | + | // bearerToken returns the token to use in Authorization headers. | |
| 207 | + | // If the client has a subject (valid JWT), it exchanges for an ephemeral token. | |
| 208 | + | // Otherwise it falls back to sending the long-lived token directly. | |
| 209 | + | func (c *Client) bearerToken() (string, error) { | |
| 210 | + | if c.token == "" { | |
| 211 | + | return "", nil | |
| 212 | + | } | |
| 213 | + | // No subject means we couldn't parse the JWT; send token as-is (backward compat). | |
| 214 | + | if c.subject == "" { | |
| 215 | + | return c.token, nil | |
| 216 | + | } | |
| 217 | + | if !c.ephemeralValid() { | |
| 218 | + | if err := c.exchangeToken(); err != nil { | |
| 219 | + | return "", err | |
| 220 | + | } | |
| 221 | + | } | |
| 222 | + | return c.ephemeral, nil | |
| 223 | + | } | |
| 224 | + | ||
| 121 | 225 | func (c *Client) applyTLSConfig() { | |
| 122 | 226 | if !c.insecureTLS { | |
| 123 | 227 | return | |
| @@ -151,18 +255,48 @@ func (c *Client) applyTLSConfig() { | |||
| 151 | 255 | ||
| 152 | 256 | // request performs an HTTP request to the admin API. | |
| 153 | 257 | func (c *Client) request(ctx context.Context, method, path string, body any) (*http.Response, error) { | |
| 154 | - | url := c.baseURL + "/admin" + path | |
| 155 | - | ||
| 156 | - | var bodyReader io.Reader | |
| 258 | + | // Marshal body once so it can be replayed on 401 retry. | |
| 259 | + | var jsonBody []byte | |
| 157 | 260 | if body != nil { | |
| 158 | - | jsonBody, err := json.Marshal(body) | |
| 261 | + | var err error | |
| 262 | + | jsonBody, err = json.Marshal(body) | |
| 159 | 263 | if err != nil { | |
| 160 | 264 | return nil, fmt.Errorf("failed to marshal request body: %w", err) | |
| 161 | 265 | } | |
| 266 | + | } | |
| 267 | + | ||
| 268 | + | resp, err := c.doRequest(ctx, method, path, jsonBody) | |
| 269 | + | if err != nil { | |
| 270 | + | return nil, err | |
| 271 | + | } | |
| 272 | + | ||
| 273 | + | // On 401: invalidate ephemeral, re-exchange, retry once. | |
| 274 | + | if resp.StatusCode == http.StatusUnauthorized && c.subject != "" && c.ephemeral != "" { | |
| 275 | + | // Drain and close the first response. | |
| 276 | + | _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxErrorBodySize)) | |
| 277 | + | resp.Body.Close() | |
| 278 | + | ||
| 279 | + | c.ephemeral = "" | |
| 280 | + | c.ephemeralExp = time.Time{} | |
| 281 | + | if exErr := c.exchangeToken(); exErr != nil { | |
| 282 | + | return nil, fmt.Errorf("token re-exchange after 401: %w", exErr) | |
| 283 | + | } | |
| 284 | + | return c.doRequest(ctx, method, path, jsonBody) | |
| 285 | + | } | |
| 286 | + | ||
| 287 | + | return resp, nil | |
| 288 | + | } | |
| 289 | + | ||
| 290 | + | // doRequest builds and executes a single HTTP request to the admin API. | |
| 291 | + | func (c *Client) doRequest(ctx context.Context, method, path string, jsonBody []byte) (*http.Response, error) { | |
| 292 | + | reqURL := c.baseURL + "/admin" + path | |
| 293 | + | ||
| 294 | + | var bodyReader io.Reader | |
| 295 | + | if jsonBody != nil { | |
| 162 | 296 | bodyReader = bytes.NewReader(jsonBody) | |
| 163 | 297 | } | |
| 164 | 298 | ||
| 165 | - | req, err := http.NewRequestWithContext(ctx, method, url, bodyReader) | |
| 299 | + | req, err := http.NewRequestWithContext(ctx, method, reqURL, bodyReader) | |
| 166 | 300 | if err != nil { | |
| 167 | 301 | return nil, fmt.Errorf("failed to create request: %w", err) | |
| 168 | 302 | } | |
| @@ -170,8 +304,12 @@ func (c *Client) request(ctx context.Context, method, path string, body any) (*h | |||
| 170 | 304 | req.Header.Set("Content-Type", "application/json") | |
| 171 | 305 | req.Header.Set("Accept", "application/json") | |
| 172 | 306 | ||
| 173 | - | if c.token != "" { | |
| 174 | - | req.Header.Set("Authorization", "Bearer "+c.token) | |
| 307 | + | bearer, err := c.bearerToken() | |
| 308 | + | if err != nil { | |
| 309 | + | return nil, err | |
| 310 | + | } | |
| 311 | + | if bearer != "" { | |
| 312 | + | req.Header.Set("Authorization", "Bearer "+bearer) | |
| 175 | 313 | } | |
| 176 | 314 | ||
| 177 | 315 | resp, err := c.httpClient.Do(req) | |
| @@ -999,18 +1137,22 @@ func (c *Client) RemoveAutoRouteAllowedDomain(ctx context.Context, pattern strin | |||
| 999 | 1137 | return parseResponse(resp, nil) | |
| 1000 | 1138 | } | |
| 1001 | 1139 | ||
| 1002 | - | // streamLogs handles SSE streaming for log endpoints. | |
| 1003 | - | func (c *Client) streamLogs(ctx context.Context, path string) (<-chan string, error) { | |
| 1004 | - | url := c.baseURL + "/admin" + path | |
| 1140 | + | // openSSEStream opens an SSE connection to the given admin path and returns the response. | |
| 1141 | + | func (c *Client) openSSEStream(ctx context.Context, path string) (*http.Response, error) { | |
| 1142 | + | streamURL := c.baseURL + "/admin" + path | |
| 1005 | 1143 | ||
| 1006 | - | req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) | |
| 1144 | + | req, err := http.NewRequestWithContext(ctx, http.MethodGet, streamURL, nil) | |
| 1007 | 1145 | if err != nil { | |
| 1008 | 1146 | return nil, fmt.Errorf("failed to create request: %w", err) | |
| 1009 | 1147 | } | |
| 1010 | 1148 | ||
| 1011 | 1149 | req.Header.Set("Accept", "text/event-stream") | |
| 1012 | - | if c.token != "" { | |
| 1013 | - | req.Header.Set("Authorization", "Bearer "+c.token) | |
| 1150 | + | bearer, err := c.bearerToken() | |
| 1151 | + | if err != nil { | |
| 1152 | + | return nil, err | |
| 1153 | + | } | |
| 1154 | + | if bearer != "" { | |
| 1155 | + | req.Header.Set("Authorization", "Bearer "+bearer) | |
| 1014 | 1156 | } | |
| 1015 | 1157 | ||
| 1016 | 1158 | // Use the same transport as the main client (honoring TLS config and | |
| @@ -1032,6 +1174,16 @@ func (c *Client) streamLogs(ctx context.Context, path string) (<-chan string, er | |||
| 1032 | 1174 | return nil, fmt.Errorf("%s: %s", resp.Status, string(body)) | |
| 1033 | 1175 | } | |
| 1034 | 1176 | ||
| 1177 | + | return resp, nil | |
| 1178 | + | } | |
| 1179 | + | ||
| 1180 | + | // streamLogs handles SSE streaming for log endpoints. | |
| 1181 | + | func (c *Client) streamLogs(ctx context.Context, path string) (<-chan string, error) { | |
| 1182 | + | resp, err := c.openSSEStream(ctx, path) | |
| 1183 | + | if err != nil { | |
| 1184 | + | return nil, err | |
| 1185 | + | } | |
| 1186 | + | ||
| 1035 | 1187 | ch := make(chan string, 100) | |
| 1036 | 1188 | ||
| 1037 | 1189 | go func() { | |
| @@ -15,6 +15,8 @@ import ( | |||||
| 15 | "strings" | 15 | "strings" | ||
| 16 | "time" | 16 | "time" | ||
| 17 | 17 | ||||
| 18 | + | "github.com/golang-jwt/jwt/v5" | |||
| 19 | + | ||||
| 18 | "github.com/bnema/gordon/internal/adapters/dto" | 20 | "github.com/bnema/gordon/internal/adapters/dto" | ||
| 19 | "github.com/bnema/gordon/internal/domain" | 21 | "github.com/bnema/gordon/internal/domain" | ||
| 20 | ) | 22 | ) | ||
| @@ -26,6 +28,11 @@ type Client struct { | |||||
| 26 | httpClient *http.Client | 28 | httpClient *http.Client | ||
| 27 | insecureTLS bool | 29 | insecureTLS bool | ||
| 28 | onTokenRefreshed func(newToken string) // optional callback to persist a refreshed token | 30 | onTokenRefreshed func(newToken string) // optional callback to persist a refreshed token | ||
| 31 | + | ||||
| 32 | + | // Ephemeral admin token exchange fields. | |||
| 33 | + | subject string // JWT subject extracted from long-lived token | |||
| 34 | + | ephemeral string // cached ephemeral admin token | |||
| 35 | + | ephemeralExp time.Time // expiry of cached ephemeral token | |||
| 29 | } | 36 | } | ||
| 30 | 37 | ||||
| 31 | var ( | 38 | var ( | ||
| @@ -60,10 +67,31 @@ func NewClient(baseURL string, opts ...ClientOption) *Client { | |||||
| 60 | } | 67 | } | ||
| 61 | 68 | ||||
| 62 | // WithToken sets the authentication token. | 69 | // WithToken sets the authentication token. | ||
| 70 | + | // If the token is a valid JWT, the subject claim is extracted for use | |||
| 71 | + | // in ephemeral token exchange. If parsing fails the subject is left | |||
| 72 | + | // empty and the token will be sent directly (backward compat). | |||
| 63 | func WithToken(token string) ClientOption { | 73 | func WithToken(token string) ClientOption { | ||
| 64 | return func(c *Client) { | 74 | return func(c *Client) { | ||
| 65 | c.token = token | 75 | c.token = token | ||
| 76 | + | c.subject = extractJWTSubject(token) | |||
| 77 | + | } | |||
| 78 | + | } | |||
| 79 | + | ||||
| 80 | + | // extractJWTSubject parses a JWT without verification and returns the | |||
| 81 | + | // "sub" claim. Returns "" if the token is not a valid JWT or has no sub. | |||
| 82 | + | func extractJWTSubject(tokenStr string) string { | |||
| 83 | + | if tokenStr == "" { | |||
| 84 | + | return "" | |||
| 66 | } | 85 | } | ||
| 86 | + | parser := jwt.NewParser() | |||
| 87 | + | claims := jwt.MapClaims{} | |||
| 88 | + | // We only need the claims; signature verification happens server-side. | |||
| 89 | + | _, _, err := parser.ParseUnverified(tokenStr, claims) | |||
| 90 | + | if err != nil { | |||
| 91 | + | return "" | |||
| 92 | + | } | |||
| 93 | + | sub, _ := claims.GetSubject() | |||
| 94 | + | return sub | |||
| 67 | } | 95 | } | ||
| 68 | 96 | ||||
| 69 | // WithHTTPClient sets a custom HTTP client. | 97 | // WithHTTPClient sets a custom HTTP client. | ||
| @@ -118,6 +146,82 @@ func (c *Client) observeTokenRotation(resp *http.Response) { | |||||
| 118 | } | 146 | } | ||
| 119 | } | 147 | } | ||
| 120 | 148 | ||||
| 149 | + | // ephemeralMargin is the safety margin before expiry at which | |||
| 150 | + | // the cached ephemeral token is considered stale. | |||
| 151 | + | const ephemeralMargin = 30 * time.Second | |||
| 152 | + | ||||
| 153 | + | // ephemeralValid reports whether the cached ephemeral token is still usable. | |||
| 154 | + | func (c *Client) ephemeralValid() bool { | |||
| 155 | + | return c.ephemeral != "" && time.Now().Before(c.ephemeralExp.Add(-ephemeralMargin)) | |||
| 156 | + | } | |||
| 157 | + | ||||
| 158 | + | // exchangeToken exchanges the long-lived token for a short-lived ephemeral | |||
| 159 | + | // admin token via /auth/token. Follows the same pattern as ExchangeRegistryToken. | |||
| 160 | + | func (c *Client) exchangeToken() error { | |||
| 161 | + | url := c.baseURL + "/auth/token?scope=admin:*:*&service=gordon-registry" | |||
| 162 | + | ||||
| 163 | + | req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, url, nil) | |||
| 164 | + | if err != nil { | |||
| 165 | + | return fmt.Errorf("ephemeral token exchange: %w", err) | |||
| 166 | + | } | |||
| 167 | + | ||||
| 168 | + | req.SetBasicAuth(c.subject, c.token) | |||
| 169 | + | req.Header.Set("Accept", "application/json") | |||
| 170 | + | ||||
| 171 | + | resp, err := c.httpClient.Do(req) | |||
| 172 | + | if err != nil { | |||
| 173 | + | return fmt.Errorf("ephemeral token exchange: %w", err) | |||
| 174 | + | } | |||
| 175 | + | defer resp.Body.Close() | |||
| 176 | + | ||||
| 177 | + | if resp.StatusCode >= 400 { | |||
| 178 | + | body, _ := io.ReadAll(io.LimitReader(resp.Body, maxErrorBodySize)) | |||
| 179 | + | var errResp struct { | |||
| 180 | + | Error string `json:"error"` | |||
| 181 | + | } | |||
| 182 | + | if err := json.Unmarshal(body, &errResp); err == nil && errResp.Error != "" { | |||
| 183 | + | return fmt.Errorf("ephemeral token exchange: %s: %s", resp.Status, errResp.Error) | |||
| 184 | + | } | |||
| 185 | + | return fmt.Errorf("ephemeral token exchange: %s: %s", resp.Status, string(body)) | |||
| 186 | + | } | |||
| 187 | + | ||||
| 188 | + | var result dto.TokenResponse | |||
| 189 | + | if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { | |||
| 190 | + | return fmt.Errorf("ephemeral token exchange: decode: %w", err) | |||
| 191 | + | } | |||
| 192 | + | ||||
| 193 | + | token := result.Token | |||
| 194 | + | if token == "" { | |||
| 195 | + | token = result.AccessToken | |||
| 196 | + | } | |||
| 197 | + | if token == "" { | |||
| 198 | + | return fmt.Errorf("ephemeral token exchange returned empty token") | |||
| 199 | + | } | |||
| 200 | + | ||||
| 201 | + | c.ephemeral = token | |||
| 202 | + | c.ephemeralExp = time.Now().Add(time.Duration(result.ExpiresIn) * time.Second) | |||
| 203 | + | return nil | |||
| 204 | + | } | |||
| 205 | + | ||||
| 206 | + | // bearerToken returns the token to use in Authorization headers. | |||
| 207 | + | // If the client has a subject (valid JWT), it exchanges for an ephemeral token. | |||
| 208 | + | // Otherwise it falls back to sending the long-lived token directly. | |||
| 209 | + | func (c *Client) bearerToken() (string, error) { | |||
| 210 | + | if c.token == "" { | |||
| 211 | + | return "", nil | |||
| 212 | + | } | |||
| 213 | + | // No subject means we couldn't parse the JWT; send token as-is (backward compat). | |||
| 214 | + | if c.subject == "" { | |||
| 215 | + | return c.token, nil | |||
| 216 | + | } | |||
| 217 | + | if !c.ephemeralValid() { | |||
| 218 | + | if err := c.exchangeToken(); err != nil { | |||
| 219 | + | return "", err | |||
| 220 | + | } | |||
| 221 | + | } | |||
| 222 | + | return c.ephemeral, nil | |||
| 223 | + | } | |||
| 224 | + | ||||
| 121 | func (c *Client) applyTLSConfig() { | 225 | func (c *Client) applyTLSConfig() { | ||
| 122 | if !c.insecureTLS { | 226 | if !c.insecureTLS { | ||
| 123 | return | 227 | return | ||
| @@ -151,18 +255,48 @@ func (c *Client) applyTLSConfig() { | |||||
| 151 | 255 | ||||
| 152 | // request performs an HTTP request to the admin API. | 256 | // request performs an HTTP request to the admin API. | ||
| 153 | func (c *Client) request(ctx context.Context, method, path string, body any) (*http.Response, error) { | 257 | func (c *Client) request(ctx context.Context, method, path string, body any) (*http.Response, error) { | ||
| 154 | - | url := c.baseURL + "/admin" + path | 258 | + | // Marshal body once so it can be replayed on 401 retry. |
| 155 | - | 259 | + | var jsonBody []byte | |
| 156 | - | var bodyReader io.Reader | |||
| 157 | if body != nil { | 260 | if body != nil { | ||
| 158 | - | jsonBody, err := json.Marshal(body) | 261 | + | var err error |
| 262 | + | jsonBody, err = json.Marshal(body) | |||
| 159 | if err != nil { | 263 | if err != nil { | ||
| 160 | return nil, fmt.Errorf("failed to marshal request body: %w", err) | 264 | return nil, fmt.Errorf("failed to marshal request body: %w", err) | ||
| 161 | } | 265 | } | ||
| 266 | + | } | |||
| 267 | + | ||||
| 268 | + | resp, err := c.doRequest(ctx, method, path, jsonBody) | |||
| 269 | + | if err != nil { | |||
| 270 | + | return nil, err | |||
| 271 | + | } | |||
| 272 | + | ||||
| 273 | + | // On 401: invalidate ephemeral, re-exchange, retry once. | |||
| 274 | + | if resp.StatusCode == http.StatusUnauthorized && c.subject != "" && c.ephemeral != "" { | |||
| 275 | + | // Drain and close the first response. | |||
| 276 | + | _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxErrorBodySize)) | |||
| 277 | + | resp.Body.Close() | |||
| 278 | + | ||||
| 279 | + | c.ephemeral = "" | |||
| 280 | + | c.ephemeralExp = time.Time{} | |||
| 281 | + | if exErr := c.exchangeToken(); exErr != nil { | |||
| 282 | + | return nil, fmt.Errorf("token re-exchange after 401: %w", exErr) | |||
| 283 | + | } | |||
| 284 | + | return c.doRequest(ctx, method, path, jsonBody) | |||
| 285 | + | } | |||
| 286 | + | ||||
| 287 | + | return resp, nil | |||
| 288 | + | } | |||
| 289 | + | ||||
| 290 | + | // doRequest builds and executes a single HTTP request to the admin API. | |||
| 291 | + | func (c *Client) doRequest(ctx context.Context, method, path string, jsonBody []byte) (*http.Response, error) { | |||
| 292 | + | reqURL := c.baseURL + "/admin" + path | |||
| 293 | + | ||||
| 294 | + | var bodyReader io.Reader | |||
| 295 | + | if jsonBody != nil { | |||
| 162 | bodyReader = bytes.NewReader(jsonBody) | 296 | bodyReader = bytes.NewReader(jsonBody) | ||
| 163 | } | 297 | } | ||
| 164 | 298 | ||||
| 165 | - | req, err := http.NewRequestWithContext(ctx, method, url, bodyReader) | 299 | + | req, err := http.NewRequestWithContext(ctx, method, reqURL, bodyReader) |
| 166 | if err != nil { | 300 | if err != nil { | ||
| 167 | return nil, fmt.Errorf("failed to create request: %w", err) | 301 | return nil, fmt.Errorf("failed to create request: %w", err) | ||
| 168 | } | 302 | } | ||
| @@ -170,8 +304,12 @@ func (c *Client) request(ctx context.Context, method, path string, body any) (*h | |||||
| 170 | req.Header.Set("Content-Type", "application/json") | 304 | req.Header.Set("Content-Type", "application/json") | ||
| 171 | req.Header.Set("Accept", "application/json") | 305 | req.Header.Set("Accept", "application/json") | ||
| 172 | 306 | ||||
| 173 | - | if c.token != "" { | 307 | + | bearer, err := c.bearerToken() |
| 174 | - | req.Header.Set("Authorization", "Bearer "+c.token) | 308 | + | if err != nil { |
| 309 | + | return nil, err | |||
| 310 | + | } | |||
| 311 | + | if bearer != "" { | |||
| 312 | + | req.Header.Set("Authorization", "Bearer "+bearer) | |||
| 175 | } | 313 | } | ||
| 176 | 314 | ||||
| 177 | resp, err := c.httpClient.Do(req) | 315 | resp, err := c.httpClient.Do(req) | ||
| @@ -999,18 +1137,22 @@ func (c *Client) RemoveAutoRouteAllowedDomain(ctx context.Context, pattern strin | |||||
| 999 | return parseResponse(resp, nil) | 1137 | return parseResponse(resp, nil) | ||
| 1000 | } | 1138 | } | ||
| 1001 | 1139 | ||||
| 1002 | - | // streamLogs handles SSE streaming for log endpoints. | 1140 | + | // openSSEStream opens an SSE connection to the given admin path and returns the response. |
| 1003 | - | func (c *Client) streamLogs(ctx context.Context, path string) (<-chan string, error) { | 1141 | + | func (c *Client) openSSEStream(ctx context.Context, path string) (*http.Response, error) { |
| 1004 | - | url := c.baseURL + "/admin" + path | 1142 | + | streamURL := c.baseURL + "/admin" + path |
| 1005 | 1143 | ||||
| 1006 | - | req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) | 1144 | + | req, err := http.NewRequestWithContext(ctx, http.MethodGet, streamURL, nil) |
| 1007 | if err != nil { | 1145 | if err != nil { | ||
| 1008 | return nil, fmt.Errorf("failed to create request: %w", err) | 1146 | return nil, fmt.Errorf("failed to create request: %w", err) | ||
| 1009 | } | 1147 | } | ||
| 1010 | 1148 | ||||
| 1011 | req.Header.Set("Accept", "text/event-stream") | 1149 | req.Header.Set("Accept", "text/event-stream") | ||
| 1012 | - | if c.token != "" { | 1150 | + | bearer, err := c.bearerToken() |
| 1013 | - | req.Header.Set("Authorization", "Bearer "+c.token) | 1151 | + | if err != nil { |
| 1152 | + | return nil, err | |||
| 1153 | + | } | |||
| 1154 | + | if bearer != "" { | |||
| 1155 | + | req.Header.Set("Authorization", "Bearer "+bearer) | |||
| 1014 | } | 1156 | } | ||
| 1015 | 1157 | ||||
| 1016 | // Use the same transport as the main client (honoring TLS config and | 1158 | // Use the same transport as the main client (honoring TLS config and | ||
| @@ -1032,6 +1174,16 @@ func (c *Client) streamLogs(ctx context.Context, path string) (<-chan string, er | |||||
| 1032 | return nil, fmt.Errorf("%s: %s", resp.Status, string(body)) | 1174 | return nil, fmt.Errorf("%s: %s", resp.Status, string(body)) | ||
| 1033 | } | 1175 | } | ||
| 1034 | 1176 | ||||
| 1177 | + | return resp, nil | |||
| 1178 | + | } | |||
| 1179 | + | ||||
| 1180 | + | // streamLogs handles SSE streaming for log endpoints. | |||
| 1181 | + | func (c *Client) streamLogs(ctx context.Context, path string) (<-chan string, error) { | |||
| 1182 | + | resp, err := c.openSSEStream(ctx, path) | |||
| 1183 | + | if err != nil { | |||
| 1184 | + | return nil, err | |||
| 1185 | + | } | |||
| 1186 | + | ||||
| 1035 | ch := make(chan string, 100) | 1187 | ch := make(chan string, 100) | ||
| 1036 | 1188 | ||||
| 1037 | go func() { | 1189 | go func() { | ||