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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
|
# Copyright (C) Internet Systems Consortium, Inc. ("ISC")
#
# SPDX-License-Identifier: MPL-2.0
#
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, you can obtain one at https://mozilla.org/MPL/2.0/.
#
# See the COPYRIGHT file distributed with this work for additional
# information regarding copyright ownership.
############################################################################
#
# This tool acts as a TCP/UDP proxy and delays all incoming packets by 500
# milliseconds.
#
# We use it to check pipelining - a client sents 8 questions over a
# pipelined connection - that require asking a normal (examplea) and a
# slow-responding (exampleb) servers:
# a.examplea
# a.exampleb
# b.examplea
# b.exampleb
# c.examplea
# c.exampleb
# d.examplea
# d.exampleb
#
# If pipelining works properly the answers will be returned out of order
# with all answers from examplea returned first, and then all answers
# from exampleb.
#
############################################################################
from __future__ import print_function
import datetime
import os
import select
import signal
import socket
import sys
import time
import threading
import struct
DELAY = 0.5
THREADS = []
def log(msg):
print(datetime.datetime.now().strftime("%d-%b-%Y %H:%M:%S.%f ") + msg)
def sigterm(*_):
log("SIGTERM received, shutting down")
for thread in THREADS:
thread.close()
thread.join()
os.remove("ans.pid")
sys.exit(0)
class TCPDelayer(threading.Thread):
"""For a given TCP connection conn we open a connection to (ip, port),
and then we delay each incoming packet by DELAY by putting it in a
queue.
In the pipelined test TCP should not be used, but it's here for
completnes.
"""
def __init__(self, conn, ip, port):
threading.Thread.__init__(self)
self.conn = conn
self.cconn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.cconn.connect((ip, port))
self.queue = []
self.running = True
def close(self):
self.running = False
def run(self):
while self.running:
curr_timeout = 0.5
try:
curr_timeout = self.queue[0][0] - time.time()
except StopIteration:
pass
if curr_timeout > 0:
if curr_timeout == 0:
curr_timeout = 0.5
rfds, _, _ = select.select(
[self.conn, self.cconn], [], [], curr_timeout
)
if self.conn in rfds:
data = self.conn.recv(65535)
if not data:
return
self.queue.append((time.time() + DELAY, data))
if self.cconn in rfds:
data = self.cconn.recv(65535)
if not data == 0:
return
self.conn.send(data)
try:
while self.queue[0][0] - time.time() < 0:
_, data = self.queue.pop(0)
self.cconn.send(data)
except StopIteration:
pass
class UDPDelayer(threading.Thread):
"""Every incoming UDP packet is put in a queue for DELAY time, then
it's sent to (ip, port). We remember the query id to send the
response we get to a proper source, responses are not delayed.
"""
def __init__(self, usock, ip, port):
threading.Thread.__init__(self)
self.sock = usock
self.csock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.dst = (ip, port)
self.queue = []
self.qid_mapping = {}
self.running = True
def close(self):
self.running = False
def run(self):
while self.running:
curr_timeout = 0.5
if self.queue:
curr_timeout = self.queue[0][0] - time.time()
if curr_timeout >= 0:
if curr_timeout == 0:
curr_timeout = 0.5
rfds, _, _ = select.select(
[self.sock, self.csock], [], [], curr_timeout
)
if self.sock in rfds:
data, addr = self.sock.recvfrom(65535)
if not data:
return
self.queue.append((time.time() + DELAY, data))
qid = struct.unpack(">H", data[:2])[0]
log("Received a query from %s, queryid %d" % (str(addr), qid))
self.qid_mapping[qid] = addr
if self.csock in rfds:
data, addr = self.csock.recvfrom(65535)
if not data:
return
qid = struct.unpack(">H", data[:2])[0]
dst = self.qid_mapping.get(qid)
if dst is not None:
self.sock.sendto(data, dst)
log(
"Received a response from %s, queryid %d, sending to %s"
% (str(addr), qid, str(dst))
)
while self.queue and self.queue[0][0] - time.time() < 0:
_, data = self.queue.pop(0)
qid = struct.unpack(">H", data[:2])[0]
log("Sending a query to %s, queryid %d" % (str(self.dst), qid))
self.csock.sendto(data, self.dst)
def main():
signal.signal(signal.SIGTERM, sigterm)
signal.signal(signal.SIGINT, sigterm)
with open("ans.pid", "w") as pidfile:
print(os.getpid(), file=pidfile)
listenip = "10.53.0.5"
serverip = "10.53.0.2"
try:
port = int(os.environ["PORT"])
except KeyError:
port = 5300
log("Listening on %s:%d" % (listenip, port))
usock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
usock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
usock.bind((listenip, port))
thread = UDPDelayer(usock, serverip, port)
thread.start()
THREADS.append(thread)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind((listenip, port))
sock.listen(1)
sock.settimeout(1)
while True:
try:
(clientsock, _) = sock.accept()
log("Accepted connection from %s" % clientsock)
thread = TCPDelayer(clientsock, serverip, port)
thread.start()
THREADS.append(thread)
except socket.timeout:
pass
if __name__ == "__main__":
main()
|
