想写个不需要登录，无需维护Session或Cookie等状态的纯RESTful API，但是又需要有身份认证和权限鉴别功能。比如我有一套爬来的金融数据API，想让指定的用户使用，其他人访问就报401。最近用FastAPI，里面提供了OAuth2以及scopes的密码式令牌鉴权机制，十分方便，但是部分场景也容易造成信息泄露。

IMPORTANT
OAuth2的密码式令牌应在可信任网络中使用。

OAuth2密码式令牌应用场景#

众所周知OAuth2有四种应用场景各不相同的鉴权方式，其中，密码式可以说是在互联网环境中最不安全的一种了，因为你需要将你在诸如微信的用户名密码一类的信息告诉一个第三方应用，而后第三方应用拿着你的用户名密码去向微信申请数据。如果是在互联网上使用这种验证方式，你必须极度信任这个第三方应用，不然就是社会性自杀。

那么，这样一个危险的鉴权方式有什么实际作用呢？那当然是可控网络环境中，对各业务系统API进行集中访问控制了——毕竟现在微服务这么火，就算是内网应用也应该试试开放几个API吐出点数据——这样PPT上就可以写打破数据孤岛了✌️。

比如，我有很多业务系统，以系统A为例。a.com开放了两个API：

查看当前用户的信息http://a.com/me；
查看当前用户拥有的数据条目http://a.com/me/items。

这时候有人想调用这俩API，而A系统管理员只希望指定的人用，此外，还有XYZ的各种业务系统，都有这种需求，如何在不使用session、cookie这种有状态身份认证的情况下（调用API还要先认证个session太麻烦了）直接在请求中附带一个通行证就调用API，还要保证安全和权限分级呢？

OAuth2密码式令牌应用环境#

密码式令牌提供了一个简单的解决方案：

在A系统上开一个发通行证access_token的窗口：http://a.com/token。

系统中有个加解密的秘钥SECRET_KEY；
通行证窗口是一个可以被A系统全信任的窗口，不私存用户名密码，就是个工具窗口。
通行证窗口接受用户认证信息，返回一个jwt库使用秘钥SECRET_KEY加密的通行证。
通行证窗口接受方式规定为HTTP POST方式。接受如下的表单，其中scope为表示权限的字符串，以空格为分隔符：
```
1
grant_type: password
2
scope: me items
3
username: johndoe
4
password: secret
```
通行证窗口拿到用户名密码后确认身份，并取出库中的权限列表与请求中的权限列表进行核对，如果请求的权限多于在册的权限，则http-401，反之则放行。

通行证窗口返回的通行证为json格式，通行证的内容包含认证用户ID、权限列表、通行证有效期，例如

1
{"access_token":"eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJqb2huZG9lIiwic2NvcGVzIjpbIm1lIiwiaXRlbXMiXSwiZXhwIjoxNTk0MTI2MjkxfQ.AimqzTZg1t7XHstWi-048prkRAPv6-m-IM9OfGYio8A","token_type":"bearer"}
2
// 其中access_token解密后的明文形为
3
{'sub': 'johndoe', 'scopes': ['me', 'items'], 'exp': 1593831859}

A系统提供了很多API：

查看当前用户的信息http://a.com/me；
查看当前用户拥有的数据条目http://a.com/me/items。
…
系统中有个加解密的秘钥SECRET_KEY，与发通行证的窗口用的一样；
这些API都有身份验证机制，具有独立的权限列表要求，需要提供OAuth2通行证令牌，jwt库使用秘钥SECRET_KEY对令牌解密，以验明正身。

A系统管理员维护着一个用户及权限表，包括用户名username（如johndoe）、密码hashed_password、权限列表scope（如['b/me', 'b/items']）。

OAuth2密码式令牌调用流程#

此时，第三方系统B想要调用业务系统A的API，比如有人想要调用http://a.com/me，那么A系统如何认证用户和权限呢？

B需要获取A库表中的某个用户名密码：

这个动作可以是A管理员给B的，比如我A系统就开一个API专用账户，你B拿着这个账号就随便临幸我A吧。（这种通常权限开的尺度比较大）
也可以是A的一个用户给B的，用户aaa表示你B系统可以去A系统拿我的某些数据。

B拿着用户名密码去A的通行证窗口申请通行证：

通行证窗口就是负责接收用户名密码，确认在A的库表中后，发通行证。
查无此人直接http-401。

B拿到通行证，在每一次对A系统的API请求Header中附带这个通行证：

例如，在访问a.com/me时加上Authorization头部字段，其值形为Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJqb2huZG9lIiwic2NvcGVzIjpbIm1lIiwiaXRlbXMiXSwiZXhwIjoxNTk0MTI2MjkxfQ.AimqzTZg1t7XHstWi-048prkRAPv6-m-IM9OfGYio8A。

A的API会拿到Authorization头部字段，取出其中的通行证，解密，获取其中的用户名、权限、有效期：

若令牌可解密，且在有效期内，比对API所需的权限与请求通行证提供的权限，若提供满足所需，则API响应并返回，反之则http-401。

OAuth2密码式令牌代码示例#

1
from datetime import datetime, timedelta
2

3
import jwt
4
from jwt import PyJWTError
5
from passlib.context import CryptContext
6

7
from fastapi import Depends, FastAPI, HTTPException, Security, status
8
from fastapi.security import (
9
    OAuth2PasswordBearer,
10
    OAuth2PasswordRequestForm,
11
    SecurityScopes,
12
)
13

14
from typing import Optional, List
15
from pydantic import BaseModel, ValidationError
16

17
# to get a string like this run:
18
# openssl rand -hex 32
19
SECRET_KEY = "4ff135e7c8c7375a00738b8f02cc016a7983ea345fe35735900495bea0cf9ce9"
20
ALGORITHM = "HS256"
21
ACCESS_TOKEN_EXPIRE_MINUTES = 30
22

23

24
app = FastAPI()
25
VERSION = '1.0'
26

27
fake_users_db = {
28
    "johndoe": {
29
        "username": "johndoe",
30
        "full_name": "John Doe",
31
        "email": "johndoe@example.com",
32
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
33
        "disabled": False,
34
    },
35
    "alice": {
36
        "username": "alice",
37
        "full_name": "Alice Chains",
38
        "email": "alicechains@example.com",
39
        "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
40
        "disabled": True,
41
    },
42
}
43

44
class Token(BaseModel):
45
    access_token: str
46
    token_type: str
47

48

49
class TokenData(BaseModel):
50
    username: str = None
51
    scopes: List[str] = []
52

53

54
class User(BaseModel):
55
    username: str
56
    email: Optional[str] = None
57
    full_name: Optional[str] = None
58
    disabled: Optional[bool] = None
59

60

61
class UserInDB(User):
62
    hashed_password: str
63

64

65
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
66
oauth2_scheme = OAuth2PasswordBearer(
67
    tokenUrl="/token",
68
    scopes={
69
        "me": "Read information about the current user.",
70
        "items": "Read items."
71
    },
72
)
73

74

75
def verify_password(plain_password, hashed_password):
76
    return pwd_context.verify(plain_password, hashed_password)
77

78

79
def get_user(db, username: str):
80
    if username in db:
81
        user_dict = db[username]
82
        return UserInDB(**user_dict)
83

84

85
def authenticate_user(fake_db, username: str, password: str):
86
    user = get_user(fake_db, username)
87
    if not user:
88
        return False
89
    if not verify_password(password, user.hashed_password):
90
        return False
91
    return user
92

93

94
def create_access_token(data: dict, expires_delta: timedelta = None):
95
    to_encode = data.copy()
96
    if expires_delta:
97
        expire = datetime.utcnow() + expires_delta
98
    else:
99
        expire = datetime.utcnow() + timedelta(minutes=15)
100
    to_encode.update({"exp": expire})
101
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
102
    return encoded_jwt
103

104

105
async def get_current_user(security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)):
106
    # Request Headers里添加了: Authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJqb2huZG9lIiwic2NvcGVzIjpbIm1lIiwiaXRlbXMiXSwiZXhwIjoxNTkzODMyMTM2fQ.Ulq6S2zg0kETKdS_3NhcFho9M4-JhzHXcXWPZQJm0Cc
107
    # curl -X GET "http://localhost:8910/users/me/" -H "accept: application/json" -H "Authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJqb2huZG9lIiwic2NvcGVzIjpbIm1lIiwiaXRlbXMiXSwiZXhwIjoxNTkzODMyMTM2fQ.Ulq6S2zg0kETKdS_3NhcFho9M4-JhzHXcXWPZQJm0Cc"
108
    if security_scopes.scopes:
109
        authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
110
    else:
111
        authenticate_value = f"Bearer"
112
    credentials_exception = HTTPException(
113
        status_code=status.HTTP_401_UNAUTHORIZED,
114
        detail="Could not validate credentials",
115
        headers={"WWW-Authenticate": authenticate_value},
116
    )
117
    try:
118
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) # payload解密为 {'sub': 'johndoe', 'scopes': ['me', 'items'], 'exp': 1593831859}
119
        username: str = payload.get("sub")
120
        if username is None:
121
            raise credentials_exception
122
        token_scopes = payload.get("scopes", [])
123
        token_data = TokenData(scopes=token_scopes, username=username)
124
    except (PyJWTError, ValidationError):
125
        raise credentials_exception
126
    user = get_user(fake_users_db, username=token_data.username)
127
    if user is None:
128
        raise credentials_exception
129
    for scope in security_scopes.scopes: # security_scopes.scopes为当前请求所要求的scopes权限，如/me为['me']，/me/items为['items', 'me']
130
        if scope not in token_data.scopes: # token_data.scopes为当前请求的token中持有的scopes权限，如果所需的scopes中有任何一条不在请求持有的的scopes时（即少持有任一条），均禁止访问
131
            raise HTTPException(
132
                status_code=status.HTTP_401_UNAUTHORIZED,
133
                detail="Not enough permissions",
134
                headers={"WWW-Authenticate": authenticate_value},
135
            )
136
    return user
137

138

139
async def get_current_active_user(current_user: User = Security(get_current_user, scopes=["me"])):
140
    if current_user.disabled:
141
        raise HTTPException(status_code=400, detail="Inactive user")
142
    return current_user
143

144

145
@app.post("/token", response_model=Token)
146
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
147
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
148
    if not user:
149
        raise HTTPException(status_code=400, detail="Incorrect username or password")
150
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
151
    access_token = create_access_token(
152
        data={"sub": user.username, "scopes": form_data.scopes},
153
        expires_delta=access_token_expires,
154
    )
155
    return {"access_token": access_token, "token_type": "bearer"}
156

157

158
@app.get("/users/me/", response_model=User)
159
async def read_users_me(current_user: User = Depends(get_current_active_user)):
160
    return current_user
161

162

163
@app.get("/users/me/items/")
164
async def read_own_items(current_user: User = Security(get_current_active_user, scopes=["items"])):
165
    return [{"item_id": "Foo", "owner": current_user.username}]
166

167

168
@app.get("/status/")
169
async def read_system_status(current_user: User = Depends(get_current_user)):
170
    return {"status": "ok"}

警告
在152行中，示例为了方便直接给令牌赋予了请求者索要的权限，在真实情况下，此处应该将请求索要的权限与数据库中记录的权限进行比对，若索要权限大于记录权限，则应产生“未授权”的异常。

总结#

在网络环境较为透明可控的场景下，使用OAuth2的密码方式对开放的API进行身份及权限验证，是一种安全、可控、简便的管理方法。除上文描述的方法外，还可以将各应用系统的权限收口统一管理，维护一个用户权限库表，将通行证的发放维护在一个入口中，再于各应用系统中约定同一个加解密秘钥，可进一步方便管理。当然，这些方法的前提必须是网络环境较为透明可控。