mirror of
https://github.com/PRVTPRO/Amnezia-Web-Panel.git
synced 2026-09-18 01:44:10 +02:00
hell
539ef92197
Both files were submitted in #143 and #147, before the exit node existed here, so their exit-specific halves had to be stripped to keep them importable. That code is in the tree since #149, and the stripped copies now test less than they could: - test_ssh_error_text.py gains ExitLinkMessageTests: an SSH whose sudo call returns -1 with no output at all (a dropped transport) must make exit_link and exit_unlink raise a message that names the exit code, not "exit link apply failed: " with nothing after the colon. That is the failure the empty str(EOFError) produced, and it is the reason #147 exists. - test_sudo_command_chains.py pins amnezia-exit, not amnezia-awg2, as the representative container that DNSManager attaches to amnezia-dns-net. Both are in vpn_containers, but only the exit one silently breaks DNS on an exit node when it falls out of the list. No production code changes.
108 lines
3.9 KiB
Python
108 lines
3.9 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)
|
|
# the exit-node container is among the ones attached
|
|
self.assertTrue(any('amnezia-exit' in c for c in 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()
|