oauth-backend/addon/scope_detection.py

from lib.report_vuln import report_vuln
from lib.utils.is_oauth_uri import is_oauth_uri
from urllib.parse import urlparse, parse_qs

class ScopeDetection:
    def get_scope_from_query(self, query: str) -> str | None:
        if not query:
            return None
        parsed = 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

        parsed = urlparse(req.pretty_url)
        query = parsed.query

        location = res.headers.get("Location", "")
        location_query = urlparse(location).query

        query_scope = self.get_scope_from_query(query)
        location_scope = self.get_scope_from_query(location_query)

        if query_scope in ["all", "*"]:
            report_vuln(
                title="OAuth Scope Value Issue",
                desc=f"Scope value issue detected in request: {query_scope}",
                status="WARNING",
                uri=req.pretty_url
            )
        if location_scope in ["all", "*"]:
            report_vuln(
                title="OAuth Scope Value Issue",
                desc=f"Scope value issue detected in response location: {location_scope}",
                status="WARNING",
                uri=location
            )

    async def test(self, flow):
        
        if not is_oauth_uri(flow.request.pretty_url):
            return

        await self.check_scope(flow)