diff --git a/webui/container-app.py b/webui/container-app.py
new file mode 100644
index 0000000..04e8bdf
--- /dev/null
+++ b/webui/container-app.py
@@ -0,0 +1,559 @@
+#!/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…
+
+
📊 Summary
+
+
+
+
+ compliance rate
+
+
+
📋 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