调用受Azure WAF保护的448端口API：限流与代理处理难题

我正在开发一个调用448端口托管API的服务，该API受Azure WAF V2保护，限流规则为：仅允许150次连续请求，之后会封禁IP 5分钟。

已尝试的规避方案均失败：

• 住宅代理：测试Smart Proxy等服务，仅支持443/80端口，无法适配448端口
• Tor网络：Tor节点IP会被API直接封禁
• Azure应用代理路由：部署Azure应用尝试路由请求，但出站请求仍显示服务器IP，未切换为代理IP

重要说明：该API并非我方所有，仅用于数据爬取流程

from fastapi import FastAPI, HTTPException, Request  
from fastapi.responses import JSONResponse
import logging
import uvicorn
import random
import asyncio
from fake_useragent import UserAgent
from curl_cffi import requests as cffi_requests  # pip install curl-cffi

# Logging configuration
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

ua = UserAgent()
app = FastAPI()

def get_random_headers():
    """Generates dynamic headers with a random User-Agent"""
    return {
        "Authority": "example.domain.com:448",
        "Accept": "application/json, text/plain, */*",
        "Accept-Language": "en-US,en;q=0.9",
        "Origin": "https://example.domain.com",
        "Referer": "https://example.domain.com/",
        "User-Agent": ua.random,
        "Priority": "u=1, i",
        "Alt-Used": "example.domain.com",
        "Upgrade-Insecure-Requests": "1"
    }

async def make_request(url, params=None):
    """Performs the request with headers and mimics browser fingerprinting"""
    await asyncio.sleep(random.uniform(1, 4))  # Add random delay
    try:
        async with cffi_requests.AsyncSession() as s:
            response = await s.get(
                url,
                params=params,
                headers={
                    **get_random_headers(),
                    "Sec-Ch-Ua": '"Not_A Brand";v="8", "Chromium";v="120"',
                    "Sec-Ch-Ua-Platform": '"Windows"',
                    "Sec-Fetch-Dest": "empty",
                    "Sec-Fetch-Mode": "cors",
                    "Sec-Fetch-Site": "same-site"
                },
                impersonate="chrome120"
            )
            
            if response.status_code != 200:
                logging.error(f"Error {response.status_code}: {response.text}")
                raise HTTPException(status_code=response.status_code, detail=response.text)
            
            return JSONResponse(content=response.json())
    except HTTPException as http_exc:
        raise http_exc
    except Exception as e:
        logging.error(f"Critical error: {str(e)}")
        raise HTTPException(status_code=500, detail="Internal server error")

# Endpoints
@app.get("/api/v2/Query")
async def query_api(request: Request, number: str, only_active: bool = False, page: int = 1):
    params = {
        "number": number,
        "onlyActive": str(only_active).lower(),
        "page": page
    }
    return await make_request("https://example.domain.com:448/api/v2/Query", params)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8080)

1. 是否有开发者遇到过类似的API限流与非标准端口代理适配问题？
2. 针对非标准端口（如448）的服务请求，需要IP轮换或代理时，有哪些可行策略？
3. 在Azure或其他云环境中，如何配置才能让出站请求使用代理IP而非服务器自身IP？

补充信息：

• API受Azure WAF V2保护，连续请求上限150次，封禁时长5分钟
• API归属第三方，仅用于数据爬取

一、非标准端口代理适配策略 #

1. 选择支持自定义端口的代理服务商 #

高端住宅/数据中心代理服务商（如BrightData、Oxylabs）普遍支持自定义端口转发，可直接配置目标API的448端口。配置方式通常为：在请求时指定代理地址+目标端口映射，或通过服务商提供的转发端点直接对接目标API的448端口。

2. 自建中转代理服务器 #

若不想依赖第三方，可在云服务器（AWS EC2、Azure VM等）上搭建Nginx/Squid中转代理，实现端口转发：

• 在中转服务器配置Nginx反向代理，将本地端口（如8080）转发到目标API的example.domain.com:448
• FastAPI服务请求中转服务器的8080端口，由中转服务器代为发起请求，目标API将识别中转服务器的IP
• 多部署几台中转服务器，手动或自动轮换IP，每台请求达到140次左右时切换，避免触发封禁阈值

二、Azure环境下出站IP切换配置 #

1. Azure Application Gateway + 代理池 #

• 配置Azure Application Gateway作为入口，转发请求到FastAPI服务
• 在Application Gateway的Outbound rules中设置支持自定义端口的代理服务器地址，所有出站请求将通过代理发送，目标API看到的是代理IP

2. Azure Functions + 代理环境变量 #

• 将请求逻辑迁移到Azure Functions，在函数配置中添加HTTP_PROXY和HTTPS_PROXY环境变量，指向支持448端口的代理服务器
• 若使用消费计划，函数实例IP会动态变化，可辅助实现IP轮换（适合低频率请求场景）

三、限流规避补充策略 #

1. 优化请求调度逻辑 #

在现有代码基础上，加入IP轮换与请求计数逻辑：

# 示例：代理IP池，格式为 "http://username:password@proxy-ip:proxy-port"
PROXY_POOL = [
    "http://proxy1:8080",
    "http://proxy2:8080",
    # 更多代理节点
]
current_proxy_index = 0
request_count_per_proxy = 0

async def make_request(url, params=None):
    global current_proxy_index, request_count_per_proxy
    await asyncio.sleep(random.uniform(1, 4))
    
    # 每140次请求切换代理，提前规避封禁阈值
    if request_count_per_proxy >= 140:
        current_proxy_index = (current_proxy_index + 1) % len(PROXY_POOL)
        request_count_per_proxy = 0
    
    proxy = PROXY_POOL[current_proxy_index]
    try:
        async with cffi_requests.AsyncSession(proxies={"https": proxy}) as s:
            response = await s.get(
                url,
                params=params,
                headers={
                    **get_random_headers(),
                    "Sec-Ch-Ua": '"Not_A Brand";v="8", "Chromium";v="120"',
                    "Sec-Ch-Ua-Platform": '"Windows"',
                    "Sec-Fetch-Dest": "empty",
                    "Sec-Fetch-Mode": "cors",
                    "Sec-Fetch-Site": "same-site"
                },
                impersonate="chrome120"
            )
            
            if response.status_code != 200:
                logging.error(f"Error {response.status_code}: {response.text}")
                # 请求失败切换代理并重试
                current_proxy_index = (current_proxy_index + 1) % len(PROXY_POOL)
                request_count_per_proxy = 0
                return await make_request(url, params)
            
            request_count_per_proxy += 1
            return JSONResponse(content=response.json())
    except Exception as e:
        logging.error(f"Critical error: {str(e)}")
        # 异常时切换代理并重试
        current_proxy_index = (current_proxy_index + 1) % len(PROXY_POOL)
        request_count_per_proxy = 0
        return await make_request(url, params)

2. 模拟真实浏览器行为 #

• 随机生成Sec-Ch-Ua-Mobile、Sec-Ch-Ua-Version等头部字段值，避免固定指纹
• 调整请求间隔的随机性，加入10%左右的概率插入30-60秒的长延迟，模拟用户停顿
• 偶尔请求API的其他公开端点，模拟真实用户的访问路径

内容的提问来源于stack exchange，提问作者Alejandro Echeverria