Merge pull request #27 from j93es/gyu

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

addon/access_token.py

@@ -119,7 +119,7 @@ class AccessTokenScanner:
         print("[TOKENDEBUG] No matched.")
         return None
 
-    def _is_implicit_flow(request_url: str) -> bool:
+    def _is_implicit_flow(self, request_url: str) -> bool:
         """
         URL의 파라미터에서 OAuth Implicit Flow 패턴을 체크합니다.
         
@@ -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/csrf_check.py

@@ -9,7 +9,7 @@ from lib.utils.is_oauth_uri import is_oauth_uri
 
 class CsrfChecker:
     nonce_params = {
-        "state", "nonce", "as", "frame_id", "csrf_token", "csrf"
+        "state", "nonce", "csrf_token", "csrf"
     }
 
     def get_header(self, headers: http.Headers, name: str) -> Optional[str]:

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 redirect_uri_check import RedirectBypassChecker
+from client_secret import ClientSecret
+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.
@@ -29,26 +32,61 @@ class AddonBase:
         else:
             self.google_login_hint = None
 
+    def should_ignore(self, flow: http.HTTPFlow) -> bool:
+        """Check if the request should be ignored."""
+        ignore_domains = [
+            ".googleapis.com",
+            "android.clients.google.com",  # Added missing comma here
+            ".adtrafficquality.google",
+            ".googlesyndication.com",
+            "cdn.jsdelivr.net",
+            "update.googleapis.com",
+            ".google-analytics.com",
+            ".gstatic.com"
+        ]
+        # Ignore .googleapis.com domains
+        for domain in ignore_domains:
+            if domain in flow.request.pretty_host:
+                return True
+        
+        # Ignore static files (JS, CSS, fonts, images, etc.)
+        # Split on '?' to remove query parameters before checking extension
+        path = flow.request.path.split('?')[0].lower()
+        static_extensions = [
+            '.js', '.css', '.woff2', '.woff', '.ttf', '.otf', '.svg',
+            '.png', '.jpg', '.jpeg', '.gif', '.webp', '.ico', '.bmp',
+            '.tiff', '.tif', '.webm', '.mp4', '.avi', '.mov', '.pdf', '.md',
+            '.txt', '.csv'
+        ]
+        
+        if any(path.endswith(ext) for ext in static_extensions):
+            return True
+            
+        return False
+
     async def request(self, flow: http.HTTPFlow):
+        if self.google_login_hint:
+            await try_catch(self.google_login_hint.request(flow))
+            
         if false_true_varifing_task.is_verifing_false_true():
             return
             
         tasks = [
-            try_catch(self.google_login_hint.request(flow)) if self.google_login_hint else None,
             try_catch(PKCEDowngradeChecker().test(flow)),
         ]
         await asyncio.gather(*tasks)
 
     async def response(self, flow: http.HTTPFlow):
-        if false_true_varifing_task.is_verifing_false_true():
+        if false_true_varifing_task.is_verifing_false_true() or self.should_ignore(flow):
             return
             
         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)