From 3df85090ff24768eb61ec2b6fac023394933b3cf Mon Sep 17 00:00:00 2001 From: gitea-spec-agent Date: Fri, 9 Oct 2026 15:06:18 +0000 Subject: [PATCH] fix(harness): verify socket permissions and report partial cleanup --- harness/lib/managed_lock_broker.py | 127 +++++++---- harness/operations/repository-cache.md | 6 +- .../tests/common/test_managed_lock_broker.py | 206 +++++++++++++++++- 3 files changed, 291 insertions(+), 48 deletions(-) diff --git a/harness/lib/managed_lock_broker.py b/harness/lib/managed_lock_broker.py index ebfd371..9e2f4af 100644 --- a/harness/lib/managed_lock_broker.py +++ b/harness/lib/managed_lock_broker.py @@ -132,9 +132,12 @@ def _write_handle(path: Path, state: dict, *, exclusive: bool = False, on_create os.close(fd) -def _update_handle(path: Path, state: dict) -> None: +def _update_handle(path: Path, state: dict, *, expected_identity: tuple[int, int] | None = None) -> None: fd = _open_handle(path, os.O_WRONLY) try: + st = os.fstat(fd) + if expected_identity is not None and (st.st_dev, st.st_ino) != expected_identity: + raise ValueError("handle identity changed") os.lseek(fd, 0, os.SEEK_SET) os.ftruncate(fd, 0) _write_all(fd, (json.dumps(state, sort_keys=True, separators=(",", ":")) + "\n").encode()) @@ -202,6 +205,7 @@ class Broker: self.handle_created = False self.handle_identity: tuple[int, int] | None = None self.started = False + self.cleanup_result: dict | None = None def _record_handle_identity(self, identity: tuple[int, int]) -> None: self.handle_identity = identity @@ -213,54 +217,74 @@ class Broker: persisted = dict(self.state) if not retain_token: persisted.pop("token", None) - _update_handle(self.handle, persisted) + _update_handle(self.handle, persisted, expected_identity=self.handle_identity) def cleanup(self, status: str) -> bool: - released = False - try: - if self.set.repository.held: - self.set.repository.release() - self.set.release_issue() - released = True - except Exception: - released = False - if released and self.handle_created: + # Keep a previous partial result: a second cleanup must not erase failures. + previous = self.cleanup_result or {} + errors = list(previous.get("errors", [])) + for name, release in (("repository-release", self.set.repository.release), + ("issue-release", self.set.release_issue)): try: - self.persist(status, retain_token=False) + if name == "issue-release" or self.set.repository.held: + release() except Exception: - released = False + errors.append(name) + released = not self.set.issue.held and not self.set.repository.held if self.server: - self.server.close() - self.server = None - if self.socket_identity is not None: + server, self.server = self.server, None try: - if self.socket_parent_fd is not None: - st = os.stat(self.sock_path.name, dir_fd=self.socket_parent_fd, follow_symlinks=False) - if (st.st_dev, st.st_ino) == self.socket_identity: - os.unlink(self.sock_path.name, dir_fd=self.socket_parent_fd) - except (FileNotFoundError, OSError): - pass - if self.handle_created and not self.started: - try: - parent_fd, name = _open_parent(self.handle) - try: - st = os.stat(name, dir_fd=parent_fd, follow_symlinks=False) - if self.handle_identity == (st.st_dev, st.st_ino): - try: - self.persist(status, retain_token=False) - except Exception: - os.unlink(name, dir_fd=parent_fd) - finally: - os.close(parent_fd) + server.close() except OSError: - pass + errors.append("server-close") + socket_removed = previous.get("socket_removed", self.socket_identity is None) + if self.socket_identity is not None and self.socket_parent_fd is not None: + try: + st = os.stat(self.sock_path.name, dir_fd=self.socket_parent_fd, follow_symlinks=False) + if (not stat.S_ISSOCK(st.st_mode) or st.st_uid != os.getuid() + or (st.st_dev, st.st_ino) != self.socket_identity): + errors.append("socket-identity-changed") + else: + os.unlink(self.sock_path.name, dir_fd=self.socket_parent_fd) + socket_removed = True + self.socket_identity = None + except FileNotFoundError: + socket_removed = True + self.socket_identity = None + except OSError: + errors.append("socket-unlink") if self.socket_parent_fd is not None: + fd, self.socket_parent_fd = self.socket_parent_fd, None try: - os.close(self.socket_parent_fd) + os.close(fd) except OSError: - pass - self.socket_parent_fd = None - return released + errors.append("socket-parent-close") + self.cleanup_result = {"lease_released": released, "socket_removed": bool(socket_removed), + "handle_saved": False, "errors": sorted(set(errors))} + if self.handle_created: + try: + # Do not claim another object's handle during failed startup. + if not self.started and self.handle_identity is not None: + fd = _open_handle(self.handle, os.O_RDONLY) + try: + st = os.fstat(fd) + if (st.st_dev, st.st_ino) != self.handle_identity: + raise ValueError("handle identity changed") + finally: + os.close(fd) + self.cleanup_result["handle_saved"] = True + self.state["cleanup"] = self.cleanup_result + terminal_status = status if released and socket_removed and not errors else "cleanup-incomplete" + self.persist(terminal_status, retain_token=not released) + except Exception: + self.cleanup_result["handle_saved"] = False + self.cleanup_result["errors"] = sorted(set(self.cleanup_result["errors"] + ["handle-save"])) + else: + self.cleanup_result["handle_saved"] = True # No owned handle to save. + if self.cleanup_result["errors"]: + print(json.dumps({"event": "cleanup-incomplete", "cleanup": self.cleanup_result}, + separators=(",", ":")), file=sys.stderr, flush=True) + return released and bool(socket_removed) and not self.cleanup_result["errors"] def authenticate(self, req: dict) -> bool: supplied = req.get("token") @@ -300,9 +324,10 @@ class Broker: self.persist("held") return {"status": "held"} if action == "release": - if not self.cleanup("released"): - return {"error": "release-failed"} + complete = self.cleanup("released") self.stopping = True + if not complete: + return {"error": "cleanup-incomplete", "cleanup": self.cleanup_result} return {"status": "released"} return {"error": "invalid action"} @@ -318,7 +343,8 @@ class Broker: except BaseException: raise st = self.handle.lstat() - self.handle_identity = (st.st_dev, st.st_ino) + if (st.st_dev, st.st_ino) != self.handle_identity: + raise ValueError("created handle identity changed") self.handle_created = True self.socket_parent_fd, socket_name = _open_parent(self.sock_path) try: @@ -328,9 +354,17 @@ class Broker: pass self.server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) self.server.bind(f"/proc/self/fd/{self.socket_parent_fd}/{socket_name}") - sock_st = os.fstat(self.server.fileno()) + # Socket FD metadata describes sockfs, not the bound filesystem node. + sock_st = os.stat(socket_name, dir_fd=self.socket_parent_fd, follow_symlinks=False) + if not stat.S_ISSOCK(sock_st.st_mode) or sock_st.st_uid != os.getuid(): + raise ValueError("unsafe bound socket") self.socket_identity = (sock_st.st_dev, sock_st.st_ino) - os.fchmod(self.server.fileno(), 0o600) + os.chmod(socket_name, 0o600, dir_fd=self.socket_parent_fd, follow_symlinks=False) + verified = os.stat(socket_name, dir_fd=self.socket_parent_fd, follow_symlinks=False) + if (not stat.S_ISSOCK(verified.st_mode) or verified.st_uid != os.getuid() + or (verified.st_dev, verified.st_ino) != self.socket_identity + or stat.S_IMODE(verified.st_mode) != 0o600): + raise ValueError("bound socket verification failed") self.server.listen(8); self.server.settimeout(0.2) self.started = True print(json.dumps({"status": "held", "handle": str(self.handle), "socket": str(self.sock_path)}, separators=(",", ":")), flush=True) @@ -352,8 +386,11 @@ class Broker: response = {"error": "invalid request"} conn.sendall((json.dumps(response, separators=(",", ":")) + "\n").encode()) finally: - if self.set.issue.held or self.handle_created or self.server: + if (self.set.issue.held or self.set.repository.held + or (self.cleanup_result is None and (self.handle_created or self.server))): self.cleanup("released" if not self.stopping else "expired") + if self.cleanup_result and self.cleanup_result["errors"]: + raise OSError("broker cleanup incomplete") def serve(args: argparse.Namespace) -> int: diff --git a/harness/operations/repository-cache.md b/harness/operations/repository-cache.md index ef9761b..baee499 100644 --- a/harness/operations/repository-cache.md +++ b/harness/operations/repository-cache.md @@ -64,7 +64,7 @@ Repository cache 경로는 고정된 machine-local 절대경로나 현재 작업 Bootstrap 이후 모든 Issue workflow는 role-neutral shared managed-lock broker 하나를 사용한다. 개별 impl/spec/review skill이 직접 `flock`을 획득하거나 broker 규칙을 재정의하지 않는다. Broker executable은 `harness/lib/managed_lock_broker.py`이며 authenticated request action은 `status`, `renew`, `acquire_repository`, `release_repository`, `release`이고 최대 TTL은 `3600`초다. - Bootstrap lock을 해제한 뒤 broker를 시작한다. checkout 밖 trusted parent 아래 role-owned mode `0700` runtime directory를 미리 만들고, run-unique handle과 socket path를 사용한다. Handle file은 mode `0600`이고 socket은 mode `0600`이며 둘 다 symlink가 아니어야 한다. -- `serve`는 tracked background process로 시작하고 ready JSON을 기다린 뒤에만 Issue 작업을 시작한다. Client는 raw token을 직접 read/log하지 않으며, broker public `request` API가 handle을 인증하고 token을 캡슐화한다. +- `serve`는 tracked background process로 시작하고 ready JSON을 기다린 뒤에만 Issue 작업을 시작한다. Ready 전에 broker는 bound **filesystem socket node**의 type·owner UID·device/inode와 exact mode `0600`을 검증한다. Socket FD의 sockfs metadata는 이 검증을 대신하지 못한다. Client는 raw token을 직접 read/log하지 않으며, broker public `request` API가 handle을 인증하고 token을 캡슐화한다. 이 broker의 보안 경계에서 filesystem owner UID는 trusted principal이다. mode `0700`/`0600`과 random bearer token은 다른 UID 또는 foreign token을 사용한 작업을 막지만, 같은 UID를 공유하는 상호 비신뢰 프로세스를 서로 격리하지는 못한다. 그런 격리가 필요한 배포는 각 role을 distinct UID로 실행해야 한다. @@ -72,6 +72,10 @@ Bootstrap 이후 모든 Issue workflow는 role-neutral shared managed-lock broke - TTL 만료 전에 `renew`를 request한다. Broker crash 뒤 kernel이 advisory lock을 해제해도 stale handle은 진단 정보일 뿐이며 `probe`가 authoritative하다. - 정상, 오류와 cancel 경로는 동일한 cleanup을 수행한다: 보유 중이면 `release_repository`, `release`, broker process wait/reap, 그리고 Issue `probe`를 실행해 `free=true`를 요구한다. 다른 Issue workflow가 합법적으로 repository lock을 먼저 획득할 수 있으므로 완료 gate는 Issue resource를 검사하고, 종료한 broker process가 자신의 repository FD를 더 이상 보유하지 않음을 wait/reap으로 증명한다. 이 순서를 검증하기 전에는 completed/log/report를 주장하지 않는다. Lock 파일은 절대 unlink하지 않는다. +- 소켓 정리는 부모 directory FD를 기준으로 기록한 filesystem device/inode·type·owner가 일치하는 경로만 삭제한다. 경로가 다른 객체나 symlink로 교체되면 그 객체는 삭제하지 않고 `socket-identity-changed`로 보고한다. 같은 UID의 경쟁 프로세스까지 원자적 격리를 보장하지 않으며, SIGKILL/crash 뒤 소켓 삭제를 보장하지 않는다. Stale 경로를 자동 삭제하거나 재사용하지 않는다. 첫 소켓 정리 실패 뒤에는 부모 FD를 닫고 경로를 다시 열어 자동 재시도하지 않는다. 잔여 경로의 제거는 운영자가 exact 경로와 소유권을 확인하는 별도 복구이며, 반복 cleanup은 이전 부분 실패를 보존한다. +- 정상 release 응답은 기존 `{"status":"released"}`를 유지한다. 정리가 일부 실패하면 broker는 `{"error":"cleanup-incomplete","cleanup":...}`를 응답하고 종료하며, `cleanup`의 `lease_released`, `socket_removed`, `handle_saved`, `errors`로 해제와 잔여 정리를 구분한다. Handle 기록 가능 시 같은 결과를 비밀값 없이 저장하고 stderr에도 오류 코드만 보고한다. 부분 실패는 nonzero process exit로 보고하며, 성공한 cleanup을 재적용하거나 실패 이력을 성공으로 덮어쓰지 않는다. 이 오류만으로 이미 해제한 lease를 재해제하거나 workflow 외부 write를 반복하지 않는다. +- 완료 gate는 기존 release → wait/reap → Issue probe 순서에 더해 broker의 정상 종료와 run-owned 소켓 경로 부재를 확인한다. `probe free=true`는 lock 해제 증거이지 소켓 권한·삭제 증거가 아니다. Lease 해제와 소켓·handle 정리 실패는 별도로 보고하고 완료로 간주하지 않는다. + ### Canonical command templates ```sh diff --git a/harness/tests/common/test_managed_lock_broker.py b/harness/tests/common/test_managed_lock_broker.py index 4b50602..6151b45 100644 --- a/harness/tests/common/test_managed_lock_broker.py +++ b/harness/tests/common/test_managed_lock_broker.py @@ -25,11 +25,12 @@ def wait_ready(proc, timeout=3): class ManagedLockBrokerTests(unittest.TestCase): - def start(self, root, owner="alice", repo="ledger", issue=7, ttl=5, role="gitea-impl-agent"): + def start(self, root, owner="alice", repo="ledger", issue=7, ttl=5, role="gitea-impl-agent", mask=-1): locks = root / "repositories" / owner / repo / "locks" locks.mkdir(parents=True) os.chmod(locks, 0o700) (locks / "repository.lock").touch(mode=0o600) + (locks / f"issue-{issue}.lock").touch(mode=0o600) runtime = root / "runtime" runtime.mkdir() os.chmod(runtime, 0o700) @@ -39,7 +40,7 @@ class ManagedLockBrokerTests(unittest.TestCase): "--repository", repo, "--issue", str(issue), "--role", role, "--run-id", "run-1", "--handle", str(handle), "--socket", str(socket), "--ttl-seconds", str(ttl), "--acquire-timeout", "0.3"] - proc = subprocess.Popen(cmd, cwd=Path.cwd(), stdout=subprocess.PIPE, stderr=subprocess.PIPE) + proc = subprocess.Popen(cmd, cwd=Path.cwd(), stdout=subprocess.PIPE, stderr=subprocess.PIPE, umask=mask) ready = wait_ready(proc) self.assertEqual(ready["status"], "held") self.assertNotIn("token", json.dumps(ready)) @@ -126,6 +127,207 @@ class ManagedLockBrokerTests(unittest.TestCase): os.umask(old) self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600) + def test_socket_mode_and_cleanup_under_umasks(self): + for mask in (0o022, 0o077, 0o777): + with self.subTest(mask=mask), tempfile.TemporaryDirectory() as d: + proc, handle, sock = self.start(Path(d), mask=mask) + try: + self.assertEqual(stat.S_IMODE(sock.lstat().st_mode), 0o600) + self.assertTrue(stat.S_ISSOCK(sock.lstat().st_mode)) + self.assertEqual(broker.request_handle(handle, "status")["status"], "held") + self.assertEqual(broker.request_handle(handle, "release")["status"], "released") + self.assertEqual(proc.wait(timeout=3), 0) + self.assertFalse(sock.exists()) + self.assertTrue(probe_free(Path(d) / "repositories/alice/ledger/locks/issue-7.lock")) + finally: + if proc.poll() is None: + proc.terminate(); proc.wait(timeout=3) + proc.stdout.close(); proc.stderr.close() + + def make_bound_instance(self, root): + import socket + locks = root / "repositories/alice/ledger/locks" + locks.mkdir(parents=True); locks.chmod(0o700) + (locks / "repository.lock").touch(mode=0o600) + runtime = root / "runtime"; runtime.mkdir(mode=0o700) + instance = broker.Broker(Namespace(root=str(root), owner="alice", repository="ledger", issue=7, + role="gitea-impl-agent", run_id="cleanup-test", handle=str(runtime / "handle.json"), + socket=str(runtime / "broker.sock"), ttl_seconds=5.0, acquire_timeout=0.0)) + instance.set.issue.acquire() + broker._write_handle(instance.handle, instance.state, exclusive=True, + on_created=instance._record_handle_identity) + instance.handle_created = True; instance.started = True + instance.socket_parent_fd = broker._open_directory(runtime) + instance.server = socket.socket(socket.AF_UNIX) + instance.server.bind(str(instance.sock_path)) + st = instance.sock_path.lstat() + instance.socket_identity = (st.st_dev, st.st_ino) + return instance + + def test_release_reports_socket_failure_separately_and_stops(self): + with tempfile.TemporaryDirectory() as d: + instance = self.make_bound_instance(Path(d)) + with patch.object(broker.os, "unlink", side_effect=PermissionError("injected")): + result = instance.handle_request({"action": "release", "token": instance.token}) + self.assertEqual(result["error"], "cleanup-incomplete") + self.assertTrue(result["cleanup"]["lease_released"]) + self.assertFalse(result["cleanup"]["socket_removed"]) + self.assertTrue(instance.stopping) + self.assertTrue(probe_free(instance.set.issue.path)) + self.assertNotIn("token", json.loads(instance.handle.read_text())) + self.assertIsNone(instance.socket_parent_fd) + self.assertIsNone(instance.server) + + def test_cleanup_preserves_replaced_path_and_reports_incomplete(self): + for kind in ("file", "symlink"): + with self.subTest(kind=kind), tempfile.TemporaryDirectory() as d: + instance = self.make_bound_instance(Path(d)) + instance.sock_path.unlink() + target = Path(d) / "target"; target.write_text("foreign") + if kind == "file": instance.sock_path.write_text("replacement") + else: instance.sock_path.symlink_to(target) + self.assertFalse(instance.cleanup("released")) + self.assertTrue(instance.sock_path.is_symlink() if kind == "symlink" else instance.sock_path.is_file()) + self.assertEqual(target.read_text(), "foreign") + self.assertTrue(instance.cleanup_result["lease_released"]) + self.assertIn("socket-identity-changed", instance.cleanup_result["errors"]) + self.assertFalse(instance.cleanup("released")) # Do not erase a partial result. + + def test_socket_removed_after_expiry_and_sigterm(self): + for action in ("expiry", "sigterm"): + with self.subTest(action=action), tempfile.TemporaryDirectory() as d: + proc, handle, sock = self.start(Path(d), ttl=1 if action == "expiry" else 5) + try: + if action == "sigterm": proc.terminate() + self.assertEqual(proc.wait(timeout=4), 0) + self.assertFalse(sock.exists()) + self.assertTrue(probe_free(Path(d) / "repositories/alice/ledger/locks/issue-7.lock")) + self.assertNotIn("token", json.loads(handle.read_text())) + finally: + if proc.poll() is None: proc.kill(); proc.wait(timeout=3) + proc.stdout.close(); proc.stderr.close() + + def test_partial_cleanup_response_delivered_before_nonzero_exit(self): + import socket + with tempfile.TemporaryDirectory() as d: + root = Path(d); proc, handle, sock_path = self.start(root) + target = root / "replacement-target"; target.write_text("untouched") + try: + data = json.loads(handle.read_text()) + with socket.socket(socket.AF_UNIX) as client: + client.settimeout(3); client.connect(str(sock_path)) + sock_path.unlink(); sock_path.symlink_to(target) + client.sendall((json.dumps({"action": "release", "token": data["token"]}) + "\n").encode()) + with client.makefile("rb") as stream: + result = json.loads(stream.readline()) + self.assertEqual(result["error"], "cleanup-incomplete") + self.assertTrue(result["cleanup"]["lease_released"]) + self.assertEqual(proc.wait(timeout=3), 2) + self.assertTrue(sock_path.is_symlink()) + self.assertEqual(target.read_text(), "untouched") + self.assertTrue(probe_free(root / "repositories/alice/ledger/locks/issue-7.lock")) + terminal = json.loads(handle.read_text()) + self.assertNotIn("token", terminal) + self.assertFalse(terminal["cleanup"]["socket_removed"]) + self.assertNotIn(data["token"], proc.stderr.read().decode()) + finally: + if proc.poll() is None: proc.kill(); proc.wait(timeout=3) + proc.stdout.close(); proc.stderr.close() + + def test_socket_permission_failure_has_no_ready_and_cleans_owned_socket(self): + import contextlib + import io + with tempfile.TemporaryDirectory() as d: + instance = self.make_bound_instance(Path(d)) + instance.cleanup("released") + args = Namespace(root=str(instance.root), owner="alice", repository="ledger", issue=7, + role="gitea-impl-agent", run_id="startup-permission", handle=str(instance.handle.parent / "new.handle"), + socket=str(instance.sock_path.parent / "new.sock"), ttl_seconds=5.0, acquire_timeout=0.0) + fresh = broker.Broker(args) + out = io.StringIO() + with patch.object(broker.os, "chmod", side_effect=PermissionError("injected")), contextlib.redirect_stdout(out): + with self.assertRaises(PermissionError): fresh.run() + self.assertEqual(out.getvalue(), "") + self.assertFalse(fresh.sock_path.exists()) + self.assertTrue(probe_free(fresh.set.issue.path)) + self.assertNotIn("token", json.loads(fresh.handle.read_text())) + + def test_cleanup_respects_lock_order_after_repository_error(self): + with tempfile.TemporaryDirectory() as d: + instance = self.make_bound_instance(Path(d)) + instance.set.repository.acquire() + with patch.object(instance.set.repository, "release", side_effect=OSError("injected")): + self.assertFalse(instance.cleanup("released")) + self.assertFalse(probe_free(instance.set.issue.path)) # Repository still held; preserve lock order. + self.assertFalse(instance.cleanup_result["lease_released"]) + self.assertTrue(instance.cleanup_result["socket_removed"]) + self.assertIn("repository-release", instance.cleanup_result["errors"]) + instance.cleanup("released") # Retry held FD, but keep previous failure visible. + self.assertTrue(probe_free(instance.set.repository.path)) + self.assertNotIn("token", json.loads(instance.handle.read_text())) + + def test_startup_handle_replacement_is_preserved(self): + import contextlib + import io + with tempfile.TemporaryDirectory() as d: + instance = self.make_bound_instance(Path(d)); instance.cleanup("released") + fresh = broker.Broker(Namespace(root=str(instance.root), owner="alice", repository="ledger", issue=7, + role="gitea-impl-agent", run_id="handle-swap", handle=str(instance.handle.parent / "swap.handle"), + socket=str(instance.sock_path.parent / "swap.sock"), ttl_seconds=5.0, acquire_timeout=0.0)) + original_write = broker._write_handle + foreign = b'{"foreign":"preserve"}\n' + def replace_after_create(path, state, **kwargs): + original_write(path, state, **kwargs) + path.rename(path.with_suffix(".original")) # Keep original inode alive. + path.write_bytes(foreign); path.chmod(0o600) + with patch.object(broker, "_write_handle", side_effect=replace_after_create), contextlib.redirect_stdout(io.StringIO()): + with self.assertRaises((OSError, ValueError)): fresh.run() + self.assertEqual(fresh.handle.read_bytes(), foreign) + self.assertTrue(probe_free(fresh.set.issue.path)) + self.assertIn("handle-save", fresh.cleanup_result["errors"]) + + def test_partial_cleanup_durable_status_is_incomplete(self): + with tempfile.TemporaryDirectory() as d: + instance = self.make_bound_instance(Path(d)) + with patch.object(broker.os, "unlink", side_effect=PermissionError("injected")): + instance.cleanup("released") + state = json.loads(instance.handle.read_text()) + self.assertEqual(state["status"], "cleanup-incomplete") + self.assertTrue(state["cleanup"]["lease_released"]) + self.assertFalse(state["cleanup"]["socket_removed"]) + self.assertNotIn("token", state) + + def test_socket_replacement_during_permission_setup_is_preserved(self): + import contextlib + import io + with tempfile.TemporaryDirectory() as d: + instance = self.make_bound_instance(Path(d)); instance.cleanup("released") + fresh = broker.Broker(Namespace(root=str(instance.root), owner="alice", repository="ledger", issue=7, + role="gitea-impl-agent", run_id="socket-swap", handle=str(instance.handle.parent / "swap.handle"), + socket=str(instance.sock_path.parent / "swap.sock"), ttl_seconds=5.0, acquire_timeout=0.0)) + real_chmod = os.chmod + def replace_after_chmod(path, mode, **kwargs): + real_chmod(path, mode, **kwargs) + fresh.sock_path.unlink() + fresh.sock_path.write_text("foreign socket replacement") + out = io.StringIO() + with patch.object(broker.os, "chmod", side_effect=replace_after_chmod), contextlib.redirect_stdout(out): + with self.assertRaises(ValueError): fresh.run() + self.assertEqual(out.getvalue(), "") + self.assertEqual(fresh.sock_path.read_text(), "foreign socket replacement") + self.assertTrue(probe_free(fresh.set.issue.path)) + self.assertEqual(json.loads(fresh.handle.read_text())["status"], "cleanup-incomplete") + + def test_persist_rejects_replaced_handle_after_ready(self): + with tempfile.TemporaryDirectory() as d: + instance = self.make_bound_instance(Path(d)) + instance.handle.rename(instance.handle.with_suffix(".original")) + instance.handle.write_text("foreign"); instance.handle.chmod(0o600) + with self.assertRaises(ValueError): instance.persist("held") + self.assertFalse(instance.cleanup("released")) + self.assertEqual(instance.handle.read_text(), "foreign") + self.assertTrue(probe_free(instance.set.issue.path)) + def test_repository_scope_release_keeps_issue(self): with tempfile.TemporaryDirectory() as d: root = Path(d); proc, handle, _ = self.start(root)