fix(api): support regionless OCI credentials

- Make OCI region optional while accepting legacy input

- Strip legacy region before storing or scanning

- Cover serializer, API, and provider kwargs behavior
This commit is contained in:
Hugo P.Brito committed 2026-07-16 11:18:41 +01:00
1 parent 59b13778e2
commit d61906ffb0
7 files changed
+378 -21

No files matched your search

@@ -0,0 +1 @@
OCI provider secrets no longer require `region`; legacy `region` input is accepted for backwards compatibility but ignored before storing or scanning
+64 -1
View File
@@ -3,7 +3,12 @@ from api.v1.serializer_utils.integrations import (
JiraCredentialSerializer,
S3ConfigSerializer,
)
from api.v1.serializers import ImageProviderSecret, KubernetesProviderSecret
from api.v1.serializer_utils.providers import ProviderSecretField
from api.v1.serializers import (
ImageProviderSecret,
KubernetesProviderSecret,
OracleCloudProviderSecret,
)
from rest_framework.exceptions import ValidationError
@@ -190,6 +195,64 @@ class TestImageProviderSecret:
assert "non_field_errors" in serializer.errors
class TestOracleCloudProviderSecret:
def valid_secret(self, **overrides):
secret = {
"user": "ocid1.user.oc1..aaaaaaaexample",
"fingerprint": "aa:bb:cc:dd:ee:ff:00:11:22:33:44:55:66:77:88:99",
"key_content": "fake-base64-key-content",
"tenancy": "ocid1.tenancy.oc1..aaaaaaaexample",
}
secret.update(overrides)
return secret
def test_accepts_regionless_secret(self):
serializer = OracleCloudProviderSecret(data=self.valid_secret())
assert serializer.is_valid(), serializer.errors
assert "region" not in serializer.validated_data
def test_accepts_and_ignores_region_field(self):
secret = self.valid_secret(region="us-phoenix-1")
serializer = OracleCloudProviderSecret(data=secret)
assert serializer.is_valid(), serializer.errors
assert "region" not in serializer.validated_data
@pytest.mark.parametrize(
"legacy_field, legacy_value",
[
("region", None),
("region", ""),
("region", {"name": "us-ashburn-1"}),
],
)
def test_accepts_and_ignores_any_legacy_region_value(
self, legacy_field, legacy_value
):
serializer = OracleCloudProviderSecret(
data=self.valid_secret(**{legacy_field: legacy_value})
)
assert serializer.is_valid(), serializer.errors
assert legacy_field not in serializer.validated_data
class TestProviderSecretFieldSchema:
def test_oraclecloud_schema_includes_legacy_region_field(self):
schema = ProviderSecretField._spectacular_annotation["field"]
oraclecloud_schema = next(
credential_schema
for credential_schema in schema["oneOf"]
if credential_schema["title"]
== "Oracle Cloud Infrastructure (OCI) API Key Credentials"
)
assert oraclecloud_schema["properties"]["region"]["deprecated"] is True
class TestKubernetesProviderSecret:
def test_valid_static_kubeconfig_is_accepted(self):
kubeconfig_content = """
+86 -3
View File
@@ -171,6 +171,53 @@ class TestInitializeProwlerProvider:
key="value", mutelist_content={"key": "value"}
)
@patch("api.utils.return_prowler_provider")
def test_initialize_oraclecloud_provider_removes_region_string(
self, mock_return_prowler_provider
):
provider = MagicMock()
provider.provider = Provider.ProviderChoices.ORACLECLOUD.value
provider.secret.secret = {
"user": "ocid1.user.oc1..fake",
"fingerprint": "00:11:22:33:44:55:66:77",
"key_content": "fake-base64-key-content",
"tenancy": "ocid1.tenancy.oc1..fake",
"region": "us-ashburn-1",
}
mock_return_prowler_provider.return_value = MagicMock()
initialize_prowler_provider(provider)
mock_return_prowler_provider.return_value.assert_called_once_with(
user="ocid1.user.oc1..fake",
fingerprint="00:11:22:33:44:55:66:77",
key_content="fake-base64-key-content",
tenancy="ocid1.tenancy.oc1..fake",
)
@patch("api.utils.return_prowler_provider")
def test_initialize_oraclecloud_provider_without_region_omits_scan_filter(
self, mock_return_prowler_provider
):
provider = MagicMock()
provider.provider = Provider.ProviderChoices.ORACLECLOUD.value
provider.secret.secret = {
"user": "ocid1.user.oc1..fake",
"fingerprint": "00:11:22:33:44:55:66:77",
"key_content": "fake-base64-key-content",
"tenancy": "ocid1.tenancy.oc1..fake",
}
mock_return_prowler_provider.return_value = MagicMock()
initialize_prowler_provider(provider)
mock_return_prowler_provider.return_value.assert_called_once_with(
user="ocid1.user.oc1..fake",
fingerprint="00:11:22:33:44:55:66:77",
key_content="fake-base64-key-content",
tenancy="ocid1.tenancy.oc1..fake",
)
class TestProwlerProviderConnectionTest:
@patch("api.utils.return_prowler_provider")
@@ -185,6 +232,37 @@ class TestProwlerProviderConnectionTest:
key="value", provider_id="1234567890", raise_on_exception=False
)
@patch("api.utils.return_prowler_provider")
def test_oraclecloud_connection_test_uses_direct_credentials_without_region(
self, mock_return_prowler_provider
):
provider = MagicMock()
provider.uid = "ocid1.tenancy.oc1..aaaaaaaexample"
provider.provider = Provider.ProviderChoices.ORACLECLOUD.value
provider.secret.secret = {
"user": "ocid1.user.oc1..aaaaaaaexample",
"fingerprint": "00:11:22:33:44:55:66:77",
"key_content": "fake-base64-key-content",
"tenancy": "ocid1.tenancy.oc1..aaaaaaaexample",
}
mock_return_prowler_provider.return_value = MagicMock()
prowler_provider_connection_test(provider)
mock_return_prowler_provider.return_value.test_connection.assert_called_once_with(
user="ocid1.user.oc1..aaaaaaaexample",
fingerprint="00:11:22:33:44:55:66:77",
key_content="fake-base64-key-content",
tenancy="ocid1.tenancy.oc1..aaaaaaaexample",
region=getattr(
OraclecloudProvider,
"_bootstrap_region",
OraclecloudProvider._home_region,
),
provider_id="ocid1.tenancy.oc1..aaaaaaaexample",
raise_on_exception=False,
)
@pytest.mark.django_db
@patch("api.utils.return_prowler_provider")
def test_prowler_provider_connection_test_without_secret(
@@ -356,7 +434,7 @@ class TestGetProwlerProviderKwargs:
expected_result = {**secret_dict, **expected_extra_kwargs}
assert result == expected_result
def test_get_prowler_provider_kwargs_oraclecloud_converts_region_string_to_set(
def test_get_prowler_provider_kwargs_oraclecloud_removes_region(
self,
):
secret_dict = {
@@ -377,8 +455,13 @@ class TestGetProwlerProviderKwargs:
result = get_prowler_provider_kwargs(provider)
expected_result = {**secret_dict, "region": {"us-ashburn-1"}}
assert result == expected_result
assert result == {
"user": "ocid1.user.oc1..fake",
"fingerprint": "00:11:22:33:44:55:66:77",
"key_content": "-----BEGIN PRIVATE KEY-----\nfake\n-----END PRIVATE KEY-----",
"tenancy": "ocid1.tenancy.oc1..fake",
"pass_phrase": "fake-passphrase",
}
def test_get_prowler_provider_kwargs_with_mutelist(self):
provider_uid = "provider_uid"
+139 -3
View File
@@ -2917,6 +2917,48 @@ class TestProviderGroupViewSet:
@pytest.mark.django_db
class TestProviderSecretViewSet:
@staticmethod
def _oraclecloud_secret(**overrides):
secret = {
"user": "ocid1.user.oc1..aaaaaaaakldibrbov4ubh25aqdeiroklxjngwka7u6w7no3glmdq3n5sxtkq",
"fingerprint": "aa:bb:cc:dd:ee:ff:00:11:22:33:44:55:66:77:88:99",
"key_content": "test-key-content",
"tenancy": "ocid1.tenancy.oc1..aaaaaaaa3dwoazoox4q7wrvriywpokp5grlhgnkwtyt6dmwyou7no6mdmzda",
}
secret.update(overrides)
return secret
def _create_oraclecloud_secret(
self,
authenticated_client,
oraclecloud_provider,
secret,
name="OCI Secret",
):
data = {
"data": {
"type": "provider-secrets",
"attributes": {
"name": name,
"secret_type": ProviderSecret.TypeChoices.STATIC,
"secret": secret,
},
"relationships": {
"provider": {
"data": {
"type": "providers",
"id": str(oraclecloud_provider.id),
}
}
},
}
}
return authenticated_client.post(
reverse("providersecret-list"),
data=json.dumps(data),
content_type="application/vnd.api+json",
)
def test_provider_secrets_list(self, authenticated_client, provider_secret_fixture):
response = authenticated_client.get(reverse("providersecret-list"))
assert response.status_code == status.HTTP_200_OK
@@ -3076,7 +3118,6 @@ current-context: test-context
"fingerprint": "aa:bb:cc:dd:ee:ff:00:11:22:33:44:55:66:77:88:99",
"key_content": "-----BEGIN RSA PRIVATE KEY-----\ntest-key-content\n-----END RSA PRIVATE KEY-----",
"tenancy": "ocid1.tenancy.oc1..aaaaaaaa3dwoazoox4q7wrvriywpokp5grlhgnkwtyt6dmwyou7no6mdmzda",
"region": "us-ashburn-1",
},
),
# OCI with API key credentials (with key_file)
@@ -3088,7 +3129,6 @@ current-context: test-context
"fingerprint": "aa:bb:cc:dd:ee:ff:00:11:22:33:44:55:66:77:88:99",
"key_file": "/path/to/oci_api_key.pem",
"tenancy": "ocid1.tenancy.oc1..aaaaaaaa3dwoazoox4q7wrvriywpokp5grlhgnkwtyt6dmwyou7no6mdmzda",
"region": "us-ashburn-1",
},
),
# OCI with API key credentials (with passphrase)
@@ -3100,7 +3140,6 @@ current-context: test-context
"fingerprint": "aa:bb:cc:dd:ee:ff:00:11:22:33:44:55:66:77:88:99",
"key_content": "-----BEGIN RSA PRIVATE KEY-----\ntest-encrypted-key\n-----END RSA PRIVATE KEY-----",
"tenancy": "ocid1.tenancy.oc1..aaaaaaaa3dwoazoox4q7wrvriywpokp5grlhgnkwtyt6dmwyou7no6mdmzda",
"region": "us-ashburn-1",
"pass_phrase": "my-secure-passphrase",
},
),
@@ -3258,6 +3297,103 @@ current-context: test-context
== data["data"]["relationships"]["provider"]["data"]["id"]
)
def test_provider_secrets_create_oraclecloud_without_region_stores_no_region(
self,
authenticated_client,
oraclecloud_provider,
):
response = self._create_oraclecloud_secret(
authenticated_client,
oraclecloud_provider,
self._oraclecloud_secret(),
)
assert response.status_code == status.HTTP_201_CREATED
provider_secret = ProviderSecret.objects.get()
assert "region" not in provider_secret.secret
def test_provider_secrets_create_oraclecloud_accepts_and_ignores_region(
self,
authenticated_client,
oraclecloud_provider,
):
response = self._create_oraclecloud_secret(
authenticated_client,
oraclecloud_provider,
self._oraclecloud_secret(
key_content=" test-key-content ", region=" us-ashburn-1 "
),
)
assert response.status_code == status.HTTP_201_CREATED
provider_secret = ProviderSecret.objects.get()
assert provider_secret.secret["key_content"] == "test-key-content"
assert "region" not in provider_secret.secret
def test_provider_secrets_update_oraclecloud_without_region_stores_no_region(
self,
authenticated_client,
oraclecloud_provider,
):
create_response = self._create_oraclecloud_secret(
authenticated_client,
oraclecloud_provider,
self._oraclecloud_secret(),
)
provider_secret = ProviderSecret.objects.get(
id=create_response.json()["data"]["id"]
)
data = {
"data": {
"type": "provider-secrets",
"id": str(provider_secret.id),
"attributes": {"secret": self._oraclecloud_secret()},
}
}
response = authenticated_client.patch(
reverse("providersecret-detail", kwargs={"pk": provider_secret.id}),
data=json.dumps(data),
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_200_OK
provider_secret.refresh_from_db()
assert "region" not in provider_secret.secret
def test_provider_secrets_update_oraclecloud_accepts_and_ignores_region(
self,
authenticated_client,
oraclecloud_provider,
):
create_response = self._create_oraclecloud_secret(
authenticated_client,
oraclecloud_provider,
self._oraclecloud_secret(),
)
provider_secret = ProviderSecret.objects.get(
id=create_response.json()["data"]["id"]
)
data = {
"data": {
"type": "provider-secrets",
"id": str(provider_secret.id),
"attributes": {
"secret": self._oraclecloud_secret(region=" us-ashburn-1 ")
},
}
}
response = authenticated_client.patch(
reverse("providersecret-detail", kwargs={"pk": provider_secret.id}),
data=json.dumps(data),
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_200_OK
provider_secret.refresh_from_db()
assert "region" not in provider_secret.secret
@pytest.mark.parametrize(
"attributes, error_code, error_pointer",
(
+48 -6
View File
@@ -252,12 +252,6 @@ def get_prowler_provider_kwargs(
**prowler_provider_kwargs,
"filter_accounts": [provider.uid],
}
elif provider.provider == Provider.ProviderChoices.ORACLECLOUD.value:
if isinstance(prowler_provider_kwargs.get("region"), str):
prowler_provider_kwargs = {
**prowler_provider_kwargs,
"region": {prowler_provider_kwargs["region"]},
}
elif provider.provider == Provider.ProviderChoices.OPENSTACK.value:
# clouds_yaml_content, clouds_yaml_cloud and provider_id are validated
# in the provider itself, so it's not needed here.
@@ -288,6 +282,11 @@ def get_prowler_provider_kwargs(
**{k: v for k, v in prowler_provider_kwargs.items() if v},
}
elif provider.provider == Provider.ProviderChoices.ORACLECLOUD.value:
prowler_provider_kwargs = _normalize_oraclecloud_provider_kwargs(
prowler_provider_kwargs
)
if mutelist_processor:
mutelist_content = mutelist_processor.configuration.get("Mutelist", {})
# IaC and Image providers don't support mutelist (both use Trivy's built-in logic)
@@ -300,6 +299,40 @@ def get_prowler_provider_kwargs(
return prowler_provider_kwargs
def _normalize_oraclecloud_provider_kwargs(secret: dict) -> dict:
"""Normalize external OCI secret fields into SDK provider kwargs."""
prowler_provider_kwargs = secret.copy()
prowler_provider_kwargs.pop("region", None)
return prowler_provider_kwargs
def _normalize_oraclecloud_connection_test_kwargs(secret: dict) -> dict:
"""Normalize external OCI secret fields into test_connection kwargs."""
from prowler.providers.oraclecloud.oraclecloud_provider import OraclecloudProvider
prowler_provider_kwargs = secret.copy()
prowler_provider_kwargs.pop("region", None)
if (
prowler_provider_kwargs.get("user")
and prowler_provider_kwargs.get("fingerprint")
and prowler_provider_kwargs.get("tenancy")
and (
prowler_provider_kwargs.get("key_content")
or prowler_provider_kwargs.get("key_file")
)
):
# Connection validation needs one OCI endpoint, but scans remain unfiltered.
prowler_provider_kwargs["region"] = getattr(
OraclecloudProvider,
"_bootstrap_region",
OraclecloudProvider._home_region,
)
return prowler_provider_kwargs
def initialize_prowler_provider(
provider: Provider,
mutelist_processor: Processor | None = None,
@@ -402,6 +435,15 @@ def prowler_provider_connection_test(provider: Provider) -> Connection:
if prowler_provider_kwargs.get("registry_token"):
image_kwargs["registry_token"] = prowler_provider_kwargs["registry_token"]
return prowler_provider.test_connection(**image_kwargs)
elif provider.provider == Provider.ProviderChoices.ORACLECLOUD.value:
oraclecloud_kwargs = _normalize_oraclecloud_connection_test_kwargs(
prowler_provider_kwargs
)
return prowler_provider.test_connection(
**oraclecloud_kwargs,
provider_id=provider.uid,
raise_on_exception=False,
)
else:
return prowler_provider.test_connection(
**prowler_provider_kwargs,
@@ -295,16 +295,21 @@ from rest_framework_json_api import serializers
"type": "string",
"description": "The OCID of the tenancy.",
},
"region": {
"type": "string",
"description": "The OCI region identifier (e.g., us-ashburn-1, us-phoenix-1).",
},
"pass_phrase": {
"type": "string",
"description": "The passphrase for the private key, if encrypted.",
},
"region": {
"type": "string",
"deprecated": True,
"description": "Legacy OCI region field accepted for backwards compatibility but ignored; OCI scans all regions.",
},
},
"required": ["user", "fingerprint", "tenancy", "region"],
"required": ["user", "fingerprint", "tenancy"],
"anyOf": [
{"required": ["key_file"]},
{"required": ["key_content"]},
],
},
{
"type": "object",
+30 -3
View File
@@ -1672,6 +1672,7 @@ class BaseWriteProviderSecretSerializer(BaseWriteSerializer):
validation_error.detail[f"secret/{key}"] = value
del validation_error.detail[key]
raise validation_error
return serializer.validated_data
class AwsProviderSecret(serializers.Serializer):
@@ -1813,14 +1814,32 @@ class IacProviderSecret(serializers.Serializer):
resource_name = "provider-secrets"
class LegacyOCIRegionField(serializers.Field):
def to_internal_value(self, data):
return data
def to_representation(self, value):
return value
class OracleCloudProviderSecret(serializers.Serializer):
user = serializers.CharField()
fingerprint = serializers.CharField()
key_file = serializers.CharField(required=False)
key_content = serializers.CharField(required=False)
tenancy = serializers.CharField()
region = serializers.CharField()
pass_phrase = serializers.CharField(required=False)
region = LegacyOCIRegionField(required=False, allow_null=True)
def validate(self, attrs):
attrs.pop("region", None)
if "key_file" not in attrs and "key_content" not in attrs:
raise serializers.ValidationError(
{"key_file": "Either key_file or key_content must be provided."}
)
return attrs
class Meta:
resource_name = "provider-secrets"
@@ -1965,7 +1984,11 @@ class ProviderSecretCreateSerializer(RLSSerializer, BaseWriteProviderSecretSeria
secret = attrs.get("secret")
validated_attrs = super().validate(attrs)
self.validate_secret_based_on_provider(provider.provider, secret_type, secret)
validated_secret = self.validate_secret_based_on_provider(
provider.provider, secret_type, secret
)
if provider.provider == Provider.ProviderChoices.ORACLECLOUD.value:
validated_attrs["secret"] = validated_secret
return validated_attrs
@@ -1997,7 +2020,11 @@ class ProviderSecretUpdateSerializer(BaseWriteProviderSecretSerializer):
secret = attrs.get("secret")
validated_attrs = super().validate(attrs)
self.validate_secret_based_on_provider(provider.provider, secret_type, secret)
validated_secret = self.validate_secret_based_on_provider(
provider.provider, secret_type, secret
)
if provider.provider == Provider.ProviderChoices.ORACLECLOUD.value:
validated_attrs["secret"] = validated_secret
return validated_attrs