Mirrors/ProxMenux
1
0
Fork 0
mirror of https://github.com/MacRimi/ProxMenux.git synced 2026-11-02 02:42:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: distinguish the waiting unattended upgrades shutdown helper

Browse Source
This commit is contained in:
VAIO73
2026-10-28 17:30:44 +00:00
parent 65b10cea09
commit bdf9075639
3 changed files with 142 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AppImage/scripts/host_update_controls.py
+14 -2
View File
@@ -29,10 +29,22 @@ def package_process_active():
if not path.name.isdigit():
continue
try:
if (path/'comm').read_text().strip() in ('apt', 'apt-get', 'dpkg', 'unattended-upgr'):
name = (path/'comm').read_text().strip()
if name == 'unattended-upgr':
# Linux comm truncates both the updater and its persistent
# shutdown helper to 15 bytes. Exempt only the exact idle
# service command (also with its system Python shebang).
argv = (path/'cmdline').read_bytes().split(b'\0')
helper = b'/usr/share/unattended-upgrades/unattended-upgrade-shutdown'
if argv in ([helper, b'--wait-for-signal', b''],
[b'/usr/bin/python3', helper, b'--wait-for-signal', b'']):
continue
return True
if name in ('apt', 'apt-get', 'dpkg'):
return True
except (FileNotFoundError, ProcessLookupError):
continue
if path.exists():
return True
except OSError:
return True
return False
tests/test_host_update_controls.py
+114 -8
View File
@@ -10,6 +10,7 @@ import sys
import tempfile
from types import ModuleType, SimpleNamespace
import unittest
from contextlib import contextmanager
from unittest.mock import Mock, patch
from flask import Flask
@@ -19,6 +20,10 @@ sys.path.insert(0, str(ROOT/'AppImage/scripts'))
import host_update_controls as controls
import proxmenux_version_status as versions
SHUTDOWN_HELPER = b'/usr/share/unattended-upgrades/unattended-upgrade-shutdown'
WAITING_HELPER = b'/usr/bin/python3\0' + SHUTDOWN_HELPER + b'\0--wait-for-signal\0'
REAL_UPDATER = b'/usr/bin/python3\0/usr/bin/unattended-upgrade\0'
class VersionStatus(unittest.TestCase):
def setUp(self):
@@ -103,6 +108,68 @@ class RebootGuards(unittest.TestCase):
self.probe=patch.object(controls,'reboot_status',return_value={'state':'required'}).start()
self.addCleanup(patch.stopall)
def package(self, command=WAITING_HELPER, name='unattended-upgr', pid='123'):
path=self.root/'proc'/pid;path.mkdir(exist_ok=True)
(path/'comm').write_text(name+'\n')
(path/'cmdline').write_bytes(command)
return path
def test_waiting_shutdown_helper_does_not_block_required_reboot_or_run_commands(self):
# Debian's service ExecStart, including the /usr/bin/python3 shebang.
for command in (WAITING_HELPER, SHUTDOWN_HELPER+b'\0--wait-for-signal\0'):
with self.subTest(command=command),patch.object(controls.subprocess,'run') as run:
self.package(command)
self.assertEqual(controls.reboot_context()['reason'],'required')
run.assert_not_called()
def test_actual_package_tools_and_unattended_updater_still_block_before_probe(self):
for name in ('apt','apt-get','dpkg','unattended-upgr'):
with self.subTest(name=name):
self.package(REAL_UPDATER,name)
self.assertEqual(controls.reboot_context()['reason'],'busy')
self.probe.assert_not_called()
def test_waiting_helper_never_hides_another_active_updater(self):
self.package()
self.package(REAL_UPDATER,pid='124')
self.assertEqual(controls.reboot_context()['reason'],'busy')
self.probe.assert_not_called()
def test_helper_without_wait_flag_and_lookalike_arguments_stay_busy(self):
commands=(SHUTDOWN_HELPER+b'\0',
b'/usr/bin/unattended-upgrade\0--wait-for-signal\0',
b'/usr/bin/unattended-upgrade\0'+SHUTDOWN_HELPER+b'\0--wait-for-signal\0',
b'/bin/sh\0-c\0'+SHUTDOWN_HELPER+b' --wait-for-signal\0',
b'/usr/bin/python3\0-c\0'+SHUTDOWN_HELPER+b'\0--wait-for-signal\0',
WAITING_HELPER+b'--unexpected\0',
b'/tmp/unattended-upgrade-shutdown\0--wait-for-signal\0')
for command in commands:
with self.subTest(command=command):
self.package(command)
self.assertEqual(controls.reboot_context()['reason'],'busy')
self.probe.assert_not_called()
def test_empty_partial_malformed_and_missing_live_cmdline_stay_busy(self):
for command in (b'',b'\0',WAITING_HELPER[:-1],WAITING_HELPER+b'\0',b'\xff\0'):
with self.subTest(command=command):
self.package(command)
self.assertEqual(controls.reboot_context()['reason'],'busy')
(self.root/'proc/123/cmdline').unlink()
self.assertEqual(controls.reboot_context()['reason'],'busy')
self.probe.assert_not_called()
def test_unreadable_cmdline_blocks_but_process_that_exits_can_be_skipped(self):
process=self.package()
with patch.object(Path,'read_bytes',side_effect=PermissionError('fixture denied')):
self.assertEqual(controls.reboot_context()['reason'],'busy')
self.probe.assert_not_called()
def exited():
for path in process.iterdir():path.unlink()
process.rmdir()
raise FileNotFoundError('fixture exited after comm read')
with patch.object(Path,'read_bytes',side_effect=exited):
self.assertTrue(controls.reboot_context()['allowed'])
def test_active_attention_starting_and_unreadable_jobs_block_before_kernel_probe(self):
for state in ('starting','running','attention','finalizing',None):
self.jobs.return_value=[{'state':state}]
@@ -163,14 +230,17 @@ class RebootGuards(unittest.TestCase):
for path in paths:Path(path).touch()
# A separate process holds the same POSIX lock, just as APT does.
code='import fcntl,sys,time; f=open(sys.argv[1],"r+"); fcntl.lockf(f,fcntl.LOCK_EX); print("locked",flush=True); time.sleep(5)'
child=subprocess.Popen([sys.executable,'-c',code,paths[0]],stdout=subprocess.PIPE,text=True)
try:
self.assertEqual(child.stdout.readline().strip(),'locked')
with patch.object(controls,'PACKAGE_LOCKS',paths),patch.object(controls.subprocess,'run') as reboot:
self.assertEqual(controls.execute_reboot(),75)
reboot.assert_not_called()
finally:
child.terminate();child.wait(timeout=2);child.stdout.close()
for held in paths:
child=subprocess.Popen([sys.executable,'-c',code,held],stdout=subprocess.PIPE,text=True)
try:
self.assertEqual(child.stdout.readline().strip(),'locked')
with patch.object(controls,'PACKAGE_LOCKS',paths),patch.object(controls.subprocess,'run') as reboot:
self.assertEqual(controls.execute_reboot(),75)
reboot.assert_not_called()
self.probe.assert_not_called()
self.assertEqual([Path(path).read_bytes() for path in paths],[b'',b''])
finally:
child.terminate();child.wait(timeout=2);child.stdout.close()
class Routes(unittest.TestCase):
@@ -191,6 +261,7 @@ class Routes(unittest.TestCase):
self.app=Flask(__name__);self.app.register_blueprint(routes.health_bp)
self.run=Mock(return_value=(True,{}));self.state=Mock(return_value={'state':'running'})
self.app.extensions['host_update_power']={'run':self.run,'state':self.state}
self.real_context=controls.reboot_context
self.context=patch.object(controls,'reboot_context',return_value={'allowed':True,'reason':'required','boot_id':'fixture-boot'}).start()
self.version=patch.object(versions,'version_status',return_value={'state':'unknown'}).start()
self.addCleanup(patch.stopall)
@@ -198,6 +269,41 @@ class Routes(unittest.TestCase):
import socket
self.body={'confirm':True,'hostname':socket.gethostname(),'boot_id':'fixture-boot'}
@contextmanager
def actual_package_context(self,command):
with tempfile.TemporaryDirectory() as folder:
root=Path(folder);proc=root/'proc';(proc/'123').mkdir(parents=True)
(proc/'123/comm').write_text('unattended-upgr\n')
(proc/'123/cmdline').write_bytes(command)
boot=root/'boot_id';boot.write_text('fixture-boot')
with patch.object(controls,'PROC',proc),patch.object(controls,'BOOT_ID',boot),\
patch.object(controls,'LOCK',str(root/'missing-lock')),\
patch.object(controls,'list_jobs',return_value=[]),\
patch.object(controls,'reboot_status',return_value={'state':'required'}) as probe,\
patch.object(controls,'reboot_context',wraps=self.real_context):
yield probe
def test_actual_context_routes_allow_waiting_helper_and_only_queue_guarded_action(self):
with self.actual_package_context(WAITING_HELPER):
response=self.client.get('/api/health/reboot-context',headers={'Authorization':'Bearer admin'})
self.assertEqual(response.status_code,200)
self.assertTrue(response.json['allowed'])
self.run.assert_not_called()
response=self.client.post('/api/health/reboot-after-update',json=self.body,headers={'Authorization':'Bearer admin'})
self.assertEqual(response.status_code,202)
self.run.assert_called_once_with(controls.UNIT,list(controls.COMMAND),description='ProxMenux: reboot after host update',extra_properties=['RuntimeMaxSec=75'])
def test_actual_context_routes_deny_active_or_uncertain_updater_without_launch(self):
for command in (REAL_UPDATER,b'',WAITING_HELPER[:-1]):
with self.subTest(command=command),self.actual_package_context(command) as probe:
response=self.client.get('/api/health/reboot-context',headers={'Authorization':'Bearer admin'})
self.assertEqual(response.status_code,200)
self.assertEqual(response.json['reason'],'busy')
response=self.client.post('/api/health/reboot-after-update',json=self.body,headers={'Authorization':'Bearer admin'})
self.assertEqual(response.status_code,409)
self.assertEqual(response.json['reason'],'busy')
probe.assert_not_called();self.run.assert_not_called()
def test_denied_auth_and_methods_do_not_probe_or_launch(self):
for token in (None,'bad','reader'):
headers={'Authorization':'Bearer '+token} if token else {}
tests/test_host_update_shell.py
+14 -4
View File
@@ -5,6 +5,7 @@ are inert executables which record calls; no root, network or host update.
"""
import os
from pathlib import Path
import re
import subprocess
import sys
import tempfile
@@ -70,12 +71,18 @@ msg_success() { echo "$*"; }
show_proxmenux_logo() { :; }
LOCAL_SCRIPTS="$FIXTURE/scripts"
''')
# Keep the real confirmation contract added upstream; stub only the
# package/host effects, not the updater's shared decision helper.
confirmation = re.search(r'^pmx_confirm_update\(\) \{\n.*?^\}\n',
(ROOT / 'scripts/global/common-functions.sh').read_text(),
re.MULTILINE | re.DOTALL)
self.assertIsNotNone(confirmation)
(scripts / 'global/common-functions.sh').write_text('''
pmx_format_upgradable() { cat; }
pmx_dpkg_changes_since() { echo 'package changes fixture'; }
lvm_repair_check() { echo lvm-poststep >> "$CALLS"; }
pmx_rebuild_dkms_after_kernel() { echo dkms-poststep >> "$CALLS"; }
''')
''' + confirmation.group(0))
(scripts / 'global/utils-install-functions.sh').write_text('ensure_repositories() { return 0; }\n')
commands = {
'pveversion': "echo 'pve-manager/9.2.24/test (running kernel: 7.0.14-25-pve)'",
@@ -93,7 +100,9 @@ else
exit "${UPGRADE_EXIT:-0}"
fi''',
'dialog': 'echo dialog >> "$CALLS"; exit "${CONFIRM_EXIT:-0}"',
'whiptail': 'echo "whiptail $*" >> "$CALLS"; exit 1',
'whiptail': '''echo "whiptail $*" >> "$CALLS"
if [[ "$1" == --title && "$2" == "Proxmox Update" ]]; then exit "${CONFIRM_EXIT:-0}"; fi
exit 1''',
'reboot': 'echo FORBIDDEN-REBOOT >> "$CALLS"; exit 99',
'clear': ':',
}
@@ -139,7 +148,8 @@ fi''',
self.assertIn('apt-get -y autoremove', calls)
self.assertIn('apt-get -y autoclean', calls)
self.assertNotIn('FORBIDDEN-REBOOT', calls)
self.assertNotIn('whiptail', calls)
self.assertIn('whiptail --title Proxmox Update --yesno', calls)
self.assertNotIn('whiptail --title Reboot Required', calls)
self.assertIn('Reboot Required', output)
self.assertIn('You can reboot later manually.', output)
self.assertFalse((self.folder / 'marker-in-progress').exists())
@@ -161,7 +171,7 @@ fi''',
proc = self.start(UPGRADE_EXIT='1')
output, _ = proc.communicate(timeout=5)
self.assertEqual(proc.returncode, 1, output)
for forbidden in ('autoremove', 'autoclean', 'REBOOT', 'poststep', 'whiptail'):
for forbidden in ('autoremove', 'autoclean', 'REBOOT', 'poststep', 'whiptail --title Reboot Required'):
self.assertNotIn(forbidden, self.calls.read_text())
def test_old_menu_cli_contract_retains_explicit_reboot_prompt(self):