fix: fix esb jwt get iss (TencentBlueKing#177)

Author: Han-Ya-Jun

sdks/apigw-manager/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "apigw-manager"
-version = "3.0.4"
+version = "3.0.5"
 description = "The SDK for managing blueking gateway resource."
 readme = "README.md"
 authors = ["blueking <blueking@tencent.com>"]

sdks/apigw-manager/src/apigw_manager/apigw/providers.py

@@ -115,13 +115,13 @@ def __init__(self, gateway_name: str, payload: dict) -> None:
 
 class JWTProvider(metaclass=abc.ABCMeta):
     def __init__(
-        self,
-        jwt_key_name: str,
-        default_gateway_name: str,
-        algorithm: str,
-        allow_invalid_jwt_token: bool,
-        public_key_provider: PublicKeyProvider,
-        **kwargs,
+            self,
+            jwt_key_name: str,
+            default_gateway_name: str,
+            algorithm: str,
+            allow_invalid_jwt_token: bool,
+            public_key_provider: PublicKeyProvider,
+            **kwargs,
     ) -> None:
         self.jwt_key_name = jwt_key_name
         self.default_gateway_name = default_gateway_name
@@ -148,6 +148,9 @@ def _decode_jwt(self, jwt_payload, public_key, algorithm):
     def _decode_jwt_header(self, jwt_payload):
         return jwt.get_unverified_header(jwt_payload)
 
+    def _decode_payload(self, jwt_payload):
+        return jwt.decode(jwt_payload, options={"verify_signature": False})
+
     def provide(self, request: HttpRequest) -> Optional[DecodedJWT]:
         jwt_token = request.META.get(self.jwt_key_name, "")
         if not jwt_token:
@@ -156,9 +159,11 @@ def provide(self, request: HttpRequest) -> Optional[DecodedJWT]:
         try:
             jwt_header = self._decode_jwt_header(jwt_token)
             gateway_name = jwt_header.get("kid") or self.default_gateway_name
-            public_key = self.public_key_provider.provide(gateway_name, jwt_header.get("iss"))
+            # 兼容bk-esb签发jwt时未在header里面添加 iss
+            iss = jwt_header.get("iss") or self._decode_payload(jwt_token).get("iss", "")
+            public_key = self.public_key_provider.provide(gateway_name, iss)
             if not public_key:
-                logger.warning("no public key found, gateway=%s, issuer=%s", gateway_name, jwt_header.get("iss"))
+                logger.warning("no public key found, gateway=%s, issuer=%s", gateway_name, iss)
                 return None
 
             algorithm = jwt_header.get("alg") or self.algorithm