mirror of
https://github.com/j93es/oauth-backend.git
synced 2026-06-04 03:41:52 +09:00
[Update] 자동 오탐 검증을 위한 라우터 추가
This commit is contained in:
parent
53db0fb14e
commit
3c5db3c1fd
5 changed files with 188 additions and 23 deletions
|
|
@ -76,10 +76,10 @@ 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="HIGH", uri=flow.request.url)
|
||||
report_vuln(title="CSRF Risk", desc="Missing nonce in redirect response", status="CRITICAL", uri=flow.request.url)
|
||||
return 1
|
||||
if orig_nonce != resp_nonce:
|
||||
report_vuln(title="CSRF Risk", desc="Nonce mismatch request↔response", status="MEDIUM", uri=flow.request.url)
|
||||
report_vuln(title="CSRF Risk", desc="Nonce mismatch request↔response", status="HIGH", uri=flow.request.url)
|
||||
return 1
|
||||
|
||||
return 0
|
||||
|
|
@ -103,11 +103,11 @@ 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="HIGH", uri=flow.request.url)
|
||||
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 교체 쿼리
|
||||
|
|
@ -120,7 +120,7 @@ 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="MEDIUM", uri=flow.request.url)
|
||||
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
|
||||
|
|
@ -130,7 +130,7 @@ class CsrfChecker:
|
|||
|
||||
# 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="HIGH", uri=flow.request.url)
|
||||
report_vuln(title="CSRF Risk", desc="Missing nonce in OAuth request", status="CRITICAL", uri=flow.request.url)
|
||||
return
|
||||
|
||||
# 2) 리다이렉트에서 nonce 검사
|
||||
|
|
|
|||
|
|
@ -9,6 +9,10 @@ from access_token import AccessTokenScanner
|
|||
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}")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue