#!/usr/bin/env python3 """ SYLink HoneyBot — service polling activation. Polle toutes les 15 s /api/honeypot/enroll/poll-activation?token=. Quand le backend renvoie `activated=true` : 1. Persiste license.json (license_token + tenant_id + honeypot_id) 2. Applique config réseau si network_config présent (netplan generate/apply) 3. Persiste hostname si imposé 4. Lance `reboot` — au prochain boot, init.py voit license.json et pose le banner « couverture » (BackupServer Corp) au lieu du banner d'activation. Service systemd : Type=simple, Restart=always, lancé après sylink-honeypot-init. """ from __future__ import annotations import json import os import socket import subprocess import sys import time import urllib.request import urllib.parse from datetime import datetime, timezone from pathlib import Path CONFIG_DIR = Path("/opt/sylink-honeypot/etc") LOG_DIR = Path("/opt/sylink-honeypot/log") PT_API_BASE = os.environ.get("HP_API_BASE", "https://api.unisoc.fr") POLL_INTERVAL = int(os.environ.get("HP_POLL_INTERVAL", "15")) def log(msg: str): LOG_DIR.mkdir(parents=True, exist_ok=True) line = f"[{datetime.now(timezone.utc).isoformat()}] {msg}" print(line, flush=True) with (LOG_DIR / "poll.log").open("a") as f: f.write(line + "\n") def read_enroll() -> dict | None: f = CONFIG_DIR / "enroll.json" if not f.exists(): return None try: return json.loads(f.read_text()) except Exception: return None def already_activated() -> bool: f = CONFIG_DIR / "license.json" if not f.exists(): return False try: return bool(json.loads(f.read_text()).get("license_token")) except Exception: return False def poll_once(token: str) -> dict | None: url = f"{PT_API_BASE}/api/honeypot/enroll/poll-activation?token={urllib.parse.quote(token)}" try: req = urllib.request.Request(url, method="GET") with urllib.request.urlopen(req, timeout=10) as resp: return json.loads(resp.read()) except urllib.error.HTTPError as e: log(f"poll HTTP {e.code} : {e.reason}") return None except Exception as e: log(f"poll erreur : {e}") return None def write_license(payload: dict): CONFIG_DIR.mkdir(parents=True, exist_ok=True) license_doc = { "license_token": payload["license_token"], "tenant_id": payload.get("tenant_id"), "honeypot_id": payload.get("honeypot_id"), "activated_at": payload.get("activated_at") or datetime.now(timezone.utc).isoformat(), } (CONFIG_DIR / "license.json").write_text(json.dumps(license_doc, indent=2)) os.chmod(CONFIG_DIR / "license.json", 0o600) log(f"license.json écrit (honeypot_id={license_doc['honeypot_id']})") def apply_hostname(hostname: str | None): if not hostname: return try: subprocess.run(["hostnamectl", "set-hostname", hostname], check=True, timeout=10) log(f"hostname défini : {hostname}") except Exception as e: log(f"hostnamectl échec : {e}") def apply_network_config(network_config: dict | None): """Applique la config réseau choisie côté portail (DHCP ou statique). Supporte netplan (Ubuntu) ET ifupdown (Debian) — détection auto.""" if not network_config: return mode = network_config.get("mode", "dhcp") iface = network_config.get("interface") or "ens192" netplan_present = Path("/etc/netplan").is_dir() # ─── Mode DHCP : restaurer la config originale ─── if mode == "dhcp": try: if netplan_present: f_unisoc = Path("/etc/netplan/99-unisoc-honeypot.yaml") if f_unisoc.exists(): f_unisoc.unlink() for f in Path("/etc/netplan").glob("*.yaml.unisoc-bak"): f.rename(f.with_suffix("")) if not list(Path("/etc/netplan").glob("*.yaml")): Path("/etc/netplan/01-dhcp.yaml").write_text( f"network:\n version: 2\n ethernets:\n {iface}:\n dhcp4: true\n" ) os.chmod("/etc/netplan/01-dhcp.yaml", 0o600) subprocess.run(["netplan", "generate"], check=False, timeout=15) subprocess.Popen(["bash", "-c", "sleep 2 && netplan apply"]) log("network: netplan DHCP restauré (apply en background)") else: dropin = Path("/etc/network/interfaces.d/99-unisoc-honeypot") if dropin.exists(): dropin.unlink() backup_path = Path("/etc/network/interfaces.unisoc-bak") interfaces_path = Path("/etc/network/interfaces") if backup_path.exists(): interfaces_path.write_text(backup_path.read_text()) else: interfaces_path.write_text( f"source /etc/network/interfaces.d/*\n\n" f"auto lo\niface lo inet loopback\n\n" f"allow-hotplug {iface}\niface {iface} inet dhcp\n" ) # Apply : ifdown/ifup en background (coupe le réseau quelques secondes) subprocess.Popen(["bash", "-c", f"sleep 2 && ifdown {iface} 2>/dev/null ; ifup {iface}"]) log("network: ifupdown DHCP restauré (apply en background)") except Exception as e: log(f"network DHCP apply échec : {e}") return # ─── Mode static : pose la config ─── if mode != "static": log(f"network_config: mode inconnu '{mode}' — skip") return ip_cidr = network_config.get("ip_cidr") gateway = network_config.get("gateway") dns = network_config.get("dns") or ["8.8.8.8", "1.1.1.1"] if not ip_cidr or not gateway: log(f"network: static incomplet (ip={ip_cidr} gw={gateway}) — skip") return try: if netplan_present: yaml = f"""# UniSOC honeypot - config statique network: version: 2 ethernets: {iface}: dhcp4: no addresses: [{ip_cidr}] routes: - to: default via: {gateway} nameservers: addresses: [{', '.join(dns)}] """ f = Path("/etc/netplan/99-unisoc-honeypot.yaml") f.write_text(yaml) os.chmod(f, 0o600) for other in Path("/etc/netplan").glob("*.yaml"): if other.name != f.name: other.rename(other.with_suffix(".yaml.unisoc-bak")) subprocess.run(["netplan", "generate"], check=False, timeout=15) log(f"network: netplan static {ip_cidr} apposé (apply au reboot)") else: # ifupdown (Debian) interfaces_path = Path("/etc/network/interfaces") backup_path = Path("/etc/network/interfaces.unisoc-bak") if interfaces_path.exists() and not backup_path.exists(): backup_path.write_text(interfaces_path.read_text()) interfaces_path.write_text( "# Geré par UniSOC honeypot\n" "source /etc/network/interfaces.d/*\n\n" "auto lo\niface lo inet loopback\n" ) dropin = Path("/etc/network/interfaces.d/99-unisoc-honeypot") dropin.write_text( f"# UniSOC honeypot - config statique apposée à l'activation\n" f"allow-hotplug {iface}\n" f"iface {iface} inet static\n" f" address {ip_cidr}\n" f" gateway {gateway}\n" f" dns-nameservers {' '.join(dns)}\n" ) os.chmod(dropin, 0o644) log(f"network: ifupdown static {ip_cidr} apposé (apply au reboot)") except Exception as e: log(f"network apply échec : {e}") def trigger_reboot(): log("=== REBOOT demandé dans 5s ===") time.sleep(5) try: subprocess.run(["systemctl", "reboot"], timeout=10) except Exception as e: log(f"reboot échec : {e}") # fallback brutal os.system("reboot") def main() -> int: log("== sylink-honeypot-poll start ==") enroll = read_enroll() if not enroll or not enroll.get("token"): log("ERREUR : enroll.json absent — l'init a-t-il tourné ?") return 1 token = enroll["token"] if already_activated(): log("VM déjà activée — rien à faire, je sors.") return 0 log(f"polling toutes les {POLL_INTERVAL}s sur {PT_API_BASE}") while True: if already_activated(): log("license.json apparu hors poll — exit.") return 0 resp = poll_once(token) if resp and resp.get("activated"): log(f"ACTIVATION reçue : tenant={resp.get('tenant_id')} hp_id={resp.get('honeypot_id')}") write_license(resp) apply_hostname(resp.get("hostname")) apply_network_config(resp.get("network_config")) # Re-run l'init.py pour basculer le banner (mais le reboot va tout réinitialiser) try: subprocess.run(["/opt/sylink-honeypot/bin/init.py"], timeout=20) except Exception: pass trigger_reboot() return 0 time.sleep(POLL_INTERVAL) if __name__ == "__main__": sys.exit(main())