feat: add an authorization API (#4363)

Add an authorization API to the LanceDB catalog. This is an API that can
be used to bind access control lists (ACLs) to individual catalog
objects. It can also be used to manage server-side roles, and introspect
principals and groups. The new API is implemented in both the Rust,
Python, and async Python styles and includes list_acls, add_acl,
delete_acl, list_role_bindings, add_role_binding, delete_role_binding,
get_principal, and get_group. This API implements the design principle
of forwards-compatibility. So unknown privilege, object, and subject
types do not cause errors, but are be supported as best as possible in
the library.

python/python/lancedb/_lancedb.pyi

@@ -955,6 +955,8 @@ class PermutationReader:
     def __init__(self, base_table: Table, permutation_table: Table): ...
 
 class Catalog:
+    @property
+    def authz(self) -> AuthorizationClient: ...
     @property
     def uri(self) -> str: ...
     async def create_database(
@@ -1001,3 +1003,60 @@ class JobListing:
     def page_token(self) -> Optional[str]: ...
     def __aiter__(self) -> JobListing: ...
     async def __anext__(self) -> JobInfo: ...
+
+class AuthorizationClient:
+    async def list_acls(self, request: str) -> str: ...
+    async def add_acl(self, request: str) -> None: ...
+    async def delete_acl(self, request: str) -> None: ...
+    async def list_role_bindings(self, request: str) -> str: ...
+    async def add_role_binding(self, request: str) -> None: ...
+    async def delete_role_binding(self, request: str) -> None: ...
+    async def get_principal(self, request: str) -> str: ...
+    async def get_group(self, request: str) -> str: ...
+
+class AuthzSubject:
+    def __init__(self, kind: str, value: str): ...
+    @staticmethod
+    def from_canonical(value: str) -> AuthzSubject: ...
+    @property
+    def kind(self) -> str: ...
+    @property
+    def value(self) -> str: ...
+    @property
+    def display_value(self) -> str: ...
+    def to_wire(self, context: Optional[str] = None) -> str: ...
+    def __str__(self) -> str: ...
+    def __repr__(self) -> str: ...
+
+class AuthzPrivilege:
+    def __init__(self, value: str): ...
+    @staticmethod
+    def known() -> list[str]: ...
+    def __str__(self) -> str: ...
+
+class AuthzObject:
+    def __init__(self, value: str): ...
+    @staticmethod
+    def system() -> AuthzObject: ...
+    @staticmethod
+    def database(name: str) -> AuthzObject: ...
+    @staticmethod
+    def namespace(database: str, namespace_path: Sequence[str]) -> AuthzObject: ...
+    @staticmethod
+    def table(
+        database: str, namespace_path: Sequence[str], name: str
+    ) -> AuthzObject: ...
+    @staticmethod
+    def view(
+        database: str, namespace_path: Sequence[str], name: str
+    ) -> AuthzObject: ...
+    @staticmethod
+    def secret(
+        database: str, namespace_path: Sequence[str], name: str
+    ) -> AuthzObject: ...
+    @staticmethod
+    def function(
+        database: str, namespace_path: Sequence[str], name: str
+    ) -> AuthzObject: ...
+    def __str__(self) -> str: ...
+    def __repr__(self) -> str: ...

python/python/lancedb/authz.py

@@ -0,0 +1,588 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+"""Deployment-wide authorization through ``catalog.authz``.
+
+ACLs grant one privilege on an object to a principal, group, or role. Role
+bindings grant a role to a principal or group. Identity-provider group membership
+is read-only. Name resolution and permission checks are performed by the server.
+"""
+
+from collections.abc import AsyncIterator, Iterator, Sequence
+from dataclasses import dataclass, field
+from enum import Enum
+import json
+from typing import Literal, Optional, TypeVar, Union, cast
+
+from . import _lancedb
+from .background_loop import LOOP
+
+__all__ = [
+    "Authorization",
+    "AsyncAuthorization",
+    "Subject",
+    "Object",
+    "Privilege",
+    "AccessControlEntry",
+    "RoleBinding",
+    "IdentityInfo",
+    "PrincipalInfo",
+    "GroupInfo",
+]
+
+_PAGE_SIZE = 100
+_T = TypeVar("_T")
+_SubjectKind = Literal[
+    "principal_id",
+    "principal_name",
+    "group_id",
+    "group_name",
+    "role",
+    "principal_api_key",
+    "unknown",
+]
+
+
+@dataclass(frozen=True)
+class Subject:
+    """An immutable identity selector. Prefer the named factory methods.
+
+    IDs are stable; names must resolve unambiguously on the server. API-key
+    selectors identify the subject of an operation, not its authenticated caller.
+    Secret values are omitted from representations. Unknown canonical subject
+    kinds returned by newer servers use ``kind="unknown"`` and preserve the full
+    wire string in ``value``; they can be passed back to ACL methods.
+    """
+
+    kind: _SubjectKind
+    _value: str = field(repr=False)
+
+    _inner: _lancedb.AuthzSubject = field(init=False, repr=False, compare=False)
+
+    def __post_init__(self):
+        object.__setattr__(
+            self, "_inner", _lancedb.AuthzSubject(self.kind, self._value)
+        )
+
+    @classmethod
+    def principal_id(cls, id: str) -> "Subject":
+        """Select a principal by canonical identity-provider ID."""
+        return cls("principal_id", id)
+
+    @classmethod
+    def principal_name(cls, name: str) -> "Subject":
+        """Select a principal by an unambiguous provider name."""
+        return cls("principal_name", name)
+
+    @classmethod
+    def group_id(cls, id: str) -> "Subject":
+        """Select a group by canonical identity-provider ID."""
+        return cls("group_id", id)
+
+    @classmethod
+    def group_name(cls, name: str) -> "Subject":
+        """Select a group by an unambiguous provider name."""
+        return cls("group_name", name)
+
+    @classmethod
+    def role(cls, name: str) -> "Subject":
+        """Select a LanceDB-managed role, without an ``r:`` prefix."""
+        return cls("role", name)
+
+    @classmethod
+    def principal_api_key(cls, api_key: str) -> "Subject":
+        """Select the principal holding this secret API key."""
+        return cls("principal_api_key", api_key)
+
+    def __repr__(self) -> str:
+        value = self._inner.display_value
+        return f"Subject.{self.kind}({value!r})"
+
+    @property
+    def value(self) -> str:
+        """The selected value, or full wire string for unknown subject kinds.
+
+        API-key values are secrets.
+        """
+        return self._value
+
+    def __reduce__(self):
+        return type(self), (self.kind, self._value)
+
+    def _wire(self) -> str:
+        return self._inner.to_wire()
+
+    @classmethod
+    def _canonical(cls, value: str) -> "Subject":
+        inner = _lancedb.AuthzSubject.from_canonical(value)
+        return cls(cast(_SubjectKind, inner.kind), inner.value)
+
+
+@dataclass(frozen=True)
+class Object:
+    """An immutable resource name in the authorization syntax.
+
+    Use the builders to avoid interpreting separators as part of a name.
+    Methods also accept raw object strings, validated by the same Rust parser.
+    Unknown resource types are preserved for compatibility with newer servers.
+    Omitted or empty namespace paths select the database's public namespace.
+    """
+
+    _value: str
+    _inner: _lancedb.AuthzObject = field(init=False, repr=False, compare=False)
+
+    def __post_init__(self):
+        object.__setattr__(self, "_inner", _lancedb.AuthzObject(self._value))
+
+    @classmethod
+    def system(cls) -> "Object":
+        """The deployment-wide system object."""
+        return cls(str(_lancedb.AuthzObject.system()))
+
+    @classmethod
+    def database(cls, name: str) -> "Object":
+        """A database; slash-containing names remain one logical database."""
+        return cls(str(_lancedb.AuthzObject.database(name)))
+
+    @classmethod
+    def namespace(
+        cls, *, database: str, namespace_path: Sequence[str] = ("public",)
+    ) -> "Object":
+        """A namespace path, defaulting to the database's public namespace."""
+        return cls(str(_lancedb.AuthzObject.namespace(database, namespace_path)))
+
+    @classmethod
+    def table(
+        cls, *, database: str, name: str, namespace_path: Sequence[str] = ("public",)
+    ) -> "Object":
+        """A table, defaulting to the database's public namespace."""
+        return cls(str(_lancedb.AuthzObject.table(database, namespace_path, name)))
+
+    @classmethod
+    def view(
+        cls, *, database: str, name: str, namespace_path: Sequence[str] = ("public",)
+    ) -> "Object":
+        """A view, defaulting to the database's public namespace."""
+        return cls(str(_lancedb.AuthzObject.view(database, namespace_path, name)))
+
+    @classmethod
+    def secret(
+        cls, *, database: str, name: str, namespace_path: Sequence[str] = ("public",)
+    ) -> "Object":
+        """A secret, defaulting to the database's public namespace."""
+        return cls(str(_lancedb.AuthzObject.secret(database, namespace_path, name)))
+
+    @classmethod
+    def function(
+        cls, *, database: str, name: str, namespace_path: Sequence[str] = ("public",)
+    ) -> "Object":
+        """A function, defaulting to the database's public namespace."""
+        return cls(str(_lancedb.AuthzObject.function(database, namespace_path, name)))
+
+    def __str__(self) -> str:
+        return str(self._inner)
+
+    def __reduce__(self):
+        return type(self), (self._value,)
+
+
+class Privilege(str, Enum):
+    """Known privileges, validated and serialized by Rust.
+
+    ACL methods also accept strings such as ``"SELECT"`` and future privilege
+    names such as ``"PRIVILEGE20"``. Future names are preserved verbatim;
+    their meaning and validity are determined by the server.
+
+    Grant ``CREATE_SECRET`` or ``CREATE_FUNCTION`` on a namespace to delegate
+    creation of those resources without transferring namespace ownership.
+    """
+
+    def __new__(cls, value: str):
+        canonical = str(_lancedb.AuthzPrivilege(value))
+        member = str.__new__(cls, canonical)
+        member._value_ = canonical
+        return member
+
+    OWNERSHIP = "OWNERSHIP"
+    USAGE = "USAGE"
+    SELECT = "SELECT"
+    INSERT = "INSERT"
+    UPDATE = "UPDATE"
+    DELETE = "DELETE"
+    CREATE_DATABASE = "CREATE_DATABASE"
+    CREATE_NAMESPACE = "CREATE_NAMESPACE"
+    CREATE_TABLE = "CREATE_TABLE"
+    CREATE_MATERIALIZED_VIEW = "CREATE_MATERIALIZED_VIEW"
+    CREATE_VIEW = "CREATE_VIEW"
+    CREATE_SECRET = "CREATE_SECRET"
+    CREATE_FUNCTION = "CREATE_FUNCTION"
+    OPERATE = "OPERATE"
+
+
+@dataclass(frozen=True)
+class AccessControlEntry:
+    """One object/subject/privilege grant; subject contains a canonical ID or role."""
+
+    object: str
+    subject: Subject
+    subject_name: Optional[str]
+    privilege: str
+
+
+@dataclass(frozen=True)
+class RoleBinding:
+    """A direct principal/group role binding, not an expanded effective role."""
+
+    subject: Subject
+    subject_name: Optional[str]
+    role: str
+
+
+@dataclass(frozen=True)
+class IdentityInfo:
+    """A canonical identity-provider ID and its optional display name."""
+
+    id: str
+    name: Optional[str] = None
+
+
+@dataclass(frozen=True)
+class PrincipalInfo:
+    """A principal's groups and effective roles, including group-inherited roles."""
+
+    principal_id: str
+    principal_name: Optional[str]
+    groups: list[IdentityInfo]
+    roles: list[str]
+
+
+@dataclass(frozen=True)
+class GroupInfo:
+    """A group's provider-managed members and LanceDB-managed direct roles."""
+
+    group_id: str
+    group_name: Optional[str]
+    principals: list[IdentityInfo]
+    roles: list[str]
+
+
+def _subject(subject: Subject, context: Optional[str] = None) -> str:
+    if not isinstance(subject, Subject):
+        raise ValueError("subject must be a Subject selector")
+    return subject._inner.to_wire(context)
+
+
+def _object(value: Union[str, Object]) -> str:
+    if not isinstance(value, (str, Object)) or not str(value):
+        raise ValueError("object must be a nonempty string or Object")
+    return str(value) if isinstance(value, Object) else str(_lancedb.AuthzObject(value))
+
+
+def _limit(limit: Optional[int]) -> None:
+    if limit is not None and (
+        isinstance(limit, bool) or not isinstance(limit, int) or limit < 0
+    ):
+        raise ValueError("limit must be a nonnegative integer or None")
+
+
+def _privilege(value: Union[str, Privilege]) -> str:
+    if not isinstance(value, str):
+        raise ValueError("privilege must be a string or Privilege")
+    return str(_lancedb.AuthzPrivilege(value))
+
+
+def _entry(value: dict) -> AccessControlEntry:
+    return AccessControlEntry(
+        value["object"],
+        Subject._canonical(value["subject_id"]),
+        value.get("subject_name"),
+        value["privilege"],
+    )
+
+
+def _binding(value: dict) -> RoleBinding:
+    return RoleBinding(
+        Subject._canonical(value["subject_id"]),
+        value.get("subject_name"),
+        value["role"],
+    )
+
+
+class AsyncAuthorization:
+    """Async authorization returned by ``AsyncCatalog.authz``.
+
+    HTTP failures raise [HttpError][lancedb.remote.errors.HttpError], preserving
+    status and request ID. Invalid local arguments raise ValueError. Lists are
+    lazy async iterators: failures can occur during iteration after some results
+    have been yielded. Iteration does not provide a consistent snapshot.
+    """
+
+    def __init__(self, inner: _lancedb.AuthorizationClient):
+        self._inner = inner
+
+    async def _pages(self, method, request: dict, key: str, limit: Optional[int]):
+        yielded = 0
+        while limit is None or yielded < limit:
+            request["max_results"] = (
+                _PAGE_SIZE if limit is None else min(_PAGE_SIZE, limit - yielded)
+            )
+            response = json.loads(await method(json.dumps(request)))
+            for value in response[key]:
+                yield value
+                yielded += 1
+                if limit is not None and yielded >= limit:
+                    return
+            token = response.get("continuation_token")
+            if not token:
+                return
+            if token == request.get("continuation_token"):
+                raise RuntimeError(
+                    "Authorization listing returned a repeated continuation token"
+                )
+            request["continuation_token"] = token
+
+    def list_acls(
+        self,
+        *,
+        object: Optional[Union[str, Object]] = None,
+        subject: Optional[Subject] = None,
+        limit: Optional[int] = None,
+    ) -> AsyncIterator[AccessControlEntry]:
+        """Iterate grants, optionally filtered by exact object and/or subject.
+
+        Requires USAGE on the supplied object, or on system if object is omitted.
+        ``limit`` bounds yielded entries across all pages; None lists all and zero
+        makes no requests. Missing matches produce an empty iterator.
+        """
+        _limit(limit)
+        request = {}
+        if object is not None:
+            request["object"] = _object(object)
+        if subject is not None:
+            request["subject"] = _subject(subject)
+
+        async def entries():
+            async for value in self._pages(
+                self._inner.list_acls, request, "entries", limit
+            ):
+                yield _entry(value)
+
+        return entries()
+
+    async def add_acl(
+        self,
+        *,
+        object: Union[str, Object],
+        subject: Subject,
+        privilege: Union[str, Privilege],
+    ) -> None:
+        """Grant one privilege; requires OWNERSHIP on the object.
+
+        A duplicate grant raises HTTP 409. Granting OWNERSHIP transfers ownership.
+        Writes are not automatically retried after ambiguous failures.
+        """
+        await self._inner.add_acl(
+            json.dumps(
+                {
+                    "object": _object(object),
+                    "subject": _subject(subject),
+                    "privilege": _privilege(privilege),
+                }
+            )
+        )
+
+    async def delete_acl(
+        self,
+        *,
+        object: Union[str, Object],
+        subject: Subject,
+        privilege: Union[str, Privilege],
+    ) -> None:
+        """Revoke one privilege; requires OWNERSHIP on the object.
+
+        A missing grant raises HTTP 404. OWNERSHIP itself cannot be deleted.
+        """
+        await self._inner.delete_acl(
+            json.dumps(
+                {
+                    "object": _object(object),
+                    "subject": _subject(subject),
+                    "privilege": _privilege(privilege),
+                }
+            )
+        )
+
+    def list_role_bindings(
+        self,
+        *,
+        subject: Optional[Subject] = None,
+        role: Optional[str] = None,
+        limit: Optional[int] = None,
+    ) -> AsyncIterator[RoleBinding]:
+        """Iterate direct bindings, optionally filtered by principal/group and role.
+
+        Requires USAGE on system. Limit semantics match ``list_acls``. Principals'
+        roles inherited through groups are available through ``get_principal``.
+        """
+        _limit(limit)
+        request = {}
+        if subject is not None:
+            request["subject"] = _subject(subject, "role_member")
+        if role is not None:
+            request["role"] = role
+
+        async def bindings():
+            async for value in self._pages(
+                self._inner.list_role_bindings, request, "bindings", limit
+            ):
+                yield _binding(value)
+
+        return bindings()
+
+    async def add_role_binding(self, *, subject: Subject, role: str) -> None:
+        """Grant a role idempotently; requires OPERATE on system."""
+        await self._inner.add_role_binding(
+            json.dumps({"subject": _subject(subject, "role_member"), "role": role})
+        )
+
+    async def delete_role_binding(self, *, subject: Subject, role: str) -> None:
+        """Idempotently revoke a role; requires OPERATE on system."""
+        await self._inner.delete_role_binding(
+            json.dumps({"subject": _subject(subject, "role_member"), "role": role})
+        )
+
+    async def get_principal(self, subject: Subject) -> PrincipalInfo:
+        """Look up groups and effective roles; requires USAGE on system.
+
+        Select by ID, name, or API key. Missing/ambiguous identities raise HTTP
+        404/409; unavailable or unconfigured resolution raises HTTP 503/501.
+        """
+        value = json.loads(
+            await self._inner.get_principal(
+                json.dumps({"principal": _subject(subject, "principal")})
+            )
+        )
+        return PrincipalInfo(
+            value["principal_id"],
+            value.get("principal_name"),
+            [IdentityInfo(**group) for group in value["groups"]],
+            value["roles"],
+        )
+
+    async def get_group(self, subject: Subject) -> GroupInfo:
+        """Look up a group's members and roles by ID/name; requires USAGE on system.
+
+        Membership is read-only and comes from the identity provider. Lookup
+        errors have the same HTTP semantics as ``get_principal``.
+        """
+        value = json.loads(
+            await self._inner.get_group(
+                json.dumps({"group": _subject(subject, "group")})
+            )
+        )
+        return GroupInfo(
+            value["group_id"],
+            value.get("group_name"),
+            [IdentityInfo(**principal) for principal in value["principals"]],
+            value["roles"],
+        )
+
+
+async def _next(iterator: AsyncIterator[_T]) -> _T:
+    return await iterator.__anext__()
+
+
+def _sync_iter(iterator: AsyncIterator[_T]) -> Iterator[_T]:
+    try:
+        while True:
+            yield LOOP.run(_next(iterator))
+    except StopAsyncIteration:
+        return
+    finally:
+        close = getattr(iterator, "aclose", None)
+        if close is not None:
+            LOOP.run(close())
+
+
+class Authorization:
+    """Synchronous authorization returned by ``Catalog.authz``.
+
+    Shares the async API's permissions, errors, and lazy pagination behavior.
+    Group membership comes from the identity provider and cannot be edited here.
+    """
+
+    def __init__(self, inner: AsyncAuthorization):
+        self._inner = inner
+
+    def list_acls(
+        self,
+        *,
+        object: Optional[Union[str, Object]] = None,
+        subject: Optional[Subject] = None,
+        limit: Optional[int] = None,
+    ) -> Iterator[AccessControlEntry]:
+        """Iterate grants.
+
+        See [list_acls][lancedb.authz.AsyncAuthorization.list_acls] for permissions.
+        """
+        return _sync_iter(
+            self._inner.list_acls(object=object, subject=subject, limit=limit)
+        )
+
+    def add_acl(
+        self,
+        *,
+        object: Union[str, Object],
+        subject: Subject,
+        privilege: Union[str, Privilege],
+    ) -> None:
+        """Grant a privilege; duplicates raise HTTP 409.
+
+        OWNERSHIP transfers ownership.
+        """
+        LOOP.run(
+            self._inner.add_acl(object=object, subject=subject, privilege=privilege)
+        )
+
+    def delete_acl(
+        self,
+        *,
+        object: Union[str, Object],
+        subject: Subject,
+        privilege: Union[str, Privilege],
+    ) -> None:
+        """Revoke a privilege. Missing grants raise HTTP 404.
+
+        OWNERSHIP cannot be deleted.
+        """
+        LOOP.run(
+            self._inner.delete_acl(object=object, subject=subject, privilege=privilege)
+        )
+
+    def list_role_bindings(
+        self,
+        *,
+        subject: Optional[Subject] = None,
+        role: Optional[str] = None,
+        limit: Optional[int] = None,
+    ) -> Iterator[RoleBinding]:
+        """Iterate direct role bindings, using the same limit semantics as list_acls."""
+        return _sync_iter(
+            self._inner.list_role_bindings(subject=subject, role=role, limit=limit)
+        )
+
+    def add_role_binding(self, *, subject: Subject, role: str) -> None:
+        """Grant a role idempotently; requires OPERATE on system."""
+        LOOP.run(self._inner.add_role_binding(subject=subject, role=role))
+
+    def delete_role_binding(self, *, subject: Subject, role: str) -> None:
+        """Idempotently revoke a role; requires OPERATE on system."""
+        LOOP.run(self._inner.delete_role_binding(subject=subject, role=role))
+
+    def get_principal(self, subject: Subject) -> PrincipalInfo:
+        """Look up groups and effective roles; requires USAGE on system."""
+        return LOOP.run(self._inner.get_principal(subject))
+
+    def get_group(self, subject: Subject) -> GroupInfo:
+        """Look up a group's members and roles; requires USAGE on system."""
+        return LOOP.run(self._inner.get_group(subject))

python/python/lancedb/catalog.py

@@ -3,11 +3,13 @@
 
 """Remote catalogs manage databases through a server's root namespace."""
 
+from functools import cached_property
 from datetime import timedelta
 from typing import Any, AsyncIterator, Iterator, Optional, Union
 
 from . import _lancedb
 from .background_loop import LOOP
+from .authz import AsyncAuthorization, Authorization
 from .db import AsyncConnection, DBConnection
 from .remote import ClientConfig, OAuthConfig
 from .remote.db import RemoteDBConnection
@@ -86,6 +88,11 @@ class AsyncCatalog:
     def __init__(self, inner: _lancedb.Catalog):
         self._inner = inner
 
+    @cached_property
+    def authz(self) -> AsyncAuthorization:
+        """Authorization sharing this catalog's scope and credentials."""
+        return AsyncAuthorization(self._inner.authz)
+
     @property
     def uri(self) -> str:
         """The catalog's root namespace endpoint."""
@@ -157,6 +164,11 @@ class Catalog:
         self._sql_host_override = sql_host_override
         self._oauth_config = oauth_config
 
+    @cached_property
+    def authz(self) -> Authorization:
+        """Authorization sharing this catalog's scope and credentials."""
+        return Authorization(self._inner.authz)
+
     @property
     def uri(self) -> str:
         """The catalog's root namespace endpoint."""

python/python/tests/test_authz.py

@@ -0,0 +1,821 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+import asyncio
+from dataclasses import FrozenInstanceError
+import json
+from unittest.mock import AsyncMock
+
+import pytest
+
+import lancedb
+from lancedb.authz import (
+    AccessControlEntry,
+    AsyncAuthorization,
+    Authorization,
+    GroupInfo,
+    IdentityInfo,
+    Object,
+    PrincipalInfo,
+    Privilege,
+    RoleBinding,
+    Subject,
+)
+from lancedb.remote import HeaderProvider
+from lancedb.remote.errors import HttpError
+from test_catalog import catalog_server as _catalog_server
+
+catalog_server = _catalog_server
+
+
+class DynamicHeaders(HeaderProvider):
+    def get_headers(self):
+        return {
+            "Authorization": "Bearer rotating-token",
+            "X-LanceDB-Database": "wrong-dynamic-db",
+            "X-LanceDB-Database-Prefix": "wrong-prefix",
+        }
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+def test_authz_operations_and_scope(catalog_server, asynchronous):
+    endpoint, requests, responses = catalog_server
+    responses.extend(
+        [
+            (200, {}),
+            (200, {}),
+            (200, {}),
+            (200, {}),
+            (
+                200,
+                {
+                    "entries": [
+                        {
+                            "subject_id": "g:team-id",
+                            "object": "table:tenant/db:a$b$events",
+                            "privilege": "PRIVILEGE20",
+                        }
+                    ]
+                },
+            ),
+            (
+                200,
+                {
+                    "bindings": [
+                        {
+                            "subject_id": "p:alice-id",
+                            "subject_name": "Alice",
+                            "role": "reader",
+                        }
+                    ]
+                },
+            ),
+            (
+                200,
+                {
+                    "principal_id": "alice-id",
+                    "groups": [{"id": "team-id"}],
+                    "roles": ["reader"],
+                },
+            ),
+            (
+                200,
+                {
+                    "group_id": "team-id",
+                    "group_name": "Engineering",
+                    "principals": [{"id": "alice-id", "name": "Alice"}],
+                    "roles": ["reader"],
+                },
+            ),
+        ]
+    )
+    config = {
+        "header_provider": DynamicHeaders(),
+        "extra_headers": {"x-lancedb-database": "wrong-static-db"},
+    }
+    obj = Object.table(database="tenant/db", namespace_path=["a", "b"], name="events")
+
+    async def exercise():
+        if asynchronous:
+            catalog = await lancedb.connect_catalog_async(
+                endpoint, client_config=config
+            )
+        else:
+            catalog = lancedb.connect_catalog(endpoint, client_config=config)
+        authz = catalog.authz
+        assert authz is catalog.authz
+        assert requests == []  # Constructing handles does not probe a database.
+
+        async def call(method, *args, **kwargs):
+            value = method(*args, **kwargs)
+            return await value if asynchronous else value
+
+        await call(
+            authz.add_acl,
+            object=obj,
+            subject=Subject.group_name("Engineering"),
+            privilege=Privilege.SELECT,
+        )
+        await call(
+            authz.delete_acl,
+            object=obj,
+            subject=Subject.group_id("team-id"),
+            privilege="SELECT",
+        )
+        await call(
+            authz.add_role_binding,
+            subject=Subject.principal_name("Alice"),
+            role="reader",
+        )
+        await call(
+            authz.delete_role_binding,
+            subject=Subject.principal_api_key("subject-secret"),
+            role="reader",
+        )
+        iterator = authz.list_acls(object=obj, subject=Subject.role("reader"))
+        entries = (
+            [entry async for entry in iterator] if asynchronous else list(iterator)
+        )
+        assert entries == [
+            AccessControlEntry(
+                str(obj), Subject.group_id("team-id"), None, "PRIVILEGE20"
+            )
+        ]
+        iterator = authz.list_role_bindings(
+            subject=Subject.group_id("team-id"), role="reader"
+        )
+        bindings = (
+            [binding async for binding in iterator] if asynchronous else list(iterator)
+        )
+        assert bindings == [
+            RoleBinding(Subject.principal_id("alice-id"), "Alice", "reader")
+        ]
+        assert await call(
+            authz.get_principal, Subject.principal_id("alice-id")
+        ) == PrincipalInfo("alice-id", None, [IdentityInfo("team-id")], ["reader"])
+        assert await call(
+            authz.get_group, Subject.group_name("Engineering")
+        ) == GroupInfo(
+            "team-id", "Engineering", [IdentityInfo("alice-id", "Alice")], ["reader"]
+        )
+
+    asyncio.run(exercise())
+    assert [path for path, _, _ in requests] == [
+        "/admin/authz/acl/add",
+        "/admin/authz/acl/delete",
+        "/admin/authz/role_binding/add",
+        "/admin/authz/role_binding/delete",
+        "/admin/authz/acl/list",
+        "/admin/authz/role_binding/list",
+        "/admin/authz/principal",
+        "/admin/authz/group",
+    ]
+    assert [body for _, _, body in requests] == [
+        {"object": str(obj), "subject": "G:Engineering", "privilege": "SELECT"},
+        {"object": str(obj), "subject": "g:team-id", "privilege": "SELECT"},
+        {"subject": "P:Alice", "role": "reader"},
+        {"subject": "a:subject-secret", "role": "reader"},
+        {"object": str(obj), "subject": "r:reader", "max_results": 100},
+        {"subject": "g:team-id", "role": "reader", "max_results": 100},
+        {"principal": "p:alice-id"},
+        {"group": "G:Engineering"},
+    ]
+    for _, headers, _ in requests:
+        headers = {key.lower(): value for key, value in headers.items()}
+        assert "x-lancedb-database" not in headers
+        assert "x-lancedb-database-prefix" not in headers
+        assert headers["authorization"] == "Bearer rotating-token"
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+@pytest.mark.parametrize("limit", [None, 0, 1, 3])
+def test_lazy_expanded_acl_pagination(catalog_server, asynchronous, limit):
+    endpoint, requests, responses = catalog_server
+
+    def entry(privilege):
+        return {
+            "object": "database:db",
+            "subject_id": "r:reader",
+            "privilege": privilege,
+        }
+
+    responses.extend(
+        [
+            (
+                200,
+                {
+                    "entries": [entry("SELECT"), entry("INSERT")],
+                    "continuation_token": "opaque/first",
+                },
+            ),
+            (200, {"entries": [], "continuation_token": "opaque/second"}),
+            (200, {"entries": [entry("UPDATE"), entry("DELETE")]}),
+        ]
+    )
+
+    async def exercise():
+        catalog = (
+            await lancedb.connect_catalog_async(endpoint, api_key="caller-secret")
+            if asynchronous
+            else lancedb.connect_catalog(endpoint, api_key="caller-secret")
+        )
+        iterator = catalog.authz.list_acls(subject=Subject.role("reader"), limit=limit)
+        assert requests == []
+        return [entry async for entry in iterator] if asynchronous else list(iterator)
+
+    values = asyncio.run(exercise())
+    expected = ["SELECT", "INSERT", "UPDATE", "DELETE"]
+    assert [entry.privilege for entry in values] == (
+        expected if limit is None else expected[:limit]
+    )
+    assert len(requests) == (0 if limit == 0 else 1 if limit == 1 else 3)
+    for _, headers, body in requests:
+        assert headers["x-api-key"] == "caller-secret"
+        assert body["subject"] == "r:reader"
+    if len(requests) == 3:
+        assert requests[1][2]["continuation_token"] == "opaque/first"
+        assert requests[2][2]["continuation_token"] == "opaque/second"
+
+
+@pytest.mark.parametrize("status", [400, 401, 403, 404, 409, 501, 503])
+def test_http_errors_preserve_status_without_replaying_writes(catalog_server, status):
+    endpoint, requests, responses = catalog_server
+    responses.append((status, {"error": "authorization failure"}))
+    authz = lancedb.connect_catalog(endpoint).authz
+    with pytest.raises(HttpError) as error:
+        authz.add_acl(
+            object="system", subject=Subject.role("operator"), privilege="OPERATE"
+        )
+    assert error.value.status_code == status
+    assert error.value.request_id
+    assert len(requests) == 1
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+def test_role_pagination_and_late_failure(catalog_server, asynchronous):
+    endpoint, requests, responses = catalog_server
+    responses.extend(
+        [
+            (
+                200,
+                {
+                    "bindings": [{"subject_id": "g:team", "role": "reader"}],
+                    "continuation_token": "next",
+                },
+            ),
+            (403, {"error": "denied"}),
+        ]
+    )
+
+    async def exercise():
+        catalog = (
+            await lancedb.connect_catalog_async(endpoint)
+            if asynchronous
+            else lancedb.connect_catalog(endpoint)
+        )
+        iterator = catalog.authz.list_role_bindings(role="reader")
+        first = await anext(iterator) if asynchronous else next(iterator)
+        assert first.subject == Subject.group_id("team")
+        assert len(requests) == 1
+        with pytest.raises(HttpError) as error:
+            if asynchronous:
+                await anext(iterator)
+            else:
+                next(iterator)
+        assert error.value.status_code == 403
+
+    asyncio.run(exercise())
+    assert requests[1][2] == {
+        "role": "reader",
+        "max_results": 100,
+        "continuation_token": "next",
+    }
+
+
+def test_early_close_does_not_fetch_next_page(catalog_server):
+    endpoint, requests, responses = catalog_server
+    responses.append(
+        (
+            200,
+            {
+                "bindings": [{"subject_id": "g:team", "role": "reader"}],
+                "continuation_token": "next",
+            },
+        )
+    )
+    iterator = lancedb.connect_catalog(endpoint).authz.list_role_bindings()
+    next(iterator)
+    iterator.close()
+    assert len(requests) == 1
+
+
+def test_selectors_objects_and_redaction():
+    secret = Subject.principal_api_key("do-not-print-this")
+    assert "do-not-print-this" not in repr(secret)
+    assert "do-not-print-this" not in str(secret)
+    with pytest.raises(FrozenInstanceError):
+        secret.kind = "role"
+    assert str(Object.system()) == "system"
+    assert str(Object.database("tenant/db")) == "database:tenant/db"
+    assert (
+        str(Object.namespace(database="db", namespace_path=["a", "b"]))
+        == "namespace:db:a$b"
+    )
+    assert str(Object.table(database="db", name="events")) == "table:db:public$events"
+    assert str(Object.view(database="db", name="recent")) == "view:db:public$recent"
+    assert Subject.principal_id("p:id")._wire() == "p:p:id"
+    assert Subject.group_id("team-id").value == "team-id"
+    for namespace in (["a$b"], [""], "public", [".."]):
+        with pytest.raises(ValueError):
+            Object.table(database="db", namespace_path=namespace, name="events")
+    for database in ("", "a//b", "a/../b", "a:b", "a%2Fb", "a$b"):
+        with pytest.raises(ValueError):
+            Object.database(database)
+
+
+@pytest.mark.asyncio
+async def test_validation_before_io_and_empty_lists():
+    inner = AsyncMock()
+    inner.list_acls.return_value = json.dumps({"entries": []})
+    inner.list_role_bindings.return_value = json.dumps({"bindings": []})
+    authz = AsyncAuthorization(inner)
+    for limit in (-1, 1.5, True):
+        with pytest.raises(ValueError):
+            authz.list_acls(limit=limit)
+        with pytest.raises(ValueError):
+            Authorization(authz).list_role_bindings(limit=limit)
+    with pytest.raises(ValueError):
+        await authz.get_group(Subject.principal_id("alice"))
+    with pytest.raises(ValueError):
+        await authz.get_principal(Subject.group_id("team"))
+    for method in (authz.add_role_binding, authz.delete_role_binding):
+        with pytest.raises(ValueError):
+            await method(subject=Subject.role("reader"), role="writer")
+    with pytest.raises(ValueError):
+        authz.list_role_bindings(subject=Subject.role("reader"))
+    assert [entry async for entry in authz.list_acls(limit=0)] == []
+    assert inner.mock_calls == []
+    assert [entry async for entry in authz.list_acls()] == []
+    assert [entry async for entry in authz.list_role_bindings()] == []
+
+
+def test_api_key_subject_is_not_logged(catalog_server):
+    import os
+    import subprocess
+    import sys
+
+    endpoint, requests, responses = catalog_server
+    responses.append((200, {"principal_id": "public-id", "groups": [], "roles": []}))
+    result = subprocess.run(
+        [
+            sys.executable,
+            "-c",
+            "import lancedb, sys; from lancedb.authz import Subject; "
+            "lancedb.connect_catalog(sys.argv[1]).authz.get_principal("
+            "Subject.principal_api_key('test-subject-secret-never-log'))",
+            endpoint,
+        ],
+        env={**os.environ, "LANCEDB_LOG": "debug"},
+        capture_output=True,
+        text=True,
+        check=True,
+        timeout=30,
+    )
+    assert requests[0][2] == {"principal": "a:test-subject-secret-never-log"}
+    assert "test-subject-secret-never-log" not in result.stdout + result.stderr
+
+
+@pytest.mark.parametrize(
+    "builder, kwargs, wire",
+    [
+        ("system", {}, "system"),
+        ("secret", {"database": "db", "name": "api-key"}, "secret:db:public$api-key"),
+        (
+            "function",
+            {"database": "db", "name": "caption"},
+            "function:db:public$caption",
+        ),
+        (
+            "secret",
+            {"database": "tenant/db", "namespace_path": ["a", "b"], "name": "api-key"},
+            "secret:tenant/db:a$b$api-key",
+        ),
+        (
+            "function",
+            {"database": "tenant/db", "namespace_path": ["a", "b"], "name": "caption"},
+            "function:tenant/db:a$b$caption",
+        ),
+        ("database", {"name": "tenant/db"}, "database:tenant/db"),
+        (
+            "namespace",
+            {"database": "tenant/db", "namespace_path": ["a", "b"]},
+            "namespace:tenant/db:a$b",
+        ),
+        (
+            "table",
+            {"database": "tenant/db", "namespace_path": ["a", "b"], "name": "events"},
+            "table:tenant/db:a$b$events",
+        ),
+        (
+            "view",
+            {"database": "db", "namespace_path": ["public"], "name": "recent"},
+            "view:db:public$recent",
+        ),
+    ],
+)
+def test_rust_backed_object_builders(builder, kwargs, wire):
+    import pickle
+    from lancedb import _lancedb
+
+    value = getattr(Object, builder)(**kwargs)
+    assert str(value) == wire
+    assert str(value._inner) == wire
+    assert str(_lancedb.AuthzObject(wire)) == wire
+    assert value == Object(wire)
+    assert hash(value) == hash(Object(wire))
+    assert pickle.loads(pickle.dumps(value)) == value
+
+
+@pytest.mark.parametrize(
+    "kind, prefix",
+    [
+        ("principal_id", "p:"),
+        ("principal_name", "P:"),
+        ("group_id", "g:"),
+        ("group_name", "G:"),
+        ("role", "r:"),
+        ("principal_api_key", "a:"),
+    ],
+)
+def test_rust_backed_subject_selectors(kind, prefix):
+    import pickle
+    from lancedb import _lancedb
+
+    value = "p:literal@example.com / 名前"
+    subject = getattr(Subject, kind)(value)
+    assert subject.kind == kind
+    assert subject.value == value
+    assert subject._wire() == prefix + value
+    assert _lancedb.AuthzSubject(kind, value).to_wire() == prefix + value
+    assert pickle.loads(pickle.dumps(subject)) == subject
+    assert hash(subject) == hash(getattr(Subject, kind)(value))
+    with pytest.raises(ValueError):
+        getattr(Subject, kind)("")
+    if kind == "principal_api_key":
+        assert value not in repr(subject._inner)
+        assert value not in str(subject._inner)
+    if kind in ("principal_id", "group_id", "role"):
+        assert Subject._canonical(prefix + value) == subject
+    else:
+        with pytest.raises(ValueError) as error:
+            Subject._canonical(prefix + value)
+        assert value not in str(error.value)
+
+
+@pytest.mark.parametrize(
+    "wire",
+    [
+        "",
+        "database:",
+        "database:a//b",
+        "database:a/../b",
+        "database:a%2Fb",
+        "namespace:db:a$$b",
+        "table:db:events",
+        "table:db:public$",
+        "table:db:public$..",
+        "table:db:a/b$events",
+        "view:db:public$café",
+        "secret:db:public$",
+        "function:db:public$..",
+        "secret:db:a$$key",
+        "function:db:caption",
+        "secret:db:public$bad/name",
+        "function:db:public$bad:name",
+        "system:db",
+    ],
+)
+def test_invalid_raw_objects_fail_before_io(catalog_server, wire):
+    endpoint, requests, _ = catalog_server
+    authz = lancedb.connect_catalog(endpoint).authz
+    with pytest.raises(ValueError):
+        Object(wire)
+    with pytest.raises(ValueError):
+        authz.list_acls(object=wire)
+    with pytest.raises(ValueError):
+        authz.add_acl(object=wire, subject=Subject.role("reader"), privilege="SELECT")
+    assert requests == []
+
+
+@pytest.mark.parametrize(
+    "wire",
+    [
+        "future:db:resource",
+        "table_extension:db:resource",
+        "secret:tenant/db:a$b$key",
+        "function:tenant/db:a$b$caption",
+    ],
+)
+def test_objects_round_trip_through_acl_requests(catalog_server, wire):
+    endpoint, requests, responses = catalog_server
+    responses.extend(
+        [
+            (200, {}),
+            (
+                200,
+                {
+                    "entries": [
+                        {
+                            "object": wire,
+                            "subject_id": "r:reader",
+                            "privilege": "SELECT",
+                        }
+                    ]
+                },
+            ),
+            (200, {}),
+        ]
+    )
+    authz = lancedb.connect_catalog(endpoint).authz
+    obj = Object(wire)
+    assert str(obj) == wire
+    authz.add_acl(object=obj, subject=Subject.role("reader"), privilege="SELECT")
+    entries = list(authz.list_acls(object=wire))
+    assert entries[0].object == wire
+    authz.delete_acl(
+        object=entries[0].object, subject=Subject.role("reader"), privilege="SELECT"
+    )
+    assert len(requests) == 3
+    for _, _, body in requests:
+        assert body["object"] == wire
+
+
+@pytest.mark.asyncio
+async def test_native_requests_validate_selector_context_before_io(catalog_server):
+    endpoint, requests, _ = catalog_server
+    catalog = await lancedb.connect_catalog_async(endpoint)
+    native = catalog.authz._inner
+    for method, request in [
+        (native.list_role_bindings, {"subject": "r:reader"}),
+        (native.add_role_binding, {"subject": "r:reader", "role": "writer"}),
+        (native.delete_role_binding, {"subject": "r:reader", "role": "writer"}),
+        (native.get_principal, {"principal": "g:group"}),
+        (native.get_group, {"group": "a:never-log-this-key"}),
+    ]:
+        with pytest.raises(ValueError) as error:
+            await method(json.dumps(request))
+        assert "never-log-this-key" not in str(error.value)
+    assert requests == []
+
+
+def test_privilege_values_use_rust_serialization():
+    import pickle
+    from lancedb import _lancedb
+
+    names = [value.value for value in Privilege]
+    assert names == _lancedb.AuthzPrivilege.known()
+    for privilege in Privilege:
+        assert json.loads(json.dumps(privilege)) == privilege.value
+        assert str(_lancedb.AuthzPrivilege(privilege)) == privilege.value
+        assert pickle.loads(pickle.dumps(privilege)) is privilege
+    for index in range(64):
+        alias = f"PRIVILEGE{index}"
+        expected = names[index] if index < len(names) else alias
+        assert str(_lancedb.AuthzPrivilege(alias)) == expected
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+@pytest.mark.parametrize(
+    "creation_privilege", [None, "CREATE_SECRET", "CREATE_FUNCTION"]
+)
+def test_privilege_names_and_future_values_round_trip(
+    catalog_server, asynchronous, creation_privilege
+):
+    endpoint, requests, responses = catalog_server
+    inputs = [
+        Privilege.SELECT,
+        "SELECT",
+        "PRIVILEGE2",
+        "PRIVILEGE20",
+        "PRIVILEGE63",
+        "PRIVILEGE64",
+        "PRIVILEGE256",
+        "PRIVILEGE18446744073709551616",
+    ]
+    expected = ["SELECT", "SELECT", "SELECT", *inputs[3:]]
+    object_id = "table:db:public$t"
+    if creation_privilege is not None:
+        inputs = [Privilege[creation_privilege], creation_privilege]
+        expected = [creation_privilege, creation_privilege]
+        object_id = "namespace:db:public"
+    for privilege in inputs:
+        responses.extend(
+            [
+                (200, {}),
+                (
+                    200,
+                    {
+                        "entries": [
+                            {
+                                "object": object_id,
+                                "subject_id": "r:reader",
+                                "privilege": privilege,
+                            }
+                        ]
+                    },
+                ),
+                (200, {}),
+            ]
+        )
+
+    async def exercise():
+        catalog = (
+            await lancedb.connect_catalog_async(endpoint)
+            if asynchronous
+            else lancedb.connect_catalog(endpoint)
+        )
+        authz = catalog.authz
+        for privilege, canonical in zip(inputs, expected):
+            added = authz.add_acl(
+                object=object_id,
+                subject=Subject.role("reader"),
+                privilege=privilege,
+            )
+            if asynchronous:
+                await added
+            iterator = authz.list_acls()
+            entries = (
+                [entry async for entry in iterator] if asynchronous else list(iterator)
+            )
+            assert entries[0].privilege == canonical
+            deleted = authz.delete_acl(
+                object=entries[0].object,
+                subject=entries[0].subject,
+                privilege=entries[0].privilege,
+            )
+            if asynchronous:
+                await deleted
+
+    asyncio.run(exercise())
+    assert len(requests) == 3 * len(inputs)
+    for index, canonical in enumerate(expected):
+        assert requests[index * 3][2]["object"] == object_id
+        assert requests[index * 3][2]["privilege"] == canonical
+        assert requests[index * 3 + 2][2]["object"] == object_id
+        assert requests[index * 3 + 2][2]["privilege"] == canonical
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+def test_invalid_privileges_fail_before_io(catalog_server, asynchronous):
+    endpoint, requests, _ = catalog_server
+    invalid = [
+        "",
+        "select",
+        "SELECT ",
+        "UNKNOWN",
+        "PRIVILEGE",
+        "PRIVILEGE-1",
+        "PRIVILEGE+1",
+        "PRIVILEGE1x",
+        None,
+        2,
+        True,
+        ["SELECT"],
+    ]
+
+    async def exercise():
+        catalog = (
+            await lancedb.connect_catalog_async(endpoint)
+            if asynchronous
+            else lancedb.connect_catalog(endpoint)
+        )
+        async_authz = catalog.authz if asynchronous else catalog.authz._inner
+        native = async_authz._inner
+        for privilege in invalid:
+            for method in [catalog.authz.add_acl, catalog.authz.delete_acl]:
+                with pytest.raises(ValueError, match="privilege"):
+                    result = method(
+                        object="system",
+                        subject=Subject.role("reader"),
+                        privilege=privilege,
+                    )
+                    if asynchronous:
+                        await result
+            for method in [
+                native.add_acl,
+                native.delete_acl,
+            ]:
+                with pytest.raises(ValueError):
+                    await method(
+                        json.dumps(
+                            {
+                                "object": "system",
+                                "subject": "r:reader",
+                                "privilege": privilege,
+                            }
+                        )
+                    )
+
+    asyncio.run(exercise())
+    assert requests == []
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+def test_acl_page_preserves_unknown_subjects(catalog_server, asynchronous):
+    import pickle
+
+    endpoint, requests, responses = catalog_server
+    responses.extend(
+        [
+            (
+                200,
+                {
+                    "entries": [
+                        {
+                            "subject_id": "p:known",
+                            "object": "system",
+                            "privilege": "USAGE",
+                        },
+                        {
+                            "subject_id": "x:future-subject",
+                            "object": "system",
+                            "privilege": "USAGE",
+                        },
+                    ]
+                },
+            ),
+            (200, {}),
+        ]
+    )
+
+    async def exercise():
+        catalog = (
+            await lancedb.connect_catalog_async(endpoint)
+            if asynchronous
+            else lancedb.connect_catalog(endpoint)
+        )
+        iterator = catalog.authz.list_acls()
+        entries = (
+            [entry async for entry in iterator] if asynchronous else list(iterator)
+        )
+        assert len(entries) == 2
+        assert entries[0].subject == Subject.principal_id("known")
+        subject = entries[1].subject
+        assert subject.kind == "unknown"
+        assert subject.value == "x:future-subject"
+        assert subject._wire() == "x:future-subject"
+        assert pickle.loads(pickle.dumps(subject)) == subject
+        result = catalog.authz.delete_acl(
+            object=entries[1].object, subject=subject, privilege=entries[1].privilege
+        )
+        if asynchronous:
+            await result
+
+    asyncio.run(exercise())
+    assert len(requests) == 2
+    assert requests[1][2]["subject"] == "x:future-subject"
+
+
+@pytest.mark.parametrize("builder", ["secret", "function"])
+def test_secret_and_function_builders_validate_components(builder):
+    from lancedb import _lancedb
+
+    for database, namespace, name in [
+        ("a//b", ["public"], "valid"),
+        ("db", ["a$b"], "valid"),
+        ("db", "public", "valid"),
+        ("db", ["public"], "bad$name"),
+        ("db", ["public"], ".."),
+    ]:
+        with pytest.raises(ValueError):
+            getattr(Object, builder)(
+                database=database, namespace_path=namespace, name=name
+            )
+        with pytest.raises(ValueError):
+            getattr(_lancedb.AuthzObject, builder)(database, namespace, name)
+
+
+@pytest.mark.parametrize(
+    "builder", ["namespace", "table", "view", "secret", "function"]
+)
+def test_empty_and_omitted_namespace_paths_select_public(builder):
+    from lancedb import _lancedb
+
+    kwargs = {"database": "db"}
+    native_args = ()
+    wire = "namespace:db:public"
+    if builder != "namespace":
+        kwargs["name"] = "item"
+        native_args = ("item",)
+        wire = f"{builder}:db:public$item"
+
+    build = getattr(Object, builder)
+    expected = build(**kwargs, namespace_path=["public"])
+    assert str(expected) == wire
+    assert build(**kwargs) == expected
+    for namespace in [[], ()]:
+        assert build(**kwargs, namespace_path=namespace) == expected
+        assert (
+            str(getattr(_lancedb.AuthzObject, builder)("db", namespace, *native_args))
+            == wire
+        )

python/python/tests/test_catalog.py

@@ -263,3 +263,45 @@ async def test_database_names_invalid_limit(catalog_server, asynchronous, page_l
             next(names)
     assert names.num_page_results() == 0
     assert requests == []
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("asynchronous", [False, True])
+async def test_catalog_rejects_invalid_database_names_before_io(
+    catalog_server, asynchronous
+):
+    endpoint, requests, _ = catalog_server
+    catalog = (
+        await lancedb.connect_catalog_async(endpoint)
+        if asynchronous
+        else lancedb.connect_catalog(endpoint)
+    )
+    for name in [
+        "",
+        ".",
+        "..",
+        "/db",
+        "db/",
+        "a//b",
+        "a/./b",
+        "a/../b",
+        "a b",
+        "a:b",
+        "a$b",
+        "a%b",
+        "a?b",
+        "a#b",
+        "a\\b",
+        "café",
+        "a\nb",
+    ]:
+        for method in [
+            catalog.create_database,
+            catalog.connect_database,
+            catalog.drop_database,
+        ]:
+            with pytest.raises(ValueError, match="Invalid database name"):
+                result = method(name)
+                if asynchronous:
+                    await result
+    assert requests == []

python/src/authz.rs

@@ -0,0 +1,280 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+use crate::error::PythonErrorExt;
+use crate::runtime::future_into_py;
+use lancedb::authz::*;
+use pyo3::exceptions::{PyRuntimeError, PyValueError};
+use pyo3::types::{PyAnyMethods, PyBytes, PyString};
+use pyo3::{Bound, PyAny, PyRef, PyResult, pyclass, pymethods};
+use std::sync::Arc;
+
+#[pyclass]
+pub struct AuthorizationClient {
+    pub(crate) inner: Arc<dyn Authorization>,
+}
+
+#[pymethods]
+impl AuthorizationClient {
+    fn list_acls<'py>(self_: PyRef<'py, Self>, request: &str) -> PyResult<Bound<'py, PyAny>> {
+        let request: ListAccessControlEntriesRequest = serde_json::from_str(request)
+            .map_err(|_| PyValueError::new_err("Invalid authorization request"))?;
+        let inner = self_.inner.clone();
+        future_into_py(self_.py(), async move {
+            let result = inner.list_acls(request).await.infer_error()?;
+            serde_json::to_string(&result).map_err(|err| PyRuntimeError::new_err(err.to_string()))
+        })
+    }
+    fn add_acl<'py>(self_: PyRef<'py, Self>, request: &str) -> PyResult<Bound<'py, PyAny>> {
+        let request: AccessControlEntryRequest = serde_json::from_str(request)
+            .map_err(|_| PyValueError::new_err("Invalid authorization request"))?;
+        let inner = self_.inner.clone();
+        future_into_py(self_.py(), async move {
+            inner.add_acl(request).await.infer_error()
+        })
+    }
+    fn delete_acl<'py>(self_: PyRef<'py, Self>, request: &str) -> PyResult<Bound<'py, PyAny>> {
+        let request: AccessControlEntryRequest = serde_json::from_str(request)
+            .map_err(|_| PyValueError::new_err("Invalid authorization request"))?;
+        let inner = self_.inner.clone();
+        future_into_py(self_.py(), async move {
+            inner.delete_acl(request).await.infer_error()
+        })
+    }
+    fn list_role_bindings<'py>(
+        self_: PyRef<'py, Self>,
+        request: &str,
+    ) -> PyResult<Bound<'py, PyAny>> {
+        let request: ListRoleBindingsRequest = serde_json::from_str(request)
+            .map_err(|_| PyValueError::new_err("Invalid authorization request"))?;
+        request.validate().infer_error()?;
+        let inner = self_.inner.clone();
+        future_into_py(self_.py(), async move {
+            let result = inner.list_role_bindings(request).await.infer_error()?;
+            serde_json::to_string(&result).map_err(|err| PyRuntimeError::new_err(err.to_string()))
+        })
+    }
+    fn add_role_binding<'py>(
+        self_: PyRef<'py, Self>,
+        request: &str,
+    ) -> PyResult<Bound<'py, PyAny>> {
+        let request: RoleBindingRequest = serde_json::from_str(request)
+            .map_err(|_| PyValueError::new_err("Invalid authorization request"))?;
+        request.validate().infer_error()?;
+        let inner = self_.inner.clone();
+        future_into_py(self_.py(), async move {
+            inner.add_role_binding(request).await.infer_error()
+        })
+    }
+    fn delete_role_binding<'py>(
+        self_: PyRef<'py, Self>,
+        request: &str,
+    ) -> PyResult<Bound<'py, PyAny>> {
+        let request: RoleBindingRequest = serde_json::from_str(request)
+            .map_err(|_| PyValueError::new_err("Invalid authorization request"))?;
+        request.validate().infer_error()?;
+        let inner = self_.inner.clone();
+        future_into_py(self_.py(), async move {
+            inner.delete_role_binding(request).await.infer_error()
+        })
+    }
+    fn get_principal<'py>(self_: PyRef<'py, Self>, request: &str) -> PyResult<Bound<'py, PyAny>> {
+        let request: GetPrincipalRequest = serde_json::from_str(request)
+            .map_err(|_| PyValueError::new_err("Invalid authorization request"))?;
+        request.validate().infer_error()?;
+        let inner = self_.inner.clone();
+        future_into_py(self_.py(), async move {
+            let result = inner.get_principal(request).await.infer_error()?;
+            serde_json::to_string(&result).map_err(|err| PyRuntimeError::new_err(err.to_string()))
+        })
+    }
+    fn get_group<'py>(self_: PyRef<'py, Self>, request: &str) -> PyResult<Bound<'py, PyAny>> {
+        let request: GetGroupRequest = serde_json::from_str(request)
+            .map_err(|_| PyValueError::new_err("Invalid authorization request"))?;
+        request.validate().infer_error()?;
+        let inner = self_.inner.clone();
+        future_into_py(self_.py(), async move {
+            let result = inner.get_group(request).await.infer_error()?;
+            serde_json::to_string(&result).map_err(|err| PyRuntimeError::new_err(err.to_string()))
+        })
+    }
+}
+
+/// The shared privilege parser and canonical wire names.
+#[pyclass(frozen)]
+pub struct AuthzPrivilege {
+    inner: Privilege,
+}
+
+#[pymethods]
+impl AuthzPrivilege {
+    #[new]
+    fn new(value: &str) -> PyResult<Self> {
+        Ok(Self {
+            inner: value.parse().map_err(PyValueError::new_err)?,
+        })
+    }
+
+    #[staticmethod]
+    fn known() -> Vec<String> {
+        Privilege::KNOWN.iter().map(ToString::to_string).collect()
+    }
+
+    fn __str__(&self) -> String {
+        self.inner.to_string()
+    }
+}
+
+#[pyclass(frozen)]
+pub struct AuthzSubject {
+    inner: Subject,
+}
+
+#[pymethods]
+impl AuthzSubject {
+    #[new]
+    fn new(kind: &str, value: &Bound<'_, PyAny>) -> PyResult<Self> {
+        let value: String = value
+            .extract()
+            .map_err(|_| PyValueError::new_err("Subject value must be a nonempty string"))?;
+        Ok(Self {
+            inner: Subject::new(kind.parse().infer_error()?, value).infer_error()?,
+        })
+    }
+
+    #[staticmethod]
+    fn from_canonical(value: &str) -> PyResult<Self> {
+        Ok(Self {
+            inner: Subject::from_canonical(value).infer_error()?,
+        })
+    }
+
+    #[getter]
+    fn kind(&self) -> &str {
+        self.inner.kind().as_str()
+    }
+    #[getter]
+    fn value(&self) -> &str {
+        self.inner.value()
+    }
+    #[getter]
+    fn display_value(&self) -> &str {
+        self.inner.display_value()
+    }
+
+    #[pyo3(signature = (context=None))]
+    fn to_wire(&self, context: Option<&str>) -> PyResult<String> {
+        match context {
+            None => {}
+            Some("principal") => self.inner.validate_principal().infer_error()?,
+            Some("group") => self.inner.validate_group().infer_error()?,
+            Some("role_member") => self.inner.validate_role_member().infer_error()?,
+            Some(_) => return Err(PyValueError::new_err("Invalid subject context")),
+        }
+        Ok(self.inner.to_wire())
+    }
+
+    fn __repr__(&self) -> String {
+        format!("{:?}", self.inner)
+    }
+    fn __str__(&self) -> String {
+        self.inner.to_string()
+    }
+}
+
+#[pyclass(frozen)]
+pub struct AuthzObject {
+    inner: Object,
+}
+
+impl AuthzObject {
+    fn validated(inner: Object) -> PyResult<Self> {
+        inner.validate().map_err(PyValueError::new_err)?;
+        Ok(Self { inner })
+    }
+}
+
+fn namespace_path(value: &Bound<'_, PyAny>) -> PyResult<NamespacePath> {
+    if value.is_instance_of::<PyString>() || value.is_instance_of::<PyBytes>() {
+        return Err(PyValueError::new_err(
+            "namespace must be a sequence of components",
+        ));
+    }
+    let components = value
+        .extract()
+        .map_err(|_| PyValueError::new_err("namespace must be a sequence of string components"))?;
+    NamespacePath::new(components).map_err(PyValueError::new_err)
+}
+
+#[pymethods]
+impl AuthzObject {
+    #[new]
+    fn new(value: &str) -> PyResult<Self> {
+        Ok(Self {
+            inner: value.parse().infer_error()?,
+        })
+    }
+
+    #[staticmethod]
+    fn system() -> Self {
+        Self {
+            inner: Object::System,
+        }
+    }
+
+    #[staticmethod]
+    fn database(name: &str) -> PyResult<Self> {
+        Self::validated(Object::Database(DatabaseObject::new(name.into())))
+    }
+
+    #[staticmethod]
+    fn namespace(database: &str, namespace: &Bound<'_, PyAny>) -> PyResult<Self> {
+        Self::validated(Object::Namespace(NamespaceObject {
+            database: database.into(),
+            namespace: namespace_path(namespace)?,
+        }))
+    }
+
+    #[staticmethod]
+    fn table(database: &str, namespace: &Bound<'_, PyAny>, name: &str) -> PyResult<Self> {
+        Self::validated(Object::Table(TableObject {
+            database: database.into(),
+            namespace: namespace_path(namespace)?,
+            table: name.into(),
+        }))
+    }
+
+    #[staticmethod]
+    fn view(database: &str, namespace: &Bound<'_, PyAny>, name: &str) -> PyResult<Self> {
+        Self::validated(Object::View(ViewObject {
+            database: database.into(),
+            namespace: namespace_path(namespace)?,
+            view: name.into(),
+        }))
+    }
+
+    #[staticmethod]
+    fn secret(database: &str, namespace: &Bound<'_, PyAny>, name: &str) -> PyResult<Self> {
+        Self::validated(Object::Secret(SecretObject {
+            database: database.into(),
+            namespace: namespace_path(namespace)?,
+            secret: name.into(),
+        }))
+    }
+
+    #[staticmethod]
+    fn function(database: &str, namespace: &Bound<'_, PyAny>, name: &str) -> PyResult<Self> {
+        Self::validated(Object::Function(FunctionObject {
+            database: database.into(),
+            namespace: namespace_path(namespace)?,
+            function: name.into(),
+        }))
+    }
+
+    fn __str__(&self) -> String {
+        self.inner.to_string()
+    }
+    fn __repr__(&self) -> String {
+        format!("{:?}", self.inner)
+    }
+}

python/src/catalog.rs

@@ -21,6 +21,13 @@ pub struct Catalog {
 
 #[pymethods]
 impl Catalog {
+    #[getter]
+    fn authz(&self) -> PyResult<crate::authz::AuthorizationClient> {
+        Ok(crate::authz::AuthorizationClient {
+            inner: self.inner.authz().infer_error()?,
+        })
+    }
+
     #[getter]
     fn uri(&self) -> &str {
         self.inner.uri()

python/src/lib.rs

@@ -22,6 +22,7 @@ use table::{
 };
 
 pub mod arrow;
+pub mod authz;
 pub mod catalog;
 pub mod connection;
 pub mod error;
@@ -70,6 +71,10 @@ pub fn _lancedb(py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<listing::FunctionListing>()?;
     m.add_class::<listing::JobListing>()?;
 
+    m.add_class::<authz::AuthorizationClient>()?;
+    m.add_class::<authz::AuthzSubject>()?;
+    m.add_class::<authz::AuthzObject>()?;
+    m.add_class::<authz::AuthzPrivilege>()?;
     m.add_function(wrap_pyfunction!(catalog::connect_catalog, m)?)?;
     m.add_class::<Session>()?;
     m.add_class::<Table>()?;