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
|
#!/usr/bin/env python3
import datetime
import logging
import os
from typing import Any, Optional
import argparse_utils
from collect.bidict import BiDict
import config
import exec_utils
import persistent
import string_utils
logger = logging.getLogger(__name__)
cfg = config.add_commandline_args(
f'MAC <--> IP Address mapping table cache ({__file__})',
'Commandline args related to MAC <--> IP Address mapping',
)
cfg.add_argument(
'--arper_cache_location',
default=f'{os.environ["HOME"]}/cache/.arp_table_cache',
metavar='FILENAME',
help='Where to cache the kernel ARP table',
)
cfg.add_argument(
'--arp_cache_max_staleness',
type=argparse_utils.valid_duration,
default=datetime.timedelta(seconds=60 * 5),
metavar='DURATION',
help='Max acceptable age of the kernel arp table cache'
)
@persistent.persistent_autoloaded_singleton()
class Arper(persistent.Persistent):
def __init__(self, cached_state: Optional[BiDict[str, str]] = None) -> None:
self.state = BiDict()
if cached_state is not None:
logger.debug('Loading Arper map from cached state.')
self.state = cached_state
else:
logger.debug('No usable cached state; calling /usr/sbin/arp')
output = exec_utils.cmd(
'/usr/sbin/arp -a',
timeout_seconds=5.0
)
for line in output.split('\n'):
line = str(line, 'ascii')
ip = string_utils.extract_ip_v4(line)
mac = string_utils.extract_mac_address(line)
mac = mac.lower()
logger.debug(f' {mac} => {ip}')
self.state[mac] = ip
def get_ip_by_mac(self, mac: str) -> Optional[str]:
mac = mac.lower()
return self.state.get(mac, None)
def get_mac_by_ip(self, ip: str) -> Optional[str]:
return self.state.inverse.get(ip, None)
def save(self) -> bool:
logger.debug(f'Persisting state to {config.config["arp_cache_location"]}')
with open(config.config['arp_cache_location'], 'w') as wf:
for (mac, ip) in self.state.items():
mac = mac.lower()
print(f'{mac}, {ip}', file=wf)
@classmethod
def load(cls) -> Any:
if persistent.was_file_written_within_n_seconds(
config.config['arp_cache_location'],
config.config['arp_cache_max_staleness'].total_seconds(),
):
logger.debug(f'Loading state from {config.config["arp_cache_location"]}')
cached_state = BiDict()
with open(config.config['arp_cache_location'], 'r') as rf:
for line in rd.readline():
(mac, ip) = line.split(',')
mac = mac.strip()
mac = mac.lower()
ip = ip.strip()
cached_state[mac] = ip
return Arper(cached_state)
logger.debug('No usable saved state found')
return None
|
