oauth-backend/addon/init.py

from mitmproxy import http
import asyncio
from pkce_check import PKCEDowngradeChecker
from ScopeDetection import ScopeDetection
from csrf_check import CsrfChecker
from nonce_check import NonceChecker
from redirect_uri_check import RedirectBypassChecker
from access_token import AccessTokenScanner

class PKCEAddon:
    def __init__(self):
        self.checker = PKCEDowngradeChecker()

    async def request(self, flow: http.HTTPFlow):
        print(
            f"[DEBUG] Processing request: {flow.request.method} {flow.request.pretty_url}"
        )
        try:
            await self.checker.test(flow)
        except Exception as e:
            print(f"[ERROR] Addon failed: {e}")
            pass
            
            
class CsrfAddon:
    def __init__(self):
        self.checker = CsrfChecker()

    async def response(self, flow: http.HTTPFlow):
        try:
            await self.checker.response(flow)
        except Exception as e:
            print(f"[ERROR] CSRF Addon failed: {e}")
            pass
        
        
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:
            await self.checker.test(flow)
        except Exception as e:
            print(f"[ERROR] ScopeDetection failed: {e}")

class NonceAddon:
    def __init__(self):
        self.checker = NonceChecker()

    async def response(self, flow: http.HTTPFlow):
        try:
            await self.checker.check_nonce_in_id_token(flow)
        except Exception as e:
            print(f"[ERROR] NonceAddon failed: {e}")
            pass

class AccessTokenAddon:
    def __init__(self):
        self.checker = AccessTokenScanner()

    async def response(self, flow: http.HTTPFlow):
        try:
            await self.checker.scan(flow)
        except Exception as e:
            print(f"[ERROR] AccessToken Addon failed: {e}")
            pass

class RedirectBypassAddon:
    def __init__(self):
        self.checker = RedirectBypassChecker()

    # request 대신 response 로 바꿔 보세요:
    async def response(self, flow: http.HTTPFlow):
        try:
            await self.checker.test(flow)
        except Exception as e:
            print(f"[ERROR] RedirectBypass Addon failed: {e}")

addons = [PKCEAddon(), ScopeAddon(), CsrfAddon(), NonceAddon(), AccessTokenAddon(), RedirectBypassAddon()]