diff --git a/sipsimple/core/_core.mediatransport.pxi b/sipsimple/core/_core.mediatransport.pxi index cabd1c44..e64ee968 100644 --- a/sipsimple/core/_core.mediatransport.pxi +++ b/sipsimple/core/_core.mediatransport.pxi @@ -1,2563 +1,2558 @@ import sys from errno import EADDRINUSE # classes cdef class RTPTransport: def __cinit__(self, *args, **kwargs): cdef int status cdef pj_pool_t *pool cdef bytes pool_name cdef char* c_pool_name cdef PJSIPUA ua ua = _get_ua() pool_name = b"RTPTransport_%d" % id(self) self.weakref = weakref.ref(self) Py_INCREF(self.weakref) self._af = pj_AF_INET() status = pj_mutex_create_recursive(ua._pjsip_endpoint._pool, "rtp_transport_lock", &self._lock) if status != 0: raise PJSIPError("failed to create lock", status) pool = ua.create_memory_pool(pool_name, 4096, 4096) self._pool = pool self.state = "NULL" def __init__(self, encryption=None, use_ice=False, ice_stun_address=None, ice_stun_port=PJ_STUN_PORT): cdef PJSIPUA ua = _get_ua() if self.state != "NULL": raise SIPCoreError("RTPTransport.__init__() was already called") self._rtp_valid_pair = None self._encryption = encryption self.use_ice = use_ice self.ice_stun_address = ice_stun_address self.ice_stun_port = ice_stun_port def __dealloc__(self): cdef PJSIPUA ua cdef pjmedia_transport *transport cdef Timer timer try: ua = _get_ua() except: return transport = self._obj if transport != NULL: transport.user_data = NULL if self._wrapped_transport != NULL: self._wrapped_transport.user_data = NULL with nogil: pjmedia_transport_media_stop(transport) pjmedia_transport_close(transport) self._obj = NULL self._wrapped_transport = NULL ua.release_memory_pool(self._pool) self._pool = NULL if self._lock != NULL: pj_mutex_destroy(self._lock) timer = Timer() try: timer.schedule(60, deallocate_weakref, self.weakref) except SIPCoreError: pass cdef PJSIPUA _check_ua(self): cdef PJSIPUA ua try: ua = _get_ua() return ua except: self.state = "INVALID" self._obj = NULL self._wrapped_transport = NULL self._pool = NULL return None cdef void _get_info(self, pjmedia_transport_info *info): cdef int status cdef pjmedia_transport *transport transport = self._obj with nogil: pjmedia_transport_info_init(info) status = pjmedia_transport_get_info(transport, info) if status != 0: raise PJSIPError("Could not get transport info", status) property local_rtp_port: def __get__(self): cdef int status cdef pj_mutex_t *lock = self._lock cdef pjmedia_transport_info info cdef PJSIPUA ua ua = self._check_ua() if ua is None: return None with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: if self.state in ["NULL", "WAIT_STUN", "INVALID"]: return None self._get_info(&info) if pj_sockaddr_has_addr(&info.sock_info.rtp_addr_name): return pj_sockaddr_get_port(&info.sock_info.rtp_addr_name) else: return None finally: with nogil: pj_mutex_unlock(lock) property local_rtp_address: def __get__(self): cdef char buf[PJ_INET6_ADDRSTRLEN] cdef int status cdef pj_mutex_t *lock = self._lock cdef pjmedia_transport_info info cdef PJSIPUA ua ua = self._check_ua() if ua is None: return None with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: if self.state in ["NULL", "WAIT_STUN", "INVALID"]: return None self._get_info(&info) if pj_sockaddr_has_addr(&info.sock_info.rtp_addr_name): return pj_sockaddr_print(&info.sock_info.rtp_addr_name, buf, PJ_INET6_ADDRSTRLEN, 0) else: return None finally: with nogil: pj_mutex_unlock(lock) property local_rtp_candidate: def __get__(self): cdef int status cdef pj_mutex_t *lock = self._lock cdef PJSIPUA ua ua = self._check_ua() if ua is None: return None with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: if self._rtp_valid_pair: return self._rtp_valid_pair.local_candidate return None finally: with nogil: pj_mutex_unlock(lock) property remote_rtp_port: def __get__(self): cdef int status cdef pj_mutex_t *lock = self._lock cdef pjmedia_transport_info info cdef PJSIPUA ua ua = self._check_ua() if ua is None: return None with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: if self.state in ["NULL", "WAIT_STUN", "INVALID"]: return None if self._ice_active() and self._rtp_valid_pair: return self._rtp_valid_pair.remote_candidate.port self._get_info(&info) if pj_sockaddr_has_addr(&info.src_rtp_name): return pj_sockaddr_get_port(&info.src_rtp_name) else: return None finally: with nogil: pj_mutex_unlock(lock) property remote_rtp_address: def __get__(self): cdef char buf[PJ_INET6_ADDRSTRLEN] cdef int status cdef pj_mutex_t *lock = self._lock cdef pjmedia_transport_info info cdef PJSIPUA ua ua = self._check_ua() if ua is None: return None with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: if self.state in ["NULL", "WAIT_STUN", "INVALID"]: return None if self._ice_active() and self._rtp_valid_pair: return self._rtp_valid_pair.remote_candidate.address self._get_info(&info) if pj_sockaddr_has_addr(&info.src_rtp_name): return pj_sockaddr_print(&info.src_rtp_name, buf, PJ_INET6_ADDRSTRLEN, 0) else: return None finally: with nogil: pj_mutex_unlock(lock) property remote_rtp_candidate: def __get__(self): cdef int status cdef pj_mutex_t *lock = self._lock cdef PJSIPUA ua ua = self._check_ua() if ua is None: return None with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: if self._rtp_valid_pair: return self._rtp_valid_pair.remote_candidate return None finally: with nogil: pj_mutex_unlock(lock) property srtp_active: def __get__(self): cdef int status cdef pj_mutex_t *lock = self._lock cdef pjmedia_srtp_info *srtp_info cdef pjmedia_transport_info info cdef PJSIPUA ua ua = self._check_ua() if ua is None: return False with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: if self.state in ["NULL", "WAIT_STUN", "INVALID"]: return False self._get_info(&info) srtp_info = pjmedia_transport_info_get_spc_info(&info, PJMEDIA_TRANSPORT_TYPE_SRTP) if srtp_info != NULL: return bool(srtp_info.active) return False finally: with nogil: pj_mutex_unlock(lock) property srtp_cipher: def __get__(self): cdef int status cdef pj_mutex_t *lock = self._lock cdef pjmedia_srtp_info *srtp_info cdef pjmedia_transport_info info cdef PJSIPUA ua ua = self._check_ua() if ua is None: return None with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: if self.state in ["NULL", "WAIT_STUN", "INVALID"]: return None self._get_info(&info) srtp_info = pjmedia_transport_info_get_spc_info(&info, PJMEDIA_TRANSPORT_TYPE_SRTP) if srtp_info == NULL or not bool(srtp_info.active): return None return _pj_str_to_bytes(srtp_info.tx_policy.name) finally: with nogil: pj_mutex_unlock(lock) property zrtp_active: def __get__(self): cdef int status cdef pj_mutex_t *lock = self._lock cdef pjmedia_zrtp_info *zrtp_info cdef pjmedia_transport_info info cdef PJSIPUA ua ua = self._check_ua() if ua is None: return False with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: if self.state in ["NULL", "WAIT_STUN", "INVALID"]: return False self._get_info(&info) zrtp_info = pjmedia_transport_info_get_spc_info(&info, PJMEDIA_TRANSPORT_TYPE_ZRTP) if zrtp_info != NULL: return bool(zrtp_info.active) return False finally: with nogil: pj_mutex_unlock(lock) cdef int _ice_active(self): # this function needs to be called with the lock held cdef pjmedia_transport_info info cdef pjmedia_ice_transport_info *ice_info if self.state in ["NULL", "WAIT_STUN", "INVALID"]: return 0 self._get_info(&info) ice_info = pjmedia_transport_info_get_spc_info(&info, PJMEDIA_TRANSPORT_TYPE_ICE) if ice_info != NULL and ice_info.sess_state == PJ_ICE_STRANS_STATE_RUNNING: return 1 return 0 property ice_active: def __get__(self): cdef int status cdef pj_mutex_t *lock = self._lock cdef PJSIPUA ua ua = self._check_ua() if ua is None: return False with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: return bool(self._ice_active()) finally: with nogil: pj_mutex_unlock(lock) cdef int _init_local_sdp(self, BaseSDPSession local_sdp, BaseSDPSession remote_sdp, int sdp_index): cdef int status cdef pj_pool_t *pool cdef pjmedia_sdp_session *pj_local_sdp cdef pjmedia_sdp_session *pj_remote_sdp cdef pjmedia_transport *transport pool = self._pool transport = self._obj pj_local_sdp = local_sdp.get_sdp_session() if remote_sdp is not None: pj_remote_sdp = remote_sdp.get_sdp_session() else: pj_remote_sdp = NULL if sdp_index < 0: raise ValueError("sdp_index argument cannot be negative") if sdp_index >= pj_local_sdp.media_count: raise ValueError("sdp_index argument out of range") with nogil: status = pjmedia_transport_media_create(transport, pool, 0, pj_remote_sdp, sdp_index) if status != 0: raise PJSIPError("Could not create media transport", status) return 0 def set_LOCAL(self, SDPSession local_sdp, int sdp_index): cdef int status cdef pj_mutex_t *lock = self._lock _get_ua() with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: if local_sdp is None: raise SIPCoreError("local_sdp argument cannot be None") if self.state == "LOCAL": return if self.state != "INIT": raise SIPCoreError('set_LOCAL can only be called in the "INIT" state, current state is "%s"' % self.state) self._init_local_sdp(local_sdp, None, sdp_index) self.state = "LOCAL" finally: with nogil: pj_mutex_unlock(lock) def set_REMOTE(self, BaseSDPSession local_sdp, BaseSDPSession remote_sdp, int sdp_index): cdef int status cdef pj_mutex_t *lock = self._lock _get_ua() with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: if None in [local_sdp, remote_sdp]: raise SIPCoreError("SDP arguments cannot be None") if self.state == "REMOTE": return if self.state != "INIT": raise SIPCoreError('set_REMOTE can only be called in the "INIT" state, current state is "%s"' % self.state) self._init_local_sdp(local_sdp, remote_sdp, sdp_index) self.state = "REMOTE" finally: with nogil: pj_mutex_unlock(lock) def set_ESTABLISHED(self, BaseSDPSession local_sdp, BaseSDPSession remote_sdp, int sdp_index): cdef int status cdef pj_mutex_t *lock = self._lock cdef pjmedia_sdp_session *pj_local_sdp cdef pjmedia_sdp_session *pj_remote_sdp cdef pjmedia_transport *transport = self._obj _get_ua() with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: transport = self._obj if None in [local_sdp, remote_sdp]: raise SIPCoreError("SDP arguments cannot be None") pj_local_sdp = local_sdp.get_sdp_session() pj_remote_sdp = remote_sdp.get_sdp_session() if self.state == "ESTABLISHED": return if self.state not in ["LOCAL", "REMOTE"]: raise SIPCoreError('set_ESTABLISHED can only be called in the "INIT" and "LOCAL" states, ' + 'current state is "%s"' % self.state) with nogil: status = pjmedia_transport_media_start(transport, self._pool, pj_local_sdp, pj_remote_sdp, sdp_index) if status != 0: raise PJSIPError("Could not start media transport", status) self.state = "ESTABLISHED" finally: with nogil: pj_mutex_unlock(lock) def set_INIT(self): global _ice_cb cdef int af cdef int i cdef int status cdef int port cdef pj_caching_pool *caching_pool cdef pj_ice_strans_cfg ice_cfg cdef pj_ice_strans *ice_st cdef pj_ice_strans_state ice_state cdef pj_mutex_t *lock = self._lock cdef pj_str_t local_ip cdef pj_str_t *local_ip_address cdef pjmedia_endpt *media_endpoint cdef pjmedia_srtp_setting srtp_setting cdef pjmedia_transport **transport_address cdef pjmedia_transport *wrapped_transport cdef pjsip_endpoint *sip_endpoint cdef bytes zid_file cdef char *c_zid_file cdef PJSIPUA ua ua = _get_ua() with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: af = self._af caching_pool = &ua._caching_pool._obj media_endpoint = ua._pjmedia_endpoint._obj sip_endpoint = ua._pjsip_endpoint._obj transport_address = &self._obj if self.state == "INIT": return if self.state in ["LOCAL", "ESTABLISHED"]: with nogil: status = pjmedia_transport_media_stop(transport_address[0]) if status != 0: raise PJSIPError("Could not stop media transport", status) self.state = "INIT" elif self.state == "NULL": if ua.ip_address is None: local_ip_address = NULL else: _str_to_pj_str(ua.ip_address, &local_ip) local_ip_address = &local_ip if self.use_ice: with nogil: pj_ice_strans_cfg_default(&ice_cfg) ice_cfg.af = self._af with nogil: pj_stun_config_init(&ice_cfg.stun_cfg, &caching_pool.factory, 0, pjmedia_endpt_get_ioqueue(media_endpoint), pjsip_endpt_get_timer_heap(sip_endpoint)) if self.ice_stun_address is not None: _str_to_pj_str(self.ice_stun_address, &ice_cfg.stun.server) ice_cfg.stun.port = self.ice_stun_port # IIRC we can't choose the port for ICE with nogil: status = pj_sockaddr_init(ice_cfg.af, &ice_cfg.stun.cfg.bound_addr, local_ip_address, 0) if status != 0: raise PJSIPError("Could not init ICE bound address", status) with nogil: status = pjmedia_ice_create2(media_endpoint, NULL, 2, &ice_cfg, &_ice_cb, 0, transport_address) if status != 0: raise PJSIPError("Could not create ICE media transport", status) else: status = PJ_EBUG for i in xrange(ua._rtp_port_index, ua._rtp_port_index + ua._rtp_port_usable_count, 2): port = ua._rtp_port_start + i % ua._rtp_port_usable_count with nogil: status = pjmedia_transport_udp_create3(media_endpoint, af, NULL, local_ip_address, port, 0, transport_address) if status != PJ_ERRNO_START_SYS + EADDRINUSE: ua._rtp_port_index = (i + 2) % ua._rtp_port_usable_count break if status != 0: raise PJSIPError("Could not create UDP/RTP media transport", status) self._obj.user_data = self.weakref if self._encryption is not None: wrapped_transport = self._wrapped_transport = self._obj self._obj = NULL if self._encryption.startswith('sdes'): with nogil: pjmedia_srtp_setting_default(&srtp_setting) if self._encryption == 'sdes_mandatory': srtp_setting.use = PJMEDIA_SRTP_MANDATORY with nogil: status = pjmedia_transport_srtp_create(media_endpoint, wrapped_transport, &srtp_setting, transport_address) if status != 0: with nogil: pjmedia_transport_close(wrapped_transport) self._wrapped_transport = NULL raise PJSIPError("Could not create SRTP media transport", status) elif self._encryption == 'zrtp': with nogil: status = pjmedia_transport_zrtp_create(media_endpoint, pjsip_endpt_get_timer_heap(sip_endpoint), wrapped_transport, transport_address, 1) if status == 0: zid_file = ua.zrtp_cache.encode(sys.getfilesystemencoding()) c_zid_file = zid_file with nogil: # Auto-enable is deactivated status = pjmedia_transport_zrtp_initialize(self._obj, c_zid_file, 0, &_zrtp_cb) if status != 0: with nogil: pjmedia_transport_close(wrapped_transport) self._wrapped_transport = NULL raise PJSIPError("Could not create ZRTP media transport", status) else: raise RuntimeError('invalid SRTP key negotiation specified: %s' % self._encryption) self._obj.user_data = self.weakref if not self.use_ice or self.ice_stun_address is None: self.state = "INIT" _add_event("RTPTransportDidInitialize", dict(obj=self)) else: self.state = "WAIT_STUN" if self.use_ice: _add_event("RTPTransportICENegotiationStateDidChange", dict(obj=self, prev_state="NULL", state="GATHERING")) ice_st = pjmedia_ice_get_strans(transport_address[0]) if ice_st != NULL: ice_state = pj_ice_strans_get_state(ice_st) if ice_state == PJ_ICE_STRANS_STATE_READY: _add_event("RTPTransportICENegotiationStateDidChange", dict(obj=self, prev_state="GATHERING", state="GATHERING_COMPLETE")) else: raise SIPCoreError('set_INIT can only be called in the "NULL", "LOCAL" and "ESTABLISHED" states, ' + 'current state is "%s"' % self.state) finally: with nogil: pj_mutex_unlock(lock) def set_zrtp_sas_verified(self, verified): cdef int status cdef int c_verified cdef pj_mutex_t *lock = self._lock cdef pjmedia_zrtp_info *zrtp_info cdef pjmedia_transport_info info cdef PJSIPUA ua ua = self._check_ua() if ua is None: return False with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: if self.state in ["NULL", "WAIT_STUN", "INVALID"]: return False self._get_info(&info) zrtp_info = pjmedia_transport_info_get_spc_info(&info, PJMEDIA_TRANSPORT_TYPE_ZRTP) if zrtp_info == NULL or not bool(zrtp_info.active): return False c_verified = int(verified) with nogil: pjmedia_transport_zrtp_setSASVerified(self._obj, c_verified) return True finally: with nogil: pj_mutex_unlock(lock) def set_zrtp_enabled(self, enabled, object master_stream): cdef int status cdef int c_enabled cdef pj_mutex_t *lock = self._lock cdef pjmedia_zrtp_info *zrtp_info cdef pjmedia_transport_info info cdef PJSIPUA ua cdef bytes multistream_params cdef char *c_multistream_params cdef int length cdef RTPTransport master_transport ua = self._check_ua() if ua is None: return with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: if self.state in ["NULL", "WAIT_STUN", "INVALID"]: return self._get_info(&info) zrtp_info = pjmedia_transport_info_get_spc_info(&info, PJMEDIA_TRANSPORT_TYPE_ZRTP) if zrtp_info == NULL: return if master_stream is not None: master_transport = master_stream._rtp_transport assert master_transport is not None # extract the multistream parameters multistream_params = master_transport.zrtp_multistream_parameters if multistream_params: # set multistream mode in ourselves c_multistream_params = multistream_params length = len(multistream_params) with nogil: pjmedia_transport_zrtp_setMultiStreamParameters(self._obj, c_multistream_params, length, master_transport._obj) c_enabled = int(enabled) with nogil: pjmedia_transport_zrtp_setEnableZrtp(self._obj, c_enabled) finally: with nogil: pj_mutex_unlock(lock) property zrtp_multistream_parameters: def __get__(self): cdef int status cdef char* c_name cdef pj_mutex_t *lock = self._lock cdef pjmedia_zrtp_info *zrtp_info cdef pjmedia_transport_info info cdef PJSIPUA ua cdef char *multistr_params cdef int length ua = self._check_ua() if ua is None: return None with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: if self.state in ["NULL", "WAIT_STUN", "INVALID"]: return None self._get_info(&info) zrtp_info = pjmedia_transport_info_get_spc_info(&info, PJMEDIA_TRANSPORT_TYPE_ZRTP) if zrtp_info == NULL or not bool(zrtp_info.active): return None with nogil: multistr_params = pjmedia_transport_zrtp_getMultiStreamParameters(self._obj, &length) if length > 0: ret = _pj_buf_len_to_str(multistr_params, length) free(multistr_params) return ret else: return None finally: with nogil: pj_mutex_unlock(lock) property zrtp_cipher: def __get__(self): cdef int status cdef char* c_name cdef pj_mutex_t *lock = self._lock cdef pjmedia_zrtp_info *zrtp_info cdef pjmedia_transport_info info cdef PJSIPUA ua ua = self._check_ua() if ua is None: return None with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: if self.state in ["NULL", "WAIT_STUN", "INVALID"]: return None self._get_info(&info) zrtp_info = pjmedia_transport_info_get_spc_info(&info, PJMEDIA_TRANSPORT_TYPE_ZRTP) if zrtp_info == NULL or not bool(zrtp_info.active): return None return _buf_to_str(zrtp_info.cipher) finally: with nogil: pj_mutex_unlock(lock) property zrtp_peer_name: def __get__(self): cdef int status cdef char* c_name cdef pj_mutex_t *lock = self._lock cdef pjmedia_zrtp_info *zrtp_info cdef pjmedia_transport_info info cdef PJSIPUA ua ua = self._check_ua() if ua is None: return '' with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: if self.state in ["NULL", "WAIT_STUN", "INVALID"]: return '' self._get_info(&info) zrtp_info = pjmedia_transport_info_get_spc_info(&info, PJMEDIA_TRANSPORT_TYPE_ZRTP) if zrtp_info == NULL or not bool(zrtp_info.active): return '' with nogil: c_name = pjmedia_transport_zrtp_getPeerName(self._obj) if c_name == NULL: return '' else: name = PyUnicode_FromString(c_name) or u'' free(c_name) return name finally: with nogil: pj_mutex_unlock(lock) def __set__(self, basestring name): cdef int status cdef char* c_name cdef pj_mutex_t *lock = self._lock cdef pjmedia_zrtp_info *zrtp_info cdef pjmedia_transport_info info cdef PJSIPUA ua ua = self._check_ua() if ua is None: return with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: if self.state in ["NULL", "WAIT_STUN", "INVALID"]: return self._get_info(&info) zrtp_info = pjmedia_transport_info_get_spc_info(&info, PJMEDIA_TRANSPORT_TYPE_ZRTP) if zrtp_info == NULL or not bool(zrtp_info.active): return name = name.encode('utf-8') c_name = name with nogil: pjmedia_transport_zrtp_putPeerName(self._obj, c_name) finally: with nogil: pj_mutex_unlock(lock) property zrtp_peer_id: def __get__(self): cdef int status cdef unsigned char name[12] # IDENTIFIER_LEN, 96bits cdef pj_mutex_t *lock = self._lock cdef pjmedia_zrtp_info *zrtp_info cdef pjmedia_transport_info info cdef PJSIPUA ua ua = self._check_ua() if ua is None: return None with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: if self.state in ["NULL", "WAIT_STUN", "INVALID"]: return None self._get_info(&info) zrtp_info = pjmedia_transport_info_get_spc_info(&info, PJMEDIA_TRANSPORT_TYPE_ZRTP) if zrtp_info == NULL or not bool(zrtp_info.active): return None with nogil: status = pjmedia_transport_zrtp_getPeerZid(self._obj, name) if status <= 0: return None else: return _pj_buf_len_to_str(name, 12) finally: with nogil: pj_mutex_unlock(lock) def update_local_sdp(self, SDPSession local_sdp, BaseSDPSession remote_sdp=None, int sdp_index=0): - print('Cython mediatransport update_local_sdp') cdef int status cdef pj_pool_t *pool cdef pjmedia_sdp_session *pj_local_sdp cdef pjmedia_sdp_session *pj_remote_sdp cdef pjmedia_transport *transport cdef SDPMediaStream local_media pool = self._pool transport = self._obj pj_local_sdp = local_sdp.get_sdp_session() if remote_sdp is not None: pj_remote_sdp = remote_sdp.get_sdp_session() else: pj_remote_sdp = NULL if sdp_index < 0: raise ValueError("sdp_index argument cannot be negative") if sdp_index >= pj_local_sdp.media_count: raise ValueError("sdp_index argument out of range") # Remove ICE and SRTP/ZRTP related attributes from SDP, they will be added by pjmedia_transport_encode_sdp - print('Cython mediatransport update_local_sdp 1') local_media = local_sdp.media[sdp_index] local_media.attributes = [ attr for attr in local_media.attributes if attr.name not in ('crypto', 'zrtp-hash', 'ice-ufrag', 'ice-pwd', 'ice-mismatch', 'candidate', 'remote-candidates')] - print('Cython mediatransport update_local_sdp 2') pj_local_sdp = local_sdp.get_sdp_session() - print('Cython mediatransport update_local_sdp 3') with nogil: status = pjmedia_transport_encode_sdp(transport, pool, pj_local_sdp, pj_remote_sdp, sdp_index) if status != 0: raise PJSIPError("Could not update SDP for media transport", status) - print('Cython mediatransport update_local_sdp 4') local_sdp._update() return 0 cdef class MediaCheckTimer(Timer): def __init__(self, media_check_interval): self.media_check_interval = media_check_interval cdef class SDPInfo: def __init__(self, BaseSDPMediaStream local_media=None, BaseSDPSession local_sdp=None, BaseSDPSession remote_sdp=None, int index=0): self.local_media = local_media self.local_sdp = local_sdp self.remote_sdp = remote_sdp self.index = index property local_media: def __get__(self): return self._local_media def __set__(self, local_media): if local_media is not None: local_media = SDPMediaStream.new(local_media) self._local_media = local_media property local_sdp: def __get__(self): return self._local_sdp def __set__(self, local_sdp): if local_sdp is not None: local_sdp = SDPSession.new(local_sdp) self._local_sdp = local_sdp property remote_sdp: def __get__(self): return self._remote_sdp def __set__(self, remote_sdp): if remote_sdp is not None: remote_sdp = SDPSession.new(remote_sdp) self._remote_sdp = remote_sdp cdef class AudioTransport: def __cinit__(self, *args, **kwargs): cdef int status cdef pj_pool_t *pool cdef bytes pool_name cdef char* c_pool_name cdef PJSIPUA ua ua = _get_ua() pool_name = b"AudioTransport_%d" % id(self) self.weakref = weakref.ref(self) Py_INCREF(self.weakref) status = pj_mutex_create_recursive(ua._pjsip_endpoint._pool, "audio_transport_lock", &self._lock) if status != 0: raise PJSIPError("failed to create lock", status) pool = ua.create_memory_pool(pool_name, 4096, 4096) self._pool = pool self._slot = -1 self._timer = None self._volume = 100 def __init__(self, AudioMixer mixer, RTPTransport transport, BaseSDPSession remote_sdp=None, int sdp_index=0, enable_silence_detection=False, list codecs=None): cdef int status cdef pj_pool_t *pool cdef pjmedia_endpt *media_endpoint cdef pjmedia_sdp_media *local_media_c cdef pjmedia_sdp_session *local_sdp_c cdef pj_sockaddr *addr cdef pjmedia_transport_info info cdef list global_codecs cdef SDPMediaStream local_media cdef SDPSession local_sdp cdef PJSIPUA ua ua = _get_ua() media_endpoint = ua._pjmedia_endpoint._obj pool = self._pool if self.transport is not None: raise SIPCoreError("AudioTransport.__init__() was already called") if mixer is None: raise ValueError("mixer argument may not be None") if transport is None: raise ValueError("transport argument cannot be None") if sdp_index < 0: raise ValueError("sdp_index argument cannot be negative") if transport.state != "INIT": raise SIPCoreError('RTPTransport object provided is not in the "INIT" state, but in the "%s" state' % transport.state) self._vad = int(bool(enable_silence_detection)) self.mixer = mixer self.transport = transport transport._get_info(&info) global_codecs = ua._pjmedia_endpoint._get_current_codecs() if codecs is None: codecs = global_codecs try: ua._pjmedia_endpoint._set_codecs(codecs) addr = &info.sock_info.rtp_addr_name with nogil: status = pjmedia_endpt_create_base_sdp(media_endpoint, pool, NULL, addr, &local_sdp_c) if status != 0: raise PJSIPError("Could not generate base SDP", status) with nogil: status = pjmedia_endpt_create_audio_sdp(media_endpoint, pool, &info.sock_info, 0, &local_media_c) if status != 0: raise PJSIPError("Could not generate SDP audio stream", status) # Create a 'fake' SDP, which only contains the audio stream, then the m line is extracted because the full # SDP is built by the Session local_sdp_c.media_count = 1 local_sdp_c.media[0] = local_media_c finally: ua._pjmedia_endpoint._set_codecs(global_codecs) local_sdp = SDPSession_create(local_sdp_c) local_media = local_sdp.media[0] if remote_sdp is None: self._is_offer = 1 self.transport.set_LOCAL(local_sdp, 0) else: self._is_offer = 0 if sdp_index != 0: local_sdp.media = [None] * (sdp_index+1) local_sdp.media[sdp_index] = local_media self.transport.set_REMOTE(local_sdp, remote_sdp, sdp_index) self._sdp_info = SDPInfo(local_media, local_sdp, remote_sdp, sdp_index) def __dealloc__(self): cdef PJSIPUA ua cdef Timer timer try: ua = _get_ua() except: return if self._obj != NULL: self.stop() ua.release_memory_pool(self._pool) self._pool = NULL if self._lock != NULL: pj_mutex_destroy(self._lock) timer = Timer() try: timer.schedule(60, deallocate_weakref, self.weakref) except SIPCoreError: pass cdef PJSIPUA _check_ua(self): cdef PJSIPUA ua try: ua = _get_ua() return ua except: self._obj = NULL self._pool = NULL return None property is_active: def __get__(self): self._check_ua() return bool(self._obj != NULL) property is_started: def __get__(self): return bool(self._is_started) property codec: def __get__(self): self._check_ua() if self._obj == NULL: return None else: return _pj_str_to_bytes(self._stream_info.fmt.encoding_name) property sample_rate: def __get__(self): self._check_ua() if self._obj == NULL: return None else: return self._stream_info.fmt.clock_rate property enable_silence_detection: def __get__(self): return bool(self._vad) property statistics: def __get__(self): cdef int status cdef pj_mutex_t *lock = self._lock cdef pjmedia_rtcp_stat stat cdef pjmedia_stream *stream cdef dict statistics = dict() cdef PJSIPUA ua ua = self._check_ua() if ua is None: return None with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: stream = self._obj if stream == NULL: return None with nogil: status = pjmedia_stream_get_stat(stream, &stat) if status != 0: raise PJSIPError("Could not get RTP statistics", status) statistics["rtt"] = _pj_math_stat_to_dict(&stat.rtt) statistics["rx"] = _pjmedia_rtcp_stream_stat_to_dict(&stat.rx) statistics["tx"] = _pjmedia_rtcp_stream_stat_to_dict(&stat.tx) return statistics finally: with nogil: pj_mutex_unlock(lock) property volume: def __get__(self): return self._volume def __set__(self, value): cdef int slot cdef int status cdef int volume cdef pj_mutex_t *lock = self._lock cdef pjmedia_conf *conf_bridge cdef PJSIPUA ua ua = self._check_ua() if ua is not None: with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: conf_bridge = self.mixer._obj slot = self._slot if value < 0: raise ValueError("volume attribute cannot be negative") if ua is not None and self._obj != NULL: volume = int(value * 1.28 - 128) with nogil: status = pjmedia_conf_adjust_rx_level(conf_bridge, slot, volume) if status != 0: raise PJSIPError("Could not set volume of audio transport", status) self._volume = value finally: if ua is not None: with nogil: pj_mutex_unlock(lock) property slot: def __get__(self): self._check_ua() if self._slot == -1: return None else: return self._slot def get_local_media(self, BaseSDPSession remote_sdp=None, int index=0, direction="sendrecv"): global valid_sdp_directions cdef int status cdef pj_mutex_t *lock = self._lock cdef object direction_attr cdef SDPAttribute attr cdef SDPSession local_sdp cdef SDPMediaStream local_media cdef pjmedia_sdp_media *c_local_media _get_ua() with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: is_offer = remote_sdp is None if is_offer and direction not in valid_sdp_directions: raise SIPCoreError("Unknown direction: %s" % direction) self._sdp_info.index = index local_sdp = self._sdp_info.local_sdp local_media = self._sdp_info.local_media local_sdp.media = [None] * (index+1) local_sdp.media[index] = local_media self.transport.update_local_sdp(local_sdp, remote_sdp, index) # updating the local SDP might have modified the connection line if local_sdp.connection is not None and local_media.connection is None: local_media.connection = SDPConnection.new(local_sdp.connection) local_media.attributes = [ attr for attr in local_media.attributes if attr.name not in valid_sdp_directions] if is_offer: direction_attr = direction else: - if self.direction is None or "recv" in self.direction: - direction_attr = "sendrecv" + if self.direction is None or "recv" in self.direction.decode(): + direction_attr = b"sendrecv" else: - direction_attr = "sendonly" - local_media.attributes.append(SDPAttribute(direction_attr, "")) + direction_attr = b"sendonly" + local_media.attributes.append(SDPAttribute(direction_attr, b"")) for attribute in local_media.attributes: - if attribute.name == 'rtcp': - attribute.value = attribute.value.split(' ', 1)[0] + if attribute.name == b'rtcp': + attribute.value = (attribute.value.decode().split(' ', 1)[0]).encode() self._sdp_info.local_media = local_media return local_media finally: with nogil: pj_mutex_unlock(lock) def start(self, BaseSDPSession local_sdp, BaseSDPSession remote_sdp, int sdp_index, int timeout=30): cdef int status cdef object desired_state cdef pj_mutex_t *lock = self._lock cdef pj_pool_t *pool cdef pjmedia_endpt *media_endpoint cdef pjmedia_port *media_port cdef pjmedia_sdp_media *local_media cdef pjmedia_sdp_session *pj_local_sdp cdef pjmedia_sdp_session *pj_remote_sdp cdef pjmedia_stream **stream_address cdef pjmedia_stream_info *stream_info_address cdef pjmedia_transport *transport cdef PJSIPUA ua ua = _get_ua() with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: pool = self._pool media_endpoint = ua._pjmedia_endpoint._obj stream_address = &self._obj stream_info_address = &self._stream_info transport = self.transport._obj if self._is_started: raise SIPCoreError("This AudioTransport was already started once") desired_state = ("LOCAL" if self._is_offer else "REMOTE") if self.transport.state != desired_state: raise SIPCoreError('RTPTransport object provided is not in the "%s" state, but in the "%s" state' % (desired_state, self.transport.state)) if None in [local_sdp, remote_sdp]: raise ValueError("SDP arguments cannot be None") pj_local_sdp = local_sdp.get_sdp_session() pj_remote_sdp = remote_sdp.get_sdp_session() if sdp_index < 0: raise ValueError("sdp_index argument cannot be negative") if local_sdp.media[sdp_index].port == 0 or remote_sdp.media[sdp_index].port == 0: raise SIPCoreError("Cannot start a rejected audio stream") if timeout < 0: raise ValueError("timeout value cannot be negative") self.transport.set_ESTABLISHED(local_sdp, remote_sdp, sdp_index) with nogil: status = pjmedia_stream_info_from_sdp(stream_info_address, pool, media_endpoint, pj_local_sdp, pj_remote_sdp, sdp_index) if status != 0: raise PJSIPError("Could not parse SDP for audio session", status) if self._stream_info.param == NULL: raise SIPCoreError("Could not parse SDP for audio session") self._stream_info.param.setting.vad = self._vad self._stream_info.use_ka = 1 with nogil: status = pjmedia_stream_create(media_endpoint, pool, stream_info_address, transport, NULL, stream_address) if status != 0: raise PJSIPError("Could not initialize RTP for audio session", status) with nogil: status = pjmedia_stream_set_dtmf_callback(stream_address[0], _AudioTransport_cb_dtmf, self.weakref) if status != 0: with nogil: pjmedia_stream_destroy(stream_address[0]) self._obj = NULL raise PJSIPError("Could not set DTMF callback for audio session", status) with nogil: status = pjmedia_stream_start(stream_address[0]) if status != 0: with nogil: pjmedia_stream_destroy(stream_address[0]) self._obj = NULL raise PJSIPError("Could not start RTP for audio session", status) with nogil: status = pjmedia_stream_get_port(stream_address[0], &media_port) if status != 0: with nogil: pjmedia_stream_destroy(stream_address[0]) self._obj = NULL raise PJSIPError("Could not get audio port for audio session", status) try: self._slot = self.mixer._add_port(ua, pool, media_port) if self._volume != 100: self.volume = self._volume except: with nogil: pjmedia_stream_destroy(stream_address[0]) self._obj = NULL raise self.update_direction(local_sdp.media[sdp_index].direction) self._sdp_info.local_media = local_sdp.media[sdp_index] self._sdp_info.local_sdp = local_sdp self._sdp_info.remote_sdp = remote_sdp self._sdp_info.index = sdp_index self._is_started = 1 if timeout > 0: self._timer = MediaCheckTimer(timeout) self._timer.schedule(timeout, self._cb_check_rtp, self) self.mixer.reset_ec() finally: with nogil: pj_mutex_unlock(lock) def stop(self): cdef int status cdef pj_mutex_t *lock = self._lock cdef pjmedia_stream *stream cdef PJSIPUA ua ua = self._check_ua() if ua is not None: with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: stream = self._obj if self._timer is not None: self._timer.cancel() self._timer = None if self._obj == NULL: return self._obj = NULL self.mixer._remove_port(ua, self._slot) with nogil: pjmedia_stream_destroy(stream) self.transport.set_INIT() finally: if ua is not None: with nogil: pj_mutex_unlock(lock) def update_direction(self, direction): cdef int status cdef pj_mutex_t *lock = self._lock _get_ua() with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: if self._obj == NULL: raise SIPCoreError("Stream is not active") if direction not in valid_sdp_directions: raise SIPCoreError("Unknown direction: %s" % direction) if direction != self.direction: self.mixer.reset_ec() self.direction = direction finally: with nogil: pj_mutex_unlock(lock) def update_sdp(self, local_sdp, remote_sdp, index): cdef int status cdef pj_mutex_t *lock = self._lock _get_ua() with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: if self._obj == NULL: raise SIPCoreError("Stream is not active") self._sdp_info.local_media = local_sdp.media[index] self._sdp_info.local_sdp = local_sdp self._sdp_info.remote_sdp = remote_sdp self._sdp_info.index = index finally: with nogil: pj_mutex_unlock(lock) def send_dtmf(self, digit): cdef int status cdef pj_mutex_t *lock = self._lock cdef pj_str_t digit_pj cdef pjmedia_stream *stream cdef PJSIPUA ua ua = _get_ua() with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: stream = self._obj if self._obj == NULL: raise SIPCoreError("Stream is not active") if len(digit) != 1 or digit not in "0123456789*#ABCD": raise SIPCoreError("Not a valid DTMF digit: %s" % digit) _str_to_pj_str(digit, &digit_pj) if not self._stream_info.tx_event_pt < 0: # If the remote doesn't support telephone-event just don't send DTMF with nogil: status = pjmedia_stream_dial_dtmf(stream, &digit_pj) if status != 0: raise PJSIPError("Could not send DTMF digit on audio stream", status) finally: with nogil: pj_mutex_unlock(lock) cdef int _cb_check_rtp(self, MediaCheckTimer timer) except -1: cdef int status cdef pj_mutex_t *lock = self._lock cdef pjmedia_rtcp_stat stat cdef pjmedia_stream *stream with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: stream = self._obj if stream == NULL: return 0 if self._timer is None: return 0 self._timer = None with nogil: status = pjmedia_stream_get_stat(stream, &stat) if status == 0: if self._packets_received == stat.rx.pkt and self.direction == "sendrecv": _add_event("RTPAudioTransportDidTimeout", dict(obj=self)) self._packets_received = stat.rx.pkt if timer.media_check_interval > 0: self._timer = MediaCheckTimer(timer.media_check_interval) self._timer.schedule(timer.media_check_interval, self._cb_check_rtp, self) finally: with nogil: pj_mutex_unlock(lock) cdef class VideoTransport: def __cinit__(self, *args, **kwargs): cdef int status cdef pj_pool_t *pool cdef bytes pool_name cdef PJSIPUA ua ua = _get_ua() endpoint = ua._pjsip_endpoint._obj pool_name = b"VideoTransport_%d" % id(self) self.weakref = weakref.ref(self) Py_INCREF(self.weakref) pool = ua.create_memory_pool(pool_name, 4096, 4096) self._pool = pool status = pj_mutex_create_recursive(pool, "video_transport_lock", &self._lock) if status != 0: raise PJSIPError("failed to create lock", status) self._timer = None def __init__(self, RTPTransport transport, BaseSDPSession remote_sdp=None, int sdp_index=0, list codecs=None): cdef int status cdef pj_pool_t *pool cdef pjmedia_endpt *media_endpoint cdef pjmedia_sdp_media *local_media_c cdef pjmedia_sdp_session *local_sdp_c cdef pjmedia_transport_info info cdef pj_sockaddr *addr cdef list global_codecs cdef SDPMediaStream local_media cdef SDPSession local_sdp cdef PJSIPUA ua ua = _get_ua() media_endpoint = ua._pjmedia_endpoint._obj pool = self._pool if self.transport is not None: raise SIPCoreError("VideoTransport.__init__() was already called") if transport is None: raise ValueError("transport argument cannot be None") if sdp_index < 0: raise ValueError("sdp_index argument cannot be negative") if transport.state != "INIT": raise SIPCoreError('RTPTransport object provided is not in the "INIT" state, but in the "%s" state' % transport.state) self.transport = transport transport._get_info(&info) global_codecs = ua._pjmedia_endpoint._get_current_video_codecs() if codecs is None: codecs = global_codecs try: ua._pjmedia_endpoint._set_video_codecs(codecs) addr = &(info.sock_info.rtp_addr_name) with nogil: status = pjmedia_endpt_create_base_sdp(media_endpoint, pool, NULL, addr, &local_sdp_c) if status != 0: raise PJSIPError("Could not generate base SDP", status) with nogil: status = pjmedia_endpt_create_video_sdp(media_endpoint, pool, &info.sock_info, 0, &local_media_c) if status != 0: raise PJSIPError("Could not generate SDP video stream", status) # Create a 'fake' SDP, which only contains the video stream, then the m line is extracted because the full # SDP is built by the Session local_sdp_c.media_count = 1 local_sdp_c.media[0] = local_media_c finally: ua._pjmedia_endpoint._set_video_codecs(global_codecs) local_sdp = SDPSession_create(local_sdp_c) local_media = local_sdp.media[0] if remote_sdp is None: self._is_offer = 1 self.transport.set_LOCAL(local_sdp, 0) else: self._is_offer = 0 if sdp_index != 0: local_sdp.media = [None] * (sdp_index+1) local_sdp.media[sdp_index] = local_media self.transport.set_REMOTE(local_sdp, remote_sdp, sdp_index) self._sdp_info = SDPInfo(local_media, local_sdp, remote_sdp, sdp_index) self.local_video = None self.remote_video = None def __dealloc__(self): cdef PJSIPUA ua cdef Timer timer try: ua = _get_ua() except SIPCoreError: return if self._obj != NULL: self.stop() if self._lock != NULL: pj_mutex_destroy(self._lock) ua.release_memory_pool(self._pool) self._pool = NULL timer = Timer() try: timer.schedule(60, deallocate_weakref, self.weakref) except SIPCoreError: pass cdef PJSIPUA _check_ua(self): cdef PJSIPUA ua try: ua = _get_ua() return ua except: self._obj = NULL self._pool = NULL return None property is_active: def __get__(self): self._check_ua() return bool(self._obj != NULL) property is_started: def __get__(self): return bool(self._is_started) property codec: def __get__(self): self._check_ua() if self._obj == NULL: return None else: return _pj_str_to_bytes(self._stream_info.codec_info.encoding_name) property sample_rate: def __get__(self): self._check_ua() if self._obj == NULL: return None else: return self._stream_info.codec_info.clock_rate property statistics: def __get__(self): cdef int status cdef pj_mutex_t *lock = self._lock cdef pjmedia_rtcp_stat stat cdef pjmedia_vid_stream *stream cdef dict statistics = dict() cdef PJSIPUA ua ua = self._check_ua() if ua is None: return None with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: stream = self._obj if stream == NULL: return None with nogil: status = pjmedia_vid_stream_get_stat(stream, &stat) if status != 0: raise PJSIPError("Could not get RTP statistics", status) statistics["rtt"] = _pj_math_stat_to_dict(&stat.rtt) statistics["rx"] = _pjmedia_rtcp_stream_stat_to_dict(&stat.rx) statistics["tx"] = _pjmedia_rtcp_stream_stat_to_dict(&stat.tx) return statistics finally: with nogil: pj_mutex_unlock(lock) def get_local_media(self, BaseSDPSession remote_sdp=None, int index=0, direction="sendrecv"): global valid_sdp_directions cdef int status cdef pj_mutex_t *lock = self._lock cdef object direction_attr cdef SDPAttribute attr cdef SDPSession local_sdp cdef SDPMediaStream local_media cdef pjmedia_sdp_media *c_local_media _get_ua() with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: is_offer = remote_sdp is None if is_offer and direction not in valid_sdp_directions: raise SIPCoreError("Unknown direction: %s" % direction) self._sdp_info.index = index local_sdp = self._sdp_info.local_sdp local_media = self._sdp_info.local_media local_sdp.media = [None] * (index+1) local_sdp.media[index] = local_media self.transport.update_local_sdp(local_sdp, remote_sdp, index) # updating the local SDP might have modified the connection line if local_sdp.connection is not None and local_media.connection is None: local_media.connection = SDPConnection.new(local_sdp.connection) local_media.attributes = [ attr for attr in local_media.attributes if attr.name not in valid_sdp_directions] if is_offer: direction_attr = direction else: - if self.direction is None or "recv" in self.direction: - direction_attr = "sendrecv" + if self.direction is None or "recv" in self.direction.decode(): + direction_attr = b"sendrecv" else: - direction_attr = "sendonly" - local_media.attributes.append(SDPAttribute(direction_attr, "")) + direction_attr = b"sendonly" + local_media.attributes.append(SDPAttribute(direction_attr, b"")) for attribute in local_media.attributes: if attribute.name == 'rtcp': - attribute.value = attribute.value.split(' ', 1)[0] + attribute.value = (attribute.value.decode().split(' ', 1)[0]).encode() return local_media finally: with nogil: pj_mutex_unlock(lock) def start(self, BaseSDPSession local_sdp, BaseSDPSession remote_sdp, int sdp_index, int timeout=30): cdef int status cdef object desired_state cdef pj_mutex_t *lock = self._lock cdef pj_pool_t *pool cdef pjmedia_endpt *media_endpoint cdef pjmedia_sdp_media *local_media cdef pjmedia_sdp_session *pj_local_sdp cdef pjmedia_sdp_session *pj_remote_sdp cdef pjmedia_vid_stream *stream cdef pjmedia_vid_stream_info *stream_info cdef pjmedia_transport *transport cdef PJSIPUA ua ua = _get_ua() with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: pool = self._pool media_endpoint = ua._pjmedia_endpoint._obj stream_info = &self._stream_info transport = self.transport._obj if self._is_started: raise SIPCoreError("This VideoTransport was already started once") desired_state = ("LOCAL" if self._is_offer else "REMOTE") if self.transport.state != desired_state: raise SIPCoreError('RTPTransport object provided is not in the "%s" state, but in the "%s" state' % (desired_state, self.transport.state)) if None in (local_sdp, remote_sdp): raise ValueError("SDP arguments cannot be None") pj_local_sdp = local_sdp.get_sdp_session() pj_remote_sdp = remote_sdp.get_sdp_session() if sdp_index < 0: raise ValueError("sdp_index argument cannot be negative") if local_sdp.media[sdp_index].port == 0 or remote_sdp.media[sdp_index].port == 0: raise SIPCoreError("Cannot start a rejected video stream") if timeout < 0: raise ValueError("timeout value cannot be negative") self.transport.set_ESTABLISHED(local_sdp, remote_sdp, sdp_index) with nogil: status = pjmedia_vid_stream_info_from_sdp(stream_info, pool, media_endpoint, pj_local_sdp, pj_remote_sdp, sdp_index) if status != 0: raise PJSIPError("Could not parse SDP for video session", status) if self._stream_info.codec_param == NULL: raise SIPCoreError("Could not parse SDP for video session") self._stream_info.use_ka = 1 with nogil: status = pjmedia_vid_stream_create(media_endpoint, pool, stream_info, transport, NULL, &stream) if status != 0: raise PJSIPError("Could not initialize RTP for video session", status) self._obj = stream with nogil: status = pjmedia_vid_stream_start(stream) if status != 0: with nogil: pjmedia_vid_stream_destroy(stream) self._obj = NULL raise PJSIPError("Could not start RTP for video session", status) with nogil: pjmedia_vid_stream_send_rtcp_sdes(stream) try: local_video = LocalVideoStream_create(stream) remote_video = RemoteVideoStream_create(stream, self._remote_video_event_handler) except PJSIPError: with nogil: pjmedia_vid_stream_destroy(stream) self._obj = NULL self.local_video = None self.remote_video = None raise self.local_video = local_video self.remote_video = remote_video self.update_direction(local_sdp.media[sdp_index].direction) self._sdp_info.local_media = local_sdp.media[sdp_index] self._sdp_info.local_sdp = local_sdp self._sdp_info.remote_sdp = remote_sdp self._sdp_info.index = sdp_index self._is_started = 1 if timeout > 0: self._timer = MediaCheckTimer(timeout) self._timer.schedule(timeout, self._cb_check_rtp, self) finally: with nogil: pj_mutex_unlock(lock) def stop(self): cdef int status cdef pj_mutex_t *lock = self._lock cdef pjmedia_vid_stream *stream cdef PJSIPUA ua ua = self._check_ua() if ua is not None: with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: stream = self._obj if self._timer is not None: self._timer.cancel() self._timer = None if self._obj == NULL: return self._obj = NULL if self.local_video is not None: self.local_video.close() self.local_video = None if self.remote_video is not None: self.remote_video.close() self.remote_video = None with nogil: pjmedia_vid_stream_send_rtcp_bye(stream) pjmedia_vid_stream_destroy(stream) self.transport.set_INIT() finally: if ua is not None: with nogil: pj_mutex_unlock(lock) def update_direction(self, direction): global valid_sdp_directions cdef int status cdef pj_mutex_t *lock = self._lock cdef pjmedia_vid_stream *stream _get_ua() with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: stream = self._obj if self._obj == NULL: raise SIPCoreError("Stream is not active") if direction not in valid_sdp_directions: raise SIPCoreError("Unknown direction: %s" % direction) self.direction = direction finally: with nogil: pj_mutex_unlock(lock) def update_sdp(self, local_sdp, remote_sdp, index): cdef int status cdef pj_mutex_t *lock = self._lock _get_ua() with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: if self._obj == NULL: raise SIPCoreError("Stream is not active") self._sdp_info.local_media = local_sdp.media[index] self._sdp_info.local_sdp = local_sdp self._sdp_info.remote_sdp = remote_sdp self._sdp_info.index = index finally: with nogil: pj_mutex_unlock(lock) def pause(self, direction="both"): cdef int status cdef pj_mutex_t *lock = self._lock cdef pjmedia_vid_stream *stream cdef pjmedia_dir pj_dir _get_ua() if direction not in ("incoming", "outgoing", "both"): raise ValueError("direction can only be one of 'incoming', 'outgoing' or 'both'") if direction == "incoming": pj_dir = PJMEDIA_DIR_RENDER elif direction == "outgoing": pj_dir = PJMEDIA_DIR_CAPTURE else: pj_dir = PJMEDIA_DIR_CAPTURE_RENDER with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: stream = self._obj if self._obj == NULL: raise SIPCoreError("Stream is not active") with nogil: status = pjmedia_vid_stream_pause(stream, pj_dir) if status != 0: raise PJSIPError("failed to pause video stream", status) finally: with nogil: pj_mutex_unlock(lock) def resume(self, direction="both"): cdef int status cdef pj_mutex_t *lock = self._lock cdef pjmedia_vid_stream *stream cdef pjmedia_dir pj_dir _get_ua() if direction not in ("incoming", "outgoing", "both"): raise ValueError("direction can only be one of 'incoming', 'outgoing' or 'both'") if direction == "incoming": pj_dir = PJMEDIA_DIR_RENDER elif direction == "outgoing": pj_dir = PJMEDIA_DIR_CAPTURE else: pj_dir = PJMEDIA_DIR_CAPTURE_RENDER with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: stream = self._obj if self._obj == NULL: raise SIPCoreError("Stream is not active") with nogil: status = pjmedia_vid_stream_resume(stream, pj_dir) if status != 0: raise PJSIPError("failed to resume video stream", status) finally: with nogil: pj_mutex_unlock(lock) def send_keyframe(self): cdef pj_mutex_t *lock = self._lock cdef pjmedia_vid_stream *stream _get_ua() with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: stream = self._obj if stream != NULL: # Do not check for errors, it's OK if we can't send it pjmedia_vid_stream_send_keyframe(stream) finally: with nogil: pj_mutex_unlock(lock) def request_keyframe(self): cdef pj_mutex_t *lock = self._lock cdef pjmedia_vid_stream *stream _get_ua() with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: stream = self._obj if stream != NULL: # Do not check for errors, it's OK if we can't send it pjmedia_vid_stream_send_rtcp_pli(stream) finally: with nogil: pj_mutex_unlock(lock) cdef int _cb_check_rtp(self, MediaCheckTimer timer) except -1: cdef int status cdef pj_mutex_t *lock = self._lock cdef pjmedia_rtcp_stat stat cdef pjmedia_vid_stream *stream with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: stream = self._obj if stream == NULL: return 0 if self._timer is None: return 0 self._timer = None with nogil: status = pjmedia_vid_stream_get_stat(stream, &stat) if status == 0: if self._packets_received == stat.rx.pkt and self.direction == "sendrecv": _add_event("RTPVideoTransportDidTimeout", dict(obj=self)) self._packets_received = stat.rx.pkt if timer.media_check_interval > 0: self._timer = MediaCheckTimer(timer.media_check_interval) self._timer.schedule(timer.media_check_interval, self._cb_check_rtp, self) finally: with nogil: pj_mutex_unlock(lock) def _remote_video_event_handler(self, str name, object data): if name == "FORMAT_CHANGED": size, framerate = data _add_event("RTPVideoTransportRemoteFormatDidChange", dict(obj=self, size=size, framerate=framerate)) elif name == "RECEIVED_KEYFRAME": _add_event("RTPVideoTransportReceivedKeyFrame", dict(obj=self)) elif name == "MISSED_KEYFRAME": _add_event("RTPVideoTransportMissedKeyFrame", dict(obj=self)) elif name == "REQUESTED_KEYFRAME": _add_event("RTPVideoTransportRequestedKeyFrame", dict(obj=self)) cdef class ICECandidate: def __init__(self, component, cand_type, address, port, priority, rel_addr=''): self.component = component self.type = cand_type self.address = address self.port = port self.priority = priority self.rel_address = rel_addr def __str__(self): return '(%s) %s:%d%s priority=%d type=%s' % (self.component, self.address, self.port, ' rel_addr=%s' % self.rel_address if self.rel_address else '', self.priority, self.type.lower()) cdef ICECandidate ICECandidate_create(pj_ice_sess_cand *cand): cdef char buf[PJ_INET6_ADDRSTRLEN] cdef str address cdef str cand_type cdef int port if cand.type == PJ_ICE_CAND_TYPE_HOST: cand_type = 'HOST' elif cand.type == PJ_ICE_CAND_TYPE_SRFLX: cand_type = 'SRFLX' elif cand.type == PJ_ICE_CAND_TYPE_PRFLX: cand_type = 'PRFLX' elif cand.type == PJ_ICE_CAND_TYPE_RELAYED: cand_type = 'RELAY' else: cand_type = 'UNKNOWN' pj_sockaddr_print(&cand.addr, buf, PJ_INET6_ADDRSTRLEN, 0) address = _buf_to_str(buf) port = pj_sockaddr_get_port(&cand.addr) if pj_sockaddr_has_addr(&cand.rel_addr): pj_sockaddr_print(&cand.rel_addr, buf, PJ_INET6_ADDRSTRLEN, 0) rel_addr = _buf_to_str(buf) else: rel_addr = '' return ICECandidate('RTP' if cand.comp_id==1 else 'RTCP', cand_type, address, port, cand.prio, rel_addr) cdef class ICECheck: def __init__(self, local_candidate, remote_candidate, state, nominated): self.local_candidate = local_candidate self.remote_candidate = remote_candidate self.state = state self.nominated = nominated def __str__(self): return '%s:%d -> %s:%d (%s, %s)' % (self.local_candidate.address, self.local_candidate.port, self.remote_candidate.address, self.remote_candidate.port, self.state.lower(), 'nominated' if self.nominated else 'not nominated') cdef ICECheck ICECheck_create(pj_ice_sess_check *check): cdef str state cdef ICECandidate lcand cdef ICECandidate rcand if check.state == PJ_ICE_SESS_CHECK_STATE_FROZEN: state = 'FROZEN' elif check.state == PJ_ICE_SESS_CHECK_STATE_WAITING: state = 'WAITING' elif check.state == PJ_ICE_SESS_CHECK_STATE_IN_PROGRESS: state = 'IN_PROGRESS' elif check.state == PJ_ICE_SESS_CHECK_STATE_SUCCEEDED: state = 'SUCCEEDED' elif check.state == PJ_ICE_SESS_CHECK_STATE_FAILED: state = 'FAILED' else: state = 'UNKNOWN' lcand = ICECandidate_create(check.lcand) rcand = ICECandidate_create(check.rcand) return ICECheck(lcand, rcand, state, bool(check.nominated)) cdef ICECheck _get_rtp_valid_pair(pj_ice_strans *ice_st): cdef pj_ice_sess_check_ptr_const ice_check ice_check = pj_ice_strans_get_valid_pair(ice_st, 1) if ice_check == NULL: return None return ICECheck_create(ice_check) # helper functions cdef dict _pj_math_stat_to_dict(pj_math_stat *stat): cdef dict retval = dict() retval["count"] = stat.n retval["max"] = stat.max retval["min"] = stat.min retval["last"] = stat.last retval["avg"] = stat.mean return retval cdef dict _pjmedia_rtcp_stream_stat_to_dict(pjmedia_rtcp_stream_stat *stream_stat): cdef dict retval = dict() retval["packets"] = stream_stat.pkt retval["bytes"] = stream_stat.bytes retval["packets_discarded"] = stream_stat.discard retval["packets_lost"] = stream_stat.loss retval["packets_reordered"] = stream_stat.reorder retval["packets_duplicate"] = stream_stat.dup retval["loss_period"] = _pj_math_stat_to_dict(&stream_stat.loss_period) retval["burst_loss"] = bool(stream_stat.loss_type.burst) retval["random_loss"] = bool(stream_stat.loss_type.random) retval["jitter"] = _pj_math_stat_to_dict(&stream_stat.jitter) return retval cdef str _ice_state_to_str(int state): if state == PJ_ICE_STRANS_STATE_NULL: return 'NULL' elif state == PJ_ICE_STRANS_STATE_INIT: return 'GATHERING' elif state == PJ_ICE_STRANS_STATE_READY: return 'GATHERING_COMPLETE' elif state == PJ_ICE_STRANS_STATE_SESS_READY: return 'NEGOTIATION_START' elif state == PJ_ICE_STRANS_STATE_NEGO: return 'NEGOTIATING' elif state == PJ_ICE_STRANS_STATE_RUNNING: return 'RUNNING' elif state == PJ_ICE_STRANS_STATE_FAILED: return 'FAILED' else: return 'UNKNOWN' cdef dict _extract_ice_session_data(pj_ice_sess *ice_sess): cdef dict data = dict() cdef pj_ice_sess_cand *cand cdef pj_ice_sess_check *check # Process local candidates local_candidates = [] for i in range(ice_sess.lcand_cnt): cand = &ice_sess.lcand[i] local_candidates.append(ICECandidate_create(cand)) data['local_candidates'] = local_candidates # Process remote candidates remote_candidates = [] for i in range(ice_sess.rcand_cnt): cand = &ice_sess.rcand[i] remote_candidates.append(ICECandidate_create(cand)) data['remote_candidates'] = remote_candidates # Process valid pairs valid_pairs = [] for i in range(ice_sess.comp_cnt): check = ice_sess.comp[i].valid_check valid_pairs.append(ICECheck_create(check)) data['valid_pairs'] = valid_pairs # Process valid list valid_list = [] for i in range(ice_sess.valid_list.count): check = &ice_sess.valid_list.checks[i] valid_list.append(ICECheck_create(check)) data['valid_list'] = valid_list return data cdef object _extract_rtp_transport(pjmedia_transport *tp): cdef void *rtp_transport_ptr = NULL if tp != NULL: rtp_transport_ptr = tp.user_data if rtp_transport_ptr == NULL: return None return ( rtp_transport_ptr)() # callback functions cdef void _RTPTransport_cb_ice_complete(pjmedia_transport *tp, pj_ice_strans_op op, int status) with gil: # Despite the name this callback is not only called when ICE negotiation ends, it depends on the # op parameter cdef int lock_status cdef double duration cdef pj_ice_strans *ice_st cdef pj_ice_sess *ice_sess cdef pj_time_val tv, start_time cdef pj_mutex_t *lock cdef RTPTransport rtp_transport cdef PJSIPUA ua try: ua = _get_ua() except: return rtp_transport = _extract_rtp_transport(tp) if rtp_transport is None: return lock = rtp_transport._lock with nogil: lock_status = pj_mutex_lock(lock) if lock_status != 0: raise PJSIPError("failed to acquire lock", status) try: if op == PJ_ICE_STRANS_OP_NEGOTIATION: if status == 0: ice_st = pjmedia_ice_get_strans(tp) if ice_st == NULL: return ice_sess = pj_ice_strans_get_session(ice_st) if ice_sess == NULL: return start_time = pj_ice_strans_get_start_time(ice_st) pj_gettimeofday(&tv) tv.sec -= start_time.sec tv.msec -= start_time.msec pj_time_val_normalize(&tv) duration = (tv.sec*1000 + tv.msec)/1000.0 data = _extract_ice_session_data(ice_sess) rtp_transport._rtp_valid_pair = _get_rtp_valid_pair(ice_st) _add_event("RTPTransportICENegotiationDidSucceed", dict(obj=rtp_transport, duration=duration, local_candidates=data['local_candidates'], remote_candidates=data['remote_candidates'], valid_pairs=data['valid_pairs'], valid_list=data['valid_list'])) else: rtp_transport._rtp_valid_pair = None _add_event("RTPTransportICENegotiationDidFail", dict(obj=rtp_transport, reason=_pj_status_to_str(status))) elif op == PJ_ICE_STRANS_OP_INIT: if status == 0: rtp_transport.state = "INIT" _add_event("RTPTransportDidInitialize", dict(obj=rtp_transport)) else: rtp_transport.state = "INVALID" _add_event("RTPTransportDidFail", dict(obj=rtp_transport, reason=_pj_status_to_str(status))) else: # silence compiler warning pass finally: with nogil: pj_mutex_unlock(lock) cdef void _RTPTransport_cb_ice_state(pjmedia_transport *tp, pj_ice_strans_state prev, pj_ice_strans_state curr) with gil: cdef int status cdef pj_mutex_t *lock cdef RTPTransport rtp_transport cdef PJSIPUA ua try: ua = _get_ua() except: return rtp_transport = _extract_rtp_transport(tp) if rtp_transport is None: return lock = rtp_transport._lock with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: _add_event("RTPTransportICENegotiationStateDidChange", dict(obj=rtp_transport, prev_state=_ice_state_to_str(prev), state=_ice_state_to_str(curr))) finally: with nogil: pj_mutex_unlock(lock) cdef void _RTPTransport_cb_ice_stop(pjmedia_transport *tp, char *reason, int err) with gil: cdef int status cdef pj_mutex_t *lock cdef RTPTransport rtp_transport cdef PJSIPUA ua try: ua = _get_ua() except: return rtp_transport = _extract_rtp_transport(tp) if rtp_transport is None: return lock = rtp_transport._lock with nogil: status = pj_mutex_lock(lock) if status != 0: raise PJSIPError("failed to acquire lock", status) try: rtp_transport._rtp_valid_pair = None _reason = reason if _reason != b"media stop requested": _add_event("RTPTransportICENegotiationDidFail", dict(obj=rtp_transport, reason=_reason)) finally: with nogil: pj_mutex_unlock(lock) cdef void _RTPTransport_cb_zrtp_secure_on(pjmedia_transport *tp, char* cipher) with gil: cdef RTPTransport rtp_transport cdef PJSIPUA ua try: ua = _get_ua() except: return try: rtp_transport = _extract_rtp_transport(tp) if rtp_transport is None: return _add_event("RTPTransportZRTPSecureOn", dict(obj=rtp_transport, cipher=bytes(cipher))) except: ua._handle_exception(1) cdef void _RTPTransport_cb_zrtp_secure_off(pjmedia_transport *tp) with gil: cdef RTPTransport rtp_transport cdef PJSIPUA ua try: ua = _get_ua() except: return try: rtp_transport = _extract_rtp_transport(tp) if rtp_transport is None: return _add_event("RTPTransportZRTPSecureOff", dict(obj=rtp_transport)) except: ua._handle_exception(1) cdef void _RTPTransport_cb_zrtp_show_sas(pjmedia_transport *tp, char* sas, int verified) with gil: cdef RTPTransport rtp_transport cdef PJSIPUA ua try: ua = _get_ua() except: return try: rtp_transport = _extract_rtp_transport(tp) if rtp_transport is None: return _add_event("RTPTransportZRTPReceivedSAS", dict(obj=rtp_transport, sas=str(sas), verified=bool(verified))) except: ua._handle_exception(1) cdef void _RTPTransport_cb_zrtp_confirm_goclear(pjmedia_transport *tp) with gil: cdef RTPTransport rtp_transport cdef PJSIPUA ua try: ua = _get_ua() except: return try: rtp_transport = _extract_rtp_transport(tp) if rtp_transport is None: return # TODO: not yet implemented by PJSIP's ZRTP transport except: ua._handle_exception(1) cdef void _RTPTransport_cb_zrtp_show_message(pjmedia_transport *tp, int severity, int sub_code) with gil: global zrtp_message_levels, zrtp_error_messages cdef RTPTransport rtp_transport cdef PJSIPUA ua try: ua = _get_ua() except: return try: rtp_transport = _extract_rtp_transport(tp) if rtp_transport is None: return level = zrtp_message_levels.get(severity, 1) message = zrtp_error_messages[level].get(sub_code, 'Unknown') _add_event("RTPTransportZRTPLog", dict(obj=rtp_transport, level=level, message=message)) except: ua._handle_exception(1) cdef void _RTPTransport_cb_zrtp_negotiation_failed(pjmedia_transport *tp, int severity, int sub_code) with gil: global zrtp_message_levels, zrtp_error_messages cdef RTPTransport rtp_transport cdef PJSIPUA ua try: ua = _get_ua() except: return try: rtp_transport = _extract_rtp_transport(tp) if rtp_transport is None: return level = zrtp_message_levels.get(severity, 1) reason = zrtp_error_messages[level].get(sub_code, 'Unknown') _add_event("RTPTransportZRTPNegotiationFailed", dict(obj=rtp_transport, reason=reason)) except: ua._handle_exception(1) cdef void _RTPTransport_cb_zrtp_not_supported_by_other(pjmedia_transport *tp) with gil: cdef RTPTransport rtp_transport cdef PJSIPUA ua try: ua = _get_ua() except: return try: rtp_transport = _extract_rtp_transport(tp) if rtp_transport is None: return _add_event("RTPTransportZRTPNotSupportedByRemote", dict(obj=rtp_transport)) except: ua._handle_exception(1) cdef void _RTPTransport_cb_zrtp_ask_enrollment(pjmedia_transport *tp, int info) with gil: cdef RTPTransport rtp_transport cdef PJSIPUA ua try: ua = _get_ua() except: return try: rtp_transport = _extract_rtp_transport(tp) if rtp_transport is None: return # TODO: implement PBX enrollment except: ua._handle_exception(1) cdef void _RTPTransport_cb_zrtp_inform_enrollment(pjmedia_transport *tp, int info) with gil: cdef RTPTransport rtp_transport cdef PJSIPUA ua try: ua = _get_ua() except: return try: rtp_transport = _extract_rtp_transport(tp) if rtp_transport is None: return # TODO: implement PBX enrollment except: ua._handle_exception(1) cdef void _AudioTransport_cb_dtmf(pjmedia_stream *stream, void *user_data, int digit) with gil: cdef AudioTransport audio_stream = ( user_data)() cdef PJSIPUA ua try: ua = _get_ua() except: return if audio_stream is None: return try: _add_event("RTPAudioStreamGotDTMF", dict(obj=audio_stream, digit=chr(digit))) except: ua._handle_exception(1) # globals cdef pjmedia_ice_cb _ice_cb _ice_cb.on_ice_complete = _RTPTransport_cb_ice_complete _ice_cb.on_ice_state = _RTPTransport_cb_ice_state _ice_cb.on_ice_stop = _RTPTransport_cb_ice_stop -valid_sdp_directions = ("sendrecv", "sendonly", "recvonly", "inactive") +valid_sdp_directions = (b"sendrecv", b"sendonly", b"recvonly", b"inactive") # ZRTP cdef pjmedia_zrtp_cb _zrtp_cb _zrtp_cb.secure_on = _RTPTransport_cb_zrtp_secure_on _zrtp_cb.secure_off = _RTPTransport_cb_zrtp_secure_off _zrtp_cb.show_sas = _RTPTransport_cb_zrtp_show_sas _zrtp_cb.confirm_go_clear = _RTPTransport_cb_zrtp_confirm_goclear _zrtp_cb.show_message = _RTPTransport_cb_zrtp_show_message _zrtp_cb.negotiation_failed = _RTPTransport_cb_zrtp_negotiation_failed _zrtp_cb.not_supported_by_other = _RTPTransport_cb_zrtp_not_supported_by_other _zrtp_cb.ask_enrollment = _RTPTransport_cb_zrtp_ask_enrollment _zrtp_cb.inform_enrollment = _RTPTransport_cb_zrtp_inform_enrollment _zrtp_cb.sign_sas = NULL _zrtp_cb.check_sas_signature = NULL # Keep these aligned with ZrtpCodes.h cdef dict zrtp_message_levels = {1: 'INFO', 2: 'WARNING', 3: 'SEVERE', 4: 'ERROR'} cdef dict zrtp_error_messages = { 'INFO': { 0: "Unknown", 1: "Hello received and prepared a Commit, ready to get peer's hello hash", #InfoHelloReceived 2: "Commit: Generated a public DH key", #InfoCommitDHGenerated 3: "Responder: Commit received, preparing DHPart1", #InfoRespCommitReceived 4: "DH1Part: Generated a public DH key", #InfoDH1DHGenerated 5: "Initiator: DHPart1 received, preparing DHPart2", #InfoInitDH1Received 6: "Responder: DHPart2 received, preparing Confirm1", #InfoRespDH2Received 7: "Initiator: Confirm1 received, preparing Confirm2", #InfoInitConf1Received 8: "Responder: Confirm2 received, preparing Conf2Ack", #InfoRespConf2Received 9: "At least one retained secrets matches - security OK", #InfoRSMatchFound 10: "Entered secure state", #InfoSecureStateOn 11: "No more security for this session", #InfoSecureStateOff }, 'WARNING': { 0: "Unknown", 1: "WarningDHAESmismatch = 1, //!< Commit contains an AES256 cipher but does not offer a Diffie-Helman 4096 - not used DH4096 was discarded", #WarningDHAESmismatch 2: "Received a GoClear message", #WarningGoClearReceived 3: "Hello offers an AES256 cipher but does not offer a Diffie-Helman 4096- not used DH4096 was discarded", #WarningDHShort 4: "No retained shared secrets available - must verify SAS", #WarningNoRSMatch 5: "Internal ZRTP packet checksum mismatch - packet dropped", #WarningCRCmismatch 6: "Dropping packet because SRTP authentication failed!", #WarningSRTPauthError 7: "Dropping packet because SRTP replay check failed!", #WarningSRTPreplayError 8: "Valid retained shared secrets availabe but no matches found - must verify SAS", #WarningNoExpectedRSMatch }, 'SEVERE': { 0: "Unknown", 1: "Hash HMAC check of Hello failed!", #SevereHelloHMACFailed 2: "Hash HMAC check of Commit failed!", #SevereCommitHMACFailed 3: "Hash HMAC check of DHPart1 failed!", #SevereDH1HMACFailed 4: "Hash HMAC check of DHPart2 failed!", #SevereDH2HMACFailed 5: "Cannot send data - connection or peer down?", #SevereCannotSend 6: "Internal protocol error occured!", #SevereProtocolError 7: "Cannot start a timer - internal resources exhausted?", #SevereNoTimer 8: "Too much retries during ZRTP negotiation - connection or peer down?", #SevereTooMuchRetries }, 'ERROR': { 0x00: "Unknown", 0x10: "Malformed packet (CRC OK, but wrong structure)", #MalformedPacket 0x20: "Critical software error", #CriticalSWError 0x30: "Unsupported ZRTP version", #UnsuppZRTPVersion 0x40: "Hello components mismatch", #HelloCompMismatch 0x51: "Hash type not supported", #UnsuppHashType 0x52: "Cipher type not supported", #UnsuppCiphertype 0x53: "Public key exchange not supported", #UnsuppPKExchange 0x54: "SRTP auth. tag not supported", #UnsuppSRTPAuthTag 0x55: "SAS scheme not supported", #UnsuppSASScheme 0x56: "No shared secret available, DH mode required", #NoSharedSecret 0x61: "DH Error: bad pvi or pvr ( == 1, 0, or p-1)", #DHErrorWrongPV 0x62: "DH Error: hvi != hashed data", #DHErrorWrongHVI 0x63: "Received relayed SAS from untrusted MiTM", #SASuntrustedMiTM 0x70: "Auth. Error: Bad Confirm pkt HMAC", #ConfirmHMACWrong 0x80: "Nonce reuse", #NonceReused 0x90: "Equal ZIDs in Hello", #EqualZIDHello 0x100: "GoClear packet received, but not allowed", #GoCleatNotAllowed 0x7fffffff: "Packet ignored", #IgnorePacket } } diff --git a/sipsimple/core/_core.sdp.pxi b/sipsimple/core/_core.sdp.pxi index 1f9316fb..f0fe5069 100644 --- a/sipsimple/core/_core.sdp.pxi +++ b/sipsimple/core/_core.sdp.pxi @@ -1,1228 +1,1227 @@ import re from application.python.descriptor import WriteOnceAttribute cdef object BaseSDPSession_richcmp(object self, object other, int op) with gil: cdef int eq = 1 if op not in [2,3]: return NotImplemented if not isinstance(other, BaseSDPSession): return NotImplemented for attr in ("id", "version", "user", "net_type", "address_type", "address", "address", "name", "connection", "start_time", "stop_time", "attributes", "bandwidth_info", "media"): if getattr(self, attr) != getattr(other, attr): eq = 0 break if op == 2: return bool(eq) else: return not eq cdef pjmedia_sdp_session* _parse_sdp_session(object sdp): cdef int status cdef pjmedia_sdp_session *sdp_session status = pjmedia_sdp_parse(_get_ua()._pjsip_endpoint._pool, _str_as_str(sdp), _str_as_size(sdp), &sdp_session) if status != 0: raise PJSIPError("failed to parse SDP", status) return sdp_session cdef class BaseSDPSession: def __init__(self, *args, **kwargs): raise TypeError("BaseSDPSession cannot be instantiated directly") def __repr__(self): return "%s(%r, %r, %r, %r, %r, %r, %r, %r, %r, %r, %r, %r, %r)" % (self.__class__.__name__, self.address, self.id, self.version, self.user, self.net_type, self.address_type, self.name, self.connection, self.start_time, self.stop_time, self.attributes, self.bandwidth_info, self.media) def __str__(self): cdef char cbuf[2048] cdef int buf_len buf_len = pjmedia_sdp_print(self.get_sdp_session(), cbuf, sizeof(cbuf)) if buf_len > -1: return _pj_buf_len_to_str(cbuf, buf_len) return '' def __richcmp__(self, other, op): return BaseSDPSession_richcmp(self, other, op) cdef pjmedia_sdp_session* get_sdp_session(self): self._sdp_session.media_count = len(self.media) for index, m in enumerate(self.media): if m is not None: self._sdp_session.media[index] = (m).get_sdp_media() else: self._sdp_session.media[index] = NULL self._sdp_session.attr_count = len(self.attributes) for index, attr in enumerate(self.attributes): self._sdp_session.attr[index] = (attr).get_sdp_attribute() self._sdp_session.bandw_count = len(self.bandwidth_info) for index, info in enumerate(self.bandwidth_info): self._sdp_session.bandw[index] = (info).get_sdp_bandwidth_info() return &self._sdp_session property has_ice_attributes: def __get__(self): return set([attr.name for attr in self.attributes]).issuperset(['ice-pwd', 'ice-ufrag']) cdef class SDPSession(BaseSDPSession): - def __init__(self, object address not None, object id=None, object version=None, object user not None="-", object net_type not None=b"IN", object address_type not None=b"IP4", + def __init__(self, object address not None, object id=None, object version=None, object user not None=b"-", object net_type not None=b"IN", object address_type not None=b"IP4", object name not None=b" ", SDPConnection connection=None, unsigned long start_time=0, unsigned long stop_time=0, list attributes=None, list bandwidth_info=None, list media=None): cdef unsigned int version_id = 2208988800UL cdef pj_time_val tv pj_gettimeofday(&tv) version_id += tv.sec self.address = address self.id = id if id is not None else version_id self.version = version if version is not None else version_id self.user = user self.net_type = net_type self.address_type = address_type self.name = name self.connection = connection self.start_time = start_time self.stop_time = stop_time self.attributes = attributes if attributes is not None else [] self.bandwidth_info = bandwidth_info if bandwidth_info is not None else [] self.media = media if media is not None else [] @classmethod def new(cls, BaseSDPSession sdp_session): connection = SDPConnection.new(sdp_session.connection) if (sdp_session.connection is not None) else None attributes = [SDPAttribute.new(attr) for attr in sdp_session.attributes] bandwidth_info = [SDPBandwidthInfo.new(info) for info in sdp_session.bandwidth_info] media = [SDPMediaStream.new(m) if m is not None else None for m in sdp_session.media] return cls(sdp_session.address, sdp_session.id, sdp_session.version, sdp_session.user, sdp_session.net_type, sdp_session.address_type, sdp_session.name, connection, sdp_session.start_time, sdp_session.stop_time, attributes, bandwidth_info, media) @classmethod def parse(cls, object sdp): cdef pjmedia_sdp_session *sdp_session sdp_session = _parse_sdp_session(sdp) return SDPSession_create(sdp_session) property address: def __get__(self): return self._address def __set__(self, str address not None): _str_to_pj_str(address, &self._sdp_session.origin.addr) self._address = address property id: def __get__(self): return self._sdp_session.origin.id def __set__(self, unsigned int id): self._sdp_session.origin.id = id property version: def __get__(self): return self._sdp_session.origin.version def __set__(self, unsigned int version): self._sdp_session.origin.version = version property user: def __get__(self): return self._user def __set__(self, str user not None): _str_to_pj_str(user, &self._sdp_session.origin.user) self._user = user property net_type: def __get__(self): return self._net_type def __set__(self, str net_type not None): _str_to_pj_str(net_type, &self._sdp_session.origin.net_type) self._net_type = net_type property address_type: def __get__(self): return self._address_type def __set__(self, str address_type not None): _str_to_pj_str(address_type, &self._sdp_session.origin.addr_type) self._address_type = address_type property name: def __get__(self): return self._name def __set__(self, str name not None): _str_to_pj_str(name, &self._sdp_session.name) self._name = name property connection: def __get__(self): return self._connection def __set__(self, SDPConnection connection): if connection is None: self._sdp_session.conn = NULL else: self._sdp_session.conn = connection.get_sdp_connection() self._connection = connection property start_time: def __get__(self): return self._sdp_session.time.start def __set__(self, unsigned long start_time): self._sdp_session.time.start = start_time property stop_time: def __get__(self): return self._sdp_session.time.stop def __set__(self, unsigned long stop_time): self._sdp_session.time.stop = stop_time property attributes: def __get__(self): return self._attributes def __set__(self, list attributes not None): if len(attributes) > PJMEDIA_MAX_SDP_ATTR: raise SIPCoreError("Too many attributes") for attr in attributes: if not isinstance(attr, SDPAttribute): raise TypeError("Items in SDPSession attribute list must be SDPAttribute instancess") if not isinstance(attributes, SDPAttributeList): attributes = SDPAttributeList(attributes) self._attributes = attributes property bandwidth_info: def __get__(self): return self._bandwidth_info def __set__(self, list infos not None): if len(infos) > PJMEDIA_MAX_SDP_BANDW: raise SIPCoreError("Too many bandwidth info attributes") for info in infos: if not isinstance(info, SDPBandwidthInfo): raise TypeError("Items in SDPSession attribute list must be SDPBandwidthInfo instancess") if not isinstance(infos, SDPBandwidthInfoList): infos = SDPBandwidthInfoList(infos) self._bandwidth_info = infos property media: def __get__(self): return self._media def __set__(self, list media not None): if len(media) > PJMEDIA_MAX_SDP_MEDIA: raise SIPCoreError("Too many media objects") for m in media: if m is not None and not isinstance(m, SDPMediaStream): raise TypeError("Items in SDPSession media list must be SDPMediaStream instancess") self._media = media cdef int _update(self) except -1: cdef SDPSession session cdef SDPMediaStream media, old_media session = SDPSession_create(&(self)._sdp_session) if len(self._media) != len(session._media): raise ValueError("Number of media streams in SDPSession got changed") if len(self._attributes) > len(session._attributes): raise ValueError("Number of attributes in SDPSession got reduced") for attr in ("id", "version", "user", "net_type", "address_type", "address", "name", "start_time", "stop_time"): setattr(self, attr, getattr(session, attr)) if session._connection is None: self.connection = None elif self._connection is None or self._connection != session._connection: self.connection = session._connection for index, attribute in enumerate(session._attributes): try: old_attribute = self._attributes[index] except IndexError: self._attributes.append(attribute) else: if old_attribute != attribute: self._attributes[index] = attribute for index, info in enumerate(session._bandwidth_info): try: old_info = self._bandwidth_info[index] except IndexError: self._bandwidth_info.append(info) else: if old_info != info: self._bandwidth_info[index] = info for index, media in enumerate(session._media): old_media = self._media[index] if old_media is not None: old_media._update(media) cdef class FrozenSDPSession(BaseSDPSession): - def __init__(self, object address not None, object id=None, object version=None, object user not None="-", object net_type not None=b"IN", object address_type not None=b"IP4", object name not None=b" ", + def __init__(self, object address not None, object id=None, object version=None, object user not None=b"-", object net_type not None=b"IN", object address_type not None=b"IP4", object name not None=b" ", FrozenSDPConnection connection=None, unsigned long start_time=0, unsigned long stop_time=0, frozenlist attributes not None=frozenlist(), frozenlist bandwidth_info not None=frozenlist(), frozenlist media not None=frozenlist()): cdef unsigned int version_id = 2208988800UL cdef pj_time_val tv if not self.initialized: if len(attributes) > PJMEDIA_MAX_SDP_ATTR: raise SIPCoreError("Too many attributes") for attr in attributes: if not isinstance(attr, FrozenSDPAttribute): raise TypeError("Items in FrozenSDPSession attribute list must be FrozenSDPAttribute instances") if len(bandwidth_info) > PJMEDIA_MAX_SDP_BANDW: raise SIPCoreError("Too many bandwidth info attributes") for info in bandwidth_info: if not isinstance(info, FrozenSDPBandwidthInfo): raise TypeError("Items in FrozenSDPSession bandwidth info attribute list must be FrozenSDPBandwidthInfo instances") if len(media) > PJMEDIA_MAX_SDP_MEDIA: raise SIPCoreError("Too many media objects") for m in media: if not isinstance(m, FrozenSDPMediaStream): raise TypeError("Items in FrozenSDPSession media list must be FrozenSDPMediaStream instancess") pj_gettimeofday(&tv) version_id += tv.sec self.address = address _str_to_pj_str(address, &self._sdp_session.origin.addr) self.id = id if id is not None else version_id self._sdp_session.origin.id = id if id is not None else version_id self.version = version if version is not None else version_id self._sdp_session.origin.version = version if version is not None else version_id self.user = user _str_to_pj_str(user, &self._sdp_session.origin.user) self.net_type = net_type _str_to_pj_str(net_type, &self._sdp_session.origin.net_type) self.address_type = address_type _str_to_pj_str(address_type, &self._sdp_session.origin.addr_type) self.name = name _str_to_pj_str(name, &self._sdp_session.name) self.connection = connection if connection is None: self._sdp_session.conn = NULL else: self._sdp_session.conn = connection.get_sdp_connection() self.start_time = start_time self._sdp_session.time.start = start_time self.stop_time = stop_time self._sdp_session.time.stop = stop_time self.attributes = FrozenSDPAttributeList(attributes) if not isinstance(attributes, FrozenSDPAttributeList) else attributes self.bandwidth_info = FrozenSDPBandwidthInfoList(bandwidth_info) if not isinstance(bandwidth_info, FrozenSDPBandwidthInfo) else bandwidth_info self.media = media self.initialized = 1 @classmethod def new(cls, BaseSDPSession sdp_session): if isinstance(sdp_session, FrozenSDPSession): return sdp_session connection = FrozenSDPConnection.new(sdp_session.connection) if (sdp_session.connection is not None) else None attributes = frozenlist([FrozenSDPAttribute.new(attr) for attr in sdp_session.attributes]) bandwidth_info = frozenlist([FrozenSDPBandwidthInfo.new(info) for info in sdp_session.bandwidth_info]) media = frozenlist([FrozenSDPMediaStream.new(m) for m in sdp_session.media]) return cls(sdp_session.address, sdp_session.id, sdp_session.version, sdp_session.user, sdp_session.net_type, sdp_session.address_type, sdp_session.name, connection, sdp_session.start_time, sdp_session.stop_time, attributes, bandwidth_info, media) @classmethod def parse(cls, object sdp): cdef pjmedia_sdp_session *sdp_session sdp_session = _parse_sdp_session(sdp) return FrozenSDPSession_create(sdp_session) def __hash__(self): return hash((self.address, self.id, self.version, self.user, self.net_type, self.address_type, self.name, self.connection, self.start_time, self.stop_time, self.attributes, self.bandwidth_info, self.media)) def __richcmp__(self, other, op): return BaseSDPSession_richcmp(self, other, op) class MediaCodec(object): name = WriteOnceAttribute() rate = WriteOnceAttribute() def __init__(self, name, rate): self.name = name self.rate = int(rate) def __repr__(self): return "%s(%r, %r)" % (self.__class__.__name__, self.name, self.rate) def __str__(self): return "%s/%s" % (self.name, self.rate) def __hash__(self): return hash(self.name) def __eq__(self, other): if isinstance(other, MediaCodec): return self.name.lower() == other.name.lower() and self.rate == other.rate elif isinstance(other, basestring): if '/' in other: return self.__str__().lower() == other.lower() else: return self.name.lower() == other.lower() return False def __ne__(self, other): return not self.__eq__(other) cdef object BaseSDPMediaStream_richcmp(object self, object other, int op) with gil: cdef int eq = 1 if op not in [2,3]: return NotImplemented if not isinstance(other, BaseSDPMediaStream): return NotImplemented for attr in ("media", "port", "port_count", "transport", "formats", "connection", "attributes", "bandwidth_info"): if getattr(self, attr) != getattr(other, attr): eq = 0 break if op == 2: return bool(eq) else: return not eq cdef class BaseSDPMediaStream: rtpmap_re = re.compile(r"""^(?P\d+)\s+(?P[-\w]+)/(?P\d+)(?:/\w+)?$""", re.IGNORECASE | re.MULTILINE) rtp_mappings = { 0: MediaCodec('PCMU', 8000), 3: MediaCodec('GSM', 8000), 4: MediaCodec('G723', 8000), 5: MediaCodec('DVI4', 8000), 6: MediaCodec('DVI4', 16000), 7: MediaCodec('LPC', 8000), 8: MediaCodec('PCMA', 8000), 9: MediaCodec('G722', 8000), 10: MediaCodec('L16', 44100), # 2 channels 11: MediaCodec('L16', 44100), # 1 channel 12: MediaCodec('QCELP', 8000), 13: MediaCodec('CN', 8000), 14: MediaCodec('MPA', 90000), 15: MediaCodec('G728', 8000), 16: MediaCodec('DVI4', 11025), 17: MediaCodec('DVI4', 22050), 18: MediaCodec('G729', 8000)} def __init__(self, *args, **kwargs): raise TypeError("BaseSDPMediaStream cannot be instantiated directly") def __repr__(self): return "%s(%r, %r, %r, %r, %r, %r, %r, %r)" % (self.__class__.__name__, self.media, self.port, self.transport, self.port_count, self.formats, self.connection, self.attributes, self.bandwidth_info) def __richcmp__(self, other, op): return BaseSDPMediaStream_richcmp(self, other, op) property direction: def __get__(self): for attribute in self.attributes: - if attribute.name in ("sendrecv", "sendonly", "recvonly", "inactive"): + if attribute.name in (b"sendrecv", b"sendonly", b"recvonly", b"inactive"): return attribute.name - return "sendrecv" + return b"sendrecv" property has_ice_attributes: def __get__(self): return set([attr.name for attr in self.attributes]).issuperset(['ice-pwd', 'ice-ufrag']) property has_ice_candidates: def __get__(self): return 'candidate' in self.attributes cdef pjmedia_sdp_media* get_sdp_media(self): self._sdp_media.attr_count = len(self.attributes) for index, attr in enumerate(self.attributes): self._sdp_media.attr[index] = (attr).get_sdp_attribute() self._sdp_media.bandw_count = len(self.bandwidth_info) for index, info in enumerate(self.bandwidth_info): self._sdp_media.bandw[index] = (info).get_sdp_bandwidth_info() return &self._sdp_media cdef class SDPMediaStream(BaseSDPMediaStream): def __init__(self, object media not None, int port, object transport not None, int port_count=1, list formats=None, SDPConnection connection=None, list attributes=None, list bandwidth_info=None): self.media = media self.port = port self.transport = transport self.port_count = port_count self.formats = formats if formats is not None else [] self.connection = connection self.attributes = attributes if attributes is not None else [] self.bandwidth_info = bandwidth_info if bandwidth_info is not None else [] @classmethod def new(cls, BaseSDPMediaStream sdp_media): connection = SDPConnection.new(sdp_media.connection) if (sdp_media.connection is not None) else None attributes = [SDPAttribute.new(attr) for attr in sdp_media.attributes] bandwidth_info = [SDPBandwidthInfo.new(bi) for bi in sdp_media.bandwidth_info] return cls(sdp_media.media, sdp_media.port, sdp_media.transport, sdp_media.port_count, list(sdp_media.formats), connection, attributes, bandwidth_info) property media: def __get__(self): return self._media def __set__(self, str media not None): _str_to_pj_str(media, &self._sdp_media.desc.media) self._media = media property port: def __get__(self): return self._sdp_media.desc.port def __set__(self, int port): self._sdp_media.desc.port = port property transport: def __get__(self): return self._transport def __set__(self, str transport not None): _str_to_pj_str(transport, &self._sdp_media.desc.transport) self._transport = transport property port_count: def __get__(self): return self._sdp_media.desc.port_count def __set__(self, int port_count): self._sdp_media.desc.port_count = port_count property formats: def __get__(self): return self._formats def __set__(self, list formats not None): if len(formats) > PJMEDIA_MAX_SDP_FMT: raise SIPCoreError("Too many formats") self._sdp_media.desc.fmt_count = len(formats) for index, format in enumerate(formats): _str_to_pj_str(format, &self._sdp_media.desc.fmt[index]) self._formats = formats property codec_list: def __get__(self): return self._codec_list property connection: def __get__(self): return self._connection def __set__(self, SDPConnection connection): if connection is None: self._sdp_media.conn = NULL else: self._sdp_media.conn = connection.get_sdp_connection() self._connection = connection property attributes: def __get__(self): return self._attributes def __set__(self, list attributes not None): if len(attributes) > PJMEDIA_MAX_SDP_ATTR: raise SIPCoreError("Too many attributes") for attr in attributes: if not isinstance(attr, SDPAttribute): raise TypeError("Items in SDPMediaStream attribute list must be SDPAttribute instances") if not isinstance(attributes, SDPAttributeList): attributes = SDPAttributeList(attributes) self._attributes = attributes if self._media in ("audio", "video"): rtp_mappings = self.rtp_mappings.copy() rtpmap_lines = '\n'.join([attr.value for attr in attributes if attr.name=='rtpmap']) # iterators are not supported -Dan rtpmap_codecs = dict([(int(type), MediaCodec(name, rate)) for type, name, rate in self.rtpmap_re.findall(rtpmap_lines)]) rtp_mappings.update(rtpmap_codecs) self._codec_list = [rtp_mappings.get(int(format), MediaCodec('Unknown', 0)) for format in self.formats] else: self._codec_list = list() property bandwidth_info: def __get__(self): return self._bandwidth_info def __set__(self, list infos not None): if len(infos) > PJMEDIA_MAX_SDP_BANDW: raise SIPCoreError("Too many bandwidth information attributes") for info in infos: if not isinstance(info, SDPBandwidthInfo): raise TypeError("Items in SDPMediaStream bandwidth_info list must be SDPBandwidthInfo instances") if not isinstance(infos, SDPBandwidthInfoList): infos = SDPBandwidthInfoList(infos) self._bandwidth_info = infos cdef int _update(self, SDPMediaStream media) except -1: if len(self._attributes) > len(media._attributes): raise ValueError("Number of attributes in SDPMediaStream got reduced") if len(self._bandwidth_info) > len(media._bandwidth_info): raise ValueError("Number of bandwidth info attributes in SDPMediaStream got reduced") for attr in ("media", "port", "transport", "port_count", "formats"): setattr(self, attr, getattr(media, attr)) if media._connection is None: self.connection = None elif self._connection is None or self._connection != media.connection: self.connection = media._connection for index, attribute in enumerate(media._attributes): try: old_attribute = self._attributes[index] except IndexError: self._attributes.append(attribute) else: if old_attribute != attribute: self._attributes[index] = attribute for index, info in enumerate(media._bandwidth_info): try: old_info = self._bandwidth_info[index] except IndexError: self._bandwidth_info.append(info) else: if old_info != info: self._bandwidth_info[index] = info cdef class FrozenSDPMediaStream(BaseSDPMediaStream): def __init__(self, object media not None, int port, object transport not None, int port_count=1, frozenlist formats not None=frozenlist(), FrozenSDPConnection connection=None, frozenlist attributes not None=frozenlist(), frozenlist bandwidth_info not None=frozenlist()): if not self.initialized: if len(formats) > PJMEDIA_MAX_SDP_FMT: raise SIPCoreError("Too many formats") if len(attributes) > PJMEDIA_MAX_SDP_ATTR: raise SIPCoreError("Too many attributes") for attr in attributes: if not isinstance(attr, FrozenSDPAttribute): raise TypeError("Items in FrozenSDPMediaStream attribute list must be FrozenSDPAttribute instances") if len(bandwidth_info) > PJMEDIA_MAX_SDP_BANDW: raise SIPCoreError("Too many bandwidth info attributes") for info in bandwidth_info: if not isinstance(info, FrozenSDPBandwidthInfo): raise TypeError("Items in FrozenSDPMediaStream bandwidth info list must be FrozenSDPBandwidthInfo instances") self.media = media _str_to_pj_str(media, &self._sdp_media.desc.media) self.port = port self._sdp_media.desc.port = port self.transport = transport _str_to_pj_str(transport, &self._sdp_media.desc.transport) self.port_count = port_count self._sdp_media.desc.port_count = port_count self.formats = formats self._sdp_media.desc.fmt_count = len(self.formats) for index, format in enumerate(self.formats): _str_to_pj_str(format, &self._sdp_media.desc.fmt[index]) self.connection = connection if connection is None: self._sdp_media.conn = NULL else: self._sdp_media.conn = connection.get_sdp_connection() self.attributes = FrozenSDPAttributeList(attributes) if not isinstance(attributes, FrozenSDPAttributeList) else attributes if self.media in ("audio", "video"): rtp_mappings = self.rtp_mappings.copy() rtpmap_lines = '\n'.join([attr.value for attr in attributes if attr.name=='rtpmap']) # iterators are not supported -Dan rtpmap_codecs = dict([(int(type), MediaCodec(name, rate)) for type, name, rate in self.rtpmap_re.findall(rtpmap_lines)]) rtp_mappings.update(rtpmap_codecs) self.codec_list = frozenlist([rtp_mappings.get(int(format) if format.isdigit() else None, MediaCodec('Unknown', 0)) for format in self.formats]) else: self.codec_list = frozenlist() self.bandwidth_info = FrozenSDPBandwidthInfoList(bandwidth_info) if not isinstance(bandwidth_info, FrozenSDPBandwidthInfoList) else bandwidth_info self.initialized = 1 @classmethod def new(cls, BaseSDPMediaStream sdp_media): if isinstance(sdp_media, FrozenSDPMediaStream): return sdp_media connection = FrozenSDPConnection.new(sdp_media.connection) if (sdp_media.connection is not None) else None attributes = frozenlist([FrozenSDPAttribute.new(attr) for attr in sdp_media.attributes]) bandwidth_info = frozenlist([FrozenSDPBandwidthInfo.new(info) for info in sdp_media.bandwidth_info]) return cls(sdp_media.media, sdp_media.port, sdp_media.transport, sdp_media.port_count, frozenlist(sdp_media.formats), connection, attributes, bandwidth_info) def __hash__(self): return hash((self.media, self.port, self.transport, self.port_count, self.formats, self.connection, self.attributes, self.bandwidth_info)) def __richcmp__(self, other, op): return BaseSDPMediaStream_richcmp(self, other, op) cdef object BaseSDPConnection_richcmp(object self, object other, int op) with gil: cdef int eq = 1 if op not in [2,3]: return NotImplemented if not isinstance(other, BaseSDPConnection): return NotImplemented for attr in ("net_type", "address_type", "address"): if getattr(self, attr) != getattr(other, attr): eq = 0 break if op == 2: return bool(eq) else: return not eq cdef class BaseSDPConnection: def __init__(self, *args, **kwargs): raise TypeError("BaseSDPConnection cannot be instantiated directly") def __repr__(self): return "%s(%r, %r, %r)" % (self.__class__.__name__, self.address, self.net_type, self.address_type) def __richcmp__(self, other, op): return BaseSDPConnection_richcmp(self, other, op) cdef pjmedia_sdp_conn* get_sdp_connection(self): return &self._sdp_connection cdef class SDPConnection(BaseSDPConnection): def __init__(self, object address not None, object net_type not None=b"IN", object address_type not None=b"IP4"): - print('SDPConnection %s %s %s %s' % (address, type(address), net_type, type())) self.address = address self.net_type = net_type self.address_type = address_type @classmethod def new(cls, BaseSDPConnection sdp_connection): return cls(sdp_connection.address, sdp_connection.net_type, sdp_connection.address_type) property address: def __get__(self): return self._address def __set__(self, str address not None): _str_to_pj_str(address, &self._sdp_connection.addr) self._address = address property net_type: def __get__(self): return self._net_type def __set__(self, str net_type not None): _str_to_pj_str(net_type, &self._sdp_connection.net_type) self._net_type = net_type property address_type: def __get__(self): return self._address_type def __set__(self, str address_type not None): _str_to_pj_str(address_type, &self._sdp_connection.addr_type) self._address_type = address_type cdef class FrozenSDPConnection(BaseSDPConnection): def __init__(self, object address not None, object net_type not None=b"IN", object address_type not None=b"IP4"): if not self.initialized: _str_to_pj_str(address, &self._sdp_connection.addr) _str_to_pj_str(net_type, &self._sdp_connection.net_type) _str_to_pj_str(address_type, &self._sdp_connection.addr_type) self.address = address self.net_type = net_type self.address_type = address_type self.initialized = 1 @classmethod def new(cls, BaseSDPConnection sdp_connection): if isinstance(sdp_connection, FrozenSDPConnection): return sdp_connection return cls(sdp_connection.address, sdp_connection.net_type, sdp_connection.address_type) def __hash__(self): return hash((self.address, self.net_type, self.address_type)) def __richcmp__(self, other, op): return BaseSDPConnection_richcmp(self, other, op) cdef class SDPAttributeList(list): def __contains__(self, item): if isinstance(item, BaseSDPAttribute): return list.__contains__(self, item) else: return item in [attr.name for attr in self] def getall(self, name): return [attr.value for attr in self if attr.name == name] def getfirst(self, name, default=None): for attr in self: if attr.name == name: return attr.value return default cdef class FrozenSDPAttributeList(frozenlist): def __contains__(self, item): if isinstance(item, BaseSDPAttribute): return list.__contains__(self, item) else: return item in [attr.name for attr in self] def getall(self, name): return [attr.value for attr in self if attr.name == name] def getfirst(self, name, default=None): for attr in self: if attr.name == name: return attr.value return default cdef object BaseSDPAttribute_richcmp(object self, object other, int op) with gil: cdef int eq = 1 if op not in [2,3]: return NotImplemented if not isinstance(other, BaseSDPAttribute): return NotImplemented for attr in ("name", "value"): if getattr(self, attr) != getattr(other, attr): eq = 0 break if op == 2: return bool(eq) else: return not eq cdef class BaseSDPAttribute: def __init__(self, *args, **kwargs): raise TypeError("BaseSDPAttribute cannot be instantiated directly") def __repr__(self): return "%s(%r, %r)" % (self.__class__.__name__, self.name, self.value) def __richcmp__(self, other, op): return BaseSDPAttribute_richcmp(self, other, op) cdef pjmedia_sdp_attr* get_sdp_attribute(self): return &self._sdp_attribute cdef class SDPAttribute(BaseSDPAttribute): def __init__(self, object name not None, object value not None): self.name = name self.value = value @classmethod def new(cls, BaseSDPAttribute sdp_attribute): return cls(sdp_attribute.name, sdp_attribute.value) property name: def __get__(self): return self._name def __set__(self, str name not None): _str_to_pj_str(name, &self._sdp_attribute.name) self._name = name property value: def __get__(self): return self._value def __set__(self, str value not None): _str_to_pj_str(value, &self._sdp_attribute.value) self._value = value cdef class FrozenSDPAttribute(BaseSDPAttribute): def __init__(self, object name not None, object value not None): if not self.initialized: _str_to_pj_str(name, &self._sdp_attribute.name) _str_to_pj_str(value, &self._sdp_attribute.value) self.name = name self.value = value self.initialized = 1 @classmethod def new(cls, BaseSDPAttribute sdp_attribute): if isinstance(sdp_attribute, FrozenSDPAttribute): return sdp_attribute return cls(sdp_attribute.name, sdp_attribute.value) def __hash__(self): return hash((self.name, self.value)) def __richcmp__(self, other, op): return BaseSDPAttribute_richcmp(self, other, op) cdef class SDPBandwidthInfoList(list): def __contains__(self, item): if isinstance(item, BaseSDPBandwidthInfo): return list.__contains__(self, item) else: return item in [attr.name for attr in self] cdef class FrozenSDPBandwidthInfoList(frozenlist): def __contains__(self, item): if isinstance(item, BaseSDPBandwidthInfo): return list.__contains__(self, item) else: return item in [info.modifier for info in self] cdef object BaseSDPBandwidthInfo_richcmp(object self, object other, int op) with gil: cdef int eq = 1 if op not in [2,3]: return NotImplemented if not isinstance(other, BaseSDPBandwidthInfo): return NotImplemented for attr in ("modifier", "value"): if getattr(self, attr) != getattr(other, attr): eq = 0 break if op == 2: return bool(eq) else: return not eq cdef class BaseSDPBandwidthInfo: def __init__(self, *args, **kwargs): raise TypeError("BaseSDPBandwidthInfo cannot be instantiated directly") def __repr__(self): return "%s(%r, %r)" % (self.__class__.__name__, self.modifier, self.value) def __richcmp__(self, other, op): return BaseSDPBandwidthInfo_richcmp(self, other, op) cdef pjmedia_sdp_bandw* get_sdp_bandwidth_info(self): return &self._sdp_bandwidth_info cdef class SDPBandwidthInfo(BaseSDPBandwidthInfo): def __init__(self, object modifier not None, object value not None): self.modifier = modifier self.value = value @classmethod def new(cls, BaseSDPBandwidthInfo sdp_bandwidth_info): return cls(sdp_bandwidth_info.modifier, sdp_bandwidth_info.value) property modifier: def __get__(self): return self._modifier def __set__(self, str modifier not None): _str_to_pj_str(modifier, &self._sdp_bandwidth_info.modifier) self._modifier = modifier property value: def __get__(self): return self._value def __set__(self, object value not None): self._value = value self._sdp_bandwidth_info.value = self._value cdef class FrozenSDPBandwidthInfo(BaseSDPBandwidthInfo): def __init__(self, object modifier not None, object value not None): if not self.initialized: _str_to_pj_str(modifier, &self._sdp_bandwidth_info.modifier) self.modifier = modifier self._sdp_bandwidth_info.value = value self.value = value self.initialized = 1 @classmethod def new(cls, BaseSDPBandwidthInfo sdp_bandwidth_info): if isinstance(sdp_bandwidth_info, FrozenSDPBandwidthInfo): return sdp_bandwidth_info return cls(sdp_bandwidth_info.modifier, sdp_bandwidth_info.value) def __hash__(self): return hash((self.modifier, self.value)) def __richcmp__(self, other, op): return BaseSDPBandwidthInfo_richcmp(self, other, op) # Factory functions # cdef SDPSession SDPSession_create(pjmedia_sdp_session_ptr_const pj_session): cdef SDPConnection connection = None cdef int i if pj_session.conn != NULL: connection = SDPConnection_create(pj_session.conn) return SDPSession(_pj_str_to_bytes(pj_session.origin.addr), pj_session.origin.id, pj_session.origin.version, _pj_str_to_bytes(pj_session.origin.user), _pj_str_to_bytes(pj_session.origin.net_type), _pj_str_to_bytes(pj_session.origin.addr_type), _pj_str_to_bytes(pj_session.name), connection, pj_session.time.start, pj_session.time.stop, [SDPAttribute_create(pj_session.attr[i]) for i in range(pj_session.attr_count)], [SDPBandwidthInfo_create(pj_session.bandw[i]) for i in range(pj_session.bandw_count)], [SDPMediaStream_create(pj_session.media[i]) if pj_session.media[i] != NULL else None for i in range(pj_session.media_count)]) cdef FrozenSDPSession FrozenSDPSession_create(pjmedia_sdp_session_ptr_const pj_session): cdef FrozenSDPConnection connection = None cdef int i if pj_session.conn != NULL: connection = FrozenSDPConnection_create(pj_session.conn) return FrozenSDPSession(_pj_str_to_bytes(pj_session.origin.addr), pj_session.origin.id, pj_session.origin.version, _pj_str_to_bytes(pj_session.origin.user), _pj_str_to_bytes(pj_session.origin.net_type), _pj_str_to_bytes(pj_session.origin.addr_type), _pj_str_to_bytes(pj_session.name), connection, pj_session.time.start, pj_session.time.stop, frozenlist([FrozenSDPAttribute_create(pj_session.attr[i]) for i in range(pj_session.attr_count)]), frozenlist([FrozenSDPBandwidthInfo_create(pj_session.bandw[i]) for i in range(pj_session.bandw_count)]), frozenlist([FrozenSDPMediaStream_create(pj_session.media[i]) if pj_session.media[i] != NULL else None for i in range(pj_session.media_count)])) cdef SDPMediaStream SDPMediaStream_create(pjmedia_sdp_media *pj_media): cdef SDPConnection connection = None cdef int i if pj_media.conn != NULL: connection = SDPConnection_create(pj_media.conn) return SDPMediaStream(_pj_str_to_bytes(pj_media.desc.media), pj_media.desc.port, _pj_str_to_bytes(pj_media.desc.transport), pj_media.desc.port_count, [_pj_str_to_bytes(pj_media.desc.fmt[i]) for i in range(pj_media.desc.fmt_count)], connection, [SDPAttribute_create(pj_media.attr[i]) for i in range(pj_media.attr_count)], [SDPBandwidthInfo_create(pj_media.bandw[i]) for i in range(pj_media.bandw_count)]) cdef FrozenSDPMediaStream FrozenSDPMediaStream_create(pjmedia_sdp_media *pj_media): cdef FrozenSDPConnection connection = None cdef int i if pj_media.conn != NULL: connection = FrozenSDPConnection_create(pj_media.conn) return FrozenSDPMediaStream(_pj_str_to_bytes(pj_media.desc.media), pj_media.desc.port, _pj_str_to_bytes(pj_media.desc.transport), pj_media.desc.port_count, frozenlist([_pj_str_to_bytes(pj_media.desc.fmt[i]) for i in range(pj_media.desc.fmt_count)]), connection, frozenlist([FrozenSDPAttribute_create(pj_media.attr[i]) for i in range(pj_media.attr_count)]), frozenlist([FrozenSDPBandwidthInfo_create(pj_media.bandw[i]) for i in range(pj_media.bandw_count)])) cdef SDPConnection SDPConnection_create(pjmedia_sdp_conn *pj_conn): return SDPConnection(_pj_str_to_bytes(pj_conn.addr), _pj_str_to_bytes(pj_conn.net_type), _pj_str_to_bytes(pj_conn.addr_type)) cdef FrozenSDPConnection FrozenSDPConnection_create(pjmedia_sdp_conn *pj_conn): return FrozenSDPConnection(_pj_str_to_bytes(pj_conn.addr), _pj_str_to_bytes(pj_conn.net_type), _pj_str_to_bytes(pj_conn.addr_type)) cdef SDPAttribute SDPAttribute_create(pjmedia_sdp_attr *pj_attr): return SDPAttribute(_pj_str_to_bytes(pj_attr.name), _pj_str_to_bytes(pj_attr.value)) cdef FrozenSDPAttribute FrozenSDPAttribute_create(pjmedia_sdp_attr *pj_attr): return FrozenSDPAttribute(_pj_str_to_bytes(pj_attr.name), _pj_str_to_bytes(pj_attr.value)) cdef SDPBandwidthInfo SDPBandwidthInfo_create(pjmedia_sdp_bandw *pj_bandw): return SDPBandwidthInfo(_pj_str_to_bytes(pj_bandw.modifier), int(pj_bandw.value)) cdef FrozenSDPBandwidthInfo FrozenSDPBandwidthInfo_create(pjmedia_sdp_bandw *pj_bandw): return FrozenSDPBandwidthInfo(_pj_str_to_bytes(pj_bandw.modifier), int(pj_bandw.value)) # SDP negotiator # cdef class SDPNegotiator: def __cinit__(self, *args, **kwargs): cdef pj_pool_t *pool cdef bytes pool_name cdef PJSIPUA ua ua = _get_ua() pool_name = b"SDPNegotiator_%d" % id(self) pool = ua.create_memory_pool(pool_name, 4096, 4096) self._pool = pool self._neg = NULL def __dealloc__(self): cdef PJSIPUA ua try: ua = _get_ua() except: return ua.release_memory_pool(self._pool) self._pool = NULL @classmethod def create_with_local_offer(cls, BaseSDPSession sdp_session): cdef int status cdef pjmedia_sdp_neg *neg cdef pj_pool_t *pool cdef SDPNegotiator obj obj = cls() pool = obj._pool status = pjmedia_sdp_neg_create_w_local_offer(pool, sdp_session.get_sdp_session(), &neg) if status != 0: raise PJSIPError("failed to create SDPNegotiator with local offer", status) obj._neg = neg return obj @classmethod def create_with_remote_offer(cls, BaseSDPSession sdp_session): cdef int status cdef pjmedia_sdp_neg *neg cdef pj_pool_t *pool cdef SDPNegotiator obj obj = cls() pool = obj._pool status = pjmedia_sdp_neg_create_w_remote_offer(pool, NULL, sdp_session.get_sdp_session(), &neg) if status != 0: raise PJSIPError("failed to create SDPNegotiator with remote offer", status) obj._neg = neg return obj def __repr__(self): return "%s, state=%s" % (self.__class__.__name__, self.state) property state: def __get__(self): if self._neg == NULL: return None return _buf_to_str(pjmedia_sdp_neg_state_str(pjmedia_sdp_neg_get_state(self._neg))) property active_local: def __get__(self): cdef int status cdef pjmedia_sdp_session_ptr_const sdp if self._neg == NULL: return None status = pjmedia_sdp_neg_get_active_local(self._neg, &sdp) if status != 0: return None return FrozenSDPSession_create(sdp) property active_remote: def __get__(self): cdef int status cdef pjmedia_sdp_session_ptr_const sdp if self._neg == NULL: return None status = pjmedia_sdp_neg_get_active_remote(self._neg, &sdp) if status != 0: return None return FrozenSDPSession_create(sdp) property current_local: def __get__(self): cdef int status cdef pjmedia_sdp_session_ptr_const sdp if self._neg == NULL: return None status = pjmedia_sdp_neg_get_neg_local(self._neg, &sdp) if status != 0: return None return FrozenSDPSession_create(sdp) property current_remote: def __get__(self): cdef int status cdef pjmedia_sdp_session_ptr_const sdp if self._neg == NULL: return None status = pjmedia_sdp_neg_get_neg_remote(self._neg, &sdp) if status != 0: return None return FrozenSDPSession_create(sdp) def _check_self(self): if self._neg == NULL: raise RuntimeError('SDPNegotiator was not properly initialized') def set_local_answer(self, BaseSDPSession sdp_session): self._check_self() cdef int status cdef pj_pool_t *pool = self._pool status = pjmedia_sdp_neg_set_local_answer(pool, self._neg, sdp_session.get_sdp_session()) if status != 0: raise PJSIPError("failed to set local answer", status) def set_local_offer(self, BaseSDPSession sdp_session): # PJSIP has an asymmetric API here. This function will modify the local session with the new SDP and treat it as a local offer. self._check_self() cdef int status cdef pj_pool_t *pool = self._pool status = pjmedia_sdp_neg_modify_local_offer(pool, self._neg, sdp_session.get_sdp_session()) if status != 0: raise PJSIPError("failed to set local offer", status) def set_remote_answer(self, BaseSDPSession sdp_session): self._check_self() cdef int status cdef pj_pool_t *pool = self._pool status = pjmedia_sdp_neg_set_remote_answer(pool, self._neg, sdp_session.get_sdp_session()) if status != 0: raise PJSIPError("failed to set remote answer", status) def set_remote_offer(self, BaseSDPSession sdp_session): self._check_self() cdef int status cdef pj_pool_t *pool = self._pool status = pjmedia_sdp_neg_set_remote_offer(pool, self._neg, sdp_session.get_sdp_session()) if status != 0: raise PJSIPError("failed to set remote offer", status) def cancel_offer(self): self._check_self() cdef int status status = pjmedia_sdp_neg_cancel_offer(self._neg) if status != 0: raise PJSIPError("failed to cancel offer", status) def negotiate(self): self._check_self() cdef int status cdef pj_pool_t *pool = self._pool status = pjmedia_sdp_neg_negotiate(pool, self._neg, 0) if status != 0: raise PJSIPError("SDP negotiation failed", status) diff --git a/sipsimple/core/_core.ua.pxi b/sipsimple/core/_core.ua.pxi index 131ceeb7..4dc482d1 100644 --- a/sipsimple/core/_core.ua.pxi +++ b/sipsimple/core/_core.ua.pxi @@ -1,1223 +1,1224 @@ import errno import heapq import re import random import sys import time import traceback import os import tempfile cdef class Timer: cdef int schedule(self, float delay, timer_callback callback, object obj) except -1: cdef PJSIPUA ua = _get_ua() if delay < 0: raise ValueError("delay must be a non-negative number") if callback == NULL: raise ValueError("callback must be non-NULL") if self._scheduled: raise RuntimeError("already scheduled") self.schedule_time = PyFloat_AsDouble(time.time() + delay) self.callback = callback self.obj = obj ua._add_timer(self) self._scheduled = 1 return 0 cdef int cancel(self) except -1: cdef PJSIPUA ua = _get_ua() if not self._scheduled: return 0 ua._remove_timer(self) self._scheduled = 0 return 0 cdef int call(self) except -1: self._scheduled = 0 self.callback(self.obj, self) def __richcmp__(self, other, op): cdef double diff if not isinstance(self, Timer) or not isinstance(other, Timer): return NotImplemented diff = (self).schedule_time - (other).schedule_time if op == 0: # < return diff < 0.0 elif op == 1: # <= return diff <= 0.0 elif op == 2: # == return diff == 0.0 elif op == 3: # != return diff != 0.0 elif op == 4: # > return diff > 0.0 elif op == 5: # >= return diff >= 0.0 return cdef class PJSIPUA: def __cinit__(self, *args, **kwargs): global _ua if _ua != NULL: raise SIPCoreError("Can only have one PJSUPUA instance at the same time") _ua = self self._threads = [] self._timers = list() self._events = {} self._incoming_events = set() self._incoming_requests = set() self._sent_messages = set() def __init__(self, event_handler, *args, **kwargs): global _event_queue_lock cdef object event cdef object method cdef list accept_types cdef int status cdef PJSTR message_method = PJSTR("MESSAGE") cdef PJSTR refer_method = PJSTR("REFER") cdef PJSTR str_norefersub = PJSTR("norefersub") cdef PJSTR str_gruu = PJSTR("gruu") self._event_handler = event_handler if kwargs["log_level"] < 0 or kwargs["log_level"] > PJ_LOG_MAX_LEVEL: raise ValueError("Log level should be between 0 and %d" % PJ_LOG_MAX_LEVEL) pj_log_set_level(kwargs["log_level"]) pj_log_set_decor(PJ_LOG_HAS_YEAR | PJ_LOG_HAS_MONTH | PJ_LOG_HAS_DAY_OF_MON | PJ_LOG_HAS_TIME | PJ_LOG_HAS_MICRO_SEC | PJ_LOG_HAS_SENDER | PJ_LOG_HAS_INDENT) pj_log_set_log_func(_cb_log) self._pjlib = PJLIB() pj_srand(random.getrandbits(32)) # rely on python seed for now self._caching_pool = PJCachingPool() self._pjmedia_endpoint = PJMEDIAEndpoint(self._caching_pool) self._pjsip_endpoint = PJSIPEndpoint(self._caching_pool, kwargs["ip_address"], kwargs["udp_port"], kwargs["tcp_port"], kwargs["tls_port"], kwargs["tls_verify_server"], kwargs["tls_ca_file"], kwargs["tls_cert_file"], kwargs["tls_privkey_file"], kwargs["tls_timeout"]) status = pj_mutex_create_simple(self._pjsip_endpoint._pool, "event_queue_lock", &_event_queue_lock) if status != 0: raise PJSIPError("Could not initialize event queue mutex", status) self._ip_address = kwargs["ip_address"] self.codecs = kwargs["codecs"] self.video_codecs = kwargs["video_codecs"] self._module_name = PJSTR("mod-core") self._module.name = self._module_name.pj_str self._module.id = -1 self._module.priority = PJSIP_MOD_PRIORITY_APPLICATION self._module.on_rx_request = _PJSIPUA_cb_rx_request self._module.on_tsx_state = _Request_cb_tsx_state status = pjsip_endpt_register_module(self._pjsip_endpoint._obj, &self._module) if status != 0: raise PJSIPError("Could not load application module", status) status = pjsip_endpt_add_capability(self._pjsip_endpoint._obj, &self._module, PJSIP_H_ALLOW, NULL, 1, &message_method.pj_str) if status != 0: raise PJSIPError("Could not add MESSAGE method to supported methods", status) status = pjsip_endpt_add_capability(self._pjsip_endpoint._obj, &self._module, PJSIP_H_ALLOW, NULL, 1, &refer_method.pj_str) if status != 0: raise PJSIPError("Could not add REFER method to supported methods", status) status = pjsip_endpt_add_capability(self._pjsip_endpoint._obj, NULL, PJSIP_H_SUPPORTED, NULL, 1, &str_norefersub.pj_str) if status != 0: raise PJSIPError("Could not add 'norefsub' to Supported header", status) status = pjsip_endpt_add_capability(self._pjsip_endpoint._obj, NULL, PJSIP_H_SUPPORTED, NULL, 1, &str_gruu.pj_str) if status != 0: raise PJSIPError("Could not add 'gruu' to Supported header", status) self._trace_sip = int(bool(kwargs["trace_sip"])) self._detect_sip_loops = int(bool(kwargs["detect_sip_loops"])) self._enable_colorbar_device = int(bool(kwargs["enable_colorbar_device"])) self._opus_fix_module_name = PJSTR("mod-core-opus-fix") self._opus_fix_module.name = self._opus_fix_module_name.pj_str self._opus_fix_module.id = -1 self._opus_fix_module.priority = PJSIP_MOD_PRIORITY_TRANSPORT_LAYER+1 self._opus_fix_module.on_rx_request = _cb_opus_fix_rx self._opus_fix_module.on_rx_response = _cb_opus_fix_rx self._opus_fix_module.on_tx_request = _cb_opus_fix_tx self._opus_fix_module.on_tx_response = _cb_opus_fix_tx status = pjsip_endpt_register_module(self._pjsip_endpoint._obj, &self._opus_fix_module) if status != 0: raise PJSIPError("Could not load opus-fix module", status) self._trace_module_name = PJSTR("mod-core-sip-trace") self._trace_module.name = self._trace_module_name.pj_str self._trace_module.id = -1 self._trace_module.priority = 0 self._trace_module.on_rx_request = _cb_trace_rx self._trace_module.on_rx_response = _cb_trace_rx self._trace_module.on_tx_request = _cb_trace_tx self._trace_module.on_tx_response = _cb_trace_tx status = pjsip_endpt_register_module(self._pjsip_endpoint._obj, &self._trace_module) if status != 0: raise PJSIPError("Could not load sip trace module", status) self._ua_tag_module_name = PJSTR("mod-core-ua-tag") self._ua_tag_module.name = self._ua_tag_module_name.pj_str self._ua_tag_module.id = -1 self._ua_tag_module.priority = PJSIP_MOD_PRIORITY_TRANSPORT_LAYER+1 self._ua_tag_module.on_tx_request = _cb_add_user_agent_hdr self._ua_tag_module.on_tx_response = _cb_add_server_hdr status = pjsip_endpt_register_module(self._pjsip_endpoint._obj, &self._ua_tag_module) if status != 0: raise PJSIPError("Could not load User-Agent/Server header tagging module", status) self._event_module_name = PJSTR("mod-core-events") self._event_module.name = self._event_module_name.pj_str self._event_module.id = -1 self._event_module.priority = PJSIP_MOD_PRIORITY_DIALOG_USAGE status = pjsip_endpt_register_module(self._pjsip_endpoint._obj, &self._event_module) if status != 0: raise PJSIPError("Could not load events module", status) status = pjmedia_aud_dev_set_observer_cb(_cb_audio_dev_process_event); if status != 0: raise PJSIPError("Could not set audio_change callbacks", status) status = pj_rwmutex_create(self._pjsip_endpoint._pool, "ua_audio_change_rwlock", &self.audio_change_rwlock) if status != 0: raise PJSIPError("Could not initialize audio change rwmutex", status) status = pj_mutex_create_recursive(self._pjsip_endpoint._pool, "ua_video_lock", &self.video_lock) if status != 0: raise PJSIPError("Could not initialize video mutex", status) self._user_agent = PJSTR(kwargs["user_agent"]) for event, accept_types in kwargs["events"].iteritems(): self.add_event(event, accept_types) for event in kwargs["incoming_events"]: if event not in self._events.iterkeys(): raise ValueError('Event "%s" is not known' % event) self._incoming_events.add(event) for method in kwargs["incoming_requests"]: if method in (b"ACK", b"BYE", b"INVITE", b"REFER", b"SUBSCRIBE"): raise ValueError('Handling incoming "%s" requests is not allowed' % method) self._incoming_requests.add(method) self.rtp_port_range = kwargs["rtp_port_range"] self.zrtp_cache = kwargs["zrtp_cache"] pj_stun_config_init(&self._stun_cfg, &self._caching_pool._obj.factory, 0, pjmedia_endpt_get_ioqueue(self._pjmedia_endpoint._obj), pjsip_endpt_get_timer_heap(self._pjsip_endpoint._obj)) property trace_sip: def __get__(self): self._check_self() return bool(self._trace_sip) def __set__(self, value): self._check_self() self._trace_sip = int(bool(value)) property detect_sip_loops: def __get__(self): self._check_self() return bool(self._detect_sip_loops) def __set__(self, value): self._check_self() self._detect_sip_loops = int(bool(value)) property enable_colorbar_device: def __get__(self): self._check_self() return bool(self._enable_colorbar_device) def __set__(self, value): self._check_self() self._enable_colorbar_device = int(bool(value)) self.refresh_video_devices() property events: def __get__(self): self._check_self() return self._events.copy() property ip_address: def __get__(self): self._check_self() return self._ip_address def add_event(self, object event, list accept_types): cdef pj_str_t event_pj cdef pj_str_t accept_types_pj[PJSIP_MAX_ACCEPT_COUNT] cdef int index cdef object accept_type cdef int accept_cnt = len(accept_types) cdef int status self._check_self() if accept_cnt == 0: raise SIPCoreError("Need at least one of accept_types") if accept_cnt > PJSIP_MAX_ACCEPT_COUNT: raise SIPCoreError("Too many accept_types") _str_to_pj_str(event.encode(), &event_pj) for index, accept_type in enumerate(accept_types): _str_to_pj_str(accept_type.encode(), &accept_types_pj[index]) status = pjsip_evsub_register_pkg(&self._event_module, &event_pj, 3600, accept_cnt, accept_types_pj) if status != 0: raise PJSIPError("Could not register event package", status) self._events[event] = accept_types[:] property incoming_events: def __get__(self): self._check_self() return self._incoming_events.copy() def add_incoming_event(self, object event): self._check_self() if event not in self._events.iterkeys(): raise ValueError('Event "%s" is not known' % event) self._incoming_events.add(event) def remove_incoming_event(self, object event): self._check_self() if event not in self._events.iterkeys(): raise ValueError('Event "%s" is not known' % event) self._incoming_events.discard(event) property incoming_requests: def __get__(self): self._check_self() return self._incoming_requests.copy() def add_incoming_request(self, object method): self._check_self() if method in (b"ACK", b"BYE", b"INVITE", b"REFER", b"SUBSCRIBE"): raise ValueError('Handling incoming "%s" requests is not allowed' % method) self._incoming_requests.add(method) def remove_incoming_request(self, object method): self._check_self() if method in (b"ACK", b"BYE", b"INVITE", b"REFER", b"SUBSCRIBE"): raise ValueError('Handling incoming "%s" requests is not allowed' % method) self._incoming_requests.discard(method) cdef pj_pool_t* create_memory_pool(self, bytes name, int initial_size, int resize_size): cdef pj_pool_t *pool cdef char *c_pool_name cdef pjsip_endpoint *endpoint c_pool_name = name endpoint = self._pjsip_endpoint._obj with nogil: pool = pjsip_endpt_create_pool(endpoint, c_pool_name, initial_size, resize_size) if pool == NULL: raise SIPCoreError("Could not allocate memory pool") return pool cdef void release_memory_pool(self, pj_pool_t* pool): cdef pjsip_endpoint *endpoint endpoint = self._pjsip_endpoint._obj if pool != NULL: with nogil: pjsip_endpt_release_pool(endpoint, pool) cdef void reset_memory_pool(self, pj_pool_t* pool): if pool != NULL: with nogil: pj_pool_reset(pool) cdef object _get_sound_devices(self, int is_output): cdef int count cdef pjmedia_aud_dev_info info cdef list retval = list() cdef int status with nogil: status = pj_rwmutex_lock_read(self.audio_change_rwlock) if status != 0: raise PJSIPError('Could not acquire audio_change_rwlock', status) try: for i in range(pjmedia_aud_dev_count()): with nogil: status = pjmedia_aud_dev_get_info(i, &info) if status != 0: raise PJSIPError("Could not get audio device info", status) if is_output: count = info.output_count else: count = info.input_count if count: retval.append(decode_device_name(info.name)) return retval finally: pj_rwmutex_unlock_read(self.audio_change_rwlock) cdef object _get_default_sound_device(self, int is_output): cdef pjmedia_aud_dev_info info cdef int dev_id cdef int status with nogil: status = pj_rwmutex_lock_read(self.audio_change_rwlock) if status != 0: raise SIPCoreError('Could not acquire audio_change_rwlock', status) try: if is_output: dev_id = PJMEDIA_AUD_DEFAULT_PLAYBACK_DEV else: dev_id = PJMEDIA_AUD_DEFAULT_CAPTURE_DEV with nogil: status = pjmedia_aud_dev_get_info(dev_id, &info) if status != 0: raise PJSIPError("Could not get audio device info", status) return decode_device_name(info.name) finally: pj_rwmutex_unlock_read(self.audio_change_rwlock) property default_output_device: def __get__(self): self._check_self() return self._get_default_sound_device(1) property default_input_device: def __get__(self): self._check_self() return self._get_default_sound_device(0) property output_devices: def __get__(self): self._check_self() return self._get_sound_devices(1) property input_devices: def __get__(self): self._check_self() return self._get_sound_devices(0) property sound_devices: def __get__(self): self._check_self() cdef int count cdef pjmedia_aud_dev_info info cdef list retval = list() cdef int status with nogil: status = pj_rwmutex_lock_read(self.audio_change_rwlock) if status != 0: raise SIPCoreError('Could not acquire audio_change_rwlock', status) try: for i in range(pjmedia_aud_dev_count()): with nogil: status = pjmedia_aud_dev_get_info(i, &info) if status == 0: retval.append(decode_device_name(info.name)) return retval finally: pj_rwmutex_unlock_read(self.audio_change_rwlock) def refresh_sound_devices(self): self._check_self() cdef int status cdef dict event_dict self.old_devices = self.sound_devices with nogil: status = pj_rwmutex_lock_write(self.audio_change_rwlock) if status != 0: raise SIPCoreError('Could not acquire audio_change_rwlock', status) with nogil: pjmedia_aud_dev_refresh() status = pj_rwmutex_unlock_write(self.audio_change_rwlock) if status != 0: raise SIPCoreError('Could not release audio_change_rwlock', status) event_dict = dict() event_dict["old_devices"] = self.old_devices event_dict["new_devices"] = self.sound_devices _add_event("AudioDevicesDidChange", event_dict) cdef object _get_video_devices(self): cdef pjmedia_vid_dev_info info cdef list retval = list() cdef int direction cdef int status for i in range(pjmedia_vid_dev_count()): with nogil: status = pjmedia_vid_dev_get_info(i, &info) if status != 0: raise PJSIPError("Could not get video device info", status) direction = info.dir if direction in (PJMEDIA_DIR_CAPTURE, PJMEDIA_DIR_CAPTURE_PLAYBACK): if (not self._enable_colorbar_device and bytes(info.driver) == "Colorbar") or bytes(info.driver) == "Null": continue retval.append(decode_device_name(info.name)) return retval cdef object _get_default_video_device(self): cdef pjmedia_vid_dev_info info cdef int status with nogil: status = pjmedia_vid_dev_get_info(PJMEDIA_VID_DEFAULT_CAPTURE_DEV, &info) if status != 0: raise PJSIPError("Could not get default video device info", status) if (not self._enable_colorbar_device and bytes(info.driver) == "Colorbar") or bytes(info.driver) == "Null": raise SIPCoreError("Could not get default video device") return decode_device_name(info.name) def refresh_video_devices(self): self._check_self() cdef int status cdef dict event_dict self.old_video_devices = self.video_devices with nogil: pjmedia_vid_dev_refresh() event_dict = dict() event_dict["old_devices"] = self.old_video_devices event_dict["new_devices"] = self.video_devices _add_event("VideoDevicesDidChange", event_dict) property default_video_device: def __get__(self): self._check_self() return self._get_default_video_device() property video_devices: def __get__(self): self._check_self() return self._get_video_devices() property available_codecs: def __get__(self): self._check_self() return self._pjmedia_endpoint._get_all_codecs() property codecs: def __get__(self): self._check_self() return self._pjmedia_endpoint._get_current_codecs() def __set__(self, value): self._check_self() self._pjmedia_endpoint._set_codecs(value) property available_video_codecs: def __get__(self): self._check_self() return self._pjmedia_endpoint._get_all_video_codecs() property video_codecs: def __get__(self): self._check_self() return self._pjmedia_endpoint._get_current_video_codecs() def __set__(self, value): self._check_self() self._pjmedia_endpoint._set_video_codecs(value) property udp_port: def __get__(self): self._check_self() if self._pjsip_endpoint._udp_transport == NULL: return None return self._pjsip_endpoint._udp_transport.local_name.port def set_udp_port(self, value): cdef int port self._check_self() if value is None: if self._pjsip_endpoint._udp_transport == NULL: return self._pjsip_endpoint._stop_udp_transport() else: port = value if not (0 <= port <= 65535): raise ValueError("Not a valid UDP port: %d" % value) if self._pjsip_endpoint._udp_transport != NULL: if port == self._pjsip_endpoint._udp_transport.local_name.port: return self._pjsip_endpoint._stop_udp_transport() self._pjsip_endpoint._start_udp_transport(port) property tcp_port: def __get__(self): self._check_self() if self._pjsip_endpoint._tcp_transport == NULL: return None return self._pjsip_endpoint._tcp_transport.addr_name.port def set_tcp_port(self, value): cdef int port self._check_self() if value is None: if self._pjsip_endpoint._tcp_transport == NULL: return self._pjsip_endpoint._stop_tcp_transport() else: port = value if not (0 <= port <= 65535): raise ValueError("Not a valid TCP port: %d" % value) if self._pjsip_endpoint._tcp_transport != NULL: if port == self._pjsip_endpoint._tcp_transport.addr_name.port: return self._pjsip_endpoint._stop_tcp_transport() self._pjsip_endpoint._start_tcp_transport(port) property tls_port: def __get__(self): self._check_self() if self._pjsip_endpoint._tls_transport == NULL: return None return self._pjsip_endpoint._tls_transport.addr_name.port property rtp_port_range: def __get__(self): self._check_self() return (self._rtp_port_start, self._rtp_port_start + self._rtp_port_count) def __set__(self, value): cdef int _rtp_port_start cdef int _rtp_port_stop cdef int _rtp_port_count cdef int _rtp_port_usable_count cdef int port self._check_self() for port in value: if not (0 <= port <= 65535): raise SIPCoreError("RTP port range values should be between 0 and 65535") _rtp_port_start, _rtp_port_stop = value _rtp_port_count = _rtp_port_stop - _rtp_port_start _rtp_port_usable_count = _rtp_port_count - _rtp_port_count % 2 # we need an even number of ports, so we won't use the last one if an odd number is provided if _rtp_port_usable_count < 2: raise SIPCoreError("RTP port range should contain at least 2 ports") self._rtp_port_start = _rtp_port_start self._rtp_port_count = _rtp_port_count self._rtp_port_usable_count = _rtp_port_usable_count self._rtp_port_index = 0 property user_agent: def __get__(self): self._check_self() return self._user_agent.str def __set__(self, value): self._check_self() self._user_agent = PJSTR("value") property log_level: def __get__(self): self._check_self() return pj_log_get_level() def __set__(self, value): self._check_self() if value < 0 or value > PJ_LOG_MAX_LEVEL: raise ValueError("Log level should be between 0 and %d" % PJ_LOG_MAX_LEVEL) pj_log_set_level(value) property tls_verify_server: def __get__(self): self._check_self() return bool(self._pjsip_endpoint._tls_verify_server) property tls_ca_file: def __get__(self): self._check_self() if self._pjsip_endpoint._tls_ca_file is None: return None else: return self._pjsip_endpoint._tls_ca_file.str property tls_cert_file: def __get__(self): self._check_self() if self._pjsip_endpoint._tls_cert_file is None: return None else: return self._pjsip_endpoint._tls_cert_file.str property tls_privkey_file: def __get__(self): self._check_self() if self._pjsip_endpoint._tls_privkey_file is None: return None else: return self._pjsip_endpoint._tls_privkey_file.str property tls_timeout: def __get__(self): self._check_self() return self._pjsip_endpoint._tls_timeout def set_tls_options(self, port=None, verify_server=False, ca_file=None, cert_file=None, privkey_file=None, int timeout=3000): cdef int c_port self._check_self() if port is None: if self._pjsip_endpoint._tls_transport == NULL: return self._pjsip_endpoint._stop_tls_transport() else: c_port = port if not (0 <= c_port <= 65535): raise ValueError("Not a valid TCP port: %d" % port) if ca_file is not None and not os.path.isfile(ca_file): raise ValueError("Cannot find the specified CA file: %s" % ca_file) if cert_file is not None and not os.path.isfile(cert_file): raise ValueError("Cannot find the specified certificate file: %s" % cert_file) if privkey_file is not None and not os.path.isfile(privkey_file): raise ValueError("Cannot find the specified private key file: %s" % privkey_file) if timeout < 0: raise ValueError("Invalid TLS timeout value: %d" % timeout) if self._pjsip_endpoint._tls_transport != NULL: self._pjsip_endpoint._stop_tls_transport() self._pjsip_endpoint._tls_verify_server = int(bool(verify_server)) if ca_file is None: self._pjsip_endpoint._tls_ca_file = None else: self._pjsip_endpoint._tls_ca_file = PJSTR(ca_file.encode(sys.getfilesystemencoding())) if cert_file is None: self._pjsip_endpoint._tls_cert_file = None else: self._pjsip_endpoint._tls_cert_file = PJSTR(cert_file.encode(sys.getfilesystemencoding())) if privkey_file is None: self._pjsip_endpoint._tls_privkey_file = None else: self._pjsip_endpoint._tls_privkey_file = PJSTR(privkey_file.encode(sys.getfilesystemencoding())) self._pjsip_endpoint._tls_timeout = timeout self._pjsip_endpoint._start_tls_transport(c_port) def detect_nat_type(self, stun_server_address, stun_server_port=PJ_STUN_PORT, object user_data=None): cdef pj_str_t stun_server_address_pj cdef pj_sockaddr_in stun_server cdef int status self._check_self() if not _is_valid_ip(pj_AF_INET(), stun_server_address.encode()): raise ValueError("Not a valid IPv4 address: %s" % stun_server_address) _str_to_pj_str(stun_server_address, &stun_server_address_pj) status = pj_sockaddr_in_init(&stun_server, &stun_server_address_pj, stun_server_port) if status != 0: raise PJSIPError("Could not init STUN server address", status) status = pj_stun_detect_nat_type(&stun_server, &self._stun_cfg, user_data, _cb_detect_nat_type) if status != 0: raise PJSIPError("Could not start NAT type detection", status) Py_INCREF(user_data) def set_nameservers(self, list nameservers): self._check_self() return self._pjsip_endpoint._set_dns_nameservers([n for n in nameservers if _re_ipv4.match(n)]) def set_h264_options(self, profile, level): self._check_self() self._pjmedia_endpoint._set_h264_options(str(profile), int(level.replace('.', ''))) def set_video_options(self, max_resolution, int max_framerate, object max_bitrate): self._check_self() self._pjmedia_endpoint._set_video_options(tuple(max_resolution), max_framerate, max_bitrate or 0.0) property zrtp_cache: def __get__(self): self._check_self() return self._zrtp_cache def __set__(self, value): self._check_self() if value is None: value = os.path.join(tempfile.gettempdir(), 'zrtp_cache_%d.db' % os.getpid()) self._zrtp_cache = value def __dealloc__(self): self.dealloc() def dealloc(self): global _ua, _dealloc_handler_queue, _event_queue_lock if _ua == NULL: return self._check_thread() pjmedia_aud_dev_set_observer_cb(NULL) if self.audio_change_rwlock != NULL: pj_rwmutex_destroy(self.audio_change_rwlock) self.audio_change_rwlock = NULL if self.video_lock != NULL: pj_mutex_destroy(self.video_lock) self.video_lock = NULL _process_handler_queue(self, &_dealloc_handler_queue) if _event_queue_lock != NULL: pj_mutex_lock(_event_queue_lock) pj_mutex_destroy(_event_queue_lock) _event_queue_lock = NULL self._pjsip_endpoint = None self._pjmedia_endpoint = None self._caching_pool = None self._pjlib = None _ua = NULL self._poll_log() cdef int _poll_log(self) except -1: cdef object event_name cdef dict event_params cdef list events events = _get_clear_event_queue() for event_name, event_params in events: self._event_handler(event_name, **event_params) def poll(self): global _post_poll_handler_queue cdef int status cdef double now cdef object retval = None cdef float max_timeout cdef pj_time_val pj_max_timeout cdef list timers cdef Timer timer self._check_self() max_timeout = 0.100 while self._timers: if not (self._timers[0])._scheduled: # timer was cancelled heapq.heappop(self._timers) else: max_timeout = min(max((self._timers[0]).schedule_time - time.time(), 0.0), max_timeout) break pj_max_timeout.sec = int(max_timeout) pj_max_timeout.msec = int(max_timeout * 1000) % 1000 with nogil: status = pjsip_endpt_handle_events(self._pjsip_endpoint._obj, &pj_max_timeout) IF UNAME_SYSNAME == "Darwin": if status not in [0, PJ_ERRNO_START_SYS + errno.EBADF]: raise PJSIPError("Error while handling events", status) ELSE: if status != 0: raise PJSIPError("Error while handling events", status) _process_handler_queue(self, &_post_poll_handler_queue) timers = list() now = time.time() while self._timers: if not (self._timers[0])._scheduled: # timer was cancelled heapq.heappop(self._timers) elif (self._timers[0]).schedule_time <= now: # timer needs to be processed timer = heapq.heappop(self._timers) timers.append(timer) else: break for timer in timers: timer.call() self._poll_log() if self._fatal_error: return True else: return False cdef int _handle_exception(self, int is_fatal) except -1: cdef object exc_type cdef object exc_val cdef object exc_tb exc_type, exc_val, exc_tb = sys.exc_info() if is_fatal: self._fatal_error = is_fatal _add_event("SIPEngineGotException", dict(type=exc_type, value=exc_val, traceback="".join(traceback.format_exception(exc_type, exc_val, exc_tb)))) return 0 cdef int _check_self(self) except -1: global _ua if _ua == NULL: raise SIPCoreError("The PJSIPUA is no longer running") self._check_thread() cdef int _check_thread(self) except -1: if not pj_thread_is_registered(): self._threads.append(PJSIPThread()) return 0 cdef int _add_timer(self, Timer timer) except -1: heapq.heappush(self._timers, timer) return 0 cdef int _remove_timer(self, Timer timer) except -1: # Don't remove it from the heap, just mark it as not scheduled timer._scheduled = 0 return 0 cdef int _cb_rx_request(self, pjsip_rx_data *rdata) except 0: global _event_hdr_name cdef int status cdef int bad_request cdef pjsip_tx_data *tdata = NULL cdef pjsip_hdr_ptr_const hdr_add cdef IncomingRequest request cdef Invitation inv cdef IncomingSubscription sub cdef IncomingReferral ref cdef list extra_headers cdef dict event_dict cdef dict message_params cdef pj_str_t tsx_key cdef pjsip_via_hdr *top_via cdef pjsip_via_hdr *via cdef pjsip_transaction *tsx = NULL cdef unsigned int options = PJSIP_INV_SUPPORT_100REL cdef pjsip_event_hdr *event_hdr cdef object method_name = _pj_str_to_bytes(rdata.msg_info.msg.line.req.method.name) if method_name != b"ACK": if self._detect_sip_loops: # Temporarily trick PJSIP into believing the last Via header is actually the first top_via = via = rdata.msg_info.via while True: rdata.msg_info.via = via via = pjsip_msg_find_hdr(rdata.msg_info.msg, PJSIP_H_VIA, ( via).next) if via == NULL: break status = pjsip_tsx_create_key(rdata.tp_info.pool, &tsx_key, PJSIP_ROLE_UAC, &rdata.msg_info.msg.line.req.method, rdata) rdata.msg_info.via = top_via if status != 0: raise PJSIPError("Could not generate transaction key for incoming request", status) tsx = pjsip_tsx_layer_find_tsx(&tsx_key, 0) if tsx != NULL: status = pjsip_endpt_create_response(self._pjsip_endpoint._obj, rdata, 482, NULL, &tdata) if status != 0: raise PJSIPError("Could not create response", status) elif method_name in self._incoming_requests: request = IncomingRequest() request.init(self, rdata) elif method_name == b"OPTIONS": status = pjsip_endpt_create_response(self._pjsip_endpoint._obj, rdata, 200, NULL, &tdata) if status != 0: raise PJSIPError("Could not create response", status) for hdr_type in [PJSIP_H_ALLOW, PJSIP_H_ACCEPT, PJSIP_H_SUPPORTED]: hdr_add = pjsip_endpt_get_capability(self._pjsip_endpoint._obj, hdr_type, NULL) if hdr_add != NULL: pjsip_msg_add_hdr(tdata.msg, pjsip_hdr_clone(tdata.pool, hdr_add)) elif method_name == b"INVITE": status = pjsip_inv_verify_request(rdata, &options, NULL, NULL, self._pjsip_endpoint._obj, &tdata) if status == 0: inv = Invitation() inv.init_incoming(self, rdata, options) elif method_name == b"SUBSCRIBE": event_hdr = pjsip_msg_find_hdr_by_name(rdata.msg_info.msg, &_event_hdr_name.pj_str, NULL) if event_hdr == NULL or _pj_str_to_bytes(event_hdr.event_type) not in self._incoming_events: status = pjsip_endpt_create_response(self._pjsip_endpoint._obj, rdata, 489, NULL, &tdata) if status != 0: raise PJSIPError("Could not create response", status) else: sub = IncomingSubscription() sub.init(self, rdata, _pj_str_to_bytes(event_hdr.event_type)) elif method_name == b"REFER": ref = IncomingReferral() ref.init(self, rdata) elif method_name == b"MESSAGE": bad_request = 0 extra_headers = list() message_params = dict() event_dict = dict() _pjsip_msg_to_dict(rdata.msg_info.msg, event_dict) message_params["request_uri"] = event_dict["request_uri"] message_params["from_header"] = event_dict["headers"].get("From", None) message_params["to_header"] = event_dict["headers"].get("To", None) message_params["headers"] = event_dict["headers"] message_params["body"] = event_dict["body"] content_type = message_params["headers"].get("Content-Type", None) if content_type is not None: message_params["content_type"] = content_type.content_type if message_params["headers"].get("Content-Length", 0) > 0 and message_params["body"] is None: bad_request = 1 extra_headers.append(WarningHeader(399, "local", "Missing body")) else: message_params["content_type"] = None if message_params["headers"].get("Content-Length", 0) > 0 and message_params["body"] is None: bad_request = 1 extra_headers.append(WarningHeader(399, "local", "Missing Content-Type header")) if bad_request: status = pjsip_endpt_create_response(self._pjsip_endpoint._obj, rdata, 400, NULL, &tdata) if status != 0: raise PJSIPError("Could not create response", status) _add_headers_to_tdata(tdata, extra_headers) else: _add_event("SIPEngineGotMessage", message_params) status = pjsip_endpt_create_response(self._pjsip_endpoint._obj, rdata, 200, NULL, &tdata) if status != 0: raise PJSIPError("Could not create response", status) elif method_name != b"ACK": status = pjsip_endpt_create_response(self._pjsip_endpoint._obj, rdata, 405, NULL, &tdata) if status != 0: raise PJSIPError("Could not create response", status) if tdata != NULL: status = pjsip_endpt_send_response2(self._pjsip_endpoint._obj, rdata, tdata, NULL, NULL) if status != 0: pjsip_tx_data_dec_ref(tdata) raise PJSIPError("Could not send response", status) return 1 cdef class PJSIPThread: def __cinit__(self): cdef object thread_name = "python_%d" % id(self) cdef int status status = pj_thread_register(thread_name, self._thread_desc, &self._obj) if status != 0: raise PJSIPError("Error while registering thread", status) # callback functions cdef void _cb_audio_dev_process_event(pjmedia_aud_dev_event event) with gil: cdef PJSIPUA ua event_dict = dict() try: ua = _get_ua() except: return try: if event in (PJMEDIA_AUD_DEV_DEFAULT_INPUT_CHANGED, PJMEDIA_AUD_DEV_DEFAULT_OUTPUT_CHANGED): event_dict["changed_input"] = event == PJMEDIA_AUD_DEV_DEFAULT_INPUT_CHANGED event_dict["changed_output"] = event == PJMEDIA_AUD_DEV_DEFAULT_OUTPUT_CHANGED _add_event("DefaultAudioDeviceDidChange", event_dict) elif event == PJMEDIA_AUD_DEV_LIST_WILL_REFRESH: ua.old_devices = ua.sound_devices with nogil: status = pj_rwmutex_lock_write(ua.audio_change_rwlock) if status != 0: raise SIPCoreError('Could not acquire audio_change_rwlock for writing', status) elif event == PJMEDIA_AUD_DEV_LIST_DID_REFRESH: with nogil: status = pj_rwmutex_unlock_write(ua.audio_change_rwlock) if status != 0: raise SIPCoreError('Could not release the audio_change_rwlock', status) event_dict["old_devices"] = ua.old_devices event_dict["new_devices"] = ua.sound_devices _add_event("AudioDevicesDidChange", event_dict) except: ua._handle_exception(1) cdef void _cb_detect_nat_type(void *user_data, pj_stun_nat_detect_result_ptr_const res) with gil: cdef PJSIPUA ua cdef dict event_dict cdef object user_data_obj = user_data Py_DECREF(user_data_obj) try: ua = _get_ua() except: return try: event_dict = dict() event_dict["succeeded"] = res.status == 0 event_dict["user_data"] = user_data_obj if res.status == 0: event_dict["nat_type"] = res.nat_type_name else: event_dict["error"] = res.status_text _add_event("SIPEngineDetectedNATType", event_dict) except: ua._handle_exception(0) cdef int _PJSIPUA_cb_rx_request(pjsip_rx_data *rdata) with gil: cdef PJSIPUA ua try: ua = _get_ua() except: return 0 try: return ua._cb_rx_request(rdata) except: ua._handle_exception(0) cdef int _cb_opus_fix_tx(pjsip_tx_data *tdata) with gil: cdef PJSIPUA ua cdef pjsip_msg_body *body cdef pjsip_msg_body *new_body cdef pjmedia_sdp_session *sdp cdef pjmedia_sdp_media *media cdef pjmedia_sdp_attr *attr cdef int i cdef int j cdef pj_str_t new_value try: ua = _get_ua() except: return 0 try: if tdata != NULL and tdata.msg != NULL: body = tdata.msg.body if body != NULL and _pj_str_to_str(body.content_type.type).lower() == "application" and _pj_str_to_str(body.content_type.subtype).lower() == "sdp": new_body = pjsip_msg_body_clone(tdata.pool, body) sdp = new_body.data for i in range(sdp.media_count): media = sdp.media[i] if _pj_str_to_str(media.desc.media).lower() != "audio": continue for j in range(media.attr_count): attr = media.attr[j] if _pj_str_to_str(attr.name).lower() != "rtpmap": continue attr_value = _pj_str_to_str(attr.value).lower() pos = attr_value.find("opus") if pos == -1: continue # this is the opus rtpmap attribute opus_line = attr_value[:pos] + "opus/48000/2" + opus_line = opus_line.encode() new_value.slen = len(opus_line) new_value.ptr = pj_pool_alloc(tdata.pool, new_value.slen) memcpy(new_value.ptr, PyBytes_AsString(opus_line), new_value.slen) attr.value = new_value break tdata.msg.body = new_body except: ua._handle_exception(0) return 0 cdef int _cb_opus_fix_rx(pjsip_rx_data *rdata) with gil: cdef PJSIPUA ua cdef pjsip_msg_body *body cdef int pos1 cdef int pos2 cdef char *body_ptr try: ua = _get_ua() except: return 0 try: if rdata != NULL and rdata.msg_info.msg != NULL: body = rdata.msg_info.msg.body if body != NULL and _pj_str_to_str(body.content_type.type).lower() == "application" and _pj_str_to_str(body.content_type.subtype).lower() == "sdp": body_ptr = body.data body_str = _pj_buf_len_to_str(body_ptr, body.len).lower() pos1 = body_str.find("opus/48000") if pos1 != -1: pos2 = body_str.find("opus/48000/2") if pos2 != -1: memcpy(body_ptr + pos2 + 11, '1', 1) else: # old opus, we must make it fail memcpy(body_ptr + pos1 + 5, 'XXXXX', 5) except: ua._handle_exception(0) return 0 cdef int _cb_trace_rx(pjsip_rx_data *rdata) with gil: cdef PJSIPUA ua try: ua = _get_ua() except: return 0 try: if ua._trace_sip: _add_event("SIPEngineSIPTrace", dict(received=True, source_ip=rdata.pkt_info.src_name, source_port=rdata.pkt_info.src_port, destination_ip=_pj_str_to_str(rdata.tp_info.transport.local_name.host), destination_port=rdata.tp_info.transport.local_name.port, data=_pj_buf_len_to_str(rdata.pkt_info.packet, rdata.pkt_info.len), transport=rdata.tp_info.transport.type_name)) except: ua._handle_exception(0) return 0 cdef int _cb_trace_tx(pjsip_tx_data *tdata) with gil: cdef PJSIPUA ua try: ua = _get_ua() except: return 0 try: if ua._trace_sip: _add_event("SIPEngineSIPTrace", dict(received=False, source_ip=_pj_str_to_str(tdata.tp_info.transport.local_name.host), source_port=tdata.tp_info.transport.local_name.port, destination_ip=tdata.tp_info.dst_name, destination_port=tdata.tp_info.dst_port, data=_pj_buf_len_to_str(tdata.buf.start, tdata.buf.cur - tdata.buf.start), transport=tdata.tp_info.transport.type_name)) except: ua._handle_exception(0) return 0 cdef int _cb_add_user_agent_hdr(pjsip_tx_data *tdata) with gil: cdef PJSIPUA ua cdef pjsip_hdr *hdr cdef void *found_hdr try: ua = _get_ua() except: return 0 try: found_hdr = pjsip_msg_find_hdr_by_name(tdata.msg, &_user_agent_hdr_name.pj_str, NULL) if found_hdr == NULL: hdr = pjsip_generic_string_hdr_create(tdata.pool, &_user_agent_hdr_name.pj_str, &ua._user_agent.pj_str) if hdr == NULL: raise SIPCoreError('Could not add "User-Agent" header to outgoing request') pjsip_msg_add_hdr(tdata.msg, hdr) except: ua._handle_exception(0) return 0 cdef int _cb_add_server_hdr(pjsip_tx_data *tdata) with gil: cdef PJSIPUA ua cdef pjsip_hdr *hdr cdef void *found_hdr try: ua = _get_ua() except: return 0 try: found_hdr = pjsip_msg_find_hdr_by_name(tdata.msg, &_server_hdr_name.pj_str, NULL) if found_hdr == NULL: hdr = pjsip_generic_string_hdr_create(tdata.pool, &_server_hdr_name.pj_str, &ua._user_agent.pj_str) if hdr == NULL: raise SIPCoreError('Could not add "Server" header to outgoing response') pjsip_msg_add_hdr(tdata.msg, hdr) except: ua._handle_exception(0) return 0 # functions cdef PJSIPUA _get_ua(): global _ua cdef PJSIPUA ua if _ua == NULL: raise SIPCoreError("PJSIPUA is not instantiated") ua = _ua ua._check_thread() return ua cdef int deallocate_weakref(object weak_ref, object timer) except -1 with gil: Py_DECREF(weak_ref) # globals cdef void *_ua = NULL cdef PJSTR _user_agent_hdr_name = PJSTR("User-Agent") cdef PJSTR _server_hdr_name = PJSTR("Server") cdef PJSTR _event_hdr_name = PJSTR("Event") cdef object _re_ipv4 = re.compile(r"^(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})$")