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

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}")

addons = [PKCEAddon(), ScopeAddon(), CsrfAddon()]