diff --git a/oci/remote/oci_keep_backup.py b/oci/remote/oci_keep_backup.py index 900985fc..665fb010 100644 --- a/oci/remote/oci_keep_backup.py +++ b/oci/remote/oci_keep_backup.py @@ -50,9 +50,12 @@ def before_update(vmid, storage): """A storage that cannot receive the file gets its own backup first.""" if dump_dir(storage) is not None: return None + # Like transaction.run(), scope PVE's creation umask to this child only. + # PBS reads the temporary LXC config as the guest-mapped UID; inheriting + # a private caller umask (077) makes its traversal/config files unreadable. subprocess.run(['vzdump', str(vmid), '--storage', storage, '--mode', 'snapshot', '--compress', 'zstd', '--notes-template', 'ProxMenux OCI: before the image update'], - check=True) + check=True, umask=0o022) return storage diff --git a/oci/tests/test_keep_backup_umask.py b/oci/tests/test_keep_backup_umask.py new file mode 100644 index 00000000..06294193 --- /dev/null +++ b/oci/tests/test_keep_backup_umask.py @@ -0,0 +1,86 @@ +"""PBS backup traversal modes must not loosen the OCI caller's private files.""" +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import textwrap +import unittest + +REMOTE = Path(__file__).resolve().parents[1] / 'remote' + + +class KeptBackupPermissions(unittest.TestCase): + def scenario(self, code, exit_code=0): + with tempfile.TemporaryDirectory() as folder: + root = Path(folder) + binary = root / 'vzdump' + binary.write_text(f'#!{sys.executable}\n' + textwrap.dedent(f''' + import os + from pathlib import Path + import sys + assert sys.argv[1:] == ['4242', '--storage', 'pbs-fixture', '--mode', + 'snapshot', '--compress', 'zstd', '--notes-template', + 'ProxMenux OCI: before the image update'] + root = Path(os.environ['BACKUP_FIXTURE']) + (root / 'pve-tmp').mkdir() + (root / 'pve-tmp/pct.conf').write_text('fixture only') + sys.exit({exit_code}) + ''')) + binary.chmod(0o700) + script = textwrap.dedent(''' + import os + from pathlib import Path + import stat + import subprocess + import sys + sys.path.insert(0, sys.argv[1]) + import oci_instances as instances + import oci_keep_backup as backup + root = Path(os.environ['BACKUP_FIXTURE']) + backup.dump_dir = lambda storage: None + instances.write(root / 'private-before/record.json', {'secret': 'fixture'}) + ''') + textwrap.dedent(code) + result = subprocess.run( + [sys.executable, '-c', script, str(REMOTE)], + env={**os.environ, 'PATH': str(root) + os.pathsep + os.environ['PATH'], + 'BACKUP_FIXTURE': str(root)}, + umask=0o077, capture_output=True, text=True, + ) + self.assertEqual(result.returncode, 0, result.stderr) + + def test_pbs_child_is_readable_while_parent_records_stay_private(self): + self.scenario(''' + assert backup.before_update(4242, 'pbs-fixture') == 'pbs-fixture' + assert stat.S_IMODE((root / 'pve-tmp').stat().st_mode) == 0o755 + assert stat.S_IMODE((root / 'pve-tmp/pct.conf').stat().st_mode) == 0o644 + assert os.umask(0o077) == 0o077 + instances.write(root / 'private-after/record.json', {'secret': 'fixture'}) + for name in ('private-before', 'private-after'): + assert stat.S_IMODE((root / name).stat().st_mode) == 0o700 + assert stat.S_IMODE((root / name / 'record.json').stat().st_mode) == 0o600 + ''') + + def test_backup_failure_still_propagates_without_changing_parent_umask(self): + self.scenario(''' + try: + backup.before_update(4242, 'pbs-fixture') + except subprocess.CalledProcessError as error: + assert error.returncode == 7 + else: + raise AssertionError('backup failure was hidden') + assert os.umask(0o077) == 0o077 + assert stat.S_IMODE((root / 'private-before/record.json').stat().st_mode) == 0o600 + ''', exit_code=7) + + def test_file_storage_keeps_the_existing_no_extra_backup_route(self): + self.scenario(''' + backup.dump_dir = lambda storage: root / 'dump' + assert backup.before_update(4242, 'pbs-fixture') is None + assert not (root / 'pve-tmp').exists() + assert os.umask(0o077) == 0o077 + ''') + + +if __name__ == '__main__': + unittest.main()