157 lines
4.9 KiB
Python
157 lines
4.9 KiB
Python
import os
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from scripts import build_mcp_hub_release_images as producer
|
|
|
|
|
|
def test_database_download_child_has_private_umask_without_changing_other_stages(tmp_path):
|
|
calls = []
|
|
|
|
def runner(command, **kwargs):
|
|
calls.append(kwargs)
|
|
return subprocess.CompletedProcess(command, 0)
|
|
|
|
for stage in ("trivy_db_download", "compose_local_build"):
|
|
producer._execute(["fixture"], env={}, cwd=tmp_path, timeout=1, stage=stage, runner=runner)
|
|
assert calls[0]["umask"] == 0o077
|
|
assert "umask" not in calls[1]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"fault",
|
|
[
|
|
None,
|
|
"owner",
|
|
"symlink",
|
|
"hardlink",
|
|
"oversize",
|
|
"parent",
|
|
"db_link",
|
|
"swap",
|
|
"parent_swap",
|
|
"after_swap",
|
|
],
|
|
)
|
|
def test_db_metadata_adapter_restricts_only_owned_anchored_inode(monkeypatch, fault):
|
|
def info(inode, mode, **extra):
|
|
fields = dict(
|
|
st_dev=1, st_ino=inode, st_mode=mode, st_uid=123, st_nlink=1, st_size=16, st_mtime_ns=1
|
|
)
|
|
fields.update(extra)
|
|
return SimpleNamespace(**fields)
|
|
|
|
parent = info(1, stat.S_IFDIR | (0o755 if fault == "parent" else 0o700))
|
|
db = info(2, stat.S_IFDIR | 0o755)
|
|
metadata = info(3, stat.S_IFREG | 0o644)
|
|
if fault == "owner":
|
|
metadata.st_uid = 999
|
|
if fault == "symlink":
|
|
metadata.st_mode = stat.S_IFLNK | 0o644
|
|
if fault == "hardlink":
|
|
metadata.st_nlink = 2
|
|
if fault == "oversize":
|
|
metadata.st_size = 65537
|
|
restricted = []
|
|
closed = []
|
|
opens = []
|
|
successful = []
|
|
cache = SimpleNamespace(lstat=lambda: parent)
|
|
|
|
def open_file(path, flags, *, dir_fd=None):
|
|
opens.append((path, dir_fd))
|
|
assert flags & 0x20000
|
|
if path is cache:
|
|
successful.append(10)
|
|
return 10
|
|
if path == "db":
|
|
assert dir_fd == 10
|
|
if fault == "db_link":
|
|
raise OSError("fixture-private-value")
|
|
successful.append(11)
|
|
return 11
|
|
assert path == "metadata.json" and dir_fd == 11
|
|
successful.append(12)
|
|
return 12
|
|
|
|
def fstat(fd):
|
|
if fd == 10:
|
|
return parent
|
|
if fd == 11:
|
|
return db
|
|
if fault == "swap":
|
|
return info(4, metadata.st_mode)
|
|
return SimpleNamespace(**vars(metadata))
|
|
|
|
def inspect(path, *, dir_fd, follow_symlinks):
|
|
assert follow_symlinks is False
|
|
if path == "db":
|
|
assert dir_fd == 10
|
|
return info(4, db.st_mode) if fault == "parent_swap" else db
|
|
assert path == "metadata.json" and dir_fd == 11
|
|
return info(4, metadata.st_mode) if fault == "after_swap" and restricted else metadata
|
|
|
|
def restrict(fd, mode):
|
|
assert fd == 12 and mode == 0o600
|
|
restricted.append((fd, mode))
|
|
metadata.st_mode = stat.S_IFREG | 0o600
|
|
|
|
monkeypatch.setattr(producer.platform, "system", lambda: "Linux")
|
|
monkeypatch.setattr(producer, "_effective_uid", lambda: 123)
|
|
monkeypatch.setattr(
|
|
producer,
|
|
"os",
|
|
SimpleNamespace(
|
|
O_RDONLY=os.O_RDONLY,
|
|
O_DIRECTORY=0x10000,
|
|
O_NOFOLLOW=0x20000,
|
|
open=open_file,
|
|
fstat=fstat,
|
|
stat=inspect,
|
|
fchmod=restrict,
|
|
close=closed.append,
|
|
),
|
|
)
|
|
if fault:
|
|
with pytest.raises(producer.BuildRejected) as error:
|
|
producer._private_db_metadata(cache)
|
|
assert "fixture-private-value" not in str(error.value)
|
|
if fault != "after_swap":
|
|
assert restricted == []
|
|
else:
|
|
producer._private_db_metadata(cache)
|
|
assert restricted == [(12, 0o600)]
|
|
assert closed == list(reversed(successful))
|
|
|
|
|
|
@pytest.mark.skipif(os.name != "posix", reason="file-descriptor permission adapter requires Linux")
|
|
def test_explicit_upstream_chmod_is_restricted_before_strict_json_reader(tmp_path):
|
|
cache = tmp_path / "cache"
|
|
db = cache / "db"
|
|
db.mkdir(parents=True)
|
|
cache.chmod(0o700)
|
|
db.chmod(0o755)
|
|
metadata = db / "metadata.json"
|
|
metadata.write_text('{"Version":2}')
|
|
metadata.chmod(0o644)
|
|
with pytest.raises(producer.BuildRejected, match="not private and bounded"):
|
|
producer._json_file(metadata, maximum=64 * 1024)
|
|
producer._private_db_metadata(cache)
|
|
assert metadata.stat().st_mode & 0o777 == 0o600
|
|
assert producer._json_file(metadata, maximum=64 * 1024) == {"Version": 2}
|
|
|
|
|
|
@pytest.mark.skipif(os.name != "posix", reason="kernel umask requires Linux")
|
|
def test_kernel_download_child_creates_private_files(tmp_path):
|
|
command = [
|
|
sys.executable,
|
|
"-c",
|
|
"from pathlib import Path; Path('metadata.json').write_text('{}')",
|
|
]
|
|
producer._execute(command, env=os.environ, cwd=tmp_path, timeout=10, stage="trivy_db_download")
|
|
assert (tmp_path / "metadata.json").stat().st_mode & 0o777 == 0o600
|