python-sdks/tests/api/test_webhook.py at 593419e88ebdc03d2d9379bdcd484f336495738f · livekit/python-sdks · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
import base64
import hashlib

import pytest  # type: ignore
from livekit.api import AccessToken, TokenVerifier, WebhookReceiver

TEST_API_KEY = "myapikey"
TEST_API_SECRET = "thiskeyistotallyunsafe"
TEST_EVENT = """
{
  "event": "room_started",
  "room": {
    "sid": "RM_hycBMAjmt6Ub",
    "name": "Demo Room",
    "emptyTimeout": 300,
    "creationTime": "1692627281",
    "turnPassword": "2Pvdj+/WV1xV4EkB8klJ9xkXDWY=",
    "enabledCodecs": [
      {
        "mime": "audio/opus"
      },
      {
        "mime": "video/H264"
      },
      {
        "mime": "video/VP8"
      },
      {
        "mime": "video/AV1"
      },
      {
        "mime": "video/H264"
      },
      {
        "mime": "audio/red"
      },
      {
        "mime": "video/VP9"
      }
    ]
  },
  "id": "EV_eugWmGhovZmm",
  "createdAt": "1692985556"
}
"""


def test_webhook_receiver():
    token_verifier = TokenVerifier(TEST_API_KEY, TEST_API_SECRET)
    receiver = WebhookReceiver(token_verifier)

    hash64 = base64.b64encode(hashlib.sha256(TEST_EVENT.encode()).digest()).decode()
    token = AccessToken(TEST_API_KEY, TEST_API_SECRET)
    token.claims.sha256 = hash64
    jwt = token.to_jwt()
    receiver.receive(TEST_EVENT, jwt)


def test_bad_hash():
    token_verifier = TokenVerifier(TEST_API_KEY, TEST_API_SECRET)
    receiver = WebhookReceiver(token_verifier)

    token = AccessToken(TEST_API_KEY, TEST_API_SECRET)
    hash64 = base64.b64encode(hashlib.sha256("wrong_hash".encode()).digest()).decode()
    token.claims.sha256 = hash64
    jwt = token.to_jwt()
    with pytest.raises(Exception):
        receiver.receive(TEST_EVENT, jwt)


def test_invalid_body():
    token_verifier = TokenVerifier(TEST_API_KEY, TEST_API_SECRET)
    receiver = WebhookReceiver(token_verifier)

    token = AccessToken(TEST_API_KEY, TEST_API_SECRET)
    body = "invalid body"
    hash64 = base64.b64encode(hashlib.sha256(body.encode()).digest()).decode()
    token.claims.sha256 = hash64
    jwt = token.to_jwt()
    with pytest.raises(Exception):
        receiver.receive(body, jwt)