4ventures-client-portal/scripts/dokploy_apply_traefik_harden.py

#!/usr/bin/env python3
"""Apply Traefik harden patches (office IP allowlist). Does NOT touch portal.yml / app / supabase."""
from __future__ import annotations

import re
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path

from playwright.sync_api import sync_playwright

BPS = Path("/Users/4ventures/Projects/bps-facturacion")
ROOT = Path("/Users/4ventures/4ventures-client-portal")
PATCH = ROOT / ".ops" / "traefik-patches"
SHOT = ROOT / "scripts" / "dokploy-shots"
BASE = "https://dokploy.4ventures.es"

# (ui label in Traefik file list, local patch file)
FILES = [
    ("middleware-hardening.yml", "middleware-hardening.yml"),
    ("gitea.yml", "gitea.yml"),
    ("n8n-hardened.yml", "n8n-hardened.yml"),
    ("rutas-manuales.yml", "rutas-manuales.yml"),
]


def env(key: str) -> str:
    import os

    v = os.environ.get(key, "").strip()
    if v:
        return v
    p = BPS / ".env"
    if not p.is_file():
        return ""
    for line in p.read_text(encoding="utf-8").splitlines():
        if line.startswith(f"{key}="):
            return line.split("=", 1)[1].strip().strip('"').strip("'")
    return ""


def set_monaco(page, val: str) -> str:
    mode = page.evaluate(
        """(val) => {
          const m = window.monaco;
          if (m && m.editor) {
            const eds = m.editor.getEditors();
            if (eds && eds[0]) { eds[0].setValue(val); return 'monaco'; }
          }
          const cm = document.querySelector('.cm-content');
          if (cm) {
            cm.focus();
            return 'cm';
          }
          return 'none';
        }""",
        val,
    )
    if mode == "cm":
        page.locator(".cm-content").first.click(force=True)
        page.keyboard.press("Meta+A")
        page.keyboard.insert_text(val)
    return mode


def probe(url: str) -> tuple[int, str]:
    try:
        req = urllib.request.Request(url, method="GET", headers={"User-Agent": "4v-harden-probe"})
        with urllib.request.urlopen(req, timeout=15) as r:
            return r.status, (r.read(80) or b"").decode("utf-8", "replace")
    except urllib.error.HTTPError as e:
        return e.code, (e.read(80) or b"").decode("utf-8", "replace")
    except Exception as e:
        return 0, type(e).__name__


def main() -> int:
    email = env("DOKPLOY_EMAIL") or "marketing@4ventures.es"
    password = env("DOKPLOY_PASSWORD")
    if not password:
        print("NO_DOKPLOY_PASSWORD", flush=True)
        return 2
    SHOT.mkdir(parents=True, exist_ok=True)

    for label, fname in FILES:
        if not (PATCH / fname).is_file():
            print("MISSING_PATCH", fname, flush=True)
            return 3

    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page(viewport={"width": 1600, "height": 1200})
        page.goto(BASE + "/", wait_until="domcontentloaded", timeout=90000)
        page.wait_for_timeout(700)
        if "/dashboard" not in page.url:
            page.locator('input[type="email"], input[name="email"]').first.fill(email)
            page.locator('input[type="password"]').first.fill(password)
            page.locator('button[type="submit"]').first.click()
            page.wait_for_url(re.compile(r"/dashboard"), timeout=90000)
        print("login ok", flush=True)

        page.goto(f"{BASE}/dashboard/traefik", wait_until="networkidle", timeout=120000)
        page.wait_for_timeout(1000)
        page.get_by_text(re.compile(r"^dynamic$", re.I)).first.click()
        page.wait_for_timeout(900)

        for label, fname in FILES:
            yaml = (PATCH / fname).read_text(encoding="utf-8")
            # find file in list
            loc = page.get_by_text(re.compile(rf"^{re.escape(label)}$", re.I))
            if not loc.count():
                loc = page.locator(f"text={label}")
            if not loc.count():
                print("FILE_MISSING_IN_UI", label, flush=True)
                page.screenshot(path=str(SHOT / f"harden_missing_{fname}.png"), full_page=True)
                browser.close()
                return 4
            loc.first.dblclick()
            page.wait_for_timeout(1200)
            unlock = page.get_by_role("button", name=re.compile(r"^Unlock$", re.I))
            if unlock.count():
                unlock.first.click()
                page.wait_for_timeout(800)
            mode = set_monaco(page, yaml)
            print("set", label, mode, "chars", len(yaml), flush=True)
            upd = page.get_by_role("button", name=re.compile(r"^Update$", re.I))
            if not upd.count():
                upd = page.get_by_role("button", name=re.compile(r"Update|Save", re.I))
            upd.first.click()
            page.wait_for_timeout(2000)
            page.screenshot(path=str(SHOT / f"harden_applied_{fname.replace('.','_')}.png"), full_page=True)
            # back to file list
            page.goto(f"{BASE}/dashboard/traefik", wait_until="networkidle", timeout=120000)
            page.wait_for_timeout(700)
            page.get_by_text(re.compile(r"^dynamic$", re.I)).first.click()
            page.wait_for_timeout(700)

        browser.close()

    time.sleep(3)
    checks = [
        ("git", "https://git.4ventures.es/"),
        ("n8n_ui", "https://n8n.4ventures.es/"),
        ("n8n_agencia_ui", "https://n8n-agencia.4ventures.es/"),
        ("app", "https://app.4ventures.es/api/health"),
        ("dokploy", "https://dokploy.4ventures.es/"),
    ]
    ok = True
    for name, url in checks:
        code, head = probe(url)
        print(f"probe {name} {code} {head[:60]!r}", flush=True)
        # From allowlisted IP: git/n8n should NOT be 403
        if name in ("git", "n8n_ui", "n8n_agencia_ui") and code == 403:
            ok = False
        if name == "app" and code != 200:
            ok = False
    print("SUCCESS" if ok else "CHECK_FAILED", flush=True)
    return 0 if ok else 5


if __name__ == "__main__":
    raise SystemExit(main())