Spaces:
Running
Running
File size: 5,389 Bytes
7cd15ab | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 | import os
import time
import logging
from datetime import datetime
from collections import deque
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional
logger = logging.getLogger(__name__)
@dataclass
class SecurityIncident:
timestamp: float
threat_type: str
severity: str
snippet: str
session_id: str
source_ip: str
details: Dict[str, Any] = field(default_factory=dict)
class SecurityMonitor:
MAX_INCIDENTS = 1000
def __init__(self):
self._incidents: deque = deque(maxlen=self.MAX_INCIDENTS)
self._stats_cache: Optional[Dict] = None
self._stats_time: float = 0
self._stats_ttl: float = 2.0
logger.info("🔒 SecurityMonitor inicializado (memoria, máx 1000 incidentes)")
def log_incident(
self,
threat_type: str,
severity: str,
snippet: str,
session_id: str = "unknown",
source_ip: str = "0.0.0.0",
details: Optional[Dict] = None,
) -> SecurityIncident:
incident = SecurityIncident(
timestamp=time.time(),
threat_type=threat_type,
severity=severity,
snippet=snippet[:120],
session_id=session_id,
source_ip=source_ip,
details=details or {},
)
self._incidents.append(incident)
self._stats_cache = None
log_msg = f"🚨 [{severity.upper()}] {threat_type} | sesión={session_id} | {snippet[:80]}"
if severity in ("high", "critical"):
logger.error(log_msg)
self._try_telegram_alert(incident)
else:
logger.warning(log_msg)
return incident
def get_stats(self) -> Dict[str, Any]:
now = time.time()
if self._stats_cache and (now - self._stats_time) < self._stats_ttl:
return self._stats_cache
total = len(self._incidents)
by_severity: Dict[str, int] = {}
by_type: Dict[str, int] = {}
for inc in self._incidents:
by_severity[inc.severity] = by_severity.get(inc.severity, 0) + 1
by_type[inc.threat_type] = by_type.get(inc.threat_type, 0) + 1
window = 3600
cutoff = time.time() - window
recent_count = sum(1 for inc in self._incidents if inc.timestamp >= cutoff)
result = {
"total_incidents": total,
"by_severity": dict(sorted(by_severity.items())),
"by_type": dict(sorted(by_type.items(), key=lambda x: x[1], reverse=True)),
"incidents_last_hour": recent_count,
"max_capacity": self.MAX_INCIDENTS,
"usage_pct": round(total / self.MAX_INCIDENTS * 100, 1) if total else 0.0,
}
self._stats_cache = result
self._stats_time = now
return result
def get_recent_incidents(self, limit: int = 50, min_severity: str = "low") -> List[Dict]:
severity_order = {"low": 0, "medium": 1, "high": 2, "critical": 3}
min_order = severity_order.get(min_severity, 0)
result = []
for inc in reversed(self._incidents):
if severity_order.get(inc.severity, 0) >= min_order:
result.append({
"timestamp": datetime.fromtimestamp(inc.timestamp).isoformat(),
"threat_type": inc.threat_type,
"severity": inc.severity,
"snippet": inc.snippet,
"session_id": inc.session_id,
"source_ip": inc.source_ip,
"details": inc.details,
})
if len(result) >= limit:
break
return result
def _try_telegram_alert(self, incident: SecurityIncident) -> None:
try:
bot_token = os.environ.get("TELEGRAM_BOT_TOKEN", "")
chat_id = os.environ.get("TELEGRAM_CHAT_ID", "")
if not bot_token or not chat_id:
logger.debug("🔒 Telegram no configurado, omitiendo alerta")
return
import requests
fecha = datetime.fromtimestamp(incident.timestamp).strftime("%d/%m/%Y %H:%M:%S")
message = (
f"🚨 *ALERTA DE SEGURIDAD*\n"
f"• Tipo: `{incident.threat_type}`\n"
f"• Severidad: *{incident.severity.upper()}*\n"
f"• Sesión: `{incident.session_id}`\n"
f"• IP: `{incident.source_ip}`\n"
f"• Fragmento: `{incident.snippet[:80]}`\n"
f"• Fecha: {fecha}\n"
f"• Tokens consumidos: 0"
)
url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
resp = requests.post(url, json={
"chat_id": chat_id,
"text": message,
"parse_mode": "Markdown",
}, timeout=15)
if resp.status_code == 200:
logger.debug("🔒 Alerta de seguridad enviada a Telegram")
else:
logger.debug(f"🔒 Telegram respondió {resp.status_code}")
except Exception as e:
logger.debug(f"🔒 Error enviando alerta Telegram: {e}")
_monitor: Optional[SecurityMonitor] = None
def get_monitor() -> SecurityMonitor:
global _monitor
if _monitor is None:
_monitor = SecurityMonitor()
return _monitor
|