73 lines
4.2 KiB
Python
73 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Isolated UDP SIP UAC/UAS on test B. No PSTN routing or remote targets."""
|
|
import json
|
|
import re
|
|
import socket
|
|
import threading
|
|
import time
|
|
import uuid
|
|
|
|
events = []
|
|
stop = threading.Event()
|
|
|
|
def header(message, name):
|
|
match = re.search(r'^' + re.escape(name) + r':\s*(.+)$', message, re.I | re.M)
|
|
return match.group(1).strip() if match else ''
|
|
|
|
def reply(message, code, reason):
|
|
to = header(message, 'To')
|
|
if ';tag=' not in to:
|
|
to += ';tag=cra-uas'
|
|
vias = re.findall(r'^Via:\s*(.+)$', message, re.I | re.M)
|
|
routes = re.findall(r'^Record-Route:\s*(.+)$', message, re.I | re.M)
|
|
lines = [f'SIP/2.0 {code} {reason}'] + [f'Via: {v.strip()}' for v in vias]
|
|
lines += [f'From: {header(message,"From")}', f'To: {to}', f'Call-ID: {header(message,"Call-ID")}', f'CSeq: {header(message,"CSeq")}']
|
|
lines += [f'Record-Route: {r.strip()}' for r in routes]
|
|
lines += ['Contact: <sip:uas@127.0.0.1:50630>', 'Content-Length: 0', '', '']
|
|
return '\r\n'.join(lines).encode()
|
|
|
|
def uas():
|
|
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
|
|
sock.bind(('127.0.0.1',50630)); sock.settimeout(.2)
|
|
while not stop.is_set():
|
|
try: data, addr = sock.recvfrom(65535)
|
|
except socket.timeout: continue
|
|
msg = data.decode(); method = msg.split()[0]
|
|
if method == 'INVITE':
|
|
number = re.search(r'INVITE sip:([^@]+)',msg).group(1)
|
|
cases = {'99101':[(180,'Ringing'),(486,'Busy Here')], '99102':[(183,'Session Progress'),(487,'Request Terminated')], '99103':[(486,'Busy Here')], '99104':[(200,'OK')], '99105':[(180,'Ringing'),(200,'OK')], '99106':[(100,'Trying'),(486,'Busy Here')]}
|
|
for code, reason in cases[number]:
|
|
sock.sendto(reply(msg,code,reason),addr)
|
|
events.append({'callId':header(msg,'Call-ID'),'direction':'UAS','code':code,'time':time.time()})
|
|
time.sleep(.15 if number!='99106' else 1.5)
|
|
elif method == 'BYE': sock.sendto(reply(msg,200,'OK'),addr)
|
|
|
|
thread = threading.Thread(target=uas,daemon=True); thread.start(); time.sleep(.3)
|
|
try:
|
|
for number in ['99101','99102','99103','99104','99105','99106']:
|
|
with socket.socket(socket.AF_INET,socket.SOCK_DGRAM) as sock:
|
|
sock.bind(('127.0.0.1',0)); sock.settimeout(8); port=sock.getsockname()[1]
|
|
callid=f'cra-{number}-{uuid.uuid4().hex}@loopback'; tag=uuid.uuid4().hex[:8]
|
|
branch='z9hG4bK'+uuid.uuid4().hex
|
|
base=[f'Via: SIP/2.0/UDP 127.0.0.1:{port};branch={branch};rport', f'From: <sip:99100001@127.0.0.1>;tag={tag}', f'To: <sip:{number}@127.0.0.1>', f'Call-ID: {callid}', f'Contact: <sip:99100001@127.0.0.1:{port}>', 'Max-Forwards: 70']
|
|
invite='\r\n'.join([f'INVITE sip:{number}@127.0.0.1:15060 SIP/2.0',*base,'CSeq: 1 INVITE','Content-Length: 0','',''])
|
|
sock.sendto(invite.encode(),('127.0.0.1',15060))
|
|
while True:
|
|
msg=sock.recv(65535).decode(); code=int(msg.split()[1]); events.append({'callId':callid,'direction':'UAC','code':code,'time':time.time()})
|
|
if code<200: continue
|
|
if code>=300:
|
|
ack=invite.replace('INVITE sip:','ACK sip:',1).replace('CSeq: 1 INVITE','CSeq: 1 ACK').replace(base[2],f'To: {header(msg,"To")}')
|
|
sock.sendto(ack.encode(),('127.0.0.1',15060)); break
|
|
routes = list(reversed(re.findall(r'^Record-Route:\s*(.+)$',msg,re.I|re.M)))
|
|
for method,cseq in [('ACK',1),('BYE',2)]:
|
|
if method=='BYE': time.sleep(.3 if number=='99105' else 1.5)
|
|
seq=[f'{method} sip:uas@127.0.0.1:50630 SIP/2.0',f'Via: SIP/2.0/UDP 127.0.0.1:{port};branch=z9hG4bK{uuid.uuid4().hex};rport',base[1],f'To: {header(msg,"To")}',base[3],*['Route: '+r.strip() for r in routes],f'CSeq: {cseq} {method}','Max-Forwards: 70','Content-Length: 0','','']
|
|
sock.sendto('\r\n'.join(seq).encode(),('127.0.0.1',15060))
|
|
if method=='BYE':
|
|
final=sock.recv(65535).decode(); assert final.startswith('SIP/2.0 200'), final
|
|
break
|
|
time.sleep(.3)
|
|
finally:
|
|
stop.set(); thread.join(timeout=2)
|
|
print(json.dumps(events))
|