aaronrene / overseer-kit public
v1_io.py python
101 lines 3.4 KB
Raw
sha256:ff22b6a82a08c117176c3e2603264a0e49fe8204ce57c3cefe7f5b165a8634f8 docs: recenter public story on governance loop Human 7 days ago
1 """Small confined-file primitives for the trusted-host v1 utility."""
2 from __future__ import annotations
3
4 import errno
5 import os
6 from pathlib import Path
7 import stat
8 import tempfile
9
10 from cli.digest import sha256_hex_raw as digest
11
12 LIMIT = 128 * 1024
13
14
15 class Refusal(Exception):
16 pass
17
18
19 def confined(root: Path, relative: str, *, parents: bool = False,
20 missing_parents: bool = False) -> Path:
21 """Reject symlink components and traversal; create only requested directories."""
22 rel = Path(relative)
23 if rel.is_absolute() or not rel.parts or any(p in (".", "..") for p in rel.parts):
24 raise Refusal("path_invalid")
25 current = root
26 for part in rel.parts[:-1]:
27 current /= part
28 if parents and not current.exists() and not current.is_symlink():
29 current.mkdir()
30 if missing_parents and not current.exists() and not current.is_symlink():
31 continue
32 if current.is_symlink() or not current.is_dir():
33 raise Refusal("path_parent_invalid")
34 target = current / rel.name
35 if target.is_symlink():
36 raise Refusal("path_symlink")
37 if target.exists():
38 st = target.stat()
39 if not stat.S_ISREG(st.st_mode) or st.st_nlink != 1:
40 raise Refusal("path_not_regular")
41 return target
42
43
44 def read(root: Path, relative: str, *, missing: bool = False) -> bytes | None:
45 path = confined(root, relative, missing_parents=missing)
46 try:
47 fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
48 except FileNotFoundError:
49 if missing:
50 return None
51 raise Refusal("file_missing: " + relative) from None
52 with os.fdopen(fd, "rb") as stream:
53 st = os.fstat(stream.fileno())
54 if not stat.S_ISREG(st.st_mode) or st.st_nlink != 1 or st.st_size > LIMIT:
55 raise Refusal("file_invalid: " + relative)
56 data = stream.read(LIMIT + 1)
57 if len(data) > LIMIT:
58 raise Refusal("file_too_large")
59 return data
60
61
62 def atomic(root: Path, relative: str, data: bytes, *, expected: str,
63 mode: int = 0o644) -> None:
64 """Compare prior bytes; fsync a unique sibling; replace and fsync its directory.
65
66 Caller serialization is deliberately not a security subsystem. The second
67 comparison catches ordinary stale writers; malicious same-UID races are out
68 of scope. NEXT callers also take a brief advisory directory lock.
69 """
70 if len(data) > LIMIT:
71 raise Refusal("file_too_large")
72 path = confined(root, relative, parents=True)
73
74 def compare() -> None:
75 old = read(root, relative, missing=True)
76 actual = "absent" if old is None else digest(old)
77 if expected != actual:
78 raise Refusal("stale_write: " + relative)
79
80 compare()
81 fd, name = tempfile.mkstemp(prefix="." + path.name + ".", dir=path.parent)
82 try:
83 with os.fdopen(fd, "wb") as stream:
84 stream.write(data)
85 os.fchmod(stream.fileno(), mode)
86 stream.flush()
87 os.fsync(stream.fileno())
88 compare()
89 os.replace(name, path)
90 directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
91 try:
92 try:
93 os.fsync(directory)
94 except OSError as exc:
95 if exc.errno not in (errno.EINVAL, errno.ENOTSUP):
96 raise
97 finally:
98 os.close(directory)
99 finally:
100 if os.path.exists(name):
101 os.unlink(name)
File History 1 commit
sha256:ff22b6a82a08c117176c3e2603264a0e49fe8204ce57c3cefe7f5b165a8634f8 docs: recenter public story on governance loop Human 7 days ago