Keep future Coulomb release evidence directories private
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3
This commit is contained in:
tegwick 2026-10-31 12:44:59 +01:00
parent d795120f98
commit 904dcb3ce3
4 changed files with 99 additions and 2 deletions

View file

@ -16,6 +16,7 @@ import subprocess
import tempfile
import time
import urllib.request
from contextlib import ExitStack
from datetime import UTC, datetime
from pathlib import Path
@ -35,6 +36,33 @@ class Refused(RuntimeError):
pass
def private_directories(root, relative):
"""Create every component privately; never repair existing unsafe state."""
if not root.is_absolute() or relative.is_absolute() or ".." in relative.parts:
raise Refused("private_directory_path_required")
with ExitStack() as stack:
def opened(name, parent=None):
fd = os.open(name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=parent)
stack.callback(os.close, fd)
return fd
fd = opened("/")
for component in root.parts[1:]:
fd = opened(component, fd)
for component in (*relative.parts, None):
info = os.fstat(fd)
if info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o700:
raise Refused("private_owned_directory_required")
if component is None:
break
try:
os.mkdir(component, mode=0o700, dir_fd=fd)
os.fsync(fd)
except FileExistsError:
pass
fd = opened(component, fd)
class Release:
def __init__(self, capsule, expected, read_only=False):
self.root = capsule
@ -73,12 +101,13 @@ class Release:
return
if self.journal.exists():
raise Refused("prior_execution_requires_review")
self.journal.mkdir(parents=True, mode=0o700)
private_directories(JOURNAL_ROOT.parent, Path(JOURNAL_ROOT.name))
self.journal.mkdir(mode=0o700)
self.root = self.journal / "capsule"
self.root.mkdir(mode=0o700)
for name, data in frozen.items():
target = self.root / name
target.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
private_directories(self.root, Path(name).parent)
fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
with os.fdopen(fd, "wb") as stream:
stream.write(data)