使用python中tornado框架实现的微信access_token中控服务器

wechat-token-server

微信token中控服务器，用于统一获取并缓存微信开发中使用的access_token和jsticket。 wechat-token-server是一个自动定时刷新微信token的服务，可以每隔一段时间自动获取token，保存在redis中，通过访问redis或web接口即可获取到缓存的token值。同时用户也可以主动刷新。 github源码地址：https://github.com/shangyexin/wechat-token-server

主要实现代码

使用tornado异步http client定时刷新token

异步请求：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
async def refreshToken(tokenType):
    # 异步客户端
    asyncHttpClient = tornado.httpclient.AsyncHTTPClient()
    request = renderRequest(tokenType)
    try:
        response = await asyncHttpClient.fetch(request)

    except Exception as e:
        logger.error('Exception: %s', e)
        logger.error('Create a request for %s failed, will retry after 10s' % str(tokenType))
        # 请求失败后10s后再次尝试刷新
        tornado.ioloop.IOLoop.instance().call_later(10, refreshToken, tokenType)
    else:
        logger.debug('Async httpclient response: %s', response)
        logger.debug('Async httpclient response body: %s', response.body)
        # response body里面json格式的字典
        resDict = json.loads(response.body.decode('utf8'))
        # 获取access_token成功时没有errcode，获取ticket成功时errcode等于0
        errcodeExist = 'errcode' in resDict
        if errcodeExist == False or resDict['errcode'] == 0:
            logger.info('Request %s success, response is :%s' % (tokenType, resDict))
            token = resDict[tokenType]
            # logger.info('Token value is %s', token)
            try:
                # 存入redis数据库并设置过期时间
                wechatRedis.set(tokenType, token, ex=config.tokenExpireTime)
                logger.info('Set %s in redis success.', tokenType)
            except Exception as e:
                logger.error(e)
        else:
            logger.error('Request for %s failed, error message is:' % str(tokenType))
            logger.error(resDict['errmsg'])
            # 10s后再次尝试刷新
            tornado.ioloop.IOLoop.instance().call_later(10, refreshToken, tokenType)

定时刷新：

1
2
3
4
5
6
7
8
# 依次刷新tokenSources中的token
async def refreshAllTokens():
    logger.info('Begin to refresh all tokens...')
    while True:
        for tokenType in config.tokenSources.keys():
            await refreshToken(tokenType)
        # 定时更新
        await tornado.gen.sleep(config.tokenExpireTime)

开始：

1
2
# 自动刷新token
tornado.ioloop.IOLoop.instance().spawn_callback(refreshAllTokens)
使用tornado同步http client强制刷新
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
# 强制刷新token,适用于应用程序判断token已过期后主动刷新
class ForceRefreshHandler(tornado.web.RequestHandler):
    def get(self):
        tokenType = self.get_argument('type', 'UnKnown')
        # 获取token的口令
        secret = self.get_argument('secret', 'UnKnown')
        ret = {}
        if tokenType != 'UnKnown' and secret == config.requestSecret:
            request = renderRequest(tokenType)
            # 同步客戶端
            syncHttpClient = tornado.httpclient.HTTPClient()
            try:
                response = syncHttpClient.fetch(request)
            except Exception as e:
                logger.error('Exception: %s', e)
                logger.error('Force refresh %s failed.' % str(tokenType))
                ret['errmsg'] = str(e)
            else:
                logger.debug('Sync httpclient response: %s', response)
                logger.debug('Sync httpclient response body: %s', response.body)
                # response body里面json格式的字典
                resDict = json.loads(response.body.decode('utf8'))
                logger.info('Force refresh %s success, response is :%s' % (tokenType, resDict))
                token = resDict[tokenType]
                try:
                    # 存入redis数据库并设置过期时间
                    wechatRedis.set(tokenType, token, ex=config.tokenExpireTime)
                    logger.info('Set %s in redis success.', tokenType)
                except Exception as e:
                    logger.error(e)
                # 返回请求
                ret[tokenType] = resDict[tokenType]
                ret['expires_in'] = config.tokenExpireTime
            syncHttpClient.close()

        self.write(json.dumps(ret, sort_keys=False))

实现功能（微信官方建议）

  • 建议公众号开发者使用中控服务器统一获取和刷新Access_token，其他业务逻辑服务器所使用的access_token均来自于该中控服务器，不应该各自去刷新，否则容易造成冲突，导致access_token覆盖而影响业务；

  • 目前Access_token的有效期通过返回的expire_in来传达，目前是7200秒之内的值。中控服务器需要根据这个有效时间提前去刷新新access_token。在刷新过程中，中控服务器可对外继续输出的老access_token，此时公众平台后台会保证在5分钟内，新老access_token都可用，这保证了第三方业务的平滑过渡；

  • Access_token的有效时间可能会在未来有调整，所以中控服务器不仅需要内部定时主动刷新，还需要提供被动刷新access_token的接口，这样便于业务服务器在API调用获知access_token已超时的情况下，可以触发access_token的刷新流程。

配置

修改config.py即可，下面是各配置选项含义。 tokenExpireTime: 中控服务器自己缓存的token过期时间
bindIp:tornado服务器绑定的IP
bindPort: 绑定端口
redisIp:redis服务器运行IP
redisPort: redis服务器运行端口
requestSecret: 获取token的口令
appid:公众号或小程序应用ID
secret: 应用秘钥 示例代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
logging.config.fileConfig("logger.conf")
logger = logging.getLogger("wechat-access-token-server")

tokenSources = {}
tokenExpireTime = 7000

bindIp = '0.0.0.0'
bindPort = 12123

redisIp = '127.0.0.1'
redisPort = 6379

# 获取token的口令
requestSecret = '*******'

# 微信access_token配置示例
tokenSources['access_token'] = {
    'url': 'https://api.weixin.qq.com/cgi-bin/token',
    'method': 'GET',
    'args': {
        'grant_type': 'client_credential',
        'appid': '**************',
        'secret': '**********************'
    },
}

# 微信jsticket配置示例
tokenSources['ticket'] = {
    'url': 'https://api.weixin.qq.com/cgi-bin/ticket/getticket',
    'method': 'GET',
    'args': {
        'type': 'jsapi',
        'access_token': '{{access_token}}'
    },
}

运行与使用

环境与依赖项
  • Ubuntu16.04
  • Python 3.6
  • tornado==5.1.1
  • redis==3.0.1
后台运行

nohup python3 main.py &

通过redis查询

如果查询程序与wechat token sever运行在同一台机器上，可以直接查询redis服务器，键值就是access_token和ticket。

通过url访问

直接访问对应的url即可，secret为自己在config.py里面设置的值

  • 查询access_token : http://127.0.0.1:12123/wechat/token?type=access_token&secret=f3b2241f967aa3c7966f537cdd82ce11
  • 查询jsticket : http://127.0.0.1:12123/wechat/token?type=ticket&secret=f3b2241f967aa3c7966f537cdd82ce11
  • 主动刷新access_token : http://127.0.0.1:12123/wechat/token/forcerefresh?type=access_token&secret=f3b2241f967aa3c7966f537cdd82ce11
  • 主动刷新jsticket : http://127.0.0.1:12123/wechat/token/forcerefresh?type=ticket&secret=f3b2241f967aa3c7966f537cdd82ce11
结果示例

查询access_token结果： {"access_token": "16_XLwKyOP2XSxwr1ZCB0tpaANb4l2cJSeJdr6aj0QQkQAp_v4q5E-eCIkmDyOcMU6V0aTSsXiJzyf-KwiVP0MHIv47XftDa_oSvCnH8jJfTz8POfBjTcl52jGBboPMoo-esVjCVW2Ll3D2_6RHAGEiAJAGMK", "expires_in": 3298}

反向代理设置

如果希望通过域名访问，可以使用Nginx设置反向代理，将你需要访问域名配置好后，访问指定的url直接转发到tornado运行的地址与端口即可。
配置示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
server {
    listen 443;
    server_name gitlab.net.cn;
    ssl on;
    root /home/yasin/html/domain/gitlab_net_cn;
    index index.html index.htm;
    ssl_certificate  /home/yasin/ssl/gitlab/gitlab.net.cn.crt;
    ssl_certificate_key /home/yasin/ssl/gitlab/gitlab.net.cn.key;
    ssl_session_timeout 5m;
    ssl_ciphers ECDHE-RSA-AES128-GCM-SHA256:ECDHE:ECDH:AES:HIGH:!NULL:!aNULL:!MD5:!ADH:!RC4;
    ssl_protocols TLSv1 TLSv1.1 TLSv1.2;
    ssl_prefer_server_ciphers on;
    location / {
        proxy_pass http://127.0.0.1:12123;
    }
}

这样配置好后，在其他任何地方直接访问https://gitlab.net.cn/wechat/token?type=access_token&secret=f3b2241f967aa3c7966f537cdd82ce11即可得到access_token值。

Tornado
#python #wechat #tornado
使用python中tornado框架实现的微信access_token中控服务器
https://www.shangyexin.com/2018/11/27/wechat-token-server/
作者
Yasin
发布于
2018年11月27日
许可协议