feat: Add ConnectionRef to DataSource for pluggable external credential resolution - #6642
Conversation
|
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #6642 +/- ##
========================================
Coverage 46.95% 46.96%
========================================
Files 417 418 +1
Lines 51393 51672 +279
Branches 7438 7484 +46
========================================
+ Hits 24132 24267 +135
- Misses 25530 25671 +141
- Partials 1731 1734 +3
... and 1 file with indirect coverage changes Continue to review full report in Codecov by Harness.
🚀 New features to boost your workflow:
|
4f3809d to
948b262
Compare
ed8fece to
35e9afc
Compare
|
@jyejare @patelchaitany @aniketpalu @Vperiodt Please review |
jyejare
left a comment
There was a problem hiding this comment.
This PR adds ConnectionRef to DataSource for pluggable external credential resolution, enabling per-datasource credentials via external providers like Kubernetes Secrets, HashiCorp Vault, and cloud secret managers. The implementation is well-architected with comprehensive documentation, but has some critical security concerns and missing integration points that need addressing.
| message ConnectionRef { | ||
| // Credential provider type: "kubernetes", "vault", "aws-secrets-manager", | ||
| // "gcp-secret-manager", "azure-key-vault", "env". | ||
| string provider = 1; | ||
|
|
||
| // Provider-specific name: K8s Secret name, Vault path, env var prefix, etc. | ||
| string name = 2; | ||
|
|
||
| // Optional scope qualifier: K8s namespace, Vault mount, AWS region, etc. |
There was a problem hiding this comment.
[Suggestion] ConnectionRef proto should include validation constraints
The ConnectionRef proto definition lacks validation constraints that could prevent common misconfigurations. Adding field validation (e.g., required provider, valid provider types, namespace format) would catch configuration errors early rather than at runtime. This could be part of next improvement PR though.
Optional Comment!
Suggested:
| message ConnectionRef { | |
| // Credential provider type: "kubernetes", "vault", "aws-secrets-manager", | |
| // "gcp-secret-manager", "azure-key-vault", "env". | |
| string provider = 1; | |
| // Provider-specific name: K8s Secret name, Vault path, env var prefix, etc. | |
| string name = 2; | |
| // Optional scope qualifier: K8s namespace, Vault mount, AWS region, etc. | |
| message ConnectionRef { | |
| // Credential provider type: "kubernetes", "vault", "aws-secrets-manager", | |
| // "gcp-secret-manager", "azure-key-vault", "env". | |
| // Required field. | |
| string provider = 1 [(validate.rules).string.min_len = 1]; | |
| // Provider-specific name: K8s Secret name, Vault path, env var prefix, etc. | |
| // Required field. | |
| string name = 2 [(validate.rules).string.min_len = 1]; |
|
|
||
| @staticmethod | ||
| def create_filesystem_and_path( | ||
| path: str, s3_endpoint_override: str | ||
| path: str, | ||
| s3_endpoint_override: str, | ||
| resolved_credentials: Optional[Dict[str, str]] = None, | ||
| ) -> Tuple[Optional[FileSystem], str]: | ||
| if path.startswith("s3://"): | ||
| s3fs = S3FileSystem( | ||
| endpoint_override=s3_endpoint_override if s3_endpoint_override else None | ||
| ) | ||
| kwargs: Dict[str, Optional[str]] = {} | ||
| if s3_endpoint_override: | ||
| kwargs["endpoint_override"] = s3_endpoint_override | ||
|
|
||
| if resolved_credentials: | ||
| access_key = resolved_credentials.get("AWS_ACCESS_KEY_ID", "") | ||
| secret_key = resolved_credentials.get("AWS_SECRET_ACCESS_KEY", "") | ||
| session_token = resolved_credentials.get("AWS_SESSION_TOKEN") | ||
| region = resolved_credentials.get("AWS_DEFAULT_REGION") | ||
| if access_key and secret_key: | ||
| kwargs["access_key"] = access_key | ||
| kwargs["secret_key"] = secret_key | ||
| if session_token: | ||
| kwargs["session_token"] = session_token | ||
| if region: | ||
| kwargs["region"] = region | ||
|
|
||
| s3fs = S3FileSystem(**kwargs) |
There was a problem hiding this comment.
[Suggestion] S3 credential handling should support all AWS credential types
The S3FileSystem credential handling only supports access_key/secret_key authentication but AWS supports many other credential types (IAM roles, STS, OIDC, etc.). The implementation should be more flexible to support the full range of AWS authentication methods.
Again this could be the part of future improvements. Or we should at least document that only access_key / secret_key type auth.
Suggested:
| @staticmethod | |
| def create_filesystem_and_path( | |
| path: str, s3_endpoint_override: str | |
| path: str, | |
| s3_endpoint_override: str, | |
| resolved_credentials: Optional[Dict[str, str]] = None, | |
| ) -> Tuple[Optional[FileSystem], str]: | |
| if path.startswith("s3://"): | |
| s3fs = S3FileSystem( | |
| endpoint_override=s3_endpoint_override if s3_endpoint_override else None | |
| ) | |
| kwargs: Dict[str, Optional[str]] = {} | |
| if s3_endpoint_override: | |
| kwargs["endpoint_override"] = s3_endpoint_override | |
| if resolved_credentials: | |
| access_key = resolved_credentials.get("AWS_ACCESS_KEY_ID", "") | |
| secret_key = resolved_credentials.get("AWS_SECRET_ACCESS_KEY", "") | |
| session_token = resolved_credentials.get("AWS_SESSION_TOKEN") | |
| region = resolved_credentials.get("AWS_DEFAULT_REGION") | |
| if access_key and secret_key: | |
| kwargs["access_key"] = access_key | |
| kwargs["secret_key"] = secret_key | |
| if session_token: | |
| kwargs["session_token"] = session_token | |
| if region: | |
| kwargs["region"] = region | |
| s3fs = S3FileSystem(**kwargs) | |
| if resolved_credentials: | |
| # Support multiple AWS credential formats | |
| access_key = resolved_credentials.get("AWS_ACCESS_KEY_ID") or resolved_credentials.get("access_key_id") | |
| secret_key = resolved_credentials.get("AWS_SECRET_ACCESS_KEY") or resolved_credentials.get("secret_access_key") | |
| session_token = resolved_credentials.get("AWS_SESSION_TOKEN") or resolved_credentials.get("session_token") | |
| region = resolved_credentials.get("AWS_DEFAULT_REGION") or resolved_credentials.get("region") | |
| if access_key and secret_key: | |
| kwargs["access_key"] = access_key | |
| kwargs["secret_key"] = secret_key | |
| # Support role-based authentication | |
| elif "role_arn" in resolved_credentials: | |
| kwargs["role_arn"] = resolved_credentials["role_arn"] | |
| if "external_id" in resolved_credentials: | |
| kwargs["external_id"] = resolved_credentials["external_id"] |
jyejare
left a comment
There was a problem hiding this comment.
This PR introduces a well-architected ConnectionRef feature for pluggable external credential resolution in Feast DataSources. The implementation follows good patterns with comprehensive documentation, built-in providers for Kubernetes/Vault/environment variables, and proper integration across all data source types. However, there are several critical security concerns around credential handling and some missing test coverage.
…al resolution Signed-off-by: ntkathole <nikhilkathole2683@gmail.com>
What this PR does / why we need it:
Feast DataSources today have no mechanism to reference external credentials - authentication relies on ambient environment variables or a single global
offline_storeconfig infeature_store.yaml.This PR introduces
ConnectionRef- which backend to use, how to authenticate, and where to connect - an optional reference onDataSourcethat points to an external credential store (Kubernetes Secrets, HashiCorp Vault, cloud secret managers, or environment variables). Credentials are resolvedat runtime by a pluggable
CredentialProviderinterface.This enables:
connection_refspecifyingconnection_type, credentials, and connection params - no server restart needed.