mirror of
https://github.com/PRVTPRO/Amnezia-Web-Panel.git
synced 2026-09-18 01:44:10 +02:00
hell
95bfd42ff8
- app.py handed every /static response `max-age=15552000, immutable`, but only the assets built through static_version() carry ?v=<mtime>. The favicon, the icons, qrcode.min.js, searchable-select.js, the vendored CodeMirror bundles and ReDoc keep one URL forever, so a redeploy could not reach a browser that had already cached them. Immutable is now limited to fingerprinted URLs; the rest get an hour plus a revalidation. - WireGuardManager._apply_bw_limits ran `docker cp ... && docker exec ...` through run_sudo_command, which only privileges the head of the chain, so the exec ran as the unprivileged SSH user. Wrapped in one `sh -c`, the pattern tests/test_sudo_command_chains.py enforces. - toggleConnection() restored the row only in the error path. When loadConnections() fails after a successful toggle, the row it was going to rebuild stays disabled with a spinning bulb. Restore now runs in `finally`. - The add-user modal starts on role "none", where the password is optional, but the label said "required" until the select fired its change event. Tests: the bandwidth chain joins the sudo-chain suite, and a new tests/test_static_cache.py pins both cache lifetimes.
108 lines
4.0 KiB
Python
108 lines
4.0 KiB
Python
"""A privileged command chain must run inside one shell.
|
|
|
|
`SSHManager.run_sudo_command` builds `echo <pw> | sudo -S -p '' <command>`, so
|
|
in `sudo a && b` only `a` is privileged: `b` runs as the panel's unprivileged
|
|
SSH user. Wherever both halves need root, the chain has to be handed to a
|
|
single `sh -c` (the pattern the rest of BackupManager already uses).
|
|
"""
|
|
|
|
import unittest
|
|
|
|
from managers.backup_manager import BackupManager
|
|
from managers.adguard_manager import AdguardManager
|
|
from managers.dns_manager import DNSManager
|
|
from managers.wireguard_manager import WireGuardManager
|
|
|
|
|
|
class RecordingSSH:
|
|
def __init__(self, code=1):
|
|
self.commands = []
|
|
self.code = code
|
|
|
|
def run_sudo_command(self, command, timeout=60):
|
|
self.commands.append(command)
|
|
return '', 'not found', self.code
|
|
|
|
def run_command(self, command, timeout=60):
|
|
self.commands.append(command)
|
|
# DNSManager gates its install on this probe
|
|
return ('Docker version 27.0.3' if 'docker --version' in command else ''), '', 0
|
|
|
|
def write_file(self, path, content):
|
|
pass
|
|
|
|
def upload_file(self, content, path):
|
|
pass
|
|
|
|
|
|
def chain_is_wrapped(command):
|
|
"""True when every `&&`/`||`/`|` sits inside a `sh -c`/`bash -c` argument."""
|
|
for token in ('&&', '||', '|'):
|
|
idx = command.find(token)
|
|
if idx == -1:
|
|
continue
|
|
shell = min((command.find(s) for s in ('sh -c', 'bash -c') if command.find(s) != -1), default=-1)
|
|
if shell == -1 or shell > idx:
|
|
return False
|
|
return True
|
|
|
|
|
|
class SudoChainTests(unittest.TestCase):
|
|
def test_backup_fetch_runs_the_chain_in_one_shell(self):
|
|
ssh = RecordingSSH()
|
|
BackupManager(ssh)._fetch_remote_archive('/opt/amnezia/backups/awg2/awg2-2026.tar.gz')
|
|
cmd = ssh.commands[0]
|
|
self.assertTrue(cmd.startswith('sh -c '), cmd)
|
|
self.assertTrue(chain_is_wrapped(cmd), cmd)
|
|
# the copy still targets the same paths
|
|
self.assertIn('/opt/amnezia/backups/awg2/awg2-2026.tar.gz', cmd)
|
|
self.assertIn('chmod 0644', cmd)
|
|
|
|
def test_dns_network_creation_runs_the_chain_in_one_shell(self):
|
|
ssh = RecordingSSH(code=0)
|
|
try:
|
|
DNSManager(ssh).install_protocol()
|
|
except Exception:
|
|
pass
|
|
network = [c for c in ssh.commands if 'amnezia-dns-net' in c and 'network' in c]
|
|
self.assertTrue(network, ssh.commands)
|
|
self.assertTrue(chain_is_wrapped(network[0]), network[0])
|
|
|
|
def test_dns_attaches_existing_containers_in_one_shell(self):
|
|
ssh = RecordingSSH(code=0)
|
|
try:
|
|
DNSManager(ssh).install_protocol()
|
|
except Exception:
|
|
pass
|
|
connects = [c for c in ssh.commands if 'network connect' in c]
|
|
self.assertTrue(connects, ssh.commands)
|
|
for cmd in connects:
|
|
self.assertTrue(chain_is_wrapped(cmd), cmd)
|
|
# a representative container is among the ones attached
|
|
self.assertTrue(any('amnezia-awg2' in c for c in connects), connects)
|
|
|
|
def test_adguard_network_creation_runs_the_chain_in_one_shell(self):
|
|
ssh = RecordingSSH(code=0)
|
|
AdguardManager(ssh)._ensure_network()
|
|
self.assertTrue(chain_is_wrapped(ssh.commands[0]), ssh.commands[0])
|
|
|
|
def test_wireguard_bw_limits_run_the_chain_in_one_shell(self):
|
|
ssh = RecordingSSH(code=0)
|
|
manager = WireGuardManager(ssh)
|
|
manager.check_container_running = lambda: True
|
|
manager._apply_bw_limits([
|
|
{'userData': {'maxSpeed': 10, 'clientIp': '10.8.2.2'}},
|
|
])
|
|
execs = [c for c in ssh.commands if 'docker exec' in c and '_wg_tc.sh' in c]
|
|
self.assertTrue(execs, ssh.commands)
|
|
for cmd in execs:
|
|
self.assertTrue(chain_is_wrapped(cmd), cmd)
|
|
|
|
def test_helper_flags_a_bare_chain(self):
|
|
self.assertFalse(chain_is_wrapped("test -f a && cp a b"))
|
|
self.assertTrue(chain_is_wrapped("sh -c 'test -f a && cp a b'"))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|