python - Alternate OAuth2 sign in for a PyQt5 application

Question

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()

Answer 1

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_())