whs-authz-authn/browser-use-oauth
1
0
Fork 0
mirror of https://github.com/j93es/browser-use-oauth.git synced 2026-06-04 05:11:53 +09:00
Code Issues Projects Releases Packages Wiki Activity

feat/refect 가져오기

Browse source
This commit is contained in:
tk 2025-06-27 23:06:35 +09:00
commit 54682cdb72
43 changed files with 702 additions and 520 deletions
Download patch file Download diff file Expand all files Collapse all files

36
.env.example Normal file
View file

@ -0,0 +1,36 @@
ANONYMIZED_TELEMETRY=false
# ========== LLM ==========
GOOGLE_API_KEY=
# 권장 (다른 모델로 교체 가능) [다른 모델로 교체시 성능 보장 불가]
GOOGLE_MODEL=gemini-2.5-flash
#GOOGLE_PLANNER_MODEL=gemini-2.5-flash # 왜 비활성화 되었나요? // Planner 모델이 오히려 문제를 일으키는 경우가 있어 비활성화했습니다. 필요시 활성화하세요.
# min(INITIAL_BACKOFF * (2 ** try_cnt), MAX_BACKOFF)만큼 API가 실패시 대기합니다.
INITIAL_BACKOFF=60
MAX_BACKOFF=600
#ENABLE_PLANNER_MODEL_OAUTH_LOGIN=true # OAuth 로그인 시 Planner 모델을 활성화합니다.
#ENABLE_PLANNER_MODEL_OAUTH_LIST=true # OAuth List를 찾을 때 Planner 모델을 활성화합니다.
# ========== Monitoring ==========
# 선택
PROXY_HOST=127.0.0.1
PROXY_PORT=11080
BACKEND_URL=http://localhost:11081
# https://docs.browser-use.com/development/observability
# Lmnr 계정이 필요합니다.
# https://lmnr.ai/
LMNR_PROJECT_API_KEY=
# 브라우저 언어 설정
LANG=en_US
# ========= Account ==========
# 필수 뒤에 있는 이메일 주소는 Google 계정의 로그인 힌트로 사용됩니다.
# 이메일의 전체를 입력해주세요
GOOGLE_ID=whs.imnya.ng@gmail.com

3
.gitignore vendored
View file

@ -83,5 +83,6 @@ my.sh
log.txt log.txt
data/ data/
!src/lib/utils/data
# End of https://www.toptal.com/developers/gitignore/api/macos,windows # End of https://www.toptal.com/developers/gitignore/api/macos,windows

29
README.md
View file

@ -20,6 +20,10 @@
> 다른 플렛폼은 수동으로 설정되어야만 합니다. > 다른 플렛폼은 수동으로 설정되어야만 합니다.
> https://docs.mitmproxy.org/stable/concepts/certificates/ > https://docs.mitmproxy.org/stable/concepts/certificates/
현재 아래와 같은 환경에서 개발되며 테스트되고 있습니다.
- ✅ MacOS 26 Tahoe Developer Beta 2 (25A5295e) en-US aarch64
- ✅ Windows 11 Pro for Workstations 24H2 (26100.4351) en-US x86_64
- ✅ NixOS 25.05.804570.c7ab75210cb8 KDE 6 / Linux 6.15 x86_64
--- ---
다음과 같은 명령어로 환경을 설정합니다. 다음과 같은 명령어로 환경을 설정합니다.
@ -112,23 +116,30 @@ uv run run.py 1 100 --skh
## 1. 파일 생성 ## 1. 파일 생성
`lib/llm/prompt` 폴더로 `lib/llm/prompt` 폴더에서 fallback 폴더를 복사하여
![](./docs/list.png) 원하는 프로바이더를 추가해줍니다. `ex) lib/llm/prompt/Google/`
fallback.py를 복사하여 ## 2. prompt.py 수정
원하는 프로바이더를 추가해줍니다. `ex) lib/llm/prompt/Google.py` Prompt에서 추가한 파일을 prompt.py에서 수정합니다.
## 2. __init__.py 수정 만약 로그인 정보를 넣고 싶다면 Sensitive
`Log into example.com as user x_username with password x_password`
![](./docs/guide.png) ## 3. model.py
Prompt에서 추가한 파일을 __init__.py에서 import합니다. 응답할 때 원하는 리턴 값을 `dict`로 받습니다.
## 3. 파일 수정 ## 4. \_\_init\_\_.py 수정
![alt text](./docs/guide_0.png)
생성한 파일에서 프롬프트를 수정합니다. 추가한 prompt에 따라 import합니다.
## 5. 사용 방법
```py
from lib.llm.prompt.fallback import prompt, model
```
# 참고하면 좋을만한 것 # 참고하면 좋을만한 것

BIN
docs/guide.png
View file

Binary file not shown.
Side by side

Before

Width:  |  Height:  |  Size: 2.6 MiB

BIN
docs/guide_0.png Normal file
View file

Binary file not shown.
Side by side

After

Width:  |  Height:  |  Size: 356 KiB

53
lib/llm/prompt/Microsoft.py
View file

@ -1,53 +0,0 @@
# Extended planner prompt
extend_planner_system_message = f"""
# 목적 : Microsoft OAuth 로그인
## ✅ Step 1 : Step 3에서
만약에 Microsoft에 로그인 되어 있는 계정(oauth test)가 없을 경우
- Microosoft.sensitive.json에서 sign in with your username(email) x_username and password is x_password
- 반드시 이 과정에서 username을 먼저 입력하고 다음 과정에서 암호 사용을 누른 후 password를 입력해야 한다.
쿠키 삭제 방법:
chrome://settings/clearBrowserData에 들어가서 삭제해주세요.
🛑 절대 아래와 같이 해석하지 말 것:
- ❌ 버튼 클릭 후 페이지 로딩만 기다리고 돌아가기
- ❌ URL 저장 없이 go_back() 호출
---
### ✨ 추가 안전 장치: "뒤로가기(go_back) 호출 조건" 제한
```text
🛑 뒤로가기(go_back)은 다음 조건이 모두 충족될 때만 사용 => 다만 로그인 실패 시, 뒤로가기 수행:
- ✅ 로그인 흐름이 완료됨 (예: redirect back to app, or callback URL)
- ✅ 현재 리디렉션 URL이 수집됨
- ✅ 결과에 저장 후 다음 버튼 탐색을 위해 복귀 필요할 때
```
---
## 🚫 Step 2: 버튼 없음 또는 예외 발생 시
* 유효한 SSO 버튼이 **전혀 없을 경우**
* 예외, 오류 등 발생 시
-> 즉시 중단
---
## 📎 중요 규칙 요약
* 🔁 단계는 반드시 순서대로 진행
* 🔐 로그인은 쿠키/세션으로 유지된 상태에서 수행
* 👀 직접 OAuth Providor ID/PW를 입력하여도 됨 가지고 있다면
* ⛔ 추측한 URL은 접속하지 않음
---
"""

10
lib/utils/__init__.py
View file

@ -1,10 +0,0 @@
# export from show_info
from lib.utils.agent_info import *
from lib.utils.backend_client import *
from lib.utils.config import *
from lib.utils.is_html import *
from lib.utils.logger import *
from lib.utils.read_txt import *
from lib.utils.browser_use import *

5
lib/utils/browser_use/__init__.py
View file

@ -1,5 +0,0 @@
from lib.utils.browser_use.clean_resources import *
from lib.utils.browser_use.func import *
from lib.utils.browser_use.model import *
from lib.utils.browser_use.init_profile import *
from lib.utils.browser_use.sensitive_data import *

425
main.py
View file

@ -1,425 +0,0 @@
import asyncio
import json
import os
import csv
import argparse
from pathlib import Path
import signal
from dotenv import load_dotenv
from browser_use import (
Agent,
BrowserSession,
Controller,
)
from patchright.async_api import async_playwright as async_patchright, Page
from pydantic import BaseModel
from lib.utils import (
notify_backend,
read_lines_between,
is_html_url,
env_cheker,
logger,
config,
GetProfile
)
from lib.utils import (
GetSensitiveData,
setup_storage_state,
clean_resources
)
from lib.llm import (
CreateChatGoogleGenerativeAI,
get_prompt
)
import lib.utils.browser_use.model as model
load_dotenv(verbose=True, override=True)
# Exponential backoff settings
INITIAL_BACKOFF = int(os.getenv("INITIAL_BACKOFF", "60")) # seconds
MAX_BACKOFF = int(os.getenv("MAX_BACKOFF", "600")) # seconds
# 진행 상황 추적을 위한 전역 변수
current_progress = {"current_index": 0, "total": 0, "current_url": "", "start_line": 0}
progress_file = Path("data/scan_progress.json")
env_cheker()
if os.getenv("LMNR_PROJECT_API_KEY"):
from lmnr import Laminar
Laminar.initialize(project_api_key=os.getenv("LMNR_PROJECT_API_KEY"))
def save_progress():
"""현재 진행 상황을 파일에 저장"""
with open(progress_file, "w", encoding="utf-8") as f:
json.dump(current_progress, f, ensure_ascii=False, indent=2)
def load_progress():
"""이전 진행 상황을 파일에서 불러오기"""
if os.path.exists(progress_file):
try:
with open(progress_file, "r", encoding="utf-8") as f:
return json.load(f)
except:
return None
return None
def signal_handler(signum, frame):
"""Ctrl+C 시그널 핸들러"""
print("\n" + "=" * 60)
print("🛑 스캔이 중단되었습니다!")
print(f"📊 진행 상황:")
print(f" - 전체: {current_progress['total']}개 URL")
print(f" - 완료: {current_progress['current_index']}개 URL")
print(f" - 현재 처리 중: {current_progress['current_url']}")
print(
f" - domains.txt의 {current_progress['start_line'] + current_progress['current_index']}번째 줄"
)
print(
f" - 진행률: {current_progress['current_index']}/{current_progress['total']} ({current_progress['current_index']/current_progress['total']*100:.1f}%)"
)
print("=" * 60)
save_progress()
print(f"💾 진행 상황이 {progress_file}에 저장되었습니다.")
exit(0)
# 시그널 핸들러 등록
signal.signal(signal.SIGINT, signal_handler)
# ── OAuth 리스트 추출 Agent ──
async def extract_oauth_list(url: str, skip_html_check: bool = False):
"""첫 번째 Agent: 로그인 페이지를 찾고 OAuth 리스트만 추출"""
await setup_storage_state()
target_url = url if url.startswith("http") else f"https://{url}"
print(f"<EFBFBD> OAuth 리스트 추출 시작: {target_url}")
# 1) URL이 HTML 페이지인지 확인
if not is_html_url(target_url) and not skip_html_check:
print(f"❌ {target_url} 은(는) HTML이 아닙니다. 스킵합니다.")
return []
agent = None
session = None
try_cnt = 0
while True:
session = BrowserSession(
playwright=(await async_patchright().start()),
browser_profile=await GetProfile(),
)
initial_actions = [{"open_tab": {"url": target_url}}]
controller = Controller(
output_model=model.OAuthList,
exclude_actions=["search_google", "unknown_action", "unkown"],
)
print("🤖 OAuth 리스트 추출 Agent 초기화...")
try:
agent = Agent(
browser_session=session,
initial_actions=initial_actions,
sensitive_data=GetSensitiveData(),
task=(
"Navigate to the login page and identify all OAuth provider buttons (excluding Passkey). "
"DO NOT click any OAuth buttons or attempt to login. "
"Just find and list all available OAuth providers with their button texts or provider names. "
"Return a list of OAuth providers found on the login page."
),
llm=CreateChatGoogleGenerativeAI(config.GOOGLE_MODEL),
planner_llm=(
CreateChatGoogleGenerativeAI(config.GOOGLE_PLANNER_MODEL)
if config.GOOGLE_PLANNER_MODEL
else None
),
controller=controller,
extend_planner_system_message=get_prompt("auth"),
)
response = await agent.run()
final_result = response.final_result()
if final_result is None:
raise ValueError("OAuth 리스트 추출 결과가 None입니다.")
data = json.loads(final_result)
oauth_providers = data["oauth_providers"] # 이제 문자열 배열
oauth_entries = [model.OAuth(provider=provider) for provider in oauth_providers]
await clean_resources(agent, session)
return oauth_entries
except Exception as e:
await clean_resources(agent, session)
# API 쿼터 문제인지 확인
if "ResourceExhausted" in str(e) or "429" in str(e):
wait = min(INITIAL_BACKOFF * (2**try_cnt), MAX_BACKOFF)
print(f"⚠️ API 쿼터 에러: {e}. {wait}초 대기 후 재시도합니다...")
await asyncio.sleep(wait)
try_cnt += 1
if try_cnt >= 3:
print(
f"❌ {url} OAuth 리스트 추출 실패: API 쿼터 문제가 지속됩니다."
)
logger(f"❌ {url} OAuth 리스트 추출 실패: API 쿼터 문제: {e}")
return []
continue
# 일반 에러 처리
try_cnt += 1
if try_cnt >= 3:
print(f"❌ {url} OAuth 리스트 추출 실패: 에러: {e}")
logger(f"❌ {url} OAuth 리스트 추출 실패: 에러: {e}")
return []
print(f"⚠️ 에러 발생: {e}. {try_cnt}번째 재시도 중...")
await asyncio.sleep(30)
continue
# ── 개별 OAuth 로그인 Agent ──
async def test_oauth_login(url: str, oauth_provider: str):
"""두 번째 Agent: 특정 OAuth 제공자로 로그인 시도"""
await setup_storage_state()
target_url = url if url.startswith("http") else f"https://{url}"
print(f"🔐 {oauth_provider} 로그인 시작: {target_url}")
agent = None
session = None
try_cnt = 0
while True:
session = BrowserSession(
playwright=(await async_patchright().start()),
browser_profile=await GetProfile(),
)
initial_actions = [{"open_tab": {"url": target_url}}]
controller = Controller(
exclude_actions=["search_google", "unknown_action", "unkown"],
)
print(f"🤖 {oauth_provider} 로그인 Agent 초기화...")
try:
agent = Agent(
browser_session=session,
initial_actions=initial_actions,
sensitive_data=GetSensitiveData(),
task=(
f"Navigate to the login page, find and click the {oauth_provider} OAuth button, "
f"then follow the complete OAuth login flow as far as possible with a real user account. "
f"Capture the final redirect URL after login completion. "
f"If login fails or encounters errors, report the issue. "
f"Focus only on {oauth_provider} - ignore other OAuth providers."
),
llm=CreateChatGoogleGenerativeAI(config.GOOGLE_MODEL),
planner_llm=(
CreateChatGoogleGenerativeAI(config.GOOGLE_PLANNER_MODEL)
if config.GOOGLE_PLANNER_MODEL
else None
),
controller=controller,
extend_planner_system_message=get_prompt(oauth_provider),
)
response = await agent.run()
final_result = response.final_result()
print(f"✅ {oauth_provider} 로그인 완료")
if final_result:
logger(f"✅ {url} - {oauth_provider} 로그인 결과: {final_result}")
await clean_resources(agent, session)
return True
except Exception as e:
await clean_resources(agent, session)
# API 쿼터 문제인지 확인
if "ResourceExhausted" in str(e) or "429" in str(e):
wait = min(INITIAL_BACKOFF * (2**try_cnt), MAX_BACKOFF)
print(f"⚠️ API 쿼터 에러: {e}. {wait}초 대기 후 재시도합니다...")
await asyncio.sleep(wait)
try_cnt += 1
if try_cnt >= 3:
print(
f"❌ {oauth_provider} 로그인 실패: API 쿼터 문제가 지속됩니다."
)
logger(
f"❌ {url} - {oauth_provider} 로그인 실패: API 쿼터 문제: {e}"
)
return False
continue
# 일반 에러 처리
try_cnt += 1
if try_cnt >= 3:
print(f"❌ {oauth_provider} 로그인 실패: 에러: {e}")
logger(f"❌ {url} - {oauth_provider} 로그인 실패: 에러: {e}")
return False
print(f"⚠️ 에러 발생: {e}. {try_cnt}번째 재시도 중...")
await asyncio.sleep(30)
continue
# ── 통합 스캔 함수 ──
async def scan_one_url(url: str, skip_html_check: bool = False):
"""URL 스캔 통합 함수: OAuth 리스트 추출 → 개별 OAuth 로그인 시도"""
target_url = url if url.startswith("http") else f"https://{url}"
print(f"🚀 스캔 시작: {target_url}")
# Backend에 스캔 시작을 알림
notify_backend(target_url)
# 1단계: OAuth 리스트 추출
oauth_entries = await extract_oauth_list(url, skip_html_check)
if not oauth_entries:
print(f"❌ {target_url}에서 OAuth 제공자를 찾을 수 없습니다.")
return
print("-" * 50)
print(f"🔗 스캔 URL: {url}")
print(f"🔐 발견된 OAuth 제공자들: {len(oauth_entries)}개")
for entry in oauth_entries:
print(f" - {entry.provider}")
print("-" * 50)
# CSV에 OAuth 리스트 저장
csv_file = "./data/oauth_providers.csv"
file_exists = os.path.isfile(csv_file)
with open(csv_file, "a", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
if not file_exists:
writer.writerow(["issuer", "provider", "oauth_uri", "login_tested"])
for entry in oauth_entries:
writer.writerow([url, entry.provider, "", "pending"]) # oauth_uri는 빈 문자열
# 2단계: 각 OAuth 제공자별로 개별 로그인 시도
for i, oauth_entry in enumerate(oauth_entries):
print(
f"\n🔄 OAuth 로그인 테스트 {i+1}/{len(oauth_entries)}: {oauth_entry.provider}"
)
# OAuth 간 대기 시간
if i > 0:
print("⏳ OAuth 테스트 간 대기 중 (30초)...")
await asyncio.sleep(30)
# 개별 OAuth 로그인 시도
success = await test_oauth_login(url, oauth_entry.provider)
# 결과를 CSV에 업데이트 (간단하게 로그만 남김)
status = "success" if success else "failed"
print(f"📝 {oauth_entry.provider} 로그인 결과: {status}")
async def loop(
filepath: str, start_line: int, end_line: int, skip_html_check: bool = False
):
# 인자값으로 받은 파일 경로와 줄 범위를 통해 도메인 리스트 생성
target_list = read_lines_between(
filepath=filepath, start_line=start_line, end_line=end_line
)
# 진행 상황 초기화
current_progress["total"] = len(target_list)
current_progress["start_line"] = start_line
current_progress["current_index"] = 0
# 이전 진행 상황 확인
prev_progress = load_progress()
if prev_progress and prev_progress.get("start_line") == start_line:
print(f"📋 이전 진행 상황을 발견했습니다:")
print(
f" - 이전 완료: {prev_progress['current_index']}/{prev_progress['total']}"
)
print(f" - 마지막 처리: {prev_progress.get('current_url', 'N/A')}")
resume = input("이어서 진행하시겠습니까? (y/n): ").lower().strip()
if resume == "y":
current_progress["current_index"] = prev_progress["current_index"]
target_list = target_list[current_progress["current_index"] :]
print(f"✅ {current_progress['current_index']}번째부터 재개합니다.")
# (필요하다면) 강제 설정이 필요한 경우, 아래 주석을 해제하여 target_list[0] 등을 덮어쓸 수 있습니다.
# target_list[0] = "velog.io"
for i, url in enumerate(target_list):
actual_index = current_progress["current_index"] + i
current_progress["current_url"] = url
current_progress["current_index"] = actual_index
print(f"\n🔄 Processing {actual_index + 1}/{current_progress['total']}: {url}")
print(f"📍 domains.txt의 {start_line + actual_index}번째 줄")
# URL들 사이에 API 쿼터 회복을 위한 대기 시간 추가
if actual_index > 0:
print("⏳ API 쿼터 보호를 위해 30초 대기 중...")
await asyncio.sleep(30)
await scan_one_url(url, skip_html_check=skip_html_check)
# 진행 상황 저장
current_progress["current_index"] = actual_index + 1
save_progress()
print(f"\n🎉 모든 스캔이 완료되었습니다! ({current_progress['total']}개 URL)")
# 완료 후 진행 상황 파일 삭제
if os.path.exists(progress_file):
os.remove(progress_file)
def main():
parser = argparse.ArgumentParser(
prog="domain_scanner",
description="도메인 목록 파일에서 지정한 줄 범위를 읽어 SSO 스캔을 수행합니다.",
)
# 커맨드라인 인자로 받을 옵션들 정의
parser.add_argument(
"-f",
"--file",
type=str,
required=True,
help="도메인 목록이 들어 있는 텍스트 파일 경로 (예: ./domains.txt)",
)
parser.add_argument(
"-s", "--start", type=int, required=True, help="읽기 시작 줄 번호 (1-based)"
)
parser.add_argument(
"-e", "--end", type=int, required=True, help="읽기 종료 줄 번호 (1-based)"
)
parser.add_argument(
"-skh",
"--skip-html-check",
type=bool,
default=False,
help="HTML 페이지 체크를 건너뛰고 모든 URL을 스캔합니다. (기본값: False)",
)
args = parser.parse_args()
# 인자값을 비동기 함수에 전달
asyncio.run(
loop(
filepath=args.file,
start_line=args.start,
end_line=args.end,
skip_html_check=args.skip_html_check,
)
)
if __name__ == "__main__":
main()

17
run.py
View file

@ -8,7 +8,7 @@ import argparse
#!/usr/bin/env python3 #!/usr/bin/env python3
# ── 설정 부분 ── # ── 설정 부분 ──
PYTHON_SCRIPT = "main.py" PYTHON_SCRIPT = "./src/main.py"
DOMAIN_FILE = "./data/domains.txt" DOMAIN_FILE = "./data/domains.txt"
# ───────────── # ─────────────
@ -35,13 +35,16 @@ def run_script(start_line, end_line, skh_option):
print(f"[{current_time}] Processing lines {start_line} to {end_line}...") print(f"[{current_time}] Processing lines {start_line} to {end_line}...")
try: try:
subprocess.run([ command = [
"uv", "run", PYTHON_SCRIPT, "uv", "run", PYTHON_SCRIPT,
"-f", DOMAIN_FILE, "-f", DOMAIN_FILE,
"-s", str(start_line), "-s", str(start_line),
"-e", str(end_line), "-e", str(end_line),
"-skh", str(skh_option) ]
], check=True) if skh_option:
command.append("--skip-html-check")
subprocess.run(command, check=True)
except subprocess.CalledProcessError: except subprocess.CalledProcessError:
print("Python 스크립트 실행 실패") print("Python 스크립트 실행 실패")
sys.exit(1) sys.exit(1)
@ -52,9 +55,9 @@ def main():
formatter_class=argparse.RawDescriptionHelpFormatter, formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=""" epilog="""
사용 예시: 사용 예시:
python run.py 10000 11000 # 10000~11000 라인 처리 uv run run.py 10000 11000 # 10000~11000 라인 처리
python run.py 10000 11000 --skh # SKH 옵션 활성화 uv run run.py 10000 11000 --skh # SKH 옵션 활성화
python run.py 10000 11000 --no-download # 다운로드 생략 uv run run.py 10000 11000 --no-download # 다운로드 생략
""" """
) )

7
src/lib/browser_use/__init__.py Normal file
View file

@ -0,0 +1,7 @@
from lib.browser_use.clean_resources import *
from lib.browser_use.func import *
from lib.browser_use.model import *
from lib.browser_use.init_profile import *
from lib.browser_use.sensitive_data import *
from lib.browser_use.agents import *
from lib.browser_use.scanner import *

175
src/lib/browser_use/agents.py Normal file
View file

@ -0,0 +1,175 @@
import asyncio
import os
import json
from browser_use import Agent, BrowserSession, Controller
from patchright.async_api import async_playwright as async_patchright
from lib.browser_use import (
GetProfile,
GetSensitiveData,
clean_resources,
)
from lib.utils import (
logger,
config,
)
from lib.llm import CreateChatGoogleGenerativeAI, get_prompt
# Exponential backoff settings
INITIAL_BACKOFF = int(os.getenv("INITIAL_BACKOFF", "60")) # seconds
MAX_BACKOFF = int(os.getenv("MAX_BACKOFF", "600")) # seconds
async def _run_agent_with_retry(agent_config):
"""Agent 실행을 위한 내부 헬퍼 함수 (재시도 로직 포함)"""
agent = None
session = None
try_cnt = 0
url = agent_config["url"]
while try_cnt < 3:
try:
session = BrowserSession(
playwright=(await async_patchright().start()),
browser_profile=await GetProfile(),
)
agent = Agent(
browser_session=session,
**agent_config["agent_params"]
)
response = await agent.run()
await clean_resources(agent, session)
return response
except Exception as e:
await clean_resources(agent, session)
if "ResourceExhausted" in str(e) or "429" in str(e):
wait = min(INITIAL_BACKOFF * (2**try_cnt), MAX_BACKOFF)
print(f"⚠️ API 쿼터 에러: {e}. {wait}초 대기 후 재시도합니다...")
await asyncio.sleep(wait)
try_cnt += 1
if try_cnt >= 3:
error_msg = f"API 쿼터 문제가 지속됩니다."
logger(f"❌ {url} - {agent_config['log_context']} 실패: {error_msg}: {e}")
print(f"❌ {url} - {agent_config['log_context']} 실패: {error_msg}")
return None
continue
# 일반 에러 처리
try_cnt += 1
if try_cnt >= 3:
error_msg = f"최대 재시도 횟수 초과."
logger(f"❌ {url} - {agent_config['log_context']} 실패: {error_msg}: {e}")
print(f"❌ {url} - {agent_config['log_context']} 실패: {error_msg}")
return None
print(f"⚠️ 에러 발생: {e}. {try_cnt}번째 재시도 중...")
await asyncio.sleep(30)
continue
return None
async def extract_oauth_list(url: str):
"""첫 번째 Agent: 로그인 페이지를 찾고 OAuth 리스트만 추출"""
target_url = url if url.startswith("http") else f"https://{url}"
print(f"🔎 OAuth 리스트 추출 시작: {target_url}")
prompt, model = get_prompt("auth")
agent_config = {
"url": target_url,
"log_context": "OAuth 리스트 추출",
"agent_params": {
"initial_actions": [{"open_tab": {"url": target_url}}],
"sensitive_data": GetSensitiveData(),
"task": (
"Navigate to the login page and identify all OAuth provider buttons (excluding Passkey). "
"DO NOT click any OAuth buttons or attempt to login. "
"Just find and list all available OAuth providers with their button texts or provider names. "
"Return a list of OAuth providers found on the login page."
),
"llm": CreateChatGoogleGenerativeAI(config.GOOGLE_MODEL),
"planner_llm": (
CreateChatGoogleGenerativeAI(config.GOOGLE_PLANNER_MODEL)
if config.GOOGLE_PLANNER_MODEL
else None
),
"controller": Controller(
output_model=model if not isinstance(model, str) else None,
exclude_actions=["search_google", "unknown_action", "unkown"],
),
"extend_planner_system_message": prompt,
}
}
response = await _run_agent_with_retry(agent_config)
if not response:
return []
final_result = response.final_result()
if not final_result:
print("OAuth 리스트 추출 결과가 없습니다.")
return []
try:
data = json.loads(final_result)
oauth_providers = data.get("oauth_providers", [])
if not oauth_providers:
print("❌ OAuth 제공자가 없습니다.")
logger(f"❌ {url} - OAuth 제공자 없음: {final_result}")
return []
print(f"✅ OAuth 제공자 추출 완료: {oauth_providers}")
return oauth_providers
except (json.JSONDecodeError, KeyError) as e:
print(f"❌ 결과 파싱 실패: {e}")
logger(f"❌ {url} 결과 파싱 실패: {final_result}")
return []
async def test_oauth_login(url: str, oauth_provider: str):
"""두 번째 Agent: 특정 OAuth 제공자로 로그인 시도"""
target_url = url if url.startswith("http") else f"https://{url}"
print(f"🔐 {oauth_provider} 로그인 시작: {target_url}")
prompt, model = get_prompt(oauth_provider)
agent_config = {
"url": target_url,
"log_context": f"{oauth_provider} 로그인",
"agent_params": {
"initial_actions": [{"open_tab": {"url": target_url}}],
"sensitive_data": GetSensitiveData(),
"task": (
f"Navigate to the login page, find and click the {oauth_provider} OAuth button, "
f"then follow the complete OAuth login flow as far as possible with a real user account. "
f"Capture the final redirect URL after login completion. "
f"If login fails or encounters errors, report the issue. "
f"Focus only on {oauth_provider} - ignore other OAuth providers."
),
"llm": CreateChatGoogleGenerativeAI(config.GOOGLE_MODEL),
"planner_llm": (
CreateChatGoogleGenerativeAI(config.GOOGLE_PLANNER_MODEL)
if config.GOOGLE_PLANNER_MODEL and os.getenv("ENABLE_PLANNER_MODEL_OAUTH_LOGIN")
else None
),
"controller": Controller(
output_model=model if not isinstance(model, str) else None,
exclude_actions=["search_google", "unknown_action", "unkown"],
),
"extend_planner_system_message": prompt,
}
}
response = await _run_agent_with_retry(agent_config)
if response and response.final_result():
final_result = response.final_result()
print(f"✅ {oauth_provider} 로그인 완료")
logger(f"✅ {url} - {oauth_provider} 로그인 결과: {final_result}")
return True
print(f"❌ {oauth_provider} 로그인 실패")
return False

0
lib/utils/browser_use/clean_resources.py → src/lib/browser_use/clean_resources.py
View file

0
lib/utils/browser_use/func.py → src/lib/browser_use/func.py
View file

2
lib/utils/browser_use/init_profile.py → src/lib/browser_use/init_profile.py
View file

@ -1,5 +1,5 @@
import os import os
from lib.utils.browser_use.func import * from lib.browser_use.func import *
# Initialize configuration # Initialize configuration
proxy_url = setup_proxy() proxy_url = setup_proxy()

0
lib/utils/browser_use/model.py → src/lib/browser_use/model.py
View file

106
src/lib/browser_use/scanner.py Normal file
View file

@ -0,0 +1,106 @@
import asyncio
import os
import csv
from lib.utils import notify_backend, read_lines_between, is_html_url
from lib.browser_use.agents import extract_oauth_list, test_oauth_login
from lib.utils.progress import current_progress, load_progress, save_progress, progress_file
async def scan_one_url(url: str, skip_html_check: bool = False):
"""URL 스캔 통합 함수: OAuth 리스트 추출 → 개별 OAuth 로그인 시도"""
target_url = url if url.startswith("http") else f"https://{url}"
print(f"🚀 스캔 시작: {target_url}")
# Backend에 스캔 시작을 알림
notify_backend(target_url)
# 1) URL이 HTML 페이지인지 확인
if not is_html_url(target_url) and not skip_html_check:
print(f"❌ {target_url} 은(는) HTML이 아닙니다. 스킵합니다.")
return
# 1단계: OAuth 리스트 추출
oauth_entries = await extract_oauth_list(target_url)
if not oauth_entries:
print(f"❌ {target_url}에서 OAuth 제공자를 찾을 수 없습니다.")
return
print("-" * 50)
print(f"🔗 스캔 URL: {url}")
print(f"🔐 발견된 OAuth 제공자들: {len(oauth_entries)}개")
for entry in oauth_entries:
print(f" - {entry.provider}")
print("-" * 50)
# CSV에 OAuth 리스트 저장
csv_file = "./data/oauth_providers.csv"
file_exists = os.path.isfile(csv_file)
with open(csv_file, "a", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
if not file_exists:
writer.writerow(["issuer", "provider", "oauth_uri", "login_tested"])
for entry in oauth_entries:
writer.writerow([url, entry.provider, "", "pending"])
# 2단계: 각 OAuth 제공자별로 개별 로그인 시도
for i, oauth_entry in enumerate(oauth_entries):
print(
f"\n🔄 OAuth 로그인 테스트 {i+1}/{len(oauth_entries)}: {oauth_entry.provider}"
)
# OAuth 간 대기 시간
if i > 0:
print("⏳ OAuth 테스트 간 대기 중 (30초)...")
await asyncio.sleep(30)
# 개별 OAuth 로그인 시도
success = await test_oauth_login(url, oauth_entry.provider)
# 결과를 CSV에 업데이트 (간단하게 로그만 남김)
status = "success" if success else "failed"
print(f"📝 {oauth_entry.provider} 로그인 결과: {status}")
async def main_loop(
filepath: str, start_line: int, end_line: int, skip_html_check: bool = False
):
"""지정된 URL 목록에 대해 스캔을 실행하는 메인 루프"""
target_list = read_lines_between(
filepath=filepath, start_line=start_line, end_line=end_line
)
current_progress["total"] = len(target_list)
current_progress["start_line"] = start_line
current_progress["current_index"] = 0
prev_progress = load_progress()
if prev_progress and prev_progress.get("start_line") == start_line:
print("📋 이전 진행 상황을 발견했습니다:")
print(f" - 이전 완료: {prev_progress['current_index']}/{prev_progress['total']}")
print(f" - 마지막 처리: {prev_progress.get('current_url', 'N/A')}")
resume = input("이어서 진행하시겠습니까? (y/n): ").lower().strip()
if resume == 'y':
start_index = prev_progress.get("current_index", 0)
current_progress["current_index"] = start_index
target_list = target_list[start_index:]
print(f"✅ {start_index}번째부터 재개합니다.")
for i, url in enumerate(target_list):
actual_index = current_progress["current_index"] + i
current_progress["current_url"] = url
print(f"\n🔄 Processing {actual_index + 1}/{current_progress['total']}: {url}")
print(f"📍 {os.path.basename(filepath)}의 {start_line + actual_index}번째 줄")
if i > 0:
print("⏳ API 쿼터 보호를 위해 30초 대기 중...")
await asyncio.sleep(30)
await scan_one_url(url, skip_html_check=skip_html_check)
current_progress["current_index"] = actual_index + 1
save_progress()
print(f"\n🎉 모든 스캔이 완료되었습니다! ({current_progress['total']}개 URL)")

0
lib/utils/browser_use/sensitive_data.py → src/lib/browser_use/sensitive_data.py
View file

0
lib/llm/__init__.py → src/lib/llm/__init__.py
View file

0
lib/llm/create.py → src/lib/llm/create.py
View file

25
src/lib/llm/prompt/__init__.py Normal file
View file

@ -0,0 +1,25 @@
from typing import Union, Type
from pydantic import BaseModel
def get_prompt(type: str) -> tuple[str, Type[BaseModel]] | str:
"""
Prompt를 반환합니다.
:param type: 'auth' {Auth List} 또는 'google' {OAuth Provider}, 'meta' {OAuth Provider}을 지정합니다.
:return: 해당하는 프롬프트 문자열 또는 (프롬프트, 모델) 튜플
"""
if type.lower() == "auth":
from lib.llm.prompt.get_oauth import prompt, model
return prompt, model
elif type.lower() in ["google", "google account"]:
from lib.llm.prompt.google import prompt, model
return prompt, model
elif type.lower() in ["microsoft", "microsoftonline"]:
from lib.llm.prompt.microsoft import prompt, model
return prompt, model
else:
from lib.llm.prompt.fallback import model, prompt
return prompt, model

2
src/lib/llm/prompt/fallback/__init__.py Normal file
View file

@ -0,0 +1,2 @@
from lib.llm.prompt.fallback.prompt import prompt
from lib.llm.prompt.fallback.model import model

6
src/lib/llm/prompt/fallback/model.py Normal file
View file

@ -0,0 +1,6 @@
from pydantic import BaseModel
class model(BaseModel):
msg: str | None = None
status: str | None = None # "success", "mfa_required", "google_blocked", "sso_not_found", "login_page_not_found", "invalid_credentials"
final_url: str | None = None

16
lib/llm/prompt/fallback.py → src/lib/llm/prompt/fallback/prompt.py
View file

@ -1,5 +1,5 @@
# Extended planner prompt # Extended planner prompt
extend_planner_system_message = f""" prompt = f"""
🎯 목적: 웹 자동화를 위한 **SSO 로그인 리디렉션 URL 수집** 🎯 목적: 웹 자동화를 위한 **SSO 로그인 리디렉션 URL 수집**
📌 주의사항 (전제 조건) 📌 주의사항 (전제 조건)
@ -96,6 +96,20 @@ chrome://settings/clearBrowserData에 들어가서 삭제해주세요.
--- ---
최종 반환:
Return the result in the following format only:
```json
{{
"msg": "Google login completed",
"status": "success" | "already_logged_in" | "mfa_required" | "captcha_triggered" | "window_blocked" | "idpw_required" | "google_blocked" | "sso_not_found" | "login_page_not_found",
"final_url": "<url_after_login_redirect or empty string>"
}}
```
---
## 📎 중요 규칙 요약 ## 📎 중요 규칙 요약
* ✅ **모든 SSO 로그인은 반드시 실행** (가능한 버튼은 모두 클릭) * ✅ **모든 SSO 로그인은 반드시 실행** (가능한 버튼은 모두 클릭)

2
src/lib/llm/prompt/get_oauth/__init__.py Normal file
View file

@ -0,0 +1,2 @@
from lib.llm.prompt.get_oauth.prompt import prompt
from lib.llm.prompt.get_oauth.model import model

6
src/lib/llm/prompt/get_oauth/model.py Normal file
View file

@ -0,0 +1,6 @@
from pydantic import BaseModel
class model(BaseModel):
msg: str | None = None
url: str | None = None
sso_list: list[str] = [] # List of SSO providers found on the login page

6
lib/llm/prompt/auth_list.py → src/lib/llm/prompt/get_oauth/prompt.py
View file

@ -1,6 +1,4 @@
# @file purpose: This file contains the prompt for extracting a list of OAuth providers from a web page. prompt = """
# OAuth 리스트 추출용 프롬프트 (클릭하지 않고 단순 식별만)
extract_oauth_list_prompt = f"""
🎯 목적: 주어진 초기 URL 내에서 **OAuth 로그인 Provider**를 찾아 아래 형식의 JSON으로 정리합니다. 🎯 목적: 주어진 초기 URL 내에서 **OAuth 로그인 Provider**를 찾아 아래 형식의 JSON으로 정리합니다.
📌 작업 목표: 📌 작업 목표:
@ -36,6 +34,4 @@ extract_oauth_list_prompt = f"""
📌 주의: 📌 주의:
결과가 없는 경우 빈 배열 `[]`로 반환합니다. 결과가 없는 경우 빈 배열 `[]`로 반환합니다.
정확한 provider 이름을 포함해 주세요. 정확한 provider 이름을 포함해 주세요.
""" """

2
src/lib/llm/prompt/google/__init__.py Normal file
View file

@ -0,0 +1,2 @@
from lib.llm.prompt.google.prompt import prompt
from lib.llm.prompt.google.model import model

6
src/lib/llm/prompt/google/model.py Normal file
View file

@ -0,0 +1,6 @@
from pydantic import BaseModel
class model(BaseModel):
msg: str | None = None
status: str | None = None # "success", "mfa_required", "google_blocked", "sso_not_found", "login_page_not_found", "invalid_credentials"
final_url: str | None = None

58
src/lib/llm/prompt/google/prompt.py Normal file
View file

@ -0,0 +1,58 @@
import os
# Extended planner prompt
prompt = f"""
You are a web automation agent.
Your task is to visit the given domain and perform a full login via the **Google SSO button**, following all steps strictly as described below.
▶ Target: Find a login page inside this domain that allows "Sign in with Google", and use it to complete login via Google.
Instructions:
1. If any cookie or privacy popups appear, dismiss or accept them.
2. Navigate through the site's UI to find the **login or sign-in page** (e.g., via buttons like "Log In", "Sign In", "Get Started").
- Only follow links within the same domain.
3. On the login page, look for a clearly labeled **Google SSO button** — typically labeled as:
- "Continue with Google"
- "Sign in with Google"
- or a button with the Google 'G' icon
4. Click the **Google login button**.
- ⚠️ The Google login flow MUST open in a **new browser tab** (not a new window or popup).
- ❌ If the login opens in a new **window** or **popup**, do NOT continue. Immediately stop and return the appropriate status.
5. Check if the user is **already logged in to Google and immediately redirected back to the original site** without showing a Google login screen.
- ✅ If so, treat the login as successful and return immediately.
6. If redirected to the Google login page:
- If a **CAPTCHA**, **MFA prompt**, or a request for **ID/password entry** appears, do NOT proceed.
- Immediately stop and return the appropriate status.
7. If login proceeds without interruptions, wait for redirection back to the original site and record the final URL.
Credentials to use for Google login:
- Email: {os.getenv("GOOGLE_EMAIL", "")}
- Password: {os.getenv("GOOGLE_PASSWORD", "")}
Constraints:
- Do NOT use search engines or guess URLs.
- Do NOT use autofill, saved sessions, or cookies.
- Do NOT proceed with login if:
- The login opens in a new window (only tabs are allowed)
- CAPTCHA or MFA appears
- ID/password input is required
- If the user is already logged in to Google and redirected back automatically, stop there and report success.
- If the login page cannot be found, return "login_page_not_found".
- If the Google login button is not found, return "sso_not_found".
- If a page such as a sign-up page appears, consider it a successful login and terminate immediately.
Final Output:
Return the result in the following format only:
```json
{{
"msg": "Google login completed",
"status": "success" | "already_logged_in" | "mfa_required" | "captcha_triggered" | "window_blocked" | "idpw_required" | "google_blocked" | "sso_not_found" | "login_page_not_found",
"final_url": "<url_after_login_redirect or empty string>"
}}
```
- Return ONLY the JSON object. Do NOT include any explanation, logging, or extra output.
"""

2
src/lib/llm/prompt/microsoft/__init__.py Normal file
View file

@ -0,0 +1,2 @@
from lib.llm.prompt.microsoft.prompt import prompt
from lib.llm.prompt.microsoft.model import model

57
src/lib/llm/prompt/microsoft/microsoft.py Normal file
View file

@ -0,0 +1,57 @@
import os
# This code snippet is used to generate a prompt for a web automation agent that performs Microsoft SSO login.
prompt = f"""
당신은 웹 자동화 에이전트입니다.
당신의 임무는 주어진 도메인에 방문하여 아래에 엄격히 설명된 모든 단계를 따라 **Microsoft SSO 버튼**을 통해 전체 로그인을 수행하는 것입니다.
▶ 목표: 이 도메인 내에서 "Microsoft로 로그인"이 가능한 로그인 페이지를 찾아 Microsoft을 통해 로그인을 완료하세요.
지침:
1. 쿠키 또는 개인정보 팝업이 나타나면 닫거나 수락하세요.
2. 사이트의 UI를 탐색하여 **로그인 또는 로그인 페이지**(예: "로그인", "Sign In", "Get Started"와 같은 버튼)를 찾으세요.
- 동일한 도메인 내의 링크만 따라가세요.
3. 로그인 페이지에서 명확하게 표시된 **Microsoft SSO 버튼**을 찾으세요. 일반적으로 다음과 같이 표시됩니다:
- "Microsoft로 계속"
- "Microsoft로 로그인"
- 또는 Microsoft 'M' 아이콘이 있는 버튼
4. **Microsoft 로그인 버튼**을 클릭하세요.
- ⚠️ Microsoft 로그인 플로우는 반드시 **새 브라우저 탭**에서 열려야 합니다 (새 창이나 팝업이 아님).
- ❌ 로그인이 새 **창**이나 **팝업**에서 열리면, 즉시 중단하고 적절한 상태를 반환하세요.
5. 사용자가 **이미 Microsoft에 로그인되어 있고 즉시 원래 사이트로 리디렉션**된다면,
- ✅ 이 경우 로그인이 성공한 것으로 간주하고 즉시 반환하세요.
6. Microsoft 로그인 페이지로 리디렉션된 경우:
- **CAPTCHA**, **MFA 프롬프트**, 또는 **ID/비밀번호 입력** 요청이 나타나면 진행하지 마세요.
- 즉시 중단하고 적절한 상태를 반환하세요.
7. 로그인에 방해가 없다면, 원래 사이트로 리디렉션될 때까지 기다리고 최종 URL을 기록하세요.
Microsoft 로그인에 사용할 자격 증명:
- 이메일: {os.getenv("MICROSOFT_EMAIL", "")}
- 비밀번호: {os.getenv("MICROSOFT_PASSWORD", "")}
제약 사항:
- 검색 엔진을 사용하거나 URL을 추측하지 마세요.
- 자동완성, 저장된 세션 또는 쿠키를 사용하지 마세요.
- 다음과 같은 경우 로그인 절차를 진행하지 마세요:
- 로그인이 새 창에서 열릴 때 (탭만 허용)
- CAPTCHA 또는 MFA가 나타날 때
- ID/비밀번호 입력이 요구될 때
- 사용자가 이미 Microsoft에 로그인되어 자동으로 리디렉션된다면, 그 즉시 성공으로 보고 종료하세요.
- 로그인 페이지를 찾을 수 없으면 "login_page_not_found"를 반환하세요.
- Microsoft 로그인 버튼을 찾을 수 없으면 "sso_not_found"를 반환하세요.
- 회원가입 페이지와 같은 화면이 나타나면 성공적인 로그인으로 간주하고 즉시 종료하세요.
최종 출력:
다음 형식으로만 결과를 반환하세요:
```json
{{
"msg": "Microsoft login completed",
"status": "success" | "already_logged_in" | "mfa_required" | "captcha_triggered" | "window_blocked" | "idpw_required" | "microsoft_blocked" | "sso_not_found" | "login_page_not_found",
"final_url": "<url_after_login_redirect or empty string>"
}}
```
- Return ONLY the JSON object. Do NOT include any explanation, logging, or extra output.
"""

6
src/lib/llm/prompt/microsoft/model.py Normal file
View file

@ -0,0 +1,6 @@
from pydantic import BaseModel
class model(BaseModel):
msg: str | None = None
status: str | None = None # "success", "mfa_required", "microsoft_blocked", "sso_not_found", "login_page_not_found", "invalid_credentials"
final_url: str | None = None

7
src/lib/utils/__init__.py Normal file
View file

@ -0,0 +1,7 @@
# export from show_info
from lib.utils.agent_info import *
from lib.utils.data import *
from lib.utils.config import *
from lib.utils.parsing.is_html import *
from lib.utils.parsing.read_txt import *

20
lib/utils/agent_info.py → src/lib/utils/agent_info.py
View file

@ -4,7 +4,9 @@ from lib.utils.config import (
GOOGLE_MODEL, GOOGLE_MODEL,
GOOGLE_PLANNER_MODEL, GOOGLE_PLANNER_MODEL,
) )
import os
from dotenv import load_dotenv
load_dotenv(override=True)
def show_info(): def show_info():
print("🔧 환경 설정:") print("🔧 환경 설정:")
@ -38,3 +40,19 @@ def browser_use_version():
def env_cheker(): def env_cheker():
if GOOGLE_API_KEY is None: if GOOGLE_API_KEY is None:
raise ValueError("GOOGLE_API_KEY 환경변수가 설정되지 않았습니다.") raise ValueError("GOOGLE_API_KEY 환경변수가 설정되지 않았습니다.")
if GOOGLE_PLANNER_MODEL != None and (not os.getenv("ENABLE_PLANNER_MODEL_OAUTH_LOGIN") or not os.getenv("ENABLE_PLANNER_MODEL_OAUTH_LIST")):
print(
"⚠️ GOOGLE_PLANNER_MODEL이 설정되어 있지만, ENABLE_PLANNER_MODEL_OAUTH_LOGIN 또는 ENABLE_PLANNER_MODEL_OAUTH_LIST가 활성화되지 않았습니다."
)
print(
"⚠️ Planner 모델을 사용하려면 .env 파일에서 ENABLE_PLANNER_MODEL_OAUTH_LOGIN과 ENABLE_PLANNER_MODEL_OAUTH_LIST를 true로 설정하세요."
)
print(
"‼️ 하지만 현재 Planner 모델을 사용하는 것이 권장되지 않습니다. 이 기능은 오작동을 일으킬 수 있습니다."
)
print(
"⚠️ 이 경고는 1초동안 정지합니다."
)
# 이 경고는 1초동안 sleep
import time
time.sleep(1)

4
lib/utils/config.py → src/lib/utils/config.py
View file

@ -4,5 +4,5 @@ load_dotenv(verbose=True, override=True)
BACKEND_URL = os.getenv("BACKEND_URL", "http://localhost:11081") BACKEND_URL = os.getenv("BACKEND_URL", "http://localhost:11081")
GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY") GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY")
GOOGLE_MODEL = os.getenv("GOOGLE_MODEL", "gemini-2.5-flash-preview-05-20") GOOGLE_MODEL = os.getenv("GOOGLE_MODEL", "gemini-2.5-flash")
GOOGLE_PLANNER_MODEL = os.getenv("GOOGLE_PLANNER_MODEL", "gemini-2.5-pro-preview-06-05") GOOGLE_PLANNER_MODEL = os.getenv("GOOGLE_PLANNER_MODEL")

2
src/lib/utils/data/__init__.py Normal file
View file

@ -0,0 +1,2 @@
from lib.utils.data.backend_client import *
from lib.utils.data.logger import *

0
lib/utils/backend_client.py → src/lib/utils/data/backend_client.py
View file

0
lib/utils/logger.py → src/lib/utils/data/logger.py
View file

0
lib/utils/is_html.py → src/lib/utils/parsing/is_html.py
View file

0
lib/utils/read_txt.py → src/lib/utils/parsing/read_txt.py
View file

48
src/lib/utils/progress.py Normal file
View file

@ -0,0 +1,48 @@
import json
import os
import signal
from pathlib import Path
# 진행 상황 추적을 위한 전역 변수
current_progress = {"current_index": 0, "total": 0, "current_url": "", "start_line": 0}
progress_file = Path("data/scan_progress.json")
def save_progress():
"""현재 진행 상황을 파일에 저장"""
progress_file.parent.mkdir(parents=True, exist_ok=True)
with open(progress_file, "w", encoding="utf-8") as f:
json.dump(current_progress, f, ensure_ascii=False, indent=2)
def load_progress():
"""이전 진행 상황을 파일에서 불러오기"""
if os.path.exists(progress_file):
try:
with open(progress_file, "r", encoding="utf-8") as f:
return json.load(f)
except (json.JSONDecodeError, IOError):
return None
return None
def signal_handler(signum, frame):
"""Ctrl+C 시그널 핸들러"""
print("\n" + "=" * 60)
print("🛑 스캔이 중단되었습니다!")
print(f"📊 진행 상황:")
print(f" - 전체: {current_progress['total']}개 URL")
print(f" - 완료: {current_progress['current_index']}개 URL")
print(f" - 현재 처리 중: {current_progress['current_url']}")
print(
f" - domains.txt의 {current_progress['start_line'] + current_progress['current_index']}번째 줄"
)
if current_progress['total'] > 0:
print(
f" - 진행률: {current_progress['current_index']}/{current_progress['total']} ({current_progress['current_index']/current_progress['total']*100:.1f}%)"
)
print("=" * 60)
save_progress()
print(f"💾 진행 상황이 {progress_file}에 저장되었습니다.")
exit(0)
def setup_signal_handler():
"""시그널 핸들러 등록"""
signal.signal(signal.SIGINT, signal_handler)

79
src/main.py Normal file
View file

@ -0,0 +1,79 @@
import asyncio
import argparse
import os
from dotenv import load_dotenv
from lib.utils import env_cheker
from lib.browser_use.scanner import main_loop
from lib.utils.progress import setup_signal_handler, progress_file
# .env 파일 로드
load_dotenv(verbose=True, override=True)
# 환경 변수 체크
env_cheker()
# Laminar 초기화 (선택적)
if os.getenv("LMNR_PROJECT_API_KEY"):
try:
from lmnr import Laminar
Laminar.initialize(project_api_key=os.getenv("LMNR_PROJECT_API_KEY"))
except ImportError:
print("⚠️ Laminar 라이브러리가 설치되지 않았습니다. 관련 기능이 비활성화됩니다.")
def main():
"""애플리케이션 메인 진입점"""
# 시그널 핸들러 설정
setup_signal_handler()
parser = argparse.ArgumentParser(
prog="domain_scanner",
description="도메인 목록 파일에서 지정한 줄 범위를 읽어 SSO 스캔을 수행합니다.",
)
parser.add_argument(
"-f",
"--file",
type=str,
required=True,
help="도메인 목록이 들어 있는 텍스트 파일 경로 (예: ./domains.txt)",
)
parser.add_argument(
"-s", "--start", type=int, required=True, help="읽기 시작 줄 번호 (1-based)"
)
parser.add_argument(
"-e", "--end", type=int, required=True, help="읽기 종료 줄 번호 (1-based)"
)
parser.add_argument(
"-skh",
"--skip-html-check",
action='store_true', # 플래그 형식으로 변경
help="HTML 페이지 체크를 건너뛰고 모든 URL을 스캔합니다.",
)
args = parser.parse_args()
try:
asyncio.run(
main_loop(
filepath=args.file,
start_line=args.start,
end_line=args.end,
skip_html_check=args.skip_html_check,
)
)
except KeyboardInterrupt:
# signal_handler가 처리하므로 여기서는 별도 처리 불필요
pass
finally:
# 정상 종료 시 진행 상황 파일 삭제
if os.path.exists(progress_file):
try:
os.remove(progress_file)
except OSError as e:
print(f"오류: 진행 상황 파일을 삭제하지 못했습니다. {e}")
if __name__ == "__main__":
main()