From 5daa39c4cbed11ae7f53752f07c568b29c8dbdb3 Mon Sep 17 00:00:00 2001 From: StylusFrost Date: Mon, 1 Jun 2026 22:09:21 +0200 Subject: [PATCH] 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 --- api/src/backend/api/tests/test_serializers.py | 66 +++++++++++-------- api/src/backend/api/v1/serializers.py | 57 +++++++++------- prowler/providers/common/provider.py | 25 ++++--- 3 files changed, 84 insertions(+), 64 deletions(-) diff --git a/api/src/backend/api/tests/test_serializers.py b/api/src/backend/api/tests/test_serializers.py index a2feb4e498..0422ac4a01 100644 --- a/api/src/backend/api/tests/test_serializers.py +++ b/api/src/backend/api/tests/test_serializers.py @@ -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 ) diff --git a/api/src/backend/api/v1/serializers.py b/api/src/backend/api/v1/serializers.py index 0a8d31cd1e..46187dc220 100644 --- a/api/src/backend/api/v1/serializers.py +++ b/api/src/backend/api/v1/serializers.py @@ -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 diff --git a/prowler/providers/common/provider.py b/prowler/providers/common/provider.py index 7462df70ad..27d46f2a35 100644 --- a/prowler/providers/common/provider.py +++ b/prowler/providers/common/provider.py @@ -244,22 +244,21 @@ 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. + def get_credentials_schema(cls) -> dict: + """Return the provider's credential schemas keyed by 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=...)``). + Maps each secret type the provider accepts (``"static"``, ``"role"`` or + ``"service_account"``) to the pydantic model that validates a secret of + that type. The provider declares which type each schema belongs to, so + the API validates a secret against the model for the secret type it is + created with and the chosen type stays bound to the shape it claims. - 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`. + Each model documents each field via ``Field(description=...)`` and + whether it is required (no default) or optional. An empty dict means no + schema is declared: the secret is accepted as an object and validated by + :meth:`test_connection`. """ - return [] + return {} def display_compliance_table( self,