feat: bind external provider secret validation to its secret type

- get_credentials_schema returns schemas keyed by secret type
- validate an external provider's secret against the schema for its
  secret_type instead of accepting any declared schema
- reject a secret_type the provider does not declare
This commit is contained in:
StylusFrost
2026-06-01 22:09:21 +02:00
parent 116fb7083d
commit 5daa39c4cb
3 changed files with 84 additions and 64 deletions
+39 -27
View File
@@ -13,43 +13,59 @@ from api.v1.serializers import (
class TestExternalProviderSecretValidation:
"""A non-built-in provider's secret is validated against the credential
schema it declares through the SDK contract, or accepted as-is when it
declares none (then validated by the provider's test_connection)."""
"""A non-built-in provider's secret is validated against the schema it
declares for the chosen secret type through the SDK contract, or accepted as
an object when it declares none (then validated by test_connection)."""
class _Credentials(BaseModel):
class _StaticCredentials(BaseModel):
api_url: str
api_key: str
def test_secret_validated_against_declared_schema(self):
class _RoleCredentials(BaseModel):
role_arn: str
def _patch(self, schemas):
provider_class = MagicMock()
provider_class.get_credentials_schema.return_value = [self._Credentials]
with patch(
provider_class.get_credentials_schema.return_value = schemas
return patch(
"api.v1.serializers.SDKProvider.get_class", return_value=provider_class
):
)
def test_secret_validated_against_its_type_schema(self):
with self._patch({"static": self._StaticCredentials}):
BaseWriteProviderSecretSerializer._validate_external_provider_secret(
"external-template", {"api_url": "u", "api_key": "k"}
"external-template", "static", {"api_url": "u", "api_key": "k"}
)
def test_secret_rejected_when_schema_violated(self):
provider_class = MagicMock()
provider_class.get_credentials_schema.return_value = [self._Credentials]
with patch(
"api.v1.serializers.SDKProvider.get_class", return_value=provider_class
):
with self._patch({"static": self._StaticCredentials}):
with pytest.raises(ValidationError):
BaseWriteProviderSecretSerializer._validate_external_provider_secret(
"external-template", {"api_url": "u"}
"external-template", "static", {"api_url": "u"}
)
def test_secret_must_match_its_type_not_another(self):
"""A secret is validated against the schema for its declared secret_type,
not "any declared schema": a role-shaped secret under secret_type=static
is rejected. See PR #11402 review (josema-xyz / Alan-TheGentleman)."""
schemas = {"static": self._StaticCredentials, "role": self._RoleCredentials}
with self._patch(schemas):
with pytest.raises(ValidationError):
BaseWriteProviderSecretSerializer._validate_external_provider_secret(
"external-template", "static", {"role_arn": "arn:aws:iam::x"}
)
def test_rejects_secret_type_not_declared_by_provider(self):
with self._patch({"static": self._StaticCredentials}):
with pytest.raises(ValidationError):
BaseWriteProviderSecretSerializer._validate_external_provider_secret(
"external-template", "role", {"role_arn": "arn"}
)
def test_secret_accepted_when_no_schema_declared(self):
provider_class = MagicMock()
provider_class.get_credentials_schema.return_value = []
with patch(
"api.v1.serializers.SDKProvider.get_class", return_value=provider_class
):
with self._patch({}):
BaseWriteProviderSecretSerializer._validate_external_provider_secret(
"external-template", {"anything": "goes"}
"external-template", "static", {"anything": "goes"}
)
@pytest.mark.parametrize("bad_secret", [["a", "b"], "a-string", None, 42])
@@ -57,14 +73,10 @@ class TestExternalProviderSecretValidation:
"""Even with no declared schema, a non-object secret must be rejected so
a list/string/null cannot be persisted and blow up later at
``{**secret}``. See PR #11402 review (Alan-TheGentleman)."""
provider_class = MagicMock()
provider_class.get_credentials_schema.return_value = []
with patch(
"api.v1.serializers.SDKProvider.get_class", return_value=provider_class
):
with self._patch({}):
with pytest.raises(ValidationError):
BaseWriteProviderSecretSerializer._validate_external_provider_secret(
"external-template", bad_secret
"external-template", "static", bad_secret
)
+33 -24
View File
@@ -1544,36 +1544,45 @@ class FindingMetadataSerializer(BaseSerializerV1):
# Provider secrets
class BaseWriteProviderSecretSerializer(BaseWriteSerializer):
@staticmethod
def _validate_external_provider_secret(provider_type: str, secret: dict):
"""Validate a non-built-in provider's secret against the credential
schemas it declares through the SDK contract (one model per secret type;
the secret must match one).
def _validate_external_provider_secret(
provider_type: str, secret_type: str, secret: dict
):
"""Validate a non-built-in provider's secret against the schema it
declares for the given secret type through the SDK contract.
Providers that declare no schema have their secret accepted as-is; the
credentials are then validated by the provider's ``test_connection``.
The provider maps each secret type to one model, so the chosen
secret_type stays bound to the shape it claims. Providers that declare
no schema have their secret accepted as an object and validated by the
provider's ``test_connection``.
"""
if not isinstance(secret, dict):
raise serializers.ValidationError({"secret": ["Must be a JSON object."]})
schemas = SDKProvider.get_class(provider_type).get_credentials_schema()
if not schemas:
return
collected_errors = []
for schema in schemas:
try:
schema.model_validate(secret)
return
except PydanticValidationError as error:
collected_errors.append(error)
raise serializers.ValidationError(
{
"secret": [
f"{'/'.join(str(loc) for loc in item['loc']) or 'secret'}: "
f"{item['msg']}"
for error in collected_errors
for item in error.errors()
]
}
)
schema = schemas.get(secret_type)
if schema is None:
raise serializers.ValidationError(
{
"secret_type": [
f"'{secret_type}' is not supported by provider "
f"'{provider_type}'. Supported types: "
f"{', '.join(sorted(schemas))}."
]
}
)
try:
schema.model_validate(secret)
except PydanticValidationError as error:
raise serializers.ValidationError(
{
"secret": [
f"{'/'.join(str(loc) for loc in item['loc']) or 'secret'}: "
f"{item['msg']}"
for item in error.errors()
]
}
)
@staticmethod
def validate_secret_based_on_provider(
@@ -1583,7 +1592,7 @@ class BaseWriteProviderSecretSerializer(BaseWriteSerializer):
# SDK contract; built-in providers keep their explicit serializers below.
if not SDKProvider.is_builtin(provider_type):
BaseWriteProviderSecretSerializer._validate_external_provider_secret(
provider_type, secret
provider_type, secret_type, secret
)
return