feat: implement MC domain filter proxy, API, dashboard
- proxy: asyncio TCP proxy with handshake parser, domain whitelist, transparent backend tunneling, SQLite logging, mtime hot reload - api: FastAPI routes for config/domains/logs/status + restart trigger - frontend: React + Vite NPM-style dashboard (dashboard/domains/logs/settings) - nginx: reverse proxy for /api -> api:8000 and / -> frontend:3000 - docker-compose: full stack with shared data volume - replace spec mc-domain-filter.md with README.md
This commit is contained in:
278
proxy/main.py
Normal file
278
proxy/main.py
Normal file
@@ -0,0 +1,278 @@
|
||||
"""MC Domain Filter Proxy.
|
||||
|
||||
asyncio 기반 TCP 프록시. 동작 순서:
|
||||
1) 클라이언트가 연결되면 첫 핸드셰이크 패킷을 읽는다.
|
||||
2) 패킷에서 server_address 를 꺼내 허용 도메인 목록과 대조한다.
|
||||
3) 허용되면 백엔드 MC 서버에 연결하고, 받은 핸드셰이크 바이트를 그대로
|
||||
forward 한 뒤 양방향으로 TCP 를 중계한다.
|
||||
4) 허용되지 않으면 즉시 연결을 종료한다 (응답을 보내지 않음).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import os
|
||||
import sqlite3
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import config as cfg_mod
|
||||
from handshake import HandshakeError, parse_handshake, read_handshake_bytes
|
||||
|
||||
LOG_DB = Path(os.environ.get("MC_LOG_DB", "/data/logs.db"))
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s %(levelname)s %(message)s",
|
||||
)
|
||||
log = logging.getLogger("proxy")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Log DB
|
||||
# ---------------------------------------------------------------------------
|
||||
def init_db() -> None:
|
||||
LOG_DB.parent.mkdir(parents=True, exist_ok=True)
|
||||
con = sqlite3.connect(LOG_DB)
|
||||
con.execute("PRAGMA journal_mode=WAL;")
|
||||
con.execute(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS connections (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
ts REAL NOT NULL,
|
||||
client_ip TEXT NOT NULL,
|
||||
domain TEXT,
|
||||
next_state INTEGER,
|
||||
action TEXT NOT NULL,
|
||||
reason TEXT
|
||||
)
|
||||
"""
|
||||
)
|
||||
con.execute("CREATE INDEX IF NOT EXISTS idx_connections_ts ON connections(ts);")
|
||||
con.commit()
|
||||
con.close()
|
||||
|
||||
|
||||
def log_event(
|
||||
client_ip: str,
|
||||
domain: str | None,
|
||||
next_state: int | None,
|
||||
action: str,
|
||||
reason: str = "",
|
||||
) -> None:
|
||||
try:
|
||||
con = sqlite3.connect(LOG_DB, timeout=2)
|
||||
con.execute(
|
||||
"INSERT INTO connections(ts, client_ip, domain, next_state, action, reason) "
|
||||
"VALUES(?,?,?,?,?,?)",
|
||||
(time.time(), client_ip, domain, next_state, action, reason),
|
||||
)
|
||||
con.commit()
|
||||
con.close()
|
||||
except Exception as exc: # noqa: BLE001 - 로그 실패는 본 흐름을 막지 않는다
|
||||
log.warning("log write failed: %s", exc)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Runtime state
|
||||
# ---------------------------------------------------------------------------
|
||||
class ProxyState:
|
||||
def __init__(self) -> None:
|
||||
self.cfg = cfg_mod.load()
|
||||
self.cfg_mtime = cfg_mod.mtime()
|
||||
self.listen_port: int = int(self.cfg["proxy"]["listen_port"])
|
||||
|
||||
def allowed(self) -> set[str]:
|
||||
return cfg_mod.allowed_domain_set(self.cfg)
|
||||
|
||||
def backend(self) -> tuple[str, int]:
|
||||
b = self.cfg["backend"]
|
||||
return b["host"], int(b["port"])
|
||||
|
||||
def enabled(self) -> bool:
|
||||
return bool(self.cfg.get("proxy", {}).get("enabled", True))
|
||||
|
||||
def reload_if_changed(self) -> bool:
|
||||
m = cfg_mod.mtime()
|
||||
if m == self.cfg_mtime:
|
||||
return False
|
||||
try:
|
||||
self.cfg = cfg_mod.load()
|
||||
self.cfg_mtime = m
|
||||
log.info(
|
||||
"config reloaded: enabled=%s backend=%s domains=%s",
|
||||
self.enabled(),
|
||||
self.backend(),
|
||||
sorted(self.allowed()),
|
||||
)
|
||||
return True
|
||||
except Exception as exc: # noqa: BLE001
|
||||
log.warning("config reload failed: %s", exc)
|
||||
return False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TCP tunneling
|
||||
# ---------------------------------------------------------------------------
|
||||
async def _pipe(
|
||||
reader: asyncio.StreamReader, writer: asyncio.StreamWriter
|
||||
) -> None:
|
||||
try:
|
||||
while True:
|
||||
data = await reader.read(8192)
|
||||
if not data:
|
||||
break
|
||||
writer.write(data)
|
||||
await writer.drain()
|
||||
except (ConnectionResetError, BrokenPipeError, asyncio.IncompleteReadError):
|
||||
pass
|
||||
except Exception as exc: # noqa: BLE001
|
||||
log.debug("pipe error: %s", exc)
|
||||
finally:
|
||||
try:
|
||||
writer.close()
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
|
||||
|
||||
async def handle_client(
|
||||
client_reader: asyncio.StreamReader,
|
||||
client_writer: asyncio.StreamWriter,
|
||||
state: ProxyState,
|
||||
) -> None:
|
||||
peer = client_writer.get_extra_info("peername") or ("?", 0)
|
||||
client_ip = peer[0]
|
||||
|
||||
if not state.enabled():
|
||||
log_event(client_ip, None, None, "blocked", "proxy disabled")
|
||||
client_writer.close()
|
||||
return
|
||||
|
||||
try:
|
||||
hs_bytes = await asyncio.wait_for(
|
||||
read_handshake_bytes(client_reader), timeout=5
|
||||
)
|
||||
hs = parse_handshake(hs_bytes)
|
||||
except (HandshakeError, asyncio.TimeoutError, asyncio.IncompleteReadError, OSError) as exc:
|
||||
log_event(client_ip, None, None, "blocked", f"handshake error: {exc}")
|
||||
log.info("BLOCK %s reason=handshake_error (%s)", client_ip, exc)
|
||||
client_writer.close()
|
||||
return
|
||||
|
||||
domain = hs.server_address.lower().strip()
|
||||
allowed = state.allowed()
|
||||
if domain not in allowed:
|
||||
log_event(client_ip, domain, hs.next_state, "blocked", "domain not allowed")
|
||||
log.info(
|
||||
"BLOCK %s domain=%r next_state=%d", client_ip, domain, hs.next_state
|
||||
)
|
||||
client_writer.close()
|
||||
return
|
||||
|
||||
backend_host, backend_port = state.backend()
|
||||
try:
|
||||
backend_reader, backend_writer = await asyncio.wait_for(
|
||||
asyncio.open_connection(backend_host, backend_port), timeout=5
|
||||
)
|
||||
except (OSError, asyncio.TimeoutError) as exc:
|
||||
log_event(
|
||||
client_ip, domain, hs.next_state, "error", f"backend connect failed: {exc}"
|
||||
)
|
||||
log.warning(
|
||||
"ERROR %s domain=%r backend=%s:%d msg=%s",
|
||||
client_ip,
|
||||
domain,
|
||||
backend_host,
|
||||
backend_port,
|
||||
exc,
|
||||
)
|
||||
client_writer.close()
|
||||
return
|
||||
|
||||
log_event(client_ip, domain, hs.next_state, "allowed")
|
||||
log.info(
|
||||
"PASS %s -> %s:%d domain=%r next_state=%d",
|
||||
client_ip,
|
||||
backend_host,
|
||||
backend_port,
|
||||
domain,
|
||||
hs.next_state,
|
||||
)
|
||||
|
||||
# 백엔드로 캡처해둔 첫 핸드셰이크 바이트를 그대로 forward
|
||||
backend_writer.write(hs_bytes)
|
||||
try:
|
||||
await backend_writer.drain()
|
||||
except Exception: # noqa: BLE001
|
||||
client_writer.close()
|
||||
return
|
||||
|
||||
await asyncio.gather(
|
||||
_pipe(client_reader, backend_writer),
|
||||
_pipe(backend_reader, client_writer),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Listener lifecycle
|
||||
# ---------------------------------------------------------------------------
|
||||
class Listener:
|
||||
def __init__(self, state: ProxyState) -> None:
|
||||
self.state = state
|
||||
self.server: asyncio.base_events.Server | None = None
|
||||
|
||||
async def start(self) -> None:
|
||||
if not self.state.enabled():
|
||||
log.info("proxy disabled by config; not listening")
|
||||
return
|
||||
self.server = await asyncio.start_server(
|
||||
lambda r, w: handle_client(r, w, self.state),
|
||||
host="0.0.0.0",
|
||||
port=self.state.listen_port,
|
||||
)
|
||||
log.info("listening on 0.0.0.0:%d", self.state.listen_port)
|
||||
|
||||
async def stop(self) -> None:
|
||||
if self.server is not None:
|
||||
self.server.close()
|
||||
await self.server.wait_closed()
|
||||
self.server = None
|
||||
log.info("listener stopped")
|
||||
|
||||
async def restart(self) -> None:
|
||||
await self.stop()
|
||||
await self.start()
|
||||
|
||||
|
||||
async def config_watcher(state: ProxyState, listener: Listener) -> None:
|
||||
while True:
|
||||
await asyncio.sleep(2)
|
||||
old_port = state.listen_port
|
||||
old_enabled = state.enabled()
|
||||
if not state.reload_if_changed():
|
||||
continue
|
||||
new_port = int(state.cfg["proxy"]["listen_port"])
|
||||
new_enabled = state.enabled()
|
||||
if new_port != old_port or new_enabled != old_enabled:
|
||||
state.listen_port = new_port
|
||||
await listener.restart()
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
init_db()
|
||||
state = ProxyState()
|
||||
listener = Listener(state)
|
||||
await listener.start()
|
||||
watcher = asyncio.create_task(config_watcher(state, listener))
|
||||
try:
|
||||
await asyncio.Event().wait()
|
||||
finally:
|
||||
watcher.cancel()
|
||||
await listener.stop()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
asyncio.run(main())
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
Reference in New Issue
Block a user