test_cross_host_signing_identity.py
python
sha256:c5d88af19bcd6c1e60fd829dea85a4fdd5a74f74c5617031ba5b2955ba6a5b95
fix: domains.py signing-identity resolution uses resolved h…
Sonnet 5
patch
15 hours ago
| 1 | """muse#20 Phase 4 — cross-host signing identity regression guard. |
| 2 | |
| 3 | Every command that talks to a remote must resolve the signing identity |
| 4 | against the ACTUAL target hostname, not the repo's default [hub] url. |
| 5 | muse#20's background bug (the ls-remote HTTP 401) and the domains.py bug |
| 6 | fixed in Phase 1 are both instances of the same mistake: computing the |
| 7 | target URL, then calling `get_signing_identity(root)` without forwarding |
| 8 | it, so the lookup silently falls back to the wrong host's key. |
| 9 | |
| 10 | This is a single integration-style test (mocked transport/network — no |
| 11 | live hub required, unlike tests/test_wire_localhost.py) that configures a |
| 12 | repo whose default [hub] url and a named remote deliberately point at |
| 13 | DIFFERENT hostnames, then drives `ls-remote`, `fetch`, and `domains |
| 14 | publish` against the named remote — the three callers the ticket's Phase 4 |
| 15 | specifically names — asserting each one resolves identity against the |
| 16 | remote's own hostname. Intended to catch this exact regression class |
| 17 | immediately whenever a new remote-talking command is added without |
| 18 | forwarding the resolved URL. |
| 19 | """ |
| 20 | from __future__ import annotations |
| 21 | |
| 22 | import json |
| 23 | import pathlib |
| 24 | from unittest.mock import patch |
| 25 | |
| 26 | from tests.cli_test_helper import CliRunner |
| 27 | from muse.core.paths import muse_dir |
| 28 | |
| 29 | cli = None |
| 30 | runner = CliRunner() |
| 31 | |
| 32 | _DEFAULT_HUB = "https://localhost:1337" |
| 33 | _TARGET_REMOTE_URL = "https://staging.musehub.ai/gabriel/muse" |
| 34 | _TARGET_HOST = "staging.musehub.ai" |
| 35 | |
| 36 | |
| 37 | def _env(root: pathlib.Path) -> dict[str, str]: |
| 38 | return {"MUSE_REPO_ROOT": str(root)} |
| 39 | |
| 40 | |
| 41 | def _init_repo(tmp_path: pathlib.Path) -> pathlib.Path: |
| 42 | """Repo whose default [hub] url and the "staging" remote point at |
| 43 | deliberately different hosts, so a wrong-hostname identity lookup is |
| 44 | observable.""" |
| 45 | dot_muse = muse_dir(tmp_path) |
| 46 | for sub in ("commits", "snapshots", "objects", "refs/heads"): |
| 47 | (dot_muse / sub).mkdir(parents=True, exist_ok=True) |
| 48 | (dot_muse / "HEAD").write_text("ref: refs/heads/main\n", encoding="utf-8") |
| 49 | (dot_muse / "repo.json").write_text( |
| 50 | json.dumps({"repo_id": "test-repo", "domain": "code"}), encoding="utf-8" |
| 51 | ) |
| 52 | (dot_muse / "config.toml").write_text( |
| 53 | f'[hub]\nurl = "{_DEFAULT_HUB}"\n\n' |
| 54 | f'[remotes.staging]\nurl = "{_TARGET_REMOTE_URL}"\n', |
| 55 | encoding="utf-8", |
| 56 | ) |
| 57 | return tmp_path |
| 58 | |
| 59 | |
| 60 | def _make_remote_info() -> dict: |
| 61 | return { |
| 62 | "repo_id": "test-repo", |
| 63 | "domain": "code", |
| 64 | "default_branch": "main", |
| 65 | "branch_heads": {"main": "sha256:" + "a" * 64}, |
| 66 | } |
| 67 | |
| 68 | |
| 69 | class TestCrossHostSigningIdentity: |
| 70 | def test_ls_remote_uses_target_host_not_default_hub(self, tmp_path: pathlib.Path) -> None: |
| 71 | repo = _init_repo(tmp_path) |
| 72 | with patch("muse.cli.commands.ls_remote.HttpTransport") as mt: |
| 73 | with patch("muse.cli.commands.ls_remote.get_signing_identity") as mock_gsi: |
| 74 | mock_gsi.return_value = None |
| 75 | mt.return_value.fetch_remote_info.return_value = _make_remote_info() |
| 76 | r = runner.invoke(cli, ["ls-remote", "staging", "--json"], env=_env(repo)) |
| 77 | assert r.exit_code == 0, r.output |
| 78 | mock_gsi.assert_called_once() |
| 79 | _, kwargs = mock_gsi.call_args |
| 80 | assert kwargs.get("remote_url") == _TARGET_REMOTE_URL |
| 81 | assert _TARGET_HOST in kwargs["remote_url"] |
| 82 | assert "localhost" not in kwargs["remote_url"] |
| 83 | |
| 84 | def test_fetch_uses_target_host_not_default_hub(self, tmp_path: pathlib.Path) -> None: |
| 85 | repo = _init_repo(tmp_path) |
| 86 | with patch("muse.cli.commands.fetch.get_signing_identity") as mock_gsi: |
| 87 | mock_gsi.return_value = None |
| 88 | with patch("muse.cli.commands.fetch.make_transport") as mt: |
| 89 | mt.return_value.fetch_remote_info.return_value = _make_remote_info() |
| 90 | r = runner.invoke(cli, ["fetch", "staging", "--json"], env=_env(repo)) |
| 91 | assert r.exit_code == 0, r.output |
| 92 | mock_gsi.assert_called_once() |
| 93 | _, kwargs = mock_gsi.call_args |
| 94 | assert kwargs.get("remote_url") == _TARGET_REMOTE_URL |
| 95 | assert _TARGET_HOST in kwargs["remote_url"] |
| 96 | assert "localhost" not in kwargs["remote_url"] |
| 97 | |
| 98 | def test_domains_publish_uses_target_host_not_default_hub(self, tmp_path: pathlib.Path) -> None: |
| 99 | repo = _init_repo(tmp_path) |
| 100 | with patch("muse.cli.commands.domains.get_signing_identity") as mock_gsi: |
| 101 | mock_gsi.return_value = None |
| 102 | runner.invoke( |
| 103 | cli, |
| 104 | [ |
| 105 | "domains", "publish", |
| 106 | "--author", "gabriel", "--slug", "genomics", |
| 107 | "--name", "Genomics", "--description", "desc", |
| 108 | "--viewer-type", "generic", "--hub", _TARGET_REMOTE_URL, |
| 109 | ], |
| 110 | env=_env(repo), |
| 111 | ) |
| 112 | mock_gsi.assert_called_once() |
| 113 | _, kwargs = mock_gsi.call_args |
| 114 | assert kwargs.get("remote_url") == _TARGET_REMOTE_URL |
| 115 | assert _TARGET_HOST in kwargs["remote_url"] |
| 116 | assert "localhost" not in kwargs["remote_url"] |
| 117 | |
| 118 | def test_no_caller_ever_falls_back_to_default_hub_host(self, tmp_path: pathlib.Path) -> None: |
| 119 | """Belt-and-suspenders: across all three callers, none resolves |
| 120 | identity against the repo's default [hub] url host once a |
| 121 | different target is explicit (named remote or --hub).""" |
| 122 | repo = _init_repo(tmp_path) |
| 123 | seen_hosts: list[str] = [] |
| 124 | |
| 125 | with patch("muse.cli.commands.ls_remote.HttpTransport") as mt1: |
| 126 | with patch("muse.cli.commands.ls_remote.get_signing_identity") as gsi1: |
| 127 | gsi1.return_value = None |
| 128 | mt1.return_value.fetch_remote_info.return_value = _make_remote_info() |
| 129 | runner.invoke(cli, ["ls-remote", "staging", "--json"], env=_env(repo)) |
| 130 | seen_hosts.append(gsi1.call_args.kwargs.get("remote_url", "")) |
| 131 | |
| 132 | with patch("muse.cli.commands.fetch.get_signing_identity") as gsi2: |
| 133 | gsi2.return_value = None |
| 134 | with patch("muse.cli.commands.fetch.make_transport") as mt2: |
| 135 | mt2.return_value.fetch_remote_info.return_value = _make_remote_info() |
| 136 | runner.invoke(cli, ["fetch", "staging", "--json"], env=_env(repo)) |
| 137 | seen_hosts.append(gsi2.call_args.kwargs.get("remote_url", "")) |
| 138 | |
| 139 | with patch("muse.cli.commands.domains.get_signing_identity") as gsi3: |
| 140 | gsi3.return_value = None |
| 141 | runner.invoke( |
| 142 | cli, |
| 143 | [ |
| 144 | "domains", "publish", |
| 145 | "--author", "gabriel", "--slug", "genomics", |
| 146 | "--name", "Genomics", "--description", "desc", |
| 147 | "--viewer-type", "generic", "--hub", _TARGET_REMOTE_URL, |
| 148 | ], |
| 149 | env=_env(repo), |
| 150 | ) |
| 151 | seen_hosts.append(gsi3.call_args.kwargs.get("remote_url", "")) |
| 152 | |
| 153 | assert all(_TARGET_HOST in h for h in seen_hosts), seen_hosts |
| 154 | assert not any("localhost" in h for h in seen_hosts), seen_hosts |
File History
1 commit
sha256:c5d88af19bcd6c1e60fd829dea85a4fdd5a74f74c5617031ba5b2955ba6a5b95
fix: domains.py signing-identity resolution uses resolved h…
Sonnet 5
patch
15 hours ago