diff --git a/api/src/backend/api/tests/test_serializers.py b/api/src/backend/api/tests/test_serializers.py index cdba488932..3a930c37ce 100644 --- a/api/src/backend/api/tests/test_serializers.py +++ b/api/src/backend/api/tests/test_serializers.py @@ -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: diff --git a/api/src/backend/api/v1/serializers.py b/api/src/backend/api/v1/serializers.py index 03c4a99e7e..693450ce2c 100644 --- a/api/src/backend/api/v1/serializers.py +++ b/api/src/backend/api/v1/serializers.py @@ -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) diff --git a/prowler/providers/common/provider.py b/prowler/providers/common/provider.py index 88b0473938..a9bcea1be3 100644 --- a/prowler/providers/common/provider.py +++ b/prowler/providers/common/provider.py @@ -243,6 +243,24 @@ class Provider(ABC): """ return {**secret} + @classmethod + def get_credentials_schema(cls) -> list: + """Return the credential schemas this provider accepts — one pydantic + model per secret type. + + Each model documents, in a single declaration the API can consume for + both validation and OpenAPI generation: + * the secret type itself, via the model docstring (schema description); + * each field, via ``Field(description=...)``; + * whether each field is required (no default) or optional + (``Optional[...] = None`` / ``Field(default=...)``). + + The API validates a stored secret against these models (it must match + one). An empty list means no schema is declared: the credentials are + accepted as-is and validated by :meth:`test_connection`. + """ + return [] + def display_compliance_table( self, _findings: list,