474 lines
18 KiB
Python
474 lines
18 KiB
Python
import typing
|
|
|
|
#This became a class so we can combine envvar initialisation with cred storage
|
|
class Py3CXEnvironmentVars:
|
|
"""Class to initialise module configuration from environment variables
|
|
|
|
Class to initialise module configuration from environment variables, but doubles as configuration storage for explicit credentials
|
|
"""
|
|
Prefix='TCX'
|
|
SiteComponent='SITE'
|
|
AuthComponent='API_AUTH'
|
|
EnvTemplate='{}_{}_{}'
|
|
URI=EnvTemplate.format(Prefix, SiteComponent, 'URI')
|
|
AUTH_USER=EnvTemplate.format(Prefix, AuthComponent, 'USERNAME')
|
|
AUTH_PWD=EnvTemplate.format(Prefix, AuthComponent, 'PASSWORD')
|
|
def __init__(self):
|
|
from os import getenv
|
|
self.uri=getenv(self.URI)
|
|
self.auth_user=getenv(self.AUTH_USER)
|
|
self.auth_pwd=getenv(self.AUTH_PWD)
|
|
|
|
#Base class for exceptions means one can 'except Py3CXException'
|
|
#TODO: should this instead be a subclass of Py3CX? (except Py3CX.Exception?)
|
|
class Py3CXException(Exception):
|
|
"""Base Py3CX Exception class"""
|
|
pass
|
|
class APIError(Py3CXException):
|
|
"""Exception class used in event of an error being returned by the API"""
|
|
pass
|
|
class ValidationError(Py3CXException):
|
|
"""Exception class used in event of a validation error internal to the class"""
|
|
pass
|
|
|
|
#Request is a basic wrapper around python-requests to remove the need to always specify the API base uri, and to handle basic result validation on behalf of the caller
|
|
class Request:
|
|
"""Internal-use wrapper around Python-Requests"""
|
|
from requests import Response, ConnectionError, exceptions
|
|
def __init__(self, uri : str, verify : bool):
|
|
"""Initialises a Request object
|
|
|
|
Initialises a Python-Requests Session object with decorative attributes and wrappers
|
|
|
|
Parameters
|
|
---
|
|
uri: Required, string representing the FQDN where the API lives
|
|
verify: Required, boolean representing whether to verify the TLS trust chain during connections
|
|
|
|
>>> rq=Request('http://myinstance.3cx.com',False)
|
|
>>> rq
|
|
<Request [http://myinstance.3cx.com,False]>
|
|
"""
|
|
from requests import Session
|
|
self.uri=uri
|
|
self.base="%s/api/{}" % uri
|
|
self.last=None
|
|
self.sess=Session()
|
|
self.sess.verify=verify
|
|
def __repr__(self):
|
|
return '<Request [%s,v%s]>' % (self.base,self.sess.verify)
|
|
def _action(self, api : str, method, params, expect=200) -> 'Response':
|
|
try:
|
|
if isinstance(method, str):
|
|
if method is 'GET':
|
|
self.last=resp=self.sess.get(url=self.base.format(api), params=params, json=params)
|
|
elif method is 'POST':
|
|
self.last=resp=self.sess.post(url=self.base.format(api), json=params, params=params)
|
|
else:
|
|
self.last=resp=method(url=self.base.format(api), json=params, params=params)
|
|
assert (resp.status_code==expect)
|
|
except AssertionError:
|
|
raise APIError("Assertion error when handling API response; response code was %s, but expected response was %s" % (resp.status_code, expect))
|
|
except self.exceptions.SSLError as e:
|
|
raise APIError("TLS error returned when communicating; use tls_verify=False or check leaf certs: %s" % str(e))
|
|
except self.exceptions.BaseHTTPError as e:
|
|
raise APIError("HTTP error raised when communicating: %s" % str(e))
|
|
except ConnectionError as e:
|
|
raise APIError("ConnectionError raised when communicating: %s" % str(e))
|
|
return resp
|
|
def get(self, api : str, params=None, expect : int =200) -> 'Response':
|
|
return self._action(api, self.sess.get, params=params, expect=expect)
|
|
def post(self, api : str, params, expect : int =200) -> 'Response':
|
|
return self._action(api, self.sess.post, params=params, expect=expect)
|
|
|
|
#BasicObject is literally just an object - hacky way of making a variable addressable via var.prop rather than var['prop']
|
|
class BasicObject(object):
|
|
pass
|
|
#APIObject is a basic object type that has some glue to reduce duplicate code
|
|
class APIObject(BasicObject):
|
|
def __init__(self, parent, api):
|
|
self.tcx=parent
|
|
self.api=api
|
|
def __repr__(self):
|
|
return '<APIObject [%s]>' % self.api
|
|
#ReadOnlyObject is a basic object type that expects to be able to GET the given API endpoint and retrieve a JSON response
|
|
class ReadOnlyObject(APIObject):
|
|
def __repr__(self):
|
|
return '<ReadOnlyObject [%s]>' % self.api
|
|
def refresh(self, params={}):
|
|
self._result=self.tcx.rq.get(self.api, params=params)
|
|
self.active=self._result.json()
|
|
#TransactionalObject is an object type that expects to retrieve details from an API via one endpoint, then use standard 3CX transactional endpoints for either cancelling the modify request, or submit modifications
|
|
class TransactionalObject(APIObject):
|
|
def __init__(self, parent, api, save='edit/save', discard='edit/cancel'):
|
|
super().__init__(parent, api)
|
|
self.save=save
|
|
self.discard=discard
|
|
def __repr__(self):
|
|
return '<TransactionalObject [%s]>' % self.api
|
|
def create(self, params):
|
|
self._result=self.tcx.rq.post(self.api, params=params)
|
|
self._session=self._result.json()['Id']
|
|
self.active=self._result.json()['ActiveObject']
|
|
def submit(self, params):
|
|
raise ValidationError("NotImplemented error")
|
|
def cancel(self):
|
|
self.tcx.rq.post(self.discard, params={
|
|
'Id':self._session})
|
|
|
|
#Main logic goes here
|
|
class Py3CX:
|
|
"""Base Py3CX class, containing all functionality for interacting with a 3CX phone system"""
|
|
class _Call(ReadOnlyObject):
|
|
def __init__(self, tcx : 'Py3CX', callid : int):
|
|
"""Returns a Py3CX.Call object
|
|
|
|
Returns a Py3CX.Call object representing a currently-ongoing or historical call on a 3CX system
|
|
|
|
Parameters
|
|
---
|
|
tcx: Required, Py3CX object representing an authenticated 3CX API session
|
|
callid: Required, integer representing the unique identifier of a given call
|
|
"""
|
|
super().__init__(tcx, 'activeCalls')
|
|
self.params=callid
|
|
def refresh(self):
|
|
"""Initialises or updates the properties of the current Py3CX.Call object
|
|
|
|
Loads in and stores the properties of the current Py3CX.Call object by performing a GET request to the 'activeCalls' API endpoint, filtering the resulting list by the Call ID.
|
|
Future: Calls not present in the activeCalls API call results should be queried against the call history database
|
|
|
|
>>> cl=tcx._Call(tcx, 1234)
|
|
>>> cl.state
|
|
Dialing
|
|
"""
|
|
self.refresh()
|
|
self.timestamp=self._result.headers.get('date')
|
|
res=list(filter(lambda cid: cid['Id'] == self.params, self._result.json()['list']))
|
|
assert (len(res)>0), "No call found in currently-active call list for ID %s" % self.params
|
|
assert(len(res)==1), "More than one active call found for ID %s" % self.params
|
|
res=res[0]
|
|
self.id=res['Id']
|
|
self.caller=res['Caller']
|
|
self.callee=res['Callee']
|
|
self.state=res['Status']
|
|
self.duration=res['Duration']
|
|
self.since=res['LastChangeStatus']
|
|
def __repr__(self):
|
|
return '<Py3CX.Call [%s %s]>' % (self.id, self.state)
|
|
def hangup(self):
|
|
"""Terminates the current call
|
|
|
|
If the call represented by this object is an ongoing call, drops the call by performing a POST request to 'activeCalls/drop' with the Call ID as the Id parameter in the JSON payload
|
|
|
|
>>> cl=tcx._Call(tcx, 1234)
|
|
>>> cl.state
|
|
Connected
|
|
>>> cl.hangup()
|
|
>>> cl.state
|
|
Terminated
|
|
"""
|
|
self.refresh()
|
|
if self.state is 'Terminated':
|
|
return
|
|
self.tcx.rq.post('activeCalls/drop', params={
|
|
'Id': self.params})
|
|
self.refresh()
|
|
class _User(TransactionalObject):
|
|
def __init__(self, parent : 'Py3CX', params : str):
|
|
"""Returns a Py3CX.User object
|
|
|
|
Returns a Py3CX.User object which represents a given user account (Extension) on a 3CX system, including more advanced attributes.
|
|
|
|
Parameters
|
|
---
|
|
parent: Required, Py3CX object representing an authenticated 3CX API session
|
|
params: Required, string representing the extension number of a given user
|
|
|
|
>>> user=tcx._User(tcx, '1234')
|
|
>>> user.enabled
|
|
True
|
|
>>> user.extension
|
|
<Py3CX.Extension [1234]>
|
|
"""
|
|
super().__init__(parent, 'ExtensionList/set')
|
|
self.params=params
|
|
self.load()
|
|
def __repr__(self):
|
|
return '<Py3CX.User [%s(%s)]>' % (self.params, self.sip_authid)
|
|
def load(self):
|
|
self.create(params={
|
|
'Id':self.params})
|
|
parms=self.active
|
|
self.id=parms['Id']
|
|
self.enabled=not parms['Disabled']['_value']
|
|
self.sip_id=parms['SIPId']['_value']
|
|
#RecordCallsOption - enum
|
|
self.callrecs=parms['RecordCalls']['selected']
|
|
self.sip_host=parms['MyPhoneLocalInterface']['selected']
|
|
self.sip_host_addrs=parms['MyPhoneLocalInterface']['possibleValues']
|
|
self.sip_authid=parms['AuthId']['_value']
|
|
self.sip_authpw=parms['AuthPassword']['_value']
|
|
self.provision_uri=parms['MyPhoneProvLink']['_value']
|
|
self.webpw=parms['AccessPassword']['_value']
|
|
self.extension=Py3CX._Extension(self.tcx, self.params)
|
|
self.cancel()
|
|
class _Extension(ReadOnlyObject):
|
|
def __init__(self, parent : 'Py3CX', params : str, populate : bool =True):
|
|
"""Returns a Py3CX.Extension object
|
|
|
|
Returns a Py3CX.Extension object representing information about the specified extension number.
|
|
|
|
Parameters
|
|
---
|
|
parent: Required, Py3CX object representing the authenticated 3CX session
|
|
params: Required, string representing the extension number to select
|
|
populate: Optional, boolean indicator as to whether the object should be populated during instantiation
|
|
|
|
>>> extension=tcx.system.list_extensions[0]
|
|
>>> extension
|
|
<Py3CX.Extension [1234]>
|
|
>>> extension=tcx._Extension(tcx, '1234')
|
|
>>> extension.number
|
|
'1234'
|
|
"""
|
|
super().__init__(parent, api='ExtensionList')
|
|
self.params=params
|
|
if populate:
|
|
self.load()
|
|
def __repr__(self):
|
|
return '<Py3CX.Extension [%s]>' % self.number
|
|
def load(self):
|
|
"""Initialises or updates the Py3CX.Extension object with the latest properties from the server
|
|
|
|
Performs a GET request to 'ExtensionList', filtering the response by matches against the specified extension number, populating the object's properties with the filtered results
|
|
|
|
>>> extension=tcx._Extension(tcx, '1234', populate=False)
|
|
>>> extension.load()
|
|
>>> extension.number
|
|
'1234'
|
|
"""
|
|
self.refresh(params=self.params)
|
|
res=self._result
|
|
self.timestamp=res.headers.get('date')
|
|
res=list(filter(lambda ext: ext['Number'] == self.params, res.json()['list']))
|
|
assert (len(res)>0), "No extension found for: %s" % self.params
|
|
assert (len(res)==1), "More than one extension found for %s" % self.params
|
|
res=res[0]
|
|
self.number=res['Number']
|
|
self.firstname=res['FirstName']
|
|
self.surname=res['LastName']
|
|
self.name="%s %s" % (self.firstname, self.surname)
|
|
self.dnd=res['DND']
|
|
self.status=res['CurrentProfile']
|
|
self.mail=res['Email']
|
|
self.cli=res['OutboundCallerId']
|
|
self.mobile=res['MobileNumber']
|
|
self.online=res['IsRegistered']
|
|
class _PhoneSystem:
|
|
class System(object):
|
|
class License(object):
|
|
pass
|
|
class Status(object):
|
|
pass
|
|
def __init__(self, tcx : 'Py3CX'):
|
|
"""Returns a Py3CX.System object
|
|
|
|
Returns a Py3CX.System object, which contains functionality for obtaining or acting on system-wide components of a 3CX system
|
|
|
|
Parameters
|
|
---
|
|
tcx: Required, Py3CX instance considered the parent object
|
|
|
|
>>> tcx._PhoneSystem(tcx)
|
|
<Py3CX.System [http://myinstance.3cx.com]>
|
|
"""
|
|
self.tcx=tcx
|
|
self.calllist=None
|
|
self.extlist=None
|
|
def __repr__(self):
|
|
return '<Py3CX.System [%s]>' % self.tcx.cnf.uri
|
|
def refresh_sysstat(self):
|
|
sysstat=self.tcx.rq.get('SystemStatus').json()
|
|
self.Status.exts_online=sysstat['ExtensionsRegistered']
|
|
self.Status.trunks_online=sysstat['TrunksRegistered']
|
|
self.Status.calls_active=sysstat['CallsActive']
|
|
self.Status.diskbytes_free=sysstat['FreeDiskSpace']
|
|
self.Status.membytes_free=sysstat['FreePhysicalMemory']
|
|
self.Status.banned_ips=sysstat['BlacklistedIpCount']
|
|
self.Status.last_backup=sysstat['LastBackupDateTime']
|
|
self.Status.sysexts_online=not sysstat['HasUnregisteredSystemExtensions']
|
|
self.Status.services_online=not sysstat['HasNotRunningServices']
|
|
self.System.routes_out=sysstat['OutboundRules']
|
|
self.System.fqdn=sysstat['FQDN']
|
|
self.System.webconf=sysstat['WebMeetingFQDN']
|
|
self.System.version=sysstat['Version']
|
|
self.System.diskbytes_total=sysstat['TotalDiskSpace']
|
|
self.System.membytes_total=sysstat['TotalPhysicalMemory']
|
|
self.System.exts_total=sysstat['ExtensionsTotal']
|
|
self.System.trunks_total=sysstat['TrunksTotal']
|
|
self.System.firebase_confed=sysstat['OwnPush']
|
|
self.System.backups_enabled=sysstat['BackupScheduled']
|
|
self.System.License.activated=sysstat['Activated']
|
|
self.System.License.calls=sysstat['MaxSimCalls']
|
|
self.System.License.meetingcalls=sysstat['MaxSimMeetingParticipants']
|
|
self.System.License.supported=sysstat['Support']
|
|
self.System.License.expire=sysstat['ExpirationDate']
|
|
self.System.License.maintexpire=sysstat['MaintenanceExpiresAt']
|
|
self.System.License.reseller=sysstat['ResellerName']
|
|
self.System.License.key=sysstat['LicenseKey']
|
|
self.System.License.sku=sysstat['ProductCode']
|
|
self.System.License.spla=sysstat['IsSpla']
|
|
def refresh_addl(self):
|
|
addl=self.tcx.rq.get('SystemStatus/AdditionalStatus').json()
|
|
self.Status.callrecs_bytes=addl['RecordingUsedSpace']
|
|
self.System.callrecs_enabled=not addl['RecordingStopped']
|
|
self.System.callrecs_quota=addl['RecordingQuota']
|
|
def refresh_inroute(self):
|
|
inroute=self.tcx.rq.get('InboundRulesList').json()['list']
|
|
self.System.routes_in=len(inroute)
|
|
def refresh_hists(self):
|
|
hists=self.tcx.rq.post('SystemStatus/getDbInformation', params={}).json()
|
|
self.System.calls_total=hists['CallHistoryCount']
|
|
self.System.chats_total=hists['ChatMessagesCount']
|
|
def refresh_certified(self):
|
|
certstat=self.tcx.rq.get('SystemStatus/GetSingleStatus').json()
|
|
self.System.License.firewallsupported=certstat['Health']['Firewall']
|
|
self.System.License.trunkssupported=certstat['Health']['Trunks']
|
|
self.System.License.phonessupported=certstat['Health']['Phones']
|
|
def refresh(self):
|
|
self.refresh_addl()
|
|
self.refresh_certified()
|
|
self.refresh_hists()
|
|
self.refresh_inroute()
|
|
self.refresh_sysstat()
|
|
@property
|
|
def list_extensions(self) -> typing.List['Py3CX._Extension']:
|
|
"""Returns a list of Py3CX.Extension objects
|
|
|
|
Returns a list of Py3CX.Extension objects representing every configured extension on a 3CX phone system
|
|
|
|
>>> tcx.system.list_extensions
|
|
[<Py3CX.Extension[0000]>, <Py3CX.Extension[1234]>]
|
|
"""
|
|
if self.extlist is None:
|
|
self.extlist=ReadOnlyObject(self.tcx, api='ExtensionList')
|
|
self.extlist.refresh()
|
|
res=self.extlist._result.json()['list']
|
|
assert (len(res)>0), "No extensions found"
|
|
ret=[]
|
|
for extension in res:
|
|
this=Py3CX._Extension(self.tcx, params=extension['Number'], populate=False)
|
|
this.firstname=extension['FirstName']
|
|
this.surname=extension['LastName']
|
|
this.name="%s %s" % (this.firstname, this.surname)
|
|
this.dnd=extension['DND']
|
|
this.status=extension['CurrentProfile']
|
|
this.mail=extension['Email']
|
|
this.cli=extension['OutboundCallerId']
|
|
this.mobile=extension['MobileNumber']
|
|
this.online=extension['IsRegistered']
|
|
ret.append(this)
|
|
return ret
|
|
@property
|
|
def list_calls(self) -> typing.List['Py3CX._Call']:
|
|
"""Returns a list of Py3CX.Call objects
|
|
|
|
Returns a list of Py3CX.Call objects representing all current active calls on the phone system
|
|
|
|
>>> tcx.system.list_calls
|
|
[<Py3CX.Call [1234 Dialing]>]
|
|
"""
|
|
if self.calllist is None:
|
|
self.calllist=ReadOnlyObject(self.tcx, api='activeCalls')
|
|
self.calllist.refresh()
|
|
res=self.calllist._result.json()['list']
|
|
if len(res)==0: return []
|
|
ret=[]
|
|
for call in res:
|
|
this=Py3CX._Call(self.tcx, call['Id'])
|
|
this.id=call['Id']
|
|
this.caller=call['Caller']
|
|
this.callee=call['Callee']
|
|
this.state=call['Status']
|
|
this.duration=call['Duration']
|
|
this.since=call['LastChangeStatus']
|
|
ret.append(this)
|
|
return ret
|
|
def __init__(self, uri=None, tls_verify : bool =True):
|
|
"""Returns a Py3CX class object instance
|
|
|
|
Returns an instance of the Py3CX class, which is the primary entrypoint of this module.
|
|
|
|
Parameters
|
|
---
|
|
uri: Optional HTTP/HTTPS URI to your 3CX FQDN. Do not include the /api/. If not explicitly specified, this can instead be set through environment variables
|
|
tls_verify: Optional boolean attribute indicating whether the TLS certificate presented by the 3CX phone system should be validated against internal root trust
|
|
|
|
>>> tcx=Py3CX(uri='http://myinstance.3cx.com')
|
|
"""
|
|
self.cnf=Py3CXEnvironmentVars()
|
|
self._tcxsystem=None
|
|
if uri is not None:
|
|
self.cnf.uri=uri
|
|
try:
|
|
assert (self.cnf.uri is not None and self.cnf.uri.startswith('http')), "No or invalid URI specified, please provide via uri= or by setting environment variable %s" % self.cnf.URI
|
|
except AssertionError as e:
|
|
raise ValidationError(str(e))
|
|
self.rq=Request(uri=self.cnf.uri, verify=tls_verify)
|
|
def __repr__(self):
|
|
return '<Py3CX [%s@%s]>' % (self.cnf.auth_user, self.cnf.uri)
|
|
def authenticate(self, username=None, password=None) -> None:
|
|
"""Authenticates against the given API, optionally using explicit credentials.
|
|
|
|
Performs a POST request to the 'login' endpoint of the current API server, using either credentials gleaned through the execution environment, or explicitly specified during the function invocation.
|
|
|
|
>>> tcx.authenticate()
|
|
>>> tcx.authenticate(username='admin', password='password')
|
|
|
|
"""
|
|
if username is not None:
|
|
self.cnf.auth_user=username
|
|
if password is not None:
|
|
self.cnf.auth_pwd=password
|
|
try:
|
|
assert (None not in (self.cnf.auth_user, self.cnf.auth_pwd))
|
|
except AssertionError as e:
|
|
raise ValidationError(str(e))
|
|
rs=self.rq.post('login', params={
|
|
'Username': self.cnf.auth_user,
|
|
'Password': self.cnf.auth_pwd})
|
|
self.rq.sess.headers.update({'x-xsrf-token':rs.cookies['XSRF-TOKEN']})
|
|
@property
|
|
def authenticated(self) -> bool:
|
|
"""Boolean property noting if the current session is logged-in.
|
|
|
|
Returns a boolean value indicating whether the Py3CX instance is authenticated against the configured 3CX server, by performing a GET request to the 'CurrentUser' API endpoint.
|
|
The expected response code is HTTP 200. If that expectation is not met, an exception is thrown which this function translates to a boolean value.
|
|
|
|
>>> tcx.authenticated
|
|
False
|
|
>>> tcx.authenticate()
|
|
>>> tcx.authenticated
|
|
True
|
|
"""
|
|
try:
|
|
self.rq.get('CurrentUser')
|
|
except APIError:
|
|
return False
|
|
return True
|
|
@property
|
|
def system(self) -> 'Py3CX._PhoneSystem':
|
|
"""Returns a Py3CX.System object
|
|
|
|
Returns the current instance of (and initialises a new instance of, if one does not exist yet) the Py3CX System class, which contains functionality for system-wide 3CX APIs
|
|
|
|
>>> tcx.system
|
|
<Py3CX.System [http://myinstance.3cx.com]>
|
|
"""
|
|
assert self.authenticated, "Py3CX not authenticated yet!"
|
|
if self._tcxsystem is None:
|
|
self._tcxsystem=Py3CX._PhoneSystem(self)
|
|
return self._tcxsystem
|
|
|