From 9424daff28299b63005ca38afb71c256f237973b Mon Sep 17 00:00:00 2001 From: Pasha Sviderski Date: Tue, 26 May 2026 10:07:06 +1000 Subject: [PATCH] feat: enable auth for local requests to Corrosion API (resolves #110) --- internal/corrosion/client.go | 70 ++++++++++++++++-------- internal/corrosion/client_test.go | 60 ++++++++++++++++++++ internal/machine/corroservice/config.go | 7 ++- internal/machine/corroservice/service.go | 2 +- internal/machine/machine.go | 23 +++++++- internal/machine/state.go | 3 + 6 files changed, 138 insertions(+), 27 deletions(-) create mode 100644 internal/corrosion/client_test.go diff --git a/internal/corrosion/client.go b/internal/corrosion/client.go index 4b3522b6..543e47ee 100644 --- a/internal/corrosion/client.go +++ b/internal/corrosion/client.go @@ -33,39 +33,44 @@ type APIClient struct { newResubBackoff func() backoff.BackOff } -// NewAPIClient creates a new Corrosion API client. The client retries on network errors using an exponential backoff -// policy with a maximum interval of 1 second and a maximum elapsed time of 10 seconds. +// NewAPIClient creates a new Corrosion API client. The bearerToken is sent in the Authorization header of every +// request to authenticate against Corrosion API. +// The client retries on network errors using an exponential backoff policy with a maximum interval of 1 second and +// a maximum elapsed time of 10 seconds. // It automatically resubscribes to active subscriptions if an error occurs using an exponential backoff policy with a // maximum interval of 1 second and a maximum elapsed time of 60 seconds. // Use the WithHTTP2Client option to provide a custom HTTP client and the WithResubscribeBackoff option to change the // backoff policy for resubscribing to a query. -func NewAPIClient(addr netip.AddrPort, opts ...APIClientOption) (*APIClient, error) { +func NewAPIClient(addr netip.AddrPort, bearerToken string, opts ...APIClientOption) (*APIClient, error) { baseURL, err := url.Parse(fmt.Sprintf("http://%s", addr)) if err != nil { return nil, fmt.Errorf("invalid URL: %w", err) } - c := &APIClient{ - baseURL: baseURL, - client: &http.Client{ - Transport: &RetryRoundTripper{ - Base: &http2.Transport{ - AllowHTTP: true, - DialTLSContext: func(ctx context.Context, network, addr string, _ *tls.Config) (net.Conn, error) { - dialer := &net.Dialer{ - Timeout: http2ConnectTimeout, - } - return dialer.DialContext(ctx, network, addr) - }, - }, - NewBackoff: func() backoff.BackOff { - return backoff.NewExponentialBackOff( - backoff.WithInitialInterval(100*time.Millisecond), - backoff.WithMaxInterval(1*time.Second), - backoff.WithMaxElapsedTime(http2MaxRetryTime), - ) + + transport := &AuthRoundTripper{ + Base: &RetryRoundTripper{ + Base: &http2.Transport{ + AllowHTTP: true, + DialTLSContext: func(ctx context.Context, network, addr string, _ *tls.Config) (net.Conn, error) { + dialer := &net.Dialer{ + Timeout: http2ConnectTimeout, + } + return dialer.DialContext(ctx, network, addr) }, }, + NewBackoff: func() backoff.BackOff { + return backoff.NewExponentialBackOff( + backoff.WithInitialInterval(100*time.Millisecond), + backoff.WithMaxInterval(1*time.Second), + backoff.WithMaxElapsedTime(http2MaxRetryTime), + ) + }, }, + Token: bearerToken, + } + c := &APIClient{ + baseURL: baseURL, + client: &http.Client{Transport: transport}, newResubBackoff: func() backoff.BackOff { return backoff.NewExponentialBackOff( backoff.WithInitialInterval(100*time.Millisecond), @@ -83,6 +88,8 @@ func NewAPIClient(addr netip.AddrPort, opts ...APIClientOption) (*APIClient, err type APIClientOption func(*APIClient) +// WithHTTP2Client replaces the client's HTTP transport. The provided client bypasses the built-in bearer-token +// injection, so the caller is responsible for setting the Authorization header. func WithHTTP2Client(client *http.Client) APIClientOption { return func(c *APIClient) { c.client = client @@ -97,6 +104,22 @@ func WithResubscribeBackoff(newBackoff func() backoff.BackOff) APIClientOption { } } +// AuthRoundTripper sets the Authorization header on every outgoing request. An empty Token leaves the header untouched. +type AuthRoundTripper struct { + Base http.RoundTripper + Token string +} + +func (rt *AuthRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { + if rt.Token == "" { + return rt.Base.RoundTrip(req) + } + // RoundTripper contract: must not mutate the caller's request. + req = req.Clone(req.Context()) + req.Header.Set("Authorization", "Bearer "+rt.Token) + return rt.Base.RoundTrip(req) +} + type RetryRoundTripper struct { Base http.RoundTripper // NewBackoff creates a new backoff policy for each request. @@ -107,8 +130,7 @@ func (rt *RetryRoundTripper) RoundTrip(req *http.Request) (*http.Response, error roundTrip := func() (*http.Response, error) { resp, err := rt.Base.RoundTrip(req) if err != nil { - var opErr *net.OpError - if errors.As(err, &opErr) { + if _, ok := errors.AsType[*net.OpError](err); ok { // Not certain, but I expect operational errors should generally be retryable. slog.Debug("Retrying corrosion API request due to network error.", "error", err) return nil, err diff --git a/internal/corrosion/client_test.go b/internal/corrosion/client_test.go new file mode 100644 index 00000000..157c3f84 --- /dev/null +++ b/internal/corrosion/client_test.go @@ -0,0 +1,60 @@ +package corrosion + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestAuthRoundTripper_SetsAuthorizationHeader(t *testing.T) { + t.Parallel() + + const token = "test-token-1234567890" + + var gotAuth string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotAuth = r.Header.Get("Authorization") + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + rt := &AuthRoundTripper{Base: http.DefaultTransport, Token: token} + client := &http.Client{Transport: rt} + + req, err := http.NewRequest(http.MethodGet, srv.URL, nil) + require.NoError(t, err) + + resp, err := client.Do(req) + require.NoError(t, err) + resp.Body.Close() + + assert.Equal(t, "Bearer "+token, gotAuth) + // Caller's request must not be mutated by the RoundTripper. + assert.Empty(t, req.Header.Get("Authorization")) +} + +func TestAuthRoundTripper_EmptyTokenSkipsHeader(t *testing.T) { + t.Parallel() + + var headerSeen bool + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, headerSeen = r.Header["Authorization"] + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + rt := &AuthRoundTripper{Base: http.DefaultTransport, Token: ""} + client := &http.Client{Transport: rt} + + req, err := http.NewRequest(http.MethodGet, srv.URL, nil) + require.NoError(t, err) + + resp, err := client.Do(req) + require.NoError(t, err) + resp.Body.Close() + + assert.False(t, headerSeen, "Authorization header should not be sent when token is empty") +} diff --git a/internal/machine/corroservice/config.go b/internal/machine/corroservice/config.go index a4d33a83..bc47fc88 100644 --- a/internal/machine/corroservice/config.go +++ b/internal/machine/corroservice/config.go @@ -37,7 +37,12 @@ type GossipConfig struct { } type APIConfig struct { - Addr netip.AddrPort `toml:"addr"` + Addr netip.AddrPort `toml:"addr"` + Authz APIAuthzConfig `toml:"authz"` +} + +type APIAuthzConfig struct { + BearerToken string `toml:"bearer-token"` } type AdminConfig struct { diff --git a/internal/machine/corroservice/service.go b/internal/machine/corroservice/service.go index d7aef164..ec521503 100644 --- a/internal/machine/corroservice/service.go +++ b/internal/machine/corroservice/service.go @@ -35,7 +35,7 @@ func WaitReady(ctx context.Context, dataDir string) error { return fmt.Errorf("unmarshal config: %w", err) } - corro, err := corrosion.NewAPIClient(config.API.Addr) + corro, err := corrosion.NewAPIClient(config.API.Addr, config.API.Authz.BearerToken) if err != nil { return fmt.Errorf("create corrosion API client: %w", err) } diff --git a/internal/machine/machine.go b/internal/machine/machine.go index 049712dc..5cd05aac 100644 --- a/internal/machine/machine.go +++ b/internal/machine/machine.go @@ -35,6 +35,7 @@ import ( machinedocker "github.com/psviderski/uncloud/internal/machine/docker" "github.com/psviderski/uncloud/internal/machine/network" "github.com/psviderski/uncloud/internal/machine/store" + "github.com/psviderski/uncloud/internal/secret" "github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/unregistry" "github.com/siderolabs/grpc-proxy/proxy" @@ -224,7 +225,21 @@ func NewMachine(config *Config) (*Machine, error) { } } - corro, err := corrosion.NewAPIClient(config.CorrosionAPIAddr) + // Generate and persist a token for Corrosion API if not already present in the state. + if len(state.CorrosionAPIToken) == 0 { + token, tErr := secret.New(16) + if tErr != nil { + return nil, fmt.Errorf("generate corrosion API token: %w", tErr) + } + + state.CorrosionAPIToken = token + if err = state.Save(); err != nil { + return nil, fmt.Errorf("save machine state with corrosion API token: %w", err) + } + slog.Info("Generated Corrosion API bearer token.") + } + + corro, err := corrosion.NewAPIClient(config.CorrosionAPIAddr, state.CorrosionAPIToken.String()) if err != nil { return nil, fmt.Errorf("create corrosion API client: %w", err) } @@ -603,6 +618,9 @@ func listenUnixSocket(path string) (net.Listener, error) { } func (m *Machine) configureCorrosion() error { + if len(m.state.CorrosionAPIToken) == 0 { + return fmt.Errorf("corrosion API token not set in machine state") + } if err := corroservice.MkDir(m.config.CorrosionDataDir, m.config.CorrosionUser); err != nil { return fmt.Errorf("create corrosion data directory: %w", err) } @@ -639,6 +657,9 @@ func (m *Machine) configureCorrosion() error { }, API: corroservice.APIConfig{ Addr: m.config.CorrosionAPIAddr, + Authz: corroservice.APIAuthzConfig{ + BearerToken: m.state.CorrosionAPIToken.String(), + }, }, Admin: corroservice.AdminConfig{ Path: m.config.CorrosionAdminSockPath, diff --git a/internal/machine/state.go b/internal/machine/state.go index 25892834..b78c7432 100644 --- a/internal/machine/state.go +++ b/internal/machine/state.go @@ -8,6 +8,7 @@ import ( "sync" "github.com/psviderski/uncloud/internal/machine/network" + "github.com/psviderski/uncloud/internal/secret" ) const ( @@ -28,6 +29,8 @@ type State struct { // Per-actor vector (Corrosion actor UUID → max applied db_version) captured from an existing // member at join time. Cleared once reached. MinStoreVersion map[string]int64 `json:",omitempty"` + // CorrosionAPIToken authenticates requests to the local Corrosion API. + CorrosionAPIToken secret.Secret `json:",omitempty"` // path is the file path config is read from and saved to. path string