148 lines
4.3 KiB
Python
148 lines
4.3 KiB
Python
"""
|
|
Google OAuth2 flow implementation.
|
|
|
|
Isolated OAuth2 client that can run without FastAPI.
|
|
"""
|
|
|
|
from typing import Optional
|
|
from google.auth.transport.requests import Request
|
|
from google.oauth2.credentials import Credentials
|
|
from google_auth_oauthlib.flow import Flow
|
|
from .config import settings
|
|
|
|
|
|
class GoogleOAuth:
|
|
"""
|
|
Google OAuth2 client.
|
|
|
|
Handles authorization flow, token exchange, and token refresh.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
client_id: Optional[str] = None,
|
|
client_secret: Optional[str] = None,
|
|
redirect_uri: Optional[str] = None,
|
|
scopes: Optional[list[str]] = None,
|
|
):
|
|
"""
|
|
Initialize OAuth client.
|
|
|
|
Falls back to settings if parameters not provided.
|
|
"""
|
|
self.client_id = client_id or settings.google_client_id
|
|
self.client_secret = client_secret or settings.google_client_secret
|
|
self.redirect_uri = redirect_uri or settings.google_redirect_uri
|
|
self.scopes = scopes or settings.google_scopes.split()
|
|
|
|
def _create_flow(self) -> Flow:
|
|
"""Create OAuth flow object."""
|
|
client_config = {
|
|
"web": {
|
|
"client_id": self.client_id,
|
|
"client_secret": self.client_secret,
|
|
"auth_uri": "https://accounts.google.com/o/oauth2/auth",
|
|
"token_uri": "https://oauth2.googleapis.com/token",
|
|
}
|
|
}
|
|
flow = Flow.from_client_config(
|
|
client_config,
|
|
scopes=self.scopes,
|
|
redirect_uri=self.redirect_uri,
|
|
)
|
|
return flow
|
|
|
|
def get_authorization_url(self, state: Optional[str] = None) -> str:
|
|
"""
|
|
Generate OAuth2 authorization URL.
|
|
|
|
Args:
|
|
state: Optional state parameter for CSRF protection
|
|
|
|
Returns:
|
|
URL to redirect user for Google authorization
|
|
"""
|
|
flow = self._create_flow()
|
|
auth_url, _ = flow.authorization_url(
|
|
access_type="offline", # Request refresh token
|
|
include_granted_scopes="true",
|
|
state=state,
|
|
)
|
|
return auth_url
|
|
|
|
def exchange_code_for_tokens(self, code: str) -> dict:
|
|
"""
|
|
Exchange authorization code for tokens.
|
|
|
|
Args:
|
|
code: Authorization code from callback
|
|
|
|
Returns:
|
|
Token dict containing:
|
|
- access_token
|
|
- refresh_token
|
|
- expires_in
|
|
- scope
|
|
- token_type
|
|
"""
|
|
flow = self._create_flow()
|
|
flow.fetch_token(code=code)
|
|
|
|
credentials = flow.credentials
|
|
return {
|
|
"access_token": credentials.token,
|
|
"refresh_token": credentials.refresh_token,
|
|
"expires_in": 3600, # Google tokens typically 1 hour
|
|
"scope": " ".join(credentials.scopes or []),
|
|
"token_type": "Bearer",
|
|
}
|
|
|
|
def refresh_access_token(self, refresh_token: str) -> dict:
|
|
"""
|
|
Refresh an expired access token.
|
|
|
|
Args:
|
|
refresh_token: The refresh token
|
|
|
|
Returns:
|
|
New token dict with fresh access_token
|
|
"""
|
|
credentials = Credentials(
|
|
token=None,
|
|
refresh_token=refresh_token,
|
|
token_uri="https://oauth2.googleapis.com/token",
|
|
client_id=self.client_id,
|
|
client_secret=self.client_secret,
|
|
)
|
|
|
|
request = Request()
|
|
credentials.refresh(request)
|
|
|
|
return {
|
|
"access_token": credentials.token,
|
|
"refresh_token": refresh_token, # Keep original refresh token
|
|
"expires_in": 3600,
|
|
"scope": " ".join(credentials.scopes or []),
|
|
"token_type": "Bearer",
|
|
}
|
|
|
|
def get_credentials(self, access_token: str, refresh_token: Optional[str] = None) -> Credentials:
|
|
"""
|
|
Create Google Credentials object from tokens.
|
|
|
|
Args:
|
|
access_token: OAuth2 access token
|
|
refresh_token: Optional refresh token
|
|
|
|
Returns:
|
|
Google Credentials object for API calls
|
|
"""
|
|
return Credentials(
|
|
token=access_token,
|
|
refresh_token=refresh_token,
|
|
token_uri="https://oauth2.googleapis.com/token",
|
|
client_id=self.client_id,
|
|
client_secret=self.client_secret,
|
|
scopes=self.scopes,
|
|
)
|