From 33a414c0956b7b5df65291f8f4c96d856ee96033 Mon Sep 17 00:00:00 2001 From: Chris Christiansen Date: Sun, 20 Sep 2026 19:47:31 +0000 Subject: [PATCH] Add deterministic deployment source normalization --- opax-mcp/deployment_policy.py | 39 ++++ opax-mcp/deployment_source.py | 185 ++++++++++++++++ opax-mcp/test_deployment_source.py | 330 +++++++++++++++++++++++++++++ 3 files changed, 554 insertions(+) create mode 100644 opax-mcp/deployment_policy.py create mode 100644 opax-mcp/deployment_source.py create mode 100644 opax-mcp/test_deployment_source.py diff --git a/opax-mcp/deployment_policy.py b/opax-mcp/deployment_policy.py new file mode 100644 index 0000000..4f05b03 --- /dev/null +++ b/opax-mcp/deployment_policy.py @@ -0,0 +1,39 @@ +"""Deployment policy for OPAX-MCP.""" + +INITIAL_DEPLOYMENT_TARGET = "opax-mcp" + +DEPLOYMENT_TARGETS = { + "opax-mcp": { + "repository": "chris/OSVauco", + "region": "us-central1", + "cloud_run_service": "opax-mcp", + "build_config": "cloudbuild.deploy.yaml", + "required_source_paths": ( + "cloudbuild.deploy.yaml", + "opax-mcp/Dockerfile", + ), + }, +} + + +def get_deployment_target(service_key: str) -> dict: + """ + Retrieves a copy of the deployment target metadata for a given service key. + + Args: + service_key: The identifier for the service. + + Returns: + A copy of the deployment target dictionary. + + Raises: + ValueError: If the service_key is unknown or invalid. + """ + if not isinstance(service_key, str) or not service_key: + raise ValueError("Invalid service key.") + + target = DEPLOYMENT_TARGETS.get(service_key) + if not target: + raise ValueError(f"Unknown deployment target: {service_key}") + + return target.copy() diff --git a/opax-mcp/deployment_source.py b/opax-mcp/deployment_source.py new file mode 100644 index 0000000..58d1403 --- /dev/null +++ b/opax-mcp/deployment_source.py @@ -0,0 +1,185 @@ +import gzip +import hashlib +import io +import tarfile +import struct +from typing import List, Tuple + +class SourceArtifactError(ValueError): + """Custom exception for source artifact processing errors.""" + pass + +MAX_SOURCE_ARCHIVE_MEMBERS = 10_000 +MAX_NORMALIZED_SOURCE_BYTES = 209_715_200 + +_SAFE_MESSAGES = { + "INVALID": "Invalid source archive.", + "LIMITS": "Source archive exceeds allowed limits.", + "UNSUPPORTED": "Source archive contains unsupported entries.", + "UNSAFE_PATHS": "Source archive has unsafe paths.", + "DUPLICATE_PATHS": "Source archive has duplicate paths.", + "MISSING_FILES": "Source archive is missing required build files.", +} + +def _validate_archive_member_path( + name: str, + *, + is_directory: bool, +) -> str: + """ + Validates and normalizes a tar member path. + """ + if not isinstance(name, str) or not name: + raise SourceArtifactError(_SAFE_MESSAGES["UNSAFE_PATHS"]) + + if is_directory and name.endswith('/'): + name = name[:-1] + + if ( + not name + or '\\' in name + or '\0' in name + or name.startswith('/') + or '//' in name + ): + raise SourceArtifactError(_SAFE_MESSAGES["UNSAFE_PATHS"]) + + parts = name.split('/') + if any(p in ('.', '..') for p in parts) or not all(parts): + raise SourceArtifactError(_SAFE_MESSAGES["UNSAFE_PATHS"]) + return name + +def _validate_ustar_output_path(path: str) -> None: + """ + Validates that a normalized path is representable in USTAR format. + """ + _validate_archive_member_path(path, is_directory=False) + try: + path_bytes = path.encode('utf-8') + except UnicodeEncodeError: + raise SourceArtifactError(_SAFE_MESSAGES["UNSAFE_PATHS"]) from None + if len(path_bytes) > 255: + raise SourceArtifactError(_SAFE_MESSAGES["UNSAFE_PATHS"]) + + if b'/' in path_bytes: + prefix_bytes, name_bytes = path_bytes.rsplit(b'/', 1) + else: + prefix_bytes, name_bytes = b"", path_bytes + if not name_bytes: + raise SourceArtifactError(_SAFE_MESSAGES["UNSAFE_PATHS"]) + if len(name_bytes) > 100 or len(prefix_bytes) > 155: + raise SourceArtifactError(_SAFE_MESSAGES["UNSAFE_PATHS"]) + + +def normalize_gitea_archive( + archive_bytes: bytes, + *, + required_paths: tuple[str, ...], +) -> tuple[bytes, dict]: + """ + Normalizes a Gitea source archive for deterministic builds. + """ + if not isinstance(archive_bytes, bytes) or not archive_bytes: + raise SourceArtifactError(_SAFE_MESSAGES["INVALID"]) + + if not isinstance(required_paths, tuple) or not required_paths: + raise SourceArtifactError(_SAFE_MESSAGES["INVALID"]) + validated_req_paths = [] + try: + for p in required_paths: + if not isinstance(p, str) or not p: + raise SourceArtifactError(_SAFE_MESSAGES["INVALID"]) + validated_req_paths.append(_validate_archive_member_path(p, is_directory=False)) + except SourceArtifactError: + raise SourceArtifactError(_SAFE_MESSAGES["INVALID"]) from None + if len(validated_req_paths) != len(set(validated_req_paths)): + raise SourceArtifactError(_SAFE_MESSAGES["INVALID"]) + + final_files: List[Tuple[tarfile.TarInfo, str]] = [] + + try: + with gzip.GzipFile(fileobj=io.BytesIO(archive_bytes), mode="rb") as gzip_file: + with tarfile.open(fileobj=gzip_file, mode="r:") as tar: + members = tar.getmembers() + if len(members) > MAX_SOURCE_ARCHIVE_MEMBERS: + raise SourceArtifactError(_SAFE_MESSAGES["LIMITS"]) + + validated_members: List[Tuple[tarfile.TarInfo, str]] = [] + total_size = 0 + for member in members: + is_dir = member.isdir() + if member.isreg() or is_dir: + normalized_path = _validate_archive_member_path(member.name, is_directory=is_dir) + if member.isreg(): + if member.size < 0: + raise SourceArtifactError(_SAFE_MESSAGES["INVALID"]) + if total_size + member.size > MAX_NORMALIZED_SOURCE_BYTES: + raise SourceArtifactError(_SAFE_MESSAGES["LIMITS"]) + total_size += member.size + validated_members.append((member, normalized_path)) + elif member.issym() or member.islnk() or member.ischr() or member.isblk() or member.isfifo(): + raise SourceArtifactError(_SAFE_MESSAGES["UNSUPPORTED"]) + else: + raise SourceArtifactError(_SAFE_MESSAGES["UNSUPPORTED"]) + + regular_files = [(m, p) for m, p in validated_members if m.isreg()] + if not regular_files: + raise SourceArtifactError(_SAFE_MESSAGES["MISSING_FILES"]) + first_segments = {path.split('/')[0] for _, path in regular_files if '/' in path} + wrapper_dir_stripped = False + if len(first_segments) == 1 and all('/' in p for _, p in regular_files): + wrapper_dir = first_segments.pop() + temp_files: List[Tuple[tarfile.TarInfo, str]] = [] + for member, path in regular_files: + new_path = path.partition(f"{wrapper_dir}/")[2] + final_path = _validate_archive_member_path(new_path, is_directory=False) + temp_files.append((member, final_path)) + final_files = temp_files + wrapper_dir_stripped = True + else: + final_files = regular_files + final_paths = [path for _, path in final_files] + if len(final_paths) != len(set(final_paths)): + raise SourceArtifactError(_SAFE_MESSAGES["DUPLICATE_PATHS"]) + if not set(validated_req_paths).issubset(set(final_paths)): + raise SourceArtifactError(_SAFE_MESSAGES["MISSING_FILES"]) + + out_buffer = io.BytesIO() + with gzip.GzipFile(fileobj=out_buffer, mode='wb', mtime=0) as gz: + with tarfile.open(fileobj=gz, mode='w:', format=tarfile.USTAR_FORMAT) as out_tar: + for member, path in sorted(final_files, key=lambda item: item[1]): + _validate_ustar_output_path(path) + content_file = tar.extractfile(member) + if content_file is None: + raise SourceArtifactError(_SAFE_MESSAGES["INVALID"]) + content_bytes = content_file.read() + + if len(content_bytes) != member.size: + raise SourceArtifactError(_SAFE_MESSAGES["INVALID"]) + info = tarfile.TarInfo(name=path) + info.size = member.size + info.mtime = 0 + info.uid = 0 + info.gid = 0 + info.uname = "" + info.gname = "" + info.mode = 0o644 + out_tar.addfile(info, io.BytesIO(content_bytes)) + except SourceArtifactError: + raise + except (gzip.BadGzipFile, tarfile.TarError, EOFError, OSError, struct.error, ValueError): + raise SourceArtifactError(_SAFE_MESSAGES["INVALID"]) from None + except Exception: + raise SourceArtifactError(_SAFE_MESSAGES["INVALID"]) from None + + normalized_bytes = out_buffer.getvalue() + sha256_hash = hashlib.sha256(normalized_bytes).hexdigest() + + manifest = { + "sha256": sha256_hash, + "source_bytes": len(normalized_bytes), + "required_paths": list(validated_req_paths), + "wrapper_directory_stripped": wrapper_dir_stripped, + } + + return normalized_bytes, manifest diff --git a/opax-mcp/test_deployment_source.py b/opax-mcp/test_deployment_source.py new file mode 100644 index 0000000..1bd6b22 --- /dev/null +++ b/opax-mcp/test_deployment_source.py @@ -0,0 +1,330 @@ +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()