chore: update
This commit is contained in:
@@ -0,0 +1,230 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
pip install pyTelegramBotAPI
|
||||
"""
|
||||
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from collections import defaultdict
|
||||
from datetime import datetime
|
||||
|
||||
import telebot
|
||||
|
||||
# ============================================================
|
||||
# CONFIGURATION — à modifier avant de lancer
|
||||
# ============================================================
|
||||
|
||||
BOT_TOKEN = "7671151228:AAHI_xJqaXkbf2RnDGotn6ZwO_fWgoY-WJk"
|
||||
CHAT_ID = "7950889398"
|
||||
|
||||
LOG_FILE = "/var/log/auth.log"
|
||||
|
||||
BRUTE_FORCE_THRESHOLD = 5
|
||||
BRUTE_FORCE_WINDOW = 60
|
||||
|
||||
# Fenêtre de déduplication : évite d'envoyer 2 fois le même type
|
||||
# d'alerte pour la même IP en moins de DEDUP_WINDOW secondes.
|
||||
# Toute IP hors whitelist est considérée malveillante → on alerte toujours.
|
||||
DEDUP_WINDOW = 10
|
||||
|
||||
WHITELIST_IPS = {"86.217.21.144"}
|
||||
|
||||
ACTIVE_RESPONSE = True
|
||||
|
||||
BAN_ON_BRUTE_FORCE = True
|
||||
BAN_ON_MAX_AUTH = True
|
||||
|
||||
banned_ips: set[str] = set()
|
||||
|
||||
RE_FAILED = re.compile(
|
||||
r"sshd\[\d+\]: Failed (?:password|publickey) for (?:invalid user )?(\S+) from ([\d.]+)"
|
||||
)
|
||||
RE_INVALID_USER = re.compile(r"sshd\[\d+\]: Invalid user (\S+) from ([\d.]+)")
|
||||
RE_ACCEPTED = re.compile(
|
||||
r"sshd\[\d+\]: Accepted (?:password|publickey) for (\S+) from ([\d.]+)"
|
||||
)
|
||||
RE_MAX_AUTH = re.compile(
|
||||
r"sshd\[\d+\]: error: maximum authentication attempts exceeded.*from ([\d.]+)"
|
||||
)
|
||||
RE_DISCONNECT_INVALID = re.compile(
|
||||
r"sshd\[\d+\]: Disconnected from invalid user (\S+) ([\d.]+)"
|
||||
)
|
||||
|
||||
|
||||
failed_attempts: dict[str, list[float]] = defaultdict(list)
|
||||
|
||||
last_sent: dict[tuple, float] = {}
|
||||
|
||||
known_ips: set[str] = set()
|
||||
|
||||
# ============================================================
|
||||
# TELEGRAM
|
||||
# ============================================================
|
||||
|
||||
bot = telebot.TeleBot(BOT_TOKEN, parse_mode="HTML")
|
||||
|
||||
|
||||
def send_alert(message: str) -> None:
|
||||
try:
|
||||
bot.send_message(CHAT_ID, message)
|
||||
except Exception as e:
|
||||
print(f"[TELEGRAM ERROR] {e}", file=sys.stderr)
|
||||
|
||||
|
||||
def dedup_ok(ip: str, event: str) -> bool:
|
||||
"""Retourne True si on peut envoyer l'alerte (pas envoyée depuis DEDUP_WINDOW s)."""
|
||||
key = (ip, event)
|
||||
now = time.time()
|
||||
if key in last_sent and now - last_sent[key] < DEDUP_WINDOW:
|
||||
return False
|
||||
last_sent[key] = now
|
||||
return True
|
||||
|
||||
|
||||
def ts() -> str:
|
||||
return datetime.now().strftime("%d/%m/%Y %H:%M:%S")
|
||||
|
||||
|
||||
# ============================================================
|
||||
# ANALYSE DES LIGNES
|
||||
# ============================================================
|
||||
|
||||
|
||||
def process_line(line: str) -> None:
|
||||
line = line.strip()
|
||||
if not line:
|
||||
return
|
||||
|
||||
m = RE_ACCEPTED.search(line)
|
||||
if m:
|
||||
user, ip = m.group(1), m.group(2)
|
||||
if ip in WHITELIST_IPS:
|
||||
return
|
||||
icon = "🔴" if user == "root" else "🟡"
|
||||
new_ip = "" if ip in known_ips else "(nouvelle IP !)"
|
||||
known_ips.add(ip)
|
||||
send_alert(
|
||||
f"{icon}IP Alertes Connecter: {new_ip}\n"
|
||||
f"👤 Utilisateur : <code>{user}</code>\n"
|
||||
f"🌐 IP : <code>{ip}</code>\n"
|
||||
f"🕐 {ts()}"
|
||||
)
|
||||
return
|
||||
|
||||
m = RE_MAX_AUTH.search(line)
|
||||
if m:
|
||||
ip = m.group(1)
|
||||
if ip in WHITELIST_IPS:
|
||||
return
|
||||
if dedup_ok(ip, "maxauth"):
|
||||
send_alert(f"🚫 Max authentifications atteint\n🌐 IP : {ip}\n🕐 {ts()}")
|
||||
return
|
||||
|
||||
# --- Utilisateur invalide ---
|
||||
m = RE_INVALID_USER.search(line)
|
||||
if m:
|
||||
user, ip = m.group(1), m.group(2)
|
||||
if ip in WHITELIST_IPS:
|
||||
return
|
||||
_record_failure(ip)
|
||||
_alert(ip, user)
|
||||
return
|
||||
|
||||
# --- Mot de passe / clé échouée ---
|
||||
m = RE_FAILED.search(line)
|
||||
if m:
|
||||
user, ip = m.group(1), m.group(2)
|
||||
if ip in WHITELIST_IPS:
|
||||
return
|
||||
_record_failure(ip)
|
||||
_alert(ip, user)
|
||||
return
|
||||
|
||||
|
||||
def _record_failure(ip: str) -> None:
|
||||
now = time.time()
|
||||
failed_attempts[ip].append(now)
|
||||
failed_attempts[ip] = [
|
||||
t for t in failed_attempts[ip] if now - t <= BRUTE_FORCE_WINDOW
|
||||
]
|
||||
|
||||
|
||||
def _alert(ip: str, user: str) -> None:
|
||||
count = len(failed_attempts[ip])
|
||||
|
||||
if count >= BRUTE_FORCE_THRESHOLD:
|
||||
if dedup_ok(ip, "bruteforce"):
|
||||
send_alert(
|
||||
f"🔥 <b>Brute force SSH détecté !</b>\n"
|
||||
f"🌐 IP : <code>{ip}</code>\n"
|
||||
f"👤 Dernier user : <code>{user}</code>\n"
|
||||
f"📊 {count} échecs en {BRUTE_FORCE_WINDOW}s\n"
|
||||
f"🕐 {ts()}"
|
||||
)
|
||||
return
|
||||
|
||||
if dedup_ok(ip, "fail"):
|
||||
send_alert(
|
||||
f"⚠️ <b>Tentative SSH échouée</b>\n"
|
||||
f"🌐 IP : <code>{ip}</code>\n"
|
||||
f"👤 User : <code>{user}</code>\n"
|
||||
f"📊 {count} échec(s) récent(s)\n"
|
||||
f"🕐 {ts()}"
|
||||
)
|
||||
|
||||
|
||||
# ============================================================
|
||||
# LECTURE EN CONTINU (tail -f)
|
||||
# ============================================================
|
||||
|
||||
|
||||
def follow(filepath: str):
|
||||
if not os.path.exists(filepath):
|
||||
print(f"[ERREUR] Fichier introuvable : {filepath}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
with open(filepath, "r", encoding="utf-8", errors="ignore") as f:
|
||||
f.seek(0, os.SEEK_END)
|
||||
print(f"[INFO] Surveillance de {filepath} démarrée.")
|
||||
send_alert(
|
||||
f"✅ <b>SSH Monitor démarré</b>\n📂 <code>{filepath}</code>\n🕐 {ts()}"
|
||||
)
|
||||
|
||||
while True:
|
||||
line = f.readline()
|
||||
if line:
|
||||
yield line
|
||||
else:
|
||||
try:
|
||||
if os.stat(filepath).st_ino != os.fstat(f.fileno()).st_ino:
|
||||
print("[INFO] Rotation du fichier détectée, réouverture.")
|
||||
f.close()
|
||||
f = open(filepath, "r", encoding="utf-8", errors="ignore")
|
||||
except OSError:
|
||||
pass
|
||||
time.sleep(0.2)
|
||||
|
||||
|
||||
# ============================================================
|
||||
# POINT D'ENTRÉE
|
||||
# ============================================================
|
||||
|
||||
if __name__ == "__main__":
|
||||
if BOT_TOKEN == "VOTRE_BOT_TOKEN" or CHAT_ID == "VOTRE_CHAT_ID":
|
||||
print("[ERREUR] Configure BOT_TOKEN et CHAT_ID avant de lancer le script.")
|
||||
sys.exit(1)
|
||||
|
||||
print(
|
||||
f"[INFO] Brute force : {BRUTE_FORCE_THRESHOLD} échecs en {BRUTE_FORCE_WINDOW}s"
|
||||
)
|
||||
print(f"[INFO] Déduplication : {DEDUP_WINDOW}s par (IP, type d'événement)")
|
||||
|
||||
try:
|
||||
for log_line in follow(LOG_FILE):
|
||||
process_line(log_line)
|
||||
except KeyboardInterrupt:
|
||||
print("\n[INFO] Arrêt du moniteur.")
|
||||
send_alert(f"🛑 <b>SSH Monitor arrêté</b>\n🕐 {ts()}")
|
||||
Reference in New Issue
Block a user