fix(oci): verify web readiness and persist private hook contracts

2026-11-03 03:12:07 +00:00 · 2026-10-28 08:52:57 +01:00
parent 3a0a7eaf14
commit 6a30d92ba7
5 changed files with 165 additions and 1 deletions
@@ -0,0 +1,54 @@
+"""Job hook contracts can be published on pmxcfs without chmod of /etc/pve/priv."""
+from pathlib import Path
+import json
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / 'remote'))
+import oci_stack_native as native
+
+class JobHookPmxcfsTests(unittest.TestCase):
+    def test_atomic_publish_does_not_change_parent_permissions(self):
+        with tempfile.TemporaryDirectory() as folder:
+            parent = Path(folder)
+            path = parent / 'proxmenux-stack-4900.json'
+            path.write_text('{"old":true}')
+            path.chmod(0o600)
+            mode = parent.stat().st_mode
+            with patch.object(native, 'private_directory', side_effect=AssertionError('No directory chmod')), \
+                 patch.object(native.os, 'chmod', side_effect=PermissionError('pmxcfs')):
+                native.write_hook_contract(path, {'schema': 1, 'dependencies': []})
+            self.assertEqual(json.loads(path.read_text()), {'schema': 1, 'dependencies': []})
+            self.assertEqual(parent.stat().st_mode, mode)
+            self.assertEqual(list(parent.iterdir()), [path])
+            self.assertEqual(path.stat().st_mode & 0o777, 0o600)
+
+    def test_failed_atomic_publish_preserves_existing_contract(self):
+        with tempfile.TemporaryDirectory() as folder:
+            path = Path(folder) / 'proxmenux-stack-4900.json'
+            path.write_text('{"old":true}')
+            path.chmod(0o600)
+            with patch.object(native.os, 'replace', side_effect=OSError('pmxcfs write failed')):
+                with self.assertRaisesRegex(OSError, 'write failed'):
+                    native.write_hook_contract(path, {'schema': 1})
+            self.assertEqual(json.loads(path.read_text()), {'old': True})
+            self.assertEqual(list(Path(folder).iterdir()), [path])
+
+    def test_untrusted_parent_or_file_is_refused_without_writing(self):
+        for parent_mode, file_mode in ((0o755, 0o600), (0o700, 0o644)):
+            with self.subTest(parent_mode=parent_mode, file_mode=file_mode), tempfile.TemporaryDirectory() as folder:
+                parent = Path(folder); parent.chmod(parent_mode)
+                path = parent / 'contract.json'; path.write_text('{"old":true}'); path.chmod(file_mode)
+                with self.assertRaisesRegex(ValueError, 'Unsafe dependency hook contract'):
+                    native.write_hook_contract(path, {'schema': 1})
+                self.assertEqual(json.loads(path.read_text()), {'old': True})
+
+    def test_symlink_contract_is_refused(self):
+        with tempfile.TemporaryDirectory() as folder:
+            real = Path(folder) / 'real.json'; real.write_text('{"old":true}'); real.chmod(0o600)
+            link = Path(folder) / 'contract.json'; link.symlink_to(real)
+            with self.assertRaisesRegex(ValueError, 'Unsafe dependency hook contract'):
+                native.write_hook_contract(link, {'schema': 1})
+            self.assertEqual(json.loads(real.read_text()), {'old': True})