tecnotel
/
iride-setup
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
iride-setup/setup_server.py

680 lines
28 KiB
Python
Raw Normal View History

iride-setup 0.2.0 - bootstrap, first-setup, web installer 8888 con licenza, dev-setup runner
2026-09-12 12:54:45 +02:00
#!/usr/bin/env python3
"""
IRIDE — Web Installer Server (porta 8888)
Tecnotel Servizi SRL
Porting di argos-setup/setup_server.py. Self-contained: solo stdlib +
python3-cryptography (pacchetto di sistema) per la verifica della licenza.
Flusso: licenza (Ed25519 + machine_id + product) → clone di tecnotel/iride
con il token della licenza → virtualenv → iride.json / users.json → config
da template → migrazioni → build frontend (se presente) → SSL (Let's Encrypt,
certificato caricato o autofirmato) → nginx → unit systemd → licenza in
/opt/iride/data → chiusura della 8888 e rimozione di /opt/iride-setup-pkg.
"""
import hashlib
import json
import os
import secrets
import shutil
import signal
import subprocess
import threading
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
from urllib.parse import urlparse
APP_DIR = Path("/opt/iride/app")
CONFIG_DIR = Path("/opt/iride/config")
DATA_DIR = Path("/opt/iride/data")
LOGS_DIR = Path("/opt/iride/logs")
CERTS_DIR = Path("/opt/iride/certs")
BACKUP_DIR = Path("/opt/iride/backups")
SETUP_DIR = Path("/opt/iride/setup")
APP_USER = "iride"
PORT = 8888
PRODUCT = "iride"
# ── Licenza — chiave pubblica Ed25519 del vendor (raw 32 byte, base64) ────────
# Stessa costante di iride/backend/core/license.py e del portale Tecnotel.
_LICENSE_PUBLIC_KEY_B64 = "GMRsZMoxOlCBiJU66EsQcj0ZO0gVd0GHB5LelEo/hns="
# ── Clone: username Basic Auth del bot Gitea al quale appartiene il token ─────
GITEA_BOT_USER = "argos-portal-bot"
GITEA_REPO_PATH = "/tecnotel/iride.git"
# ── Heartbeat verso il portale vendor (stessa chiave di registrazione ARGOS) ──
VENDOR_HEARTBEAT_URL = "https://license.argosdefense.io"
VENDOR_INSTALL_KEY = "5b1ab5c872383f686d3a25a5e123adca"
SERVICES = ["iride-api", "iride-worker", "iride-scheduler"]
install_log = []
install_done = False
install_error = False
def get_machine_id() -> str:
"""Fingerprint univoco del server: SHA256 hex di
/etc/machine-id | hostname | MAC prima interfaccia fisica.
IDENTICO a iride/backend/core/license.py e a ARGOS core.get_machine_id()."""
import socket as _sock
parts = []
try:
with open("/etc/machine-id") as f:
parts.append(f.read().strip())
except Exception:
parts.append("")
try:
parts.append(_sock.gethostname())
except Exception:
parts.append("")
try:
r = subprocess.run(["cat", "/sys/class/net/eth0/address"], capture_output=True, text=True, timeout=2)
mac = r.stdout.strip()
if not mac or mac == "00:00:00:00:00:00":
r = subprocess.run(["ip", "-o", "link", "show"], capture_output=True, text=True, timeout=2)
for line in r.stdout.splitlines():
if "link/ether" in line and "00:00:00:00:00:00" not in line:
if "docker" in line or "br-" in line or "veth" in line:
continue
mac = line.split("link/ether")[1].split()[0].strip()
break
parts.append(mac or "")
except Exception:
parts.append("")
return hashlib.sha256("|".join(parts).encode()).hexdigest()
def verify_license(raw_bytes):
"""Firma Ed25519 + product + machine_id + scadenza + credenziali Gitea.
Ritorna (ok, license_dict, errore)."""
try:
raw = json.loads(raw_bytes)
except Exception as e:
return (False, None, f"File non è JSON valido: {e}")
if not isinstance(raw, dict):
return (False, None, "Formato licenza non riconosciuto")
try:
import base64
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
sig = raw.pop("signature", "")
if not sig:
return (False, None, "Licenza senza firma (campo 'signature' mancante)")
payload = json.dumps(raw, sort_keys=True, separators=(",", ":"))
raw["signature"] = sig
pub = Ed25519PublicKey.from_public_bytes(base64.b64decode(_LICENSE_PUBLIC_KEY_B64))
try:
pub.verify(base64.b64decode(sig), payload.encode())
except InvalidSignature:
return (False, None, "Firma non valida: licenza manomessa o emessa da un altro vendor.")
except ImportError:
return (False, None, "Libreria 'cryptography' non disponibile: apt install python3-cryptography")
except Exception as e:
return (False, None, f"Errore verifica firma: {e}")
product = raw.get("product")
if product is not None and product != PRODUCT:
return (False, None, f"Licenza per il prodotto '{product}', non per IRIDE.")
lic_machine = raw.get("machine_id", "")
if not lic_machine:
return (False, None, "Licenza senza machine_id: formato non supportato")
cur_machine = get_machine_id()
if lic_machine != cur_machine:
return (False, None, f"Machine ID non corrisponde: licenza per {lic_machine[:12]}..., "
f"questo server è {cur_machine[:12]}... La licenza non vale per questa macchina.")
expires = raw.get("expires_at", "")
if expires and expires < datetime.now().strftime("%Y-%m-%d"):
return (False, None, f"Licenza scaduta il {expires}")
if not raw.get("gitea_url") or not raw.get("gitea_token"):
return (False, None, "Licenza priva di credenziali Gitea: contattare Tecnotel per riemetterla.")
return (True, raw, "")
def log(msg):
line = f"[{datetime.now().strftime('%H:%M:%S')}] {msg}"
install_log.append(line)
print(line, flush=True)
def run(cmd, check=True):
log(f"$ {cmd}")
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
if result.stdout.strip(): log(result.stdout.strip()[-2000:])
if result.stderr.strip(): log(result.stderr.strip()[-2000:])
if check and result.returncode != 0:
raise RuntimeError(f"Comando fallito (exit {result.returncode}): {cmd}")
return result
def chown(path):
run(f"chown -R {APP_USER}:{APP_USER} {path}", check=False)
def venv_python() -> str:
return str(APP_DIR / "backend/venv/bin/python")
# ── Generazione configurazione ────────────────────────────────────────────────
def generate_iride_json(data):
hostname = data.get("domain", "").strip().lower()
aliases = [a.strip().lower() for a in data.get("aliases", "").split() if a.strip()]
provider = data.get("ai_provider", "fake") or "fake"
if provider not in ("fake", "openai", "anthropic"):
provider = "fake"
api_key = data.get("ai_api_key", "").strip()
cfg = {
"_version": "1.0",
"_installed": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
"cliente": {
"name": data.get("cliente_name", "").strip(),
"full_name": data.get("cliente_full", "").strip() or data.get("cliente_name", "").strip(),
"domain": hostname,
"email_domain": data.get("cliente_domain", "").strip().lower(),
"type": data.get("cliente_type", "customer_service"),
"ai_context": data.get("ai_context", "").strip(),
},
"network": {"hostname": hostname, "aliases": aliases},
"system": {
"secret_key": secrets.token_hex(32),
"internal_api_key": secrets.token_hex(24),
"timezone": "Europe/Rome",
"vendor_heartbeat": {
"enabled": True,
"url": VENDOR_HEARTBEAT_URL,
"install_key": VENDOR_INSTALL_KEY,
"interval_h": 6,
},
},
"ai": {
"text": {"provider": provider,
"options": {"model": data.get("ai_model", "").strip() or "gpt-4.1-mini", "temperature": 0.3}},
"realtime": {"provider": "openai", "options": {"model": ""}},
"providers": {
"openai": {"api_key": api_key if provider == "openai" else "", "base_url": "https://api.openai.com/v1"},
"anthropic": {"api_key": api_key if provider == "anthropic" else ""},
},
"limits": {"max_ai_turns": 20, "daily_budget_micros": 5000000},
},
"ports": {"api": 8080, "voice": 8081},
"paths": {"config_dir": str(CONFIG_DIR), "data_dir": str(DATA_DIR), "logs_dir": str(LOGS_DIR),
"backups": str(BACKUP_DIR)},
"backup": {"keep_snapshots": 14},
"license": {"file": str(DATA_DIR / "license.json")},
}
return cfg
def create_admin_user(data):
"""users.json con hash bcrypt calcolato dal backend IRIDE (stesso modulo
che lo verifica al login). La password passa via stdin, mai in argv."""
username = (data.get("admin_username") or "admin").strip().lower()
password = data.get("admin_password", "")
if not username or not password:
log("WARN: credenziali admin mancanti — skip creazione utente")
return
proc = subprocess.run(
[venv_python(), "-c", "import sys; from core.auth import hash_password; print(hash_password(sys.stdin.read()))"],
input=password, capture_output=True, text=True, cwd=str(APP_DIR / "backend"),
env={**os.environ, "PYTHONPATH": str(APP_DIR / "backend"), "IRIDE_CONFIG_DIR": str(CONFIG_DIR),
"IRIDE_DATA_DIR": str(DATA_DIR)})
if proc.returncode != 0:
raise RuntimeError(f"hash della password fallito: {proc.stderr.strip()[-300:]}")
pw_hash = proc.stdout.strip()
users_file = CONFIG_DIR / "users.json"
try:
users = json.loads(users_file.read_text())
except Exception:
users = {"users": {}}
users.setdefault("users", {})[username] = {
"password_hash": pw_hash, "roles": ["admin"], "totp_secret": "",
"email": (data.get("admin_email_user") or "").strip(), "enabled": True,
}
users_file.write_text(json.dumps(users, indent=2, ensure_ascii=False) + "\n")
os.chmod(users_file, 0o600)
chown(users_file)
log(f"Utente admin '{username}' creato in users.json")
# ── Installazione ─────────────────────────────────────────────────────────────
def install(data):
global install_done, install_error
try:
log("=== AVVIO INSTALLAZIONE IRIDE ===")
# 0. Licenza (già validata da /api/license/upload)
log("── Verifica licenza IRIDE ──")
lic_path = SETUP_DIR / "license.json"
if not lic_path.exists():
raise RuntimeError("license.json non trovata in /opt/iride/setup/: caricare una licenza valida.")
ok, lic, err = verify_license(lic_path.read_bytes())
if not ok:
raise RuntimeError(f"Licenza non valida: {err}")
gitea_url = lic.get("gitea_url", "").rstrip("/")
gitea_token = lic.get("gitea_token", "")
gitea_host = gitea_url[:-len("/api/v1")] if gitea_url.endswith("/api/v1") else gitea_url
log(f"Licenza OK: {lic.get('customer')} / {lic.get('tier')} / exp {lic.get('expires_at')}")
# 1. Clone (URL autenticato temporaneo: il token NON resta in .git/config)
log("── Clone repository IRIDE ──")
if (APP_DIR / ".git").exists():
log("Repository già presente — skip clone")
else:
auth_url = f"https://{GITEA_BOT_USER}:{gitea_token}@{gitea_host[len('https://'):]}{GITEA_REPO_PATH}"
APP_DIR.parent.mkdir(parents=True, exist_ok=True)
run(f"git config --global --add safe.directory {APP_DIR}")
run(f"git clone {auth_url} {APP_DIR}")
run(f"git -C {APP_DIR} remote set-url origin {gitea_host}{GITEA_REPO_PATH}")
chown(APP_DIR)
log(f"Repository IRIDE pronto ({(APP_DIR / 'VERSION').read_text().strip() if (APP_DIR / 'VERSION').exists() else '?'})")
# 2. Virtualenv
log("── Virtualenv Python ──")
venv_dir = APP_DIR / "backend/venv"
if not venv_dir.exists():
run(f"python3 -m venv {venv_dir}")
run(f"{venv_dir}/bin/pip install --upgrade pip -q")
run(f"{venv_dir}/bin/pip install -r {APP_DIR}/backend/requirements.txt -q")
chown(venv_dir)
log(f"Virtualenv pronto ({run(f'{venv_dir}/bin/python --version', check=False).stdout.strip()})")
# 3. iride.json
log("── Generazione iride.json ──")
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
cfg_path = CONFIG_DIR / "iride.json"
cfg_path.write_text(json.dumps(generate_iride_json(data), indent=2, ensure_ascii=False) + "\n")
os.chmod(cfg_path, 0o600)
chown(CONFIG_DIR)
log("iride.json creato")
# 4. Config da template .example (glob, idempotente: mai clobber)
_GENERATED = {"iride.json", "users.json"}
for src in sorted((APP_DIR / "config").glob("*.json.example")):
name = src.name[:-len(".example")]
if name in _GENERATED:
continue
dst = CONFIG_DIR / name
if dst.exists():
continue
shutil.copy(src, dst)
os.chmod(dst, 0o600)
chown(dst)
log(f"{name} copiato da template")
# 5. Logo cliente
logo_src = SETUP_DIR / "logo_cliente.png"
if logo_src.exists():
(CONFIG_DIR / "assets").mkdir(parents=True, exist_ok=True)
shutil.copy(logo_src, CONFIG_DIR / "assets" / "logo_cliente.png")
chown(CONFIG_DIR / "assets")
log("Logo cliente copiato")
# 6. Utente admin (bcrypt dal backend)
log("── Creazione utente admin ──")
create_admin_user(data)
# 7. Migrazioni DB
log("── Migrazioni database ──")
DATA_DIR.mkdir(parents=True, exist_ok=True)
chown(DATA_DIR)
run(f"cd {APP_DIR}/backend && sudo -u {APP_USER} env IRIDE_CONFIG_DIR={CONFIG_DIR} IRIDE_DATA_DIR={DATA_DIR} "
f"IRIDE_LOGS_DIR={LOGS_DIR} IRIDE_DB={DATA_DIR}/iride.db {venv_dir}/bin/python db.py")
log("Schema allineato")
# 8. Frontend (solo se presente: B-067)
if (APP_DIR / "frontend/package.json").exists():
log("── Build frontend ──")
run(f"cd {APP_DIR}/frontend && npm ci --silent")
run(f"cd {APP_DIR}/frontend && npm run build")
chown(APP_DIR / "frontend")
run(f"chmod 755 /opt/iride /opt/iride/app /opt/iride/app/frontend")
run(f"chmod -R 755 {APP_DIR}/frontend/dist/", check=False)
log("Frontend compilato")
else:
log("Nessun frontend/package.json: nginx serve la sola API (B-067)")
# 9. SSL — stesse tre modalità di ARGOS
log("── Configurazione SSL ──")
domain = data.get("domain", "").strip()
aliases = data.get("aliases", "").strip()
ssl_mode = data.get("ssl_mode", "letsencrypt")
all_names = (domain + " " + aliases).strip()
CERTS_DIR.mkdir(parents=True, exist_ok=True)
if ssl_mode == "manual":
crt_src, key_src = SETUP_DIR / "uploaded.crt", SETUP_DIR / "uploaded.key"
if not crt_src.exists() or not key_src.exists():
raise RuntimeError("File SSL .crt o .key non trovati in /opt/iride/setup/")
shutil.copy(crt_src, CERTS_DIR / "fullchain.pem")
shutil.copy(key_src, CERTS_DIR / "privkey.pem")
os.chmod(CERTS_DIR / "privkey.pem", 0o600)
ssl_crt, ssl_key = str(CERTS_DIR / "fullchain.pem"), str(CERTS_DIR / "privkey.pem")
log("Certificato SSL caricato dal wizard")
elif ssl_mode == "selfsigned":
log("Generazione certificato autofirmato (RSA 4096, validità 10 anni)")
crt_path, key_path, cnf_path = CERTS_DIR / "fullchain.pem", CERTS_DIR / "privkey.pem", CERTS_DIR / "openssl-selfsigned.cnf"
san_dns = [n for n in all_names.split() if n]
try:
server_ip = subprocess.check_output(["hostname", "-I"], text=True).strip().split()[0]
except Exception:
server_ip = ""
san_lines = "\n".join(f"DNS.{i+1} = {n}" for i, n in enumerate(san_dns)) or "DNS.1 = iride.local"
if server_ip:
san_lines += f"\nIP.1 = {server_ip}"
client_full = data.get("cliente_full") or data.get("cliente_name") or "IRIDE"
cn = domain or "iride.local"
cnf_path.write_text(f"""[req]
default_bits = 4096
prompt = no
default_md = sha256
distinguished_name = dn
req_extensions = req_ext
x509_extensions = v3_ext
[dn]
C = IT
O = {client_full}
OU = IRIDE
CN = {cn}
[req_ext]
subjectAltName = @alt_names
[v3_ext]
subjectAltName = @alt_names
basicConstraints = critical, CA:FALSE
keyUsage = critical, digitalSignature, keyEncipherment
extendedKeyUsage = serverAuth
[alt_names]
{san_lines}
""")
run(f"openssl req -x509 -nodes -days 3650 -newkey rsa:4096 -keyout {key_path} -out {crt_path} -config {cnf_path}")
os.chmod(key_path, 0o600)
os.chmod(crt_path, 0o644)
ssl_crt, ssl_key = str(crt_path), str(key_path)
log(f"Certificato autofirmato generato (CN={cn}, SAN: {len(san_dns)} DNS{' + 1 IP' if server_ip else ''})")
log("ATTENZIONE: i browser lo segnaleranno come non attendibile; per WhatsApp serve un certificato valido.")
else:
_write_nginx_http(all_names)
run("nginx -t && systemctl restart nginx")
certbot_d = " ".join(f"-d {n}" for n in all_names.split())
email = data.get("admin_email", "admin@tecnotelsrl.com")
run(f"certbot --nginx {certbot_d} --non-interactive --agree-tos -m {email}")
ssl_crt = f"/etc/letsencrypt/live/{domain}/fullchain.pem"
ssl_key = f"/etc/letsencrypt/live/{domain}/privkey.pem"
log("Certificato Let's Encrypt ottenuto")
chown(CERTS_DIR)
# 10. Nginx finale
log("── Nginx configurazione finale ──")
_write_nginx_final(all_names or "_", ssl_crt, ssl_key)
run("nginx -t && systemctl restart nginx")
log("Nginx configurato")
# 11. Servizi systemd (unit dal repo: update.sh le tiene allineate)
log("── Creazione e avvio servizi ──")
for svc in SERVICES:
shutil.copy(APP_DIR / "deploy/systemd" / f"{svc}.service", f"/etc/systemd/system/{svc}.service")
if (APP_DIR / "deploy/logrotate/iride").exists():
shutil.copy(APP_DIR / "deploy/logrotate/iride", "/etc/logrotate.d/iride")
run("systemctl daemon-reload")
LOGS_DIR.mkdir(parents=True, exist_ok=True)
chown(LOGS_DIR)
for svc in SERVICES:
run(f"systemctl enable --now {svc}")
log(f"{svc} avviato")
# 12. Licenza in posizione finale
log("── Copia licenza in posizione finale ──")
final_lic = DATA_DIR / "license.json"
shutil.copy(lic_path, final_lic)
os.chmod(final_lic, 0o600)
chown(final_lic)
log(f"Licenza copiata in {final_lic}")
# 13. Health check e strumento di verifica permanente
health = Path(__file__).parent / "checks" / "health.sh"
if health.exists():
shutil.copy(health, "/usr/local/bin/iride-health")
os.chmod("/usr/local/bin/iride-health", 0o755)
res = run("bash /usr/local/bin/iride-health", check=False)
if res.returncode != 0:
log("ATTENZIONE: health check con errori (vedi sopra): verificare i log in /opt/iride/logs")
# 14. Chiusura web installer
log("── Chiusura web installer ──")
run("systemctl disable --now iride-setup", check=False)
run("ufw delete allow 8888/tcp", check=False)
log("Porta 8888 chiusa — web installer disabilitato")
log("=== INSTALLAZIONE COMPLETATA ===")
_schedule_cleanup()
install_done = True
def shutdown():
import time
time.sleep(15)
os.kill(os.getpid(), signal.SIGTERM)
threading.Thread(target=shutdown, daemon=True).start()
except Exception as e:
log(f"ERRORE: {e}")
install_log.append(f"__ERROR__: {e}")
install_error = True
# ── Nginx ─────────────────────────────────────────────────────────────────────
def _write_nginx_http(all_names):
_write_nginx_conf(f"""server {{
listen 80;
server_name {all_names};
location /.well-known/acme-challenge/ {{ root /var/www/html; }}
location / {{ return 301 https://$host$request_uri; }}
}}
""")
def _write_nginx_final(all_names, ssl_crt, ssl_key):
has_frontend = (APP_DIR / "frontend/dist/index.html").exists()
root_block = (f""" location / {{
root {APP_DIR}/frontend/dist;
try_files $uri $uri/ /index.html;
expires 1h;
}}""" if has_frontend else
""" location = / {
return 200 'IRIDE installata. Portale in arrivo (B-067). API: /api/health';
add_header Content-Type text/plain;
}""")
conf = f"""limit_req_zone $binary_remote_addr zone=iride:10m rate=20r/s;
server {{
listen 80;
server_name {all_names};
location / {{ return 301 https://$host$request_uri; }}
location /.well-known/acme-challenge/ {{ root /var/www/html; }}
}}
server {{
listen 443 ssl http2;
server_name {all_names};
ssl_certificate {ssl_crt};
ssl_certificate_key {ssl_key};
ssl_protocols TLSv1.2 TLSv1.3;
ssl_ciphers HIGH:!aNULL:!MD5;
add_header X-Frame-Options SAMEORIGIN;
add_header X-Content-Type-Options nosniff;
add_header Strict-Transport-Security "max-age=31536000" always;
client_max_body_size 25m;
# Webhook dei canali: nessun rate limit per IP (Meta e Telegram arrivano da pochi IP)
location /api/v1/webhooks/ {{
proxy_pass http://127.0.0.1:8080;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto https;
proxy_read_timeout 30s;
}}
location /api/ {{
limit_req zone=iride burst=40 nodelay;
proxy_pass http://127.0.0.1:8080;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto https;
proxy_connect_timeout 30s;
proxy_send_timeout 120s;
proxy_read_timeout 120s;
}}
# WebSocket dell'inbox operatori (B-016)
location /ws/ {{
proxy_pass http://127.0.0.1:8080;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_set_header Host $host;
proxy_read_timeout 3600s;
}}
# Relay vocale (POC-01): attivare solo con iride-voice installato
# location /v1/voice/ {{
# proxy_pass http://127.0.0.1:8081;
# proxy_http_version 1.1;
# proxy_set_header Upgrade $http_upgrade;
# proxy_set_header Connection "upgrade";
# proxy_read_timeout 3600s;
# proxy_send_timeout 3600s;
# }}
location /widget/ {{
alias {APP_DIR}/frontend/packages/widget/dist/;
add_header Cache-Control "public, max-age=3600";
}}
{root_block}
access_log {LOGS_DIR}/nginx-access.log;
error_log {LOGS_DIR}/nginx-error.log;
}}
"""
_write_nginx_conf(conf)
def _write_nginx_conf(conf):
Path("/etc/nginx/sites-available/iride").write_text(conf)
p = Path("/etc/nginx/sites-enabled/iride")
if not p.exists():
p.symlink_to("/etc/nginx/sites-available/iride")
for f in ["/etc/nginx/sites-enabled/default", "/etc/nginx/sites-enabled/iride-setup"]:
if Path(f).exists():
Path(f).unlink()
# ── HTTP ──────────────────────────────────────────────────────────────────────
class SetupHandler(BaseHTTPRequestHandler):
def log_message(self, *args):
pass
def do_GET(self):
path = urlparse(self.path).path
if path in ("/", "/setup"):
html_path = Path(__file__).parent / "setup.html"
if not html_path.exists():
self.send_response(404); self.end_headers(); return
html = html_path.read_bytes()
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", len(html))
self.end_headers()
self.wfile.write(html)
elif path == "/api/status":
self._json({"done": install_done, "error": install_error, "log": install_log[-60:]})
elif path == "/api/machine-id":
self._json({"machine_id": get_machine_id()})
else:
self.send_response(404); self.end_headers()
def do_POST(self):
path = urlparse(self.path).path
length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(length)
if path == "/api/install":
try:
data = json.loads(body)
threading.Thread(target=install, args=(data,), daemon=True).start()
self._json({"ok": True})
except Exception as e:
self._json({"ok": False, "error": str(e)}, 400)
elif path in ("/api/upload/cert", "/api/upload/key", "/api/upload/logo"):
SETUP_DIR.mkdir(parents=True, exist_ok=True)
name = {"cert": "uploaded.crt", "key": "uploaded.key", "logo": "logo_cliente.png"}[path.rsplit("/", 1)[1]]
(SETUP_DIR / name).write_bytes(body)
os.chmod(SETUP_DIR / name, 0o600)
self._json({"ok": True})
elif path == "/api/license/upload":
ok, lic, err = verify_license(body)
if not ok:
self._json({"ok": False, "error": err}, 400)
return
SETUP_DIR.mkdir(parents=True, exist_ok=True)
lic_path = SETUP_DIR / "license.json"
lic_path.write_bytes(body)
os.chmod(lic_path, 0o600)
self._json({"ok": True, "summary": {
"customer": lic.get("customer", ""),
"tier": lic.get("tier", ""),
"issued_to": lic.get("issued_to", ""),
"issued_at": lic.get("issued_at", ""),
"expires_at": lic.get("expires_at", ""),
"product": lic.get("product", "") or "(non indicato)",
"has_gitea": bool(lic.get("gitea_token")),
}})
else:
self.send_response(404); self.end_headers()
def _json(self, data, code=200):
body = json.dumps(data).encode()
self.send_response(code)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", len(body))
self.end_headers()
self.wfile.write(body)
def _schedule_cleanup():
"""Disabilita iride-setup e rimuove /opt/iride-setup-pkg via transient
unit systemd-run, indipendente dal processo padre (che sta per morire)."""
script = r"""#!/bin/bash
sleep 5
systemctl stop iride-setup.service 2>/dev/null || true
systemctl disable iride-setup.service 2>/dev/null || true
rm -f /etc/systemd/system/iride-setup.service
systemctl daemon-reload
ufw delete allow 8888/tcp 2>/dev/null || true
rm -rf /opt/iride-setup-pkg
echo "iride-setup cleanup completato $(date -Iseconds)" >> /var/log/iride-setup-cleanup.log
"""
script_path = "/tmp/iride-setup-cleanup.sh"
try:
import time
Path(script_path).write_text(script)
os.chmod(script_path, 0o755)
unit_name = f"iride-setup-cleanup-{int(time.time())}.service"
subprocess.Popen(["systemd-run", "--no-block", "--unit", unit_name, "/bin/bash", script_path],
stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
start_new_session=True)
log(f"Cleanup schedulato via systemd-run come {unit_name} (delay 5s)")
except Exception as e:
log(f"Errore schedulazione cleanup: {e}")
if __name__ == "__main__":
print(f"\n{'='*55}\n IRIDE — Web Installer\n Tecnotel Servizi SRL\n Apri: http://<IP_SERVER>:{PORT}\n{'='*55}\n")
HTTPServer(("0.0.0.0", PORT), SetupHandler).serve_forever()