前言

林舟和许岚都在蓝桥项目工作，但两人拥有不同的行动记录。只传项目标识，服务器还不知道调用者是谁；用户在对话里填写姓名，也不能证明这个姓名对应他的账号。查询自己的待办时，需要先用有效登录凭据确定业务用户，再按项目和状态筛选。姓名可以用于显示，不能代替这一步身份核对。

账号接入涉及小艺 Claw、华为账号服务器和业务服务，三者需要按实际凭据的传递顺序连接。我们先通过本地模拟观察授权、查询、过期、刷新和解绑，确认业务服务怎样根据凭据选择数据。

一、先确认真实接入条件，再决定使用哪条操作路径

从目录打开 examples/account/README.md，检查服务文件、配置说明与真实补验要求。server.py 包含两个明确的账号来源：huawei 会向账号服务器发送真实 HTTPS 表单，mock 使用两名虚构用户。当前没有真实授权码和生产密钥，可以先执行后面的完整模拟，观察自己服务器的处理过程。准备真实接入时则必须选择 huawei，缺少凭据要停下来补配置，不能靠模拟用户继续返回数据。

当前手机号方案需要 AGC 应用对应的 OAuth 客户端标识、客户端密钥和获批手机号权限。应用图标与身份信息还会出现在授权卡上，所以要确认它们属于实际提供行动查询的应用。手机号权限的申请条件限定企业开发者，支持 Phone、Tablet；这个条件约束这里使用的手机号关联方案，不代表普通文字 Skill 都需要企业账号。

这条流程里有两层凭据，接收方和用途不同。用户同意绑定后，小艺把 authCode 交给服务器，服务器用它换取华为 access_token 并读取已授权用户信息；关联业务用户以后，再签发自己的 loginToken 与 loginRefreshToken。后续查询使用业务登录凭据，刷新使用业务刷新凭据。

二、从完整模拟开始，观察同一项目怎样返回两份数据

将终端的工作目录设为根目录再执行下面两行命令。第一行进入 examples/account/，该目录中应该能看到 server.py 和 reproduce.py，第二行运行复现脚本。程序只使用 Python 标准库，不需要预先启动常驻服务器，也不用配置真实密钥。找不到脚本或相邻模块时，先核对命令的起始目录与实际文件，不要修改导入语句来迁就路径。

cd examples/account
python3 reproduce.py

复现脚本会建立临时 SQLite 数据库，在本机随机端口启动 HTTP 服务，再为两名教学用户分别执行模拟授权。它使用的测试接入码和密钥只用于这个临时进程，程序结束后会关闭服务并清理临时目录。终端最终打印脱敏 JSON，不打印授权码、手机号或 token 原值。

已有一次实际执行的输出位于 samples/10-account-actual-mock.json。打开原件，先确认根字段 mode 为 mock，realHuaweiAuthorizationVerified 为 false，再展开两项查询。先看这些身份字段，能够防止把后面的查询成功读成华为授权已经成功。它们也是保存自己复现结果时应该保留的说明，不应为了让画面更像成功记录而删除。

linzhouQuery 使用 blue-bridge 与 open，返回林舟的提交接口清单；xulanQuery 使用相同项目和状态，返回许岚的确认联调时间。两次请求切换的是登录凭据，因此需要比较 items 中的标识和负责人，确认数据确实随账号变化。只有显示姓名改变而记录仍然相同，还不足以说明服务器已经按用户筛选。

这些记录由账号服务在模拟模式下创建，与前面公开教学查询服务的记录分别保存。如果模拟输出的某个 items 为空，先检查临时数据是否创建，再检查项目和状态；如果返回凭据错误，则先检查绑定。两种返回需要不同处理。真实用户查询没有待办时也可以合法返回空数组，但服务器在确认身份失败时不能用空数组掩盖账号问题。

三、打开身份查询代码，确认姓名无法改变数据归属

沿着模拟结果回到 server.py 的 actions 方法。方法先拒绝请求中的 userId，随后读取项目和状态，再用 login-token 对应的摘要找到绑定。绑定不存在、已经撤销或已经过期时，返回账号错误；绑定有效时，把其中的 user_id 与项目标识一起放入 SQL 条件。两名用户的查询结果之所以不同，依赖的是这条条件关系。

回调和查询还采用了不同的鉴权条件。平台回调使用 AK/SK，供服务器核对配置中的调用者；行动查询使用业务登录凭据，供服务器确定数据所属用户。回调鉴权通过不会替不带登录凭据的查询补出身份，查询凭据也不能替代回调签名。遇到错误时先定位接口，再检查它实际要求的请求头，避免把同一组密钥填进不同位置。

业务用户标识由 clientId 与 openID 的组合生成摘要，当前实现没有把手机号写入数据库或普通日志。这个选择使手机号变化不会直接改变长期数据归属，也避免把会议文字中的姓名当成身份。真实业务已有账号系统时，还需要按已有绑定规则关联原用户，处理冲突与重复关联。这里两名教学用户只用于检查凭据与记录的关系，没有实现企业账户合并或历史数据迁移。

Skill 请求业务时，使用 GET /api/actions?projectId=blue-bridge&status=open，登录凭据放在 login-token 请求头。projectId 和 status 筛选当前用户的数据，不指定当前用户是谁。修改项目以后，服务器仍然使用同一绑定中的身份；添加一个姓名参数也不能让它变成另一个账号。如果需要切换真实账号，应从授权和绑定流程完成，再用新凭据查询。

继续查看服务的错误返回。HTTP 401 和代码 4201 表示当前登录凭据无法继续使用，界面应该引导刷新或重新授权，不能回答没有行动记录。反过来，有效绑定返回空 items 才说明当前筛选下没有匹配数据。技能对错误的解释需要保留服务器判断，否则账号服务已经拒绝请求，用户仍可能被误导为查询完成。

四、在原始输出中观察过期、刷新与解绑

完整模拟在两名用户首次查询后，把服务使用的测试时钟向后推进八天，再用林舟的旧登录凭据查询。当前登录凭据有效七天，所以 expiredQuery 返回 HTTP 401、代码 4201。时间推进发生在脚本注入的时钟里，没有修改电脑日期，也没有让真实华为凭据提前过期。复现时无需手工等待八天或更改系统设置。

refresh 中的 error.code 为零，newLoginTokenPresent 为 true，随后 refreshedQuery 又返回林舟的记录，这些结果共同说明新业务登录凭据能够用于查询。输出已隐藏 token 原值，不能靠截图中的字符串比较变化。配套检查还确认刷新后旧 token 被拒绝，因此核对刷新过程时，需要同时看新值查询和旧值失效，不能只看刷新接口的完成状态。

刷新内部使用事务读取绑定，检查撤销状态与刷新截止时间，再替换登录凭据摘要。下面的片段保持当前实现，读取的是自己的业务绑定表，没有请求华为刷新接口。刷新凭据的截止时间仍沿用原值，当前设定为六个自然月，不能理解成每次刷新又获得新的半年期限。

row = db.execute('SELECT * FROM bindings WHERE refresh_hash=?', (digest(refresh),)).fetchone()
if row is None or row['revoked'] or row['refresh_exp'] <= now:
    raise AdapterError(4201, '绑定已失效，请重新授权')
db.execute('UPDATE bindings SET token_hash=?,token_exp=? WHERE id=?', (digest(token), token_exp, row['id']))

输出随后执行 deauthorize，unbind 返回空 result 与 error.code=0，再执行查询时得到 afterUnbindQuery 的 HTTP 401 和代码 4201。这组相邻结果证明模拟服务把绑定撤销以后，仍未过期的新登录凭据也无法使用。解绑没有删除行动数据，重新建立合法绑定后的数据处理仍由业务规则决定。

需要单独复现服务检查时，在同一目录执行 python3 -m unittest -v test_account.py。现有十项检查覆盖拒绝授权、跨用户查询、过期与刷新、解绑、状态重读、请求签名与重放限制。Provider 请求字段使用可控 HTTP 响应检查，仍不构成真实授权。看见全部通过以后，可以继续配置真实服务，但不据此填写设备已绑定或环境已更新。

五、配置生产环境，缺少密钥时让服务明确停止

打开 .env.example，逐项核对 ACCOUNT_MODE、客户端标识与密钥、回调接入码与密钥、数据库路径和端口。当前服务直接读取进程环境变量，不会自动加载名为 .env 的文件。因此，只复制配置说明再启动，进程收到的值仍可能为空。使用部署工具管理变量时，需要由部署配置将这些值传入进程。

检查环境配置时，先确认必需变量都已传入，再核对客户端标识、客户端密钥和回调密钥是否属于相应配置。变量非空只能排除缺值，不能证明它与目标应用匹配。数据库路径和端口可以按部署安排设置，账号身份与鉴权值则要回到真实配置逐项核对，名称相近不能作为一致的依据。

在本地服务目录中单独保存一份私有配置，文件名例如 .env.local，保持 ACCOUNT_MODE=huawei，填写真实应用和平台配置的值。回调密钥需要与账号鉴权中的 AK/SK 匹配，不能使用模拟脚本的测试值。文件内容应符合 Shell 赋值语法，包含空格的值要正确引用。确认它是自己维护的可信配置后，可以在服务目录显式导入，再启动程序。

set -a
. ./.env.local
set +a
python3 server.py

启动前检查 HUAWEI_CLIENT_ID 与 HUAWEI_CLIENT_SECRET 已设置。缺少它们时，生产 Provider 构造会报错，程序不会转入 mock。没有明确 ACCOUNT_MODE 时同样停止。看到这类配置错误，就返回环境设置补齐真实值；不要把模式改成模拟后仍然按生产成功继续操作。当前本地服务监听环回地址，正式回调还需要自己的 HTTPS 代理与部署配置。

HUAWEI_REDIRECT_URI 只在取得授权码的请求使用注册回调 URI 时填写，并与注册值一致。当前实现没有猜测 Claw 内部地址，也没有提供一个通用占位值。ACCOUNT_DB 指向需要持久保存的业务状态，服务重启后继续读取相同数据库；换成临时路径会改变绑定状态，排查时也需要核对实际打开的是哪一份数据库。密钥与凭据不进入文章、截图或普通日志，观察结果时只保留字段名与脱敏状态。

六、在账号鉴权页面把回调和业务域名对应起来

生产服务准备好以后，打开小艺开放平台的 授权管理，进入 账号鉴权，选择 新建账号鉴权。应用名称选择真实 AGC 应用，图标与描述对应实际授权用途，API URL 指向服务的 POST /rpc，使用账号的域名填写 Skill 请求业务数据的实际域名。回调入口负责签发业务凭据，业务域名负责使用它，两处需要与真正部署的地址对应。

在这个表单临时选择已有的会议随记 Pro 后，客户端标识自动出现。API地址和密钥没有填写，表单已经取消，没有建立新的账号绑定。当前画面没有单列使用账号的域名字段，填写前需要按当前应用配置继续核对。

当前服务实现 AK/SK，因此配置中选择这一方式，再把同一接入码和密钥传入服务环境。平台每次回调携带 accessKey、毫秒时间戳 ts 与 sign，服务器计算 Base64(HMAC-SHA256(secretKey, ts)) 并比较。这个签名只覆盖时间戳，HTTPS继续保护请求正文。服务另外设置五分钟时间窗口与二十四小时消息标识去重，这是当前服务的限制，调试时要让时钟、消息标识和真实请求一起核对。

回调参数放在 params.message.parts 的 data 对象里。授权使用 authCode 与 clientId，刷新和解绑使用 loginRefreshToken。下面保留参数位置说明，其中中文文本用于说明参数含义，真实调用需要换成有效凭据。真实回调应该读取实际应用标识和本次授权码，不能把这段说明原样发给华为服务器。

{
  "jsonrpc": "2.0",
  "id": "一次新的全局消息标识",
  "method": "authorize",
  "params": {
    "message": {
      "role": "user",
      "parts": [{"kind":"data","data":{"authCode":"本次授权码","clientId":"实际客户端标识"}}]
    }
  }
}

clientId 不一致时返回参数错误，AK/SK不匹配时先检查回调鉴权，时间戳太旧时核对请求时钟。重复消息和授权码重放也会被当前实现拒绝，所以不能反复用同一请求试到成功。账号样例的解绑方法带一个前导空格，服务兼容这一已知形式；真实平台究竟使用哪个值，仍需读回请求。

七、换取用户信息失败时，不签发业务登录凭据

服务器收到授权请求后，先检查客户端标识并登记授权码已尝试使用，再发送表单交换。字段包含 grant_type=authorization_code、code、client_id 和 client_secret，必要时追加匹配的回调 URI。表单编码交给标准库，避免授权码中的加号或斜杠在手工拼接时改变含义。得到有效 access_token 后，再读取用户信息，两个请求按顺序执行。

fields = {'grant_type': 'authorization_code', 'code': auth_code,
          'client_id': self.client_id, 'client_secret': self.client_secret}
if self.redirect_uri:
    fields['redirect_uri'] = self.redirect_uri
token = self.post_form(TOKEN_URL, fields)
access_token = token.get('access_token')
if not isinstance(access_token, str) or not access_token:
    raise AdapterError(4104, '账号响应缺少access_token')
user = self.post_form(USER_URL, {'access_token': access_token})
open_id, phone = user.get('openID'), user.get('mobileNumber')
if not isinstance(open_id, str) or not open_id or not isinstance(phone, str) or not phone:
    raise AdapterError(4105, '账号响应缺少openID或手机号；本接入方案要求相应授权')

缺少 access_token 时返回 4104，缺少 openID 或本方案要求的 mobileNumber 时返回 4105。HTTP请求正常结束以后，服务还检查 NSP_STATUS 和正文错误，权限不足或用户拒绝会进入错误分支。遇到这些结果时，先核对应用、权限与实际响应，不能在缺少手机号时编造一个身份继续绑定。是否采用其他账号关联方式需要重新决定业务方案，当前服务不会自行更换。

4103 表示网络超时，服务没有取得可用结果，但授权码可能已经被消费。当前实现不会自动重试，同一授权码再次提交还会被本地重放检查拒绝，因此继续操作需要取得新的合法授权。正式部署也仍需完善回调交付、限流和故障恢复，避免服务器已经签发凭据，设备却没有收到。

八、让Skill重新读取凭据，再核对真实设备返回

打开 examples/account/meeting-actions/SKILL.md 与同目录的 module.json，先确认正在修改自己的接入副本。在 SKILL.md 顶部的 env 中找到 CLIENT_ID_PENDING_login_token 和 CLIENT_ID_PENDING_login_token_expire_time，将两个键中的 CLIENT_ID_PENDING 都替换为实际应用的同一个客户端标识，保留各自的后缀。再核对使用条件中的占位说明，并让第一项环境说明及查询规则中调用 huawei_id_tool 时使用的 clientId 与这个前缀一致。skillName 继续使用文件顶部的名称 meeting-actions，不随客户端标识改变。

当前 module.json 只有 version、availableOn 和 toolDependencies，没有 CLIENT_ID_PENDING 或 clientId 字段，所以客户端值不写入这份配置。模板中的待绑定标识没有替换时，查询规则要求停止。Claw 环境路径为 /home/sandbox/.openclaw/.xiaoyienv，后续读取的登录凭据与过期时间键必须采用刚才核对的真实前缀。这个环境属于技能运行侧，电脑上的 .env.local 则供服务器使用，两份文件不能互相代替。

凭据为空或已经过期时，技能调用一次 huawei_id_tool，参数使用真实 clientId 和 skillName。工具完成后重新读取环境，再把新登录凭据放入 login-token 请求头。

在Claw中可以从使用技能的对话开始，凭据缺失时观察授权卡，再核对卡上的应用与用途。确认对应真实应用后，选择 同意并绑定，继续观察服务器收到的授权回调；不接受授权时，当前查询就应停止。另一条路径是在Claw账号管理页选择对应应用完成绑定或解绑，再回到技能查询。操作前后都要确认同一应用，避免账号管理显示绑定成功，技能却仍在读取另一个客户端前缀。

服务返回的过期时间包含东八区偏移，比较时需要保留时区，不能只截取日期。如果刷新接口返回成功，环境中的值却没有更新，先检查工具返回与重新读取环境的过程，避免继续发送旧凭据。如果环境已更新，业务查询仍被拒绝，再检查服务器绑定状态和请求头。分别记录这两种现象，才能判断问题发生在凭据更新还是业务鉴权。

在真实设备上，用两名测试用户查询同一项目，分别检查返回项；随后检查刷新后的查询与解绑后的拒绝。服务器回调结构、Claw环境更新和业务结果需要对应到同一次操作，才能定位失败发生在哪里。授权卡没有出现时先检查技能的客户端与域名配置，回调失败时检查鉴权和服务响应，回调成功但查询失败时检查新凭据读取。每个继续条件都回到具体返回，不让一张授权卡代替完整业务检查。

总结

服务器需要先用有效凭据确定业务用户，项目和状态才能用于筛选这个用户的行动。本地模拟已经取得两名用户各自的查询结果，以及过期、刷新和解绑后的返回，能够据此检查服务逻辑。真实接入时，还要把授权回调、Claw 环境中的新凭据和业务查询对应到同一次操作，才能确认账号绑定最终改变了哪份数据的访问。

配套文件

1. 账号版技能的完整查询规则

文件路径：examples/account/meeting-actions/SKILL.md，展示完整内容。

这份规则要求先读取有效业务凭据，再查询当前用户的数据。客户端前缀和真实业务域名尚待绑定，原有占位说明继续保留。

---
name: meeting-actions
description: 在用户授权的账号下查询本人项目行动记录，显示真实查询结果；不读取其他用户数据，不保存任务或发送提醒。
version: 1.0.0
env:
  - CLIENT_ID_PENDING_login_token: 从/home/sandbox/.openclaw/.xiaoyienv读取账号凭证；若为空或过期，调用一次huawei_id_tool，参数clientId必须替换成实际客户端标识，skillName=meeting-actions；刷新完成后重新读取环境文件。标识未绑定时停止，不提交占位请求。
  - CLIENT_ID_PENDING_login_token_expire_time: 账号凭证过期时间，凭证过期时先刷新再查询。
---

## 使用条件

这是一份本地作者账号版规则模板，尚未绑定平台客户端和业务域名，不能直接视为已接受的包。账号鉴权中的使用账号的域名必须与实际HTTPS业务域名一致。CLIENT_ID_PENDING必须同时替换为真实clientId及其环境键前缀。账号服务API URL指向部署后的/rpc；本机127.0.0.1不能填作云侧可访问地址。

## 查询规则

读取有效凭证后，请求GET /api/actions?projectId=用户确认的项目标识&status=open，HTTP请求头携带login-token环境值。业务服务根据凭证确定用户；不得从对话提供userId来改变查询身份。

凭证缺失、过期或接口返回账号异常时只调用一次huawei_id_tool，使用实际clientId与skillName=meeting-actions。用户拒绝或刷新没有返回有效凭证时停止查询，说明授权未完成。刷新成功后重新读取环境值，不沿用刷新前的login-token。

接口items为空时报告当前账号没有符合条件的记录。接口失败时不能用虚构数据代替。凭证、授权码、手机号与密钥不进入回答、正文、普通日志或截图。此版本不提供手机号查询能力，不执行写入、提交或提醒。

服务器的私有环境配置与技能运行侧的凭据环境分别管理，不能把服务器密钥复制到技能描述中。

2. 账号版技能配置

文件路径：examples/account/meeting-actions/module.json，展示完整内容。

这份配置声明技能版本、运行位置和当前工具依赖。

{"version":"1.0.0","availableOn":["cloudSandbox"],"toolDependencies":[]}

真实客户端标识不写入这份配置，应按第八节核对技能规则中的环境键前缀。

3. 服务环境变量模板

文件路径：examples/account/.env.example，展示完整内容。

模板只列变量名与已有默认配置，密钥值保持为空。服务读取进程环境变量，不会自动载入这个文件。

# Mandatory explicit provider; huawei never falls back to mock.
ACCOUNT_MODE=huawei
HUAWEI_CLIENT_ID=
HUAWEI_CLIENT_SECRET=
# Fill only when the OAuth authorization request uses the registered redirect URI.
HUAWEI_REDIRECT_URI=
CALLBACK_ACCESS_KEY=
CALLBACK_SECRET_KEY=
ACCOUNT_DB=account.sqlite3
PORT=8766

真实配置应保存在自己维护的私有文件或部署环境中，不能将填入密钥的副本放进文章。

4. 账号回调与用户查询的完整服务

文件路径：examples/account/server.py，展示完整内容。

完整实现把授权来源、回调鉴权、绑定状态、凭据刷新、解绑和用户查询放在同一份文件中。保留这些相邻处理，便于沿一次凭据的签发与使用检查数据归属。

"""Author account adapter. Mock mode is explicit; no Huawei authorization is simulated silently."""
import base64
import calendar
import datetime as dt
import hashlib
import hmac
import json
import os
import secrets
import sqlite3
import time
import urllib.parse
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

UTC = dt.timezone.utc
CST = dt.timezone(dt.timedelta(hours=8))
TOKEN_URL = 'https://oauth-login.cloud.huawei.com/oauth2/v3/token'
USER_URL = 'https://account.cloud.huawei.com/rest.php?nsp_svc=GOpen.User.getInfo'


class AdapterError(Exception):
    def __init__(self, code, message):
        self.code, self.message = code, message
        super().__init__(message)


def digest(token):
    return hashlib.sha256(token.encode()).hexdigest()


def expires_string(value):
    return dt.datetime.fromtimestamp(value, UTC).astimezone(CST).isoformat(timespec='seconds')


def six_months(value):
    start = dt.datetime.fromtimestamp(value, UTC)
    month = start.month + 6
    year = start.year + (month - 1) // 12
    month = (month - 1) % 12 + 1
    day = min(start.day, calendar.monthrange(year, month)[1])
    return start.replace(year=year, month=month, day=day).timestamp()


class HuaweiProvider:
    def __init__(self, client_id, client_secret, redirect_uri='', opener=urllib.request.urlopen):
        if not client_id or not client_secret:
            raise ValueError('HUAWEI_CLIENT_ID and HUAWEI_CLIENT_SECRET are required in huawei mode')
        self.client_id, self.client_secret, self.redirect_uri = client_id, client_secret, redirect_uri
        self.opener = opener

    def post_form(self, url, fields):
        body = urllib.parse.urlencode(fields).encode()
        request = urllib.request.Request(url, data=body,
            headers={'Content-Type': 'application/x-www-form-urlencoded'}, method='POST')
        try:
            with self.opener(request, timeout=10) as response:
                data = json.loads(response.read(1024 * 1024))
                nsp = response.headers.get('NSP_STATUS', '0')
                if response.status != 200 or nsp not in ('0', '') or data.get('error'):
                    raise AdapterError(4102, '华为账号接口拒绝了授权，请检查权限或重新授权')
                return data
        except AdapterError:
            raise
        except Exception as error:
            raise AdapterError(4103, '华为账号接口未返回可用结果；不要自动重放授权码') from error

    def exchange(self, auth_code):
        fields = {'grant_type': 'authorization_code', 'code': auth_code,
                  'client_id': self.client_id, 'client_secret': self.client_secret}
        if self.redirect_uri:
            fields['redirect_uri'] = self.redirect_uri
        token = self.post_form(TOKEN_URL, fields)
        access_token = token.get('access_token')
        if not isinstance(access_token, str) or not access_token:
            raise AdapterError(4104, '账号响应缺少access_token')
        user = self.post_form(USER_URL, {'access_token': access_token})
        open_id, phone = user.get('openID'), user.get('mobileNumber')
        if not isinstance(open_id, str) or not open_id or not isinstance(phone, str) or not phone:
            raise AdapterError(4105, '账号响应缺少openID或手机号；本接入方案要求相应授权')
        return {'openID': open_id, 'mobileNumber': phone}


class MockProvider:
    """Fictional identity responses; only used with ACCOUNT_MODE=mock."""
    def exchange(self, auth_code):
        identities = {'mock-linzhou': ('mock-open-linzhou', 'mock-phone-linzhou'),
                      'mock-xulan': ('mock-open-xulan', 'mock-phone-xulan')}
        if auth_code not in identities:
            raise AdapterError(4102, '模拟授权被拒绝')
        open_id, phone = identities[auth_code]
        return {'openID': open_id, 'mobileNumber': phone}


class AccountAdapter:
    def __init__(self, path, client_id, access_key, secret_key, provider, mode, clock=time.time):
        if mode not in ('mock', 'huawei') or not client_id or not access_key or not secret_key:
            raise ValueError('Explicit mode, client ID and callback AK/SK are required')
        self.path, self.client_id, self.access_key, self.secret_key = str(path), client_id, access_key, secret_key
        self.provider, self.mode, self.clock = provider, mode, clock
        Path(self.path).parent.mkdir(parents=True, exist_ok=True)
        with self.connect() as db:
            db.executescript('''
              CREATE TABLE IF NOT EXISTS users(id TEXT PRIMARY KEY);
              CREATE TABLE IF NOT EXISTS actions(id TEXT PRIMARY KEY,user_id TEXT NOT NULL,
                project_id TEXT NOT NULL,title TEXT NOT NULL,owner TEXT NOT NULL,status TEXT NOT NULL);
              CREATE TABLE IF NOT EXISTS bindings(id TEXT PRIMARY KEY,user_id TEXT NOT NULL,
                refresh_hash TEXT UNIQUE NOT NULL,refresh_exp REAL NOT NULL,
                token_hash TEXT UNIQUE NOT NULL,token_exp REAL NOT NULL,revoked INTEGER NOT NULL DEFAULT 0);
              CREATE TABLE IF NOT EXISTS callbacks(id TEXT PRIMARY KEY,received_at REAL NOT NULL);
              CREATE TABLE IF NOT EXISTS auth_codes(code_hash TEXT PRIMARY KEY,received_at REAL NOT NULL);
            ''')
        os.chmod(self.path, 0o600)

    def connect(self):
        connection = sqlite3.connect(self.path, timeout=10)
        connection.row_factory = sqlite3.Row
        return connection

    def user_id(self, open_id):
        return digest(self.client_id + ':' + open_id)

    def seed_mock(self):
        if self.mode != 'mock':
            raise ValueError('Only mock mode permits fictional data seeding')
        with self.connect() as db:
            for who, title, name in [('linzhou', '提交接口清单', '林舟'), ('xulan', '确认联调时间', '许岚')]:
                user = self.user_id('mock-open-' + who)
                db.execute('INSERT OR IGNORE INTO users VALUES(?)', (user,))
                db.execute('INSERT OR IGNORE INTO actions VALUES(?,?,?,?,?,?)',
                           ('mock-action-' + who, user, 'blue-bridge', title, name, 'open'))

    def check_caller(self, headers):
        stamp = headers.get('ts', '')
        if not stamp.isdigit() or abs(self.clock() * 1000 - int(stamp)) > 300000:
            raise AdapterError(4001, '回调时间戳无效或超出服务器允许窗口')
        expected = base64.b64encode(hmac.new(self.secret_key.encode(), stamp.encode(), hashlib.sha256).digest()).decode()
        if not hmac.compare_digest(headers.get('accessKey', ''), self.access_key) or \
                not hmac.compare_digest(headers.get('sign', ''), expected):
            raise AdapterError(4002, '回调鉴权失败')

    def rpc(self, message, headers):
        ident = message.get('id') if isinstance(message, dict) else None
        try:
            self.check_caller(headers)
            if not isinstance(message, dict) or message.get('jsonrpc') != '2.0' or not isinstance(ident, str) or not ident:
                raise AdapterError(-32600, 'JSON-RPC请求结构无效')
            method = message.get('method')
            # S16 deauthorize sample has a leading space; accept that one known variant only.
            if method == ' deauthorize':
                method = 'deauthorize'
            if method not in ('authorize', 'refreshToken', 'deauthorize'):
                raise AdapterError(-32601, '不支持的账号方法')
            wrapper = message.get('params', {}).get('message', {})
            parts = wrapper.get('parts', [])
            if wrapper.get('role') != 'user' or len(parts) != 1 or parts[0].get('kind') != 'data' or \
                    not isinstance(parts[0].get('data'), dict):
                raise AdapterError(-32602, '账号参数需要是唯一data part')
            data = parts[0]['data']
            now = self.clock()
            with self.connect() as db:
                db.execute('DELETE FROM callbacks WHERE received_at < ?', (now - 86400,))
                try:
                    db.execute('INSERT INTO callbacks VALUES(?,?)', (ident, now))
                except sqlite3.IntegrityError:
                    raise AdapterError(4003, '消息标识已处理，请先核对绑定结果')
            if method == 'authorize':
                result = self.authorize(data, now)
            else:
                refresh = data.get('loginRefreshToken')
                if not isinstance(refresh, str) or not refresh:
                    raise AdapterError(-32602, '缺少loginRefreshToken')
                result = self.refresh(refresh, now) if method == 'refreshToken' else self.deauthorize(refresh)
            return {'jsonrpc': '2.0', 'id': ident, 'result': result, 'error': {'code': 0, 'message': ''}}
        except (AdapterError, AttributeError, TypeError, KeyError, IndexError) as error:
            if not isinstance(error, AdapterError):
                error = AdapterError(-32602, '账号参数结构无效')
            return {'jsonrpc': '2.0', 'id': ident, 'result': {},
                    'error': {'code': error.code, 'message': error.message}}

    def authorize(self, data, now):
        code = data.get('authCode')
        if data.get('clientId') != self.client_id or not isinstance(code, str) or not code:
            raise AdapterError(-32602, 'clientId不匹配或缺少authCode')
        with self.connect() as db:
            try:
                db.execute('INSERT INTO auth_codes VALUES(?,?)', (digest(code), now))
            except sqlite3.IntegrityError:
                raise AdapterError(4106, '授权码已经使用或尝试使用，请重新授权')
        identity = self.provider.exchange(code)
        user = self.user_id(identity['openID'])
        token, refresh = secrets.token_urlsafe(32), secrets.token_urlsafe(48)
        token_exp, refresh_exp = now + 7 * 86400, six_months(now)
        with self.connect() as db:
            db.execute('INSERT OR IGNORE INTO users VALUES(?)', (user,))
            db.execute('INSERT INTO bindings VALUES(?,?,?,?,?,?,0)',
                       (secrets.token_hex(16), user, digest(refresh), refresh_exp, digest(token), token_exp))
        return {'loginToken': token, 'loginTokenExpireTime': expires_string(token_exp),
                'loginRefreshToken': refresh, 'loginRefreshTokenExpireTime': expires_string(refresh_exp)}

    def refresh(self, refresh, now):
        token, token_exp = secrets.token_urlsafe(32), now + 7 * 86400
        with self.connect() as db:
            db.execute('BEGIN IMMEDIATE')
            row = db.execute('SELECT * FROM bindings WHERE refresh_hash=?', (digest(refresh),)).fetchone()
            if row is None or row['revoked'] or row['refresh_exp'] <= now:
                raise AdapterError(4201, '绑定已失效，请重新授权')
            db.execute('UPDATE bindings SET token_hash=?,token_exp=? WHERE id=?', (digest(token), token_exp, row['id']))
        return {'loginToken': token, 'loginTokenExpireTime': expires_string(token_exp)}

    def deauthorize(self, refresh):
        with self.connect() as db:
            db.execute('UPDATE bindings SET revoked=1 WHERE refresh_hash=?', (digest(refresh),))
        return {}

    def actions(self, token, query):
        if 'userId' in query:
            raise AdapterError(4004, '业务身份只能由login-token确定')
        project = query.get('projectId', [''])[0]
        status = query.get('status', ['open'])[0]
        if not project or status not in ('open', 'done', 'all'):
            raise AdapterError(4004, 'projectId或status无效')
        with self.connect() as db:
            row = db.execute('SELECT * FROM bindings WHERE token_hash=?', (digest(token),)).fetchone()
            if row is None or row['revoked'] or row['token_exp'] <= self.clock():
                raise AdapterError(4201, '账号凭证失效，请重新授权或刷新')
            sql = 'SELECT id,title,owner,status FROM actions WHERE user_id=? AND project_id=?'
            values = [row['user_id'], project]
            if status != 'all':
                sql += ' AND status=?'
                values.append(status)
            items = [dict(item) for item in db.execute(sql, values)]
        return {'ok': True, 'projectId': project, 'status': status, 'items': items}


def make_server(adapter, address=('127.0.0.1', 0)):
    class Handler(BaseHTTPRequestHandler):
        def log_message(self, *args):
            pass  # Never log headers, tokens, authorization codes, phone numbers or request bodies.

        def send_json(self, status, body):
            self.send_response(status)
            self.send_header('Content-Type', 'application/json; charset=utf-8')
            self.send_header('Cache-Control', 'no-store')
            self.end_headers()
            self.wfile.write(json.dumps(body, ensure_ascii=False).encode())

        def do_POST(self):
            if self.path != '/rpc':
                return self.send_json(404, {'ok': False, 'error': 'NOT_FOUND'})
            try:
                size = int(self.headers.get('Content-Length', '0'))
                if size <= 0 or size > 65536:
                    return self.send_json(413, {'ok': False, 'error': 'BODY_SIZE'})
                message = json.loads(self.rfile.read(size))
            except (ValueError, json.JSONDecodeError):
                return self.send_json(400, {'ok': False, 'error': 'INVALID_JSON'})
            self.send_json(200, adapter.rpc(message, self.headers))

        def do_GET(self):
            path = urllib.parse.urlsplit(self.path)
            if path.path == '/':
                return self.send_json(200, {'service': 'meeting-actions account adapter', 'mode': adapter.mode,
                                           'note': 'mock只验证作者服务；不代表华为授权成功'})
            if path.path != '/api/actions':
                return self.send_json(404, {'ok': False, 'error': 'NOT_FOUND'})
            try:
                result = adapter.actions(self.headers.get('login-token', ''), urllib.parse.parse_qs(path.query))
                self.send_json(200, result)
            except AdapterError as error:
                self.send_json(401 if error.code == 4201 else 400, {'ok': False, 'code': error.code, 'message': error.message})
    return ThreadingHTTPServer(address, Handler)


def from_environment():
    mode = os.environ.get('ACCOUNT_MODE')
    client = os.environ.get('HUAWEI_CLIENT_ID', '')
    if mode == 'mock':
        provider = MockProvider()
    elif mode == 'huawei':
        provider = HuaweiProvider(client, os.environ.get('HUAWEI_CLIENT_SECRET', ''),
                                  os.environ.get('HUAWEI_REDIRECT_URI', ''))
    else:
        raise ValueError('Set ACCOUNT_MODE explicitly to huawei or mock; there is no fallback')
    adapter = AccountAdapter(os.environ.get('ACCOUNT_DB', 'account.sqlite3'), client,
        os.environ.get('CALLBACK_ACCESS_KEY', ''), os.environ.get('CALLBACK_SECRET_KEY', ''), provider, mode)
    if mode == 'mock':
        adapter.seed_mock()
    return adapter


if __name__ == '__main__':
    application = from_environment()
    server = make_server(application, ('127.0.0.1', int(os.environ.get('PORT', '8766'))))
    print(json.dumps({'mode': application.mode, 'listen': '127.0.0.1', 'port': server.server_port,
                      'realHuaweiAuthorizationVerified': False}))
    server.serve_forever()

该文件包含真实请求路径与明确模拟模式。这里展示的是已有实现，没有重新取得真实华为授权或设备绑定结果。

5. 两名用户的完整模拟复现程序

文件路径：examples/account/reproduce.py，展示完整内容。

这份程序在临时目录中建立本地服务，依次比较两名用户的查询、凭据过期、刷新和解绑。测试接入码与密钥仅供这个模拟进程使用，输出不包含凭据原值。

"""Run a complete author-server mock HTTP scenario, print only redacted evidence."""
import base64
import hashlib
import hmac
import json
import tempfile
import threading
import urllib.request
from pathlib import Path
from server import AccountAdapter, MockProvider, make_server


def main():
    now = [1791072000.0]
    with tempfile.TemporaryDirectory() as folder:
        adapter = AccountAdapter(Path(folder) / 'state.sqlite3', 'mock-client', 'test-ak', 'test-sk',
                                 MockProvider(), 'mock', clock=lambda: now[0])
        adapter.seed_mock()
        httpd = make_server(adapter)
        worker = threading.Thread(target=httpd.serve_forever, daemon=True)
        worker.start()
        base = 'http://127.0.0.1:' + str(httpd.server_port)
        seq = [0]
        def rpc(method, data):
            seq[0] += 1
            stamp = str(int(now[0] * 1000))
            sign = base64.b64encode(hmac.new(b'test-sk', stamp.encode(), hashlib.sha256).digest()).decode()
            body = {'jsonrpc': '2.0', 'id': 'reproduce-' + str(seq[0]), 'method': method,
                    'params': {'message': {'role': 'user', 'parts': [{'kind': 'data', 'data': data}]}}}
            request = urllib.request.Request(base + '/rpc', data=json.dumps(body).encode(),
                headers={'Content-Type': 'application/json', 'accessKey': 'test-ak', 'ts': stamp, 'sign': sign})
            with urllib.request.urlopen(request) as response:
                return json.load(response)
        def query(token):
            request = urllib.request.Request(base + '/api/actions?projectId=blue-bridge&status=open', headers={'login-token': token})
            try:
                with urllib.request.urlopen(request) as response:
                    return json.load(response)
            except urllib.error.HTTPError as failure:
                return {'httpStatus': failure.code, **json.load(failure)}
        try:
            lin = rpc('authorize', {'authCode': 'mock-linzhou', 'clientId': 'mock-client'})['result']
            xu = rpc('authorize', {'authCode': 'mock-xulan', 'clientId': 'mock-client'})['result']
            report = {'mode': 'mock', 'realHuaweiAuthorizationVerified': False,
                      'linzhouQuery': query(lin['loginToken']), 'xulanQuery': query(xu['loginToken'])}
            now[0] += 8 * 86400
            report['expiredQuery'] = query(lin['loginToken'])
            refreshed = rpc('refreshToken', {'loginRefreshToken': lin['loginRefreshToken']})
            report['refresh'] = {'error': refreshed['error'], 'newLoginTokenPresent': bool(refreshed['result'].get('loginToken'))}
            report['refreshedQuery'] = query(refreshed['result']['loginToken'])
            report['unbind'] = rpc('deauthorize', {'loginRefreshToken': lin['loginRefreshToken']})
            report['afterUnbindQuery'] = query(refreshed['result']['loginToken'])
            print(json.dumps(report, ensure_ascii=False, indent=2))
        finally:
            httpd.shutdown()
            httpd.server_close()
            worker.join()


if __name__ == '__main__':
    main()

运行入口仍为 python3 reproduce.py。它不会访问华为账号服务器，也不会建立真实账号绑定。

6. 脱敏后的历史模拟结果

文件路径：samples/10-account-actual-mock.json，展示完整内容。

下面保留已有模拟结果，用于核对相同项目怎样随登录凭据返回不同用户的数据。

{
  "mode": "mock",
  "realHuaweiAuthorizationVerified": false,
  "linzhouQuery": {
    "ok": true,
    "projectId": "blue-bridge",
    "status": "open",
    "items": [
      {
        "id": "mock-action-linzhou",
        "title": "提交接口清单",
        "owner": "林舟",
        "status": "open"
      }
    ]
  },
  "xulanQuery": {
    "ok": true,
    "projectId": "blue-bridge",
    "status": "open",
    "items": [
      {
        "id": "mock-action-xulan",
        "title": "确认联调时间",
        "owner": "许岚",
        "status": "open"
      }
    ]
  },
  "expiredQuery": {
    "httpStatus": 401,
    "ok": false,
    "code": 4201,
    "message": "账号凭证失效，请重新授权或刷新"
  },
  "refresh": {
    "error": {
      "code": 0,
      "message": ""
    },
    "newLoginTokenPresent": true
  },
  "refreshedQuery": {
    "ok": true,
    "projectId": "blue-bridge",
    "status": "open",
    "items": [
      {
        "id": "mock-action-linzhou",
        "title": "提交接口清单",
        "owner": "林舟",
        "status": "open"
      }
    ]
  },
  "unbind": {
    "jsonrpc": "2.0",
    "id": "reproduce-4",
    "result": {},
    "error": {
      "code": 0,
      "message": ""
    }
  },
  "afterUnbindQuery": {
    "httpStatus": 401,
    "ok": false,
    "code": 4201,
    "message": "账号凭证失效，请重新授权或刷新"
  }
}