PyQt5 应用程序的备用 OAuth2 登录 [英] Alternate OAuth2 sign in for a PyQt5 application

查看:72
本文介绍了PyQt5 应用程序的备用 OAuth2 登录的处理方法,对大家解决问题具有一定的参考价值,需要的朋友们下面随着小编来一起学习吧!

问题描述

我有一个带有 Google 登录的 PyQt5 应用程序,它是使用 oauth2client 实现的.登录页面使用 QWebEngineView 显示在嵌入式浏览器中.但是使用 从 2021 年 1 月 4 日起,使用嵌入式浏览器的 Google 阻止登录工作流程,我的应用程序需要进行更改,以改为打开系统浏览器,然后接收来自该浏览器的授权响应.为此,我使用 google-auth-oauthlib,这是也用于 Google Python 快速入门文档.

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.

我创建了一个仅实现此工作流程的小型 POC,但我面临几个问题:

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

  1. 登录页面在新的浏览器窗口或选项卡中打开.但这可能不是一个很好的用户体验,因为用户必须在登录后关闭浏览器,然后再返回到应用程序.在这里打开弹出式浏览器似乎是一个更好的用户体验.我查看了负责打开浏览器的run_local_server方法的源码,他们好像使用了webbrowser模块,可惜没有打开弹窗的方法.

  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.

如果用户在未登录的情况下关闭了使用 run_local_server 方法打开的浏览器,则调用它的应用程序只会冻结并需要强制退出.我也没有注意到任何控制台错误.有没有办法用我正在使用的库来处理这个问题?

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?

这是最小的工作示例:

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

推荐答案

QWebEngineView 可用作浏览器进行身份验证,但您必须设置有效的用户代理.另一方面,google-auth-oauthlib 请求被阻塞,因此它们必须在不同的线程中执行并通过信号通知结果:

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

这篇关于PyQt5 应用程序的备用 OAuth2 登录的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!

查看全文
登录 关闭
扫码关注1秒登录
发送“验证码”获取 | 15天全站免登陆