feat(oci): add storage admin delete exclusion check (#11523)

This commit is contained in:
Hugo Pereira Brito
2026-06-12 11:10:46 +02:00
committed by GitHub
parent 2e82f1564f
commit 56831a7392
6 changed files with 545 additions and 1 deletions
+1
View File
@@ -9,6 +9,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
- `securityhub_delegated_admin_enabled_all_regions` check for AWS provider, verifying that Security Hub has a delegated administrator, is active in all opted-in regions, and has organization auto-enable on [(#11259)](https://github.com/prowler-cloud/prowler/pull/11259)
- `config_delegated_admin_and_org_aggregator_all_regions` check for AWS provider, verifying that AWS Config has a delegated administrator and an organization aggregator covering all AWS regions [(#11259)](https://github.com/prowler-cloud/prowler/pull/11259)
- `sagemaker_clarify_exists` check for AWS provider [(#11211)](https://github.com/prowler-cloud/prowler/pull/11211)
- `identity_storage_service_level_admins_scoped` check for OCI provider CIS 3.1 control 1.15, ensuring storage service-level administrators exclude delete permissions [(#11523)](https://github.com/prowler-cloud/prowler/pull/11523)
---
@@ -302,7 +302,9 @@
{
"Id": "1.15",
"Description": "Ensure storage service-level admins cannot delete resources they manage",
"Checks": [],
"Checks": [
"identity_storage_service_level_admins_scoped"
],
"Attributes": [
{
"Section": "1. Identity and Access Management",
@@ -0,0 +1,39 @@
{
"Provider": "oraclecloud",
"CheckID": "identity_storage_service_level_admins_scoped",
"CheckTitle": "OCI IAM storage service-level admin policies exclude delete permissions",
"CheckType": [],
"ServiceName": "identity",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "Policy",
"ResourceGroup": "IAM",
"Description": "**OCI IAM policies** are reviewed to ensure storage service-level administrator statements that grant `manage` permissions exclude the relevant storage delete permissions with `request.permission`. This supports CIS OCI 3.1 control 1.15 separation of duties for Block Volume, File Storage, and Object Storage administrators.",
"Risk": "Storage service-level administrators with unrestricted `manage` permissions can delete the resources they administer, including volumes, backups, file systems, mount targets, export sets, objects, or buckets. This weakens separation of duties and can lead to data loss, service disruption, or destructive insider activity.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://docs.oracle.com/en-us/iaas/Content/Identity/policyreference/policyreference.htm",
"https://docs.oracle.com/en-us/iaas/Content/Block/home.htm",
"https://docs.oracle.com/en-us/iaas/Content/File/home.htm",
"https://docs.oracle.com/en-us/iaas/Content/Object/home.htm"
],
"Remediation": {
"Code": {
"CLI": "oci iam policy update --policy-id <policy-ocid> --statements \"[\\\"Allow group VolumeUsers to manage volumes in tenancy where request.permission!='VOLUME_DELETE'\\\"]\"",
"NativeIaC": "",
"Other": "1. In OCI Console, go to Identity & Security > Policies\n2. Open each active policy that grants storage service-level administrators `manage` permissions\n3. Edit storage manage statements to exclude the relevant delete permission with `request.permission`\n4. Example: Allow group BucketUsers to manage buckets in tenancy where request.permission!='BUCKET_DELETE'\n5. Save changes",
"Terraform": "```hcl\nresource \"oci_identity_policy\" \"storage_admins\" {\n compartment_id = var.compartment_id\n name = \"storage-admins\"\n description = \"Storage administrators without delete permissions\"\n\n statements = [\n \"Allow group VolumeUsers to manage volumes in tenancy where request.permission!='VOLUME_DELETE'\",\n \"Allow group VolumeUsers to manage volume-backups in tenancy where request.permission!='VOLUME_BACKUP_DELETE'\",\n \"Allow group FileUsers to manage file-systems in tenancy where request.permission!='FILE_SYSTEM_DELETE'\",\n \"Allow group FileUsers to manage mount-targets in tenancy where request.permission!='MOUNT_TARGET_DELETE'\",\n \"Allow group FileUsers to manage export-sets in tenancy where request.permission!='EXPORT_SET_DELETE'\",\n \"Allow group BucketUsers to manage objects in tenancy where request.permission!='OBJECT_DELETE'\",\n \"Allow group BucketUsers to manage buckets in tenancy where request.permission!='BUCKET_DELETE'\"\n ]\n}\n```"
},
"Recommendation": {
"Text": "Exclude delete permissions from storage service-level administrator policies. Use `request.permission!='VOLUME_DELETE'`, `request.permission!='VOLUME_BACKUP_DELETE'`, `request.permission!='FILE_SYSTEM_DELETE'`, `request.permission!='MOUNT_TARGET_DELETE'`, `request.permission!='EXPORT_SET_DELETE'`, `request.permission!='OBJECT_DELETE'`, and `request.permission!='BUCKET_DELETE'` as appropriate for each storage manage statement.",
"Url": "https://hub.prowler.com/check/identity_storage_service_level_admins_scoped"
}
},
"Categories": [
"identity-access"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,176 @@
"""Check storage service-level administrators cannot delete managed resources."""
import re
from prowler.lib.check.models import Check, Check_Report_OCI
from prowler.providers.oraclecloud.services.identity.identity_client import (
identity_client,
)
STORAGE_DELETE_PERMISSIONS_BY_RESOURCE = {
"volumes": {"VOLUME_DELETE"},
"volume-backups": {"VOLUME_BACKUP_DELETE"},
"file-systems": {"FILE_SYSTEM_DELETE"},
"mount-targets": {"MOUNT_TARGET_DELETE"},
"export-sets": {"EXPORT_SET_DELETE"},
"objects": {"OBJECT_DELETE"},
"buckets": {"BUCKET_DELETE"},
"volume-family": {"VOLUME_DELETE", "VOLUME_BACKUP_DELETE"},
"file-family": {"FILE_SYSTEM_DELETE", "MOUNT_TARGET_DELETE", "EXPORT_SET_DELETE"},
"object-family": {"OBJECT_DELETE", "BUCKET_DELETE"},
}
ALL_STORAGE_DELETE_PERMISSIONS = set().union(
*STORAGE_DELETE_PERMISSIONS_BY_RESOURCE.values()
)
STORAGE_DELETE_PERMISSIONS_BY_RESOURCE["all-resources"] = ALL_STORAGE_DELETE_PERMISSIONS
MANAGE_STATEMENT_PATTERN = re.compile(
r"\ballow\s+group\b.+?\bto\s+manage\s+(?P<resource>[a-z-]+)\b",
re.IGNORECASE,
)
QUOTED_LITERAL_PATTERN = re.compile(r"'(?:\\.|[^'\\])*'|\"(?:\\.|[^\"\\])*\"")
def _normalize_statement(statement: str) -> str:
"""Collapse whitespace in an OCI policy statement."""
return " ".join(statement.strip().split())
def _has_disjunctive_condition(statement: str) -> bool:
"""Return True when the WHERE condition can allow alternate branches."""
condition = re.split(r"\bwhere\b", statement, flags=re.IGNORECASE, maxsplit=1)
if len(condition) != 2:
return False
condition_without_literals = QUOTED_LITERAL_PATTERN.sub("", condition[1])
return bool(
re.search(r"\b(any|or)\b|\|\|", condition_without_literals, re.IGNORECASE)
)
def _storage_manage_resource(statement: str) -> str | None:
"""Return the managed storage resource in a policy statement, if any."""
normalized_statement = _normalize_statement(statement)
match = MANAGE_STATEMENT_PATTERN.search(normalized_statement)
if not match:
return None
resource = match.group("resource").lower()
if resource not in STORAGE_DELETE_PERMISSIONS_BY_RESOURCE:
return None
return resource
def _excluded_permissions(statement: str) -> set[str]:
"""Return delete permissions explicitly excluded with request.permission != value."""
if _has_disjunctive_condition(statement):
return set()
exclusions = set()
for permission in ALL_STORAGE_DELETE_PERMISSIONS:
pattern = re.compile(
rf"\brequest\.permission\s*!=\s*['\"]?{re.escape(permission)}['\"]?\b",
re.IGNORECASE,
)
if pattern.search(statement):
exclusions.add(permission)
return exclusions
def _missing_delete_exclusions(statement: str) -> tuple[str, set[str]] | None:
"""Return the storage resource and missing delete exclusions for a statement."""
normalized_statement = _normalize_statement(statement)
resource = _storage_manage_resource(normalized_statement)
if not resource:
return None
required_permissions = STORAGE_DELETE_PERMISSIONS_BY_RESOURCE[resource]
excluded_permissions = _excluded_permissions(normalized_statement)
missing_permissions = required_permissions - excluded_permissions
if not missing_permissions:
return None
return resource, missing_permissions
class identity_storage_service_level_admins_scoped(Check):
"""Ensure storage service-level admins cannot delete resources they manage."""
def execute(self) -> list[Check_Report_OCI]:
"""Execute the storage service-level administrators scoped check.
Returns:
A list of OCI check reports for active non-tenant-admin policies.
"""
findings = []
for policy in identity_client.policies:
if policy.lifecycle_state != "ACTIVE":
continue
if policy.name.upper() == "TENANT ADMIN POLICY":
continue
region = policy.region if hasattr(policy, "region") else "global"
violations = []
has_storage_manage_statement = False
for statement in policy.statements:
if _storage_manage_resource(statement):
has_storage_manage_statement = True
missing_result = _missing_delete_exclusions(statement)
if not missing_result:
continue
resource, missing_permissions = missing_result
violations.append(
f"statement `{_normalize_statement(statement)}` manages {resource} without excluding: {', '.join(sorted(missing_permissions))}"
)
if not has_storage_manage_statement:
continue
report = Check_Report_OCI(
metadata=self.metadata(),
resource=policy,
region=region,
resource_id=policy.id,
resource_name=policy.name,
compartment_id=policy.compartment_id,
)
if violations:
report.status = "FAIL"
report.status_extended = (
f"Policy '{policy.name}' allows storage service-level administrators to manage storage resources without explicitly excluding required delete permissions: "
+ "; ".join(violations)
+ "."
)
else:
report.status = "PASS"
report.status_extended = f"Policy '{policy.name}' excludes required storage delete permissions from storage manage statements."
findings.append(report)
if not findings:
region = (
identity_client.audited_regions[0].key
if identity_client.audited_regions
else "global"
)
report = Check_Report_OCI(
metadata=self.metadata(),
resource={},
region=region,
resource_id=identity_client.audited_tenancy,
resource_name="Tenancy",
compartment_id=identity_client.audited_tenancy,
)
report.status = "PASS"
report.status_extended = "No active storage service-level administrator policies grant manage permissions without excluding delete permissions."
findings.append(report)
return findings
@@ -0,0 +1,326 @@
from datetime import datetime
from unittest import mock
import pytest
from prowler.lib.check.models import Check_Report_OCI
from prowler.providers.oraclecloud.services.identity.identity_service import Policy
from tests.providers.oraclecloud.oci_fixtures import (
OCI_COMPARTMENT_ID,
OCI_REGION,
OCI_TENANCY_ID,
set_mocked_oraclecloud_provider,
)
CHECK_PATH = "prowler.providers.oraclecloud.services.identity.identity_storage_service_level_admins_scoped.identity_storage_service_level_admins_scoped"
def _policy(
name: str, statements: list[str], lifecycle_state: str = "ACTIVE"
) -> Policy:
return Policy(
id=f"ocid1.policy.oc1..{name.lower().replace(' ', '-')}",
name=name,
description="Test policy",
compartment_id=OCI_COMPARTMENT_ID,
statements=statements,
time_created=datetime.now(),
lifecycle_state=lifecycle_state,
region=OCI_REGION,
)
def _identity_client(policies: list[Policy]) -> mock.MagicMock:
identity_client = mock.MagicMock()
identity_client.policies = policies
identity_client.audited_tenancy = OCI_TENANCY_ID
identity_client.audited_regions = [mock.MagicMock(key=OCI_REGION)]
return identity_client
def _run_check(policies: list[Policy]) -> list[Check_Report_OCI]:
identity_client = _identity_client(policies)
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_oraclecloud_provider(),
),
mock.patch(f"{CHECK_PATH}.identity_client", new=identity_client),
):
from prowler.providers.oraclecloud.services.identity.identity_storage_service_level_admins_scoped.identity_storage_service_level_admins_scoped import (
identity_storage_service_level_admins_scoped,
)
return identity_storage_service_level_admins_scoped().execute()
class Test_identity_storage_service_level_admins_scoped:
def test_no_policies_passes_with_tenancy_finding(self):
result = _run_check([])
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].resource_id == OCI_TENANCY_ID
assert result[0].resource_name == "Tenancy"
assert (
result[0].status_extended
== "No active storage service-level administrator policies grant manage permissions without excluding delete permissions."
)
def test_manage_volumes_without_delete_exclusion_fails(self):
result = _run_check(
[
_policy(
"Volume Admins",
["Allow group VolumeUsers to manage volumes in tenancy"],
)
]
)
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].resource_name == "Volume Admins"
assert "VOLUME_DELETE" in result[0].status_extended
assert (
"Allow group VolumeUsers to manage volumes in tenancy"
in result[0].status_extended
)
def test_manage_volumes_with_delete_exclusion_passes(self):
result = _run_check(
[
_policy(
"Volume Admins",
[
"Allow group VolumeUsers to manage volumes in tenancy where request.permission!='VOLUME_DELETE'"
],
)
]
)
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "Policy 'Volume Admins' excludes required storage delete permissions from storage manage statements."
)
def test_delete_exclusion_parser_is_case_and_whitespace_insensitive(self):
result = _run_check(
[
_policy(
"Volume Admins",
[
" allow group VolumeUsers TO manage volumes in tenancy WHERE request.permission != 'volume_delete' "
],
)
]
)
assert len(result) == 1
assert result[0].status == "PASS"
def test_generic_where_clause_does_not_pass(self):
result = _run_check(
[
_policy(
"Bucket Admins",
[
"Allow group BucketUsers to manage buckets in tenancy where request.region='iad'"
],
)
]
)
assert len(result) == 1
assert result[0].status == "FAIL"
assert "BUCKET_DELETE" in result[0].status_extended
assert "request.region='iad'" in result[0].status_extended
@pytest.mark.parametrize(
"statement",
[
"Allow group BucketUsers to manage buckets in tenancy where ANY {request.permission!='BUCKET_DELETE', request.region='iad'}",
"Allow group BucketUsers to manage buckets in tenancy where request.permission!='BUCKET_DELETE' OR request.region='iad'",
],
)
def test_disjunctive_delete_exclusion_does_not_pass(self, statement):
result = _run_check([_policy("Bucket Admins", [statement])])
assert len(result) == 1
assert result[0].status == "FAIL"
assert "BUCKET_DELETE" in result[0].status_extended
def test_quoted_literals_do_not_make_delete_exclusion_disjunctive(self):
result = _run_check(
[
_policy(
"Bucket Admins",
[
"Allow group BucketUsers to manage buckets in tenancy where request.permission!='BUCKET_DELETE' and target.tag.namespace='any-tag'"
],
)
]
)
assert len(result) == 1
assert result[0].status == "PASS"
@pytest.mark.parametrize(
"resource,permission",
[
("file-systems", "FILE_SYSTEM_DELETE"),
("mount-targets", "MOUNT_TARGET_DELETE"),
("export-sets", "EXPORT_SET_DELETE"),
("volumes", "VOLUME_DELETE"),
("volume-backups", "VOLUME_BACKUP_DELETE"),
("objects", "OBJECT_DELETE"),
("buckets", "BUCKET_DELETE"),
],
)
def test_storage_resources_require_matching_delete_exclusion(
self, resource, permission
):
fail_result = _run_check(
[
_policy(
"Storage Admins",
[f"Allow group StorageUsers to manage {resource} in tenancy"],
)
]
)
pass_result = _run_check(
[
_policy(
"Storage Admins",
[
f"Allow group StorageUsers to manage {resource} in tenancy where request.permission != '{permission}'"
],
)
]
)
assert len(fail_result) == 1
assert fail_result[0].status == "FAIL"
assert permission in fail_result[0].status_extended
assert len(pass_result) == 1
assert pass_result[0].status == "PASS"
def test_file_family_fails_until_all_delete_permissions_are_excluded(self):
partial_result = _run_check(
[
_policy(
"File Admins",
[
"Allow group FileUsers to manage file-family in tenancy where ALL {request.permission!='FILE_SYSTEM_DELETE', request.permission!='MOUNT_TARGET_DELETE'}"
],
)
]
)
complete_result = _run_check(
[
_policy(
"File Admins",
[
"Allow group FileUsers to manage file-family in tenancy where ALL {request.permission!='FILE_SYSTEM_DELETE', request.permission!='MOUNT_TARGET_DELETE', request.permission!='EXPORT_SET_DELETE'}"
],
)
]
)
assert len(partial_result) == 1
assert partial_result[0].status == "FAIL"
assert "EXPORT_SET_DELETE" in partial_result[0].status_extended
assert len(complete_result) == 1
assert complete_result[0].status == "PASS"
@pytest.mark.parametrize(
"family,missing_permission,statement",
[
(
"volume-family",
"VOLUME_BACKUP_DELETE",
"Allow group VolumeUsers to manage volume-family in tenancy where request.permission!='VOLUME_DELETE'",
),
(
"object-family",
"BUCKET_DELETE",
"Allow group BucketUsers to manage object-family in tenancy where request.permission!='OBJECT_DELETE'",
),
(
"all-resources",
"BUCKET_DELETE",
"Allow group StorageUsers to manage all-resources in tenancy where ALL {request.permission!='VOLUME_DELETE', request.permission!='VOLUME_BACKUP_DELETE', request.permission!='FILE_SYSTEM_DELETE', request.permission!='MOUNT_TARGET_DELETE', request.permission!='EXPORT_SET_DELETE', request.permission!='OBJECT_DELETE'}",
),
],
)
def test_families_and_all_resources_fail_unless_all_delete_permissions_are_excluded(
self, family, missing_permission, statement
):
result = _run_check([_policy("Storage Admins", [statement])])
assert len(result) == 1
assert result[0].status == "FAIL"
assert family in result[0].status_extended
assert missing_permission in result[0].status_extended
def test_all_resources_passes_when_all_storage_delete_permissions_are_excluded(
self,
):
result = _run_check(
[
_policy(
"Storage Admins",
[
"Allow group StorageUsers to manage all-resources in tenancy where ALL {request.permission!='VOLUME_DELETE', request.permission!='VOLUME_BACKUP_DELETE', request.permission!='FILE_SYSTEM_DELETE', request.permission!='MOUNT_TARGET_DELETE', request.permission!='EXPORT_SET_DELETE', request.permission!='OBJECT_DELETE', request.permission!='BUCKET_DELETE'}"
],
)
]
)
assert len(result) == 1
assert result[0].status == "PASS"
def test_inactive_policies_are_ignored(self):
result = _run_check(
[
_policy(
"Inactive Volume Admins",
["Allow group VolumeUsers to manage volumes in tenancy"],
lifecycle_state="INACTIVE",
)
]
)
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].resource_name == "Tenancy"
def test_tenant_admin_policy_is_ignored(self):
result = _run_check(
[
_policy(
"Tenant Admin Policy",
["Allow group Administrators to manage all-resources in tenancy"],
)
]
)
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].resource_name == "Tenancy"
def test_policies_without_storage_manage_statements_are_ignored(self):
result = _run_check(
[
_policy(
"Network Admins",
["Allow group NetworkUsers to manage vcns in tenancy"],
)
]
)
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].resource_name == "Tenancy"