From 358bbdcb06c066e0a056cdce2ce3961751aa240b Mon Sep 17 00:00:00 2001 From: Scot Wells Date: Sun, 12 Jul 2026 20:56:17 -0500 Subject: [PATCH] Add 'datumctl api proxy' for tokenless local API access Co-Authored-By: Claude Fable 5 --- docs/api-proxy.md | 134 +++++++ internal/apiproxy/handler.go | 127 ++++++ internal/apiproxy/logging.go | 160 ++++++++ internal/apiproxy/logging_test.go | 219 +++++++++++ internal/apiproxy/proxy.go | 199 ++++++++++ internal/apiproxy/proxy_test.go | 476 +++++++++++++++++++++++ internal/apiproxy/streaming_test.go | 304 +++++++++++++++ internal/apiproxy/token_test.go | 309 +++++++++++++++ internal/authutil/credentials.go | 72 ++-- internal/authutil/credentials_test.go | 191 +++++++++ internal/authutil/login.go | 2 +- internal/authutil/migrate_test.go | 17 +- internal/authutil/serviceaccount.go | 16 +- internal/client/factory.go | 32 +- internal/client/session_endpoint.go | 119 ++++++ internal/client/session_endpoint_test.go | 224 +++++++++++ internal/cmd/api/api.go | 23 ++ internal/cmd/api/proxy.go | 307 +++++++++++++++ internal/cmd/api/proxy_lifecycle_test.go | 139 +++++++ internal/cmd/api/proxy_test.go | 231 +++++++++++ internal/cmd/auth/auth.go | 2 +- internal/cmd/auth/get_token.go | 2 +- internal/cmd/root.go | 5 + 23 files changed, 3247 insertions(+), 63 deletions(-) create mode 100644 docs/api-proxy.md create mode 100644 internal/apiproxy/handler.go create mode 100644 internal/apiproxy/logging.go create mode 100644 internal/apiproxy/logging_test.go create mode 100644 internal/apiproxy/proxy.go create mode 100644 internal/apiproxy/proxy_test.go create mode 100644 internal/apiproxy/streaming_test.go create mode 100644 internal/apiproxy/token_test.go create mode 100644 internal/authutil/credentials_test.go create mode 100644 internal/client/session_endpoint.go create mode 100644 internal/client/session_endpoint_test.go create mode 100644 internal/cmd/api/api.go create mode 100644 internal/cmd/api/proxy.go create mode 100644 internal/cmd/api/proxy_lifecycle_test.go create mode 100644 internal/cmd/api/proxy_test.go diff --git a/docs/api-proxy.md b/docs/api-proxy.md new file mode 100644 index 0000000..746569f --- /dev/null +++ b/docs/api-proxy.md @@ -0,0 +1,134 @@ +--- +title: "API Proxy" +sidebar: + order: 4 +--- + +`datumctl api proxy` starts a local HTTP proxy that forwards every request to +the Datum Cloud API endpoint of your datumctl session, adding your credentials +automatically and refreshing them as needed. Point any local tool — a dev +server, a test harness, `curl` — at the printed URL and it can talk to the +platform with no tokens to copy and no expiry to manage: + +``` +$ datumctl api proxy --port 8001 +$ curl http://127.0.0.1:8001/apis/resourcemanager.miloapis.com/v1alpha1/organizations +``` + +## Starting the proxy + +``` +# Start a proxy on a fixed port for a dev server +datumctl api proxy --port 8001 + +# Start on a random free port; the URL is printed on the first stdout line +datumctl api proxy + +# Pin a non-active session (see 'datumctl auth list' for names) +datumctl api proxy --session sam@datum.net@api.staging.env.datum.net + +# Suppress per-request log lines +datumctl api proxy --quiet +``` + +By default the proxy picks a random free port, so starting a second proxy +never fails. The bound URL is always printed: + +- A human-readable banner on **stderr** names the session, upstream, scope, + and listen address, so you can verify at a glance whose credentials the + proxy serves. +- The bare proxy URL (for example `http://127.0.0.1:52347`) is printed as the + **first and only line on stdout**, after the listener is serving. Scripts + and test harnesses can read that one line as their readiness signal: + +```go +cmd := exec.Command("datumctl", "api", "proxy", "--quiet", "--project", testProject) +stdout, _ := cmd.StdoutPipe() +_ = cmd.Start() +apiURL, _ := bufio.NewReader(stdout).ReadString('\n') // first line = ready + address +``` + +Press `Ctrl+C` to stop the proxy. In-flight requests get a short grace period +before their connections are closed. + +## Paths and scoping + +By default the proxy is a pure passthrough of the platform API surface: the +same paths that work against the real API endpoint work against the local +port, including the scoped organization/project control-plane prefixes: + +``` +# Watch DNS zones on a project control plane through the proxy +curl "http://127.0.0.1:8001/apis/resourcemanager.miloapis.com/v1alpha1/projects/my-project/control-plane/apis/networking.datumapis.com/v1alpha/dnszones?watch=true" +``` + +With an explicit `--project` or `--organization` flag, the proxy instead +serves that single control plane at its root, so URLs lose the long +control-plane prefix: + +``` +$ datumctl api proxy --port 8001 --project my-project +$ curl "http://127.0.0.1:8001/apis/networking.datumapis.com/v1alpha/dnszones?watch=true" +``` + +The session and the scope are **pinned when the proxy starts** and shown in +the banner. Switching your active account (`datumctl auth switch`) or context +(`datumctl ctx use`) does not affect a running proxy, and the proxy never +inherits a scope from your current context — scoping is always an explicit +flag. Restart the proxy to pick up a new session or scope. + +## Streaming + +Streaming responses — watch requests, server-sent events, chunked transfer — +pass through unbuffered: each event is flushed to your client the moment the +upstream sends it, and response duration is never limited by the proxy. This +makes the proxy suitable for long-lived watch clients. + +## Credentials and token refresh + +Every proxied request carries a fresh access token for the pinned session; +the proxy refreshes the token before expiry and persists refreshed tokens +exactly as other datumctl commands do. Service-account sessions are supported +the same way. + +If the session cannot be refreshed (expired, revoked, or logged out with +`datumctl logout`), the proxy stays up and answers requests with a +synthesized `502 Bad Gateway` carrying a JSON `Status` body, the +`X-Datum-Proxy-Error: true` marker header, and a message telling you to run +`datumctl login`. The proxy recovers automatically — without a restart — once +you log back in to the same session. A `401` or `403` from the platform +itself passes through unchanged, without the marker header, so your client +can always tell a proxy-local authentication problem from a platform answer. + +## Security model + +- **Loopback only.** The proxy listens on `127.0.0.1` and there is no flag to + bind other addresses. Anything that should be reachable remotely deserves a + tunnel whose security model you own. +- **Local clients are trusted.** Like other local developer proxies, the + proxy does not authenticate local clients: while it runs, any process on + your machine can make API calls as the pinned session. The banner names the + identity it serves, requests are logged by default, and the credential + itself is never exposed — a local client can act through the proxy but + cannot take your token with it. +- **Host-header validation.** Requests whose `Host` is not `localhost`, + `127.0.0.1`, or `[::1]` are rejected with `403`, which defeats DNS-rebinding + attacks from web pages. +- **No CORS headers.** Browsers refuse scripted cross-origin reads of proxy + responses; the proxy is meant for server-side and command-line clients. +- **Authorization discipline.** Any `Authorization` header your local client + sends is stripped and replaced with the session's real token, and tokens + never appear in the request log. + +## Request logging + +One line per request is written to stderr (silence with `--quiet`): + +``` +10:42:03 GET /apis/resourcemanager.miloapis.com/v1alpha1/organizations 200 143ms 8.1kB +10:42:05 GET /apis/…/projects/my-project/control-plane/…/portalplugins?watch=true 200 …streaming +``` + +Streaming responses log once when headers arrive (marked `…streaming`) and +again when the stream ends, with total duration and bytes — so an abruptly +closed watch is visible. diff --git a/internal/apiproxy/handler.go b/internal/apiproxy/handler.go new file mode 100644 index 0000000..e8dd57a --- /dev/null +++ b/internal/apiproxy/handler.go @@ -0,0 +1,127 @@ +package apiproxy + +import ( + "encoding/json" + "errors" + "fmt" + "net" + "net/http" + "net/http/httputil" + "net/url" + "strconv" + "strings" + + customerrors "go.datum.net/datumctl/internal/errors" +) + +// proxyErrorHeader marks responses synthesized by the proxy itself, so a +// client can tell a proxy-local failure from something the upstream said. +// Upstream responses — including 401/403 — pass through without it. +const proxyErrorHeader = "X-Datum-Proxy-Error" + +// rewriteTo returns the ReverseProxy Rewrite func: the inbound path and +// query are forwarded verbatim, joined under the upstream root (which may +// carry a control-plane path prefix), with the upstream host as the +// outbound Host header. +func rewriteTo(upstream *url.URL) func(*httputil.ProxyRequest) { + return func(pr *httputil.ProxyRequest) { + pr.SetURL(upstream) + pr.Out.Host = upstream.Host + // Never forward a locally supplied credential upstream — the + // oauth2.Transport injects the real one. Deleting rather than + // overwriting also keeps stale tokens baked into client configs + // from half-working. + pr.Out.Header.Del("Authorization") + // No SetXForwarded: the upstream gains nothing from knowing + // about 127.0.0.1. + } +} + +// hostValidator rejects any request whose Host header is not a local +// address, before the upstream sees anything — the DNS-rebinding defense: a +// malicious page that rebinds its hostname to 127.0.0.1 still sends that +// hostname in Host. +type hostValidator struct { + next http.Handler +} + +func (h *hostValidator) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if !allowedHost(r.Host) { + writeStatus(w, http.StatusForbidden, "Forbidden", + fmt.Sprintf("host %q is not a local address; the datumctl proxy only serves local clients", r.Host)) + return + } + h.next.ServeHTTP(w, r) +} + +// allowedHost reports whether hostport is localhost, 127.0.0.1, or [::1], +// with or without a port. +func allowedHost(hostport string) bool { + host := hostport + if h, _, err := net.SplitHostPort(hostport); err == nil { + host = h + } + host = strings.TrimPrefix(host, "[") + host = strings.TrimSuffix(host, "]") + switch strings.ToLower(host) { + case "localhost", "127.0.0.1", "::1": + return true + } + return false +} + +// handleProxyError synthesizes a 502 for failures that happened proxy-side — +// deliberately not 401, so a client's re-auth logic never misfires on a +// proxy-local problem. Token-source failures and upstream connection +// failures get distinct Status reasons. +func handleProxyError(w http.ResponseWriter, r *http.Request, err error) { + reason := "ProxyUpstreamError" + var refreshErr *tokenRefreshError + if errors.As(err, &refreshErr) { + reason = "ProxyAuthenticationFailed" + } + writeStatus(w, http.StatusBadGateway, reason, errorMessage(err)) +} + +// errorMessage flattens err into a single Status message line, preferring a +// UserError's message and hint so clients see actionable guidance such as +// "run `datumctl login`". +func errorMessage(err error) string { + if userErr, ok := customerrors.IsUserError(err); ok { + if userErr.Hint != "" { + return userErr.Message + " — " + userErr.Hint + } + return userErr.Message + } + return err.Error() +} + +// kubeStatus is the subset of the Kubernetes Status object the proxy +// synthesizes — the dialect the target clients already parse. +type kubeStatus struct { + Kind string `json:"kind"` + APIVersion string `json:"apiVersion"` + Status string `json:"status"` + Code int `json:"code"` + Reason string `json:"reason"` + Message string `json:"message"` +} + +// writeStatus writes a proxy-synthesized error response, carrying the marker +// header that distinguishes it from anything the upstream returned. +func writeStatus(w http.ResponseWriter, code int, reason, message string) { + body, _ := json.Marshal(kubeStatus{ + Kind: "Status", + APIVersion: "v1", + Status: "Failure", + Code: code, + Reason: reason, + Message: message, + }) + h := w.Header() + h.Set("Content-Type", "application/json") + h.Set("Content-Length", strconv.Itoa(len(body))) + h.Set(proxyErrorHeader, "true") + w.WriteHeader(code) + _, _ = w.Write(body) +} diff --git a/internal/apiproxy/logging.go b/internal/apiproxy/logging.go new file mode 100644 index 0000000..eeb5cf5 --- /dev/null +++ b/internal/apiproxy/logging.go @@ -0,0 +1,160 @@ +package apiproxy + +import ( + "fmt" + "io" + "net/http" + "strings" + "time" +) + +const timeFormat = "15:04:05" + +// redactedQueryKeys are query parameter names whose values are redacted in +// log lines. The platform API never puts credentials in query strings; this +// is a defensive pass. +var redactedQueryKeys = map[string]bool{ + "access_token": true, + "token": true, + "authorization": true, +} + +// requestLogger writes one line per request — and for streaming responses, +// one when headers arrive and one on stream end — recording method, path, +// status, duration, and bytes. Headers and token values are never logged. +type requestLogger struct { + next http.Handler + out io.Writer + quiet bool +} + +func (l *requestLogger) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if l.quiet { + l.next.ServeHTTP(w, r) + return + } + logged := &loggedResponse{ + ResponseWriter: w, + out: l.out, + method: r.Method, + path: redactedRequestPath(r), + isHead: r.Method == http.MethodHead, + start: time.Now(), + } + // Deferred so an aborted stream (client gone, upstream cut) still gets + // its completion line. + defer logged.logCompletion() + l.next.ServeHTTP(logged, r) +} + +// redactedRequestPath returns the request path plus query with token-shaped +// query values redacted, preserving parameter order. +func redactedRequestPath(r *http.Request) string { + rawQuery := r.URL.RawQuery + if rawQuery == "" { + return r.URL.Path + } + pairs := strings.Split(rawQuery, "&") + for i, pair := range pairs { + key, _, found := strings.Cut(pair, "=") + if found && redactedQueryKeys[strings.ToLower(key)] { + pairs[i] = key + "=REDACTED" + } + } + return r.URL.Path + "?" + strings.Join(pairs, "&") +} + +// loggedResponse observes the response as it is written. A response with no +// declared Content-Length is treated as streaming: it logs a line the moment +// headers arrive and another with totals on completion. +type loggedResponse struct { + http.ResponseWriter + out io.Writer + method string + path string + isHead bool + start time.Time + + wroteHeader bool + streaming bool + status int + bytes int64 +} + +func (l *loggedResponse) WriteHeader(code int) { + if code >= 100 && code < 200 { + // Informational responses don't settle the final status. + l.ResponseWriter.WriteHeader(code) + return + } + if !l.wroteHeader { + l.wroteHeader = true + l.status = code + l.streaming = l.detectStreaming(code) + if l.streaming { + fmt.Fprintf(l.out, "%s %-4s %s %d …streaming\n", + time.Now().Format(timeFormat), l.method, l.path, code) + } + } + l.ResponseWriter.WriteHeader(code) +} + +func (l *loggedResponse) Write(b []byte) (int, error) { + if !l.wroteHeader { + l.WriteHeader(http.StatusOK) + } + n, err := l.ResponseWriter.Write(b) + l.bytes += int64(n) + return n, err +} + +// Flush and Unwrap keep the wrapped writer streamable: the reverse proxy +// reaches the underlying Flusher (and Hijacker, for upgrades) through them. +func (l *loggedResponse) Flush() { + if flusher, ok := l.ResponseWriter.(http.Flusher); ok { + flusher.Flush() + } +} + +func (l *loggedResponse) Unwrap() http.ResponseWriter { return l.ResponseWriter } + +// detectStreaming reports whether the response is stream-shaped: no declared +// length on a status/method that can carry a body. +func (l *loggedResponse) detectStreaming(code int) bool { + if l.isHead || code == http.StatusNoContent || code == http.StatusNotModified { + return false + } + return l.Header().Get("Content-Length") == "" +} + +func (l *loggedResponse) logCompletion() { + status := l.status + if !l.wroteHeader { + // The handler wrote nothing; net/http sends an implicit 200. + status = http.StatusOK + } + fmt.Fprintf(l.out, "%s %-4s %s %d %s %s\n", + time.Now().Format(timeFormat), l.method, l.path, status, + formatDuration(time.Since(l.start)), formatBytes(l.bytes)) +} + +func formatDuration(d time.Duration) string { + if d < time.Second { + return fmt.Sprintf("%dms", d.Milliseconds()) + } + return fmt.Sprintf("%.1fs", d.Seconds()) +} + +func formatBytes(n int64) string { + if n < 1000 { + return fmt.Sprintf("%dB", n) + } + value := float64(n) + for _, unit := range []string{"kB", "MB", "GB", "TB"} { + value /= 1000 + if value < 1000 { + return fmt.Sprintf("%.1f%s", value, unit) + } + } + return fmt.Sprintf("%.1fPB", value/1000) +} diff --git a/internal/apiproxy/logging_test.go b/internal/apiproxy/logging_test.go new file mode 100644 index 0000000..37cfd9d --- /dev/null +++ b/internal/apiproxy/logging_test.go @@ -0,0 +1,219 @@ +package apiproxy + +import ( + "bytes" + "errors" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +// doRecorded drives the handler synchronously with a recorder, so by the +// time it returns every log line for the request has been written. +func doRecorded(t *testing.T, h http.Handler, target string) *httptest.ResponseRecorder { + t.Helper() + req := httptest.NewRequest(http.MethodGet, target, nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + return rec +} + +func logLines(buf *bytes.Buffer) []string { + var lines []string + for _, line := range strings.Split(buf.String(), "\n") { + if line != "" { + lines = append(lines, line) + } + } + return lines +} + +func newLoggingProxy(t *testing.T, upstream string, buf *bytes.Buffer, quiet bool, source ...func(*Config)) http.Handler { + t.Helper() + cfg := Config{ + Upstream: parseURL(t, upstream), + TokenSource: staticToken("tok"), + LogWriter: buf, + Quiet: quiet, + } + for _, m := range source { + m(&cfg) + } + server, err := New(cfg) + if err != nil { + t.Fatalf("New: %v", err) + } + return server.Handler() +} + +func TestRequestLogLineRedactsTokenQueryValues(t *testing.T) { + upstream := newRecordingUpstream(t, nil) // responds 200 "ok" with Content-Length + var buf bytes.Buffer + handler := newLoggingProxy(t, upstream.server.URL, &buf, false) + + rec := doRecorded(t, handler, + "http://127.0.0.1:8001/apis/foo?watch=true&access_token=supersecret&TOKEN=alsosecret&authorization=hush&limit=5") + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200", rec.Code) + } + + lines := logLines(&buf) + if len(lines) != 1 { + t.Fatalf("log lines = %q, want exactly one for a non-streaming request", lines) + } + line := lines[0] + for _, want := range []string{ + "GET", "/apis/foo?", "watch=true", "limit=5", " 200 ", + "access_token=REDACTED", "TOKEN=REDACTED", "authorization=REDACTED", + " 2B", // "ok" + } { + if !strings.Contains(line, want) { + t.Errorf("log line %q missing %q", line, want) + } + } + for _, banned := range []string{"supersecret", "alsosecret", "hush", "Bearer", "…streaming"} { + if strings.Contains(line, banned) { + t.Errorf("log line %q leaks %q", line, banned) + } + } +} + +func TestStreamingLogsHeaderArrivalAndCompletion(t *testing.T) { + events := []string{`{"type":"ADDED"}`, `{"type":"DELETED"}`} + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + flusher := w.(http.Flusher) + for _, event := range events { + io.WriteString(w, event+"\n") + flusher.Flush() + } + })) + defer upstream.Close() + var buf bytes.Buffer + handler := newLoggingProxy(t, upstream.URL, &buf, false) + + rec := doRecorded(t, handler, "http://127.0.0.1:8001/apis/foo?watch=true") + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200", rec.Code) + } + + lines := logLines(&buf) + if len(lines) != 2 { + t.Fatalf("log lines = %q, want a streaming-start line and a completion line", lines) + } + if !strings.Contains(lines[0], "…streaming") || !strings.Contains(lines[0], " 200") { + t.Errorf("start line %q must mark the stream and its status", lines[0]) + } + totalBytes := 0 + for _, event := range events { + totalBytes += len(event) + 1 + } + if want := formatBytes(int64(totalBytes)); !strings.Contains(lines[1], " "+want) { + t.Errorf("completion line %q missing total bytes %q", lines[1], want) + } + if strings.Contains(lines[1], "…streaming") { + t.Errorf("completion line %q must carry totals, not the streaming marker", lines[1]) + } +} + +func TestQuietSuppressesRequestLines(t *testing.T) { + upstream := newRecordingUpstream(t, nil) + var buf bytes.Buffer + handler := newLoggingProxy(t, upstream.server.URL, &buf, true) + + rec := doRecorded(t, handler, "http://127.0.0.1:8001/apis/foo?watch=true") + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200", rec.Code) + } + if buf.Len() != 0 { + t.Fatalf("quiet mode logged %q, want nothing", buf.String()) + } +} + +func TestRefreshFailureLoggedOncePerWindowEvenWhenQuiet(t *testing.T) { + source := &fakeTokenSource{token: "stale", expired: true, refreshErr: errTokenDead} + upstream := newRecordingUpstream(t, nil) + var buf bytes.Buffer + handler := newLoggingProxy(t, upstream.server.URL, &buf, true, + func(c *Config) { c.TokenSource = source }) + + for range 3 { + rec := doRecorded(t, handler, "http://127.0.0.1:8001/apis/foo") + if rec.Code != http.StatusBadGateway { + t.Fatalf("status = %d, want 502", rec.Code) + } + } + + lines := logLines(&buf) + if len(lines) != 1 || !strings.Contains(lines[0], "token refresh failed") { + t.Fatalf("log lines = %q, want exactly one refresh-failure line for the whole cooldown window", lines) + } + if strings.Contains(buf.String(), "/apis/foo") { + t.Errorf("quiet mode must not log request lines, got %q", buf.String()) + } +} + +func TestFormatDuration(t *testing.T) { + cases := []struct { + d time.Duration + want string + }{ + {143 * time.Millisecond, "143ms"}, + {0, "0ms"}, + {3200 * time.Millisecond, "3.2s"}, + {time.Minute, "60.0s"}, + } + for _, c := range cases { + if got := formatDuration(c.d); got != c.want { + t.Errorf("formatDuration(%v) = %q, want %q", c.d, got, c.want) + } + } +} + +func TestFormatBytes(t *testing.T) { + cases := []struct { + n int64 + want string + }{ + {0, "0B"}, + {999, "999B"}, + {8100, "8.1kB"}, + {1500000, "1.5MB"}, + {2000000000, "2.0GB"}, + } + for _, c := range cases { + if got := formatBytes(c.n); got != c.want { + t.Errorf("formatBytes(%d) = %q, want %q", c.n, got, c.want) + } + } +} + +func TestRedactedRequestPath(t *testing.T) { + cases := []struct { + target string + want string + }{ + {"http://localhost/apis/foo", "/apis/foo"}, + {"http://localhost/apis/foo?watch=true", "/apis/foo?watch=true"}, + { + "http://localhost/apis/foo?a=1&access_token=x&b=2", + "/apis/foo?a=1&access_token=REDACTED&b=2", + }, + { + // Order preserved, case-insensitive keys, valueless keys untouched. + "http://localhost/apis/foo?Token=x&watch&AUTHORIZATION=y", + "/apis/foo?Token=REDACTED&watch&AUTHORIZATION=REDACTED", + }, + } + for _, c := range cases { + req := httptest.NewRequest(http.MethodGet, c.target, nil) + if got := redactedRequestPath(req); got != c.want { + t.Errorf("redactedRequestPath(%q) = %q, want %q", c.target, got, c.want) + } + } +} + +var errTokenDead = errors.New("refresh token is dead") diff --git a/internal/apiproxy/proxy.go b/internal/apiproxy/proxy.go new file mode 100644 index 0000000..836ff58 --- /dev/null +++ b/internal/apiproxy/proxy.go @@ -0,0 +1,199 @@ +// Package apiproxy implements the engine behind `datumctl api proxy`: a +// local reverse proxy that forwards every request to the configured Datum +// Cloud upstream, injecting a bearer token from the session's token source +// and streaming responses through unbuffered. +// +// The engine is deliberately self-contained: the upstream URL, token source, +// TLS settings, and log destination are all injected, so it can be tested — +// and reasoned about — without Cobra, the keyring, or a real network. +package apiproxy + +import ( + "context" + "crypto/tls" + "fmt" + "io" + "log" + "net" + "net/http" + "net/http/httputil" + "net/url" + "strings" + "sync" + "time" + + "golang.org/x/oauth2" +) + +// Timeouts are asymmetric by design: connection setup is bounded, response +// duration is not — watches are infinite on purpose. Nothing in this package +// may introduce an overall request or response-duration timeout. +const ( + dialTimeout = 10 * time.Second + tlsHandshakeTimeout = 10 * time.Second + responseHeaderTimeout = 30 * time.Second + + // Local listener hygiene. There is deliberately no ReadTimeout or + // WriteTimeout on the local server: either one would kill long-lived + // streams. + readHeaderTimeout = 10 * time.Second + idleTimeout = 2 * time.Minute + + // After a token refresh failure, requests fail fast with the cached + // error for this long instead of re-attempting a refresh per request. + refreshCooldown = 5 * time.Second +) + +// Config carries everything the proxy engine needs. All dependencies are +// injected; the engine never reads datumctl config or the keyring itself. +type Config struct { + // Upstream is the proxy root every request is forwarded under: the + // endpoint root by default, or a control-plane URL whose path prefix + // local request paths are joined onto. + Upstream *url.URL + + // TokenSource supplies the bearer token injected into each outbound + // request (typically authutil.GetTokenSourceForUser). + TokenSource oauth2.TokenSource + + // TLSClientConfig configures TLS to the upstream. Nil means defaults. + TLSClientConfig *tls.Config + + // LogWriter is the request log destination (stderr in production). + // Nil discards log output. + LogWriter io.Writer + + // Quiet suppresses per-request log lines. Token refresh failures are + // still logged, once per refresh attempt. + Quiet bool +} + +// Server is a configured proxy engine. Callers either mount Handler on a +// server of their own or hand a listener to Serve. +type Server struct { + handler http.Handler + httpServer *http.Server +} + +// New builds a proxy engine from cfg. +func New(cfg Config) (*Server, error) { + if cfg.Upstream == nil { + return nil, fmt.Errorf("apiproxy: Upstream is required") + } + if cfg.Upstream.Scheme == "" || cfg.Upstream.Host == "" { + return nil, fmt.Errorf("apiproxy: Upstream must be an absolute URL, got %q", cfg.Upstream) + } + if cfg.TokenSource == nil { + return nil, fmt.Errorf("apiproxy: TokenSource is required") + } + logWriter := cfg.LogWriter + if logWriter == nil { + logWriter = io.Discard + } + + upstreamTransport := &http.Transport{ + Proxy: http.ProxyFromEnvironment, + DialContext: (&net.Dialer{ + Timeout: dialTimeout, + KeepAlive: 30 * time.Second, + }).DialContext, + ForceAttemptHTTP2: true, + TLSClientConfig: cfg.TLSClientConfig, + TLSHandshakeTimeout: tlsHandshakeTimeout, + ResponseHeaderTimeout: responseHeaderTimeout, + IdleConnTimeout: 90 * time.Second, + } + + proxy := &httputil.ReverseProxy{ + Rewrite: rewriteTo(cfg.Upstream), + Transport: &oauth2.Transport{ + Source: newCooldownTokenSource(cfg.TokenSource, refreshCooldown, logWriter), + Base: upstreamTransport, + }, + // Flush every upstream write to the client immediately: unbuffered + // streaming (watch, SSE, chunked transfer) is the point of the proxy. + FlushInterval: -1, + ErrorHandler: handleProxyError, + ErrorLog: log.New(logWriter, "", 0), + } + + var handler http.Handler = &hostValidator{next: proxy} + handler = &requestLogger{next: handler, out: logWriter, quiet: cfg.Quiet} + + return &Server{ + handler: handler, + httpServer: &http.Server{ + Handler: handler, + ReadHeaderTimeout: readHeaderTimeout, + IdleTimeout: idleTimeout, + // No ReadTimeout/WriteTimeout: they would cut long-lived streams. + }, + }, nil +} + +// Handler returns the proxy handler: host validation, request logging, and +// the reverse proxy itself. +func (s *Server) Handler() http.Handler { return s.handler } + +// Serve accepts connections on l until Shutdown is called, applying +// stream-safe local server settings: ReadHeaderTimeout and a keep-alive +// IdleTimeout only, never a ReadTimeout or WriteTimeout. It returns +// http.ErrServerClosed after Shutdown. +func (s *Server) Serve(l net.Listener) error { return s.httpServer.Serve(l) } + +// Shutdown gracefully shuts the proxy down: the listener closes and +// in-flight requests get until ctx expires before their connections are cut. +func (s *Server) Shutdown(ctx context.Context) error { return s.httpServer.Shutdown(ctx) } + +// tokenRefreshError marks a failure that came from the token source rather +// than the upstream connection, so the error handler can report +// ProxyAuthenticationFailed instead of a generic upstream error. +type tokenRefreshError struct{ err error } + +func (e *tokenRefreshError) Error() string { return e.err.Error() } +func (e *tokenRefreshError) Unwrap() error { return e.err } + +// cooldownTokenSource serializes token retrieval — at most one refresh in +// flight — and, after a refresh failure, fails fast with the cached error +// for a cooldown window instead of re-attempting a refresh per proxied +// request. Combined, the auth server sees at most one refresh attempt per +// window no matter how hot the local client polls. +type cooldownTokenSource struct { + source oauth2.TokenSource + cooldown time.Duration + out io.Writer + now func() time.Time + + mu sync.Mutex + lastErr *tokenRefreshError + failedAt time.Time +} + +func newCooldownTokenSource(source oauth2.TokenSource, cooldown time.Duration, out io.Writer) *cooldownTokenSource { + return &cooldownTokenSource{source: source, cooldown: cooldown, out: out, now: time.Now} +} + +// Token implements oauth2.TokenSource. +func (c *cooldownTokenSource) Token() (*oauth2.Token, error) { + c.mu.Lock() + defer c.mu.Unlock() + + if c.lastErr != nil { + if c.now().Sub(c.failedAt) < c.cooldown { + return nil, c.lastErr + } + c.lastErr = nil + } + + token, err := c.source.Token() + if err != nil { + c.lastErr = &tokenRefreshError{err: err} + c.failedAt = c.now() + // Logged here, once per refresh attempt, rather than once per + // request in the error handler. + fmt.Fprintf(c.out, "%s token refresh failed: %s\n", + c.now().Format(timeFormat), strings.ReplaceAll(err.Error(), "\n", " — ")) + return nil, c.lastErr + } + return token, nil +} diff --git a/internal/apiproxy/proxy_test.go b/internal/apiproxy/proxy_test.go new file mode 100644 index 0000000..b58dfc3 --- /dev/null +++ b/internal/apiproxy/proxy_test.go @@ -0,0 +1,476 @@ +package apiproxy + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "sync" + "testing" + + "golang.org/x/oauth2" +) + +// corsHeaders must never appear on any proxy response, synthesized or +// passed through. +var corsHeaders = []string{ + "Access-Control-Allow-Origin", + "Access-Control-Allow-Methods", + "Access-Control-Allow-Headers", + "Access-Control-Allow-Credentials", +} + +func assertNoCORS(t *testing.T, h http.Header) { + t.Helper() + for _, name := range corsHeaders { + if v := h.Get(name); v != "" { + t.Errorf("response carries CORS header %s: %q; the proxy must never emit CORS headers", name, v) + } + } +} + +// upstreamRequest is one request as observed by the fake upstream. +type upstreamRequest struct { + method string + path string + rawQuery string + host string + header http.Header + body []byte +} + +// recordingUpstream is an httptest upstream that records every request it +// receives before delegating to respond (200 "ok" if nil). +type recordingUpstream struct { + server *httptest.Server + + mu sync.Mutex + requests []upstreamRequest +} + +func newRecordingUpstream(t *testing.T, respond http.HandlerFunc) *recordingUpstream { + t.Helper() + u := &recordingUpstream{} + u.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + u.mu.Lock() + u.requests = append(u.requests, upstreamRequest{ + method: r.Method, + path: r.URL.Path, + rawQuery: r.URL.RawQuery, + host: r.Host, + header: r.Header.Clone(), + body: body, + }) + u.mu.Unlock() + if respond != nil { + respond(w, r) + return + } + io.WriteString(w, "ok") + })) + t.Cleanup(u.server.Close) + return u +} + +func (u *recordingUpstream) count() int { + u.mu.Lock() + defer u.mu.Unlock() + return len(u.requests) +} + +func (u *recordingUpstream) last(t *testing.T) upstreamRequest { + t.Helper() + u.mu.Lock() + defer u.mu.Unlock() + if len(u.requests) == 0 { + t.Fatal("upstream received no requests") + } + return u.requests[len(u.requests)-1] +} + +func (u *recordingUpstream) all() []upstreamRequest { + u.mu.Lock() + defer u.mu.Unlock() + return append([]upstreamRequest(nil), u.requests...) +} + +func parseURL(t *testing.T, raw string) *url.URL { + t.Helper() + u, err := url.Parse(raw) + if err != nil { + t.Fatalf("parse %q: %v", raw, err) + } + return u +} + +func staticToken(token string) oauth2.TokenSource { + return oauth2.StaticTokenSource(&oauth2.Token{AccessToken: token}) +} + +// newTestProxy builds a proxy engine against upstream and serves its handler +// on an httptest server (which binds 127.0.0.1, so Host validation passes). +func newTestProxy(t *testing.T, upstream string, mutate ...func(*Config)) *httptest.Server { + t.Helper() + cfg := Config{ + Upstream: parseURL(t, upstream), + TokenSource: staticToken("good-token"), + } + for _, m := range mutate { + m(&cfg) + } + server, err := New(cfg) + if err != nil { + t.Fatalf("New: %v", err) + } + proxy := httptest.NewServer(server.Handler()) + t.Cleanup(proxy.Close) + return proxy +} + +func decodeStatus(t *testing.T, body io.Reader) kubeStatus { + t.Helper() + var status kubeStatus + if err := json.NewDecoder(body).Decode(&status); err != nil { + t.Fatalf("decode Status body: %v", err) + } + return status +} + +func TestPassthrough(t *testing.T) { + upstream := newRecordingUpstream(t, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("X-Upstream", "yes") + w.WriteHeader(http.StatusCreated) + io.WriteString(w, "created") + }) + proxy := newTestProxy(t, upstream.server.URL) + + req, err := http.NewRequest(http.MethodPost, + proxy.URL+"/apis/example.com/v1/things?labelSelector=a%3Db&limit=2", + strings.NewReader("hello body")) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Authorization", "Bearer locally-smuggled") + req.Header.Set("X-Request-ID", "rid-123") + req.Header.Set("Proxy-Connection", "keep-alive") + req.Header.Set("Keep-Alive", "timeout=5") + + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + + if resp.StatusCode != http.StatusCreated { + t.Fatalf("status = %d, want 201", resp.StatusCode) + } + if string(body) != "created" { + t.Errorf("body = %q, want %q", body, "created") + } + if resp.Header.Get("X-Upstream") != "yes" { + t.Error("upstream response header did not pass through") + } + if resp.Header.Get(proxyErrorHeader) != "" { + t.Error("passthrough response must not carry the proxy error marker") + } + assertNoCORS(t, resp.Header) + + got := upstream.last(t) + if got.method != http.MethodPost { + t.Errorf("upstream method = %q, want POST", got.method) + } + if got.path != "/apis/example.com/v1/things" { + t.Errorf("upstream path = %q", got.path) + } + if got.rawQuery != "labelSelector=a%3Db&limit=2" { + t.Errorf("upstream query = %q, want it forwarded verbatim", got.rawQuery) + } + if string(got.body) != "hello body" { + t.Errorf("upstream body = %q", got.body) + } + if auth := got.header.Get("Authorization"); auth != "Bearer good-token" { + t.Errorf("upstream Authorization = %q, want the injected token, never the inbound one", auth) + } + if rid := got.header.Get("X-Request-ID"); rid != "rid-123" { + t.Errorf("upstream X-Request-ID = %q, want passthrough", rid) + } + for _, hop := range []string{"Proxy-Connection", "Keep-Alive"} { + if v := got.header.Get(hop); v != "" { + t.Errorf("hop-by-hop header %s reached upstream: %q", hop, v) + } + } + wantHost := parseURL(t, upstream.server.URL).Host + if got.host != wantHost { + t.Errorf("upstream Host = %q, want %q", got.host, wantHost) + } +} + +func TestHostValidationRejectsNonLocal(t *testing.T) { + upstream := newRecordingUpstream(t, nil) + proxy := newTestProxy(t, upstream.server.URL) + + req, err := http.NewRequest(http.MethodGet, proxy.URL+"/apis/foo", nil) + if err != nil { + t.Fatal(err) + } + req.Host = "evil.example" + + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusForbidden { + t.Fatalf("status = %d, want 403", resp.StatusCode) + } + if resp.Header.Get(proxyErrorHeader) != "true" { + t.Error("proxy-synthesized 403 must carry the proxy error marker") + } + assertNoCORS(t, resp.Header) + if status := decodeStatus(t, resp.Body); status.Kind != "Status" || status.Code != http.StatusForbidden { + t.Errorf("403 body = %+v, want a Status object with code 403", status) + } + if n := upstream.count(); n != 0 { + t.Fatalf("upstream saw %d request(s); a rejected Host must never reach it", n) + } +} + +func TestHostValidationAllowsLocalAliases(t *testing.T) { + upstream := newRecordingUpstream(t, nil) + proxy := newTestProxy(t, upstream.server.URL) + + for _, host := range []string{ + "localhost", "localhost:9999", "LOCALHOST", + "127.0.0.1", "127.0.0.1:8001", + "[::1]", "[::1]:8001", + } { + req, err := http.NewRequest(http.MethodGet, proxy.URL+"/apis/foo", nil) + if err != nil { + t.Fatal(err) + } + req.Host = host + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("Host %q: %v", host, err) + } + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Errorf("Host %q: status = %d, want 200", host, resp.StatusCode) + } + } +} + +func TestAllowedHost(t *testing.T) { + cases := []struct { + host string + want bool + }{ + {"localhost", true}, + {"localhost:8001", true}, + {"LocalHost:8001", true}, + {"127.0.0.1", true}, + {"127.0.0.1:52347", true}, + {"[::1]", true}, + {"[::1]:8001", true}, + {"::1", true}, + {"evil.example", false}, + {"evil.example:80", false}, + {"127.0.0.2", false}, + {"127.0.0.1.evil.example", false}, + {"localhost.evil.example", false}, + {"[2001:db8::1]:8001", false}, + {"", false}, + } + for _, c := range cases { + if got := allowedHost(c.host); got != c.want { + t.Errorf("allowedHost(%q) = %v, want %v", c.host, got, c.want) + } + } +} + +func TestUpstreamAuthErrorsPassThroughUnmarked(t *testing.T) { + for _, code := range []int{http.StatusUnauthorized, http.StatusForbidden} { + t.Run(fmt.Sprint(code), func(t *testing.T) { + upstreamBody := fmt.Sprintf(`{"kind":"Status","apiVersion":"v1","status":"Failure","code":%d,"reason":"upstream-said-no"}`, code) + upstream := newRecordingUpstream(t, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(code) + io.WriteString(w, upstreamBody) + }) + proxy := newTestProxy(t, upstream.server.URL) + + resp, err := http.Get(proxy.URL + "/apis/foo") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + + if resp.StatusCode != code { + t.Fatalf("status = %d, want %d", resp.StatusCode, code) + } + if string(body) != upstreamBody { + t.Errorf("body = %q, want the upstream body byte-for-byte", body) + } + if resp.Header.Get(proxyErrorHeader) != "" { + t.Errorf("upstream %d must pass through without the proxy error marker", code) + } + assertNoCORS(t, resp.Header) + }) + } +} + +func TestUpstreamUnreachableSynthesizes502(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + deadAddr := listener.Addr().String() + listener.Close() + + proxy := newTestProxy(t, "http://"+deadAddr) + resp, err := http.Get(proxy.URL + "/apis/foo") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusBadGateway { + t.Fatalf("status = %d, want 502", resp.StatusCode) + } + if resp.Header.Get(proxyErrorHeader) != "true" { + t.Error("synthesized 502 must carry the proxy error marker") + } + assertNoCORS(t, resp.Header) + status := decodeStatus(t, resp.Body) + if status.Kind != "Status" || status.APIVersion != "v1" || status.Status != "Failure" { + t.Errorf("502 body = %+v, want kind Status / apiVersion v1 / status Failure", status) + } + if status.Code != http.StatusBadGateway { + t.Errorf("Status.code = %d, want 502", status.Code) + } + if status.Reason != "ProxyUpstreamError" { + t.Errorf("Status.reason = %q, want ProxyUpstreamError for a non-token failure", status.Reason) + } +} + +func TestScopedUpstreamJoinsPathPrefix(t *testing.T) { + upstream := newRecordingUpstream(t, nil) + prefix := "/apis/resourcemanager.miloapis.com/v1alpha1/projects/x/control-plane" + proxy := newTestProxy(t, upstream.server.URL+prefix) + + resp, err := http.Get(proxy.URL + "/apis/foo?watch=true") + if err != nil { + t.Fatal(err) + } + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200", resp.StatusCode) + } + + got := upstream.last(t) + if want := prefix + "/apis/foo"; got.path != want { + t.Errorf("upstream path = %q, want %q (local path joined under the scoped prefix)", got.path, want) + } + if got.rawQuery != "watch=true" { + t.Errorf("upstream query = %q, want %q", got.rawQuery, "watch=true") + } +} + +func TestServerSettingsAreStreamSafe(t *testing.T) { + server, err := New(Config{ + Upstream: parseURL(t, "https://api.example.test"), + TokenSource: staticToken("tok"), + }) + if err != nil { + t.Fatal(err) + } + hs := server.httpServer + if hs.ReadHeaderTimeout != readHeaderTimeout { + t.Errorf("ReadHeaderTimeout = %v, want %v", hs.ReadHeaderTimeout, readHeaderTimeout) + } + if hs.IdleTimeout == 0 { + t.Error("IdleTimeout must be set for keep-alive hygiene") + } + if hs.ReadTimeout != 0 || hs.WriteTimeout != 0 { + t.Errorf("ReadTimeout/WriteTimeout = %v/%v, must stay zero — they would kill streams", + hs.ReadTimeout, hs.WriteTimeout) + } +} + +func TestServeAndShutdown(t *testing.T) { + upstream := newRecordingUpstream(t, nil) + server, err := New(Config{ + Upstream: parseURL(t, upstream.server.URL), + TokenSource: staticToken("tok"), + }) + if err != nil { + t.Fatal(err) + } + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + served := make(chan error, 1) + go func() { served <- server.Serve(listener) }() + + resp, err := http.Get("http://" + listener.Addr().String() + "/apis/foo") + if err != nil { + t.Fatal(err) + } + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200", resp.StatusCode) + } + + if err := server.Shutdown(context.Background()); err != nil { + t.Fatalf("Shutdown: %v", err) + } + if err := <-served; !errors.Is(err, http.ErrServerClosed) { + t.Fatalf("Serve returned %v, want http.ErrServerClosed", err) + } +} + +func TestNewValidatesConfig(t *testing.T) { + valid := func() Config { + return Config{ + Upstream: parseURL(t, "https://api.example.test"), + TokenSource: staticToken("tok"), + } + } + + cfg := valid() + cfg.Upstream = nil + if _, err := New(cfg); err == nil { + t.Error("New must reject a nil Upstream") + } + + cfg = valid() + cfg.Upstream = parseURL(t, "/relative/only") + if _, err := New(cfg); err == nil { + t.Error("New must reject a relative Upstream URL") + } + + cfg = valid() + cfg.TokenSource = nil + if _, err := New(cfg); err == nil { + t.Error("New must reject a nil TokenSource") + } + + if _, err := New(valid()); err != nil { + t.Errorf("New with a valid config: %v", err) + } +} diff --git a/internal/apiproxy/streaming_test.go b/internal/apiproxy/streaming_test.go new file mode 100644 index 0000000..a26f61a --- /dev/null +++ b/internal/apiproxy/streaming_test.go @@ -0,0 +1,304 @@ +package apiproxy + +import ( + "bufio" + "fmt" + "io" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +// lineReader delivers response lines on a channel so tests can assert each +// line the moment it clears the proxy. The timeouts below are hang guards +// that turn a buffering proxy into a loud failure, not timing assertions: +// the pass path never waits on a clock. +type lineReader struct { + lines chan string + errs chan error +} + +func newLineReader(r io.Reader) *lineReader { + lr := &lineReader{lines: make(chan string), errs: make(chan error, 1)} + go func() { + reader := bufio.NewReader(r) + for { + line, err := reader.ReadString('\n') + if line != "" { + lr.lines <- line + } + if err != nil { + lr.errs <- err + return + } + } + }() + return lr +} + +func (lr *lineReader) next(t *testing.T) string { + t.Helper() + select { + case line := <-lr.lines: + return line + case err := <-lr.errs: + t.Fatalf("stream ended before the expected line: %v", err) + case <-time.After(10 * time.Second): + t.Fatal("timed out waiting for a line: the proxy is buffering the stream") + } + return "" +} + +func (lr *lineReader) expectEOF(t *testing.T) { + t.Helper() + select { + case err := <-lr.errs: + if err != io.EOF { + t.Fatalf("stream ended with %v, want io.EOF", err) + } + case line := <-lr.lines: + t.Fatalf("unexpected extra line %q", line) + case <-time.After(10 * time.Second): + t.Fatal("timed out waiting for stream end") + } +} + +// TestWatchStreamingIsUnbuffered is the load-bearing streaming test. The +// upstream writes one watch event, flushes, then blocks until the test has +// read that event through the proxy. Receiving event N while the upstream is +// still blocked before producing event N+1 proves zero proxy-side buffering +// with no timing dependence. It runs twice against the same proxy to cover +// the repeat-watch case. +func TestWatchStreamingIsUnbuffered(t *testing.T) { + events := []string{ + `{"type":"ADDED","object":{"metadata":{"name":"zone-a"}}}`, + `{"type":"MODIFIED","object":{"metadata":{"name":"zone-a"}}}`, + `{"type":"DELETED","object":{"metadata":{"name":"zone-a"}}}`, + } + proceed := make(chan struct{}) + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + flusher := w.(http.Flusher) + for i, event := range events { + io.WriteString(w, event+"\n") + flusher.Flush() + // Block until the test confirms the event arrived through the + // proxy: a buffering proxy can never unblock this. + if i < len(events)-1 { + <-proceed + } + } + })) + defer upstream.Close() + proxy := newTestProxy(t, upstream.URL) + + runWatch := func() { + resp, err := http.Get(proxy.URL + "/apis/networking.datumapis.com/v1alpha/dnszones?watch=true") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200", resp.StatusCode) + } + if cl := resp.Header.Get("Content-Length"); cl != "" { + t.Fatalf("watch response carries Content-Length %q; it must stream", cl) + } + reader := newLineReader(resp.Body) + for i, want := range events { + if got := reader.next(t); got != want+"\n" { + t.Fatalf("event %d = %q, want %q", i, got, want) + } + if i < len(events)-1 { + proceed <- struct{}{} + } + } + reader.expectEOF(t) + } + + runWatch() + // A second watch through the same proxy must stream just as well. + runWatch() +} + +func TestSSEStreamingIsUnbuffered(t *testing.T) { + events := []string{"one", "two", "three"} + proceed := make(chan struct{}) + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + flusher := w.(http.Flusher) + for i, event := range events { + fmt.Fprintf(w, "data: %s\n\n", event) + flusher.Flush() + if i < len(events)-1 { + <-proceed + } + } + })) + defer upstream.Close() + proxy := newTestProxy(t, upstream.URL) + + resp, err := http.Get(proxy.URL + "/events") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if ct := resp.Header.Get("Content-Type"); ct != "text/event-stream" { + t.Fatalf("Content-Type = %q, want text/event-stream passthrough", ct) + } + + reader := newLineReader(resp.Body) + for i, event := range events { + if got, want := reader.next(t), "data: "+event+"\n"; got != want { + t.Fatalf("event %d = %q, want %q", i, got, want) + } + if got := reader.next(t); got != "\n" { + t.Fatalf("event %d separator = %q, want blank line", i, got) + } + if i < len(events)-1 { + proceed <- struct{}{} + } + } + reader.expectEOF(t) +} + +// TestWatchStreamingIsUnbufferedThroughScopedProxy re-runs the channel-gated +// watch streaming test against a scoped proxy: the upstream root carries a +// control-plane path prefix, and each event must still clear the proxy while +// the upstream is blocked before producing the next one. +func TestWatchStreamingIsUnbufferedThroughScopedProxy(t *testing.T) { + const prefix = "/apis/resourcemanager.miloapis.com/v1alpha1/projects/my-project/control-plane" + events := []string{ + `{"type":"ADDED","object":{"metadata":{"name":"zone-a"}}}`, + `{"type":"MODIFIED","object":{"metadata":{"name":"zone-a"}}}`, + } + proceed := make(chan struct{}) + var gotPath string + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotPath = r.URL.Path + w.Header().Set("Content-Type", "application/json") + flusher := w.(http.Flusher) + for i, event := range events { + io.WriteString(w, event+"\n") + flusher.Flush() + if i < len(events)-1 { + <-proceed + } + } + })) + defer upstream.Close() + proxy := newTestProxy(t, upstream.URL+prefix) + + resp, err := http.Get(proxy.URL + "/apis/networking.datumapis.com/v1alpha/dnszones?watch=true") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200", resp.StatusCode) + } + + reader := newLineReader(resp.Body) + for i, want := range events { + if got := reader.next(t); got != want+"\n" { + t.Fatalf("event %d = %q, want %q", i, got, want) + } + if i < len(events)-1 { + proceed <- struct{}{} + } + } + reader.expectEOF(t) + + if want := prefix + "/apis/networking.datumapis.com/v1alpha/dnszones"; gotPath != want { + t.Errorf("upstream path = %q, want %q (watch path joined under the scoped prefix)", gotPath, want) + } +} + +// TestSlowTrickleSmallWritesAreUnbuffered covers the slow-trickle streaming +// variant: the upstream emits tiny sub-line fragments, flushing after each, +// and blocks until the client has observed the fragment through the proxy. A +// proxy that coalesces or buffers small writes can never unblock it. +func TestSlowTrickleSmallWritesAreUnbuffered(t *testing.T) { + fragments := []string{"{", `"type":`, `"ADD`, `ED"`, "}\n"} + proceed := make(chan struct{}) + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + flusher := w.(http.Flusher) + for i, fragment := range fragments { + io.WriteString(w, fragment) + flusher.Flush() + if i < len(fragments)-1 { + <-proceed + } + } + })) + defer upstream.Close() + proxy := newTestProxy(t, upstream.URL) + + resp, err := http.Get(proxy.URL + "/apis/foo?watch=true") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200", resp.StatusCode) + } + + for i, fragment := range fragments { + // The upstream is blocked before producing the next fragment, so at + // most len(fragment) bytes can be in flight; ReadFull returning is + // itself the proof this fragment cleared the proxy unbuffered. + buf := make([]byte, len(fragment)) + readDone := make(chan error, 1) + go func() { + _, err := io.ReadFull(resp.Body, buf) + readDone <- err + }() + select { + case err := <-readDone: + if err != nil { + t.Fatalf("fragment %d: read error: %v", i, err) + } + case <-time.After(10 * time.Second): + t.Fatalf("fragment %d: timed out — the proxy is buffering small writes", i) + } + if got := string(buf); got != fragment { + t.Fatalf("fragment %d = %q, want %q", i, got, fragment) + } + if i < len(fragments)-1 { + proceed <- struct{}{} + } + } + if n, err := resp.Body.Read(make([]byte, 1)); err != io.EOF { + t.Fatalf("after the last fragment: read %d bytes, err %v, want io.EOF", n, err) + } +} + +func TestUpstreamClosesMidStream(t *testing.T) { + event := `{"type":"ADDED","object":{"metadata":{"name":"zone-a"}}}` + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + io.WriteString(w, event+"\n") + w.(http.Flusher).Flush() + // Returning here cuts the stream: watch timeout, token expiry, + // anything. The client must see a normal stream end. + })) + defer upstream.Close() + proxy := newTestProxy(t, upstream.URL) + + resp, err := http.Get(proxy.URL + "/apis/foo?watch=true") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200", resp.StatusCode) + } + + reader := newLineReader(resp.Body) + if got := reader.next(t); got != event+"\n" { + t.Fatalf("event = %q, want %q", got, event) + } + reader.expectEOF(t) +} diff --git a/internal/apiproxy/token_test.go b/internal/apiproxy/token_test.go new file mode 100644 index 0000000..47e4c46 --- /dev/null +++ b/internal/apiproxy/token_test.go @@ -0,0 +1,309 @@ +package apiproxy + +import ( + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + customerrors "go.datum.net/datumctl/internal/errors" + "golang.org/x/oauth2" +) + +// fakeTokenSource simulates authutil's refreshing source: it hands out its +// current token until told it is expired, then "refreshes" by minting the +// configured next token (or failing). The gap between the expired check and +// the refresh is deliberately not atomic, so only the engine's serialization +// makes refresh single-flight — which is exactly what the tests prove. +type fakeTokenSource struct { + mu sync.Mutex + token string + next string + expired bool + refreshErr error + + refreshCount atomic.Int32 + onRefresh func() +} + +func (f *fakeTokenSource) Token() (*oauth2.Token, error) { + f.mu.Lock() + expired := f.expired + current := f.token + refreshErr := f.refreshErr + f.mu.Unlock() + + if !expired { + return &oauth2.Token{AccessToken: current}, nil + } + + f.refreshCount.Add(1) + if f.onRefresh != nil { + f.onRefresh() + } + if refreshErr != nil { + return nil, refreshErr + } + f.mu.Lock() + f.token = f.next + f.expired = false + minted := f.token + f.mu.Unlock() + return &oauth2.Token{AccessToken: minted}, nil +} + +// expire marks the current token stale; the next Token call refreshes to next. +func (f *fakeTokenSource) expire(next string) { + f.mu.Lock() + f.expired = true + f.next = next + f.mu.Unlock() +} + +func TestConcurrentRequestsSingleFlightRefresh(t *testing.T) { + const parallel = 8 + started := make(chan struct{}, parallel) + release := make(chan struct{}) + source := &fakeTokenSource{ + token: "stale", + next: "fresh", + expired: true, + onRefresh: func() { + started <- struct{}{} + <-release + }, + } + upstream := newRecordingUpstream(t, nil) + proxy := newTestProxy(t, upstream.server.URL, func(c *Config) { c.TokenSource = source }) + + var wg sync.WaitGroup + failures := make(chan error, parallel) + for range parallel { + wg.Add(1) + go func() { + defer wg.Done() + resp, err := http.Get(proxy.URL + "/apis/foo") + if err != nil { + failures <- err + return + } + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + if resp.StatusCode != http.StatusOK { + failures <- fmt.Errorf("status = %d, want 200", resp.StatusCode) + } + }() + } + + // One refresh is now in flight and blocked; every other request is + // queued behind it. Releasing it must satisfy them all. + <-started + close(release) + wg.Wait() + close(failures) + for err := range failures { + t.Fatal(err) + } + + if got := source.refreshCount.Load(); got != 1 { + t.Fatalf("refresh attempts = %d, want exactly 1 across %d concurrent requests", got, parallel) + } + for _, req := range upstream.all() { + if auth := req.header.Get("Authorization"); auth != "Bearer fresh" { + t.Errorf("upstream Authorization = %q, want the refreshed token", auth) + } + } +} + +func TestStreamSurvivesTokenExpiryMidStream(t *testing.T) { + source := &fakeTokenSource{token: "token-A"} + + var authMu sync.Mutex + var authHeaders []string + proceed := make(chan struct{}) + events := []string{ + `{"type":"ADDED","object":{"metadata":{"name":"zone-a"}}}`, + `{"type":"MODIFIED","object":{"metadata":{"name":"zone-a"}}}`, + } + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + authMu.Lock() + authHeaders = append(authHeaders, r.Header.Get("Authorization")) + authMu.Unlock() + if r.URL.Query().Get("watch") != "true" { + io.WriteString(w, "ok") + return + } + w.Header().Set("Content-Type", "application/json") + flusher := w.(http.Flusher) + io.WriteString(w, events[0]+"\n") + flusher.Flush() + <-proceed + io.WriteString(w, events[1]+"\n") + flusher.Flush() + })) + defer upstream.Close() + proxy := newTestProxy(t, upstream.URL, func(c *Config) { c.TokenSource = source }) + + resp, err := http.Get(proxy.URL + "/apis/foo?watch=true") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + reader := newLineReader(resp.Body) + if got := reader.next(t); got != events[0]+"\n" { + t.Fatalf("event 0 = %q", got) + } + + // The token expires while the stream is open. The stream must not be + // interrupted: authentication happens at request start only. + source.expire("token-B") + proceed <- struct{}{} + if got := reader.next(t); got != events[1]+"\n" { + t.Fatalf("event 1 after expiry = %q", got) + } + reader.expectEOF(t) + + // The next NEW request refreshes and carries the new token. + resp2, err := http.Get(proxy.URL + "/apis/foo") + if err != nil { + t.Fatal(err) + } + io.Copy(io.Discard, resp2.Body) + resp2.Body.Close() + if resp2.StatusCode != http.StatusOK { + t.Fatalf("post-expiry request status = %d, want 200", resp2.StatusCode) + } + + authMu.Lock() + defer authMu.Unlock() + if want := []string{"Bearer token-A", "Bearer token-B"}; len(authHeaders) != 2 || + authHeaders[0] != want[0] || authHeaders[1] != want[1] { + t.Fatalf("upstream Authorization sequence = %q, want %q", authHeaders, want) + } + if got := source.refreshCount.Load(); got != 1 { + t.Fatalf("refresh attempts = %d, want 1", got) + } +} + +func TestRefreshFailureSynthesizes502(t *testing.T) { + userErr := customerrors.NewUserErrorWithHint( + "Authentication session has expired or refresh token is no longer valid.", + "Please re-authenticate using: `datumctl login`", + ) + source := &fakeTokenSource{token: "stale", expired: true, refreshErr: userErr} + upstream := newRecordingUpstream(t, nil) + proxy := newTestProxy(t, upstream.server.URL, func(c *Config) { c.TokenSource = source }) + + const parallel = 8 + var wg sync.WaitGroup + failures := make(chan error, parallel) + for range parallel { + wg.Add(1) + go func() { + defer wg.Done() + resp, err := http.Get(proxy.URL + "/apis/foo") + if err != nil { + failures <- err + return + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusBadGateway { + failures <- fmt.Errorf("status = %d, want 502", resp.StatusCode) + return + } + if resp.Header.Get(proxyErrorHeader) != "true" { + failures <- errors.New("502 must carry the proxy error marker") + return + } + if ct := resp.Header.Get("Content-Type"); ct != "application/json" { + failures <- fmt.Errorf("Content-Type = %q, want application/json", ct) + return + } + status := kubeStatus{} + if err := jsonDecode(resp.Body, &status); err != nil { + failures <- err + return + } + if status.Kind != "Status" || status.APIVersion != "v1" || status.Status != "Failure" || + status.Code != http.StatusBadGateway { + failures <- fmt.Errorf("body = %+v, want a 502 Failure Status", status) + return + } + if status.Reason != "ProxyAuthenticationFailed" { + failures <- fmt.Errorf("Status.reason = %q, want ProxyAuthenticationFailed", status.Reason) + return + } + if !strings.Contains(status.Message, userErr.Message) || !strings.Contains(status.Message, userErr.Hint) { + failures <- fmt.Errorf("Status.message = %q, want the UserError message and hint", status.Message) + } + }() + } + wg.Wait() + close(failures) + for err := range failures { + t.Fatal(err) + } + + if got := source.refreshCount.Load(); got != 1 { + t.Fatalf("refresh attempts = %d, want 1 per cooldown window under concurrent load", got) + } + if n := upstream.count(); n != 0 { + t.Fatalf("upstream saw %d request(s); token failures must never reach it", n) + } +} + +func TestCooldownWindowGatesRefreshRetries(t *testing.T) { + source := &fakeTokenSource{token: "stale", expired: true, refreshErr: errors.New("boom")} + current := time.Unix(1_000_000, 0) + cooled := newCooldownTokenSource(source, 5*time.Second, io.Discard) + cooled.now = func() time.Time { return current } + + _, err := cooled.Token() + if err == nil { + t.Fatal("want an error from the failing source") + } + var refreshErr *tokenRefreshError + if !errors.As(err, &refreshErr) { + t.Fatalf("error %T does not mark a token refresh failure", err) + } + if got := source.refreshCount.Load(); got != 1 { + t.Fatalf("refresh attempts = %d, want 1", got) + } + + // Inside the window: fail fast with the cached error, no new attempt. + if _, err2 := cooled.Token(); err2 != err { + t.Fatalf("in-window error = %v, want the cached %v", err2, err) + } + current = current.Add(4 * time.Second) + if _, _ = cooled.Token(); source.refreshCount.Load() != 1 { + t.Fatalf("refresh attempts = %d after 4s, want still 1", source.refreshCount.Load()) + } + + // Past the window: one new attempt is allowed — and it can recover. + current = current.Add(2 * time.Second) + source.mu.Lock() + source.refreshErr = nil + source.next = "fresh" + source.mu.Unlock() + token, err := cooled.Token() + if err != nil { + t.Fatalf("post-window Token: %v", err) + } + if token.AccessToken != "fresh" { + t.Fatalf("post-window token = %q, want %q", token.AccessToken, "fresh") + } + if got := source.refreshCount.Load(); got != 2 { + t.Fatalf("refresh attempts = %d, want 2 (one per window)", got) + } +} + +func jsonDecode(r io.Reader, v any) error { + return json.NewDecoder(r).Decode(v) +} diff --git a/internal/authutil/credentials.go b/internal/authutil/credentials.go index 5e2dcd1..ce1f699 100644 --- a/internal/authutil/credentials.go +++ b/internal/authutil/credentials.go @@ -27,7 +27,7 @@ const KnownUsersKey = "known_users" // ErrNoActiveUser indicates that no active user is set in the keyring. var ErrNoActiveUser = customerrors.NewUserErrorWithHint( "No active user found.", - "Please login first using: `datumctl auth login`", + "Please login first using: `datumctl login`", ) // IsNoActiveUser reports whether err wraps ErrNoActiveUser. @@ -117,35 +117,67 @@ func GetStoredCredentials(userKey string) (*StoredCredentials, error) { return &creds, nil } -// persistingTokenSource wraps an oauth2.TokenSource and persists token updates to the keyring. +// persistingTokenSource is an oauth2.TokenSource over the credentials stored +// in the keyring: it hands out the stored token while it is valid, refreshes +// it through the session's OAuth endpoints when it is not, and persists +// refreshed tokens back to the keyring. type persistingTokenSource struct { ctx context.Context - source oauth2.TokenSource userKey string creds *StoredCredentials mu sync.Mutex } // Token implements oauth2.TokenSource. -// It retrieves a token from the underlying source and persists it to the keyring if refreshed. +// +// When a refresh is needed, the keyring is re-read first, so long-running +// consumers (`datumctl api proxy`, MCP) track the stored session across their +// lifetime: a deleted entry (`datumctl logout`) fails with actionable +// guidance instead of silently refreshing — and re-persisting — a session the +// user deliberately ended, and a replaced entry (re-login, or a refresh by +// another datumctl process) is adopted without a restart. func (p *persistingTokenSource) Token() (*oauth2.Token, error) { p.mu.Lock() defer p.mu.Unlock() - currentAccessToken := "" - if p.creds.Token != nil { - currentAccessToken = p.creds.Token.AccessToken + // Fast path: the in-memory token is still fresh. + if p.creds.Token.Valid() { + return p.creds.Token, nil } - // Get token from the underlying source (may trigger refresh) - newToken, err := p.source.Token() + stored, err := GetStoredCredentials(p.userKey) + if err != nil { + return nil, customerrors.WrapUserErrorWithHint( + fmt.Sprintf("The stored credentials for %s are no longer available — the session may have been logged out.", p.userKey), + "Run 'datumctl login' to re-authenticate.", + err, + ) + } + p.creds = stored + if p.creds.Token.Valid() { + return p.creds.Token, nil + } + + // Rebuild the oauth2.Config needed for refreshing from the (possibly + // re-read) credentials. + conf := &oauth2.Config{ + ClientID: p.creds.ClientID, + Scopes: p.creds.Scopes, + Endpoint: oauth2.Endpoint{ + AuthURL: p.creds.EndpointAuthURL, + TokenURL: p.creds.EndpointTokenURL, + }, + // RedirectURL not needed for token refresh + } + + newToken, err := conf.TokenSource(p.ctx, p.creds.Token).Token() if err != nil { var retrieveErr *oauth2.RetrieveError if errors.As(err, &retrieveErr) { if retrieveErr.ErrorCode == "invalid_grant" || retrieveErr.ErrorCode == "invalid_request" { return nil, customerrors.WrapUserErrorWithHint( "Authentication session has expired or refresh token is no longer valid.", - "Please re-authenticate using: `datumctl auth login`", + "Please re-authenticate using: `datumctl login`", err, ) } @@ -154,7 +186,7 @@ func (p *persistingTokenSource) Token() (*oauth2.Token, error) { } // Persist the token if it was refreshed - if newToken.AccessToken != currentAccessToken { + if newToken.AccessToken != p.creds.Token.AccessToken { p.creds.Token = newToken credsJSON, marshalErr := json.Marshal(p.creds) @@ -202,24 +234,10 @@ func tokenSourceFor(ctx context.Context, userKey string, creds *StoredCredential }, nil } - // Rebuild the oauth2.Config needed for refreshing - conf := &oauth2.Config{ - ClientID: creds.ClientID, - Scopes: creds.Scopes, - Endpoint: oauth2.Endpoint{ - AuthURL: creds.EndpointAuthURL, - TokenURL: creds.EndpointTokenURL, - }, - // RedirectURL not needed for token refresh - } - - // Create the base TokenSource with the stored token - baseSource := conf.TokenSource(ctx, creds.Token) - - // Wrap it with our persisting source + // The source re-reads the keyring and rebuilds its refresh config + // whenever the stored token needs refreshing. return &persistingTokenSource{ ctx: ctx, - source: baseSource, userKey: userKey, creds: creds, }, nil diff --git a/internal/authutil/credentials_test.go b/internal/authutil/credentials_test.go new file mode 100644 index 0000000..6df6507 --- /dev/null +++ b/internal/authutil/credentials_test.go @@ -0,0 +1,191 @@ +package authutil + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + customerrors "go.datum.net/datumctl/internal/errors" + "go.datum.net/datumctl/internal/keyring" + "golang.org/x/oauth2" +) + +const testUserKey = "maya@datum.net@auth.datum.net" + +// seedInteractiveCreds stores interactive credentials for testUserKey and +// returns them. expiry in the past forces the next Token() call to refresh. +func seedInteractiveCreds(t *testing.T, accessToken, refreshToken, tokenURL string, expiry time.Time) *StoredCredentials { + t.Helper() + creds := &StoredCredentials{ + Hostname: "auth.datum.net", + APIHostname: "api.datum.net", + ClientID: "client-id", + EndpointTokenURL: tokenURL, + UserEmail: "maya@datum.net", + Token: &oauth2.Token{ + AccessToken: accessToken, + RefreshToken: refreshToken, + Expiry: expiry, + }, + } + blob, err := json.Marshal(creds) + if err != nil { + t.Fatalf("marshal creds: %v", err) + } + if err := keyring.Set(ServiceName, testUserKey, string(blob)); err != nil { + t.Fatalf("seed keyring: %v", err) + } + return creds +} + +func TestPersistingTokenSource_ValidTokenFastPath(t *testing.T) { + mockKeyring(t) + seedInteractiveCreds(t, "fresh-token", "refresh-token", "http://127.0.0.1:1/token", time.Now().Add(time.Hour)) + + source, err := GetTokenSourceForUser(context.Background(), testUserKey) + if err != nil { + t.Fatalf("GetTokenSourceForUser: %v", err) + } + token, err := source.Token() + if err != nil { + t.Fatalf("Token: %v", err) + } + if token.AccessToken != "fresh-token" { + t.Errorf("access token = %q, want the stored token with no refresh", token.AccessToken) + } +} + +// TestPersistingTokenSource_LogoutMidRunFailsWithoutResurrection pins the +// proxy's mid-run logout contract: once the keyring entry is deleted, the +// next refresh fails with a UserError pointing at 'datumctl login', and the +// deleted entry is NOT silently re-created from in-memory state. +func TestPersistingTokenSource_LogoutMidRunFailsWithoutResurrection(t *testing.T) { + mockKeyring(t) + seedInteractiveCreds(t, "old-token", "refresh-token", "http://127.0.0.1:1/token", time.Now().Add(-time.Minute)) + + source, err := GetTokenSourceForUser(context.Background(), testUserKey) + if err != nil { + t.Fatalf("GetTokenSourceForUser: %v", err) + } + + // The user logs out: the keyring entry disappears. + if err := keyring.Delete(ServiceName, testUserKey); err != nil { + t.Fatalf("delete creds: %v", err) + } + + _, err = source.Token() + if err == nil { + t.Fatal("Token after logout: want an error, got none") + } + userErr, ok := customerrors.IsUserError(err) + if !ok { + t.Fatalf("error = %v (%T), want a UserError", err, err) + } + if !strings.Contains(userErr.Hint, "datumctl login") { + t.Errorf("hint = %q, want it to point at 'datumctl login'", userErr.Hint) + } + if _, getErr := keyring.Get(ServiceName, testUserKey); !errors.Is(getErr, keyring.ErrNotFound) { + t.Errorf("keyring entry after failed refresh: err = %v, want ErrNotFound (the logged-out session must not be resurrected)", getErr) + } +} + +// TestPersistingTokenSource_RecoversAfterReLogin pins the recovery half of +// the same contract: when the user logs back in to the same session, an +// already-running token source adopts the new stored credentials on its next +// refresh — no restart required. +func TestPersistingTokenSource_RecoversAfterReLogin(t *testing.T) { + mockKeyring(t) + seedInteractiveCreds(t, "stale-token", "dead-refresh-token", "http://127.0.0.1:1/token", time.Now().Add(-time.Minute)) + + source, err := GetTokenSourceForUser(context.Background(), testUserKey) + if err != nil { + t.Fatalf("GetTokenSourceForUser: %v", err) + } + + // The user re-logs in: the keyring entry is replaced with fresh creds. + seedInteractiveCreds(t, "relogin-token", "new-refresh-token", "http://127.0.0.1:1/token", time.Now().Add(time.Hour)) + + token, err := source.Token() + if err != nil { + t.Fatalf("Token after re-login: %v", err) + } + if token.AccessToken != "relogin-token" { + t.Errorf("access token = %q, want the re-login token adopted from the keyring", token.AccessToken) + } +} + +// TestPersistingTokenSource_RefreshPersistsToKeyring exercises a real refresh +// against a fake token endpoint and asserts the refreshed token is written +// back to the keyring. +func TestPersistingTokenSource_RefreshPersistsToKeyring(t *testing.T) { + mockKeyring(t) + + tokenEndpoint := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{"access_token":"refreshed-token","token_type":"Bearer","refresh_token":"refresh-token-2","expires_in":3600}`) + })) + defer tokenEndpoint.Close() + + seedInteractiveCreds(t, "expired-token", "refresh-token", tokenEndpoint.URL, time.Now().Add(-time.Minute)) + + source, err := GetTokenSourceForUser(context.Background(), testUserKey) + if err != nil { + t.Fatalf("GetTokenSourceForUser: %v", err) + } + token, err := source.Token() + if err != nil { + t.Fatalf("Token: %v", err) + } + if token.AccessToken != "refreshed-token" { + t.Errorf("access token = %q, want %q", token.AccessToken, "refreshed-token") + } + + persisted, err := GetStoredCredentials(testUserKey) + if err != nil { + t.Fatalf("GetStoredCredentials after refresh: %v", err) + } + if persisted.Token.AccessToken != "refreshed-token" { + t.Errorf("persisted access token = %q, want the refreshed token", persisted.Token.AccessToken) + } +} + +// TestPersistingTokenSource_InvalidGrantIsUserError pins the flagship failure +// message: a dead refresh token surfaces a UserError whose hint names the +// real login command ('datumctl login', not the nonexistent 'auth login'). +func TestPersistingTokenSource_InvalidGrantIsUserError(t *testing.T) { + mockKeyring(t) + + tokenEndpoint := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + fmt.Fprint(w, `{"error":"invalid_grant","error_description":"refresh token revoked"}`) + })) + defer tokenEndpoint.Close() + + seedInteractiveCreds(t, "expired-token", "revoked-refresh-token", tokenEndpoint.URL, time.Now().Add(-time.Minute)) + + source, err := GetTokenSourceForUser(context.Background(), testUserKey) + if err != nil { + t.Fatalf("GetTokenSourceForUser: %v", err) + } + _, err = source.Token() + if err == nil { + t.Fatal("Token with a revoked refresh token: want an error") + } + userErr, ok := customerrors.IsUserError(err) + if !ok { + t.Fatalf("error = %v (%T), want a UserError", err, err) + } + if !strings.Contains(userErr.Hint, "`datumctl login`") { + t.Errorf("hint = %q, want it to reference `datumctl login`", userErr.Hint) + } + if strings.Contains(userErr.Hint, "auth login") { + t.Errorf("hint = %q references the nonexistent 'datumctl auth login'", userErr.Hint) + } +} diff --git a/internal/authutil/login.go b/internal/authutil/login.go index d629645..4e5b894 100644 --- a/internal/authutil/login.go +++ b/internal/authutil/login.go @@ -201,7 +201,7 @@ func runPKCEFlow(ctx context.Context, provider *oidc.Provider, clientID string, fmt.Println("Please visit this URL manually to authenticate:") fmt.Printf("\n%s\n\n", authURL) fmt.Println("Tip: in a headless environment (CI, SSH without forwarding, or a container) use") - fmt.Println("'datumctl auth login --no-browser' — it uses a device-code flow that doesn't need a local browser.") + fmt.Println("'datumctl login --no-browser' — it uses a device-code flow that doesn't need a local browser.") } else { fmt.Println("Please complete the authentication in your browser.") } diff --git a/internal/authutil/migrate_test.go b/internal/authutil/migrate_test.go index 210890e..1c56c90 100644 --- a/internal/authutil/migrate_test.go +++ b/internal/authutil/migrate_test.go @@ -25,6 +25,17 @@ func TestUserKeyFor(t *testing.T) { } } +// mockKeyring points HOME at an empty temp dir before installing the mock +// keyring provider. Without the HOME isolation, a real fallback credentials +// file at ~/.datumctl/credentials.json would silently hijack the mock (the +// wrapper prefers an existing on-disk store), making these tests read — and +// write — the developer's real credential store. +func mockKeyring(t *testing.T) { + t.Helper() + t.Setenv("HOME", t.TempDir()) + keyring.MockInit() +} + // storeCreds writes a StoredCredentials blob to the keyring under key. func storeCreds(t *testing.T, key, authHostname string) { t.Helper() @@ -64,7 +75,7 @@ func collidingConfig() *datumconfig.ConfigV1Beta1 { } func TestMigrateUserKeys_CollisionRecoversMatchingEnvironment(t *testing.T) { - keyring.MockInit() + mockKeyring(t) // The single shared blob currently holds the staging token (last written). storeCreds(t, "swells@datum.net", "auth.staging.env.datum.net") @@ -99,7 +110,7 @@ func TestMigrateUserKeys_CollisionRecoversMatchingEnvironment(t *testing.T) { } func TestMigrateUserKeys_Idempotent(t *testing.T) { - keyring.MockInit() + mockKeyring(t) storeCreds(t, "swells@datum.net", "auth.staging.env.datum.net") cfg := collidingConfig() @@ -117,7 +128,7 @@ func TestMigrateUserKeys_Idempotent(t *testing.T) { } func TestMigrateUserKeys_AlreadyQualifiedNoop(t *testing.T) { - keyring.MockInit() + mockKeyring(t) cfg := &datumconfig.ConfigV1Beta1{ Sessions: []datumconfig.Session{{ diff --git a/internal/authutil/serviceaccount.go b/internal/authutil/serviceaccount.go index ae3a6a9..4eee6e2 100644 --- a/internal/authutil/serviceaccount.go +++ b/internal/authutil/serviceaccount.go @@ -26,10 +26,10 @@ import ( // ServiceAccountCredentials is the on-disk JSON format downloaded from the Datum Cloud portal. type ServiceAccountCredentials struct { - Type string `json:"type"` // "datum_service_account" - APIEndpoint string `json:"api_endpoint"` // "https://api.datum.net" - TokenURI string `json:"token_uri"` // "https://auth.datum.net/oauth/v2/token" - Scope string `json:"scope"` // OAuth2 scope string, e.g. "openid profile email urn:zitadel:..." + Type string `json:"type"` // "datum_service_account" + APIEndpoint string `json:"api_endpoint"` // "https://api.datum.net" + TokenURI string `json:"token_uri"` // "https://auth.datum.net/oauth/v2/token" + Scope string `json:"scope"` // OAuth2 scope string, e.g. "openid profile email urn:zitadel:..." ProjectID string `json:"project_id"` ClientEmail string `json:"client_email"` // identity e-mail, used as display name ClientID string `json:"client_id"` // numeric Zitadel user ID (iss / sub) @@ -205,7 +205,7 @@ func (m *serviceAccountTokenSource) Token() (*oauth2.Token, error) { if readErr != nil { return nil, customerrors.WrapUserErrorWithHint( "failed to read service account private key from "+sa.PrivateKeyPath, - "re-run 'datumctl auth login --credentials '; you may need to download a new service account credentials file from the Datum portal if the original is no longer available", + "re-run 'datumctl login --credentials '; you may need to download a new service account credentials file from the Datum portal if the original is no longer available", readErr, ) } @@ -213,7 +213,7 @@ func (m *serviceAccountTokenSource) Token() (*oauth2.Token, error) { if pemKey == "" { return nil, customerrors.WrapUserErrorWithHint( "service account session is missing its private key", - "re-run 'datumctl auth login --credentials '; you may need to download a new service account credentials file from the Datum portal if the original is no longer available", + "re-run 'datumctl login --credentials '; you may need to download a new service account credentials file from the Datum portal if the original is no longer available", nil, ) } @@ -222,7 +222,7 @@ func (m *serviceAccountTokenSource) Token() (*oauth2.Token, error) { if err != nil { return nil, customerrors.WrapUserErrorWithHint( "Failed to mint JWT for service account authentication.", - "Please re-authenticate using: `datumctl auth login --credentials `", + "Please re-authenticate using: `datumctl login --credentials `", err, ) } @@ -231,7 +231,7 @@ func (m *serviceAccountTokenSource) Token() (*oauth2.Token, error) { if err != nil { return nil, customerrors.WrapUserErrorWithHint( "Failed to exchange JWT for access token.", - "Please re-authenticate using: `datumctl auth login --credentials `", + "Please re-authenticate using: `datumctl login --credentials `", err, ) } diff --git a/internal/client/factory.go b/internal/client/factory.go index 9d1f6e5..87efa2e 100644 --- a/internal/client/factory.go +++ b/internal/client/factory.go @@ -2,7 +2,6 @@ package client import ( "context" - "encoding/base64" "fmt" "net/http" "os" @@ -34,8 +33,8 @@ type errorClientConfig struct{ err error } func (e *errorClientConfig) RawConfig() (clientcmdapi.Config, error) { return clientcmdapi.Config{}, e.err } -func (e *errorClientConfig) ClientConfig() (*rest.Config, error) { return nil, e.err } -func (e *errorClientConfig) Namespace() (string, bool, error) { return "default", false, nil } +func (e *errorClientConfig) ClientConfig() (*rest.Config, error) { return nil, e.err } +func (e *errorClientConfig) Namespace() (string, bool, error) { return "default", false, nil } func (e *errorClientConfig) ConfigAccess() clientcmd.ConfigAccess { return nil } type DatumCloudFactory struct { @@ -81,7 +80,7 @@ func (c *CustomConfigFlags) ToRESTConfig() (*rest.Config, error) { config.ServerName = *c.TLSServerName } - userKey, err := c.resolveUserKey(session) + userKey, err := sessionUserKey(session) if err != nil { return nil, err } @@ -128,11 +127,11 @@ func (c *CustomConfigFlags) ToRESTConfig() (*rest.Config, error) { config.Insecure = true } if len(config.CAData) == 0 && ep.CertificateAuthorityData != "" { - decoded, err := base64.StdEncoding.DecodeString(ep.CertificateAuthorityData) + epTLS, err := sessionEndpointTLS(session) if err != nil { - return nil, fmt.Errorf("decode certificate authority data for session %q: %w", session.Name, err) + return nil, err } - config.CAData = decoded + config.CAData = epTLS.CAData } } @@ -254,25 +253,14 @@ func (c *CustomConfigFlags) loadDatumContext() (*datumconfig.DiscoveredContext, return ctxEntry, session, nil } -func (c *CustomConfigFlags) resolveUserKey(session *datumconfig.Session) (string, error) { - if session != nil && session.UserKey != "" { - return session.UserKey, nil - } - return authutil.GetUserKey() -} - +// resolveBaseServer picks the base API server: the --server flag when set, +// else the shared session-endpoint resolution (session endpoint, falling back +// to the keyring API hostname). func (c *CustomConfigFlags) resolveBaseServer(userKey string, session *datumconfig.Session) (string, error) { if c.APIServer != nil && *c.APIServer != "" { return datumconfig.CleanBaseServer(datumconfig.EnsureScheme(*c.APIServer)), nil } - if session != nil && session.Endpoint.Server != "" { - return datumconfig.CleanBaseServer(datumconfig.EnsureScheme(session.Endpoint.Server)), nil - } - apiHostname, err := authutil.GetAPIHostnameForUser(userKey) - if err != nil { - return "", err - } - return datumconfig.CleanBaseServer(datumconfig.EnsureScheme(apiHostname)), nil + return sessionBaseServer(userKey, session) } // ResolvedScope returns the effective project, organization, and platform-wide diff --git a/internal/client/session_endpoint.go b/internal/client/session_endpoint.go new file mode 100644 index 0000000..c900508 --- /dev/null +++ b/internal/client/session_endpoint.go @@ -0,0 +1,119 @@ +package client + +import ( + "encoding/base64" + "fmt" + + "go.datum.net/datumctl/internal/authutil" + "go.datum.net/datumctl/internal/datumconfig" + customerrors "go.datum.net/datumctl/internal/errors" +) + +// EndpointTLS carries the TLS settings a session's endpoint declares: +// the SNI/verification server name, whether verification is skipped, and the +// decoded certificate authority bundle. +type EndpointTLS struct { + ServerName string + InsecureSkipTLSVerify bool + CAData []byte +} + +// SessionEndpoint is the resolved connection identity for one datumctl +// session: the keyring user key that owns its credentials, the base API +// server URL (scheme included, no trailing slash), and the endpoint's TLS +// settings. +type SessionEndpoint struct { + UserKey string + BaseServer string + TLS EndpointTLS +} + +// ResolveSessionEndpoint picks a session from cfg and resolves its endpoint, +// using the same resolution order CustomConfigFlags.ToRESTConfig performs: +// the named session when sessionName is given (error if unknown), else the +// active session, else the keyring active user for pre-session setups; the +// base server from the session endpoint, falling back to the keyring API +// hostname; TLS from the session endpoint. The returned session is nil on +// the keyring-fallback path. +func ResolveSessionEndpoint(cfg *datumconfig.ConfigV1Beta1, sessionName string) (*datumconfig.Session, *SessionEndpoint, error) { + var session *datumconfig.Session + if sessionName != "" { + session = cfg.SessionByName(sessionName) + if session == nil { + return nil, nil, customerrors.NewUserErrorWithHint( + fmt.Sprintf("No session named %q.", sessionName), + "Run 'datumctl auth list' to see the available session names.", + ) + } + } else { + session = cfg.ActiveSessionEntry() + } + + userKey, err := sessionUserKey(session) + if err != nil { + return nil, nil, err + } + baseServer, err := sessionBaseServer(userKey, session) + if err != nil { + return nil, nil, err + } + tlsSettings, err := sessionEndpointTLS(session) + if err != nil { + return nil, nil, err + } + + return session, &SessionEndpoint{ + UserKey: userKey, + BaseServer: baseServer, + TLS: tlsSettings, + }, nil +} + +// sessionUserKey returns the keyring user key for a session, falling back to +// the keyring active user (which bootstraps pre-session setups) when the +// session is nil or carries no key. +func sessionUserKey(session *datumconfig.Session) (string, error) { + if session != nil && session.UserKey != "" { + return session.UserKey, nil + } + return authutil.GetUserKey() +} + +// sessionBaseServer returns the base API server URL for a session, falling +// back to the API hostname recorded in the user's stored credentials. +func sessionBaseServer(userKey string, session *datumconfig.Session) (string, error) { + if session != nil && session.Endpoint.Server != "" { + return datumconfig.CleanBaseServer(datumconfig.EnsureScheme(session.Endpoint.Server)), nil + } + apiHostname, err := authutil.GetAPIHostnameForUser(userKey) + if err != nil { + return "", err + } + return datumconfig.CleanBaseServer(datumconfig.EnsureScheme(apiHostname)), nil +} + +// sessionEndpointTLS extracts the endpoint TLS settings from a session, +// decoding the base64 certificate authority data. A nil session has no TLS +// settings. +func sessionEndpointTLS(session *datumconfig.Session) (EndpointTLS, error) { + if session == nil { + return EndpointTLS{}, nil + } + ep := session.Endpoint + settings := EndpointTLS{ + ServerName: ep.TLSServerName, + InsecureSkipTLSVerify: ep.InsecureSkipTLSVerify, + } + if ep.CertificateAuthorityData != "" { + decoded, err := base64.StdEncoding.DecodeString(ep.CertificateAuthorityData) + if err != nil { + return EndpointTLS{}, customerrors.WrapUserErrorWithHint( + fmt.Sprintf("Could not decode the certificate authority data stored for session %q.", session.Name), + "Run 'datumctl login' again to refresh the session's endpoint settings.", + err, + ) + } + settings.CAData = decoded + } + return settings, nil +} diff --git a/internal/client/session_endpoint_test.go b/internal/client/session_endpoint_test.go new file mode 100644 index 0000000..49b445f --- /dev/null +++ b/internal/client/session_endpoint_test.go @@ -0,0 +1,224 @@ +package client + +import ( + "bytes" + "encoding/base64" + "strings" + "testing" + + "go.datum.net/datumctl/internal/datumconfig" + customerrors "go.datum.net/datumctl/internal/errors" +) + +// The fixtures below always carry a UserKey and an endpoint Server so that +// resolution never falls back to the keyring, keeping these tests hermetic. + +func configWithSessions(active string, sessions ...datumconfig.Session) *datumconfig.ConfigV1Beta1 { + return &datumconfig.ConfigV1Beta1{ + APIVersion: datumconfig.V1Beta1APIVersion, + Kind: datumconfig.DefaultKind, + Sessions: sessions, + ActiveSession: active, + } +} + +// TestResolveSessionEndpoint_BaseServer pins the base-server normalization to +// what ToRESTConfig's resolution applies to session.Endpoint.Server: +// EnsureScheme then CleanBaseServer. +func TestResolveSessionEndpoint_BaseServer(t *testing.T) { + tests := []struct { + name string + server string + want string + }{ + {name: "scheme kept", server: "https://api.datum.net", want: "https://api.datum.net"}, + {name: "scheme added", server: "api.staging.env.datum.net", want: "https://api.staging.env.datum.net"}, + {name: "trailing slash stripped", server: "https://api.datum.net/", want: "https://api.datum.net"}, + {name: "http scheme kept", server: "http://localhost:8080", want: "http://localhost:8080"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + cfg := configWithSessions("s", datumconfig.Session{ + Name: "s", + UserKey: "user@auth.datum.net", + UserEmail: "user@datum.net", + Endpoint: datumconfig.Endpoint{Server: tc.server}, + }) + session, endpoint, err := ResolveSessionEndpoint(cfg, "") + if err != nil { + t.Fatalf("ResolveSessionEndpoint: %v", err) + } + if session == nil || session.Name != "s" { + t.Fatalf("session = %+v, want the active session", session) + } + if endpoint.BaseServer != tc.want { + t.Errorf("BaseServer = %q, want %q", endpoint.BaseServer, tc.want) + } + if endpoint.UserKey != "user@auth.datum.net" { + t.Errorf("UserKey = %q, want the session's user key", endpoint.UserKey) + } + }) + } +} + +// TestResolveSessionEndpoint_TLS pins the endpoint TLS handling to what +// ToRESTConfig applies: TLSServerName and InsecureSkipTLSVerify carried +// as-is, CertificateAuthorityData base64-decoded. +func TestResolveSessionEndpoint_TLS(t *testing.T) { + caPEM := []byte("-----BEGIN CERTIFICATE-----\nMIIB\n-----END CERTIFICATE-----\n") + + tests := []struct { + name string + endpoint datumconfig.Endpoint + want EndpointTLS + wantErr bool + wantErrPart string + }{ + { + name: "no TLS settings", + endpoint: datumconfig.Endpoint{Server: "https://api.datum.net"}, + want: EndpointTLS{}, + }, + { + name: "server name and insecure", + endpoint: datumconfig.Endpoint{ + Server: "https://api.staging.env.datum.net", + TLSServerName: "api.internal", + InsecureSkipTLSVerify: true, + }, + want: EndpointTLS{ServerName: "api.internal", InsecureSkipTLSVerify: true}, + }, + { + name: "certificate authority data decoded", + endpoint: datumconfig.Endpoint{ + Server: "https://api.staging.env.datum.net", + CertificateAuthorityData: base64.StdEncoding.EncodeToString(caPEM), + }, + want: EndpointTLS{CAData: caPEM}, + }, + { + name: "invalid certificate authority data", + endpoint: datumconfig.Endpoint{ + Server: "https://api.staging.env.datum.net", + CertificateAuthorityData: "not!!!base64", + }, + wantErr: true, + wantErrPart: "tls-session", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + cfg := configWithSessions("tls-session", datumconfig.Session{ + Name: "tls-session", + UserKey: "user@auth.datum.net", + UserEmail: "user@datum.net", + Endpoint: tc.endpoint, + }) + _, endpoint, err := ResolveSessionEndpoint(cfg, "") + if tc.wantErr { + if err == nil { + t.Fatal("expected an error") + } + if !strings.Contains(err.Error(), tc.wantErrPart) { + t.Errorf("error = %q, want it to contain %q", err, tc.wantErrPart) + } + return + } + if err != nil { + t.Fatalf("ResolveSessionEndpoint: %v", err) + } + if endpoint.TLS.ServerName != tc.want.ServerName { + t.Errorf("ServerName = %q, want %q", endpoint.TLS.ServerName, tc.want.ServerName) + } + if endpoint.TLS.InsecureSkipTLSVerify != tc.want.InsecureSkipTLSVerify { + t.Errorf("InsecureSkipTLSVerify = %v, want %v", + endpoint.TLS.InsecureSkipTLSVerify, tc.want.InsecureSkipTLSVerify) + } + if !bytes.Equal(endpoint.TLS.CAData, tc.want.CAData) { + t.Errorf("CAData = %q, want %q", endpoint.TLS.CAData, tc.want.CAData) + } + }) + } +} + +func TestResolveSessionEndpoint_NamedSessionWinsOverActive(t *testing.T) { + cfg := configWithSessions("active", + datumconfig.Session{ + Name: "active", + UserKey: "active@auth.datum.net", + UserEmail: "active@datum.net", + Endpoint: datumconfig.Endpoint{Server: "https://api.datum.net"}, + }, + datumconfig.Session{ + Name: "pinned", + UserKey: "pinned@auth.staging.env.datum.net", + UserEmail: "pinned@datum.net", + Endpoint: datumconfig.Endpoint{Server: "https://api.staging.env.datum.net"}, + }, + ) + + session, endpoint, err := ResolveSessionEndpoint(cfg, "pinned") + if err != nil { + t.Fatalf("ResolveSessionEndpoint: %v", err) + } + if session.Name != "pinned" { + t.Errorf("session = %q, want the pinned session", session.Name) + } + if endpoint.BaseServer != "https://api.staging.env.datum.net" { + t.Errorf("BaseServer = %q, want the pinned session's endpoint", endpoint.BaseServer) + } + if endpoint.UserKey != "pinned@auth.staging.env.datum.net" { + t.Errorf("UserKey = %q, want the pinned session's user key", endpoint.UserKey) + } +} + +func TestResolveSessionEndpoint_UnknownNameIsUserError(t *testing.T) { + cfg := configWithSessions("active", datumconfig.Session{ + Name: "active", + UserKey: "active@auth.datum.net", + UserEmail: "active@datum.net", + Endpoint: datumconfig.Endpoint{Server: "https://api.datum.net"}, + }) + + _, _, err := ResolveSessionEndpoint(cfg, "missing") + if err == nil { + t.Fatal("expected an error for an unknown session name") + } + userErr, ok := customerrors.IsUserError(err) + if !ok { + t.Fatalf("error = %v (%T), want a UserError", err, err) + } + if !strings.Contains(userErr.Hint, "datumctl auth list") { + t.Errorf("hint = %q, want it to point at 'datumctl auth list'", userErr.Hint) + } +} + +// TestResolveSessionEndpoint_ActiveSessionFromContext pins ActiveSessionEntry +// semantics: with no explicit active session, the session referenced by the +// current context is used. +func TestResolveSessionEndpoint_ActiveSessionFromContext(t *testing.T) { + cfg := configWithSessions("", datumconfig.Session{ + Name: "ctx-session", + UserKey: "user@auth.datum.net", + UserEmail: "user@datum.net", + Endpoint: datumconfig.Endpoint{Server: "https://api.datum.net"}, + }) + cfg.Contexts = []datumconfig.DiscoveredContext{{ + Name: "org-a", + Session: "ctx-session", + OrganizationID: "org-a", + }} + cfg.CurrentContext = "org-a" + + session, endpoint, err := ResolveSessionEndpoint(cfg, "") + if err != nil { + t.Fatalf("ResolveSessionEndpoint: %v", err) + } + if session == nil || session.Name != "ctx-session" { + t.Fatalf("session = %+v, want the context's session", session) + } + if endpoint.BaseServer != "https://api.datum.net" { + t.Errorf("BaseServer = %q, want %q", endpoint.BaseServer, "https://api.datum.net") + } +} diff --git a/internal/cmd/api/api.go b/internal/cmd/api/api.go new file mode 100644 index 0000000..7a8c153 --- /dev/null +++ b/internal/cmd/api/api.go @@ -0,0 +1,23 @@ +// Package api implements the 'datumctl api' command group: commands for +// working with the Datum Cloud API directly over HTTP. +package api + +import ( + "github.com/spf13/cobra" + "k8s.io/kubectl/pkg/util/templates" + + "go.datum.net/datumctl/internal/client" +) + +// Command returns the parent 'api' command group. +func Command(factory *client.DatumCloudFactory) *cobra.Command { + cmd := &cobra.Command{ + Use: "api", + Short: "Work with the Datum Cloud API directly", + Long: templates.LongDesc(` + Commands for working with the Datum Cloud API directly over HTTP, + such as running a local authenticated proxy for your own tools.`), + } + cmd.AddCommand(proxyCommand(factory)) + return cmd +} diff --git a/internal/cmd/api/proxy.go b/internal/cmd/api/proxy.go new file mode 100644 index 0000000..f9730d6 --- /dev/null +++ b/internal/cmd/api/proxy.go @@ -0,0 +1,307 @@ +package api + +import ( + "context" + "crypto/tls" + "crypto/x509" + "errors" + "fmt" + "net" + "net/http" + "net/url" + "os" + "os/signal" + "syscall" + "time" + + "github.com/spf13/cobra" + "k8s.io/kubectl/pkg/util/templates" + + "go.datum.net/datumctl/internal/apiproxy" + "go.datum.net/datumctl/internal/authutil" + "go.datum.net/datumctl/internal/client" + "go.datum.net/datumctl/internal/datumconfig" + customerrors "go.datum.net/datumctl/internal/errors" + "go.datum.net/datumctl/internal/miloapi" +) + +// shutdownGrace is how long in-flight requests get to finish after the first +// interrupt before their connections are cut. +const shutdownGrace = 2 * time.Second + +func proxyCommand(factory *client.DatumCloudFactory) *cobra.Command { + var ( + port int + sessionName string + quiet bool + ) + + cmd := &cobra.Command{ + Use: "proxy", + Short: "Start a local proxy that authenticates requests to the Datum Cloud API", + Long: templates.LongDesc(` + Start a local proxy that authenticates requests to the Datum Cloud API. + + The proxy listens on 127.0.0.1 and forwards every request to the API + endpoint of your datumctl session, adding your credentials automatically + and refreshing them as needed. Point any local tool at the printed URL — + no tokens to copy, no expiry to manage. + + By default the proxy serves the full API endpoint, so requests use the + same paths as the real API. Pass --project or --organization to serve a + single control plane instead, with shorter paths. + + The session and scope are pinned when the proxy starts. Switching your + active account or context does not affect a running proxy.`), + Example: templates.Examples(` + # Start a proxy on a fixed port for a dev server + datumctl api proxy --port 8001 + + # Start on a random free port; the URL is printed on the first stdout line + datumctl api proxy + + # List organizations through the proxy + curl http://127.0.0.1:8001/apis/resourcemanager.miloapis.com/v1alpha1/organizations + + # Watch DNS zones on a project control plane through the proxy + curl "http://127.0.0.1:8001/apis/resourcemanager.miloapis.com/v1alpha1/projects/my-project/control-plane/apis/networking.datumapis.com/v1alpha/dnszones?watch=true" + + # Serve one project's control plane directly, for shorter URLs + datumctl api proxy --port 8001 --project my-project + curl "http://127.0.0.1:8001/apis/networking.datumapis.com/v1alpha/dnszones?watch=true" + + # Pin a non-active session + datumctl api proxy --session sam@datum.net@api.staging.env.datum.net`), + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, args []string) error { + return runProxy(cmd, factory, port, sessionName, quiet) + }, + } + + cmd.Flags().IntVar(&port, "port", 0, "Local port to listen on (default: a random free port)") + cmd.Flags().StringVar(&sessionName, "session", "", "Pin a specific session by name (defaults to the active session; see 'datumctl auth list')") + cmd.Flags().BoolVarP(&quiet, "quiet", "q", false, "Suppress per-request log lines") + return cmd +} + +// proxyTarget is everything about the upstream that gets pinned when the +// proxy starts: the session it serves, the resolved endpoint identity, the +// upstream root, and the human-readable scope shown in the banner. +type proxyTarget struct { + session *datumconfig.Session + endpoint *client.SessionEndpoint + upstream *url.URL + scope string +} + +// resolveProxyTarget resolves the pinned session and upstream for the proxy. +// +// Unlike resource commands, scope comes only from the explicit flags passed +// here: with no scope flag the upstream is the endpoint root — never the +// active context and never DATUM_* environment variables. Tools cache the +// proxy URL, so its meaning must not depend on ambient state at launch (see +// the api-proxy proposal, "Path semantics"). +func resolveProxyTarget(cfg *datumconfig.ConfigV1Beta1, sessionName, project, organization string, platformWide bool) (*proxyTarget, error) { + if platformWide && (project != "" || organization != "") { + return nil, customerrors.NewUserError("--platform-wide cannot be used with --project or --organization") + } + if project != "" && organization != "" { + return nil, customerrors.NewUserError("only one of --project or --organization may be set") + } + + session, endpoint, err := client.ResolveSessionEndpoint(cfg, sessionName) + if err != nil { + if authutil.IsNoActiveUser(err) { + return nil, customerrors.NewUserErrorWithHint( + "No datumctl session found.", + "Run 'datumctl login' to authenticate, then start the proxy again.", + ) + } + return nil, err + } + + upstream := endpoint.BaseServer + scope := "full endpoint (use --project/--organization to serve one control plane)" + switch { + case platformWide: + scope = "platform-wide" + case organization != "": + upstream = miloapi.OrgControlPlaneURL(endpoint.BaseServer, organization) + scope = "organization " + organization + case project != "": + upstream = miloapi.ProjectControlPlaneURL(endpoint.BaseServer, project) + scope = "project " + project + } + + upstreamURL, err := url.Parse(upstream) + if err != nil { + return nil, customerrors.WrapUserErrorWithHint( + fmt.Sprintf("The session's API endpoint produced an invalid upstream URL (%q).", upstream), + "Run 'datumctl login' again to refresh the session's endpoint, or check the session with 'datumctl auth list'.", + err, + ) + } + + return &proxyTarget{ + session: session, + endpoint: endpoint, + upstream: upstreamURL, + scope: scope, + }, nil +} + +// tlsClientConfig converts the session endpoint's TLS settings into a +// *tls.Config for the upstream transport. Returns nil when the endpoint +// declares nothing, keeping Go's defaults. +func tlsClientConfig(epTLS client.EndpointTLS) (*tls.Config, error) { + if epTLS.ServerName == "" && !epTLS.InsecureSkipTLSVerify && len(epTLS.CAData) == 0 { + return nil, nil + } + cfg := &tls.Config{ + ServerName: epTLS.ServerName, + InsecureSkipVerify: epTLS.InsecureSkipTLSVerify, + } + if len(epTLS.CAData) > 0 { + pool := x509.NewCertPool() + if !pool.AppendCertsFromPEM(epTLS.CAData) { + return nil, customerrors.NewUserErrorWithHint( + "No certificates could be parsed from the session's certificate authority data.", + "Run 'datumctl login' again to refresh the session's endpoint settings.", + ) + } + cfg.RootCAs = pool + } + return cfg, nil +} + +// sessionLabel names the identity the proxy serves, for the startup banner. +func sessionLabel(target *proxyTarget) string { + if target.session != nil { + return fmt.Sprintf("%s (%s)", target.session.UserEmail, datumconfig.StripScheme(target.endpoint.BaseServer)) + } + return target.endpoint.UserKey +} + +func runProxy(cmd *cobra.Command, factory *client.DatumCloudFactory, port int, sessionName string, quiet bool) error { + cfg, err := datumconfig.LoadAuto() + if err != nil { + return err + } + if err := authutil.EnsureUserKeysMigrated(cfg); err != nil { + return err + } + + flags := factory.ConfigFlags + target, err := resolveProxyTarget(cfg, sessionName, + stringValue(flags.Project), stringValue(flags.Organization), boolValue(flags.PlatformWide)) + if err != nil { + return err + } + + tokenSource, err := authutil.GetTokenSourceForUser(cmd.Context(), target.endpoint.UserKey) + if err != nil { + if _, isUser := customerrors.IsUserError(err); isUser { + return err + } + return customerrors.WrapUserErrorWithHint( + fmt.Sprintf("No stored credentials found for %s.", sessionLabel(target)), + "Run 'datumctl login' to authenticate, then start the proxy again.", + err, + ) + } + + tlsConfig, err := tlsClientConfig(target.endpoint.TLS) + if err != nil { + return err + } + + errOut := cmd.ErrOrStderr() + server, err := apiproxy.New(apiproxy.Config{ + Upstream: target.upstream, + TokenSource: tokenSource, + TLSClientConfig: tlsConfig, + LogWriter: errOut, + Quiet: quiet, + }) + if err != nil { + return err + } + + // Loopback only, by design: there is no flag to bind other addresses. + listener, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", port)) + if err != nil { + if port != 0 { + return customerrors.WrapUserErrorWithHint( + fmt.Sprintf("Could not listen on 127.0.0.1:%d.", port), + "The port may already be in use — pass a different --port, or omit --port to pick a random free port.", + err, + ) + } + return customerrors.WrapUserError("Could not open a local listener on 127.0.0.1.", err) + } + localURL := "http://" + listener.Addr().String() + + // Register the signal handler before advertising readiness, so a harness + // that reads the URL line and later interrupts the proxy can never signal + // an unhandled process. + signals := make(chan os.Signal, 2) + signal.Notify(signals, os.Interrupt, syscall.SIGTERM) + defer signal.Stop(signals) + + fmt.Fprintf(errOut, " Session: %s\n", sessionLabel(target)) + fmt.Fprintf(errOut, " Upstream: %s\n", target.upstream) + fmt.Fprintf(errOut, " Scope: %s\n", target.scope) + fmt.Fprintf(errOut, " Listening: %s\n", localURL) + fmt.Fprintln(errOut) + if quiet { + fmt.Fprintln(errOut, " Press Ctrl+C to stop.") + } else { + fmt.Fprintln(errOut, " Press Ctrl+C to stop. Requests are logged below (silence with --quiet).") + } + fmt.Fprintln(errOut) + + // Machine-readable readiness contract: the bare URL is the first and only + // stdout line, printed only after the listener is bound. + fmt.Fprintln(cmd.OutOrStdout(), localURL) + + serveErr := make(chan error, 1) + go func() { serveErr <- server.Serve(listener) }() + + select { + case err := <-serveErr: + if err != nil && !errors.Is(err, http.ErrServerClosed) { + return err + } + return nil + case <-signals: + } + + // Graceful shutdown: the listener closes and in-flight requests get the + // grace period before their connections are cut. A second signal exits + // immediately. + fmt.Fprintln(errOut, "Shutting down...") + shutdownDone := make(chan struct{}) + go func() { + ctx, cancel := context.WithTimeout(context.Background(), shutdownGrace) + defer cancel() + _ = server.Shutdown(ctx) + close(shutdownDone) + }() + + select { + case <-shutdownDone: + case <-signals: + } + return nil +} + +func stringValue(p *string) string { + if p == nil { + return "" + } + return *p +} + +func boolValue(p *bool) bool { + return p != nil && *p +} diff --git a/internal/cmd/api/proxy_lifecycle_test.go b/internal/cmd/api/proxy_lifecycle_test.go new file mode 100644 index 0000000..59aacea --- /dev/null +++ b/internal/cmd/api/proxy_lifecycle_test.go @@ -0,0 +1,139 @@ +package api + +import ( + "bufio" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "net/url" + "os" + "strings" + "syscall" + "testing" + "time" + + "golang.org/x/oauth2" + + "go.datum.net/datumctl/internal/authutil" + "go.datum.net/datumctl/internal/client" + "go.datum.net/datumctl/internal/datumconfig" + "go.datum.net/datumctl/internal/keyring" +) + +// TestProxyLifecycle drives the real 'datumctl api proxy' command end to end, +// hermetically: HOME points at a temp dir (so the config and any credential +// fallback file live there), the keyring is mocked, and the session endpoint +// is an httptest upstream. It pins the readiness contract — with --port 0 the +// first stdout line parses as a URL and a request to it succeeds — and that +// SIGINT terminates the command within the shutdown grace period. +func TestProxyLifecycle(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + keyring.MockInit() + + const userKey = "maya@datum.net@auth.datum.net" + + var authHeader string + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + authHeader = r.Header.Get("Authorization") + io.WriteString(w, "ok") + })) + defer upstream.Close() + + // Config fixture: one active session whose endpoint is the upstream. + cfg := datumconfig.NewV1Beta1() + cfg.Sessions = []datumconfig.Session{{ + Name: "maya@datum.net@api.datum.net", + UserKey: userKey, + UserEmail: "maya@datum.net", + Endpoint: datumconfig.Endpoint{ + Server: upstream.URL, + AuthHostname: "auth.datum.net", + }, + }} + cfg.ActiveSession = "maya@datum.net@api.datum.net" + if err := datumconfig.SaveV1Beta1(cfg); err != nil { + t.Fatalf("save config fixture: %v", err) + } + + // Keyring fixture: a token valid well past the test, so no refresh runs. + blob, err := json.Marshal(authutil.StoredCredentials{ + Hostname: "auth.datum.net", + APIHostname: "api.datum.net", + UserEmail: "maya@datum.net", + Token: &oauth2.Token{ + AccessToken: "lifecycle-token", + Expiry: time.Now().Add(time.Hour), + }, + }) + if err != nil { + t.Fatalf("marshal creds: %v", err) + } + if err := keyring.Set(authutil.ServiceName, userKey, string(blob)); err != nil { + t.Fatalf("seed keyring: %v", err) + } + + factory := &client.DatumCloudFactory{ConfigFlags: &client.CustomConfigFlags{}} + cmd := proxyCommand(factory) + stdoutR, stdoutW := io.Pipe() + cmd.SetOut(stdoutW) + cmd.SetErr(io.Discard) + cmd.SetArgs([]string{"--port", "0", "--quiet"}) + + done := make(chan error, 1) + go func() { done <- cmd.Execute() }() + + // Readiness contract: the first stdout line is the bare proxy URL, + // printed once the listener is serving (and the signal handler is armed). + lines := make(chan string, 1) + go func() { + line, _ := bufio.NewReader(stdoutR).ReadString('\n') + lines <- line + }() + var readyLine string + select { + case readyLine = <-lines: + case err := <-done: + t.Fatalf("proxy exited before printing its URL: %v", err) + case <-time.After(10 * time.Second): + t.Fatal("timed out waiting for the readiness line on stdout") + } + proxyURL, err := url.Parse(strings.TrimSpace(readyLine)) + if err != nil { + t.Fatalf("first stdout line %q does not parse as a URL: %v", readyLine, err) + } + if proxyURL.Scheme != "http" || !strings.HasPrefix(proxyURL.Host, "127.0.0.1:") { + t.Fatalf("proxy URL = %q, want http://127.0.0.1:", proxyURL) + } + + // A request to the printed URL succeeds and carries the injected token. + resp, err := http.Get(proxyURL.String() + "/apis/resourcemanager.miloapis.com/v1alpha1/organizations") + if err != nil { + t.Fatalf("request through the proxy: %v", err) + } + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200", resp.StatusCode) + } + if string(body) != "ok" { + t.Fatalf("body = %q, want the upstream body", body) + } + if authHeader != "Bearer lifecycle-token" { + t.Fatalf("upstream Authorization = %q, want the injected session token", authHeader) + } + + // SIGINT terminates within the grace period with a clean exit. + if err := syscall.Kill(os.Getpid(), syscall.SIGINT); err != nil { + t.Fatalf("send SIGINT: %v", err) + } + select { + case err := <-done: + if err != nil { + t.Fatalf("proxy exited with error after SIGINT: %v", err) + } + case <-time.After(shutdownGrace + 5*time.Second): + t.Fatal("proxy did not terminate within the shutdown grace period after SIGINT") + } + stdoutW.Close() +} diff --git a/internal/cmd/api/proxy_test.go b/internal/cmd/api/proxy_test.go new file mode 100644 index 0000000..20b4656 --- /dev/null +++ b/internal/cmd/api/proxy_test.go @@ -0,0 +1,231 @@ +package api + +import ( + "strings" + "testing" + + "go.datum.net/datumctl/internal/client" + "go.datum.net/datumctl/internal/datumconfig" + customerrors "go.datum.net/datumctl/internal/errors" +) + +// fixtureConfig returns a config with an active session AND an active +// project-scoped context. Every session carries a UserKey and an endpoint +// Server so resolution never has to touch the keyring. +func fixtureConfig() *datumconfig.ConfigV1Beta1 { + return &datumconfig.ConfigV1Beta1{ + APIVersion: datumconfig.V1Beta1APIVersion, + Kind: datumconfig.DefaultKind, + Sessions: []datumconfig.Session{ + { + Name: "maya@datum.net@api.datum.net", + UserKey: "sub-maya@auth.datum.net", + UserEmail: "maya@datum.net", + Endpoint: datumconfig.Endpoint{ + Server: "https://api.datum.net", + AuthHostname: "auth.datum.net", + }, + }, + { + Name: "sam@datum.net@api.staging.env.datum.net", + UserKey: "sub-sam@auth.staging.env.datum.net", + UserEmail: "sam@datum.net", + Endpoint: datumconfig.Endpoint{ + Server: "https://api.staging.env.datum.net", + AuthHostname: "auth.staging.env.datum.net", + }, + }, + }, + Contexts: []datumconfig.DiscoveredContext{ + { + Name: "ctx-org/ctx-project", + Session: "maya@datum.net@api.datum.net", + OrganizationID: "ctx-org", + ProjectID: "ctx-project", + }, + }, + CurrentContext: "ctx-org/ctx-project", + ActiveSession: "maya@datum.net@api.datum.net", + } +} + +// TestResolveProxyTarget_NoScopeFlagsTargetsEndpointRoot pins the +// no-context-inheritance rule: even with an active project context configured +// and DATUM_* scope variables set, a bare proxy targets the endpoint root. +func TestResolveProxyTarget_NoScopeFlagsTargetsEndpointRoot(t *testing.T) { + t.Setenv("DATUM_PROJECT", "env-project") + t.Setenv("DATUM_ORGANIZATION", "env-org") + + target, err := resolveProxyTarget(fixtureConfig(), "", "", "", false) + if err != nil { + t.Fatalf("resolveProxyTarget: %v", err) + } + + if got, want := target.upstream.String(), "https://api.datum.net"; got != want { + t.Errorf("upstream = %q, want the endpoint root %q (no active-context or env inheritance)", got, want) + } + if !strings.Contains(target.scope, "full endpoint") { + t.Errorf("scope = %q, want it to describe the full endpoint", target.scope) + } +} + +func TestResolveProxyTarget_ScopedFlags(t *testing.T) { + tests := []struct { + name string + project string + organization string + platformWide bool + wantUpstream string + wantScope string + }{ + { + name: "project scope", + project: "my-project", + wantUpstream: "https://api.datum.net/apis/resourcemanager.miloapis.com/v1alpha1/projects/my-project/control-plane", + wantScope: "project my-project", + }, + { + name: "organization scope", + organization: "my-org", + wantUpstream: "https://api.datum.net/apis/resourcemanager.miloapis.com/v1alpha1/organizations/my-org/control-plane", + wantScope: "organization my-org", + }, + { + name: "platform-wide scope", + platformWide: true, + wantUpstream: "https://api.datum.net", + wantScope: "platform-wide", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + target, err := resolveProxyTarget(fixtureConfig(), "", tc.project, tc.organization, tc.platformWide) + if err != nil { + t.Fatalf("resolveProxyTarget: %v", err) + } + if got := target.upstream.String(); got != tc.wantUpstream { + t.Errorf("upstream = %q, want %q", got, tc.wantUpstream) + } + if target.scope != tc.wantScope { + t.Errorf("scope = %q, want %q", target.scope, tc.wantScope) + } + }) + } +} + +// TestResolveProxyTarget_ScopeUsesPinnedSessionEndpoint proves scoped mode +// builds the control-plane URL on the pinned session's endpoint, not the +// active session's. +func TestResolveProxyTarget_ScopeUsesPinnedSessionEndpoint(t *testing.T) { + target, err := resolveProxyTarget(fixtureConfig(), "sam@datum.net@api.staging.env.datum.net", "my-project", "", false) + if err != nil { + t.Fatalf("resolveProxyTarget: %v", err) + } + want := "https://api.staging.env.datum.net/apis/resourcemanager.miloapis.com/v1alpha1/projects/my-project/control-plane" + if got := target.upstream.String(); got != want { + t.Errorf("upstream = %q, want %q", got, want) + } +} + +func TestResolveProxyTarget_PinnedSession(t *testing.T) { + target, err := resolveProxyTarget(fixtureConfig(), "sam@datum.net@api.staging.env.datum.net", "", "", false) + if err != nil { + t.Fatalf("resolveProxyTarget: %v", err) + } + if got, want := target.upstream.String(), "https://api.staging.env.datum.net"; got != want { + t.Errorf("upstream = %q, want %q", got, want) + } + if target.session == nil || target.session.UserEmail != "sam@datum.net" { + t.Errorf("session = %+v, want the pinned staging session", target.session) + } + if got, want := target.endpoint.UserKey, "sub-sam@auth.staging.env.datum.net"; got != want { + t.Errorf("user key = %q, want %q", got, want) + } +} + +func TestResolveProxyTarget_UnknownSessionIsUserError(t *testing.T) { + _, err := resolveProxyTarget(fixtureConfig(), "nobody@datum.net@api.datum.net", "", "", false) + if err == nil { + t.Fatal("expected an error for an unknown --session") + } + userErr, ok := customerrors.IsUserError(err) + if !ok { + t.Fatalf("error = %v (%T), want a UserError", err, err) + } + if !strings.Contains(userErr.Message, "nobody@datum.net@api.datum.net") { + t.Errorf("message = %q, want it to name the unknown session", userErr.Message) + } + if !strings.Contains(userErr.Hint, "datumctl auth list") { + t.Errorf("hint = %q, want it to point at 'datumctl auth list'", userErr.Hint) + } +} + +func TestResolveProxyTarget_ConflictingScopeFlags(t *testing.T) { + tests := []struct { + name string + project string + organization string + platformWide bool + }{ + {name: "project and organization", project: "p", organization: "o"}, + {name: "platform-wide and project", project: "p", platformWide: true}, + {name: "platform-wide and organization", organization: "o", platformWide: true}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + _, err := resolveProxyTarget(fixtureConfig(), "", tc.project, tc.organization, tc.platformWide) + if err == nil { + t.Fatal("expected an error for conflicting scope flags") + } + if _, ok := customerrors.IsUserError(err); !ok { + t.Errorf("error = %v (%T), want a UserError", err, err) + } + }) + } +} + +func TestTLSClientConfig(t *testing.T) { + t.Run("empty settings keep defaults", func(t *testing.T) { + cfg, err := tlsClientConfig(client.EndpointTLS{}) + if err != nil { + t.Fatalf("tlsClientConfig: %v", err) + } + if cfg != nil { + t.Errorf("config = %+v, want nil for default TLS", cfg) + } + }) + + t.Run("server name and insecure carry over", func(t *testing.T) { + cfg, err := tlsClientConfig(client.EndpointTLS{ + ServerName: "internal.example", + InsecureSkipTLSVerify: true, + }) + if err != nil { + t.Fatalf("tlsClientConfig: %v", err) + } + if cfg == nil { + t.Fatal("config = nil, want TLS settings applied") + } + if cfg.ServerName != "internal.example" { + t.Errorf("ServerName = %q, want %q", cfg.ServerName, "internal.example") + } + if !cfg.InsecureSkipVerify { + t.Error("InsecureSkipVerify = false, want true") + } + }) + + t.Run("invalid CA data errors", func(t *testing.T) { + _, err := tlsClientConfig(client.EndpointTLS{CAData: []byte("not a pem")}) + if err == nil { + t.Fatal("expected an error for unparseable CA data") + } + userErr, ok := customerrors.IsUserError(err) + if !ok { + t.Fatalf("error = %v (%T), want a UserError with actionable guidance", err, err) + } + if !strings.Contains(userErr.Hint, "datumctl login") { + t.Errorf("hint = %q, want it to point at 'datumctl login'", userErr.Hint) + } + }) +} diff --git a/internal/cmd/auth/auth.go b/internal/cmd/auth/auth.go index 51d7729..2091dfa 100644 --- a/internal/cmd/auth/auth.go +++ b/internal/cmd/auth/auth.go @@ -23,7 +23,7 @@ Advanced — kubectl integration: If you use kubectl and want to point it at a Datum Cloud control plane, see 'datumctl auth update-kubeconfig --help'.`, Example: ` # Log in to Datum Cloud - datumctl auth login + datumctl login # Show all logged-in accounts datumctl auth list diff --git a/internal/cmd/auth/get_token.go b/internal/cmd/auth/get_token.go index 8278761..9806029 100644 --- a/internal/cmd/auth/get_token.go +++ b/internal/cmd/auth/get_token.go @@ -91,7 +91,7 @@ func runGetToken(cmd *cobra.Command, args []string) error { tokenSource, err = authutil.GetTokenSource(ctx) if err != nil { if errors.Is(err, authutil.ErrNoActiveUser) { - return errors.New("no active user found in keyring. Please login first using 'datumctl auth login'") + return errors.New("no active user found in keyring. Please login first using 'datumctl login'") } return fmt.Errorf("failed to get token source: %w", err) } diff --git a/internal/cmd/root.go b/internal/cmd/root.go index 65eff19..d0a1f70 100644 --- a/internal/cmd/root.go +++ b/internal/cmd/root.go @@ -29,6 +29,7 @@ import ( "go.datum.net/datumctl/internal/client" aicmd "go.datum.net/datumctl/internal/cmd/ai" + apicmd "go.datum.net/datumctl/internal/cmd/api" "go.datum.net/datumctl/internal/cmd/auth" "go.datum.net/datumctl/internal/cmd/console" "go.datum.net/datumctl/internal/cmd/create" @@ -716,6 +717,10 @@ Specify the resource type and name to view its history.` consoleCmd.GroupID = "other" rootCmd.AddCommand(consoleCmd) + apiCmd := apicmd.Command(factory) + apiCmd.GroupID = "other" + rootCmd.AddCommand(apiCmd) + pluginCommand := plugincmd.Command(factory) pluginCommand.GroupID = "other" rootCmd.AddCommand(pluginCommand)