如何使用Twisted查看OAuth2.0认证的Gmail
How to use Twisted to check Gmail with OAuth2.0 authentication
我有一个用于 Google 邮件的 IMAP 客户端,但它最近停止工作了。我认为问题是 gmail 不再允许 TTL username/password 登录,但现在需要 OAuth2.0.
我想知道更改下面示例的最佳方法,以便我的扭曲 IMAP 客户端使用 OAuth2.0 进行身份验证。 (并且在没有 Google API 包的情况下这样做,如果可能的话。)
示例使用 username/password 登录(不再有效)
class AriSBDGmailImap4Client(imap4.IMAP4Client):
'''
client to fetch and process SBD emails from gmail. the messages
contained in the emails are sent to the AriSBDStationProtocol for
this sbd modem.
'''
def __init__(self, contextFactory=None):
imap4.IMAP4Client.__init__(self, contextFactory)
@defer.inlineCallbacks
def serverGreeting(self, caps):
# log in
try:
# the line below no longer works for gmail
yield self.login(mailuser, mailpass)
try:
yield self.uponAuthentication()
except Exception as e:
uponFail(e, "uponAuthentication")
except Exception as e:
uponFail(e, "logging in")
# done. log out
try:
yield self.logout()
except Exception as e:
uponFail(e, "logging out")
@defer.inlineCallbacks
def uponAuthentication(self):
try:
yield self.select('Inbox')
try:
# read messages, etc, etc
pass
except Exception as e:
uponFail(e, "searching unread")
except Exception as e:
uponFail(e, "selecting inbox")
我为这个客户准备了一个简单的工厂。它通过使用 reactor.connectSSL
和 Google 邮件主机 url 和端口开始。
我已经按照 https://developers.google.com/gmail/api/quickstart/quickstart-python 的说明进行了 "installed app"(但我不知道这是否是正确的选择)。我可以 运行 他们的 "quickstart.py" 示例成功。
我快速而肮脏的尝试(不起作用)
@defer.inlineCallbacks
def serverGreeting(self, caps):
# log in
try:
#yield self.login(mailuser, mailpass)
flow = yield threads.deferToThread(
oauth2client.client.flow_from_clientsecrets,
filename=CLIENT_SECRET_FILE,
scope=OAUTH_SCOPE)
http = httplib2.Http()
credentials = yield threads.deferToThread( STORAGE.get )
if credentials is None or credentials.invalid:
parser = argparse.ArgumentParser(
parents=[oauth2client.tools.argparser])
flags = yield threads.deferToThread( parser.parse_args )
credentials = yield threads.deferToThread(
oauth2client.tools.run_flow,
flow=flow,
storage=STORAGE,
flags=flags, http=http)
http = yield threads.deferToThread(
credentials.authorize, http)
gmail_service = yield threads.deferToThread(
apiclient.discovery.build,
serviceName='gmail',
version='v1',
http=http)
self.state = 'auth'
try:
yield self.uponAuthentication()
except Exception as e:
uponFail(e, "uponAuthentication")
except Exception as e:
uponFail(e, "logging in")
# done. log out
try:
yield self.logout()
except Exception as e:
uponFail(e, "logging out")
我基本上只是将 "quickstart.py" 复制到 serverGreeting
,然后尝试将客户端状态设置为 "auth"。
这样验证就好了,但是后来扭曲了无法select收件箱:
[AriSBDGmailImap4Client (TLSMemoryBIOProtocol),client] FAIL: Unknown command {random gibberish}
随机乱码有字母和数字,每次 select 收件箱命令失败时都不同。
感谢您的帮助!
经过大量阅读和测试,我终于能够使用 OAuth2 实现 gmail 的有效登录。
一个重要的注意事项是,使用 "service account" 的两步过程对我来说 不 有效。我仍然不清楚为什么不能使用此过程,但服务帐户似乎无法访问同一帐户中的 gmail。即使服务帐户具有 "can edit" 权限并且启用了 Gmail API,也是如此。
有用的参考资料
使用概览
OAuth2 https://developers.google.com/identity/protocols/OAuth2
将 OAuth2 与 "Installed Applications" 结合使用的指南
https://developers.google.com/identity/protocols/OAuth2InstalledApp
设置帐户以将 OAuth2 与 "Installed Applications" 一起使用的指南
https://developers.google.com/api-client-library/python/auth/installed-app
没有完整的 OAuth2 例程的集合 Google API
https://code.google.com/p/google-mail-oauth2-tools/wiki/OAuth2DotPyRunThrough
第 1 步 - 获取 Google 客户端 ID
使用 gmail 帐户登录 https://console.developers.google.com/
启动一个项目,启用 Gmail API 并为已安装的应用程序创建一个新的客户端 ID。 https://developers.google.com/api-client-library/python/auth/installed-app#creatingcred
处的说明
单击 "Download JSON" 按钮并将此文件保存在 public 无法访问的位置(因此可能不在代码存储库中)。
步骤 2 - 获取 Google OAuth2 Python 工具
从 https://code.google.com/p/google-mail-oauth2-tools/wiki/OAuth2DotPyRunThrough
下载 oauth2.py 脚本
第 3 步 - 获取授权 URL
使用第 2 步中的脚本获取 URL 允许您授权您的 Google 项目。
在终端中:
python oauth2.py --user={myaccount@gmail.com} --client_id={your client_id from the json file} --client_secret={your client_secret from the json file} --generate_oauth2_token
第 4 步 -- 获取授权码
将步骤 3 中的 URL 粘贴到您的浏览器中,然后单击 "accept" 按钮。
从网页复制代码。
将代码粘贴到终端中并按回车键。您将获得:
To authorize token, visit this url and follow the directions: https://accounts.google.com/o/oauth2/auth?client_id{...}
Enter verification code: {...}
Refresh Token: {...}
Access Token: {...}
Access Token Expiration Seconds: 3600
第 5 步 - 保存刷新令牌
从终端复制刷新令牌并将其保存在某处。在本例中,我将其保存为 json 格式的文本文件,关键字为 "Refresh Token"。但它也可以保存到私有数据库中。
确保 public!
无法访问刷新令牌
第 6 步 - 制作扭曲的身份验证器
这是一个 OAuth2 身份验证器的工作示例。它需要步骤 2 中的 oauth2.py 脚本。
import json
import oauth2
from zope.interface import implementer
from twisted.internet import threads
MY_GMAIL = {your gmail address}
REFRESH_TOKEN_SECRET_FILE = {name of your refresh token file from Step 5}
CLIENT_SECRET_FILE = {name of your cliend json file from Step 1}
@implementer(imap4.IClientAuthentication)
class GmailOAuthAuthenticator():
authName = "XOAUTH2"
tokenTimeout = 3300 # 5 mins short of the real timeout (1 hour)
def __init__(self, reactr):
self.token = None
self.reactor = reactr
self.expire = None
@defer.inlineCallbacks
def getToken(self):
if ( (self.token==None) or (self.reactor.seconds() > self.expire) ):
rt = None
with open(REFRESH_TOKEN_SECRET_FILE) as f:
rt = json.load(f)
cl = None
with open(CLIENT_SECRET_FILE) as f:
cl = json.load(f)
self.token = yield threads.deferToThread(
oauth2.RefreshToken,
client_id = cl['installed']['client_id'],
client_secret = cl['installed']['client_secret'],
refresh_token = rt['Refresh Token'] )
self.expire = self.reactor.seconds() + self.tokenTimeout
def getName(self):
return self.authName
def challengeResponse(self, secret, chal):
# we MUST already have the token
# (allow an exception to be thrown if not)
t = self.token['access_token']
ret = oauth2.GenerateOAuth2String(MY_GMAIL, t, False)
return ret
步骤 7 - 为协议注册 Authenitcator
在 IMAP4ClientFactory 中:
def buildProtocol(self, addr):
p = self.protocol(self.ctx)
p.factory = self
x = GmailOAuthAuthenticator(self.reactor)
p.registerAuthenticator(x)
return p
步骤 8 - 使用访问令牌进行身份验证
不使用 "login",而是获取访问令牌(如有必要),然后使用身份验证。
更改问题中的示例代码:
@defer.inlineCallbacks
def serverGreeting(self, caps):
# log in
try:
# the line below no longer works for gmail
# yield self.login(mailuser, mailpass)
if GmailOAuthAuthenticator.authName in self.authenticators:
yield self.authenticators[AriGmailOAuthAuthenticator.authName].getToken()
yield self.authenticate("")
try:
yield self.uponAuthentication()
except Exception as e:
uponFail(e, "uponAuthentication")
except Exception as e:
uponFail(e, "logging in")
我有一个用于 Google 邮件的 IMAP 客户端,但它最近停止工作了。我认为问题是 gmail 不再允许 TTL username/password 登录,但现在需要 OAuth2.0.
我想知道更改下面示例的最佳方法,以便我的扭曲 IMAP 客户端使用 OAuth2.0 进行身份验证。 (并且在没有 Google API 包的情况下这样做,如果可能的话。)
示例使用 username/password 登录(不再有效)
class AriSBDGmailImap4Client(imap4.IMAP4Client):
'''
client to fetch and process SBD emails from gmail. the messages
contained in the emails are sent to the AriSBDStationProtocol for
this sbd modem.
'''
def __init__(self, contextFactory=None):
imap4.IMAP4Client.__init__(self, contextFactory)
@defer.inlineCallbacks
def serverGreeting(self, caps):
# log in
try:
# the line below no longer works for gmail
yield self.login(mailuser, mailpass)
try:
yield self.uponAuthentication()
except Exception as e:
uponFail(e, "uponAuthentication")
except Exception as e:
uponFail(e, "logging in")
# done. log out
try:
yield self.logout()
except Exception as e:
uponFail(e, "logging out")
@defer.inlineCallbacks
def uponAuthentication(self):
try:
yield self.select('Inbox')
try:
# read messages, etc, etc
pass
except Exception as e:
uponFail(e, "searching unread")
except Exception as e:
uponFail(e, "selecting inbox")
我为这个客户准备了一个简单的工厂。它通过使用 reactor.connectSSL
和 Google 邮件主机 url 和端口开始。
我已经按照 https://developers.google.com/gmail/api/quickstart/quickstart-python 的说明进行了 "installed app"(但我不知道这是否是正确的选择)。我可以 运行 他们的 "quickstart.py" 示例成功。
我快速而肮脏的尝试(不起作用)
@defer.inlineCallbacks
def serverGreeting(self, caps):
# log in
try:
#yield self.login(mailuser, mailpass)
flow = yield threads.deferToThread(
oauth2client.client.flow_from_clientsecrets,
filename=CLIENT_SECRET_FILE,
scope=OAUTH_SCOPE)
http = httplib2.Http()
credentials = yield threads.deferToThread( STORAGE.get )
if credentials is None or credentials.invalid:
parser = argparse.ArgumentParser(
parents=[oauth2client.tools.argparser])
flags = yield threads.deferToThread( parser.parse_args )
credentials = yield threads.deferToThread(
oauth2client.tools.run_flow,
flow=flow,
storage=STORAGE,
flags=flags, http=http)
http = yield threads.deferToThread(
credentials.authorize, http)
gmail_service = yield threads.deferToThread(
apiclient.discovery.build,
serviceName='gmail',
version='v1',
http=http)
self.state = 'auth'
try:
yield self.uponAuthentication()
except Exception as e:
uponFail(e, "uponAuthentication")
except Exception as e:
uponFail(e, "logging in")
# done. log out
try:
yield self.logout()
except Exception as e:
uponFail(e, "logging out")
我基本上只是将 "quickstart.py" 复制到 serverGreeting
,然后尝试将客户端状态设置为 "auth"。
这样验证就好了,但是后来扭曲了无法select收件箱:
[AriSBDGmailImap4Client (TLSMemoryBIOProtocol),client] FAIL: Unknown command {random gibberish}
随机乱码有字母和数字,每次 select 收件箱命令失败时都不同。
感谢您的帮助!
经过大量阅读和测试,我终于能够使用 OAuth2 实现 gmail 的有效登录。
一个重要的注意事项是,使用 "service account" 的两步过程对我来说 不 有效。我仍然不清楚为什么不能使用此过程,但服务帐户似乎无法访问同一帐户中的 gmail。即使服务帐户具有 "can edit" 权限并且启用了 Gmail API,也是如此。
有用的参考资料
使用概览 OAuth2 https://developers.google.com/identity/protocols/OAuth2
将 OAuth2 与 "Installed Applications" 结合使用的指南 https://developers.google.com/identity/protocols/OAuth2InstalledApp
设置帐户以将 OAuth2 与 "Installed Applications" 一起使用的指南 https://developers.google.com/api-client-library/python/auth/installed-app
没有完整的 OAuth2 例程的集合 Google API https://code.google.com/p/google-mail-oauth2-tools/wiki/OAuth2DotPyRunThrough
第 1 步 - 获取 Google 客户端 ID
使用 gmail 帐户登录 https://console.developers.google.com/
启动一个项目,启用 Gmail API 并为已安装的应用程序创建一个新的客户端 ID。 https://developers.google.com/api-client-library/python/auth/installed-app#creatingcred
处的说明单击 "Download JSON" 按钮并将此文件保存在 public 无法访问的位置(因此可能不在代码存储库中)。
步骤 2 - 获取 Google OAuth2 Python 工具
从 https://code.google.com/p/google-mail-oauth2-tools/wiki/OAuth2DotPyRunThrough
下载 oauth2.py 脚本第 3 步 - 获取授权 URL
使用第 2 步中的脚本获取 URL 允许您授权您的 Google 项目。
在终端中:
python oauth2.py --user={myaccount@gmail.com} --client_id={your client_id from the json file} --client_secret={your client_secret from the json file} --generate_oauth2_token
第 4 步 -- 获取授权码
将步骤 3 中的 URL 粘贴到您的浏览器中,然后单击 "accept" 按钮。
从网页复制代码。
将代码粘贴到终端中并按回车键。您将获得:
To authorize token, visit this url and follow the directions: https://accounts.google.com/o/oauth2/auth?client_id{...}
Enter verification code: {...}
Refresh Token: {...}
Access Token: {...}
Access Token Expiration Seconds: 3600
第 5 步 - 保存刷新令牌
从终端复制刷新令牌并将其保存在某处。在本例中,我将其保存为 json 格式的文本文件,关键字为 "Refresh Token"。但它也可以保存到私有数据库中。
确保 public!
无法访问刷新令牌第 6 步 - 制作扭曲的身份验证器
这是一个 OAuth2 身份验证器的工作示例。它需要步骤 2 中的 oauth2.py 脚本。
import json
import oauth2
from zope.interface import implementer
from twisted.internet import threads
MY_GMAIL = {your gmail address}
REFRESH_TOKEN_SECRET_FILE = {name of your refresh token file from Step 5}
CLIENT_SECRET_FILE = {name of your cliend json file from Step 1}
@implementer(imap4.IClientAuthentication)
class GmailOAuthAuthenticator():
authName = "XOAUTH2"
tokenTimeout = 3300 # 5 mins short of the real timeout (1 hour)
def __init__(self, reactr):
self.token = None
self.reactor = reactr
self.expire = None
@defer.inlineCallbacks
def getToken(self):
if ( (self.token==None) or (self.reactor.seconds() > self.expire) ):
rt = None
with open(REFRESH_TOKEN_SECRET_FILE) as f:
rt = json.load(f)
cl = None
with open(CLIENT_SECRET_FILE) as f:
cl = json.load(f)
self.token = yield threads.deferToThread(
oauth2.RefreshToken,
client_id = cl['installed']['client_id'],
client_secret = cl['installed']['client_secret'],
refresh_token = rt['Refresh Token'] )
self.expire = self.reactor.seconds() + self.tokenTimeout
def getName(self):
return self.authName
def challengeResponse(self, secret, chal):
# we MUST already have the token
# (allow an exception to be thrown if not)
t = self.token['access_token']
ret = oauth2.GenerateOAuth2String(MY_GMAIL, t, False)
return ret
步骤 7 - 为协议注册 Authenitcator
在 IMAP4ClientFactory 中:
def buildProtocol(self, addr):
p = self.protocol(self.ctx)
p.factory = self
x = GmailOAuthAuthenticator(self.reactor)
p.registerAuthenticator(x)
return p
步骤 8 - 使用访问令牌进行身份验证
不使用 "login",而是获取访问令牌(如有必要),然后使用身份验证。
更改问题中的示例代码:
@defer.inlineCallbacks
def serverGreeting(self, caps):
# log in
try:
# the line below no longer works for gmail
# yield self.login(mailuser, mailpass)
if GmailOAuthAuthenticator.authName in self.authenticators:
yield self.authenticators[AriGmailOAuthAuthenticator.authName].getToken()
yield self.authenticate("")
try:
yield self.uponAuthentication()
except Exception as e:
uponFail(e, "uponAuthentication")
except Exception as e:
uponFail(e, "logging in")