Adapt Cloudflare Web Security

.gitmodules

@@ -0,0 +1,4 @@
+[submodule "cloudflare_security_utils"]
+	path = cloudflare_security_utils
+	url = https://github.com/DGP-Studio/cloudflare-api-security.git
+	branch = main

config.py

@@ -8,8 +8,8 @@
 VALID_PROJECT_KEYS = ["snap-hutao", "snap-hutao-deployment"]
 
 IMAGE_NAME = os.getenv("IMAGE_NAME", "generic-api")
-SERVER_TYPE = os.getenv("SERVER_TYPE", "[Unknown Server Type]")
-IS_DEBUG = True if "alpha" in IMAGE_NAME.lower() or "dev" in IMAGE_NAME.lower() else False
+SERVER_TYPE = os.getenv("SERVER_TYPE", "unknown").lower()
+IS_DEBUG = True if "alpha" in SERVER_TYPE.lower() or "dev" in SERVER_TYPE.lower() else False
 IS_DEV = True if os.getenv("IS_DEV", "False").lower() == "true" or SERVER_TYPE in ["dev"] else False
 if IS_DEV:
     BUILD_NUMBER = "DEV"

main.py

@@ -9,7 +9,8 @@
 from datetime import datetime
 from contextlib import asynccontextmanager
 from routers import (enka_network, metadata, patch_next, static, net, wallpaper, strategy, crowdin, system_email,
-                     client_feature, mgnt)
+                     client_feature)
+from cloudflare_security_utils import mgnt
 from base_logger import get_logger
 from config import (MAIN_SERVER_DESCRIPTION, TOS_URL, CONTACT_INFO, LICENSE_INFO, VALID_PROJECT_KEYS,
                     IS_DEBUG, IS_DEV, SERVER_TYPE, REDIS_HOST, SENTRY_URL, BUILD_NUMBER, CURRENT_COMMIT_HASH)

routers/client_feature.py

@@ -1,6 +1,7 @@
-from fastapi import APIRouter, Request
+from fastapi import APIRouter, Request, Depends
 from fastapi.responses import RedirectResponse
 from redis import asyncio as aioredis
+from cloudflare_security_utils.safety import enhanced_safety_check
 
 
 china_router = APIRouter(tags=["Client Feature"], prefix="/client")
@@ -9,16 +10,14 @@
 
 
 @china_router.get("/{file_path:path}")
-async def china_client_feature_request_handler(request: Request, file_path: str) -> RedirectResponse:
-    """
-    Handle requests to client feature metadata files.
+async def china_client_feature_request_handler(
+    request: Request,
+    file_path: str,
+    safety_check: bool | RedirectResponse = Depends(enhanced_safety_check)
+) -> RedirectResponse:
+    if isinstance(safety_check, RedirectResponse):
+        return safety_check
 
-    :param request: Request object from FastAPI
-
-    :param file_path: Path to the metadata file
-
-    :return: HTTP 301 redirect to the file based on censorship status of the file
-    """
     redis_client = aioredis.Redis.from_pool(request.app.state.redis)
 
     host_for_normal_files = await redis_client.get("url:china:client-feature")
@@ -28,16 +27,14 @@ async def china_client_feature_request_handler(request: Request, file_path: str)
 
 
 @global_router.get("/{file_path:path}")
-async def global_client_feature_request_handler(request: Request, file_path: str) -> RedirectResponse:
-    """
-    Handle requests to client feature metadata files.
-
-    :param request: Request object from FastAPI
+async def global_client_feature_request_handler(
+    request: Request,
+    file_path: str,
+    safety_check: bool | RedirectResponse = Depends(enhanced_safety_check)
+) -> RedirectResponse:
+    if isinstance(safety_check, RedirectResponse):
+        return safety_check
 
-    :param file_path: Path to the metadata file
-
-    :return: HTTP 301 redirect to the file based on censorship status of the file
-    """
     redis_client = aioredis.Redis.from_pool(request.app.state.redis)
 
     host_for_normal_files = await redis_client.get("url:global:client-feature")
@@ -47,16 +44,14 @@ async def global_client_feature_request_handler(request: Request, file_path: str
 
 
 @fujian_router.get("/{file_path:path}")
-async def fujian_client_feature_request_handler(request: Request, file_path: str) -> RedirectResponse:
-    """
-    Handle requests to client feature metadata files.
-
-    :param request: Request object from FastAPI
-
-    :param file_path: Path to the metadata file
+async def fujian_client_feature_request_handler(
+    request: Request,
+    file_path: str,
+    safety_check: bool | RedirectResponse = Depends(enhanced_safety_check)
+) -> RedirectResponse:
+    if isinstance(safety_check, RedirectResponse):
+        return safety_check
 
-    :return: HTTP 301 redirect to the file based on censorship status of the file
-    """
     redis_client = aioredis.Redis.from_pool(request.app.state.redis)
 
     host_for_normal_files = await redis_client.get("url:fujian:client-feature")

routers/enka_network.py

@@ -1,7 +1,7 @@
 from fastapi import APIRouter, Depends, Request
 from fastapi.responses import RedirectResponse
 from redis import asyncio as aioredis
-from utils.dgp_utils import validate_client_is_updated
+from cloudflare_security_utils.safety import validate_client_is_updated
 
 
 china_router = APIRouter(tags=["Enka Network"], prefix="/enka")

routers/metadata.py

@@ -2,14 +2,15 @@
 from fastapi.responses import RedirectResponse
 from redis import asyncio as aioredis
 from mysql_app.schemas import StandardResponse
-from utils.dgp_utils import validate_client_is_updated
+from cloudflare_security_utils.safety import validate_client_is_updated
 from base_logger import get_logger
 import httpx
 import os
 
 china_router = APIRouter(tags=["Hutao Metadata"], prefix="/metadata")
 global_router = APIRouter(tags=["Hutao Metadata"], prefix="/metadata")
 fujian_router = APIRouter(tags=["Hutao Metadata"], prefix="/metadata")
+logger = get_logger(__name__)
 
 
 async def fetch_metadata_repo_file_list(redis_client: aioredis.Redis) -> None:

utils/dgp_utils.py

@@ -1,9 +1,6 @@
 import json
 import os
 import httpx
-from fastapi import HTTPException, status, Header, Request
-from redis import asyncio as aioredis
-from typing import Annotated
 from base_logger import get_logger
 from config import github_headers, IS_DEBUG
 
@@ -14,9 +11,6 @@
     WHITE_LIST_REPOSITORIES = {}
     logger.error("Failed to load WHITE_LIST_REPOSITORIES from environment variable.")
     logger.info(os.environ.get("WHITE_LIST_REPOSITORIES"))
-BYPASS_CLIENT_VERIFICATION = os.environ.get("BYPASS_CLIENT_VERIFICATION", "False").lower() == "true"
-if BYPASS_CLIENT_VERIFICATION:
-    logger.warning("Client verification is bypassed in this server.")
 
 # Helper: HTTP GET with retry
 async def fetch_with_retry(url, max_retries=3):
@@ -102,34 +96,3 @@ async def update_recent_versions(redis_client) -> list[str]:
     return new_user_agents
 
 
-async def validate_client_is_updated(request: Request, user_agent: Annotated[str, Header()]) -> bool:
-    requested_hostname = request.headers.get("Host")
-    if "snapgenshin.cn" in requested_hostname:
-        return True
-    redis_client = aioredis.Redis.from_pool(request.app.state.redis)
-    if BYPASS_CLIENT_VERIFICATION:
-        logger.debug("Client verification is bypassed.")
-        return True
-    logger.info(f"Received request from user agent: {user_agent}")
-    if user_agent.startswith("Snap Hutao/2025"):
-        logger.info("Client is Snap Hutao Alpha, allowed.")
-        return True
-    if user_agent.startswith("PaimonsNotebook/"):
-        logger.info("Client is Paimon's Notebook, allowed.")
-        return True
-    if user_agent.startswith("Reqable/"):
-        logger.info("Client is Reqable, allowed.")
-        return True
-
-    allowed_user_agents = await redis_client.get("allowed_user_agents")
-    if allowed_user_agents:
-        allowed_user_agents = json.loads(allowed_user_agents)
-    else:
-        # redis data is expired
-        logger.info("Updating allowed user agents from GitHub")
-        allowed_user_agents = await update_recent_versions(redis_client)
-
-    if user_agent not in allowed_user_agents:
-        logger.info(f"Client is outdated: {user_agent}, not in the allowed list: {allowed_user_agents}")
-        raise HTTPException(status_code=status.HTTP_418_IM_A_TEAPOT, detail="Client is outdated.")
-    return True