Database Security

# Install: pip install bcrypt "psycopg[binary]"
import bcrypt
import psycopg

conn = psycopg.connect("host=localhost dbname=demo user=demo password=demo")
cur = conn.cursor()


# 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. The plaintext password is never
    # stored, and never leaves this function.
    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()

    # An unknown user and a user with no hash both land here. Treat them the
    # same as a wrong password: never reveal which of the three it was.
    if not result or result[0] is None:
        print("Invalid credentials")
        return False

    if bcrypt.checkpw(password.encode('utf-8'), result[0].encode('utf-8')):
        print("Login successful!")
        return True

    print("Invalid credentials")
    return False


register_user("carol", "correct horse battery staple")
login_user("carol", "correct horse battery staple")
login_user("carol", "wrong password")
login_user("nobody", "whatever")

cur.execute("SELECT password_hash FROM users WHERE username = 'carol'")
print("\nStored value:", cur.fetchone()[0])

-- Database schema for RBAC. "users" already exists from the setup above,
-- so this adds only the tables around it.
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 alone proves nothing: with no rows, every permission check
-- returns false and the next example can only demonstrate denial.
INSERT INTO roles (role_name) VALUES ('admin'), ('editor'), ('viewer');
INSERT INTO permissions (permission_name) VALUES
    ('read_posts'), ('write_posts'), ('delete_users');

-- admin gets everything, editor can read and write, viewer only reads.
INSERT INTO role_permissions (role_id, permission_id)
SELECT r.id, p.id FROM roles r, permissions p
WHERE (r.role_name = 'admin')
   OR (r.role_name = 'editor' AND p.permission_name IN ('read_posts', 'write_posts'))
   OR (r.role_name = 'viewer' AND p.permission_name = 'read_posts');

-- A spare account for the deletion demo further down.
INSERT INTO users (username, email) VALUES ('dave', 'dave@example.com');

INSERT INTO user_roles (user_id, role_id)
SELECT u.id, r.id FROM users u, roles r
WHERE (u.username = 'admin' AND r.role_name = 'admin')
   OR (u.username = 'alice' AND r.role_name = 'editor')
   OR (u.username = 'bob'   AND r.role_name = 'viewer');

SELECT u.username, r.role_name, p.permission_name
FROM users u
JOIN user_roles ur       ON ur.user_id = u.id
JOIN roles r             ON r.id = ur.role_id
JOIN role_permissions rp ON rp.role_id = r.id
JOIN permissions p       ON p.id = rp.permission_id
ORDER BY u.username, p.permission_name;

from functools import wraps

import psycopg

conn = psycopg.connect("host=localhost dbname=demo user=demo password=demo")
cur = conn.cursor()


def user_id_for(username):
    cur.execute("SELECT id FROM users WHERE username = %s", (username,))
    return cur.fetchone()[0]


# 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
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


@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")


# Look the ids up rather than hardcoding them: they depend on how many
# times you have run the earlier blocks.
admin_id = user_id_for("admin")
alice_id = user_id_for("alice")
dave_id = user_id_for("dave")

print("admin can delete_users:", has_permission(admin_id, "delete_users"))
print("alice can delete_users:", has_permission(alice_id, "delete_users"))
print("alice can write_posts: ", has_permission(alice_id, "write_posts"))
print()

# The editor is refused before the DELETE is ever issued.
try:
    delete_user(user_id=alice_id, target_user_id=dave_id)
except PermissionError as e:
    print(f"Access denied: {e}")

# The admin is allowed through.
delete_user(user_id=admin_id, target_user_id=dave_id)

# Python application sets user context
import psycopg

# Connect as the dedicated non-superuser role, NOT as the "demo" superuser
conn = psycopg.connect("host=localhost dbname=demo user=app_user password=app_password")
cur = conn.cursor()

# Set current user for this session.
# Note: plain SET does not accept bind parameters ("SET app.user_id = %s"
# raises a syntax error). Use the set_config() function instead, which does.
user_id = 42
cur.execute("SELECT set_config('app.user_id', %s, false)", (str(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}")

# Install: pip install cryptography "psycopg[binary]"
# Application-level encryption with Fernet (AES-128)
import psycopg
from cryptography.fernet import Fernet

conn = psycopg.connect("host=localhost dbname=demo user=demo password=demo")
cur = conn.cursor()

cur.execute("CREATE TABLE IF NOT EXISTS user_data (user_id INT, encrypted_ssn BYTEA)")
conn.commit()

# Generate encryption key (store in a secrets manager, NOT in code, and NOT
# regenerated per run: a new key makes every existing row undecryptable).
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}")

# What an attacker with a database dump actually gets:
cur.execute("SELECT encrypted_ssn FROM user_data WHERE user_id = 1")
print("On disk:", bytes(cur.fetchone()[0])[:40])

# The pg-demo container from the top of this lesson ships with TLS off, so
# turn it on. A self-signed certificate is fine for a local demo and is
# exactly what a real deployment must NOT use (see below).

# The alpine image has no openssl binary, so add it, then generate a cert
# INSIDE the data directory where the postgres user already owns the files.
docker exec -u root pg-demo sh -c "apk add --no-cache openssl && \
  openssl req -new -x509 -days 365 -nodes -text \
    -out /var/lib/postgresql/data/server.crt \
    -keyout /var/lib/postgresql/data/server.key \
    -subj '/CN=localhost' && \
  chmod 600 /var/lib/postgresql/data/server.key && \
  chown postgres:postgres /var/lib/postgresql/data/server.key \
                          /var/lib/postgresql/data/server.crt"

# PostgreSQL reads the certificate at startup, so this needs a restart, not
# a reload. Get the cert wrong and the server refuses to come back up with
# "could not load server certificate file".
docker exec pg-demo psql -U demo -d demo -c "ALTER SYSTEM SET ssl = on;"
docker restart pg-demo

docker exec pg-demo psql -U demo -d demo -c "SHOW ssl;"

# PostgreSQL connection with SSL/TLS
import psycopg

conn = psycopg.connect(
    host="localhost",
    port=5432,
    dbname="demo",
    user="demo",
    password="demo",
    sslmode="require",  # Refuse to connect unless the link is encrypted
)

# Ask the server, not the client: pg_stat_ssl reports what this very
# backend negotiated.
with conn.cursor() as cur:
    cur.execute("""
        SELECT ssl, version, cipher
        FROM pg_stat_ssl
        WHERE pid = pg_backend_pid()
    """)
    ssl, version, cipher = cur.fetchone()
    print(f"ssl={ssl} version={version} cipher={cipher}")

conn.close()

# sslmode="require" encrypts, but does NOT check who is on the other end.
# verify-full is what authenticates the server, and it rejects this
# self-signed certificate because no trusted CA vouches for it.
try:
    psycopg.connect(
        host="localhost", port=5432, dbname="demo",
        user="demo", password="demo", sslmode="verify-full",
    )
    print("verify-full: connected")
except Exception as e:
    print("verify-full rejected it:", str(e).strip().splitlines()[0])

import json

import psycopg

conn = psycopg.connect("host=localhost dbname=demo user=demo password=demo")
cur = conn.cursor()


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("Email updated and logged")


# Usage. Look the id up: hardcoding one that does not exist makes
# fetchone() return None and the whole thing dies on a TypeError.
cur.execute("SELECT id FROM users WHERE username = 'alice'")
alice_id = cur.fetchone()[0]

update_user_email(user_id=alice_id, new_email="alice@newdomain.com",
                  current_user_id=1, ip_address="192.168.1.100")

cur.execute("""
    SELECT username, action, table_name, record_id,
           old_values, new_values, ip_address, success
    FROM audit_log ORDER BY id DESC LIMIT 1
""")
row = cur.fetchone()
print("\nAudit row:")
for name, value in zip(
    ["username", "action", "table", "record_id", "old", "new", "ip", "success"], row
):
    print(f"  {name:<10} {value}")

-- 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;

-- The function is table-agnostic: TG_TABLE_NAME and row_to_json() mean it
-- works on anything with an "id" column. A second table shows that off,
-- and nothing earlier in this lesson created one.
CREATE TABLE transactions (
    id      SERIAL PRIMARY KEY,
    user_id INTEGER REFERENCES users(id),
    amount  NUMERIC(10,2)
);

-- 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();

-- Three ordinary statements. No application code calls log_audit here.
INSERT INTO transactions (user_id, amount) VALUES (2, 99.50);
UPDATE transactions SET amount = 120.00 WHERE id = 1;
DELETE FROM transactions WHERE id = 1;

SELECT action, table_name, record_id,
       old_values->>'amount' AS old_amount,
       new_values->>'amount' AS new_amount
FROM audit_log
WHERE table_name = 'transactions'
ORDER BY id;

# Complete user data deletion (GDPR compliance).
# Reuses log_audit() and the conn/cur from the audit-logging block above.
import json

import psycopg

conn = psycopg.connect("host=localhost dbname=demo user=demo password=demo")
cur = conn.cursor()


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()


# Tables that reference users(id), in the order they must be cleared:
# children first, or the foreign keys reject the delete.
CHILD_TABLES = ["user_roles", "transactions"]


def gdpr_delete_user(user_id, admin_user_id, reason):
    # Log the request BEFORE deleting, with success=False. If the delete
    # fails halfway, the request itself is still on record.
    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
    )

    for table in CHILD_TABLES:
        # The table name is interpolated because it cannot be a bind
        # parameter. It is safe ONLY because CHILD_TABLES is a fixed list
        # in this file; never build this from user input.
        cur.execute(f"DELETE FROM {table} WHERE user_id = %s", (user_id,))
        print(f"Deleted {cur.rowcount} row(s) from {table}")

    # users is keyed by "id", not "user_id": the same WHERE clause does not
    # work on the parent table.
    cur.execute("DELETE FROM users WHERE id = %s", (user_id,))
    print(f"Deleted {cur.rowcount} row(s) from users")
    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)")


# Build a user with related rows so the deletion has something to clear.
cur.execute("INSERT INTO users (username, email) VALUES ('erin', 'erin@example.com') RETURNING id")
erin_id = cur.fetchone()[0]
cur.execute("INSERT INTO transactions (user_id, amount) VALUES (%s, 42.00)", (erin_id,))
cur.execute("INSERT INTO user_roles (user_id, role_id) SELECT %s, id FROM roles WHERE role_name = 'viewer'", (erin_id,))
conn.commit()

gdpr_delete_user(erin_id, admin_user_id=1, reason="User requested deletion via GDPR form")

cur.execute("SELECT action, record_id, success FROM audit_log WHERE action = 'GDPR_DELETE' ORDER BY id")
print("\nAudit trail:")
for action, record_id, success in cur.fetchall():
    print(f"  {action} record_id={record_id} success={success}")