560 lines
22 KiB
Python
560 lines
22 KiB
Python
#!/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"""<!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 SL2 — Compliance Tester</title>
|
|
<style>
|
|
:root{{--bg:#0d1117;--fg:#c9d1d9;--accent:#58a6ff;--green:#3fb950;
|
|
--red:#f85149;--card:#161b22;--border:#30363d;--btn:#21262d;--btn-hover:#30363d}}
|
|
*{{box-sizing:border-box;margin:0;padding:0}}
|
|
body{{font:14px/1.6 -apple-system,BlinkMacSystemFont,sans-serif;background:var(--bg);
|
|
color:var(--fg);max-width:1024px;margin:0 auto;padding:24px}}
|
|
h1{{color:var(--accent);font-size:22px;margin-bottom:4px}}
|
|
h2{{color:var(--accent);font-size:15px;margin:24px 0 10px;text-transform:uppercase;letter-spacing:.5px}}
|
|
.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(200px,1fr));gap:10px}}
|
|
.btn{{display:block;width:100%;padding:10px 14px;border:1px solid var(--border);border-radius:6px;
|
|
background:var(--btn);color:var(--fg);font-size:13px;cursor:pointer;text-align:left;transition:all .15s}}
|
|
.btn:hover{{background:var(--btn-hover);border-color:var(--accent)}}
|
|
.btn.run{{color:var(--green);font-weight:600}}
|
|
.output{{background:#0d1117;border:1px solid var(--border);border-radius:6px;
|
|
padding:14px;font:12px/ui-monospace,monospace;white-space:pre-wrap;
|
|
max-height:420px;overflow:auto;margin-top:12px;display:none;color:#7ee787}}
|
|
.output.visible{{display:block}}
|
|
.reports table{{width:100%;border-collapse:collapse;font-size:13px}}
|
|
.reports th{{text-align:left;padding:8px 12px;border-bottom:1px solid var(--border);color:var(--accent)}}
|
|
.reports td{{padding:8px 12px;border-bottom:1px solid var(--border)}}
|
|
.reports a{{color:var(--accent);text-decoration:none;margin-right:8px;font-size:12px;
|
|
padding:3px 8px;border:1px solid var(--border);border-radius:4px}}
|
|
.reports a:hover{{border-color:var(--accent);background:var(--btn-hover)}}
|
|
.status{{font-size:12px;color:#8b949e;margin-top:16px;display:flex;align-items:center;gap:8px}}
|
|
.status-dot{{width:8px;height:8px;border-radius:50%;display:inline-block}}
|
|
.status-dot.idle{{background:var(--green)}}
|
|
.status-dot.running{{background:#d29922;animation:pulse 1s infinite}}
|
|
@keyframes pulse{{50%{{opacity:.4}}}}
|
|
.rate{{font-size:28px;font-weight:700;color:var(--green)}}
|
|
.rate.low{{color:var(--red)}}
|
|
.summary-grid{{display:grid;grid-template-columns:repeat(4,1fr);gap:12px;margin-top:12px}}
|
|
.summary-item{{text-align:center;padding:12px;border-radius:6px;background:var(--btn)}}
|
|
.summary-item .num{{font-size:24px;font-weight:700}}
|
|
.summary-item .label{{font-size:11px;color:#8b949e;margin-top:4px}}
|
|
.summary-item.pass .num{{color:var(--green)}}
|
|
.summary-item.fail .num{{color:var(--red)}}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
|
|
<h1>⚡ IEC 62443-3-3 SL2</h1>
|
|
<p style="color:#8b949e;font-size:13px">Industrial control system security compliance validation</p>
|
|
|
|
<h2>▶ Run Tests</h2>
|
|
<div class="card">
|
|
<div class="grid">{playbook_buttons}</div>
|
|
<pre class="output" id="output">Select a playbook to run…</pre>
|
|
</div>
|
|
|
|
<div id="summary-section" style="display:none">
|
|
<h2>📊 Summary</h2>
|
|
<div class="card">
|
|
<div class="summary-grid" id="summary-grid"></div>
|
|
<div style="text-align:center;margin-top:12px">
|
|
<span class="rate" id="compliance-rate"></span>
|
|
<span style="color:#8b949e;font-size:12px;margin-left:8px">compliance rate</span>
|
|
</div>
|
|
</div>
|
|
</div>
|
|
|
|
<h2>📋 Reports</h2>
|
|
<div class="card reports">
|
|
<table>
|
|
<thead><tr><th>Report</th><th>Size</th><th style="text-align:right">Download</th></tr></thead>
|
|
<tbody id="report-list">{report_rows}</tbody>
|
|
</table>
|
|
</div>
|
|
|
|
<div class="status">
|
|
<span class="status-dot idle" id="status-dot"></span>
|
|
<span id="status-text">Ready</span>
|
|
</div>
|
|
|
|
<script>
|
|
async function runPlaybook(name) {{
|
|
const out = document.getElementById("output");
|
|
const dot = document.getElementById("status-dot");
|
|
const txt = document.getElementById("status-text");
|
|
out.classList.add("visible");
|
|
out.textContent = "▸ Starting " + name + "…\\n";
|
|
dot.className = "status-dot running";
|
|
txt.textContent = "Running " + name + "…";
|
|
|
|
try {{
|
|
const res = await fetch("/api/run", {{
|
|
method: "POST",
|
|
headers: {{"Content-Type": "application/x-www-form-urlencoded"}},
|
|
body: "playbook=" + encodeURIComponent(name)
|
|
}});
|
|
const data = await res.json();
|
|
out.textContent = data.output || data.error || "No output";
|
|
dot.className = "status-dot idle";
|
|
txt.textContent = data.ok ? "✓ " + name + " — passed" : "✗ " + name + " — issues found";
|
|
|
|
// Refresh summary & reports if run succeeded
|
|
if (data.latest_report) {{
|
|
loadSummary(data.latest_report);
|
|
}}
|
|
loadReports();
|
|
}} catch(e) {{
|
|
out.textContent += "\\n✗ Error: " + e;
|
|
dot.className = "status-dot idle";
|
|
txt.textContent = "Error running " + name;
|
|
}}
|
|
}}
|
|
|
|
async function loadSummary(reportFile) {{
|
|
try {{
|
|
const res = await fetch("/api/summary?file=" + encodeURIComponent(reportFile));
|
|
const s = await res.json();
|
|
document.getElementById("summary-section").style.display = "block";
|
|
document.getElementById("summary-grid").innerHTML =
|
|
`<div class="summary-item pass"><div class="num">${{s.passed}}</div><div class="label">Passed</div></div>
|
|
<div class="summary-item fail"><div class="num">${{s.failed}}</div><div class="label">Failed</div></div>
|
|
<div class="summary-item"><div class="num">${{s.total}}</div><div class="label">Total</div></div>
|
|
<div class="summary-item"><div class="num">${{s.skipped || 0}}</div><div class="label">Skipped</div></div>`;
|
|
const rate = document.getElementById("compliance-rate");
|
|
rate.textContent = (s.passed/s.total*100).toFixed(1) + "%";
|
|
rate.className = "rate" + (s.passed/s.total < 0.8 ? " low" : "");
|
|
}} catch(e) {{}}
|
|
}}
|
|
|
|
async function loadReports() {{
|
|
try {{
|
|
const res = await fetch("/api/reports");
|
|
const reports = await res.json();
|
|
let rows = "";
|
|
reports.forEach(r => {{
|
|
rows += `<tr>
|
|
<td>${{r.name}}</td>
|
|
<td style="color:#8b949e">${{r.size}}</td>
|
|
<td style="text-align:right">
|
|
<a href="/api/reports/${{r.name}}">JSON</a>
|
|
<a href="/api/reports/${{r.name}}/md">MD</a>
|
|
<a href="/api/reports/${{r.name}}/pdf">PDF</a>
|
|
</td></tr>`;
|
|
}});
|
|
document.getElementById("report-list").innerHTML =
|
|
rows || `<tr><td colspan="3" style="color:#8b949e">No reports yet</td></tr>`;
|
|
}} catch(e) {{}}
|
|
}}
|
|
</script>
|
|
</body>
|
|
</html>"""
|
|
|
|
# ── 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'<button class="btn run" '
|
|
f'onclick="runPlaybook(\'{p}\')">{label}</button>\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'<tr><td>{name}</td><td style="color:#8b949e">{szs}</td>'
|
|
f'<td style="text-align:right">'
|
|
f'<a href="/api/reports/{name}">JSON</a>'
|
|
f'<a href="/api/reports/{name}/md">MD</a>'
|
|
f'<a href="/api/reports/{name}/pdf">PDF</a>'
|
|
f'</td></tr>\n'
|
|
)
|
|
|
|
self._send(200, "text/html", HTML.format(
|
|
playbook_buttons=buttons or "<p style='color:#8b949e'>No playbooks found in /ansible/playbooks</p>",
|
|
report_rows=rows or '<tr><td colspan="3" style="color:#8b949e">No reports yet — run a test</td></tr>',
|
|
))
|
|
|
|
# ── 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
|