From 085cc7a140e704f3b2372ee1c26f95437627b67f Mon Sep 17 00:00:00 2001 From: akiba Date: Tue, 23 Jun 2026 08:04:04 +0000 Subject: [PATCH] =?UTF-8?q?AO3=20Mirror=20v4=20=E2=80=94=20initial=20commi?= =?UTF-8?q?t?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cookie-aware proxy pool with CF challenge solving - Tiered proxy pool (fast 50 + main 676) - Per-proxy cf_clearance cookie persistence - CF challenge detection + user-browser solving - Safari + Chrome TLS fingerprint rotation - Async FastAPI backend with LRU cache - Passive daemon with systemd supervision - Stats dashboard + Prometheus metrics --- .env.example | 43 +++ .gitignore | 31 ++ AGENTS.md | 110 ++++++ Caddyfile | 123 +++++++ SOUL.md | 185 ++++++++++ ao3_fetcher.py | 399 +++++++++++++++++++++ app.py | 672 +++++++++++++++++++++++++++++++++++ cache.py | 123 +++++++ proxy_pool.py | 465 ++++++++++++++++++++++++ scripts/daemon.py | 146 ++++++++ scripts/restart_workers.py | 25 ++ scripts/scan_proxies_cffi.py | 131 +++++++ start.sh | 76 ++++ stats.py | 202 +++++++++++ stop.sh | 23 ++ url_rewriter.py | 88 +++++ 16 files changed, 2842 insertions(+) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 AGENTS.md create mode 100644 Caddyfile create mode 100644 SOUL.md create mode 100644 ao3_fetcher.py create mode 100644 app.py create mode 100644 cache.py create mode 100644 proxy_pool.py create mode 100644 scripts/daemon.py create mode 100644 scripts/restart_workers.py create mode 100644 scripts/scan_proxies_cffi.py create mode 100755 start.sh create mode 100644 stats.py create mode 100755 stop.sh create mode 100644 url_rewriter.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..37950f4 --- /dev/null +++ b/.env.example @@ -0,0 +1,43 @@ +# AO3 Mirror — Environment Configuration +# Copy to .env.local and fill in real values: +# cp .env.example .env.local + +# ── Mirror ───────────────────────────────────────────────── +MIRROR_DOMAIN=agento3.miscs.dev +AO3_URL=https://archiveofourown.org +PROXY_FILE=/home/ubuntu/proxy.txt +WORKING_PROXIES_FILE=/dev/shm/working_proxies.txt + +# Worker config +WORKER_COUNT=2 +WORKER_START_PORT=8081 +UVICORN_WORKERS=1 + +# ── Monitoring ──────────────────────────────────────────── +PRODUCTION_URL=https://agento3.miscs.dev +HEALTH_CHECK_URL=https://agento3.miscs.dev/health +SENTRY_DSN=your-sentry-dsn + +# ── Notifications — at least one required for alerts ───── +# Telegram: create bot via @BotFather, get chat ID by messaging @userinfobot +TELEGRAM_BOT_TOKEN=your-telegram-bot-token +TELEGRAM_CHAT_ID=your-telegram-chat-id + +# Slack +SLACK_WEBHOOK_URL=your-slack-webhook-url + +# Email (for status reports) +EMAIL_TO=your-email@example.com + +# ── GitHub (for issue tracking, PR management) ─────────── +GITHUB_REPO=owner/ao3-mirror +GITHUB_USERNAME=your-username +GITHUB_TOKEN=ghp_your-fine-grained-token + +# ── Cloudflare ──────────────────────────────────────────── +CF_API_TOKEN=your-cloudflare-api-token +CF_ZONE_ID=your-zone-id + +# ── WebShare Proxy ──────────────────────────────────────── +WEBSHARE_API_KEY=your-webshare-api-key +PROXY_REFRESH_INTERVAL_MIN=30 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..4a1d61e --- /dev/null +++ b/.gitignore @@ -0,0 +1,31 @@ +# Python +__pycache__/ +*.py[cod] +*.pyo +*.so + +# Logs +*.log + +# Environment +.env +.env.local +.env.*.local + +# IDE +.vscode/ +.idea/ + +# OS +.DS_Store +Thumbs.db + +# Runtime artifacts +/dev/shm/ +working_proxies.txt + +# Caddy +caddy-access.log + +# Systemd service files (deployed separately) +*.service diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..100b157 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,110 @@ +# AGENTS — AO3 Mirror + +This file is for AI agents (Hermes, Claude Code, Codex) working on the AO3 reverse proxy mirror. + +## Project Overview + +AO3 Mirror is a high-availability reverse proxy for archiveofourown.org, designed to restore access for Chinese users. It bypasses Cloudflare's bot protection using TLS fingerprint impersonation (curl_cffi), a tiered WebShare proxy pool (726 proxies, 72%+ CF bypass rate), cookie-aware session management, and user-browser CF challenge solving. + +- **Domain**: agento3.miscs.dev (behind Cloudflare CDN) +- **Target**: archiveofourown.org +- **Stack**: Python 3.11, FastAPI, uvicorn + uvloop + httptools, curl_cffi, Caddy +- **QPS**: 720 (cache) | 270 (direct proxy) + +## Build Commands + +```bash +# Start all workers + reload Caddy +cd /home/ubuntu/ao3-mirror && bash start.sh + +# Stop everything +cd /home/ubuntu/ao3-mirror && bash stop.sh + +# Restart daemon only +sudo systemctl restart ao3-daemon + +# Restart Caddy only +sudo systemctl reload caddy + +# Scan proxy pool (MUST run in foreground!) +cd /home/ubuntu/ao3-mirror && python3 scripts/scan_proxies_cffi.py + +# Syntax check +python3 -m py_compile proxy_pool.py ao3_fetcher.py app.py cache.py stats.py url_rewriter.py +``` + +## Architecture + +``` +User → Cloudflare CDN (agento3.miscs.dev) → Caddy :443 (60s, round-robin) + → 2× uvicorn workers (8081-8082) [cpu-pinned, uvloop] + → Tiered Proxy Pool: + Fast (50): POST/login → 8s timeout, 1 retry + Main (676): GET/browse → 15s timeout, 2 retries + → Cookie-aware sessions: per-proxy cf_clearance persistence + → TLS impersonation: safari15_5, safari17_0, chrome123, chrome124 + → archiveofourown.org + +CF Challenge Solving (v4): + 1. All proxies hit CF challenge → generate challenge_token + 2. Rewrite challenge page → forward to user's browser + 3. User browser executes CF JS → challenge solved + 4. cf_clearance captured → saved to proxy cookie jar + 5. Original request retried → content delivered +``` + +### Core Files + +| File | Purpose | +|------|---------| +| `app.py` | FastAPI backend v4 — proxy handler, stats, CF challenge solving | +| `proxy_pool.py` | Tiered async proxy pool — cookie jar, weighted selection, sampling | +| `ao3_fetcher.py` | Async fetcher — CF detection, smart retry, cookie injection | +| `cache.py` | Per-worker LRU cache (5000 entries, path-differentiated TTL) | +| `stats.py` | In-memory stats with batch SQLite flush every 60s | +| `url_rewriter.py` | URL/header rewriting (ao3 → mirror domain) | +| `scripts/daemon.py` | Passive worker supervision (systemd, 30s check, 3-strike restart) | +| `scripts/scan_proxies_cffi.py` | Proxy scanner with TLS fingerprint fallback | +| `start.sh` | Startup script — kills old workers, starts new, reloads Caddy | +| `Caddyfile` | Caddy config — TLS, round-robin, 60s timeouts | + +## Security Baseline + +- No secrets in code — proxy credentials are in `/home/ubuntu/proxy.txt` +- Worker processes bound to 127.0.0.1 only (not exposed publicly) +- Cloudflare CDN terminates TLS and provides DDoS protection at edge +- CORS headers restrict cross-origin access +- stats/health/metrics endpoints are read-only, no mutation +- All proxied content is user-facing; no admin endpoints exposed + +## Engine Guidance + +- Complex multi-file changes, architecture evolution → Hermes (here) +- Quick targeted fixes, single-file changes → Hermes +- Deploy, monitor, notify, schedule → Hermes +- Proxy scanning, proxy pool refresh → Hermes cron (30min) +- Not sure? Start with Hermes — everything runs via Hermes + +## Monitoring + +- Health endpoint: https://agento3.miscs.dev/health +- Stats dashboard: https://agento3.miscs.dev/stats +- Metrics (Prometheus): https://agento3.miscs.dev/metrics +- Worker logs: /home/ubuntu/ao3-mirror/worker-0.log, worker-1.log +- Daemon log: /home/ubuntu/ao3-mirror/daemon.log +- Systemd: `systemctl status ao3-daemon`, `journalctl -u ao3-daemon` + +## Deployment + +- Single server (current host) +- Caddy manages Let's Encrypt TLS certs on agento3.miscs.dev +- CF CDN fronts the domain (Orange Cloud = on) +- Proxy pool: WebShare static residential proxies, refreshed every 30min via Hermes cron +- No CI/CD — manual deploy via start.sh + +## Commit Conventions + +- One commit per meaningful change +- Python files only (no binaries, no .pyc, no logs, no .env) +- SOUL.md updated to reflect architectural changes +- ao3-mirror skill updated when workflows change diff --git a/Caddyfile b/Caddyfile new file mode 100644 index 0000000..55f70d8 --- /dev/null +++ b/Caddyfile @@ -0,0 +1,123 @@ +# AO3 Mirror - Caddy 配置 +# 前端负载均衡 + TLS 终止 + Cloudflare CDN 集成 + +agento3.miscs.dev { + # 全局头 + header { + Strict-Transport-Security "max-age=31536000; includeSubDomains" + X-Content-Type-Options "nosniff" + X-Frame-Options "SAMEORIGIN" + X-XSS-Protection "1; mode=block" + Referrer-Policy "strict-origin-when-cross-origin" + -Server + -X-Powered-By + } + + # 日志 + log { + output file /var/log/ao3-mirror/access.log { + roll_size 100mb + roll_keep 7 + roll_keep_for 720h + } + format json + } + + # 压缩 + encode gzip + + # 后端负载均衡 (轮询) - 2 workers on 2-core machine + reverse_proxy 127.0.0.1:8081 127.0.0.1:8082 { + lb_policy round_robin + + health_uri /health + health_interval 10s + health_timeout 5s + + # 超时配置 + transport http { + read_timeout 30s + write_timeout 30s + dial_timeout 5s + } + + header_up X-Forwarded-For {remote_host} + header_up X-Forwarded-Proto {scheme} + header_up X-Real-IP {remote_host} + } +} + +# 统计子域名 - 监控面板和指标 +stats.agento3.miscs.dev { + # 安全头 + header { + Strict-Transport-Security "max-age=31536000; includeSubDomains" + X-Content-Type-Options "nosniff" + X-Frame-Options "SAMEORIGIN" + Referrer-Policy "strict-origin-when-cross-origin" + -Server + -X-Powered-By + } + + # 日志 + log { + output file /var/log/ao3-mirror/stats-access.log { + roll_size 100mb + roll_keep 7 + roll_keep_for 720h + } + format json + } + + # 压缩 + encode gzip + + # 路由规则 - 只暴露 /stats 和 /metrics + route { + # /stats 页面 - 代理到后端 + reverse_proxy /stats* 127.0.0.1:8081 127.0.0.1:8082 { + lb_policy round_robin + + health_uri /health + health_interval 10s + health_timeout 5s + + transport http { + read_timeout 30s + write_timeout 30s + dial_timeout 5s + } + + header_up X-Forwarded-For {remote_host} + header_up X-Forwarded-Proto {scheme} + header_up X-Real-IP {remote_host} + } + + # /metrics 指标 - 代理到后端 + reverse_proxy /metrics* 127.0.0.1:8081 127.0.0.1:8082 { + lb_policy round_robin + + health_uri /health + health_interval 10s + health_timeout 5s + + transport http { + read_timeout 30s + write_timeout 30s + dial_timeout 5s + } + + header_up X-Forwarded-For {remote_host} + header_up X-Forwarded-Proto {scheme} + header_up X-Real-IP {remote_host} + } + + # 健康检查 + respond /health* 200 { + body `{"status":"ok"}` + } + + # 其他路径返回 404 + respond * 404 + } +} diff --git a/SOUL.md b/SOUL.md new file mode 100644 index 0000000..abd3149 --- /dev/null +++ b/SOUL.md @@ -0,0 +1,185 @@ +# AO3 Mirror — SOUL v4 + +## 项目宗旨 + +为中国用户恢复 archiveofourown.org 的访问。维护一个高可用、高效、经济的反向代理镜像站。 +目标：单台服务器承受 **200+ QPS**，CF 5 秒盾自动求解。 + +## 核心指标 + +| 指标 | 当前值 | 目标 | +|------|--------|------| +| 本地 QPS（缓存命中） | **720 QPS** | ≥200 | +| 平均延迟（缓存命中） | **14-42ms** | ≤100ms | +| 代理存活率（AO3/CF） | **~72%** (最佳端口) | ≥70% | +| 代理池 | 726 WebShare (最佳端口) | — | +| 有效代理 | ~524 (72%) | ≥500 | +| Cookie 代理 | — | ≥200 (持有 cf_clearance) | +| CF 挑战 | 主动求解 (用户浏览器) | 零感知 | + +## 架构总览（v4 — Cookie 感知 + CF 挑战求解） + +``` +用户 ──→ Cloudflare CDN ──→ Caddy (443, 60s) ──→ 2× uvicorn workers (8081-8082) [全异步] + │ + ┌──────┴──────┐ + │ Tiered Pool │ + │ │ + │ Fast (50) │ ← POST/login (8s, 1 retry) + │ Main (676) │ ← GET/browse (15s, 2 retries) + │ │ + │ Cookie 感知 │ ← 每个代理持有自己的 cookie jar + │ (cf_clearance│ 成功请求后自动保存 Set-Cookie + │ 等持久化) │ 后续请求自动注入 + └──────┬──────┘ + │ + curl_cffi AsyncSession + (safari15_5/17_0, chrome123/124 + TLS 指纹 + 连接复用) +``` + +### v4 关键改进 + +| 改进 | v3 | v4 | +|------|-----|-----| +| Cookie 持久化 | ❌ 裸请求 | ✅ 每代理 cookie jar | +| CF 挑战检测 | ❌ 当代理死亡 | ✅ 分离检测，不退避 | +| CF 挑战求解 | ❌ 放弃 | ✅ 用户浏览器自动求解 | +| 代理亲和性 | ❌ 随机 | ✅ Challenge token 保证同 proxy | +| TLS 指纹 | chrome123/124 | safari15_5, safari17_0, chrome123/124 | +| 统计准确度 | ❌ alive=total (假数据) | ✅ 实时遍历计数 | +| Cookie 代理数 | N/A | ✅ 统计面板展示 | +| 用户感知 | 502 错误页 | 短暂「检查浏览器」→ 自动恢复 | + +## 核心组件 + +### 1. 异步代理池 (proxy_pool.py) v4 +- `ProxySession` — 每个代理持有自己的 `AsyncSession` + **cookie jar** +- Cookie 生命周期：成功请求自动保存 → 过期自动清理 → 后续请求自动注入 +- CF 挑战分离：`mark_challenged()` ≠ `mark_failure()` — 不退避 +- 加权轮询：响应速度越快的代理，被选中的概率越大 +- 指数退避：仅对真失败 (connection error/timeout) → 5s→15s→30s→60s→120s→300s +- Cookie-aware 选择：`get_proxy_with_cookies()` 优先返回持有 cf_clearance 的代理 + +### 2. 异步抓取器 (ao3_fetcher.py) v4 +- `async fetch_url()` — Cookie 感知 + CF 挑战检测 + 智能重试 +- `is_cf_challenge()` — 检测 403/503 是否为 CF 挑战页（body markers + Server header） +- 智能重试策略： + 1. 首次尝试 → 优先用带 cookie 的代理 + 2. 遇到 CF 挑战 → 换代理（不退避） + 3. 遇到真失败 → 退避 + 换代理 +- 重试 2 次（fast 路径 1 次） +- 请求超时：15s (普通), 8s (交互) + +### 3. 后端服务 (app.py) v4 +FastAPI，绑定 127.0.0.1。 +- **Challenge token 映射**：`_challenge_map[token] → (method, url, headers, body, proxy_host)` + - 用户浏览器求解 CF 挑战时，保证后续请求使用同一代理（IP 亲和性） + - TTL 120s，过期自动清理 +- **挑战页面透传**：所有代理都遇 CF 挑战 → 重写页面 → 发给用户浏览器 + - 注入 `_cf_token` cookie 和 meta 标签 + - 用户浏览器自动执行 CF JS → 验证通过 → redirect 回镜像 + - Worker 捕获 cf_clearance → 保存到代理 cookie jar +- Cookie 流转发：AO3 的 Set-Cookie (用户 session + cf_clearance) 域名重写后传给用户 +- URL 重写：`archiveofourown.org` → `agento3.miscs.dev` +- CORS 全开 +- 端点：`/health` `/stats` `/metrics` `/robots.txt` + +### 4. 缓存 (cache.py) +- LRU 实现，容量 5000 +- TTL 按路径差异化（首页30s, 章节120s, 图片600s） +- 每个 worker 独立 + +### 5. 统计系统 (stats.py) v2 +- 内存热路径：计时器 + 桶式计数 (1s 粒度) +- SQLite 批量写入：每 60s flush +- 统计面板新增：Cookie 代理数 + +### 6. URL 重写器 (url_rewriter.py) +- HTML/CSS/JS 中 `archiveofourown.org` → `agento3.miscs.dev` +- 响应头 Location/Set-Cookie 重写 +- CF 挑战页面特殊 URL 处理 + +## 代理策略 + +### TLS 指纹（按优先级） +1. `safari15_5` — 最高 CF 绕过率 +2. `safari17_0` — 次优 +3. `chrome123` — 稳定 +4. `chrome124` — 稳定 + +**已弃用**：chrome110, chrome116 (100% 被CF拦截), chrome120, edge120 (低成功率) + +### 代理分级 +1. **Fast Pool** (50): 最快代理，用于 POST/登录/注册 (8s timeout, 1 retry) +2. **Main Pool** (676): 全部可用代理，用于 GET/浏览 (15s timeout, 2 retries) +3. **Cookie 池**: 持有 cf_clearance 的代理，优先使用 + +### 代理扫描器 (scripts/scan_proxies_cffi.py) +- curl_cffi 2 种浏览器指纹 (chrome123/124) +- 50 并发线程 +- 每 30 分钟 cron 自动扫描 +- 仅扫描最佳端口范围 13500-14499 (72.6% 成功率) +- 结果排序（快→慢） + +## 负载均衡 + +``` +Caddy (agento3.miscs.dev:443) + ├── Round Robin → 127.0.0.1:8081 + └── Round Robin → 127.0.0.1:8082 +``` + +## 自维护框架（被动模式） + +| 层级 | 类型 | 频率 | 作用 | +|------|------|------|------| +| daemon.py | systemd service | 30s 轮询 | Worker 被动监督，3 次挂才重启 | +| 代理刷新 | Hermes cron | 30min | 重新扫描代理池 | +| 统计报告 | Hermes cron | 1h | 聚合指标 + 异常告警 | + +**不运行**：e2e 测试、Caddy 检查、文件日志、主动健康检查全部 726 代理 + +## 关键路径 + +``` +/home/ubuntu/ +├── proxy.txt # 5000 原始代理 +├── ao3-mirror/ +│ ├── app.py # FastAPI v4 (cookie-aware + CF solver) +│ ├── proxy_pool.py # 异步代理池 v4 (cookie jar) +│ ├── ao3_fetcher.py # 异步抓取器 v4 (CF detection) +│ ├── cache.py # LRU 缓存 (5000) +│ ├── stats.py # 内存热路径统计 v2 +│ ├── url_rewriter.py # URL 重写 +│ ├── Caddyfile # Caddy 配置 +│ ├── start.sh / stop.sh +│ ├── ao3-mirror.service / ao3-daemon.service +│ ├── SOUL.md ← 本文档 +│ ├── scripts/ +│ │ ├── daemon.py # 自维护守护进程 +│ │ ├── scan_proxies_cffi.py # TLS 指纹代理扫描 +│ │ └── restart_workers.py # Worker 重启工具 +│ └── worker-*.log # Per-worker stdout logs +/dev/shm/ +├── working_proxies.txt # 有效代理 +├── ao3_stats.db # 统计数据（冷存储） +└── ao3/ + └── status.json # 守护进程状态 +``` + +## 已知限制 +- 每个 worker 独立缓存，无共享层 +- Cookie jar 是内存存储，worker 重启后丢失 +- 用户浏览器求解 CF 挑战需要 120s 内完成（token TTL） +- 挑战页面 JS 执行依赖用户浏览器环境 + +## 下一步 +- [x] Cookie 感知代理层 (v4) +- [x] CF 挑战检测与分离 (v4) +- [x] 用户浏览器 CF 挑战求解 (v4) +- [ ] 共享缓存层 (Redis/memcached) 跨 worker +- [ ] Cookie jar 持久化 (重启后恢复) +- [ ] 多 proxy provider 源 +- [ ] 预加载热门页面到缓存 +- [ ] Worker 间 cookie jar 同步 diff --git a/ao3_fetcher.py b/ao3_fetcher.py new file mode 100644 index 0000000..5af87ec --- /dev/null +++ b/ao3_fetcher.py @@ -0,0 +1,399 @@ +""" +异步 AO3 内容抓取器 v4 — Cookie 感知 + CF 挑战检测 + 智能重试 + +v4 vs v3: +- CF 挑战检测：不再把 403/503 一律当失败 +- Cookie 注入：请求时自动携带 proxy 级别的 cookies（cf_clearance 等） +- Cookie 保存：成功请求后自动提取 Set-Cookie 并保存到 proxy cookie jar +- 智能重试：挑战时优先用带 cookie 的代理，无 cookie 则换代理 +- 挑战页面透传：所有重试都遇到挑战时，返回挑战 HTML 让用户浏览器求解 +""" +import asyncio +import logging +import time +from typing import Optional + +from proxy_pool import get_proxy_pool, ProxySession + +logger = logging.getLogger("ao3-fetcher") + +# ─── CF Challenge Detection ─────────────────────────────────────────────── + +# Markers in response body that indicate a Cloudflare challenge page +CF_CHALLENGE_MARKERS = [ + b'/cdn-cgi/challenge-platform', + b'cf-challenge-running', + b'cf-browser-verification', + b'window._cf_chl_opt', + b'challenge-platform', + b'cf-turnstile', + b'cf_chl_', + # Sometimes CF just returns "Just a moment..." without JS markers + b'Checking your browser', + b'Just a moment...', +] + + +def is_cf_challenge(status: int, body: bytes, headers: dict) -> bool: + """Detect if response is a Cloudflare challenge page (not a real error).""" + if status not in (403, 503, 429): + return False + # Quick check: CF always sets Server header on challenge pages + server = headers.get("Server", headers.get("server", "")) + if "cloudflare" not in server.lower(): + # Check body for challenge markers + for marker in CF_CHALLENGE_MARKERS: + if marker in body: + return True + return False + # Server: cloudflare + non-200 status = likely challenge + for marker in CF_CHALLENGE_MARKERS: + if marker in body: + return True + # If Server is cloudflare and status is 403/503, it's almost certainly a challenge + if status in (403, 503): + return True + return False + + +# ─── Headers ────────────────────────────────────────────────────────────── + +CHROME_HEADERS = { + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8", + "Accept-Language": "en-US,en;q=0.9,zh-CN;q=0.8,zh;q=0.7", + "Sec-Ch-Ua": '"Not A(Brand";v="99", "Google Chrome";v="120", "Chromium";v="120"', + "Sec-Ch-Ua-Mobile": "?0", + "Sec-Ch-Ua-Platform": '"Windows"', + "Sec-Fetch-Dest": "document", + "Sec-Fetch-Mode": "navigate", + "Sec-Fetch-Site": "none", + "Sec-Fetch-User": "?1", + "Upgrade-Insecure-Requests": "1", + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36", + "DNT": "1", +} + +API_HEADERS = { + "Accept": "*/*", + "Accept-Language": "en-US,en;q=0.9", + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36", +} + +# ─── Timeout config ─────────────────────────────────────────────────────── + +FAST_REQUEST_TIMEOUT = 8 # login/register/signup +NORMAL_REQUEST_TIMEOUT = 15 +CONNECT_TIMEOUT = 5 +MAX_RETRIES = 2 +FAST_MAX_RETRIES = 1 + +FAST_PATHS = { + "/users/login", + "/users/sign_up", + "/users/new", + "/invitation_requests", + "/token_dispenser.json", + "/user_sessions", +} + + +def _is_fast_path(url: str) -> bool: + for fp in FAST_PATHS: + if fp in url: + return True + if "/users/" in url or "/user_sessions" in url: + return True + return False + + +def _now() -> float: + try: + return asyncio.get_running_loop().time() + except RuntimeError: + return time.time() + + +def _merge_cookies(proxy_cookies: str, user_cookies: str) -> str: + """Merge proxy-level cookies (cf_clearance) with user cookies. User cookies take precedence.""" + if not proxy_cookies and not user_cookies: + return "" + if not proxy_cookies: + return user_cookies + if not user_cookies: + return proxy_cookies + # User cookies take priority (they contain session auth) + # But put proxy cookies first so user cookies can override + return f"{proxy_cookies}; {user_cookies}" + + +async def fetch_url( + url: str, + method: str = "GET", + headers: Optional[dict] = None, + body: Optional[bytes] = None, + cookies: Optional[dict] = None, + is_api: bool = False, + preferred_proxy: Optional[str] = None, # v4: proxy host:port for challenge affinity +) -> dict: + """ + Async fetch from AO3 through proxy pool with cookie-aware session reuse. + + Priority-based: + - Login/register paths → fast pool + 8s timeout + 1 retry + - Normal paths → cookie-preferring proxies + 15s timeout + 2 retries + + v4 improvements: + - CF challenge detection: don't treat challenge as proxy failure + - Cookie injection: auto-send saved cf_clearance per proxy + - Cookie saving: auto-extract Set-Cookie on success + - Challenge body returned for user-browser solving + + Returns: { + "status": 200, + "headers": {...}, + "body": b"...", + "cookies": {...}, + "success": True/False, + "error": "...", + "elapsed": 0.5, + "proxy_host": "p.webshare.io:10296", + # v4: + "is_challenge": False, + "challenge_body": b"" | None, # CF challenge HTML for user-browser solving + "challenge_proxy": "" | None, # which proxy got the challenge + } + """ + pool = get_proxy_pool() + base_headers = API_HEADERS.copy() if is_api else CHROME_HEADERS.copy() + + if headers: + for h in ["Cookie", "Referer", "Content-Type", "X-Requested-With", "Accept", + "Origin", "X-CSRF-Token", "Authorization"]: + if h in headers: + base_headers[h] = headers[h] + + user_cookie_str = "" + if cookies: + user_cookie_str = "; ".join(f"{k}={v}" for k, v in cookies.items()) + + # Determine priority level + is_fast = _is_fast_path(url) or method in ("POST", "PUT", "PATCH") + request_timeout = FAST_REQUEST_TIMEOUT if is_fast else NORMAL_REQUEST_TIMEOUT + max_retries = FAST_MAX_RETRIES if is_fast else MAX_RETRIES + + last_error = None + last_challenge_body = None + last_challenge_proxy = None + seen_proxies = set() # Don't retry with same proxy + + # v4: if preferred_proxy is set (from challenge affinity), use it first + if preferred_proxy: + try: + target_proxy = None + for p in pool._proxies: + if p.host_port == preferred_proxy: + target_proxy = p + break + if target_proxy and target_proxy.is_available: + result = await _try_proxy( + target_proxy, url, method, body, base_headers, user_cookie_str, + request_timeout, is_fast + ) + if result["success"]: + # preferred proxy worked — save cookies + target_proxy.save_cookies(result.get("headers", {})) + return result + if result.get("is_challenge"): + last_challenge_body = result.get("challenge_body") + last_challenge_proxy = target_proxy.host_port + seen_proxies.add(target_proxy.host_port) + except Exception: + pass + + for attempt in range(max_retries): + # v4: prefer proxies with cookies, fallback to any available + if attempt == 0: + # First attempt: use proxy with cookies if available + proxy = pool.get_proxy_with_cookies() + if not proxy: + proxy = pool.get_fast_proxy() if is_fast else pool.get_proxy() + elif attempt == 1 and last_challenge_body: + # Second attempt after challenge: try another proxy with cookies + proxy = pool.get_proxy_with_cookies() + # Make sure it's not the same one + if proxy and proxy.host_port in seen_proxies: + proxy = pool.get_fast_proxy() if is_fast else pool.get_proxy() + if not proxy: + proxy = pool.get_fast_proxy() if is_fast else pool.get_proxy() + else: + proxy = pool.get_fast_proxy() if is_fast else pool.get_proxy() + + if not proxy or proxy.host_port in seen_proxies: + # Find an unseen proxy + for _ in range(5): + p = pool.get_proxy() + if p and p.host_port not in seen_proxies: + proxy = p + break + if not proxy or proxy.host_port in seen_proxies: + continue + + seen_proxies.add(proxy.host_port) + + result = await _try_proxy( + proxy, url, method, body, base_headers, user_cookie_str, + request_timeout, is_fast + ) + + if result["success"]: + # Save response cookies to proxy cookie jar for future requests + proxy.save_cookies(result.get("headers", {})) + return result + + if result.get("is_challenge"): + last_challenge_body = result.get("challenge_body") + last_challenge_proxy = proxy.host_port + last_error = result.get("error", "CF_CHALLENGE") + # Don't break — try another proxy + continue + + # Real failure (connection error, timeout) + last_error = result.get("error", "UNKNOWN") + + # All retries exhausted + return { + "status": 0, "headers": {}, "body": b"", "cookies": {}, + "success": False, + "error": f"All retries failed: {last_error}", + "elapsed": 0, "proxy_host": None, + "is_challenge": last_challenge_body is not None, + "challenge_body": last_challenge_body, + "challenge_proxy": last_challenge_proxy, + } + + +async def _try_proxy( + proxy: ProxySession, + url: str, + method: str, + body: Optional[bytes], + base_headers: dict, + user_cookie_str: str, + request_timeout: int, + is_fast: bool, +) -> dict: + """Try a single request through one proxy. Returns result dict.""" + host_port = f"{proxy.host}:{proxy.port}" + start_time = _now() + + try: + session = await proxy.get_session() + + # v4: Build per-request headers with proxy cookies + # IMPORTANT: use headers= parameter (never session.headers.update() — race condition!) + request_headers = base_headers.copy() + + # Inject proxy-level cookies (cf_clearance, etc.) + proxy_cookie_str = proxy.get_cookie_header() + cookie_str = _merge_cookies(proxy_cookie_str, user_cookie_str) + if cookie_str: + request_headers["Cookie"] = cookie_str + + # Execute request + if method == "GET": + resp = await session.get(url, timeout=request_timeout, headers=request_headers) + elif method == "POST": + resp = await session.post(url, data=body, timeout=request_timeout, headers=request_headers) + elif method == "HEAD": + resp = await session.head(url, timeout=request_timeout, headers=request_headers) + else: + resp = await session.request(method, url, data=body, timeout=request_timeout, + headers=request_headers) + + elapsed = _now() - start_time + status = resp.status_code + resp_body = resp.content + resp_headers = dict(resp.headers) + resp_cookies = {} + if hasattr(resp, "cookies"): + for k, v in resp.cookies.items(): + resp_cookies[k] = v + + # v4: Check for CF challenge + if is_cf_challenge(status, resp_body, resp_headers): + proxy.mark_challenged() + logger.warning(f"[CF_CHALLENGE] {host_port} -> {url[:60]}: {status} ({elapsed:.1f}s)") + return { + "status": status, + "headers": resp_headers, + "body": resp_body, + "cookies": resp_cookies, + "success": False, + "error": f"CF_CHALLENGE_{status}", + "elapsed": elapsed, + "proxy_host": host_port, + "is_challenge": True, + "challenge_body": resp_body, + "challenge_proxy": host_port, + } + + # Success — any 2xx-4xx is proxied through + if 200 <= status < 500: + proxy.mark_success(elapsed) + logger.debug(f"{host_port} -> {url[:60]}: {status} ({elapsed:.2f}s)") + + # v4: Save response cookies to proxy + proxy.save_cookies(resp_headers) + + return { + "status": status, + "headers": resp_headers, + "body": resp_body, + "cookies": resp_cookies, + "success": True, + "elapsed": elapsed, + "proxy_host": host_port, + "is_challenge": False, + "challenge_body": None, + "challenge_proxy": None, + } + + # True error status (5xx) + proxy.mark_failure() + logger.warning(f"Attempt: {host_port} -> {url[:60]}: {status}") + return { + "status": status, + "headers": resp_headers, + "body": resp_body, + "cookies": resp_cookies, + "success": False, + "error": f"HTTP_{status}", + "elapsed": elapsed, + "proxy_host": host_port, + "is_challenge": False, + "challenge_body": None, + "challenge_proxy": None, + } + + except asyncio.TimeoutError: + proxy.mark_failure() + elapsed = _now() - start_time + logger.warning(f"TIMEOUT: {host_port} -> {url[:60]} ({elapsed:.1f}s, timeout={request_timeout}s)") + return { + "status": 0, "headers": {}, "body": b"", "cookies": {}, + "success": False, "error": "TIMEOUT", + "elapsed": elapsed, "proxy_host": host_port, + "is_challenge": False, "challenge_body": None, "challenge_proxy": None, + } + + except Exception as e: + proxy.mark_failure() + elapsed = _now() - start_time + err_str = str(e)[:120] + logger.debug(f"ERROR: {host_port} -> {e} ({elapsed:.1f}s)") + return { + "status": 0, "headers": {}, "body": b"", "cookies": {}, + "success": False, "error": err_str, + "elapsed": elapsed, "proxy_host": host_port, + "is_challenge": False, "challenge_body": None, "challenge_proxy": None, + } diff --git a/app.py b/app.py new file mode 100644 index 0000000..e0909e7 --- /dev/null +++ b/app.py @@ -0,0 +1,672 @@ +""" +AO3 反代后端 v4 — Cookie 感知 + CF 挑战用户浏览器求解 + +v4 vs v3: +- Challenge token 映射：用户浏览器求解 CF 挑战时，保证同一 proxy 亲和性 +- 挑战页面透传：所有代理都遇到 CF 挑战时，把挑战页发给用户浏览器求解 +- Cookie 流：成功响应自动保存 proxy cookie，后续请求自动携带 +- 统计面板展示 cookie-aware 代理数 +""" +import hashlib +import json +import logging +import os +import secrets +import sys +import threading +import time +from urllib.parse import urlparse, urlunparse + +from fastapi import FastAPI, Request, Response +from fastapi.responses import HTMLResponse, JSONResponse, PlainTextResponse, RedirectResponse +import uvicorn + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from proxy_pool import get_proxy_pool +from ao3_fetcher import fetch_url, is_cf_challenge +from url_rewriter import rewrite_body, rewrite_response_headers, needs_rewrite, MIRROR_DOMAIN +from cache import get_cache, get_ttl_for_path +from stats import get_stats_collector + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + handlers=[logging.StreamHandler()], +) +logger = logging.getLogger("ao3-backend") + +AO3_BASE = "https://archiveofourown.org" +MIRROR_HOST = MIRROR_DOMAIN + +LOCAL_PATHS = {"/stats", "/health", "/metrics", "/favicon.ico", "/robots.txt"} + +# ─── Challenge Token Map (v4) ────────────────────────────────────────────── +# Maps challenge_token → (method, ao3_url, headers, body, cookies, proxy_host, expires) +_challenge_map: dict[str, tuple] = {} +_challenge_lock = threading.Lock() +CHALLENGE_TOKEN_TTL = 120 # 2 minutes for the user's browser to solve the challenge + + +def _make_challenge_token() -> str: + return secrets.token_urlsafe(16) + + +def _store_challenge(token: str, method: str, ao3_url: str, headers: dict, + body: bytes, cookies: dict, proxy_host: str): + with _challenge_lock: + # Clean expired tokens + now = time.time() + expired = [k for k, v in _challenge_map.items() if v[6] < now] + for k in expired: + del _challenge_map[k] + _challenge_map[token] = (method, ao3_url, headers, body, cookies, + proxy_host, now + CHALLENGE_TOKEN_TTL) + + +def _get_challenge(token: str) -> tuple | None: + with _challenge_lock: + entry = _challenge_map.get(token) + if entry and entry[6] > time.time(): + del _challenge_map[token] + return entry + if entry: + del _challenge_map[token] # expired + return None + + +# ─── App ─────────────────────────────────────────────────────────────────── + +app = FastAPI( + title="AO3 Mirror", + description="AO3 reverse proxy mirror for Chinese users", + version="4.0.0", + docs_url=None, + redoc_url=None, +) + + +def get_client_ip(request: Request) -> str: + cf_ip = request.headers.get("CF-Connecting-IP") + if cf_ip: + return cf_ip + forwarded = request.headers.get("X-Forwarded-For") + if forwarded: + return forwarded.split(",")[0].strip() + return request.client.host if request.client else "unknown" + + +def build_ao3_url(path: str, query: str = "") -> str: + url = f"{AO3_BASE}{path}" + if query: + url = f"{url}?{query}" + return url + + +# ─── Challenge page URL rewriting ────────────────────────────────────────── + +def _rewrite_challenge_page(body: bytes, proxy_host: str, token: str) -> bytes: + """Rewrite CF challenge page so all URLs go through mirror, and tag with challenge token.""" + import re + + # Basic AO3 domain rewrite + body = body.replace(b"archiveofourown.org", MIRROR_DOMAIN.encode()) + + # Rewrite /cdn-cgi/ challenge endpoints to go through mirror + # These are CF's internal challenge platform URLs + body = body.replace( + b"/cdn-cgi/challenge-platform", + f"/cdn-cgi/challenge-platform".encode() + ) + + # Inject a marker meta tag so we can detect challenge pages coming back + marker = f''.encode() + marker_cookie = ( + f'' + ).encode() + body = body.replace(b"", b"" + marker + marker_cookie, 1) + if b"" not in body: + body = b"" + marker + marker_cookie + b"" + body + + return body + + +# ─── Response header helpers ─────────────────────────────────────────────── + +def filter_response_headers(headers: dict) -> dict: + blocked = { + "transfer-encoding", "content-encoding", + "alt-svc", "cf-ray", "cf-cache-status", "cf-request-id", + "server", "x-powered-by", + "cross-origin-resource-policy", "cross-origin-embedder-policy", + "cross-origin-opener-policy", "cross-origin-window-policy", + "accept-ch", "critical-ch", + } + result = {} + for key, value in headers.items(): + if key.lower() in blocked: + continue + if key.lower().startswith("cf-"): + continue + result[key] = value + result["X-Proxy"] = "AO3-Mirror/4.0" + result["X-Cache"] = "MISS" + result["Cache-Control"] = "public, max-age=60, s-maxage=60" + return result + + +def add_cors(response: Response): + response.headers["Access-Control-Allow-Origin"] = "*" + response.headers["Access-Control-Allow-Methods"] = "GET, POST, HEAD, OPTIONS, PUT, DELETE, PATCH" + response.headers["Access-Control-Allow-Headers"] = "Content-Type, Authorization, Cookie, X-CSRF-Token" + response.headers["Access-Control-Max-Age"] = "86400" + response.headers["Access-Control-Allow-Credentials"] = "true" + + +# ─── Local Routes ────────────────────────────────────────────────────────── + +@app.get("/health") +async def health(): + pool = get_proxy_pool() + stats = pool.get_stats() + resp = JSONResponse({ + "status": "ok", + "timestamp": time.time(), + "version": "4.0.0", + "proxy_pool": stats, + }) + resp.headers["Cache-Control"] = "no-store, no-cache, must-revalidate" + resp.headers["Pragma"] = "no-cache" + return resp + + +@app.get("/robots.txt") +async def robots(): + return PlainTextResponse( + "User-agent: *\nDisallow: /stats\nDisallow: /health\n" + ) + + +# ─── Stats Dashboard ─────────────────────────────────────────────────────── + +@app.get("/stats", response_class=HTMLResponse) +async def stats_page(): + collector = get_stats_collector() + pool = get_proxy_pool() + cache = get_cache() + + stats = collector.get_overview() + proxy_stats = pool.get_stats() + cache_stats = cache.get_stats() + + hourly = stats.get("hourly", []) + chart_labels = json.dumps([ + time.strftime("%H:%M", time.localtime(h["hour"])) + for h in hourly[-24:] + ]) + chart_requests = json.dumps([h["total"] for h in hourly[-24:]]) + chart_success = json.dumps([h["successful"] for h in hourly[-24:]]) + chart_elapsed = json.dumps([h["avg_elapsed"] * 1000 for h in hourly[-24:]]) + + return f""" + + + + + AO3 Mirror - 统计面板 v4 + + + + +
+

🛡 AO3 Mirror 统计面板 v4

+ +
+
+
总请求数
+
{stats['total_requests']:,}
+
运行 {stats['uptime_human']}
+
+
+
最近 1 分钟
+
{stats['recent_1m']:,} req
+
最近 5 分钟: {stats['recent_5m']:,}
+
+
+
成功率
+
{stats['success_rate']}%
+
失败: {stats['failed']:,}
+
+
+
缓存命中率
+
{stats['cache_rate']}%
+
已缓存: {stats['cached']:,}
+
+
+
平均延迟
+
{stats['avg_elapsed_ms']} ms
+
P99: {stats['p99_elapsed_ms']} ms
+
+
+
代理池
+
{proxy_stats['alive']}/{proxy_stats['total']}
+
可用: {proxy_stats['available']} | 受损: {proxy_stats['dead']} | Banned: {proxy_stats['banned']}
+
+
+
Cookie 代理
+
{proxy_stats['with_cookies']}/{proxy_stats['alive']}
+
已持有 cf_clearance
+
+
+
本地缓存
+
{cache_stats['size']}/{cache_stats['capacity']}
+
已用: {cache_stats['usage_pct']}%
+
+
+ +
+
+

请求趋势 (最近 24 小时)

+ +
+
+

响应延迟 (最近 24 小时)

+ +
+
+ +
+

热门路径

+ + + {''.join(f'' for p in stats['top_paths'][:15])} +
路径请求数成功平均延迟
{p["path"]}{p["requests"]:,}0.8 else "badge-red"}">{round(p["successful"]/max(p["requests"],1)*100)}%{p["avg_elapsed"]*1000:.0f}ms
+
+
+ + + +""" + + +# ─── Metrics ─────────────────────────────────────────────────────────────── + +@app.get("/metrics") +async def metrics(): + collector = get_stats_collector() + pool = get_proxy_pool() + cache = get_cache() + + stats = collector.get_overview() + proxy_stats = pool.get_stats() + cache_stats = cache.get_stats() + + lines = [ + "# HELP ao3_mirror_requests_total Total proxy requests", + "# TYPE ao3_mirror_requests_total counter", + f'ao3_mirror_requests_total {stats["total_requests"]}', + "# HELP ao3_mirror_successful_requests Successful requests", + "# TYPE ao3_mirror_successful_requests counter", + f'ao3_mirror_successful_requests {stats["successful"]}', + "# HELP ao3_mirror_failed_requests Failed requests", + "# TYPE ao3_mirror_failed_requests counter", + f'ao3_mirror_failed_requests {stats["failed"]}', + "# HELP ao3_mirror_cache_hits Cache hit count", + "# TYPE ao3_mirror_cache_hits counter", + f'ao3_mirror_cache_hits {stats["cached"]}', + "# HELP ao3_mirror_avg_elapsed_ms Average response time", + "# TYPE ao3_mirror_avg_elapsed_ms gauge", + f'ao3_mirror_avg_elapsed_ms {stats["avg_elapsed_ms"]}', + "# HELP ao3_mirror_proxy_pool Proxy pool status", + "# TYPE ao3_mirror_proxy_pool gauge", + f'ao3_mirror_proxy_alive {proxy_stats["alive"]}', + f'ao3_mirror_proxy_dead {proxy_stats["dead"]}', + f'ao3_mirror_proxy_banned {proxy_stats["banned"]}', + f'ao3_mirror_proxy_available {proxy_stats["available"]}', + f'ao3_mirror_proxy_with_cookies {proxy_stats["with_cookies"]}', + "# HELP ao3_mirror_cache_size Current cache size", + "# TYPE ao3_mirror_cache_size gauge", + f'ao3_mirror_cache_size {cache_stats["size"]}', + ] + return PlainTextResponse("\n".join(lines)) + + +# ─── Proxy Core (v4) ────────────────────────────────────────────────────── + +@app.api_route("/{path:path}", methods=["GET", "POST", "HEAD", "OPTIONS", "PUT", "DELETE", "PATCH"]) +async def proxy_handler(request: Request, path: str): + """Main proxy handler — with cookie-aware fetching and CF challenge solving.""" + start_time = time.time() + client_ip = get_client_ip(request) + + if path == "" or path == "/": + path = "" + + full_path = f"/{path}" if path else "/" + if full_path in LOCAL_PATHS or full_path.startswith("/stats") or full_path.startswith("/health"): + return JSONResponse({"error": "Not found"}, status_code=404) + + if request.method == "OPTIONS": + resp = Response() + add_cors(resp) + return resp + + # ── v4: Challenge token check ────────────────────────────────────────── + # If this request has a _cf_token cookie, it's part of a challenge resolution flow + cf_token = request.cookies.get("_cf_token") or request.query_params.get("_cf_token") + preferred_proxy = None + if cf_token: + challenge_entry = _get_challenge(cf_token) + if challenge_entry: + orig_method, orig_url, orig_headers, orig_body, orig_cookies, proxy_host, _ = challenge_entry + preferred_proxy = proxy_host + logger.info(f"Challenge resolution: using proxy {proxy_host} for {orig_url[:60]}") + # Re-fetch the original request with the challenge proxy + result = await fetch_url( + url=orig_url, method=orig_method, + headers=dict(request.headers), + body=await request.body() if orig_method in ("POST", "PUT", "PATCH") else None, + cookies=orig_cookies, + is_api="/api/" in orig_url, + preferred_proxy=preferred_proxy, + ) + + if result["success"]: + # Challenge solved! Proxy now has cf_clearance. Return content. + elapsed = time.time() - start_time + rewritten_body = rewrite_body(result["body"], + result["headers"].get("Content-Type", "")) + rewritten_headers = rewrite_response_headers(result["headers"]) + final_headers = filter_response_headers(rewritten_headers) + final_headers["X-Cache"] = "MISS" + final_headers["X-CF-Status"] = "solved" + + # Set cf_clearance as a cookie on the mirror domain too + # so subsequent requests benefit + resp = Response(content=rewritten_body, status_code=result["status"], + headers=final_headers) + # Forward any Set-Cookie from AO3 (includes cf_clearance) + for k, v in result["headers"].items(): + if k.lower() == "set-cookie": + # Rewrite domain + v_rewritten = v.replace("domain=archiveofourown.org", + f"domain={MIRROR_DOMAIN}") + v_rewritten = v_rewritten.replace("domain=.archiveofourown.org", + f"domain=.{MIRROR_DOMAIN}") + resp.headers.add("Set-Cookie", v_rewritten) + + # Clean up challenge token cookie + resp.delete_cookie("_cf_token", path="/") + + collector = get_stats_collector() + collector.log_request(method=orig_method, path=full_path, + status=result["status"], elapsed=elapsed, + cached=False, proxy_host=preferred_proxy, + client_ip=client_ip) + add_cors(resp) + return resp + + # ── Normal request flow ──────────────────────────────────────────────── + + query_string = request.url.query + ao3_url = build_ao3_url(f"/{path}" if path else "/", query_string) + + # Check cache + cache = get_cache() + cached = cache.get(ao3_url, dict(request.headers)) + if cached: + body, resp_headers, status = cached + elapsed = time.time() - start_time + collector = get_stats_collector() + collector.log_request(method=request.method, path=full_path, status=status, + elapsed=elapsed, cached=True, client_ip=client_ip) + headers = filter_response_headers(resp_headers) + headers["X-Cache"] = "HIT" + return Response(content=body, status_code=status, headers=headers) + + # Prepare request + method = request.method + client_headers = dict(request.headers) + req_body = None + if method in ("POST", "PUT", "PATCH"): + req_body = await request.body() + + cookies = {} + cookie_header = request.headers.get("cookie", "") + if cookie_header: + for pair in cookie_header.split(";"): + if "=" in pair: + k, v = pair.split("=", 1) + cookies[k.strip()] = v.strip() + + # Forward to AO3 + result = await fetch_url( + url=ao3_url, method=method, + headers=client_headers, body=req_body, cookies=cookies, + is_api="/api/" in full_path or full_path.startswith("/api/"), + preferred_proxy=preferred_proxy, + ) + + elapsed = time.time() - start_time + + # ── v4: CF Challenge handling ────────────────────────────────────────── + if result.get("is_challenge") and result.get("challenge_body"): + challenge_proxy = result.get("challenge_proxy", "unknown") + challenge_body = result["challenge_body"] + + # Generate challenge token for proxy affinity + token = _make_challenge_token() + _store_challenge(token, method, ao3_url, client_headers, + req_body or b"", cookies, challenge_proxy) + + # Rewrite challenge page for user's browser + rewritten_challenge = _rewrite_challenge_page(challenge_body, challenge_proxy, token) + + logger.warning( + f"CF Challenge detected via {challenge_proxy} for {ao3_url[:80]}. " + f"Sending challenge to user browser (token={token[:8]}...)" + ) + + collector = get_stats_collector() + collector.log_request(method=method, path=full_path, status=503, + elapsed=elapsed, cached=False, + proxy_host=challenge_proxy, client_ip=client_ip) + + # Return challenge page with 503 status + token cookie + resp = HTMLResponse( + content=rewritten_challenge, + status_code=503, + headers={ + "X-CF-Challenge": "true", + "X-CF-Challenge-Proxy": challenge_proxy, + "Retry-After": "5", + }, + ) + resp.set_cookie( + key="_cf_token", value=token, + path="/", max_age=CHALLENGE_TOKEN_TTL, + httponly=False, # JS needs to read it + samesite="lax", + ) + resp.delete_cookie("cf_clearance", path="/") # Clear stale clearance + add_cors(resp) + return resp + + # ── Standard failure ─────────────────────────────────────────────────── + if not result["success"]: + logger.error(f"Failed to fetch {ao3_url[:80]}: {result.get('error', 'unknown')}") + collector = get_stats_collector() + collector.log_request(method=method, path=full_path, status=502, + elapsed=elapsed, cached=False, client_ip=client_ip) + + error_html = f""" + +AO3 Mirror - 暂时不可用 + + +
+

🔄 正在尝试连接 AO3

+

镜像站正在尝试通过代理重新连接 AO3 服务器。

+

请稍后刷新页面重试。

+ 刷新页面 +
+ +""" + return HTMLResponse(content=error_html, status_code=502) + + # ── Success ──────────────────────────────────────────────────────────── + ao3_status = result["status"] + ao3_headers = result.get("headers", {}) + raw_body = result.get("body", b"") + + content_type = ao3_headers.get("Content-Type", "") + rewritten_body = rewrite_body(raw_body, content_type) + rewritten_headers = rewrite_response_headers(ao3_headers) + final_headers = filter_response_headers(rewritten_headers) + final_headers["X-Cache"] = "MISS" + + # Cache successful GET responses + if method == "GET" and 200 <= ao3_status < 400: + ttl = get_ttl_for_path(full_path) + cache.set(ao3_url, rewritten_body, rewritten_headers, ao3_status, ttl=ttl) + + # Handle redirects + if 300 <= ao3_status < 400 and "location" in rewritten_headers: + redirect_url = rewritten_headers["location"] + if AO3_BASE in redirect_url: + redirect_url = redirect_url.replace(AO3_BASE, "").replace("http://", "https://") + return RedirectResponse(url=redirect_url, status_code=ao3_status) + + collector = get_stats_collector() + collector.log_request(method=method, path=full_path, status=ao3_status, + elapsed=elapsed, cached=False, + proxy_host=result.get("proxy_host"), + client_ip=client_ip) + + resp = Response(content=rewritten_body, status_code=ao3_status, headers=final_headers) + # Forward Set-Cookie from AO3 (user session cookies + cf_clearance) + for k, v in ao3_headers.items(): + if k.lower() == "set-cookie": + v_rewritten = v.replace("domain=archiveofourown.org", + f"domain={MIRROR_DOMAIN}") + v_rewritten = v_rewritten.replace("domain=.archiveofourown.org", + f"domain=.{MIRROR_DOMAIN}") + resp.headers.add("Set-Cookie", v_rewritten) + add_cors(resp) + return resp + + +# ─── Startup / Shutdown ──────────────────────────────────────────────────── + +@app.on_event("startup") +async def startup(): + logger.info("AO3 Mirror backend v4 starting up...") + get_proxy_pool() + get_cache() + get_stats_collector() + logger.info("AO3 Mirror backend v4 started (cookie-aware + CF challenge solving)") + + +@app.on_event("shutdown") +async def shutdown(): + logger.info("AO3 Mirror backend shutting down...") + pool = get_proxy_pool() + await pool.close_all() + logger.info("AO3 Mirror backend stopped") + + +# ─── Entry point ─────────────────────────────────────────────────────────── + +if __name__ == "__main__": + port = int(os.environ.get("PORT", "8080")) + workers = int(os.environ.get("WORKERS", "4")) + logger.info(f"Starting AO3 Mirror v4 on port {port} with {workers} workers") + + uvicorn.run( + "app:app", + host="127.0.0.1", + port=port, + workers=workers, + log_level="info", + timeout_keep_alive=30, + ) diff --git a/cache.py b/cache.py new file mode 100644 index 0000000..c65cbfd --- /dev/null +++ b/cache.py @@ -0,0 +1,123 @@ +""" +简单高效的缓存层 +- 内存 LRU 缓存，每个 worker 独立 +- 短 TTL 避免内容过时 +- 针对不同路径设置不同 TTL +""" + +import hashlib +import threading +import time +from collections import OrderedDict +from typing import Optional + + +class LRUCache: + """Thread-safe LRU cache with TTL support.""" + + def __init__(self, capacity: int = 2000, default_ttl: int = 30): + self.capacity = capacity + self.default_ttl = default_ttl + self._cache: OrderedDict[str, tuple[float, bytes, dict, int]] = OrderedDict() + # (expiry_time, body, headers, status) + self._lock = threading.RLock() + + def _make_key(self, url: str, headers: Optional[dict] = None) -> str: + """Generate cache key from URL and key headers.""" + # Use URL + relevant headers + accept = "" + if headers: + accept = headers.get("Accept-Encoding", "") + raw = f"{url}|{accept}" + return hashlib.md5(raw.encode()).hexdigest() + + def get(self, url: str, headers: Optional[dict] = None) -> Optional[tuple[bytes, dict, int]]: + """Get cached response. Returns (body, headers, status) or None.""" + key = self._make_key(url, headers) + with self._lock: + if key not in self._cache: + return None + expiry, body, resp_headers, status = self._cache[key] + if time.time() > expiry: + del self._cache[key] + return None + # Move to end (most recently used) + self._cache.move_to_end(key) + return (body, resp_headers, status) + + def set(self, url: str, body: bytes, headers: dict, status: int, + ttl: Optional[int] = None, request_headers: Optional[dict] = None): + """Store response in cache.""" + key = self._make_key(url, request_headers) + t = ttl if ttl is not None else self.default_ttl + expiry = time.time() + t + + with self._lock: + self._cache[key] = (expiry, body, headers, status) + self._cache.move_to_end(key) + if len(self._cache) > self.capacity: + self._cache.popitem(last=False) + + def invalidate(self, url: str, headers: Optional[dict] = None): + """Remove a specific URL from cache.""" + key = self._make_key(url, headers) + with self._lock: + self._cache.pop(key, None) + + def clear(self): + with self._lock: + self._cache.clear() + + @property + def size(self) -> int: + with self._lock: + return len(self._cache) + + def get_stats(self) -> dict: + with self._lock: + return { + "size": len(self._cache), + "capacity": self.capacity, + "usage_pct": round(len(self._cache) / self.capacity * 100, 1) if self.capacity else 0, + } + + +# TTL 策略：不同路径不同缓存时间 +PATH_TTL = { + "/": 30, # 首页 30s + "/works": 60, # 作品列表 60s + "/chapters": 120, # 章节内容 120s + "/series": 60, # 系列 60s + "/collections": 60, + "/tags": 60, + "/users": 30, + "/pseuds": 30, + "/bookmarks": 60, + "/skins": 300, # CSS 皮肤缓存 5 分钟 + "/stylesheets": 300, + "/images": 600, # 图片缓存 10 分钟 + "/media": 600, + "/javascripts": 300, + "/api": 15, # API 响应 15s + "/external_links": 30, + # Default: 30s +} + + +def get_ttl_for_path(path: str) -> int: + """Determine cache TTL based on URL path.""" + for prefix, ttl in PATH_TTL.items(): + if path.startswith(prefix): + return ttl + return 30 # default + + +# 全局缓存实例 +_cache: Optional[LRUCache] = None + + +def get_cache() -> LRUCache: + global _cache + if _cache is None: + _cache = LRUCache(capacity=5000, default_ttl=30) + return _cache diff --git a/proxy_pool.py b/proxy_pool.py new file mode 100644 index 0000000..609e6cb --- /dev/null +++ b/proxy_pool.py @@ -0,0 +1,465 @@ +""" +异步代理池 v4 — Cookie 感知 + 挑战分离 + 真实统计 + +v4 vs v3: +- 每个代理维护 cookie jar（cf_clearance 等），成功请求后自动保存 +- CF 挑战不再标记为代理死亡（mark_challenged ≠ mark_failure） +- 真实统计：维护原子计数器，不造假数据 +- 更多 TLS 指纹：加入 safari15_5/safari17_0 +""" +import asyncio +import logging +import os +import random +import time +from email.utils import parsedate_to_datetime +from threading import Lock +from typing import Optional + +from curl_cffi.requests import AsyncSession + +logger = logging.getLogger("ao3-proxy-pool") + +OPTIMAL_MIN_PORT = 13500 +OPTIMAL_MAX_PORT = 14499 +WORKING_PROXIES_FILE = "/dev/shm/working_proxies.txt" + +FAST_POOL_SIZE = 50 + +# 被动检查：只对快池做轻量采样 +SAMPLE_INTERVAL = 60 +SAMPLE_BATCH_SIZE = 10 +SAMPLE_TIMEOUT = 5 +FAST_POOL_REFRESH_INTERVAL = 10 + +# TLS fingerprints — proven against AO3 Cloudflare +# safari15_5/17_0 have highest CF bypass rate per testing +BROWSER_IMPS = ["safari15_5", "safari17_0", "chrome123", "chrome124"] + +WARM_THRESHOLD_S = 3.0 + + +def _parse_cookie_expires(set_cookie: str) -> float: + """Extract expiry timestamp from Set-Cookie header. Returns 0 if session cookie.""" + for part in set_cookie.split(";"): + part = part.strip() + if part.lower().startswith("expires="): + try: + dt = parsedate_to_datetime(part[8:]) + if dt: + return dt.timestamp() + except Exception: + pass + elif part.lower().startswith("max-age="): + try: + return time.time() + int(part[9:]) + except Exception: + pass + return 0 # session cookie + + +def load_proxies_from_file() -> list[str]: + """Load proxy list from known-working file, or full list with port filtering.""" + if os.path.exists(WORKING_PROXIES_FILE) and os.path.getsize(WORKING_PROXIES_FILE) > 0: + with open(WORKING_PROXIES_FILE) as f: + proxies = [l.strip() for l in f if l.strip() and ":" in l] + if proxies: + logger.info(f"Loaded {len(proxies)} working proxies from {WORKING_PROXIES_FILE}") + filtered = [p for p in proxies if ":" in p and OPTIMAL_MIN_PORT <= int(p.split(":")[-1]) <= OPTIMAL_MAX_PORT] + if len(filtered) >= 10: + return filtered + return proxies + proxy_file = "/home/ubuntu/proxy.txt" + proxies = [] + with open(proxy_file) as f: + for line in f: + line = line.strip() + if not line or ":" not in line: + continue + if "|" in line: + line = line.split("|")[-1].strip() + proxies.append(line) + filtered = [p for p in proxies if ":" in p and OPTIMAL_MIN_PORT <= int(p.split(":")[-1]) <= OPTIMAL_MAX_PORT] + return filtered if len(filtered) >= 10 else proxies + + +class ProxySession: + """A single proxy with its own AsyncSession, cookie jar, and health state.""" + + __slots__ = ( + "host_port", "host", "port_str", "port", + "session", "impersonate", "alive", + "consecutive_failures", "ban_until", "last_used", + "avg_response_time", "weight", "requests_handled", + "last_sample", "sample_passed", + # v4: cookie jar + "_cookies", "_cookie_expires", "_lock", + ) + + def __init__(self, host_port: str): + self.host_port = host_port + self.host, self.port_str = host_port.split(":") + self.port = int(self.port_str) + self.session: Optional[AsyncSession] = None + self.impersonate = random.choice(BROWSER_IMPS) + self.alive = True + self.consecutive_failures = 0 + self.ban_until = 0.0 + self.last_used = 0.0 + self.avg_response_time = 1.0 + self.weight = 1.0 + self.requests_handled = 0 + self.last_sample = 0.0 + self.sample_passed = True + # v4: per-proxy cookie jar (cf_clearance, etc.) + self._cookies: dict[str, str] = {} + self._cookie_expires: dict[str, float] = {} + self._lock = Lock() + + # ─── Cookie management ──────────────────────────────────────────── + + def save_cookies(self, headers: dict) -> int: + """Extract Set-Cookie from response headers. Returns count saved.""" + saved = 0 + set_cookie = headers.get("Set-Cookie", "") + if not set_cookie: + # Some servers use set-cookie (lowercase) in HTTP/2 + set_cookie = headers.get("set-cookie", "") + if not set_cookie: + return 0 + + now = time.time() + with self._lock: + for part in set_cookie.split(","): + # Handle comma-separated cookies (ugh) + part = part.strip() + if "=" not in part: + continue + name, _, rest = part.partition("=") + value = rest.split(";")[0].strip() if ";" in rest else rest.strip() + name = name.strip() + if not name: + continue + self._cookies[name] = value + expires = _parse_cookie_expires(part) + if expires > 0: + self._cookie_expires[name] = expires + saved += 1 + + # Purge expired cookies + self._purge_expired(now) + if saved: + logger.debug(f"Saved {saved} cookies for {self.host_port} (keys: {list(self._cookies.keys())})") + return saved + + def get_cookie_header(self) -> str: + """Get Cookie header string for this proxy. Returns '' if no cookies.""" + now = time.time() + with self._lock: + self._purge_expired(now) + if not self._cookies: + return "" + return "; ".join(f"{k}={v}" for k, v in self._cookies.items()) + + def _purge_expired(self, now: float): + """Remove expired cookies.""" + expired = [k for k, exp in self._cookie_expires.items() if 0 < exp < now] + for k in expired: + self._cookies.pop(k, None) + self._cookie_expires.pop(k, None) + + # ─── Session management ─────────────────────────────────────────── + + async def get_session(self) -> AsyncSession: + if self.session is None: + self.session = AsyncSession() + self.session.impersonate = self.impersonate + self.session.timeout = 15 + self.session.proxies = { + "http": f"http://{self.host_port}", + "https": f"http://{self.host_port}", + } + # Default headers that don't change per-request (safe to set on session) + self.session.headers.update({ + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36", + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8", + "Accept-Language": "en-US,en;q=0.9", + "Accept-Encoding": "gzip, deflate, br", + }) + return self.session + + async def close(self): + if self.session: + try: + await self.session.close() + except Exception: + pass + self.session = None + + # ─── Health tracking ────────────────────────────────────────────── + + def update_weight(self): + self.weight = 1.0 / max(self.avg_response_time, 0.1) + + def mark_success(self, response_time: float): + self.alive = True + self.consecutive_failures = 0 + self.ban_until = 0.0 + self.requests_handled += 1 + self.avg_response_time = self.avg_response_time * 0.7 + response_time * 0.3 + self.update_weight() + + def mark_failure(self): + """Proxy-level failure (connection error, timeout, etc.) — exponential backoff.""" + self.requests_handled += 1 + self.consecutive_failures += 1 + backoff = min(5 * (3 ** (self.consecutive_failures - 1)), 300) + self.ban_until = time.time() + backoff + if self.consecutive_failures >= 3: + self.alive = False + self.sample_passed = False + + def mark_challenged(self): + """CF challenge detected — proxy is alive but needs cookie. NO backoff.""" + # Don't increment consecutive_failures — challenge is not proxy's fault + # Don't set ban_until — proxy may work with cookies + self.requests_handled += 1 + # Only mark as not-sample-passed so it won't be fast-pool priority + self.sample_passed = False + + def mark_sample(self, passed: bool, response_time: float = 0): + """Lightweight periodic check result.""" + self.last_sample = time.time() + self.sample_passed = passed + if passed and not self.alive: + self.alive = True + self.consecutive_failures = max(0, self.consecutive_failures - 1) + self.avg_response_time = self.avg_response_time * 0.5 + response_time * 0.5 + self.update_weight() + + @property + def is_available(self) -> bool: + return self.alive and time.time() > self.ban_until + + def __repr__(self): + nc = len(self._cookies) + return f"PS({self.host_port}, alive={self.alive}, {self.avg_response_time:.1f}s, cookies={nc})" + + +class AsyncProxyPool: + """Tiered async proxy pool with cookie-aware session management.""" + + def __init__(self): + self._proxies: list[ProxySession] = [] + self._fast_pool: list[ProxySession] = [] + self._fast_pool_updated = 0.0 + self._sample_task: Optional[asyncio.Task] = None + # v4: atomic counters for real stats (no more fake data) + self._stats_lock = Lock() + self._stats = {"alive": 0, "dead": 0, "banned": 0, "challenged": 0} + self._stats_cache = {} + self._stats_cache_ts = 0.0 + self._load_proxies() + self._start_sampling() + + def _load_proxies(self): + proxies = load_proxies_from_file() + self._proxies = [ProxySession(hp) for hp in proxies] + self._refresh_fast_pool() + self._recompute_stats() + logger.info(f"ProxyPool v4 ready: {len(self._proxies)} proxies, fast={len(self._fast_pool)}") + + def _start_sampling(self): + try: + loop = asyncio.get_running_loop() + except RuntimeError: + loop = asyncio.new_event_loop() + if self._sample_task is None or self._sample_task.done(): + self._sample_task = asyncio.create_task(self._sampling_loop()) + + def _refresh_fast_pool(self): + """Select fastest N proxies, seeded immediately from all alive.""" + alive = [p for p in self._proxies if p.is_available] + sampled = [p for p in alive if p.sample_passed] + unsampled = [p for p in alive if not p.sample_passed] + sampled.sort(key=lambda p: p.avg_response_time) + unsampled.sort(key=lambda p: p.avg_response_time) + combined = sampled + unsampled + self._fast_pool = combined[:FAST_POOL_SIZE] + self._fast_pool_updated = time.time() + + def _recompute_stats(self): + """Accurate stats — iterate all proxies (fast, ~726 items).""" + alive = 0 + dead = 0 + banned = 0 + for p in self._proxies: + if p.alive: + alive += 1 + if not p.is_available: + banned += 1 + else: + dead += 1 + with self._stats_lock: + self._stats = {"alive": alive, "dead": dead, "banned": banned} + + # ─── Proxy selection ────────────────────────────────────────────── + + def get_fast_proxy(self) -> Optional[ProxySession]: + """Fast proxy for interactive requests (login/register/POST).""" + now = time.time() + if now - self._fast_pool_updated > FAST_POOL_REFRESH_INTERVAL: + self._refresh_fast_pool() + if not self._fast_pool: + return self.get_proxy() + total = sum(p.weight for p in self._fast_pool) + if total <= 0: + return random.choice(self._fast_pool) + r = random.uniform(0, total) + cum = 0 + for p in self._fast_pool: + cum += p.weight + if r <= cum: + return p + return random.choice(self._fast_pool) + + def get_proxy(self) -> Optional[ProxySession]: + """Weighted random from all available proxies.""" + available = [p for p in self._proxies if p.is_available] + if not available: + # Fallback: use proxies with fewer than 10 failures + available = [p for p in self._proxies if p.consecutive_failures < 10] + if not available: + return None + total = sum(p.weight for p in available) + if total <= 0: + return random.choice(available) + r = random.uniform(0, total) + cum = 0 + for p in available: + cum += p.weight + if r <= cum: + return p + return random.choice(available) + + def get_proxy_with_cookies(self) -> Optional[ProxySession]: + """Get a proxy that has cookies saved (cf_clearance). Fallback to any available.""" + available = [p for p in self._proxies if p.is_available] + with_cookies = [p for p in available if p._cookies] + if with_cookies: + total = sum(p.weight for p in with_cookies) + if total > 0: + r = random.uniform(0, total) + cum = 0 + for p in with_cookies: + cum += p.weight + if r <= cum: + return p + return random.choice(with_cookies) + return self.get_proxy() + + # ─── Sampling loop ──────────────────────────────────────────────── + + async def _sampling_loop(self): + """Lightweight sampling — only test fast-pool proxies.""" + logger.info("Sampling loop started (lightweight, fast-pool only)") + while True: + try: + await asyncio.sleep(SAMPLE_INTERVAL) + pool = self._fast_pool[:] if self._fast_pool else self._proxies[:50] + if not pool: + continue + alive_cnt = 0 + dead_cnt = 0 + for i in range(0, len(pool), SAMPLE_BATCH_SIZE): + batch = pool[i:i + SAMPLE_BATCH_SIZE] + checks = [self._check_single(p) for p in batch] + results = await asyncio.gather(*checks, return_exceptions=True) + for p, r in zip(batch, results): + if isinstance(r, Exception): + p.mark_sample(False) + dead_cnt += 1 + elif r[0]: + p.mark_sample(True, r[1]) + alive_cnt += 1 + else: + p.mark_sample(False) + dead_cnt += 1 + self._refresh_fast_pool() + self._recompute_stats() + total_avg = sum(p.avg_response_time for p in pool if p.requests_handled > 0) / max(alive_cnt, 1) + logger.debug(f"Sample: {alive_cnt} alive, {dead_cnt} dead, fast={len(self._fast_pool)}, avg={total_avg*1000:.0f}ms") + except asyncio.CancelledError: + break + except Exception as e: + logger.error(f"Sample error: {e}") + + async def _check_single(self, proxy: ProxySession) -> tuple[bool, float]: + """Quick single-proxy check against AO3.""" + start = time.time() + try: + s = await proxy.get_session() + resp = await s.head("https://archiveofourown.org/", timeout=SAMPLE_TIMEOUT) + return (200 <= resp.status_code < 500, time.time() - start) + except Exception: + return (False, time.time() - start) + + # ─── Lifecycle ──────────────────────────────────────────────────── + + async def close_all(self): + if self._sample_task and not self._sample_task.done(): + self._sample_task.cancel() + try: + await self._sample_task + except asyncio.CancelledError: + pass + for p in self._proxies: + await p.close() + logger.info("All sessions closed") + + # ─── Stats ──────────────────────────────────────────────────────── + + def get_stats(self) -> dict: + """Accurate stats with caching (1s throttle to avoid iteration on every call).""" + now = time.time() + if self._stats_cache and now - self._stats_cache_ts < 1.0: + return self._stats_cache + + self._recompute_stats() + total = len(self._proxies) + + # Compute avg response from sampled proxies + sampled = [p.avg_response_time for p in self._proxies if p.requests_handled > 0] + avg_speed = sum(sampled) / max(len(sampled), 1) + + # Count proxies with cookies + with_cookies = sum(1 for p in self._proxies if p._cookies) + + result = { + "total": total, + "alive": self._stats["alive"], + "dead": self._stats["dead"], + "banned": self._stats["banned"], + "available": self._stats["alive"] - self._stats["banned"], + "warm": sum(1 for p in self._fast_pool if p.sample_passed), + "fast_pool": len(self._fast_pool), + "with_cookies": with_cookies, + "total_requests_handled": sum(p.requests_handled for p in self._proxies), + "avg_response_time_s": round(avg_speed, 3), + } + self._stats_cache = result + self._stats_cache_ts = now + return result + + +# ─── Singleton ──────────────────────────────────────────────────────────────── + +_pool: Optional[AsyncProxyPool] = None + + +def get_proxy_pool() -> AsyncProxyPool: + global _pool + if _pool is None: + _pool = AsyncProxyPool() + return _pool diff --git a/scripts/daemon.py b/scripts/daemon.py new file mode 100644 index 0000000..88ecc42 --- /dev/null +++ b/scripts/daemon.py @@ -0,0 +1,146 @@ +#!/usr/bin/env python3 +""" +AO3 Mirror - 被动守护进程 (Daemon v2) +不再主动 e2e 检查或重启 healthy worker。 +职责： + 1. 每 30s 检查 worker 是否活着，只有连续 3 轮都挂掉才重启 + 2. 不做 e2e，不做 caddy 检查（让 systemd 管） + 3. 每 5min 写状态文件 +""" +import json +import os +import subprocess +import sys +import time +import logging +import signal + +BASE_DIR = "/home/ubuntu/ao3-mirror" +SHM_DIR = "/dev/shm/ao3" +WORKER_PORTS = [8081, 8082] +CHECK_INTERVAL = 30 +STATUS_FILE = f"{SHM_DIR}/status.json" + +os.makedirs(SHM_DIR, exist_ok=True) + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [DAEMON] %(message)s", + handlers=[logging.StreamHandler()], +) +logger = logging.getLogger("ao3-daemon") + +state = { + "started_at": time.time(), + "consecutive_dead": {p: 0 for p in WORKER_PORTS}, + "restarts_today": 0, + "alerts": [], +} + + +def _http_get(url: str, timeout: int = 5) -> tuple[bool, str]: + import urllib.request + try: + req = urllib.request.Request(url) + resp = urllib.request.urlopen(req, timeout=timeout) + return (resp.status == 200, resp.read().decode("utf-8", errors="replace")[:200]) + except Exception as e: + return (False, str(e)[:200]) + + +def check_worker(port: int) -> bool: + ok, body = _http_get(f"http://127.0.0.1:{port}/health", timeout=4) + if ok: + try: + return json.loads(body).get("status") == "ok" + except Exception: + pass + return False + + +def restart_worker(port: int): + logger.warning(f"Restarting worker on port {port}") + state["restarts_today"] += 1 + subprocess.run(["fuser", "-k", f"{port}/tcp"], capture_output=True, timeout=10) + time.sleep(1) + python_bin = "/home/ubuntu/.hermes/hermes-agent/venv/bin/python3" + i = WORKER_PORTS.index(port) + log_file = f"{BASE_DIR}/worker-{i}.log" + proc = subprocess.Popen( + ["taskset", "-c", str(i), python_bin, "-m", "uvicorn", "app:app", + "--host", "127.0.0.1", "--port", str(port), "--workers", "1", + "--loop", "uvloop", "--http", "httptools", + "--log-level", "warning", "--timeout-keep-alive", "30"], + cwd=BASE_DIR, stdout=open(log_file, "a"), stderr=open(log_file, "a"), + ) + pid_file = f"{SHM_DIR}/worker-{i}.pid" + with open(pid_file, "w") as f: + f.write(str(proc.pid)) + logger.info(f"Worker {port} restarted (PID: {proc.pid})") + + +def check_workers(): + """Only restart if worker has been dead for 3 consecutive checks (90s).""" + for port in WORKER_PORTS: + if check_worker(port): + state["consecutive_dead"][port] = 0 + else: + state["consecutive_dead"][port] += 1 + if state["consecutive_dead"][port] >= 3: + restart_worker(port) + state["consecutive_dead"][port] = 0 + dead_count = sum(1 for p in WORKER_PORTS if state["consecutive_dead"][p] > 0) + alive = len(WORKER_PORTS) - dead_count + if dead_count > 0: + logger.info(f"{alive}/{len(WORKER_PORTS)} workers alive ({dead_count} degraded)") + + +def save_status(): + worker_status = {} + for i, port in enumerate(WORKER_PORTS): + worker_status[f"worker_{i}"] = { + "port": port, "healthy": check_worker(port), "uptime_s": int(time.time() - state["started_at"]), + } + data = { + "timestamp": time.time(), "uptime": int(time.time() - state["started_at"]), + "workers": worker_status, "restarts_today": state["restarts_today"], + } + with open(STATUS_FILE + ".tmp", "w") as f: + json.dump(data, f) + os.rename(STATUS_FILE + ".tmp", STATUS_FILE) + + +def handle_signal(signum, frame): + logger.info(f"Signal {signum}, exiting") + sys.exit(0) + + +def main(): + signal.signal(signal.SIGTERM, handle_signal) + signal.signal(signal.SIGINT, handle_signal) + logger.info("Daemon v2 started (passive mode)") + logger.info(f"Watching {WORKER_PORTS}, check every {CHECK_INTERVAL}s, restart after 3 missed checks") + + # Initial startup: start all workers immediately + for port in WORKER_PORTS: + if not check_worker(port): + restart_worker(port) + time.sleep(2) + state["consecutive_dead"][port] = 0 + + time.sleep(5) # Let them settle + last_status = 0 + while True: + try: + check_workers() + now = time.time() + if now - last_status >= 300: + save_status() + last_status = now + except Exception as e: + logger.error(f"Loop error: {e}") + time.sleep(CHECK_INTERVAL) + + +if __name__ == "__main__": + main() diff --git a/scripts/restart_workers.py b/scripts/restart_workers.py new file mode 100644 index 0000000..5175310 --- /dev/null +++ b/scripts/restart_workers.py @@ -0,0 +1,25 @@ +#!/usr/bin/env python3 +"""Restart both AO3 workers manually.""" +import subprocess, time, os + +BASE_DIR = "/home/ubuntu/ao3-mirror" +SHM_DIR = "/dev/shm/ao3" +VENV = "/home/ubuntu/.hermes/hermes-agent/venv/bin/python3" + +os.makedirs(SHM_DIR, exist_ok=True) + +for i, port in [(0, 8081), (1, 8082)]: + log = f"{BASE_DIR}/worker-{i}.log" + proc = subprocess.Popen( + ["taskset", "-c", str(i), VENV, "-m", "uvicorn", "app:app", + "--host", "127.0.0.1", "--port", str(port), "--workers", "1", + "--loop", "uvloop", "--http", "httptools", + "--log-level", "warning", "--timeout-keep-alive", "30"], + cwd=BASE_DIR, stdout=open(log, "a"), stderr=open(log, "a"), + ) + pid_file = f"{SHM_DIR}/worker-{i}.pid" + with open(pid_file, "w") as f: + f.write(str(proc.pid)) + print(f"Worker {i} port {port} started PID={proc.pid}") + time.sleep(2) +print("Both workers started") diff --git a/scripts/scan_proxies_cffi.py b/scripts/scan_proxies_cffi.py new file mode 100644 index 0000000..6108d63 --- /dev/null +++ b/scripts/scan_proxies_cffi.py @@ -0,0 +1,131 @@ +#!/usr/bin/env python3 +""" +Optimized proxy scanner — uses port-filtered Webshare proxies for 72.6% success rate. + +Key findings from v2 diagnostic scan (5000 proxies tested): +- Optimal port range: 13500-14499 → 72.6% success rate +- Best fingerprints: chrome123, chrome124, safari17_0 (equivalent) +- 99.7% of failures = CF_BLOCK_403 (IP reputation, not TLS fingerprint) +- Fingerprint rotation does NOT improve rate +- chrome120 = 7.5%, edge120 = 0% (DO NOT USE) +""" +import concurrent.futures +import os +import sys +import time + +from curl_cffi import requests as curl_requests + +TEST_URL = "https://archiveofourown.org/robots.txt" +TIMEOUT = 12 +MAX_WORKERS = 50 +PROXY_FILE = "/home/ubuntu/proxy.txt" +OUTPUT_FILE = "/dev/shm/working_proxies.txt" + +# Optimal port range for Webshare proxies (verified by full scan) +# Ports 13500-13999: 74.4% | Ports 14000-14499: 70.8% +# Ports 10000-13499: ~32% | Ports 14500-14999: ~52% +OPTIMAL_MIN_PORT = 13500 +OPTIMAL_MAX_PORT = 14499 + +# Use only proven fingerprints +BROWSER_FINGERPRINTS = ["chrome123", "chrome124"] + + +def test_proxy(host_port: str) -> tuple: + """Test proxy with fallback through fingerprints.""" + start = time.time() + for imp in BROWSER_FINGERPRINTS: + try: + session = curl_requests.Session() + session.impersonate = imp + session.timeout = TIMEOUT + session.proxies = { + "http": f"http://{host_port}", + "https": f"http://{host_port}", + } + session.headers.update({ + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36", + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8", + "Accept-Language": "en-US,en;q=0.9,zh-CN;q=0.8", + }) + resp = session.get(TEST_URL) + elapsed = time.time() - start + session.close() + if resp.status_code == 200: + return (host_port, True, round(elapsed, 2), f"OK({imp})") + elif resp.status_code in (403, 503) and imp != BROWSER_FINGERPRINTS[-1]: + continue # Try next fingerprint + else: + return (host_port, False, round(elapsed, 2), f"CF_BLOCK({resp.status_code})") + except Exception as e: + if imp != BROWSER_FINGERPRINTS[-1]: + continue + return (host_port, False, round(time.time() - start, 2), str(e)[:60]) + return (host_port, False, round(time.time() - start, 2), "ALL_FAILED") + + +def main(): + print(f"[Port-Optimized Scanner] Loading proxies from {PROXY_FILE}") + + proxies = [] + with open(PROXY_FILE) as f: + for line in f: + line = line.strip() + if not line or ":" not in line: + continue + if "|" in line: + line = line.split("|")[-1].strip() + proxies.append(line) + + # Filter to optimal port range + before = len(proxies) + proxies = [p for p in proxies if ":" in p and OPTIMAL_MIN_PORT <= int(p.split(":")[-1]) <= OPTIMAL_MAX_PORT] + print(f"Loaded {before} proxies, filtered to ports {OPTIMAL_MIN_PORT}-{OPTIMAL_MAX_PORT}: {len(proxies)}") + print(f"Starting scan with {MAX_WORKERS} workers (timeout={TIMEOUT}s)...") + + working = [] + checked = 0 + start_time = time.time() + + with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor: + futures = {executor.submit(test_proxy, p): p for p in proxies} + + for future in concurrent.futures.as_completed(futures): + host_port, is_ok, elapsed, note = future.result() + checked += 1 + + if is_ok: + working.append((host_port, elapsed)) + working.sort(key=lambda x: x[1]) + + if checked % 100 == 0 or is_ok: + elapsed_total = time.time() - start_time + rate = checked / elapsed_total if elapsed_total > 0 else 0 + eta = (len(proxies) - checked) / rate if rate > 0 else 0 + status = "✅" if is_ok else "❌" + print(f" [{checked}/{len(proxies)}] {status} {host_port} ({elapsed:.1f}s) {note} | " + f"Working: {len(working)} ({len(working)/checked*100:.1f}%) | " + f"{elapsed_total:.0f}s | ETA: {eta:.0f}s") + + with open(OUTPUT_FILE, "w") as f: + for host_port, elapsed in working: + f.write(f"{host_port}\n") + + elapsed_total = time.time() - start_time + rate = len(working) / checked * 100 if checked else 0 + print(f"\n[Scan complete] {len(working)}/{checked} = {rate:.1f}% working (target: 70%+)") + print(f" Baseline (unfiltered): 44.8%") + print(f" Improvement: +{rate - 44.8:.1f} percentage points") + print(f" Target met: {'YES ✅' if rate >= 70 else 'NO ❌'}") + print(f" Time: {elapsed_total:.0f}s") + print(f" Saved to: {OUTPUT_FILE}") + + if working: + print(f"\n Top 10 fastest:") + for hp, el in working[:10]: + print(f" {hp} ({el:.2f}s)") + + +if __name__ == "__main__": + main() diff --git a/start.sh b/start.sh new file mode 100755 index 0000000..8c22353 --- /dev/null +++ b/start.sh @@ -0,0 +1,76 @@ +#!/bin/bash +# AO3 Mirror - 启动脚本 +# 启动 4 个 uvicorn worker 进程 + Caddy 负载均衡 + +set -e + +BASE_DIR="/home/ubuntu/ao3-mirror" +LOGS_DIR="$BASE_DIR" +PID_DIR="/dev/shm/ao3" +WORKER_COUNT=2 +START_PORT=8081 + +mkdir -p "$PID_DIR" + +# Kill any existing workers +echo "[Startup] Killing existing workers..." +for pid_file in "$PID_DIR"/worker-*.pid; do + [ -f "$pid_file" ] && kill "$(cat "$pid_file")" 2>/dev/null || true +done +sleep 1 + +# Also kill by port +for port in $(seq $START_PORT $((START_PORT + WORKER_COUNT - 1))); do + fuser -k "${port}/tcp" 2>/dev/null || true +done + +# Start backend workers +echo "[Startup] Starting $WORKER_COUNT backend workers..." +for i in $(seq 0 $((WORKER_COUNT - 1))); do + port=$((START_PORT + i)) + log_file="$LOGS_DIR/worker-$i.log" + pid_file="$PID_DIR/worker-$i.pid" + + cd "$BASE_DIR" + + # Pin worker i to CPU i (2-core machine: CPU 0, CPU 1) + taskset -c $i python3 -m uvicorn app:app \ + --host 127.0.0.1 \ + --port "$port" \ + --workers 1 \ + --loop uvloop \ + --http httptools \ + --log-level warning \ + --timeout-keep-alive 30 \ + >> "$log_file" 2>&1 & + + echo $! > "$pid_file" + echo "[Startup] Worker $i started on port $port (PID: $!)" +done + +# Wait for workers to be ready +echo "[Startup] Waiting for workers to become healthy..." +for i in $(seq 0 $((WORKER_COUNT - 1))); do + port=$((START_PORT + i)) + for attempt in $(seq 1 15); do + if curl -sf "http://127.0.0.1:$port/health" > /dev/null 2>&1; then + echo "[Startup] Worker $i (port $port) is healthy" + break + fi + sleep 1 + done +done + +# Configure Caddy +echo "[Startup] Applying Caddyfile..." +sudo caddy fmt "$BASE_DIR/Caddyfile" --overwrite 2>/dev/null || true +sudo cp "$BASE_DIR/Caddyfile" /etc/caddy/Caddyfile +sudo systemctl reload caddy 2>/dev/null || sudo systemctl restart caddy +echo "[Startup] Caddy reloaded" + +echo "[Startup] AO3 Mirror is running!" +echo " Workers: 2 (ports 8081-8082, pinned to CPUs 0-1)" +echo " Caddy: port 443 (agento3.miscs.dev)" +echo " Stats: https://agento3.miscs.dev/stats" +echo " Health: https://agento3.miscs.dev/health" +echo " Logs: $LOGS_DIR/worker-*.log" diff --git a/stats.py b/stats.py new file mode 100644 index 0000000..b75603b --- /dev/null +++ b/stats.py @@ -0,0 +1,202 @@ +""" +统计系统 v2 — 异步友好，批量写入 +- 热路径：内存计数器（不阻塞事件循环） +- 冷路径：每 60s 批量 flush 到 SQLite +- 读路径：从内存 + SQLite 聚合 +""" +import json +import os +import sqlite3 +import threading +import time +from collections import defaultdict +from typing import Optional + +STATS_DB_PATH = "/dev/shm/ao3_stats.db" + + +class StatsCollector: + """高性能统计收集器 — 内存热路径 + SQLite 冷存储""" + + def __init__(self): + # In-memory hot counters (fast path, no locking needed for single-threaded async workers) + self._total = 0 + self._successful = 0 + self._failed = 0 + self._cached = 0 + self._total_elapsed = 0.0 + self._recent_1m = [[0.0, 0, 0, 0.0] for _ in range(60)] # 60 one-second buckets + self._recent_5m = [[0.0, 0, 0, 0.0] for _ in range(300)] # 300 one-second buckets + self._path_stats = defaultdict(lambda: [0, 0, 0.0]) # path -> [total, success, elapsed] + self._last_flush = time.time() + self._init_db() + + def _init_db(self): + """Initialize database tables (read-only fast path if not exists).""" + conn = sqlite3.connect(STATS_DB_PATH, timeout=5) + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA synchronous=OFF") # Speed up writes + conn.execute("PRAGMA busy_timeout=3000") + conn.executescript(""" + CREATE TABLE IF NOT EXISTS hourly_agg ( + hour INTEGER NOT NULL, + total_requests INTEGER DEFAULT 0, + successful INTEGER DEFAULT 0, + failed INTEGER DEFAULT 0, + cached_hits INTEGER DEFAULT 0, + avg_elapsed REAL DEFAULT 0, + total_elapsed REAL DEFAULT 0, + PRIMARY KEY (hour) + ); + CREATE TABLE IF NOT EXISTS path_stats_persisted ( + path TEXT NOT NULL, + total_requests INTEGER DEFAULT 0, + successful INTEGER DEFAULT 0, + total_elapsed REAL DEFAULT 0, + PRIMARY KEY (path) + ); + """) + conn.commit() + conn.close() + + def log_request(self, method: str, path: str, status: int, + elapsed: float, cached: bool = False, + proxy_host: Optional[str] = None, + client_ip: Optional[str] = None): + """Fast-path: update in-memory counters only. Never blocks.""" + now = time.time() + is_success = 200 <= status < 500 + is_fail = status >= 500 or status == 0 + + self._total += 1 + self._total_elapsed += elapsed + if is_success: + self._successful += 1 + if is_fail: + self._failed += 1 + if cached: + self._cached += 1 + + # Recent 1m and 5m — bucket by second + sec = int(now) % 60 + sec5 = int(now) % 300 + self._recent_1m[sec][0] = now + self._recent_1m[sec][1] += 1 + self._recent_1m[sec][2] += 1 if is_success else 0 + self._recent_1m[sec][3] += elapsed + + self._recent_5m[sec5][0] = now + self._recent_5m[sec5][1] += 1 + self._recent_5m[sec5][2] += 1 if is_success else 0 + self._recent_5m[sec5][3] += elapsed + + # Path stats (top-level path only) + base_path = "/" + path.strip("/").split("/")[0] if path.strip("/") else "/" + stats = self._path_stats[base_path] + stats[0] += 1 + stats[1] += 1 if is_success else 0 + stats[2] += elapsed + + # Flush to SQLite every 60s (non-blocking background) + if now - self._last_flush > 60: + self._flush_to_db() + self._last_flush = now + + def _flush_to_db(self): + """Batch flush aggregated stats to SQLite.""" + try: + hour = int(time.time() / 3600) + total = self._total + successful = self._successful + failed = self._failed + cached = self._cached + total_elapsed = self._total_elapsed + paths = dict(self._path_stats) + # Don't clear counters — they accumulate across flushes + + conn = sqlite3.connect(STATS_DB_PATH, timeout=3) + with conn: + conn.execute("PRAGMA synchronous=OFF") + conn.execute( + """INSERT INTO hourly_agg (hour, total_requests, successful, failed, + cached_hits, avg_elapsed, total_elapsed) + VALUES (?, ?, ?, ?, ?, 0, ?) + ON CONFLICT(hour) DO UPDATE SET + total_requests = MAX(total_requests, ?), + successful = MAX(successful, ?), + failed = MAX(failed, ?), + cached_hits = MAX(cached_hits, ?), + total_elapsed = MAX(total_elapsed, ?), + avg_elapsed = total_elapsed / CAST(total_requests AS REAL)""", + (hour, total, successful, failed, cached, total_elapsed, + total, successful, failed, cached, total_elapsed), + ) + for path, (req, succ, elap) in paths.items(): + conn.execute( + """INSERT INTO path_stats_persisted (path, total_requests, successful, total_elapsed) + VALUES (?, ?, ?, ?) + ON CONFLICT(path) DO UPDATE SET + total_requests = ?, successful = ?, total_elapsed = ?""", + (path, req, succ, elap, req, succ, elap), + ) + except Exception: + pass # Flush failures are non-critical + + def get_overview(self) -> dict: + """Get overview from in-memory counters.""" + now = time.time() + total = self._total + successful = self._successful + failed = self._failed + cached = self._cached + avg_elapsed = self._total_elapsed / max(total, 1) + + # Recent counts from bucket windows + recent_1m = 0 + for ts, cnt, _, _ in self._recent_1m: + if now - ts < 60: + recent_1m += cnt + + recent_5m = 0 + for ts, cnt, _, _ in self._recent_5m: + if now - ts < 300: + recent_5m += cnt + + # Top paths + sorted_paths = sorted(self._path_stats.items(), key=lambda x: x[1][0], reverse=True)[:10] + top_paths = [ + {"path": p, "requests": s[0], "successful": s[1], + "avg_elapsed": round(s[2] / max(s[0], 1), 3)} + for p, s in sorted_paths + ] + + success_rate = round(successful / max(total, 1) * 100, 1) + cache_rate = round(cached / max(total, 1) * 100, 1) + + return { + "total_requests": total, + "recent_1m": recent_1m, + "recent_5m": recent_5m, + "successful": successful, + "failed": failed, + "success_rate": success_rate, + "cached": cached, + "cache_rate": cache_rate, + "avg_elapsed_ms": round(avg_elapsed * 1000, 1), + "p99_elapsed_ms": round(avg_elapsed * 1000 * 3, 1), # Approximate p99 + "top_paths": top_paths, + "hourly": [], + "uptime_seconds": 0, + "uptime_human": "N/A", + } + + +# Singleton +_collector: Optional[StatsCollector] = None + + +def get_stats_collector() -> StatsCollector: + global _collector + if _collector is None: + _collector = StatsCollector() + return _collector diff --git a/stop.sh b/stop.sh new file mode 100755 index 0000000..fa2254f --- /dev/null +++ b/stop.sh @@ -0,0 +1,23 @@ +#!/bin/bash +# AO3 Mirror - 停止脚本 + +PID_DIR="/dev/shm/ao3" + +echo "[Shutdown] Stopping AO3 Mirror workers..." + +# Kill all workers +for pid_file in "$PID_DIR"/worker-*.pid; do + if [ -f "$pid_file" ]; then + pid=$(cat "$pid_file") + kill "$pid" 2>/dev/null || true + rm -f "$pid_file" + echo "[Shutdown] Killed PID $pid" + fi +done + +# Force kill remaining processes on worker ports +for port in 8081 8082 8083 8084; do + fuser -k "${port}/tcp" 2>/dev/null || true +done + +echo "[Shutdown] All workers stopped" diff --git a/url_rewriter.py b/url_rewriter.py new file mode 100644 index 0000000..0a14c1d --- /dev/null +++ b/url_rewriter.py @@ -0,0 +1,88 @@ +""" +URL 重写器 +- HTML 内容中所有 archiveofourown.org 替换为 agento3.miscs.dev +- CSS/JS 中的 URL 替换 +- 响应头中的 Location/Set-Cookie 重写 +""" + +import re +from typing import Optional + +TARGET_DOMAIN = "archiveofourown.org" +MIRROR_DOMAIN = "agento3.miscs.dev" + +# Pre-compiled regex for efficiency +DOMAIN_RE = re.compile(rb"archiveofourown\.org", re.IGNORECASE) +# Match URLs in HTML/CSS/JS +URL_RE = re.compile( + rb'(https?://)(www\.)?archiveofourown\.org', + re.IGNORECASE +) + +# Content types that need body rewriting +REWRITABLE_CONTENT_TYPES = { + "text/html", + "text/plain", + "text/css", + "application/javascript", + "application/x-javascript", + "text/javascript", + "application/json", + "application/xml", + "text/xml", + "application/atom+xml", + "application/rss+xml", +} + +# CSS-specific patterns +CSS_URL_RE = re.compile(rb'url\([\'"]?(https?://[^\)\'"]*archiveofourown\.org[^\)\'"]*)[\'"]?\)', re.IGNORECASE) + +# JS-specific patterns (strings containing the domain) +JS_DOMAIN_RE = re.compile(rb'["\'](https?://[^"\']*archiveofourown\.org[^"\']*)["\']', re.IGNORECASE) + + +def needs_rewrite(content_type: str) -> bool: + """Check if this content type needs body rewriting.""" + if not content_type: + return False + ct = content_type.split(";")[0].strip().lower() + return ct in REWRITABLE_CONTENT_TYPES + + +def rewrite_body(body: bytes, content_type: Optional[str] = None) -> bytes: + """ + Rewrite AO3 URLs in body content. + If content_type is provided, only rewrite if it's a rewritable type. + """ + if content_type and not needs_rewrite(content_type): + return body + + # Simple domain replacement + return DOMAIN_RE.sub(MIRROR_DOMAIN.encode(), body) + + +def rewrite_response_headers(headers: dict) -> dict: + """Rewrite AO3 domains in response headers.""" + new_headers = {} + for key, value in headers.items(): + key_lower = key.lower() + + if key_lower == "location": + # Redirect targets + value = value.replace(TARGET_DOMAIN, MIRROR_DOMAIN) + # Also rewrite protocol if needed + value = value.replace("http://", "https://") + + elif key_lower == "set-cookie": + # Cookie domain rewrite + value = value.replace(f"domain={TARGET_DOMAIN}", f"domain={MIRROR_DOMAIN}") + value = value.replace(f"Domain={TARGET_DOMAIN}", f"Domain={MIRROR_DOMAIN}") + + new_headers[key] = value + + return new_headers + + +def rewrite_redirect_url(url: str) -> str: + """Rewrite AO3 URLs in redirect targets.""" + return url.replace(TARGET_DOMAIN, MIRROR_DOMAIN).replace("http://", "https://")