diff --git a/datajunction-server/datajunction_server/internal/nodes.py b/datajunction-server/datajunction_server/internal/nodes.py index 7674faf3b..0a4d4e5ab 100644 --- a/datajunction-server/datajunction_server/internal/nodes.py +++ b/datajunction-server/datajunction_server/internal/nodes.py @@ -60,7 +60,9 @@ ) from datajunction_server.internal.access.authorization import ( AccessChecker, + AccessDenialMode, ) +from datajunction_server.internal.access.authorization.context import AuthContext from datajunction_server.internal.caching.interface import Cache from datajunction_server.internal.history import ActivityType, EntityType from datajunction_server.internal.materializations import ( @@ -73,6 +75,11 @@ validate_node_data, validate_node_data_v2, ) +from datajunction_server.models.access import ( + Resource, + ResourceAction, + ResourceRequest, +) from datajunction_server.models.attribute import ( AttributeTypeIdentifier, ColumnAttributes, @@ -160,6 +167,7 @@ async def create_a_source_node( access_checker=access_checker, background_tasks=background_tasks, save_history=save_history, + access_target=_create_access_target(data), ): return recreated_node @@ -251,6 +259,8 @@ async def create_a_node( await raise_if_node_exists(session, data.name) + create_access_target = _create_access_target(data) + # if the node previously existed and now is inactive if recreated_node := await create_node_from_inactive( new_node_type=node_type, @@ -263,6 +273,7 @@ async def create_a_node( access_checker=access_checker, save_history=save_history, cache=cache, + access_target=create_access_target, ): return recreated_node # pragma: no cover @@ -311,11 +322,18 @@ async def create_a_node( # For metric nodes, derive the referenced frozen measures and save them if node.type == NodeType.METRIC: - background_tasks.add_task(derive_frozen_measures, node_revision.id) + background_tasks.add_task( + derive_frozen_measures, + node_revision.id, + current_user=current_user, + access_target=create_access_target, + ) background_tasks.add_task( save_column_level_lineage, node_revision_id=node_revision.id, + current_user=current_user, + access_target=create_access_target, ) return await Node.get_by_name( # type: ignore @@ -349,6 +367,7 @@ async def create_a_cube( background_tasks=background_tasks, access_checker=access_checker, save_history=save_history, + access_target=_create_access_target(data), ): return recreated_node # pragma: no cover @@ -711,7 +730,62 @@ async def create_cube_node_revision( return node_revision -async def derive_frozen_measures(node_revision_id: int) -> list[FrozenMeasure]: +def _create_access_target( + data: CreateSourceNode | CreateNode | CreateCubeNode, +) -> Resource: + """ + The resource a create endpoint governs: the node's target namespace. + + Background work scheduled by a create re-authorizes this rather than the node + (see ``_background_write_allowed``), including on the re-creation path, which + routes through the update helper. + """ + return Resource.from_namespace(data.namespace or get_namespace_from_name(data.name)) + + +async def _background_write_allowed( + session: AsyncSession, + access_target: Resource, + current_user: User, + action_description: str, +) -> bool: + """ + Whether ``current_user`` may WRITE ``access_target``. + + Background tasks mutate after the response has returned, outside the request's + AccessChecker, so they re-authorize here instead of trusting the scheduling + caller. The target is supplied by that caller rather than derived here, + because the endpoints do not all govern the same resource: node creation + authorizes the target namespace while updates authorize the node. Deriving a + node request here would deny the create path, whose grant may be a namespace + scope that no node pattern matches. + + Denials are logged and skip the mutation rather than raising, since the + callers swallow exceptions and an exception would be indistinguishable from + a failure. + """ + access_checker = AccessChecker(await AuthContext.from_user(session, current_user)) + access_checker.add_request( + ResourceRequest(verb=ResourceAction.WRITE, access_object=access_target), + ) + decisions = await access_checker.check(on_denied=AccessDenialMode.RETURN) + if any(not decision.approved for decision in decisions): + _logger.warning( + "Skipping %s for %s %s: %s lacks WRITE", + action_description, + access_target.resource_type.value, + access_target.name, + current_user.username, + ) + return False + return True + + +async def derive_frozen_measures( + node_revision_id: int, + current_user: User, + access_target: Resource, +) -> list[FrozenMeasure]: """ Find or create frozen measures for a metric. @@ -723,9 +797,20 @@ async def derive_frozen_measures(node_revision_id: int) -> list[FrozenMeasure]: the derivation runs asynchronously. Opens its own session and commits on completion. The deployment path uses ``derive_frozen_measures_bulk`` instead to batch DB work across all metrics in a single transaction. + + Runs after the response, so it authorizes ``current_user`` for WRITE on + ``access_target`` -- the resource the scheduling endpoint governed -- rather + than trusting that caller to have checked (#2234 step 0). """ try: async with session_context() as session: + if not await _background_write_allowed( + session, + access_target, + current_user, + "deriving frozen measures", + ): + return [] result = await _derive_frozen_measures_impl(node_revision_id, session) await session.commit() return result @@ -1258,11 +1343,17 @@ async def update_node_with_query( access_checker: AccessChecker, save_history: Callable, cache: Cache | None = None, + access_target: Resource | None = None, ) -> Node: """ Update the named node with the changes defined in the UpdateNode object. Propagate these changes to all of the node's downstream children. + ``access_target`` is the resource the calling endpoint authorized, used by the + background work scheduled below. It defaults to the node, which is what the + update endpoint governs; re-creating an inactive node routes here from the + create endpoint, which governs the target namespace instead. + Note: this function works for both source nodes and nodes with query (transforms, dimensions, metrics). We should update it to separate out the logic for source nodes """ @@ -1372,6 +1463,8 @@ async def update_node_with_query( background_tasks.add_task( save_column_level_lineage, node_revision_id=new_revision.id, + current_user=current_user, + access_target=access_target or Resource.from_node(new_revision), ) # TODO: Do not save this until: # 1. We get to the bottom of why there are query building discrepancies @@ -1866,6 +1959,16 @@ async def _propagate_update_downstream( - altered column names: may invalidate downstream nodes - altered column types: may invalidate downstream nodes - new columns: won't affect downstream nodes + + Authorization: this revalidates nodes the updater may not hold WRITE on, and + that is deliberate (#2234 step 0). A node is only downstream because its own + owner pointed it at this upstream, so the blast radius is opt-in rather than + caller-controlled, and revalidation only makes stored status reflect reality. + Requiring WRITE here would either block owners from updating their own nodes + whenever another team depends on them, or skip the denied ones and leave the + graph asserting VALID for nodes that are now broken -- silently, since the + caller above swallows exceptions. Pinned by + tests/internal/nodes/background_authz_test.py. """ _logger.info("Propagating update of node %s downstream", node.name) downstreams = await get_downstream_nodes( @@ -2008,10 +2111,15 @@ async def create_node_from_inactive( background_tasks: BackgroundTasks = None, access_checker: AccessChecker | None = None, cache: Cache | None = None, + access_target: Resource | None = None, ) -> Node | None: """ If the node existed and is inactive the re-creation takes different steps than creating it from scratch. + + ``access_target`` is forwarded to the update path so background work + re-authorizes what the *create* endpoint governed (the namespace), not the + node that path would otherwise assume. """ previous_inactive_node = await Node.get_by_name( session, @@ -2059,6 +2167,7 @@ async def create_node_from_inactive( access_checker=access_checker, # type: ignore save_history=save_history, cache=cache, + access_target=access_target, ) else: await update_cube_node( @@ -2386,12 +2495,27 @@ async def create_new_revision_from_existing( return new_revision -async def save_column_level_lineage(node_revision_id: int): +async def save_column_level_lineage( + node_revision_id: int, + current_user: User, + access_target: Resource, +): """ Saves the column-level lineage for a node revision + + Runs after the response, so it authorizes ``current_user`` for WRITE on + ``access_target`` -- the resource the scheduling endpoint governed -- rather + than trusting that caller to have checked (#2234 step 0). """ try: async with session_context() as session: + if not await _background_write_allowed( + session, + access_target, + current_user, + "saving column-level lineage", + ): + return statement = ( select(NodeRevision) .where(NodeRevision.id == node_revision_id) @@ -3815,7 +3939,13 @@ async def revalidate_node( # For metric nodes, derive frozen measures (ensures they exist even for # metrics created via deployment or updated after initial creation) if current_node_revision.type == NodeType.METRIC and background_tasks: - background_tasks.add_task(derive_frozen_measures, node.current.id) # type: ignore + background_tasks.add_task( + derive_frozen_measures, + node.current.id, # type: ignore + current_user=current_user, + # Revalidation is governed on the node itself. + access_target=Resource.from_node(node), # type: ignore + ) return node_validator diff --git a/datajunction-server/scripts/backfill_derived_expression.py b/datajunction-server/scripts/backfill_derived_expression.py index a21f5f88a..fcdeb6d61 100644 --- a/datajunction-server/scripts/backfill_derived_expression.py +++ b/datajunction-server/scripts/backfill_derived_expression.py @@ -28,7 +28,7 @@ from sqlalchemy import select from datajunction_server.database.node import Node, NodeRevision -from datajunction_server.internal.nodes import derive_frozen_measures +from datajunction_server.internal.nodes import derive_frozen_measures_bulk from datajunction_server.models.node_type import NodeType from datajunction_server.utils import session_context @@ -67,9 +67,12 @@ async def backfill(batch_size: int = 100, dry_run: bool = False) -> None: batch = targets[i : i + batch_size] for nr_id, name in batch: try: - # derive_frozen_measures opens its own session_context and - # commits inside; safe to call in a loop. - await derive_frozen_measures(nr_id) + # Operator tool with no request user, so it uses the + # system-facing bulk API; the per-revision entry point + # authorizes a user for WRITE on the metric. + async with session_context() as session: + await derive_frozen_measures_bulk(session, [nr_id]) + await session.commit() done += 1 except Exception as exc: failed += 1 diff --git a/datajunction-server/scripts/migrate-measures.py b/datajunction-server/scripts/migrate-measures.py deleted file mode 100644 index b80ff3fd4..000000000 --- a/datajunction-server/scripts/migrate-measures.py +++ /dev/null @@ -1,83 +0,0 @@ -import asyncio - -import sqlalchemy as sa -from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine -from sqlalchemy.orm import joinedload, selectinload, sessionmaker - -from datajunction_server.database.node import Node, NodeRevision, NodeType -from datajunction_server.internal.nodes import derive_frozen_measures -from datajunction_server.utils import get_settings - -settings = get_settings() - - -async def backfill_measures(): - engine = create_async_engine(settings.writer_db.uri) - async_session = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession) - - async with async_session() as session: - async with session.begin(): - # Get all latest metric node revisions - metric_revisions = [ - metric - for metric in ( - await session.execute( - sa.select(NodeRevision) - .join( - Node, - (NodeRevision.node_id == Node.id) - & (NodeRevision.version == Node.current_version), - ) - .where( - NodeRevision.type == NodeType.METRIC, - sa.not_(NodeRevision.name.like("system.temp%")), - ) - .options( - selectinload(NodeRevision.parents).options( - joinedload(Node.current), - ), - selectinload(NodeRevision.frozen_measures), - ), - ) - ) - .unique() - .scalars() - .all() - if not metric.name.startswith("system.temp") - ] - print(f"Found {len(metric_revisions)} metric revisions") - for idx, revision in enumerate(metric_revisions): - try: - print( - f"[{idx + 1}/{len(metric_revisions)}] Processing metric revision {revision.name}@{revision.version}", - ) - derived_measures = [ - m for m in await derive_frozen_measures(session, revision) if m - ] - print( - f"[{idx + 1}/{len(metric_revisions)}] Derived the following frozen measures: {[m.name for m in derived_measures]}", - ) - - for frozen_measure in derived_measures: - session.add(frozen_measure) - with session.no_autoflush: - if frozen_measure not in revision.frozen_measures: - revision.frozen_measures.append(frozen_measure) - print( - f"[{idx + 1}/{len(metric_revisions)}] Added frozen measures: {[m.name for m in derived_measures]}", - ) - session.add(revision) - print("---") - except Exception as exc: - print( - "[{idx+1}/{len(metric_revisions)}] Failed to process", - derived_measures, - ) - print(exc) - raise exc - - await session.commit() - - -if __name__ == "__main__": - asyncio.run(backfill_measures()) diff --git a/datajunction-server/tests/internal/nodes/background_authz_test.py b/datajunction-server/tests/internal/nodes/background_authz_test.py new file mode 100644 index 000000000..2b0dd0177 --- /dev/null +++ b/datajunction-server/tests/internal/nodes/background_authz_test.py @@ -0,0 +1,346 @@ +""" +Authorization on the background mutation path (#2234 step 0). + +These tasks run after the response, in their own session, outside the request's +AccessChecker. Two contracts follow: + +* ``derive_frozen_measures`` and ``save_column_level_lineage`` re-authorize the + resource their endpoint governed -- the namespace for creation, the node for + updates. Authorizing the node instead would deny work scheduled by a create + that a namespace-scoped grant allowed, since a namespace scope reaches a node + request only by pattern matching. + +* ``propagate_update_downstream`` deliberately does not, since a node is only + downstream once its own owner points it at the upstream. Gating it would block + owners whose nodes have dependents, or leave the graph asserting VALID for + broken nodes -- silently, as propagation swallows exceptions. +""" + +from functools import partial + +import pytest +from httpx import AsyncClient +from sqlalchemy import select + +from datajunction_server.database.node import NodeRevision +from datajunction_server.database.rbac import Role, RoleAssignment, RoleScope +from datajunction_server.internal.access.authorization import ( + AuthorizationService, + RBACAuthorizationService, +) +from datajunction_server.internal.nodes import ( + _background_write_allowed, + derive_frozen_measures, + save_column_level_lineage, +) +from datajunction_server.models import access +from datajunction_server.models.node import NodeStatus + +# Patch target: the name as imported into the validator module, where +# AccessChecker.check() looks the authorization service up. +VALIDATOR_AUTH_SERVICE = ( + "datajunction_server.internal.access.authorization." + "validator.get_authorization_service" +) +LINEAGE_TASK = "datajunction_server.internal.nodes.save_column_level_lineage" +DERIVE_TASK = "datajunction_server.internal.nodes.derive_frozen_measures" + +NAMESPACE = "bgauthz" +SOURCE = f"{NAMESPACE}.events" +TRANSFORM = f"{NAMESPACE}.users_per_country" +METRIC = f"{NAMESPACE}.total_users" +# Creation is governed on the target namespace, so that is what its background +# work must re-authorize. +CREATE_TARGET = access.Resource.from_namespace(NAMESPACE) + + +def deny(_request: access.ResourceRequest) -> bool: + return False + + +def allow(_request: access.ResourceRequest) -> bool: + return True + + +class RecordingAuthorizationService(AuthorizationService): + """ + Decides each request by ``approves``, recording requests and the principal so + tests can assert the *right* resource was authorized for the *right* user. + """ + + name = "test_recording_background" + + def __init__(self, approves=allow): + self.approves = approves + self.requests: list[access.ResourceRequest] = [] + self.usernames: list[str] = [] + + def authorize(self, auth_context, requests): + self.requests.extend(requests) + self.usernames.append(auth_context.username) + return [ + access.AccessDecision(request=request, approved=self.approves(request)) + for request in requests + ] + + +async def _post(client: AsyncClient, url: str, json=None) -> None: + response = await client.post(url, json=json) + assert response.status_code in (200, 201, 409), response.text + + +async def _create_metric(client: AsyncClient, name: str, description: str) -> None: + await _post( + client, + "/nodes/metric/", + { + "name": name, + "description": description, + "query": f"SELECT SUM(num_users) FROM {TRANSFORM}", + "mode": "published", + }, + ) + + +@pytest.fixture +async def metric_graph(client: AsyncClient) -> None: + """A source, a transform over it, and a metric over the transform.""" + await _post(client, "/catalogs/", {"name": "warehouse"}) + await _post(client, f"/namespaces/{NAMESPACE}/") + await _post( + client, + "/nodes/source/", + { + "name": SOURCE, + "columns": [ + {"name": "id", "type": "int"}, + {"name": "country", "type": "string"}, + ], + "mode": "published", + "catalog": "warehouse", + "schema_": "db", + "table": "events", + }, + ) + await _post( + client, + "/nodes/transform/", + { + "name": TRANSFORM, + "query": ( + f"SELECT country, COUNT(DISTINCT id) AS num_users " + f"FROM {SOURCE} GROUP BY 1" + ), + "mode": "published", + }, + ) + await _create_metric(client, METRIC, "Total users") + + +async def _revision(session, name: str) -> NodeRevision: + """Latest revision of ``name``, read fresh -- the tasks commit elsewhere.""" + session.expire_all() + query = select(NodeRevision).where(NodeRevision.name == name) + revision = (await session.execute(query.order_by(NodeRevision.id.desc()))).scalars() + return revision.first() + + +async def _assert_gated_on_write(mocker, task, username, ran) -> None: + """``task`` must ask for WRITE on the create target, and run only if granted.""" + for approves, expected in ((deny, False), (allow, True)): + recorder = RecordingAuthorizationService(approves) + mocker.patch(VALIDATOR_AUTH_SERVICE, lambda: recorder) + await task() + + assert recorder.usernames == [username] + assert [ + (request.verb, request.access_object) for request in recorder.requests + ] == [(access.ResourceAction.WRITE, CREATE_TARGET)] + assert await ran() is expected, ( + f"task ran={not expected} with WRITE {'denied' if expected else 'granted'}" + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("recreate", [False, True], ids=["create", "recreate"]) +async def test_create_schedules_background_work_against_the_namespace( + recreate, + client: AsyncClient, + metric_graph, + mocker, +): + """ + Creation hands its background tasks the namespace it authorized. + + Re-creating a deleted node routes through ``create_node_from_inactive`` into + the update path, whose default target is the node -- which an exact namespace + grant does not match, so lineage would be silently skipped. + """ + lineage = mocker.patch(LINEAGE_TASK) + derive = mocker.patch(DERIVE_TASK) + + if recreate: + assert (await client.delete(f"/nodes/{METRIC}/")).status_code == 200 + # Recreate with a change; an identical recreate makes no new revision + # and so schedules nothing. + await _create_metric(client, METRIC, "Total users, again") + else: + await _create_metric(client, f"{METRIC}_again", "Total users") + assert derive.call_args.kwargs["access_target"] == CREATE_TARGET + + assert lineage.call_args.kwargs["access_target"] == CREATE_TARGET + + +@pytest.mark.asyncio +async def test_exact_namespace_grant_authorizes_create_background_work( + session, + current_user, + mocker, +): + """ + An exact namespace grant -- enough to create the node -- must also satisfy the + background work that create schedules. Uses the real RBAC matcher, since the + resource type decides it: `bgauthz` does not match `bgauthz.total_users`. + """ + settings = mocker.patch( + "datajunction_server.internal.access.authorization.service.settings", + ) + settings.authorization_provider = "rbac" + settings.default_access_policy = "restrictive" + + role = Role(name="bgauthz-namespace-writer", created_by_id=current_user.id) + session.add(role) + await session.flush() + session.add_all( + [ + RoleScope( + role_id=role.id, + action=access.ResourceAction.WRITE, + scope_type=access.ResourceType.NAMESPACE, + scope_value=NAMESPACE, + ), + RoleAssignment( + principal_id=current_user.id, + role_id=role.id, + granted_by_id=current_user.id, + ), + ], + ) + await session.commit() + mocker.patch(VALIDATOR_AUTH_SERVICE, lambda: RBACAuthorizationService()) + + allowed = partial(_background_write_allowed, session, current_user=current_user) + assert await allowed(CREATE_TARGET, action_description="test") is True, ( + "the grant that allowed the create did not allow its background work" + ) + # The node target the create path must not use, denied by that same grant. + node_target = access.Resource(name=METRIC, resource_type=access.ResourceType.NODE) + assert await allowed(node_target, action_description="test") is False + + +@pytest.mark.asyncio +async def test_lineage_gated_on_the_create_target( + metric_graph, + session, + current_user, + mocker, +): + """Lineage is written only when the caller may write the create target.""" + revision = await _revision(session, METRIC) + revision.lineage = None + await session.commit() + + async def ran() -> bool: + return bool((await _revision(session, METRIC)).lineage) + + await _assert_gated_on_write( + mocker, + partial( + save_column_level_lineage, + node_revision_id=revision.id, + current_user=current_user, + access_target=CREATE_TARGET, + ), + current_user.username, + ran, + ) + + +@pytest.mark.asyncio +async def test_frozen_measures_gated_on_the_create_target( + metric_graph, + session, + current_user, + mocker, +): + """ + Derivation runs only when the caller may write the create target. Observes + whether the body was reached, since the return value is empty for several + unrelated reasons. + """ + revision = await _revision(session, METRIC) + derive = mocker.patch( + "datajunction_server.internal.nodes._derive_frozen_measures_impl", + return_value=[], + ) + + async def ran() -> bool: + return derive.called + + await _assert_gated_on_write( + mocker, + partial( + derive_frozen_measures, + node_revision_id=revision.id, + current_user=current_user, + access_target=CREATE_TARGET, + ), + current_user.username, + ran, + ) + + +@pytest.mark.asyncio +async def test_downstream_revalidation_not_gated_on_caller_write( + client: AsyncClient, + metric_graph, + session, + mocker, +): + """ + A caller who may write only the upstream still gets downstream revalidation. + Fails if a WRITE check is added there -- which would break cross-namespace + graphs. + """ + assert (await client.get(f"/nodes/{METRIC}/")).json()["status"] == NodeStatus.VALID + + # WRITE only on the upstream; reads stay allowed so the assertions below work. + mocker.patch( + VALIDATOR_AUTH_SERVICE, + lambda: RecordingAuthorizationService( + lambda request: ( + request.verb != access.ResourceAction.WRITE + or request.access_object.name == TRANSFORM + ), + ), + ) + + # Rename the column the downstream metric selects, invalidating it. + response = await client.patch( + f"/nodes/{TRANSFORM}/", + json={ + "query": ( + f"SELECT country, COUNT(DISTINCT id) AS user_count " + f"FROM {SOURCE} GROUP BY 1" + ), + }, + ) + assert response.status_code == 200, response.text + + session.expire_all() + assert (await client.get(f"/nodes/{METRIC}/")).json()[ + "status" + ] == NodeStatus.INVALID, ( + "downstream was not revalidated; propagation appears to be gated on the " + "caller's WRITE, which would break cross-namespace graphs" + ) diff --git a/datajunction-server/tests/internal/nodes/background_tasks_test.py b/datajunction-server/tests/internal/nodes/background_tasks_test.py index a986618ea..43231e0de 100644 --- a/datajunction-server/tests/internal/nodes/background_tasks_test.py +++ b/datajunction-server/tests/internal/nodes/background_tasks_test.py @@ -57,6 +57,10 @@ async def test_derive_frozen_measures_swallows_exceptions(caplog): mock_ctx.return_value.__aexit__ = AsyncMock(return_value=False) with ( + patch( + "datajunction_server.internal.nodes._background_write_allowed", + return_value=True, + ), patch( "datajunction_server.internal.nodes._derive_frozen_measures_impl", side_effect=RuntimeError("boom"), @@ -66,7 +70,11 @@ async def test_derive_frozen_measures_swallows_exceptions(caplog): logger="datajunction_server.internal.nodes", ), ): - result = await derive_frozen_measures(node_revision_id=99) + result = await derive_frozen_measures( + node_revision_id=99, + current_user=MagicMock(), + access_target=MagicMock(), + ) assert result == [] assert any("deriving frozen measures" in r.message.lower() for r in caplog.records) @@ -82,11 +90,21 @@ async def test_save_column_level_lineage_swallows_exceptions(caplog): mock_ctx.return_value.__aexit__ = AsyncMock(return_value=False) mock_session.execute.side_effect = RuntimeError("boom") - with caplog.at_level( - logging.ERROR, - logger="datajunction_server.internal.nodes", + with ( + patch( + "datajunction_server.internal.nodes._background_write_allowed", + return_value=True, + ), + caplog.at_level( + logging.ERROR, + logger="datajunction_server.internal.nodes", + ), ): - await save_column_level_lineage(node_revision_id=99) + await save_column_level_lineage( + node_revision_id=99, + current_user=MagicMock(), + access_target=MagicMock(), + ) # The exception must be folded into the message itself (not just exc_info), # so backends that retain only the formatted message stay diagnosable.