Lab 11: Cloud Database Security — Neon Branching & Connection Hardening¶

-- Each Neon branch has its own endpoint (visible as app_name)
SELECT
  current_database()           AS database,
  current_user                 AS connected_as,
  inet_server_addr()           AS server_ip,
  inet_server_port()           AS port,
  pg_postmaster_start_time()   AS server_started;

CREATE TABLE IF NOT EXISTS public.branch_registry (
  branch_name   TEXT PRIMARY KEY,
  purpose       TEXT NOT NULL,
  created_at    TIMESTAMPTZ DEFAULT NOW(),
  ssl_enforced  BOOLEAN DEFAULT TRUE,
  max_conns     INT DEFAULT 10,
  environment   TEXT NOT NULL
                  CHECK (environment IN ('prod','staging','dev','test'))
);

INSERT INTO branch_registry VALUES
  ('main',    'Production database',    NOW(), TRUE, 100, 'prod'),
  ('lab-11',  'Lab 11 branch',          NOW(), TRUE, 10,  'dev'),
  ('staging', 'Pre-production testing', NOW(), TRUE, 50,  'staging')
ON CONFLICT (branch_name) DO NOTHING;

SELECT * FROM branch_registry ORDER BY environment;

-- Neon always uses SSL; verify TLS version and cipher
SELECT
  ssl,
  version,
  cipher,
  bits,
  client_dn
FROM pg_stat_ssl
WHERE pid = pg_backend_pid();

-- Check current connection count vs limit
SELECT
  datname,
  numbackends AS current_connections,
  (SELECT setting::INT FROM pg_settings
   WHERE name = 'max_connections') AS max_connections
FROM pg_stat_database
WHERE datname = current_database();

SELECT
  pid,
  usename,
  application_name,
  client_addr,
  state,
  wait_event_type,
  query_start,
  LEFT(query, 80) AS current_query
FROM pg_stat_activity
WHERE datname = current_database()
  AND pid != pg_backend_pid()
ORDER BY query_start DESC NULLS LAST;

CREATE TABLE IF NOT EXISTS public.cloud_security_controls (
  control_id   TEXT PRIMARY KEY,
  category     TEXT NOT NULL,
  responsible  TEXT NOT NULL
                 CHECK (responsible IN ('Neon','Customer','Shared')),
  description  TEXT NOT NULL,
  implemented  BOOLEAN DEFAULT FALSE
);

INSERT INTO cloud_security_controls VALUES
  ('NEON-01', 'Physical Security',     'Neon',     'Data center physical access control',                TRUE),
  ('NEON-02', 'Infrastructure',        'Neon',     'Hypervisor and OS patching',                         TRUE),
  ('NEON-03', 'Network',               'Neon',     'DDoS protection and network isolation',              TRUE),
  ('NEON-04', 'Encryption in transit', 'Neon',     'TLS enforcement on all connections',                 TRUE),
  ('CUST-01', 'Authentication',        'Customer', 'Role passwords and connection limits',               FALSE),
  ('CUST-02', 'Access Control',        'Customer', 'GRANT/REVOKE, RLS, schema privileges',              FALSE),
  ('CUST-03', 'Encryption at rest',    'Customer', 'Column-level encryption with pgcrypto',             FALSE),
  ('CUST-04', 'Auditing',              'Customer', 'Audit triggers and log monitoring',                  FALSE),
  ('CUST-05', 'Branch isolation',      'Customer', 'Use separate branches per environment',              FALSE),
  ('SHR-01',  'Monitoring',            'Shared',   'Neon provides metrics; customer monitors app queries', FALSE)
ON CONFLICT (control_id) DO NOTHING;

SELECT
  responsible,
  COUNT(*)                                           AS control_count,
  SUM(CASE WHEN implemented THEN 1 ELSE 0 END)       AS implemented
FROM cloud_security_controls
GROUP BY responsible
ORDER BY responsible;

UPDATE public.cloud_security_controls
SET implemented = TRUE
WHERE responsible = 'Customer'
  AND control_id IN ('CUST-01','CUST-02','CUST-03','CUST-04');

-- Verify implementation status
SELECT
  control_id,
  category,
  responsible,
  implemented
FROM public.cloud_security_controls
ORDER BY responsible, control_id;

SELECT
  rolname,
  CASE WHEN rolsuper
       THEN 'DANGER: superuser'      ELSE 'ok' END AS superuser_check,
  CASE WHEN rolcreaterole
       THEN 'RISK: can create roles' ELSE 'ok' END AS createrole_check,
  CASE WHEN rolcreatedb
       THEN 'RISK: can create dbs'   ELSE 'ok' END AS createdb_check,
  rolconnlimit,
  rolvaliduntil
FROM pg_roles
WHERE rolcanlogin = TRUE
  AND rolname NOT LIKE 'pg_%'
ORDER BY rolname;

-- VERIFY LAB 11
SELECT
  -- Branch registry exists with entries
  (SELECT COUNT(*) FROM public.branch_registry)::INT
    AS branch_entries,

  -- Shared responsibility table filled
  (SELECT COUNT(*) FROM public.cloud_security_controls)::INT
    AS control_count,

  -- Customer controls marked implemented
  (SELECT COUNT(*) FROM public.cloud_security_controls
   WHERE responsible = 'Customer' AND implemented = TRUE)::INT
    AS customer_controls_implemented,

  -- SSL is active (Neon always on)
  (SELECT ssl FROM pg_stat_ssl WHERE pid = pg_backend_pid())
    AS ssl_active;

CREATE OR REPLACE FUNCTION public.check_connection_policy()
RETURNS TABLE(
  policy_name  TEXT,
  status       TEXT,
  detail       TEXT
) AS $$
BEGIN
  -- Check 1: password encryption method is scram-sha-256
  RETURN QUERY
  SELECT
    'Password Encryption'::TEXT,
    CASE WHEN setting = 'scram-sha-256' THEN 'PASS'
         WHEN setting = 'md5'           THEN 'FAIL'
         ELSE                                'WARN'
    END::TEXT,
    ('Current setting: ' || setting)::TEXT
  FROM pg_settings WHERE name = 'password_encryption';

  -- Check 2: current connection uses SSL
  RETURN QUERY
  SELECT
    'SSL Enforcement'::TEXT,
    CASE WHEN ssl THEN 'PASS' ELSE 'FAIL' END::TEXT,
    CASE WHEN ssl
         THEN ('TLS active — cipher: ' || cipher)
         ELSE 'Connection is NOT encrypted'
    END::TEXT
  FROM pg_stat_ssl WHERE pid = pg_backend_pid();

  -- Check 3: no login role (except postgres) is superuser
  RETURN QUERY
  SELECT
    'No Unexpected Superusers'::TEXT,
    CASE WHEN COUNT(*) = 0 THEN 'PASS' ELSE 'FAIL' END::TEXT,
    CASE WHEN COUNT(*) = 0
         THEN 'No unexpected superuser login roles found'
         ELSE COUNT(*)::TEXT || ' superuser login role(s) found: '
              || STRING_AGG(rolname, ', ')
    END::TEXT
  FROM pg_roles
  WHERE rolcanlogin = TRUE
    AND rolsuper     = TRUE
    AND rolname NOT IN ('postgres')
    AND rolname NOT LIKE 'pg_%';

  -- Check 4: no login role has unlimited connections (rolconnlimit = -1)
  RETURN QUERY
  SELECT
    'Connection Limits Set'::TEXT,
    CASE WHEN COUNT(*) = 0 THEN 'PASS' ELSE 'WARN' END::TEXT,
    CASE WHEN COUNT(*) = 0
         THEN 'All login roles have explicit connection limits'
         ELSE COUNT(*)::TEXT || ' role(s) with unlimited connections: '
              || STRING_AGG(rolname, ', ')
    END::TEXT
  FROM pg_roles
  WHERE rolcanlogin    = TRUE
    AND rolconnlimit   = -1
    AND rolname NOT LIKE 'pg_%';
END;
$$ LANGUAGE plpgsql;

-- Run the policy check
SELECT * FROM public.check_connection_policy();