1. Introducción: Es hora de tocar el código

¡Hola! En la primera parte discutimos las razones para construir un sistema de despliegue automático de GitHub Webhook y los materiales necesarios, y en la segunda parte examinamos la arquitectura general y el papel clave del servicio webhook de FastAPI. Ahora es el momento de trasladar la imagen que tenemos en nuestra mente a código real.

Si no has visto las partes anteriores, puedes consultarlas en los siguientes enlaces.

Construcción de un sistema de despliegue automático personalizado utilizando GitHub Webhook

1 - ¿Por qué implementar esto por ti mismo? 2 - Diseño de arquitectura y procesos

En esta tercera parte, finalizaremos la configuración básica del entorno del servidor de staging y crearemos juntos el esqueleto básico del servidor webhook de FastAPI, que podrá recibir y verificar solicitudes de GitHub Webhook de manera segura. ¡Es el momento en el que el sistema de despliegue automático, que ha parecido abstracto, se convierte en una realidad concreta!

2. Configuración inicial del servidor de staging

Lo primero que debemos hacer es preparar el entorno del servidor de staging donde ocurrirá el despliegue, para que podamos ejecutar el código. Conéctate al servidor vía SSH y sigue los siguientes pasos.

2.1. Preparación del entorno de desarrollo en Python

Configurar un entorno virtual de Python en el servidor es una buena práctica para evitar conflictos de dependencias entre proyectos y mantener un entorno de desarrollo ordenado. Instalamos los paquetes necesarios en un entorno venv usando pip.

# Instalación de FastAPI, Uvicorn y python-dotenv para leer el archivo .env
pip install fastapi uvicorn python-dotenv

2.2. Verificación de la instalación de Git, Docker y Docker Compose

Como mencionamos en la parte anterior, Git, Docker y Docker Compose son herramientas clave para el despliegue automático. Asumimos que, como lectores de este artículo, ya tienes estas herramientas instaladas en el servidor.

Si no están instaladas, verifica su disponibilidad con los siguientes comandos.

git --version
docker --version
docker compose version # o docker-compose --version (dependiendo de cómo se haya instalado)

Si no están instaladas, consulta la documentación oficial de cada herramienta para proceder con la instalación. Si necesitas una guía detallada sobre la instalación de Docker, puedes consultar el siguiente artículo: Guía de instalación de Docker

Nota: Después de instalar Docker, es necesario otorgar permisos del grupo Docker al usuario actual y cerrar sesión antes de volver a iniciar sesión para poder usar el comando docker sin sudo.

sudo usermod -aG docker $USER

Después de otorgar permisos de Docker, asegúrate de cerrar sesión y volver a iniciar sesión en el servidor para que los cambios se apliquen. Verifica ejecutando el comando docker run hello-world sin sudo para asegurar que funciona correctamente.

Imagen de resumen del servidor webhook de FastAPI

3. Crear la estructura básica del servidor webhook de FastAPI

Ahora creamos el esqueleto básico de la aplicación FastAPI que aceptará solicitudes webhook. Crea un archivo main.py dentro del directorio ~/projects/webhook_server y escribe el siguiente código.

3.1. Creación del archivo main.py e inicialización de la app de FastAPI

He escrito toda la lógica en main.py para fines de explicación, pero te recomiendo que en tu proyecto real lo dividas en archivos de módulo individuales para cada función de utilidad y los importes, de modo que se vean ordenados según tu estilo. Aquí hay un ejemplo de cómo hacerlo.

# ~/projects/webhook_server/main.py
import hmac
import hashlib
import os
import logging
import subprocess
from pathlib import Path
from typing import Optional

from fastapi import FastAPI, Request, HTTPException, BackgroundTasks
from dotenv import load_dotenv, dotenv_values

# Cargar el archivo .env (para usar variables de entorno)
load_dotenv()

# Configuración del registro
logging.basicConfig(
    level=logging.INFO, # Registrar logs de nivel INFO y superiores
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("webhook_server.log"), # Guardar logs en un archivo
        logging.StreamHandler() # También imprimir logs en consola
    ]
)
logger = logging.getLogger(__name__)

app = FastAPI()

# Cargar el secreto de GitHub Webhook desde las variables de entorno
GITHUB_WEBHOOK_SECRET = os.getenv("GITHUB_WEBHOOK_SECRET")

# Ruta de endpoint para pruebas: es recomendable eliminarla por seguridad en producción.
# Cuando está expuesta a Internet, no es necesario anunciar su existencia.
@app.get("/")
async def read_root():
    return {"message": "¡Servidor webhook en funcionamiento!"}

# Endpoint para recibir el Webhook de GitHub
@app.post("/webhook")
async def github_webhook(request: Request, background_tasks: BackgroundTasks):
    logger.info("Solicitud de Webhook recibida.")

    # 1. Verificación del secreto de GitHub Webhook (primera etapa de seguridad a realizar)
    if GITHUB_WEBHOOK_SECRET:
        signature = request.headers.get("X-Hub-Signature-256")
        if not signature:
            logger.warning("Falta el encabezado de firma. Abortando.")
            raise HTTPException(status_code=401, detail="Falta el encabezado X-Hub-Signature-256")

        body = await request.body()
        expected_signature = "sha256=" + hmac.new(
            GITHUB_WEBHOOK_SECRET.encode('utf-8'),
            body,
            hashlib.sha256
        ).hexdigest()

        if not hmac.compare_digest(expected_signature, signature):
            logger.warning("Firma inválida. Abortando.")
            raise HTTPException(status_code=401, detail="Firma inválida")
        else:
            logger.info("Firma verificada correctamente.")
    else:
        logger.warning("GITHUB_WEBHOOK_SECRET no está configurado. Saltando la verificación de la firma (NO RECOMENDADO PARA PRODUCCIÓN).")


    # 2. Análisis del cuerpo HTTP (Payload)
    try:
        payload = await request.json()
        event_type = request.headers.get("X-GitHub-Event")
        # Extraer el nombre del repositorio del payload de GitHub Webhook.
        # Usamos 'name' en lugar de 'full_name'.
        repo_name = payload.get("repository", {}).get("name", "unknown_repository")
        ref = payload.get("ref", "unknown_ref")
        pusher_name = payload.get("pusher", {}).get("name", "unknown_pusher")

        logger.info(f"Evento: {event_type}, Repo: {repo_name}, Ref: {ref}, Pusher: {pusher_name}")

        # Ejecutar la lógica de despliegue en segundo plano
        background_tasks.add_task(handle_deploy, repo_name)

    except Exception as e:
        logger.error(f"Error al analizar el payload del webhook o procesar la solicitud: {e}")
        raise HTTPException(status_code=400, detail=f"Error procesando la solicitud: {e}")

    # Devolver respuesta inmediata a GitHub
    return {"message": "Webhook recibido y procesamiento iniciado en segundo plano!"}

# Función que maneja la lógica del despliegue
def handle_deploy(repo_name: str):
    # El repositorio llamado 'deployer' se usa para actualizar el propio servidor webhook, por lo que se omite
    if repo_name == "deployer":
        logger.info("⚙️ Actualización del Deployer omitida.")
        return

    # Leer la ruta local de cada repositorio desde las variables de entorno.
    # Este diccionario debe mapear los nombres de tus proyectos reales y sus rutas en el servidor.
    # Ejemplo: la ruta del servidor del proyecto debe estar almacenada en la variable de entorno SAMPLE_PROJECT_1_PATH.
    repo_paths = {
        "sample_project1": os.getenv("SAMPLE_PROJECT_1_PATH"),
        "sample_project2": os.getenv("SAMPLE_PROJECT_2_PATH"),
        "sample_project3": os.getenv("SAMPLE_PROJECT_3_PATH"),
    }

    # Es recomendable mapear de nuevo los nombres de los proyectos si difieren del nombre de los proyectos en Docker Compose.
    # Esto permitirá usarlos con la opción -p (nombre del proyecto) de Docker Compose.
    # Ten en cuenta que, debido a la naturaleza de Docker, no se permiten mayúsculas ni guiones en la opción -p.
    p_name = {
        "sample_project1": "p_name_1",
        "sample_project2": "p_name_2",
        "sample_project3": "p_name_3",
    }

    # Encontrar la ruta local del repositorio que recibió la solicitud webhook.
    repo_path: Optional[str] = repo_paths.get(repo_name)

    # Si el repositorio no está definido, dejar un log de advertencia y salir.
    if not repo_path:
        logger.warning(f"🚫 Repositorio desconocido: {repo_name}")
        return

    # Convertir a un objeto Path para manejar fácilmente la ruta del sistema de archivos.
    repo_path = Path(repo_path)

    # Cargar los valores DEBUG y COLOR desde el archivo .env del repositorio.
    # Esto puede ser útil para distinguir entre entornos de desarrollo/producción por proyecto o para configurar dinámicamente el nombre del proyecto de Docker Compose.
    env_path = repo_path / ".env"
    env_vars = dotenv_values(env_path) if env_path.exists() else {}
    is_debug = env_vars.get("DEBUG", "false").lower() == "true" # Convertir el string "true" a boolean
    deploy_color = env_vars.get("COLOR", "red") # El valor por defecto es "red"
    project_name = str(deploy_color) + p_name.get(repo_name, repo_name) # Nombre del proyecto de Docker Compose

    # También se puede decidir qué archivo de Docker Compose usar según el valor DEBUG.
    docker_compose_file = "docker-compose.dev.yml" if is_debug else "docker-compose.prod.yml"

    # Definir la ruta del archivo de Docker Compose.
    compose_file = repo_path / docker_compose_file

    logger.info(f"Obteniendo el código más reciente para {repo_name} en {repo_path}...")
    # Ejecutar Git Pull: obtener el código más reciente desde la ruta del repositorio.
    # La opción check=True genera una excepción subprocess.CalledProcessError si falla el comando.
    try:
        subprocess.run(["git", "-C", repo_path, "pull"], check=True, capture_output=True, text=True)
        logger.info(f"Pull de Git exitoso para {repo_name}.")
    except subprocess.CalledProcessError as e:
        logger.error(f"Error en el pull de Git para {repo_name}: {e.stderr}")
        return # Salir en caso de error en el despliegue

    logger.info(f"Verificando si hay cambios en {repo_name} para decidir reconstruir...")
    # Comparar cambios: determinar si se necesita reconstruir la imagen de Docker.
    changed = should_rebuild(repo_path)

    try:
        if changed:
            logger.info(f"🚀 Cambios detectados. Construyendo y desplegando {repo_name} con {docker_compose_file}...")
            # Si hay cambios, ejecutar Docker Compose con la opción --build para reconstruir las imágenes.
            subprocess.run(["docker", "compose", "-p", project_name, "-f", compose_file, "up", "-d", "--build"], check=True, capture_output=True, text=True)
            logger.info(f"Docker compose build y up exitoso para {repo_name}.")
        else:
            logger.info(f"✨ No hay cambios significativos. Desplegando {repo_name} sin reconstruir imágenes usando {docker_compose_file}...")
            # Si no hay cambios, ejecutar Docker Compose sin la opción --build.
            subprocess.run(["docker", "compose", "-p", project_name, "-f", compose_file, "up", "-d"], check=True, capture_output=True, text=True)
            logger.info(f"Docker compose up exitoso para {repo_name}.")
    except subprocess.CalledProcessError as e:
        logger.error(f"Error en Docker compose para {repo_name}: {e.stderr}")
        return # Salir en caso de fallo en el despliegue

    logger.info(f"✅ Tarea de despliegue para {repo_name} finalizada.")

# Función que determina si es necesario reconstruir la imagen de Docker
def should_rebuild(repo_path: Path) -> bool:
    # Estos archivos requieren la reconstrucción de la imagen de Docker si cambian.
    trigger_files = [
        "requirements.txt", "Dockerfile", "Dockerfile.dev", "Dockerfile.prod",
        ".dockerignore", ".env" # Cambios en .env también disparan la reconstrucción
    ]

    logger.info(f"Verificando git diff para {repo_path}...")
    # Obtener la lista de archivos cambiados entre HEAD~1 (el commit anterior) y HEAD actual usando git diff.
    # La opción --name-only recupera solamente los nombres de los archivos cambiados.
    try:
        result = subprocess.run(
            ["git", "diff", "--name-only", "HEAD~1"],
            cwd=repo_path,
            capture_output=True, text=True, check=True
        )
        changed_files_in_git = result.stdout.strip().splitlines()
        logger.info(f"Archivos cambiados en git: {changed_files_in_git}")
    except subprocess.CalledProcessError as e:
        logger.error(f"Error en git diff para {repo_path}: {e.stderr}")
        # En caso de fallo, es más seguro intentar reconstruir.
        return True

    # Verificar si hay archivos disparadores en la lista de archivos cambiados.
    if any(f in changed_files_in_git for f in trigger_files):
        logger.info("Cambios de archivo disparador detectados a través de git diff. Se requiere reconstrucción.")
        return True

    # Verificar si existe un archivo disparador de reconstrucción manual (REBUILD_TRIGGER).
    # Si existe, se considera un mandato para forzar la reconstrucción y se elimina el archivo.
    rebuild_trigger_file = repo_path / "REBUILD_TRIGGER"
    if rebuild_trigger_file.exists():
        try:
            rebuild_trigger_file.unlink() # Eliminar archivo
            logger.info("Archivo REBUILD_TRIGGER detectado. Se requiere reconstrucción y el archivo disparador fue eliminado.")
            return True
        except Exception as e:
            logger.error(f"Error al eliminar el archivo REBUILD_TRIGGER: {e}")
            # También es seguro intentar reconstruir si falla la eliminación del archivo.
            return True

    logger.info("No se encontraron archivos de disparo para la reconstrucción. No se requiere reconstrucción.")
    return False

3.2. Definición del endpoint /webhook

En el código anterior, la función asíncrona github_webhook definida bajo el decorador app.post("/webhook") es el endpoint que recibirá solicitudes de Webhook de GitHub. Esta función maneja las solicitudes POST entrantes y permite acceder a los encabezados y el cuerpo (Payload) mediante el objeto Request. BackgroundTasks es una poderosa funcionalidad de FastAPI que permite ejecutar tareas que tardan mucho tiempo, como el despliegue, en segundo plano mientras se manda una respuesta inmediata a GitHub.

4. Implementación de la lógica de verificación del secreto de GitHub Webhook

El aspecto de seguridad más crucial en el código anterior es la lógica de verificación usando GITHUB_WEBHOOK_SECRET.

4.1. Importancia del valor Secret

Al configurar el webhook de GitHub, puedes establecer un Secret value. Este valor permite a GitHub crear un hash (Hash) combinando el cuerpo de la solicitud y el valor del secreto y enviar esto en el encabezado X-Hub-Signature-256. Nuestro servidor webhook generará un hash de la misma manera y luego lo comparará con el hash que envió GitHub para confirmar que la solicitud proviene efectivamente de GitHub y que no ha sido alterada en tránsito.

4.2. Verificación usando el encabezado X-Hub-Signature-256

Como se vio en el código, se usa el módulo hmac y hashlib para realizar la verificación del secreto.

  1. Configuración del secreto: Primero, crea un archivo .env en el directorio ~/projects/webhook_server y escribe el valor del secreto que usarás en la configuración del webhook de GitHub. Asegúrate de agregar este archivo a .gitignore para evitar que sea subido al repositorio Git. Este valor del secreto se utilizará más adelante en la configuración del Webhook del repositorio de GitHub.
# ~/projects/webhook_server/.env
GITHUB_WEBHOOK_SECRET="Introduce aquí el valor secreto que usarás para configurar el webhook de GitHub"
  1. Código de FastAPI: En main.py, se utiliza la librería python-dotenv para cargar este archivo .env y obtener el valor de GITHUB_WEBHOOK_SECRET usando os.getenv(). Luego, cada vez que se reciba una solicitud de webhook, se recupera el valor de la firma usando Request.headers.get("X-Hub-Signature-256") y se compara generando el valor esperado usando hmac.new().

Esta lógica de verificación es un escudo necesario para proteger el endpoint webhook de ataques externos, así que nunca la omitas.

5. Configuración del sistema de registro

En el código, hemos realizado una configuración básica de registro usando el módulo logging.

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("webhook_server.log"), # Guardar en el archivo webhook_server.log
        logging.StreamHandler() # También imprimir en consola
    ]
)
logger = logging.getLogger(__name__)

Con esta configuración, cada vez que se reciba una solicitud webhook, se registrará información en el archivo webhook_server.log y se imprimirá en el terminal. Cuando la lógica de despliegue se vuelva más compleja, esta funcionalidad de registro te ayudará a identificar fácilmente en qué etapa ocurrió un problema.

6. Ejecutar el servidor webhook con Uvicorn y realizar pruebas locales

Ahora probemos brevemente si el servidor webhook de FastAPI que hemos creado funciona correctamente en el entorno local. En el directorio ~/projects/webhook_server del servidor de staging, asegurándote de que el entorno virtual esté activado, ejecuta el siguiente comando.

# Verifica si el entorno virtual está activado (deberías ver (venv) al principio)
# source venv/bin/activate # Si no está activado, ejecuta este comando nuevamente

uvicorn main:app --host 0.0.0.0 --port 8000 --reload

La opción --reload reinicia automáticamente el servidor cuando se realizan cambios en el código durante el desarrollo, lo que resulta conveniente.

Accede a 127.0.0.1:8000/docs en tu navegador para verificar que la documentación de la API se haya generado correctamente. Si la documentación se muestra, la aplicación está funcionando bien. Si conoces la dirección IP del servidor (por ejemplo, 192.168.1.100), puedes probar enviando una solicitud de prueba usando el siguiente comando curl desde tu máquina local.

 

# Reemplaza la dirección IP del servidor por 'YOUR_SERVER_IP'.
# Cambia 'YOUR_WEBHOOK_SECRET' por el valor real en el archivo .env.

# Incluye un JSON de prueba simple.
curl -X POST http://YOUR_SERVER_IP:8000/webhook \
     -H "Content-Type: application/json" \
     -H "X-GitHub-Event: push" \
     -H "X-Hub-Signature-256: sha256=$(echo -n '{"repository":{"name":"your-repo"},"ref":"refs/heads/main","pusher":{"name":"test-user"}}' | openssl dgst -sha256 -hmac 'YOUR_WEBHOOK_SECRET' | awk '{print $2}')" \
     -d '{"repository":{"name":"your-repo"},"ref":"refs/heads/main","pusher":{"name":"test-user"}}'

 

Verifica que en la terminal aparezca el mensaje Webhook recibido y procesamiento iniciado en segundo plano! junto con los logs. Asegúrate de que también se registre en el archivo de logs del servidor (webhook_server.log). Si ocurre un error, podrás depurar usando el mensaje de error que aparece en la terminal.

7. Conclusión: Adelanto de la próxima parte

En esta parte hemos configurado el entorno del servidor de staging y completado la estructura básica del servidor webhook de FastAPI que puede recibir solicitudes de GitHub Webhook y verificar el secreto. Ahora, el servidor webhook está listo para recibir solicitudes externas de manera segura.

En la próxima cuarta parte, implementaremos la lógica real de despliegue (Git Pull, decisión sobre la reconstrucción de imagen de Docker, ejecución de Docker Compose) utilizando el módulo subprocess y explicaremos con detalle cómo registrar este servidor webhook de FastAPI como un servicio de Systemd para que se ejecute automáticamente después de un reinicio del servidor. ¡Espero que te emocione!