External/watermarks-remover
1
0
Fork 0
mirror of https://github.com/guillaumemeyer/watermarks-remover.git synced 2026-08-22 13:11:57 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #6 from CarlosMaeda/fix/harden-cleaners

Browse Source 
fix: harden cleaners against argv injection and resource exhaustion
This commit is contained in:
Guillaume Meyer (The Opinionated Man)
2026-08-12 20:49:43 -07:00
committed by GitHub
parent 4b4c942635 a3253d3675
commit a443019bc8
6 changed files with 142 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
skills/remove-ai-marks/scripts/clean_file.py
+7
View File
@@ -5,6 +5,7 @@ from __future__ import annotations
import argparse
import json
import os
import sys
from pathlib import Path
@@ -15,6 +16,8 @@ from container_meta import clean_container, detect_container_format # noqa: E40
from image_meta import clean_image, detect_format as detect_image_format # noqa: E402
from text_unicode import clean_text # noqa: E402
MAX_INPUT_BYTES = int(os.environ.get("WATERMARKS_MAX_INPUT_BYTES", str(1 << 30)))
IMAGE_EXTS = {".png", ".jpg", ".jpeg"}
CONTAINER_EXTS = {".svg", ".pdf", ".docx", ".odt", ".html", ".htm", ".md", ".markdown", ".mdx"}
TEXT_EXTS = {
@@ -74,6 +77,10 @@ def main() -> int:
eprint(f"not a file: {args.path}")
return 2
if args.path.stat().st_size > MAX_INPUT_BYTES:
eprint(f"refusing input larger than {MAX_INPUT_BYTES} bytes: {args.path}")
return 2
kind = args.force_type if args.force_type != "auto" else classify(args.path)
if args.in_place:
skills/remove-ai-marks/scripts/common.py
+11
View File
@@ -43,3 +43,14 @@ def which(cmd: str) -> str | None:
from shutil import which as _which
return _which(cmd)
def safe_arg(path: str) -> str:
"""Guard paths passed to option-parsing CLIs (exiftool, c2patool).
A filename starting with '-' would otherwise be interpreted as an option
(e.g. exiftool's -@argfile), turning a crafted filename into argv injection.
"""
if path.startswith("-"):
return "./" + path
return path
skills/remove-ai-marks/scripts/container_meta.py
+29 -8
View File
@@ -13,7 +13,7 @@ import zipfile
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
from common import which
from common import safe_arg, which
from image_meta import AI_META_HINTS, C2PA_MARKERS, run_optional_tools
# Frontmatter / meta keys that often carry AI provenance
@@ -385,23 +385,38 @@ def _zip_namelist(data: bytes) -> list[str]:
return zf.namelist()
MAX_ZIP_DECOMPRESSED_BYTES = 512 * 1024 * 1024
def _check_zip_budget(info: zipfile.ZipInfo, budget: list[int]) -> None:
"""Reject zip bombs before decompression (ZipInfo.file_size is stored)."""
budget[0] += info.file_size
if budget[0] > MAX_ZIP_DECOMPRESSED_BYTES:
raise ValueError(
"zip decompressed size exceeds cap "
f"({MAX_ZIP_DECOMPRESSED_BYTES} bytes); refusing to process"
)
def inspect_docx(data: bytes) -> tuple[bool, bool, list[str], dict]:
findings: list[str] = []
has_c2pa = False
has_ai = False
parts: list[str] = []
budget = [0]
try:
with zipfile.ZipFile(io.BytesIO(data)) as zf:
parts = zf.namelist()
for name in parts:
raw = zf.read(name)
for info in zf.infolist():
_check_zip_budget(info, budget)
raw = zf.read(info.filename)
c2, ai, hits = _blob_hits(raw)
if c2 or ai:
if c2:
has_c2pa = True
if ai:
has_ai = True
findings.append(f"{name}: {', '.join(hits[:6])}")
findings.append(f"{info.filename}: {', '.join(hits[:6])}")
# always flag customXml presence lightly
custom = [n for n in parts if n.startswith("customXml/")]
if custom:
@@ -414,11 +429,13 @@ def inspect_docx(data: bytes) -> tuple[bool, bool, list[str], dict]:
def clean_docx(data: bytes) -> tuple[bytes, list[str]]:
actions: list[str] = []
out_buf = io.BytesIO()
budget = [0]
with zipfile.ZipFile(io.BytesIO(data)) as zin, zipfile.ZipFile(
out_buf, "w", compression=zipfile.ZIP_DEFLATED
) as zout:
for info in zin.infolist():
name = info.filename
_check_zip_budget(info, budget)
raw = zin.read(name)
# Drop entire customXml trees (often provenance injects)
if name.startswith("customXml/"):
@@ -495,17 +512,19 @@ def inspect_odt(data: bytes) -> tuple[bool, bool, list[str], dict]:
findings: list[str] = []
has_c2pa = False
has_ai = False
budget = [0]
try:
with zipfile.ZipFile(io.BytesIO(data)) as zf:
for name in zf.namelist():
raw = zf.read(name)
for info in zf.infolist():
_check_zip_budget(info, budget)
raw = zf.read(info.filename)
c2, ai, hits = _blob_hits(raw)
if c2 or ai:
if c2:
has_c2pa = True
if ai:
has_ai = True
findings.append(f"{name}: {', '.join(hits[:6])}")
findings.append(f"{info.filename}: {', '.join(hits[:6])}")
if "meta.xml" in zf.namelist():
meta = zf.read("meta.xml").decode("utf-8", errors="replace")
if re.search(r"generator|claude|openai|anthropic|gemini", meta, re.I):
@@ -519,11 +538,13 @@ def inspect_odt(data: bytes) -> tuple[bool, bool, list[str], dict]:
def clean_odt(data: bytes) -> tuple[bytes, list[str]]:
actions: list[str] = []
out_buf = io.BytesIO()
budget = [0]
with zipfile.ZipFile(io.BytesIO(data)) as zin, zipfile.ZipFile(
out_buf, "w", compression=zipfile.ZIP_DEFLATED
) as zout:
for info in zin.infolist():
name = info.filename
_check_zip_budget(info, budget)
raw = zin.read(name)
if name == "meta.xml":
text = raw.decode("utf-8", errors="replace")
@@ -607,7 +628,7 @@ def clean_pdf(path: Path, dest: Path) -> tuple[list[str], dict]:
exiftool,
"-all=",
"-overwrite_original",
str(dest),
safe_arg(str(dest)),
],
capture_output=True,
text=True,
skills/remove-ai-marks/scripts/image_meta.py
+4 -4
View File
@@ -13,7 +13,7 @@ from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
from common import which
from common import safe_arg, which
SCRIPTS_DIR = Path(__file__).resolve().parent
@@ -197,7 +197,7 @@ def run_optional_tools(path: Path) -> dict[str, Any]:
if c2patool:
try:
r = subprocess.run(
[c2patool, str(path)],
[c2patool, safe_arg(str(path))],
capture_output=True,
text=True,
timeout=30,
@@ -227,7 +227,7 @@ def run_optional_tools(path: Path) -> dict[str, Any]:
if exiftool:
try:
r = subprocess.run(
[exiftool, "-G1", "-a", "-s", str(path)],
[exiftool, "-G1", "-a", "-s", safe_arg(str(path))],
capture_output=True,
text=True,
timeout=30,
@@ -490,7 +490,7 @@ def clean_image(
exiftool,
"-all=",
"-overwrite_original",
str(dest),
safe_arg(str(dest)),
],
capture_output=True,
text=True,
skills/remove-ai-marks/scripts/inspect_file.py
+7
View File
@@ -4,6 +4,7 @@
from __future__ import annotations
import argparse
import os
import sys
from pathlib import Path
@@ -15,6 +16,8 @@ from image_meta import detect_format as detect_image_format # noqa: E402
from image_meta import inspect_image # noqa: E402
from text_unicode import human_report, inspect_text # noqa: E402
MAX_INPUT_BYTES = int(os.environ.get("WATERMARKS_MAX_INPUT_BYTES", str(1 << 30)))
TEXT_EXTS = {".txt", ".text", ".md", ".markdown", ".mdx", ".html", ".htm", ".css", ".js", ".py", ".rs", ".go", ".json", ".yaml", ".yml", ".toml", ".csv"}
IMAGE_EXTS = {".png", ".jpg", ".jpeg"}
CONTAINER_EXTS = {".svg", ".pdf", ".docx", ".odt", ".html", ".htm", ".md", ".markdown", ".mdx"}
@@ -55,6 +58,10 @@ def main() -> int:
eprint(f"not a file: {args.path}")
return 2
if args.path.stat().st_size > MAX_INPUT_BYTES:
eprint(f"refusing input larger than {MAX_INPUT_BYTES} bytes: {args.path}")
return 2
kind = args.force_type if args.force_type != "auto" else classify(args.path)
if kind == "text":
tests/test_security_hardening.py
+84
View File
@@ -0,0 +1,84 @@
"""Tests for the cleaners security hardening (safe argv, resource caps)."""
from __future__ import annotations
import io
import sys
import zipfile
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
SCRIPTS = ROOT / "skills" / "remove-ai-marks" / "scripts"
sys.path.insert(0, str(SCRIPTS))
from common import safe_arg # noqa: E402
from container_meta import ( # noqa: E402
MAX_ZIP_DECOMPRESSED_BYTES,
_check_zip_budget,
inspect_docx,
)
def test_safe_arg_prefixes_leading_dash():
assert safe_arg("-@evil") == "./-@evil"
assert safe_arg("--argfile") == "./--argfile"
def test_safe_arg_leaves_normal_paths_alone():
assert safe_arg("photo.png") == "photo.png"
assert safe_arg("dir/file.svg") == "dir/file.svg"
assert safe_arg("/abs/path.pdf") == "/abs/path.pdf"
assert safe_arg(".") == "."
def test_zip_budget_rejects_oversized_member():
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w") as zf:
zf.writestr("word/document.xml", "<w:document/>")
with zipfile.ZipFile(io.BytesIO(buf.getvalue())) as zf:
info = zf.infolist()[0]
info.file_size = MAX_ZIP_DECOMPRESSED_BYTES + 1
raised = False
try:
_check_zip_budget(info, [0])
except ValueError:
raised = True
assert raised
def test_zip_budget_accumulates_across_members():
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w") as zf:
zf.writestr("a.xml", b"a")
zf.writestr("b.xml", b"b")
with zipfile.ZipFile(io.BytesIO(buf.getvalue())) as zf:
infos = zf.infolist()
for info in infos:
info.file_size = MAX_ZIP_DECOMPRESSED_BYTES // 2 + 1024
budget = [0]
raised = False
for info in infos:
try:
_check_zip_budget(info, budget)
except ValueError:
raised = True
break
assert raised
def test_inspect_docx_with_ai_markers_does_not_crash():
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w") as zf:
zf.writestr("word/document.xml", "<w:document/>")
zf.writestr(
"docProps/app.xml",
"<Properties><Application>Claude AI Writer</Application></Properties>",
)
zf.writestr(
"docProps/core.xml",
"<cp:coreProperties><dc:creator>Anthropic</dc:creator></cp:coreProperties>",
)
has_c2pa, has_ai, findings, _ = inspect_docx(buf.getvalue())
assert has_ai
assert findings
assert not has_c2pa or has_ai