mirror of
https://github.com/j93es/oauth-backend.git
synced 2026-06-04 07:51:51 +09:00
Compare commits
4 commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2bfe057889 | ||
|
|
568d3f0ce5 | ||
|
|
4059cc7adb | ||
|
|
cba2d545b6 |
2 changed files with 96 additions and 1 deletions
76
addon/cleintsecret_check.py
Normal file
76
addon/cleintsecret_check.py
Normal file
|
|
@ -0,0 +1,76 @@
|
||||||
|
# clientsecret_check.py
|
||||||
|
from mitmproxy import http
|
||||||
|
from urllib.parse import urlparse, parse_qs
|
||||||
|
from typing import List
|
||||||
|
import lib.target as target
|
||||||
|
from lib.report import save_report
|
||||||
|
|
||||||
|
|
||||||
|
class ClientSecretChecker:
|
||||||
|
def is_oauth_uri(self, uri: str) -> bool:
|
||||||
|
qs = parse_qs(urlparse(uri).query)
|
||||||
|
keys = qs.keys()
|
||||||
|
return "client_id" in keys and ("client_secret" in keys or "client_secret" in uri)
|
||||||
|
|
||||||
|
def has_client_secret_in_uri(self, uri: str) -> bool:
|
||||||
|
qs = parse_qs(urlparse(uri).query)
|
||||||
|
return "client_secret" in qs
|
||||||
|
|
||||||
|
def is_post_with_client_secret(self, flow: http.HTTPFlow) -> bool:
|
||||||
|
if flow.request.method != "POST":
|
||||||
|
return False
|
||||||
|
content_type = flow.request.headers.get("Content-Type", "")
|
||||||
|
if "application/x-www-form-urlencoded" in content_type:
|
||||||
|
body = parse_qs(flow.request.get_text())
|
||||||
|
return "client_secret" in body
|
||||||
|
return False
|
||||||
|
|
||||||
|
def is_exposed_in_referer(self, flow: http.HTTPFlow) -> bool:
|
||||||
|
referer = flow.request.headers.get("Referer", "")
|
||||||
|
return "client_secret" in referer
|
||||||
|
|
||||||
|
def check_client_secret_leak(self, flow: http.HTTPFlow) -> List[str]:
|
||||||
|
messages = []
|
||||||
|
|
||||||
|
if self.has_client_secret_in_uri(flow.request.url):
|
||||||
|
messages.append("client_secret found in URL query string")
|
||||||
|
|
||||||
|
if self.is_post_with_client_secret(flow):
|
||||||
|
messages.append("client_secret found in POST body")
|
||||||
|
|
||||||
|
if self.is_exposed_in_referer(flow):
|
||||||
|
messages.append("client_secret exposed in Referer header")
|
||||||
|
|
||||||
|
return messages
|
||||||
|
|
||||||
|
def _report(self, flow: http.HTTPFlow, issues: List[str], direction: str):
|
||||||
|
desc = " | ".join(issues)
|
||||||
|
report_data = [{
|
||||||
|
'target': target.load(),
|
||||||
|
'status': "HIGH",
|
||||||
|
'title': f"OAuth Client Secret Exposure ({direction})",
|
||||||
|
'description': desc,
|
||||||
|
'uri': flow.request.url,
|
||||||
|
}]
|
||||||
|
save_report(report_data)
|
||||||
|
print(f"[INFO] Client Secret Leak Detected ({direction}): {desc}")
|
||||||
|
|
||||||
|
def request(self, flow: http.HTTPFlow) -> None:
|
||||||
|
try:
|
||||||
|
if not self.is_oauth_uri(flow.request.url):
|
||||||
|
return
|
||||||
|
issues = self.check_client_secret_leak(flow)
|
||||||
|
if issues:
|
||||||
|
self._report(flow, issues, "request")
|
||||||
|
except Exception as e:
|
||||||
|
print(f"[ERROR] Client Secret Check (request) failed: {e}")
|
||||||
|
|
||||||
|
def response(self, flow: http.HTTPFlow) -> None:
|
||||||
|
try:
|
||||||
|
if not self.is_oauth_uri(flow.request.url):
|
||||||
|
return
|
||||||
|
issues = self.check_client_secret_leak(flow)
|
||||||
|
if issues:
|
||||||
|
self._report(flow, issues, "response")
|
||||||
|
except Exception as e:
|
||||||
|
print(f"[ERROR] Client Secret Check (response) failed: {e}")
|
||||||
|
|
@ -4,6 +4,7 @@ from pkce_check import PKCEDowngradeChecker
|
||||||
from ScopeDetection import ScopeDetection
|
from ScopeDetection import ScopeDetection
|
||||||
from csrf_check import CsrfChecker
|
from csrf_check import CsrfChecker
|
||||||
from nonce_check import NonceChecker
|
from nonce_check import NonceChecker
|
||||||
|
from cleintsecret_check import ClientSecretChecker
|
||||||
from redirect_uri_check import RedirectBypassChecker
|
from redirect_uri_check import RedirectBypassChecker
|
||||||
from access_token import AccessTokenScanner
|
from access_token import AccessTokenScanner
|
||||||
|
|
||||||
|
|
@ -65,6 +66,24 @@ class NonceAddon:
|
||||||
print(f"[ERROR] NonceAddon failed: {e}")
|
print(f"[ERROR] NonceAddon failed: {e}")
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
class ClientSecretAddon:
|
||||||
|
def __init__(self):
|
||||||
|
self.checker = ClientSecretChecker()
|
||||||
|
|
||||||
|
async def request(self, flow: http.HTTPFlow):
|
||||||
|
try:
|
||||||
|
self.checker.request(flow)
|
||||||
|
except Exception as e:
|
||||||
|
print(f"[ERROR] ClientSecretAddon request failed: {e}")
|
||||||
|
pass
|
||||||
|
|
||||||
|
async def response(self, flow: http.HTTPFlow):
|
||||||
|
try:
|
||||||
|
self.checker.response(flow)
|
||||||
|
except Exception as e:
|
||||||
|
print(f"[ERROR] ClientSecretAddon response failed: {e}")
|
||||||
|
pass
|
||||||
|
|
||||||
class AccessTokenAddon:
|
class AccessTokenAddon:
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
self.checker = AccessTokenScanner()
|
self.checker = AccessTokenScanner()
|
||||||
|
|
@ -87,4 +106,4 @@ class RedirectBypassAddon:
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
print(f"[ERROR] RedirectBypass Addon failed: {e}")
|
print(f"[ERROR] RedirectBypass Addon failed: {e}")
|
||||||
|
|
||||||
addons = [PKCEAddon(), ScopeAddon(), CsrfAddon(), NonceAddon(), AccessTokenAddon(), RedirectBypassAddon()]
|
addons = [PKCEAddon(), ScopeAddon(), CsrfAddon(), NonceAddon(), ClientSecretAddon(), AccessTokenAddon(), RedirectBypassAddon()]
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue