178 lines
8 KiB
Python
178 lines
8 KiB
Python
#!/usr/bin/env python3
|
|
"""File-backed authorization attacks and real ledger tests, never runtime.
|
|
|
|
CONTRACT: immutable LAB/historical files are never edited by negative tests.
|
|
Alternate records model received corrupt inputs; actual verification functions
|
|
still execute. Manifest attacks use real disposable files and real SHA checks.
|
|
"""
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
import handoff as h
|
|
|
|
LAB = None
|
|
|
|
|
|
class Gate(unittest.TestCase):
|
|
def auth_attack(self, key, value):
|
|
auth = h.read(LAB / 'results/authorization.json')
|
|
auth[key] = value
|
|
with tempfile.TemporaryDirectory(prefix='nested-auth-negative-') as tmp:
|
|
p = Path(tmp) / 'authorization.json'
|
|
p.write_text(json.dumps(auth))
|
|
with self.assertRaises((ValueError, AssertionError, OSError, KeyError)):
|
|
h.preflight(LAB, p, review=False)
|
|
|
|
def record_attack(self, name, change):
|
|
actual = h.read
|
|
def read(path):
|
|
value = actual(path)
|
|
if Path(path) == LAB / 'results' / name:
|
|
value = copy.deepcopy(value)
|
|
change(value)
|
|
return value
|
|
with patch.object(h, 'read', read):
|
|
with self.assertRaises((ValueError, AssertionError, OSError, KeyError)):
|
|
h.preflight(LAB, LAB / 'results/authorization.json', review=False)
|
|
|
|
def test_positive_real_preflight_without_review(self):
|
|
r = h.preflight(LAB, LAB / 'results/authorization.json', review=False)
|
|
self.assertEqual(r['handoff'], 'PENDING_INDEPENDENT_REVIEW')
|
|
self.assertFalse(r['run_reserved'])
|
|
|
|
def test_no_authorization(self):
|
|
with self.assertRaises(ValueError):
|
|
h.preflight(LAB, None, review=False)
|
|
|
|
def test_old_ticket(self):
|
|
self.auth_attack('ticket', 'SWAYFX_MOSAIC_STEP20C_NESTED_OFFLINE_FREEZE_AND_RECEIPT_INTEGRITY_V1')
|
|
|
|
def test_wrong_lab(self):
|
|
self.auth_attack('registered_lab', str(h.OLD_LAB))
|
|
|
|
def test_exhausted_historical_lab(self):
|
|
auth = h.read(LAB / 'results/authorization.json')
|
|
old = Path('/home/fy59/swayfx-mosaic-static-nested-grab-m0a5e2qh')
|
|
auth['registered_lab'] = str(old)
|
|
with tempfile.TemporaryDirectory(prefix='nested-old-lab-negative-') as tmp:
|
|
p = Path(tmp) / 'authorization.json'
|
|
p.write_text(json.dumps(auth))
|
|
with self.assertRaises(ValueError):
|
|
h.preflight(old, p, review=False)
|
|
|
|
def test_wrong_owner(self):
|
|
self.auth_attack('owner_registry', '/tmp/nonexistent-old-campaign-owner')
|
|
|
|
def test_wrong_uid(self):
|
|
self.auth_attack('uid', h.os.getuid() + 1)
|
|
|
|
def test_wrong_binary(self):
|
|
self.auth_attack('binary', '/usr/bin/sway')
|
|
|
|
def test_wrong_binary_hash(self):
|
|
self.auth_attack('binary_sha256', '0' * 64)
|
|
|
|
def test_wrong_wlroots(self):
|
|
self.auth_attack('wlroots_sha256', '0' * 64)
|
|
|
|
def test_wrong_fixture_hash(self):
|
|
self.auth_attack('fixture_sha256', '0' * 64)
|
|
|
|
def test_review_not_accepted_original_gate(self):
|
|
# The unchanged authorize() reads the actual file-backed gate/review.
|
|
# Preserve original accepted bytes in an isolated negative-test LAB,
|
|
# then downgrade its review; no accepted review is invented.
|
|
with tempfile.TemporaryDirectory(prefix='nested-review-negative-') as tmp:
|
|
lab = Path(tmp)
|
|
(lab / 'results').mkdir()
|
|
auth = h.read(LAB / 'results/authorization.json')
|
|
auth['registered_lab'] = str(lab)
|
|
(lab / 'auth.json').write_text(json.dumps(auth))
|
|
(lab / 'results/offline-gate.json').write_bytes((LAB / 'results/offline-gate.json').read_bytes())
|
|
review = h.read(LAB / 'results/independent-review.json')
|
|
review['status'] = 'CORRECTION_REQUIRED'
|
|
(lab / 'results/independent-review.json').write_text(json.dumps(review))
|
|
with self.assertRaises(ValueError):
|
|
h.authorize(lab, lab / 'auth.json')
|
|
|
|
def test_offline_gate_not_ready_original_gate(self):
|
|
with tempfile.TemporaryDirectory(prefix='nested-ready-negative-') as tmp:
|
|
lab = Path(tmp)
|
|
(lab / 'results').mkdir()
|
|
auth = h.read(LAB / 'results/authorization.json')
|
|
auth['registered_lab'] = str(lab)
|
|
(lab / 'auth.json').write_text(json.dumps(auth))
|
|
gate = h.read(LAB / 'results/offline-gate.json')
|
|
gate['marker'] = 'INCONCLUSIVE'
|
|
(lab / 'results/offline-gate.json').write_text(json.dumps(gate))
|
|
(lab / 'results/independent-review.json').write_bytes((LAB / 'results/independent-review.json').read_bytes())
|
|
with self.assertRaises(ValueError):
|
|
h.authorize(lab, lab / 'auth.json')
|
|
|
|
def test_missing_manifest(self):
|
|
actual = h.verify
|
|
with tempfile.TemporaryDirectory(prefix='nested-manifest-negative-') as tmp:
|
|
missing = Path(tmp) / 'absent'
|
|
def verify(path):
|
|
return actual(missing if Path(path) == LAB / 'results/fixture-freeze-sha256.txt' else path)
|
|
with patch.object(h, 'verify', verify), self.assertRaises(OSError):
|
|
h.preflight(LAB, LAB / 'results/authorization.json', review=False)
|
|
|
|
def test_changed_manifest_entry(self):
|
|
actual = h.verify
|
|
with tempfile.TemporaryDirectory(prefix='nested-manifest-negative-') as tmp:
|
|
manifest = Path(tmp) / 'fixture-freeze-sha256.txt'
|
|
rows = (LAB / 'results/fixture-freeze-sha256.txt').read_text().splitlines(True)
|
|
rows[0] = '0' * 64 + rows[0][64:]
|
|
manifest.write_text(''.join(rows))
|
|
def verify(path):
|
|
return actual(manifest if Path(path) == LAB / 'results/fixture-freeze-sha256.txt' else path)
|
|
with patch.object(h, 'verify', verify), self.assertRaises(AssertionError):
|
|
h.preflight(LAB, LAB / 'results/authorization.json', review=False)
|
|
|
|
def test_run_already_consumed(self):
|
|
def consume(b):
|
|
b['used']['run'] = 1
|
|
b['events'] = [dict(seq=0, kind='run', ns=1)]
|
|
self.record_attack('budget.json', consume)
|
|
|
|
def test_dispatch_already_consumed(self):
|
|
with tempfile.TemporaryDirectory(prefix='nested-consumed-negative-') as tmp:
|
|
p = Path(tmp) / 'spent'
|
|
p.write_text('consumed')
|
|
with patch.object(h, 'SPENT', p), self.assertRaises(ValueError):
|
|
h.preflight(LAB, LAB / 'results/authorization.json', review=False)
|
|
|
|
def test_durable_exclusive_consumption(self):
|
|
with tempfile.TemporaryDirectory(prefix='nested-exclusive-test-') as tmp:
|
|
p = Path(tmp) / 'spent'
|
|
h.exclusive(p, dict(ticket=h.TICKET))
|
|
before = p.read_bytes()
|
|
with self.assertRaises(FileExistsError):
|
|
h.exclusive(p, dict(ticket='second'))
|
|
self.assertEqual(p.read_bytes(), before)
|
|
|
|
def test_ledger_exhaustion_and_atomic_bundle(self):
|
|
from ledger import initial_for_offline_test, reserve
|
|
with tempfile.TemporaryDirectory(prefix='geometry-ledger-test-') as tmp:
|
|
p = Path(tmp) / 'budget.json'
|
|
initial_for_offline_test(p)
|
|
reserve(p, 'run')
|
|
before = p.read_bytes()
|
|
with self.assertRaises(ValueError):
|
|
reserve(p, 'run', 'click')
|
|
self.assertEqual(p.read_bytes(), before)
|
|
self.assertEqual(h.valid(h.read(p))['used']['run'], 1)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
import sys
|
|
LAB = Path(sys.argv[1]).resolve()
|
|
result = unittest.TextTestRunner(verbosity=2).run(unittest.defaultTestLoader.loadTestsFromTestCase(Gate))
|
|
h.exclusive(LAB / 'results/preflight-tests.json', dict(status='PASS' if result.wasSuccessful() else 'FAIL',
|
|
tests=result.testsRun, failures=len(result.failures), errors=len(result.errors),
|
|
source_sha256=h.sha(Path(__file__)), run_reserved=False, runtime_runs=0))
|
|
raise SystemExit(0 if result.wasSuccessful() else 1)
|