"""GitHub 组织成员资格驱动 —— 邀请、移除、枚举。 与同目录的 ``github_software.py`` 是**两个东西**,不是一个的两种模式: 那个走 SCIM 2.0 管席位,需要 ``scim:enterprise``,而且只在开了 SAML 的 Enterprise Cloud 组织上工作;这个走组织 REST 端点管成员资格,需要 ``admin:org`` 级别的授权,在普通组织上也工作。它们的凭证、API 面和语义 都不同,合成一个驱动就必须在某处撒谎。 ## 两种鉴权,同一套调用面 ``auth_mode`` 决定怎么拿到 ``Authorization`` 头: * ``github_app``(推荐)—— 用 App 私钥签一个 10 分钟的 RS256 JWT,再用它 换取**每个组织各自的** installation token(1 小时)。token 按组织缓存, 提前 5 分钟过期。好处是权限最小(只勾 Members: Read and write)、轮换 独立、审计里看得出是哪个 App 干的。 * ``pat`` —— 直接用 personal access token。简单,但那个 token 天然带着 这个人的全部权限,而且这个人离职时它会一起消失。 两种模式下所有业务方法完全一样 —— 上层不需要知道用的是哪一种,这正是 「凭证方式最后再定」这件事能成立的原因。 ## 三条不肯让步的规则 1. **传输失败 ≠ 不存在。** 每一个喂给「要不要写」的查询,都必须把 「GitHub 说没有」和「我们没问到」分开。在这个域里,这条规则的具体 含义是:一次网络抖动绝不能被读成「这个人已经不在组织里了」,然后被 写成「已回收」。平台的驱动规则里已经为 ``github_software`` 记过这条 (``_find_user_id`` 的第三个返回值),这里从第一天就照做。 2. **邀请不是加入。** ``invite_member`` 成功只意味着邀请发出去了。 ``GithubOrgActionResult.invited`` 与 ``already_done`` 因此是两个字段。 3. **凭证从不进日志。** 错误片段落进 ``provisioning_tasks.error_message`` 与集成的 ``last_error_message``,那两处管理员看得见。 """ from __future__ import annotations import asyncio import logging import re import time from typing import Any from urllib.parse import quote as _url_quote import httpx from app.integrations.driver_base import ( GithubOrgActionResult, GithubOrgDriver, GithubOrgInvitation, GithubOrgMember, GithubOrgSeatUsage, GithubUserProfile, ) from app.integrations.registry import register_driver from app.integrations.schemas import ( IntegrationFieldInfo, IntegrationTestResult, ) logger = logging.getLogger(__name__) _DEFAULT_BASE = "https://api.github.com" _HTTP_TIMEOUT = httpx.Timeout(30.0, connect=10.0) _API_VERSION = "2022-11-28" #: 退避阶梯。三次重试足够穿过一次二级限流,又不会让一个离职 tick 卡住。 _RETRY_BACKOFF_SECONDS = (1.0, 3.0, 8.0) #: 值得重试的状态码。**403 不在里面** —— GitHub 用 403 同时表达「权限不足」 #: 和「二级限流」,只有带 ``Retry-After`` 或 ``x-ratelimit-remaining: 0`` #: 的那种才是后者,由 ``_is_retryable`` 单独判。 _TRANSIENT_STATUSES = frozenset({429, 500, 502, 503, 504}) #: installation token 提前多久算过期。GitHub 给 1 小时,提前 5 分钟换新的, #: 免得一个长事务在 token 到期那一刻发出请求。 _TOKEN_SKEW_SECONDS = 300 #: installation token 缓存里代表「随便哪一个 installation」的键。组织名 #: 一律小写,而 GitHub 的组织名不含 ``*``,所以它不可能撞上一个真组织。 _ANY_INSTALLATION = "*" #: 一次分页最多抓多少页。GitHub 每页最多 100,100 页 = 10000 个成员 —— #: 远超任何真实组织,但它保证一个坏掉的 Link 头不会让 worker 无限转。 _MAX_PAGES = 100 _LINK_NEXT_RE = re.compile(r'<([^>]+)>\s*;\s*rel="next"') #: 只有这些计划的 ``plan.seats`` 是「这个组织的、我们能动的」上限。 #: #: 实测 2026-08-31(生产 5 个组织):``free`` 计划报 ``seats=92`` 而 #: ``filled_seats=203`` —— 那个 92 不是上限,按字面算余量是 -111,会变成一个 #: 永久假警报;``enterprise`` 报的是整个 enterprise 的池子(tikv 9901/50000 #: 是 CNCF 的),既不是这个组织的账单也不是我们能动的东西。 _SEAT_CAPPED_PLANS = frozenset({"team", "business", "business_plus"}) #: GitHub 用 403 同时表达「权限不足」和「坐席用满」。这几个片段用来把后者 #: 分出来 —— 它的处置完全不同:权限不足去修集成,坐席用满要 org owner / #: billing manager 去加坐席。 #: #: ⚠️ 片段是 GitHub 的**文案**,不是稳定契约。文案改了这里就认不出来, #: 那时它会退回成一条普通的 permission_denied 失败(仍然不静默、仍然有 #: last_error),而不是产生一个错误的结论 —— 这个降级方向是刻意的。 _SEAT_LIMIT_MARKERS = ( "seat", "licensed", "license limit", "exceeded your", "add more seats", "purchase", "billing", ) def _redact(blob: str, *known_secrets: str) -> str: """把凭证从错误片段里抹掉。 与 ``github_software._redact_secrets`` 同款,额外多抹 PEM 私钥体 —— GitHub App 模式下配置里躺着一整段私钥,一次 traceback 就能把它带进 管理员看得见的错误字段。 """ if not blob: return blob for secret in known_secrets: if secret and len(secret) >= 8: blob = blob.replace(secret, "***") blob = re.sub( r"(?i)(authorization\s*:?\s*(?:bearer|token|basic)\s+)[A-Za-z0-9._\-+/=]+", r"\1***", blob, ) for prefix in ("ghp_", "ghs_", "ghu_", "gho_", "ghr_"): blob = re.sub(rf"\b{prefix}[A-Za-z0-9]{{20,}}\b", f"{prefix}***", blob) blob = re.sub(r"\bgithub_pat_[A-Za-z0-9_]{20,}\b", "github_pat_***", blob) # PEM 私钥:整段换掉,不留头尾 —— 留着头尾等于告诉读日志的人这里曾经 # 有一把私钥,而抹掉的只是中间。 blob = re.sub( r"-----BEGIN [A-Z ]*PRIVATE KEY-----.*?-----END [A-Z ]*PRIVATE KEY-----", "***PRIVATE KEY REDACTED***", blob, flags=re.DOTALL, ) return blob def _seg(value: str) -> str: """URL 路径段转义。 组织名与 handle 都来自我们自己的库,但它们**曾经**来自用户输入(自助 申请的表单)。一个带 ``../`` 或 ``%2F`` 的值拼进路径,就是一次指向别的 端点的请求 —— 转义在这里不是形式主义。 """ return _url_quote(str(value), safe="") class _GithubOrgApiError(RuntimeError): """带状态码的 API 失败,用于把「问不到」与「答案是没有」分开。""" def __init__(self, message: str, *, status: int | None = None) -> None: super().__init__(message) self.status = status @register_driver class GithubOrgApiDriver(GithubOrgDriver): provider = "github" display_name = "GitHub 组织" description = ( "GitHub 组织成员资格 —— 邀请入组、移出组织、枚举成员与待接受邀请。" "支持两种鉴权:GitHub App(推荐,权限最小、可独立轮换)与 " "Personal Access Token。一个凭证可以覆盖多个组织,组织清单来自 " "Helix 的 GitHub 组织登记表,不写在这里。" ) supports_multi_instance = True config_schema = [ IntegrationFieldInfo( key="auth_mode", label="鉴权方式", is_secret=False, required=True, default="github_app", placeholder="github_app", hint=( "github_app 或 pat。推荐 github_app:权限可以只勾 " "Members: Read and write,轮换与其它系统互不牵连。" ), ), IntegrationFieldInfo( key="app_id", label="GitHub App ID", is_secret=False, required=False, placeholder="123456", hint="鉴权方式为 github_app 时必填。App 设置页顶部的 App ID。", ), IntegrationFieldInfo( key="private_key", label="GitHub App 私钥 (PEM)", is_secret=True, required=False, placeholder="-----BEGIN RSA PRIVATE KEY-----", hint=( "鉴权方式为 github_app 时必填。App 设置页生成的 .pem 文件" "全文,含 BEGIN/END 两行。" ), ), IntegrationFieldInfo( key="token", label="Personal Access Token", is_secret=True, required=False, placeholder="ghp_... / github_pat_...", hint=( "鉴权方式为 pat 时必填,需要 admin:org 权限。" "注意:它带着这个账号的全部权限,且账号注销时一起失效。" ), ), IntegrationFieldInfo( key="base_url", label="API Base URL", is_secret=False, required=False, default=_DEFAULT_BASE, placeholder=_DEFAULT_BASE, hint="通常留空;GitHub Enterprise Server 自托管才需要改。", ), IntegrationFieldInfo( key="probe_org", label="连接测试用的组织", is_secret=False, required=False, placeholder="pingcap-inc", hint=( "「测试连接」时拿它验证凭证真的能读这个组织的成员。" "留空则只验证凭证本身有效(App 模式列安装,PAT 模式查 /user)。" ), ), ] binding_schema = [] def __init__(self, *args: Any, **kwargs: Any) -> None: super().__init__(*args, **kwargs) #: 按组织缓存的 installation token:``org_lower -> (token, expires_at)`` self._inst_tokens: dict[str, tuple[str, float]] = {} self._installations: dict[str, int] | None = None # ── 配置读取 ────────────────────────────────────────────────── @property def _base(self) -> str: return (self.config.get("base_url") or _DEFAULT_BASE).rstrip("/") @property def _auth_mode(self) -> str: raw = (self.config.get("auth_mode") or "github_app").strip().lower() return raw if raw in ("github_app", "pat") else "github_app" def _known_secrets(self) -> tuple[str, ...]: return ( str(self.config.get("token") or ""), str(self.config.get("private_key") or ""), ) # ── 鉴权 ────────────────────────────────────────────────────── def _app_jwt(self) -> str: """给 App 签一个短命 JWT。 ``iat`` 回拨 60 秒:GitHub 会拒绝 ``iat`` 在未来的 JWT,而两台机器 的时钟差几十秒是常态。这是 GitHub 官方文档自己给的建议,不是我们 在掩盖什么。 """ import jwt as _jwt # noqa: PLC0415 app_id = str(self.config.get("app_id") or "").strip() private_key = str(self.config.get("private_key") or "") if not app_id or not private_key: raise _GithubOrgApiError( "GitHub App auth requires both app_id and private_key", ) now = int(time.time()) return _jwt.encode( {"iat": now - 60, "exp": now + 540, "iss": app_id}, private_key, algorithm="RS256", ) async def _installation_id(self, org: str) -> int: """这个组织上装的是哪个 installation。 用 ``GET /orgs/{org}/installation`` 直接问,而不是列出全部 installation 再在里面找:后者在 App 装了几十个组织时要翻页,而且「列表里没有」 与「这个组织没装」是同一个结果却少一条明确的 404。 """ key = org.lower() if self._installations and key in self._installations: return self._installations[key] resp = await self._raw_request( "GET", f"/orgs/{_seg(org)}/installation", auth_header=f"Bearer {self._app_jwt()}", ) if resp.status_code == 404: raise _GithubOrgApiError( f"GitHub App is not installed on organization '{org}'", status=404, ) if resp.status_code >= 400: raise _GithubOrgApiError( f"installation lookup failed: HTTP {resp.status_code} " f"{_redact(resp.text[:300], *self._known_secrets())}", status=resp.status_code, ) inst_id = int(resp.json()["id"]) self._installations = {**(self._installations or {}), key: inst_id} return inst_id async def _any_installation_id(self) -> int: """随便哪一个 installation 的 id —— 给**与组织无关**的端点用。 ``GET /users/{login}`` 读的是公开资料,不属于任何组织,但在 App 模式 下它仍然需要一个 **installation token**:App JWT 只对 ``/app*`` 有效, 打在 ``/users/`` 上 GitHub 返回 401 ``Bad credentials`` —— 一个看起来 像「token 过期了」的错误,而私钥其实完全正确。生产实测(2026-09-01): 同一个端点,App JWT 401,installation token 200。 这里刻意**不**要求调用方给一个组织:哪个 installation 都能读公开 资料,而逼调用方选一个,等于让「读某人的公开资料」这件事依赖一个 与它无关的组织参数 —— 那个参数迟早会被填错或填成一个 App 没装的 组织,然后这条路径以 404 的形态死掉。 """ cached = (self._installations or {}).get(_ANY_INSTALLATION) if cached: return cached resp = await self._raw_request( "GET", "/app/installations", auth_header=f"Bearer {self._app_jwt()}", params={"per_page": 1}, ) if resp.status_code >= 400: raise _GithubOrgApiError( f"installation list failed: HTTP {resp.status_code} " f"{_redact(resp.text[:300], *self._known_secrets())}", status=resp.status_code, ) rows = resp.json() or [] if not rows: raise _GithubOrgApiError( "GitHub App is not installed anywhere, so it cannot read " "public user profiles", status=404, ) inst_id = int(rows[0]["id"]) self._installations = { **(self._installations or {}), _ANY_INSTALLATION: inst_id, } return inst_id async def _auth_header( self, org: str | None, *, any_installation: bool = False, ) -> str: if self._auth_mode == "pat": token = str(self.config.get("token") or "").strip() if not token: raise _GithubOrgApiError("PAT auth requires a token") return f"Bearer {token}" if org is None and not any_installation: # 没有组织上下文时(例如只验证凭证本身),App JWT 就够了 —— # 它能读 /app,不能读组织成员,这正是这条路径要的。 return f"Bearer {self._app_jwt()}" # 与组织无关、但仍然要 installation token 的端点(公开资料)。 key = _ANY_INSTALLATION if org is None else org.lower() cached = self._inst_tokens.get(key) if cached and cached[1] - _TOKEN_SKEW_SECONDS > time.time(): return f"Bearer {cached[0]}" inst_id = ( await self._any_installation_id() if org is None else await self._installation_id(org) ) resp = await self._raw_request( "POST", f"/app/installations/{inst_id}/access_tokens", auth_header=f"Bearer {self._app_jwt()}", ) if resp.status_code >= 400: raise _GithubOrgApiError( f"installation token exchange failed: HTTP {resp.status_code} " f"{_redact(resp.text[:300], *self._known_secrets())}", status=resp.status_code, ) body = resp.json() token = body["token"] # ``expires_at`` 是 ISO8601。解析失败就按 1 小时算 —— 保守方向是 # 更早换新的,不是更晚。 expires = time.time() + 3600 raw_exp = body.get("expires_at") if raw_exp: try: from datetime import datetime # noqa: PLC0415 expires = datetime.fromisoformat( str(raw_exp).replace("Z", "+00:00"), ).timestamp() except (ValueError, TypeError): pass self._inst_tokens[key] = (token, expires) return f"Bearer {token}" # ── HTTP ────────────────────────────────────────────────────── @staticmethod def _is_retryable(resp: httpx.Response) -> bool: if resp.status_code in _TRANSIENT_STATUSES: return True # GitHub 用 403 同时表达「权限不足」和「二级限流」。只有配额确实 # 见底、或者明说了 Retry-After 的那种才值得重试;把两者混为一谈 # 会让一个凭证配错的集成在每次调用上白等 12 秒。 if resp.status_code == 403: if resp.headers.get("retry-after"): return True if resp.headers.get("x-ratelimit-remaining") == "0": return True return False async def _raw_request( self, method: str, path: str, *, auth_header: str, json_body: Any | None = None, params: dict[str, Any] | None = None, retry_on_transient: bool = True, ) -> httpx.Response: url = path if path.startswith("http") else f"{self._base}{path}" from app.integrations._url_safety import ( # noqa: PLC0415 PinnedAsyncHTTPTransport, resolve_safe_external_url_async, ) target = await resolve_safe_external_url_async(url, allowed_suffixes=None) headers = { "Authorization": auth_header, "Accept": "application/vnd.github+json", "X-GitHub-Api-Version": _API_VERSION, } if json_body is not None: headers["Content-Type"] = "application/json" attempts = len(_RETRY_BACKOFF_SECONDS) + 1 resp: httpx.Response | None = None for attempt in range(attempts): transport = PinnedAsyncHTTPTransport(target) async with httpx.AsyncClient( timeout=_HTTP_TIMEOUT, trust_env=False, transport=transport, ) as client: resp = await client.request( method, url, headers=headers, json=json_body, params=params, ) if ( retry_on_transient and attempt < len(_RETRY_BACKOFF_SECONDS) and self._is_retryable(resp) ): backoff = _RETRY_BACKOFF_SECONDS[attempt] retry_after = resp.headers.get("retry-after") if retry_after: try: backoff = max(backoff, float(retry_after)) except ValueError: pass logger.warning( "github_org %s %s transient HTTP %s (attempt %d/%d); " "backing off %.1fs", method, path, resp.status_code, attempt + 1, attempts, backoff, ) await asyncio.sleep(backoff) continue return resp assert resp is not None # noqa: S101 — 循环至少跑一次 return resp async def _request( self, method: str, path: str, *, org: str | None, json_body: Any | None = None, params: dict[str, Any] | None = None, any_installation: bool = False, ) -> httpx.Response: return await self._raw_request( method, path, auth_header=await self._auth_header( org, any_installation=any_installation, ), json_body=json_body, params=params, ) async def _paged( self, path: str, *, org: str, params: dict[str, Any] | None = None, ) -> list[dict[str, Any]]: """按 Link 头翻页,聚合所有结果。 页数封顶(``_MAX_PAGES``)并在触顶时 **warning**:静默截断会让 「这个组织有多少人」变成一个偷偷变小的数,而对账正是靠这个数 判断差异。 """ out: list[dict[str, Any]] = [] query = {"per_page": 100, **(params or {})} url: str | None = path pages = 0 while url and pages < _MAX_PAGES: resp = await self._request( "GET", url, org=org, params=query if pages == 0 else None, ) if resp.status_code >= 400: raise _GithubOrgApiError( f"GET {path} failed: HTTP {resp.status_code} " f"{_redact(resp.text[:300], *self._known_secrets())}", status=resp.status_code, ) body = resp.json() if isinstance(body, list): out.extend(body) pages += 1 link = resp.headers.get("link") or "" match = _LINK_NEXT_RE.search(link) url = match.group(1) if match else None if url: logger.warning( "github_org paging hit the %d-page cap on %s (org=%s) — " "the result is TRUNCATED and reconciliation built on it " "will under-count", _MAX_PAGES, path, org, ) return out # ── 读 ──────────────────────────────────────────────────────── async def list_members( self, org: str, *, role: str | None = None, ) -> list[GithubOrgMember]: params: dict[str, Any] = {} if role in ("admin", "member"): params["role"] = role rows = await self._paged(f"/orgs/{_seg(org)}/members", org=org, params=params) # ``/members`` 不返回每个人的角色,要角色得单独问 ``?role=admin`` # 再做差集。只有调用方明确要 admin 时才多发这一次请求。 admin_logins: set[str] = set() if role is None: admin_rows = await self._paged( f"/orgs/{_seg(org)}/members", org=org, params={"role": "admin"}, ) admin_logins = { str(r.get("login", "")).lower() for r in admin_rows if r.get("login") } return [ GithubOrgMember( login=str(r["login"]), state="active", role=( "admin" if (role == "admin" or str(r["login"]).lower() in admin_logins) else "member" ), user_id=int(r["id"]) if r.get("id") is not None else None, ) for r in rows if r.get("login") ] async def list_pending_invitations( self, org: str, ) -> list[GithubOrgInvitation]: rows = await self._paged(f"/orgs/{_seg(org)}/invitations", org=org) out: list[GithubOrgInvitation] = [] for r in rows: inv_id = r.get("id") if inv_id is None: continue out.append( GithubOrgInvitation( invitation_id=str(inv_id), login=(str(r["login"]) if r.get("login") else None), email=(str(r["email"]) if r.get("email") else None), role=str(r.get("role") or "direct_member"), created_at=( str(r["created_at"]) if r.get("created_at") else None ), failed_reason=( str(r["failed_reason"]) if r.get("failed_reason") else None ), ) ) return out async def get_user(self, login: str) -> GithubOrgMember | None: # ⚠️ ``any_installation=True`` 不是可选的优化:App 模式下 org=None # 会拿到 App JWT,而 App JWT 打 /users/ 是 401(见 # ``_any_installation_id`` 的文档)。整条「靠公开资料认出这个 # handle 是谁」的链路都挂在这一个参数上。 resp = await self._request( "GET", f"/users/{_seg(login)}", org=None, any_installation=True, ) if resp.status_code == 404: return None if resp.status_code >= 400: # 明确抛,不返回 None。"我们没问到"绝不能长得像"这个人不存在" # —— 后者会让一张审批通过的单子被判成 handle 打错。 raise _GithubOrgApiError( f"user lookup failed: HTTP {resp.status_code} " f"{_redact(resp.text[:300], *self._known_secrets())}", status=resp.status_code, ) body = resp.json() return GithubOrgMember( login=str(body.get("login") or login), state="unknown", role="member", user_id=int(body["id"]) if body.get("id") is not None else None, ) async def get_user_profile(self, login: str) -> GithubUserProfile | None: # 同一个坑,见 ``get_user`` 上那段注释:App JWT 打 /users/ 是 401。 resp = await self._request( "GET", f"/users/{_seg(login)}", org=None, any_installation=True, ) if resp.status_code == 404: return None if resp.status_code >= 400: raise _GithubOrgApiError( f"user profile lookup failed: HTTP {resp.status_code} " f"{_redact(resp.text[:300], *self._known_secrets())}", status=resp.status_code, ) body = resp.json() or {} def _s(key: str) -> str | None: # GitHub 把没填的字段返回成 null,但也见过返回空串和纯空白的。 # 三种都要归到 None:界面上「company 是空串」和「没填 company」 # 应该长得一样,否则人会以为拉到了什么。 v = body.get(key) if not isinstance(v, str): return None v = v.strip() return v or None return GithubUserProfile( login=str(body.get("login") or login), user_id=int(body["id"]) if body.get("id") is not None else None, name=_s("name"), company=_s("company"), email=_s("email"), bio=_s("bio"), location=_s("location"), blog=_s("blog"), account_type=_s("type"), ) async def list_teams(self, org: str) -> list[dict[str, Any]]: rows = await self._paged(f"/orgs/{_seg(org)}/teams", org=org) return [ { "id": r.get("id"), "slug": r.get("slug"), "name": r.get("name"), "privacy": r.get("privacy"), } for r in rows ] # ── 写 ──────────────────────────────────────────────────────── @staticmethod def _github_role(role: str) -> str: """Helix 的 member/admin → GitHub membership API 的取值。 `PUT /orgs/{org}/memberships/{username}` 收的是 ``member`` / ``admin``,而 `/invitations` 收的是 ``direct_member`` / ``admin``。 两套词表,所以这里显式转换而不是原样透传。 """ return "admin" if role == "admin" else "member" async def invite_member( self, org: str, login: str, *, role: str = "member", ) -> GithubOrgActionResult: """``PUT /orgs/{org}/memberships/{username}``。 选它而不是 ``POST /orgs/{org}/invitations``,有两个理由:它按 **handle** 而不是数字 user id 定位(省掉一次查人),而且它天然幂等 —— 对方已经是成员时返回 ``state=active`` 而不是报错,对方已有待接受 邀请时重复调用也只是刷新那张邀请。这两点正是「飞书重投一次回调」 需要的性质。 """ try: resp = await self._request( "PUT", f"/orgs/{_seg(org)}/memberships/{_seg(login)}", org=org, json_body={"role": self._github_role(role)}, ) except _GithubOrgApiError as e: return GithubOrgActionResult( ok=False, error_message=_redact(str(e), *self._known_secrets()), permission_denied=(e.status in (401, 403)), ) if resp.status_code == 404: # 这个 handle 在 GitHub 上不存在。一个真实且可行动的答案 —— # 审批单上的 handle 打错了,该去找申请人,不该重试。 return GithubOrgActionResult( ok=False, not_found=True, error_message=f"GitHub user '{login}' does not exist", ) if resp.status_code in (401, 403, 422): body = (resp.text or "").lower() if self._looks_like_seat_limit(resp.status_code, body): # 坐席用满 —— 与权限不足分开。处置完全不同:这个要 org # owner / billing manager 去加坐席,IT 查凭证是白费功夫。 return GithubOrgActionResult( ok=False, seat_limit=True, error_message=_redact( f"organization '{org}' has no seat available for " f"{login}: HTTP {resp.status_code} {resp.text[:200]}", *self._known_secrets(), ), ) if resp.status_code == 422: return GithubOrgActionResult( ok=False, error_message=_redact( f"HTTP 422 {resp.text[:300]}", *self._known_secrets(), ), ) return GithubOrgActionResult( ok=False, permission_denied=True, error_message=_redact( f"not authorised to manage members of '{org}' " f"(HTTP {resp.status_code}) {resp.text[:200]}", *self._known_secrets(), ), ) if resp.status_code >= 400: return GithubOrgActionResult( ok=False, error_message=_redact( f"HTTP {resp.status_code} {resp.text[:300]}", *self._known_secrets(), ), ) payload: dict[str, Any] = resp.json() if resp.content else {} state = str(payload.get("state") or "").lower() return GithubOrgActionResult( ok=True, external_ref=str((payload.get("user") or {}).get("id") or "") or None, # state=active 意味着这个人本来就在组织里(GitHub 不会因为一次 # PUT 就把 pending 变 active),所以它是幂等命中而不是新加入。 already_done=(state == "active"), invited=(state == "pending"), ) async def remove_member( self, org: str, login: str, ) -> GithubOrgActionResult: """``DELETE /orgs/{org}/memberships/{username}``。 这一个调用同时收掉活跃成员资格**和**待接受的邀请 —— GitHub 的 membership 资源涵盖两种状态,所以不需要先查再分支。 404 读作「已经不在里面了」,是幂等成功。这条读法只在**响应确实 到达**时才成立:一次连接失败没有告诉我们任何关于世界的事,把它 读成「已回收」正是那条平台级驱动规则要防的东西。 """ try: resp = await self._request( "DELETE", f"/orgs/{_seg(org)}/memberships/{_seg(login)}", org=org, ) except _GithubOrgApiError as e: return GithubOrgActionResult( ok=False, error_message=_redact(str(e), *self._known_secrets()), permission_denied=(e.status in (401, 403)), ) except httpx.HTTPError as e: return GithubOrgActionResult( ok=False, error_message=_redact( f"{type(e).__name__}: {e}", *self._known_secrets(), ), ) if resp.status_code in (204, 200): return GithubOrgActionResult(ok=True) if resp.status_code == 404: return GithubOrgActionResult(ok=True, already_done=True) if resp.status_code in (401, 403): return GithubOrgActionResult( ok=False, permission_denied=True, error_message=( f"not authorised to manage members of '{org}' " f"(HTTP {resp.status_code})" ), ) return GithubOrgActionResult( ok=False, error_message=_redact( f"HTTP {resp.status_code} {resp.text[:300]}", *self._known_secrets(), ), ) async def cancel_invitation( self, org: str, invitation_id: str, ) -> GithubOrgActionResult: resp = await self._request( "DELETE", f"/orgs/{_seg(org)}/invitations/{_seg(invitation_id)}", org=org, ) if resp.status_code in (204, 200): return GithubOrgActionResult(ok=True) if resp.status_code == 404: return GithubOrgActionResult(ok=True, already_done=True) return GithubOrgActionResult( ok=False, permission_denied=(resp.status_code in (401, 403)), error_message=_redact( f"HTTP {resp.status_code} {resp.text[:300]}", *self._known_secrets(), ), ) @staticmethod def _looks_like_seat_limit(status: int, body_lower: str) -> bool: """403/422 是不是「坐席用满」而不是「权限不足」。 按**响应正文的文案片段**判,因为 GitHub 没给这两者不同的状态码。 认不出来时返回 False —— 退化成一条普通的权限失败,仍然响亮、仍然 带 ``last_error``,只是通知发给了 IT 而不是 org owner。宁可通知错 对象,也不要因为文案变了就把一次失败读成成功。 """ if status not in (403, 422): return False return any(m in body_lower for m in _SEAT_LIMIT_MARKERS) async def get_seat_usage(self, org: str) -> GithubOrgSeatUsage: """``GET /orgs/{org}`` 的 ``plan`` 块。 ``members:write`` 的 App 就能读到 —— 所以坐席不足是**可以主动监控** 的,不必等一次邀请失败才发现。 """ resp = await self._request("GET", f"/orgs/{_seg(org)}", org=org) if resp.status_code >= 400: raise _GithubOrgApiError( f"seat usage lookup failed for {org}: HTTP " f"{resp.status_code} " f"{_redact(resp.text[:200], *self._known_secrets())}", status=resp.status_code, ) plan = (resp.json() or {}).get("plan") or {} name = str(plan.get("name") or "") or None filled = plan.get("filled_seats") total = plan.get("seats") return GithubOrgSeatUsage( plan=name, filled_seats=int(filled) if isinstance(filled, int) else None, total_seats=int(total) if isinstance(total, int) else None, has_cap=bool(name and name.lower() in _SEAT_CAPPED_PLANS), ) # ── 健康探针 ────────────────────────────────────────────────── async def test_connection(self) -> IntegrationTestResult: """验证凭证,并且在给了 ``probe_org`` 时验证它**对那个组织**有效。 分两级是有意的:一个 App 私钥可能完全正确,却没装在你关心的那个 组织上 —— 只验证凭证本身会给出一个绿灯,然后第一次真实邀请才失败。 """ secrets = self._known_secrets() probe_org = (self.config.get("probe_org") or "").strip() try: if self._auth_mode == "pat": resp = await self._request("GET", "/user", org=None) if resp.status_code >= 400: return IntegrationTestResult( ok=False, message=_redact( f"PAT rejected: HTTP {resp.status_code} " f"{resp.text[:200]}", *secrets, ), detail={"auth_mode": "pat"}, ) identity = str(resp.json().get("login") or "?") summary = f"PAT OK · 身份 {identity}" else: resp = await self._raw_request( "GET", "/app", auth_header=f"Bearer {self._app_jwt()}", ) if resp.status_code >= 400: return IntegrationTestResult( ok=False, message=_redact( f"App credentials rejected: HTTP " f"{resp.status_code} {resp.text[:200]}", *secrets, ), detail={"auth_mode": "github_app"}, ) identity = str(resp.json().get("slug") or "?") summary = f"GitHub App OK · {identity}" if not probe_org: return IntegrationTestResult( ok=True, message=( f"{summary} · 未填「连接测试用的组织」,只验证了凭证" "本身 —— 凭证有效不等于它对某个具体组织有权限。" ), detail={"auth_mode": self._auth_mode, "probe_org": None}, ) members = await self.list_members(probe_org, role="admin") # 🔴 再探一次「读公开资料」这条路 —— 它用的是**另一把**凭证。 # # 2026-09-01 的事故形状:App 私钥完全正确,``/app`` 200、组织成员 # 可读、离职对账每天正常跑 —— 而 ``/users/{login}`` 全部 401, # 因为那条路径当时传的是 App JWT。连接测试从头到尾绿灯,代价是 # 「靠公开资料认出这个 handle 是谁」在生产上一年也没工作过, # 而界面上只表现为「规则命中 0 个」。 # # 探针失败**不**把整体判成红:公开资料只影响「认人」,不影响 # 邀请与回收,把它算成致命会让一次 GitHub 抖动挡住真正要紧的 # 那两件事。它作为一句独立结论出现在消息里,让绿灯说得出自己 # 到底验了什么。 profile_note = "" try: probe = await self.get_user_profile("octocat") profile_note = ( " · 公开资料可读" if probe is not None else " · ⚠️ 公开资料探针拿不到 octocat" ) except Exception as e: # noqa: BLE001 profile_note = ( f" · ⚠️ 公开资料读不了({type(e).__name__})—— " "「认出这个 handle 是谁」会退化成只能人工判断" ) return IntegrationTestResult( ok=True, message=( f"{summary} · 组织 {probe_org} 可读," f"{len(members)} 名 owner{profile_note}" ), detail={ "auth_mode": self._auth_mode, "probe_org": probe_org, "owner_count": len(members), "profile_readable": "⚠️" not in profile_note, }, ) except Exception as e: # noqa: BLE001 — 探针必须返回结果而不是抛 return IntegrationTestResult( ok=False, message=_redact(f"{type(e).__name__}: {e}", *secrets), detail={"auth_mode": self._auth_mode, "type": type(e).__name__}, )