From 7af7f477806f4add78e47e1b598edaea5722015b Mon Sep 17 00:00:00 2001 From: lilydu Date: Tue, 25 Aug 2026 16:57:47 -0700 Subject: [PATCH 01/10] feat(oauth): deduplicate signin/tokenExchange callbacks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Teams fans `signin/tokenExchange` out to every signed-in client endpoint, so the same logical exchange reaches the bot several times. Without dedup each copy ran a full exchange, re-emitted `sign_in`, re-invoked the flow's success handlers, and re-ran the middleware chain. Two cooperating layers, mirroring the C# and TypeScript SDKs: An in-flight guard keyed by exchange id. The first request installs a future and owns the exchange; concurrent duplicates await it and mirror its result, so a caller that lost the race still learns the exchange failed instead of being told the sign-in succeeded. The claim sequence holds no `await`, which is what makes check-and-insert atomic on a single event loop. A completed marker for duplicates that arrive after the original settles. Held in memory (5-minute TTL, capped at 1000 entries) and persisted to conversation state under the reserved `__oauth:exchange:{id}` key, so a duplicate handled by another process instance short-circuits too. Turn state is last-write-wins with no compare-and-set, so that cross-instance layer is best-effort; the in-memory guard stays authoritative for the same instance. C# carries the identical caveat. The marker is never cleared on completion — a late duplicate from a second endpoint arrives after the exchange finishes, and a cleared marker would let it run anew. Deduplicated requests return a 200 no-op and skip `ctx.next()`, so the sign-in side effects and the rest of the middleware chain run exactly once per exchange. The owning request keeps PR4's guarantee that `next` runs on every path. `signin/verifyState` and `signin/failure` are left undeduplicated: the verify code is single-use and failure is a single informational notice. An exchange with no id is run undeduplicated rather than collapsed onto a shared empty key, which would drop unrelated sign-ins. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../src/microsoft_teams/apps/app_oauth.py | 221 ++++- .../apps/diagnostics/_constants.py | 1 + .../src/microsoft_teams/apps/oauth_state.py | 94 +++ packages/apps/tests/test_app_oauth.py | 18 +- packages/apps/tests/test_app_oauth_dedup.py | 780 ++++++++++++++++++ 5 files changed, 1104 insertions(+), 10 deletions(-) create mode 100644 packages/apps/tests/test_app_oauth_dedup.py diff --git a/packages/apps/src/microsoft_teams/apps/app_oauth.py b/packages/apps/src/microsoft_teams/apps/app_oauth.py index a579549c..971865ce 100644 --- a/packages/apps/src/microsoft_teams/apps/app_oauth.py +++ b/packages/apps/src/microsoft_teams/apps/app_oauth.py @@ -3,9 +3,12 @@ Licensed under the MIT License. """ +import asyncio import logging -from time import perf_counter -from typing import Optional, Union +from asyncio import Future +from dataclasses import dataclass +from time import perf_counter, time +from typing import Dict, Optional, Union from httpx import HTTPStatusError from microsoft_teams.api import ( @@ -32,10 +35,30 @@ from .events import ErrorEvent, EventType, SignInEvent, SignInFailureEvent from .oauth_connection import connection_lookup_key from .oauth_flow import OAuthFlow, OAuthFlowRegistry +from .oauth_state import ( + TOKEN_EXCHANGE_DEDUP_TTL_SECONDS, + has_completed_token_exchange, + record_completed_token_exchange, +) from .routing import ActivityContext logger = logging.getLogger(__name__) +TokenExchangeResult = Union[TokenExchangeInvokeResponseType, InvokeResponse[TokenExchangeInvokeResponseType]] + +_TOKEN_EXCHANGE_DEDUP_MAX_ENTRIES = 1000 +"""Cap on the in-memory completed-marker set, so a long-lived process cannot grow it +without bound even if entries are added faster than the TTL retires them.""" + + +@dataclass(frozen=True) +class _TokenExchangeOutcome: + """How an owned token exchange settled, replayed to concurrent duplicates.""" + + response: TokenExchangeResult = None + error: Optional[BaseException] = None + token_redeemed: bool = False + class OauthHandlers: def __init__( @@ -47,13 +70,66 @@ def __init__( self.default_connection_name = default_connection_name self.event_emitter = event_emitter self.oauth_registry = oauth_registry + # Dedup bookkeeping is per-``App`` instance state rather than module-level + # globals, so two apps in one process (and two tests in one session) never + # short-circuit each other's exchanges. + self._token_exchange_in_flight: Dict[str, Future[_TokenExchangeOutcome]] = {} + self._token_exchange_completed: Dict[str, float] = {} async def sign_in_token_exchange( self, ctx: ActivityContext[SignInTokenExchangeInvokeActivity] - ) -> Union[TokenExchangeInvokeResponseType, InvokeResponse[TokenExchangeInvokeResponseType]]: - """ - Decorator to register a function that handles the sign-in token exchange. + ) -> TokenExchangeResult: + """Handle ``signin/tokenExchange``, deduplicating repeats of the same exchange. + + Teams fans the same exchange out to every signed-in client endpoint, so the + bot can see it several times. Duplicates short-circuit to a ``200`` no-op and + deliberately skip ``ctx.next()``: the whole point of dedup is that the sign-in + side effects — the ``sign_in`` event, the flow callbacks, and the rest of the + middleware chain — run exactly once per exchange. """ + exchange_id = ctx.activity.value.id + if not exchange_id: + # Teams normally stamps every exchange with an id. Without one there is + # nothing safe to key on, and collapsing every id-less exchange onto a + # shared empty key would drop unrelated sign-ins, so run undeduplicated. + return await self._run_token_exchange(ctx) + + # Claim the exchange. The in-flight lookup, the in-flight insert and the + # completed-marker check below contain no ``await``, so the event loop cannot + # switch coroutines part way through: on a single loop this claim is atomic, + # which is what stops two concurrent duplicates from both starting an exchange. + # + # In-flight is checked first so a running exchange always wins over its own + # completed marker. The marker is stamped the moment the token is redeemed, + # while the owner still has sign-in callbacks to run, and a duplicate that + # landed in that window must mirror the owner rather than answer ahead of it. + in_flight = self._token_exchange_in_flight.get(exchange_id) + if in_flight is not None: + return await self._await_token_exchange(ctx, exchange_id, in_flight) + if self._is_completed_token_exchange(ctx, exchange_id): + return self._replay_completed_token_exchange(ctx, exchange_id) + owned: Future[_TokenExchangeOutcome] = asyncio.get_running_loop().create_future() + self._token_exchange_in_flight[exchange_id] = owned + + try: + response = await self._run_token_exchange(ctx) + except BaseException as error: + # ``BaseException`` so cancellation also releases the entry and wakes + # waiters, instead of leaking the id until the process restarts. + self._settle_token_exchange(exchange_id, owned, _TokenExchangeOutcome(error=error)) + raise + self._settle_token_exchange( + exchange_id, + owned, + _TokenExchangeOutcome( + response=response, + token_redeemed=exchange_id in self._token_exchange_completed, + ), + ) + return response + + async def _run_token_exchange(self, ctx: ActivityContext[SignInTokenExchangeInvokeActivity]) -> TokenExchangeResult: + """Perform the exchange itself for the request that owns this exchange id.""" activity = ctx.activity api = ctx.api next_handler = ctx.next @@ -149,6 +225,11 @@ async def sign_in_token_exchange( span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_result, result) raise + # Recorded as soon as the exchange succeeds, before any sign-in side + # effects: the exchange token is spent at this point, so a retry could + # never succeed anyway. A failed exchange is never marked, leaving the + # id free for a genuine retry. + self._record_completed_token_exchange(ctx, activity.value.id) ctx.is_signed_in = True ctx.user_token = token.token self.oauth_registry._clear_pending( # pyright: ignore[reportPrivateUsage] @@ -175,6 +256,136 @@ async def sign_in_token_exchange( ) await next_handler() + def _is_completed_token_exchange( + self, ctx: ActivityContext[SignInTokenExchangeInvokeActivity], exchange_id: str + ) -> bool: + self._prune_completed_token_exchanges() + if exchange_id in self._token_exchange_completed: + return True + return has_completed_token_exchange(ctx.state, exchange_id) + + def _record_completed_token_exchange( + self, ctx: ActivityContext[SignInTokenExchangeInvokeActivity], exchange_id: str + ) -> None: + if not exchange_id: + return + + self._prune_completed_token_exchanges() + self._token_exchange_completed[exchange_id] = time() + while len(self._token_exchange_completed) > _TOKEN_EXCHANGE_DEDUP_MAX_ENTRIES: + del self._token_exchange_completed[next(iter(self._token_exchange_completed))] + # Persisted too, so a duplicate handled by another process instance still sees + # it. Best-effort only: state has no compare-and-set, so the in-memory layer + # above remains the authoritative same-instance guard. + record_completed_token_exchange(ctx.state, exchange_id) + + def _prune_completed_token_exchanges(self) -> None: + cutoff = time() - TOKEN_EXCHANGE_DEDUP_TTL_SECONDS + expired = [ + exchange_id + for exchange_id, completed_at in self._token_exchange_completed.items() + if completed_at <= cutoff + ] + for exchange_id in expired: + del self._token_exchange_completed[exchange_id] + + def _settle_token_exchange( + self, exchange_id: str, owned: Future[_TokenExchangeOutcome], outcome: _TokenExchangeOutcome + ) -> None: + self._token_exchange_in_flight.pop(exchange_id, None) + if not owned.done(): + owned.set_result(outcome) + + def _stamp_completed_token_exchange( + self, ctx: ActivityContext[SignInTokenExchangeInvokeActivity], exchange_id: str + ) -> None: + """Copy the completed marker into a deduplicated request's own turn state. + + Each turn loads its own state snapshot, and saves are last-write-wins with no + compare-and-set. A duplicate whose snapshot predates the owner's write would + otherwise erase that freshly persisted marker when its own save lands last. + Only the in-memory marker is left alone here, so the TTL stays anchored to the + moment the token was actually redeemed rather than being extended by every + duplicate that arrives. + """ + if not has_completed_token_exchange(ctx.state, exchange_id): + record_completed_token_exchange(ctx.state, exchange_id) + + def _replay_completed_token_exchange( + self, ctx: ActivityContext[SignInTokenExchangeInvokeActivity], exchange_id: str + ) -> TokenExchangeResult: + """Answer a duplicate that arrived after its exchange already completed.""" + logger.debug("Duplicate signin/tokenExchange with id '%s' - returning 200 no-op.", exchange_id) + self._stamp_completed_token_exchange(ctx, exchange_id) + connection_name = ctx.activity.value.connection_name + started_at = perf_counter() + try: + with get_tracer().start_as_current_span( + APP_SPAN_NAMES.oauth_token_exchange, + record_exception=False, + set_status_on_exception=False, + ) as span: + span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_connection, connection_name) + span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_operation, APP_OAUTH_OPERATIONS.token_exchange) + span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_result, APP_OAUTH_RESULTS.duplicate) + span.set_attribute(APP_ATTRIBUTE_NAMES.invoke_response_status, 200) + return InvokeResponse(status=200) + finally: + record_oauth_operation( + connection_name, + APP_OAUTH_OPERATIONS.token_exchange, + APP_OAUTH_RESULTS.duplicate, + (perf_counter() - started_at) * 1000, + ) + + async def _await_token_exchange( + self, + ctx: ActivityContext[SignInTokenExchangeInvokeActivity], + exchange_id: str, + in_flight: Future[_TokenExchangeOutcome], + ) -> TokenExchangeResult: + """Answer a duplicate that arrived while its exchange is still running. + + The waiter mirrors whatever the owning request produced, so a caller that lost + the race still learns that the exchange failed (``412``) instead of being told + the sign-in succeeded. + """ + connection_name = ctx.activity.value.connection_name + result = APP_OAUTH_RESULTS.duplicate + started_at = perf_counter() + try: + with get_tracer().start_as_current_span( + APP_SPAN_NAMES.oauth_token_exchange, + record_exception=False, + set_status_on_exception=False, + ) as span: + span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_connection, connection_name) + span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_operation, APP_OAUTH_OPERATIONS.token_exchange) + # Shielded: cancelling this waiter must not cancel the future the + # owning request still has to resolve. + outcome = await asyncio.shield(in_flight) + if outcome.error is not None: + result = APP_OAUTH_RESULTS.failure + span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_result, result) + raise outcome.error + if outcome.token_redeemed: + self._stamp_completed_token_exchange(ctx, exchange_id) + response = outcome.response + # The owning request signals success by returning ``None``, which the + # activity processor materializes as a 200. Duplicates say so + # explicitly instead, matching the C# SDK's 200 no-op. + status = response.status if isinstance(response, InvokeResponse) else 200 + span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_result, result) + span.set_attribute(APP_ATTRIBUTE_NAMES.invoke_response_status, status) + return response if isinstance(response, InvokeResponse) else InvokeResponse(status=200) + finally: + record_oauth_operation( + connection_name, + APP_OAUTH_OPERATIONS.token_exchange, + result, + (perf_counter() - started_at) * 1000, + ) + async def sign_in_failure( self, ctx: ActivityContext[SignInFailureInvokeActivity] ) -> Optional[InvokeResponse[None]]: diff --git a/packages/apps/src/microsoft_teams/apps/diagnostics/_constants.py b/packages/apps/src/microsoft_teams/apps/diagnostics/_constants.py index 8fa0cab6..3827c8fb 100644 --- a/packages/apps/src/microsoft_teams/apps/diagnostics/_constants.py +++ b/packages/apps/src/microsoft_teams/apps/diagnostics/_constants.py @@ -81,6 +81,7 @@ class _AppOAuthOperations: @dataclass(frozen=True) class _AppOAuthResults: + duplicate: str = "duplicate" failure: str = "failure" no_token: str = "no_token" notified: str = "notified" diff --git a/packages/apps/src/microsoft_teams/apps/oauth_state.py b/packages/apps/src/microsoft_teams/apps/oauth_state.py index 1190c3b8..c1648720 100644 --- a/packages/apps/src/microsoft_teams/apps/oauth_state.py +++ b/packages/apps/src/microsoft_teams/apps/oauth_state.py @@ -32,6 +32,19 @@ _PENDING_OAUTH_MAX_AGE_SECONDS = 5 * 60 _PENDING_OAUTH_MAX_CLOCK_SKEW_SECONDS = 60 +# Reserved conversation-state key prefix holding the completed marker for a single +# ``signin/tokenExchange``. The full key is ``__oauth:exchange:{id}`` and the value is an +# ISO 8601 UTC timestamp, mirroring the C# SDK (``OAuthFlow.cs``) so all three SDKs +# describe this state identically. That is design parity, not wire compatibility: the +# SDKs encode the enclosing scope key differently, so they never resolve to the same +# stored document. Conversation scope rather than user scope, because duplicates arrive +# from several of the user's clients but always on the same conversation. Treat the key +# as private: app code should neither read nor write it. +_COMPLETED_EXCHANGE_STATE_KEY_PREFIX = "__oauth:exchange:" + +# Completed markers age out on the same schedule as pending sign-ins. +TOKEN_EXCHANGE_DEDUP_TTL_SECONDS = _PENDING_OAUTH_MAX_AGE_SECONDS + @dataclass(frozen=True) class PendingOAuthSignIn: @@ -280,3 +293,84 @@ def _parse_timestamp(raw: Any) -> Optional[float]: if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) return parsed.timestamp() + + +def completed_token_exchange_state_key(exchange_id: str) -> str: + """Reserved conversation-state key holding the completed marker for ``exchange_id``.""" + return f"{_COMPLETED_EXCHANGE_STATE_KEY_PREFIX}{exchange_id}" + + +def has_completed_token_exchange(state: Optional[TurnStateContainer], exchange_id: str) -> bool: + """Whether ``exchange_id`` has already been redeemed on this conversation. + + Read from conversation state so a duplicate routed to a different process instance + still short-circuits. Turn state is a last-write-wins document with no ETag/CAS in + ``Storage``, so this cross-instance layer is best-effort; the caller's in-process + guard is what makes same-instance dedup atomic. The C# SDK carries the same caveat. + """ + if state is None or not exchange_id: + return False + + key = completed_token_exchange_state_key(exchange_id) + raw = state.conversation.get(key) + if raw is None: + return False + + completed_at = _parse_completed_at(raw) + if completed_at is None: + logger.warning("Discarding malformed completed OAuth token exchange state.") + state.conversation.pop(key, None) + return False + if time() - completed_at > TOKEN_EXCHANGE_DEDUP_TTL_SECONDS: + state.conversation.pop(key, None) + return False + return True + + +def record_completed_token_exchange(state: Optional[TurnStateContainer], exchange_id: str) -> None: + """Persist the completed marker for ``exchange_id``. + + The marker is deliberately never cleared when the exchange finishes: a late + duplicate from a second Teams endpoint can arrive after the original settles, and + an already-removed marker would let it run as a brand new exchange. Markers are + pruned only once they age past :data:`TOKEN_EXCHANGE_DEDUP_TTL_SECONDS`. + """ + if state is None or not exchange_id: + return + _write_completed_token_exchange(state, exchange_id) + + +def _write_completed_token_exchange(state: TurnStateContainer, exchange_id: str) -> None: + """Single write chokepoint for completed markers. + + Pruning here keeps the invariant that the conversation document never carries an + expired or unparsable marker, no matter which caller wrote it. + """ + _prune_completed_token_exchanges(state) + state.conversation[completed_token_exchange_state_key(exchange_id)] = _format_timestamp(time()) + + +def _prune_completed_token_exchanges(state: TurnStateContainer) -> None: + """Drop expired or malformed markers so the conversation document stays bounded.""" + now = time() + for key in list(state.conversation): + if not key.startswith(_COMPLETED_EXCHANGE_STATE_KEY_PREFIX): + continue + completed_at = _parse_completed_at(state.conversation.get(key)) + if completed_at is None or now - completed_at > TOKEN_EXCHANGE_DEDUP_TTL_SECONDS: + state.conversation.pop(key, None) + + +def _parse_completed_at(raw: Any) -> Optional[float]: + """Parse a stored marker, rejecting timestamps too far in the future. + + ``_parse_timestamp`` already rejects anything that is not a parseable ISO 8601 + string. The extra guard is for clock skew between instances: a marker stamped well + ahead of this instance's clock would otherwise read as fresh long past its TTL. + """ + completed_at = _parse_timestamp(raw) + if completed_at is None: + return None + if completed_at > time() + _PENDING_OAUTH_MAX_CLOCK_SKEW_SECONDS: + return None + return completed_at diff --git a/packages/apps/tests/test_app_oauth.py b/packages/apps/tests/test_app_oauth.py index a32c4606..ac3d7cc7 100644 --- a/packages/apps/tests/test_app_oauth.py +++ b/packages/apps/tests/test_app_oauth.py @@ -1005,9 +1005,14 @@ async def still_runs(_): mock_context.next.assert_awaited_once() @pytest.mark.asyncio - async def test_duplicate_token_exchange_is_not_deduplicated_in_pr4( + async def test_duplicate_token_exchange_runs_sign_in_side_effects_once( self, oauth_handlers, mock_context, token_exchange_activity, mock_token_response ): + """A repeat of the same exchange id is a 200 no-op. + + Broader dedup coverage lives in ``test_app_oauth_dedup.py``; this guards the + interaction with the PR4 routing path that resolves the flow. + """ flow = oauth_handlers.oauth_registry.add(OAuthFlow("test-connection")) callback_count = 0 @@ -1019,11 +1024,14 @@ async def on_signin(_): mock_context.activity = token_exchange_activity mock_context.api.users.exchange_token.return_value = mock_token_response - await oauth_handlers.sign_in_token_exchange(mock_context) - await oauth_handlers.sign_in_token_exchange(mock_context) + assert await oauth_handlers.sign_in_token_exchange(mock_context) is None + duplicate = await oauth_handlers.sign_in_token_exchange(mock_context) - assert mock_context.api.users.exchange_token.await_count == 2 - assert callback_count == 2 + assert isinstance(duplicate, InvokeResponse) + assert duplicate.status == 200 + assert mock_context.api.users.exchange_token.await_count == 1 + assert callback_count == 1 + assert mock_context.next.await_count == 1 @pytest.mark.asyncio async def test_verify_state_routes_to_pending_non_default_flow( diff --git a/packages/apps/tests/test_app_oauth_dedup.py b/packages/apps/tests/test_app_oauth_dedup.py new file mode 100644 index 00000000..ea9081bb --- /dev/null +++ b/packages/apps/tests/test_app_oauth_dedup.py @@ -0,0 +1,780 @@ +""" +Copyright (c) Microsoft Corporation. All rights reserved. +Licensed under the MIT License. + +Teams fans ``signin/tokenExchange`` out to every signed-in client endpoint, so the +same exchange reaches the bot several times. These tests pin down the resulting +deduplication contract end to end through the public handler entry points. +""" + +import asyncio +import logging +from datetime import datetime, timezone +from typing import Any, List, Optional +from unittest.mock import AsyncMock, MagicMock + +import pytest +from httpx import HTTPStatusError, Request, Response +from microsoft_teams.api import ( + InvokeResponse, + SignInFailureInvokeActivity, + SignInTokenExchangeInvokeActivity, + SignInVerifyStateInvokeActivity, +) +from microsoft_teams.api.models import ( + Account, + ConversationAccount, + SignInExchangeToken, + SignInFailure, + SignInStateVerifyQuery, + TokenResponse, +) +from microsoft_teams.apps.app_oauth import OauthHandlers +from microsoft_teams.apps.oauth_flow import OAuthFlow, OAuthFlowRegistry +from microsoft_teams.apps.routing import ActivityContext +from microsoft_teams.apps.state import TurnState, TurnStateContainer +from microsoft_teams.common import EventEmitter + +# pyright: basic + +DEDUP_TTL_SECONDS = 5 * 60 +"""Mirror of the production TTL. Duplicated rather than imported so a silent change +to the production value shows up here as a failure instead of passing vacuously.""" + +DEDUP_MAX_ENTRIES = 1000 +"""Mirror of the production cap on the in-memory completed-marker set.""" + +CONNECTION_NAME = "test-connection" +EXCHANGE_STATE_KEY_PREFIX = "__oauth:exchange:" + + +def iso(epoch_seconds: float) -> str: + """Render an epoch as the ISO 8601 UTC string a marker is stored as. + + Mirrors the production storage boundary rather than importing it, so a change to + the persisted format surfaces here as a failure instead of passing vacuously. + """ + return datetime.fromtimestamp(epoch_seconds, tz=timezone.utc).isoformat() + + +def token_response() -> TokenResponse: + return TokenResponse(connection_name=CONNECTION_NAME, token="access-token", expiration="2024-12-31T23:59:59Z") + + +def oauth_http_error(status: int, message: str = "boom") -> HTTPStatusError: + request = Request("GET", "https://token.example") + response = Response(status, request=request) + return HTTPStatusError(message, request=request, response=response) + + +def exchange_activity(exchange_id: str = "exchange-1", token: str = "sso-token") -> SignInTokenExchangeInvokeActivity: + return SignInTokenExchangeInvokeActivity( + type="invoke", + id="activity-789", + from_=Account(id="user-123", name="Test User", role="user"), + recipient=Account(id="bot-456", name="Test Bot", role="bot"), + conversation=ConversationAccount(id="conv-456", conversation_type="personal"), + channel_id="msteams", + name="signin/tokenExchange", + value=SignInExchangeToken(id=exchange_id, connection_name=CONNECTION_NAME, token=token), + ) + + +def verify_state_activity() -> SignInVerifyStateInvokeActivity: + return SignInVerifyStateInvokeActivity( + type="invoke", + id="activity-789", + from_=Account(id="user-123", name="Test User", role="user"), + recipient=Account(id="bot-456", name="Test Bot", role="bot"), + conversation=ConversationAccount(id="conv-456", conversation_type="personal"), + channel_id="msteams", + name="signin/verifyState", + value=SignInStateVerifyQuery(state="verify-code"), + ) + + +def failure_activity() -> SignInFailureInvokeActivity: + return SignInFailureInvokeActivity( + type="invoke", + id="activity-789", + from_=Account(id="user-123", name="Test User", role="user"), + recipient=Account(id="bot-456", name="Test Bot", role="bot"), + conversation=ConversationAccount(id="conv-456", conversation_type="personal"), + channel_id="msteams", + name="signin/failure", + value=SignInFailure(code="invokeerror", message="nope"), + ) + + +def make_api(*, exchange: Any = None, get_token: Any = None) -> MagicMock: + """A stand-in ``ctx.api``. Shared between contexts so call counts span requests.""" + api = MagicMock() + api.users.exchange_token = exchange if exchange is not None else AsyncMock(return_value=token_response()) + api.users.get_token = get_token if get_token is not None else AsyncMock(return_value=token_response()) + return api + + +def slow_exchange(*outcomes: Any, delay: float = 0.01) -> AsyncMock: + """An ``exchange_token`` mock that yields to the event loop before answering. + + The suspension is what lets a second concurrent request reach the dedup gate while + the first exchange is still in flight. + """ + queued: List[Any] = list(outcomes) or [token_response()] + + async def _exchange(_params: Any) -> Any: + await asyncio.sleep(delay) + outcome = queued.pop(0) if len(queued) > 1 else queued[0] + if isinstance(outcome, BaseException): + raise outcome + return outcome + + return AsyncMock(side_effect=_exchange) + + +def make_context(activity: Any, api: MagicMock, state: Optional[TurnStateContainer] = None) -> MagicMock: + ctx = MagicMock(spec=ActivityContext) + ctx.activity = activity + ctx.api = api + ctx.logger = MagicMock() + ctx.next = AsyncMock() + ctx.state = state + return ctx + + +def make_state() -> TurnStateContainer: + return TurnStateContainer( + conversation=TurnState(), + conversation_id="conv-456", + user=TurnState(), + user_id="user-123", + ) + + +def make_handlers() -> tuple[OauthHandlers, MagicMock, OAuthFlow]: + emitter = MagicMock(spec=EventEmitter) + registry = OAuthFlowRegistry() + flow = registry.add(OAuthFlow(CONNECTION_NAME)) + return OauthHandlers(CONNECTION_NAME, emitter, registry), emitter, flow + + +def status_of(result: Any) -> int: + """The status Teams ends up seeing. + + The owning request signals success by returning ``None``, which the activity + processor materializes as a 200; duplicates return that 200 explicitly. + """ + return result.status if isinstance(result, InvokeResponse) else 200 + + +def emitted(emitter: MagicMock, name: str) -> List[Any]: + return [call.args[1] for call in emitter.emit_async.await_args_list if call.args[0] == name] + + +class TestConcurrentTokenExchangeDedup: + @pytest.mark.asyncio + async def test_concurrent_duplicates_run_sign_in_side_effects_exactly_once(self): + handlers, emitter, flow = make_handlers() + signin_calls: List[str] = [] + + @flow.on_signin + async def on_signin(event): + signin_calls.append(event.connection_name) + + api = make_api(exchange=slow_exchange()) + first = make_context(exchange_activity(), api) + second = make_context(exchange_activity(), api) + + results = await asyncio.gather( + handlers.sign_in_token_exchange(first), + handlers.sign_in_token_exchange(second), + ) + + assert [status_of(result) for result in results] == [200, 200] + assert api.users.exchange_token.await_count == 1 + assert len(emitted(emitter, "sign_in")) == 1 + assert signin_calls == [CONNECTION_NAME] + # The winner runs the middleware chain; the duplicate is a no-op, so `next` + # fires once across the pair rather than once per request. + assert sorted([first.next.await_count, second.next.await_count]) == [0, 1] + + @pytest.mark.asyncio + async def test_concurrent_duplicates_mirror_the_original_failure(self): + handlers, emitter, flow = make_handlers() + + @flow.on_signin + async def on_signin(_event): + pytest.fail("sign-in handlers must not run when the exchange fails") + + api = make_api(exchange=slow_exchange(oauth_http_error(400, "bad exchange"))) + first = make_context(exchange_activity("exchange-fail"), api) + second = make_context(exchange_activity("exchange-fail"), api) + + results = await asyncio.gather( + handlers.sign_in_token_exchange(first), + handlers.sign_in_token_exchange(second), + ) + + assert api.users.exchange_token.await_count == 1 + assert [status_of(result) for result in results] == [412, 412] + for result in results: + assert isinstance(result, InvokeResponse) + assert result.body is not None + assert result.body.id == "exchange-fail" + assert result.body.connection_name == CONNECTION_NAME + assert emitted(emitter, "sign_in") == [] + + @pytest.mark.asyncio + async def test_concurrent_duplicate_sees_the_same_handler_exception(self): + handlers, emitter, flow = make_handlers() + + @flow.on_signin + async def on_signin(_event): + raise RuntimeError("handler failed") + + api = make_api(exchange=slow_exchange()) + first = make_context(exchange_activity("exchange-raise"), api) + second = make_context(exchange_activity("exchange-raise"), api) + + results = await asyncio.gather( + handlers.sign_in_token_exchange(first), + handlers.sign_in_token_exchange(second), + return_exceptions=True, + ) + + assert api.users.exchange_token.await_count == 1 + assert all(isinstance(result, RuntimeError) for result in results) + assert all(str(result) == "handler failed" for result in results) + + @pytest.mark.asyncio + async def test_duplicate_arriving_during_sign_in_callbacks_awaits_the_owner(self): + """The completed marker is stamped the moment the token is redeemed. + + A duplicate that lands in the window between redemption and the sign-in + callbacks finishing must still mirror the owner, not answer ahead of it. + """ + handlers, _emitter, flow = make_handlers() + handler_started = asyncio.Event() + release_handler = asyncio.Event() + + @flow.on_signin + async def on_signin(_event): + handler_started.set() + await release_handler.wait() + raise RuntimeError("handler failed") + + api = make_api() + first = make_context(exchange_activity(), api) + second = make_context(exchange_activity(), api) + + owner = asyncio.create_task(handlers.sign_in_token_exchange(first)) + await handler_started.wait() + duplicate = asyncio.create_task(handlers.sign_in_token_exchange(second)) + await asyncio.sleep(0) + release_handler.set() + + results = await asyncio.gather(owner, duplicate, return_exceptions=True) + + assert api.users.exchange_token.await_count == 1 + assert all(isinstance(result, RuntimeError) for result in results) + assert all(str(result) == "handler failed" for result in results) + + @pytest.mark.asyncio + async def test_concurrent_exchanges_with_distinct_ids_both_run(self): + handlers, emitter, _flow = make_handlers() + api = make_api(exchange=slow_exchange()) + first = make_context(exchange_activity("exchange-a"), api) + second = make_context(exchange_activity("exchange-b"), api) + + results = await asyncio.gather( + handlers.sign_in_token_exchange(first), + handlers.sign_in_token_exchange(second), + ) + + assert results == [None, None] + assert api.users.exchange_token.await_count == 2 + assert len(emitted(emitter, "sign_in")) == 2 + assert first.next.await_count == 1 + assert second.next.await_count == 1 + + @pytest.mark.asyncio + async def test_concurrent_exchanges_without_an_id_are_never_collapsed(self): + """An empty id is not a shared key: two unrelated sign-ins must both proceed.""" + handlers, emitter, _flow = make_handlers() + api = make_api(exchange=slow_exchange()) + first = make_context(exchange_activity("", token="sso-token-a"), api) + second = make_context(exchange_activity("", token="sso-token-b"), api) + + results = await asyncio.gather( + handlers.sign_in_token_exchange(first), + handlers.sign_in_token_exchange(second), + ) + + assert results == [None, None] + assert api.users.exchange_token.await_count == 2 + exchanged = [call.args[0].exchange_request.token for call in api.users.exchange_token.await_args_list] + assert sorted(exchanged) == ["sso-token-a", "sso-token-b"] + assert len(emitted(emitter, "sign_in")) == 2 + assert first.next.await_count == 1 + assert second.next.await_count == 1 + + +class TestLateTokenExchangeDedup: + @pytest.mark.asyncio + async def test_late_duplicate_is_a_200_no_op(self): + handlers, emitter, flow = make_handlers() + signin_calls: List[str] = [] + + @flow.on_signin + async def on_signin(event): + signin_calls.append(event.connection_name) + + api = make_api() + first = make_context(exchange_activity(), api) + second = make_context(exchange_activity(), api) + + assert await handlers.sign_in_token_exchange(first) is None + late = await handlers.sign_in_token_exchange(second) + + assert isinstance(late, InvokeResponse) + assert late.status == 200 + assert late.body is None + assert api.users.exchange_token.await_count == 1 + assert signin_calls == [CONNECTION_NAME] + assert len(emitted(emitter, "sign_in")) == 1 + assert first.next.await_count == 1 + assert second.next.await_count == 0 + + @pytest.mark.asyncio + async def test_sequential_exchanges_with_distinct_ids_both_run(self): + handlers, emitter, _flow = make_handlers() + api = make_api() + + first = make_context(exchange_activity("exchange-a"), api) + second = make_context(exchange_activity("exchange-b"), api) + assert await handlers.sign_in_token_exchange(first) is None + assert await handlers.sign_in_token_exchange(second) is None + + assert api.users.exchange_token.await_count == 2 + assert len(emitted(emitter, "sign_in")) == 2 + + @pytest.mark.asyncio + async def test_sequential_exchanges_without_an_id_are_never_collapsed(self): + handlers, emitter, _flow = make_handlers() + api = make_api() + + first = make_context(exchange_activity(""), api) + second = make_context(exchange_activity(""), api) + assert await handlers.sign_in_token_exchange(first) is None + assert await handlers.sign_in_token_exchange(second) is None + + assert api.users.exchange_token.await_count == 2 + assert len(emitted(emitter, "sign_in")) == 2 + assert first.next.await_count == 1 + assert second.next.await_count == 1 + + @pytest.mark.asyncio + async def test_dedup_works_without_state_configured(self): + handlers, _emitter, _flow = make_handlers() + api = make_api() + first = make_context(exchange_activity(), api, state=None) + second = make_context(exchange_activity(), api, state=None) + + await handlers.sign_in_token_exchange(first) + late = await handlers.sign_in_token_exchange(second) + + assert status_of(late) == 200 + assert api.users.exchange_token.await_count == 1 + + @pytest.mark.asyncio + async def test_a_second_app_instance_does_not_share_in_memory_dedup(self): + """Guards against the marker set living in module scope instead of per app.""" + first_handlers, _first_emitter, _first_flow = make_handlers() + second_handlers, _second_emitter, _second_flow = make_handlers() + api = make_api() + + await first_handlers.sign_in_token_exchange(make_context(exchange_activity(), api)) + assert await second_handlers.sign_in_token_exchange(make_context(exchange_activity(), api)) is None + + assert api.users.exchange_token.await_count == 2 + + @pytest.mark.asyncio + async def test_sign_in_handler_failure_still_marks_the_exchange_as_spent(self): + """The exchange token is single-use, so a duplicate could never redeem it again.""" + handlers, _emitter, flow = make_handlers() + + @flow.on_signin + async def on_signin(_event): + raise RuntimeError("handler failed") + + api = make_api() + first = make_context(exchange_activity(), api) + second = make_context(exchange_activity(), api) + + with pytest.raises(RuntimeError, match="handler failed"): + await handlers.sign_in_token_exchange(first) + # PR4 guarantee: the owning request still advances the middleware chain. + assert first.next.await_count == 1 + + assert status_of(await handlers.sign_in_token_exchange(second)) == 200 + assert api.users.exchange_token.await_count == 1 + + +class TestTokenExchangeDedupFailures: + @pytest.mark.asyncio + async def test_failed_exchange_can_be_retried_with_the_same_id(self): + """The in-flight entry is released on settle and a failure is never marked.""" + handlers, emitter, _flow = make_handlers() + api = make_api(exchange=AsyncMock(side_effect=[oauth_http_error(400, "bad"), token_response()])) + + first = make_context(exchange_activity("exchange-retry"), api) + second = make_context(exchange_activity("exchange-retry"), api) + + failed = await handlers.sign_in_token_exchange(first) + assert status_of(failed) == 412 + + assert await handlers.sign_in_token_exchange(second) is None + assert api.users.exchange_token.await_count == 2 + assert len(emitted(emitter, "sign_in")) == 1 + assert first.next.await_count == 1 + assert second.next.await_count == 1 + + @pytest.mark.asyncio + async def test_unexpected_service_error_keeps_its_status_and_is_not_marked_complete(self): + handlers, emitter, _flow = make_handlers() + api = make_api(exchange=AsyncMock(side_effect=[oauth_http_error(503, "unavailable"), token_response()])) + + first = make_context(exchange_activity("exchange-503"), api) + second = make_context(exchange_activity("exchange-503"), api) + + result = await handlers.sign_in_token_exchange(first) + + assert isinstance(result, InvokeResponse) + assert result.status == 503 + assert result.body is None + assert len(emitted(emitter, "error")) == 1 + assert emitted(emitter, "sign_in") == [] + + assert await handlers.sign_in_token_exchange(second) is None + assert api.users.exchange_token.await_count == 2 + assert len(emitted(emitter, "sign_in")) == 1 + + @pytest.mark.asyncio + async def test_concurrent_duplicate_mirrors_an_unexpected_service_status(self): + handlers, emitter, _flow = make_handlers() + api = make_api(exchange=slow_exchange(oauth_http_error(503, "unavailable"))) + first = make_context(exchange_activity("exchange-503"), api) + second = make_context(exchange_activity("exchange-503"), api) + + results = await asyncio.gather( + handlers.sign_in_token_exchange(first), + handlers.sign_in_token_exchange(second), + ) + + assert [status_of(result) for result in results] == [503, 503] + assert api.users.exchange_token.await_count == 1 + # Only the request that actually talked to the service reports the error. + assert len(emitted(emitter, "error")) == 1 + + +class TestTokenExchangeDedupState: + @pytest.mark.asyncio + async def test_completion_is_persisted_under_the_reserved_conversation_key(self): + handlers, _emitter, _flow = make_handlers() + api = make_api() + state = make_state() + + await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, state)) + + marker = state.conversation[f"{EXCHANGE_STATE_KEY_PREFIX}exchange-1"] + # A bare ISO 8601 UTC string, matching the value shape the C# SDK stores. + assert isinstance(marker, str) + assert datetime.fromisoformat(marker).tzinfo is not None + # Conversation scope, not user scope — a duplicate can arrive from any of the + # user's clients but always on the same conversation. + assert state.user is not None + assert f"{EXCHANGE_STATE_KEY_PREFIX}exchange-1" not in state.user + + @pytest.mark.asyncio + async def test_persisted_marker_dedups_across_app_instances(self): + first_handlers, _first_emitter, _first_flow = make_handlers() + second_handlers, second_emitter, _second_flow = make_handlers() + api = make_api() + state = make_state() + + await first_handlers.sign_in_token_exchange(make_context(exchange_activity(), api, state)) + late = make_context(exchange_activity(), api, state) + result = await second_handlers.sign_in_token_exchange(late) + + assert status_of(result) == 200 + assert api.users.exchange_token.await_count == 1 + assert emitted(second_emitter, "sign_in") == [] + assert late.next.await_count == 0 + + @pytest.mark.asyncio + async def test_failure_is_not_persisted_as_a_completion(self): + handlers, _emitter, _flow = make_handlers() + api = make_api(exchange=AsyncMock(side_effect=oauth_http_error(400, "bad"))) + state = make_state() + + await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, state)) + + assert [key for key in state.conversation if key.startswith(EXCHANGE_STATE_KEY_PREFIX)] == [] + + @pytest.mark.asyncio + async def test_concurrent_duplicate_stamps_the_marker_into_its_own_snapshot(self): + """Each turn saves its own snapshot, last-write-wins. + + An unstamped duplicate snapshot would erase the owner's freshly written marker + if its save happened to land last. + """ + handlers, _emitter, _flow = make_handlers() + api = make_api(exchange=slow_exchange()) + owner_state = make_state() + waiter_state = make_state() + + await asyncio.gather( + handlers.sign_in_token_exchange(make_context(exchange_activity(), api, owner_state)), + handlers.sign_in_token_exchange(make_context(exchange_activity(), api, waiter_state)), + ) + + key = f"{EXCHANGE_STATE_KEY_PREFIX}exchange-1" + assert isinstance(owner_state.conversation[key], str) + assert isinstance(waiter_state.conversation[key], str) + assert api.users.exchange_token.await_count == 1 + + @pytest.mark.asyncio + async def test_late_duplicate_stamps_the_marker_into_its_own_snapshot(self): + handlers, _emitter, _flow = make_handlers() + api = make_api() + owner_state = make_state() + late_state = make_state() + + await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, owner_state)) + await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, late_state)) + + key = f"{EXCHANGE_STATE_KEY_PREFIX}exchange-1" + assert isinstance(late_state.conversation[key], str) + assert api.users.exchange_token.await_count == 1 + + @pytest.mark.asyncio + async def test_duplicate_does_not_extend_an_existing_marker(self, monkeypatch): + """Re-stamping a snapshot that already has the marker would push out its TTL.""" + handlers, _emitter, _flow = make_handlers() + api = make_api() + state = make_state() + clock = {"now": 1_000.0} + monkeypatch.setattr("microsoft_teams.apps.app_oauth.time", lambda: clock["now"]) + monkeypatch.setattr("microsoft_teams.apps.oauth_state.time", lambda: clock["now"]) + + await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, state)) + clock["now"] += 60 + await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, state)) + + assert state.conversation[f"{EXCHANGE_STATE_KEY_PREFIX}exchange-1"] == iso(1_000.0) + + @pytest.mark.parametrize( + "corrupt", + [ + "not-an-iso-timestamp", + "", + 123, + None, + True, + [], + {"version": 1, "completed_at": 1.0}, + ], + ids=[ + "unparseable-string", + "empty-string", + "number", + "null", + "bool", + "list", + "legacy-dict-format", + ], + ) + @pytest.mark.asyncio + async def test_corrupt_persisted_marker_does_not_crash_the_turn(self, corrupt): + handlers, emitter, _flow = make_handlers() + api = make_api() + state = make_state() + state.conversation[f"{EXCHANGE_STATE_KEY_PREFIX}exchange-1"] = corrupt + + result = await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, state)) + + assert result is None + assert api.users.exchange_token.await_count == 1 + assert len(emitted(emitter, "sign_in")) == 1 + # The unusable value is replaced by a well-formed marker. + assert isinstance(state.conversation[f"{EXCHANGE_STATE_KEY_PREFIX}exchange-1"], str) + + @pytest.mark.asyncio + async def test_corrupt_persisted_marker_is_logged(self, caplog): + handlers, _emitter, _flow = make_handlers() + api = make_api() + state = make_state() + state.conversation[f"{EXCHANGE_STATE_KEY_PREFIX}exchange-1"] = "not-an-iso-timestamp" + + with caplog.at_level(logging.WARNING, logger="microsoft_teams.apps.oauth_state"): + await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, state)) + + assert "malformed completed OAuth token exchange state" in caplog.text + + @pytest.mark.asyncio + async def test_marker_leaves_unrelated_state_untouched(self): + handlers, _emitter, _flow = make_handlers() + api = make_api() + state = make_state() + state.conversation["app-data"] = {"counter": 1} + + await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, state)) + + assert state.conversation["app-data"] == {"counter": 1} + + +class TestTokenExchangeDedupExpiry: + @pytest.mark.asyncio + async def test_marker_expires_after_the_ttl(self, monkeypatch): + handlers, emitter, _flow = make_handlers() + api = make_api() + state = make_state() + clock = {"now": 1_000.0} + monkeypatch.setattr("microsoft_teams.apps.app_oauth.time", lambda: clock["now"]) + monkeypatch.setattr("microsoft_teams.apps.oauth_state.time", lambda: clock["now"]) + + await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, state)) + + clock["now"] += DEDUP_TTL_SECONDS - 1 + assert status_of(await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, state))) == 200 + assert api.users.exchange_token.await_count == 1 + + clock["now"] += 2 + assert await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, state)) is None + assert api.users.exchange_token.await_count == 2 + assert len(emitted(emitter, "sign_in")) == 2 + + @pytest.mark.asyncio + async def test_expired_persisted_markers_are_pruned_from_conversation_state(self, monkeypatch): + handlers, _emitter, _flow = make_handlers() + api = make_api() + state = make_state() + clock = {"now": 1_000.0} + monkeypatch.setattr("microsoft_teams.apps.app_oauth.time", lambda: clock["now"]) + monkeypatch.setattr("microsoft_teams.apps.oauth_state.time", lambda: clock["now"]) + + await handlers.sign_in_token_exchange(make_context(exchange_activity("old"), api, state)) + clock["now"] += DEDUP_TTL_SECONDS + 1 + await handlers.sign_in_token_exchange(make_context(exchange_activity("new"), api, state)) + + markers = [key for key in state.conversation if key.startswith(EXCHANGE_STATE_KEY_PREFIX)] + assert markers == [f"{EXCHANGE_STATE_KEY_PREFIX}new"] + + @pytest.mark.asyncio + async def test_marker_from_the_future_is_treated_as_corrupt(self, monkeypatch): + """A clock jump backwards must not pin a marker in place forever.""" + handlers, _emitter, _flow = make_handlers() + api = make_api() + state = make_state() + monkeypatch.setattr("microsoft_teams.apps.oauth_state.time", lambda: 1_000.0) + # A well-formed ISO value, so this exercises the clock-skew guard rather than + # merely failing to parse. + state.conversation[f"{EXCHANGE_STATE_KEY_PREFIX}exchange-1"] = iso(1_000.0 + (60 * 60)) + + assert await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, state)) is None + assert api.users.exchange_token.await_count == 1 + + @pytest.mark.asyncio + async def test_in_memory_markers_are_bounded(self): + """Past the cap the oldest ids are evicted, so the set cannot grow forever.""" + handlers, _emitter, _flow = make_handlers() + api = make_api() + + for index in range(DEDUP_MAX_ENTRIES + 1): + await handlers.sign_in_token_exchange(make_context(exchange_activity(f"exchange-{index}"), api)) + assert api.users.exchange_token.await_count == DEDUP_MAX_ENTRIES + 1 + + # The newest id is still remembered... + newest = f"exchange-{DEDUP_MAX_ENTRIES}" + assert status_of(await handlers.sign_in_token_exchange(make_context(exchange_activity(newest), api))) == 200 + assert api.users.exchange_token.await_count == DEDUP_MAX_ENTRIES + 1 + + # ...while the oldest was evicted to keep the set bounded. + assert await handlers.sign_in_token_exchange(make_context(exchange_activity("exchange-0"), api)) is None + assert api.users.exchange_token.await_count == DEDUP_MAX_ENTRIES + 2 + + @pytest.mark.asyncio + async def test_expired_in_memory_markers_are_pruned_without_state(self, monkeypatch): + handlers, _emitter, _flow = make_handlers() + api = make_api() + clock = {"now": 1_000.0} + monkeypatch.setattr("microsoft_teams.apps.app_oauth.time", lambda: clock["now"]) + + await handlers.sign_in_token_exchange(make_context(exchange_activity(), api)) + clock["now"] += DEDUP_TTL_SECONDS + 1 + + assert await handlers.sign_in_token_exchange(make_context(exchange_activity(), api)) is None + assert api.users.exchange_token.await_count == 2 + + +class TestOtherSignInCallbacksAreNotDeduplicated: + @pytest.mark.asyncio + async def test_verify_state_is_not_deduplicated(self): + """The verify code is single-use, so repeats are naturally idempotent.""" + handlers, emitter, flow = make_handlers() + signin_calls: List[str] = [] + + @flow.on_signin + async def on_signin(event): + signin_calls.append(event.connection_name) + + api = make_api() + first = make_context(verify_state_activity(), api) + second = make_context(verify_state_activity(), api) + + assert await handlers.sign_in_verify_state(first) is None + assert await handlers.sign_in_verify_state(second) is None + + assert api.users.get_token.await_count == 2 + assert len(emitted(emitter, "sign_in")) == 2 + assert signin_calls == [CONNECTION_NAME, CONNECTION_NAME] + assert first.next.await_count == 1 + assert second.next.await_count == 1 + + @pytest.mark.asyncio + async def test_concurrent_verify_states_are_not_deduplicated(self): + handlers, emitter, _flow = make_handlers() + get_token = AsyncMock(side_effect=slow_exchange().side_effect) + api = make_api(get_token=get_token) + first = make_context(verify_state_activity(), api) + second = make_context(verify_state_activity(), api) + + await asyncio.gather( + handlers.sign_in_verify_state(first), + handlers.sign_in_verify_state(second), + ) + + assert get_token.await_count == 2 + assert len(emitted(emitter, "sign_in")) == 2 + + @pytest.mark.asyncio + async def test_sign_in_failure_is_not_deduplicated(self): + """Failure is a single informational notice, not a state-changing operation.""" + handlers, emitter, flow = make_handlers() + failures: List[str] = [] + + @flow.on_signin_failure + async def on_failure(event): + failures.append(event.code or "") + + api = make_api() + first = make_context(failure_activity(), api) + second = make_context(failure_activity(), api) + + assert await handlers.sign_in_failure(first) is None + assert await handlers.sign_in_failure(second) is None + + assert len(emitted(emitter, "sign_in_failure")) == 2 + assert failures == ["invokeerror", "invokeerror"] + assert first.next.await_count == 1 + assert second.next.await_count == 1 From 3012cbc9740ffb0baae9443ccd22bb2a16860f76 Mon Sep 17 00:00:00 2001 From: lilydu Date: Thu, 27 Aug 2026 10:45:31 -0700 Subject: [PATCH 02/10] feat(oauth): demonstrate multi-connection OAuth in the example Rewrite examples/oauth around two registered flows -- "profile" (User.Read) and "mail" (Mail.Read) -- driven through add_oauth_flow, replacing the single default-connection ctx.sign_in() path. Each flow gets its own on_signin/on_signin_failure handler, and those handlers call Microsoft Graph with that connection's token so a completed sign-in returns real per-connection data rather than a static string. A "status" command reports both connections independently, showing that a user can be signed in to one and out of the other. The global @app.event("sign_in") handler is kept to show that it now carries connection_name alongside the per-flow handlers. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- examples/oauth/pyproject.toml | 2 + examples/oauth/src/main.py | 107 +++++++++++++++++++++++++--------- uv.lock | 2 + 3 files changed, 82 insertions(+), 29 deletions(-) diff --git a/examples/oauth/pyproject.toml b/examples/oauth/pyproject.toml index 87fce76b..590b9558 100644 --- a/examples/oauth/pyproject.toml +++ b/examples/oauth/pyproject.toml @@ -7,7 +7,9 @@ requires-python = ">=3.11,<4.0" dependencies = [ "dotenv>=0.9.9", "microsoft-teams-apps", + "microsoft-teams-graph", ] [tool.uv.sources] microsoft-teams-apps = { workspace = true } +microsoft-teams-graph = { workspace = true } diff --git a/examples/oauth/src/main.py b/examples/oauth/src/main.py index 9215f91d..334a57b1 100644 --- a/examples/oauth/src/main.py +++ b/examples/oauth/src/main.py @@ -7,55 +7,104 @@ import logging from microsoft_teams.api import MessageActivity -from microsoft_teams.api.activities.invoke.sign_in import SignInFailureInvokeActivity from microsoft_teams.apps import ActivityContext, App, SignInEvent -from microsoft_teams.apps.events.types import ErrorEvent +from microsoft_teams.apps.events.types import ErrorEvent, SignInFailureEvent from microsoft_teams.common import ConsoleFormatter +from microsoft_teams.graph import get_graph_client # Setup logging -logging.getLogger().setLevel(logging.DEBUG) +logging.getLogger().setLevel(logging.INFO) stream_handler = logging.StreamHandler() stream_handler.setFormatter(ConsoleFormatter()) logging.getLogger().addHandler(stream_handler) logger = logging.getLogger(__name__) -app = App() +# Pending sign-in hints are stored in per-turn state, which lets connection-less +# callbacks be routed back to the flow that started them. +app = App(state=True) +# Multi-connection OAuth: two named connections registered with add_oauth_flow, each +# driven through the OAuthFlow it returns. +# +# Bot needs two OAuth connections configured in Azure, named "profile" and +# "mail", matching the names passed below. Grant "User.Read" to the first and +# "Mail.Read" to the second. +profile = app.add_oauth_flow("profile", oauth_card_text="Sign in to read your profile") +mail = app.add_oauth_flow("mail", oauth_card_text="Sign in to read your mail") + + +@profile.on_signin +async def on_profile_signin(event: SignInEvent) -> None: + """Only fires for the `profile` connection, using that connection's token.""" + client = get_graph_client(event.token_response.token) + me = await client.me.get() + name = me.display_name if me else "unknown" + await event.activity_ctx.send(f"Signed in as **{name}**.") + + +@mail.on_signin +async def on_mail_signin(event: SignInEvent) -> None: + """Only fires for the `mail` connection — a token the profile flow does not have.""" + client = get_graph_client(event.token_response.token) + page = await client.me.messages.get() + subjects = [m.subject or "(no subject)" for m in (page.value or [])[:3]] if page else [] + body = "\n".join(f"- {s}" for s in subjects) if subjects else "_no messages_" + await event.activity_ctx.send(f"Latest mail:\n{body}") + + +@profile.on_signin_failure +async def on_profile_failure(event: SignInFailureEvent) -> None: + await event.activity_ctx.send(f"Profile sign-in failed: {event.code} - {event.message}") -@app.on_message -async def handle_message(ctx: ActivityContext[MessageActivity]): - """Handle message activities using the new generated handler system.""" - print(f"[GENERATED onMessage] Message received: {ctx.activity.text}") - print(f"[GENERATED onMessage] From: {ctx.activity.from_}") - logger.info("User requested sign-in.") - if ctx.is_signed_in: - await ctx.send("You are already signed in. Logging you out.") - await ctx.sign_out() - else: - await ctx.sign_in() +@mail.on_signin_failure +async def on_mail_failure(event: SignInFailureEvent) -> None: + await event.activity_ctx.send(f"Mail sign-in failed: {event.code} - {event.message}") @app.event("sign_in") -async def handle_sign_in(event: SignInEvent): - """Handle sign-in events.""" - await event.activity_ctx.send("You are now signed in!") +async def on_any_signin(event: SignInEvent) -> None: + """Fires for every connection. `connection_name` says which one completed.""" + logger.info("sign-in completed on connection %r", event.connection_name) + + +@app.on_message +async def handle_message(ctx: ActivityContext[MessageActivity]) -> None: + text = (ctx.activity.text or "").strip().lower() + + if text == "sign in profile": + # Returns the token directly when one is already cached, otherwise sends a + # card and returns None; the flow's on_signin handler fires once it completes. + if await profile.sign_in(ctx): + await ctx.send("Already signed in for profile access.") + return + + if text == "sign in mail": + if await mail.sign_in(ctx): + await ctx.send("Already signed in for mail access.") + return + + if text == "sign out": + await profile.sign_out(ctx) + await mail.sign_out(ctx) + await ctx.send("Signed out of both connections.") + return + if text == "status": + # Tokens are tracked per connection, so these can differ. + lines: list[str] = [] + for flow in (profile, mail): + state = "signed in" if await flow.is_signed_in(ctx) else "signed out" + lines.append(f"- `{flow.connection_name}`: {state}") + await ctx.send("\n".join(lines)) + return -@app.on_signin_failure() -async def handle_signin_failure(ctx: ActivityContext[SignInFailureInvokeActivity]): - """Handle sign-in failure events.""" - failure = ctx.activity.value - print(f"Sign-in failed: {failure.code} - {failure.message}") - await ctx.send("Sign-in failed.") + await ctx.send("Try `sign in profile`, `sign in mail`, `status`, or `sign out`.") @app.event("error") -async def handle_error(event: ErrorEvent): - """Handle error events.""" - print(f"Error occurred: {event.error}") - if event.context: - print(f"Context: {event.context}") +async def handle_error(event: ErrorEvent) -> None: + logger.error("error: %s (context=%s)", event.error, event.context) if __name__ == "__main__": diff --git a/uv.lock b/uv.lock index 1fd5729d..48b16f0c 100644 --- a/uv.lock +++ b/uv.lock @@ -2354,12 +2354,14 @@ source = { virtual = "examples/oauth" } dependencies = [ { name = "dotenv" }, { name = "microsoft-teams-apps" }, + { name = "microsoft-teams-graph" }, ] [package.metadata] requires-dist = [ { name = "dotenv", specifier = ">=0.9.9" }, { name = "microsoft-teams-apps", editable = "packages/apps" }, + { name = "microsoft-teams-graph", editable = "packages/graph" }, ] [[package]] From bba47b0a03c9a49bb3ee545c27a1aa29601bd17f Mon Sep 17 00:00:00 2001 From: lilydu Date: Thu, 27 Aug 2026 14:33:47 -0700 Subject: [PATCH 03/10] feat(oauth): bound persisted token-exchange dedup markers Completed-exchange markers were pruned by TTL but never counted, so a conversation that began many distinct sign-ins inside the five-minute window could grow its stored document without limit. The in-memory completed set was already capped at 1000; this gives the persisted set the same ceiling. Enforcement runs at the existing write chokepoint, after the marker is written, so the exchange being recorded can never be the one evicted to make room. Expired markers are pruned first, so eviction only ever gives up live coverage as a last resort, and it starts with the oldest markers because those are closest to ageing out on their own. Ties are broken on the key so two instances holding the same document evict identically rather than following dict insertion order. The TTL remains the primary bound and the only one expected to bind in practice: markers are scoped to one conversation and every duplicate of an exchange reuses its id. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../src/microsoft_teams/apps/oauth_state.py | 56 ++++++- packages/apps/tests/test_oauth_state.py | 151 ++++++++++++++++++ 2 files changed, 204 insertions(+), 3 deletions(-) diff --git a/packages/apps/src/microsoft_teams/apps/oauth_state.py b/packages/apps/src/microsoft_teams/apps/oauth_state.py index c1648720..00a0d2ea 100644 --- a/packages/apps/src/microsoft_teams/apps/oauth_state.py +++ b/packages/apps/src/microsoft_teams/apps/oauth_state.py @@ -45,6 +45,15 @@ # Completed markers age out on the same schedule as pending sign-ins. TOKEN_EXCHANGE_DEDUP_TTL_SECONDS = _PENDING_OAUTH_MAX_AGE_SECONDS +# Hard ceiling on how many completed markers one conversation document may carry. +# The TTL above is the primary bound and the only one that should ever bind in +# practice: these markers are scoped to a single conversation, and every duplicate of +# an exchange reuses its id, so a conversation would have to begin a thousand +# *distinct* sign-ins inside five minutes to reach this. The cap is a backstop that +# keeps a pathological burst from growing the stored document without limit, matching +# the 1000-entry bound the TypeScript SDK places on its completed list. +_COMPLETED_EXCHANGE_MAX_ENTRIES = 1000 + @dataclass(frozen=True) class PendingOAuthSignIn: @@ -333,7 +342,9 @@ def record_completed_token_exchange(state: Optional[TurnStateContainer], exchang The marker is deliberately never cleared when the exchange finishes: a late duplicate from a second Teams endpoint can arrive after the original settles, and an already-removed marker would let it run as a brand new exchange. Markers are - pruned only once they age past :data:`TOKEN_EXCHANGE_DEDUP_TTL_SECONDS`. + dropped only once they age past :data:`TOKEN_EXCHANGE_DEDUP_TTL_SECONDS`, or -- far + more rarely -- when a conversation exceeds + :data:`_COMPLETED_EXCHANGE_MAX_ENTRIES` and the oldest are trimmed to fit. """ if state is None or not exchange_id: return @@ -344,10 +355,49 @@ def _write_completed_token_exchange(state: TurnStateContainer, exchange_id: str) """Single write chokepoint for completed markers. Pruning here keeps the invariant that the conversation document never carries an - expired or unparsable marker, no matter which caller wrote it. + expired or unparsable marker, no matter which caller wrote it. The ceiling is + enforced afterwards, with the new marker already in place, so the exchange being + recorded can never be the one evicted to make room. Every marker enters through + this function, so bounding on write bounds the stored set for good. """ _prune_completed_token_exchanges(state) - state.conversation[completed_token_exchange_state_key(exchange_id)] = _format_timestamp(time()) + key = completed_token_exchange_state_key(exchange_id) + state.conversation[key] = _format_timestamp(time()) + _enforce_completed_token_exchange_cap(state, key) + + +def _enforce_completed_token_exchange_cap(state: TurnStateContainer, keep: str) -> None: + """Drop the oldest markers once a conversation exceeds the cap. + + Expired markers are pruned before this runs, so everything still stored is live and + evicting any of it costs real dedup coverage. Oldest-first is the least damaging + order available: those markers are the closest to ageing out on their own, so they + have the least protection left to give. + + ``keep`` is the marker just written and is never evicted -- it is the one the + current exchange depends on, and it would otherwise become a candidate whenever its + timestamp ties with another. Remaining ties are broken on the key so that eviction + is deterministic across instances instead of following dict insertion order. + """ + keys = [key for key in state.conversation if key.startswith(_COMPLETED_EXCHANGE_STATE_KEY_PREFIX)] + overflow = len(keys) - _COMPLETED_EXCHANGE_MAX_ENTRIES + if overflow <= 0: + return + + def _age_order(key: str) -> tuple[float, str]: + completed_at = _parse_completed_at(state.conversation.get(key)) + # Unparsable markers sort first. Pruning should have removed them already, and + # anything that slipped through carries no usable expiry, so it is the safest + # thing to give up. + return (completed_at if completed_at is not None else float("-inf"), key) + + for key in sorted(keys, key=_age_order): + if overflow <= 0: + break + if key == keep: + continue + state.conversation.pop(key, None) + overflow -= 1 def _prune_completed_token_exchanges(state: TurnStateContainer) -> None: diff --git a/packages/apps/tests/test_oauth_state.py b/packages/apps/tests/test_oauth_state.py index bbf23a49..267d1a61 100644 --- a/packages/apps/tests/test_oauth_state.py +++ b/packages/apps/tests/test_oauth_state.py @@ -10,10 +10,16 @@ import pytest from microsoft_teams.apps import TurnState, TurnStateContainer from microsoft_teams.apps.oauth_state import ( + _COMPLETED_EXCHANGE_MAX_ENTRIES, # pyright: ignore[reportPrivateUsage] + TOKEN_EXCHANGE_DEDUP_TTL_SECONDS, PendingOAuthSignIn, + _enforce_completed_token_exchange_cap, # pyright: ignore[reportPrivateUsage] clear_pending_oauth_sign_in, + completed_token_exchange_state_key, get_pending_oauth_sign_ins, + has_completed_token_exchange, mark_pending_oauth_sso_consumed, + record_completed_token_exchange, record_pending_oauth_sign_in, replace_pending_oauth_sign_ins, ) @@ -302,3 +308,148 @@ def test_helpers_are_no_ops_without_a_user_scope(self) -> None: clear_pending_oauth_sign_in(state, "Graph") mark_pending_oauth_sso_consumed(state, "Graph") replace_pending_oauth_sign_ins(state, []) + + +# Declared rather than imported, so a change to the stored layout has to be a +# deliberate edit here too. Mirrors ``test_app_oauth_dedup.py``. +EXCHANGE_STATE_KEY_PREFIX = "__oauth:exchange:" + + +def completed_keys(state: TurnStateContainer) -> set[str]: + return {key for key in state.conversation if key.startswith(EXCHANGE_STATE_KEY_PREFIX)} + + +def seed_completed(state: TurnStateContainer, exchange_ids: list[str], *, age_seconds: float) -> None: + """Write markers straight into the document, bypassing the prune-and-cap write path.""" + stamp = iso(time.time() - age_seconds) + for exchange_id in exchange_ids: + state.conversation[completed_token_exchange_state_key(exchange_id)] = stamp + + +class TestCompletedExchangeCap: + def test_cap_is_one_thousand(self): + """Pinned deliberately: it is the bound the TypeScript SDK also promises.""" + assert _COMPLETED_EXCHANGE_MAX_ENTRIES == 1000 + + def test_writing_up_to_the_cap_evicts_nothing(self): + state = make_state() + seed_completed( + state, + [f"old-{index:05d}" for index in range(_COMPLETED_EXCHANGE_MAX_ENTRIES - 1)], + age_seconds=10, + ) + + record_completed_token_exchange(state, "newest") + + # Exactly at the limit, so every marker survives. + assert len(completed_keys(state)) == _COMPLETED_EXCHANGE_MAX_ENTRIES + assert has_completed_token_exchange(state, "newest") + assert has_completed_token_exchange(state, "old-00000") + + def test_overflow_evicts_the_oldest_marker(self): + state = make_state() + now = time.time() + # Ages ascending, so ``old-00000`` is the oldest and first to go. + for index in range(_COMPLETED_EXCHANGE_MAX_ENTRIES): + key = completed_token_exchange_state_key(f"old-{index:05d}") + state.conversation[key] = iso(now - 250 + index * 0.01) + + record_completed_token_exchange(state, "newest") + + assert len(completed_keys(state)) == _COMPLETED_EXCHANGE_MAX_ENTRIES + assert not has_completed_token_exchange(state, "old-00000") + assert has_completed_token_exchange(state, "old-00001") + assert has_completed_token_exchange(state, "newest") + + def test_expired_markers_are_pruned_before_the_cap_applies(self): + """Expiry must run first, or a live marker is evicted while dead ones stay.""" + state = make_state() + seed_completed( + state, + [f"stale-{index:05d}" for index in range(_COMPLETED_EXCHANGE_MAX_ENTRIES)], + age_seconds=TOKEN_EXCHANGE_DEDUP_TTL_SECONDS + 60, + ) + seed_completed(state, ["live"], age_seconds=1) + + record_completed_token_exchange(state, "newest") + + assert completed_keys(state) == { + completed_token_exchange_state_key("live"), + completed_token_exchange_state_key("newest"), + } + + def test_ties_are_broken_deterministically_across_instances(self): + """Identical timestamps must not leave eviction up to dict insertion order.""" + first = make_state() + second = make_state() + exchange_ids = [f"tie-{index:05d}" for index in range(_COMPLETED_EXCHANGE_MAX_ENTRIES)] + stamp = iso(time.time() - 5) + for exchange_id in exchange_ids: + first.conversation[completed_token_exchange_state_key(exchange_id)] = stamp + for exchange_id in reversed(exchange_ids): + second.conversation[completed_token_exchange_state_key(exchange_id)] = stamp + + record_completed_token_exchange(first, "newest") + record_completed_token_exchange(second, "newest") + + assert completed_keys(first) == completed_keys(second) + assert not has_completed_token_exchange(first, "tie-00000") + + def test_the_marker_being_written_is_never_evicted(self): + """The current exchange still depends on the marker being recorded for it.""" + state = make_state() + stamp = iso(time.time() - 5) + keep = completed_token_exchange_state_key("aaa-current") + state.conversation[keep] = stamp + for index in range(_COMPLETED_EXCHANGE_MAX_ENTRIES + 25): + state.conversation[completed_token_exchange_state_key(f"tie-{index:05d}")] = stamp + + # Everything shares a timestamp and ``aaa-current`` sorts first on the key + # tie-break, so an unguarded sweep would drop the very marker being written. + _enforce_completed_token_exchange_cap(state, keep) + + assert keep in completed_keys(state) + assert len(completed_keys(state)) == _COMPLETED_EXCHANGE_MAX_ENTRIES + + def test_malformed_markers_are_dropped_rather_than_filling_the_cap(self): + state = make_state() + state.conversation[completed_token_exchange_state_key("unparsable")] = "not-a-timestamp" + state.conversation[completed_token_exchange_state_key("wrong-type")] = 12345 + state.conversation[completed_token_exchange_state_key("far-future")] = iso(time.time() + 3600) + + record_completed_token_exchange(state, "newest") + + assert completed_keys(state) == {completed_token_exchange_state_key("newest")} + + def test_capping_leaves_unrelated_conversation_state_alone(self): + state = make_state() + state.conversation["app:counter"] = 7 + seed_completed( + state, + [f"old-{index:05d}" for index in range(_COMPLETED_EXCHANGE_MAX_ENTRIES + 40)], + age_seconds=10, + ) + + record_completed_token_exchange(state, "newest") + + assert state.conversation["app:counter"] == 7 + assert len(completed_keys(state)) == _COMPLETED_EXCHANGE_MAX_ENTRIES + + def test_recorded_marker_round_trips_after_an_overflowing_write(self): + state = make_state() + now = time.time() + for index in range(_COMPLETED_EXCHANGE_MAX_ENTRIES + 10): + key = completed_token_exchange_state_key(f"old-{index:05d}") + state.conversation[key] = iso(now - 250 + index * 0.01) + + record_completed_token_exchange(state, "round-trip") + + assert len(completed_keys(state)) == _COMPLETED_EXCHANGE_MAX_ENTRIES + # Reload the way conversation state is actually rehydrated on the next turn. + reloaded = TurnStateContainer( + conversation=TurnState(dict(state.conversation)), + conversation_id="conv-1", + user=TurnState(), + user_id="user-1", + ) + assert has_completed_token_exchange(reloaded, "round-trip") From 4ccbbdc91e3cdca010b010dfcf20b4db9482b494 Mon Sep 17 00:00:00 2001 From: lilydu Date: Thu, 27 Aug 2026 16:45:42 -0700 Subject: [PATCH 04/10] test(oauth): align dedup tests with flow handler isolation The rebase onto the latest sign-in routing work brought in handler isolation: a raising on_signin listener is now logged and skipped rather than propagated, so it no longer turns a successful callback into a failed invoke response. Three dedup tests still asserted the old propagating contract and failed on the new base. Their dedup intent is unchanged, only the contract they assert: - the concurrent-exception test now proves a broken listener does not break dedup -- one exchange, one sign_in event, both callers get 200 - the mid-callback test drops the raise and instead pins the invariant it was always about, asserting the duplicate is still parked on the owner's future and that the owner's callbacks finish first - the spent-marker test keeps its subject, that the marker is stamped even when a listener fails, without expecting the failure to surface Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- packages/apps/tests/test_app_oauth_dedup.py | 32 +++++++++++++++------ 1 file changed, 23 insertions(+), 9 deletions(-) diff --git a/packages/apps/tests/test_app_oauth_dedup.py b/packages/apps/tests/test_app_oauth_dedup.py index ea9081bb..43eac165 100644 --- a/packages/apps/tests/test_app_oauth_dedup.py +++ b/packages/apps/tests/test_app_oauth_dedup.py @@ -225,7 +225,13 @@ async def on_signin(_event): assert emitted(emitter, "sign_in") == [] @pytest.mark.asyncio - async def test_concurrent_duplicate_sees_the_same_handler_exception(self): + async def test_broken_sign_in_handler_does_not_break_dedup(self): + """A raising ``on_signin`` listener is contained by the flow. + + Handler isolation means one broken listener must not turn a successful + callback into a failed invoke response -- so neither the owner nor its + duplicate fails, and the token is still exchanged exactly once. + """ handlers, emitter, flow = make_handlers() @flow.on_signin @@ -243,8 +249,9 @@ async def on_signin(_event): ) assert api.users.exchange_token.await_count == 1 - assert all(isinstance(result, RuntimeError) for result in results) - assert all(str(result) == "handler failed" for result in results) + assert not any(isinstance(result, BaseException) for result in results) + assert [status_of(result) for result in results] == [200, 200] + assert len(emitted(emitter, "sign_in")) == 1 @pytest.mark.asyncio async def test_duplicate_arriving_during_sign_in_callbacks_awaits_the_owner(self): @@ -256,12 +263,13 @@ async def test_duplicate_arriving_during_sign_in_callbacks_awaits_the_owner(self handlers, _emitter, flow = make_handlers() handler_started = asyncio.Event() release_handler = asyncio.Event() + finished: List[str] = [] @flow.on_signin async def on_signin(_event): handler_started.set() await release_handler.wait() - raise RuntimeError("handler failed") + finished.append("handler") api = make_api() first = make_context(exchange_activity(), api) @@ -271,13 +279,18 @@ async def on_signin(_event): await handler_started.wait() duplicate = asyncio.create_task(handlers.sign_in_token_exchange(second)) await asyncio.sleep(0) + + # Still parked on the owner's future: the marker alone must not let it answer. + assert not duplicate.done() + assert finished == [] release_handler.set() - results = await asyncio.gather(owner, duplicate, return_exceptions=True) + results = await asyncio.gather(owner, duplicate) assert api.users.exchange_token.await_count == 1 - assert all(isinstance(result, RuntimeError) for result in results) - assert all(str(result) == "handler failed" for result in results) + # The owner's callbacks ran to completion before the duplicate was answered. + assert finished == ["handler"] + assert [status_of(result) for result in results] == [200, 200] @pytest.mark.asyncio async def test_concurrent_exchanges_with_distinct_ids_both_run(self): @@ -411,8 +424,9 @@ async def on_signin(_event): first = make_context(exchange_activity(), api) second = make_context(exchange_activity(), api) - with pytest.raises(RuntimeError, match="handler failed"): - await handlers.sign_in_token_exchange(first) + # Handler isolation contains the failure, so the exchange still succeeds -- + # but the marker must be stamped either way, which is what the duplicate proves. + assert await handlers.sign_in_token_exchange(first) is None # PR4 guarantee: the owning request still advances the middleware chain. assert first.next.await_count == 1 From 8b3cb195d1e937345864ee841878b67cca999531 Mon Sep 17 00:00:00 2001 From: lilydu Date: Thu, 27 Aug 2026 16:58:59 -0700 Subject: [PATCH 05/10] fix(oauth): harden token-exchange dedup against state and cancellation failures Addresses five review findings on the dedup path: - Persisted-marker reads run before `_run_token_exchange` opens its try/finally, so a state error escaped past `ctx.next()` and stalled the middleware chain. Reads and writes are now best-effort, degrading to in-memory dedup rather than taking the turn down. - The completed marker is flushed mid-turn instead of at end of turn. The owner still has sign-in callbacks to run after redeeming the token, and a duplicate racing on another process instance would otherwise load a snapshot with no marker and redeem the spent exchange again. - A failure *after* the token was redeemed reported `token_redeemed=False` to waiters, and waiters checked `outcome.error` before stamping. Either alone lost the completion marker to a waiter's last-write-wins save. Both are fixed; `_clear_pending` is the reachable trigger, since flow listener failures are isolated and cannot fail the exchange. - Waiters no longer re-raise the owner's exception object. A `CancelledError` made an uncancelled waiter claim it was cancelled, and one instance shared across waiters had them all append frames to the same traceback. Waiters now mirror the failure as a 412, matching the TypeScript SDK. - Both duplicate paths resolve the connection name through the registry, so the casing Teams echoes back from the card does not split one connection into several telemetry series. Each fix is covered by a test verified to fail when the fix is reverted. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../src/microsoft_teams/apps/app_oauth.py | 104 +++++++++-- packages/apps/tests/test_app_oauth_dedup.py | 167 ++++++++++++++++++ 2 files changed, 258 insertions(+), 13 deletions(-) diff --git a/packages/apps/src/microsoft_teams/apps/app_oauth.py b/packages/apps/src/microsoft_teams/apps/app_oauth.py index 971865ce..d4e8fdd2 100644 --- a/packages/apps/src/microsoft_teams/apps/app_oauth.py +++ b/packages/apps/src/microsoft_teams/apps/app_oauth.py @@ -116,7 +116,17 @@ async def sign_in_token_exchange( except BaseException as error: # ``BaseException`` so cancellation also releases the entry and wakes # waiters, instead of leaking the id until the process restarts. - self._settle_token_exchange(exchange_id, owned, _TokenExchangeOutcome(error=error)) + self._settle_token_exchange( + exchange_id, + owned, + # ``token_redeemed`` matters on this path too: the exchange can fail + # *after* the token was already spent, and a waiter that does not know + # the marker was written would let its own stale snapshot erase it. + _TokenExchangeOutcome( + error=error, + token_redeemed=exchange_id in self._token_exchange_completed, + ), + ) raise self._settle_token_exchange( exchange_id, @@ -229,7 +239,7 @@ async def _run_token_exchange(self, ctx: ActivityContext[SignInTokenExchangeInvo # effects: the exchange token is spent at this point, so a retry could # never succeed anyway. A failed exchange is never marked, leaving the # id free for a genuine retry. - self._record_completed_token_exchange(ctx, activity.value.id) + await self._record_completed_token_exchange(ctx, activity.value.id) ctx.is_signed_in = True ctx.user_token = token.token self.oauth_registry._clear_pending( # pyright: ignore[reportPrivateUsage] @@ -262,9 +272,42 @@ def _is_completed_token_exchange( self._prune_completed_token_exchanges() if exchange_id in self._token_exchange_completed: return True - return has_completed_token_exchange(ctx.state, exchange_id) + return self._read_persisted_marker(ctx, exchange_id) + + def _resolved_connection_name(self, connection_name: str) -> str: + """The registered casing for ``connection_name``. + + Teams echoes back whatever casing the sign-in card carried, so the same + connection can arrive as ``Graph`` on one request and ``graph`` on the next. + The duplicate paths report this name to telemetry, and an unresolved name + would split one connection into several series. + + Resolution is a registry dict lookup, exactly what ``_run_token_exchange`` + does for ``event_connection_name``. Note that ``_run_token_exchange`` still + reports the *raw* name to its own telemetry; making that consistent is a + wider change than this dedup path and is deliberately left alone here. + """ + flow = self.oauth_registry.get(connection_name) + return flow.connection_name if flow is not None else connection_name + + def _read_persisted_marker(self, ctx: ActivityContext[SignInTokenExchangeInvokeActivity], exchange_id: str) -> bool: + """Read the cross-instance marker without ever failing the turn. + + This runs before the owning request reaches ``_run_token_exchange``'s + ``try``/``finally``, so an escaping state error would skip ``ctx.next()`` and + stall the middleware chain. The persisted layer is best-effort by design -- the + in-memory guard is the authoritative same-instance one -- so an unreadable + store degrades to in-memory-only dedup rather than taking the turn down. - def _record_completed_token_exchange( + ``Exception``, not ``BaseException``: a cancellation still belongs to the task. + """ + try: + return has_completed_token_exchange(ctx.state, exchange_id) + except Exception: + logger.exception("Unable to read persisted OAuth token exchange state; deduplicating in memory only.") + return False + + async def _record_completed_token_exchange( self, ctx: ActivityContext[SignInTokenExchangeInvokeActivity], exchange_id: str ) -> None: if not exchange_id: @@ -277,7 +320,17 @@ def _record_completed_token_exchange( # Persisted too, so a duplicate handled by another process instance still sees # it. Best-effort only: state has no compare-and-set, so the in-memory layer # above remains the authoritative same-instance guard. - record_completed_token_exchange(ctx.state, exchange_id) + # + # Flushed mid-turn rather than at end of turn: the owner still has sign-in + # callbacks to run, and a duplicate racing on another instance would otherwise + # load a snapshot with no marker and redeem the exchange a second time. This + # mirrors the mid-turn save ``ctx.sign_in()`` performs for its pending hint. + try: + record_completed_token_exchange(ctx.state, exchange_id) + if ctx.state is not None: + await ctx.state._save() # pyright: ignore[reportPrivateUsage] + except Exception: + logger.exception("Unable to persist completed OAuth token exchange; deduplicating in memory only.") def _prune_completed_token_exchanges(self) -> None: cutoff = time() - TOKEN_EXCHANGE_DEDUP_TTL_SECONDS @@ -307,9 +360,15 @@ def _stamp_completed_token_exchange( Only the in-memory marker is left alone here, so the TTL stays anchored to the moment the token was actually redeemed rather than being extended by every duplicate that arrives. + + Best-effort like every other persisted-marker touch: a state failure must not + turn a successful duplicate into a failed invoke response. """ - if not has_completed_token_exchange(ctx.state, exchange_id): - record_completed_token_exchange(ctx.state, exchange_id) + try: + if not has_completed_token_exchange(ctx.state, exchange_id): + record_completed_token_exchange(ctx.state, exchange_id) + except Exception: + logger.exception("Unable to stamp completed OAuth token exchange into turn state.") def _replay_completed_token_exchange( self, ctx: ActivityContext[SignInTokenExchangeInvokeActivity], exchange_id: str @@ -317,7 +376,7 @@ def _replay_completed_token_exchange( """Answer a duplicate that arrived after its exchange already completed.""" logger.debug("Duplicate signin/tokenExchange with id '%s' - returning 200 no-op.", exchange_id) self._stamp_completed_token_exchange(ctx, exchange_id) - connection_name = ctx.activity.value.connection_name + connection_name = self._resolved_connection_name(ctx.activity.value.connection_name) started_at = perf_counter() try: with get_tracer().start_as_current_span( @@ -348,9 +407,10 @@ async def _await_token_exchange( The waiter mirrors whatever the owning request produced, so a caller that lost the race still learns that the exchange failed (``412``) instead of being told - the sign-in succeeded. + the sign-in succeeded. It mirrors the owner's *result*, never its exception + object -- see the failure branch below. """ - connection_name = ctx.activity.value.connection_name + connection_name = self._resolved_connection_name(ctx.activity.value.connection_name) result = APP_OAUTH_RESULTS.duplicate started_at = perf_counter() try: @@ -364,12 +424,30 @@ async def _await_token_exchange( # Shielded: cancelling this waiter must not cancel the future the # owning request still has to resolve. outcome = await asyncio.shield(in_flight) + # Stamped before the failure check: an exchange can fail after the + # token was already spent, and the marker still has to survive this + # request's own last-write-wins save. + if outcome.token_redeemed: + self._stamp_completed_token_exchange(ctx, exchange_id) if outcome.error is not None: result = APP_OAUTH_RESULTS.failure span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_result, result) - raise outcome.error - if outcome.token_redeemed: - self._stamp_completed_token_exchange(ctx, exchange_id) + span.set_attribute(APP_ATTRIBUTE_NAMES.invoke_response_status, 412) + # Reported as a 412 rather than re-raised. Re-raising would hand + # this task an exception it never incurred: a ``CancelledError`` + # from the owner would make an uncancelled waiter report itself as + # cancelled and trip ``except CancelledError`` cleanup, and one + # exception object shared between several waiters would have them + # all append frames to the same ``__traceback__``. Mirroring the + # result is what the TypeScript SDK does and what Teams needs. + return InvokeResponse( + status=412, + body=TokenExchangeInvokeResponse( + id=exchange_id, + connection_name=connection_name, + failure_detail=str(outcome.error) or "unable to exchange token...", + ), + ) response = outcome.response # The owning request signals success by returning ``None``, which the # activity processor materializes as a 200. Duplicates say so diff --git a/packages/apps/tests/test_app_oauth_dedup.py b/packages/apps/tests/test_app_oauth_dedup.py index 43eac165..6ac8bc1d 100644 --- a/packages/apps/tests/test_app_oauth_dedup.py +++ b/packages/apps/tests/test_app_oauth_dedup.py @@ -792,3 +792,170 @@ async def on_failure(event): assert failures == ["invokeerror", "invokeerror"] assert first.next.await_count == 1 assert second.next.await_count == 1 + + +class TestDedupFailureIsolation: + """Dedup must not turn a state hiccup, a cancellation, or a post-redemption + failure into a stalled turn or a lost completion marker.""" + + @pytest.mark.asyncio + async def test_unreadable_marker_state_still_completes_the_turn(self, monkeypatch): + """An unreadable store degrades to in-memory dedup instead of stalling. + + The persisted read happens before ``_run_token_exchange`` opens its + ``try``/``finally``, so an escaping error would skip ``ctx.next()`` and leave + the middleware chain hanging. + """ + handlers, emitter, _flow = make_handlers() + monkeypatch.setattr( + "microsoft_teams.apps.app_oauth.has_completed_token_exchange", + MagicMock(side_effect=RuntimeError("state store unavailable")), + ) + + api = make_api() + ctx = make_context(exchange_activity(), api, make_state()) + + assert await handlers.sign_in_token_exchange(ctx) is None + assert api.users.exchange_token.await_count == 1 + assert len(emitted(emitter, "sign_in")) == 1 + assert ctx.next.await_count == 1 + + @pytest.mark.asyncio + async def test_completed_marker_is_flushed_before_sign_in_callbacks_run(self): + """The marker reaches storage mid-turn, not at end of turn. + + The owner still has callbacks to run after redeeming the token. A duplicate + racing on another process instance loads its own snapshot, so the marker has + to be durable before those callbacks start or the duplicate exchanges again. + """ + handlers, _emitter, flow = make_handlers() + state = make_state() + marker_key = f"{EXCHANGE_STATE_KEY_PREFIX}exchange-1" + order: List[str] = [] + + async def capture_save() -> None: + order.append("save" if marker_key in state.conversation else "save-without-marker") + + state._save = capture_save + + @flow.on_signin + async def on_signin(_event): + order.append("signin") + + ctx = make_context(exchange_activity(), make_api(), state) + await handlers.sign_in_token_exchange(ctx) + + assert order == ["save", "signin"] + + @pytest.mark.asyncio + async def test_failure_after_redemption_still_lets_duplicates_stamp_the_marker(self): + """A failure *after* the token is spent must not cost the completion marker. + + ``_clear_pending`` runs once the marker is already recorded, which is the + window this covers. A raising sign-in handler cannot reach it -- the flow + isolates listener failures -- so a state operation is the honest trigger. + + Both halves of the bug are load-bearing here: if the owner reports + ``token_redeemed=False``, or if the waiter checks ``outcome.error`` before + stamping, the waiter's last-write-wins save erases the owner's completion and + a later duplicate redeems the spent exchange again. + """ + handlers, _emitter, _flow = make_handlers() + handlers.oauth_registry._clear_pending = MagicMock(side_effect=RuntimeError("state store unavailable")) + + api = make_api(exchange=slow_exchange()) + owner_state, waiter_state = make_state(), make_state() + owner = make_context(exchange_activity(), api, owner_state) + waiter = make_context(exchange_activity(), api, waiter_state) + + owner_result, waiter_result = await asyncio.gather( + handlers.sign_in_token_exchange(owner), + handlers.sign_in_token_exchange(waiter), + return_exceptions=True, + ) + + marker_key = f"{EXCHANGE_STATE_KEY_PREFIX}exchange-1" + assert api.users.exchange_token.await_count == 1 + assert isinstance(owner_result, RuntimeError) + assert marker_key in owner_state.conversation + assert isinstance(waiter_result, InvokeResponse) + assert waiter_result.status == 412 + assert marker_key in waiter_state.conversation + + @pytest.mark.asyncio + async def test_cancelling_the_owner_does_not_cancel_its_duplicate(self): + """A duplicate never incurred the owner's cancellation, so it must not report one. + + Re-raising the owner's ``CancelledError`` would make an uncancelled task claim + it was cancelled and trip any ``except CancelledError`` cleanup above it. + """ + handlers, _emitter, _flow = make_handlers() + api = make_api(exchange=slow_exchange(delay=0.05)) + owner = make_context(exchange_activity(), api) + waiter = make_context(exchange_activity(), api) + + owner_task = asyncio.create_task(handlers.sign_in_token_exchange(owner)) + await asyncio.sleep(0.01) + waiter_task = asyncio.create_task(handlers.sign_in_token_exchange(waiter)) + await asyncio.sleep(0.01) + owner_task.cancel() + + result = await waiter_task + + assert owner_task.cancelled() + assert not waiter_task.cancelled() + assert isinstance(result, InvokeResponse) + assert result.status == 412 + + @pytest.mark.asyncio + async def test_duplicates_are_not_handed_the_owners_exception_object(self): + """Several waiters must not share one exception instance. + + A shared instance has every waiter appending frames to the same + ``__traceback__``, so each report contaminates the others. + """ + handlers, _emitter, _flow = make_handlers() + api = make_api(exchange=slow_exchange(RuntimeError("token service exploded"))) + contexts = [make_context(exchange_activity(), api) for _ in range(3)] + + owner_result, *waiter_results = await asyncio.gather( + *(handlers.sign_in_token_exchange(ctx) for ctx in contexts), + return_exceptions=True, + ) + + assert api.users.exchange_token.await_count == 1 + assert isinstance(owner_result, RuntimeError) + assert [status_of(result) for result in waiter_results] == [412, 412] + for result in waiter_results: + assert isinstance(result, InvokeResponse) + assert waiter_results[0] is not waiter_results[1] + + @pytest.mark.asyncio + async def test_duplicate_paths_report_the_registered_connection_casing(self, monkeypatch): + """Teams echoes the card's casing back, which must not split one connection + into several telemetry series.""" + emitter = MagicMock(spec=EventEmitter) + registry = OAuthFlowRegistry() + registry.add(OAuthFlow("Test-Connection")) + handlers = OauthHandlers("Test-Connection", emitter, registry) + + recorded: List[str] = [] + monkeypatch.setattr( + "microsoft_teams.apps.app_oauth.record_oauth_operation", + lambda connection_name, *_args, **_kwargs: recorded.append(connection_name), + ) + + api = make_api(exchange=slow_exchange()) + activity = exchange_activity() + activity.value.connection_name = "test-connection" + + # Concurrent pair exercises the in-flight waiter, then a late request + # exercises the completed-marker replay. + await asyncio.gather( + handlers.sign_in_token_exchange(make_context(activity, api)), + handlers.sign_in_token_exchange(make_context(activity, api)), + ) + await handlers.sign_in_token_exchange(make_context(activity, api)) + + assert recorded.count("Test-Connection") == 2 + assert "test-connection" not in recorded[1:] From 37d907b479b3de66edda00df20552e81f318df00 Mon Sep 17 00:00:00 2001 From: lilydu Date: Thu, 27 Aug 2026 17:15:40 -0700 Subject: [PATCH 06/10] fix(oauth): report the registered connection casing to token-exchange telemetry Teams echoes back whatever casing the sign-in card carried, so an app that registers "Graph" can receive "graph" on the wire. `_run_token_exchange` already resolved the registered name for the `SignInEvent`, but reported the raw name to the span attribute, both error counters and the operations and duration metrics, splitting one connection into several series. The registry lookup is hoisted above the `try` so the metric write in `finally` can always reach it. Resolving it inside the `try`, as before, left the name unbound whenever anything above it raised, which would have turned that `finally` into an `UnboundLocalError` masking the original exception. Deliberately unchanged: the name sent to the Token Service stays as Teams sent it, since rewriting the wire value is a behavior change rather than a telemetry fix, and the unregistered-connection warning keeps the raw name -- it only fires when no flow matched, where raw and registered are the same string anyway. This also makes the owner path agree with the duplicate paths, which were canonicalized in the previous commit. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../src/microsoft_teams/apps/app_oauth.py | 21 ++-- packages/apps/tests/test_app_oauth_dedup.py | 108 +++++++++++++++++- 2 files changed, 120 insertions(+), 9 deletions(-) diff --git a/packages/apps/src/microsoft_teams/apps/app_oauth.py b/packages/apps/src/microsoft_teams/apps/app_oauth.py index d4e8fdd2..f8824815 100644 --- a/packages/apps/src/microsoft_teams/apps/app_oauth.py +++ b/packages/apps/src/microsoft_teams/apps/app_oauth.py @@ -144,6 +144,16 @@ async def _run_token_exchange(self, ctx: ActivityContext[SignInTokenExchangeInvo api = ctx.api next_handler = ctx.next connection_name = activity.value.connection_name + # Resolved before the ``try`` so the ``finally`` below can always report it. The + # lookup is a plain dict access; computing it inside the ``try`` left it unbound + # whenever anything above it raised, which would turn the metric write in + # ``finally`` into an ``UnboundLocalError`` masking the original exception. + flow = self.oauth_registry.get(connection_name) + # Teams echoes back whatever casing the sign-in card carried, so telemetry keyed + # on the raw name splits one connection into several series. ``connection_name`` + # stays raw: it is what goes on the wire to the Token Service, and rewriting that + # would be a behavior change rather than a telemetry fix. + event_connection_name = flow.connection_name if flow is not None else connection_name result = APP_OAUTH_RESULTS.failure started_at = perf_counter() try: @@ -152,12 +162,9 @@ async def _run_token_exchange(self, ctx: ActivityContext[SignInTokenExchangeInvo record_exception=False, set_status_on_exception=False, ) as span: - span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_connection, connection_name) + span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_connection, event_connection_name) span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_operation, APP_OAUTH_OPERATIONS.token_exchange) - flow = self.oauth_registry.get(connection_name) - event_connection_name = flow.connection_name if flow is not None else connection_name - if ( connection_lookup_key(connection_name) != connection_lookup_key(self.default_connection_name) and flow is None @@ -194,7 +201,7 @@ async def _run_token_exchange(self, ctx: ActivityContext[SignInTokenExchangeInvo error_type = APP_OAUTH_ERROR_TYPES.http_error span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_error_type, error_type) record_exception(span, e) - record_oauth_error(connection_name, APP_OAUTH_OPERATIONS.token_exchange, error_type) + record_oauth_error(event_connection_name, APP_OAUTH_OPERATIONS.token_exchange, error_type) status = status or 500 result = APP_OAUTH_RESULTS.failure span.set_attribute(APP_ATTRIBUTE_NAMES.invoke_response_status, status) @@ -230,7 +237,7 @@ async def _run_token_exchange(self, ctx: ActivityContext[SignInTokenExchangeInvo error_type = APP_OAUTH_ERROR_TYPES.exception span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_error_type, error_type) record_exception(span, e) - record_oauth_error(connection_name, APP_OAUTH_OPERATIONS.token_exchange, error_type) + record_oauth_error(event_connection_name, APP_OAUTH_OPERATIONS.token_exchange, error_type) result = APP_OAUTH_RESULTS.failure span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_result, result) raise @@ -259,7 +266,7 @@ async def _run_token_exchange(self, ctx: ActivityContext[SignInTokenExchangeInvo return None finally: record_oauth_operation( - connection_name, + event_connection_name, APP_OAUTH_OPERATIONS.token_exchange, result, (perf_counter() - started_at) * 1000, diff --git a/packages/apps/tests/test_app_oauth_dedup.py b/packages/apps/tests/test_app_oauth_dedup.py index 6ac8bc1d..2f2f9fca 100644 --- a/packages/apps/tests/test_app_oauth_dedup.py +++ b/packages/apps/tests/test_app_oauth_dedup.py @@ -957,5 +957,109 @@ async def test_duplicate_paths_report_the_registered_connection_casing(self, mon ) await handlers.sign_in_token_exchange(make_context(activity, api)) - assert recorded.count("Test-Connection") == 2 - assert "test-connection" not in recorded[1:] + # Owner, in-flight waiter and late replay all land on one series. + assert recorded == ["Test-Connection"] * 3 + assert "test-connection" not in recorded + + +class TestConnectionNameTelemetryCasing: + """Teams echoes back whatever casing the sign-in card carried, so one connection + must not fan out into several telemetry series.""" + + @staticmethod + def _capture(monkeypatch) -> tuple[List[str], List[str]]: + operations: List[str] = [] + errors: List[str] = [] + monkeypatch.setattr( + "microsoft_teams.apps.app_oauth.record_oauth_operation", + lambda connection_name, *_a, **_kw: operations.append(connection_name), + ) + monkeypatch.setattr( + "microsoft_teams.apps.app_oauth.record_oauth_error", + lambda connection_name, *_a, **_kw: errors.append(connection_name), + ) + return operations, errors + + @staticmethod + def _handlers() -> OauthHandlers: + registry = OAuthFlowRegistry() + registry.add(OAuthFlow("Test-Connection")) + return OauthHandlers("Test-Connection", MagicMock(spec=EventEmitter), registry) + + @staticmethod + def _activity() -> SignInTokenExchangeInvokeActivity: + activity = exchange_activity() + # The casing Teams echoes back, which differs from the registered name. + activity.value.connection_name = "test-connection" + return activity + + @pytest.mark.asyncio + async def test_success_reports_the_registered_casing(self, monkeypatch): + operations, _errors = self._capture(monkeypatch) + handlers = self._handlers() + api = make_api() + + await handlers.sign_in_token_exchange(make_context(self._activity(), api)) + + assert operations == ["Test-Connection"] + # The wire call keeps the name Teams sent: canonicalizing it would change what + # reaches the Token Service, which is a behavior change, not a telemetry fix. + assert api.users.exchange_token.await_args.args[0].connection_name == "test-connection" + + @pytest.mark.asyncio + async def test_failure_reports_the_registered_casing(self, monkeypatch): + """The failure path is what exercises the ``finally`` after an exception. + + A 500 rather than a 400: 404/400/412 are expected exchange misses that report + no error, so only an unexpected status reaches ``record_oauth_error``. + """ + operations, errors = self._capture(monkeypatch) + handlers = self._handlers() + api = make_api(exchange=AsyncMock(side_effect=oauth_http_error(500))) + + await handlers.sign_in_token_exchange(make_context(self._activity(), api)) + + assert operations == ["Test-Connection"] + assert errors == ["Test-Connection"] + + @pytest.mark.asyncio + async def test_expected_miss_reports_the_registered_casing(self, monkeypatch): + """A 412 fall-back-to-card miss still has to land on the canonical series.""" + operations, errors = self._capture(monkeypatch) + handlers = self._handlers() + api = make_api(exchange=AsyncMock(side_effect=oauth_http_error(404))) + + result = await handlers.sign_in_token_exchange(make_context(self._activity(), api)) + + assert isinstance(result, InvokeResponse) + assert result.status == 412 + assert operations == ["Test-Connection"] + assert errors == [] + + @pytest.mark.asyncio + async def test_registry_failure_does_not_mask_itself_in_the_finally(self, monkeypatch): + """The name must be resolved before the ``try``. + + Resolving inside it left the variable unbound when anything above raised, so + the metric write in ``finally`` died with ``UnboundLocalError`` and buried the + real exception. + """ + self._capture(monkeypatch) + handlers = self._handlers() + handlers.oauth_registry.get = MagicMock(side_effect=RuntimeError("registry exploded")) + + with pytest.raises(RuntimeError, match="registry exploded"): + await handlers.sign_in_token_exchange(make_context(self._activity(), make_api())) + + @pytest.mark.asyncio + async def test_unregistered_connection_is_reported_as_teams_sent_it(self, monkeypatch): + """An unknown connection has no registered casing to fall back to, and the + diagnostic is only useful if it shows what actually arrived.""" + operations, _errors = self._capture(monkeypatch) + handlers = self._handlers() + activity = exchange_activity() + activity.value.connection_name = "not-registered" + + await handlers.sign_in_token_exchange(make_context(activity, make_api())) + + assert operations == ["not-registered"] From aad1e48a300d06d145387d1663c6ae187d35f0fc Mon Sep 17 00:00:00 2001 From: lilydu Date: Fri, 28 Aug 2026 09:16:07 -0700 Subject: [PATCH 07/10] fix(oauth-example): strip the bot mention before matching commands The example is @mentioned in group chats and channels, so the activity text arrives as "BotName sign in profile" and never matches the command strings. Every command silently did nothing outside 1:1 chat, which defeats the point of an example meant to demonstrate multi-connection OAuth across scopes. Verified in a group chat and a channel. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- examples/oauth/src/main.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/examples/oauth/src/main.py b/examples/oauth/src/main.py index 334a57b1..7dfd459d 100644 --- a/examples/oauth/src/main.py +++ b/examples/oauth/src/main.py @@ -70,7 +70,9 @@ async def on_any_signin(event: SignInEvent) -> None: @app.on_message async def handle_message(ctx: ActivityContext[MessageActivity]) -> None: - text = (ctx.activity.text or "").strip().lower() + # The bot is @mentioned in group chats and channels, so drop the mention + # before dispatching to keep the same commands working in every scope. + text = (ctx.activity.strip_mentions_text().text or "").strip().lower() if text == "sign in profile": # Returns the token directly when one is already cached, otherwise sends a From e3ba811124bcf4132d41e2f037c0d18fffb6e265 Mon Sep 17 00:00:00 2001 From: lilydu Date: Fri, 28 Aug 2026 10:11:18 -0700 Subject: [PATCH 08/10] refactor(oauth): rename get_token_status and drop the flow's hop through the deprecated API `ctx.get_token_status` shadowed `api.users.get_token_status` while behaving differently: the context method adds a silent-SSO correction pass that the Token Service endpoint does not. One name, two layers, two behaviors. Rename the context method to `get_connection_status`, matching `getConnectionStatus` in teams.ts and `GetConnectionStatusAsync` in teams.net. The API client keeps its name, since that one is the transport endpoint and has already shipped. The context method landed in #561, after every published tag, so no released version ever exposed the old name. Point `OAuthFlow.sign_out`, `get_token` and `is_signed_in` straight at `api.users.*`, as teams.ts does, instead of delegating to the deprecated single-connection methods on `ActivityContext`. Verified equivalent: identical request params, log output, return values, and 404/non-404 handling. `sign_in` still delegates; extracting its card, pending-hint and rollback logic is a follow-up. Mark the legacy single-connection OAuth surface deprecated in docstrings only, matching the JSDoc-only approach in teams.ts. No decorators, so the supported flow API never warns callers about an API they did not call. Migrate examples/graph to the flow API, fixing the example's own version of the multi-connection bug: it built a Graph client from `ctx.user_token`, which is whichever connection last exchanged, rather than from the Graph connection's token. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- examples/graph/src/main.py | 26 +++---- .../src/microsoft_teams/apps/oauth_flow.py | 27 ++++++- .../apps/src/microsoft_teams/apps/options.py | 14 +++- .../apps/routing/activity_context.py | 37 +++++++++- packages/apps/tests/test_activity_context.py | 55 ++++++++++----- packages/apps/tests/test_oauth_flow.py | 70 +++++++++++++++---- 6 files changed, 181 insertions(+), 48 deletions(-) diff --git a/examples/graph/src/main.py b/examples/graph/src/main.py index dad84d1c..f36df860 100644 --- a/examples/graph/src/main.py +++ b/examples/graph/src/main.py @@ -9,7 +9,7 @@ from azure.core.exceptions import ClientAuthenticationError from microsoft_teams.api import MessageActivity -from microsoft_teams.apps import ActivityContext, App, AppOptions, ErrorEvent, SignInEvent +from microsoft_teams.apps import ActivityContext, App, ErrorEvent, SignInEvent from microsoft_teams.common import ConsoleFormatter from microsoft_teams.graph import get_graph_client from msgraph.generated.users.item.messages.messages_request_builder import ( # type: ignore @@ -23,8 +23,8 @@ logging.getLogger().addHandler(stream_handler) logger = logging.getLogger(__name__) -app_options = AppOptions(default_connection_name=os.getenv("CONNECTION_NAME", "graph")) -app = App(**app_options) +app = App() +graph_flow = app.add_oauth_flow(os.getenv("CONNECTION_NAME", "graph")) async def get_authenticated_graph_client(ctx: ActivityContext[MessageActivity]): @@ -34,40 +34,40 @@ async def get_authenticated_graph_client(ctx: ActivityContext[MessageActivity]): Returns: Graph client if successful, None if authentication failed. """ - # Check if user is signed in - if not ctx.is_signed_in: + token = await graph_flow.get_token(ctx) + if token is None: await ctx.send("🔐 Please sign in first to access Microsoft Graph.") - await ctx.sign_in() + await graph_flow.sign_in(ctx) return None try: # Create Graph client using the user token - return get_graph_client(ctx.user_token) + return get_graph_client(token) except Exception as e: logger.error(f"Failed to create Graph client: {e}") await ctx.send("🔐 Failed to create authenticated client. Please try signing in again.") - await ctx.sign_in() + await graph_flow.sign_in(ctx) return None @app.on_message_pattern("signin") async def handle_signin_command(ctx: ActivityContext[MessageActivity]): """Handle sign-in command.""" - if ctx.is_signed_in: + if await graph_flow.is_signed_in(ctx): await ctx.send("✅ You are already signed in!") else: await ctx.send("🔐 Please sign in to access Microsoft Graph...") - await ctx.sign_in() + await graph_flow.sign_in(ctx) @app.on_message_pattern("signout") async def handle_signout_command(ctx: ActivityContext[MessageActivity]): """Handle sign-out command.""" - if not ctx.is_signed_in: + if not await graph_flow.is_signed_in(ctx): await ctx.send("ℹ️ You are not currently signed in.") else: - await ctx.sign_out() + await graph_flow.sign_out(ctx) await ctx.send("👋 You have been signed out successfully!") @@ -98,7 +98,7 @@ async def handle_profile_command(ctx: ActivityContext[MessageActivity]): except ClientAuthenticationError as e: logger.error(f"Authentication error: {e}") await ctx.send("🔐 Authentication failed. Please try signing in again.") - await ctx.sign_in() + await graph_flow.sign_in(ctx) except Exception as e: logger.error(f"Error getting profile: {e}") diff --git a/packages/apps/src/microsoft_teams/apps/oauth_flow.py b/packages/apps/src/microsoft_teams/apps/oauth_flow.py index aa198fd3..38df8f64 100644 --- a/packages/apps/src/microsoft_teams/apps/oauth_flow.py +++ b/packages/apps/src/microsoft_teams/apps/oauth_flow.py @@ -9,6 +9,9 @@ from dataclasses import replace from typing import Any, Awaitable, Callable, Iterator, List, Mapping, Optional, Sequence, Tuple, TypeVar, Union +from httpx import HTTPStatusError +from microsoft_teams.api import GetUserTokenParams, SignOutUserParams + from .events import SignInEvent, SignInFailureEvent from .oauth_connection import connection_lookup_key, normalize_connection_name from .oauth_state import ( @@ -150,15 +153,33 @@ async def sign_in(self, ctx: ActivityContext[Any], options: Optional[SignInOptio async def sign_out(self, ctx: ActivityContext[Any]) -> None: """Sign the user out of this connection.""" - await ctx.sign_out(connection_name=self.connection_name) + await ctx.api.users.sign_out( + SignOutUserParams( + channel_id=ctx.activity.channel_id, + user_id=ctx.activity.from_.id, + connection_name=self.connection_name, + ) + ) async def get_token(self, ctx: ActivityContext[Any]) -> Optional[str]: """The user's token for this connection, or ``None`` if not signed in.""" - return await ctx.get_user_token(connection_name=self.connection_name) + try: + res = await ctx.api.users.get_token( + GetUserTokenParams( + channel_id=ctx.activity.channel_id, + user_id=ctx.activity.from_.id, + connection_name=self.connection_name, + ) + ) + return res.token + except HTTPStatusError as e: + if e.response.status_code == 404: + return None + raise async def is_signed_in(self, ctx: ActivityContext[Any]) -> bool: """Whether the user currently has a token for this connection.""" - return await ctx.get_user_token(connection_name=self.connection_name) is not None + return await self.get_token(ctx) is not None class OAuthFlowRegistry(Mapping[str, OAuthFlow]): diff --git a/packages/apps/src/microsoft_teams/apps/options.py b/packages/apps/src/microsoft_teams/apps/options.py index 56fa2f69..8f4eb227 100644 --- a/packages/apps/src/microsoft_teams/apps/options.py +++ b/packages/apps/src/microsoft_teams/apps/options.py @@ -120,7 +120,14 @@ class AppOptions(TypedDict, total=False): # OAuth default_connection_name: Optional[str] - """The OAuth connection name to use for authentication. Defaults to 'graph'.""" + """The OAuth connection name to use for authentication. Defaults to 'graph'. + + .. deprecated:: + Names a single connection for the whole app, which does not generalize + past one. Register each connection with ``app.add_oauth_flow(name)`` and + hold on to the returned ``OAuthFlow`` instead. Still honoured as the + default for the deprecated ``ctx.sign_in`` / ``ctx.sign_out`` / + ``ctx.get_user_token`` surface.""" fetch_user_token: Optional[bool] """Whether to eagerly look up the user's OAuth token on every inbound activity. @@ -163,7 +170,10 @@ class InternalAppOptions: dangerously_allow_unauthenticated_requests: bool = False """Whether to accept incoming requests without JWT validation.""" default_connection_name: str = "graph" - """The OAuth connection name to use for authentication.""" + """The OAuth connection name to use for authentication. + + .. deprecated:: + Use ``app.add_oauth_flow(name)`` and the returned ``OAuthFlow``.""" fetch_user_token: bool = False """When True, eagerly looks up the user's OAuth token on every inbound activity. The token is used to compute ``ctx.is_signed_in`` and ``ctx.user_token``, and to authenticate diff --git a/packages/apps/src/microsoft_teams/apps/routing/activity_context.py b/packages/apps/src/microsoft_teams/apps/routing/activity_context.py index 57863f2c..14aec1b3 100644 --- a/packages/apps/src/microsoft_teams/apps/routing/activity_context.py +++ b/packages/apps/src/microsoft_teams/apps/routing/activity_context.py @@ -97,7 +97,18 @@ class SignInOptions: class ActivityContext(Generic[T]): - """Context object passed to activity handlers with middleware support.""" + """Context object passed to activity handlers with middleware support. + + .. note:: + The single-connection OAuth surface here - :attr:`is_signed_in`, + :attr:`user_token`, :attr:`user_graph`, :meth:`sign_in`, + :meth:`sign_out` and :meth:`get_user_token` - predates + multi-connection support and is deprecated. + + Register connections with + ``app.add_oauth_flow(name)`` and use the returned ``OAuthFlow``, which + pins every operation to the connection that owns it. + """ def __init__( self, @@ -162,6 +173,14 @@ def user_graph(self) -> "GraphServiceClient": """ Get a Microsoft Graph client configured with the user's token. + .. deprecated:: + Built from :attr:`user_token`, which holds the token for whichever + connection signed in most recently rather than the one that talks to + Microsoft Graph. + + Read the token from the owning flow instead:: + token = await app.get_oauth_flow("graph").get_token(ctx) + Raises: ValueError: If the user is not signed in or doesn't have a valid token. RuntimeError: If the graph client cannot be created. @@ -350,6 +369,12 @@ async def sign_in(self, options: Optional[SignInOptions] = None) -> Optional[str """ Initiate a sign-in flow for the user. + .. deprecated:: + Targets the app's single ``connection_name``, which does not + generalize past one connection. Use + ``app.get_oauth_flow(name).sign_in(ctx, options)``, which pins the + connection to the flow that owns it. + Args: options: Optional signin options to customize the flow @@ -480,6 +505,10 @@ async def sign_out(self, connection_name: Optional[str] = None) -> None: This method will remove the user's token from the storage. + .. deprecated:: + Use ``app.get_oauth_flow(name).sign_out(ctx)``, which cannot sign + out of the wrong connection. + Args: connection_name: The connection to sign out of. Defaults to the app's default connection. @@ -502,6 +531,10 @@ async def get_user_token(self, connection_name: Optional[str] = None) -> Optiona """ Get the user's token for a connection. + .. deprecated:: + Use ``app.get_oauth_flow(name).get_token(ctx)``, which reads the + token from the flow that owns the connection. + Args: connection_name: The connection to read. Defaults to the app's default connection. @@ -530,7 +563,7 @@ async def get_user_token(self, connection_name: Optional[str] = None) -> Optiona return None raise - async def get_token_status(self) -> List[TokenStatus]: + async def get_connection_status(self) -> List[TokenStatus]: """ Get the token status for every OAuth connection registered on the bot. diff --git a/packages/apps/tests/test_activity_context.py b/packages/apps/tests/test_activity_context.py index bc9567a0..bf2198ae 100644 --- a/packages/apps/tests/test_activity_context.py +++ b/packages/apps/tests/test_activity_context.py @@ -1079,7 +1079,7 @@ async def test_sign_out_propagates_token_service_failure(self) -> None: class TestActivityContextTokenHelpers: - """Tests for sign_out(connection_name=), get_user_token, and get_token_status.""" + """Tests for sign_out(connection_name=), get_user_token, and get_connection_status.""" @pytest.mark.asyncio async def test_sign_out_uses_override_connection_name(self) -> None: @@ -1190,8 +1190,8 @@ async def test_get_user_token_propagates_non_http_errors(self) -> None: await ctx.get_user_token() @pytest.mark.asyncio - async def test_get_token_status_returns_all_connections(self) -> None: - """get_token_status makes a single call and returns the status list unfiltered.""" + async def test_get_connection_status_returns_all_connections(self) -> None: + """get_connection_status makes a single call and returns the status list unfiltered.""" mock_activity = MagicMock() mock_activity.channel_id = "msteams" mock_activity.from_.id = "user-1" @@ -1200,7 +1200,7 @@ async def test_get_token_status_returns_all_connections(self) -> None: statuses = [MagicMock(), MagicMock()] ctx.api.users.get_token_status = AsyncMock(return_value=statuses) - result = await ctx.get_token_status() + result = await ctx.get_connection_status() assert result is statuses ctx.api.users.get_token_status.assert_awaited_once() @@ -1210,8 +1210,8 @@ async def test_get_token_status_returns_all_connections(self) -> None: assert params.channel_id == "msteams" @pytest.mark.asyncio - async def test_get_token_status_propagates_errors(self) -> None: - """Unlike get_user_token, get_token_status lets service failures surface.""" + async def test_get_connection_status_propagates_errors(self) -> None: + """Unlike get_user_token, get_connection_status lets service failures surface.""" mock_activity = MagicMock() mock_activity.channel_id = "msteams" mock_activity.from_.id = "user-1" @@ -1220,7 +1220,7 @@ async def test_get_token_status_propagates_errors(self) -> None: ctx.api.users.get_token_status = AsyncMock(side_effect=RuntimeError("service down")) with pytest.raises(RuntimeError): - await ctx.get_token_status() + await ctx.get_connection_status() class TestActivityContextPromptPreview: @@ -1520,7 +1520,7 @@ async def test_override_send_failure_rolls_back_pending(self) -> None: class TestGetTokenStatusRegistryAware: - """get_token_status corrects the bulk call for registered flows.""" + """get_connection_status corrects the bulk call for registered flows.""" @staticmethod def _context(connection_names: list[str] | None = None): @@ -1546,7 +1546,7 @@ async def test_without_registered_flows_the_bulk_result_is_returned_verbatim(sel ctx.api.users.get_token_status = AsyncMock(return_value=statuses) ctx.api.users.get_token = AsyncMock(side_effect=AssertionError("must not probe")) - assert await ctx.get_token_status() is statuses + assert await ctx.get_connection_status() is statuses @pytest.mark.asyncio async def test_false_status_for_a_registered_flow_is_corrected(self) -> None: @@ -1559,7 +1559,7 @@ async def test_false_status_for_a_registered_flow_is_corrected(self) -> None: ctx.api.users.get_token_status = AsyncMock(return_value=[self._status("graph", False)]) ctx.api.users.get_token = AsyncMock(return_value=MagicMock(token="a-token")) - result = await ctx.get_token_status() + result = await ctx.get_connection_status() assert [(s.connection_name, s.has_token) for s in result] == [("graph", True)] @@ -1570,7 +1570,7 @@ async def test_a_genuine_miss_stays_false(self) -> None: ctx.api.users.get_token_status = AsyncMock(return_value=[self._status("graph", False)]) ctx.api.users.get_token = AsyncMock(side_effect=_http_status_error(404)) - result = await ctx.get_token_status() + result = await ctx.get_connection_status() assert [(s.connection_name, s.has_token) for s in result] == [("graph", False)] @@ -1581,7 +1581,7 @@ async def test_a_registered_flow_missing_from_the_bulk_result_is_added(self) -> ctx.api.users.get_token_status = AsyncMock(return_value=[self._status("legacy", True)]) ctx.api.users.get_token = AsyncMock(return_value=MagicMock(token="a-token")) - result = await ctx.get_token_status() + result = await ctx.get_connection_status() assert [(s.connection_name, s.has_token) for s in result] == [("legacy", True), ("graph", True)] @@ -1599,7 +1599,7 @@ async def probe(params): ctx.api.users.get_token = AsyncMock(side_effect=probe) - result = await ctx.get_token_status() + result = await ctx.get_connection_status() assert result[0] is legacy assert probes == [] @@ -1611,7 +1611,7 @@ async def test_registry_casing_is_matched_case_insensitively(self) -> None: ctx.api.users.get_token_status = AsyncMock(return_value=[self._status("GRAPH", False)]) ctx.api.users.get_token = AsyncMock(return_value=MagicMock(token="a-token")) - result = await ctx.get_token_status() + result = await ctx.get_connection_status() # Corrected in place: one row, keeping the service's own casing. assert [(s.connection_name, s.has_token) for s in result] == [("GRAPH", True)] @@ -1625,7 +1625,7 @@ async def test_order_and_shape_are_preserved(self) -> None: ) ctx.api.users.get_token = AsyncMock(side_effect=_http_status_error(404)) - result = await ctx.get_token_status() + result = await ctx.get_connection_status() assert [s.connection_name for s in result] == ["a", "b", "c", "missing"] @@ -1637,4 +1637,27 @@ async def test_non_404_lookup_failure_propagates(self) -> None: ctx.api.users.get_token = AsyncMock(side_effect=_http_status_error(503)) with pytest.raises(HTTPStatusError): - await ctx.get_token_status() + await ctx.get_connection_status() + + +class TestConnectionStatusRename: + """``ctx.get_token_status`` was renamed to ``ctx.get_connection_status``.""" + + def test_old_context_name_is_gone(self) -> None: + """Renamed outright rather than aliased - the method was never released.""" + ctx, _ = _create_activity_context() + + assert not hasattr(ctx, "get_token_status") + assert hasattr(ctx, "get_connection_status") + + def test_api_client_keeps_the_endpoint_name(self) -> None: + """The API layer must keep its name; renaming both would undo the fix. + + ``api.users.get_token_status`` is the Token Service endpoint and returns + the service answer verbatim, while ``ctx.get_connection_status`` adds the + silent-SSO correction pass. One name for two behaviors was the problem. + """ + from microsoft_teams.api.clients.user.client import UserClient + + assert hasattr(UserClient, "get_token_status") + assert not hasattr(UserClient, "get_connection_status") diff --git a/packages/apps/tests/test_oauth_flow.py b/packages/apps/tests/test_oauth_flow.py index fbaf42ac..c5d7294b 100644 --- a/packages/apps/tests/test_oauth_flow.py +++ b/packages/apps/tests/test_oauth_flow.py @@ -9,14 +9,34 @@ from unittest.mock import AsyncMock, MagicMock import pytest +from httpx import HTTPStatusError, Request, Response from microsoft_teams.apps import App, OAuthFlow, OAuthFlowRegistry from microsoft_teams.apps.routing import SignInOptions from microsoft_teams.apps.state import StateOptions, TurnStateLoader from microsoft_teams.common.storage import LocalStorage +def _api_ctx() -> MagicMock: + """Context mock whose activity fields survive pydantic validation. + + ``sign_out`` and ``get_token`` build Token Service params themselves, so + ``channel_id`` and ``from_.id`` must be real values rather than MagicMocks. + """ + ctx = MagicMock() + ctx.activity.channel_id = "msteams" + ctx.activity.from_.id = "user-1" + return ctx + + +def _http_status_error(status_code: int) -> HTTPStatusError: + """Build an httpx.HTTPStatusError carrying the given status code.""" + request = Request("GET", "https://token.example/api/usertoken/GetToken") + response = Response(status_code, request=request) + return HTTPStatusError(f"HTTP {status_code}", request=request, response=response) + + class TestOAuthFlowOperations: - """sign_in / sign_out / get_token / is_signed_in delegate to the context.""" + """sign_in / sign_out / get_token / is_signed_in, always on this flow's connection.""" @pytest.mark.asyncio async def test_sign_in_forces_flow_connection_name(self) -> None: @@ -70,37 +90,63 @@ async def test_sign_in_without_state_remains_supported(self) -> None: @pytest.mark.asyncio async def test_sign_out_targets_flow_connection(self) -> None: flow = OAuthFlow("graph") - ctx = MagicMock() - ctx.sign_out = AsyncMock(return_value=None) + ctx = _api_ctx() + ctx.api.users.sign_out = AsyncMock(return_value=None) await flow.sign_out(ctx) - ctx.sign_out.assert_awaited_once_with(connection_name="graph") + ctx.api.users.sign_out.assert_awaited_once() + params = ctx.api.users.sign_out.call_args[0][0] + assert params.connection_name == "graph" + assert params.channel_id == "msteams" + assert params.user_id == "user-1" @pytest.mark.asyncio - async def test_get_token_returns_ctx_token(self) -> None: + async def test_get_token_targets_flow_connection(self) -> None: flow = OAuthFlow("graph") - ctx = MagicMock() - ctx.get_user_token = AsyncMock(return_value="tok") + ctx = _api_ctx() + ctx.api.users.get_token = AsyncMock(return_value=MagicMock(token="tok")) result = await flow.get_token(ctx) assert result == "tok" - ctx.get_user_token.assert_awaited_once_with(connection_name="graph") + params = ctx.api.users.get_token.call_args[0][0] + assert params.connection_name == "graph" + assert params.channel_id == "msteams" + assert params.user_id == "user-1" + + @pytest.mark.asyncio + async def test_get_token_returns_none_when_no_token_cached(self) -> None: + """404 is the Token Service saying "not signed in", not a failure.""" + flow = OAuthFlow("graph") + ctx = _api_ctx() + ctx.api.users.get_token = AsyncMock(side_effect=_http_status_error(404)) + + assert await flow.get_token(ctx) is None + + @pytest.mark.asyncio + async def test_get_token_reraises_non_404(self) -> None: + """An outage must not be reported as a logged-out user.""" + flow = OAuthFlow("graph") + ctx = _api_ctx() + ctx.api.users.get_token = AsyncMock(side_effect=_http_status_error(503)) + + with pytest.raises(HTTPStatusError): + await flow.get_token(ctx) @pytest.mark.asyncio async def test_is_signed_in_true_when_token_present(self) -> None: flow = OAuthFlow("graph") - ctx = MagicMock() - ctx.get_user_token = AsyncMock(return_value="tok") + ctx = _api_ctx() + ctx.api.users.get_token = AsyncMock(return_value=MagicMock(token="tok")) assert await flow.is_signed_in(ctx) is True @pytest.mark.asyncio async def test_is_signed_in_false_when_no_token(self) -> None: flow = OAuthFlow("graph") - ctx = MagicMock() - ctx.get_user_token = AsyncMock(return_value=None) + ctx = _api_ctx() + ctx.api.users.get_token = AsyncMock(side_effect=_http_status_error(404)) assert await flow.is_signed_in(ctx) is False From cb64af54a375edbd65d0ef54ff7111108401b8ac Mon Sep 17 00:00:00 2001 From: lilydu Date: Fri, 28 Aug 2026 10:14:18 -0700 Subject: [PATCH 09/10] style: strip trailing whitespace from OAuth deprecation docstrings Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../src/microsoft_teams/apps/routing/activity_context.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/packages/apps/src/microsoft_teams/apps/routing/activity_context.py b/packages/apps/src/microsoft_teams/apps/routing/activity_context.py index 14aec1b3..b553b534 100644 --- a/packages/apps/src/microsoft_teams/apps/routing/activity_context.py +++ b/packages/apps/src/microsoft_teams/apps/routing/activity_context.py @@ -103,8 +103,8 @@ class ActivityContext(Generic[T]): The single-connection OAuth surface here - :attr:`is_signed_in`, :attr:`user_token`, :attr:`user_graph`, :meth:`sign_in`, :meth:`sign_out` and :meth:`get_user_token` - predates - multi-connection support and is deprecated. - + multi-connection support and is deprecated. + Register connections with ``app.add_oauth_flow(name)`` and use the returned ``OAuthFlow``, which pins every operation to the connection that owns it. @@ -177,7 +177,7 @@ def user_graph(self) -> "GraphServiceClient": Built from :attr:`user_token`, which holds the token for whichever connection signed in most recently rather than the one that talks to Microsoft Graph. - + Read the token from the owning flow instead:: token = await app.get_oauth_flow("graph").get_token(ctx) From 7d23a0478cbc30862a9c31e36e1ab153493fcaeb Mon Sep 17 00:00:00 2001 From: lilydu Date: Fri, 28 Aug 2026 11:44:47 -0700 Subject: [PATCH 10/10] fix(oauth): clear pending sign-in hints to match C# and TS A failed sign-in left its pending hint behind, so a later callback could be attributed to an attempt that was already over. Retire the hint on every path that ends a sign-in, matching the C# and TS SDKs: - signin/failure clears both the base hint and its SSO marker, on every flow it notifies (previously only the SSO marker, only on a resolved target) - signin/tokenExchange clears on an unexpected status - signin/verifyState clears on an unexpected status Expected 400/404/412 responses still keep the hint, since those hand off to the interactive fallback rather than ending the sign-in. Hints are cleared before the failure events are emitted, not inside the callback loop. A global sign_in_failure handler may start a fresh sign-in, and clearing afterwards would wipe the replacement hint it just recorded. C# clears before its callback likewise; TS clears immediately before each flow's handler and has no global event in between. This retires mark_pending_oauth_sso_consumed and its helper chain, which only existed to keep the base hint alive past a failure. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../src/microsoft_teams/apps/app_oauth.py | 15 ++- .../src/microsoft_teams/apps/oauth_flow.py | 5 - .../apps/oauth_pending_local.py | 12 -- .../src/microsoft_teams/apps/oauth_state.py | 25 ---- packages/apps/tests/test_app_oauth.py | 127 +++++++++++++++--- .../apps/tests/test_oauth_pending_local.py | 10 +- packages/apps/tests/test_oauth_state.py | 17 +-- 7 files changed, 132 insertions(+), 79 deletions(-) diff --git a/packages/apps/src/microsoft_teams/apps/app_oauth.py b/packages/apps/src/microsoft_teams/apps/app_oauth.py index f8824815..7be5c9f5 100644 --- a/packages/apps/src/microsoft_teams/apps/app_oauth.py +++ b/packages/apps/src/microsoft_teams/apps/app_oauth.py @@ -190,6 +190,9 @@ async def _run_token_exchange(self, ctx: ActivityContext[SignInTokenExchangeInvo except HTTPStatusError as e: status = e.response.status_code if status not in (404, 400, 412): + self.oauth_registry._clear_pending( # pyright: ignore[reportPrivateUsage] + ctx, event_connection_name + ) logger.error( f"Error exchanging token for user {activity.from_.id} in " f"conversation {activity.conversation.id}: {e}" @@ -558,9 +561,11 @@ async def sign_in_failure( f"registration has 'Expose an API' configured with the correct " f"Application ID URI matching your OAuth connection's Token Exchange URL." ) - if target_flow is not None: - self.oauth_registry._mark_sso_consumed( # pyright: ignore[reportPrivateUsage] - ctx, target_flow.connection_name + callback_flows = [target_flow] if target_flow is not None else registered_flows + + for flow in callback_flows: + self.oauth_registry._clear_pending( # pyright: ignore[reportPrivateUsage] + ctx, flow.connection_name ) await self.event_emitter.emit_async( "error", @@ -578,7 +583,6 @@ async def sign_in_failure( await self.event_emitter.emit_async("sign_in_failure", event) span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_callback_invoked, True) span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_result, result) - callback_flows = [target_flow] if target_flow is not None else registered_flows for flow in callback_flows: flow_event = ( event @@ -696,6 +700,9 @@ async def sign_in_verify_state( f"{activity.conversation.id} (HTTP {status})." ) continue + self.oauth_registry._clear_pending( # pyright: ignore[reportPrivateUsage] + ctx, connection_name + ) logger.error( f"Error verifying sign-in state for user {activity.from_.id} in conversation" f"{activity.conversation.id}: {e}" diff --git a/packages/apps/src/microsoft_teams/apps/oauth_flow.py b/packages/apps/src/microsoft_teams/apps/oauth_flow.py index 38df8f64..9689ca27 100644 --- a/packages/apps/src/microsoft_teams/apps/oauth_flow.py +++ b/packages/apps/src/microsoft_teams/apps/oauth_flow.py @@ -17,7 +17,6 @@ from .oauth_state import ( clear_pending_oauth_sign_in, get_pending_oauth_sign_ins, - mark_pending_oauth_sso_consumed, ) from .routing import ActivityContext, SignInOptions @@ -240,10 +239,6 @@ def _clear_pending(self, ctx: ActivityContext[Any], connection_name: Optional[st conversation_id, user_id = _pending_scope(ctx) clear_pending_oauth_sign_in(ctx.state, connection_name, conversation_id, user_id) - def _mark_sso_consumed(self, ctx: ActivityContext[Any], connection_name: str) -> None: - conversation_id, user_id = _pending_scope(ctx) - mark_pending_oauth_sso_consumed(ctx.state, connection_name, conversation_id, user_id) - def _pending_scope(ctx: ActivityContext[Any]) -> Tuple[Optional[str], Optional[str]]: """Conversation and user identifiers used to scope process-local pending hints.""" diff --git a/packages/apps/src/microsoft_teams/apps/oauth_pending_local.py b/packages/apps/src/microsoft_teams/apps/oauth_pending_local.py index c521b059..32c28518 100644 --- a/packages/apps/src/microsoft_teams/apps/oauth_pending_local.py +++ b/packages/apps/src/microsoft_teams/apps/oauth_pending_local.py @@ -103,18 +103,6 @@ def replace(conversation_id: str, user_id: str, restored: List[_Entry]) -> None: _entries[(conversation_id, user_id, connection_key)] = (name.strip(), created_at, sso_offered) -def mark_sso_consumed(conversation_id: str, user_id: str, connection_name: str) -> None: - """Retire a hint's silent-SSO marker, keeping it available for routing.""" - key = _key(conversation_id, user_id, connection_name) - if key is None: - return - - with _lock: - existing = _entries.get(key) - if existing is not None: - _entries[key] = (existing[0], existing[1], False) - - def _key(conversation_id: str, user_id: str, connection_name: str) -> Optional[_Key]: if not conversation_id or not user_id: return None diff --git a/packages/apps/src/microsoft_teams/apps/oauth_state.py b/packages/apps/src/microsoft_teams/apps/oauth_state.py index 00a0d2ea..ddba809b 100644 --- a/packages/apps/src/microsoft_teams/apps/oauth_state.py +++ b/packages/apps/src/microsoft_teams/apps/oauth_state.py @@ -128,31 +128,6 @@ def clear_pending_oauth_sign_in( _remove_connection(state, connection_name) -def mark_pending_oauth_sso_consumed( - state: Optional[TurnStateContainer], - connection_name: str, - conversation_id: Optional[str] = None, - user_id: Optional[str] = None, -) -> None: - """Retire a hint's silent-SSO marker while keeping it for callback routing. - - Teams renders the sign-in button on the same OAuth card after a silent-SSO - failure, so the sign-in is still pending even though its SSO attempt is - spent. Dropping ``sso_offered`` stops the hint from re-attributing later - ``signin/failure`` callbacks while a follow-up ``signin/verifyState`` can - still be routed to the right connection. ``created_at`` is preserved so the - hint expires on its original schedule. - """ - if state is None or state.user is None: - oauth_pending_local.mark_sso_consumed(conversation_id or "", user_id or "", connection_name) - return - - target = connection_lookup_key(connection_name) - for key, name, is_sso_marker in _iter_markers(state.user): - if is_sso_marker and connection_lookup_key(name) == target: - state.user.pop(key, None) - - def replace_pending_oauth_sign_ins( state: Optional[TurnStateContainer], pending: List[PendingOAuthSignIn], diff --git a/packages/apps/tests/test_app_oauth.py b/packages/apps/tests/test_app_oauth.py index ac3d7cc7..a2527e4d 100644 --- a/packages/apps/tests/test_app_oauth.py +++ b/packages/apps/tests/test_app_oauth.py @@ -1103,13 +1103,14 @@ async def on_github(event): assert pending_marker_keys(state) == {"__oauth:pending:graph", "__oauth:pending:sso:graph"} @pytest.mark.asyncio - async def test_sso_failure_keeps_hint_so_button_click_still_routes_to_that_flow( + async def test_sso_failure_clears_hint_and_verify_state_still_resolves_by_probing( self, oauth_handlers, mock_context, failure_activity, verify_state_activity, mock_token_response ): - """After silent SSO fails, Teams shows the sign-in button on the same card. + """The retired hint costs extra probes, not correctness. - The follow-up verify-state carries no connection name, so the retired hint is what - keeps it on GitHub instead of probing (and possibly mis-attributing to) Graph. + Clearing on failure matches C#/TS. The follow-up verify-state carries no + connection name, so it probes each flow; the code is single-use and + connection-scoped, so only the issuing connection accepts it. """ oauth_handlers.oauth_registry.add(OAuthFlow("Graph")) github = oauth_handlers.oauth_registry.add(OAuthFlow("GitHub")) @@ -1125,19 +1126,28 @@ async def on_github(event): mock_context.activity = failure_activity await oauth_handlers.sign_in_failure(mock_context) + # Both keys retired, so nothing routes the follow-up. + assert state.user is not None + assert pending_marker_keys(state) == set() + + # Only GitHub issued this code; Graph rejects it the way the Token Service would. + def get_token(params): + if params.connection_name == "GitHub": + return mock_token_response + raise oauth_http_error(400, "wrong connection") + mock_context.activity = verify_state_activity - mock_context.api.users.get_token.return_value = mock_token_response + mock_context.api.users.get_token.side_effect = get_token + result = await oauth_handlers.sign_in_verify_state(mock_context) assert result is None attempted = [call.args[0].connection_name for call in mock_context.api.users.get_token.await_args_list] - assert attempted == ["GitHub"] + assert attempted == ["Graph", "GitHub"] assert calls == ["github:GitHub"] - assert state.user is not None - assert pending_marker_keys(state) == set() @pytest.mark.asyncio - async def test_retired_sso_hint_does_not_attribute_a_second_failure( + async def test_cleared_hint_does_not_attribute_a_second_failure( self, oauth_handlers, mock_context, failure_activity ): github = oauth_handlers.oauth_registry.add(OAuthFlow("GitHub")) @@ -1153,13 +1163,12 @@ async def on_github(event): await oauth_handlers.sign_in_failure(mock_context) assert calls == ["github:GitHub"] - # Second failure: the hint's SSO marker is spent, so this falls back to the + # Second failure: the hint is gone, so this falls back to the # notify-all-registered-flows path rather than re-attributing to GitHub. await oauth_handlers.sign_in_failure(mock_context) assert calls == ["github:GitHub", "github:GitHub"] assert mock_context.state.user is not None - # The SSO marker is retired; the sign-in itself is still pending. - assert pending_marker_keys(mock_context.state) == {"__oauth:pending:GitHub"} + assert pending_marker_keys(mock_context.state) == set() @pytest.mark.asyncio async def test_legacy_default_connection_hint_is_cleared_without_warning( @@ -1216,12 +1225,11 @@ async def on_github(event): "github:GitHub", ] assert state.user is not None - # The failed connection keeps its hint (the card's sign-in button is still live) but - # loses its SSO marker so it cannot re-attribute a second failure. + # Only the flow that was notified loses its hint; the unrelated pending + # sign-in on test-connection is left intact. assert pending_marker_keys(state) == { "__oauth:pending:test-connection", "__oauth:pending:sso:test-connection", - "__oauth:pending:GitHub", } @pytest.mark.asyncio @@ -1235,7 +1243,9 @@ async def test_sign_in_failure_preserves_replacement_hint_created_by_global_hand async def replace_hint(event): assert event.connection_name == "GitHub" assert state.user is not None - assert pending_marker_keys(state) == {"__oauth:pending:GitHub"} + # Already retired by the time handlers run, so what this records is a + # genuinely new sign-in rather than a survivor of the failed one. + assert pending_marker_keys(state) == set() state.user["__oauth:pending:GitHub"] = datetime.now(timezone.utc).isoformat() state.user["__oauth:pending:sso:GitHub"] = state.user["__oauth:pending:GitHub"] @@ -1251,6 +1261,91 @@ async def replace_hint(event): f"__oauth:pending:sso:{github.connection_name}", } + @pytest.mark.asyncio + async def test_sign_in_failure_clears_both_keys_for_the_resolved_flow( + self, oauth_handlers, mock_context, failure_activity + ): + """Parity with C#/TS: the base hint retires alongside its SSO marker.""" + oauth_handlers.oauth_registry.add(OAuthFlow("GitHub")) + mock_context.activity = failure_activity + state = create_pending_state(("GitHub", time.time(), True)) + mock_context.state = state + assert pending_marker_keys(state) == {"__oauth:pending:GitHub", "__oauth:pending:sso:GitHub"} + + await oauth_handlers.sign_in_failure(mock_context) + + assert pending_marker_keys(state) == set() + + @pytest.mark.asyncio + async def test_sign_in_failure_without_a_hint_clears_every_notified_flow( + self, oauth_handlers, mock_context, failure_activity + ): + """With nothing to attribute the callback, every notified flow retires its hint.""" + oauth_handlers.oauth_registry.add(OAuthFlow("Graph")) + oauth_handlers.oauth_registry.add(OAuthFlow("GitHub")) + mock_context.activity = failure_activity + # Non-SSO hints, so none of them can resolve a target for this callback. + state = create_pending_state( + ("Graph", time.time(), False), + ("GitHub", time.time(), False), + ) + mock_context.state = state + + await oauth_handlers.sign_in_failure(mock_context) + + assert pending_marker_keys(state) == set() + + @pytest.mark.asyncio + async def test_unexpected_token_exchange_error_clears_pending( + self, oauth_handlers, mock_context, token_exchange_activity + ): + """An operational fault ends the sign-in, so its hint must not outlive it.""" + mock_context.activity = token_exchange_activity + state = create_pending_state(("test-connection", time.time(), True)) + mock_context.state = state + mock_context.api.users.exchange_token.side_effect = oauth_http_error(500, "boom") + + result = await oauth_handlers.sign_in_token_exchange(mock_context) + + assert result is not None and result.status == 500 + assert pending_marker_keys(state) == set() + + @pytest.mark.asyncio + async def test_expected_token_exchange_failure_keeps_pending( + self, oauth_handlers, mock_context, token_exchange_activity + ): + """Regression guard: the 412 fallback needs its hint to route the button click.""" + mock_context.activity = token_exchange_activity + state = create_pending_state(("test-connection", time.time(), True)) + mock_context.state = state + mock_context.api.users.exchange_token.side_effect = oauth_http_error(404, "no token") + + result = await oauth_handlers.sign_in_token_exchange(mock_context) + + assert result is not None and result.status == 412 + assert pending_marker_keys(state) == { + "__oauth:pending:test-connection", + "__oauth:pending:sso:test-connection", + } + + @pytest.mark.asyncio + async def test_unexpected_verify_state_error_clears_pending( + self, oauth_handlers, mock_context, verify_state_activity + ): + """An operational fault ends the sign-in, unlike an expected candidate miss.""" + # Registered so the hint survives candidate resolution and reaches the probe; + # an unregistered connection's hint is discarded before this branch is hit. + oauth_handlers.oauth_registry.add(OAuthFlow("test-connection")) + mock_context.activity = verify_state_activity + state = create_pending_state(("test-connection", time.time(), True)) + mock_context.state = state + mock_context.api.users.get_token.side_effect = oauth_http_error(500, "boom") + + result = await oauth_handlers.sign_in_verify_state(mock_context) + + assert result is not None and result.status == 500 + assert pending_marker_keys(state) == set() + @pytest.mark.asyncio async def test_sign_in_failure_ignores_more_recent_non_sso_hint( self, oauth_handlers, mock_context, failure_activity diff --git a/packages/apps/tests/test_oauth_pending_local.py b/packages/apps/tests/test_oauth_pending_local.py index 68f681f8..04538b67 100644 --- a/packages/apps/tests/test_oauth_pending_local.py +++ b/packages/apps/tests/test_oauth_pending_local.py @@ -214,15 +214,13 @@ def test_clearing_without_a_name_clears_the_whole_scope(self) -> None: assert get_pending_oauth_sign_ins(None, "c", "u") == [] - def test_consuming_sso_keeps_the_hint_for_routing(self) -> None: - """The sign-in is still pending; only its silent-SSO attempt is spent.""" + def test_clearing_retires_an_sso_hint_entirely(self) -> None: + """The whole hint goes, SSO marker included, so nothing survives to re-route.""" record_pending_oauth_sign_in(None, "graph", sso_offered=True, conversation_id="c", user_id="u") - local.mark_sso_consumed("c", "u", "GRAPH") + clear_pending_oauth_sign_in(None, "GRAPH", "c", "u") - hints = get_pending_oauth_sign_ins(None, "c", "u") - assert len(hints) == 1 - assert hints[0].sso_offered is False + assert get_pending_oauth_sign_ins(None, "c", "u") == [] def test_replace_restores_the_original_timestamps(self) -> None: """Rollback puts back what was there, not a fresh set of hints.""" diff --git a/packages/apps/tests/test_oauth_state.py b/packages/apps/tests/test_oauth_state.py index 267d1a61..a6053c77 100644 --- a/packages/apps/tests/test_oauth_state.py +++ b/packages/apps/tests/test_oauth_state.py @@ -18,7 +18,6 @@ completed_token_exchange_state_key, get_pending_oauth_sign_ins, has_completed_token_exchange, - mark_pending_oauth_sso_consumed, record_completed_token_exchange, record_pending_oauth_sign_in, replace_pending_oauth_sign_ins, @@ -158,18 +157,16 @@ def test_equal_timestamps_break_ties_deterministically(self) -> None: class TestSsoMarkerHandling: - def test_marking_sso_consumed_retires_only_the_sso_marker(self) -> None: + def test_clearing_retires_the_hint_and_its_sso_marker_together(self) -> None: + """Both keys retire as a unit, so no orphan can re-attribute a later callback.""" state = make_state() record_pending_oauth_sign_in(state, "Graph", sso_offered=True) - assert state.user is not None - original = state.user["__oauth:pending:Graph"] + assert stored_keys(state) == {"__oauth:pending:Graph", "__oauth:pending:sso:Graph"} - mark_pending_oauth_sso_consumed(state, "graph") + clear_pending_oauth_sign_in(state, "graph") - # The sign-in is still pending on its original schedule; only SSO is spent. - assert stored_keys(state) == {"__oauth:pending:Graph"} - assert state.user["__oauth:pending:Graph"] == original - assert [(h.connection_name, h.sso_offered) for h in get_pending_oauth_sign_ins(state)] == [("Graph", False)] + assert stored_keys(state) == set() + assert get_pending_oauth_sign_ins(state) == [] def test_connection_named_sso_is_not_mistaken_for_an_sso_marker(self) -> None: # ``sso:`` is a legal start to a connection name. Without its own base marker, @@ -292,7 +289,6 @@ def test_helpers_are_no_ops_without_state(self) -> None: assert get_pending_oauth_sign_ins(None) == [] record_pending_oauth_sign_in(None, "Graph", sso_offered=True) clear_pending_oauth_sign_in(None, "Graph") - mark_pending_oauth_sso_consumed(None, "Graph") replace_pending_oauth_sign_ins(None, []) def test_helpers_are_no_ops_without_a_user_scope(self) -> None: @@ -306,7 +302,6 @@ def test_helpers_are_no_ops_without_a_user_scope(self) -> None: assert get_pending_oauth_sign_ins(state) == [] record_pending_oauth_sign_in(state, "Graph", sso_offered=True) clear_pending_oauth_sign_in(state, "Graph") - mark_pending_oauth_sso_consumed(state, "Graph") replace_pending_oauth_sign_ins(state, [])