#!/usr/bin/env python3
"""IEC 62443-3-3 Compliance Tester — Container Web UI.
Runs inside the ansible-node Docker image. Serves on :8080.
Features: run playbooks, export results as JSON / Markdown / PDF.
Dependencies: fpdf2 (pure Python PDF), render_report.py (bundled), Python 3 stdlib.
"""
import http.server
import json
import os
import subprocess
import urllib.parse
import shutil
import io
import time
from pathlib import Path
try:
from fpdf import FPDF
HAS_FPDF = True
except ImportError:
HAS_FPDF = False
PLAYBOOKS_DIR = "/ansible/playbooks"
REPORTS_DIR = "/ansible/reports"
INVENTORY = "/ansible/inventory/inventory.ini"
RENDER_MD = "/ansible/reports/render_report.py"
BIND = ("0.0.0.0", 8080)
# ── HTML template ──────────────────────────────────────────
HTML = r"""
IEC 62443-3-3 SL2 — Compliance Tester
⚡ IEC 62443-3-3 SL2
Industrial control system security compliance validation
▶ Run Tests
{playbook_buttons}
Select a playbook to run…
📋 Reports
| Report | Size | Download |
|---|
{report_rows}
Ready
"""
# ── PDF Generator ──────────────────────────────────────────
def generate_pdf(json_path: str) -> bytes:
"""Generate a clean PDF report from a test-results JSON file."""
with open(json_path) as f:
data = json.load(f)
meta = data.get("meta", {})
summary = data.get("summary", {})
results = data.get("results", [])
failures= data.get("failures", [])
pdf = FPDF()
pdf.set_auto_page_break(True, 20)
pdf.add_page()
# ── Cover / Header ──────────────────────────────────
pdf.set_font("Helvetica", "B", 22)
pdf.set_text_color(0, 74, 173)
pdf.cell(0, 12, "IEC 62443-3-3 SL2", new_x="LMARGIN", new_y="NEXT")
pdf.set_font("Helvetica", "", 14)
pdf.set_text_color(100, 100, 100)
pdf.cell(0, 8, "Compliance Validation Report", new_x="LMARGIN", new_y="NEXT")
pdf.ln(6)
# Meta info
pdf.set_font("Helvetica", "", 10)
pdf.set_text_color(80, 80, 80)
for label, key in [("Target:", "target"), ("Date:", "timestamp"),
("Standard:", "standard"), ("Security Level:", "security_level")]:
val = meta.get(key, "—")
pdf.cell(35, 6, label)
pdf.set_text_color(40, 40, 40)
pdf.cell(0, 6, str(val), new_x="LMARGIN", new_y="NEXT")
pdf.set_text_color(80, 80, 80)
pdf.ln(8)
# ── Summary box ─────────────────────────────────────
total = summary.get("total", 0)
passed = summary.get("passed", 0)
failed = summary.get("failed", 0)
rate = (passed / total * 100) if total > 0 else 0
pdf.set_fill_color(240, 248, 255)
pdf.rect(10, pdf.get_y(), 190, 22, style="F")
pdf.set_xy(14, pdf.get_y() + 4)
pdf.set_font("Helvetica", "B", 12)
pdf.set_text_color(0, 74, 173)
pdf.cell(50, 6, f"Passed: {passed}")
pdf.set_text_color(180, 40, 40)
pdf.cell(50, 6, f"Failed: {failed}")
pdf.set_text_color(40, 40, 40)
pdf.cell(50, 6, f"Total: {total}")
pdf.set_text_color(0, 120, 0)
pdf.cell(40, 6, f"Rate: {rate:.1f}%")
pdf.ln(26)
# ── Results by category ─────────────────────────────
by_cat = {}
for r in results:
cat = r.get("category", "Uncategorized")
by_cat.setdefault(cat, []).append(r)
for cat, items in by_cat.items():
# Category header
pdf.set_font("Helvetica", "B", 11)
pdf.set_text_color(0, 74, 173)
pdf.cell(0, 8, cat, new_x="LMARGIN", new_y="NEXT")
# Column headers
pdf.set_font("Helvetica", "B", 8)
pdf.set_fill_color(230, 235, 245)
pdf.set_text_color(60, 60, 60)
cols = [("Test ID", 22), ("Description", 72), ("Status", 18),
("Severity", 22), ("Expected", 56)]
for label, w in cols:
pdf.cell(w, 6, label, fill=True)
pdf.ln()
# Results
for r in items:
pid = r.get("test_id", "?")
desc = r.get("description", "")[:65]
p = r.get("passed")
sev = r.get("severity", "low")
exp = r.get("expected", "")[:45]
icon = "PASS" if p is True else ("FAIL" if p is False else "REVIEW")
pdf.set_font("Helvetica", "", 8)
if p is False:
pdf.set_text_color(180, 40, 40)
elif p is True:
pdf.set_text_color(0, 100, 0)
else:
pdf.set_text_color(180, 130, 0)
pdf.cell(22, 5, pid)
pdf.set_text_color(40, 40, 40)
pdf.cell(72, 5, desc)
pdf.set_text_color(180 if p is False else (0, 100, 0) if p is True else (180, 130, 0))
pdf.cell(18, 5, icon)
pdf.set_text_color(100, 100, 100)
pdf.cell(22, 5, sev.upper() if p is False else sev)
pdf.set_text_color(40, 40, 40)
pdf.cell(56, 5, exp)
pdf.ln()
pdf.ln(4)
# ── Failure details ──────────────────────────────────
if failures:
pdf.add_page()
pdf.set_font("Helvetica", "B", 14)
pdf.set_text_color(180, 40, 40)
pdf.cell(0, 10, "Failure Details & Remediation", new_x="LMARGIN", new_y="NEXT")
pdf.ln(4)
for f in failures:
pdf.set_font("Helvetica", "B", 10)
pdf.set_text_color(180, 40, 40)
pdf.cell(0, 7, f"[{f.get('test_id', '?')}] {f.get('description', '')}",
new_x="LMARGIN", new_y="NEXT")
pdf.set_font("Helvetica", "", 9)
pdf.set_text_color(80, 80, 80)
pdf.cell(0, 5, f" Expected: {f.get('expected', '—')}",
new_x="LMARGIN", new_y="NEXT")
pdf.cell(0, 5, f" Actual: {f.get('actual', '—')}",
new_x="LMARGIN", new_y="NEXT")
pdf.cell(0, 5, f" Remediation: {f.get('remediation', '—')}",
new_x="LMARGIN", new_y="NEXT")
pdf.set_draw_color(220, 220, 220)
pdf.line(10, pdf.get_y() + 2, 200, pdf.get_y() + 2)
pdf.ln(6)
return pdf.output()
# ── HTTP Handler ───────────────────────────────────────────
class Handler(http.server.BaseHTTPRequestHandler):
def log_message(self, fmt, *args):
pass
def _send(self, code, ct, body):
self.send_response(code)
self.send_header("Content-Type", ct)
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Cache-Control", "no-cache")
self.end_headers()
self.wfile.write(body if isinstance(body, bytes) else body.encode())
def _json(self, code, obj):
self._send(code, "application/json", json.dumps(obj, indent=2))
def _list_playbooks(self):
pbs = sorted(
[f.name for f in Path(PLAYBOOKS_DIR).rglob("*.yml")
if not f.name.startswith(".")],
key=lambda x: (x != "site.yml", x)
)
return pbs
def _list_reports(self):
try:
return sorted(
Path(REPORTS_DIR).glob("*.json"),
key=lambda f: f.stat().st_mtime, reverse=True
)[:50]
except Exception:
return []
# ── Routing ────────────────────────────────────────
def do_GET(self):
p = urllib.parse.urlparse(self.path)
path = p.path
if path == "/":
self._serve_index()
elif path == "/api/reports":
self._api_reports()
elif path == "/api/summary":
qs = urllib.parse.parse_qs(p.query)
fn = qs.get("file", [""])[0]
self._api_summary(fn)
elif path.startswith("/api/reports/"):
rest = path[len("/api/reports/"):]
if rest.endswith("/md"):
self._serve_markdown(rest[:-3])
elif rest.endswith("/pdf"):
self._serve_pdf(rest[:-4])
else:
self._serve_json(rest)
else:
self._send(404, "text/plain", "Not found")
def do_POST(self):
p = urllib.parse.urlparse(self.path)
if p.path == "/api/run":
cl = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(cl).decode()
qs = urllib.parse.parse_qs(body)
playbook = qs.get("playbook", [""])[0]
self._run_playbook(playbook)
else:
self._send(405, "text/plain", "Method not allowed")
# ── Pages ──────────────────────────────────────────
def _serve_index(self):
pbs = self._list_playbooks()
buttons = ""
for p in pbs:
label = p.replace(".yml", "").replace("_", " ").title()
if p == "site.yml":
label = "🚀 Run All Tests"
buttons += (f'\n')
reps = self._list_reports()
rows = ""
for r in reps:
name = r.name
try:
sz = r.stat().st_size
szs = f"{sz/1024:.0f} KB"
except Exception:
szs = "?"
rows += (
f'| {name} | {szs} | '
f''
f'JSON'
f'MD'
f'PDF'
f' |
\n'
)
self._send(200, "text/html", HTML.format(
playbook_buttons=buttons or "No playbooks found in /ansible/playbooks
",
report_rows=rows or '| No reports yet — run a test |
',
))
# ── API ────────────────────────────────────────────
def _api_reports(self):
result = []
for r in self._list_reports():
sz = r.stat().st_size
szs = f"{sz/1024:.0f} KB"
result.append({
"name": r.name,
"size": szs,
})
self._json(200, result)
def _api_summary(self, filename):
fpath = os.path.join(REPORTS_DIR, os.path.basename(filename))
if not os.path.isfile(fpath):
self._json(404, {"error": "Report not found"})
return
try:
with open(fpath) as f:
data = json.load(f)
s = data.get("summary", {})
self._json(200, {
"total": s.get("total", 0),
"passed": s.get("passed", 0),
"failed": s.get("failed", 0),
"skipped": s.get("skipped", 0),
})
except Exception as e:
self._json(500, {"error": str(e)})
# ── File serving ───────────────────────────────────
def _serve_json(self, name):
name = os.path.basename(name)
fpath = os.path.join(REPORTS_DIR, name)
if not os.path.isfile(fpath):
self._send(404, "text/plain", "File not found"); return
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Disposition", f'attachment; filename="{name}"')
self.end_headers()
with open(fpath, "rb") as f:
shutil.copyfileobj(f, self.wfile)
def _serve_markdown(self, name):
name = os.path.basename(name)
fpath = os.path.join(REPORTS_DIR, name)
if not os.path.isfile(fpath):
self._send(404, "text/plain", "File not found"); return
out = io.StringIO()
try:
# Use bundled render_report.py for markdown conversion
r = subprocess.run(
["python3", RENDER_MD, fpath, "--format", "md"],
capture_output=True, text=True, timeout=30, cwd=REPORTS_DIR
)
md = r.stdout or f"# Error converting report\n\n{r.stderr}"
except Exception:
md = f"# Error\n\nCould not convert {name} to Markdown"
self.send_response(200)
self.send_header("Content-Type", "text/markdown; charset=utf-8")
self.send_header("Content-Disposition",
f'attachment; filename="{name.replace(".json", ".md")}"')
self.end_headers()
self.wfile.write(md.encode())
def _serve_pdf(self, name):
if not HAS_FPDF:
self._send(500, "text/plain", "PDF support not installed (missing fpdf2)")
return
name = os.path.basename(name)
fpath = os.path.join(REPORTS_DIR, name)
if not os.path.isfile(fpath):
self._send(404, "text/plain", "File not found"); return
try:
pdf_bytes = generate_pdf(fpath)
except Exception as e:
self._send(500, "text/plain", f"PDF generation failed: {e}")
return
self.send_response(200)
self.send_header("Content-Type", "application/pdf")
self.send_header("Content-Disposition",
f'attachment; filename="{name.replace(".json", ".pdf")}"')
self.send_header("Content-Length", str(len(pdf_bytes)))
self.end_headers()
self.wfile.write(pdf_bytes)
# ── Run playbook ───────────────────────────────────
def _run_playbook(self, playbook):
playbook = os.path.basename(playbook)
if not playbook or ".." in playbook:
self._json(400, {"error": "Invalid playbook name"}); return
pb_path = None
for f in Path(PLAYBOOKS_DIR).rglob(playbook):
pb_path = str(f); break
if not pb_path:
self._json(404, {"error": f"Not found: {playbook}"}); return
rel = os.path.relpath(pb_path, PLAYBOOKS_DIR)
cmd = [
"ansible-playbook",
f"/ansible/playbooks/{rel}",
"-i", INVENTORY,
]
try:
result = subprocess.run(cmd, capture_output=True, text=True,
timeout=300, cwd="/ansible")
output = (result.stdout + "\n" + result.stderr)[-80000:]
# Find latest report
latest = ""
reps = sorted(Path(REPORTS_DIR).glob("*.json"),
key=lambda f: f.stat().st_mtime, reverse=True)
if reps:
latest = reps[0].name
self._json(200, {
"ok": result.returncode == 0,
"exit_code": result.returncode,
"output": output,
"latest_report": latest,
})
except subprocess.TimeoutExpired:
self._json(500, {"error": "Timed out after 5 minutes"})
except Exception as e:
self._json(500, {"error": str(e)})
if __name__ == "__main__":
os.makedirs(PLAYBOOKS_DIR, exist_ok=True)
os.makedirs(REPORTS_DIR, exist_ok=True)
print(f"Listening on http://{BIND[0]}:{BIND[1]}")
httpd = http.server.HTTPServer(BIND, Handler)
try:
httpd.serve_forever()
except KeyboardInterrupt:
pass