diff --git a/.superpowers/sdd/evidence-task-4-report.md b/.superpowers/sdd/evidence-task-4-report.md index e139caba..bd18ce30 100644 --- a/.superpowers/sdd/evidence-task-4-report.md +++ b/.superpowers/sdd/evidence-task-4-report.md @@ -55,3 +55,33 @@ tests; no Task 4 file is among them. L2 tests remain deselected by the repositor `JobContext.run_dir` for staging and perform their own final atomic publish only after validation. - A dry run is an execution mode foundation: the runner exposes and records it; individual stages remain responsible for suppressing external mutations. + +## Review hardening follow-up + +Four post-implementation findings were fixed test-first: + +1. Resume compatibility is now a canonical SHA-256 fingerprint over checkpoint schema version, + hashed workspace identity, job type, dry-run mode, explicit spec/pipeline versions, + configuration/input fingerprints, and the exact ordered explicit `stage_ids`. Any insertion, + removal, reorder, mode, identity, version, config, or input change rejects resume before a stage + executes. Omitting `resume_run_id` remains the explicit safe path for a new run. +2. Lock traversal now uses directory file descriptors with `O_DIRECTORY` and `O_NOFOLLOW`. + Lock files use `O_NOFOLLOW | O_CLOEXEC`; `fstat` requires a regular file owned by the current + UID with one link, and permissions are forced to `0600` (`0700` for private directories). + Pre-existing lock-file and lock-directory symlinks are rejected. +3. Stage failures now serialize only the fixed safe tuple `internal` / `stage_exception` / + `stage execution failed`. Neither exception class names nor messages are inspected for output; + a hostile exception-name/message regression test proves a terminal failed report is retained. +4. Job/run directory creation is no-follow, owner-checked, private, and durable. Each newly created + parent is fsynced, the run directory is fsynced before the first atomic file write, and the + existing file-fsync → replace → directory-fsync ordering has an explicit regression test. + +Follow-up verification: + +```text +focused job/lock suite: 27 passed in 0.45s +full harness suite: 613 passed, 5 deselected, 17 warnings in 29.65s +Task 4 scoped Ruff: All checks passed +``` + +Repository-wide Ruff continues to report the same 34 unrelated pre-existing legacy-test findings. diff --git a/harness/tests/test_job_locking.py b/harness/tests/test_job_locking.py index 0166812f..0d952583 100644 --- a/harness/tests/test_job_locking.py +++ b/harness/tests/test_job_locking.py @@ -1,4 +1,5 @@ import multiprocessing +import os from pathlib import Path import pytest @@ -46,6 +47,36 @@ def test_lock_keys_cannot_escape_lock_directory(tmp_path): WorkspaceJobLock(tmp_path, "demo", "../evidence") +def test_preexisting_lock_symlink_is_rejected(tmp_path): + lock = WorkspaceJobLock(tmp_path, "demo", "evidence") + lock.path.parent.mkdir(parents=True) + target = tmp_path / "target" + target.write_text("do not modify") + lock.path.symlink_to(target) + with pytest.raises(OSError): + lock.acquire() + assert target.read_text() == "do not modify" + + +def test_preexisting_locks_directory_symlink_is_rejected(tmp_path): + jobs = tmp_path / ".tht-jobs" + jobs.mkdir() + outside = tmp_path / "outside" + outside.mkdir() + (jobs / ".locks").symlink_to(outside, target_is_directory=True) + with pytest.raises(OSError): + WorkspaceJobLock(tmp_path, "demo", "evidence").acquire() + assert list(outside.iterdir()) == [] + + +def test_lock_file_is_owner_only_regular_single_link(tmp_path): + with WorkspaceJobLock(tmp_path, "demo", "evidence") as lock: + stat = os.stat(lock.path, follow_symlinks=False) + assert stat.st_uid == os.getuid() + assert stat.st_nlink == 1 + assert stat.st_mode & 0o777 == 0o600 + + def test_lock_is_recoverable_after_process_crash_without_stale_deletion(tmp_path): context = multiprocessing.get_context("spawn") ready = context.Event() diff --git a/harness/tests/test_job_runner.py b/harness/tests/test_job_runner.py index 36a25a1f..d0821271 100644 --- a/harness/tests/test_job_runner.py +++ b/harness/tests/test_job_runner.py @@ -1,9 +1,11 @@ import json +import os import pytest from pydantic import ValidationError from tht.jobs.models import JobSpec from tht.jobs.runner import CorruptCheckpointError, run_job +import tht.jobs.runner as runner_module def _spec(tmp_path, **updates): @@ -11,6 +13,11 @@ def _spec(tmp_path, **updates): "workspace_id": "demo", "job_type": "evidence", "workspace_root": tmp_path, + "spec_version": "jobs-v1", + "pipeline_version": "evidence-v1", + "config_fingerprint": "sha256:" + "1" * 64, + "input_fingerprint": "sha256:" + "2" * 64, + "stage_ids": ("stage",), } values.update(updates) return JobSpec(**values) @@ -36,11 +43,12 @@ def test_failed_stage_is_resumable_and_skips_completed_stage(tmp_path): calls.append(("acquire", False)) raise RuntimeError("source /customer/alice token=secret unavailable") - first = run_job(_spec(tmp_path), [discover, acquire]) + first = run_job(_spec(tmp_path, stage_ids=("discover", "acquire")), [discover, acquire]) assert first.status == "failed" assert [stage.status for stage in first.stages] == ["succeeded", "failed"] assert first.stages[1].error.model_dump() == { - "category": "RuntimeError", + "category": "internal", + "code": "stage_exception", "message": "stage execution failed", } @@ -48,7 +56,7 @@ def test_failed_stage_is_resumable_and_skips_completed_stage(tmp_path): calls.append(("recovered", False)) second = run_job( - _spec(tmp_path, resume_run_id=first.run_id), + _spec(tmp_path, resume_run_id=first.run_id, stage_ids=("discover", "acquire")), [discover, acquire], ) assert second.resumed_from == first.run_id @@ -125,3 +133,103 @@ def test_stage_timestamps_are_aware_and_ordered(tmp_path): assert stage.finished_at.tzinfo is not None assert stage.started_at <= stage.finished_at assert report.started_at <= stage.started_at <= report.finished_at + + +@pytest.mark.parametrize( + ("update", "replacement"), + [ + ("dry_run", True), + ("spec_version", "jobs-v2"), + ("pipeline_version", "evidence-v2"), + ("config_fingerprint", "sha256:" + "3" * 64), + ("input_fingerprint", "sha256:" + "4" * 64), + ("workspace_id", "other"), + ("job_type", "dwh"), + ], +) +def test_resume_rejects_changed_identity_or_inputs_before_stage_execution( + tmp_path, update, replacement +): + first = run_job(_spec(tmp_path), [lambda _context: None]) + called = False + + def stage(_context): + nonlocal called + called = True + + values = {update: replacement, "resume_run_id": first.run_id} + with pytest.raises(CorruptCheckpointError, match="incompatible"): + run_job(_spec(tmp_path, **values), [stage]) + assert called is False + + +@pytest.mark.parametrize("stages", [[], [lambda _context: None, lambda _context: None]]) +def test_resume_rejects_removed_or_inserted_stages(tmp_path, stages): + def first_stage(_context): + pass + + first = run_job(_spec(tmp_path), [first_stage]) + with pytest.raises(CorruptCheckpointError, match="incompatible"): + run_job(_spec(tmp_path, resume_run_id=first.run_id), stages) + + +def test_resume_rejects_reordered_stages(tmp_path): + def one(_context): + pass + + def two(_context): + pass + + first = run_job(_spec(tmp_path, stage_ids=("one", "two")), [one, two]) + with pytest.raises(CorruptCheckpointError, match="incompatible"): + run_job( + _spec(tmp_path, resume_run_id=first.run_id, stage_ids=("two", "one")), + [two, one], + ) + + +def test_hostile_exception_identity_never_enters_terminal_report(tmp_path): + Hostile = type("ApiKey_secret_/customer/alice", (Exception,), {}) + + def fail(_context): + raise Hostile("password=hunter2") + + report = run_job(_spec(tmp_path), [fail]) + payload = report.model_dump_json() + assert report.status == "failed" + assert report.stages[0].error.model_dump() == { + "category": "internal", + "code": "stage_exception", + "message": "stage execution failed", + } + assert "secret" not in payload + assert "alice" not in payload + assert "hunter2" not in payload + + +def test_new_run_without_resume_allows_intentional_spec_change(tmp_path): + first = run_job(_spec(tmp_path), [lambda _context: None]) + second = run_job(_spec(tmp_path, input_fingerprint="sha256:" + "9" * 64), [lambda _context: None]) + assert second.status == "succeeded" + assert second.run_id != first.run_id + assert second.resumed_from is None + + +def test_run_directories_are_private_and_fsynced_before_atomic_replace(tmp_path, monkeypatch): + events = [] + real_replace = os.replace + + monkeypatch.setattr(runner_module.os, "fsync", lambda _fd: events.append("fsync")) + + def tracked_replace(source, destination): + events.append("replace") + real_replace(source, destination) + + monkeypatch.setattr(runner_module.os, "replace", tracked_replace) + report = run_job(_spec(tmp_path), [lambda _context: None]) + run_dir = tmp_path / ".tht-jobs" / "evidence" / "runs" / report.run_id + + assert run_dir.stat().st_mode & 0o777 == 0o700 + first_replace = events.index("replace") + assert "fsync" in events[:first_replace] + assert "fsync" in events[first_replace + 1 :] diff --git a/harness/tht/jobs/locking.py b/harness/tht/jobs/locking.py index 80f24e6d..26ed43d1 100644 --- a/harness/tht/jobs/locking.py +++ b/harness/tht/jobs/locking.py @@ -6,6 +6,7 @@ import fcntl import hashlib import os import re +import stat from pathlib import Path from types import TracebackType @@ -36,13 +37,42 @@ class WorkspaceJobLock: def acquire(self) -> "WorkspaceJobLock": if self._fd is not None: raise RuntimeError("job lock is already held by this object") - self.path.parent.mkdir(parents=True, exist_ok=True) - fd = os.open(self.path, os.O_RDWR | os.O_CREAT, 0o600) + root_fd = os.open(self.path.parents[2], os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) try: - fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) - except BlockingIOError as error: - os.close(fd) - raise JobAlreadyRunningError("this workspace job is already running") from error + jobs_fd = _open_owned_directory(root_fd, ".tht-jobs") + try: + locks_fd = _open_owned_directory(jobs_fd, ".locks") + try: + fd = os.open( + self.path.name, + os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_CLOEXEC, + 0o600, + dir_fd=locks_fd, + ) + try: + info = os.fstat(fd) + if ( + not stat.S_ISREG(info.st_mode) + or info.st_uid != os.getuid() + or info.st_nlink != 1 + ): + raise OSError("unsafe job lock file") + os.fchmod(fd, 0o600) + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError as error: + raise JobAlreadyRunningError( + "this workspace job is already running" + ) from error + except BaseException: + os.close(fd) + raise + finally: + os.close(locks_fd) + finally: + os.close(jobs_fd) + finally: + os.close(root_fd) self._fd = fd return self @@ -65,3 +95,21 @@ class WorkspaceJobLock: traceback: TracebackType | None, ) -> None: self.release() + + +def _open_owned_directory(parent_fd: int, name: str) -> int: + try: + os.mkdir(name, 0o700, dir_fd=parent_fd) + os.fsync(parent_fd) + except FileExistsError: + pass + fd = os.open(name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=parent_fd) + try: + info = os.fstat(fd) + if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid(): + raise OSError("unsafe job lock directory") + os.fchmod(fd, 0o700) + except BaseException: + os.close(fd) + raise + return fd diff --git a/harness/tht/jobs/models.py b/harness/tht/jobs/models.py index a2210c75..0572ec08 100644 --- a/harness/tht/jobs/models.py +++ b/harness/tht/jobs/models.py @@ -12,6 +12,7 @@ from pydantic import BaseModel, ConfigDict, Field, field_serializer, field_valid _JOB_KEY = re.compile(r"^[a-z][a-z0-9_-]{0,63}$") _RUN_ID = re.compile(r"^[0-9a-f]{32}$") +_FINGERPRINT = re.compile(r"^sha256:[0-9a-f]{64}$") JobStatus = Literal["pending", "running", "succeeded", "failed"] StageStatus = Literal["pending", "running", "succeeded", "failed"] @@ -54,18 +55,38 @@ class JobSpec(_FrozenModel): workspace_id: str job_type: str workspace_root: Path = Field(exclude=True) + spec_version: str = Field(min_length=1, max_length=64) + pipeline_version: str = Field(min_length=1, max_length=64) + config_fingerprint: str + input_fingerprint: str + stage_ids: tuple[str, ...] dry_run: bool = False resume_run_id: str | None = None _workspace_key = field_validator("workspace_id")(_validate_job_key) _job_type_key = field_validator("job_type")(_validate_job_key) + _version_keys = field_validator("spec_version", "pipeline_version")(_validate_job_key) _resume_id = field_validator("resume_run_id")(_validate_run_id) + _config_fingerprint = field_validator("config_fingerprint")( + lambda value: value if _FINGERPRINT.fullmatch(value) else _invalid_fingerprint() + ) + _input_fingerprint = field_validator("input_fingerprint")( + lambda value: value if _FINGERPRINT.fullmatch(value) else _invalid_fingerprint() + ) + _stage_ids = field_validator("stage_ids")( + lambda values: tuple(_validate_job_key(value) for value in values) + ) def model_copy(self, *, update=None, deep: bool = False) -> Self: data = { "workspace_id": self.workspace_id, "job_type": self.job_type, "workspace_root": self.workspace_root, + "spec_version": self.spec_version, + "pipeline_version": self.pipeline_version, + "config_fingerprint": self.config_fingerprint, + "input_fingerprint": self.input_fingerprint, + "stage_ids": self.stage_ids, "dry_run": self.dry_run, "resume_run_id": self.resume_run_id, } @@ -78,7 +99,8 @@ class JobSpec(_FrozenModel): class StageError(_FrozenModel): - category: str = Field(pattern=r"^[A-Za-z][A-Za-z0-9_]{0,127}$") + category: Literal["internal"] = "internal" + code: Literal["stage_exception"] = "stage_exception" message: Literal["stage execution failed"] = "stage execution failed" @@ -97,7 +119,13 @@ class JobRun(_FrozenModel): schema_version: Literal[1] = 1 run_id: str + compatibility_fingerprint: str + workspace_fingerprint: str job_type: str + spec_version: str + pipeline_version: str + config_fingerprint: str + input_fingerprint: str dry_run: bool status: JobStatus started_at: datetime @@ -106,9 +134,20 @@ class JobRun(_FrozenModel): stages: tuple[StageRun, ...] = () _run_id = field_validator("run_id")(_validate_run_id) + _compatibility = field_validator("compatibility_fingerprint", "workspace_fingerprint")( + lambda value: value if _FINGERPRINT.fullmatch(value) else _invalid_fingerprint() + ) + _input_fingerprints = field_validator("config_fingerprint", "input_fingerprint")( + lambda value: value if _FINGERPRINT.fullmatch(value) else _invalid_fingerprint() + ) _job_type = field_validator("job_type")(_validate_job_key) + _persisted_versions = field_validator("spec_version", "pipeline_version")(_validate_job_key) _resumed_from = field_validator("resumed_from")(_validate_run_id) class JobReport(JobRun): """Public machine-readable terminal report (contains no paths or stage outputs).""" + + +def _invalid_fingerprint(): + raise ValueError("fingerprint must be sha256 followed by 64 lowercase hexadecimal characters") diff --git a/harness/tht/jobs/runner.py b/harness/tht/jobs/runner.py index 46849df3..19056370 100644 --- a/harness/tht/jobs/runner.py +++ b/harness/tht/jobs/runner.py @@ -3,8 +3,10 @@ from __future__ import annotations import json +import hashlib import os import uuid +import stat from collections.abc import Callable, Sequence from dataclasses import dataclass from pathlib import Path @@ -33,7 +35,6 @@ Stage = Callable[[JobContext], Any] def _atomic_write(path: Path, payload: str) -> None: - path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp") fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) try: @@ -66,20 +67,21 @@ def _load_checkpoint(path: Path) -> JobRun: raise CorruptCheckpointError("checkpoint is invalid and cannot be resumed") from error -def _stage_name(stage: Stage) -> str: - name = getattr(stage, "__name__", "") - if name == "": - name = "stage" - return name.replace("_", "-") - - def _new_run(spec: JobSpec, run_id: str, stages: Sequence[Stage]) -> JobRun: - names = [_stage_name(stage) for stage in stages] + names = list(spec.stage_ids) + if len(names) != len(stages): + raise ValueError("stage_ids must identify every stage exactly once") if len(names) != len(set(names)): raise ValueError("stage names must be unique") return JobRun( run_id=run_id, + compatibility_fingerprint=_compatibility_fingerprint(spec, names), + workspace_fingerprint=_value_fingerprint(spec.workspace_id), job_type=spec.job_type, + spec_version=spec.spec_version, + pipeline_version=spec.pipeline_version, + config_fingerprint=spec.config_fingerprint, + input_fingerprint=spec.input_fingerprint, dry_run=spec.dry_run, status="running", started_at=utc_now(), @@ -89,9 +91,14 @@ def _new_run(spec: JobSpec, run_id: str, stages: Sequence[Stage]) -> JobRun: def _resume_run(spec: JobSpec, run_id: str, stages: Sequence[Stage], source: JobRun) -> JobRun: - if source.job_type != spec.job_type: - raise CorruptCheckpointError("checkpoint job type does not match resume request") - requested_names = [_stage_name(stage) for stage in stages] + requested_names = list(spec.stage_ids) + if len(requested_names) != len(stages) or len(requested_names) != len(set(requested_names)): + raise CorruptCheckpointError("resume checkpoint is incompatible with requested stages") + expected = _compatibility_fingerprint(spec, requested_names) + if source.compatibility_fingerprint != expected: + raise CorruptCheckpointError( + "resume checkpoint is incompatible; start an intentional new run without resume" + ) source_by_name = {stage.name: stage for stage in source.stages} resumed_stages = [] for name in requested_names: @@ -103,7 +110,13 @@ def _resume_run(spec: JobSpec, run_id: str, stages: Sequence[Stage], source: Job ) return JobRun( run_id=run_id, + compatibility_fingerprint=source.compatibility_fingerprint, + workspace_fingerprint=source.workspace_fingerprint, job_type=spec.job_type, + spec_version=spec.spec_version, + pipeline_version=spec.pipeline_version, + config_fingerprint=spec.config_fingerprint, + input_fingerprint=spec.input_fingerprint, dry_run=spec.dry_run, status="running", started_at=utc_now(), @@ -118,11 +131,20 @@ def run_job(spec: JobSpec, stages: Sequence[Stage]) -> JobReport: jobs_root = spec.workspace_root / ".tht-jobs" / spec.job_type / "runs" run_id = uuid.uuid4().hex run_dir = jobs_root / run_id + _prepare_run_directory(spec.workspace_root, spec.job_type, run_id) checkpoint_path = run_dir / "checkpoint.json" if spec.resume_run_id is None: run = _new_run(spec, run_id, stages) else: source_path = jobs_root / spec.resume_run_id / "checkpoint.json" + if not source_path.exists(): + matches = list( + (spec.workspace_root / ".tht-jobs").glob( + f"*/runs/{spec.resume_run_id}/checkpoint.json" + ) + ) + if len(matches) == 1: + source_path = matches[0] source = _load_checkpoint(source_path) run = _resume_run(spec, run_id, stages, source) _persist(checkpoint_path, run) @@ -140,12 +162,12 @@ def run_job(spec: JobSpec, stages: Sequence[Stage]) -> JobReport: _persist(checkpoint_path, run) try: stage_callable(context) - except Exception as error: + except Exception: failed = stage.model_copy( update={ "status": "failed", "finished_at": utc_now(), - "error": StageError(category=type(error).__name__), + "error": StageError(), } ) run = run.model_copy( @@ -169,3 +191,49 @@ def run_job(spec: JobSpec, stages: Sequence[Stage]) -> JobReport: report = JobReport.model_validate(run.model_dump()) _atomic_write(run_dir / "report.json", report.model_dump_json(indent=2) + "\n") return report + + +def _value_fingerprint(value: str) -> str: + return "sha256:" + hashlib.sha256(value.encode("utf-8")).hexdigest() + + +def _compatibility_fingerprint(spec: JobSpec, stage_ids: list[str]) -> str: + payload = { + "schema_version": 1, + "workspace": _value_fingerprint(spec.workspace_id), + "job_type": spec.job_type, + "dry_run": spec.dry_run, + "spec_version": spec.spec_version, + "pipeline_version": spec.pipeline_version, + "config_fingerprint": spec.config_fingerprint, + "input_fingerprint": spec.input_fingerprint, + "stage_ids": stage_ids, + } + canonical = json.dumps(payload, sort_keys=True, separators=(",", ":")) + return _value_fingerprint(canonical) + + +def _prepare_run_directory(workspace_root: Path, job_type: str, run_id: str) -> None: + parent_fd = os.open(workspace_root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + try: + for component in (".tht-jobs", job_type, "runs", run_id): + try: + os.mkdir(component, 0o700, dir_fd=parent_fd) + os.fsync(parent_fd) + except FileExistsError: + pass + child_fd = os.open( + component, + os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, + dir_fd=parent_fd, + ) + info = os.fstat(child_fd) + if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid(): + os.close(child_fd) + raise OSError("unsafe job run directory") + os.fchmod(child_fd, 0o700) + os.close(parent_fd) + parent_fd = child_fd + os.fsync(parent_fd) + finally: + os.close(parent_fd)