用Python写一个UDP端口测试工具（一）
需求
最近有个运维需求，需要测试客户端的UDP端口与服务器的连通性。
需求也很简单：客户端测往服务端发UDP包，服务端收到包后响应客户端，当客户端能收到服务端的响应则可断定端口是可达的。但是客户端需要测试的端口范有很多，是个很大的端口范围。
思路
虽然nc等工具可以测试端口，但是面对多端口测试场景，就显得捉襟见肘了，因此就想到使用Python的socket编程来自己写一个工具来实现这个功能。
具体思路如下：
- 由客户端指定本地端口范围，也可以不指定，不指定则交由系统使用随机端口
- 当指定了本地端口，循环结束后停止运行，并输出统计结果
- 服务端为固定端口
- 如果需要服务端也需要变更为端口范围，需要有一个额外的线程来控制协商端口号，因为涉及到系统可能会存在端口被占用、端口不通等异常情况需要处理，因此简单的一个循环不能解决问题，可以参考下一篇用Python写一个UDP端口测试工具（二）
show you the code
服务端
#!/usr/bin/env python
from __future__ import print_function
import socket
import sys
import signal
import os
def h():
print(""" usage:""")
print(""" this_program <listen_port>""")
print(""" listen_port is not specified, default is 4000""")
print()
print(" examples:")
print(" ./udpserver.py 4000")
print()
def signal_handler(signal, frame):
os._exit(0)
signal.signal(signal.SIGINT, signal_handler)
if len(sys.argv)>1:
try:
port = int(sys.argv[1])
except:
h()
exit()
else:
port = 4000
sock = socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
sock.bind(('0.0.0.0', port))
while True:
recv_data,addr = sock.recvfrom(65536)
sock.sendto(recv_data, addr)
客户端
#!/usr/bin/env python
from __future__ import print_function
import socket
import sys
import time
import string
import random
import signal
import sys
import os
INTERVAL = 1000 #unit ms
LEN =64
IP=""
PORT=0
SRC_PORT_RANGE=False
count=0
count_of_received=0
rtt_sum=0.0
rtt_min=99999999.0
rtt_max=0.0
def signal_handler(signal, frame):
if count!=0 and count_of_received!=0:
print('')
print('--- ping statistics ---')
if count!=0:
print('%d packets transmitted, %d received, %.2f%% packet loss'%(count,count_of_received, (count-count_of_received)*100.0/count))
if count_of_received!=0:
print('rtt min/avg/max = %.2f/%.2f/%.2f ms'%(rtt_min,rtt_sum/count_of_received,rtt_max))
os._exit(0)
def random_string(length):
return ''.join(random.choice(string.ascii_letters+ string.digits ) for m in range(length))
def h():
print(""" usage:""")
print(""" this_program <dest_ip> <dest_port>""")
print(""" this_program <dest_ip> <dest_port> "<options>" """)
print()
print(""" options:""")
print(""" LEN the length of payload, unit:byte""")
print(""" INTERVAL the seconds waited between sending each packet, as well as the timeout for reply packet, unit: ms""")
print(""" SRC_PORT_RANGE the source port range, will be stoped when end of loop""")
print()
print(" examples:")
print(" ./udpclient.py 44.55.66.77 4000")
print(' ./udpclient.py 44.55.66.77 4000 "LEN=400;INTERVAL=2000;SRC_PORT_RANGE=\'20000:30000\'"')
print(" ./udpclient.py fe80::5400:ff:aabb:ccdd 4000")
print()
if len(sys.argv) != 3 and len(sys.argv)!=4 :
h()
exit()
IP=socket.gethostbyname(sys.argv[1])
PORT=int(sys.argv[2])
is_ipv6=0;
if IP.find(":")!=-1:
is_ipv6=1;
if len(sys.argv)==4:
try:
exec(sys.argv[3])
except:
h()
exit()
signal.signal(signal.SIGINT, signal_handler)
print("UDPping %s via port %d with %d bytes of payload"% (IP,PORT,LEN))
sys.stdout.flush()
if SRC_PORT_RANGE:
start, end = SRC_PORT_RANGE.split(':')
start = int(start)
end = int(end)
while True:
sock = socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
if SRC_PORT_RANGE:
if start > end:
signal_handler(None,None)
break
try:
sock.bind(('0.0.0.0', start))
except:
print("%s already in use" % start)
start += 1
continue
start += 1
payload= random_string(LEN)
sock.sendto(payload.encode(), (IP, PORT))
time_of_send=time.time()
deadline = time.time() + INTERVAL/1000.0
received=0
rtt=0.0
retrans=0
timeout=0.5
while True:
if retrans >= 3:
print("%s packet loss, retrans %s times, " % (time.strftime("%Y-%m-%d %H:%M:%S"), retrans),"%s:%s" % sock.getsockname(), "-> %s:%s" % (IP, PORT))
sys.stdout.flush()
break
sock.settimeout(timeout)
try:
recv_data,addr = sock.recvfrom(65536)
if recv_data== payload.encode() and addr[1]==PORT:
rtt=((time.time()-time_of_send)*1000)
print("%s:%s"%sock.getsockname(),"-> %s:%s"%(IP, PORT),"Reply from","%s:%s"%(IP,PORT),"seq=%d"%count, "time=%.2f"%(rtt),"ms")
sys.stdout.flush()
received=1
break
except socket.timeout:
sock.sendto(payload.encode(), (IP, PORT))
retrans+=1
timeout=timeout*2
except Exception as e:
pass
count+= 1
if received==1:
count_of_received+=1
rtt_sum+=rtt
rtt_max=max(rtt_max,rtt)
rtt_min=min(rtt_min,rtt)
else:
#print("Request timed out")
sys.stdout.flush()
time_remaining=deadline-time.time()
if(time_remaining>0):
time.sleep(time_remaining)