🔒 [보안] CLI 무제한 파일 읽기 취약점 수정

.github/scripts/repair_pr_811.py

@@ -0,0 +1,124 @@
+from __future__ import annotations
+
+import subprocess
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def run(*args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
+    """Run one repository command and propagate failures by default."""
+    return subprocess.run(args, cwd=ROOT, check=check, text=True)
+
+
+def replace_once(path: Path, old: str, new: str, label: str) -> None:
+    """Replace one exact source fragment or fail before mutating the branch."""
+    text = path.read_text(encoding="utf-8")
+    if text.count(old) != 1:
+        raise RuntimeError(f"unexpected {label} shape")
+    path.write_text(text.replace(old, new), encoding="utf-8")
+
+
+def implement_fail_closed_argument_dispatch() -> None:
+    """Reject every malformed explicit argument mode before standard input."""
+    cli_path = ROOT / "services/analysis-engine/src/bandscope_analysis/cli.py"
+    replace_once(
+        cli_path,
+        '''        if cli_args[0] == "--status":
+            json.dump(get_analysis_status(), sys.stdout)
+            return 0
+        if cli_args[0] == "--job":
+''',
+        '''        if cli_args[0] == "--status":
+            if len(cli_args) != 1:
+                json.dump(
+                    failed_cli_response("--status does not accept additional arguments"),
+                    sys.stdout,
+                )
+                return 1
+            json.dump(get_analysis_status(), sys.stdout)
+            return 0
+        if cli_args[0] == "--job":
+''',
+        "status argument dispatch",
+    )
+    replace_once(
+        cli_path,
+        '''                except Exception:
+                    json.dump(failed_cli_response("Failed to read job file"), sys.stdout)
+                    return 1
+
+    if input_data is None:
+''',
+        '''                except Exception:
+                    json.dump(failed_cli_response("Failed to read job file"), sys.stdout)
+                    return 1
+        else:
+            json.dump(failed_cli_response("Unsupported CLI arguments"), sys.stdout)
+            return 1
+
+    if input_data is None:
+''',
+        "unknown argument dispatch",
+    )
+
+    changelog_path = ROOT / "CHANGELOG.md"
+    marker = "## [Unreleased]\n"
+    addition = (
+        "\n### Fixed\n\n"
+        "- Reject unknown CLI arguments and extra `--status` operands before reading standard "
+        "input, so malformed explicit invocations fail immediately instead of blocking on an "
+        "unrelated open pipe.\n"
+    )
+    replace_once(changelog_path, marker, marker + addition, "Unreleased heading")
+
+
+def main() -> None:
+    """Execute focused RED/GREEN, full verification, and workflow self-removal."""
+    run("uv", "sync", "--project", "services/analysis-engine", "--group", "dev", "--frozen")
+    red = run(
+        "uv",
+        "run",
+        "--project",
+        "services/analysis-engine",
+        "pytest",
+        "-q",
+        "services/analysis-engine/tests/test_cli_unknown_arguments.py",
+        check=False,
+    )
+    if red.returncode == 0:
+        raise RuntimeError("expected malformed explicit argument dispatch to fail before repair")
+
+    implement_fail_closed_argument_dispatch()
+    run(
+        "uv",
+        "run",
+        "--project",
+        "services/analysis-engine",
+        "pytest",
+        "-q",
+        "services/analysis-engine/tests/test_cli_unknown_arguments.py",
+        "services/analysis-engine/tests/test_cli_input_bounds.py",
+        "services/analysis-engine/tests/test_cli.py",
+    )
+    run("./scripts/harness/quickcheck.sh")
+
+    (ROOT / ".github/workflows/repair-pr-811-argument-dispatch.yml").unlink()
+    Path(__file__).unlink()
+    run("git", "config", "user.name", "CWL repair bot")
+    run("git", "config", "user.email", "actions@users.noreply.github.com")
+    run(
+        "git",
+        "add",
+        "CHANGELOG.md",
+        "services/analysis-engine/src/bandscope_analysis/cli.py",
+        "services/analysis-engine/tests/test_cli_unknown_arguments.py",
+        ".github/workflows/repair-pr-811-argument-dispatch.yml",
+        ".github/scripts/repair_pr_811.py",
+    )
+    run("git", "commit", "-m", "fix(cli): reject unsupported explicit arguments")
+    run("git", "push", "origin", "HEAD:fix-cli-unbounded-read-5165758910965089497")
+
+
+if __name__ == "__main__":
+    main()

.github/workflows/repair-pr-811-argument-dispatch.yml

@@ -0,0 +1,34 @@
+name: Repair PR 811 explicit argument dispatch
+
+on:
+  push:
+    branches:
+      - fix-cli-unbounded-read-5165758910965089497
+    paths:
+      - .github/workflows/repair-pr-811-argument-dispatch.yml
+
+permissions:
+  contents: write
+
+env:
+  GIT_CONFIG_COUNT: "1"
+  GIT_CONFIG_KEY_0: init.defaultBranch
+  GIT_CONFIG_VALUE_0: develop
+
+jobs:
+  repair:
+    if: >-
+      github.repository == 'ContextualWisdomLab/bandscope' &&
+      github.ref == 'refs/heads/fix-cli-unbounded-read-5165758910965089497'
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        with:
+          fetch-depth: 0
+          ref: fix-cli-unbounded-read-5165758910965089497
+      - uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        with:
+          version: "0.8.6"
+          enable-cache: false
+      - name: Execute bounded test-first repair
+        run: python3 .github/scripts/repair_pr_811.py

services/analysis-engine/src/bandscope_analysis/cli.py

@@ -8,10 +8,17 @@
 from datetime import UTC, datetime
 
 from bandscope_analysis.api import get_analysis_status, run_analysis_job, run_analysis_job_updates
-from bandscope_analysis.temporal import TemporalAnalyzer
+from bandscope_analysis.temporal import TemporalAnalyzer as _TemporalAnalyzer
 
 logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
 
+MAX_JSON_FILE_SIZE = 10 * 1024 * 1024  # 10 MB
+
+# Compatibility hook for existing CLI-level tests and downstream monkeypatches.
+# The CLI intentionally does not invoke temporal analysis before request validation;
+# validated orchestration owns every local-audio file access.
+TemporalAnalyzer = _TemporalAnalyzer
+
 
 def failed_cli_response(message: str) -> dict[str, object]:
     """Return a typed CLI failure envelope for malformed stdin payloads."""
@@ -28,28 +35,81 @@ def failed_cli_response(message: str) -> dict[str, object]:
     }
 
 
+def _read_bounded_stdin() -> tuple[str | None, int]:
+    """Read one bounded UTF-8 stdin payload and return text plus an exit code.
+
+    Standard process stdin exposes ``buffer``; enforce the allocation bound on
+    raw bytes before UTF-8 decoding. Text-only injected streams are already
+    decoded outside this boundary, so retain a compatibility path for in-process
+    callers. Failures are emitted here so callers never need to retain rejected
+    payload content.
+    """
+    binary_stdin = getattr(sys.stdin, "buffer", None)
+    if binary_stdin is None:
+        raw_text = sys.stdin.read(MAX_JSON_FILE_SIZE + 1)
+        raw_bytes = raw_text.encode("utf-8")
+    else:
+        raw_bytes = binary_stdin.read(MAX_JSON_FILE_SIZE + 1)
+    if len(raw_bytes) > MAX_JSON_FILE_SIZE:
+        json.dump(failed_cli_response("Job input exceeds maximum size limit"), sys.stdout)
+        return None, 1
+    try:
+        raw_text = raw_bytes.decode("utf-8")
+    except UnicodeDecodeError:
+        json.dump(failed_cli_response("Job input must be valid UTF-8"), sys.stdout)
+        return None, 1
+    return raw_text.strip(), 0
+
+
 def main() -> int:
-    """Read a job payload from stdin and print a structured job response to stdout."""
-    # Read all input from stdin first
-    input_data = sys.stdin.read().strip()
+    """Read one explicit argument or bounded stdin job and print its response."""
     progress_jsonl = "--progress-jsonl" in sys.argv[1:]
     cli_args = [arg for arg in sys.argv[1:] if arg != "--progress-jsonl"]
+    input_data: str | None = None
 
-    # Check if there are command line arguments (fallback for manual testing)
+    # Explicit argument modes own their input source. Resolve them before touching
+    # stdin so ``--status`` and ``--job`` cannot block on an unrelated open pipe or
+    # consume data that the caller did not select as the job payload.
     if cli_args:
         if cli_args[0] == "--status":
             json.dump(get_analysis_status(), sys.stdout)
             return 0
-        elif cli_args[0] == "--job" and len(cli_args) > 1:
+        if cli_args[0] == "--job":
+            if len(cli_args) != 2:
+                json.dump(
+                    failed_cli_response(
+                        "--job requires exactly one JSON payload or file path"
+                    ),
+                    sys.stdout,
+                )
+                return 1
             input_data = cli_args[1]
-            if not input_data.startswith("{"):
+            if input_data.startswith("{"):
+                if len(input_data.encode("utf-8")) > MAX_JSON_FILE_SIZE:
+                    json.dump(
+                        failed_cli_response("Job input exceeds maximum size limit"), sys.stdout
+                    )
+                    return 1
+            else:
                 try:
-                    with open(input_data, "r", encoding="utf-8") as f:
-                        input_data = f.read()
+                    with open(input_data, "rb") as f:
+                        input_bytes = f.read(MAX_JSON_FILE_SIZE + 1)
+                        if len(input_bytes) > MAX_JSON_FILE_SIZE:
+                            json.dump(
+                                failed_cli_response("Job file exceeds maximum size limit"),
+                                sys.stdout,
+                            )
+                            return 1
+                        input_data = input_bytes.decode("utf-8")
                 except Exception:
                     json.dump(failed_cli_response("Failed to read job file"), sys.stdout)
                     return 1
 
+    if input_data is None:
+        input_data, stdin_exit_code = _read_bounded_stdin()
+        if input_data is None:
+            return stdin_exit_code
+
     if not input_data:
         json.dump(failed_cli_response("Empty input"), sys.stdout)
         return 0
@@ -74,29 +134,6 @@ def main() -> int:
         return 0
 
     request = payload.get("request")
-
-    # Temporary: Inject temporal analyzer call if it's a local file, just to prove it works
-    # before full orchestrator integration
-    if (
-        isinstance(request, dict)
-        and request.get("sourceKind") == "local_audio"
-        and "localSource" in request
-    ):
-        local_source = request["localSource"]
-        audio_path = local_source.get("sourcePath")
-        file_name = local_source.get("fileName", "selected audio")
-        if audio_path:
-            logging.info("Extracting temporal features from %s...", file_name)
-            try:
-                temporal_analyzer = TemporalAnalyzer()
-                features = temporal_analyzer.analyze(audio_path)
-                logging.info(f"Extracted BPM: {features['bpm']}")
-            except Exception:
-                logging.warning(
-                    "Temporal analysis failed for %s; continuing with safe fallback.",
-                    file_name,
-                )
-
     requested_at = datetime.now(UTC).isoformat().replace("+00:00", "Z")
     if progress_jsonl:
         for update in run_analysis_job_updates(job_id, request, requested_at):

services/analysis-engine/tests/test_cli.py

@@ -487,3 +487,23 @@ def fake_stem_separation(*args: Any, **kwargs: Any) -> dict[str, Any]:
     ]
     assert updates[-1]["state"] == "succeeded"
     assert updates[-1]["progressPercent"] == 100
+
+
+def test_cli_main_job_arg_rejects_large_file(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None:
+    """Ensure --job rejects files larger than MAX_JSON_FILE_SIZE."""
+    stdin = io.StringIO("")
+    stdout = io.StringIO()
+    job_file = tmp_path / "large_job.json"
+
+    # Create a dummy file larger than MAX_JSON_FILE_SIZE
+    from bandscope_analysis.cli import MAX_JSON_FILE_SIZE
+
+    with open(job_file, "wb") as f:
+        f.seek(MAX_JSON_FILE_SIZE + 1024)
+        f.write(b"0")
+
+    monkeypatch.setattr(cli.sys, "argv", ["cli.py", "--job", str(job_file)])
+    monkeypatch.setattr(cli.sys, "stdin", stdin)
+    monkeypatch.setattr(cli.sys, "stdout", stdout)
+    assert cli.main() == 1
+    assert "Job file exceeds maximum size limit" in stdout.getvalue()