diff --git a/webui/app.py b/webui/app.py new file mode 100644 index 0000000..4ef6264 --- /dev/null +++ b/webui/app.py @@ -0,0 +1,282 @@ +#!/usr/bin/env python3 +"""IEC 62443-3-3 Compliance Tester — Minimal Web UI. + +Zero dependencies beyond Python 3 stdlib. +Serves on :8080, executes playbooks via docker, serves reports.""" + +import http.server +import json +import os +import subprocess +import glob +import urllib.parse +import shutil +from pathlib import Path + +PLAYBOOKS_DIR = "/ansible/playbooks" +REPORTS_DIR = "/ansible/reports" +INVENTORY = "/ansible/inventory/inventory.ini" +ANSIBLE_IMAGE = "ansible-node" + +# ── HTML template (inline) ───────────────────────────────── +HTML = r""" + + + + +IEC 62443-3-3 Compliance Tester + + + + +

⚡ IEC 62443-3-3 SL2

+

Industrial control system security compliance validation

+ +

▶ Run Tests

+
+
+ {playbook_buttons} +
+ 
Select a playbook to run...
+
+ +

📋 Reports

+
+ +
+ +
{status}
+ + + +""" + + +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 do_GET(self): + p = urllib.parse.urlparse(self.path) + if p.path == "/" or p.path == "/index.html": + self._serve_index() + elif p.path == "/api/reports": + self._api_reports() + elif p.path.startswith("/reports/"): + self._serve_file(REPORTS_DIR, p.path[9:]) + 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] + limit = qs.get("limit", ["all"])[0] + self._run_playbook(playbook, limit) + else: + self._send(405, "text/plain", "Method not allowed") + + def _serve_index(self): + # List playbooks + pbs = sorted( + [f.name for f in Path(PLAYBOOKS_DIR).rglob("*.yml") + if not f.name.startswith(".")], + key=lambda x: (x != "site.yml", x) + ) + buttons = "" + for p in pbs: + label = p.replace(".yml", "").replace("_", " ").title() + if p == "site.yml": + label = "🚀 Run All Tests" + buttons += (f'\n') + + # List reports + try: + reps = sorted( + Path(REPORTS_DIR).glob("*.json"), + key=lambda f: f.stat().st_mtime, reverse=True + )[:20] + items = "" + for r in reps: + try: + sz = r.stat().st_size + szs = f"{sz/1024:.0f}KB" + except Exception: + szs = "?" + items += (f'
  • {r.name} {szs} ' + f'Download
    • \n') + except Exception: + items = "
  • No reports yet
    • " + + html = HTML.format( + playbook_buttons=buttons or "

    No playbooks found

    ", + report_items=items or "
  • No reports yet
    • ", + status="Ready" + ) + self._send(200, "text/html", html) + + def _api_reports(self): + try: + reps = sorted( + Path(REPORTS_DIR).glob("*.json"), + key=lambda f: f.stat().st_mtime, reverse=True + )[:20] + result = [] + for r in reps: + sz = r.stat().st_size + szs = f"{sz/1024:.0f}KB" + result.append({ + "name": r.name, + "size": szs, + "json_url": f"/reports/{r.name}", + }) + self._json(200, result) + except Exception as e: + self._json(500, {"error": str(e)}) + + def _serve_file(self, base, name): + name = os.path.basename(name) + fpath = os.path.join(base, name) + if not os.path.isfile(fpath): + self._send(404, "text/plain", "File not found") + return + ct = "application/json" if name.endswith(".json") else "application/octet-stream" + self.send_response(200) + self.send_header("Content-Type", ct) + self.send_header("Content-Disposition", + f'attachment; filename="{name}"') + self.end_headers() + with open(fpath, "rb") as f: + shutil.copyfileobj(f, self.wfile) + + def _run_playbook(self, playbook, limit): + 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"Playbook not found: {playbook}"}) + return + + rel = os.path.relpath(pb_path, PLAYBOOKS_DIR) + cmd = [ + "docker", "run", "--rm", + "-v", f"{PLAYBOOKS_DIR}:/ansible/playbooks:ro", + "-v", f"{REPORTS_DIR}:/ansible/reports", + "-v", f"{os.path.dirname(INVENTORY)}:/ansible/inventory:ro", + ANSIBLE_IMAGE, + f"/ansible/playbooks/{rel}", + "-i", "/ansible/inventory/inventory.ini", + ] + if limit and limit != "all": + cmd += ["--limit", limit] + + try: + result = subprocess.run( + cmd, capture_output=True, text=True, timeout=300 + ) + output = result.stdout + "\n" + result.stderr + self._json(200, { + "ok": result.returncode == 0, + "exit_code": result.returncode, + "output": output[-50000:] + }) + except subprocess.TimeoutExpired: + self._json(500, {"error": "Playbook timed out after 5 min"}) + 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("Listening on http://0.0.0.0:8080") + httpd = http.server.HTTPServer(("0.0.0.0", 8080), Handler) + try: + httpd.serve_forever() + except KeyboardInterrupt: + pass