-
Notifications
You must be signed in to change notification settings - Fork 3
/
Copy pathssl_throughput.py
68 lines (57 loc) · 1.61 KB
/
ssl_throughput.py
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
"""
This benchmarks runs a trivial Twisted TLSv1 echo server using a certificate
with a 2048 bit RSA key as well as a client which pumps as much data to that
server as it can in a fixed period of time.
"""
from tcp_throughput import Client, driver
from twisted.internet.endpoints import SSL4ClientEndpoint
from twisted.internet.protocol import ServerFactory
from twisted.internet.ssl import (
DN,
CertificateOptions,
KeyPair,
PrivateCertificate,
)
from twisted.protocols.wire import Echo
# Generate a new self-signed certificate
key = KeyPair.generate(size=2048)
req = key.certificateRequest(
DN(commonName='localhost'), digestAlgorithm='sha1'
)
cert = PrivateCertificate.load(
key.signCertificateRequest(
DN(commonName='localhost'),
req,
lambda dn: True,
1,
digestAlgorithm='sha1',
),
key,
)
def main(reactor, duration):
chunkSize = 16384
server = ServerFactory()
server.protocol = Echo
port = reactor.listenSSL(0, server, cert.options())
client = Client(
reactor,
SSL4ClientEndpoint(
reactor,
'127.0.0.1',
port.getHost().port,
CertificateOptions(
verify=True, requireCertificate=True, caCerts=[cert.original]
),
),
)
d = client.run(duration, chunkSize)
def cleanup(passthrough):
d = port.stopListening()
d.addCallback(lambda ignored: passthrough)
return d
d.addCallback(cleanup)
return d
if __name__ == '__main__':
import sys
import ssl_throughput
driver(ssl_throughput.main, sys.argv)