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
|
"""Test server utils."""
import asyncio
from datetime import timedelta
import os
import pytest
from snitun.client.client_peer import ClientPeer
from snitun.client.connector import Connector
from snitun.exceptions import SniTunConnectionError
from snitun.server.listener_peer import PeerListener
from snitun.server.peer_manager import PeerManager
from snitun.utils import server
from ..conftest import Client
from ..server.const_fernet import FERNET_TOKENS
async def test_fernet_token(
peer_listener: PeerListener,
peer_manager: PeerManager,
test_endpoint: list[Client],
) -> None:
"""Test fernet token created by server."""
client = ClientPeer("127.0.0.1", "8893")
connector = Connector("127.0.0.1", "8822")
assert not peer_manager.peer_available("localhost")
valid = timedelta(days=1)
aes_key = os.urandom(32)
aes_iv = os.urandom(16)
hostname = "localhost"
fernet_token = server.generate_client_token(
FERNET_TOKENS,
valid,
hostname,
aes_key,
aes_iv,
)
await client.start(connector, fernet_token, aes_key, aes_iv)
await asyncio.sleep(0.1)
assert peer_manager.peer_available("localhost")
await client.stop()
await asyncio.sleep(0.1)
assert not peer_manager.peer_available("localhost")
async def test_fernet_token_date(
peer_listener: PeerListener,
peer_manager: PeerManager,
test_endpoint: list[Client],
) -> None:
"""Test fernet token created by server as invalid."""
client = ClientPeer("127.0.0.1", "8893")
connector = Connector("127.0.0.1", "8822")
assert not peer_manager.peer_available("localhost")
valid = timedelta(days=-1)
aes_key = os.urandom(32)
aes_iv = os.urandom(16)
hostname = "localhost"
fernet_token = server.generate_client_token(
FERNET_TOKENS,
valid,
hostname,
aes_key,
aes_iv,
)
with pytest.raises(SniTunConnectionError):
await client.start(connector, fernet_token, aes_key, aes_iv)
await asyncio.sleep(0.1)
assert not peer_manager.peer_available("localhost")
|
