diff --git a/pkg-py/src/commons/_catalog/__init__.py b/pkg-py/src/commons/_catalog/__init__.py index 2590ec74..c7f12af4 100644 --- a/pkg-py/src/commons/_catalog/__init__.py +++ b/pkg-py/src/commons/_catalog/__init__.py @@ -4,9 +4,11 @@ types are authoritative, identifier case normalizes per backend, and an ambiguous relative name is an error rather than a guess. -Everything here is a pure function over the rows a warehouse listing returns. -Running the queries that produce those rows belongs to the per-backend -readers, which keeps this testable without a warehouse. +Interpreting a warehouse listing is kept to pure functions over the rows it +returns, and running the queries that produce them belongs to the per-backend +readers, which keeps the interpretation testable without a warehouse. The +session and access checks in `_security` are the exception, since asking the +warehouse is the whole point of them. """ from . import _databricks, _snowflake @@ -23,19 +25,41 @@ search, table_registry, ) +from ._security import ( + CatalogAccessError, + CatalogAuthorizationError, + CatalogSessionChangedError, + CatalogTransientError, + SessionSnapshot, + check_session, + ensure_queryable, + require_queryable, + require_queryable_relations, + session_snapshot, +) __all__ = [ + "CatalogAccessError", + "CatalogAuthorizationError", + "CatalogSessionChangedError", + "CatalogTransientError", "Manifest", "MergedDictionary", "Relation", "Selector", + "SessionSnapshot", "_databricks", "_snowflake", "check_exclude", + "check_session", + "ensure_queryable", "excluded", "id_type", "merge_dictionary", "normalize_identifier", + "require_queryable", + "require_queryable_relations", "search", + "session_snapshot", "table_registry", ] diff --git a/pkg-py/src/commons/_catalog/_core.py b/pkg-py/src/commons/_catalog/_core.py index ee7fb1a3..fc7b4853 100644 --- a/pkg-py/src/commons/_catalog/_core.py +++ b/pkg-py/src/commons/_catalog/_core.py @@ -174,12 +174,18 @@ def table_registry( # An entry naming a table is kept whether or not the warehouse # has it, and is always validated. Dropping a missing one turns # "that table is not there" into a quietly smaller selection. - table_id = _selector_id(selector) found = exact_relation(selector) - relations.append( - found if found is not None else Relation(id=table_id, discovered=False) + # Keyed by the relation's own id rather than the selector's: an + # entry naming a bare table is qualified with the connection's + # namespace once the warehouse answers, and the two lists have to + # agree on the label or the access check cannot pair them up. + relation = ( + found + if found is not None + else Relation(id=_selector_id(selector), discovered=False) ) - validate.append(table_id) + relations.append(relation) + validate.append(relation.id) continue namespace_selected = True relations.extend(list_relations(selector)) @@ -220,7 +226,9 @@ class Manifest: objects: dict[str, Relation] searchable: bool = False access: dict[str, str] = field(default_factory=dict) - access_errors: dict[str, str] = field(default_factory=dict) + # The driver's own failure, kept for the relations whose refusal is + # cached, so a later refusal can still be raised from what caused it. + access_errors: dict[str, BaseException] = field(default_factory=dict) @classmethod def build( diff --git a/pkg-py/src/commons/_catalog/_security.py b/pkg-py/src/commons/_catalog/_security.py new file mode 100644 index 00000000..4dae817b --- /dev/null +++ b/pkg-py/src/commons/_catalog/_security.py @@ -0,0 +1,353 @@ +"""Session identity and query access for a warehouse catalog. + +A warehouse decides what a principal may read, so commons never tries to +answer that itself: it asks, with a query that returns no rows, and reads the +answer off the failure. The classification matters because it decides what +happens next. An authorization refusal is stable, so it is remembered and the +relation is not probed again; a transient one is not, so the next touch +retries; anything unrecognized is neither cached nor treated as a refusal. + +The session snapshot exists for the same reason. Access was decided for the +principal, role, and namespace in force at discovery, so if any of those +change the answers no longer apply and the source has to be rebuilt. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import Any, NoReturn + +from .._data_source import TableId +from ._core import Manifest, Relation + +__all__ = [ + "CatalogAccessError", + "CatalogAuthorizationError", + "CatalogSessionChangedError", + "CatalogTransientError", + "Probe", + "SessionSnapshot", + "changed_session_fields", + "check_session", + "classify_access_error", + "ensure_queryable", + "probe_relation", + "probe_sql", + "require_queryable", + "require_queryable_relations", + "session_snapshot", +] + +_AUTHORIZATION_MESSAGE = re.compile( + "not authorized|insufficient privilege|permission denied|" + "access denied|does not have.*privilege|not permitted|" + "permission_denied|sql access control error|not allowed to access", + re.IGNORECASE, +) + +_TRANSIENT_MESSAGE = re.compile( + "temporar|timed? ?out|unavailable|connection.*(closed|reset|failed)|" + "warehouse.*(starting|stopped|unavailable)|throttl|rate limit|" + "network|socket|http (429|503)|unexpected eof", + re.IGNORECASE, +) + +_TRANSIENT_SQLSTATE = re.compile("^08|^HYT|^40|^57P01") + + +class CatalogAccessError(Exception): + """Query access to a relation could not be verified.""" + + +class CatalogAuthorizationError(CatalogAccessError): + """The current principal may not query the relation.""" + + +class CatalogTransientError(CatalogAccessError): + """Access could not be verified now, but might be on a later try.""" + + +class CatalogSessionChangedError(Exception): + """The connection's identity moved after the catalog was discovered.""" + + +@dataclass(frozen=True) +class SessionSnapshot: + """The connection identity the catalog's access answers were decided for.""" + + backend: str + principal: str | None + catalog: str | None + schema: str | None + role: str | None = None + secondary_roles: str | None = None + + +@dataclass(frozen=True) +class Probe: + state: str + error: BaseException | None = None + + +def session_snapshot(backend: Any) -> SessionSnapshot | None: + """Read the identity a Snowflake or Databricks connection is acting under. + + Any other backend returns None: nothing else commons supports scopes + catalog access to a role that can change underneath the source. + """ + dialect = backend.dialect() + if dialect == "snowflake": + sql = ( + "SELECT CURRENT_USER() AS principal, CURRENT_ROLE() AS role, " + "CURRENT_SECONDARY_ROLES() AS secondary_roles, " + "CURRENT_DATABASE() AS catalog, CURRENT_SCHEMA() AS schema" + ) + elif dialect == "databricks": + sql = ( + "SELECT CURRENT_USER() AS principal, " + "CURRENT_CATALOG() AS catalog, CURRENT_SCHEMA() AS schema" + ) + else: + return None + + try: + rows = backend.query(sql) + except Exception as error: + raise RuntimeError(f"Failed to read the {dialect} session identity.") from error + return _session_row(rows, dialect) + + +def _session_row(rows: list[dict[str, Any]], dialect: str) -> SessionSnapshot: + has_roles = dialect == "snowflake" + required = ["principal", "catalog", "schema"] + if has_roles: + required += ["role", "secondary_roles"] + row = {str(key).lower(): value for key, value in rows[0].items()} if rows else {} + if len(rows) != 1 or not all(field in row for field in required): + raise ValueError(f"{dialect} returned an invalid session identity response.") + return SessionSnapshot( + backend=dialect, + principal=_session_value(row["principal"]), + catalog=_session_value(row["catalog"]), + schema=_session_value(row["schema"]), + role=_session_value(row["role"]) if has_roles else None, + secondary_roles=(_session_value(row["secondary_roles"]) if has_roles else None), + ) + + +def _session_value(value: Any) -> str | None: + if value is None or str(value) == "": + return None + return str(value) + + +def check_session(backend: Any, snapshot: SessionSnapshot | None) -> None: + """Refuse to go on when the connection is no longer who it was.""" + if snapshot is None: + return + current = session_snapshot(backend) + if current == snapshot: + return + changed = [ + _FIELD_NAMES[field] for field in changed_session_fields(current, snapshot) + ] + raise CatalogSessionChangedError( + f"The connection {_listed(changed) or 'identity'} changed after " + f"catalog discovery; rebuild the data source." + ) + + +def _listed(items: Any) -> str: + """A comma-separated list a person would read out loud.""" + items = list(items) + if len(items) < 2: + return "".join(items) + return f"{', '.join(items[:-1])} and {items[-1]}" + + +# How each snapshot field is worded in the refusal. +_FIELD_NAMES = { + "principal": "principal", + "role": "active role", + "secondary_roles": "secondary roles", + "catalog": "catalog", + "schema": "schema", +} + + +def changed_session_fields( + current: SessionSnapshot | None, snapshot: SessionSnapshot +) -> list[str]: + """Which parts of the session identity differ, in snapshot order. + + The refusal names what moved rather than everything it compares, because + a backend need not have every field: Databricks reports no role, and + naming one there sends the user looking for something that cannot change. + """ + if current is None: + return [] + return [ + field + for field in _FIELD_NAMES + if getattr(current, field) != getattr(snapshot, field) + ] + + +def probe_sql(backend: Any, sql: str) -> Probe: + try: + backend.query(sql) + except Exception as error: # noqa: BLE001 - the failure is the answer + return Probe(classify_access_error(error), error) + return Probe("queryable") + + +def probe_relation(backend: Any, table_id: TableId) -> Probe: + return probe_sql(backend, f"SELECT * FROM {backend.quote(table_id)} WHERE 1 = 0") + + +def classify_access_error(error: BaseException) -> str: + """Read a driver's failure as authorization, transient, or neither. + + Conservative in both directions: a refusal is only called authorization + when the driver said so, and everything unrecognized stays unknown so it + is neither cached nor retried on its own. + """ + sqlstate = _sqlstate(error) + message = str(error) + if ( + sqlstate.startswith("28") + or sqlstate == "42501" + or _AUTHORIZATION_MESSAGE.search(message) + ): + return "authorization" + if _TRANSIENT_SQLSTATE.match(sqlstate) or _TRANSIENT_MESSAGE.search(message): + return "transient" + return "unknown" + + +def _sqlstate(error: BaseException) -> str: + """The SQLSTATE a driver reported, wherever it hung it. + + DBAPI drivers put it on the exception, SQLAlchemy wraps that exception in + one of its own, so the cause is read too. + """ + for candidate in (error, getattr(error, "orig", None), error.__cause__): + if candidate is None: + continue + for attribute in ("sqlstate", "state"): + value = getattr(candidate, attribute, None) + if isinstance(value, str) and value: + return value.upper() + return "" + + +def require_queryable( + backend: Any, table_id: TableId, label: str | None = None +) -> None: + probe = probe_relation(backend, table_id) + if probe.state != "queryable": + _abort_access(probe, label or table_id.label) + + +def require_queryable_relations( + backend: Any, + validate: dict[str, TableId], + relations: dict[str, Relation] | None = None, +) -> None: + """Check every explicitly named relation before the source is built. + + A name the warehouse never reported is missing rather than refused, and + saying so is more use than an access error about a table that is not + there. An unexplained failure is only reported as missing when the + backend can confirm the relation's absence. + + Every relation the listing did report is probed before anything is + raised. Stopping at the first refusal would report one problem at a time, + and would report it ahead of a name the caller simply got wrong. + """ + missing = [ + label + for label in validate + if relations is not None and not relations[label].discovered + ] + # Only the first refusal is raised, so only the first is kept: a + # selection may run to thousands of relations. + refused: tuple[Probe, str] | None = None + for label, table_id in validate.items(): + if label in missing: + continue + probe = probe_relation(backend, table_id) + if probe.state == "queryable": + continue + if probe.state == "unknown" and _relation_exists(backend, table_id) is False: + missing.append(label) + continue + refused = refused or (probe, label) + if missing: + _abort_missing(missing) + if refused is not None: + _abort_access(*refused) + + +def _relation_exists(backend: Any, table_id: TableId) -> bool | None: + inspector = backend.inspector() + if inspector is None: + return None + try: + return bool(inspector(table_id)) + except Exception: # noqa: BLE001 - inconclusive; the caller keeps the probe + return None + + +def ensure_queryable( + backend: Any, manifest: Manifest | None, label: str, table_id: TableId +) -> None: + """Check access to one relation at the moment it is about to be used. + + Construction only probes what the caller named and what the dictionary + matched, so a relation that arrived through a namespace listing is first + checked here. Its caller is the first-touch path that describes a table + to the agent, which lands with the retrieval surface; until then nothing + in commons resolves a label at query time. + """ + if manifest is None: + return + state = manifest.access.get(label, "unknown") + if state == "queryable": + return + if state == "authorization": + _abort_access(Probe(state, manifest.access_errors.get(label)), label) + + probe = probe_relation(backend, table_id) + if probe.state == "queryable": + manifest.access[label] = "queryable" + return + # Only stable authorization failures are cached; other failures retry. + if probe.state == "authorization": + manifest.access[label] = probe.state + if probe.error is not None: + manifest.access_errors[label] = probe.error + _abort_access(probe, label) + + +def _abort_missing(missing: list[str]) -> NoReturn: + noun = "a table" if len(missing) == 1 else "tables" + raise ValueError( + f"tables must not name {noun} the connection does not have: " + f"{_listed(repr(label) for label in missing)}." + ) + + +def _abort_access(probe: Probe, label: str) -> NoReturn: + if probe.state == "authorization": + raise CatalogAuthorizationError( + f"The current principal is not authorized to query {label!r}." + ) from probe.error + if probe.state == "transient": + raise CatalogTransientError( + f"Query access to {label!r} is temporarily unavailable." + ) from probe.error + raise CatalogAccessError( + f"Could not verify query access to {label!r}." + ) from probe.error diff --git a/pkg-py/tests/test_catalog.py b/pkg-py/tests/test_catalog.py index b80e3818..b786b598 100644 --- a/pkg-py/tests/test_catalog.py +++ b/pkg-py/tests/test_catalog.py @@ -111,6 +111,31 @@ def test_a_namespace_selection_lists_and_excludes(): assert registry.namespace_selected is True +def test_a_bare_relation_is_validated_under_the_label_it_came_back_with(): + # A selection entry naming a bare table is qualified by the warehouse + # from the connection's namespace. The access check pairs the two lists + # by label, so validate has to carry the qualified one. + registry = table_registry( + selectors=[Selector(table="orders")], + exact_relation=lambda selector: relation("main.sales.orders", kind="table"), + list_relations=lambda selector: [], + ) + + assert list(registry.relations) == ["main.sales.orders"] + assert list(registry.validate) == ["main.sales.orders"] + + +def test_a_bare_relation_the_warehouse_lacks_keeps_the_name_that_was_asked_for(): + registry = table_registry( + selectors=[Selector(table="orders")], + exact_relation=lambda selector: None, + list_relations=lambda selector: [], + ) + + assert list(registry.validate) == ["orders"] + assert registry.relations["orders"].discovered is False + + def test_a_selection_above_the_object_limit_is_refused(): with pytest.raises(ValueError, match="above the supported limit"): table_registry( diff --git a/pkg-py/tests/test_catalog_security.py b/pkg-py/tests/test_catalog_security.py new file mode 100644 index 00000000..f9ec3a1b --- /dev/null +++ b/pkg-py/tests/test_catalog_security.py @@ -0,0 +1,327 @@ +"""Session and access checks for a warehouse catalog. + +A fake backend stands in for the network here, not for a warehouse's access +rules: the point of each test is what commons does with the reply, since the +warehouse is the one deciding whether a principal may read a relation. +""" + +from typing import Any + +import pytest + +from commons._catalog import Manifest, Relation +from commons._catalog._security import ( + CatalogAccessError, + CatalogAuthorizationError, + CatalogSessionChangedError, + CatalogTransientError, + SessionSnapshot, + check_session, + classify_access_error, + ensure_queryable, + probe_relation, + require_queryable, + require_queryable_relations, + session_snapshot, +) +from commons._data_source import TableId + + +class DriverError(Exception): + """An error carrying a SQLSTATE, the way a warehouse driver's does.""" + + def __init__(self, message, sqlstate=None): + super().__init__(message) + self.sqlstate = sqlstate + + +class FakeBackend: + """Replays a canned reply, or raises, for each query it is given.""" + + def __init__(self, replies=None, dialect="snowflake", exists=None): + self._replies = list(replies or []) + self._dialect = dialect + self._exists = exists + self.queries: list[str] = [] + + def query(self, sql: str): + self.queries.append(sql) + reply = self._replies.pop(0) if self._replies else [] + if isinstance(reply, BaseException): + raise reply + return reply + + def quote(self, table_id: TableId) -> str: + return ".".join(f'"{part}"' for part in table_id.parts) + + def dialect(self) -> str: + return self._dialect + + def inspector(self): + if self._exists is None: + return None + return lambda table_id: self._exists + + +SALES = TableId(catalog="ANALYTICS", schema="PUBLIC", table="SALES") + + +def snowflake_row(role="REPORTER") -> dict[str, Any]: + return { + "PRINCIPAL": "ANALYST", + "ROLE": role, + "SECONDARY_ROLES": '{"roles":"READER","value":"ALL"}', + "CATALOG": "ANALYTICS", + "SCHEMA": "PUBLIC", + } + + +def test_session_snapshots_retain_authority_bearing_fields(): + backend = FakeBackend([[snowflake_row()]]) + + snapshot = session_snapshot(backend) + + assert snapshot == SessionSnapshot( + backend="snowflake", + principal="ANALYST", + role="REPORTER", + secondary_roles='{"roles":"READER","value":"ALL"}', + catalog="ANALYTICS", + schema="PUBLIC", + ) + + +def test_databricks_snapshots_carry_no_role(): + rows = [ + [{"principal": "analyst@example.com", "catalog": "main", "schema": "default"}] + ] + backend = FakeBackend(rows, dialect="databricks") + + snapshot = session_snapshot(backend) + + assert snapshot is not None + assert snapshot.role is None + assert snapshot.secondary_roles is None + assert (snapshot.catalog, snapshot.schema) == ("main", "default") + + +def test_other_backends_have_no_session_to_snapshot(): + assert session_snapshot(FakeBackend(dialect="duckdb")) is None + + +def test_an_empty_session_value_is_absent_rather_than_blank(): + row = snowflake_row() + row["ROLE"] = "" + row["SCHEMA"] = None + + snapshot = session_snapshot(FakeBackend([[row]])) + + assert snapshot is not None + assert snapshot.role is None + assert snapshot.schema is None + + +def test_an_unreadable_session_identity_is_an_error(): + backend = FakeBackend([DriverError("connection closed")]) + + with pytest.raises(RuntimeError, match="session identity"): + session_snapshot(backend) + + +def test_an_invalid_session_reply_is_an_error(): + with pytest.raises(ValueError, match="invalid session identity"): + session_snapshot(FakeBackend([[]])) + + with pytest.raises(ValueError, match="invalid session identity"): + session_snapshot(FakeBackend([[{"PRINCIPAL": "ANALYST"}]])) + + +def test_catalog_operations_reject_a_changed_session(): + taken = session_snapshot(FakeBackend([[snowflake_row()]])) + backend = FakeBackend([[snowflake_row(role="ADMIN")]]) + + with pytest.raises(CatalogSessionChangedError, match="active role changed"): + check_session(backend, taken) + + +def test_a_databricks_refusal_never_names_a_role(): + rows = {"principal": "analyst@example.com", "catalog": "main", "schema": "default"} + taken = session_snapshot(FakeBackend([[rows]], dialect="databricks")) + assert taken is not None + moved = FakeBackend([[{**rows, "principal": "other@example.com"}]], "databricks") + + with pytest.raises(CatalogSessionChangedError) as refusal: + check_session(moved, taken) + + assert "principal changed" in str(refusal.value) + assert "role" not in str(refusal.value) + + +def test_an_unchanged_session_passes(): + taken = session_snapshot(FakeBackend([[snowflake_row()]])) + backend = FakeBackend([[snowflake_row()]]) + + check_session(backend, taken) + + +def test_a_source_without_a_session_is_not_checked(): + backend = FakeBackend() + + check_session(backend, None) + + assert backend.queries == [] + + +def test_a_probe_reads_no_rows_from_the_relation(): + backend = FakeBackend() + + probe = probe_relation(backend, SALES) + + assert probe.state == "queryable" + assert backend.queries == ['SELECT * FROM "ANALYTICS"."PUBLIC"."SALES" WHERE 1 = 0'] + + +def test_require_queryable_names_the_relation_it_refused(): + backend = FakeBackend([DriverError("hidden", sqlstate="42501")]) + + with pytest.raises(CatalogAuthorizationError, match="ANALYTICS.PUBLIC.SALES"): + require_queryable(backend, SALES) + + +def test_transient_access_failures_remain_retryable(): + relations = {"sales": Relation(id=SALES)} + manifest = Manifest.build(relations) + backend = FakeBackend([DriverError("timed out"), []]) + + with pytest.raises(CatalogTransientError): + ensure_queryable(backend, manifest, "sales", SALES) + assert manifest.access["sales"] == "unknown" + + ensure_queryable(backend, manifest, "sales", SALES) + assert manifest.access["sales"] == "queryable" + assert len(backend.queries) == 2 + + +def test_authorization_failures_are_cached_per_relation(): + relations = {"sales": Relation(id=SALES)} + manifest = Manifest.build(relations) + backend = FakeBackend([DriverError("permission denied")]) + + for _ in range(2): + with pytest.raises(CatalogAuthorizationError): + ensure_queryable(backend, manifest, "sales", SALES) + + assert manifest.access["sales"] == "authorization" + assert len(backend.queries) == 1 + + +def test_a_relation_already_known_queryable_is_not_probed_again(): + manifest = Manifest.build({"sales": Relation(id=SALES)}) + manifest.access["sales"] = "queryable" + backend = FakeBackend() + + ensure_queryable(backend, manifest, "sales", SALES) + + assert backend.queries == [] + + +def test_a_source_without_a_manifest_has_nothing_to_check(): + backend = FakeBackend() + + ensure_queryable(backend, None, "sales", SALES) + + assert backend.queries == [] + + +def test_exact_relations_use_classified_access_probes(): + backend = FakeBackend([DriverError("warehouse is starting")]) + + with pytest.raises(CatalogTransientError): + require_queryable_relations(backend, {"ANALYTICS.PUBLIC.SALES": SALES}) + + +def test_an_undiscovered_relation_fails_before_the_access_probe(): + backend = FakeBackend() + relations = {"ANALYTICS.PUBLIC.SALES": Relation(id=SALES, discovered=False)} + + with pytest.raises(ValueError, match="must not name"): + require_queryable_relations( + backend, {"ANALYTICS.PUBLIC.SALES": SALES}, relations + ) + + assert backend.queries == [] + + +def test_an_unexplained_failure_on_an_absent_relation_reports_it_missing(): + backend = FakeBackend([DriverError("object not found")], exists=False) + + with pytest.raises(ValueError, match="must not name"): + require_queryable_relations(backend, {"ANALYTICS.PUBLIC.SALES": SALES}) + + +def test_an_unexplained_failure_is_kept_when_the_relation_is_there(): + backend = FakeBackend([DriverError("something odd")], exists=True) + + with pytest.raises(CatalogAccessError, match="Could not verify"): + require_queryable_relations(backend, {"ANALYTICS.PUBLIC.SALES": SALES}) + + +def test_a_backend_that_cannot_answer_existence_keeps_the_access_error(): + backend = FakeBackend([DriverError("something odd")]) + + with pytest.raises(CatalogAccessError, match="Could not verify"): + require_queryable_relations(backend, {"ANALYTICS.PUBLIC.SALES": SALES}) + + +def test_a_relation_the_listing_reported_is_probed_even_beside_a_missing_one(): + orders = TableId(catalog="ANALYTICS", schema="PUBLIC", table="ORDERS") + backend = FakeBackend([DriverError("permission denied")]) + relations = { + "ANALYTICS.PUBLIC.SALES": Relation(id=SALES, discovered=False), + "ANALYTICS.PUBLIC.ORDERS": Relation(id=orders), + } + + with pytest.raises(ValueError, match="ANALYTICS.PUBLIC.SALES"): + require_queryable_relations( + backend, + {"ANALYTICS.PUBLIC.SALES": SALES, "ANALYTICS.PUBLIC.ORDERS": orders}, + relations, + ) + + # The refused relation was still probed, so fixing the missing name is + # not a round trip spent to be told about the next problem. + assert len(backend.queries) == 1 + + +def test_a_missing_relation_is_reported_alongside_a_refused_one(): + orders = TableId(catalog="ANALYTICS", schema="PUBLIC", table="ORDERS") + backend = FakeBackend( + [DriverError("object not found"), DriverError("permission denied")], + exists=False, + ) + + with pytest.raises(ValueError, match="ANALYTICS.PUBLIC.SALES"): + require_queryable_relations( + backend, + {"ANALYTICS.PUBLIC.SALES": SALES, "ANALYTICS.PUBLIC.ORDERS": orders}, + ) + + +def test_a_failing_existence_check_keeps_the_access_error(): + backend = FakeBackend([DriverError("something odd")]) + backend.inspector = lambda: _raise_on_call + + with pytest.raises(CatalogAccessError, match="Could not verify"): + require_queryable_relations(backend, {"ANALYTICS.PUBLIC.SALES": SALES}) + + +def _raise_on_call(table_id): + raise DriverError("the inspector failed too") + + +def test_a_chained_driver_error_still_yields_its_sqlstate(): + inner = DriverError("hidden", sqlstate="42501") + outer = Exception("statement failed") + outer.__cause__ = inner + + assert classify_access_error(outer) == "authorization" diff --git a/pkg-py/tests/test_catalog_security_fixtures.py b/pkg-py/tests/test_catalog_security_fixtures.py new file mode 100644 index 00000000..29fadb2a --- /dev/null +++ b/pkg-py/tests/test_catalog_security_fixtures.py @@ -0,0 +1,133 @@ +"""Access-error classification and session comparison, against the shared +fixtures. + +Which failures are authorization, transient, or neither decides what the user +is told and whether the answer is cached, and which parts of a session are +reported as changed decides what they are told to look at, so both are pinned +once and read by both suites. +""" + +import pytest + +from commons._catalog import Relation +from commons._catalog._security import ( + CatalogAccessError, + CatalogAuthorizationError, + CatalogTransientError, + SessionSnapshot, + changed_session_fields, + classify_access_error, + require_queryable_relations, +) +from commons._data_source import TableId +from tests._shared import load_shared_fixture + + +class DriverError(Exception): + def __init__(self, message, sqlstate): + super().__init__(message) + self.sqlstate = sqlstate + + +def test_access_errors_match_the_shared_contract(): + cases = load_shared_fixture("catalog-access-errors")["cases"] + assert cases + + for case in cases: + error = DriverError(case["message"], case["sqlstate"] or None) + assert classify_access_error(error) == case["kind"], case["name"] + + +def test_changed_session_fields_match_the_shared_contract(): + cases = load_shared_fixture("catalog-session-changed")["cases"] + assert cases + + for case in cases: + before, after = _snapshot(case["before"]), _snapshot(case["after"]) + assert changed_session_fields(after, before) == case["changed"], case["name"] + + +def _snapshot(fields): + return SessionSnapshot( + backend=fields["backend"], + principal=fields["principal"], + catalog=fields["catalog"], + schema=fields["schema"], + role=fields["role"], + secondary_roles=fields["secondary_roles"], + ) + + +class _PrecedenceBackend: + """Answers each relation's probe from the fixture's own script.""" + + def __init__(self, relations): + self._script = {item["label"]: item for item in relations} + self.probed: list[str] = [] + + def query(self, sql: str): + label = sql.split('"')[1] + self.probed.append(label) + state = self._script[label]["probe"] + if state == "queryable": + return [] + raise _PROBE_ERRORS[state]() + + def quote(self, table_id): + return f'"{table_id.table}"' + + def dialect(self): + return "snowflake" + + def inspector(self): + def exists(table_id): + answer = self._script[table_id.table].get("exists", "unknown") + if answer == "unknown": + raise RuntimeError("the backend cannot say") + return answer == "true" + + return exists + + +_PROBE_ERRORS = { + "authorization": lambda: DriverError("permission denied", None), + "transient": lambda: DriverError("timed out", None), + "unknown": lambda: DriverError("something odd", None), +} + +_OUTCOMES = { + "missing": ValueError, + "authorization": CatalogAuthorizationError, + "transient": CatalogTransientError, + "access": CatalogAccessError, +} + + +def test_access_precedence_matches_the_shared_contract(): + cases = load_shared_fixture("catalog-access-precedence")["cases"] + assert cases + + for case in cases: + backend = _PrecedenceBackend(case["relations"]) + validate = { + item["label"]: TableId(table=item["label"]) for item in case["relations"] + } + relations = { + item["label"]: Relation( + id=validate[item["label"]], discovered=item["discovered"] == "true" + ) + for item in case["relations"] + } + expected = case["expected"] + if expected["outcome"] == "ok": + require_queryable_relations(backend, validate, relations) + else: + with pytest.raises(_OUTCOMES[expected["outcome"]]) as refusal: + require_queryable_relations(backend, validate, relations) + for label in expected["labels"]: + assert label in str(refusal.value), case["name"] + # The contract is that nothing is raised until every relation the + # listing reported has been probed, which only the probes can show. + assert backend.probed == [ + item["label"] for item in case["relations"] if item["discovered"] == "true" + ], case["name"] diff --git a/pkg-r/R/catalog-security.R b/pkg-r/R/catalog-security.R index 579343c3..5c40dae7 100644 --- a/pkg-r/R/catalog-security.R +++ b/pkg-r/R/catalog-security.R @@ -111,10 +111,16 @@ catalog_check_session_snapshot <- function( ) { current <- catalog_session_snapshot(con, call = call) if (!identical(current, snapshot)) { + fields <- catalog_session_changed_fields(current, snapshot) + # cli reads a character vector out as a list, commas and "and" included. + changed <- unname(catalog_session_field_names[fields]) + if (length(changed) == 0L) { + changed <- "identity" + } cli::cli_abort( paste( - "The connection principal, active role, or namespace changed after", - "catalog discovery; rebuild the data source." + "The connection {changed} changed after catalog discovery;", + "rebuild the data source." ), class = "commons_catalog_session_changed", call = call @@ -123,6 +129,38 @@ catalog_check_session_snapshot <- function( invisible(snapshot) } +# How each snapshot field is worded in the refusal. +catalog_session_field_names <- c( + principal = "principal", + role = "active role", + secondary_roles = "secondary roles", + catalog = "catalog", + schema = "schema" +) + +# Name what moved rather than everything the snapshot compares: Databricks +# has no role, so a fixed list would name a field that backend never had. +catalog_session_changed_fields <- function(current, snapshot) { + fields <- names(catalog_session_field_names) + fields[vapply( + fields, + function(field) { + !identical( + catalog_session_field(current, field), + catalog_session_field(snapshot, field) + ) + }, + logical(1) + )] +} + +catalog_session_field <- function(snapshot, field) { + if (field %in% c("catalog", "schema")) { + return(snapshot$namespace[[field]]) + } + snapshot[[field]] +} + catalog_probe_relation <- function(con, id) { catalog_probe_sql( con, @@ -164,7 +202,7 @@ catalog_access_error_kind <- function(err) { paste( "not authorized|insufficient privilege|permission denied|", "access denied|does not have.*privilege|not permitted|", - "permission_denied|sql access control error", + "permission_denied|sql access control error|not allowed to access", sep = "" ), conditionMessage(err), @@ -220,10 +258,16 @@ catalog_require_queryable_relations <- function( logical(1) )] } - if (length(missing)) { - catalog_abort_missing_relations(missing, call = call) - } + # Every relation the listing did report is probed before anything is + # raised, so a name the caller got wrong and a relation they cannot read + # are found in one pass rather than one round trip each. + # Only the first refusal is raised, so only the first is kept: a selection + # may run to thousands of relations. + refused <- NULL for (i in seq_along(registry$ids)) { + if (registry$labels[[i]] %in% missing) { + next + } probe <- catalog_probe_relation(con, registry$ids[[i]]) if (identical(probe$state, "queryable")) { next @@ -235,11 +279,16 @@ catalog_require_queryable_relations <- function( missing <- c(missing, registry$labels[[i]]) next } - catalog_abort_access(probe, registry$labels[[i]], call = call) + if (is.null(refused)) { + refused <- list(probe = probe, label = registry$labels[[i]]) + } } if (length(missing)) { catalog_abort_missing_relations(missing, call = call) } + if (!is.null(refused)) { + catalog_abort_access(refused$probe, refused$label, call = call) + } invisible(registry) } diff --git a/pkg-r/R/data-source.R b/pkg-r/R/data-source.R index c1437429..cabbc015 100644 --- a/pkg-r/R/data-source.R +++ b/pkg-r/R/data-source.R @@ -78,10 +78,11 @@ #' data frames, commons additionally disables extension loading and filesystem #' access. These are safeguards, not a sandbox: when you supply your own #' connection, still open it in read-only mode where the backend supports it. -#' Snowflake and Databricks sources snapshot the principal, active role, and -#' namespace at creation, then reject catalog access and trusted calculations -#' after those values change. Authored and native semantic material is exposed -#' only after a zero-row query succeeds for the current principal. +#' Snowflake and Databricks sources snapshot the principal and namespace at +#' creation, and Snowflake its active and secondary roles as well, then reject +#' catalog access and trusted calculations after any of those change. Authored +#' and native semantic material is exposed only after a zero-row query +#' succeeds for the current principal. #' #' @return A `commons_data_source` R6 object. Its internals are private and may #' change without notice. diff --git a/pkg-r/man/data_source.Rd b/pkg-r/man/data_source.Rd index ad45e24d..b150a105 100644 --- a/pkg-r/man/data_source.Rd +++ b/pkg-r/man/data_source.Rd @@ -98,10 +98,11 @@ rejected before reaching the database. For the in-process DuckDB built from data frames, commons additionally disables extension loading and filesystem access. These are safeguards, not a sandbox: when you supply your own connection, still open it in read-only mode where the backend supports it. -Snowflake and Databricks sources snapshot the principal, active role, and -namespace at creation, then reject catalog access and trusted calculations -after those values change. Authored and native semantic material is exposed -only after a zero-row query succeeds for the current principal. +Snowflake and Databricks sources snapshot the principal and namespace at +creation, and Snowflake its active and secondary roles as well, then reject +catalog access and trusted calculations after any of those change. Authored +and native semantic material is exposed only after a zero-row query +succeeds for the current principal. } \examples{ diff --git a/pkg-r/tests/testthat/fixtures/shared/catalog-access-errors.json b/pkg-r/tests/testthat/fixtures/shared/catalog-access-errors.json new file mode 100644 index 00000000..bce22ef1 --- /dev/null +++ b/pkg-r/tests/testthat/fixtures/shared/catalog-access-errors.json @@ -0,0 +1,101 @@ +{ + "description": "How a failed catalog access probe is classified from the driver's error. The classification decides which error a user sees and whether the failure is cached: an authorization failure is stable and is remembered per relation, a transient one is retried on the next touch, and an unknown one is neither. Drivers differ in what they raise, so each case gives only the two things every driver carries, a SQLSTATE (empty when the driver reported none) and a message.", + "cases": [ + { + "name": "insufficient privilege sqlstate", + "sqlstate": "42501", + "message": "hidden", + "kind": "authorization" + }, + { + "name": "invalid authorization sqlstate class", + "sqlstate": "28000", + "message": "hidden", + "kind": "authorization" + }, + { + "name": "snowflake access control message", + "sqlstate": "", + "message": "SQL access control error: Insufficient privileges to operate on table 'SALES'", + "kind": "authorization" + }, + { + "name": "databricks permission message", + "sqlstate": "", + "message": "PERMISSION_DENIED: User does not have SELECT on Table 'main.default.sales'", + "kind": "authorization" + }, + { + "name": "plain permission denied message", + "sqlstate": "", + "message": "Permission denied on relation sales", + "kind": "authorization" + }, + { + "name": "a network policy refusal is authorization, not a network fault", + "sqlstate": "", + "message": "Incoming request with IP 10.0.0.1 is not allowed to access Snowflake", + "kind": "authorization" + }, + { + "name": "connection exception sqlstate class", + "sqlstate": "08006", + "message": "hidden", + "kind": "transient" + }, + { + "name": "timeout sqlstate", + "sqlstate": "HYT00", + "message": "hidden", + "kind": "transient" + }, + { + "name": "serialization failure sqlstate", + "sqlstate": "40001", + "message": "hidden", + "kind": "transient" + }, + { + "name": "admin shutdown sqlstate", + "sqlstate": "57P01", + "message": "hidden", + "kind": "transient" + }, + { + "name": "warehouse starting message", + "sqlstate": "", + "message": "Warehouse 'COMPUTE_WH' is starting", + "kind": "transient" + }, + { + "name": "a temporarily unavailable warehouse", + "sqlstate": "", + "message": "Warehouse is temporarily unavailable", + "kind": "transient" + }, + { + "name": "rate limited message", + "sqlstate": "", + "message": "HTTP 429 too many requests", + "kind": "transient" + }, + { + "name": "lowercase sqlstate is still matched", + "sqlstate": "hyt00", + "message": "hidden", + "kind": "transient" + }, + { + "name": "syntax error with a sqlstate", + "sqlstate": "42601", + "message": "syntax error at or near SELCT", + "kind": "unknown" + }, + { + "name": "unrecognized failure without a sqlstate", + "sqlstate": "", + "message": "column BOGUS not found", + "kind": "unknown" + } + ] +} diff --git a/pkg-r/tests/testthat/fixtures/shared/catalog-access-precedence.json b/pkg-r/tests/testthat/fixtures/shared/catalog-access-precedence.json new file mode 100644 index 00000000..33fbbd81 --- /dev/null +++ b/pkg-r/tests/testthat/fixtures/shared/catalog-access-precedence.json @@ -0,0 +1,75 @@ +{ + "description": "What a source construction reports when several explicitly named relations fail at once. Every relation the listing did report is probed before anything is raised, so a name the caller got wrong and a relation they cannot read are found in one pass rather than one round trip each. A name the warehouse never listed is reported ahead of a refusal: it is the more actionable problem, and an access error about a table that is not there reads as a permissions problem the caller does not have. Booleans travel as strings so the file reads the same from both JSON readers. `discovered` is whether the listing reported the relation, `probe` is what the zero-row query did, and `exists` is what the backend answers about an unexplained failure.", + "cases": [ + { + "name": "every relation is readable", + "relations": [ + {"label": "A", "discovered": "true", "probe": "queryable"}, + {"label": "B", "discovered": "true", "probe": "queryable"} + ], + "expected": {"outcome": "ok", "labels": []} + }, + { + "name": "a name the listing never reported", + "relations": [ + {"label": "A", "discovered": "false", "probe": "queryable"}, + {"label": "B", "discovered": "true", "probe": "queryable"} + ], + "expected": {"outcome": "missing", "labels": ["A"]} + }, + { + "name": "a missing name is reported ahead of a refusal", + "relations": [ + {"label": "A", "discovered": "false", "probe": "queryable"}, + {"label": "B", "discovered": "true", "probe": "authorization"} + ], + "expected": {"outcome": "missing", "labels": ["A"]} + }, + { + "name": "a relation the probe proves absent is missing too", + "relations": [ + {"label": "A", "discovered": "true", "probe": "unknown", "exists": "false"}, + {"label": "B", "discovered": "true", "probe": "authorization"} + ], + "expected": {"outcome": "missing", "labels": ["A"]} + }, + { + "name": "every missing name is reported at once", + "relations": [ + {"label": "A", "discovered": "false", "probe": "queryable"}, + {"label": "B", "discovered": "true", "probe": "unknown", "exists": "false"} + ], + "expected": {"outcome": "missing", "labels": ["A", "B"]} + }, + { + "name": "the first refusal is the one raised", + "relations": [ + {"label": "A", "discovered": "true", "probe": "authorization"}, + {"label": "B", "discovered": "true", "probe": "transient"} + ], + "expected": {"outcome": "authorization", "labels": ["A"]} + }, + { + "name": "a transient refusal on its own", + "relations": [ + {"label": "A", "discovered": "true", "probe": "queryable"}, + {"label": "B", "discovered": "true", "probe": "transient"} + ], + "expected": {"outcome": "transient", "labels": ["B"]} + }, + { + "name": "an unexplained failure the backend cannot explain away", + "relations": [ + {"label": "A", "discovered": "true", "probe": "unknown", "exists": "unknown"} + ], + "expected": {"outcome": "access", "labels": ["A"]} + }, + { + "name": "an unexplained failure on a relation that is there", + "relations": [ + {"label": "A", "discovered": "true", "probe": "unknown", "exists": "true"} + ], + "expected": {"outcome": "access", "labels": ["A"]} + } + ] +} diff --git a/pkg-r/tests/testthat/fixtures/shared/catalog-session-changed.json b/pkg-r/tests/testthat/fixtures/shared/catalog-session-changed.json new file mode 100644 index 00000000..957b0223 --- /dev/null +++ b/pkg-r/tests/testthat/fixtures/shared/catalog-session-changed.json @@ -0,0 +1,41 @@ +{ + "description": "Which parts of a warehouse session identity a refusal reports as changed. The comparison runs before every catalog operation, and the answer decides what the user is told to look at, so it is pinned rather than described twice. A field is absent (null) when the backend has none: Databricks reports no role, so a refusal there must never name one. The names here are the snapshot's own fields; each implementation words them for its own message.", + "cases": [ + { + "name": "a snowflake role change", + "before": {"backend": "snowflake", "principal": "ANALYST", "role": "REPORTER", "secondary_roles": "ALL", "catalog": "ANALYTICS", "schema": "PUBLIC"}, + "after": {"backend": "snowflake", "principal": "ANALYST", "role": "ADMIN", "secondary_roles": "ALL", "catalog": "ANALYTICS", "schema": "PUBLIC"}, + "changed": ["role"] + }, + { + "name": "a snowflake secondary role change", + "before": {"backend": "snowflake", "principal": "ANALYST", "role": "REPORTER", "secondary_roles": "ALL", "catalog": "ANALYTICS", "schema": "PUBLIC"}, + "after": {"backend": "snowflake", "principal": "ANALYST", "role": "REPORTER", "secondary_roles": "NONE", "catalog": "ANALYTICS", "schema": "PUBLIC"}, + "changed": ["secondary_roles"] + }, + { + "name": "a databricks principal change names no role", + "before": {"backend": "databricks", "principal": "analyst@example.com", "role": null, "secondary_roles": null, "catalog": "main", "schema": "default"}, + "after": {"backend": "databricks", "principal": "other@example.com", "role": null, "secondary_roles": null, "catalog": "main", "schema": "default"}, + "changed": ["principal"] + }, + { + "name": "a databricks namespace change names both parts", + "before": {"backend": "databricks", "principal": "analyst@example.com", "role": null, "secondary_roles": null, "catalog": "main", "schema": "default"}, + "after": {"backend": "databricks", "principal": "analyst@example.com", "role": null, "secondary_roles": null, "catalog": "sandbox", "schema": "scratch"}, + "changed": ["catalog", "schema"] + }, + { + "name": "several fields at once, in snapshot order", + "before": {"backend": "snowflake", "principal": "ANALYST", "role": "REPORTER", "secondary_roles": "ALL", "catalog": "ANALYTICS", "schema": "PUBLIC"}, + "after": {"backend": "snowflake", "principal": "ADMIN_USER", "role": "ADMIN", "secondary_roles": "ALL", "catalog": "ANALYTICS", "schema": "STAGING"}, + "changed": ["principal", "role", "schema"] + }, + { + "name": "a role that was dropped rather than swapped", + "before": {"backend": "snowflake", "principal": "ANALYST", "role": "REPORTER", "secondary_roles": "ALL", "catalog": "ANALYTICS", "schema": "PUBLIC"}, + "after": {"backend": "snowflake", "principal": "ANALYST", "role": null, "secondary_roles": "ALL", "catalog": "ANALYTICS", "schema": "PUBLIC"}, + "changed": ["role"] + } + ] +} diff --git a/pkg-r/tests/testthat/helper-catalog-rows.R b/pkg-r/tests/testthat/helper-catalog-rows.R index d3292833..d73df2e8 100644 --- a/pkg-r/tests/testthat/helper-catalog-rows.R +++ b/pkg-r/tests/testthat/helper-catalog-rows.R @@ -54,3 +54,47 @@ catalog_rows_expect_columns <- function(columns, expected, info) { } } } + +# A session snapshot from the shared fixture's field-by-field spelling. +catalog_session_fixture_snapshot <- function(fields) { + list( + backend = fields$backend, + principal = fields$principal, + role = fields$role, + secondary_roles = fields$secondary_roles, + namespace = list(catalog = fields$catalog, schema = fields$schema) + ) +} + +# The access-precedence fixture's relations, keyed by label. +catalog_precedence_script <- function(relations) { + stats::setNames(relations, vapply(relations, `[[`, character(1), "label")) +} + +catalog_precedence_probe <- function(state) { + if (identical(state, "queryable")) { + return(list(state = "queryable", error = NULL)) + } + list(state = state, error = simpleError(state)) +} + +# Every refusal here is a cli condition, so the missing-relation outcome is +# told apart by what it says rather than by a class it shares with the rest. +catalog_precedence_expectation <- function(outcome) { + switch( + outcome, + missing = list(class = "rlang_error", regexp = "not on the connection"), + authorization = list(class = "commons_catalog_authorization_error"), + transient = list(class = "commons_catalog_transient_error"), + list(class = "commons_catalog_access_error") + ) +} + +# The relations a run must have probed: every one the listing reported. +catalog_precedence_probed <- function(script) { + names(script)[vapply( + script, + function(item) identical(item$discovered, "true"), + logical(1) + )] +} diff --git a/pkg-r/tests/testthat/test-catalog-security.R b/pkg-r/tests/testthat/test-catalog-security.R index 8df42beb..3c2d9e3c 100644 --- a/pkg-r/tests/testthat/test-catalog-security.R +++ b/pkg-r/tests/testthat/test-catalog-security.R @@ -73,6 +73,11 @@ test_that("catalog operations reject changed sessions", { catalog_search(source, "sales"), class = "commons_catalog_session_changed" ) + # The refusal names what moved rather than every field it compares. + expect_error( + source_query(source, "SELECT * FROM sales"), + regexp = "active role" + ) }) test_that("transient access failures remain retryable", { @@ -122,23 +127,68 @@ test_that("authorization failures are cached per relation", { expect_equal(calls, 1L) }) -test_that("warehouse access errors are classified conservatively", { - authorization <- structure( - list(message = "hidden", call = NULL, sqlstate = "42501"), - class = c("error", "condition") +test_that("warehouse access errors match the shared contract", { + cases <- shared_fixture("catalog-access-errors")$cases + expect_gt(length(cases), 0L) + + for (case in cases) { + err <- structure( + list(message = case$message, call = NULL, sqlstate = case$sqlstate), + class = c("error", "condition") + ) + expect_equal(catalog_access_error_kind(err), case$kind, info = case$name) + } +}) + +test_that("a databricks refusal never names a role", { + before <- list( + backend = "databricks", + principal = "analyst@example.com", + namespace = list(catalog = "main", schema = "default") ) + local_mocked_bindings( + catalog_session_snapshot = function(...) { + list( + backend = "databricks", + principal = "other@example.com", + namespace = list(catalog = "main", schema = "default") + ) + } + ) + + err <- expect_error( + catalog_check_session_snapshot(DBI::ANSI(), before), + class = "commons_catalog_session_changed" + ) + expect_match(conditionMessage(err), "principal changed") + expect_no_match(conditionMessage(err), "role") +}) - expect_equal(catalog_access_error_kind(authorization), "authorization") +test_that("changed session fields match the shared contract", { + cases <- shared_fixture("catalog-session-changed")$cases + expect_gt(length(cases), 0L) + + for (case in cases) { + expect_equal( + catalog_session_changed_fields( + catalog_session_fixture_snapshot(case$after), + catalog_session_fixture_snapshot(case$before) + ), + unlist(case$changed), + info = case$name + ) + } +}) + +test_that("an NA sqlstate is read as no sqlstate", { + # R-only: the fixture carries an empty string for a driver that reported + # no sqlstate, and only a DBI driver can hand back NA_character_ instead. missing_sqlstate <- structure( list(message = "bad syntax", call = NULL, sqlstate = NA_character_), class = c("error", "condition") ) + expect_equal(catalog_access_error_kind(missing_sqlstate), "unknown") - expect_equal( - catalog_access_error_kind(simpleError("warehouse is temporarily unavailable")), - "transient" - ) - expect_equal(catalog_access_error_kind(simpleError("bad syntax")), "unknown") }) test_that("catalog SQL probes bind typed nulls", { @@ -344,6 +394,55 @@ test_that("exact missing warehouse relations retain their diagnostic", { ) }) +test_that("access precedence matches the shared contract", { + cases <- shared_fixture("catalog-access-precedence")$cases + expect_gt(length(cases), 0L) + + for (case in cases) { + script <- catalog_precedence_script(case$relations) + registry <- list( + labels = names(script), + ids = lapply(names(script), function(label) DBI::Id(table = label)) + ) + relations <- lapply(script, function(item) { + list(id = DBI::Id(table = item$label), discovered = item$discovered == "true") + }) + probed <- character() + local_mocked_bindings( + catalog_probe_relation = function(con, id) { + label <- id@name[["table"]] + probed <<- c(probed, label) + catalog_precedence_probe(script[[label]]$probe) + }, + catalog_relation_exists = function(con, id) { + answer <- script[[id@name[["table"]]]]$exists %||% "unknown" + if (identical(answer, "unknown")) NULL else identical(answer, "true") + } + ) + + expected <- case$expected + if (identical(expected$outcome, "ok")) { + expect_no_error( + catalog_require_queryable_relations(DBI::ANSI(), registry, relations) + ) + expect_equal(probed, catalog_precedence_probed(script), info = case$name) + next + } + expectation <- catalog_precedence_expectation(expected$outcome) + err <- expect_error( + catalog_require_queryable_relations(DBI::ANSI(), registry, relations), + class = expectation$class + ) + if (!is.null(expectation$regexp)) { + expect_match(conditionMessage(err), expectation$regexp, info = case$name) + } + for (label in unlist(expected$labels)) { + expect_match(conditionMessage(err), label, fixed = TRUE, info = case$name) + } + expect_equal(probed, catalog_precedence_probed(script), info = case$name) + } +}) + test_that("discovered relations may have an unknown kind", { registry <- list( labels = "hive_metastore.default.sales", diff --git a/pkg-r/vignettes/governance.Rmd b/pkg-r/vignettes/governance.Rmd index 7f4a5549..75c66bf2 100644 --- a/pkg-r/vignettes/governance.Rmd +++ b/pkg-r/vignettes/governance.Rmd @@ -38,7 +38,7 @@ These checks provide defense in depth, but they are not a SQL parser or a databa The `tables` argument to `data_source()` controls which tables commons describes to the model. It is not an authorization boundary: SQL written by the agent can query any object available to the connection. -On Posit Connect, [viewer OAuth integrations](https://docs.posit.co/connect/admin/access-controls/) can give an interactive application the current viewer's Snowflake or Databricks credentials. If the application creates its connection from those credentials, the warehouse continues to enforce that viewer's existing access policies, including row- and column-level security. commons snapshots the connection's principal, active role, and namespace when it creates a Snowflake or Databricks data source, and rejects subsequent operations if that identity changes. +On Posit Connect, [viewer OAuth integrations](https://docs.posit.co/connect/admin/access-controls/) can give an interactive application the current viewer's Snowflake or Databricks credentials. If the application creates its connection from those credentials, the warehouse continues to enforce that viewer's existing access policies, including row- and column-level security. commons snapshots the connection's principal and namespace when it creates a Snowflake or Databricks data source, and its active and secondary roles as well on Snowflake, and rejects subsequent operations if that identity changes. Viewer credentials are not automatic: commons uses the DBI connection supplied by the application. When using viewer credentials, create the connection and the commons agent inside the Shiny server function so that each session has the correct database identity. diff --git a/tests/shared/README.md b/tests/shared/README.md index d735f1b4..272c44f0 100644 --- a/tests/shared/README.md +++ b/tests/shared/README.md @@ -36,6 +36,10 @@ The Python suite reads this directory directly. The R suite cannot. `testthat` n - **Definition expansion and rendering.** `definition-rendering.json` pins what happens to a governed definition after the compiler is done with it: which `{{token}}` queries expand and to what, the one-line gist shown at first touch and in retrieval, and the kind index under a character cap. It carries a bank of export records that each package hydrates into its own shape. A refused query pins the refusal and a reason slug rather than the message, because the wording belongs to each language. - **Catalog rows.** `catalog-rows.json` pins how a warehouse listing becomes relations and columns: which rows are relations at all, what kind each is, which comments count as prose, and where a `DESCRIBE` reply stops being columns. The rows are what Snowflake's `SHOW OBJECTS` and `DESC TABLE`, and Databricks' `system.information_schema.tables` and `DESCRIBE TABLE`, actually return. Running those queries is each language's own business; agreeing on their replies is not. Hand-maintained, since no binary generates it, and booleans travel as strings so the file reads the same from both JSON readers. +- **Catalog access errors.** `catalog-access-errors.json` pins how a failed access probe is read: a SQLSTATE and a message become `authorization`, `transient`, or `unknown`, which decides what the user is told and whether the answer is cached per relation. Hand-maintained, since the cases are the failures real drivers report rather than anything a binary emits. An absent SQLSTATE travels as an empty string, because JSON has no way to spell R's `NA_character_`. + +- **Session identity and access precedence.** `catalog-session-changed.json` pins which parts of a warehouse session a refusal reports as changed, so neither package tells a Databricks user that a role moved on a backend that has none. `catalog-access-precedence.json` pins what construction reports when several named relations fail at once: every relation the listing reported is probed before anything is raised, and a name the warehouse never listed is reported ahead of a refusal. Both are hand-maintained, and both carry field names rather than message text, since the wording belongs to each language. + - **Trace file naming.** `trace(-[0-9]+)?\.jsonl`, one OTLP envelope per line. ## Conventions diff --git a/tests/shared/catalog-access-errors.json b/tests/shared/catalog-access-errors.json new file mode 100644 index 00000000..bce22ef1 --- /dev/null +++ b/tests/shared/catalog-access-errors.json @@ -0,0 +1,101 @@ +{ + "description": "How a failed catalog access probe is classified from the driver's error. The classification decides which error a user sees and whether the failure is cached: an authorization failure is stable and is remembered per relation, a transient one is retried on the next touch, and an unknown one is neither. Drivers differ in what they raise, so each case gives only the two things every driver carries, a SQLSTATE (empty when the driver reported none) and a message.", + "cases": [ + { + "name": "insufficient privilege sqlstate", + "sqlstate": "42501", + "message": "hidden", + "kind": "authorization" + }, + { + "name": "invalid authorization sqlstate class", + "sqlstate": "28000", + "message": "hidden", + "kind": "authorization" + }, + { + "name": "snowflake access control message", + "sqlstate": "", + "message": "SQL access control error: Insufficient privileges to operate on table 'SALES'", + "kind": "authorization" + }, + { + "name": "databricks permission message", + "sqlstate": "", + "message": "PERMISSION_DENIED: User does not have SELECT on Table 'main.default.sales'", + "kind": "authorization" + }, + { + "name": "plain permission denied message", + "sqlstate": "", + "message": "Permission denied on relation sales", + "kind": "authorization" + }, + { + "name": "a network policy refusal is authorization, not a network fault", + "sqlstate": "", + "message": "Incoming request with IP 10.0.0.1 is not allowed to access Snowflake", + "kind": "authorization" + }, + { + "name": "connection exception sqlstate class", + "sqlstate": "08006", + "message": "hidden", + "kind": "transient" + }, + { + "name": "timeout sqlstate", + "sqlstate": "HYT00", + "message": "hidden", + "kind": "transient" + }, + { + "name": "serialization failure sqlstate", + "sqlstate": "40001", + "message": "hidden", + "kind": "transient" + }, + { + "name": "admin shutdown sqlstate", + "sqlstate": "57P01", + "message": "hidden", + "kind": "transient" + }, + { + "name": "warehouse starting message", + "sqlstate": "", + "message": "Warehouse 'COMPUTE_WH' is starting", + "kind": "transient" + }, + { + "name": "a temporarily unavailable warehouse", + "sqlstate": "", + "message": "Warehouse is temporarily unavailable", + "kind": "transient" + }, + { + "name": "rate limited message", + "sqlstate": "", + "message": "HTTP 429 too many requests", + "kind": "transient" + }, + { + "name": "lowercase sqlstate is still matched", + "sqlstate": "hyt00", + "message": "hidden", + "kind": "transient" + }, + { + "name": "syntax error with a sqlstate", + "sqlstate": "42601", + "message": "syntax error at or near SELCT", + "kind": "unknown" + }, + { + "name": "unrecognized failure without a sqlstate", + "sqlstate": "", + "message": "column BOGUS not found", + "kind": "unknown" + } + ] +} diff --git a/tests/shared/catalog-access-precedence.json b/tests/shared/catalog-access-precedence.json new file mode 100644 index 00000000..33fbbd81 --- /dev/null +++ b/tests/shared/catalog-access-precedence.json @@ -0,0 +1,75 @@ +{ + "description": "What a source construction reports when several explicitly named relations fail at once. Every relation the listing did report is probed before anything is raised, so a name the caller got wrong and a relation they cannot read are found in one pass rather than one round trip each. A name the warehouse never listed is reported ahead of a refusal: it is the more actionable problem, and an access error about a table that is not there reads as a permissions problem the caller does not have. Booleans travel as strings so the file reads the same from both JSON readers. `discovered` is whether the listing reported the relation, `probe` is what the zero-row query did, and `exists` is what the backend answers about an unexplained failure.", + "cases": [ + { + "name": "every relation is readable", + "relations": [ + {"label": "A", "discovered": "true", "probe": "queryable"}, + {"label": "B", "discovered": "true", "probe": "queryable"} + ], + "expected": {"outcome": "ok", "labels": []} + }, + { + "name": "a name the listing never reported", + "relations": [ + {"label": "A", "discovered": "false", "probe": "queryable"}, + {"label": "B", "discovered": "true", "probe": "queryable"} + ], + "expected": {"outcome": "missing", "labels": ["A"]} + }, + { + "name": "a missing name is reported ahead of a refusal", + "relations": [ + {"label": "A", "discovered": "false", "probe": "queryable"}, + {"label": "B", "discovered": "true", "probe": "authorization"} + ], + "expected": {"outcome": "missing", "labels": ["A"]} + }, + { + "name": "a relation the probe proves absent is missing too", + "relations": [ + {"label": "A", "discovered": "true", "probe": "unknown", "exists": "false"}, + {"label": "B", "discovered": "true", "probe": "authorization"} + ], + "expected": {"outcome": "missing", "labels": ["A"]} + }, + { + "name": "every missing name is reported at once", + "relations": [ + {"label": "A", "discovered": "false", "probe": "queryable"}, + {"label": "B", "discovered": "true", "probe": "unknown", "exists": "false"} + ], + "expected": {"outcome": "missing", "labels": ["A", "B"]} + }, + { + "name": "the first refusal is the one raised", + "relations": [ + {"label": "A", "discovered": "true", "probe": "authorization"}, + {"label": "B", "discovered": "true", "probe": "transient"} + ], + "expected": {"outcome": "authorization", "labels": ["A"]} + }, + { + "name": "a transient refusal on its own", + "relations": [ + {"label": "A", "discovered": "true", "probe": "queryable"}, + {"label": "B", "discovered": "true", "probe": "transient"} + ], + "expected": {"outcome": "transient", "labels": ["B"]} + }, + { + "name": "an unexplained failure the backend cannot explain away", + "relations": [ + {"label": "A", "discovered": "true", "probe": "unknown", "exists": "unknown"} + ], + "expected": {"outcome": "access", "labels": ["A"]} + }, + { + "name": "an unexplained failure on a relation that is there", + "relations": [ + {"label": "A", "discovered": "true", "probe": "unknown", "exists": "true"} + ], + "expected": {"outcome": "access", "labels": ["A"]} + } + ] +} diff --git a/tests/shared/catalog-session-changed.json b/tests/shared/catalog-session-changed.json new file mode 100644 index 00000000..957b0223 --- /dev/null +++ b/tests/shared/catalog-session-changed.json @@ -0,0 +1,41 @@ +{ + "description": "Which parts of a warehouse session identity a refusal reports as changed. The comparison runs before every catalog operation, and the answer decides what the user is told to look at, so it is pinned rather than described twice. A field is absent (null) when the backend has none: Databricks reports no role, so a refusal there must never name one. The names here are the snapshot's own fields; each implementation words them for its own message.", + "cases": [ + { + "name": "a snowflake role change", + "before": {"backend": "snowflake", "principal": "ANALYST", "role": "REPORTER", "secondary_roles": "ALL", "catalog": "ANALYTICS", "schema": "PUBLIC"}, + "after": {"backend": "snowflake", "principal": "ANALYST", "role": "ADMIN", "secondary_roles": "ALL", "catalog": "ANALYTICS", "schema": "PUBLIC"}, + "changed": ["role"] + }, + { + "name": "a snowflake secondary role change", + "before": {"backend": "snowflake", "principal": "ANALYST", "role": "REPORTER", "secondary_roles": "ALL", "catalog": "ANALYTICS", "schema": "PUBLIC"}, + "after": {"backend": "snowflake", "principal": "ANALYST", "role": "REPORTER", "secondary_roles": "NONE", "catalog": "ANALYTICS", "schema": "PUBLIC"}, + "changed": ["secondary_roles"] + }, + { + "name": "a databricks principal change names no role", + "before": {"backend": "databricks", "principal": "analyst@example.com", "role": null, "secondary_roles": null, "catalog": "main", "schema": "default"}, + "after": {"backend": "databricks", "principal": "other@example.com", "role": null, "secondary_roles": null, "catalog": "main", "schema": "default"}, + "changed": ["principal"] + }, + { + "name": "a databricks namespace change names both parts", + "before": {"backend": "databricks", "principal": "analyst@example.com", "role": null, "secondary_roles": null, "catalog": "main", "schema": "default"}, + "after": {"backend": "databricks", "principal": "analyst@example.com", "role": null, "secondary_roles": null, "catalog": "sandbox", "schema": "scratch"}, + "changed": ["catalog", "schema"] + }, + { + "name": "several fields at once, in snapshot order", + "before": {"backend": "snowflake", "principal": "ANALYST", "role": "REPORTER", "secondary_roles": "ALL", "catalog": "ANALYTICS", "schema": "PUBLIC"}, + "after": {"backend": "snowflake", "principal": "ADMIN_USER", "role": "ADMIN", "secondary_roles": "ALL", "catalog": "ANALYTICS", "schema": "STAGING"}, + "changed": ["principal", "role", "schema"] + }, + { + "name": "a role that was dropped rather than swapped", + "before": {"backend": "snowflake", "principal": "ANALYST", "role": "REPORTER", "secondary_roles": "ALL", "catalog": "ANALYTICS", "schema": "PUBLIC"}, + "after": {"backend": "snowflake", "principal": "ANALYST", "role": null, "secondary_roles": "ALL", "catalog": "ANALYTICS", "schema": "PUBLIC"}, + "changed": ["role"] + } + ] +}