跳转至

payment_accounts

feishu.agent.payment_accounts

收款账户（payment account）解析：从用户**本人**历史审批实例中恢复其银行账户，句柄优先、内存暂存、严格按用户隔离。

飞书没有枚举用户绑定收款账户的接口，但用户本人过去提交的审批实例里带有完整的账户值（自包含对象，而非不可逆 令牌），可原样重新提交。本模块只读取**请求用户本人**的历史实例（经 feishu.approval.instances 的 query 按其 open_id 过滤——绝不触及他人），抽取账户值，向模型只暴露**不可逆句柄 + 脱敏标签**；完整账户值仅存于内存 （绝不落盘），并只在提交瞬间由 feishu.agent.payment_accounts.PaymentAccountResolver.resolve 重新带入。

因此即使模型被越权操控（jailbreak），它既看不到完整卡号，也无法触达他人的收款账户。详见 feishu.agent。

PaymentAccount dataclass

一个收款账户：模型可见的句柄 + 脱敏标签，外加仅服务端可见的完整控件值。

源代码位于： feishu/agent/payment_accounts.py
Python
56
57
58
59
60
61
62
63
64
65
66
67
@dataclass
class PaymentAccount:
    r"""一个收款账户：模型可见的句柄 + 脱敏标签，外加仅服务端可见的完整控件值。"""

    account_id: str  # opaque handle (pa_...), model-safe — parallels SharedFile.file_id
    label: str  # privacy-masked, model-safe (e.g. "杭州银行 ****8383 (张三)")
    account_value: dict[str, Any]  # the full account widget value — server-side ONLY, never to the model
    user_keys: tuple[str, ...] = ()  # owning user's alias keys — parallels SharedFile.user_keys

    def summary(self) -> dict[str, str]:
        r"""模型安全视图：仅句柄 + 脱敏标签，绝不含完整账户值 / 卡号。"""
        return {"account_id": self.account_id, "label": self.label}

summary

Python
summary() -> dict[str, str]

模型安全视图：仅句柄 + 脱敏标签，绝不含完整账户值 / 卡号。

源代码位于： feishu/agent/payment_accounts.py
Python
65
66
67
def summary(self) -> dict[str, str]:
    r"""模型安全视图：仅句柄 + 脱敏标签，绝不含完整账户值 / 卡号。"""
    return {"account_id": self.account_id, "label": self.label}

PaymentAccountResolver

从「请求用户本人」的历史审批实例解析其收款账户，句柄优先、内存暂存、严格按用户隔离。

recent 经 feishu.approval.instances.InstancesNamespace.query 仅按用户 open_id 拉取其本人实例， 再抽取账户值、去重、生成句柄与脱敏标签；完整账户值仅缓存在内存（绝不落盘）。resolve 仅在提交瞬间把句柄 还原为完整值，供 feishu.agent.toolkit.approvals.create_approval_instance 填入账户控件。模型自始至终 只接触句柄与脱敏标签。命名与 feishu.agent.shared_files.SharedFileResolver 对齐（recent / user_keys）。

源代码位于： feishu/agent/payment_accounts.py
Python
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
class PaymentAccountResolver:
    r"""
    从「请求用户本人」的历史审批实例解析其收款账户，句柄优先、内存暂存、严格按用户隔离。

    `recent` 经 [feishu.approval.instances.InstancesNamespace.query][] 仅按用户 `open_id` 拉取其本人实例，
    再抽取账户值、去重、生成句柄与脱敏标签；完整账户值仅缓存在内存（绝不落盘）。`resolve` 仅在提交瞬间把句柄
    还原为完整值，供 [feishu.agent.toolkit.approvals.create_approval_instance][] 填入账户控件。模型自始至终
    只接触句柄与脱敏标签。命名与 [feishu.agent.shared_files.SharedFileResolver][] 对齐（`recent` / `user_keys`）。
    """

    def __init__(self, client: Any, *, lookback_ms: int = 730 * 24 * 3600 * 1000) -> None:
        self._client = client
        self._lookback_ms = lookback_ms
        # user_key -> {account_id: PaymentAccount}; full values live here in memory only, never persisted.
        self._cache: dict[str, dict[str, PaymentAccount]] = {}

    async def recent(
        self, user: Mapping[str, Any], *, approval_code: str | None = None, limit: int = 10
    ) -> list[PaymentAccount]:
        r"""返回请求用户本人历史里出现过的去重收款账户（句柄 + 脱敏标签）；无 `open_id` 时拒绝（绝不枚举他人）。"""
        user_keys = payment_account_keys(user)
        open_id = user.get("open_id") if isinstance(user, Mapping) else None
        if not user_keys or not open_id:
            # Without the user's own open_id we cannot scope the query to them — refuse rather than over-read.
            return []
        end = int(time.time() * 1000)
        start = end - self._lookback_ms
        try:
            items = await self._client.approval.instances.query(
                user_id=open_id,
                approval_code=approval_code,
                user_id_type="open_id",
                start_time=str(start),
                end_time=str(end),
                max_items=50,
            )
        except Exception:  # noqa: BLE001 - an account lookup must never crash the tool; degrade to "none found"
            logging.getLogger("feishu").debug("payment account query failed for the requesting user", exc_info=True)
            return []
        accounts: dict[str, PaymentAccount] = {}
        for item in items:
            instance_node = item.get("instance") if isinstance(item, Mapping) else None
            code = instance_node.get("code") if isinstance(instance_node, Mapping) else None
            if not code:
                continue
            try:
                instance = await self._client.approval.instances.get(code)
            except Exception:  # noqa: BLE001 - skip an instance we cannot read rather than failing the whole lookup
                logging.getLogger("feishu").debug(
                    "could not read instance %s for payment accounts", code, exc_info=True
                )
                continue
            if not _instance_belongs_to(instance, user):
                continue  # defense in depth: never harvest an instance that is not the requesting user's own
            for widget in approval_account_widgets(instance):
                value = widget.get("value")
                number = approval_account_number(value)
                if not number:
                    continue
                handle = payment_account_handle(number)
                if handle not in accounts:
                    accounts[handle] = PaymentAccount(
                        account_id=handle,
                        label=approval_account_label(value),
                        account_value=dict(value),
                        user_keys=user_keys,
                    )
                if len(accounts) >= limit:
                    break
            if len(accounts) >= limit:
                break
        if user_keys:
            self._cache.setdefault(user_keys[0], {}).update(accounts)
        return list(accounts.values())

    async def resolve(self, user: Mapping[str, Any], account_id: str) -> dict[str, Any] | None:
        r"""把句柄还原为完整账户控件值，严格限定为请求用户本人；缓存未命中时回源一次再试，仍无则返回 `None`。"""
        user_keys = payment_account_keys(user)
        if not user_keys:
            return None
        primary = user_keys[0]
        account = self._cache.get(primary, {}).get(account_id)
        if account is None:
            await self.recent(user)  # repopulate (process restart / first use), then retry
            account = self._cache.get(primary, {}).get(account_id)
        if account is None:
            return None
        if not set(user_keys) & set(account.user_keys):
            return None  # the handle must belong to THIS user
        return dict(account.account_value)

recent async

Python
recent(user: Mapping[str, Any], *, approval_code: str | None = None, limit: int = 10) -> list[PaymentAccount]

返回请求用户本人历史里出现过的去重收款账户（句柄 + 脱敏标签）；无 open_id 时拒绝（绝不枚举他人）。

源代码位于： feishu/agent/payment_accounts.py
Python
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
async def recent(
    self, user: Mapping[str, Any], *, approval_code: str | None = None, limit: int = 10
) -> list[PaymentAccount]:
    r"""返回请求用户本人历史里出现过的去重收款账户（句柄 + 脱敏标签）；无 `open_id` 时拒绝（绝不枚举他人）。"""
    user_keys = payment_account_keys(user)
    open_id = user.get("open_id") if isinstance(user, Mapping) else None
    if not user_keys or not open_id:
        # Without the user's own open_id we cannot scope the query to them — refuse rather than over-read.
        return []
    end = int(time.time() * 1000)
    start = end - self._lookback_ms
    try:
        items = await self._client.approval.instances.query(
            user_id=open_id,
            approval_code=approval_code,
            user_id_type="open_id",
            start_time=str(start),
            end_time=str(end),
            max_items=50,
        )
    except Exception:  # noqa: BLE001 - an account lookup must never crash the tool; degrade to "none found"
        logging.getLogger("feishu").debug("payment account query failed for the requesting user", exc_info=True)
        return []
    accounts: dict[str, PaymentAccount] = {}
    for item in items:
        instance_node = item.get("instance") if isinstance(item, Mapping) else None
        code = instance_node.get("code") if isinstance(instance_node, Mapping) else None
        if not code:
            continue
        try:
            instance = await self._client.approval.instances.get(code)
        except Exception:  # noqa: BLE001 - skip an instance we cannot read rather than failing the whole lookup
            logging.getLogger("feishu").debug(
                "could not read instance %s for payment accounts", code, exc_info=True
            )
            continue
        if not _instance_belongs_to(instance, user):
            continue  # defense in depth: never harvest an instance that is not the requesting user's own
        for widget in approval_account_widgets(instance):
            value = widget.get("value")
            number = approval_account_number(value)
            if not number:
                continue
            handle = payment_account_handle(number)
            if handle not in accounts:
                accounts[handle] = PaymentAccount(
                    account_id=handle,
                    label=approval_account_label(value),
                    account_value=dict(value),
                    user_keys=user_keys,
                )
            if len(accounts) >= limit:
                break
        if len(accounts) >= limit:
            break
    if user_keys:
        self._cache.setdefault(user_keys[0], {}).update(accounts)
    return list(accounts.values())

resolve async

Python
resolve(user: Mapping[str, Any], account_id: str) -> dict[str, Any] | None

把句柄还原为完整账户控件值，严格限定为请求用户本人；缓存未命中时回源一次再试，仍无则返回 None。

源代码位于： feishu/agent/payment_accounts.py
Python
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
async def resolve(self, user: Mapping[str, Any], account_id: str) -> dict[str, Any] | None:
    r"""把句柄还原为完整账户控件值，严格限定为请求用户本人；缓存未命中时回源一次再试，仍无则返回 `None`。"""
    user_keys = payment_account_keys(user)
    if not user_keys:
        return None
    primary = user_keys[0]
    account = self._cache.get(primary, {}).get(account_id)
    if account is None:
        await self.recent(user)  # repopulate (process restart / first use), then retry
        account = self._cache.get(primary, {}).get(account_id)
    if account is None:
        return None
    if not set(user_keys) & set(account.user_keys):
        return None  # the handle must belong to THIS user
    return dict(account.account_value)

payment_account_keys

Python
payment_account_keys(user: Mapping[str, Any]) -> tuple[str, ...]

用户的稳定多别名键；feishu.auth.user_tokens.user_identity_keys 的别名（全库统一表示）。

源代码位于： feishu/agent/payment_accounts.py
Python
46
47
48
def payment_account_keys(user: Mapping[str, Any]) -> tuple[str, ...]:
    r"""用户的稳定多别名键；[feishu.auth.user_tokens.user_identity_keys][] 的别名（全库统一表示）。"""
    return user_identity_keys(user)

payment_account_handle

Python
payment_account_handle(number: str) -> str

把卡号映射为稳定、不可逆的句柄（pa_<sha256[:16]>）；模型只见句柄，绝不见卡号。

源代码位于： feishu/agent/payment_accounts.py
Python
51
52
53
def payment_account_handle(number: str) -> str:
    r"""把卡号映射为稳定、不可逆的句柄（`pa_<sha256[:16]>`）；模型只见句柄，绝不见卡号。"""
    return "pa_" + hashlib.sha256(number.encode("utf-8")).hexdigest()[:16]