mirror of
https://github.com/j93es/oauth-backend.git
synced 2026-06-04 05:01:51 +09:00
53 lines
1.7 KiB
Python
53 lines
1.7 KiB
Python
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)
|