feat(auth): add native AgentID sign-in and account binding

deploy/login.example.toml

@@ -13,6 +13,15 @@ client_id = "REPLACE_WITH_GOOGLE_CLIENT_ID.apps.googleusercontent.com"
 credential_file = "/etc/msgd/google-login-client.secret"
 token_auth_method = "client_secret_post"
 
+# Registered AgentID client. Bind to an existing MSG account before signing in.
+# Redirect: https://YOUR_HOST/-/login/callback/agentid
+# Initiate login: https://YOUR_HOST/login/agentid
+[login.providers.agentid]
+enabled = false
+client_id = "REPLACE_WITH_AGENTID_CLIENT_ID"
+credential_file = "/etc/msgd/agentid-login-client.secret"
+token_auth_method = "client_secret_basic"
+
 [login.providers.github]
 enabled = false
 client_id = "REPLACE_WITH_GITHUB_CLIENT_ID"

docs/ACCOUNT_LOGIN.md

@@ -7,6 +7,7 @@
 | Google | 允许，使用平台托管密钥 | 允许 | 进入原账号 |
 | 邮箱 | 允许，使用平台托管密钥 | 允许 | 进入原账号 |
 | GitHub | 禁止 | 允许 | 进入原账号 |
+| AgentID | 禁止 | 允许 | 进入原账号 |
 | Passkey | 禁止 | 允许，并可移除 | 进入原账号 |
 | ChatGPT | 禁止 | 有独立获准的网站身份客户端时允许 | 进入原账号 |
 
@@ -48,6 +49,31 @@ GitHub 使用数值 `id`，不能使用可变的 `login`。公开身份无需额
 
 当前 adapter 的 Google 授权范围为 `openid profile email`，GitHub 为 `read:user`，不请求 `user:email`，不读取私有邮箱。示例不提供 `scopes` 配置字段；上述最小 scope 说明用于核对提供方权限。
 
+### AgentID
+
+AgentID 使用本站原生登录和账号绑定，不需要 Better Auth 或新数据库表。
+在 `[login.providers.agentid]` 配置本站注册的 `client_id`、原始 secret 的
+`credential_file`，以及控制台选择的 `token_auth_method`（默认 `client_secret_basic`）。
+生产回调为 `https://msg.lmm.best/-/login/callback/agentid`；登录页为
+`https://msg.lmm.best/login`；控制台的 Initiate login URL 为
+`https://msg.lmm.best/login/agentid`。自托管实例替换域名并单独登记回调。
+不要使用 Better Auth 示例的 `/api/auth/callback/agentid`。
+
+在 `/account/login-methods` 将 AgentID 绑定到已有 MSG 账号后，即可用它登录。
+此接入保持现有注册政策：仅 Google／邮箱可创建账号；AgentID 不会自动创建、
+按邮箱合并账号，或改变已有自持密钥与权限。只请求 `openid profile email`，
+不请求 owner scopes、`offline_access` 或刷新凭据；绑定键仅使用固定 issuer 和 `sub`。
+
+服务端校验发现文档、ES256/P-256 签名、issuer、audience、到期时间、nonce，
+并验证回调的 `iss`。登录使用 PKCE S256；`state` 和加密 verifier 单次有效，
+从发起登录起保留十分钟，支持 AgentID 的等待页。控制台传入的 `login_hint`
+仅转发到固定授权端点，不能指定任意返回地址、绑定目标或权限。
+
+secret 只保存在服务账号可读的独立私有文件中，不能进入仓库或浏览器。
+登录入口默认关闭；代码测试不表示生产已启用，也不能代替真实 AgentID 登录。
+详见 [AgentID 自定义接入](https://www.agentid.com/docs/custom)与
+[签名验证](https://www.agentid.com/docs)。
+
 ### ChatGPT 网站身份登录
 
 截至 2026-10-05，OpenAI 已公开第三方网站的 Sign in with ChatGPT 文档，范围仍是选定商业合作方的有限试用；网站需申请自己的客户端和精确回调。它并非仅用于 Codex，也不是任何网站均可自行注册启用。[网站接入](https://developers.openai.com/siwc/website)、[客户端申请](https://developers.openai.com/siwc/request-client-id)。

src/msg/login_config.py

@@ -6,7 +6,8 @@
 
 from msg.core.errors import require
 
-PROVIDERS = frozenset({'google', 'github', 'chatgpt', 'email', 'passkey'})
+OAUTH_PROVIDERS = frozenset({'google', 'github', 'chatgpt', 'agentid'})
+PROVIDERS = OAUTH_PROVIDERS | {'email', 'passkey'}
 REGISTER_PROVIDERS = frozenset({'google', 'email'})
 
 
@@ -54,7 +55,7 @@ def load_login(data, service):
         item = raw.get(name, {})
         allowed = (
             {'enabled', 'client_id', 'credential_file', 'token_auth_method'}
-            if name in {'google', 'github', 'chatgpt'}
+            if name in OAUTH_PROVIDERS
             else {'enabled', 'transport', 'credential_file', 'sender'}
             if name == 'email'
             else {'enabled'}
@@ -64,7 +65,12 @@ def load_login(data, service):
         require(type(active) is bool, 'invalid_login_provider')
         client_id, credential_file = item.get('client_id', ''), item.get('credential_file', '')
         token_auth_method = item.get(
-            'token_auth_method', 'client_secret_post' if name in {'google', 'github'} else 'none'
+            'token_auth_method',
+            {
+                'google': 'client_secret_post',
+                'github': 'client_secret_post',
+                'agentid': 'client_secret_basic',
+            }.get(name, 'none'),
         )
         transport, sender = item.get('transport', 'sequenzy'), item.get('sender', '')
         require(
@@ -85,7 +91,17 @@ def load_login(data, service):
             and isinstance(sender, str)
             and len(sender) <= 512
             and not any(ord(c) < 32 for c in sender)
-            and (name != 'chatgpt' or not active or client_id.startswith('oaiapp_')),
+            and (name != 'chatgpt' or not active or client_id.startswith('oaiapp_'))
+            and (
+                name != 'agentid'
+                or not active
+                or (
+                    client_id
+                    and client_id.isascii()
+                    and credential_file
+                    and token_auth_method in {'client_secret_basic', 'client_secret_post'}
+                )
+            ),
             'invalid_login_provider',
         )
         providers.append(

src/msg/security/login_flow.py

@@ -9,6 +9,7 @@
 
 from msg.core.codec import b64, unb64
 from msg.core.errors import Failure, require
+from msg.login_config import OAUTH_PROVIDERS
 from msg.security.oauth import get, put, save, secret, state_id
 
 TTL = 600
@@ -33,7 +34,7 @@ def _open(app, state, value):
 
 
 def start_flow(app, tx, provider, browser, *, mode='login', source=None, handle=None):
-    require(provider in {'google', 'github', 'chatgpt'}, 'login_provider_disabled')
+    require(provider in OAUTH_PROVIDERS, 'login_provider_disabled')
     require(mode in {'login', 'register', 'bind'}, 'invalid_login_mode')
     require(mode != 'register' or provider == 'google', 'login_registration_forbidden')
     require(mode != 'bind' or source is not None, 'authentication_required')

src/msg/security/login_jwt.py

@@ -12,6 +12,7 @@
 from datetime import datetime
 
 import jwt
+from cryptography.hazmat.primitives.asymmetric.ec import SECP256R1, EllipticCurvePublicKey
 from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey
 
 from msg.core.errors import Failure
@@ -68,14 +69,15 @@ def _token_parts(token: str) -> tuple[str, str, str]:
     return parts[0], parts[1], parts[2]
 
 
-def token_key_id(token: str) -> str:
-    """Read only a bounded RS256 key ID; header URLs never cause a network request."""
+def token_key_id(token: str, *, algorithm: str = 'RS256') -> str:
+    """Read a bounded key ID with the caller-pinned algorithm; header URLs never cause a network request."""
     try:
         header_part, _, _ = _token_parts(token)
         header = _json_part(header_part, limit=_MAX_HEADER_BYTES)
         key_id = header.get('kid')
         if (
-            header.get('alg') != 'RS256'
+            algorithm not in {'RS256', 'ES256'}
+            or header.get('alg') != algorithm
             or 'crit' in header
             or header.get('b64', True) is not True
             or not isinstance(key_id, str)
@@ -87,7 +89,9 @@ def token_key_id(token: str) -> str:
         raise _invalid() from None
 
 
-def _public_key(jwks: Mapping[str, object], key_id: str) -> RSAPublicKey:
+def _public_key(
+    jwks: Mapping[str, object], key_id: str, algorithm: str
+) -> RSAPublicKey | EllipticCurvePublicKey:
     if not isinstance(jwks, Mapping):
         raise _invalid()
     keys = jwks.get('keys')
@@ -103,9 +107,10 @@ def _public_key(jwks: Mapping[str, object], key_id: str) -> RSAPublicKey:
             not isinstance(candidate_id, str)
             or not _KEY_ID.fullmatch(candidate_id)
             or candidate_id in seen
-            or key.get('kty') != 'RSA'
+            or key.get('kty') != ('EC' if algorithm == 'ES256' else 'RSA')
+            or (algorithm == 'ES256' and key.get('crv') != 'P-256')
             or key.get('use', 'sig') != 'sig'
-            or key.get('alg', 'RS256') != 'RS256'
+            or key.get('alg', algorithm) != algorithm
             or ('key_ops' in key and key['key_ops'] != ['verify'])
             or any(
                 private_field in key for private_field in ('d', 'p', 'q', 'dp', 'dq', 'qi', 'oth')
@@ -117,16 +122,22 @@ def _public_key(jwks: Mapping[str, object], key_id: str) -> RSAPublicKey:
             selected = key
     if selected is None:
         raise _invalid()
-    for field in ('n', 'e'):
+    for field in ('x', 'y') if algorithm == 'ES256' else ('n', 'e'):
         value = selected.get(field)
         if (
             not isinstance(value, str)
             or not 1 <= len(value) <= 1400
+            or (algorithm == 'ES256' and len(value) != 43)
             or not _COMPACT_PART.fullmatch(value)
         ):
             raise _invalid()
-    public_key = jwt.PyJWK.from_dict(selected, algorithm='RS256').key
-    if not isinstance(public_key, RSAPublicKey) or not 2048 <= public_key.key_size <= 8192:
+    public_key = jwt.PyJWK.from_dict(selected, algorithm=algorithm).key
+    if algorithm == 'ES256':
+        if not isinstance(public_key, EllipticCurvePublicKey) or not isinstance(
+            public_key.curve, SECP256R1
+        ):
+            raise _invalid()
+    elif not isinstance(public_key, RSAPublicKey) or not 2048 <= public_key.key_size <= 8192:
         raise _invalid()
     return public_key
 
@@ -143,10 +154,11 @@ def verify_identity_token(
     audience: str,
     nonce: str,
     now: datetime,
+    algorithm: str = 'RS256',
 ) -> TokenIdentity:
     """Verify a real signature and return only the provider's immutable identity."""
     try:
-        key_id = token_key_id(token)
+        key_id = token_key_id(token, algorithm=algorithm)
         if (
             isinstance(issuers, str)
             or not isinstance(issuers, Collection)
@@ -167,8 +179,8 @@ def verify_identity_token(
         _json_part(payload_part, limit=_MAX_TOKEN_BYTES)
         claims = jwt.decode(
             token,
-            _public_key(jwks, key_id),
-            algorithms=['RS256'],
+            _public_key(jwks, key_id, algorithm),
+            algorithms=[algorithm],
             options={
                 'require': list(_REQUIRED_CLAIMS),
                 'verify_signature': True,

src/msg/transports/account_login.py

@@ -4,15 +4,16 @@
 from dataclasses import replace
 from html import escape
 from types import SimpleNamespace
-from urllib.parse import quote, urlsplit
+from urllib.parse import quote, urlencode, urlsplit
 
 from starlette.requests import Request
 from starlette.responses import RedirectResponse
 
 from msg.core.errors import Failure, require
+from msg.login_config import OAUTH_PROVIDERS
 from msg.security.email_login import begin_email, verify_email
 from msg.security.login import VerifiedLogin, bind_login, complete_login, remove_login
-from msg.security.login_flow import consume_flow, flow, receive_callback, start_flow
+from msg.security.login_flow import TTL, consume_flow, flow, receive_callback, start_flow
 from msg.security.oauth import OAuthService, get, put, secret, state_id
 from msg.transports.account_login_page import (
     approval_page,
@@ -21,7 +22,7 @@
     login_page,
     methods_page,
 )
-from msg.transports.login_providers import authorization_url, exchange_identity
+from msg.transports.login_providers import AGENTID_ISSUER, authorization_url, exchange_identity
 from msg.transports.oauth_http import HEADERS, csrf, form, json, page
 from msg.transports.url_safety import require_matching_host, require_safe_request_target
 
@@ -50,9 +51,7 @@ def provider(self, name):
 
     def providers(self):
         return tuple(
-            p.name
-            for p in self.config.providers
-            if p.enabled and p.name in {'google', 'github', 'chatgpt'}
+            p.name for p in self.config.providers if p.enabled and p.name in OAUTH_PROVIDERS
         )
 
     def enabled(self, name):
@@ -261,7 +260,7 @@ async def dispatch(self, request):
                     if self.enabled('passkey') and path != '/register'
                     else '',
                 )
-            self.cookie(response, self.flow_cookie, browser, 600)
+            self.cookie(response, self.flow_cookie, browser, TTL)
             return response
         if path == '/login/finish':
             require(request.method == 'GET', 'method_not_allowed')
@@ -272,12 +271,29 @@ async def dispatch(self, request):
                 body = flow(app, tx, state, browser)
                 require(body['phase'] == 'returned', 'invalid_login_transaction')
             return finish_page(csrf(browser), state)
-        if path == '/-/login/start':
-            args, browser = await self.args(
-                request, {'csrf', 'provider', 'mode', 'handle', 'custody'}
+        if path == '/login/agentid':
+            require(request.method == 'GET', 'method_not_allowed')
+            self.provider('agentid')
+            require(set(request.query_params) <= {'iss', 'login_hint'}, 'invalid_request')
+            # This public alias only redirects. The existing transaction route creates state.
+            return RedirectResponse(
+                '/-/login/start?' + urlencode({'provider': 'agentid', **request.query_params}),
+                303,
+                headers=HEADERS,
             )
+        if path == '/-/login/start':
+            if request.method == 'GET':
+                args = dict(request.query_params)
+                require(set(args) <= {'provider', 'iss', 'login_hint'}, 'invalid_request')
+                require(args.get('provider') == 'agentid', 'method_not_allowed')
+                require(args.get('iss') in {None, AGENTID_ISSUER}, 'login_provider_invalid_issuer')
+                browser = secret()
+            else:
+                args, browser = await self.args(
+                    request, {'csrf', 'provider', 'mode', 'handle', 'custody'}
+                )
             provider = self.provider(args.get('provider'))
-            require(provider.name in {'google', 'github', 'chatgpt'}, 'login_provider_disabled')
+            require(provider.name in OAUTH_PROVIDERS, 'login_provider_disabled')
             mode, handle = self.mode(args)
             source = await self.source(request) if mode == 'bind' else None
             async with app.metadata.transaction(write=True) as tx:
@@ -286,12 +302,23 @@ async def dispatch(self, request):
                     app, tx, provider.name, browser, mode=mode, source=source, handle=handle
                 )
             callback = app.settings.service_url + '/-/login/callback/' + provider.name
-            url = authorization_url(provider.name, provider, state, verifier, nonce, callback)
-            return RedirectResponse(url, 303, headers=HEADERS)
+            url = authorization_url(
+                provider.name,
+                provider,
+                state,
+                verifier,
+                nonce,
+                callback,
+                **({'login_hint': args.get('login_hint')} if provider.name == 'agentid' else {}),
+            )
+            response = RedirectResponse(url, 303, headers=HEADERS)
+            # Start a fresh ten-minute window even when the sign-in page was left open.
+            self.cookie(response, self.flow_cookie, browser, TTL)
+            return response
         if path.startswith('/-/login/callback/'):
             require(request.method == 'GET', 'method_not_allowed')
             provider = self.provider(path.rsplit('/', 1)[-1])
-            require(provider.name in {'google', 'github', 'chatgpt'}, 'login_provider_disabled')
+            require(provider.name in OAUTH_PROVIDERS, 'login_provider_disabled')
             require(
                 set(request.query_params)
                 <= {
@@ -306,6 +333,10 @@ async def dispatch(self, request):
                 },
                 'invalid_request',
             )
+            require(
+                provider.name != 'agentid' or request.query_params.get('iss') == AGENTID_ISSUER,
+                'login_provider_invalid_issuer',
+            )
             require('error' not in request.query_params, 'login_provider_denied')
             state = request.query_params.get('state', '')
             async with app.metadata.transaction(write=True) as tx:
@@ -442,7 +473,13 @@ async def dispatch(self, request):
     async def __call__(self, scope, receive, send):
         path = scope.get('path', '')
         enabled = getattr(getattr(self.service.settings, 'login', None), 'enabled', False)
-        routes = {'/login', '/register', '/login/finish', '/account/login-methods'}
+        routes = {
+            '/login',
+            '/login/agentid',
+            '/register',
+            '/login/finish',
+            '/account/login-methods',
+        }
         if self.config.registration == 'provider_only':
             routes.add('/oauth/signup')
         if (

src/msg/transports/account_login_page.py

@@ -5,6 +5,7 @@
 from msg.transports.oauth_http import fields, page
 
 PROVIDER_NAMES = {
+    'agentid': 'AgentID',
     'google': 'Google',
     'github': 'GitHub',
     'chatgpt': 'ChatGPT',
@@ -76,8 +77,10 @@ def login_page(
     else:
         for provider in providers:
             body += provider_form(provider, token)
-        if 'github' in providers or 'chatgpt' in providers:
-            body += '<p>GitHub／ChatGPT 须先绑定已有 MSG 账号；相同邮箱不会自动合并账号。</p>'
+        if any(provider in providers for provider in ('github', 'chatgpt', 'agentid')):
+            body += (
+                '<p>GitHub／ChatGPT／AgentID 须先绑定已有 MSG 账号；相同邮箱不会自动合并账号。</p>'
+            )
     if email:
         body += email_form(token, mode='register' if register else 'login', handle=register)
     if passkey and not register: