gabriel / muse public
test_core_transport.py python
764 lines 30.3 KB
Raw
sha256:79ffe87f5fe2ec146e35f05521218bbf54dffdb0440c07f970bad05f16efb89f chore: merge main — carry all urllib/typing/test fixes from dev Sonnet 4.6 minor ⚠ breaking 47 days ago
1 """Tests for muse.core.transport — HttpTransport and response parsers."""
2
3 from __future__ import annotations
4
5 import json
6 import signal
7 import socket
8 import threading
9 import time
10 import unittest.mock
11 import urllib.error
12 import urllib.request
13 from io import BytesIO
14
15 import msgpack
16 import pytest
17
18 from muse.core._types import MsgpackDict
19 from muse.core.pack import PackBundle, RemoteInfo
20 from muse.core.transport import (
21 HttpTransport,
22 SigningIdentity,
23 TransportError,
24 _parse_bundle,
25 _parse_push_result,
26 _parse_remote_info,
27 build_msign_header,
28 )
29
30
31 # ---------------------------------------------------------------------------
32 # Helpers
33 # ---------------------------------------------------------------------------
34
35
36 def _make_signing() -> "SigningIdentity":
37 """Generate a fresh Ed25519 SigningIdentity for tests."""
38 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
39 from muse.core.transport import SigningIdentity
40 return SigningIdentity(handle="testuser", private_key=Ed25519PrivateKey.generate())
41
42
43 def _mock_response(
44 body: bytes,
45 status: int = 200,
46 content_type: str = "application/x-msgpack",
47 streaming: bool = False,
48 ) -> unittest.mock.MagicMock:
49 """Return a mock urllib response context manager.
50
51 When *streaming* is True, ``read()`` returns *body* on the first call and
52 ``b""`` on all subsequent calls — matching the chunked-read loop used by
53 :meth:`~muse.core.transport.HttpTransport._execute_fetch`.
54 """
55 resp = unittest.mock.MagicMock()
56 if streaming:
57 resp.read.side_effect = [body, b""]
58 else:
59 resp.read.return_value = body
60 resp.headers = {"Content-Type": content_type}
61 resp.__enter__ = lambda s: s
62 resp.__exit__ = unittest.mock.MagicMock(return_value=False)
63 return resp
64
65
66 def _mp(data: MsgpackDict) -> bytes:
67 """Encode data as msgpack."""
68 return msgpack.packb(data, use_bin_type=True)
69
70
71 def _http_error(code: int, body: bytes = b"") -> urllib.error.HTTPError:
72 return urllib.error.HTTPError(
73 url="https://example.com",
74 code=code,
75 msg=str(code),
76 hdrs=None,
77 fp=BytesIO(body),
78 )
79
80
81 # ---------------------------------------------------------------------------
82 # _parse_remote_info
83 # ---------------------------------------------------------------------------
84
85
86 class TestParseRemoteInfo:
87 def test_valid_response(self) -> None:
88 raw = _mp(
89 {
90 "repo_id": "r123",
91 "domain": "midi",
92 "default_branch": "main",
93 "branch_heads": {"main": "abc123", "dev": "def456"},
94 }
95 )
96 info = _parse_remote_info(raw)
97 assert info["repo_id"] == "r123"
98 assert info["domain"] == "midi"
99 assert info["default_branch"] == "main"
100 assert info["branch_heads"] == {"main": "abc123", "dev": "def456"}
101
102 def test_invalid_msgpack_raises_transport_error(self) -> None:
103 with pytest.raises(TransportError):
104 _parse_remote_info(b"\xff\xff\xff\xff\xff invalid")
105
106 def test_non_dict_response_returns_defaults(self) -> None:
107 raw = _mp([1, 2, 3])
108 info = _parse_remote_info(raw)
109 assert info["repo_id"] == ""
110 assert info["branch_heads"] == {}
111
112 def test_missing_fields_get_defaults(self) -> None:
113 raw = _mp({"repo_id": "x"})
114 info = _parse_remote_info(raw)
115 assert info["repo_id"] == "x"
116 assert info["domain"] == "midi"
117 assert info["default_branch"] == "main"
118 assert info["branch_heads"] == {}
119
120 def test_non_string_branch_heads_excluded(self) -> None:
121 raw = _mp({"branch_heads": {"main": "abc", "bad": 123}})
122 info = _parse_remote_info(raw)
123 assert "main" in info["branch_heads"]
124 assert "bad" not in info["branch_heads"]
125
126 def test_pack_origin_populated_when_present(self) -> None:
127 raw = _mp(
128 {
129 "repo_id": "r1",
130 "domain": "code",
131 "default_branch": "main",
132 "branch_heads": {},
133 "pack_origin": "https://worker.example.workers.dev",
134 }
135 )
136 info = _parse_remote_info(raw)
137 assert info.get("pack_origin") == "https://worker.example.workers.dev"
138
139 def test_pack_origin_absent_when_not_in_response(self) -> None:
140 raw = _mp({"repo_id": "r1", "branch_heads": {}})
141 info = _parse_remote_info(raw)
142 assert "pack_origin" not in info
143
144 def test_pack_origin_whitespace_stripped(self) -> None:
145 raw = _mp(
146 {"repo_id": "r1", "branch_heads": {}, "pack_origin": " https://w.example.dev "}
147 )
148 info = _parse_remote_info(raw)
149 assert info.get("pack_origin") == "https://w.example.dev"
150
151 def test_pack_origin_none_value_not_set(self) -> None:
152 raw = _mp({"repo_id": "r1", "branch_heads": {}, "pack_origin": None})
153 info = _parse_remote_info(raw)
154 assert "pack_origin" not in info
155
156 def test_pack_origin_empty_string_not_set(self) -> None:
157 raw = _mp({"repo_id": "r1", "branch_heads": {}, "pack_origin": ""})
158 info = _parse_remote_info(raw)
159 assert "pack_origin" not in info
160
161 def test_pack_origin_whitespace_only_not_set(self) -> None:
162 raw = _mp({"repo_id": "r1", "branch_heads": {}, "pack_origin": " "})
163 info = _parse_remote_info(raw)
164 assert "pack_origin" not in info
165
166
167 # ---------------------------------------------------------------------------
168 # build_msign_header — module-level signing utility
169 # ---------------------------------------------------------------------------
170
171
172 class TestBuildMsignHeader:
173 def _make_signing(self) -> SigningIdentity:
174 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
175
176 return SigningIdentity(handle="testuser", private_key=Ed25519PrivateKey.generate())
177
178 def test_header_format(self) -> None:
179 header = build_msign_header(self._make_signing(), "GET", "https://example.com/path", None)
180 assert header.startswith('MSign handle="testuser"')
181 assert " ts=" in header
182 assert " sig=" in header
183
184 def test_timestamp_is_recent(self) -> None:
185 import time
186
187 before = int(time.time())
188 header = build_msign_header(self._make_signing(), "GET", "https://example.com/p", None)
189 after = int(time.time())
190 ts_part = next(p for p in header.split() if p.startswith("ts="))
191 ts = int(ts_part[3:])
192 assert before <= ts <= after + 1
193
194 def test_signature_is_verifiable(self) -> None:
195 """The sig= value must verify against the Ed25519 public key for the canonical input."""
196 import base64
197 import hashlib
198
199 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
200
201 private_key = Ed25519PrivateKey.generate()
202 signing = SigningIdentity(handle="testuser", private_key=private_key)
203 method = "POST"
204 url = "https://hub.example.com/owner/repo/push"
205 body = b"some body data"
206
207 header = build_msign_header(signing, method, url, body)
208
209 parts: dict[str, str] = {}
210 for part in header[len("MSign "):].split():
211 k, _, v = part.partition("=")
212 parts[k] = v.strip('"')
213
214 ts = int(parts["ts"])
215 sig_bytes = base64.urlsafe_b64decode(parts["sig"] + "==")
216
217 body_hash = hashlib.sha256(body).hexdigest()
218 canonical = f"{method}\n/owner/repo/push\n{ts}\n{body_hash}".encode()
219
220 # raises cryptography.exceptions.InvalidSignature on failure
221 private_key.public_key().verify(sig_bytes, canonical)
222
223 def test_query_string_included_in_canonical(self) -> None:
224 """Query parameters must be part of the signed path, not dropped."""
225 import base64
226 import hashlib
227
228 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
229
230 private_key = Ed25519PrivateKey.generate()
231 signing = SigningIdentity(handle="u", private_key=private_key)
232 url = "https://hub.example.com/path?foo=bar&baz=1"
233
234 header = build_msign_header(signing, "GET", url, None)
235
236 parts: dict[str, str] = {}
237 for part in header[len("MSign "):].split():
238 k, _, v = part.partition("=")
239 parts[k] = v.strip('"')
240
241 ts = int(parts["ts"])
242 sig_bytes = base64.urlsafe_b64decode(parts["sig"] + "==")
243 body_hash = hashlib.sha256(b"").hexdigest()
244 canonical = f"GET\n/path?foo=bar&baz=1\n{ts}\n{body_hash}".encode()
245
246 private_key.public_key().verify(sig_bytes, canonical)
247
248 def test_none_body_treated_as_empty_bytes(self) -> None:
249 """None and b'' must produce the same SHA-256 body hash in the canonical form."""
250 import base64
251 import hashlib
252
253 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
254
255 private_key = Ed25519PrivateKey.generate()
256 signing = SigningIdentity(handle="u", private_key=private_key)
257 url = "https://hub.example.com/path"
258
259 header = build_msign_header(signing, "GET", url, None)
260
261 parts: dict[str, str] = {}
262 for part in header[len("MSign "):].split():
263 k, _, v = part.partition("=")
264 parts[k] = v.strip('"')
265
266 ts = int(parts["ts"])
267 sig_bytes = base64.urlsafe_b64decode(parts["sig"] + "==")
268 # body=None → b"" → sha256("") is the canonical body hash
269 body_hash = hashlib.sha256(b"").hexdigest()
270 canonical = f"GET\n/path\n{ts}\n{body_hash}".encode()
271
272 private_key.public_key().verify(sig_bytes, canonical)
273
274 def test_different_methods_produce_different_headers(self) -> None:
275 """Two calls with different methods on the same URL must differ (method is in canonical)."""
276
277 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
278
279 private_key = Ed25519PrivateKey.generate()
280 signing = SigningIdentity(handle="u", private_key=private_key)
281
282 with unittest.mock.patch("muse.core.transport.time") as mt:
283 mt.time.return_value = 1_700_000_000
284 h_get = build_msign_header(signing, "GET", "https://example.com/x", b"")
285 h_post = build_msign_header(signing, "POST", "https://example.com/x", b"")
286
287 assert h_get != h_post
288
289 def test_different_bodies_produce_different_sigs(self) -> None:
290 """Body content must influence the signature (body hash is in canonical)."""
291
292 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
293
294 private_key = Ed25519PrivateKey.generate()
295 signing = SigningIdentity(handle="u", private_key=private_key)
296
297 with unittest.mock.patch("muse.core.transport.time") as mt:
298 mt.time.return_value = 1_700_000_000
299 h1 = build_msign_header(signing, "POST", "https://example.com/x", b"body-a")
300 h2 = build_msign_header(signing, "POST", "https://example.com/x", b"body-b")
301
302 assert h1 != h2
303
304 def test_handle_embedded_in_header(self) -> None:
305 """The MSign header must carry the signing identity's handle."""
306
307 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
308
309 private_key = Ed25519PrivateKey.generate()
310 signing = SigningIdentity(handle="my-agent-42", private_key=private_key)
311 header = build_msign_header(signing, "GET", "https://example.com/x", None)
312 assert 'handle="my-agent-42"' in header
313
314
315 # ---------------------------------------------------------------------------
316 # _parse_bundle
317 # ---------------------------------------------------------------------------
318
319
320 class TestParseBundle:
321 def test_empty_msgpack_object_returns_empty_bundle(self) -> None:
322 bundle = _parse_bundle(_mp({}))
323 assert bundle == {}
324
325 def test_non_dict_returns_empty_bundle(self) -> None:
326 bundle = _parse_bundle(_mp([]))
327 assert bundle == {}
328
329 def test_commits_extracted(self) -> None:
330 raw = _mp(
331 {
332 "commits": [
333 {
334 "commit_id": "c1",
335 "repo_id": "r1",
336 "branch": "main",
337 "snapshot_id": "1" * 64,
338 "message": "test",
339 "committed_at": "2026-01-01T00:00:00+00:00",
340 "parent_commit_id": None,
341 "parent2_commit_id": None,
342 "author": "bob",
343 "metadata": {},
344 }
345 ]
346 }
347 )
348 bundle = _parse_bundle(raw)
349 commits = bundle.get("commits") or []
350 assert len(commits) == 1
351 assert commits[0]["commit_id"] == "c1"
352
353 def test_objects_extracted(self) -> None:
354 raw = _mp(
355 {
356 "objects": [
357 {
358 "object_id": "abc123",
359 "content": b"hello",
360 }
361 ]
362 }
363 )
364 bundle = _parse_bundle(raw)
365 objs = bundle.get("objects") or []
366 assert len(objs) == 1
367 assert objs[0]["object_id"] == "abc123"
368 assert objs[0]["content"] == b"hello"
369
370 def test_object_missing_content_excluded(self) -> None:
371 raw = _mp({"objects": [{"object_id": "abc"}]})
372 bundle = _parse_bundle(raw)
373 assert (bundle.get("objects") or []) == []
374
375 def test_branch_heads_extracted(self) -> None:
376 raw = _mp({"branch_heads": {"main": "abc123"}})
377 bundle = _parse_bundle(raw)
378 assert bundle.get("branch_heads") == {"main": "abc123"}
379
380
381 # ---------------------------------------------------------------------------
382 # _parse_push_result
383 # ---------------------------------------------------------------------------
384
385
386 class TestParsePushResult:
387 def test_success_response(self) -> None:
388 raw = _mp({"ok": True, "message": "pushed", "branch_heads": {"main": "abc"}})
389 result = _parse_push_result(raw)
390 assert result["ok"] is True
391 assert result["message"] == "pushed"
392 assert result["branch_heads"] == {"main": "abc"}
393
394 def test_failure_response(self) -> None:
395 raw = _mp({"ok": False, "message": "rejected", "branch_heads": {}})
396 result = _parse_push_result(raw)
397 assert result["ok"] is False
398 assert result["message"] == "rejected"
399
400 def test_non_msgpack_raises_transport_error(self) -> None:
401 with pytest.raises(TransportError):
402 _parse_push_result(b"\xff\xff invalid msgpack")
403
404 def test_missing_ok_defaults_false(self) -> None:
405 raw = _mp({"message": "hm", "branch_heads": {}})
406 result = _parse_push_result(raw)
407 assert result["ok"] is False
408
409
410 # ---------------------------------------------------------------------------
411 # HttpTransport — mocked urlopen
412 # ---------------------------------------------------------------------------
413
414
415 class TestHttpTransportFetchRemoteInfo:
416 def test_calls_correct_endpoint(self) -> None:
417 body = _mp(
418 {
419 "repo_id": "r1",
420 "domain": "midi",
421 "default_branch": "main",
422 "branch_heads": {"main": "abc"},
423 }
424 )
425 mock_resp = _mock_response(body)
426 with unittest.mock.patch("muse.core.transport._open_url", return_value=mock_resp) as m:
427 transport = HttpTransport()
428 info = transport.fetch_remote_info("https://hub.example.com/repos/r1", None)
429 req = m.call_args[0][0]
430 assert req.full_url == "https://hub.example.com/repos/r1/refs"
431 assert info["repo_id"] == "r1"
432
433 def test_msign_header_sent(self) -> None:
434 body = _mp(
435 {"repo_id": "r1", "domain": "midi", "default_branch": "main", "branch_heads": {}}
436 )
437 mock_resp = _mock_response(body)
438 signing = _make_signing()
439 with unittest.mock.patch("muse.core.transport._open_url", return_value=mock_resp) as m:
440 HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1", signing)
441 req = m.call_args[0][0]
442 assert req.get_header("Authorization").startswith("MSign handle=\"testuser\"")
443
444 def test_no_token_no_auth_header(self) -> None:
445 body = _mp(
446 {"repo_id": "r1", "domain": "midi", "default_branch": "main", "branch_heads": {}}
447 )
448 mock_resp = _mock_response(body)
449 with unittest.mock.patch("muse.core.transport._open_url", return_value=mock_resp) as m:
450 HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1", None)
451 req = m.call_args[0][0]
452 assert req.get_header("Authorization") is None
453
454 def test_http_401_raises_transport_error(self) -> None:
455 with unittest.mock.patch(
456 "muse.core.transport._open_url", side_effect=_http_error(401, b"Unauthorized")
457 ):
458 with pytest.raises(TransportError) as exc_info:
459 HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1", None)
460 assert exc_info.value.status_code == 401
461
462 def test_http_404_raises_transport_error(self) -> None:
463 with unittest.mock.patch(
464 "muse.core.transport._open_url", side_effect=_http_error(404)
465 ):
466 with pytest.raises(TransportError) as exc_info:
467 HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1", None)
468 assert exc_info.value.status_code == 404
469
470 def test_http_500_raises_transport_error(self) -> None:
471 with unittest.mock.patch(
472 "muse.core.transport._open_url", side_effect=_http_error(500, b"Internal Error")
473 ):
474 with pytest.raises(TransportError) as exc_info:
475 HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1", None)
476 assert exc_info.value.status_code == 500
477
478 def test_url_error_raises_transport_error_with_code_0(self) -> None:
479 with unittest.mock.patch(
480 "muse.core.transport._open_url",
481 side_effect=urllib.error.URLError("Name or service not known"),
482 ):
483 with pytest.raises(TransportError) as exc_info:
484 HttpTransport().fetch_remote_info("https://bad.host/r", None)
485 assert exc_info.value.status_code == 0
486
487 def test_trailing_slash_stripped_from_url(self) -> None:
488 body = _mp(
489 {"repo_id": "r", "domain": "midi", "default_branch": "main", "branch_heads": {}}
490 )
491 mock_resp = _mock_response(body)
492 with unittest.mock.patch("muse.core.transport._open_url", return_value=mock_resp) as m:
493 HttpTransport().fetch_remote_info("https://hub.example.com/repos/r1/", None)
494 req = m.call_args[0][0]
495 assert req.full_url == "https://hub.example.com/repos/r1/refs"
496
497
498 class TestHttpTransportFetchPack:
499 def test_posts_to_fetch_endpoint(self) -> None:
500 bundle_body = _mp(
501 {
502 "commits": [],
503 "snapshots": [],
504 "objects": [],
505 "branch_heads": {"main": "abc"},
506 }
507 )
508 # fetch_pack uses _execute_fetch which reads in a chunked loop until
509 # read() returns b"" — use streaming=True so the mock terminates.
510 mock_resp = _mock_response(bundle_body, streaming=True)
511 with unittest.mock.patch("muse.core.transport._open_url", return_value=mock_resp) as m:
512 transport = HttpTransport()
513 bundle = transport.fetch_pack(
514 "https://hub.example.com/repos/r1",
515 _make_signing(),
516 want=["abc"],
517 have=["def"],
518 )
519 req = m.call_args[0][0]
520 assert req.full_url == "https://hub.example.com/repos/r1/fetch"
521 sent = msgpack.unpackb(req.data, raw=False)
522 assert sent["want"] == ["abc"]
523 assert sent["have"] == ["def"]
524 assert bundle.get("branch_heads") == {"main": "abc"}
525
526 def test_http_409_raises_transport_error(self) -> None:
527 with unittest.mock.patch(
528 "muse.core.transport._open_url", side_effect=_http_error(409)
529 ):
530 with pytest.raises(TransportError) as exc_info:
531 HttpTransport().fetch_pack("https://hub.example.com/r", None, [], [])
532 assert exc_info.value.status_code == 409
533
534
535 class TestHttpTransportPushPack:
536 def test_posts_to_push_endpoint(self) -> None:
537 push_body = _mp({"ok": True, "message": "ok", "branch_heads": {"main": "new"}})
538 mock_resp = _mock_response(push_body)
539 bundle: PackBundle = {"commits": [], "snapshots": [], "objects": []}
540 with unittest.mock.patch("muse.core.transport._open_url", return_value=mock_resp) as m:
541 result = HttpTransport().push_pack(
542 "https://hub.example.com/repos/r1", _make_signing(), bundle, "main", False
543 )
544 req = m.call_args[0][0]
545 assert req.full_url == "https://hub.example.com/repos/r1/push"
546 sent = msgpack.unpackb(req.data, raw=False)
547 assert sent["branch"] == "main"
548 assert sent["force"] is False
549 assert result["ok"] is True
550
551 def test_force_flag_sent(self) -> None:
552 push_body = _mp({"ok": True, "message": "", "branch_heads": {}})
553 mock_resp = _mock_response(push_body)
554 bundle: PackBundle = {}
555 with unittest.mock.patch("muse.core.transport._open_url", return_value=mock_resp) as m:
556 HttpTransport().push_pack("https://hub.example.com/r", None, bundle, "main", True)
557 req = m.call_args[0][0]
558 sent = msgpack.unpackb(req.data, raw=False)
559 assert sent["force"] is True
560
561 def test_push_rejected_raises_transport_error(self) -> None:
562 with unittest.mock.patch(
563 "muse.core.transport._open_url", side_effect=_http_error(409, b"non-fast-forward")
564 ):
565 with pytest.raises(TransportError) as exc_info:
566 HttpTransport().push_pack(
567 "https://hub.example.com/r", None, {}, "main", False
568 )
569 assert exc_info.value.status_code == 409
570
571
572 # ---------------------------------------------------------------------------
573 # HttpTransport._build_request — credential security and loopback allowlist
574 # ---------------------------------------------------------------------------
575
576
577 class TestBuildRequest:
578 """_build_request enforces HTTPS for non-loopback URLs with signing identity."""
579
580 def _build(self, url: str, with_signing: bool = True) -> urllib.request.Request:
581 signing = _make_signing() if with_signing else None
582 # Stub TOFU so unit tests don't make real TLS connections to verify
583 # certificate fingerprints — that's network state, not request structure.
584 with unittest.mock.patch("muse.core.hub_trust.check_and_pin"):
585 return HttpTransport()._build_request("GET", url, signing)
586
587 # ── Loopback hosts allowed over plain HTTP ────────────────────────────
588
589 def test_localhost_http_with_token_allowed(self) -> None:
590 req = self._build("http://localhost:10003/repo/refs")
591 assert req.get_header("Authorization").startswith("MSign ")
592
593 def test_127_0_0_1_http_with_token_allowed(self) -> None:
594 req = self._build("http://127.0.0.1:10003/repo/refs")
595 assert req.get_header("Authorization").startswith("MSign ")
596
597 def test_ipv6_loopback_http_with_token_allowed(self) -> None:
598 req = self._build("http://[::1]:10003/repo/refs")
599 assert req.get_header("Authorization").startswith("MSign ")
600
601 def test_host_docker_internal_http_with_token_allowed(self) -> None:
602 """host.docker.internal is Docker Desktop's alias for the host loopback.
603
604 Agent swarms run inside Docker and use http://host.docker.internal:10003
605 to reach a local MuseHub instance. Credentials must be sent over this
606 plain-HTTP connection — the traffic never leaves the machine.
607 """
608 req = self._build("http://host.docker.internal:10003/gabriel/repo/refs")
609 assert req.get_header("Authorization").startswith("MSign ")
610
611 def test_https_any_host_with_token_allowed(self) -> None:
612 req = self._build("https://musehub.ai/gabriel/repo/refs")
613 assert req.get_header("Authorization").startswith("MSign ")
614
615 # ── Non-loopback HTTP with token must be rejected ─────────────────────
616
617 def test_non_loopback_http_token_raises_transport_error(self) -> None:
618 with pytest.raises(TransportError, match="non-HTTPS"):
619 self._build("http://musehub.ai/gabriel/repo/refs")
620
621 def test_arbitrary_hostname_http_token_raises(self) -> None:
622 with pytest.raises(TransportError, match="non-HTTPS"):
623 self._build("http://evil.example.com/steal")
624
625 def test_localhost_lookalike_http_token_raises(self) -> None:
626 """'localhost.evil.com' must NOT be mistaken for the loopback interface."""
627 with pytest.raises(TransportError, match="non-HTTPS"):
628 self._build("http://localhost.evil.com/repo/refs")
629
630 def test_host_docker_internal_lookalike_http_token_raises(self) -> None:
631 """'host.docker.internal.evil.com' must not bypass the check."""
632 with pytest.raises(TransportError, match="non-HTTPS"):
633 self._build("http://host.docker.internal.evil.com/repo")
634
635 # ── No token — scheme restriction does not apply ──────────────────────
636
637 def test_non_loopback_http_without_token_allowed(self) -> None:
638 req = self._build("http://musehub.ai/public/repo/refs", with_signing=False)
639 assert req.get_header("Authorization") is None
640
641 # ── Request structure ─────────────────────────────────────────────────
642
643 def test_accept_header_always_set(self) -> None:
644 req = self._build("https://musehub.ai/repo/refs")
645 assert "msgpack" in req.get_header("Accept")
646
647 def test_method_preserved(self) -> None:
648 req = HttpTransport()._build_request("POST", "https://musehub.ai/x", None)
649 assert req.get_method() == "POST"
650
651 def test_body_sets_content_type(self) -> None:
652 req = HttpTransport()._build_request(
653 "POST", "https://musehub.ai/x", None, body_bytes=b"data"
654 )
655 assert req.get_header("Content-type") == "application/x-msgpack"
656
657 def test_no_body_omits_content_type(self) -> None:
658 req = self._build("https://musehub.ai/x", with_signing=False)
659 assert req.get_header("Content-type") is None
660
661
662 # ---------------------------------------------------------------------------
663 # SIGPIPE regression — large push body with early-close server
664 # ---------------------------------------------------------------------------
665
666
667 def _early_close_server(
668 server_sock: socket.socket,
669 response_code: int,
670 resp_body: bytes,
671 ) -> None:
672 """Accept one connection, read HTTP headers, send response, close immediately.
673
674 Simulates the scenario where the server sends a 4xx/5xx response while
675 the client is still uploading a large request body. Without the
676 ``_ignore_sigpipe`` guard in ``_open_url``, the client process dies with
677 exit code 141 (SIGPIPE) instead of raising ``TransportError``.
678 """
679 try:
680 conn, _ = server_sock.accept()
681 conn.settimeout(5.0)
682 try:
683 buf = b""
684 deadline = time.time() + 5
685 while time.time() < deadline:
686 try:
687 chunk = conn.recv(4096)
688 if not chunk:
689 break
690 buf += chunk
691 if b"\r\n\r\n" in buf:
692 break
693 except socket.timeout:
694 break
695 status_line = f"HTTP/1.1 {response_code} Error\r\n"
696 resp_headers = (
697 "Content-Type: application/json\r\n"
698 f"Content-Length: {len(resp_body)}\r\n"
699 "Connection: close\r\n\r\n"
700 )
701 conn.sendall((status_line + resp_headers).encode() + resp_body)
702 finally:
703 conn.close()
704 except Exception:
705 pass
706 finally:
707 server_sock.close()
708
709
710 class TestSigpipeRegression:
711 """Regression tests for SIGPIPE on large push bodies.
712
713 ``muse/cli/app.py`` sets ``SIGPIPE = SIG_DFL`` so that piping output to
714 ``head``/``grep``/``jq`` exits cleanly. Without the ``_ignore_sigpipe``
715 guard in ``_open_url`` the push command dies with exit 141 when the server
716 closes the connection while the client is still uploading a large body.
717 """
718
719 def _run_early_close_scenario(self, payload_mb: float, response_code: int) -> None:
720 """Assert that TransportError is raised, not a process-killing SIGPIPE."""
721 # Reproduce app.py's startup action — SIG_DFL kills the process on SIGPIPE.
722 if hasattr(signal, "SIGPIPE"):
723 original = signal.signal(signal.SIGPIPE, signal.SIG_DFL)
724 else:
725 original = None
726
727 body = b"X" * int(payload_mb * 1024 * 1024)
728 resp_body = b'{"detail":"test error"}'
729
730 server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
731 server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
732 server_sock.bind(("127.0.0.1", 0))
733 server_sock.listen(1)
734 port = server_sock.getsockname()[1]
735
736 t = threading.Thread(
737 target=_early_close_server,
738 args=(server_sock, response_code, resp_body),
739 daemon=True,
740 )
741 t.start()
742
743 try:
744 url = f"http://127.0.0.1:{port}/owner/repo/push"
745 transport = HttpTransport()
746 req = transport._build_request("POST", url, None, body, "application/x-msgpack")
747 with pytest.raises(TransportError):
748 transport._execute(req)
749 finally:
750 t.join(timeout=2)
751 if original is not None and hasattr(signal, "SIGPIPE"):
752 signal.signal(signal.SIGPIPE, original)
753
754 def test_sigpipe_not_fatal_409_large_body(self) -> None:
755 """20 MB body + server closes after headers → TransportError, not exit 141."""
756 self._run_early_close_scenario(payload_mb=20.0, response_code=409)
757
758 def test_sigpipe_not_fatal_401_large_body(self) -> None:
759 """15 MB body + server sends 401 early → TransportError, not SIGPIPE crash."""
760 self._run_early_close_scenario(payload_mb=15.0, response_code=401)
761
762 def test_sigpipe_not_fatal_500_large_body(self) -> None:
763 """10 MB body + server crashes (500) → TransportError, not SIGPIPE crash."""
764 self._run_early_close_scenario(payload_mb=10.0, response_code=500)
File History 2 commits
sha256:79ffe87f5fe2ec146e35f05521218bbf54dffdb0440c07f970bad05f16efb89f chore: merge main — carry all urllib/typing/test fixes from dev Sonnet 4.6 minor ⚠ 47 days ago
sha256:0bea7600d1eee83e87950be49933b1006fa9dc2c71e7c4ee748d324f61138156 chore: bump version to 0.2.0rc11; fix typing audit violatio… Sonnet 4.6 minor ⚠ 47 days ago