PostgreSQL Security Best Practices¶

# encryption/field_encryption.py
import psycopg2
from cryptography.fernet import Fernet
import base64
import os

class PostgreSQLFieldEncryption:
    def __init__(self, encryption_key=None):
        if encryption_key:
            self.cipher = Fernet(encryption_key)
        else:
            # Generate new key
            key = Fernet.generate_key()
            self.cipher = Fernet(key)
            print(f"Generated encryption key: {key.decode()}")

    def encrypt_field(self, value):
        """Encrypt a field value."""
        if value is None:
            return None
        encrypted = self.cipher.encrypt(value.encode())
        return base64.b64encode(encrypted).decode()

    def decrypt_field(self, encrypted_value):
        """Decrypt a field value."""
        if encrypted_value is None:
            return None
        try:
            encrypted = base64.b64decode(encrypted_value.encode())
            decrypted = self.cipher.decrypt(encrypted)
            return decrypted.decode()
        except Exception as e:
            print(f"Decryption error: {e}")
            return None

    def create_encrypted_table(self, connection):
        """Create table with encrypted fields."""
        with connection.cursor() as cur:
            cur.execute("""
                CREATE TABLE IF NOT EXISTS encrypted_users (
                    id SERIAL PRIMARY KEY,
                    username VARCHAR(50),
                    email VARCHAR(100),
                    encrypted_ssn TEXT,
                    encrypted_credit_card TEXT,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                );
            """)
            connection.commit()

    def insert_encrypted_data(self, connection, username, email, ssn, credit_card):
        """Insert data with encrypted fields."""
        encrypted_ssn = self.encrypt_field(ssn)
        encrypted_credit_card = self.encrypt_field(credit_card)

        with connection.cursor() as cur:
            cur.execute("""
                INSERT INTO encrypted_users (username, email, encrypted_ssn, encrypted_credit_card)
                VALUES (%s, %s, %s, %s)
            """, (username, email, encrypted_ssn, encrypted_credit_card))
            connection.commit()

    def retrieve_encrypted_data(self, connection, user_id):
        """Retrieve and decrypt data."""
        with connection.cursor() as cur:
            cur.execute("""
                SELECT id, username, email, encrypted_ssn, encrypted_credit_card
                FROM encrypted_users WHERE id = %s
            """, (user_id,))

            row = cur.fetchone()
            if row:
                return {
                    'id': row[0],
                    'username': row[1],
                    'email': row[2],
                    'ssn': self.decrypt_field(row[3]),
                    'credit_card': self.decrypt_field(row[4])
                }
            return None

# Usage
encryption = PostgreSQLFieldEncryption()
conn = psycopg2.connect(
    host='localhost',
    database='production',
    user='app_user',
    password='app_password'
)

encryption.create_encrypted_table(conn)
encryption.insert_encrypted_data(conn, 'user1', 'user1@example.com', '123-45-6789', '4111-1111-1111-1111')
user_data = encryption.retrieve_encrypted_data(conn, 1)
print(user_data)

# monitoring/security_monitor.py
import psycopg2
import json
import time
from datetime import datetime, timedelta
import smtplib
from email.mime.text import MIMEText

class PostgreSQLSecurityMonitor:
    def __init__(self, connection_params, alert_config):
        self.conn_params = connection_params
        self.alert_config = alert_config
        self.conn = None

    def connect(self):
        """Connect to PostgreSQL."""
        self.conn = psycopg2.connect(**self.conn_params)

    def check_failed_logins(self, hours=1):
        """Check for failed login attempts."""
        with self.conn.cursor() as cur:
            cur.execute("""
                SELECT 
                    usename,
                    client_addr,
                    application_name,
                    count(*) as failed_attempts
                FROM pg_stat_activity 
                WHERE state = 'idle in transaction (aborted)'
                AND query_start > NOW() - INTERVAL '%s hours'
                GROUP BY usename, client_addr, application_name
                HAVING count(*) > 5
                ORDER BY failed_attempts DESC;
            """, (hours,))
            return cur.fetchall()

    def check_privilege_escalation(self):
        """Check for privilege escalation attempts."""
        with self.conn.cursor() as cur:
            cur.execute("""
                SELECT 
                    usename,
                    client_addr,
                    application_name,
                    query
                FROM pg_stat_activity 
                WHERE query ILIKE '%GRANT%' 
                OR query ILIKE '%REVOKE%'
                OR query ILIKE '%ALTER USER%'
                OR query ILIKE '%CREATE USER%'
                ORDER BY query_start DESC;
            """)
            return cur.fetchall()

    def check_suspicious_queries(self):
        """Check for suspicious query patterns."""
        with self.conn.cursor() as cur:
            cur.execute("""
                SELECT 
                    usename,
                    client_addr,
                    application_name,
                    query,
                    query_start
                FROM pg_stat_activity 
                WHERE query ILIKE '%DROP%'
                OR query ILIKE '%TRUNCATE%'
                OR query ILIKE '%DELETE FROM%'
                OR query ILIKE '%UPDATE%'
                ORDER BY query_start DESC;
            """)
            return cur.fetchall()

    def check_audit_log_anomalies(self, hours=24):
        """Check audit log for anomalies."""
        with self.conn.cursor() as cur:
            cur.execute("""
                SELECT 
                    changed_by,
                    client_ip,
                    application_name,
                    count(*) as operations
                FROM audit.audit_log 
                WHERE changed_at > NOW() - INTERVAL '%s hours'
                GROUP BY changed_by, client_ip, application_name
                HAVING count(*) > 100
                ORDER BY operations DESC;
            """, (hours,))
            return cur.fetchall()

    def send_security_alert(self, subject, message):
        """Send security alert via email."""
        try:
            msg = MIMEText(message)
            msg['Subject'] = subject
            msg['From'] = self.alert_config['from_email']
            msg['To'] = self.alert_config['to_email']

            server = smtplib.SMTP(self.alert_config['smtp_server'], self.alert_config['smtp_port'])
            server.starttls()
            server.login(self.alert_config['username'], self.alert_config['password'])
            server.send_message(msg)
            server.quit()
            print(f"Security alert sent: {subject}")
        except Exception as e:
            print(f"Failed to send alert: {e}")

    def run_security_checks(self):
        """Run all security checks."""
        alerts = []

        # Check failed logins
        failed_logins = self.check_failed_logins()
        if failed_logins:
            alerts.append({
                'type': 'failed_logins',
                'message': f"Multiple failed login attempts detected: {failed_logins}",
                'severity': 'high'
            })

        # Check privilege escalation
        privilege_attempts = self.check_privilege_escalation()
        if privilege_attempts:
            alerts.append({
                'type': 'privilege_escalation',
                'message': f"Privilege escalation attempts detected: {privilege_attempts}",
                'severity': 'critical'
            })

        # Check suspicious queries
        suspicious_queries = self.check_suspicious_queries()
        if suspicious_queries:
            alerts.append({
                'type': 'suspicious_queries',
                'message': f"Suspicious queries detected: {suspicious_queries}",
                'severity': 'medium'
            })

        # Send alerts
        for alert in alerts:
            self.send_security_alert(
                f"PostgreSQL Security Alert: {alert['type']}",
                alert['message']
            )

        return alerts

# Usage
if __name__ == "__main__":
    monitor = PostgreSQLSecurityMonitor(
        connection_params={
            'host': 'localhost',
            'database': 'production',
            'user': 'security_monitor',
            'password': 'monitor_password'
        },
        alert_config={
            'from_email': 'alerts@company.com',
            'to_email': 'security@company.com',
            'smtp_server': 'smtp.company.com',
            'smtp_port': 587,
            'username': 'alerts@company.com',
            'password': 'smtp_password'
        }
    )

    monitor.connect()
    alerts = monitor.run_security_checks()
    print(f"Security check completed. {len(alerts)} alerts generated.")

# anonymization/data_anonymizer.py
import psycopg2
import random
import string
from faker import Faker

class PostgreSQLDataAnonymizer:
    def __init__(self, connection_params):
        self.conn_params = connection_params
        self.fake = Faker()
        self.conn = None

    def connect(self):
        """Connect to PostgreSQL."""
        self.conn = psycopg2.connect(**self.conn_params)

    def anonymize_users(self):
        """Anonymize user data."""
        with self.conn.cursor() as cur:
            # Get all users
            cur.execute("SELECT id, username, email FROM users")
            users = cur.fetchall()

            for user_id, username, email in users:
                # Generate fake data
                fake_username = self.fake.user_name()
                fake_email = self.fake.email()

                # Update user data
                cur.execute("""
                    UPDATE users 
                    SET username = %s, email = %s
                    WHERE id = %s
                """, (fake_username, fake_email, user_id))

            self.conn.commit()
            print(f"Anonymized {len(users)} users")

    def anonymize_posts(self):
        """Anonymize post content."""
        with self.conn.cursor() as cur:
            # Get all posts
            cur.execute("SELECT id, title, content FROM posts")
            posts = cur.fetchall()

            for post_id, title, content in posts:
                # Generate fake content
                fake_title = self.fake.sentence(nb_words=6)
                fake_content = self.fake.text(max_nb_chars=500)

                # Update post data
                cur.execute("""
                    UPDATE posts 
                    SET title = %s, content = %s
                    WHERE id = %s
                """, (fake_title, fake_content, post_id))

            self.conn.commit()
            print(f"Anonymized {len(posts)} posts")

    def create_anonymized_copy(self, source_db, target_db):
        """Create anonymized copy of database."""
        with self.conn.cursor() as cur:
            # Create target database
            cur.execute(f"CREATE DATABASE {target_db}")
            self.conn.commit()

        # Connect to target database
        target_conn = psycopg2.connect(
            host=self.conn_params['host'],
            database=target_db,
            user=self.conn_params['user'],
            password=self.conn_params['password']
        )

        with target_conn.cursor() as cur:
            # Copy schema
            cur.execute(f"CREATE SCHEMA IF NOT EXISTS public")

            # Copy tables structure
            cur.execute("""
                SELECT table_name FROM information_schema.tables 
                WHERE table_schema = 'public'
            """)
            tables = cur.fetchall()

            for table_name, in tables:
                # Create table in target database
                cur.execute(f"CREATE TABLE {table_name} (LIKE {source_db}.{table_name})")

            target_conn.commit()

        target_conn.close()
        print(f"Created anonymized copy: {target_db}")

# Usage
if __name__ == "__main__":
    anonymizer = PostgreSQLDataAnonymizer({
        'host': 'localhost',
        'database': 'production',
        'user': 'anonymizer',
        'password': 'anonymizer_password'
    })

    anonymizer.connect()
    anonymizer.anonymize_users()
    anonymizer.anonymize_posts()
    anonymizer.create_anonymized_copy('production', 'production_anonymized')

# Complete PostgreSQL security setup
def setup_postgresql_security():
    # 1. Authentication & authorization
    # 2. Network security
    # 3. Data encryption
    # 4. Audit logging
    # 5. Data masking
    # 6. Security monitoring
    # 7. Compliance
    # 8. Incident response

    print("PostgreSQL security setup complete!")