from __future__ import annotations
|
|
import argparse
|
import hashlib
|
import json
|
import os
|
from pathlib import Path
|
import stat
|
import sys
|
from typing import Any
|
|
from .manifests import utc_now
|
from .models import ContractError, ErrorCode
|
from .performance import (build_execution_context, validate_execution_context_bytes)
|
|
|
def _is_reparse(path: Path) -> bool:
|
try:
|
attributes = os.lstat(path).st_file_attributes
|
except AttributeError:
|
return path.is_symlink()
|
except OSError as exc:
|
raise ContractError(ErrorCode.RECOVERY_UNKNOWN, "path", "object stat failed") from exc
|
return bool(attributes & getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0x400))
|
|
|
def _require_ordinary_directory(path: Path, field: str) -> None:
|
try:
|
valid = path.is_dir() and not path.is_symlink() and not _is_reparse(path)
|
except OSError as exc:
|
raise ContractError(ErrorCode.RECOVERY_UNKNOWN, field, "directory state unknown") from exc
|
if not valid:
|
raise ContractError(ErrorCode.PERSIST_LATE, field, "ordinary directory required")
|
|
|
def _lexical_absolute(path: Path) -> Path:
|
lexical = Path(os.path.abspath(os.fspath(path)))
|
if not lexical.is_absolute():
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "evidence_root", "absolute path required")
|
try:
|
str(lexical).encode("ascii", "strict")
|
except UnicodeError as exc:
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "evidence_root", "ASCII path required") from exc
|
return lexical
|
|
|
def _validate_lexical_ancestors(root: Path) -> None:
|
# Walk from the filesystem anchor to the immediate parent without calling
|
# resolve: resolving first would erase the very junction/reparse evidence
|
# this gate must reject.
|
for ancestor in reversed(root.parents):
|
if not os.path.lexists(ancestor):
|
raise ContractError(ErrorCode.PERSIST_LATE, "evidence_root_parent",
|
"existing ancestor required")
|
if _is_reparse(ancestor):
|
raise ContractError(ErrorCode.PERSIST_LATE, "evidence_root_parent",
|
"reparse ancestor rejected")
|
_require_ordinary_directory(ancestor, "evidence_root_parent")
|
|
|
def _validate_lexical_identity(root: Path) -> None:
|
try:
|
resolved = root.resolve(strict=False)
|
except OSError as exc:
|
raise ContractError(ErrorCode.RECOVERY_UNKNOWN, "evidence_root", "resolve failed") from exc
|
if os.path.normcase(str(resolved)) != os.path.normcase(str(root)):
|
raise ContractError(ErrorCode.PERSIST_LATE, "evidence_root", "resolved path drift")
|
|
|
def _write_exclusive(path: Path, data: bytes) -> None:
|
flags = os.O_CREAT | os.O_EXCL | os.O_WRONLY | getattr(os, "O_BINARY", 0)
|
fd = os.open(path, flags, 0o600)
|
try:
|
remaining = memoryview(data)
|
while remaining:
|
written = os.write(fd, remaining)
|
if written <= 0:
|
raise OSError("short write")
|
remaining = remaining[written:]
|
os.fsync(fd)
|
finally:
|
os.close(fd)
|
|
|
def materialize_execution_context(
|
evidence_root: Path, execution_started_at_utc: str, *,
|
preimage: bytes | None = None) -> dict[str, Any]:
|
root = _lexical_absolute(evidence_root)
|
_validate_lexical_ancestors(root)
|
_validate_lexical_identity(root)
|
|
# All semantic and byte validation is complete before the first mkdir/open.
|
if preimage is None:
|
context, data = build_execution_context(root, execution_started_at_utc)
|
else:
|
data = bytes(preimage)
|
context = validate_execution_context_bytes(data, root, execution_started_at_utc)
|
|
if os.path.lexists(root):
|
raise ContractError(ErrorCode.PERSIST_LATE, "evidence_root", "must be absent")
|
|
# Repeat the no-follow checks immediately before the first mutation so a
|
# late ancestor replacement cannot be hidden by canonicalization.
|
_validate_lexical_ancestors(root)
|
_validate_lexical_identity(root)
|
if os.path.lexists(root):
|
raise ContractError(ErrorCode.PERSIST_LATE, "evidence_root", "must remain absent")
|
os.mkdir(root)
|
_require_ordinary_directory(root, "evidence_root")
|
control = root / "control"
|
os.mkdir(control)
|
_require_ordinary_directory(control, "control")
|
|
context_path = control / "execution_context.json"
|
_write_exclusive(context_path, data)
|
if (not context_path.is_file() or context_path.is_symlink() or
|
_is_reparse(context_path)):
|
raise ContractError(ErrorCode.PERSIST_LATE, "execution_context", "ordinary file required")
|
try:
|
reopened = context_path.read_bytes()
|
except OSError as exc:
|
raise ContractError(ErrorCode.RECOVERY_UNKNOWN, "execution_context", "reopen failed") from exc
|
if reopened != data:
|
raise ContractError(ErrorCode.PERSIST_LATE, "execution_context", "readback mismatch")
|
observed = validate_execution_context_bytes(reopened, root, execution_started_at_utc)
|
if observed != context:
|
raise ContractError(ErrorCode.PERSIST_LATE, "execution_context", "semantic readback mismatch")
|
return {
|
"path": str(context_path),
|
"bytes": len(data),
|
"sha256": hashlib.sha256(data).hexdigest(),
|
"execution_started_at_utc": execution_started_at_utc,
|
}
|
|
|
def main(argv: list[str] | None = None) -> int:
|
parser = argparse.ArgumentParser(prog="hibor-fast-performance-coordinator")
|
parser.add_argument("--evidence-root", type=Path, required=True)
|
parser.add_argument("--started-at-utc", default=None)
|
args = parser.parse_args(argv)
|
started_at_utc = args.started_at_utc or utc_now()
|
try:
|
receipt = materialize_execution_context(args.evidence_root, started_at_utc)
|
print(json.dumps(receipt, ensure_ascii=True, separators=(",", ":")))
|
return 0
|
except (OSError, UnicodeError, ValueError, ContractError) as exc:
|
code = exc.code.value if isinstance(exc, ContractError) else "UNEXPECTED_EXCEPTION"
|
print(json.dumps({"status": "STOPPED", "error_code": code},
|
ensure_ascii=True, separators=(",", ":")), file=sys.stderr)
|
return 12
|
|
|
if __name__ == "__main__":
|
raise SystemExit(main())
|
