Seguro Cifrado con Hashicorp Vault

Una pregunta que siempre aparece cuando trabajamos con bases de datos es: ¿Qué pasa si alguien accede directamente a la base y extrae la información? Si no hay ningún tipo de protección, cualquier dato sensible —como DNIs, CUITs o tarjetas— puede quedar completamente expuesto. Para responder a ese riesgo, armé un proceso paso a paso para proteger esta información utilizando HashiCorp Vault como sistema de cifrado y control de acceso. La solución se integra con una base de datos MySQL y una aplicación desarrollada en Python, aunque el lenguaje es indistinto: lo importante es que todo se hace de forma segura, programática y auditable. Vamos a probar con una app Python para simplificar, pero este enfoque puede aplicarse a cualquier tecnología que pueda integrarse con Vault.

🎯 ¿Qué queríamos lograr?

Cifrar y descifrar DNIs mediante Vault desde una app Python, podria ser el dato que desees.
Guardar los valores cifrados en una base MySQL.
Auditar todos los accesos a Vault (lectura/escritura).
Hacer todo esto de forma segura y programática, sin intervención manual, usando AppRole.

Tabla de MySQL

Cree una base de datos llamada appdb, donde generaremos esta tabla.

CREATE TABLE usuarios (
  id INT AUTO_INCREMENT PRIMARY KEY,
  nombre VARCHAR(100),
  apellido VARCHAR(100),
  dni_cifrado TEXT,
  fecha_creacion TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

Ya tenemos nuestra base de datos y Vault.

Vault

Perfecto, si ya tenés Vault funcionando, vamos a hacer un paso a paso bien claro para:

Habilitar el motor Transit.
Crear la llave, la llamaremos dni-key.
Crear una política en Vault que permita encriptar y desencriptar.
Crear un Role de AppRole que use esa política.
Obtener el Role ID y el Secret ID para que tu app Python lo use.

Habilitar el motor Transit

vault secrets enable transit

🔑 Crear una clave para cifrado

vault write -f transit/keys/dni-key

Creamos la política en Vault transit-app.hcl

path "transit/keys/dni-key" {
  capabilities = ["create", "read", "update"]
}

path "transit/encrypt/dni-key" {
  capabilities = ["update"]
}

path "transit/decrypt/dni-key" {
  capabilities = ["update"]
}

Creamos el AppRole y la asociamos.

vault auth enable approle

vault write auth/approle/role/python-app \
  token_policies="transit-app" \
  token_ttl=1h \
  token_max_ttl=4h

🧠 ¿Por qué es útil?

Estás limitando el tiempo de vida del token, lo que reduce el impacto si es comprometido.

Asignás una política clara y específica (transit-app), evitando privilegios excesivos.

Te permite mantener el control desde el lado de Vault, y no desde la aplicación.

🔐 Obtenemos Role ID y Secret ID

vault read auth/approle/role/python-app/role-id
vault write -f auth/approle/role/python-app/secret-id

Guardamos el role-id y el secret-id.

Ahora vamos a ver la aplicacion de Python para poder insertar cifrado el DNI, se las dejo aca.

import os
import requests
import mysql.connector
import base64
from dotenv import load_dotenv

# Cargar .env
load_dotenv()

# ======== CONFIG ========
VAULT_ADDR = os.getenv("VAULT_ADDR", "https://vault.esprueba.com")
ROLE_ID = os.getenv("VAULT_ROLE_ID")
SECRET_ID = os.getenv("VAULT_SECRET_ID")
TRANSIT_KEY_NAME = "dni-key"

MYSQL_HOST = "127.0.0.1"
MYSQL_PORT = 3306
MYSQL_USER = "admin"
MYSQL_PASSWORD = "TUPASS"
MYSQL_DATABASE = "appdb"

# ======== VAULT AUTH ========
def get_vault_token():
    url = f"{VAULT_ADDR}/v1/auth/approle/login"
    headers = {"Content-Type": "application/json"}
    data = {"role_id": ROLE_ID, "secret_id": SECRET_ID}
    try:
        resp = requests.post(url, headers=headers, json=data, timeout=10)
        resp.raise_for_status()
        token = resp.json()["auth"]["client_token"]
        return token
    except requests.exceptions.RequestException as e:
        print("❌ Error autenticando en Vault:", e)
        exit(1)

# ======== VAULT ENCRYPT ========
def encrypt_dni(vault_token, dni):
    url = f"{VAULT_ADDR}/v1/transit/encrypt/{TRANSIT_KEY_NAME}"
    headers = {"X-Vault-Token": vault_token}
    dni_b64 = base64.b64encode(dni.encode("utf-8")).decode("utf-8")
    data = {"plaintext": dni_b64}
    try:
        resp = requests.post(url, headers=headers, json=data)
        resp.raise_for_status()
        return resp.json()["data"]["ciphertext"]
    except requests.exceptions.RequestException as e:
        print("❌ Error cifrando DNI:", e)
        exit(1)

# ======== MYSQL INSERT ========
def insertar_usuario(nombre, apellido, dni_cifrado):
    try:
        conn = mysql.connector.connect(
            host=MYSQL_HOST,
            port=MYSQL_PORT,
            user=MYSQL_USER,
            password=MYSQL_PASSWORD,
            database=MYSQL_DATABASE
        )
        cursor = conn.cursor()
        cursor.execute(
            "INSERT INTO usuarios (nombre, apellido, dni_cifrado) VALUES (%s, %s, %s)",
            (nombre, apellido, dni_cifrado)
        )
        conn.commit()
        conn.close()
        print(f"✅ Usuario '{nombre} {apellido}' insertado correctamente.")
    except mysql.connector.Error as err:
        print("❌ Error al conectar con MySQL:", err)
        exit(1)

# ======== VALIDACIÓN DNI ========
def pedir_dni():
    while True:
        dni = input("🪪 Ingresá el DNI (8 dígitos): ").strip()
        if dni.isdigit() and len(dni) == 8:
            return dni
        print("❌ DNI inválido. Debe tener exactamente 8 dígitos numéricos.")

# ======== MAIN ========
if __name__ == "__main__":
    print("🚀 Iniciando carga de usuario...")
    nombre = input("🧑 Ingresá el nombre: ").strip()
    apellido = input("🧑 Ingresá el apellido: ").strip()
    dni = pedir_dni()

    print("🔐 Autenticando con Vault...")
    token = get_vault_token()

    print("🔒 Cifrando DNI...")
    dni_cifrado = encrypt_dni(token, dni)

    print("💾 Insertando en base de datos...")
    insertar_usuario(nombre, apellido, dni_cifrado)

La ejecutamos y uala! Vamos a revisar como impacto en la base de datos.

Ahora les dejo el codigo, para poder revisar el campo.

import os
import mysql.connector
import requests
import base64
from dotenv import load_dotenv

# ======== CONFIGURACIÓN ========
load_dotenv()

VAULT_ADDR = os.getenv("VAULT_ADDR", "https://vault.esprueba.com")
ROLE_ID = os.getenv("VAULT_ROLE_ID")
SECRET_ID = os.getenv("VAULT_SECRET_ID")
VAULT_KEY = "dni-key"

DB_HOST = os.getenv("DB_HOST", "127.0.0.1")
DB_PORT = int(os.getenv("DB_PORT", 3306))
DB_USER = os.getenv("DB_USER", "admin")
DB_PASSWORD = os.getenv("DB_PASSWORD", "TUPASS")
DB_NAME = os.getenv("DB_NAME", "appdb")

# ======== VAULT ========
def get_vault_token():
    print("🔐 Autenticando con Vault via AppRole...")
    try:
        url = f"{VAULT_ADDR}/v1/auth/approle/login"
        payload = {"role_id": ROLE_ID, "secret_id": SECRET_ID}
        resp = requests.post(url, json=payload)
        resp.raise_for_status()
        print("✅ Token obtenido.")
        return resp.json()["auth"]["client_token"]
    except Exception as e:
        print(f"❌ Error autenticando con Vault: {e}")
        exit(1)

def decrypt_dni(vault_token, ciphertext):
    url = f"{VAULT_ADDR}/v1/transit/decrypt/{VAULT_KEY}"
    headers = {"X-Vault-Token": vault_token}
    payload = {"ciphertext": ciphertext}
    try:
        resp = requests.post(url, headers=headers, json=payload)
        resp.raise_for_status()
        plaintext_b64 = resp.json()["data"]["plaintext"]
        dni = base64.b64decode(plaintext_b64).decode("utf-8")
        return dni
    except base64.binascii.Error as e:
        print(f"⚠️ Base64 decode error: {e}")
        print(f"🔍 Base64 recibido de Vault: {plaintext_b64}")
        return "<Error al decodificar>"
    except Exception as e:
        print(f"❌ Error descifrando con Vault: {e}")
        return "<Error de Vault>"

# ======== MYSQL ========
def get_usuarios():
    try:
        conn = mysql.connector.connect(
            host=DB_HOST,
            port=DB_PORT,
            user=DB_USER,
            password=DB_PASSWORD,
            database=DB_NAME
        )
        cursor = conn.cursor()
        cursor.execute("SELECT nombre, apellido, dni_cifrado FROM usuarios")
        resultados = cursor.fetchall()
        conn.close()
        return resultados
    except mysql.connector.Error as err:
        print("❌ Error al conectar con MySQL:", err)
        exit(1)

# ======== MAIN ========
if __name__ == "__main__":
    vault_token = get_vault_token()

    print("🔍 Buscando usuarios...\n")
    for nombre, apellido, cifrado in get_usuarios():
        print(f"🔎 Descifrando ciphertext de {nombre} {apellido}...")
        dni = decrypt_dni(vault_token, cifrado)
        print(f"👤 {nombre} {apellido} - 🪪 DNI: {dni}")

Aca funcionando con ambos programas.

Este flujo es una forma sencilla pero potente de aplicar seguridad a nivel de aplicación usando HashiCorp Vault. La tokenización o cifrado a través de Vault con transit garantiza que incluso si la base es comprometida, los datos sensibles no sean legibles sin acceso a Vault.

Espero que les sirva.

Cifrado de Datos Sensibles con Hashicorp Vault