56 lines
2.1 KiB
Python
56 lines
2.1 KiB
Python
|
"""ITs for the WebRTC mesh TURN-credentials endpoint — Phase C of
|
||
|
|
[[my-sea-invite-voice-blueprint]]. Verifies the coturn `use-auth-secret`
|
||
|
|
REST scheme (HMAC-SHA1 username/credential) + auth gating.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import base64
|
||
|
|
import hashlib
|
||
|
|
import hmac
|
||
|
|
|
||
|
|
from django.test import TestCase, override_settings
|
||
|
|
from django.urls import reverse
|
||
|
|
|
||
|
|
from apps.lyric.models import User
|
||
|
|
|
||
|
|
|
||
|
|
@override_settings(
|
||
|
|
COTURN_SHARED_SECRET="testsecret",
|
||
|
|
COTURN_TURN_HOST="turn.test",
|
||
|
|
COTURN_TTL=86400,
|
||
|
|
)
|
||
|
|
class TURNCredentialsTest(TestCase):
|
||
|
|
def setUp(self):
|
||
|
|
self.user = User.objects.create(email="turn@test.io", username="turner")
|
||
|
|
self.url = reverse("api_turn_credentials")
|
||
|
|
|
||
|
|
def test_requires_authentication(self):
|
||
|
|
resp = self.client.get(self.url)
|
||
|
|
self.assertIn(resp.status_code, (401, 403))
|
||
|
|
|
||
|
|
def test_returns_ice_servers_and_ttl(self):
|
||
|
|
self.client.force_login(self.user)
|
||
|
|
data = self.client.get(self.url).json()
|
||
|
|
self.assertEqual(data["ttl"], 86400)
|
||
|
|
urls = [s for srv in data["iceServers"] for s in srv["urls"]]
|
||
|
|
self.assertTrue(any(u.startswith("stun:turn.test") for u in urls))
|
||
|
|
self.assertTrue(any("turn:turn.test" in u and "transport=udp" in u for u in urls))
|
||
|
|
self.assertTrue(any("turn:turn.test" in u and "transport=tcp" in u for u in urls))
|
||
|
|
|
||
|
|
def test_username_is_expiry_colon_user_id(self):
|
||
|
|
self.client.force_login(self.user)
|
||
|
|
data = self.client.get(self.url).json()
|
||
|
|
expiry_str, _, uid = data["username"].partition(":")
|
||
|
|
self.assertTrue(expiry_str.isdigit())
|
||
|
|
self.assertEqual(uid, str(self.user.id))
|
||
|
|
|
||
|
|
def test_credential_is_hmac_sha1_of_username(self):
|
||
|
|
self.client.force_login(self.user)
|
||
|
|
data = self.client.get(self.url).json()
|
||
|
|
expected = base64.b64encode(
|
||
|
|
hmac.new(b"testsecret", data["username"].encode(), hashlib.sha1).digest()
|
||
|
|
).decode()
|
||
|
|
self.assertEqual(data["credential"], expected)
|
||
|
|
# The TURN iceServer entry carries the same credential.
|
||
|
|
turn = [s for s in data["iceServers"] if "credential" in s][0]
|
||
|
|
self.assertEqual(turn["credential"], expected)
|