forked from calebmadrigal/trackerjacker
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathconfig_management.py
More file actions
executable file
·290 lines (244 loc) · 13.8 KB
/
config_management.py
File metadata and controls
executable file
·290 lines (244 loc) · 13.8 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
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
#!/usr/bin/env python3
# pylint: disable=C0111, C0103, W0703, R0902, R0903, R0912, R0913, R0914, R0915, C0413
import os
import copy
import json
import argparse
from .common import TJException
from . import plugin_parser
# Default config
DEFAULT_CONFIG = {'log_path': None,
'log_level': 'INFO',
'iface': None,
'devices_to_watch': [],
'aps_to_watch': [],
'threshold': None,
'power': None,
'threshold_window': 10,
'do_map': True,
'do_track': False,
'map_file': 'wifi_map.yaml',
'map_save_interval': 10,
'trigger_plugin': None,
'plugin_config': None,
'trigger_command': None,
'trigger_cooldown': 30,
'channels_to_monitor': None,
'channel_switch_scheme': 'default',
'time_per_channel': 0.5,
'display_matching_packets': False,
'display_all_packets': False,
'beep_on_trigger': False}
def get_arg_parser():
"""Returns the configured argparse object."""
parser = argparse.ArgumentParser()
# Modes
parser.add_argument('--map', action='store_true', dest='do_map',
help='Map mode - output map to wifi_map.yaml')
parser.add_argument('--track', action='store_true', dest='do_track',
help='Track mode')
parser.add_argument('--monitor-mode-on', action='store_true', dest='do_enable_monitor_mode',
help='Enables monitor mode on the specified interface and exit')
parser.add_argument('--monitor-mode-off', action='store_true', dest='do_disable_monitor_mode',
help='Disables monitor mode on the specified interface and exit')
parser.add_argument('--set-channel', metavar='CHANNEL', dest='set_channel', nargs=1,
help='Set the specified wireless interface to the specified channel and exit')
parser.add_argument('--mac-lookup', type=str, dest='mac_lookup',
help='Lookup the vendor of the specified MAC address and exit')
parser.add_argument('--print-default-config', action='store_true', dest='print_default_config',
help='Print boilerplate config file and exit')
# Normal switches
parser.add_argument('-v', '--version', action='store_true', dest='version',
help='Display trackerjacker version')
parser.add_argument('-i', '--interface', type=str, dest='iface',
help='Network interface to use; if empty, try to find monitor inferface')
parser.add_argument('-m', '--macs', type=str, dest='devices_to_watch',
help='MAC(s) to track; comma separated for multiple')
parser.add_argument('-a', '--access-points', type=str, dest='aps_to_watch',
help='Access point(s) to track - specified by BSSID; comma separated for multiple')
parser.add_argument('--channels-to-monitor', type=str, dest='channels_to_monitor',
help='Channels to monitor; comma separated for multiple')
parser.add_argument('--channel-switch-scheme', type=str, dest='channel_switch_scheme',
help='Options: "round_robin" or "traffic_based"', default='default')
parser.add_argument('--time-per-channel', type=float, dest='time_per_channel',
help='Seconds spent on each channel before hopping')
parser.add_argument('-w', '--time-window', type=int, dest='threshold_window',
help='Time window (in seconds) which alert threshold is applied to')
parser.add_argument('--map-save-interval', type=float, dest='map_save_interval',
help='Number of seconds between saving the wifi map to disk')
parser.add_argument('--threshold', type=int, dest='threshold',
help='Default data threshold (unless overridden on a per-dev basis) for triggering')
parser.add_argument('--power', type=int, dest='power',
help='Default power threshold (unless overridden on a per-dev basis) for triggering')
parser.add_argument('--plugin', type=str, dest='trigger_plugin',
help='Python trigger plugin file path; for more information')
parser.add_argument('--trigger-plugin', type=str, dest='trigger_plugin',
help='Python trigger plugin file path; for more information')
parser.add_argument('--plugin-config', type=str, dest='plugin_config',
help='Config to pass to python trigger plugin. Must be a python dict or json obj.')
parser.add_argument('--trigger-command', type=str, dest='trigger_command',
help='Command to execute upon alert')
parser.add_argument('--trigger-cooldown', type=str, dest='trigger_cooldown',
help='Time in seconds between trigger executions for a particular device')
parser.add_argument('--display-all-packets', action='store_true', dest='display_all_packets',
help='If true, displays all packets matching filters')
parser.add_argument('--beep-on-trigger', action='store_true', dest='beep_on_trigger',
help='If enabled, beep each time a trigger hits (off by default)')
parser.add_argument('--map-file', type=str, dest='map_file', default='wifi_map.yaml',
help='File path to which to output wifi map; default: wifi_map.yaml')
parser.add_argument('--log-path', type=str, dest='log_path', default=None,
help='Log path; default is stdout')
parser.add_argument('--log-level', type=str, dest='log_level', default='INFO',
help='Log level; Options: DEBUG, INFO, WARNING, ERROR, CRITICAL')
parser.add_argument('-c', '--config', type=str, dest='config',
help='Path to config json file; For example config file, use --print-default-config')
return parser
def parse_command_line_watch_list(watch_str):
"""Parse string that represents devices to watch.
Valid examples:
* aa:bb:cc:dd:ee:ff
- Threshold of 1 for the given MAC address
* aa:bb:cc:dd:ee:ff,11:22:33:44:55:66
- This means look for any traffic from either address
* aa:bb:cc:dd:ee:ff=1337, 11:22:33:44:55:66=1000
- This means look for 1337 bytes for the first address, and 1000 for the second
* my_ssid, 11:22:33:44:55:66=1000
- This means look for 1 byte from my_ssid or 1000 for the second
* 11:22:33:44:55:66=-30
- This means trigger if 11:22:33:44:55:66 is seen at a power level >= -30dBm (negative value implies power)
Returns dict in this format:
{'aa:bb:cc:dd:ee:ff': {'threshold': 100, 'power': None},
'11:22:33:44:55:66': {'threshold': None, 'power': -30}}
"""
watch_list = [i.strip() for i in watch_str.split(',')]
watch_dict = {}
for watch_part in watch_list:
power = None
threshold = None
if '=' in watch_part:
# dev_id is a MAC, BSSID, or SSID
dev_id, val = [i.strip() for i in watch_part.split('=')]
try:
val = int(val)
except ValueError:
# Can't parse with "dev_id=threshold" formula, so assume '=' sign was part of ssid
dev_id = watch_part
if val > 0:
threshold = val
else:
power = val
else:
dev_id = watch_part
watch_dict[dev_id] = {'threshold': threshold, 'power': power}
return watch_dict
def determine_watch_list(to_watch_from_args,
to_watch_from_config,
generic_threshold,
generic_power):
"""Builds the list of devices to watch.
Coalesces the to_watch list from the command-line arguments, the config file,
and the the general threshold, power, and trigger command. The main idea here is to look
for the config values set on a per-device basis, and prioritize those, but if they are not there,
fall back to the "generic_*" version. And if those have not been specified, fall back to defaults.
Example input:
to_watch_from_args = 'aa:bb:cc:dd:ee:ff=1337, 11:22:33:44:55:66=100, ff:ee:dd:cc:bb:aa',
to_watch_from_config = {}
generic_threshold = 1337
generic_power = None
Returns a dict in this format:
{
'aa:bb:cc:dd:ee:ff': {'threshold': None, 'power': -40},
'11:22:33:44:55:66': {'threshold': 100, 'power': None},
'ff:ee:dd:cc:bb:aa': {'threshold': 1337, 'power': None}
}
"""
# Converts from cli param format like: "aa:bb:cc:dd:ee:ff=-40,11:22:33:44:55:66=100' to a map like:
# {'aa:bb:cc:dd:ee:ff': {'threshold': None, 'power': -40},
# '11:22:33:44:55:66': {'threshold': 100, 'power': None}
if to_watch_from_args:
to_watch_from_args = parse_command_line_watch_list(to_watch_from_args)
if not to_watch_from_args:
to_watch_from_args = {}
if not to_watch_from_config:
to_watch_from_config = {}
watch_config_dict = {}
for dev_id in to_watch_from_args.keys() | to_watch_from_config.keys():
watch_entry = {'threshold': None,
'power': None}
watch_entry['threshold'] = (to_watch_from_args.get(dev_id, {}).get('threshold', None) or
to_watch_from_config.get(dev_id, {}).get('threshold', None))
watch_entry['power'] = (to_watch_from_args.get(dev_id, {}).get('power', None) or
to_watch_from_config.get(dev_id, {}).get('power', None))
if not watch_entry['threshold'] and not watch_entry['power']:
if generic_threshold and not generic_power:
watch_entry['threshold'] = generic_threshold
elif generic_power:
watch_entry['power'] = generic_power
else:
watch_entry['threshold'] = 1
watch_config_dict[dev_id] = watch_entry
return watch_config_dict
def build_config(args):
"""Builds the config from the command-line args, config file, and defaults."""
config = copy.deepcopy(DEFAULT_CONFIG)
devices_from_config = {}
aps_from_config = {}
if args.config:
try:
with open(args.config, 'r') as f:
config_from_file = json.loads(f.read())
# If there are any keys defined in the config file not allowed, error out
invalid_keys = set(config_from_file.keys()) - set(config.keys())
if invalid_keys:
raise TJException('Invalid keys found in config file: {}'.format(invalid_keys))
devices_from_config = config_from_file.pop('devices_to_watch', {})
aps_from_config = config_from_file.pop('aps_to_watch', {})
config.update(config_from_file)
print('Loaded configuration from {}'.format(args.config))
except (IOError, OSError, json.decoder.JSONDecodeError) as e:
raise TJException('Error loading config file ({}): {}'.format(args.config, e))
non_config_args = {'config', 'devices_to_watch', 'aps_to_watch', 'do_enable_monitor_mode', 'version',
'do_disable_monitor_mode', 'set_channel', 'print_default_config', 'mac_lookup'}
config_from_args = vars(args)
config_from_args = {k: v for k, v in config_from_args.items()
if v is not None and k not in non_config_args}
# Config from args trumps default or config file
config.update(config_from_args)
# Allow any plugins to override config
if config['trigger_plugin']:
trigger_plugin_path = get_real_plugin_path(config['trigger_plugin'])
parsed_trigger_plugin = plugin_parser.parse_trigger_plugin(trigger_plugin_path,
config['plugin_config'],
parse_only=True)
# Allow plugin to override any config parameters
if 'config' in parsed_trigger_plugin:
trigger_config = parsed_trigger_plugin['config']
config.update(**trigger_config)
try:
config['trigger_cooldown'] = float(config['trigger_cooldown'])
except ValueError:
raise TJException('trigger_cooldown must be a number')
# If we're in track mode and no other threshold info is set, default to a 1 byte data threshold
if config['do_track']:
if not config['threshold'] and not config['power']:
config['threshold'] = 1
config['devices_to_watch'] = determine_watch_list(args.devices_to_watch,
devices_from_config,
config['threshold'],
config['power'])
config['aps_to_watch'] = determine_watch_list(args.aps_to_watch,
aps_from_config,
config['threshold'],
config['power'])
if args.channels_to_monitor:
channels_to_monitor = args.channels_to_monitor.split(',')
config['channels_to_monitor'] = channels_to_monitor
return config
def get_real_plugin_path(trigger_plugin):
if not trigger_plugin.lower().endswith('.py') and '/' not in trigger_plugin:
possible_builtin_path = os.path.join(os.path.dirname(__file__),
'plugins',
'{}.py'.format(trigger_plugin))
if os.path.exists(possible_builtin_path):
trigger_plugin = possible_builtin_path
return trigger_plugin