ptnqoxywl
/
sqlmap
1
0
Fork 1
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

liguanwei_change

Browse Source
liguanwei_branch
liguanwei 3 months ago
parent 5472a4aaa3
commit c1bd472c8d
3 changed files with 225 additions and 166 deletions
Show Stats Download Patch File Download Diff File

247
src/sqlmap-master/thirdparty/socks/socks.py vendored
Unescape View File

@ -121,6 +121,7 @@ def unwrapmodule(module):
module.socket.socket = _orgsocket
module.socket.create_connection = _orgcreateconnection
class socksocket(socket.socket):
"""socksocket([family[, type[, proto]]]) -> socket object
Open a SOCKS enabled socket. The parameters are the same as
@ -129,11 +130,17 @@ class socksocket(socket.socket):
"""
def __init__(self, family=socket.AF_INET, type=socket.SOCK_STREAM, proto=0, _sock=None):
# 初始化父类 socket
_orgsocket.__init__(self, family, type, proto, _sock)
# 根据默认代理设置代理
if _defaultproxy != None:
self.__proxy = _defaultproxy
else:
# 初始化代理信息，如果没有默认代理则设为空元组
self.__proxy = (None, None, None, None, None, None)
# 用于保存代理关联的本地socket信息
self.__proxysockname = None
self.__proxypeername = None
@ -142,121 +149,111 @@ class socksocket(socket.socket):
Receive EXACTLY the number of bytes requested from the socket.
Blocks until the required number of bytes have been received.
"""
# 从socket接收指定字节数数据
data = self.recv(count)
# 当接收的数据少于请求的字节数时继续接收
while len(data) < count:
d = self.recv(count-len(data))
if not d: raise GeneralProxyError((0, "connection closed unexpectedly"))
data = data + d
d = self.recv(count - len(data))
if not d:
raise GeneralProxyError((0, "connection closed unexpectedly")) # 如果连接关闭则抛出异常
data = data + d # 累加接收到的数据
return data
def setproxy(self, proxytype=None, addr=None, port=None, rdns=True, username=None, password=None):
"""setproxy(proxytype, addr[, port[, rdns[, username[, password]]]])
Sets the proxy to be used.
proxytype - The type of the proxy to be used. Three types
are supported: PROXY_TYPE_SOCKS4 (including socks4a),
PROXY_TYPE_SOCKS5 and PROXY_TYPE_HTTP
addr - The address of the server (IP or DNS).
port - The port of the server. Defaults to 1080 for SOCKS
servers and 8080 for HTTP proxy servers.
rdns - Should DNS queries be preformed on the remote side
(rather than the local side). The default is True.
Note: This has no effect with SOCKS4 servers.
username - Username to authenticate with to the server.
The default is no authentication.
password - Password to authenticate with to the server.
Only relevant when username is also provided.
"""
# 设置代理类型、地址、端口等参数
self.__proxy = (proxytype, addr, port, rdns, username, password)
def __negotiatesocks5(self, destaddr, destport):
"""__negotiatesocks5(self,destaddr,destport)
Negotiates a connection through a SOCKS5 server.
"""
# First we'll send the authentication packages we support.
if (self.__proxy[4]!=None) and (self.__proxy[5]!=None):
# The username/password details were supplied to the
# setproxy method so we support the USERNAME/PASSWORD
# authentication (in addition to the standard none).
self.sendall(struct.pack('BBBB', 0x05, 0x02, 0x00, 0x02))
# 发送支持的认证方法
if (self.__proxy[4] != None) and (self.__proxy[5] != None):
# 如果提供了用户名和密码,则支持用户名/密码认证
self.sendall(struct.pack('BBBB', 0x05, 0x02, 0x00, 0x02)) # 0x02 代表用户名/密码认证
else:
# No username/password were entered, therefore we
# only support connections with no authentication.
self.sendall(struct.pack('BBB', 0x05, 0x01, 0x00))
# We'll receive the server's response to determine which
# method was selected
# 仅支持无认证连接
self.sendall(struct.pack('BBB', 0x05, 0x01, 0x00)) # 0x01 代表无认证
# 接收服务器的响应以确定选定的身份验证方法
chosenauth = self.__recvall(2)
if chosenauth[0:1] != b'\x05':
self.close()
raise GeneralProxyError((1, _generalerrors[1]))
# Check the chosen authentication method
raise GeneralProxyError((1, _generalerrors[1])) # 无效的协议版本
# 检查所选的身份验证方法
if chosenauth[1:2] == b'\x00':
# No authentication is required
# 无需身份验证
pass
elif chosenauth[1:2] == b'\x02':
# Okay, we need to perform a basic username/password
# authentication.
self.sendall(b'\x01' + chr(len(self.__proxy[4])).encode() + self.__proxy[4].encode() + chr(len(self.__proxy[5])).encode() + self.__proxy[5].encode())
authstat = self.__recvall(2)
# 需要进行用户名/密码身份验证
self.sendall(b'\x01' + chr(len(self.__proxy[4])).encode() + self.__proxy[4].encode() + chr(
len(self.__proxy[5])).encode() + self.__proxy[5].encode())
authstat = self.__recvall(2) # 接收身份验证状态
if authstat[0:1] != b'\x01':
# Bad response
self.close()
raise GeneralProxyError((1, _generalerrors[1]))
raise GeneralProxyError((1, _generalerrors[1])) # 无效的认证响应
if authstat[1:2] != b'\x00':
# Authentication failed
self.close()
raise Socks5AuthError((3, _socks5autherrors[3]))
# Authentication succeeded
raise Socks5AuthError((3, _socks5autherrors[3])) # 身份验证失败
else:
# Reaching here is always bad
# 非法的身份验证方法
self.close()
if chosenauth[1:2] == b'\xff':
raise Socks5AuthError((2, _socks5autherrors[2]))
raise Socks5AuthError((2, _socks5autherrors[2])) # 不支持的身份验证
else:
raise GeneralProxyError((1, _generalerrors[1]))
# Now we can request the actual connection
req = struct.pack('BBB', 0x05, 0x01, 0x00)
# If the given destination address is an IP address, we'll
# use the IPv4 address request even if remote resolving was specified.
# 请求实际连接
req = struct.pack('BBB', 0x05, 0x01, 0x00) # 0x01 表示连接请求
try:
# 尝试将目标地址作为IP地址处理
ipaddr = socket.inet_aton(destaddr)
req = req + b'\x01' + ipaddr
req = req + b'\x01' + ipaddr # IPv4地址请求
except socket.error:
# Well it's not an IP number, so it's probably a DNS name.
if self.__proxy[3]:
# Resolve remotely
# 目标地址不是IP地址，可能是DNS名称
if self.__proxy[3]: # 检查是否需要远程解析
ipaddr = None
req = req + chr(0x03).encode() + chr(len(destaddr)).encode() + (destaddr if isinstance(destaddr, bytes) else destaddr.encode())
req = req + chr(0x03).encode() + chr(len(destaddr)).encode() + (
destaddr if isinstance(destaddr, bytes) else destaddr.encode())
else:
# Resolve locally
# 本地解析
ipaddr = socket.inet_aton(socket.gethostbyname(destaddr))
req = req + chr(0x01).encode() + ipaddr
req = req + struct.pack(">H", destport)
self.sendall(req)
# Get the response
req = req + struct.pack(">H", destport) # 添加目标端口
self.sendall(req) # 发送请求
# 获取响应
resp = self.__recvall(4)
if resp[0:1] != chr(0x05).encode():
self.close()
raise GeneralProxyError((1, _generalerrors[1]))
raise GeneralProxyError((1, _generalerrors[1])) # 协议错误
elif resp[1:2] != chr(0x00).encode():
# Connection failed
# 连接失败
self.close()
if ord(resp[1:2])<=8:
if ord(resp[1:2]) <= 8:
raise Socks5Error((ord(resp[1:2]), _socks5errors[ord(resp[1:2])]))
else:
raise Socks5Error((9, _socks5errors[9]))
# Get the bound address/port
# 解析绑定地址/端口
elif resp[3:4] == chr(0x01).encode():
boundaddr = self.__recvall(4)
boundaddr = self.__recvall(4) # IPv4绑定地址
elif resp[3:4] == chr(0x03).encode():
resp = resp + self.recv(1)
boundaddr = self.__recvall(ord(resp[4:5]))
resp = resp + self.recv(1) # 读取后续字节
boundaddr = self.__recvall(ord(resp[4:5])) # DNS名
else:
self.close()
raise GeneralProxyError((1,_generalerrors[1]))
raise GeneralProxyError((1, _generalerrors[1]))
# 获取绑定端口
boundport = struct.unpack(">H", self.__recvall(2))[0]
self.__proxysockname = (boundaddr, boundport)
self.__proxysockname = (boundaddr, boundport) # 保存绑定的socket信息
if ipaddr != None:
self.__proxypeername = (socket.inet_ntoa(ipaddr), destport)
self.__proxypeername = (socket.inet_ntoa(ipaddr), destport) # 保存peer名称
else:
self.__proxypeername = (destaddr, destport)
@ -264,135 +261,153 @@ class socksocket(socket.socket):
"""getsockname() -> address info
Returns the bound IP address and port number at the proxy.
"""
return self.__proxysockname
return self.__proxysockname # 返回代理的socket信息
def getproxypeername(self):
"""getproxypeername() -> address info
Returns the IP and port number of the proxy.
"""
return _orgsocket.getpeername(self)
return _orgsocket.getpeername(self) # 返回代理的peer信息
def getpeername(self):
"""getpeername() -> address info
Returns the IP address and port number of the destination
machine (note: getproxypeername returns the proxy)
"""
return self.__proxypeername
return self.__proxypeername # 返回目标机器的peer名称
def __negotiatesocks4(self,destaddr,destport):
def __negotiatesocks4(self, destaddr, destport):
"""__negotiatesocks4(self,destaddr,destport)
Negotiates a connection through a SOCKS4 server.
"""
# Check if the destination address provided is an IP address
# 检查目标地址是否为IP地址
rmtrslv = False
try:
ipaddr = socket.inet_aton(destaddr)
except socket.error:
# It's a DNS name. Check where it should be resolved.
# 目标是DNS名称, 检查是否应进行远程解析
if self.__proxy[3]:
ipaddr = struct.pack("BBBB", 0x00, 0x00, 0x00, 0x01)
ipaddr = struct.pack("BBBB", 0x00, 0x00, 0x00, 0x01) # 默认IP地址
rmtrslv = True
else:
ipaddr = socket.inet_aton(socket.gethostbyname(destaddr))
# Construct the request packet
req = struct.pack(">BBH", 0x04, 0x01, destport) + ipaddr
# The username parameter is considered userid for SOCKS4
ipaddr = socket.inet_aton(socket.gethostbyname(destaddr)) # 本地解析
# 构造请求报文
req = struct.pack(">BBH", 0x04, 0x01, destport) + ipaddr # SOCKS4 请求
# 用户名参数视为userid
if self.__proxy[4] != None:
req = req + self.__proxy[4]
req = req + chr(0x00).encode()
# DNS name if remote resolving is required
# NOTE: This is actually an extension to the SOCKS4 protocol
# called SOCKS4A and may not be supported in all cases.
req = req + chr(0x00).encode() # 请求结束
# 需要远程解析的情况下使用DNS名称
if rmtrslv:
req = req + destaddr + chr(0x00).encode()
self.sendall(req)
# Get the response from the server
req = req + destaddr + chr(0x00).encode() # SOCKS4A扩展
self.sendall(req) # 发送请求
# 获取服务器响应
resp = self.__recvall(8)
if resp[0:1] != chr(0x00).encode():
# Bad data
# 出现错误
self.close()
raise GeneralProxyError((1,_generalerrors[1]))
raise GeneralProxyError((1, _generalerrors[1])) # 无效数据
if resp[1:2] != chr(0x5A).encode():
# Server returned an error
# 服务器返回错误
self.close()
if ord(resp[1:2]) in (91, 92, 93):
self.close()
raise Socks4Error((ord(resp[1:2]), _socks4errors[ord(resp[1:2]) - 90]))
raise Socks4Error((ord(resp[1:2]), _socks4errors[ord(resp[1:2]) - 90])) # 处理错误代码
else:
raise Socks4Error((94, _socks4errors[4]))
# Get the bound address/port
self.__proxysockname = (socket.inet_ntoa(resp[4:]), struct.unpack(">H", resp[2:4])[0])
raise Socks4Error((94, _socks4errors[4])) # 一般性错误
# 获取绑定的地址和端口
self.__proxysockname = (socket.inet_ntoa(resp[4:]), struct.unpack(">H", resp[2:4])[0]) # 保存绑定信息
if rmtrslv != None:
self.__proxypeername = (socket.inet_ntoa(ipaddr), destport)
self.__proxypeername = (socket.inet_ntoa(ipaddr), destport) # 目标地址
else:
self.__proxypeername = (destaddr, destport)
self.__proxypeername = (destaddr, destport) # 目标地址
def __negotiatehttp(self, destaddr, destport):
"""__negotiatehttp(self,destaddr,destport)
Negotiates a connection through an HTTP server.
"""
# If we need to resolve locally, we do this now
# 本地解析
if not self.__proxy[3]:
addr = socket.gethostbyname(destaddr)
else:
addr = destaddr
self.sendall(("CONNECT " + addr + ":" + str(destport) + " HTTP/1.1\r\n" + "Host: " + destaddr + "\r\n\r\n").encode())
# We read the response until we get the string "\r\n\r\n"
# 发送HTTP的CONNECT请求
self.sendall(
("CONNECT " + addr + ":" + str(destport) + " HTTP/1.1\r\n" + "Host: " + destaddr + "\r\n\r\n").encode())
# 读取响应直到遇到"\r\n\r\n"
resp = self.recv(1)
while resp.find("\r\n\r\n".encode()) == -1:
resp = resp + self.recv(1)
# We just need the first line to check if the connection
# was successful
# 仅需第一行检查连接是否成功
statusline = resp.splitlines()[0].split(" ".encode(), 2)
if statusline[0] not in ("HTTP/1.0".encode(), "HTTP/1.1".encode()):
self.close()
raise GeneralProxyError((1, _generalerrors[1]))
raise GeneralProxyError((1, _generalerrors[1])) # 无效的HTTP协议
try:
statuscode = int(statusline[1])
statuscode = int(statusline[1]) # 获取状态码
except ValueError:
self.close()
raise GeneralProxyError((1, _generalerrors[1]))
raise GeneralProxyError((1, _generalerrors[1])) # 状态码无效
if statuscode != 200:
self.close()
raise HTTPError((statuscode, statusline[2]))
self.__proxysockname = ("0.0.0.0", 0)
self.__proxypeername = (addr, destport)
raise HTTPError((statuscode, statusline[2])) # 显示状态码错误
# 连接成功后的代理信息
self.__proxysockname = ("0.0.0.0", 0) # 绑定在0.0.0.0
self.__proxypeername = (addr, destport) # 保存peer信息
def connect(self, destpair):
"""connect(self, despair)
"""connect(self, destpair)
Connects to the specified destination through a proxy.
destpar - A tuple of the IP/DNS address and the port number.
destpair - A tuple of the IP/DNS address and the port number.
(identical to socket's connect).
To select the proxy server use setproxy().
"""
# Do a minimal input check first
if (not type(destpair) in (list,tuple)) or (len(destpair) < 2) or (type(destpair[0]) != type('')) or (type(destpair[1]) != int):
# 校验输入参数
if (not type(destpair) in (list, tuple)) or (len(destpair) < 2) or (type(destpair[0]) != type('')) or (
type(destpair[1]) != int):
raise GeneralProxyError((5, _generalerrors[5]))
# 根据代理类型选择相应的握手协议
if self.__proxy[0] == PROXY_TYPE_SOCKS5:
if self.__proxy[2] != None:
portnum = self.__proxy[2]
else:
portnum = 1080
_orgsocket.connect(self, (self.__proxy[1], portnum))
self.__negotiatesocks5(destpair[0], destpair[1])
portnum = 1080 # 默认端口
_orgsocket.connect(self, (self.__proxy[1], portnum)) # 连接到代理
self.__negotiatesocks5(destpair[0], destpair[1]) # 握手
elif self.__proxy[0] == PROXY_TYPE_SOCKS4:
if self.__proxy[2] != None:
portnum = self.__proxy[2]
else:
portnum = 1080
_orgsocket.connect(self,(self.__proxy[1], portnum))
self.__negotiatesocks4(destpair[0], destpair[1])
portnum = 1080 # 默认端口
_orgsocket.connect(self, (self.__proxy[1], portnum)) # 连接到代理
self.__negotiatesocks4(destpair[0], destpair[1]) # 握手
elif self.__proxy[0] == PROXY_TYPE_HTTP:
if self.__proxy[2] != None:
portnum = self.__proxy[2]
else:
portnum = 8080
_orgsocket.connect(self,(self.__proxy[1], portnum))
self.__negotiatehttp(destpair[0], destpair[1])
portnum = 8080 # 默认端口
_orgsocket.connect(self, (self.__proxy[1], portnum)) # 连接到代理
self.__negotiatehttp(destpair[0], destpair[1]) # 握手
elif self.__proxy[0] == None:
# 没有代理时直接连接目标
_orgsocket.connect(self, (destpair[0], destpair[1]))
else:
raise GeneralProxyError((4, _generalerrors[4]))
raise GeneralProxyError((4, _generalerrors[4])) # 不支持的代理类型
def create_connection(address, timeout=socket._GLOBAL_DEFAULT_TIMEOUT,
source_address=None):

44
src/sqlmap-master/thirdparty/termcolor/termcolor.py vendored
Unescape View File

@ -79,12 +79,14 @@ COLORS = dict(
))
)
# 添加亮色到 COLORS 字典
COLORS.update(dict(("light%s" % color, COLORS[color] + 60) for color in COLORS))
# Reference: https://misc.flogisoft.com/bash/tip_colors_and_formatting
COLORS["lightgrey"] = 37
COLORS["darkgrey"] = 90
# 带有参考的颜色定义
COLORS["lightgrey"] = 37 # 亮灰色
COLORS["darkgrey"] = 90 # 深灰色
# 定义重置颜色的转义序列
RESET = '\033[0m'
@ -104,34 +106,37 @@ def colored(text, color=None, on_color=None, attrs=None):
colored('Hello, World!', 'red', 'on_grey', ['blue', 'blink'])
colored('Hello, World!', 'green')
"""
# 检查环境变量，以决定是否支持 ANSI 颜色
if os.getenv('ANSI_COLORS_DISABLED') is None:
fmt_str = '\033[%dm%s'
fmt_str = '\033[%dm%s' # ANSI 格式化字符串
if color is not None:
text = fmt_str % (COLORS[color], text)
text = fmt_str % (COLORS[color], text) # 将文本颜色应用于文本
if on_color is not None:
text = fmt_str % (HIGHLIGHTS[on_color], text)
text = fmt_str % (HIGHLIGHTS[on_color], text) # 将背景颜色应用于文本
if attrs is not None:
for attr in attrs:
text = fmt_str % (ATTRIBUTES[attr], text)
text = fmt_str % (ATTRIBUTES[attr], text) # 应用文本属性
text += RESET
text += RESET # 重置颜色到默认值
return text
def cprint(text, color=None, on_color=None, attrs=None, **kwargs):
"""Print colorize text.
"""打印带颜色的文本。
It accepts arguments of print function.
支持 print 函数的所有参数。
"""
print((colored(text, color, on_color, attrs)), **kwargs)
print((colored(text, color, on_color, attrs)), **kwargs) # 调用 colored 函数并进行打印
if __name__ == '__main__':
print('Current terminal type: %s' % os.getenv('TERM'))
# 用于测试的主模块
print('Current terminal type: %s' % os.getenv('TERM')) # 打印当前终端类型
print('Test basic colors:')
# 测试基本颜色
cprint('Grey color', 'grey')
cprint('Red color', 'red')
cprint('Green color', 'green')
@ -143,6 +148,7 @@ if __name__ == '__main__':
print(('-' * 78))
print('Test highlights:')
# 测试背景高亮
cprint('On grey color', on_color='on_grey')
cprint('On red color', on_color='on_red')
cprint('On green color', on_color='on_green')
@ -154,6 +160,7 @@ if __name__ == '__main__':
print('-' * 78)
print('Test attributes:')
# 测试文本属性
cprint('Bold grey color', 'grey', attrs=['bold'])
cprint('Dark red color', 'red', attrs=['dark'])
cprint('Underline green color', 'green', attrs=['underline'])
@ -161,13 +168,12 @@ if __name__ == '__main__':
cprint('Reversed blue color', 'blue', attrs=['reverse'])
cprint('Concealed Magenta color', 'magenta', attrs=['concealed'])
cprint('Bold underline reverse cyan color', 'cyan',
attrs=['bold', 'underline', 'reverse'])
attrs=['bold', 'underline', 'reverse'])
cprint('Dark blink concealed white color', 'white',
attrs=['dark', 'blink', 'concealed'])
attrs=['dark', 'blink', 'concealed'])
print(('-' * 78))
print('Test mixing:')
cprint('Underline red on grey color', 'red', 'on_grey',
['underline'])
cprint('Reversed green on red color', 'green', 'on_red', ['reverse'])
# 测试混合使用颜色和属性
cprint('Underline red on grey color', 'red', 'on_grey', ['underline'])
cprint('Reversed green on red color', 'green', 'on_red', ['reverse'])

100
src/sqlmap-master/thirdparty/wininetpton/win_inet_pton.py vendored
Unescape View File

@ -9,30 +9,51 @@ import ctypes
import os
# 定义 sockaddr 结构体，以便与底层 socket API 交互
class sockaddr(ctypes.Structure):
_fields_ = [("sa_family", ctypes.c_short),
("__pad1", ctypes.c_ushort),
("ipv4_addr", ctypes.c_byte * 4),
("ipv6_addr", ctypes.c_byte * 16),
("__pad2", ctypes.c_ulong)]
_fields_ = [
("sa_family", ctypes.c_short), # 地址族，例如 AF_INET 或 AF_INET6
("__pad1", ctypes.c_ushort), # 填充字段用于对齐
("ipv4_addr", ctypes.c_byte * 4), # IPv4 地址（4 字节）
("ipv6_addr", ctypes.c_byte * 16), # IPv6 地址（16 字节）
("__pad2", ctypes.c_ulong) # 额外填充字段
]
# 检测是否在 Windows 平台上，并加载相应的 Windows socket 函数
if hasattr(ctypes, 'windll'):
WSAStringToAddressA = ctypes.windll.ws2_32.WSAStringToAddressA
WSAAddressToStringA = ctypes.windll.ws2_32.WSAAddressToStringA
WSAStringToAddressA = ctypes.windll.ws2_32.WSAStringToAddressA # 字符串转换为地址
WSAAddressToStringA = ctypes.windll.ws2_32.WSAAddressToStringA # 地址转换为字符串
else:
def not_windows():
raise SystemError(
"Invalid platform. ctypes.windll must be available."
)
WSAStringToAddressA = not_windows
WSAAddressToStringA = not_windows
WSAStringToAddressA = not_windows # 非 Windows 平台时的占位符
WSAAddressToStringA = not_windows # 非 Windows 平台时的占位符
def inet_pton(address_family, ip_string):
addr = sockaddr()
addr.sa_family = address_family
addr_size = ctypes.c_int(ctypes.sizeof(addr))
"""
Convert an IP address from text to binary format.
Args:
address_family: The address family (AF_INET for IPv4 or AF_INET6 for IPv6).
ip_string: The IP address in string form (e.g., "192.168.1.1").
Returns:
A packed binary representation of the IP address.
Raises:
socket.error: If the conversion fails.
"""
addr = sockaddr() # 创建 sockaddr 实例
addr.sa_family = address_family # 设置地址族
addr_size = ctypes.c_int(ctypes.sizeof(addr)) # 地址结构的大小
# 调用 WSAStringToAddressA 将字符串 IP 转换为二进制地址
if WSAStringToAddressA(
ip_string,
address_family,
@ -40,34 +61,50 @@ def inet_pton(address_family, ip_string):
ctypes.byref(addr),
ctypes.byref(addr_size)
) != 0:
raise socket.error(ctypes.FormatError())
raise socket.error(ctypes.FormatError()) # 转换失败时抛出异常
# 根据地址族返回相应的打包 IP 地址
if address_family == socket.AF_INET:
return ctypes.string_at(addr.ipv4_addr, 4)
return ctypes.string_at(addr.ipv4_addr, 4) # IPv4 地址
if address_family == socket.AF_INET6:
return ctypes.string_at(addr.ipv6_addr, 16)
return ctypes.string_at(addr.ipv6_addr, 16) # IPv6 地址
raise socket.error('unknown address family')
raise socket.error('unknown address family') # 抛出未知地址族的异常
def inet_ntop(address_family, packed_ip):
addr = sockaddr()
addr.sa_family = address_family
addr_size = ctypes.c_int(ctypes.sizeof(addr))
ip_string = ctypes.create_string_buffer(128)
ip_string_size = ctypes.c_int(ctypes.sizeof(ip_string))
"""
Convert a binary IP address to its text representation.
Args:
address_family: The address family (AF_INET for IPv4 or AF_INET6 for IPv6).
packed_ip: The packed binary IP address.
Returns:
The IP address in string form.
Raises:
socket.error: If the conversion fails.
"""
addr = sockaddr() # 创建 sockaddr 实例
addr.sa_family = address_family # 设置地址族
addr_size = ctypes.c_int(ctypes.sizeof(addr)) # 地址结构的大小
ip_string = ctypes.create_string_buffer(128) # 用于存储 IP 字符串
ip_string_size = ctypes.c_int(ctypes.sizeof(ip_string)) # 字符串大小
# 根据地址族填充 sockaddr 结构
if address_family == socket.AF_INET:
if len(packed_ip) != ctypes.sizeof(addr.ipv4_addr):
raise socket.error('packed IP wrong length for inet_ntoa')
ctypes.memmove(addr.ipv4_addr, packed_ip, 4)
raise socket.error('packed IP wrong length for inet_ntoa') # IPv4 地址长度错误
ctypes.memmove(addr.ipv4_addr, packed_ip, 4) # 复制 IPv4 地址
elif address_family == socket.AF_INET6:
if len(packed_ip) != ctypes.sizeof(addr.ipv6_addr):
raise socket.error('packed IP wrong length for inet_ntoa')
ctypes.memmove(addr.ipv6_addr, packed_ip, 16)
raise socket.error('packed IP wrong length for inet_ntoa') # IPv6 地址长度错误
ctypes.memmove(addr.ipv6_addr, packed_ip, 16) # 复制 IPv6 地址
else:
raise socket.error('unknown address family')
raise socket.error('unknown address family') # 抛出未知地址族的异常
# 调用 WSAAddressToStringA 将二进制地址转换为字符串
if WSAAddressToStringA(
ctypes.byref(addr),
addr_size,
@ -75,11 +112,12 @@ def inet_ntop(address_family, packed_ip):
ip_string,
ctypes.byref(ip_string_size)
) != 0:
raise socket.error(ctypes.FormatError())
raise socket.error(ctypes.FormatError()) # 转换失败时抛出异常
return ip_string[:ip_string_size.value - 1] # 返回字符串表示形式，移除尾部的空字符
return ip_string[:ip_string_size.value - 1]
# Adding our two functions to the socket library
# 如果在 Windows 平台，将自定义的函数添加到 socket 库中
if os.name == 'nt':
socket.inet_pton = inet_pton
socket.inet_ntop = inet_ntop
socket.inet_pton = inet_pton # 添加 inet_pton 函数
socket.inet_ntop = inet_ntop # 添加 inet_ntop 函数
Write Preview
Loading…
Cancel
Save