ArchiveBox/archivebox/tests/test_server_security_browser.py

#!/usr/bin/env python3
"""Browser-level security mode tests using the existing Node/Puppeteer runtime."""

from __future__ import annotations

import json
import os
import time
import subprocess
import textwrap
from pathlib import Path
from urllib.parse import urlencode

import pytest
import requests
import yaml

from .conftest import resolve_abxpkg_chrome_env, run_python_cwd
from .conftest import (
    cli_env,
    get_free_port,
    run_archivebox_cmd,
    start_archivebox_server as start_daemon_server,
    stop_archivebox_process,
    get_http_response,
    wait_for_log_pattern,
)


PUPPETEER_PROBE_SCRIPT = """\
const fs = require("node:fs");
const puppeteer = require("puppeteer");

async function login(page, config) {
  const result = {
    reachable: false,
    succeeded: false,
    finalUrl: null,
    status: null,
    error: null,
  };

  try {
    const response = await page.goto(config.adminLoginUrl, {
      waitUntil: "networkidle2",
      timeout: 15000,
    });
    result.reachable = true;
    result.status = response ? response.status() : null;

    const usernameInput = await page.$('input[name="username"]');
    const passwordInput = await page.$('input[name="password"]');
    if (!usernameInput || !passwordInput) {
      result.finalUrl = page.url();
      return result;
    }

    await usernameInput.type(config.username);
    await passwordInput.type(config.password);
    await Promise.all([
      page.waitForNavigation({waitUntil: "networkidle2", timeout: 15000}),
      page.click('button[type="submit"], input[type="submit"]'),
    ]);

    result.finalUrl = page.url();
    result.succeeded = !page.url().includes("/admin/login/");
    return result;
  } catch (error) {
    result.error = String(error);
    result.finalUrl = page.url();
    return result;
  }
}

async function main() {
  const config = JSON.parse(fs.readFileSync(0, "utf8"));
  const browser = await puppeteer.launch({
    executablePath: config.chromePath,
    headless: true,
    args: [
      "--no-sandbox",
      "--disable-dev-shm-usage",
      "--disable-background-networking",
    ],
  });

  const loginPage = await browser.newPage();
  const loginResult = await login(loginPage, config);
  await loginPage.close();

  const page = await browser.newPage();
  const consoleMessages = [];
  const requestFailures = [];
  page.on("console", (message) => {
    consoleMessages.push({type: message.type(), text: message.text()});
  });
  page.on("pageerror", (error) => {
    consoleMessages.push({type: "pageerror", text: String(error)});
  });
  page.on("requestfailed", (request) => {
    requestFailures.push({
      url: request.url(),
      error: request.failure() ? request.failure().errorText : "unknown",
    });
  });

  const response = await page.goto(config.dangerousUrl, {
    waitUntil: "networkidle2",
    timeout: 15000,
  });

  await page.waitForFunction(
    () => window.__dangerousScriptRan !== true || window.__probeResults !== undefined,
    {timeout: 15000},
  );

  const pageState = await page.evaluate(() => ({
    href: location.href,
    scriptRan: window.__dangerousScriptRan === true,
    probeResults: window.__probeResults || null,
    bodyText: document.body ? document.body.innerText.slice(0, 600) : "",
  }));

  const output = {
    mode: config.mode,
    login: loginResult,
    dangerousPage: {
      status: response ? response.status() : null,
      finalUrl: page.url(),
      contentSecurityPolicy: response ? response.headers()["content-security-policy"] || null : null,
      archiveboxSecurityMode: response ? response.headers()["x-archivebox-security-mode"] || null : null,
    },
    pageState,
    consoleMessages,
    requestFailures,
  };

  console.log(JSON.stringify(output));
  await browser.close();
}

main().catch((error) => {
  console.error(String(error));
  process.exit(1);
});
"""


PUPPETEER_FIRST_RUN_SETUP_SCRIPT = """\
const fs = require("node:fs");
const puppeteer = require("puppeteer");

async function main() {
  const config = JSON.parse(fs.readFileSync(0, "utf8"));
  const browser = await puppeteer.launch({
    executablePath: config.chromePath,
    headless: true,
    args: [
      "--no-sandbox",
      "--disable-dev-shm-usage",
      `--host-resolver-rules=MAP ${config.hostname} 127.0.0.1`,
    ],
  });
  const page = await browser.newPage();

  await page.goto(config.loginUrl, {waitUntil: "networkidle2", timeout: 15000});
  const firstAdminForm = await page.$("#first-admin-form");
  if (!firstAdminForm) throw new Error("First admin setup form was not shown");
  const firstAdminText = await page.$eval("body", (element) => element.innerText);
  if (firstAdminText.includes("archivebox manage createsuperuser")) {
    throw new Error("First admin setup showed the alternate CLI account flow");
  }
  if (firstAdminText.includes("Ask an ArchiveBox superuser")) {
    throw new Error("First admin setup incorrectly asked for an existing superuser");
  }
  await page.type('input[name="username"]', config.username);
  await page.type('input[name="password1"]', config.password);
  await page.type('input[name="password2"]', config.password);
  await Promise.all([
    page.waitForNavigation({waitUntil: "networkidle2", timeout: 15000}),
    page.click('button[type="submit"], input[type="submit"]'),
  ]);
  await page.waitForSelector("#archivebox-setup-wizard", {timeout: 15000});

  await page.$eval(
    'input[name="archivebox-hosting-location"][value="private"]',
    (input) => input.click(),
  );
  const privateHostingSelected = await page.$eval(
    'input[name="archivebox-hosting-location"][value="private"]',
    (input) => input.checked,
  );
  if (!privateHostingSelected) throw new Error("Private Server option was not selected");
  await page.$eval('input[name="archivebox-dns-mode"][value="single"]', (input) => input.click());
  await page.$eval('input[name="archivebox-tls-mode"][value="none"]', (input) => input.click());
  await page.select("#archivebox-setup-security-mode", "safe-onedomain-nojsreplay");
  await page.waitForFunction(
    () => !document.querySelector("#archivebox-setup-review").disabled,
    {timeout: 30000},
  );
  await Promise.all([
    page.waitForNavigation({waitUntil: "networkidle2", timeout: 15000}),
    page.$eval("#archivebox-setup-review", (button) => button.click()),
  ]);

  await page.waitForFunction(() => {
    const value = document.querySelector('input[name="config"]')?.value || "{}";
    const saved = JSON.parse(value);
    return saved.BASE_URL && saved.SERVER_SECURITY_MODE === "safe-onedomain-nojsreplay";
  }, {timeout: 15000});
  await Promise.all([
    page.waitForNavigation({waitUntil: "networkidle2", timeout: 15000}),
    page.$eval('button[name="_continue"][form="machine_form"]', (button) => button.click()),
  ]);
  await page.reload({waitUntil: "networkidle2", timeout: 15000});
  if (await page.$("#archivebox-setup-wizard")) {
    throw new Error("Setup wizard remained visible after BASE_URL was saved");
  }

  console.log(JSON.stringify({finalUrl: page.url(), bodyText: await page.$eval("body", el => el.innerText.slice(0, 500))}));
  await browser.close();
}

main().catch((error) => {
  console.error(String(error));
  process.exit(1);
});
"""


PUPPETEER_WACZ_PREVIEW_SCRIPT = """\
const fs = require("node:fs");
const puppeteer = require("puppeteer");

function isDescendantOf(frame, ancestor) {
  let parent = frame.parentFrame();
  while (parent) {
    if (parent === ancestor) return true;
    parent = parent.parentFrame();
  }
  return false;
}

async function frameText(frame) {
  try {
    return await frame.evaluate(() => {
      const body = document.body ? document.body.innerText : "";
      const root = document.documentElement ? document.documentElement.innerText : "";
      return body || root || "";
    });
  } catch (_error) {
    return "";
  }
}

async function readPreviewText(page, expectedText) {
  const frames = page.frames();
  const previewFrame = frames.find((frame) => {
    const url = frame.url();
    return url.includes("/archivewebpage/archivewebpage.wacz") && url.includes("preview=1");
  });

  const frameState = [];
  for (const frame of frames) {
    const text = await frameText(frame);
    frameState.push({
      name: frame.name(),
      url: frame.url(),
      isPreview: frame === previewFrame,
      underPreview: previewFrame ? isDescendantOf(frame, previewFrame) : false,
      textSample: text.slice(0, 240),
    });
    if (previewFrame && (frame === previewFrame || isDescendantOf(frame, previewFrame)) && text.includes(expectedText)) {
      return {
        matched: true,
        previewUrl: previewFrame.url(),
        matchedFrameUrl: frame.url(),
        matchedFrameName: frame.name(),
        textSample: text.slice(0, 400),
      };
    }
  }
  return {matched: false, frames: frameState};
}

async function waitForPreviewReady(frame) {
  return await frame.evaluate(() => {
    if (window.archiveboxReplayError) throw new Error(window.archiveboxReplayError);
    if (window.archiveboxReplayReady) return window.archiveboxReplayReady;
    return new Promise((resolve, reject) => {
      window.addEventListener("archivebox-replay-ready", (event) => resolve(event.detail), {once: true});
      window.addEventListener("archivebox-replay-error", (event) => reject(new Error(event.detail)), {once: true});
    });
  });
}

async function main() {
  const config = JSON.parse(fs.readFileSync(0, "utf8"));
  const browser = await puppeteer.launch({
    executablePath: config.chromePath,
    headless: true,
    args: [
      "--no-sandbox",
      "--disable-dev-shm-usage",
      "--disable-background-networking",
    ],
  });

  const page = await browser.newPage();
  const consoleMessages = [];
  const requestFailures = [];
  page.on("console", (message) => {
    consoleMessages.push({type: message.type(), text: message.text()});
  });
  page.on("pageerror", (error) => {
    consoleMessages.push({type: "pageerror", text: String(error)});
  });
  page.on("requestfailed", (request) => {
    requestFailures.push({
      url: request.url(),
      error: request.failure() ? request.failure().errorText : "unknown",
    });
  });

  const response = await page.goto(config.detailUrl, {
    waitUntil: "domcontentloaded",
    timeout: 30000,
  });
  const previewFrame = await page.waitForFrame((frame) => {
    const url = frame.url();
    return url.includes("/archivewebpage/archivewebpage.wacz") && url.includes("preview=1");
  });
  await waitForPreviewReady(previewFrame);
  const previewResult = await readPreviewText(page, config.expectedText);

  console.log(JSON.stringify({
    detailUrl: config.detailUrl,
    status: response ? response.status() : null,
    finalUrl: page.url(),
    previewResult,
    consoleMessages,
    requestFailures,
  }));
  await browser.close();
}

main().catch((error) => {
  console.error(String(error));
  process.exit(1);
});
"""


@pytest.fixture
def browser_runtime(initialized_archive: Path):
    shared_lib = initialized_archive / "lib"
    env = cli_env(
        ABXPKG_LIB_DIR=str(shared_lib),
        CHROME_HEADLESS="True",
        CHROME_SANDBOX="False",
        CHROME_ISOLATION="snapshot",
    )
    env.pop("CHROME_BINARY", None)
    install_result = run_archivebox_cmd(
        ["install", "chrome"],
        cwd=initialized_archive,
        env=env,
        timeout=900,
    )
    assert install_result.returncode == 0, install_result.stderr or install_result.stdout

    resolved_env = resolve_abxpkg_chrome_env(shared_lib, env)

    return {
        "lib_dir": shared_lib,
        "node_modules_dir": Path(resolved_env["NODE_MODULES_DIR"]),
        "node_path": resolved_env["NODE_PATH"],
        "node_binary": Path(resolved_env["NODE_BINARY"]),
        "chrome_binary": Path(resolved_env["CHROME_BINARY"]),
    }


def _seed_archive(data_dir: Path) -> dict[str, object]:
    script = textwrap.dedent(
        """
        import json
        import os
        from pathlib import Path
        from django.utils import timezone

        os.environ.setdefault("DJANGO_SETTINGS_MODULE", "archivebox.core.settings")
        import django
        django.setup()

        from django.contrib.auth import get_user_model
        from archivebox.core.models import Snapshot
        from archivebox.crawls.models import Crawl

        User = get_user_model()
        admin, _ = User.objects.get_or_create(
            username="testadmin",
            defaults={"email": "admin@example.com", "is_staff": True, "is_superuser": True},
        )
        admin.set_password("testpassword")
        admin.save()

        snapshots = {}
        fixture_specs = (
            ("attacker", "https://attacker.example/entry", "Attacker Snapshot", "ATTACKER_SECRET"),
            ("victim", "https://victim.example/private", "Victim Snapshot", "VICTIM_SECRET"),
        )

        for slug, url, title, secret in fixture_specs:
            crawl = Crawl.objects.create(
                urls=url,
                created_by=admin,
                status=Crawl.StatusChoices.SEALED,
                retry_at=timezone.now(),
            )
            snapshot = Snapshot.objects.create(
                url=url,
                title=title,
                crawl=crawl,
                status=Snapshot.StatusChoices.SEALED,
                downloaded_at=timezone.now(),
            )
            output_dir = Path(snapshot.output_dir)
            output_dir.mkdir(parents=True, exist_ok=True)
            (output_dir / "safe.json").write_text(
                json.dumps({"slug": slug, "secret": secret}),
                encoding="utf-8",
            )
            if slug == "attacker":
                (output_dir / "dangerous.html").write_text(
                    '''
                    <!doctype html>
                    <html>
                      <body>
                        <h1>Dangerous Replay Fixture</h1>
                        <script>
                          window.__dangerousScriptRan = true;
                          (async () => {
                            const params = new URLSearchParams(location.search);
                            const targets = {
                              own: params.get("own") || "safe.json",
                              victim: params.get("victim"),
                              admin: params.get("admin"),
                              api: params.get("api"),
                            };
                            const results = {};
                            for (const [label, url] of Object.entries(targets)) {
                              if (!url) continue;
                              try {
                                const response = await fetch(url, {credentials: "include"});
                                const text = await response.text();
                                results[label] = {
                                  ok: true,
                                  status: response.status,
                                  url: response.url,
                                  sample: text.slice(0, 120),
                                };
                              } catch (error) {
                                results[label] = {
                                  ok: false,
                                  error: String(error),
                                };
                              }
                            }
                            window.__probeResults = results;
                            const pre = document.createElement("pre");
                            pre.id = "probe-results";
                            pre.textContent = JSON.stringify(results);
                            document.body.appendChild(pre);
                          })().catch((error) => {
                            window.__probeResults = {fatal: String(error)};
                          });
                        </script>
                      </body>
                    </html>
                    ''',
                    encoding="utf-8",
                )
            snapshots[slug] = {
                "id": str(snapshot.id),
                "domain": snapshot.domain,
            }

        print(json.dumps({
            "username": "testadmin",
            "password": "testpassword",
            "snapshots": snapshots,
        }))
        """,
    )
    stdout, stderr, returncode = run_python_cwd(script, cwd=data_dir, timeout=120)
    assert returncode == 0, stderr
    return json.loads(stdout.strip())


def _build_probe_config(
    mode: str,
    port: int,
    fixture: dict[str, object],
    runtime: dict[str, Path | str],
) -> dict[str, str]:
    snapshots = fixture["snapshots"]
    attacker = snapshots["attacker"]
    victim = snapshots["victim"]
    base_origin = f"http://archivebox.localhost:{port}"
    attacker_id = attacker["id"]
    victim_id = victim["id"]

    if mode == "safe-subdomains-fullreplay":
        attacker_origin = f"http://{attacker_id}.archivebox.localhost:{port}"
        victim_url = f"http://{victim_id}.archivebox.localhost:{port}/safe.json"
        dangerous_base = f"{attacker_origin}/dangerous.html"
        admin_origin = f"http://admin.archivebox.localhost:{port}"
    else:
        attacker_origin = base_origin
        victim_url = f"{base_origin}/snapshot/{victim_id}/safe.json"
        dangerous_base = f"{base_origin}/snapshot/{attacker_id}/dangerous.html"
        admin_origin = base_origin

    query = urlencode(
        {
            "own": "safe.json",
            "victim": victim_url,
            "admin": f"{admin_origin}/admin/",
            "api": f"{admin_origin}/api/v1/docs",
        },
    )

    return {
        "mode": mode,
        "chromePath": str(runtime["chrome_binary"]),
        "adminLoginUrl": f"{admin_origin}/admin/login/",
        "dangerousUrl": f"{dangerous_base}?{query}",
        "username": fixture["username"],
        "password": fixture["password"],
    }


def _run_browser_probe(
    data_dir: Path,
    runtime: dict[str, Path | str],
    mode: str,
    fixture: dict[str, object],
    tmp_path: Path,
) -> dict[str, object]:
    port = get_free_port()
    server_env = os.environ.copy()
    server_env.pop("DATA_DIR", None)
    server_env.update(
        {
            "PYTHONPATH": str(Path(__file__).resolve().parents[2]),
            "BIND_ADDR": f"127.0.0.1:{port}",
            "BASE_URL": f"http://archivebox.localhost:{port}",
            "ALLOWED_HOSTS": "*",
            "SERVER_SECURITY_MODE": mode,
            "USE_COLOR": "False",
            "SHOW_PROGRESS": "False",
            "SAVE_ARCHIVEDOTORG": "False",
            "SAVE_TITLE": "False",
            "SAVE_FAVICON": "False",
            "SAVE_WGET": "False",
            "SAVE_WARC": "False",
            "SAVE_PDF": "False",
            "SAVE_SCREENSHOT": "False",
            "SAVE_DOM": "False",
            "SAVE_SINGLEFILE": "False",
            "SAVE_READABILITY": "False",
            "SAVE_MERCURY": "False",
            "SAVE_GIT": "False",
            "SAVE_YTDLP": "False",
            "SAVE_HEADERS": "False",
            "SAVE_HTMLTOTEXT": "False",
            "USE_CHROME": "False",
        },
    )
    server_log_path = tmp_path / f"{mode}_server.log"
    with server_log_path.open("w", encoding="utf-8") as server_log_file:
        process = run_archivebox_cmd(
            ["server", "--debug", "--nothreading", f"127.0.0.1:{port}"],
            cwd=data_dir,
            env=server_env,
            stdout=server_log_file,
            stderr=subprocess.STDOUT,
            start_new_session=True,
            wait=False,
        )
    try:
        supervisord_log_path = data_dir / "logs" / "supervisord.log"
        runserver_log_path = data_dir / "logs" / "worker_runserver.log"
        wait_for_log_pattern(
            supervisord_log_path,
            r"success: worker_runserver entered RUNNING state,",
            timeout=30,
        )
        deadline = time.monotonic() + 30
        last_exc: Exception | None = None
        while time.monotonic() < deadline:
            try:
                get_http_response(
                    port,
                    f"archivebox.localhost:{port}",
                    timeout=2,
                    process=process,
                )
                break
            except (AssertionError, requests.RequestException) as exc:
                last_exc = exc
                time.sleep(0.2)
        else:
            raise AssertionError(f"server did not become HTTP-ready: {last_exc}")
    except AssertionError as exc:
        stop_archivebox_process(process)
        server_log = server_log_path.read_text(encoding="utf-8", errors="replace")
        supervisord_log = supervisord_log_path.read_text(encoding="utf-8", errors="replace") if supervisord_log_path.exists() else ""
        runserver_log = runserver_log_path.read_text(encoding="utf-8", errors="replace") if runserver_log_path.exists() else ""
        raise AssertionError(
            f"{exc}\n\nSERVER LOG:\n{server_log}\n\nSUPERVISORD LOG:\n{supervisord_log}\n\nRUNSERVER LOG:\n{runserver_log}",
        ) from exc

    probe_path = tmp_path / "server_security_probe.js"
    probe_path.write_text(PUPPETEER_PROBE_SCRIPT, encoding="utf-8")
    probe_config = _build_probe_config(mode, port, fixture, runtime)

    env = os.environ.copy()
    env["NODE_PATH"] = str(runtime["node_path"])
    env["NODE_MODULES_DIR"] = str(runtime["node_modules_dir"])
    env["CHROME_BINARY"] = str(runtime["chrome_binary"])
    env["USE_COLOR"] = "False"

    try:
        result = subprocess.run(
            [str(runtime["node_binary"]), str(probe_path)],
            cwd=data_dir,
            env=env,
            input=json.dumps(probe_config),
            capture_output=True,
            text=True,
            timeout=120,
        )
    finally:
        stop_archivebox_process(process)
        server_log = server_log_path.read_text(encoding="utf-8", errors="replace")

    assert result.returncode == 0, f"{result.stderr}\n\nSERVER LOG:\n{server_log}"
    return json.loads(result.stdout.strip())


def _get_archivewebpage_capture(data_dir: Path, url: str) -> dict[str, str]:
    from archivebox.core.models import ArchiveResult, Snapshot
    from archivebox.tests.test_orm_helpers import use_archivebox_db

    with use_archivebox_db(data_dir):
        snapshot = Snapshot.objects.get(url=url)
        result = ArchiveResult.objects.get(
            snapshot=snapshot,
            plugin="archivewebpage",
            hook_name="on_Snapshot__65_archivewebpage_stop",
        )
        wacz_path = Path(snapshot.output_dir) / "archivewebpage" / "archivewebpage.wacz"
        assert snapshot.status == Snapshot.StatusChoices.SEALED
        assert result.status == ArchiveResult.StatusChoices.SUCCEEDED
        assert wacz_path.is_file()
        return {
            "snapshot_id": str(snapshot.id),
            "wacz_path": str(wacz_path),
        }


@pytest.mark.timeout(180)
def test_unconfigured_public_host_superuser_can_reach_setup_wizard(tmp_path: Path, browser_runtime) -> None:
    port = get_free_port()
    public_hostname = "archivebox.example.test"
    public_host = f"{public_hostname}:{port}"
    env = cli_env(
        port=port,
        disable_extractors=True,
        ALLOWED_HOSTS="*",
        BIND_ADDR=f"127.0.0.1:{port}",
    )
    env.pop("BASE_URL", None)
    env.pop("SERVER_SECURITY_MODE", None)
    init_result = run_archivebox_cmd(["init", "--quick"], cwd=tmp_path, env=env, timeout=60)
    assert init_result.returncode == 0, init_result.stderr or init_result.stdout

    process = start_daemon_server(
        tmp_path,
        port=port,
        env=env,
        daemonize=False,
        log_name="first_run_public_host.log",
    )
    probe_path = tmp_path / "first_run_setup.js"
    probe_path.write_text(PUPPETEER_FIRST_RUN_SETUP_SCRIPT, encoding="utf-8")
    browser_env = os.environ.copy()
    browser_env["NODE_PATH"] = str(browser_runtime["node_path"])

    try:
        for blocked_path in (
            "/api/v1/auth/get_api_token",
            "/admin/machine/machine/00000000-0000-0000-0000-000000000000/change/",
        ):
            blocked_response = requests.post(
                f"http://127.0.0.1:{port}{blocked_path}",
                headers={"Host": public_host},
                timeout=15,
            )
            assert blocked_response.status_code == 403
            assert "control plane disabled" in blocked_response.text.lower()

        result = subprocess.run(
            [str(browser_runtime["node_binary"]), str(probe_path)],
            cwd=tmp_path,
            env=browser_env,
            input=json.dumps(
                {
                    "chromePath": str(browser_runtime["chrome_binary"]),
                    "hostname": public_hostname,
                    "loginUrl": f"http://{public_host}/admin/login/?next=/admin/",
                    "username": "testadmin",
                    "password": "ArchiveBox-test-9vK!",
                },
            ),
            capture_output=True,
            text=True,
            timeout=90,
        )
        assert result.returncode == 0, result.stderr or result.stdout
    finally:
        stop_archivebox_process(process)

    for key, expected in (
        ("BASE_URL", f"http://{public_host}"),
        ("SERVER_SECURITY_MODE", "safe-onedomain-nojsreplay"),
    ):
        config_result = run_archivebox_cmd(["config", "--get", key], cwd=tmp_path, env=env, timeout=60)
        assert config_result.returncode == 0, config_result.stderr or config_result.stdout
        assert expected in config_result.stdout

    compose_text = (Path(__file__).resolve().parents[2] / "docker-compose.yml").read_text()
    compose = yaml.safe_load(compose_text)
    archivebox_environment = compose["services"]["archivebox"]["environment"]
    assert "Open /admin/ on the hostname or IP used to reach ArchiveBox" in compose_text
    assert "manage createsuperuser" not in compose_text
    assert compose["services"]["archivebox"]["restart"] == "unless-stopped"
    assert "BASE_URL" in archivebox_environment
    assert "SERVER_SECURITY_MODE" in archivebox_environment
    assert all("ARCHIVEBOX_INGRESS_BASE_URL" not in value for value in archivebox_environment)
    assert compose["services"]["archivebox"]["ports"] == ["${ARCHIVEBOX_PORT:-8000}:8000"]
    assert set(compose["services"]) == {"archivebox"}


def _run_wacz_preview_probe(
    data_dir: Path,
    runtime: dict[str, Path | str],
    detail_url: str,
    tmp_path: Path,
) -> dict[str, object]:
    probe_path = tmp_path / "wacz_preview_probe.js"
    probe_path.write_text(PUPPETEER_WACZ_PREVIEW_SCRIPT, encoding="utf-8")

    env = os.environ.copy()
    env["NODE_PATH"] = str(runtime["node_path"])
    env["NODE_MODULES_DIR"] = str(runtime["node_modules_dir"])
    env["CHROME_BINARY"] = str(runtime["chrome_binary"])
    env["USE_COLOR"] = "False"

    result = subprocess.run(
        [str(runtime["node_binary"]), str(probe_path)],
        cwd=data_dir,
        env=env,
        input=json.dumps(
            {
                "chromePath": str(runtime["chrome_binary"]),
                "detailUrl": detail_url,
                "expectedText": "This domain is for use in documentation examples",
            },
        ),
        capture_output=True,
        text=True,
        timeout=120,
    )
    assert result.returncode == 0, result.stderr or result.stdout
    return json.loads(result.stdout.strip())


@pytest.mark.parametrize(
    ("mode", "expected"),
    [
        (
            "safe-subdomains-fullreplay",
            {
                "login_succeeds": True,
                "script_ran": True,
                "victim_ok": False,
                "admin_ok": False,
                "admin_status": None,
                "api_ok": False,
                "api_status": None,
                "csp_contains": None,
            },
        ),
        (
            "safe-onedomain-nojsreplay",
            {
                "login_succeeds": True,
                "script_ran": False,
                "victim_ok": None,
                "admin_ok": None,
                "admin_status": None,
                "api_ok": None,
                "api_status": None,
                "csp_contains": "sandbox",
            },
        ),
        (
            "unsafe-onedomain-noadmin",
            {
                "login_succeeds": False,
                "login_status": 403,
                "script_ran": True,
                "victim_ok": True,
                "victim_status": 200,
                "admin_ok": True,
                "admin_status": 403,
                "api_ok": True,
                "api_status": 403,
                "csp_contains": None,
            },
        ),
        (
            "danger-onedomain-fullreplay",
            {
                "login_succeeds": True,
                "script_ran": True,
                "victim_ok": True,
                "victim_status": 200,
                "admin_ok": True,
                "admin_status": 200,
                "api_ok": True,
                "api_status": 200,
                "csp_contains": None,
            },
        ),
    ],
)
def test_server_security_modes_in_chrome(
    initialized_archive: Path,
    browser_runtime,
    tmp_path: Path,
    mode: str,
    expected: dict[str, object],
) -> None:
    fixture = _seed_archive(initialized_archive)
    result = _run_browser_probe(initialized_archive, browser_runtime, mode, fixture, tmp_path)

    login = result["login"]
    dangerous_page = result["dangerousPage"]
    page_state = result["pageState"]
    probe_results = page_state["probeResults"] or {}
    console_texts = [entry["text"] for entry in result["consoleMessages"]]

    assert dangerous_page["status"] == 200
    assert dangerous_page["archiveboxSecurityMode"] == mode
    assert page_state["scriptRan"] is expected["script_ran"]
    assert login["succeeded"] is expected["login_succeeds"]

    login_status = expected.get("login_status")
    if login_status is not None:
        assert login["status"] == login_status

    csp_contains = expected.get("csp_contains")
    if csp_contains:
        csp = dangerous_page["contentSecurityPolicy"] or ""
        assert csp_contains in csp
    else:
        assert dangerous_page["contentSecurityPolicy"] is None

    if mode == "safe-subdomains-fullreplay":
        assert probe_results["own"]["ok"] is True
        assert probe_results["own"]["status"] == 200
        assert "ATTACKER_SECRET" in probe_results["own"]["sample"]
        assert probe_results["victim"]["ok"] is expected["victim_ok"]
        assert probe_results["admin"]["ok"] is expected["admin_ok"]
        assert probe_results["api"]["ok"] is expected["api_ok"]
        assert any("CORS policy" in text for text in console_texts)
    elif mode == "safe-onedomain-nojsreplay":
        assert probe_results == {}
        assert "Dangerous Replay Fixture" in page_state["bodyText"]
        assert any("Blocked script execution" in text for text in console_texts)
    else:
        assert probe_results["own"]["ok"] is True
        assert probe_results["own"]["status"] == 200
        assert "ATTACKER_SECRET" in probe_results["own"]["sample"]
        assert probe_results["victim"]["ok"] is expected["victim_ok"]
        assert probe_results["victim"]["status"] == expected["victim_status"]
        assert "VICTIM_SECRET" in probe_results["victim"]["sample"]
        assert probe_results["admin"]["ok"] is expected["admin_ok"]
        assert probe_results["admin"]["status"] == expected["admin_status"]
        assert probe_results["api"]["ok"] is expected["api_ok"]
        assert probe_results["api"]["status"] == expected["api_status"]

        if mode == "unsafe-onedomain-noadmin":
            assert "control plane disabled" in probe_results["admin"]["sample"].lower()
            assert "control plane disabled" in probe_results["api"]["sample"].lower()
        elif mode == "danger-onedomain-fullreplay":
            assert "ArchiveBox" in probe_results["admin"]["sample"]
            assert "swagger" in probe_results["api"]["sample"].lower()


@pytest.mark.django_db(transaction=True)
@pytest.mark.timeout(600)
def test_archivewebpage_wacz_preview_serves_real_capture_frame(initialized_archive: Path, browser_runtime, tmp_path: Path) -> None:
    from archivebox.core.routes_util import get_snapshot_subdomain

    url = "https://example.com"
    port = get_free_port()
    env = cli_env(
        port=port,
        server=True,
        PLUGINS="archivewebpage",
        BASE_URL=f"http://archivebox.localhost:{port}",
        URL_ALLOWLIST="",
        PUBLIC_INDEX="True",
        PUBLIC_ADD_VIEW="True",
        SERVER_SECURITY_MODE="safe-subdomains-fullreplay",
        USE_CHROME="True",
        CHROME_BINARY=str(browser_runtime["chrome_binary"]),
        CHROME_HEADLESS="True",
        CHROME_SANDBOX="False",
        CHROME_ISOLATION="snapshot",
        ARCHIVEWEBPAGE_ENABLED="True",
        ARCHIVEWEBPAGE_TIMEOUT="90",
        TIMEOUT="90",
    )
    env["ABXPKG_LIB_DIR"] = str(browser_runtime["lib_dir"])
    env["NODE_PATH"] = str(browser_runtime["node_path"])
    env["NODE_MODULES_DIR"] = str(browser_runtime["node_modules_dir"])
    env["CHROMEWEBSTORE_EXTENSIONS_DIR"] = str(
        browser_runtime["lib_dir"] / "chromewebstore" / "extensions",
    )

    process = None
    try:
        install_result = run_archivebox_cmd(
            ["install", "archivewebpage"],
            cwd=initialized_archive,
            env=env,
            timeout=600,
        )
        assert install_result.returncode == 0, (
            f"archivebox install archivewebpage failed:\nSTDOUT:\n{install_result.stdout}\nSTDERR:\n{install_result.stderr}"
        )

        process = start_daemon_server(
            initialized_archive,
            env=env,
            port=port,
            daemonize=False,
            log_name="archivewebpage_server.log",
        )
        get_http_response(port, host=f"archivebox.localhost:{port}", path="/")
        _cmd_result = run_archivebox_cmd(
            ["add", "--depth=0", "--max-urls=1", "--plugins=archivewebpage", url],
            cwd=initialized_archive,
            env=env,
            timeout=120,
        )
        stdout, stderr, returncode = _cmd_result.stdout, _cmd_result.stderr, _cmd_result.returncode
        assert returncode == 0, f"archivebox add failed:\nSTDOUT:\n{stdout}\nSTDERR:\n{stderr}"

        capture = _get_archivewebpage_capture(initialized_archive, url)
        snapshot_host = f"{get_snapshot_subdomain(capture['snapshot_id'])}.archivebox.localhost:{port}"
        detail_url = f"http://{snapshot_host}/#archivewebpage/archivewebpage.wacz"
        result = _run_wacz_preview_probe(initialized_archive, browser_runtime, detail_url, tmp_path)
    finally:
        if process is not None:
            stop_archivebox_process(process)

    assert result["status"] == 200
    assert result["previewResult"]["matched"], json.dumps(result, indent=2)
    assert "/archivewebpage/archivewebpage.wacz" in result["previewResult"]["previewUrl"]
    assert result["previewResult"]["matchedFrameUrl"] != result["finalUrl"]