# Copyright (c) 2023-present Plane Software, Inc. and contributors # SPDX-License-Identifier: AGPL-3.0-only # See the LICENSE file for details. # Python imports import os from datetime import datetime from urllib.parse import urlencode import pytz # Module imports from plane.authentication.adapter.oauth import OauthAdapter from plane.license.utils.instance_value import get_configuration_value from plane.authentication.adapter.error import ( AUTHENTICATION_ERROR_CODES, AuthenticationException, ) class GoogleOAuthProvider(OauthAdapter): token_url = "https://oauth2.googleapis.com/token" userinfo_url = "https://www.googleapis.com/oauth2/v2/userinfo" scope = "https://www.googleapis.com/auth/userinfo.email https://www.googleapis.com/auth/userinfo.profile" provider = "google" def __init__(self, request, code=None, state=None, callback=None): (GOOGLE_CLIENT_ID, GOOGLE_CLIENT_SECRET) = get_configuration_value( [ { "key": "GOOGLE_CLIENT_ID", "default": os.environ.get("GOOGLE_CLIENT_ID"), }, { "key": "GOOGLE_CLIENT_SECRET", "default": os.environ.get("GOOGLE_CLIENT_SECRET"), }, ] ) if not (GOOGLE_CLIENT_ID and GOOGLE_CLIENT_SECRET): raise AuthenticationException( error_code=AUTHENTICATION_ERROR_CODES["GOOGLE_NOT_CONFIGURED"], error_message="GOOGLE_NOT_CONFIGURED", ) client_id = GOOGLE_CLIENT_ID client_secret = GOOGLE_CLIENT_SECRET redirect_uri = f"""{"https" if request.is_secure() else "http"}://{request.get_host()}/auth/google/callback/""" url_params = { "client_id": client_id, "scope": self.scope, "redirect_uri": redirect_uri, "response_type": "code", "access_type": "offline", "prompt": "consent", "state": state, } auth_url = f"https://accounts.google.com/o/oauth2/v2/auth?{urlencode(url_params)}" super().__init__( request, self.provider, client_id, self.scope, redirect_uri, auth_url, self.token_url, self.userinfo_url, client_secret, code, callback=callback, ) def set_token_data(self): data = { "code": self.code, "client_id": self.client_id, "client_secret": self.client_secret, "redirect_uri": self.redirect_uri, "grant_type": "authorization_code", } token_response = self.get_user_token(data=data) super().set_token_data( { "access_token": token_response.get("access_token"), "refresh_token": token_response.get("refresh_token", None), "access_token_expired_at": ( datetime.fromtimestamp(token_response.get("expires_in"), tz=pytz.utc) if token_response.get("expires_in") else None ), "refresh_token_expired_at": ( datetime.fromtimestamp(token_response.get("refresh_token_expired_at"), tz=pytz.utc) if token_response.get("refresh_token_expired_at") else None ), "id_token": token_response.get("id_token", ""), } ) def set_user_data(self): user_info_response = self.get_user_response() user_data = { "email": user_info_response.get("email"), "user": { "avatar": user_info_response.get("picture"), "first_name": user_info_response.get("given_name"), "last_name": user_info_response.get("family_name"), "provider_id": user_info_response.get("id"), "is_password_autoset": True, }, } super().set_user_data(user_data)