mirror of
https://github.com/PRVTPRO/Amnezia-Web-Panel.git
synced 2026-09-18 01:44:10 +02:00
hell
69893ba340
Today a client leaves to the internet from the same server it connected to.
Exit nodes split the two: clients connect to an *entry* node (reachable,
possibly behind DPI) and egress from an *exit* node (clean IP, other country).
One exit serves many entries and keeps serving its own AWG clients.
client --AWG--> [entry] amnezia-awg2 --WG/AWG link--> [exit] amnezia-exit --> internet
awg0 10.8.1.0/24 transit0 10.9.0.0/24
exit0 10.9.0.N (Table=off) one [Peer] per entry, /32
Exit side: a new protocol `exit` = container `amnezia-exit`, an AmneziaWG
listener on 55520/udp over the transit subnet 10.9.0.0/24. Peers are panel
nodes, never end users, and live in their own peersTable. ExitManager
subclasses AWGManager and closes every client-facing method.
Entry side: a second interface `exit0` inside the existing AWG container with
`Table = off`, plus policy routing - `from <client subnet> lookup 200`,
`default dev exit0` in table 200, and `blackhole default metric 1000` as a
kill-switch installed *before* the client tunnel comes up, so a container
restart cannot leak. Clients are MASQUERADEd into the entry's transit /32, so
the identical 10.8.1.0/24 of every instance can never collide at the exit.
TCP MSS is clamped, and rules 190/191 keep client<->client and AmneziaDNS
traffic local. State is file-driven: exit0.conf present means linked, and the
same shell block runs at container start and on a live (un)link.
Link crypto is WireGuard; AWG obfuscation is a per-exit-node property for hops
that cross DPI, and S3/S4 are deliberately not written on the transit - they
pad every transport packet and would eat the MTU budget of a tunnel inside a
tunnel. Private keys are generated inside the containers; the panel carries
public keys and a per-peer PSK only.
exit_link_service.py orchestrates linking with per-(entry, protocol) locks,
SSH off the event loop, a 15-second handshake gate that rolls back, and
lifecycle hooks for every way a link can go stale: exit uninstalled, stopped,
deleted, host cleared, entry reinstalled, container gone, backup restored.
Servers get a stable `uid` (already upstream) so reorder and delete cannot
break a link.
IPv4 only for now: while linked, client IPv6 is refused with `unreachable`
rather than leaking out of the entry.
Also included: an egress check that curls a public IP service bound to the
tunnel gateway (the client's exact path) and compares it with the direct one;
optional DNS resolution at the exit node; a default exit node applied to new
AWG installs; and reconciliation after a protocol restore.
244 tests pass. CURRENT_VERSION is untouched.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
83 lines
4.2 KiB
Python
83 lines
4.2 KiB
Python
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import tempfile
|
|
import unittest
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
import app as panel
|
|
|
|
|
|
def seed_servers():
|
|
return [
|
|
{'name': 'Paris', 'host': '198.51.100.1', 'ssh_port': 22, 'username': 'root', 'password': 'x',
|
|
'private_key': '', 'server_info': {}, 'uid': 'entry-uid',
|
|
'protocols': {'awg2': {'installed': True, 'port': '55424', 'awg_params': {}, 'base_protocol': 'awg2',
|
|
'instance': 1, 'display_name': 'AmneziaWG 2.0', 'container_name': 'amnezia-awg2',
|
|
'exit_link': {'exit_uid': 'exit-uid', 'exit_name': 'Berlin-1', 'transit_ip': '10.9.0.7',
|
|
'exit_public_key': 'k', 'endpoint_host': '203.0.113.5',
|
|
'endpoint_port': '55520', 'obfuscation': False, 'dns_via_exit': False,
|
|
'linked_at': 't', 'stale': None}}}},
|
|
{'name': 'Berlin-1', 'host': '203.0.113.5', 'ssh_port': 22, 'username': 'root', 'password': 'x',
|
|
'private_key': '', 'server_info': {}, 'uid': 'exit-uid',
|
|
'protocols': {'exit': {'installed': True, 'port': '55520', 'awg_params': {}, 'subnet': '10.9.0.0/24',
|
|
'public_key': 'EXITPUB', 'obfuscation': False, 'base_protocol': 'exit',
|
|
'instance': 1, 'display_name': 'Exit Node', 'container_name': 'amnezia-exit'}}},
|
|
]
|
|
|
|
|
|
class ServerPageRenderTests(unittest.TestCase):
|
|
"""Render the real server page through the app (startup creates the
|
|
default admin, the session comes from a real login) and make sure the
|
|
exit-node markup, scripts and translations come out of Jinja intact."""
|
|
|
|
def setUp(self):
|
|
self.tmp = tempfile.mkdtemp()
|
|
self._data_file = panel.DATA_FILE
|
|
panel.DATA_FILE = os.path.join(self.tmp, 'data.json')
|
|
with open(panel.DATA_FILE, 'w', encoding='utf-8') as f:
|
|
json.dump({'servers': seed_servers(), 'users': [], 'user_connections': []}, f)
|
|
# keep the background SSH monitor out of the test
|
|
self._monitor_flag = panel._conn_monitor_started
|
|
panel._conn_monitor_started = True
|
|
|
|
def tearDown(self):
|
|
panel.DATA_FILE = self._data_file
|
|
panel._conn_monitor_started = self._monitor_flag
|
|
shutil.rmtree(self.tmp, ignore_errors=True)
|
|
|
|
def render(self, client, path, lang='en'):
|
|
client.cookies.set('lang', lang)
|
|
response = client.get(path)
|
|
self.assertEqual(response.status_code, 200, path)
|
|
return response.text
|
|
|
|
def test_server_pages_render_with_exit_node_ui(self):
|
|
with TestClient(panel.app) as client:
|
|
login = client.post('/api/auth/login', json={'username': 'admin', 'password': 'admin'})
|
|
self.assertEqual(login.status_code, 200, login.text)
|
|
|
|
for sid in (0, 1):
|
|
page = self.render(client, f'/server/{sid}')
|
|
for marker in ('id="proto-exit"', 'id="exitLinkModal"', 'id="exitPeersModal"', 'id="exitOptions"',
|
|
"proto: 'exit'", 'function openExitLinkModal', 'function showExitPeers'):
|
|
self.assertTrue(marker in page, f'/server/{sid} lacks {marker}')
|
|
# a missing translation renders as the raw key
|
|
self.assertEqual(re.findall(r'>\s*exit_[a-z0-9_]+\s*<', page), [], f'/server/{sid} leaks raw keys')
|
|
self.assertTrue('Exit node: a transit endpoint' in page, f'/server/{sid} lacks the English description')
|
|
|
|
page = self.render(client, '/server/0', lang='ru')
|
|
self.assertTrue('Выходная нода' in page, 'ru page lacks the Russian card text')
|
|
i18n = json.loads(re.search(r'window\.I18N = (\{.*?\});', page).group(1))
|
|
self.assertEqual(i18n['exit_link_saved'], 'Выходная нода привязана')
|
|
self.assertTrue('dataset.serverUid = "entry-uid"' in page, 'own server uid not stamped on the page')
|
|
|
|
exits = client.get('/api/exit-nodes').json()
|
|
self.assertEqual([n['uid'] for n in exits['exit_nodes']], ['exit-uid'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|