We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[日志] auth fail的日志没有区分度
大量的相同的错误信息, 无法快速精确答复`调用方`的疑惑

```
msg = "Invalid token header. No credentials provided."
```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/api/bkuser_core/enhanced_account/authentication.py`
Content:
```
1 # -*- coding: utf-8 -*-
2 """
3 TencentBlueKing is pleased to support the open source community by making 蓝鲸智云-用户管理(Bk-User) available.
4 Copyright (C) 2017-2021 THL A29 Limited, a Tencent company. All rights reserved.
5 Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License.
6 You may obtain a copy of the License at http://opensource.org/licenses/MIT
7 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
8 an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
9 specific language governing permissions and limitations under the License.
10 """
11 import base64
12 import logging
13 import re
14 
15 import jwt
16 from django.conf import settings
17 from django.contrib.auth import get_user_model
18 from django.core.cache import caches
19 from rest_framework import exceptions
20 from rest_framework.authentication import BaseAuthentication, get_authorization_header
21 
22 from bkuser_core.esb_sdk.shortcuts import get_client_by_raw_username
23 
24 logger = logging.getLogger(__name__)
25 
26 
27 HEADER_JWT_KEY_NAME = "HTTP_X_BKAPI_JWT"
28 HEADER_APP_CODE_KEY_NAME = "HTTP_X_BK_APP_CODE"
29 HEADER_APP_SECRET_KEY_NAME = "HTTP_X_BK_APP_SECRET"
30 
31 ESB_PUBLIC_KEY_CACHE_KEY = "bk_user:esb_public_key"
32 
33 TOKEN_KEY_NAME = "token"
34 
35 
36 def create_user(username="admin"):
37     return get_user_model()(username=username, is_staff=True, is_superuser=True)
38 
39 
40 class InternalTokenAuthentication(BaseAuthentication):
41 
42     keyword = "iBearer"
43     model = None
44 
45     query_params_keyword = TOKEN_KEY_NAME
46 
47     def get_token_from_query_params(self, request):
48         try:
49             return request.query_params[self.query_params_keyword]
50         except KeyError:
51             msg = "Invalid token header. No credentials provided."
52             raise exceptions.AuthenticationFailed(msg)
53 
54     def get_token_from_header(self, request):
55         auth = get_authorization_header(request).split()
56 
57         if not auth or auth[0].lower() != self.keyword.lower().encode():
58             msg = "Invalid token header. No credentials provided."
59             raise exceptions.AuthenticationFailed(msg)
60 
61         if len(auth) == 1:
62             msg = "Invalid token header. No credentials provided."
63             raise exceptions.AuthenticationFailed(msg)
64         elif len(auth) > 2:
65             msg = "Invalid token header. Token string should not contain spaces."
66             raise exceptions.AuthenticationFailed(msg)
67 
68         try:
69             token = auth[1].decode()
70         except UnicodeError:
71             msg = "Invalid token header. Token string should not contain invalid characters."
72             raise exceptions.AuthenticationFailed(msg)
73 
74         return token
75 
76     def authenticate(self, request):
77         for white_url in settings.AUTH_EXEMPT_PATHS:
78             if re.search(white_url, request.path):
79                 logger.info("%s path in white_url<%s>, exempting auth", request.path, white_url)
80                 return None, None
81 
82         try:
83             token = self.get_token_from_query_params(request)
84         except exceptions.AuthenticationFailed:
85             logger.debug("no token from query params, trying to get from header instead")
86             token = self.get_token_from_header(request)
87 
88         return self.authenticate_credentials(token)
89 
90     def authenticate_credentials(self, key):
91         """Use access token to identify user"""
92         if key in settings.INTERNAL_AUTH_TOKENS:
93             user_info = settings.INTERNAL_AUTH_TOKENS[key]
94             return create_user(user_info["username"]), None
95         raise exceptions.AuthenticationFailed("request failed: Invalid token header. No credentials provided.")
96 
97 
98 class ESBOrAPIGatewayAuthentication(BaseAuthentication):
99     def authenticate(self, request):
100         # get jwt from header
101         jwt_content = request.META.get(HEADER_JWT_KEY_NAME, "")
102         if not jwt_content:
103             return None, None
104 
105         # get the public key
106         jwt_header = jwt.get_unverified_header(jwt_content)
107         api_name = jwt_header.get("kid") or ""
108         public_key = self._get_public_key(api_name)
109         algorithm = jwt_header.get("alg") or "RS512"
110 
111         # do decode, without verify issuer
112         try:
113             jwt_playload = jwt.decode(jwt_content, public_key, algorithm)
114         except Exception:  # pylint: disable=broad-except
115             logger.exception("JWT decode failed! jwt_payload: %s, public_key: %s", jwt_content, public_key)
116             raise exceptions.AuthenticationFailed("decode jwt token fail")
117 
118         # username = self._get_username_from_jwt_payload(payload)
119         app_code = self._get_app_code_from_jwt_payload(jwt_playload)
120         request.bk_app_code = app_code
121 
122         username = "APIGW" if api_name == settings.BK_APIGW_NAME else "ESB"
123         return create_user(username), None
124 
125     def _get_public_key(self, api_name):
126         # it's from apigateway
127         if api_name == settings.BK_APIGW_NAME:
128             return self._get_apigw_public_key()
129         # it's from esb
130         else:
131             return self._get_esb_public_key()
132 
133     def _get_apigw_public_key(self):
134         """
135         获取APIGW的PUBLIC KEY
136         由于配置文件里的public key 是来着环境变量，且使用了base64编码的，所以需要获取后解码
137         """
138         # 如果BK_APIGW_PUBLIC_KEY为空，则直接报错
139         if not settings.BK_APIGW_PUBLIC_KEY:
140             raise exceptions.AuthenticationFailed("BK_APIGW_PUBLIC_KEY can not be empty")
141 
142         # base64解码
143         try:
144             public_key = base64.b64decode(settings.BK_APIGW_PUBLIC_KEY).decode("utf-8")
145         except Exception:  # pylint: disable=broad-except
146             logger.exception(
147                 "BK_APIGW_PUBLIC_KEY is not valid base64 string! public_key=%s", settings.BK_APIGW_PUBLIC_KEY
148             )
149             raise exceptions.AuthenticationFailed("BK_APIGW_PUBLIC_KEY is not valid base64 string!")
150 
151         return public_key
152 
153     def _get_esb_public_key(self):
154         cache = caches["locmem"]
155         public_key = cache.get(ESB_PUBLIC_KEY_CACHE_KEY)
156         if public_key is not None:
157             return public_key
158 
159         # get from esb
160         client = get_client_by_raw_username("admin")
161         try:
162             ret = client.esb.get_public_key()
163         except Exception:  # pylint: disable=broad-except
164             logger.exception("Get ESB Public Key failed!")
165             raise exceptions.AuthenticationFailed("Get ESB Public Key failed!")
166 
167         if not ret.get("result", False):
168             msg = ret.get("message", "unknown error")
169             logger.error("Get ESB Public Key failed! %s", msg)
170             raise exceptions.AuthenticationFailed(f"Get ESB Public Key failed! {msg}")
171 
172         public_key = ret.get("data", {}).get("public_key", "")
173         if public_key is None:
174             logger.error("Get ESB Public Key failed! public_key is empty, ret=%s", ret)
175             raise exceptions.AuthenticationFailed("Get ESB Public Key failed! the public key is empty")
176 
177         cache.set(ESB_PUBLIC_KEY_CACHE_KEY, public_key, 60 * 60)
178         return public_key
179 
180     def _get_app_code_from_jwt_payload(self, jwt_payload):
181         """从jwt里获取app_code"""
182         app = jwt_payload.get("app", {})
183 
184         verified = app.get("verified", False)
185         if not verified:
186             raise exceptions.AuthenticationFailed("app is not verified")
187 
188         app_code = app.get("bk_app_code", "") or app.get("app_code", "")
189         # 虽然app_code为空对于后续的鉴权一定是不通过的，但鉴权不通过有很多原因，这里提前log便于问题排查
190         if not app_code:
191             raise exceptions.AuthenticationFailed("could not get app_code from esb/apigateway jwt payload! it's empty")
192 
193         return app_code
194 
195 
196 class AppCodeAppSecretAuthentication(BaseAuthentication):
197     """
198     通过app_code和app_secret进行鉴权
199     """
200 
201     def authenticate(self, request):
202         # get app_code and app_secret from header
203         app_code = request.META.get(HEADER_APP_CODE_KEY_NAME, "")
204         app_secret = request.META.get(HEADER_APP_SECRET_KEY_NAME, "")
205 
206         if app_code == settings.APP_ID and app_secret == settings.APP_TOKEN:
207             return create_user("SAAS"), None
208 
209         return None, None
210 
211 
212 class MultipleAuthentication(BaseAuthentication):
213     """it's a dispatcher"""
214 
215     def authenticate(self, request):
216         # FIXME: 最终, 下掉token, 只保留 jwt + app_code/app_secret
217         # withe list
218         for white_url in settings.AUTH_EXEMPT_PATHS:
219             if re.search(white_url, request.path):
220                 logger.debug("%s path in white_url<%s>, exempting auth", request.path, white_url)
221                 return None, None
222 
223         # app_code and app_secret
224         if HEADER_APP_CODE_KEY_NAME in request.META and HEADER_APP_SECRET_KEY_NAME in request.META:
225             return AppCodeAppSecretAuthentication().authenticate(request)
226 
227         # FIXME: should remove this totally
228         # NOTE: some case we want to use token as credentials, call through APIGateway(set default headers)
229         # so we should verify others first, not jwt
230         if get_authorization_header(request) or request.query_params.get(TOKEN_KEY_NAME):
231             # token
232             return InternalTokenAuthentication().authenticate(request)
233 
234         # jwt
235         if HEADER_JWT_KEY_NAME in request.META:
236             return ESBOrAPIGatewayAuthentication().authenticate(request)
237 
238         raise exceptions.AuthenticationFailed(
239             "no valid authentication credentials provided! should call through APIGateway/ESB"
240         )
241
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/api/bkuser_core/enhanced_account/authentication.py b/src/api/bkuser_core/enhanced_account/authentication.py
--- a/src/api/bkuser_core/enhanced_account/authentication.py
+++ b/src/api/bkuser_core/enhanced_account/authentication.py
@@ -48,21 +48,24 @@
         try:
             return request.query_params[self.query_params_keyword]
         except KeyError:
-            msg = "Invalid token header. No credentials provided."
+            msg = f"Invalid token header. No credentials provided. {self.query_params_keyword} is not in query params"
             raise exceptions.AuthenticationFailed(msg)
 
     def get_token_from_header(self, request):
         auth = get_authorization_header(request).split()
 
         if not auth or auth[0].lower() != self.keyword.lower().encode():
-            msg = "Invalid token header. No credentials provided."
+            msg = "Invalid token header. No credentials provided. The format should be `iBearer THE_TOKEN`"
             raise exceptions.AuthenticationFailed(msg)
 
         if len(auth) == 1:
-            msg = "Invalid token header. No credentials provided."
+            msg = "Invalid token header. No credentials provided. The size of auth array credentials is 0"
             raise exceptions.AuthenticationFailed(msg)
         elif len(auth) > 2:
-            msg = "Invalid token header. Token string should not contain spaces."
+            msg = (
+                "Invalid token header. Token string should not contain spaces. "
+                + "The size of auth array credentials is more than 2"
+            )
             raise exceptions.AuthenticationFailed(msg)
 
         try:
@@ -74,11 +77,6 @@
         return token
 
     def authenticate(self, request):
-        for white_url in settings.AUTH_EXEMPT_PATHS:
-            if re.search(white_url, request.path):
-                logger.info("%s path in white_url<%s>, exempting auth", request.path, white_url)
-                return None, None
-
         try:
             token = self.get_token_from_query_params(request)
         except exceptions.AuthenticationFailed:
@@ -92,7 +90,9 @@
         if key in settings.INTERNAL_AUTH_TOKENS:
             user_info = settings.INTERNAL_AUTH_TOKENS[key]
             return create_user(user_info["username"]), None
-        raise exceptions.AuthenticationFailed("request failed: Invalid token header. No credentials provided.")
+        raise exceptions.AuthenticationFailed(
+            "request failed: Invalid token header. No credentials provided or Wrong credentials."
+        )
 
 
 class ESBOrAPIGatewayAuthentication(BaseAuthentication):

{"golden_diff": "diff --git a/src/api/bkuser_core/enhanced_account/authentication.py b/src/api/bkuser_core/enhanced_account/authentication.py\n--- a/src/api/bkuser_core/enhanced_account/authentication.py\n+++ b/src/api/bkuser_core/enhanced_account/authentication.py\n@@ -48,21 +48,24 @@\n         try:\n             return request.query_params[self.query_params_keyword]\n         except KeyError:\n-            msg = \"Invalid token header. No credentials provided.\"\n+            msg = f\"Invalid token header. No credentials provided. {self.query_params_keyword} is not in query params\"\n             raise exceptions.AuthenticationFailed(msg)\n \n     def get_token_from_header(self, request):\n         auth = get_authorization_header(request).split()\n \n         if not auth or auth[0].lower() != self.keyword.lower().encode():\n-            msg = \"Invalid token header. No credentials provided.\"\n+            msg = \"Invalid token header. No credentials provided. The format should be `iBearer THE_TOKEN`\"\n             raise exceptions.AuthenticationFailed(msg)\n \n         if len(auth) == 1:\n-            msg = \"Invalid token header. No credentials provided.\"\n+            msg = \"Invalid token header. No credentials provided. The size of auth array credentials is 0\"\n             raise exceptions.AuthenticationFailed(msg)\n         elif len(auth) > 2:\n-            msg = \"Invalid token header. Token string should not contain spaces.\"\n+            msg = (\n+                \"Invalid token header. Token string should not contain spaces. \"\n+                + \"The size of auth array credentials is more than 2\"\n+            )\n             raise exceptions.AuthenticationFailed(msg)\n \n         try:\n@@ -74,11 +77,6 @@\n         return token\n \n     def authenticate(self, request):\n-        for white_url in settings.AUTH_EXEMPT_PATHS:\n-            if re.search(white_url, request.path):\n-                logger.info(\"%s path in white_url<%s>, exempting auth\", request.path, white_url)\n-                return None, None\n-\n         try:\n             token = self.get_token_from_query_params(request)\n         except exceptions.AuthenticationFailed:\n@@ -92,7 +90,9 @@\n         if key in settings.INTERNAL_AUTH_TOKENS:\n             user_info = settings.INTERNAL_AUTH_TOKENS[key]\n             return create_user(user_info[\"username\"]), None\n-        raise exceptions.AuthenticationFailed(\"request failed: Invalid token header. No credentials provided.\")\n+        raise exceptions.AuthenticationFailed(\n+            \"request failed: Invalid token header. No credentials provided or Wrong credentials.\"\n+        )\n \n \n class ESBOrAPIGatewayAuthentication(BaseAuthentication):\n", "issue": "[\u65e5\u5fd7] auth fail\u7684\u65e5\u5fd7\u6ca1\u6709\u533a\u5206\u5ea6\n\u5927\u91cf\u7684\u76f8\u540c\u7684\u9519\u8bef\u4fe1\u606f, \u65e0\u6cd5\u5feb\u901f\u7cbe\u786e\u7b54\u590d`\u8c03\u7528\u65b9`\u7684\u7591\u60d1\r\n\r\n```\r\nmsg = \"Invalid token header. No credentials provided.\"\r\n```\n", "before_files": [{"content": "# -*- coding: utf-8 -*-\n\"\"\"\nTencentBlueKing is pleased to support the open source community by making \u84dd\u9cb8\u667a\u4e91-\u7528\u6237\u7ba1\u7406(Bk-User) available.\nCopyright (C) 2017-2021 THL A29 Limited, a Tencent company. All rights reserved.\nLicensed under the MIT License (the \"License\"); you may not use this file except in compliance with the License.\nYou may obtain a copy of the License at http://opensource.org/licenses/MIT\nUnless required by applicable law or agreed to in writing, software distributed under the License is distributed on\nan \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the\nspecific language governing permissions and limitations under the License.\n\"\"\"\nimport base64\nimport logging\nimport re\n\nimport jwt\nfrom django.conf import settings\nfrom django.contrib.auth import get_user_model\nfrom django.core.cache import caches\nfrom rest_framework import exceptions\nfrom rest_framework.authentication import BaseAuthentication, get_authorization_header\n\nfrom bkuser_core.esb_sdk.shortcuts import get_client_by_raw_username\n\nlogger = logging.getLogger(__name__)\n\n\nHEADER_JWT_KEY_NAME = \"HTTP_X_BKAPI_JWT\"\nHEADER_APP_CODE_KEY_NAME = \"HTTP_X_BK_APP_CODE\"\nHEADER_APP_SECRET_KEY_NAME = \"HTTP_X_BK_APP_SECRET\"\n\nESB_PUBLIC_KEY_CACHE_KEY = \"bk_user:esb_public_key\"\n\nTOKEN_KEY_NAME = \"token\"\n\n\ndef create_user(username=\"admin\"):\n    return get_user_model()(username=username, is_staff=True, is_superuser=True)\n\n\nclass InternalTokenAuthentication(BaseAuthentication):\n\n    keyword = \"iBearer\"\n    model = None\n\n    query_params_keyword = TOKEN_KEY_NAME\n\n    def get_token_from_query_params(self, request):\n        try:\n            return request.query_params[self.query_params_keyword]\n        except KeyError:\n            msg = \"Invalid token header. No credentials provided.\"\n            raise exceptions.AuthenticationFailed(msg)\n\n    def get_token_from_header(self, request):\n        auth = get_authorization_header(request).split()\n\n        if not auth or auth[0].lower() != self.keyword.lower().encode():\n            msg = \"Invalid token header. No credentials provided.\"\n            raise exceptions.AuthenticationFailed(msg)\n\n        if len(auth) == 1:\n            msg = \"Invalid token header. No credentials provided.\"\n            raise exceptions.AuthenticationFailed(msg)\n        elif len(auth) > 2:\n            msg = \"Invalid token header. Token string should not contain spaces.\"\n            raise exceptions.AuthenticationFailed(msg)\n\n        try:\n            token = auth[1].decode()\n        except UnicodeError:\n            msg = \"Invalid token header. Token string should not contain invalid characters.\"\n            raise exceptions.AuthenticationFailed(msg)\n\n        return token\n\n    def authenticate(self, request):\n        for white_url in settings.AUTH_EXEMPT_PATHS:\n            if re.search(white_url, request.path):\n                logger.info(\"%s path in white_url<%s>, exempting auth\", request.path, white_url)\n                return None, None\n\n        try:\n            token = self.get_token_from_query_params(request)\n        except exceptions.AuthenticationFailed:\n            logger.debug(\"no token from query params, trying to get from header instead\")\n            token = self.get_token_from_header(request)\n\n        return self.authenticate_credentials(token)\n\n    def authenticate_credentials(self, key):\n        \"\"\"Use access token to identify user\"\"\"\n        if key in settings.INTERNAL_AUTH_TOKENS:\n            user_info = settings.INTERNAL_AUTH_TOKENS[key]\n            return create_user(user_info[\"username\"]), None\n        raise exceptions.AuthenticationFailed(\"request failed: Invalid token header. No credentials provided.\")\n\n\nclass ESBOrAPIGatewayAuthentication(BaseAuthentication):\n    def authenticate(self, request):\n        # get jwt from header\n        jwt_content = request.META.get(HEADER_JWT_KEY_NAME, \"\")\n        if not jwt_content:\n            return None, None\n\n        # get the public key\n        jwt_header = jwt.get_unverified_header(jwt_content)\n        api_name = jwt_header.get(\"kid\") or \"\"\n        public_key = self._get_public_key(api_name)\n        algorithm = jwt_header.get(\"alg\") or \"RS512\"\n\n        # do decode, without verify issuer\n        try:\n            jwt_playload = jwt.decode(jwt_content, public_key, algorithm)\n        except Exception:  # pylint: disable=broad-except\n            logger.exception(\"JWT decode failed! jwt_payload: %s, public_key: %s\", jwt_content, public_key)\n            raise exceptions.AuthenticationFailed(\"decode jwt token fail\")\n\n        # username = self._get_username_from_jwt_payload(payload)\n        app_code = self._get_app_code_from_jwt_payload(jwt_playload)\n        request.bk_app_code = app_code\n\n        username = \"APIGW\" if api_name == settings.BK_APIGW_NAME else \"ESB\"\n        return create_user(username), None\n\n    def _get_public_key(self, api_name):\n        # it's from apigateway\n        if api_name == settings.BK_APIGW_NAME:\n            return self._get_apigw_public_key()\n        # it's from esb\n        else:\n            return self._get_esb_public_key()\n\n    def _get_apigw_public_key(self):\n        \"\"\"\n        \u83b7\u53d6APIGW\u7684PUBLIC KEY\n        \u7531\u4e8e\u914d\u7f6e\u6587\u4ef6\u91cc\u7684public key \u662f\u6765\u7740\u73af\u5883\u53d8\u91cf\uff0c\u4e14\u4f7f\u7528\u4e86base64\u7f16\u7801\u7684\uff0c\u6240\u4ee5\u9700\u8981\u83b7\u53d6\u540e\u89e3\u7801\n        \"\"\"\n        # \u5982\u679cBK_APIGW_PUBLIC_KEY\u4e3a\u7a7a\uff0c\u5219\u76f4\u63a5\u62a5\u9519\n        if not settings.BK_APIGW_PUBLIC_KEY:\n            raise exceptions.AuthenticationFailed(\"BK_APIGW_PUBLIC_KEY can not be empty\")\n\n        # base64\u89e3\u7801\n        try:\n            public_key = base64.b64decode(settings.BK_APIGW_PUBLIC_KEY).decode(\"utf-8\")\n        except Exception:  # pylint: disable=broad-except\n            logger.exception(\n                \"BK_APIGW_PUBLIC_KEY is not valid base64 string! public_key=%s\", settings.BK_APIGW_PUBLIC_KEY\n            )\n            raise exceptions.AuthenticationFailed(\"BK_APIGW_PUBLIC_KEY is not valid base64 string!\")\n\n        return public_key\n\n    def _get_esb_public_key(self):\n        cache = caches[\"locmem\"]\n        public_key = cache.get(ESB_PUBLIC_KEY_CACHE_KEY)\n        if public_key is not None:\n            return public_key\n\n        # get from esb\n        client = get_client_by_raw_username(\"admin\")\n        try:\n            ret = client.esb.get_public_key()\n        except Exception:  # pylint: disable=broad-except\n            logger.exception(\"Get ESB Public Key failed!\")\n            raise exceptions.AuthenticationFailed(\"Get ESB Public Key failed!\")\n\n        if not ret.get(\"result\", False):\n            msg = ret.get(\"message\", \"unknown error\")\n            logger.error(\"Get ESB Public Key failed! %s\", msg)\n            raise exceptions.AuthenticationFailed(f\"Get ESB Public Key failed! {msg}\")\n\n        public_key = ret.get(\"data\", {}).get(\"public_key\", \"\")\n        if public_key is None:\n            logger.error(\"Get ESB Public Key failed! public_key is empty, ret=%s\", ret)\n            raise exceptions.AuthenticationFailed(\"Get ESB Public Key failed! the public key is empty\")\n\n        cache.set(ESB_PUBLIC_KEY_CACHE_KEY, public_key, 60 * 60)\n        return public_key\n\n    def _get_app_code_from_jwt_payload(self, jwt_payload):\n        \"\"\"\u4ecejwt\u91cc\u83b7\u53d6app_code\"\"\"\n        app = jwt_payload.get(\"app\", {})\n\n        verified = app.get(\"verified\", False)\n        if not verified:\n            raise exceptions.AuthenticationFailed(\"app is not verified\")\n\n        app_code = app.get(\"bk_app_code\", \"\") or app.get(\"app_code\", \"\")\n        # \u867d\u7136app_code\u4e3a\u7a7a\u5bf9\u4e8e\u540e\u7eed\u7684\u9274\u6743\u4e00\u5b9a\u662f\u4e0d\u901a\u8fc7\u7684\uff0c\u4f46\u9274\u6743\u4e0d\u901a\u8fc7\u6709\u5f88\u591a\u539f\u56e0\uff0c\u8fd9\u91cc\u63d0\u524dlog\u4fbf\u4e8e\u95ee\u9898\u6392\u67e5\n        if not app_code:\n            raise exceptions.AuthenticationFailed(\"could not get app_code from esb/apigateway jwt payload! it's empty\")\n\n        return app_code\n\n\nclass AppCodeAppSecretAuthentication(BaseAuthentication):\n    \"\"\"\n    \u901a\u8fc7app_code\u548capp_secret\u8fdb\u884c\u9274\u6743\n    \"\"\"\n\n    def authenticate(self, request):\n        # get app_code and app_secret from header\n        app_code = request.META.get(HEADER_APP_CODE_KEY_NAME, \"\")\n        app_secret = request.META.get(HEADER_APP_SECRET_KEY_NAME, \"\")\n\n        if app_code == settings.APP_ID and app_secret == settings.APP_TOKEN:\n            return create_user(\"SAAS\"), None\n\n        return None, None\n\n\nclass MultipleAuthentication(BaseAuthentication):\n    \"\"\"it's a dispatcher\"\"\"\n\n    def authenticate(self, request):\n        # FIXME: \u6700\u7ec8, \u4e0b\u6389token, \u53ea\u4fdd\u7559 jwt + app_code/app_secret\n        # withe list\n        for white_url in settings.AUTH_EXEMPT_PATHS:\n            if re.search(white_url, request.path):\n                logger.debug(\"%s path in white_url<%s>, exempting auth\", request.path, white_url)\n                return None, None\n\n        # app_code and app_secret\n        if HEADER_APP_CODE_KEY_NAME in request.META and HEADER_APP_SECRET_KEY_NAME in request.META:\n            return AppCodeAppSecretAuthentication().authenticate(request)\n\n        # FIXME: should remove this totally\n        # NOTE: some case we want to use token as credentials, call through APIGateway(set default headers)\n        # so we should verify others first, not jwt\n        if get_authorization_header(request) or request.query_params.get(TOKEN_KEY_NAME):\n            # token\n            return InternalTokenAuthentication().authenticate(request)\n\n        # jwt\n        if HEADER_JWT_KEY_NAME in request.META:\n            return ESBOrAPIGatewayAuthentication().authenticate(request)\n\n        raise exceptions.AuthenticationFailed(\n            \"no valid authentication credentials provided! should call through APIGateway/ESB\"\n        )\n", "path": "src/api/bkuser_core/enhanced_account/authentication.py"}], "after_files": [{"content": "# -*- coding: utf-8 -*-\n\"\"\"\nTencentBlueKing is pleased to support the open source community by making \u84dd\u9cb8\u667a\u4e91-\u7528\u6237\u7ba1\u7406(Bk-User) available.\nCopyright (C) 2017-2021 THL A29 Limited, a Tencent company. All rights reserved.\nLicensed under the MIT License (the \"License\"); you may not use this file except in compliance with the License.\nYou may obtain a copy of the License at http://opensource.org/licenses/MIT\nUnless required by applicable law or agreed to in writing, software distributed under the License is distributed on\nan \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the\nspecific language governing permissions and limitations under the License.\n\"\"\"\nimport base64\nimport logging\nimport re\n\nimport jwt\nfrom django.conf import settings\nfrom django.contrib.auth import get_user_model\nfrom django.core.cache import caches\nfrom rest_framework import exceptions\nfrom rest_framework.authentication import BaseAuthentication, get_authorization_header\n\nfrom bkuser_core.esb_sdk.shortcuts import get_client_by_raw_username\n\nlogger = logging.getLogger(__name__)\n\n\nHEADER_JWT_KEY_NAME = \"HTTP_X_BKAPI_JWT\"\nHEADER_APP_CODE_KEY_NAME = \"HTTP_X_BK_APP_CODE\"\nHEADER_APP_SECRET_KEY_NAME = \"HTTP_X_BK_APP_SECRET\"\n\nESB_PUBLIC_KEY_CACHE_KEY = \"bk_user:esb_public_key\"\n\nTOKEN_KEY_NAME = \"token\"\n\n\ndef create_user(username=\"admin\"):\n    return get_user_model()(username=username, is_staff=True, is_superuser=True)\n\n\nclass InternalTokenAuthentication(BaseAuthentication):\n\n    keyword = \"iBearer\"\n    model = None\n\n    query_params_keyword = TOKEN_KEY_NAME\n\n    def get_token_from_query_params(self, request):\n        try:\n            return request.query_params[self.query_params_keyword]\n        except KeyError:\n            msg = f\"Invalid token header. No credentials provided. {self.query_params_keyword} is not in query params\"\n            raise exceptions.AuthenticationFailed(msg)\n\n    def get_token_from_header(self, request):\n        auth = get_authorization_header(request).split()\n\n        if not auth or auth[0].lower() != self.keyword.lower().encode():\n            msg = \"Invalid token header. No credentials provided. The format should be `iBearer THE_TOKEN`\"\n            raise exceptions.AuthenticationFailed(msg)\n\n        if len(auth) == 1:\n            msg = \"Invalid token header. No credentials provided. The size of auth array credentials is 0\"\n            raise exceptions.AuthenticationFailed(msg)\n        elif len(auth) > 2:\n            msg = (\n                \"Invalid token header. Token string should not contain spaces. \"\n                + \"The size of auth array credentials is more than 2\"\n            )\n            raise exceptions.AuthenticationFailed(msg)\n\n        try:\n            token = auth[1].decode()\n        except UnicodeError:\n            msg = \"Invalid token header. Token string should not contain invalid characters.\"\n            raise exceptions.AuthenticationFailed(msg)\n\n        return token\n\n    def authenticate(self, request):\n        try:\n            token = self.get_token_from_query_params(request)\n        except exceptions.AuthenticationFailed:\n            logger.debug(\"no token from query params, trying to get from header instead\")\n            token = self.get_token_from_header(request)\n\n        return self.authenticate_credentials(token)\n\n    def authenticate_credentials(self, key):\n        \"\"\"Use access token to identify user\"\"\"\n        if key in settings.INTERNAL_AUTH_TOKENS:\n            user_info = settings.INTERNAL_AUTH_TOKENS[key]\n            return create_user(user_info[\"username\"]), None\n        raise exceptions.AuthenticationFailed(\n            \"request failed: Invalid token header. No credentials provided or Wrong credentials.\"\n        )\n\n\nclass ESBOrAPIGatewayAuthentication(BaseAuthentication):\n    def authenticate(self, request):\n        # get jwt from header\n        jwt_content = request.META.get(HEADER_JWT_KEY_NAME, \"\")\n        if not jwt_content:\n            return None, None\n\n        # get the public key\n        jwt_header = jwt.get_unverified_header(jwt_content)\n        api_name = jwt_header.get(\"kid\") or \"\"\n        public_key = self._get_public_key(api_name)\n        algorithm = jwt_header.get(\"alg\") or \"RS512\"\n\n        # do decode, without verify issuer\n        try:\n            jwt_playload = jwt.decode(jwt_content, public_key, algorithm)\n        except Exception:  # pylint: disable=broad-except\n            logger.exception(\"JWT decode failed! jwt_payload: %s, public_key: %s\", jwt_content, public_key)\n            raise exceptions.AuthenticationFailed(\"decode jwt token fail\")\n\n        # username = self._get_username_from_jwt_payload(payload)\n        app_code = self._get_app_code_from_jwt_payload(jwt_playload)\n        request.bk_app_code = app_code\n\n        username = \"APIGW\" if api_name == settings.BK_APIGW_NAME else \"ESB\"\n        return create_user(username), None\n\n    def _get_public_key(self, api_name):\n        # it's from apigateway\n        if api_name == settings.BK_APIGW_NAME:\n            return self._get_apigw_public_key()\n        # it's from esb\n        else:\n            return self._get_esb_public_key()\n\n    def _get_apigw_public_key(self):\n        \"\"\"\n        \u83b7\u53d6APIGW\u7684PUBLIC KEY\n        \u7531\u4e8e\u914d\u7f6e\u6587\u4ef6\u91cc\u7684public key \u662f\u6765\u7740\u73af\u5883\u53d8\u91cf\uff0c\u4e14\u4f7f\u7528\u4e86base64\u7f16\u7801\u7684\uff0c\u6240\u4ee5\u9700\u8981\u83b7\u53d6\u540e\u89e3\u7801\n        \"\"\"\n        # \u5982\u679cBK_APIGW_PUBLIC_KEY\u4e3a\u7a7a\uff0c\u5219\u76f4\u63a5\u62a5\u9519\n        if not settings.BK_APIGW_PUBLIC_KEY:\n            raise exceptions.AuthenticationFailed(\"BK_APIGW_PUBLIC_KEY can not be empty\")\n\n        # base64\u89e3\u7801\n        try:\n            public_key = base64.b64decode(settings.BK_APIGW_PUBLIC_KEY).decode(\"utf-8\")\n        except Exception:  # pylint: disable=broad-except\n            logger.exception(\n                \"BK_APIGW_PUBLIC_KEY is not valid base64 string! public_key=%s\", settings.BK_APIGW_PUBLIC_KEY\n            )\n            raise exceptions.AuthenticationFailed(\"BK_APIGW_PUBLIC_KEY is not valid base64 string!\")\n\n        return public_key\n\n    def _get_esb_public_key(self):\n        cache = caches[\"locmem\"]\n        public_key = cache.get(ESB_PUBLIC_KEY_CACHE_KEY)\n        if public_key is not None:\n            return public_key\n\n        # get from esb\n        client = get_client_by_raw_username(\"admin\")\n        try:\n            ret = client.esb.get_public_key()\n        except Exception:  # pylint: disable=broad-except\n            logger.exception(\"Get ESB Public Key failed!\")\n            raise exceptions.AuthenticationFailed(\"Get ESB Public Key failed!\")\n\n        if not ret.get(\"result\", False):\n            msg = ret.get(\"message\", \"unknown error\")\n            logger.error(\"Get ESB Public Key failed! %s\", msg)\n            raise exceptions.AuthenticationFailed(f\"Get ESB Public Key failed! {msg}\")\n\n        public_key = ret.get(\"data\", {}).get(\"public_key\", \"\")\n        if public_key is None:\n            logger.error(\"Get ESB Public Key failed! public_key is empty, ret=%s\", ret)\n            raise exceptions.AuthenticationFailed(\"Get ESB Public Key failed! the public key is empty\")\n\n        cache.set(ESB_PUBLIC_KEY_CACHE_KEY, public_key, 60 * 60)\n        return public_key\n\n    def _get_app_code_from_jwt_payload(self, jwt_payload):\n        \"\"\"\u4ecejwt\u91cc\u83b7\u53d6app_code\"\"\"\n        app = jwt_payload.get(\"app\", {})\n\n        verified = app.get(\"verified\", False)\n        if not verified:\n            raise exceptions.AuthenticationFailed(\"app is not verified\")\n\n        app_code = app.get(\"bk_app_code\", \"\") or app.get(\"app_code\", \"\")\n        # \u867d\u7136app_code\u4e3a\u7a7a\u5bf9\u4e8e\u540e\u7eed\u7684\u9274\u6743\u4e00\u5b9a\u662f\u4e0d\u901a\u8fc7\u7684\uff0c\u4f46\u9274\u6743\u4e0d\u901a\u8fc7\u6709\u5f88\u591a\u539f\u56e0\uff0c\u8fd9\u91cc\u63d0\u524dlog\u4fbf\u4e8e\u95ee\u9898\u6392\u67e5\n        if not app_code:\n            raise exceptions.AuthenticationFailed(\"could not get app_code from esb/apigateway jwt payload! it's empty\")\n\n        return app_code\n\n\nclass AppCodeAppSecretAuthentication(BaseAuthentication):\n    \"\"\"\n    \u901a\u8fc7app_code\u548capp_secret\u8fdb\u884c\u9274\u6743\n    \"\"\"\n\n    def authenticate(self, request):\n        # get app_code and app_secret from header\n        app_code = request.META.get(HEADER_APP_CODE_KEY_NAME, \"\")\n        app_secret = request.META.get(HEADER_APP_SECRET_KEY_NAME, \"\")\n\n        if app_code == settings.APP_ID and app_secret == settings.APP_TOKEN:\n            return create_user(\"SAAS\"), None\n\n        return None, None\n\n\nclass MultipleAuthentication(BaseAuthentication):\n    \"\"\"it's a dispatcher\"\"\"\n\n    def authenticate(self, request):\n        # FIXME: \u6700\u7ec8, \u4e0b\u6389token, \u53ea\u4fdd\u7559 jwt + app_code/app_secret\n        # withe list\n        for white_url in settings.AUTH_EXEMPT_PATHS:\n            if re.search(white_url, request.path):\n                logger.debug(\"%s path in white_url<%s>, exempting auth\", request.path, white_url)\n                return None, None\n\n        # app_code and app_secret\n        if HEADER_APP_CODE_KEY_NAME in request.META and HEADER_APP_SECRET_KEY_NAME in request.META:\n            return AppCodeAppSecretAuthentication().authenticate(request)\n\n        # FIXME: should remove this totally\n        # NOTE: some case we want to use token as credentials, call through APIGateway(set default headers)\n        # so we should verify others first, not jwt\n        if get_authorization_header(request) or request.query_params.get(TOKEN_KEY_NAME):\n            # token\n            return InternalTokenAuthentication().authenticate(request)\n\n        # jwt\n        if HEADER_JWT_KEY_NAME in request.META:\n            return ESBOrAPIGatewayAuthentication().authenticate(request)\n\n        raise exceptions.AuthenticationFailed(\n            \"no valid authentication credentials provided! should call through APIGateway/ESB\"\n        )\n", "path": "src/api/bkuser_core/enhanced_account/authentication.py"}]}