#!/usr/bin/env python3 import logging import mitel_ommclient2 import time import hashlib from .extensions import ExtensionConfig logger = logging.getLogger("fieldpoc.dect") class Dect: def __init__(self, fp): self.fp = fp self.c = None # OOMClient2 not initialized at the beginning def _init_client(self): self.c = mitel_ommclient2.OMMClient2( host=self.fp.config.dect.host, username=self.fp.config.dect.username, password=self.fp.config.dect.password, ommsync=True, ) @property def temp_num_prefix(self): return next(self.fp.extensions.extensions_by_type("temp")).num def load_temp_extensions(self): current_temp_extension = 0 used_temp_extensions = self.c.find_users(lambda u: u.num.startswith(self.temp_num_prefix)) for u in used_temp_extensions: temp_num = u.num self.fp.temp_extensions[temp_num] = ExtensionConfig( num=temp_num, name=f"Temp {temp_num[4:]}", type="dect", trunk=False, dialout_allowed=False, ) def get_temp_number(self): current_temp_extension = 0 used_temp_extensions = [num[len(self.temp_num_prefix):] for num, ext in self.fp.temp_extensions.items()] while "{:0>4}".format(current_temp_extension) in used_temp_extensions: current_temp_extension += 1 return "{}{:0>4}".format(self.temp_num_prefix, current_temp_extension) def get_sip_password_for_number(self, num): return hashlib.sha256(bytes.fromhex((self.fp.config.dect.sipsecret + str(num))[-16:])).hexdigest()[:16] def create_and_bind_user(self, d, num): u = self.c.create_user(num) self.c.attach_user_device(u.uid, d.ppn) self.c.set_user_relation_fixed(u.uid) self.c.set_user_sipauth(u.uid, num, self.get_sip_password_for_number(num)) return u def sync(self): logger.info("syncing") # load DECT extensions from configuration extensions = self.fp.extensions.extensions_by_type("dect") # accessible by extension number extensions_by_num = {e.num: e for e in extensions} # accessible by device IPEI, for devices with static IPEI in configuration extensions_by_ipei = {e.dect_ipei: e for _, e in extensions_by_num.items() if e.dect_ipei is not None} # signal if new extensions got created created_tmp_ext = False # collect users users_by_ext = {} users_by_uid = {} # check all users on OMM for user in self.c.get_users(): # check user on OMM against fieldpoc configuration e = extensions_by_num.get(user.num) # user in OMM, but no DECT extension in configuration if not e: # the user might have a temporary extension assigned, collect it if user.num.startswith(next(self.fp.extensions.extensions_by_type("temp")).num): users_by_ext[user.num] = user users_by_uid[user.uid] = user # we have no configuration for the user else: # TODO: delete in omm pass continue # update user in OMM if name of extension changed elif e.name != user.name: self.c.set_user_name(user.uid, e.name) if e.dect_ipei is not None and user.relType != mitel_ommclient2.types.PPRelTypeType("Unbound"): d = self.c.get_device(user.ppn) if d.ipei != e.dect_ipei: logger.debug(f"Detaching {user} {d}") self.c.detach_user_device(user.uid, user.ppn) self.c.set_user_sipauth(user.uid, e.num, self.get_sip_password_for_number(e.num)) users_by_ext[user.num] = user users_by_uid[user.uid] = user # check all devices on OMM for d in self.c.get_devices(): # find extension in fieldpoc configuration with static IPEI of the device on OMM e = extensions_by_ipei.get(d.ipei) # in case the IPEI is statically assigned to an extension if e: # find user on OMM configured for the selectied extension u = users_by_ext.get(e.num) # bind the user to the device if user exist and device is not bound to any user if u and d.relType == mitel_ommclient2.types.PPRelTypeType("Unbound"): logger.debug(f'Binding user for {d}') self.c.attach_user_device(u.uid, d.ppn) self.c.set_user_relation_fixed(u.uid) # fix user settings on OMM if device on OMM is already connected to a user elif d.relType != mitel_ommclient2.types.PPRelTypeType("Unbound"): ui = users_by_uid.get(d.uid) if ui.num != e.num: logger.debug(f'User for {d} has wrong number') if self.fp.temp_extensions.get(ui.num): self.fp.temp_extensions.pop(ui.num) self.c.set_user_num(d.uid, e.num) self.c.set_user_sipauth(d.uid, e.num, self.get_sip_password_for_number(e.num)) self.c.set_user_name(user.uid, e.name) # create a new user on OMM if none exist and bind it to not bound device on OMM else: logger.debug(f'Creating and binding user for {d}') user = self.create_and_bind_user(d, e.num) self.c.set_user_name(user.uid, e.name) # assign any unbound device without static assigned IPEI a temporary extension elif d.relType == mitel_ommclient2.types.PPRelTypeType("Unbound"): temp_num = self.get_temp_number() logger.debug(f'Creating and binding tmp-user for {d}: {temp_num}') user = self.create_and_bind_user(d, temp_num) self.c.set_user_name(user.uid, f"Temp {temp_num[4:]}") self.fp.temp_extensions[temp_num] = ExtensionConfig( num=temp_num, name=f"Temp {temp_num[4:]}", type="dect", trunk=False, dialout_allowed=False, ) created_tmp_ext = True # update rounting when new extensions got created if created_tmp_ext: logger.info("notify for routing update") self.fp.queues['routing'].put({"type": "sync"}) logger.info("finished sync") def claim_extension(self, current_extension, token): new_extension = None # find an extension that can be claimed with the provided token for ext in self.fp.extensions.extensions_by_type("dect"): if ext.dect_claim_token == token: new_extension = ext break if e: # find user on OMM related to the current extension user = next(self.c.find_users(lambda u: u.num == current_extension)) # if the current extension is a temporary extension, free it up again if self.fp.temp_extensions.get(user.num): self.fp.temp_extensions.pop(user.num) # assign new extension to the user on OMM self.c.set_user_num(user.uid, new_extension.num) self.c.set_user_sipauth(user.uid, e.num, self.get_sip_password_for_number(new_extension.num)) self.c.set_user_name(user.uid, new_extension.name) def run(self): logger.info("starting dect thread") while True: msg = self.fp.queues["dect"].get() self.fp.queues["dect"].task_done() if msg.get("type") == "stop": self.restart_loop = False logger.info("stopped") break else: try: # setup connection to OMM if self.c is None: logger.info("initialising connection to OMM") self._init_client() self.load_temp_extensions() # handle tasks if msg.get("type") == "sync": self.sync() elif msg.get("type") == "claim": self.claim_extenstion(msg.get("extension"), msg.get("token")[4:]) except: logger.exception("task failed") # reset connection to OMM if self.c is not None: try: self.c.connection.close() except: pass self.c = None time.sleep(2) logger.info("requeue failed task") self.fp.queues["dect"].put(msg) if self.c is not None: self.c.connection.close()