"""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()