#!/usr/bin/env python3
# FemtoSIP -- A minimal SIP client
# Copyright (C) 2017-2023 Andreas Stöckel
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
import collections
import hashlib
import socket
import select
import random
import re
import time
import logging
# Create the logger
logger = logging.getLogger('femtosip')
###############################################################################
# Helper functions #
###############################################################################
def format_sip_header_field(key):
"""
Brings SIP header fields to a canonical form. E.g. 'content-length' becomes
'Content-Length', cseq becomes 'CSeq' and call-id becomes 'Call-ID'.
"""
if isinstance(key, bytes) or isinstance(key, bytearray):
key = str(key, 'ascii')
key = key.lower()
# Special cases
if key == 'call-id':
return 'Call-ID'
elif key == 'cseq':
return 'CSeq'
elif key == 'www-authenticate':
return 'WWW-Authenticate'
# Generic case
res = ''
for i in range(len(key)):
if i == 0 or key[i - 1] == '-':
res += key[i].upper()
else:
res += key[i]
return res
###############################################################################
# Digest handling (see RFC 7616) #
###############################################################################
class DigestTokenizer:
def __init__(self):
self.buf = ""
self.tokens = []
self.status = 0
def emit(self, special=None):
if self.buf or (self.status != 0):
self.tokens.append((False, self.buf))
if special is not None:
self.tokens.append((True, special))
self.buf = ""
def feed(self, data, last=True):
STATUS_INITIAL = 0
STATUS_IN_QUOTES = 1
STATUS_IN_QUOTES_ESCAPE = 2
for i in range(len(data)):
is_ws = data[i].isspace()
if self.status == STATUS_INITIAL:
if is_ws:
self.emit()
elif (data[i] == '=') or (data[i] == ","):
self.emit(data[i])
elif data[i] == "\"":
self.emit()
self.status = STATUS_IN_QUOTES
else:
self.buf += data[i]
elif self.status == STATUS_IN_QUOTES:
if data[i] == "\"":
self.emit()
self.status = STATUS_INITIAL
elif data[i] == "\\":
self.status = STATUS_IN_QUOTES_ESCAPE
else:
self.buf += data[i]
elif self.status == STATUS_IN_QUOTES_ESCAPE:
self.buf += data[i]
self.status = STATUS_IN_QUOTES
if last:
self.emit()
return self.tokens
@staticmethod
def tokenize(data):
return DigestTokenizer().feed(data)
def parse_digest(data):
ST_INITIAL, ST_KEY, ST_EQ, ST_VALUE, ST_COMMA = 0, 1, 2, 3, 4
tokens = DigestTokenizer.tokenize(data)
status, fields, key = ST_INITIAL, {}, None
for token in tokens:
if status == ST_INITIAL:
if token[0] or token[1].lower() != "digest":
return None
elif status == ST_KEY:
if token[0]:
return None
key = token[1].lower()
elif status == ST_EQ:
if (not token[0]) or (token[1] != "="):
return None
elif status == ST_VALUE:
if token[0]:
return None
fields[key] = token[1]
elif status == ST_COMMA:
if (not token[0]) or (token[1] != ","):
return None
status = max(1, (status + 1) % 5) # Wrap around back to ST_KEY
return fields
def digest_response(user, password, realm, nonce, method, uri):
ha1 = hashlib.md5()
ha1.update(
user.encode('utf-8') + b':' +
realm.encode('utf-8') + b':' +
password.encode('utf-8')
)
ha2 = hashlib.md5()
ha2.update(
method.upper().encode('utf-8') + b':' +
uri.encode('utf-8')
)
res = hashlib.md5()
res.update(
ha1.hexdigest().lower().encode('ascii') + b':' +
nonce.encode('utf-8') + b':' +
ha2.hexdigest().lower().encode('ascii')
)
return res.hexdigest().lower()
##############################################################################
# SIP response (i.e., HTTP-like header) parser #
##############################################################################
class ResponseParser:
"""
Parses HTTP-like response headers (such as responses from a SIP server).
"""
def __init__(self):
self.reset()
def reset(self):
self._had_lf = False
self._n_linebreaks = 0
self._status = 0
self._key = bytearray()
self._value = bytearray()
self._content_length = 0
self._skip_ws = True
self.protocol = bytearray()
self.code = bytearray()
self.message = bytearray()
self.fields = collections.OrderedDict()
self.body = bytearray()
def feed(self, data, callback):
"""
Parses the given bytes/bytearray and calls the given callback with each
parsed response. First and single parameter to the callback function is
this parser instance. Access the protocol, code, message, fields and
body member variables for the parsed content.
"""
response = [False]
def call_callback():
# Convert the message code to an integer
try:
self.code = int(str(self.code, 'ascii'))
except:
logger.error('Received invalid response code')
self.code = -1
# Convert the protocol and the message to a string
try:
self.protocol = str(self.protocol, 'ascii')
self.message = str(self.message, 'ascii')
except:
logger.error('Invalid protocol or message')
# Convert the body to "bytes"
self.body = bytes(self.body)
# Call the callback
response[0] = True
self._status = STATUS_INITIAL
callback(self)
STATUS_INITIAL = 0
STATUS_PROTOCOL = 1
STATUS_CODE = 2
STATUS_MESSAGE = 3
STATUS_HEADER_KEY = 4
STATUS_HEADER_VALUE = 5
STATUS_BODY = 6
# Iterate over the initial bytes and assemble the result
i = 0
while i < len(data):
# Skip whitespace if requested
is_ws = (data[i] == b' '[0] or data[i] == b'\t'[0])
if self._skip_ws and is_ws:
i += 1
continue
self._skip_ws = False
# Re-initialise
if self._status == STATUS_INITIAL:
self.reset()
self._status = STATUS_PROTOCOL
continue
# Handle linebreaks
if self._status != STATUS_BODY:
# Handle '\r'
if data[i] == b'\r'[0]:
self._had_lf = True
i += 1
continue
# Handle '\n\r'
if data[i] == b'\n'[0] and self._had_lf:
# Go to the next state if a line-feed is found
if self._status < STATUS_HEADER_KEY:
self._status = STATUS_HEADER_KEY
elif (self._status == STATUS_HEADER_KEY or
self._status == STATUS_HEADER_VALUE):
if len(self._key) > 0:
try:
key = format_sip_header_field(self._key)
if key == 'Content-Length':
try:
self._content_length = int(self._value)
except:
logger.error('Received invalid Content-Length')
self._content_length = 0
self.fields[key] = bytes(self._value.strip())
except:
logger.error('Invalid header key')
raise
self._key = bytearray()
self._value = bytearray()
self._status = STATUS_HEADER_KEY
# Count linebreaks, body starts with the second linebreak
self._n_linebreaks += 1
if self._n_linebreaks == 2:
self._status = STATUS_BODY
# We've handled this character, continue
i += 1
continue
# This is not a linebreak, reset the linebreak data
self._had_lf = False
self._n_linebreaks = 0
# Switch between fields in the first response line
if self._status < STATUS_MESSAGE and is_ws:
self._skip_ws = True
self._status += 1
continue
# Switch betwen states
if self._status == STATUS_PROTOCOL:
self.protocol.append(data[i])
elif self._status == STATUS_CODE:
self.code.append(data[i])
elif self._status == STATUS_MESSAGE:
self.message.append(data[i])
elif self._status == STATUS_HEADER_KEY:
if (data[i] == b':'[0]):
i += 1
self._skip_ws = True
self._status = STATUS_HEADER_VALUE
continue
self._key.append(data[i])
elif self._status == STATUS_HEADER_VALUE:
self._value.append(data[i])
elif self._status == STATUS_BODY:
if self._content_length > 0:
self.body.append(data[i])
self._content_length -= 1
if self._content_length == 0:
i += 1
call_callback()
continue
# Increase the read pointer by one
i += 1
# Explicitly call the callback if this is the end of the data
if self._status == STATUS_BODY and self._content_length == 0:
call_callback()
# Return true if the callback has been called
return response[0]
###############################################################################
# Main SIP state-machine #
###############################################################################
class SIP:
"""
The SIP class implements a minimal, incomplete, and likely broken SIP
endpoint that is capable of initiating a phone call to a third party
(without any NAT traversal).
"""
ALLOW = 'INVITE, ACK, CANCEL, OPTIONS, BYE, REFER, NOTIFY, MESSAGE, SUBSCRIBE, INFO'
def __init__(self, user, password, gateway, port, display_name=None, local_ip = None, protocol = "tcp"):
# Copy the parameters
self.user = user
self.password = password
self.local_ip, self.local_port = local_ip, None # Allow IP to be specified. Enables use on WAN.
self.gateway = gateway
self.port = port
self.display_name = display_name
self.protocol = protocol
# Create the response parser instance
self.response_parser = ResponseParser()
# Initilise the session parameters
self.seq = 0
self.session_id = self.make_random_digits(4)
self.session_version = self.make_random_digits(4)
@staticmethod
def make_sip_packet(method, uri, fields, data=b''):
"""
The sip_packet method assembles a single sip_packet consisting of a
request method, request uri, header fields, and some payload data.
The ContentLength header field is automatically added to the header
depending on the given payload data. Returns a bytes instance containg
the bytes that should be sent to the server.
method: is the request method. Will be converted to upper-case and
encoded as ascii.
uri: is the request URI that will be appended to the method. Will be
converted to ascii.
fields: is a dict containing the header fields as key-value pairs.
data: is optional payload data. Must by a bytes object.
"""
res = (method.upper().encode('ascii') + b' ' +
uri.encode('ascii') + b' SIP/2.0\r\n')
for key, value in fields.items():
res += format_sip_header_field(key).encode('ascii') + b': ' + value.encode('utf-8') + b'\r\n'
res += b'Content-Length: ' + str(len(data)).encode('ascii') + b'\r\n\r\n'
res += data
return res
@staticmethod
def make_random_digits(len=10):
res = ''
for i in range(len):
res += str(random.randint(1 if i == 0 else 0, 9))
return res
def make_branch(self):
return 'z9hG4bK' + self.make_random_digits(10)
def make_from_field(self, remote_host, tag):
from_field = '<sip:' + self.user + '@' + remote_host + '>;tag=' + tag
if self.display_name:
from_field = '"' + self.display_name + '" ' + from_field
return from_field
def make_invite_sip_packet(self,
remote_id, remote_host,
branch, tag, call_id, seq, realm=None, nonce=None):
# Assemble the request uri
uri = 'sip:' + remote_id + '@' + remote_host
# Assemble the header fields
fields = collections.OrderedDict()
fields['Via'] = (
'SIP/2.0/' + self.protocol.upper() + ' ' + self.local_ip_header + ':' + str(self.port) +
';rport;branch=' + branch)
fields['From'] = self.make_from_field(remote_host, tag)
fields['To'] = (
'<sip:' + remote_id + '@' + remote_host + '>')
fields['Call-ID'] = str(call_id)
fields['CSeq'] = str(seq) + ' INVITE'
fields['Contact'] = (
'<sip:' + self.user + '@' + self.local_ip_header +
':' + str(self.local_port) + ';transport=' + self.protocol.lower() + '>')
fields['Content-Type'] = 'application/sdp'
fields['Allow'] = self.ALLOW
fields['Max-Forwards'] = '70'
if (not realm is None) and (not nonce is None):
fields['Authorization'] = (
'Digest username=\"' + self.user + "\", " +
"realm=\"" + realm + "\", " +
"nonce=\"" + nonce + "\", " +
"uri=\"" + uri + "\", " +
"response=\"" + digest_response(
self.user, self.password,
realm, nonce, 'INVITE', uri) + "\", " +
"algorithm=\"MD5\"")
return self.make_sip_packet('INVITE', uri, fields)
def make_cancel_sip_packet(self, remote_id, remote_host, branch, tag, call_id, seq):
# Assemble the request uri
uri = 'sip:' + remote_id + '@' + remote_host
# Assemble the header fields
fields = collections.OrderedDict()
fields['Via'] = (
'SIP/2.0/' + self.protocol.upper() + ' ' + self.local_ip_header + ':' + str(self.port) +
';rport;branch=' + branch)
fields['From'] = self.make_from_field(remote_host, tag)
fields['To'] = (
'<sip:' + remote_id + '@' + remote_host + '>')
fields['Call-ID'] = str(call_id)
fields['CSeq'] = str(seq) + ' CANCEL'
fields['Max-Forwards'] = '70'
return self.make_sip_packet('CANCEL', uri, fields)
def make_bye_sip_packet(self, remote_id, remote_host, branch, tag, remote_tag, call_id, seq):
# Assemble the request uri
uri = 'sip:' + remote_id + '@' + remote_host
# Assemble the header fields
fields = collections.OrderedDict()
fields['Via'] = (
'SIP/2.0/' + self.protocol.upper() + ' ' + self.local_ip_header + ':' + str(self.port) +
';rport;branch=' + branch)
fields['From'] = self.make_from_field(remote_host, tag)
fields['To'] = (
'<sip:' + remote_id + '@' + remote_host + '>;tag=' + remote_tag)
fields['Call-ID'] = str(call_id)
fields['CSeq'] = str(seq) + ' BYE'
fields['Max-Forwards'] = '70'
return self.make_sip_packet('BYE', uri, fields)
def make_socket(self):
if self.protocol.lower() == "udp":
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
elif self.protocol.lower() == "tcp":
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
elif self.protocol.lower() == "udp6":
sock = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
elif self.protocol.lower() == "tcp6":
sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
else:
raise Exception('Unsupported protocol.')
sock.connect((self.gateway,self.port))
if self.local_ip:
self.local_port = sock.getsockname()[1]
else:
self.local_ip, self.local_port = sock.getsockname()[0:2]
sock.setblocking(0)
return sock
def call(self, remote_id, delay=15.0, timeout=1.0):
# Generate a call_id and increase the sequence number
self.seq += 1
tag = self.make_random_digits()
call_id = self.make_random_digits()
# Object containing the state of the s
state = {
'done': False,
'status': 'send_invite',
'tries': 0,
'realm': None,
'nonce': None,
'last_request': time.time(),
'delay_start': 0,
'remote_tag': None,
}
def error(msg):
logger.error(msg)
state['done'] = True
# Function advancing the state machine
def handle_response(res):
logger.info('response: ' + res.protocol + ' ' +
str(res.code) + ' ' + res.message)
# Handle the individual response codes
if res.code == 401:
# Increment the number of tries
state['tries'] += 1
# Abort if we get more than one authentication error in a row
if state['tries'] > 1:
error('Authentication failed. Check password and username.')
return
# Read realm and nonce
if not 'WWW-Authenticate' in res.fields:
error('Did not find "WWW-Authenticate" field')
return
auth = str(res.fields['WWW-Authenticate'], 'ascii')
auth_fields = parse_digest(auth)
if (auth_fields is None) or ('realm' not in auth_fields) or ('nonce' not in auth_fields):
error('Could not parse "WWW-Authenticate" header, authentication methods other than digest are not supported.')
state['realm'] = auth_fields['realm']
state['nonce'] = auth_fields['nonce']
# Try again
self.seq += 1
if state['status'].startswith('done_'):
state['status'] = state['status'][5:]
elif res.code == 100 or res.code == 101:
# Ignore this response, everything is fine
pass
elif res.code == 183 or res.code == 180:
if not 'From' in res.fields:
error('Did not find "To" field')
return
state['status'] = 'delay' # Phones are ringing, wait
state['delay_start'] = time.time()
elif res.code == 603: # Decline
state['done'] = True
elif res.code == 200:
if state['status'] == 'delay':
self.seq += 1
state['remote_tag'] = str(res.fields['To'].split(b';', 2)[-1].split(b'=', 2)[-1], 'ascii')
state['status'] = 'send_bye'
if state['status'] == 'done_send_bye':
state['done'] = True
elif res.code == 487 and state['status'] == 'done_send_cancel':
state['done'] = True
elif res.code >= 400:
error('Unhandled error.')
state['done'] = True
writebuf = bytearray()
with self.make_socket() as sock:
# IPv6 heuristics
self.protocol = self.protocol[:3]
self.local_ip_header = '[' + self.local_ip + ']' if ':' in self.local_ip else self.local_ip
self.gateway_header = '[' + self.gateway + ']' if ':' in self.gateway else self.gateway
while not state['done']:
now = time.time()
try:
if state['status'] == 'send_invite':
logger.info('request: INVITE sip:'
+ remote_id + '@' + self.gateway_header)
branch = self.make_branch()
writebuf += self.make_invite_sip_packet(
remote_id, self.gateway_header,
branch, tag, call_id, self.seq,
state['realm'], state['nonce'])
state['status'] = 'done_send_invite'
elif state['status'] == 'send_cancel':
logger.info('request: CANCEL sip:'
+ remote_id + '@' + self.gateway_header)
writebuf += self.make_cancel_sip_packet(
remote_id, self.gateway_header,
branch, tag, call_id, self.seq)
state['status'] = 'done_send_cancel'
elif state['status'] == 'send_bye':
logger.info('request: BYE sip:'
+ remote_id + '@' + self.gateway_header)
branch = self.make_branch()
writebuf += self.make_bye_sip_packet(
remote_id, self.gateway_header,
branch, tag, state['remote_tag'],
call_id, self.seq)
state['status'] = 'done_send_bye'
elif state['status'] == 'delay':
if now - state['delay_start'] > delay:
state['status'] = 'send_cancel'
elif now - state['last_request'] > timeout:
error('Timeout while waiting for server response')
# Check whether we can read or write from the socket
can_read, can_write, in_error = \
select.select([sock], [sock], [sock], 10e-3)
if len(in_error) > 0:
error('Socket error')
else:
if len(can_read) > 0:
readbuf = sock.recv(4096)
self.response_parser.feed(readbuf, handle_response)
if len(can_write) > 0 and len(writebuf) > 0:
state['last_request'] = time.time()
sent = sock.send(writebuf)
if sent == 0:
error('Error while writing to socket')
writebuf = writebuf[sent:]
except KeyboardInterrupt:
if state['status'] == 'delay':
state['status'] = 'send_cancel'
else:
state['done'] = True
#
# Main program
#
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser(
description='A microscopic SIP client that can be used to ring a ' +
'phone.')
parser.add_argument('--gateway', required=True,
help='Hostname or IP address of the SIP server')
parser.add_argument('--port', default=5060, type=int,
help='Port of the SIP server (default 5060)')
parser.add_argument('--user', required=True,
help='Username used for authentication at the SIP server')
parser.add_argument('--password', default='',
help='Password used in conjunction with the user for authentication ' +
'at the SIP server. (default '')')
parser.add_argument('--displayname', default=None,
help='Alter the displayed caller name. (defaults to SIP configuration)')
parser.add_argument('--localip', default=None)
parser.add_argument('--protocol', default="tcp",
help='supported protocols: IPv4: udp/tcp, IPv6: udp6/tcp6')
parser.add_argument('--call', required=True,
help='Phone number of the endpoint that will be called')
parser.add_argument('--delay', default=15.0, type=float,
help='Pause in seconds until the call is canceled (default 15.0)')
parser.add_argument('--timeout', default=1.0, type=float,
help='Period in seconds the SIP server must respond within (default 1.0)')
args = parser.parse_args()
logging.basicConfig(format='%(asctime)s %(message)s', level=logging.INFO)
sip = SIP(args.user, args.password, args.gateway, args.port, args.displayname, args.localip, args.protocol)
sip.call(args.call, args.delay, args.timeout)