胖子百科

公众号旧文

什么是 Looking Glass（lg）？

作者：胖子阅读 —
#微信公众号#网络测速

十几年前就看到过 Looking Glass 这个词，一直都懵懵懂懂的。

Looking Glass 直译过来是 观测镜，但我一直不知道是用来观测什么的。

比较官方的说法是：

Looking Glass 是基于 Web 的网络探测工具，可直接从当前节点发起网络诊断，帮助用户快速了解网络连通性、路由路径以及服务可用性。

说白了就是：

上线一台服务器，在网页上对外提供这个服务器的一些探测能力。

我这个站点上线已经很长时间了，长到我都忘了自己还有这么一个站。

http://45.202.240.112/

没用域名是因为，Looking Glass 观测的是从服务器到某个目标的结果。

比如用户自己的公网 IP，此时应该叫“服务器回程”。

而去程就需要服务器的 IP 了，于是我就直接把 IP 给放出来了。

目前支持以下功能：

  • Ping：检测目标主机是否可达，并查看延迟、丢包率等网络质量信息。
  • Traceroute：追踪数据包到目标主机的完整路由路径，展示每一跳的节点及往返延迟，方便分析网络出口和路由情况。
  • DNS 查询：查询域名解析结果，验证 DNS 是否正常生效。
  • HTTP 检测：获取目标网站的 HTTP/HTTPS 响应头，检查 Web 服务是否正常访问及返回状态。
  • TCP 检测：检测指定 TCP 端口是否能够建立连接，用于验证常见服务端口的可达性。

为了保障服务的安全与稳定，本 Looking Glass 有多项安全机制，包括：

  • 请求频率限制（Rate Limit）
  • 最大并发控制
  • SSRF 防护
  • 私有地址及保留地址过滤
  • 操作日志记录

所有探测均从当前节点发起，可用于排查不同地区节点之间的网络质量、路由路径及服务连通性。

本项目仍在持续完善中，后续将继续增加更多网络诊断功能。

比如：

  • DNS 自动识别域名 / IP
  • HTTP 显示所有解析到的 IP
  • 给每个接口显示耗时
  • Trace 支持 IPv6

欢迎体验并提出宝贵建议。

最后放一下源码：

from flask import Flask, request, jsonify, render_template_string
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
import subprocess
import ipaddress
import socket
import re
import time
from datetime import datetime
from threading import BoundedSemaphore, Lock

app = Flask(__name__)

limiter = Limiter(
    get_remote_address,
    app=app,
    default_limits=["30 per minute"]
)

MAX_RUNNING = 10
SEM = BoundedSemaphore(MAX_RUNNING)
RUNNING_COUNT = 0
RUNNING_LOCK = Lock()

PRIVATE_NETS = [
    ipaddress.ip_network("0.0.0.0/8"),
    ipaddress.ip_network("10.0.0.0/8"),
    ipaddress.ip_network("127.0.0.0/8"),
    ipaddress.ip_network("169.254.0.0/16"),
    ipaddress.ip_network("172.16.0.0/12"),
    ipaddress.ip_network("192.168.0.0/16"),
    ipaddress.ip_network("224.0.0.0/4"),
    ipaddress.ip_network("240.0.0.0/4"),
    ipaddress.ip_network("::1/128"),
    ipaddress.ip_network("fc00::/7"),
    ipaddress.ip_network("fe80::/10"),
]

DOMAIN_RE = re.compile(r"^[a-zA-Z0-9.-]{1,253}$")
ALLOWED_PORTS = {80, 443, 22, 25, 53}

INFO_CACHE = {"time": 0, "data": None}


def check_public_ip(ip_text):
    ip = ipaddress.ip_address(ip_text)
    for net in PRIVATE_NETS:
        if ip in net:
            raise ValueError("不允许探测内网、保留地址或本机地址")
    return True


def resolve_all_ips(target):
    target = target.strip()

    if not DOMAIN_RE.match(target):
        raise ValueError("目标格式不合法")

    try:
        ip = ipaddress.ip_address(target)
    except ValueError:
        ip = None

    if ip is not None:
        check_public_ip(str(ip))
        return [str(ip)]

    infos = socket.getaddrinfo(target, None, proto=socket.IPPROTO_TCP)
    ips = sorted(set([i[4][0] for i in infos]))

    if not ips:
        raise ValueError("解析失败")

    for ip in ips:
        check_public_ip(ip)

    return ips


def resolve_target(target):
    return resolve_all_ips(target)[0]


def run_cmd(cmd, timeout=12):
    global RUNNING_COUNT

    with SEM:
        with RUNNING_LOCK:
            RUNNING_COUNT += 1

        try:
            p = subprocess.run(
                cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                text=True,
                timeout=timeout
            )
            return p.stdout[-12000:]
        finally:
            with RUNNING_LOCK:
                RUNNING_COUNT -= 1


def client_ip():
    return request.headers.get("X-Real-IP") or request.remote_addr or "-"


def write_log(kind, target, ip="-", port="-", ok=True):
    line = (
        f"{datetime.utcnow().isoformat()}Z\t"
        f"{client_ip()}\t"
        f"{kind}\t"
        f"{target}\t"
        f"{ip}\t"
        f"{port}\t"
        f"{ok}\n"
    )
    with open("/var/log/net-lg.log", "a", encoding="utf-8") as f:
        f.write(line)


def safe_log(kind, target, ip="-", port="-", ok=True):
    try:
        write_log(kind, target, ip, port, ok)
    except Exception:
        pass


def get_node_info():
    now = time.time()

    if INFO_CACHE["data"] and now - INFO_CACHE["time"] < 600:
        return INFO_CACHE["data"]

    hostname = subprocess.check_output(
        ["hostname"],
        text=True,
        timeout=3
    ).strip()

    geo_raw = subprocess.check_output(
        ["curl", "-s", "--max-time", "5", "myip.ipip.net"],
        text=True,
        timeout=6
    ).strip()

    ip = ""
    location = geo_raw

    if "当前 IP：" in geo_raw:
        ip = geo_raw.split("当前 IP：", 1)[1].split()[0].strip()

    if "来自于：" in geo_raw:
        location = geo_raw.split("来自于：", 1)[1].strip()
    elif "来自于" in geo_raw:
        location = geo_raw.split("来自于", 1)[1].strip()

    data = {
        "hostname": hostname,
        "ip": ip,
        "location": location,
        "raw": geo_raw
    }

    INFO_CACHE["time"] = now
    INFO_CACHE["data"] = data
    return data


@app.route("/api/info")
def api_info():
    try:
        return jsonify(ok=True, **get_node_info())
    except Exception as e:
        return jsonify(ok=False, error=str(e)), 500


@app.route("/api/status")
def api_status():
    with RUNNING_LOCK:
        running = RUNNING_COUNT

    return jsonify(
        ok=True,
        running=running,
        max_running=MAX_RUNNING,
        available=MAX_RUNNING - running
    )


@app.route("/")
def index():
    return render_template_string("""
<!doctype html>
<html>
<head>
<meta charset="utf-8">
<title>Network Looking Glass</title>

<style>
body { font-family: Arial, sans-serif; max-width: 900px; margin: 40px auto; padding: 0 20px; }
input, select, button { font-size: 16px; padding: 8px; }
input { width: 360px; }
pre { background: #111; color: #eee; padding: 16px; overflow: auto; border-radius: 8px; }
.card { border: 1px solid #ddd; padding: 20px; border-radius: 10px; }
#nodeinfo { color: #666; margin-bottom: 8px; }
#loadinfo { color: #666; margin-bottom: 15px; }
#type-buttons { margin: 10px 0; }
#type-buttons button { margin-right: 6px; padding: 8px 12px; cursor: pointer; border: 1px solid #ccc; background: #f7f7f7; border-radius: 6px; }
#type-buttons button.active { background: #222; color: white; border-color: #222; }
</style>

</head>
<body>
<h1>Network Looking Glass</h1>
<p id="nodeinfo">加载节点信息中...</p>
<p id="loadinfo">节点负载：加载中...</p>

<div class="card">
  <p>从本节点发起 ping / traceroute / DNS / HTTP / TCP 查询。</p>
  <input id="target" placeholder="example.com 或 8.8.8.8">
  <input id="port" placeholder="端口(默认443)" style="width:120px; display:none;">
  <div id="type-buttons">
    <button type="button" data-type="ping">Ping</button>
    <button type="button" data-type="trace">Trace</button>
    <button type="button" data-type="dns">DNS</button>
    <button type="button" data-type="http">HTTP</button>
    <button type="button" data-type="tcp">TCP</button>
  </div>
  <button onclick="run()">开始探测</button>
</div>

<h3>结果</h3>
<pre id="out">等待输入...</pre>

<script>
async function loadInfo() {
  try {
    let r = await fetch("/api/info");
    let j = await r.json();

    if (j.ok) {
      let text = "节点：" + j.hostname;
      if (j.ip) text += " | 出口IP：" + j.ip;
      if (j.location) text += " | 位置：" + j.location;
      document.getElementById("nodeinfo").textContent = text;
    } else {
      document.getElementById("nodeinfo").textContent = "节点信息获取失败";
    }
  } catch (e) {
    document.getElementById("nodeinfo").textContent = "节点信息获取失败";
  }
}

async function loadStatus() {
  try {
    let r = await fetch("/api/status");
    let j = await r.json();

    if (j.ok) {
      document.getElementById("loadinfo").textContent =
        "节点负载：当前并发 " + j.running + " / " + j.max_running;
    } else {
      document.getElementById("loadinfo").textContent = "节点负载：获取失败";
    }
  } catch (e) {
    document.getElementById("loadinfo").textContent = "节点负载：获取失败";
  }
}

loadInfo();
loadStatus();
setInterval(loadStatus, 3000);

let currentType = "ping";

function updatePortBox() {
  let portBox = document.getElementById("port");
  portBox.style.display = currentType === "tcp" ? "inline-block" : "none";
}

function setType(type) {
  currentType = type;

  document.querySelectorAll("#type-buttons button").forEach(btn => {
    if (btn.dataset.type === type) {
      btn.classList.add("active");
    } else {
      btn.classList.remove("active");
    }
  });

  updatePortBox();
}

document.querySelectorAll("#type-buttons button").forEach(btn => {
  btn.addEventListener("click", function () {
    setType(this.dataset.type);
  });
});

setType("ping");

async function run() {
  let target = document.getElementById("target").value;
  let type = currentType;
  let port = document.getElementById("port").value || "443";

  let out = document.getElementById("out");
  out.textContent = "运行中...";

  let url = "/api/" + type + "?target=" + encodeURIComponent(target);

  if (type === "tcp") {
    url += "&port=" + encodeURIComponent(port);
  }

  try {
    let r = await fetch(url);

    if (r.status === 429) {
      out.textContent = "⚠️ 请求过于频繁，请稍后再试（限速触发）";
      loadStatus();
      return;
    }

    let j = await r.json();

    if (!r.ok) {
      out.textContent = j.error
        ? ("错误: " + j.error)
        : ("❌ 请求失败，HTTP 状态码：" + r.status);
      loadStatus();
      return;
    }

    out.textContent = j.ok ? j.output : ("错误: " + j.error);
    loadStatus();

  } catch (e) {
    out.textContent = "❌ 网络错误：" + e;
    loadStatus();
  }
}
</script>
</body>
</html>
""")


@app.route("/api/ping")
@limiter.limit("10 per minute")
def api_ping():
    target = request.args.get("target", "")
    try:
        ip = resolve_target(target)
        output = run_cmd(["ping", "-c", "4", "-W", "2", ip], timeout=8)
        safe_log("ping", target, ip, "-", True)
        return jsonify(ok=True, target=target, ip=ip, output=output)
    except Exception as e:
        safe_log("ping", target, "-", "-", False)
        return jsonify(ok=False, error=str(e)), 400


@app.route("/api/trace")
@limiter.limit("5 per minute")
def api_trace():
    target = request.args.get("target", "")
    try:
        ip = resolve_target(target)
        output = run_cmd(["traceroute", "-m", "20", "-w", "2", ip], timeout=15)
        safe_log("trace", target, ip, "-", True)
        return jsonify(ok=True, target=target, ip=ip, output=output)
    except Exception as e:
        safe_log("trace", target, "-", "-", False)
        return jsonify(ok=False, error=str(e)), 400


@app.route("/api/dns")
@limiter.limit("20 per minute")
def api_dns():
    target = request.args.get("target", "")
    try:
        if not DOMAIN_RE.match(target):
            raise ValueError("域名格式不合法")
        output = run_cmd(["dig", "+short", target], timeout=5)
        safe_log("dns", target, "-", "-", True)
        return jsonify(ok=True, target=target, output=output)
    except Exception as e:
        safe_log("dns", target, "-", "-", False)
        return jsonify(ok=False, error=str(e)), 400


@app.route("/api/http")
@limiter.limit("10 per minute")
def api_http():
    target = request.args.get("target", "").strip()
    try:
        if not DOMAIN_RE.match(target):
            raise ValueError("目标格式不合法")

        ips = resolve_all_ips(target)
        ip = ips[0]

        https_output = run_cmd(
            [
                "curl",
                "-I",
                "-sS",
                "--max-time",
                "6",
                "--user-agent",
                "Net-LG/1.0",
                "--resolve",
                f"{target}:443:{ip}",
                f"https://{target}"
            ],
            timeout=8
        )

        if "HTTP/" in https_output and "curl:" not in https_output:
            safe_log("http", target, ip, 443, True)
            return jsonify(
                ok=True,
                target=target,
                ip=ip,
                all_ips=ips,
                protocol="https",
                output="=== HTTPS ===\n" + https_output
            )

        http_output = run_cmd(
            [
                "curl",
                "-I",
                "-sS",
                "--max-time",
                "6",
                "--user-agent",
                "Net-LG/1.0",
                "--resolve",
                f"{target}:80:{ip}",
                f"http://{target}"
            ],
            timeout=8
        )

        ok = "HTTP/" in http_output
        safe_log("http", target, ip, "443/80", ok)

        return jsonify(
            ok=True,
            target=target,
            ip=ip,
            all_ips=ips,
            protocol="http_fallback",
            output="=== HTTPS 失败，尝试 HTTP ===\n"
                   + https_output
                   + "\n\n=== HTTP ===\n"
                   + http_output
        )

    except Exception as e:
        safe_log("http", target, "-", "-", False)
        return jsonify(ok=False, error=str(e)), 400


@app.route("/api/tcp")
@limiter.limit("10 per minute")
def api_tcp():
    target = request.args.get("target", "")
    try:
        port = int(request.args.get("port", "443"))

        if port not in ALLOWED_PORTS:
            raise ValueError("端口不允许")

        ip = resolve_target(target)

        output = run_cmd(
            ["nc", "-vz", "-w", "5", ip, str(port)],
            timeout=6
        )

        safe_log("tcp", target, ip, port, True)
        return jsonify(ok=True, target=target, ip=ip, port=port, output=output)
    except Exception as e:
        safe_log("tcp", target, "-", request.args.get("port", "-"), False)
        return jsonify(ok=False, error=str(e)), 400


if __name__ == "__main__":
    app.run(host="127.0.0.1", port=8090)