feat: validate external provider secrets via SDK credential schema

- Add get_credentials_schema to the provider contract
- Validate non-built-in secrets against the declared schema; built-ins unchanged
- Accept secrets as-is when no schema is declared (validated at connection)
This commit is contained in:
StylusFrost
2026-06-01 00:32:47 +02:00
parent 51581c35ec
commit b8d3312577
3 changed files with 106 additions and 1 deletions
+49 -1
View File
@@ -1,8 +1,56 @@
from unittest.mock import MagicMock, patch
import pytest
from pydantic import BaseModel
from rest_framework.exceptions import ValidationError
from api.v1.serializer_utils.integrations import S3ConfigSerializer
from api.v1.serializers import ImageProviderSecret, ProviderEnumSerializerField
from api.v1.serializers import (
BaseWriteProviderSecretSerializer,
ImageProviderSecret,
ProviderEnumSerializerField,
)
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)."""
class _Credentials(BaseModel):
api_url: str
api_key: str
def test_secret_validated_against_declared_schema(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
):
BaseWriteProviderSecretSerializer._validate_external_provider_secret(
"external-template", {"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 pytest.raises(ValidationError):
BaseWriteProviderSecretSerializer._validate_external_provider_secret(
"external-template", {"api_url": "u"}
)
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
):
BaseWriteProviderSecretSerializer._validate_external_provider_secret(
"external-template", {"anything": "goes"}
)
class TestProviderEnumSerializerField:
+39
View File
@@ -10,6 +10,7 @@ from django.core.exceptions import ValidationError as DjangoValidationError
from django.db import IntegrityError
from drf_spectacular.utils import extend_schema_field
from jwt.exceptions import InvalidKeyError
from pydantic import ValidationError as PydanticValidationError
from rest_framework.reverse import reverse
from rest_framework.validators import UniqueTogetherValidator
from rest_framework_json_api import serializers
@@ -1544,10 +1545,48 @@ 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).
Providers that declare no schema have their secret accepted as-is; the
credentials are then validated by the provider's ``test_connection``.
"""
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()
]
}
)
@staticmethod
def validate_secret_based_on_provider(
provider_type: str, secret_type: ProviderSecret.TypeChoices, secret: dict
):
# External providers validate against the schemas they declare via the
# 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
)
return
if secret_type == ProviderSecret.TypeChoices.STATIC:
if provider_type == Provider.ProviderChoices.AWS.value:
serializer = AwsProviderSecret(data=secret)