126 lines
6.4 KiB
Python
126 lines
6.4 KiB
Python
import importlib.util
|
|
import json
|
|
import multiprocessing
|
|
import os
|
|
import stat
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
SCRIPT = Path(__file__).resolve().parents[2] / 'agents/review/skills/gitea-code-review/scripts/publication_journal.py'
|
|
|
|
def load():
|
|
spec = importlib.util.spec_from_file_location('publication_journal_safety', SCRIPT)
|
|
m = importlib.util.module_from_spec(spec); spec.loader.exec_module(m); return m
|
|
|
|
def plan(n='a'):
|
|
return {'target': {'owner':'o','repository':'r','issue':1,'pull':2,'head':n*40},
|
|
'contract_sha256':'b'*64, 'reviewer': {'id':3}}
|
|
|
|
def validate(p): return p
|
|
|
|
class SafetyTests(unittest.TestCase):
|
|
def test_binding_record_contains_plan_fingerprint_and_strict_schema(self):
|
|
m=load()
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root=Path(td)/'root'; m.root_for=lambda _: root
|
|
p=plan(); e=m.open_binding(m.canonical_path(p,anchor_file=__file__,origin='x',validate=validate),p,anchor_file=__file__,origin='x',validate=validate)
|
|
record=json.loads((e/'binding.json').read_text())
|
|
self.assertEqual(set(record), {'schema','binding','fingerprint'})
|
|
self.assertEqual(record['fingerprint'], m.plan_fingerprint(p))
|
|
self.assertEqual(m.assert_bound(e,p,anchor_file=__file__,origin='x',validate=validate), e)
|
|
|
|
def test_different_plan_same_binding_conflicts_without_overwrite(self):
|
|
m=load()
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root=Path(td)/'root'; m.root_for=lambda _: root
|
|
p=plan(); path=m.canonical_path(p,anchor_file=__file__,origin='x',validate=validate); m.open_binding(path,p,anchor_file=__file__,origin='x',validate=validate)
|
|
q=dict(p); q['extra']='different'
|
|
with self.assertRaises(ValueError): m.assert_bound(path,q,anchor_file=__file__,origin='x',validate=validate)
|
|
|
|
def test_strict_json_rejects_duplicate_and_nonfinite(self):
|
|
m=load()
|
|
with tempfile.TemporaryDirectory() as td:
|
|
p=Path(td)/'x'; p.write_bytes(b'{"a":1,"a":2}') ; os.chmod(p,0o600)
|
|
with self.assertRaises(ValueError): m._strict_load(m._secure_file(p))
|
|
p.write_bytes(b'{"a":NaN}')
|
|
with self.assertRaises(ValueError): m._strict_load(m._secure_file(p))
|
|
p.write_bytes(b'{"a":1e999}')
|
|
with self.assertRaises(ValueError): m._strict_load(m._secure_file(p))
|
|
|
|
def test_require_new_seeds_exact_pending_registration_sentinel(self):
|
|
m=load()
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root=Path(td)/'root'; m.root_for=lambda _: root
|
|
p=plan(); path=m.canonical_path(p,anchor_file=__file__,origin='x',validate=validate)
|
|
entry=m.open_binding(path,p,anchor_file=__file__,origin='x',validate=validate,require_new=True)
|
|
self.assertEqual(json.loads((entry/'import.pending.json').read_text()), {'schema':'gitea-agent.native-review-import/v1','state':'pending'})
|
|
with self.assertRaises(ValueError):
|
|
m.open_binding(path,p,anchor_file=__file__,origin='x',validate=validate)
|
|
|
|
def test_fifo_does_not_block_and_symlink_hardlink_rejected(self):
|
|
m=load()
|
|
with tempfile.TemporaryDirectory() as td:
|
|
d=Path(td); fifo=d/'fifo'; os.mkfifo(fifo,0o600)
|
|
with self.assertRaises(ValueError): m._secure_file(fifo,cap=10)
|
|
real=d/'real'; real.write_bytes(b'{}'); os.chmod(real,0o600)
|
|
os.symlink(real,d/'link')
|
|
with self.assertRaises(Exception): m._secure_file(d/'link')
|
|
os.link(real,d/'hard')
|
|
with self.assertRaises(ValueError): m._secure_file(real)
|
|
|
|
def test_deleted_binding_record_is_not_reconstructed(self):
|
|
m = load()
|
|
with tempfile.TemporaryDirectory() as td:
|
|
m.root_for = lambda _: Path(td) / 'root'
|
|
p = plan(); path = m.canonical_path(p, anchor_file=__file__, origin='x', validate=validate)
|
|
m.open_binding(path, p, anchor_file=__file__, origin='x', validate=validate)
|
|
(path / 'binding.json').unlink()
|
|
with self.assertRaises(ValueError):
|
|
m.open_binding(path, p, anchor_file=__file__, origin='x', validate=validate)
|
|
self.assertFalse((path / 'binding.json').exists())
|
|
|
|
def test_first_use_bootstraps_private_runtime_parent(self):
|
|
m = load()
|
|
with tempfile.TemporaryDirectory() as td:
|
|
runtime = Path(td) / '.runtime'
|
|
m.root_for = lambda _: runtime / 'root'
|
|
p = plan(); path = m.canonical_path(p, anchor_file=__file__, origin='x', validate=validate)
|
|
self.assertEqual(m.open_binding(path, p, anchor_file=__file__, origin='x', validate=validate), path)
|
|
self.assertEqual(stat.S_IMODE(runtime.stat().st_mode), 0o700)
|
|
|
|
def test_prewrite_presence_validation_rejects_schema_corruption(self):
|
|
m = load()
|
|
for kind in ('root', 'entry'):
|
|
with self.subTest(kind=kind), tempfile.TemporaryDirectory() as td:
|
|
m.root_for = lambda _: Path(td) / 'root'
|
|
p = plan(); path = m.canonical_path(p, anchor_file=__file__, origin='x', validate=validate)
|
|
m.open_binding(path, p, anchor_file=__file__, origin='x', validate=validate)
|
|
marker = Path(td) / ('root.presence' if kind == 'root' else path.name + '.presence')
|
|
record = json.loads(marker.read_text()); record['schema'] = 'corrupt'
|
|
marker.write_text(json.dumps(record))
|
|
with self.assertRaises(ValueError):
|
|
m.assert_bound(path, p, anchor_file=__file__, origin='x', validate=validate)
|
|
|
|
def test_unsafe_mutex_failure_does_not_leak_descriptor(self):
|
|
m = load()
|
|
with tempfile.TemporaryDirectory() as td:
|
|
path = Path(td) / 'publication.lock'; path.write_text(''); path.chmod(0o644)
|
|
def live_fds():
|
|
found = set()
|
|
for name in os.listdir('/proc/self/fd'):
|
|
try: os.readlink('/proc/self/fd/' + name)
|
|
except FileNotFoundError: continue
|
|
found.add(name)
|
|
return found
|
|
before = live_fds()
|
|
with self.assertRaises(ValueError): m.acquire_lock(Path(td))
|
|
after = live_fds()
|
|
leaked = after - before
|
|
for value in leaked:
|
|
try: os.close(int(value))
|
|
except OSError: pass
|
|
self.assertEqual(after, before)
|
|
|
|
if __name__ == '__main__': unittest.main()
|