| 1234567891011121314151617181920212223242526 |
- #!/usr/bin/env python
- # -*- coding: utf-8 -*-
- from starlette.authentication import AuthenticationBackend, SimpleUser, AuthCredentials
- from db.asyncredis import redis_client
- from utils.depends import check_access_token
- class JwtAuth(AuthenticationBackend):
- async def authenticate(self, conn):
- if "Authorization" not in conn.headers:
- return None
- token = conn.headers["Authorization"].split(" ")[-1]
- token_payload = check_access_token(token)
- # 检验token是否过期？
- token_key = "{}:{}:access-token".format(
- token_payload.sub.get("school", None), token_payload.sub["sub"])
- access_token = redis_client.get(token_key)
- if not access_token:
- return None
- return AuthCredentials(["authenticated"]), SimpleUser(token_payload.sub)
|
