authz

python/python/lancedb/_lancedb.pyi

@@ -1,6 +1,6 @@
 from datetime import date, datetime, timedelta
 from decimal import Decimal
-from typing import Dict, List, Optional, Tuple, Any, TypedDict, Union, Literal
+from typing import Dict, List, Optional, Tuple, Any, TypedDict, Union, Literal, Sequence
 from uuid import UUID
 
 import pyarrow as pa
@@ -951,6 +951,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(
@@ -997,3 +999,52 @@ 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,585 @@
+# 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.
+    """
+
+    _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]) -> "Object":
+        """A namespace path in an explicitly selected database."""
+        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,792 @@
+# 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=["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=["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=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": ["a", "b"], "name": "api-key"},
+            "secret:tenant/db:a$b$api-key",
+        ),
+        (
+            "function",
+            {"database": "tenant/db", "namespace": ["a", "b"], "name": "caption"},
+            "function:tenant/db:a$b$caption",
+        ),
+        ("database", {"name": "tenant/db"}, "database:tenant/db"),
+        (
+            "namespace",
+            {"database": "tenant/db", "namespace": ["a", "b"]},
+            "namespace:tenant/db:a$b",
+        ),
+        (
+            "table",
+            {"database": "tenant/db", "namespace": ["a", "b"], "name": "events"},
+            "table:tenant/db:a$b$events",
+        ),
+        (
+            "view",
+            {"database": "db", "namespace": ["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}"
+        assert str(_lancedb.AuthzPrivilege(alias)) == alias
+
+
+@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", *inputs[1:]]
+    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", [], "valid"),
+        ("db", ["a$b"], "valid"),
+        ("db", "public", "valid"),
+        ("db", ["public"], "bad$name"),
+        ("db", ["public"], ".."),
+    ]:
+        with pytest.raises(ValueError):
+            getattr(Object, builder)(database=database, namespace=namespace, name=name)
+        with pytest.raises(ValueError):
+            getattr(_lancedb.AuthzObject, builder)(database, namespace, name)

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>()?;