202 lines
7.8 KiB
Python
202 lines
7.8 KiB
Python
"""Portweiterleitungen (TCP/UDP) für Clients ohne Proxy-Unterstützung,
|
||
z. B. WireGuard, Outlook-SMTP/IMAP, alte Geräte.
|
||
|
||
Jeder Listener leitet an genau ein festes Ziel weiter. Zugriff wird über
|
||
erlaubte Quell-Netze gesteuert; die Regeln (Black-/Whitelist) gelten zusätzlich."""
|
||
import asyncio
|
||
import socket
|
||
import time
|
||
|
||
from . import config
|
||
from .core import Blocked, Conn, check_and_resolve, history, log, open_upstream, peer_ip, register, relay
|
||
from .db import db
|
||
from .policy import ip_in, parse_networks
|
||
|
||
UDP_IDLE = 180
|
||
|
||
|
||
class UdpSession(asyncio.DatagramProtocol):
|
||
def __init__(self, fwd, client_addr, listener):
|
||
self.fwd = fwd
|
||
self.client_addr = client_addr
|
||
self.listener = listener
|
||
self.transport = None
|
||
self.last = time.time()
|
||
self.start = time.time()
|
||
self.up = 0
|
||
self.down = 0
|
||
|
||
def connection_made(self, transport):
|
||
self.transport = transport
|
||
|
||
def datagram_received(self, data, addr):
|
||
self.last = time.time()
|
||
self.down += len(data)
|
||
self.listener.transport.sendto(data, self.client_addr)
|
||
|
||
def send(self, data: bytes):
|
||
self.last = time.time()
|
||
self.up += len(data)
|
||
self.transport.sendto(data)
|
||
|
||
def close(self):
|
||
if self.transport:
|
||
self.transport.close()
|
||
cip = self.client_addr[0]
|
||
history.add(ts=self.start, username=None, client_ip=cip[7:] if cip.startswith("::ffff:") else cip,
|
||
proto="fwd-udp", method="UDP", host=self.fwd["target_host"], port=self.fwd["target_port"],
|
||
url=f'{self.fwd["target_host"]}:{self.fwd["target_port"]}', status="ok",
|
||
detail=f'Weiterleitung "{self.fwd["name"]}" (:{self.fwd["listen_port"]})',
|
||
bytes_up=self.up, bytes_down=self.down, duration=round(time.time() - self.start, 3))
|
||
|
||
|
||
class UdpListener(asyncio.DatagramProtocol):
|
||
def __init__(self, fwd, nets):
|
||
self.fwd = fwd
|
||
self.nets = nets
|
||
self.transport = None
|
||
self.sessions: dict[tuple, UdpSession] = {}
|
||
self.pending: dict[tuple, list] = {}
|
||
self.denied_logged: set = set()
|
||
|
||
def connection_made(self, transport):
|
||
self.transport = transport
|
||
|
||
def datagram_received(self, data, addr):
|
||
key = (addr[0], addr[1])
|
||
s = self.sessions.get(key)
|
||
if s:
|
||
s.send(data)
|
||
return
|
||
if key in self.pending:
|
||
if len(self.pending[key]) < 32:
|
||
self.pending[key].append(data)
|
||
return
|
||
ip = addr[0][7:] if addr[0].startswith("::ffff:") else addr[0]
|
||
if self.nets and not ip_in(ip, self.nets):
|
||
if ip not in self.denied_logged:
|
||
self.denied_logged.add(ip)
|
||
history.add(client_ip=ip, proto="fwd-udp", method="UDP", host=self.fwd["target_host"],
|
||
port=self.fwd["target_port"], status="blocked",
|
||
detail=f'Quelle nicht erlaubt für "{self.fwd["name"]}"')
|
||
return
|
||
self.pending[key] = [data]
|
||
asyncio.ensure_future(self._open(key, addr))
|
||
|
||
async def _open(self, key, addr):
|
||
loop = asyncio.get_running_loop()
|
||
try:
|
||
addrs = await check_and_resolve(None, self.fwd["target_host"], self.fwd["target_port"], socket.SOCK_DGRAM)
|
||
family, sockaddr = addrs[0]
|
||
_, sess = await loop.create_datagram_endpoint(lambda: UdpSession(self.fwd, addr, self),
|
||
remote_addr=(sockaddr[0], self.fwd["target_port"]),
|
||
family=family)
|
||
except (Blocked, OSError) as e:
|
||
self.pending.pop(key, None)
|
||
reason = e.decision.reason if isinstance(e, Blocked) else str(e)
|
||
history.add(client_ip=addr[0], proto="fwd-udp", method="UDP", host=self.fwd["target_host"],
|
||
port=self.fwd["target_port"], status="blocked" if isinstance(e, Blocked) else "error",
|
||
detail=reason)
|
||
return
|
||
self.sessions[key] = sess
|
||
for d in self.pending.pop(key, []):
|
||
sess.send(d)
|
||
|
||
def cleanup(self, force=False):
|
||
now = time.time()
|
||
for key, s in list(self.sessions.items()):
|
||
if force or now - s.last > UDP_IDLE:
|
||
s.close()
|
||
del self.sessions[key]
|
||
|
||
|
||
class ForwardManager:
|
||
def __init__(self):
|
||
self.running: dict[int, dict] = {} # id -> {"sig", "server"/"transport", "proto"}
|
||
self.status: dict[int, str] = {}
|
||
|
||
@staticmethod
|
||
def _sig(f):
|
||
return (f["protocol"], f["listen_port"], f["target_host"], f["target_port"], f["allowed_sources"],
|
||
f["enabled"], f["name"])
|
||
|
||
async def reload(self):
|
||
rows = {f["id"]: f for f in db.query("SELECT * FROM forwards")}
|
||
for fid in list(self.running):
|
||
if fid not in rows or self._sig(rows[fid]) != self.running[fid]["sig"]:
|
||
await self._stop(fid)
|
||
for fid, f in rows.items():
|
||
if fid in self.running:
|
||
continue
|
||
if not f["enabled"]:
|
||
self.status[fid] = "deaktiviert"
|
||
continue
|
||
try:
|
||
await self._start(f)
|
||
self.status[fid] = "aktiv"
|
||
except OSError as e:
|
||
self.status[fid] = f"Fehler: {e.strerror or e}"
|
||
log.warning("Weiterleitung %s konnte nicht starten: %s", f["name"], e)
|
||
for fid in list(self.status):
|
||
if fid not in rows:
|
||
del self.status[fid]
|
||
|
||
async def _start(self, f):
|
||
nets = parse_networks(f["allowed_sources"])
|
||
loop = asyncio.get_running_loop()
|
||
entry = {"sig": self._sig(f)}
|
||
if f["protocol"] == "tcp":
|
||
async def handler(reader, writer, f=f, nets=nets):
|
||
await self._tcp_client(f, nets, reader, writer)
|
||
entry["server"] = await asyncio.start_server(handler, host=config.LISTEN_HOST, port=f["listen_port"])
|
||
else:
|
||
transport, proto = await loop.create_datagram_endpoint(lambda: UdpListener(f, nets),
|
||
local_addr=(config.LISTEN_HOST, f["listen_port"]))
|
||
entry["transport"], entry["proto"] = transport, proto
|
||
self.running[f["id"]] = entry
|
||
|
||
async def _stop(self, fid):
|
||
entry = self.running.pop(fid, None)
|
||
if not entry:
|
||
return
|
||
if "server" in entry:
|
||
entry["server"].close()
|
||
if "transport" in entry:
|
||
entry["proto"].cleanup(force=True)
|
||
entry["transport"].close()
|
||
|
||
async def _tcp_client(self, f, nets, reader, writer):
|
||
ip = peer_ip(writer)
|
||
conn = Conn(None, ip, "fwd-tcp", f["target_host"], f["target_port"], "FORWARD",
|
||
f'{f["target_host"]}:{f["target_port"]}')
|
||
label = f'Weiterleitung "{f["name"]}" (:{f["listen_port"]})'
|
||
try:
|
||
if nets and not ip_in(ip, nets):
|
||
conn.finish("blocked", f"Quelle nicht erlaubt – {label}")
|
||
return
|
||
try:
|
||
r_reader, r_writer = await open_upstream(None, f["target_host"], f["target_port"])
|
||
except Blocked as b:
|
||
conn.finish("blocked", b.decision.reason)
|
||
return
|
||
except OSError as e:
|
||
conn.finish("error", str(e))
|
||
return
|
||
register(conn)
|
||
try:
|
||
await relay(conn, reader, writer, r_reader, r_writer)
|
||
finally:
|
||
conn.finish("ok", label)
|
||
finally:
|
||
writer.close()
|
||
|
||
async def housekeeping(self):
|
||
while True:
|
||
await asyncio.sleep(30)
|
||
for entry in self.running.values():
|
||
if "proto" in entry:
|
||
entry["proto"].cleanup()
|
||
|
||
|
||
forwards = ForwardManager()
|