104 lines
6.5 KiB
Python
104 lines
6.5 KiB
Python
"""Ephemeral test peer: generated identity and synthetic media, no saved keys."""
|
|
import asyncio, base64, fractions, hashlib, json, time
|
|
import aiohttp
|
|
from cryptography.hazmat.primitives.asymmetric import ec
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
from aiortc import RTCPeerConnection, RTCConfiguration, RTCIceServer, RTCSessionDescription, VideoStreamTrack, AudioStreamTrack
|
|
from aiortc.sdp import candidate_from_sdp
|
|
from av import VideoFrame
|
|
|
|
class TestVideo(VideoStreamTrack):
|
|
async def recv(self):
|
|
pts, tb = await self.next_timestamp()
|
|
frame = VideoFrame(320, 240, 'yuv420p')
|
|
for index, plane in enumerate(frame.planes): plane.update(bytes([100 if index == 0 else 128]) * plane.buffer_size)
|
|
frame.pts, frame.time_base = pts, tb
|
|
return frame
|
|
|
|
async def main():
|
|
async with aiohttp.ClientSession() as http:
|
|
token = None
|
|
async def rpc(op, payload):
|
|
async with http.post(f'http://127.0.0.1:5099/api/ikar/requests/{op}',json={'payload':payload},headers={'Authorization':f'Bearer {token}'} if token else {}) as r:
|
|
data = await r.json()
|
|
if data.get('cmd') != 1: raise RuntimeError(f'RPC {op}: {data.get("payload")}')
|
|
return data['payload']
|
|
key=ec.generate_private_key(ec.SECP256R1())
|
|
public=key.public_key().public_bytes(serialization.Encoding.DER,serialization.PublicFormat.SubjectPublicKeyInfo)
|
|
challenge=await rpc(17,{'phone':f'+999{int(time.time()*1000)}','installationId':hashlib.sha256(public).hexdigest()[:32],'devicePublicKey':base64.b64encode(public).decode()})
|
|
signature=key.sign(base64.b64decode(challenge['signaturePayload']),ec.ECDSA(hashes.SHA256()))
|
|
verified=await rpc(18,{'token':challenge['token'],'deviceSignature':base64.b64encode(signature).decode()})
|
|
token=verified['tokenAttrs']['REGISTER']['token']
|
|
profile=await rpc(23,{'firstName':'IKAR video test','lastName':'Synthetic media'})
|
|
token=profile['token']; user_id=profile['accountId']
|
|
pc=None
|
|
try:
|
|
target=(await rpc(46,{'phone':'+12025550101'}))['contact']['id']
|
|
call=await rpc(78,{'calleeIds':[target],'isVideo':True,'conversationId':f'ikar-rtp-test-{time.time_ns()}'})
|
|
endpoint=json.loads(call['internalCallerParams'])['endpoint']
|
|
# The own test peer uses the local server socket; native APK uses HTTPS.
|
|
endpoint=endpoint.replace('wss://ikar.kusoft.xyz','ws://127.0.0.1:5099')
|
|
async with http.ws_connect(endpoint,heartbeat=10) as socket:
|
|
initial=await socket.receive_json()
|
|
cfg=initial['conversationParams']
|
|
servers=[]
|
|
for kind in ('stun','turn'):
|
|
s=cfg.get(kind)
|
|
if s: servers.append(RTCIceServer(s['urls'],username=s.get('username'),credential=s.get('credential')))
|
|
pc=RTCPeerConnection(RTCConfiguration(iceServers=servers))
|
|
pc.addTrack(AudioStreamTrack()); pc.addTrack(TestVideo())
|
|
frames={'audio':0,'video':0}
|
|
readers=[]
|
|
@pc.on('track')
|
|
def on_track(track):
|
|
async def consume():
|
|
try:
|
|
while True:
|
|
await track.recv(); frames[track.kind]+=1
|
|
except Exception: pass
|
|
readers.append(asyncio.create_task(consume()))
|
|
@pc.on('connectionstatechange')
|
|
async def state(): print(json.dumps({'state':pc.connectionState}),flush=True)
|
|
sequence=0
|
|
async def transmit(data):
|
|
nonlocal sequence
|
|
sequence+=1
|
|
await socket.send_json({'command':'transmit-data','sequence':sequence,'data':data,'participantId':target})
|
|
# Wait for the native user to accept before negotiating.
|
|
print(json.dumps({'status':'CALLING','target':target,'testPeer':user_id}),flush=True)
|
|
deadline=time.monotonic()+150
|
|
offered=False
|
|
while time.monotonic()<deadline:
|
|
try: event=await asyncio.wait_for(socket.receive_json(),timeout=2)
|
|
except asyncio.TimeoutError: event={}
|
|
if event.get('notification') in ('accepted-call','participant-joined') and not offered:
|
|
offered=True
|
|
await pc.setLocalDescription(await pc.createOffer())
|
|
await transmit({'sdp':{'type':pc.localDescription.type,'sdp':pc.localDescription.sdp}})
|
|
if event.get('notification')=='transmitted-data':
|
|
data=event.get('data',{})
|
|
if 'sdp' in data:
|
|
description=data['sdp']
|
|
await pc.setRemoteDescription(RTCSessionDescription(description['sdp'],description['type']))
|
|
if description['type']=='offer':
|
|
await pc.setLocalDescription(await pc.createAnswer())
|
|
await transmit({'sdp':{'type':pc.localDescription.type,'sdp':pc.localDescription.sdp}})
|
|
if data.get('candidate',{}).get('candidate'):
|
|
candidate=data['candidate']; ice=candidate_from_sdp(candidate['candidate'].removeprefix('candidate:'))
|
|
ice.sdpMid=candidate.get('sdpMid'); ice.sdpMLineIndex=candidate.get('sdpMLineIndex')
|
|
await pc.addIceCandidate(ice)
|
|
if frames['audio']>25 and frames['video']>5:
|
|
stats=await pc.getStats()
|
|
outbound=[{'kind':s.kind,'packetsSent':s.packetsSent,'bytesSent':s.bytesSent} for s in stats.values() if s.type=='outbound-rtp']
|
|
report={'passed':True,'connection':pc.connectionState,'receivedFrames':frames,'outbound':outbound,'syntheticMedia':True,'nativeDevice':'Android emulator API35'}
|
|
print(json.dumps(report),flush=True)
|
|
await socket.send_json({'command':'hangup','sequence':sequence+1,'reason':'FINISHED'})
|
|
break
|
|
else: raise RuntimeError(f'Media timed out: {pc.connectionState}, received={frames}')
|
|
for reader in readers: reader.cancel()
|
|
finally:
|
|
if pc: await pc.close()
|
|
await rpc(199,{})
|
|
|
|
asyncio.run(main())
|