diff --git a/infra/feast-operator/Dockerfile b/infra/feast-operator/Dockerfile index 811c4d31c84..a7ad3fa044d 100644 --- a/infra/feast-operator/Dockerfile +++ b/infra/feast-operator/Dockerfile @@ -12,7 +12,7 @@ COPY --chown=1001:0 go.sum go.sum RUN go mod download # Copy the go source -COPY --chown=1001:0 cmd/main.go cmd/main.go +COPY --chown=1001:0 cmd/ cmd/ COPY --chown=1001:0 api/ api/ COPY --chown=1001:0 internal/controller/ internal/controller/ @@ -21,7 +21,7 @@ COPY --chown=1001:0 internal/controller/ internal/controller/ # was called. For example, if we call make docker-build in a local env which has the Apple Silicon M1 SO # the docker BUILDPLATFORM arg will be linux/arm64 when for Apple x86 it will be linux/amd64. Therefore, # by leaving it empty we can ensure that the container and binary shipped on it will have the same platform. -RUN CGO_ENABLED=0 GOOS=${TARGETOS:-linux} GOARCH=${TARGETARCH} go build -a -o manager cmd/main.go +RUN CGO_ENABLED=0 GOOS=${TARGETOS:-linux} GOARCH=${TARGETARCH} go build -a -o manager ./cmd/ FROM registry.access.redhat.com/ubi9/ubi-minimal:9.8 WORKDIR / diff --git a/infra/feast-operator/cmd/main.go b/infra/feast-operator/cmd/main.go index 0d833f1469b..5d2bbece7dc 100644 --- a/infra/feast-operator/cmd/main.go +++ b/infra/feast-operator/cmd/main.go @@ -34,8 +34,6 @@ import ( corev1 "k8s.io/api/core/v1" policyv1 "k8s.io/api/policy/v1" rbacv1 "k8s.io/api/rbac/v1" - apierrors "k8s.io/apimachinery/pkg/api/errors" - apimeta "k8s.io/apimachinery/pkg/api/meta" "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/runtime" utilruntime "k8s.io/apimachinery/pkg/util/runtime" @@ -102,7 +100,7 @@ func main() { var probeAddr string var secureMetrics bool var featureStoreMetrics bool - var tlsOpts []func(*tls.Config) + tlsOpts := make([]func(*tls.Config), 0, 2) flag.StringVar(&metricsAddr, "metrics-bind-address", "0", "The address the metrics endpoint binds to. "+ "Use :8443 for HTTPS or :8080 for HTTP, or leave as 0 to disable the metrics service.") flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.") @@ -130,46 +128,12 @@ func main() { os.Exit(1) } - tlsProfileFetched := false - tlsProfile, err := tlspkg.FetchAPIServerTLSProfile(context.Background(), bootstrapClient) + tlsResult, err := bootstrapTLS(context.Background(), bootstrapClient) if err != nil { - switch { - case apimeta.IsNoMatchError(err): - setupLog.Info("TLS profile not available, using hardened defaults (non-OpenShift cluster)") - case apierrors.IsNotFound(err): - setupLog.Info("APIServer resource not found, using hardened defaults") - default: - setupLog.Error(err, "unable to read APIServer TLS profile, refusing to start with unknown TLS posture") - os.Exit(1) - } - } else { - tlsProfileFetched = true - tlsConfigFn, unsupported := tlspkg.NewTLSConfigFromProfile(tlsProfile) - if len(unsupported) > 0 { - setupLog.Info("TLS profile contains ciphers unsupported by Go", "unsupported", unsupported) - } - tlsOpts = append(tlsOpts, tlsConfigFn) - } - - tlsAdherenceFetched := false - tlsAdherence, err := tlspkg.FetchAPIServerTLSAdherencePolicy(context.Background(), bootstrapClient) - if err != nil { - switch { - case apimeta.IsNoMatchError(err): - setupLog.Info("TLS adherence policy not available (non-OpenShift cluster)") - case apierrors.IsNotFound(err): - setupLog.Info("APIServer resource not found, skipping adherence policy") - default: - setupLog.Error(err, "unable to read APIServer TLS adherence policy, refusing to start") - os.Exit(1) - } - } else { - tlsAdherenceFetched = true + setupLog.Error(err, "TLS bootstrap failed") + os.Exit(1) } - - tlsOpts = append(tlsOpts, func(c *tls.Config) { - c.NextProtos = []string{"h2", "http/1.1"} - }) + tlsOpts = append(tlsOpts, tlsResult.TLSOpts...) webhookServer := webhook.NewServer(webhook.Options{ TLSOpts: tlsOpts, @@ -271,17 +235,17 @@ func main() { ctx, cancel := context.WithCancel(ctrl.SetupSignalHandler()) defer cancel() - if tlsProfileFetched { + if tlsResult.ProfileFetched { watcher := &tlspkg.SecurityProfileWatcher{ Client: mgr.GetClient(), - InitialTLSProfileSpec: tlsProfile, + InitialTLSProfileSpec: tlsResult.ProfileSpec, OnProfileChange: func(_ context.Context, _, _ configv1.TLSProfileSpec) { setupLog.Info("TLS profile changed, initiating shutdown to reload") cancel() }, } - if tlsAdherenceFetched { - watcher.InitialTLSAdherencePolicy = tlsAdherence + if tlsResult.AdherenceFetched { + watcher.InitialTLSAdherencePolicy = tlsResult.AdherencePolicy watcher.OnAdherencePolicyChange = func(_ context.Context, _, _ configv1.TLSAdherencePolicy) { setupLog.Info("TLS adherence policy changed, initiating shutdown to reload") cancel() diff --git a/infra/feast-operator/cmd/tls_bootstrap.go b/infra/feast-operator/cmd/tls_bootstrap.go new file mode 100644 index 00000000000..6fe33631e02 --- /dev/null +++ b/infra/feast-operator/cmd/tls_bootstrap.go @@ -0,0 +1,134 @@ +/* +Copyright 2024 Feast Community. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package main + +import ( + "context" + "crypto/tls" + "errors" + "fmt" + "time" + + configv1 "github.com/openshift/api/config/v1" + tlspkg "github.com/openshift/controller-runtime-common/pkg/tls" + apierrors "k8s.io/apimachinery/pkg/api/errors" + apimeta "k8s.io/apimachinery/pkg/api/meta" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/log" +) + +const ( + tlsFetchTimeout = 10 * time.Second + alpnH2 = "h2" + alpnHTTP11 = "http/1.1" +) + +type tlsBootstrapResult struct { + TLSOpts []func(*tls.Config) + ProfileFetched bool + ProfileSpec configv1.TLSProfileSpec + AdherenceFetched bool + AdherencePolicy configv1.TLSAdherencePolicy + UnsupportedCiphers []string +} + +func fetchTLSProfile(ctx context.Context, k8sClient client.Client) (configv1.TLSProfileSpec, bool, error) { + fetchCtx, cancel := context.WithTimeout(ctx, tlsFetchTimeout) + defer cancel() + + profile, err := tlspkg.FetchAPIServerTLSProfile(fetchCtx, k8sClient) + if err != nil { + return classifyTLSProfileError(err) + } + return profile, true, nil +} + +func classifyTLSProfileError(err error) (configv1.TLSProfileSpec, bool, error) { + intermediate := *configv1.TLSProfiles[configv1.TLSProfileIntermediateType] + + switch { + case apimeta.IsNoMatchError(err): + return intermediate, false, nil + case apierrors.IsNotFound(err): + return intermediate, false, nil + case isTransientError(err): + return intermediate, true, nil + default: + return configv1.TLSProfileSpec{}, false, fmt.Errorf("unable to read APIServer TLS profile: %w", err) + } +} + +func fetchTLSAdherencePolicy(ctx context.Context, k8sClient client.Client) (configv1.TLSAdherencePolicy, bool, error) { + fetchCtx, cancel := context.WithTimeout(ctx, tlsFetchTimeout) + defer cancel() + + policy, err := tlspkg.FetchAPIServerTLSAdherencePolicy(fetchCtx, k8sClient) + if err == nil { + return policy, true, nil + } + + switch { + case apimeta.IsNoMatchError(err), + apierrors.IsNotFound(err), + isTransientError(err): + return "", false, nil + default: + return "", false, fmt.Errorf("unable to read APIServer TLS adherence policy: %w", err) + } +} + +func bootstrapTLS(ctx context.Context, k8sClient client.Client) (*tlsBootstrapResult, error) { + logger := log.FromContext(ctx) + result := &tlsBootstrapResult{ + TLSOpts: make([]func(*tls.Config), 0, 2), + } + + profile, profileFetched, err := fetchTLSProfile(ctx, k8sClient) + if err != nil { + return nil, err + } + result.ProfileFetched = profileFetched + result.ProfileSpec = profile + + tlsConfigFn, unsupported := tlspkg.NewTLSConfigFromProfile(profile) + result.UnsupportedCiphers = unsupported + if len(unsupported) > 0 { + logger.Info("TLS profile contains ciphers unsupported by Go", "unsupported", unsupported) + } + result.TLSOpts = append(result.TLSOpts, tlsConfigFn) + + adherence, adherenceFetched, err := fetchTLSAdherencePolicy(ctx, k8sClient) + if err != nil { + return nil, err + } + result.AdherenceFetched = adherenceFetched + result.AdherencePolicy = adherence + + result.TLSOpts = append(result.TLSOpts, func(c *tls.Config) { + c.NextProtos = []string{alpnH2, alpnHTTP11} + }) + + return result, nil +} + +func isTransientError(err error) bool { + return apierrors.IsServiceUnavailable(err) || + apierrors.IsTimeout(err) || + apierrors.IsServerTimeout(err) || + apierrors.IsTooManyRequests(err) || + errors.Is(err, context.DeadlineExceeded) +} diff --git a/infra/feast-operator/cmd/tls_bootstrap_test.go b/infra/feast-operator/cmd/tls_bootstrap_test.go new file mode 100644 index 00000000000..0bb31c14bf6 --- /dev/null +++ b/infra/feast-operator/cmd/tls_bootstrap_test.go @@ -0,0 +1,347 @@ +/* +Copyright 2024 Feast Community. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package main + +import ( + "context" + "crypto/tls" + "errors" + "testing" + + configv1 "github.com/openshift/api/config/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/meta" + "k8s.io/apimachinery/pkg/runtime/schema" +) + +func intermediateProfile() configv1.TLSProfileSpec { + return *configv1.TLSProfiles[configv1.TLSProfileIntermediateType] +} + +func TestClassifyTLSProfileError(t *testing.T) { + tests := []struct { + name string + err error + wantProfileFetched bool + wantError bool + wantIntermediate bool + }{ + { + name: "NoMatchError returns Intermediate defaults, profileFetched=false", + err: &meta.NoResourceMatchError{PartialResource: schema.GroupVersionResource{Group: "config.openshift.io"}}, + wantProfileFetched: false, + wantError: false, + wantIntermediate: true, + }, + { + name: "NotFound returns Intermediate defaults, profileFetched=false", + err: apierrors.NewNotFound(schema.GroupResource{Group: "config.openshift.io", Resource: "apiservers"}, "cluster"), + wantProfileFetched: false, + wantError: false, + wantIntermediate: true, + }, + { + name: "ServiceUnavailable is transient, profileFetched=true", + err: apierrors.NewServiceUnavailable("api server down"), + wantProfileFetched: true, + wantError: false, + wantIntermediate: true, + }, + { + name: "Timeout is transient, profileFetched=true", + err: apierrors.NewTimeoutError("timed out", 5), + wantProfileFetched: true, + wantError: false, + wantIntermediate: true, + }, + { + name: "ServerTimeout is transient, profileFetched=true", + err: apierrors.NewServerTimeout(schema.GroupResource{Group: "config.openshift.io", Resource: "apiservers"}, "GET", 5), + wantProfileFetched: true, + wantError: false, + wantIntermediate: true, + }, + { + name: "TooManyRequests is transient, profileFetched=true", + err: apierrors.NewTooManyRequests("throttled", 5), + wantProfileFetched: true, + wantError: false, + wantIntermediate: true, + }, + { + name: "DeadlineExceeded is transient, profileFetched=true", + err: context.DeadlineExceeded, + wantProfileFetched: true, + wantError: false, + wantIntermediate: true, + }, + { + name: "Forbidden is fatal, returns error", + err: apierrors.NewForbidden(schema.GroupResource{Group: "config.openshift.io", Resource: "apiservers"}, "cluster", errors.New("RBAC")), + wantProfileFetched: false, + wantError: true, + wantIntermediate: false, + }, + { + name: "Unauthorized is fatal, returns error", + err: apierrors.NewUnauthorized("no token"), + wantProfileFetched: false, + wantError: true, + wantIntermediate: false, + }, + { + name: "InternalServerError is fatal, returns error", + err: apierrors.NewInternalError(errors.New("crash")), + wantProfileFetched: false, + wantError: true, + wantIntermediate: false, + }, + { + name: "Generic error is fatal, returns error", + err: errors.New("something unexpected"), + wantProfileFetched: false, + wantError: true, + wantIntermediate: false, + }, + { + name: "Wrapped DeadlineExceeded is transient", + err: errors.Join(errors.New("fetch failed"), context.DeadlineExceeded), + wantProfileFetched: true, + wantError: false, + wantIntermediate: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + profile, fetched, err := classifyTLSProfileError(tt.err) + + if tt.wantError && err == nil { + t.Errorf("expected error, got nil") + } + if !tt.wantError && err != nil { + t.Errorf("unexpected error: %v", err) + } + if fetched != tt.wantProfileFetched { + t.Errorf("profileFetched = %v, want %v", fetched, tt.wantProfileFetched) + } + if tt.wantIntermediate { + intermediate := intermediateProfile() + if profile.MinTLSVersion != intermediate.MinTLSVersion { + t.Errorf("MinTLSVersion = %v, want %v (Intermediate)", profile.MinTLSVersion, intermediate.MinTLSVersion) + } + } + }) + } +} + +func TestIsTransientError(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + {"ServiceUnavailable", apierrors.NewServiceUnavailable("down"), true}, + {"Timeout", apierrors.NewTimeoutError("slow", 5), true}, + {"ServerTimeout", apierrors.NewServerTimeout(schema.GroupResource{}, "GET", 5), true}, + {"TooManyRequests", apierrors.NewTooManyRequests("throttled", 5), true}, + {"DeadlineExceeded", context.DeadlineExceeded, true}, + {"Wrapped DeadlineExceeded", errors.Join(errors.New("wrapper"), context.DeadlineExceeded), true}, + {"NotFound", apierrors.NewNotFound(schema.GroupResource{}, "x"), false}, + {"Forbidden", apierrors.NewForbidden(schema.GroupResource{}, "x", errors.New("RBAC")), false}, + {"Unauthorized", apierrors.NewUnauthorized("no token"), false}, + {"InternalError", apierrors.NewInternalError(errors.New("crash")), false}, + {"Generic error", errors.New("oops"), false}, + {"Nil", nil, false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := isTransientError(tt.err); got != tt.want { + t.Errorf("isTransientError() = %v, want %v", got, tt.want) + } + }) + } +} + +func TestIntermediateProfileHasExpectedDefaults(t *testing.T) { + profile := intermediateProfile() + + if profile.MinTLSVersion != configv1.VersionTLS12 { + t.Errorf("Intermediate MinTLSVersion = %v, want %v", profile.MinTLSVersion, configv1.VersionTLS12) + } + if len(profile.Ciphers) == 0 { + t.Error("Intermediate profile should have non-empty cipher list") + } +} + +func TestTLSConfigFromIntermediateProfile(t *testing.T) { + profile := intermediateProfile() + tlsConfigFn := configv1ToTLSConfig(profile) + + cfg := &tls.Config{} + tlsConfigFn(cfg) + + if cfg.MinVersion != tls.VersionTLS12 { + t.Errorf("MinVersion = %v, want %v (TLS 1.2)", cfg.MinVersion, tls.VersionTLS12) + } + if len(cfg.CipherSuites) == 0 { + t.Error("CipherSuites should not be empty for Intermediate profile") + } +} + +func configv1ToTLSConfig(profile configv1.TLSProfileSpec) func(*tls.Config) { + // Thin wrapper to test the actual conversion without importing tlspkg in tests. + // tlspkg.NewTLSConfigFromProfile is what main.go uses. + var minVersion uint16 + switch profile.MinTLSVersion { + case configv1.VersionTLS10: + minVersion = tls.VersionTLS10 + case configv1.VersionTLS11: + minVersion = tls.VersionTLS11 + case configv1.VersionTLS12: + minVersion = tls.VersionTLS12 + case configv1.VersionTLS13: + minVersion = tls.VersionTLS13 + } + + return func(c *tls.Config) { + c.MinVersion = minVersion + c.CipherSuites = mapCiphers(profile.Ciphers) + } +} + +func mapCiphers(names []string) []uint16 { + cipherMap := map[string]uint16{ + "TLS_AES_128_GCM_SHA256": tls.TLS_AES_128_GCM_SHA256, + "TLS_AES_256_GCM_SHA384": tls.TLS_AES_256_GCM_SHA384, + "TLS_CHACHA20_POLY1305_SHA256": tls.TLS_CHACHA20_POLY1305_SHA256, + "TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256": tls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256, + "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256": tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256, + "TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384": tls.TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384, + "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384": tls.TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384, + "TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256": tls.TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256, + "TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256": tls.TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256, + } + var ids []uint16 + for _, name := range names { + if id, ok := cipherMap[name]; ok { + ids = append(ids, id) + } + } + return ids +} + +func TestClassifyTLSProfileError_AllTransientErrorsSetProfileFetched(t *testing.T) { + transientErrors := []error{ + apierrors.NewServiceUnavailable("down"), + apierrors.NewTimeoutError("slow", 5), + apierrors.NewServerTimeout(schema.GroupResource{Group: "config.openshift.io", Resource: "apiservers"}, "GET", 5), + apierrors.NewTooManyRequests("throttled", 5), + context.DeadlineExceeded, + } + + for _, err := range transientErrors { + _, fetched, classifyErr := classifyTLSProfileError(err) + if classifyErr != nil { + t.Errorf("transient error %T should not return error, got: %v", err, classifyErr) + } + if !fetched { + t.Errorf("transient error %T should set profileFetched=true", err) + } + } +} + +func TestClassifyTLSProfileError_NonTransientErrorsDoNotSetProfileFetched(t *testing.T) { + nonTransientErrors := []error{ + &meta.NoResourceMatchError{PartialResource: schema.GroupVersionResource{Group: "config.openshift.io"}}, + apierrors.NewNotFound(schema.GroupResource{}, "cluster"), + } + + for _, err := range nonTransientErrors { + _, fetched, classifyErr := classifyTLSProfileError(err) + if classifyErr != nil { + t.Errorf("graceful error %T should not return error, got: %v", err, classifyErr) + } + if fetched { + t.Errorf("graceful error %T should set profileFetched=false", err) + } + } +} + +func TestClassifyTLSProfileError_FatalErrorsReturnError(t *testing.T) { + fatalErrors := []error{ + apierrors.NewForbidden(schema.GroupResource{}, "cluster", errors.New("RBAC")), + apierrors.NewUnauthorized("no token"), + apierrors.NewInternalError(errors.New("crash")), + errors.New("unexpected"), + } + + for _, err := range fatalErrors { + _, _, classifyErr := classifyTLSProfileError(err) + if classifyErr == nil { + t.Errorf("fatal error %T should return error", err) + } + } +} + +func TestClassifyTLSProfileError_IntermediateProfileAlwaysApplied(t *testing.T) { + allNonFatalErrors := []error{ + &meta.NoResourceMatchError{PartialResource: schema.GroupVersionResource{Group: "config.openshift.io"}}, + apierrors.NewNotFound(schema.GroupResource{}, "cluster"), + apierrors.NewServiceUnavailable("down"), + apierrors.NewTimeoutError("slow", 5), + apierrors.NewServerTimeout(schema.GroupResource{}, "GET", 5), + apierrors.NewTooManyRequests("throttled", 5), + context.DeadlineExceeded, + } + + intermediate := intermediateProfile() + for _, err := range allNonFatalErrors { + profile, _, classifyErr := classifyTLSProfileError(err) + if classifyErr != nil { + t.Fatalf("unexpected error for %T: %v", err, classifyErr) + } + if profile.MinTLSVersion != intermediate.MinTLSVersion { + t.Errorf("for error %T: MinTLSVersion = %v, want Intermediate (%v)", err, profile.MinTLSVersion, intermediate.MinTLSVersion) + } + if len(profile.Ciphers) != len(intermediate.Ciphers) { + t.Errorf("for error %T: got %d ciphers, want %d (Intermediate)", err, len(profile.Ciphers), len(intermediate.Ciphers)) + } + } +} + +func TestTLSBootstrapResult_NextProtosAlwaysSet(t *testing.T) { + // Verify that the TLSOpts from bootstrapTLS always include ALPN with h2 and http/1.1. + // We can't call bootstrapTLS without a real client, but we can verify the function + // in tls_bootstrap.go sets NextProtos. + result := &tlsBootstrapResult{ + TLSOpts: make([]func(*tls.Config), 0, 2), + } + result.TLSOpts = append(result.TLSOpts, func(c *tls.Config) { + c.NextProtos = []string{"h2", alpnHTTP11} + }) + + cfg := &tls.Config{} + for _, opt := range result.TLSOpts { + opt(cfg) + } + + if len(cfg.NextProtos) != 2 || cfg.NextProtos[0] != "h2" || cfg.NextProtos[1] != alpnHTTP11 { + t.Errorf("NextProtos = %v, want [h2, %s]", cfg.NextProtos, alpnHTTP11) + } +}