basestorage.py 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526
  1. #!/usr/bin/python
  2. # -*- coding: utf-8 -*-
  3. from __future__ import print_function, unicode_literals
  4. import logging
  5. import re
  6. import time
  7. import ffstatus
  8. from .exceptions import VpnKeyFormatError
  9. def sanitize_node(data, include_raw_data=False):
  10. """
  11. Filters potentially harmful entries from the node's data.
  12. """
  13. export = ffstatus.dict_merge({}, data)
  14. # remove fields from output: __RAW__
  15. if '__RAW__' in export and not include_raw_data:
  16. del export['__RAW__']
  17. return export
  18. class BaseStorage(object):
  19. """
  20. Provides operations on the storage data.
  21. This class gets subclassed to actually write the data
  22. to a file, database, whatever.
  23. """
  24. DATAKEY_VPN = '__VPN__'
  25. FIELDKEY_UPDATED = '__UPDATED__'
  26. metric_handler = None
  27. def open(self):
  28. """
  29. When overridden in a subclass,
  30. opens the persistent storage.
  31. """
  32. pass
  33. def save(self):
  34. """
  35. When overriden in a subclass,
  36. stores the data to a persistent storage.
  37. """
  38. pass
  39. def close(self):
  40. """
  41. When overridden in a subclass,
  42. closes the persistent storage.
  43. """
  44. pass
  45. @property
  46. def status(self):
  47. """Gets status information on the storage."""
  48. nodes = 0
  49. nodes_active = 0
  50. gateways = 0
  51. gateways_active = 0
  52. sum_clients = 0
  53. clients = set()
  54. for node in self.get_nodes():
  55. nodetype = node.get('type', 'node')
  56. if nodetype == 'gateway':
  57. gateways += 1
  58. if self.get_nodestatus(node=node) == 'active':
  59. gateways_active += 1
  60. continue
  61. nodes += 1
  62. nodemacs = [x for x in node.get('macs', [])]
  63. if 'mac' in node:
  64. nodemacs.append(node['mac'])
  65. if self.get_nodestatus(node=node) == 'active':
  66. nodes_active += 1
  67. sum_clients += node.get('clientcount', 0)
  68. for client in node.get('clients', []):
  69. if client in nodemacs:
  70. continue
  71. clients.add(client)
  72. return {
  73. 'clients_sum': sum_clients,
  74. 'clients_unique': len(clients),
  75. 'gateways': gateways,
  76. 'gateways_active': gateways_active,
  77. 'nodes': nodes,
  78. 'nodes_active': nodes_active,
  79. 'now': int(time.time()),
  80. }
  81. def __merge_alias_node(self, item, alias):
  82. # start by using standard dict_merge()
  83. update = ffstatus.dict_merge(item, alias, overwrite_lists=False)
  84. # extract some fields for further inspection
  85. update_macs = update.get('macs', []) or []
  86. # field 'node_id': keep original value
  87. update['node_id'] = item['node_id']
  88. # field 'mac': keep original value
  89. if 'mac' in item:
  90. update['mac'] = item['mac']
  91. if 'mac' in alias:
  92. update_macs.append(alias['mac'])
  93. # field 'macs': get rid of duplicates and primary mac
  94. primary_mac = update.get('mac')
  95. macs = []
  96. for x in update_macs:
  97. if x != primary_mac and x not in macs:
  98. macs.append(x)
  99. update['macs'] = update_macs = macs
  100. # field 'type': keep special node type
  101. item_type = item.get('type', 'node')
  102. if item_type != 'node' and update.get('type', 'node') == 'node':
  103. update['type'] = item_type
  104. return update
  105. def __send_metric(self, key, value, ts=None):
  106. if ts is None:
  107. ts = time.time()
  108. func = self.metric_handler
  109. if func is None or not callable(func):
  110. # no handler -> do nothing
  111. return
  112. func(self, key, value, ts)
  113. def merge_new_data(self, newdata):
  114. """Updates data in the storage by merging the new data."""
  115. ts = time.time()
  116. if newdata is None or not isinstance(newdata, dict):
  117. raise ValueError("Expected a dict as new data.")
  118. # keep a list of aliased nodes so they can be removed from the result
  119. aliased_nodes = {}
  120. # start merge on a copy of the current data
  121. current = {}
  122. for node in self.get_nodes():
  123. item_id = node['node_id']
  124. current[item_id] = ffstatus.dict_merge(node, {})
  125. current[item_id]['aliases'] = []
  126. current[item_id]['clients.bak'] = current[item_id].get('clients', [])
  127. current[item_id]['clients'] = []
  128. current[item_id]['neighbours'] = {}
  129. current[item_id]['type'] = 'node'
  130. if not item_id in newdata:
  131. continue
  132. if not '__RAW__' in current[item_id]:
  133. current[item_id]['__RAW__'] = {}
  134. if '__RAW__' in newdata[item_id]:
  135. for key in newdata[item_id]['__RAW__']:
  136. if key in current[item_id]['__RAW__']:
  137. del current[item_id]['__RAW__'][key]
  138. # merge the dictionaries
  139. updated = {}
  140. for itemid in newdata:
  141. if not itemid in current:
  142. # new element which did not exist in storage before, that's easy
  143. updated[itemid] = newdata[itemid]
  144. else:
  145. # merge the old and new element
  146. update = ffstatus.dict_merge(current[itemid], newdata[itemid])
  147. updated[itemid] = update
  148. for alias_id in updated[itemid].get('aliases', []):
  149. if alias_id in aliased_nodes:
  150. aliased_nodes[alias_id].append(itemid)
  151. else:
  152. aliased_nodes[alias_id] = [itemid]
  153. # merge aliased nodes
  154. for alias_id in aliased_nodes:
  155. if len(aliased_nodes[alias_id]) != 1:
  156. logging.warn("Node '%s' is aliased by multiple nodes: %s",
  157. alias_id, aliased_nodes[alias_id])
  158. continue
  159. # target's id is the single entry of the alias list
  160. item_id = aliased_nodes[alias_id][0]
  161. if alias_id == item_id:
  162. # the node has itself as alias -> remove the alias entry
  163. if alias_id in updated and 'aliases' in updated[alias_id]:
  164. updated[alias_id]['aliases'].remove(alias_id)
  165. logging.debug("Removed self-alias of '%s'.", alias_id)
  166. continue
  167. # look for alias node
  168. alias = updated.get(alias_id, current.get(alias_id))
  169. if alias is None:
  170. # no alias node present already, as we're trying to achieve here
  171. continue
  172. # look for target node
  173. item = updated.get(item_id, current.get(item_id))
  174. if item is None:
  175. logging.warn("Alias node '%s' is missing its target '%s.",
  176. alias_id, item_id)
  177. continue
  178. # ensure both target and alias node have 'node_id' field set
  179. if not 'node_id' in item:
  180. item['node_id'] = item_id
  181. alias['node_id'] = alias_id
  182. # merge data
  183. update = self.__merge_alias_node(item, alias)
  184. updated[item_id] = update
  185. logging.debug("Merged alias '%s' into '%s'.", alias_id, item_id)
  186. # mark alias node for deletion
  187. updated[alias_id] = None
  188. # sanitize each item's data
  189. for itemid in updated:
  190. if itemid.startswith('__'):
  191. continue
  192. item = updated[itemid]
  193. # delete node if requested
  194. if item is None:
  195. self.set_node_data(itemid, None)
  196. continue
  197. # ensure 'node_id' is set
  198. if not 'node_id' in item:
  199. item['node_id'] = itemid
  200. # remove node's MACs from clients list
  201. item_mac = item.get('mac') or ffstatus.guess_mac_from_nodeid(itemid)
  202. clients = [x for x in item.get('clients', [])]
  203. if item_mac is not None and item_mac in clients:
  204. clients.remove(item_mac)
  205. for mac in item.get('macs', []):
  206. if mac in clients:
  207. clients.remove(mac)
  208. # set clientcount
  209. item['clientcount'] = len(clients)
  210. # compute client delta
  211. prev_clients = item.get('clients.bak', [])
  212. new_clients = item.get('clients', [])
  213. if 'clients.bak' in item:
  214. del(item['clients.bak'])
  215. diff_added = [x for x in new_clients if x not in prev_clients]
  216. diff_removed = [x for x in prev_clients if x not in new_clients]
  217. self.__send_metric("nodes." + item_mac + ".clients_added",
  218. len(diff_added), ts)
  219. self.__send_metric("nodes." + item_mac + ".clients_removed",
  220. len(diff_removed), ts)
  221. # finally, set each new data
  222. self.set_node_data(itemid, item)
  223. def get_nodes(self, sortby=None, include_raw_data=False):
  224. """Gets a list of all known nodes."""
  225. nodes = self.get_all_nodes_raw()
  226. sorted_ids = [x for x in nodes]
  227. if sortby is not None:
  228. if sortby == 'name':
  229. sortkey = lambda x: nodes[x]['hostname'].lower()
  230. sorted_ids = sorted(sorted_ids, key=sortkey)
  231. elif sortby == 'id':
  232. sorted_ids = sorted(sorted_ids)
  233. result = []
  234. for nodeid in sorted_ids:
  235. if nodeid.startswith('__'):
  236. continue
  237. node = sanitize_node(nodes[nodeid], include_raw_data)
  238. result.append(node)
  239. return result
  240. def find_node(self, rawid, include_raw_data=False, search_aliases=True):
  241. """
  242. Fetch node data by given id.
  243. If necessary, look through node aliases.
  244. """
  245. # try direct match, first
  246. node = self.get_node(rawid)
  247. if node is not None:
  248. return sanitize_node(node, include_raw_data=include_raw_data)
  249. # look through all nodes
  250. found = None
  251. nodes = self.get_all_nodes_raw()
  252. for nodeid in nodes:
  253. node = nodes[nodeid]
  254. # if we have a direct hit, return it immediately
  255. if nodeid == rawid:
  256. return sanitize_node(node, include_raw_data=include_raw_data)
  257. # search through aliases
  258. if search_aliases and rawid in node.get('aliases', []):
  259. found = node
  260. # return found node
  261. if not found is None:
  262. return sanitize_node(found, include_raw_data=include_raw_data)
  263. else:
  264. return None
  265. def find_node_by_mac(self, mac):
  266. """Fetch node data by given MAC address."""
  267. needle = mac.lower()
  268. # iterate over all nodes
  269. for node in self.get_nodes():
  270. # check node's primary MAC
  271. if 'mac' in node and needle == node['mac'].lower():
  272. return sanitize_node(node)
  273. # check alias MACs
  274. if 'macs' in node:
  275. haystack = [x.lower() for x in node['macs']]
  276. if mac in haystack:
  277. return sanitize_node(node)
  278. # MAC address not found
  279. return None
  280. def get_nodestatus(self, rawid=None, node=None):
  281. """Determine node's status."""
  282. # search node by the given id
  283. if node is None and not rawid is None:
  284. node = self.find_node(rawid)
  285. # handle unknown nodes
  286. if node is None:
  287. return None
  288. # check that the last batadv update is noted in the data
  289. updated = node.get(self.FIELDKEY_UPDATED, None)
  290. if updated is None:
  291. return 'unknown'
  292. u = updated.get('batadv', updated.get('batctl'))
  293. if u is None:
  294. return 'unknown'
  295. # make decision based on time of last batadv update
  296. diff = time.time() - u
  297. if diff < 150:
  298. return 'active'
  299. elif diff < 300:
  300. return 'stale'
  301. else:
  302. return 'offline'
  303. def set_node_data(self, key, data):
  304. """
  305. Overwrite data for the node with the given key.
  306. Specifying 'None' as data effectively means deleting the key.
  307. """
  308. raise NotImplementedError("set_node_data was not overridden")
  309. def check_vpn_key(self, key):
  310. if key is None or re.match(r'^[a-fA-F0-9]+$', key) is None:
  311. raise VpnKeyFormatError(key)
  312. def get_vpn_keys(self):
  313. """Gets a list of VPN keys."""
  314. raise NotImplementedError("get_vpn_keys was not overriden")
  315. def get_vpn_item(self, key, create=False):
  316. self.check_vpn_key(key)
  317. raise NotImplementedError("store_vpn_item was not overriden")
  318. def store_vpn_item(self, key, data):
  319. raise NotImplementedError("store_vpn_item was not overriden")
  320. def resolve_vpn_remotes(self):
  321. """Iterates all remotes and resolves IP blocks not yet resolved."""
  322. vpn = self.get_vpn_keys()
  323. init_vpn_cache = {}
  324. for key in vpn:
  325. entry = self.get_vpn_item(key)
  326. entry_modified = False
  327. for mode in entry:
  328. if not isinstance(entry[mode], dict):
  329. continue
  330. for gateway in entry[mode]:
  331. if not isinstance(entry[mode][gateway], dict):
  332. continue
  333. item = entry[mode][gateway]
  334. if 'remote' in item and not 'remote_raw' in item:
  335. item['remote_raw'] = item['remote']
  336. resolved = None
  337. if item['remote'] in init_vpn_cache:
  338. resolved = init_vpn_cache[item['remote']]
  339. else:
  340. resolved = ffstatus.resolve_ipblock(item['remote'])
  341. init_vpn_cache[item['remote']] = resolved
  342. if resolved is not None:
  343. logging.info(
  344. 'Resolved VPN entry \'%s\' to net \'%s\'.',
  345. item['remote'],
  346. resolved['name'],
  347. )
  348. if resolved is not None:
  349. item['remote'] = resolved
  350. entry_modified = True
  351. if entry_modified:
  352. self.store_vpn_item(key, entry)
  353. def get_vpn_gateways(self):
  354. gateways = set()
  355. vpn = self.get_vpn_keys()
  356. for key in vpn:
  357. entry = self.get_vpn_item(key)
  358. for conntype in entry:
  359. for gateway in entry[conntype]:
  360. gateways.add(gateway)
  361. return sorted(gateways)
  362. def get_vpn_connections(self):
  363. conntypes = ['active', 'last']
  364. result = []
  365. vpnkeys = self.get_vpn_keys()
  366. for key in vpnkeys:
  367. vpn_entry = self.get_vpn_item(key)
  368. if not isinstance(vpn_entry, dict):
  369. continue
  370. item = {
  371. 'key': key,
  372. 'count': {},
  373. 'remote': {},
  374. }
  375. names = set()
  376. for conntype in conntypes:
  377. item['count'][conntype] = 0
  378. item['remote'][conntype] = {}
  379. if conntype in vpn_entry:
  380. for gateway in vpn_entry[conntype]:
  381. if 'remote' in vpn_entry[conntype][gateway]:
  382. remote = vpn_entry[conntype][gateway]['remote']
  383. if remote is None or remote == '':
  384. continue
  385. item['count'][conntype] += 1
  386. item['remote'][conntype][gateway] = remote
  387. if 'peer' in vpn_entry[conntype][gateway]:
  388. names.add(vpn_entry[conntype][gateway]['peer'])
  389. item['names'] = sorted(names)
  390. item['online'] = item['count']['active'] > 0
  391. result.append(item)
  392. return result
  393. def log_vpn_connect(self, key, peername, remote, gateway, timestamp):
  394. item = self.get_vpn_item(key, create=True)
  395. # resolve remote addr to its netblock
  396. remote_raw = remote
  397. remote_resolved = None
  398. if remote is not None:
  399. remote_resolved = ffstatus.resolve_ipblock(remote)
  400. if remote_resolved is not None:
  401. logging.debug('Resolved IP \'{0}\' to block \'{1}\'.'.format(
  402. remote, remote_resolved['name'],
  403. ))
  404. remote = remote_resolved
  405. # store connection info
  406. item['active'][gateway] = {
  407. 'establish': timestamp,
  408. 'peer': peername,
  409. 'remote': remote,
  410. 'remote_raw': remote_raw,
  411. }
  412. self.store_vpn_item(key, item)
  413. def log_vpn_disconnect(self, key, gateway, timestamp):
  414. item = self.get_vpn_item(key, create=True)
  415. active = {}
  416. if gateway in item['active']:
  417. active = item['active'][gateway]
  418. del item['active'][gateway]
  419. active['disestablish'] = timestamp
  420. item['last'][gateway] = active
  421. self.store_vpn_item(key, item)