#!/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(
'''
Dangerous Replay Fixture
''',
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"]