refactor(obfuscate_release): enhance asset processing and validation logic for JS and CSS files

scripts/obfuscate_release.py

@@ -3,15 +3,17 @@ from __future__ import annotations
 
 import argparse
 import hashlib
-import os
 import re
 import shutil
 import subprocess
 import sys
+import tempfile
+from collections import defaultdict
 from pathlib import Path
 
 
 TEXT_EXTENSIONS = {".php", ".html", ".htm", ".xml", ".txt", ".js", ".css"}
+HASH_SUFFIX_RE = re.compile(r"\.[a-f0-9]{12}$", re.I)
 
 
 def strip_comments_keep_strings(text: str) -> str:
@@ -174,42 +176,93 @@ def minify_js_fallback(text: str) -> str:
     return text.strip()
 
 
-def run_cmd(command: list[str], cwd: Path, input_text: str | None = None) -> str:
+def canonical_asset_base(stem: str) -> str:
+    name = stem
+    while HASH_SUFFIX_RE.search(name):
+        name = HASH_SUFFIX_RE.sub("", name)
+    return name
+
+
+def is_skipped_asset(path: Path) -> bool:
+    lowered = path.as_posix().lower()
+    if ".min." in path.name or ".obf." in path.name or ".deob." in path.name:
+        return True
+    if "deobfuscated" in lowered:
+        return True
+    return False
+
+
+def is_valid_source_content(content: str) -> bool:
+    if "[javascript-obfuscator-cli]" in content:
+        return False
+    return len(content.strip()) >= 20
+
+
+def collect_asset_groups(asset_root: Path) -> dict[tuple[Path, str, str], list[Path]]:
+    groups: dict[tuple[Path, str, str], list[Path]] = defaultdict(list)
+    for ext in (".js", ".css"):
+        for file_path in sorted(asset_root.rglob(f"*{ext}")):
+            if is_skipped_asset(file_path):
+                continue
+            base = canonical_asset_base(file_path.stem)
+            key = (file_path.parent, base, ext)
+            groups[key].append(file_path)
+    return groups
+
+
+def pick_source_file(paths: list[Path], base: str, ext: str) -> Path:
+    plain = paths[0].parent / f"{base}{ext}"
+    if plain in paths:
+        return plain
+    return min(paths, key=lambda p: len(p.name))
+
+
+def run_cmd(command: list[str], cwd: Path) -> None:
     proc = subprocess.run(
         command,
         cwd=str(cwd),
-        input=input_text,
         text=True,
         capture_output=True,
         check=False,
     )
     if proc.returncode != 0:
-        raise RuntimeError(proc.stderr.strip() or "command failed")
-    return proc.stdout
+        raise RuntimeError(proc.stderr.strip() or proc.stdout.strip() or "command failed")
 
 
 def process_js(path: Path, cwd: Path) -> None:
     original = path.read_text(encoding="utf-8")
+    if not is_valid_source_content(original):
+        raise ValueError(f"invalid or corrupted JS source: {path}")
+
     if shutil.which("npx"):
+        tmpdir = Path(tempfile.mkdtemp())
         try:
-            minified = run_cmd(
+            src = tmpdir / "input.js"
+            out = tmpdir / "output.js"
+            src.write_text(original, encoding="utf-8")
+            run_cmd(
                 [
                     "npx",
                     "--yes",
                     "terser",
+                    str(src),
+                    "-o",
+                    str(src),
                     "--compress",
                     "--mangle",
                     "--comments",
                     "false",
                 ],
                 cwd,
-                input_text=original,
             )
-            obfuscated = run_cmd(
+            run_cmd(
                 [
                     "npx",
                     "--yes",
                     "javascript-obfuscator",
+                    str(src),
+                    "--output",
+                    str(out),
                     "--compact",
                     "true",
                     "--control-flow-flattening",
@@ -224,38 +277,51 @@ def process_js(path: Path, cwd: Path) -> None:
                     "browser-no-eval",
                     "--source-map",
                     "false",
-                    "--output",
-                    "stdout",
                 ],
                 cwd,
-                input_text=minified,
             )
-            path.write_text(obfuscated.strip() + "\n", encoding="utf-8")
+            if not out.exists():
+                raise RuntimeError("obfuscator produced no output file")
+            result = out.read_text(encoding="utf-8")
+            if not is_valid_source_content(result):
+                raise RuntimeError("obfuscator output looks invalid")
+            path.write_text(result.strip() + "\n", encoding="utf-8")
             return
         except Exception:
             pass
+        finally:
+            shutil.rmtree(tmpdir, ignore_errors=True)
+
     path.write_text(minify_js_fallback(original) + "\n", encoding="utf-8")
 
 
 def process_css(path: Path, cwd: Path) -> None:
     original = path.read_text(encoding="utf-8")
     if shutil.which("npx"):
+        tmpdir = Path(tempfile.mkdtemp())
         try:
-            minified = run_cmd(
+            src = tmpdir / "input.css"
+            out = tmpdir / "output.css"
+            src.write_text(original, encoding="utf-8")
+            run_cmd(
                 [
                     "npx",
                     "--yes",
                     "clean-css-cli",
+                    str(src),
+                    "-o",
+                    str(out),
                     "--skip-rebase",
                     "-O2",
                 ],
                 cwd,
-                input_text=original,
             )
-            path.write_text(minified.strip() + "\n", encoding="utf-8")
+            path.write_text(out.read_text(encoding="utf-8").strip() + "\n", encoding="utf-8")
             return
         except Exception:
             pass
+        finally:
+            shutil.rmtree(tmpdir, ignore_errors=True)
     path.write_text(minify_css_fallback(original) + "\n", encoding="utf-8")
 
 
@@ -266,8 +332,7 @@ def process_php(path: Path) -> None:
 
 
 def hash_file(path: Path) -> str:
-    digest = hashlib.sha256(path.read_bytes()).hexdigest()[:12]
-    return digest
+    return hashlib.sha256(path.read_bytes()).hexdigest()[:12]
 
 
 def replace_references(root: Path, mapping: dict[str, str]) -> None:
@@ -279,7 +344,7 @@ def replace_references(root: Path, mapping: dict[str, str]) -> None:
         except UnicodeDecodeError:
             continue
         updated = content
-        for src, dst in mapping.items():
+        for src, dst in sorted(mapping.items(), key=lambda item: len(item[0]), reverse=True):
             updated = updated.replace(src, dst)
             updated = updated.replace("/" + src, "/" + dst)
         if updated != content:
@@ -289,17 +354,30 @@ def replace_references(root: Path, mapping: dict[str, str]) -> None:
 def build_hash_mapping(public_root: Path) -> dict[str, str]:
     mapping: dict[str, str] = {}
     asset_root = public_root / "assets"
-    for ext in (".js", ".css"):
-        for file_path in sorted(asset_root.rglob(f"*{ext}")):
-            if ".min." in file_path.name or ".obf." in file_path.name:
-                continue
-            digest = hash_file(file_path)
-            new_name = f"{file_path.stem}.{digest}{file_path.suffix}"
-            new_path = file_path.with_name(new_name)
-            file_path.rename(new_path)
-            rel_old = file_path.relative_to(public_root).as_posix()
-            rel_new = new_path.relative_to(public_root).as_posix()
-            mapping[rel_old] = rel_new
+    if not asset_root.exists():
+        return mapping
+
+    groups = collect_asset_groups(asset_root)
+    for (parent, base, ext), paths in groups.items():
+        source = pick_source_file(paths, base, ext)
+        digest = hash_file(source)
+        target = parent / f"{base}.{digest}{ext}"
+        rel_new = target.relative_to(public_root).as_posix()
+
+        for old in paths:
+            rel_old = old.relative_to(public_root).as_posix()
+            if rel_old != rel_new:
+                mapping[rel_old] = rel_new
+
+        if source != target:
+            if target.exists():
+                target.unlink()
+            source.replace(target)
+
+        for old in paths:
+            if old != target and old.exists():
+                old.unlink()
+
     return mapping
 
 
@@ -316,11 +394,17 @@ def main() -> int:
         print("public directory not found", file=sys.stderr)
         return 1
 
-    for js in sorted(public_root.rglob("*.js")):
-        process_js(js, repo_root)
-    for css in sorted(public_root.rglob("*.css")):
-        process_css(css, repo_root)
-    for php in sorted((repo_root / "public").rglob("*.php")):
+    asset_root = public_root / "assets"
+    if asset_root.exists():
+        groups = collect_asset_groups(asset_root)
+        for (_parent, base, ext), paths in sorted(groups.items()):
+            source = pick_source_file(paths, base, ext)
+            if ext == ".js":
+                process_js(source, repo_root)
+            else:
+                process_css(source, repo_root)
+
+    for php in sorted(public_root.rglob("*.php")):
         process_php(php)
     for php in sorted((repo_root / "backend").rglob("*.php")):
         process_php(php)