#!/usr/bin/env python3 """ CIS OCI Foundations Benchmark 3.0 — Compliance Checker Performs all 48 CIS checks against an OCI tenancy (no OBP). Generates report.json + report.html. Usage: python3 cis_runner.py --config /path/to/oci/config --output /path/to/dir \ --tenancy-name mytenancy [--regions r1,r2] [--level 1] """ import argparse, json, datetime, sys, logging, time, re from pathlib import Path from concurrent.futures import ThreadPoolExecutor, as_completed import oci log = logging.getLogger("cis_runner") logging.basicConfig(level=logging.INFO, format="[%(levelname)s] %(name)s: %(message)s") # ─── Check catalog ──────────────────────────────────────────────────────────── CHECKS = { "1.1": {"id":"IAM-1","section":"Identity and Access Management","title":"Ensure service level admins are created to manage resources of particular service","level":1}, "1.2": {"id":"IAM-2","section":"Identity and Access Management","title":"Ensure permissions on all resources are given only to the tenancy administrator group","level":1}, "1.3": {"id":"IAM-3","section":"Identity and Access Management","title":"Ensure IAM administrators cannot update tenancy Administrators group","level":1}, "1.4": {"id":"IAM-4","section":"Identity and Access Management","title":"Ensure IAM password policy requires minimum length of 14 or greater","level":1}, "1.5": {"id":"IAM-5","section":"Identity and Access Management","title":"Ensure IAM password policy expires passwords within 365 days","level":1}, "1.6": {"id":"IAM-6","section":"Identity and Access Management","title":"Ensure IAM password policy prevents password reuse","level":1}, "1.7": {"id":"IAM-7","section":"Identity and Access Management","title":"Ensure MFA is enabled for all users with a console password","level":1}, "1.8": {"id":"IAM-8","section":"Identity and Access Management","title":"Ensure user API keys rotate within 90 days","level":1}, "1.9": {"id":"IAM-9","section":"Identity and Access Management","title":"Ensure user customer secret keys rotate within 90 days","level":1}, "1.10": {"id":"IAM-10","section":"Identity and Access Management","title":"Ensure user auth tokens rotate within 90 days","level":1}, "1.11": {"id":"IAM-11","section":"Identity and Access Management","title":"Ensure user IAM Database Passwords rotate within 90 days","level":1}, "1.12": {"id":"IAM-12","section":"Identity and Access Management","title":"Ensure API keys are not created for tenancy administrator users","level":1}, "1.13": {"id":"IAM-13","section":"Identity and Access Management","title":"Ensure all OCI IAM user accounts have a valid and current email address","level":2}, "1.14": {"id":"IAM-14","section":"Identity and Access Management","title":"Ensure Instance Principal authentication is used","level":1}, "1.15": {"id":"IAM-15","section":"Identity and Access Management","title":"Ensure storage service-level admins cannot delete resources they manage","level":1}, "1.16": {"id":"IAM-16","section":"Identity and Access Management","title":"Ensure OCI IAM credentials unused for 45 days or more are disabled","level":2}, "1.17": {"id":"IAM-17","section":"Identity and Access Management","title":"Ensure there is only one active API Key per user","level":2}, "2.1": {"id":"NTW-1","section":"Networking","title":"Ensure no security lists allow ingress from 0.0.0.0/0 to port 22","level":1}, "2.2": {"id":"NTW-2","section":"Networking","title":"Ensure no security lists allow ingress from 0.0.0.0/0 to port 3389","level":1}, "2.3": {"id":"NTW-3","section":"Networking","title":"Ensure no network security groups allow ingress from 0.0.0.0/0 to port 22","level":1}, "2.4": {"id":"NTW-4","section":"Networking","title":"Ensure no network security groups allow ingress from 0.0.0.0/0 to port 3389","level":1}, "2.5": {"id":"NTW-5","section":"Networking","title":"Ensure the default security list of every VCN restricts all traffic except ICMP","level":1}, "2.6": {"id":"NTW-6","section":"Networking","title":"Ensure Oracle Integration Cloud (OIC) access is restricted","level":2}, "2.7": {"id":"NTW-7","section":"Networking","title":"Ensure Oracle Analytics Cloud (OAC) access is restricted or deployed within a VCN","level":2}, "2.8": {"id":"NTW-8","section":"Networking","title":"Ensure Oracle Autonomous Shared Database access is restricted or deployed within a VCN","level":2}, "3.1": {"id":"COM-1","section":"Compute","title":"Ensure Compute Instance Legacy Metadata service endpoint is disabled","level":2}, "3.2": {"id":"COM-2","section":"Compute","title":"Ensure Secure Boot is enabled on Compute Instance","level":2}, "3.3": {"id":"COM-3","section":"Compute","title":"Ensure In-transit Encryption is enabled on Compute Instance","level":1}, "4.1": {"id":"LAM-1","section":"Logging and Monitoring","title":"Ensure default tags are used on resources","level":1}, "4.2": {"id":"LAM-2","section":"Logging and Monitoring","title":"Create at least one notification topic and subscription","level":1}, "4.3": {"id":"LAM-3","section":"Logging and Monitoring","title":"Ensure a notification is configured for Identity Provider changes","level":1}, "4.4": {"id":"LAM-4","section":"Logging and Monitoring","title":"Ensure a notification is configured for IdP group mapping changes","level":1}, "4.5": {"id":"LAM-5","section":"Logging and Monitoring","title":"Ensure a notification is configured for IAM group changes","level":1}, "4.6": {"id":"LAM-6","section":"Logging and Monitoring","title":"Ensure a notification is configured for IAM policy changes","level":1}, "4.7": {"id":"LAM-7","section":"Logging and Monitoring","title":"Ensure a notification is configured for user changes","level":1}, "4.8": {"id":"LAM-8","section":"Logging and Monitoring","title":"Ensure a notification is configured for VCN changes","level":1}, "4.9": {"id":"LAM-9","section":"Logging and Monitoring","title":"Ensure a notification is configured for changes to route tables","level":1}, "4.10": {"id":"LAM-10","section":"Logging and Monitoring","title":"Ensure a notification is configured for security list changes","level":1}, "4.11": {"id":"LAM-11","section":"Logging and Monitoring","title":"Ensure a notification is configured for network security group changes","level":1}, "4.12": {"id":"LAM-12","section":"Logging and Monitoring","title":"Ensure a notification is configured for changes to network gateways","level":1}, "4.13": {"id":"LAM-13","section":"Logging and Monitoring","title":"Ensure VCN flow logging is enabled for all subnets","level":2}, "4.14": {"id":"LAM-14","section":"Logging and Monitoring","title":"Ensure Cloud Guard is enabled in the root compartment","level":1}, "4.15": {"id":"LAM-15","section":"Logging and Monitoring","title":"Ensure a notification is configured for Cloud Guard problems detected","level":2}, "4.16": {"id":"LAM-16","section":"Logging and Monitoring","title":"Ensure customer created CMK is rotated at least annually","level":1}, "4.17": {"id":"LAM-17","section":"Logging and Monitoring","title":"Ensure write level Object Storage logging is enabled for all buckets","level":2}, "4.18": {"id":"LAM-18","section":"Logging and Monitoring","title":"Ensure a notification is configured for Local OCI User Authentication","level":1}, "5.1.1":{"id":"STO-1-1","section":"Storage - Object Storage","title":"Ensure no Object Storage buckets are publicly visible","level":1}, "5.1.2":{"id":"STO-1-2","section":"Storage - Object Storage","title":"Ensure Object Storage Buckets are encrypted with a CMK","level":2}, "5.1.3":{"id":"STO-1-3","section":"Storage - Object Storage","title":"Ensure Versioning is Enabled for Object Storage Buckets","level":2}, "5.2.1":{"id":"STO-2-1","section":"Storage - Block Volumes","title":"Ensure Block Volumes are encrypted with Customer Managed Keys","level":2}, "5.2.2":{"id":"STO-2-2","section":"Storage - Block Volumes","title":"Ensure Boot Volumes are encrypted with Customer Managed Key","level":2}, "5.3.1":{"id":"STO-3-1","section":"Storage - File Storage Service","title":"Ensure File Storage Systems are encrypted with Customer Managed Keys","level":2}, "6.1": {"id":"AM-1","section":"Asset Management","title":"Create at least one compartment in your tenancy","level":1}, "6.2": {"id":"AM-2","section":"Asset Management","title":"Ensure no resources are created in the root compartment","level":1}, } # Event types required for monitoring checks 4.3-4.12, 4.15, 4.18 CIS_MONITORING_EVENTS = { "4.3": [ "com.oraclecloud.identitycontrolplane.createidentityprovider", "com.oraclecloud.identitycontrolplane.deleteidentityprovider", "com.oraclecloud.identitycontrolplane.updateidentityprovider", ], "4.4": [ "com.oraclecloud.identitycontrolplane.createidpgroupmapping", "com.oraclecloud.identitycontrolplane.deleteidpgroupmapping", "com.oraclecloud.identitycontrolplane.updateidpgroupmapping", ], "4.5": [ "com.oraclecloud.identitycontrolplane.creategroup", "com.oraclecloud.identitycontrolplane.deletegroup", "com.oraclecloud.identitycontrolplane.updategroup", ], "4.6": [ "com.oraclecloud.identitycontrolplane.createpolicy", "com.oraclecloud.identitycontrolplane.deletepolicy", "com.oraclecloud.identitycontrolplane.updatepolicy", ], "4.7": [ "com.oraclecloud.identitycontrolplane.createuser", "com.oraclecloud.identitycontrolplane.deleteuser", "com.oraclecloud.identitycontrolplane.updateuser", ], "4.8": [ "com.oraclecloud.virtualnetwork.createvcn", "com.oraclecloud.virtualnetwork.deletevcn", "com.oraclecloud.virtualnetwork.updatevcn", ], "4.9": [ "com.oraclecloud.virtualnetwork.changeroutetablecompartment", "com.oraclecloud.virtualnetwork.createroutetable", "com.oraclecloud.virtualnetwork.deleteroutetable", "com.oraclecloud.virtualnetwork.updateroutetable", ], "4.10": [ "com.oraclecloud.virtualnetwork.changesecuritylistcompartment", "com.oraclecloud.virtualnetwork.createsecuritylist", "com.oraclecloud.virtualnetwork.deletesecuritylist", "com.oraclecloud.virtualnetwork.updatesecuritylist", ], "4.11": [ "com.oraclecloud.virtualnetwork.changenetworksecuritygroupcompartment", "com.oraclecloud.virtualnetwork.createnetworksecuritygroup", "com.oraclecloud.virtualnetwork.deletenetworksecuritygroup", "com.oraclecloud.virtualnetwork.updatenetworksecuritygroup", ], "4.12": [ "com.oraclecloud.virtualnetwork.createdrg", "com.oraclecloud.virtualnetwork.deletedrg", "com.oraclecloud.virtualnetwork.updatedrg", "com.oraclecloud.virtualnetwork.createinternetgateway", "com.oraclecloud.virtualnetwork.deleteinternetgateway", "com.oraclecloud.virtualnetwork.updateinternetgateway", "com.oraclecloud.virtualnetwork.createlocalpeering gateway", "com.oraclecloud.virtualnetwork.deletelocalpeering gateway", "com.oraclecloud.virtualnetwork.updatelocalpeering gateway", ], "4.15": [ "com.oraclecloud.cloudguard.problemdetected", ], "4.18": [ "com.oraclecloud.identitycontrolplane.createpolicy", "com.oraclecloud.identitycontrolplane.deletepolicy", "com.oraclecloud.identitycontrolplane.updatepolicy", "com.oraclecloud.identitycontrolplane.createuser", "com.oraclecloud.identitycontrolplane.deleteuser", "com.oraclecloud.identitycontrolplane.updateuser", "com.oraclecloud.identitycontrolplane.updateauthenticationpolicy", ], } # ─── CIS Compliance Checker ────────────────────────────────────────────────── class CISComplianceChecker: def __init__(self, config_path: str, filter_regions: list[str] | None = None, level: int | None = None): self.config_path = config_path self.filter_regions = [r.strip() for r in filter_regions] if filter_regions else None self.level = level # None = all levels self.config = oci.config.from_file(config_path, "DEFAULT") oci.config.validate_config(self.config) self.tenancy_id = self.config["tenancy"] self.home_region = None self.regions = [] self.compartments = [] # Collected data stores self._policies = [] self._users = [] self._groups = [] self._admin_group_id = None self._admin_user_ids = set() self._auth_policy = None self._dynamic_groups = [] self._tag_defaults = [] self._user_api_keys = {} self._user_secret_keys = {} self._user_auth_tokens = {} self._user_db_credentials = {} self._user_mfa = {} self._user_memberships = {} # Regional data (keyed by region) self._security_lists = [] self._nsgs = [] self._nsg_rules = {} self._vcns = [] self._subnets = [] self._instances = [] self._event_rules = [] self._topics = [] self._subscriptions = [] self._log_groups = [] self._logs = [] self._cloud_guard_status = None self._vaults = [] self._keys = [] self._buckets = [] self._block_volumes = [] self._boot_volumes = [] self._file_systems = [] self._oic_instances = [] self._oac_instances = [] self._adbs = [] self._root_resources_count = 0 # Results self.findings = {} for cid, ck in CHECKS.items(): self.findings[cid] = {**ck, "status": "REVIEW", "findings": [], "total": []} # ── Helpers ──────────────────────────────────────────────────────────────── def _safe(self, fn, *args, default=None, **kwargs): """Call OCI SDK function with error handling and retry on 429.""" for attempt in range(3): try: return fn(*args, **kwargs) except oci.exceptions.ServiceError as e: if e.status == 429: wait = 2 ** attempt log.warning(f"Rate limited, retrying in {wait}s...") time.sleep(wait) continue if e.status in (401, 403): log.warning(f"Auth/permission error ({e.status}): {e.message[:200]}") return default if e.status == 404: return default log.warning(f"OCI API error {e.status}: {e.message[:200]}") return default except oci.exceptions.ClientError as e: log.warning(f"Client error: {e}") return default except Exception as e: log.warning(f"Unexpected error: {e}") return default return default def _paginate(self, fn, **kwargs): """Paginate through all results of an OCI list operation.""" results = [] resp = self._safe(fn, **kwargs) if resp is None: return results results.extend(resp.data) while resp.has_next_page: resp = self._safe(fn, page=resp.next_page, **kwargs) if resp is None: break results.extend(resp.data) return results def _make_config(self, region: str) -> dict: """Return OCI config dict targeting a specific region.""" cfg = dict(self.config) cfg["region"] = region return cfg def _set_status(self, cid: str, passed: bool): """Set check status to PASS or FAIL.""" self.findings[cid]["status"] = "PASS" if passed else "FAIL" def _add_finding(self, cid: str, desc: str): """Add a non-compliant resource finding.""" self.findings[cid]["findings"].append(desc) def _add_total(self, cid: str, desc: str): """Add an evaluated resource.""" self.findings[cid]["total"].append(desc) def _should_skip(self, cid: str) -> bool: """Check if a check should be skipped based on level filter.""" if self.level is None: return False return CHECKS[cid]["level"] > self.level # ── Region Discovery ────────────────────────────────────────────────────── def _discover_regions(self): identity = oci.identity.IdentityClient(self.config) tenancy = self._safe(identity.get_tenancy, self.tenancy_id) if tenancy: self.home_region_key = tenancy.data.home_region_key subs = self._safe(identity.list_region_subscriptions, self.tenancy_id) if not subs: log.error("Cannot discover regions — using config region only") self.regions = [self.config["region"]] self.home_region = self.config["region"] return for r in subs.data: if r.is_home_region: self.home_region = r.region_name if self.filter_regions: if r.region_name in self.filter_regions: self.regions.append(r.region_name) else: if r.status == "READY": self.regions.append(r.region_name) if not self.home_region: self.home_region = self.config["region"] if not self.regions: self.regions = [self.home_region] log.info(f"Home region: {self.home_region}") log.info(f"Regions to scan: {', '.join(self.regions)}") # ── Compartment Discovery ───────────────────────────────────────────────── def _discover_compartments(self): identity = oci.identity.IdentityClient(self._make_config(self.home_region)) comps = self._paginate(identity.list_compartments, compartment_id=self.tenancy_id, compartment_id_in_subtree=True, access_level="ACCESSIBLE", lifecycle_state="ACTIVE") self.compartments = comps # Include root compartment root = type('obj', (object,), {'id': self.tenancy_id, 'name': 'root', 'lifecycle_state': 'ACTIVE'})() self.compartments.insert(0, root) log.info(f"Discovered {len(self.compartments)} compartments") def _comp_ids(self): return [c.id for c in self.compartments] # ── IAM Data Collection ─────────────────────────────────────────────────── def _collect_iam(self): log.info("Collecting IAM data...") identity = oci.identity.IdentityClient(self._make_config(self.home_region)) # Policies across all compartments for cid in self._comp_ids(): pols = self._paginate(identity.list_policies, compartment_id=cid) self._policies.extend(pols) # Users self._users = self._paginate(identity.list_users, compartment_id=self.tenancy_id) # Groups self._groups = self._paginate(identity.list_groups, compartment_id=self.tenancy_id) # Find Administrators group for g in self._groups: if g.name == "Administrators": self._admin_group_id = g.id break # Group memberships and user credentials now = datetime.datetime.now(datetime.timezone.utc) for u in self._users: uid = u.id self._user_api_keys[uid] = self._safe(identity.list_api_keys, uid, default=type('R', (), {'data': []})()).data self._user_secret_keys[uid] = self._safe(identity.list_customer_secret_keys, uid, default=type('R', (), {'data': []})()).data self._user_auth_tokens[uid] = self._safe(identity.list_auth_tokens, uid, default=type('R', (), {'data': []})()).data try: self._user_db_credentials[uid] = self._safe(identity.list_db_credentials, uid, default=type('R', (), {'data': []})()).data except Exception: self._user_db_credentials[uid] = [] mfa_resp = self._safe(identity.list_mfa_totp_devices, uid, default=type('R', (), {'data': []})()) self._user_mfa[uid] = mfa_resp.data if mfa_resp else [] memberships = self._safe(identity.list_user_group_memberships, compartment_id=self.tenancy_id, user_id=uid, default=type('R', (), {'data': []})()) self._user_memberships[uid] = memberships.data if memberships else [] # Determine admin users if self._admin_group_id: for uid, memberships in self._user_memberships.items(): for m in memberships: if m.group_id == self._admin_group_id: self._admin_user_ids.add(uid) # Password policy auth_policy = self._safe(identity.get_authentication_policy, self.tenancy_id) self._auth_policy = auth_policy.data if auth_policy else None # Dynamic groups self._dynamic_groups = self._paginate(identity.list_dynamic_groups, compartment_id=self.tenancy_id) # Tag defaults self._tag_defaults = self._paginate(identity.list_tag_defaults, compartment_id=self.tenancy_id) log.info(f"IAM: {len(self._policies)} policies, {len(self._users)} users, {len(self._groups)} groups") # ── Network Data Collection (per region) ────────────────────────────────── def _collect_network(self, region: str): log.info(f"Collecting network data for {region}...") cfg = self._make_config(region) vn = oci.core.VirtualNetworkClient(cfg) sls, nsgs, vcns, subnets = [], [], [], [] for cid in self._comp_ids(): vcns.extend(self._paginate(vn.list_vcns, compartment_id=cid)) sls.extend(self._paginate(vn.list_security_lists, compartment_id=cid)) nsgs.extend(self._paginate(vn.list_network_security_groups, compartment_id=cid)) subnets.extend(self._paginate(vn.list_subnets, compartment_id=cid)) nsg_rules = {} for nsg in nsgs: rules = self._paginate(vn.list_network_security_group_security_rules, network_security_group_id=nsg.id) nsg_rules[nsg.id] = rules # OIC, OAC, ADB for checks 2.6-2.8 oic_instances, oac_instances, adbs = [], [], [] try: oic_client = oci.integration.IntegrationInstanceClient(cfg) for cid in self._comp_ids(): oic_instances.extend(self._paginate(oic_client.list_integration_instances, compartment_id=cid)) except Exception as e: log.warning(f"OIC collection skipped: {e}") try: oac_client = oci.analytics.AnalyticsClient(cfg) for cid in self._comp_ids(): oac_instances.extend(self._paginate(oac_client.list_analytics_instances, compartment_id=cid)) except Exception as e: log.warning(f"OAC collection skipped: {e}") try: db_client = oci.database.DatabaseClient(cfg) for cid in self._comp_ids(): adbs.extend(self._paginate(db_client.list_autonomous_databases, compartment_id=cid)) except Exception as e: log.warning(f"ADB collection skipped: {e}") return { "security_lists": sls, "nsgs": nsgs, "nsg_rules": nsg_rules, "vcns": vcns, "subnets": subnets, "oic": oic_instances, "oac": oac_instances, "adbs": adbs, } # ── Compute Data Collection (per region) ────────────────────────────────── def _collect_compute(self, region: str): log.info(f"Collecting compute data for {region}...") cfg = self._make_config(region) compute = oci.core.ComputeClient(cfg) instances = [] for cid in self._comp_ids(): insts = self._paginate(compute.list_instances, compartment_id=cid) instances.extend([i for i in insts if i.lifecycle_state == "RUNNING"]) return instances # ── Monitoring Data Collection (per region) ─────────────────────────────── def _collect_monitoring(self, region: str): log.info(f"Collecting monitoring data for {region}...") cfg = self._make_config(region) events_client = oci.events.EventsClient(cfg) ons_cp = oci.ons.NotificationControlPlaneClient(cfg) ons_dp = oci.ons.NotificationDataPlaneClient(cfg) logging_client = oci.logging.LoggingManagementClient(cfg) rules, topics, subscriptions, log_groups_all, logs_all = [], [], [], [], [] for cid in self._comp_ids(): rules.extend(self._paginate(events_client.list_rules, compartment_id=cid)) topics.extend(self._paginate(ons_cp.list_topics, compartment_id=cid)) subscriptions.extend(self._paginate(ons_dp.list_subscriptions, compartment_id=cid)) lgs = self._paginate(logging_client.list_log_groups, compartment_id=cid) log_groups_all.extend(lgs) for lg in lgs: log_list = self._paginate(logging_client.list_logs, log_group_id=lg.id) logs_all.extend(log_list) # Vaults and keys vaults, keys = [], [] try: kms_vault_client = oci.key_management.KmsVaultClient(cfg) for cid in self._comp_ids(): vs = self._paginate(kms_vault_client.list_vaults, compartment_id=cid) for v in vs: if v.lifecycle_state != "ACTIVE": continue vaults.append(v) try: kms_mgmt = oci.key_management.KmsManagementClient(cfg, service_endpoint=v.management_endpoint) ks = self._paginate(kms_mgmt.list_keys, compartment_id=cid) for k in ks: if k.lifecycle_state == "ENABLED": key_detail = self._safe(kms_mgmt.get_key, k.id) if key_detail: keys.append(key_detail.data) except Exception as e: log.warning(f"Key collection for vault {v.id}: {e}") except Exception as e: log.warning(f"Vault collection skipped: {e}") return { "event_rules": rules, "topics": topics, "subscriptions": subscriptions, "log_groups": log_groups_all, "logs": logs_all, "vaults": vaults, "keys": keys, } # ── Storage Data Collection (per region) ────────────────────────────────── def _collect_storage(self, region: str): log.info(f"Collecting storage data for {region}...") cfg = self._make_config(region) os_client = oci.object_storage.ObjectStorageClient(cfg) block_client = oci.core.BlockstorageClient(cfg) buckets, block_volumes, boot_volumes, file_systems = [], [], [], [] ns = self._safe(os_client.get_namespace) namespace = ns.data if ns else None if namespace: for cid in self._comp_ids(): bucket_list = self._paginate(os_client.list_buckets, namespace_name=namespace, compartment_id=cid) for b in bucket_list: detail = self._safe(os_client.get_bucket, namespace_name=namespace, bucket_name=b.name, fields=["approximateCount", "approximateSize", "autoTiering"]) if detail: buckets.append(detail.data) for cid in self._comp_ids(): block_volumes.extend(self._paginate(block_client.list_volumes, compartment_id=cid)) # Boot volumes and file systems need availability domains identity = oci.identity.IdentityClient(cfg) ads = self._safe(identity.list_availability_domains, compartment_id=self.tenancy_id) ad_names = [ad.name for ad in (ads.data if ads else [])] for cid in self._comp_ids(): for ad in ad_names: boot_volumes.extend(self._paginate(block_client.list_boot_volumes, compartment_id=cid, availability_domain=ad)) try: fs_client = oci.file_storage.FileStorageClient(cfg) for cid in self._comp_ids(): for ad in ad_names: file_systems.extend(self._paginate(fs_client.list_file_systems, compartment_id=cid, availability_domain=ad)) except Exception as e: log.warning(f"File storage collection skipped: {e}") return { "buckets": buckets, "block_volumes": block_volumes, "boot_volumes": boot_volumes, "file_systems": file_systems, } # ── Cloud Guard (home region only) ──────────────────────────────────────── def _collect_cloud_guard(self): log.info("Collecting Cloud Guard status...") cfg = self._make_config(self.home_region) try: cg = oci.cloud_guard.CloudGuardClient(cfg) resp = self._safe(cg.get_configuration, compartment_id=self.tenancy_id) self._cloud_guard_status = resp.data.status if resp else None except Exception as e: log.warning(f"Cloud Guard collection skipped: {e}") # ── Asset / Resource Search (home region) ───────────────────────────────── def _collect_assets(self): log.info("Collecting asset data...") cfg = self._make_config(self.home_region) try: search_client = oci.resource_search.ResourceSearchClient(cfg) query = f"query all resources where compartmentId = '{self.tenancy_id}'" details = oci.resource_search.models.StructuredSearchDetails(query=query, type="Structured") resp = self._safe(search_client.search_resources, details) if resp: # Exclude resource types that normally live in root root_only_types = {"Compartment", "TagNamespace", "TagDefault", "Policy", "Group", "DynamicGroup", "User", "IdentityProvider", "NetworkSource", "AuthenticationPolicy", "Tenancy"} non_root = [r for r in resp.data.items if r.resource_type not in root_only_types] self._root_resources_count = len(non_root) except Exception as e: log.warning(f"Resource search skipped: {e}") # ═══════════════════════════════════════════════════════════════════════════ # CIS CHECKS # ═══════════════════════════════════════════════════════════════════════════ # ── IAM Policy Checks: 1.1, 1.2, 1.3, 1.15 ────────────────────────────── def _check_iam_policies(self): log.info("Running IAM policy checks (1.1, 1.2, 1.3, 1.15)...") admin_group_name = "Administrators" all_statements = [] for p in self._policies: if p.lifecycle_state != "ACTIVE": continue for stmt in (p.statements or []): all_statements.append(stmt.lower().strip()) # 1.1 — Service-level admins exist if not self._should_skip("1.1"): service_keywords = ["virtual-network-family", "object-family", "instance-family", "volume-family", "database-family", "file-family", "cluster-family"] has_service_admin = False for stmt in all_statements: if "manage" in stmt or "use" in stmt: for kw in service_keywords: if kw in stmt and "administrators" not in stmt.split("group")[1] if "group" in stmt else True: has_service_admin = True break self._set_status("1.1", has_service_admin) if not has_service_admin: self._add_finding("1.1", "No service-level admin groups found — only tenancy-wide Administrators") # 1.2 — Only Administrators has manage all-resources if not self._should_skip("1.2"): violating = [] pattern = re.compile(r"allow\s+group\s+(\S+)\s+to\s+manage\s+all-resources\s+in\s+tenancy") for stmt in all_statements: m = pattern.search(stmt) if m: group_name = m.group(1).strip("'\"") if group_name.lower() != admin_group_name.lower(): violating.append(group_name) self._set_status("1.2", len(violating) == 0) for g in violating: self._add_finding("1.2", f"Group '{g}' has 'manage all-resources in tenancy'") # 1.3 — IAM admins cannot update Administrators group if not self._should_skip("1.3"): can_modify_admins = False for stmt in all_statements: if ("manage" in stmt and ("users" in stmt or "groups" in stmt) and "tenancy" in stmt and "administrators" not in stmt.split("group")[0] if "group" in stmt else True): if "where" not in stmt: can_modify_admins = True self._set_status("1.3", not can_modify_admins) if can_modify_admins: self._add_finding("1.3", "Non-admin groups may modify Administrators group (no where clause restriction)") # 1.15 — Storage admins cannot delete resources if not self._should_skip("1.15"): can_delete = False storage_resources = ["object-family", "volume-family", "file-family", "buckets", "objects", "volumes", "boot-volumes", "file-systems"] for stmt in all_statements: if "manage" in stmt: for sr in storage_resources: if sr in stmt and "where" not in stmt: can_delete = True self._add_finding("1.15", f"Policy allows unrestricted manage on '{sr}' without delete restriction") self._set_status("1.15", not can_delete) # ── Password Policy Checks: 1.4, 1.5, 1.6 ─────────────────────────────── def _check_password_policies(self): log.info("Running password policy checks (1.4, 1.5, 1.6)...") pp = self._auth_policy.password_policy if self._auth_policy else None if not pp: for cid in ["1.4", "1.5", "1.6"]: if not self._should_skip(cid): self.findings[cid]["status"] = "REVIEW" self._add_finding(cid, "Could not retrieve authentication policy") return # 1.4 — Min length >= 14 if not self._should_skip("1.4"): min_len = getattr(pp, "minimum_password_length", None) if min_len is None: min_len = getattr(pp, "minimum_password_length_in_characters", 0) passed = min_len is not None and min_len >= 14 self._set_status("1.4", passed) self._add_total("1.4", f"Minimum password length: {min_len}") if not passed: self._add_finding("1.4", f"Password minimum length is {min_len}, should be >= 14") # 1.5 — Passwords expire if not self._should_skip("1.5"): # OCI uses is_password_expires_in_applicable or password_expire_warning_in_days pass_expires = getattr(pp, "is_password_expires_in_applicable", None) if pass_expires is None: # Check for numeric expiry field pass_expires = getattr(pp, "password_expiry_in_days", None) is not None self._set_status("1.5", bool(pass_expires)) if not pass_expires: self._add_finding("1.5", "Password expiration is not enabled") # 1.6 — Prevent reuse if not self._should_skip("1.6"): num_previous = getattr(pp, "num_previous_passwords_to_restrict", 0) or 0 passed = num_previous > 0 self._set_status("1.6", passed) if not passed: self._add_finding("1.6", "Password reuse prevention is not configured") # ── User Checks: 1.7-1.13, 1.16, 1.17 ─────────────────────────────────── def _check_users(self): log.info("Running user checks (1.7-1.13, 1.16, 1.17)...") now = datetime.datetime.now(datetime.timezone.utc) rotation_days = 90 inactive_days = 45 for u in self._users: uid = u.id name = u.name active = u.lifecycle_state == "ACTIVE" self._add_total("1.7", name) # 1.7 — MFA for console users if not self._should_skip("1.7") and active: has_console = getattr(u, "can_use_console_password", None) if has_console is None: has_console = getattr(u, 'capabilities', None) if has_console: has_console = getattr(has_console, 'can_use_console_password', False) is_mfa = getattr(u, "is_mfa_activated", False) if has_console and not is_mfa: self._add_finding("1.7", f"User '{name}' has console password but MFA not activated") # 1.8 — API key rotation if not self._should_skip("1.8") and active: for key in self._user_api_keys.get(uid, []): if getattr(key, 'lifecycle_state', 'ACTIVE') != 'ACTIVE': continue self._add_total("1.8", f"{name}/{key.fingerprint}") created = key.time_created if created and (now - created).days > rotation_days: self._add_finding("1.8", f"User '{name}' API key {key.fingerprint} is {(now - created).days} days old") # 1.9 — Secret key rotation if not self._should_skip("1.9") and active: for key in self._user_secret_keys.get(uid, []): if getattr(key, 'lifecycle_state', 'ACTIVE') != 'ACTIVE': continue self._add_total("1.9", f"{name}/{key.id}") created = key.time_created if created and (now - created).days > rotation_days: self._add_finding("1.9", f"User '{name}' secret key is {(now - created).days} days old") # 1.10 — Auth token rotation if not self._should_skip("1.10") and active: for tok in self._user_auth_tokens.get(uid, []): if getattr(tok, 'lifecycle_state', 'ACTIVE') != 'ACTIVE': continue self._add_total("1.10", f"{name}/{tok.id}") created = tok.time_created if created and (now - created).days > rotation_days: self._add_finding("1.10", f"User '{name}' auth token is {(now - created).days} days old") # 1.11 — DB credentials rotation if not self._should_skip("1.11") and active: for cred in self._user_db_credentials.get(uid, []): if getattr(cred, 'lifecycle_state', 'ACTIVE') != 'ACTIVE': continue self._add_total("1.11", f"{name}/{cred.id}") created = cred.time_created if created and (now - created).days > rotation_days: self._add_finding("1.11", f"User '{name}' DB credential is {(now - created).days} days old") # 1.12 — No API keys for admin users if not self._should_skip("1.12") and uid in self._admin_user_ids: active_keys = [k for k in self._user_api_keys.get(uid, []) if getattr(k, 'lifecycle_state', 'ACTIVE') == 'ACTIVE'] self._add_total("1.12", name) if active_keys: self._add_finding("1.12", f"Admin user '{name}' has {len(active_keys)} active API key(s)") # 1.13 — Valid email if not self._should_skip("1.13") and active: email = getattr(u, "email", None) self._add_total("1.13", name) if not email or "@" not in email: self._add_finding("1.13", f"User '{name}' has no valid email address") # 1.16 — Inactive credentials disabled if not self._should_skip("1.16") and active: last_login = getattr(u, "last_successful_login_time", None) ref_time = last_login or u.time_created if ref_time and (now - ref_time).days > inactive_days: self._add_total("1.16", name) self._add_finding("1.16", f"User '{name}' inactive for {(now - ref_time).days} days but still ACTIVE") # 1.17 — Max 1 active API key if not self._should_skip("1.17") and active: active_keys = [k for k in self._user_api_keys.get(uid, []) if getattr(k, 'lifecycle_state', 'ACTIVE') == 'ACTIVE'] self._add_total("1.17", name) if len(active_keys) > 1: self._add_finding("1.17", f"User '{name}' has {len(active_keys)} active API keys") # Set statuses for cid in ["1.7", "1.8", "1.9", "1.10", "1.11", "1.12", "1.13", "1.16", "1.17"]: if not self._should_skip(cid): self._set_status(cid, len(self.findings[cid]["findings"]) == 0) # ── Dynamic Groups Check: 1.14 ─────────────────────────────────────────── def _check_dynamic_groups(self): log.info("Running dynamic groups check (1.14)...") if self._should_skip("1.14"): return has_instance_principal = False for dg in self._dynamic_groups: if dg.lifecycle_state != "ACTIVE": continue rules = getattr(dg, "matching_rule", "") or "" if "instance.compartment.id" in rules.lower() or "instance.id" in rules.lower(): has_instance_principal = True break self._set_status("1.14", has_instance_principal) self._add_total("1.14", f"{len(self._dynamic_groups)} dynamic groups found") if not has_instance_principal: self._add_finding("1.14", "No dynamic group found using Instance Principal matching rules") # ── Network Checks: 2.1-2.8 ────────────────────────────────────────────── def _check_network(self): log.info("Running network checks (2.1-2.8)...") def _port_in_range(port_range, port): if port_range is None: return True # All ports return port_range.min <= port <= port_range.max def _is_open_source(source): return source in ("0.0.0.0/0", "::/0") # 2.1, 2.2 — Security Lists for sl in self._security_lists: sl_name = f"{sl.display_name} ({sl.id})" for rule in (sl.ingress_security_rules or []): if not _is_open_source(getattr(rule, "source", "")): continue proto = str(getattr(rule, "protocol", "")) if proto == "6": # TCP tcp_opts = getattr(rule, "tcp_options", None) dst_range = getattr(tcp_opts, "destination_port_range", None) if tcp_opts else None if not self._should_skip("2.1"): self._add_total("2.1", sl_name) if _port_in_range(dst_range, 22): self._add_finding("2.1", f"SL '{sl.display_name}' allows 0.0.0.0/0 to port 22") if not self._should_skip("2.2"): self._add_total("2.2", sl_name) if _port_in_range(dst_range, 3389): self._add_finding("2.2", f"SL '{sl.display_name}' allows 0.0.0.0/0 to port 3389") elif proto == "all": if not self._should_skip("2.1"): self._add_finding("2.1", f"SL '{sl.display_name}' allows ALL protocols from 0.0.0.0/0") if not self._should_skip("2.2"): self._add_finding("2.2", f"SL '{sl.display_name}' allows ALL protocols from 0.0.0.0/0") # 2.3, 2.4 — NSG rules for nsg in self._nsgs: nsg_name = f"{nsg.display_name} ({nsg.id})" for rule in self._nsg_rules.get(nsg.id, []): if getattr(rule, "direction", "") != "INGRESS": continue if not _is_open_source(getattr(rule, "source", "")): continue proto = str(getattr(rule, "protocol", "")) if proto == "6": tcp_opts = getattr(rule, "tcp_options", None) dst_range = getattr(tcp_opts, "destination_port_range", None) if tcp_opts else None if not self._should_skip("2.3"): self._add_total("2.3", nsg_name) if _port_in_range(dst_range, 22): self._add_finding("2.3", f"NSG '{nsg.display_name}' allows 0.0.0.0/0 to port 22") if not self._should_skip("2.4"): self._add_total("2.4", nsg_name) if _port_in_range(dst_range, 3389): self._add_finding("2.4", f"NSG '{nsg.display_name}' allows 0.0.0.0/0 to port 3389") elif proto == "all": if not self._should_skip("2.3"): self._add_finding("2.3", f"NSG '{nsg.display_name}' allows ALL protocols from 0.0.0.0/0") if not self._should_skip("2.4"): self._add_finding("2.4", f"NSG '{nsg.display_name}' allows ALL protocols from 0.0.0.0/0") # 2.5 — Default SL restricts all except ICMP if not self._should_skip("2.5"): for vcn in self._vcns: default_sl_id = getattr(vcn, "default_security_list_id", None) if not default_sl_id: continue default_sl = next((sl for sl in self._security_lists if sl.id == default_sl_id), None) if not default_sl: continue self._add_total("2.5", f"{vcn.display_name} default SL") for rule in (default_sl.ingress_security_rules or []): source = getattr(rule, "source", "") proto = str(getattr(rule, "protocol", "")) if _is_open_source(source) and proto != "1": # 1 = ICMP self._add_finding("2.5", f"VCN '{vcn.display_name}' default SL allows non-ICMP from 0.0.0.0/0 (proto={proto})") # Set statuses for 2.1-2.5 for cid in ["2.1", "2.2", "2.3", "2.4", "2.5"]: if not self._should_skip(cid): self._set_status(cid, len(self.findings[cid]["findings"]) == 0) # 2.6 — OIC access restricted if not self._should_skip("2.6"): for oic in self._oic_instances: self._add_total("2.6", oic.display_name) net_ep = getattr(oic, "network_endpoint_details", None) if net_ep: ep_type = getattr(net_ep, "network_endpoint_type", "PUBLIC") if ep_type == "PUBLIC": allowlist = getattr(net_ep, "allowlisted_http_ips", None) if not allowlist: self._add_finding("2.6", f"OIC '{oic.display_name}' has public access without IP allowlist") self._set_status("2.6", len(self.findings["2.6"]["findings"]) == 0) # 2.7 — OAC access restricted if not self._should_skip("2.7"): for oac in self._oac_instances: self._add_total("2.7", oac.name) net_ep = getattr(oac, "network_endpoint_details", None) if net_ep: ep_type = getattr(net_ep, "network_endpoint_type", "PUBLIC") if ep_type == "PUBLIC": allowlist = getattr(net_ep, "whitelisted_ips", None) or getattr(net_ep, "allowlisted_ips", None) if not allowlist: self._add_finding("2.7", f"OAC '{oac.name}' has public access without restrictions") self._set_status("2.7", len(self.findings["2.7"]["findings"]) == 0) # 2.8 — ADB access restricted if not self._should_skip("2.8"): for adb in self._adbs: if adb.lifecycle_state not in ("AVAILABLE", "PROVISIONING"): continue self._add_total("2.8", adb.display_name) has_acl = getattr(adb, "is_access_control_enabled", False) has_subnet = getattr(adb, "subnet_id", None) wl_ips = getattr(adb, "whitelisted_ips", None) or [] if not has_subnet and not has_acl and not wl_ips: self._add_finding("2.8", f"ADB '{adb.display_name}' has unrestricted public access") self._set_status("2.8", len(self.findings["2.8"]["findings"]) == 0) # ── Compute Checks: 3.1-3.3 ────────────────────────────────────────────── def _check_compute(self): log.info("Running compute checks (3.1-3.3)...") for inst in self._instances: name = f"{inst.display_name} ({inst.id[-12:]})" # 3.1 — Legacy IMDS disabled if not self._should_skip("3.1"): self._add_total("3.1", name) inst_opts = getattr(inst, "instance_options", None) legacy_disabled = getattr(inst_opts, "are_legacy_imds_endpoints_disabled", False) if inst_opts else False if not legacy_disabled: self._add_finding("3.1", f"Instance '{inst.display_name}' has legacy IMDS enabled") # 3.2 — Secure Boot if not self._should_skip("3.2"): self._add_total("3.2", name) pc = getattr(inst, "platform_config", None) secure_boot = getattr(pc, "is_secure_boot_enabled", False) if pc else False if not secure_boot: self._add_finding("3.2", f"Instance '{inst.display_name}' does not have Secure Boot enabled") # 3.3 — In-transit encryption if not self._should_skip("3.3"): self._add_total("3.3", name) lo = getattr(inst, "launch_options", None) in_transit = getattr(lo, "is_pv_encryption_in_transit_enabled", False) if lo else False if not in_transit: self._add_finding("3.3", f"Instance '{inst.display_name}' does not have in-transit encryption") for cid in ["3.1", "3.2", "3.3"]: if not self._should_skip(cid): self._set_status(cid, len(self.findings[cid]["findings"]) == 0) # ── Logging & Monitoring Checks: 4.1-4.18 ──────────────────────────────── def _check_monitoring(self): log.info("Running monitoring checks (4.1-4.18)...") # 4.1 — Default tags if not self._should_skip("4.1"): self._add_total("4.1", f"{len(self._tag_defaults)} tag defaults found") self._set_status("4.1", len(self._tag_defaults) > 0) if not self._tag_defaults: self._add_finding("4.1", "No default tags configured in the tenancy") # 4.2 — At least one topic with subscription if not self._should_skip("4.2"): active_topics = [t for t in self._topics if getattr(t, "lifecycle_state", "") == "ACTIVE"] active_subs = [s for s in self._subscriptions if getattr(s, "lifecycle_state", "") == "ACTIVE"] has_topic_with_sub = False topic_ids_with_subs = {s.topic_id for s in active_subs} for t in active_topics: if t.topic_id in topic_ids_with_subs: has_topic_with_sub = True break self._add_total("4.2", f"{len(active_topics)} topics, {len(active_subs)} subscriptions") self._set_status("4.2", has_topic_with_sub) if not has_topic_with_sub: self._add_finding("4.2", "No notification topic with an active subscription found") # 4.3-4.12, 4.15, 4.18 — Event rule notifications active_rules = [r for r in self._event_rules if getattr(r, "lifecycle_state", "") == "ACTIVE" and getattr(r, "is_enabled", False)] for check_id, required_events in CIS_MONITORING_EVENTS.items(): if self._should_skip(check_id): continue found = False for rule in active_rules: condition = getattr(rule, "condition", "") or "" try: cond_json = json.loads(condition) event_types = [] if isinstance(cond_json, dict): event_types = cond_json.get("eventType", []) if not event_types: # Handle nested conditions data = cond_json.get("data", {}) if isinstance(data, dict): event_types = data.get("eventType", []) except (json.JSONDecodeError, TypeError): continue event_types_lower = [e.lower() for e in event_types] all_matched = all(req.lower() in event_types_lower for req in required_events) if all_matched: # Verify rule has ONS action actions = getattr(rule, "actions", None) if actions: action_list = getattr(actions, "actions", []) for a in action_list: if getattr(a, "action_type", "") in ("ONS", "FAAS", "OSS"): found = True break if found: break self._add_total(check_id, f"Checked {len(active_rules)} active event rules") self._set_status(check_id, found) if not found: self._add_finding(check_id, f"No event rule found covering required events for {check_id}") # 4.13 — VCN flow logging for all subnets if not self._should_skip("4.13"): logged_resources = set() for lg in self._logs: config = getattr(lg, "configuration", None) if not config: continue source = getattr(config, "source", None) if not source: continue service = getattr(source, "service", "") resource = getattr(source, "resource", "") if service == "flowlogs" and resource: logged_resources.add(resource) for subnet in self._subnets: self._add_total("4.13", f"{subnet.display_name}") if subnet.id not in logged_resources: self._add_finding("4.13", f"Subnet '{subnet.display_name}' ({subnet.id[-12:]}) has no VCN flow logging") self._set_status("4.13", len(self.findings["4.13"]["findings"]) == 0) # 4.14 — Cloud Guard enabled if not self._should_skip("4.14"): enabled = self._cloud_guard_status == "ENABLED" self._add_total("4.14", f"Cloud Guard status: {self._cloud_guard_status or 'UNKNOWN'}") self._set_status("4.14", enabled) if not enabled: self._add_finding("4.14", f"Cloud Guard is not enabled (status: {self._cloud_guard_status or 'UNKNOWN'})") # 4.16 — CMK rotation (365 days) if not self._should_skip("4.16"): now = datetime.datetime.now(datetime.timezone.utc) for key in self._keys: key_name = getattr(key, "display_name", key.id) self._add_total("4.16", key_name) # Check key versions for rotation current_version = getattr(key, "current_key_version", None) time_created = getattr(key, "time_created", None) # Get key version time kv_time = None key_versions = getattr(key, "key_versions", None) if current_version: # Use current key version's time if available kv_time = getattr(current_version, "time_created", None) if hasattr(current_version, "time_created") else None if kv_time is None: kv_time = time_created if kv_time and (now - kv_time).days > 365: self._add_finding("4.16", f"Key '{key_name}' not rotated in {(now - kv_time).days} days") self._set_status("4.16", len(self.findings["4.16"]["findings"]) == 0) # 4.17 — Write level Object Storage logging if not self._should_skip("4.17"): logged_buckets = set() for lg in self._logs: config = getattr(lg, "configuration", None) if not config: continue source = getattr(config, "source", None) if not source: continue service = getattr(source, "service", "") resource = getattr(source, "resource", "") category = getattr(source, "category", "") if service == "objectstorage" and category == "write": logged_buckets.add(resource) for b in self._buckets: self._add_total("4.17", b.name) if b.name not in logged_buckets: self._add_finding("4.17", f"Bucket '{b.name}' has no write-level logging enabled") self._set_status("4.17", len(self.findings["4.17"]["findings"]) == 0) # ── Storage Checks: 5.1.1-5.3.1 ────────────────────────────────────────── def _check_storage(self): log.info("Running storage checks (5.x)...") # 5.1.1 — No public buckets if not self._should_skip("5.1.1"): for b in self._buckets: self._add_total("5.1.1", b.name) pa = getattr(b, "public_access_type", "NoPublicAccess") if pa != "NoPublicAccess": self._add_finding("5.1.1", f"Bucket '{b.name}' is publicly visible ({pa})") self._set_status("5.1.1", len(self.findings["5.1.1"]["findings"]) == 0) # 5.1.2 — Bucket CMK if not self._should_skip("5.1.2"): for b in self._buckets: self._add_total("5.1.2", b.name) if not getattr(b, "kms_key_id", None): self._add_finding("5.1.2", f"Bucket '{b.name}' not encrypted with CMK") self._set_status("5.1.2", len(self.findings["5.1.2"]["findings"]) == 0) # 5.1.3 — Bucket versioning if not self._should_skip("5.1.3"): for b in self._buckets: self._add_total("5.1.3", b.name) versioning = getattr(b, "versioning", "Disabled") if versioning != "Enabled": self._add_finding("5.1.3", f"Bucket '{b.name}' versioning not enabled ({versioning})") self._set_status("5.1.3", len(self.findings["5.1.3"]["findings"]) == 0) # 5.2.1 — Block Volume CMK if not self._should_skip("5.2.1"): for bv in self._block_volumes: if bv.lifecycle_state != "AVAILABLE": continue self._add_total("5.2.1", bv.display_name) if not getattr(bv, "kms_key_id", None): self._add_finding("5.2.1", f"Block volume '{bv.display_name}' not encrypted with CMK") self._set_status("5.2.1", len(self.findings["5.2.1"]["findings"]) == 0) # 5.2.2 — Boot Volume CMK if not self._should_skip("5.2.2"): for bv in self._boot_volumes: if bv.lifecycle_state != "AVAILABLE": continue self._add_total("5.2.2", bv.display_name) if not getattr(bv, "kms_key_id", None): self._add_finding("5.2.2", f"Boot volume '{bv.display_name}' not encrypted with CMK") self._set_status("5.2.2", len(self.findings["5.2.2"]["findings"]) == 0) # 5.3.1 — File Storage CMK if not self._should_skip("5.3.1"): for fs in self._file_systems: if fs.lifecycle_state != "ACTIVE": continue self._add_total("5.3.1", fs.display_name) if not getattr(fs, "kms_key_id", None): self._add_finding("5.3.1", f"File system '{fs.display_name}' not encrypted with CMK") self._set_status("5.3.1", len(self.findings["5.3.1"]["findings"]) == 0) # ── Asset Checks: 6.1, 6.2 ─────────────────────────────────────────────── def _check_assets(self): log.info("Running asset checks (6.1, 6.2)...") # 6.1 — At least one compartment if not self._should_skip("6.1"): non_root = [c for c in self.compartments if c.id != self.tenancy_id] self._add_total("6.1", f"{len(non_root)} compartments (excluding root)") self._set_status("6.1", len(non_root) > 0) if not non_root: self._add_finding("6.1", "No compartments created — all resources are in the root compartment") # 6.2 — No resources in root if not self._should_skip("6.2"): self._add_total("6.2", f"{self._root_resources_count} non-IAM resources in root compartment") self._set_status("6.2", self._root_resources_count == 0) if self._root_resources_count > 0: self._add_finding("6.2", f"{self._root_resources_count} resource(s) found in root compartment") # ═══════════════════════════════════════════════════════════════════════════ # ORCHESTRATOR # ═══════════════════════════════════════════════════════════════════════════ def _step(self, n, total, msg): """Print a numbered progress step (flushed for real-time capture).""" print(f"[{n}/{total}] {msg}", flush=True) def run_all(self) -> dict: """Run all data collection and CIS checks, return findings dict.""" T = 12 # total steps self._step(1, T, "Discovering regions...") self._discover_regions() self._step(2, T, "Discovering compartments...") self._discover_compartments() # Home-region-only collections self._step(3, T, "Collecting IAM data (policies, users, groups)...") try: self._collect_iam() except Exception as e: log.error(f"IAM collection failed: {e}") for cid in [f"1.{i}" for i in range(1, 18)]: if cid in self.findings: self.findings[cid]["status"] = "REVIEW" self._add_finding(cid, f"IAM data collection failed: {e}") self._step(4, T, "Collecting Cloud Guard status...") self._collect_cloud_guard() self._step(5, T, "Collecting asset data...") self._collect_assets() # Per-region collections in parallel self._step(6, T, f"Collecting regional data ({len(self.regions)} regions: {', '.join(self.regions)})...") def _collect_region(region): print(f" Scanning {region}...", flush=True) net = self._collect_network(region) comp = self._collect_compute(region) mon = self._collect_monitoring(region) stor = self._collect_storage(region) print(f" {region} done.", flush=True) return region, net, comp, mon, stor with ThreadPoolExecutor(max_workers=min(3, len(self.regions))) as pool: futures = {pool.submit(_collect_region, r): r for r in self.regions} for future in as_completed(futures): r = futures[future] try: _, net, comp, mon, stor = future.result() self._security_lists.extend(net["security_lists"]) self._nsgs.extend(net["nsgs"]) self._nsg_rules.update(net["nsg_rules"]) self._vcns.extend(net["vcns"]) self._subnets.extend(net["subnets"]) self._oic_instances.extend(net["oic"]) self._oac_instances.extend(net["oac"]) self._adbs.extend(net["adbs"]) self._instances.extend(comp) self._event_rules.extend(mon["event_rules"]) self._topics.extend(mon["topics"]) self._subscriptions.extend(mon["subscriptions"]) self._log_groups.extend(mon["log_groups"]) self._logs.extend(mon["logs"]) self._vaults.extend(mon["vaults"]) self._keys.extend(mon["keys"]) self._buckets.extend(stor["buckets"]) self._block_volumes.extend(stor["block_volumes"]) self._boot_volumes.extend(stor["boot_volumes"]) self._file_systems.extend(stor["file_systems"]) except Exception as e: log.error(f"Region {r} collection failed: {e}") # Run all checks self._step(7, T, "Running IAM checks (1.1-1.17)...") check_methods = [ (self._check_iam_policies, ["1.1", "1.2", "1.3", "1.15"], None), (self._check_password_policies, ["1.4", "1.5", "1.6"], None), (self._check_users, ["1.7", "1.8", "1.9", "1.10", "1.11", "1.12", "1.13", "1.16", "1.17"], None), (self._check_dynamic_groups, ["1.14"], None), (self._check_network, ["2.1", "2.2", "2.3", "2.4", "2.5", "2.6", "2.7", "2.8"], (8, T, "Running network checks (2.1-2.8)...")), (self._check_compute, ["3.1", "3.2", "3.3"], (9, T, "Running compute checks (3.1-3.3)...")), (self._check_monitoring, ["4.1", "4.2", "4.3", "4.4", "4.5", "4.6", "4.7", "4.8", "4.9", "4.10", "4.11", "4.12", "4.13", "4.14", "4.15", "4.16", "4.17", "4.18"], (10, T, "Running monitoring checks (4.1-4.18)...")), (self._check_storage, ["5.1.1", "5.1.2", "5.1.3", "5.2.1", "5.2.2", "5.3.1"], (11, T, "Running storage checks (5.x)...")), (self._check_assets, ["6.1", "6.2"], (12, T, "Running asset management checks (6.1-6.2)...")), ] for method, check_ids, step_info in check_methods: if step_info: self._step(*step_info) try: method() except Exception as e: log.error(f"{method.__name__} failed: {e}") for cid in check_ids: if self.findings[cid]["status"] == "REVIEW": self._add_finding(cid, f"Check failed: {e}") print("[DONE] All checks completed.", flush=True) return self.findings # ─── HTML Report Generator ─────────────────────────────────────────────────── def gen_html(findings, tenancy, output): now = datetime.datetime.now() sections = {} for cid, ck in findings.items(): sec = ck["section"] sections.setdefault(sec, {"total":0,"passed":0,"failed":0}) sections[sec]["total"] += 1 if ck["status"] == "PASS": sections[sec]["passed"] += 1 elif ck["status"] == "FAIL": sections[sec]["failed"] += 1 tot = sum(s["total"] for s in sections.values()) pas = sum(s["passed"] for s in sections.values()) fai = sum(s["failed"] for s in sections.values()) sec_rows = "".join(f'{s}{v["total"]}' f'{v["failed"]}' f'{v["passed"]}' for s,v in sorted(sections.items())) det_rows = "" for cid in sorted(findings.keys(), key=lambda x: [int(p) if p.isdigit() else p for p in x.replace('.',' ').split()]): ck = findings[cid]; st = ck["status"] cls = "pass" if st=="PASS" else "fail" if st=="FAIL" else "review" finding_details = "" if ck.get("findings"): items = "".join(f"
  • {f}
    • " for f in ck["findings"][:10]) if len(ck["findings"]) > 10: items += f"
  • ... and {len(ck['findings'])-10} more
    • " finding_details = f'
    ' det_rows += f'{ck["id"]}{cid}{ck["title"]}{finding_details}{st}{ck["section"]}' html = f""" Cloud Security Assessment - {tenancy}

    Cloud Security Assessment

    Tenancy: {tenancy}
    OCI CIS Foundations Benchmark 3.0

    {now.strftime('%B, %Y')}, Version [1.0]
    Extract date: {now.isoformat()[:19]}

    Disclaimer

    This document, in any form, software or printed matter, contains proprietary information that is the exclusive property of Oracle. Your access to and use of this confidential material is subject to the terms and conditions of your Oracle software license and service agreement, which has been executed and with which you agree to comply.

    This document is for informational purposes only and is intended solely to assist you in planning for the implementation and upgrade of the product features described. It is not a commitment to deliver any material, code, or functionality.

    Findings Overview

    Resumo por dominio (Section)

    {tot}
    Total Controls
    {pas}
    Passed
    {fai}
    Failed
    {sec_rows}
    DomainsTotal ControlsFailedPassed

    Total: {tot} · Passed: {pas} · Failed: {fai}

    Detailed Findings

    {det_rows}
    IDCheck #TitleStatusSection
    Generated by OCI CIS AI Agent · CIS OCI Foundations Benchmark 3.0 · {now.strftime('%Y-%m-%d %H:%M:%S')}
    """ Path(output).write_text(html, encoding="utf-8") # ─── CLI Entry Point ───────────────────────────────────────────────────────── def main(): ap = argparse.ArgumentParser(description="CIS OCI Foundations Benchmark 3.0 Compliance Checker") ap.add_argument("--config", required=True, help="Path to OCI config file") ap.add_argument("--output", required=True, help="Output directory for reports") ap.add_argument("--tenancy-name", default="Tenancy", help="Tenancy display name") ap.add_argument("--regions", default=None, help="Comma-separated regions to scan") ap.add_argument("--level", type=int, default=None, choices=[1, 2], help="CIS level filter (1 or 2)") # Compatibility args (accepted but unused for now) ap.add_argument("--mcp-module", default=None, help="(reserved)") ap.add_argument("--adb-dsn", default=None, help="(reserved)") ap.add_argument("--adb-user", default=None, help="(reserved)") ap.add_argument("--adb-wallet", default=None, help="(reserved)") args = ap.parse_args() out = Path(args.output) out.mkdir(parents=True, exist_ok=True) print(f"[CIS Runner] Config: {args.config}") print(f"[CIS Runner] Output: {args.output}") print(f"[CIS Runner] Tenancy: {args.tenancy_name}") if not Path(args.config).exists(): print(f"[ERROR] Config not found: {args.config}", file=sys.stderr) sys.exit(1) filter_regions = args.regions.split(",") if args.regions else None try: checker = CISComplianceChecker(args.config, filter_regions, args.level) findings = checker.run_all() except Exception as e: log.error(f"Fatal error: {e}") # Generate partial report with all REVIEW findings = {} for cid, ck in CHECKS.items(): findings[cid] = {**ck, "status": "REVIEW", "findings": [f"Execution error: {e}"], "total": []} # Extract compartment names for report metadata compartment_names = [] try: compartment_names = [c.name for c in checker.compartments] if hasattr(checker, 'compartments') else [] except Exception: pass report = { "tenancy": args.tenancy_name, "generated_at": datetime.datetime.now().isoformat(), "benchmark": "CIS OCI Foundations Benchmark 3.0", "regions": filter_regions or ["all"], "compartments": compartment_names, "summary": { "total": len(findings), "passed": sum(1 for f in findings.values() if f["status"] == "PASS"), "failed": sum(1 for f in findings.values() if f["status"] == "FAIL"), "review": sum(1 for f in findings.values() if f["status"] == "REVIEW"), }, "findings": findings, } (out / "report.json").write_text(json.dumps(report, indent=2, default=str)) gen_html(findings, args.tenancy_name, str(out / "report.html")) print("[CIS Runner] Reports generated successfully") if __name__ == "__main__": main()