138 lines
7.5 KiB
Python
138 lines
7.5 KiB
Python
|
import importlib.util
|
||
|
|
from pathlib import Path
|
||
|
|
import unittest
|
||
|
|
from unittest.mock import patch, MagicMock
|
||
|
|
import tempfile
|
||
|
|
import json
|
||
|
|
|
||
|
|
spec=importlib.util.spec_from_file_location('repair', Path(__file__).parents[1]/'repair-npm-tunnel.py')
|
||
|
|
m=importlib.util.module_from_spec(spec)
|
||
|
|
spec.loader.exec_module(m)
|
||
|
|
DROP='[Container]\nPublishPort=10.77.0.2:18080:80/tcp\nPublishPort=10.77.0.2:18443:443/tcp\n'
|
||
|
|
RULES='''table inet web_tunnel {
|
||
|
|
chain input {
|
||
|
|
type filter hook input priority -10; policy accept;
|
||
|
|
iifname != "wg-web" return
|
||
|
|
ct state established,related accept
|
||
|
|
ip saddr 10.77.0.1 icmp type echo-request accept
|
||
|
|
ip saddr 10.77.0.1 ip daddr 10.77.0.2 tcp dport { 18080, 18443 } accept
|
||
|
|
counter drop
|
||
|
|
}
|
||
|
|
chain forward {
|
||
|
|
type filter hook forward priority -10; policy accept;
|
||
|
|
iifname "wg-web" counter drop
|
||
|
|
oifname "wg-web" counter drop
|
||
|
|
}
|
||
|
|
}'''
|
||
|
|
|
||
|
|
class Plan(unittest.TestCase):
|
||
|
|
def test_preserves_peer_https_and_restricts_redirect(self):
|
||
|
|
drop,rules,dst=m.plan(DROP,RULES)
|
||
|
|
self.assertEqual(dst,'10.77.0.2')
|
||
|
|
self.assertIn('PublishPort=10.77.0.2:18081:80/tcp',drop)
|
||
|
|
self.assertIn('PublishPort=10.77.0.2:18443:443/tcp',drop)
|
||
|
|
self.assertIn('iifname "wg-web" ip saddr 10.77.0.1 ip daddr 10.77.0.2 tcp dport 18080 redirect to :18081',rules)
|
||
|
|
self.assertNotIn('tcp dport { 18080, 18443 } accept',rules)
|
||
|
|
self.assertIn('iifname "wg-web" counter drop',rules)
|
||
|
|
def test_idempotent(self):
|
||
|
|
d,r,_=m.plan(DROP,RULES)
|
||
|
|
self.assertIsNone(m.plan(d,r))
|
||
|
|
def test_standard_fresh_install_untouched(self):
|
||
|
|
self.assertIsNone(m.plan('[Container]\nPublishPort=127.0.0.1:8081:81/tcp',''))
|
||
|
|
def test_no_hardcoded_deployment_address(self):
|
||
|
|
d,r,dst=m.plan(DROP.replace('10.77.0.','10.55.0.'),RULES.replace('10.77.0.','10.55.0.'))
|
||
|
|
self.assertEqual(dst,'10.55.0.2');self.assertIn('ip saddr 10.55.0.1',r)
|
||
|
|
def test_custom_firewall_preserved(self):
|
||
|
|
for r in [RULES+'\ntable inet extra {}',RULES.replace('counter drop','accept'),RULES.replace('wg-web','wg-custom')]:
|
||
|
|
with self.assertRaises(ValueError): m.plan(DROP,r)
|
||
|
|
def test_ambiguous_mapping(self):
|
||
|
|
with self.assertRaises(ValueError): m.plan(DROP+DROP,RULES)
|
||
|
|
def test_wrong_destination(self):
|
||
|
|
with self.assertRaises(ValueError): m.plan(DROP.replace('10.77.0.2','10.77.0.3'),RULES)
|
||
|
|
def test_already_used_mapping(self):
|
||
|
|
with self.assertRaises(ValueError): m.plan(DROP+'PublishPort=10.77.0.2:18081:80/tcp\n',RULES)
|
||
|
|
|
||
|
|
class Migration(unittest.TestCase):
|
||
|
|
def setUp(self):
|
||
|
|
self.temp=tempfile.TemporaryDirectory()
|
||
|
|
self.addCleanup(self.temp.cleanup)
|
||
|
|
self.home=Path(self.temp.name)
|
||
|
|
self.drop=self.home/'.config/containers/systemd/nginx-proxy-manager.container.d/web-tunnel.conf'
|
||
|
|
self.drop.parent.mkdir(parents=True)
|
||
|
|
self.drop.write_text(DROP)
|
||
|
|
self.calls=[];self.rules=RULES;self.fail=None
|
||
|
|
self.state=self.home/'.local/state/archipelago/npm-tunnel-migration'
|
||
|
|
def command(self,*args,input=None):
|
||
|
|
self.calls.append((args,input))
|
||
|
|
if self.fail and self.fail(args):
|
||
|
|
self.fail=None
|
||
|
|
raise RuntimeError('injected failure')
|
||
|
|
if 'cat' in args or ('list' in args and 'nft' in args): return self.rules
|
||
|
|
if 'grep' in args: return 'PreUp = nft -f /etc/wireguard/wg-web.nft\nPostDown = nft delete table inet web_tunnel'
|
||
|
|
if 'show' in args: return 'active'
|
||
|
|
return ''
|
||
|
|
def run_migration(self):
|
||
|
|
with patch.object(Path,'home',return_value=self.home),patch.object(m,'command',side_effect=self.command),patch.object(m,'root_write') as write,patch.object(m.socket,'socket'):
|
||
|
|
m.main()
|
||
|
|
return write
|
||
|
|
def test_success_and_second_run_noop(self):
|
||
|
|
write=self.run_migration()
|
||
|
|
self.assertIn(':18081:80/tcp',self.drop.read_text())
|
||
|
|
self.assertEqual(json.loads((self.state/'before.json').read_text())['drop'],DROP)
|
||
|
|
self.assertFalse((self.state/'pending.json').exists())
|
||
|
|
self.assertEqual(write.call_count,1)
|
||
|
|
self.calls.clear();self.run_migration();self.assertEqual(self.calls,[])
|
||
|
|
def test_no_native_service_commands(self):
|
||
|
|
self.run_migration()
|
||
|
|
for args,_ in self.calls:
|
||
|
|
self.assertNotIn('lnd.service',args);self.assertNotIn('bitcoin-core.service',args)
|
||
|
|
def test_validation_failure_does_not_stop_or_write(self):
|
||
|
|
self.fail=lambda a:'--check' in a
|
||
|
|
with self.assertRaises(RuntimeError):self.run_migration()
|
||
|
|
self.assertEqual(self.drop.read_text(),DROP)
|
||
|
|
self.assertFalse(self.state.exists())
|
||
|
|
self.assertFalse(any('stop' in a for a,_ in self.calls))
|
||
|
|
def test_apply_failure_restores_files_and_firewall(self):
|
||
|
|
self.fail=lambda a:'nft' in a and '-f' in a and '--check' not in a
|
||
|
|
with self.assertRaises(RuntimeError):self.run_migration()
|
||
|
|
self.assertEqual(self.drop.read_text(),DROP)
|
||
|
|
self.assertFalse((self.state/'pending.json').exists())
|
||
|
|
self.assertTrue(any(v=='delete table inet web_tunnel\n'+RULES for _,v in self.calls))
|
||
|
|
def test_crash_journal_recovers_and_retries(self):
|
||
|
|
self.state.mkdir(parents=True)
|
||
|
|
(self.state/'pending.json').write_text(json.dumps({'drop':DROP,'rules':RULES,'was_active':'active'}))
|
||
|
|
self.drop.write_text(m.plan(DROP,RULES)[0])
|
||
|
|
self.run_migration()
|
||
|
|
self.assertIn(':18081:80/tcp',self.drop.read_text())
|
||
|
|
self.assertFalse((self.state/'pending.json').exists())
|
||
|
|
def test_fresh_install_executes_no_commands(self):
|
||
|
|
self.drop.unlink();self.run_migration();self.assertEqual(self.calls,[])
|
||
|
|
def test_busy_replacement_port_does_not_mutate(self):
|
||
|
|
with patch.object(Path,'home',return_value=self.home),patch.object(m,'command',side_effect=self.command),patch.object(m.socket,'socket') as socket:
|
||
|
|
socket.return_value.__enter__.return_value.bind.side_effect=OSError('in use')
|
||
|
|
with self.assertRaises(OSError):m.main()
|
||
|
|
self.assertFalse(self.state.exists())
|
||
|
|
self.assertFalse(any('stop' in a for a,_ in self.calls))
|
||
|
|
def test_failed_rollback_keeps_recovery_journal(self):
|
||
|
|
with patch.object(Path,'home',return_value=self.home),patch.object(m,'command',side_effect=self.command),patch.object(m,'root_write',side_effect=RuntimeError('write failed')),patch.object(m.socket,'socket'):
|
||
|
|
with self.assertRaises(RuntimeError):m.main()
|
||
|
|
self.assertTrue((self.state/'pending.json').exists())
|
||
|
|
self.assertEqual(self.drop.read_text(),DROP)
|
||
|
|
def test_stopped_app_is_not_started(self):
|
||
|
|
original=self.command
|
||
|
|
def stopped(*args,input=None):
|
||
|
|
return 'inactive' if 'show' in args else original(*args,input=input)
|
||
|
|
with patch.object(Path,'home',return_value=self.home),patch.object(m,'command',side_effect=stopped),patch.object(m,'root_write'),patch.object(m.socket,'socket'):
|
||
|
|
m.main()
|
||
|
|
self.assertFalse(any('restart' in a for a,_ in self.calls))
|
||
|
|
def test_live_only_firewall_changes_are_not_discarded(self):
|
||
|
|
original=self.command
|
||
|
|
def different(*args,input=None):
|
||
|
|
result=original(*args,input=input)
|
||
|
|
return result+' table inet custom {}' if 'list' in args else result
|
||
|
|
with patch.object(Path,'home',return_value=self.home),patch.object(m,'command',side_effect=different),patch.object(m.socket,'socket'):
|
||
|
|
with self.assertRaises(ValueError):m.main()
|
||
|
|
self.assertEqual(self.drop.read_text(),DROP)
|
||
|
|
|
||
|
|
if __name__=='__main__': unittest.main()
|