Merge pull request #27 from j93es/gyu

[REFACTOR]: Access Token 탐지 조건 완화

.env

@@ -1,2 +1,2 @@
 # Google OAuth 설정
-GOOGLE_ID=oauth.j93es@gmail.com
+GOOGLE_ID=whs.imnya.ng@gmail.com

addon/access_token.py

@@ -135,7 +135,7 @@ class AccessTokenScanner:
             query_params = parse_qs(parsed_url.query)
             
             # 필요한 파라미터들이 모두 존재하는지 확인
-            required_params = ['client_id', 'redirect_uri', 'response_type']
+            required_params = ['redirect_uri', 'response_type']
             
             for param in required_params:
                 if param not in query_params:
@@ -145,7 +145,7 @@ class AccessTokenScanner:
             response_type_values = query_params.get('response_type', [])
             
             # response_type 파라미터가 존재하고 값 중에 'token'이 있는지 확인
-            return 'token' in response_type_values
+            return 'token' in response_type_values or 'id_token' in response_type_values
             
         except Exception:
             return False

addon/client_secret.py

@@ -0,0 +1,29 @@
+from lib.report_vuln import report_vuln
+from urllib.parse import urlparse, parse_qs
+
+class ClientSecret:
+    def get_target_from_query(self, query: str, target: str) -> str | None:
+        if not query:
+            return None
+        parsed = parse_qs(query)
+        scope_values = parsed.get(target, [])
+        if scope_values:
+            return scope_values[0]
+        return None
+
+    async def test(self, flow):
+        req = flow.request
+        
+        parsed = urlparse(req.pretty_url)
+        query = parsed.query
+
+        query_client_id = self.get_target_from_query(query, "client_id")
+        query_client_secret = self.get_target_from_query(query, "client_secret")
+        
+        if query_client_id and query_client_secret:
+            report_vuln(
+                title="OAuth Client Secret Exposure",
+                desc=f"Client ID and Secret found in request: {query_client_id}, {query_client_secret}",
+                status="CRITICAL",
+                uri=req.pretty_url
+            )

addon/google_login_hint.py

@@ -44,8 +44,8 @@ class GoogleLoginHint:
                 
                 # 요청 URL 수정 - URL과 호스트 모두 업데이트
                 flow.request.url = new_url
-                flow.request.pretty_url = new_url
                 print(f"🔄 Modified URL: {new_url}")
+                
     def _is_google_oauth_url(self, url):
         """Google OAuth URL인지 확인"""
         google_oauth_domains = [

addon/google_response_type_token.py

@@ -0,0 +1,52 @@
+from lib.report_vuln import report_vuln
+import httpx
+from lib.utils.is_oauth_uri import is_oauth_uri
+from urllib.parse import urlparse, parse_qs
+
+class GoogleResponseTypeToken:
+    def get_taregt_from_query(self, query: str, target: str) -> str | None:
+        if not query:
+            return None
+        parsed = parse_qs(query)
+        scope_values = parsed.get(target, [])
+        if scope_values:
+            return scope_values[0]
+        return None
+
+    async def test(self, flow):
+        req = flow.request
+        
+        if not is_oauth_uri(req.pretty_url):
+            return
+        
+        if req.pretty_host != "accounts.google.com":
+            return
+        
+        if "response_type=token" in req.pretty_url:
+            return
+        
+        url = f"{req.pretty_url}".replace("response_type=code", "response_type=token")
+        
+        async with httpx.AsyncClient(follow_redirects=True) as cli:
+            response = await cli.request(
+                method=req.method,
+                url=url,
+                headers=req.headers,
+                content=req.get_content(),
+            )
+
+            
+        if response.status_code >= 400:
+            return
+        
+        if "<b>400.</b>" in response.text:
+            return
+        
+        if "response_type=token" in str(response.url):
+            report_vuln(
+                "Google Response Type Token",
+                f"Response type token allowed in {req.pretty_url}",
+                "HIGH",
+                str(response.url)
+            )
+            

addon/init.py

@@ -3,10 +3,11 @@ import asyncio
 from pkce_check import PKCEDowngradeChecker
 from addon.scope_detection import ScopeDetection
 from csrf_check import CsrfChecker
-from nonce_check import NonceChecker
+from client_secret import ClientSecret
-from redirect_uri_check import RedirectBypassChecker
+from addon.open_redirect_check import OpenRedirectChecker
 from access_token import AccessTokenScanner
 from addon.google_login_hint import GoogleLoginHint
+from addon.google_response_type_token import GoogleResponseTypeToken
 import os
 from dotenv import load_dotenv
 from lib.utils.try_catch import try_catch
@@ -17,6 +18,8 @@ false_true_varifing_task = FalseTrueVarifingTask()
 
 load_dotenv(override=True)
 
+_open_redirect_checker = OpenRedirectChecker()
+
 class AddonBase:
     """
     Base class for addons.
@@ -61,8 +64,6 @@ class AddonBase:
             
         return False
 
-
-
     async def request(self, flow: http.HTTPFlow):
         if self.google_login_hint:
             await try_catch(self.google_login_hint.request(flow))
@@ -82,9 +83,10 @@ class AddonBase:
         tasks = [
             try_catch(CsrfChecker().response(flow)),
             try_catch(ScopeDetection().test(flow)),
-            # try_catch(NonceChecker().check_nonce_in_request(flow)),
+            try_catch(ClientSecret().test(flow)),
             try_catch(AccessTokenScanner().scan(flow)),
-            try_catch(RedirectBypassChecker().test(flow)),
+            try_catch(GoogleResponseTypeToken().test(flow)),
+            try_catch(_open_redirect_checker.test(flow)),
         ]
         await asyncio.gather(*tasks)