Add deterministic deployment source normalization
This commit is contained in:
parent
07626c83f4
commit
33a414c095
39
opax-mcp/deployment_policy.py
Normal file
39
opax-mcp/deployment_policy.py
Normal file
|
|
@ -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()
|
||||
185
opax-mcp/deployment_source.py
Normal file
185
opax-mcp/deployment_source.py
Normal file
|
|
@ -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
|
||||
330
opax-mcp/test_deployment_source.py
Normal file
330
opax-mcp/test_deployment_source.py
Normal file
|
|
@ -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()
|
||||
Loading…
Reference in New Issue
Block a user