Handling Database Credentials Securely

Database credentials are among the highest-value secrets: a stolen database password grants direct access to all user data, financial records, and proprietary information. Many breaches occur because developers store database passwords in plaintext in configuration files, Docker environment, or hardcoded strings. Python database drivers (psycopg2 for PostgreSQL, PyMySQL for MySQL, pymongo for MongoDB) all support secure credential patterns: connection URLs (DSNs), environment variable injection, and integration with secret managers. This article covers the security-by-default approach to database credentials, avoiding pitfalls that leak data.

Using Database URLs (DSNs) Safely

A Database Source Name (DSN), also called a connection URI or connection string, is a URL that encodes credentials and connection details in a single string. For example:

postgresql://username:password@host:5432/database_name
mysql://user:password@localhost:3306/mydb
mongodb+srv://user:[email protected]/database

Always build DSNs from environment variables, never hardcode credentials:

import os
import psycopg2

# WRONG: Hardcoded credentials
# conn = psycopg2.connect("postgresql://admin:mypass123@localhost:5432/mydb")

# CORRECT: Environment variables
db_user = os.getenv("DATABASE_USER")
db_password = os.getenv("DATABASE_PASSWORD")
db_host = os.getenv("DATABASE_HOST")
db_port = os.getenv("DATABASE_PORT", "5432")
db_name = os.getenv("DATABASE_NAME")

if not all([db_user, db_password, db_host, db_name]):
    raise ValueError("Missing required database environment variables")

# Build DSN from environment
dsn = f"postgresql://{db_user}:{db_password}@{db_host}:{db_port}/{db_name}"

# Escape special characters in password
from urllib.parse import quote_plus
safe_password = quote_plus(db_password)
dsn = f"postgresql://{db_user}:{safe_password}@{db_host}:{db_port}/{db_name}"

conn = psycopg2.connect(dsn)

The quote_plus() function escapes special characters (like @ or : in passwords), preventing parsing errors.

Using Connection Pools to Reduce Exposure

Every time your application creates a new database connection, credentials are transmitted and new connection state is allocated. A connection pool reuses connections, reducing the number of times credentials are sent and improving performance. Connection pools also support automatic connection validation and refresh:

import psycopg2.pool
import os

# Create a connection pool
pool = psycopg2.pool.SimpleConnectionPool(
    minconn=1,
    maxconn=5,
    host=os.getenv("DATABASE_HOST"),
    user=os.getenv("DATABASE_USER"),
    password=os.getenv("DATABASE_PASSWORD"),
    database=os.getenv("DATABASE_NAME")
)

# Get a connection from the pool
conn = pool.getconn()

try:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users LIMIT 1")
    print(cursor.fetchone())
    cursor.close()
finally:
    pool.putconn(conn)  # Return connection to pool

# Clean up at application shutdown
pool.closeall()

For web frameworks (Flask, Django), use an ORM or connection pooling library:

# Flask + SQLAlchemy with connection pooling
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
import os

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = os.getenv("DATABASE_URL")
app.config["SQLALCHEMY_ENGINE_OPTIONS"] = {
    "pool_size": 10,
    "pool_recycle": 3600,  # Recycle connections older than 1 hour
    "pool_pre_ping": True,  # Test connection before use
}

db = SQLAlchemy(app)

The pool_recycle option closes old connections periodically (useful if your database server has a connection timeout). The pool_pre_ping option tests each connection before handing it to the application, avoiding "connection lost" errors.

Storing Credentials in Environment Variables vs Secret Managers

For development and small projects, environment variables are sufficient:

export DATABASE_PASSWORD="dev_password_123"
python app.py

For production, use a secret manager:

import os
import hvac

# Fetch database credentials from Vault
client = hvac.Client(url=os.getenv("VAULT_ADDR"))
client.auth.approle.login(
    role_id=os.getenv("VAULT_ROLE_ID"),
    secret_id=os.getenv("VAULT_SECRET_ID")
)

# Request dynamic database credentials from Vault
db_creds = client.secrets.database.read_dynamic_credentials(
    name="myapp",
    mount_point="database"
)

db_user = db_creds["data"]["data"]["username"]
db_password = db_creds["data"]["data"]["password"]
db_host = os.getenv("DATABASE_HOST")

dsn = f"postgresql://{db_user}:{db_password}@{db_host}:5432/mydb"

With dynamic credentials, Vault creates a new temporary username and password for each request, automatically revokes it after a TTL, and logs all access.

Handling Connection Errors Without Leaking Credentials

When a database connection fails, never log the full DSN (it contains the password):

import psycopg2
import logging
import os

logger = logging.getLogger(__name__)

def get_db_connection():
    try:
        conn = psycopg2.connect(
            host=os.getenv("DATABASE_HOST"),
            user=os.getenv("DATABASE_USER"),
            password=os.getenv("DATABASE_PASSWORD"),
            database=os.getenv("DATABASE_NAME")
        )
        return conn
    except psycopg2.OperationalError as e:
        # WRONG: logs the connection string with password
        # logger.error(f"Connection failed: {e}")
        
        # CORRECT: log only safe information
        logger.error(
            f"Failed to connect to database {os.getenv('DATABASE_HOST')}. "
            f"Check DATABASE_USER, DATABASE_PASSWORD, and DATABASE_NAME."
        )
        raise

Create a custom logger that redacts sensitive keys:

import logging
import re

class RedactingFormatter(logging.Formatter):
    """Redact sensitive keys from log messages."""
    
    REDACTED = "***REDACTED***"
    PATTERNS = [
        r"password[=:]\s*[^\s]+",
        r"api_key[=:]\s*[^\s]+",
        r"token[=:]\s*[^\s]+",
        r"secret[=:]\s*[^\s]+",
    ]
    
    def format(self, record):
        msg = super().format(record)
        for pattern in self.PATTERNS:
            msg = re.sub(pattern, f"\\g<0>.split('=')[0]={self.REDACTED}", msg, flags=re.IGNORECASE)
        return msg

# Apply to all handlers
handler = logging.StreamHandler()
handler.setFormatter(RedactingFormatter())
logging.getLogger().addHandler(handler)

Database Credential Rotation Strategy

Database passwords should be rotated every 30–90 days. With static credentials, rotation requires updating code or redeploying. With dynamic credentials (Vault, AWS Secrets Manager), rotation is automatic:

1. Vault generates a new temporary credential.
2. Your application uses the new credential.
3. Vault automatically revokes the old credential after TTL.

For static credentials (when dynamic is not available), implement rotation manually:

import psycopg2
import os
from datetime import datetime, timedelta

def rotate_database_password():
    """
    Manually rotate the database password.
    Connect with old credentials, create new ones, update environment.
    """
    old_password = os.getenv("DATABASE_PASSWORD")
    old_user = os.getenv("DATABASE_USER")
    
    # Connect with old credentials
    conn = psycopg2.connect(
        host=os.getenv("DATABASE_HOST"),
        user=old_user,
        password=old_password,
        database="postgres"
    )
    
    cursor = conn.cursor()
    
    # Generate new password
    import secrets
    new_password = secrets.token_urlsafe(32)
    
    # Update password in database
    cursor.execute(
        f"ALTER USER {old_user} WITH PASSWORD %s",
        (new_password,)
    )
    conn.commit()
    cursor.close()
    conn.close()
    
    # Update environment and config
    os.environ["DATABASE_PASSWORD"] = new_password
    
    # Persist to secret manager or config service
    import boto3
    sm_client = boto3.client("secretsmanager")
    sm_client.update_secret(
        SecretId="prod/database/postgres",
        SecretString=f'{{"user": "{old_user}", "password": "{new_password}"}}'
    )
    
    print(f"Database password rotated at {datetime.now()}")

# Schedule rotation (e.g., via APScheduler)
from apscheduler.schedulers.background import BackgroundScheduler

scheduler = BackgroundScheduler()
scheduler.add_job(rotate_database_password, "cron", days=45)
scheduler.start()

Comparison: Credential Storage Methods

Method	Security	Rotation	Complexity	Best For
Hardcoded in code	Worst	Manual (redeploy)	Low	Never—don't do this
Environment variables	Good	Manual (redeploy)	Low	Development, small projects
.env file	Fair	Manual	Low	Local development
Secret Managers (static)	Excellent	Manual or automatic	Medium	Production (AWS, GCP)
Vault (dynamic)	Excellent	Automatic	High	Large teams, compliance

Key Takeaways

• Always build database DSNs from environment variables or secret managers, never hardcode credentials.
• Use connection pools to reuse connections, reducing credential exposure and improving performance.
• Redact passwords from error logs and implement a custom logger that filters sensitive keys.
• Use dynamic credentials (Vault, AWS RDS Proxy) in production to automatically rotate passwords without downtime.
• Test credential rotation by breaking the old credential and verifying the application switches to the new one.

Frequently Asked Questions

How do I safely pass database credentials to Docker containers?

Use Docker secrets (Swarm) or Kubernetes Secrets (mounted as environment variables or files), never pass them as command-line arguments or in Dockerfiles. For local development: pass via docker run -e DATABASE_PASSWORD=... or a .env file with docker-compose --env-file .env. Never commit .env to Git.

What if my database connection has special characters in the password?

Use urllib.parse.quote_plus() to escape the password when building the DSN: password = "p@ss:word"; safe = quote_plus(password). Alternatively, use a database driver that accepts credentials as separate arguments (not in the DSN): psycopg2.connect(host=..., password=...).

How do I know if a database credential has been compromised?

Check database logs for failed authentication attempts. Enable audit logging in your database (PostgreSQL: log_statement = "all") and forward logs to a SIEM (security information and event management system). AWS RDS CloudTrail logs, Vault audit logs, and Secrets Manager access logs provide visibility. If compromised, rotate immediately.

Can I use read-only database credentials for my application?

Yes, and it is highly recommended. Create a database user with only SELECT permissions, and use those credentials in your application. If the credential is compromised, an attacker can read data but cannot modify or delete it. Use separate read-write credentials only for admin operations.

How do I handle database connections in async Python (asyncio)?

Use asyncpg (PostgreSQL), aiomysql (MySQL), or motor (MongoDB) with their built-in connection pools. Example: pool = await asyncpg.create_pool(user=..., password=..., host=...). Pools work the same way but for async connections.

Further Reading

• PostgreSQL Connection Strings
• OWASP: Database Security
• SQLAlchemy Connection Pooling
• AWS RDS Proxy for Credential Rotation