Search code examples
pythonpyqt5google-oauth

Alternate OAuth2 sign in for a PyQt5 application


I have a PyQt5 application with a Google sign-in which is implemented using oauth2client. And the sign in page is shown in an embedded browser using QWebEngineView. But with Google blocking sign in workflows using embedded browsers from Jan 4, 2021, there will be a change required in my application to open system browser instead and then receive the authorization response from that. For that, I am using google-auth-oauthlib, which is also used in the Google Python Quickstart Documentation.

I have created a small POC that just implements this workflow, but I am facing a couple of issues:

  1. The sign in page is opened in a new browser window or a tab. But this might not be a great UX as the user has to close the browser after signing in and then get back to the application. Opening a popup browser seems like a better UX here. I checked the source code of run_local_server method that is responsible for opening the browser, and they seem to use the webbrowser module, which unfortunately does not have a way of opening a popup.

  2. If the user closes the browser opened using run_local_server method without signing in, the application that is calling it just freezes and needs to be force quit. I did not notice any console errors as well. Is there even a way to handle this with the library that I am using?

Here is the minimum working example:

import sys

from PyQt5.QtCore import QStandardPaths
from PyQt5.QtWidgets import QWidget, QPushButton, QApplication
from google_auth_oauthlib.flow import InstalledAppFlow

class GoogleSignIn(QWidget):

    def __init__(self):
        super().__init__()

        self.flow = InstalledAppFlow.from_client_secrets_file(
            "credentials.json", # This file should be placed in the correct folder
            scopes=["https://www.googleapis.com/auth/userinfo.profile", "openid",
                    "https://www.googleapis.com/auth/userinfo.email"])

        self.initUI()

    def initUI(self):
        self.sign_in_btn = QPushButton('Sign In', self)
        self.sign_in_btn.move(135, 135)
        self.sign_in_btn.setFixedSize(100, 40)
        self.sign_in_btn.clicked.connect(self.open_google_sign_in)

        self.setFixedSize(350, 350)
        self.setWindowTitle('Google Sign in Test')
        self.show()

    def open_google_sign_in(self):
        self.flow.run_local_server(port=0)

        session = self.flow.authorized_session()

        profile_info = session.get('https://www.googleapis.com/userinfo/v2/me').json()
        print(profile_info)


def main():
    app = QApplication(sys.argv)
    ex = GoogleSignIn()
    sys.exit(app.exec_())


if __name__ == '__main__':
    main()

Solution

  • QWebEngineView can be used as a browser for authentication but you must set a valid user-agent. On the other hand, google-auth-oauthlib requests are blocking so they must be executed in a different thread and notify the result through signals:

    import functools
    import logging
    import os
    import pickle
    import sys
    import threading
    
    from PyQt5 import QtCore, QtGui, QtWidgets, QtWebEngineWidgets
    
    from google_auth_oauthlib.flow import InstalledAppFlow
    from google.auth.transport.requests import Request
    from googleapiclient.discovery import build
    
    
    SCOPES = [
        "https://www.googleapis.com/auth/userinfo.profile",
        "openid",
        "https://www.googleapis.com/auth/userinfo.email",
    ]
    CURRENT_DIR = os.path.dirname(os.path.realpath(__file__))
    
    
    logging.basicConfig(level=logging.DEBUG)
    
    
    class Reply(QtCore.QObject):
        finished = QtCore.pyqtSignal()
    
        def __init__(self, func, args=(), kwargs=None, parent=None):
            super().__init__(parent)
            self._results = None
            self._is_finished = False
            self._error_str = ""
            threading.Thread(
                target=self._execute, args=(func, args, kwargs), daemon=True
            ).start()
    
        @property
        def results(self):
            return self._results
    
        @property
        def error_str(self):
            return self._error_str
    
        def is_finished(self):
            return self._is_finished
    
        def has_error(self):
            return bool(self._error_str)
    
        def _execute(self, func, args, kwargs):
            if kwargs is None:
                kwargs = {}
            try:
                self._results = func(*args, **kwargs)
            except Exception as e:
                self._error_str = str(e)
            self._is_finished = True
            self.finished.emit()
    
    
    def convert_to_reply(func):
        def wrapper(*args, **kwargs):
            reply = Reply(func, args, kwargs)
            return reply
    
        return wrapper
    
    
    class Backend(QtCore.QObject):
        started = QtCore.pyqtSignal(QtCore.QUrl)
        finished = QtCore.pyqtSignal()
    
        def __init__(self, parent=None):
            super().__init__(parent)
            self._service = None
    
        @property
        def service(self):
            if self._service is None:
                reply = self._update_credentials()
                loop = QtCore.QEventLoop()
                reply.finished.connect(loop.quit)
                loop.exec_()
                if not reply.has_error():
                    self._service = reply.results
                else:
                    logging.debug(reply.error_str)
            return self._service
    
        @convert_to_reply
        def _update_credentials(self):
            creds = None
            if os.path.exists("token.pickle"):
                with open("token.pickle", "rb") as token:
                    creds = pickle.load(token)
            if not creds or not creds.valid:
                if creds and creds.expired and creds.refresh_token:
                    creds.refresh(Request())
                else:
                    flow = InstalledAppFlow.from_client_secrets_file(
                        "credentials.json", SCOPES
                    )
                    host = "localhost"
                    port = 8080
                    state = "default"
                    QtCore.QTimer.singleShot(
                        0, functools.partial(self.get_url, flow, host, port, state)
                    )
                    creds = flow.run_local_server(
                        host=host, port=port, open_browser=False, state=state
                    )
                    self.finished.emit()
                with open("token.pickle", "wb") as token:
                    pickle.dump(creds, token)
            return build("oauth2", "v2", credentials=creds)
    
        def get_url(self, flow, host, port, state):
            flow.redirect_uri = "http://{}:{}/".format(host, port)
            redirect_uri, _ = flow.authorization_url(state=state)
            self.started.emit(QtCore.QUrl.fromUserInput(redirect_uri))
    
        @convert_to_reply
        def get_user_info(self):
            return self.service.userinfo().get().execute()
    
    
    class Widget(QtWidgets.QWidget):
        def __init__(self, parent=None):
            super().__init__(parent)
    
            self.backend = Backend()
    
            self.webengineview = QtWebEngineWidgets.QWebEngineView()
            self.webengineview.page().profile().setHttpUserAgent(
                "Mozilla/5.0 (X11; Linux x86_64; rv:57.0) Gecko/20100101 Firefox/57.0"
            )
            self.webengineview.hide()
            button = QtWidgets.QPushButton("Sign in")
    
            lay = QtWidgets.QVBoxLayout(self)
            lay.addWidget(button)
            lay.addWidget(self.webengineview)
    
            button.clicked.connect(self.sign_in)
            self.backend.started.connect(self.handle_url_changed)
            self.backend.finished.connect(self.webengineview.hide)
    
            self.resize(640, 480)
    
        def sign_in(self):
            reply = self.backend.get_user_info()
            wrapper = functools.partial(self.handle_finished_user_info, reply)
            reply.finished.connect(wrapper)
    
        def handle_finished_user_info(self, reply):
            if reply.has_error():
                logging.debug(reply.error_str)
            else:
                profile_info = reply.results
                print(profile_info)
    
        def handle_url_changed(self, url):
            self.webengineview.load(url)
            self.webengineview.show()
    
    
    if __name__ == "__main__":
        import sys
    
        app = QtWidgets.QApplication(sys.argv)
        w = Widget()
        w.show()
        sys.exit(app.exec_())