163 lines
8.4 KiB
Python
163 lines
8.4 KiB
Python
"""Pure ACL semantics and real kernel xattr round trips on disposable files."""
|
|
import os
|
|
from pathlib import Path
|
|
import stat
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
import json
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
import c01_audio_devices as audio
|
|
|
|
|
|
class ACLTests(unittest.TestCase):
|
|
def test_basic_audio_group_access_becomes_exact_named_uid(self):
|
|
result = audio.decode(audio.grant(None, 0o660, 0, 29, [20,27,29,1000]), 0o660)
|
|
self.assertEqual(audio.permissions(result,0,29,1000,[1000]),6)
|
|
self.assertEqual(audio.permissions(result,0,29,2000,[2000]),0)
|
|
self.assertEqual(audio.permissions(result,0,29,2000,[29]),6)
|
|
self.assertEqual(audio.permissions(result,0,29,0,[0]),6)
|
|
|
|
def test_mask_expansion_never_activates_masked_other_entries(self):
|
|
# Original UID1000 reaches OTHER rw; mask r hides write/execute from
|
|
# named users/groups. Adding its explicit rw must not expose those bits.
|
|
rows = [(1,6,audio.UNDEFINED),(2,7,2000),(4,7,audio.UNDEFINED),
|
|
(8,7,44),(16,4,audio.UNDEFINED),(32,6,audio.UNDEFINED)]
|
|
new = audio.decode(audio.grant(audio.encode(rows),0o646,0,29,[1000]),0o646)
|
|
self.assertEqual(audio.permissions(new,0,29,1000,[1000]),6)
|
|
for uid,groups in ((2000,[2000]),(2001,[29]),(2002,[44]),(2003,[29,44]),(2004,[55]),(0,[0])):
|
|
self.assertEqual(audio.permissions(new,0,29,uid,groups),audio.permissions(rows,0,29,uid,groups))
|
|
|
|
def test_existing_named_uid_takes_precedence_over_supplementary_audio_group(self):
|
|
rows = [(1,6,audio.UNDEFINED),(2,4,1000),(4,6,audio.UNDEFINED),
|
|
(16,6,audio.UNDEFINED),(32,0,audio.UNDEFINED)]
|
|
new = audio.decode(audio.grant(audio.encode(rows),0o660,0,29,[29,1000]),0o660)
|
|
self.assertEqual(audio.permissions(new,0,29,1000,[1000]),4)
|
|
|
|
def test_all_other_group_class_effective_permissions_preserved(self):
|
|
for mask in range(8):
|
|
for user_permissions in range(8):
|
|
rows=[(1,6,audio.UNDEFINED),(2,user_permissions,2000),(4,7,audio.UNDEFINED),
|
|
(8,7,44),(16,mask,audio.UNDEFINED),(32,6,audio.UNDEFINED)]
|
|
new=audio.decode(audio.grant(audio.encode(rows),0o606|(mask<<3),0,29,[1000]),0)
|
|
for uid,groups in ((2000,[29,44]),(2001,[29]),(2002,[44]),(2003,[29,44]),(2004,[55])):
|
|
self.assertEqual(audio.permissions(new,0,29,uid,groups),audio.permissions(rows,0,29,uid,groups))
|
|
|
|
def test_no_prior_access_is_not_an_authority_to_grant(self):
|
|
with self.assertRaises(ValueError):audio.grant(None,0o600,0,29,[29,1000])
|
|
with self.assertRaises(ValueError):audio.grant(None,0o670,0,29,[29,1000])
|
|
|
|
def test_malformed_acl_refused(self):
|
|
for raw in (b'', b'\x02'*28, audio.encode([(1,6,audio.UNDEFINED)]),
|
|
audio.encode([(1,6,audio.UNDEFINED),(4,6,audio.UNDEFINED),(16,6,audio.UNDEFINED),
|
|
(16,6,audio.UNDEFINED),(32,0,audio.UNDEFINED)])):
|
|
with self.subTest(raw=raw),self.assertRaises(ValueError):audio.decode(raw,0o660)
|
|
|
|
def test_real_kernel_acl_roundtrip_preserves_preimage_and_mode(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
path=Path(root)/'fake-alsa-object'
|
|
path.touch(mode=0o660)
|
|
os.chmod(path,0o660)
|
|
before=audio.acl(path)
|
|
after=audio.grant(before,0o660,0,29,[29,1000])
|
|
os.setxattr(path,audio.ATTRIBUTE,after)
|
|
self.assertEqual(audio.acl(path),after)
|
|
audio.restore(path,before)
|
|
os.chmod(path,0o660)
|
|
self.assertEqual(audio.acl(path),before)
|
|
self.assertEqual(stat.S_IMODE(path.stat().st_mode),0o660)
|
|
|
|
def test_retained_path_handle_does_not_modify_replacement_inode(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
path=Path(root)/'fake-alsa-object';path.touch(mode=0o660)
|
|
fd=os.open(path,os.O_PATH|os.O_NOFOLLOW)
|
|
try:
|
|
retained=Path('/proc/self/fd')/str(fd)
|
|
os.rename(path,Path(root)/'retained-preimage')
|
|
path.touch(mode=0o600)
|
|
after=audio.grant(None,0o660,0,29,[29,1000])
|
|
os.setxattr(retained,audio.ATTRIBUTE,after)
|
|
self.assertEqual(audio.acl(retained),after)
|
|
self.assertIsNone(audio.acl(path))
|
|
self.assertEqual(stat.S_IMODE(path.stat().st_mode),0o600)
|
|
finally:os.close(fd)
|
|
|
|
|
|
class ReconciliationTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary=tempfile.TemporaryDirectory()
|
|
self.root=Path(self.temporary.name)
|
|
self.path=self.root/'device';self.path.touch();self.path.chmod(0o660)
|
|
self.after=audio.grant(None,0o660,0,29,[29,1000])
|
|
self.record=dict(phase='granted',boot_id=Path('/proc/sys/kernel/random/boot_id').read_text().strip(),
|
|
device={'name':'controlC0'},identity=audio.identity(self.path.stat()),mode=0o660,before=None,after=self.after.hex())
|
|
self.journal=self.root/'receipt.json'
|
|
self.journal.write_text(json.dumps(self.record))
|
|
path=self.path
|
|
class FakeDevice:
|
|
def __init__(self,_):
|
|
self.fd=os.open(path,os.O_PATH|os.O_NOFOLLOW)
|
|
self.handle=Path('/proc/self/fd')/str(self.fd)
|
|
self.original_identity=audio.identity(os.fstat(self.fd))
|
|
def close(self):os.close(self.fd)
|
|
self.patches=[patch.object(audio,'STATE',self.root),patch.object(audio,'protected',side_effect=lambda p,**_:Path(p)),
|
|
patch.object(audio,'Device',FakeDevice),patch.object(audio,'device_path',return_value=self.path),
|
|
patch.object(audio,'runtime_retired')]
|
|
for p in self.patches:p.start()
|
|
|
|
def tearDown(self):
|
|
for p in reversed(self.patches):p.stop()
|
|
self.temporary.cleanup()
|
|
|
|
def test_grant_is_restored_only_after_full_runtime_retirement(self):
|
|
os.setxattr(self.path,audio.ATTRIBUTE,self.after)
|
|
audio.runtime_retired.side_effect=ValueError('runtime still populated')
|
|
with self.assertRaises(ValueError):audio.reconcile()
|
|
self.assertEqual(audio.acl(self.path),self.after)
|
|
audio.runtime_retired.side_effect=None
|
|
audio.reconcile()
|
|
self.assertIsNone(audio.acl(self.path))
|
|
self.assertEqual(json.loads(self.journal.read_text())['phase'],'revoked')
|
|
|
|
def test_unapplied_intent_is_reconciled_without_granting(self):
|
|
self.record['phase']='intent';self.journal.write_text(json.dumps(self.record))
|
|
audio.reconcile()
|
|
self.assertIsNone(audio.acl(self.path))
|
|
self.assertEqual(json.loads(self.journal.read_text())['phase'],'revoked')
|
|
|
|
def test_independent_acl_edit_is_not_rolled_back(self):
|
|
different=audio.encode([(1,6,audio.UNDEFINED),(2,4,1000),(4,6,audio.UNDEFINED),
|
|
(16,6,audio.UNDEFINED),(32,0,audio.UNDEFINED)])
|
|
os.setxattr(self.path,audio.ATTRIBUTE,different)
|
|
with self.assertRaisesRegex(ValueError,'changed independently'):audio.reconcile()
|
|
self.assertEqual(audio.acl(self.path),different)
|
|
self.assertEqual(json.loads(self.journal.read_text())['phase'],'granted')
|
|
|
|
def test_replaced_incarnation_never_receives_old_preimage(self):
|
|
self.path.rename(self.root/'old-device');self.path.touch(mode=0o600)
|
|
with self.assertRaisesRegex(ValueError,'old ALSA incarnation'):audio.reconcile()
|
|
self.assertEqual(stat.S_IMODE(self.path.stat().st_mode),0o600)
|
|
|
|
def test_unplugged_inode_is_retired_without_touching_replacement(self):
|
|
# Keep an O_PATH pin so this fixture cannot accidentally reuse its inode.
|
|
fd=os.open(self.path,os.O_PATH)
|
|
try:
|
|
self.path.unlink();self.path.touch(mode=0o600)
|
|
audio.reconcile()
|
|
self.assertEqual(stat.S_IMODE(self.path.stat().st_mode),0o600)
|
|
self.assertIsNone(audio.acl(self.path))
|
|
self.assertEqual(json.loads(self.journal.read_text())['phase'],'detached')
|
|
finally:os.close(fd)
|
|
|
|
def test_prior_boot_inode_reuse_never_restores_old_acl(self):
|
|
self.record['boot_id']='different-boot';self.journal.write_text(json.dumps(self.record))
|
|
os.setxattr(self.path,audio.ATTRIBUTE,self.after)
|
|
audio.reconcile()
|
|
self.assertEqual(audio.acl(self.path),self.after)
|
|
self.assertEqual(json.loads(self.journal.read_text())['phase'],'expired-boot')
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|