[Refactor] 리팩터링

.gitignore

@@ -9,6 +9,8 @@ wheels/
 # Virtual environments
 .venv
 
+.env
+
 data/
 
 

addon/access_token.py

@@ -4,8 +4,8 @@ from typing import List, Dict, Optional, Any
 
 from mitmproxy.http import HTTPFlow
 
-import lib.target as target
+import lib.cur_target_url as cur_target_url
-from lib.report import save_report
+from lib.report_vuln import save_report
 
 # 결과 리포트 저장용 데이터 클래스
 @dataclass
@@ -32,7 +32,7 @@ class AccessTokenScanner:
         findings.extend(await self._scan_response(flow.response, flow.request.url))
 
         if findings:
-            target_value = target.load()
+            target_value = cur_target_url.load()
             save_report([f.to_report(target_value) for f in findings])
 
     # 내부 구현

addon/csrf_check.py

@@ -4,8 +4,8 @@ from urllib.parse import urlparse, parse_qs, unquote
 import httpx
 from typing import Optional, Union, List
 
-import lib.target as target
+import lib.cur_target_url as cur_target_url
-from lib.report import save_report
+from lib.report_vuln import save_report
 
 class CsrfChecker:
     nonce_params = {
@@ -153,7 +153,7 @@ class CsrfChecker:
                 desc = " | ".join(msgs)
                 status = "MEDIUM"
                 report_data = [{
-                    'target': target.load(),
+                    'target': cur_target_url.load(),
                     'status': status,
                     'title': "CSRF Risk",
                     'description': desc,

addon/google_login_hint.py

@@ -0,0 +1,67 @@
+import os
+from urllib.parse import urlparse, parse_qs, urlencode, urlunparse
+from dotenv import load_dotenv
+
+# .env 파일 로드
+load_dotenv(override=True)
+
+class GoogleLoginHint:
+    def __init__(self):
+        self.google_id = os.getenv('GOOGLE_ID', '')
+        if not self.google_id:
+            print("⚠️  Warning: GOOGLE_ID not found in .env file")
+    
+    async def request(self, flow):
+        """Google OAuth 요청을 가로채서 login_hint를 추가하거나 수정"""
+        req = flow.request
+        url = req.pretty_url
+        
+        # Google OAuth 인증 URL인지 확인
+        if self._is_google_oauth_url(url):
+            print(f"🔍 Google OAuth URL detected: {url}")
+            
+            # URL 파싱
+            parsed_url = urlparse(url)
+            query_params = parse_qs(parsed_url.query)
+            
+            # login_hint 추가 또는 수정
+            if self.google_id:
+                query_params['login_hint'] = [self.google_id]
+                print(f"✅ Added/Updated login_hint: {self.google_id}")
+                
+                # 새로운 쿼리 스트링 생성
+                new_query = urlencode(query_params, doseq=True)
+                
+                # 새로운 URL 생성
+                new_url = urlunparse((
+                    parsed_url.scheme,
+                    parsed_url.netloc,
+                    parsed_url.path,
+                    parsed_url.params,
+                    new_query,
+                    parsed_url.fragment
+                ))
+                
+                # 요청 URL 수정 - URL과 호스트 모두 업데이트
+                flow.request.url = new_url
+                flow.request.pretty_url = new_url
+                print(f"🔄 Modified URL: {new_url}")
+    def _is_google_oauth_url(self, url):
+        """Google OAuth URL인지 확인"""
+        google_oauth_domains = [
+            'accounts.google.com',
+            'oauth2.googleapis.com'
+        ]
+        
+        parsed_url = urlparse(url)
+        domain = parsed_url.netloc.lower()
+        
+        # Google OAuth 도메인 확인
+        for google_domain in google_oauth_domains:
+            if google_domain in domain:
+                # OAuth 관련 경로 확인
+                path = parsed_url.path.lower()
+                if any(oauth_path in path for oauth_path in ['/oauth2', '/auth', '/login']):
+                    return True
+        
+        return False

addon/init.py

@@ -1,12 +1,12 @@
 from mitmproxy import http
 import asyncio
 from pkce_check import PKCEDowngradeChecker
-from ScopeDetection import ScopeDetection
+from addon.scope_detection import ScopeDetection
 from csrf_check import CsrfChecker
 from nonce_check import NonceChecker
 from redirect_uri_check import RedirectBypassChecker
 from access_token import AccessTokenScanner
-from GoogleLoginHint import GoogleLoginHint
+from addon.google_login_hint import GoogleLoginHint
 import os
 from dotenv import load_dotenv
 

addon/nonce_check.py

@@ -3,8 +3,8 @@ from urllib.parse import urlparse, parse_qs
 from typing import Union
 import httpx
 
-import lib.target as target
+import lib.cur_target_url as cur_target_url
-from lib.report import save_report
+from lib.report_vuln import save_report
 
 class NonceChecker:
     def is_oidc_flow(self, flow) -> bool:
@@ -76,7 +76,7 @@ class NonceChecker:
         url = req.pretty_url
         if not nonce:
             report_data = [{
-                'target': target.load(),
+                'target': cur_target_url.load(),
                 'status': "CRITICAL",
                 'title': "nonce is missing in id_token",
                 'description': "Nonce is present in the request but missing in the id_token.",

addon/pkce_check.py

@@ -3,8 +3,8 @@ import asyncio
 import httpx
 from typing import Dict, List
 
-import lib.target as target
+import lib.cur_target_url as cur_target_url
-from lib.report import save_report
+from lib.report_vuln import save_report
 
 
 class PKCEDowngradeChecker:
@@ -170,7 +170,7 @@ class PKCEDowngradeChecker:
         self, status: str, title: str, description: str, uri: str
     ) -> Dict[str, str]:
         return {
-            "target": target.load(),
+            "target": cur_target_url.load(),
             "status": status,
             "title": title,
             "description": description,

addon/redirect_uri_check.py

@@ -4,8 +4,8 @@ import asyncio
 import random
 import time
 from urllib.parse import urlparse, parse_qs, urlencode, urlunparse  
-import lib.target as target  
+import lib.cur_target_url as cur_target_url  
-from lib.report import save_report 
+from lib.report_vuln import save_report 
 
 class RedirectRateLimiter:
     """redirect_uri_check 전용 rate limiter"""
@@ -1385,7 +1385,7 @@ class RedirectBypassChecker:
         )
         
         report_data = [{
-            "target": target.load(),
+            "target": cur_target_url.load(),
             "status": "CRITICAL",  
             "title": "Redirect URI Bypass Vulnerability",
             "description": description,

addon/scope_detection.py

@@ -0,0 +1,53 @@
+import lib.cur_target_url as cur_target_url
+from lib.report_vuln 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': cur_target_url.load(),
+                'status': "WARNING",
+                'title': "OAuth scope value issue",
+                'description': f"{method} {url}: {', '.join(result)}",
+                'uri': url
+            }]
+            save_report(report_data)

lib/cur_target_url.py

@@ -0,0 +1,32 @@
+# save ./data/target.temp file as domain string
+import os
+
+def save(target: str, file_path: str = "./data/target.dump") -> None:
+    """
+    Save the target domain to a temporary file.
+
+    :param target: Target domain to be saved.
+    """
+    # Create directory if it doesn't exist
+    os.makedirs(os.path.dirname(file_path), exist_ok=True)
+    
+    with open(file_path, 'w', encoding='utf-8') as f:
+        f.write(target)
+    
+    print(f"Target saved to {file_path}")  # Debug message
+
+def load(file_path: str = "./data/target.dump") -> str:
+    """
+    Load the target domain from a temporary file.
+
+    :return: Target domain string.
+    """
+    if not os.path.exists(file_path):
+        print(f"[ERROR] Target file {file_path} does not exist.")
+        return ""
+
+    with open(file_path, 'r', encoding='utf-8') as f:
+        target = f.read().strip()
+    
+    print(f"Loaded target from {file_path}: {target}")  # Debug message
+    return target

lib/report_vuln.py

@@ -0,0 +1,32 @@
+# save as data/report.csv
+import os
+import csv
+from typing import List, Dict, Any
+
+# target, status, title, description, uri
+
+# file path는 'data/report.csv'로 고정
+def save_report(report_data: List[Dict[str, Any]], file_path: str = 'data/report.csv') -> None:
+    os.makedirs(os.path.dirname(file_path), exist_ok=True)
+    
+
+    """
+    report_data 안의 각 레포트를 한 줄씩 CSV에 추가로 저장합니다.
+    파일이 없으면 헤더를 먼저 쓰고, 있으면 레코드만 이어서 씁니다.
+    """
+    fieldnames = ['target', 'status', 'title', 'description', 'uri']
+    file_exists = os.path.exists(file_path)
+
+    with open(file_path, mode='a', newline='', encoding='utf-8') as csvfile:
+        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
+        # 파일이 없던 새로 만들 때만 헤더 작성
+        if not file_exists:
+            writer.writeheader()
+
+        for row in report_data:
+            # None 방지 & 줄바꿈 이스케이프
+            escaped = {
+                k: str(v).replace('\n', '\\n') if v is not None else ''
+                for k, v in row.items()
+            }
+            writer.writerow(escaped)

runner/backend/__init__.py

@@ -1,6 +1,6 @@
 from fastapi import FastAPI, Query, HTTPException
 from fastapi.responses import Response
-import lib.target as target
+import lib.cur_target_url as cur_target_url
 
 app = FastAPI()
 
@@ -8,7 +8,7 @@ app = FastAPI()
 @app.post("/start")
 async def start(url: str = Query(None)):
     if url:
-        target.save(url)
+        cur_target_url.save(url)
         print(f"Target URL set to: {url}")
         return {"message": f"Target URL set to: {url}"}
     return {"error": "No URL provided"}