"""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()