#!/usr/bin/env python3 """ SYLink PenTest VM — mini UI web port 8080. Design refresh — style cohérent avec le portail client UniSOC (slate + amber). Endpoints : GET / — page principale (activation si pas activée + gestion réseau) GET /api/network-state — JSON état réseau live POST /api/network-config — applique config DHCP/static POST /api/route — ajoute route statique POST /api/route/delete — supprime route """ from __future__ import annotations import json import os import shutil import subprocess from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import urlparse CONFIG_DIR = Path("/opt/sylink-pentest/etc") LOG_DIR = Path("/opt/sylink-pentest/log") NETPLAN_FILE = Path("/etc/netplan/99-sylink-pentest.yaml") NETPLAN_BACKUP = CONFIG_DIR / "netplan_backup.yaml" PORT = int(os.environ.get("SETUP_UI_PORT", "8080")) def load_state() -> dict: state = {"activated": False} enroll = CONFIG_DIR / "enroll.json" license_file = CONFIG_DIR / "license.json" if enroll.exists(): state.update(json.loads(enroll.read_text())) if license_file.exists(): try: lic = json.loads(license_file.read_text()) if lic.get("api_key"): state["activated"] = True state["tenant_id"] = lic.get("tenant_id") state["expires_at"] = lic.get("expires_at") state["enabled_services"] = lic.get("enabled_services") or [] except Exception: pass return state def get_resources_state() -> dict: """Collecte CPU, RAM, disque, réseau (rx/tx bytes), uptime.""" out = {} # CPU loadavg try: with open("/proc/loadavg") as f: parts = f.read().split() out["load_1"] = float(parts[0]) out["load_5"] = float(parts[1]) out["load_15"] = float(parts[2]) except Exception: pass # CPU count try: out["cpu_count"] = os.cpu_count() or 1 except Exception: out["cpu_count"] = 1 # RAM try: with open("/proc/meminfo") as f: mem = {} for line in f: k, v = line.split(":", 1) mem[k.strip()] = int(v.strip().split()[0]) * 1024 # bytes total = mem.get("MemTotal", 1) avail = mem.get("MemAvailable", 0) out["mem_total"] = total out["mem_used"] = total - avail out["mem_used_pct"] = round(100 * (total - avail) / total, 1) except Exception: pass # Disk root try: st = os.statvfs("/") used = (st.f_blocks - st.f_bavail) * st.f_frsize total = st.f_blocks * st.f_frsize out["disk_total"] = total out["disk_used"] = used out["disk_used_pct"] = round(100 * used / total, 1) except Exception: pass # Network rx/tx (sum of all real interfaces) try: rx_total = 0 tx_total = 0 with open("/proc/net/dev") as f: for line in f.readlines()[2:]: # skip headers parts = line.split() iface = parts[0].rstrip(":") if iface in ("lo",) or iface.startswith(("veth", "docker", "br-")): continue rx_total += int(parts[1]) tx_total += int(parts[9]) out["net_rx_bytes"] = rx_total out["net_tx_bytes"] = tx_total except Exception: pass # Uptime try: with open("/proc/uptime") as f: out["uptime_seconds"] = int(float(f.read().split()[0])) except Exception: pass return out def fmt_bytes(n: int) -> str: if n is None: return "—" units = ["B", "KB", "MB", "GB", "TB"] for u in units: if n < 1024: return f"{n:.1f} {u}" if u != "B" else f"{n} {u}" n /= 1024 return f"{n:.1f} PB" def fmt_uptime(s: int) -> str: if not s: return "—" days = s // 86400 hours = (s % 86400) // 3600 minutes = (s % 3600) // 60 parts = [] if days: parts.append(f"{days}j") if hours: parts.append(f"{hours}h") if minutes or not parts: parts.append(f"{minutes}min") return " ".join(parts) def get_network_state() -> dict: out = {"interfaces": [], "default_gateway": None, "dns_servers": [], "mode": "unknown", "routes_static": []} try: ip_addr = subprocess.run(["ip", "-j", "addr"], capture_output=True, text=True, timeout=5).stdout for iface in json.loads(ip_addr): name = iface.get("ifname", "") if name in ("lo",) or name.startswith(("veth", "docker", "br-", "podman")): continue addrs = iface.get("addr_info", []) ipv4 = next((f"{a['local']}/{a['prefixlen']}" for a in addrs if a.get("family") == "inet"), None) out["interfaces"].append({ "name": name, "mac": iface.get("address"), "ipv4": ipv4, "operstate": iface.get("operstate"), "mtu": iface.get("mtu"), }) except Exception: pass try: rt = subprocess.run(["ip", "-j", "route"], capture_output=True, text=True, timeout=5).stdout for r in json.loads(rt): if r.get("dst") == "default": out["default_gateway"] = r.get("gateway") elif r.get("dst") and r["dst"] != "default": out["routes_static"].append({"cidr": r["dst"], "via": r.get("gateway"), "dev": r.get("dev")}) except Exception: pass try: with open("/etc/resolv.conf") as f: for line in f: if line.startswith("nameserver"): out["dns_servers"].append(line.split()[1]) except Exception: pass if NETPLAN_FILE.exists(): try: content = NETPLAN_FILE.read_text() if "dhcp4: true" in content: out["mode"] = "dhcp" elif "addresses:" in content: out["mode"] = "static" except Exception: pass return out def primary_interface() -> str: try: out = subprocess.run(["ip", "-o", "route", "show", "default"], capture_output=True, text=True, timeout=5).stdout parts = out.split() if "dev" in parts: return parts[parts.index("dev") + 1] except Exception: pass return "enp6s18" def is_rfc1918(cidr_or_ip: str) -> bool: import ipaddress try: net = ipaddress.ip_network(cidr_or_ip, strict=False) return net.is_private and not net.is_loopback and not net.is_multicast except Exception: return False def apply_netplan_config(cfg: dict) -> tuple[bool, str]: iface = primary_interface() if cfg.get("mode") == "static": if not cfg.get("ipv4_address") or not cfg.get("ipv4_gateway"): return False, "Adresse IP et gateway requises en mode statique" if not is_rfc1918(cfg["ipv4_address"]) or not is_rfc1918(cfg["ipv4_gateway"]): return False, "L'IP et la gateway doivent être en RFC1918 (10/8, 172.16/12, 192.168/16)" ns = cfg.get("dns_servers") or ["9.9.9.9", "1.1.1.1"] eth_block = f""" dhcp4: false addresses: ["{cfg['ipv4_address']}"] routes: - to: default via: {cfg['ipv4_gateway']} nameservers: addresses: [{', '.join(ns)}]""" else: eth_block = " dhcp4: true" yaml_content = f"""network: version: 2 renderer: networkd ethernets: {iface}: {eth_block} """ NETPLAN_BACKUP.parent.mkdir(parents=True, exist_ok=True) if NETPLAN_FILE.exists(): shutil.copyfile(NETPLAN_FILE, NETPLAN_BACKUP) NETPLAN_FILE.write_text(yaml_content) os.chmod(NETPLAN_FILE, 0o600) res = subprocess.run(["netplan", "apply"], capture_output=True, text=True, timeout=30) if res.returncode != 0: if NETPLAN_BACKUP.exists(): shutil.copyfile(NETPLAN_BACKUP, NETPLAN_FILE) subprocess.run(["netplan", "apply"], capture_output=True, timeout=30) return False, res.stderr or "netplan apply a échoué" return True, "OK" def add_static_route(cidr: str, via: str) -> tuple[bool, str]: if cidr in ("0.0.0.0/0", "::/0"): return False, "La default route (0.0.0.0/0) est interdite" if not is_rfc1918(cidr) or not is_rfc1918(via): return False, "RFC1918 obligatoire" res = subprocess.run(["ip", "route", "add", cidr, "via", via], capture_output=True, text=True, timeout=10) return res.returncode == 0, res.stderr or "OK" def del_static_route(cidr: str) -> tuple[bool, str]: res = subprocess.run(["ip", "route", "del", cidr], capture_output=True, text=True, timeout=10) return res.returncode == 0, res.stderr or "OK" HTML_TEMPLATE = """ SYLink PenTest — Setup
{hostname}
{status_banner}
{activation_card} {specs_card}
{network_card} {routes_card}
Documentation : guide d'installation  ·  Logs locaux : journalctl -u sylink-pentest-agent  ·  Support : contact@unisoc.fr
""" def render_status_banner(state: dict) -> str: if state.get("activated"): services = ', '.join(state.get('enabled_services') or []) or '—' exp = state.get('expires_at') or '—' if exp != '—': try: exp = exp[:10] # YYYY-MM-DD only except Exception: pass return f"""""" return """""" def render_activation_card(state: dict) -> str: """Bloc d'activation — affiché UNIQUEMENT si pas activée.""" if state.get("activated"): return "" try: qr_text = "" content = Path("/etc/issue").read_text() if Path("/etc/issue").exists() else "" qr_text = content[: content.find("Code d'activation")] if "Code d'activation" in content else "" qr_text = "\n".join(l for l in qr_text.splitlines() if l.strip()) except Exception: qr_text = "(QR indisponible — voir logs init)" return f"""

📱
Activation

{qr_text}
1Scanner le QR avec votre téléphone connecté au portail UniSOC (recommandé)
2Code d'activation manuel
{state.get('short_code', '—')}
3URL directe {state.get('qr_url', '—')}
""" def render_resources_card(res: dict) -> str: """Card monitoring système — CPU/RAM/disque/réseau/uptime.""" def progress_bar(pct, color="primary"): if pct is None: return "—" color_map = {"primary": "var(--primary)", "danger": "var(--danger)", "success": "var(--success)"} c = color_map.get(color, color_map["primary"]) if pct >= 90: c = color_map["danger"] elif pct >= 75: c = "#f59e0b" return f"""
{pct}%
""" cpu_load = res.get("load_1", 0) cpu_count = res.get("cpu_count", 1) cpu_pct = min(round(100 * cpu_load / cpu_count, 1), 100) if cpu_count else 0 return f"""

📊
Gestion des ressources

LIVE
CPU charge (1 min)
{progress_bar(cpu_pct)}
Load avg
{res.get('load_1', '?'):.2f} / {res.get('load_5', '?'):.2f} / {res.get('load_15', '?'):.2f}
RAM
{progress_bar(res.get('mem_used_pct'))}
RAM utilisée / totale
{fmt_bytes(res.get('mem_used'))} / {fmt_bytes(res.get('mem_total'))}
Disque /
{progress_bar(res.get('disk_used_pct'))}
Disque utilisé / total
{fmt_bytes(res.get('disk_used'))} / {fmt_bytes(res.get('disk_total'))}
Réseau reçu (cumul)
{fmt_bytes(res.get('net_rx_bytes'))} ↓
Réseau envoyé (cumul)
{fmt_bytes(res.get('net_tx_bytes'))} ↑
Uptime
{fmt_uptime(res.get('uptime_seconds'))}

Rafraîchi à chaque chargement de page. Les compteurs réseau sont cumulés depuis le boot.

""" def render_specs_card(state: dict) -> str: return f"""

⚙
Spécifications

Hostname
{os.uname().nodename}
Fingerprint
{(state.get('machine_fingerprint') or '—')[:24]}…
API
{state.get('pt_api_base', 'https://pentest.unisoc.fr')}
Portail SOC
{state.get('portal_base', 'https://client.unisoc.fr')}
""" def render_network_card(net: dict) -> str: iface_rows = "" current_ipv4 = "" for i in net["interfaces"]: operstate_badge = 'UP' if i.get("operstate") == "UP" else f'{i.get("operstate","?")}' iface_rows += f'
{i["name"]}
{i.get("ipv4") or "—"} {operstate_badge}
' if i.get("ipv4") and not current_ipv4: current_ipv4 = i["ipv4"] mode_badge = 'DHCP' if net["mode"] == "dhcp" else ( 'IP FIXE' if net["mode"] == "static" else '?' ) dhcp_selected = "selected" if net["mode"] != "static" else "" static_selected = "selected" if net["mode"] == "static" else "" static_display = "block" if net["mode"] == "static" else "none" gateway = net["default_gateway"] or "—" dns_str = ", ".join(net["dns_servers"]) or "—" current_gw = net["default_gateway"] or "" current_dns = ", ".join(net["dns_servers"]) return f"""

🌐
Gestion réseau

{mode_badge}
État actuel
{iface_rows}
Gateway
{gateway}
DNS
{dns_str}
Modifier la configuration
En cas de perte de connectivité > 5 min après reconfig, rollback automatique vers la config précédente.
""" def render_routes_card(net: dict) -> str: if net["routes_static"]: routes_rows = "" for r in net["routes_static"]: routes_rows += f'
{r["cidr"]} via {r["via"] or "—"}
' else: routes_rows = '
Aucune route additionnelle
' return f"""

🛣
Routes statiques

Pour scanner d'autres réseaux internes (VLAN, sites distants via VPN). RFC1918 obligatoire ; 0.0.0.0/0 interdit.

Routes actives
{routes_rows}
Ajouter une route
""" def render_page() -> str: state = load_state() net = get_network_state() res = get_resources_state() return HTML_TEMPLATE.format( hostname=os.uname().nodename, portal_base=state.get("portal_base", "https://client.unisoc.fr"), status_banner=render_status_banner(state), activation_card=render_activation_card(state), specs_card=render_specs_card(state) + render_resources_card(res), network_card=render_network_card(net), routes_card=render_routes_card(net), ) class Handler(BaseHTTPRequestHandler): def _json(self, code, data): body = json.dumps(data).encode() self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _read_body(self): length = int(self.headers.get("Content-Length", "0")) return json.loads(self.rfile.read(length)) if length else {} def do_GET(self): path = urlparse(self.path).path if path == "/": page = render_page() self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() self.wfile.write(page.encode()) elif path == "/api/network-state": self._json(200, get_network_state()) else: self.send_response(404); self.end_headers() def do_POST(self): path = urlparse(self.path).path try: data = self._read_body() except Exception as e: self._json(400, {"success": False, "error": f"JSON invalide: {e}"}); return if path == "/api/network-config": ok, msg = apply_netplan_config(data) self._json(200 if ok else 400, {"success": ok, "error": None if ok else msg}) elif path == "/api/route": ok, msg = add_static_route(data.get("cidr", ""), data.get("via", "")) self._json(200 if ok else 400, {"success": ok, "error": None if ok else msg}) elif path == "/api/route/delete": ok, msg = del_static_route(data.get("cidr", "")) self._json(200 if ok else 400, {"success": ok, "error": None if ok else msg}) else: self.send_response(404); self.end_headers() def log_message(self, format, *args): pass def main(): server = ThreadingHTTPServer(("0.0.0.0", PORT), Handler) print(f"sylink-pentest-setup-ui listening on 0.0.0.0:{PORT}") server.serve_forever() if __name__ == "__main__": main()