"""Pure temporary/mocked protocol tests: never SSH, curl, GPU, or real Git.""" import contextlib import hashlib import io import json from pathlib import Path import subprocess import tempfile import unittest from unittest.mock import patch import lan_lfs_transfer as M BASE = "https://" + M.HOST + M.PREFIX TOKEN = "synthetic-fixture-" + "credential-value" class AdapterTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) / "clone" self.git = self.root / ".git" self.git.mkdir(parents=True) self.data = b"synthetic content\x00\xff\n" self.oid = hashlib.sha256(self.data).hexdigest() self.size = len(self.data) self.events = [] self.agent = object.__new__(M.Agent) self.agent.root, self.agent.git_dir = self.root, self.git self.agent.operation, self.agent.emit = "upload", self.events.append self.source = self.git / "lfs" / "objects" / self.oid[:2] / self.oid[2:4] / self.oid self.source.parent.mkdir(parents=True) self.source.write_bytes(self.data) def request(self, operation="upload", **extra): value = {"event": operation, "oid": self.oid, "size": self.size, "action": None} if operation == "upload": value["path"] = str(self.source) value.update(extra) return value def init_message(self, **extra): return dict(event="init", operation="upload", remote="origin", concurrent=True, concurrenttransfers=2, **extra) def action(self, suffix="objects/" ): return {"href": BASE + "/" + suffix + self.oid, "header": {"Authorization": "Bearer " + TOKEN}} def test_strict_url_policy(self): self.assertEqual(M.safe_url(BASE + "/objects/" + self.oid + "?signed=opaque"), BASE + "/objects/" + self.oid + "?signed=opaque") invalid = [BASE.replace("https:", "http:"), BASE.replace(M.HOST, M.IP), BASE.replace(M.HOST, "other.example"), BASE.replace(M.HOST, M.HOST + ":444"), BASE.replace(M.HOST, "user:opaque@" + M.HOST), BASE + "-other/objects/x", BASE + "/../other", BASE + "/%2e%2e/other", BASE + "/objects/x#fragment", BASE + "//objects/x", BASE + "/objects\\x", BASE + "\n"] for url in invalid: with self.subTest(url=url), self.assertRaises(M.TransferError): M.safe_url(url) def test_headers_reject_injection_and_routing_overrides(self): for headers in ({"Authorization": "x\r\nHost: bad"}, {"Host": "bad"}, {"Authorization": "a", "authorization": "b"}, {"Range": "bytes=1-"}, {"bad\nkey": "x"}, {"X": "\x00"}): with self.subTest(headers=headers), self.assertRaises(M.TransferError): M.safe_headers(headers) def test_init_checks_origin_and_isolated_git_directory(self): values = [str(self.root), M.ORIGIN, str(self.git)] with patch.object(M, "git_output", side_effect=values): actual = M.Agent(self.init_message(), self.events.append, cwd=self.root) self.assertEqual(actual.git_dir, self.git) with patch.object(M, "git_output", side_effect=[str(self.root), "ssh://other/repo"]): with self.assertRaisesRegex(M.TransferError, "origin-refused"): M.Agent(self.init_message(), self.events.append, cwd=self.root) def test_original_source_and_excess_concurrency_refused(self): with patch.object(M, "SOURCE", self.root): with self.assertRaisesRegex(M.TransferError, "scientific-source-cwd"): M.Agent(self.init_message(), self.events.append, cwd=self.root) message = self.init_message() message["concurrenttransfers"] = 3 with self.assertRaisesRegex(M.TransferError, "concurrency"): M.Agent(message, self.events.append, cwd=self.root) def test_ssh_authentication_is_strict_and_operation_scoped(self): response = {"href": BASE, "header": {"Authorization": "Bearer " + TOKEN}} result = subprocess.CompletedProcess([], 0, stdout=json.dumps(response).encode(), stderr=b"") with patch.object(M.subprocess, "run", return_value=result) as call: href, headers = self.agent.authenticate() args = call.call_args.args[0] self.assertIn("BatchMode=yes", args) self.assertIn("StrictHostKeyChecking=yes", args) self.assertEqual(args[-3:], ["git-lfs-authenticate", "qishizhe/digital-drosophila.git", "upload"]) self.assertEqual(href, BASE) self.assertIn(TOKEN, headers["Authorization"]) self.assertNotIn(TOKEN, json.dumps(args)) def test_curl_secrets_only_in_stdin_and_secure_defaults(self): result = subprocess.CompletedProcess([], 0, stdout=b"{}\n200", stderr=b"") with patch.object(M.subprocess, "run", return_value=result) as call: body = self.agent.curl("PUT", BASE + "/objects/" + self.oid + "?signature=" + TOKEN, {"Authorization": "Bearer " + TOKEN}, upload=self.source) self.assertEqual(body, b"{}") self.assertEqual(call.call_args.args[0], ["curl", "-q", "--config", "-"]) self.assertNotIn(TOKEN, json.dumps(call.call_args.args[0])) config = call.call_args.kwargs["input"].decode() for wanted in ('noproxy = "*"', M.HOST + ":443:" + M.IP, 'proto = "=https"', "max-redirs = 0", "Accept: application/vnd.git-lfs+json", "Content-Type: application/octet-stream", TOKEN): self.assertIn(wanted, config) self.assertNotIn("insecure", config) self.assertNotIn("location", config) self.assertNotIn("retry", config) def test_redirect_or_curl_error_is_sanitized(self): for rc, output in ((0, b"redirect body\n302"), (22, b"")): result = subprocess.CompletedProcess([], rc, stdout=output, stderr=TOKEN.encode()) with patch.object(M.subprocess, "run", return_value=result): with self.assertRaises(M.TransferError) as caught: self.agent.curl("GET", BASE + "/objects/" + self.oid, {}) self.assertNotIn(TOKEN, str(caught.exception)) def test_post_json_sets_both_gitea_headers(self): result = subprocess.CompletedProcess([], 0, stdout=b"\n204", stderr=b"") with patch.object(M.subprocess, "run", return_value=result) as call: self.agent.curl("POST", BASE + "/verify", {}, body={"oid": self.oid, "size": self.size}) config = call.call_args.kwargs["input"].decode() self.assertIn("Accept: application/vnd.git-lfs+json", config) self.assertIn("Content-Type: application/vnd.git-lfs+json", config) def test_batch_requests_basic_and_checks_exact_identity(self): response = {"objects": [{"oid": self.oid, "size": self.size, "actions": {"upload": self.action()}}]} with patch.object(self.agent, "authenticate", return_value=(BASE, {"Authorization": TOKEN})), \ patch.object(self.agent, "curl", return_value=json.dumps(response).encode()) as call: self.assertIn("upload", self.agent.batch(self.oid, self.size)) args, kwargs = call.call_args self.assertEqual(args[1], BASE + "/objects/batch") self.assertEqual(args[2]["Accept"], "application/vnd.git-lfs+json") self.assertEqual(kwargs["body"]["transfers"], ["basic"]) response["objects"][0]["size"] += 1 with patch.object(self.agent, "authenticate", return_value=(BASE, {})), \ patch.object(self.agent, "curl", return_value=json.dumps(response).encode()): with self.assertRaisesRegex(M.TransferError, "identity"): self.agent.batch(self.oid, self.size) def test_batch_action_host_rejected(self): action = self.action() action["href"] = action["href"].replace(M.HOST, "other.example") response = {"objects": [{"oid": self.oid, "size": self.size, "actions": {"upload": action}}]} with patch.object(self.agent, "authenticate", return_value=(BASE, {})), \ patch.object(self.agent, "curl", return_value=json.dumps(response).encode()): with self.assertRaisesRegex(M.TransferError, "authority"): self.agent.batch(self.oid, self.size) def test_upload_chunked_exact_value_allowed_in_action_and_put(self): action = self.action() action["header"]["Transfer-Encoding"] = "chunked" response = {"objects": [{"oid": self.oid, "size": self.size, "actions": {"upload": action}}]} with patch.object(self.agent, "authenticate", return_value=(BASE, {})), \ patch.object(self.agent, "curl", return_value=json.dumps(response).encode()): self.assertEqual(self.agent.batch(self.oid, self.size)["upload"]["header"]["Transfer-Encoding"], "chunked") result = subprocess.CompletedProcess([], 0, stdout=b"\n200", stderr=b"") with patch.object(M.subprocess, "run", return_value=result) as call: self.agent.curl("PUT", action["href"], action["header"], upload=self.source) config = call.call_args.kwargs["input"].decode() self.assertIn('header = "Transfer-Encoding: chunked"', config) self.assertIn('request = "PUT"', config) self.assertIn("upload-file = ", config) def test_chunked_forbidden_outside_upload_or_for_other_values(self): for value in ("gzip", "CHUNKED", "chunked, gzip", "chunked\r\nX: y", " chunked"): with self.subTest(value=value), self.assertRaises(M.TransferError): self.agent.curl("PUT", BASE + "/objects/" + self.oid, {"Transfer-Encoding": value}, upload=self.source) for method, kwargs in (("GET", {}), ("POST", {"body": {}}), ("PUT", {})): with self.subTest(method=method, kwargs=kwargs), self.assertRaises(M.TransferError): self.agent.curl(method, BASE + "/objects/" + self.oid, {"Transfer-Encoding": "chunked"}, **kwargs) with self.assertRaises(M.TransferError): M.safe_headers({"Transfer-Encoding": "chunked"}) for operation, name in (("upload", "verify"), ("download", "download")): self.agent.operation = operation action = self.action() action["header"]["Transfer-Encoding"] = "chunked" response = {"objects": [{"oid": self.oid, "size": self.size, "actions": {name: action}}]} with patch.object(self.agent, "authenticate", return_value=(BASE, {})), \ patch.object(self.agent, "curl", return_value=json.dumps(response).encode()): with self.subTest(operation=operation, action=name), self.assertRaises(M.TransferError): self.agent.batch(self.oid, self.size) def test_upload_existing_object_needs_no_put(self): with patch.object(self.agent, "batch", return_value={}) as batch, patch.object(self.agent, "curl") as curl: result = self.agent.transfer(self.request()) self.assertEqual(result, {"event": "complete", "oid": self.oid}) batch.assert_called_once() curl.assert_not_called() self.assertEqual(self.events[-1]["bytesSoFar"], self.size) def test_upload_put_then_verification_post(self): actions = {"upload": self.action(), "verify": self.action("verify/")} with patch.object(self.agent, "batch", return_value=actions), \ patch.object(self.agent, "curl", return_value=b"") as curl: self.agent.transfer(self.request()) self.assertEqual([c.args[0] for c in curl.call_args_list], ["PUT", "POST"]) self.assertEqual(curl.call_args_list[1].kwargs["body"], {"oid": self.oid, "size": self.size}) def test_upload_wrong_sha_fails_before_network(self): self.source.write_bytes(b"x" * self.size) with patch.object(self.agent, "batch") as batch: with self.assertRaisesRegex(M.TransferError, "sha"): self.agent.transfer(self.request()) batch.assert_not_called() def test_upload_outside_lfs_store_refused(self): other = self.root / self.oid other.write_bytes(self.data) with self.assertRaisesRegex(M.TransferError, "outside-current-lfs"): self.agent.transfer(self.request(path=str(other))) def test_upload_source_mutation_detected(self): def change(*args, **kwargs): self.source.write_bytes(b"changed") with patch.object(self.agent, "batch", return_value={"upload": self.action()}), \ patch.object(self.agent, "curl", side_effect=change): with self.assertRaisesRegex(M.TransferError, "upload-source-changed"): self.agent.transfer(self.request()) def test_download_is_fresh_unique_and_verified(self): self.agent.operation = "download" def fetch(*args, **kwargs): kwargs["download"].write_bytes(self.data) return b"" with patch.object(self.agent, "batch", return_value={"download": self.action()}), \ patch.object(self.agent, "curl", side_effect=fetch) as curl: first = self.agent.transfer(self.request("download")) second = self.agent.transfer(self.request("download")) self.assertNotEqual(first["path"], second["path"]) self.assertEqual(Path(first["path"]).read_bytes(), self.data) self.assertEqual(curl.call_count, 2) self.assertEqual(self.source.read_bytes(), self.data) def test_download_hash_failure_retains_partial(self): self.agent.operation = "download" def fetch(*args, **kwargs): kwargs["download"].write_bytes(b"x" * self.size) with patch.object(self.agent, "batch", return_value={"download": self.action()}), \ patch.object(self.agent, "curl", side_effect=fetch): with self.assertRaisesRegex(M.TransferError, "sha"): self.agent.transfer(self.request("download")) partials = list((self.git / "lfs" / "lan-transfer-tmp").glob("*.partial")) self.assertEqual(len(partials), 1) self.assertEqual(partials[0].read_bytes(), b"x" * self.size) def test_protocol_init_progress_completion_terminate(self): requests = [self.init_message(), self.request(), {"event": "terminate"}] incoming = io.StringIO("".join(json.dumps(x) + "\n" for x in requests)) with patch.object(self.agent, "batch", return_value={}): code = M.serve(incoming, self.events.append, factory=lambda message, send: self.agent) self.assertEqual(code, 0) self.assertEqual(self.events[0], {}) self.assertEqual(self.events[-1], {"event": "complete", "oid": self.oid}) def test_transfer_error_continues_and_never_echoes_exception_secret(self): incoming = io.StringIO("\n".join(json.dumps(x) for x in [self.init_message(), self.request(), self.request(), {"event": "terminate"}]) + "\n") with patch.object(self.agent, "transfer", side_effect=[RuntimeError(TOKEN), {"event": "complete", "oid": self.oid}]): code = M.serve(incoming, self.events.append, factory=lambda message, send: self.agent) self.assertEqual(code, 0) self.assertIn("error", self.events[1]) self.assertNotIn(TOKEN, json.dumps(self.events)) self.assertNotIn("error", self.events[2]) def test_bad_initialization_uses_init_error_schema(self): incoming = io.StringIO(json.dumps(self.init_message()) + "\n") with patch.object(M, "git_output", side_effect=RuntimeError(TOKEN)): code = M.serve(incoming, self.events.append) self.assertEqual(code, 1) self.assertEqual(self.events[0]["error"]["code"], 32) self.assertNotIn(TOKEN, json.dumps(self.events)) if __name__ == "__main__": unittest.main()