diff --git a/Cargo.lock b/Cargo.lock index eecf3d0d1..10f1a43c0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5845,6 +5845,7 @@ dependencies = [ "serial_test", "sha2 0.10.9", "snafu 0.8.9", + "strum_macros 0.28.0", "tempfile", "test-log", "tokenizers", diff --git a/docs/src/python/python.md b/docs/src/python/python.md index 3776283dc..01e3e69f5 100644 --- a/docs/src/python/python.md +++ b/docs/src/python/python.md @@ -48,6 +48,168 @@ are ordinary connections. Dropping a database requires it to be empty. ::: lancedb.catalog.DatabaseNames +## Authorization + +Some catalogs support the authorization API. This is an API that can be used to +bind access control lists (ACLs) to individual catalog objects. + +Each ACL contains a subject, an object, and privilege. + +### Subjects + +There are three kinds of subjects: principals, groups, and roles. + +A principal uniquely identifies each user accessing the system. Each LanceDB +Enterprise API key corresponds to a unique principal, and each OpenID user has +a unique principal. + +All principals have an immutable ID which never changes, and is safe to include +in log files. For example, the API key +`sk_AAAQEAYEAUDAOCAJBIFQYDIOB4IBCEQTCQKRMFYYDENBWHA5DYPQ====` maps to the +principal ID `pk_aaaqeayeaudaocajbifqydiob4`. APIs such as `list_acls` will +return the principal ID, never the API key. + +OpenID principals also typically have a human-readable name associated with +them. For example, a principal might be named `bob@example.com` and might have +a principal ID of `8a004400-9443-4fab-9997-b9372a7ca8fdCopy`. Whenever +possible, APIs will return the principal name for convenience. However, +principal names are mutable and can be changed by the identity provider. + +Here is an example of getting information about an API key: +```python +from lancedb.authz import Subject +catalog = lancedb.connect_catalog("https://catalog.example.com", api_key=api_key) +principal = catalog.authz.get_principal(Subject.principal_api_key(api_key)) +print(f"Found principal {principal.principal_id}") +``` + +OpenID principals can be members of groups. Like principals, groups also have +IDs and human-readable names. It is not possible to modify group membership via +this API; it is managed by the OpenID identity provider. + +Here is an example of getting information about a group based on its name: +```python +from lancedb.authz import Subject +catalog = lancedb.connect_catalog("https://catalog.example.com", api_key=api_key) +group = catalog.authz.get_group(Subject.group_name("Engineering")) +print(f"Found group {group.group_id} with name {group.group_name}") +``` + +Roles can contain either principals or groups. Unlike groups, they are managed +by the catalog, not the identity provider. Roles can also contain API key +principals. + +Here is an example of managing the `reader` role: +```python +from lancedb.authz import Subject +catalog = lancedb.connect_catalog("https://catalog.example.com", api_key=api_key) +catalog.authz.add_role_binding( + subject=Subject.principal_name("Alice"), + role="reader" +) +``` +It may take a few milliseconds for changes to role bindings to become visible +to all clients. + +More subject types may be added in the future. + +### Objects + +An object is any resource in the system that we can attach authorization rules to. +The Python builders and Rust client share resource validation and wire encoding. +Methods also accept literal strings such as `table:tenant/db:public$events`; +malformed names for known resource types raise `ValueError` locally before a +request is sent. Unknown resource types are preserved for compatibility with +newer servers. + +Object types include: + +The System object, a singleton which represents the entire system. You can +refer to it as `Object.system()`. + +Database objects which represents specific databases. For example, +`Object.database("analytics")` represents the "analytics" database. + +Table objects which represent specific tables. For example, +`Object.table(database="analytics", name="events")` represents the "events" +table inside the "analytics" database. Tables are in the default namespace (which +is named "public") unless a specific namespaces is given. + +Similarly, View objects represent specific views. For example, +`Object.view(database="analytics", name="recent_events")` represents the +"recent_events" view inside the "analytics" table. Views are in the default +namespace (which is named "public") unless a specific namespaces is given. + +Secret objects identify named credentials, for example, +`Object.secret(database="analytics", name="api-key")`. Function objects identify +registered functions, for example, +`Object.function(database="analytics", name="caption")`. Both default to the +`public` namespace. Pass `namespace=["platform", "ml"]` to select a nested +namespace. These objects identify resources for ACL operations; a secret object +contains the credential's name, not its value. + +More object types may be added in the future. + +### Access control lists + +An access control list attaches a specific set of privileges to a specific +(subject, object) pair. So, for example, we might have an access control list +that gives `bob@example.com` the ability to read the foo table, that looks like +this: + +```python +from lancedb.authz import Object, Subject, Privilege + +catalog = lancedb.connect_catalog("https://catalog.example.com", api_key=api_key) +catalog.authz.add_acl( + object=Object.table(database="analytics", name="foo"), + subject=Subject.principal_name("bob@example.com"), + privilege=Privilege.SELECT, +) +``` + +Reading a table generally requires USAGE on its database and namespace, plus +SELECT on the table. Each object has at most one owner. + +OWNERSHIP is a special privilege which grants all the other privileges. By +default, the principal who created an object owns it. Only one subject can own +an object; granting OWNERSHIP to a new subject removes it from the previous +owner. OWNERSHIP cannot be deleted. + +To delegate secret or function creation, grant `CREATE_SECRET` or +`CREATE_FUNCTION` on the containing namespace. This preserves namespace ownership: + +```python +for privilege in [Privilege.CREATE_SECRET, Privilege.CREATE_FUNCTION]: + catalog.authz.add_acl( + object=Object.namespace(database="analytics", namespace=["public"]), + subject=Subject.role("developer"), + privilege=privilege, + ) +``` + +Both privileges also accept their string names in synchronous and asynchronous +ACL methods. + +By default, the `list_acls` function lists all the access control entries in +the system. By adding a function parameter, the output is filtered to just ACLs +matching that parameter. For example, this code lists all the ACLs for the +"reader" role: + +```python +from lancedb.authz import Subject +catalog = lancedb.connect_catalog("https://catalog.example.com", api_key=api_key) +for entry in catalog.authz.list_acls(subject=Subject.role("reader")): + print(entry) +``` + +Or, when using the async API: +```python +catalog = await lancedb.connect_catalog_async(endpoint, api_key=api_key) +async for entry in catalog.authz.list_acls(subject=Subject.role("reader")): + print(entry) +``` + ## Remote SQL Submit SQL against a remote LanceDB database through the connection. @@ -359,6 +521,8 @@ still work. Queries return descriptors. Call ## Exceptions +::: lancedb.remote.errors.HttpError + ::: lancedb.exceptions.MissingValueError ::: lancedb.exceptions.MissingColumnError diff --git a/nodejs/__test__/catalog.test.ts b/nodejs/__test__/catalog.test.ts index 945956b5b..9128d6083 100644 --- a/nodejs/__test__/catalog.test.ts +++ b/nodejs/__test__/catalog.test.ts @@ -266,8 +266,41 @@ describe("remote catalog", () => { "limit", ); } - await expect(catalog.connectDatabase("a$b")).rejects.toThrow( - "Invalid database name", - ); }); + + it.each([ + "", + ".", + "..", + "/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", + ])( + "rejects invalid database name %j before sending requests", + async (name) => { + await withCatalog([], async (catalog, requests) => { + for (const method of [ + "createDatabase", + "connectDatabase", + "dropDatabase", + ] as const) { + await expect(catalog[method](name)).rejects.toThrow( + "invalid database name", + ); + } + expect(requests).toEqual([]); + }); + }, + ); }); diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index a8f02e458..e8031131e 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/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: ... diff --git a/python/python/lancedb/authz.py b/python/python/lancedb/authz.py new file mode 100644 index 000000000..b7e21af27 --- /dev/null +++ b/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)) diff --git a/python/python/lancedb/catalog.py b/python/python/lancedb/catalog.py index 70aab1bbb..666d414a4 100644 --- a/python/python/lancedb/catalog.py +++ b/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.""" diff --git a/python/python/tests/test_authz.py b/python/python/tests/test_authz.py new file mode 100644 index 000000000..4f1ecc35b --- /dev/null +++ b/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) diff --git a/python/python/tests/test_catalog.py b/python/python/tests/test_catalog.py index e7b7b1ec2..b2036b7a6 100644 --- a/python/python/tests/test_catalog.py +++ b/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 == [] diff --git a/python/src/authz.rs b/python/src/authz.rs new file mode 100644 index 000000000..6ad3e69b1 --- /dev/null +++ b/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, +} + +#[pymethods] +impl AuthorizationClient { + fn list_acls<'py>(self_: PyRef<'py, Self>, request: &str) -> PyResult> { + 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> { + 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> { + 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> { + 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> { + 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> { + 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> { + 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> { + 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 { + Ok(Self { + inner: value.parse().map_err(PyValueError::new_err)?, + }) + } + + #[staticmethod] + fn known() -> Vec { + 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 { + 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 { + 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 { + 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 { + inner.validate().map_err(PyValueError::new_err)?; + Ok(Self { inner }) + } +} + +fn namespace_path(value: &Bound<'_, PyAny>) -> PyResult { + if value.is_instance_of::() || value.is_instance_of::() { + 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 { + Ok(Self { + inner: value.parse().infer_error()?, + }) + } + + #[staticmethod] + fn system() -> Self { + Self { + inner: Object::System, + } + } + + #[staticmethod] + fn database(name: &str) -> PyResult { + Self::validated(Object::Database(DatabaseObject::new(name.into()))) + } + + #[staticmethod] + fn namespace(database: &str, namespace: &Bound<'_, PyAny>) -> PyResult { + Self::validated(Object::Namespace(NamespaceObject { + database: database.into(), + namespace: namespace_path(namespace)?, + })) + } + + #[staticmethod] + fn table(database: &str, namespace: &Bound<'_, PyAny>, name: &str) -> PyResult { + 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::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::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::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) + } +} diff --git a/python/src/catalog.rs b/python/src/catalog.rs index e7cc56c49..9e375b314 100644 --- a/python/src/catalog.rs +++ b/python/src/catalog.rs @@ -21,6 +21,13 @@ pub struct Catalog { #[pymethods] impl Catalog { + #[getter] + fn authz(&self) -> PyResult { + Ok(crate::authz::AuthorizationClient { + inner: self.inner.authz().infer_error()?, + }) + } + #[getter] fn uri(&self) -> &str { self.inner.uri() diff --git a/python/src/lib.rs b/python/src/lib.rs index 9f88030ca..066e85927 100644 --- a/python/src/lib.rs +++ b/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::()?; m.add_class::()?; + m.add_class::()?; + m.add_class::()?; + m.add_class::()?; + m.add_class::()?; m.add_function(wrap_pyfunction!(catalog::connect_catalog, m)?)?; m.add_class::()?; m.add_class::()?; diff --git a/rust/lancedb/Cargo.toml b/rust/lancedb/Cargo.toml index b15c24ca5..b695d7202 100644 --- a/rust/lancedb/Cargo.toml +++ b/rust/lancedb/Cargo.toml @@ -66,6 +66,7 @@ serde = { workspace = true } serde_json = { workspace = true } async-openai = { version = "0.20.0", optional = true } serde_with = { version = "3.8.1" } +strum_macros = "0.28" tempfile = { workspace = true } aws-smithy-types = { workspace = true, optional = true } aws-sdk-bedrockruntime = { version = "1.27.0", optional = true } diff --git a/rust/lancedb/src/authz/api.rs b/rust/lancedb/src/authz/api.rs new file mode 100644 index 000000000..38e7d381b --- /dev/null +++ b/rust/lancedb/src/authz/api.rs @@ -0,0 +1,239 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +//! Authorization requests, responses, and the deployment-wide API. + +use super::{Object, Privilege, Subject}; +use crate::Result; +use serde::{Deserialize, Serialize}; + +/// Filters and pagination for a page of ACL grants. Omitted filters match all. +#[derive(Clone, Default, Serialize, Deserialize)] +#[non_exhaustive] +pub struct ListAccessControlEntriesRequest { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub object: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub subject: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub continuation_token: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + /// Maximum stored subject/object pairs, not expanded privilege entries. + /// The server caps this at 100; None uses its default. Must be positive. + pub max_results: Option, +} + +/// One privilege granted to a canonical subject on an object. +#[derive(Clone, Serialize, Deserialize, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub struct AccessControlEntry { + #[serde(deserialize_with = "Subject::deserialize_canonical")] + pub subject_id: Subject, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub subject_name: Option, + pub object: Object, + pub privilege: Privilege, +} + +/// One page of grants. A stored subject/object pair may expand to several entries. +#[derive(Clone, Default, Serialize, Deserialize, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub struct ListAccessControlEntriesResponse { + pub entries: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub continuation_token: Option, +} + +/// One grant to add or delete. The subject may be selected by ID, name, or API key. +#[derive(Clone, Serialize, Deserialize)] +#[non_exhaustive] +pub struct AccessControlEntryRequest { + pub subject: Subject, + pub object: Object, + /// A privilege such as [`Privilege::Select`] or [`Privilege::Usage`]. + /// [`Privilege::Unknown`] preserves future privilege names. + pub privilege: Privilege, +} + +/// Filters and pagination for direct principal/group role bindings. +#[derive(Clone, Default, Serialize, Deserialize)] +#[non_exhaustive] +pub struct ListRoleBindingsRequest { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub subject: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub role: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub continuation_token: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + /// Maximum bindings on this page; the server caps this at 100. + pub max_results: Option, +} + +/// A direct principal or group membership in a LanceDB-managed role. +#[derive(Clone, Serialize, Deserialize, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub struct RoleBinding { + #[serde(deserialize_with = "Subject::deserialize_canonical")] + pub subject_id: Subject, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub subject_name: Option, + pub role: String, +} + +/// One page of direct bindings, without expanding group membership. +#[derive(Clone, Default, Serialize, Deserialize, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub struct ListRoleBindingsResponse { + pub bindings: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub continuation_token: Option, +} + +/// A role to grant to or revoke from a principal or group. Roles cannot nest. +#[derive(Clone, Serialize, Deserialize)] +#[non_exhaustive] +pub struct RoleBindingRequest { + pub subject: Subject, + pub role: String, +} + +/// Select a principal by canonical ID, provider name, or API key. +#[derive(Clone, Serialize, Deserialize)] +#[non_exhaustive] +pub struct GetPrincipalRequest { + pub principal: Subject, +} + +/// Select a group by canonical ID or provider name. +#[derive(Clone, Serialize, Deserialize)] +#[non_exhaustive] +pub struct GetGroupRequest { + pub group: Subject, +} + +/// A canonical identity-provider ID and optional display name. +#[derive(Clone, Default, Serialize, Deserialize, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub struct IdentityInfo { + pub id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub name: Option, +} + +/// A principal, its provider-managed groups, and roles held directly or via groups. +#[derive(Clone, Default, Serialize, Deserialize, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub struct PrincipalInfo { + pub principal_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub principal_name: Option, + pub groups: Vec, + pub roles: Vec, +} + +/// A group, its provider-managed members, and its directly granted roles. +#[derive(Clone, Default, Serialize, Deserialize, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub struct GroupInfo { + pub group_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub group_name: Option, + pub principals: Vec, + pub roles: Vec, +} + +/// Authorization operations on one deployment, independent of database selection. +/// +/// ACL writes preserve server conflict/not-found errors and must not be replayed +/// after ambiguous transport failures. Role-binding writes are idempotent. +#[async_trait::async_trait] +pub trait Authorization: Send + Sync + std::fmt::Debug + 'static { + /// List a page of ACLs. Requires USAGE on the object, or system if unfiltered. + async fn list_acls( + &self, + request: ListAccessControlEntriesRequest, + ) -> Result; + /// Grant a privilege; requires OWNERSHIP. Duplicate grants return HTTP 409. + /// Granting OWNERSHIP transfers it from the previous owner. + async fn add_acl(&self, request: AccessControlEntryRequest) -> Result<()>; + /// Revoke a privilege; requires OWNERSHIP. Missing grants return HTTP 404. + /// OWNERSHIP cannot be deleted. + async fn delete_acl(&self, request: AccessControlEntryRequest) -> Result<()>; + /// List a page of direct bindings; requires USAGE on system. + async fn list_role_bindings( + &self, + request: ListRoleBindingsRequest, + ) -> Result; + /// Idempotently grant a role; requires OPERATE on system. + async fn add_role_binding(&self, request: RoleBindingRequest) -> Result<()>; + /// Idempotently revoke a role; requires OPERATE on system. + async fn delete_role_binding(&self, request: RoleBindingRequest) -> Result<()>; + /// Inspect a principal and effective roles; requires USAGE on system. + async fn get_principal(&self, request: GetPrincipalRequest) -> Result; + /// Inspect a group and direct roles; requires USAGE on system. + async fn get_group(&self, request: GetGroupRequest) -> Result; +} + +impl AccessControlEntryRequest { + /// Build one grant from validated resource and identity selectors. + pub fn new(object: Object, subject: Subject, privilege: Privilege) -> Self { + Self { + object, + subject, + privilege, + } + } +} + +impl ListRoleBindingsRequest { + /// Reject role subjects, since roles cannot contain other roles. + pub fn validate(&self) -> Result<()> { + if let Some(subject) = &self.subject { + subject.validate_role_member()?; + } + Ok(()) + } +} + +impl RoleBindingRequest { + /// Build a role grant/revocation for a principal or group. + pub fn new(subject: Subject, role: impl Into) -> Result { + subject.validate_role_member()?; + Ok(Self { + subject, + role: role.into(), + }) + } + + /// Validate the member selector, including deserialized requests. + pub fn validate(&self) -> Result<()> { + self.subject.validate_role_member() + } +} + +impl GetPrincipalRequest { + /// Select a principal by ID, name, or API key. + pub fn new(principal: Subject) -> Result { + principal.validate_principal()?; + Ok(Self { principal }) + } + + /// Validate the lookup selector, including deserialized requests. + pub fn validate(&self) -> Result<()> { + self.principal.validate_principal() + } +} + +impl GetGroupRequest { + /// Select a group by ID or name. + pub fn new(group: Subject) -> Result { + group.validate_group()?; + Ok(Self { group }) + } + + /// Validate the lookup selector, including deserialized requests. + pub fn validate(&self) -> Result<()> { + self.group.validate_group() + } +} diff --git a/rust/lancedb/src/authz/mod.rs b/rust/lancedb/src/authz/mod.rs new file mode 100644 index 000000000..3191627e0 --- /dev/null +++ b/rust/lancedb/src/authz/mod.rs @@ -0,0 +1,48 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +//! Deployment-wide authorization, accessed through a catalog. +//! +//! Subjects use `p:ID`, `P:NAME`, `g:ID`, `G:NAME`, `r:ROLE`, or `a:API_KEY`. +//! Objects use literal names such as `table:tenant/db:public$events`. +//! Names are resolved by the server. Request types intentionally do not implement +//! Debug because their subject selectors may contain secret API keys. +//! +//! ```no_run +//! # #[cfg(feature = "remote")] +//! # async fn example() -> lancedb::Result<()> { +//! let catalog = lancedb::connect_catalog("https://example.com") +//! .api_key("secret") +//! .execute().await?; +//! let authz = catalog.authz()?; +//! use lancedb::authz::{AccessControlEntryRequest, Object, TableObject, NamespacePath, Subject, Privilege}; +//! let grant = AccessControlEntryRequest::new( +//! Object::Table(TableObject { +//! database: "analytics".into(), +//! namespace: NamespacePath::from_component("public"), +//! table: "foo".into(), +//! }), +//! Subject::role("reader")?, +//! Privilege::Select, +//! ); +//! authz.add_acl(grant).await?; +//! let page = authz.list_acls(Default::default()).await?; +//! for entry in page.entries { +//! println!("{} {} {}", entry.object, entry.subject_id, entry.privilege); +//! } +//! # Ok(()) +//! # } +//! ``` + +pub mod api; +pub mod object; +pub mod privilege; +pub mod subject; + +pub use api::*; +pub use object::{ + DatabaseObject, FunctionObject, NamespaceObject, NamespacePath, Object, SecretObject, + TableObject, ViewObject, +}; +pub use privilege::*; +pub use subject::{Subject, SubjectKind}; diff --git a/rust/lancedb/src/authz/object.rs b/rust/lancedb/src/authz/object.rs new file mode 100644 index 000000000..8b8bf1611 --- /dev/null +++ b/rust/lancedb/src/authz/object.rs @@ -0,0 +1,925 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +use crate::utils::{ + validate_database_name, validate_namespace_component, validate_secret_name, validate_table_name, +}; +use serde::{Deserialize, Deserializer, Serialize, Serializer, de::Error as _, ser::Error as _}; + +/// The default namespace for tables, views, secrets, and functions. +pub const DEFAULT_NAMESPACE: &str = "public"; + +pub const SYSTEM_TYPE_STRING: &str = "system"; +pub const DATABASE_TYPE_STRING: &str = "database"; +pub const NAMESPACE_TYPE_STRING: &str = "namespace"; +pub const TABLE_TYPE_STRING: &str = "table"; +pub const VIEW_TYPE_STRING: &str = "view"; +pub const SECRET_TYPE_STRING: &str = "secret"; +pub const FUNCTION_TYPE_STRING: &str = "function"; +pub const UNKNOWN_TYPE_STRING: &str = "unknown"; + +/// A non-empty path through the namespace hierarchy. +/// +/// The `$` separator is only the canonical wire encoding. Authorization code +/// should use the components and prefix helpers instead of interpreting the +/// encoded form. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +pub struct NamespacePath(Vec); + +impl NamespacePath { + /// Construct a path, validating each component without splitting separators. + pub fn new(components: Vec) -> Result { + if components.is_empty() { + return Err("namespace paths must contain at least one component".to_string()); + } + for component in &components { + validate_namespace_component(component).map_err(|error| error.to_string())?; + } + Ok(Self(components)) + } + + /// Construct a single-component path. + /// + /// # Panics + /// Panics if the component is invalid. Use [`Self::new`] for untrusted input. + pub fn from_component(component: impl Into) -> Self { + Self::new(vec![component.into()]).expect("namespace components must be valid") + } + + pub fn components(&self) -> &[String] { + &self.0 + } + + pub fn parent(&self) -> Option { + (self.0.len() > 1).then(|| Self(self.0[..self.0.len() - 1].to_vec())) + } + + pub fn prefixes(&self) -> impl Iterator + '_ { + (1..=self.0.len()).map(|length| Self(self.0[..length].to_vec())) + } + + pub fn is_descendant_of(&self, ancestor: &Self) -> bool { + self.0.len() > ancestor.0.len() && self.0.starts_with(&ancestor.0) + } + + pub fn parse(path: &str) -> Result { + Self::new(path.split('$').map(str::to_string).collect()) + } +} + +impl std::fmt::Display for NamespacePath { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.0.join("$")) + } +} + +/// An authorization resource, serialized using its canonical wire name. +/// +/// Unknown resource types are preserved so clients can read rules created by +/// newer servers. Known resource types validate names during parsing and serialization. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +pub enum Object { + System, + Database(DatabaseObject), + Namespace(NamespaceObject), + Table(TableObject), + View(ViewObject), + Secret(SecretObject), + Function(FunctionObject), + Unknown(String), +} + +impl Object { + /// Validate resource names, including objects recovered from storage keys. + pub fn validate(&self) -> Result<(), String> { + match self { + Self::System => Ok(()), + Self::Database(object) => object.validate(), + Self::Namespace(object) => object.validate(), + Self::Table(object) => object.validate(), + Self::View(object) => object.validate(), + Self::Secret(object) => object.validate(), + Self::Function(object) => object.validate(), + Self::Unknown(value) if value.is_empty() => Err("object must not be empty".into()), + Self::Unknown(_) => Ok(()), + } + } + + /// The database an object belongs to, or None. + pub fn database(&self) -> Option<&str> { + match self { + Self::System => None, + Self::Database(object) => Some(&object.database), + Self::Namespace(object) => Some(&object.database), + Self::Table(object) => Some(&object.database), + Self::View(object) => Some(&object.database), + Self::Secret(object) => Some(&object.database), + Self::Function(object) => Some(&object.database), + Self::Unknown(_) => None, + } + } + + pub fn type_name(&self) -> &'static str { + match self { + Self::System => SYSTEM_TYPE_STRING, + Self::Database(_) => DATABASE_TYPE_STRING, + Self::Namespace(_) => NAMESPACE_TYPE_STRING, + Self::Table(_) => TABLE_TYPE_STRING, + Self::View(_) => VIEW_TYPE_STRING, + Self::Secret(_) => SECRET_TYPE_STRING, + Self::Function(_) => FUNCTION_TYPE_STRING, + Self::Unknown(_) => UNKNOWN_TYPE_STRING, + } + } +} + +impl std::str::FromStr for Object { + type Err = crate::Error; + + fn from_str(value: &str) -> crate::Result { + Self::deserialize(serde::de::value::StrDeserializer::::new(value)) + .map_err(|error| crate::Error::InvalidInput { + message: error.to_string(), + }) + } +} + +impl std::fmt::Display for Object { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::System => f.write_str(SYSTEM_TYPE_STRING), + Self::Database(object) => object.fmt(f), + Self::Namespace(object) => object.fmt(f), + Self::Table(object) => object.fmt(f), + Self::View(object) => object.fmt(f), + Self::Secret(object) => object.fmt(f), + Self::Function(object) => object.fmt(f), + Self::Unknown(s) => f.write_str(s), + } + } +} + +impl Serialize for Object { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + self.validate().map_err(S::Error::custom)?; + self.to_string().serialize(serializer) + } +} + +impl<'de> Deserialize<'de> for Object { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let s = String::deserialize(deserializer)?; + let kind = s.split(':').next().unwrap_or_default(); + if s == SYSTEM_TYPE_STRING { + Ok(Self::System) + } else if s.is_empty() || kind == SYSTEM_TYPE_STRING { + Err(D::Error::custom(format!("invalid Object {s:?}"))) + } else if kind == DATABASE_TYPE_STRING { + let database = DatabaseObject::deserialize(serde::de::value::StringDeserializer::< + D::Error, + >::new(s))?; + Ok(Self::Database(database)) + } else if kind == NAMESPACE_TYPE_STRING { + let namespace = NamespaceObject::deserialize(serde::de::value::StringDeserializer::< + D::Error, + >::new(s))?; + Ok(Self::Namespace(namespace)) + } else if kind == TABLE_TYPE_STRING { + let table = + TableObject::deserialize(serde::de::value::StringDeserializer::::new(s))?; + Ok(Self::Table(table)) + } else if kind == VIEW_TYPE_STRING { + let view = + ViewObject::deserialize(serde::de::value::StringDeserializer::::new(s))?; + Ok(Self::View(view)) + } else if kind == SECRET_TYPE_STRING { + let secret = SecretObject::deserialize( + serde::de::value::StringDeserializer::::new(s), + )?; + Ok(Self::Secret(secret)) + } else if kind == FUNCTION_TYPE_STRING { + let function = FunctionObject::deserialize(serde::de::value::StringDeserializer::< + D::Error, + >::new(s))?; + Ok(Self::Function(function)) + } else { + Ok(Self::Unknown(s)) + } + } +} + +/// A database resource, encoded as `database:NAME`. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +pub struct DatabaseObject { + pub database: String, +} + +impl DatabaseObject { + pub fn new(database: String) -> Self { + Self { database } + } + + fn validate(&self) -> Result<(), String> { + validate_database_name(&self.database).map_err(|error| error.to_string())?; + Ok(()) + } +} + +impl std::fmt::Display for DatabaseObject { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "database:{}", self.database) + } +} + +impl Serialize for DatabaseObject { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + self.validate().map_err(S::Error::custom)?; + self.to_string().serialize(serializer) + } +} + +impl<'de> Deserialize<'de> for DatabaseObject { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let s = String::deserialize(deserializer)?; + let database = s + .strip_prefix("database:") + .ok_or_else(|| D::Error::custom(format!("invalid DatabaseObject {s:?}")))?; + let object = Self { + database: database.to_string(), + }; + object.validate().map_err(D::Error::custom)?; + Ok(object) + } +} + +/// A namespace resource, encoded as `namespace:DATABASE:PATH`. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +pub struct NamespaceObject { + pub database: String, + pub namespace: NamespacePath, +} + +impl NamespaceObject { + fn validate(&self) -> Result<(), String> { + validate_database_name(&self.database).map_err(|error| error.to_string())?; + Ok(()) + } +} + +impl std::fmt::Display for NamespaceObject { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "namespace:{}:{}", self.database, self.namespace) + } +} + +impl Serialize for NamespaceObject { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + self.validate().map_err(S::Error::custom)?; + self.to_string().serialize(serializer) + } +} + +impl<'de> Deserialize<'de> for NamespaceObject { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let s = String::deserialize(deserializer)?; + let invalid = || D::Error::custom(format!("invalid NamespaceObject {s:?}")); + let rest = s.strip_prefix("namespace:").ok_or_else(invalid)?; + let (database, namespace) = rest.split_once(':').ok_or_else(invalid)?; + let object = Self { + database: database.to_string(), + namespace: NamespacePath::parse(namespace).map_err(D::Error::custom)?, + }; + object.validate().map_err(D::Error::custom)?; + Ok(object) + } +} + +/// A table resource, encoded as `table:DATABASE:NAMESPACE$TABLE`. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +pub struct TableObject { + pub database: String, + pub namespace: NamespacePath, + pub table: String, +} + +impl TableObject { + fn validate(&self) -> Result<(), String> { + validate_database_name(&self.database).map_err(|error| error.to_string())?; + validate_table_name(&self.table).map_err(|error| error.to_string())?; + Ok(()) + } +} + +impl std::fmt::Display for TableObject { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "table:{}:{}${}", + self.database, self.namespace, self.table + ) + } +} + +impl Serialize for TableObject { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + self.validate().map_err(S::Error::custom)?; + self.to_string().serialize(serializer) + } +} + +impl<'de> Deserialize<'de> for TableObject { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let s = String::deserialize(deserializer)?; + let invalid = || D::Error::custom(format!("invalid TableObject {s:?}")); + let rest = s.strip_prefix("table:").ok_or_else(invalid)?; + let (database, path) = rest.split_once(':').ok_or_else(invalid)?; + let (namespace, table) = path.rsplit_once('$').ok_or_else(invalid)?; + let object = Self { + database: database.to_string(), + namespace: NamespacePath::parse(namespace).map_err(D::Error::custom)?, + table: table.to_string(), + }; + object.validate().map_err(D::Error::custom)?; + Ok(object) + } +} + +/// A view resource, encoded as `view:DATABASE:NAMESPACE$VIEW`. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +pub struct ViewObject { + pub database: String, + pub namespace: NamespacePath, + pub view: String, +} + +impl ViewObject { + fn validate(&self) -> Result<(), String> { + validate_database_name(&self.database).map_err(|error| error.to_string())?; + validate_table_name(&self.view).map_err(|error| error.to_string())?; + Ok(()) + } +} + +impl std::fmt::Display for ViewObject { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "view:{}:{}${}", self.database, self.namespace, self.view) + } +} + +impl Serialize for ViewObject { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + self.validate().map_err(S::Error::custom)?; + self.to_string().serialize(serializer) + } +} + +impl<'de> Deserialize<'de> for ViewObject { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let s = String::deserialize(deserializer)?; + let invalid = || D::Error::custom(format!("invalid ViewObject {s:?}")); + let rest = s.strip_prefix("view:").ok_or_else(invalid)?; + let (database, path) = rest.split_once(':').ok_or_else(invalid)?; + let (namespace, view) = path.rsplit_once('$').ok_or_else(invalid)?; + let object = Self { + database: database.to_string(), + namespace: NamespacePath::parse(namespace).map_err(D::Error::custom)?, + view: view.to_string(), + }; + object.validate().map_err(D::Error::custom)?; + Ok(object) + } +} +/// A secret resource, encoded as `secret:DATABASE:NAMESPACE$SECRET`. +/// +/// ``` +/// use lancedb::authz::{SecretObject, NamespacePath, Object}; +/// +/// let object = Object::Secret(SecretObject { +/// database: "analytics".into(), +/// namespace: NamespacePath::from_component("public"), +/// secret: "example".into(), +/// }); +/// assert_eq!(object.to_string(), "secret:analytics:public$example"); +/// ``` +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +pub struct SecretObject { + pub database: String, + pub namespace: NamespacePath, + pub secret: String, +} + +impl SecretObject { + fn validate(&self) -> Result<(), String> { + validate_database_name(&self.database).map_err(|error| error.to_string())?; + validate_secret_name(&self.secret).map_err(|error| error.to_string())?; + Ok(()) + } +} + +impl std::fmt::Display for SecretObject { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "secret:{}:{}${}", + self.database, self.namespace, self.secret + ) + } +} + +impl Serialize for SecretObject { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + self.validate().map_err(S::Error::custom)?; + self.to_string().serialize(serializer) + } +} + +impl<'de> Deserialize<'de> for SecretObject { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let s = String::deserialize(deserializer)?; + let invalid = || D::Error::custom(format!("invalid SecretObject {s:?}")); + let rest = s.strip_prefix("secret:").ok_or_else(invalid)?; + let (database, path) = rest.split_once(':').ok_or_else(invalid)?; + let (namespace, secret) = path.rsplit_once('$').ok_or_else(invalid)?; + let object = Self { + database: database.to_string(), + namespace: NamespacePath::parse(namespace).map_err(D::Error::custom)?, + secret: secret.to_string(), + }; + object.validate().map_err(D::Error::custom)?; + Ok(object) + } +} + +/// A function resource, encoded as `function:DATABASE:NAMESPACE$FUNCTION`. +/// +/// ``` +/// use lancedb::authz::{FunctionObject, NamespacePath, Object}; +/// +/// let object = Object::Function(FunctionObject { +/// database: "analytics".into(), +/// namespace: NamespacePath::from_component("public"), +/// function: "example".into(), +/// }); +/// assert_eq!(object.to_string(), "function:analytics:public$example"); +/// ``` +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +pub struct FunctionObject { + pub database: String, + pub namespace: NamespacePath, + pub function: String, +} + +impl FunctionObject { + fn validate(&self) -> Result<(), String> { + validate_database_name(&self.database).map_err(|error| error.to_string())?; + validate_table_name(&self.function).map_err(|error| error.to_string())?; + Ok(()) + } +} + +impl std::fmt::Display for FunctionObject { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "function:{}:{}${}", + self.database, self.namespace, self.function + ) + } +} + +impl Serialize for FunctionObject { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + self.validate().map_err(S::Error::custom)?; + self.to_string().serialize(serializer) + } +} + +impl<'de> Deserialize<'de> for FunctionObject { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let s = String::deserialize(deserializer)?; + let invalid = || D::Error::custom(format!("invalid FunctionObject {s:?}")); + let rest = s.strip_prefix("function:").ok_or_else(invalid)?; + let (database, path) = rest.split_once(':').ok_or_else(invalid)?; + let (namespace, function) = path.rsplit_once('$').ok_or_else(invalid)?; + let object = Self { + database: database.to_string(), + namespace: NamespacePath::parse(namespace).map_err(D::Error::custom)?, + function: function.to_string(), + }; + object.validate().map_err(D::Error::custom)?; + Ok(object) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn system_object_round_trip() { + let encoded = serde_json::to_string(&Object::System).unwrap(); + assert_eq!(encoded, r#""system""#); + assert_eq!( + serde_json::from_str::(&encoded).unwrap(), + Object::System + ); + } + + #[test] + fn object_database_round_trip() { + let object = Object::Database(DatabaseObject { + database: "db".to_string(), + }); + let encoded = serde_json::to_string(&object).unwrap(); + assert_eq!(encoded, r#""database:db""#); + assert_eq!(serde_json::from_str::(&encoded).unwrap(), object); + } + + #[test] + fn object_namespace_round_trip() { + let object = Object::Namespace(NamespaceObject { + database: "db".to_string(), + namespace: NamespacePath::from_component("ns"), + }); + let encoded = serde_json::to_string(&object).unwrap(); + assert_eq!(encoded, r#""namespace:db:ns""#); + assert_eq!(serde_json::from_str::(&encoded).unwrap(), object); + } + + #[test] + fn object_table_round_trip() { + let object = Object::Table(TableObject { + database: "db".to_string(), + namespace: NamespacePath::from_component("ns"), + table: "foo".to_string(), + }); + let encoded = serde_json::to_string(&object).unwrap(); + assert_eq!(encoded, r#""table:db:ns$foo""#); + assert_eq!(serde_json::from_str::(&encoded).unwrap(), object); + } + + #[test] + fn object_view_round_trip() { + let object = Object::View(ViewObject { + database: "db".to_string(), + namespace: NamespacePath::from_component("ns"), + view: "foo".to_string(), + }); + let encoded = serde_json::to_string(&object).unwrap(); + assert_eq!(encoded, r#""view:db:ns$foo""#); + assert_eq!(serde_json::from_str::(&encoded).unwrap(), object); + } + + #[test] + fn secret_objects_round_trip_and_validate() { + let resource = SecretObject { + database: "tenant/db".into(), + namespace: NamespacePath::new(vec!["a".into(), "b".into()]).unwrap(), + secret: "example".into(), + }; + let wire = "secret:tenant/db:a$b$example"; + assert_eq!(serde_json::to_value(&resource).unwrap(), wire); + assert_eq!( + serde_json::from_value::(serde_json::json!(wire)).unwrap(), + resource + ); + let object = Object::Secret(resource.clone()); + assert_eq!(object.database(), Some("tenant/db")); + assert_eq!(object.type_name(), "secret"); + assert_eq!(wire.parse::().unwrap(), object); + assert_eq!(serde_json::to_value(object).unwrap(), wire); + for name in ["", ".", "..", "bad$name", "bad/name", "bad:name"] { + let invalid = SecretObject { + secret: name.into(), + ..resource.clone() + }; + assert!(serde_json::to_value(&invalid).is_err()); + assert!(serde_json::to_value(Object::Secret(invalid)).is_err()); + } + } + + #[test] + fn function_objects_round_trip_and_validate() { + let resource = FunctionObject { + database: "tenant/db".into(), + namespace: NamespacePath::new(vec!["a".into(), "b".into()]).unwrap(), + function: "example".into(), + }; + let wire = "function:tenant/db:a$b$example"; + assert_eq!(serde_json::to_value(&resource).unwrap(), wire); + assert_eq!( + serde_json::from_value::(serde_json::json!(wire)).unwrap(), + resource + ); + let object = Object::Function(resource.clone()); + assert_eq!(object.database(), Some("tenant/db")); + assert_eq!(object.type_name(), "function"); + assert_eq!(wire.parse::().unwrap(), object); + assert_eq!(serde_json::to_value(object).unwrap(), wire); + for name in ["", ".", "..", "bad$name", "bad/name", "bad:name"] { + let invalid = FunctionObject { + function: name.into(), + ..resource.clone() + }; + assert!(serde_json::to_value(&invalid).is_err()); + assert!(serde_json::to_value(Object::Function(invalid)).is_err()); + } + } + + #[test] + fn unknown_objects_round_trip() { + for wire in [ + "future:db:resource", + "table_extension:db:resource", + "garbage", + ] { + let object: Object = wire.parse().unwrap(); + assert_eq!(object, Object::Unknown(wire.into())); + assert_eq!(object.database(), None); + assert_eq!(object.type_name(), UNKNOWN_TYPE_STRING); + assert_eq!(object.to_string(), wire); + assert_eq!(serde_json::to_value(&object).unwrap(), wire); + assert_eq!( + serde_json::from_value::(serde_json::json!(wire)).unwrap(), + object + ); + } + } + + #[test] + fn database_object_round_trip() { + let object = DatabaseObject { + database: "db".to_string(), + }; + let encoded = serde_json::to_string(&object).unwrap(); + assert_eq!(encoded, r#""database:db""#); + assert_eq!( + serde_json::from_str::(&encoded).unwrap(), + object + ); + } + + #[test] + fn namespace_object_round_trip() { + let object = NamespaceObject { + database: "db".to_string(), + namespace: NamespacePath::from_component("ns"), + }; + let encoded = serde_json::to_string(&object).unwrap(); + assert_eq!(encoded, r#""namespace:db:ns""#); + assert_eq!( + serde_json::from_str::(&encoded).unwrap(), + object + ); + } + + #[test] + fn namespace_object_rejects_missing_namespace() { + assert!(serde_json::from_str::(r#""namespace:db""#).is_err()); + } + + #[test] + fn table_object_round_trip() { + let object = TableObject { + database: "db".to_string(), + namespace: NamespacePath::from_component("ns"), + table: "foo".to_string(), + }; + let encoded = serde_json::to_string(&object).unwrap(); + assert_eq!(encoded, r#""table:db:ns$foo""#); + assert_eq!( + serde_json::from_str::(&encoded).unwrap(), + object + ); + } + + #[test] + fn table_object_rejects_missing_prefix() { + assert!(serde_json::from_str::(r#""foo""#).is_err()); + } + + #[test] + fn table_object_rejects_missing_table() { + assert!(serde_json::from_str::(r#""table:db:ns""#).is_err()); + } + + #[test] + fn objects_order_by_variant_then_path_components() { + let database = Object::Database(DatabaseObject::new("z".to_string())); + let namespace = Object::Namespace(NamespaceObject { + database: "a".to_string(), + namespace: NamespacePath::from_component("a"), + }); + assert!(Object::System < database); + assert!(database < namespace); + + let first_table = Object::Table(TableObject { + database: "db".to_string(), + namespace: NamespacePath::from_component("a"), + table: "z".to_string(), + }); + let second_table = Object::Table(TableObject { + database: "db".to_string(), + namespace: NamespacePath::from_component("b"), + table: "a".to_string(), + }); + assert!(first_table < second_table); + } + + #[test] + fn namespace_path_returns_cumulative_prefixes() { + let path = NamespacePath::new(vec!["ns1".into(), "ns2".into(), "ns3".into()]).unwrap(); + assert_eq!( + path.prefixes() + .map(|prefix| prefix.to_string()) + .collect::>(), + vec!["ns1", "ns1$ns2", "ns1$ns2$ns3"] + ); + assert_eq!(path.parent().unwrap().to_string(), "ns1$ns2"); + } + + #[test] + fn nested_objects_round_trip_without_escaping() { + for wire in [ + "database:tenant/db", + "namespace:tenant/db:a$b", + "table:tenant/db:a$b$c", + "view:tenant/db:a$b$c", + "secret:tenant/db:a$b$c", + "function:tenant/db:a$b$c", + ] { + let object: Object = serde_json::from_value(serde_json::json!(wire)).unwrap(); + assert_eq!(wire.parse::().unwrap(), object); + assert_eq!(object.database(), Some("tenant/db")); + assert_eq!(object.to_string(), wire); + assert_eq!(serde_json::to_value(object).unwrap(), wire); + } + } + + #[test] + fn objects_reject_invalid_names_and_legacy_forms() { + for database in [ + "", + ".", + "..", + "db:bad", + "db%2Fchild", + "db$child", + "/db", + "db/", + "db//child", + "db/../child", + "db name", + ] { + for wire in [ + format!("database:{database}"), + format!("namespace:{database}:ns"), + format!("table:{database}:ns$t"), + format!("view:{database}:ns$v"), + format!("secret:{database}:ns$s"), + format!("function:{database}:ns$f"), + ] { + assert!( + serde_json::from_value::(serde_json::json!(wire)).is_err(), + "{wire}" + ); + } + } + for name in ["", "bad:name", "bad/name", "bad%24name", "bad name"] { + for wire in [ + format!("namespace:db:{name}"), + format!("table:db:{name}$t"), + format!("view:db:{name}$v"), + format!("table:db:ns${name}"), + format!("view:db:ns${name}"), + format!("secret:db:{name}$s"), + format!("function:db:{name}$f"), + format!("secret:db:ns${name}"), + format!("function:db:ns${name}"), + ] { + assert!( + serde_json::from_value::(serde_json::json!(wire)).is_err(), + "{wire}" + ); + } + } + for wire in [ + "", + "system:db", + "database", + "namespace", + "table", + "view", + "namespace:db/ns", + "table:db/ns/t", + "view:db/ns/v", + "table:db:ns", + "view:db:ns", + "namespace:db:a$$b", + "table:db:ns$.", + "view:db:ns$..", + "table:db:a$$t", + "view:db:$v", + "secret", + "function", + "secret:db/ns/s", + "function:db/ns/f", + "secret:db:ns", + "function:db:ns", + "secret:db:a$$s", + "function:db:$f", + "secret:db:ns$.", + "function:db:ns$..", + ] { + assert!( + serde_json::from_value::(serde_json::json!(wire)).is_err(), + "{wire}" + ); + assert!(matches!( + wire.parse::(), + Err(crate::Error::InvalidInput { .. }) + )); + } + } + + #[test] + fn namespace_components_use_namespace_validation() { + for component in ["a$b", "c/d", "e%f", "a:b", "", ".", ".."] { + assert!(NamespacePath::new(vec![component.into()]).is_err()); + } + assert!(NamespacePath::new(vec!["a.b".into(), "c..d".into()]).is_ok()); + } + + #[test] + fn serialization_rejects_invalid_directly_constructed_objects() { + let database = DatabaseObject::new("db:bad".into()); + let namespace = NamespaceObject { + database: "db%2Fbad".into(), + namespace: NamespacePath::from_component("ns"), + }; + let table = TableObject { + database: "db".into(), + namespace: NamespacePath::from_component("ns"), + table: "bad$name".into(), + }; + let view = ViewObject { + database: "db".into(), + namespace: NamespacePath::from_component("ns"), + view: "..".into(), + }; + assert!(serde_json::to_string(&database).is_err()); + assert!(serde_json::to_string(&namespace).is_err()); + assert!(serde_json::to_string(&table).is_err()); + assert!(serde_json::to_string(&view).is_err()); + for object in [ + Object::Database(database), + Object::Namespace(namespace), + Object::Table(table), + Object::View(view), + ] { + assert!(serde_json::to_string(&object).is_err()); + } + } +} diff --git a/rust/lancedb/src/authz/privilege.rs b/rust/lancedb/src/authz/privilege.rs new file mode 100644 index 000000000..d8e94596f --- /dev/null +++ b/rust/lancedb/src/authz/privilege.rs @@ -0,0 +1,330 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +use std::{fmt, str::FromStr}; + +use serde::{Deserialize, Deserializer, Serialize, Serializer, de::Error as _}; + +/// A LanceDB privilege, serialized by name (for example, `SELECT`). +/// Future privilege names use `PRIVILEGE` followed by decimal digits. They are +/// preserved verbatim; their meaning and validity are determined by the server. +/// +/// ``` +/// use lancedb::authz::Privilege; +/// +/// let privilege: Privilege = "PRIVILEGE256".parse().unwrap(); +/// assert_eq!(privilege.to_string(), "PRIVILEGE256"); +/// +/// // Delegate creation on a namespace without transferring its ownership. +/// for privilege in [Privilege::CreateSecret, Privilege::CreateFunction] { +/// assert!(privilege.valid_for_namespace()); +/// } +/// ``` +#[derive(Debug, Clone, PartialEq, Eq, strum_macros::Display)] +#[strum(serialize_all = "SCREAMING_SNAKE_CASE")] +pub enum Privilege { + /// A special privilege that conveys ownership. + Ownership, + + /// On a database: allows accessing the namespaces inside. + /// On a namespace: allows accessing the tables inside. + /// + /// When looking for Usage on a Database, if there are no ACLs + /// found for that database, we assume it doesn't exist yet and + /// check for CreateDatabase on System. + /// + /// When looking for Usage on a public (aka default) Namespace, + /// if there are no ACLs found for that Namespace, we assume it + /// doesn't exist yet and check for CreateNamespace on the + /// containing database instead. + Usage, + + /// On a table: allows reading rows. + Select, + + /// On a table: allows appending rows. + Insert, + + /// On a table: allows modifying existing rows. + Update, + + /// On a table: allows deleting rows. + Delete, + + /// On the system object: can create a database. + CreateDatabase, + + /// On a database or namespace: can create immediate child namespaces. + CreateNamespace, + + /// On a namespace: can create a table in that namespace. + CreateTable, + + /// On a namespace: can create a materialized view in that namespace. + CreateMaterializedView, + + /// On a namespace: can create a view in that namespace. + CreateView, + + /// On a namespace: can create a secret without taking ownership of the namespace. + CreateSecret, + + /// On a namespace: can create a function without taking ownership of the namespace. + CreateFunction, + + /// On the system object: + /// - can read server configuration. + Operate, + + /// A future privilege name, preserved verbatim. + #[strum(to_string = "{0}")] + Unknown(UnknownPrivilege), +} + +/// An unknown privilege. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct UnknownPrivilege(String); + +impl fmt::Display for UnknownPrivilege { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + self.0.fmt(f) + } +} + +impl Privilege { + pub const KNOWN: [Self; 14] = [ + Self::Ownership, + Self::Usage, + Self::Select, + Self::Insert, + Self::Update, + Self::Delete, + Self::CreateDatabase, + Self::CreateNamespace, + Self::CreateTable, + Self::CreateMaterializedView, + Self::CreateView, + Self::CreateSecret, + Self::CreateFunction, + Self::Operate, + ]; + + pub fn valid_for_system(&self) -> bool { + matches!( + self, + Self::Operate | Self::CreateDatabase | Self::Usage | Self::Unknown(_) + ) + } + + pub fn valid_for_database(&self) -> bool { + matches!( + self, + Self::Ownership | Self::Usage | Self::CreateNamespace | Self::Unknown(_) + ) + } + + pub fn valid_for_namespace(&self) -> bool { + matches!( + self, + Self::Ownership + | Self::Usage + | Self::CreateNamespace + | Self::CreateTable + | Self::CreateMaterializedView + | Self::CreateView + | Self::CreateSecret + | Self::CreateFunction + | Self::Unknown(_) + ) + } + + pub fn valid_for_table(&self) -> bool { + matches!( + self, + Self::Ownership + | Self::Select + | Self::Insert + | Self::Update + | Self::Delete + | Self::Unknown(_) + ) + } + + pub fn valid_for_view(&self) -> bool { + matches!(self, Self::Ownership | Self::Select | Self::Unknown(_)) + } + + /// Whether this privilege can mutate database contents or schema. + /// Unknown privileges are conservatively treated as writes. + pub fn is_write(&self) -> bool { + !matches!(self, Self::Usage | Self::Select | Self::Operate) + } +} + +impl FromStr for Privilege { + type Err = String; + + fn from_str(name: &str) -> Result { + if let Some(privilege) = Self::KNOWN.into_iter().find(|p| p.to_string() == name) { + return Ok(privilege); + } + if let Some(digits) = name.strip_prefix("PRIVILEGE") + && !digits.is_empty() + && digits.bytes().all(|b| b.is_ascii_digit()) + { + return Ok(Self::Unknown(UnknownPrivilege(name.to_owned()))); + } + Err(format!("invalid privilege {name:?}")) + } +} + +impl Serialize for Privilege { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + serializer.collect_str(self) + } +} + +impl<'de> Deserialize<'de> for Privilege { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + String::deserialize(deserializer)? + .parse() + .map_err(D::Error::custom) + } +} + +#[cfg(test)] +mod tests { + use super::Privilege; + + #[test] + fn display_matches_serde() { + for privilege in Privilege::KNOWN { + assert_eq!( + serde_json::to_string(&privilege).unwrap(), + format!("\"{privilege}\""), + ); + } + } + + #[test] + fn secret_and_function_creation_privileges() { + for (name, privilege) in [ + ("CREATE_SECRET", Privilege::CreateSecret), + ("CREATE_FUNCTION", Privilege::CreateFunction), + ] { + assert_eq!(name.parse::().unwrap(), privilege); + assert_eq!(privilege.to_string(), name); + assert_eq!(serde_json::to_value(&privilege).unwrap(), name); + assert_eq!( + serde_json::from_value::(serde_json::json!(name)).unwrap(), + privilege + ); + assert!(privilege.valid_for_namespace()); + assert!(!privilege.valid_for_system()); + assert!(!privilege.valid_for_database()); + assert!(!privilege.valid_for_table()); + assert!(!privilege.valid_for_view()); + assert!(privilege.is_write()); + } + } + + #[test] + fn future_privilege_names_round_trip() { + for name in [ + "PRIVILEGE0", + "PRIVILEGE2", + "PRIVILEGE20", + "PRIVILEGE64", + "PRIVILEGE256", + "PRIVILEGE18446744073709551616", + "PRIVILEGE002", + ] { + let privilege: Privilege = name.parse().unwrap(); + assert!(matches!(privilege, Privilege::Unknown(_))); + assert_eq!(privilege.to_string(), name); + assert_eq!(serde_json::to_value(&privilege).unwrap(), name); + assert_eq!( + serde_json::from_value::(serde_json::json!(name)).unwrap(), + privilege + ); + } + assert_ne!( + "PRIVILEGE2".parse::().unwrap(), + Privilege::Select + ); + } + + #[test] + fn rejects_invalid_privileges() { + for name in [ + "", + "select", + " SELECT", + "SELECT ", + "PRIVILEGE", + "PRIVILEGE-1", + "PRIVILEGE+1", + "PRIVILEGE 1", + "PRIVILEGE1x", + "privilege12", + "UNKNOWN", + ] { + assert!(name.parse::().is_err(), "{name}"); + assert!( + serde_json::from_value::(serde_json::json!(name)).is_err(), + "{name}" + ); + } + } + + #[test] + fn unknown_privileges_are_valid_for_all_objects_and_treated_as_writes() { + for name in [ + "PRIVILEGE0", + "PRIVILEGE2", + "PRIVILEGE20", + "PRIVILEGE64", + "PRIVILEGE256", + ] { + let privilege: Privilege = name.parse().unwrap(); + assert!(privilege.valid_for_system()); + assert!(privilege.valid_for_database()); + assert!(privilege.valid_for_namespace()); + assert!(privilege.valid_for_table()); + assert!(privilege.valid_for_view()); + assert!(privilege.is_write()); + } + assert!(!Privilege::Ownership.valid_for_system()); + assert!(!Privilege::Select.valid_for_database()); + } + + #[test] + fn test_privilege_is_write() { + for privilege in Privilege::KNOWN { + assert_eq!( + privilege.is_write(), + matches!( + privilege, + Privilege::Ownership + | Privilege::Insert + | Privilege::Update + | Privilege::Delete + | Privilege::CreateDatabase + | Privilege::CreateNamespace + | Privilege::CreateTable + | Privilege::CreateMaterializedView + | Privilege::CreateView + | Privilege::CreateSecret + | Privilege::CreateFunction + ), + "unexpected write classification for {privilege}" + ); + } + } +} diff --git a/rust/lancedb/src/authz/subject.rs b/rust/lancedb/src/authz/subject.rs new file mode 100644 index 000000000..2cc607fc4 --- /dev/null +++ b/rust/lancedb/src/authz/subject.rs @@ -0,0 +1,378 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +//! Identity selectors. Provider names resolve on the server; IDs remain literal. + +use std::{fmt, str::FromStr}; + +use serde::{Deserialize, Deserializer, Serialize, Serializer, de::Error as _}; + +use crate::{Error, Result}; + +/// The kind of identifier supplied to an authorization operation. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +#[non_exhaustive] +pub enum SubjectKind { + PrincipalId, + PrincipalName, + GroupId, + GroupName, + Role, + PrincipalApiKey, + /// An unrecognized canonical subject. Its full wire string is stored as the value. + Unknown, +} + +impl SubjectKind { + /// Stable name used by language bindings. + pub fn as_str(self) -> &'static str { + match self { + Self::PrincipalId => "principal_id", + Self::PrincipalName => "principal_name", + Self::GroupId => "group_id", + Self::GroupName => "group_name", + Self::Role => "role", + Self::PrincipalApiKey => "principal_api_key", + Self::Unknown => "unknown", + } + } + + fn prefix(self) -> &'static str { + match self { + Self::PrincipalId => "p:", + Self::PrincipalName => "P:", + Self::GroupId => "g:", + Self::GroupName => "G:", + Self::Role => "r:", + Self::PrincipalApiKey => "a:", + Self::Unknown => "", + } + } +} + +const KINDS: [SubjectKind; 6] = [ + SubjectKind::PrincipalId, + SubjectKind::PrincipalName, + SubjectKind::GroupId, + SubjectKind::GroupName, + SubjectKind::Role, + SubjectKind::PrincipalApiKey, +]; + +impl FromStr for SubjectKind { + type Err = Error; + + fn from_str(value: &str) -> Result { + if value == "unknown" { + return Ok(Self::Unknown); + } + KINDS + .into_iter() + .find(|kind| kind.as_str() == value) + .ok_or_else(|| invalid("Invalid subject kind")) + } +} + +/// An immutable validated principal, group, or role selector. +/// +/// Unrecognized canonical prefixes are preserved as [`SubjectKind::Unknown`], +/// with the complete wire string in [`Self::value`], so future subject kinds can +/// be listed and sent back to the server. +/// +/// ``` +/// use lancedb::authz::{Subject, SubjectKind}; +/// let subject = Subject::from_canonical("x:future-subject")?; +/// assert_eq!(subject.kind(), SubjectKind::Unknown); +/// assert_eq!(subject.to_wire(), "x:future-subject"); +/// # Ok::<(), lancedb::Error>(()) +/// ``` +/// +/// Debug and Display redact API keys. [`Self::to_wire`] and serialization +/// intentionally expose the key for transport and must not be logged. +/// +/// ``` +/// use lancedb::authz::Subject; +/// let subject = Subject::group_name("Engineering")?; +/// assert_eq!(subject.to_wire(), "G:Engineering"); +/// assert_eq!(subject, "G:Engineering".parse::()?); +/// # Ok::<(), lancedb::Error>(()) +/// ``` +#[derive(Clone, PartialEq, Eq, Hash)] +pub struct Subject { + kind: SubjectKind, + value: String, +} + +fn invalid(message: &str) -> Error { + // Never include the selector: it may contain an API key. + Error::InvalidInput { + message: message.into(), + } +} + +impl Subject { + /// Construct a selector without resolving it against an identity provider. + pub fn new(kind: SubjectKind, value: impl Into) -> Result { + let value = value.into(); + if value.is_empty() { + return Err(invalid("Subject value must be a nonempty string")); + } + if kind == SubjectKind::Unknown + && (!value + .split_once(':') + .is_some_and(|(prefix, value)| !prefix.is_empty() && !value.is_empty()) + || KINDS.iter().any(|kind| value.starts_with(kind.prefix()))) + { + return Err(invalid("Invalid unknown subject selector")); + } + Ok(Self { kind, value }) + } + + /// Select a principal by canonical ID, without stripping any prefix in the ID. + pub fn principal_id(id: impl Into) -> Result { + Self::new(SubjectKind::PrincipalId, id) + } + /// Select a principal by an unambiguous provider name. + pub fn principal_name(name: impl Into) -> Result { + Self::new(SubjectKind::PrincipalName, name) + } + /// Select a group by canonical ID. + pub fn group_id(id: impl Into) -> Result { + Self::new(SubjectKind::GroupId, id) + } + /// Select a group by an unambiguous provider name. + pub fn group_name(name: impl Into) -> Result { + Self::new(SubjectKind::GroupName, name) + } + /// Select a LanceDB-managed role. + pub fn role(name: impl Into) -> Result { + Self::new(SubjectKind::Role, name) + } + /// Select the principal holding this secret API key, not the request's caller. + pub fn principal_api_key(key: impl Into) -> Result { + Self::new(SubjectKind::PrincipalApiKey, key) + } + + /// The selector kind. + pub fn kind(&self) -> SubjectKind { + self.kind + } + /// The selected ID, name, role, or API key. API-key values are secrets. + /// For unknown kinds, this is the full canonical wire string. + pub fn value(&self) -> &str { + &self.value + } + /// The value safe for display; API keys are replaced with ``. + pub fn display_value(&self) -> &str { + if self.kind == SubjectKind::PrincipalApiKey { + "" + } else { + &self.value + } + } + /// Encode the literal wire selector. The result can contain a secret API key. + pub fn to_wire(&self) -> String { + format!("{}{}", self.kind.prefix(), self.value) + } + + /// Parse a returned canonical identity, rejecting names and secret selectors. + pub fn from_canonical(value: &str) -> Result { + let subject: Self = value.parse()?; + if !matches!( + subject.kind, + SubjectKind::PrincipalId + | SubjectKind::GroupId + | SubjectKind::Role + | SubjectKind::Unknown + ) { + return Err(invalid("Server returned a non-canonical subject")); + } + Ok(subject) + } + + /// Deserialize a canonical response subject, rejecting secrets and name selectors. + pub fn deserialize_canonical<'de, D: Deserializer<'de>>( + deserializer: D, + ) -> std::result::Result { + Self::from_canonical(&String::deserialize(deserializer)?).map_err(D::Error::custom) + } + + /// Require a principal selector for a principal lookup. + pub fn validate_principal(&self) -> Result<()> { + if matches!( + self.kind, + SubjectKind::PrincipalId | SubjectKind::PrincipalName | SubjectKind::PrincipalApiKey + ) { + Ok(()) + } else { + Err(invalid("Subject kind is not valid for a principal lookup")) + } + } + /// Require a group selector for a group lookup. + pub fn validate_group(&self) -> Result<()> { + if matches!(self.kind, SubjectKind::GroupId | SubjectKind::GroupName) { + Ok(()) + } else { + Err(invalid("Subject kind is not valid for a group lookup")) + } + } + /// Require a principal/group selector. Roles cannot contain other roles. + pub fn validate_role_member(&self) -> Result<()> { + if self.kind == SubjectKind::Role { + Err(invalid("Roles cannot be members of other roles")) + } else { + Ok(()) + } + } +} + +impl FromStr for Subject { + type Err = Error; + fn from_str(value: &str) -> Result { + for kind in KINDS { + if let Some(value) = value.strip_prefix(kind.prefix()) { + return Self::new(kind, value); + } + } + Self::new(SubjectKind::Unknown, value) + } +} + +impl fmt::Debug for Subject { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("Subject") + .field("kind", &self.kind) + .field("value", &self.display_value()) + .finish() + } +} +impl fmt::Display for Subject { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}{}", self.kind.prefix(), self.display_value()) + } +} +impl Serialize for Subject { + fn serialize(&self, serializer: S) -> std::result::Result { + serializer.serialize_str(&self.to_wire()) + } +} +impl<'de> Deserialize<'de> for Subject { + fn deserialize>(deserializer: D) -> std::result::Result { + String::deserialize(deserializer)? + .parse() + .map_err(D::Error::custom) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn selectors_roundtrip_without_normalizing_values() { + for kind in KINDS { + for value in [ + "alice@example.com", + "p:literal-id", + "Engineering / R&D", + "名前", + ] { + let subject = Subject::new(kind, value).unwrap(); + assert_eq!(subject.to_wire().parse::().unwrap(), subject); + let json = serde_json::to_string(&subject).unwrap(); + assert_eq!(serde_json::from_str::(&json).unwrap(), subject); + assert_eq!(subject.value(), value); + assert!(Subject::new(kind, "").is_err()); + } + } + } + + #[test] + fn contexts_and_canonical_ids() { + for kind in KINDS { + let subject = Subject::new(kind, "value").unwrap(); + assert_eq!( + subject.validate_principal().is_ok(), + matches!( + kind, + SubjectKind::PrincipalId + | SubjectKind::PrincipalName + | SubjectKind::PrincipalApiKey + ) + ); + assert_eq!( + subject.validate_group().is_ok(), + matches!(kind, SubjectKind::GroupId | SubjectKind::GroupName) + ); + assert_eq!( + subject.validate_role_member().is_ok(), + kind != SubjectKind::Role + ); + assert_eq!( + Subject::from_canonical(&subject.to_wire()).is_ok(), + matches!( + kind, + SubjectKind::PrincipalId | SubjectKind::GroupId | SubjectKind::Role + ) + ); + } + } + + #[test] + fn unknown_subjects_preserve_canonical_wire_strings() { + for wire in ["x:future-subject", "future:literal:identifier"] { + let subject = Subject::from_canonical(wire).unwrap(); + assert_eq!(subject.kind(), SubjectKind::Unknown); + assert_eq!(subject.value(), wire); + assert_eq!(subject.to_wire(), wire); + assert_eq!(serde_json::to_value(&subject).unwrap(), wire); + assert_eq!( + serde_json::from_value::(serde_json::json!(wire)).unwrap(), + subject + ); + assert!(subject.validate_principal().is_err()); + assert!(subject.validate_group().is_err()); + } + for wire in [ + "", + "no-prefix", + ":value", + "x:", + "p:value", + "P:value", + "g:value", + "G:value", + "r:value", + "a:secret", + ] { + assert!(Subject::new(SubjectKind::Unknown, wire).is_err()); + } + } + + #[test] + fn acl_page_preserves_known_and_unknown_subjects() { + let payload = serde_json::json!({"entries": [ + {"subject_id": "p:known", "object": "system", "privilege": "USAGE"}, + {"subject_id": "x:future-subject", "object": "system", "privilege": "USAGE"} + ]}); + let page: crate::authz::ListAccessControlEntriesResponse = + serde_json::from_value(payload.clone()).unwrap(); + assert_eq!(page.entries.len(), 2); + assert_eq!(page.entries[0].subject_id.kind(), SubjectKind::PrincipalId); + assert_eq!(page.entries[1].subject_id.kind(), SubjectKind::Unknown); + assert_eq!(serde_json::to_value(page).unwrap(), payload); + } + + #[test] + fn secrets_are_only_exposed_by_explicit_serialization() { + let secret = "never-log-this-key"; + let subject = Subject::principal_api_key(secret).unwrap(); + assert!(!format!("{subject:?} {subject}").contains(secret)); + assert!(serde_json::to_string(&subject).unwrap().contains(secret)); + let error = Subject::from_canonical(&subject.to_wire()).unwrap_err(); + assert!(!error.to_string().contains(secret)); + assert!(!format!("{error:?}").contains(secret)); + let error = secret.parse::().unwrap_err(); + assert!(!error.to_string().contains(secret)); + } +} diff --git a/rust/lancedb/src/catalog.rs b/rust/lancedb/src/catalog.rs index 576eb8e88..ad8674b77 100644 --- a/rust/lancedb/src/catalog.rs +++ b/rust/lancedb/src/catalog.rs @@ -193,6 +193,13 @@ impl Stream for DatabaseNames { /// A backend that manages databases. Implementations own database lifecycle semantics. #[async_trait::async_trait] pub trait Catalog: Send + Sync + std::fmt::Debug + 'static { + /// Obtain deployment-wide authorization, if supported by this backend. + fn authz(&self) -> Result> { + Err(crate::Error::NotSupported { + message: "Authorization administration is not supported by this catalog".into(), + }) + } + /// Catalog endpoint or location. fn uri(&self) -> &str; /// Create a database, or open it when `exist_ok` permits. @@ -239,6 +246,11 @@ impl CatalogConnection { pub fn uri(&self) -> &str { self.catalog.uri() } + /// Obtain a handle sharing this catalog's deployment scope and credentials. + pub fn authz(&self) -> Result> { + self.catalog.authz() + } + /// Access the underlying backend. pub fn catalog(&self) -> &Arc { &self.catalog diff --git a/rust/lancedb/src/lib.rs b/rust/lancedb/src/lib.rs index e215f3c6b..7750f66b5 100644 --- a/rust/lancedb/src/lib.rs +++ b/rust/lancedb/src/lib.rs @@ -173,6 +173,7 @@ #![recursion_limit = "256"] pub mod arrow; +pub mod authz; pub mod blob; pub mod catalog; pub mod connection; diff --git a/rust/lancedb/src/remote.rs b/rust/lancedb/src/remote.rs index bc534c0f4..a6b2f0e62 100644 --- a/rust/lancedb/src/remote.rs +++ b/rust/lancedb/src/remote.rs @@ -6,6 +6,7 @@ //! building client/server applications with LanceDB or as a client for some //! other custom LanceDB service. +mod authz; pub mod catalog; pub(crate) mod client; pub(crate) mod db; diff --git a/rust/lancedb/src/remote/authz.rs b/rust/lancedb/src/remote/authz.rs new file mode 100644 index 000000000..88c2841ce --- /dev/null +++ b/rust/lancedb/src/remote/authz.rs @@ -0,0 +1,254 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +use super::client::{HttpSend, RequestResultExt, RestfulLanceDbClient, Sender}; +use crate::Result; +use crate::authz::*; +use serde::{Serialize, de::DeserializeOwned}; + +pub struct RemoteAuthorization { + pub(crate) client: RestfulLanceDbClient, +} + +impl std::fmt::Debug for RemoteAuthorization { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("RemoteAuthorization") + .finish_non_exhaustive() + } +} + +impl RemoteAuthorization { + async fn post( + &self, + route: &str, + request: &T, + ) -> Result<(String, reqwest::Response)> { + // Structured objects can be constructed with invalid names. Serialize + // before building the request so validation errors remain local errors. + let body = serde_json::to_value(request).map_err(|error| crate::Error::InvalidInput { + message: error.to_string(), + })?; + let request = self + .client + .post(&format!("/admin/authz/{route}")) + .json(&body); + // Every subject selector can contain a secret. Never log these bodies, + // and never replay an ACL mutation after an ambiguous response. + let (id, response) = self.client.send_suppressing_body(request).await?; + let response = self.client.check_response(&id, response).await?; + Ok((id, response)) + } + + async fn read( + &self, + route: &str, + request: &T, + ) -> Result { + let (id, response) = self.post(route, request).await?; + response.json().await.err_to_http(id) + } +} + +#[async_trait::async_trait] +impl Authorization for RemoteAuthorization { + async fn list_acls( + &self, + request: ListAccessControlEntriesRequest, + ) -> Result { + self.read("acl/list", &request).await + } + async fn add_acl(&self, request: AccessControlEntryRequest) -> Result<()> { + self.post("acl/add", &request).await?; + Ok(()) + } + async fn delete_acl(&self, request: AccessControlEntryRequest) -> Result<()> { + self.post("acl/delete", &request).await?; + Ok(()) + } + async fn list_role_bindings( + &self, + request: ListRoleBindingsRequest, + ) -> Result { + request.validate()?; + self.read("role_binding/list", &request).await + } + async fn add_role_binding(&self, request: RoleBindingRequest) -> Result<()> { + request.validate()?; + self.post("role_binding/add", &request).await?; + Ok(()) + } + async fn delete_role_binding(&self, request: RoleBindingRequest) -> Result<()> { + request.validate()?; + self.post("role_binding/delete", &request).await?; + Ok(()) + } + async fn get_principal(&self, request: GetPrincipalRequest) -> Result { + request.validate()?; + self.read("principal", &request).await + } + async fn get_group(&self, request: GetGroupRequest) -> Result { + request.validate()?; + self.read("group", &request).await + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::remote::client::test_utils::client_with_handler; + use serde_json::json; + use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }; + + #[tokio::test] + async fn list_preserves_wire_identity_and_continuation() { + let authz = RemoteAuthorization { + client: client_with_handler(|request| { + assert_eq!(request.url().path(), "/admin/authz/acl/list"); + let body: serde_json::Value = + serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap(); + assert_eq!( + body, + json!({"object": "table:tenant/db:a$b$events", "subject": "G:Engineering", "max_results": 100, "continuation_token": "opaque/token"}) + ); + http::Response::builder().status(200).body(json!({ + "entries": [{"object": "table:tenant/db:a$b$events", "subject_id": "g:team-id", "privilege": "PRIVILEGE20"}], + "continuation_token": "next/token" + }).to_string()).unwrap() + }), + }; + let result = authz + .list_acls(ListAccessControlEntriesRequest { + object: Some("table:tenant/db:a$b$events".parse().unwrap()), + subject: Some(Subject::group_name("Engineering").unwrap()), + max_results: Some(100), + continuation_token: Some("opaque/token".into()), + }) + .await + .unwrap(); + assert_eq!(result.continuation_token.as_deref(), Some("next/token")); + assert_eq!( + result.entries[0].subject_id, + Subject::group_id("team-id").unwrap() + ); + assert_eq!(result.entries[0].subject_name, None); + assert_eq!( + result.entries[0].privilege, + "PRIVILEGE20".parse::().unwrap() + ); + } + + #[tokio::test] + async fn acl_failure_is_not_retried_or_reclassified() { + for status in [404, 409, 503] { + let calls = Arc::new(AtomicUsize::new(0)); + let count = calls.clone(); + let authz = RemoteAuthorization { + client: client_with_handler(move |request| { + count.fetch_add(1, Ordering::SeqCst); + let body: serde_json::Value = + serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()) + .unwrap(); + assert_eq!(body["privilege"], "USAGE"); + http::Response::builder() + .status(status) + .body("failure") + .unwrap() + }), + }; + let result = authz + .add_acl(AccessControlEntryRequest { + object: Object::Database(DatabaseObject::new("db".into())), + subject: Subject::principal_api_key("secret").unwrap(), + privilege: Privilege::Usage, + }) + .await; + let Err(crate::Error::Http { + status_code, + request_id, + .. + }) = result + else { + panic!("expected Http error") + }; + assert_eq!(status_code.unwrap().as_u16(), status); + assert!(!request_id.is_empty()); + assert_eq!(calls.load(Ordering::SeqCst), 1); + } + } + + #[tokio::test] + async fn rejects_invalid_objects_before_sending() { + let authz = RemoteAuthorization { + client: client_with_handler(|_| -> http::Response { + panic!("invalid objects must not send a request") + }), + }; + let request = AccessControlEntryRequest::new( + Object::Database(DatabaseObject::new("bad:database".into())), + Subject::principal_api_key("secret").unwrap(), + Privilege::Usage, + ); + for result in [ + authz.add_acl(request.clone()).await, + authz.delete_acl(request.clone()).await, + authz + .list_acls(ListAccessControlEntriesRequest { + object: Some(request.object), + ..Default::default() + }) + .await + .map(|_| ()), + ] { + assert!(matches!(result, Err(crate::Error::InvalidInput { .. }))); + assert!(!result.unwrap_err().to_string().contains("secret")); + } + } + + #[tokio::test] + async fn rejects_incompatible_subjects_before_sending() { + let authz = RemoteAuthorization { + client: client_with_handler(|_| -> http::Response { + panic!("invalid selectors must not send a request") + }), + }; + let request = RoleBindingRequest { + subject: Subject::role("reader").unwrap(), + role: "writer".into(), + }; + assert!(matches!( + authz.add_role_binding(request.clone()).await, + Err(crate::Error::InvalidInput { .. }) + )); + assert!(matches!( + authz.delete_role_binding(request).await, + Err(crate::Error::InvalidInput { .. }) + )); + let request = ListRoleBindingsRequest { + subject: Some(Subject::role("reader").unwrap()), + ..Default::default() + }; + assert!(matches!( + authz.list_role_bindings(request).await, + Err(crate::Error::InvalidInput { .. }) + )); + assert!(matches!( + authz + .get_principal(GetPrincipalRequest { + principal: Subject::group_id("group").unwrap() + }) + .await, + Err(crate::Error::InvalidInput { .. }) + )); + assert!(matches!( + authz + .get_group(GetGroupRequest { + group: Subject::principal_api_key("secret").unwrap() + }) + .await, + Err(crate::Error::InvalidInput { .. }) + )); + } +} diff --git a/rust/lancedb/src/remote/catalog.rs b/rust/lancedb/src/remote/catalog.rs index fa6b185cc..df3782368 100644 --- a/rust/lancedb/src/remote/catalog.rs +++ b/rust/lancedb/src/remote/catalog.rs @@ -176,6 +176,10 @@ impl RemoteCatalog { #[async_trait] impl Catalog for RemoteCatalog { + fn authz(&self) -> Result> { + Ok(self.root.authz()) + } + fn uri(&self) -> &str { &self.endpoint } diff --git a/rust/lancedb/src/remote/db.rs b/rust/lancedb/src/remote/db.rs index 28582cdc3..a0f4aa953 100644 --- a/rust/lancedb/src/remote/db.rs +++ b/rust/lancedb/src/remote/db.rs @@ -297,6 +297,12 @@ pub struct RemoteHostOverrides { } impl RemoteDatabase { + pub(crate) fn authz(&self) -> Arc { + Arc::new(super::authz::RemoteAuthorization { + client: self.client.clone(), + }) + } + pub(crate) fn try_new( uri: &str, api_key: &str,