From bb00814b187a3a43f72986b5889e47dc062f7bbe Mon Sep 17 00:00:00 2001 From: edvinas Date: Thu, 18 Jun 2026 20:47:15 +0300 Subject: [PATCH] Initial commit --- .gitignore | 29 + README.md | 124 ++ cis_auditor/__init__.py | 1 + cis_auditor/benchmarks/__init__.py | 20 + cis_auditor/benchmarks/base.py | 212 +++ cis_auditor/benchmarks/debian/__init__.py | 49 + cis_auditor/benchmarks/debian/loader.py | 284 ++++ cis_auditor/benchmarks/nginx/__init__.py | 15 + cis_auditor/core/__init__.py | 1 + cis_auditor/core/executor.py | 85 + cis_auditor/core/models.py | 66 + cis_auditor/core/report.py | 277 ++++ main.py | 210 +++ nginx_audit/__init__.py | 1 + nginx_audit/checks.py | 1724 +++++++++++++++++++++ nginx_audit/executor.py | 85 + nginx_audit/models.py | 66 + nginx_audit/report.py | 277 ++++ nginx_audit/runner.py | 70 + requirements.txt | 2 + 20 files changed, 3598 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 cis_auditor/__init__.py create mode 100644 cis_auditor/benchmarks/__init__.py create mode 100644 cis_auditor/benchmarks/base.py create mode 100644 cis_auditor/benchmarks/debian/__init__.py create mode 100644 cis_auditor/benchmarks/debian/loader.py create mode 100644 cis_auditor/benchmarks/nginx/__init__.py create mode 100644 cis_auditor/core/__init__.py create mode 100644 cis_auditor/core/executor.py create mode 100644 cis_auditor/core/models.py create mode 100644 cis_auditor/core/report.py create mode 100644 main.py create mode 100644 nginx_audit/__init__.py create mode 100644 nginx_audit/checks.py create mode 100644 nginx_audit/executor.py create mode 100644 nginx_audit/models.py create mode 100644 nginx_audit/report.py create mode 100644 nginx_audit/runner.py create mode 100644 requirements.txt diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..8dc1fa2 --- /dev/null +++ b/.gitignore @@ -0,0 +1,29 @@ +# Python +__pycache__/ +*.py[cod] +*.pyo +*.pyd +.Python +*.egg-info/ +dist/ +build/ +.eggs/ +*.egg +.venv/ +venv/ +env/ +.env + +# CIS benchmark XLSX files — provide your own (not redistributable) +*.xlsx +*.xls + +# OS +.DS_Store +Thumbs.db + +# IDE +.vscode/ +.idea/ +*.swp +*.swo diff --git a/README.md b/README.md new file mode 100644 index 0000000..b67f037 --- /dev/null +++ b/README.md @@ -0,0 +1,124 @@ +# CIS Benchmark Auditor + +Automated security auditing tool for CIS Benchmarks. Runs checks against a +local machine or a remote host via SSH and produces a colour-coded XLSX report +showing passed, failed, and manual-review items with remediation guidance. + +## Supported benchmarks + +| Benchmark | Version | Checks | Implementation | +|-----------|---------|--------|----------------| +| CIS NGINX | v3.0.0 | 44 | Hand-coded Python functions | +| CIS Debian Linux 13 | v1.0.0 | 343 | Data-driven XLSX parser | + +## Requirements + +``` +pip install -r requirements.txt +``` + +`paramiko` is only needed for SSH mode. The rest of the standard library covers +local execution. + +## Usage + +### List available profiles + +```bash +python main.py nginx --list-profiles +python main.py debian --list-profiles --benchmark-file CIS_Debian_Linux_13_Benchmark_v1.0.0.xlsx +``` + +### Audit local machine + +```bash +python main.py nginx --profile "Level 1 - Webserver" + +python main.py debian --profile "Level 1 - Server" \ + --benchmark-file CIS_Debian_Linux_13_Benchmark_v1.0.0.xlsx +``` + +### Audit remote host via SSH + +```bash +python main.py debian \ + --mode ssh --host 10.0.0.5 --user admin --key ~/.ssh/id_rsa \ + --profile "Level 1 - Server" \ + --benchmark-file CIS_Debian_Linux_13_Benchmark_v1.0.0.xlsx +``` + +### Run specific checks only + +```bash +python main.py debian \ + --profile "Level 1 - Server" \ + --benchmark-file CIS_Debian_Linux_13_Benchmark_v1.0.0.xlsx \ + --checks 1.1.2.1.1,2.1.1,5.1.1 +``` + +### All options + +``` +positional: + benchmark nginx | debian + +options: + --benchmark-file XLSX Path to benchmark XLSX (required for: debian) + --mode {local,ssh} Execution mode (default: local) + --host HOST SSH host + --port PORT SSH port (default: 22) + --user USER SSH username (default: root) + --key KEY Path to SSH private key + --password PASSWORD SSH password + --profile PROFILE Profile to audit + --checks CHECKS Comma-separated check IDs + --list-profiles Print profiles and exit + --output XLSX Output report path (default: auto-named) + --verbose, -v Print check progress +``` + +## Report format + +The XLSX output contains four sheets: + +- **Summary** — total counts, timestamp, target host +- **Passed** — checks that passed with actual vs expected values +- **Failed** — checks that failed with actual vs expected values and remediation steps +- **Manual Review** — checks that require human judgment, with collected command output +- **Errors** (if any) — checks that could not be executed + +## Project structure + +``` +main.py # CLI entry point (all benchmarks) +requirements.txt +cis_auditor/ + core/ + models.py # CheckResult, AuditReport dataclasses + executor.py # LocalExecutor, SSHExecutor + report.py # XLSX report generator + benchmarks/ + base.py # CheckDefinition, BenchmarkInfo + nginx/ + __init__.py # Wraps hand-coded nginx_audit checks + debian/ + __init__.py # Lazy-load wrapper + loader.py # XLSX parser → CheckDefinition objects +nginx_audit/ # Hand-coded NGINX check functions (44 checks) + checks.py + executor.py + models.py + report.py + runner.py +``` + +## Adding a new benchmark + +1. Create `cis_auditor/benchmarks//` +2. Implement `loader.py` (data-driven from XLSX) or `checks.py` (hand-coded) +3. Expose `load(xlsx_path) -> BenchmarkInfo` in `__init__.py` +4. Add the name to `BENCHMARK_NAMES` in `main.py` + +The `CheckDefinition.run()` method handles all four pass-condition types +(`empty_output`, `non_empty_output`, `contains_pass`, `manual`) automatically, +so new benchmarks only need to supply the audit script and the right condition. diff --git a/cis_auditor/__init__.py b/cis_auditor/__init__.py new file mode 100644 index 0000000..7b03c04 --- /dev/null +++ b/cis_auditor/__init__.py @@ -0,0 +1 @@ +"""CIS Benchmark Auditor — multi-benchmark edition.""" diff --git a/cis_auditor/benchmarks/__init__.py b/cis_auditor/benchmarks/__init__.py new file mode 100644 index 0000000..22d1a30 --- /dev/null +++ b/cis_auditor/benchmarks/__init__.py @@ -0,0 +1,20 @@ +"""Benchmark registry — maps benchmark names to their check loaders.""" +from __future__ import annotations +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from ..benchmarks.base import BenchmarkInfo + +# Lazy imports to avoid loading all benchmarks at startup +def get_benchmark(name: str) -> "BenchmarkInfo": + name = name.lower() + if name == "nginx": + from .nginx import benchmark + return benchmark + elif name == "debian": + from .debian import benchmark + return benchmark + else: + raise ValueError(f"Unknown benchmark '{name}'. Available: nginx, debian") + +AVAILABLE = ["nginx", "debian"] diff --git a/cis_auditor/benchmarks/base.py b/cis_auditor/benchmarks/base.py new file mode 100644 index 0000000..08b42af --- /dev/null +++ b/cis_auditor/benchmarks/base.py @@ -0,0 +1,212 @@ +""" +Base classes shared by all benchmarks. + +BenchmarkInfo — metadata + profile/check registry for one benchmark +CheckDefinition — a single check: its audit script, pass condition, remediation + and a run() method that handles all evaluation logic + +Pass conditions +─────────────── +empty_output Most common. Command output empty → PASS, any output → FAIL. + Used by: grep -v checks (mount options, kernel params), most + "find bad things" style commands. + +non_empty_output Output present → PASS, empty → FAIL. + Used by: "verify X is installed/enabled" checks where the + command echoes a confirmation string. + +contains_pass Script explicitly prints "PASS" or "FAIL" in its output. + Used by: embedded CIS bash scripts with l_output/l_output2 + variables that evaluate multiple sub-conditions. + +manual Requires human judgment. Script is still run and output is + collected, but status is always MANUAL so the reviewer just + reads the data without re-running. +""" +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from typing import Callable, Dict, List, Optional + +from ..core.models import CheckResult, AuditReport, PASS, FAIL, MANUAL, ERROR +from ..core.executor import LocalExecutor, SSHExecutor + +Executor = LocalExecutor | SSHExecutor + + +# ── CheckDefinition ─────────────────────────────────────────────────────────── + +@dataclass +class CheckDefinition: + """ + A single benchmark check in data-driven form. + + The run() method handles all pass/fail logic so callers never need to + inspect audit text or implement evaluation themselves. + """ + check_id: str + title: str + profiles: List[str] + assessment_status: str # 'Automated' | 'Manual' + audit_script: str # bash command or script to execute + pass_condition: str # 'empty_output' | 'non_empty_output' | 'contains_pass' | 'manual' + expected_description: str # human-readable expected state + remediation: str + default_value: str = "" + notes: str = "" + + def run(self, executor: Executor) -> CheckResult: + """Execute this check and return a CheckResult.""" + + if not self.audit_script.strip() or self.pass_condition == "manual": + out, _ = executor.run(self.audit_script) if self.audit_script.strip() else ("", 0) + return CheckResult( + check_id=self.check_id, + title=self.title, + profiles=self.profiles, + status=MANUAL, + actual_value=out.strip() or "(manual check — no automatable command)", + expected_value=self.expected_description, + commands_run=[self.audit_script] if self.audit_script.strip() else [], + remediation=self.remediation, + notes=self.notes, + ) + + # ── run the script ──────────────────────────────────────────────────── + # Multi-line bash scripts need to be run via bash explicitly + script = self.audit_script.strip() + if "\n" in script or script.startswith("#!/"): + # Use heredoc so both local and SSH executors handle it correctly + cmd = f"bash <<'__SCRIPT_EOF__'\n{script}\n__SCRIPT_EOF__" + else: + cmd = script + + try: + out, exit_code = executor.run(cmd) + except Exception as exc: + return CheckResult( + check_id=self.check_id, + title=self.title, + profiles=self.profiles, + status=ERROR, + actual_value=f"Execution error: {exc}", + expected_value=self.expected_description, + commands_run=[script], + remediation=self.remediation, + ) + + out = out.strip() + + # ── evaluate pass condition ─────────────────────────────────────────── + if self.pass_condition == "empty_output": + status = PASS if not out else FAIL + actual = "(no issues found — compliant)" if not out else out + + elif self.pass_condition == "non_empty_output": + status = PASS if out else FAIL + actual = out if out else "(no output — expected confirmation of compliance)" + + elif self.pass_condition == "contains_pass": + # CIS bash scripts print lines like " - PASS: ..." or " - FAIL: ..." + has_pass = bool(re.search(r"\bPASS\b", out, re.IGNORECASE)) + has_fail = bool(re.search(r"\bFAIL\b", out, re.IGNORECASE)) + if has_pass and not has_fail: + status = PASS + elif has_fail: + status = FAIL + else: + # Script didn't emit PASS/FAIL — fall back to empty=PASS heuristic + status = PASS if not out else FAIL + actual = out or "(no output)" + + else: + status = MANUAL + actual = out or "(no output)" + + return CheckResult( + check_id=self.check_id, + title=self.title, + profiles=self.profiles, + status=status, + actual_value=actual, + expected_value=self.expected_description, + commands_run=[script], + remediation=self.remediation, + notes=self.notes, + ) + + +# ── BenchmarkInfo ───────────────────────────────────────────────────────────── + +@dataclass +class BenchmarkInfo: + """ + Describes one benchmark (e.g. NGINX, Debian 13). + + profiles ordered list of profile names + checks_by_profile profile → [check_id, ...] + check_registry check_id → CheckDefinition (for data-driven benchmarks) + OR check_id → Callable[[Executor], CheckResult] (for hand-coded) + """ + name: str + version: str + profiles: List[str] + checks_by_profile: Dict[str, List[str]] + check_registry: Dict[str, CheckDefinition | Callable] + + def run_profile( + self, + executor: Executor, + profile: str, + check_ids: Optional[List[str]] = None, + verbose: bool = False, + ) -> AuditReport: + """Run all (or a subset of) checks for *profile*.""" + if profile not in self.checks_by_profile: + raise ValueError( + f"Unknown profile '{profile}' for benchmark '{self.name}'. " + f"Valid: {self.profiles}" + ) + + ids = check_ids or self.checks_by_profile[profile] + report = AuditReport(target=executor.name, profile=f"{self.name} — {profile}") + total = len(ids) + + for i, cid in enumerate(ids, 1): + if verbose: + print(f" [{i:>3}/{total}] {cid:<12} …", end="\r", flush=True) + + entry = self.check_registry.get(cid) + if entry is None: + report.results.append(CheckResult( + check_id=cid, title=f"Check {cid}", profiles=[profile], + status=ERROR, + actual_value="No implementation found for this check ID", + expected_value="N/A", commands_run=[], remediation="", + notes="Missing from check_registry", + )) + continue + + try: + if callable(entry) and not isinstance(entry, CheckDefinition): + # hand-coded function: check_X_Y_Z(executor) -> CheckResult + result = entry(executor) + else: + # data-driven CheckDefinition + result = entry.run(executor) + except Exception as exc: + result = CheckResult( + check_id=cid, title=f"Check {cid}", profiles=[profile], + status=ERROR, + actual_value=f"Unexpected error: {exc}", + expected_value="N/A", commands_run=[], remediation="", + notes=str(exc), + ) + + report.results.append(result) + + if verbose: + print(" " * 60, end="\r") + + return report diff --git a/cis_auditor/benchmarks/debian/__init__.py b/cis_auditor/benchmarks/debian/__init__.py new file mode 100644 index 0000000..5f28929 --- /dev/null +++ b/cis_auditor/benchmarks/debian/__init__.py @@ -0,0 +1,49 @@ +""" +CIS Debian Linux 13 Benchmark v1.0.0. + +The benchmark object is built lazily — call debian.load(xlsx_path) before +running any checks. This avoids requiring the XLSX on systems that only +run (not build) the auditor. +""" +from __future__ import annotations + +from pathlib import Path +from typing import Optional + +from ..base import BenchmarkInfo + +_benchmark: Optional[BenchmarkInfo] = None +_loaded_from: Optional[Path] = None + + +def load(xlsx_path: str | Path) -> BenchmarkInfo: + """Parse the Debian benchmark XLSX and return (and cache) the BenchmarkInfo.""" + global _benchmark, _loaded_from + xlsx_path = Path(xlsx_path) + + if _benchmark is not None and _loaded_from == xlsx_path: + return _benchmark + + from .loader import load_benchmark + registry, by_profile, profiles = load_benchmark(xlsx_path) + + _benchmark = BenchmarkInfo( + name="CIS Debian Linux 13 Benchmark", + version="v1.0.0", + profiles=profiles, + checks_by_profile=by_profile, + check_registry=registry, + ) + _loaded_from = xlsx_path + return _benchmark + + +# Expose as module-level 'benchmark' only after load() is called +def __getattr__(name: str): + if name == "benchmark": + if _benchmark is None: + raise RuntimeError( + "Debian benchmark not loaded yet. Call debian.load(xlsx_path) first." + ) + return _benchmark + raise AttributeError(name) diff --git a/cis_auditor/benchmarks/debian/loader.py b/cis_auditor/benchmarks/debian/loader.py new file mode 100644 index 0000000..844eee2 --- /dev/null +++ b/cis_auditor/benchmarks/debian/loader.py @@ -0,0 +1,284 @@ +""" +Parse the CIS Debian Linux 13 Benchmark XLSX and produce CheckDefinition objects. + +Key insight +----------- +327 of 343 checks are "Automated" and the benchmark already embeds a runnable +bash script or command in the Audit Procedure column. Two patterns appear: + + A. CIS bash script (~100 checks) + Starts with #!/usr/bin/env bash, contains l_output/l_output2 variables, + and explicitly prints " - PASS: ..." or " - FAIL: ...". + -> pass_condition = 'contains_pass' + + B. Simple shell command (~227 checks) + One-liners like: + findmnt -kn /tmp | grep -v nodev + dpkg-query -s autofs && echo "autofs is installed" + Pass condition is inferred from surrounding prose (see _infer_pass_condition). + -> pass_condition = 'empty_output' | 'non_empty_output' + + C. Manual checks (16 checks) + Commands are extracted for reviewer convenience but status is always MANUAL. + -> pass_condition = 'manual' +""" +from __future__ import annotations + +import re +from collections import defaultdict +from pathlib import Path +from typing import Dict, List, Tuple + +import openpyxl + +from ..base import CheckDefinition + + +# ── script extraction ───────────────────────────────────────────────────────── + +_CODE_BLOCK_RE = re.compile(r"```(?:bash|shell|sh|)?\n?(.*?)```", re.DOTALL) + + +def _is_command_block(block: str) -> bool: + """Return True if the code block looks like a command/script rather than example output.""" + stripped = block.strip() + if not stripped: + return False + if stripped.startswith("#!/usr/bin/env bash") or stripped.startswith("#!/bin/bash"): + return True + if "l_output" in stripped: + return True + if stripped.startswith("#"): + if re.match(r"^#\s*(File:|Mode:|Access:|Owner:|Group:)", stripped): + return False + return True + if re.match( + r"^(find|stat|grep|awk|sed|cat|ls|dpkg|apt|systemctl|sysctl|" + r"findmnt|mount|chkconfig|service|ps|ss|netstat|ip|" + r"passwd|getent|useradd|usermod|chage|last|who|w|" + r"ufw|iptables|nft|firewall-cmd|modprobe|lsmod|" + r"openssl|ssh|curl|wget|auditctl|ausearch|aureport)", + stripped, + ): + return True + return False + + +def _extract_scripts(audit_text: str) -> List[str]: + """Extract all command/script code blocks from audit procedure markdown.""" + blocks = _CODE_BLOCK_RE.findall(audit_text) + return [b.strip() for b in blocks if _is_command_block(b.strip())] + + +def _looks_like_example_output(line: str) -> bool: + """Return True if a line looks like CIS example output rather than a command.""" + s = line.strip() + if not s: + return False + return bool(re.match( + r"^(" + r"[A-Z][a-z]+:.*|" + r"\d{4}-\d{2}-\d{2}.*|" + r"(enabled|disabled|generated|active|inactive|masked|running)\s*$|" + r"[a-z0-9._/-]+\s+\d{3,4}\s*$|" + r"-----BEGIN|" + r"[a-z]+\s+is\s+(installed|enabled|active|running)\s*$|" + r"-+\s*(OR|AND)\s*-+|" + r"nothing\s+is\s+returned\s*$|" + r"\(no output\)\s*$" + r")", + s, + re.IGNORECASE, + )) + + +def _clean_simple_command(raw: str) -> str: + """ + Strip leading '# ' comment prefix and remove trailing example-output lines. + + CIS XLSX often puts expected output in the same fenced block as the command, + separated by a blank line. Strategy: find the last line that is a real shell + command (not example output, not blank) and discard everything after it. + """ + lines = raw.strip().splitlines() + cleaned = [re.sub(r"^#\s*", "", line) for line in lines] + + last_cmd_idx = -1 + for i, line in enumerate(cleaned): + if line.strip() and not _looks_like_example_output(line): + last_cmd_idx = i + + if last_cmd_idx == -1: + return raw + + result_lines = cleaned[: last_cmd_idx + 1] + non_empty = [l for l in result_lines if l.strip()] + if not non_empty: + return raw + if len(non_empty) == 1: + return non_empty[0] + return "\n".join(non_empty) + + +# ── pass condition inference ────────────────────────────────────────────────── + +_EMPTY_PASS_PHRASES = [ + "nothing should be returned", + "no output", + "should return nothing", + "output should be empty", + "nothing is returned", + "should produce no output", + "should not return", + "if nothing", + "if no output", + "empty string", + "no files", + "no results", +] + +_NON_EMPTY_PASS_PHRASES = [ + "is installed", + "is enabled", + "is active", + "verify.*installed", + "confirm.*installed", + "should be installed", + "should return.*installed", + "output.*installed", + "output.*enabled", + "output.*active", +] + + +def _infer_pass_condition(audit_text: str, script: str) -> str: + """ + Determine pass condition for a check. + Returns one of: 'contains_pass', 'empty_output', 'non_empty_output', 'manual' + """ + is_bash = "#!/usr/bin/env bash" in script or "#!/bin/bash" in script + has_pass_fail_output = bool(re.search(r'"[^"]*\bPASS\b|[^"]*\bFAIL\b', script)) + if is_bash and ("l_output" in script or has_pass_fail_output): + return "contains_pass" + + text_lower = audit_text.lower() + + for phrase in _EMPTY_PASS_PHRASES: + if phrase in text_lower: + return "empty_output" + + for phrase in _NON_EMPTY_PASS_PHRASES: + if re.search(phrase, text_lower): + if re.search(r"(not installed|not enabled|not active|should not be|" + r"should be removed|should not be installed)", text_lower): + return "empty_output" + return "non_empty_output" + + if re.search(r"findmnt.*grep.*-v", script): + return "empty_output" + + if re.search(r"sysctl.*grep", script) or re.search(r"grep.*sysctl", script): + return "empty_output" + + if re.search(r"grep\s+-[a-zA-Z]*v", script): + return "empty_output" + + if re.match(r"stat\s+", script) and "|" not in script: + return "manual" + + return "empty_output" + + +def _extract_expected_description(audit_text: str) -> str: + """Pull a short expected-state description from the audit procedure text.""" + m = re.search( + r"(?:example output|expected output|verify.*?:)(.*?)(?:```|\n\n|\Z)", + audit_text, + re.IGNORECASE | re.DOTALL, + ) + if m: + snippet = m.group(1).strip()[:200] + if snippet: + return snippet + first = re.split(r"\.\s|\n", audit_text.strip())[0] + return first[:200] if first else "See audit procedure" + + +# ── main parser ─────────────────────────────────────────────────────────────── + +def load_benchmark(xlsx_path: str | Path) -> Tuple[ + Dict[str, CheckDefinition], + Dict[str, List[str]], + List[str], +]: + """ + Parse the Debian benchmark XLSX and return: + - check_registry: {check_id: CheckDefinition} + - checks_by_profile: {profile: [check_ids]} + - profiles: ordered list of profile names + """ + wb = openpyxl.load_workbook(xlsx_path, read_only=True, data_only=True) + ws = wb["Combined Profiles"] + + check_registry: Dict[str, CheckDefinition] = {} + checks_by_profile: Dict[str, List[str]] = defaultdict(list) + profile_order: List[str] = [] + + for row in ws.iter_rows(min_row=2, values_only=True): + check_id = row[1] + profile = row[2] + title = row[3] + status = row[4] + remediation = str(row[8]) if row[8] else "" + audit_text = str(row[9]) if row[9] else "" + default_val = str(row[25]) if len(row) > 25 and row[25] else "" + + if not check_id or not profile: + continue + + if profile not in profile_order: + profile_order.append(profile) + + checks_by_profile[profile].append(check_id) + + if check_id in check_registry: + if profile not in check_registry[check_id].profiles: + check_registry[check_id].profiles.append(profile) + continue + + # extract audit script + scripts = _extract_scripts(audit_text) + bash_scripts = [s for s in scripts if "l_output" in s or s.startswith("#!/")] + simple_cmds = [s for s in scripts if s not in bash_scripts] + + if bash_scripts: + primary_script = bash_scripts[0] + elif simple_cmds: + primary_script = _clean_simple_command(simple_cmds[0]) + else: + primary_script = "" + + # pass condition + if status == "Manual": + pass_cond = "manual" + elif not primary_script: + pass_cond = "manual" + else: + pass_cond = _infer_pass_condition(audit_text, primary_script) + + expected = _extract_expected_description(audit_text) + + check_registry[check_id] = CheckDefinition( + check_id=check_id, + title=title or f"Check {check_id}", + profiles=[profile], + assessment_status=status or "Manual", + audit_script=primary_script, + pass_condition=pass_cond, + expected_description=expected, + remediation=remediation, + default_value=default_val, + ) + + wb.close() + return check_registry, dict(checks_by_profile), profile_order diff --git a/cis_auditor/benchmarks/nginx/__init__.py b/cis_auditor/benchmarks/nginx/__init__.py new file mode 100644 index 0000000..3bfbdf5 --- /dev/null +++ b/cis_auditor/benchmarks/nginx/__init__.py @@ -0,0 +1,15 @@ +"""CIS NGINX Benchmark v3.0.0.""" +from __future__ import annotations +from ..base import BenchmarkInfo +# Import the hand-coded check functions and profile maps from the original module +import sys, os +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', '..')) +from nginx_audit.checks import ALL_CHECKS, PROFILE_CHECKS + +benchmark = BenchmarkInfo( + name="CIS NGINX Benchmark", + version="v3.0.0", + profiles=list(PROFILE_CHECKS.keys()), + checks_by_profile=PROFILE_CHECKS, + check_registry=ALL_CHECKS, # callable functions — BenchmarkInfo.run_profile handles both +) diff --git a/cis_auditor/core/__init__.py b/cis_auditor/core/__init__.py new file mode 100644 index 0000000..1b6e3d1 --- /dev/null +++ b/cis_auditor/core/__init__.py @@ -0,0 +1 @@ +"""Core shared components: models, executors, report generator.""" diff --git a/cis_auditor/core/executor.py b/cis_auditor/core/executor.py new file mode 100644 index 0000000..0ee1c5a --- /dev/null +++ b/cis_auditor/core/executor.py @@ -0,0 +1,85 @@ +"""Command executors: run shell commands locally or over SSH.""" +from __future__ import annotations + +import subprocess +from typing import Tuple + + +class ExecutorError(Exception): + pass + + +class LocalExecutor: + """Run shell commands on the local machine.""" + + name = "localhost" + + def run(self, cmd: str, timeout: int = 30) -> Tuple[str, int]: + """ + Execute *cmd* in a shell and return (combined_output, exit_code). + stdout and stderr are merged so callers get one string to inspect. + """ + try: + result = subprocess.run( + cmd, + shell=True, + capture_output=True, + text=True, + timeout=timeout, + ) + combined = (result.stdout + result.stderr).strip() + return combined, result.returncode + except subprocess.TimeoutExpired: + return f"[TIMEOUT after {timeout}s]", 1 + except Exception as exc: + return f"[ERROR: {exc}]", 1 + + def close(self) -> None: + pass # nothing to clean up + + +class SSHExecutor: + """Run shell commands on a remote host over SSH (requires paramiko).""" + + def __init__( + self, + host: str, + port: int = 22, + username: str = "root", + password: str | None = None, + key_file: str | None = None, + ) -> None: + try: + import paramiko # noqa: F401 + except ImportError as exc: + raise ExecutorError( + "paramiko is required for SSH mode: pip install paramiko" + ) from exc + + import paramiko + + self.name = host + self._client = paramiko.SSHClient() + self._client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + self._client.connect( + host, + port=port, + username=username, + password=password, + key_filename=key_file, + timeout=10, + ) + + def run(self, cmd: str, timeout: int = 30) -> Tuple[str, int]: + """Execute *cmd* on the remote host and return (combined_output, exit_code).""" + try: + _, stdout, stderr = self._client.exec_command(cmd, timeout=timeout) + exit_code = stdout.channel.recv_exit_status() + out = stdout.read().decode(errors="replace") + err = stderr.read().decode(errors="replace") + return (out + err).strip(), exit_code + except Exception as exc: + return f"[ERROR: {exc}]", 1 + + def close(self) -> None: + self._client.close() diff --git a/cis_auditor/core/models.py b/cis_auditor/core/models.py new file mode 100644 index 0000000..27409bf --- /dev/null +++ b/cis_auditor/core/models.py @@ -0,0 +1,66 @@ +"""Data models for the CIS NGINX benchmark auditor.""" +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime +from typing import List, Optional + + +# Possible check outcomes +PASS = "PASS" +FAIL = "FAIL" +MANUAL = "MANUAL" # check ran but result needs human judgment +ERROR = "ERROR" # could not run the check (command failed, not found, etc.) + + +@dataclass +class CheckResult: + """Result of running a single CIS check.""" + check_id: str + title: str + profiles: List[str] # which CIS profiles this check belongs to + status: str # PASS | FAIL | MANUAL | ERROR + actual_value: str # what was found on the system + expected_value: str # what the benchmark requires + commands_run: List[str] # shell commands executed + remediation: str # how to fix (from benchmark) + notes: str = "" # extra context or error detail + + +@dataclass +class AuditReport: + """Complete audit result for a target system.""" + target: str # hostname or "localhost" + profile: str # e.g. "Level 1 - Webserver" + timestamp: datetime = field(default_factory=datetime.utcnow) + results: List[CheckResult] = field(default_factory=list) + + # ── convenience aggregates ────────────────────────────────────────────── + + @property + def passed(self) -> List[CheckResult]: + return [r for r in self.results if r.status == PASS] + + @property + def failed(self) -> List[CheckResult]: + return [r for r in self.results if r.status == FAIL] + + @property + def manual(self) -> List[CheckResult]: + return [r for r in self.results if r.status == MANUAL] + + @property + def errors(self) -> List[CheckResult]: + return [r for r in self.results if r.status == ERROR] + + @property + def total(self) -> int: + return len(self.results) + + def summary(self) -> str: + return ( + f"Target: {self.target} | Profile: {self.profile} | " + f"Total: {self.total} | PASS: {len(self.passed)} | " + f"FAIL: {len(self.failed)} | MANUAL: {len(self.manual)} | " + f"ERROR: {len(self.errors)}" + ) diff --git a/cis_auditor/core/report.py b/cis_auditor/core/report.py new file mode 100644 index 0000000..63adf3b --- /dev/null +++ b/cis_auditor/core/report.py @@ -0,0 +1,277 @@ +""" +Generate an XLSX audit report with four sheets: + + 1. Summary – score, counts, target info + 2. Passed – checks that passed, actual vs expected + 3. Failed – checks that failed, actual vs expected, remediation + 4. Manual Review – checks requiring human judgment, command output + criteria + 5. Errors – checks that could not run + +Uses openpyxl (bundled with most Python environments). +""" +from __future__ import annotations + +from datetime import datetime +from pathlib import Path +from typing import List + +import openpyxl +from openpyxl.styles import ( + Alignment, Font, PatternFill, Border, Side +) +from openpyxl.utils import get_column_letter + +from .models import AuditReport, CheckResult, PASS, FAIL, MANUAL, ERROR + + +# ── colour palette ──────────────────────────────────────────────────────────── +C_HEADER_BG = "1F4E79" # dark blue – header row +C_HEADER_FG = "FFFFFF" # white +C_PASS_BG = "E2EFDA" # light green +C_FAIL_BG = "FCE4D6" # light red/orange +C_MANUAL_BG = "FFF2CC" # light yellow +C_ERROR_BG = "F2CEEF" # light purple +C_ALT_BG = "F5F5F5" # very light grey – alternating rows +C_SUMMARY_BG = "DEEAF1" # light blue – summary labels + +# thin border +_THIN = Side(style="thin", color="BFBFBF") +_BORDER = Border(left=_THIN, right=_THIN, top=_THIN, bottom=_THIN) + + +# ── helpers ─────────────────────────────────────────────────────────────────── + +def _header_font() -> Font: + return Font(name="Calibri", bold=True, color=C_HEADER_FG, size=11) + + +def _bold(color: str = "000000") -> Font: + return Font(name="Calibri", bold=True, color=color, size=11) + + +def _normal(size: int = 11) -> Font: + return Font(name="Calibri", size=size) + + +def _fill(hex_color: str) -> PatternFill: + return PatternFill("solid", fgColor=hex_color) + + +def _wrap_align() -> Alignment: + return Alignment(wrap_text=True, vertical="top") + + +def _set_header_row(ws, headers: list[str], row: int = 1) -> None: + for col, title in enumerate(headers, 1): + cell = ws.cell(row=row, column=col, value=title) + cell.font = _header_font() + cell.fill = _fill(C_HEADER_BG) + cell.alignment = Alignment(wrap_text=True, vertical="center", horizontal="center") + cell.border = _BORDER + + +def _write_row(ws, row: int, values: list, bg_color: str | None = None) -> None: + for col, val in enumerate(values, 1): + cell = ws.cell(row=row, column=col, value=str(val) if val is not None else "") + cell.font = _normal() + cell.alignment = _wrap_align() + cell.border = _BORDER + if bg_color: + cell.fill = _fill(bg_color) + elif row % 2 == 0: + cell.fill = _fill(C_ALT_BG) + + +def _auto_width(ws, min_w: int = 12, max_w: int = 60) -> None: + for col_cells in ws.columns: + length = max( + len(str(cell.value or "")) for cell in col_cells + ) + col_letter = get_column_letter(col_cells[0].column) + ws.column_dimensions[col_letter].width = max(min_w, min(length + 2, max_w)) + + +def _freeze(ws, cell: str = "A2") -> None: + ws.freeze_panes = cell + + +# ── sheet builders ──────────────────────────────────────────────────────────── + +def _build_summary(wb: openpyxl.Workbook, report: AuditReport) -> None: + ws = wb.active + ws.title = "Summary" + ws.sheet_view.showGridLines = False + + rows = [ + ("CIS NGINX Benchmark v3.0.0 – Audit Report", ""), + ("", ""), + ("Target", report.target), + ("Profile", report.profile), + ("Timestamp", report.timestamp.strftime("%Y-%m-%d %H:%M:%S UTC")), + ("", ""), + ("Total Checks", report.total), + ("✅ PASS", len(report.passed)), + ("❌ FAIL", len(report.failed)), + ("⚠️ MANUAL", len(report.manual)), + ("🔴 ERROR", len(report.errors)), + ("", ""), + ] + + if report.total: + pct_pass = round(len(report.passed) / report.total * 100, 1) + pct_fail = round(len(report.failed) / report.total * 100, 1) + rows.append(("Pass Rate", f"{pct_pass}%")) + rows.append(("Fail Rate", f"{pct_fail}%")) + + for r_idx, (label, value) in enumerate(rows, 1): + if r_idx == 1: + cell = ws.cell(row=r_idx, column=1, value=label) + cell.font = Font(name="Calibri", bold=True, size=16, color=C_HEADER_BG) + cell.alignment = Alignment(vertical="center") + ws.row_dimensions[r_idx].height = 30 + ws.merge_cells(start_row=r_idx, start_column=1, end_row=r_idx, end_column=2) + else: + lbl_cell = ws.cell(row=r_idx, column=1, value=label) + val_cell = ws.cell(row=r_idx, column=2, value=value) + if label and label not in ("",): + lbl_cell.font = _bold(C_HEADER_BG) + lbl_cell.fill = _fill(C_SUMMARY_BG) + val_cell.font = _normal() + + ws.column_dimensions["A"].width = 22 + ws.column_dimensions["B"].width = 40 + + +def _build_passed(wb: openpyxl.Workbook, results: list[CheckResult]) -> None: + ws = wb.create_sheet("Passed") + ws.sheet_view.showGridLines = False + headers = ["Check ID", "Title", "Commands Run", "Actual Value", "Expected Value"] + _set_header_row(ws, headers) + + for r_idx, res in enumerate(results, 2): + _write_row(ws, r_idx, [ + res.check_id, + res.title, + "\n".join(res.commands_run), + res.actual_value, + res.expected_value, + ], bg_color=C_PASS_BG) + ws.row_dimensions[r_idx].height = max( + 30, min(len(res.actual_value) // 3, 120) + ) + + _auto_width(ws) + _freeze(ws) + ws.column_dimensions["A"].width = 12 + ws.column_dimensions["B"].width = 52 + ws.column_dimensions["C"].width = 55 + ws.column_dimensions["D"].width = 55 + ws.column_dimensions["E"].width = 45 + + +def _build_failed(wb: openpyxl.Workbook, results: list[CheckResult]) -> None: + ws = wb.create_sheet("Failed") + ws.sheet_view.showGridLines = False + headers = [ + "Check ID", "Title", "Commands Run", + "Actual Value", "Expected Value", "Remediation Steps", "Notes" + ] + _set_header_row(ws, headers) + + for r_idx, res in enumerate(results, 2): + _write_row(ws, r_idx, [ + res.check_id, + res.title, + "\n".join(res.commands_run), + res.actual_value, + res.expected_value, + res.remediation, + res.notes, + ], bg_color=C_FAIL_BG) + ws.row_dimensions[r_idx].height = max( + 30, min(max(len(res.actual_value), len(res.remediation)) // 3, 150) + ) + + _auto_width(ws) + _freeze(ws) + ws.column_dimensions["A"].width = 12 + ws.column_dimensions["B"].width = 52 + ws.column_dimensions["C"].width = 55 + ws.column_dimensions["D"].width = 55 + ws.column_dimensions["E"].width = 45 + ws.column_dimensions["F"].width = 60 + ws.column_dimensions["G"].width = 40 + + +def _build_manual(wb: openpyxl.Workbook, results: list[CheckResult]) -> None: + ws = wb.create_sheet("Manual Review") + ws.sheet_view.showGridLines = False + headers = [ + "Check ID", "Title", "Commands Run", + "Command Output (Actual)", "What to Look For (Expected)", "Notes" + ] + _set_header_row(ws, headers) + + for r_idx, res in enumerate(results, 2): + _write_row(ws, r_idx, [ + res.check_id, + res.title, + "\n".join(res.commands_run), + res.actual_value, + res.expected_value, + res.notes, + ], bg_color=C_MANUAL_BG) + ws.row_dimensions[r_idx].height = max( + 30, min(len(res.actual_value) // 3, 150) + ) + + _auto_width(ws) + _freeze(ws) + ws.column_dimensions["A"].width = 12 + ws.column_dimensions["B"].width = 52 + ws.column_dimensions["C"].width = 55 + ws.column_dimensions["D"].width = 60 + ws.column_dimensions["E"].width = 55 + ws.column_dimensions["F"].width = 40 + + +def _build_errors(wb: openpyxl.Workbook, results: list[CheckResult]) -> None: + if not results: + return + ws = wb.create_sheet("Errors") + ws.sheet_view.showGridLines = False + headers = ["Check ID", "Title", "Commands Run", "Error Detail", "Notes"] + _set_header_row(ws, headers) + + for r_idx, res in enumerate(results, 2): + _write_row(ws, r_idx, [ + res.check_id, + res.title, + "\n".join(res.commands_run), + res.actual_value, + res.notes, + ], bg_color=C_ERROR_BG) + + _auto_width(ws) + _freeze(ws) + + +# ── public API ──────────────────────────────────────────────────────────────── + +def generate_report(report: AuditReport, output_path: str | Path) -> Path: + """ + Write the audit report as an XLSX file to *output_path*. + Returns the resolved Path. + """ + wb = openpyxl.Workbook() + + _build_summary(wb, report) + _build_passed(wb, report.passed) + _build_failed(wb, report.failed) + _build_manual(wb, report.manual) + _build_errors(wb, report.errors) + + path = Path(output_path) + path.parent.mkdir(parents=True, exist_ok=True) + wb.save(path) + return path diff --git a/main.py b/main.py new file mode 100644 index 0000000..29ee215 --- /dev/null +++ b/main.py @@ -0,0 +1,210 @@ +#!/usr/bin/env python3 +""" +CIS Benchmark Auditor — multi-benchmark CLI +============================================ +Supports: nginx (CIS NGINX Benchmark v3.0.0) + debian (CIS Debian Linux 13 Benchmark v1.0.0) + +Usage examples +-------------- +# List available profiles for a benchmark +python main_v2.py nginx --list-profiles +python main_v2.py debian --list-profiles --benchmark-file CIS_Debian_Linux_13_Benchmark_v1.0.0.xlsx + +# Audit local machine +python main_v2.py nginx --profile "Level 1 - Webserver" +python main_v2.py debian --profile "Level 1 - Server" \ + --benchmark-file CIS_Debian_Linux_13_Benchmark_v1.0.0.xlsx + +# Audit remote host via SSH +python main_v2.py debian \ + --mode ssh --host 10.0.0.5 --user admin --key ~/.ssh/id_rsa \ + --profile "Level 1 - Server" \ + --benchmark-file CIS_Debian_Linux_13_Benchmark_v1.0.0.xlsx + +# Run only specific checks +python main_v2.py debian --profile "Level 1 - Server" \ + --benchmark-file CIS_Debian_Linux_13_Benchmark_v1.0.0.xlsx \ + --checks 1.1.2.1.1,2.1.1,5.1.1 + +Adding a new benchmark +---------------------- +1. Create cis_auditor/benchmarks// +2. Implement loader.py (data-driven) or checks.py (hand-coded) +3. Expose a load(xlsx_path) -> BenchmarkInfo in __init__.py +4. Add the name to BENCHMARK_NAMES below +""" + +import argparse +import sys +from datetime import datetime +from pathlib import Path + +BENCHMARK_NAMES = ["nginx", "debian"] + + +def build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + description="CIS Benchmark Auditor", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=__doc__, + ) + p.add_argument( + "benchmark", + choices=BENCHMARK_NAMES, + help="Which benchmark to run", + ) + + # Benchmark file (required for data-driven benchmarks like debian) + p.add_argument( + "--benchmark-file", + metavar="XLSX", + help="Path to the benchmark XLSX file (required for: debian)", + ) + + # Execution mode + p.add_argument("--mode", choices=["local", "ssh"], default="local") + + # SSH options + ssh = p.add_argument_group("SSH options") + ssh.add_argument("--host") + ssh.add_argument("--port", type=int, default=22) + ssh.add_argument("--user", default="root") + ssh.add_argument("--key", help="Path to SSH private key") + ssh.add_argument("--password") + + # Scope + p.add_argument("--profile", metavar="PROFILE", + help="Profile to audit (use --list-profiles to see options)") + p.add_argument("--checks", + help="Comma-separated check IDs to run, e.g. 1.1.2.1.1,2.1.1") + p.add_argument("--list-profiles", action="store_true", + help="Print available profiles and exit") + + # Output + p.add_argument("--output", default=None, + help="Output XLSX path (default: auto-named)") + p.add_argument("--verbose", "-v", action="store_true") + + return p + + +def load_benchmark(name: str, xlsx_path: str | None): + """Load and return a BenchmarkInfo for *name*.""" + if name == "nginx": + from cis_auditor.benchmarks.nginx import benchmark + return benchmark + + elif name == "debian": + if not xlsx_path: + print("[!] --benchmark-file is required for the debian benchmark.", file=sys.stderr) + sys.exit(1) + from cis_auditor.benchmarks import debian + return debian.load(xlsx_path) + + else: + print(f"[!] Unknown benchmark '{name}'.", file=sys.stderr) + sys.exit(1) + + +def main() -> int: + parser = build_parser() + args = parser.parse_args() + + # ── load benchmark ──────────────────────────────────────────────────────── + bm = load_benchmark(args.benchmark, args.benchmark_file) + + # ── --list-profiles ─────────────────────────────────────────────────────── + if args.list_profiles: + print(f"\n{bm.name} {bm.version} — available profiles:\n") + for prof in bm.profiles: + n = len(bm.checks_by_profile.get(prof, [])) + print(f" {prof!r:<45} ({n} checks)") + print() + return 0 + + if not args.profile: + parser.error("--profile is required (use --list-profiles to see options)") + + if args.mode == "ssh" and not args.host: + parser.error("--host is required when --mode ssh") + + # ── build executor ──────────────────────────────────────────────────────── + from cis_auditor.core.executor import LocalExecutor, SSHExecutor, ExecutorError + try: + if args.mode == "local": + executor = LocalExecutor() + print(f"[*] Benchmark : {bm.name} {bm.version}") + print(f"[*] Mode : LOCAL") + else: + print(f"[*] Benchmark : {bm.name} {bm.version}") + print(f"[*] Mode : SSH → {args.user}@{args.host}:{args.port}") + executor = SSHExecutor( + host=args.host, port=args.port, username=args.user, + password=args.password, key_file=args.key, + ) + except ExecutorError as exc: + print(f"[!] {exc}", file=sys.stderr) + return 1 + except Exception as exc: + print(f"[!] Connection error: {exc}", file=sys.stderr) + return 1 + + # ── check subset ────────────────────────────────────────────────────────── + check_ids = None + if args.checks: + check_ids = [c.strip() for c in args.checks.split(",") if c.strip()] + + print(f"[*] Profile : {args.profile}") + total_checks = len(check_ids or bm.checks_by_profile.get(args.profile, [])) + print(f"[*] Checks : {total_checks}") + print(f"[*] Running audit …\n") + + # ── run audit ───────────────────────────────────────────────────────────── + try: + report = bm.run_profile( + executor=executor, + profile=args.profile, + check_ids=check_ids, + verbose=args.verbose, + ) + finally: + executor.close() + + # ── print summary ───────────────────────────────────────────────────────── + print(f"\n{'─' * 65}") + print(f" {report.summary()}") + print(f"{'─' * 65}") + + if report.failed: + print(f"\n Failed checks ({len(report.failed)}):") + for r in report.failed: + print(f" ✗ {r.check_id:<18} {r.title[:55]}") + + if report.manual: + print(f"\n Manual review required ({len(report.manual)}):") + for r in report.manual: + print(f" ⚠ {r.check_id:<18} {r.title[:55]}") + + if report.errors: + print(f"\n Errors ({len(report.errors)}):") + for r in report.errors: + print(f" ✗ {r.check_id:<18} {str(r.actual_value)[:60]}") + + # ── generate report ─────────────────────────────────────────────────────── + if args.output: + output_path = Path(args.output) + else: + safe_bm = args.benchmark + safe_prof = args.profile.replace(" ", "_").replace("-", "").replace("__", "_") + ts = datetime.utcnow().strftime("%Y%m%d_%H%M%S") + output_path = Path(f"cis_{safe_bm}_{safe_prof}_{ts}.xlsx") + + from cis_auditor.core.report import generate_report + generate_report(report, output_path) + print(f"\n[✓] Report saved to: {output_path.resolve()}\n") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/nginx_audit/__init__.py b/nginx_audit/__init__.py new file mode 100644 index 0000000..c575e43 --- /dev/null +++ b/nginx_audit/__init__.py @@ -0,0 +1 @@ +"""CIS NGINX Benchmark v3.0.0 Auditor.""" diff --git a/nginx_audit/checks.py b/nginx_audit/checks.py new file mode 100644 index 0000000..770ea58 --- /dev/null +++ b/nginx_audit/checks.py @@ -0,0 +1,1724 @@ +""" +CIS NGINX Benchmark v3.0.0 — all 44 check implementations. + +Each public function matches the pattern: + + def check_X_Y_Z(executor) -> CheckResult + +The function runs one or more shell commands via *executor*, evaluates the +output, and returns a CheckResult with status PASS / FAIL / MANUAL / ERROR. + +MANUAL is returned when the check requires human judgment (site-specific +expected values, ambiguous output, etc.). The command output is still +collected so the reviewer only has to read data, not re-run commands. +""" +from __future__ import annotations + +import re +from typing import List, Optional + +from .executor import LocalExecutor, SSHExecutor +from .models import CheckResult, PASS, FAIL, MANUAL, ERROR + +# Type alias used through the module +Executor = LocalExecutor | SSHExecutor + + +# ── helpers ────────────────────────────────────────────────────────────────── + +def _run(executor: Executor, cmd: str) -> tuple[str, int]: + return executor.run(cmd) + + +def _ok(check_id: str, title: str, profiles: List[str], + actual: str, expected: str, commands: List[str], + remediation: str, notes: str = "") -> CheckResult: + return CheckResult(check_id, title, profiles, PASS, + actual, expected, commands, remediation, notes) + + +def _fail(check_id: str, title: str, profiles: List[str], + actual: str, expected: str, commands: List[str], + remediation: str, notes: str = "") -> CheckResult: + return CheckResult(check_id, title, profiles, FAIL, + actual, expected, commands, remediation, notes) + + +def _manual(check_id: str, title: str, profiles: List[str], + actual: str, expected: str, commands: List[str], + remediation: str, notes: str = "") -> CheckResult: + return CheckResult(check_id, title, profiles, MANUAL, + actual, expected, commands, remediation, notes) + + +def _error(check_id: str, title: str, profiles: List[str], + msg: str, commands: List[str], remediation: str) -> CheckResult: + return CheckResult(check_id, title, profiles, ERROR, + msg, "N/A – check could not run", commands, remediation) + + +def _parse_timeout(value_str: str) -> Optional[int]: + """Convert NGINX timeout strings like '10', '10s', '1m' to seconds.""" + m = re.match(r"(\d+)\s*([smh]?)", value_str.strip().lower()) + if not m: + return None + n, unit = int(m.group(1)), m.group(2) + return n * {"s": 1, "m": 60, "h": 3600, "": 1}[unit] + + +def _octal_to_int(s: str) -> Optional[int]: + try: + return int(s, 8) + except (ValueError, TypeError): + return None + + +# ── Section 1 — Initial Setup ──────────────────────────────────────────────── + +PROFILES_L1_ALL = [ + "Level 1 - Webserver", "Level 1 - Proxy", "Level 1 - Loadbalancer" +] + +REMEDIATIONS = { + "1.1.1": ( + "Install or upgrade NGINX to version 1.28.0 or later from " + "https://nginx.org/en/linux_packages.html" + ), + "1.2.1": ( + "Configure package manager to point to the official NGINX repository " + "or your OS vendor's trusted repository." + ), + "1.2.2": ( + "Run the OS package manager update command (e.g. `dnf update nginx` " + "or `apt upgrade nginx`) to install the latest available version." + ), + "2.1.1": ( + "Remove unused dynamic modules from the NGINX configuration by deleting " + "or commenting out the relevant `load_module` directives." + ), + "2.2.1": ( + "Add `user nginx;` (or another dedicated non-root account) to nginx.conf " + "and ensure the account is not root." + ), + "2.2.2": ( + "Lock the NGINX service account: `passwd -l nginx` (RHEL) or " + "`usermod -L nginx` (Debian/Ubuntu)." + ), + "2.2.3": ( + "Set the NGINX service account shell to an invalid login shell: " + "`usermod -s /sbin/nologin nginx`." + ), + "2.3.1": ( + "Change ownership of all NGINX config files and directories to root: " + "`chown -R root:root /etc/nginx`." + ), + "2.3.2": ( + "Restrict permissions: directories to 750 or less, files to 640 or less. " + "Example: `find /etc/nginx -type d -exec chmod 750 {} +; " + "find /etc/nginx -type f -exec chmod 640 {} +`." + ), + "2.3.3": ( + "Set PID file ownership and permissions: " + "`chown root:root /run/nginx.pid && chmod 644 /run/nginx.pid`." + ), + "2.4.1": ( + "Review all `listen` directives and remove any that expose unauthorized ports." + ), + "2.4.2": ( + "Add a default_server block that rejects unknown host names:\n" + "server {\n listen 443 ssl default_server;\n ssl_reject_handshake on;\n}" + ), + "2.4.3": ( + "Add `keepalive_timeout 10;` to the http block in nginx.conf." + ), + "2.4.4": ( + "Add `send_timeout 10;` to the http or server block in nginx.conf." + ), + "2.5.1": ( + "Add `server_tokens off;` to the http block in nginx.conf." + ), + "2.5.2": ( + "Create custom error pages that do not contain any NGINX branding, " + "and reference them with `error_page 404 /custom_404.html;`." + ), + "2.5.3": ( + "Add a location block to deny access to hidden files:\n" + "location ~ /\\. { deny all; return 404; }" + ), + "2.5.4": ( + "Add `proxy_hide_header Server; proxy_hide_header X-Powered-By;` " + "to the server or location block." + ), + "3.1": ( + "Define a detailed log format and reference it in access_log:\n" + 'log_format main \'$remote_addr - $remote_user [$time_local] "$request" ...\'' + ), + "3.2": ( + "Ensure access_log is enabled and points to a valid log file:\n" + "`access_log /var/log/nginx/access.log main;`" + ), + "3.3": ( + "Set `error_log /var/log/nginx/error.log info;` in nginx.conf." + ), + "3.4": ( + "Add to your proxy location block:\n" + "proxy_set_header X-Real-IP $remote_addr;\n" + "proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;" + ), + "4.1.1": ( + "Add an HTTP server block that redirects to HTTPS:\n" + "server { listen 80; return 301 https://$host$request_uri; }" + ), + "4.1.2": ( + "Install a certificate from a trusted CA and set ssl_certificate / " + "ssl_certificate_key paths in your server block." + ), + "4.1.3": ( + "Restrict private key permissions: " + "`find /etc/nginx -name '*.key' -exec chmod 400 {} +`." + ), + "4.1.4": ( + "Set `ssl_protocols TLSv1.2 TLSv1.3;` and remove any older protocols." + ), + "4.1.5": ( + "Set a strong cipher suite, e.g.:\n" + "ssl_ciphers ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256:...;\n" + "ssl_prefer_server_ciphers on;" + ), + "4.1.6": ( + "Generate a strong DH parameter file:\n" + "`openssl dhparam -out /etc/nginx/dhparam.pem 4096`\n" + "Then add `ssl_dhparam /etc/nginx/dhparam.pem;` to nginx.conf." + ), + "4.1.7": ( + "Enable OCSP stapling:\n" + "ssl_stapling on;\nssl_stapling_verify on;\n" + "ssl_trusted_certificate /path/to/chain.pem;\nresolver 8.8.8.8 valid=300s;" + ), + "4.1.8": ( + "Add HSTS header:\n" + 'add_header Strict-Transport-Security "max-age=63072000; ' + 'includeSubDomains; preload" always;' + ), + "4.1.9": ( + "Add client certificate for upstream:\n" + "proxy_ssl_certificate /etc/nginx/client.crt;\n" + "proxy_ssl_certificate_key /etc/nginx/client.key;" + ), + "4.1.10": ( + "Enable upstream certificate verification:\n" + "proxy_ssl_verify on;\n" + "proxy_ssl_trusted_certificate /etc/nginx/trusted_ca.crt;" + ), + "4.1.11": ( + "Configure session resumption:\n" + "ssl_session_cache shared:SSL:10m;\n" + "ssl_session_timeout 1d;\n" + "ssl_session_tickets off;" + ), + "4.1.12": ( + "Enable HTTP/3 with QUIC:\n" + "listen 443 quic reuseport;\n" + 'add_header Alt-Svc \'h3=":443"; ma=86400\';' + ), + "5.1.1": ( + "Add allow/deny rules inside your server or location blocks to restrict " + "access to trusted IP ranges. Deny all other traffic by default." + ), + "5.1.2": ( + "Restrict allowed HTTP methods using limit_except or a request_method check:\n" + "limit_except GET HEAD POST { deny all; }" + ), + "5.2.1": ( + "Set client timeouts in the http block:\n" + "client_header_timeout 10;\nclient_body_timeout 10;" + ), + "5.2.2": ( + "Set a maximum request body size in the http block:\n" + "client_max_body_size 1m;" + ), + "5.2.3": ( + "Set large_client_header_buffers in the http block:\n" + "large_client_header_buffers 2 1k;" + ), + "5.2.4": ( + "Add connection limiting to the http block:\n" + "limit_conn_zone $binary_remote_addr zone=addr:10m;\n" + "limit_conn addr 10;" + ), + "5.2.5": ( + "Add rate limiting to the http block:\n" + "limit_req_zone $binary_remote_addr zone=req:10m rate=10r/s;\n" + "limit_req zone=req burst=20 nodelay;" + ), + "5.3.1": ( + 'Add `add_header X-Content-Type-Options "nosniff" always;` ' + "to the server block." + ), + "5.3.2": ( + "Add a Content-Security-Policy header appropriate to your application:\n" + 'add_header Content-Security-Policy "default-src \'self\';" always;' + ), + "5.3.3": ( + "Add a Referrer-Policy header:\n" + 'add_header Referrer-Policy "strict-origin-when-cross-origin" always;' + ), +} + + +# ── 1.1.1 ──────────────────────────────────────────────────────────────────── + +def check_1_1_1(executor: Executor) -> CheckResult: + cid, title = "1.1.1", "Ensure NGINX is installed" + profiles = PROFILES_L1_ALL + cmd = "nginx -V 2>&1" + out, _ = _run(executor, cmd) + + # Extract version string + m = re.search(r"nginx/(\d+)\.(\d+)\.(\d+)", out) + if not m: + return _fail(cid, title, profiles, + actual="NGINX not found or version unreadable", + expected="nginx/1.28.0 or later installed", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + major, minor, patch = int(m.group(1)), int(m.group(2)), int(m.group(3)) + version_str = f"nginx/{m.group(1)}.{m.group(2)}.{m.group(3)}" + min_version = (1, 28, 0) + + if (major, minor, patch) >= min_version: + return _ok(cid, title, profiles, + actual=version_str, + expected="nginx/1.28.0 or later", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + return _fail(cid, title, profiles, + actual=version_str, + expected="nginx/1.28.0 or later", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + +# ── 1.2.1 ──────────────────────────────────────────────────────────────────── + +def check_1_2_1(executor: Executor) -> CheckResult: + cid, title = "1.2.1", "Ensure package manager repositories are properly configured" + profiles = PROFILES_L1_ALL + # OS-agnostic: try both RPM and DEB + cmds = [ + "dnf repolist -v 2>/dev/null | grep -i nginx || true", + "apt-cache policy nginx 2>/dev/null || true", + ] + outputs = [] + for cmd in cmds: + out, _ = _run(executor, cmd) + if out: + outputs.append(out) + + combined = "\n".join(outputs).strip() or "(no output)" + return _manual(cid, title, profiles, + actual=combined, + expected="Repository URL points to nginx.org or OS vendor's trusted domain; repo is enabled", + commands=cmds, + remediation=REMEDIATIONS[cid], + notes="Verify the repo URL is trusted and the repo is enabled.") + + +# ── 1.2.2 ──────────────────────────────────────────────────────────────────── + +def check_1_2_2(executor: Executor) -> CheckResult: + cid, title = "1.2.2", "Ensure the latest software package is installed" + profiles = PROFILES_L1_ALL + cmds = [ + "dnf info nginx 2>/dev/null | grep -E '(Version|Available)' || true", + "apt-cache policy nginx 2>/dev/null | grep -E '(Installed|Candidate)' || true", + ] + outputs = [] + for cmd in cmds: + out, _ = _run(executor, cmd) + if out: + outputs.append(out) + + combined = "\n".join(outputs).strip() or "(no output)" + return _manual(cid, title, profiles, + actual=combined, + expected="Installed version matches the latest available version", + commands=cmds, + remediation=REMEDIATIONS[cid], + notes="Compare Installed vs Candidate/Available to confirm up-to-date.") + + +# ── 2.1.1 ──────────────────────────────────────────────────────────────────── + +def check_2_1_1(executor: Executor) -> CheckResult: + cid, title = "2.1.1", "Ensure only required dynamic modules are loaded" + profiles = PROFILES_L1_ALL + cmd = "nginx -T 2>/dev/null | grep 'load_module'" + out, _ = _run(executor, cmd) + + if not out.strip(): + return _ok(cid, title, profiles, + actual="No dynamic modules loaded", + expected="Only required dynamic modules loaded (none is acceptable)", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + return _manual(cid, title, profiles, + actual=out, + expected="Each listed module must be required for the application", + commands=[cmd], + remediation=REMEDIATIONS[cid], + notes="Review each load_module line and remove any that are not needed.") + + +# ── 2.2.1 ──────────────────────────────────────────────────────────────────── + +def check_2_2_1(executor: Executor) -> CheckResult: + cid, title = "2.2.1", "Ensure NGINX is run using a non-privileged, dedicated service account" + profiles = PROFILES_L1_ALL + cmd = "nginx -T 2>/dev/null | grep -i '^user'" + out, _ = _run(executor, cmd) + + if not out.strip(): + return _fail(cid, title, profiles, + actual="No 'user' directive found – NGINX may run as the launching user or nobody", + expected="user ;", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + # Extract user name from "user nginx;" or "user www-data nginx;" + m = re.match(r"user\s+(\S+)", out.strip()) + nginx_user = m.group(1).rstrip(";") if m else out.strip() + + if nginx_user.lower() in ("root", "0"): + return _fail(cid, title, profiles, + actual=f"user directive: {out.strip()} → running as root", + expected="Non-root dedicated service account", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + return _ok(cid, title, profiles, + actual=f"user directive: {out.strip()}", + expected="Non-root dedicated service account", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + +# ── 2.2.2 ──────────────────────────────────────────────────────────────────── + +def check_2_2_2(executor: Executor) -> CheckResult: + cid, title = "2.2.2", "Ensure the NGINX service account is locked" + profiles = PROFILES_L1_ALL + + cmd_user = "nginx -T 2>/dev/null | grep -i '^user'" + out_user, _ = _run(executor, cmd_user) + + m = re.match(r"user\s+(\S+)", out_user.strip()) + nginx_user = m.group(1).rstrip(";") if m else "nginx" + + cmd_lock = f"passwd -S {nginx_user} 2>&1" + out_lock, _ = _run(executor, cmd_lock) + commands = [cmd_user, cmd_lock] + + if not out_lock.strip() or "unknown" in out_lock.lower(): + return _error(cid, title, profiles, + msg=f"Could not determine lock status for '{nginx_user}': {out_lock}", + commands=commands, remediation=REMEDIATIONS[cid]) + + # RHEL shows "LK" or "L"; Debian/Ubuntu shows "L" field in a status line + locked = bool(re.search(r"\bL\b|\bLK\b", out_lock)) + status_label = f"passwd -S {nginx_user}: {out_lock}" + + if locked: + return _ok(cid, title, profiles, + actual=status_label, + expected=f"Account '{nginx_user}' is locked (L or LK in passwd -S output)", + commands=commands, + remediation=REMEDIATIONS[cid]) + else: + return _fail(cid, title, profiles, + actual=status_label, + expected=f"Account '{nginx_user}' is locked (L or LK in passwd -S output)", + commands=commands, + remediation=REMEDIATIONS[cid]) + + +# ── 2.2.3 ──────────────────────────────────────────────────────────────────── + +def check_2_2_3(executor: Executor) -> CheckResult: + cid, title = "2.2.3", "Ensure the NGINX service account has an invalid shell" + profiles = PROFILES_L1_ALL + + cmd_user = "nginx -T 2>/dev/null | grep -i '^user'" + out_user, _ = _run(executor, cmd_user) + m = re.match(r"user\s+(\S+)", out_user.strip()) + nginx_user = m.group(1).rstrip(";") if m else "nginx" + + cmd_shell = f"getent passwd {nginx_user} 2>&1" + out_shell, _ = _run(executor, cmd_shell) + commands = [cmd_user, cmd_shell] + + if not out_shell.strip(): + return _error(cid, title, profiles, + msg=f"User '{nginx_user}' not found in passwd database.", + commands=commands, remediation=REMEDIATIONS[cid]) + + # passwd entry: name:x:uid:gid:comment:home:shell + parts = out_shell.strip().split(":") + shell = parts[-1] if parts else "" + invalid_shells = ("/sbin/nologin", "/bin/false", "/usr/sbin/nologin", + "/dev/null", "/bin/nologin") + + if shell in invalid_shells: + return _ok(cid, title, profiles, + actual=f"Shell for '{nginx_user}': {shell}", + expected="Non-login shell (/sbin/nologin, /bin/false, etc.)", + commands=commands, + remediation=REMEDIATIONS[cid]) + else: + return _fail(cid, title, profiles, + actual=f"Shell for '{nginx_user}': {shell}", + expected="Non-login shell (/sbin/nologin, /bin/false, etc.)", + commands=commands, + remediation=REMEDIATIONS[cid]) + + +# ── 2.3.1 ──────────────────────────────────────────────────────────────────── + +def check_2_3_1(executor: Executor) -> CheckResult: + cid, title = "2.3.1", "Ensure NGINX directories and files are owned by root" + profiles = PROFILES_L1_ALL + cmd = ( + r"find /etc/nginx -name '*' \( -not -user root -o -not -group root \)" + r" -exec ls -ld {} \; 2>/dev/null" + ) + out, _ = _run(executor, cmd) + + if not out.strip(): + return _ok(cid, title, profiles, + actual="All files and directories in /etc/nginx are owned by root:root", + expected="All files owned by root:root", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + return _fail(cid, title, profiles, + actual=f"Files/dirs NOT owned by root:root:\n{out}", + expected="All files owned by root:root", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + +# ── 2.3.2 ──────────────────────────────────────────────────────────────────── + +def check_2_3_2(executor: Executor) -> CheckResult: + cid, title = "2.3.2", "Ensure access to NGINX directories and files is restricted" + profiles = PROFILES_L1_ALL + cmd_dirs = r'find /etc/nginx -type d -exec stat -Lc "%n %a" {} + 2>/dev/null' + cmd_files = r'find /etc/nginx -type f -exec stat -Lc "%n %a" {} + 2>/dev/null' + out_dirs, _ = _run(executor, cmd_dirs) + out_files, _ = _run(executor, cmd_files) + commands = [cmd_dirs, cmd_files] + + bad = [] + # Directories: max 750 (rwxr-x---) + for line in out_dirs.splitlines(): + parts = line.rsplit(" ", 1) + if len(parts) == 2: + perm = _octal_to_int(parts[1]) + if perm is not None and perm > 0o750: + bad.append(f"DIR perm={parts[1]}: {parts[0]}") + + # Files: max 640 (rw-r-----) + for line in out_files.splitlines(): + parts = line.rsplit(" ", 1) + if len(parts) == 2: + perm = _octal_to_int(parts[1]) + if perm is not None and perm > 0o640: + bad.append(f"FILE perm={parts[1]}: {parts[0]}") + + if not bad: + return _ok(cid, title, profiles, + actual="All directories ≤ 750, all files ≤ 640", + expected="Directories max 750, files max 640", + commands=commands, + remediation=REMEDIATIONS[cid]) + else: + return _fail(cid, title, profiles, + actual="Over-permissive entries found:\n" + "\n".join(bad), + expected="Directories max 750, files max 640", + commands=commands, + remediation=REMEDIATIONS[cid]) + + +# ── 2.3.3 ──────────────────────────────────────────────────────────────────── + +def check_2_3_3(executor: Executor) -> CheckResult: + cid, title = "2.3.3", "Ensure the NGINX process ID (PID) file is secured" + profiles = PROFILES_L1_ALL + cmds = [ + r'stat -Lc "%U:%G %a" /run/nginx.pid 2>/dev/null', + r'stat -Lc "%U:%G %a" /var/run/nginx.pid 2>/dev/null', + ] + out = "" + for cmd in cmds: + out, _ = _run(executor, cmd) + if out.strip(): + break + + if not out.strip(): + # PID file may not exist if NGINX is not running – mark as manual + return _manual(cid, title, profiles, + actual="PID file not found (NGINX may not be running)", + expected="root:root with permissions 644 or stricter", + commands=cmds, + remediation=REMEDIATIONS[cid], + notes="Start NGINX and re-run this check.") + + # Expected: owner root:root, perms 644 or stricter (<=644) + m = re.match(r"(\S+:\S+)\s+(\d+)", out.strip()) + if not m: + return _error(cid, title, profiles, + msg=f"Could not parse stat output: {out}", + commands=cmds, remediation=REMEDIATIONS[cid]) + + owner, perm_str = m.group(1), m.group(2) + perm = _octal_to_int(perm_str) + + issues = [] + if owner != "root:root": + issues.append(f"owner is {owner}, expected root:root") + if perm is not None and perm > 0o644: + issues.append(f"permissions {perm_str} are too permissive (max 644)") + + actual = f"owner={owner}, permissions={perm_str}" + if issues: + return _fail(cid, title, profiles, + actual=actual, + expected="root:root with permissions 644 or stricter", + commands=cmds, + remediation=REMEDIATIONS[cid]) + else: + return _ok(cid, title, profiles, + actual=actual, + expected="root:root with permissions 644 or stricter", + commands=cmds, + remediation=REMEDIATIONS[cid]) + + +# ── 2.4.1 ──────────────────────────────────────────────────────────────────── + +def check_2_4_1(executor: Executor) -> CheckResult: + cid, title = "2.4.1", "Ensure NGINX only listens for network connections on authorized ports" + profiles = PROFILES_L1_ALL + cmd = "nginx -T 2>/dev/null | grep -E '\\blisten\\b'" + out, _ = _run(executor, cmd) + + return _manual(cid, title, profiles, + actual=out.strip() or "(no listen directives found)", + expected="Only authorized ports (typically 80, 443) should be listed", + commands=[cmd], + remediation=REMEDIATIONS[cid], + notes="Review each listen directive and confirm all ports are authorized.") + + +# ── 2.4.2 ──────────────────────────────────────────────────────────────────── + +def check_2_4_2(executor: Executor) -> CheckResult: + cid, title = "2.4.2", "Ensure requests for unknown host names are rejected" + profiles = PROFILES_L1_ALL + cmd = "nginx -T 2>/dev/null | grep -Ei 'listen.*default_server|ssl_reject_handshake'" + out, _ = _run(executor, cmd) + + has_default = "default_server" in out.lower() + has_reject = "ssl_reject_handshake" in out.lower() + + if has_default or has_reject: + return _ok(cid, title, profiles, + actual=out.strip(), + expected="A default_server block with ssl_reject_handshake on is present", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + return _fail(cid, title, profiles, + actual=out.strip() or "(neither default_server nor ssl_reject_handshake found)", + expected="A default_server block with ssl_reject_handshake on is present", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + +# ── 2.4.3 ──────────────────────────────────────────────────────────────────── + +def check_2_4_3(executor: Executor) -> CheckResult: + cid, title = "2.4.3", "Ensure keepalive_timeout is 10 seconds or less, but not 0" + profiles = PROFILES_L1_ALL + cmd = "nginx -T 2>/dev/null | grep -i keepalive_timeout" + out, _ = _run(executor, cmd) + + if not out.strip(): + return _fail(cid, title, profiles, + actual="keepalive_timeout not set (default is 75s)", + expected="keepalive_timeout between 1 and 10 (seconds)", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + # Find value – take last occurrence (most specific context wins in NGINX) + values = re.findall(r"keepalive_timeout\s+(\S+)", out) + if not values: + return _error(cid, title, profiles, + msg=f"Could not parse keepalive_timeout value from: {out}", + commands=[cmd], remediation=REMEDIATIONS[cid]) + + raw = values[-1].rstrip(";") + seconds = _parse_timeout(raw) + + if seconds is None: + return _error(cid, title, profiles, + msg=f"Unrecognized timeout format: {raw}", + commands=[cmd], remediation=REMEDIATIONS[cid]) + + if 1 <= seconds <= 10: + return _ok(cid, title, profiles, + actual=f"keepalive_timeout = {raw} ({seconds}s)", + expected="1 to 10 seconds", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + return _fail(cid, title, profiles, + actual=f"keepalive_timeout = {raw} ({seconds}s)", + expected="1 to 10 seconds", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + +# ── 2.4.4 ──────────────────────────────────────────────────────────────────── + +def check_2_4_4(executor: Executor) -> CheckResult: + cid, title = "2.4.4", "Ensure send_timeout is set to 10 seconds or less, but not 0" + profiles = PROFILES_L1_ALL + cmd = "nginx -T 2>/dev/null | grep -i send_timeout" + out, _ = _run(executor, cmd) + + if not out.strip(): + return _fail(cid, title, profiles, + actual="send_timeout not set (default is 60s)", + expected="send_timeout between 1 and 10 seconds", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + values = re.findall(r"send_timeout\s+(\S+)", out) + if not values: + return _error(cid, title, profiles, + msg=f"Could not parse send_timeout value from: {out}", + commands=[cmd], remediation=REMEDIATIONS[cid]) + + raw = values[-1].rstrip(";") + seconds = _parse_timeout(raw) + + if seconds is None: + return _error(cid, title, profiles, + msg=f"Unrecognized timeout format: {raw}", + commands=[cmd], remediation=REMEDIATIONS[cid]) + + if 1 <= seconds <= 10: + return _ok(cid, title, profiles, + actual=f"send_timeout = {raw} ({seconds}s)", + expected="1 to 10 seconds", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + return _fail(cid, title, profiles, + actual=f"send_timeout = {raw} ({seconds}s)", + expected="1 to 10 seconds", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + +# ── 2.5.1 ──────────────────────────────────────────────────────────────────── + +def check_2_5_1(executor: Executor) -> CheckResult: + cid, title = "2.5.1", "Ensure server_tokens directive is set to `off`" + profiles = PROFILES_L1_ALL + cmd = "nginx -T 2>/dev/null | grep -i server_tokens" + out, _ = _run(executor, cmd) + + if not out.strip(): + return _fail(cid, title, profiles, + actual="server_tokens not set (default is 'on')", + expected="server_tokens off;", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + if re.search(r"server_tokens\s+off", out, re.IGNORECASE): + return _ok(cid, title, profiles, + actual=out.strip(), + expected="server_tokens off;", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + return _fail(cid, title, profiles, + actual=out.strip(), + expected="server_tokens off;", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + +# ── 2.5.2 ──────────────────────────────────────────────────────────────────── + +def check_2_5_2(executor: Executor) -> CheckResult: + cid, title = "2.5.2", "Ensure default error and index.html pages do not reference NGINX" + profiles = PROFILES_L1_ALL + cmd_pages = "nginx -T 2>/dev/null | grep -i error_page" + cmd_curl = "curl -sk http://127.0.0.1/non-existent-page 2>&1 | grep -i nginx | head -5" + out_pages, _ = _run(executor, cmd_pages) + out_curl, _ = _run(executor, cmd_curl) + commands = [cmd_pages, cmd_curl] + + if out_curl.strip(): + return _fail(cid, title, profiles, + actual=f"NGINX branding found in error page response:\n{out_curl}", + expected="Error pages contain no NGINX version information", + commands=commands, + remediation=REMEDIATIONS[cid]) + else: + return _ok(cid, title, profiles, + actual="No NGINX branding detected in 404 error response", + expected="Error pages contain no NGINX version information", + commands=commands, + remediation=REMEDIATIONS[cid]) + + +# ── 2.5.3 ──────────────────────────────────────────────────────────────────── + +def check_2_5_3(executor: Executor) -> CheckResult: + cid, title = "2.5.3", "Ensure hidden file serving is disabled" + profiles = PROFILES_L1_ALL + cmd = r"nginx -T 2>/dev/null | grep -A2 'location.*\\\\.'" + out, _ = _run(executor, cmd) + + if re.search(r"deny\s+all", out, re.IGNORECASE): + return _ok(cid, title, profiles, + actual=out.strip(), + expected="location ~/\\. block with 'deny all' present", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + return _fail(cid, title, profiles, + actual=out.strip() or "(no hidden-file location block found)", + expected="location ~/\\. block with 'deny all' present", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + +# ── 2.5.4 ──────────────────────────────────────────────────────────────────── + +PROFILES_PROXY_LB_L1 = ["Level 1 - Proxy", "Level 1 - Loadbalancer"] + + +def check_2_5_4(executor: Executor) -> CheckResult: + cid, title = "2.5.4", "Ensure the NGINX reverse proxy does not enable information disclosure" + profiles = PROFILES_PROXY_LB_L1 + cmd = "nginx -T 2>/dev/null | grep -Ei '(proxy|fastcgi)_hide_header'" + out, _ = _run(executor, cmd) + + hides_server = bool(re.search(r"hide_header\s+Server", out, re.IGNORECASE)) + hides_xpb = bool(re.search(r"hide_header\s+X-Powered-By", out, re.IGNORECASE)) + + if hides_server and hides_xpb: + return _ok(cid, title, profiles, + actual=out.strip(), + expected="proxy_hide_header Server and proxy_hide_header X-Powered-By both present", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + missing = [] + if not hides_server: + missing.append("proxy_hide_header Server") + if not hides_xpb: + missing.append("proxy_hide_header X-Powered-By") + return _fail(cid, title, profiles, + actual=out.strip() or "(no hide_header directives found)", + expected="proxy_hide_header Server and proxy_hide_header X-Powered-By both present", + commands=[cmd], + remediation=REMEDIATIONS[cid], + notes=f"Missing: {', '.join(missing)}") + + +# ── 3.1 ────────────────────────────────────────────────────────────────────── + +def check_3_1(executor: Executor) -> CheckResult: + cid, title = "3.1", "Ensure detailed logging is enabled" + profiles = PROFILES_L1_ALL + cmd_fmt = "nginx -T 2>/dev/null | grep -i log_format" + cmd_acc = "nginx -T 2>/dev/null | grep -i access_log" + out_fmt, _ = _run(executor, cmd_fmt) + out_acc, _ = _run(executor, cmd_acc) + commands = [cmd_fmt, cmd_acc] + + has_format = bool(out_fmt.strip()) + has_access = bool(out_acc.strip()) + + if has_format and has_access: + return _ok(cid, title, profiles, + actual=f"log_format defined: YES\naccess_log directive: {out_acc.strip()[:200]}", + expected="log_format defined and referenced in access_log", + commands=commands, + remediation=REMEDIATIONS[cid]) + else: + missing = [] + if not has_format: + missing.append("log_format not defined") + if not has_access: + missing.append("access_log not configured") + return _fail(cid, title, profiles, + actual="\n".join(missing), + expected="log_format defined and referenced in access_log", + commands=commands, + remediation=REMEDIATIONS[cid]) + + +# ── 3.2 ────────────────────────────────────────────────────────────────────── + +def check_3_2(executor: Executor) -> CheckResult: + cid, title = "3.2", "Ensure access logging is enabled" + profiles = PROFILES_L1_ALL + cmd = "nginx -T 2>/dev/null | grep -i access_log" + out, _ = _run(executor, cmd) + + if not out.strip(): + return _fail(cid, title, profiles, + actual="access_log directive not found", + expected="access_log enabled (not 'off')", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + if re.search(r"access_log\s+off", out, re.IGNORECASE): + return _fail(cid, title, profiles, + actual=out.strip(), + expected="access_log enabled (not 'off')", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + return _ok(cid, title, profiles, + actual=out.strip(), + expected="access_log enabled (not 'off')", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + +# ── 3.3 ────────────────────────────────────────────────────────────────────── + +def check_3_3(executor: Executor) -> CheckResult: + cid, title = "3.3", "Ensure error logging is enabled and set to the info logging level" + profiles = PROFILES_L1_ALL + cmd = "nginx -T 2>/dev/null | grep -i error_log" + out, _ = _run(executor, cmd) + + if not out.strip(): + return _fail(cid, title, profiles, + actual="error_log directive not found", + expected="error_log at 'info' level or more verbose", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + # Log levels from most verbose to least: debug, info, notice, warn, error, crit, alert, emerg + verbose_levels = {"debug", "info", "notice", "warn"} + m = re.search(r"error_log\s+\S+\s+(\w+)", out) + if m: + level = m.group(1).lower() + if level in verbose_levels: + return _ok(cid, title, profiles, + actual=out.strip(), + expected="error_log at 'info' level or more verbose", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + return _fail(cid, title, profiles, + actual=out.strip(), + expected="error_log at 'info' level or more verbose", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + # error_log with no level specified defaults to 'error' – not sufficient + return _fail(cid, title, profiles, + actual=out.strip() + " (no log level specified – defaults to 'error')", + expected="error_log at 'info' level or more verbose", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + +# ── 3.4 ────────────────────────────────────────────────────────────────────── + +def check_3_4(executor: Executor) -> CheckResult: + cid, title = "3.4", "Ensure proxies pass source IP information" + profiles = PROFILES_PROXY_LB_L1 + cmd = "nginx -T 2>/dev/null | grep -E 'proxy_set_header.*(X-Real-IP|X-Forwarded-For)'" + out, _ = _run(executor, cmd) + + has_real_ip = bool(re.search(r"X-Real-IP", out, re.IGNORECASE)) + has_forwarded = bool(re.search(r"X-Forwarded-For", out, re.IGNORECASE)) + + if has_real_ip and has_forwarded: + return _ok(cid, title, profiles, + actual=out.strip(), + expected="proxy_set_header X-Real-IP and X-Forwarded-For both set", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + missing = [] + if not has_real_ip: + missing.append("X-Real-IP") + if not has_forwarded: + missing.append("X-Forwarded-For") + return _fail(cid, title, profiles, + actual=out.strip() or "(no proxy_set_header for IP headers found)", + expected="proxy_set_header X-Real-IP and X-Forwarded-For both set", + commands=[cmd], + remediation=REMEDIATIONS[cid], + notes=f"Missing headers: {', '.join(missing)}") + + +# ── 4.1.1 ──────────────────────────────────────────────────────────────────── + +def check_4_1_1(executor: Executor) -> CheckResult: + cid, title = "4.1.1", "Ensure HTTP is redirected to HTTPS" + profiles = PROFILES_L1_ALL + cmd = "nginx -T 2>/dev/null | grep -A10 'listen 80'" + out, _ = _run(executor, cmd) + + if re.search(r"return\s+30[12]\s+https", out, re.IGNORECASE): + return _ok(cid, title, profiles, + actual=out.strip()[:400], + expected="HTTP server block contains 'return 301/302 https://' redirect", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + return _fail(cid, title, profiles, + actual=out.strip()[:400] or "(no listen 80 block or redirect found)", + expected="HTTP server block contains 'return 301/302 https://' redirect", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + +# ── 4.1.2 ──────────────────────────────────────────────────────────────────── + +def check_4_1_2(executor: Executor) -> CheckResult: + cid, title = "4.1.2", "Ensure a trusted certificate and trust chain is installed" + profiles = PROFILES_L1_ALL + cmd_cert = "grep -ir ssl_certificate /etc/nginx/ 2>/dev/null | grep -v ssl_certificate_key" + out_cert, _ = _run(executor, cmd_cert) + commands = [cmd_cert] + + return _manual(cid, title, profiles, + actual=out_cert.strip() or "(no ssl_certificate directive found)", + expected="ssl_certificate points to a valid cert from a trusted CA with complete chain", + commands=commands, + remediation=REMEDIATIONS[cid], + notes="Verify the certificate path, expiry date, and that the full chain is included.") + + +# ── 4.1.3 ──────────────────────────────────────────────────────────────────── + +def check_4_1_3(executor: Executor) -> CheckResult: + cid, title = "4.1.3", "Ensure private key permissions are restricted" + profiles = PROFILES_L1_ALL + cmd = r"find /etc/nginx/ -name '*.key' -exec stat -Lc '%n %a' {} + 2>/dev/null" + out, _ = _run(executor, cmd) + + if not out.strip(): + return _manual(cid, title, profiles, + actual="No .key files found under /etc/nginx/", + expected="Private key files with permissions 400 or 600", + commands=[cmd], + remediation=REMEDIATIONS[cid], + notes="Key files may be stored outside /etc/nginx. Verify manually.") + + bad = [] + for line in out.strip().splitlines(): + parts = line.rsplit(" ", 1) + if len(parts) == 2: + perm = _octal_to_int(parts[1]) + if perm is not None and perm not in (0o400, 0o600): + bad.append(f"perm={parts[1]}: {parts[0]}") + + if not bad: + return _ok(cid, title, profiles, + actual=out.strip(), + expected="Private key files with permissions 400 or 600", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + return _fail(cid, title, profiles, + actual="Incorrect permissions:\n" + "\n".join(bad), + expected="Private key files with permissions 400 or 600", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + +# ── 4.1.4 ──────────────────────────────────────────────────────────────────── + +def check_4_1_4(executor: Executor) -> CheckResult: + cid, title = "4.1.4", "Ensure only modern TLS protocols are used" + profiles = PROFILES_L1_ALL + cmd = "nginx -T 2>/dev/null | grep -i ssl_protocols" + out, _ = _run(executor, cmd) + + if not out.strip(): + return _fail(cid, title, profiles, + actual="ssl_protocols not set (NGINX default may include older protocols)", + expected="ssl_protocols TLSv1.2 TLSv1.3; (no SSLv2/3, TLSv1.0, TLSv1.1)", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + weak = re.findall(r"(SSLv2|SSLv3|TLSv1\.0|TLSv1\b(?!\.)|TLSv1\.1)", out, re.IGNORECASE) + if weak: + return _fail(cid, title, profiles, + actual=out.strip(), + expected="ssl_protocols TLSv1.2 TLSv1.3; only", + commands=[cmd], + remediation=REMEDIATIONS[cid], + notes=f"Weak/deprecated protocols found: {', '.join(set(weak))}") + else: + return _ok(cid, title, profiles, + actual=out.strip(), + expected="ssl_protocols TLSv1.2 TLSv1.3; only", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + +# ── 4.1.5 ──────────────────────────────────────────────────────────────────── + +WEAK_CIPHERS = [ + "RC4", "DES", "3DES", "MD5", "NULL", "EXPORT", "ANON", + "ADH", "AECDH", "SHA1", "RC2", "IDEA", +] + + +def check_4_1_5(executor: Executor) -> CheckResult: + cid, title = "4.1.5", "Disable weak ciphers" + profiles = PROFILES_L1_ALL + cmd = "nginx -T 2>/dev/null | grep -i ssl_ciphers" + out, _ = _run(executor, cmd) + + if not out.strip(): + return _manual(cid, title, profiles, + actual="ssl_ciphers not explicitly set (using OpenSSL defaults)", + expected="Strong cipher suite without RC4, DES, MD5, NULL, EXPORT, etc.", + commands=[cmd], + remediation=REMEDIATIONS[cid], + notes="Review the compiled OpenSSL defaults or explicitly set ssl_ciphers.") + + found_weak = [c for c in WEAK_CIPHERS if re.search(c, out, re.IGNORECASE)] + if found_weak: + return _fail(cid, title, profiles, + actual=out.strip(), + expected="No RC4, DES, 3DES, MD5, NULL, EXPORT, or ANON ciphers", + commands=[cmd], + remediation=REMEDIATIONS[cid], + notes=f"Weak cipher patterns found: {', '.join(found_weak)}") + else: + return _ok(cid, title, profiles, + actual=out.strip(), + expected="No RC4, DES, 3DES, MD5, NULL, EXPORT, or ANON ciphers", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + +# ── 4.1.6 ──────────────────────────────────────────────────────────────────── + +def check_4_1_6(executor: Executor) -> CheckResult: + cid, title = "4.1.6", "Ensure awareness of TLS 1.3 new Diffie-Hellman parameters" + profiles = PROFILES_L1_ALL + cmd = "nginx -T 2>/dev/null | grep -i ssl_dhparam" + out, _ = _run(executor, cmd) + + return _manual(cid, title, profiles, + actual=out.strip() or "(ssl_dhparam not set)", + expected="For TLS 1.2: ssl_dhparam pointing to a 2048-bit+ DH param file; " + "TLS 1.3 uses ephemeral key exchange (no dhparam needed)", + commands=[cmd], + remediation=REMEDIATIONS[cid], + notes="If TLS 1.2 is in use, ensure a strong DH param file is configured.") + + +# ── 4.1.7 ──────────────────────────────────────────────────────────────────── + +def check_4_1_7(executor: Executor) -> CheckResult: + cid, title = "4.1.7", "Ensure Online Certificate Status Protocol (OCSP) stapling is enabled" + profiles = PROFILES_L1_ALL + cmd = "nginx -T 2>/dev/null | grep -Ei '^\\s*(ssl_stapling|ssl_stapling_verify|ssl_trusted_certificate|resolver)'" + out, _ = _run(executor, cmd) + + has_stapling = bool(re.search(r"ssl_stapling\s+on", out, re.IGNORECASE)) + has_verify = bool(re.search(r"ssl_stapling_verify\s+on", out, re.IGNORECASE)) + has_resolver = bool(re.search(r"resolver\s+", out, re.IGNORECASE)) + + if has_stapling and has_verify and has_resolver: + return _ok(cid, title, profiles, + actual=out.strip(), + expected="ssl_stapling on; ssl_stapling_verify on; resolver configured", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + missing = [] + if not has_stapling: + missing.append("ssl_stapling on") + if not has_verify: + missing.append("ssl_stapling_verify on") + if not has_resolver: + missing.append("resolver") + return _fail(cid, title, profiles, + actual=out.strip() or "(none of the required OCSP directives found)", + expected="ssl_stapling on; ssl_stapling_verify on; resolver configured", + commands=[cmd], + remediation=REMEDIATIONS[cid], + notes=f"Missing: {', '.join(missing)}") + + +# ── 4.1.8 ──────────────────────────────────────────────────────────────────── + +MIN_HSTS_AGE = 63072000 # 2 years in seconds + + +def check_4_1_8(executor: Executor) -> CheckResult: + cid, title = "4.1.8", "Ensure HTTP Strict Transport Security (HSTS) is enabled" + profiles = PROFILES_L1_ALL + cmd = "nginx -T 2>/dev/null | grep -i Strict-Transport-Security" + out, _ = _run(executor, cmd) + + if not out.strip(): + return _fail(cid, title, profiles, + actual="HSTS header not configured", + expected=f'add_header Strict-Transport-Security "max-age={MIN_HSTS_AGE}; includeSubDomains" always;', + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + m = re.search(r"max-age=(\d+)", out, re.IGNORECASE) + if m: + age = int(m.group(1)) + if age >= MIN_HSTS_AGE: + return _ok(cid, title, profiles, + actual=out.strip(), + expected=f"max-age >= {MIN_HSTS_AGE} (2 years)", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + return _fail(cid, title, profiles, + actual=out.strip(), + expected=f"max-age >= {MIN_HSTS_AGE} (2 years)", + commands=[cmd], + remediation=REMEDIATIONS[cid], + notes=f"max-age={age} is below the recommended minimum of {MIN_HSTS_AGE}.") + else: + return _fail(cid, title, profiles, + actual=out.strip(), + expected=f"max-age >= {MIN_HSTS_AGE} (2 years)", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + +# ── 4.1.9 ──────────────────────────────────────────────────────────────────── + +def check_4_1_9(executor: Executor) -> CheckResult: + cid, title = "4.1.9", "Ensure upstream server traffic is authenticated with a client certificate" + profiles = PROFILES_PROXY_LB_L1 + cmd = "nginx -T 2>/dev/null | grep -E '^\\s*(proxy_ssl_certificate|proxy_ssl_certificate_key)'" + out, _ = _run(executor, cmd) + + has_cert = bool(re.search(r"proxy_ssl_certificate\s+", out)) + has_key = bool(re.search(r"proxy_ssl_certificate_key\s+", out)) + + if has_cert and has_key: + return _ok(cid, title, profiles, + actual=out.strip(), + expected="proxy_ssl_certificate and proxy_ssl_certificate_key both configured", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + missing = [] + if not has_cert: + missing.append("proxy_ssl_certificate") + if not has_key: + missing.append("proxy_ssl_certificate_key") + return _fail(cid, title, profiles, + actual=out.strip() or "(neither directive found)", + expected="proxy_ssl_certificate and proxy_ssl_certificate_key both configured", + commands=[cmd], + remediation=REMEDIATIONS[cid], + notes=f"Missing: {', '.join(missing)}") + + +# ── 4.1.10 ─────────────────────────────────────────────────────────────────── + +PROFILES_PROXY_LB_L2 = ["Level 2 - Proxy", "Level 2 - Loadbalancer"] + + +def check_4_1_10(executor: Executor) -> CheckResult: + cid, title = "4.1.10", "Ensure the upstream traffic server certificate is trusted" + profiles = PROFILES_PROXY_LB_L2 + cmd = "nginx -T 2>/dev/null | grep -E '^\\s*(proxy_ssl_verify|proxy_ssl_trusted_certificate|proxy_ssl_name)'" + out, _ = _run(executor, cmd) + + has_verify = bool(re.search(r"proxy_ssl_verify\s+on", out, re.IGNORECASE)) + has_trusted = bool(re.search(r"proxy_ssl_trusted_certificate\s+", out)) + + if has_verify and has_trusted: + return _ok(cid, title, profiles, + actual=out.strip(), + expected="proxy_ssl_verify on and proxy_ssl_trusted_certificate configured", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + missing = [] + if not has_verify: + missing.append("proxy_ssl_verify on") + if not has_trusted: + missing.append("proxy_ssl_trusted_certificate") + return _fail(cid, title, profiles, + actual=out.strip() or "(neither directive found)", + expected="proxy_ssl_verify on and proxy_ssl_trusted_certificate configured", + commands=[cmd], + remediation=REMEDIATIONS[cid], + notes=f"Missing: {', '.join(missing)}") + + +# ── 4.1.11 ─────────────────────────────────────────────────────────────────── + +def check_4_1_11(executor: Executor) -> CheckResult: + cid, title = "4.1.11", "Ensure Secure Session Resumption is Enabled" + profiles = PROFILES_L1_ALL + cmd = "nginx -T 2>/dev/null | grep -E '^\\s*(ssl_session_cache|ssl_session_timeout|ssl_session_tickets)'" + out, _ = _run(executor, cmd) + + has_cache = bool(re.search(r"ssl_session_cache\s+shared:", out, re.IGNORECASE)) + has_tickets_off = bool(re.search(r"ssl_session_tickets\s+off", out, re.IGNORECASE)) + has_timeout = bool(re.search(r"ssl_session_timeout\s+", out)) + + if has_cache and has_tickets_off and has_timeout: + return _ok(cid, title, profiles, + actual=out.strip(), + expected="ssl_session_cache shared:SSL:Xm; ssl_session_timeout set; ssl_session_tickets off", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + missing = [] + if not has_cache: + missing.append("ssl_session_cache shared:SSL:") + if not has_timeout: + missing.append("ssl_session_timeout") + if not has_tickets_off: + missing.append("ssl_session_tickets off") + return _fail(cid, title, profiles, + actual=out.strip() or "(no session resumption directives found)", + expected="ssl_session_cache shared:SSL:Xm; ssl_session_timeout set; ssl_session_tickets off", + commands=[cmd], + remediation=REMEDIATIONS[cid], + notes=f"Missing/incorrect: {', '.join(missing)}") + + +# ── 4.1.12 ─────────────────────────────────────────────────────────────────── + +PROFILES_WS_L2 = ["Level 2 - Webserver"] + + +def check_4_1_12(executor: Executor) -> CheckResult: + cid, title = "4.1.12", "Ensure HTTP/3.0 is used" + profiles = PROFILES_WS_L2 + cmd = "nginx -T 2>/dev/null | grep -E '^\\s*(listen.*quic|listen.*http3|add_header.*Alt-Svc)'" + out, _ = _run(executor, cmd) + + has_quic = bool(re.search(r"listen.*quic", out, re.IGNORECASE)) + has_alt_svc = bool(re.search(r"Alt-Svc", out, re.IGNORECASE)) + + if has_quic and has_alt_svc: + return _ok(cid, title, profiles, + actual=out.strip(), + expected="listen ... quic; and add_header Alt-Svc directive present", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + missing = [] + if not has_quic: + missing.append("listen 443 quic") + if not has_alt_svc: + missing.append("add_header Alt-Svc") + return _fail(cid, title, profiles, + actual=out.strip() or "(no HTTP/3 configuration found)", + expected="listen ... quic; and add_header Alt-Svc directive present", + commands=[cmd], + remediation=REMEDIATIONS[cid], + notes=f"Missing: {', '.join(missing)}") + + +# ── 5.1.1 ──────────────────────────────────────────────────────────────────── + +PROFILES_ALL_L2 = [ + "Level 2 - Webserver", "Level 2 - Proxy", "Level 2 - Loadbalancer" +] + + +def check_5_1_1(executor: Executor) -> CheckResult: + cid, title = "5.1.1", "Ensure allow and deny filters limit access to specific IP addresses" + profiles = PROFILES_ALL_L2 + cmd = "nginx -T 2>/dev/null | grep -E '^\\s*(allow|deny)'" + out, _ = _run(executor, cmd) + + return _manual(cid, title, profiles, + actual=out.strip() or "(no allow/deny directives found)", + expected="allow/deny rules restricting access to trusted IP ranges; 'deny all' as default", + commands=[cmd], + remediation=REMEDIATIONS[cid], + notes="Verify that IP ranges are appropriately restricted for your environment.") + + +# ── 5.1.2 ──────────────────────────────────────────────────────────────────── + +def check_5_1_2(executor: Executor) -> CheckResult: + cid, title = "5.1.2", "Ensure only approved HTTP methods are allowed" + profiles = PROFILES_L1_ALL + cmd = "nginx -T 2>/dev/null | grep -E '(\\$request_method|limit_except)'" + out, _ = _run(executor, cmd) + + if out.strip(): + return _ok(cid, title, profiles, + actual=out.strip(), + expected="HTTP method restriction via limit_except or $request_method conditional", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + return _fail(cid, title, profiles, + actual="(no HTTP method restriction found)", + expected="HTTP method restriction via limit_except or $request_method conditional", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + +# ── 5.2.1 ──────────────────────────────────────────────────────────────────── + +def check_5_2_1(executor: Executor) -> CheckResult: + cid, title = "5.2.1", "Ensure timeout values for reading the client header and body are set correctly" + profiles = PROFILES_L1_ALL + cmd = "nginx -T 2>/dev/null | grep -E '^\\s*(client_header_timeout|client_body_timeout)'" + out, _ = _run(executor, cmd) + + if not out.strip(): + return _fail(cid, title, profiles, + actual="Neither client_header_timeout nor client_body_timeout is set (defaults: 60s each)", + expected="client_header_timeout and client_body_timeout both set to 10s or less", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + bad = [] + for line in out.splitlines(): + m = re.match(r"\s*(client_(?:header|body)_timeout)\s+(\S+)", line) + if m: + directive, raw = m.group(1), m.group(2).rstrip(";") + secs = _parse_timeout(raw) + if secs is not None and secs > 10: + bad.append(f"{directive} = {raw} ({secs}s) — exceeds 10s") + + if not bad: + return _ok(cid, title, profiles, + actual=out.strip(), + expected="client_header_timeout and client_body_timeout ≤ 10s", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + return _fail(cid, title, profiles, + actual="\n".join(bad), + expected="client_header_timeout and client_body_timeout ≤ 10s", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + +# ── 5.2.2 ──────────────────────────────────────────────────────────────────── + +def check_5_2_2(executor: Executor) -> CheckResult: + cid, title = "5.2.2", "Ensure the maximum request body size is set correctly" + profiles = PROFILES_L1_ALL + cmd = "nginx -T 2>/dev/null | grep -E '^\\s*client_max_body_size'" + out, _ = _run(executor, cmd) + + if not out.strip(): + return _fail(cid, title, profiles, + actual="client_max_body_size not set (default: 1m; may allow large uploads)", + expected="client_max_body_size explicitly configured", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + return _ok(cid, title, profiles, + actual=out.strip(), + expected="client_max_body_size explicitly configured", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + +# ── 5.2.3 ──────────────────────────────────────────────────────────────────── + +def check_5_2_3(executor: Executor) -> CheckResult: + cid, title = "5.2.3", "Ensure the maximum buffer size for URIs is defined" + profiles = PROFILES_L1_ALL + cmd = "nginx -T 2>/dev/null | grep large_client_header_buffers" + out, _ = _run(executor, cmd) + + if not out.strip(): + return _fail(cid, title, profiles, + actual="large_client_header_buffers not set (default: 4 8k)", + expected="large_client_header_buffers explicitly configured (e.g. 2 1k)", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + return _ok(cid, title, profiles, + actual=out.strip(), + expected="large_client_header_buffers explicitly configured", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + +# ── 5.2.4 ──────────────────────────────────────────────────────────────────── + +def check_5_2_4(executor: Executor) -> CheckResult: + cid, title = "5.2.4", "Ensure the number of connections per IP address is limited" + profiles = PROFILES_ALL_L2 + cmd = "nginx -T 2>/dev/null | grep -E '^\\s*(limit_conn_zone|limit_conn)'" + out, _ = _run(executor, cmd) + + has_zone = bool(re.search(r"limit_conn_zone\s+", out)) + has_conn = bool(re.search(r"\blimit_conn\s+", out)) + + if has_zone and has_conn: + return _ok(cid, title, profiles, + actual=out.strip(), + expected="limit_conn_zone and limit_conn both configured", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + missing = [] + if not has_zone: + missing.append("limit_conn_zone") + if not has_conn: + missing.append("limit_conn") + return _fail(cid, title, profiles, + actual=out.strip() or "(neither limit_conn_zone nor limit_conn found)", + expected="limit_conn_zone and limit_conn both configured", + commands=[cmd], + remediation=REMEDIATIONS[cid], + notes=f"Missing: {', '.join(missing)}") + + +# ── 5.2.5 ──────────────────────────────────────────────────────────────────── + +def check_5_2_5(executor: Executor) -> CheckResult: + cid, title = "5.2.5", "Ensure rate limits by IP address are set" + profiles = PROFILES_ALL_L2 + cmd = "nginx -T 2>/dev/null | grep -E '^\\s*(limit_req_zone|limit_req)'" + out, _ = _run(executor, cmd) + + has_zone = bool(re.search(r"limit_req_zone\s+", out)) + has_req = bool(re.search(r"\blimit_req\s+", out)) + + if has_zone and has_req: + return _ok(cid, title, profiles, + actual=out.strip(), + expected="limit_req_zone and limit_req both configured", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + missing = [] + if not has_zone: + missing.append("limit_req_zone") + if not has_req: + missing.append("limit_req") + return _fail(cid, title, profiles, + actual=out.strip() or "(neither limit_req_zone nor limit_req found)", + expected="limit_req_zone and limit_req both configured", + commands=[cmd], + remediation=REMEDIATIONS[cid], + notes=f"Missing: {', '.join(missing)}") + + +# ── 5.3.1 ──────────────────────────────────────────────────────────────────── + +PROFILES_WS_L1 = ["Level 1 - Webserver"] + + +def check_5_3_1(executor: Executor) -> CheckResult: + cid, title = "5.3.1", "Ensure X-Content-Type-Options header is configured and enabled" + profiles = PROFILES_WS_L1 + cmd = "nginx -T 2>/dev/null | grep -i X-Content-Type-Options" + out, _ = _run(executor, cmd) + + if re.search(r"nosniff", out, re.IGNORECASE): + return _ok(cid, title, profiles, + actual=out.strip(), + expected='add_header X-Content-Type-Options "nosniff" always;', + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + return _fail(cid, title, profiles, + actual=out.strip() or "(X-Content-Type-Options header not configured)", + expected='add_header X-Content-Type-Options "nosniff" always;', + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + +# ── 5.3.2 ──────────────────────────────────────────────────────────────────── + +def check_5_3_2(executor: Executor) -> CheckResult: + cid, title = "5.3.2", "Ensure that Content Security Policy (CSP) is enabled and configured properly" + profiles = PROFILES_WS_L2 + cmd = "nginx -T 2>/dev/null | grep -i Content-Security-Policy" + out, _ = _run(executor, cmd) + + if out.strip(): + return _manual(cid, title, profiles, + actual=out.strip(), + expected="Content-Security-Policy header with appropriate directives for the application", + commands=[cmd], + remediation=REMEDIATIONS[cid], + notes="CSP is site-specific. Verify the policy is appropriate and restrictive.") + else: + return _fail(cid, title, profiles, + actual="Content-Security-Policy header not configured", + expected="Content-Security-Policy header present with appropriate directives", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + +# ── 5.3.3 ──────────────────────────────────────────────────────────────────── + +def check_5_3_3(executor: Executor) -> CheckResult: + cid, title = "5.3.3", "Ensure the Referrer Policy is enabled and configured properly" + profiles = PROFILES_WS_L2 + cmd = "nginx -T 2>/dev/null | grep -i Referrer-Policy" + out, _ = _run(executor, cmd) + + valid_policies = { + "no-referrer", "no-referrer-when-downgrade", "origin", + "origin-when-cross-origin", "same-origin", "strict-origin", + "strict-origin-when-cross-origin", "unsafe-url", + } + + if out.strip(): + found_policy = re.search( + r"(" + "|".join(valid_policies) + r")", out, re.IGNORECASE + ) + if found_policy: + return _ok(cid, title, profiles, + actual=out.strip(), + expected="Referrer-Policy header with a valid policy value", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + return _fail(cid, title, profiles, + actual=out.strip(), + expected="Referrer-Policy header with a recognized policy value", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + else: + return _fail(cid, title, profiles, + actual="Referrer-Policy header not configured", + expected="Referrer-Policy header with a valid policy value", + commands=[cmd], + remediation=REMEDIATIONS[cid]) + + +# ── Registry: map check_id → function ──────────────────────────────────────── + +ALL_CHECKS = { + "1.1.1": check_1_1_1, + "1.2.1": check_1_2_1, + "1.2.2": check_1_2_2, + "2.1.1": check_2_1_1, + "2.2.1": check_2_2_1, + "2.2.2": check_2_2_2, + "2.2.3": check_2_2_3, + "2.3.1": check_2_3_1, + "2.3.2": check_2_3_2, + "2.3.3": check_2_3_3, + "2.4.1": check_2_4_1, + "2.4.2": check_2_4_2, + "2.4.3": check_2_4_3, + "2.4.4": check_2_4_4, + "2.5.1": check_2_5_1, + "2.5.2": check_2_5_2, + "2.5.3": check_2_5_3, + "2.5.4": check_2_5_4, + "3.1": check_3_1, + "3.2": check_3_2, + "3.3": check_3_3, + "3.4": check_3_4, + "4.1.1": check_4_1_1, + "4.1.2": check_4_1_2, + "4.1.3": check_4_1_3, + "4.1.4": check_4_1_4, + "4.1.5": check_4_1_5, + "4.1.6": check_4_1_6, + "4.1.7": check_4_1_7, + "4.1.8": check_4_1_8, + "4.1.9": check_4_1_9, + "4.1.10": check_4_1_10, + "4.1.11": check_4_1_11, + "4.1.12": check_4_1_12, + "5.1.1": check_5_1_1, + "5.1.2": check_5_1_2, + "5.2.1": check_5_2_1, + "5.2.2": check_5_2_2, + "5.2.3": check_5_2_3, + "5.2.4": check_5_2_4, + "5.2.5": check_5_2_5, + "5.3.1": check_5_3_1, + "5.3.2": check_5_3_2, + "5.3.3": check_5_3_3, +} + + +# Profile → set of applicable check IDs +PROFILE_CHECKS: dict[str, list[str]] = { + "Level 1 - Webserver": [ + "1.1.1","1.2.1","1.2.2","2.1.1","2.2.1","2.2.2","2.2.3", + "2.3.1","2.3.2","2.3.3","2.4.1","2.4.2","2.4.3","2.4.4", + "2.5.1","2.5.2","2.5.3","3.1","3.2","3.3","4.1.1","4.1.2", + "4.1.3","4.1.4","4.1.5","4.1.6","4.1.7","4.1.8","4.1.11", + "5.1.2","5.2.1","5.2.2","5.2.3","5.3.1", + ], + "Level 1 - Proxy": [ + "1.1.1","1.2.1","1.2.2","2.1.1","2.2.1","2.2.2","2.2.3", + "2.3.1","2.3.2","2.3.3","2.4.1","2.4.2","2.4.3","2.4.4", + "2.5.1","2.5.2","2.5.3","2.5.4","3.1","3.2","3.3","3.4", + "4.1.1","4.1.2","4.1.3","4.1.4","4.1.5","4.1.6","4.1.7", + "4.1.8","4.1.9","4.1.11","5.1.2","5.2.1","5.2.2","5.2.3", + ], + "Level 1 - Loadbalancer": [ + "1.1.1","1.2.1","1.2.2","2.1.1","2.2.1","2.2.2","2.2.3", + "2.3.1","2.3.2","2.3.3","2.4.1","2.4.2","2.4.3","2.4.4", + "2.5.1","2.5.2","2.5.3","2.5.4","3.1","3.2","3.3","3.4", + "4.1.1","4.1.2","4.1.3","4.1.4","4.1.5","4.1.6","4.1.7", + "4.1.8","4.1.9","4.1.11","5.1.2","5.2.1","5.2.2","5.2.3", + ], + "Level 2 - Webserver": [ + "1.1.1","1.2.1","1.2.2","2.1.1","2.2.1","2.2.2","2.2.3", + "2.3.1","2.3.2","2.3.3","2.4.1","2.4.2","2.4.3","2.4.4", + "2.5.1","2.5.2","2.5.3","3.1","3.2","3.3","4.1.1","4.1.2", + "4.1.3","4.1.4","4.1.5","4.1.6","4.1.7","4.1.8","4.1.11", + "4.1.12","5.1.1","5.1.2","5.2.1","5.2.2","5.2.3","5.2.4", + "5.2.5","5.3.1","5.3.2","5.3.3", + ], + "Level 2 - Proxy": [ + "1.1.1","1.2.1","1.2.2","2.1.1","2.2.1","2.2.2","2.2.3", + "2.3.1","2.3.2","2.3.3","2.4.1","2.4.2","2.4.3","2.4.4", + "2.5.1","2.5.2","2.5.3","2.5.4","3.1","3.2","3.3","3.4", + "4.1.1","4.1.2","4.1.3","4.1.4","4.1.5","4.1.6","4.1.7", + "4.1.8","4.1.9","4.1.10","4.1.11","5.1.1","5.1.2","5.2.1", + "5.2.2","5.2.3","5.2.4","5.2.5", + ], + "Level 2 - Loadbalancer": [ + "1.1.1","1.2.1","1.2.2","2.1.1","2.2.1","2.2.2","2.2.3", + "2.3.1","2.3.2","2.3.3","2.4.1","2.4.2","2.4.3","2.4.4", + "2.5.1","2.5.2","2.5.3","2.5.4","3.1","3.2","3.3","3.4", + "4.1.1","4.1.2","4.1.3","4.1.4","4.1.5","4.1.6","4.1.7", + "4.1.8","4.1.9","4.1.10","4.1.11","5.1.1","5.1.2","5.2.1", + "5.2.2","5.2.3","5.2.4","5.2.5", + ], +} diff --git a/nginx_audit/executor.py b/nginx_audit/executor.py new file mode 100644 index 0000000..0ee1c5a --- /dev/null +++ b/nginx_audit/executor.py @@ -0,0 +1,85 @@ +"""Command executors: run shell commands locally or over SSH.""" +from __future__ import annotations + +import subprocess +from typing import Tuple + + +class ExecutorError(Exception): + pass + + +class LocalExecutor: + """Run shell commands on the local machine.""" + + name = "localhost" + + def run(self, cmd: str, timeout: int = 30) -> Tuple[str, int]: + """ + Execute *cmd* in a shell and return (combined_output, exit_code). + stdout and stderr are merged so callers get one string to inspect. + """ + try: + result = subprocess.run( + cmd, + shell=True, + capture_output=True, + text=True, + timeout=timeout, + ) + combined = (result.stdout + result.stderr).strip() + return combined, result.returncode + except subprocess.TimeoutExpired: + return f"[TIMEOUT after {timeout}s]", 1 + except Exception as exc: + return f"[ERROR: {exc}]", 1 + + def close(self) -> None: + pass # nothing to clean up + + +class SSHExecutor: + """Run shell commands on a remote host over SSH (requires paramiko).""" + + def __init__( + self, + host: str, + port: int = 22, + username: str = "root", + password: str | None = None, + key_file: str | None = None, + ) -> None: + try: + import paramiko # noqa: F401 + except ImportError as exc: + raise ExecutorError( + "paramiko is required for SSH mode: pip install paramiko" + ) from exc + + import paramiko + + self.name = host + self._client = paramiko.SSHClient() + self._client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + self._client.connect( + host, + port=port, + username=username, + password=password, + key_filename=key_file, + timeout=10, + ) + + def run(self, cmd: str, timeout: int = 30) -> Tuple[str, int]: + """Execute *cmd* on the remote host and return (combined_output, exit_code).""" + try: + _, stdout, stderr = self._client.exec_command(cmd, timeout=timeout) + exit_code = stdout.channel.recv_exit_status() + out = stdout.read().decode(errors="replace") + err = stderr.read().decode(errors="replace") + return (out + err).strip(), exit_code + except Exception as exc: + return f"[ERROR: {exc}]", 1 + + def close(self) -> None: + self._client.close() diff --git a/nginx_audit/models.py b/nginx_audit/models.py new file mode 100644 index 0000000..27409bf --- /dev/null +++ b/nginx_audit/models.py @@ -0,0 +1,66 @@ +"""Data models for the CIS NGINX benchmark auditor.""" +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime +from typing import List, Optional + + +# Possible check outcomes +PASS = "PASS" +FAIL = "FAIL" +MANUAL = "MANUAL" # check ran but result needs human judgment +ERROR = "ERROR" # could not run the check (command failed, not found, etc.) + + +@dataclass +class CheckResult: + """Result of running a single CIS check.""" + check_id: str + title: str + profiles: List[str] # which CIS profiles this check belongs to + status: str # PASS | FAIL | MANUAL | ERROR + actual_value: str # what was found on the system + expected_value: str # what the benchmark requires + commands_run: List[str] # shell commands executed + remediation: str # how to fix (from benchmark) + notes: str = "" # extra context or error detail + + +@dataclass +class AuditReport: + """Complete audit result for a target system.""" + target: str # hostname or "localhost" + profile: str # e.g. "Level 1 - Webserver" + timestamp: datetime = field(default_factory=datetime.utcnow) + results: List[CheckResult] = field(default_factory=list) + + # ── convenience aggregates ────────────────────────────────────────────── + + @property + def passed(self) -> List[CheckResult]: + return [r for r in self.results if r.status == PASS] + + @property + def failed(self) -> List[CheckResult]: + return [r for r in self.results if r.status == FAIL] + + @property + def manual(self) -> List[CheckResult]: + return [r for r in self.results if r.status == MANUAL] + + @property + def errors(self) -> List[CheckResult]: + return [r for r in self.results if r.status == ERROR] + + @property + def total(self) -> int: + return len(self.results) + + def summary(self) -> str: + return ( + f"Target: {self.target} | Profile: {self.profile} | " + f"Total: {self.total} | PASS: {len(self.passed)} | " + f"FAIL: {len(self.failed)} | MANUAL: {len(self.manual)} | " + f"ERROR: {len(self.errors)}" + ) diff --git a/nginx_audit/report.py b/nginx_audit/report.py new file mode 100644 index 0000000..63adf3b --- /dev/null +++ b/nginx_audit/report.py @@ -0,0 +1,277 @@ +""" +Generate an XLSX audit report with four sheets: + + 1. Summary – score, counts, target info + 2. Passed – checks that passed, actual vs expected + 3. Failed – checks that failed, actual vs expected, remediation + 4. Manual Review – checks requiring human judgment, command output + criteria + 5. Errors – checks that could not run + +Uses openpyxl (bundled with most Python environments). +""" +from __future__ import annotations + +from datetime import datetime +from pathlib import Path +from typing import List + +import openpyxl +from openpyxl.styles import ( + Alignment, Font, PatternFill, Border, Side +) +from openpyxl.utils import get_column_letter + +from .models import AuditReport, CheckResult, PASS, FAIL, MANUAL, ERROR + + +# ── colour palette ──────────────────────────────────────────────────────────── +C_HEADER_BG = "1F4E79" # dark blue – header row +C_HEADER_FG = "FFFFFF" # white +C_PASS_BG = "E2EFDA" # light green +C_FAIL_BG = "FCE4D6" # light red/orange +C_MANUAL_BG = "FFF2CC" # light yellow +C_ERROR_BG = "F2CEEF" # light purple +C_ALT_BG = "F5F5F5" # very light grey – alternating rows +C_SUMMARY_BG = "DEEAF1" # light blue – summary labels + +# thin border +_THIN = Side(style="thin", color="BFBFBF") +_BORDER = Border(left=_THIN, right=_THIN, top=_THIN, bottom=_THIN) + + +# ── helpers ─────────────────────────────────────────────────────────────────── + +def _header_font() -> Font: + return Font(name="Calibri", bold=True, color=C_HEADER_FG, size=11) + + +def _bold(color: str = "000000") -> Font: + return Font(name="Calibri", bold=True, color=color, size=11) + + +def _normal(size: int = 11) -> Font: + return Font(name="Calibri", size=size) + + +def _fill(hex_color: str) -> PatternFill: + return PatternFill("solid", fgColor=hex_color) + + +def _wrap_align() -> Alignment: + return Alignment(wrap_text=True, vertical="top") + + +def _set_header_row(ws, headers: list[str], row: int = 1) -> None: + for col, title in enumerate(headers, 1): + cell = ws.cell(row=row, column=col, value=title) + cell.font = _header_font() + cell.fill = _fill(C_HEADER_BG) + cell.alignment = Alignment(wrap_text=True, vertical="center", horizontal="center") + cell.border = _BORDER + + +def _write_row(ws, row: int, values: list, bg_color: str | None = None) -> None: + for col, val in enumerate(values, 1): + cell = ws.cell(row=row, column=col, value=str(val) if val is not None else "") + cell.font = _normal() + cell.alignment = _wrap_align() + cell.border = _BORDER + if bg_color: + cell.fill = _fill(bg_color) + elif row % 2 == 0: + cell.fill = _fill(C_ALT_BG) + + +def _auto_width(ws, min_w: int = 12, max_w: int = 60) -> None: + for col_cells in ws.columns: + length = max( + len(str(cell.value or "")) for cell in col_cells + ) + col_letter = get_column_letter(col_cells[0].column) + ws.column_dimensions[col_letter].width = max(min_w, min(length + 2, max_w)) + + +def _freeze(ws, cell: str = "A2") -> None: + ws.freeze_panes = cell + + +# ── sheet builders ──────────────────────────────────────────────────────────── + +def _build_summary(wb: openpyxl.Workbook, report: AuditReport) -> None: + ws = wb.active + ws.title = "Summary" + ws.sheet_view.showGridLines = False + + rows = [ + ("CIS NGINX Benchmark v3.0.0 – Audit Report", ""), + ("", ""), + ("Target", report.target), + ("Profile", report.profile), + ("Timestamp", report.timestamp.strftime("%Y-%m-%d %H:%M:%S UTC")), + ("", ""), + ("Total Checks", report.total), + ("✅ PASS", len(report.passed)), + ("❌ FAIL", len(report.failed)), + ("⚠️ MANUAL", len(report.manual)), + ("🔴 ERROR", len(report.errors)), + ("", ""), + ] + + if report.total: + pct_pass = round(len(report.passed) / report.total * 100, 1) + pct_fail = round(len(report.failed) / report.total * 100, 1) + rows.append(("Pass Rate", f"{pct_pass}%")) + rows.append(("Fail Rate", f"{pct_fail}%")) + + for r_idx, (label, value) in enumerate(rows, 1): + if r_idx == 1: + cell = ws.cell(row=r_idx, column=1, value=label) + cell.font = Font(name="Calibri", bold=True, size=16, color=C_HEADER_BG) + cell.alignment = Alignment(vertical="center") + ws.row_dimensions[r_idx].height = 30 + ws.merge_cells(start_row=r_idx, start_column=1, end_row=r_idx, end_column=2) + else: + lbl_cell = ws.cell(row=r_idx, column=1, value=label) + val_cell = ws.cell(row=r_idx, column=2, value=value) + if label and label not in ("",): + lbl_cell.font = _bold(C_HEADER_BG) + lbl_cell.fill = _fill(C_SUMMARY_BG) + val_cell.font = _normal() + + ws.column_dimensions["A"].width = 22 + ws.column_dimensions["B"].width = 40 + + +def _build_passed(wb: openpyxl.Workbook, results: list[CheckResult]) -> None: + ws = wb.create_sheet("Passed") + ws.sheet_view.showGridLines = False + headers = ["Check ID", "Title", "Commands Run", "Actual Value", "Expected Value"] + _set_header_row(ws, headers) + + for r_idx, res in enumerate(results, 2): + _write_row(ws, r_idx, [ + res.check_id, + res.title, + "\n".join(res.commands_run), + res.actual_value, + res.expected_value, + ], bg_color=C_PASS_BG) + ws.row_dimensions[r_idx].height = max( + 30, min(len(res.actual_value) // 3, 120) + ) + + _auto_width(ws) + _freeze(ws) + ws.column_dimensions["A"].width = 12 + ws.column_dimensions["B"].width = 52 + ws.column_dimensions["C"].width = 55 + ws.column_dimensions["D"].width = 55 + ws.column_dimensions["E"].width = 45 + + +def _build_failed(wb: openpyxl.Workbook, results: list[CheckResult]) -> None: + ws = wb.create_sheet("Failed") + ws.sheet_view.showGridLines = False + headers = [ + "Check ID", "Title", "Commands Run", + "Actual Value", "Expected Value", "Remediation Steps", "Notes" + ] + _set_header_row(ws, headers) + + for r_idx, res in enumerate(results, 2): + _write_row(ws, r_idx, [ + res.check_id, + res.title, + "\n".join(res.commands_run), + res.actual_value, + res.expected_value, + res.remediation, + res.notes, + ], bg_color=C_FAIL_BG) + ws.row_dimensions[r_idx].height = max( + 30, min(max(len(res.actual_value), len(res.remediation)) // 3, 150) + ) + + _auto_width(ws) + _freeze(ws) + ws.column_dimensions["A"].width = 12 + ws.column_dimensions["B"].width = 52 + ws.column_dimensions["C"].width = 55 + ws.column_dimensions["D"].width = 55 + ws.column_dimensions["E"].width = 45 + ws.column_dimensions["F"].width = 60 + ws.column_dimensions["G"].width = 40 + + +def _build_manual(wb: openpyxl.Workbook, results: list[CheckResult]) -> None: + ws = wb.create_sheet("Manual Review") + ws.sheet_view.showGridLines = False + headers = [ + "Check ID", "Title", "Commands Run", + "Command Output (Actual)", "What to Look For (Expected)", "Notes" + ] + _set_header_row(ws, headers) + + for r_idx, res in enumerate(results, 2): + _write_row(ws, r_idx, [ + res.check_id, + res.title, + "\n".join(res.commands_run), + res.actual_value, + res.expected_value, + res.notes, + ], bg_color=C_MANUAL_BG) + ws.row_dimensions[r_idx].height = max( + 30, min(len(res.actual_value) // 3, 150) + ) + + _auto_width(ws) + _freeze(ws) + ws.column_dimensions["A"].width = 12 + ws.column_dimensions["B"].width = 52 + ws.column_dimensions["C"].width = 55 + ws.column_dimensions["D"].width = 60 + ws.column_dimensions["E"].width = 55 + ws.column_dimensions["F"].width = 40 + + +def _build_errors(wb: openpyxl.Workbook, results: list[CheckResult]) -> None: + if not results: + return + ws = wb.create_sheet("Errors") + ws.sheet_view.showGridLines = False + headers = ["Check ID", "Title", "Commands Run", "Error Detail", "Notes"] + _set_header_row(ws, headers) + + for r_idx, res in enumerate(results, 2): + _write_row(ws, r_idx, [ + res.check_id, + res.title, + "\n".join(res.commands_run), + res.actual_value, + res.notes, + ], bg_color=C_ERROR_BG) + + _auto_width(ws) + _freeze(ws) + + +# ── public API ──────────────────────────────────────────────────────────────── + +def generate_report(report: AuditReport, output_path: str | Path) -> Path: + """ + Write the audit report as an XLSX file to *output_path*. + Returns the resolved Path. + """ + wb = openpyxl.Workbook() + + _build_summary(wb, report) + _build_passed(wb, report.passed) + _build_failed(wb, report.failed) + _build_manual(wb, report.manual) + _build_errors(wb, report.errors) + + path = Path(output_path) + path.parent.mkdir(parents=True, exist_ok=True) + wb.save(path) + return path diff --git a/nginx_audit/runner.py b/nginx_audit/runner.py new file mode 100644 index 0000000..bf601c1 --- /dev/null +++ b/nginx_audit/runner.py @@ -0,0 +1,70 @@ +"""Orchestrate running all checks for a given profile.""" +from __future__ import annotations + +import sys +from typing import List, Optional + +from .checks import ALL_CHECKS, PROFILE_CHECKS +from .executor import LocalExecutor, SSHExecutor +from .models import AuditReport, CheckResult, ERROR + + +def run_audit( + executor: LocalExecutor | SSHExecutor, + profile: str, + check_ids: Optional[List[str]] = None, + verbose: bool = False, +) -> AuditReport: + """ + Run all checks for *profile* (or a specific subset via *check_ids*) + against *executor*. Returns a populated AuditReport. + """ + if profile not in PROFILE_CHECKS: + raise ValueError( + f"Unknown profile '{profile}'. " + f"Valid: {list(PROFILE_CHECKS.keys())}" + ) + + ids_to_run = check_ids if check_ids else PROFILE_CHECKS[profile] + report = AuditReport(target=executor.name, profile=profile) + + total = len(ids_to_run) + for i, cid in enumerate(ids_to_run, 1): + if verbose: + print(f" [{i}/{total}] Running check {cid} ...", end="\r", flush=True) + + func = ALL_CHECKS.get(cid) + if func is None: + result = CheckResult( + check_id=cid, + title=f"Check {cid}", + profiles=[profile], + status=ERROR, + actual_value="No implementation found for this check ID", + expected_value="N/A", + commands_run=[], + remediation="", + notes="This check ID is not implemented in the current version.", + ) + else: + try: + result = func(executor) + except Exception as exc: + result = CheckResult( + check_id=cid, + title=f"Check {cid}", + profiles=[profile], + status=ERROR, + actual_value=f"Unexpected error: {exc}", + expected_value="N/A", + commands_run=[], + remediation="", + notes=str(exc), + ) + + report.results.append(result) + + if verbose: + print(" " * 60, end="\r") # clear progress line + + return report diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..7997a7f --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +openpyxl>=3.1.0 +paramiko>=3.0.0 # only required for SSH mode