diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 1bc7292..5d9124c 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -17,7 +17,7 @@ jobs: - uses: actions/setup-python@v5 with: python-version: ${{ matrix.python-version }} - - run: pip install . + - run: pip install ".[examples]" - run: python -m unittest discover -s tests -v typecheck: @@ -27,5 +27,5 @@ jobs: - uses: actions/setup-python@v5 with: python-version: "3.12" - - run: pip install mypy + - run: pip install mypy ".[examples]" - run: mypy diff --git a/.gitignore b/.gitignore index c50d910..8465220 100644 --- a/.gitignore +++ b/.gitignore @@ -13,3 +13,5 @@ htmlcov/ *.png *.jpeg .DS_Store +reports/ +.env.demo diff --git a/README.md b/README.md index 5bfa339..9977fcf 100644 --- a/README.md +++ b/README.md @@ -33,6 +33,12 @@ two spellings as the same project). Building an AI agent with an MCP-capable client instead? Use the sibling MCP server: `npx -y site-shot-mcp`. Working in Node? `npm install site-shot-sdk`. +## CSV → local report example + +For a validated CSV, bounded capture workers, saved PNGs, per-row outcomes and an +escaped local HTML report, see [the runnable CSV report example](examples/README.md). +It uses the optional `examples` extra; the SDK itself stays zero-dependency. + ## Capture cleanly (ads and cookie banners removed) ```python diff --git a/examples/README.md b/examples/README.md new file mode 100644 index 0000000..18a9b87 --- /dev/null +++ b/examples/README.md @@ -0,0 +1,167 @@ +# CSV screenshots → local HTML report + +A runnable, public-page example using the official Python SDK. Each CSV row makes +**one client capture call**. The run saves PNG files, a per-row JSON manifest and +an HTML index for human review. It does not retry failed rows, reuse an old image, +reduce quality on failure, publish files or export PDF. The service's own rendering +and routing policy still applies inside each API request. + +**API prerequisite:** an account, confirmed email, active paid API plan and API +key. See [Quickstart](https://www.site-shot.com/start/) and +[pricing](https://www.site-shot.com/pricing/). The separate free browser tool is +not a free API trial. Captures consume your plan's allowance; a timeout can happen +after server work has started. Confirm the budget before running a CSV. + +## Install and run + +From the cloned SDK repository, with Python 3.9 or newer: + +```bash +python3 -m venv .venv +source .venv/bin/activate +python -m pip install -e '.[examples]' +``` + +The optional `examples` extra adds Pillow for full PNG decoding/validation. The +SDK itself remains standard-library-only. Keep the example and SDK checkout at +the same revision when following this guide. + +Set `SITESHOT_API_KEY` through your secret manager or a masked terminal prompt. +Leading and trailing whitespace is removed, matching the SDK; an empty key is +rejected before creating a report or submitting captures. +For Bash, this avoids putting the key in shell history or the process arguments: + +```bash +read -r -s -p 'Site-Shot API key: ' SITESHOT_API_KEY +printf '\n' +export SITESHOT_API_KEY +python examples/csv_report.py examples/report_urls.csv --output reports +unset SITESHOT_API_KEY +``` + +The supplied CSV contains **five real public Site-Shot pages**, not customer data. +It is a proposed demo input, not evidence of a completed live capture. Inspect or +replace it before spending API quota. No live run is performed by installation, +tests, imports or `--help`. + +### CSV contract + +```csv +id,url +home,https://www.site-shot.com/ +pricing,https://www.site-shot.com/pricing/ +``` + +- UTF-8 (optional BOM), exactly `id,url` as the header, one to 1,000 data rows. +- Stable IDs: 1–64 ASCII letters, digits, `_` or `-`; first character must be a + letter or digit. IDs must be unique ignoring case, so files remain safe on + case-insensitive filesystems. Reserved device names such as `CON` are rejected. + IDs, not URL-derived names, name the PNGs. +- Absolute public HTTP(S) URLs on standard web ports. This intentionally narrow + example rejects credentials, query strings, fragments, whitespace, local + hostnames and non-public IP literals. It does not resolve DNS during validation; + hostname syntax checks are not a proof that a hostname is publicly reachable. + Ambiguous shortened/octal/hex numeric host spellings are rejected as well. +- Use only pages you are entitled to capture. Private/authenticated pages and + token-bearing URLs are out of scope. No auth cookies, custom headers or injected + JavaScript are accepted by this CLI. +- The entire CSV is validated **before** any capture. Invalid input exits `2`; + nothing is silently skipped or trimmed. + +### Fixed capture budget + +Every row uses the same settings, recorded once in `manifest.json`: + +| Setting | Value | +|---|---| +| Viewport | 1280 × 800 | +| Full page / maximum height | `full_size=True`, `max_height=4000` | +| Format | PNG | +| Cleanup | `no_ads=True`, `no_cookie_popup=True` | +| Wait before capture | `delay_time=1500` milliseconds | +| Server render deadline | `timeout=60000` milliseconds | +| Whole client exchange deadline | 90 seconds | +| Client retries | `0` | + +These are **example settings, not optimal values for every site**. Pages taller +than 4,000 pixels can be clipped. The client deadline includes response download; +this example has no separate total-batch deadline. A maximum-size input can take +a long time and spend significant quota. Cleanup is not guaranteed on every site. + +Default concurrency is **one worker**. Increase `--workers` only after checking +your account's concurrent-request allowance and budget; the CLI accepts 1–8, +which is an example safety bound, **not a plan entitlement**. For example, +`--workers 2` is appropriate only if your plan permits it. Each worker constructs +its own SDK client. The script never discovers a higher allowance or increases +concurrency automatically. + +## Read the result + +A run creates a new UTC-stamped, randomly suffixed folder, for example: + +```text +reports/run--/ + index.html + manifest.json + images/ + home.png + pricing.png +``` + +Open that run's `index.html` locally in your browser. It has no scripts, remote +fonts, tracking or external image loads. The source-page links are explicit +outbound links. Do not publish the directory automatically: the manifest and +report contain the supplied URLs and the screenshots themselves. + +The manifest records each row's ID, source URL, UTC start/end, duration, status, +image path/dimensions/hash or a safe error code. Settings, client deadline, +retries and worker count are recorded once for the whole run. CSV order is +preserved even when workers finish out of order. A second run cannot overwrite +the first. + +PNG bytes must pass signature, container integrity and full pixel decoding before +an image gets its final filename. This example rejects images over 40 MiB, +32 million pixels, or multiple frames. JSON, HTML, JPEG or corrupt/truncated +responses are errors, not `.png` screenshots. A failed local image write may +leave a `.tmp` file, never a report-linked PNG. A report-storage failure or an +interrupted run leaves an incomplete directory; do not treat its existence as a +completed report. + +**Decoding is not a content-quality check.** Before sharing, open every image and +confirm the intended page, readable content, geometry and acceptable clipping; +reject blank, challenge, login or target-error pages. Even an exit code of `0` +means only that image capture/decoding and local report creation completed; +visual review is still required. This is not visual regression testing, a +functional test, semantic comparison, an accessibility audit or legal evidence. + +### Exit statuses and failures + +- `0`: every row produced a decoded PNG; human review still required. +- `1`: at least one row failed; the HTML explicitly says **Incomplete capture + batch** and keeps failed rows visible. Other rows are still attempted once. +- `2`: invalid input/missing key or a local report-storage problem. The CLI states + whether requests were not sent or the run could not be completed. +- `130`: interrupted with Ctrl+C. Not-yet-started captures are cancelled, including + if interruption happens while the batch is being submitted. Already in-flight + requests are allowed to finish and can still consume quota. The script waits + for them (up to their remaining client deadlines) before exiting; that partial + directory is not a completed report. + +Errors expose only a fixed code (such as `capture_timeout`, +`quota_or_payment_required`, `authentication_failed`, `invalid_png` or +`local_file_error`) and, when available, an HTTP status. Raw exception messages, +bodies, URLs built by the SDK and API keys are not logged or copied into the +report. Fix the reported issue deliberately before deciding whether to start a +**new** billable run; there is no automatic retry/resume path. + +## Test without API calls + +```bash +python -m unittest discover -s tests -v +``` + +The example tests use generated local images and fake clients. They cover CSV and +URL validation, secret-safe failures, incomplete reports, decoding, unique run +folders, escaped HTML and bounded concurrency. They do not prove live API +parameters, target-page quality or paid-plan concurrency. Those remain separate, +explicitly approved demo checks before publishing a tutorial or recording video. diff --git a/examples/__init__.py b/examples/__init__.py new file mode 100644 index 0000000..f26ede1 --- /dev/null +++ b/examples/__init__.py @@ -0,0 +1 @@ +"""Runnable examples; not part of the installed SDK package.""" diff --git a/examples/csv_report.py b/examples/csv_report.py new file mode 100644 index 0000000..2b0f4e6 --- /dev/null +++ b/examples/csv_report.py @@ -0,0 +1,369 @@ +"""Capture a public-URL CSV once per row; produce a local, review-required report. + +Instructions and limits: examples/README.md. No client retries or live test mode. +""" + +from __future__ import annotations + +import argparse +import csv +import hashlib +import ipaddress +import json +import os +import re +import secrets +import sys +import time +import warnings +from collections.abc import Sequence +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass +from datetime import datetime, timezone +from html import escape +from io import BytesIO +from pathlib import Path +from typing import Any, Callable, Protocol +from urllib.parse import urlsplit + +from PIL import Image + +from site_shot import ( + APIError, + AuthError, + CountryUnavailableError, + InvalidParamsError, + QuotaError, + SiteShot, + SiteShotError, + SiteShotTimeoutError, +) + +CAPTURE_SETTINGS: dict[str, Any] = { + "width": 1280, + "height": 800, + "full_size": True, + "max_height": 4000, + "format": "png", + "no_ads": True, + "no_cookie_popup": True, + "delay_time": 1500, + "timeout": 60000, +} +CLIENT_TIMEOUT_SECONDS = 90 +MAX_ROWS = 1000 +MAX_WORKERS = 8 +MAX_IMAGE_BYTES = 40 * 1024 * 1024 +MAX_IMAGE_PIXELS = 32_000_000 +ID_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9_-]{0,63}\Z") +RESERVED_IDS = {"con", "prn", "aux", "nul"} | { + f"{prefix}{number}" for prefix in ("com", "lpt") for number in range(1, 10) +} + + +class InputError(ValueError): + """An input failed validation before any API requests were submitted.""" + + +class InvalidImage(ValueError): + """Response bytes are not a complete, bounded PNG image.""" + + +class CaptureClient(Protocol): + def capture(self, url: str, /, **params: Any) -> bytes: ... + + +@dataclass(frozen=True) +class Row: + id: str + url: str + + +@dataclass(frozen=True) +class BatchResult: + directory: Path + captured: int + failed: int + + @property + def exit_code(self) -> int: + return 1 if self.failed else 0 + + +def utc_now() -> str: + return datetime.now(timezone.utc).isoformat(timespec="milliseconds") + + +def validate_url(value: str) -> None: + # This deliberately small public-page example never accepts auth/share tokens. + if not value or any(c.isspace() or ord(c) < 32 or ord(c) == 127 for c in value): + raise InputError("URL must not contain whitespace or control characters") + if "\\" in value: + raise InputError("URL must not contain backslashes") + try: + url = urlsplit(value) + host = url.hostname + if url.scheme not in ("http", "https") or not host: + raise InputError("Use an absolute HTTP(S) URL") + if url.username is not None or url.password is not None or url.query or url.fragment: + raise InputError("Use a public URL without credentials, query strings or fragments") + if url.port not in (None, 80, 443): + raise InputError("Only standard public web ports are supported") + # Classify the same ASCII hostname a browser resolves, including Unicode + # digits and dot separators, rather than accepting their pre-IDNA form. + host = host.encode("idna").decode("ascii").lower().rstrip(".") + try: + address = ipaddress.ip_address(host) + except ValueError: + # Browsers accept shortened, octal and hex IPv4 spellings that + # ipaddress intentionally rejects. Never treat those as DNS names. + if re.fullmatch( + r"(?:0[xX][0-9A-Fa-f]+|[0-9]+)(?:\.(?:0[xX][0-9A-Fa-f]+|[0-9]+))*", host + ): + raise InputError("Use an unambiguous public hostname or IP address") + if ( + "." not in host + or host.endswith((".local", ".localhost", ".internal")) + or len(host) > 253 + or any( + not re.fullmatch(r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?", label) + for label in host.split(".") + ) + ): + raise InputError("Use a public hostname, not a local/private address") + else: + if not address.is_global: + raise InputError("Use a public hostname, not a local/private address") + except (UnicodeError, ValueError) as exc: + if isinstance(exc, InputError): + raise + raise InputError("Malformed HTTP(S) URL") from None + + +def validate_rows(rows: Sequence[Row]) -> None: + if not rows or len(rows) > MAX_ROWS: + raise InputError(f"Supply between 1 and {MAX_ROWS} rows") + seen: set[str] = set() + for number, row in enumerate(rows, 2): + if ( + not ID_PATTERN.fullmatch(row.id) + or row.id.casefold() in seen + or row.id.casefold() in RESERVED_IDS + ): + raise InputError(f"CSV row {number} needs a unique, filename-safe ID") + try: + validate_url(row.url) + except InputError as exc: + raise InputError(f"CSV row {number}: {exc}") from None + seen.add(row.id.casefold()) + + +def read_rows(path: Path) -> list[Row]: + rows: list[Row] = [] + try: + with path.open(encoding="utf-8-sig", newline="") as stream: + reader = csv.reader(stream, strict=True) + if next(reader, None) != ["id", "url"]: + raise InputError("CSV must start with exactly the header id,url") + for number, values in enumerate(reader, 2): + if len(values) != 2: + raise InputError(f"CSV row {number} must have exactly two fields") + rows.append(Row(*values)) + if len(rows) > MAX_ROWS: + raise InputError(f"This example accepts at most {MAX_ROWS} rows") + except (OSError, UnicodeError, csv.Error): + raise InputError("CSV could not be read as valid UTF-8 CSV") from None + validate_rows(rows) + return rows + + +def validate_png(data: bytes) -> tuple[int, int]: + if not isinstance(data, bytes) or not data.startswith(b"\x89PNG\r\n\x1a\n"): + raise InvalidImage("Response is not PNG") + if len(data) > MAX_IMAGE_BYTES: + raise InvalidImage("Image exceeds this example's size limit") + try: + with warnings.catch_warnings(): + warnings.simplefilter("error", Image.DecompressionBombWarning) + with Image.open(BytesIO(data)) as image: + if image.format != "PNG" or getattr(image, "n_frames", 1) != 1: + raise InvalidImage("Expected one PNG frame") + width, height = image.size + if width < 1 or height < 1 or width * height > MAX_IMAGE_PIXELS: + raise InvalidImage("Image exceeds this example's pixel limit") + image.verify() + # verify checks the container; load also requires complete pixel data. + with Image.open(BytesIO(data)) as image: + image.load() + return width, height + except InvalidImage: + raise + except Exception: # noqa: BLE001 -- decoder failures must not expose untrusted response data + raise InvalidImage("PNG could not be completely decoded") from None + + +ERROR_CODES = ( + (AuthError, "authentication_failed"), + (QuotaError, "quota_or_payment_required"), + (CountryUnavailableError, "country_unavailable"), + (InvalidParamsError, "invalid_capture_parameters"), + (SiteShotTimeoutError, "capture_timeout"), + (APIError, "api_error"), + (SiteShotError, "capture_error"), + (InvalidImage, "invalid_png"), + (OSError, "local_file_error"), +) + + +def capture_row(row: Row, directory: Path, factory: Callable[[], CaptureClient]) -> dict[str, Any]: + start = time.monotonic() + result: dict[str, Any] = {"id": row.id, "url": row.url, "started_at": utc_now()} + try: + data = factory().capture(row.url, **CAPTURE_SETTINGS) + width, height = validate_png(data) + filename = f"images/{row.id}.png" + # A failed local write can leave only a .tmp, never a report-linked PNG. + temporary = directory / f"images/{row.id}.png.tmp" + with temporary.open("xb") as stream: + stream.write(data) + temporary.replace(directory / filename) + result.update( + status="captured", + image=filename, + width=width, + height=height, + sha256=hashlib.sha256(data).hexdigest(), + visual_review="required", + ) + except Exception as exc: # noqa: BLE001 -- per-row failure accounting with secret-safe codes + code = next((code for cls, code in ERROR_CODES if isinstance(exc, cls)), "unexpected_error") + result.update(status="failed", error=code) + # Never serialize exception text/body/cause: upstream errors may echo secrets. + if isinstance(exc, SiteShotError) and isinstance(exc.http_status, int): + result["http_status"] = exc.http_status + result.update(finished_at=utc_now(), duration_ms=round((time.monotonic() - start) * 1000)) + return result + + +def write_html(directory: Path, manifest: dict[str, Any]) -> None: + cards = [] + for row in manifest["rows"]: + label, url = escape(row["id"]), escape(row["url"], quote=True) + if row["status"] == "captured": + filename = escape(row["image"], quote=True) + content = ( + f"

Captured; visual review required ({row['width']} × {row['height']}).

" + f'Open full-size PNG' + ) + else: + content = f'

Failed: {escape(row["error"])}

' + cards.append( + f'' + ) + title = "Incomplete capture batch" if manifest["failed"] else "Capture batch: review required" + html = f""" + + +{title} + +

{title}

{manifest["captured"]} captured; {manifest["failed"]} failed.

+

Image decoding is not a content check. Review every image for the intended page, +readable content, clipping, blanks and challenge/error pages before sharing this report.

+

Run started: {escape(manifest["started_at"])}. Times are UTC. Manifest.

+
Settings used for every row
{escape(json.dumps(manifest["capture_settings"], indent=2))}
+{"".join(cards)}""" + (directory / "index.html").write_text(html, encoding="utf-8") + + +def run_batch( + rows: Sequence[Row], output: Path, factory: Callable[[], CaptureClient], workers: int = 1 +) -> BatchResult: + if not 1 <= workers <= MAX_WORKERS: + raise InputError(f"workers must be between 1 and {MAX_WORKERS}") + validate_rows(rows) + output.mkdir(parents=True, exist_ok=True) + stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S.%fZ") + directory = output / f"run-{stamp}-{secrets.token_hex(4)}" + directory.mkdir() + (directory / "images").mkdir() + started = utc_now() + executor = ThreadPoolExecutor(max_workers=workers) + try: + results = list(executor.map(lambda row: capture_row(row, directory, factory), rows)) + finally: + # map can be interrupted while still submitting, before its iterator's + # own cancellation cleanup exists. Never start queued billable work then. + executor.shutdown(wait=True, cancel_futures=True) + failed = sum(row["status"] == "failed" for row in results) + manifest: dict[str, Any] = { + "schema_version": 1, + "started_at": started, + "finished_at": utc_now(), + "status": "incomplete" if failed else "captures_completed_review_required", + "capture_settings": dict(CAPTURE_SETTINGS), + "client_timeout_seconds": CLIENT_TIMEOUT_SECONDS, + "client_retries": 0, + "workers": workers, + "captured": len(results) - failed, + "failed": failed, + "rows": results, + } + (directory / "manifest.json").write_text(json.dumps(manifest, indent=2), encoding="utf-8") + write_html(directory, manifest) + return BatchResult(directory, len(results) - failed, failed) + + +def main(argv: Sequence[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description="Capture a public-URL CSV into a local HTML report." + ) + parser.add_argument("csv", type=Path, help="UTF-8 CSV with exactly id,url columns") + parser.add_argument( + "--output", type=Path, default=Path("reports"), help="Parent of a unique run folder" + ) + parser.add_argument( + "--workers", + type=int, + choices=range(1, MAX_WORKERS + 1), + default=1, + help="Default 1; increase only after confirming your plan's concurrency allowance", + ) + args = parser.parse_args(argv) + key = os.environ.get("SITESHOT_API_KEY", "").strip() + if not key: + print("Set SITESHOT_API_KEY in your environment; no requests were sent.", file=sys.stderr) + return 2 + try: + rows = read_rows(args.csv) + if key in str(args.output) or any(key in row.url or key in row.id for row in rows): + raise InputError("The API key must not occur in input data or the output path") + result = run_batch( + rows, + args.output, + lambda: SiteShot(key, timeout=CLIENT_TIMEOUT_SECONDS, retries=0), + args.workers, + ) + except InputError as exc: + print(f"Input error: {exc}. No requests were sent.", file=sys.stderr) + return 2 + except OSError: + print("Local report storage failed; this run is not complete.", file=sys.stderr) + return 2 + except KeyboardInterrupt: + print( + "Interrupted; queued captures cancelled. In-flight captures may have consumed quota. " + "This run is not a completed report.", + file=sys.stderr, + ) + return 130 + print( + f"Captured: {result.captured}; failed: {result.failed}. Report: {result.directory / 'index.html'}" + ) + print("Review every captured image before treating it as a usable result.") + return result.exit_code + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/examples/report_urls.csv b/examples/report_urls.csv new file mode 100644 index 0000000..55b1590 --- /dev/null +++ b/examples/report_urls.csv @@ -0,0 +1,6 @@ +id,url +home,https://www.site-shot.com/ +quickstart,https://www.site-shot.com/start/ +pricing,https://www.site-shot.com/pricing/ +blog,https://www.site-shot.com/blog/ +scheduled,https://www.site-shot.com/scheduled-screenshots/ diff --git a/pyproject.toml b/pyproject.toml index 417c8be..337a039 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -42,6 +42,9 @@ classifiers = [ ] dependencies = [] +[project.optional-dependencies] +examples = ["Pillow"] + [project.urls] Homepage = "https://www.site-shot.com/" Documentation = "https://www.site-shot.com/#documentation" @@ -58,6 +61,7 @@ include = [ "/src", "/tests", "/scripts", + "/examples", "/README.md", "/LICENSE", ] @@ -70,7 +74,7 @@ mypy_path = "src" # TypedDict broke, and checking only src/ is why that shipped. Bodies are # checked everywhere; only the legacy test module is excused from annotating # every def. -files = ["src/site_shot", "tests"] +files = ["src/site_shot", "tests", "examples/csv_report.py"] [[tool.mypy.overrides]] module = ["test_client", "tests.test_client"] diff --git a/tests/test_csv_report.py b/tests/test_csv_report.py new file mode 100644 index 0000000..339eb05 --- /dev/null +++ b/tests/test_csv_report.py @@ -0,0 +1,338 @@ +"""No live API calls: the runnable report is tested through fake clients.""" + +from __future__ import annotations + +import json +import os +import tempfile +import threading +import time +import unittest +from concurrent.futures import Future, ThreadPoolExecutor +from contextlib import redirect_stderr, redirect_stdout +from io import BytesIO, StringIO +from pathlib import Path +from typing import Any +from unittest.mock import Mock, patch + +from PIL import Image + +from examples import csv_report as report +from site_shot import APIError, AuthError, SiteShotTimeoutError + + +def png_bytes() -> bytes: + stream = BytesIO() + Image.new("RGB", (12, 8), "white").save(stream, format="PNG") + return stream.getvalue() + + +class FakeClient: + def __init__(self, response: Any = None) -> None: + self.response = png_bytes() if response is None else response + self.calls: list[Any] = [] + + def capture(self, url: str, /, **params: Any) -> bytes: + self.calls.append((url, params)) + if isinstance(self.response, Exception): + raise self.response + return bytes(self.response) + + +class CSVReportTests(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name) + + def csv(self, text: str) -> Path: + path = self.root / "input.csv" + path.write_text(text, encoding="utf-8") + return path + + def test_public_fixture_has_five_real_owned_pages(self) -> None: + rows = report.read_rows(Path(__file__).parents[1] / "examples/report_urls.csv") + self.assertEqual(len(rows), 5) + self.assertTrue(all(r.url.startswith("https://www.site-shot.com/") for r in rows)) + + def test_bad_csv_rejected(self) -> None: + for text in ( + "url,id\nhttps://example.com,a", + "id,url\n", + "id,url\na,https://example.com\nA,https://example.com/b", + "id,url\n../a,https://example.com", + "id,url\nCON,https://example.com", + "id,url\na,https://example.com,extra", + 'id,url\na,"broken', + ): + with self.subTest(text=text), self.assertRaises(report.InputError): + report.read_rows(self.csv(text)) + + def test_bom_and_stable_ids(self) -> None: + rows = report.read_rows( + self.csv("\ufeffid,url\na-1,https://example.com/\nb_2,https://example.org/\n") + ) + self.assertEqual([r.id for r in rows], ["a-1", "b_2"]) + + def test_untrusted_url_inputs(self) -> None: + for url in ( + "javascript:alert(1)", + "https://user:secret@example.com/", + "https://example.com/?token=secret", + "https://example.com/#secret", + "http://127.0.0.1/", + "http://127.1/", + "http://0177.0.0.1/", + "http://0x7f.0.0.1/", + "http://127.0.0.1./", + "http://192.168.1/", + "http://2130706433/", + "http://１２７.０.０.１/", + "http://127。0。0。1/", + "http://０１７７.０.０.１/", + "http://０ｘ７ｆ.０.０.１/", + "http://127.0.0.1。/", + "http://[::1]/", + "http://localhost/", + "https://test.local/", + "https://example.com:8080/", + "https://example.com/\nx", + "https://example.com\\@other.org/", + "https://bad_host.com/", + ): + with self.subTest(url=url), self.assertRaises(report.InputError): + report.validate_url(url) + + def test_public_idns_and_global_ip_literals_remain_allowed(self) -> None: + for url in ( + "https://bücher.de/", + "https://münchen.de/", + "https://8.8.8.8/", + "https://[2001:4860:4860::8888]/", + ): + with self.subTest(url=url): + report.validate_url(url) + + def test_binary_validation_requires_decoded_png(self) -> None: + self.assertEqual(report.validate_png(png_bytes()), (12, 8)) + jpg = BytesIO() + Image.new("RGB", (12, 8)).save(jpg, format="JPEG") + for data in ( + b'{"error":"secret"}', + b"Error", + jpg.getvalue(), + b"\x89PNG\r\n\x1a\n{}", + png_bytes()[:40], + ): + with self.subTest(data=data[:8]), self.assertRaises(report.InvalidImage): + report.validate_png(data) + with patch.object(report, "MAX_IMAGE_PIXELS", 2), self.assertRaises(report.InvalidImage): + report.validate_png(png_bytes()) + + def test_unique_run_folders_and_settings(self) -> None: + client = FakeClient() + rows = [report.Row("a", "https://example.com/")] + first = report.run_batch(rows, self.root, lambda: client) + second = report.run_batch(rows, self.root, lambda: client) + self.assertNotEqual(first.directory, second.directory) + self.assertEqual(first.exit_code, 0) + self.assertEqual((first.directory / "images/a.png").read_bytes(), png_bytes()) + manifest = json.loads((first.directory / "manifest.json").read_text()) + self.assertEqual(manifest["client_retries"], 0) + self.assertEqual(manifest["workers"], 1) + self.assertEqual(manifest["capture_settings"], report.CAPTURE_SETTINGS) + self.assertEqual(manifest["rows"][0]["visual_review"], "required") + self.assertTrue(all(params == report.CAPTURE_SETTINGS for _, params in client.calls)) + + def test_failure_is_visible_without_secret_or_retry(self) -> None: + secret = "secret-key-do-not-log" + for error in ( + AuthError(secret, http_status=401, body=secret), + SiteShotTimeoutError(secret), + APIError(secret), + ValueError(secret), + b'{"error":"' + secret.encode() + b'"}', + ): + client = FakeClient(error) + factory = Mock(return_value=client) + run = report.run_batch([report.Row("a", "https://example.com/")], self.root, factory) + self.assertEqual(run.exit_code, 1) + self.assertEqual(len(client.calls), 1) + manifest = (run.directory / "manifest.json").read_text() + html = (run.directory / "index.html").read_text() + self.assertNotIn(secret, manifest + html) + self.assertIn("Incomplete capture batch", html) + self.assertFalse((run.directory / "images/a.png").exists()) + + def test_html_escapes_url_path_and_shows_failed_row(self) -> None: + rows = [ + report.Row("a", 'https://example.com/