Fix missing object-level authorization (IDOR) in Python and Flask

def get_invoice(db: sqlite3.Connection, user: User, invoice_id: int) -> dict | None:
    row = db.execute("""
        SELECT id, tenant_id, owner_id, number, amount_cents, status FROM invoices
        WHERE id = ?""",
        (invoice_id,)).fetchone()
    return None if row is None else dict(row)


def list_invoices(db: sqlite3.Connection, user: User) -> list[dict]:
    rows = db.execute("""
        SELECT id, tenant_id, owner_id, number, amount_cents, status FROM invoices
        WHERE tenant_id = ?
        ORDER BY id""",
        (user.tenant_id,))
    return [dict(row) for row in rows]


def export_invoices(db: sqlite3.Connection, user: User, ids: list[int]) -> list[dict]:
    rows = db.execute("""
        SELECT id, tenant_id, owner_id, number, amount_cents, status FROM invoices
        WHERE id IN (SELECT value FROM json_each(?))
        ORDER BY id""",
        (json.dumps(ids),))
    return [dict(row) for row in rows]

# Members read their own invoices; billing admins read every invoice in their own
# tenant. The tenant test sits outside the parentheses, so it binds admins too:
#   tenant_id = :tenant AND (owner_id = :user OR :is_billing_admin)
def _scope(user: User) -> dict:
    return {"tenant": user.tenant_id, "user": user.id, "is_billing_admin": user.role == "billing_admin"}


def get_invoice(db: sqlite3.Connection, user: User, invoice_id: int) -> dict | None:
    row = db.execute("""
        SELECT id, tenant_id, owner_id, number, amount_cents, status FROM invoices
        WHERE id = :id AND tenant_id = :tenant AND (owner_id = :user OR :is_billing_admin)""",
        {"id": invoice_id, **_scope(user)}).fetchone()
    return None if row is None else dict(row)


def list_invoices(db: sqlite3.Connection, user: User) -> list[dict]:
    rows = db.execute("""
        SELECT id, tenant_id, owner_id, number, amount_cents, status FROM invoices
        WHERE tenant_id = :tenant AND (owner_id = :user OR :is_billing_admin)
        ORDER BY id""",
        _scope(user))
    return [dict(row) for row in rows]


def export_invoices(db: sqlite3.Connection, user: User, ids: list[int]) -> list[dict]:
    rows = db.execute("""
        SELECT id, tenant_id, owner_id, number, amount_cents, status FROM invoices
        WHERE id IN (SELECT value FROM json_each(:ids))
          AND tenant_id = :tenant AND (owner_id = :user OR :is_billing_admin)
        ORDER BY id""",
        {"ids": json.dumps(ids), **_scope(user)})
    return [dict(row) for row in rows]

def _can_read(user: User, invoice: sqlite3.Row) -> bool:
    return invoice["owner_id"] == user.id or user.role == "billing_admin"


def get_invoice(db: sqlite3.Connection, user: User, invoice_id: int) -> dict | None:
    row = db.execute("""
        SELECT id, tenant_id, owner_id, number, amount_cents, status FROM invoices
        WHERE id = ?""",
        (invoice_id,)).fetchone()
    return dict(row) if row is not None and _can_read(user, row) else None


def list_invoices(db: sqlite3.Connection, user: User) -> list[dict]:
    rows = db.execute("""
        SELECT id, tenant_id, owner_id, number, amount_cents, status FROM invoices
        ORDER BY id""")
    return [dict(row) for row in rows if _can_read(user, row)]


def export_invoices(db: sqlite3.Connection, user: User, ids: list[int]) -> list[dict]:
    rows = db.execute("""
        SELECT id, tenant_id, owner_id, number, amount_cents, status FROM invoices
        WHERE id IN (SELECT value FROM json_each(?))
        ORDER BY id""",
        (json.dumps(ids),))
    return [dict(row) for row in rows if _can_read(user, row)]

@app.before_request
def load_signed_in_user() -> None:
    # Flask signs the session cookie with SECRET_KEY, so a client cannot change
    # user_id. Tenant and role come from the users table, never from the request.
    user_id = session.get("user_id")
    row = None if user_id is None else get_db().execute(
        "SELECT id, tenant_id, role FROM users WHERE id = ?", (user_id,)).fetchone()
    g.user = None if row is None else User(**row)

def login_required(view):
    @wraps(view)
    def wrapped(**kwargs):
        if g.user is None:
            return {"error": "sign in first"}, 401
        return view(**kwargs)
    return wrapped

@app.get("/invoices")
@login_required
def list_invoices():
    return impl.list_invoices(get_db(), g.user)

@app.get("/invoices/<int(max=999999999):invoice_id>")
@login_required
def get_invoice(invoice_id: int):
    invoice = impl.get_invoice(get_db(), g.user, invoice_id)
    # Missing and not yours get the same answer, so IDs cannot be probed.
    return invoice if invoice is not None else ({"error": "not found"}, 404)

@app.get("/invoices/export")
@login_required
def export_invoices():
    ids = parse_ids(request.args.get("ids"))
    if ids is None:
        return {"error": f"ids must be 1 to {MAX_EXPORT_IDS} invoice IDs, comma-separated"}, 400
    invoices = impl.export_invoices(get_db(), g.user, ids)
    # All or nothing: one missing or unreadable ID refuses the whole export, so a
    # partial file is never mistaken for the full one.
    if len(invoices) != len(ids):
        return {"error": "not found"}, 404
    return to_csv(invoices), {"Content-Type": "text/csv; charset=utf-8"}