283 lines
10 KiB
Python
283 lines
10 KiB
Python
#!/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"""<!DOCTYPE html>
|
|
<html lang="en">
|
|
<head>
|
|
<meta charset="utf-8">
|
|
<meta name="viewport" content="width=device-width,initial-scale=1">
|
|
<title>IEC 62443-3-3 Compliance Tester</title>
|
|
<style>
|
|
:root{{--bg:#1a1a2e;--fg:#e0e0e0;--accent:#00d4aa;--err:#ff6b6b;
|
|
--card:#16213e;--border:#0f3460;--btn:#0f3460;--btn-hover:#1a508b}}
|
|
*{{box-sizing:border-box;margin:0;padding:0}}
|
|
body{{font:14px/1.6 system-ui,sans-serif;background:var(--bg);color:var(--fg);
|
|
max-width:960px;margin:0 auto;padding:20px}}
|
|
h1{{color:var(--accent);margin-bottom:8px}}
|
|
h2{{color:var(--accent);margin:24px 0 12px;font-size:1.1em}}
|
|
.card{{background:var(--card);border:1px solid var(--border);
|
|
border-radius:8px;padding:16px;margin-bottom:16px}}
|
|
.grid{{display:grid;grid-template-columns:repeat(auto-fill,minmax(220px,1fr));gap:12px}}
|
|
.btn{{display:block;width:100%;padding:12px 16px;border:none;border-radius:6px;
|
|
background:var(--btn);color:var(--fg);font-size:14px;cursor:pointer;
|
|
text-align:left;transition:background .2s}}
|
|
.btn:hover{{background:var(--btn-hover)}}
|
|
.btn.run{{color:var(--accent);font-weight:bold}}
|
|
.output{{background:#000;color:#0f0;padding:12px;border-radius:6px;
|
|
font:12px monospace;white-space:pre-wrap;max-height:420px;overflow:auto;
|
|
margin-top:12px;display:none}}
|
|
.output.visible{{display:block}}
|
|
.badge{{display:inline-block;padding:2px 8px;border-radius:4px;font-size:11px;
|
|
margin-left:8px;opacity:.7}}
|
|
.report-list{{list-style:none}}
|
|
.report-list li{{padding:8px 0;border-bottom:1px solid var(--border)}}
|
|
.report-list a{{color:var(--accent);text-decoration:none;margin-right:12px}}
|
|
.report-list a:hover{{text-decoration:underline}}
|
|
.status{{font-size:12px;opacity:.7;margin-top:16px}}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
|
|
<h1>⚡ IEC 62443-3-3 SL2</h1>
|
|
<p>Industrial control system security compliance validation</p>
|
|
|
|
<h2>▶ Run Tests</h2>
|
|
<div class="card">
|
|
<div class="grid" id="playbook-list">
|
|
{playbook_buttons}
|
|
</div>
|
|
<pre class="output" id="output">Select a playbook to run...</pre>
|
|
</div>
|
|
|
|
<h2>📋 Reports</h2>
|
|
<div class="card">
|
|
<ul class="report-list" id="report-list">
|
|
{report_items}
|
|
</ul>
|
|
</div>
|
|
|
|
<div class="status" id="status">{status}</div>
|
|
|
|
<script>
|
|
async function runPlaybook(name) {{
|
|
const out = document.getElementById("output");
|
|
out.classList.add("visible");
|
|
out.textContent = "Starting " + name + "...\\n";
|
|
document.getElementById("status").textContent = "Running " + name + "...";
|
|
|
|
try {{
|
|
const res = await fetch("/api/run", {{
|
|
method: "POST",
|
|
headers: {{"Content-Type": "application/x-www-form-urlencoded"}},
|
|
body: "playbook=" + encodeURIComponent(name) +
|
|
"&limit=" + encodeURIComponent(document.getElementById("limit")?.value || "all")
|
|
}});
|
|
const data = await res.json();
|
|
out.textContent = data.output || data.error || "No output";
|
|
document.getElementById("status").textContent =
|
|
data.ok ? "✓ " + name + " completed" : "✗ " + name + " failed";
|
|
|
|
// Refresh report list
|
|
const rr = await fetch("/api/reports");
|
|
const reports = await rr.json();
|
|
let items = "";
|
|
reports.forEach(r => {{
|
|
items += `<li>${{r.name}} <span class="badge">${{r.size}}</span>
|
|
<a href="${{r.json_url}}">JSON</a></li>`;
|
|
}});
|
|
document.getElementById("report-list").innerHTML =
|
|
items || "<li>No reports yet</li>";
|
|
}} catch(e) {{
|
|
out.textContent += "\\nError: " + e;
|
|
document.getElementById("status").textContent = "Connection lost";
|
|
}}
|
|
}}
|
|
</script>
|
|
</body>
|
|
</html>"""
|
|
|
|
|
|
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'<button class="btn run" '
|
|
f'onclick="runPlaybook(\'{p}\')">{label}</button>\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'<li>{r.name} <span class="badge">{szs}</span> '
|
|
f'<a href="/reports/{r.name}">Download</a></li>\n')
|
|
except Exception:
|
|
items = "<li>No reports yet</li>"
|
|
|
|
html = HTML.format(
|
|
playbook_buttons=buttons or "<p>No playbooks found</p>",
|
|
report_items=items or "<li>No reports yet</li>",
|
|
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
|