Database Security

Protecting your data from threats, breaches, and compliance violations

Why Database Security is Non-Negotiable

A single SQL injection vulnerability exposed 143 million Equifax customer records. Capital One lost 100 million records because of misconfigured S3 permissions. Facebook paid $5 billion in GDPR fines for privacy violations. Database security isn't optional, it's the difference between a successful business and a catastrophic breach that destroys customer trust and triggers massive fines. Beyond attackers stealing credit cards or social security numbers, insecure databases face compliance violations (GDPR, HIPAA, SOC 2), data loss from ransomware, and privilege escalationwhere low-level users access sensitive data. This lesson covers SQL injection prevention(parameterized queries, ORMs), authentication & authorization (RBAC, row-level security), encryption (at rest with AES-256, in transit with TLS), audit logging for compliance, and secrets management(never hardcode passwords!). You'll learn the OWASP Top 10 database vulnerabilities, how to pass compliance audits, and production patterns from companies like Stripe and GitHub.

Real-World Impact:The average cost of a data breach in 2024 is $4.45 million. 60% of breaches involve database vulnerabilities. Companies that can't demonstrate GDPR/HIPAA compliance face fines up to 4% of annual revenue or $50,000 per violation. Security isn't just IT, it's business survival.

SQL Injection: The #1 Database Vulnerability

SQL injection occurs when user input is concatenated directly into SQL queries, allowing attackers to execute arbitrary SQL commands. It's been #1 on the OWASP Top 10 for over a decade.

❌ Vulnerable Code (NEVER DO THIS!)

# DANGEROUS: User input directly in SQL string
import psycopg2

conn = psycopg2.connect("dbname=mydb user=postgres")
cur = conn.cursor()

# User provides: username = "admin' OR '1'='1"
username = input("Enter username: ")
password = input("Enter password: ")

# String concatenation = SQL INJECTION VULNERABILITY
query = f"SELECT * FROM users WHERE username = '{username}' AND password = '{password}'"
cur.execute(query)

user = cur.fetchone()
if user:
    print("Login successful!")
else:
    print("Login failed")
Attack Scenario:
Input: username = "admin' OR '1'='1"
       password = "anything"

Generated SQL:
SELECT * FROM users WHERE username = 'admin' OR '1'='1' AND password = 'anything'

Result: Bypasses authentication! OR '1'='1' is always true, logs in as first user.

✅ Solution 1: Parameterized Queries (Prepared Statements)

# SECURE: Use parameterized queries
import psycopg2

conn = psycopg2.connect("dbname=mydb user=postgres")
cur = conn.cursor()

username = input("Enter username: ")
password = input("Enter password: ")

# Use %s placeholders and pass values separately
# Database driver escapes input automatically
cur.execute(
    "SELECT * FROM users WHERE username = %s AND password = %s",
    (username, password)  # Tuple of parameters
)

user = cur.fetchone()
if user:
    print("Login successful!")
else:
    print("Login failed")
Result: Attack input is treated as literal string, not SQL code. Even "admin' OR '1'='1" is safely escaped as a username search. SQL injection prevented!

✅ Solution 2: Use an ORM (SQLAlchemy)

# ORMs automatically use parameterized queries
from sqlalchemy import create_engine, Column, String, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    username = Column(String)
    password = Column(String)

engine = create_engine('postgresql://postgres@localhost/mydb')
Session = sessionmaker(bind=engine)
session = Session()

# ORM query - automatically parameterized
username = input("Enter username: ")
password = input("Enter password: ")

user = session.query(User).filter(
    User.username == username,
    User.password == password
).first()

if user:
    print(f"Login successful! Welcome {user.username}")
else:
    print("Login failed")
Result: SQLAlchemy automatically uses parameterized queries. No SQL injection possible. Plus: cleaner code, database-agnostic!

Other Injection Attack Vectors

  • ORDER BY injection: Never use user input in ORDER BY (use whitelist: allowed = ['name', 'date'])
  • LIMIT injection: Validate integers:int(user_input) before using in LIMIT
  • Table/column names: Can't be parameterized, use strict whitelists and validation
  • NoSQL injection: MongoDB, Redis also vulnerable (use parameterized queries in drivers)

Authentication & Authorization

Authentication = Who are you? Authorization = What can you do? Both are critical: authenticating users isn't enough if they can access data they shouldn't see.

Step 1: Never Store Plaintext Passwords

# Use bcrypt for password hashing
import bcrypt

# REGISTRATION: Hash password before storing
def register_user(username, password):
    # Generate salt and hash password
    salt = bcrypt.gensalt(rounds=12)  # 12 rounds = strong but not too slow
    password_hash = bcrypt.hashpw(password.encode('utf-8'), salt)

    # Store username and password_hash in database
    cur.execute(
        "INSERT INTO users (username, password_hash) VALUES (%s, %s)",
        (username, password_hash.decode('utf-8'))
    )
    conn.commit()
    print(f"User {username} registered successfully")

# LOGIN: Verify password against hash
def login_user(username, password):
    cur.execute(
        "SELECT password_hash FROM users WHERE username = %s",
        (username,)
    )
    result = cur.fetchone()

    if result and bcrypt.checkpw(password.encode('utf-8'), result[0].encode('utf-8')):
        print("Login successful!")
        return True
    else:
        print("Invalid credentials")
        return False
Result: Passwords are hashed with bcrypt (includes salt). Even if database is breached, passwords can't be reversed. Never use MD5/SHA1 (too fast, crackable with GPUs)!

Step 2: Role-Based Access Control (RBAC)

# Database schema for RBAC
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    username VARCHAR(255) UNIQUE,
    password_hash TEXT
);

CREATE TABLE roles (
    id SERIAL PRIMARY KEY,
    role_name VARCHAR(50) UNIQUE  -- 'admin', 'editor', 'viewer'
);

CREATE TABLE user_roles (
    user_id INT REFERENCES users(id),
    role_id INT REFERENCES roles(id),
    PRIMARY KEY (user_id, role_id)
);

CREATE TABLE permissions (
    id SERIAL PRIMARY KEY,
    permission_name VARCHAR(100)  -- 'read_posts', 'write_posts', 'delete_users'
);

CREATE TABLE role_permissions (
    role_id INT REFERENCES roles(id),
    permission_id INT REFERENCES permissions(id),
    PRIMARY KEY (role_id, permission_id)
);
Schema Design: Users have roles, roles have permissions. This allows flexible access control (add permissions to roles without touching code).

Step 3: Enforce RBAC in Application

# Check if user has permission
def has_permission(user_id, permission_name):
    cur.execute("""
        SELECT COUNT(*) FROM permissions p
        JOIN role_permissions rp ON p.id = rp.permission_id
        JOIN user_roles ur ON rp.role_id = ur.role_id
        WHERE ur.user_id = %s AND p.permission_name = %s
    """, (user_id, permission_name))

    count = cur.fetchone()[0]
    return count > 0

# Decorator for permission-based access control
from functools import wraps

def require_permission(permission):
    def decorator(func):
        @wraps(func)
        def wrapper(user_id, *args, **kwargs):
            if not has_permission(user_id, permission):
                raise PermissionError(f"User lacks '{permission}' permission")
            return func(user_id, *args, **kwargs)
        return wrapper
    return decorator

# Usage
@require_permission('delete_users')
def delete_user(user_id, target_user_id):
    cur.execute("DELETE FROM users WHERE id = %s", (target_user_id,))
    conn.commit()
    print(f"User {target_user_id} deleted")

# Attempt deletion
try:
    delete_user(user_id=5, target_user_id=10)
except PermissionError as e:
    print(f"Access denied: {e}")
Result: Only users with 'delete_users' permission can execute the function. Others get PermissionError.

Step 4: Row-Level Security (PostgreSQL RLS)

-- Enable Row-Level Security on table
ALTER TABLE documents ENABLE ROW LEVEL SECURITY;

-- Create policy: users can only see their own documents
CREATE POLICY user_documents ON documents
    FOR SELECT
    USING (user_id = current_user_id());

-- Create policy: users can only update their own documents
CREATE POLICY user_documents_update ON documents
    FOR UPDATE
    USING (user_id = current_user_id());

-- Function to get current user ID from session
CREATE OR REPLACE FUNCTION current_user_id()
RETURNS INTEGER AS $$
BEGIN
    RETURN current_setting('app.user_id')::INTEGER;
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;
# Python application sets user context
import psycopg2

conn = psycopg2.connect("dbname=mydb user=postgres")
cur = conn.cursor()

# Set current user for this session
user_id = 42
cur.execute("SET app.user_id = %s", (user_id,))

# Query automatically filtered by RLS
cur.execute("SELECT * FROM documents")
results = cur.fetchall()

# Returns only documents where user_id = 42
# Even if query has no WHERE clause!
print(f"Found {len(results)} documents for user {user_id}")
Result: Database enforces access control at the row level. Even if application has a bug, users can't access other users' data. Defense in depth!

Encryption: At Rest & In Transit

Encryption protects data from attackers who gain physical access (stolen drives, cloud snapshots) or intercept network traffic (man-in-the-middle attacks).

Encryption at Rest (Database Files)

Transparent Data Encryption (TDE)

Database automatically encrypts data files with AES-256. No application changes needed.

PostgreSQL: pgcrypto extension
MySQL: InnoDB encryption
SQL Server: TDE built-in

Application-Level Encryption

Encrypt sensitive fields before INSERT. More control but requires key management.

Use case: SSN, credit cards, PII
Library: cryptography (Python)
Algorithm: AES-256-GCM
# Application-level encryption with Fernet (AES-128)
from cryptography.fernet import Fernet
import os

# Generate encryption key (store in secrets manager, NOT in code!)
key = Fernet.generate_key()
cipher = Fernet(key)

# Encrypt sensitive data before storing
def store_sensitive_data(user_id, ssn):
    encrypted_ssn = cipher.encrypt(ssn.encode())

    cur.execute(
        "INSERT INTO user_data (user_id, encrypted_ssn) VALUES (%s, %s)",
        (user_id, encrypted_ssn)
    )
    conn.commit()
    print("Data stored with encryption")

# Decrypt when retrieving
def get_sensitive_data(user_id):
    cur.execute(
        "SELECT encrypted_ssn FROM user_data WHERE user_id = %s",
        (user_id,)
    )
    encrypted_ssn = cur.fetchone()[0]
    decrypted_ssn = cipher.decrypt(encrypted_ssn).decode()
    return decrypted_ssn

# Usage
store_sensitive_data(user_id=1, ssn="123-45-6789")
ssn = get_sensitive_data(user_id=1)
print(f"Retrieved SSN: {ssn}")
Result: SSN stored as encrypted bytes. Even if database is compromised, data is unreadable without encryption key.

Encryption in Transit (TLS/SSL)

# PostgreSQL connection with SSL/TLS
import psycopg2

conn = psycopg2.connect(
    host="db.example.com",
    port=5432,
    dbname="mydb",
    user="postgres",
    password="secure_password",
    sslmode="require",  # Force SSL/TLS
    sslrootcert="/path/to/server-ca.pem",  # Verify server certificate
    sslcert="/path/to/client-cert.pem",    # Client certificate (mutual TLS)
    sslkey="/path/to/client-key.pem"
)

print("Connected via TLS")
Result: All data between application and database is encrypted with TLS 1.2+. Prevents eavesdropping and man-in-the-middle attacks.

Encryption Key Management Best Practices

  • Never hardcode keys: Use environment variables or secrets managers (AWS KMS, HashiCorp Vault)
  • Rotate keys regularly: Re-encrypt data with new keys every 90-365 days
  • Separate key storage: Keys in different location than encrypted data
  • Use key hierarchies: Master key encrypts data keys (envelope encryption)

Audit Logging & Compliance

Audit logs track who accessed what data and when. Required for compliance (GDPR, HIPAA, SOC 2) and forensic analysis after security incidents.

Step 1: Create Audit Log Table

-- Comprehensive audit log table
CREATE TABLE audit_log (
    id SERIAL PRIMARY KEY,
    timestamp TIMESTAMP DEFAULT NOW(),
    user_id INTEGER,
    username VARCHAR(255),
    action VARCHAR(50),  -- 'SELECT', 'INSERT', 'UPDATE', 'DELETE', 'LOGIN'
    table_name VARCHAR(100),
    record_id INTEGER,
    old_values JSONB,
    new_values JSONB,
    ip_address INET,
    user_agent TEXT,
    success BOOLEAN
);

-- Index for fast queries
CREATE INDEX idx_audit_timestamp ON audit_log(timestamp);
CREATE INDEX idx_audit_user ON audit_log(user_id);
CREATE INDEX idx_audit_table ON audit_log(table_name, record_id);
Schema Design: Captures who, what, when, where, and before/after state. JSONB stores flexible data.

Step 2: Implement Audit Logging

import json
from datetime import datetime

def log_audit(user_id, username, action, table_name, record_id=None,
              old_values=None, new_values=None, ip_address=None, success=True):
    cur.execute("""
        INSERT INTO audit_log
        (user_id, username, action, table_name, record_id,
         old_values, new_values, ip_address, success)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
    """, (
        user_id, username, action, table_name, record_id,
        json.dumps(old_values) if old_values else None,
        json.dumps(new_values) if new_values else None,
        ip_address, success
    ))
    conn.commit()

# Example: Log user data update
def update_user_email(user_id, new_email, current_user_id, ip_address):
    # Get old value
    cur.execute("SELECT email FROM users WHERE id = %s", (user_id,))
    old_email = cur.fetchone()[0]

    # Perform update
    cur.execute(
        "UPDATE users SET email = %s WHERE id = %s",
        (new_email, user_id)
    )
    conn.commit()

    # Log the change
    log_audit(
        user_id=current_user_id,
        username="admin",
        action="UPDATE",
        table_name="users",
        record_id=user_id,
        old_values={"email": old_email},
        new_values={"email": new_email},
        ip_address=ip_address,
        success=True
    )

    print(f"Email updated and logged")

# Usage
update_user_email(user_id=10, new_email="new@example.com",
                  current_user_id=1, ip_address="192.168.1.100")
Result: Every data change is logged with who made it, what changed, and when. Perfect for compliance audits!

Step 3: Automatic Audit Logging with Triggers

-- Trigger function to log all changes
CREATE OR REPLACE FUNCTION audit_trigger_func()
RETURNS TRIGGER AS $$
BEGIN
    IF TG_OP = 'INSERT' THEN
        INSERT INTO audit_log (action, table_name, record_id, new_values)
        VALUES ('INSERT', TG_TABLE_NAME, NEW.id, row_to_json(NEW));
        RETURN NEW;

    ELSIF TG_OP = 'UPDATE' THEN
        INSERT INTO audit_log (action, table_name, record_id, old_values, new_values)
        VALUES ('UPDATE', TG_TABLE_NAME, NEW.id, row_to_json(OLD), row_to_json(NEW));
        RETURN NEW;

    ELSIF TG_OP = 'DELETE' THEN
        INSERT INTO audit_log (action, table_name, record_id, old_values)
        VALUES ('DELETE', TG_TABLE_NAME, OLD.id, row_to_json(OLD));
        RETURN OLD;
    END IF;
END;
$$ LANGUAGE plpgsql;

-- Attach trigger to sensitive tables
CREATE TRIGGER audit_users
    AFTER INSERT OR UPDATE OR DELETE ON users
    FOR EACH ROW EXECUTE FUNCTION audit_trigger_func();

CREATE TRIGGER audit_transactions
    AFTER INSERT OR UPDATE OR DELETE ON transactions
    FOR EACH ROW EXECUTE FUNCTION audit_trigger_func();
Result: Database automatically logs all changes to sensitive tables. No application code needed, can't be bypassed!

Compliance Requirements (GDPR, HIPAA, SOC 2)

StandardKey RequirementsDatabase Controls
GDPR
(EU Privacy)		• Right to be forgotten
• Data portability
• Breach notification (72hr)
• Consent tracking		• Audit logs (who accessed PII)
• Data deletion procedures
• Encryption at rest
• Export capabilities
HIPAA
(US Healthcare)		• PHI protection
• Access controls
• Audit trails
• Encryption required		• RBAC for patient data
• Audit log retention (6 years)
• TLS for all connections
• Automatic session timeout
SOC 2
(Security/Privacy)		• Access reviews
• Change management
• Incident response
• Monitoring		• Quarterly access audits
• Database change logs
• Anomaly detection alerts
• Security patch management

GDPR Example: Right to be Forgotten

# Complete user data deletion (GDPR compliance)
def gdpr_delete_user(user_id, admin_user_id, reason):
    # Log deletion request
    log_audit(
        user_id=admin_user_id,
        username="admin",
        action="GDPR_DELETE",
        table_name="users",
        record_id=user_id,
        new_values={"reason": reason},
        success=False  # Set to True after completion
    )

    # Delete user data across all tables
    tables_to_clean = [
        "user_preferences",
        "user_sessions",
        "user_comments",
        "user_orders",
        "users"
    ]

    for table in tables_to_clean:
        cur.execute(f"DELETE FROM {table} WHERE user_id = %s", (user_id,))
        print(f"Deleted from {table}")

    conn.commit()

    # Update audit log to mark success
    log_audit(
        user_id=admin_user_id,
        username="admin",
        action="GDPR_DELETE",
        table_name="users",
        record_id=user_id,
        new_values={"reason": reason},
        success=True
    )

    print(f"User {user_id} completely deleted (GDPR compliant)")

# Usage
gdpr_delete_user(
    user_id=123,
    admin_user_id=1,
    reason="User requested deletion via GDPR form"
)
Result: All user data deleted across tables. Audit log shows who approved deletion and why. GDPR compliant!

Secrets Management

Database credentials, API keys, and encryption keys must NEVER be hardcoded or committed to version control. Use secrets managers and environment variables.

❌ What NOT to Do

Common Mistakes (DON'T DO THESE!)

1. Hardcoded credentials in code:

# DANGEROUS!
conn = psycopg2.connect(
    host="prod-db.company.com",
    user="postgres",
    password="SuperSecret123!"  # ← Will be in Git history forever!
)

2. Credentials in config files committed to Git:

# config.json (committed to repo)
{
  "db_password": "prod_password_123"  # ← Exposed to all devs
}

3. Credentials in environment variables on laptops:

Better than hardcoding, but still risky if laptop is compromised

✅ Solution 1: Environment Variables (Basic)

# .env file (add to .gitignore!)
DB_HOST=localhost
DB_PORT=5432
DB_NAME=mydb
DB_USER=postgres
DB_PASSWORD=secure_password_here

# Load with python-dotenv
from dotenv import load_dotenv
import os

load_dotenv()

conn = psycopg2.connect(
    host=os.getenv('DB_HOST'),
    port=os.getenv('DB_PORT'),
    dbname=os.getenv('DB_NAME'),
    user=os.getenv('DB_USER'),
    password=os.getenv('DB_PASSWORD')
)

print("Connected using environment variables")
Result: Credentials not in code. Different environments (dev/staging/prod) use different .env files. Important: Add .env to .gitignore!

✅ Solution 2: AWS Secrets Manager (Production)

# Install: pip install boto3

import boto3
import json
from botocore.exceptions import ClientError

def get_secret(secret_name, region_name="us-east-1"):
    # Create a Secrets Manager client
    session = boto3.session.Session()
    client = session.client(
        service_name='secretsmanager',
        region_name=region_name
    )

    try:
        response = client.get_secret_value(SecretId=secret_name)
        secret = json.loads(response['SecretString'])
        return secret
    except ClientError as e:
        raise e

# Get database credentials from Secrets Manager
db_secret = get_secret("prod/database/credentials")

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

print("Connected using AWS Secrets Manager")
Result: Credentials stored encrypted in AWS. Automatic rotation supported. Audit logs show who accessed secrets. Production-grade!

✅ Solution 3: HashiCorp Vault (Enterprise)

# Install: pip install hvac

import hvac

# Authenticate to Vault
client = hvac.Client(url='https://vault.company.com')
client.auth.approle.login(
    role_id='your-role-id',
    secret_id='your-secret-id'
)

# Read database credentials
secret = client.secrets.kv.v2.read_secret_version(
    path='database/postgres/prod'
)

db_creds = secret['data']['data']

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

print("Connected using HashiCorp Vault")
Result: Centralized secrets management. Dynamic credentials (Vault generates short-lived DB passwords). Full audit trail. Enterprise-grade security!

Secrets Management Best Practices

  • Never commit secrets to Git: Use .gitignore for .env files, scan repos with tools like git-secrets
  • Rotate credentials regularly: Database passwords every 90 days, API keys every 6-12 months
  • Use different credentials per environment: Dev, staging, prod should never share passwords
  • Least privilege: Application should have minimum permissions needed (not superuser!)
  • Audit access: Log who retrieved secrets and when

Database Security Checklist

✅ Application Security

  • Use parameterized queries (no SQL injection)
  • Hash passwords with bcrypt/scrypt
  • Implement RBAC/permissions
  • Validate all user input
  • Use ORMs when possible
  • Implement rate limiting

✅ Database Configuration

  • Enable TLS/SSL for connections
  • Disable remote root login
  • Configure firewall rules (whitelist IPs)
  • Enable audit logging
  • Set up automated backups
  • Apply security patches promptly

✅ Access Control

  • Principle of least privilege
  • Separate users per application/service
  • Revoke unused accounts
  • Implement row-level security
  • Multi-factor authentication for admins
  • Quarterly access reviews

✅ Monitoring & Response

  • Monitor failed login attempts
  • Alert on unusual query patterns
  • Track privilege escalations
  • Log all schema changes
  • Test incident response plan
  • Regular security assessments

Key Takeaways

  • SQL injection is preventable: Always use parameterized queries or ORMs. Never concatenate user input into SQL strings.
  • Defense in depth: Combine application security (RBAC, input validation) with database security (row-level security, encryption).
  • Encrypt everything: Data at rest (TDE or app-level), data in transit (TLS), and backups.
  • Audit trails are mandatory: Log all access to sensitive data for compliance (GDPR, HIPAA) and incident response.
  • Secrets management is critical: Never hardcode credentials. Use secrets managers (AWS Secrets Manager, HashiCorp Vault) in production.
Databases AdvancedLesson 21