diff --git a/README.md b/README.md index 4d69556..e5238f7 100644 --- a/README.md +++ b/README.md @@ -69,6 +69,13 @@ window). Optional: `zbar` (QR decoding) and `tesseract` (OCR search) — without them the picker still works and shows the `omarchy pkg add …` command to add them. No sudo or pkexec is required by the plugin itself. +The clipboard owner is treated as untrusted. Payloads are streamed through a +capped reader and dropped when they exceed 32 MiB (images) or 4 MiB (text), or +take longer than 5 s to deliver; QR decoding and OCR only run on images under +40 megapixels, as read from the file header without decoding. Override with +`CLIPBOARD_MAX_IMAGE_BYTES`, `CLIPBOARD_MAX_TEXT_BYTES`, and +`CLIPBOARD_MAX_PARSE_PIXELS` in the shell's environment. + Omarchy's default `Super+Ctrl+V` routes to it via the clone mechanism. An existing `Super+Shift+V` binding targeting `omarchy.clipboard` routes to it too. For a custom binding, edit the **live** config — diff --git a/capture.py b/capture.py index 55d111f..50c587e 100755 --- a/capture.py +++ b/capture.py @@ -11,12 +11,19 @@ the current clipboard. Emits exactly one JSON line per capture: Sensitive clips (x-kde-passwordManagerHint / CLIPBOARD_STATE=sensitive) and binary payloads are skipped silently. + +The clipboard owner is untrusted: every payload is read through a bounded +reader (byte cap + deadline) and QR/OCR only run on images whose header +dimensions are under a pixel cap, so a hostile or runaway source cannot make +the plugin buffer an unbounded payload or decode a decompression bomb. """ import hashlib import json import os +import select import shutil +import struct import subprocess import sys import tempfile @@ -34,6 +41,24 @@ IMAGE_EXT = {"image/png": "png", "image/jpeg": "jpg", "image/webp": "webp", TEXT_TYPES = ["text/plain;charset=utf-8", "text/plain", "UTF8_STRING", "STRING", "TEXT", "COMPOUND_TEXT"] +def _env_int(name, default): + try: + return max(0, int(os.environ.get(name, "") or default)) + except ValueError: + return default + + +# Payload limits. Anything larger is dropped (not truncated) so the history +# never holds a partial clip. Overridable per environment. +MAX_IMAGE_BYTES = _env_int("CLIPBOARD_MAX_IMAGE_BYTES", 32 * 1024 * 1024) +MAX_TEXT_BYTES = _env_int("CLIPBOARD_MAX_TEXT_BYTES", 4 * 1024 * 1024) +# QR decoding and OCR decode the full bitmap; skip them above this many +# pixels (the image is still recorded and previewed by Qt, which has its own +# allocation limits). +MAX_PARSE_PIXELS = _env_int("CLIPBOARD_MAX_PARSE_PIXELS", 40 * 1000 * 1000) +READ_TIMEOUT = 5.0 + + def run(args, timeout=5): try: r = subprocess.run(args, capture_output=True, timeout=timeout) @@ -42,6 +67,115 @@ def run(args, timeout=5): return None +def read_bounded(args, limit, timeout=READ_TIMEOUT): + """Run `args` and return its stdout, or None if it exits non-zero, writes + more than `limit` bytes, or does not finish within `timeout` seconds. + Output is streamed and the process is killed as soon as the cap is hit, + so memory use is bounded by `limit` regardless of what the source sends.""" + try: + proc = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) + except Exception: + return None + chunks = [] + total = 0 + ok = True + deadline = time.monotonic() + timeout + try: + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + ok = False + break + ready, _, _ = select.select([proc.stdout], [], [], remaining) + if not ready: + ok = False + break + chunk = os.read(proc.stdout.fileno(), 65536) + if not chunk: + break + total += len(chunk) + if total > limit: + ok = False + break + chunks.append(chunk) + except Exception: + ok = False + finally: + if not ok: + proc.kill() + try: + proc.wait(timeout=2) + except Exception: + proc.kill() + proc.stdout.close() + if not ok or proc.returncode != 0: + return None + return b"".join(chunks) + + +def read_clipboard(mime, limit): + return read_bounded(["wl-paste", "--type", mime, "--no-newline"], limit) + + +def image_size(data, mime): + """(width, height) from the container header alone — no pixel decoding. + Returns None when the header is not understood.""" + try: + if mime == "image/png" and data[:8] == b"\x89PNG\r\n\x1a\n" and data[12:16] == b"IHDR": + return struct.unpack(">II", data[16:24]) + if mime == "image/gif" and data[:6] in (b"GIF87a", b"GIF89a"): + return struct.unpack("> 14) & 0x3FFF) + 1 + if chunk == b"VP8 ": + w, h = struct.unpack("H", data[i + 2:i + 4])[0] + if marker in (0xC0, 0xC1, 0xC2, 0xC3, 0xC5, 0xC6, 0xC7, 0xC9, 0xCA, 0xCB, 0xCD, 0xCE, 0xCF): + h, w = struct.unpack(">HH", data[i + 5:i + 9]) + return w, h + i += 2 + length + if mime == "image/tiff" and data[:4] in (b"II*\x00", b"MM\x00*"): + end = "<" if data[:2] == b"II" else ">" + off = struct.unpack(end + "I", data[4:8])[0] + count = struct.unpack(end + "H", data[off:off + 2])[0] + w = h = None + for k in range(min(count, 64)): + e = off + 2 + k * 12 + tag, typ = struct.unpack(end + "HH", data[e:e + 4]) + val = struct.unpack(end + ("H" if typ == 3 else "I"), data[e + 8:e + 8 + (2 if typ == 3 else 4)])[0] + if tag == 256: + w = val + elif tag == 257: + h = val + if w and h: + return w, h + except Exception: + return None + return None + + def list_types(): out = run(["wl-paste", "--list-types"]) if not out: @@ -86,22 +220,22 @@ def capture_image(types, app): mime = next((m for m in IMAGE_MIMES if m in types), None) if not mime: return - data = run(["wl-paste", "--type", mime, "--no-newline"]) + data = read_clipboard(mime, MAX_IMAGE_BYTES) if not data: - return + # Oversized, slow, or empty: skip the clip entirely (an image mime was + # offered, so we do not fall through to a text capture of it either). + return True os.makedirs(IMAGE_DIR, exist_ok=True) digest = hashlib.sha256(data).hexdigest() path = os.path.join(IMAGE_DIR, f"{digest}.{IMAGE_EXT[mime]}") entry = {"type": "image", "mime": mime, "path": path, "bytes": len(data), "app": app} - # Dimensions via Pillow when available; harmless without it. - try: - from PIL import Image # noqa: PLC0415 - import io # noqa: PLC0415 - with Image.open(io.BytesIO(data)) as im: - entry["w"], entry["h"] = im.size - except Exception: - pass + # Dimensions from the header only; the bitmap is never decoded here. + size = image_size(data, mime) + if size: + entry["w"], entry["h"] = int(size[0]), int(size[1]) + # Unknown or huge dimensions → no QR/OCR pass (both decode the full image). + parse_ok = bool(size) and size[0] * size[1] <= MAX_PARSE_PIXELS if not os.path.exists(path): fd, tmp = tempfile.mkstemp(dir=IMAGE_DIR) @@ -116,11 +250,11 @@ def capture_image(types, app): pass return - qr = decode_qr(path) if os.environ.get("CLIPBOARD_QR", "1") != "0" else None + qr = decode_qr(path) if parse_ok and os.environ.get("CLIPBOARD_QR", "1") != "0" else None if qr: entry["qr"] = qr - ocr = decode_ocr(path, os.environ.get("CLIPBOARD_OCR_LANG", "eng")) + ocr = decode_ocr(path, os.environ.get("CLIPBOARD_OCR_LANG", "eng")) if parse_ok else None if ocr: entry["ocr"] = ocr @@ -175,7 +309,7 @@ def decode_ocr(path, lang): def capture_uri_list(types, app): if "text/uri-list" not in types: return False - data = run(["wl-paste", "--type", "text/uri-list", "--no-newline"]) + data = read_clipboard("text/uri-list", MAX_TEXT_BYTES) if not data: return False text = decode_text(data) @@ -200,7 +334,7 @@ def capture_text(types, app): mime = next((m for m in TEXT_TYPES if m in types), None) if not mime: return - data = run(["wl-paste", "--type", mime, "--no-newline"]) + data = read_clipboard(mime, MAX_TEXT_BYTES) if not data: return text = decode_text(data) diff --git a/scripts/backfill-ocr.py b/scripts/backfill-ocr.py index 2ec2902..30c3007 100755 --- a/scripts/backfill-ocr.py +++ b/scripts/backfill-ocr.py @@ -11,7 +11,9 @@ import os import shutil import subprocess import sys -import time + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +import capture # noqa: E402 (image_size / MAX_PARSE_PIXELS shared with the daemon) STATE = os.path.join( os.environ.get("XDG_STATE_HOME", os.path.expanduser("~/.local/state")), @@ -56,6 +58,14 @@ def main(): continue if not os.path.exists(entry["path"]): continue + # Same guard as the capture daemon: never decode an image whose header + # dimensions are unknown or above the pixel cap. + with open(entry["path"], "rb") as f: + head = f.read(65536) + size = capture.image_size(head, entry.get("mime") or "image/png") + if not size or size[0] * size[1] > capture.MAX_PARSE_PIXELS: + print(f" - {entry['id']}: skipped (size {size or 'unknown'})") + continue text = ocr(entry["path"], args.lang) if text: entry["ocr"] = text