118 lines
6.3 KiB
Python
118 lines
6.3 KiB
Python
"""Activation guard tests; privileged success is qualified in the installed VM."""
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).parents[1]))
|
|
import os_updater_activate as A
|
|
|
|
|
|
class ActivationGuards(unittest.TestCase):
|
|
def test_owner_policy_allows_required_locks_but_not_os_package_roots(self):
|
|
for root in A.OS_ROOTS:
|
|
paths=A.owner_policy(root).decode().split('ReadWritePaths=')[1].split()
|
|
self.assertEqual(set(paths),{'/var/lib/dpkg/lock-frontend','/var/lib/dpkg/lock',
|
|
str(root/'transactions'),str(root/'executor.lock')})
|
|
self.assertNotIn(str(root),paths)
|
|
self.assertNotIn('/var/lib/dpkg',paths)
|
|
self.assertEqual(A.OWNER_POLICY,A.owner_policy(A.OS_ROOTS[0]))
|
|
|
|
def test_only_pinned_fixed_executor_storage_abi_is_accepted(self):
|
|
for root in A.OS_ROOTS:
|
|
raw=f"ROOT = pathlib.Path({str(root)!r})\n".encode()
|
|
with patch.object(A.bootstrap,'read',return_value=raw):
|
|
self.assertEqual(A.declared_root(Path('/pinned-package')),root)
|
|
for raw in (b"ROOT = pathlib.Path('/home/archipelago/os')\n",
|
|
b"ROOT = pathlib.Path(os.environ['ROOT'])\n",
|
|
b"ROOT = pathlib.Path('/var/lib/archipelago-os-update')\nROOT = evil()\n"):
|
|
with patch.object(A.bootstrap,'read',return_value=raw), self.assertRaises((ValueError,TypeError)):
|
|
A.declared_root(Path('/pinned-package'))
|
|
|
|
def test_unprotected_destination_is_rejected_without_creation(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
destination = Path(directory)/'os-update/plans'
|
|
with self.assertRaisesRegex(ValueError,'custody'):
|
|
A.destination_custody(destination)
|
|
self.assertFalse(destination.parent.exists())
|
|
|
|
def test_interrupted_activation_refuses_before_any_command(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
(root/'activation-interrupted').mkdir()
|
|
(root/'activation-interrupted/INTENT').write_bytes(b'consumed')
|
|
approval = dict(version=1, prepared_sha256='0'*64,package_root='/unused',
|
|
package_files={},admission_sha256='1'*64,os_inputs={},helper_sha256='2'*64,
|
|
owner_dropin_sha256=A.bootstrap.digest(A.OWNER_POLICY))
|
|
with patch.object(A, 'BASE', root), patch.object(A.os,'geteuid',return_value=0), patch.object(
|
|
A,'same',return_value=json.dumps(approval).encode()), patch.object(A,'protected_dir'), patch.object(
|
|
A.bootstrap,'secure',side_effect=lambda p: os.open(p,os.O_RDONLY)), patch.object(A,'command') as command:
|
|
with self.assertRaisesRegex(ValueError, 'existing activation intent'):
|
|
A.activate(root/'approval.json','3'*64)
|
|
command.assert_not_called()
|
|
self.assertEqual((root/'activation-interrupted/INTENT').read_bytes(),b'consumed')
|
|
|
|
def test_nonroot_never_reads_approval(self):
|
|
with patch.object(A.os, 'geteuid', return_value=1000), patch.object(A, 'same') as read:
|
|
with self.assertRaisesRegex(ValueError, 'root administrator'):
|
|
A.activate(Path('/not-read'), '0'*64)
|
|
read.assert_not_called()
|
|
|
|
def test_all_package_writers_must_be_masked_and_drained(self):
|
|
for bad in ('ActiveState=active\nUnitFileState=masked\n',
|
|
'ActiveState=inactive\nUnitFileState=enabled\n'):
|
|
with patch.object(A, 'command', return_value=bad):
|
|
with self.assertRaisesRegex(ValueError, 'writer exclusion'):
|
|
A.current_exclusion()
|
|
def state(*args):
|
|
return ('MainPID=42\nActiveState=active\n' if args[2]=='archipelago.service'
|
|
else 'ActiveState=inactive\nUnitFileState=masked\n')
|
|
with patch.object(A, 'command', side_effect=state):
|
|
with self.assertRaisesRegex(ValueError, 'externally drained'):
|
|
A.current_exclusion()
|
|
|
|
def test_unresolved_os_history_not_reinitialized(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
(root/'transactions').mkdir()
|
|
(root/'transactions/old-authority').mkdir()
|
|
sentinel = root/'transactions/old-authority/status.json'
|
|
sentinel.write_bytes(b'APPLY_INTENT')
|
|
with patch.object(A, 'OS_ROOT', root), patch.object(A, 'OS_ROOTS', (root,)), patch.object(
|
|
A, 'D09', root/'d09'), patch.object(A, 'protected_dir'), patch.object(
|
|
A, 'Path', side_effect=lambda p: root/'binding.json' if p=='/etc/archipelago-d09/binding.json' else Path(p)):
|
|
with self.assertRaisesRegex(ValueError, 'existing OS transaction'):
|
|
A.untouched_authority()
|
|
self.assertEqual(sentinel.read_bytes(), b'APPLY_INTENT')
|
|
|
|
def test_d09_history_not_reinitialized(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
(root/'ledger.json').write_bytes(b'uncertain')
|
|
with patch.object(A, 'D09', root):
|
|
with self.assertRaisesRegex(ValueError, 'existing D09 authority'):
|
|
A.untouched_authority()
|
|
self.assertEqual((root/'ledger.json').read_bytes(), b'uncertain')
|
|
|
|
def test_package_inventory_checks_files_and_directory_aliases(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
(root/'file').write_bytes(b'code')
|
|
with patch.object(A, 'protected_dir'), patch.object(A.bootstrap, 'read', side_effect=lambda p: p.read_bytes()):
|
|
self.assertEqual(A.package_inventory(root), {'file': A.bootstrap.digest(b'code')})
|
|
(root/'alias').symlink_to(root, target_is_directory=True)
|
|
def protected(path):
|
|
if path.is_symlink():
|
|
raise ValueError('alias')
|
|
with patch.object(A, 'protected_dir', side_effect=protected), patch.object(A.bootstrap, 'read', side_effect=lambda p: p.read_bytes()):
|
|
with self.assertRaisesRegex(ValueError, 'alias'):
|
|
A.package_inventory(root)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|