Add Wazo webrtc lines to users that don't have any

add-webrtc-line.py

#!/usr/bin/python3

# Copyright 2020 The Wazo Authors  (see the AUTHORS file)
# SPDX-License-Identifier: GPL-3.0-or-later

import random
import string
from wazo_auth_client import Client as Auth
from wazo_confd_client import Client as Confd

# Please add a web service user with acl confd.#
# in order to use ./add-webrtc-line

host = 'localhost'  # Fill with your Wazo hostname if running the script remotely
username = "api-user"  # Fill with your username
password = "secret"  # Fill with your password
extension_start = '1100'  # Start of the extension range for selecting users
extension_end = '1199'  # End of the extension range.
tenant_uuid = 'aaa-bbb-ccc'  # Fill with your tenant UUID
webrtc_video = True  # Set to False if you don't want to enable WebRTC video

# ############### Do no modify ##############


def find_internal_context(tenant_uuid):
    contexts = confd.contexts.list(tenant_uuid=tenant_uuid)['items']
    internal_contexts = [context['name'] for context in contexts if context['type'] == 'internal']
    if len(internal_contexts) != 1:
        raise Exception(f'Tenant {tenant_uuid} has {len(internal_contexts)} internal contexts')
    return internal_contexts.pop()


def extension_in_range(extension, start, end):
    return start <= extension <= end


def find_user_extension(user, context):
    user_extensions = set((extension['id'], extension['exten']) for line in user['lines'] for extension in line['extensions'] if extension['context'] == context)
    if len(user_extensions) != 1:
        return None, None
    return user_extensions.pop()


def find_users_in_range(context, extension_start, extension_end):
    all_users = confd.users.list()['items']
    for user in all_users:
        extension_id, extension = find_user_extension(user, context)
        if not extension:
            print(f'Skipping user {user["firstname"]} {user["lastname"]}: user does not have only one extension')
            continue
        if extension_in_range(extension, extension_start, extension_end):
            yield extension_id, user


def endpoint_is_webrtc(endpoint):
    labels = set(template['label'] for template in endpoint['templates'])
    return 'webrtc' in labels


def find_webrtc_endpoint_uuids():
    all_endpoints = confd.endpoints_sip.list()['items']
    return (endpoint['uuid'] for endpoint in all_endpoints if endpoint_is_webrtc(endpoint))


def find_sip_templates():
    all_templates = confd.endpoints_sip_templates.list()['items']
    return {template['label']: template for template in all_templates}


def user_has_webrtc_line(user, webrtc_endpoint_uuids):
    endpoint_sip_uuids = set(line['endpoint_sip']['uuid'] for line in user['lines'] if line['endpoint_sip'])
    return not set(endpoint_sip_uuids).isdisjoint(set(webrtc_endpoint_uuids))


def generate_string():
    return ''.join(random.choice(string.ascii_lowercase) for _ in range(8))


def add_webrtc_line_to_user(user, context, extension_id, global_template, webrtc_template, webrtc_video_template):
    username = generate_string()
    password = generate_string()
    endpoint_sip_body = {
        'templates': [
            global_template,
            webrtc_template,
        ],
        'auth_section_options': [
            ('username', username),
            ('password', password),
        ],
        'name': username,
    }
    if webrtc_video:
        endpoint_sip_body['templates'].append(webrtc_video_template)
    line_body = {
        'name': username,
        'context': context
    }
    endpoint_sip = None
    line = None
    try:
        endpoint_sip = confd.endpoints_sip.create(endpoint_sip_body)
        line = confd.lines.create(line_body)

        print('Associating line', line['id'], 'with endpoint', endpoint_sip['uuid'], 'and user', user['uuid'])
        confd.endpoints_sip.relations(endpoint_sip['uuid']).associate_line(line['id'])
        confd.users.relations(user['uuid']).add_line(line['id'])
        confd.extensions.relations(extension_id).add_line(line['id'])
    except Exception as e:
        print('Error while creating WebRTC line for user', user['firstname'], user['lastname'])
        print(e)
        try:
            if endpoint_sip:
                confd.endpoints_sip.delete(endpoint_sip['uuid'])
            if line:
                confd.line.delete(line['id'])
        except Exception:
            print('Error while removing endpoint_sip', endpoint_sip['uuid'], 'or line', line['id'])


print('Creating token...')
auth = Auth(host, username=username, password=password, verify_certificate=False)
token = auth.token.new('wazo_user', expiration=60)['token']
confd = Confd(host, verify_certificate=False, token=token, tenant=tenant_uuid)
print('Done.')

print('Finding context')
context = find_internal_context(tenant_uuid)
print('Found context', context)
print('Finding WebRTC endpoints...')
webrtc_endpoints = set(find_webrtc_endpoint_uuids())
print('Found', len(webrtc_endpoints), 'WebRTC endpoints.')
print('Finding SIP templates...')
templates = find_sip_templates()
global_template = templates['global']
webrtc_template = templates['webrtc']
webrtc_video_template = templates['webrtc_video']
print('Found', len(templates), 'SIP templates.')

for extension_id, user in find_users_in_range(context, extension_start, extension_end):
    if not user_has_webrtc_line(user, webrtc_endpoints):
        print('Processing user', user['firstname'], user['lastname'])
        add_webrtc_line_to_user(user, context, extension_id, global_template, webrtc_template, webrtc_video_template)
    else:
        print('Skipping user', user['firstname'], user['lastname'], ': user already has webrtc line')


auth.token.revoke(token)