763 lines
35 KiB
Python
763 lines
35 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import contextlib
|
|
import dataclasses
|
|
import json
|
|
import re
|
|
import socket
|
|
import sqlite3
|
|
import struct
|
|
import time
|
|
import traceback
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
import uuid
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
APP_DIR = Path(__file__).resolve().parent
|
|
DEFAULT_CONFIG = APP_DIR / "config.json"
|
|
DEFAULT_DB = APP_DIR / "data" / "monitor.db"
|
|
|
|
|
|
def now_ts() -> int:
|
|
return int(time.time())
|
|
|
|
|
|
def iso(ts: int | None = None) -> str:
|
|
return datetime.fromtimestamp(ts or now_ts()).strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
|
|
def load_json(path: Path) -> dict[str, Any]:
|
|
if not path.exists():
|
|
return {}
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def save_json(path: Path, data: dict[str, Any]) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = path.with_suffix(path.suffix + ".tmp")
|
|
tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
tmp.replace(path)
|
|
|
|
|
|
def deep_copy(obj: Any) -> Any:
|
|
return json.loads(json.dumps(obj, ensure_ascii=False))
|
|
|
|
|
|
def default_config() -> dict[str, Any]:
|
|
return {
|
|
"interval_seconds": 300,
|
|
"timeout_seconds": 8,
|
|
"target_host": "www.baidu.com",
|
|
"target_port": 443,
|
|
"fail_threshold": 3,
|
|
"recover_threshold": 1,
|
|
"db_path": str(DEFAULT_DB),
|
|
"proxies": [],
|
|
"notifications": {"enabled": True, "timeout_seconds": 10, "channels": []},
|
|
}
|
|
|
|
|
|
def merge_defaults(cfg: dict[str, Any]) -> dict[str, Any]:
|
|
base = default_config()
|
|
base.update(cfg or {})
|
|
notifications = base.get("notifications") or {}
|
|
base["notifications"] = {"enabled": True, "timeout_seconds": 10, "channels": [], **notifications}
|
|
if base.get("target_host") in (None, "", "1.1.1.1", "8.8.8.8", "google.com"):
|
|
base["target_host"] = "www.baidu.com"
|
|
base["target_port"] = int(base.get("target_port") or 443)
|
|
base["interval_seconds"] = int(base.get("interval_seconds") or 300)
|
|
base.setdefault("proxies", [])
|
|
for proxy in base["proxies"]:
|
|
if proxy.get("target_host") in (None, "", "1.1.1.1", "8.8.8.8", "google.com"):
|
|
proxy["target_host"] = base["target_host"]
|
|
proxy["target_port"] = int(proxy.get("target_port") or base["target_port"])
|
|
proxy["interval_seconds"] = int(proxy.get("interval_seconds") or base["interval_seconds"])
|
|
base["notifications"].setdefault("channels", [])
|
|
base["notifications"]["channels"] = [normalize_channel(c) for c in base["notifications"].get("channels", [])]
|
|
return base
|
|
|
|
|
|
def public_proxy(proxy: dict[str, Any]) -> dict[str, Any]:
|
|
p = deep_copy(proxy)
|
|
if p.get("password"):
|
|
p["password_masked"] = "********"
|
|
p["password"] = ""
|
|
return p
|
|
|
|
|
|
def normalize_channel(channel: dict[str, Any]) -> dict[str, Any]:
|
|
"""Return canonical notification channel shape: common fields + type-specific config.
|
|
|
|
Backward compatible with the old flat schema, but output never mixes fields
|
|
across Bark / PushPlus / Autman / Webhook.
|
|
"""
|
|
c = dict(channel or {})
|
|
typ = c.get("type") or "webhook"
|
|
cfg = dict(c.get("config") or {})
|
|
if typ == "bark":
|
|
cfg = {
|
|
"server": cfg.get("server") or c.get("server") or "https://api.day.app",
|
|
"key": cfg.get("key") or c.get("key") or "",
|
|
"group": cfg.get("group") or c.get("group") or "SOCKS5监控",
|
|
}
|
|
elif typ == "pushplus":
|
|
cfg = {
|
|
"token": cfg.get("token") or c.get("token") or c.get("key") or "",
|
|
"template": cfg.get("template") or c.get("template") or "txt",
|
|
"topic": cfg.get("topic") or c.get("topic") or c.get("group") or "",
|
|
}
|
|
elif typ == "autman":
|
|
cfg = {
|
|
"url": cfg.get("url") or c.get("url") or c.get("webhook") or "",
|
|
"access_token": cfg.get("access_token") or cfg.get("token") or c.get("access_token") or c.get("token") or "",
|
|
}
|
|
elif typ == "webhook":
|
|
cfg = {
|
|
"url": cfg.get("url") or c.get("url") or "",
|
|
"headers": cfg.get("headers") if isinstance(cfg.get("headers"), dict) else (c.get("headers") if isinstance(c.get("headers"), dict) else {}),
|
|
"payload": cfg.get("payload") if isinstance(cfg.get("payload"), dict) else (c.get("payload") if isinstance(c.get("payload"), dict) else {}),
|
|
}
|
|
return {
|
|
"id": c.get("id") or f"ch-{uuid.uuid4().hex[:8]}",
|
|
"type": typ,
|
|
"name": c.get("name") or typ,
|
|
"enabled": bool(c.get("enabled", True)),
|
|
"config": cfg,
|
|
}
|
|
|
|
|
|
class Store:
|
|
def __init__(self, path: Path):
|
|
self.path = path
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
self.conn = sqlite3.connect(self.path, check_same_thread=False)
|
|
self.conn.row_factory = sqlite3.Row
|
|
self.lock = asyncio.Lock()
|
|
self.init()
|
|
|
|
def init(self) -> None:
|
|
self.conn.executescript(
|
|
"""
|
|
create table if not exists checks(
|
|
id integer primary key autoincrement,
|
|
proxy_id text not null,
|
|
ts integer not null,
|
|
ok integer not null,
|
|
latency_ms integer,
|
|
error text,
|
|
exit_ip text
|
|
);
|
|
create index if not exists idx_checks_proxy_ts on checks(proxy_id, ts desc);
|
|
create table if not exists events(
|
|
id integer primary key autoincrement,
|
|
proxy_id text not null,
|
|
ts integer not null,
|
|
old_status text,
|
|
new_status text,
|
|
message text
|
|
);
|
|
create index if not exists idx_events_ts on events(ts desc);
|
|
create table if not exists notification_logs(
|
|
id integer primary key autoincrement,
|
|
ts integer not null,
|
|
channel_id text,
|
|
channel_name text,
|
|
ok integer not null,
|
|
title text,
|
|
message text,
|
|
result text
|
|
);
|
|
create index if not exists idx_notification_logs_ts on notification_logs(ts desc);
|
|
"""
|
|
)
|
|
self.conn.commit()
|
|
|
|
async def add_check(self, proxy_id: str, ok: bool, latency_ms: int | None, error: str | None, exit_ip: str | None) -> None:
|
|
async with self.lock:
|
|
self.conn.execute(
|
|
"insert into checks(proxy_id,ts,ok,latency_ms,error,exit_ip) values(?,?,?,?,?,?)",
|
|
(proxy_id, now_ts(), 1 if ok else 0, latency_ms, error, exit_ip),
|
|
)
|
|
self.conn.commit()
|
|
|
|
async def add_event(self, proxy_id: str, old_status: str | None, new_status: str, message: str) -> None:
|
|
async with self.lock:
|
|
self.conn.execute(
|
|
"insert into events(proxy_id,ts,old_status,new_status,message) values(?,?,?,?,?)",
|
|
(proxy_id, now_ts(), old_status, new_status, message),
|
|
)
|
|
self.conn.commit()
|
|
|
|
async def add_notification(self, channel: dict[str, Any], ok: bool, title: str, message: str, result: str) -> None:
|
|
async with self.lock:
|
|
self.conn.execute(
|
|
"insert into notification_logs(ts,channel_id,channel_name,ok,title,message,result) values(?,?,?,?,?,?,?)",
|
|
(now_ts(), channel.get("id"), channel.get("name") or channel.get("type"), 1 if ok else 0, title, message, result),
|
|
)
|
|
self.conn.commit()
|
|
|
|
def recent_checks(self, proxy_id: str, limit: int = 50) -> list[dict[str, Any]]:
|
|
rows = self.conn.execute("select * from checks where proxy_id=? order by ts desc limit ?", (proxy_id, limit)).fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
def recent_events(self, limit: int = 50) -> list[dict[str, Any]]:
|
|
rows = self.conn.execute("select * from events order by ts desc limit ?", (limit,)).fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
def recent_notifications(self, limit: int = 50) -> list[dict[str, Any]]:
|
|
rows = self.conn.execute("select * from notification_logs order by ts desc limit ?", (limit,)).fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class RuntimeState:
|
|
status: str = "unknown"
|
|
last_ok: bool | None = None
|
|
last_ts: int | None = None
|
|
latency_ms: int | None = None
|
|
error: str | None = None
|
|
exit_ip: str | None = None
|
|
fail_count: int = 0
|
|
ok_count: int = 0
|
|
notified_down: bool = False
|
|
|
|
|
|
class Notifier:
|
|
def __init__(self, cfg: dict[str, Any], store: Store | None = None):
|
|
self.cfg = cfg or {}
|
|
self.store = store
|
|
self.timeout = int(self.cfg.get("timeout_seconds", 10))
|
|
|
|
async def send_all(self, title: str, body: str) -> list[dict[str, Any]]:
|
|
if not self.cfg.get("enabled", True):
|
|
return [{"channel": "all", "ok": False, "error": "notifications disabled"}]
|
|
channels = [c for c in self.cfg.get("channels", []) if c.get("enabled", True)]
|
|
return await self.send_channels(channels, title, body)
|
|
|
|
async def send_channel_id(self, channel_id: str, title: str, body: str) -> dict[str, Any]:
|
|
ch = next((c for c in self.cfg.get("channels", []) if c.get("id") == channel_id), None)
|
|
if not ch:
|
|
return {"channel": channel_id, "ok": False, "error": "channel not found"}
|
|
# 单通道测试应可验证配置本身，即使该渠道平时处于停用状态。
|
|
test_ch = dict(ch)
|
|
test_ch["enabled"] = True
|
|
res = await self.send_channels([test_ch], title, body)
|
|
return res[0] if res else {"channel": channel_id, "ok": False, "error": "not sent"}
|
|
|
|
async def send_channels(self, channels: list[dict[str, Any]], title: str, body: str) -> list[dict[str, Any]]:
|
|
results = []
|
|
for ch in channels:
|
|
res = await asyncio.to_thread(self._send_one, ch, title, body)
|
|
results.append(res)
|
|
if self.store:
|
|
await self.store.add_notification(ch, bool(res.get("ok")), title, body, json.dumps(res, ensure_ascii=False))
|
|
return results
|
|
|
|
def _http_get(self, url: str) -> tuple[bool, str]:
|
|
try:
|
|
with urllib.request.urlopen(url, timeout=self.timeout) as r:
|
|
text = r.read(4096).decode("utf-8", "replace")
|
|
return 200 <= r.status < 300, f"HTTP {r.status} {text[:300]}"
|
|
except Exception as e:
|
|
return False, repr(e)
|
|
|
|
def _http_post_json(self, url: str, payload: dict[str, Any], headers: dict[str, str] | None = None) -> tuple[bool, str]:
|
|
data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
|
req = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json", **(headers or {})}, method="POST")
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=self.timeout) as r:
|
|
text = r.read(4096).decode("utf-8", "replace")
|
|
return 200 <= r.status < 300, f"HTTP {r.status} {text[:300]}"
|
|
except Exception as e:
|
|
return False, repr(e)
|
|
|
|
def _send_one(self, ch: dict[str, Any], title: str, body: str) -> dict[str, Any]:
|
|
ch = normalize_channel(ch)
|
|
typ = ch.get("type")
|
|
cfg = ch.get("config") or {}
|
|
name = ch.get("name") or typ
|
|
if not ch.get("enabled", True):
|
|
return {"channel": name, "ok": False, "error": "disabled"}
|
|
try:
|
|
if typ == "bark":
|
|
base = (cfg.get("server") or "https://api.day.app").rstrip("/")
|
|
key = cfg.get("key") or ""
|
|
group = cfg.get("group") or "SOCKS5监控"
|
|
if not key:
|
|
return {"channel": name, "ok": False, "error": "missing Bark key"}
|
|
url = f"{base}/{urllib.parse.quote(key)}/{urllib.parse.quote(title)}/{urllib.parse.quote(body)}?group={urllib.parse.quote(group)}"
|
|
ok, msg = self._http_get(url)
|
|
elif typ == "pushplus":
|
|
token = cfg.get("token") or ""
|
|
if not token:
|
|
return {"channel": name, "ok": False, "error": "missing PushPlus token"}
|
|
payload = {"token": token, "title": title, "content": body, "template": cfg.get("template", "txt")}
|
|
if cfg.get("topic"):
|
|
payload["topic"] = cfg.get("topic")
|
|
ok, msg = self._http_post_json("https://www.pushplus.plus/send", payload)
|
|
elif typ == "autman":
|
|
url = cfg.get("url") or ""
|
|
if not url:
|
|
return {"channel": name, "ok": False, "error": "missing autman url"}
|
|
message = f"标题:{title}\n内容:{body}"
|
|
payload = {"message": message, "access_token": cfg.get("access_token") or ""}
|
|
ok, msg = self._http_post_json(url, payload, {})
|
|
elif typ == "webhook":
|
|
url = cfg.get("url") or ""
|
|
if not url:
|
|
return {"channel": name, "ok": False, "error": "missing webhook url"}
|
|
payload = dict(cfg.get("payload", {}) if isinstance(cfg.get("payload"), dict) else {})
|
|
payload.update({"title": title, "content": body, "ts": now_ts()})
|
|
ok, msg = self._http_post_json(url, payload, cfg.get("headers") or {})
|
|
else:
|
|
return {"channel": name, "ok": False, "error": f"unknown type {typ}"}
|
|
return {"channel": name, "ok": ok, "message": msg}
|
|
except Exception as e:
|
|
return {"channel": name, "ok": False, "error": repr(e)}
|
|
|
|
|
|
async def socks5_open_connection(proxy: dict[str, Any], timeout: float, target_host: str | None = None, target_port: int | None = None) -> tuple[asyncio.StreamReader, asyncio.StreamWriter]:
|
|
host = proxy["host"]
|
|
port = int(proxy["port"])
|
|
username = proxy.get("username") or ""
|
|
password = proxy.get("password") or ""
|
|
target_host = str(target_host or proxy.get("target_host") or "www.baidu.com")
|
|
target_port = int(target_port or proxy.get("target_port", 443))
|
|
reader, writer = await asyncio.wait_for(asyncio.open_connection(host, port), timeout=timeout)
|
|
try:
|
|
methods = [0x00]
|
|
if username or password:
|
|
methods.append(0x02)
|
|
writer.write(bytes([0x05, len(methods), *methods])); await writer.drain()
|
|
resp = await asyncio.wait_for(reader.readexactly(2), timeout=timeout)
|
|
if resp[0] != 0x05: raise RuntimeError(f"invalid socks version {resp[0]}")
|
|
method = resp[1]
|
|
if method == 0xFF: raise RuntimeError("socks auth method rejected")
|
|
if method == 0x02:
|
|
ub, pb = username.encode(), password.encode()
|
|
writer.write(bytes([0x01, len(ub)]) + ub + bytes([len(pb)]) + pb); await writer.drain()
|
|
auth = await asyncio.wait_for(reader.readexactly(2), timeout=timeout)
|
|
if auth[1] != 0x00: raise RuntimeError("socks username/password auth failed")
|
|
elif method != 0x00:
|
|
raise RuntimeError(f"unsupported socks method {method}")
|
|
try:
|
|
addr = socket.inet_aton(target_host); atyp = 0x01; body = addr
|
|
except OSError:
|
|
hb = target_host.encode(); atyp = 0x03; body = bytes([len(hb)]) + hb
|
|
writer.write(bytes([0x05, 0x01, 0x00, atyp]) + body + struct.pack("!H", target_port)); await writer.drain()
|
|
head = await asyncio.wait_for(reader.readexactly(4), timeout=timeout)
|
|
if head[1] != 0x00: raise RuntimeError(f"socks connect failed rep=0x{head[1]:02x}")
|
|
atyp = head[3]
|
|
if atyp == 0x01: await reader.readexactly(4)
|
|
elif atyp == 0x03: await reader.readexactly((await reader.readexactly(1))[0])
|
|
elif atyp == 0x04: await reader.readexactly(16)
|
|
await reader.readexactly(2)
|
|
return reader, writer
|
|
except Exception:
|
|
writer.close()
|
|
with contextlib.suppress(Exception): await writer.wait_closed()
|
|
raise
|
|
|
|
|
|
async def socks5_connect(proxy: dict[str, Any], timeout: float) -> None:
|
|
reader, writer = await socks5_open_connection(proxy, timeout)
|
|
writer.close()
|
|
with contextlib.suppress(Exception): await writer.wait_closed()
|
|
|
|
|
|
async def http_get_via_socks(proxy: dict[str, Any], timeout: float, host: str, path: str = "/") -> str:
|
|
reader, writer = await socks5_open_connection(proxy, timeout, host, 80)
|
|
try:
|
|
req = (
|
|
f"GET {path} HTTP/1.1\r\n"
|
|
f"Host: {host}\r\n"
|
|
"User-Agent: socks5-monitor/1.0\r\n"
|
|
"Accept: text/plain, application/json;q=0.9, */*;q=0.1\r\n"
|
|
"Connection: close\r\n\r\n"
|
|
).encode("ascii")
|
|
writer.write(req); await writer.drain()
|
|
chunks = []
|
|
deadline = asyncio.get_running_loop().time() + timeout
|
|
while True:
|
|
remaining = max(0.1, deadline - asyncio.get_running_loop().time())
|
|
if remaining <= 0.1 and chunks:
|
|
break
|
|
try:
|
|
chunk = await asyncio.wait_for(reader.read(4096), timeout=remaining)
|
|
except asyncio.TimeoutError:
|
|
break
|
|
if not chunk:
|
|
break
|
|
chunks.append(chunk)
|
|
if sum(len(x) for x in chunks) >= 65536:
|
|
break
|
|
raw = b"".join(chunks)
|
|
return raw.decode("utf-8", "replace")
|
|
finally:
|
|
writer.close()
|
|
with contextlib.suppress(Exception): await writer.wait_closed()
|
|
|
|
|
|
def extract_ip(text: str) -> str | None:
|
|
body = text.split("\r\n\r\n", 1)[-1].strip()
|
|
m = re.search(r"\b(?:(?:25[0-5]|2[0-4]\d|1?\d?\d)\.){3}(?:25[0-5]|2[0-4]\d|1?\d?\d)\b", body)
|
|
if m:
|
|
return m.group(0)
|
|
# Very small IPv6 matcher, enough for plain text IP services.
|
|
m = re.search(r"\b(?:[0-9a-fA-F]{1,4}:){2,}[0-9a-fA-F]{0,4}\b", body)
|
|
return m.group(0) if m else None
|
|
|
|
|
|
async def get_exit_ip_via_socks(proxy: dict[str, Any], timeout: float) -> str | None:
|
|
"""Fetch the public egress IP through the SOCKS5 proxy without failing health checks."""
|
|
endpoints = [
|
|
("api.ipify.org", "/?format=text"),
|
|
("ipv4.icanhazip.com", "/"),
|
|
("ifconfig.me", "/ip"),
|
|
("ident.me", "/"),
|
|
("checkip.amazonaws.com", "/"),
|
|
]
|
|
errors = []
|
|
per_endpoint_timeout = max(2.0, min(float(timeout), 6.0))
|
|
for host, path in endpoints:
|
|
try:
|
|
text = await http_get_via_socks(proxy, per_endpoint_timeout, host, path)
|
|
ip = extract_ip(text)
|
|
if ip:
|
|
return ip
|
|
errors.append(f"{host}: no ip in response")
|
|
except Exception as e:
|
|
errors.append(f"{host}: {e}")
|
|
proxy["last_exit_ip_error"] = "; ".join(errors[-3:]) or "exit ip lookup failed: empty response from all providers"
|
|
return None
|
|
|
|
|
|
async def check_proxy(proxy: dict[str, Any]) -> tuple[bool, int | None, str | None, str | None]:
|
|
timeout = float(proxy.get("timeout_seconds", 8))
|
|
start = time.perf_counter()
|
|
try:
|
|
await socks5_connect(proxy, timeout)
|
|
latency = int((time.perf_counter() - start) * 1000)
|
|
exit_ip = await get_exit_ip_via_socks(proxy, timeout)
|
|
err = None if exit_ip else (proxy.get("last_exit_ip_error") or "exit ip lookup failed")
|
|
return True, latency, err, exit_ip
|
|
except Exception as e:
|
|
return False, int((time.perf_counter() - start) * 1000), str(e), None
|
|
|
|
|
|
class App:
|
|
def __init__(self, config_path: Path):
|
|
self.config_path = config_path
|
|
self.config = merge_defaults(load_json(config_path))
|
|
if not config_path.exists(): save_json(config_path, self.config)
|
|
self.store = Store(Path(self.config.get("db_path") or DEFAULT_DB))
|
|
self.notifier = Notifier(self.config.get("notifications") or {}, self.store)
|
|
self.states: dict[str, RuntimeState] = {}
|
|
self.monitor_tasks: dict[str, asyncio.Task] = {}
|
|
self.config_lock = asyncio.Lock()
|
|
|
|
@property
|
|
def proxies(self) -> list[dict[str, Any]]:
|
|
return self.config.setdefault("proxies", [])
|
|
|
|
@property
|
|
def channels(self) -> list[dict[str, Any]]:
|
|
return self.config.setdefault("notifications", {}).setdefault("channels", [])
|
|
|
|
def state_for(self, proxy_id: str) -> RuntimeState:
|
|
return self.states.setdefault(proxy_id, RuntimeState())
|
|
|
|
async def save_config(self) -> None:
|
|
async with self.config_lock:
|
|
save_json(self.config_path, self.config)
|
|
self.notifier = Notifier(self.config.get("notifications") or {}, self.store)
|
|
|
|
def get_proxy(self, proxy_id: str) -> dict[str, Any] | None:
|
|
return next((p for p in self.proxies if p.get("id") == proxy_id), None)
|
|
|
|
def get_channel(self, channel_id: str) -> dict[str, Any] | None:
|
|
return next((c for c in self.channels if c.get("id") == channel_id), None)
|
|
|
|
def validate_proxy(self, data: dict[str, Any], existing: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
p = dict(existing or {})
|
|
p.update(data)
|
|
p.setdefault("id", re.sub(r"[^a-zA-Z0-9_-]+", "-", (p.get("name") or "proxy").strip()).strip("-") or f"proxy-{uuid.uuid4().hex[:8]}")
|
|
p.setdefault("name", p["id"])
|
|
if not p.get("host"): raise ValueError("host required")
|
|
p["port"] = int(p.get("port") or 1080)
|
|
p["target_host"] = p.get("target_host") or self.config.get("target_host") or "www.baidu.com"
|
|
p["target_port"] = int(p.get("target_port") or self.config.get("target_port") or 443)
|
|
p["timeout_seconds"] = int(p.get("timeout_seconds") or self.config.get("timeout_seconds") or 8)
|
|
p["interval_seconds"] = max(5, int(p.get("interval_seconds") or self.config.get("interval_seconds") or 300))
|
|
p["fail_threshold"] = int(p.get("fail_threshold") or self.config.get("fail_threshold") or 3)
|
|
p["recover_threshold"] = int(p.get("recover_threshold") or self.config.get("recover_threshold") or 1)
|
|
p["enabled"] = bool(p.get("enabled", True))
|
|
return p
|
|
|
|
def validate_channel(self, data: dict[str, Any], existing: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
incoming = normalize_channel(data or {})
|
|
if existing:
|
|
incoming["id"] = existing.get("id")
|
|
typ = incoming.get("type")
|
|
if typ not in {"bark", "pushplus", "autman", "webhook"}:
|
|
raise ValueError("type must be bark/pushplus/autman/webhook")
|
|
incoming.setdefault("name", typ)
|
|
incoming["enabled"] = bool(incoming.get("enabled", True))
|
|
cfg = incoming["config"]
|
|
if typ == "bark" and not cfg.get("server"):
|
|
cfg["server"] = "https://api.day.app"
|
|
if typ == "pushplus" and not cfg.get("template"):
|
|
cfg["template"] = "txt"
|
|
return incoming
|
|
|
|
def validate_settings(self, data: dict[str, Any]) -> dict[str, Any]:
|
|
out = {}
|
|
if "interval_seconds" in data:
|
|
out["interval_seconds"] = max(5, int(data.get("interval_seconds") or 300))
|
|
if "timeout_seconds" in data:
|
|
out["timeout_seconds"] = max(1, int(data.get("timeout_seconds") or 8))
|
|
if "target_host" in data:
|
|
host = str(data.get("target_host") or "").strip()
|
|
if not host:
|
|
raise ValueError("target_host required")
|
|
out["target_host"] = host
|
|
if "target_port" in data:
|
|
port = int(data.get("target_port") or 443)
|
|
if port < 1 or port > 65535:
|
|
raise ValueError("target_port must be 1..65535")
|
|
out["target_port"] = port
|
|
if "fail_threshold" in data:
|
|
out["fail_threshold"] = max(1, int(data.get("fail_threshold") or 3))
|
|
if "recover_threshold" in data:
|
|
out["recover_threshold"] = max(1, int(data.get("recover_threshold") or 1))
|
|
return out
|
|
|
|
async def update_settings(self, data: dict[str, Any]) -> dict[str, Any]:
|
|
settings = self.validate_settings(data)
|
|
self.config.update(settings)
|
|
# Apply changed global defaults to nodes that were still using old built-in defaults.
|
|
for p in self.proxies:
|
|
if settings.get("target_host") and p.get("target_host") in (None, "", "1.1.1.1", "8.8.8.8", "google.com"):
|
|
p["target_host"] = settings["target_host"]
|
|
if settings.get("target_port") and not p.get("target_port"):
|
|
p["target_port"] = settings["target_port"]
|
|
if settings.get("interval_seconds") and not p.get("interval_seconds"):
|
|
p["interval_seconds"] = settings["interval_seconds"]
|
|
await self.save_config()
|
|
for p in list(self.proxies):
|
|
await self.restart_proxy_task(p["id"])
|
|
return {k: self.config.get(k) for k in ["interval_seconds", "timeout_seconds", "target_host", "target_port", "fail_threshold", "recover_threshold"]}
|
|
|
|
async def upsert_proxy(self, data: dict[str, Any], proxy_id: str | None = None) -> dict[str, Any]:
|
|
existing = self.get_proxy(proxy_id) if proxy_id else None
|
|
p = self.validate_proxy(data, existing)
|
|
if proxy_id and existing:
|
|
idx = self.proxies.index(existing); self.proxies[idx] = p
|
|
else:
|
|
if self.get_proxy(p["id"]): raise ValueError("proxy id already exists")
|
|
self.proxies.append(p)
|
|
await self.save_config(); await self.restart_proxy_task(p["id"])
|
|
return public_proxy(p)
|
|
|
|
async def delete_proxy(self, proxy_id: str) -> bool:
|
|
p = self.get_proxy(proxy_id)
|
|
if not p: return False
|
|
self.proxies.remove(p)
|
|
task = self.monitor_tasks.pop(proxy_id, None)
|
|
if task: task.cancel()
|
|
self.states.pop(proxy_id, None)
|
|
await self.save_config()
|
|
return True
|
|
|
|
async def upsert_channel(self, data: dict[str, Any], channel_id: str | None = None) -> dict[str, Any]:
|
|
existing = self.get_channel(channel_id) if channel_id else None
|
|
c = self.validate_channel(data, existing)
|
|
if channel_id and existing:
|
|
idx = self.channels.index(existing); self.channels[idx] = c
|
|
else:
|
|
if self.get_channel(c["id"]): raise ValueError("channel id already exists")
|
|
self.channels.append(c)
|
|
await self.save_config()
|
|
return c
|
|
|
|
async def delete_channel(self, channel_id: str) -> bool:
|
|
c = self.get_channel(channel_id)
|
|
if not c: return False
|
|
self.channels.remove(c)
|
|
await self.save_config()
|
|
return True
|
|
|
|
async def run_once(self, proxy: dict[str, Any]) -> None:
|
|
if not proxy.get("enabled", True): return
|
|
proxy_id = proxy["id"]
|
|
st = self.state_for(proxy_id)
|
|
ok, latency, err, exit_ip = await check_proxy(proxy)
|
|
await self.store.add_check(proxy_id, ok, latency, err, exit_ip)
|
|
st.last_ts = now_ts(); st.last_ok = ok; st.latency_ms = latency; st.error = err; st.exit_ip = exit_ip
|
|
fail_threshold = int(proxy.get("fail_threshold", self.config.get("fail_threshold", 3)))
|
|
recover_threshold = int(proxy.get("recover_threshold", self.config.get("recover_threshold", 1)))
|
|
old = st.status
|
|
if ok:
|
|
st.ok_count += 1; st.fail_count = 0
|
|
if st.status == "down" and st.ok_count >= recover_threshold:
|
|
st.status = "up"; st.notified_down = False
|
|
msg = f"{proxy.get('name', proxy_id)} 已恢复。延迟 {latency}ms，时间 {iso()}"
|
|
await self.store.add_event(proxy_id, old, st.status, msg)
|
|
await self.notifier.send_all(f"SOCKS5恢复：{proxy.get('name', proxy_id)}", msg)
|
|
elif st.status == "unknown":
|
|
st.status = "up"
|
|
await self.store.add_event(proxy_id, old, st.status, f"{proxy.get('name', proxy_id)} 首次检测正常")
|
|
else:
|
|
st.fail_count += 1; st.ok_count = 0
|
|
if st.fail_count >= fail_threshold:
|
|
st.status = "down"
|
|
if not st.notified_down:
|
|
msg = f"{proxy.get('name', proxy_id)} 不通，连续失败 {st.fail_count} 次。错误：{err}。时间 {iso()}"
|
|
await self.store.add_event(proxy_id, old, st.status, msg)
|
|
await self.notifier.send_all(f"SOCKS5故障：{proxy.get('name', proxy_id)}", msg)
|
|
st.notified_down = True
|
|
|
|
async def monitor_loop(self, proxy_id: str) -> None:
|
|
while True:
|
|
proxy = self.get_proxy(proxy_id)
|
|
if not proxy: return
|
|
if proxy.get("enabled", True):
|
|
await self.run_once(proxy)
|
|
await asyncio.sleep(int(proxy.get("interval_seconds", self.config.get("interval_seconds", 60))))
|
|
|
|
async def restart_proxy_task(self, proxy_id: str) -> None:
|
|
old = self.monitor_tasks.pop(proxy_id, None)
|
|
if old: old.cancel()
|
|
if self.get_proxy(proxy_id):
|
|
self.monitor_tasks[proxy_id] = asyncio.create_task(self.monitor_loop(proxy_id))
|
|
|
|
async def start_tasks(self) -> None:
|
|
for p in self.proxies:
|
|
await self.restart_proxy_task(p["id"])
|
|
|
|
def status_payload(self) -> dict[str, Any]:
|
|
items = []
|
|
for p in self.proxies:
|
|
sid = p["id"]; st = self.state_for(sid)
|
|
items.append({"proxy": public_proxy(p), "state": dataclasses.asdict(st), "recent": self.store.recent_checks(sid, 20)})
|
|
return {
|
|
"ok": True, "now": now_ts(), "now_text": iso(),
|
|
"settings": {k: self.config.get(k) for k in ["interval_seconds", "timeout_seconds", "target_host", "target_port", "fail_threshold", "recover_threshold"]},
|
|
"proxies": items,
|
|
"notifications": self.config.get("notifications", {}),
|
|
"events": self.store.recent_events(50),
|
|
"notification_logs": self.store.recent_notifications(30),
|
|
}
|
|
|
|
|
|
def get_index_html() -> str:
|
|
static_index = APP_DIR / "static" / "index.html"
|
|
if static_index.exists():
|
|
return static_index.read_text(encoding="utf-8")
|
|
return "<h1>SOCKS5 Sentinel</h1><p>static/index.html missing</p>"
|
|
|
|
|
|
|
|
def json_response(obj: Any, status: int = 200) -> tuple[int, str, bytes]:
|
|
return status, "application/json; charset=utf-8", json.dumps(obj, ensure_ascii=False).encode("utf-8")
|
|
|
|
|
|
async def read_request(reader: asyncio.StreamReader) -> tuple[str, str, dict[str, str], bytes]:
|
|
head = await reader.readuntil(b"\r\n\r\n")
|
|
text = head.decode("iso-8859-1")
|
|
lines = text.split("\r\n")
|
|
method, path, _ = lines[0].split(" ", 2)
|
|
headers = {}
|
|
for line in lines[1:]:
|
|
if ":" in line:
|
|
k, v = line.split(":", 1); headers[k.lower()] = v.strip()
|
|
length = int(headers.get("content-length") or "0")
|
|
body = await reader.readexactly(length) if length else b""
|
|
return method, path, headers, body
|
|
|
|
|
|
async def handle_http(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, app: App) -> None:
|
|
try:
|
|
method, path, headers, body = await read_request(reader)
|
|
parsed = urllib.parse.urlparse(path); p = parsed.path
|
|
status, ctype, resp = 200, "text/plain; charset=utf-8", b""
|
|
data = json.loads(body.decode("utf-8")) if body else {}
|
|
if method == "GET" and p == "/":
|
|
ctype = "text/html; charset=utf-8"; resp = get_index_html().encode("utf-8")
|
|
elif method == "GET" and p == "/api/status":
|
|
status, ctype, resp = json_response(app.status_payload())
|
|
elif method == "PUT" and p == "/api/settings":
|
|
status, ctype, resp = json_response({"ok": True, "settings": await app.update_settings(data)})
|
|
elif method == "POST" and p == "/api/proxies":
|
|
status, ctype, resp = json_response({"ok": True, "proxy": await app.upsert_proxy(data)}, 201)
|
|
elif method == "PUT" and p.startswith("/api/proxies/"):
|
|
pid = urllib.parse.unquote(p.rsplit("/", 1)[-1])
|
|
status, ctype, resp = json_response({"ok": True, "proxy": await app.upsert_proxy(data, pid)})
|
|
elif method == "DELETE" and p.startswith("/api/proxies/"):
|
|
pid = urllib.parse.unquote(p.rsplit("/", 1)[-1])
|
|
status, ctype, resp = json_response({"ok": await app.delete_proxy(pid)})
|
|
elif method == "POST" and p.startswith("/api/check/"):
|
|
pid = urllib.parse.unquote(p.rsplit("/", 1)[-1]); proxy = app.get_proxy(pid)
|
|
if not proxy: status, ctype, resp = json_response({"ok": False, "error": "proxy not found"}, 404)
|
|
else:
|
|
await app.run_once(proxy)
|
|
status, ctype, resp = json_response({"ok": True, "state": dataclasses.asdict(app.state_for(pid))})
|
|
elif method == "POST" and p == "/api/channels":
|
|
status, ctype, resp = json_response({"ok": True, "channel": await app.upsert_channel(data)}, 201)
|
|
elif method == "PUT" and p.startswith("/api/channels/"):
|
|
cid = urllib.parse.unquote(p.rsplit("/", 1)[-1])
|
|
status, ctype, resp = json_response({"ok": True, "channel": await app.upsert_channel(data, cid)})
|
|
elif method == "DELETE" and p.startswith("/api/channels/"):
|
|
cid = urllib.parse.unquote(p.rsplit("/", 1)[-1])
|
|
status, ctype, resp = json_response({"ok": await app.delete_channel(cid)})
|
|
elif method == "POST" and p.startswith("/api/channels/") and p.endswith("/test"):
|
|
cid = urllib.parse.unquote(p.split("/")[-2])
|
|
res = await app.notifier.send_channel_id(cid, "SOCKS5监控测试", f"测试通知 {iso()}")
|
|
status, ctype, resp = json_response({"ok": bool(res.get("ok")), "result": res})
|
|
elif method == "POST" and p == "/api/notify/test":
|
|
res = await app.notifier.send_all("SOCKS5监控测试", f"测试通知 {iso()}")
|
|
status, ctype, resp = json_response({"ok": True, "results": res})
|
|
else:
|
|
status, ctype, resp = json_response({"ok": False, "error": "not found"}, 404)
|
|
except Exception as e:
|
|
traceback.print_exc()
|
|
status, ctype, resp = json_response({"ok": False, "error": str(e)}, 500)
|
|
try:
|
|
reason = {200:"OK",201:"Created",400:"Bad Request",404:"Not Found",500:"Internal Server Error"}.get(status,"OK")
|
|
writer.write(f"HTTP/1.1 {status} {reason}\r\nContent-Type: {ctype}\r\nContent-Length: {len(resp)}\r\nConnection: close\r\n\r\n".encode()+resp)
|
|
await writer.drain()
|
|
finally:
|
|
writer.close()
|
|
with contextlib.suppress(Exception): await writer.wait_closed()
|
|
|
|
|
|
async def main_async(args: argparse.Namespace) -> None:
|
|
app = App(Path(args.config))
|
|
await app.start_tasks()
|
|
server = await asyncio.start_server(lambda r,w: handle_http(r,w,app), args.host, args.port)
|
|
print(f"SOCKS5 monitor panel: http://{args.host}:{args.port}", flush=True)
|
|
print(f"Config: {args.config}", flush=True)
|
|
async with server:
|
|
await server.serve_forever()
|
|
|
|
|
|
def main() -> None:
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--config", default=str(DEFAULT_CONFIG))
|
|
ap.add_argument("--host", default="0.0.0.0")
|
|
ap.add_argument("--port", type=int, default=8787)
|
|
args = ap.parse_args()
|
|
asyncio.run(main_async(args))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|