ChatWork OAuth2换取access token时出现401错误，请求协助排查

开发一款向ChatWork发送消息的应用，已完成登录授权并获取到authorization code，但调用token接口换取access token时收到401错误。以下是实现代码：

import requests
from urllib.parse import urlencode, urlparse, parse_qs
import hashlib
import base64
import secrets

# Your ChatWork OAuth 2.0 client ID and redirect URI
client_id = "my_client_ID"
client_secret = "my_client_secret"
redirect_uri = "https://example.com/"
state = "343ab3341331218786ef" # This is for CSRF staff

# URL for the OAuth 2.0 authorization and token endpoints
auth_endpoint = "https://www.chatwork.com/packages/oauth2/login.php"
token_endpoint = "https://oauth.chatwork.com/token"

# Generate a code verifier
code_verifier = base64.urlsafe_b64encode(secrets.token_bytes(32)).rstrip(b'=').decode('utf-8')

# Calculate the code challenge
code_challenge = base64.urlsafe_b64encode(hashlib.sha256(code_verifier.encode('utf-8')).digest()).decode('utf-8').replace('=', '')

# Parameters for the authorization request
params = {
    "response_type": "code",
    "client_id": client_id,
    "redirect_uri": redirect_uri,
    "scope": "rooms.all:read_write",
    "state": state,  
    "code_challenge": code_challenge,
    "code_challenge_method": "S256"
}

# Construct the authorization URL
auth_url = auth_endpoint + "?" + urlencode(params)

# Print the authorization URL
print("Open this URL in your browser and authorize the application:")
print(auth_url)

# After the user authorizes the application, they will be redirected to your redirect URI
# Parse the authorization code from the redirect URI
authorization_code = input("Enter the authorization code from the redirect URI (or leave empty if denied): ")
parsed_url = urlparse(authorization_code)
query_params = parse_qs(parsed_url.query)
code = query_params["code"][0]
print(code)

if not code:
    print("Authorization denied by user.")
    # Handle the denial, such as displaying a message to the user or redirecting them to a different page
else:
    # Parameters for the token request
    token_params = {
        "grant_type": "authorization_code",
        "code": code,
        "client_id": client_id,
        "redirect_uri": redirect_uri,
        "code_verifier": code_verifier
    }

    # Make a POST request to the token endpoint to exchange the authorization code for an access token
    response = requests.post(token_endpoint, data=token_params, headers={"Authorization": "Basic " + base64.b64encode(f"{client_id}:{client_secret}".encode('utf-8')).decode('utf-8')})
    print(response)

    # Parse the access token from the response
    access_token = response.json().get('access_token')
    print(access_token)

已知客户端类型已设置为confidential，可正常获取client_secret。

可能的问题点及解决方案 #

1. 授权码解析逻辑缺陷
   当前代码默认用户输入完整重定向URL，若用户直接输入code字符串，urlparse无法解析出query参数，会触发KeyError并传递错误code到token接口。

   修正方式：增加兼容逻辑，自动判断输入类型：

   authorization_input = input("请输入重定向URI或授权码：").strip()
   code = None
   if authorization_input:
       parsed_url = urlparse(authorization_input)
       if parsed_url.query:
           query_params = parse_qs(parsed_url.query)
           code = query_params.get("code", [None])[0]
       else:
           # 用户直接输入了code
           code = authorization_input
   if not code:
       print("授权被拒绝或输入无效")
   

2. 未查看具体错误详情
   仅打印response对象无法得知401的具体原因（如invalid_client、invalid_code、code_verifier_mismatch等）。建议补充打印响应文本：

   print("响应状态码：", response.status_code)
   print("响应内容：", response.text)
   

3. Basic Auth编码问题
   确保client_id与client_secret拼接时无多余空格，单独提取编码逻辑避免出错：

   auth_credentials = f"{client_id}:{client_secret}".encode('utf-8')
   auth_header = "Basic " + base64.b64encode(auth_credentials).decode('utf-8').strip()
   headers = {"Authorization": auth_header}
   

4. redirect_uri完全一致性
   授权请求与token请求中的redirect_uri必须与ChatWork开发者后台配置的完全一致，包括协议、域名、路径及末尾斜杠。

5. PKCE参数配对验证
   确认授权阶段生成的code_challenge与token阶段传递的code_verifier是配对的，可在生成后打印两者的值做校验。

修改后的完整代码示例 #

import requests
from urllib.parse import urlencode, urlparse, parse_qs
import hashlib
import base64
import secrets

# 配置信息
client_id = "my_client_ID"
client_secret = "my_client_secret"
redirect_uri = "https://example.com/"
state = "343ab3341331218786ef"

# 端点URL
auth_endpoint = "https://www.chatwork.com/packages/oauth2/login.php"
token_endpoint = "https://oauth.chatwork.com/token"

# 生成PKCE参数
code_verifier = base64.urlsafe_b64encode(secrets.token_bytes(32)).rstrip(b'=').decode('utf-8')
code_challenge = base64.urlsafe_b64encode(hashlib.sha256(code_verifier.encode('utf-8')).digest()).decode('utf-8').replace('=', '')

# 构建授权URL
auth_params = {
    "response_type": "code",
    "client_id": client_id,
    "redirect_uri": redirect_uri,
    "scope": "rooms.all:read_write",
    "state": state,
    "code_challenge": code_challenge,
    "code_challenge_method": "S256"
}
auth_url = auth_endpoint + "?" + urlencode(auth_params)
print("授权URL：", auth_url)

# 获取授权码
authorization_input = input("请输入重定向URI或授权码：").strip()
code = None
if authorization_input:
    parsed = urlparse(authorization_input)
    if parsed.query:
        query_params = parse_qs(parsed.query)
        code = query_params.get("code", [None])[0]
    else:
        code = authorization_input

if not code:
    print("授权被拒绝或输入无效")
else:
    # 构建token请求参数
    token_params = {
        "grant_type": "authorization_code",
        "code": code,
        "client_id": client_id,
        "redirect_uri": redirect_uri,
        "code_verifier": code_verifier
    }
    # 生成Basic Auth头
    auth_cred = f"{client_id}:{client_secret}".encode('utf-8')
    auth_header = "Basic " + base64.b64encode(auth_cred).decode('utf-8')
    headers = {"Authorization": auth_header}

    # 发送token请求
    response = requests.post(token_endpoint, data=token_params, headers=headers)
    print("响应状态码：", response.status_code)
    print("响应内容：", response.text)

    # 解析结果
    if response.status_code == 200:
        access_token = response.json().get('access_token')
        print("获取到的AccessToken：", access_token)

内容的提问来源于stack exchange，提问作者Luis Alejandro Vargas Ramos