Public Access
70 lines
2.8 KiB
Python
70 lines
2.8 KiB
Python
import json
|
|
import os
|
|
import time
|
|
import requests
|
|
from datetime import datetime
|
|
from modules.logger import log_event
|
|
|
|
LOG_FILE = "aethas_ip_logs.json"
|
|
|
|
# --- DÉFINITION CENTRALISÉE DES PROFILS PRIVILÉGIÉS ---
|
|
UTILISATEURS_VIP = ["xavier", "céline", "anaëlle", "renan", "audric", "admin"]
|
|
|
|
def verifier_et_bloquer_vpn(ip_address, utilisateur="visiteur"):
|
|
"""
|
|
Vérifie si une IP appartient à un VPN, un Proxy ou un Datacenter.
|
|
Intègre une liste blanche (Whitelist) pour les réseaux locaux et les VIP.
|
|
"""
|
|
# 1. Bypass matériel : On ne bloque jamais les requêtes locales (Serveur, MeshCentral, Beszel, Gluetun)
|
|
if ip_address in ["127.0.0.1", "localhost", "::1"] or ip_address.startswith(("192.168.", "10.", "172.")):
|
|
return False
|
|
|
|
# 2. Bypass profil : Passe-droit pour les VIP et Administrateurs
|
|
if utilisateur.lower() in UTILISATEURS_VIP:
|
|
log_event("INFO", "SECURITY", f"Passe-droit VPN accordé à l'utilisateur VIP : {utilisateur} ({ip_address})")
|
|
return False
|
|
|
|
try:
|
|
# Requête ultra-légère, timeout court pour ne pas ralentir le site
|
|
reponse = requests.get(f"https://blackbox.ipinfo.app/lookup/{ip_address}", timeout=3)
|
|
if reponse.text.strip() == "Y":
|
|
log_event("WARNING", "SECURITY", f"Tentative de connexion VPN bloquée pour l'IP : {ip_address}")
|
|
return True # True = C'est un VPN public/datacenter, il faut bloquer
|
|
except requests.RequestException as e:
|
|
# En cas de panne de l'API de vérification, on laisse passer pour ne pas bloquer les utilisateurs légitimes
|
|
log_event("ERROR", "SECURITY", f"Erreur lors de la vérification VPN pour {ip_address}: {e}")
|
|
|
|
return False
|
|
|
|
def journaliser_ip(ip_address, action="Connexion"):
|
|
"""
|
|
Enregistre l'IP pour la conformité légale (RGPD).
|
|
Purge automatique des logs vieux de plus de 365 jours.
|
|
"""
|
|
logs = []
|
|
|
|
# 1. Chargement de l'historique existant
|
|
if os.path.exists(LOG_FILE):
|
|
with open(LOG_FILE, 'r', encoding='utf-8') as f:
|
|
try:
|
|
logs = json.load(f)
|
|
except json.JSONDecodeError:
|
|
logs = []
|
|
|
|
# 2. Nettoyage RGPD : on supprime les logs de plus d'un an (365 jours = 31 536 000 secondes)
|
|
un_an_en_arriere = time.time() - 31536000
|
|
logs = [log for log in logs if log.get("timestamp", 0) > un_an_en_arriere]
|
|
|
|
# 3. Ajout de la nouvelle entrée
|
|
logs.append({
|
|
"ip": ip_address,
|
|
"action": action,
|
|
"date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
|
"timestamp": time.time()
|
|
})
|
|
|
|
# 4. Sauvegarde
|
|
with open(LOG_FILE, 'w', encoding='utf-8') as f:
|
|
json.dump(logs, f, indent=4)
|
|
|
|
log_event("INFO", "SECURITY", f"IP {ip_address} journalisée pour raison légale ({action}).") |