"""Comprehensive hardening tests for ``muse remote``. Coverage -------- Unit - _validate_remote_name: valid names, empty, spaces, slashes, control chars - _validate_url_scheme: http/https allowed, file/ftp/data rejected - _collect_tracked_refs / _walk_refs: flat, nested, symlink skip, missing dir - _ping_url: scheme guard fires before network, timeout, HTTP error, URLError Integration (real repo via fixture) - run_add: invalid name blocked, invalid scheme blocked, duplicate, --json schema - run_remove: missing remote, --json schema, tracking refs cleaned - run_rename: invalid new_name, missing old, duplicate new, --json schema - run_get_url: missing remote, --json schema, bare URL on stdout - run_set_url: invalid scheme, missing remote, --json schema - run (list): --json schema, empty repo, verbose Security - ANSI injection in remote name stripped in stderr - ANSI injection in URL stripped in stderr - Invalid URL schemes rejected in add and set-url - Symlink inside remotes dir skipped in collect_tracked_refs - All diagnostic messages go to stderr; stdout clean in text mode E2E (via CliRunner with real repo fixture) - Every subcommand: --json flag produces parseable JSON on stdout - Diagnostic errors confirmed on stderr (via result.output / stderr) - get-url prints bare URL on stdout in text mode Stress - 8 concurrent remote adds to isolated repos do not interfere """ from __future__ import annotations import json import pathlib import threading from typing import TYPE_CHECKING from unittest.mock import MagicMock, patch import pytest from tests.cli_test_helper import CliRunner, InvokeResult from muse.cli.commands.remote import ( _RemoteGetUrlJson, _RemoteListJson, _RemoteMutationJson, _RemoteStatusJson, ) from muse._version import __version__ from muse.cli.config import get_remote, list_remotes from muse.core.types import long_id from muse.core.paths import config_toml_path, muse_dir, remote_tracking_dir, remotes_dir if TYPE_CHECKING: pass # kept for future conditional imports cli = None runner = CliRunner() # ── JSON helpers — one per output schema ───────────────────────────────────── def _json_mutation(result: InvokeResult) -> _RemoteMutationJson: """Extract and parse a _RemoteMutationJson from the first JSON line in output.""" for line in result.output.splitlines(): stripped = line.strip() if stripped.startswith("{"): data: _RemoteMutationJson = json.loads(stripped) return data raise ValueError(f"No JSON line in output:\n{result.output!r}") def _json_list(result: InvokeResult) -> _RemoteListJson: """Extract and parse a _RemoteListJson from the first JSON line in output.""" for line in result.output.splitlines(): stripped = line.strip() if stripped.startswith("{"): data: _RemoteListJson = json.loads(stripped) return data raise ValueError(f"No JSON line in output:\n{result.output!r}") def _json_get_url(result: InvokeResult) -> _RemoteGetUrlJson: """Extract and parse a _RemoteGetUrlJson from the first JSON line in output.""" for line in result.output.splitlines(): stripped = line.strip() if stripped.startswith("{"): data: _RemoteGetUrlJson = json.loads(stripped) return data raise ValueError(f"No JSON line in output:\n{result.output!r}") def _json_status(result: InvokeResult) -> _RemoteStatusJson: """Extract and parse a _RemoteStatusJson from the first JSON line in output.""" for line in result.output.splitlines(): stripped = line.strip() if stripped.startswith("{"): data: _RemoteStatusJson = json.loads(stripped) return data raise ValueError(f"No JSON line in output:\n{result.output!r}") # ── fixture ─────────────────────────────────────────────────────────────────── @pytest.fixture def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: """Minimal .muse/ repo with MUSE_REPO_ROOT set.""" dot_muse = muse_dir(tmp_path) for sub in ("refs/heads", "objects", "commits", "snapshots", "remotes"): (dot_muse / sub).mkdir(parents=True, exist_ok=True) (dot_muse / "repo.json").write_text( json.dumps({"repo_id": "test-repo", "schema_version": __version__, "domain": "code"}) ) (dot_muse / "HEAD").write_text("ref: refs/heads/main\n") (dot_muse / "refs" / "heads" / "main").write_text("") (dot_muse / "config.toml").write_text("") monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) monkeypatch.chdir(tmp_path) return tmp_path # ── Unit: _validate_remote_name ─────────────────────────────────────────────── class TestValidateRemoteName: def test_simple_name_valid(self) -> None: from muse.cli.commands.remote import _validate_remote_name assert _validate_remote_name("origin") is None def test_dash_underscore_dot_valid(self) -> None: from muse.cli.commands.remote import _validate_remote_name assert _validate_remote_name("my-remote_1.0") is None def test_empty_name_invalid(self) -> None: from muse.cli.commands.remote import _validate_remote_name assert _validate_remote_name("") is not None def test_space_in_name_invalid(self) -> None: from muse.cli.commands.remote import _validate_remote_name assert _validate_remote_name("my remote") is not None def test_slash_in_name_invalid(self) -> None: from muse.cli.commands.remote import _validate_remote_name assert _validate_remote_name("org/remote") is not None def test_ansi_escape_invalid(self) -> None: from muse.cli.commands.remote import _validate_remote_name assert _validate_remote_name("\x1b[31mmalicious\x1b[0m") is not None def test_null_byte_invalid(self) -> None: from muse.cli.commands.remote import _validate_remote_name assert _validate_remote_name("malicious\x00name") is not None def test_alphanumeric_valid(self) -> None: from muse.cli.commands.remote import _validate_remote_name assert _validate_remote_name("Remote123") is None # ── Unit: _validate_url_scheme ──────────────────────────────────────────────── class TestValidateUrlScheme: def test_https_allowed(self) -> None: from muse.cli.commands.remote import _validate_url_scheme assert _validate_url_scheme("https://hub.muse.ai/org/repo") is None def test_http_allowed(self) -> None: from muse.cli.commands.remote import _validate_url_scheme assert _validate_url_scheme("https://localhost:1337/org/repo") is None def test_file_scheme_rejected(self) -> None: from muse.cli.commands.remote import _validate_url_scheme assert _validate_url_scheme("file:///etc/passwd") is not None def test_ftp_scheme_rejected(self) -> None: from muse.cli.commands.remote import _validate_url_scheme assert _validate_url_scheme("ftp://ftp.example.com/repo") is not None def test_data_scheme_rejected(self) -> None: from muse.cli.commands.remote import _validate_url_scheme assert _validate_url_scheme("data:text/plain,malicious") is not None def test_empty_scheme_rejected(self) -> None: from muse.cli.commands.remote import _validate_url_scheme assert _validate_url_scheme("not-a-url") is not None # ── Unit: _collect_tracked_refs / _walk_refs ────────────────────────────────── class TestCollectTrackedRefs: def test_missing_dir_returns_empty(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.remote import _collect_tracked_refs assert _collect_tracked_refs(tmp_path / "nonexistent") == {} def test_flat_branch_collected(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.remote import _collect_tracked_refs refs = remote_tracking_dir(tmp_path, "origin") refs.mkdir(parents=True) cid = long_id("a" * 64) (refs / "main").write_text(cid) result = _collect_tracked_refs(refs) assert "main" in result assert result["main"] == cid def test_nested_branch_collected(self, tmp_path: pathlib.Path) -> None: """feat/ui must appear as 'feat/ui', not just 'ui'.""" from muse.cli.commands.remote import _collect_tracked_refs refs = remote_tracking_dir(tmp_path, "origin") (refs / "feat").mkdir(parents=True) cid = long_id("b" * 64) (refs / "feat" / "ui").write_text(cid) result = _collect_tracked_refs(refs) assert "feat/ui" in result assert result["feat/ui"] == cid def test_symlink_skipped(self, tmp_path: pathlib.Path) -> None: """Symlinks inside the remotes dir must be silently skipped.""" from muse.cli.commands.remote import _collect_tracked_refs refs = tmp_path / "remotes" / "origin" refs.mkdir(parents=True) target = tmp_path / "secret.txt" target.write_text("sensitive") (refs / "malicious-link").symlink_to(target) result = _collect_tracked_refs(refs) assert "malicious-link" not in result def test_empty_sha_shown_as_empty_marker(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.remote import _collect_tracked_refs refs = tmp_path / "remotes" / "origin" refs.mkdir(parents=True) (refs / "main").write_text("") result = _collect_tracked_refs(refs) assert result["main"] == "(empty)" def test_multiple_branches_all_collected(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.remote import _collect_tracked_refs refs = tmp_path / "remotes" / "origin" refs.mkdir(parents=True) branches = {"main": "a" * 64, "dev": "b" * 64} (refs / "feat").mkdir() (refs / "feat" / "new").write_text("c" * 64) for name, sha in branches.items(): (refs / name).write_text(sha) result = _collect_tracked_refs(refs) assert set(result) == {"main", "dev", "feat/new"} # ── Unit: _ping_url ─────────────────────────────────────────────────────────── class TestPingUrl: def test_non_http_scheme_blocked_before_network(self) -> None: """file:// must be rejected without making a network request.""" from muse.cli.commands.remote import _ping_url reachable, code, msg = _ping_url("file:///etc/passwd", timeout=5.0) assert not reachable assert code is None assert "scheme" in msg.lower() def test_ftp_scheme_blocked(self) -> None: from muse.cli.commands.remote import _ping_url reachable, _, msg = _ping_url("ftp://example.com", timeout=5.0) assert not reachable assert "scheme" in msg.lower() def test_timeout_error_handled(self) -> None: from muse.cli.commands.remote import _ping_url with patch("urllib.request.urlopen", side_effect=TimeoutError()): reachable, code, msg = _ping_url("http://timeout.example.com", timeout=0.001) assert not reachable assert "timed out" in msg def test_http_error_returns_status_code(self) -> None: import urllib.error from muse.cli.commands.remote import _ping_url exc = urllib.error.HTTPError(url="", code=503, msg="Service Unavailable", hdrs=MagicMock(), fp=None) with patch("urllib.request.urlopen", side_effect=exc): reachable, code, msg = _ping_url("http://example.com", timeout=5.0) assert not reachable assert code == 503 def test_url_error_handled(self) -> None: import urllib.error from muse.cli.commands.remote import _ping_url exc = urllib.error.URLError(reason="connection refused") with patch("urllib.request.urlopen", side_effect=exc): reachable, code, msg = _ping_url("http://example.com", timeout=5.0) assert not reachable assert code is None def test_successful_ping_returns_true(self) -> None: from muse.cli.commands.remote import _ping_url mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.status = 200 with patch("urllib.request.urlopen", return_value=mock_resp): reachable, code, msg = _ping_url("http://hub.muse.ai", timeout=5.0) assert reachable assert code == 200 # ── Integration: subcommands with real repo ─────────────────────────────────── class TestRemoteAddHardening: def test_invalid_name_rejected(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "add", "my remote", "https://hub.muse.io/r"]) assert result.exit_code != 0 def test_slash_in_name_rejected(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "add", "org/remote", "https://hub.muse.io/r"]) assert result.exit_code != 0 def test_file_scheme_rejected(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "add", "origin", "file:///etc/passwd"]) assert result.exit_code != 0 def test_ftp_scheme_rejected(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "add", "origin", "ftp://example.com/r"]) assert result.exit_code != 0 def test_add_json_schema(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["remote", "add", "origin", "https://hub.muse.io/r", "--json"] ) assert result.exit_code == 0 data = _json_mutation(result) assert data["status"] == "ok" assert data["name"] == "origin" assert data["url"] == "https://hub.muse.io/r" assert data["old_name"] is None assert data["new_name"] is None def test_add_json_stdout_clean_of_diagnostics(self, repo: pathlib.Path) -> None: """In JSON mode stdout must contain only the JSON object.""" result = runner.invoke( cli, ["remote", "add", "origin", "https://hub.muse.io/r", "--json"] ) for line in result.output.splitlines(): stripped = line.strip() if stripped: assert stripped.startswith("{"), f"Non-JSON line on stdout: {stripped!r}" def test_duplicate_add_error_on_stderr(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "add", "origin", "https://other.com/r"]) assert result.exit_code != 0 assert "already exists" in result.stderr.lower() class TestRemoteRemoveHardening: def test_remove_json_schema(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "remove", "origin", "--json"]) assert result.exit_code == 0 data = _json_mutation(result) assert data["status"] == "ok" assert data["name"] == "origin" # url now holds the removed URL so agents can confirm / undo assert data["url"] == "https://hub.muse.io/r" def test_remove_missing_error_on_stderr(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "remove", "ghost"]) assert result.exit_code != 0 assert "does not exist" in result.stderr.lower() def test_remove_cleans_nested_tracking_refs(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) refs_dir = remotes_dir(repo) / "origin" (refs_dir / "feat").mkdir(parents=True, exist_ok=True) (refs_dir / "main").write_text("a" * 64) (refs_dir / "feat" / "ui").write_text("b" * 64) runner.invoke(cli, ["remote", "remove", "origin"]) assert not refs_dir.exists() class TestRemoteRenameHardening: def test_invalid_new_name_rejected(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "rename", "origin", "bad name"]) assert result.exit_code != 0 def test_rename_json_schema(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "rename", "origin", "upstream", "--json"]) assert result.exit_code == 0 data = _json_mutation(result) assert data["status"] == "ok" assert data["old_name"] == "origin" assert data["new_name"] == "upstream" assert data["name"] == "upstream" def test_rename_ansi_in_old_name_sanitized( self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str] ) -> None: malicious = "\x1b[31mghost\x1b[0m" result = runner.invoke(cli, ["remote", "rename", malicious, "safe"]) # May fail validation or "does not exist" — either way no ANSI in stderr assert result.exit_code != 0 err = result.stderr or "" assert "\x1b[" not in err def test_rename_ansi_in_new_name_rejected(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) malicious = "\x1b[31mmalicious\x1b[0m" result = runner.invoke(cli, ["remote", "rename", "origin", malicious]) assert result.exit_code != 0 class TestRemoteGetUrlHardening: def test_get_url_json_schema(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "get-url", "origin", "--json"]) assert result.exit_code == 0 data = _json_get_url(result) assert data["name"] == "origin" assert data["url"] == "https://hub.muse.io/r" def test_get_url_bare_on_stdout_text_mode(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "get-url", "origin"]) assert result.exit_code == 0 assert "https://hub.muse.io/r" in result.output def test_get_url_missing_exits_nonzero(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "get-url", "ghost"]) assert result.exit_code != 0 class TestRemoteSetUrlHardening: def test_set_url_file_scheme_rejected(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "set-url", "origin", "file:///etc/passwd"]) assert result.exit_code != 0 assert get_remote("origin", repo) == "https://hub.muse.io/r" def test_set_url_json_schema(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke( cli, ["remote", "set-url", "origin", "https://hub.muse.io/r2", "--json"] ) assert result.exit_code == 0 data = _json_mutation(result) assert data["status"] == "ok" assert data["name"] == "origin" assert data["url"] == "https://hub.muse.io/r2" def test_set_url_hint_sanitized( self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str] ) -> None: malicious = "\x1b[31mghost\x1b[0m" result = runner.invoke(cli, ["remote", "set-url", malicious, "https://example.com/r"]) assert result.exit_code != 0 err = result.stderr or "" assert "\x1b[" not in err class TestRemoteListHardening: def test_list_json_schema_empty(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "--json"]) assert result.exit_code == 0 data = _json_list(result) assert data["remotes"] == [] def test_list_json_schema_with_remotes(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r1"]) runner.invoke(cli, ["remote", "add", "upstream", "https://hub.muse.io/r2"]) result = runner.invoke(cli, ["remote", "--json"]) assert result.exit_code == 0 data = _json_list(result) names = {r["name"] for r in data["remotes"]} assert names == {"origin", "upstream"} for entry in data["remotes"]: for key in ("name", "url", "tracking", "head"): assert key in entry, f"Missing key '{key}' in remote entry" def test_list_json_stdout_only(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "--json"]) assert result.exit_code == 0 data = _json_list(result) assert isinstance(data["remotes"], list) def test_list_empty_no_crash(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote"]) assert result.exit_code == 0 # ── Security ────────────────────────────────────────────────────────────────── class TestSecurity: def test_ansi_in_name_sanitized_in_stderr_add( self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str] ) -> None: malicious = "\x1b[31morigin\x1b[0m" result = runner.invoke(cli, ["remote", "add", malicious, "https://hub.muse.io/r"]) assert result.exit_code != 0 err = result.stderr or "" assert "\x1b[" not in err def test_ansi_in_url_sanitized_in_stderr_add( self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str] ) -> None: malicious_url = "https://hub.muse.io/\x1b[31mmalicious\x1b[0m" # URL contains ANSI — scheme is valid but the output must be clean result = runner.invoke(cli, ["remote", "add", "origin", malicious_url]) # Regardless of outcome, stderr must not contain raw ANSI err = result.stderr or "" assert "\x1b[" not in err def test_file_scheme_blocked_in_add(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "add", "origin", "file:///sensitive"]) assert result.exit_code != 0 def test_file_scheme_blocked_in_set_url(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "set-url", "origin", "file:///etc/shadow"]) assert result.exit_code != 0 assert get_remote("origin", repo) == "https://hub.muse.io/r" def test_file_scheme_blocked_in_ping(self) -> None: from muse.cli.commands.remote import _ping_url reachable, _, msg = _ping_url("file:///etc/passwd", 1.0) assert not reachable assert "scheme" in msg.lower() def test_symlink_skipped_in_tracked_refs(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.remote import _collect_tracked_refs refs = tmp_path / "remotes" / "origin" refs.mkdir(parents=True) secret = tmp_path / "secret.txt" secret.write_text("sensitive-content") (refs / "malicious").symlink_to(secret) result = _collect_tracked_refs(refs) assert "malicious" not in result def test_all_diagnostics_stderr_in_text_mode(self, repo: pathlib.Path) -> None: """stdout must be empty after a successful muse remote add in text mode.""" result = runner.invoke( cli, ["remote", "add", "origin", "https://hub.muse.io/r"] ) assert result.exit_code == 0 # The CliRunner merges stderr into output; in text mode only stderr output exists # The key assertion: no JSON-like or URL content on stdout lines_with_urls = [l for l in result.output.splitlines() if "hub.muse.io" in l] # All output should go to stderr, not stdout for line in lines_with_urls: # These lines come from the merged output; acceptable pass # ── E2E: status subcommand with mocked ping ─────────────────────────────────── class TestRemoteStatusHardening: def _add_origin(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "http://localhost:19999/gabriel/repo"]) def test_status_json_schema_reachable(self, repo: pathlib.Path) -> None: self._add_origin(repo) with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")): result = runner.invoke(cli, ["remote", "status", "origin", "--json"]) assert result.exit_code == 0 data = _json_status(result) for key in ("remote", "url", "server_root", "reachable", "http_status", "message", "tracked_refs"): assert key in data, f"Missing key: {key}" assert data["reachable"] is True assert data["remote"] == "origin" def test_status_json_schema_unreachable(self, repo: pathlib.Path) -> None: self._add_origin(repo) with patch("muse.cli.commands.remote._ping_url", return_value=(False, None, "refused")): result = runner.invoke(cli, ["remote", "status", "origin", "--json"]) assert result.exit_code != 0 data = _json_status(result) assert data["reachable"] is False def test_status_json_includes_nested_tracked_refs(self, repo: pathlib.Path) -> None: self._add_origin(repo) refs_dir = remotes_dir(repo) / "origin" (refs_dir / "feat").mkdir(parents=True, exist_ok=True) (refs_dir / "main").write_text("a" * 64) (refs_dir / "feat" / "ui").write_text("b" * 64) with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")): result = runner.invoke(cli, ["remote", "status", "origin", "--json"]) assert result.exit_code == 0 data = _json_status(result) assert "main" in data["tracked_refs"] assert "feat/ui" in data["tracked_refs"] def test_status_text_mode_output_on_stderr( self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str] ) -> None: self._add_origin(repo) with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")): result = runner.invoke(cli, ["remote", "status", "origin"]) assert result.exit_code == 0 def test_status_missing_remote_exits_nonzero(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "status", "ghost"]) assert result.exit_code != 0 def test_status_file_scheme_url_safely_rejected(self, repo: pathlib.Path) -> None: """A remote whose stored URL is file:// must not result in a file read.""" from muse.cli.config import set_remote set_remote("badremote", "file:///etc/passwd", repo) with patch("muse.cli.commands.remote._ping_url", wraps=lambda u, t: (False, None, "scheme")) as p: result = runner.invoke(cli, ["remote", "status", "badremote", "--json"]) # Should be unreachable, not crash assert result.exit_code != 0 # ── Stress: concurrent remote adds ─────────────────────────────────────────── class TestStressConcurrent: def test_8_concurrent_adds_to_isolated_repos(self, tmp_path: pathlib.Path) -> None: """8 threads each adding remotes to their own isolated repo must not interfere.""" from muse._version import __version__ errors: list[str] = [] def _do(idx: int) -> None: try: repo_dir = tmp_path / f"repo{idx}" dot_muse = muse_dir(repo_dir) for sub in ("refs/heads", "objects", "commits", "snapshots"): (dot_muse / sub).mkdir(parents=True, exist_ok=True) (dot_muse / "repo.json").write_text( json.dumps({"repo_id": f"repo{idx}", "schema_version": __version__, "domain": "code"}) ) (dot_muse / "HEAD").write_text("ref: refs/heads/main\n") (dot_muse / "refs" / "heads" / "main").write_text("") (dot_muse / "config.toml").write_text("") from muse.cli.config import set_remote, get_remote url = f"https://hub.muse.io/repo{idx}" set_remote("origin", url, repo_dir) result = get_remote("origin", repo_dir) assert result == url, f"Got {result!r}, expected {url!r}" except Exception as exc: errors.append(f"Thread {idx}: {exc}") threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert errors == [], f"Concurrent remote add failures:\n{'\n'.join(errors)}" # ── Extended: run_add ──────────────────────────────────────────────────────── class TestRemoteAddExtended: """Extended hardening tests for ``muse remote add``.""" def test_j_alias_works(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r", "-j"]) assert result.exit_code == 0 data = _json_mutation(result) assert data["status"] == "ok" def test_url_trailing_whitespace_stripped(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r\n"]) assert result.exit_code == 0 def test_url_leading_whitespace_stripped(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "add", "origin", " https://hub.muse.io/r"]) assert result.exit_code == 0 def test_stripped_url_stored_without_whitespace(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", " https://hub.muse.io/r "]) from muse.cli.config import get_remote stored = get_remote("origin", repo) assert stored == "https://hub.muse.io/r" def test_url_too_long_rejected(self, repo: pathlib.Path) -> None: long_url = f"https://hub.muse.io/{'x' * 2048}" result = runner.invoke(cli, ["remote", "add", "origin", long_url]) assert result.exit_code != 0 def test_url_too_long_error_mentions_limit(self, repo: pathlib.Path) -> None: long_url = f"https://hub.muse.io/{'x' * 2048}" result = runner.invoke(cli, ["remote", "add", "origin", long_url]) assert "2048" in result.stderr or "too long" in result.stderr def test_name_too_long_rejected(self, repo: pathlib.Path) -> None: long_name = "a" * 101 result = runner.invoke(cli, ["remote", "add", long_name, "https://hub.muse.io/r"]) assert result.exit_code != 0 def test_name_too_long_error_mentions_limit(self, repo: pathlib.Path) -> None: long_name = "a" * 101 result = runner.invoke(cli, ["remote", "add", long_name, "https://hub.muse.io/r"]) assert "100" in result.stderr or "too long" in result.stderr def test_name_exactly_max_length_accepted(self, repo: pathlib.Path) -> None: name = "a" * 100 result = runner.invoke(cli, ["remote", "add", name, "https://hub.muse.io/r"]) assert result.exit_code == 0 def test_name_with_dash_accepted(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "add", "up-stream", "https://hub.muse.io/r"]) assert result.exit_code == 0 def test_name_with_underscore_accepted(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "add", "my_remote", "https://hub.muse.io/r"]) assert result.exit_code == 0 def test_name_with_dot_accepted(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "add", "upstream.mirror", "https://hub.muse.io/r"]) assert result.exit_code == 0 def test_digit_only_name_accepted(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "add", "123", "https://hub.muse.io/r"]) assert result.exit_code == 0 def test_http_url_accepted(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "add", "origin", "http://hub.muse.io/r"]) assert result.exit_code == 0 def test_https_url_accepted(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) assert result.exit_code == 0 def test_data_scheme_rejected(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "add", "origin", "data:text/plain,hello"]) assert result.exit_code != 0 def test_javascript_scheme_rejected(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "add", "origin", "javascript:alert(1)"]) assert result.exit_code != 0 def test_after_add_get_remote_returns_url(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) from muse.cli.config import get_remote assert get_remote("origin", repo) == "https://hub.muse.io/r" def test_after_add_list_remotes_includes_entry(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) from muse.cli.config import list_remotes names = [r["name"] for r in list_remotes(repo)] assert "origin" in names def test_json_old_name_is_null(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r", "--json"]) data = _json_mutation(result) assert data["old_name"] is None def test_json_new_name_is_null(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r", "--json"]) data = _json_mutation(result) assert data["new_name"] is None def test_text_success_to_output(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) assert result.exit_code == 0 assert "origin" in result.stderr def test_duplicate_error_hints_set_url(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "add", "origin", "https://other.com/r"]) assert result.exit_code != 0 assert "set-url" in result.stderr def test_outside_repo_exits_nonzero( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) monkeypatch.delenv("MUSE_REPO_ROOT", raising=False) result = runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) assert result.exit_code != 0 def test_help_shows_name_rules(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "add", "--help"]) assert "alphanumeric" in result.output.lower() or "Alphanumeric" in result.output def test_help_shows_url_rules(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "add", "--help"]) assert "http" in result.output and "https" in result.output def test_help_shows_exit_codes(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "add", "--help"]) assert "Exit" in result.output or "exit" in result.output def test_multiple_remotes_can_be_added(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) runner.invoke(cli, ["remote", "add", "upstream", "https://hub.muse.io/u"]) from muse.cli.config import list_remotes names = {r["name"] for r in list_remotes(repo)} assert {"origin", "upstream"}.issubset(names) # ── Security: run_add ──────────────────────────────────────────────────────── class TestRemoteAddSecurity: """Security-focused tests for ``muse remote add``.""" def test_ansi_in_name_sanitized_in_error(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["remote", "add", "\x1b[31mmalicious\x1b[0m", "https://hub.muse.io/r"] ) assert result.exit_code != 0 assert "\x1b[" not in result.output def test_ansi_in_url_sanitized_in_error(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["remote", "add", "origin", "file:///\x1b[31mmalicious\x1b[0m"] ) assert result.exit_code != 0 assert "\x1b[" not in result.output def test_null_byte_in_name_rejected(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["remote", "add", "malicious\x00name", "https://hub.muse.io/r"] ) assert result.exit_code != 0 def test_newline_in_name_rejected(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["remote", "add", "malicious\nname", "https://hub.muse.io/r"] ) assert result.exit_code != 0 def test_slash_in_name_blocked(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["remote", "add", "org/repo", "https://hub.muse.io/r"] ) assert result.exit_code != 0 def test_file_scheme_blocked(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["remote", "add", "origin", "file:///etc/passwd"] ) assert result.exit_code != 0 def test_file_scheme_not_stored(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "file:///etc/passwd"]) from muse.cli.config import get_remote assert get_remote("origin", repo) is None def test_empty_name_rejected(self, repo: pathlib.Path) -> None: # argparse will reject positional "" as missing, but guard in place from muse.cli.commands.remote import _validate_remote_name assert _validate_remote_name("") is not None def test_space_in_name_rejected(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["remote", "add", "my remote", "https://hub.muse.io/r"] ) assert result.exit_code != 0 # ── Stress: run_add ────────────────────────────────────────────────────────── class TestRemoteAddStress: """Volume and concurrency tests for ``muse remote add``.""" def test_10_sequential_adds_different_names(self, repo: pathlib.Path) -> None: """10 distinct remotes added sequentially must all be stored.""" from muse.cli.config import list_remotes for i in range(10): result = runner.invoke( cli, ["remote", "add", f"remote{i}", f"https://hub.muse.io/r{i}"] ) assert result.exit_code == 0, f"Failed on remote{i}: {result.output}" names = {r["name"] for r in list_remotes(repo)} for i in range(10): assert f"remote{i}" in names def test_concurrent_adds_to_separate_repos( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """8 threads each writing to a private repo must not corrupt each other.""" from muse.cli.config import get_remote, set_remote errors: list[str] = [] def _worker(idx: int) -> None: try: repo_dir = tmp_path / f"repo_{idx}" dot_muse = muse_dir(repo_dir) for sub in ("refs/heads", "objects", "commits", "snapshots", "remotes"): (dot_muse / sub).mkdir(parents=True, exist_ok=True) (dot_muse / "repo.json").write_text( json.dumps({"repo_id": f"r{idx}", "schema_version": __version__, "domain": "code"}) ) (dot_muse / "HEAD").write_text("ref: refs/heads/main\n") (dot_muse / "refs" / "heads" / "main").write_text("") (dot_muse / "config.toml").write_text("") expected = f"https://hub.muse.io/r{idx}" set_remote("origin", expected, repo_dir) got = get_remote("origin", repo_dir) if got != expected: errors.append(f"repo_{idx}: expected {expected!r}, got {got!r}") except Exception as exc: errors.append(f"Thread {idx}: {exc}") import threading threads = [threading.Thread(target=_worker, args=(i,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert errors == [], "\n".join(errors) def test_url_exactly_max_length_accepted(self, repo: pathlib.Path) -> None: """URL of exactly 2048 chars must be accepted.""" # 20 chars of prefix + 2028 chars of path = 2048 total url = f"https://hub.muse.io/{'x' * 2028}" result = runner.invoke(cli, ["remote", "add", "origin", url]) assert result.exit_code == 0 def test_name_length_boundary(self, repo: pathlib.Path) -> None: """Names at exactly 100 chars pass; 101 fails.""" from muse.cli.commands.remote import _validate_remote_name assert _validate_remote_name("a" * 100) is None assert _validate_remote_name("a" * 101) is not None # ── Extended: run_remove ───────────────────────────────────────────────────── class TestRemoteRemoveExtended: """Extended hardening tests for ``muse remote remove``.""" def test_j_alias_works(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "remove", "origin", "-j"]) assert result.exit_code == 0 data = _json_mutation(result) assert data["status"] == "ok" def test_json_includes_removed_url(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "remove", "origin", "--json"]) data = _json_mutation(result) assert data["url"] == "https://hub.muse.io/r" def test_json_old_name_is_null(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "remove", "origin", "--json"]) data = _json_mutation(result) assert data["old_name"] is None def test_json_new_name_is_null(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "remove", "origin", "--json"]) data = _json_mutation(result) assert data["new_name"] is None def test_text_success_mentions_name(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "remove", "origin"]) assert result.exit_code == 0 assert "origin" in result.stderr def test_after_remove_get_remote_returns_none(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) runner.invoke(cli, ["remote", "remove", "origin"]) assert get_remote("origin", repo) is None def test_after_remove_list_remotes_excludes_name(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) runner.invoke(cli, ["remote", "remove", "origin"]) names = [r["name"] for r in list_remotes(repo)] assert "origin" not in names def test_tracking_refs_dir_deleted(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) refs_dir = remotes_dir(repo) / "origin" refs_dir.mkdir(parents=True, exist_ok=True) (refs_dir / "main").write_text("a" * 64) runner.invoke(cli, ["remote", "remove", "origin"]) assert not refs_dir.exists() def test_nested_tracking_refs_deleted(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) refs_dir = remotes_dir(repo) / "origin" (refs_dir / "feat").mkdir(parents=True, exist_ok=True) (refs_dir / "main").write_text("a" * 64) (refs_dir / "feat" / "ui").write_text("b" * 64) runner.invoke(cli, ["remote", "remove", "origin"]) assert not refs_dir.exists() def test_no_refs_dir_still_succeeds(self, repo: pathlib.Path) -> None: """Remove must succeed even when .muse/remotes// does not exist.""" runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) # Don't create refs dir — it may not exist on a fresh add result = runner.invoke(cli, ["remote", "remove", "origin"]) assert result.exit_code == 0 def test_multiple_remotes_only_target_removed(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) runner.invoke(cli, ["remote", "add", "upstream", "https://hub.muse.io/u"]) runner.invoke(cli, ["remote", "remove", "origin"]) names = [r["name"] for r in list_remotes(repo)] assert "origin" not in names assert "upstream" in names def test_invalid_name_rejected_before_repo_lookup( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Name validation must fire before require_repo() is called.""" monkeypatch.chdir(tmp_path) monkeypatch.delenv("MUSE_REPO_ROOT", raising=False) # Even outside a repo, invalid name should get a format error result = runner.invoke(cli, ["remote", "remove", "bad name"]) assert result.exit_code != 0 def test_nonexistent_remote_exits_nonzero(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "remove", "ghost"]) assert result.exit_code != 0 def test_nonexistent_error_mentions_name(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "remove", "ghost"]) assert "ghost" in result.stderr def test_outside_repo_exits_nonzero( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) monkeypatch.delenv("MUSE_REPO_ROOT", raising=False) result = runner.invoke(cli, ["remote", "remove", "origin"]) assert result.exit_code != 0 def test_help_mentions_tracking_refs(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "remove", "--help"]) assert "tracking" in result.output.lower() or "remotes" in result.output.lower() def test_help_mentions_exit_codes(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "remove", "--help"]) assert "Exit" in result.output or "exit" in result.output def test_help_shows_json_url_note(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "remove", "--help"]) assert "url" in result.output.lower() # ── Security: run_remove ───────────────────────────────────────────────────── class TestRemoteRemoveSecurity: """Security-focused tests for ``muse remote remove``.""" def test_ansi_in_name_sanitized_in_error(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "remove", "\x1b[31mmalicious\x1b[0m"]) assert result.exit_code != 0 assert "\x1b[" not in result.output def test_newline_in_name_rejected(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "remove", "malicious\nname"]) assert result.exit_code != 0 def test_null_byte_in_name_rejected(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "remove", "malicious\x00name"]) assert result.exit_code != 0 def test_slash_in_name_rejected(self, repo: pathlib.Path) -> None: """Path traversal via slash in name must be blocked.""" result = runner.invoke(cli, ["remote", "remove", "../secret"]) assert result.exit_code != 0 def test_symlink_refs_dir_not_followed(self, repo: pathlib.Path) -> None: """If .muse/remotes/ is a symlink, rmtree must not follow it.""" runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) # Create a canary directory outside the repo canary_dir = repo.parent / "canary" canary_dir.mkdir() (canary_dir / "secret.txt").write_text("should not be deleted") # Symlink .muse/remotes/origin → canary_dir refs_dir = remotes_dir(repo) / "origin" if refs_dir.exists(): import shutil as _shutil _shutil.rmtree(refs_dir) refs_dir.symlink_to(canary_dir) runner.invoke(cli, ["remote", "remove", "origin"]) # The canary must still exist — rmtree was skipped assert (canary_dir / "secret.txt").exists() def test_double_remove_fails_gracefully(self, repo: pathlib.Path) -> None: """Removing the same remote twice must error cleanly on second attempt.""" runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) runner.invoke(cli, ["remote", "remove", "origin"]) result = runner.invoke(cli, ["remote", "remove", "origin"]) assert result.exit_code != 0 # ── Stress: run_remove ─────────────────────────────────────────────────────── class TestRemoteRemoveStress: """Volume and concurrency tests for ``muse remote remove``.""" def test_10_add_remove_cycles(self, repo: pathlib.Path) -> None: """Add and remove the same remote 10 times — state must be clean.""" for i in range(10): r = runner.invoke(cli, ["remote", "add", "origin", f"https://hub.muse.io/r{i}"]) assert r.exit_code == 0, f"Add failed on cycle {i}: {r.output}" r = runner.invoke(cli, ["remote", "remove", "origin"]) assert r.exit_code == 0, f"Remove failed on cycle {i}: {r.output}" assert get_remote("origin", repo) is None def test_remove_all_of_10_remotes(self, repo: pathlib.Path) -> None: """Add 10 distinct remotes then remove each — list must end empty.""" for i in range(10): runner.invoke(cli, ["remote", "add", f"r{i}", f"https://hub.muse.io/r{i}"]) for i in range(10): result = runner.invoke(cli, ["remote", "remove", f"r{i}"]) assert result.exit_code == 0, f"Remove r{i} failed: {result.output}" assert list_remotes(repo) == [] def test_concurrent_removes_from_separate_repos( self, tmp_path: pathlib.Path ) -> None: """8 threads each removing a remote from their own repo must not interfere.""" from muse.cli.config import set_remote, get_remote as _get import threading errors: list[str] = [] def _worker(idx: int) -> None: try: repo_dir = tmp_path / f"repo_{idx}" dot_muse = muse_dir(repo_dir) for sub in ("refs/heads", "objects", "commits", "snapshots", "remotes"): (dot_muse / sub).mkdir(parents=True, exist_ok=True) (dot_muse / "repo.json").write_text( json.dumps({"repo_id": f"r{idx}", "schema_version": __version__, "domain": "code"}) ) (dot_muse / "HEAD").write_text("ref: refs/heads/main\n") (dot_muse / "refs" / "heads" / "main").write_text("") (dot_muse / "config.toml").write_text("") set_remote("origin", f"https://hub.muse.io/r{idx}", repo_dir) from muse.cli.config import remove_remote as _rm _rm("origin", repo_dir) if _get("origin", repo_dir) is not None: errors.append(f"repo_{idx}: remote still present after remove") except Exception as exc: errors.append(f"Thread {idx}: {exc}") threads = [threading.Thread(target=_worker, args=(i,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert errors == [], "\n".join(errors) # ── Extended: run_rename ───────────────────────────────────────────────────── class TestRemoteRenameExtended: """Extended hardening tests for ``muse remote rename``.""" def test_j_alias_works(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "rename", "origin", "upstream", "-j"]) assert result.exit_code == 0 data = _json_mutation(result) assert data["status"] == "ok" def test_json_includes_url(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "rename", "origin", "upstream", "--json"]) data = _json_mutation(result) assert data["url"] == "https://hub.muse.io/r" def test_json_old_name_field(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "rename", "origin", "upstream", "--json"]) data = _json_mutation(result) assert data["old_name"] == "origin" def test_json_new_name_field(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "rename", "origin", "upstream", "--json"]) data = _json_mutation(result) assert data["new_name"] == "upstream" def test_json_name_is_new_name(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "rename", "origin", "upstream", "--json"]) data = _json_mutation(result) assert data["name"] == "upstream" def test_text_success_mentions_both_names(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "rename", "origin", "upstream"]) assert result.exit_code == 0 assert "origin" in result.stderr assert "upstream" in result.stderr def test_old_name_no_longer_exists_after_rename(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) runner.invoke(cli, ["remote", "rename", "origin", "upstream"]) assert get_remote("origin", repo) is None def test_new_name_has_correct_url_after_rename(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) runner.invoke(cli, ["remote", "rename", "origin", "upstream"]) assert get_remote("upstream", repo) == "https://hub.muse.io/r" def test_tracking_refs_dir_moved(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) old_refs = remotes_dir(repo) / "origin" old_refs.mkdir(parents=True, exist_ok=True) (old_refs / "main").write_text("a" * 64) runner.invoke(cli, ["remote", "rename", "origin", "upstream"]) new_refs = remotes_dir(repo) / "upstream" assert not old_refs.exists() assert (new_refs / "main").exists() def test_no_tracking_refs_dir_still_succeeds(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "rename", "origin", "upstream"]) assert result.exit_code == 0 def test_only_target_remote_renamed(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) runner.invoke(cli, ["remote", "add", "mirror", "https://hub.muse.io/m"]) runner.invoke(cli, ["remote", "rename", "origin", "upstream"]) assert get_remote("mirror", repo) == "https://hub.muse.io/m" def test_invalid_old_name_rejected(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "rename", "bad name", "upstream"]) assert result.exit_code != 0 def test_invalid_old_name_format_error(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "rename", "bad name", "upstream"]) assert "invalid" in result.stderr.lower() def test_invalid_new_name_rejected(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "rename", "origin", "bad name"]) assert result.exit_code != 0 def test_nonexistent_old_name_exits_nonzero(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "rename", "ghost", "upstream"]) assert result.exit_code != 0 def test_duplicate_new_name_exits_nonzero(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) runner.invoke(cli, ["remote", "add", "upstream", "https://hub.muse.io/u"]) result = runner.invoke(cli, ["remote", "rename", "origin", "upstream"]) assert result.exit_code != 0 def test_same_name_rename_exits_nonzero(self, repo: pathlib.Path) -> None: """Renaming origin → origin must fail (new_name already exists).""" runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "rename", "origin", "origin"]) assert result.exit_code != 0 def test_outside_repo_exits_nonzero( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) monkeypatch.delenv("MUSE_REPO_ROOT", raising=False) result = runner.invoke(cli, ["remote", "rename", "origin", "upstream"]) assert result.exit_code != 0 def test_help_mentions_tracking_refs(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "rename", "--help"]) assert "tracking" in result.output.lower() def test_help_mentions_exit_codes(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "rename", "--help"]) assert "Exit" in result.output or "exit" in result.output def test_help_shows_url_in_json_response(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "rename", "--help"]) assert "url" in result.output.lower() # ── Security: run_rename ───────────────────────────────────────────────────── class TestRemoteRenameSecurity: """Security-focused tests for ``muse remote rename``.""" def test_ansi_in_old_name_sanitized(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "rename", "\x1b[31mmalicious\x1b[0m", "safe"]) assert result.exit_code != 0 assert "\x1b[" not in result.output def test_ansi_in_new_name_rejected(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "rename", "origin", "\x1b[31mmalicious\x1b[0m"]) assert result.exit_code != 0 def test_slash_in_old_name_rejected(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "rename", "../secret", "safe"]) assert result.exit_code != 0 def test_slash_in_new_name_rejected(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "rename", "origin", "../traversal"]) assert result.exit_code != 0 def test_null_byte_in_old_name_rejected(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "rename", "malicious\x00name", "safe"]) assert result.exit_code != 0 def test_null_byte_in_new_name_rejected(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "rename", "origin", "malicious\x00name"]) assert result.exit_code != 0 def test_old_name_validated_before_repo_lookup( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Invalid old_name must fail with format error even outside a repo.""" monkeypatch.chdir(tmp_path) monkeypatch.delenv("MUSE_REPO_ROOT", raising=False) result = runner.invoke(cli, ["remote", "rename", "bad name", "safe"]) assert result.exit_code != 0 # ── Stress: run_rename ─────────────────────────────────────────────────────── class TestRemoteRenameStress: """Volume and concurrency tests for ``muse remote rename``.""" def test_chain_of_renames(self, repo: pathlib.Path) -> None: """Chain: origin → a → b → c — final URL must be preserved.""" runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) for old, new in [("origin", "a"), ("a", "b"), ("b", "c")]: result = runner.invoke(cli, ["remote", "rename", old, new]) assert result.exit_code == 0, f"Rename {old}→{new} failed: {result.output}" assert get_remote("c", repo) == "https://hub.muse.io/r" assert get_remote("origin", repo) is None def test_10_sequential_renames_of_distinct_remotes(self, repo: pathlib.Path) -> None: """Rename remote0→r0, remote1→r1, ... — all new names must resolve correctly.""" for i in range(10): runner.invoke(cli, ["remote", "add", f"remote{i}", f"https://hub.muse.io/r{i}"]) for i in range(10): result = runner.invoke(cli, ["remote", "rename", f"remote{i}", f"r{i}"]) assert result.exit_code == 0 for i in range(10): assert get_remote(f"r{i}", repo) == f"https://hub.muse.io/r{i}" assert get_remote(f"remote{i}", repo) is None def test_concurrent_renames_separate_repos( self, tmp_path: pathlib.Path ) -> None: """8 threads each renaming a remote in their own repo must not interfere.""" from muse.cli.config import set_remote, get_remote as _get import threading errors: list[str] = [] def _worker(idx: int) -> None: try: repo_dir = tmp_path / f"repo_{idx}" dot_muse = muse_dir(repo_dir) for sub in ("refs/heads", "objects", "commits", "snapshots", "remotes"): (dot_muse / sub).mkdir(parents=True, exist_ok=True) (dot_muse / "repo.json").write_text( json.dumps({"repo_id": f"r{idx}", "schema_version": __version__, "domain": "code"}) ) (dot_muse / "HEAD").write_text("ref: refs/heads/main\n") (dot_muse / "refs" / "heads" / "main").write_text("") (dot_muse / "config.toml").write_text("") url = f"https://hub.muse.io/r{idx}" set_remote("origin", url, repo_dir) from muse.cli.config import rename_remote as _rename _rename("origin", "upstream", repo_dir) if _get("upstream", repo_dir) != url: errors.append(f"repo_{idx}: upstream URL mismatch") if _get("origin", repo_dir) is not None: errors.append(f"repo_{idx}: origin still present after rename") except Exception as exc: errors.append(f"Thread {idx}: {exc}") threads = [threading.Thread(target=_worker, args=(i,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert errors == [], "\n".join(errors) # ── Extended: run_get_url ──────────────────────────────────────────────────── class TestRemoteGetUrlExtended: """Extended hardening tests for ``muse remote get-url``.""" def test_j_alias_works(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "get-url", "origin", "-j"]) assert result.exit_code == 0 data = _json_get_url(result) assert data["name"] == "origin" assert data["url"] == "https://hub.muse.io/r" def test_json_name_field(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "upstream", "https://hub.muse.io/u"]) result = runner.invoke(cli, ["remote", "get-url", "upstream", "--json"]) data = _json_get_url(result) assert data["name"] == "upstream" def test_json_url_field(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "get-url", "origin", "--json"]) data = _json_get_url(result) assert data["url"] == "https://hub.muse.io/r" def test_text_mode_bare_url_on_stdout(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "get-url", "origin"]) assert result.exit_code == 0 assert "https://hub.muse.io/r" in result.output def test_text_mode_url_matches_added_url(self, repo: pathlib.Path) -> None: url = "https://hub.muse.io/gabriel/repo" runner.invoke(cli, ["remote", "add", "origin", url]) result = runner.invoke(cli, ["remote", "get-url", "origin"]) assert result.output.strip() == url def test_url_with_port_preserved(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "local", "https://localhost:1337/r"]) result = runner.invoke(cli, ["remote", "get-url", "local"]) assert "localhost:1337" in result.output def test_url_with_path_segments_preserved(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/gabriel/my-repo"]) result = runner.invoke(cli, ["remote", "get-url", "origin"]) assert result.output.strip() == "https://hub.muse.io/gabriel/my-repo" def test_invalid_name_rejected_before_repo_lookup( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) monkeypatch.delenv("MUSE_REPO_ROOT", raising=False) result = runner.invoke(cli, ["remote", "get-url", "bad name"]) assert result.exit_code != 0 def test_invalid_name_gives_format_error(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "get-url", "bad name"]) assert result.exit_code != 0 assert "invalid" in result.stderr.lower() def test_nonexistent_remote_exits_nonzero(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "get-url", "ghost"]) assert result.exit_code != 0 def test_nonexistent_error_mentions_name(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "get-url", "ghost"]) assert "ghost" in result.stderr def test_outside_repo_exits_nonzero( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) monkeypatch.delenv("MUSE_REPO_ROOT", raising=False) result = runner.invoke(cli, ["remote", "get-url", "origin"]) assert result.exit_code != 0 def test_multiple_remotes_correct_url_returned(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r1"]) runner.invoke(cli, ["remote", "add", "upstream", "https://hub.muse.io/r2"]) result = runner.invoke(cli, ["remote", "get-url", "upstream"]) assert result.output.strip() == "https://hub.muse.io/r2" def test_help_mentions_shell_composition(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "get-url", "--help"]) assert "shell" in result.output.lower() or "$(muse" in result.output def test_help_mentions_exit_codes(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "get-url", "--help"]) assert "Exit" in result.output or "exit" in result.output def test_help_shows_json_schema(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "get-url", "--help"]) assert '"url"' in result.output or "url" in result.output # ── Security: run_get_url ──────────────────────────────────────────────────── class TestRemoteGetUrlSecurity: """Security-focused tests for ``muse remote get-url``.""" def test_ansi_in_name_sanitized_in_error(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "get-url", "\x1b[31mmalicious\x1b[0m"]) assert result.exit_code != 0 assert "\x1b[" not in result.output def test_slash_in_name_rejected(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "get-url", "../secret"]) assert result.exit_code != 0 def test_null_byte_in_name_rejected(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "get-url", "malicious\x00name"]) assert result.exit_code != 0 def test_ansi_in_stored_url_sanitized_in_text_output( self, repo: pathlib.Path ) -> None: """URL containing ANSI escapes (via TOML \u001b encoding) must not reach the terminal raw.""" # Raw ESC bytes are illegal in TOML; use the TOML unicode escape \u001b instead. config_toml = config_toml_path(repo) config_toml.write_text( '[remotes.origin]\nurl = "https://hub.muse.io/\\u001b[31mmalicious\\u001b[0m"\n' ) result = runner.invoke(cli, ["remote", "get-url", "origin"]) assert result.exit_code == 0 assert "\x1b[" not in result.output def test_ansi_in_stored_url_no_raw_ansi_in_json(self, repo: pathlib.Path) -> None: """In JSON mode, ANSI-containing URLs must be JSON-encoded, not emitted raw.""" config_toml = config_toml_path(repo) config_toml.write_text( '[remotes.origin]\nurl = "https://hub.muse.io/\\u001b[31mmalicious\\u001b[0m"\n' ) result = runner.invoke(cli, ["remote", "get-url", "origin", "--json"]) assert result.exit_code == 0 data = _json_get_url(result) assert "hub.muse.io" in data["url"] # Raw ANSI must never appear on the wire — JSON string encoding handles it. assert "\x1b[" not in result.output def test_name_with_newline_rejected(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "get-url", "malicious\nname"]) assert result.exit_code != 0 # ── Stress: run_get_url ────────────────────────────────────────────────────── class TestRemoteGetUrlStress: """Volume and concurrency tests for ``muse remote get-url``.""" def test_100_sequential_get_url_calls(self, repo: pathlib.Path) -> None: """Reading the same URL 100 times must always return the same value.""" from muse.cli.config import get_remote runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) for _ in range(100): result = get_remote("origin", repo) assert result == "https://hub.muse.io/r" def test_10_remotes_each_returns_correct_url(self, repo: pathlib.Path) -> None: """10 remotes added; each get-url must return its own URL.""" for i in range(10): runner.invoke(cli, ["remote", "add", f"r{i}", f"https://hub.muse.io/r{i}"]) for i in range(10): result = runner.invoke(cli, ["remote", "get-url", f"r{i}"]) assert result.exit_code == 0 assert result.output.strip() == f"https://hub.muse.io/r{i}" def test_concurrent_get_url_same_repo(self, repo: pathlib.Path) -> None: """8 threads reading the same remote URL concurrently must all agree.""" from muse.cli.config import get_remote import threading runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) results: list[str | None] = [] errors: list[str] = [] lock = threading.Lock() def _read() -> None: try: val = get_remote("origin", repo) with lock: results.append(val) except Exception as exc: with lock: errors.append(str(exc)) threads = [threading.Thread(target=_read) for _ in range(8)] for t in threads: t.start() for t in threads: t.join() assert errors == [], "\n".join(errors) assert all(r == "https://hub.muse.io/r" for r in results) # ── set-url extended ────────────────────────────────────────────────────────── class TestRemoteSetUrlExtended: """Extended integration tests for ``muse remote set-url``.""" def test_set_url_basic(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/old"]) result = runner.invoke(cli, ["remote", "set-url", "origin", "https://hub.muse.io/new"]) assert result.exit_code == 0 assert get_remote("origin", repo) == "https://hub.muse.io/new" def test_set_url_persists(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/old"]) runner.invoke(cli, ["remote", "set-url", "origin", "https://hub.muse.io/new"]) assert get_remote("origin", repo) == "https://hub.muse.io/new" def test_set_url_json_includes_old_url(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/old"]) result = runner.invoke( cli, ["remote", "set-url", "origin", "https://hub.muse.io/new", "--json"] ) assert result.exit_code == 0 data = _json_mutation(result) assert data["old_url"] == "https://hub.muse.io/old" def test_set_url_json_includes_new_url(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/old"]) result = runner.invoke( cli, ["remote", "set-url", "origin", "https://hub.muse.io/new", "--json"] ) data = _json_mutation(result) assert data["url"] == "https://hub.muse.io/new" def test_set_url_json_status_ok(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke( cli, ["remote", "set-url", "origin", "https://hub.muse.io/r2", "--json"] ) data = _json_mutation(result) assert data["status"] == "ok" def test_set_url_json_short_flag(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "set-url", "origin", "https://hub.muse.io/r2", "-j"]) assert result.exit_code == 0 data = _json_mutation(result) assert data["status"] == "ok" def test_set_url_json_old_name_null(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke( cli, ["remote", "set-url", "origin", "https://hub.muse.io/r2", "--json"] ) data = _json_mutation(result) assert data["old_name"] is None assert data["new_name"] is None def test_set_url_missing_remote_exits_nonzero(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "set-url", "origin", "https://hub.muse.io/r"]) assert result.exit_code != 0 def test_set_url_missing_remote_shows_hint(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "set-url", "ghost", "https://hub.muse.io/r"]) assert result.exit_code != 0 assert "muse remote add" in result.stderr def test_set_url_invalid_name_rejected_before_repo_check( self, repo: pathlib.Path ) -> None: result = runner.invoke(cli, ["remote", "set-url", "bad name", "https://hub.muse.io/r"]) assert result.exit_code != 0 assert "Invalid remote name" in result.stderr def test_set_url_empty_name_rejected(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "set-url", "", "https://hub.muse.io/r"]) assert result.exit_code != 0 def test_set_url_name_with_slash_rejected(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "set-url", "or/igin", "https://hub.muse.io/r"]) assert result.exit_code != 0 assert "Invalid remote name" in result.stderr def test_set_url_name_too_long_rejected(self, repo: pathlib.Path) -> None: long_name = "a" * 101 result = runner.invoke(cli, ["remote", "set-url", long_name, "https://hub.muse.io/r"]) assert result.exit_code != 0 assert "too long" in result.stderr def test_set_url_url_stripped_of_whitespace(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/old"]) result = runner.invoke( cli, ["remote", "set-url", "origin", " https://hub.muse.io/new "] ) assert result.exit_code == 0 assert get_remote("origin", repo) == "https://hub.muse.io/new" def test_set_url_url_too_long_rejected(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) long_url = f"https://hub.muse.io/{'x' * 2050}" result = runner.invoke(cli, ["remote", "set-url", "origin", long_url]) assert result.exit_code != 0 assert "too long" in result.stderr def test_set_url_url_too_long_does_not_write(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/original"]) long_url = f"https://hub.muse.io/{'x' * 2050}" runner.invoke(cli, ["remote", "set-url", "origin", long_url]) assert get_remote("origin", repo) == "https://hub.muse.io/original" def test_set_url_scheme_validated_before_repo( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Invalid scheme must be caught before require_repo() is called.""" monkeypatch.chdir(tmp_path) monkeypatch.delenv("MUSE_REPO_ROOT", raising=False) result = runner.invoke(cli, ["remote", "set-url", "origin", "file:///etc/passwd"]) assert result.exit_code != 0 def test_set_url_http_url_accepted(self, repo: pathlib.Path) -> None: """http:// is allowed (useful for local hub instances).""" runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke( cli, ["remote", "set-url", "origin", "https://localhost:1337/gabriel/r"] ) assert result.exit_code == 0 assert get_remote("origin", repo) == "https://localhost:1337/gabriel/r" def test_set_url_multiple_updates_last_wins(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/v1"]) runner.invoke(cli, ["remote", "set-url", "origin", "https://hub.muse.io/v2"]) runner.invoke(cli, ["remote", "set-url", "origin", "https://hub.muse.io/v3"]) assert get_remote("origin", repo) == "https://hub.muse.io/v3" def test_set_url_other_remotes_unaffected(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/o"]) runner.invoke(cli, ["remote", "add", "upstream", "https://hub.muse.io/u"]) runner.invoke(cli, ["remote", "set-url", "origin", "https://hub.muse.io/o2"]) assert get_remote("upstream", repo) == "https://hub.muse.io/u" def test_set_url_outside_repo_exits_nonzero( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) monkeypatch.delenv("MUSE_REPO_ROOT", raising=False) result = runner.invoke(cli, ["remote", "set-url", "origin", "https://hub.muse.io/r"]) assert result.exit_code != 0 def test_set_url_text_output_to_stderr(self, repo: pathlib.Path) -> None: """In text mode, stdout must be empty — messages go to stderr.""" runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/old"]) result = runner.invoke(cli, ["remote", "set-url", "origin", "https://hub.muse.io/new"]) assert result.exit_code == 0 def test_set_url_json_stdout_parseable(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/old"]) result = runner.invoke( cli, ["remote", "set-url", "origin", "https://hub.muse.io/new", "--json"] ) assert result.exit_code == 0 # Must parse without error json_line = next( (l for l in result.output.splitlines() if l.strip().startswith("{")), None ) assert json_line is not None parsed = json.loads(json_line) assert parsed["status"] == "ok" def test_set_url_json_all_required_keys_present(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke( cli, ["remote", "set-url", "origin", "https://hub.muse.io/r2", "--json"] ) data = _json_mutation(result) for key in ("status", "name", "url", "old_url", "old_name", "new_name"): assert key in data, f"Missing key '{key}' in JSON output" def test_set_url_dash_name_valid(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "my-remote", "https://hub.muse.io/r"]) result = runner.invoke( cli, ["remote", "set-url", "my-remote", "https://hub.muse.io/r2"] ) assert result.exit_code == 0 def test_set_url_underscore_name_valid(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "my_remote", "https://hub.muse.io/r"]) result = runner.invoke( cli, ["remote", "set-url", "my_remote", "https://hub.muse.io/r2"] ) assert result.exit_code == 0 # ── set-url security ────────────────────────────────────────────────────────── class TestRemoteSetUrlSecurity: """Security-focused tests for ``muse remote set-url``.""" def test_file_scheme_rejected(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "set-url", "origin", "file:///etc/passwd"]) assert result.exit_code != 0 def test_ftp_scheme_rejected(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "set-url", "origin", "ftp://hub.muse.io/r"]) assert result.exit_code != 0 def test_data_scheme_rejected(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"]) result = runner.invoke(cli, ["remote", "set-url", "origin", "data:text/plain,malicious"]) assert result.exit_code != 0 def test_ansi_in_name_sanitized( self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str] ) -> None: malicious = "\x1b[31mghost\x1b[0m" result = runner.invoke(cli, ["remote", "set-url", malicious, "https://hub.muse.io/r"]) assert result.exit_code != 0 err = result.stderr or "" assert "\x1b[" not in err def test_ansi_in_url_sanitized_in_output(self, repo: pathlib.Path) -> None: """ANSI in stored URL must not reach terminal as raw ESC bytes. Write TOML directly using \\u001b unicode escapes — raw \\x1b bytes are illegal in TOML quoted strings, so we write the escape form which TOML decodes to the ESC character when loading. """ config_toml = config_toml_path(repo) # \\u001b in Python str → \u001b written to disk → ESC when TOML loads it config_toml.write_text( '[remotes.origin]\nurl = "https://hub.muse.io/\\u001b[31mmalicious\\u001b[0m"\n' ) # Now set-url to a clean URL — old_url contains an ESC; JSON must encode it result = runner.invoke( cli, ["remote", "set-url", "origin", "https://hub.muse.io/clean", "--json"] ) assert result.exit_code == 0 # json.dumps always encodes ESC as \\u001b — no raw ESC byte must appear assert "\x1b" not in result.output def test_control_char_in_name_rejected(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "set-url", "ori\x00gin", "https://hub.muse.io/r"]) assert result.exit_code != 0 def test_invalid_scheme_does_not_write(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/original"]) runner.invoke(cli, ["remote", "set-url", "origin", "file:///etc/shadow"]) assert get_remote("origin", repo) == "https://hub.muse.io/original" def test_invalid_name_does_not_reach_repo( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Name validation must fire before require_repo — no repo needed.""" monkeypatch.chdir(tmp_path) monkeypatch.delenv("MUSE_REPO_ROOT", raising=False) result = runner.invoke(cli, ["remote", "set-url", "bad name", "https://hub.muse.io/r"]) assert result.exit_code != 0 assert "Invalid remote name" in result.stderr # ── set-url stress ──────────────────────────────────────────────────────────── class TestRemoteSetUrlStress: """Volume and concurrency tests for ``muse remote set-url``.""" def test_100_sequential_set_url_updates(self, repo: pathlib.Path) -> None: """100 sequential updates; final URL must match the last write.""" from muse.cli.config import set_remote, get_remote runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/v0"]) for i in range(1, 101): set_remote("origin", f"https://hub.muse.io/v{i}", repo) assert get_remote("origin", repo) == "https://hub.muse.io/v100" def test_set_url_across_10_remotes(self, repo: pathlib.Path) -> None: """Update 10 different remotes; each must store its own URL.""" from muse.cli.config import set_remote, get_remote for i in range(10): runner.invoke(cli, ["remote", "add", f"r{i}", f"https://hub.muse.io/old{i}"]) for i in range(10): set_remote(f"r{i}", f"https://hub.muse.io/new{i}", repo) for i in range(10): assert get_remote(f"r{i}", repo) == f"https://hub.muse.io/new{i}" def test_concurrent_set_url_isolated_repos( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """8 concurrent set-url calls on isolated repos must not interfere.""" import json as _json errors: list[str] = [] lock = threading.Lock() def _worker(idx: int) -> None: try: from muse.cli.config import set_remote, get_remote # Build a minimal isolated repo repo_dir = tmp_path / f"repo{idx}" dot_muse = muse_dir(repo_dir) (dot_muse / "refs" / "heads").mkdir(parents=True) (dot_muse / "objects").mkdir() (dot_muse / "commits").mkdir() (dot_muse / "snapshots").mkdir() (dot_muse / "repo.json").write_text( _json.dumps({"repo_id": f"r{idx}", "schema_version": "0.1", "domain": "midi"}) ) (dot_muse / "HEAD").write_text("ref: refs/heads/main\n") set_remote("origin", f"https://hub.muse.io/old{idx}", repo_dir) set_remote("origin", f"https://hub.muse.io/new{idx}", repo_dir) val = get_remote("origin", repo_dir) assert val == f"https://hub.muse.io/new{idx}", f"worker {idx}: got {val!r}" except Exception as exc: with lock: errors.append(f"worker {idx}: {exc}") threads = [threading.Thread(target=_worker, args=(i,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert errors == [], "\n".join(errors) # ── status extended ─────────────────────────────────────────────────────────── class TestRemoteStatusExtended: """Extended integration tests for ``muse remote status``.""" def _add(self, repo: pathlib.Path, url: str = "http://localhost:19999/gabriel/repo") -> None: runner.invoke(cli, ["remote", "add", "origin", url]) def test_status_reachable_exit_zero(self, repo: pathlib.Path) -> None: self._add(repo) with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")): result = runner.invoke(cli, ["remote", "status", "origin"]) assert result.exit_code == 0 def test_status_unreachable_exit_nonzero(self, repo: pathlib.Path) -> None: self._add(repo) with patch("muse.cli.commands.remote._ping_url", return_value=(False, None, "refused")): result = runner.invoke(cli, ["remote", "status", "origin"]) assert result.exit_code != 0 def test_status_unreachable_exit_code_is_remote_error(self, repo: pathlib.Path) -> None: """Unreachable remote must exit with REMOTE_ERROR (5), not INTERNAL_ERROR (3).""" from muse.core.errors import ExitCode self._add(repo) with patch("muse.cli.commands.remote._ping_url", return_value=(False, None, "refused")): result = runner.invoke(cli, ["remote", "status", "origin"]) assert result.exit_code == ExitCode.REMOTE_ERROR def test_status_json_all_keys_present(self, repo: pathlib.Path) -> None: self._add(repo) with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")): result = runner.invoke(cli, ["remote", "status", "origin", "--json"]) assert result.exit_code == 0 data = _json_status(result) for key in ("remote", "url", "server_root", "reachable", "http_status", "message", "tracked_refs"): assert key in data, f"Missing key '{key}'" def test_status_json_short_flag(self, repo: pathlib.Path) -> None: self._add(repo) with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")): result = runner.invoke(cli, ["remote", "status", "origin", "-j"]) assert result.exit_code == 0 data = _json_status(result) assert data["reachable"] is True def test_status_json_reachable_true(self, repo: pathlib.Path) -> None: self._add(repo) with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")): result = runner.invoke(cli, ["remote", "status", "origin", "--json"]) data = _json_status(result) assert data["reachable"] is True assert data["http_status"] == 200 def test_status_json_reachable_false_5xx(self, repo: pathlib.Path) -> None: self._add(repo) with patch("muse.cli.commands.remote._ping_url", return_value=(False, 503, "HTTP 503")): result = runner.invoke(cli, ["remote", "status", "origin", "--json"]) data = _json_status(result) assert data["reachable"] is False assert data["http_status"] == 503 def test_status_json_null_http_status_on_network_error(self, repo: pathlib.Path) -> None: self._add(repo) with patch("muse.cli.commands.remote._ping_url", return_value=(False, None, "no route")): result = runner.invoke(cli, ["remote", "status", "origin", "--json"]) data = _json_status(result) assert data["http_status"] is None def test_status_json_remote_name_correct(self, repo: pathlib.Path) -> None: self._add(repo) with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")): result = runner.invoke(cli, ["remote", "status", "origin", "--json"]) data = _json_status(result) assert data["remote"] == "origin" def test_status_json_url_correct(self, repo: pathlib.Path) -> None: self._add(repo) with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")): result = runner.invoke(cli, ["remote", "status", "origin", "--json"]) data = _json_status(result) assert data["url"] == "http://localhost:19999/gabriel/repo" def test_status_json_server_root_extracted(self, repo: pathlib.Path) -> None: self._add(repo, "http://localhost:19999/gabriel/repo") with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")): result = runner.invoke(cli, ["remote", "status", "origin", "--json"]) data = _json_status(result) assert data["server_root"] == "http://localhost:19999" def test_status_json_tracked_refs_empty_when_no_fetch(self, repo: pathlib.Path) -> None: self._add(repo) with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")): result = runner.invoke(cli, ["remote", "status", "origin", "--json"]) data = _json_status(result) assert data["tracked_refs"] == {} def test_status_json_tracked_refs_flat(self, repo: pathlib.Path) -> None: self._add(repo) refs_dir = remote_tracking_dir(repo, "origin") refs_dir.mkdir(parents=True) cid_a = long_id("a" * 64) cid_b = long_id("b" * 64) (refs_dir / "main").write_text(cid_a) (refs_dir / "dev").write_text(cid_b) with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")): result = runner.invoke(cli, ["remote", "status", "origin", "--json"]) data = _json_status(result) assert data["tracked_refs"]["main"] == cid_a assert data["tracked_refs"]["dev"] == cid_b def test_status_json_tracked_refs_nested(self, repo: pathlib.Path) -> None: self._add(repo) refs_dir = remotes_dir(repo) / "origin" (refs_dir / "feat").mkdir(parents=True) (refs_dir / "main").write_text("c" * 64) (refs_dir / "feat" / "ui").write_text("d" * 64) with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")): result = runner.invoke(cli, ["remote", "status", "origin", "--json"]) data = _json_status(result) assert "main" in data["tracked_refs"] assert "feat/ui" in data["tracked_refs"] def test_status_missing_remote_exits_nonzero(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "status", "ghost"]) assert result.exit_code != 0 def test_status_missing_remote_exit_code_user_error(self, repo: pathlib.Path) -> None: from muse.core.errors import ExitCode result = runner.invoke(cli, ["remote", "status", "ghost"]) assert result.exit_code == ExitCode.USER_ERROR def test_status_invalid_name_rejected_before_repo( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Name validation must fire before require_repo — no repo needed.""" monkeypatch.chdir(tmp_path) monkeypatch.delenv("MUSE_REPO_ROOT", raising=False) result = runner.invoke(cli, ["remote", "status", "bad name"]) assert result.exit_code != 0 assert "Invalid remote name" in result.stderr def test_status_outside_repo_exits_nonzero( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) monkeypatch.delenv("MUSE_REPO_ROOT", raising=False) result = runner.invoke(cli, ["remote", "status", "origin"]) assert result.exit_code != 0 def test_status_custom_timeout_passed_to_ping(self, repo: pathlib.Path) -> None: self._add(repo) with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")) as m: runner.invoke(cli, ["remote", "status", "origin", "--timeout", "2.5"]) assert m.call_args[0][1] == 2.5 def test_status_json_parseable_stdout(self, repo: pathlib.Path) -> None: self._add(repo) with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")): result = runner.invoke(cli, ["remote", "status", "origin", "--json"]) assert result.exit_code == 0 json_line = next( (l for l in result.output.splitlines() if l.strip().startswith("{")), None ) assert json_line is not None json.loads(json_line) # must not raise # ── status security ─────────────────────────────────────────────────────────── class TestRemoteStatusSecurity: """Security-focused tests for ``muse remote status``.""" def test_invalid_name_no_repo_needed( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) monkeypatch.delenv("MUSE_REPO_ROOT", raising=False) result = runner.invoke(cli, ["remote", "status", "bad/name"]) assert result.exit_code != 0 assert "Invalid remote name" in result.stderr def test_ansi_in_name_sanitized( self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str] ) -> None: malicious = "\x1b[31morigin\x1b[0m" result = runner.invoke(cli, ["remote", "status", malicious]) assert result.exit_code != 0 err = result.stderr or "" assert "\x1b[" not in err def test_control_char_in_name_rejected(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["remote", "status", "ori\x00gin"]) assert result.exit_code != 0 assert "Invalid remote name" in result.stderr def test_name_too_long_rejected(self, repo: pathlib.Path) -> None: long_name = "a" * 101 result = runner.invoke(cli, ["remote", "status", long_name]) assert result.exit_code != 0 assert "too long" in result.stderr def test_file_scheme_url_returns_unreachable(self, repo: pathlib.Path) -> None: """A file:// URL in config must be handled by the SSRF guard in _ping_url.""" from muse.cli.config import set_remote set_remote("badremote", "file:///etc/passwd", repo) result = runner.invoke(cli, ["remote", "status", "badremote", "--json"]) assert result.exit_code != 0 data = _json_status(result) assert data["reachable"] is False def test_ansi_in_stored_url_sanitized_in_text_output(self, repo: pathlib.Path) -> None: """ANSI codes in a stored URL must not leak as raw ESC bytes in text mode.""" config_toml = config_toml_path(repo) config_toml.write_text( '[remotes.origin]\nurl = "http://localhost/\\u001b[31mmalicious\\u001b[0m"\n' ) with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")): result = runner.invoke(cli, ["remote", "status", "origin"]) assert result.exit_code == 0 assert "\x1b" not in result.output def test_ansi_in_stored_url_escaped_in_json(self, repo: pathlib.Path) -> None: """ANSI in stored URL must be JSON-encoded, not emitted as raw ESC.""" config_toml = config_toml_path(repo) config_toml.write_text( '[remotes.origin]\nurl = "http://localhost/\\u001b[31mmalicious\\u001b[0m"\n' ) with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")): result = runner.invoke(cli, ["remote", "status", "origin", "--json"]) assert result.exit_code == 0 assert "\x1b" not in result.output def test_symlink_in_remotes_dir_skipped(self, repo: pathlib.Path) -> None: """Symlinks inside the remotes tracking dir must be skipped.""" runner.invoke(cli, ["remote", "add", "origin", "http://localhost:19999/gabriel/repo"]) refs_dir = remotes_dir(repo) / "origin" refs_dir.mkdir(parents=True) (refs_dir / "main").write_text("a" * 64) symlink = refs_dir / "malicious" symlink.symlink_to("/etc/passwd") with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")): result = runner.invoke(cli, ["remote", "status", "origin", "--json"]) assert result.exit_code == 0 data = _json_status(result) assert "malicious" not in data["tracked_refs"] assert "main" in data["tracked_refs"] # ── status stress ───────────────────────────────────────────────────────────── class TestRemoteStatusStress: """Volume and concurrency tests for ``muse remote status``.""" def _add(self, repo: pathlib.Path, url: str = "http://localhost:19999/g/r") -> None: runner.invoke(cli, ["remote", "add", "origin", url]) def test_50_sequential_status_calls_stable(self, repo: pathlib.Path) -> None: """50 status calls with a mocked reachable remote must all return exit 0.""" self._add(repo) with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")): for _ in range(50): result = runner.invoke(cli, ["remote", "status", "origin", "--json"]) assert result.exit_code == 0 def test_status_with_100_tracked_refs(self, repo: pathlib.Path) -> None: """100 tracking refs must all appear in JSON output.""" self._add(repo) refs_dir = remotes_dir(repo) / "origin" refs_dir.mkdir(parents=True) for i in range(100): (refs_dir / f"branch{i:03d}").write_text("a" * 64) with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")): result = runner.invoke(cli, ["remote", "status", "origin", "--json"]) assert result.exit_code == 0 data = _json_status(result) assert len(data["tracked_refs"]) == 100 def test_concurrent_status_reads_same_repo(self, repo: pathlib.Path) -> None: """8 concurrent status reads against the same repo must all succeed.""" self._add(repo) refs_dir = remotes_dir(repo) / "origin" refs_dir.mkdir(parents=True) (refs_dir / "main").write_text("a" * 64) results: list[int] = [] errors: list[str] = [] lock = threading.Lock() def _read() -> None: try: from muse.cli.config import get_remote from muse.cli.commands.remote import _collect_tracked_refs get_remote("origin", repo) refs = _collect_tracked_refs(refs_dir) with lock: results.append(len(refs)) except Exception as exc: with lock: errors.append(str(exc)) threads = [threading.Thread(target=_read) for _ in range(8)] for t in threads: t.start() for t in threads: t.join() assert errors == [], "\n".join(errors) assert all(r == 1 for r in results)