feat: add SDK contract for dynamic provider construction args

- Add get_scan_arguments/get_connection_arguments to the provider contract
- Route non-built-in providers through the contract; built-ins unchanged
- Cover the external dynamic path in tests
This commit is contained in:
StylusFrost committed 2026-06-01 00:19:17 +02:00
1 parent cd15ed07eb
commit 51581c35ec
3 files changed
+106 -5

No files matched your search

+50 -3
View File
@@ -164,6 +164,7 @@ class TestInitializeProwlerProvider:
@patch("api.utils.return_prowler_provider")
def test_initialize_prowler_provider(self, mock_return_prowler_provider):
provider = MagicMock()
provider.provider = "aws"
provider.secret.secret = {"key": "value"}
mock_return_prowler_provider.return_value = MagicMock()
@@ -175,6 +176,7 @@ class TestInitializeProwlerProvider:
self, mock_return_prowler_provider
):
provider = MagicMock()
provider.provider = "aws"
provider.secret.secret = {"key": "value"}
mutelist_processor = MagicMock()
mutelist_processor.configuration = {"Mutelist": {"key": "value"}}
@@ -190,6 +192,7 @@ class TestProwlerProviderConnectionTest:
@patch("api.utils.return_prowler_provider")
def test_prowler_provider_connection_test(self, mock_return_prowler_provider):
provider = MagicMock()
provider.provider = "aws"
provider.uid = "1234567890"
provider.secret.secret = {"key": "value"}
mock_return_prowler_provider.return_value = MagicMock()
@@ -199,6 +202,29 @@ class TestProwlerProviderConnectionTest:
key="value", provider_id="1234567890", raise_on_exception=False
)
@patch("api.utils.return_prowler_provider")
def test_prowler_provider_connection_test_external_uses_contract(
self, mock_return_prowler_provider
):
"""External providers build connection kwargs through the SDK contract
(get_connection_arguments), with no provider_id forced by the API."""
provider = MagicMock()
provider.provider = "external-template"
provider.uid = "acme-1"
provider.secret.secret = {"api_key": "k"}
mock_return_prowler_provider.return_value.get_connection_arguments.return_value = {
"api_key": "k"
}
prowler_provider_connection_test(provider)
mock_return_prowler_provider.return_value.get_connection_arguments.assert_called_once_with(
"acme-1", {"api_key": "k"}
)
mock_return_prowler_provider.return_value.test_connection.assert_called_once_with(
api_key="k", raise_on_exception=False
)
@pytest.mark.django_db
@patch("api.utils.return_prowler_provider")
def test_prowler_provider_connection_test_without_secret(
@@ -573,7 +599,8 @@ class TestGetProwlerProviderKwargs:
assert result == expected_result
def test_get_prowler_provider_kwargs_unsupported_provider(self):
# Setup
# A non-built-in provider is resolved dynamically; one that is neither a
# built-in nor an installed entry-point provider cannot be resolved.
provider_uid = "provider_uid"
secret_dict = {"key": "value"}
secret_mock = MagicMock()
@@ -584,10 +611,30 @@ class TestGetProwlerProviderKwargs:
provider.secret = secret_mock
provider.uid = provider_uid
with pytest.raises(ValueError):
get_prowler_provider_kwargs(provider)
@patch("api.utils.return_prowler_provider")
def test_get_prowler_provider_kwargs_external_uses_contract(
self, mock_return_prowler_provider
):
"""External providers build constructor kwargs through the SDK contract
(get_scan_arguments), not a hardcoded branch in the API."""
provider = MagicMock()
provider.provider = "external-template"
provider.uid = "acme-1"
provider.secret.secret = {"api_key": "k"}
mock_return_prowler_provider.return_value.get_scan_arguments.return_value = {
"api_key": "k",
"custom": "acme-1",
}
result = get_prowler_provider_kwargs(provider)
expected_result = secret_dict.copy()
assert result == expected_result
mock_return_prowler_provider.return_value.get_scan_arguments.assert_called_once_with(
"acme-1", {"api_key": "k"}, None
)
assert result == {"api_key": "k", "custom": "acme-1"}
def test_get_prowler_provider_kwargs_no_secret(self):
# Setup
+22
View File
@@ -89,6 +89,18 @@ def get_prowler_provider_kwargs(
Returns:
dict: The provider kwargs for the corresponding provider class.
"""
# External providers declare their own uid/secret -> kwargs mapping through
# the SDK contract; built-in providers keep the explicit mapping below.
if not SDKProvider.is_builtin(provider.provider):
mutelist_content = (
mutelist_processor.configuration.get("Mutelist", {})
if mutelist_processor
else None
)
return return_prowler_provider(provider).get_scan_arguments(
provider.uid, provider.secret.secret, mutelist_content
)
prowler_provider_kwargs = provider.secret.secret
if provider.provider == Provider.ProviderChoices.AZURE.value:
prowler_provider_kwargs = {
@@ -205,6 +217,16 @@ def prowler_provider_connection_test(provider: Provider) -> Connection:
except Provider.secret.RelatedObjectDoesNotExist as secret_error:
return Connection(is_connected=False, error=secret_error)
# External providers declare their own connection kwargs through the SDK
# contract; built-in providers keep the explicit mapping below.
if not SDKProvider.is_builtin(provider.provider):
return prowler_provider.test_connection(
**prowler_provider.get_connection_arguments(
provider.uid, prowler_provider_kwargs
),
raise_on_exception=False,
)
# For IaC provider, construct the kwargs properly for test_connection
if provider.provider == Provider.ProviderChoices.IAC.value:
# Don't pass repository_url from secret, use scan_repository_url with the UID
+34 -2
View File
@@ -211,6 +211,38 @@ class Provider(ABC):
f"{self.__class__.__name__} has not implemented get_mutelist_finding_args()"
)
@classmethod
def get_scan_arguments(
cls,
provider_uid: str,
secret: dict,
mutelist_content: Optional[dict] = None,
) -> dict:
"""Build the provider constructor kwargs from a stored uid and secret.
This is the programmatic construction interface used by callers that
persist a provider as a single ``uid`` plus a ``secret`` dict (e.g. the
API), as opposed to the CLI which passes explicit per-provider flags.
The base implementation passes the secret through and adds the mutelist;
providers whose constructor needs the uid (e.g. as a subscription id) or
that rename/filter secret keys override this.
"""
kwargs = {**secret}
if mutelist_content:
kwargs["mutelist_content"] = mutelist_content
return kwargs
@classmethod
def get_connection_arguments(cls, provider_uid: str, secret: dict) -> dict:
"""Build the ``test_connection`` kwargs from a stored uid and secret.
Companion to :meth:`get_scan_arguments` for the connection check, which
often needs a different shape than the constructor. The base passes the
secret through; providers add their identity kwarg (and ``provider_id``
where their ``test_connection`` expects it).
"""
return {**secret}
def display_compliance_table(
self,
_findings: list,
@@ -303,8 +335,8 @@ class Provider(ABC):
# Kept for downstream forks that may extend the dispatch below
# with their own custom built-in branches and reference this name.
# The upstream chain dispatches by `arguments.provider` directly.
provider_class_name = (
f"{arguments.provider.capitalize()}Provider" # noqa: F841
provider_class_name = ( # noqa: F841
f"{arguments.provider.capitalize()}Provider"
)
fixer_config = load_and_validate_config_file(