whs-authz-authn/oauth-backend
1
0
Fork 0
mirror of https://github.com/j93es/oauth-backend.git synced 2026-06-04 02:31:51 +09:00
Code Issues Projects Releases Packages Wiki Activity

[Update] save vuln report logic

Browse source
This commit is contained in:
tv0924@icloud.com 2025-06-26 12:20:41 +09:00
commit 3a1422a2f2
9 changed files with 121 additions and 190 deletions
Download patch file Download diff file Expand all files Collapse all files

87
addon/access_token.py
View file

@ -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.cur_target_url as cur_target_url
from lib.report_vuln 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 = cur_target_url.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",

58
addon/csrf_check.py
View file

@ -4,23 +4,14 @@ from urllib.parse import urlparse, parse_qs, unquote
import httpx
from typing import Optional, Union, List
import lib.cur_target_url as cur_target_url
from lib.report_vuln 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,19 @@ 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="HIGH", 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="MEDIUM", 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,7 +103,8 @@ 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="HIGH", uri=flow.request.url)
return 1
# (2) 두 번의 리다이렉트 비교
async with httpx.AsyncClient(follow_redirects=False) as cli:
@ -127,42 +120,25 @@ 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="MEDIUM", 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="HIGH", 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': cur_target_url.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

2
addon/init.py
View file

@ -107,4 +107,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()]

20
addon/nonce_check.py
View file

@ -1,10 +1,8 @@
import jwt
from urllib.parse import urlparse, parse_qs
from typing import Union
import httpx
import lib.cur_target_url as cur_target_url
from lib.report_vuln 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': cur_target_url.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

57
addon/pkce_check.py
View file

@ -4,7 +4,7 @@ import httpx
from typing import Dict, List
import lib.cur_target_url as cur_target_url
from lib.report_vuln import save_report
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": cur_target_url.load(),
"status": status,
"title": title,
"description": description,
"uri": uri,
}
def save(self, report_data: List[Dict[str, str]]):
save_report(report_data)

37
addon/redirect_uri_check.py
View file

@ -4,8 +4,7 @@ import asyncio
import random
import time
from urllib.parse import urlparse, parse_qs, urlencode, urlunparse
import lib.cur_target_url as cur_target_url
from lib.report_vuln 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": cur_target_url.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}")

16
addon/scope_detection.py
View file

@ -1,5 +1,5 @@
import lib.cur_target_url as cur_target_url
from lib.report_vuln import save_report
from lib.report_vuln import report_vuln
class ScopeDetection:
def get_scope_from_query(self, query: str) -> str | None:
@ -43,11 +43,9 @@ class ScopeDetection:
result = await self.check_scope(flow)
if result != 0:
report_data = [{
'target': cur_target_url.load(),
'status': "WARNING",
'title': "OAuth scope value issue",
'description': f"{method} {url}: {', '.join(result)}",
'uri': url
}]
save_report(report_data)
report_vuln(
title="OAuth Scope Value Issue",
desc=f"Detected scope value issue in {method} {url}: {', '.join(result)}",
status="WARNING",
uri=url
)

22
lib/report_vuln.py
View file

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