blob: 753ab2fee4f40d4a6f2ad95b3fe82ea9d604e729 (
plain)
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
|
# Test DTLS functionality including timeout handling
try:
from tls import PROTOCOL_DTLS_CLIENT, PROTOCOL_DTLS_SERVER, SSLContext, CERT_NONE
import io
except ImportError:
print("SKIP")
raise SystemExit
class DummySocket(io.IOBase):
def __init__(self):
self.write_buffer = bytearray()
self.read_buffer = bytearray()
def write(self, data):
return len(data)
def readinto(self, buf):
# This is a placeholder socket that doesn't actually read anything
# so the read buffer is always empty.
return None
def ioctl(self, req, arg):
if req == 4: # MP_STREAM_CLOSE
return 0
return -1
# Create dummy sockets for testing
server_socket = DummySocket()
client_socket = DummySocket()
# Wrap the DTLS Server
dtls_server_ctx = SSLContext(PROTOCOL_DTLS_SERVER)
dtls_server_ctx.verify_mode = CERT_NONE
dtls_server = dtls_server_ctx.wrap_socket(
server_socket, do_handshake_on_connect=False, client_id=b"dummy_client_id"
)
print("Wrapped DTLS Server")
# wrap DTLS server with invalid client_id
try:
dtls_server = dtls_server_ctx.wrap_socket(
server_socket, do_handshake_on_connect=False, client_id=4
)
except OSError:
print("Failed to wrap DTLS Server with invalid client_id")
# Wrap the DTLS Client
dtls_client_ctx = SSLContext(PROTOCOL_DTLS_CLIENT)
dtls_client_ctx.verify_mode = CERT_NONE
dtls_client = dtls_client_ctx.wrap_socket(client_socket, do_handshake_on_connect=False)
print("Wrapped DTLS Client")
# Trigger the timing check multiple times with different elapsed times
for i in range(10): # Try multiple iterations to hit the timing window
dtls_client.write(b"test")
data = dtls_server.read(1024) # This should eventually hit the timing condition
print("OK")
|
