81 lines
2.9 KiB
Python
81 lines
2.9 KiB
Python
import http.server
|
||
import socket
|
||
import socketserver
|
||
import ipaddress
|
||
import logging
|
||
from datetime import datetime
|
||
|
||
# 日志配置（线程安全），按时间和级别输出到 stdout
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s %(levelname)s %(message)s",
|
||
datefmt="%Y-%m-%d %H:%M:%S",
|
||
)
|
||
|
||
class TCPServer6(socketserver.TCPServer):
|
||
address_family = socket.AF_INET6
|
||
allow_reuse_address = True
|
||
|
||
class MyHTTPRequestHandler(http.server.SimpleHTTPRequestHandler):
|
||
def log_response(self, status_code, client_ip, x_forwarded_for):
|
||
# 记录每次响应的日志
|
||
logging.info(
|
||
"method=%s path=%s client_ip=%s x_forwarded_for=%s status=%d",
|
||
self.command,
|
||
self.path,
|
||
client_ip,
|
||
x_forwarded_for if x_forwarded_for is not None else "-",
|
||
status_code,
|
||
)
|
||
|
||
def do_GET(self):
|
||
# 获取请求头中的 'X-Forwarded-For' 或 'Remote-Addr'
|
||
x_forwarded_for = self.headers.get('X-Forwarded-For')
|
||
remote_addr = self.client_address[0]
|
||
|
||
# 客户端 IP 优先选择 X-Forwarded-For
|
||
client_ip = (x_forwarded_for.split(',')[0].strip() if x_forwarded_for else remote_addr)
|
||
|
||
# 将 IPv4-mapped IPv6 地址转换为纯 IPv4（例如 ::ffff:127.0.0.1 -> 127.0.0.1）
|
||
try:
|
||
ip_obj = ipaddress.ip_address(client_ip)
|
||
if isinstance(ip_obj, ipaddress.IPv6Address) and ip_obj.ipv4_mapped:
|
||
client_ip = str(ip_obj.ipv4_mapped)
|
||
except ValueError:
|
||
# 如果解析失败（非标准格式），尝试手工处理以防某些代理返回带方括号或端口
|
||
# 移除 IPv6 方括号及端口（如 [::ffff:127.0.0.1]:8080）
|
||
s = client_ip
|
||
if s.startswith('[') and ']' in s:
|
||
s = s.split(']', 1)[0].lstrip('[')
|
||
if s.count(':') >= 2 and s.startswith('::ffff:'):
|
||
# 处理可能带端口的情况
|
||
s = s.split(':')[-1]
|
||
client_ip = s
|
||
|
||
# 发送响应
|
||
status_code = 200
|
||
self.send_response(status_code)
|
||
self.send_header("Content-type", "text/plain")
|
||
self.end_headers()
|
||
self.wfile.write(client_ip.encode())
|
||
|
||
# 在响应后记录日志
|
||
try:
|
||
self.log_response(status_code, client_ip, x_forwarded_for)
|
||
except Exception:
|
||
logging.exception("Failed to log response")
|
||
|
||
if __name__ == "__main__":
|
||
PORT = 8080
|
||
# 对于 AF_INET6 使用四元组绑定所有 IPv6 地址
|
||
server_address = ("::", PORT, 0, 0)
|
||
with TCPServer6(server_address, MyHTTPRequestHandler) as httpd:
|
||
# 可选：在绑定前设置 IPV6_V6ONLY，若需要同时接受 IPv4 映射地址可设为 0
|
||
try:
|
||
httpd.socket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0)
|
||
except Exception:
|
||
pass
|
||
logging.info("Serving on [::]:%d", PORT)
|
||
httpd.serve_forever()
|
||
|