"""Runs the GpgFrontend screenshot mode in an isolated throwaway environment. Both the render (`--screenshot `) and the environment probe (`--screenshot-env`) are built from the same resolved manifest entry and run with the same subprocess environment, so their env records are comparable. """ from __future__ import annotations import argparse import json import os import shutil import subprocess import sys from dataclasses import dataclass, field from pathlib import Path from typing import Any, Optional from PIL import Image import envhash import hkpmock import fonts import gitutil from common import (DEFAULT_CACHE, DEFAULT_MANIFEST, PipelineError, UsageError, canonical_json, load_json, split_ids) from manifest import Entry, Manifest, load_manifest XVFB_SCREEN = "-screen 0 3200x2000x24" EXIT_REASONS = { 2: "usage-or-precondition", 3: "timeout", 4: "output-unwritable", 5: "unexpected-modal", 6: "environment-gate", } CAPTURE_MODES = ("window", "dialog", "composite") TIMEOUT_MARGIN_S = 60 # Unix sockets are limited to 108 bytes. The demo homedir ends up at # $XDG_DATA_HOME/BKTUS//profiles/.tmp-<16hex>/db and gpg-agent # puts S.gpg-agent.browser inside it. _SOCKET_LIMIT = 107 _SOCKET_TAIL = "/BKTUS/GpgFrontendXXXXXXXXXXXX/profiles/.tmp-0123456789abcdef/db/S.gpg-agent.browser" # Paths that must not leak from the parent environment into a render. _SCRUB_ENV = ( "GNUPGHOME", "QT_SCALE_FACTOR", "QT_SCREEN_SCALE_FACTORS", "QT_AUTO_SCREEN_SCALE_FACTOR", "QT_SCALE_FACTOR_ROUNDING_POLICY", "QT_ENABLE_HIGHDPI_SCALING", "QT_FONT_DPI", "QT_QPA_PLATFORM", "QT_STYLE_OVERRIDE", "QT_QPA_PLATFORMTHEME", ) APPDATA_ROOTS_ENV = "GF_SHOTS_APPDATA_ROOTS" # Where GpgFrontend creates a temporary profile's tree (core/profile). SESSION_STORAGE_ENV = "GF_PROFILE_PACKAGE_STORAGE_DIR" NO_XVFB_ENV = "GF_SHOTS_NO_XVFB" def default_work_root() -> Path: # Fixed and free of the user's name: on Windows the temporary profile # lives under TEMP, which points in here, and the settings page shows # that path. A per-user %TEMP% would put the account name into the manual. if os.name == "nt": drive = os.environ.get("SystemDrive") or "C:" return Path(drive + "\\") / "gf-shot" return Path(f"/tmp/gf-shot-{os.getuid()}") def fresh_session_base() -> Optional[Path]: """Linux: the memory-backed folder the temporary profile is created in, emptied first. The profile has the same name on every run, so the folder it lands in must be one the pipeline owns and empties: a session a crash left in the application's own default folder could not be found again (its pointer lived in the throwaway directory) and would stand in the next run's way. On Windows the session is made under TEMP, which already points into the throwaway directory. """ shm = Path("/dev/shm") if os.name == "nt" or not shm.is_dir(): return None base = shm / f"gf-shot-{os.getuid()}" if base.is_symlink() or (base.exists() and base.stat().st_uid != os.getuid()): raise PipelineError(f"{base} is a symlink or not owned by this user") if base.exists(): shutil.rmtree(base) base.mkdir(mode=0o700) return base def default_xvfb_cmd() -> list[str]: """xvfb-run on Linux. GF_SHOTS_NO_XVFB=1 runs without it (fake binaries in tests); such renders get a non-reference platform tag.""" if os.environ.get(NO_XVFB_ENV) == "1": return [] if sys.platform.startswith("linux"): return ["xvfb-run", "-a", "-s", XVFB_SCREEN] return [] def real_appdata_roots() -> list[Path]: """Where a real (non-ephemeral) GpgFrontend profile would write.""" override = os.environ.get(APPDATA_ROOTS_ENV) if override is not None: return [Path(p) for p in override.split(os.pathsep) if p] if os.name == "nt": return [Path(os.environ[v]) / "BKTUS" for v in ("LOCALAPPDATA", "APPDATA") if os.environ.get(v)] home = Path.home() return [ Path(os.environ.get("XDG_DATA_HOME") or home / ".local" / "share") / "BKTUS", Path(os.environ.get("XDG_CONFIG_HOME") or home / ".config") / "BKTUS", Path(os.environ.get("XDG_STATE_HOME") or home / ".local" / "state") / "BKTUS", Path(os.environ.get("XDG_CACHE_HOME") or home / ".cache") / "BKTUS", home / "Library" / "Application Support" / "BKTUS", ] def footprint_snapshot(roots: list[Path]) -> set[str]: """Every path under /GpgFrontend* for each root.""" seen: set[str] = set() for root in roots: if not root.is_dir(): continue for child in root.iterdir(): if not child.name.startswith("GpgFrontend"): continue seen.add(str(child)) if child.is_dir(): for dirpath, dirnames, filenames in os.walk(child): for name in dirnames + filenames: seen.add(os.path.join(dirpath, name)) return seen @dataclass class EnvInfo: record: dict[str, Any] core_hash: str env_hash: str platform: str font_files: list[dict[str, Any]] def lock_env(self) -> dict[str, Any]: env = envhash.strip_volatile(self.record) env["platform"] = self.platform env["font_files"] = self.font_files return env @dataclass class RunOutcome: ok: bool exit_code: Optional[int] reason: str = "" detail: str = "" env: Optional[EnvInfo] = None png: Optional[Path] = None json_path: Optional[Path] = None geometry: Optional[dict[str, Any]] = None footprint: list[str] = field(default_factory=list) @property def failure(self) -> str: return f"{self.reason}: {self.detail}" if self.detail else self.reason @dataclass class RenderConfig: binary: Path gpgfrontend: Path work_root: Optional[Path] = None keep_work: bool = False xvfb_cmd: Optional[list[str]] = None appdata_roots: Optional[list[Path]] = None font_resolver: fonts.FontResolver = fonts.resolve_font_file def resolved_xvfb(self) -> list[str]: return default_xvfb_cmd() if self.xvfb_cmd is None else list(self.xvfb_cmd) class Renderer: def __init__(self, cfg: RenderConfig) -> None: self.cfg = cfg self.binary = Path(cfg.binary) if not self.binary.is_file(): raise UsageError(f"binary not found: {self.binary}") self.work_root = Path(cfg.work_root) if cfg.work_root else default_work_root() self.xvfb = cfg.resolved_xvfb() if self.xvfb and not shutil.which(self.xvfb[0]): raise UsageError(f"{self.xvfb[0]} not found; it is needed for Linux previews") self._font_cache: dict[str, dict] = {} self._probe_cache: dict[tuple, RunOutcome] = {} self._key_server: Optional[hkpmock.MockKeyServer] = None def key_server(self) -> Optional[hkpmock.MockKeyServer]: """The local mock key server, started on first use; None without demo keys.""" directory = self.fixtures().parent / "keyserver" if self._key_server is None and (directory / "listing.colons").is_file(): self._key_server = hkpmock.MockKeyServer(directory) return self._key_server def close(self) -> None: if self._key_server is not None: self._key_server.close() self._key_server = None # Paths ----------------------------------------------------------------- def demo_gnupghome(self) -> Path: return Path(self.cfg.gpgfrontend) / "resource" / "screenshot" / "demo-gnupghome" def fixtures(self) -> Path: return Path(self.cfg.gpgfrontend) / "resource" / "screenshot" / "fixtures" def _prepare_root(self) -> None: root = self.work_root root.mkdir(parents=True, exist_ok=True, mode=0o700) if os.name != "nt": st = os.lstat(root) if os.path.islink(root) or st.st_uid != os.getuid(): raise PipelineError(f"work root {root} is a symlink or not owned by this user") if len(str(root)) + 4 + len(_SOCKET_TAIL) > _SOCKET_LIMIT: raise UsageError( f"work root {root} is too long: the gpg-agent socket path would exceed " f"{_SOCKET_LIMIT} bytes. Use a shorter --work-root.") def _new_workdir(self) -> Path: """One fixed directory, emptied first: the application shows paths under it, so its name must be the same on every run. Renders run one at a time; two pipelines sharing a work root would collide.""" self._prepare_root() candidate = self.work_root / "run" if candidate.is_symlink(): raise PipelineError(f"{candidate} is a symlink; refusing to use it") if candidate.exists(): shutil.rmtree(candidate) candidate.mkdir(mode=0o700) return candidate def subprocess_env(self, work: Path) -> dict[str, str]: env = {k: v for k, v in os.environ.items() if k not in _SCRUB_ENV} env["TZ"] = "UTC" if os.name == "nt": layout = {"LOCALAPPDATA": "la", "APPDATA": "ra", "TEMP": "t", "TMP": "t", "USERPROFILE": "alice"} else: layout = {"HOME": "alice", "XDG_DATA_HOME": "d", "XDG_CONFIG_HOME": "c", "XDG_STATE_HOME": "s", "XDG_CACHE_HOME": "k", "TMPDIR": "t"} for var, sub in layout.items(): path = work / sub path.mkdir(exist_ok=True) env[var] = str(path) session = fresh_session_base() if session is not None: env[SESSION_STORAGE_ENV] = str(session) key_server = self.key_server() if key_server is not None: env[hkpmock.PROXY_ENV] = key_server.address return env # Command lines ----------------------------------------------------------- def command(self, entry: Entry, out: Path, probe: bool) -> list[str]: mode = ["--screenshot-env"] if probe else ["--screenshot", entry.scenario or ""] cmd = [ str(self.binary), "--temp-profile", *mode, "--locale", entry.locale, "--theme", entry.theme, "--font-family", ",".join(entry.fonts), *(["--size", entry.size_str] if entry.size is not None else []), "--clock", entry.clock, "--demo-gnupghome", str(self.demo_gnupghome()), "--fixtures", str(self.fixtures()), "--out", str(out), "--timeout", str(entry.timeout_ms), ] return self.xvfb + cmd def list_scenarios(self) -> dict[str, dict[str, str]]: work = self._new_workdir() try: proc = subprocess.run( self.xvfb + [str(self.binary), "--screenshot-list"], env=self.subprocess_env(work), capture_output=True, timeout=120, check=False, ) finally: self._cleanup(work) if proc.returncode != 0: raise PipelineError(f"--screenshot-list exited {proc.returncode}: " f"{proc.stderr.decode('utf-8', 'replace').strip()[-400:]}") scenarios = {} for line in proc.stdout.decode("utf-8", "replace").splitlines(): parts = line.strip().split("\t") if len(parts) == 3: scenarios[parts[0]] = {"size": parts[1], "capture_mode": parts[2]} return scenarios # Running ----------------------------------------------------------------- def _cleanup(self, work: Path) -> None: if not self.cfg.keep_work: shutil.rmtree(work, ignore_errors=True) def _run(self, entry: Entry, probe: bool, dest_dir: Optional[Path], stem: str) -> RunOutcome: work = self._new_workdir() out_dir = work / "o" out_dir.mkdir() out = out_dir / (f"{stem}.json" if probe else f"{stem}.png") json_path = out.with_suffix(".json") roots = self.cfg.appdata_roots if self.cfg.appdata_roots is not None else real_appdata_roots() before = footprint_snapshot(roots) try: try: proc = subprocess.run( self.command(entry, out, probe), env=self.subprocess_env(work), capture_output=True, timeout=entry.timeout_ms / 1000 + TIMEOUT_MARGIN_S, check=False, ) code: Optional[int] = proc.returncode stderr = proc.stderr.decode("utf-8", "replace").strip() except subprocess.TimeoutExpired: code, stderr = None, "the binary did not exit; killed by the runner" except OSError as exc: return RunOutcome(False, None, "launch-failed", str(exc)) new = sorted(footprint_snapshot(roots) - before) outcome = self._evaluate(entry, probe, code, stderr, out, json_path) outcome.footprint = new if new and outcome.ok: outcome.ok = False outcome.reason = "footprint" outcome.detail = "new files outside the throwaway directory: " + ", ".join(new[:10]) if dest_dir is not None: dest_dir.mkdir(parents=True, exist_ok=True) if not probe and out.exists(): outcome.png = dest_dir / out.name shutil.copyfile(out, outcome.png) if json_path.exists(): outcome.json_path = dest_dir / json_path.name shutil.copyfile(json_path, outcome.json_path) return outcome finally: self._cleanup(work) def _evaluate(self, entry: Entry, probe: bool, code: Optional[int], stderr: str, out: Path, json_path: Path) -> RunOutcome: record: Optional[dict[str, Any]] = None if json_path.exists(): try: record = json.loads(json_path.read_bytes().decode("utf-8")) except (ValueError, UnicodeDecodeError) as exc: return RunOutcome(False, code, "invalid-json", str(exc)) if code is None: return RunOutcome(False, None, "timeout", stderr[-400:]) if code != 0: reason = EXIT_REASONS.get(code, f"exit-{code}") detail = "" if isinstance(record, dict): err = record.get("error") or {} if isinstance(err, dict) and err.get("reason"): detail = str(err.get("reason")) if err.get("detail"): detail += f" ({err['detail']})" env_part = record if probe else record.get("env") if isinstance(env_part, dict) and env_part.get("gate_failures"): detail += " gates: " + ",".join(map(str, env_part["gate_failures"])) return RunOutcome(False, code, reason, (detail or stderr[-400:]).strip()) if record is None: return RunOutcome(False, code, "no-json", f"{json_path.name} was not written") env_record = record if probe else record.get("env") if not isinstance(env_record, dict): return RunOutcome(False, code, "no-env", "the record carries no env") try: env = self.env_info(env_record) except PipelineError as exc: return RunOutcome(False, code, "env-invalid", str(exc)) if env_record.get("gate_failures"): return RunOutcome(False, code, "environment-gate", ",".join(map(str, env_record["gate_failures"])), env) if probe: return RunOutcome(True, code, env=env, geometry=None) problem = check_geometry(record, out, entry.size) if problem: return RunOutcome(False, code, "geometry", problem, env) return RunOutcome(True, code, env=env, geometry=record) def env_info(self, record: dict[str, Any]) -> EnvInfo: if record.get("schema") != envhash.ENV_SCHEMA: raise PipelineError(f"unsupported env schema {record.get('schema')!r}") core = envhash.env_hash_core(record) if record.get("env_hash_core") != core: raise PipelineError( f"env_hash_core cross-check failed: binary says {record.get('env_hash_core')}, " f"recomputed {core}") files = fonts.font_files(envhash.font_file_list(record), self.cfg.font_resolver, self._font_cache) platform = envhash.platform_tag(record, under_xvfb=bool(self.xvfb)) return EnvInfo(record, core, envhash.final_env_hash(core, platform, files), platform, files) def probe(self, entry: Entry) -> RunOutcome: key = entry.probe_key() if key not in self._probe_cache: self._probe_cache[key] = self._run(entry, probe=True, dest_dir=None, stem="env") return self._probe_cache[key] def render(self, entry: Entry, dest_dir: Path) -> RunOutcome: return self._run(entry, probe=False, dest_dir=dest_dir, stem=entry.id) def check_geometry(geometry: dict[str, Any], png: Path, size: Optional[tuple[int, int]] = None) -> str: """Whether a render record is well formed and matches its PNG. With @p size (logical, as the entry asks for it), the captured window must also be exactly that size, so a window resized after the binary's own check is not taken for a good render. """ if geometry.get("schema") != 1: return f"unsupported geometry schema {geometry.get('schema')!r}" if geometry.get("error"): return f"record carries an error: {geometry['error']}" if geometry.get("capture_mode") not in CAPTURE_MODES: return f"unknown capture_mode {geometry.get('capture_mode')!r}" if geometry.get("dpr") != 2: return f"dpr-mismatch: expected 2, got {geometry.get('dpr')!r}" if not png.exists(): return f"{png.name} was not written" with Image.open(png) as img: actual = list(img.size) if list(geometry.get("image_size") or []) != actual: return f"image_size {geometry.get('image_size')} does not match the PNG {actual}" if not isinstance(geometry.get("widgets"), dict) or not isinstance(geometry.get("missing", []), list): return "widgets/missing malformed" if size is not None: expected = [size[0] * 2, size[1] * 2] window = next((layer for layer in geometry.get("layers") or [] if isinstance(layer, dict) and layer.get("name") == "window"), None) rect = window.get("rect") if window else None if not isinstance(rect, list) or len(rect) != 4: return "no window layer to check the size against" if list(rect[2:]) != expected: return f"size-mismatch: window captured at {rect[2]}x{rect[3]}, expected {expected[0]}x{expected[1]}" return "" def comparable_json(path: Path) -> bytes: """A render JSON without the fields that legitimately differ per run.""" data = load_json(path) env = data.get("env") if isinstance(data, dict) else None if isinstance(env, dict): data["env"] = envhash.strip_volatile(env) return canonical_json(data) def freshness_warnings(binary: Path, gpgfrontend: Path, source_paths: list[str]) -> list[str]: """A heuristic, not proof that the binary matches the source.""" warnings: list[str] = [] binary = Path(binary) for candidate in (binary.parent / "BUILD-INFO.json", binary.parent.parent / "BUILD-INFO.json"): if candidate.exists(): info = load_json(candidate, {}) or {} head = gitutil.head(gpgfrontend) if info.get("commit") and info.get("commit") != head: warnings.append(f"{candidate}: commit {info.get('commit')} differs from GpgFrontend HEAD {head}") return warnings try: bin_mtime = binary.stat().st_mtime except OSError: return warnings newest: tuple[float, str] = (0.0, "") for rel in source_paths: path = Path(gpgfrontend) / rel files = [path] if path.is_file() else (p for p in path.rglob("*") if p.is_file()) if path.is_dir() else [] for f in files: m = f.stat().st_mtime if m > newest[0]: newest = (m, str(f)) if newest[0] > bin_mtime: warnings.append(f"binary {binary} is older than {newest[1]}; it may not contain the latest source") return warnings def main(argv: Optional[list[str]] = None) -> int: parser = argparse.ArgumentParser(description="Render or probe manifest entries (debug helper).") parser.add_argument("--gpgfrontend", required=True, type=Path) parser.add_argument("--binary", type=Path, default=os.environ.get("GF_SCREENSHOT_BINARY")) parser.add_argument("--manifest", type=Path, default=DEFAULT_MANIFEST) parser.add_argument("--only", required=True, help="comma separated entry ids") parser.add_argument("--probe", action="store_true", help="run --screenshot-env instead") parser.add_argument("--out-dir", type=Path, default=DEFAULT_CACHE / "render") parser.add_argument("--work-root", type=Path) parser.add_argument("--keep-work", action="store_true") args = parser.parse_args(argv) try: if not args.binary: raise UsageError("--binary or GF_SCREENSHOT_BINARY is required") manifest: Manifest = load_manifest(args.manifest) by_id = manifest.by_id() renderer = Renderer(RenderConfig(args.binary, args.gpgfrontend, args.work_root, args.keep_work)) rc = 0 for entry_id in split_ids(args.only): if entry_id not in by_id or not by_id[entry_id].renderable: raise UsageError(f"unknown or non-renderable id: {entry_id}") entry = by_id[entry_id] outcome = renderer.probe(entry) if args.probe else renderer.render(entry, args.out_dir) if outcome.ok and outcome.env: print(f"{entry_id}: ok platform={outcome.env.platform} env_hash={outcome.env.env_hash}" + (f" png={outcome.png}" if outcome.png else "")) else: rc = 1 print(f"{entry_id}: failed:{outcome.failure}") return rc except UsageError as exc: print(f"error: {exc}", file=sys.stderr) return 2 except PipelineError as exc: print(f"error: {exc}", file=sys.stderr) return 1 if __name__ == "__main__": sys.exit(main())