BotBuilder-schema contains the serialized data sent across the wire between user and bot when using Bot Framework
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
OAuth token management, authentication flows, and signin processes including token requests, responses, and exchange mechanisms for secure bot authentication in Bot Framework applications.
Response model containing OAuth tokens returned from authentication providers.
class TokenResponse(Model):
def __init__(self, *, connection_name: str = None, token: str = None,
expiration: str = None, channel_id: str = None, **kwargs): ...from botbuilder.schema import TokenResponse
# Token response from OAuth provider
token_response = TokenResponse(
connection_name="MyOAuthConnection",
token="eyJhbGciOiJSUzI1NiIsImtpZCI6...",
expiration="2023-12-31T23:59:59Z",
channel_id="webchat"
)Request model for initiating OAuth token requests.
class TokenRequest(Model):
def __init__(self, *, provider: str = None, settings: dict = None, **kwargs): ...from botbuilder.schema import TokenRequest
token_request = TokenRequest(
provider="AzureAD",
settings={
"scopes": ["User.Read", "Mail.Read"],
"tenant": "common"
}
)Model for token exchange operations in Bot Framework applications.
class TokenExchangeInvokeRequest(Model):
def __init__(self, *, id: str = None, connection_name: str = None,
token: str = None, properties: dict = None, **kwargs): ...from botbuilder.schema import TokenExchangeInvokeRequest
exchange_request = TokenExchangeInvokeRequest(
id="unique-exchange-id",
connection_name="MyOAuthConnection",
token="original-token-to-exchange",
properties={"scope": "additional-permissions"}
)Response model for token exchange operations.
class TokenExchangeInvokeResponse(Model):
def __init__(self, *, id: str = None, connection_name: str = None,
failure_detail: str = None, **kwargs): ...from botbuilder.schema import TokenExchangeInvokeResponse
# Successful exchange response
success_response = TokenExchangeInvokeResponse(
id="unique-exchange-id",
connection_name="MyOAuthConnection"
)
# Failed exchange response
failure_response = TokenExchangeInvokeResponse(
id="unique-exchange-id",
connection_name="MyOAuthConnection",
failure_detail="Token expired or invalid"
)State information for token exchange flows.
class TokenExchangeState(Model):
def __init__(self, *, connection_name: str = None, conversation = None,
relates_to = None, bot_url: str = None, **kwargs): ...from botbuilder.schema import TokenExchangeState, ConversationReference
exchange_state = TokenExchangeState(
connection_name="MyOAuthConnection",
conversation=conversation_reference,
bot_url="https://mybot.azurewebsites.net"
)Card specifically designed for OAuth authentication flows.
class OAuthCard(Model):
def __init__(self, *, text: str = None, connection_name: str = None,
buttons: List[CardAction] = None, **kwargs): ...from botbuilder.schema import OAuthCard, CardAction
oauth_card = OAuthCard(
text="Please sign in to access your account",
connection_name="MyOAuthConnection",
buttons=[
CardAction(
type="signin",
title="Sign In",
value="https://oauth.provider.com/signin"
)
]
)Generic signin card for authentication prompts.
class SigninCard(Model):
def __init__(self, *, text: str = None, buttons: List[CardAction] = None, **kwargs): ...from botbuilder.schema import SigninCard, CardAction
signin_card = SigninCard(
text="Authentication required to continue",
buttons=[
CardAction(type="signin", title="Sign In", value="signin_url"),
CardAction(type="imBack", title="Cancel", value="cancel")
]
)Constants for signin operations and event names.
class SignInConstants:
verify_state_operation_name: str = "signin/verifyState"
token_exchange_operation_name: str = "signin/tokenExchange"
token_response_event_name: str = "tokens/response"from botbuilder.schema import SignInConstants
# Use constants for consistent event handling
if activity.name == SignInConstants.token_response_event_name:
# Handle token response
pass
elif activity.name == SignInConstants.verify_state_operation_name:
# Handle state verification
passConstants for identifying Bot Framework channels and authentication contexts.
class CallerIdConstants:
public_azure_channel: str = "urn:botframework:azure"
us_gov_channel: str = "urn:botframework:azureusgov"
bot_to_bot_prefix: str = "urn:botframework:aadappid:"from botbuilder.schema import CallerIdConstants
def is_azure_channel(caller_id: str) -> bool:
return caller_id == CallerIdConstants.public_azure_channel
def is_bot_to_bot(caller_id: str) -> bool:
return caller_id and caller_id.startswith(CallerIdConstants.bot_to_bot_prefix)from botbuilder.schema import (
Activity, ActivityTypes, Attachment, OAuthCard,
CardAction, TokenResponse
)
async def start_oauth_flow(connection_name: str) -> Activity:
"""Initiate OAuth authentication flow"""
oauth_card = OAuthCard(
text="Please sign in to continue using the bot",
connection_name=connection_name,
buttons=[
CardAction(type="signin", title="Sign In")
]
)
attachment = Attachment(
content_type="application/vnd.microsoft.card.oauth",
content=oauth_card
)
return Activity(
type=ActivityTypes.message,
text="Authentication required",
attachments=[attachment]
)
async def handle_token_response(activity: Activity) -> str:
"""Handle token response from OAuth provider"""
if activity.name == "tokens/response":
token_response = TokenResponse(**activity.value)
return token_response.token
return Nonefrom botbuilder.schema import (
TokenExchangeInvokeRequest, TokenExchangeInvokeResponse,
InvokeResponse
)
async def handle_token_exchange(request: TokenExchangeInvokeRequest) -> InvokeResponse:
"""Handle token exchange invoke"""
try:
# Attempt to exchange token
exchanged_token = await exchange_token(
request.token,
request.connection_name
)
response = TokenExchangeInvokeResponse(
id=request.id,
connection_name=request.connection_name
)
return InvokeResponse(
status=200,
body=response
)
except Exception as e:
error_response = TokenExchangeInvokeResponse(
id=request.id,
connection_name=request.connection_name,
failure_detail=str(e)
)
return InvokeResponse(
status=412, # Precondition Failed
body=error_response
)from botbuilder.schema import SignInConstants
async def verify_signin_state(activity: Activity) -> bool:
"""Verify signin state for security"""
if activity.name == SignInConstants.verify_state_operation_name:
state = activity.value.get('state')
# Verify state matches expected value
return await validate_state(state)
return Falsefrom botbuilder.schema import Activity, ActivityTypes
class AuthenticationMiddleware:
def __init__(self, connection_name: str):
self.connection_name = connection_name
async def on_message_activity(self, turn_context, next_handler):
"""Check authentication before processing messages"""
token = await self.get_user_token(turn_context)
if not token:
# Send OAuth card
oauth_activity = await self.create_oauth_prompt()
await turn_context.send_activity(oauth_activity)
return
# User is authenticated, continue processing
await next_handler()
async def get_user_token(self, turn_context) -> str:
"""Get user token from token store"""
# Implementation depends on token store
pass
async def create_oauth_prompt(self) -> Activity:
"""Create OAuth authentication prompt"""
return await start_oauth_flow(self.connection_name)import aiohttp
from botbuilder.schema import TokenResponse
async def make_authenticated_api_call(token: str, api_url: str):
"""Make API call with OAuth token"""
headers = {
'Authorization': f'Bearer {token}',
'Content-Type': 'application/json'
}
async with aiohttp.ClientSession() as session:
async with session.get(api_url, headers=headers) as response:
if response.status == 401:
raise AuthenticationError("Token expired or invalid")
return await response.json()
class AuthenticationError(Exception):
"""Custom exception for authentication errors"""
passfrom datetime import datetime
from botbuilder.schema import TokenResponse
def is_token_valid(token_response: TokenResponse) -> bool:
"""Validate token expiration"""
if not token_response.expiration:
return True # No expiration set
expiration = datetime.fromisoformat(token_response.expiration.replace('Z', '+00:00'))
return datetime.now(expiration.tzinfo) < expiration
def validate_token_scopes(token: str, required_scopes: list) -> bool:
"""Validate token has required scopes"""
# Implementation depends on token format (JWT, etc.)
passimport hashlib
import secrets
def generate_state() -> str:
"""Generate secure state parameter"""
return secrets.token_urlsafe(32)
def validate_state(received_state: str, expected_state: str) -> bool:
"""Validate state parameter to prevent CSRF"""
return secrets.compare_digest(received_state, expected_state)from botbuilder.schema import Activity, ActivityTypes
def create_auth_error_response(error_message: str) -> Activity:
"""Create user-friendly authentication error response"""
return Activity(
type=ActivityTypes.message,
text=f"Authentication failed: {error_message}. Please try signing in again."
)
def handle_oauth_errors(error_type: str) -> Activity:
"""Handle different OAuth error scenarios"""
error_messages = {
'access_denied': "Access was denied. Please try again or contact support.",
'invalid_request': "Invalid authentication request. Please try again.",
'server_error': "Authentication server error. Please try again later.",
'temporarily_unavailable': "Authentication service temporarily unavailable."
}
message = error_messages.get(error_type, "Unknown authentication error occurred.")
return create_auth_error_response(message)Install with Tessl CLI
npx tessl i tessl/pypi-botbuilder-schema