feat(runtime): install pinned artifacts in a private owner store
Assistant: codex Assistant-Model: gpt-5.6-luna Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
This commit is contained in:
parent
174dba17b6
commit
df5e941814
6 changed files with 333 additions and 1 deletions
89
src/sandboxer/extensions/runtime_store.py
Normal file
89
src/sandboxer/extensions/runtime_store.py
Normal file
|
|
@ -0,0 +1,89 @@
|
|||
"""Publish a reviewed runtime in the local owner's private artifact store."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import fcntl
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import stat
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
from sandboxer.extensions.runtime import verified_runtime
|
||||
|
||||
|
||||
def _owned_directory(path: Path, *, private: bool = False) -> None:
|
||||
info = path.lstat()
|
||||
if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid():
|
||||
raise ValueError("runtime store requires owner-controlled directories")
|
||||
mode = stat.S_IMODE(info.st_mode)
|
||||
if mode & 0o022 or (private and mode != 0o700):
|
||||
raise ValueError("runtime store permissions are not protected")
|
||||
|
||||
|
||||
def _prepare_store(store: Path) -> None:
|
||||
home = Path.home().resolve(strict=True)
|
||||
if not store.is_absolute() or store.resolve() != store or store == home:
|
||||
raise ValueError("store must be a canonical absolute directory below the owner home")
|
||||
if not store.is_relative_to(home):
|
||||
raise ValueError("store must be below the owner home")
|
||||
_owned_directory(home)
|
||||
current = home
|
||||
for part in store.relative_to(home).parts:
|
||||
current /= part
|
||||
current.mkdir(mode=0o700, exist_ok=True)
|
||||
_owned_directory(current, private=current == store)
|
||||
|
||||
|
||||
def _protected_tree(root: Path) -> None:
|
||||
for path in [root, *root.rglob("*")]:
|
||||
info = path.lstat()
|
||||
if info.st_uid != os.getuid():
|
||||
raise ValueError("runtime artifact is not owned by the installing owner")
|
||||
if not stat.S_ISLNK(info.st_mode) and stat.S_IMODE(info.st_mode) & 0o7022:
|
||||
raise ValueError("runtime artifact has unsafe permission bits")
|
||||
|
||||
|
||||
def install_runtime(source: Path, store: Path, sha256: str) -> dict:
|
||||
"""Verify, copy, reverify and atomically publish; never replace an artifact.
|
||||
|
||||
The local owner remains trusted and can alter its own files. This store
|
||||
protects against other local users; the broker separately mounts the
|
||||
artifact read-only and verifies its digest for every sandbox creation.
|
||||
This function enables no profile, credentials, egress or schedule.
|
||||
"""
|
||||
if not re.fullmatch(r"[0-9a-f]{64}", sha256):
|
||||
raise ValueError("expected runtime digest must be a lowercase SHA-256")
|
||||
config = {"runtime": {"path": str(source), "sha256": sha256}}
|
||||
verified_runtime(config)
|
||||
if source.is_relative_to(store) or store.is_relative_to(source):
|
||||
raise ValueError("source and runtime store must not overlap")
|
||||
_prepare_store(store)
|
||||
destination = store / sha256
|
||||
fd = os.open(store / ".install.lock", os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
|
||||
with os.fdopen(fd, "rb") as lock:
|
||||
info = os.fstat(lock.fileno())
|
||||
if (not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid()
|
||||
or stat.S_IMODE(info.st_mode) != 0o600):
|
||||
raise ValueError("runtime install lock is not private and owner-controlled")
|
||||
fcntl.flock(lock, fcntl.LOCK_EX)
|
||||
reused = destination.exists() or destination.is_symlink()
|
||||
if reused:
|
||||
verified_runtime({"runtime": {"path": str(destination), "sha256": sha256}})
|
||||
_protected_tree(destination)
|
||||
else:
|
||||
staging = Path(tempfile.mkdtemp(prefix=".install-", dir=store))
|
||||
try:
|
||||
shutil.copytree(source, staging, symlinks=True, dirs_exist_ok=True)
|
||||
verified_runtime({"runtime": {"path": str(staging), "sha256": sha256}})
|
||||
_protected_tree(staging)
|
||||
staging.rename(destination)
|
||||
finally:
|
||||
if staging.exists():
|
||||
shutil.rmtree(staging)
|
||||
return {
|
||||
"runtime": {"path": str(destination), "sha256": sha256},
|
||||
"owner_uid": os.getuid(), "store_mode": "0700", "reused": reused,
|
||||
"profile_activated": False, "credential_delivery_configured": False,
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue