# 1. Install the full python3 suite (needed for virtual environments)
sudo apt update && sudo apt install python3-full -y
# 2. Create a virtual environment named "venv" in your current directory
python3 -m venv venv
# 3. Activate the virtual environment
source venv/bin/activate
# 4. Install your packages safely inside this isolated environment
pip install ollama requests
import sys
import socket
import requests
import ollama
import json
import urllib3
import re
import time
import os
from urllib.parse import urlparse
# --- IMPORTACIÓN DE SENSORES NVIDIA ---
try:
import pynvml
pynvml.nvmlInit()
HAS_GPU = True
except Exception:
HAS_GPU = False
# Desactivar advertencias de certificados SSL no válidos
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
MODEL_NAME = "gemma4:e2b"
# --- MÉTRICAS GLOBALES ---
METRICAS_LLM = {
"total_tokens_eval": 0,
"total_tokens_gen": 0,
"total_tiempo_ollama": 0.0,
"vram_muestras": []
}
def obtener_vram_actual() -> float:
"""Obtiene el consumo de VRAM actual de la primera GPU
Nvidia en Gigabytes (GB)."""
if not HAS_GPU:
return 0.0
try:
handle =
pynvml.nvmlDeviceGetHandleByIndex(0)
info =
pynvml.nvmlDeviceGetMemoryInfo(handle)
return
info.used / (1024 ** 3)
except
Exception:
return 0.0
def obtener_ip(url: str) -> str:
"""Extrae el dominio de la URL y resuelve su dirección
IP."""
try:
parsed_url
= urlparse(url)
dominio =
parsed_url.netloc if parsed_url.netloc else parsed_url.path.split('/')[0]
if ':' in
dominio:
dominio
= dominio.split(':')[0]
return
socket.gethostbyname(dominio)
except
Exception:
return
"No se pudo resolver la IP"
def limpiar_nombre_dominio(url: str) -> str:
"""Limpia la URL para dejar un nombre de archivo válido
basado en el dominio."""
parsed_url =
urlparse(url)
dominio =
parsed_url.netloc if parsed_url.netloc else parsed_url.path.split('/')[0]
dominio =
dominio.split(':')[0]
return
re.sub(r'[^\w\.-]', '_', dominio)
def consultar_al_modelo(prompt: str) -> str:
"""Envía un prompt a gemma4:e2b y acumula métricas de
rendimiento."""
try:
response =
ollama.generate(
model=MODEL_NAME,
prompt=prompt,
options={
"temperature": 0.1,
"top_p": 0.9
}
)
METRICAS_LLM["total_tokens_eval"] +=
response.get("prompt_eval_count", 0)
METRICAS_LLM["total_tokens_gen"] +=
response.get("eval_count", 0)
tiempo_segundos = response.get("total_duration", 0) /
1_000_000_000
METRICAS_LLM["total_tiempo_ollama"] += tiempo_segundos
return
response['response']
except
Exception as e:
return
f"Error al conectar con el modelo: {e}"
def auditar_punto_owasp(base_url: str, categoria_owasp:
str, endpoint: str, lista_resumen: list):
print(f"\n[+] Analizando {categoria_owasp} en
{base_url}{endpoint}...")
prompt_payload
= f"""
Actúa como un
experto en ciberseguridad. Para la categoría '{categoria_owasp}',
genera una
lista de 3 payloads comunes en formato JSON plano (ej. ["payload1",
"payload2", "payload3"]).
Solo devuelve
la lista JSON, nada de texto adicional.
"""
respuesta_payloads = consultar_al_modelo(prompt_payload)
try:
payloads =
json.loads(respuesta_payloads)
except:
payloads =
["' OR 1=1 --", "<script>alert(1)</script>",
"../etc/passwd"]
vulnerable_en_categoria = "NO"
for payload in
payloads:
url_prueba
= f"{base_url}{endpoint}"
params =
{"input": payload, "search": payload, "id":
payload, "url": payload}
try:
print(f" [*] Probando
payload: {payload}")
response = requests.get(url_prueba, params=params, timeout=5,
verify=False)
if
response.status_code == 404:
print(f" [!] Advertencia:
El endpoint devuelve 404 (No encontrado).")
elif
response.status_code == 403:
print(f" [!] Advertencia:
Acceso denegado 403 (WAF / Cloudflare activo).")
prompt_analisis = f"""
Analiza
la siguiente respuesta HTTP del servidor para determinar si el payload
'{payload}'
explotó
con éxito una vulnerabilidad de la categoría {categoria_owasp}.
Código
de Estado: {response.status_code}
Cuerpo
de la respuesta (primeros 500 caracteres):
{response.text[:500] if response.text else '[Cuerpo Vacío]'}
Responde estrictamente con este formato:
DIAMETRO: [SÍ o NO o SOSPECHOSO]
RAZÓN:
(Tu breve explicación aquí)
"""
analisis = consultar_al_modelo(prompt_analisis)
print(f" [Resultado
AI]:\n{analisis}")
lineas_analisis = analisis.upper().split('\n')
for
linea in lineas_analisis:
if
"DIAMETRO:" in linea:
if "SÍ" in linea or "SI" in linea:
vulnerable_en_categoria = "SÍ"
break
elif "SOSPECHOSO" in linea and vulnerable_en_categoria !=
"SÍ":
vulnerable_en_categoria = "SOSPECHOSO"
except
requests.exceptions.RequestException as e:
print(f" [-] Error de
conexión en este payload: {e}")
vulnerable_en_categoria = "ERROR DE CONEXIÓN"
vram_actual =
obtener_vram_actual()
if HAS_GPU:
print(f" [GPU Info]: VRAM
en uso actual: {vram_actual:.2f} GB")
METRICAS_LLM["vram_muestras"].append(vram_actual)
lista_resumen.append({
"categoria": categoria_owasp,
"endpoint": endpoint,
"estado": vulnerable_en_categoria,
"vram_alcanzada": vram_actual if HAS_GPU else None
})
if __name__ == "__main__":
if
len(sys.argv) < 2:
print("\n[!] Error: No especificaste el sitio web objetivo.")
print("Uso correcto: python3 audit.py <URL_DEL_SITIO>")
print("Ejemplo: python3
audit.py https://n1g.cl\n")
sys.exit(1)
target_url =
sys.argv[1].rstrip("/")
target_ip =
obtener_ip(target_url)
nombre_limpio =
limpiar_nombre_dominio(target_url)
user_home =
os.path.expanduser("~")
report_file =
os.path.join(user_home, f"reporte_{nombre_limpio}.txt")
print(f"Iniciando escáner OWASP asistido por AI
({MODEL_NAME})")
print(f"Objetivo base: {target_url}")
print(f"IP
objetivo: {target_ip}")
print(f"Archivo de salida asignado: {report_file}")
if not HAS_GPU:
print("[!] Advertencia: No se detectó GPU Nvidia compatible.
Monitoreo VRAM desactivado.")
resumen_final =
[]
pruebas = [
{"categoria": "A01:2021-Broken Access Control",
"endpoint": "/view.php"},
{"categoria": "A02:2021-Cryptographic Failures",
"endpoint": "/config.json"},
{"categoria": "A03:2021-Injection",
"endpoint": "/login.php"},
{"categoria": "A04:2021-Insecure Design",
"endpoint": "/checkout"},
{"categoria": "A05:2021-Security Misconfiguration",
"endpoint": "/admin/"},
{"categoria": "A06:2021-Vulnerable and Outdated
Components", "endpoint": "/api/v1/"},
{"categoria": "A07:2021-Identification and Authentication
Failures", "endpoint": "/reset-password"},
{"categoria": "A08:2021-Software and Data Integrity
Failures", "endpoint": "/update"},
{"categoria": "A09:2021-Security Logging and Monitoring
Failures", "endpoint": "/logs/"},
{"categoria": "A10:2021-Server-Side Request Forgery
(SSRF)", "endpoint": "/webhook"}
]
inicio_proceso
= time.time()
for prueba in
pruebas:
auditar_punto_owasp(target_url, prueba["categoria"],
prueba["endpoint"], resumen_final)
tiempo_total_proceso = time.time() - inicio_proceso
tps_generacion
= 0.0
if
METRICAS_LLM["total_tiempo_ollama"] > 0:
tps_generacion = METRICAS_LLM["total_tokens_gen"] /
METRICAS_LLM["total_tiempo_ollama"]
vram_promedio =
0.0
if
METRICAS_LLM["vram_muestras"]:
vram_promedio = sum(METRICAS_LLM["vram_muestras"]) /
len(METRICAS_LLM["vram_muestras"])
lineas_reporte
= []
lineas_reporte.append("\n" + "="*50)
lineas_reporte.append(" RESUMEN DE AUDITORÍA OWASP TOP 10")
lineas_reporte.append(f" Objetivo: {target_url}")
lineas_reporte.append(f" IP:
{target_ip}")
lineas_reporte.append("="*50)
for i, item in
enumerate(resumen_final, start=1):
lineas_reporte.append(f"{i}. {item['categoria']}")
lineas_reporte.append(f" -
Objetivo: {item['endpoint']}")
lineas_reporte.append(f" -
Estado: {item['estado']}")
if
item['vram_alcanzada'] is not None:
lineas_reporte.append(f" -
VRAM Usada: {item['vram_alcanzada']:.2f} GB")
lineas_reporte.append("-" * 50)
lineas_reporte.append("\n" + "="*50)
lineas_reporte.append(" MÉTRICAS DE RENDIMIENTO Y
PROCESAMIENTO")
lineas_reporte.append("="*50)
lineas_reporte.append(f" Tiempo total del proceso: {tiempo_total_proceso:.2f}
segundos")
lineas_reporte.append(f" Tiempo neto en Ollama (LLM): {METRICAS_LLM['total_tiempo_ollama']:.2f}
segundos")
lineas_reporte.append(f" Tokens de entrada procesados: {METRICAS_LLM['total_tokens_eval']}
tokens")
lineas_reporte.append(f" Tokens de salida generados: {METRICAS_LLM['total_tokens_gen']}
tokens")
lineas_reporte.append(f" Velocidad de respuesta (LLM): {tps_generacion:.2f} tokens/seg")
if HAS_GPU:
lineas_reporte.append(f" Consumo de VRAM Promedio: {vram_promedio:.2f} GB")
lineas_reporte.append("="*50)
texto_final =
"\n".join(lineas_reporte)
print(texto_final)
try:
with
open(report_file, "w", encoding="utf-8") as f:
f.write(texto_final)
print(f"\n[+] El reporte ha sido guardado exitosamente en:
{report_file}\n")
except
Exception as error_guardado:
print(f"\n[!] Error inesperado al escribir el reporte en disco:
{error_guardado}")
if HAS_GPU:
try:
pynvml.nvmlShutdown()
except
Exception:
pass
No comments:
Post a Comment