diff --git a/src/op_mode/ping.py b/src/op_mode/ping.py
index 934cfdb52..f1d87a118 100755
--- a/src/op_mode/ping.py
+++ b/src/op_mode/ping.py
@@ -1,290 +1,300 @@
#! /usr/bin/env python3
# Copyright (C) 2020 VyOS maintainers and contributors
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or later as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import os
import sys
import socket
import ipaddress
from vyos.utils.network import get_all_vrfs
from vyos.ifconfig import Section
def interface_list() -> list:
"""
Get list of interfaces in system
:rtype: list
"""
return Section.interfaces()
def vrf_list() -> list:
"""
Get list of VRFs in system
:rtype: list
"""
return list(get_all_vrfs().keys())
options = {
'audible': {
'ping': '{command} -a',
'type': 'noarg',
'help': 'Make a noise on ping'
},
'adaptive': {
'ping': '{command} -A',
'type': 'noarg',
'help': 'Adativly set interpacket interval'
},
'allow-broadcast': {
'ping': '{command} -b',
'type': 'noarg',
'help': 'Ping broadcast address'
},
'bypass-route': {
'ping': '{command} -r',
'type': 'noarg',
'help': 'Bypass normal routing tables'
},
'count': {
'ping': '{command} -c {value}',
'type': '<requests>',
'help': 'Number of requests to send'
},
'deadline': {
'ping': '{command} -w {value}',
'type': '<seconds>',
'help': 'Number of seconds before ping exits'
},
'do-not-fragment': {
'ping': '{command} -M do',
'type': 'noarg',
'help': 'Set DF-bit flag to 1 for no fragmentation'
},
'flood': {
'ping': 'sudo {command} -f',
'type': 'noarg',
'help': 'Send 100 requests per second'
},
'interface': {
'ping': '{command} -I {value}',
'type': '<interface>',
'helpfunction': interface_list,
'help': 'Source interface'
},
'interval': {
'ping': '{command} -i {value}',
'type': '<seconds>',
'help': 'Number of seconds to wait between requests'
},
+ 'ipv4': {
+ 'ping': '{command} -4',
+ 'type': 'noarg',
+ 'help': 'Use IPv4 only'
+ },
+ 'ipv6': {
+ 'ping': '{command} -6',
+ 'type': 'noarg',
+ 'help': 'Use IPv6 only'
+ },
'mark': {
'ping': '{command} -m {value}',
'type': '<fwmark>',
'help': 'Mark request for special processing'
},
'numeric': {
'ping': '{command} -n',
'type': 'noarg',
'help': 'Do not resolve DNS names'
},
'no-loopback': {
'ping': '{command} -L',
'type': 'noarg',
'help': 'Supress loopback of multicast pings'
},
'pattern': {
'ping': '{command} -p {value}',
'type': '<pattern>',
'help': 'Pattern to fill out the packet'
},
'timestamp': {
'ping': '{command} -D',
'type': 'noarg',
'help': 'Print timestamp of output'
},
'tos': {
'ping': '{command} -Q {value}',
'type': '<tos>',
'help': 'Mark packets with specified TOS'
},
'quiet': {
'ping': '{command} -q',
'type': 'noarg',
'help': 'Only print summary lines'
},
'record-route': {
'ping': '{command} -R',
'type': 'noarg',
'help': 'Record route the packet takes'
},
'size': {
'ping': '{command} -s {value}',
'type': '<bytes>',
'help': 'Number of bytes to send'
},
'source-address': {
'ping': '{command} -I {value}',
'type': '<x.x.x.x> <h:h:h:h:h:h:h:h>',
},
'ttl': {
'ping': '{command} -t {value}',
'type': '<ttl>',
'help': 'Maximum packet lifetime'
},
'vrf': {
'ping': 'sudo ip vrf exec {value} {command}',
'type': '<vrf>',
'help': 'Use specified VRF table',
'helpfunction': vrf_list,
'dflt': 'default',
},
'verbose': {
'ping': '{command} -v',
'type': 'noarg',
'help': 'Verbose output'}
}
ping = {
4: '/bin/ping',
6: '/bin/ping6',
}
class List(list):
def first(self):
return self.pop(0) if self else ''
def last(self):
return self.pop() if self else ''
def prepend(self, value):
self.insert(0, value)
def completion_failure(option: str) -> None:
"""
Shows failure message after TAB when option is wrong
:param option: failure option
:type str:
"""
sys.stderr.write('\n\n Invalid option: {}\n\n'.format(option))
sys.stdout.write('<nocomps>')
sys.exit(1)
def expension_failure(option, completions):
reason = 'Ambiguous' if completions else 'Invalid'
sys.stderr.write(
'\n\n {} command: {} [{}]\n\n'.format(reason, ' '.join(sys.argv),
option))
if completions:
sys.stderr.write(' Possible completions:\n ')
sys.stderr.write('\n '.join(completions))
sys.stderr.write('\n')
sys.stdout.write('<nocomps>')
sys.exit(1)
def complete(prefix):
return [o for o in options if o.startswith(prefix)]
def convert(command, args):
while args:
shortname = args.first()
longnames = complete(shortname)
if len(longnames) != 1:
expension_failure(shortname, longnames)
longname = longnames[0]
if options[longname]['type'] == 'noarg':
command = options[longname]['ping'].format(
command=command, value='')
elif not args:
sys.exit(f'ping: missing argument for {longname} option')
else:
command = options[longname]['ping'].format(
command=command, value=args.first())
return command
if __name__ == '__main__':
args = List(sys.argv[1:])
host = args.first()
if not host:
sys.exit("ping: Missing host")
if host == '--get-options':
args.first() # pop ping
args.first() # pop IP
usedoptionslist = []
while args:
option = args.first() # pop option
matched = complete(option) # get option parameters
usedoptionslist.append(option) # list of used options
# Select options
if not args:
# remove from Possible completions used options
for o in usedoptionslist:
if o in matched:
matched.remove(o)
sys.stdout.write(' '.join(matched))
sys.exit(0)
if len(matched) > 1:
sys.stdout.write(' '.join(matched))
sys.exit(0)
# If option doesn't have value
if matched:
if options[matched[0]]['type'] == 'noarg':
continue
else:
# Unexpected option
completion_failure(option)
value = args.first() # pop option's value
if not args:
matched = complete(option)
helplines = options[matched[0]]['type']
# Run helpfunction to get list of possible values
if 'helpfunction' in options[matched[0]]:
result = options[matched[0]]['helpfunction']()
if result:
helplines = '\n' + ' '.join(result)
sys.stdout.write(helplines)
sys.exit(0)
for name, option in options.items():
if 'dflt' in option and name not in args:
args.append(name)
args.append(option['dflt'])
try:
ip = socket.gethostbyname(host)
except UnicodeError:
sys.exit(f'ping: Unknown host: {host}')
except socket.gaierror:
ip = host
try:
version = ipaddress.ip_address(ip).version
except ValueError:
sys.exit(f'ping: Unknown host: {host}')
command = convert(ping[version], args)
# print(f'{command} {host}')
os.system(f'{command} {host}')