test_security_code_porcelain.py
python
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9
docs: expand cache plan with all seven testing tiers and do…
Sonnet 4.6
132 days ago
| 1 | """Security regression tests for Muse code domain porcelain commands. |
| 2 | |
| 3 | Red-hat findings from the semantic audit, turned into blue-hat defences: |
| 4 | |
| 5 | 1. ANSI / OSC terminal injection — 30+ commands printed user-controlled strings |
| 6 | (symbol addresses, commit messages, file paths) without sanitize_display. |
| 7 | Any commit with an OSC-52 payload in its message could hijack the clipboard |
| 8 | of every developer whose terminal renders that output. |
| 9 | |
| 10 | 2. Integer denial-of-service — 13 commands accept unbounded int arguments |
| 11 | (--top, --max-commits, --workers, --context, --limit, --min-co-changes, |
| 12 | --window, --predict). Passing 2147483647 triggers enormous allocations or |
| 13 | infinite-feeling loops that exhaust memory and CPU. |
| 14 | |
| 15 | 3. Output-path traversal — docs_cmd wrote to pathlib.Path(args.output) without |
| 16 | contain_path, allowing --output /etc/cron.d/evil to escape the repo. |
| 17 | """ |
| 18 | |
| 19 | from __future__ import annotations |
| 20 | |
| 21 | import datetime |
| 22 | import json |
| 23 | import pathlib |
| 24 | import time |
| 25 | |
| 26 | import pytest |
| 27 | |
| 28 | from tests.cli_test_helper import CliRunner |
| 29 | from muse.core._types import fake_id, blob_id |
| 30 | from muse.core.object_store import write_object as _write_obj_store |
| 31 | |
| 32 | cli = None # post-argparse migration stub |
| 33 | runner = CliRunner() |
| 34 | |
| 35 | # --------------------------------------------------------------------------- |
| 36 | # OSC-52 payload — NOT stripped by CliRunner._strip_ansi (which only strips |
| 37 | # \x1b[...m sequences), but IS stripped by sanitize_display (which removes |
| 38 | # every C0/C1 control character including ESC = 0x1B and BEL = 0x07). |
| 39 | # --------------------------------------------------------------------------- |
| 40 | _ANSI_PAYLOAD: str = "sec\x1b]52;c;HACKED==\x07end" |
| 41 | _ANSI_MARKER: str = "\x1b" |
| 42 | |
| 43 | |
| 44 | # --------------------------------------------------------------------------- |
| 45 | # Shared repo helpers |
| 46 | # --------------------------------------------------------------------------- |
| 47 | |
| 48 | def _env(root: pathlib.Path) -> Manifest: |
| 49 | return {"MUSE_REPO_ROOT": str(root)} |
| 50 | |
| 51 | |
| 52 | def _init_code_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]: |
| 53 | muse_dir = tmp_path / ".muse" |
| 54 | muse_dir.mkdir() |
| 55 | repo_id = fake_id("repo") |
| 56 | (muse_dir / "repo.json").write_text( |
| 57 | json.dumps({ |
| 58 | "repo_id": repo_id, |
| 59 | "domain": "code", |
| 60 | "default_branch": "main", |
| 61 | "created_at": "2025-01-01T00:00:00+00:00", |
| 62 | }), |
| 63 | encoding="utf-8", |
| 64 | ) |
| 65 | (muse_dir / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8") |
| 66 | (muse_dir / "refs" / "heads").mkdir(parents=True) |
| 67 | (muse_dir / "snapshots").mkdir() |
| 68 | (muse_dir / "commits").mkdir() |
| 69 | (muse_dir / "objects").mkdir() |
| 70 | return tmp_path, repo_id |
| 71 | |
| 72 | |
| 73 | def _store_object(root: pathlib.Path, content: bytes) -> str: |
| 74 | """Write *content* into the object store and return its sha256:-prefixed id.""" |
| 75 | oid = blob_id(content) |
| 76 | _write_obj_store(root, oid, content) |
| 77 | return oid |
| 78 | |
| 79 | |
| 80 | def _make_commit( |
| 81 | root: pathlib.Path, |
| 82 | repo_id: str, |
| 83 | message: str = "init", |
| 84 | manifest: Manifest | None = None, |
| 85 | ) -> str: |
| 86 | from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot |
| 87 | from muse.core.snapshot import compute_snapshot_id, compute_commit_id |
| 88 | |
| 89 | ref_file = root / ".muse" / "refs" / "heads" / "main" |
| 90 | parent_id = ref_file.read_text().strip() if ref_file.exists() else None |
| 91 | m: Manifest = manifest or {} |
| 92 | snap_id = compute_snapshot_id(m) |
| 93 | committed_at = datetime.datetime.now(datetime.timezone.utc) |
| 94 | commit_id = compute_commit_id( |
| 95 | repo_id=repo_id, |
| 96 | parent_ids=[parent_id] if parent_id else [], |
| 97 | snapshot_id=snap_id, |
| 98 | message=message, |
| 99 | committed_at_iso=committed_at.isoformat(), |
| 100 | ) |
| 101 | write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=m)) |
| 102 | write_commit(root, CommitRecord( |
| 103 | commit_id=commit_id, |
| 104 | repo_id=repo_id, |
| 105 | created_on_branch="main", |
| 106 | snapshot_id=snap_id, |
| 107 | message=message, |
| 108 | committed_at=committed_at, |
| 109 | parent_commit_id=parent_id, |
| 110 | )) |
| 111 | ref_file.parent.mkdir(parents=True, exist_ok=True) |
| 112 | ref_file.write_text(commit_id, encoding="utf-8") |
| 113 | return commit_id |
| 114 | |
| 115 | |
| 116 | def _ansi_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]: |
| 117 | """Create a repo whose most-recent commit message contains an OSC-52 payload. |
| 118 | |
| 119 | Also includes a Python source file so symbol-related commands have data |
| 120 | to work with. The commit message injection is the primary ANSI attack |
| 121 | vector tested here — it affects every command that echoes messages. |
| 122 | """ |
| 123 | root, repo_id = _init_code_repo(tmp_path) |
| 124 | |
| 125 | # First commit — clean Python file |
| 126 | py_src = b"def alpha():\n return 1\n\ndef beta():\n return 2\n" |
| 127 | oid = _store_object(root, py_src) |
| 128 | # Create the physical file so working-tree commands can resolve it. |
| 129 | src_dir = root / "src" |
| 130 | src_dir.mkdir() |
| 131 | (src_dir / "module.py").write_bytes(py_src) |
| 132 | _make_commit(root, repo_id, "initial commit", {"src/module.py": oid}) |
| 133 | |
| 134 | # Second commit — same file, different body (creates churn for hotspots/stable) |
| 135 | py_src2 = b"def alpha():\n return 99\n\ndef beta():\n return 2\n" |
| 136 | oid2 = _store_object(root, py_src2) |
| 137 | (src_dir / "module.py").write_bytes(py_src2) |
| 138 | # Message carries the OSC-52 payload — the primary injection vector. |
| 139 | _make_commit(root, repo_id, _ANSI_PAYLOAD, {"src/module.py": oid2}) |
| 140 | |
| 141 | return root, repo_id |
| 142 | |
| 143 | |
| 144 | # --------------------------------------------------------------------------- |
| 145 | # § 1 — ANSI / OSC terminal injection |
| 146 | # |
| 147 | # Each test invokes one code porcelain command against the ANSI fixture repo |
| 148 | # and asserts that ESC (0x1B) is absent from the captured output. |
| 149 | # |
| 150 | # CliRunner._strip_ansi removes \x1b[...m sequences but NOT OSC sequences |
| 151 | # like \x1b]52;...\x07. sanitize_display (which the commands must call) |
| 152 | # removes ALL C0/C1 control characters including ESC. |
| 153 | # --------------------------------------------------------------------------- |
| 154 | |
| 155 | class TestAnsiInjectionCommit: |
| 156 | """Commands that display commit messages must not echo raw ESC bytes.""" |
| 157 | |
| 158 | def test_symbol_log_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 159 | root, _ = _ansi_repo(tmp_path) |
| 160 | r = runner.invoke(cli, ["code", "symbol-log", "src/module.py::alpha"], env=_env(root)) |
| 161 | assert _ANSI_MARKER not in r.output |
| 162 | |
| 163 | def test_blame_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 164 | root, _ = _ansi_repo(tmp_path) |
| 165 | r = runner.invoke(cli, ["code", "blame", "src/module.py::alpha"], env=_env(root)) |
| 166 | assert _ANSI_MARKER not in r.output |
| 167 | |
| 168 | def test_find_symbol_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 169 | root, _ = _ansi_repo(tmp_path) |
| 170 | r = runner.invoke(cli, ["code", "find-symbol", "--name", "alpha"], env=_env(root)) |
| 171 | assert _ANSI_MARKER not in r.output |
| 172 | |
| 173 | def test_narrative_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 174 | root, _ = _ansi_repo(tmp_path) |
| 175 | r = runner.invoke( |
| 176 | cli, |
| 177 | ["code", "narrative", "src/module.py::alpha", "--max-commits", "5"], |
| 178 | env=_env(root), |
| 179 | ) |
| 180 | assert _ANSI_MARKER not in r.output |
| 181 | |
| 182 | def test_contract_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 183 | root, _ = _ansi_repo(tmp_path) |
| 184 | r = runner.invoke( |
| 185 | cli, |
| 186 | ["code", "contract", "src/module.py::alpha", "--max-commits", "5"], |
| 187 | env=_env(root), |
| 188 | ) |
| 189 | assert _ANSI_MARKER not in r.output |
| 190 | |
| 191 | def test_detect_refactor_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 192 | root, _ = _ansi_repo(tmp_path) |
| 193 | r = runner.invoke( |
| 194 | cli, ["code", "detect-refactor", "--max-commits", "5"], env=_env(root) |
| 195 | ) |
| 196 | assert _ANSI_MARKER not in r.output |
| 197 | |
| 198 | def test_query_history_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 199 | root, _ = _ansi_repo(tmp_path) |
| 200 | r = runner.invoke(cli, ["code", "query-history", "kind=function"], env=_env(root)) |
| 201 | assert _ANSI_MARKER not in r.output |
| 202 | |
| 203 | |
| 204 | class TestAnsiInjectionAddress: |
| 205 | """Commands that display symbol addresses must not echo raw ESC bytes. |
| 206 | |
| 207 | We verify the sanitize_display path is called for address output. The |
| 208 | OSC-52 injection payload is embedded in the commit message (guaranteed |
| 209 | to appear in commands that echo messages). Symbol-address injection via |
| 210 | filesystem paths is impossible on most OSes; we rely on the code-review |
| 211 | audit and sanitize_display application at all print sites for that case. |
| 212 | """ |
| 213 | |
| 214 | def test_hotspots_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 215 | root, _ = _ansi_repo(tmp_path) |
| 216 | r = runner.invoke(cli, ["code", "hotspots", "--top", "10"], env=_env(root)) |
| 217 | assert _ANSI_MARKER not in r.output |
| 218 | |
| 219 | def test_stable_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 220 | root, _ = _ansi_repo(tmp_path) |
| 221 | r = runner.invoke(cli, ["code", "stable", "--top", "10"], env=_env(root)) |
| 222 | assert _ANSI_MARKER not in r.output |
| 223 | |
| 224 | def test_symbols_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 225 | root, _ = _ansi_repo(tmp_path) |
| 226 | r = runner.invoke(cli, ["code", "symbols"], env=_env(root)) |
| 227 | assert _ANSI_MARKER not in r.output |
| 228 | |
| 229 | def test_grep_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 230 | root, _ = _ansi_repo(tmp_path) |
| 231 | r = runner.invoke(cli, ["code", "grep", "alpha"], env=_env(root)) |
| 232 | assert _ANSI_MARKER not in r.output |
| 233 | |
| 234 | def test_cat_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 235 | root, _ = _ansi_repo(tmp_path) |
| 236 | r = runner.invoke( |
| 237 | cli, ["code", "cat", "src/module.py::alpha"], env=_env(root) |
| 238 | ) |
| 239 | assert _ANSI_MARKER not in r.output |
| 240 | |
| 241 | def test_blast_risk_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 242 | root, _ = _ansi_repo(tmp_path) |
| 243 | r = runner.invoke( |
| 244 | cli, |
| 245 | ["code", "blast-risk", "--top", "5", "--max-commits", "5"], |
| 246 | env=_env(root), |
| 247 | ) |
| 248 | assert _ANSI_MARKER not in r.output |
| 249 | |
| 250 | def test_age_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 251 | root, _ = _ansi_repo(tmp_path) |
| 252 | r = runner.invoke( |
| 253 | cli, |
| 254 | ["code", "age", "src/module.py::alpha", "--max-commits", "5"], |
| 255 | env=_env(root), |
| 256 | ) |
| 257 | assert _ANSI_MARKER not in r.output |
| 258 | |
| 259 | def test_velocity_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 260 | root, _ = _ansi_repo(tmp_path) |
| 261 | r = runner.invoke( |
| 262 | cli, |
| 263 | ["code", "velocity", "--top", "5", "--max-commits", "5"], |
| 264 | env=_env(root), |
| 265 | ) |
| 266 | assert _ANSI_MARKER not in r.output |
| 267 | |
| 268 | def test_entangle_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 269 | root, _ = _ansi_repo(tmp_path) |
| 270 | r = runner.invoke( |
| 271 | cli, |
| 272 | ["code", "entangle", "--top", "5", "--max-commits", "5"], |
| 273 | env=_env(root), |
| 274 | ) |
| 275 | assert _ANSI_MARKER not in r.output |
| 276 | |
| 277 | def test_gravity_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 278 | root, _ = _ansi_repo(tmp_path) |
| 279 | r = runner.invoke( |
| 280 | cli, |
| 281 | ["code", "gravity", "src/module.py::alpha", "--max-commits", "5"], |
| 282 | env=_env(root), |
| 283 | ) |
| 284 | assert _ANSI_MARKER not in r.output |
| 285 | |
| 286 | def test_impact_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 287 | root, _ = _ansi_repo(tmp_path) |
| 288 | r = runner.invoke( |
| 289 | cli, ["code", "impact", "src/module.py::alpha"], env=_env(root) |
| 290 | ) |
| 291 | assert _ANSI_MARKER not in r.output |
| 292 | |
| 293 | def test_deps_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 294 | root, _ = _ansi_repo(tmp_path) |
| 295 | r = runner.invoke( |
| 296 | cli, ["code", "deps", "src/module.py"], env=_env(root) |
| 297 | ) |
| 298 | assert _ANSI_MARKER not in r.output |
| 299 | |
| 300 | def test_coverage_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 301 | root, _ = _ansi_repo(tmp_path) |
| 302 | r = runner.invoke( |
| 303 | cli, ["code", "coverage", "src/module.py::alpha"], env=_env(root) |
| 304 | ) |
| 305 | assert _ANSI_MARKER not in r.output |
| 306 | |
| 307 | def test_lineage_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 308 | root, _ = _ansi_repo(tmp_path) |
| 309 | r = runner.invoke( |
| 310 | cli, ["code", "lineage", "src/module.py::alpha"], env=_env(root) |
| 311 | ) |
| 312 | assert _ANSI_MARKER not in r.output |
| 313 | |
| 314 | def test_api_surface_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 315 | root, _ = _ansi_repo(tmp_path) |
| 316 | r = runner.invoke(cli, ["code", "api-surface"], env=_env(root)) |
| 317 | assert _ANSI_MARKER not in r.output |
| 318 | |
| 319 | def test_dead_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 320 | root, _ = _ansi_repo(tmp_path) |
| 321 | r = runner.invoke(cli, ["code", "dead"], env=_env(root)) |
| 322 | assert _ANSI_MARKER not in r.output |
| 323 | |
| 324 | def test_clones_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 325 | root, _ = _ansi_repo(tmp_path) |
| 326 | r = runner.invoke(cli, ["code", "clones"], env=_env(root)) |
| 327 | assert _ANSI_MARKER not in r.output |
| 328 | |
| 329 | def test_codemap_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 330 | root, _ = _ansi_repo(tmp_path) |
| 331 | r = runner.invoke(cli, ["code", "codemap", "--top", "5"], env=_env(root)) |
| 332 | assert _ANSI_MARKER not in r.output |
| 333 | |
| 334 | def test_coupling_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 335 | root, _ = _ansi_repo(tmp_path) |
| 336 | r = runner.invoke( |
| 337 | cli, ["code", "coupling", "--top", "5", "--min", "1"], env=_env(root) |
| 338 | ) |
| 339 | assert _ANSI_MARKER not in r.output |
| 340 | |
| 341 | def test_compare_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 342 | root, _ = _ansi_repo(tmp_path) |
| 343 | r = runner.invoke( |
| 344 | cli, ["code", "compare", "HEAD~1", "HEAD"], env=_env(root) |
| 345 | ) |
| 346 | assert _ANSI_MARKER not in r.output |
| 347 | |
| 348 | def test_semantic_test_coverage_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 349 | root, _ = _ansi_repo(tmp_path) |
| 350 | r = runner.invoke( |
| 351 | cli, ["code", "semantic-test-coverage", "--max-commits", "5"], env=_env(root) |
| 352 | ) |
| 353 | assert _ANSI_MARKER not in r.output |
| 354 | |
| 355 | def test_predict_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 356 | root, _ = _ansi_repo(tmp_path) |
| 357 | r = runner.invoke( |
| 358 | cli, ["code", "predict", "--top", "5", "--max-commits", "5"], env=_env(root) |
| 359 | ) |
| 360 | assert _ANSI_MARKER not in r.output |
| 361 | |
| 362 | def test_patch_error_message_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 363 | """patch echoes the address back on error — must sanitize it.""" |
| 364 | root, _ = _ansi_repo(tmp_path) |
| 365 | evil_addr = f"src/module.py::\x1b]52;c;evil\x07func" |
| 366 | r = runner.invoke( |
| 367 | cli, ["code", "patch", evil_addr, "--body", "-"], |
| 368 | env=_env(root), input="def func(): pass", |
| 369 | ) |
| 370 | assert _ANSI_MARKER not in r.output |
| 371 | |
| 372 | def test_checkout_symbol_error_message_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 373 | """checkout-symbol echoes the address on error — must sanitize.""" |
| 374 | root, _ = _ansi_repo(tmp_path) |
| 375 | evil_addr = f"src/module.py::\x1b]52;c;evil\x07func" |
| 376 | r = runner.invoke( |
| 377 | cli, ["code", "checkout-symbol", evil_addr], env=_env(root) |
| 378 | ) |
| 379 | assert _ANSI_MARKER not in r.output |
| 380 | |
| 381 | def test_semantic_cherry_pick_error_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 382 | root, _ = _ansi_repo(tmp_path) |
| 383 | evil_addr = f"src/module.py::\x1b]52;c;evil\x07func" |
| 384 | r = runner.invoke( |
| 385 | cli, ["code", "semantic-cherry-pick", evil_addr, "--from", "HEAD~1"], |
| 386 | env=_env(root), |
| 387 | ) |
| 388 | assert _ANSI_MARKER not in r.output |
| 389 | |
| 390 | def test_query_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 391 | root, _ = _ansi_repo(tmp_path) |
| 392 | r = runner.invoke(cli, ["code", "query", "kind=function"], env=_env(root)) |
| 393 | assert _ANSI_MARKER not in r.output |
| 394 | |
| 395 | def test_docs_cmd_no_ansi(self, tmp_path: pathlib.Path) -> None: |
| 396 | root, _ = _ansi_repo(tmp_path) |
| 397 | r = runner.invoke( |
| 398 | cli, |
| 399 | ["code", "docs", "history", "src/module.py::alpha"], |
| 400 | env=_env(root), |
| 401 | ) |
| 402 | assert _ANSI_MARKER not in r.output |
| 403 | |
| 404 | |
| 405 | # --------------------------------------------------------------------------- |
| 406 | # § 2 — Integer denial-of-service |
| 407 | # |
| 408 | # Commands with unbounded --top / --max-commits / --workers etc. must reject |
| 409 | # extreme values rather than allocating gigabytes of memory or looping for |
| 410 | # unbounded time. |
| 411 | # |
| 412 | # The test passes if: the command returns exit_code != 0 (clamped and |
| 413 | # rejected) OR it completes within a generous 5-second wall-clock budget |
| 414 | # (the correct behaviour after clamping is applied). |
| 415 | # --------------------------------------------------------------------------- |
| 416 | |
| 417 | _DOS_BUDGET_S: float = 5.0 # max wall-clock seconds for a command with huge arg |
| 418 | |
| 419 | |
| 420 | class TestIntegerDoS: |
| 421 | """Unbounded numeric args must be clamped; commands must not hang or OOM.""" |
| 422 | |
| 423 | def _check( |
| 424 | self, |
| 425 | root: pathlib.Path, |
| 426 | args: list[str], |
| 427 | huge_value: str = "2147483647", |
| 428 | ) -> None: |
| 429 | """Run the command with *huge_value* injected at the right position. |
| 430 | |
| 431 | Asserts: either exit_code != 0 (arg rejected) OR elapsed < _DOS_BUDGET_S. |
| 432 | A command that simply produces no output in time is fine; one that |
| 433 | hangs indefinitely is not. |
| 434 | """ |
| 435 | t0 = time.monotonic() |
| 436 | r = runner.invoke(cli, args, env=_env(root)) |
| 437 | elapsed = time.monotonic() - t0 |
| 438 | if r.exit_code == 0: |
| 439 | assert elapsed < _DOS_BUDGET_S, ( |
| 440 | f"Command {args} took {elapsed:.1f}s > budget {_DOS_BUDGET_S}s " |
| 441 | "with max-int arg — clamp_int guard is missing" |
| 442 | ) |
| 443 | # exit_code != 0 means the guard rejected the huge value (preferred) |
| 444 | |
| 445 | def test_hotspots_top_dos(self, tmp_path: pathlib.Path) -> None: |
| 446 | root, _ = _ansi_repo(tmp_path) |
| 447 | self._check(root, ["code", "hotspots", "--top", "2147483647"]) |
| 448 | |
| 449 | def test_hotspots_max_commits_dos(self, tmp_path: pathlib.Path) -> None: |
| 450 | root, _ = _ansi_repo(tmp_path) |
| 451 | self._check(root, ["code", "hotspots", "--max-commits", "2147483647"]) |
| 452 | |
| 453 | def test_stable_top_dos(self, tmp_path: pathlib.Path) -> None: |
| 454 | root, _ = _ansi_repo(tmp_path) |
| 455 | self._check(root, ["code", "stable", "--top", "2147483647"]) |
| 456 | |
| 457 | def test_coupling_top_dos(self, tmp_path: pathlib.Path) -> None: |
| 458 | root, _ = _ansi_repo(tmp_path) |
| 459 | self._check(root, ["code", "coupling", "--top", "2147483647"]) |
| 460 | |
| 461 | def test_coupling_min_dos(self, tmp_path: pathlib.Path) -> None: |
| 462 | root, _ = _ansi_repo(tmp_path) |
| 463 | self._check(root, ["code", "coupling", "--min", "2147483647"]) |
| 464 | |
| 465 | def test_blast_risk_top_dos(self, tmp_path: pathlib.Path) -> None: |
| 466 | root, _ = _ansi_repo(tmp_path) |
| 467 | self._check(root, ["code", "blast-risk", "--top", "2147483647"]) |
| 468 | |
| 469 | def test_blast_risk_max_commits_dos(self, tmp_path: pathlib.Path) -> None: |
| 470 | root, _ = _ansi_repo(tmp_path) |
| 471 | self._check(root, ["code", "blast-risk", "--max-commits", "2147483647"]) |
| 472 | |
| 473 | def test_age_max_commits_dos(self, tmp_path: pathlib.Path) -> None: |
| 474 | root, _ = _ansi_repo(tmp_path) |
| 475 | self._check( |
| 476 | root, ["code", "age", "src/module.py::alpha", "--max-commits", "2147483647"] |
| 477 | ) |
| 478 | |
| 479 | def test_velocity_top_dos(self, tmp_path: pathlib.Path) -> None: |
| 480 | root, _ = _ansi_repo(tmp_path) |
| 481 | self._check(root, ["code", "velocity", "--top", "2147483647"]) |
| 482 | |
| 483 | def test_velocity_max_commits_dos(self, tmp_path: pathlib.Path) -> None: |
| 484 | root, _ = _ansi_repo(tmp_path) |
| 485 | self._check(root, ["code", "velocity", "--max-commits", "2147483647"]) |
| 486 | |
| 487 | def test_entangle_top_dos(self, tmp_path: pathlib.Path) -> None: |
| 488 | root, _ = _ansi_repo(tmp_path) |
| 489 | self._check(root, ["code", "entangle", "--top", "2147483647"]) |
| 490 | |
| 491 | def test_entangle_max_commits_dos(self, tmp_path: pathlib.Path) -> None: |
| 492 | root, _ = _ansi_repo(tmp_path) |
| 493 | self._check(root, ["code", "entangle", "--max-commits", "2147483647"]) |
| 494 | |
| 495 | def test_entangle_min_co_changes_dos(self, tmp_path: pathlib.Path) -> None: |
| 496 | root, _ = _ansi_repo(tmp_path) |
| 497 | self._check(root, ["code", "entangle", "--min-co-changes", "2147483647"]) |
| 498 | |
| 499 | def test_find_symbol_limit_dos(self, tmp_path: pathlib.Path) -> None: |
| 500 | root, _ = _ansi_repo(tmp_path) |
| 501 | self._check( |
| 502 | root, ["code", "find-symbol", "--name", "alpha", "--limit", "2147483647"] |
| 503 | ) |
| 504 | |
| 505 | def test_dead_workers_dos(self, tmp_path: pathlib.Path) -> None: |
| 506 | root, _ = _ansi_repo(tmp_path) |
| 507 | self._check(root, ["code", "dead", "--workers", "99999"]) |
| 508 | |
| 509 | def test_codemap_top_dos(self, tmp_path: pathlib.Path) -> None: |
| 510 | root, _ = _ansi_repo(tmp_path) |
| 511 | self._check(root, ["code", "codemap", "--top", "2147483647"]) |
| 512 | |
| 513 | def test_cat_context_dos(self, tmp_path: pathlib.Path) -> None: |
| 514 | root, _ = _ansi_repo(tmp_path) |
| 515 | self._check( |
| 516 | root, |
| 517 | ["code", "cat", "src/module.py::alpha", "--context", "2147483647"], |
| 518 | ) |
| 519 | |
| 520 | def test_detect_refactor_max_commits_dos(self, tmp_path: pathlib.Path) -> None: |
| 521 | root, _ = _ansi_repo(tmp_path) |
| 522 | self._check(root, ["code", "detect-refactor", "--max-commits", "2147483647"]) |
| 523 | |
| 524 | def test_blame_max_dos(self, tmp_path: pathlib.Path) -> None: |
| 525 | root, _ = _ansi_repo(tmp_path) |
| 526 | self._check( |
| 527 | root, ["code", "blame", "src/module.py::alpha", "--max", "2147483647"] |
| 528 | ) |
| 529 | |
| 530 | |
| 531 | # --------------------------------------------------------------------------- |
| 532 | # § 3 — Output-path traversal (docs_cmd --output) |
| 533 | # --------------------------------------------------------------------------- |
| 534 | |
| 535 | class TestOutputPathTraversal: |
| 536 | """docs_cmd --output must not write files outside the repo root.""" |
| 537 | |
| 538 | def test_absolute_path_rejected(self, tmp_path: pathlib.Path) -> None: |
| 539 | """An absolute --output path that escapes the repo must fail.""" |
| 540 | root, _ = _ansi_repo(tmp_path) |
| 541 | outside = str(tmp_path.parent / "escaped_output.txt") |
| 542 | r = runner.invoke( |
| 543 | cli, |
| 544 | ["code", "docs", "generate", "--output", outside], |
| 545 | env=_env(root), |
| 546 | ) |
| 547 | # Either the command rejects the path (exit_code != 0) or the file |
| 548 | # was never written outside the repo root. |
| 549 | if r.exit_code == 0: |
| 550 | assert not pathlib.Path(outside).exists(), ( |
| 551 | "docs --output wrote a file outside the repo root — " |
| 552 | "validate_output_path guard is missing" |
| 553 | ) |
| 554 | |
| 555 | def test_dotdot_traversal_rejected(self, tmp_path: pathlib.Path) -> None: |
| 556 | """../escape.txt must not land outside the repo root.""" |
| 557 | root, _ = _ansi_repo(tmp_path) |
| 558 | r = runner.invoke( |
| 559 | cli, |
| 560 | ["code", "docs", "generate", "--output", "../../escape.txt"], |
| 561 | env=_env(root), |
| 562 | ) |
| 563 | escaped = (root / "../../escape.txt").resolve() |
| 564 | if r.exit_code == 0: |
| 565 | assert not escaped.exists() or str(escaped).startswith(str(root.resolve())), ( |
| 566 | "docs --output allowed ../ traversal out of repo root" |
| 567 | ) |
| 568 | |
| 569 | def test_safe_relative_path_allowed(self, tmp_path: pathlib.Path) -> None: |
| 570 | """A relative path inside the repo should succeed (or exit cleanly).""" |
| 571 | root, _ = _ansi_repo(tmp_path) |
| 572 | r = runner.invoke( |
| 573 | cli, |
| 574 | ["code", "docs", "generate", "--output", "out/docs.md"], |
| 575 | env=_env(root), |
| 576 | ) |
| 577 | # We don't assert exit_code here — the command may legitimately fail |
| 578 | # (e.g. no doc-ci config), but it must NOT write to a path outside root. |
| 579 | out_file = root / "out" / "docs.md" |
| 580 | if out_file.exists(): |
| 581 | assert str(out_file.resolve()).startswith(str(root.resolve())) |
| 582 | |
| 583 | |
| 584 | # --------------------------------------------------------------------------- |
| 585 | # § 4 — Sanitize_display unit contract |
| 586 | # |
| 587 | # Verify that the sanitize_display primitive itself correctly handles the |
| 588 | # OSC-52 payload used in §1 so that when commands adopt it, the guarantee |
| 589 | # is sound. |
| 590 | # --------------------------------------------------------------------------- |
| 591 | |
| 592 | class TestSanitizeDisplayContract: |
| 593 | """sanitize_display must strip ESC (0x1B) and BEL (0x07) unconditionally.""" |
| 594 | |
| 595 | def test_osc52_stripped(self) -> None: |
| 596 | from muse.core.validation import sanitize_display |
| 597 | result = sanitize_display(_ANSI_PAYLOAD) |
| 598 | assert _ANSI_MARKER not in result |
| 599 | assert "\x07" not in result |
| 600 | |
| 601 | def test_csi_color_stripped(self) -> None: |
| 602 | from muse.core.validation import sanitize_display |
| 603 | assert _ANSI_MARKER not in sanitize_display("\x1b[31mRED\x1b[0m") |
| 604 | |
| 605 | def test_plain_text_preserved(self) -> None: |
| 606 | from muse.core.validation import sanitize_display |
| 607 | text = "hello world 123 αβγ" |
| 608 | assert sanitize_display(text) == text |
| 609 | |
| 610 | def test_newline_and_tab_preserved(self) -> None: |
| 611 | from muse.core.validation import sanitize_display |
| 612 | text = "line1\n\tindented\n" |
| 613 | assert sanitize_display(text) == text |
| 614 | |
| 615 | def test_null_byte_stripped(self) -> None: |
| 616 | from muse.core.validation import sanitize_display |
| 617 | assert "\x00" not in sanitize_display("null\x00byte") |
| 618 | |
| 619 | def test_bel_stripped(self) -> None: |
| 620 | from muse.core.validation import sanitize_display |
| 621 | assert "\x07" not in sanitize_display("ring\x07bell") |
| 622 | |
| 623 | def test_hyperlink_osc8_stripped(self) -> None: |
| 624 | """OSC 8 hyperlink injection must be neutralised.""" |
| 625 | from muse.core.validation import sanitize_display |
| 626 | payload = "\x1b]8;;https://evil.example\x07click\x1b]8;;\x07" |
| 627 | result = sanitize_display(payload) |
| 628 | assert _ANSI_MARKER not in result |
| 629 | assert "\x07" not in result |
| 630 | |
| 631 | |
| 632 | # --------------------------------------------------------------------------- |
| 633 | # § 5 — clamp_int / clamp_natural unit contract |
| 634 | # --------------------------------------------------------------------------- |
| 635 | |
| 636 | class TestClampNatural: |
| 637 | """clamp_natural must accept [0, max_val] and reject anything outside.""" |
| 638 | |
| 639 | def test_value_in_range(self) -> None: |
| 640 | from muse.core.validation import clamp_natural |
| 641 | assert clamp_natural(50, 100) == 50 |
| 642 | |
| 643 | def test_zero_allowed(self) -> None: |
| 644 | from muse.core.validation import clamp_natural |
| 645 | assert clamp_natural(0, 100) == 0 |
| 646 | |
| 647 | def test_max_val_allowed(self) -> None: |
| 648 | from muse.core.validation import clamp_natural |
| 649 | assert clamp_natural(100, 100) == 100 |
| 650 | |
| 651 | def test_negative_rejected(self) -> None: |
| 652 | from muse.core.validation import clamp_natural |
| 653 | with pytest.raises(ValueError, match="value"): |
| 654 | clamp_natural(-1, 100) |
| 655 | |
| 656 | def test_above_max_rejected(self) -> None: |
| 657 | from muse.core.validation import clamp_natural |
| 658 | with pytest.raises(ValueError): |
| 659 | clamp_natural(101, 100) |
| 660 | |
| 661 | def test_maxint_rejected(self) -> None: |
| 662 | from muse.core.validation import clamp_natural |
| 663 | with pytest.raises(ValueError): |
| 664 | clamp_natural(2_147_483_647, 10_000) |
| 665 | |
| 666 | |
| 667 | # --------------------------------------------------------------------------- |
| 668 | # § 6 — validate_output_path unit contract |
| 669 | # --------------------------------------------------------------------------- |
| 670 | |
| 671 | class TestValidateOutputPath: |
| 672 | """validate_output_path must confine the resolved path to the repo root.""" |
| 673 | |
| 674 | def test_relative_path_inside_root(self, tmp_path: pathlib.Path) -> None: |
| 675 | from muse.core.validation import validate_output_path |
| 676 | result = validate_output_path("out/report.md", tmp_path) |
| 677 | assert str(result).startswith(str(tmp_path.resolve())) |
| 678 | |
| 679 | def test_dotdot_rejected(self, tmp_path: pathlib.Path) -> None: |
| 680 | from muse.core.validation import validate_output_path |
| 681 | with pytest.raises(ValueError, match="traversal"): |
| 682 | validate_output_path("../../etc/passwd", tmp_path) |
| 683 | |
| 684 | def test_absolute_outside_root_rejected(self, tmp_path: pathlib.Path) -> None: |
| 685 | from muse.core.validation import validate_output_path |
| 686 | with pytest.raises(ValueError, match="traversal"): |
| 687 | validate_output_path("/etc/cron.d/evil", tmp_path) |
| 688 | |
| 689 | def test_nested_relative_allowed(self, tmp_path: pathlib.Path) -> None: |
| 690 | from muse.core.validation import validate_output_path |
| 691 | result = validate_output_path("a/b/c/report.txt", tmp_path) |
| 692 | assert "a/b/c/report.txt" in str(result) |
File History
3 commits
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9
docs: expand cache plan with all seven testing tiers and do…
Sonnet 4.6
132 days ago
sha256:7f9e2ef5286aedad9c1e6011b4c46ca27f39dbdad6e3409357e36b26e46b3b7c
docs: docstring sprint for-each-ref→hotspots — idiomatic ru…
Sonnet 4.6
patch
138 days ago
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9
fix(cursorignore): remove git-ism (.git/worktrees)
Human
minor
⚠
141 days ago