edvinas/CIS_Benchmarks
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Initial commit

Browse source
This commit is contained in:
edvinas 2026-06-18 20:47:15 +03:00
commit bb00814b18
20 changed files with 3598 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

29
.gitignore vendored Normal file
View file

@ -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

124
README.md Normal file
View file

@ -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/<name>/`
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.

1
cis_auditor/__init__.py Normal file
View file

@ -0,0 +1 @@
"""CIS Benchmark Auditor — multi-benchmark edition."""

20
cis_auditor/benchmarks/__init__.py Normal file
View file

@ -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"]

212
cis_auditor/benchmarks/base.py Normal file
View file

@ -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

49
cis_auditor/benchmarks/debian/__init__.py Normal file
View file

@ -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)

284
cis_auditor/benchmarks/debian/loader.py Normal file
View file

@ -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

15
cis_auditor/benchmarks/nginx/__init__.py Normal file
View file

@ -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
)

1
cis_auditor/core/__init__.py Normal file
View file

@ -0,0 +1 @@
"""Core shared components: models, executors, report generator."""

85
cis_auditor/core/executor.py Normal file
View file

@ -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()

66
cis_auditor/core/models.py Normal file
View file

@ -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)}"
)

277
cis_auditor/core/report.py Normal file
View file

@ -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

210
main.py Normal file
View file

@ -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/<name>/
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())

1
nginx_audit/__init__.py Normal file
View file

@ -0,0 +1 @@
"""CIS NGINX Benchmark v3.0.0 Auditor."""

1724
nginx_audit/checks.py Normal file
View file

File diff suppressed because it is too large Load diff

85
nginx_audit/executor.py Normal file
View file

@ -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()

66
nginx_audit/models.py Normal file
View file

@ -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)}"
)

277
nginx_audit/report.py Normal file
View file

@ -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

70
nginx_audit/runner.py Normal file
View file

@ -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

2
requirements.txt Normal file
View file

@ -0,0 +1,2 @@
openpyxl>=3.1.0
paramiko>=3.0.0 # only required for SSH mode