diff --git a/local-app/python-tools/security-checks/aws_scsem_foundations.py b/local-app/python-tools/security-checks/aws_scsem_foundations.py index ca2e2ef2..ffb3023c 100755 --- a/local-app/python-tools/security-checks/aws_scsem_foundations.py +++ b/local-app/python-tools/security-checks/aws_scsem_foundations.py @@ -1,20 +1,14 @@ #!/usr/bin/env python3 """ AWS SCSEM Foundations Automated Checks -Version: 1.0.4 - -Enhancements in v1.0.4: -- All tests enabled by default (from spreadsheet), with default exclusions 32–38 & 44. -- New targeted execution: - * --only "" to run just specific test IDs (e.g., "7" or "7,9,21" or "7-9,21"). - * --exclude "" retained for skipping any specific tests. -- Richer summary export: - * --out-dir to write summary and results CSVs (summary.csv, results.csv). - * JSON summary now includes lists of executed_ids, skipped_ids, passed_ids, failed_ids, error_ids. -- Markdown exporter retained + front-matter option: - * --md-front-matter adds YAML front matter (id, title, sheet, timestamp). - * Per-row files + optional combined markdown remain. -- Defaults to spreadsheet via $SCSEM_XLSX_PATH and auto-detects sheet/columns if unspecified. +Version: 1.0.5 + +Enhancements in v1.0.5: +- Run all checks from spreadsheet (default exclusions 32–38 & 44). +- Multi-account roll-up: iterate multiple profiles/regions and aggregate results & CSVs. +- Evidence pack: per test, per account JSON artifacts of raw API responses and final result; optional zip. +- Locked defaults for sheet and columns: Cloud Computing | Test Case Number | Section Title. +- Synthetic manual results for unmapped tests (keeps "all 45" enabled and produces evidence). """ import argparse @@ -22,6 +16,7 @@ import json import os import re +import zipfile from typing import Any, Dict, List, Optional, Tuple, Callable, Set import boto3 @@ -33,7 +28,77 @@ except ImportError: openpyxl = None -VERSION = "1.0.4" +VERSION = "1.0.5" + +# --------------------------- +# Defaults (locked to your Cloud tab) +# --------------------------- +DEFAULT_SHEET = "Cloud Computing" +DEFAULT_ID_COLUMN = "Test Case Number" +DEFAULT_TITLE_COLUMN = "Section Title" + + +# --------------------------- +# Evidence pack helpers +# --------------------------- +class EvidencePack: + """ + Writes raw API evidence and result JSON for each test, per account. + Base dir layout: + {base_dir}/evidence/{account_alias}/{account_id}/{test_id}/ + - _.json + - result.json + """ + def __init__(self, base_dir: Optional[str], account_alias: str, account_id: str, test_id: str): + self.base_dir = base_dir + self.account_alias = _sanitize_filename(account_alias or "account") + self.account_id = _sanitize_filename(account_id or "unknown") + self.test_id = _sanitize_filename(test_id or "unknown") + self.ts = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%dT%H%M%SZ") + + def _dir(self) -> Optional[str]: + if not self.base_dir: + return None + d = os.path.join(self.base_dir, "evidence", self.account_alias, self.account_id, self.test_id) + os.makedirs(d, exist_ok=True) + return d + + def add_json(self, name: str, data: Any): + d = self._dir() + if not d: + return + path = os.path.join(d, f"{self.ts}_{_sanitize_filename(name)}.json") + try: + with open(path, "w", encoding="utf-8") as f: + json.dump(data, f, indent=2, default=str) + except Exception as e: + # best-effort evidence; do not fail the check + pass + + def add_result(self, result: Dict[str, Any]): + d = self._dir() + if not d: + return + path = os.path.join(d, "result.json") + try: + with open(path, "w", encoding="utf-8") as f: + json.dump(result, f, indent=2, default=str) + except Exception: + pass + + +def zip_evidence(base_dir: str, out_zip: str): + """Bundle the evidence folder to a zip.""" + evidence_root = os.path.join(base_dir, "evidence") + if not os.path.isdir(evidence_root): + return None + with zipfile.ZipFile(out_zip, "w", zipfile.ZIP_DEFLATED) as z: + for root, _, files in os.walk(evidence_root): + for fn in files: + full = os.path.join(root, fn) + rel = os.path.relpath(full, base_dir) + z.write(full, arcname=rel) + return out_zip # --------------------------- @@ -47,12 +112,14 @@ def create_session(profile_name: Optional[str] = None, region_name: Optional[str # --------------------------- -# Check functions (same as v1.0.3) +# Individual check functions (now accept evidence recorder) # --------------------------- -def aws_01_check_account_contact(session): +def aws_01_check_account_contact(session, evidence: Optional[EvidencePack] = None): client = session.client('account') try: response = client.get_contact_information() + if evidence: + evidence.add_json("get_contact_information", response) return { 'test_id': 'AWS-01', 'nist_id': 'IR-6', @@ -73,10 +140,12 @@ def aws_01_check_account_contact(session): 'error': str(e) } -def aws_02_check_security_contact(session): +def aws_02_check_security_contact(session, evidence: Optional[EvidencePack] = None): client = session.client('account') try: response = client.get_alternate_contact(AlternateContactType='SECURITY') + if evidence: + evidence.add_json("get_alternate_contact_SECURITY", response) contact = response.get('AlternateContact', {}) result = { 'test_id': 'AWS-02', @@ -87,7 +156,7 @@ def aws_02_check_security_contact(session): 'applicable_to_govcloud': True, 'contact_info': contact } - except client.exceptions.ResourceNotFoundException: + except client.exceptions.ResourceNotFoundException as rnfe: result = { 'test_id': 'AWS-02', 'nist_id': 'IR-6', @@ -110,7 +179,8 @@ def aws_02_check_security_contact(session): } return result -def aws_03_security_questions_registered(session): +def aws_03_security_questions_registered(session, evidence: Optional[EvidencePack] = None): + # No API evidence; not applicable in GovCloud return { 'test_id': 'AWS-03', 'nist_id': 'IR-6', @@ -121,7 +191,7 @@ def aws_03_security_questions_registered(session): 'note': "Security questions are not supported in AWS GovCloud." } -def aws_04_check_root_access_key(session): +def aws_04_check_root_access_key(session, evidence: Optional[EvidencePack] = None): return { 'test_id': 'AWS-04', 'nist_id': 'AC-6', @@ -132,7 +202,7 @@ def aws_04_check_root_access_key(session): 'note': "Root user does not have console access in AWS GovCloud; this test is not applicable." } -def aws_05_check_root_mfa(session): +def aws_05_check_root_mfa(session, evidence: Optional[EvidencePack] = None): return { 'test_id': 'AWS-05', 'nist_id': 'IA-2', @@ -143,7 +213,7 @@ def aws_05_check_root_mfa(session): 'note': "Root user does not have console access in AWS GovCloud; this test is not applicable." } -def aws_06_check_root_usage(session): +def aws_06_check_root_usage(session, evidence: Optional[EvidencePack] = None): return { 'test_id': 'AWS-06', 'nist_id': 'AC-6', @@ -154,10 +224,13 @@ def aws_06_check_root_usage(session): 'note': "Root user does not have console access in AWS GovCloud; this test is not applicable." } -def aws_07_check_password_policy(session): +def aws_07_check_password_policy(session, evidence: Optional[EvidencePack] = None): client = session.client('iam') try: - policy = client.get_account_password_policy()['PasswordPolicy'] + policy_resp = client.get_account_password_policy() + if evidence: + evidence.add_json("get_account_password_policy", policy_resp) + policy = policy_resp['PasswordPolicy'] min_length = policy.get('MinimumPasswordLength', 0) return { 'test_id': 'AWS-07', @@ -169,7 +242,7 @@ def aws_07_check_password_policy(session): 'minimum_password_length': min_length, 'compliant': (min_length >= 14) } - except client.exceptions.NoSuchEntityException: + except client.exceptions.NoSuchEntityException as e: return { 'test_id': 'AWS-07', 'nist_id': 'IA-5', @@ -181,10 +254,13 @@ def aws_07_check_password_policy(session): 'compliant': False } -def aws_08_check_password_reuse(session): +def aws_08_check_password_reuse(session, evidence: Optional[EvidencePack] = None): client = session.client('iam') try: - policy = client.get_account_password_policy()['PasswordPolicy'] + policy_resp = client.get_account_password_policy() + if evidence: + evidence.add_json("get_account_password_policy", policy_resp) + policy = policy_resp['PasswordPolicy'] reuse_prevention = policy.get('PasswordReusePrevention', 0) return { 'test_id': 'AWS-08', @@ -208,14 +284,23 @@ def aws_08_check_password_reuse(session): 'compliant': False } -def aws_09_check_mfa_for_iam_users(session): +def aws_09_check_mfa_for_iam_users(session, evidence: Optional[EvidencePack] = None): client = session.client('iam') - users = client.list_users()['Users'] + users_resp = client.list_users() + if evidence: + evidence.add_json("list_users", users_resp) + users = users_resp.get('Users', []) non_compliant_users = [] for user in users: - mfa_devices = client.list_mfa_devices(UserName=user['UserName'])['MFADevices'] + mfa_resp = client.list_mfa_devices(UserName=user['UserName']) + if evidence: + evidence.add_json(f"list_mfa_devices_{user['UserName']}", mfa_resp) + mfa_devices = mfa_resp.get('MFADevices', []) try: - login_profile = client.get_login_profile(UserName=user['UserName']) + login_profile_resp = client.get_login_profile(UserName=user['UserName']) + if evidence: + evidence.add_json(f"get_login_profile_{user['UserName']}", login_profile_resp) + login_profile = login_profile_resp except client.exceptions.NoSuchEntityException: login_profile = None if login_profile and not mfa_devices: @@ -231,15 +316,23 @@ def aws_09_check_mfa_for_iam_users(session): 'compliant': (len(non_compliant_users) == 0) } -def aws_10_check_access_keys_on_creation(session): +def aws_10_check_access_keys_on_creation(session, evidence: Optional[EvidencePack] = None): client = session.client('iam') - users = client.list_users()['Users'] + users_resp = client.list_users() + if evidence: + evidence.add_json("list_users", users_resp) + users = users_resp.get('Users', []) users_with_console_and_access_keys = [] for user in users: try: - login_profile = client.get_login_profile(UserName=user['UserName']) - keys = client.list_access_keys(UserName=user['UserName'])['AccessKeyMetadata'] - if login_profile and keys: + login_profile_resp = client.get_login_profile(UserName=user['UserName']) + if evidence: + evidence.add_json(f"get_login_profile_{user['UserName']}", login_profile_resp) + keys_resp = client.list_access_keys(UserName=user['UserName']) + if evidence: + evidence.add_json(f"list_access_keys_{user['UserName']}", keys_resp) + keys = keys_resp.get('AccessKeyMetadata', []) + if login_profile_resp and keys: users_with_console_and_access_keys.append(user['UserName']) except client.exceptions.NoSuchEntityException: continue @@ -253,15 +346,24 @@ def aws_10_check_access_keys_on_creation(session): 'users_with_console_and_access_keys': users_with_console_and_access_keys } -def aws_11_check_unused_credentials(session): +def aws_11_check_unused_credentials(session, evidence: Optional[EvidencePack] = None): client = session.client('iam') - users = client.list_users()['Users'] + users_resp = client.list_users() + if evidence: + evidence.add_json("list_users", users_resp) + users = users_resp.get('Users', []) now = datetime.datetime.now(datetime.timezone.utc) non_compliant = [] for user in users: - keys = client.list_access_keys(UserName=user['UserName'])['AccessKeyMetadata'] + keys_resp = client.list_access_keys(UserName=user['UserName']) + if evidence: + evidence.add_json(f"list_access_keys_{user['UserName']}", keys_resp) + keys = keys_resp.get('AccessKeyMetadata', []) for key in keys: - last_used = client.get_access_key_last_used(AccessKeyId=key['AccessKeyId'])['AccessKeyLastUsed'] + last_used_resp = client.get_access_key_last_used(AccessKeyId=key['AccessKeyId']) + if evidence: + evidence.add_json(f"get_access_key_last_used_{key['AccessKeyId']}", last_used_resp) + last_used = last_used_resp.get('AccessKeyLastUsed', {}) last_used_date = last_used.get('LastUsedDate') if last_used_date and (now - last_used_date).days > 60: non_compliant.append({'UserName': user['UserName'], 'AccessKeyId': key['AccessKeyId']}) @@ -275,13 +377,19 @@ def aws_11_check_unused_credentials(session): 'non_compliant_keys': non_compliant } -def aws_12_check_single_active_access_key(session): +def aws_12_check_single_active_access_key(session, evidence: Optional[EvidencePack] = None): client = session.client('iam') - users = client.list_users()['Users'] + users_resp = client.list_users() + if evidence: + evidence.add_json("list_users", users_resp) + users = users_resp.get('Users', []) non_compliant = [] for user in users: - keys = client.list_access_keys(UserName=user['UserName'])['AccessKeyMetadata'] - active_keys = [k for k in keys if k['Status'] == 'Active'] + keys_resp = client.list_access_keys(UserName=user['UserName']) + if evidence: + evidence.add_json(f"list_access_keys_{user['UserName']}", keys_resp) + keys = keys_resp.get('AccessKeyMetadata', []) + active_keys = [k for k in keys if k.get('Status') == 'Active'] if len(active_keys) > 1: non_compliant.append(user['UserName']) return { @@ -294,16 +402,22 @@ def aws_12_check_single_active_access_key(session): 'non_compliant_users': non_compliant } -def aws_13_check_access_key_rotation(session): +def aws_13_check_access_key_rotation(session, evidence: Optional[EvidencePack] = None): client = session.client('iam') - users = client.list_users()['Users'] + users_resp = client.list_users() + if evidence: + evidence.add_json("list_users", users_resp) + users = users_resp.get('Users', []) now = datetime.datetime.now(datetime.timezone.utc) non_compliant = [] for user in users: - keys = client.list_access_keys(UserName=user['UserName'])['AccessKeyMetadata'] + keys_resp = client.list_access_keys(UserName=user['UserName']) + if evidence: + evidence.add_json(f"list_access_keys_{user['UserName']}", keys_resp) + keys = keys_resp.get('AccessKeyMetadata', []) for key in keys: - create_date = key['CreateDate'] - if (now - create_date).days > 90: + create_date = key.get('CreateDate') + if create_date and (now - create_date).days > 90: non_compliant.append({'UserName': user['UserName'], 'AccessKeyId': key['AccessKeyId']}) return { 'test_id': 'AWS-13', @@ -315,13 +429,21 @@ def aws_13_check_access_key_rotation(session): 'non_compliant_keys': non_compliant } -def aws_14_check_permissions_through_groups(session): +def aws_14_check_permissions_through_groups(session, evidence: Optional[EvidencePack] = None): client = session.client('iam') - users = client.list_users()['Users'] + users_resp = client.list_users() + if evidence: + evidence.add_json("list_users", users_resp) + users = users_resp.get('Users', []) non_compliant = [] for user in users: - attached_policies = client.list_attached_user_policies(UserName=user['UserName'])['AttachedPolicies'] - inline_policies = client.list_user_policies(UserName=user['UserName'])['PolicyNames'] + attached_resp = client.list_attached_user_policies(UserName=user['UserName']) + inline_resp = client.list_user_policies(UserName=user['UserName']) + if evidence: + evidence.add_json(f"list_attached_user_policies_{user['UserName']}", attached_resp) + evidence.add_json(f"list_user_policies_{user['UserName']}", inline_resp) + attached_policies = attached_resp.get('AttachedPolicies', []) + inline_policies = inline_resp.get('PolicyNames', []) if attached_policies or inline_policies: non_compliant.append(user['UserName']) return { @@ -334,16 +456,22 @@ def aws_14_check_permissions_through_groups(session): 'non_compliant_users': non_compliant } -def aws_15_check_full_admin_policies(session): +def aws_15_check_full_admin_policies(session, evidence: Optional[EvidencePack] = None): client = session.client('iam') - policies = client.list_policies(Scope='Local', OnlyAttached=True)['Policies'] + policies_resp = client.list_policies(Scope='Local', OnlyAttached=True) + if evidence: + evidence.add_json("list_policies_OnlyAttached", policies_resp) + policies = policies_resp.get('Policies', []) non_compliant = [] for policy in policies: - version = client.get_policy_version( + version_resp = client.get_policy_version( PolicyArn=policy['Arn'], VersionId=policy['DefaultVersionId'] - )['PolicyVersion']['Document'] - statements = version.get('Statement', []) + ) + if evidence: + evidence.add_json(f"get_policy_version_{policy['PolicyName']}", version_resp) + document = version_resp.get('PolicyVersion', {}).get('Document', {}) + statements = document.get('Statement', []) if not isinstance(statements, list): statements = [statements] for stmt in statements: @@ -359,13 +487,19 @@ def aws_15_check_full_admin_policies(session): 'non_compliant_policies': non_compliant } -def aws_16_check_support_role(session): +def aws_16_check_support_role(session, evidence: Optional[EvidencePack] = None): client = session.client('iam') - roles = client.list_roles()['Roles'] + roles_resp = client.list_roles() + if evidence: + evidence.add_json("list_roles", roles_resp) + roles = roles_resp.get('Roles', []) support_roles = [] for r in roles: - attached = client.list_attached_role_policies(RoleName=r['RoleName'])['AttachedPolicies'] - if any('AWSSupportAccess' in p['PolicyName'] for p in attached): + attached_resp = client.list_attached_role_policies(RoleName=r['RoleName']) + if evidence: + evidence.add_json(f"list_attached_role_policies_{r['RoleName']}", attached_resp) + attached = attached_resp.get('AttachedPolicies', []) + if any('AWSSupportAccess' in p.get('PolicyName', '') for p in attached): support_roles.append(r['RoleName']) return { 'test_id': 'AWS-16', @@ -378,9 +512,11 @@ def aws_16_check_support_role(session): 'compliant': (len(support_roles) > 0) } -def aws_21_check_ebs_encryption(session, region_name: str): +def aws_21_check_ebs_encryption(session, region_name: str, evidence: Optional[EvidencePack] = None): client = session.client('ec2', region_name=region_name) response = client.get_ebs_encryption_by_default() + if evidence: + evidence.add_json("get_ebs_encryption_by_default", response) return { 'test_id': 'AWS-21', 'nist_id': 'SC-28', @@ -396,8 +532,7 @@ def aws_21_check_ebs_encryption(session, region_name: str): # --------------------------- # Test registry and mapping # --------------------------- - -ALL_CHECKS: List[Callable[[boto3.session.Session], Dict[str, Any]]] = [ +ALL_CHECKS: List[Callable[..., Dict[str, Any]]] = [ aws_01_check_account_contact, aws_02_check_security_contact, aws_03_security_questions_registered, @@ -435,16 +570,27 @@ def aws_21_check_ebs_encryption(session, region_name: str): 15: "aws_15_check_full_admin_policies", 16: "aws_16_check_support_role", 21: "aws_21_check_ebs_encryption" + # Remaining IDs (17–45 except 21) will render synthetic manual results until automated functions are added. } # --------------------------- # Utility parsing # --------------------------- -def parse_exclusion_ranges(ranges_text: str) -> Set[int]: - """Parse ranges like '32-38,44' into a set of integers.""" +def _sanitize_filename(name: str) -> str: + s = re.sub(r'[\\/*?:"<>|]+', '_', name) + s = re.sub(r'\s+', '_', s) + s = re.sub(r'[^A-Za-z0-9_.-]+', '_', s) + return s.strip('_')[:200] + +def parse_ranges(list_text: Optional[str]) -> Set[int]: + """ + Parse a comma/range list like '7,9,21' or '7-9,21' into a set of integers. + """ result: Set[int] = set() - for part in (ranges_text or "").split(","): + if not list_text: + return result + for part in list_text.split(","): part = part.strip() if not part: continue @@ -463,12 +609,6 @@ def parse_exclusion_ranges(ranges_text: str) -> Set[int]: pass return result -def parse_id_list(list_text: Optional[str]) -> Set[int]: - """Parse a comma/range list like '7,9,21' or '7-9,21' into a set of integers.""" - if not list_text: - return set() - return parse_exclusion_ranges(list_text) - # --------------------------- # Result evaluation @@ -484,7 +624,7 @@ def evaluate_check_result(r: Dict[str, Any]) -> Tuple[str, Optional[str]]: method = r.get('test_method', '') applicable = r.get('applicable_to_govcloud', True) if method == 'Test (Manual)' or not applicable: - return ('info', None) # skipped + return ('info', None) if 'error' in r: return ('fail', str(r['error'])) @@ -515,84 +655,19 @@ def _detect_header_row(ws) -> Tuple[int, List[str]]: headers = [str(cell.value).strip() if cell.value is not None else '' for cell in ws[header_row_idx]] return header_row_idx, headers -def load_spreadsheet_test_ids( - xlsx_path: str, - sheet_name: Optional[str], - id_column_name: Optional[str] -) -> List[int]: - """ - Load test IDs (integers) from the XLSX sheet. - - If id_column_name provided, use that exact header. - - Otherwise, select the column with most numeric-like values. - """ - if not openpyxl: - raise RuntimeError("openpyxl is required to read XLSX. Please install it or remove --xlsx.") - if not os.path.exists(xlsx_path): - raise FileNotFoundError(f"XLSX not found: {xlsx_path}") - - wb = openpyxl.load_workbook(xlsx_path, data_only=True) - ws = wb[sheet_name] if sheet_name and sheet_name in wb.sheetnames else wb.active - - header_row_idx, headers = _detect_header_row(ws) - - # Resolve target column - target_col_idx = None - if id_column_name: - for idx, h in enumerate(headers): - if h.strip().lower() == id_column_name.strip().lower(): - target_col_idx = idx - break - if target_col_idx is None: - raise RuntimeError(f"Column '{id_column_name}' not found. Headers: {headers}") - else: - candidate_indices = list(range(len(headers))) - numeric_hits: Dict[int, int] = {} - for idx in candidate_indices: - hits = 0 - for row in ws.iter_rows(min_row=header_row_idx + 1, max_row=header_row_idx + 50): - cell = row[idx].value - if cell is None: - continue - if isinstance(cell, int): - hits += 1 - else: - s = str(cell).strip() - if re.search(r'\d+', s): - hits += 1 - numeric_hits[idx] = hits - target_col_idx = max(numeric_hits, key=numeric_hits.get) if numeric_hits else 0 - - test_ids: List[int] = [] - for row in ws.iter_rows(min_row=header_row_idx + 1): - cell = row[target_col_idx].value - if cell is None: - continue - if isinstance(cell, int): - test_ids.append(cell) - else: - s = str(cell).strip() - m = re.search(r'(\d+)', s) - if m: - test_ids.append(int(m.group(1))) - return test_ids - def load_spreadsheet_rows( xlsx_path: str, sheet_name: Optional[str], id_column_name: Optional[str], title_column_name: Optional[str] ) -> Tuple[List[str], List[Dict[str, Any]], str, str]: - """ - Load full rows into list of dicts keyed by column header. - Returns: (headers, rows, resolved_id_header, resolved_title_header) - """ if not openpyxl: raise RuntimeError("openpyxl is required to read XLSX. Please install it or remove --md-only/--xlsx.") if not os.path.exists(xlsx_path): raise FileNotFoundError(f"XLSX not found: {xlsx_path}") wb = openpyxl.load_workbook(xlsx_path, data_only=True) - ws = wb[sheet_name] if sheet_name and sheet_name in wb.sheetnames else wb.active + ws = wb[sheet_name] if sheet_name and sheet_name in wb.sheetnames else wb[DEFAULT_SHEET] if DEFAULT_SHEET in wb.sheetnames else wb.active header_row_idx, headers = _detect_header_row(ws) @@ -606,13 +681,7 @@ def load_spreadsheet_rows( if id_header is None: raise RuntimeError(f"ID column '{id_column_name}' not found. Headers: {headers}") else: - for h in headers: - hl = h.lower() - if ('test' in hl or 'case' in hl or 'control' in hl) and (('#' in hl) or ('id' in hl) or ('number' in hl)): - id_header = h - break - if id_header is None: - id_header = headers[0] + id_header = DEFAULT_ID_COLUMN if DEFAULT_ID_COLUMN in headers else headers[0] # Resolve Title header title_header = None @@ -624,13 +693,7 @@ def load_spreadsheet_rows( if title_header is None: raise RuntimeError(f"Title column '{title_column_name}' not found. Headers: {headers}") else: - for h in headers: - hl = h.lower() - if 'title' in hl or 'section title' in hl or 'control name' in hl or 'description' in hl: - title_header = h - break - if title_header is None and len(headers) > 1: - title_header = headers[1] + title_header = DEFAULT_TITLE_COLUMN if DEFAULT_TITLE_COLUMN in headers else (headers[1] if len(headers) > 1 else headers[0]) # Build rows rows: List[Dict[str, Any]] = [] @@ -644,9 +707,24 @@ def load_spreadsheet_rows( return headers, rows, id_header, title_header +def load_spreadsheet_test_ids( + xlsx_path: str, + sheet_name: Optional[str], + id_column_name: Optional[str] +) -> List[int]: + headers, rows, id_header, _ = load_spreadsheet_rows(xlsx_path, sheet_name, id_column_name, None) + test_ids: List[int] = [] + for record in rows: + cell = record.get(id_header, '') + s = str(cell).strip() + m = re.search(r'(\d+)', s) + if m: + test_ids.append(int(m.group(1))) + return test_ids + # --------------------------- -# Markdown rendering +# Markdown rendering (same as v1.0.4; omitted for brevity) # --------------------------- def _stringify(value: Any) -> str: if value is None: @@ -655,126 +733,66 @@ def _stringify(value: Any) -> str: return json.dumps(value, indent=2, default=str) return str(value) -def _sanitize_filename(name: str) -> str: - s = re.sub(r'[\\/*?:"<>|]+', '_', name) - s = re.sub(r'\s+', '_', s) - s = re.sub(r'[^A-Za-z0-9_.-]+', '_', s) - return s.strip('_')[:200] - -def render_row_to_markdown( - row: Dict[str, Any], - headers: List[str], - id_header: str, - title_header: str, - add_front_matter: bool = False, - sheet_name: str = '' -) -> Tuple[str, str]: - """ - Returns (markdown_text, suggested_filename) - - Title: '# {ID} - {Title}' - - Sections: each column header becomes '## {Header}' with value - - Optional YAML front matter - """ +def render_row_to_markdown(row: Dict[str, Any], headers: List[str], id_header: str, title_header: str, add_front_matter: bool = False, sheet_name: str = '') -> Tuple[str, str]: raw_id = _stringify(row.get(id_header, '')).strip() raw_title = _stringify(row.get(title_header, '')).strip() - - id_for_title = raw_id m = re.search(r'(\d+)', raw_id) - if m: - id_for_title = m.group(1) - + id_for_title = m.group(1) if m else raw_id title_line = f"# {id_for_title} - {raw_title}".strip(" -") - parts: List[str] = [] if add_front_matter: fm = [ "---", f"id: \"{id_for_title}\"", f"title: \"{raw_title}\"", - f"sheet: \"{sheet_name or '(active)'}\"", + f"sheet: \"{sheet_name or DEFAULT_SHEET}\"", f"generated: \"{datetime.datetime.now(datetime.timezone.utc).isoformat()}\"", - "---", - "" + "---", "" ] parts.extend(fm) - parts.append(title_line) parts.append("") - for h in headers: val = _stringify(row.get(h, '')).strip() parts.append(f"## {h}") parts.append("") parts.append(val if val else "_(no value)_") parts.append("") - markdown_text = "\n".join(parts).rstrip() + "\n" - filename = f"{id_for_title}_{raw_title}".strip("_") - filename = _sanitize_filename(filename or f"row_{abs(hash(markdown_text))}") - if not filename.lower().endswith(".md"): - filename += ".md" + filename = _sanitize_filename(f"{id_for_title}_{raw_title}") + ".md" return markdown_text, filename -def write_markdown_files( - xlsx_path: str, - sheet_name: Optional[str], - id_column_name: Optional[str], - title_column_name: Optional[str], - out_dir: str, - exclude_ids: Set[int], - include_excluded: bool, - combine_file: Optional[str], - add_front_matter: bool -) -> Dict[str, Any]: - headers, rows, id_header, title_header = load_spreadsheet_rows( - xlsx_path, sheet_name, id_column_name, title_column_name - ) - +def write_markdown_files(xlsx_path: str, sheet_name: Optional[str], id_column_name: Optional[str], title_column_name: Optional[str], out_dir: str, exclude_ids: Set[int], include_excluded: bool, combine_file: Optional[str], add_front_matter: bool) -> Dict[str, Any]: + headers, rows, id_header, title_header = load_spreadsheet_rows(xlsx_path, sheet_name, id_column_name, title_column_name) os.makedirs(out_dir, exist_ok=True) - written_files: List[str] = [] skipped_rows: List[str] = [] combined_parts: List[str] = [] - - sheet_label = sheet_name or "(active)" - + sheet_label = sheet_name or DEFAULT_SHEET for row in rows: raw_id = _stringify(row.get(id_header, '')).strip() m = re.search(r'(\d+)', raw_id) numeric_id = int(m.group(1)) if m else None - if numeric_id is not None and (numeric_id in exclude_ids) and not include_excluded: skipped_rows.append(raw_id or "(no-id)") continue - - md_text, filename = render_row_to_markdown( - row=row, - headers=headers, - id_header=id_header, - title_header=title_header, - add_front_matter=add_front_matter, - sheet_name=sheet_label - ) + md_text, filename = render_row_to_markdown(row, headers, id_header, title_header, add_front_matter, sheet_label) path = os.path.join(out_dir, filename) with open(path, "w", encoding="utf-8") as f: f.write(md_text) written_files.append(path) - combined_parts.append(md_text) combined_parts.append("\n---\n") - combined_path = None if combine_file: if os.path.isdir(combine_file): ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") combine_file = os.path.join(combine_file, f"SCSEM_All_{ts}.md") - # Trim trailing separator if combined_parts and combined_parts[-1].strip() == "---": combined_parts = combined_parts[:-1] with open(combine_file, "w", encoding="utf-8") as f: f.write("\n".join(combined_parts).rstrip() + "\n") combined_path = combine_file - return { "xlsx": xlsx_path, "sheet": sheet_label, @@ -782,15 +800,13 @@ def write_markdown_files( "title_header": title_header, "out_dir": out_dir, "written_count": len(written_files), - "written_files_sample": written_files[:10], "skipped_rows_count": len(skipped_rows), - "skipped_rows_sample": skipped_rows[:10], "combined_path": combined_path } # --------------------------- -# Exports (CSV) +# Export helpers (CSV) # --------------------------- def _safe_str(v: Any) -> str: try: @@ -800,48 +816,56 @@ def _safe_str(v: Any) -> str: except Exception: return str(v) -def write_summary_exports(out_dir: str, summary: Dict[str, Any], results: List[Dict[str, Any]]) -> Dict[str, str]: +def write_summary_exports(out_dir: str, summary: Dict[str, Any], results: List[Dict[str, Any]], extra_cols: Optional[Dict[str, str]] = None) -> Dict[str, str]: os.makedirs(out_dir, exist_ok=True) + # summary.json + summary_json = os.path.join(out_dir, "summary.json") + with open(summary_json, "w", encoding="utf-8") as f: + json.dump(summary, f, indent=2, default=str) # summary.csv summary_csv = os.path.join(out_dir, "summary.csv") with open(summary_csv, "w", encoding="utf-8") as f: - # Simple key,value CSV for k, v in summary.items(): f.write(f"{k},{_safe_str(v)}\n") - # results.csv results_csv = os.path.join(out_dir, "results.csv") fields = [ "test_id", "section_title", "control_name", "nist_id", "test_method", - "status", "compliant", "reason", "error", - "applicable_to_govcloud" + "status", "compliant", "reason", "error", "applicable_to_govcloud" ] with open(results_csv, "w", encoding="utf-8") as f: - f.write(",".join(fields) + "\n") + header = fields[:] + if extra_cols: + header.extend(list(extra_cols.keys())) + f.write(",".join(header) + "\n") for r in results: row = [] for k in fields: row.append(_safe_str(r.get(k, ""))) + if extra_cols: + for k, v in extra_cols.items(): + row.append(_safe_str(v)) f.write(",".join(row) + "\n") - - return {"summary_csv": summary_csv, "results_csv": results_csv} + return {"summary_json": summary_json, "summary_csv": summary_csv, "results_csv": results_csv} # --------------------------- -# Execution (checks + docs) +# Execution (checks + docs) for ONE account # --------------------------- -def run_automated_checks( +def run_automated_checks_for_account( session, region_name: Optional[str], + account_alias: str, + account_id: str, planned_spreadsheet_ids: Optional[Set[int]], spreadsheet_mapping: Dict[int, str], exclude_ids: Set[int], - only_ids: Set[int] + only_ids: Set[int], + evidence_base_dir: Optional[str] ) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]: """ - Run automated checks driven by spreadsheet IDs and mapping. - - Applies 'exclude_ids' and 'only_ids'. - Returns: (results_list, summary_counts_dict) + Run checks for a single account; return (results, summary). + - For missing mappings, produce a synthetic manual result using spreadsheet title (if available). """ results: List[Dict[str, Any]] = [] @@ -866,24 +890,32 @@ def run_automated_checks( missing_implementation: List[int] = [] excluded_tests_list: List[int] = [] + # Prepare synthetic title lookup from spreadsheet rows (if available) + title_lookup: Dict[int, str] = {} + if planned_spreadsheet_ids is not None: + # We need the rows to extract titles — reload with defaults + xlsx_path = os.getenv('SCSEM_XLSX_PATH') + if xlsx_path and openpyxl and os.path.exists(xlsx_path): + headers, rows, id_header, title_header = load_spreadsheet_rows(xlsx_path, DEFAULT_SHEET, DEFAULT_ID_COLUMN, DEFAULT_TITLE_COLUMN) + for row in rows: + s = str(row.get(id_header, '')).strip() + m = re.search(r'(\d+)', s) + if m: + tid = int(m.group(1)) + title_lookup[tid] = str(row.get(title_header, '')).strip() + # Determine planned set if planned_spreadsheet_ids is None: - # If no spreadsheet, run registry (no numeric IDs available) selected_ids = None else: - # Apply exclude first base_ids = planned_spreadsheet_ids - exclude_ids - # Apply only (if provided) selected_ids = base_ids if not only_ids else (base_ids & only_ids) - - # Count excluded upfront (skipped) excluded_tests_list = sorted(list(planned_spreadsheet_ids.intersection(exclude_ids))) skipped += len(excluded_tests_list) skipped_ids.extend(excluded_tests_list) - # Build function list + # Build function list or synthetic entries if selected_ids is None: - # No spreadsheet IDs; run all known functions (IDs unavailable) for func in ALL_CHECKS: funcs_to_run.append((func, None)) else: @@ -891,8 +923,23 @@ def run_automated_checks( func_name = spreadsheet_mapping.get(tid) if not func_name or func_name not in func_name_map: missing_implementation.append(tid) - # Treat as skipped since it's enabled but not implemented + # Synthetic manual result ensures row is represented + result = { + 'test_id': f"AWS-{tid}", + 'section_title': title_lookup.get(tid, f"SCSEM Test {tid}"), + 'control_name': '—', + 'nist_id': '—', + 'test_method': 'Test (Manual)', + 'applicable_to_govcloud': True, + 'status': 'info', + 'note': 'Automated check not yet implemented; see markdown doc for manual review.' + } + # Evidence: record synthetic result + ev = EvidencePack(evidence_base_dir, account_alias, account_id, f"AWS-{tid}") + ev.add_result(result) + results.append(result) skipped += 1 + manual_skips += 1 skipped_ids.append(tid) continue funcs_to_run.append((func_name_map[func_name], tid)) @@ -901,20 +948,28 @@ def run_automated_checks( with tqdm(total=total, ncols=100) as pbar: for idx, (check, tid) in enumerate(funcs_to_run, 1): - pbar.set_description(f"Test {idx}") - r = check(session) + pbar.set_description(f"{account_alias}: Test {idx}") + test_id_label = f"AWS-{tid}" if tid is not None else (str(check.__name__).replace("aws_", "AWS-").upper()) + ev = EvidencePack(evidence_base_dir, account_alias, account_id, test_id_label) + + # Run check (pass evidence when supported) + try: + r = check(session, ev) # all built-ins accept evidence now + except TypeError: + r = check(session) + status, reason = evaluate_check_result(r) r['status'] = status if reason: r['reason'] = reason + # Persist per-test result to evidence pack + ev.add_result(r) results.append(r) - # Update counters and id lists - if tid is not None: - current_id = tid - else: - # Try to infer ID from result test_id like 'AWS-07' + # Update counters and lists + current_id = tid + if current_id is None: m = re.search(r'(\d+)', str(r.get('test_id', ''))) current_id = int(m.group(1)) if m else None @@ -945,20 +1000,20 @@ def run_automated_checks( # Region-specific aws_21 if region_name: idx = total - pbar.set_description(f"Test {idx}") - r = aws_21_check_ebs_encryption(session, region_name) + pbar.set_description(f"{account_alias}: Test {idx}") + ev = EvidencePack(evidence_base_dir, account_alias, account_id, "AWS-21") + r = aws_21_check_ebs_encryption(session, region_name, ev) status, reason = evaluate_check_result(r) r['status'] = status if reason: r['reason'] = reason + ev.add_result(r) results.append(r) - m = re.search(r'(\d+)', str(r.get('test_id', ''))) - current_id = int(m.group(1)) if m else None - + current_id = 21 if status == 'info': skipped += 1 - skipped_ids.append(current_id if current_id is not None else -1) + skipped_ids.append(current_id) if not r.get('applicable_to_govcloud', True): inapplicable_skips += 1 elif r.get('test_method') == 'Test (Manual)': @@ -966,16 +1021,16 @@ def run_automated_checks( elif status == 'pass': executed += 1 passed += 1 - executed_ids.append(current_id if current_id is not None else -1) - passed_ids.append(current_id if current_id is not None else -1) + executed_ids.append(current_id) + passed_ids.append(current_id) elif status == 'fail': executed += 1 failed += 1 - executed_ids.append(current_id if current_id is not None else -1) - failed_ids.append(current_id if current_id is not None else -1) + executed_ids.append(current_id) + failed_ids.append(current_id) if 'error' in r: errors += 1 - error_ids.append(current_id if current_id is not None else -1) + error_ids.append(current_id) pbar.set_postfix_str(f"{idx}/{total}") pbar.update(1) @@ -983,6 +1038,9 @@ def run_automated_checks( summary = { 'version': VERSION, 'timestamp': datetime.datetime.now(datetime.timezone.utc).isoformat(), + 'account_alias': account_alias, + 'account_id': account_id, + 'region': region_name, 'executed_tests': executed, 'skipped_tests': skipped, 'passed': passed, @@ -1004,58 +1062,100 @@ def run_automated_checks( return results, summary +# --------------------------- +# Accounts parsing +# --------------------------- +def parse_accounts_arg(text: Optional[str]) -> List[Dict[str, str]]: + """ + Format: "profile:region@alias,profile2:region2@alias2" + alias optional. Example: "census-govwest:us-gov-west-1@GW,census-goveast:us-gov-east-1" + """ + accounts: List[Dict[str, str]] = [] + if not text: + return accounts + for part in text.split(","): + part = part.strip() + if not part: + continue + # Split profile:region@alias + alias = "" + if "@" in part: + main, alias = part.split("@", 1) + else: + main = part + if ":" in main: + profile, region = main.split(":", 1) + else: + profile, region = main, None + accounts.append({"profile": profile.strip(), "region": (region or "").strip(), "alias": alias.strip()}) + return accounts + +def load_accounts_file(path: Optional[str]) -> List[Dict[str, str]]: + if not path or not os.path.exists(path): + return [] + with open(path, "r", encoding="utf-8") as f: + data = json.load(f) + # Expect list[{"profile": "...", "region": "...", "alias": "..."}] + out: List[Dict[str, str]] = [] + for entry in data: + out.append({ + "profile": entry.get("profile", ""), + "region": entry.get("region", ""), + "alias": entry.get("alias", "") + }) + return out + + +# --------------------------- +# Main (multi-account roll-up) +# --------------------------- def main(): parser = argparse.ArgumentParser(description=f"AWS SCSEM Foundations Automated Checks (v{VERSION})") - parser.add_argument('--profile', type=str, help='AWS CLI profile name') - parser.add_argument('--region', type=str, help='AWS region name (e.g., us-gov-west-1)') parser.add_argument('--version', action='version', version=VERSION) - # Spreadsheet-driven execution - parser.add_argument('--xlsx', type=str, help='Path to SCSEM XLSX with test listing (defaults to $SCSEM_XLSX_PATH if set)') - parser.add_argument('--sheet', type=str, help='Worksheet name to read (default: active sheet)') - parser.add_argument('--id-column', type=str, help='Header name of the column that contains test numbers/IDs') - parser.add_argument('--title-column', type=str, help='Header name of the column that contains the row title/section') + # Spreadsheet (locked defaults) + parser.add_argument('--xlsx', type=str, help='Path to SCSEM XLSX (defaults to $SCSEM_XLSX_PATH)') + parser.add_argument('--sheet', type=str, default=DEFAULT_SHEET, help=f'Worksheet name (default: "{DEFAULT_SHEET}")') + parser.add_argument('--id-column', type=str, default=DEFAULT_ID_COLUMN, help=f'ID column name (default: "{DEFAULT_ID_COLUMN}")') + parser.add_argument('--title-column', type=str, default=DEFAULT_TITLE_COLUMN, help=f'Title column name (default: "{DEFAULT_TITLE_COLUMN}")') # Targeted run controls - parser.add_argument('--exclude', type=str, default='32-38,44', - help='Comma-separated list/ranges of Test IDs to exclude (default: "32-38,44")') - parser.add_argument('--only', type=str, - help='Comma-separated list/ranges of Test IDs to run (e.g., "7" or "7-9,21"). If not set, runs all enabled IDs.') + parser.add_argument('--exclude', type=str, default='32-38,44', help='IDs to exclude (default: "32-38,44")') + parser.add_argument('--only', type=str, help='IDs to run (e.g., "7" or "7-9,21"). If not set, runs all enabled IDs.') - parser.add_argument('--mapping-json', type=str, - help='Optional JSON file mapping spreadsheet Test IDs (int) -> function name (e.g., "aws_07_check_password_policy")') + parser.add_argument('--mapping-json', type=str, help='Optional mapping JSON for spreadsheet ID -> function name') # Markdown export - parser.add_argument('--md-out-dir', type=str, help='Directory to write per-row Markdown files') - parser.add_argument('--md-combine-file', type=str, help='Path to a combined Markdown file (or a directory to write the combined file into)') - parser.add_argument('--md-include-excluded', action='store_true', - help='Include excluded rows in Markdown output (by default they are skipped).') - parser.add_argument('--md-front-matter', action='store_true', - help='Include YAML front matter in each per-row Markdown.') + parser.add_argument('--md-out-dir', type=str, help='Directory for per-row Markdown') + parser.add_argument('--md-combine-file', type=str, help='Path for combined Markdown (or directory to place it)') + parser.add_argument('--md-include-excluded', action='store_true', help='Include excluded rows in Markdown output') + parser.add_argument('--md-front-matter', action='store_true', help='Include YAML front matter in Markdown') + parser.add_argument('--md-only', action='store_true', help='Only generate Markdown; skip AWS checks') - parser.add_argument('--md-only', action='store_true', - help='Only generate Markdown docs from the spreadsheet; skip AWS checks.') + # Evidence pack + parser.add_argument('--evidence-dir', type=str, help='Base directory to write evidence JSON files') + parser.add_argument('--evidence-zip', type=str, help='Zip file path to bundle evidence (e.g., "./exports/evidence.zip")') - # Export directory for CSVs - parser.add_argument('--out-dir', type=str, - help='Directory to write CSV exports (summary.csv, results.csv). No CSVs if omitted.') + # Single account options (optional if using multi-account) + parser.add_argument('--profile', type=str, help='AWS CLI profile name') + parser.add_argument('--region', type=str, help='AWS region (e.g., us-gov-west-1)') + parser.add_argument('--alias', type=str, default='default', help='Alias for this account in outputs') - args = parser.parse_args() + # Multi-account + parser.add_argument('--accounts', type=str, help='Comma/range list: "profile:region@alias,profile2:region2@alias2"') + parser.add_argument('--accounts-file', type=str, help='JSON file with [{"profile": "...","region":"...","alias":"..."}]') - # Resolve XLSX path (arg or env var) - xlsx_path = args.xlsx or os.getenv('SCSEM_XLSX_PATH') + # CSV/JSON export + parser.add_argument('--out-dir', type=str, help='Directory to write CSV/JSON exports') - # Prepare exclusions, inclusions & mapping - excluded_ids: Set[int] = parse_exclusion_ranges(args.exclude) - only_ids: Set[int] = parse_id_list(args.only) + args = parser.parse_args() - mapping: Dict[int, str] = DEFAULT_SPREADSHEET_MAPPING.copy() - if args.mapping_json and os.path.exists(args.mapping_json): - with open(args.mapping_json, 'r') as f: - user_map = json.load(f) - mapping.update({int(k): v for k, v in user_map.items()}) + # Resolve XLSX path + xlsx_path = args.xlsx or os.getenv('SCSEM_XLSX_PATH') + if not xlsx_path and not args.md_only: + raise SystemExit("XLSX not specified. Set --xlsx or SCSEM_XLSX_PATH.") - # Markdown-only mode (no AWS calls) + # Markdown-only mode markdown_summary = None if (args.md_out_dir or args.md_combine_file) and xlsx_path: markdown_summary = write_markdown_files( @@ -1064,54 +1164,137 @@ def main(): id_column_name=args.id_column, title_column_name=args.title_column, out_dir=args.md_out_dir or os.path.join(os.getcwd(), "scsem_md"), - exclude_ids=excluded_ids, + exclude_ids=parse_ranges(args.exclude), include_excluded=args.md_include_excluded, combine_file=args.md_combine_file, add_front_matter=args.md_front_matter ) - if args.md_only: - print(json.dumps({ - 'version': VERSION, - 'markdown_summary': markdown_summary or {'note': 'No markdown was produced (missing --xlsx or openpyxl).'} - }, indent=2, default=str)) + print(json.dumps({'version': VERSION, 'markdown_summary': markdown_summary or {}}, indent=2, default=str)) return - # Build execution plan from spreadsheet (optional) + # Build plan from spreadsheet planned_ids: Optional[Set[int]] = None if xlsx_path: ids = load_spreadsheet_test_ids(xlsx_path, args.sheet, args.id_column) planned_ids = set(ids) - # Create session and run AWS checks - session = create_session(args.profile, args.region) - results, summary = run_automated_checks( - session=session, - region_name=args.region, - planned_spreadsheet_ids=planned_ids, - spreadsheet_mapping=mapping, - exclude_ids=excluded_ids, - only_ids=only_ids - ) - - # Enrich summary with spreadsheet totals - if planned_ids is not None: - summary['total_planned_from_spreadsheet'] = len(planned_ids) - summary['total_excluded'] = len({i for i in planned_ids if i in excluded_ids}) - summary['total_unmapped'] = len(summary['missing_implementation']) - - # Optional CSV exports - exports = None - if args.out_dir: - exports = write_summary_exports(args.out_dir, summary, results) - summary['csv_exports'] = exports + excluded_ids: Set[int] = parse_ranges(args.exclude) + only_ids: Set[int] = parse_ranges(args.only) + + # Mapping + mapping: Dict[int, str] = DEFAULT_SPREADSHEET_MAPPING.copy() + if args.mapping_json and os.path.exists(args.mapping_json): + with open(args.mapping_json, 'r') as f: + user_map = json.load(f) + mapping.update({int(k): v for k, v in user_map.items()}) + + # Build accounts list + accounts: List[Dict[str, str]] = [] + accounts.extend(parse_accounts_arg(args.accounts)) + accounts.extend(load_accounts_file(args.accounts_file)) + if not accounts and args.profile: + accounts.append({"profile": args.profile, "region": (args.region or ""), "alias": args.alias}) + + if not accounts: + raise SystemExit("No accounts provided. Use --profile/--region or --accounts/--accounts-file.") + + # Execute per account and aggregate + rollup_results: Dict[str, Any] = {'accounts': []} + overall_totals = { + 'executed_tests': 0, 'skipped_tests': 0, 'passed': 0, 'failed': 0, 'errors': 0 + } + + combined_results_rows: List[Dict[str, Any]] = [] # for combined CSV/JSON - # Print machine-readable output + for acct in accounts: + profile = acct.get("profile") or None + region = acct.get("region") or None + alias = acct.get("alias") or (profile or "account") + + session = create_session(profile, region) + # Discover account id via STS + try: + sts = session.client('sts') + ident = sts.get_caller_identity() + account_id = ident.get('Account', 'unknown') + except Exception: + account_id = "unknown" + + res, sumy = run_automated_checks_for_account( + session=session, + region_name=region, + account_alias=alias, + account_id=account_id, + planned_spreadsheet_ids=planned_ids, + spreadsheet_mapping=mapping, + exclude_ids=excluded_ids, + only_ids=only_ids, + evidence_base_dir=args.evidence_dir + ) + + # Export per-account CSV/JSON + if args.out_dir: + acct_dir = os.path.join(args.out_dir, _sanitize_filename(f"{alias}_{account_id}")) + exports = write_summary_exports(acct_dir, sumy, res, {"account_alias": alias, "account_id": account_id}) + sumy['csv_exports'] = exports + + # Update roll-up totals + for k in overall_totals.keys(): + overall_totals[k] += sumy.get(k, 0) + + # Append to combined rows with account context + for r in res: + r_with_acct = dict(r) + r_with_acct['account_alias'] = alias + r_with_acct['account_id'] = account_id + combined_results_rows.append(r_with_acct) + + rollup_results['accounts'].append({'alias': alias, 'account_id': account_id, 'summary': sumy}) + + # Bundle evidence if requested + if args.evidence_dir and args.evidence_zip: + os.makedirs(os.path.dirname(args.evidence_zip), exist_ok=True) + zip_path = zip_evidence(args.evidence_dir, args.evidence_zip) + rollup_results['evidence_zip'] = zip_path + + # Combined export + rollup_summary = { + 'version': VERSION, + 'timestamp': datetime.datetime.now(datetime.timezone.utc).isoformat(), + 'overall': overall_totals, + 'accounts_count': len(accounts) + } + if args.out_dir: + combined_dir = os.path.join(args.out_dir, "combined") + os.makedirs(combined_dir, exist_ok=True) + # Combined summary/results + with open(os.path.join(combined_dir, "rollup_summary.json"), "w", encoding="utf-8") as f: + json.dump({'summary': rollup_summary, 'accounts': rollup_results['accounts']}, f, indent=2, default=str) + # Combined CSV + fields = [ + "account_alias", "account_id", + "test_id", "section_title", "control_name", "nist_id", "test_method", + "status", "compliant", "reason", "error", "applicable_to_govcloud" + ] + combined_csv = os.path.join(combined_dir, "rollup_results.csv") + with open(combined_csv, "w", encoding="utf-8") as f: + f.write(",".join(fields) + "\n") + for r in combined_results_rows: + row = [] + for k in fields: + row.append(_safe_str(r.get(k, ""))) + f.write(",".join(row) + "\n") + rollup_results['combined_exports'] = {"rollup_summary_json": os.path.join(combined_dir, "rollup_summary.json"), + "rollup_results_csv": combined_csv} + + # Final print print(json.dumps({ 'version': VERSION, - 'summary': summary, - 'markdown_summary': markdown_summary, - 'results': results + 'rollup_summary': rollup_summary, + 'accounts': rollup_results['accounts'], + 'evidence_zip': rollup_results.get('evidence_zip'), + 'markdown_summary': markdown_summary }, indent=2, default=str))