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