Create safety.py

safety.py

@@ -0,0 +1,56 @@
+import shlex
+from pathlib import Path
+
+DISALLOWED_TOKENS = [";", "&&", "||", "|", ">", ">>", "<", "`", "$(", "${"]
+DISALLOWED_STRINGS = [
+    "pipe:", "concat:", "subfile:", "crypto:", "data:", "ftp:", "gopher:", "rtmp:", "rtsp:",
+    "-filter_script", "-f tee", "tee:"
+]
+
+def parse_ffmpeg_cmd(cmd: str) -> list[str]:
+    if not cmd or not cmd.strip():
+        raise ValueError("Empty ffmpeg command.")
+
+    for t in DISALLOWED_TOKENS:
+        if t in cmd:
+            raise ValueError(f"Unsafe token found: {t}")
+
+    lowered = cmd.lower()
+    for s in DISALLOWED_STRINGS:
+        if s in lowered:
+            raise ValueError(f"Disallowed ffmpeg feature/protocol: {s}")
+
+    parts = shlex.split(cmd)
+    if not parts or parts[0] != "ffmpeg":
+        raise ValueError("Command must start with 'ffmpeg'.")
+    return parts
+
+def enforce_local_inputs(parts: list[str], job_dir: Path):
+    input_dir = (job_dir / "input").resolve()
+    i = 0
+    while i < len(parts):
+        if parts[i] == "-i" and i + 1 < len(parts):
+            p = parts[i + 1]
+            if "://" in p:
+                raise ValueError("FFmpeg must not fetch URLs. Provide URLs via file_urls (server downloads them).")
+
+            abs_p = (job_dir / p).resolve() if not Path(p).is_absolute() else Path(p).resolve()
+            if not str(abs_p).startswith(str(input_dir)):
+                raise ValueError("FFmpeg inputs must be within ./input/")
+            i += 2
+        else:
+            i += 1
+
+def make_exec_args(parts: list[str], forced_output: Path) -> list[str]:
+    # ensure no interaction / overwrite
+    if "-nostdin" not in parts:
+        parts.insert(1, "-nostdin")
+    if "-y" not in parts:
+        parts.insert(1, "-y")
+
+    # remove last arg if it looks like output
+    if len(parts) > 1 and not parts[-1].startswith("-"):
+        parts = parts[:-1]
+
+    parts.append(str(forced_output))
+    return parts