python/jc
1
0
Fork 0
mirror of https://github.com/kellyjonbrazil/jc.git synced 2025-07-13 01:20:24 +02:00
Code Issues Releases Activity

linting

Browse Source
This commit is contained in:
Kelly Brazil
2021-09-24 09:16:44 -07:00
parent 486282b985
commit bbed9e274b
1 changed files with 147 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files

293
jc/parsers/ping_s.py
View File

@ -174,11 +174,12 @@ def _error_type(line):
if err_type in line: if err_type in line:
return code return code
return None
def _bsd_parse(line, s): def _bsd_parse(line, s):
output_line = {} output_line = {}
if line.startswith('PING '): if line.startswith('PING '):
s.destination_ip = line.split()[2].lstrip('(').rstrip(':').rstrip(')') s.destination_ip = line.split()[2].lstrip('(').rstrip(':').rstrip(')')
s.sent_bytes = line.split()[3] s.sent_bytes = line.split()[3]
@ -204,95 +205,75 @@ def _bsd_parse(line, s):
s.duplicates = line.split()[6].lstrip('+') s.duplicates = line.split()[6].lstrip('+')
return None return None
else: s.packets_transmitted = line.split()[0]
s.packets_transmitted = line.split()[0] s.packets_received = line.split()[3]
s.packets_received = line.split()[3] s.packet_loss_percent = line.split()[6].rstrip('%')
s.packet_loss_percent = line.split()[6].rstrip('%') s.duplicates = '0'
s.duplicates = '0' return None
return None
else: split_line = line.split(' = ')[1]
split_line = line.split(' = ')[1] split_line = split_line.split('/')
split_line = split_line.split('/')
output_line = {
'type': 'summary',
'destination_ip': s.destination_ip or None,
'sent_bytes': s.sent_bytes or None,
'pattern': s.pattern or None,
'packets_transmitted': s.packets_transmitted or None,
'packets_received': s.packets_received or None,
'packet_loss_percent': s.packet_loss_percent or None,
'duplicates': s.duplicates or None,
'round_trip_ms_min': split_line[0],
'round_trip_ms_avg': split_line[1],
'round_trip_ms_max': split_line[2],
'round_trip_ms_stddev': split_line[3].replace(' ms', '')
}
return output_line
# ping response lines
# ipv4 lines
if not _ipv6_in(line):
# request timeout
if line.startswith('Request timeout for '):
output_line = { output_line = {
'type': 'summary', 'type': 'timeout',
'destination_ip': s.destination_ip or None, 'destination_ip': s.destination_ip or None,
'sent_bytes': s.sent_bytes or None, 'sent_bytes': s.sent_bytes or None,
'pattern': s.pattern or None, 'pattern': s.pattern or None,
'packets_transmitted': s.packets_transmitted or None, 'icmp_seq': line.split()[4]
'packets_received': s.packets_received or None,
'packet_loss_percent': s.packet_loss_percent or None,
'duplicates': s.duplicates or None,
'round_trip_ms_min': split_line[0],
'round_trip_ms_avg': split_line[1],
'round_trip_ms_max': split_line[2],
'round_trip_ms_stddev': split_line[3].replace(' ms', '')
} }
return output_line return output_line
# ping response lines # catch error responses
else: err = _error_type(line)
# ipv4 lines if err:
if not _ipv6_in(line): output_line = {
'type': err
}
# request timeout try:
if line.startswith('Request timeout for '): output_line['bytes'] = line.split()[0]
output_line = { output_line['destination_ip'] = s.destination_ip
'type': 'timeout', output_line['response_ip'] = line.split()[4].strip(':').strip('(').strip(')')
'destination_ip': s.destination_ip or None, except Exception:
'sent_bytes': s.sent_bytes or None, pass
'pattern': s.pattern or None,
'icmp_seq': line.split()[4]
}
return output_line return output_line
# catch error responses # normal response
err = _error_type(line) if ' bytes from ' in line:
if err: line = line.replace(':', ' ').replace('=', ' ')
output_line = {
'type': err
}
try:
output_line['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
# normal response
elif ' bytes from ' in line:
line = line.replace(':', ' ').replace('=', ' ')
output_line = {
'type': 'reply',
'destination_ip': s.destination_ip or None,
'sent_bytes': s.sent_bytes or None,
'pattern': s.pattern or None,
'response_bytes': line.split()[0],
'response_ip': line.split()[3],
'icmp_seq': line.split()[5],
'ttl': line.split()[7],
'time_ms': line.split()[9]
}
return output_line
# ipv6 lines
elif ' bytes from ' in line:
line = line.replace(',', ' ').replace('=', ' ')
output_line = { output_line = {
'type': 'reply', 'type': 'reply',
'destination_ip': s.destination_ip or None, 'destination_ip': s.destination_ip or None,
'sent_bytes': s.sent_bytes or None, 'sent_bytes': s.sent_bytes or None,
'pattern': s.pattern or None, 'pattern': s.pattern or None,
'bytes': line.split()[0], 'response_bytes': line.split()[0],
'response_ip': line.split()[3], 'response_ip': line.split()[3],
'icmp_seq': line.split()[5], 'icmp_seq': line.split()[5],
'ttl': line.split()[7], 'ttl': line.split()[7],
@ -301,6 +282,24 @@ def _bsd_parse(line, s):
return output_line return output_line
# ipv6 lines
elif ' bytes from ' in line:
line = line.replace(',', ' ').replace('=', ' ')
output_line = {
'type': 'reply',
'destination_ip': s.destination_ip or None,
'sent_bytes': s.sent_bytes or None,
'pattern': s.pattern or None,
'bytes': line.split()[0],
'response_ip': line.split()[3],
'icmp_seq': line.split()[5],
'ttl': line.split()[7],
'time_ms': line.split()[9]
}
return output_line
def _linux_parse(line, s): def _linux_parse(line, s):
""" """
@ -318,7 +317,7 @@ def _linux_parse(line, s):
output_line = {} output_line = {}
if line.startswith('PING '): if line.startswith('PING '):
s.ipv4 = True if 'bytes of data' in line else False s.ipv4 = 'bytes of data' in line
if s.ipv4 and line[5] not in string.digits: if s.ipv4 and line[5] not in string.digits:
s.hostname = True s.hostname = True
@ -356,98 +355,94 @@ def _linux_parse(line, s):
s.packet_loss_percent = line.split()[7].rstrip('%') s.packet_loss_percent = line.split()[7].rstrip('%')
s.duplicates = line.split()[5].lstrip('+') s.duplicates = line.split()[5].lstrip('+')
s.time_ms = line.split()[11].replace('ms', '') s.time_ms = line.split()[11].replace('ms', '')
return None return None
else: s.packets_transmitted = line.split()[0]
s.packets_transmitted = line.split()[0] s.packets_received = line.split()[3]
s.packets_received = line.split()[3] s.packet_loss_percent = line.split()[5].rstrip('%')
s.packet_loss_percent = line.split()[5].rstrip('%') s.duplicates = '0'
s.duplicates = '0' s.time_ms = line.split()[9].replace('ms', '')
s.time_ms = line.split()[9].replace('ms', '') return None
return None split_line = line.split(' = ')[1]
split_line = split_line.split('/')
output_line = {
'type': 'summary',
'destination_ip': s.destination_ip or None,
'sent_bytes': s.sent_bytes or None,
'pattern': s.pattern or None,
'packets_transmitted': s.packets_transmitted or None,
'packets_received': s.packets_received or None,
'packet_loss_percent': s.packet_loss_percent or None,
'duplicates': s.duplicates or None,
'time_ms': s.time_ms or None,
'round_trip_ms_min': split_line[0],
'round_trip_ms_avg': split_line[1],
'round_trip_ms_max': split_line[2],
'round_trip_ms_stddev': split_line[3].split()[0]
}
else: return output_line
split_line = line.split(' = ')[1]
split_line = split_line.split('/')
output_line = {
'type': 'summary',
'destination_ip': s.destination_ip or None,
'sent_bytes': s.sent_bytes or None,
'pattern': s.pattern or None,
'packets_transmitted': s.packets_transmitted or None,
'packets_received': s.packets_received or None,
'packet_loss_percent': s.packet_loss_percent or None,
'duplicates': s.duplicates or None,
'time_ms': s.time_ms or None,
'round_trip_ms_min': split_line[0],
'round_trip_ms_avg': split_line[1],
'round_trip_ms_max': split_line[2],
'round_trip_ms_stddev': split_line[3].split()[0]
}
return output_line
# ping response lines # ping response lines
else:
# 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 # request timeout
if line[0] == '[': if 'no answer yet for icmp_seq=' in line:
timestamp = True timestamp = False
isequence = 6 isequence = 5
output_line = { # if timestamp option is specified, then shift icmp sequence field right by one
'type': 'timeout', if line[0] == '[':
'destination_ip': s.destination_ip or None, timestamp = True
'sent_bytes': s.sent_bytes or None, isequence = 6
'pattern': s.pattern or None,
'timestamp': line.split()[0].lstrip('[').rstrip(']') if timestamp else None,
'icmp_seq': line.replace('=', ' ').split()[isequence]
}
return output_line 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]
}
# normal responses return output_line
elif ' bytes from ' in line:
line = line.replace('(', ' ').replace(')', ' ').replace('=', ' ') # normal responses
if ' bytes from ' in line:
# positions of items depend on whether ipv4/ipv6 and/or ip/hostname is used line = line.replace('(', ' ').replace(')', ' ').replace('=', ' ')
if s.ipv4 and not s.hostname:
bts, rip, iseq, t2l, tms = (0, 3, 5, 7, 9)
elif s.ipv4 and s.hostname:
bts, rip, iseq, t2l, tms = (0, 4, 7, 9, 11)
elif not s.ipv4 and not s.hostname:
bts, rip, iseq, t2l, tms = (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 # positions of items depend on whether ipv4/ipv6 and/or ip/hostname is used
timestamp = False if s.ipv4 and not s.hostname:
if line[0] == '[': bts, rip, iseq, t2l, tms = (0, 3, 5, 7, 9)
timestamp = True elif s.ipv4 and s.hostname:
bts, rip, iseq, t2l, tms = (bts + 1, rip + 1, iseq + 1, t2l + 1, tms + 1) bts, rip, iseq, t2l, tms = (0, 4, 7, 9, 11)
elif not s.ipv4 and not s.hostname:
bts, rip, iseq, t2l, tms = (0, 3, 5, 7, 9)
elif not s.ipv4 and s.hostname:
bts, rip, iseq, t2l, tms = (0, 4, 7, 9, 11)
output_line = { # if timestamp option is specified, then shift everything right by one
'type': 'reply', timestamp = False
'destination_ip': s.destination_ip or None, if line[0] == '[':
'sent_bytes': s.sent_bytes or None, timestamp = True
'pattern': s.pattern or None, bts, rip, iseq, t2l, tms = (bts + 1, rip + 1, iseq + 1, t2l + 1, tms + 1)
'timestamp': line.split()[0].lstrip('[').rstrip(']') if timestamp else None,
'response_bytes': line.split()[bts],
'response_ip': line.split()[rip].rstrip(':'),
'icmp_seq': line.split()[iseq],
'ttl': line.split()[t2l],
'time_ms': line.split()[tms],
'duplicate': True if 'DUP!' in line else False
}
return output_line output_line = {
'type': 'reply',
'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,
'response_bytes': line.split()[bts],
'response_ip': line.split()[rip].rstrip(':'),
'icmp_seq': line.split()[iseq],
'ttl': line.split()[t2l],
'time_ms': line.split()[tms],
'duplicate': 'DUP!' in line
}
return output_line
def parse(data, raw=False, quiet=False, ignore_exceptions=False): def parse(data, raw=False, quiet=False, ignore_exceptions=False):