splinter.sql

set local search_path = '';

-- Used for 0025_public_bucket_allows_listing
do $$
declare
    buckets text;
begin
    if pg_catalog.to_regclass('storage.buckets') is not null then
        select coalesce(
            pg_catalog.jsonb_agg(
                pg_catalog.jsonb_build_object('bucket_id', id::text, 'bucket_name', name::text)
                order by id
            ),
            '[]'::pg_catalog.jsonb
        )::text
        from storage.buckets
        where public = true
        into buckets;
    else
        buckets := '[]';
    end if;
    perform pg_catalog.set_config('splinter.public_buckets', buckets, true);
end $$;

(
with foreign_keys as (
    select
        ns.nspname as schema_name,
        cl.relname as table_name,
        cl.oid as table_oid,
        ct.conname as fkey_name,
        ct.conkey as col_attnums
    from
        pg_catalog.pg_constraint ct
        join pg_catalog.pg_class cl -- fkey owning table
            on ct.conrelid = cl.oid
        join pg_catalog.pg_namespace ns
            on cl.relnamespace = ns.oid
        left join pg_catalog.pg_depend d
            on d.objid = cl.oid
            and d.deptype = 'e'
            and d.classid = 'pg_catalog.pg_class'::regclass
    where
        ct.contype = 'f' -- foreign key constraints
        and d.objid is null -- exclude tables that are dependencies of extensions
        and ns.nspname not in (
            'pg_catalog', 'information_schema', 'auth', 'storage', 'vault', 'extensions'
        )
),
index_ as (
    select
        pi.indrelid as table_oid,
        indexrelid::regclass as index_,
        string_to_array(indkey::text, ' ')::smallint[] as col_attnums
    from
        pg_catalog.pg_index pi
    where
        indisvalid
)
select
    'unindexed_foreign_keys' as name,
    'Unindexed foreign keys' as title,
    'INFO' as level,
    'EXTERNAL' as facing,
    array['PERFORMANCE'] as categories,
    'Identifies foreign key constraints without a covering index, which can impact database performance.' as description,
    format(
        'Table \`%s.%s\` has a foreign key \`%s\` without a covering index. This can lead to suboptimal query performance.',
        fk.schema_name,
        fk.table_name,
        fk.fkey_name
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0001_unindexed_foreign_keys' as remediation,
    jsonb_build_object(
        'schema', fk.schema_name,
        'name', fk.table_name,
        'type', 'table',
        'fkey_name', fk.fkey_name,
        'fkey_columns', fk.col_attnums
    ) as metadata,
    format('unindexed_foreign_keys_%s_%s_%s', fk.schema_name, fk.table_name, fk.fkey_name) as cache_key
from
    foreign_keys fk
    left join index_ idx
        on fk.table_oid = idx.table_oid
        and fk.col_attnums = idx.col_attnums[1:array_length(fk.col_attnums, 1)]
    left join pg_catalog.pg_depend dep
        on idx.table_oid = dep.objid
        and dep.deptype = 'e'
        and dep.classid = 'pg_catalog.pg_class'::regclass
where
    idx.index_ is null
    and fk.schema_name not in (
        '_timescaledb_cache', '_timescaledb_catalog', '_timescaledb_config', '_timescaledb_internal', 'auth', 'cron', 'extensions', 'graphql', 'graphql_public', 'information_schema', 'net', 'pgmq', 'pgroonga', 'pgsodium', 'pgsodium_masks', 'pgtle', 'pgbouncer', 'pg_catalog', 'realtime', 'repack', 'storage', 'supabase_functions', 'supabase_migrations', 'tiger', 'topology', 'vault'
    )
    and dep.objid is null -- exclude tables owned by extensions
order by
    fk.schema_name,
    fk.table_name,
    fk.fkey_name)
union all
(
select
    'auth_users_exposed' as name,
    'Exposed Auth Users' as title,
    'ERROR' as level,
    'EXTERNAL' as facing,
    array['SECURITY'] as categories,
    'Detects if auth.users is exposed to anon or authenticated roles via a view or materialized view in schemas exposed to PostgREST, potentially compromising user data security.' as description,
    format(
        'View/Materialized View "%s" in the public schema may expose \`auth.users\` data to anon or authenticated roles.',
        c.relname
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0002_auth_users_exposed' as remediation,
    jsonb_build_object(
        'schema', n.nspname,
        'name', c.relname,
        'type', 'view',
        'exposed_to', array_remove(array_agg(DISTINCT case when pg_catalog.has_table_privilege('anon', c.oid, 'SELECT') then 'anon' when pg_catalog.has_table_privilege('authenticated', c.oid, 'SELECT') then 'authenticated' end), null)
    ) as metadata,
    format('auth_users_exposed_%s_%s', n.nspname, c.relname) as cache_key
from
    -- Identify the oid for auth.users
    pg_catalog.pg_class auth_users_pg_class
    join pg_catalog.pg_namespace auth_users_pg_namespace
        on auth_users_pg_class.relnamespace = auth_users_pg_namespace.oid
        and auth_users_pg_class.relname = 'users'
        and auth_users_pg_namespace.nspname = 'auth'
    -- Depends on auth.users
    join pg_catalog.pg_depend d
        on d.refobjid = auth_users_pg_class.oid
    join pg_catalog.pg_rewrite r
        on r.oid = d.objid
    join pg_catalog.pg_class c
        on c.oid = r.ev_class
    join pg_catalog.pg_namespace n
        on n.oid = c.relnamespace
    join pg_catalog.pg_class pg_class_auth_users
        on d.refobjid = pg_class_auth_users.oid
where
    d.deptype = 'n'
    and (
      pg_catalog.has_table_privilege('anon', c.oid, 'SELECT')
      or pg_catalog.has_table_privilege('authenticated', c.oid, 'SELECT')
    )
    and n.nspname = any(array(select trim(unnest(string_to_array(coalesce(current_setting('pgrst.db_schemas', 't'), 'public'), ',')))))
    -- Exclude self
    and c.relname <> '0002_auth_users_exposed'
    -- There are 3 insecure configurations
    and
    (
        -- Materialized views don't support RLS so this is insecure by default
        (c.relkind in ('m')) -- m for materialized view
        or
        -- Standard View, accessible to anon or authenticated that is security_definer
        (
            c.relkind = 'v' -- v for view
            -- Exclude security invoker views
            and not (
                lower(coalesce(c.reloptions::text,'{}'))::text[]
                && array[
                    'security_invoker=1',
                    'security_invoker=true',
                    'security_invoker=yes',
                    'security_invoker=on'
                ]
            )
        )
        or
        -- Standard View, security invoker, but no RLS enabled on auth.users
        (
            c.relkind in ('v') -- v for view
            -- is security invoker
            and (
                lower(coalesce(c.reloptions::text,'{}'))::text[]
                && array[
                    'security_invoker=1',
                    'security_invoker=true',
                    'security_invoker=yes',
                    'security_invoker=on'
                ]
            )
            and not pg_class_auth_users.relrowsecurity
        )
    )
group by
    n.nspname,
    c.relname,
    c.oid)
union all
(
with policies as (
    select
        nsp.nspname as schema_name,
        pb.tablename as table_name,
        pc.relrowsecurity as is_rls_active,
        polname as policy_name,
        polpermissive as is_permissive, -- if not, then restrictive
        (select array_agg(r::regrole) from unnest(polroles) as x(r)) as roles,
        case polcmd
            when 'r' then 'SELECT'
            when 'a' then 'INSERT'
            when 'w' then 'UPDATE'
            when 'd' then 'DELETE'
            when '*' then 'ALL'
        end as command,
        qual,
        with_check
    from
        pg_catalog.pg_policy pa
        join pg_catalog.pg_class pc
            on pa.polrelid = pc.oid
        join pg_catalog.pg_namespace nsp
            on pc.relnamespace = nsp.oid
        join pg_catalog.pg_policies pb
            on pc.relname = pb.tablename
            and nsp.nspname = pb.schemaname
            and pa.polname = pb.policyname
)
select
    'auth_rls_initplan' as name,
    'Auth RLS Initialization Plan' as title,
    'WARN' as level,
    'EXTERNAL' as facing,
    array['PERFORMANCE'] as categories,
    'Detects if calls to \`current_setting()\` and \`auth.<function>()\` in RLS policies are being unnecessarily re-evaluated for each row' as description,
    format(
        'Table \`%s.%s\` has a row level security policy \`%s\` that re-evaluates current_setting() or auth.<function>() for each row. This produces suboptimal query performance at scale. Resolve the issue by replacing \`auth.<function>()\` with \`(select auth.<function>())\`. See [docs](https://supabase.com/docs/guides/database/postgres/row-level-security#call-functions-with-select) for more info.',
        schema_name,
        table_name,
        policy_name
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0003_auth_rls_initplan' as remediation,
    jsonb_build_object(
        'schema', schema_name,
        'name', table_name,
        'type', 'table'
    ) as metadata,
    format('auth_rls_init_plan_%s_%s_%s', schema_name, table_name, policy_name) as cache_key
from
    policies
where
    is_rls_active
    -- NOTE: does not include realtime in support of monitoring policies on realtime.messages
    and schema_name not in (
        '_timescaledb_cache', '_timescaledb_catalog', '_timescaledb_config', '_timescaledb_internal', 'auth', 'cron', 'extensions', 'graphql', 'graphql_public', 'information_schema', 'net', 'pgmq', 'pgroonga', 'pgsodium', 'pgsodium_masks', 'pgtle', 'pgbouncer', 'pg_catalog', 'repack', 'storage', 'supabase_functions', 'supabase_migrations', 'tiger', 'topology', 'vault'
    )
    and (
        -- Example: auth.uid()
        (
            qual like '%auth.uid()%'
            and lower(qual) not like '%select auth.uid()%'
        )
        or (
            qual like '%auth.jwt()%'
            and lower(qual) not like '%select auth.jwt()%'
        )
        or (
            qual like '%auth.role()%'
            and lower(qual) not like '%select auth.role()%'
        )
        or (
            qual like '%auth.email()%'
            and lower(qual) not like '%select auth.email()%'
        )
        or (
            qual like '%current\_setting(%)%'
            and lower(qual) not like '%select current\_setting(%)%'
        )
        or (
            with_check like '%auth.uid()%'
            and lower(with_check) not like '%select auth.uid()%'
        )
        or (
            with_check like '%auth.jwt()%'
            and lower(with_check) not like '%select auth.jwt()%'
        )
        or (
            with_check like '%auth.role()%'
            and lower(with_check) not like '%select auth.role()%'
        )
        or (
            with_check like '%auth.email()%'
            and lower(with_check) not like '%select auth.email()%'
        )
        or (
            with_check like '%current\_setting(%)%'
            and lower(with_check) not like '%select current\_setting(%)%'
        )
    ))
union all
(
select
    'no_primary_key' as name,
    'No Primary Key' as title,
    'INFO' as level,
    'EXTERNAL' as facing,
    array['PERFORMANCE'] as categories,
    'Detects if a table does not have a primary key. Tables without a primary key can be inefficient to interact with at scale.' as description,
    format(
        'Table \`%s.%s\` does not have a primary key',
        pgns.nspname,
        pgc.relname
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0004_no_primary_key' as remediation,
     jsonb_build_object(
        'schema', pgns.nspname,
        'name', pgc.relname,
        'type', 'table'
    ) as metadata,
    format(
        'no_primary_key_%s_%s',
        pgns.nspname,
        pgc.relname
    ) as cache_key
from
    pg_catalog.pg_class pgc
    join pg_catalog.pg_namespace pgns
        on pgns.oid = pgc.relnamespace
    left join pg_catalog.pg_index pgi
        on pgi.indrelid = pgc.oid
    left join pg_catalog.pg_depend dep
        on pgc.oid = dep.objid
        and dep.deptype = 'e'
        and dep.classid = 'pg_catalog.pg_class'::regclass
where
    pgc.relkind = 'r' -- regular tables
    and pgns.nspname not in (
        '_timescaledb_cache', '_timescaledb_catalog', '_timescaledb_config', '_timescaledb_internal', 'auth', 'cron', 'extensions', 'graphql', 'graphql_public', 'information_schema', 'net', 'pgmq', 'pgroonga', 'pgsodium', 'pgsodium_masks', 'pgtle', 'pgbouncer', 'pg_catalog', 'realtime', 'repack', 'storage', 'supabase_functions', 'supabase_migrations', 'tiger', 'topology', 'vault'
    )
    and dep.objid is null -- exclude tables owned by extensions
group by
    pgc.oid,
    pgns.nspname,
    pgc.relname
having
    max(coalesce(pgi.indisprimary, false)::int) = 0)
union all
(
select
    'unused_index' as name,
    'Unused Index' as title,
    'INFO' as level,
    'EXTERNAL' as facing,
    array['PERFORMANCE'] as categories,
    'Detects if an index has never been used and may be a candidate for removal.' as description,
    format(
        'Index \`%s\` on table \`%s.%s\` has not been used',
        psui.indexrelname,
        psui.schemaname,
        psui.relname
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0005_unused_index' as remediation,
    jsonb_build_object(
        'schema', psui.schemaname,
        'name', psui.relname,
        'type', 'table'
    ) as metadata,
    format(
        'unused_index_%s_%s_%s',
        psui.schemaname,
        psui.relname,
        psui.indexrelname
    ) as cache_key

from
    pg_catalog.pg_stat_user_indexes psui
    join pg_catalog.pg_index pi
        on psui.indexrelid = pi.indexrelid
    left join pg_catalog.pg_depend dep
        on psui.relid = dep.objid
        and dep.deptype = 'e'
        and dep.classid = 'pg_catalog.pg_class'::regclass
where
    psui.idx_scan = 0
    and not pi.indisunique
    and not pi.indisprimary
    and dep.objid is null -- exclude tables owned by extensions
    and psui.schemaname not in (
        '_timescaledb_cache', '_timescaledb_catalog', '_timescaledb_config', '_timescaledb_internal', 'auth', 'cron', 'extensions', 'graphql', 'graphql_public', 'information_schema', 'net', 'pgmq', 'pgroonga', 'pgsodium', 'pgsodium_masks', 'pgtle', 'pgbouncer', 'pg_catalog', 'realtime', 'repack', 'storage', 'supabase_functions', 'supabase_migrations', 'tiger', 'topology', 'vault'
    ))
union all
(
select
    'multiple_permissive_policies' as name,
    'Multiple Permissive Policies' as title,
    'WARN' as level,
    'EXTERNAL' as facing,
    array['PERFORMANCE'] as categories,
    'Detects if multiple permissive row level security policies are present on a table for the same \`role\` and \`action\` (e.g. insert). Multiple permissive policies are suboptimal for performance as each policy must be executed for every relevant query.' as description,
    format(
        'Table \`%s.%s\` has multiple permissive policies for role \`%s\` for action \`%s\`. Policies include \`%s\`',
        n.nspname,
        c.relname,
        r.rolname,
        act.cmd,
        array_agg(p.polname order by p.polname)
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0006_multiple_permissive_policies' as remediation,
    jsonb_build_object(
        'schema', n.nspname,
        'name', c.relname,
        'type', 'table'
    ) as metadata,
    format(
        'multiple_permissive_policies_%s_%s_%s_%s',
        n.nspname,
        c.relname,
        r.rolname,
        act.cmd
    ) as cache_key
from
    pg_catalog.pg_policy p
    join pg_catalog.pg_class c
        on p.polrelid = c.oid
    join pg_catalog.pg_namespace n
        on c.relnamespace = n.oid
    join pg_catalog.pg_roles r
        on p.polroles @> array[r.oid]
        or p.polroles = array[0::oid]
    left join pg_catalog.pg_depend dep
        on c.oid = dep.objid
        and dep.deptype = 'e'
        and dep.classid = 'pg_catalog.pg_class'::regclass,
    lateral (
        select x.cmd
        from unnest((
            select
                case p.polcmd
                    when 'r' then array['SELECT']
                    when 'a' then array['INSERT']
                    when 'w' then array['UPDATE']
                    when 'd' then array['DELETE']
                    when '*' then array['SELECT', 'INSERT', 'UPDATE', 'DELETE']
                    else array['ERROR']
                end as actions
        )) x(cmd)
    ) act(cmd)
where
    c.relkind = 'r' -- regular tables
    and p.polpermissive -- policy is permissive
    and n.nspname not in (
        '_timescaledb_cache', '_timescaledb_catalog', '_timescaledb_config', '_timescaledb_internal', 'auth', 'cron', 'extensions', 'graphql', 'graphql_public', 'information_schema', 'net', 'pgmq', 'pgroonga', 'pgsodium', 'pgsodium_masks', 'pgtle', 'pgbouncer', 'pg_catalog', 'realtime', 'repack', 'storage', 'supabase_functions', 'supabase_migrations', 'tiger', 'topology', 'vault'
    )
    and r.rolname not like 'pg_%'
    and r.rolname not like 'supabase%admin'
    and not r.rolbypassrls
    and dep.objid is null -- exclude tables owned by extensions
group by
    n.nspname,
    c.relname,
    r.rolname,
    act.cmd
having
    count(1) > 1)
union all
(
select
    'policy_exists_rls_disabled' as name,
    'Policy Exists RLS Disabled' as title,
    'ERROR' as level,
    'EXTERNAL' as facing,
    array['SECURITY'] as categories,
    'Detects cases where row level security (RLS) policies have been created, but RLS has not been enabled for the underlying table.' as description,
    format(
        'Table \`%s.%s\` has RLS policies but RLS is not enabled on the table. Policies include %s.',
        n.nspname,
        c.relname,
        array_agg(p.polname order by p.polname)
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0007_policy_exists_rls_disabled' as remediation,
    jsonb_build_object(
        'schema', n.nspname,
        'name', c.relname,
        'type', 'table'
    ) as metadata,
    format(
        'policy_exists_rls_disabled_%s_%s',
        n.nspname,
        c.relname
    ) as cache_key
from
    pg_catalog.pg_policy p
    join pg_catalog.pg_class c
        on p.polrelid = c.oid
    join pg_catalog.pg_namespace n
        on c.relnamespace = n.oid
    left join pg_catalog.pg_depend dep
        on c.oid = dep.objid
        and dep.deptype = 'e'
        and dep.classid = 'pg_catalog.pg_class'::regclass
where
    c.relkind = 'r' -- regular tables
    and n.nspname not in (
        '_timescaledb_cache', '_timescaledb_catalog', '_timescaledb_config', '_timescaledb_internal', 'auth', 'cron', 'extensions', 'graphql', 'graphql_public', 'information_schema', 'net', 'pgmq', 'pgroonga', 'pgsodium', 'pgsodium_masks', 'pgtle', 'pgbouncer', 'pg_catalog', 'realtime', 'repack', 'storage', 'supabase_functions', 'supabase_migrations', 'tiger', 'topology', 'vault'
    )
    -- RLS is disabled
    and not c.relrowsecurity
    and dep.objid is null -- exclude tables owned by extensions
group by
    n.nspname,
    c.relname)
union all
(
select
    'rls_enabled_no_policy' as name,
    'RLS Enabled No Policy' as title,
    'INFO' as level,
    'EXTERNAL' as facing,
    array['SECURITY'] as categories,
    'Detects cases where row level security (RLS) has been enabled on a table but no RLS policies have been created.' as description,
    format(
        'Table \`%s.%s\` has RLS enabled, but no policies exist',
        n.nspname,
        c.relname
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0008_rls_enabled_no_policy' as remediation,
    jsonb_build_object(
        'schema', n.nspname,
        'name', c.relname,
        'type', 'table'
    ) as metadata,
    format(
        'rls_enabled_no_policy_%s_%s',
        n.nspname,
        c.relname
    ) as cache_key
from
    pg_catalog.pg_class c
    left join pg_catalog.pg_policy p
        on p.polrelid = c.oid
    join pg_catalog.pg_namespace n
        on c.relnamespace = n.oid
    left join pg_catalog.pg_depend dep
        on c.oid = dep.objid
        and dep.deptype = 'e'
        and dep.classid = 'pg_catalog.pg_class'::regclass
where
    c.relkind = 'r' -- regular tables
    and n.nspname not in (
        '_timescaledb_cache', '_timescaledb_catalog', '_timescaledb_config', '_timescaledb_internal', 'auth', 'cron', 'extensions', 'graphql', 'graphql_public', 'information_schema', 'net', 'pgmq', 'pgroonga', 'pgsodium', 'pgsodium_masks', 'pgtle', 'pgbouncer', 'pg_catalog', 'realtime', 'repack', 'storage', 'supabase_functions', 'supabase_migrations', 'tiger', 'topology', 'vault'
    )
    -- RLS is enabled
    and c.relrowsecurity
    and p.polname is null
    and dep.objid is null -- exclude tables owned by extensions
group by
    n.nspname,
    c.relname)
union all
(
select
    'duplicate_index' as name,
    'Duplicate Index' as title,
    'WARN' as level,
    'EXTERNAL' as facing,
    array['PERFORMANCE'] as categories,
    'Detects cases where two ore more identical indexes exist.' as description,
    format(
        'Table \`%s.%s\` has identical indexes %s. Drop all except one of them',
        n.nspname,
        c.relname,
        array_agg(pi.indexname order by pi.indexname)
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0009_duplicate_index' as remediation,
    jsonb_build_object(
        'schema', n.nspname,
        'name', c.relname,
        'type', case
            when c.relkind = 'r' then 'table'
            when c.relkind = 'm' then 'materialized view'
            else 'ERROR'
        end,
        'indexes', array_agg(pi.indexname order by pi.indexname)
    ) as metadata,
    format(
        'duplicate_index_%s_%s_%s',
        n.nspname,
        c.relname,
        array_agg(pi.indexname order by pi.indexname)
    ) as cache_key
from
    pg_catalog.pg_indexes pi
    join pg_catalog.pg_namespace n
        on n.nspname  = pi.schemaname
    join pg_catalog.pg_class c
        on pi.tablename = c.relname
        and n.oid = c.relnamespace
    left join pg_catalog.pg_depend dep
        on c.oid = dep.objid
        and dep.deptype = 'e'
        and dep.classid = 'pg_catalog.pg_class'::regclass
where
    c.relkind in ('r', 'm') -- tables and materialized views
    and n.nspname not in (
        '_timescaledb_cache', '_timescaledb_catalog', '_timescaledb_config', '_timescaledb_internal', 'auth', 'cron', 'extensions', 'graphql', 'graphql_public', 'information_schema', 'net', 'pgmq', 'pgroonga', 'pgsodium', 'pgsodium_masks', 'pgtle', 'pgbouncer', 'pg_catalog', 'realtime', 'repack', 'storage', 'supabase_functions', 'supabase_migrations', 'tiger', 'topology', 'vault'
    )
    and dep.objid is null -- exclude tables owned by extensions
group by
    n.nspname,
    c.relkind,
    c.relname,
    replace(pi.indexdef, pi.indexname, '')
having
    count(*) > 1)
union all
(
select
    'security_definer_view' as name,
    'Security Definer View' as title,
    'ERROR' as level,
    'EXTERNAL' as facing,
    array['SECURITY'] as categories,
    'Detects views defined with the SECURITY DEFINER property. These views enforce Postgres permissions and row level security policies (RLS) of the view creator, rather than that of the querying user' as description,
    format(
        'View \`%s.%s\` is defined with the SECURITY DEFINER property',
        n.nspname,
        c.relname
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0010_security_definer_view' as remediation,
    jsonb_build_object(
        'schema', n.nspname,
        'name', c.relname,
        'type', 'view'
    ) as metadata,
    format(
        'security_definer_view_%s_%s',
        n.nspname,
        c.relname
    ) as cache_key
from
    pg_catalog.pg_class c
    join pg_catalog.pg_namespace n
        on n.oid = c.relnamespace
    left join pg_catalog.pg_depend dep
        on c.oid = dep.objid
        and dep.deptype = 'e'
        and dep.classid = 'pg_catalog.pg_class'::regclass
where
    c.relkind = 'v'
    and (
        pg_catalog.has_table_privilege('anon', c.oid, 'SELECT')
        or pg_catalog.has_table_privilege('authenticated', c.oid, 'SELECT')
    )
    and substring(pg_catalog.version() from 'PostgreSQL ([0-9]+)') >= '15' -- security invoker was added in pg15
    and n.nspname = any(array(select trim(unnest(string_to_array(coalesce(current_setting('pgrst.db_schemas', 't'), 'public'), ',')))))
    and n.nspname not in (
        '_timescaledb_cache', '_timescaledb_catalog', '_timescaledb_config', '_timescaledb_internal', 'auth', 'cron', 'extensions', 'graphql', 'graphql_public', 'information_schema', 'net', 'pgmq', 'pgroonga', 'pgsodium', 'pgsodium_masks', 'pgtle', 'pgbouncer', 'pg_catalog', 'realtime', 'repack', 'storage', 'supabase_functions', 'supabase_migrations', 'tiger', 'topology', 'vault'
    )
    and dep.objid is null -- exclude views owned by extensions
    and not (
        lower(coalesce(c.reloptions::text,'{}'))::text[]
        && array[
            'security_invoker=1',
            'security_invoker=true',
            'security_invoker=yes',
            'security_invoker=on'
        ]
    ))
union all
(
select
    'function_search_path_mutable' as name,
    'Function Search Path Mutable' as title,
    'WARN' as level,
    'EXTERNAL' as facing,
    array['SECURITY'] as categories,
    'Detects functions where the search_path parameter is not set.' as description,
    format(
        'Function \`%s.%s\` has a role mutable search_path',
        n.nspname,
        p.proname
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0011_function_search_path_mutable' as remediation,
    jsonb_build_object(
        'schema', n.nspname,
        'name', p.proname,
        'type', 'function'
    ) as metadata,
    format(
        'function_search_path_mutable_%s_%s_%s',
        n.nspname,
        p.proname,
        md5(p.prosrc) -- required when function is polymorphic
    ) as cache_key
from
    pg_catalog.pg_proc p
    join pg_catalog.pg_namespace n
        on p.pronamespace = n.oid
    left join pg_catalog.pg_depend dep
        on p.oid = dep.objid
        and dep.deptype = 'e'
        and dep.classid = 'pg_catalog.pg_proc'::regclass
where
    n.nspname not in (
        '_timescaledb_cache', '_timescaledb_catalog', '_timescaledb_config', '_timescaledb_internal', 'auth', 'cron', 'extensions', 'graphql', 'graphql_public', 'information_schema', 'net', 'pgmq', 'pgroonga', 'pgsodium', 'pgsodium_masks', 'pgtle', 'pgbouncer', 'pg_catalog', 'realtime', 'repack', 'storage', 'supabase_functions', 'supabase_migrations', 'tiger', 'topology', 'vault'
    )
    and dep.objid is null -- exclude functions owned by extensions
    -- Search path not set
    and not exists (
        select 1
        from unnest(coalesce(p.proconfig, '{}')) as config
        where config like 'search_path=%'
    ))
union all
(
select
    'rls_disabled_in_public' as name,
    'RLS Disabled in Public' as title,
    'ERROR' as level,
    'EXTERNAL' as facing,
    array['SECURITY'] as categories,
    'Detects cases where row level security (RLS) has not been enabled on tables in schemas exposed to PostgREST' as description,
    format(
        'Table \`%s.%s\` is public, but RLS has not been enabled.',
        n.nspname,
        c.relname
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0013_rls_disabled_in_public' as remediation,
    jsonb_build_object(
        'schema', n.nspname,
        'name', c.relname,
        'type', 'table'
    ) as metadata,
    format(
        'rls_disabled_in_public_%s_%s',
        n.nspname,
        c.relname
    ) as cache_key
from
    pg_catalog.pg_class c
    join pg_catalog.pg_namespace n
        on c.relnamespace = n.oid
where
    c.relkind = 'r' -- regular tables
    -- RLS is disabled
    and not c.relrowsecurity
    and (
        pg_catalog.has_table_privilege('anon', c.oid, 'SELECT')
        or pg_catalog.has_table_privilege('authenticated', c.oid, 'SELECT')
    )
    and n.nspname = any(array(select trim(unnest(string_to_array(coalesce(current_setting('pgrst.db_schemas', 't'), 'public'), ',')))))
    and n.nspname not in (
        '_timescaledb_cache', '_timescaledb_catalog', '_timescaledb_config', '_timescaledb_internal', 'auth', 'cron', 'extensions', 'graphql', 'graphql_public', 'information_schema', 'net', 'pgmq', 'pgroonga', 'pgsodium', 'pgsodium_masks', 'pgtle', 'pgbouncer', 'pg_catalog', 'realtime', 'repack', 'storage', 'supabase_functions', 'supabase_migrations', 'tiger', 'topology', 'vault'
    ))
union all
(
select
    'extension_in_public' as name,
    'Extension in Public' as title,
    'WARN' as level,
    'EXTERNAL' as facing,
    array['SECURITY'] as categories,
    'Detects extensions installed in the \`public\` schema.' as description,
    format(
        'Extension \`%s\` is installed in the public schema. Move it to another schema.',
        pe.extname
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0014_extension_in_public' as remediation,
    jsonb_build_object(
        'schema', pe.extnamespace::regnamespace,
        'name', pe.extname,
        'type', 'extension'
    ) as metadata,
    format(
        'extension_in_public_%s',
        pe.extname
    ) as cache_key
from
    pg_catalog.pg_extension pe
where
    -- plpgsql is installed by default in public and outside user control
    -- confirmed safe
    pe.extname not in ('plpgsql')
    -- Scoping this to public is not optimal. Ideally we would use the postgres
    -- search path. That currently isn't available via SQL. In other lints
    -- we have used has_schema_privilege('anon', 'extensions', 'USAGE') but that
    -- is not appropriate here as it would evaluate true for the extensions schema
    and pe.extnamespace::regnamespace::text = 'public')
union all
(
with policies as (
    select
        nsp.nspname as schema_name,
        pb.tablename as table_name,
        polname as policy_name,
        qual,
        with_check
    from
        pg_catalog.pg_policy pa
        join pg_catalog.pg_class pc
            on pa.polrelid = pc.oid
        join pg_catalog.pg_namespace nsp
            on pc.relnamespace = nsp.oid
        join pg_catalog.pg_policies pb
            on pc.relname = pb.tablename
            and nsp.nspname = pb.schemaname
            and pa.polname = pb.policyname
)
select
    'rls_references_user_metadata' as name,
    'RLS references user metadata' as title,
    'ERROR' as level,
    'EXTERNAL' as facing,
    array['SECURITY'] as categories,
    'Detects when Supabase Auth user_metadata is referenced insecurely in a row level security (RLS) policy.' as description,
    format(
        'Table \`%s.%s\` has a row level security policy \`%s\` that references Supabase Auth \`user_metadata\`. \`user_metadata\` is editable by end users and should never be used in a security context.',
        schema_name,
        table_name,
        policy_name
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0015_rls_references_user_metadata' as remediation,
    jsonb_build_object(
        'schema', schema_name,
        'name', table_name,
        'type', 'table'
    ) as metadata,
    format('rls_references_user_metadata_%s_%s_%s', schema_name, table_name, policy_name) as cache_key
from
    policies
where
    schema_name not in (
        '_timescaledb_cache', '_timescaledb_catalog', '_timescaledb_config', '_timescaledb_internal', 'auth', 'cron', 'extensions', 'graphql', 'graphql_public', 'information_schema', 'net', 'pgmq', 'pgroonga', 'pgsodium', 'pgsodium_masks', 'pgtle', 'pgbouncer', 'pg_catalog', 'realtime', 'repack', 'storage', 'supabase_functions', 'supabase_migrations', 'tiger', 'topology', 'vault'
    )
    and (
        -- Example: auth.jwt() -> 'user_metadata'
        -- False positives are possible, but it isn't practical to string match
        -- If false positive rate is too high, this expression can iterate
        qual like '%auth.jwt()%user_metadata%'
        or qual like '%current_setting(%request.jwt.claims%)%user_metadata%'
        or with_check like '%auth.jwt()%user_metadata%'
        or with_check like '%current_setting(%request.jwt.claims%)%user_metadata%'
    ))
union all
(
select
    'materialized_view_in_api' as name,
    'Materialized View in API' as title,
    'WARN' as level,
    'EXTERNAL' as facing,
    array['SECURITY'] as categories,
    'Detects materialized views that are accessible over the Data APIs.' as description,
    format(
        'Materialized view \`%s.%s\` is selectable by anon or authenticated roles',
        n.nspname,
        c.relname
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0016_materialized_view_in_api' as remediation,
    jsonb_build_object(
        'schema', n.nspname,
        'name', c.relname,
        'type', 'materialized view'
    ) as metadata,
    format(
        'materialized_view_in_api_%s_%s',
        n.nspname,
        c.relname
    ) as cache_key
from
    pg_catalog.pg_class c
    join pg_catalog.pg_namespace n
        on n.oid = c.relnamespace
    left join pg_catalog.pg_depend dep
        on c.oid = dep.objid
        and dep.deptype = 'e'
        and dep.classid = 'pg_catalog.pg_class'::regclass
where
    c.relkind = 'm'
    and (
        pg_catalog.has_table_privilege('anon', c.oid, 'SELECT')
        or pg_catalog.has_table_privilege('authenticated', c.oid, 'SELECT')
    )
    and n.nspname = any(array(select trim(unnest(string_to_array(coalesce(current_setting('pgrst.db_schemas', 't'), 'public'), ',')))))
    and n.nspname not in (
        '_timescaledb_cache', '_timescaledb_catalog', '_timescaledb_config', '_timescaledb_internal', 'auth', 'cron', 'extensions', 'graphql', 'graphql_public', 'information_schema', 'net', 'pgmq', 'pgroonga', 'pgsodium', 'pgsodium_masks', 'pgtle', 'pgbouncer', 'pg_catalog', 'realtime', 'repack', 'storage', 'supabase_functions', 'supabase_migrations', 'tiger', 'topology', 'vault'
    )
    and dep.objid is null)
union all
(
select
    'foreign_table_in_api' as name,
    'Foreign Table in API' as title,
    'WARN' as level,
    'EXTERNAL' as facing,
    array['SECURITY'] as categories,
    'Detects foreign tables that are accessible over APIs. Foreign tables do not respect row level security policies.' as description,
    format(
        'Foreign table \`%s.%s\` is accessible over APIs',
        n.nspname,
        c.relname
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0017_foreign_table_in_api' as remediation,
    jsonb_build_object(
        'schema', n.nspname,
        'name', c.relname,
        'type', 'foreign table'
    ) as metadata,
    format(
        'foreign_table_in_api_%s_%s',
        n.nspname,
        c.relname
    ) as cache_key
from
    pg_catalog.pg_class c
    join pg_catalog.pg_namespace n
        on n.oid = c.relnamespace
    left join pg_catalog.pg_depend dep
        on c.oid = dep.objid
        and dep.deptype = 'e'
        and dep.classid = 'pg_catalog.pg_class'::regclass
where
    c.relkind = 'f'
    and (
        pg_catalog.has_table_privilege('anon', c.oid, 'SELECT')
        or pg_catalog.has_table_privilege('authenticated', c.oid, 'SELECT')
    )
    and n.nspname = any(array(select trim(unnest(string_to_array(coalesce(current_setting('pgrst.db_schemas', 't'), 'public'), ',')))))
    and n.nspname not in (
        '_timescaledb_cache', '_timescaledb_catalog', '_timescaledb_config', '_timescaledb_internal', 'auth', 'cron', 'extensions', 'graphql', 'graphql_public', 'information_schema', 'net', 'pgmq', 'pgroonga', 'pgsodium', 'pgsodium_masks', 'pgtle', 'pgbouncer', 'pg_catalog', 'realtime', 'repack', 'storage', 'supabase_functions', 'supabase_migrations', 'tiger', 'topology', 'vault'
    )
    and dep.objid is null)
union all
(
select
    'unsupported_reg_types' as name,
    'Unsupported reg types' as title,
    'WARN' as level,
    'EXTERNAL' as facing,
    array['SECURITY'] as categories,
    'Identifies columns using unsupported reg* types outside pg_catalog schema, which prevents database upgrades using pg_upgrade.' as description,
    format(
        'Table \`%s.%s\` has a column \`%s\` with unsupported reg* type \`%s\`.',
        n.nspname,
        c.relname,
        a.attname,
        t.typname
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=unsupported_reg_types' as remediation,
    jsonb_build_object(
        'schema', n.nspname,
        'name', c.relname,
        'column', a.attname,
        'type', 'table'
    ) as metadata,
    format(
        'unsupported_reg_types_%s_%s_%s',
        n.nspname,
        c.relname,
        a.attname
    ) AS cache_key
from
    pg_catalog.pg_attribute a
    join pg_catalog.pg_class c
        on a.attrelid = c.oid
    join pg_catalog.pg_namespace n
        on c.relnamespace = n.oid
    join pg_catalog.pg_type t
        on a.atttypid = t.oid
    join pg_catalog.pg_namespace tn
        on t.typnamespace = tn.oid
where
    tn.nspname = 'pg_catalog'
    and t.typname in ('regcollation', 'regconfig', 'regdictionary', 'regnamespace', 'regoper', 'regoperator', 'regproc', 'regprocedure')
    and n.nspname not in ('pg_catalog', 'information_schema', 'pgsodium'))
union all
(
select
    'insecure_queue_exposed_in_api' as name,
    'Insecure Queue Exposed in API' as title,
    'ERROR' as level,
    'EXTERNAL' as facing,
    array['SECURITY'] as categories,
    'Detects cases where an insecure Queue is exposed over Data APIs' as description,
    format(
        'Table \`%s.%s\` is public, but RLS has not been enabled.',
        n.nspname,
        c.relname
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0019_insecure_queue_exposed_in_api' as remediation,
    jsonb_build_object(
        'schema', n.nspname,
        'name', c.relname,
        'type', 'table'
    ) as metadata,
    format(
        'rls_disabled_in_public_%s_%s',
        n.nspname,
        c.relname
    ) as cache_key
from
    pg_catalog.pg_class c
    join pg_catalog.pg_namespace n
        on c.relnamespace = n.oid
where
    c.relkind in ('r', 'I') -- regular or partitioned tables
    and not c.relrowsecurity -- RLS is disabled
    and (
        pg_catalog.has_table_privilege('anon', c.oid, 'SELECT')
        or pg_catalog.has_table_privilege('authenticated', c.oid, 'SELECT')
    )
    and n.nspname = 'pgmq' -- tables in the pgmq schema
    and c.relname like 'q_%' -- only queue tables
    -- Constant requirements
    and 'pgmq_public' = any(array(select trim(unnest(string_to_array(coalesce(current_setting('pgrst.db_schemas', 't'), 'public'), ','))))))
union all
(
with constants as (
    select current_setting('block_size')::numeric as bs, 23 as hdr, 4 as ma
),

bloat_info as (
    select
        ma,
        bs,
        schemaname,
        tablename,
        (datawidth + (hdr + ma - (case when hdr % ma = 0 then ma else hdr % ma end)))::numeric as datahdr,
        (maxfracsum * (nullhdr + ma - (case when nullhdr % ma = 0 then ma else nullhdr % ma end))) as nullhdr2
    from (
        select
            schemaname,
            tablename,
            hdr,
            ma,
            bs,
            sum((1 - null_frac) * avg_width) as datawidth,
            max(null_frac) as maxfracsum,
            hdr + (
                select 1 + count(*) / 8
                from pg_stats s2
                where
                    null_frac <> 0
                    and s2.schemaname = s.schemaname
                    and s2.tablename = s.tablename
            ) as nullhdr
        from pg_stats s, constants
        group by 1, 2, 3, 4, 5
    ) as foo
),

table_bloat as (
    select
        schemaname,
        tablename,
        cc.relpages,
        bs,
        ceil((cc.reltuples * ((datahdr + ma -
          (case when datahdr % ma = 0 then ma else datahdr % ma end)) + nullhdr2 + 4)) / (bs - 20::float)) as otta
    from
        bloat_info
        join pg_class cc
            on cc.relname = bloat_info.tablename
        join pg_namespace nn
            on cc.relnamespace = nn.oid
            and nn.nspname = bloat_info.schemaname
            and nn.nspname <> 'information_schema'
        where
            cc.relkind = 'r'
            and cc.relam = (select oid from pg_am where amname = 'heap')
),

bloat_data as (
    select
        'table' as type,
        schemaname,
        tablename as object_name,
        round(case when otta = 0 then 0.0 else table_bloat.relpages / otta::numeric end, 1) as bloat,
        case when relpages < otta then 0 else (bs * (table_bloat.relpages - otta)::bigint)::bigint end as raw_waste
    from
        table_bloat
)

select
    'table_bloat' as name,
    'Table Bloat' as title,
    'WARN' as level,
    'EXTERNAL' as facing,
    array['PERFORMANCE'] as categories,
    'Detects if a table has excess bloat and may benefit from maintenance operations like vacuum full or cluster.' as description,
    format(
        'Table `%s`.`%s` has excessive bloat',
        bloat_data.schemaname,
        bloat_data.object_name
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0020_table_bloat' as remediation,
    jsonb_build_object(
        'schema', bloat_data.schemaname,
        'name', bloat_data.object_name,
        'type', bloat_data.type
    ) as metadata,
    format(
        'table_bloat_%s_%s',
        bloat_data.schemaname,
        bloat_data.object_name
    ) as cache_key
from
    bloat_data
where
    bloat > 70.0
    and raw_waste > (20 * 1024 * 1024) -- filter for waste > 200 MB
order by
    schemaname,
    object_name)
union all
(
select
    'fkey_to_auth_unique' as name,
    'Foreign Key to Auth Unique Constraint' as title,
    'ERROR' as level,
    'EXTERNAL' as facing,
    array['SECURITY'] as categories,
    'Detects user defined foreign keys to unique constraints in the auth schema.' as description,
    format(
        'Table `%s`.`%s` has a foreign key `%s` referencing an auth unique constraint',
        n.nspname, -- referencing schema
        c_rel.relname, -- referencing table
        c.conname -- fkey name
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0021_fkey_to_auth_unique' as remediation,
    jsonb_build_object(
        'schema', n.nspname,
        'name', c_rel.relname,
        'foreign_key', c.conname
    ) as metadata,
    format(
        'fkey_to_auth_unique_%s_%s_%s',
        n.nspname, -- referencing schema
        c_rel.relname, -- referencing table
        c.conname
    ) as cache_key
from
    pg_catalog.pg_constraint c
    join pg_catalog.pg_class c_rel
        on c.conrelid = c_rel.oid
    join pg_catalog.pg_namespace n
        on c_rel.relnamespace = n.oid
    join pg_catalog.pg_class ref_rel
        on c.confrelid = ref_rel.oid
    join pg_catalog.pg_namespace cn
        on ref_rel.relnamespace = cn.oid
    join pg_catalog.pg_index i
        on c.conindid = i.indexrelid
where c.contype = 'f'
    and cn.nspname = 'auth'
    and i.indisunique
    and not i.indisprimary)
union all
(
select
    'extension_versions_outdated' as name,
    'Extension Versions Outdated' as title,
    'WARN' as level,
    'EXTERNAL' as facing,
    array['SECURITY'] as categories,
    'Detects extensions that are not using the default (recommended) version.' as description,
    format(
        'Extension `%s` is using version `%s` but version `%s` is available. Using outdated extension versions may expose the database to security vulnerabilities.',
        ext.name,
        ext.installed_version,
        ext.default_version
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0022_extension_versions_outdated' as remediation,
    jsonb_build_object(
        'extension_name', ext.name,
        'installed_version', ext.installed_version,
        'default_version', ext.default_version
    ) as metadata,
    format(
        'extension_versions_outdated_%s_%s',
        ext.name,
        ext.installed_version
    ) as cache_key
from
    pg_catalog.pg_available_extensions ext
join
    -- ignore versions not in pg_available_extension_versions
    -- e.g. residue of pg_upgrade
    pg_catalog.pg_available_extension_versions extv
    on extv.name = ext.name and extv.installed
where
    ext.installed_version is not null
    and ext.default_version is not null
    and ext.installed_version != ext.default_version
order by
    ext.name)
union all
(
-- Detects tables exposed via API that contain columns with sensitive names
-- Inspired by patterns from security scanners that detect PII/credential exposure
with sensitive_patterns as (
    select unnest(array[
        -- Authentication & Credentials
        'password', 'passwd', 'pwd', 'passphrase',
        'secret', 'secret_key', 'private_key', 'api_key', 'apikey',
        'auth_key', 'token', 'jwt', 'access_token', 'refresh_token',
        'oauth_token', 'session_token', 'bearer_token', 'auth_code',
        'session_id', 'session_key', 'session_secret',
        'recovery_code', 'backup_code', 'verification_code',
        'otp', 'two_factor', '2fa_secret', '2fa_code',
        -- Personal Identifiers
        'ssn', 'social_security', 'social_security_number',
        'driver_license', 'drivers_license', 'license_number',
        'passport_number', 'passport_id', 'national_id', 'tax_id',
        -- Financial Information
        'credit_card', 'card_number', 'cvv', 'cvc', 'cvn',
        'bank_account', 'account_number', 'routing_number',
        'iban', 'swift_code', 'bic',
        -- Health & Medical
        'health_record', 'medical_record', 'patient_id',
        'insurance_number', 'health_insurance', 'medical_insurance',
        'treatment',
        -- Device Identifiers
        'mac_address', 'macaddr', 'imei', 'device_uuid',
        -- Digital Keys & Certificates
        'pgp_key', 'gpg_key', 'ssh_key', 'certificate',
        'license_key', 'activation_key',
        -- Biometric Data
        'facial_recognition'
    ]) as pattern
),
exposed_tables as (
    select
        n.nspname as schema_name,
        c.relname as table_name,
        c.oid as table_oid
    from
        pg_catalog.pg_class c
        join pg_catalog.pg_namespace n
            on c.relnamespace = n.oid
    where
        c.relkind = 'r' -- regular tables
        and (
            pg_catalog.has_table_privilege('anon', c.oid, 'SELECT')
            or pg_catalog.has_table_privilege('authenticated', c.oid, 'SELECT')
        )
        and n.nspname = any(array(select trim(unnest(string_to_array(coalesce(current_setting('pgrst.db_schemas', 't'), 'public'), ',')))))
        and n.nspname not in (
            '_timescaledb_cache', '_timescaledb_catalog', '_timescaledb_config', '_timescaledb_internal', 'auth', 'cron', 'extensions', 'graphql', 'graphql_public', 'information_schema', 'net', 'pgmq', 'pgroonga', 'pgsodium', 'pgsodium_masks', 'pgtle', 'pgbouncer', 'pg_catalog', 'realtime', 'repack', 'storage', 'supabase_functions', 'supabase_migrations', 'tiger', 'topology', 'vault'
        )
        -- Only flag tables without RLS enabled
        and not c.relrowsecurity
),
sensitive_columns as (
    select
        et.schema_name,
        et.table_name,
        a.attname as column_name,
        sp.pattern as matched_pattern
    from
        exposed_tables et
        join pg_catalog.pg_attribute a
            on a.attrelid = et.table_oid
            and a.attnum > 0
            and not a.attisdropped
        cross join sensitive_patterns sp
    where
        -- Match column name against sensitive patterns (case insensitive), allowing '-'/'_' variants
        replace(lower(a.attname), '-', '_') = sp.pattern
)
select
    'sensitive_columns_exposed' as name,
    'Sensitive Columns Exposed' as title,
    'ERROR' as level,
    'EXTERNAL' as facing,
    array['SECURITY'] as categories,
    'Detects tables exposed via API that contain columns with potentially sensitive data (PII, credentials, financial info) without RLS protection.' as description,
    format(
        'Table `%s.%s` is exposed via API without RLS and contains potentially sensitive column(s): %s. This may lead to data exposure.',
        schema_name,
        table_name,
        string_agg(distinct column_name, ', ' order by column_name)
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0023_sensitive_columns_exposed' as remediation,
    jsonb_build_object(
        'schema', schema_name,
        'name', table_name,
        'type', 'table',
        'sensitive_columns', array_agg(distinct column_name order by column_name),
        'matched_patterns', array_agg(distinct matched_pattern order by matched_pattern)
    ) as metadata,
    format(
        'sensitive_columns_exposed_%s_%s',
        schema_name,
        table_name
    ) as cache_key
from
    sensitive_columns
group by
    schema_name,
    table_name
order by
    schema_name,
    table_name)
union all
(
-- Detects RLS policies that are overly permissive (e.g., USING (true), USING (1=1))
-- These policies effectively disable row-level security while giving a false sense of security
with policies as (
    select
        nsp.nspname as schema_name,
        pb.tablename as table_name,
        pc.relrowsecurity as is_rls_active,
        pa.polname as policy_name,
        pa.polpermissive as is_permissive,
        pa.polroles as role_oids,
        (select array_agg(r::regrole::text) from unnest(pa.polroles) as x(r)) as roles,
        case pa.polcmd
            when 'r' then 'SELECT'
            when 'a' then 'INSERT'
            when 'w' then 'UPDATE'
            when 'd' then 'DELETE'
            when '*' then 'ALL'
        end as command,
        pb.qual,
        pb.with_check,
        -- Normalize expressions by removing whitespace and lowercasing
        replace(replace(replace(lower(coalesce(pb.qual, '')), ' ', ''), E'\n', ''), E'\t', '') as normalized_qual,
        replace(replace(replace(lower(coalesce(pb.with_check, '')), ' ', ''), E'\n', ''), E'\t', '') as normalized_with_check
    from
        pg_catalog.pg_policy pa
        join pg_catalog.pg_class pc
            on pa.polrelid = pc.oid
        join pg_catalog.pg_namespace nsp
            on pc.relnamespace = nsp.oid
        join pg_catalog.pg_policies pb
            on pc.relname = pb.tablename
            and nsp.nspname = pb.schemaname
            and pa.polname = pb.policyname
    where
        pc.relkind = 'r' -- regular tables
        and nsp.nspname not in (
            '_timescaledb_cache', '_timescaledb_catalog', '_timescaledb_config', '_timescaledb_internal', 'auth', 'cron', 'extensions', 'graphql', 'graphql_public', 'information_schema', 'net', 'pgmq', 'pgroonga', 'pgsodium', 'pgsodium_masks', 'pgtle', 'pgbouncer', 'pg_catalog', 'realtime', 'repack', 'storage', 'supabase_functions', 'supabase_migrations', 'tiger', 'topology', 'vault'
        )
),
permissive_patterns as (
    select
        p.*,
        -- Check for always-true USING clause patterns
        -- Note: SELECT with (true) is often intentional and documented, so we only flag UPDATE/DELETE
        case when (
            command in ('UPDATE', 'DELETE', 'ALL')
            and (
                normalized_qual in ('true', '(true)', '1=1', '(1=1)')
                -- Empty or null qual on permissive policy means allow all
                or (qual is null and is_permissive)
            )
        ) then true else false end as has_permissive_using,
        -- Check for always-true WITH CHECK clause patterns
        case when (
            normalized_with_check in ('true', '(true)', '1=1', '(1=1)')
            -- Empty with_check on INSERT means allow all (INSERT has no USING to fall back on)
            or (with_check is null and is_permissive and command = 'INSERT')
            -- Empty with_check on UPDATE/ALL with permissive USING means allow all writes
            or (with_check is null and is_permissive and command in ('UPDATE', 'ALL')
                and normalized_qual in ('true', '(true)', '1=1', '(1=1)'))
        ) then true else false end as has_permissive_with_check
    from
        policies p
    where
        -- Only check tables with RLS enabled (otherwise it's a different lint)
        is_rls_active
        -- Only check permissive policies (restrictive policies with true are less dangerous)
        and is_permissive
        -- Only flag policies that apply to anon or authenticated roles (or public/all roles)
        and (
            role_oids = array[0::oid] -- public (all roles)
            or exists (
                select 1
                from unnest(role_oids) as r
                where r::regrole::text in ('anon', 'authenticated')
            )
        )
)
select
    'rls_policy_always_true' as name,
    'RLS Policy Always True' as title,
    'WARN' as level,
    'EXTERNAL' as facing,
    array['SECURITY'] as categories,
    'Detects RLS policies that use overly permissive expressions like \`USING (true)\` or \`WITH CHECK (true)\` for UPDATE, DELETE, or INSERT operations. SELECT policies with \`USING (true)\` are intentionally excluded as this pattern is often used deliberately for public read access.' as description,
    format(
        'Table `%s.%s` has an RLS policy `%s` for `%s` that allows unrestricted access%s. This effectively bypasses row-level security for %s.',
        schema_name,
        table_name,
        policy_name,
        command,
        case
            when has_permissive_using and has_permissive_with_check then ' (both USING and WITH CHECK are always true)'
            when has_permissive_using then ' (USING clause is always true)'
            when has_permissive_with_check then ' (WITH CHECK clause is always true)'
            else ''
        end,
        array_to_string(roles, ', ')
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0024_permissive_rls_policy' as remediation,
    jsonb_build_object(
        'schema', schema_name,
        'name', table_name,
        'type', 'table',
        'policy_name', policy_name,
        'command', command,
        'roles', roles,
        'qual', qual,
        'with_check', with_check,
        'permissive_using', has_permissive_using,
        'permissive_with_check', has_permissive_with_check
    ) as metadata,
    format(
        'rls_policy_always_true_%s_%s_%s',
        schema_name,
        table_name,
        policy_name
    ) as cache_key
from
    permissive_patterns
where
    has_permissive_using or has_permissive_with_check
order by
    schema_name,
    table_name,
    policy_name)
union all
(
with public_buckets as (
    -- Read storage.buckets at runtime so the lint can load even when storage is not installed.
    select
        bucket->>'bucket_id' as bucket_id,
        bucket->>'bucket_name' as bucket_name
    from pg_catalog.jsonb_array_elements(
        coalesce(
            nullif(pg_catalog.current_setting('splinter.public_buckets', true), '')::pg_catalog.jsonb,
            '[]'::pg_catalog.jsonb
        )
    ) as buckets(bucket)
),
matching_policies as (
    select
        b.bucket_id,
        b.bucket_name,
        p.policyname
    from
        public_buckets b
        join pg_catalog.pg_policies p
            on p.schemaname = 'storage'
            and p.tablename = 'objects'
            and p.cmd in ('SELECT', 'ALL')
            and p.permissive = 'PERMISSIVE'
            and p.roles && array['public'::name, 'anon'::name, 'authenticated'::name]
    where
        (
            p.qual is null
            or replace(replace(replace(lower(p.qual), ' ', ''), E'\n', ''), E'\t', '')
                in ('true', '(true)', '1=1', '(1=1)')
            or exists (
                select
                    1
                from
                    pg_catalog.regexp_match(
                        p.qual,
                        $re$\A\s*\(*\s*bucket_id\s*=\s*('(?:[^']|'')*')(\s*::\s*[[:alnum:]_\.]+)?\s*\)*\s*\Z$re$,
                        'i'
                    ) as bucket_match(matches)
                where
                    bucket_match.matches[1] = '''' || replace(b.bucket_id, '''', '''''') || ''''
            )
        )
),
affected_buckets as (
    select
        bucket_id,
        bucket_name,
        array_agg(policyname order by policyname) as policy_names,
        count(*)::int as policy_count
    from
        matching_policies
    group by
        bucket_id,
        bucket_name
)
select
    'public_bucket_allows_listing' as name,
    'Public Bucket Allows Listing' as title,
    'WARN' as level,
    'EXTERNAL' as facing,
    array['SECURITY'] as categories,
    'Detects public storage buckets with a broad SELECT policy on `storage.objects`, which allows clients to list all files in the bucket.' as description,
    format(
        'Public bucket `%s` has %s broad SELECT %s on `storage.objects` (%s), allowing clients to list all files. Public buckets don''t need this for object URL access and it may expose more data than intended.',
        bucket_name,
        policy_count,
        case
            when policy_count = 1 then 'policy'
            else 'policies'
        end,
        array_to_string(policy_names, ', ')
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0025_public_bucket_allows_listing' as remediation,
    jsonb_build_object(
        'schema', 'storage',
        'name', bucket_name,
        'type', 'bucket',
        'bucket_id', bucket_id,
        'bucket_name', bucket_name,
        'policy_names', policy_names,
        'policy_count', policy_count
    ) as metadata,
    format('public_bucket_allows_listing_%s', bucket_id) as cache_key
from
    affected_buckets
order by
    bucket_id)
union all
(
-- Detects tables, views, materialized views, and foreign tables whose
-- schema is visible to the `anon` role through pg_graphql introspection.
-- pg_graphql introspection reflects the calling role's Postgres
-- privileges: if `anon` can SELECT an object, its name, columns, and
-- relationships are visible via /graphql/v1 to anyone with the public
-- anon key, even when RLS is enabled.
--
-- See lint 0027 for the equivalent check against the `authenticated`
-- role. In default Supabase projects the two roles share the same
-- default-privilege grants, so revoking from `anon` alone often leaves
-- the introspection response served to any signed-up user unchanged.
--
-- The privilege check uses has_column_privilege rather than
-- has_table_privilege so that column-level grants such as
-- `GRANT SELECT (some_col) ON t TO anon` are also caught: pg_graphql
-- exposes a relation in introspection if any column is selectable to
-- the calling role (`load_sql_context.sql:327`,
-- `is_column_selectable`).
--
-- The relkind set ('r','v','m','f') matches pg_graphql's own filter in
-- load_sql_context.sql. Partitioned table roots ('p') are excluded
-- because pg_graphql does not expose them their leaf partitions ('r')
-- are still picked up.
with graphql_installed as (
    select 1 as installed
    from pg_catalog.pg_extension
    where extname = 'pg_graphql'
),
exposed_objects as (
    select
        n.nspname as schema_name,
        c.relname as object_name,
        c.relkind as object_relkind,
        case c.relkind
            when 'r' then 'table'
            when 'v' then 'view'
            when 'm' then 'materialized view'
            when 'f' then 'foreign table'
        end as object_type
    from
        pg_catalog.pg_class c
        join pg_catalog.pg_namespace n
            on c.relnamespace = n.oid
    where
        c.relkind in ('r', 'v', 'm', 'f') -- tables, views, materialized views, foreign tables matches pg_graphql
        and exists (
            -- Any selectable column (table-level or column-level grant)
            select 1
            from pg_catalog.pg_attribute a
            where a.attrelid = c.oid
                and a.attnum > 0
                and not a.attisdropped
                and pg_catalog.has_column_privilege('anon', c.oid, a.attnum, 'SELECT')
        )
        and exists (select 1 from graphql_installed)
        and n.nspname not in (
            '_timescaledb_cache', '_timescaledb_catalog', '_timescaledb_config', '_timescaledb_internal', 'auth', 'cron', 'extensions', 'graphql', 'graphql_public', 'information_schema', 'net', 'pgmq', 'pgroonga', 'pgsodium', 'pgsodium_masks', 'pgtle', 'pgbouncer', 'pg_catalog', 'realtime', 'repack', 'storage', 'supabase_functions', 'supabase_migrations', 'tiger', 'topology', 'vault'
        )
)
select
    'pg_graphql_anon_table_exposed' as name,
    'Public Can See Object in GraphQL Schema' as title,
    'WARN' as level,
    'EXTERNAL' as facing,
    array['SECURITY'] as categories,
    'Detects tables, views, materialized views, and foreign tables that are visible in the GraphQL schema to anyone using your public anon key. Revoke `SELECT` from `anon` for objects that should not be discoverable before sign-in, and check lint 0027 for the matching signed-in-user exposure.' as description,
    format(
        '%s `%s.%s` is visible in the GraphQL schema because the `anon` role can `SELECT` it. Revoke `SELECT` from `anon` if it should not be discoverable without signing in.',
        object_type,
        schema_name,
        object_name
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0026_pg_graphql_anon_table_exposed' as remediation,
    jsonb_build_object(
        'schema', schema_name,
        'name', object_name,
        'type', object_type
    ) as metadata,
    format(
        'pg_graphql_anon_table_exposed_%s_%s',
        schema_name,
        object_name
    ) as cache_key
from
    exposed_objects
order by
    schema_name,
    object_name)
union all
(
-- Detects tables, views, materialized views, and foreign tables whose
-- schema is visible to the `authenticated` role through pg_graphql
-- introspection. pg_graphql introspection reflects the calling role's
-- Postgres privileges: if `authenticated` can SELECT an object, its
-- name, columns, and relationships are visible via /graphql/v1 to any
-- signed-up user, even when RLS is enabled.
--
-- See lint 0026 for the equivalent check against the `anon` role. In
-- default Supabase projects the two roles share the same
-- default-privilege grants — `ALTER DEFAULT PRIVILEGES FOR ROLE
-- supabase_admin ... TO anon, authenticated, service_role` — so an
-- object readable by `anon` is typically also readable by
-- `authenticated`. If signup is open or auto-confirm, "authenticated"
-- is in practice anyone with a throwaway email rather than a
-- meaningfully smaller audience.
--
-- The privilege check uses has_column_privilege rather than
-- has_table_privilege so that column-level grants such as
-- `GRANT SELECT (some_col) ON t TO authenticated` are also caught:
-- pg_graphql exposes a relation in introspection if any column is
-- selectable to the calling role (`load_sql_context.sql:327`,
-- `is_column_selectable`).
--
-- The relkind set ('r','v','m','f') matches pg_graphql's own filter in
-- load_sql_context.sql. Partitioned table roots ('p') are excluded
-- because pg_graphql does not expose them their leaf partitions ('r')
-- are still picked up.
with graphql_installed as (
    select 1 as installed
    from pg_catalog.pg_extension
    where extname = 'pg_graphql'
),
exposed_objects as (
    select
        n.nspname as schema_name,
        c.relname as object_name,
        c.relkind as object_relkind,
        case c.relkind
            when 'r' then 'table'
            when 'v' then 'view'
            when 'm' then 'materialized view'
            when 'f' then 'foreign table'
        end as object_type
    from
        pg_catalog.pg_class c
        join pg_catalog.pg_namespace n
            on c.relnamespace = n.oid
    where
        c.relkind in ('r', 'v', 'm', 'f') -- tables, views, materialized views, foreign tables matches pg_graphql
        and exists (
            -- Any selectable column (table-level or column-level grant)
            select 1
            from pg_catalog.pg_attribute a
            where a.attrelid = c.oid
                and a.attnum > 0
                and not a.attisdropped
                and pg_catalog.has_column_privilege('authenticated', c.oid, a.attnum, 'SELECT')
        )
        and exists (select 1 from graphql_installed)
        and n.nspname not in (
            '_timescaledb_cache', '_timescaledb_catalog', '_timescaledb_config', '_timescaledb_internal', 'auth', 'cron', 'extensions', 'graphql', 'graphql_public', 'information_schema', 'net', 'pgmq', 'pgroonga', 'pgsodium', 'pgsodium_masks', 'pgtle', 'pgbouncer', 'pg_catalog', 'realtime', 'repack', 'storage', 'supabase_functions', 'supabase_migrations', 'tiger', 'topology', 'vault'
        )
)
select
    'pg_graphql_authenticated_table_exposed' as name,
    'Signed-In Users Can See Object in GraphQL Schema' as title,
    'WARN' as level,
    'EXTERNAL' as facing,
    array['SECURITY'] as categories,
    'Detects tables, views, materialized views, and foreign tables that are visible in the GraphQL schema to signed-in users. Revoke `SELECT` from `authenticated` for objects that signed-in users should not discover, and check lint 0026 for the matching public exposure.' as description,
    format(
        '%s `%s.%s` is visible in the GraphQL schema to signed-in users because the `authenticated` role can `SELECT` it. Revoke `SELECT` from `authenticated` if it should not be discoverable to every account.',
        object_type,
        schema_name,
        object_name
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0027_pg_graphql_authenticated_table_exposed' as remediation,
    jsonb_build_object(
        'schema', schema_name,
        'name', object_name,
        'type', object_type
    ) as metadata,
    format(
        'pg_graphql_authenticated_table_exposed_%s_%s',
        schema_name,
        object_name
    ) as cache_key
from
    exposed_objects
order by
    schema_name,
    object_name)
union all
(
-- Detects SECURITY DEFINER functions that the `anon` role has EXECUTE
-- on. SECURITY DEFINER functions run with the privileges of their
-- owner (typically a superuser-equivalent role) rather than the
-- caller, so they bypass RLS on every table they read or write.
-- Granting EXECUTE to `anon` therefore lets any holder of the public
-- anon key invoke a privileged operation through PostgREST
-- `/rest/v1/rpc/<name>`, and through `/graphql/v1` Query/Mutation
-- fields when pg_graphql is installed and the return type is
-- supported.
--
-- See lint 0029 for the equivalent check against the `authenticated`
-- role. Together with 0026/0027 they cover the four direct
-- exposure paths: anon-vs-authenticated × table-vs-function.
--
-- This lint deliberately does not gate on pg_graphql being installed:
-- PostgREST exposes `/rest/v1/rpc` independently and is the more
-- commonly deployed surface for the same risk.
select
    'anon_security_definer_function_executable' as name,
    'Public Can Execute SECURITY DEFINER Function' as title,
    'WARN' as level,
    'EXTERNAL' as facing,
    array['SECURITY'] as categories,
    'Detects `SECURITY DEFINER` functions that are callable without signing in. Revoke `EXECUTE`, switch the function to `SECURITY INVOKER`, or move it out of your exposed API schema if it is not meant to be public.' as description,
    format(
        'Function `%s.%s(%s)` can be executed by the `anon` role as a `SECURITY DEFINER` function via `/rest/v1/rpc/%s`. Revoke `EXECUTE` or switch it to `SECURITY INVOKER` if that is not intentional.',
        schema_name,
        function_name,
        function_args,
        function_name
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0028_anon_security_definer_function_executable' as remediation,
    jsonb_build_object(
        'schema', schema_name,
        'name', function_name,
        'arguments', function_args,
        'language', function_language,
        'security_definer', true
    ) as metadata,
    format(
        'anon_security_definer_function_executable_%s_%s_%s',
        schema_name,
        function_name,
        function_args
    ) as cache_key
from
    (
        select
            n.nspname as schema_name,
            p.proname as function_name,
            pg_catalog.pg_get_function_identity_arguments(p.oid) as function_args,
            l.lanname as function_language
        from
            pg_catalog.pg_proc p
            join pg_catalog.pg_namespace n
                on p.pronamespace = n.oid
            join pg_catalog.pg_language l
                on p.prolang = l.oid
        where
            p.prosecdef = true
            and pg_catalog.has_function_privilege('anon', p.oid, 'EXECUTE')
            and n.nspname = any(array(select trim(unnest(string_to_array(coalesce(current_setting('pgrst.db_schemas', 't'), 'public'), ',')))))
            and n.nspname not in (
                '_timescaledb_cache', '_timescaledb_catalog', '_timescaledb_config', '_timescaledb_internal', 'auth', 'cron', 'extensions', 'graphql', 'graphql_public', 'information_schema', 'net', 'pgmq', 'pgroonga', 'pgsodium', 'pgsodium_masks', 'pgtle', 'pgbouncer', 'pg_catalog', 'realtime', 'repack', 'storage', 'supabase_functions', 'supabase_migrations', 'tiger', 'topology', 'vault'
            )
    ) exposed_functions
order by
    schema_name,
    function_name,
    function_args)
union all
(
-- Detects SECURITY DEFINER functions that the `authenticated` role
-- has EXECUTE on. SECURITY DEFINER functions run with the privileges
-- of their owner (typically a superuser-equivalent role) rather than
-- the caller, so they bypass RLS on every table they read or write.
-- Granting EXECUTE to `authenticated` therefore lets any signed-up
-- user invoke a privileged operation through PostgREST
-- `/rest/v1/rpc/<name>`, and through `/graphql/v1` Query/Mutation
-- fields when pg_graphql is installed and the return type is
-- supported. Under open or auto-confirm signup, "authenticated" is in
-- practice anyone with a throwaway email.
--
-- See lint 0028 for the equivalent check against the `anon` role.
-- Together with 0026/0027 they cover the four direct exposure paths:
-- anon-vs-authenticated × table-vs-function.
--
-- This lint deliberately does not gate on pg_graphql being installed:
-- PostgREST exposes `/rest/v1/rpc` independently and is the more
-- commonly deployed surface for the same risk.
select
    'authenticated_security_definer_function_executable' as name,
    'Signed-In Users Can Execute SECURITY DEFINER Function' as title,
    'WARN' as level,
    'EXTERNAL' as facing,
    array['SECURITY'] as categories,
    'Detects `SECURITY DEFINER` functions that are callable by signed-in users. Revoke `EXECUTE`, switch the function to `SECURITY INVOKER`, or move it out of your exposed API schema if signed-in users should not call it.' as description,
    format(
        'Function `%s.%s(%s)` can be executed by the `authenticated` role as a `SECURITY DEFINER` function via `/rest/v1/rpc/%s`. Revoke `EXECUTE` or switch it to `SECURITY INVOKER` if that is not intentional.',
        schema_name,
        function_name,
        function_args,
        function_name
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0029_authenticated_security_definer_function_executable' as remediation,
    jsonb_build_object(
        'schema', schema_name,
        'name', function_name,
        'arguments', function_args,
        'language', function_language,
        'security_definer', true
    ) as metadata,
    format(
        'authenticated_security_definer_function_executable_%s_%s_%s',
        schema_name,
        function_name,
        function_args
    ) as cache_key
from
    (
        select
            n.nspname as schema_name,
            p.proname as function_name,
            pg_catalog.pg_get_function_identity_arguments(p.oid) as function_args,
            l.lanname as function_language
        from
            pg_catalog.pg_proc p
            join pg_catalog.pg_namespace n
                on p.pronamespace = n.oid
            join pg_catalog.pg_language l
                on p.prolang = l.oid
        where
            p.prosecdef = true
            and pg_catalog.has_function_privilege('authenticated', p.oid, 'EXECUTE')
            and n.nspname = any(array(select trim(unnest(string_to_array(coalesce(current_setting('pgrst.db_schemas', 't'), 'public'), ',')))))
            and n.nspname not in (
                '_timescaledb_cache', '_timescaledb_catalog', '_timescaledb_config', '_timescaledb_internal', 'auth', 'cron', 'extensions', 'graphql', 'graphql_public', 'information_schema', 'net', 'pgmq', 'pgroonga', 'pgsodium', 'pgsodium_masks', 'pgtle', 'pgbouncer', 'pg_catalog', 'realtime', 'repack', 'storage', 'supabase_functions', 'supabase_migrations', 'tiger', 'topology', 'vault'
            )
    ) exposed_functions
order by
    schema_name,
    function_name,
    function_args)
union all
(select
    'autovacuum_disabled' as name,
    'Autovacuum Disabled' as title,
    'INFO' as level,
    'EXTERNAL' as facing,
    array['PERFORMANCE'] as categories,
    'Table has autovacuum_enabled=false set as a storage parameter. Without autovacuum, dead tuples accumulate and cause table bloat — see the docs for legitimate exceptions before acting.' as description,
    format(
        'Table `%s`.`%s` has autovacuum_enabled=false set as a storage parameter.',
        nsp.nspname,
        cls.relname
    ) as detail,
    'https://supabase.com/docs/guides/database/database-linter?lint=0030_autovacuum_disabled' as remediation,
    jsonb_build_object(
        'schema', nsp.nspname,
        'name', cls.relname,
        'type', 'table'
    ) as metadata,
    format('autovacuum_disabled_%s_%s', nsp.nspname, cls.relname) as cache_key
from pg_catalog.pg_class cls
join pg_catalog.pg_namespace nsp on cls.relnamespace = nsp.oid
where
    cls.relkind = 'r'
    and 'autovacuum_enabled=false' = any(cls.reloptions)
    and nsp.nspname not in (
        '_timescaledb_cache', '_timescaledb_catalog', '_timescaledb_config',
        '_timescaledb_internal', 'auth', 'cron', 'extensions', 'graphql',
        'graphql_public', 'information_schema', 'net', 'pgmq', 'pgroonga',
        'pgsodium', 'pgsodium_masks', 'pgtle', 'pgbouncer', 'pg_catalog',
        'realtime', 'repack', 'storage', 'supabase_functions',
        'supabase_migrations', 'tiger', 'topology', 'vault'
    )
order by
    nsp.nspname,
    cls.relname)