From 0d10c6412b6923627d190f50e3495804c96b04b4 Mon Sep 17 00:00:00 2001 From: Sam Welborn Date: Fri, 17 Jul 2026 10:47:13 -0400 Subject: [PATCH] end-to-end delegated-credential security boundary coverage --- tests/v2/test_credential_boundaries.py | 362 +++++++++++++++++++++++++ 1 file changed, 362 insertions(+) create mode 100644 tests/v2/test_credential_boundaries.py diff --git a/tests/v2/test_credential_boundaries.py b/tests/v2/test_credential_boundaries.py new file mode 100644 index 0000000..f413f74 --- /dev/null +++ b/tests/v2/test_credential_boundaries.py @@ -0,0 +1,362 @@ +"""Security boundaries for delegated credentials and durable execution.""" + +from collections.abc import Mapping +from contextlib import nullcontext +from datetime import UTC, datetime, timedelta +from inspect import signature +from pathlib import Path +from types import SimpleNamespace +from typing import Any, cast +from unittest.mock import ANY, AsyncMock +from uuid import UUID + +import pytest +from sqlalchemy.ext.asyncio import AsyncSession + +from lclstream_api.v2 import service, workflows +from lclstream_api.v2.auth import AuthenticatedUser +from lclstream_api.v2.clients.fastcache import FastcacheClient +from lclstream_api.v2.core import logs as lcore, producer as pcore, transfer as tcore +from lclstream_api.v2.exceptions import ( + CacheShutdownBlocked, + InsufficientTokenLifetime, + NotFound, +) +from lclstream_api.v2.models import CacheMode, TransferCancelOutcome, TransferState + +NOW = datetime(2026, 7, 16, 12, 0, tzinfo=UTC) +TRANSFER_ID = UUID("00000000-0000-0000-0000-000000000042") +RAW_TOKEN = "request-only-secret-token" + + +def _user( + *, + issuer: str = "https://issuer.example", + subject: str = "subject-1", + email: str = "user@example.org", + token: str = RAW_TOKEN, +) -> AuthenticatedUser: + return AuthenticatedUser( + issuer=issuer, + subject=subject, + email=email, + token=token, + expires_at=datetime.now(UTC) + timedelta(hours=12), + ) + + +def _assert_not_durable(value: object) -> None: + """Reject request credentials anywhere in a captured DBOS invocation.""" + + if isinstance(value, (tuple, list)): + for item in value: + _assert_not_durable(item) + return + if isinstance(value, dict): + for key, item in value.items(): + _assert_not_durable(key) + _assert_not_durable(item) + return + assert not isinstance(value, AuthenticatedUser) + assert value != RAW_TOKEN + + +def _await_args(mock: AsyncMock) -> tuple[tuple[Any, ...], Mapping[str, Any]]: + """Return an already-awaited mock's (args, kwargs), narrowed from Optional.""" + call = mock.await_args + assert call is not None + return call.args, call.kwargs + + +@pytest.fixture +def io(monkeypatch: pytest.MonkeyPatch, make_producer_settings) -> SimpleNamespace: + """Pre-patch every external collaborator `service.py`/`workflows.py` reach + through a module-level singleton (repo, IRI, fastcache, DBOS, config). + Each test sets ``return_value``/``side_effect`` only on the handles it + cares about; everything else keeps a harmless default.""" + mocks = SimpleNamespace( + insert_transfer=AsyncMock(), + get_owned_transfer=AsyncMock(return_value=None), + get_transfer=AsyncMock(return_value=None), + record_state=AsyncMock(return_value=True), + lock_active_cache=AsyncMock(return_value=None), + count_active_transfers_by_cache=AsyncMock(return_value=0), + retire_cache=AsyncMock(), + start_workflow_async=AsyncMock(), + cancel_job=AsyncMock(), + tail=AsyncMock(), + submit_job=AsyncMock(), + delete_cache=AsyncMock(), + token_for=AsyncMock(return_value=RAW_TOKEN), + ) + for name in ( + "insert_transfer", + "get_owned_transfer", + "get_transfer", + "record_state", + "lock_active_cache", + "count_active_transfers_by_cache", + "retire_cache", + ): + monkeypatch.setattr(service.repo, name, getattr(mocks, name)) + monkeypatch.setattr(service, "uuid4", lambda: TRANSFER_ID) + monkeypatch.setattr(service, "SetWorkflowID", lambda _: nullcontext()) + monkeypatch.setattr(service.config, "get_producer", make_producer_settings) + monkeypatch.setattr( + service.config, + "get_credentials", + lambda: SimpleNamespace(lifecycle_grace_seconds=0), + ) + monkeypatch.setattr( + service.DBOS, "start_workflow_async", mocks.start_workflow_async + ) + # service.iri and workflows.iri are the same module (`clients.iri`) — one + # patch on its `client` factory covers every caller. + monkeypatch.setattr( + service.iri, + "client", + lambda: SimpleNamespace( + cancel_job=mocks.cancel_job, + tail=mocks.tail, + submit_job=mocks.submit_job, + ), + ) + monkeypatch.setattr( + service.fastcache, + "client", + lambda: SimpleNamespace(delete_cache=mocks.delete_cache), + ) + monkeypatch.setattr(workflows, "_token_for", mocks.token_for) + return mocks + + +@pytest.mark.asyncio +async def test_transfer_creation_starts_workflow_without_request_credential( + io: SimpleNamespace, fake_session: AsyncSession, make_params +) -> None: + user = _user() + + async def insert_transfer(_session, **values): + return SimpleNamespace( + id=values["transfer_id"], + created_at=NOW, + owner_email=values["owner_email"], + state=TransferState.provisioning, + experiment=values["experiment"], + run=values["run"], + cache_mode=values["cache_mode"], + cache_hostname=None, + push_port=None, + ) + + io.insert_transfer.side_effect = insert_transfer + + await service.create_transfer( + fake_session, user, make_params(), experiment="mfxl1001", run="42" + ) + + io.start_workflow_async.assert_awaited_once_with( + workflows.provision_transfer, TRANSFER_ID + ) + args, kwargs = _await_args(io.start_workflow_async) + _assert_not_durable(args) + _assert_not_durable(kwargs) + + +@pytest.mark.asyncio +async def test_transfer_creation_rejects_short_lived_token_before_persistence( + io: SimpleNamespace, fake_session: AsyncSession, make_params +) -> None: + user = AuthenticatedUser( + issuer="https://issuer.example", + subject="subject-1", + email="user@example.org", + token=RAW_TOKEN, + expires_at=datetime.now(UTC) + timedelta(seconds=100), + ) + + with pytest.raises(InsufficientTokenLifetime) as exc_info: + await service.create_transfer( + fake_session, user, make_params(), experiment="mfxl1001", run="42" + ) + + assert exc_info.value.required_seconds == 3600 + assert exc_info.value.remaining_seconds <= 100 + io.start_workflow_async.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_non_owner_cancel_cannot_reuse_owner_credential( + io: SimpleNamespace, fake_session: AsyncSession +) -> None: + caller = _user(subject="intruder") + + with pytest.raises(NotFound): + await service.cancel_transfer(fake_session, TRANSFER_ID, caller) + + io.get_owned_transfer.assert_awaited_once_with( + ANY, TRANSFER_ID, owner_issuer=caller.issuer, owner_subject=caller.subject + ) + io.cancel_job.assert_not_awaited() + io.start_workflow_async.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_owner_cancel_uses_request_token_but_not_in_workflow_args( + io: SimpleNamespace, fake_session: AsyncSession +) -> None: + owner = _user() + io.get_owned_transfer.return_value = SimpleNamespace( + state=TransferState.ready, + producer_job_id="producer-42", + cache_id=None, + cache_mode=CacheMode.per_transfer, + ) + + outcome = await service.cancel_transfer(fake_session, TRANSFER_ID, owner) + + assert outcome is TransferCancelOutcome.canceling + io.cancel_job.assert_awaited_once_with("producer-42", RAW_TOKEN) + assert io.start_workflow_async.await_count == 1 + args, kwargs = _await_args(io.start_workflow_async) + _assert_not_durable(args) + _assert_not_durable(kwargs) + + +@pytest.mark.asyncio +async def test_log_read_uses_current_caller_token_not_owner_credential( + io: SimpleNamespace, fake_session: AsyncSession +) -> None: + caller = _user(subject="observer", token="observer-current-token") + io.get_transfer.return_value = SimpleNamespace( + experiment="mfxl1001", + run="42", + cache_mode=CacheMode.per_transfer, + owner_subject="different-owner", + ) + io.tail.return_value = "log output" + + result = await service.read_transfer_log( + fake_session, TRANSFER_ID, lcore.LogStream.producer_stdout, caller, lines=20 + ) + + assert result == "log output" + path, token = _await_args(io.tail)[0] + assert isinstance(path, Path) + assert token == "observer-current-token" + assert _await_args(io.tail)[1] == {"lines": 20, "bytes_": None} + + +@pytest.mark.asyncio +async def test_iri_step_resolves_token_internally_and_returns_only_result( + io: SimpleNamespace, +) -> None: + io.submit_job.return_value = "producer-42" + + owner = tcore.Principal(issuer="https://issuer.example", subject="subject-1") + result = await workflows._submit_producer(pcore.DEFAULT_JOB_SPEC, owner) + + assert result == "producer-42" + assert result != RAW_TOKEN + io.token_for.assert_awaited_once_with(owner) + io.submit_job.assert_awaited_once_with(pcore.DEFAULT_JOB_SPEC, RAW_TOKEN) + + iri_steps = ( + workflows._submit_producer, + workflows._upload_config, + workflows._delete_config, + workflows._delete_work_dir, + workflows._get_producer_state, + workflows._cancel_producer, + ) + for step in iri_steps: + parameters = signature(step).parameters + assert "token" not in parameters + assert "user" not in parameters + + +def test_fastcache_control_plane_has_no_bearer_credential_parameter() -> None: + for method in ( + FastcacheClient.create_cache, + FastcacheClient.get_cache, + FastcacheClient.delete_cache, + ): + parameters = signature(method).parameters + assert "token" not in parameters + assert "authorization" not in parameters + + +@pytest.mark.asyncio +async def test_shutdown_rejects_unknown_or_non_shared_cache( + io: SimpleNamespace, fake_session: AsyncSession +) -> None: + with pytest.raises(NotFound): + await service.shutdown_cache(fake_session, TRANSFER_ID) + + io.delete_cache.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_shutdown_blocks_active_shared_cache_under_registry_lock( + io: SimpleNamespace, fake_session: AsyncSession +) -> None: + io.lock_active_cache.return_value = SimpleNamespace(id=TRANSFER_ID, retired_at=None) + io.count_active_transfers_by_cache.return_value = 2 + + with pytest.raises(CacheShutdownBlocked): + await service.shutdown_cache(fake_session, TRANSFER_ID) + + io.delete_cache.assert_not_awaited() + io.retire_cache.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_shutdown_retires_shared_cache_in_delete_transaction( + io: SimpleNamespace, fake_session: AsyncSession +) -> None: + cache = SimpleNamespace(id=TRANSFER_ID, retired_at=None) + io.lock_active_cache.return_value = cache + + await service.shutdown_cache(fake_session, TRANSFER_ID) + + io.delete_cache.assert_awaited_once_with(TRANSFER_ID) + io.retire_cache.assert_awaited_once() + args, kwargs = _await_args(io.retire_cache) + assert args == (ANY, cache) + assert kwargs["retired_at"].tzinfo is UTC + + +@pytest.mark.asyncio +async def test_shared_cache_attachment_rejects_retired_registry_row() -> None: + transfer = SimpleNamespace( + cache_mode=CacheMode.shared, + experiment="mfxl1001", + cache_id=None, + cache_hostname=None, + pull_port=None, + push_port=None, + ) + retired = SimpleNamespace( + experiment="mfxl1001", + retired_at=NOW, + ) + result = SimpleNamespace(scalar_one=lambda: retired) + session = cast( + AsyncSession, + SimpleNamespace( + get=AsyncMock(return_value=transfer), + execute=AsyncMock(side_effect=[None, result]), + ), + ) + + with pytest.raises(LookupError, match="has been shut down"): + await service.repo.set_cache_endpoints( + session, + TRANSFER_ID, + cache_id=TRANSFER_ID, + hostname="dtn.example", + pull_port=5000, + push_port=5001, + ) + + assert transfer.cache_id is None