docqube_backend/scripts/b1_preflight.py

"""
B1 step 2 — is this database safe to turn tenant isolation on against?

**Read-only.** It opens a connection, runs SELECTs, and prints. It writes
nothing, changes no settings, and is safe to point at production. That is the
whole design: the decision it informs is a production decision, so the check has
to run there.

    APP_ENV=production python scripts/b1_preflight.py
    python scripts/b1_preflight.py --database-url postgresql://.../docqube

Run it *before* deleting the `tenant_id IS NULL` fallback (step 3) and before
enabling `TENANT_FILTER_ENABLED` (step 4). The order matters: enable the filter
while the fallback still exists and any request that loses its tenant becomes a
superadmin request — the exact escalation B1 exists to remove.

Exit code is 0 if nothing needs a decision, 1 if something does. Nothing here is
a hard failure — a finding means "fix the data or accept it knowingly", not
"stop".
"""

import argparse
import os
import sys
from pathlib import Path

os.environ.setdefault("PYTHONIOENCODING", "utf-8")
sys.path.insert(0, str(Path(__file__).parent.parent))

# Read the table list from the migration itself, so the two cannot disagree.
# Loaded by path rather than imported: `alembic` resolves to the installed
# package, and `alembic/versions/` is a plain directory, not a subpackage of it.
#
# The isolation ratchet (tests/probes/test_isolation_ratchet.py) fails if a
# tenant-owned model is missing from that migration, so this list cannot drift
# silently away from the models either.
import importlib.util  # noqa: E402

_spec = importlib.util.spec_from_file_location(
    "_b1_3_rls",
    Path(__file__).parent.parent / "alembic" / "versions" / "b1_3_row_level_security.py",
)
_rls = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(_rls)
TENANT_TABLES = _rls.TENANT_TABLES

FINDINGS: list[str] = []


def _rule(title: str) -> None:
    print(f"\n{title}\n{'-' * len(title)}")


def check_migration_applied(conn) -> bool:
    from sqlalchemy import text

    exists = conn.execute(
        text(
            "SELECT 1 FROM information_schema.columns "
            "WHERE table_name = 'users' AND column_name = 'is_superadmin'"
        )
    ).first()
    _rule("1. Has b1_0 been deployed?")
    if not exists:
        print("  NO — users.is_superadmin does not exist.")
        print("  Steps 3-5 cannot proceed. Deploy the migration first (step 1).")
        FINDINGS.append("b1_0 not applied")
        return False
    print("  Yes — users.is_superadmin exists.")
    return True


def check_fallback_dependents(conn) -> None:
    """
    Who currently gets privilege *only* from `tenant_id IS NULL`?

    These are the people who lose access the moment the fallback is deleted.
    b1_0 backfilled the ones that existed when it ran; anyone created since,
    with a null tenant and the flag unset, is relying on the fallback today.
    """
    from sqlalchemy import text

    _rule("2. Who relies on the `tenant_id IS NULL` fallback?")
    rows = conn.execute(
        text(
            "SELECT id, email, is_deleted FROM users "
            "WHERE tenant_id IS NULL AND is_superadmin = false "
            "ORDER BY is_deleted, id"
        )
    ).fetchall()

    live = [r for r in rows if not r[2]]
    if not live:
        print("  None. Every tenant-less operator carries the explicit flag.")
        if rows:
            print(f"  ({len(rows)} deleted user(s) also match; they do not matter.)")
        return

    print(f"  {len(live)} ACTIVE user(s) HAVE ALREADY LOST superadmin:")
    for uid, email, _ in live:
        print(f"    id={uid}  {email}")
    print()
    print("  This is no longer a warning about a future step. The fallback has")
    print("  been deleted, so on the deploy carrying that change these accounts")
    print("  stop being superadmins. Fix the data before deploying:")
    print("    UPDATE users SET is_superadmin = true WHERE id IN (...);")
    print()
    print("  If this database has not had migration b1_0 applied yet, that is")
    print("  the real answer — apply it first; it backfills exactly these rows.")
    FINDINGS.append(f"{len(live)} active user(s) lose superadmin on deploy")


def check_superadmin_population(conn) -> None:
    from sqlalchemy import text

    _rule("3. How many superadmins are there?")
    total, supers = conn.execute(
        text(
            "SELECT count(*), count(*) FILTER (WHERE is_superadmin) "
            "FROM users WHERE is_deleted = false"
        )
    ).first()
    print(f"  {supers} of {total} active users hold is_superadmin.")
    if total and supers / total > 0.10:
        print("  That is a large share. A superadmin bypasses tenant scoping")
        print("  entirely, so this is worth eyeballing before step 4.")
        FINDINGS.append(f"{supers}/{total} users are superadmin")
    for (email,) in conn.execute(
        text(
            "SELECT email FROM users WHERE is_superadmin AND is_deleted = false "
            "ORDER BY id LIMIT 20"
        )
    ):
        print(f"    {email}")


def check_null_tenant_rows(conn) -> None:
    """
    The one that actually bites.

    The RLS policy reads `... OR tenant_id IS NULL OR tenant_id = <session>`.
    The null branch is deliberate — it keeps shared system records such as
    global roles readable by everyone. But it applies to *every* row in the
    table, so a data row that lost its tenant is not hidden by RLS; it is
    published to every tenant at once.
    """
    from sqlalchemy import text

    _rule("4. Rows with no tenant (visible to EVERY tenant under RLS)")
    offenders = []
    for table in TENANT_TABLES:
        present = conn.execute(
            text(
                "SELECT 1 FROM information_schema.columns "
                "WHERE table_name = :t AND column_name = 'tenant_id'"
            ),
            {"t": table},
        ).first()
        if not present:
            print(f"  {table:<28} (no tenant_id column — skipped)")
            continue
        n = conn.execute(text(f"SELECT count(*) FROM {table} WHERE tenant_id IS NULL")).scalar()
        if n:
            offenders.append((table, n))
        print(f"  {table:<28}{n:>10} null-tenant row(s)")

    if not offenders:
        print("\n  Clean. Every row carries a tenant.")
        return

    print("\n  These rows will be readable by every tenant once RLS is on.")
    print("  For `roles` a shared row is usually intentional (global role")
    print("  templates). For drive_files, projects or notifications it is not —")
    print("  assign the tenant, or delete the row, before step 5.")
    FINDINGS.append(
        "null-tenant rows in: " + ", ".join(f"{t} ({n})" for t, n in offenders)
    )


def check_connection_role(conn) -> None:
    """
    RLS does not apply to superusers, and it does not apply to a role with
    BYPASSRLS. Deploying the policy under such a role silently does nothing —
    every test passes, every query still returns everything.
    """
    from sqlalchemy import text

    _rule("5. Can the application's role bypass RLS?")
    name, is_super, bypass = conn.execute(
        text(
            "SELECT current_user, "
            "       (SELECT rolsuper FROM pg_roles WHERE rolname = current_user), "
            "       (SELECT rolbypassrls FROM pg_roles WHERE rolname = current_user)"
        )
    ).first()
    print(f"  connected as: {name}  (superuser={is_super}, bypassrls={bypass})")
    if is_super or bypass:
        print("\n  This role IGNORES row-level security. If the application")
        print("  connects as this role, enabling RLS changes nothing at all.")
        print("  Create the unprivileged role first:")
        print("    python scripts/create_app_role.py --role docqube_app --password ...")
        FINDINGS.append(f"connection role {name} bypasses RLS")
    else:
        print("  No — RLS will apply to this connection.")


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--database-url",
        help="Override the URL from settings. Use to point at production "
        "without changing APP_ENV.",
    )
    args = parser.parse_args()

    from sqlalchemy import create_engine

    url = args.database_url
    if not url:
        from app.core.settings import settings

        url = settings.DATABASE_URL

    shown = url.split("@")[-1] if "@" in url else url
    print(f"B1 preflight — read-only checks against {shown}")

    engine = create_engine(url)
    try:
        with engine.connect() as conn:
            if check_migration_applied(conn):
                check_fallback_dependents(conn)
                check_superadmin_population(conn)
            check_null_tenant_rows(conn)
            check_connection_role(conn)
    finally:
        engine.dispose()

    print("\n" + "=" * 60)
    if not FINDINGS:
        print("Nothing needs a decision. Steps 3-5 are safe to sequence.")
        return 0
    print(f"{len(FINDINGS)} thing(s) need a decision before proceeding:")
    for f in FINDINGS:
        print(f"  - {f}")
    print("\nNone of these is automatically fatal. Each is a data fix or an")
    print("informed acceptance — but make it knowingly, not by deploying past it.")
    return 1


if __name__ == "__main__":
    raise SystemExit(main())