准备工作
需要用到的模块:
Odoo/auth_oauth
OCA/server-auth/auth_oidc
因为钉钉的Oauth授权流程比较特殊,需要继承改造res.users上的几个方法
models\res_users.py
class ResUsers(models.Model):
_inherit = "res.users"
dingtalk_userid = fields.Char('Dingtalk User ID', copy=False)
# 授权回调后换取acccess_token
def _auth_oauth_get_tokens_auth_code_flow(self, oauth_provider, params):
if 'dingtalk' in oauth_provider.auth_endpoint:
# https://openid.net/specs/openid-connect-core-1_0.html#AuthResponse
code = params.get("code")
# https://openid.net/specs/openid-connect-core-1_0.html#TokenRequest
auth = None
if oauth_provider.client_secret:
auth = (oauth_provider.client_id, oauth_provider.client_secret)
response = requests.post(
oauth_provider.token_endpoint,
json=dict(
clientId=oauth_provider.client_id,
clientSecret=oauth_provider.client_secret,
grantType="authorization_code",
code=code,
),
auth=auth,
timeout=10,
)
response.raise_for_status()
response_json = response.json()
# https://openid.net/specs/openid-connect-core-1_0.html#TokenResponse
return response_json.get("accessToken"), response_json.get("refreshToken")
else:
return super()._auth_oauth_get_tokens_auth_code_flow(oauth_provider, params)
@api.model
def auth_oauth(self, provider, params):
oauth_provider = self.env["auth.oauth.provider"].browse(provider)
if oauth_provider.flow == "id_token_code" and 'dingtalk' in oauth_provider.auth_endpoint:
access_token, refresh_token = self._auth_oauth_get_tokens_auth_code_flow(
oauth_provider, params
)
if not access_token:
_logger.error("No access_token in response.")
raise AccessDenied()
# 获取unionid
resp = requests.get(oauth_provider.validation_endpoint, headers={'x-acs-dingtalk-access-token': access_token})
res = resp.json()
if resp.status_code != 200:
_logger.error("Failed to get user info.\n" % res)
raise AccessDenied()
# 通过unionid换取钉钉用户id
dingtalk_userid = self.env['dingtalk.client'].get_userid_by_unionid(res['unionId'])
if not dingtalk_userid:
raise AccessDenied()
# 查询内部用户
user_id = self.env['res.users'].sudo().search([('dingtalk_userid', '=', dingtalk_userid)], limit=1)
if not user_id.login:
raise AccessDenied()
user_id.write({'oauth_uid': res['unionId'], 'oauth_access_token': access_token})
self.delete_other_session()
# return user credentials
return (self.env.cr.dbname, user_id.login, access_token)
else:
return super().auth_oauth(provider, params)
需要预先用服务端接口把钉钉的userid同步进odoo上
controllers\oauth_login.py
# -*- coding: utf-8 -*-
from odoo.addons.auth_oauth.controllers.main import OAuthLogin
class DingtalkAuthLogin(OAuthLogin):
# 针对钉钉授权登录中的prompt参数,配置在auth_endpoint里:https://login.dingtalk.com/oauth2/auth?prompt=consent
def list_providers(self):
providers = super().list_providers()
for provider in providers:
if provider['auth_link'].count('?') > 1:
auth_params = provider['auth_link'].split('?')
provider['auth_link'] = '%s?%s' % (auth_params[0], '&'.join(auth_params[1:]))
return providers