Spaces:
Running
Running
| """Root-owned immutable executable identities for the sealed-exec boundary. | |
| The policy is deliberately small. It is not a general application sandbox: | |
| it answers one question for the native listener, namely whether a newly | |
| requested executable image is the exact root-controlled object that an | |
| operator selected before launch. | |
| """ | |
| from __future__ import annotations | |
| import hashlib | |
| import os | |
| import re | |
| import stat | |
| import time | |
| from pathlib import Path | |
| from typing import Any | |
| from .jsonio import JsonInputError, canonical_bytes, load_regular_json | |
| from .records import write_atomic | |
| SCHEMA = "lumi-eggcracker.exec-policy.v1" | |
| NAME = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,63}\Z") | |
| HEX = re.compile(r"[0-9a-f]{64}\Z") | |
| POLICY_ID = re.compile(r"[0-9a-f]{24}\Z") | |
| MAX_EXECUTABLES = 32 | |
| MAX_EXECUTABLE_BYTES = 512 * 1024 * 1024 | |
| STATE_DIR = Path("/var/lib/lumi-eggcracker/exec-policies") | |
| def _root_controlled(path: Path) -> bool: | |
| if not path.is_absolute() or Path(os.path.normpath(str(path))) != path: | |
| return False | |
| for item in (path, *path.parents): | |
| try: | |
| metadata = item.lstat() | |
| except OSError: | |
| return False | |
| if metadata.st_uid != 0: | |
| return False | |
| if not stat.S_ISLNK(metadata.st_mode) and metadata.st_mode & 0o022: | |
| return False | |
| return True | |
| def _hash_descriptor(descriptor: int, metadata: os.stat_result) -> str: | |
| if not stat.S_ISREG(metadata.st_mode) or not 1 <= metadata.st_size <= MAX_EXECUTABLE_BYTES: | |
| raise JsonInputError("execution policy executable is outside the supported size bound") | |
| before = ( | |
| metadata.st_dev, | |
| metadata.st_ino, | |
| metadata.st_size, | |
| metadata.st_mtime_ns, | |
| metadata.st_ctime_ns, | |
| ) | |
| os.lseek(descriptor, 0, os.SEEK_SET) | |
| digest = hashlib.sha256() | |
| remaining = metadata.st_size | |
| while remaining: | |
| block = os.read(descriptor, min(1024 * 1024, remaining)) | |
| if not block: | |
| raise JsonInputError("execution policy executable changed during hashing") | |
| digest.update(block) | |
| remaining -= len(block) | |
| after_metadata = os.fstat(descriptor) | |
| after = ( | |
| after_metadata.st_dev, | |
| after_metadata.st_ino, | |
| after_metadata.st_size, | |
| after_metadata.st_mtime_ns, | |
| after_metadata.st_ctime_ns, | |
| ) | |
| if after != before: | |
| raise JsonInputError("execution policy executable changed during hashing") | |
| return digest.hexdigest() | |
| def inspect_executable(path_text: str) -> dict[str, Any]: | |
| """Capture one immutable root-controlled executable identity.""" | |
| if not isinstance(path_text, str) or not path_text or len(path_text) > 4096: | |
| raise JsonInputError("execution policy executable path is invalid") | |
| supplied = Path(path_text) | |
| if not _root_controlled(supplied): | |
| raise JsonInputError("execution policy executable path must be root-controlled") | |
| try: | |
| resolved = supplied.resolve(strict=True) | |
| except OSError as error: | |
| raise JsonInputError("execution policy executable cannot be resolved") from error | |
| if not _root_controlled(resolved): | |
| raise JsonInputError("resolved execution policy executable is not root-controlled") | |
| flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0) | |
| try: | |
| descriptor = os.open(resolved, flags) | |
| except OSError as error: | |
| raise JsonInputError("execution policy executable cannot be opened") from error | |
| try: | |
| metadata = os.fstat(descriptor) | |
| if ( | |
| not stat.S_ISREG(metadata.st_mode) | |
| or metadata.st_uid != 0 | |
| or metadata.st_mode & 0o022 | |
| or not (metadata.st_mode & 0o111) | |
| ): | |
| raise JsonInputError("execution policy target must be a root-owned executable") | |
| digest = _hash_descriptor(descriptor, metadata) | |
| return { | |
| "device": metadata.st_dev, | |
| "inode": metadata.st_ino, | |
| "mode": stat.S_IMODE(metadata.st_mode), | |
| "path": str(resolved), | |
| "sha256": digest, | |
| "size": metadata.st_size, | |
| "uid": metadata.st_uid, | |
| } | |
| finally: | |
| os.close(descriptor) | |
| def _validate_entry(value: Any) -> dict[str, Any]: | |
| if not isinstance(value, dict) or set(value) != { | |
| "device", "inode", "mode", "path", "sha256", "size", "uid" | |
| }: | |
| raise JsonInputError("execution policy executable identity is invalid") | |
| if ( | |
| not isinstance(value["path"], str) | |
| or not value["path"].startswith("/") | |
| or not isinstance(value["sha256"], str) | |
| or HEX.fullmatch(value["sha256"]) is None | |
| ): | |
| raise JsonInputError("execution policy executable identity is invalid") | |
| for key in ("device", "inode", "mode", "size", "uid"): | |
| if isinstance(value[key], bool) or not isinstance(value[key], int) or value[key] < 0: | |
| raise JsonInputError("execution policy executable integer is invalid") | |
| if value["uid"] != 0 or value["size"] < 1 or value["size"] > MAX_EXECUTABLE_BYTES: | |
| raise JsonInputError("execution policy executable owner or size is invalid") | |
| return value | |
| def _digest(value: dict[str, Any]) -> str: | |
| body = {key: value[key] for key in value if key != "digest"} | |
| return hashlib.sha256(canonical_bytes(body)).hexdigest() | |
| def validate(value: dict[str, Any]) -> dict[str, Any]: | |
| expected = { | |
| "created_monotonic_ns", "creator_uid", "digest", "executables", "generation", | |
| "name", "policy_id", "revoked", "schema_version", | |
| } | |
| if set(value) != expected or value.get("schema_version") != SCHEMA: | |
| raise JsonInputError("execution policy schema is invalid") | |
| if not isinstance(value["name"], str) or NAME.fullmatch(value["name"]) is None: | |
| raise JsonInputError("execution policy name is invalid") | |
| if not isinstance(value["policy_id"], str) or POLICY_ID.fullmatch(value["policy_id"]) is None: | |
| raise JsonInputError("execution policy identity is invalid") | |
| if not isinstance(value["digest"], str) or HEX.fullmatch(value["digest"]) is None: | |
| raise JsonInputError("execution policy digest is invalid") | |
| for key in ("created_monotonic_ns", "creator_uid", "generation"): | |
| if isinstance(value[key], bool) or not isinstance(value[key], int) or value[key] < 0: | |
| raise JsonInputError("execution policy integer is invalid") | |
| if value["creator_uid"] != 0 or value["generation"] < 1 or not isinstance(value["revoked"], bool): | |
| raise JsonInputError("execution policy authority or state is invalid") | |
| entries = value["executables"] | |
| if not isinstance(entries, list) or not 1 <= len(entries) <= MAX_EXECUTABLES: | |
| raise JsonInputError("execution policy executable count is invalid") | |
| checked = [_validate_entry(entry) for entry in entries] | |
| identities = {(item["device"], item["inode"], item["size"], item["sha256"]) for item in checked} | |
| if len(identities) != len(checked): | |
| raise JsonInputError("execution policy executable identity is duplicated") | |
| if _digest(value) != value["digest"]: | |
| raise JsonInputError("execution policy digest does not match contents") | |
| return value | |
| def _path(root: Path, name: str) -> Path: | |
| if not isinstance(name, str) or NAME.fullmatch(name) is None: | |
| raise JsonInputError("execution policy name is invalid") | |
| return root / f"{name}.json" | |
| def create(root: Path, *, name: str, paths: list[str], creator_uid: int = 0) -> dict[str, Any]: | |
| if creator_uid != 0 or not isinstance(paths, list) or not 1 <= len(paths) <= MAX_EXECUTABLES: | |
| raise JsonInputError("execution policy arguments are invalid") | |
| destination = _path(root, name) | |
| if destination.exists() or destination.is_symlink(): | |
| raise JsonInputError("execution policy name is unavailable") | |
| entries = [inspect_executable(item) for item in paths] | |
| value = { | |
| "created_monotonic_ns": time.monotonic_ns(), | |
| "creator_uid": 0, | |
| "digest": "0" * 64, | |
| "executables": entries, | |
| "generation": 1, | |
| "name": name, | |
| "policy_id": os.urandom(12).hex(), | |
| "revoked": False, | |
| "schema_version": SCHEMA, | |
| } | |
| value["digest"] = _digest(value) | |
| value = validate(value) | |
| root.mkdir(mode=0o700, parents=True, exist_ok=True) | |
| os.chown(root, 0, 0) | |
| os.chmod(root, 0o700) | |
| write_atomic(destination, value) | |
| os.chown(destination, 0, 0) | |
| os.chmod(destination, 0o600) | |
| return value | |
| def load_all(root: Path = STATE_DIR) -> list[dict[str, Any]]: | |
| if not root.exists(): | |
| return [] | |
| if root.is_symlink() or not root.is_dir(): | |
| raise JsonInputError("execution policy root is invalid") | |
| values: list[dict[str, Any]] = [] | |
| for path in sorted(root.glob("*.json")): | |
| value = validate(load_regular_json(path)) | |
| if value["name"] != path.stem: | |
| raise JsonInputError("execution policy name/path mismatch") | |
| values.append(value) | |
| return values | |
| def load(root: Path, policy_id: str) -> dict[str, Any]: | |
| for value in load_all(root): | |
| if value["policy_id"] == policy_id: | |
| if value["revoked"]: | |
| raise JsonInputError("execution policy is revoked") | |
| return value | |
| raise JsonInputError("execution policy is unavailable") | |
| def revoke(root: Path, policy_id: str) -> dict[str, Any]: | |
| values = load_all(root) | |
| for value in values: | |
| if value["policy_id"] == policy_id: | |
| path = _path(root, value["name"]) | |
| replacement = dict(value) | |
| replacement["revoked"] = True | |
| replacement["generation"] = value["generation"] + 1 | |
| replacement["digest"] = _digest(replacement) | |
| write_atomic(path, validate(replacement)) | |
| return {"policy_id": policy_id, "result": "REVOKED"} | |
| raise JsonInputError("execution policy is unavailable") | |
| def public(value: dict[str, Any]) -> dict[str, Any]: | |
| checked = validate(value) | |
| return { | |
| "digest": checked["digest"], | |
| "executable_count": len(checked["executables"]), | |
| "generation": checked["generation"], | |
| "name": checked["name"], | |
| "policy_id": checked["policy_id"], | |
| "revoked": checked["revoked"], | |
| } | |
| def ephemeral(path: str, run_id: str) -> dict[str, Any]: | |
| """Build a non-persistent policy for the default initial executable.""" | |
| entry = inspect_executable(path) | |
| value = { | |
| "created_monotonic_ns": time.monotonic_ns(), | |
| "creator_uid": 0, | |
| "digest": "0" * 64, | |
| "executables": [entry], | |
| "generation": 1, | |
| "name": f"sealed-{run_id}", | |
| "policy_id": run_id, | |
| "revoked": False, | |
| "schema_version": SCHEMA, | |
| } | |
| value["digest"] = _digest(value) | |
| return validate(value) | |
| def match_identity(policy: dict[str, Any], identity: dict[str, Any]) -> bool: | |
| validate(policy) | |
| return any( | |
| all(identity.get(key) == entry.get(key) for key in ("device", "inode", "size", "sha256")) | |
| for entry in policy["executables"] | |
| ) | |