Merge pull request #7 from j93es/nonce_check

nonceCheck

addon/init.py

@@ -3,6 +3,7 @@ import asyncio
 from pkce_check import PKCEDowngradeChecker
 from ScopeDetection import ScopeDetection
 from csrf_check import CsrfChecker
+from nonce_check import NonceChecker
 
 class PKCEAddon:
     def __init__(self):
@@ -49,4 +50,15 @@ class ScopeAddon:
         except Exception as e:
             print(f"[ERROR] ScopeDetection failed: {e}")
 
-addons = [PKCEAddon(), ScopeAddon(), CsrfAddon()]
+class NonceAddon:
+    def __init__(self):
+        self.checker = NonceChecker()
+
+    async def response(self, flow: http.HTTPFlow):
+        try:
+            await self.checker.response(flow)
+        except Exception as e:
+            print(f"[ERROR] NonceAddon failed: {e}")
+            pass
+
+addons = [PKCEAddon(), ScopeAddon(), CsrfAddon(), NonceAddon()]

addon/nonce_check.py

@@ -0,0 +1,88 @@
+import jwt
+from urllib.parse import urlparse, parse_qs
+from typing import Union
+import httpx
+
+import lib.target as target
+from lib.report import save_report
+
+class NonceChecker:
+    def is_oidc_flow(self, flow) -> bool:
+        req = flow.request
+        res = flow.response
+        url = req.pretty_url
+        parsed = urlparse(url)
+        query = parse_qs(parsed.query)
+        location = res.headers.get("location", "")
+        content_type = res.headers.get("content-type", "")
+
+        if "/authorize" in url and "response_type" in query and "openid" in query.get("scope", [""])[0]:
+            return True
+
+        if "application/json" in content_type:
+            if "id_token" in res.text:
+                return True
+
+        if res.status_code in [302, 303]:
+            if isinstance(location, list):
+                location = location[0]
+            if "id_token=" in location:
+                return True
+
+        if "/authorize" in url and "nonce" in query:
+            return True
+
+        return False
+
+
+    def extract_id_token(self, response) -> Union[str, None]:
+        """
+        응답에서 id_token을 추출하는 함수.
+        """
+        # 1. JSON 응답에 id_token 있음
+        try:
+            if "application/json" in response.headers.get("content-type", ""):
+                data = response.json()
+                return data.get("id_token")
+        except Exception:
+            pass
+
+        # 2. Location 헤더에서 id_token 파싱 (예: #id_token=...&access_token=...)
+        location = response.headers.get("location", "")
+        if location:
+            if "#" in location:
+                fragment = location.split("#")[1]
+                params = parse_qs(fragment)
+                return params.get("id_token", [None])[0]
+            elif "?" in location:
+                query = location.split("?")[1]
+                params = parse_qs(query)
+                return params.get("id_token", [None])[0]
+
+        return None
+
+
+    def decode_id_token(self, id_token: str) -> dict:
+        try:
+            return jwt.decode(id_token, options={"verify_signature": False})
+        except Exception as e:
+            return {}
+        
+
+    def check_nonce_in_id_token(self, flow, id_token: str) -> bool:
+        decoded = self.decode_id_token(id_token)
+        nonce = decoded.get("nonce")
+        req = flow.request
+        url = req.pretty_url
+        if not nonce:
+            report_data = [{
+                'target': target.load(),
+                'status': "CRITICAL",
+                'title': "nonce is missing in id_token",
+                'description': "Nonce is present in the request but missing in the id_token.",
+                'uri': f"Original: {url}\nDecoded ID Token: {decoded}",
+            }]
+            save_report(report_data)
+            return False
+        else:
+            return True

pyproject.toml

@@ -10,4 +10,5 @@ dependencies = [
     "httpx>=0.28.1",
     "fastapi[standard]>=0.115.12",
     "granian>=2.3.2",
+    "PyJWT>=2.10.1",
 ]

uv.lock

@@ -755,6 +755,7 @@ dependencies = [
     { name = "granian" },
     { name = "httpx" },
     { name = "mitmproxy" },
+    { name = "pyjwt" },
 ]
 
 [package.metadata]
@@ -764,6 +765,7 @@ requires-dist = [
     { name = "granian", specifier = ">=2.3.2" },
     { name = "httpx", specifier = ">=0.28.1" },
     { name = "mitmproxy", specifier = ">=12.1.1" },
+    { name = "pyjwt", specifier = ">=2.10.1" },
 ]
 
 [[package]]
@@ -916,6 +918,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/8a/0b/9fcc47d19c48b59121088dd6da2488a49d5f72dacf8262e2790a1d2c7d15/pygments-2.19.1-py3-none-any.whl", hash = "sha256:9ea1544ad55cecf4b8242fab6dd35a93bbce657034b0611ee383099054ab6d8c", size = 1225293, upload-time = "2025-01-06T17:26:25.553Z" },
 ]
 
+[[package]]
+name = "pyjwt"
+version = "2.10.1"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/e7/46/bd74733ff231675599650d3e47f361794b22ef3e3770998dda30d3b63726/pyjwt-2.10.1.tar.gz", hash = "sha256:3cc5772eb20009233caf06e9d8a0577824723b44e6648ee0a2aedb6cf9381953", size = 87785, upload-time = "2024-11-28T03:43:29.933Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/61/ad/689f02752eeec26aed679477e80e632ef1b682313be70793d798c1d5fc8f/PyJWT-2.10.1-py3-none-any.whl", hash = "sha256:dcdd193e30abefd5debf142f9adfcdd2b58004e644f25406ffaebd50bd98dacb", size = 22997, upload-time = "2024-11-28T03:43:27.893Z" },
+]
+
 [[package]]
 name = "pylsqpack"
 version = "0.3.22"