From 4d5dc40bac3b9a6ae416249c10f2b7d4062cf54e Mon Sep 17 00:00:00 2001 From: badra001 Date: Wed, 6 May 2026 15:26:16 -0400 Subject: [PATCH] update --- .../python-tools/security-checks/README.md | 14 + .../security-checks/aws_scsem_foundations.py | 296 +++++++++++++----- 2 files changed, 224 insertions(+), 86 deletions(-) diff --git a/local-app/python-tools/security-checks/README.md b/local-app/python-tools/security-checks/README.md index 0ee58cd1..c4e4595b 100644 --- a/local-app/python-tools/security-checks/README.md +++ b/local-app/python-tools/security-checks/README.md @@ -90,3 +90,17 @@ If your spreadsheet uses different numbering, define it here (keys = spreadsheet - `excluded_tests`: explicit list of excluded test IDs. - Plus normal `passed`, `failed`, `errors` counters. + +python aws_scsem_foundations.py \ + --profile census-govcloud \ + --region us-gov-west-1 \ + --out-dir ./exports + +python aws_scsem_foundations.py \ + --md-out-dir ./scsem_md \ + --md-combine-file ./scsem_md/SCSEM_All.md \ + --md-front-matter \ + --sheet "Cloud Computing" \ + --id-column "Test Case Number" \ + --title-column "Section Title" + 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 19b90aa5..ca2e2ef2 100755 --- a/local-app/python-tools/security-checks/aws_scsem_foundations.py +++ b/local-app/python-tools/security-checks/aws_scsem_foundations.py @@ -1,16 +1,20 @@ #!/usr/bin/env python3 """ AWS SCSEM Foundations Automated Checks -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). +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. """ import argparse @@ -29,7 +33,7 @@ except ImportError: openpyxl = None -VERSION = "1.0.3" +VERSION = "1.0.4" # --------------------------- @@ -43,9 +47,8 @@ def create_session(profile_name: Optional[str] = None, region_name: Optional[str # --------------------------- -# Individual check functions +# Check functions (same as v1.0.3) # --------------------------- - def aws_01_check_account_contact(session): client = session.client('account') try: @@ -114,7 +117,7 @@ def aws_03_security_questions_registered(session): 'control_name': 'Incident Reporting', 'test_method': 'Test (Manual)', 'section_title': 'Ensure security questions are registered in the AWS account', - 'applicable_to_govcloud': False, # Not supported in GovCloud + 'applicable_to_govcloud': False, 'note': "Security questions are not supported in AWS GovCloud." } @@ -391,7 +394,7 @@ def aws_21_check_ebs_encryption(session, region_name: str): # --------------------------- -# Test registry and helpers +# Test registry and mapping # --------------------------- ALL_CHECKS: List[Callable[[boto3.session.Session], Dict[str, Any]]] = [ @@ -414,7 +417,6 @@ def aws_21_check_ebs_encryption(session, region_name: str): # aws_21 is region-specific, handled separately ] -# Default mapping from SCSEM Test Number -> function name DEFAULT_SPREADSHEET_MAPPING: Dict[int, str] = { 1: "aws_01_check_account_contact", 2: "aws_02_check_security_contact", @@ -436,10 +438,11 @@ def aws_21_check_ebs_encryption(session, region_name: str): } +# --------------------------- +# Utility parsing +# --------------------------- def parse_exclusion_ranges(ranges_text: str) -> Set[int]: - """ - Parse ranges like '32-38,44' into a set of integers. - """ + """Parse ranges like '32-38,44' into a set of integers.""" result: Set[int] = set() for part in (ranges_text or "").split(","): part = part.strip() @@ -460,14 +463,23 @@ 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 +# --------------------------- def evaluate_check_result(r: Dict[str, Any]) -> Tuple[str, Optional[str]]: """ Return ('pass'|'fail'|'info', reason_or_none). - Automated: 'compliant' True -> pass; False -> fail - Errors -> fail - - Manual or Not Applicable -> info (will be counted as skipped) - - If no explicit 'compliant', infer failure if a known non_compliant list is non-empty. + - Manual or Not Applicable -> info (counted as skipped) + - If no 'compliant', infer fail if common 'non_compliant_*' containers have values. """ method = r.get('test_method', '') applicable = r.get('applicable_to_govcloud', True) @@ -489,9 +501,8 @@ def evaluate_check_result(r: Dict[str, Any]) -> Tuple[str, Optional[str]]: # --------------------------- -# XLSX ingestion (IDs + full rows) +# XLSX ingestion (IDs + 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): @@ -504,7 +515,6 @@ 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], @@ -512,8 +522,8 @@ def load_spreadsheet_test_ids( ) -> List[int]: """ Load test IDs (integers) from the XLSX sheet. - - If id_column_name is provided, use that exact header. - - Otherwise, pick the column that yields the most numeric-like values. + - 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.") @@ -566,7 +576,6 @@ def load_spreadsheet_test_ids( test_ids.append(int(m.group(1))) return test_ids - def load_spreadsheet_rows( xlsx_path: str, sheet_name: Optional[str], @@ -574,7 +583,7 @@ def load_spreadsheet_rows( 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. + Load full rows into list of dicts keyed by column header. Returns: (headers, rows, resolved_id_header, resolved_title_header) """ if not openpyxl: @@ -597,14 +606,12 @@ def load_spreadsheet_rows( 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 @@ -617,14 +624,12 @@ def load_spreadsheet_rows( 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 @@ -634,7 +639,6 @@ def load_spreadsheet_rows( 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) @@ -644,7 +648,6 @@ def load_spreadsheet_rows( # --------------------------- # Markdown rendering # --------------------------- - def _stringify(value: Any) -> str: if value is None: return '' @@ -653,50 +656,65 @@ def _stringify(value: Any) -> 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 + return s.strip('_')[:200] def render_row_to_markdown( row: Dict[str, Any], headers: List[str], id_header: str, - title_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 """ 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, ""] + + parts: List[str] = [] + if add_front_matter: + fm = [ + "---", + f"id: \"{id_for_title}\"", + f"title: \"{raw_title}\"", + f"sheet: \"{sheet_name or '(active)'}\"", + 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() - body_lines.append(f"## {h}") - body_lines.append("") - body_lines.append(val if val else "_(no value)_") - body_lines.append("") # blank line between sections + parts.append(f"## {h}") + parts.append("") + parts.append(val if val else "_(no value)_") + parts.append("") - markdown_text = "\n".join(body_lines).rstrip() + "\n" + 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" return markdown_text, filename - def write_markdown_files( xlsx_path: str, sheet_name: Optional[str], @@ -705,12 +723,9 @@ def write_markdown_files( out_dir: str, exclude_ids: Set[int], include_excluded: bool, - combine_file: Optional[str] + combine_file: Optional[str], + add_front_matter: bool ) -> 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 ) @@ -718,10 +733,11 @@ def write_markdown_files( os.makedirs(out_dir, exist_ok=True) written_files: List[str] = [] - skipped_rows: List[str] = [] # list of IDs skipped due to exclusion - + skipped_rows: List[str] = [] combined_parts: List[str] = [] + sheet_label = sheet_name or "(active)" + for row in rows: raw_id = _stringify(row.get(id_header, '')).strip() m = re.search(r'(\d+)', raw_id) @@ -731,37 +747,42 @@ def write_markdown_files( skipped_rows.append(raw_id or "(no-id)") continue - md_text, filename = render_row_to_markdown(row, headers, id_header, title_header) + 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 + ) 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_parts.append("\n---\n") 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") + # 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: - # 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)", + "sheet": sheet_label, "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 + "written_files_sample": written_files[:10], "skipped_rows_count": len(skipped_rows), "skipped_rows_sample": skipped_rows[:10], "combined_path": combined_path @@ -769,18 +790,57 @@ def write_markdown_files( # --------------------------- -# Execution (checks + docs) +# Exports (CSV) # --------------------------- +def _safe_str(v: Any) -> str: + try: + if isinstance(v, (dict, list)): + return json.dumps(v, ensure_ascii=False) + return "" if v is None else str(v) + except Exception: + return str(v) +def write_summary_exports(out_dir: str, summary: Dict[str, Any], results: List[Dict[str, Any]]) -> Dict[str, str]: + os.makedirs(out_dir, exist_ok=True) + # 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" + ] + with open(results_csv, "w", encoding="utf-8") as f: + f.write(",".join(fields) + "\n") + for r in results: + row = [] + for k in fields: + row.append(_safe_str(r.get(k, ""))) + f.write(",".join(row) + "\n") + + return {"summary_csv": summary_csv, "results_csv": results_csv} + + +# --------------------------- +# Execution (checks + docs) +# --------------------------- def run_automated_checks( session, region_name: Optional[str], planned_spreadsheet_ids: Optional[Set[int]], spreadsheet_mapping: Dict[int, str], - exclude_ids: Set[int] + exclude_ids: Set[int], + only_ids: Set[int] ) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]: """ - Run automated checks driven by spreadsheet IDs and mapping (if provided). + Run automated checks driven by spreadsheet IDs and mapping. + - Applies 'exclude_ids' and 'only_ids'. Returns: (results_list, summary_counts_dict) """ results: List[Dict[str, Any]] = [] @@ -791,40 +851,56 @@ def run_automated_checks( failed = 0 errors = 0 - # Skipped breakdown manual_skips = 0 inapplicable_skips = 0 + executed_ids: List[int] = [] + skipped_ids: List[int] = [] + passed_ids: List[int] = [] + failed_ids: List[int] = [] + error_ids: List[int] = [] + func_name_map: Dict[str, Callable] = {f.__name__: f for f in ALL_CHECKS} - funcs_to_run: List[Callable] = [] + funcs_to_run: List[Tuple[Callable, Optional[int]]] = [] missing_implementation: List[int] = [] excluded_tests_list: List[int] = [] - if planned_spreadsheet_ids is not None: + # 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) - if planned_spreadsheet_ids is not None: - for tid in sorted(planned_spreadsheet_ids): - if tid in exclude_ids: - continue + # Build function list + 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: + for tid in sorted(selected_ids): func_name = spreadsheet_mapping.get(tid) - if not func_name: - missing_implementation.append(tid) - continue - func = func_name_map.get(func_name) - if not func: + 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 + skipped += 1 + skipped_ids.append(tid) continue - funcs_to_run.append(func) - else: - funcs_to_run = ALL_CHECKS + funcs_to_run.append((func_name_map[func_name], tid)) total = len(funcs_to_run) + (1 if region_name else 0) with tqdm(total=total, ncols=100) as pbar: - for idx, check in enumerate(funcs_to_run, 1): + for idx, (check, tid) in enumerate(funcs_to_run, 1): pbar.set_description(f"Test {idx}") r = check(session) status, reason = evaluate_check_result(r) @@ -834,8 +910,17 @@ def run_automated_checks( 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' + m = re.search(r'(\d+)', str(r.get('test_id', ''))) + current_id = int(m.group(1)) if m else None + if status == 'info': skipped += 1 + skipped_ids.append(current_id if current_id is not None else -1) if not r.get('applicable_to_govcloud', True): inapplicable_skips += 1 elif r.get('test_method') == 'Test (Manual)': @@ -843,11 +928,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) 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) if 'error' in r: errors += 1 + error_ids.append(current_id if current_id is not None else -1) pbar.set_postfix_str(f"{idx}/{total}") pbar.update(1) @@ -863,8 +953,12 @@ def run_automated_checks( r['reason'] = reason results.append(r) + m = re.search(r'(\d+)', str(r.get('test_id', ''))) + current_id = int(m.group(1)) if m else None + if status == 'info': skipped += 1 + skipped_ids.append(current_id if current_id is not None else -1) if not r.get('applicable_to_govcloud', True): inapplicable_skips += 1 elif r.get('test_method') == 'Test (Manual)': @@ -872,11 +966,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) 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) if 'error' in r: errors += 1 + error_ids.append(current_id if current_id is not None else -1) pbar.set_postfix_str(f"{idx}/{total}") pbar.update(1) @@ -895,7 +994,12 @@ def run_automated_checks( 'excluded': len(excluded_tests_list) }, 'missing_implementation': sorted(set(missing_implementation)), - 'excluded_tests': excluded_tests_list + 'excluded_tests': excluded_tests_list, + 'executed_ids': [x for x in executed_ids if x is not None], + 'skipped_ids': [x for x in skipped_ids if x is not None], + 'passed_ids': [x for x in passed_ids if x is not None], + 'failed_ids': [x for x in failed_ids if x is not None], + 'error_ids': [x for x in error_ids if x is not None], } return results, summary @@ -911,8 +1015,13 @@ def main(): 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') + + # 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('--mapping-json', type=str, help='Optional JSON file mapping spreadsheet Test IDs (int) -> function name (e.g., "aws_07_check_password_policy")') @@ -921,16 +1030,24 @@ def main(): 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-only', action='store_true', help='Only generate Markdown docs from the spreadsheet; skip AWS checks.') + # 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.') + args = parser.parse_args() # Resolve XLSX path (arg or env var) xlsx_path = args.xlsx or os.getenv('SCSEM_XLSX_PATH') - # Prepare exclusions & mapping + # Prepare exclusions, inclusions & mapping excluded_ids: Set[int] = parse_exclusion_ranges(args.exclude) + only_ids: Set[int] = parse_id_list(args.only) mapping: Dict[int, str] = DEFAULT_SPREADSHEET_MAPPING.copy() if args.mapping_json and os.path.exists(args.mapping_json): @@ -949,7 +1066,8 @@ def main(): 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 + combine_file=args.md_combine_file, + add_front_matter=args.md_front_matter ) if args.md_only: @@ -972,7 +1090,8 @@ def main(): region_name=args.region, planned_spreadsheet_ids=planned_ids, spreadsheet_mapping=mapping, - exclude_ids=excluded_ids + exclude_ids=excluded_ids, + only_ids=only_ids ) # Enrich summary with spreadsheet totals @@ -981,6 +1100,12 @@ def main(): 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 + # Print machine-readable output print(json.dumps({ 'version': VERSION, @@ -992,4 +1117,3 @@ def main(): if __name__ == "__main__": main() -