Spaces:
Build error
Build error
File size: 5,504 Bytes
0827183 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 |
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
from abc import ABC
from botbuilder.core import TurnContext
from botbuilder.core.bot_framework_adapter import TokenExchangeRequest
from botbuilder.core.oauth import ConnectorClientBuilder, ExtendedUserTokenProvider
from botbuilder.schema import TokenResponse
from botframework.connector import ConnectorClient
from botframework.connector.auth import ClaimsIdentity, ConnectorFactory
from botframework.connector.auth.user_token_client import UserTokenClient
from botframework.connector.token_api.models import SignInUrlResponse
from .prompts.oauth_prompt_settings import OAuthPromptSettings
class _UserTokenAccess(ABC):
@staticmethod
async def get_user_token(
turn_context: TurnContext, settings: OAuthPromptSettings, magic_code: str
) -> TokenResponse:
user_token_client: UserTokenClient = turn_context.turn_state.get(
UserTokenClient.__name__, None
)
if user_token_client:
return await user_token_client.get_user_token(
turn_context.activity.from_property.id,
settings.connection_name,
turn_context.activity.channel_id,
magic_code,
)
if isinstance(turn_context.adapter, ExtendedUserTokenProvider):
return await turn_context.adapter.get_user_token(
turn_context,
settings.connection_name,
magic_code,
settings.oath_app_credentials,
)
raise TypeError("OAuthPrompt is not supported by the current adapter")
@staticmethod
async def get_sign_in_resource(
turn_context: TurnContext, settings: OAuthPromptSettings
) -> SignInUrlResponse:
user_token_client: UserTokenClient = turn_context.turn_state.get(
UserTokenClient.__name__, None
)
if user_token_client:
return await user_token_client.get_sign_in_resource(
settings.connection_name, turn_context.activity, None
)
if isinstance(turn_context.adapter, ExtendedUserTokenProvider):
return await turn_context.adapter.get_sign_in_resource_from_user_and_credentials(
turn_context,
settings.oath_app_credentials,
settings.connection_name,
(
turn_context.activity.from_property.id
if turn_context.activity and turn_context.activity.from_property
else None
),
)
raise TypeError("OAuthPrompt is not supported by the current adapter")
@staticmethod
async def sign_out_user(turn_context: TurnContext, settings: OAuthPromptSettings):
user_token_client: UserTokenClient = turn_context.turn_state.get(
UserTokenClient.__name__, None
)
if user_token_client:
return await user_token_client.sign_out_user(
turn_context.activity.from_property.id,
settings.connection_name,
turn_context.activity.channel_id,
)
if isinstance(turn_context.adapter, ExtendedUserTokenProvider):
return await turn_context.adapter.sign_out_user(
turn_context,
settings.connection_name,
(
turn_context.activity.from_property.id
if turn_context.activity and turn_context.activity.from_property
else None
),
settings.oath_app_credentials,
)
raise TypeError("OAuthPrompt is not supported by the current adapter")
@staticmethod
async def exchange_token(
turn_context: TurnContext,
settings: OAuthPromptSettings,
token_exchange_request: TokenExchangeRequest,
) -> TokenResponse:
user_token_client: UserTokenClient = turn_context.turn_state.get(
UserTokenClient.__name__, None
)
user_id = turn_context.activity.from_property.id
if user_token_client:
channel_id = turn_context.activity.channel_id
return await user_token_client.exchange_token(
user_id,
channel_id,
token_exchange_request,
)
if isinstance(turn_context.adapter, ExtendedUserTokenProvider):
return await turn_context.adapter.exchange_token(
turn_context,
settings.connection_name,
user_id,
token_exchange_request,
)
raise TypeError("OAuthPrompt is not supported by the current adapter")
@staticmethod
async def create_connector_client(
turn_context: TurnContext,
service_url: str,
claims_identity: ClaimsIdentity,
audience: str,
) -> ConnectorClient:
connector_factory: ConnectorFactory = turn_context.turn_state.get(
ConnectorFactory.__name__, None
)
if connector_factory:
return await connector_factory.create(service_url, audience)
if isinstance(turn_context.adapter, ConnectorClientBuilder):
return await turn_context.adapter.create_connector_client(
service_url,
claims_identity,
audience,
)
raise TypeError("OAuthPrompt is not supported by the current adapter")
|