轻量级 IPv6 在线检测系统：一键部署与使用指南

　　这是一套基于 Python + Flask 的极简、安全、轻量的 IPv6 在线检测系统，专为服务器 IPv6 连通性、域名 IPv6 解析、TCP 端口开放状态检测设计。

直接在服务器上创建项目目录并进入。

mkdir -p ~/ipv6-checker
cd ~/ipv6-checker

项目包含 3 个核心文件：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from flask import Flask, request, jsonify
import socket
import ipaddress
from flask_cors import CORS
import re
import subprocess
import time
from datetime import datetime
from functools import wraps

app = Flask(__name__)
CORS(app)  # 启用 CORS 跨域支持

# ============ 频率限制 ============
RATE_LIMIT = {}
RATE_LIMIT_WINDOW = 60
RATE_LIMIT_MAX = 20

def rate_limit_check(client_ip):
    now = time.time()
    if client_ip not in RATE_LIMIT:
        RATE_LIMIT[client_ip] = []
    RATE_LIMIT[client_ip] = [t for t in RATE_LIMIT[client_ip] if now - t < RATE_LIMIT_WINDOW]
    if len(RATE_LIMIT[client_ip]) >= RATE_LIMIT_MAX:
        return False
    RATE_LIMIT[client_ip].append(now)
    return True

def rate_limit_decorator(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        client_ip = request.remote_addr
        if not rate_limit_check(client_ip):
            return jsonify({
                'success': False,
                'error': f'Rate limit exceeded. Max {RATE_LIMIT_MAX} requests per {RATE_LIMIT_WINDOW} seconds'
            }), 429
        return f(*args, **kwargs)
    return decorated_function

# ============ 验证函数 ============
def is_valid_ipv6(address):
    try:
        ipaddress.IPv6Address(address)
        return True
    except:
        return False

def is_valid_domain(domain):
    pattern = r'^([a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]{2,}$'
    return re.match(pattern, domain) is not None

def resolve_ipv6(hostname):
    try:
        addr_info = socket.getaddrinfo(hostname, None, socket.AF_INET6)
        return addr_info[0][4][0]
    except Exception as e:
        return None

# ============ Ping 功能 ============
def ping_ipv6(target, count=4, timeout=2):
    try:
        cmd = ['ping6', '-c', str(count), '-W', str(timeout), target]
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
        
        output = result.stdout + result.stderr
        
        loss_match = re.search(r'(\d+)% packet loss', output)
        loss = int(loss_match.group(1)) if loss_match else 100
        
        avg_match = re.search(r'rtt min/avg/max/mdev = [\d.]+/([\d.]+)/', output)
        avg_time = float(avg_match.group(1)) if avg_match else None
        
        min_match = re.search(r'rtt min/avg/max/mdev = ([\d.]+)/', output)
        min_time = float(min_match.group(1)) if min_match else None
        
        max_match = re.search(r'rtt min/avg/max/mdev = [\d.]+/[\d.]+/([\d.]+)/', output)
        max_time = float(max_match.group(1)) if max_match else None
        
        return {
            'success': loss < 100,
            'loss': loss,
            'avg_time': avg_time,
            'min_time': min_time,
            'max_time': max_time,
            'transmitted': count,
            'received': count - int(count * loss / 100)
        }
    except subprocess.TimeoutExpired:
        return {'success': False, 'loss': 100, 'error': 'Ping command timeout'}
    except Exception as e:
        return {'success': False, 'loss': 100, 'error': str(e)}

# ============ TCP 端口检测 ============
def tcp_check(ip, port, timeout=5):
    try:
        start_time = datetime.now()
        sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
        sock.settimeout(timeout)
        
        result = sock.connect_ex((ip, port))
        end_time = datetime.now()
        
        elapsed_ms = int((end_time - start_time).total_seconds() * 1000)
        sock.close()
        
        if result == 0:
            return {'success': True, 'time_ms': elapsed_ms}
        else:
            error_map = {
                10060: 'Connection timeout',
                10061: 'Connection refused',
                111: 'Connection refused',
                110: 'Connection timeout',
                113: 'No route to host'
            }
            return {'success': False, 'error': error_map.get(result, f'Error code: {result}')}
    except socket.timeout:
        return {'success': False, 'error': f'Connection timeout ({timeout}s)'}
    except Exception as e:
        return {'success': False, 'error': str(e)}

# ============ API 接口 ============
@app.route('/', methods=['GET'])
def index():
    return jsonify({
        'service': 'IPv6 Checker API',
        'version': '1.0',
        'endpoints': {
            '/api/ping': {
                'method': 'POST',
                'description': 'Test IPv6 connectivity using ICMP ping',
                'body': {'target': 'ipv6 address or domain'}
            },
            '/api/tcp': {
                'method': 'POST',
                'description': 'Test TCP port connectivity',
                'body': {'target': 'ipv6 address or domain', 'port': 'port number', 'timeout': 'seconds (optional)'}
            },
            '/health': {
                'method': 'GET',
                'description': 'Health check'
            }
        }
    })

@app.route('/health', methods=['GET'])
def health():
    return jsonify({'status': 'ok', 'timestamp': datetime.now().isoformat()})

@app.route('/api/ping', methods=['POST'])
@rate_limit_decorator
def api_ping():
    data = request.get_json()
    if not data:
        return jsonify({'success': False, 'error': 'Invalid JSON body'}), 400
    
    target = data.get('target', '').strip()
    if not target:
        return jsonify({'success': False, 'error': 'Missing target parameter'}), 400
    
    is_ip = is_valid_ipv6(target)
    is_domain = is_valid_domain(target)
    
    if not is_ip and not is_domain:
        return jsonify({'success': False, 'error': 'Invalid IPv6 address or domain format'}), 400
    
    ip_address = target
    if is_domain:
        ip_address = resolve_ipv6(target)
        if not ip_address:
            return jsonify({'success': False, 'error': 'Failed to resolve domain to IPv6'}), 400
    
    count = data.get('count', 4)
    timeout = data.get('timeout', 2)
    result = ping_ipv6(ip_address, count, timeout)
    
    result['ip'] = ip_address
    result['target'] = target
    result['timestamp'] = datetime.now().isoformat()
    
    return jsonify(result)

@app.route('/api/tcp', methods=['POST'])
@rate_limit_decorator
def api_tcp():
    data = request.get_json()
    if not data:
        return jsonify({'success': False, 'error': 'Invalid JSON body'}), 400
    
    target = data.get('target', '').strip()
    port = data.get('port')
    
    if not target:
        return jsonify({'success': False, 'error': 'Missing target parameter'}), 400
    
    if not port:
        return jsonify({'success': False, 'error': 'Missing port parameter'}), 400
    
    try:
        port = int(port)
        if port < 1 or port > 65535:
            raise ValueError
    except:
        return jsonify({'success': False, 'error': 'Port must be between 1 and 65535'}), 400
    
    timeout = data.get('timeout', 5)
    
    is_ip = is_valid_ipv6(target)
    is_domain = is_valid_domain(target)
    
    if not is_ip and not is_domain:
        return jsonify({'success': False, 'error': 'Invalid IPv6 address or domain format'}), 400
    
    ip_address = target
    if is_domain:
        ip_address = resolve_ipv6(target)
        if not ip_address:
            return jsonify({'success': False, 'error': 'Failed to resolve domain to IPv6'}), 400
    
    result = tcp_check(ip_address, port, timeout)
    
    result['ip'] = ip_address
    result['target'] = target
    result['port'] = port
    result['timestamp'] = datetime.now().isoformat()
    
    return jsonify(result)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080, debug=False, threaded=True)

Flask==2.3.3
flask-cors==4.0.0

FROM python:3.11-alpine

# 安装 ping6
RUN apk add --no-cache iputils

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple
RUN pip install --no-cache-dir gunicorn

COPY app.py .

# 创建非 root 用户
RUN addgroup -g 1000 -S appgroup && \
    adduser -u 1000 -S appuser -G appgroup

USER appuser

EXPOSE 8080

# 生产环境推荐（使用 gunicorn）
CMD ["gunicorn", "-b", "0.0.0.0:8080", "-w", "4", "app:app"]

进入目录后执行构建命令。

cd ~/ipv6-checker

# 构建镜像
docker build -t ipv6-checker .

使用 host 网络模式运行，确保 IPv6 正常使用，并配置开机自启。

# 运行容器（使用 host 网络）
docker run -d \
  --name ipv6-checker \
  --network host \
  --restart unless-stopped \
  ipv6-checker

# 验证
docker logs ipv6-checker
curl http://localhost:8080/health
curl -X POST http://localhost:8080/api/ping -H "Content-Type: application/json" -d '{"target": "www.qq.com"}'

所有接口返回统一 JSON 格式，包含状态、延迟、丢包率、错误信息等。

curl -X POST http://localhost:8080/api/ping \
  -H "Content-Type: application/json" \
  -d '{"target": "2001:4860:4860::8888"}'

curl -X POST http://localhost:8080/api/tcp \
  -H "Content-Type: application/json" \
  -d '{"target": "www.qq.com", "port": 80}'

六、前端对接说明

• 已内置 flask-cors 开启全跨域，前端可直接调用

• 无需额外配置 Nginx 即可使用
• 如需域名访问，可使用 Nginx 反向代理
• 注：为避免对其它api干扰，ad80a1/ipv6-checker:latest镜像内置跨域功能未开启

七、Nginx 反向代理配置

八、日常运维命令

# 查看日志
docker logs -f ipv6-checker

# 停止容器
docker stop ipv6-checker

# 启动容器
docker start ipv6-checker

# 重启容器
docker restart ipv6-checker

# 删除容器
docker stop ipv6-checker && docker rm ipv6-checker

# 进入容器调试
docker exec -it ipv6-checker sh

九、后续一键部署方案

docker run -d \
  --name ipv6-checker \
  --network host \
  --restart unless-stopped \
  ad80a1/ipv6-checker:latest

十、常见问题

1. 无法检测 IPv6检查服务器是否获取公网 IPv6 地址，关闭防火墙 IPv6 限制。
2. Ping 全部丢包确认目标支持 IPv6，或检查本机 IPv6 路由与防火墙规则。
3. API 访问报错 429触发频率限制，等待 60 秒后重试，可自行修改代码调整限流策略。
4. Docker 启动失败检查端口 8080 是否被占用，或更换映射端口。

十一、总结

• 个人 IPv6 网络检测工具
• 企业内网 IPv6 服务巡检平台
• 网站 IPv6 支持性检测入口
• 集成到运维系统的检测模块