#!/usr/bin/env python3 """Validate ORC config and build a worktree-bound Codex launch plan.""" from __future__ import annotations import argparse import hashlib import json import os import pwd import re import shlex import stat import subprocess import sys from pathlib import Path from typing import Any LEVELS = ("low", "mid", "high") STAGES = ("code", "release", "deb", "docker") EXPECTED_EFFORT = {"low": "low", "mid": "medium", "high": "high"} TOP_LEVEL_KEYS = { "version", "defaultLevel", "stageDefaults", "allowedWorktrees", "profiles", } PROFILE_KEYS = { "cli", "model", "reasoningEffort", "permissionMode", "approvalPolicy", } MODEL_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:/-]{0,127}\Z") MAX_CONFIG_SIZE = 64 * 1024 MAX_CONTROL_OUTPUT = 64 * 1024 CONTROL_TIMEOUT_SECONDS = 15 CONFIG_RELATIVE_PATH = Path("docs/orc/config.yaml") COMMON_ENVIRONMENT_NAMES = { "COLORTERM", "LANG", "LC_ALL", "NO_COLOR", "TERM", "TZ", } MODEL_AUTH_ENVIRONMENT = { "codex-login": frozenset(), "openai": frozenset({"OPENAI_API_KEY"}), "azure-openai": frozenset({"AZURE_OPENAI_API_KEY"}), } REMOTE_AUTH_ENVIRONMENT = { "none": frozenset(), "github-token": frozenset({"GITHUB_TOKEN"}), "gitlab-token": frozenset({"GITLAB_TOKEN"}), "gitea-token": frozenset({"GITEA_TOKEN"}), "forgejo-token": frozenset({"FORGEJO_TOKEN"}), "ssh-agent": frozenset({"SSH_AUTH_SOCK"}), "deb-token": frozenset({"DEB_TOKEN"}), } STAGE_REMOTE_AUTH = { "code": frozenset( { "none", "github-token", "gitlab-token", "gitea-token", "forgejo-token", "ssh-agent", } ), "release": frozenset( { "none", "github-token", "gitlab-token", "gitea-token", "forgejo-token", "ssh-agent", } ), "deb": frozenset({"none", "deb-token", "ssh-agent"}), "docker": frozenset({"none"}), } class ConfigError(ValueError): """ORC configuration or launch state is invalid or unsafe.""" def _json_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]: result: dict[str, Any] = {} for key, value in pairs: if key in result: raise ConfigError("JSON-compatible YAML contains a duplicate key") result[key] = value return result def _read_bounded_regular_file(path: Path) -> tuple[str, dict[str, int]]: absolute = path.absolute() flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) descriptor: int | None = None try: descriptor = os.open(absolute, flags) with os.fdopen(descriptor, "rb") as stream: descriptor = None before = os.fstat(stream.fileno()) if not stat.S_ISREG(before.st_mode): raise ConfigError(f"config must be a regular file: {absolute}") if before.st_size > MAX_CONFIG_SIZE: raise ConfigError(f"config exceeds {MAX_CONFIG_SIZE} bytes") content = stream.read(MAX_CONFIG_SIZE + 1) after = os.fstat(stream.fileno()) except OSError as exc: raise ConfigError(f"cannot safely read config {absolute}: {exc}") from exc finally: if descriptor is not None: os.close(descriptor) if len(content) > MAX_CONFIG_SIZE: raise ConfigError(f"config exceeds {MAX_CONFIG_SIZE} bytes") identity_before = (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns) identity_after = (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns) if identity_before != identity_after: raise ConfigError("config changed while it was being read") try: text = content.decode("utf-8") except UnicodeDecodeError as exc: raise ConfigError("config must be valid UTF-8") from exc return text, { "device": before.st_dev, "inode": before.st_ino, "size": before.st_size, "mtimeNs": before.st_mtime_ns, } def _mapping(value: Any, path: str) -> dict[str, Any]: if not isinstance(value, dict): raise ConfigError(f"{path} must be a mapping") return value def _exact_keys( value: dict[str, Any], *, required: set[str], path: str, ) -> None: missing = sorted(required - set(value)) unknown = sorted(set(value) - required) if missing: raise ConfigError(f"{path} is missing fields: {', '.join(missing)}") if unknown: raise ConfigError(f"{path} has unknown fields: {', '.join(unknown)}") def _enum(value: Any, allowed: tuple[str, ...] | set[str], path: str) -> str: if not isinstance(value, str) or value not in allowed: raise ConfigError(f"{path} must be one of: {', '.join(sorted(allowed))}") return value def validate_config(document: Any) -> dict[str, Any]: config = _mapping(document, "config") _exact_keys(config, required=TOP_LEVEL_KEYS, path="config") version = config["version"] if isinstance(version, bool) or version != 1: raise ConfigError("version must be integer 1") _enum(config["defaultLevel"], LEVELS, "defaultLevel") stage_defaults = _mapping(config["stageDefaults"], "stageDefaults") unknown_stages = sorted(set(stage_defaults) - set(STAGES)) if unknown_stages: raise ConfigError( f"stageDefaults has unknown fields: {', '.join(unknown_stages)}" ) for stage, level in stage_defaults.items(): _enum(level, LEVELS, f"stageDefaults.{stage}") allowed = config["allowedWorktrees"] if not isinstance(allowed, list) or not allowed: raise ConfigError("allowedWorktrees must be a non-empty list") seen_worktrees: set[str] = set() for index, entry in enumerate(allowed): if ( not isinstance(entry, str) or not entry or entry != entry.strip() or any(character in entry for character in ("\x00", "\n", "\r")) ): raise ConfigError( f"allowedWorktrees[{index}] must be a safe non-empty path" ) if entry != "." and not Path(entry).is_absolute(): raise ConfigError( f"allowedWorktrees[{index}] must be '.' or an absolute path" ) normalized = entry if entry == "." else str(Path(entry).absolute()) if normalized in seen_worktrees: raise ConfigError(f"allowedWorktrees contains duplicate path: {entry}") seen_worktrees.add(normalized) profiles = _mapping(config["profiles"], "profiles") _exact_keys(profiles, required=set(LEVELS), path="profiles") for level in LEVELS: profile = _mapping(profiles[level], f"profiles.{level}") _exact_keys(profile, required=PROFILE_KEYS, path=f"profiles.{level}") if profile["cli"] != "codex": raise ConfigError(f"profiles.{level}.cli must be codex in ORC v1") model = profile["model"] if not isinstance(model, str) or not MODEL_RE.fullmatch(model): raise ConfigError(f"profiles.{level}.model is not a safe exact model ID") expected_effort = EXPECTED_EFFORT[level] if profile["reasoningEffort"] != expected_effort: raise ConfigError( f"profiles.{level}.reasoningEffort must be {expected_effort}" ) if profile["permissionMode"] != "workspace-write": raise ConfigError( f"profiles.{level}.permissionMode must be workspace-write in ORC v1" ) _enum( profile["approvalPolicy"], {"untrusted", "on-request", "never"}, f"profiles.{level}.approvalPolicy", ) return config def load_config_snapshot(path: Path) -> tuple[dict[str, Any], dict[str, Any]]: raw, identity = _read_bounded_regular_file(path) try: document = json.loads(raw, object_pairs_hook=_json_object) except ConfigError: raise except Exception as exc: raise ConfigError("invalid JSON-compatible YAML config") from exc config = validate_config(document) snapshot = { "sha256": hashlib.sha256(raw.encode("utf-8")).hexdigest(), **identity, } return config, snapshot def load_config(path: Path) -> dict[str, Any]: config, _ = load_config_snapshot(path) return config def resolve_profile( config: dict[str, Any], *, stage: str, global_level: str | None = None, stage_level: str | None = None, model_auth: str = "codex-login", remote_auth: str = "none", ) -> dict[str, Any]: _enum(stage, STAGES, "stage") if global_level is not None: _enum(global_level, LEVELS, "global level") if stage_level is not None: _enum(stage_level, LEVELS, "stage level") _enum(model_auth, set(MODEL_AUTH_ENVIRONMENT), "model auth") _enum(remote_auth, STAGE_REMOTE_AUTH[stage], f"{stage} remote auth") if stage_level is not None: level, source = stage_level, "request.stage" elif global_level is not None: level, source = global_level, "request.global" elif stage in config["stageDefaults"]: level, source = config["stageDefaults"][stage], f"config.stageDefaults.{stage}" else: level, source = config["defaultLevel"], "config.defaultLevel" profile = config["profiles"].get(level) if profile is None: raise ConfigError(f"requested profile does not exist: {level}") worker_args = [ "--model", profile["model"], "-c", f'model_reasoning_effort="{profile["reasoningEffort"]}"', "--sandbox", profile["permissionMode"], "--ask-for-approval", profile["approvalPolicy"], "--strict-config", ] return { "stage": stage, "level": level, "selectionSource": source, "modelAuth": model_auth, "remoteAuth": remote_auth, "profile": dict(profile), "workerArgs": worker_args, } def _path_has_parent_reference(value: str) -> bool: return ".." in Path(value).parts def _assert_no_symlink_components(path: Path, label: str) -> None: if not path.is_absolute(): raise ConfigError(f"{label} must be absolute: {path}") current = Path(path.anchor) for part in path.parts[1:]: current /= part try: metadata = os.lstat(current) except OSError as exc: raise ConfigError(f"{label} does not exist: {current}") from exc if stat.S_ISLNK(metadata.st_mode): raise ConfigError(f"{label} must not contain symlinks: {current}") def canonical_directory(value: str | Path, label: str) -> Path: text = str(value) if ( not text or text != text.strip() or any(character in text for character in ("\x00", "\n", "\r")) or _path_has_parent_reference(text) ): raise ConfigError(f"{label} must be a safe canonical absolute path") raw = Path(text) _assert_no_symlink_components(raw, label) try: resolved = raw.resolve(strict=True) except OSError as exc: raise ConfigError(f"{label} does not exist: {raw}") from exc if resolved != raw or resolved == Path(resolved.anchor) or not resolved.is_dir(): raise ConfigError(f"{label} must be a canonical non-root directory: {resolved}") return resolved def account_identity() -> tuple[Path, str]: account = pwd.getpwuid(os.getuid()) account_home = Path(account.pw_dir).resolve(strict=True) if not account_home.is_dir(): raise ConfigError("current account home is unavailable") return account_home, account.pw_name def trusted_path_entries() -> list[Path]: account_home, _ = account_identity() candidates = [ account_home / ".local" / "bin", account_home / ".local" / "share" / "mise" / "shims", account_home / ".cargo" / "bin", Path("/home/linuxbrew/.linuxbrew/bin"), Path("/usr/local/bin"), Path("/usr/bin"), Path("/bin"), ] result: list[Path] = [] for candidate in candidates: try: resolved = candidate.resolve(strict=True) except OSError: continue if resolved.is_dir() and resolved not in result: result.append(resolved) return result def _trusted_executable(path: Path, expected_name: str) -> Path | None: try: candidate_metadata = os.lstat(path) resolved = path.resolve(strict=True) metadata = resolved.stat() except OSError: return None if not ( stat.S_ISREG(candidate_metadata.st_mode) or stat.S_ISLNK(candidate_metadata.st_mode) ): return None if not stat.S_ISREG(metadata.st_mode) or not os.access(resolved, os.X_OK): return None if metadata.st_uid not in {0, os.getuid()}: return None if stat.S_IMODE(metadata.st_mode) & 0o022: return None if expected_name == "python3": if not resolved.name.startswith("python3"): return None elif resolved.name != expected_name: return None return resolved def resolve_trusted_executable(name: str) -> Path: if name not in {"codex", "git", "orca"}: raise ConfigError(f"unsupported executable: {name}") search_paths = trusted_path_entries() if name == "git": search_paths = [ path for path in search_paths if str(path) in {"/usr/local/bin", "/usr/bin", "/bin"} ] for directory in search_paths: resolved = _trusted_executable(directory / name, name) if resolved is not None: return resolved raise ConfigError(f"trusted {name} executable was not found in fixed directories") def resolve_trusted_python() -> Path: for candidate in (Path("/usr/bin/python3"), Path("/usr/local/bin/python3")): resolved = _trusted_executable(candidate, "python3") if resolved is not None and resolved.stat().st_uid == 0: return resolved raise ConfigError("a root-owned Python executable was not found in fixed paths") def control_environment() -> dict[str, str]: account_home, username = account_identity() result = { "HOME": str(account_home), "LOGNAME": username, "PATH": os.pathsep.join(str(path) for path in trusted_path_entries()), "USER": username, } for name, value in os.environ.items(): if ( name in COMMON_ENVIRONMENT_NAMES or name.startswith("LC_") ) and "\x00" not in value: result[name] = value return result def selected_credential_names( stage: str, *, model_auth: str, remote_auth: str, ) -> frozenset[str]: _enum(stage, STAGES, "stage") _enum(model_auth, set(MODEL_AUTH_ENVIRONMENT), "model auth") _enum(remote_auth, STAGE_REMOTE_AUTH[stage], f"{stage} remote auth") names = MODEL_AUTH_ENVIRONMENT[model_auth] | REMOTE_AUTH_ENVIRONMENT[remote_auth] for name in names: value = os.environ.get(name) if not value or "\x00" in value: raise ConfigError( f"selected authentication variable is unavailable: {name}" ) if name == "SSH_AUTH_SOCK": socket_path = Path(value) if ( not socket_path.is_absolute() or value != value.strip() or any(character in value for character in ("\n", "\r")) or _path_has_parent_reference(value) ): raise ConfigError("selected SSH_AUTH_SOCK is not a safe absolute path") _assert_no_symlink_components(socket_path, "selected SSH_AUTH_SOCK") try: metadata = os.lstat(socket_path) except OSError as exc: raise ConfigError("selected SSH_AUTH_SOCK is unavailable") from exc if ( not stat.S_ISSOCK(metadata.st_mode) or metadata.st_uid != os.getuid() or stat.S_IMODE(metadata.st_mode) & 0o022 ): raise ConfigError("selected SSH_AUTH_SOCK is not a trusted user socket") return names def worker_environment( stage: str, *, model_auth: str, remote_auth: str, ) -> dict[str, str]: result = control_environment() for name in selected_credential_names( stage, model_auth=model_auth, remote_auth=remote_auth, ): result[name] = os.environ[name] return result def _run_control(argv: list[str], label: str) -> str: try: completed = subprocess.run( argv, shell=False, check=False, capture_output=True, text=True, timeout=CONTROL_TIMEOUT_SECONDS, env=control_environment(), ) except (OSError, subprocess.TimeoutExpired) as exc: raise ConfigError(f"{label} could not be executed safely") from exc if completed.returncode != 0: raise ConfigError(f"{label} failed with exit {completed.returncode}") output = completed.stdout.strip() if not output or len(output.encode("utf-8")) > MAX_CONTROL_OUTPUT: raise ConfigError(f"{label} returned invalid output") return output def _git_path(value: str, cwd: Path, label: str) -> Path: candidate = Path(value) if not candidate.is_absolute(): candidate = cwd / candidate _assert_no_symlink_components(candidate, label) try: resolved = candidate.resolve(strict=True) except OSError as exc: raise ConfigError(f"{label} is invalid") from exc if not resolved.is_dir(): raise ConfigError(f"{label} is not a directory") return resolved def validate_worktree( config: dict[str, Any], *, project_root_value: str | Path, worktree_value: str | Path, ) -> dict[str, Any]: project_root = canonical_directory(project_root_value, "project root") worktree = canonical_directory(worktree_value, "target worktree") git = resolve_trusted_executable("git") top_level = canonical_directory( _run_control( [str(git), "-C", str(project_root), "rev-parse", "--show-toplevel"], "Git project-root check", ), "Git project root", ) if top_level != project_root: raise ConfigError("project root is not the repository top level") common_dir = _git_path( _run_control( [str(git), "-C", str(project_root), "rev-parse", "--git-common-dir"], "Git common-directory check", ), project_root, "Git common directory", ) listed = _run_control( [str(git), "-C", str(project_root), "worktree", "list", "--porcelain"], "Git worktree listing", ) registered: set[Path] = set() for line in listed.splitlines(): if line.startswith("worktree "): try: registered.add( canonical_directory(line[9:], "registered worktree") ) except ConfigError: continue allowed: set[Path] = set() for entry in config["allowedWorktrees"]: allowed.add( project_root if entry == "." else canonical_directory(entry, "allowed worktree") ) if not allowed <= registered: raise ConfigError("allowedWorktrees contains an unregistered Git worktree") if worktree not in allowed: raise ConfigError("target worktree is not in allowedWorktrees") target_top = canonical_directory( _run_control( [str(git), "-C", str(worktree), "rev-parse", "--show-toplevel"], "Git target-worktree check", ), "target Git worktree", ) target_common = _git_path( _run_control( [str(git), "-C", str(worktree), "rev-parse", "--git-common-dir"], "target Git common-directory check", ), worktree, "target Git common directory", ) if target_top != worktree or target_common != common_dir: raise ConfigError("target worktree does not belong to the project repository") worktree_metadata = worktree.stat() common_metadata = common_dir.stat() return { "projectRoot": str(project_root), "worktree": str(worktree), "gitCommonDir": str(common_dir), "worktreeIdentity": { "device": worktree_metadata.st_dev, "inode": worktree_metadata.st_ino, "gitCommonDevice": common_metadata.st_dev, "gitCommonInode": common_metadata.st_ino, }, } def _file_facts(path: Path) -> dict[str, Any]: metadata = path.stat() return { "path": str(path), "device": metadata.st_dev, "inode": metadata.st_ino, "size": metadata.st_size, "mtimeNs": metadata.st_mtime_ns, } def _executable_facts(path: Path) -> dict[str, Any]: return { **_file_facts(path), "version": _run_control([str(path), "--version"], "Codex version check"), } def _python_facts(path: Path) -> dict[str, Any]: version = _run_control( [ str(path), "-I", "-S", "-c", "import sys; print(sys.version.split()[0])", ], "Python version check", ) return { **_file_facts(path), "version": version, } def _fingerprint(value: dict[str, Any]) -> str: canonical = json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") return "sha256:" + hashlib.sha256(canonical).hexdigest() def build_launch_plan( config_path: Path, *, project_root: str | Path, worktree: str | Path, stage: str, global_level: str | None = None, stage_level: str | None = None, model_auth: str = "codex-login", remote_auth: str = "none", ) -> dict[str, Any]: project = canonical_directory(project_root, "project root") expected_config = project / CONFIG_RELATIVE_PATH if config_path.absolute() != expected_config: raise ConfigError(f"config path must be {expected_config}") _assert_no_symlink_components(expected_config, "config path") config, config_snapshot = load_config_snapshot(expected_config) worktree_facts = validate_worktree( config, project_root_value=project, worktree_value=worktree, ) profile = resolve_profile( config, stage=stage, global_level=global_level, stage_level=stage_level, model_auth=model_auth, remote_auth=remote_auth, ) selected_credential_names( stage, model_auth=model_auth, remote_auth=remote_auth, ) codex = resolve_trusted_executable("codex") executable = _executable_facts(codex) orca = resolve_trusted_executable("orca") orca_executable = _file_facts(orca) python = resolve_trusted_python() python_executable = _python_facts(python) script = Path(__file__).resolve(strict=True) script_metadata = script.stat() script_sha256 = hashlib.sha256(script.read_bytes()).hexdigest() launch_facts = { **profile, **worktree_facts, "config": { "path": str(expected_config), **config_snapshot, }, "executable": executable, "orca": orca_executable, "python": python_executable, "launcher": { "path": str(script), "device": script_metadata.st_dev, "inode": script_metadata.st_ino, "size": script_metadata.st_size, "mtimeNs": script_metadata.st_mtime_ns, "sha256": script_sha256, }, } fingerprint = _fingerprint(launch_facts) worker_argv = [str(codex), *profile["workerArgs"]] launcher_argv = [ str(python), "-I", "-S", str(script), "_launch", str(expected_config), "--project-root", str(project), "--worktree", str(worktree_facts["worktree"]), "--stage", stage, "--expected-fingerprint", fingerprint, ] if global_level is not None: launcher_argv.extend(["--global-level", global_level]) if stage_level is not None: launcher_argv.extend(["--stage-level", stage_level]) if model_auth != "codex-login": launcher_argv.extend(["--model-auth", model_auth]) if remote_auth != "none": launcher_argv.extend(["--remote-auth", remote_auth]) terminal_command = shlex.join(launcher_argv) worktree_selector = "path:" + str(worktree_facts["worktree"]) terminal_title = f"ORC-{stage}-{profile['level']}-{fingerprint[7:15]}" terminal_create_argv = [ str(orca), "terminal", "create", "--worktree", worktree_selector, "--title", terminal_title, "--command", terminal_command, "--json", ] return { **launch_facts, "argv": worker_argv, "launchFingerprint": fingerprint, "launcherArgv": launcher_argv, "terminalCommand": terminal_command, "worktreeSelector": worktree_selector, "terminalTitle": terminal_title, "terminalCreateArgv": terminal_create_argv, "terminalCreateShellCommand": shlex.join(terminal_create_argv), } def execute_launch(args: argparse.Namespace) -> int: plan = build_launch_plan( args.config, project_root=args.project_root, worktree=args.worktree, stage=args.stage, global_level=args.global_level, stage_level=args.stage_level, model_auth=args.model_auth, remote_auth=args.remote_auth, ) if plan["launchFingerprint"] != args.expected_fingerprint: raise ConfigError("launch fingerprint changed; resolve the profile again") current = canonical_directory(Path.cwd(), "launcher working directory") if current != Path(plan["worktree"]): raise ConfigError("launcher working directory does not match target worktree") current_python = Path(sys.executable).resolve(strict=True) if current_python != Path(plan["python"]["path"]): raise ConfigError("launcher Python does not match the resolved interpreter") if _python_facts(current_python) != plan["python"]: raise ConfigError("Python executable changed before launch") executable = Path(plan["executable"]["path"]) if _executable_facts(executable) != plan["executable"]: raise ConfigError("Codex executable changed before launch") try: os.execve( executable, plan["argv"], worker_environment( plan["stage"], model_auth=plan["modelAuth"], remote_auth=plan["remoteAuth"], ), ) except OSError as exc: raise ConfigError("Codex worker could not be launched") from exc return 1 # pragma: no cover - os.execve does not return on success def _add_resolution_arguments(parser: argparse.ArgumentParser) -> None: parser.add_argument("config", type=Path) parser.add_argument("--project-root", required=True) parser.add_argument("--worktree", required=True) parser.add_argument("--stage", required=True, choices=STAGES) parser.add_argument("--global-level", choices=LEVELS) parser.add_argument("--stage-level", choices=LEVELS) parser.add_argument( "--model-auth", choices=tuple(MODEL_AUTH_ENVIRONMENT), default="codex-login", ) parser.add_argument( "--remote-auth", choices=tuple(REMOTE_AUTH_ENVIRONMENT), default="none", ) def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) subparsers = parser.add_subparsers(dest="command", required=True) validate = subparsers.add_parser("validate", help="validate config only") validate.add_argument("config", type=Path) resolve = subparsers.add_parser("resolve", help="resolve a bound launch plan") _add_resolution_arguments(resolve) launch = subparsers.add_parser("_launch", help=argparse.SUPPRESS) _add_resolution_arguments(launch) launch.add_argument("--expected-fingerprint", required=True) return parser def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) try: if args.command == "validate": config = load_config(args.config) result: dict[str, Any] = { "ok": True, "config": str(args.config.absolute()), "levels": list(LEVELS), "stages": list(STAGES), "defaultLevel": config["defaultLevel"], } elif args.command == "resolve": result = build_launch_plan( args.config, project_root=args.project_root, worktree=args.worktree, stage=args.stage, global_level=args.global_level, stage_level=args.stage_level, model_auth=args.model_auth, remote_auth=args.remote_auth, ) else: return execute_launch(args) except ConfigError as exc: print(f"ORC config error: {exc}", file=sys.stderr) return 1 print(json.dumps(result, ensure_ascii=False, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())