#core/CLI.py
import logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(name)s: %(message)s"
)
import subprocess
import typer
logger = logging.getLogger(__name__)
app = typer.Typer()
def docker_commands(*args, mode=None):
try:
if mode == "test":
subprocess.run(
["docker", "compose", "-f", "docker-compose.dev.yml", *args],
check=True
)
else:
subprocess.run(
["docker", "compose", *args],
check=True
)
except FileNotFoundError:
typer.echo("Docker no está instalado o no está en el PATH")
raise typer.Exit(code=127) # 127 = "command not found" en convención Unix
except KeyboardInterrupt:
raise typer.Exit(code=0)
except subprocess.CalledProcessError as e:
if e.returncode == -2: # SIGINT
raise typer.Exit(code=0)
if e.returncode < 0:
code = 128 + abs(e.returncode)
else:
code = e.returncode
code = min(code, 255)
typer.echo(f"docker terminó con error (código {code})")
raise typer.Exit(code=code)
#local docker
@app.command()
def local_up():
docker_commands("up", mode="test")
#produccion docker
@app.command()
def up():
docker_commands("up", "-d", "--build")
@app.command()
def down():
docker_commands("down")
@app.command()
def stop():
docker_commands("stop")
@app.command()
def start():
docker_commands("start")
@app.command()
def logs():
docker_commands("logs", "-f")
if __name__ == "__main__":
app()
#__________________
#app/main.py
from app.start import Class_Start
from contextlib import asynccontextmanager
from fastapi import FastAPI
import logging
logger = logging.getLogger(__name__)
@asynccontextmanager
async def lifespan(app: FastAPI):
inicio = Class_Start()
logger.info("corriendo programa")
try:
await inicio.start_up()
except Exception:
logger.exception("error en start_up")
raise
try:
yield
finally:
await inicio.shut_down()
app = FastAPI(
lifespan=lifespan
)
#__________________
#app/start.py
import logging
import asyncio
from core.Sql_Data import DB #instancia
from temp_workflows.create_temporal_client import CLASS_REGISTER #Instancia
from app.socket_run import Class_Socket #clase
from core.redis_ import REDIS_CONNECT #instancia
from temp_workflows.Workflow_ import Class_workflow #clase
class Class_Start:
def __init__(self):
self.Class_Register = CLASS_REGISTER
self.socket_class = Class_Socket()
self.redis = REDIS_CONNECT
self.workflow_class = Class_workflow()
self.event = asyncio.Event()
self.logger = logging.getLogger(__name__)
async def start_up(self):
try:
await DB.create_tables()
except Exception:
self.logger.exception("error creando tablas en SQL")
raise
try:
await self.redis.connect()
except Exception:
self.logger.exception("Redis no está disponible")
raise
try:
async with asyncio.TaskGroup() as tg:
tg.create_task(self.Class_Register.register_client(self.event))
tg.create_task(self.socket_class.socket_server_start(self.redis.r_connect))
tg.create_task(self.workflow_class.workflow_disparador(self.redis.r_connect, self.event))
except* Exception as eg:
for exc in eg.exceptions:
self.logger.error("error inesperado",exc_info=(type(exc), exc, exc.__traceback__))
raise
async def shut_down(self):
try:
await self.socket_class.stop()
except Exception:
self.logger.exception("error cerrando servidor de sockets")
try:
await DB.close()
except Exception:
self.logger.exception("error cerrando conexión SQL")
try:
await self.redis.close_redis()
except Exception:
self.logger.exception("error cerrando conexión Redis")
#__________________
#app/socket_run.py
import socket
import asyncio
import ipaddress
import re
from functools import partial
import logging
import redis.exceptions as redis_exceptions
from core.config import(
SERVER_HOST,
SERVER_PORT,
SOCKET_TIMEOUT,
MAX_MESSAGE_SIZE,
DOMINIO_PROHIBIDO,
TIMEOUT_CONEXIONES_ACTIVAS_START_SERVER,
SEMAFORO
)
PATRON_DOMINIO = re.compile(
r'^(?!-)[A-Za-z0-9-]{1,63}(?<!-)(\.[A-Za-z0-9-]{1,63}(?<!-))*\.[A-Za-z]{2,63}$'
)
class Class_Socket:
def __init__(self):
self.semaforo = None
self.logger = logging.getLogger(__name__)
self.dominios_prohibidos = {d.lower().strip() for d in DOMINIO_PROHIBIDO}
self.conexiones_activas: set[asyncio.Task] = set()
self.server: asyncio.AbstractServer | None = None
async def socket_server_start(self, _redis_queue):
self.semaforo = asyncio.Semaphore(SEMAFORO)
try:
self.server = await asyncio.start_server(
partial(self.on_new_connection, _redis=_redis_queue),
SERVER_HOST,
SERVER_PORT
)
async with server:
await server.serve_forever()
except PermissionError:
self.logger.warning("Sin permisos para abrir el puerto")
raise
except socket.gaierror:
self.logger.warning("Host inválido")
raise
except OSError:
self.logger.exception("Error del sistema")
raise
except asyncio.CancelledError:
self.logger.info("servidor cancelado: %s/%s", SERVER_HOST, SERVER_PORT)
raise
except Exception:
self.logger.exception("Error inesperado")
raise
async def on_new_connection(self, reader, writer, _redis):
if self.semaforo.locked():
writer.close()
await writer.wait_closed()
return
task = asyncio.current_task()
self.conexiones_activas.add(task)
try:
async with self.semaforo:
await self.handle_client(reader, writer, _redis)
finally:
self.conexiones_activas.discard(task)
async def stop(self):
if self.server is not None:
self.server.close()
await self.server.wait_closed()
self.server = None
await self.cerrar_conexiones_activas()
async def cerrar_conexiones_activas(self):
if not self.conexiones_activas:
return
self.logger.info("esperando cierre de %d conexiones activas", len(self.conexiones_activas))
pendientes = list(self.conexiones_activas)
_, aun_pendientes = await asyncio.wait(
pendientes,
timeout=TIMEOUT_CONEXIONES_ACTIVAS_START_SERVER
)
if aun_pendientes:
self.logger.warning("cancelando %d conexiones que no cerraron a tiempo", len(aun_pendientes))
for tarea in aun_pendientes:
tarea.cancel()
await asyncio.gather(*aun_pendientes, return_exceptions=True)
async def handle_client(self, reader, writer, _redis):
peername = writer.get_extra_info("peername")
self.logger.info("informacion del host: %s", peername)
try:
header = await asyncio.wait_for(
reader.readexactly(8),
timeout=SOCKET_TIMEOUT
)
try:
longitud_header = int(header.decode().strip())
except (ValueError, UnicodeDecodeError):
self.logger.warning("header inválido, no se pudo interpretar como longitud: %r", header)
await self.enviar_respuesta(writer, "header invalido")
return
if longitud_header <= 0 or longitud_header > MAX_MESSAGE_SIZE:
self.logger.warning("longitud bytes del mensaje incorrecta: %d, longitud sugerida: %d", longitud_header ,MAX_MESSAGE_SIZE)
await self.enviar_respuesta(writer, "longitud incorrecta")
return
data = await asyncio.wait_for(
reader.readexactly(longitud_header),
timeout=SOCKET_TIMEOUT
)
try:
dominio_data = data.decode().strip()
except UnicodeDecodeError:
self.logger.warning("datos recibidos no son UTF-8 valido")
await self.enviar_respuesta(writer, "encoding invalido")
return
if not self.verificar_dominio(dominio_data):
self.logger.warning("dominio prohibido rechazado: %s", dominio_data)
await self.enviar_respuesta(writer, "dominio prohibido")
return
try:
await _redis.xadd("cola:dominios", {"dominio": dominio_data})
except redis_exceptions.ConnectionError as e:
self.logger.warning("pool de redis agotado o conexion caida: %s", e)
await self.enviar_respuesta(writer, "servicio no disponible, intenta mas tarde")
return
await self.enviar_respuesta(writer, f"dominio recibido correctamente: {dominio_data}")
except ConnectionAbortedError as e:
self.logger.warning("Conexión abortada: %s", e)
except BrokenPipeError as e:
self.logger.warning("Broken pipe: %s", e)
except (asyncio.TimeoutError, TimeoutError):
self.logger.warning("tiempo de espera agotado en reader del host")
except asyncio.IncompleteReadError as e:
self.logger.warning("conexión cerrada antes de recibir los 8 bytes esperados: %s", e)
except ConnectionResetError as e:
self.logger.warning("conexión reiniciada por el cliente: %s", e)
except asyncio.CancelledError:
self.logger.info("conexion cancelada")
raise
except Exception:
self.logger.exception("error inesperado leyendo del socket")
finally:
try:
writer.close()
await writer.wait_closed()
except Exception:
self.logger.exception("error al cerrar conexion")
async def enviar_respuesta(self, writer, mensaje: str):
try:
writer.write(f"{mensaje}\n".encode())
await writer.drain()
except (ConnectionResetError, BrokenPipeError) as e:
self.logger.warning("no se pudo enviar respuesta al cliente: %s", e)
def verificar_dominio(self, data_dom: str) -> bool:
dominio = data_dom.lower().strip()
if self.is_ip(dominio):
self.logger.warning("se envio ip en lugar de dominio: %s", dominio)
return False
if not self.dominio_valido(dominio):
self.logger.warning("el formato del dominio es invalido: %s", dominio)
return False
if dominio in self.dominios_prohibidos:
return False
for prohibido in self.dominios_prohibidos:
if dominio.endswith(f".{prohibido}"):
return False
return True
def is_ip(self, dominio: str) -> bool:
try:
ipaddress.ip_address(dominio)
return True
except ValueError:
return False
def dominio_valido(self, dominio: str) -> bool:
if len(dominio) > 253:
return False
return bool(PATRON_DOMINIO.match(dominio))
#__________________
#core/redis_.py
import redis.asyncio as redis
from redis.asyncio import ConnectionPool
import logging
from core.config import REDIS_HOST, REDIS_PORT, REDIS_DB, REDIS_PASSWORD, REDIS_MAX_CONNECTIONS
class Class_Redis:
def __init__(self):
self.pool = None
self.r_connect = None
self.logger = logging.getLogger(__name__)
async def connect(self):
self.pool = ConnectionPool(
host=REDIS_HOST,
password=REDIS_PASSWORD,
port=REDIS_PORT,
db=REDIS_DB,
decode_responses=True,
max_connections=REDIS_MAX_CONNECTIONS,
)
self.r_connect = redis.Redis(connection_pool=self.pool)
await self.r_connect.ping() # verifica conexión aquí
async def close_redis(self):
if self.r_connect is not None:
try:
await self.r_connect.aclose()
except Exception:
self.logger.exception("error al cerrar conexion del pool redis")
if self.pool is not None:
try:
await self.pool.disconnect()
except Exception:
self.logger.exception("error al cerrar el pool de conexiones redis")
REDIS_CONNECT = Class_Redis()
#__________________
#temp_workflows/create_temporal_client.py
from temporalio.client import Client
from temporalio.worker import Worker
from temporalio.service import RPCError
from core.config import TEMPORAL_HOST, TASK_QUEUE, TEMPORAL_NAMESPACE
from temporalio.api.workflowservice.v1 import DescribeTaskQueueRequest
from temp_workflows.ReconWorkflow_ import ReconWorkflow
from temp_workflows.activity import sample_activity
import logging
import asyncio
class Class_Register:
def __init__(self):
self.client = None
self.logger = logging.getLogger(__name__)
self.Healthy_attemp = 5
async def Init_client(self):
try:
self.client = await Client.connect(TEMPORAL_HOST, namespace=TEMPORAL_NAMESPACE)
self.logger.info("Cliente conectado a Temporal en %s", TEMPORAL_HOST)
except RPCError as e:
self.logger.error("Temporal rechazó la conexión: %s", e)
raise
except ValueError as e:
self.logger.error("TEMPORAL_HOST mal configurado: %s", e)
raise
except (ConnectionRefusedError, OSError) as e:
self.logger.error("No se pudo alcanzar el host de Temporal: %s", e)
raise
except TimeoutError as e:
self.logger.error("Timeout al conectar con Temporal: %s", e)
raise
except Exception:
self.logger.exception("error inesperado al conectar con el cliente")
raise
async def register_client(self, event):
if self.client is None:
await self.Init_client()
while True:
reconectar = False
try:
worker_task = asyncio.create_task(self.register_worker())
await asyncio.sleep(5)
if worker_task.done():
self.logger.error("worker falló al iniciar: %s", worker_task.exception())
reconectar = True
else:
resp = await self.client.workflow_service.describe_task_queue(
DescribeTaskQueueRequest(namespace=TEMPORAL_NAMESPACE, task_queue={"name": TASK_QUEUE}))
if resp.pollers:
self.logger.info("worker confirmado activo")
event.set()
await worker_task # se queda aquí mientras el worker viva
except RPCError as e:
self.logger.error(e)
reconectar = True
except ValueError as e:
self.logger.error("configuración inválida, no se puede continuar: %s", e)
raise
except asyncio.CancelledError:
self.logger.info("cliente temporal cancelado")
raise
except Exception:
self.logger.exception("error inesperado")
reconectar = True
if reconectar:
await self._close_client()
await asyncio.sleep(5)
await self.Init_client()
async def _close_client(self):
if self.client is None:
return
try:
if hasattr(self.client, "close"):
await self.client.close()
except Exception as e:
self.logger.warning("error al cerrar cliente temporal (ignorado): %s", e)
finally:
self.client = None
async def register_worker(self):
try:
n_worker = Worker(
self.client,
task_queue=TASK_QUEUE,
workflows=[ReconWorkflow],
activities=[sample_activity]
)
await n_worker.run()
except ValueError as e:
self.logger.error("Configuración inválida del worker (falta decorador @activity.defn o @workflow.defn?): %s", e)
self.logger.info("worker.run dejo de correr")
raise
except RPCError as e:
self.logger.error("Worker perdió conexión con el servidor de Temporal: %s", e)
self.logger.info("worker.run dejo de correr, se reintentara")
raise
except Exception as e:
self.logger.exception("error inesperado: %s", e)
self.logger.info("worker.run dejo de correr, se reintentara")
raise
CLASS_REGISTER = Class_Register()
#__________________
#temp_workflows/Workflow_.py
import logging
import asyncio
import uuid
from datetime import timedelta
from temporalio.common import RetryPolicy
from temp_workflows.ReconWorkflow_ import ReconWorkflow
from core.Sql_Data import DB
from temporalio.client import Client
from temporalio.service import RPCError
from temporalio.exceptions import WorkflowFailureError
from core.config import TEMPORAL_HOST, TASK_QUEUE, TEMPORAL_NAMESPACE
from redis.exceptions import ResponseError
class Class_workflow:
def __init__(self):
self.client = None
self.logger = logging.getLogger(__name__)
self.max_workflow_attemp = 3
async def connect_client_workflow(self):
try:
self.client = await Client.connect(TEMPORAL_HOST, namespace=TEMPORAL_NAMESPACE)
self.logger.info("Cliente conectado a Temporal en %s", TEMPORAL_HOST)
except RPCError as e:
self.logger.error("Temporal rechazó la conexión: %s", e)
raise
except ValueError as e:
self.logger.error("TEMPORAL_HOST mal configurado: %s", e)
raise
except (ConnectionRefusedError, OSError) as e:
self.logger.error("No se pudo alcanzar el host de Temporal: %s", e)
raise
except TimeoutError as e:
self.logger.error("Timeout al conectar con Temporal: %s", e)
raise
except Exception:
self.logger.exception("error inesperado al conectar con el cliente")
raise
async def asegurar_grupo(self, redis, stream="cola:dominios", grupo="workers"):
try:
await redis.xgroup_create(name=stream, groupname=grupo, id="0", mkstream=True)
self.logger.info("grupo de consumidores '%s' creado", grupo)
except ResponseError as e:
if "BUSYGROUP" in str(e):
self.logger.info("grupo de consumidores '%s' ya existe", grupo)
else:
self.logger.error("error creando grupo de consumidores: %s", e)
raise
async def workflow_disparador(self, redis, event):
await event.wait()
await self.asegurar_grupo(redis)
while True:
if self.client is None:
try:
await self.connect_client_workflow()
except Exception:
self.logger.warning("no se pudo conectar, reintentando en 5s")
await asyncio.sleep(5)
continue
try:
messages = await redis.xreadgroup(
groupname="workers",
consumername="worker-1",
streams={"cola:dominios": ">"},
count=1,
block=0
)
except Exception as e:
self.logger.error("error al leer de redis: %s", e)
await asyncio.sleep(5)
continue
if not messages:
self.logger.warning("xreadgroup no devolvió mensajes, reintentando")
continue
stream_name, entries = messages[0]
if not entries:
self.logger.warning("stream sin entradas, reintentando")
continue
message_id, data = entries[0]
dominio_data = data["dominio"]
intentos = 0
while True:
error_except = None
try:
handle = await self.client.start_workflow(
ReconWorkflow.run,
dominio_data,
id=f"mainworkflow-{uuid.uuid4()}",
task_queue=TASK_QUEUE,
execution_timeout=timedelta(minutes=30),
retry_policy=RetryPolicy(
maximum_attempts=5,
initial_interval=timedelta(seconds=10))
)
resultado = await handle.result()
await redis.xack(
"cola:dominios",
"workers",
message_id
)
self.logger.info("workflow arrancado para %s", dominio_data)
self.logger.info("handle del workflow: %s", handle.id)
break
except RPCError as e:
self.logger.error("no se pudo comunicar con el servidor temporal: %s", e)
error_except = e
intentos += 1
if self.client is not None:
try:
await self.client.close()
except Exception as close_err:
self.logger.warning("no se pudo cerrar la conexión: %s", close_err)
finally:
self.client = None
except TimeoutError as e:
self.logger.error("el workflow para %s excedió el tiempo máximo: %s", dominio_data, e)
intentos += 1
error_except = e
except asyncio.CancelledError as e:
self.logger.info("procesamiento de dominio cancelado")
try:
await asyncio.shield(
DB.insert_dead_letter(str(e), intentos, "guardando datos cancelled")
)
except Exception:
self.logger.exception("error al guardar dato en SQL")
raise
except Exception as e:
self.logger.exception("error al ejecutar workflow para %s: %s", dominio_data, e)
intentos += 1
error_except = e
if intentos >= self.max_workflow_attemp:
try:
await DB.insert_dead_letter(str(error_except) if error_except else "error desconocido", intentos, dominio_data)
except Exception:
self.logger.warning("no se pudo guardar datos de error")
self.logger.warning("cantidad de intentos superada: %d/%d", intentos, self.max_workflow_attemp)
await redis.xadd("cola:dominios", {"dominio": dominio_data})
break
if intentos < self.max_workflow_attemp:
await asyncio.sleep(5)
#__________________
#temp_workflows/ReconWorkflow_.py
import logging
import uuid
from temp_workflows.activity import sample_activity
from temporalio import workflow
from datetime import timedelta
from temporalio.common import RetryPolicy
from temporalio import activity
from core.config import TASK_QUEUE
@workflow.defn
class ReconWorkflow:
@workflow.run
async def run(self, dominio_data: str) -> dict:
return await workflow.execute_activity(
sample_activity,
dominio_data,
schedule_to_close_timeout=timedelta(minutes=25),
schedule_to_start_timeout=timedelta(minutes=5),
start_to_close_timeout=timedelta(minutes=20),
heartbeat_timeout=timedelta(seconds=30),
retry_policy=RetryPolicy(maximum_attempts=5),
cancellation_type=activity.ActivityCancellationType.TRY_CANCEL,
task_queue=TASK_QUEUE,
activity_id=f"graph-{uuid.uuid4()}",
versioning_intent=None,
summary="ejecución del grafo que contiene las herramientas"
)
#__________________
#temp_workflows/activity.py
from temporalio import activity
from graph.graph_start import graph_ainvoke
import logging
logger = logging.getLogger(__name__)
@activity.defn
async def sample_activity(dominio_data: str):
try:
result = await graph_ainvoke(dominio_data)
return result
except Exception:
logger.exception("error con graph_start")
raise
#__________________
#graph/graph_start.py
from graph.graph_init_compile import grafo_compilado
import logging
import asyncio
logger = logging.getLogger(__name__)
async def graph_ainvoke(dominio_data):
platilla = {
"dominio": dominio_data,
"subdominios": {},
"dominio_ips": {},
"resultado_nmap": {},
"resultado_nuclei": {},
"resultado_filtro": {},
"resultado_hydra": {},
"resultado_final": {}
}
try:
resultado = await grafo_compilado.ainvoke(platilla)
return resultado
except KeyError as e:
logger.error("campo faltante en el estado, revisa el orden de los edges: %s", e)
raise
except asyncio.TimeoutError:
logger.error("el pipeline para %s excedió el tiempo máximo", dominio_data)
raise
except asyncio.CancelledError:
logger.info("ejecución del grafo cancelada")
raise
except Exception as e:
logger.exception("error inesperado ejecutando el pipeline para %s: %s", dominio_data, e)
raise
#__________________
#graph/graph_init_compile.py
from graph.graph_config import config_grafo
grafo_compilado = config_grafo()
#__________________
#graph/graph_config.py
from langgraph.graph import StateGraph, END
from typing import TypedDict
import logging
from tools._amass import analizer_amass
from tools._resolver_dns import DNS_resolve
from tools._nmap import analizer_nmap
from tools._nuclei import analizer_nuclei
from tools._filtro import analizer_filtro
from tools._hydra import analizer_hydra
resolver = DNS_resolve()
logger = logging.getLogger(__name__)
class State(TypedDict):
dominio: str
subdominios: dict[str, list[str]]
dominio_ips: dict[str, str]
resultado_nmap: dict[str, list[str]]
resultado_nuclei: dict[str, dict]
resultado_filtro: dict[str, list[str]]
class PartialState(TypedDict, total=False):
subdominios: dict[str, list[str]]
dominio_ips: dict[str, str]
resultado_nmap: dict[str, list[str]]
resultado_nuclei: dict[str, dict]
resultado_filtro: dict[str, list[str]]
async def amass(state: State) -> PartialState:
dominio = state.get("dominio", "desconocido")
try:
resultado = await analizer_amass(dominio)
return {"subdominios": resultado}
except Exception as e:
logger.exception("fallo crítico en amass para %s: %s", dominio, e)
raise
async def resolver_dns(state: State) -> PartialState:
dominio = state.get("dominio", "desconocido")
try:
resultado = await resolver.analizer_resolver_dns(state["subdominios"])
return {"dominio_ips": resultado}
except Exception as e:
logger.exception("error critico en resolver DNS para %s: %s", dominio, e)
raise
async def nmap(state: State) -> PartialState:
dominio = state.get("dominio", "desconocido")
try:
resultado = await analizer_nmap(state["dominio_ips"])
return {"resultado_nmap": resultado}
except Exception as e:
logger.exception("error en analisis de nmap para %s: %s", dominio, e)
raise
async def nuclei(state: State) -> PartialState:
dominio = state.get("dominio", "desconocido")
try:
resultado = await analizer_nuclei(state["dominio_ips"])
return {"resultado_nuclei": resultado}
except Exception as e:
logger.exception("error en analisis de nuclei para %s: %s", dominio, e)
raise
async def filtro(state: State) -> PartialState:
dominio = state.get("dominio", "desconocido")
try:
resultados_nmap = state.get("resultado_nmap")
resultados_nuclei = state.get("resultado_nuclei")
#recien agregado
dominios = state.get("dominio_ips")
if resultados_nmap is None or resultados_nuclei is None:
logger.warning("filtro llamado antes de tener los 2 resultados para: %s", dominio)
raise ValueError(f"faltan resultados de nmap/nuclei para completar filtro en {dominio}")
resultado = await analizer_filtro(resultados_nmap, resultados_nuclei, dominios)
return {"resultado_filtro": resultado}
except Exception as e:
logger.exception("error en analisis para nuclei en %s: %s", dominio, e)
raise
def config_grafo():
grafo = StateGraph(State)
grafo.add_node("amass", amass)
grafo.add_node("resolver_dns", resolver_dns)
grafo.add_node("nmap", nmap)
grafo.add_node("nuclei", nuclei)
grafo.add_node("filtro", filtro)
grafo.set_entry_point("amass")
grafo.add_edge("amass", "resolver_dns")
grafo.add_edge("resolver_dns", "nmap")
grafo.add_edge("resolver_dns", "nuclei")
grafo.add_edge("nmap", "filtro")
grafo.add_edge("nuclei", "filtro")
grafo.add_edge("filtro", END)
try:
return grafo.compile()
except Exception:
logger.critical("no se pudo compilar el grafo")
raise
#__________________
#tools/_amass.py
import asyncio
import json
import logging
from pathlib import Path
from core.config import PATH_AMASS
logger = logging.getLogger(__name__)
async def analizer_amass(dominio_url: str) -> dict:
tmp = Path("/tmp")
tmp.mkdir(exist_ok=True)
salida_json = tmp / f"amass_{dominio_url}.json"
try:
resultado = await asyncio.create_subprocess_exec(
PATH_AMASS, "enum",
"-d", dominio_url,
"-timeout", "20",
"-json", salida_json,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
try:
stdout, stderr = await asyncio.wait_for(
resultado.communicate(),
timeout=1300
)
except asyncio.TimeoutError:
logger.error("amass excedio el tiempo limite de python")
resultado.kill()
await resultado.wait()
raise
if resultado.returncode != 0:
raise RuntimeError(f"error en amass {dominio_url}: {stderr.decode()}")
subdominios_con_ip = {}
subdominios_sin_ip = {}
try:
with open(salida_json) as f:
for linea in f:
if linea.strip():
dato = json.loads(linea)
nombre = dato["name"]
direccion = dato.get("addresses", [])
if direccion:
ip = [addr["ip"] for addr in direccion]
subdominios_con_ip[nombre] = ip
else:
subdominios_sin_ip[nombre] = []
except FileNotFoundError:
logger.error("amass no encontro nada o fallo al escribir el archivo")
raise
finally:
try:
salida_json.unlink()
except FileNotFoundError:
pass
except OSError as e:
logger.warning("no se pudo borrar el archivo temporal %s: %s", salida_json, e)
except FileNotFoundError:
logger.error("amass no está instalado o no se encuentra en el PATH")
raise
except PermissionError:
logger.error("sin permisos para ejecutar amass")
raise
except OSError as e:
logger.error("error del sistema al lanzar amass: %s", e)
raise
except ValueError:
logger.error("parametros del comando incorrecta")
raise
return {
"con_ip": subdominios_con_ip,
"sin_ip": subdominios_sin_ip
}
#__________________
#tools/_resolver_dns.py
import logging
import dns.asyncresolver
import dns.asyncresolver
import dns.resolver
import dns.exception
import asyncio
class DNS_resolve:
def __init__(self):
self.logger = logging.getLogger(__name__)
self.semaforo = asyncio.Semaphore(20)
async def analizer_resolver_dns(self, datos_DNS: dict) -> dict:
con_ip = datos_DNS.get("con_ip", {})
sin_ip = datos_DNS.get("sin_ip", {})
dict_final = {}
lista_dominios_sin_ip = []
if not con_ip and not sin_ip:
self.logger.error("no se encontraron datos en los resultados de amass")
raise ValueError("resultados de amass sin datos")
for tipo, dict_dominios in datos_DNS.items():
for dominio, ips in dict_dominios.items():
if ips:
dict_final[dominio] = ips
else:
lista_dominios_sin_ip.append(dominio)
tarea = [self.resolver_dominio(dominio) for dominio in lista_dominios_sin_ip]
resultado = await asyncio.gather(*tarea)
for tupla in resultado:
if not tupla:
continue
sub, ips = tupla
dict_final[sub] = ips
return dict_final
async def resolver_dominio(self, dominio: str) -> tuple:
resolver = dns.asyncresolver.Resolver()
resolver.timeout = 5
resolver.lifetime = 10
async with self.semaforo:
try:
respuesta = await resolver.resolve(dominio, "A")
lista = [str(ip) for ip in respuesta]
return dominio, lista
except dns.resolver.NXDOMAIN:
self.logger.info("%s no existe (NXDOMAIN)", dominio)
return ()
except dns.resolver.NoAnswer:
self.logger.info("%s no tiene registro A", dominio)
return ()
except dns.resolver.LifetimeTimeout:
self.logger.warning("timeout resolviendo %s", dominio)
return ()
except dns.exception.DNSException as e:
self.logger.warning("error DNS para %s: %s", dominio, e)
return ()
#__________________
#tools/_nmap.py
import subprocess
import logging
import asyncio
from pathlib import Path
import tempfile
from collections import defaultdict
from xmltodict import ExpatError
import json
import xmltodict
from core.Sql_Data import DB, Socket_nmap
from core.config import PATH_NMAP, NMAP_TIMEOUT
logger = logging.getLogger(__name__)
db = DB
async def analizer_nmap(dominios_y_ip: dict) -> dict:
dict_resultados = defaultdict(dict)
max_intentos = 3
if not dominios_y_ip:
raise ValueError("datos de entrada para nmap vacíos")
folder_temp = Path(tempfile.gettempdir())
for _, dict_1 in dominios_y_ip.items():
for dominio, lista_ip in dict_1.items():
for ip in lista_ip:
file_nmap = folder_temp / f"nmap_{ip}.xml"
intento = 0
bucle = True
while bucle:
intento += 1
try:
resultado = await asyncio.create_subprocess_exec(
PATH_NMAP, ip, "-oX", file_nmap,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await asyncio.wait_for(
resultado.communicate(),
timeout= NMAP_TIMEOUT
)
if resultado.returncode != 0:
error_msg = stderr.decode() if stderr else "Código de retorno no cero"
raise RuntimeError(f"nmap falló: {error_msg}")
if stderr:
logger.warning("nmap emitió advertencias: %s", stderr.decode())
parser = parsear_salida(file_nmap)
logger.info("Dominio %s - IP %s escaneado correctamente", dominio, ip)
try:
await db.Upsert(Socket_nmap, parser, ip)
logger.info("guardado correctamente")
except Exception:
logger.error("Error al guardar datos para %s - %s", dominio, ip)
dict_resultados[dominio][ip] = parser
break
except FileNotFoundError:
logger.error("Nmap no está instalado o no está en PATH")
raise
except TimeoutError:
logger.warning("Timeout escaneando %s, reintento %d/%d", ip, intento, max_intentos)
if intento >= max_intentos:
bucle = False
except (PermissionError, OSError, ValueError, TypeError) as e:
logger.error(f"Error no recuperable: {e}")
bucle = False
except asyncio.CancelledError:
logger.error("La operación fue cancelada")
try:
if file_nmap.exists():
file_nmap.unlink()
except Exception as cleanup_error:
logger.error("Error en limpieza: %s", cleanup_error)
raise
except Exception:
logger.exception("error inesperado")
bucle = False
return dict(dict_resultados)
def parsear_salida(ruta):
if not ruta.exists():
logger.error("archivo no existe: %s", ruta)
raise FileNotFoundError(f"Archivo no encontrado: {ruta}")
try:
with open(ruta, "r") as f:
data = xmltodict.parse(f.read())
if not data or data == {}:
raise ValueError(f"archivo xml vacio: {ruta}")
if not data or 'nmap' not in data:
raise ValueError(f"XML sin estructura válida: {ruta}")
return data
finally:
try:
if ruta.exists():
ruta.unlink()
logger.info("Archivo temporal eliminado: %s", ruta)
else:
logger.info("el archivo que intenta eliminar no existe: %s", ruta)
except Exception as e:
logger.error("no se pudo eliminar el archivo %s: %s", ruta, e)
#__________________
#tools/_nuclei.py
import logging
import subprocess
import asyncio
import json
import tempfile
from pathlib import Path
from collections import defaultdict
from core.Sql_Data import DB, Nuclei
from core.config import PATH_NUCLEI, NUCLEI_TIMEOUT
logger = logging.getLogger(__name__)
async def analizer_nuclei(dominio_ips: dict) -> dict:
dict_final = defaultdict(dict)
max_intentos = 3
if not dominio_ips:
logger.warning("entrada con datos vacios")
raise ValueError("datos recibidos por resolver_dns vacios")
folder_file = tempfile.gettempdir()
#URLS = [f"https://{dominio}" if not dominio.startswith("https://") else dominio for dominio in dominio_ips.keys()]
for dominio in dominio_ips.keys():
url = f"https://{dominio}" if not dominio.startswith("https://") else dominio
intento = 0
nuclei_file = Path(f"{folder_file}/{dominio}.json")
bucle = True
while bucle:
intento += 1
try:
resultado = await asyncio.create_subprocess_exec(
PATH_NUCLEI, "-u", dominio, "-json-export", nuclei_file,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await asyncio.wait_for(
resultado.communicate(),
timeout=NUCLEI_TIMEOUT
)
if resultado.returncode != 0:
msge = stderr.decode() if stderr else "codigo de retorno no es 0"
raise RuntimeError("error en la salida de subprocess: %s", msge)
if stderr:
logger.warning("nuclei emitió advertencias: %s", stderr.decode())
parser = await parsear_salida(nuclei_file, dominio)
try:
await DB.Upsert(Nuclei, parser, dominio)
logger.info("datos de %s guardado correctamente", dominio)
except Exception as e:
logger.error("error al guardar datos de %s: %s", dominio, e)
dict_final[dominio] = parser
break
except FileNotFoundError:
logger.error("Nuclei no está instalado o no está en PATH")
raise
except TimeoutError:
logger.warning("Timeout escaneando %s, reintento %d/%d", dominio, intento, max_intentos)
if intento >= max_intentos:
bucle = False
except (PermissionError, OSError, ValueError, TypeError) as e:
logger.error(f"Error no recuperable: {e}")
bucle = False
except asyncio.CancelledError:
logger.error("La operación fue cancelada")
try:
if nuclei_file.exists():
nuclei_file.unlink()
except Exception as cleanup_error:
logger.error("Error en limpieza: %s", cleanup_error)
raise
except Exception:
logger.exception("error inesperado")
bucle = False
return dict(dict_final)
async def parsear_salida(ruta, dominio):
if not ruta.exists():
logger.warning("no esxite archivo nuclei de %s del dominio: %s", ruta, dominio)
raise FileNotFoundError(f"archivo nuclei {ruta} no encontrado del dominio: {dominio}")
try:
with open(ruta, "r") as f:
contenido = f.read()
if not contenido.strip():
logger.warning("el archivo %s está vacío", ruta)
raise ValueError(f"archivo {ruta} sin datos para procesar")
data = json.loads(contenido)
if not data:
logger.warning("el archivo %s no contiene datos útiles tras parsear", ruta)
raise ValueError(f"archivo {ruta} sin datos útiles")
return data
finally:
try:
if ruta.exists():
ruta.unlink()
logger.info("archivo temporal %s eliminado", ruta)
except Exception as e:
logger.warning("no se pudo eliminar el archivo %s: %s", ruta, e)
#__________________
#tools/_filtro.py
from urllib.parse import urlparse
import logging
from core.Sql_Data import Salida_final, DB
import json
logger = logging.getLogger(__name__)
async def analizer_filtro(resultado_nmap: dict, resultado_nuclei: dict, dominios: dict) -> dict:
formato_dict = {}
for d in dominios.keys():
formato_dict[d] = {
"ip": "",
"data": [],
"url_login": [],
"descriptions": [],
}
nmap_data = {}
puertos = []
try:
nmap_data = resultado_nmap.get(d) or {}
if not nmap_data:
logger.info("datos nmap de %s vacios", d)
else:
address = nmap_data.get("address") or {}
if isinstance(address, dict):
address = [address]
ipv4 = next((a for a in address if a.get("@addrtype") == "ipv4"), None)
if ipv4 is None:
ipv4 = address[0] if address else {}
formato_dict[d]["ip"] = ipv4.get("@addr", "")
puertos = (nmap_data.get("ports") or {}).get("port") or []
if isinstance(puertos, dict):
puertos = [puertos]
elif not isinstance(puertos, list):
logger.warning("formato de puertos inesperado en %s: %r", d, puertos)
puertos = []
except (AttributeError, TypeError, KeyError, IndexError) as e:
logger.warning("error procesando IP de %s: %s", d, e)
nmap_data, puertos = {}, []
except Exception as e:
logger.exception("error inesperado procesando nmap de %s: %s", d, e)
nmap_data, puertos = {}, []
for port in puertos:
try:
if not port:
logger.info("datos de puertos de %s vacios", d)
continue
estado = (port.get("state") or {}).get("@state")
if estado != "open":
logger.debug("puerto %s/%s de %s no está abierto (%s)", port.get("@portid"), port.get("@protocol"), d, estado)
continue
service = port.get("service") or {}
formato_dict[d]["data"].append({
"puerto": port.get("@portid"),
"protocolo": port.get("@protocol"),
"servicio": service.get("@name"),
"tecnologia": service.get("@product"),
})
except (AttributeError, TypeError, KeyError, IndexError) as e:
logger.warning("error procesando %s: %s", d, e)
except Exception as e:
logger.exception("error inesperado, %s, %s", d, e)
try:
nuclei_data = resultado_nuclei.get(d) or []
if isinstance(nuclei_data, dict):
nuclei_data = [nuclei_data]
elif not isinstance(nuclei_data, list):
logger.warning("formato de nuclei inesperado en %s: %r", d, nuclei_data)
nuclei_data = []
descriptions = []
for item in nuclei_data:
try:
desc = (item.get("info") or {}).get("description")
except AttributeError:
logger.warning("item de nuclei inválido en %s", d)
continue
if desc:
descriptions.append(desc)
formato_dict[d]["descriptions"] = list(dict.fromkeys(descriptions))
for item in nuclei_data:
try:
url = item.get("matched-at")
except AttributeError:
logger.warning("item de nuclei inválido en %s", d)
continue
if not url:
logger.debug("item de nuclei en %s sin 'matched-at', se omite", d)
continue
path = urlparse(url).path.lower()
if (any(ext in path for ext in [".php", ".asp", ".jsp"]) or
any(seg in path for seg in ["login", "admin", "xmlrpc"])):
formato_dict[d]["url_login"].append(url)
try:
await DB.Upsert(Salida_final, formato_dict[d], d)
logger.info("datos de %s guardado correctamente", d)
except Exception as e:
logger.error("error al guardar datos de %s: %s", d, e)
formato_dict[d]["url_login"] = list(dict.fromkeys(formato_dict[d]["url_login"]))
except (AttributeError, TypeError, KeyError, IndexError) as e:
logger.warning("error procesando %s: %s", d, e)
except Exception as e:
logger.exception("error inesperado, %s, %s", d, e)
#aqui crear el json
with open("datos.json", "w", encoding="utf-8") as f:
json.dump(formato_dict, f, ensure_ascii=False, indent=4)
return formato_dict
#__________________
#core/Sql_Data.py
import logging
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import JSON, DateTime, Integer, String
from sqlalchemy.dialects.postgresql import insert as pg_insert
from datetime import datetime, timezone
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.exc import IntegrityError, OperationalError, ProgrammingError
from core.config import DB_URL, SQL_ECHO, MAX_WORKERS, MAX_OVERFLOWS
logger = logging.getLogger(__name__)
class Base(DeclarativeBase):
pass
#tabla nmap
class Socket_nmap(Base):
__tablename__ = "nmap_socket"
id: Mapped[int] = mapped_column(primary_key=True)
result: Mapped[dict] = mapped_column(JSON)
domain: Mapped[str] = mapped_column(String, unique=True, index=True)
#tabla filtro final.
class Salida_final(Base):
__tablename__ = "salida_final_dominio"
id: Mapped[int] = mapped_column(primary_key=True)
result: Mapped[dict] = mapped_column(JSON)
domain: Mapped[str] = mapped_column(String, unique=True, index=True)
#tablas nuclei
class Nuclei(Base):
__tablename__ = "nuclei"
id: Mapped[int] = mapped_column(primary_key=True)
result: Mapped[dict] = mapped_column(JSON)
domain: Mapped[str] = mapped_column(String, unique=True, index=True)
#tabla dead letter
class DeadLetter(Base):
__tablename__ = "dead_letter"
id: Mapped[int] = mapped_column(primary_key=True)
error: Mapped[str] = mapped_column(String)
intentos: Mapped[int] = mapped_column(Integer, default=0)
domain: Mapped[str] = mapped_column(String)
date: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=lambda:datetime.now(timezone.utc)
)
class Base_Sql:
def __init__(self):
logger.info("Inicializando conexión a base de datos...")
self.engine = create_async_engine(
DB_URL,
echo=ECHO,
pool_size=MAX_WORKERS,
max_overflow=MAX_OVERFLOWS,
pool_pre_ping=True
)
self._session_factory = async_sessionmaker(
self.engine,
expire_on_commit=False
)
logger.info("Conexión a base de datos inicializada")
#esto es para que cuando se necesite la sesion solo se llame la funcion por si se modifica async_sessionmaker
def get_session(self) -> AsyncSession:
"""Obtiene una sesión asíncrona."""
return self._session_factory()
async def create_tables(self):
try:
async with self.engine.begin() as conn:
await conn.run_sync(
Base.metadata.create_all
)
logger.info("Tablas creadas exitosamente")
except OperationalError as e:
logger.warning("no se pudo conectar a la base de datos %s", e)
raise
except ProgrammingError as e:
logger.warning("error de permisos o esquema al crear tablas: %s", e)
raise
except Exception as e:
logger.exception("Error inesperado creando tablas: %s", e)
raise
async def close(self):
await self.engine.dispose()
async def insert_dead_letter(self, error: str, intentos: int, domain: str):
try:
async with self._session_factory() as session:
async with session.begin():
session.add(DeadLetter(
error=error,
intentos=intentos,
domain=domain
))
logger.info("Dead letter insertada para %s", domain)
except Exception as e:
logger.error("Error insertando dead letter para %s: %s", domain, e)
raise
async def Upsert(self, model, Result, Domain):
try:
async with self._session_factory() as session:
stmt = pg_insert(model).values(result=Result, domain=Domain)
stmt = stmt.on_conflict_do_update(
index_elements=["domain"],
set_={"result": Result}
)
await session.execute(stmt)
logger.info("Datos upserted para dominio: %s", Domain)
except IntegrityError as e:
logger.error(f"Error de integridad al insertar {Domain}: {e}")
#await session.rollback()
raise
except OperationalError as e:
logger.error(f"Error operacional en BD para {Domain}: {e}")
#await session.rollback()
raise
except Exception as e:
logger.exception(f"Error inesperado en Upsert para {Domain}: {e}")
#await session.rollback()
raise
DB = Base_Sql()
#__________________
#core./config.py
from pathlib import Path
import os
from dotenv import load_dotenv
load_dotenv(Path(__file__).with_name(".env"))
def get_env(name: str, default: str | None = None) -> str:
value = os.getenv(name, default)
if value is None or value == "":
raise RuntimeError(f"Falta la variable de entorno requerida: {name}")
return value
TEMPORAL_HOST = get_env("TEMPORAL_HOST")
TEMPORAL_NAMESPACE = get_env("TEMPORAL_NAMESPACE")
TASK_QUEUE = get_env("TASK_QUEUE")
DB_URL = get_env("DB_URL")
POSTGRES_USER = get_env("POSTGRES_USER")
POSTGRES_PASSWORD = get_env("POSTGRES_PASSWORD")
REDIS_PASSWORD = get_env("REDIS_PASSWORD")
SQL_ECHO = get_env("SQL_ECHO", "False").lower() == "true"
MAX_WORKERS = int(get_env("MAX_WORKERS"))
MAX_OVERFLOWS = int(get_env("MAX_OVERFLOWS"))
SERVER_HOST = get_env("SERVER_HOST")
SERVER_PORT = int(get_env("SERVER_PORT"))
SOCKET_TIMEOUT = int(get_env("SOCKET_TIMEOUT"))
TIMEOUT_CONEXIONES_ACTIVAS_START_SERVER = int(float(get_env("TIMEOUT_CONEXIONES_ACTIVAS_START_SERVER")))
SEMAFORO = int(get_env("SEMAFORO"))
NMAP_TIMEOUT = int(get_env("NMAP_TIMEOUT"))
NUCLEI_TIMEOUT = int(get_env("NUCLEI_TIMEOUT"))
MAX_MESSAGE_SIZE = int(get_env("MAX_MESSAGE_SIZE"))
REDIS_HOST = get_env("REDIS_HOST")
REDIS_PORT = int(get_env("REDIS_PORT"))
REDIS_DB = int(get_env("REDIS_DB"))
REDIS_MAX_CONNECTIONS = int(get_env("REDIS_MAX_CONNECTIONS"))
PATH_AMASS = get_env("PATH_AMASS")
PATH_NMAP = get_env("PATH_NMAP")
PATH_NUCLEI = get_env("PATH_NUCLEI")
DOMINIO_PROHIBIDO = [
"localhost",
"example.com",
"example.org",
"example.net",
]
#__________________
# core/.env
TEMPORAL_HOST="localhost:7233"
TEMPORAL_NAMESPACE="default"
TASK_QUEUE="framework_2"
DB_URL="postgresql+asyncpg://postgres:postgres@postgres:5432/proyect_framework"
POSTGRES_USER=app_user
POSTGRES_PASSWORD="jeisonpostgres"
REDIS_PASSWORD="jeisonredis"
SQL_ECHO=False
MAX_WORKERS=10
MAX_OVERFLOWS=20
SERVER_HOST=0.0.0.0
SERVER_PORT=9000
SOCKET_TIMEOUT=120
TIMEOUT_CONEXIONES_ACTIVAS_START_SERVER=0.5
SEMAFORO=100
NMAP_TIMEOUT=120
NUCLEI_TIMEOUT=120
MAX_MESSAGE_SIZE=8192
REDIS_HOST="localhost"
REDIS_PORT=6379
REDIS_DB=0
REDIS_MAX_CONNECTIONS=50
PATH_AMASS=/usr/bin/amass
PATH_NMAP=/usr/bin/nmap
PATH_NUCLEI=/usr/bin/nuclei
#__________________
#Dockerfile
# syntax=docker/dockerfile:1
ARG AMASS_VERSION=5.1.1
ARG NUCLEI_VERSION=3.8.0
# ===================== ETAPA 1: builder =====================
FROM python:3.12-slim AS builder
ARG AMASS_VERSION
ARG NUCLEI_VERSION
RUN DEBIAN_FRONTEND=noninteractive apt-get update && \
apt-get install -y --no-install-recommends \
curl \
unzip \
ca-certificates \
&& rm -rf /var/lib/apt/lists/*
RUN mkdir -p /tools && \
curl -fsSL "https://[Log in to view URL]" \
-o /tmp/amass.tar.gz && \
curl -fsSL "https://[Log in to view URL]" \
-o /tmp/amass_checksums.txt && \
grep "amass_linux_amd64.tar.gz" /tmp/amass_checksums.txt | sed 's|amass_linux_amd64.tar.gz|/tmp/amass.tar.gz|' | sha256sum --check --status && \
mkdir -p /tmp/amass && \
tar -xzf /tmp/amass.tar.gz -C /tmp/amass && \
find /tmp/amass -type f -name amass -exec cp {} /tools/amass \; && \
rm -rf /tmp/amass.tar.gz /tmp/amass_checksums.txt /tmp/amass && \
chmod 755 /tools/amass
RUN curl -fsSL "https://[Log in to view URL]" \
-o /tmp/nuclei.zip && \
curl -fsSL "https://[Log in to view URL]" \
-o /tmp/nuclei_checksums.txt && \
grep "nuclei_${NUCLEI_VERSION}_linux_amd64.zip" /tmp/nuclei_checksums.txt | sed "s|nuclei_${NUCLEI_VERSION}_linux_amd64.zip|/tmp/nuclei.zip|" | sha256sum --check --status && \
unzip /tmp/nuclei.zip -d /tmp/nuclei && \
mv /tmp/nuclei/nuclei /tools/nuclei && \
rm -rf /tmp/nuclei.zip /tmp/nuclei_checksums.txt /tmp/nuclei && \
chmod 755 /tools/nuclei
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir --upgrade pip && \
pip install --no-cache-dir --user -r requirements.txt
# ===================== ETAPA 2: runtime =====================
FROM python:3.12-slim AS runtime
LABEL org.opencontainers.image.authors="equipo@empresa.com"
LABEL org.opencontainers.image.version="1.0"
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
PATH="/tools:/home/app/.local/bin:${PATH}"
RUN DEBIAN_FRONTEND=noninteractive apt-get update && \
apt-get install -y --no-install-recommends \
nmap \
curl \
ca-certificates \
&& rm -rf /var/lib/apt/lists/*
RUN addgroup --system app && adduser --system --group app
WORKDIR /app
# Copiamos solo lo ya listo desde la etapa builder — sin curl, unzip, tar.gz, ni caché de apt
COPY --from=builder /tools /tools
COPY --from=builder /root/.local /home/app/.local
COPY --chown=app:app . .
RUN chown -R app:app /home/app/.local
USER app
EXPOSE 8000
EXPOSE 9000
HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
CMD curl -f http://localhost:8000/health || exit 1
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
#__________________
#requirements.txt
#__________________
# .gitignore
downloads/*
!downloads/.gitkeep
.env
#__________________
#downloads
#carpeta donde van los archivos JSON
#__________________
#.github/workflows/ci-cd.yml
name: CI/CD
on:
push:
branches: [main]
pull_request:
branches: [main]
env:
IMAGE_NAME: ${{ github.repository }}
jobs:
# ===================== CI: Build + Test =====================
build-and-test:
runs-on: ubuntu-latest
steps:
- name: Checkout código
uses: actions/checkout@v4
- name: Configurar Docker Buildx
uses: docker/setup-buildx-action@v3
- name: Construir imagen (solo validación, sin publicar)
uses: docker/build-push-action@v6
with:
context: .
push: false
tags: ${{ env.IMAGE_NAME }}:test
cache-from: type=gha
cache-to: type=gha,mode=max
- name: Levantar stack para pruebas
run: docker compose -f docker-compose.yml up -d --build
env:
POSTGRES_USER: ci_user
POSTGRES_PASSWORD: ci_password_temporal
REDIS_PASSWORD: ci_redis_temporal
- name: Esperar a que la app esté healthy
run: |
for i in {1..15}; do
status=$(docker inspect --format='{{.State.Health.Status}}' $(docker compose ps -q app) 2>/dev/null || echo "starting")
if [ "$status" == "healthy" ]; then
echo "App healthy"
exit 0
fi
echo "Esperando... ($status)"
sleep 5
done
echo "La app no llegó a healthy a tiempo"
docker compose logs
exit 1
- name: Ejecutar tests
run: docker compose exec -T app pytest
- name: Apagar stack
if: always()
run: docker compose down -v
# ===================== CD: Build + Push + Deploy (solo en main) =====================
build-push-deploy:
needs: build-and-test
if: github.ref == 'refs/heads/main' && github.event_name == 'push'
runs-on: ubuntu-latest
steps:
- name: Checkout código
uses: actions/checkout@v4
- name: Login al registry
uses: docker/login-action@v3
with:
registry: ghcr.io
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- name: Configurar Docker Buildx
uses: docker/setup-buildx-action@v3
- name: Construir y publicar imagen
uses: docker/build-push-action@v6
with:
context: .
push: true
tags: |
ghcr.io/${{ env.IMAGE_NAME }}:latest
ghcr.io/${{ env.IMAGE_NAME }}:${{ github.sha }}
cache-from: type=gha
cache-to: type=gha,mode=max
- name: Desplegar en el servidor por SSH
uses: appleboy/ssh-action@v1.0.3
with:
host: ${{ secrets.DEPLOY_HOST }}
username: ${{ secrets.DEPLOY_USER }}
key: ${{ secrets.DEPLOY_SSH_KEY }}
script: |
cd /ruta/a/tu/proyecto #aqui colocar ruta
git pull origin main
docker compose pull
docker compose up -d --build
docker image prune -f
#__________________
# docker-compose.yml
#este es para producción
services:
# ===================== REVERSE PROXY (único punto expuesto) =====================
nginx:
image: nginx:1.27-alpine
restart: unless-stopped
ports:
- "80:80"
volumes:
- ./nginx/nginx.conf:/etc/nginx/nginx.conf:ro
- ./nginx/.htpasswd:/etc/nginx/.htpasswd:ro
networks:
- public
- internal
depends_on:
app:
condition: service_healthy
logging:
driver: "json-file"
options:
max-size: "10m"
max-file: "3"
# ===================== APP (imagen ya construida y publicada en ghcr.io) =====================
app:
image: ghcr.io/TU_USUARIO/TU_REPOSITORIO:latest #esto se cambia
restart: unless-stopped
expose:
- "8000"
env_file:
- .env
volumes:
- ./downloads:/downloads
environment:
DATABASE_URL: postgresql://${POSTGRES_USER}:${POSTGRES_PASSWORD}@postgres:5432/proyect_framework
REDIS_URL: redis://:${REDIS_PASSWORD}@redis:6379/0
TEMPORAL_ADDRESS: temporal:7233
networks:
- internal
depends_on:
postgres:
condition: service_healthy
redis:
condition: service_healthy
temporal:
condition: service_healthy
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
interval: 30s
timeout: 5s
retries: 3
# ===================== POSTGRES (sin puertos publicados) =====================
postgres:
image: postgres:16-alpine
restart: unless-stopped
env_file:
- .env
environment:
POSTGRES_USER: ${POSTGRES_USER}
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
POSTGRES_DB: proyect_framework
volumes:
- postgres_data:/var/lib/postgresql/data
- ./postgres-init:/docker-entrypoint-initdb.d:ro
networks:
- internal
healthcheck:
test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER}"]
interval: 10s
timeout: 5s
retries: 5
# ===================== REDIS (con password, sin puertos publicados) =====================
redis:
image: redis:7-alpine
restart: unless-stopped
command: ["redis-server", "--requirepass", "${REDIS_PASSWORD}"]
env_file:
- .env
volumes:
- redis_data:/data
networks:
- internal
healthcheck:
test: ["CMD", "redis-cli", "-a", "${REDIS_PASSWORD}", "ping"]
interval: 10s
timeout: 5s
retries: 5
# ===================== TEMPORAL (sin puertos publicados) =====================
temporal:
image: temporalio/auto-setup:1.24
restart: unless-stopped
environment:
DB: postgres12
DB_PORT: 5432
POSTGRES_USER: ${POSTGRES_USER}
POSTGRES_PWD: ${POSTGRES_PASSWORD}
POSTGRES_SEEDS: postgres
DBNAME: temporal
VISIBILITY_DBNAME: temporal_visibility
networks:
- internal
depends_on:
postgres:
condition: service_healthy
healthcheck:
test: ["CMD", "tctl", "--address", "temporal:7233", "cluster", "health"]
interval: 15s
timeout: 5s
retries: 5
# ===================== TEMPORAL UI (solo vía nginx, con autenticación) =====================
temporal-ui:
image: temporalio/ui:2.31.2
restart: unless-stopped
environment:
TEMPORAL_ADDRESS: temporal:7233
networks:
- internal
depends_on:
temporal:
condition: service_healthy
networks:
public:
driver: bridge
internal:
driver: bridge
internal: true
volumes:
postgres_data:
redis_data:
#__________________
# docker-compose.dev.yml
#este es para probar local
services:
# ===================== REVERSE PROXY =====================
nginx:
image: nginx:1.27-alpine
restart: unless-stopped
ports:
- "80:80"
volumes:
- ./nginx/nginx.conf:/etc/nginx/nginx.conf:ro
- ./nginx/.htpasswd:/etc/nginx/.htpasswd:ro
networks:
- public
- internal
depends_on:
app:
condition: service_healthy
# ===================== APP (construida localmente, no descargada) =====================
app:
build:
context: .
dockerfile: Dockerfile
restart: unless-stopped
expose:
- "8000"
env_file:
- .env
environment:
DATABASE_URL: postgresql://${POSTGRES_USER}:${POSTGRES_PASSWORD}@postgres:5432/proyect_framework
REDIS_URL: redis://:${REDIS_PASSWORD}@redis:6379/0
TEMPORAL_ADDRESS: temporal:7233
networks:
- internal
depends_on:
postgres:
condition: service_healthy
redis:
condition: service_healthy
temporal:
condition: service_healthy
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
interval: 30s
timeout: 5s
retries: 3
volumes:
- ./app:/app/app # monta tu código en vivo, para no reconstruir en cada cambio
# ===================== POSTGRES =====================
postgres:
image: postgres:16-alpine
restart: unless-stopped
env_file:
- .env
environment:
POSTGRES_USER: ${POSTGRES_USER}
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
POSTGRES_DB: proyect_framework
volumes:
- postgres_data_dev:/var/lib/postgresql/data
- ./postgres-init:/docker-entrypoint-initdb.d:ro
networks:
- internal
healthcheck:
test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER}"]
interval: 10s
timeout: 5s
retries: 5
# ===================== REDIS =====================
redis:
image: redis:7-alpine
restart: unless-stopped
command: ["redis-server", "--requirepass", "${REDIS_PASSWORD}"]
env_file:
- .env
volumes:
- redis_data_dev:/data
networks:
- internal
healthcheck:
test: ["CMD", "redis-cli", "-a", "${REDIS_PASSWORD}", "ping"]
interval: 10s
timeout: 5s
retries: 5
# ===================== TEMPORAL =====================
temporal:
image: temporalio/auto-setup:1.24
restart: unless-stopped
environment:
DB: postgres12
DB_PORT: 5432
POSTGRES_USER: ${POSTGRES_USER}
POSTGRES_PWD: ${POSTGRES_PASSWORD}
POSTGRES_SEEDS: postgres
DBNAME: temporal
VISIBILITY_DBNAME: temporal_visibility
networks:
- internal
depends_on:
postgres:
condition: service_healthy
healthcheck:
test: ["CMD", "tctl", "--address", "temporal:7233", "cluster", "health"]
interval: 15s
timeout: 5s
retries: 5
# ===================== TEMPORAL UI =====================
temporal-ui:
image: temporalio/ui:2.31.2
restart: unless-stopped
environment:
TEMPORAL_ADDRESS: temporal:7233
networks:
- internal
depends_on:
temporal:
condition: service_healthy
networks:
public:
driver: bridge
internal:
driver: bridge
volumes:
postgres_data_dev:
redis_data_dev:
#__________________
# nginx/nginx.conf
user nginx;
worker_processes auto;
events {
worker_connections 1024;
}
http {
include /etc/nginx/mime.types;
default_type application/octet-stream;
sendfile on;
keepalive_timeout 65;
access_log /var/log/nginx/access.log;
error_log /var/log/nginx/error.log warn;
server_tokens off;
client_max_body_size 10M;
# ===================== App principal (acceso por IP, sin TLS) =====================
server {
listen 80 default_server;
server_name _;
add_header X-Frame-Options "SAMEORIGIN" always;
add_header X-Content-Type-Options "nosniff" always;
add_header Referrer-Policy "strict-origin-when-cross-origin" always;
location / {
proxy_pass http://app:8000;
proxy_http_version 1.1;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_read_timeout 60s;
proxy_connect_timeout 10s;
}
location /health {
proxy_pass http://app:8000/health;
access_log off;
}
# Temporal UI, protegido con autenticación básica, bajo una ruta distinta
location /temporal/ {
auth_basic "Acceso restringido";
auth_basic_user_file /etc/nginx/.htpasswd;
rewrite ^/temporal/(.*)$ /$1 break;
proxy_pass http://temporal-ui:8080;
proxy_http_version 1.1;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
}
}
}
#esto sudo apt-get install apache2-utils
#htpasswd -c ./nginx/.htpasswd admin
#crea Cómo generar el nginx/.htpasswd
#(comando, no archivo de contenido fijo) del archivo nginx
#__________________
#postgres-init:/docker-entrypoint-initdb
#!/bin/bash
set -e
psql -v ON_ERROR_STOP=1 --username "$POSTGRES_USER" <<-EOSQL
CREATE DATABASE proyect_framework;
CREATE DATABASE temporal;
CREATE DATABASE temporal_visibility;
EOSQL
To embed this project on your website, copy the following code and paste it into your website's HTML: