Skip to content

Instantly share code, notes, and snippets.

@realsama
Forked from sduthil/add-webrtc-line.py
Created November 13, 2021 03:04
Show Gist options
  • Save realsama/7a6184769b31a4aeda5e8de880ea0540 to your computer and use it in GitHub Desktop.
Save realsama/7a6184769b31a4aeda5e8de880ea0540 to your computer and use it in GitHub Desktop.
Add Wazo webrtc lines to users that don't have any
#!/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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment