131 lines
5.5 KiB
Python
Executable file
131 lines
5.5 KiB
Python
Executable file
#!/usr/bin/env python3
|
||
"""
|
||
RKN Active Probing Monitor & Telegram Alerting
|
||
Отслеживает попытки сканирования со стороны подсетей РКН / ГРЧЦ (212.192.156.0/22)
|
||
и отправляет сводку в топик 18 супергруппы «Майа-Штаб» (-1004488599816).
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import json
|
||
import socket
|
||
import urllib.request
|
||
import subprocess
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
# Конфигурация (читается из .env или переменных окружения)
|
||
ENV_PATH = Path(__file__).resolve().parent.parent / ".env"
|
||
config = {}
|
||
if ENV_PATH.exists():
|
||
with open(ENV_PATH, "r", encoding="utf-8") as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if line and not line.startswith("#") and "=" in line:
|
||
k, v = line.split("=", 1)
|
||
config[k.strip()] = v.strip().strip("'\"")
|
||
|
||
BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN", config.get("TELEGRAM_BOT_TOKEN", "8065539545:AAECBnyjinUVVGv3vhtoWZ99tUmnT9sze-k"))
|
||
CHAT_ID = os.getenv("TELEGRAM_CHAT_ID", config.get("TELEGRAM_CHAT_ID", "-1004488599816"))
|
||
TOPIC_ID = int(os.getenv("TELEGRAM_TOPIC_ID", config.get("TELEGRAM_TOPIC_ID", "18")))
|
||
|
||
STATE_FILE = Path("/tmp/rkn_probe_monitor_state.json")
|
||
RKN_COMMENT = "RKN_ACTIVE_PROBING_DROP"
|
||
|
||
def get_iptables_stats():
|
||
"""Считывает счетчики пакетов и байт из iptables для правила RKN."""
|
||
try:
|
||
cmd = ["iptables", "-L", "INPUT", "-v", "-n", "-x"]
|
||
res = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=True)
|
||
for line in res.stdout.splitlines():
|
||
if RKN_COMMENT in line:
|
||
parts = line.strip().split()
|
||
if len(parts) >= 2:
|
||
packets = int(parts[0])
|
||
bytes_count = int(parts[1])
|
||
return packets, bytes_count
|
||
except Exception as e:
|
||
print(f"Ошибка чтения iptables: {e}", file=sys.stderr)
|
||
return None, None
|
||
|
||
def get_server_identity():
|
||
"""Возвращает имя хоста и внешний IP."""
|
||
hostname = socket.gethostname()
|
||
external_ip = "Unknown"
|
||
try:
|
||
req = urllib.request.Request("https://ifconfig.me/ip", headers={"User-Agent": "curl/7.68.0"})
|
||
with urllib.request.urlopen(req, timeout=5) as resp:
|
||
external_ip = resp.read().decode().strip()
|
||
except Exception:
|
||
pass
|
||
return hostname, external_ip
|
||
|
||
def send_telegram_alert(text):
|
||
"""Отправляет сообщение в топик супергруппы Телеграм."""
|
||
url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"
|
||
payload = {
|
||
"chat_id": CHAT_ID,
|
||
"text": text,
|
||
"parse_mode": "HTML",
|
||
"message_thread_id": TOPIC_ID
|
||
}
|
||
req = urllib.request.Request(
|
||
url,
|
||
data=json.dumps(payload).encode("utf-8"),
|
||
headers={"Content-Type": "application/json"}
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=10) as resp:
|
||
data = json.loads(resp.read().decode())
|
||
return data.get("ok", False)
|
||
except Exception as e:
|
||
print(f"Ошибка отправки в Telegram: {e}", file=sys.stderr)
|
||
return False
|
||
|
||
def main():
|
||
pkts, bytes_total = get_iptables_stats()
|
||
if pkts is None:
|
||
print("Правило iptables с меткой RKN_ACTIVE_PROBING_DROP не найдено.")
|
||
sys.exit(0)
|
||
|
||
# Читаем предыдущее состояние
|
||
prev_pkts = 0
|
||
prev_bytes = 0
|
||
if STATE_FILE.exists():
|
||
try:
|
||
with open(STATE_FILE, "r", encoding="utf-8") as f:
|
||
state = json.load(f)
|
||
prev_pkts = state.get("packets", 0)
|
||
prev_bytes = state.get("bytes", 0)
|
||
except Exception:
|
||
pass
|
||
|
||
delta_pkts = pkts - prev_pkts
|
||
delta_bytes = bytes_total - prev_bytes
|
||
|
||
# Сохраняем текущее состояние
|
||
with open(STATE_FILE, "w", encoding="utf-8") as f:
|
||
json.dump({"packets": pkts, "bytes": bytes_total, "updated_at": datetime.now().isoformat()}, f)
|
||
|
||
if delta_pkts > 0:
|
||
hostname, ext_ip = get_server_identity()
|
||
domain = config.get("DOMAIN", hostname)
|
||
now_str = datetime.now().strftime("%d.%m.%Y %H:%M:%S")
|
||
|
||
msg = (
|
||
f"🚨 <b>[ACTIVE PROBING DETECTED] Попытка сканирования от РКН!</b>\n\n"
|
||
f"🌐 <b>Сервер:</b> <code>{domain}</code> ({ext_ip})\n"
|
||
f"🎯 <b>Источник:</b> AS61280 ФГУП «ГРЧЦ» (ЦМУ ССОП)\n"
|
||
f"🚫 <b>Подсеть сканера:</b> <code>212.192.156.0/22</code>\n\n"
|
||
f"📊 <b>Новых пакетов:</b> <code>+{delta_pkts}</code> ({delta_bytes / 1024:.1f} КБ)\n"
|
||
f"📈 <b>Всего заблокировано:</b> <code>{pkts}</code> пакетов ({bytes_total / 1024 / 1024:.2f} МБ)\n"
|
||
f"🛡️ <b>Результат:</b> Пакеты жестко сброшены в <b>DROP</b>. Сканер не получил ответа и считает хост мертвым.\n"
|
||
f"🕒 <i>{now_str}</i>"
|
||
)
|
||
print(f"Обнаружен пробинг: +{delta_pkts} пакетов! Отправка алерта...")
|
||
send_telegram_alert(msg)
|
||
else:
|
||
print(f"Новых попыток сканирования не зафиксировано. Всего дропнуто: {pkts} пакетов.")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|