# Cifrado de Datos Sensibles 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.

![](https://cdn.hashnode.com/res/hashnode/image/upload/v1750102400458/0c8c9726-aad7-49c9-937a-b8c3ea6a0356.png align="center")

```sql
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.

![](https://cdn.hashnode.com/res/hashnode/image/upload/v1750102734989/fb5bcb9a-f299-4961-afa1-24f8ae6c371f.png align="center")

## Vault

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

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

#### Habilitar el motor Transit

```bash
vault secrets enable transit
```

#### 🔑 Crear una clave para cifrado

```bash
vault write -f transit/keys/dni-key
```

![](https://cdn.hashnode.com/res/hashnode/image/upload/v1750103555830/8a4ccb52-c5f2-4c44-af44-8bb21ecad580.png align="center")

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

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

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

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

![](https://cdn.hashnode.com/res/hashnode/image/upload/v1750104742136/9d33513f-df64-4aba-8fe5-473b723f9b48.png align="center")

Creamos el AppRole y la asociamos.

```bash
vault auth enable approle
```

```bash
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

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

![](https://cdn.hashnode.com/res/hashnode/image/upload/v1750103814036/7c28c294-3edc-40ba-af2c-0518a04a3e40.png align="center")

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.

```python
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.

![](https://cdn.hashnode.com/res/hashnode/image/upload/v1750104408561/1d83b686-b1aa-47da-b37a-bbff88b80f85.png align="center")

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

```python
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.

![](https://cdn.hashnode.com/res/hashnode/image/upload/v1750104554928/bfda9876-b7c8-4ea2-8c43-f88a780cde64.png align="center")

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.
