diff --git a/sip-message3 b/sip-message3 index 4a49eb2..9675ce7 100755 --- a/sip-message3 +++ b/sip-message3 @@ -1,508 +1,513 @@ #!/usr/bin/env python3 import atexit import os import select import signal import sys import termios import uuid from datetime import datetime from optparse import OptionParser from threading import Thread from time import sleep from application import log from application.notification import NotificationCenter, NotificationData from application.python.queue import EventQueue from application.python import Null from sipsimple.core import FromHeader, Message, RouteHeader, SIPCoreError, SIPURI, ToHeader from sipsimple.account import Account, AccountManager, BonjourAccount from sipsimple.application import SIPApplication from sipsimple.configuration import ConfigurationError from sipsimple.configuration.settings import SIPSimpleSettings from sipsimple.core import Engine from sipsimple.lookup import DNSLookup, DNSLookupError from sipsimple.storage import FileStorage from sipsimple.streams.msrp.chat import CPIMPayload, SimplePayload, CPIMParserError, CPIMHeader, ChatIdentity, CPIMNamespace from sipsimple.payloads.imdn import IMDNDocument, DisplayNotification, DeliveryNotification from sipclient.configuration import config_directory from sipclient.configuration.account import AccountExtension, BonjourAccountExtension from sipclient.configuration.settings import SIPSimpleSettingsExtension from sipclient.log import Logger from sipclient.system import IPAddressMonitor from sipsimple.util import ISOTimestamp from sipsimple.threading.green import run_in_green_thread class InputThread(Thread): def __init__(self, read_message, batch_mode): Thread.__init__(self) self.setDaemon(True) self.read_message = read_message self.batch_mode = batch_mode self._old_terminal_settings = None def start(self): atexit.register(self._termios_restore) Thread.start(self) def run(self): notification_center = NotificationCenter() if self.read_message: lines = [] try: while True: lines.append(input()) except EOFError: message = '\n'.join(lines) notification_center.post_notification('SIPApplicationGotInputMessage', sender=self, data=NotificationData(message=message)) if not self.batch_mode: while True: chars = list(self._getchars()) while chars: char = chars.pop(0) if char == '\x1b': # escape if len(chars) >= 2 and chars[0] == '[' and chars[1] in ('A', 'B', 'C', 'D'): # one of the arrow keys char = char + chars.pop(0) + chars.pop(0) notification_center.post_notification('SIPApplicationGotInput', sender=self, data=NotificationData(input=char)) def stop(self): self._termios_restore() def _termios_restore(self): if self._old_terminal_settings is not None: termios.tcsetattr(sys.stdin.fileno(), termios.TCSADRAIN, self._old_terminal_settings) def _getchars(self): fd = sys.stdin.fileno() if os.isatty(fd): self._old_terminal_settings = termios.tcgetattr(fd) new = termios.tcgetattr(fd) new[3] = new[3] & ~termios.ICANON & ~termios.ECHO new[6][termios.VMIN] = b'\000' try: termios.tcsetattr(fd, termios.TCSADRAIN, new) if select.select([fd], [], [], None)[0]: return sys.stdin.read(4192) finally: self._termios_restore() else: return os.read(fd, 4192) class SIPMessageApplication(SIPApplication): def __init__(self): self.account = None self.options = None self.target = None self.routes = [] self.registration_succeeded = False self.input = None self.output = None self.ip_address_monitor = IPAddressMonitor() self.logger = None def _write(self, message): sys.stdout.write(message) sys.stdout.flush() def start(self, target, options): notification_center = NotificationCenter() self.options = options self.message = options.message self.target = target self.input = InputThread(read_message=self.target is not None and options.message is None, batch_mode=options.batch_mode) self.output = EventQueue(self._write) self.logger = Logger(sip_to_stdout=options.trace_sip, pjsip_to_stdout=options.trace_pjsip, notifications_to_stdout=options.trace_notifications) notification_center.add_observer(self, sender=self) notification_center.add_observer(self, sender=self.input) notification_center.add_observer(self, name='SIPEngineGotMessage') if self.input: self.input.start() self.output.start() log.level.current = log.level.WARNING # get rid of twisted messages Account.register_extension(AccountExtension) BonjourAccount.register_extension(BonjourAccountExtension) SIPSimpleSettings.register_extension(SIPSimpleSettingsExtension) try: SIPApplication.start(self, FileStorage(options.config_directory or config_directory)) except ConfigurationError as e: self.output.put("Failed to load sipclient's configuration: %s\n" % str(e)) self.output.put("If an old configuration file is in place, delete it or move it and recreate the configuration using the sip_settings script.\n") self.output.stop() def _NH_SIPApplicationWillStart(self, notification): account_manager = AccountManager() notification_center = NotificationCenter() settings = SIPSimpleSettings() for account in account_manager.iter_accounts(): if isinstance(account, Account): account.sip.register = False account.presence.enabled = False account.xcap.enabled = False account.message_summary.enabled = False if self.options.account is None: self.account = account_manager.default_account else: possible_accounts = [account for account in account_manager.iter_accounts() if self.options.account in account.id and account.enabled] if len(possible_accounts) > 1: self.output.put('More than one account exists which matches %s: %s\n' % (self.options.account, ', '.join(sorted(account.id for account in possible_accounts)))) self.output.stop() self.stop() return elif len(possible_accounts) == 0: self.output.put('No enabled account that matches %s was found. Available and enabled accounts: %s\n' % (self.options.account, ', '.join(sorted(account.id for account in account_manager.get_accounts() if account.enabled)))) self.output.stop() self.stop() return else: self.account = possible_accounts[0] if isinstance(self.account, Account) and self.target is None: self.account.sip.register = True self.account.presence.enabled = False self.account.xcap.enabled = False self.account.message_summary.enabled = False notification_center.add_observer(self, sender=self.account) self.output.put('Using account %s\n' % self.account.id) self.logger.start() if settings.logs.trace_sip and self.logger._siptrace_filename is not None: self.output.put('Logging SIP trace to file "%s"\n' % self.logger._siptrace_filename) if settings.logs.trace_pjsip and self.logger._pjsiptrace_filename is not None: self.output.put('Logging PJSIP trace to file "%s"\n' % self.logger._pjsiptrace_filename) if settings.logs.trace_notifications and self.logger._notifications_filename is not None: self.output.put('Logging notifications trace to file "%s"\n' % self.logger._notifications_filename) def _NH_SIPApplicationDidStart(self, notification): notification_center = NotificationCenter() settings = SIPSimpleSettings() engine = Engine() engine.trace_sip = self.logger.sip_to_stdout or settings.logs.trace_sip engine.log_level = settings.logs.pjsip_level if (self.logger.pjsip_to_stdout or settings.logs.trace_pjsip) else 0 self.ip_address_monitor.start() if isinstance(self.account, BonjourAccount) and self.target is None: for transport in settings.sip.transport_list: try: self.output.put('Listening on: %s\n' % self.account.contact[transport]) except KeyError: pass if self.target is not None: if '@' not in self.target: self.target = '%s@%s' % (self.target, self.account.id.domain) if not self.target.startswith('sip:') and not self.target.startswith('sips:'): self.target = 'sip:' + self.target try: self.target = SIPURI.parse(self.target) except SIPCoreError: self.output.put('Illegal SIP URI: %s\n' % self.target) self.stop() if self.message is None: self.output.put('Press Ctrl+D on an empty line to end input and send the MESSAGE request.\n') else: settings = SIPSimpleSettings() lookup = DNSLookup() notification_center.add_observer(self, sender=lookup) if isinstance(self.account, Account) and self.account.sip.outbound_proxy is not None: uri = SIPURI(host=self.account.sip.outbound_proxy.host, port=self.account.sip.outbound_proxy.port, parameters={'transport': self.account.sip.outbound_proxy.transport}) elif isinstance(self.account, Account) and self.account.sip.always_use_my_proxy: uri = SIPURI(host=self.account.id.domain) else: uri = self.target lookup.lookup_sip_proxy(uri, settings.sip.transport_list, tls_name=self.account.sip.tls_name or uri.host) else: self.output.put('Press Ctrl+D to stop the program.\n') def _NH_SIPApplicationWillEnd(self, notification): self.ip_address_monitor.stop() def _NH_SIPApplicationDidEnd(self, notification): if self.input: self.input.stop() self.output.stop() self.output.join() def _NH_SIPApplicationGotInput(self, notification): if notification.data.input == '\x04': self.stop() def _NH_SIPApplicationGotInputMessage(self, notification): if not notification.data.message: self.stop() else: notification_center = NotificationCenter() settings = SIPSimpleSettings() self.message = notification.data.message lookup = DNSLookup() notification_center.add_observer(self, sender=lookup) if isinstance(self.account, Account) and self.account.sip.outbound_proxy is not None: uri = SIPURI(host=self.account.sip.outbound_proxy.host, port=self.account.sip.outbound_proxy.port, parameters={'transport': self.account.sip.outbound_proxy.transport}) elif isinstance(self.account, Account) and self.account.sip.always_use_my_proxy: uri = SIPURI(host=self.account.id.domain) else: uri = self.target lookup.lookup_sip_proxy(uri, settings.sip.transport_list, tls_name=self.account.sip.tls_name or uri.host) def _NH_SIPEngineGotException(self, notification): self.output.put('An exception occured within the SIP core:\n%s\n' % notification.data.traceback) def _NH_SIPAccountRegistrationDidSucceed(self, notification): if self.registration_succeeded: return contact_header = notification.data.contact_header contact_header_list = notification.data.contact_header_list expires = notification.data.expires registrar = notification.data.registrar message = '%s Registered contact "%s" for sip:%s at %s:%d;transport=%s (expires in %d seconds).\n' % (datetime.now().replace(microsecond=0), contact_header.uri, self.account.id, registrar.address, registrar.port, registrar.transport, expires) if len(contact_header_list) > 1: message += 'Other registered contacts:\n%s\n' % '\n'.join([' %s (expires in %s seconds)' % (str(other_contact_header.uri), other_contact_header.expires) for other_contact_header in contact_header_list if other_contact_header.uri != notification.data.contact_header.uri]) self.output.put(message) self.registration_succeeded = True def _NH_SIPAccountRegistrationDidFail(self, notification): self.output.put('%s Failed to register contact for sip:%s: %s (retrying in %.2f seconds)\n' % (datetime.now().replace(microsecond=0), self.account.id, notification.data.error, notification.data.retry_after)) self.registration_succeeded = False def _NH_SIPAccountRegistrationDidEnd(self, notification): self.output.put('%s Registration ended.\n' % datetime.now().replace(microsecond=0)) def _NH_DNSLookupDidSucceed(self, notification): self.routes = notification.data.result self._send_message() def _NH_DNSLookupDidFail(self, notification): self.output.put('DNS lookup failed: %s\n' % notification.data.error) self.stop() def _NH_SIPEngineGotMessage(self, notification): content_type = notification.data.content_type data = notification.data from_header = FromHeader.new(notification.data.from_header) from_header.parameters = {} from_header.uri.parameters = {} identity = str(from_header.uri) if from_header.display_name: identity = '"%s" <%s>' % (from_header.display_name, identity) body = notification.data.body cpim_imdn_events = None imdn_timestamp = None imdn_id = None is_cpim = False if content_type == 'message/cpim': try: cpim_message = CPIMPayload.decode(data.body) except CPIMParserError: self.output.put("Incoming SMS from %s has invalid CPIM content" % identity) return else: is_cpim = True content = cpim_message.content content_type = cpim_message.content_type sender_identity = cpim_message.sender or data.from_header imdn_timestamp = cpim_message.timestamp for h in cpim_message.additional_headers: if h.name == "Message-ID": imdn_id = h.value if h.name == "Disposition-Notification": cpim_imdn_events = h.value else: content = data.body content_type = data.content_type sender_identity = data.from_header if content_type == 'message/imdn+xml': document = IMDNDocument.parse(content) imdn_message_id = document.message_id.value imdn_status = document.notification.status.__str__() self.output.put("%s Got IMDN for message %s to %s: %s\n" % (datetime.now().replace(microsecond=0), imdn_message_id, identity, imdn_status)) return self.output.put("Got %s MESSAGE from %s: \n%s\n" % (content_type, identity, content.decode())) if cpim_imdn_events and imdn_timestamp: self.output.put("IMDN delivery notification requested for: %s\n" % cpim_imdn_events) if self.account.sms.enable_imdn: if 'delivery' in cpim_imdn_events: self.send_imdn_notification(imdn_id, imdn_timestamp, identity, sender_identity, 'delivered') if 'display' in cpim_imdn_events: self.send_imdn_notification(imdn_id, imdn_timestamp, identity, sender_identity, 'displayed') @run_in_green_thread def send_imdn_notification(self, imdn_id, imdn_timestamp, recipient, sender_identity, event): - self.output.put("Sending IMDN %s notification to %s\n" % (event, recipient)) + self.output.put("Send %s notification for %s to %s\n" % (event, imdn_id, recipient)) if event == 'delivered': notification = DeliveryNotification('delivered') elif event == 'displayed': notification = DisplayNotification(event) else: return content = IMDNDocument.create(message_id=imdn_id, datetime=imdn_timestamp, recipient_uri=recipient, notification=notification) imdn_id = str(uuid.uuid4()) ns = CPIMNamespace('urn:ietf:params:imdn', 'imdn') additional_headers = [CPIMHeader('Message-ID', ns, imdn_id)] payload = CPIMPayload(content, IMDNDocument.content_type, charset='utf-8', sender=ChatIdentity(self.account.uri, self.account.display_name), recipients=[ChatIdentity(sender_identity, None)], timestamp=ISOTimestamp.now(), additional_headers=additional_headers) payload, content_type = payload.encode() settings = SIPSimpleSettings() if isinstance(self.account, Account) and self.account.sip.outbound_proxy is not None: uri = SIPURI(host=self.account.sip.outbound_proxy.host, port=self.account.sip.outbound_proxy.port, parameters={'transport': self.account.sip.outbound_proxy.transport}) elif isinstance(self.account, Account) and self.account.sip.always_use_my_proxy: uri = SIPURI(host=self.account.id.domain) else: uri = SIPURI.parse(recipient) target = SIPURI.parse(recipient) lookup = DNSLookup() try: routes = lookup.lookup_sip_proxy(uri, settings.sip.transport_list, tls_name=self.account.sip.tls_name).wait() except DNSLookupError as e: self.output.put('Failed to lookp destination for %s' % uri) else: route = routes[0] settings = SIPSimpleSettings() from_uri = self.account.uri if self.account is BonjourAccount(): parameters = {'instance_id': settings.instance_id} from_uri.parameters.update(parameters) message_request = Message(FromHeader(from_uri, self.account.display_name), ToHeader(target), RouteHeader(route.uri), content_type, payload, credentials=self.account.credentials) message_request.send() def _NH_SIPMessageDidSucceed(self, notification): data = notification.data user_agent = data.headers.get('User-Agent', Null).body client = data.headers.get('Client', Null).body server = data.headers.get('Server', Null).body entity = user_agent or server or client self.output.put('MESSAGE was accepted by %s\n' % entity) self.stop() def _NH_SIPMessageDidFail(self, notification): notification_center = NotificationCenter() notification_center.remove_observer(self, sender=notification.sender) self.output.put('Could not deliver MESSAGE: %d %s\n' % (notification.data.code, notification.data.reason)) self._send_message() def _send_message(self): notification_center = NotificationCenter() if self.routes: route = self.routes.pop(0) identity = str(self.account.uri) if self.account.display_name: identity = '"%s" <%s>' % (self.account.display_name, identity) self.output.put("Sending MESSAGE from '%s' to '%s' using proxy %s\n" % (identity, self.target, route)) self.output.put('Press Ctrl+D to stop the program.\n') - content_type = 'text/plain' + content_type = self.options.content_type or 'text/plain' + if content_type == 'token': + content_type = "application/sylk-api-token" + additional_cpim_headers = [] additional_sip_headers = [] - if self.account.sms.enable_imdn: + if self.account.sms.enable_imdn and content_type != "application/sylk-api-token": ns = CPIMNamespace('urn:ietf:params:imdn', 'imdn') id = str(uuid.uuid4()) additional_cpim_headers = [CPIMHeader('Message-ID', ns, id)] additional_cpim_headers.append(CPIMHeader('Disposition-Notification', ns, 'positive-delivery, display')) - if self.account.sms.use_cpim: + if self.account.sms.use_cpim and not self.options.disable_cpim and content_type != "application/sylk-api-token": payload = CPIMPayload(self.message, content_type, charset='utf-8', sender=ChatIdentity(self.account.uri, self.account.display_name), recipients=[ChatIdentity(self.target, None)], timestamp=ISOTimestamp.now(), additional_headers=additional_cpim_headers) payload, content_type = payload.encode() else: payload = self.message settings = SIPSimpleSettings() from_uri = self.account.uri if self.account is BonjourAccount(): parameters = {'instance_id': settings.instance_id} from_uri.parameters.update(parameters) message_request = Message(FromHeader(from_uri, self.account.display_name), ToHeader(self.target), RouteHeader(route.uri), content_type, payload, credentials=self.account.credentials, extra_headers=additional_sip_headers) notification_center.add_observer(self, sender=message_request) message_request.send() else: self.output.put('No more routes to try. Aborting.\n') self.stop() if __name__ == '__main__': description = "This script will either sit idle waiting for an incoming MESSAGE request, or send a MESSAGE request to the specified SIP target. In outgoing mode the program will read the contents of the messages to be sent from standard input, Ctrl+D signalling EOF as usual. In listen mode the program will quit when Ctrl+D is pressed." usage = '%prog [options] [user@domain]' parser = OptionParser(usage=usage, description=description) parser.print_usage = parser.print_help + parser.add_option('-t', '--type', type='string', dest='content_type', help='The content type of the message body') parser.add_option('-a', '--account', type='string', dest='account', help='The account name to use for any outgoing traffic. If not supplied, the default account will be used.', metavar='NAME') parser.add_option('-c', '--config-directory', type='string', dest='config_directory', help='The configuration directory to use. This overrides the default location.') parser.add_option('-s', '--trace-sip', action='store_true', dest='trace_sip', default=False, help='Dump the raw contents of incoming and outgoing SIP messages.') + parser.add_option('-d', '--disable-cpim', action='store_true', dest='disable_cpim', default=False, help='Disable the use of CPIM envelope') parser.add_option('-j', '--trace-pjsip', action='store_true', dest='trace_pjsip', default=False, help='Print PJSIP logging output.') parser.add_option('-n', '--trace-notifications', action='store_true', dest='trace_notifications', default=False, help='Print all notifications (disabled by default).') parser.add_option('-b', '--batch', action='store_true', dest='batch_mode', default=False, help='Run the program in batch mode: reading control input from the console is disabled. This is particularly useful when running this script in a non-interactive environment.') parser.add_option('-m', '--message', type='string', dest='message', help='Contents of the message to send. This disables reading the message from standard input.') options, args = parser.parse_args() target = args[0] if args else None application = SIPMessageApplication() application.start(target, options) signal.signal(signal.SIGINT, signal.SIG_DFL) application.output.join() sleep(0.1)