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 de5cd2fe..19b90aa5 100755 --- a/local-app/python-tools/security-checks/aws_scsem_foundations.py +++ b/local-app/python-tools/security-checks/aws_scsem_foundations.py @@ -1,15 +1,16 @@ #!/usr/bin/env python3 """ AWS SCSEM Foundations Automated Checks -Version: 1.0.2 - -Enhancements in v1.0.2: -- Robust summary counters: executed, skipped, passed, failed, errors. +Version: 1.0.3 + +Enhancements in v1.0.3: +- Row-to-Markdown exporter: + * Per-row Markdown files: '# {ID} - {Title}' then '## {Column}' sections with values. + * Optional combined Markdown output that concatenates all rows with separators. + * Auto-detect ID/Title columns, or use --id-column/--title-column. + * Defaults to spreadsheet from $SCSEM_XLSX_PATH; auto-reads first sheet and first numeric ID column if unspecified. +- Robust summary counters remain: executed, skipped, passed, failed, errors. - Skipped breakdown: manual, inapplicable, excluded (32–38, 44 by default). -- CLI-driven exclusions and spreadsheet ingestion (XLSX), with auto-detection of sheet/ID column. -- Environment variable fallback for XLSX path: SCSEM_XLSX_PATH. -- Default mapping of SCSEM test numbers -> implemented check functions (IDs 1–16, 21). -- Region-aware EBS encryption check preserved. """ import argparse @@ -28,7 +29,7 @@ except ImportError: openpyxl = None -VERSION = "1.0.2" +VERSION = "1.0.3" # --------------------------- @@ -38,7 +39,6 @@ def create_session(profile_name: Optional[str] = None, region_name: Optional[str session_kwargs = {} if profile_name: session_kwargs['profile_name'] = profile_name - # boto3.Session accepts region_name; clients can override. return boto3.Session(region_name=region_name, **session_kwargs) @@ -351,7 +351,7 @@ def aws_15_check_full_admin_policies(session): 'nist_id': 'AC-3', 'control_name': 'Access Enforcement', 'test_method': 'Test (Manual)', - 'section_title': 'Ensure IAM policies that allow full \"*:*\" privileges are not attached', + 'section_title': 'Ensure IAM policies that allow full "*:*" privileges are not attached', 'applicable_to_govcloud': True, 'non_compliant_policies': non_compliant } @@ -415,7 +415,6 @@ def aws_21_check_ebs_encryption(session, region_name: str): ] # Default mapping from SCSEM Test Number -> function name -# (Assumes spreadsheet IDs align to these AWS-XX items; adjust as needed.) DEFAULT_SPREADSHEET_MAPPING: Dict[int, str] = { 1: "aws_01_check_account_contact", 2: "aws_02_check_security_contact", @@ -490,9 +489,22 @@ def evaluate_check_result(r: Dict[str, Any]) -> Tuple[str, Optional[str]]: # --------------------------- -# XLSX ingestion +# XLSX ingestion (IDs + full rows) # --------------------------- +def _detect_header_row(ws) -> Tuple[int, List[str]]: + header_row_idx = None + for i, row in enumerate(ws.iter_rows(min_row=1, max_row=10), start=1): + values = [str(cell.value).strip() if cell.value is not None else '' for cell in row] + if any(values): + header_row_idx = i + break + if header_row_idx is None: + raise RuntimeError("Unable to find header row in spreadsheet.") + 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], @@ -501,7 +513,7 @@ def load_spreadsheet_test_ids( """ Load test IDs (integers) from the XLSX sheet. - If id_column_name is provided, use that exact header. - - Otherwise, pick the first column that yields numeric-looking values. + - Otherwise, pick the column that yields the most numeric-like values. """ if not openpyxl: raise RuntimeError("openpyxl is required to read XLSX. Please install it or remove --xlsx.") @@ -511,21 +523,11 @@ def load_spreadsheet_test_ids( 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 - # Determine header row (first non-empty row) - header_row_idx = None - for i, row in enumerate(ws.iter_rows(min_row=1, max_row=10), start=1): - values = [str(cell.value).strip() if cell.value is not None else '' for cell in row] - if any(values): - header_row_idx = i - break - if header_row_idx is None: - raise RuntimeError("Unable to find header row in spreadsheet.") + header_row_idx, headers = _detect_header_row(ws) - headers = [str(cell.value).strip() if cell.value is not None else '' for cell in ws[header_row_idx]] + # Resolve target column target_col_idx = None - if id_column_name: - # Exact match preferred for idx, h in enumerate(headers): if h.strip().lower() == id_column_name.strip().lower(): target_col_idx = idx @@ -533,9 +535,7 @@ def load_spreadsheet_test_ids( if target_col_idx is None: raise RuntimeError(f"Column '{id_column_name}' not found. Headers: {headers}") else: - # Heuristics: look for first column that produces many numeric IDs candidate_indices = list(range(len(headers))) - target_col_idx = None numeric_hits: Dict[int, int] = {} for idx in candidate_indices: hits = 0 @@ -550,7 +550,6 @@ def load_spreadsheet_test_ids( if re.search(r'\d+', s): hits += 1 numeric_hits[idx] = hits - # Choose the column with the most numeric-like entries target_col_idx = max(numeric_hits, key=numeric_hits.get) if numeric_hits else 0 test_ids: List[int] = [] @@ -568,8 +567,209 @@ def load_spreadsheet_test_ids( 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 a 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 + + header_row_idx, headers = _detect_header_row(ws) + + # Resolve ID header + id_header = None + if id_column_name: + for h in headers: + if h.strip().lower() == id_column_name.strip().lower(): + id_header = h + break + if id_header is None: + raise RuntimeError(f"ID column '{id_column_name}' not found. Headers: {headers}") + else: + # heuristic: header containing 'test' and ('#' or 'id' or 'number') + 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: + # fallback to the first header + id_header = headers[0] + + # Resolve Title header + title_header = None + if title_column_name: + for h in headers: + if h.strip().lower() == title_column_name.strip().lower(): + title_header = h + break + if title_header is None: + raise RuntimeError(f"Title column '{title_column_name}' not found. Headers: {headers}") + else: + # heuristic candidates + 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: + # fallback to second header + title_header = headers[1] + + # Build rows + rows: List[Dict[str, Any]] = [] + for row in ws.iter_rows(min_row=header_row_idx + 1): + record: Dict[str, Any] = {} + for idx, h in enumerate(headers): + val = row[idx].value + record[h] = val if val is not None else '' + # discard completely empty rows + if any(str(v).strip() for v in record.values()): + rows.append(record) + + return headers, rows, id_header, title_header + + +# --------------------------- +# Markdown rendering +# --------------------------- + +def _stringify(value: Any) -> str: + if value is None: + return '' + if isinstance(value, (dict, list, tuple, set)): + return json.dumps(value, indent=2, default=str) + return str(value) + +def _sanitize_filename(name: str) -> str: + # Replace problematic chars, collapse whitespace to underscores + s = re.sub(r'[\\/*?:"<>|]+', '_', name) + s = re.sub(r'\s+', '_', s) + # Keep alphanum + _.- only + s = re.sub(r'[^A-Za-z0-9_.-]+', '_', s) + return s.strip('_')[:200] # avoid excessively long filenames + +def render_row_to_markdown( + row: Dict[str, Any], + headers: List[str], + id_header: str, + title_header: str +) -> Tuple[str, str]: + """ + Returns (markdown_text, suggested_filename) + - Title: '# {ID} - {Title}' + - Sections: each column header becomes '## {Header}' with value + """ + raw_id = _stringify(row.get(id_header, '')).strip() + raw_title = _stringify(row.get(title_header, '')).strip() + + # Derive printable ID (numeric piece if present) + id_for_title = raw_id + m = re.search(r'(\d+)', raw_id) + if m: + id_for_title = m.group(1) + + title_line = f"# {id_for_title} - {raw_title}".strip(" -") + body_lines = [title_line, ""] + for h in headers: + val = _stringify(row.get(h, '')).strip() + body_lines.append(f"## {h}") + body_lines.append("") + body_lines.append(val if val else "_(no value)_") + body_lines.append("") # blank line between sections + + markdown_text = "\n".join(body_lines).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" + 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] +) -> Dict[str, Any]: + """ + Processes the spreadsheet into per-row markdown files (and optional combined file). + Returns a summary dict with counts and paths. + """ + 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] = [] # list of IDs skipped due to exclusion + + combined_parts: List[str] = [] + + 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, headers, id_header, title_header) + path = os.path.join(out_dir, filename) + with open(path, "w", encoding="utf-8") as f: + f.write(md_text) + written_files.append(path) + + # Add a separator for combined output + combined_parts.append(md_text) + combined_parts.append("\n---\n") # page-like separator + + combined_path = None + if combine_file: + # If combine path points to a directory, write default name inside it + 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") + with open(combine_file, "w", encoding="utf-8") as f: + # Trim trailing separator + if combined_parts and combined_parts[-1].strip() == "---": + combined_parts = combined_parts[:-1] + f.write("\n".join(combined_parts).rstrip() + "\n") + combined_path = combine_file + + return { + "xlsx": xlsx_path, + "sheet": sheet_name or "(active)", + "id_header": id_header, + "title_header": title_header, + "out_dir": out_dir, + "written_count": len(written_files), + "written_files_sample": written_files[:10], # sample to keep JSON smaller + "skipped_rows_count": len(skipped_rows), + "skipped_rows_sample": skipped_rows[:10], + "combined_path": combined_path + } + + # --------------------------- -# Execution +# Execution (checks + docs) # --------------------------- def run_automated_checks( @@ -595,15 +795,12 @@ def run_automated_checks( manual_skips = 0 inapplicable_skips = 0 - # Helper to produce function lookup by name func_name_map: Dict[str, Callable] = {f.__name__: f for f in ALL_CHECKS} - # Determine which checks to run funcs_to_run: List[Callable] = [] missing_implementation: List[int] = [] excluded_tests_list: List[int] = [] - # Count exclusions as skipped up-front (if spreadsheet is provided) if planned_spreadsheet_ids is not None: excluded_tests_list = sorted(list(planned_spreadsheet_ids.intersection(exclude_ids))) skipped += len(excluded_tests_list) @@ -622,7 +819,6 @@ def run_automated_checks( continue funcs_to_run.append(func) else: - # Run all (no spreadsheet). Region-specific handled later. funcs_to_run = ALL_CHECKS total = len(funcs_to_run) + (1 if region_name else 0) @@ -640,7 +836,6 @@ def run_automated_checks( if status == 'info': skipped += 1 - # Attribute skip reason if not r.get('applicable_to_govcloud', True): inapplicable_skips += 1 elif r.get('test_method') == 'Test (Manual)': @@ -715,31 +910,62 @@ def main(): 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') 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('--mapping-json', type=str, help='Optional JSON file mapping spreadsheet Test IDs (int) -> function name (e.g., "aws_07_check_password_policy")') + # 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-only', action='store_true', + help='Only generate Markdown docs from the spreadsheet; skip AWS checks.') + args = parser.parse_args() - # Build execution plan from spreadsheet (optional) + # Resolve XLSX path (arg or env var) + xlsx_path = args.xlsx or os.getenv('SCSEM_XLSX_PATH') + + # Prepare exclusions & mapping excluded_ids: Set[int] = parse_exclusion_ranges(args.exclude) 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) - # Normalize keys to int mapping.update({int(k): v for k, v in user_map.items()}) - # Resolve XLSX path (arg or env var) - xlsx_path = args.xlsx or os.getenv('SCSEM_XLSX_PATH') + # Markdown-only mode (no AWS calls) + markdown_summary = None + if (args.md_out_dir or args.md_combine_file) and xlsx_path: + markdown_summary = write_markdown_files( + xlsx_path=xlsx_path, + sheet_name=args.sheet, + 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, + include_excluded=args.md_include_excluded, + combine_file=args.md_combine_file + ) + + 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)) + return + + # Build execution plan from spreadsheet (optional) 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 + # Create session and run AWS checks session = create_session(args.profile, args.region) results, summary = run_automated_checks( session=session, @@ -749,7 +975,7 @@ def main(): exclude_ids=excluded_ids ) - # If we have spreadsheet context, enrich summary with totals + # 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}) @@ -759,9 +985,11 @@ def main(): print(json.dumps({ 'version': VERSION, 'summary': summary, + 'markdown_summary': markdown_summary, 'results': results }, indent=2, default=str)) if __name__ == "__main__": main() +