Supercharge secrets management with Vault and Gitleaks: Insights

"""
production_vault_client.py

A production-grade HashiCorp Vault client with:
- Kubernetes service account authentication
- Dynamic secret leasing with automatic renewal
- In-memory caching with TTL
- Exponential backoff retry logic
- Structured logging
- Health check endpoint

Requirements:
    pip install hvac httpx

Environment variables:
    VAULT_ADDR: Vault server URL (e.g., https://vault.internal:8200)
    VAULT_ROLE: Kubernetes auth role name
    VAULT_SECRET_PATH: Path to the secret in Vault (e.g., secret/data/myapp/db)
    VAULT_MOUNT_POINT: Kubernetes auth mount point (default: kubernetes)
    CACHE_TTL_SECONDS: TTL for cached secrets (default: 300)
    MAX_RETRIES: Maximum retry attempts (default: 5)
    LOG_LEVEL: Logging level (default: INFO)
"""

import os
import time
import logging
import threading
from typing import Any, Dict, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta

import hvac
import httpx

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------

VAULT_ADDR = os.environ.get("VAULT_ADDR", "https://vault.internal:8200")
VAULT_ROLE = os.environ.get("VAULT_ROLE", "myapp")
VAULT_SECRET_PATH = os.environ.get("VAULT_SECRET_PATH", "secret/data/myapp/db")
VAULT_MOUNT_POINT = os.environ.get("VAULT_MOUNT_POINT", "kubernetes")
CACHE_TTL_SECONDS = int(os.environ.get("CACHE_TTL_SECONDS", "300"))
MAX_RETRIES = int(os.environ.get("MAX_RETRIES", "5"))
LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO")

# Kubernetes service account token path (default mount)
K8S_TOKEN_PATH = "/var/run/secrets/kubernetes.io/serviceaccount/token"

# ---------------------------------------------------------------------------
# Logging setup
# ---------------------------------------------------------------------------

logger = logging.getLogger("vault_client")
logger.setLevel(LOG_LEVEL)
handler = logging.StreamHandler()
formatter = logging.Formatter(
    "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
handler.setFormatter(formatter)
logger.addHandler(handler)

# ---------------------------------------------------------------------------
# Data classes
# ---------------------------------------------------------------------------

@dataclass
class CachedSecret:
    """Represents a cached secret with metadata."""
    data: Dict[str, Any]
    lease_id: str
    lease_duration: int
    cached_at: datetime = field(default_factory=datetime.utcnow)
    renewable: bool = False

    @property
    def is_expired(self) -> bool:
        """Check if the cached secret has exceeded its TTL."""
        elapsed = (datetime.utcnow() - self.cached_at).total_seconds()
        # Refresh at 80% of TTL to avoid serving stale secrets
        return elapsed > (self.lease_duration * 0.8)


@dataclass
class VaultHealth:
    """Vault health status."""
    initialized: bool
    sealed: bool
    version: str
    cluster_name: str
    replication_state: str


# ---------------------------------------------------------------------------
# Custom exceptions
# ---------------------------------------------------------------------------

class VaultAuthenticationError(Exception):
    """Raised when Vault authentication fails."""
    pass

class VaultSecretNotFoundError(Exception):
    """Raised when the requested secret path does not exist."""
    pass

class VaultConnectionError(Exception):
    """Raised when the Vault server is unreachable."""
    pass

class VaultLeaseError(Exception):
    """Raised when lease operations fail."""
    pass


# ---------------------------------------------------------------------------
# Vault client
# ---------------------------------------------------------------------------

class ProductionVaultClient:
    """
    A production-grade Vault client that handles authentication,
    secret retrieval, caching, and lease renewal automatically.
    """

    def __init__(
        self,
        vault_addr: str = VAULT_ADDR,
        role: str = VAULT_ROLE,
        mount_point: str = VAULT_MOUNT_POINT,
        cache_ttl: int = CACHE_TTL_SECONDS,
        max_retries: int = MAX_RETRIES,
    ):
        self.vault_addr = vault_addr
        self.role = role
        self.mount_point = mount_point
        self.cache_ttl = cache_ttl
        self.max_retries = max_retries

        # Internal state
        self._client: Optional[hvac.Client] = None
        self._cache: Dict[str, CachedSecret] = {}
        self._cache_lock = threading.Lock()
        self._renewal_thread: Optional[threading.Thread] = None
        self._shutdown_event = threading.Event()

    # -----------------------------------------------------------------------
    # Authentication
    # -----------------------------------------------------------------------

    def _read_k8s_token(self) -> str:
        """Read the Kubernetes service account token from the pod."""
        try:
            with open(K8S_TOKEN_PATH, "r") as f:
                token = f.read().strip()
            if not token:
                raise VaultAuthenticationError(
                    "Kubernetes service account token is empty"
                )
            logger.debug("Successfully read Kubernetes service account token")
            return token
        except FileNotFoundError:
            raise VaultAuthenticationError(
                f"Kubernetes token not found at {K8S_TOKEN_PATH}. "
                "Ensure the pod has a service account token mounted."
            )
        except PermissionError:
            raise VaultAuthenticationError(
                f"Permission denied reading {K8S_TOKEN_PATH}"
            )

    def authenticate(self) -> None:
        """
        Authenticate to Vault using Kubernetes service account auth.
        Falls back to VAULT_TOKEN env var for local development.
        """
        self._client = hvac.Client(url=self.vault_addr)

        # For local development, allow VAULT_TOKEN
        dev_token = os.environ.get("VAULT_TOKEN")
        if dev_token:
            logger.info("Using VAULT_TOKEN for local development")
            self._client.token = dev_token
            if self._client.is_authenticated():
                logger.info("Successfully authenticated to Vault via token")
                return
            raise VaultAuthenticationError("VAULT_TOKEN is invalid or expired")

        # Production: Kubernetes auth
        try:
            k8s_token = self._read_k8s_token()
            auth_response = self._client.auth.kubernetes.login(
                role=self.role,
                jwt=k8s_token,
                mount_point=self.mount_point,
            )

            if not auth_response or "auth" not in auth_response:
                raise VaultAuthenticationError(
                    "Kubernetes auth returned empty response"
                )

            client_token = auth_response["auth"]["client_token"]
            lease_duration = auth_response["auth"]["lease_duration"]

            self._client.token = client_token
            logger.info(
                "Successfully authenticated to Vault via Kubernetes auth. "
                f"Lease duration: {lease_duration}s"
            )

            # Start background token renewal
            self._start_token_renewal(lease_duration)

        except hvac.exceptions.VaultError as e:
            raise VaultAuthenticationError(
                f"Vault Kubernetes auth failed: {e}"
            ) from e
        except httpx.ConnectError as e:
            raise VaultConnectionError(
                f"Cannot connect to Vault at {self.vault_addr}: {e}"
            ) from e

    # -----------------------------------------------------------------------
    # Token renewal
    # -----------------------------------------------------------------------

    def _start_token_renewal(self, lease_duration: int) -> None:
        """Start a background thread to renew the client token before expiry."""
        def _renew():
            # Renew at 50% of lease duration
            sleep_time = max(lease_duration * 0.5, 30)
            while not self._shutdown_event.is_set():
                self._shutdown_event.wait(timeout=sleep_time)
                if self._shutdown_event.is_set():
                    break
                try:
                    self._client.auth.token.renew_self()
                    logger.debug("Successfully renewed Vault client token")
                except Exception as e:
                    logger.error(f"Token renewal failed: {e}")
                    # Re-authenticate from scratch
                    try:
                        self.authenticate()
                        logger.info("Re-authenticated to Vault after renewal failure")
                    except VaultAuthenticationError:
                        logger.critical(
                            "Failed to re-authenticate to Vault. "
                            "Secret retrieval will fail."
                        )

        self._renewal_thread = threading.Thread(
            target=_renew, daemon=True, name="vault-token-renewer"
        )
        self._renewal_thread.start()
        logger.debug("Started Vault token renewal thread")

    # -----------------------------------------------------------------------
    # Secret retrieval with caching
    # -----------------------------------------------------------------------

    def get_secret(
        self, path: str = VAULT_SECRET_PATH
    ) -> Dict[str, Any]:
        """
        Retrieve a secret from Vault, using cache if available and fresh.

        Args:
            path: The secret path (e.g., 'secret/data/myapp/db')

        Returns:
            Dictionary containing the secret data.

        Raises:
            VaultSecretNotFoundError: If the path doesn't exist.
            VaultConnectionError: If Vault is unreachable.
        """
        # Check cache first
        with self._cache_lock:
            cached = self._cache.get(path)
            if cached and not cached.is_expired:
                logger.debug(f"Cache hit for secret path: {path}")
                return cached.data

        # Cache miss or expired — fetch from Vault with retry
        secret_data = self._fetch_secret_with_retry(path)

        # Update cache
        with self._cache_lock:
            self._cache[path] = CachedSecret(
                data=secret_data,
                lease_id=f"static-{path}",  # KV v2 doesn't return lease_id
                lease_duration=self.cache_ttl,
                renewable=False,
            )

        return secret_data

    def _fetch_secret_with_retry(
        self, path: str
    ) -> Dict[str, Any]:
        """
        Fetch a secret from Vault with exponential backoff retry.
        """
        last_exception = None

        for attempt in range(1, self.max_retries + 1):
            try:
                if not self._client or not self._client.is_authenticated():
                    logger.warning("Vault client not authenticated, re-authenticating")
                    self.authenticate()

                # Read the secret (KV v2)
                response = self._client.secrets.kv.v2.read_secret_version(
                    path=path,
                )

                if not response or "data" not in response:
                    raise VaultSecretNotFoundError(
                        f"No data returned for path: {path}"
                    )

                secret_data = response["data"]["data"]
                logger.info(f"Successfully fetched secret from path: {path}")
                return secret_data

            except hvac.exceptions.InvalidPath:
                raise VaultSecretNotFoundError(
                    f"Secret not found at path: {path}"
                )
            except hvac.exceptions.Forbidden as e:
                raise VaultAuthenticationError(
                    f"Access denied to path {path}: {e}"
                ) from e
            except (
                httpx.ConnectError,
                httpx.TimeoutException,
                hvac.exceptions.VaultDown,
            ) as e:
                last_exception = e
                wait_time = min(2 ** attempt, 30)  # Cap at 30 seconds
                logger.warning(
                    f"Vault request failed (attempt {attempt}/{self.max_retries}): "
                    f"{e}. Retrying in {wait_time}s..."
                )
                time.sleep(wait_time)

        raise VaultConnectionError(
            f"Failed to fetch secret after {self.max_retries} attempts. "
            f"Last error: {last_exception}"
        )

    # -----------------------------------------------------------------------
    # Dynamic database credentials (example)
    # -----------------------------------------------------------------------

    def get_dynamic_db_credentials(
        self, db_role: str = "myapp-readonly"
    ) -> Dict[str, str]:
        """
        Generate dynamic database credentials from Vault's database secrets engine.

        Returns:
            Dict with 'username', 'password', 'lease_id', 'lease_duration'
        """
        try:
            response = self._client.secrets.database.generate_credentials(
                name=db_role,
            )

            if not response or "data" not in response:
                raise VaultSecretNotFoundError(
                    f"No credentials returned for DB role: {db_role}"
                )

            creds = {
                "username": response["data"]["username"],
                "password": response["data"]["password"],
                "lease_id": response["lease_id"],
                "lease_duration": response["lease_duration"],
                "renewable": response["renewable"],
            }

            logger.info(
                f"Generated dynamic DB credentials for role {db_role}. "
                f"Username: {creds['username']}, "
                f"Lease duration: {creds['lease_duration']}s"
            )

            # Start lease renewal if renewable
            if creds["renewable"]:
                self._start_lease_renewal(
                    creds["lease_id"], creds["lease_duration"]
                )

            return creds

        except hvac.exceptions.InvalidPath:
            raise VaultSecretNotFoundError(
                f"Database role not found: {db_role}"
            )
        except hvac.exceptions.VaultError as e:
            raise VaultConnectionError(
                f"Failed to generate DB credentials: {e}"
            ) from e

    def _start_lease_renewal(
        self, lease_id: str, lease_duration: int
    ) -> None:
        """Start a background thread to renew a dynamic secret lease."""
        def _renew():
            sleep_time = max(lease_duration * 0.5, 30)
            while not self._shutdown_event.is_set():
                self._shutdown_event.wait(timeout=sleep_time)
                if self._shutdown_event.is_set():
                    break
                try:
                    self._client.sys.renew_lease(lease_id=lease_id)
                    logger.debug(f"Renewed lease: {lease_id}")
                except Exception as e:
                    logger.error(f"Lease renewal failed for {lease_id}: {e}")
                    break

        thread = threading.Thread(
            target=_renew,
            daemon=True,
            name=f"lease-renewer-{lease_id[:8]}",
        )
        thread.start()

    # -----------------------------------------------------------------------
    # Health check
    # -----------------------------------------------------------------------

    def health_check(self) -> VaultHealth:
        """Check Vault server health."""
        try:
            response = self._client.sys.read_health_status(
                method="GET",
            )
            return VaultHealth(
                initialized=response.get("initialized", False),
                sealed=response.get("sealed", True),
                version=response.get("version", "unknown"),
                cluster_name=response.get("cluster_name", "unknown"),
                replication_state=response.get("replication_dr_mode", "unknown"),
            )
        except Exception as e:
            logger.error(f"Vault health check failed: {e}")
            return VaultHealth(
                initialized=False,
                sealed=True,
                version="unreachable",
                cluster_name="unknown",
                replication_state="unknown",
            )

    # -----------------------------------------------------------------------
    # Cleanup
    # -----------------------------------------------------------------------

    def close(self) -> None:
        """Gracefully shut down the Vault client."""
        logger.info("Shutting down Vault client")
        self._shutdown_event.set()

        # Revoke all cached leases
        with self._cache_lock:
            for path, cached in self._cache.items():
                if cached.renewable:
                    try:
                        self._client.sys.revoke_lease(
                            lease_id=cached.lease_id
                        )
                        logger.debug(f"Revoked lease for {path}")
                    except Exception as e:
                        logger.warning(
                            f"Failed to revoke lease for {path}: {e}"
                        )
            self._cache.clear()

        logger.info("Vault client shut down complete")


# ---------------------------------------------------------------------------
# Usage example
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    client = ProductionVaultClient()

    try:
        # Authenticate
        client.authenticate()

        # Health check
        health = client.health_check()
        print(f"Vault version: {health.version}")
        print(f"Sealed: {health.sealed}")

        # Get a static secret
        secret = client.get_secret("secret/myapp/config")
        print(f"Database host: {secret.get('db_host')}")

        # Get dynamic database credentials
        db_creds = client.get_dynamic_db_credentials("myapp-readonly")
        print(f"Dynamic DB user: {db_creds['username']}")

    except VaultAuthenticationError as e:
        logger.critical(f"Authentication failed: {e}")
        raise SystemExit(1)
    except VaultConnectionError as e:
        logger.critical(f"Connection failed: {e}")
        raise SystemExit(1)
    finally:
        client.close()

# .gitleaks.toml
# Production Gitleaks configuration
# Docs: https://github.com/gitleaks/gitleaks

# -----------------------------------------------------------------------------
# Global settings
# -----------------------------------------------------------------------------

[allowlist]
  # Files and paths to exclude from scanning
  paths = [
    '''gitleaks\.toml''',
    '''(.*?)(jpg|gif|doc|pdf|bin|svg|socket)$''',
    '''node_modules''',
    '''vendor''',
    '''\.min\.js$''',
    '''testdata''',
    '''_test\.go$''',
  ]

  # Specific commits to allowlist (e.g., legacy commits you can't rewrite)
  commits = [
    "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2",
  ]

  # Regex patterns that are known false positives
  regexes = [
    # Example: placeholder values in documentation
    '''EXAMPLE_[A-Z_]+_PLACEHOLDER''',
    # Example: test fixtures with obviously fake data
    '''test-password-12345''',
  ]

# -----------------------------------------------------------------------------
# Built-in rules are included by default.
# Add custom rules below for organization-specific patterns.
# -----------------------------------------------------------------------------

# Custom rule: Internal API keys
[[rules]]
  id = "internal-api-key"
  description = "Internal API Key Pattern"
  regex = '''(?i)(?:api[_-]?key|apikey)\s*[:=]\s*['\"]?(sk-[a-zA-Z0-9]{32,})['\"]?'''
  tags = ["api-key", "internal"]
  [rules.allowlist]
    regexes = [
      '''sk-test-.*''',  # Allow test keys
    ]

# Custom rule: Database connection strings with passwords
[[rules]]
  id = "db-connection-string"
  description = "Database Connection String with Password"
  regex = '''(?i)(?:mongodb(\+srv)?|postgres(ql)?|mysql|redis)://[^:]+:[^@]+@'''
  tags = ["database", "connection-string"]

# Custom rule: Private keys
[[rules]]
  id = "private-key-rsa"
  description = "RSA Private Key"
  regex = '''-----BEGIN RSA PRIVATE KEY-----'''
  tags = ["key", "private-key"]

# Custom rule: AWS access key ID
[[rules]]
  id = "aws-access-key"
  description = "AWS Access Key ID"
  regex = '''(A3T[A-Z0-9]|AKIA|AGPA|AIDA|AROA|AIPA|ANPA|ANVA|ASIA)[A-Z0-9]{16}'''
  tags = ["aws", "cloud"]

# Custom rule: Generic high-entropy strings that look like secrets
[[rules]]
  id = "high-entropy-secret"
  description = "High Entropy String (potential secret)"
  regex = '''(?i)(?:secret|token|key|password|passwd)\s*[:=]\s*['\"]?([a-zA-Z0-9+/]{32,}=*)['\"]?'''
  entropy = 4.5
  tags = ["entropy", "generic"]
  [rules.allowlist]
    regexes = [
      '''your-secret-here''',
      '''changeme''',
      '''password123''',
      '''admin''',
    ]

# Custom rule: Slack tokens
[[rules]]
  id = "slack-token"
  description = "Slack Token"
  regex = '''xox[bprs]-[a-zA-Z0-9]{10,48}'''
  tags = ["slack", "token"]

# Custom rule: GitHub personal access tokens
[[rules]]
  id = "github-pat"
  description = "GitHub Personal Access Token"
  regex = '''gh[pousr]_[A-Za-z0-9_]{36,}'''
  tags = ["github", "token"]

# .github/workflows/gitleaks.yml
# GitHub Actions workflow for Gitleaks secret scanning
# This runs on every push and pull request to main branches.

name: Gitleaks Secret Scan

on:
  push:
    branches:
      - main
      - develop
      - "release/**"
  pull_request:
    branches:
      - main
      - develop
  # Allow manual triggering from the Actions tab
  workflow_dispatch:
    inputs:
      scan_depth:
        description: "Number of commits to scan (0 for all)"
        required: false
        default: "0"

# Cancel in-progress runs for the same branch
concurrency:
  group: gitleaks-${{ github.ref }}
  cancel-in-progress: true

permissions:
  contents: read
  security-events: write  # Required for SARIF upload
  pull-requests: write    # Required for PR comments

jobs:
  gitleaks-scan:
    name: "🔍 Gitleaks Secret Detection"
    runs-on: ubuntu-latest
    timeout-minutes: 15

    steps:
      # -----------------------------------------------------------------------
      # Step 1: Checkout the repository with full history
      # -----------------------------------------------------------------------
      - name: Checkout repository
        uses: actions/checkout@v4
        with:
          # Fetch full history so Gitleaks can scan all commits
          fetch-depth: ${{ github.event.inputs.scan_depth || 0 }}

      # -----------------------------------------------------------------------
      # Step 2: Install Gitleaks
      # -----------------------------------------------------------------------
      - name: Install Gitleaks
        run: |
          GITLEAKS_VERSION="8.18.4"
          echo "Installing Gitleaks v${GITLEAKS_VERSION}"

          # Download the latest release
          curl -sSfL \
            "https://github.com/gitleaks/gitleaks/releases/download/v${GITLEAKS_VERSION}/gitleaks_${GITLEAKS_VERSION}_linux_x64.tar.gz" \
            -o /tmp/gitleaks.tar.gz

          # Verify checksum (recommended for production)
          curl -sSfL \
            "https://github.com/gitleaks/gitleaks/releases/download/v${GITLEAKS_VERSION}/gitleaks_${GITLEAKS_VERSION}_checksums.txt" \
            -o /tmp/checksums.txt

          cd /tmp
          sha256sum -c checksums.txt --ignore-missing

          # Extract and install
          tar xzf gitleaks.tar.gz
          sudo mv gitleaks /usr/local/bin/

          # Verify installation
          gitleaks version

      # -----------------------------------------------------------------------
      # Step 3: Run Gitleaks scan
      # -----------------------------------------------------------------------
      - name: Run Gitleaks
        id: gitleaks
        run: |
          echo "Starting Gitleaks scan..."

          # Determine scan arguments based on event type
          if [ "${{ github.event_name }}" = "pull_request" ]; then
            SCAN_ARGS="--log-opts='--no-merges ${{ github.event.pull_request.base.sha }}..${{ github.event.pull_request.head.sha }}'"
            echo "Scanning PR commits: ${{ github.event.pull_request.base.sha }}..${{ github.event.pull_request.head.sha }}"
          else
            SCAN_ARGS=""
            echo "Scanning full repository history"
          fi

          # Run Gitleaks with SARIF output for GitHub Security tab
          gitleaks detect \
            --source=. \
            --config=.gitleaks.toml \
            --report-format=sarif \
            --report-path=gitleaks-results.sarif \
            --exit-code=1 \
            --verbose \
            --no-banner \
            $SCAN_ARGS \
            2>&1 | tee gitleaks-output.log

          SCAN_EXIT_CODE=${PIPESTATUS[0]}

          # Store the exit code for later steps
          echo "exit_code=$SCAN_EXIT_CODE" >> $GITHUB_OUTPUT

          if [ $SCAN_EXIT_CODE -ne 0 ]; then
            echo "❌ Gitleaks detected potential secrets!"
            echo "leaks_found=true" >> $GITHUB_OUTPUT
          else
            echo "✅ No secrets detected."
            echo "leaks_found=false" >> $GITHUB_OUTPUT
          fi
        continue-on-error: true

      # -----------------------------------------------------------------------
      # Step 4: Upload SARIF results to GitHub Security tab
      # -----------------------------------------------------------------------
      - name: Upload SARIF results
        if: always()
        uses: github/codeql-action/upload-sarif@v3
        with:
          sarif_file: gitleaks-results.sarif
          category: gitleaks

      # -----------------------------------------------------------------------
      # Step 5: Comment on PR with results
      # -----------------------------------------------------------------------
      - name: Comment on PR
        if: github.event_name == 'pull_request' && steps.gitleaks.outputs.leaks_found == 'true'
        uses: actions/github-script@v7
        with:
          script: |
            const fs = require('fs');
            const output = fs.readFileSync('gitleaks-output.log', 'utf8');

            const body = `## 🔴 Gitleaks Secret Detection Alert

            Potential secrets were detected in this pull request.


            Click to expand findings

            \`\`\`
            ${output.substring(0, 3000)}
            ${output.length > 3000 ? '\n... (truncated)' : ''}
            \`\`\`



            **Next steps:**
            1. Review the findings above
            2. If false positives, update \`.gitleaks.toml\` allowlist
            3. If real secrets, rotate them immediately and use git-filter-repo to remove from history
            4. Use HashiCorp Vault for secret management going forward

            See [Security Runbook](https://wiki.internal/security/secret-leak) for detailed remediation steps.`;

            github.rest.issues.createComment({
              owner: context.repo.owner,
              repo: context.repo.repo,
              issue_number: context.issue.number,
              body: body
            });

      # -----------------------------------------------------------------------
      # Step 6: Fail the workflow if secrets were found
      # -----------------------------------------------------------------------
      - name: Fail on secrets detected
        if: steps.gitleaks.outputs.leaks_found == 'true'
        run: |
          echo "::error::Gitleaks detected potential secrets. Failing the build."
          exit 1

      # -----------------------------------------------------------------------
      # Step 7: Upload scan artifacts for audit trail
      # -----------------------------------------------------------------------
      - name: Upload scan artifacts
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: gitleaks-results-${{ github.run_id }}
          path: |
            gitleaks-results.sarif
            gitleaks-output.log
          retention-days: 90

# .pre-commit-config.yaml
# Pre-commit hooks configuration
# Install: pip install pre-commit && pre-commit install
# Run manually: pre-commit run --all-files

repos:
  # Gitleaks - Secret detection
  - repo: https://github.com/gitleaks/gitleaks
    rev: v8.18.4
    hooks:
      - id: gitleaks
        name: "Gitleaks: Detect secrets"
        args: ["--verbose", "--redact"]
        # The --redact flag masks detected secrets in output
        # to avoid leaking them in CI logs

  # Additional security hooks
  - repo: https://github.com/pre-commit/pre-commit-hooks
    rev: v4.6.0
    hooks:
      - id: detect-private-key
        name: "Detect private keys"
      - id: check-added-large-files
        name: "Check for large files"
        args: ["--maxkb=500"]

  # Custom local hook for additional checks
  - repo: local
    hooks:
      - id: check-env-files
        name: "Check for .env files"
        entry: 'Files matching .env patterns are not allowed. Use Vault instead.'
        language: fail
        files: '\.env(\..+)?$'
        exclude: '\.env\.example$'

# Create a wrapped secret (run by CI pipeline or admin)
vault kv get -wrap_ttl=3600 -format=json secret/myapp/bootstrap
# Returns: {"wrap_info": {"token": "s.xxxxxxxx", "ttl": 3600, ...}}

# Unwrap the secret (run by the target service)
VAULT_TOKEN="s.xxxxxxxx" vault unwrap -format=json
# Returns the actual secret data. Token is now invalid.

# Run Gitleaks in verbose mode to collect findings
gitleaks detect --source=. --config=.gitleaks.toml --verbose --no-git \
  --report-format=json --report-path=findings.json

# Analyze findings and update allowlist
cat findings.json | jq -r '.[].RuleID' | sort | uniq -c | sort -rn
# Output:
#   15 high-entropy-secret
#    8 internal-api-key
#    3 db-connection-string

#!/bin/bash
# rotate-secret.sh — Emergency secret rotation script
# Usage: ./rotate-secret.sh secret/myapp/api-key

set -euo pipefail

SECRET_PATH="${1:?Usage: $0 }"
VAULT_ADDR="${VAULT_ADDR:?VAULT_ADDR must be set}"

echo "🔄 Rotating secret: $SECRET_PATH"

# Step 1: Generate new secret value
NEW_SECRET=$(openssl rand -base64 48 | tr -d '\n')
echo "Generated new secret value"

# Step 2: Update Vault
vault kv put "$SECRET_PATH" value="$NEW_SECRET"
echo "✅ Updated Vault"

# Step 3: Force cache invalidation by reducing TTL
# (Services with short cache TTLs will pick this up quickly)
echo "Waiting 60 seconds for cache propagation..."
sleep 60

# Step 4: Verify the new secret is accessible
VERIFY=$(vault kv get -field=value "$SECRET_PATH")
if [ "$VERIFY" = "$NEW_SECRET" ]; then
  echo "✅ Secret rotation verified"
else
  echo "❌ Secret rotation verification failed!"
  exit 1
fi

# Step 5: Scan Git history for the old secret
# (Use gitleaks with a custom rule for the specific secret pattern)
echo "🔍 Scanning Git history for old secret value..."
gitleaks detect --source=. --no-git --verbose || true

echo "✅ Rotation complete. Update dependent services if needed."