Support IPv6 & Add tool recv.py

Author: WillowSauceR

README.md

@@ -5,21 +5,21 @@
 
 ## 直接在 [Releases](https://github.com/WillowSauceR/BedrockProtocolTools/releases/latest) 页面下载最新的版本，然后解压打开其中的exe文件进行使用
 
-## 或者使用 ``git clone https://github.com/WillowSauceR/BedrockProtocolTools``将本项目克隆到本地或直接下载本项目，然后使用python3来运
+## 或者使用 ``git clone https://github.com/WillowSauceR/BedrockProtocolTools``将本项目克隆到本地或直接下载本项目，然后使用Python3来运行
 
 ## 行每个工具都可以单独使用参数 ``-h``来获取英文帮助，如 ``python3 scan.py -h``
 
 ## scan.py
 
-### 使用方法: ``python3 scan.py [目标地址] -i [可选：发包间隔] -p [可选：本地端口] -do [可选：只显示玩家在线数量大于或等于此数值的服务器] -e [在扫描到服务器后立刻执行的cmd命令，需要用双引号包含，具体使用请使用 -h 参数获取帮助]``
+### 使用方法: ``python3 scan.py [目标地址] -i [可选：发包间隔] -p [可选：本地端口] -do [可选：只显示玩家在线数量大于或等于此数值的服务器] -e [在扫描到服务器后立刻执行的cmd命令，需要用双引号包含，具体使用请使用 -h 参数获取帮助] -v6 [使用IPv6协议]``
 
 #### 描述: 扫描IP上的所有BE协议服务器
 
 #### 注意：目标地址可以填域名，IP或者IP段。如 mc.163.com，11.4.5.14，191.191.81-255.0-255
 
 ## send.py
 
-### 使用方法: ``python3 send.py [目标地址] [载体包文件] -p [可选：端口：默认19132] -l [可选：次数，默认1] -i [可选：间隔:秒，默认10] -d(自动显示MOTD) -pu(使用代理) -pc [代理的国家，如cn, ru, us] -q [安静模式，不输出任何日志]``
+### 使用方法: ``python3 send.py [目标地址] [载体包文件] -p [可选：端口：默认19132] -l [可选：次数，默认1] -i [可选：间隔:秒，默认10] -d(自动显示MOTD) -pu(使用代理) -pc [代理的国家，如cn, ru, us] -v6 [使用IPv6协议] -q [安静模式，不输出任何日志]``
 
 #### 描述: 发包复现工具, 需要一个内含byte数据的文件，一般为.dmp后缀名
 
@@ -32,3 +32,9 @@
 ### 使用方法: ``python3 motd.py [目标地址] [端口] -t [可选：超时时间]``
 
 #### 描述: motd一个BE服务器，支持自动解析返回的数据
+
+## recv.py
+
+### 使用方法：``python3 recv.py -i [可选：本地地址] -p [可选：本地端口] -v6 [可选：使用IPv6协议] -de [可选：关闭错误显示]``
+
+#### 描述：在指定的地址和端口上接受UDP数据包并打印其原始数据到控制台

api.py

@@ -1,6 +1,5 @@
 import socket
 import time
-import sys
 import random
 
 MOTD_PKT = b'\x01\x00\x00\x00\x00$\r\x12\xd3\x00\xff\xff\x00\xfe\xfe\xfe\xfe\xfd\xfd\xfd\xfd\x124Vx\n'
@@ -25,17 +24,13 @@ def log(*content, level: str = "INFO", info: str = "", quiet: bool = False):
     if quiet:
         return
     date = time.strftime('%H:%M:%S')
-    strs = ""
-    for string in content:
-        strs += str(string)
-    content = strs[0:]
     if info != "":
-        print(f"[{date} {info}] {content}")
+        print(f"[{date} {info}] ", *content, sep="")
     else:
-        print(f"[{date}] {content}")
+        print(f"[{date}] ", *content, sep="")
 
 
-def get_ip_list(addr: str):
+def get_ip_list(addr: str) -> list:
     # ip = "42.186.0-255.0-255" -> len(ip_list) == 65536
     try:
         int(addr[-1])
@@ -65,7 +60,7 @@ def get_ip_list(addr: str):
     return ip_list
 
 
-def decode_unicode(string: str):
+def decode_unicode(string: str) -> str:
     is_special_unicode = False
     for index in range(0, len(string), 5):
         if string[index] != "u":
@@ -85,14 +80,28 @@ def decode_unicode(string: str):
         return string.encode("utf-8").decode("unicode-escape")
 
 
-def get_udp_socket(port=random.randint(1024, 65535), timeout: int=1) -> socket.socket:
-    if not port:
-        port = random.randint(1024, 65535)
-    sockets = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-    sockets.settimeout(timeout)
-    sockets.bind(("", port))
+def get_udp_socket(loc_port=random.randint(1024, 65535), timeout: int=1, use_ipv6: bool=False, loc_addr: str="") -> socket.socket:
+    if not loc_port:
+        loc_port = random.randint(1024, 65535)
+    if use_ipv6:
+        sockets = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
+    else:
+        sockets = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    if timeout:
+        sockets.settimeout(timeout)
+    sockets.bind((loc_addr, loc_port))
     return sockets
 
+def is_ipv6_addr(addr: str) -> bool:
+    try:
+        addr = socket.gethostbyname(addr)
+    except socket.gaierror:
+        pass
+    try:
+        socket.inet_pton(socket.AF_INET6, addr)
+        return True
+    except socket.error:
+        return False
 
 def parse_raw_pkt(pkt):
     server_data, addr = pkt

motd.py

@@ -1,10 +1,10 @@
 import argparse
 import socket
-from api import get_udp_socket, log, parse_raw_pkt, MOTD_PKT
+from api import get_udp_socket, is_ipv6_addr, log, parse_raw_pkt, MOTD_PKT
 
 
 def send_pkt(addr, port, timeout: float=3.0, local_port: int=None):
-    udp_skt = get_udp_socket(local_port, timeout)
+    udp_skt = get_udp_socket(local_port, timeout, is_ipv6_addr(addr))
     udp_skt.sendto(
         MOTD_PKT,
         (addr, port))

recv.py

@@ -0,0 +1,40 @@
+import argparse
+from threading import Thread
+from api import get_udp_socket, log
+
+
+def recv_pkt(loc_addr, loc_port, disable_errors=False):
+    udp_skt = get_udp_socket(loc_port=loc_port, timeout=0, use_ipv6=use_ipv6, loc_addr=loc_addr)
+    err_count = 0
+    while True:
+        try:
+            data, addr = udp_skt.recvfrom(10240)
+            log(data, info=addr)
+        except Exception as exc:
+            err_count += 1
+            log(exc, level="ERROR", info=err_count, quiet=disable_errors)
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("-i", "--address", type=str,
+                        default="", help="local address to bind")
+    parser.add_argument("-p", "--port", type=int,
+                        default=19132, help="local port to bind")
+    parser.add_argument("-v6", "--use-ipv6", action="store_true",
+                        default=False, help="use IPv6 instead of IPv4")
+    parser.add_argument("-de", "--disable-errors", action="store_true",
+                        default=False, help="disable error message output")
+    args = parser.parse_args()
+
+    loc_addr = args.address
+    loc_port = args.port
+    use_ipv6 = args.use_ipv6
+    disable_errors = args.disable_errors
+
+    log(f"Starting listening {loc_addr}:{loc_port}, type q to exit.")
+    Thread(target=recv_pkt, args=(loc_addr, loc_port, disable_errors), daemon=True).start()
+    while True:
+        if input() == "q":
+            log("Exit...")
+            exit()

scan.py

@@ -137,6 +137,8 @@ def exec_cmd_async(cmd: str, variables: dict = [None]) -> None:
                         help="the cmd command that is executed immediately after scanning the server can use {var} as a variable, "
                         + "the command needs to be enclosed in double quotes, the available variables are: motd, version_id, "
                         + "version, online, max_player, unique_id, map, gamemode, source_port_v4, source_port_v6, addr and ip")
+    parser.add_argument("-v6", "--use-ipv6", action="store_true", default=False, 
+                        help="use IPv6 instead of IPv4")
 
     args, unparsed = parser.parse_known_args()
 
@@ -145,8 +147,9 @@ def exec_cmd_async(cmd: str, variables: dict = [None]) -> None:
     local_port = args.port
     display_online = args.display_online
     exec_cmd = args.exec_cmd
+    use_ipv6 = args.use_ipv6
 
-    udp_skt = get_udp_socket(local_port)
+    udp_skt = get_udp_socket(local_port, use_ipv6=use_ipv6)
 
     for addr in get_ip_list(addr):
         scanner(udp_skt, addr, interval)

send.py

@@ -6,7 +6,7 @@
 import argparse
 
 from motd import send_pkt as send_motd_pkt
-from api import get_udp_socket, log
+from api import get_udp_socket, is_ipv6_addr, log
 
 
 def get_proxy() -> list:
@@ -42,7 +42,7 @@ def create_socket():
         if proxy_used:
             log(f"Used proxy: {proxy_addr}:{proxy_port}")
     else:
-        udp_skt = get_udp_socket(local_port)
+        udp_skt = get_udp_socket(local_port, 1, use_ipv6)
     return local_port, udp_skt
 
 
@@ -91,6 +91,7 @@ def send_pkt(target_addr: str, port: int, payload_file: str, loops: int, interva
     parser.add_argument("-d", "--display_motd", action="store_true", default=False, help="Display Motd")
     parser.add_argument("-pu", "--proxy_used", action="store_true", default=False, help="Use Proxy")
     parser.add_argument("-pc", "--proxy_country", type=str, default="us", help="ProxyCountry (like cn, ru, us or enter to use all)")
+    parser.add_argument("-v6", "--use-ipv6", action="store_true", default=False, help="use IPv6 instead of IPv4")
     args = parser.parse_args()
 
     target = args.target
@@ -102,6 +103,7 @@ def send_pkt(target_addr: str, port: int, payload_file: str, loops: int, interva
     quiet_mode = args.quiet
     proxy_used = args.proxy_used
     proxyCountry = args.proxy_country
+    use_ipv6 = args.use_ipv6
 
     if proxy_used:
         try:
@@ -110,7 +112,10 @@ def send_pkt(target_addr: str, port: int, payload_file: str, loops: int, interva
             import requests
         except Exception as error:
             print(error)
-            log("Import module error! Please run \"pip install -r requirements.txt\"")
+            log("Import module error! Run \"pip install -r requirements.txt\" to install dependent modules? [y/n] ")
+            if input() == "y":
+                os.system("pip install -r requirements.txt")
+                log("Success. Please restart this program!")
             os._exit(1)
         log("Proxy mode is under development!")