fail loudly on an empty/truncated frozen harvest bundle. Tighten as much as we can the remote harvesting
This commit is contained in:
parent
a2dc054882
commit
ddb18403c8
GPG key ID:
03906B4110AAD3B8
4 changed files with 358 additions and 40 deletions
|
|
@ -354,11 +354,32 @@ def freeze_directory_bundle(
|
|||
pass
|
||||
|
||||
file_count = 0
|
||||
|
||||
def _on_walk_error(exc: OSError) -> None:
|
||||
# os.walk() defaults to *silently swallowing* directory-listing
|
||||
# errors (e.g. an unreadable subdirectory raises os.scandir() ->
|
||||
# PermissionError, which os.walk would otherwise drop). A swallowed
|
||||
# error produces a partial frozen tree with no indication that
|
||||
# content was omitted, which is exactly the "fail loudly instead of
|
||||
# producing a partial frozen copy" guarantee this helper is meant to
|
||||
# provide. Re-raise as an ArtifactSafetyError so the whole freeze
|
||||
# aborts rather than returning a silently-truncated bundle.
|
||||
raise ArtifactSafetyError(
|
||||
f"{label} could not be fully read while freezing "
|
||||
f"({exc.__class__.__name__}: {exc}); refusing to produce a "
|
||||
f"partial frozen copy"
|
||||
)
|
||||
|
||||
# followlinks=False: do not descend into symlinked directories. Each
|
||||
# discovered file is independently re-opened no-follow before copying, so
|
||||
# a symlinked directory cannot smuggle content into the frozen tree even
|
||||
# if it is swapped in mid-walk.
|
||||
for cur, dirs, files in os.walk(src_root, followlinks=False):
|
||||
#
|
||||
# onerror=_on_walk_error: fail closed on any unreadable directory rather
|
||||
# than silently skipping it (see callback above).
|
||||
for cur, dirs, files in os.walk(
|
||||
src_root, followlinks=False, onerror=_on_walk_error
|
||||
):
|
||||
cur_p = Path(cur)
|
||||
|
||||
# Refuse symlinked subdirectories rather than silently skipping them,
|
||||
|
|
|
|||
158
enroll/remote.py
158
enroll/remote.py
|
|
@ -1,6 +1,7 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import getpass
|
||||
import hashlib
|
||||
import os
|
||||
import shlex
|
||||
import shutil
|
||||
|
|
@ -232,11 +233,21 @@ def _safe_extract_tar(tar: tarfile.TarFile, dest: Path) -> None:
|
|||
tar.extract(m, path=dest)
|
||||
|
||||
|
||||
def _build_enroll_pyz(tmpdir: Path) -> Path:
|
||||
def _build_enroll_pyz(tmpdir: Path) -> tuple[Path, str]:
|
||||
"""Build a self-contained enroll zipapp (pyz) on the local machine.
|
||||
|
||||
The resulting file is stdlib-only and can be executed on the remote host
|
||||
as long as it has Python 3 available.
|
||||
|
||||
Returns ``(pyz_path, sha256_hex)``. The digest is computed on the exact
|
||||
bytes written locally so the caller can verify, on the remote side, that the
|
||||
file that is about to be executed as root is byte-for-byte the one we built
|
||||
(see ``_remote_verify_pyz_sha256``). This is transport/staging integrity
|
||||
defence-in-depth: it detects a swap of the staged file between upload and
|
||||
execution by anyone who gained write access to the staging directory. It is
|
||||
NOT a defence against a remote host that is already root-compromised -- such
|
||||
a host can subvert the interpreter regardless, and is out of scope per
|
||||
SECURITY.md.
|
||||
"""
|
||||
import enroll as pkg
|
||||
|
||||
|
|
@ -244,15 +255,58 @@ def _build_enroll_pyz(tmpdir: Path) -> Path:
|
|||
stage = tmpdir / "stage"
|
||||
(stage / "enroll").mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def _ignore(d: str, names: list[str]) -> set[str]:
|
||||
return {
|
||||
n
|
||||
for n in names
|
||||
if n in {"__pycache__", ".pytest_cache"} or n.endswith(".pyc")
|
||||
}
|
||||
# Names that must never end up in the remote zipapp. The remote only ever
|
||||
# runs ``harvest``; test suites, caches, editor/VCS scratch, and compiled
|
||||
# artifacts are never needed at runtime and should not be shipped to (or
|
||||
# executed on) a harvested host. Excluding them keeps the payload minimal
|
||||
# and avoids transferring irrelevant code to every target.
|
||||
_ignore_dirs = {
|
||||
"__pycache__",
|
||||
".pytest_cache",
|
||||
".mypy_cache",
|
||||
".ruff_cache",
|
||||
".git",
|
||||
".hg",
|
||||
".svn",
|
||||
"tests",
|
||||
"test",
|
||||
}
|
||||
_ignore_suffixes = (".pyc", ".pyo", ".orig", ".rej", ".bak")
|
||||
|
||||
def _ignore(directory: str, names: list[str]) -> set[str]:
|
||||
dropped: set[str] = set()
|
||||
for n in names:
|
||||
if n in _ignore_dirs:
|
||||
dropped.add(n)
|
||||
continue
|
||||
if n.endswith(_ignore_suffixes):
|
||||
dropped.add(n)
|
||||
continue
|
||||
# Defensive: never ship test modules even if they are colocated in
|
||||
# the package directory by a future packaging change.
|
||||
if n.startswith("test_") and n.endswith(".py"):
|
||||
dropped.add(n)
|
||||
continue
|
||||
if n == "conftest.py":
|
||||
dropped.add(n)
|
||||
continue
|
||||
return dropped
|
||||
|
||||
shutil.copytree(pkg_dir, stage / "enroll", dirs_exist_ok=True, ignore=_ignore)
|
||||
|
||||
# The JSON Schema is a required runtime data file for ``validate``/``manifest``
|
||||
# consumers of the harvest; the remote harvest itself does not validate, but
|
||||
# the bundle it produces is validated locally, and the schema travels with
|
||||
# the package. Fail loudly if a future ignore rule ever drops it rather than
|
||||
# silently shipping a package that cannot self-validate.
|
||||
staged_schema = stage / "enroll" / "schema" / "state.schema.json"
|
||||
if not staged_schema.is_file():
|
||||
raise RuntimeError(
|
||||
"internal error: enroll.pyz staging is missing the bundled JSON "
|
||||
"schema (schema/state.schema.json); refusing to build an "
|
||||
"incomplete remote payload"
|
||||
)
|
||||
|
||||
pyz_path = tmpdir / "enroll.pyz"
|
||||
zipapp.create_archive(
|
||||
stage,
|
||||
|
|
@ -260,7 +314,83 @@ def _build_enroll_pyz(tmpdir: Path) -> Path:
|
|||
main="enroll.cli:main",
|
||||
compressed=True,
|
||||
)
|
||||
return pyz_path
|
||||
|
||||
sha256_hex = _sha256_file(pyz_path)
|
||||
return pyz_path, sha256_hex
|
||||
|
||||
|
||||
def _sha256_file(path: Path) -> str:
|
||||
"""Return the hex SHA-256 of a file, read in chunks."""
|
||||
h = hashlib.sha256()
|
||||
with open(path, "rb") as f:
|
||||
for chunk in iter(lambda: f.read(1024 * 1024), b""):
|
||||
h.update(chunk)
|
||||
return h.hexdigest()
|
||||
|
||||
|
||||
def _remote_verify_pyz_sha256(
|
||||
ssh,
|
||||
remote_pyz_path: str,
|
||||
expected_sha256: str,
|
||||
*,
|
||||
remote_python: str,
|
||||
) -> None:
|
||||
"""Verify the uploaded zipapp's SHA-256 on the remote before executing it.
|
||||
|
||||
This is transport/staging integrity defence-in-depth. The check runs on the
|
||||
remote, immediately before the (root) execution of the zipapp, and fails
|
||||
closed if the digest does not match the bytes we built locally. It shrinks
|
||||
the window in which a *non-root* tamperer who somehow gained write access to
|
||||
the staging directory could swap the file between upload and execution.
|
||||
|
||||
It deliberately does NOT establish trust in a root-compromised remote: a
|
||||
host that is already root can forge any check it runs about itself. Per
|
||||
SECURITY.md, such a host is outside Enroll's threat model. The value here is
|
||||
catching accidental corruption and unprivileged-local-user staging races,
|
||||
not defeating a compromised root.
|
||||
|
||||
The hashing is done with Python's hashlib (already required to run the
|
||||
zipapp) rather than a ``sha256sum`` binary, so it does not depend on
|
||||
coreutils being present or on PATH resolution of a hashing tool.
|
||||
"""
|
||||
|
||||
# Hash the staged file using the same interpreter that will execute it.
|
||||
# ``-I`` isolates the interpreter from site/user customisation; the script
|
||||
# prints only the lowercase hex digest.
|
||||
hash_script = (
|
||||
"import hashlib,sys;"
|
||||
"h=hashlib.sha256();"
|
||||
"f=open(sys.argv[1],'rb');"
|
||||
"[h.update(c) for c in iter(lambda:f.read(1048576),b'')];"
|
||||
"sys.stdout.write(h.hexdigest())"
|
||||
)
|
||||
cmd = " ".join(
|
||||
shlex.quote(tok)
|
||||
for tok in (remote_python, "-I", "-c", hash_script, remote_pyz_path)
|
||||
)
|
||||
rc, out, err = _ssh_run(ssh, cmd, get_pty=False)
|
||||
if rc != 0:
|
||||
raise RuntimeError(
|
||||
"Failed to verify the integrity of the uploaded enroll.pyz on the "
|
||||
f"remote host.\nCommand: {cmd}\nExit code: {rc}\nStderr: {err.strip()}"
|
||||
)
|
||||
remote_digest = out.strip().lower()
|
||||
expected = expected_sha256.strip().lower()
|
||||
if not remote_digest:
|
||||
raise RuntimeError(
|
||||
"Remote integrity check returned an empty SHA-256 for enroll.pyz; "
|
||||
"refusing to execute it."
|
||||
)
|
||||
if remote_digest != expected:
|
||||
raise RuntimeError(
|
||||
"Integrity check failed for the uploaded enroll.pyz: the staged "
|
||||
"file's SHA-256 on the remote does not match the locally built "
|
||||
"payload. Refusing to execute it as root.\n"
|
||||
f" expected: {expected}\n"
|
||||
f" remote: {remote_digest}\n"
|
||||
"This can indicate the staging directory was tampered with between "
|
||||
"upload and execution, or transfer corruption."
|
||||
)
|
||||
|
||||
|
||||
def _ssh_run(
|
||||
|
|
@ -444,7 +574,7 @@ def _remote_harvest(
|
|||
# Build a zipapp locally and upload it to the remote.
|
||||
with tempfile.TemporaryDirectory(prefix="enroll-remote-") as td:
|
||||
td_path = Path(td)
|
||||
pyz = _build_enroll_pyz(td_path)
|
||||
pyz, pyz_sha256 = _build_enroll_pyz(td_path)
|
||||
local_tgz = td_path / "bundle.tgz"
|
||||
|
||||
ssh = paramiko.SSHClient()
|
||||
|
|
@ -595,6 +725,16 @@ def _remote_harvest(
|
|||
rapp = f"{rtmp}/enroll.pyz"
|
||||
sftp.put(str(pyz), rapp)
|
||||
|
||||
# Before executing the uploaded zipapp (as root, under sudo), verify
|
||||
# on the remote that the staged bytes match what we built locally.
|
||||
# This is staging/transport integrity defence-in-depth: it fails
|
||||
# closed if the file was swapped or corrupted between upload and
|
||||
# execution. It does not (and cannot) defend against a remote that
|
||||
# is already root-compromised; see _remote_verify_pyz_sha256.
|
||||
_remote_verify_pyz_sha256(
|
||||
ssh, rapp, pyz_sha256, remote_python=remote_python
|
||||
)
|
||||
|
||||
if not no_sudo:
|
||||
# The remote zipapp is staged as the SSH user, but the harvest
|
||||
# itself runs as root. Root must not write its bundle under the
|
||||
|
|
|
|||
|
|
@ -213,3 +213,25 @@ def test_freeze_directory_bundle_rejects_hardlinked_file(tmp_path: Path):
|
|||
os.link(secret, hard)
|
||||
with pytest.raises(ArtifactSafetyError, match="hardlink"):
|
||||
freeze_directory_bundle(bundle)
|
||||
|
||||
|
||||
def test_freeze_directory_bundle_fails_loudly_on_unreadable_subdir(tmp_path: Path):
|
||||
"""An unreadable subtree must abort the freeze, not silently truncate it.
|
||||
|
||||
Regression test: os.walk() defaults to swallowing directory-listing errors,
|
||||
which previously produced a partial frozen bundle with no error. The freeze
|
||||
now passes onerror= so a PermissionError aborts with ArtifactSafetyError.
|
||||
"""
|
||||
if os.geteuid() == 0:
|
||||
pytest.skip("root can read 0000 directories; cannot simulate the case")
|
||||
|
||||
bundle = _write_bundle(tmp_path)
|
||||
locked = bundle / "artifacts" / "app" / "etc" / "app"
|
||||
assert locked.is_dir()
|
||||
os.chmod(locked, 0o000)
|
||||
try:
|
||||
with pytest.raises(ArtifactSafetyError, match="could not be fully read"):
|
||||
freeze_directory_bundle(bundle)
|
||||
finally:
|
||||
# Restore perms so tmp_path cleanup can remove the tree.
|
||||
os.chmod(locked, 0o755)
|
||||
|
|
|
|||
|
|
@ -8,6 +8,30 @@ from pathlib import Path
|
|||
import pytest
|
||||
|
||||
|
||||
# The remote harvest now SHA-256-verifies the uploaded zipapp on the remote
|
||||
# before executing it. Tests mock _build_enroll_pyz to write these fixed bytes
|
||||
# and return (path, _FAKE_PYZ_SHA256); the fake SSH routers below answer the
|
||||
# verification command with the same digest so the happy paths proceed.
|
||||
_FAKE_PYZ_BYTES = b"PYZ"
|
||||
_FAKE_PYZ_SHA256 = "d6f4e1dbf7ba69af6c798c6f6f67383c978e68f4201bf31902275ef37e6263e1"
|
||||
|
||||
|
||||
def _fake_build_enroll_pyz(td) -> tuple[Path, str]:
|
||||
"""Stand-in for remote._build_enroll_pyz used across remote tests.
|
||||
|
||||
Writes a tiny fixed payload instead of a real zipapp and returns the
|
||||
(path, sha256) tuple the production builder now returns.
|
||||
"""
|
||||
p = Path(td) / "enroll.pyz"
|
||||
p.write_bytes(_FAKE_PYZ_BYTES)
|
||||
return p, _FAKE_PYZ_SHA256
|
||||
|
||||
|
||||
def _is_pyz_verify_cmd(cmd: str) -> bool:
|
||||
"""True if *cmd* is the remote pyz SHA-256 integrity check."""
|
||||
return "hashlib.sha256" in cmd and "enroll.pyz" in cmd
|
||||
|
||||
|
||||
def _make_tgz_bytes(files: dict[str, bytes]) -> bytes:
|
||||
bio = io.BytesIO()
|
||||
with tarfile.open(fileobj=bio, mode="w:gz") as tf:
|
||||
|
|
@ -56,12 +80,7 @@ def test_remote_harvest_happy_path(tmp_path: Path, monkeypatch):
|
|||
import enroll.remote as r
|
||||
|
||||
# Avoid building a real zipapp; just create a file.
|
||||
def fake_build(_td: Path) -> Path:
|
||||
p = _td / "enroll.pyz"
|
||||
p.write_bytes(b"PYZ")
|
||||
return p
|
||||
|
||||
monkeypatch.setattr(r, "_build_enroll_pyz", fake_build)
|
||||
monkeypatch.setattr(r, "_build_enroll_pyz", _fake_build_enroll_pyz)
|
||||
|
||||
# Prepare a tiny harvest bundle tar stream from the "remote".
|
||||
tgz = _make_tgz_bytes({"state.json": b'{"ok": true}\n'})
|
||||
|
|
@ -156,6 +175,9 @@ def test_remote_harvest_happy_path(tmp_path: Path, monkeypatch):
|
|||
return self._sftp
|
||||
|
||||
def exec_command(self, cmd: str, *, get_pty: bool = False, **_kwargs):
|
||||
# Integrity check of the uploaded pyz (added with SHA-256 verify).
|
||||
if _is_pyz_verify_cmd(cmd):
|
||||
return (None, _Stdout(_FAKE_PYZ_SHA256.encode()), _Stderr())
|
||||
calls.append((cmd, bool(get_pty)))
|
||||
# The tar stream uses exec_command directly.
|
||||
if cmd.startswith("tar -cz -C"):
|
||||
|
|
@ -264,12 +286,7 @@ def test_remote_harvest_no_sudo_does_not_request_pty_or_chown(
|
|||
|
||||
import enroll.remote as r
|
||||
|
||||
monkeypatch.setattr(
|
||||
r,
|
||||
"_build_enroll_pyz",
|
||||
lambda td: (Path(td) / "enroll.pyz").write_bytes(b"PYZ")
|
||||
or (Path(td) / "enroll.pyz"),
|
||||
)
|
||||
monkeypatch.setattr(r, "_build_enroll_pyz", _fake_build_enroll_pyz)
|
||||
|
||||
tgz = _make_tgz_bytes({"state.json": b"{}"})
|
||||
calls: list[tuple[str, bool]] = []
|
||||
|
|
@ -357,6 +374,9 @@ def test_remote_harvest_no_sudo_does_not_request_pty_or_chown(
|
|||
return self._sftp
|
||||
|
||||
def exec_command(self, cmd: str, *, get_pty: bool = False, **_kwargs):
|
||||
# Integrity check of the uploaded pyz (added with SHA-256 verify).
|
||||
if _is_pyz_verify_cmd(cmd):
|
||||
return (None, _Stdout(_FAKE_PYZ_SHA256.encode()), _Stderr())
|
||||
calls.append((cmd, bool(get_pty)))
|
||||
if cmd == "mktemp -d":
|
||||
return (None, _Stdout(b"/tmp/enroll-remote-456\n"), _Stderr())
|
||||
|
|
@ -409,12 +429,7 @@ def test_remote_harvest_sudo_password_retry_uses_sudo_s_and_writes_password(
|
|||
import enroll.remote as r
|
||||
|
||||
# Avoid building a real zipapp; just create a file.
|
||||
monkeypatch.setattr(
|
||||
r,
|
||||
"_build_enroll_pyz",
|
||||
lambda td: (Path(td) / "enroll.pyz").write_bytes(b"PYZ")
|
||||
or (Path(td) / "enroll.pyz"),
|
||||
)
|
||||
monkeypatch.setattr(r, "_build_enroll_pyz", _fake_build_enroll_pyz)
|
||||
|
||||
tgz = _make_tgz_bytes({"state.json": b'{"ok": true}\n'})
|
||||
calls: list[tuple[str, bool]] = []
|
||||
|
|
@ -514,6 +529,9 @@ def test_remote_harvest_sudo_password_retry_uses_sudo_s_and_writes_password(
|
|||
return self._sftp
|
||||
|
||||
def exec_command(self, cmd: str, *, get_pty: bool = False, **_kwargs):
|
||||
# Integrity check of the uploaded pyz (added with SHA-256 verify).
|
||||
if _is_pyz_verify_cmd(cmd):
|
||||
return (None, _Stdout(_FAKE_PYZ_SHA256.encode()), _Stderr())
|
||||
calls.append((cmd, bool(get_pty)))
|
||||
|
||||
# Tar stream
|
||||
|
|
@ -800,12 +818,7 @@ def test_remote_harvest_ssh_key_passphrase_retry(monkeypatch, tmp_path: Path):
|
|||
|
||||
import enroll.remote as r
|
||||
|
||||
monkeypatch.setattr(
|
||||
r,
|
||||
"_build_enroll_pyz",
|
||||
lambda td: (Path(td) / "enroll.pyz").write_bytes(b"PYZ")
|
||||
or (Path(td) / "enroll.pyz"),
|
||||
)
|
||||
monkeypatch.setattr(r, "_build_enroll_pyz", _fake_build_enroll_pyz)
|
||||
|
||||
tgz = _make_tgz_bytes({"state.json": b'{"ok": true}\n'})
|
||||
|
||||
|
|
@ -901,6 +914,9 @@ def test_remote_harvest_ssh_key_passphrase_retry(monkeypatch, tmp_path: Path):
|
|||
return self._sftp
|
||||
|
||||
def exec_command(self, cmd: str, *, get_pty: bool = False, **_kwargs):
|
||||
# Integrity check of the uploaded pyz (added with SHA-256 verify).
|
||||
if _is_pyz_verify_cmd(cmd):
|
||||
return (None, _Stdout(_FAKE_PYZ_SHA256.encode()), _Stderr())
|
||||
if cmd.startswith("tar -cz -C"):
|
||||
return (_Stdin(cmd), _Stdout(tgz, rc=0), _Stderr(b""))
|
||||
if cmd == "mktemp -d":
|
||||
|
|
@ -952,12 +968,7 @@ def test_remote_harvest_ssh_key_passphrase_raises_when_not_interactive(
|
|||
|
||||
import enroll.remote as r
|
||||
|
||||
monkeypatch.setattr(
|
||||
r,
|
||||
"_build_enroll_pyz",
|
||||
lambda td: (Path(td) / "enroll.pyz").write_bytes(b"PYZ")
|
||||
or (Path(td) / "enroll.pyz"),
|
||||
)
|
||||
monkeypatch.setattr(r, "_build_enroll_pyz", _fake_build_enroll_pyz)
|
||||
|
||||
class _Chan:
|
||||
def __init__(self):
|
||||
|
|
@ -1022,6 +1033,9 @@ def test_remote_harvest_ssh_key_passphrase_raises_when_not_interactive(
|
|||
return self._sftp
|
||||
|
||||
def exec_command(self, cmd: str, **_kwargs):
|
||||
# Integrity check of the uploaded pyz (added with SHA-256 verify).
|
||||
if _is_pyz_verify_cmd(cmd):
|
||||
return (None, _Stdout(_FAKE_PYZ_SHA256.encode()), _Stderr())
|
||||
return (_Stdout(), _Stdout(), _Stderr())
|
||||
|
||||
def close(self):
|
||||
|
|
@ -1048,3 +1062,124 @@ def test_remote_harvest_ssh_key_passphrase_raises_when_not_interactive(
|
|||
remote_host="example.com",
|
||||
stdin=io.StringIO(),
|
||||
)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Integrity of the uploaded zipapp before remote (root) execution.
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
class _VerifyChan:
|
||||
"""Minimal Paramiko-channel stand-in for _ssh_run consumption."""
|
||||
|
||||
def __init__(self, out: bytes = b"", err: bytes = b"", rc: int = 0):
|
||||
self._out, self._err, self._rc = out, err, rc
|
||||
self._oi = self._ei = 0
|
||||
self._closed = False
|
||||
|
||||
def recv_ready(self) -> bool:
|
||||
return (not self._closed) and self._oi < len(self._out)
|
||||
|
||||
def recv(self, n: int) -> bytes:
|
||||
chunk = self._out[self._oi : self._oi + n]
|
||||
self._oi += len(chunk)
|
||||
return chunk
|
||||
|
||||
def recv_stderr_ready(self) -> bool:
|
||||
return (not self._closed) and self._ei < len(self._err)
|
||||
|
||||
def recv_stderr(self, n: int) -> bytes:
|
||||
chunk = self._err[self._ei : self._ei + n]
|
||||
self._ei += len(chunk)
|
||||
return chunk
|
||||
|
||||
def exit_status_ready(self) -> bool:
|
||||
return self._closed or (
|
||||
self._oi >= len(self._out) and self._ei >= len(self._err)
|
||||
)
|
||||
|
||||
def recv_exit_status(self) -> int:
|
||||
return self._rc
|
||||
|
||||
def shutdown_write(self) -> None:
|
||||
return
|
||||
|
||||
def close(self) -> None:
|
||||
self._closed = True
|
||||
|
||||
|
||||
class _VerifyStdio:
|
||||
def __init__(self, chan: _VerifyChan):
|
||||
self.channel = chan
|
||||
|
||||
|
||||
def _verify_ssh(out: bytes, *, rc: int = 0):
|
||||
chan = _VerifyChan(out=out, rc=rc)
|
||||
|
||||
class FakeSSH:
|
||||
def exec_command(self, cmd, *, get_pty=False, **_kw):
|
||||
assert "hashlib.sha256" in cmd # this is the verify command
|
||||
return (io.StringIO(), _VerifyStdio(chan), _VerifyStdio(chan))
|
||||
|
||||
return FakeSSH()
|
||||
|
||||
|
||||
def test_remote_verify_pyz_sha256_rejects_mismatch():
|
||||
import enroll.remote as r
|
||||
|
||||
ssh = _verify_ssh(b"0" * 64)
|
||||
with pytest.raises(RuntimeError, match="Integrity check failed"):
|
||||
r._remote_verify_pyz_sha256(
|
||||
ssh, "/tmp/x/enroll.pyz", "a" * 64, remote_python="python3"
|
||||
)
|
||||
|
||||
|
||||
def test_remote_verify_pyz_sha256_accepts_match():
|
||||
import enroll.remote as r
|
||||
|
||||
expected = "a" * 64
|
||||
ssh = _verify_ssh(expected.encode())
|
||||
# Should not raise.
|
||||
r._remote_verify_pyz_sha256(
|
||||
ssh, "/tmp/x/enroll.pyz", expected, remote_python="python3"
|
||||
)
|
||||
|
||||
|
||||
def test_remote_verify_pyz_sha256_rejects_empty():
|
||||
import enroll.remote as r
|
||||
|
||||
ssh = _verify_ssh(b"")
|
||||
with pytest.raises(RuntimeError, match="empty SHA-256"):
|
||||
r._remote_verify_pyz_sha256(
|
||||
ssh, "/tmp/x/enroll.pyz", "a" * 64, remote_python="python3"
|
||||
)
|
||||
|
||||
|
||||
def test_remote_verify_pyz_sha256_rejects_nonzero_rc():
|
||||
import enroll.remote as r
|
||||
|
||||
ssh = _verify_ssh(b"", rc=2)
|
||||
with pytest.raises(RuntimeError, match="verify the integrity"):
|
||||
r._remote_verify_pyz_sha256(
|
||||
ssh, "/tmp/x/enroll.pyz", "a" * 64, remote_python="python3"
|
||||
)
|
||||
|
||||
|
||||
def test_build_enroll_pyz_excludes_tests_and_caches_and_returns_sha(tmp_path: Path):
|
||||
import zipfile
|
||||
|
||||
import enroll.remote as r
|
||||
|
||||
pyz, sha = r._build_enroll_pyz(tmp_path)
|
||||
assert isinstance(sha, str) and len(sha) == 64
|
||||
assert sha == r._sha256_file(pyz)
|
||||
|
||||
names = zipfile.ZipFile(pyz).namelist()
|
||||
assert any(n.endswith("schema/state.schema.json") for n in names)
|
||||
for n in names:
|
||||
leaf = n.rsplit("/", 1)[-1]
|
||||
assert "__pycache__" not in n
|
||||
assert not n.endswith((".pyc", ".pyo"))
|
||||
assert "/tests/" not in n
|
||||
assert not leaf.startswith("test_")
|
||||
assert leaf != "conftest.py"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue