From 71e5d0410fac8a2a103ae732f1168e95435596b8 Mon Sep 17 00:00:00 2001 From: Rui Zhang <36744357+ruizhang0519@users.noreply.github.com> Date: Tue, 4 Aug 2026 15:20:04 +0000 Subject: [PATCH 1/6] Authorize node mutations on the background task path --- .../datajunction_server/internal/nodes.py | 91 ++++++++++++++++++- .../scripts/backfill_derived_expression.py | 11 ++- .../internal/nodes/background_tasks_test.py | 34 +++++-- 3 files changed, 120 insertions(+), 16 deletions(-) diff --git a/datajunction-server/datajunction_server/internal/nodes.py b/datajunction-server/datajunction_server/internal/nodes.py index 3cb8aa4e2b..ba3e49874a 100644 --- a/datajunction-server/datajunction_server/internal/nodes.py +++ b/datajunction-server/datajunction_server/internal/nodes.py @@ -16,8 +16,11 @@ 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.models.access import ResourceAction from datajunction_server.models.deployment import DeploymentResult from datajunction_server.models.query import QueryCreate from datajunction_server.api.helpers import ( @@ -298,11 +301,16 @@ 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, + ) background_tasks.add_task( save_column_level_lineage, node_revision_id=node_revision.id, + current_user=current_user, ) return await Node.get_by_name( # type: ignore @@ -713,7 +721,47 @@ async def create_cube_node_revision( return node_revision -async def derive_frozen_measures(node_revision_id: int) -> list[FrozenMeasure]: +async def _background_write_allowed( + session: AsyncSession, + node_revision_id: int, + current_user: User, + action_description: str, +) -> bool: + """ + Whether ``current_user`` may WRITE the node owning ``node_revision_id``. + + Background tasks mutate after the response has returned, outside the request's + AccessChecker, so they re-authorize here instead of trusting the scheduling + caller. Denials are logged and skip the mutation rather than raising, since + the callers swallow exceptions and an exception would be indistinguishable + from a failure. + """ + node_name = ( + await session.execute( + select(NodeRevision.name).where(NodeRevision.id == node_revision_id), + ) + ).scalar_one_or_none() + if node_name is None: # pragma: no cover + return False + + access_checker = AccessChecker(await AuthContext.from_user(session, current_user)) + access_checker.add_request_by_node_name(node_name, ResourceAction.WRITE) + decisions = await access_checker.check(on_denied=AccessDenialMode.RETURN) + if any(not decision.approved for decision in decisions): + _logger.warning( + "Skipping %s for node %s: %s lacks WRITE", + action_description, + node_name, + current_user.username, + ) + return False + return True + + +async def derive_frozen_measures( + node_revision_id: int, + current_user: User, +) -> list[FrozenMeasure]: """ Find or create frozen measures for a metric. @@ -725,9 +773,19 @@ 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 the + metric itself rather than trusting the scheduling caller (#2234 step 0). """ try: async with session_context() as session: + if not await _background_write_allowed( + session, + node_revision_id, + current_user, + "deriving frozen measures", + ): + return [] result = await _derive_frozen_measures_impl(node_revision_id, session) await session.commit() return result @@ -1374,6 +1432,7 @@ async def update_node_with_query( background_tasks.add_task( save_column_level_lineage, node_revision_id=new_revision.id, + current_user=current_user, ) # TODO: Do not save this until: # 1. We get to the bottom of why there are query building discrepancies @@ -1724,6 +1783,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/api/background_propagation_test.py. """ _logger.info("Propagating update of node %s downstream", node.name) downstreams = await get_downstream_nodes( @@ -2244,12 +2313,22 @@ 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): """ Saves the column-level lineage for a node revision + + Runs after the response, so it authorizes ``current_user`` for WRITE on the + node itself rather than trusting the scheduling caller (#2234 step 0). """ try: async with session_context() as session: + if not await _background_write_allowed( + session, + node_revision_id, + current_user, + "saving column-level lineage", + ): + return statement = ( select(NodeRevision) .where(NodeRevision.id == node_revision_id) @@ -3673,7 +3752,11 @@ 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, + ) return node_validator diff --git a/datajunction-server/scripts/backfill_derived_expression.py b/datajunction-server/scripts/backfill_derived_expression.py index 9752a5ef59..05916bcd0c 100644 --- a/datajunction-server/scripts/backfill_derived_expression.py +++ b/datajunction-server/scripts/backfill_derived_expression.py @@ -27,7 +27,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 @@ -66,9 +66,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/tests/internal/nodes/background_tasks_test.py b/datajunction-server/tests/internal/nodes/background_tasks_test.py index 476bf6817a..c3ccd2717b 100644 --- a/datajunction-server/tests/internal/nodes/background_tasks_test.py +++ b/datajunction-server/tests/internal/nodes/background_tasks_test.py @@ -54,15 +54,24 @@ async def test_derive_frozen_measures_swallows_exceptions(caplog): mock_ctx.return_value.__aenter__ = AsyncMock(return_value=mock_session) mock_ctx.return_value.__aexit__ = AsyncMock(return_value=False) - with patch( - "datajunction_server.internal.nodes._derive_frozen_measures_impl", - side_effect=RuntimeError("boom"), + 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"), + ), ): with caplog.at_level( logging.ERROR, 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(), + ) assert result == [] assert any("deriving frozen measures" in r.message.lower() for r in caplog.records) @@ -78,11 +87,20 @@ 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(), + ) # The exception must be folded into the message itself (not just exc_info), # so backends that retain only the formatted message stay diagnosable. From 44fb9f434f99031f56a91216173a164f56686231 Mon Sep 17 00:00:00 2001 From: Rui Zhang <36744357+ruizhang0519@users.noreply.github.com> Date: Tue, 4 Aug 2026 15:20:28 +0000 Subject: [PATCH 2/6] commit tests --- .../tests/api/background_propagation_test.py | 147 +++++++++++++++ .../internal/nodes/background_authz_test.py | 170 ++++++++++++++++++ 2 files changed, 317 insertions(+) create mode 100644 datajunction-server/tests/api/background_propagation_test.py create mode 100644 datajunction-server/tests/internal/nodes/background_authz_test.py diff --git a/datajunction-server/tests/api/background_propagation_test.py b/datajunction-server/tests/api/background_propagation_test.py new file mode 100644 index 0000000000..29953d1410 --- /dev/null +++ b/datajunction-server/tests/api/background_propagation_test.py @@ -0,0 +1,147 @@ +""" +Authorization contract for downstream revalidation (#2234 step 0). + +Updating a node schedules ``propagate_update_downstream``, which revalidates +every node downstream in the DAG -- updating status, parents, and possibly +bumping revisions on nodes the updater does not own. + +That is deliberately *not* gated on the updater's WRITE. A node is only +downstream because its own owner pointed it at the upstream, so the blast radius +is opt-in rather than caller-controlled, and the propagation only makes stored +status reflect reality. Requiring WRITE on downstreams 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. +Propagation also swallows exceptions, so denials would be silent. + +The test below pins that contract: propagation must still run for downstreams the +caller cannot write. +""" + +import pytest +from httpx import AsyncClient + +from datajunction_server.internal.access.authorization import AuthorizationService +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" +) + +UPSTREAM = "bgauthz.agg" +DOWNSTREAM = "bgauthz.total_users" + + +class WriteOnlyOnAuthorizationService(AuthorizationService): + """Approves WRITE only for ``allowed`` names; approves every other action.""" + + name = "test_write_only_on" + + def __init__(self, allowed: set[str]): + self.allowed = allowed + + def authorize(self, auth_context, requests): + return [ + access.AccessDecision( + request=request, + approved=( + request.verb != access.ResourceAction.WRITE + or request.access_object.name in self.allowed + ), + ) + for request in requests + ] + + +@pytest.mark.asyncio +async def test_downstream_revalidation_not_gated_on_caller_write( + client: AsyncClient, + session, + mocker, +): + """ + A caller with WRITE on only the upstream still gets downstream revalidation. + + If someone later adds a WRITE check to the propagation path, this fails -- + which is the point: that check would break cross-namespace graphs. + """ + response = await client.post("/catalogs/", json={"name": "warehouse"}) + assert response.status_code in (200, 201, 409) + response = await client.post("/namespaces/bgauthz/") + assert response.status_code in (200, 201, 409) + + response = await client.post( + "/nodes/source/", + json={ + "name": "bgauthz.events", + "description": "Raw events", + "columns": [ + {"name": "id", "type": "int"}, + {"name": "country", "type": "string"}, + ], + "mode": "published", + "catalog": "warehouse", + "schema_": "db", + "table": "events", + }, + ) + assert response.status_code in (200, 201), response.text + + response = await client.post( + "/nodes/transform/", + json={ + "name": UPSTREAM, + "description": "Users per country", + "query": ( + "SELECT country, COUNT(DISTINCT id) AS num_users " + "FROM bgauthz.events GROUP BY 1" + ), + "mode": "published", + }, + ) + assert response.status_code in (200, 201), response.text + + response = await client.post( + "/nodes/metric/", + json={ + "name": DOWNSTREAM, + "description": "Total users", + "query": f"SELECT SUM(num_users) FROM {UPSTREAM}", + "mode": "published", + }, + ) + assert response.status_code in (200, 201), response.text + assert (await client.get(f"/nodes/{DOWNSTREAM}/")).json()[ + "status" + ] == NodeStatus.VALID + + # From here the caller may write the upstream but not the downstream. + mocker.patch( + VALIDATOR_AUTH_SERVICE, + lambda: WriteOnlyOnAuthorizationService({UPSTREAM}), + ) + + # Rename the column the downstream metric selects, invalidating it. + response = await client.patch( + f"/nodes/{UPSTREAM}/", + json={ + "query": ( + "SELECT country, COUNT(DISTINCT id) AS user_count " + "FROM bgauthz.events GROUP BY 1" + ), + }, + ) + assert response.status_code == 200, response.text + + # Propagation commits in its own session, so expire the shared test session + # before reading or the assertion can see the pre-propagation revision. + session.expire_all() + assert (await client.get(f"/nodes/{DOWNSTREAM}/")).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_authz_test.py b/datajunction-server/tests/internal/nodes/background_authz_test.py new file mode 100644 index 0000000000..b1d45a31ee --- /dev/null +++ b/datajunction-server/tests/internal/nodes/background_authz_test.py @@ -0,0 +1,170 @@ +""" +Authorization on the background mutation path (#2234 step 0). + +``derive_frozen_measures`` and ``save_column_level_lineage`` run after the HTTP +response has returned, in their own session, and used to take only a revision id +-- so the mutation carried no authorization context of its own and relied +entirely on whichever caller scheduled it having checked first. + +Every current scheduling site is a user-initiated create/update/revalidate that +already required WRITE on that node, so requiring it again here changes nothing +for legitimate flows; it makes the internal path independently fail-closed. + +(Downstream revalidation is the deliberate exception -- see +``_propagate_update_downstream`` and tests/api/background_propagation_test.py.) +""" + +import pytest +from httpx import AsyncClient +from sqlalchemy import select + +from datajunction_server.database.node import NodeRevision +from datajunction_server.internal.access.authorization import AuthorizationService +from datajunction_server.internal.nodes import ( + derive_frozen_measures, + save_column_level_lineage, +) +from datajunction_server.models import access + +VALIDATOR_AUTH_SERVICE = ( + "datajunction_server.internal.access.authorization." + "validator.get_authorization_service" +) + +METRIC = "bgwrite.total_events" + + +class DenyWriteAuthorizationService(AuthorizationService): + """Approves everything except WRITE -- a caller without write access.""" + + name = "test_deny_write_background" + + def authorize(self, auth_context, requests): + return [ + access.AccessDecision( + request=request, + approved=request.verb != access.ResourceAction.WRITE, + ) + for request in requests + ] + + +class AllowAllAuthorizationService(AuthorizationService): + """Approves everything -- the control for the denial cases.""" + + name = "test_allow_all_background" + + def authorize(self, auth_context, requests): + return [ + access.AccessDecision(request=request, approved=True) + for request in requests + ] + + +async def _create_metric(client: AsyncClient) -> None: + response = await client.post("/catalogs/", json={"name": "warehouse"}) + assert response.status_code in (200, 201, 409) + response = await client.post("/namespaces/bgwrite/") + assert response.status_code in (200, 201, 409) + response = await client.post( + "/nodes/source/", + json={ + "name": "bgwrite.events", + "description": "Raw events", + "columns": [ + {"name": "id", "type": "int"}, + {"name": "country", "type": "string"}, + ], + "mode": "published", + "catalog": "warehouse", + "schema_": "db", + "table": "events", + }, + ) + assert response.status_code in (200, 201), response.text + response = await client.post( + "/nodes/metric/", + json={ + "name": METRIC, + "description": "Total events", + "query": "SELECT COUNT(DISTINCT id) FROM bgwrite.events", + "mode": "published", + }, + ) + assert response.status_code in (200, 201), response.text + + +async def _current_revision(session, name: str) -> NodeRevision: + revision = ( + ( + await session.execute( + select(NodeRevision) + .where(NodeRevision.name == name) + .order_by(NodeRevision.id.desc()), + ) + ) + .scalars() + .first() + ) + assert revision is not None + return revision + + +@pytest.mark.asyncio +async def test_save_column_level_lineage_denied_without_write( + client: AsyncClient, + session, + current_user, + mocker, +): + """Lineage is not written for a caller who cannot write the node.""" + await _create_metric(client) + revision = await _current_revision(session, METRIC) + revision.lineage = None + await session.commit() + + mocker.patch(VALIDATOR_AUTH_SERVICE, lambda: DenyWriteAuthorizationService()) + await save_column_level_lineage( + node_revision_id=revision.id, + current_user=current_user, + ) + + session.expire_all() + revision = await _current_revision(session, METRIC) + assert not revision.lineage, "lineage was written despite WRITE being denied" + + +@pytest.mark.asyncio +async def test_derive_frozen_measures_denied_without_write( + client: AsyncClient, + session, + current_user, + mocker, +): + """ + Derivation never runs for a caller who cannot write the node. + + Asserts the derivation body is not reached rather than inspecting the return + value, which is empty for several unrelated reasons. + """ + await _create_metric(client) + revision = await _current_revision(session, METRIC) + derive = mocker.patch( + "datajunction_server.internal.nodes._derive_frozen_measures_impl", + return_value=[], + ) + + mocker.patch(VALIDATOR_AUTH_SERVICE, lambda: DenyWriteAuthorizationService()) + await derive_frozen_measures( + node_revision_id=revision.id, + current_user=current_user, + ) + derive.assert_not_called() + + # Control: the same call proceeds once WRITE is allowed. + mocker.patch(VALIDATOR_AUTH_SERVICE, lambda: AllowAllAuthorizationService()) + await derive_frozen_measures( + node_revision_id=revision.id, + current_user=current_user, + ) + derive.assert_called_once() From 887ec2cfcae5d9d3587b41333027eacb4122dc5e Mon Sep 17 00:00:00 2001 From: Rui Zhang <36744357+ruizhang0519@users.noreply.github.com> Date: Tue, 4 Aug 2026 18:18:25 +0000 Subject: [PATCH 3/6] Authorize the resource each background task's endpoint governed --- .../datajunction_server/internal/nodes.py | 70 ++++-- .../scripts/migrate-measures.py | 40 +--- .../internal/nodes/background_authz_test.py | 214 +++++++++++++++--- .../internal/nodes/background_tasks_test.py | 2 + 4 files changed, 236 insertions(+), 90 deletions(-) diff --git a/datajunction-server/datajunction_server/internal/nodes.py b/datajunction-server/datajunction_server/internal/nodes.py index 7c97f358cd..407d570d86 100644 --- a/datajunction-server/datajunction_server/internal/nodes.py +++ b/datajunction-server/datajunction_server/internal/nodes.py @@ -67,7 +67,11 @@ validate_node_data, validate_node_data_v2, ) -from datajunction_server.models.access import ResourceAction +from datajunction_server.models.access import ( + Resource, + ResourceAction, + ResourceRequest, +) from datajunction_server.models.attribute import ( AttributeTypeIdentifier, ColumnAttributes, @@ -299,18 +303,24 @@ async def create_a_node( save_history=save_history, ) + # Node creation is governed on the target namespace, so the background work + # re-authorizes that same resource (see _background_write_allowed). + create_access_target = Resource.from_namespace(data.namespace) + # 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, 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 @@ -708,35 +718,36 @@ async def create_cube_node_revision( async def _background_write_allowed( session: AsyncSession, - node_revision_id: int, + access_target: Resource, current_user: User, action_description: str, ) -> bool: """ - Whether ``current_user`` may WRITE the node owning ``node_revision_id``. + 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. Denials are logged and skip the mutation rather than raising, since - the callers swallow exceptions and an exception would be indistinguishable - from a failure. - """ - node_name = ( - await session.execute( - select(NodeRevision.name).where(NodeRevision.id == node_revision_id), - ) - ).scalar_one_or_none() - if node_name is None: # pragma: no cover - return False + 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_by_node_name(node_name, ResourceAction.WRITE) + 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 node %s: %s lacks WRITE", + "Skipping %s for %s %s: %s lacks WRITE", action_description, - node_name, + access_target.resource_type.value, + access_target.name, current_user.username, ) return False @@ -746,6 +757,7 @@ async def _background_write_allowed( 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. @@ -759,14 +771,15 @@ async def derive_frozen_measures( 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 the - metric itself rather than trusting the scheduling caller (#2234 step 0). + 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, - node_revision_id, + access_target, current_user, "deriving frozen measures", ): @@ -1418,6 +1431,8 @@ async def update_node_with_query( save_column_level_lineage, node_revision_id=new_revision.id, current_user=current_user, + # Node updates are governed on the node itself. + access_target=Resource.from_node(new_revision), ) # TODO: Do not save this until: # 1. We get to the bottom of why there are query building discrepancies @@ -2300,18 +2315,23 @@ async def create_new_revision_from_existing( return new_revision -async def save_column_level_lineage(node_revision_id: int, current_user: User): +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 the - node itself rather than trusting the scheduling caller (#2234 step 0). + 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, - node_revision_id, + access_target, current_user, "saving column-level lineage", ): @@ -3743,6 +3763,8 @@ async def revalidate_node( 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/migrate-measures.py b/datajunction-server/scripts/migrate-measures.py index b80ff3fd48..aafde9773f 100644 --- a/datajunction-server/scripts/migrate-measures.py +++ b/datajunction-server/scripts/migrate-measures.py @@ -5,7 +5,7 @@ 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.internal.nodes import derive_frozen_measures_bulk from datajunction_server.utils import get_settings settings = get_settings() @@ -46,37 +46,15 @@ async def backfill_measures(): 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 - + # Operator tool with no request user, so it uses the system-facing + # bulk API (which also links the measures to each revision); the + # per-revision entry point authorizes a user for WRITE. + await derive_frozen_measures_bulk( + session, + [revision.id for revision in metric_revisions], + ) await session.commit() + print(f"Derived frozen measures for {len(metric_revisions)} revisions") if __name__ == "__main__": diff --git a/datajunction-server/tests/internal/nodes/background_authz_test.py b/datajunction-server/tests/internal/nodes/background_authz_test.py index b1d45a31ee..4bc92b4e16 100644 --- a/datajunction-server/tests/internal/nodes/background_authz_test.py +++ b/datajunction-server/tests/internal/nodes/background_authz_test.py @@ -6,9 +6,10 @@ -- so the mutation carried no authorization context of its own and relied entirely on whichever caller scheduled it having checked first. -Every current scheduling site is a user-initiated create/update/revalidate that -already required WRITE on that node, so requiring it again here changes nothing -for legitimate flows; it makes the internal path independently fail-closed. +They now re-authorize the resource their scheduling endpoint governed, which is +not always the node: creation is governed on the target namespace, updates and +revalidation on the node. The tests below pin the actor and that exact resource, +not merely that "some WRITE" was denied. (Downstream revalidation is the deliberate exception -- see ``_propagate_update_downstream`` and tests/api/background_propagation_test.py.) @@ -19,8 +20,13 @@ from sqlalchemy import select from datajunction_server.database.node import NodeRevision -from datajunction_server.internal.access.authorization import AuthorizationService +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, ) @@ -31,45 +37,64 @@ "validator.get_authorization_service" ) -METRIC = "bgwrite.total_events" - +NAMESPACE = "bgwrite" +METRIC = f"{NAMESPACE}.total_events" -class DenyWriteAuthorizationService(AuthorizationService): - """Approves everything except WRITE -- a caller without write access.""" - name = "test_deny_write_background" - - def authorize(self, auth_context, requests): - return [ - access.AccessDecision( - request=request, - approved=request.verb != access.ResourceAction.WRITE, - ) - for request in requests - ] +class RecordingAuthorizationService(AuthorizationService): + """ + Records what it was asked to authorize, and approves or denies wholesale. + Recording the requests (and the acting principal) is what lets a test assert + the implementation authorized the *right* resource for the *right* user, + rather than only that it consulted authorization at all. + """ -class AllowAllAuthorizationService(AuthorizationService): - """Approves everything -- the control for the denial cases.""" + name = "test_recording_background" - name = "test_allow_all_background" + def __init__(self, approve: bool): + self.approve = approve + 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=True) + access.AccessDecision(request=request, approved=self.approve) for request in requests ] +def assert_authorized_namespace_write( + recorder: RecordingAuthorizationService, + username: str, +) -> None: + """The task must request WRITE on the namespace create was governed on.""" + assert recorder.usernames == [username], ( + f"built authorization context for {recorder.usernames}, expected [{username}]" + ) + assert [ + (request.verb, request.access_object.resource_type, request.access_object.name) + for request in recorder.requests + ] == [ + ( + access.ResourceAction.WRITE, + access.ResourceType.NAMESPACE, + NAMESPACE, + ), + ] + + async def _create_metric(client: AsyncClient) -> None: response = await client.post("/catalogs/", json={"name": "warehouse"}) assert response.status_code in (200, 201, 409) - response = await client.post("/namespaces/bgwrite/") + response = await client.post(f"/namespaces/{NAMESPACE}/") assert response.status_code in (200, 201, 409) response = await client.post( "/nodes/source/", json={ - "name": "bgwrite.events", + "name": f"{NAMESPACE}.events", "description": "Raw events", "columns": [ {"name": "id", "type": "int"}, @@ -87,7 +112,7 @@ async def _create_metric(client: AsyncClient) -> None: json={ "name": METRIC, "description": "Total events", - "query": "SELECT COUNT(DISTINCT id) FROM bgwrite.events", + "query": f"SELECT COUNT(DISTINCT id) FROM {NAMESPACE}.events", "mode": "published", }, ) @@ -111,60 +136,179 @@ async def _current_revision(session, name: str) -> NodeRevision: @pytest.mark.asyncio -async def test_save_column_level_lineage_denied_without_write( +async def test_create_schedules_background_work_against_the_namespace( client: AsyncClient, + mocker, +): + """ + Node creation must hand its background tasks the namespace it authorized. + + Pins the wiring, not just the helper: creation is governed on + ``data.namespace``, so scheduling a node target would deny the work under a + namespace-scoped grant (see the RBAC test above). + """ + derive = mocker.patch( + "datajunction_server.internal.nodes.derive_frozen_measures", + ) + lineage = mocker.patch( + "datajunction_server.internal.nodes.save_column_level_lineage", + ) + + await _create_metric(client) + + expected = access.Resource.from_namespace(NAMESPACE) + assert derive.call_args.kwargs["access_target"] == expected + assert lineage.call_args.kwargs["access_target"] == expected + + +@pytest.mark.asyncio +async def test_exact_namespace_grant_authorizes_create_background_work( session, current_user, mocker, ): - """Lineage is not written for a caller who cannot write the node.""" + """ + An exact namespace WRITE grant -- enough to create the node -- must also + satisfy the background work scheduled by that create. + + Run against the real RBAC matcher under a restrictive policy, because the + resource type is what decides this: a NAMESPACE scope only covers a NODE + request through pattern matching, and `bgwrite` does not match the node name + `bgwrite.total_events`. Authorizing the node here would therefore leave a + successfully created node without its derived metadata. + """ + settings = mocker.patch( + "datajunction_server.internal.access.authorization.service.settings", + ) + settings.authorization_provider = "rbac" + settings.default_access_policy = "restrictive" + + role = Role(name="bgwrite-namespace-writer", created_by_id=current_user.id) + session.add(role) + await session.flush() + session.add( + RoleScope( + role_id=role.id, + action=access.ResourceAction.WRITE, + scope_type=access.ResourceType.NAMESPACE, + scope_value=NAMESPACE, + ), + ) + session.add( + 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()) + + assert ( + await _background_write_allowed( + session, + access.Resource.from_namespace(NAMESPACE), + current_user, + "test", + ) + is True + ), "the grant that allowed the create did not allow its background work" + + # The node target the create path must NOT use, shown denied by the same grant. + assert ( + await _background_write_allowed( + session, + access.Resource( + name=METRIC, + resource_type=access.ResourceType.NODE, + ), + current_user, + "test", + ) + is False + ) + + +@pytest.mark.asyncio +async def test_save_column_level_lineage_authorizes_create_target( + client: AsyncClient, + session, + current_user, + mocker, +): + """Lineage is written only when the caller may write the create target.""" await _create_metric(client) revision = await _current_revision(session, METRIC) revision.lineage = None await session.commit() + access_target = access.Resource.from_namespace(NAMESPACE) - mocker.patch(VALIDATOR_AUTH_SERVICE, lambda: DenyWriteAuthorizationService()) + denied = RecordingAuthorizationService(approve=False) + mocker.patch(VALIDATOR_AUTH_SERVICE, lambda: denied) await save_column_level_lineage( node_revision_id=revision.id, current_user=current_user, + access_target=access_target, ) + assert_authorized_namespace_write(denied, current_user.username) session.expire_all() revision = await _current_revision(session, METRIC) assert not revision.lineage, "lineage was written despite WRITE being denied" + # Allow-path control: the same call writes lineage once WRITE is granted. + allowed = RecordingAuthorizationService(approve=True) + mocker.patch(VALIDATOR_AUTH_SERVICE, lambda: allowed) + await save_column_level_lineage( + node_revision_id=revision.id, + current_user=current_user, + access_target=access_target, + ) + assert_authorized_namespace_write(allowed, current_user.username) + + session.expire_all() + revision = await _current_revision(session, METRIC) + assert revision.lineage, "lineage was not written when WRITE was granted" + @pytest.mark.asyncio -async def test_derive_frozen_measures_denied_without_write( +async def test_derive_frozen_measures_authorizes_create_target( client: AsyncClient, session, current_user, mocker, ): """ - Derivation never runs for a caller who cannot write the node. + Derivation runs only when the caller may write the create target. - Asserts the derivation body is not reached rather than inspecting the return - value, which is empty for several unrelated reasons. + Asserts the derivation body is or is not reached rather than inspecting the + return value, which is empty for several unrelated reasons. """ await _create_metric(client) revision = await _current_revision(session, METRIC) + access_target = access.Resource.from_namespace(NAMESPACE) derive = mocker.patch( "datajunction_server.internal.nodes._derive_frozen_measures_impl", return_value=[], ) - mocker.patch(VALIDATOR_AUTH_SERVICE, lambda: DenyWriteAuthorizationService()) + denied = RecordingAuthorizationService(approve=False) + mocker.patch(VALIDATOR_AUTH_SERVICE, lambda: denied) await derive_frozen_measures( node_revision_id=revision.id, current_user=current_user, + access_target=access_target, ) + assert_authorized_namespace_write(denied, current_user.username) derive.assert_not_called() - # Control: the same call proceeds once WRITE is allowed. - mocker.patch(VALIDATOR_AUTH_SERVICE, lambda: AllowAllAuthorizationService()) + # Allow-path control: the same call proceeds once WRITE is granted. + allowed = RecordingAuthorizationService(approve=True) + mocker.patch(VALIDATOR_AUTH_SERVICE, lambda: allowed) await derive_frozen_measures( node_revision_id=revision.id, current_user=current_user, + access_target=access_target, ) + assert_authorized_namespace_write(allowed, current_user.username) derive.assert_called_once() diff --git a/datajunction-server/tests/internal/nodes/background_tasks_test.py b/datajunction-server/tests/internal/nodes/background_tasks_test.py index b6864a383a..43231e0de3 100644 --- a/datajunction-server/tests/internal/nodes/background_tasks_test.py +++ b/datajunction-server/tests/internal/nodes/background_tasks_test.py @@ -73,6 +73,7 @@ async def test_derive_frozen_measures_swallows_exceptions(caplog): result = await derive_frozen_measures( node_revision_id=99, current_user=MagicMock(), + access_target=MagicMock(), ) assert result == [] @@ -102,6 +103,7 @@ async def test_save_column_level_lineage_swallows_exceptions(caplog): 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), From f0439aeb18bd44a76643ff747c8883de49ebfa52 Mon Sep 17 00:00:00 2001 From: Rui Zhang <36744357+ruizhang0519@users.noreply.github.com> Date: Tue, 4 Aug 2026 19:40:58 +0000 Subject: [PATCH 4/6] update tests --- .../datajunction_server/internal/nodes.py | 2 +- .../tests/api/background_propagation_test.py | 147 -------- .../internal/nodes/background_authz_test.py | 331 ++++++++++-------- 3 files changed, 192 insertions(+), 288 deletions(-) delete mode 100644 datajunction-server/tests/api/background_propagation_test.py diff --git a/datajunction-server/datajunction_server/internal/nodes.py b/datajunction-server/datajunction_server/internal/nodes.py index 407d570d86..cc64f313e2 100644 --- a/datajunction-server/datajunction_server/internal/nodes.py +++ b/datajunction-server/datajunction_server/internal/nodes.py @@ -1794,7 +1794,7 @@ async def _propagate_update_downstream( 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/api/background_propagation_test.py. + tests/internal/nodes/background_authz_test.py. """ _logger.info("Propagating update of node %s downstream", node.name) downstreams = await get_downstream_nodes( diff --git a/datajunction-server/tests/api/background_propagation_test.py b/datajunction-server/tests/api/background_propagation_test.py deleted file mode 100644 index 29953d1410..0000000000 --- a/datajunction-server/tests/api/background_propagation_test.py +++ /dev/null @@ -1,147 +0,0 @@ -""" -Authorization contract for downstream revalidation (#2234 step 0). - -Updating a node schedules ``propagate_update_downstream``, which revalidates -every node downstream in the DAG -- updating status, parents, and possibly -bumping revisions on nodes the updater does not own. - -That is deliberately *not* gated on the updater's WRITE. A node is only -downstream because its own owner pointed it at the upstream, so the blast radius -is opt-in rather than caller-controlled, and the propagation only makes stored -status reflect reality. Requiring WRITE on downstreams 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. -Propagation also swallows exceptions, so denials would be silent. - -The test below pins that contract: propagation must still run for downstreams the -caller cannot write. -""" - -import pytest -from httpx import AsyncClient - -from datajunction_server.internal.access.authorization import AuthorizationService -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" -) - -UPSTREAM = "bgauthz.agg" -DOWNSTREAM = "bgauthz.total_users" - - -class WriteOnlyOnAuthorizationService(AuthorizationService): - """Approves WRITE only for ``allowed`` names; approves every other action.""" - - name = "test_write_only_on" - - def __init__(self, allowed: set[str]): - self.allowed = allowed - - def authorize(self, auth_context, requests): - return [ - access.AccessDecision( - request=request, - approved=( - request.verb != access.ResourceAction.WRITE - or request.access_object.name in self.allowed - ), - ) - for request in requests - ] - - -@pytest.mark.asyncio -async def test_downstream_revalidation_not_gated_on_caller_write( - client: AsyncClient, - session, - mocker, -): - """ - A caller with WRITE on only the upstream still gets downstream revalidation. - - If someone later adds a WRITE check to the propagation path, this fails -- - which is the point: that check would break cross-namespace graphs. - """ - response = await client.post("/catalogs/", json={"name": "warehouse"}) - assert response.status_code in (200, 201, 409) - response = await client.post("/namespaces/bgauthz/") - assert response.status_code in (200, 201, 409) - - response = await client.post( - "/nodes/source/", - json={ - "name": "bgauthz.events", - "description": "Raw events", - "columns": [ - {"name": "id", "type": "int"}, - {"name": "country", "type": "string"}, - ], - "mode": "published", - "catalog": "warehouse", - "schema_": "db", - "table": "events", - }, - ) - assert response.status_code in (200, 201), response.text - - response = await client.post( - "/nodes/transform/", - json={ - "name": UPSTREAM, - "description": "Users per country", - "query": ( - "SELECT country, COUNT(DISTINCT id) AS num_users " - "FROM bgauthz.events GROUP BY 1" - ), - "mode": "published", - }, - ) - assert response.status_code in (200, 201), response.text - - response = await client.post( - "/nodes/metric/", - json={ - "name": DOWNSTREAM, - "description": "Total users", - "query": f"SELECT SUM(num_users) FROM {UPSTREAM}", - "mode": "published", - }, - ) - assert response.status_code in (200, 201), response.text - assert (await client.get(f"/nodes/{DOWNSTREAM}/")).json()[ - "status" - ] == NodeStatus.VALID - - # From here the caller may write the upstream but not the downstream. - mocker.patch( - VALIDATOR_AUTH_SERVICE, - lambda: WriteOnlyOnAuthorizationService({UPSTREAM}), - ) - - # Rename the column the downstream metric selects, invalidating it. - response = await client.patch( - f"/nodes/{UPSTREAM}/", - json={ - "query": ( - "SELECT country, COUNT(DISTINCT id) AS user_count " - "FROM bgauthz.events GROUP BY 1" - ), - }, - ) - assert response.status_code == 200, response.text - - # Propagation commits in its own session, so expire the shared test session - # before reading or the assertion can see the pre-propagation revision. - session.expire_all() - assert (await client.get(f"/nodes/{DOWNSTREAM}/")).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_authz_test.py b/datajunction-server/tests/internal/nodes/background_authz_test.py index 4bc92b4e16..afca34b4f5 100644 --- a/datajunction-server/tests/internal/nodes/background_authz_test.py +++ b/datajunction-server/tests/internal/nodes/background_authz_test.py @@ -1,20 +1,25 @@ """ Authorization on the background mutation path (#2234 step 0). -``derive_frozen_measures`` and ``save_column_level_lineage`` run after the HTTP -response has returned, in their own session, and used to take only a revision id --- so the mutation carried no authorization context of its own and relied -entirely on whichever caller scheduled it having checked first. - -They now re-authorize the resource their scheduling endpoint governed, which is -not always the node: creation is governed on the target namespace, updates and -revalidation on the node. The tests below pin the actor and that exact resource, -not merely that "some WRITE" was denied. - -(Downstream revalidation is the deliberate exception -- see -``_propagate_update_downstream`` and tests/api/background_propagation_test.py.) +These tasks run after the HTTP response has returned, in their own session, so +they sit outside the request's AccessChecker. Two contracts follow from that: + +* ``derive_frozen_measures`` and ``save_column_level_lineage`` re-authorize the + resource their scheduling endpoint governed -- the target namespace for + creation, the node for updates -- rather than trusting that caller to have + checked. The resource matters: a namespace scope reaches a node request only + through pattern matching, so authorizing the node would deny work scheduled by + a create that a namespace-scoped grant allowed. + +* ``propagate_update_downstream`` deliberately does *not* authorize, because a + node is only downstream once its own owner points it at the upstream. Gating it + would either block owners whose nodes have dependents, or skip the denied ones + and leave the graph asserting VALID for nodes that are now broken -- silently, + since propagation swallows exceptions. """ +from functools import partial + import pytest from httpx import AsyncClient from sqlalchemy import select @@ -31,29 +36,45 @@ 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" ) -NAMESPACE = "bgwrite" -METRIC = f"{NAMESPACE}.total_events" +NAMESPACE = "bgauthz" +SOURCE = f"{NAMESPACE}.events" +TRANSFORM = f"{NAMESPACE}.users_per_country" +METRIC = f"{NAMESPACE}.total_users" +# Node creation is governed on the target namespace, so that is the resource 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): """ - Records what it was asked to authorize, and approves or denies wholesale. + Decides each request by ``approves``, recording what it was asked. - Recording the requests (and the acting principal) is what lets a test assert + Recording the requests and the acting principal is what lets a test assert the implementation authorized the *right* resource for the *right* user, rather than only that it consulted authorization at all. """ name = "test_recording_background" - def __init__(self, approve: bool): - self.approve = approve + def __init__(self, approves=allow): + self.approves = approves self.requests: list[access.ResourceRequest] = [] self.usernames: list[str] = [] @@ -61,40 +82,39 @@ def authorize(self, auth_context, requests): self.requests.extend(requests) self.usernames.append(auth_context.username) return [ - access.AccessDecision(request=request, approved=self.approve) + access.AccessDecision(request=request, approved=self.approves(request)) for request in requests ] -def assert_authorized_namespace_write( - recorder: RecordingAuthorizationService, - username: str, -) -> None: - """The task must request WRITE on the namespace create was governed on.""" - assert recorder.usernames == [username], ( - f"built authorization context for {recorder.usernames}, expected [{username}]" +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) -> None: + await _post( + client, + "/nodes/metric/", + { + "name": name, + "description": "Total users", + "query": f"SELECT SUM(num_users) FROM {TRANSFORM}", + "mode": "published", + }, ) - assert [ - (request.verb, request.access_object.resource_type, request.access_object.name) - for request in recorder.requests - ] == [ - ( - access.ResourceAction.WRITE, - access.ResourceType.NAMESPACE, - NAMESPACE, - ), - ] -async def _create_metric(client: AsyncClient) -> None: - response = await client.post("/catalogs/", json={"name": "warehouse"}) - assert response.status_code in (200, 201, 409) - response = await client.post(f"/namespaces/{NAMESPACE}/") - assert response.status_code in (200, 201, 409) - response = await client.post( +@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/", - json={ - "name": f"{NAMESPACE}.events", + { + "name": SOURCE, "description": "Raw events", "columns": [ {"name": "id", "type": "int"}, @@ -106,20 +126,25 @@ async def _create_metric(client: AsyncClient) -> None: "table": "events", }, ) - assert response.status_code in (200, 201), response.text - response = await client.post( - "/nodes/metric/", - json={ - "name": METRIC, - "description": "Total events", - "query": f"SELECT COUNT(DISTINCT id) FROM {NAMESPACE}.events", + await _post( + client, + "/nodes/transform/", + { + "name": TRANSFORM, + "description": "Users per country", + "query": ( + f"SELECT country, COUNT(DISTINCT id) AS num_users " + f"FROM {SOURCE} GROUP BY 1" + ), "mode": "published", }, ) - assert response.status_code in (200, 201), response.text + await _create_metric(client, METRIC) -async def _current_revision(session, name: str) -> NodeRevision: +async def _revision(session, name: str) -> NodeRevision: + """Latest revision of ``name``, read fresh -- the tasks commit elsewhere.""" + session.expire_all() revision = ( ( await session.execute( @@ -135,30 +160,47 @@ async def _current_revision(session, name: str) -> NodeRevision: return revision +async def _run_task(mocker, task, approves) -> RecordingAuthorizationService: + """Run ``task`` under a recording service and hand it back for assertions.""" + recorder = RecordingAuthorizationService(approves) + mocker.patch(VALIDATOR_AUTH_SERVICE, lambda: recorder) + await task() + return recorder + + +def assert_write_authorized( + recorder: RecordingAuthorizationService, + username: str, + target: access.Resource = CREATE_TARGET, +) -> None: + """The task must have asked for WRITE on ``target``, acting as ``username``.""" + assert recorder.usernames == [username] + assert [(request.verb, request.access_object) for request in recorder.requests] == [ + (access.ResourceAction.WRITE, target), + ] + + @pytest.mark.asyncio async def test_create_schedules_background_work_against_the_namespace( client: AsyncClient, + metric_graph, mocker, ): """ - Node creation must hand its background tasks the namespace it authorized. + Creation must hand its background tasks the namespace it authorized. - Pins the wiring, not just the helper: creation is governed on - ``data.namespace``, so scheduling a node target would deny the work under a - namespace-scoped grant (see the RBAC test above). + Pins the wiring rather than the helper: scheduling a node target would deny + the work under the namespace-scoped grant that allowed the create. """ - derive = mocker.patch( - "datajunction_server.internal.nodes.derive_frozen_measures", - ) + derive = mocker.patch("datajunction_server.internal.nodes.derive_frozen_measures") lineage = mocker.patch( "datajunction_server.internal.nodes.save_column_level_lineage", ) - await _create_metric(client) + await _create_metric(client, f"{METRIC}_again") - expected = access.Resource.from_namespace(NAMESPACE) - assert derive.call_args.kwargs["access_target"] == expected - assert lineage.call_args.kwargs["access_target"] == expected + assert derive.call_args.kwargs["access_target"] == CREATE_TARGET + assert lineage.call_args.kwargs["access_target"] == CREATE_TARGET @pytest.mark.asyncio @@ -169,13 +211,11 @@ async def test_exact_namespace_grant_authorizes_create_background_work( ): """ An exact namespace WRITE grant -- enough to create the node -- must also - satisfy the background work scheduled by that create. + satisfy the background work that create schedules. - Run against the real RBAC matcher under a restrictive policy, because the - resource type is what decides this: a NAMESPACE scope only covers a NODE - request through pattern matching, and `bgwrite` does not match the node name - `bgwrite.total_events`. Authorizing the node here would therefore leave a - successfully created node without its derived metadata. + Runs the real RBAC matcher under a restrictive policy, since resource type is + what decides it: `bgauthz` does not match the node name `bgauthz.total_users`, + so authorizing the node would leave a created node without derived metadata. """ settings = mocker.patch( "datajunction_server.internal.access.authorization.service.settings", @@ -183,7 +223,7 @@ async def test_exact_namespace_grant_authorizes_create_background_work( settings.authorization_provider = "rbac" settings.default_access_policy = "restrictive" - role = Role(name="bgwrite-namespace-writer", created_by_id=current_user.id) + role = Role(name="bgauthz-namespace-writer", created_by_id=current_user.id) session.add(role) await session.flush() session.add( @@ -204,76 +244,49 @@ async def test_exact_namespace_grant_authorizes_create_background_work( await session.commit() mocker.patch(VALIDATOR_AUTH_SERVICE, lambda: RBACAuthorizationService()) - assert ( - await _background_write_allowed( - session, - access.Resource.from_namespace(NAMESPACE), - current_user, - "test", - ) - is True - ), "the grant that allowed the create did not allow its background work" - - # The node target the create path must NOT use, shown denied by the same grant. - assert ( - await _background_write_allowed( - session, - access.Resource( - name=METRIC, - resource_type=access.ResourceType.NODE, - ), - current_user, - "test", - ) - is False + 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_save_column_level_lineage_authorizes_create_target( - client: AsyncClient, +async def test_lineage_authorizes_the_create_target( + metric_graph, session, current_user, mocker, ): """Lineage is written only when the caller may write the create target.""" - await _create_metric(client) - revision = await _current_revision(session, METRIC) + revision = await _revision(session, METRIC) revision.lineage = None await session.commit() - access_target = access.Resource.from_namespace(NAMESPACE) - - denied = RecordingAuthorizationService(approve=False) - mocker.patch(VALIDATOR_AUTH_SERVICE, lambda: denied) - await save_column_level_lineage( + task = partial( + save_column_level_lineage, node_revision_id=revision.id, current_user=current_user, - access_target=access_target, + access_target=CREATE_TARGET, ) - assert_authorized_namespace_write(denied, current_user.username) - - session.expire_all() - revision = await _current_revision(session, METRIC) - assert not revision.lineage, "lineage was written despite WRITE being denied" - # Allow-path control: the same call writes lineage once WRITE is granted. - allowed = RecordingAuthorizationService(approve=True) - mocker.patch(VALIDATOR_AUTH_SERVICE, lambda: allowed) - await save_column_level_lineage( - node_revision_id=revision.id, - current_user=current_user, - access_target=access_target, + denied = await _run_task(mocker, task, deny) + assert_write_authorized(denied, current_user.username) + assert not (await _revision(session, METRIC)).lineage, ( + "lineage was written despite WRITE being denied" ) - assert_authorized_namespace_write(allowed, current_user.username) - session.expire_all() - revision = await _current_revision(session, METRIC) - assert revision.lineage, "lineage was not written when WRITE was granted" + granted = await _run_task(mocker, task, allow) + assert_write_authorized(granted, current_user.username) + assert (await _revision(session, METRIC)).lineage, ( + "lineage was not written when WRITE was granted" + ) @pytest.mark.asyncio -async def test_derive_frozen_measures_authorizes_create_target( - client: AsyncClient, +async def test_frozen_measures_authorize_the_create_target( + metric_graph, session, current_user, mocker, @@ -281,34 +294,72 @@ async def test_derive_frozen_measures_authorizes_create_target( """ Derivation runs only when the caller may write the create target. - Asserts the derivation body is or is not reached rather than inspecting the + Asserts whether the derivation body is reached rather than inspecting the return value, which is empty for several unrelated reasons. """ - await _create_metric(client) - revision = await _current_revision(session, METRIC) - access_target = access.Resource.from_namespace(NAMESPACE) + revision = await _revision(session, METRIC) derive = mocker.patch( "datajunction_server.internal.nodes._derive_frozen_measures_impl", return_value=[], ) - - denied = RecordingAuthorizationService(approve=False) - mocker.patch(VALIDATOR_AUTH_SERVICE, lambda: denied) - await derive_frozen_measures( + task = partial( + derive_frozen_measures, node_revision_id=revision.id, current_user=current_user, - access_target=access_target, + access_target=CREATE_TARGET, ) - assert_authorized_namespace_write(denied, current_user.username) + + denied = await _run_task(mocker, task, deny) + assert_write_authorized(denied, current_user.username) derive.assert_not_called() - # Allow-path control: the same call proceeds once WRITE is granted. - allowed = RecordingAuthorizationService(approve=True) - mocker.patch(VALIDATOR_AUTH_SERVICE, lambda: allowed) - await derive_frozen_measures( - node_revision_id=revision.id, - current_user=current_user, - access_target=access_target, - ) - assert_authorized_namespace_write(allowed, current_user.username) + granted = await _run_task(mocker, task, allow) + assert_write_authorized(granted, current_user.username) derive.assert_called_once() + + +@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 ever added to the propagation path -- which is the + point: such a check 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" + ) From 2e96a230a776b46f78e218d79f60e45e7fcba241 Mon Sep 17 00:00:00 2001 From: Rui Zhang <36744357+ruizhang0519@users.noreply.github.com> Date: Tue, 4 Aug 2026 20:15:37 +0000 Subject: [PATCH 5/6] bug fix and tests update --- .../datajunction_server/internal/nodes.py | 37 ++- .../internal/nodes/background_authz_test.py | 227 ++++++++---------- 2 files changed, 135 insertions(+), 129 deletions(-) diff --git a/datajunction-server/datajunction_server/internal/nodes.py b/datajunction-server/datajunction_server/internal/nodes.py index cc64f313e2..a8a562d901 100644 --- a/datajunction-server/datajunction_server/internal/nodes.py +++ b/datajunction-server/datajunction_server/internal/nodes.py @@ -154,6 +154,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 @@ -245,6 +246,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, @@ -257,6 +260,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 @@ -303,10 +307,6 @@ async def create_a_node( save_history=save_history, ) - # Node creation is governed on the target namespace, so the background work - # re-authorizes that same resource (see _background_write_allowed). - create_access_target = Resource.from_namespace(data.namespace) - # For metric nodes, derive the referenced frozen measures and save them if node.type == NodeType.METRIC: background_tasks.add_task( @@ -354,6 +354,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 @@ -716,6 +717,19 @@ async def create_cube_node_revision( return node_revision +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, @@ -1316,11 +1330,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 """ @@ -1431,8 +1451,7 @@ async def update_node_with_query( save_column_level_lineage, node_revision_id=new_revision.id, current_user=current_user, - # Node updates are governed on the node itself. - access_target=Resource.from_node(new_revision), + 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 @@ -1937,10 +1956,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, @@ -1988,6 +2012,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( diff --git a/datajunction-server/tests/internal/nodes/background_authz_test.py b/datajunction-server/tests/internal/nodes/background_authz_test.py index afca34b4f5..2b0dd0177b 100644 --- a/datajunction-server/tests/internal/nodes/background_authz_test.py +++ b/datajunction-server/tests/internal/nodes/background_authz_test.py @@ -1,21 +1,19 @@ """ Authorization on the background mutation path (#2234 step 0). -These tasks run after the HTTP response has returned, in their own session, so -they sit outside the request's AccessChecker. Two contracts follow from that: +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 scheduling endpoint governed -- the target namespace for - creation, the node for updates -- rather than trusting that caller to have - checked. The resource matters: a namespace scope reaches a node request only - through pattern matching, so authorizing the node would deny work scheduled by - a create that a namespace-scoped grant allowed. - -* ``propagate_update_downstream`` deliberately does *not* authorize, because a - node is only downstream once its own owner points it at the upstream. Gating it - would either block owners whose nodes have dependents, or skip the denied ones - and leave the graph asserting VALID for nodes that are now broken -- silently, - since propagation swallows exceptions. + 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 @@ -44,13 +42,15 @@ "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" -# Node creation is governed on the target namespace, so that is the resource its -# background work must re-authorize. +# Creation is governed on the target namespace, so that is what its background +# work must re-authorize. CREATE_TARGET = access.Resource.from_namespace(NAMESPACE) @@ -64,11 +64,8 @@ def allow(_request: access.ResourceRequest) -> bool: class RecordingAuthorizationService(AuthorizationService): """ - Decides each request by ``approves``, recording what it was asked. - - Recording the requests and the acting principal is what lets a test assert - the implementation authorized the *right* resource for the *right* user, - rather than only that it consulted authorization at all. + 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" @@ -92,13 +89,13 @@ async def _post(client: AsyncClient, url: str, json=None) -> None: assert response.status_code in (200, 201, 409), response.text -async def _create_metric(client: AsyncClient, name: str) -> None: +async def _create_metric(client: AsyncClient, name: str, description: str) -> None: await _post( client, "/nodes/metric/", { "name": name, - "description": "Total users", + "description": description, "query": f"SELECT SUM(num_users) FROM {TRANSFORM}", "mode": "published", }, @@ -115,7 +112,6 @@ async def metric_graph(client: AsyncClient) -> None: "/nodes/source/", { "name": SOURCE, - "description": "Raw events", "columns": [ {"name": "id", "type": "int"}, {"name": "country", "type": "string"}, @@ -131,7 +127,6 @@ async def metric_graph(client: AsyncClient) -> None: "/nodes/transform/", { "name": TRANSFORM, - "description": "Users per country", "query": ( f"SELECT country, COUNT(DISTINCT id) AS num_users " f"FROM {SOURCE} GROUP BY 1" @@ -139,67 +134,60 @@ async def metric_graph(client: AsyncClient) -> None: "mode": "published", }, ) - await _create_metric(client, METRIC) + 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() - revision = ( - ( - await session.execute( - select(NodeRevision) - .where(NodeRevision.name == name) - .order_by(NodeRevision.id.desc()), - ) + 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'}" ) - .scalars() - .first() - ) - assert revision is not None - return revision - - -async def _run_task(mocker, task, approves) -> RecordingAuthorizationService: - """Run ``task`` under a recording service and hand it back for assertions.""" - recorder = RecordingAuthorizationService(approves) - mocker.patch(VALIDATOR_AUTH_SERVICE, lambda: recorder) - await task() - return recorder - - -def assert_write_authorized( - recorder: RecordingAuthorizationService, - username: str, - target: access.Resource = CREATE_TARGET, -) -> None: - """The task must have asked for WRITE on ``target``, acting as ``username``.""" - assert recorder.usernames == [username] - assert [(request.verb, request.access_object) for request in recorder.requests] == [ - (access.ResourceAction.WRITE, target), - ] @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 must hand its background tasks the namespace it authorized. + Creation hands its background tasks the namespace it authorized. - Pins the wiring rather than the helper: scheduling a node target would deny - the work under the namespace-scoped grant that allowed the create. + 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. """ - derive = mocker.patch("datajunction_server.internal.nodes.derive_frozen_measures") - lineage = mocker.patch( - "datajunction_server.internal.nodes.save_column_level_lineage", - ) - - await _create_metric(client, f"{METRIC}_again") + 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 derive.call_args.kwargs["access_target"] == CREATE_TARGET assert lineage.call_args.kwargs["access_target"] == CREATE_TARGET @@ -210,12 +198,9 @@ async def test_exact_namespace_grant_authorizes_create_background_work( mocker, ): """ - An exact namespace WRITE grant -- enough to create the node -- must also - satisfy the background work that create schedules. - - Runs the real RBAC matcher under a restrictive policy, since resource type is - what decides it: `bgauthz` does not match the node name `bgauthz.total_users`, - so authorizing the node would leave a created node without derived metadata. + 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", @@ -226,20 +211,20 @@ async def test_exact_namespace_grant_authorizes_create_background_work( role = Role(name="bgauthz-namespace-writer", created_by_id=current_user.id) session.add(role) await session.flush() - session.add( - RoleScope( - role_id=role.id, - action=access.ResourceAction.WRITE, - scope_type=access.ResourceType.NAMESPACE, - scope_value=NAMESPACE, - ), - ) - session.add( - RoleAssignment( - principal_id=current_user.id, - role_id=role.id, - granted_by_id=current_user.id, - ), + 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()) @@ -254,7 +239,7 @@ async def test_exact_namespace_grant_authorizes_create_background_work( @pytest.mark.asyncio -async def test_lineage_authorizes_the_create_target( +async def test_lineage_gated_on_the_create_target( metric_graph, session, current_user, @@ -264,58 +249,55 @@ async def test_lineage_authorizes_the_create_target( revision = await _revision(session, METRIC) revision.lineage = None await session.commit() - task = partial( - save_column_level_lineage, - node_revision_id=revision.id, - current_user=current_user, - access_target=CREATE_TARGET, - ) - denied = await _run_task(mocker, task, deny) - assert_write_authorized(denied, current_user.username) - assert not (await _revision(session, METRIC)).lineage, ( - "lineage was written despite WRITE being denied" - ) + async def ran() -> bool: + return bool((await _revision(session, METRIC)).lineage) - granted = await _run_task(mocker, task, allow) - assert_write_authorized(granted, current_user.username) - assert (await _revision(session, METRIC)).lineage, ( - "lineage was not written when WRITE was granted" + 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_authorize_the_create_target( +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. - - Asserts whether the derivation body is reached rather than inspecting the - return value, which is empty for several unrelated reasons. + 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=[], ) - task = partial( - derive_frozen_measures, - node_revision_id=revision.id, - current_user=current_user, - access_target=CREATE_TARGET, - ) - denied = await _run_task(mocker, task, deny) - assert_write_authorized(denied, current_user.username) - derive.assert_not_called() + async def ran() -> bool: + return derive.called - granted = await _run_task(mocker, task, allow) - assert_write_authorized(granted, current_user.username) - derive.assert_called_once() + 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 @@ -327,9 +309,8 @@ async def test_downstream_revalidation_not_gated_on_caller_write( ): """ A caller who may write only the upstream still gets downstream revalidation. - - Fails if a WRITE check is ever added to the propagation path -- which is the - point: such a check would break cross-namespace graphs. + 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 From df0bdf2fa7ed4eaf2463e99464ba2b4944b7d65a Mon Sep 17 00:00:00 2001 From: Rui Zhang <36744357+ruizhang0519@users.noreply.github.com> Date: Thu, 6 Aug 2026 15:44:18 -0400 Subject: [PATCH 6/6] Delete datajunction-server/scripts/migrate-measures.py --- .../scripts/migrate-measures.py | 61 ------------------- 1 file changed, 61 deletions(-) delete mode 100644 datajunction-server/scripts/migrate-measures.py diff --git a/datajunction-server/scripts/migrate-measures.py b/datajunction-server/scripts/migrate-measures.py deleted file mode 100644 index aafde9773f..0000000000 --- a/datajunction-server/scripts/migrate-measures.py +++ /dev/null @@ -1,61 +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_bulk -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") - # Operator tool with no request user, so it uses the system-facing - # bulk API (which also links the measures to each revision); the - # per-revision entry point authorizes a user for WRITE. - await derive_frozen_measures_bulk( - session, - [revision.id for revision in metric_revisions], - ) - await session.commit() - print(f"Derived frozen measures for {len(metric_revisions)} revisions") - - -if __name__ == "__main__": - asyncio.run(backfill_measures())