247LIKE-BLOG
large picture of the cover

高性能 BGP & Geo 资产溯源 API (双栈终极版)

🌐 高性能 BGP & Geo 资产溯源 API (双栈终极版)

📖 项目背景与核心优势

本项目是一个基于 Python (FastAPI) 构建的超高性能网络资产测绘微服务。它通过自动拉取并解析 IPinfo 官方静态产权库 (Country + ASN 融合库)，将全球近 150 万条分配数据加载到纯内存中进行极速检索。

🌟 核心解决的行业痛点：

  1. 基础设施 IP 隐身问题：传统 BGP 路由表抓不到骨干网核心路由器 IP（例如联通国际的内网互联段），本项目基于物理产权分配表，实现了 100% 物理资产无死角覆盖。
  2. 地理 CIDR 碎片化缝合：由于跨国 ASN 的 IP 会被不同国家切割成极小的碎片（如 /32），系统内置了严格的数学无损路由聚合算法 (CIDR Collapse)，反查 ASN 时会自动将其重新缝合为最精简的大网段，零冗余、防误杀，完美适配防火墙（IPTables/ROS）规则下发。
  3. 极速单树双栈检索：利用 Python 原生 bisect 二分查找算法操作内存数组，单核轻松应对极高的 QPS 并发。

🛠️ 部署指南

1. 前置准备

  • 宿主机已安装 Docker 和 Docker Compose。
  • 前往 IPinfo 官网 注册免费账号，并在控制台首页获取你的 Access Token。

2. 项目目录结构

在服务器创建工作目录（如 /opt/asn-api），包含以下 4 个文件：

📄 docker-compose.yml

services:
  asn-api:
    build: .
    container_name: bgp-asn-api
    restart: always
    ports:
      - "8000:8000"
    environment:
      - TZ=Asia/Shanghai

📄 Dockerfile

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .

RUN pip install --no-cache-dir -r requirements.txt

COPY main.py .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

📄 requirements.txt

fastapi==0.104.1
uvicorn==0.24.0

📄 main.py (核心业务逻辑)

⚠️ 部署必读：请将代码中第 19 行的 YOUR_TOKEN_HERE 替换为你申请的 IPinfo Token。

import urllib.request
import gzip
import csv
import ipaddress
import bisect
import threading
import time
import logging
from typing import Optional
from fastapi import FastAPI, HTTPException, Query

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")

app = FastAPI(title="IPinfo Country & ASN API", description="融合 ASN 与国家级地理位置的双栈 IP 检索服务")

# ==========================================
# 🛑 请填入你在 IPinfo 官网申请的免费 Token
# ==========================================
IPINFO_TOKEN = "YOUR_TOKEN_HERE"

# 全局内存数据结构 (双栈读写分离)
start_ips_v4, ranges_v4 = [], []
start_ips_v6, ranges_v6 = [], []
asn_map = {} 

def update_database():
    """下载最新的 Country + ASN 融合数据库并解析到内存中"""
    if IPINFO_TOKEN == "YOUR_TOKEN_HERE":
        logging.error("未配置 IPINFO_TOKEN！请前往 https://ipinfo.io 注册获取。")
        return

    logging.info("开始从 IPinfo 下载 Country+ASN 融合数据库...")
    url = f"https://ipinfo.io/data/free/country_asn.csv.gz?token={IPINFO_TOKEN}"
    
    try:
        req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
        with urllib.request.urlopen(req, timeout=30) as response:
            with gzip.open(response, 'rt', encoding='utf-8') as f:
                reader = csv.reader(f)
                next(reader, None) # 跳过表头
                
                temp_v4_ips, temp_v4_ranges = [], []
                temp_v6_ips, temp_v6_ranges = [], []
                temp_asn_map = {}
                
                count = 0
                for row in reader:
                    if len(row) < 9: continue
                    
                    start_ip_str = row[0]
                    end_ip_str = row[1]
                    country_code = row[2]
                    country_name = row[3]
                    asn_raw = row[6].upper().replace("AS", "")
                    as_name = row[7]
                    as_domain = row[8]
                    
                    if not asn_raw: 
                        continue
                        
                    try:
                        s_ip = ipaddress.ip_address(start_ip_str)
                        e_ip = ipaddress.ip_address(end_ip_str)
                        ip_version = s_ip.version
                        
                        # 构建核心元组
                        data_tuple = (int(e_ip), asn_raw, as_name, as_domain, country_code, country_name)
                        
                        if ip_version == 4:
                            temp_v4_ips.append(int(s_ip))
                            temp_v4_ranges.append(data_tuple)
                        else:
                            temp_v6_ips.append(int(s_ip))
                            temp_v6_ranges.append(data_tuple)
                        
                        if asn_raw not in temp_asn_map:
                            temp_asn_map[asn_raw] = {"ipv4": [], "ipv6": []}
                        
                        # 转换并存储 CIDR 碎片
                        cidrs = [str(net) for net in ipaddress.summarize_address_range(s_ip, e_ip)]
                        if ip_version == 4:
                            temp_asn_map[asn_raw]["ipv4"].extend(cidrs)
                        else:
                            temp_asn_map[asn_raw]["ipv6"].extend(cidrs)
                            
                        count += 1
                    except ValueError:
                        continue
                
                # 🚀 终极 CIDR 路由聚合 (缝合地理碎片)
                logging.info("正在执行 CIDR 路由聚合优化，合并地理碎片网段...")
                for asn_key, data in temp_asn_map.items():
                    if data["ipv4"]:
                        v4_nets = [ipaddress.IPv4Network(cidr) for cidr in data["ipv4"]]
                        data["ipv4"] = [str(net) for net in ipaddress.collapse_addresses(v4_nets)]
                    if data["ipv6"]:
                        v6_nets = [ipaddress.IPv6Network(cidr) for cidr in data["ipv6"]]
                        data["ipv6"] = [str(net) for net in ipaddress.collapse_addresses(v6_nets)]

                # 线程安全的热重载
                global start_ips_v4, ranges_v4, start_ips_v6, ranges_v6, asn_map
                start_ips_v4, ranges_v4 = temp_v4_ips, temp_v4_ranges
                start_ips_v6, ranges_v6 = temp_v6_ips, temp_v6_ranges
                asn_map = temp_asn_map
                
        logging.info(f"🎉 融合数据库更新成功！共处理 {count} 个分配区间。涵盖 ASN: {len(asn_map)} 个。")
    except Exception as e:
        logging.error(f"❌ 数据库更新失败: {e}")

def background_updater():
    """后台静默更新，每 24 小时执行一次"""
    while True:
        time.sleep(86400) 
        update_database()

@app.on_event("startup")
def startup_event():
    update_database()
    threading.Thread(target=background_updater, daemon=True).start()

@app.get("/ip/{ip_address}")
def query_ip(ip_address: str):
    """双栈 IP 正向查询接口"""
    try:
        ip_obj = ipaddress.ip_address(ip_address)
        ip_int = int(ip_obj)
        ip_version = ip_obj.version
    except ValueError:
        raise HTTPException(status_code=400, detail="无效的 IP 地址格式")
    
    target_ips = start_ips_v4 if ip_version == 4 else start_ips_v6
    target_ranges = ranges_v4 if ip_version == 4 else ranges_v6

    idx = bisect.bisect_right(target_ips, ip_int) - 1
    
    if idx >= 0:
        end_ip_int, asn, as_name, as_domain, country_code, country_name = target_ranges[idx]
        if ip_int <= end_ip_int:
            return {
                "ip": ip_address,
                "version": f"IPv{ip_version}",
                "asn": f"AS{asn}",
                "asn_name": as_name,
                "asn_domain": as_domain,
                "country_code": country_code,
                "country_name": country_name
            }
            
    raise HTTPException(status_code=404, detail="未找到该 IP 的归属信息")

@app.get("/asn/{asn}")
def query_asn(
    asn: str, 
    version: Optional[str] = Query(None, description="填 'v4' 或 'v6' 单独获取对应 IP 段")
):
    """双栈 ASN 反查接口"""
    clean_asn = asn.upper().replace("AS", "")
    
    if clean_asn not in asn_map:
        raise HTTPException(status_code=404, detail="未找到该 ASN")
        
    data = asn_map[clean_asn]
    
    if version and version.lower() in ["4", "v4", "ipv4"]:
        return {"asn": f"AS{clean_asn}", "version": "IPv4", "total_prefixes": len(data["ipv4"]), "prefixes": data["ipv4"]}
        
    elif version and version.lower() in ["6", "v6", "ipv6"]:
        return {"asn": f"AS{clean_asn}", "version": "IPv6", "total_prefixes": len(data["ipv6"]), "prefixes": data["ipv6"]}
        
    return {
        "asn": f"AS{clean_asn}",
        "total_ipv4_prefixes": len(data["ipv4"]),
        "total_ipv6_prefixes": len(data["ipv6"]),
        "ipv4_prefixes": data["ipv4"],
        "ipv6_prefixes": data["ipv6"]
    }

3. 一键启动

docker compose up -d --build
docker compose logs -f

📡 API 调用文档与完整返回示例

服务默认监听 http://你的IP:8000。提供 Swagger 可视化文档地址：http://你的IP:8000/docs。

接口 1：IP 正向查询 (/ip/{ip_address})

✅ 场景 A：成功查询（公网/隐藏骨干网 IP）

请求：GET /ip/43.255.168.66

{
  "ip": "43.255.168.66",
  "version": "IPv4",
  "asn": "AS10099",
  "asn_name": "China Unicom Global",
  "asn_domain": "chinaunicomglobal.com",
  "country_code": "HK",
  "country_name": "Hong Kong"
}

❌ 场景 B：查询失败（未分配的 IP 或局域网私有 IP）

请求：GET /ip/10.0.0.1 （内网 IP 未在产权库中注册）
HTTP 状态码：404 Not Found

{
  "detail": "未找到该 IP 的归属信息"
}

❌ 场景 C：查询失败（输入了乱码或非法的 IP 格式）

请求：GET /ip/999.999.999.999
HTTP 状态码：400 Bad Request

{
  "detail": "无效的 IP 地址格式"
}

接口 2：ASN 反向查询 (/asn/{asn})

✅ 场景 A：成功查询（全量双栈数据导出）

请求：GET /asn/AS10099 （输入 10099 或 as10099 均自动兼容）

{
  "asn": "AS10099",
  "total_ipv4_prefixes": 36,
  "total_ipv6_prefixes": 1,
  "ipv4_prefixes": [
    "38.69.152.0/24",
    "43.251.12.0/22",
    "162.255.48.0/22" 
    // ... 已深度聚合，零碎片冗余
  ],
  "ipv6_prefixes": [
    "2401:8a00::/32"
  ]
}

✅ 场景 B：成功查询（带版本参数，按需过滤）

请求：GET /asn/AS10099?version=v6 （支持 v4, 4, ipv4, v6, 6, ipv6）

{
  "asn": "AS10099",
  "version": "IPv6",
  "total_prefixes": 1,
  "prefixes": [
    "2401:8a00::/32"
  ]
}

❌ 场景 C：查询失败（查询了不存在的 ASN 或无 IP 资产的空壳 ASN）

请求：GET /asn/AS999999999
HTTP 状态码：404 Not Found

{
  "detail": "未找到该 ASN"
}

site logo

Hi,Friend

© 2026 247LIKE-BLOG