233 lines
7.4 KiB
Python
233 lines
7.4 KiB
Python
"""Hermetic filesystem and environment fixtures for the Nono evidence tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import os
|
|
import tempfile
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Iterator, Mapping
|
|
|
|
|
|
_CREDENTIAL_NAME_HINTS = (
|
|
"API_KEY",
|
|
"TOKEN",
|
|
"SECRET",
|
|
"PASSWORD",
|
|
"PRIVATE_KEY",
|
|
"CREDENTIAL",
|
|
"SSH_AUTH_SOCK",
|
|
"AWS_ACCESS_KEY_ID",
|
|
)
|
|
|
|
|
|
def _mode(path: Path, permissions: int) -> None:
|
|
path.mkdir(parents=True, exist_ok=True, mode=permissions)
|
|
os.chmod(path, permissions)
|
|
|
|
|
|
def _write_executable(path: Path, content: str) -> None:
|
|
path.write_text(content, encoding="utf-8")
|
|
os.chmod(path, 0o700)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FixtureSession:
|
|
"""Paths belonging to one disposable, independent session."""
|
|
|
|
name: str
|
|
root: Path
|
|
home: Path
|
|
state: Path
|
|
cache: Path
|
|
runtime: Path
|
|
tmp: Path
|
|
repo: Path
|
|
|
|
def environment(self, base: Mapping[str, str]) -> dict[str, str]:
|
|
"""Return a minimal environment whose state paths point at this session."""
|
|
return _minimal_environment(
|
|
base,
|
|
home=self.home,
|
|
config=self.home / ".config",
|
|
state=self.state,
|
|
cache=self.cache,
|
|
runtime=self.runtime,
|
|
tmp=self.tmp,
|
|
)
|
|
|
|
|
|
class IsolatedFixture:
|
|
"""Context-managed fixture with no dependency on or mutation of host state."""
|
|
|
|
def __init__(self, *, prefix: str = "nono-evidence-") -> None:
|
|
self._temporary: tempfile.TemporaryDirectory[str] | None = None
|
|
self.prefix = prefix
|
|
self.root: Path | None = None
|
|
self.home: Path | None = None
|
|
self.xdg_config_home: Path | None = None
|
|
self.xdg_state_home: Path | None = None
|
|
self.xdg_cache_home: Path | None = None
|
|
self.xdg_runtime_dir: Path | None = None
|
|
self.tmpdir: Path | None = None
|
|
self.repos: Path | None = None
|
|
self.sessions: Path | None = None
|
|
self.stub_bin: Path | None = None
|
|
self.env: dict[str, str] = {}
|
|
self._session_map: dict[str, FixtureSession] = {}
|
|
|
|
def __enter__(self) -> "IsolatedFixture":
|
|
self._temporary = tempfile.TemporaryDirectory(prefix=self.prefix)
|
|
self.root = Path(self._temporary.name)
|
|
os.chmod(self.root, 0o700)
|
|
|
|
self.home = self.root / "home"
|
|
self.xdg_config_home = self.root / "xdg-config"
|
|
self.xdg_state_home = self.root / "xdg-state"
|
|
self.xdg_cache_home = self.root / "xdg-cache"
|
|
self.xdg_runtime_dir = self.root / "xdg-runtime"
|
|
self.tmpdir = self.root / "tmp"
|
|
self.repos = self.root / "repos"
|
|
self.sessions = self.root / "sessions"
|
|
self.stub_bin = self.root / "bin"
|
|
for directory in (
|
|
self.home,
|
|
self.xdg_config_home,
|
|
self.xdg_state_home,
|
|
self.xdg_cache_home,
|
|
self.xdg_runtime_dir,
|
|
self.tmpdir,
|
|
self.repos,
|
|
self.sessions,
|
|
self.stub_bin,
|
|
):
|
|
_mode(directory, 0o700)
|
|
|
|
self.env = _minimal_environment(
|
|
os.environ,
|
|
home=self.home,
|
|
config=self.xdg_config_home,
|
|
state=self.xdg_state_home,
|
|
cache=self.xdg_cache_home,
|
|
runtime=self.xdg_runtime_dir,
|
|
tmp=self.tmpdir,
|
|
stub_bin=self.stub_bin,
|
|
)
|
|
self._create_fixture_repositories()
|
|
self._create_stub_bin()
|
|
self.new_session("one")
|
|
self.new_session("two")
|
|
return self
|
|
|
|
def __exit__(self, *_: object) -> None:
|
|
if self._temporary is not None:
|
|
self._temporary.cleanup()
|
|
self._temporary = None
|
|
self.root = None
|
|
|
|
def new_session(self, name: str) -> FixtureSession:
|
|
"""Create (or return) a distinct set of state and scratch directories."""
|
|
if self.sessions is None or self.repos is None:
|
|
raise RuntimeError("fixture is not active")
|
|
if not name or name in {".", ".."} or "/" in name:
|
|
raise ValueError("session names must be simple path components")
|
|
if name in self._session_map:
|
|
return self._session_map[name]
|
|
root = self.sessions / name
|
|
home = root / "home"
|
|
state = root / "state"
|
|
cache = root / "cache"
|
|
runtime = root / "runtime"
|
|
tmp = root / "tmp"
|
|
repo = self.repos / name
|
|
for directory in (root, home, home / ".config", state, cache, runtime, tmp, repo):
|
|
_mode(directory, 0o700)
|
|
session = FixtureSession(name, root, home, state, cache, runtime, tmp, repo)
|
|
self._session_map[name] = session
|
|
return session
|
|
|
|
def session(self, name: str) -> FixtureSession:
|
|
if name not in self._session_map:
|
|
return self.new_session(name)
|
|
return self._session_map[name]
|
|
|
|
def path(self, *parts: str) -> Path:
|
|
if self.root is None:
|
|
raise RuntimeError("fixture is not active")
|
|
return self.root.joinpath(*parts)
|
|
|
|
def _create_fixture_repositories(self) -> None:
|
|
assert self.repos is not None
|
|
for path in (
|
|
self.repos / "primary",
|
|
self.repos / "sibling-worktree",
|
|
self.repos / "unrelated",
|
|
):
|
|
_mode(path, 0o700)
|
|
(self.repos / "primary" / "README.txt").write_text(
|
|
"fixture repository\n", encoding="utf-8"
|
|
)
|
|
(self.repos / "sibling-worktree" / "README.txt").write_text(
|
|
"sibling fixture\n", encoding="utf-8"
|
|
)
|
|
|
|
def _create_stub_bin(self) -> None:
|
|
assert self.stub_bin is not None
|
|
_write_executable(
|
|
self.stub_bin / "safe-probe",
|
|
"#!/bin/sh\nprintf '%s\\n' safe-probe\n",
|
|
)
|
|
_write_executable(
|
|
self.stub_bin / "credential-name-probe",
|
|
"#!/bin/sh\nprintf '%s\\n' GITHUB_TOKEN\n",
|
|
)
|
|
|
|
|
|
def _minimal_environment(
|
|
inherited: Mapping[str, str],
|
|
*,
|
|
home: Path,
|
|
config: Path,
|
|
state: Path,
|
|
cache: Path,
|
|
runtime: Path,
|
|
tmp: Path,
|
|
stub_bin: Path | None = None,
|
|
) -> dict[str, str]:
|
|
"""Build an allowlist environment; no inherited credential-like names survive."""
|
|
path = f"{stub_bin}:/usr/bin:/bin" if stub_bin is not None else "/usr/bin:/bin"
|
|
environment = {
|
|
"HOME": str(home),
|
|
"XDG_CONFIG_HOME": str(config),
|
|
"XDG_STATE_HOME": str(state),
|
|
"XDG_CACHE_HOME": str(cache),
|
|
"XDG_RUNTIME_DIR": str(runtime),
|
|
"TMPDIR": str(tmp),
|
|
"PATH": path,
|
|
"LANG": inherited.get("LANG", "C"),
|
|
"LC_ALL": inherited.get("LC_ALL", "C"),
|
|
"PYTHONDONTWRITEBYTECODE": "1",
|
|
}
|
|
# Deliberately construct this mapping instead of copying os.environ. Keep this
|
|
# assertion close to the allowlist so a newly added variable is reviewed.
|
|
for name in environment:
|
|
upper = name.upper()
|
|
if any(hint in upper for hint in _CREDENTIAL_NAME_HINTS):
|
|
raise AssertionError(f"credential-like variable accidentally allowlisted: {name}")
|
|
return environment
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def isolated_environment(prefix: str = "nono-evidence-") -> Iterator[IsolatedFixture]:
|
|
"""Yield an :class:`IsolatedFixture` and clean every path on exit."""
|
|
with IsolatedFixture(prefix=prefix) as fixture:
|
|
yield fixture
|
|
|
|
|
|
__all__ = [
|
|
"FixtureSession",
|
|
"IsolatedFixture",
|
|
"isolated_environment",
|
|
]
|