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

뭐 대충했어요

Browse source
This commit is contained in:
imnyang 2025-06-07 15:12:15 +09:00
commit 5f84d24973
5 changed files with 138 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

88
addon/pkce_check.py
View file

@ -3,6 +3,12 @@
from urllib.parse import urlparse, parse_qs, urlencode, urlunparse
import asyncio
import httpx
import csv
import os
from typing import List, Dict, Any
import lib.target as target
from lib.report import save_report
class PKCEDowngradeChecker:
required_keys = ["client_id", "response_type", "code_challenge", "code_challenge_method"]
@ -35,11 +41,27 @@ class PKCEDowngradeChecker:
challenge_val = query.get("code_challenge", [None])[0]
if not method_val or not challenge_val:
self.report("PKCE parameters missing or incomplete.", url, is_openid)
status = "MEDIUM" if is_openid else "LOW"
report_data = [{
'target': target.load(),
'status': status,
'title': "PKCE Parameters Missing",
'description': "PKCE parameters are missing or incomplete.",
'uri': url
}]
save_report(report_data)
return
if method_val.lower() == "plain":
self.report("PKCE method is set to 'plain'. Possible downgrade.", url, is_openid)
status = "CRITICAL"
report_data = [{
'target': target.load(),
'status': status,
'title': "PKCE Plain Method",
'description': "PKCE method is set to 'plain'. Possible downgrade.",
'uri': url
}]
save_report(report_data)
return
# Create downgraded request
@ -49,6 +71,13 @@ class PKCEDowngradeChecker:
new_query = urlencode(downgraded_query, doseq=True)
downgraded_url = urlunparse(parsed._replace(query=new_query))
# Ensure downgraded_url is a string, not bytes
if isinstance(downgraded_url, bytes):
downgraded_url = downgraded_url.decode('utf-8')
# Ensure it's definitely a string
downgraded_url = str(downgraded_url)
print(f"[DEBUG] Testing downgraded URL: {downgraded_url}")
@ -81,12 +110,15 @@ class PKCEDowngradeChecker:
if orig_success and down_success:
# If both redirect and both have code, it's a clear vulnerability
if both_redirect and both_have_code:
self.report(
"PKCE downgrade vulnerability detected! Both URLs returned authorization code.",
f"Original: {url}\nDowngraded: {downgraded_url}",
is_openid,
critical=True
)
status = "CRITICAL"
report_data = [{
'target': target.load(),
'status': status,
'title': "PKCE Downgrade Vulnerability",
'description': "PKCE downgrade vulnerability detected! Both URLs returned authorization code.",
'uri': f"Original: {url}\nDowngraded: {downgraded_url}"
}]
save_report(report_data)
# If responses are similar (both redirect to similar pages), it might be vulnerable
elif both_redirect:
# Check if the redirect locations are similar (excluding PKCE parameters)
@ -94,30 +126,28 @@ class PKCEDowngradeChecker:
down_parsed = urlparse(down_loc)
if orig_parsed.path == down_parsed.path and orig_parsed.netloc == down_parsed.netloc:
self.report(
"Potential PKCE downgrade vulnerability! Server accepts requests without PKCE.",
f"Original: {url}\nDowngraded: {downgraded_url}",
is_openid,
critical=False
)
# If downgraded request doesn't fail, it's concerning
status = "MEDIUM" if is_openid else "LOW"
report_data = [{
'target': target.load(),
'status': status,
'title': "Potential PKCE Downgrade",
'description': "Potential PKCE downgrade vulnerability! Server accepts requests without PKCE.",
'uri': f"Original: {url}\nDowngraded: {downgraded_url}"
}]
save_report(report_data)
elif down_resp.status_code != 400: # 400 would be expected for missing required parameters
self.report(
"Server accepts OAuth request without PKCE parameters.",
f"Original: {url}\nDowngraded: {downgraded_url}",
is_openid,
critical=False
)
status = "MEDIUM" if is_openid else "LOW"
report_data = [{
'target': target.load(),
'status': status,
'title': "PKCE Not Enforced",
'description': "Server accepts OAuth request without PKCE parameters.",
'uri': f"Original: {url}\nDowngraded: {downgraded_url}"
}]
save_report(report_data)
else:
print(f"[DEBUG] Requests had different success status - likely PKCE is enforced")
except Exception as e:
print(f"[ERROR] Request failed: {e}")
def report(self, msg, context_url, is_openid, critical=False):
tag = "[CRITICAL]" if critical else "[WARN]"
flow_type = "OpenID" if is_openid else "OAuth2"
# ANSI color codes for red text
RED = '\033[91m'
RESET = '\033[0m'
print(f"{RED}{tag} {flow_type} - {msg}\n → {context_url}{RESET}\n")