mirror of
https://github.com/jumpserver/jumpserver.git
synced 2025-09-26 07:22:27 +00:00
feat: 系统工具增加服务器时间及nmap工具 (#11078)
This commit is contained in:
@@ -1,7 +1,8 @@
|
||||
# coding: utf-8
|
||||
#
|
||||
#
|
||||
|
||||
from .ldap import *
|
||||
from .common import *
|
||||
from .ping import *
|
||||
from .telnet import *
|
||||
from .nmap import *
|
||||
|
60
apps/settings/utils/nmap.py
Normal file
60
apps/settings/utils/nmap.py
Normal file
@@ -0,0 +1,60 @@
|
||||
import time
|
||||
import nmap
|
||||
|
||||
from IPy import IP
|
||||
|
||||
from common.utils.timezone import local_now_display
|
||||
|
||||
|
||||
def generate_ips(ip_string):
|
||||
# 支持的格式
|
||||
# 192.168.1.1-12 | 192.168.1.1-192.168.1.12 | 192.168.1.0/30 | 192.168.1.1
|
||||
ip_list = ip_string.split('-')
|
||||
ips = []
|
||||
try:
|
||||
if len(ip_list) == 2:
|
||||
start_ip, end_ip = ip_list
|
||||
if ip_list[1].find('.') == -1:
|
||||
end_ip = start_ip[:start_ip.rindex('.') + 1] + end_ip
|
||||
for ip in range(IP(start_ip).int(), IP(end_ip).int() + 1):
|
||||
ips.extend(IP(ip))
|
||||
else:
|
||||
ips.extend(IP(ip_list[0]))
|
||||
except Exception:
|
||||
ips = []
|
||||
return ips
|
||||
|
||||
|
||||
def once_nmap(nm, ip, ports, timeout, display):
|
||||
nmap_version = '.'.join(map(lambda x: str(x), nm.nmap_version()))
|
||||
display(f'Starting Nmap {nmap_version} at {local_now_display()} for {ip}')
|
||||
try:
|
||||
is_ok = True
|
||||
nm.scan(ip, arguments='-sS -sU -F', ports=ports, timeout=timeout)
|
||||
tcp_port = nm[ip].get('tcp', {})
|
||||
udp_port = nm[ip].get('udp', {})
|
||||
display(f'PORT\tSTATE\tSERVICE')
|
||||
for port, info in tcp_port.items():
|
||||
display(f"{port}\t{info.get('state', 'unknown')}\t{info.get('name', 'unknown')}")
|
||||
for port, info in udp_port.items():
|
||||
display(f"{port}\t{info.get('state', 'unknown')}\t{info.get('name', 'unknown')}")
|
||||
except Exception:
|
||||
is_ok = False
|
||||
display(f'Nmap scan report for {ip} error.')
|
||||
return is_ok
|
||||
|
||||
|
||||
def verbose_nmap(dest_ip, dest_port=None, timeout=None, display=print):
|
||||
dest_port = ','.join(list(dest_port)) if dest_port else None
|
||||
|
||||
ips = generate_ips(dest_ip)
|
||||
nm = nmap.PortScanner()
|
||||
success_num, start_time = 0, time.time()
|
||||
display(f'[Summary] Nmap: {len(ips)} IP addresses were scanned')
|
||||
for ip in ips:
|
||||
ok = once_nmap(nm, str(ip), dest_port, timeout, display)
|
||||
if ok:
|
||||
success_num += 1
|
||||
display('')
|
||||
display(f'[Done] Nmap: {len(ips)} IP addresses ({success_num} hosts up) '
|
||||
f'scanned in {round(time.time() - start_time, 2)} seconds')
|
@@ -128,30 +128,37 @@ def ping(dest_addr, timeout, psize, flag=0):
|
||||
return delay
|
||||
|
||||
|
||||
def verbose_ping(dest_addr, timeout=2, count=5, psize=64, display=None):
|
||||
def verbose_ping(dest_ip, timeout=2, count=5, psize=64, display=None):
|
||||
"""
|
||||
Send `count' ping with `psize' size to `dest_addr' with
|
||||
the given `timeout' and display the result.
|
||||
"""
|
||||
ip = lookup_domain(dest_addr)
|
||||
ip = lookup_domain(dest_ip)
|
||||
if not ip:
|
||||
return
|
||||
if display is None:
|
||||
display = print
|
||||
display("PING %s (%s): 56 data bytes" % (dest_addr, ip))
|
||||
error_count = 0
|
||||
display("PING %s (%s): 56 data bytes" % (dest_ip, ip))
|
||||
for i in range(count):
|
||||
try:
|
||||
delay = ping(dest_addr, timeout, psize)
|
||||
delay = ping(dest_ip, timeout, psize)
|
||||
except socket.gaierror as e:
|
||||
display("failed. (socket error: '%s')" % str(e))
|
||||
error_count += 1
|
||||
break
|
||||
|
||||
if delay is None:
|
||||
display("Request timeout for icmp_seq %i" % i)
|
||||
error_count += 1
|
||||
else:
|
||||
delay = delay * 1000
|
||||
delay *= 1000
|
||||
display("64 bytes from %s: icmp_seq=0 ttl=115 time=%.3f ms" % (ip, delay))
|
||||
time.sleep(1)
|
||||
display(f'--- {dest_ip} ping statistics ---')
|
||||
display(f'{count} packets transmitted, '
|
||||
f'{count - error_count} packets received, '
|
||||
f'{(error_count / count) * 100}% packet loss')
|
||||
print()
|
||||
|
||||
|
||||
|
@@ -18,21 +18,21 @@ def telnet(dest_addr, port_number=23, timeout=10):
|
||||
return True, output.decode('utf-8', 'ignore')
|
||||
|
||||
|
||||
def verbose_telnet(dest_addr, port_number=23, timeout=10, display=None):
|
||||
def verbose_telnet(dest_ip, dest_port=23, timeout=10, display=None):
|
||||
if display is None:
|
||||
display = print
|
||||
ip = lookup_domain(dest_addr)
|
||||
ip = lookup_domain(dest_ip)
|
||||
if not ip:
|
||||
return
|
||||
msg = 'Trying %s (%s:%s)' % (dest_addr, ip, port_number)
|
||||
msg = 'Trying %s (%s:%s)' % (dest_ip, ip, dest_port)
|
||||
display(msg)
|
||||
try:
|
||||
is_connective, resp = telnet(dest_addr, port_number, timeout)
|
||||
is_connective, resp = telnet(dest_ip, dest_port, timeout)
|
||||
if is_connective:
|
||||
template = 'Connected to {0} {1}.\r\n{2}Connection closed by foreign host.'
|
||||
else:
|
||||
template = 'telnet: connect to {0} {1} {2}\r\ntelnet: Unable to connect to remote host'
|
||||
msg = template.format(dest_addr, port_number, resp)
|
||||
msg = template.format(dest_ip, dest_port, resp)
|
||||
except Exception as e:
|
||||
msg = 'Error: %s' % e
|
||||
display(msg)
|
||||
|
Reference in New Issue
Block a user