331 lines
16 KiB
Python
331 lines
16 KiB
Python
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()
|