import gzip import hashlib import io import tarfile import unittest from pathlib import Path from unittest import mock import sys # Add the opax-mcp directory to the path for local imports REPO_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(REPO_ROOT / "opax-mcp")) import deployment_policy import deployment_source REQUIRED_PATHS = ( "cloudbuild.deploy.yaml", "opax-mcp/Dockerfile", ) def create_test_archive(members): """Creates an in-memory .tar.gz archive from a list of member descriptions.""" out_buffer = io.BytesIO() with gzip.GzipFile(fileobj=out_buffer, mode='wb', mtime=0) as gz: with tarfile.open(fileobj=gz, mode='w:') as tar: for member_info in members: info = tarfile.TarInfo(name=member_info["name"]) info.mtime = member_info.get("mtime", 12345) info.uid = member_info.get("uid", 1000) info.gid = member_info.get("gid", 1000) info.uname = member_info.get("uname", "testuser") info.gname = member_info.get("gname", "testgroup") typ = member_info["type"] if typ == "file": content = member_info.get("content", b"") info.type = tarfile.REGTYPE info.size = len(content) info.mode = member_info.get("mode", 0o644) tar.addfile(info, io.BytesIO(content)) elif typ == "dir": info.type = tarfile.DIRTYPE info.mode = member_info.get("mode", 0o755) tar.addfile(info) elif typ == "symlink": info.type = tarfile.SYMTYPE info.linkname = member_info.get("linkname", "target") info.mode = 0o777 tar.addfile(info) elif typ == 'hardlink': info.type = tarfile.LNKTYPE info.linkname = member_info.get("linkname", "target") tar.addfile(info) elif typ == 'char': info.type = tarfile.CHRTYPE tar.addfile(info) elif typ == 'block': info.type = tarfile.BLKTYPE tar.addfile(info) elif typ == 'fifo': info.type = tarfile.FIFOTYPE tar.addfile(info) return out_buffer.getvalue() VALID_ARCHIVE_MEMBERS = [ {"name": "wrapper-root/", "type": "dir"}, {"name": "wrapper-root/cloudbuild.deploy.yaml", "type": "file", "content": b"steps: []"}, {"name": "wrapper-root/opax-mcp/", "type": "dir"}, {"name": "wrapper-root/opax-mcp/Dockerfile", "type": "file", "content": b"FROM scratch"}, {"name": "wrapper-root/opax-mcp/server.py", "type": "file", "content": b"print('hello')"}, {"name": "wrapper-root/README.md", "type": "file", "content": b"readme"}, ] class TestDeploymentPolicy(unittest.TestCase): """Tests for the deployment_policy module.""" def test_get_known_target_succeeds(self): target = deployment_policy.get_deployment_target("opax-mcp") self.assertEqual(target["repository"], "chris/OSVauco") self.assertEqual(target["build_config"], "cloudbuild.deploy.yaml") self.assertIn("cloudbuild.deploy.yaml", target["required_source_paths"]) def test_rejects_unknown_key(self): with self.assertRaises(ValueError): deployment_policy.get_deployment_target("unknown-service") def test_rejects_invalid_keys(self): for key in [None, "", " ", 123]: with self.subTest(key=key): with self.assertRaises(ValueError): deployment_policy.get_deployment_target(key) def test_returned_dict_is_a_copy(self): target = deployment_policy.get_deployment_target("opax-mcp") target["repository"] = "mutated" original = deployment_policy.DEPLOYMENT_TARGETS["opax-mcp"]["repository"] self.assertEqual(original, "chris/OSVauco") class TestNormalizeGiteaArchive(unittest.TestCase): """Tests for deployment_source.normalize_gitea_archive.""" def _assert_raises_with_message(self, message, func, *args, **kwargs): with self.assertRaises(deployment_source.SourceArtifactError) as cm: func(*args, **kwargs) self.assertEqual(cm.exception.args[0], message) def test_wrapper_root_archive_succeeds(self): archive = create_test_archive(VALID_ARCHIVE_MEMBERS) _, manifest = deployment_source.normalize_gitea_archive( archive, required_paths=REQUIRED_PATHS ) self.assertTrue(manifest["wrapper_directory_stripped"]) def test_already_rooted_archive_succeeds(self): members = [ {"name": "cloudbuild.deploy.yaml", "type": "file", "content": b"steps: []"}, {"name": "opax-mcp/Dockerfile", "type": "file", "content": b"FROM scratch"}, ] archive = create_test_archive(members) _, manifest = deployment_source.normalize_gitea_archive( archive, required_paths=REQUIRED_PATHS ) self.assertFalse(manifest["wrapper_directory_stripped"]) def test_deterministic_output(self): archive1 = create_test_archive(VALID_ARCHIVE_MEMBERS) members2 = [ {"name": "wrapper-root/", "type": "dir", "uid": 4000, "gid": 5000, "uname": "dir-user", "gname": "dir-group", "mtime": 1, "mode": 0o700}, {"name": "wrapper-root/cloudbuild.deploy.yaml", "type": "file", "content": b"steps: []", "uid": 2000, "gid": 3000, "uname": "other-user", "gname": "other-group", "mtime": 9999, "mode": 0o600}, {"name": "wrapper-root/opax-mcp/", "type": "dir", "uid": 4000, "gid": 5000, "uname": "dir-user", "gname": "dir-group", "mtime": 1, "mode": 0o700}, {"name": "wrapper-root/opax-mcp/Dockerfile", "type": "file", "content": b"FROM scratch", "uid": 2000, "gid": 3000, "uname": "other-user", "gname": "other-group", "mtime": 9999, "mode": 0o600}, {"name": "wrapper-root/opax-mcp/server.py", "type": "file", "content": b"print('hello')", "uid": 2000, "gid": 3000, "uname": "other-user", "gname": "other-group", "mtime": 9999, "mode": 0o600}, {"name": "wrapper-root/README.md", "type": "file", "content": b"readme", "uid": 2000, "gid": 3000, "uname": "other-user", "gname": "other-group", "mtime": 9999, "mode": 0o600}, ] archive2 = create_test_archive(members2) norm_bytes1, manifest1 = deployment_source.normalize_gitea_archive( archive1, required_paths=REQUIRED_PATHS ) norm_bytes2, manifest2 = deployment_source.normalize_gitea_archive( archive2, required_paths=REQUIRED_PATHS ) self.assertEqual(norm_bytes1, norm_bytes2) self.assertEqual(manifest1["sha256"], manifest2["sha256"]) self.assertEqual(manifest1["source_bytes"], len(norm_bytes1)) self.assertEqual(manifest1["sha256"], hashlib.sha256(norm_bytes1).hexdigest()) def test_output_metadata_is_normalized(self): archive = create_test_archive(VALID_ARCHIVE_MEMBERS) norm_bytes, _ = deployment_source.normalize_gitea_archive( archive, required_paths=REQUIRED_PATHS ) with gzip.GzipFile(fileobj=io.BytesIO(norm_bytes), mode="rb") as gz: with tarfile.open(fileobj=gz, mode="r:") as tar: members = tar.getmembers() self.assertTrue(len(members) > 0) for member in members: self.assertTrue(member.isreg()) self.assertEqual(member.uid, 0) self.assertEqual(member.gid, 0) self.assertEqual(member.uname, "") self.assertEqual(member.gname, "") self.assertEqual(member.mtime, 0) self.assertEqual(member.mode, 0o644) def test_invalid_source_archive_rejections(self): message = "Invalid source archive." valid_archive = create_test_archive(VALID_ARCHIVE_MEMBERS) test_cases = { "none_input": (None, REQUIRED_PATHS), "string_input": ("not bytes", REQUIRED_PATHS), "empty_bytes": (b"", REQUIRED_PATHS), "malformed_gzip": (b"not a gzip file", REQUIRED_PATHS), "invalid_tar": (gzip.compress(b"this is not a tar file"), REQUIRED_PATHS), "invalid_req_paths_type": (valid_archive, ["list"]), "empty_req_paths_tuple": (valid_archive, tuple()), "duplicate_req_paths": (valid_archive, ("a", "a")), "unsafe_req_path": (valid_archive, ("../unsafe",)), } for name, (archive, req_paths) in test_cases.items(): with self.subTest(name=name): self._assert_raises_with_message( message, deployment_source.normalize_gitea_archive, archive, required_paths=req_paths ) def test_missing_build_files_rejections(self): message = "Source archive is missing required build files." members = [{"name": "wrapper-root/README.md", "type": "file", "content": b"readme"}] archive = create_test_archive(members) self._assert_raises_with_message( message, deployment_source.normalize_gitea_archive, archive, required_paths=REQUIRED_PATHS ) archive_no_files = create_test_archive([{"name": "wrapper-root/", "type": "dir"}]) self._assert_raises_with_message( message, deployment_source.normalize_gitea_archive, archive_no_files, required_paths=REQUIRED_PATHS ) def test_validate_archive_member_path_directly(self): """Layer A: Direct helper tests for raw unsafe paths.""" message = "Source archive has unsafe paths." unsafe_paths = [ "", "/etc/passwd", "../secret", "wrapper/docs/../secret", "./README.md", "wrapper//README.md", "wrapper\\README.md", "wrapper/\0README.md" ] for path in unsafe_paths: with self.subTest(path=path): with self.assertRaises(deployment_source.SourceArtifactError) as cm: deployment_source._validate_archive_member_path(path, is_directory=False) self.assertEqual(cm.exception.args[0], message) def test_archive_level_unsafe_path_rejections(self): """Layer B: Archive-level tests for serializable unsafe paths.""" message = "Source archive has unsafe paths." base_files = [ {"name": "root-marker.txt", "type": "file", "content": b"marker"}, {"name": "a/b", "type": "file", "content": b"content"}, ] unsafe_paths = [ "/etc/passwd", "../secret", "wrapper/docs/../secret", "./README.md", "wrapper//README.md", "wrapper\\README.md" ] for path in unsafe_paths: with self.subTest(path=path): archive = create_test_archive(base_files + [{"name": path, "type": "file"}]) self._assert_raises_with_message( message, deployment_source.normalize_gitea_archive, archive, required_paths=("a/b",) ) def test_ustar_path_limit_rejections(self): message = "Source archive has unsafe paths." base_files = [ {"name": "root-marker.txt", "type": "file", "content": b"marker"}, {"name": "a/b", "type": "file", "content": b"content"}, ] long_paths = { "filename_>_100": "a/" + ("b" * 101), "prefix_>_155": ("a/" * 79) + "file.txt", "total_>_255": ("a/" * 127) + "file.txt", } for name, path in long_paths.items(): with self.subTest(name=name): archive = create_test_archive(base_files + [{"name": path, "type": "file", "content": b""}]) self._assert_raises_with_message( message, deployment_source.normalize_gitea_archive, archive, required_paths=("a/b",) ) def test_unsupported_member_rejections(self): message = "Source archive contains unsupported entries." base = {"name": "wrapper/ok", "type": "file", "content": b"ok"} unsafe_members = [ {"name": "w/symlink", "type": "symlink"}, {"name": "w/hardlink", "type": "hardlink"}, {"name": "w/char", "type": "char"}, {"name": "w/block", "type": "block"}, {"name": "w/fifo", "type": "fifo"}, ] for member in unsafe_members: with self.subTest(type=member["type"]): archive = create_test_archive([base, member]) self._assert_raises_with_message( message, deployment_source.normalize_gitea_archive, archive, required_paths=("wrapper/ok",) ) def test_negative_member_size_rejection(self): message = "Invalid source archive." archive = create_test_archive([{"name": "wrapper/ok", "type": "file", "content": b"ok"}]) fake_member = mock.MagicMock() fake_member.name = "wrapper/ok" fake_member.size = -1 fake_member.isreg.return_value = True fake_member.isdir.return_value = False fake_member.issym.return_value = False fake_member.islnk.return_value = False fake_member.ischr.return_value = False fake_member.isblk.return_value = False fake_member.isfifo.return_value = False with mock.patch.object(tarfile.TarFile, "getmembers", return_value=[fake_member]): self._assert_raises_with_message( message, deployment_source.normalize_gitea_archive, archive, required_paths=("ok",) ) def test_limits_rejections(self): message = "Source archive exceeds allowed limits." with mock.patch.object(deployment_source, 'MAX_SOURCE_ARCHIVE_MEMBERS', 5): members = [{"name": f"f{i}", "type": "file"} for i in range(6)] archive = create_test_archive(members) self._assert_raises_with_message( message, deployment_source.normalize_gitea_archive, archive, required_paths=("f0",) ) with mock.patch.object(deployment_source, 'MAX_NORMALIZED_SOURCE_BYTES', 100): members = [{"name": "f1", "type": "file", "content": b"a"*50}, {"name": "f2", "type": "file", "content": b"b"*51}] archive = create_test_archive(members) self._assert_raises_with_message( message, deployment_source.normalize_gitea_archive, archive, required_paths=("f1",) ) def test_duplicate_path_rejection(self): message = "Source archive has duplicate paths." members = [ {"name": "wrapper/duplicate.txt", "type": "file", "content": b"1"}, {"name": "wrapper/duplicate.txt", "type": "file", "content": b"2"}, ] archive = create_test_archive(members) self._assert_raises_with_message( message, deployment_source.normalize_gitea_archive, archive, required_paths=("duplicate.txt",) ) def test_extraction_failures(self): message = "Invalid source archive." archive = create_test_archive(VALID_ARCHIVE_MEMBERS) with mock.patch("tarfile.TarFile.extractfile", return_value=None): self._assert_raises_with_message( message, deployment_source.normalize_gitea_archive, archive, required_paths=REQUIRED_PATHS ) mock_file = io.BytesIO(b"not the original content") with mock.patch("tarfile.TarFile.extractfile", return_value=mock_file): self._assert_raises_with_message( message, deployment_source.normalize_gitea_archive, archive, required_paths=REQUIRED_PATHS ) if __name__ == "__main__": unittest.main()