authentik.sources.oauth.types.wechat

  1"""WeChat (Weixin) OAuth Views"""
  2
  3from typing import Any
  4
  5from requests.exceptions import RequestException
  6
  7from authentik.sources.oauth.clients.oauth2 import OAuth2Client
  8from authentik.sources.oauth.models import OAuthSource
  9from authentik.sources.oauth.types.registry import SourceType, registry
 10from authentik.sources.oauth.views.callback import OAuthCallback
 11from authentik.sources.oauth.views.redirect import OAuthRedirect
 12
 13
 14class WeChatOAuth2Client(OAuth2Client):
 15    """
 16    WeChat OAuth2 Client
 17
 18    Handles the non-standard parts of the WeChat OAuth2 flow.
 19    """
 20
 21    def get_access_token(self, **request_kwargs) -> dict[str, Any] | None:
 22        """
 23        Get access token from WeChat.
 24
 25        WeChat uses a non-standard GET request for the token exchange,
 26        unlike the standard OAuth2 POST request. The AppID (client_id)
 27        and AppSecret (client_secret) are passed as URL query parameters.
 28        """
 29        if not self.check_application_state():
 30            self.logger.warning("Application state check failed.")
 31            return {"error": "State check failed."}
 32
 33        code = self.get_request_arg("code", None)
 34        if not code:
 35            return None
 36
 37        token_url = self.source.source_type.access_token_url
 38        params = {
 39            "appid": self.get_client_id(),
 40            "secret": self.get_client_secret(),
 41            "code": code,
 42            "grant_type": "authorization_code",
 43        }
 44
 45        # Send the GET request using the base class's session handler
 46        try:
 47            response = self.do_request("get", token_url, params=params)
 48            response.raise_for_status()
 49        except RequestException as exc:
 50            self.logger.warning("Unable to fetch wechat token", exc=exc)
 51            return None
 52
 53        data = response.json()
 54
 55        # Handle WeChat's specific error format (JSON with 'errcode' and 'errmsg')
 56        if "errcode" in data:
 57            self.logger.warning(
 58                "Unable to fetch wechat token",
 59                errcode=data.get("errcode"),
 60                errmsg=data.get("errmsg"),
 61            )
 62            return None
 63
 64        return data
 65
 66    def get_profile_info(self, token: dict[str, Any]) -> dict[str, Any] | None:
 67        """
 68        Get Userinfo from WeChat.
 69
 70        This API call requires both the 'access_token' and the 'openid'
 71        (which was returned during the token exchange).
 72        """
 73        profile_url = self.source.source_type.profile_url
 74        params = {
 75            "access_token": token.get("access_token"),
 76            "openid": token.get("openid"),
 77            "lang": "en",  # or 'zh_CN' (Simplified Chinese), 'zh_TW' (Traditional)
 78        }
 79
 80        response = self.do_request("get", profile_url, params=params)
 81
 82        try:
 83            response.raise_for_status()
 84        except RequestException as exc:
 85            self.logger.warning("Unable to fetch wechat userinfo", exc=exc)
 86            return None
 87
 88        data = response.json()
 89
 90        # Handle WeChat's specific error format
 91        if "errcode" in data:
 92            self.logger.warning(
 93                "Unable to fetch wechat userinfo",
 94                errcode=data.get("errcode"),
 95                errmsg=data.get("errmsg"),
 96            )
 97            return None
 98
 99        return data
100
101    def get_redirect_args(self) -> dict[str, str]:
102        """Get request parameters for redirect url."""
103        args = super().get_redirect_args()
104        args["appid"] = args.pop("client_id")
105        return args
106
107
108class WeChatOAuthRedirect(OAuthRedirect):
109    """WeChat OAuth2 Redirect"""
110
111    client_class = WeChatOAuth2Client
112
113    def get_additional_parameters(self, source: OAuthSource):  # pragma: no cover
114        # WeChat (Weixin) for Websites official documentation requires 'snsapi_login'
115        # as the *only* scope for the QR code-based login flow.
116        # Ref: https://developers.weixin.qq.com/doc/oplatform/Website_App/WeChat_Login/Wechat_Login.html (Step 1)  # noqa: E501
117        return {
118            "scope": ["snsapi_login"],
119        }
120
121
122class WeChatOAuth2Callback(OAuthCallback):
123    """WeChat OAuth2 Callback"""
124
125    # Specify our custom Client to handle the non-standard WeChat flow
126    client_class = WeChatOAuth2Client
127
128    def get_user_id(self, info: dict[str, Any]) -> str | None:
129        return info.get("unionid", info.get("openid"))
130
131
132@registry.register()
133class WeChatType(SourceType):
134    """WeChat Type definition"""
135
136    callback_view = WeChatOAuth2Callback
137    redirect_view = WeChatOAuthRedirect
138    verbose_name = "WeChat"
139    name = "wechat"
140
141    # WeChat API URLs are fixed and not customizable
142    urls_customizable = False
143
144    # URLs for the WeChat "Login for Websites" authorization flow
145    authorization_url = "https://open.weixin.qq.com/connect/qrconnect"
146    # This is a public URL, not a hardcoded secret
147    access_token_url = "https://api.weixin.qq.com/sns/oauth2/access_token"  # nosec B105
148    profile_url = "https://api.weixin.qq.com/sns/userinfo"
149
150    # Note: 'authorization_code_auth_method' is intentionally omitted.
151    # The base OAuth2Client defaults to POST_BODY, but our custom
152    # WeChatOAuth2Client overrides get_access_token() to use GET,
153    # so this setting would be misleading.
154
155    def get_base_user_properties(self, info: dict[str, Any], **kwargs) -> dict[str, Any]:
156        """
157        Map WeChat userinfo to authentik user properties.
158        """
159        # The WeChat userinfo API (sns/userinfo) does *not* return an email address.
160        # We explicitly set 'email' to None. Authentik will typically
161        # prompt the user to provide one on their first login if it's required.
162
163        # 'unionid' is the preferred unique identifier as it's consistent
164        # across multiple apps under the same WeChat Open Platform account.
165        # 'openid' is the fallback, which is only unique to this specific AppID.
166        return {
167            "username": info.get("unionid", info.get("openid")),
168            "email": None,  # WeChat API does not provide Email
169            "name": info.get("nickname"),
170            "attributes": {
171                # Save all other relevant info as user attributes
172                "headimgurl": info.get("headimgurl"),
173                "sex": info.get("sex"),
174                "city": info.get("city"),
175                "province": info.get("province"),
176                "country": info.get("country"),
177                "unionid": info.get("unionid"),
178                "openid": info.get("openid"),
179            },
180        }