pengyuthon/gzth
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

免登录

Browse Source
This commit is contained in:
pengyuthon
2026-02-23 11:27:13 +08:00
parent c855b1925e
commit 64c5d6e01a
5 changed files with 611 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
yuthon_wechat/controllers/main.py
+12
View File
@@ -10,6 +10,8 @@ from odoo.http import request
from odoo.exceptions import AccessDenied
from werkzeug.wrappers import Response as WerkzeugResponse
from ..utils.common import WeChatOAuthException
_logger = logging.getLogger(__name__)
@@ -209,6 +211,16 @@ class Home(http.Controller):
}),
headers=[('Content-Type', 'application/json')]
)
except WeChatOAuthException as e:
_logger.error(e)
return request.make_response(
json.dumps({
'code': e.errcode,
'message': f'企业微信认证失败: {str(e.errmsg)}',
'data': None
}),
headers=[('Content-Type', 'application/json')]
)
except Exception as e:
_logger.error(f"企业微信OAuth认证异常: {str(e)}", exc_info=True)
return request.make_response(
yuthon_wechat/controllers/main2.py
+343
View File
@@ -0,0 +1,343 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from urllib.parse import quote
import json
import uuid
import logging
from odoo import http
from odoo.http import request
from odoo.exceptions import AccessDenied
from werkzeug.wrappers import Response as WerkzeugResponse
from ..utils.common import WeChatOAuthException
_logger = logging.getLogger(__name__)
class Home(http.Controller):
@http.route('/WW_verify_fC7AX158Ffby8Ed1.txt', type='http', auth="public")
def WW_verify_UP42BAWNwtWK60RH(self):
return 'fC7AX158Ffby8Ed1'
@http.route('/wechat/login', type='http', auth="none", methods=['GET'])
def wechat_login(self):
base_url = request.env['ir.config_parameter'].sudo().get_param('web.base.url')
redirect_uri = quote(f'{base_url}/wechat/login_by_code')
redirect_uri = redirect_uri.replace(' ', '%20')
appid = 'wwc955c117b738b052'
agent_id = '1000120'
url = f'https://open.weixin.qq.com/connect/oauth2/authorize?appid={appid}&redirect_uri={redirect_uri}&response_type=code&scope=snsapi_base&state=STATE&agentid={agent_id}#wechat_redirect'
return http.request.redirect(url, 301, False)
@http.route('/wechat/login_by_code', type='http', auth="none", methods=['GET'])
def wechat_login_by_code(self, code, state=None, **kw):
"""
企业微信OAuth回调接口，支持state参数重定向到目标URL
参数:
code: 企业微信OAuth授权码
state: 目标URL（编码后的），登录成功后重定向到此URL
"""
provider = request.env.ref('yuthon_wechat.wechat_corp_web_auth_oauth_provider')
(db, login, oauth_code) = request.env['res.users'].sudo().auth_oauth_wechat(provider.sudo(), {'code': code})
# 如果没有state参数，返回原始值，保持与原本方法行为一致
if not state:
return (db, login, oauth_code)
# 如果有state参数，执行登录并重定向到目标URL
try:
# 查找用户并创建session
user = request.env['res.users'].sudo().search([('login', '=', login)], limit=1)
if not user:
_logger.error(f"OAuth登录后未找到用户: {login}")
return http.request.redirect('/web/login?error=user_not_found', 302)
# 创建session
dbname = request.env.cr.dbname
try:
uid = request.session.authenticate(dbname, user.login, user.login)
except Exception:
# 如果authenticate失败，直接设置session
request.session.uid = user.id
request.session.login = user.login
request.session.db = dbname
uid = user.id
_logger.info(f"企业微信OAuth登录成功: 用户 {user.login} (ID: {user.id})")
# 解码并重定向到目标URL
try:
from urllib.parse import unquote
target_url = unquote(state)
# 确保是相对路径或同域名
if target_url.startswith('http') and 'oa.thtzjt.com' not in target_url:
# 如果不是同域名，默认重定向到/web
target_url = '/web'
_logger.info(f"OAuth登录成功，重定向到: {target_url}")
return http.request.redirect(target_url, 302)
except Exception as e:
_logger.error(f"解析state参数失败: {str(e)}")
return http.request.redirect('/web', 302)
except AccessDenied as e:
_logger.error(f"企业微信OAuth登录失败: {str(e)}")
return http.request.redirect('/web/login?error=access_denied', 302)
except Exception as e:
_logger.exception(f"企业微信OAuth回调处理异常: {str(e)}")
return http.request.redirect('/web/login?error=oauth_error', 302)
@http.route('/wechat/login_by_user_id', type='http', auth="none", methods=['GET', 'POST'], csrf=False)
def wechat_login_by_user_id(self, user_id=None, login=None, **kw):
"""
免登录接口：通过用户ID或登录名直接登录，绕过密码校验
/wechat/login_by_user_id?user_id=2&redirect_url=/web#action=3584
/wechat/login_by_user_id?user_id=2&redirect_url=/web#action=3584
https://oa.thtzjt.com/web#action=3584&cids=3-1-2-4&menu_id=9668
参数:
user_id: 用户ID (res.users的id)
login: 用户登录名 (res.users的login字段)
返回:
JSON格式的响应，包含登录结果
"""
try:
# 优先使用user_id，如果没有则使用login查找
if user_id:
try:
user_id = int(user_id)
user = request.env['res.users'].sudo().browse(user_id)
except (ValueError, TypeError):
return json.dumps({
'success': False,
'message': '用户ID格式错误'
}), 200, [('Content-Type', 'application/json')]
elif login:
user = request.env['res.users'].sudo().search([('login', '=', login)], limit=1)
else:
return json.dumps({
'success': False,
'message': '请提供user_id或login参数'
}), 200, [('Content-Type', 'application/json')]
# 验证用户是否存在且激活
if not user or not user.exists():
return json.dumps({
'success': False,
'message': '用户不存在'
}), 200, [('Content-Type', 'application/json')]
if not user.active:
return json.dumps({
'success': False,
'message': '用户已被禁用'
}), 200, [('Content-Type', 'application/json')]
# 直接设置session来绕过密码验证
dbname = request.env.cr.dbname
# 方法1: 尝试使用authenticate（如果用户密码恰好是login）
try:
uid = request.session.authenticate(dbname, user.login, user.login)
except Exception:
# 方法2: 如果authenticate失败，直接设置session
# 这是绕过密码验证的关键步骤
request.session.uid = user.id
request.session.login = user.login
request.session.db = dbname
uid = user.id
_logger.info(f"免登录接口：用户 {user.login} (ID: {user.id}) 登录成功")
# 重定向到主页或返回成功信息
redirect_url = kw.get('redirect_url', '/web')
if kw.get('return_json', 'false').lower() == 'true':
return json.dumps({
'success': True,
'message': '登录成功',
'user_id': user.id,
'login': user.login,
'name': user.name
}), 200, [('Content-Type', 'application/json')]
else:
return http.request.redirect(redirect_url, 302)
except Exception as e:
_logger.exception(f"免登录接口错误: {str(e)}")
return json.dumps({
'success': False,
'message': f'登录失败: {str(e)}'
}), 200, [('Content-Type', 'application/json')]
@http.route('/phone/wechat/login', type='http', auth="none", csrf=False, cors='*', methods=['POST'])
def phone_wechat_login(self, **kw):
"""手机端企业微信OAuth登录接口"""
try:
code = kw.get('code')
if not code:
return request.make_response(
json.dumps({
'code': 400,
'message': '缺少code参数',
'data': None
}),
headers=[('Content-Type', 'application/json')]
)
try:
provider = request.env.ref('yuthon_wechat.wechat_corp_web_auth_oauth_provider').sudo()
except Exception as e:
_logger.error(f"获取企业微信OAuth provider失败: {str(e)}")
return request.make_response(
json.dumps({
'code': 500,
'message': '企业微信配置错误',
'data': None
}),
headers=[('Content-Type', 'application/json')]
)
try:
(db, login, oauth_code) = request.env['res.users'].sudo().auth_oauth_wechat(provider, {'code': code})
except AccessDenied as e:
_logger.warning(f"企业微信登录失败: {str(e)}")
return request.make_response(
json.dumps({
'code': 403,
'message': str(e) or '未找到对应的用户账号',
'data': None
}),
headers=[('Content-Type', 'application/json')]
)
except WeChatOAuthException as e:
_logger.error(e)
return request.make_response(
json.dumps({
'code': e.errcode,
'message': f'企业微信认证失败: {str(e.errmsg)}',
'data': None
}),
headers=[('Content-Type', 'application/json')]
)
except Exception as e:
_logger.error(f"企业微信OAuth认证异常: {str(e)}", exc_info=True)
return request.make_response(
json.dumps({
'code': 500,
'message': f'企业微信认证失败: {str(e)}',
'data': None
}),
headers=[('Content-Type', 'application/json')]
)
user = request.env['res.users'].sudo().search([('login', '=', login)], limit=1)
if not user:
_logger.warning(f"未找到用户: {login}")
return request.make_response(
json.dumps({
'code': 404,
'message': '未找到对应的用户账号',
'data': None
}),
headers=[('Content-Type', 'application/json')]
)
employee = user.with_company(user.company_id.id).employee_id
if not employee:
oauth_uid = user.oauth_uid
if oauth_uid:
employee = request.env['hr.employee'].sudo().search([
('wecom_userid', '=', oauth_uid)
], limit=1)
if not employee:
_logger.warning(f"未找到对应的员工信息, user: {login}")
return request.make_response(
json.dumps({
'code': 404,
'message': '未找到对应的员工信息',
'data': None
}),
headers=[('Content-Type', 'application/json')]
)
# access_list = []
# applet_security_setting_ids = request.env['applet.security'].search(
# [('employee_id', '=', employee.id)]).applet_security_setting_ids
# access_list = [access.name for access in applet_security_setting_ids]
# _logger.info(f"企业微信登录用户{user.name},权限组{access_list}---")
applet_security_setting_ids = request.env['applet.security'].sudo().search(
[('employee_id', '=', employee.id)]).applet_security_setting_ids
access_list = [access.name for access in applet_security_setting_ids]
_logger.info(f"企业微信登录用户{user.name},权限组{access_list}---")
company_list = []
for company in user.company_ids:
company_dict = {
'state': 1,
'company_id': company.id,
'company_name': company.name,
}
company_list.append(company_dict)
token = str(uuid.uuid4())
data = {
'name': employee.name,
'id': user.id,
'access': access_list,
'phone': employee.mobile_phone,
'company_data': company_list,
'token': token,
}
return request.make_response(
json.dumps({
'code': 200,
'message': '',
'data': data
}),
headers=[('Content-Type', 'application/json')]
)
except Exception as e:
_logger.error(f"企业微信登录异常: {str(e)}", exc_info=True)
return request.make_response(
json.dumps({
'code': 500,
'message': f'登录失败: {str(e)}',
'data': None
}),
headers=[('Content-Type', 'application/json')]
)
@http.route('/phone/wechat/callback', type='http', auth="none", csrf=False, cors='*', methods=['GET'])
def phone_wechat_callback(self, code=None, state=None, **kw):
try:
frontend_base_url = 'https://phone.thtzjt.com'
if code:
redirect_url = f"{frontend_base_url}/phone/WechatCallback?code={code}"
if state:
redirect_url += f"&state={state}"
_logger.info(f"手机端OAuth回调，重定向到: {redirect_url}")
else:
redirect_url = f"{frontend_base_url}/phone/Login"
_logger.warning("手机端OAuth回调缺少code参数，重定向到登录页")
response = WerkzeugResponse(
'',
status=302,
headers=[('Location', redirect_url)]
)
return response
except Exception as e:
_logger.error(f"企业微信回调处理异常: {str(e)}", exc_info=True)
frontend_base_url = 'https://phone.thtzjt.com'
redirect_url = f"{frontend_base_url}/phone/Login"
return WerkzeugResponse(
'',
status=302,
headers=[('Location', redirect_url)]
)
yuthon_wechat/models/res_users.py
+50 -58
View File
@@ -10,27 +10,24 @@ from odoo import api, fields, models
from odoo.tools.misc import ustr
from odoo.exceptions import AccessDenied
from odoo.http import request
from odoo.addons.auth_signup.models.res_users import SignupError
# from odoo.addons.auth_signup.models.res_users import SignupError
from odoo.addons.auth_oauth.controllers.main import OAuthLogin
import logging
from ..utils.common import WeChatOAuthException
_logger = logging.getLogger(__name__)
# 模块级内存缓存，存储企业微信 access_token，避免写入数据库触发 registry signaling
_wechat_token_cache = {
'access_token': None,
'expire_time': 0,
}
class OAuthLoginWechat(OAuthLogin):
def list_providers(self):
providers = super(OAuthLoginWechat, self).list_providers()
for provider in providers:
if provider['auth_endpoint'].find('weixin') >= 0:
#对微信的回调链接做特殊处理
#企业微信扫码登录链接 https://open.work.weixin.qq.com/wwopen/sso/qrConnect?appid=CORPID&agentid=AGENTID&redirect_uri=REDIRECT_URI&state=STATE
#企业微信内直接登录链接 https://open.weixin.qq.com/connect/oauth2/authorize?appid=CORPID&redirect_uri=REDIRECT_URI&response_type=code&scope=snsapi_base&state=STATE#wechat_redirect
if provider['auth_endpoint'].find('weixin') >= 0:
# 对微信的回调链接做特殊处理
# 企业微信扫码登录链接 https://open.work.weixin.qq.com/wwopen/sso/qrConnect?appid=CORPID&agentid=AGENTID&redirect_uri=REDIRECT_URI&state=STATE
# 企业微信内直接登录链接 https://open.weixin.qq.com/connect/oauth2/authorize?appid=CORPID&redirect_uri=REDIRECT_URI&response_type=code&scope=snsapi_base&state=STATE#wechat_redirect
# 如果auth_endpoint链接中含有 “?create_user” 字样，则自动创建用户
wechat_url = provider['auth_endpoint']
create_user = False
@@ -38,7 +35,7 @@ class OAuthLoginWechat(OAuthLogin):
if len(auth_endpoint) >= 2:
wechat_url = auth_endpoint[0]
create_user = auth_endpoint[1].find('create_user') >= 0
client_ids = provider['client_id'].split(' ')
return_url = request.httprequest.url_root + 'auth_oauth/signin'
state = self.get_state(provider)
@@ -50,7 +47,7 @@ class OAuthLoginWechat(OAuthLogin):
redirect_uri=return_url,
state=json.dumps(state),
)
if wechat_url.find('authorize') >= 0: # 企业微信内直接登录
if wechat_url.find('authorize') >= 0: # 企业微信内直接登录
params.update({
'response_type': 'code',
'scope': 'snsapi_base',
@@ -74,98 +71,93 @@ class ResUsers(models.Model):
@api.model
def auth_oauth(self, provider, params):
oauth_provider = self.env['auth.oauth.provider'].browse(provider)
if oauth_provider.auth_endpoint.find('weixin') >= 0: #微信认证
if oauth_provider.auth_endpoint.find('weixin') >= 0: # 微信认证
return self.auth_oauth_wechat(oauth_provider, params)
else:
return super(ResUsers, self).auth_oauth(provider, params)
@api.model
def auth_oauth_wechat(self, provider, params):
#微信获取访问用户身份的URL https://qyapi.weixin.qq.com/cgi-bin/user/getuserinfo?access_token=ACCESS_TOKEN&code=CODE
#微信获取access_token的URL https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=ID&corpsecret=SECRET
# 微信获取访问用户身份的URL https://qyapi.weixin.qq.com/cgi-bin/user/getuserinfo?access_token=ACCESS_TOKEN&code=CODE
# 微信获取access_token的URL https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=ID&corpsecret=SECRET
#
#
def gettoken(url, corpid, corpsecret):
url_token = "%s?corpid=%s&corpsecret=%s" % (url, corpid, corpsecret)
headers={'Content-Type': 'application/json'}
response = requests.get(url_token, headers = headers)
headers = {'Content-Type': 'application/json'}
response = requests.get(url_token, headers=headers)
dict_token = response.json()
if dict_token["errcode"] == 0:
return dict_token["access_token"]
else:
raise OAuthLogin("微信获取access_token错误：err_code=%s, err_msg=%s" % (dict_token["errcode"], dict_token["errmsg"] ) )
raise OAuthLogin(
"微信获取access_token错误：err_code=%s, err_msg=%s" % (dict_token["errcode"], dict_token["errmsg"]))
client_ids = provider.client_id.split(' ')
if not client_ids or len(client_ids) != 3:
raise OAuthLogin("请正确设置微信登录验证的corpid agentid corpsecret！" )
raise OAuthLogin("请正确设置微信登录验证的corpid agentid corpsecret！")
corpid = client_ids[0]
corpsecret = client_ids[2]
# 使用内存缓存 access_token，完全避免数据库写入，防止触发 registry signaling 导致 session 过期
import time as _time
access_token = _wechat_token_cache.get('access_token')
if not access_token or _time.time() > _wechat_token_cache.get('expire_time', 0):
access_token = gettoken(provider.validation_endpoint, corpid, corpsecret)
_wechat_token_cache['access_token'] = access_token
_wechat_token_cache['expire_time'] = _time.time() + 7000 # 微信 token 有效期 7200s，提前 200s 刷新
access_token = provider.scope
code = params.get('code', False)
if not access_token:
access_token = gettoken(provider.validation_endpoint, corpid, corpsecret)
if not code:
raise OAuthLogin("微信扫码错误：没有 code！" )
raise OAuthLogin("微信扫码错误：没有 code！")
provider.scope = access_token
oauth_uid = False
headers={'Content-Type': 'application/json'}
headers = {'Content-Type': 'application/json'}
user_url = "%s?access_token=%s&code=%s" % (provider.data_endpoint, access_token, code)
response = requests.get(user_url, headers = headers)
response = requests.get(user_url, headers=headers)
dict_user = response.json()
errcode = dict_user.get("errcode", 0)
if errcode in [42001]:
access_token = gettoken(provider.validation_endpoint, corpid, corpsecret)
_wechat_token_cache['access_token'] = access_token
_wechat_token_cache['expire_time'] = _time.time() + 7000
user_url = "%s?access_token=%s&code=%s" % (provider.data_endpoint, access_token, code)
response = requests.get(user_url, headers=headers)
dict_user = response.json()
_logger.warning("------------------"+str(dict_user))
_logger.warning("------------------" + str(dict_user))
if dict_user["errcode"] == 0:
oauth_uid = dict_user["UserId"]
elif dict_user["errcode"] == 40014 or dict_user["errcode"] == 42001: # 40014: 不合法的access_token, 42001: access_token expired
elif dict_user["errcode"] == 40014 or dict_user[
"errcode"] == 42001: # 40014: 不合法的access_token, 42001: access_token expired
access_token = gettoken(provider.validation_endpoint, corpid, corpsecret)
_wechat_token_cache['access_token'] = access_token
_wechat_token_cache['expire_time'] = _time.time() + 7000
provider.scope = access_token
user_url = "%s?access_token=%s&code=%s" % (provider.data_endpoint, access_token, code)
response = requests.get(user_url, headers = headers)
response = requests.get(user_url, headers=headers)
dict_user = response.json()
_logger.warning("------------------"+str(dict_user))
_logger.warning("------------------" + str(dict_user))
oauth_uid = dict_user["UserId"]
else:
raise OAuthLogin("微信获取访问用户身份错误：err_code=%s, err_msg=%s" % (dict_user["errcode"], dict_user["errmsg"] ) )
raise WeChatOAuthException(dict_user["errcode"], "微信获取访问用户身份错误：err_code=%s, err_msg=%s" % (dict_user["errcode"], dict_user["errmsg"]))
# raise OAuthLogin(
# "微信获取访问用户身份错误：err_code=%s, err_msg=%s" % (dict_user["errcode"], dict_user["errmsg"]))
user_id = self.search([('oauth_uid', '=', oauth_uid), ('oauth_uid', '!=', False)], limit=1)
if not user_id:
if self.env.context.get('wechat_create_user', False): # 自动创建企业微信User
if self.env.context.get('wechat_create_user', False): # 自动创建企业微信User
user_id = self.create_from_wechat(provider, access_token, oauth_uid, code)
if not user_id:
_logger.exception("OAuth Wechat: 微信访问用户【UserId：%s】没有关联Odoo账号，且自动创建失败！" % (oauth_uid, ))
raise AccessDenied("微信访问用户【UserId：%s】没有关联系统账号！" % (oauth_uid, ))
_logger.exception(
"OAuth Wechat: 微信访问用户【UserId：%s】没有关联Odoo账号，且自动创建失败！" % (oauth_uid,))
raise AccessDenied("微信访问用户【UserId：%s】没有关联系统账号！" % (oauth_uid,))
else:
_logger.exception("OAuth Wechat: 微信访问用户【UserId：%s】没有关联Odoo账号！" % (oauth_uid, ))
raise AccessDenied("微信访问用户【UserId：%s】没有关联系统账号！" % (oauth_uid, ))
_logger.exception("OAuth Wechat: 微信访问用户【UserId：%s】没有关联Odoo账号！" % (oauth_uid,))
raise AccessDenied("微信访问用户【UserId：%s】没有关联系统账号！" % (oauth_uid,))
else:
# 不再写入 oauth_access_token，避免触发 res.users 写入导致 registry cache invalidation
# 从而挤掉其他用户的 session
pass
user_id.oauth_access_token = code
return (self.env.cr.dbname, user_id.login, code)
@api.model
def create_from_wechat(self, provider, access_token, userid, code):
#读取企业通讯录的成员详情: https://qyapi.weixin.qq.com/cgi-bin/user/get?access_token=ACCESS_TOKEN&userid=USERID
headers={'Content-Type': 'application/json'}
# 读取企业通讯录的成员详情: https://qyapi.weixin.qq.com/cgi-bin/user/get?access_token=ACCESS_TOKEN&userid=USERID
headers = {'Content-Type': 'application/json'}
user_url = "https://qyapi.weixin.qq.com/cgi-bin/user/get"
user_url = "%s?access_token=%s&userid=%s" % (user_url, access_token, userid)
response = requests.get(user_url, headers = headers)
response = requests.get(user_url, headers=headers)
dict_user = response.json()
if dict_user["errcode"] == 0:
mobile = dict_user.get("mobile", '')
@@ -181,7 +173,7 @@ class ResUsers(models.Model):
'oauth_uid': userid,
'oauth_provider_id': provider.id,
'oauth_access_token': code,
'active': True,
'active': True,
}
template_user = self.env.ref("base.default_user")
try:
@@ -192,6 +184,6 @@ class ResUsers(models.Model):
_logger.exception("OAuth Wechat: 微信访问用户【UserId：%s】自动创建失败【Odoo错误：%s】！" % (userid, ustr(e)))
raise OAuthLogin(ustr(e))
else:
_logger.exception("OAuth Wechat: 微信访问用户【UserId：%s】自动创建失败【微信错误码：%s】！" % (userid, dict_user["errcode"]))
_logger.exception(
"OAuth Wechat: 微信访问用户【UserId：%s】自动创建失败【微信错误码：%s】！" % (userid, dict_user["errcode"]))
return False
yuthon_wechat/models/res_users副本.py
+199
View File
@@ -0,0 +1,199 @@
# -*- coding: utf-8 -*-
import werkzeug.urls
from urllib.parse import urlparse
import urllib.request
import requests
import json
from odoo import api, fields, models
from odoo.tools.misc import ustr
from odoo.exceptions import AccessDenied
from odoo.http import request
from odoo.addons.auth_signup.models.res_users import SignupError
from odoo.addons.auth_oauth.controllers.main import OAuthLogin
import logging
from ..utils.common import WeChatOAuthException
_logger = logging.getLogger(__name__)
# 模块级内存缓存，存储企业微信 access_token，避免写入数据库触发 registry signaling
_wechat_token_cache = {
'access_token': None,
'expire_time': 0,
}
class OAuthLoginWechat(OAuthLogin):
def list_providers(self):
providers = super(OAuthLoginWechat, self).list_providers()
for provider in providers:
if provider['auth_endpoint'].find('weixin') >= 0:
#对微信的回调链接做特殊处理
#企业微信扫码登录链接 https://open.work.weixin.qq.com/wwopen/sso/qrConnect?appid=CORPID&agentid=AGENTID&redirect_uri=REDIRECT_URI&state=STATE
#企业微信内直接登录链接 https://open.weixin.qq.com/connect/oauth2/authorize?appid=CORPID&redirect_uri=REDIRECT_URI&response_type=code&scope=snsapi_base&state=STATE#wechat_redirect
# 如果auth_endpoint链接中含有 “?create_user” 字样，则自动创建用户
wechat_url = provider['auth_endpoint']
create_user = False
auth_endpoint = wechat_url.split('?')
if len(auth_endpoint) >= 2:
wechat_url = auth_endpoint[0]
create_user = auth_endpoint[1].find('create_user') >= 0
client_ids = provider['client_id'].split(' ')
return_url = request.httprequest.url_root + 'auth_oauth/signin'
state = self.get_state(provider)
if create_user:
state['c'] = {'wechat_create_user': True}
params = dict(
appid=client_ids[0].strip(),
agentid=client_ids[1].strip(),
redirect_uri=return_url,
state=json.dumps(state),
)
if wechat_url.find('authorize') >= 0: # 企业微信内直接登录
params.update({
'response_type': 'code',
'scope': 'snsapi_base',
})
provider['auth_link'] = "%s?%s#wechat_redirect" % (wechat_url, werkzeug.urls.url_encode(params))
else:
provider['auth_link'] = "%s?%s" % (wechat_url, werkzeug.urls.url_encode(params))
return providers
class ResUsers(models.Model):
_inherit = 'res.users'
wecom_fields = fields.Char(string="免登录验证字段", compute='_compute_wecom_fields', stroe=True)
@api.depends('login')
def _compute_wecom_fields(self):
for i in self:
i.wecom_fields = i.login
@api.model
def auth_oauth(self, provider, params):
oauth_provider = self.env['auth.oauth.provider'].browse(provider)
if oauth_provider.auth_endpoint.find('weixin') >= 0: #微信认证
return self.auth_oauth_wechat(oauth_provider, params)
else:
return super(ResUsers, self).auth_oauth(provider, params)
@api.model
def auth_oauth_wechat(self, provider, params):
#微信获取访问用户身份的URL https://qyapi.weixin.qq.com/cgi-bin/user/getuserinfo?access_token=ACCESS_TOKEN&code=CODE
#微信获取access_token的URL https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=ID&corpsecret=SECRET
#
#
def gettoken(url, corpid, corpsecret):
url_token = "%s?corpid=%s&corpsecret=%s" % (url, corpid, corpsecret)
headers={'Content-Type': 'application/json'}
response = requests.get(url_token, headers = headers)
dict_token = response.json()
if dict_token["errcode"] == 0:
return dict_token["access_token"]
else:
raise OAuthLogin("微信获取access_token错误：err_code=%s, err_msg=%s" % (dict_token["errcode"], dict_token["errmsg"] ) )
client_ids = provider.client_id.split(' ')
if not client_ids or len(client_ids) != 3:
raise OAuthLogin("请正确设置微信登录验证的corpid agentid corpsecret！" )
corpid = client_ids[0]
corpsecret = client_ids[2]
# 使用内存缓存 access_token，完全避免数据库写入，防止触发 registry signaling 导致 session 过期
import time as _time
access_token = _wechat_token_cache.get('access_token')
if not access_token or _time.time() > _wechat_token_cache.get('expire_time', 0):
access_token = gettoken(provider.validation_endpoint, corpid, corpsecret)
_wechat_token_cache['access_token'] = access_token
_wechat_token_cache['expire_time'] = _time.time() + 7000 # 微信 token 有效期 7200s，提前 200s 刷新
code = params.get('code', False)
if not code:
raise OAuthLogin("微信扫码错误：没有 code！" )
oauth_uid = False
headers={'Content-Type': 'application/json'}
user_url = "%s?access_token=%s&code=%s" % (provider.data_endpoint, access_token, code)
response = requests.get(user_url, headers = headers)
dict_user = response.json()
errcode = dict_user.get("errcode", 0)
if errcode in [42001]:
access_token = gettoken(provider.validation_endpoint, corpid, corpsecret)
_wechat_token_cache['access_token'] = access_token
_wechat_token_cache['expire_time'] = _time.time() + 7000
user_url = "%s?access_token=%s&code=%s" % (provider.data_endpoint, access_token, code)
response = requests.get(user_url, headers=headers)
dict_user = response.json()
_logger.warning("------------------"+str(dict_user))
if dict_user["errcode"] == 0:
oauth_uid = dict_user["UserId"]
elif dict_user["errcode"] == 40014 or dict_user["errcode"] == 42001: # 40014: 不合法的access_token, 42001: access_token expired
access_token = gettoken(provider.validation_endpoint, corpid, corpsecret)
_wechat_token_cache['access_token'] = access_token
_wechat_token_cache['expire_time'] = _time.time() + 7000
user_url = "%s?access_token=%s&code=%s" % (provider.data_endpoint, access_token, code)
response = requests.get(user_url, headers = headers)
dict_user = response.json()
_logger.warning("------------------"+str(dict_user))
oauth_uid = dict_user["UserId"]
else:
raise WeChatOAuthException(dict_user["errcode"], "微信获取访问用户身份错误：err_code=%s, err_msg=%s" % (dict_user["errcode"], dict_user["errmsg"]))
user_id = self.search([('oauth_uid', '=', oauth_uid), ('oauth_uid', '!=', False)], limit=1)
if not user_id:
if self.env.context.get('wechat_create_user', False): # 自动创建企业微信User
user_id = self.create_from_wechat(provider, access_token, oauth_uid, code)
if not user_id:
_logger.exception("OAuth Wechat: 微信访问用户【UserId：%s】没有关联Odoo账号，且自动创建失败！" % (oauth_uid, ))
raise AccessDenied("微信访问用户【UserId：%s】没有关联系统账号！" % (oauth_uid, ))
else:
_logger.exception("OAuth Wechat: 微信访问用户【UserId：%s】没有关联Odoo账号！" % (oauth_uid, ))
raise AccessDenied("微信访问用户【UserId：%s】没有关联系统账号！" % (oauth_uid, ))
else:
# 不再写入 oauth_access_token，避免触发 res.users 写入导致 registry cache invalidation
# 从而挤掉其他用户的 session
pass
return (self.env.cr.dbname, user_id.login, code)
@api.model
def create_from_wechat(self, provider, access_token, userid, code):
#读取企业通讯录的成员详情: https://qyapi.weixin.qq.com/cgi-bin/user/get?access_token=ACCESS_TOKEN&userid=USERID
headers={'Content-Type': 'application/json'}
user_url = "https://qyapi.weixin.qq.com/cgi-bin/user/get"
user_url = "%s?access_token=%s&userid=%s" % (user_url, access_token, userid)
response = requests.get(user_url, headers = headers)
dict_user = response.json()
if dict_user["errcode"] == 0:
mobile = dict_user.get("mobile", '')
name = dict_user.get("name", '')
email = dict_user.get("email", '')
telephone = dict_user.get("telephone", '')
values = {
'login': email or mobile or userid,
'name': name or userid,
'mobile': mobile,
'email': email,
'phone': telephone,
'oauth_uid': userid,
'oauth_provider_id': provider.id,
'oauth_access_token': code,
'active': True,
}
template_user = self.env.ref("base.default_user")
try:
with self.env.cr.savepoint():
user_id = template_user.sudo().with_context(no_reset_password=True).copy(values)
return user_id
except Exception as e:
_logger.exception("OAuth Wechat: 微信访问用户【UserId：%s】自动创建失败【Odoo错误：%s】！" % (userid, ustr(e)))
raise OAuthLogin(ustr(e))
else:
_logger.exception("OAuth Wechat: 微信访问用户【UserId：%s】自动创建失败【微信错误码：%s】！" % (userid, dict_user["errcode"]))
return False
yuthon_wechat/utils/common.py
+7
View File
@@ -29,3 +29,10 @@ def os_path_join(path, filename):
image_path = os.path.join(str(image_path), str(path))
image = os.path.join(image_path, filename)
return image
class WeChatOAuthException(Exception):
def __init__(self, errcode, errmsg):
self.errcode = errcode
self.errmsg = errmsg
super().__init__(f"[{errcode}] {errmsg}")