python/jc
1
0
Fork 0
mirror of https://github.com/kellyjonbrazil/jc.git synced 2025-06-19 00:17:51 +02:00
Code Issues Releases Activity

Add error type support for Linux Ping (#575)

Browse Source 
* feat: add icmp error handle into linux_parse

* refactor: fixed timestamp offset logic(including error-response-type condition)
This commit is contained in:
Mabuchin
2024-09-08 11:46:43 +09:00
committed by GitHub
parent 8a22f8a468
commit 1593d0bf79
3 changed files with 83 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

102
jc/parsers/ping_s.py
View File

@ -169,7 +169,7 @@ def _ipv6_in(line):
return ipv6
def _error_type(line):
def _error_type_v4(line):
# from https://github.com/dgibson/iputils/blob/master/ping.c
# https://android.googlesource.com/platform/external/ping/+/8fc3c91cf9e7f87bc20b9e6d3ea2982d87b70d9a/ping.c
# https://opensource.apple.com/source/network_cmds/network_cmds-328/ping.tproj/ping.c
@ -207,6 +207,37 @@ def _error_type(line):
return None
def _error_type_v6(line):
type_map = {
'Destination unreachable': 'destination_unreachable',
'Packet too big': 'packet_too_big',
'Time exceeded:': 'time_exceeded',
'Parameter problem:': 'parameter_problem',
}
code_map = {
'destination_unreachable': {
'No route': 'no_route',
'Administratively prohibited': 'administratively_prohibited',
"Beyond scope of source address": 'beyond_scope_of_source_address',
'Address unreachable': 'address_unreachable',
'Port unreachable': 'port_unreachable',
},
'time_exceeded': {
'Hop limit': 'hop_limit',
'Fragment reassembly time exceeded': 'fragment_reassembly_time_exceeded',
},
}
return_code = None
for err_type, code in type_map.items():
if err_type in line:
return_code = code
for err_code, code_name in code_map[code].items():
if err_code in line:
return_code += '_' + code_name
return return_code
def _bsd_parse(line, s):
output_line = {}
@ -263,6 +294,24 @@ def _bsd_parse(line, s):
# ping response lines
err = None
if s.ipv4:
err = _error_type_v4(line)
else:
err = _error_type_v6(line)
if err:
output_line = {
'type': err
}
try:
output_line['sent_bytes'] = line.split()[0]
output_line['destination_ip'] = s.destination_ip
output_line['response_ip'] = line.split()[4].strip(':').strip('(').strip(')')
except Exception:
pass
return output_line
# ipv4 lines
if not _ipv6_in(line):
@ -279,7 +328,7 @@ def _bsd_parse(line, s):
return output_line
# catch error responses
err = _error_type(line)
err = _error_type_v4(line)
if err:
output_line = {
'type': err
@ -444,25 +493,40 @@ def _linux_parse(line, s):
}
return output_line
# if timestamp option is specified, then shift icmp sequence field right by one
timestamp = False
if line[0] == '[':
timestamp = True
timestamp_offset = 1 if timestamp else 0
# ping response lines
err = None
if s.ipv4:
err = _error_type_v4(line)
else:
err = _error_type_v6(line)
if err:
output_line = {
'type': err,
'destination_ip': s.destination_ip or None,
'sent_bytes': s.sent_bytes or None,
'response_ip': line.split()[timestamp_offset + 1] if type != 'timeout' else None,
'icmp_seq': line.replace('=', ' ').split()[timestamp_offset + 3],
'timestamp': line.split()[0].lstrip('[').rstrip(']') if timestamp else None,
}
return output_line
# request timeout
if 'no answer yet for icmp_seq=' in line:
timestamp = False
isequence = 5
# if timestamp option is specified, then shift icmp sequence field right by one
if line[0] == '[':
timestamp = True
isequence = 6
output_line = {
'type': 'timeout',
'destination_ip': s.destination_ip or None,
'sent_bytes': s.sent_bytes or None,
'pattern': s.pattern or None,
'timestamp': line.split()[0].lstrip('[').rstrip(']') if timestamp else None,
'icmp_seq': line.replace('=', ' ').split()[isequence]
'icmp_seq': line.replace('=', ' ').split()[timestamp_offset + 5]
}
return output_line
@ -473,20 +537,16 @@ def _linux_parse(line, s):
line = line.replace('(', ' ').replace(')', ' ').replace('=', ' ')
# positions of items depend on whether ipv4/ipv6 and/or ip/hostname is used
param_positions = None
if s.ipv4 and not s.hostname:
bts, rip, iseq, t2l, tms = (0, 3, 5, 7, 9)
param_positions = (0, 3, 5, 7, 9)
elif s.ipv4 and s.hostname:
bts, rip, iseq, t2l, tms = (0, 4, 7, 9, 11)
param_positions = (0, 4, 7, 9, 11)
elif not s.ipv4 and not s.hostname:
bts, rip, iseq, t2l, tms = (0, 3, 5, 7, 9)
param_positions = (0, 3, 5, 7, 9)
elif not s.ipv4 and s.hostname:
bts, rip, iseq, t2l, tms = (0, 4, 7, 9, 11)
# if timestamp option is specified, then shift everything right by one
timestamp = False
if line[0] == '[':
timestamp = True
bts, rip, iseq, t2l, tms = (bts + 1, rip + 1, iseq + 1, t2l + 1, tms + 1)
param_positions = (0, 4, 7, 9, 11)
bts, rip, iseq, t2l, tms = (x + timestamp_offset for x in param_positions)
output_line = {
'type': 'reply',

2
tests/fixtures/ubuntu-22.04/ping-dest-unreachable-streaming.json vendored
View File

@ -1 +1 @@
{"type":"summary","destination_ip":"10.0.3.22","sent_bytes":56,"pattern":null,"packets_transmitted":3,"packets_received":0,"packet_loss_percent":100.0,"duplicates":0,"errors":3,"corrupted":null,"time_ms":2049.0,"round_trip_ms_min":null,"round_trip_ms_avg":null,"round_trip_ms_max":null,"round_trip_ms_stddev":null}
[{"type": "destination_host_unreachable", "destination_ip": "10.0.3.22", "sent_bytes": 56, "response_ip": "10.0.0.1", "icmp_seq": 1, "timestamp": null}, {"type": "destination_host_unreachable", "destination_ip": "10.0.3.22", "sent_bytes": 56, "response_ip": "10.0.0.1", "icmp_seq": 2, "timestamp": null}, {"type": "destination_host_unreachable", "destination_ip": "10.0.3.22", "sent_bytes": 56, "response_ip": "10.0.0.1", "icmp_seq": 3, "timestamp": null}, {"type": "summary", "destination_ip": "10.0.3.22", "sent_bytes": 56, "pattern": null, "packets_transmitted": 3, "packets_received": 0, "packet_loss_percent": 100.0, "duplicates": 0, "errors": 3, "corrupted": null, "time_ms": 2049.0, "round_trip_ms_min": null, "round_trip_ms_avg": null, "round_trip_ms_max": null, "round_trip_ms_stddev": null}]

2
tests/test_ping_s.py
View File

@ -597,7 +597,7 @@ class MyTests(unittest.TestCase):
"""
Test 'ping' on Ubuntu 22.4 with destination unreachable message
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.ubuntu_22_4_ping_dest_unreachable.splitlines(), quiet=True)), [self.ubuntu_22_4_ping_dest_unreachable_streaming_json])
self.assertEqual(list(jc.parsers.ping_s.parse(self.ubuntu_22_4_ping_dest_unreachable.splitlines(), quiet=True)), self.ubuntu_22_4_ping_dest_unreachable_streaming_json)
def test_ping_s_hostname_I_ubuntu_22_4(self):