-
Notifications
You must be signed in to change notification settings - Fork 185
Expand file tree
/
Copy pathcommon.py
More file actions
214 lines (176 loc) · 6.74 KB
/
Copy pathcommon.py
File metadata and controls
214 lines (176 loc) · 6.74 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
# Copyright (c) 2023, FoxIO, LLC.
# All rights reserved.
# Patent Pending
# JA4 is Open-Source, Licensed under BSD 3-Clause
# JA4+ (JA4S, JA4H, JA4L, JA4X, JA4SSH) are licenced under the FoxIO License 1.1. For full license text, see the repo root.
#
from hashlib import sha256
from datetime import datetime, timedelta
conn_cache = {}
quic_cache = {}
http_cache = {}
ssh_cache = {}
TLS_MAPPER = {'0x0002': "s2",
'0x0300': "s3",
'0x0301': "10",
'0x0302': "11",
'0x0303': "12",
'0x0304': "13"}
GREASE_TABLE = {'0x0a0a': True, '0x1a1a': True, '0x2a2a': True, '0x3a3a': True,
'0x4a4a': True, '0x5a5a': True, '0x6a6a': True, '0x7a7a': True,
'0x8a8a': True, '0x9a9a': True, '0xaaaa': True, '0xbaba': True,
'0xcaca': True, '0xdada': True, '0xeaea': True, '0xfafa': True}
def _parse_tls_int(value):
if isinstance(value, int):
return value
s = str(value).strip().lower()
if s.startswith('0x'):
return int(s, 16)
if any(ch in 'abcdef' for ch in s):
return int(s, 16)
return int(s)
def normalize_tls_list(values, width=4, prefix='0x'):
if values is None:
return []
if not isinstance(values, list):
values = [values]
return [ f"{prefix}{_parse_tls_int(v):0{width}x}" for v in values ]
def normalize_tls_value(value, width=4, prefix='0x'):
values = normalize_tls_list(value, width, prefix)
return values[0] if values else None
def normalize_tls_fields(packet, extensions_prefix='0x'):
if not packet:
return packet
if 'extensions' in packet:
packet['extensions'] = normalize_tls_list(packet['extensions'], prefix=extensions_prefix)
if 'ciphers' in packet:
packet['ciphers'] = normalize_tls_list(packet['ciphers'])
if 'supported_versions' in packet:
packet['supported_versions'] = normalize_tls_list(packet['supported_versions'])
if 'version' in packet:
packet['version'] = normalize_tls_value(packet['version'])
if 'signature_algorithms' in packet:
packet['signature_algorithms'] = normalize_tls_list(packet['signature_algorithms'])
return packet
def delete_keys(keys, x):
for key in keys:
if key in x:
del(x[key])
######## SIMPLE CACHE FUNCTIONS #############################
# The idea is to record quic packets into a quic_cache
# and record tcp tls packets into a conn_cache
# The cache is indexed by the stream number and hold all the
# required data including timestamps
# we print final results from the cache
def get_cache(x):
if x['hl'] in [ 'http', 'http2']:
return http_cache
elif x['hl'] == 'quic':
return quic_cache
else:
return conn_cache
def clean_cache(x):
cache = get_cache(x)
if x['stream'] in cache:
del(cache[x['stream']])
# Updates the cache and records timestamps
def cache_update(x, field, value, debug_stream=-1):
cache = get_cache(x)
stream = int(x['stream'])
update = False
if field == 'stream' and stream not in cache:
cache[stream] = { 'stream': stream}
return
# Do not update main tuple fields if they are already in
if field in [ 'stream', 'src', 'dst', 'srcport', 'dstport', 'A', 'B', 'JA4S', 'D', 'server_extensions', 'count', 'stats'] and field in cache[stream]:
return
# update protos only if we have extra information
if field == 'protos':
if field in cache[stream] and len(value) <= len(cache[stream][field]):
return
# special requirement for ja4c when the C timestamp needs to be the
# the last before D
if field == 'C' and 'D' in cache[stream]:
return
if stream in cache:
if stream == debug_stream:
print (f'updating ({"quic" if x["quic"] else "tcp"}) stream {stream} {field} {value}')
cache[stream][field] = value
update = True
return update
###### END OF CACHE FUNCTIONS
# Joins an array by commas in the order they are presented
# and returns the first 12 chars of the sha256 hash
def sha_encode(values):
if isinstance(values, list):
return sha256(','.join(values).encode('utf8')).hexdigest()[:12]
else:
return sha256(values.encode('utf8')).hexdigest()[:12]
# processes ciphers found in a packet
# tshark keeps the ciphers either as a list or as a single value
# based on whether it is ciphersuites or ciphersuite
def get_hex_sorted(entry, field, sort=True):
values = entry[field]
if not isinstance(values, list):
values = [ values ]
# remove GREASE and calculate length
c = [ x[2:] for x in values if x not in GREASE_TABLE ]
actual_length = min(len(c), 99)
# now remove SNI and ALPN values
if field == 'extensions' and sort:
c = [ x for x in c if x not in ['0000', '0010']]
c.sort() if sort else None
return ','.join(c), '{:02d}'.format(actual_length), sha_encode(c)
def get_supported_version(v):
if not isinstance(v, list):
v = [ v ]
versions = [ k for k in v if k not in GREASE_TABLE ]
versions.sort()
return versions[-1]
def parse_timestamp(timestamp):
"""
Parse a timestamp supporting multiple formats.
Parameters:
timestamp: Unix timestamp (float/int) or ISO 8601 string
Returns:
datetime object
"""
if isinstance(timestamp, str):
# ISO 8601 format, remove trailing 'Z'
return datetime.fromisoformat(timestamp.rstrip('Z'))
else:
# Unix timestamp
return datetime.fromtimestamp(float(timestamp))
## Time diff of epoch times / 2
## computes t2 - t1
## returns diff in seconds
def epoch_diff(t1, t2):
dt1 = parse_timestamp(t1)
dt2 = parse_timestamp(t2)
# timedelta.microseconds only holds the sub-second component, so any
# difference of a second or more was silently truncated before.
return (dt2 - dt1) // timedelta(microseconds=2)
# Scan for tls
def scan_tls(layer):
if not layer:
return None
if not isinstance(layer, list):
if 'tls_tls_handshake_type' in layer:
return layer
else:
for l in layer:
if 'tls_tls_handshake_type' in l:
return l
# Get the right signature algorithms
def get_signature_algorithms(packet):
if 'sig_alg_lengths' in packet and isinstance(packet['sig_alg_lengths'], list):
alg_lengths = [ int(int(x)/2) for x in packet['sig_alg_lengths'] ]
extensions = packet['extensions']
idx = 0
try:
if extensions.index('13') > extensions.index('35'):
idx = 1
except Exception as e:
pass
packet['signature_algorithms'] = packet['signature_algorithms'][alg_lengths[idx]:]
return [ x for x in packet.get('signature_algorithms', []) if x not in GREASE_TABLE ]