#!/usr/bin/env python3 """ v2_dashboard.py - Read-only dashboard per Account 2 (live_deploy_v2). VINCOLI Mattia 2026-08-05: - Read-only: NO send/close/modify ordini, NO restart servizi, NO trigger webhook. - Data sources: SOLO /opt/charter-live/live_deploy_v2/ (no cross-contamination con 1° account). - NO proxy a 5503/5504/5510 (servizi inesistenti su v2). - /v2 → 5581 (webhook) resta intatto. - POST non ammesso. ENDPOINTS: - GET / → HTML dashboard - GET /json → JSON dump di tutti i dati - GET /healthz → 200 OK con timestamp Mavis 2026-08-05. """ import os import sys import json import time import hmac import base64 import subprocess from http.server import HTTPServer, BaseHTTPRequestHandler from datetime import datetime, timezone # ==== LOCKED: paths v2 ==== V2_DIR = "/opt/charter-live/live_deploy_v2" LOG_DIR = f"{V2_DIR}/logs" WEBHOOK_LOG = f"{LOG_DIR}/webhook_receiver.log" ORDERS_LOG = f"{LOG_DIR}/orders.log" AC_STATE = f"{LOG_DIR}/regime_state.json" RETT_STATE = f"{LOG_DIR}/rettangolo_signal_state_v2.json" ENV_FILE = f"{V2_DIR}/.env.vps" # ==== Basic Auth (opzionale via env) ==== # Se V2_DASH_USER e V2_DASH_PASS sono settate (via systemd EnvironmentFile), # tutte le richieste richiedono Basic Auth. /healthz resta aperto per monitoring. AUTH_USER = os.environ.get("V2_DASH_USER", "").strip() AUTH_PASS = os.environ.get("V2_DASH_PASS", "").strip() AUTH_ENABLED = os.environ.get("V2_DASH_AUTH_ENABLED", "0").strip().lower() in {"1", "true", "yes"} and bool(AUTH_USER and AUTH_PASS) # porta PORT = int(os.environ.get("V2_DASHBOARD_PORT", "10002")) # ============= utility ============= def safe_read_file(path, max_bytes=16384): try: with open(path, "r", encoding="utf-8", errors="replace") as f: return f.read(max_bytes) except Exception as e: return None # file may not exist def load_v2_env(): """Carica credenziali SOLO da .env.vps. Mai da altri file.""" env = {} try: with open(ENV_FILE, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line or line.startswith("#") or "=" not in line: continue k, v = line.split("=", 1) env[k.strip()] = v.strip() except Exception as e: return None, f"env read err: {e}" return env, None # ============= data sources ============= def get_vps_health(): out = {"uptime": "?", "load": "?", "memory": "?"} try: out["uptime"] = subprocess.check_output(["uptime", "-p"], text=True, timeout=5).strip() except Exception: pass try: with open("/proc/loadavg") as f: out["load"] = f.read().split()[0] except Exception: pass try: with open("/proc/meminfo") as f: lines = f.readlines() total = int(lines[0].split()[1]) // 1024 avail = int(lines[2].split()[1]) // 1024 out["memory"] = f"{total - avail}/{total} MB" except Exception: pass return out def get_v2_services(): """SOLO charter-v2-* (no altri servizi).""" services = ["charter-v2-webhook", "charter-v2-rettangolo"] result = {} for svc in services: try: r = subprocess.run( ["systemctl", "is-active", f"{svc}.service"], capture_output=True, text=True, timeout=5 ) state = r.stdout.strip() result[svc] = "RUNNING" if state == "active" else "STOPPED" except Exception as e: result[svc] = f"ERR: {e}" return result def get_account_info(): """Read-only Bybit: balance + posizioni SOLO account 2.""" out = {"balance_usdt": "?", "equity_usdt": "?", "positions": [], "error": None} env, err = load_v2_env() if err: out["error"] = err return out api_key = env.get("BYBIT_HETZNER_API_KEY") api_secret = env.get("BYBIT_HETZNER_SECRET") if not api_key or not api_secret: out["error"] = "BYBIT_HETZNER creds missing in .env.vps" return out # bybit_demo_client si aspetta BYBIT_DEMO_API_KEY e BYBIT_DEMO_SECRET_KEY. # mappiamo da .env.vps SOLO in os.environ (no file modifiche). os.environ["BYBIT_DEMO_API_KEY"] = api_key os.environ["BYBIT_DEMO_SECRET_KEY"] = api_secret # patch temporanea del ENV_FILE che bybit_demo_client cerca di caricare # (è hardcoded a /opt/charter-live/API_KEY_BYBIT.env, che contiene le creds del 1° account) # ⇒ scriviamo un ENV_FILE temporaneo v2 e facciamo cred-by-cred override # Alternativa sicura: costruiamo una env temporanea e patch via cwd. # Soluzione più semplice: import bybit_demo_client, creare client, e # POI settare le creds sul client instance PRIMA di fetch_balance/fetch_positions. try: # aggiungi v2 a path per importare il client sys.path.insert(0, V2_DIR) from bybit_demo_client import BybitDemoClient # noqa # forza ENV_FILE temporaneo per non caricare il 1° account import bybit_demo_client as bdc # patch runtime: ENV_FILE del client bdc.ENV_FILE = type(bdc.ENV_FILE)(ENV_FILE + ".__dummy__") # fa fallire load_dotenv client = BybitDemoClient.__new__(BybitDemoClient) client.api_key = api_key client.secret = api_secret import requests as _req client.session = _req.Session() except Exception as e: out["error"] = f"client init: {e}" return out # balance try: bal = client.fetch_balance() coins = bal.get("coin", []) if isinstance(bal, dict) else [] for coin in coins: if coin.get("coin") == "USDT": out["balance_usdt"] = coin.get("walletBalance", "?") out["equity_usdt"] = coin.get("equity", "?") break except Exception as e: out["balance_usdt"] = "err" out["error"] = f"balance: {e}" # positions try: positions = client.fetch_positions() or [] for p in positions: out["positions"].append({ "symbol": p.get("symbol"), "side": p.get("side"), "size": p.get("size"), "avgPrice": p.get("avgPrice"), "markPrice": p.get("markPrice"), "unrealisedPnl": p.get("unrealisedPnl"), "leverage": p.get("leverage"), }) except Exception as e: if not out["error"]: out["error"] = f"positions: {e}" return out def get_recent_orders(n=10): """Ultimi N ordini da orders.log (write-only del webhook).""" content = safe_read_file(ORDERS_LOG) if content is None: return [] lines = [l.strip() for l in content.splitlines() if l.strip()] return lines[-n:] def get_webhook_stats(): """Stats da webhook_receiver.log: ACCEPTED, REJECTED, ultimi eventi. Redazione preventiva di secret/api_key/token nei log mostrati.""" import re as _re SECRET_PATTERNS = [ (_re.compile(r'(secret[\'":= ]+)[\w\-]{4,}', _re.IGNORECASE), r'\1'), (_re.compile(r'(api[_-]?key[\'":= ]+)[\w\-]{4,}', _re.IGNORECASE), r'\1'), (_re.compile(r'(token[\'":= ]+)[\w\-]{4,}', _re.IGNORECASE), r'\1'), (_re.compile(r'(password[\'":= ]+)[\w\-]{4,}', _re.IGNORECASE), r'\1'), (_re.compile(r'(auth[\'":= ]+)[\w\-]{4,}', _re.IGNORECASE), r'\1'), ] def redact(text: str) -> str: for pat, repl in SECRET_PATTERNS: text = pat.sub(repl, text) return text out = {"accepted": 0, "rejected": 0, "wrong_secret": 0, "missing_field": 0, "last_entries": []} content = safe_read_file(WEBHOOK_LOG) if content is None: out["error"] = "log not readable" return out out["accepted"] = content.count("ACCEPTED") out["rejected"] = content.count("REJECTED") out["wrong_secret"] = content.count("WRONG_SECRET") out["missing_field"] = content.count("missing") lines = [l.strip() for l in content.splitlines() if l.strip()] # ultimi 5 eventi significativi (no DB init/no pending) interesting = [l for l in lines if any(k in l for k in ("ACCEPTED", "REJECT", "ERROR", "WARNING"))] # redazione + truncate out["last_entries"] = [redact(l)[:300] for l in interesting[-5:]] return out def get_anti_cluster_state(): """Stato anti-cluster-loss + regime + signal state v2.""" out = {} for label, path in [("regime_state", AC_STATE), ("rettangolo_signal", RETT_STATE)]: content = safe_read_file(path, max_bytes=4096) if content is None: out[label] = None continue try: out[label] = json.loads(content) except Exception as e: out[label] = f"parse err: {e}" return out # ============= HTML rendering ============= CSS = """ body{font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',sans-serif;background:#0a0e27;color:#e0e0e0;padding:20px;margin:0;line-height:1.5;} h1{color:#4fc3f7;margin:0 0 5px 0;} h2{color:#4fc3f7;margin:25px 0 10px 0;border-bottom:1px solid #1a1f3a;padding-bottom:5px;} h3{color:#80cbc4;margin:15px 0 8px 0;} .card{background:#1a1f3a;padding:15px 20px;border-radius:6px;margin:10px 0;} .ok{color:#4caf50;font-weight:bold;}.err{color:#f44336;font-weight:bold;}.warn{color:#ff9800;font-weight:bold;} code{background:#0a0e27;padding:2px 6px;border-radius:3px;color:#4fc3f7;font-size:0.9em;font-family:'Cascadia Code','Consolas',monospace;} pre{background:#0a0e27;padding:10px;border-radius:4px;overflow-x:auto;color:#b0bec5;font-size:0.85em;} table{border-collapse:collapse;width:100%;margin-top:8px;} th,td{padding:8px 10px;text-align:left;border-bottom:1px solid #1a1f3a;} th{color:#4fc3f7;background:#0a0e27;font-weight:600;} .tag{display:inline-block;background:#f44336;color:white;padding:3px 10px;border-radius:3px;font-size:0.7em;letter-spacing:0.5px;margin-left:10px;vertical-align:middle;} .tag-blue{background:#1976d2;} .empty{color:#666;font-style:italic;} .footer{color:#555;font-size:0.8em;margin-top:30px;text-align:center;} """ def render_html(vps, services, account, orders, webhook, ac_state): # services svc_html = "
    " for s, status in services.items(): cls = "ok" if status == "RUNNING" else "err" svc_html += f"
  • {s} → {status}
    • " svc_html += "
" # balance / equity bal = account.get("balance_usdt", "?") eq = account.get("equity_usdt", "?") acc_err = account.get("error") err_html = f"

{acc_err}

" if acc_err else "" # positions if account.get("positions"): pos_html = "" for p in account["positions"]: pnl = p.get("unrealisedPnl", "?") pnl_cls = "ok" if isinstance(pnl, (int, float)) and pnl >= 0 else "err" pos_html += "" pos_html += f"" pos_html += f"" pos_html += f"" pos_html += f"" pos_html += f"" pos_html += f"" pos_html += f"" pos_html += "" pos_html += "
SymbolSideSizeAvg PriceMarkPnLLev
{p.get('symbol','?')}{p.get('side','?')}{p.get('size','?')}{p.get('avgPrice','?')}{p.get('markPrice','?')}{pnl}{p.get('leverage','?')}
" else: pos_html = "

Nessuna posizione aperta

" # orders if orders: orders_html = "
    " for o in orders: orders_html += f"
  • {o[:300]}
    • " orders_html += "
" else: orders_html = "

orders.log vuoto

" # auth notice auth_badge = "" if AUTH_ENABLED: auth_badge = ' BASIC AUTH' # webhook wh_html = ( f"

ACCEPTED: {webhook.get('accepted', 0)} | " f"REJECTED: {webhook.get('rejected', 0)} " f"(WRONG_SECRET: {webhook.get('wrong_secret', 0)}, missing: {webhook.get('missing_field', 0)})

" ) if webhook.get("last_entries"): wh_html += "
    " for e in webhook["last_entries"]: wh_html += f"
  • {e[:300]}
    • " wh_html += "
" else: wh_html += "

Nessun evento recente

" # anti-cluster regime = ac_state.get("regime_state") rett = ac_state.get("rettangolo_signal") regime_html = f"
{json.dumps(regime, indent=2, ensure_ascii=False) if regime else 'N/A'}
" rett_html = f"
{json.dumps(rett, indent=2, ensure_ascii=False) if rett else 'N/A'}
" # timestamp ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") html = f""" Charter Account 2 — Dashboard Read-Only

Charter Account 2 Demo READ-ONLY v2{auth_badge}

Data sources locked to {V2_DIR} · No order actions · Refresh 15s · {ts}

VPS Health

Uptime: {vps.get('uptime','?')} · Load (1m): {vps.get('load','?')} · Memory used: {vps.get('memory','?')}

v2 Services (solo charter-v2-*)

{svc_html}

Account 2 — Bybit (read-only)

Balance USDT: {bal} · Equity USDT: {eq}

{err_html}

Open Positions

{pos_html}

Recent Orders (orders.log, ultimi {len(orders)})

{orders_html}

Webhook Stats (webhook_receiver.log)

{wh_html}

Regime State

{regime_html}

Rettangolo Signal State v2

{rett_html}

Charter v2 dashboard · v2_dashboard.py · read-only · no proxy to 5503/5504/5510 · /v2 → 5581 webhook intatto

""" return html # ============= HTTP handler ============= class V2DashboardHandler(BaseHTTPRequestHandler): def do_GET(self): # /healthz resta SEMPRE aperto per monitoring (anche se AUTH_ENABLED) if self.path == "/healthz": self._send_json(200, {"ok": True, "service": "v2-dashboard-readonly", "ts": datetime.now(timezone.utc).isoformat(), "auth": "enabled" if AUTH_ENABLED else "disabled"}) return # Auth check per tutte le altre route if AUTH_ENABLED and not self._check_auth(): self._send_unauthorized() return if self.path == "/json": try: data = { "ts": datetime.now(timezone.utc).isoformat(), "vps": get_vps_health(), "services": get_v2_services(), "account": get_account_info(), "orders_last": get_recent_orders(10), "webhook": get_webhook_stats(), "anti_cluster": get_anti_cluster_state(), } self._send_json(200, data) except Exception as e: self._send_json(500, {"error": str(e)}) return if self.path in ("/", "/index.html"): try: vps = get_vps_health() services = get_v2_services() account = get_account_info() orders = get_recent_orders(10) webhook = get_webhook_stats() ac_state = get_anti_cluster_state() html = render_html(vps, services, account, orders, webhook, ac_state) self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Cache-Control", "no-store") body = html.encode("utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) except Exception as e: self._send_json(500, {"error": str(e)}) return # qualsiasi altro path self._send_json(404, {"error": "not found (read-only dashboard)"}) def do_POST(self): # blocca TUTTE le POST: read-only self._send_json(405, {"error": "method not allowed (read-only)"}) def do_PUT(self): # Auth check anche su PUT (anche se viene rifiutato dopo) if AUTH_ENABLED and not self._check_auth(): self._send_unauthorized() return self._send_json(405, {"error": "method not allowed (read-only)"}) def do_DELETE(self): if AUTH_ENABLED and not self._check_auth(): self._send_unauthorized() return self._send_json(405, {"error": "method not allowed (read-only)"}) def _check_auth(self) -> bool: """Verifica Basic Auth. Constant-time compare.""" if not AUTH_ENABLED: return True auth = self.headers.get("Authorization", "") if not auth.startswith("Basic "): return False try: decoded = base64.b64decode(auth[6:]).decode("utf-8", errors="replace") user, _, pwd = decoded.partition(":") user_ok = hmac.compare_digest(user.encode("utf-8"), AUTH_USER.encode("utf-8")) pwd_ok = hmac.compare_digest(pwd.encode("utf-8"), AUTH_PASS.encode("utf-8")) return user_ok and pwd_ok except Exception: return False def _send_unauthorized(self): body = b"401 Unauthorized\n" self.send_response(401) self.send_header("WWW-Authenticate", 'Basic realm="Charter v2 Dashboard", charset="UTF-8"') self.send_header("Content-Type", "text/plain; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _send_json(self, code, data): body = json.dumps(data, default=str).encode("utf-8") 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 log_message(self, fmt, *args): # silenzia log per non intasare journald pass def main(): auth_status = "ENABLED" if AUTH_ENABLED else "DISABLED" print(f"[v2-dashboard] READ-ONLY listening on 0.0.0.0:{PORT}") print(f"[v2-dashboard] data source: {V2_DIR}") print(f"[v2-dashboard] Basic Auth: {auth_status}") print(f"[v2-dashboard] NO actions. NO proxy to 5503/5504/5510. /v2 → 5581 intact.") httpd = HTTPServer(("0.0.0.0", PORT), V2DashboardHandler) httpd.serve_forever() if __name__ == "__main__": main()