mirror of
https://github.com/j93es/oauth-backend.git
synced 2026-06-04 06:11:52 +09:00
Compare commits
No commits in common. "main" and "fix/access-token" have entirely different histories.
main
...
fix/access
8 changed files with 1410 additions and 1815 deletions
2
.env
2
.env
|
|
@ -1,2 +1,2 @@
|
||||||
# Google OAuth 설정
|
# Google OAuth 설정
|
||||||
GOOGLE_ID=whs.imnya.ng@gmail.com
|
GOOGLE_ID=oauth.j93es@gmail.com
|
||||||
|
|
|
||||||
|
|
@ -135,7 +135,7 @@ class AccessTokenScanner:
|
||||||
query_params = parse_qs(parsed_url.query)
|
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:
|
for param in required_params:
|
||||||
if param not in query_params:
|
if param not in query_params:
|
||||||
|
|
@ -145,7 +145,7 @@ class AccessTokenScanner:
|
||||||
response_type_values = query_params.get('response_type', [])
|
response_type_values = query_params.get('response_type', [])
|
||||||
|
|
||||||
# response_type 파라미터가 존재하고 값 중에 'token'이 있는지 확인
|
# response_type 파라미터가 존재하고 값 중에 'token'이 있는지 확인
|
||||||
return 'token' in response_type_values or 'id_token' in response_type_values
|
return 'token' in response_type_values
|
||||||
|
|
||||||
except Exception:
|
except Exception:
|
||||||
return False
|
return False
|
||||||
|
|
@ -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
|
|
||||||
)
|
|
||||||
|
|
@ -44,8 +44,8 @@ class GoogleLoginHint:
|
||||||
|
|
||||||
# 요청 URL 수정 - URL과 호스트 모두 업데이트
|
# 요청 URL 수정 - URL과 호스트 모두 업데이트
|
||||||
flow.request.url = new_url
|
flow.request.url = new_url
|
||||||
|
flow.request.pretty_url = new_url
|
||||||
print(f"🔄 Modified URL: {new_url}")
|
print(f"🔄 Modified URL: {new_url}")
|
||||||
|
|
||||||
def _is_google_oauth_url(self, url):
|
def _is_google_oauth_url(self, url):
|
||||||
"""Google OAuth URL인지 확인"""
|
"""Google OAuth URL인지 확인"""
|
||||||
google_oauth_domains = [
|
google_oauth_domains = [
|
||||||
|
|
|
||||||
|
|
@ -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)
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
@ -3,11 +3,10 @@ import asyncio
|
||||||
from pkce_check import PKCEDowngradeChecker
|
from pkce_check import PKCEDowngradeChecker
|
||||||
from addon.scope_detection import ScopeDetection
|
from addon.scope_detection import ScopeDetection
|
||||||
from csrf_check import CsrfChecker
|
from csrf_check import CsrfChecker
|
||||||
from client_secret import ClientSecret
|
from nonce_check import NonceChecker
|
||||||
from addon.open_redirect_check import OpenRedirectChecker
|
from redirect_uri_check import RedirectBypassChecker
|
||||||
from access_token import AccessTokenScanner
|
from access_token import AccessTokenScanner
|
||||||
from addon.google_login_hint import GoogleLoginHint
|
from addon.google_login_hint import GoogleLoginHint
|
||||||
from addon.google_response_type_token import GoogleResponseTypeToken
|
|
||||||
import os
|
import os
|
||||||
from dotenv import load_dotenv
|
from dotenv import load_dotenv
|
||||||
from lib.utils.try_catch import try_catch
|
from lib.utils.try_catch import try_catch
|
||||||
|
|
@ -18,8 +17,6 @@ false_true_varifing_task = FalseTrueVarifingTask()
|
||||||
|
|
||||||
load_dotenv(override=True)
|
load_dotenv(override=True)
|
||||||
|
|
||||||
_open_redirect_checker = OpenRedirectChecker()
|
|
||||||
|
|
||||||
class AddonBase:
|
class AddonBase:
|
||||||
"""
|
"""
|
||||||
Base class for addons.
|
Base class for addons.
|
||||||
|
|
@ -64,6 +61,8 @@ class AddonBase:
|
||||||
|
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
async def request(self, flow: http.HTTPFlow):
|
async def request(self, flow: http.HTTPFlow):
|
||||||
if self.google_login_hint:
|
if self.google_login_hint:
|
||||||
await try_catch(self.google_login_hint.request(flow))
|
await try_catch(self.google_login_hint.request(flow))
|
||||||
|
|
@ -83,10 +82,9 @@ class AddonBase:
|
||||||
tasks = [
|
tasks = [
|
||||||
try_catch(CsrfChecker().response(flow)),
|
try_catch(CsrfChecker().response(flow)),
|
||||||
try_catch(ScopeDetection().test(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(AccessTokenScanner().scan(flow)),
|
||||||
try_catch(GoogleResponseTypeToken().test(flow)),
|
try_catch(RedirectBypassChecker().test(flow)),
|
||||||
try_catch(_open_redirect_checker.test(flow)),
|
|
||||||
]
|
]
|
||||||
await asyncio.gather(*tasks)
|
await asyncio.gather(*tasks)
|
||||||
|
|
||||||
|
|
|
||||||
File diff suppressed because it is too large
Load diff
1400
addon/redirect_uri_check.py
Normal file
1400
addon/redirect_uri_check.py
Normal file
File diff suppressed because it is too large
Load diff
Loading…
Add table
Add a link
Reference in a new issue