Compare commits

..

No commits in common. "main" and "feat/ignore" have entirely different histories.

8 changed files with 1413 additions and 1822 deletions

View file

@ -119,7 +119,7 @@ class AccessTokenScanner:
print("[TOKENDEBUG] No matched.")
return None
def _is_implicit_flow(self, request_url: str) -> bool:
def _is_implicit_flow(request_url: str) -> bool:
"""
URL의 파라미터에서 OAuth Implicit Flow 패턴을 체크합니다.
@ -135,7 +135,7 @@ class AccessTokenScanner:
query_params = parse_qs(parsed_url.query)
# 필요한 파라미터들이 모두 존재하는지 확인
required_params = ['redirect_uri', 'response_type']
required_params = ['client_id', 'redirect_uri', 'response_type']
for param in required_params:
if param not in query_params:
@ -145,7 +145,7 @@ class AccessTokenScanner:
response_type_values = query_params.get('response_type', [])
# response_type 파라미터가 존재하고 값 중에 'token'이 있는지 확인
return 'token' in response_type_values or 'id_token' in response_type_values
return 'token' in response_type_values
except Exception:
return False

View file

@ -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
)

View file

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

View file

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

View file

@ -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 "<b>400.</b>" in response.text:
return
if "response_type=token" in str(response.url):
report_vuln(
"Google Response Type Token",
f"Response type token allowed in {req.pretty_url}",
"HIGH",
str(response.url)
)

View file

@ -3,11 +3,10 @@ import asyncio
from pkce_check import PKCEDowngradeChecker
from addon.scope_detection 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
import os
from dotenv import load_dotenv
from lib.utils.try_catch import try_catch
@ -18,8 +17,6 @@ false_true_varifing_task = FalseTrueVarifingTask()
load_dotenv(override=True)
_open_redirect_checker = OpenRedirectChecker()
class AddonBase:
"""
Base class for addons.
@ -41,8 +38,6 @@ class AddonBase:
".googlesyndication.com",
"cdn.jsdelivr.net",
"update.googleapis.com",
".google-analytics.com",
".gstatic.com"
]
# Ignore .googleapis.com domains
for domain in ignore_domains:
@ -64,14 +59,14 @@ class AddonBase:
return False
async def request(self, flow: http.HTTPFlow):
if self.google_login_hint:
await try_catch(self.google_login_hint.request(flow))
if false_true_varifing_task.is_verifing_false_true():
if false_true_varifing_task.is_verifing_false_true() or self.should_ignore(flow):
return
tasks = [
try_catch(self.google_login_hint.request(flow)) if self.google_login_hint else None,
try_catch(PKCEDowngradeChecker().test(flow)),
]
await asyncio.gather(*tasks)
@ -83,10 +78,9 @@ class AddonBase:
tasks = [
try_catch(CsrfChecker().response(flow)),
try_catch(ScopeDetection().test(flow)),
try_catch(ClientSecret().test(flow)),
# try_catch(NonceChecker().check_nonce_in_request(flow)),
try_catch(AccessTokenScanner().scan(flow)),
try_catch(GoogleResponseTypeToken().test(flow)),
try_catch(_open_redirect_checker.test(flow)),
try_catch(RedirectBypassChecker().test(flow)),
]
await asyncio.gather(*tasks)

File diff suppressed because it is too large Load diff

1400
addon/redirect_uri_check.py Normal file

File diff suppressed because it is too large Load diff