feat: 新增ping、telnet系统工具 (#8666)

* feat: 新增ping、telnet系统工具

* perf: 消息返回

Co-authored-by: halo <wuyihuangw@gmail.com>

apps/settings/utils/__init__.py

@@ -3,3 +3,5 @@
 
 from .ldap import *
 from .common import *
+from .ping import *
+from .telnet import *

apps/settings/utils/ping.py

@@ -0,0 +1,154 @@
+# -*- coding: utf-8 -*-
+#
+
+import os
+import select
+import socket
+import struct
+import time
+
+# From /usr/include/linux/icmp.h; your milage may vary.
+ICMP_ECHO_REQUEST = 8  # Seems to be the same on Solaris.
+
+
+def checksum(source_string):
+    """
+    I'm not too confident that this is right but testing seems
+    to suggest that it gives the same answers as in_cksum in ping.c
+    """
+    sum = 0
+    count_to = int((len(source_string) / 2) * 2)
+    for count in range(0, count_to, 2):
+        this = source_string[count + 1] * 256 + source_string[count]
+        sum = sum + this
+        sum = sum & 0xffffffff  # Necessary?
+
+    if count_to < len(source_string):
+        sum = sum + ord(source_string[len(source_string) - 1])
+        sum = sum & 0xffffffff  # Necessary?
+
+    sum = (sum >> 16) + (sum & 0xffff)
+    sum = sum + (sum >> 16)
+    answer = ~sum
+    answer = answer & 0xffff
+
+    # Swap bytes. Bugger me if I know why.
+    answer = answer >> 8 | (answer << 8 & 0xff00)
+
+    return answer
+
+
+def receive_one_ping(my_socket, id, timeout):
+    """
+    Receive the ping from the socket.
+    """
+    time_left = timeout
+    while True:
+        started_select = time.time()
+        what_ready = select.select([my_socket], [], [], time_left)
+        how_long_in_select = time.time() - started_select
+        if not what_ready[0]:  # Timeout
+            return
+
+        time_received = time.time()
+        received_packet, addr = my_socket.recvfrom(1024)
+        icmpHeader = received_packet[20:28]
+        type, code, checksum, packet_id, sequence = struct.unpack("bbHHh", icmpHeader)
+        if packet_id == id:
+            bytes = struct.calcsize("d")
+            time_sent = struct.unpack("d", received_packet[28: 28 + bytes])[0]
+            return time_received - time_sent
+
+        time_left = time_left - how_long_in_select
+        if time_left <= 0:
+            return
+
+
+def send_one_ping(my_socket, dest_addr, id, psize):
+    """
+    Send one ping to the given >dest_addr<.
+    """
+    dest_addr = socket.gethostbyname(dest_addr)
+
+    # Remove header size from packet size
+    # psize = psize - 8
+    # laixintao edit:
+    # Do not need to remove header here. From BSD ping man:
+    #     The default is 56, which translates into 64 ICMP data
+    #     bytes when combined with the 8 bytes of ICMP header data.
+
+    # Header is type (8), code (8), checksum (16), id (16), sequence (16)
+    my_checksum = 0
+
+    # Make a dummy heder with a 0 checksum.
+    header = struct.pack("bbHHh", ICMP_ECHO_REQUEST, 0, my_checksum, id, 1)
+    bytes = struct.calcsize("d")
+    data = (psize - bytes) * b"Q"
+    data = struct.pack("d", time.time()) + data
+
+    # Calculate the checksum on the data and the dummy header.
+    my_checksum = checksum(header + data)
+
+    # Now that we have the right checksum, we put that in. It's just easier
+    # to make up a new header than to stuff it into the dummy.
+    header = struct.pack(
+        "bbHHh", ICMP_ECHO_REQUEST, 0, socket.htons(my_checksum), id, 1
+    )
+    packet = header + data
+    my_socket.sendto(packet, (dest_addr, 1))  # Don't know about the 1
+
+
+def ping(dest_addr, timeout, psize, flag=0):
+    """
+    Returns either the delay (in seconds) or none on timeout.
+    """
+    icmp = socket.getprotobyname("icmp")
+    try:
+        if os.getuid() != 0:
+            my_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, icmp)
+        else:
+            my_socket = socket.socket(socket.AF_INET, socket.SOCK_RAW, icmp)
+    except socket.error as e:
+        if e.errno == 1:
+            # Operation not permitted
+            msg = str(e)
+            raise socket.error(msg)
+        raise  # raise the original error
+
+    process_pre = os.getpid() & 0xFF00
+    flag = flag & 0x00FF
+    my_id = process_pre | flag
+
+    send_one_ping(my_socket, dest_addr, my_id, psize)
+    delay = receive_one_ping(my_socket, my_id, timeout)
+
+    my_socket.close()
+    return delay
+
+
+def verbose_ping(dest_addr, timeout=2, count=5, psize=64):
+    """
+    Send `count' ping with `psize' size to `dest_addr' with
+    the given `timeout' and display the result.
+    """
+    for i in range(count):
+        print("ping %s with ..." % dest_addr, end="")
+        try:
+            delay = ping(dest_addr, timeout, psize)
+        except socket.gaierror as e:
+            print("failed. (socket error: '%s')" % str(e))
+            break
+
+        if delay is None:
+            print("failed. (timeout within %ssec.)" % timeout)
+        else:
+            delay = delay * 1000
+            print("get ping in %0.4fms" % delay)
+    print()
+
+
+if __name__ == "__main__":
+    verbose_ping("google.com")
+    verbose_ping("192.168.4.1")
+    verbose_ping("www.baidu.com")
+    verbose_ping("sssssss")

apps/settings/utils/telnet.py

@@ -0,0 +1,25 @@
+# -*- coding: utf-8 -*-
+#
+import socket
+import telnetlib
+
+PROMPT_REGEX = r'[\<|\[](.*)[\>|\]]'
+
+
+def telnet(dest_addr, port_number=23, timeout=10):
+    try:
+        connection = telnetlib.Telnet(dest_addr, port_number, timeout)
+    except (ConnectionRefusedError, socket.timeout, socket.gaierror) as e:
+        return False, str(e)
+    expected_regexes = [bytes(PROMPT_REGEX, encoding='ascii')]
+    index, prompt_regex, output = connection.expect(expected_regexes, timeout=3)
+    return True, output.decode('ascii')
+
+
+if __name__ == "__main__":
+    print(telnet(dest_addr='1.1.1.1', port_number=2222))
+    print(telnet(dest_addr='baidu.com', port_number=80))
+    print(telnet(dest_addr='baidu.com', port_number=8080))
+    print(telnet(dest_addr='192.168.4.1', port_number=2222))
+    print(telnet(dest_addr='192.168.4.1', port_number=2223))
+    print(telnet(dest_addr='ssssss', port_number=-1))