openmedialibrary/oml/localnodes.py
2016-03-26 17:26:19 +01:00

156 lines
5.4 KiB
Python

# -*- coding: utf-8 -*-
# vi:si:et:sw=4:sts=4:ts=4
import socket
import netifaces
from zeroconf import (
ServiceBrowser, ServiceInfo, ServiceStateChange, Zeroconf
)
from tornado.ioloop import PeriodicCallback
import settings
import state
from tor_request import get_opener
import logging
logger = logging.getLogger(__name__)
def can_connect(data):
try:
opener = get_opener(data['id'])
headers = {
'User-Agent': settings.USER_AGENT,
'X-Node-Protocol': settings.NODE_PROTOCOL,
'Accept-Encoding': 'gzip',
}
if ':' in data['host']:
url = 'https://[{host}]:{port}'.format(**data)
else:
url = 'https://{host}:{port}'.format(**data)
opener.addheaders = list(zip(headers.keys(), headers.values()))
opener.timeout = 1
r = opener.open(url)
version = r.headers.get('X-Node-Protocol', None)
if version != settings.NODE_PROTOCOL:
logger.debug('version does not match local: %s remote %s (%s)', settings.NODE_PROTOCOL, version, data['id'])
return False
c = r.read()
return True
except:
pass
#logger.debug('failed to connect to local node %s', data, exc_info=True)
return False
def get_broadcast_interfaces():
return list(set(
addr['addr']
for iface in netifaces.interfaces()
for addr in netifaces.ifaddresses(iface).get(socket.AF_INET, [])
if addr.get('netmask') != '255.255.255.255' and addr.get('broadcast')
))
class LocalNodes(dict):
service_type = '_oml._tcp.local.'
local_info = None
local_ips = None
def __init__(self):
if not settings.server.get('localnode_discovery'):
return
self.setup()
self._ip_changed = PeriodicCallback(self._update_if_ip_changed, 60000)
def setup(self):
self.local_ips = get_broadcast_interfaces()
self.zeroconf = {ip: Zeroconf(interfaces=[ip]) for ip in self.local_ips}
self.register_service()
self.browse()
def _update_if_ip_changed(self):
local_ips = get_broadcast_interfaces()
if local_ips != self.local_ips:
self.close()
self.setup()
def browse(self):
self.browser = {
ip: ServiceBrowser(self.zeroconf[ip], self.service_type, handlers=[self.on_service_state_change])
for ip in self.zeroconf
}
def register_service(self):
if self.local_info:
for local_ip, local_info in self.local_info:
self.zeroconf[local_ip].unregister_service(local_info)
self.local_info = None
local_name = socket.gethostname().partition('.')[0] + '.local.'
port = settings.server['node_port']
desc = {
'username': settings.preferences.get('username', 'anonymous')
}
self.local_info = []
for i, local_ip in enumerate(get_broadcast_interfaces()):
if i:
name = '%s-%s [%s].%s' % (desc['username'], i+1, settings.USER_ID, self.service_type)
else:
name = '%s [%s].%s' % (desc['username'], settings.USER_ID, self.service_type)
local_info = ServiceInfo(self.service_type, name,
socket.inet_aton(local_ip), port, 0, 0, desc, local_name)
self.zeroconf[local_ip].register_service(local_info)
self.local_info.append((local_ip, local_info))
def __del__(self):
self.close()
def close(self):
if self.local_info:
for local_ip, local_info in self.local_info:
try:
self.zeroconf[local_ip].unregister_service(local_info)
except:
logger.debug('exception closing zeroconf', exc_info=True)
self.local_info = None
if self.zeroconf:
for local_ip in self.zeroconf:
try:
self.zeroconf[local_ip].close()
except:
logger.debug('exception closing zeroconf', exc_info=True)
self.zeroconf = None
for id in list(self):
self.pop(id, None)
def on_service_state_change(self, zeroconf, service_type, name, state_change):
if not '[' in name:
id = name.split('.')[0]
else:
id = name.split('[')[1].split(']')[0]
if id == settings.USER_ID:
return
if state_change is ServiceStateChange.Added:
info = zeroconf.get_service_info(service_type, name)
if info:
self[id] = {
'id': id,
'host': socket.inet_ntoa(info.address),
'port': info.port
}
if info.properties:
for key, value in info.properties.items():
key = key.decode()
self[id][key] = value.decode()
logger.debug('add: %s', self[id])
if state.tasks:
state.tasks.queue('addlocalinfo', self[id])
elif state_change is ServiceStateChange.Removed:
logger.debug('remove: %s', id)
self.pop(id, None)
if state.tasks:
state.tasks.queue('removelocalinfo', id)
def get(self, user_id):
if user_id in self and can_connect(self[user_id]):
return self[user_id]