#!/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