mirror of
https://github.com/j93es/oauth-backend.git
synced 2026-06-04 06:21:51 +09:00
James
GitHub
commit
00c81f365a
16 changed files with 397 additions and 256 deletions
2
.gitignore
vendored
2
.gitignore
vendored
|
|
@ -9,6 +9,8 @@ wheels/
|
|||
# Virtual environments
|
||||
.venv
|
||||
|
||||
.env
|
||||
|
||||
data/
|
||||
|
||||
|
||||
|
|
|
|||
19
README.md
19
README.md
|
|
@ -27,7 +27,7 @@ venv와 패키지가 설치가 됩니다.
|
|||
> 그렇지 않으면 실행되지 않습니다.
|
||||
>
|
||||
> 윈도우 환경에서는 `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`으로 인증합니다.
|
||||
|
|
@ -58,8 +58,15 @@ class LoggerAddon:
|
|||
self.checker = Example()
|
||||
|
||||
def request(self, flow: http.HTTPFlow): # 비동기가 필요할 경우 async def로 할 것
|
||||
self.checker.test(flow)
|
||||
# 오탐 검사하고 있을때는 검증하지 않음
|
||||
if false_true_varifing_task.is_verifing_false_true():
|
||||
return
|
||||
self.checker.test(flow)
|
||||
|
||||
def response(self, flow: http.HTTPFlow): # 비동기가 필요할 경우 async def로 할 것
|
||||
# 오탐 검사하고 있을때는 검증하지 않음
|
||||
if false_true_varifing_task.is_verifing_false_true():
|
||||
return
|
||||
self.checker.test(flow)
|
||||
|
||||
```
|
||||
|
|
@ -88,3 +95,11 @@ class Example:
|
|||
```
|
||||
|
||||
이러한 예제를 참고하여 작성하여주세요.
|
||||
|
||||
# 백엔드 API DOCS
|
||||
|
||||
`uv run main.py`으로 백엔드를 실행한 후에, 다음의 url에 접속합니다.
|
||||
|
||||
```
|
||||
http://localhost:11081/redoc
|
||||
```
|
||||
|
|
|
|||
|
|
@ -1,53 +0,0 @@
|
|||
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)
|
||||
|
|
@ -1,23 +1,11 @@
|
|||
import re
|
||||
from dataclasses import dataclass, asdict
|
||||
from typing import List, Dict, Optional, Any
|
||||
import asyncio
|
||||
|
||||
from mitmproxy.http import HTTPFlow
|
||||
|
||||
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)}
|
||||
from lib.report_vuln import report_vuln
|
||||
|
||||
|
||||
# 요청/응답에서 액세스 토큰 노출 여부를 검사하는 스캐너
|
||||
|
|
@ -26,31 +14,25 @@ class AccessTokenScanner:
|
|||
async def scan(self, flow: HTTPFlow) -> None:
|
||||
"""단일 HTTPFlow(request + response)에 대해 요청과 응답을 모두 검사."""
|
||||
print(f"[TOKENDEBUG] Request URL: {flow.request.url}")
|
||||
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_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)
|
||||
|
||||
# 내부 구현
|
||||
async def _scan_request(self, request: Any) -> List[TokenLeakResult]:
|
||||
results: List[TokenLeakResult] = []
|
||||
async def _scan_request(self, request: Any):
|
||||
|
||||
print("[TOKENDEBUG] ==scan 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"
|
||||
)
|
||||
report_vuln(
|
||||
title="Token Leak in Request URL",
|
||||
desc=f"요청 URL에 토큰이 포함되어 있습니다 (앞 20자): {token[:20]}…",
|
||||
status="MEDIUM" if has_bearer else "LOW",
|
||||
uri=request.url
|
||||
)
|
||||
|
||||
# Body 검사 (텍스트 컨텐츠인 경우)
|
||||
|
|
@ -59,22 +41,17 @@ class AccessTokenScanner:
|
|||
token_result = self._extract_token(body_text)
|
||||
if token_result:
|
||||
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"
|
||||
)
|
||||
report_vuln(
|
||||
title="Token Leak in Request Body",
|
||||
desc=f"요청 본문에 토큰이 포함되어 있습니다 (앞 20자): {token[:20]}…",
|
||||
status="MEDIUM" if has_bearer else "LOW",
|
||||
uri=request.url
|
||||
)
|
||||
|
||||
return results
|
||||
|
||||
async def _scan_response(self, response: Optional[Any], request_url: str) -> List[TokenLeakResult]:
|
||||
async def _scan_response(self, response: Optional[Any], request_url: str):
|
||||
if response is None:
|
||||
return []
|
||||
return
|
||||
|
||||
results: List[TokenLeakResult] = []
|
||||
print("[TOKENDEBUG] ==scan response==")
|
||||
# Location 헤더 검사 (리다이렉트)
|
||||
if location_header := response.headers.get("Location"):
|
||||
|
|
@ -82,12 +59,11 @@ class AccessTokenScanner:
|
|||
if token_result:
|
||||
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,
|
||||
)
|
||||
report_vuln(
|
||||
title="Token Leak in Redirect URL (Location header)",
|
||||
desc=f"Location 헤더에 토큰이 노출되었습니다 (앞 20자): {token[:20]}…",
|
||||
status="MEDIUM",
|
||||
uri=location_header,
|
||||
)
|
||||
|
||||
# Body 검사 (텍스트 컨텐츠인 경우)
|
||||
|
|
@ -97,16 +73,13 @@ class AccessTokenScanner:
|
|||
if token_result:
|
||||
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,
|
||||
)
|
||||
report_vuln(
|
||||
title="Token Leak in Response Body",
|
||||
desc=f"응답 본문에 토큰이 노출되었습니다 (앞 20자): {token[:20]}…",
|
||||
status="MEDIUM",
|
||||
uri=request_url,
|
||||
)
|
||||
|
||||
return results
|
||||
|
||||
# 토큰 탐지 키워드드
|
||||
_TOKEN_KEYS = [
|
||||
"access_token",
|
||||
|
|
|
|||
|
|
@ -4,23 +4,14 @@ from urllib.parse import urlparse, parse_qs, unquote
|
|||
import httpx
|
||||
from typing import Optional, Union, List
|
||||
|
||||
import lib.target as target
|
||||
from lib.report import save_report
|
||||
from lib.report_vuln import report_vuln
|
||||
from lib.utils.is_oauth_uri import is_oauth_uri
|
||||
|
||||
class CsrfChecker:
|
||||
nonce_params = {
|
||||
"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)
|
||||
|
|
@ -40,7 +31,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 self.is_oauth_uri(loc)
|
||||
return 300 <= code < 400 and is_oauth_uri(loc)
|
||||
|
||||
def check_nonce_in_request(self, flow: http.HTTPFlow) -> bool:
|
||||
qs = parse_qs(urlparse(flow.request.url).query)
|
||||
|
|
@ -71,10 +62,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 (self.is_oauth_uri(flow.request.url)
|
||||
if not (is_oauth_uri(flow.request.url)
|
||||
and self.check_nonce_in_request(flow)
|
||||
and self.is_oauth_redirect(flow)):
|
||||
return 0
|
||||
|
|
@ -85,18 +76,29 @@ class CsrfChecker:
|
|||
resp_nonce = self.get_query_param(loc, param) if param else None
|
||||
|
||||
if resp_nonce is None:
|
||||
return ["Missing nonce in redirect"]
|
||||
report_vuln(
|
||||
title="CSRF Risk",
|
||||
desc="Missing nonce in redirect response",
|
||||
status="CRITICAL",
|
||||
uri=flow.request.url
|
||||
)
|
||||
return 1
|
||||
if orig_nonce != resp_nonce:
|
||||
return ["Nonce mismatch request↔response"]
|
||||
report_vuln(
|
||||
title="CSRF Risk",
|
||||
desc="Nonce mismatch request↔response",
|
||||
status="HIGH",
|
||||
uri=flow.request.url
|
||||
)
|
||||
return 1
|
||||
|
||||
return 0
|
||||
|
||||
async def check_nonce_reuse(self, flow: http.HTTPFlow) -> Union[int, List[str]]:
|
||||
# OAuth Request가 아니면서, OAuth 리다이렉트 응답인 경우만 검사
|
||||
if self.is_oauth_uri(flow.request.url) or not self.is_oauth_redirect(flow):
|
||||
if 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)
|
||||
|
|
@ -111,10 +113,16 @@ class CsrfChecker:
|
|||
if new_nonce is None:
|
||||
return 0
|
||||
if new_nonce == orig_nonce:
|
||||
return ["Nonce reused without cookies"]
|
||||
report_vuln(
|
||||
title="CSRF Risk",
|
||||
desc="Nonce reused without cookies",
|
||||
status="CRITICAL",
|
||||
uri=flow.request.url
|
||||
)
|
||||
return 1
|
||||
|
||||
# (2) 두 번의 리다이렉트 비교
|
||||
async with httpx.AsyncClient(follow_redirects=False) as cli:
|
||||
async with httpx.AsyncClient(follow_redirects=True) as cli:
|
||||
# 원본 쿼리
|
||||
req1 = await cli.get(loc0, params=qs0, headers=flow.request.headers)
|
||||
# nonce 교체 쿼리
|
||||
|
|
@ -127,42 +135,35 @@ class CsrfChecker:
|
|||
and urlparse(req1.headers.get("location", "")).path
|
||||
== urlparse(req2.headers.get("location", "")).path
|
||||
):
|
||||
return ["Identical redirects on nonce swap → potential CSRF"]
|
||||
report_vuln(
|
||||
title="CSRF Risk",
|
||||
desc="Identical redirects on nonce swap → potential CSRF",
|
||||
status="NOT-VERIFIED-HIGH",
|
||||
uri=flow.request.url
|
||||
)
|
||||
return 1
|
||||
|
||||
return 0
|
||||
|
||||
async def response(self, flow: http.HTTPFlow) -> None:
|
||||
try:
|
||||
msgs: List[str] = []
|
||||
|
||||
# 1) 요청에 nonce 없으면
|
||||
if self.is_oauth_uri(flow.request.url) and not self.check_nonce_in_request(flow):
|
||||
msgs.append("Missing state/nonce in request")
|
||||
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
|
||||
|
||||
# 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
|
||||
|
|
|
|||
|
|
@ -1,14 +1,18 @@
|
|||
from mitmproxy import http
|
||||
import asyncio
|
||||
from pkce_check import PKCEDowngradeChecker
|
||||
from ScopeDetection import ScopeDetection
|
||||
from addon.scope_detection import ScopeDetection
|
||||
from csrf_check import CsrfChecker
|
||||
from nonce_check import NonceChecker
|
||||
from redirect_uri_check import RedirectBypassChecker
|
||||
from access_token import AccessTokenScanner
|
||||
from GoogleLoginHint import GoogleLoginHint
|
||||
from addon.google_login_hint import GoogleLoginHint
|
||||
import os
|
||||
from dotenv import load_dotenv
|
||||
from lib.false_true_varifing_task import FalseTrueVarifingTask
|
||||
|
||||
# Initialize the singleton task manager
|
||||
false_true_varifing_task = FalseTrueVarifingTask()
|
||||
|
||||
load_dotenv(override=True)
|
||||
|
||||
|
|
@ -21,6 +25,10 @@ class PKCEAddon:
|
|||
f"[DEBUG] Processing request: {flow.request.method} {flow.request.pretty_url}"
|
||||
)
|
||||
try:
|
||||
# 오탐 검사하고 있을때는 검증하지 않음
|
||||
if false_true_varifing_task.is_verifing_false_true():
|
||||
return
|
||||
|
||||
await self.checker.test(flow)
|
||||
except Exception as e:
|
||||
print(f"[ERROR] Addon failed: {e}")
|
||||
|
|
@ -33,6 +41,9 @@ class CsrfAddon:
|
|||
|
||||
async def response(self, flow: http.HTTPFlow):
|
||||
try:
|
||||
# 오탐 검사하고 있을때는 검증하지 않음
|
||||
if false_true_varifing_task.is_verifing_false_true():
|
||||
return
|
||||
await self.checker.response(flow)
|
||||
except Exception as e:
|
||||
print(f"[ERROR] CSRF Addon failed: {e}")
|
||||
|
|
@ -42,21 +53,18 @@ class CsrfAddon:
|
|||
class ScopeAddon:
|
||||
def __init__(self):
|
||||
self.checker = ScopeDetection()
|
||||
self._flow_map = {} # 요청 정보를 저장
|
||||
|
||||
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:
|
||||
# 오탐 검사하고 있을때는 검증하지 않음
|
||||
if false_true_varifing_task.is_verifing_false_true():
|
||||
return
|
||||
await self.checker.test(flow)
|
||||
except Exception as e:
|
||||
print(f"[ERROR] ScopeDetection failed: {e}")
|
||||
|
||||
|
||||
|
||||
class NonceAddon:
|
||||
def __init__(self):
|
||||
self.checker = NonceChecker()
|
||||
|
|
@ -70,12 +78,17 @@ class NonceAddon:
|
|||
print(f"[ERROR] NonceAddon failed: {e}")
|
||||
pass
|
||||
|
||||
|
||||
|
||||
class AccessTokenAddon:
|
||||
def __init__(self):
|
||||
self.checker = AccessTokenScanner()
|
||||
|
||||
async def response(self, flow: http.HTTPFlow):
|
||||
try:
|
||||
# 오탐 검사하고 있을때는 검증하지 않음
|
||||
if false_true_varifing_task.is_verifing_false_true():
|
||||
return
|
||||
await self.checker.scan(flow)
|
||||
except Exception as e:
|
||||
print(f"[ERROR] AccessToken Addon failed: {e}")
|
||||
|
|
@ -88,6 +101,9 @@ class RedirectBypassAddon:
|
|||
# request 대신 response 로 바꿔 보세요:
|
||||
async def response(self, flow: http.HTTPFlow):
|
||||
try:
|
||||
# 오탐 검사하고 있을때는 검증하지 않음
|
||||
if false_true_varifing_task.is_verifing_false_true():
|
||||
return
|
||||
await self.checker.test(flow)
|
||||
except Exception as e:
|
||||
print(f"[ERROR] RedirectBypass Addon failed: {e}")
|
||||
|
|
@ -107,4 +123,4 @@ class GoogleLoginHintAddon():
|
|||
except Exception as e:
|
||||
print(f"[ERROR] GoogleLoginHint Addon failed: {e}")
|
||||
|
||||
addons = [PKCEAddon(), ScopeAddon(), CsrfAddon(), NonceAddon(), AccessTokenAddon(), RedirectBypassAddon(), GoogleLoginHintAddon()]
|
||||
addons = [PKCEAddon(), ScopeAddon(), CsrfAddon(), NonceAddon(), AccessTokenAddon(), GoogleLoginHintAddon(), RedirectBypassAddon()]
|
||||
|
|
|
|||
|
|
@ -1,10 +1,8 @@
|
|||
import jwt
|
||||
from urllib.parse import urlparse, parse_qs
|
||||
from typing import Union
|
||||
import httpx
|
||||
|
||||
import lib.target as target
|
||||
from lib.report import save_report
|
||||
from lib.report_vuln import report_vuln
|
||||
|
||||
class NonceChecker:
|
||||
def is_oidc_flow(self, flow) -> bool:
|
||||
|
|
@ -72,17 +70,13 @@ 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_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)
|
||||
report_vuln(
|
||||
title="Nonce Check Failed",
|
||||
desc="id_token에 nonce가 없습니다.",
|
||||
status="HIGH",
|
||||
uri=flow.request.url
|
||||
)
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
|
|
|
|||
|
|
@ -3,8 +3,8 @@ import asyncio
|
|||
import httpx
|
||||
from typing import Dict, List
|
||||
|
||||
import lib.target as target
|
||||
from lib.report import save_report
|
||||
import lib.cur_target_url as cur_target_url
|
||||
from lib.report_vuln import report_vuln
|
||||
|
||||
|
||||
class PKCEDowngradeChecker:
|
||||
|
|
@ -58,27 +58,19 @@ class PKCEDowngradeChecker:
|
|||
|
||||
async def report_missing_parameters(self, url: str, is_openid: bool):
|
||||
status = "MEDIUM" if is_openid else "LOW"
|
||||
self.save(
|
||||
[
|
||||
self.make_report(
|
||||
status,
|
||||
"PKCE Parameters Missing",
|
||||
"PKCE parameters are missing or incomplete.",
|
||||
url,
|
||||
)
|
||||
]
|
||||
report_vuln(
|
||||
title="PKCE Parameters Missing",
|
||||
desc="PKCE parameters are missing or incomplete.",
|
||||
status=status,
|
||||
uri=url,
|
||||
)
|
||||
|
||||
async def report_plain_method(self, url: str):
|
||||
self.save(
|
||||
[
|
||||
self.make_report(
|
||||
"CRITICAL",
|
||||
"PKCE Plain Method",
|
||||
"PKCE method is set to 'plain'. Possible downgrade.",
|
||||
url,
|
||||
)
|
||||
]
|
||||
report_vuln(
|
||||
title="PKCE Plain Method",
|
||||
desc="PKCE method is set to 'plain'. Possible downgrade.",
|
||||
status="CRITICAL",
|
||||
uri=url,
|
||||
)
|
||||
|
||||
def create_downgraded_url(self, parsed, query):
|
||||
|
|
@ -150,15 +142,11 @@ class PKCEDowngradeChecker:
|
|||
else:
|
||||
return # Likely safe
|
||||
|
||||
self.save(
|
||||
[
|
||||
self.make_report(
|
||||
status,
|
||||
title,
|
||||
description,
|
||||
f"Original: {original_url}\nDowngraded: {downgraded_url}",
|
||||
)
|
||||
]
|
||||
report_vuln(
|
||||
title=title,
|
||||
desc=description,
|
||||
status=status,
|
||||
uri=f"Original: {original_url}\nDowngraded: {downgraded_url}",
|
||||
)
|
||||
|
||||
def same_redirect_destination(self, orig_loc, down_loc):
|
||||
|
|
@ -166,16 +154,3 @@ 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)
|
||||
|
|
|
|||
|
|
@ -4,8 +4,7 @@ import asyncio
|
|||
import random
|
||||
import time
|
||||
from urllib.parse import urlparse, parse_qs, urlencode, urlunparse
|
||||
import lib.target as target
|
||||
from lib.report import save_report
|
||||
from lib.report_vuln import report_vuln
|
||||
|
||||
class RedirectRateLimiter:
|
||||
"""redirect_uri_check 전용 rate limiter"""
|
||||
|
|
@ -1287,13 +1286,21 @@ class RedirectBypassChecker:
|
|||
except:
|
||||
return ""
|
||||
|
||||
def _is_code_in_location(self, location: str) -> bool:
|
||||
return self._extract_code_from_location(location) != ""
|
||||
|
||||
""" mitmproxy용 - 실제 OAuth 플로우 가로채서 분석 """
|
||||
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:
|
||||
|
||||
# location 헤더에 code가 없으면 스킵
|
||||
location = flow.response.headers.get("Location", "")
|
||||
if not self._is_code_in_location(location):
|
||||
return
|
||||
|
||||
if not query or "redirect_uri" not in query:
|
||||
return
|
||||
|
||||
original_redirect_uri = query["redirect_uri"][0]
|
||||
|
|
@ -1347,15 +1354,13 @@ class RedirectBypassChecker:
|
|||
# 리다이렉트 응답이 아니면 스킵
|
||||
if status not in [301, 302, 303, 307, 308]:
|
||||
return False
|
||||
|
||||
# Location 헤더에서 code 추출
|
||||
auth_code = self._extract_code_from_location(location)
|
||||
|
||||
# 베이스라인 검증
|
||||
is_valid = self._is_baseline_valid(bypassed_uri, original_url)
|
||||
|
||||
if auth_code and not is_valid:
|
||||
|
||||
if self._is_code_in_location(location) and not is_valid:
|
||||
# 취약점 발견 시에만 로그
|
||||
auth_code = self._extract_code_from_location(location)
|
||||
print(f"[🎯 VULNERABILITY] {payload.name} 우회 성공!")
|
||||
await self._report_vulnerability(original_url, test_url, bypassed_uri, location, auth_code, payload)
|
||||
return True
|
||||
|
|
@ -1384,14 +1389,12 @@ class RedirectBypassChecker:
|
|||
f"• 발급된 인가 코드: {auth_code[:10]}...\n\n"
|
||||
)
|
||||
|
||||
report_data = [{
|
||||
"target": target.load(),
|
||||
"status": "CRITICAL",
|
||||
"title": "Redirect URI Bypass Vulnerability",
|
||||
"description": description,
|
||||
"uri": test_url # uri 필드 추가
|
||||
}]
|
||||
report_vuln(
|
||||
title="Redirect URI Bypass Vulnerability",
|
||||
desc=description,
|
||||
status="CRITICAL",
|
||||
uri=test_url
|
||||
)
|
||||
|
||||
save_report(report_data)
|
||||
print(f"[🎯 CRITICAL] Redirect URI 우회 취약점 발견 및 보고 완료!")
|
||||
print(f"[INFO] 패턴: {pattern_name}, 우회 URI: {bypassed_uri}")
|
||||
48
addon/scope_detection.py
Normal file
48
addon/scope_detection.py
Normal file
|
|
@ -0,0 +1,48 @@
|
|||
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 check_scope(self, flow):
|
||||
req = flow.request
|
||||
res = flow.response
|
||||
|
||||
parsed = urlparse(req.pretty_url)
|
||||
query = parsed.query
|
||||
|
||||
location = res.headers.get("Location", "")
|
||||
location_query = urlparse(location).query
|
||||
|
||||
query_scope = self.get_scope_from_query(query)
|
||||
location_scope = self.get_scope_from_query(location_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
|
||||
)
|
||||
if location_scope in ["all", "*"]:
|
||||
report_vuln(
|
||||
title="OAuth Scope Value Issue",
|
||||
desc=f"Scope value issue detected in response location: {location_scope}",
|
||||
status="WARNING",
|
||||
uri=location
|
||||
)
|
||||
|
||||
async def test(self, flow):
|
||||
|
||||
if not is_oauth_uri(flow.request.pretty_url):
|
||||
return
|
||||
|
||||
await self.check_scope(flow)
|
||||
65
lib/false_true_varifing_task.py
Normal file
65
lib/false_true_varifing_task.py
Normal file
|
|
@ -0,0 +1,65 @@
|
|||
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
|
||||
|
|
@ -1,14 +1,16 @@
|
|||
# save as data/report.csv
|
||||
import os
|
||||
import csv
|
||||
from typing import List, Dict, Any
|
||||
from mitmproxy import http
|
||||
import lib.cur_target_url as cur_target_url
|
||||
|
||||
# target, status, title, description, uri
|
||||
|
||||
# file path는 '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)
|
||||
def report_vuln(title: str, desc: str, status: str, uri: str) -> None:
|
||||
file_path: str = 'data/report.csv'
|
||||
|
||||
os.makedirs(os.path.dirname(file_path), exist_ok=True)
|
||||
|
||||
"""
|
||||
report_data 안의 각 레포트를 한 줄씩 CSV에 추가로 저장합니다.
|
||||
|
|
@ -23,10 +25,10 @@ def save_report(report_data: List[Dict[str, Any]], file_path: str = 'data/report
|
|||
if not file_exists:
|
||||
writer.writeheader()
|
||||
|
||||
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)
|
||||
writer.writerow({
|
||||
'target': cur_target_url.load(),
|
||||
'status': status,
|
||||
'title': title,
|
||||
'description': desc,
|
||||
'uri': uri,
|
||||
})
|
||||
10
lib/utils/is_oauth_uri.py
Normal file
10
lib/utils/is_oauth_uri.py
Normal file
|
|
@ -0,0 +1,10 @@
|
|||
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
|
||||
|
|
@ -1,17 +1,107 @@
|
|||
from fastapi import FastAPI, Query, HTTPException
|
||||
from fastapi.responses import Response
|
||||
import lib.target as target
|
||||
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()
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
|
||||
@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.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.exception_handler(404)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue