馃洝️ PASAIA LAB: Monitor de Ciberseguridad con Big Data e IA Descentralizada
WALLET - MONEDERO: INGRESOS BTC
En PASAIA LAB, hemos desarrollado un sistema de defensa cibern茅tica de vanguardia que integra la potencia de Big Data con la inteligencia colectiva de redes descentralizadas como Bittensor (TAO) y la automatizaci贸n de agentes aut贸nomos en NEAR Protocol.
El Desaf铆o: Ataques Cada Vez M谩s Sofisticados
Los ataques cibern茅ticos ya no son simples. Requieren una defensa que no solo reaccione a lo conocido, sino que aprenda, se adapte y act煤e de forma aut贸noma. Aqu铆 es donde nuestro "Escudo Pasaia 2026" marca la diferencia.
¿C贸mo Funciona el Escudo de PASAIA LAB?
- Vigilancia Global (Bittensor - TAO): Antes de que un dato entre a nuestra red, se consulta a una vasta red de modelos de IA en Bittensor. Si la inteligencia colectiva global detecta patrones maliciosos (ej. phishing, malware de d铆a cero), el tr谩fico es bloqueado preventivamente.
- Agentes Aut贸nomos (NEAR Protocol): Si una anomal铆a es detectada internamente, un "Agente de Seguridad" aut贸nomo desplegado en NEAR ejecuta un Smart Contract para, por ejemplo, congelar credenciales, aislar un dispositivo o desviar el tr谩fico sospechoso. Todo esto ocurre en milisegundos.
- Memoria y Aprendizaje (Big Data Interno): Todos los eventos, normales y an贸malos, se registran en nuestro "脕rbol de Datos". Esto no solo cumple con auditor铆as, sino que tambi茅n sirve para re-entrenar nuestros modelos de IA, haciendo el sistema m谩s robusto con cada incidente.
Monitor en Tiempo Real: La Sala de Control de PASAIA LAB
Para visualizar este proceso, hemos creado un monitor en Python que simula la detecci贸n de anomal铆as en el tr谩fico de red. Utiliza el algoritmo de Z-Score para identificar picos de actividad inusuales, que podr铆an indicar exfiltraci贸n de datos o un ataque.
import numpy as np
import time
import json
from datetime import datetime
import matplotlib.pyplot as plt
import matplotlib.animation as animation
from collections import deque
class PasaiaShield:
def __init__(self, threshold=3.0):
self.threshold = threshold
self.history = deque(maxlen=100)
self.audit_log_file = "audit_pasaia_lab.json"
# Para el monitor visual
self.x_data = deque(maxlen=50)
self.y_data = deque(maxlen=50)
self.z_scores = deque(maxlen=50)
self.alerts = deque(maxlen=50)
def ingest_traffic(self, packet_size):
self.history.append(packet_size)
def log_attack_to_json(self, packet_size, z_score):
attack_event = {
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"event_type": "ANOMALY_DETECTED",
"packet_size_kb": packet_size,
"severity_score": round(z_score, 2),
"protocol_action": "NEAR_SMART_CONTRACT_BLOCK",
"threat_intelligence": "TAO_SUBNET_REPORTED"
}
try:
try:
with open(self.audit_log_file, "r") as f:
data = json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
data = []
data.append(attack_event)
with open(self.audit_log_file, "w") as f:
json.dump(data, f, indent=4)
print(f"馃捑 Evento registrado en {self.audit_log_file}")
except Exception as e:
print(f"❌ Error al guardar auditor铆a: {e}")
def analyze_risk(self, current_packet, index):
if len(self.history) < 10:
return "ENTRENANDO...", 0, False
mean = np.mean(self.history)
std_dev = np.std(self.history)
z_score = abs(current_packet - mean) / std_dev if std_dev > 0 else 0
is_alert = False
if z_score > self.threshold:
self.log_attack_to_json(current_packet, z_score)
is_alert = True
self.x_data.append(index)
self.y_data.append(current_packet)
self.z_scores.append(z_score)
self.alerts.append(is_alert)
return "⚠️ ALERTA: ANOMAL脥A DETECTADA" if is_alert else "✅ TR脕FICO NORMAL", z_score, is_alert
def animate(i, shield_instance, line_packet, line_zscore, ax1, ax2):
if i % 10 == 0 and i > 0:
packet = np.random.normal(5000, 100) if np.random.rand() < 0.2 else np.random.normal(500, 50)
else:
packet = np.random.normal(500, 50)
shield_instance.ingest_traffic(packet)
status, score, is_alert = shield_instance.analyze_risk(packet, i)
line_packet.set_data(list(shield_instance.x_data), list(shield_instance.y_data))
ax1.set_xlim(shield_instance.x_data[0], shield_instance.x_data[-1] + 1)
ax1.set_ylim(min(shield_instance.y_data) * 0.9, max(shield_instance.y_data) * 1.1)
line_zscore.set_data(list(shield_instance.x_data), list(shield_instance.z_scores))
ax2.set_xlim(shield_instance.x_data[0], shield_instance.x_data[-1] + 1)
ax2.set_ylim(0, max(max(shield_instance.z_scores) * 1.2, shield_instance.threshold * 1.5))
ax2.axhline(shield_instance.threshold, color='r', linestyle='--', label=f'Umbral Z-Score ({shield_instance.threshold})')
alert_x = [shield_instance.x_data[j] for j, alert in enumerate(shield_instance.alerts) if alert]
alert_y = [shield_instance.y_data[j] for j, alert in enumerate(shield_instance.alerts) if alert]
ax1.plot(alert_x, alert_y, 'ro', markersize=8, fillstyle='none')
ax1.set_title(f"PASAIA LAB: Monitor de Tr谩fico | {status}", color='red' if is_alert else 'green')
return line_packet, line_zscore,
if __name__ == "__main__":
escudo = PasaiaShield(threshold=3.0)
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 8))
fig.suptitle('PASAIA LAB: Escudo de Ciberseguridad IA', fontsize=16)
line_packet, = ax1.plot([], [], 'g-', label='Tama帽o de Paquete (KB)')
ax1.set_ylabel('Tama帽o de Paquete (KB)')
ax1.legend()
ax1.grid(True)
line_zscore, = ax2.plot([], [], 'b-', label='Z-Score')
ax2.set_xlabel('Tiempo (Iteraciones)')
ax2.set_ylabel('Z-Score')
ax2.legend()
ax2.grid(True)
ax2.axhline(escudo.threshold, color='r', linestyle='--', label=f'Umbral Z-Score ({escudo.threshold})')
ani = animation.FuncAnimation(fig, animate, fargs=(escudo, line_packet, line_zscore, ax1, ax2),
interval=100, blit=True, cache_frame_data=False)
plt.tight_layout(rect=[0, 0.03, 1, 0.95])
plt.show()
import numpy as np
import time
import json
from datetime import datetime
import matplotlib.pyplot as plt
import matplotlib.animation as animation
from collections import deque # Para almacenar datos de forma eficiente
class PasaiaShield:
def __init__(self, threshold=3.0):
self.threshold = threshold
self.history = deque(maxlen=100) # Usamos deque para eficiencia
self.audit_log_file = "audit_pasaia_lab.json"
# Para el monitor visual
self.x_data = deque(maxlen=50) # Tiempo o 铆ndices
self.y_data = deque(maxlen=50) # Tama帽o de paquete
self.z_scores = deque(maxlen=50) # Z-Score calculado
self.alerts = deque(maxlen=50) # Marcar alertas
def ingest_traffic(self, packet_size):
"""Simula la entrada de datos al sistema"""
self.history.append(packet_size)
def log_attack_to_json(self, packet_size, z_score):
"""Guarda el ataque en la base de datos de auditor铆a"""
attack_event = {
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"event_type": "ANOMALY_DETECTED",
"packet_size_kb": packet_size,
"severity_score": round(z_score, 2),
"protocol_action": "NEAR_SMART_CONTRACT_BLOCK",
"threat_intelligence": "TAO_SUBNET_REPORTED"
}
try:
try:
with open(self.audit_log_file, "r") as f:
data = json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
data = []
data.append(attack_event)
with open(self.audit_log_file, "w") as f:
json.dump(data, f, indent=4)
print(f"馃捑 Evento registrado en {self.audit_log_file}")
except Exception as e:
print(f"❌ Error al guardar auditor铆a: {e}")
def analyze_risk(self, current_packet, index):
if len(self.history) < 10:
return "ENTRENANDO...", 0, False
mean = np.mean(self.history)
std_dev = np.std(self.history)
z_score = abs(current_packet - mean) / std_dev if std_dev > 0 else 0
is_alert = False
if z_score > self.threshold:
self.log_attack_to_json(current_packet, z_score)
is_alert = True
# Actualiza datos para el monitor
self.x_data.append(index)
self.y_data.append(current_packet)
self.z_scores.append(z_score)
self.alerts.append(is_alert)
return "⚠️ ALERTA: ANOMAL脥A DETECTADA" if is_alert else "✅ TR脕FICO NORMAL", z_score, is_alert
# --- FUNCI脫N DE ACTUALIZACI脫N DEL MONITOR ---
def animate(i, shield_instance, line_packet, line_zscore, ax1, ax2):
# Genera un paquete de tr谩fico (simulaci贸n)
if i % 10 == 0 and i > 0: # Simula un ataque cada cierto tiempo
packet = np.random.normal(5000, 100) if np.random.rand() < 0.2 else np.random.normal(500, 50)
else:
packet = np.random.normal(500, 50) # Tr谩fico normal
shield_instance.ingest_traffic(packet)
status, score, is_alert = shield_instance.analyze_risk(packet, i)
# Actualiza el gr谩fico de tama帽o de paquete
line_packet.set_data(list(shield_instance.x_data), list(shield_instance.y_data))
ax1.set_xlim(shield_instance.x_data[0], shield_instance.x_data[-1] + 1)
ax1.set_ylim(min(shield_instance.y_data) * 0.9, max(shield_instance.y_data) * 1.1)
# Actualiza el gr谩fico de Z-Score
line_zscore.set_data(list(shield_instance.x_data), list(shield_instance.z_scores))
ax2.set_xlim(shield_instance.x_data[0], shield_instance.x_data[-1] + 1)
ax2.set_ylim(0, max(max(shield_instance.z_scores) * 1.2, shield_instance.threshold * 1.5))
ax2.axhline(shield_instance.threshold, color='r', linestyle='--', label=f'Umbral Z-Score ({shield_instance.threshold})')
# Marcar alertas
alert_x = [shield_instance.x_data[j] for j, alert in enumerate(shield_instance.alerts) if alert]
alert_y = [shield_instance.y_data[j] for j, alert in enumerate(shield_instance.alerts) if alert]
ax1.plot(alert_x, alert_y, 'ro', markersize=8, fillstyle='none') # C铆rculos rojos en los paquetes an贸malos
# T铆tulo din谩mico
ax1.set_title(f"PASAIA LAB: Monitor de Tr谩fico | {status}", color='red' if is_alert else 'green')
return line_packet, line_zscore,
# --- CONFIGURACI脫N DEL MONITOR ---
if __name__ == "__main__":
escudo = PasaiaShield(threshold=3.0) # Umbral de alerta m谩s estricto
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 8))
fig.suptitle('PASAIA LAB: Escudo de Ciberseguridad IA', fontsize=16)
# Gr谩fico 1: Tama帽o de Paquete
line_packet, = ax1.plot([], [], 'g-', label='Tama帽o de Paquete (KB)')
ax1.set_ylabel('Tama帽o de Paquete (KB)')
ax1.legend()
ax1.grid(True)
# Gr谩fico 2: Z-Score de Anomal铆a
line_zscore, = ax2.plot([], [], 'b-', label='Z-Score')
ax2.set_xlabel('Tiempo (Iteraciones)')
ax2.set_ylabel('Z-Score')
ax2.legend()
ax2.grid(True)
ax2.axhline(escudo.threshold, color='r', linestyle='--', label=f'Umbral Z-Score ({escudo.threshold})') # L铆nea de umbral
# Inicia la animaci贸n
ani = animation.FuncAnimation(fig, animate, fargs=(escudo, line_packet, line_zscore, ax1, ax2),
interval=100, blit=True, cache_frame_data=False) # Intervalo en ms
plt.tight_layout(rect=[0, 0.03, 1, 0.95])
plt.show()
Visualizaci贸n en Acci贸n (Captura del Monitor)
Aqu铆 puedes insertar una imagen (JPG/PNG) o un GIF animado de c贸mo se ve el monitor en tiempo real. Esto har谩 que tu publicaci贸n sea mucho m谩s atractiva y f谩cil de entender.
Captura de pantalla de la interfaz de monitorizaci贸n de PASAIA LAB.
Conclusi贸n: Hacia una Ciberseguridad Inteligente y Descentralizada
El "Escudo Pasaia 2026" representa el futuro de la ciberseguridad: un sistema proactivo, aut贸nomo y globalmente inteligente. No solo protegemos nuestros datos, sino que contribuimos a una red de defensa m谩s robusta para todos.
Equipo de Seguridad, PASAIA LAB
https://substack.com/@agustintxo
https://agustintxo.substack.com/
BRAINSTORMING
- Tormenta de Ideas de PASAIA LAB © 2025 by Jos茅
Agust铆n Font谩n Varela is licensed under CC
BY-NC-ND 4.0
BRAINSTORMING
- Tormenta de Ideas de PASAIA LAB © 2025 by Jos茅
Agust铆n Font谩n Varela is licensed under Creative
Commons Attribution-NonCommercial-NoDerivatives 4.0 International
Tormenta Work Free Intelligence + IA Free Intelligence Laboratory by Jos茅 Agust铆n Font谩n Varela is licensed under CC BY-NC-ND 4.0
