PostgreSQL Row Level Security (RLS): Complete Implementation Guide

PostgreSQL Row Level Security (RLS): Complete Implementation Guide

Row Level Security (RLS) in PostgreSQL provides fine-grained access control at the row level, allowing you to restrict which rows users can access based on policies. This is essential for multi-tenant applications, data segregation, and implementing complex authorization requirements.

Understanding Row Level Security

What is Row Level Security?

Row Level Security (RLS) is a PostgreSQL feature that enables administrators and developers to define policies that restrict access to individual rows in a table based on the characteristics of the user attempting to access those rows.

Key Concepts

Policies: Rules that define which rows are visible or modifiable to specific users or roles.

USING clause: Determines which rows are visible for SELECT, UPDATE, and DELETE operations.

WITH CHECK clause: Determines which rows can be inserted or updated.

Policy Commands: Policies can apply to SELECT, INSERT, UPDATE, DELETE, or ALL commands.

Basic RLS Implementation

1. Enable Row Level Security

-- Create a sample table
CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    title TEXT NOT NULL,
    content TEXT,
    user_id INTEGER NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Enable RLS on the table
ALTER TABLE documents ENABLE ROW LEVEL SECURITY;

2. Create Basic Policies

-- Policy: Users can only see their own documents
CREATE POLICY user_documents_select ON documents
    FOR SELECT
    USING (user_id = current_user_id());

-- Policy: Users can only insert documents for themselves
CREATE POLICY user_documents_insert ON documents
    FOR INSERT
    WITH CHECK (user_id = current_user_id());

-- Policy: Users can only update their own documents
CREATE POLICY user_documents_update ON documents
    FOR UPDATE
    USING (user_id = current_user_id())
    WITH CHECK (user_id = current_user_id());

-- Policy: Users can only delete their own documents
CREATE POLICY user_documents_delete ON documents
    FOR DELETE
    USING (user_id = current_user_id());

-- Helper function to get current user ID
CREATE OR REPLACE FUNCTION current_user_id()
RETURNS INTEGER AS $$
    SELECT current_setting('app.user_id')::INTEGER;
$$ LANGUAGE SQL STABLE;

3. Test the Policies

-- Set the current user context
SET app.user_id = 1;

-- This user will only see documents where user_id = 1
SELECT * FROM documents;

-- Insert will automatically enforce user_id = 1
INSERT INTO documents (title, content, user_id)
VALUES ('My Document', 'Content here', 1);  -- Succeeds

INSERT INTO documents (title, content, user_id)
VALUES ('Other Document', 'Content', 2);  -- Fails WITH CHECK violation

Multi-Tenant Application Example

Complete Multi-Tenant Setup

-- Create tenants table
CREATE TABLE tenants (
    id SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Create users table with tenant association
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    username TEXT UNIQUE NOT NULL,
    email TEXT UNIQUE NOT NULL,
    tenant_id INTEGER REFERENCES tenants(id),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Create tenant-specific data table
CREATE TABLE customer_data (
    id SERIAL PRIMARY KEY,
    tenant_id INTEGER REFERENCES tenants(id),
    customer_name TEXT NOT NULL,
    customer_email TEXT,
    data JSONB,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Enable RLS on customer_data
ALTER TABLE customer_data ENABLE ROW LEVEL SECURITY;

-- Create policy for tenant isolation
CREATE POLICY tenant_isolation ON customer_data
    FOR ALL
    USING (tenant_id = current_setting('app.tenant_id')::INTEGER)
    WITH CHECK (tenant_id = current_setting('app.tenant_id')::INTEGER);

-- Create policy for admins (bypass tenant isolation)
CREATE POLICY admin_all_access ON customer_data
    FOR ALL
    TO admin_role
    USING (true)
    WITH CHECK (true);

Application Integration

# Python example with psycopg2
import psycopg2

def get_tenant_connection(tenant_id, db_config):
    """Create a connection with tenant context"""
    conn = psycopg2.connect(**db_config)
    cursor = conn.cursor()

    # Set the tenant context for RLS
    cursor.execute("SET app.tenant_id = %s", (tenant_id,))

    return conn

# Usage
conn = get_tenant_connection(tenant_id=5, db_config={
    'host': 'localhost',
    'database': 'myapp',
    'user': 'app_user',
    'password': 'secure_password'
})

# All queries will automatically be filtered by tenant_id = 5
cursor = conn.cursor()
cursor.execute("SELECT * FROM customer_data")
# Only returns rows where tenant_id = 5
// Node.js example with pg
const { Pool } = require('pg');

async function getTenantClient(tenantId, pool) {
    const client = await pool.connect();

    // Set tenant context
    await client.query('SET app.tenant_id = $1', [tenantId]);

    return client;
}

// Usage
const pool = new Pool({
    host: 'localhost',
    database: 'myapp',
    user: 'app_user',
    password: 'secure_password'
});

const client = await getTenantClient(5, pool);

// All queries filtered by tenant_id = 5
const result = await client.query('SELECT * FROM customer_data');

Advanced RLS Patterns

1. Hierarchical Access Control

-- Create organizational hierarchy
CREATE TABLE departments (
    id SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    parent_id INTEGER REFERENCES departments(id)
);

CREATE TABLE employees (
    id SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    department_id INTEGER REFERENCES departments(id),
    role TEXT NOT NULL
);

CREATE TABLE records (
    id SERIAL PRIMARY KEY,
    department_id INTEGER REFERENCES departments(id),
    content TEXT,
    confidential BOOLEAN DEFAULT false
);

-- Enable RLS
ALTER TABLE records ENABLE ROW LEVEL SECURITY;

-- Policy: Users can see records from their department and child departments
CREATE POLICY department_hierarchy ON records
    FOR SELECT
    USING (
        department_id IN (
            WITH RECURSIVE dept_tree AS (
                -- Start with user's department
                SELECT id FROM departments
                WHERE id = current_setting('app.department_id')::INTEGER

                UNION ALL

                -- Include child departments
                SELECT d.id FROM departments d
                INNER JOIN dept_tree dt ON d.parent_id = dt.id
            )
            SELECT id FROM dept_tree
        )
    );

-- Policy: Managers can see confidential records
CREATE POLICY manager_confidential ON records
    FOR SELECT
    USING (
        NOT confidential
        OR current_setting('app.user_role') = 'manager'
    );

2. Time-Based Access Control

CREATE TABLE time_restricted_data (
    id SERIAL PRIMARY KEY,
    content TEXT,
    available_from TIMESTAMP,
    available_until TIMESTAMP,
    user_id INTEGER
);

ALTER TABLE time_restricted_data ENABLE ROW LEVEL SECURITY;

-- Policy: Data is only accessible during specified time window
CREATE POLICY time_based_access ON time_restricted_data
    FOR SELECT
    USING (
        user_id = current_setting('app.user_id')::INTEGER
        AND CURRENT_TIMESTAMP >= available_from
        AND CURRENT_TIMESTAMP <= available_until
    );

3. Attribute-Based Access Control (ABAC)

CREATE TABLE sensitive_documents (
    id SERIAL PRIMARY KEY,
    title TEXT,
    classification TEXT,  -- 'public', 'internal', 'confidential', 'secret'
    department TEXT,
    owner_id INTEGER
);

ALTER TABLE sensitive_documents ENABLE ROW LEVEL SECURITY;

-- Policy: Complex attribute-based access
CREATE POLICY abac_document_access ON sensitive_documents
    FOR SELECT
    USING (
        classification = 'public'
        OR (
            classification = 'internal'
            AND current_setting('app.user_department') = department
        )
        OR (
            classification = 'confidential'
            AND current_setting('app.clearance_level')::INTEGER >= 2
        )
        OR (
            classification = 'secret'
            AND current_setting('app.clearance_level')::INTEGER >= 3
            AND current_setting('app.user_department') = department
        )
        OR owner_id = current_setting('app.user_id')::INTEGER
    );

4. Role-Based RLS

-- Create different policies for different roles
CREATE TABLE projects (
    id SERIAL PRIMARY KEY,
    name TEXT,
    owner_id INTEGER,
    team_id INTEGER,
    status TEXT
);

ALTER TABLE projects ENABLE ROW LEVEL SECURITY;

-- Policy for regular users: can see projects they own or are team members of
CREATE POLICY user_projects ON projects
    FOR ALL
    TO regular_user
    USING (
        owner_id = current_setting('app.user_id')::INTEGER
        OR team_id IN (
            SELECT team_id FROM team_members
            WHERE user_id = current_setting('app.user_id')::INTEGER
        )
    );

-- Policy for managers: can see all projects in their department
CREATE POLICY manager_projects ON projects
    FOR ALL
    TO manager_role
    USING (
        team_id IN (
            SELECT id FROM teams
            WHERE department_id = current_setting('app.department_id')::INTEGER
        )
    );

-- Policy for admins: can see everything
CREATE POLICY admin_projects ON projects
    FOR ALL
    TO admin_role
    USING (true)
    WITH CHECK (true);

RLS with JOINs

Handling Related Tables

-- Main table with RLS
CREATE TABLE orders (
    id SERIAL PRIMARY KEY,
    customer_id INTEGER,
    total DECIMAL(10,2),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Related table with RLS
CREATE TABLE order_items (
    id SERIAL PRIMARY KEY,
    order_id INTEGER REFERENCES orders(id),
    product_name TEXT,
    quantity INTEGER,
    price DECIMAL(10,2)
);

-- Enable RLS on both tables
ALTER TABLE orders ENABLE ROW LEVEL SECURITY;
ALTER TABLE order_items ENABLE ROW LEVEL SECURITY;

-- Policy for orders
CREATE POLICY customer_orders ON orders
    FOR ALL
    USING (customer_id = current_setting('app.customer_id')::INTEGER);

-- Policy for order_items (enforces through JOIN)
CREATE POLICY customer_order_items ON order_items
    FOR ALL
    USING (
        order_id IN (
            SELECT id FROM orders
            WHERE customer_id = current_setting('app.customer_id')::INTEGER
        )
    );

-- Query with JOIN (RLS applied automatically)
SET app.customer_id = 123;

SELECT o.id, o.total, oi.product_name, oi.quantity
FROM orders o
JOIN order_items oi ON o.id = oi.order_id
WHERE o.created_at > CURRENT_DATE - INTERVAL '30 days';
-- Only returns orders and items for customer_id = 123

Performance Considerations

1. Index Strategy

-- Create indexes on columns used in RLS policies
CREATE INDEX idx_documents_user_id ON documents(user_id);
CREATE INDEX idx_customer_data_tenant_id ON customer_data(tenant_id);
CREATE INDEX idx_orders_customer_id ON orders(customer_id);

-- Composite index for complex policies
CREATE INDEX idx_records_dept_confidential
ON records(department_id, confidential);

2. Optimize Complex Policies

-- Bad: Complex subquery in every row check
CREATE POLICY slow_policy ON large_table
    FOR SELECT
    USING (
        id IN (
            SELECT item_id FROM complex_join_query
            WHERE user_id = current_setting('app.user_id')::INTEGER
        )
    );

-- Better: Materialized view or simplified logic
CREATE MATERIALIZED VIEW user_accessible_items AS
SELECT user_id, item_id
FROM complex_join_query;

CREATE INDEX idx_user_accessible ON user_accessible_items(user_id, item_id);

CREATE POLICY fast_policy ON large_table
    FOR SELECT
    USING (
        id IN (
            SELECT item_id FROM user_accessible_items
            WHERE user_id = current_setting('app.user_id')::INTEGER
        )
    );

-- Refresh materialized view periodically
REFRESH MATERIALIZED VIEW CONCURRENTLY user_accessible_items;

3. Monitor RLS Performance

-- Enable timing to see RLS impact
EXPLAIN (ANALYZE, BUFFERS) SELECT * FROM customer_data;

-- Check if RLS policies are using indexes
EXPLAIN (ANALYZE, BUFFERS, VERBOSE)
SELECT * FROM documents
WHERE created_at > CURRENT_DATE - INTERVAL '7 days';

Security Best Practices

1. Force RLS for All Users

-- Make RLS mandatory (even for table owner)
ALTER TABLE sensitive_table FORCE ROW LEVEL SECURITY;

-- Without FORCE, table owners bypass RLS
-- With FORCE, policies apply to everyone except BYPASSRLS users

2. Limit BYPASSRLS Privilege

-- Check which roles can bypass RLS
SELECT rolname, rolbypassrls
FROM pg_roles
WHERE rolbypassrls = true;

-- Revoke BYPASSRLS from roles that don't need it
ALTER ROLE app_user NOBYPASSRLS;

-- Only grant to specific admin roles
ALTER ROLE security_admin BYPASSRLS;

3. Validate Policy Logic

-- Create test users for different scenarios
CREATE ROLE test_user1 LOGIN PASSWORD 'test123';
CREATE ROLE test_user2 LOGIN PASSWORD 'test123';

-- Grant necessary permissions
GRANT SELECT, INSERT, UPDATE, DELETE ON customer_data TO test_user1;
GRANT SELECT, INSERT, UPDATE, DELETE ON customer_data TO test_user2;

-- Test isolation between users
SET ROLE test_user1;
SET app.tenant_id = 1;
SELECT * FROM customer_data;  -- Should only see tenant_id = 1

SET ROLE test_user2;
SET app.tenant_id = 2;
SELECT * FROM customer_data;  -- Should only see tenant_id = 2

4. Audit RLS Policies

-- List all tables with RLS enabled
SELECT schemaname, tablename, rowsecurity
FROM pg_tables
WHERE rowsecurity = true;

-- List all policies
SELECT
    schemaname,
    tablename,
    policyname,
    permissive,
    roles,
    cmd,
    qual,
    with_check
FROM pg_policies
ORDER BY schemaname, tablename, policyname;

Common RLS Patterns

1. Shared Data with Owner Override

CREATE TABLE shared_files (
    id SERIAL PRIMARY KEY,
    filename TEXT,
    owner_id INTEGER,
    shared_with_group INTEGER
);

ALTER TABLE shared_files ENABLE ROW LEVEL SECURITY;

-- Users can see files they own or files shared with their group
CREATE POLICY shared_files_access ON shared_files
    FOR SELECT
    USING (
        owner_id = current_setting('app.user_id')::INTEGER
        OR shared_with_group = current_setting('app.group_id')::INTEGER
    );

2. Approval Workflow

CREATE TABLE purchase_requests (
    id SERIAL PRIMARY KEY,
    requester_id INTEGER,
    approver_id INTEGER,
    amount DECIMAL(10,2),
    status TEXT,  -- 'pending', 'approved', 'rejected'
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

ALTER TABLE purchase_requests ENABLE ROW LEVEL SECURITY;

-- Requesters can see their own requests
CREATE POLICY requester_view ON purchase_requests
    FOR SELECT
    USING (requester_id = current_setting('app.user_id')::INTEGER);

-- Approvers can see pending requests assigned to them
CREATE POLICY approver_view ON purchase_requests
    FOR SELECT
    USING (
        approver_id = current_setting('app.user_id')::INTEGER
        AND status = 'pending'
    );

-- Only approvers can update status
CREATE POLICY approver_update ON purchase_requests
    FOR UPDATE
    USING (approver_id = current_setting('app.user_id')::INTEGER)
    WITH CHECK (status IN ('approved', 'rejected'));

3. Geographic Restrictions

CREATE TABLE regional_data (
    id SERIAL PRIMARY KEY,
    region TEXT,
    data JSONB
);

ALTER TABLE regional_data ENABLE ROW LEVEL SECURITY;

-- Users can only access data from their assigned regions
CREATE POLICY regional_access ON regional_data
    FOR ALL
    USING (
        region = ANY(string_to_array(
            current_setting('app.user_regions'),
            ','
        ))
    );

-- Set user context with multiple regions
SET app.user_regions = 'US,EU';

Troubleshooting RLS

Common Issues

Issue 1: No rows returned

-- Check if RLS is enabled
SELECT tablename, rowsecurity, forcerowsecurity
FROM pg_tables
WHERE tablename = 'your_table';

-- Check active policies
SELECT * FROM pg_policies WHERE tablename = 'your_table';

-- Verify context is set
SHOW app.tenant_id;

Issue 2: Permission denied

-- User needs SELECT permission on table
GRANT SELECT ON table_name TO user_role;

-- User also needs USAGE on schema
GRANT USAGE ON SCHEMA schema_name TO user_role;

Issue 3: RLS bypassed

-- Check if user has BYPASSRLS
SELECT current_user,
       EXISTS(SELECT 1 FROM pg_roles
              WHERE rolname = current_user
              AND rolbypassrls = true) as can_bypass_rls;

-- Check if FORCE ROW LEVEL SECURITY is enabled
SELECT forcerowsecurity FROM pg_tables WHERE tablename = 'your_table';

Debugging Policies

-- Enable verbose logging
SET client_min_messages = DEBUG1;

-- Check policy evaluation
EXPLAIN (VERBOSE) SELECT * FROM customer_data;

-- Test with different context values
SET app.tenant_id = 1;
SELECT count(*) FROM customer_data;  -- Should return count for tenant 1

SET app.tenant_id = 2;
SELECT count(*) FROM customer_data;  -- Should return count for tenant 2

Migration to RLS

Step-by-Step Migration

-- Step 1: Add tenant_id to existing table
ALTER TABLE existing_table ADD COLUMN tenant_id INTEGER;

-- Step 2: Populate tenant_id based on existing data
UPDATE existing_table e
SET tenant_id = u.tenant_id
FROM users u
WHERE e.user_id = u.id;

-- Step 3: Make tenant_id NOT NULL after population
ALTER TABLE existing_table ALTER COLUMN tenant_id SET NOT NULL;

-- Step 4: Create index
CREATE INDEX idx_existing_table_tenant ON existing_table(tenant_id);

-- Step 5: Enable RLS (but don't force yet)
ALTER TABLE existing_table ENABLE ROW LEVEL SECURITY;

-- Step 6: Create policy
CREATE POLICY tenant_access ON existing_table
    FOR ALL
    USING (tenant_id = current_setting('app.tenant_id')::INTEGER);

-- Step 7: Test thoroughly with existing users

-- Step 8: Force RLS after validation
ALTER TABLE existing_table FORCE ROW LEVEL SECURITY;

Further Reading

Pulse - Elasticsearch Operations Done Right

Pulse can solve your Elasticsearch issues

Subscribe to the Pulse Newsletter

Get early access to new Pulse features, insightful blogs & exclusive events , webinars, and workshops.

We use cookies to provide an optimized user experience and understand our traffic. To learn more, read our use of cookies; otherwise, please choose 'Accept Cookies' to continue using our website.