Gestión de Secrets en CI/CD: HashiCorp Vault, SOPS y Testing con Secrets

# vault-config.hcl
storage "consul" {
  address = "127.0.0.1:8500"
  path    = "vault/"
}

listener "tcp" {
  address     = "0.0.0.0:8200"
  tls_disable = 0
  tls_cert_file = "/etc/vault/tls/vault.crt"
  tls_key_file  = "/etc/vault/tls/vault.key"
}

api_addr = "https://vault.example.com:8200"
ui = true

#!/bin/bash
# scripts/vault-setup.sh

# Habilitar KV v2 secrets engine
vault secrets enable -path=secret kv-v2

# Crear secrets para diferentes entornos
vault kv put secret/test/database \
  username=test_user \
  password=secure_password \
  host=test-db.example.com \
  port=5432

# Habilitar AppRole auth para CI/CD
vault auth enable approle

# Crear política para acceso CI/CD
vault policy write cicd-policy - <<EOF
path "secret/data/test/*" {
  capabilities = ["read", "list"]
}

path "auth/token/renew-self" {
  capabilities = ["update"]
}
EOF

# Crear AppRole para CI/CD
vault write auth/approle/role/cicd-role \
  token_policies="cicd-policy" \
  token_ttl=1h \
  token_max_ttl=4h

# tests/vault_client.py
import hvac
import os

class VaultClient:
    def __init__(self, vault_addr=None, approle_id=None, approle_secret=None):
        self.vault_addr = vault_addr or os.getenv('VAULT_ADDR')
        self.approle_id = approle_id or os.getenv('VAULT_ROLE_ID')
        self.approle_secret = approle_secret or os.getenv('VAULT_SECRET_ID')

        self.client = hvac.Client(url=self.vault_addr)
        self._authenticate()

    def _authenticate(self):
        """Autenticar usando AppRole"""
        response = self.client.auth.approle.login(
            role_id=self.approle_id,
            secret_id=self.approle_secret
        )
        self.client.token = response['auth']['client_token']

    def get_secret(self, path: str):
        """Recuperar secret de Vault"""
        secret = self.client.secrets.kv.v2.read_secret_version(
            path=path,
            mount_point='secret'
        )
        return secret['data']['data']

    def get_database_credentials(self, environment: str):
        """Obtener credenciales de BD para entorno especificado"""
        return self.get_secret(f"{environment}/database")

@pytest.fixture(scope="session")
def vault_client():
    """Pytest fixture para cliente Vault"""
    return VaultClient()

def test_database_connection_with_vault(vault_client):
    """Probar conexión a BD usando credenciales Vault"""
    db_creds = vault_client.get_database_credentials('test')

    conn = psycopg2.connect(
        host=db_creds['host'],
        port=db_creds['port'],
        user=db_creds['username'],
        password=db_creds['password']
    )

    cursor = conn.cursor()
    cursor.execute("SELECT 1")
    assert cursor.fetchone()[0] == 1
    conn.close()

# .gitlab-ci.yml
variables:
  VAULT_ADDR: "https://vault.example.com:8200"

.vault_auth: &vault_auth
  before_script:
    - |
      # Autenticar en Vault usando JWT
      export VAULT_TOKEN=$(curl -s \
        --request POST \
        --data "{\"jwt\": \"${CI_JOB_JWT}\", \"role\": \"cicd-role\"}" \
        ${VAULT_ADDR}/v1/auth/jwt/login | jq -r '.auth.client_token')
    - |
      # Recuperar secrets
      export DB_PASSWORD=$(curl -s \
        --header "X-Vault-Token: ${VAULT_TOKEN}" \
        ${VAULT_ADDR}/v1/secret/data/test/database | jq -r '.data.data.password')

test_with_vault_secrets:
  stage: test
  <<: *vault_auth
  script:
    - pytest tests/ -v --db-password=$DB_PASSWORD
  after_script:
    - unset VAULT_TOKEN
    - unset DB_PASSWORD

# .sops.yaml
creation_rules:
  - path_regex: secrets/test/.*\.yaml$
    kms: 'arn:aws:kms:us-east-1:123456789012:key/test-key-id'
    pgp: 'FBC7B9E2A4F9289AC0C1D4843D16CEE4A27381B4'

  - path_regex: secrets/staging/.*\.yaml$
    kms: 'arn:aws:kms:us-east-1:123456789012:key/staging-key-id'
    pgp: 'FBC7B9E2A4F9289AC0C1D4843D16CEE4A27381B4'

# tests/sops_secrets.py
import subprocess
import yaml

class SOPSSecretsManager:
    def __init__(self, secrets_path: str):
        self.secrets_path = secrets_path

    def decrypt_file(self, file_path: str):
        """Descifrar archivo cifrado con SOPS"""
        result = subprocess.run(
            ['sops', '--decrypt', file_path],
            capture_output=True,
            text=True,
            check=True
        )
        return yaml.safe_load(result.stdout)

    def get_secret(self, environment: str, secret_name: str):
        """Obtener secret específico para entorno"""
        file_path = f"{self.secrets_path}/{environment}/{secret_name}.enc.yaml"
        return self.decrypt_file(file_path)

@pytest.fixture(scope="session")
def sops_secrets():
    """Pytest fixture para secrets SOPS"""
    return SOPSSecretsManager('secrets')

def test_database_connection_with_sops(sops_secrets):
    """Probar conexión a BD usando secrets cifrados SOPS"""
    db_config = sops_secrets.get_secret('test', 'database')

    conn = psycopg2.connect(
        host=db_config['database']['host'],
        port=db_config['database']['port'],
        user=db_config['database']['username'],
        password=db_config['database']['password']
    )

    assert conn is not None
    conn.close()

# .github/workflows/test-with-sops.yml
name: Test with SOPS Secrets

on:
  pull_request:
  push:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v3

      - name: Install SOPS
        run: |
          wget https://github.com/mozilla/sops/releases/download/v3.8.1/sops-v3.8.1.linux.amd64
          sudo mv sops-v3.8.1.linux.amd64 /usr/local/bin/sops
          sudo chmod +x /usr/local/bin/sops

      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v2
        with:
          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          aws-region: us-east-1

      - name: Decrypt secrets
        run: |
          sops --decrypt secrets/test/database.enc.yaml > /tmp/database.yaml

      - name: Run tests
        run: pytest tests/ -v

      - name: Cleanup secrets
        if: always()
        run: rm -f /tmp/database.yaml

# tests/test_secret_rotation.py
class SecretRotationTester:
    def test_database_credential_rotation(self):
        """Probar proceso de rotación de credenciales de BD"""
        # Obtener credenciales iniciales
        initial_creds = self.vault.get_database_credentials('test')
        initial_password = initial_creds['password']

        # Verificar conexión con credenciales iniciales
        conn1 = psycopg2.connect(
            host=initial_creds['host'],
            user=initial_creds['username'],
            password=initial_password
        )
        assert conn1 is not None
        conn1.close()

        # Disparar rotación de credenciales
        self._rotate_database_credentials('test')
        time.sleep(5)

        # Obtener credenciales rotadas
        rotated_creds = self.vault.get_database_credentials('test')
        rotated_password = rotated_creds['password']

        # Verificar que la contraseña cambió
        assert rotated_password != initial_password

        # Verificar conexión con nuevas credenciales
        conn2 = psycopg2.connect(
            host=rotated_creds['host'],
            user=rotated_creds['username'],
            password=rotated_password
        )
        assert conn2 is not None
        conn2.close()

# tests/test_secret_detection.py
import re
from pathlib import Path

class SecretLeakDetector:
    PATTERNS = {
        'aws_access_key': r'AKIA[0-9A-Z]{16}',
        'github_token': r'ghp_[0-9a-zA-Z]{36}',
        'private_key': r'-----BEGIN (RSA |EC )?PRIVATE KEY-----',
        'jwt': r'eyJ[A-Za-z0-9_-]*\\.eyJ[A-Za-z0-9_-]*\\.[A-Za-z0-9_-]*'
    }

    def scan_file(self, file_path: Path):
        """Escanear archivo en busca de secrets potenciales"""
        findings = []

        with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
            content = f.read()

            for secret_type, pattern in self.PATTERNS.items():
                matches = re.finditer(pattern, content, re.IGNORECASE)
                for match in matches:
                    findings.append({
                        'file': str(file_path),
                        'type': secret_type,
                        'line': content[:match.start()].count('\n') + 1
                    })

        return findings

def test_no_secrets_in_code():
    """Verificar que no hay secrets en el repositorio"""
    detector = SecretLeakDetector()

    findings = detector.scan_repository(exclude_patterns=[
        '*.enc.yaml',
        '.git/*',
        'tests/fixtures/*'
    ])

    assert len(findings) == 0, f"Se encontraron secrets potenciales: {findings}"