Tornado is a Python web framework and asynchronous networking library designed for applications requiring long-lived connections to many users.
Authentication and authorization support for third-party services including OAuth, OAuth2, OpenID, and integration with major providers like Google, Facebook, and Twitter.
Abstract base classes for implementing OAuth 1.0 and 1.0a authentication flows.
class OAuthMixin:
"""Abstract implementation of OAuth 1.0 and 1.0a."""
def get_auth_http_client(self):
"""Get HTTP client for auth requests."""
def oauth_request_token_url(self) -> str:
"""Get OAuth request token URL."""
def oauth_access_token_url(self) -> str:
"""Get OAuth access token URL."""
def oauth_authorize_url(self) -> str:
"""Get OAuth authorization URL."""
def oauth_consumer_key(self) -> str:
"""Get OAuth consumer key."""
def oauth_consumer_secret(self) -> str:
"""Get OAuth consumer secret."""
def oauth_get_user(self, access_token: str, callback):
"""Get user info with access token."""
def authenticate_redirect(self, callback_uri: str = None):
"""Start OAuth authentication flow."""
def get_authenticated_user(self, callback):
"""Complete OAuth authentication."""Abstract base class for OAuth 2.0 authentication flows with authorization code grant.
class OAuth2Mixin:
"""Abstract implementation of OAuth 2.0."""
def get_auth_http_client(self):
"""Get HTTP client for auth requests."""
def oauth2_request_token_url(self) -> str:
"""Get OAuth2 token request URL."""
def oauth2_authorize_url(self) -> str:
"""Get OAuth2 authorization URL."""
def oauth2_client_id(self) -> str:
"""Get OAuth2 client ID."""
def oauth2_client_secret(self) -> str:
"""Get OAuth2 client secret."""
def oauth2_scope(self) -> str:
"""Get OAuth2 scope."""
def oauth2_access_token_url(self) -> str:
"""Get OAuth2 access token URL."""
def authorize_redirect(self, redirect_uri: str = None, client_id: str = None, client_secret: str = None, extra_params=None, scope: str = None, response_type: str = "code"):
"""
Start OAuth2 authorization flow.
Args:
redirect_uri: Redirect URI after authorization
client_id: OAuth2 client ID
client_secret: OAuth2 client secret
extra_params: Additional parameters
scope: OAuth2 scope
response_type: OAuth2 response type
"""
def get_authenticated_user(self, redirect_uri: str, client_id: str, client_secret: str, code: str, callback, extra_fields=None):
"""
Complete OAuth2 authentication.
Args:
redirect_uri: Redirect URI
client_id: OAuth2 client ID
client_secret: OAuth2 client secret
code: Authorization code
callback: Completion callback
extra_fields: Additional fields to request
"""OpenID authentication with Attribute Exchange support.
class OpenIdMixin:
"""Abstract implementation of OpenID and Attribute Exchange."""
def authenticate_redirect(self, callback_uri: str = None, ax_attrs: List[str] = None):
"""
Start OpenID authentication.
Args:
callback_uri: Callback URI
ax_attrs: Attribute Exchange attributes to request
"""
def get_authenticated_user(self, callback, http_client=None):
"""
Complete OpenID authentication.
Args:
callback: Completion callback
http_client: HTTP client for requests
"""
def get_auth_http_client(self):
"""Get HTTP client for auth requests."""Google-specific OAuth2 implementation with Google API integration.
class GoogleOAuth2Mixin(OAuth2Mixin):
"""Google OAuth2 authentication."""
_OAUTH_AUTHORIZE_URL = "https://accounts.google.com/o/oauth2/auth"
_OAUTH_ACCESS_TOKEN_URL = "https://accounts.google.com/o/oauth2/token"
_OAUTH_NO_CALLBACKS = False
def get_google_oauth_settings(self) -> Dict[str, str]:
"""
Get Google OAuth settings.
Returns:
Dict with google_oauth client_id and client_secret
"""
def oauth2_client_id(self) -> str:
"""Get Google OAuth2 client ID."""
def oauth2_client_secret(self) -> str:
"""Get Google OAuth2 client secret."""
def oauth2_scope(self) -> str:
"""Get Google OAuth2 scope."""
def oauth2_authorize_url(self) -> str:
"""Get Google OAuth2 authorization URL."""
def oauth2_access_token_url(self) -> str:
"""Get Google OAuth2 access token URL."""Facebook Graph API authentication and user data access.
class FacebookGraphMixin(OAuth2Mixin):
"""Facebook Graph API authentication."""
_OAUTH_AUTHORIZE_URL = "https://www.facebook.com/dialog/oauth"
_OAUTH_ACCESS_TOKEN_URL = "https://graph.facebook.com/oauth/access_token"
_OAUTH_NO_CALLBACKS = False
def facebook_request(self, path: str, callback, access_token: str = None, post_args=None, **args):
"""
Make Facebook Graph API request.
Args:
path: API path
callback: Response callback
access_token: Access token
post_args: POST arguments
**args: Additional arguments
"""
def oauth2_client_id(self) -> str:
"""Get Facebook app ID."""
def oauth2_client_secret(self) -> str:
"""Get Facebook app secret."""
def oauth2_scope(self) -> str:
"""Get Facebook OAuth scope."""
def oauth2_authorize_url(self) -> str:
"""Get Facebook OAuth authorization URL."""
def oauth2_access_token_url(self) -> str:
"""Get Facebook OAuth access token URL."""Twitter-specific OAuth 1.0a implementation.
class TwitterMixin(OAuthMixin):
"""Twitter OAuth authentication."""
_OAUTH_REQUEST_TOKEN_URL = "https://api.twitter.com/oauth/request_token"
_OAUTH_ACCESS_TOKEN_URL = "https://api.twitter.com/oauth/access_token"
_OAUTH_AUTHORIZE_URL = "https://api.twitter.com/oauth/authorize"
_OAUTH_AUTHENTICATE_URL = "https://api.twitter.com/oauth/authenticate"
_OAUTH_NO_CALLBACKS = False
def twitter_request(self, path: str, callback, access_token: dict = None, post_args=None, **args):
"""
Make Twitter API request.
Args:
path: API path
callback: Response callback
access_token: Access token dict
post_args: POST arguments
**args: Additional arguments
"""
def oauth_consumer_key(self) -> str:
"""Get Twitter consumer key."""
def oauth_consumer_secret(self) -> str:
"""Get Twitter consumer secret."""
def oauth_request_token_url(self) -> str:
"""Get Twitter request token URL."""
def oauth_access_token_url(self) -> str:
"""Get Twitter access token URL."""
def oauth_authorize_url(self) -> str:
"""Get Twitter authorization URL."""import tornado.web
import tornado.auth
class GoogleAuth(tornado.web.RequestHandler, tornado.auth.GoogleOAuth2Mixin):
def get_google_oauth_settings(self):
return {
'google_oauth': {
'client_id': 'your-client-id',
'client_secret': 'your-client-secret'
}
}
async def get(self):
if self.get_argument('code', False):
# Handle OAuth callback
user = await self.get_authenticated_user(
redirect_uri='http://localhost:8888/auth/google',
code=self.get_argument('code')
)
if user:
self.write(f"Hello, {user['name']}!")
else:
self.write("Authentication failed")
else:
# Start OAuth flow
self.authorize_redirect(
redirect_uri='http://localhost:8888/auth/google',
client_id=self.get_google_oauth_settings()['google_oauth']['client_id'],
scope=['profile', 'email'],
response_type='code'
)
app = tornado.web.Application([
(r"/auth/google", GoogleAuth),
])import tornado.web
import tornado.auth
class FacebookAuth(tornado.web.RequestHandler, tornado.auth.FacebookGraphMixin):
@tornado.web.asynchronous
def get(self):
if self.get_argument("code", False):
self.get_authenticated_user(
redirect_uri='http://localhost:8888/auth/facebook',
client_id='your-app-id',
client_secret='your-app-secret',
code=self.get_argument("code"),
callback=self.on_auth
)
else:
self.authorize_redirect(
redirect_uri='http://localhost:8888/auth/facebook',
client_id='your-app-id',
extra_params={"scope": "read_stream,offline_access"}
)
def on_auth(self, user):
if user:
self.write(f"Hello, {user['name']}!")
else:
self.write("Authentication failed")
self.finish()
app = tornado.web.Application([
(r"/auth/facebook", FacebookAuth),
])import tornado.web
import tornado.auth
class TwitterAuth(tornado.web.RequestHandler, tornado.auth.TwitterMixin):
@tornado.web.asynchronous
def get(self):
if self.get_argument("oauth_token", None):
self.get_authenticated_user(self.on_auth)
else:
self.authenticate_redirect(callback_uri='http://localhost:8888/auth/twitter')
def on_auth(self, user):
if user:
self.write(f"Hello, {user['username']}!")
else:
self.write("Authentication failed")
self.finish()
app = tornado.web.Application([
(r"/auth/twitter", TwitterAuth),
])# OAuth token types
OAuthToken = Dict[str, str]
OAuth2Token = Dict[str, Any]
# User info type
UserInfo = Dict[str, Any]
# OAuth callback type
OAuthCallback = Callable[[UserInfo], None]
# HTTP client type for auth
AuthHTTPClient = tornado.httpclient.AsyncHTTPClientclass AuthError(Exception):
"""Base exception for authentication errors."""
def __init__(self, message: str):
"""
Initialize auth error.
Args:
message: Error message
"""Install with Tessl CLI
npx tessl i tessl/pypi-tornado