diff --git a/addon/access_token.py b/addon/access_token.py index 4f7c3ad..786c5c2 100644 --- a/addon/access_token.py +++ b/addon/access_token.py @@ -1,9 +1,9 @@ import re -from typing import Optional, Any import asyncio +from typing import Optional, Any from mitmproxy.http import HTTPFlow - +from urllib.parse import urlparse, parse_qs from lib.report_vuln import report_vuln @@ -24,13 +24,12 @@ class AccessTokenScanner: print("[TOKENDEBUG] ==scan request==") # URL 검사 - token_result = self._extract_token(request.url) - if token_result: - token, has_bearer = token_result + if self._is_implicit_flow(request.url): + print("[TOKENDEBUG] OAuth Implicit Flow detected.") report_vuln( title="Token Leak in Request URL", - desc=f"요청 URL에 토큰이 포함되어 있습니다 (앞 20자): {token[:20]}…", - status="MEDIUM" if has_bearer else "LOW", + desc="취약한 Grant Type입니다 (Implicit Grant Type)", + status="MEDIUM", uri=request.url ) @@ -39,11 +38,10 @@ class AccessTokenScanner: body_text = request.get_text(strict=False) token_result = self._extract_token(body_text) if token_result: - token, has_bearer = token_result report_vuln( title="Token Leak in Request Body", - desc=f"요청 본문에 토큰이 포함되어 있습니다 (앞 20자): {token[:20]}…", - status="MEDIUM" if has_bearer else "LOW", + desc=f"요청 본문에 토큰이 포함되어 있습니다 (앞 20자): {token_result[:20]}…", + status="LOW", uri=request.url ) @@ -56,28 +54,24 @@ class AccessTokenScanner: if location_header := response.headers.get("Location"): token_result = self._extract_token(location_header) if token_result: - token, has_bearer = token_result - if has_bearer: - report_vuln( - title="Token Leak in Redirect URL (Location header)", - desc=f"Location 헤더에 토큰이 노출되었습니다 (앞 20자): {token[:20]}…", - status="MEDIUM", - uri=location_header, - ) + report_vuln( + title="Token Leak in Redirect URL (Location header)", + desc=f"Location 헤더에 토큰이 노출되었습니다 (앞 20자): {token_result[:20]}…", + status="MEDIUM", + uri=location_header, + ) # Body 검사 (텍스트 컨텐츠인 경우) if response.content: body_text = response.get_text(strict=False) token_result = self._extract_token(body_text) if token_result: - token, has_bearer = token_result - if has_bearer: - report_vuln( - title="Token Leak in Response Body", - desc=f"응답 본문에 토큰이 노출되었습니다 (앞 20자): {token[:20]}…", - status="MEDIUM", - uri=request_url, - ) + report_vuln( + title="Token Leak in Response Body", + desc=f"응답 본문에 토큰이 노출되었습니다 (앞 20자): {token_result[:20]}…", + status="LOW", + uri=request_url, + ) # 토큰 탐지 키워드드 _TOKEN_KEYS = [ @@ -87,8 +81,6 @@ class AccessTokenScanner: "refreshtoken", "auth_token", "session_token", - "secret_token", - "ssoauth", ] # "bearer" 표시가 동시에 존재할 때만 토큰으로 판단하여 false positive를 줄임 @@ -122,6 +114,38 @@ class AccessTokenScanner: if (m := pattern.search(text)) and m.group(1): print(f"[TOKENDEBUG] token: {m.group(1)}") print(f"[TOKENDEBUG] has_bearer: {has_bearer}") - return m.group(1), has_bearer + if has_bearer: + return m.group(1) print("[TOKENDEBUG] No matched.") return None + + def _is_implicit_flow(request_url: str) -> bool: + """ + URL의 파라미터에서 OAuth Implicit Flow 패턴을 체크합니다. + + Args: + request_url: 체크할 요청 URL + + Returns: + bool: client_id, redirect_uri, response_type이 모두 존재하고 + response_type 값이 'token'인 경우 True, 그렇지 않으면 False + """ + try: + parsed_url = urlparse(request_url) + query_params = parse_qs(parsed_url.query) + + # 필요한 파라미터들이 모두 존재하는지 확인 + required_params = ['client_id', 'redirect_uri', 'response_type'] + + for param in required_params: + if param not in query_params: + return False + + # response_type 값이 'token'인지 확인 + response_type_values = query_params.get('response_type', []) + + # response_type 파라미터가 존재하고 값 중에 'token'이 있는지 확인 + return 'token' in response_type_values + + except Exception: + return False \ No newline at end of file