zemion
2ffdb23f69
Verified with the coordinated workspace changes by devkit full run 2026-09-08T225814-186389-0000-3e3ed7cd (all seven phases passed). This shared UI pass does not mark the individual module reviews complete.
233 lines
7.5 KiB
Python
Executable File
233 lines
7.5 KiB
Python
Executable File
"""Bounded local records and predictable output; no network or AI dependency."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import stat
|
|
import tempfile
|
|
import time
|
|
|
|
META_ROOT = Path(__file__).resolve().parents[3]
|
|
MAX_JSON_BYTES = 8 * 1024 * 1024
|
|
IDENTIFIER = re.compile(r"[a-zA-Z0-9][a-zA-Z0-9_.-]{0,127}\Z")
|
|
|
|
|
|
def now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def canonical(value: object) -> bytes:
|
|
return json.dumps(
|
|
value, sort_keys=True, separators=(",", ":"), ensure_ascii=True, allow_nan=False
|
|
).encode()
|
|
|
|
|
|
def digest(value: object) -> str:
|
|
return hashlib.sha256(canonical(value)).hexdigest()
|
|
|
|
|
|
def identifier(value: str) -> str:
|
|
if (
|
|
not isinstance(value, str)
|
|
or not IDENTIFIER.fullmatch(value)
|
|
or value in {".", ".."}
|
|
):
|
|
raise ValueError("Invalid record identifier")
|
|
return value
|
|
|
|
|
|
def state_root(workspace_root: Path, state_dir: Path | None = None) -> Path:
|
|
base = (
|
|
state_dir
|
|
or Path(os.environ.get("XDG_STATE_HOME", str(Path.home() / ".local/state")))
|
|
/ "govoplan/devkit"
|
|
)
|
|
# Preserve spelling until symlink validation; resolving here would hide an unsafe alias.
|
|
base = Path(os.path.abspath(os.fspath(base.expanduser())))
|
|
return base / ("workspace-" + digest(str(workspace_root.resolve()))[:24])
|
|
|
|
|
|
def reject_symlinks(path: Path) -> None:
|
|
for item in (path, *path.parents):
|
|
if item.is_symlink():
|
|
raise ValueError("State and evidence paths must not contain symlinks")
|
|
|
|
|
|
def private_directory(path: Path) -> None:
|
|
reject_symlinks(path)
|
|
path.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
reject_symlinks(path)
|
|
metadata = path.stat()
|
|
if not stat.S_ISDIR(metadata.st_mode) or (
|
|
hasattr(os, "getuid") and metadata.st_uid != os.getuid()
|
|
):
|
|
raise ValueError("State directory must be owned by the current user")
|
|
path.chmod(0o700)
|
|
|
|
|
|
def read_bounded_bytes(path: Path, max_bytes: int = MAX_JSON_BYTES) -> bytes:
|
|
reject_symlinks(path)
|
|
descriptor = os.open(
|
|
path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0)
|
|
)
|
|
with os.fdopen(descriptor, "rb") as handle:
|
|
metadata = os.fstat(handle.fileno())
|
|
if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > max_bytes:
|
|
raise ValueError("Evidence must be a bounded regular file")
|
|
encoded = handle.read(max_bytes + 1)
|
|
if len(encoded) > max_bytes:
|
|
raise ValueError("Evidence file exceeds its size bound")
|
|
return encoded
|
|
|
|
|
|
def read_json(path: Path, max_bytes: int = MAX_JSON_BYTES) -> object:
|
|
encoded = read_bounded_bytes(path, max_bytes)
|
|
|
|
def unique(pairs):
|
|
result = {}
|
|
for key, value in pairs:
|
|
if key in result:
|
|
raise ValueError("Duplicate JSON keys are not accepted")
|
|
result[key] = value
|
|
return result
|
|
|
|
try:
|
|
return json.loads(
|
|
encoded,
|
|
object_pairs_hook=unique,
|
|
parse_constant=lambda _: (_ for _ in ()).throw(
|
|
ValueError("Non-finite JSON number")
|
|
),
|
|
)
|
|
except (RecursionError, UnicodeError) as exc:
|
|
raise ValueError("JSON nesting or encoding is unsupported") from exc
|
|
|
|
|
|
def atomic_text(path: Path, value: str, max_bytes: int = MAX_JSON_BYTES) -> None:
|
|
encoded = value.encode("utf-8")
|
|
if len(encoded) > max_bytes:
|
|
raise ValueError("Output exceeds its size bound")
|
|
reject_symlinks(path.parent)
|
|
path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
reject_symlinks(path.parent)
|
|
reject_symlinks(path)
|
|
if path.exists() and not path.is_file():
|
|
raise ValueError("Output target must be a regular file")
|
|
descriptor, temporary = tempfile.mkstemp(prefix=".devkit-", dir=path.parent)
|
|
try:
|
|
with os.fdopen(descriptor, "wb") as handle:
|
|
os.fchmod(handle.fileno(), 0o600)
|
|
handle.write(encoded)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
os.replace(temporary, path)
|
|
directory = os.open(path.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
|
|
try:
|
|
os.fsync(directory)
|
|
finally:
|
|
os.close(directory)
|
|
finally:
|
|
if os.path.exists(temporary):
|
|
os.unlink(temporary)
|
|
|
|
|
|
def atomic_json(path: Path, payload: object) -> None:
|
|
atomic_text(
|
|
path,
|
|
json.dumps(
|
|
payload, indent=2, sort_keys=True, ensure_ascii=True, allow_nan=False
|
|
)
|
|
+ "\n",
|
|
)
|
|
|
|
|
|
def redact(text: str) -> str:
|
|
"""Best-effort display hygiene, not permission to include secrets in commands."""
|
|
for key, value in os.environ.items():
|
|
if (
|
|
re.search(r"TOKEN|SECRET|PASSWORD|API_KEY|PRIVATE_KEY", key, re.I)
|
|
and len(value) >= 4
|
|
):
|
|
text = text.replace(value, "[redacted]")
|
|
text = re.sub(r"(?im)(authorization\s*[:=]\s*)([^\r\n]+)", r"\1[redacted]", text)
|
|
text = re.sub(r"(?i)(https?://)[^/\s:@]+:[^/\s@]+@", r"\1[redacted]@", text)
|
|
text = re.sub(
|
|
r"(?i)((?:token|password|secret|api[_-]?key)\s*[=:]\s*)[^\s,;]+",
|
|
r"\1[redacted]",
|
|
text,
|
|
)
|
|
return text
|
|
|
|
|
|
def redact_argv(argv: list[str]) -> list[str]:
|
|
result, hide_next = [], False
|
|
for argument in argv:
|
|
if hide_next:
|
|
result.append("[redacted]")
|
|
hide_next = False
|
|
continue
|
|
if re.fullmatch(
|
|
r"--?(?:password|passwd|token|secret|api[-_]key|access[-_]token|authorization)",
|
|
argument,
|
|
re.I,
|
|
):
|
|
hide_next = True
|
|
result.append(redact(argument))
|
|
return result
|
|
|
|
|
|
def safe_output(value):
|
|
"""Redact presentation, not immutable identity hashes or execution inputs."""
|
|
if isinstance(value, dict):
|
|
return {
|
|
key: redact_argv(item)
|
|
if key == "argv"
|
|
and isinstance(item, list)
|
|
and all(isinstance(arg, str) for arg in item)
|
|
else "[redacted]"
|
|
if re.fullmatch(
|
|
r"password|passwd|token|secret|api[_-]?key|authorization",
|
|
str(key),
|
|
re.I,
|
|
)
|
|
else safe_output(item)
|
|
for key, item in value.items()
|
|
}
|
|
if isinstance(value, list):
|
|
return [safe_output(item) for item in value]
|
|
return redact(value) if isinstance(value, str) else value
|
|
|
|
|
|
@contextmanager
|
|
def resource_lock(directory: Path, name: str, timeout: float = 0):
|
|
"""Host-local advisory lock, released by the OS even after a process crash."""
|
|
import fcntl
|
|
|
|
private_directory(directory)
|
|
path = directory / (hashlib.sha256(name.encode()).hexdigest() + ".lock")
|
|
reject_symlinks(path)
|
|
descriptor = os.open(
|
|
path, os.O_CREAT | os.O_RDWR | getattr(os, "O_NOFOLLOW", 0), 0o600
|
|
)
|
|
try:
|
|
if not stat.S_ISREG(os.fstat(descriptor).st_mode):
|
|
raise ValueError("Lock target must be a regular file")
|
|
deadline = time.monotonic() + timeout
|
|
while True:
|
|
try:
|
|
fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
break
|
|
except BlockingIOError:
|
|
if time.monotonic() >= deadline:
|
|
raise RuntimeError(f"Resource is busy: {name}") from None
|
|
time.sleep(min(0.1, max(0, deadline - time.monotonic())))
|
|
yield
|
|
finally:
|
|
os.close(descriptor)
|