fix: close workspace ownership and API escapes

harness/tests/test_workspace_writer_lock.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
 import os
+import fcntl
+import struct
+import sys
 
 import pytest
 
@@ -14,6 +17,9 @@ def test_verifier_checks_fd_identity(tmp_path):
     root=tmp_path / "root"; root.mkdir(mode=0o700); writer=root / "writer.lock"; writer.touch(mode=0o600)
     r=os.open(root, os.O_RDONLY); w=os.open(writer, os.O_RDWR)
     try:
+        fcntl.flock(w, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        if hasattr(fcntl, "F_OFD_SETLK") and sys.platform.startswith("linux"):
+            fcntl.fcntl(w, fcntl.F_OFD_SETLK, struct.pack("hhqqi", fcntl.F_WRLCK, os.SEEK_SET, 0, 0, 0))
         st=os.fstat(r); env={"THOTH_WORKSPACE_ID":"abc-workspace", "THOTH_WORKSPACE_REVISION":"a"*40, "THOTH_WORKSPACE_DEVICE":str(st.st_dev), "THOTH_WORKSPACE_INODE":str(st.st_ino)}
         cap=verify_workspace_writer_fds(writer_fd=w, root_fd=r, env=env)
         assert cap.inode == st.st_ino
@@ -33,3 +39,17 @@ def test_verifier_rejects_unrelated_lock(tmp_path):
             verify_workspace_writer_fds(writer_fd=forged_fd, root_fd=root_fd, env=env)
     finally:
         os.close(forged_fd); os.close(root_fd)
+
+
+def test_verifier_rejects_independently_opened_unlocked_writer_fd(tmp_path):
+    root = tmp_path / "root"; root.mkdir(mode=0o700)
+    writer = root / "writer.lock"; writer.touch(mode=0o600)
+    root_fd = os.open(root, os.O_RDONLY)
+    writer_fd = os.open(writer, os.O_RDWR)
+    try:
+        st = os.fstat(root_fd)
+        env = {"THOTH_WORKSPACE_ID": "abc-workspace", "THOTH_WORKSPACE_REVISION": "a" * 40, "THOTH_WORKSPACE_DEVICE": str(st.st_dev), "THOTH_WORKSPACE_INODE": str(st.st_ino)}
+        with pytest.raises(WorkspaceWriterConflict):
+            verify_workspace_writer_fds(writer_fd=writer_fd, root_fd=root_fd, env=env)
+    finally:
+        os.close(writer_fd); os.close(root_fd)