diff --git a/.secrets.baseline b/.secrets.baseline index 7ddbfb4a3f6..4d389cae97c 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -1165,14 +1165,14 @@ "filename": "infra/feast-operator/internal/controller/services/repo_config.go", "hashed_secret": "44e17306b837162269a410204daaa5ecee4ec22c", "is_verified": false, - "line_number": 198 + "line_number": 241 }, { "type": "Secret Keyword", "filename": "infra/feast-operator/internal/controller/services/repo_config.go", "hashed_secret": "e2fb052132fd6a07a56af2013e0b62a1f510572c", "is_verified": false, - "line_number": 306 + "line_number": 345 } ], "infra/feast-operator/internal/controller/services/services.go": [ @@ -1564,5 +1564,5 @@ } ] }, - "generated_at": "2026-10-06T10:38:10Z" + "generated_at": "2026-10-07T13:22:05Z" } diff --git a/infra/feast-operator/internal/controller/services/repo_config.go b/infra/feast-operator/internal/controller/services/repo_config.go index ff1d89b066b..3c390c674d3 100644 --- a/infra/feast-operator/internal/controller/services/repo_config.go +++ b/infra/feast-operator/internal/controller/services/repo_config.go @@ -52,7 +52,8 @@ func (feast *FeastServices) getServiceRepoConfig() (RepoConfig, error) { } func (feast *FeastServices) getLineageFeatureStoreYamlBase64() (string, error) { - repoConfig, err := feast.getLineageRepoConfig() + odhCaBundleExists := feast.GetCustomCertificatesBundle().IsDefined + repoConfig, err := feast.getLineageRepoConfig(odhCaBundleExists) if err != nil { return "", err } @@ -63,7 +64,7 @@ func (feast *FeastServices) getLineageFeatureStoreYamlBase64() (string, error) { return base64.StdEncoding.EncodeToString(yamlBytes), nil } -func (feast *FeastServices) getLineageRepoConfig() (RepoConfig, error) { +func (feast *FeastServices) getLineageRepoConfig(odhCaBundleExists bool) (RepoConfig, error) { cr := feast.Handler.FeatureStore applied := cr.Status.Applied @@ -108,6 +109,28 @@ func (feast *FeastServices) getLineageRepoConfig() (RepoConfig, error) { } } + // Apply auth configuration so the lineage server enforces the same + // authentication as the other Feast servers. Without this the + // standalone lineage endpoints would be unauthenticated even when + // Kubernetes or OIDC auth is configured on the FeatureStore CR. + if applied.AuthzConfig != nil && applied.AuthzConfig.OidcAuthz != nil { + authzConfig, err := resolveOidcServerAuthzConfig(applied.AuthzConfig.OidcAuthz, feast.extractConfigFromSecret, odhCaBundleExists) + if err != nil { + return repoConfig, err + } + repoConfig.AuthzConfig = authzConfig + } else if applied.AuthzConfig != nil { + if applied.AuthzConfig.NoAuth != nil && *applied.AuthzConfig.NoAuth { + repoConfig.AuthzConfig = AuthzConfig{Type: NoAuthAuthType} + } else if applied.AuthzConfig.KubernetesAuthz != nil { + repoConfig.AuthzConfig = AuthzConfig{Type: KubernetesAuthType} + } else { + return repoConfig, fmt.Errorf("authzConfig must specify OIDC, Kubernetes, or no-auth") + } + } else { + repoConfig.AuthzConfig = defaultAuthzConfig + } + return repoConfig, nil } @@ -190,54 +213,11 @@ func getBaseServiceRepoConfig( } appliedSpec := featureStore.Status.Applied if appliedSpec.AuthzConfig != nil && appliedSpec.AuthzConfig.OidcAuthz != nil { - repoConfig.AuthzConfig = AuthzConfig{Type: OidcAuthType} - oidcAuthz := appliedSpec.AuthzConfig.OidcAuthz - oidcParameters := map[string]interface{}{} - - var secretProperties map[string]interface{} - if oidcAuthz.SecretRef != nil { - var err error - secretProperties, err = secretExtractionFunc("", oidcAuthz.SecretRef.Name, oidcAuthz.SecretKeyName) - if err != nil { - return repoConfig, err - } - for _, prop := range OidcOptionalSecretProperties { - if val, exists := secretProperties[string(prop)]; exists { - // Secret values are YAML-parsed on extraction, so an - // all-digits audience or issuer arrives as an int and - // would render unquoted, which the SDK's OidcAuthConfig - // rejects (Optional[str]). Coerce the claim keys back to - // strings; the five original keys keep their historical - // typing. - if prop == OidcAudience || prop == OidcIssuer { - if _, isString := val.(string); !isString { - val = fmt.Sprintf("%v", val) - } - } - oidcParameters[string(prop)] = val - } - } - } - - discoveryUrl, err := resolveAuthDiscoveryUrl(oidcAuthz, secretProperties) + authzConfig, err := resolveOidcServerAuthzConfig(appliedSpec.AuthzConfig.OidcAuthz, secretExtractionFunc, odhCaBundleExists) if err != nil { return repoConfig, err } - oidcParameters[string(OidcAuthDiscoveryUrl)] = discoveryUrl - - if oidcAuthz.VerifySSL != nil { - oidcParameters[string(OidcVerifySsl)] = *oidcAuthz.VerifySSL - } - if oidcAuthz.JwksCacheLifespanSeconds != nil { - oidcParameters[string(OidcJwksCacheLifespanSeconds)] = *oidcAuthz.JwksCacheLifespanSeconds - } - if oidcAuthz.JwksRequestTimeoutSeconds != nil { - oidcParameters[string(OidcJwksRequestTimeoutSeconds)] = *oidcAuthz.JwksRequestTimeoutSeconds - } - if caCertPath := resolveOidcCACertPath(oidcAuthz, odhCaBundleExists); caCertPath != "" { - oidcParameters[string(OidcCaCertPath)] = caCertPath - } - repoConfig.AuthzConfig.OidcParameters = oidcParameters + repoConfig.AuthzConfig = authzConfig } else { repoConfig.AuthzConfig = clientRepoConfig.AuthzConfig } @@ -245,6 +225,65 @@ func getBaseServiceRepoConfig( return repoConfig, nil } +// resolveOidcServerAuthzConfig builds the server-side OIDC AuthzConfig from the +// CR's OidcAuthz spec. Used by both getBaseServiceRepoConfig and +// getLineageRepoConfig to avoid duplicating the secret extraction, discovery +// URL resolution, and TLS/CA logic. +func resolveOidcServerAuthzConfig( + oidcAuthz *feastdevv1.OidcAuthz, + secretExtractionFunc func(storeType string, secretRef string, secretKeyName string) (map[string]interface{}, error), + odhCaBundleExists bool, +) (AuthzConfig, error) { + authzConfig := AuthzConfig{Type: OidcAuthType} + oidcParameters := map[string]interface{}{} + + var secretProperties map[string]interface{} + if oidcAuthz.SecretRef != nil { + var err error + secretProperties, err = secretExtractionFunc("", oidcAuthz.SecretRef.Name, oidcAuthz.SecretKeyName) + if err != nil { + return authzConfig, err + } + for _, prop := range OidcOptionalSecretProperties { + if val, exists := secretProperties[string(prop)]; exists { + // Secret values are YAML-parsed on extraction, so an + // all-digits audience or issuer arrives as an int and + // would render unquoted, which the SDK's OidcAuthConfig + // rejects (Optional[str]). Coerce the claim keys back to + // strings; the five original keys keep their historical + // typing. + if prop == OidcAudience || prop == OidcIssuer { + if _, isString := val.(string); !isString { + val = fmt.Sprintf("%v", val) + } + } + oidcParameters[string(prop)] = val + } + } + } + + discoveryUrl, err := resolveAuthDiscoveryUrl(oidcAuthz, secretProperties) + if err != nil { + return authzConfig, err + } + oidcParameters[string(OidcAuthDiscoveryUrl)] = discoveryUrl + + if oidcAuthz.VerifySSL != nil { + oidcParameters[string(OidcVerifySsl)] = *oidcAuthz.VerifySSL + } + if oidcAuthz.JwksCacheLifespanSeconds != nil { + oidcParameters[string(OidcJwksCacheLifespanSeconds)] = *oidcAuthz.JwksCacheLifespanSeconds + } + if oidcAuthz.JwksRequestTimeoutSeconds != nil { + oidcParameters[string(OidcJwksRequestTimeoutSeconds)] = *oidcAuthz.JwksRequestTimeoutSeconds + } + if caCertPath := resolveOidcCACertPath(oidcAuthz, odhCaBundleExists); caCertPath != "" { + oidcParameters[string(OidcCaCertPath)] = caCertPath + } + authzConfig.OidcParameters = oidcParameters + return authzConfig, nil +} + // resolveAuthDiscoveryUrl determines the OIDC discovery URL from the first available source. // Priority: CR issuerUrl > Secret auth_discovery_url > OIDC_ISSUER_URL env var. func resolveAuthDiscoveryUrl(oidcAuthz *feastdevv1.OidcAuthz, secretProperties map[string]interface{}) (string, error) { diff --git a/infra/feast-operator/internal/controller/services/repo_config_test.go b/infra/feast-operator/internal/controller/services/repo_config_test.go index 3d26929a46b..b159b0c01e1 100644 --- a/infra/feast-operator/internal/controller/services/repo_config_test.go +++ b/infra/feast-operator/internal/controller/services/repo_config_test.go @@ -845,7 +845,7 @@ var _ = Describe("Repo Config", func() { feast := FeastServices{ Handler: handler.FeastHandler{FeatureStore: featureStore}, } - repoConfig, err := feast.getLineageRepoConfig() + repoConfig, err := feast.getLineageRepoConfig(false) Expect(err).NotTo(HaveOccurred()) Expect(repoConfig.Registry.RegistryType).To(Equal(RegistryRemoteConfigType)) Expect(repoConfig.Registry.Path).To(Equal(remoteHost)) @@ -882,7 +882,7 @@ var _ = Describe("Repo Config", func() { feast := FeastServices{ Handler: handler.FeastHandler{FeatureStore: featureStore}, } - repoConfig, err := feast.getLineageRepoConfig() + repoConfig, err := feast.getLineageRepoConfig(false) Expect(err).NotTo(HaveOccurred()) Expect(repoConfig.Registry.RegistryType).To(Equal(RegistryRemoteConfigType)) Expect(repoConfig.Registry.Path).To(Equal(resolvedHost)) @@ -922,6 +922,125 @@ var _ = Describe("Repo Config", func() { Expect(err).NotTo(HaveOccurred()) }) + It("should include kubernetes auth config in lineage repo config", func() { + featureStore := minimalFeatureStore() + remoteHost := "feast-banking-registry.feast.svc.cluster.local:443" + featureStore.Spec.Services = &feastdevv1.FeatureStoreServices{ + Registry: &feastdevv1.Registry{ + Remote: &feastdevv1.RemoteRegistryConfig{ + Hostname: &remoteHost, + }, + }, + } + featureStore.Spec.AuthzConfig = &feastdevv1.AuthzConfig{ + KubernetesAuthz: &feastdevv1.KubernetesAuthz{}, + } + featureStore.Spec.OpenLineage = &feastdevv1.OpenLineageConfig{ + Enabled: true, + Consumer: &feastdevv1.OpenLineageConsumerConfig{ + Enabled: true, + LineageServer: &feastdevv1.LineageServerConfig{ + Replicas: ptr.To[int32](1), + }, + }, + } + ApplyDefaultsToStatus(featureStore) + featureStore.Status.ServiceHostnames.Registry = remoteHost + + feast := FeastServices{ + Handler: handler.FeastHandler{FeatureStore: featureStore}, + } + repoConfig, err := feast.getLineageRepoConfig(false) + Expect(err).NotTo(HaveOccurred()) + Expect(repoConfig.AuthzConfig.Type).To(Equal(KubernetesAuthType)) + Expect(repoConfig.Registry.RegistryType).To(Equal(RegistryRemoteConfigType)) + Expect(repoConfig.Registry.Path).To(Equal(remoteHost)) + }) + + It("should default to kubernetes auth in lineage repo config when authzConfig is nil", func() { + featureStore := minimalFeatureStore() + featureStore.Spec.OpenLineage = &feastdevv1.OpenLineageConfig{ + Enabled: true, + Consumer: &feastdevv1.OpenLineageConsumerConfig{ + Enabled: true, + LineageServer: &feastdevv1.LineageServerConfig{ + Replicas: ptr.To[int32](1), + }, + }, + } + ApplyDefaultsToStatus(featureStore) + + feast := FeastServices{ + Handler: handler.FeastHandler{FeatureStore: featureStore}, + } + repoConfig, err := feast.getLineageRepoConfig(false) + Expect(err).NotTo(HaveOccurred()) + Expect(repoConfig.AuthzConfig.Type).To(Equal(KubernetesAuthType)) + }) + + It("should include no_auth config in lineage repo config when NoAuth is set", func() { + featureStore := minimalFeatureStore() + featureStore.Spec.AuthzConfig = &feastdevv1.AuthzConfig{ + NoAuth: boolPtr(true), + } + featureStore.Spec.OpenLineage = &feastdevv1.OpenLineageConfig{ + Enabled: true, + Consumer: &feastdevv1.OpenLineageConsumerConfig{ + Enabled: true, + LineageServer: &feastdevv1.LineageServerConfig{ + Replicas: ptr.To[int32](1), + }, + }, + } + ApplyDefaultsToStatus(featureStore) + + feast := FeastServices{ + Handler: handler.FeastHandler{FeatureStore: featureStore}, + } + repoConfig, err := feast.getLineageRepoConfig(false) + Expect(err).NotTo(HaveOccurred()) + Expect(repoConfig.AuthzConfig.Type).To(Equal(NoAuthAuthType)) + }) + + It("should include OIDC auth config in lineage repo config with issuerUrl", func() { + featureStore := minimalFeatureStore() + remoteHost := "feast-banking-registry.feast.svc.cluster.local:443" + featureStore.Spec.Services = &feastdevv1.FeatureStoreServices{ + Registry: &feastdevv1.Registry{ + Remote: &feastdevv1.RemoteRegistryConfig{ + Hostname: &remoteHost, + }, + }, + } + featureStore.Spec.AuthzConfig = &feastdevv1.AuthzConfig{ + OidcAuthz: &feastdevv1.OidcAuthz{ + IssuerUrl: "https://keycloak.example.com/realms/test", + }, + } + featureStore.Spec.OpenLineage = &feastdevv1.OpenLineageConfig{ + Enabled: true, + Consumer: &feastdevv1.OpenLineageConsumerConfig{ + Enabled: true, + LineageServer: &feastdevv1.LineageServerConfig{ + Replicas: ptr.To[int32](1), + }, + }, + } + ApplyDefaultsToStatus(featureStore) + featureStore.Status.ServiceHostnames.Registry = remoteHost + + feast := FeastServices{ + Handler: handler.FeastHandler{FeatureStore: featureStore}, + } + repoConfig, err := feast.getLineageRepoConfig(false) + Expect(err).NotTo(HaveOccurred()) + Expect(repoConfig.AuthzConfig.Type).To(Equal(OidcAuthType)) + Expect(repoConfig.AuthzConfig.OidcParameters).To(HaveKeyWithValue( + string(OidcAuthDiscoveryUrl), + "https://keycloak.example.com/realms/test/.well-known/openid-configuration", + )) + }) + It("should reject lineageServer without any registry when authz is set", func() { featureStore := minimalFeatureStore() featureStore.Spec.AuthzConfig = &feastdevv1.AuthzConfig{ diff --git a/sdk/python/feast/lineage_server.py b/sdk/python/feast/lineage_server.py index f29130d5233..4c142657322 100644 --- a/sdk/python/feast/lineage_server.py +++ b/sdk/python/feast/lineage_server.py @@ -26,7 +26,7 @@ import asyncio import logging -from fastapi import FastAPI +from fastapi import Depends, FastAPI logger = logging.getLogger(__name__) @@ -103,6 +103,37 @@ def _get_allowed(): return _get_allowed +def _init_auth(store): + """Initialize auth/security managers if auth is configured.""" + from feast.permissions.server.utils import ( + ServerType, + init_auth_manager, + init_security_manager, + str_to_auth_manager_type, + ) + + auth_cfg = store.config.auth_config + dependencies: list = [] + if auth_cfg and auth_cfg.type != "no_auth": + from feast.permissions.server.rest import inject_user_details + + auth_type = str_to_auth_manager_type(auth_cfg.type) + init_security_manager(auth_type=auth_type, fs=store) + init_auth_manager( + auth_type=auth_type, + server_type=ServerType.REST, + auth_config=auth_cfg, + ) + dependencies.append(Depends(inject_user_details)) + logger.info("Lineage server auth initialized (type=%s)", auth_cfg.type) + else: + logger.warning( + "Lineage server running with auth_type=no_auth. " + "All endpoints are unauthenticated." + ) + return dependencies + + def create_lineage_app(store) -> FastAPI: """Create the standalone lineage FastAPI application.""" from feast.openlineage.config import ( @@ -191,12 +222,15 @@ def create_lineage_app(store) -> FastAPI: get_allowed_namespaces=get_allowed_namespaces, ) + auth_dependencies = _init_auth(store) + app = FastAPI( title="Feast OpenLineage Server", description="Standalone OpenLineage consumer for Feast lineage data", version="1.0.0", docs_url="/", redoc_url="/docs", + dependencies=auth_dependencies, ) app.include_router(consumer_router, prefix="/api/v1")