3
from pycurl import error as PyCurlError
5
from landscape import VERSION
6
from landscape.broker.transport import HTTPTransport
7
from landscape.lib import bpickle
9
from landscape.tests.helpers import LandscapeTest, LogKeeperHelper
11
from twisted.web import server, resource
12
from twisted.internet import reactor
13
from twisted.internet.ssl import DefaultOpenSSLContextFactory
14
from twisted.internet.threads import deferToThread
18
return os.path.join(os.path.dirname(__file__), path)
21
PRIVKEY = sibpath("private.ssl")
22
PUBKEY = sibpath("public.ssl")
23
BADPRIVKEY = sibpath("badprivate.ssl")
24
BADPUBKEY = sibpath("badpublic.ssl")
27
class DataCollectingResource(resource.Resource):
28
request = content = None
29
def getChild(self, request, name):
32
def render(self, request):
33
self.request = request
34
self.content = request.content.read()
35
return bpickle.dumps("Great.")
38
class HTTPTransportTest(LandscapeTest):
40
helpers = [LogKeeperHelper]
43
super(HTTPTransportTest, self).setUp()
47
super(HTTPTransportTest, self).tearDown()
48
for port in self.ports:
51
def test_get_url(self):
52
url = "http://example/ooga"
53
transport = HTTPTransport(url)
54
self.assertEquals(transport.get_url(), url)
56
def test_request_data(self):
58
When a request is sent with HTTPTransport.exchange, it should
59
include the (optional) computer ID, a user agent, and the
60
message API version as HTTP headers, and the payload as a
61
bpickled request body.
63
r = DataCollectingResource()
64
port = reactor.listenTCP(0, server.Site(r), interface="127.0.0.1")
65
self.ports.append(port)
66
transport = HTTPTransport("http://localhost:%d/"
67
% (port.getHost().port,))
68
result = deferToThread(transport.exchange, "HI", computer_id="34",
71
def got_result(ignored):
72
self.assertEquals(r.request.received_headers["x-computer-id"], "34")
73
self.assertEquals(r.request.received_headers["user-agent"],
74
"landscape-client/%s" % (VERSION,))
75
self.assertEquals(r.request.received_headers["x-message-api"],
77
self.assertEquals(bpickle.loads(r.content), "HI")
78
result.addCallback(got_result)
81
def test_ssl_verification_positive(self):
83
The client transport should complete an upload of messages to
84
a host which provides SSL data which can be verified by the
87
r = DataCollectingResource()
88
context_factory = DefaultOpenSSLContextFactory(PRIVKEY,
90
port = reactor.listenSSL(0, server.Site(r), context_factory,
91
interface="127.0.0.1")
92
self.ports.append(port)
93
transport = HTTPTransport("https://localhost:%d/"
94
% (port.getHost().port,),
96
result = deferToThread(transport.exchange, "HI", computer_id="34",
99
def got_result(ignored):
100
self.assertEquals(r.request.received_headers["x-computer-id"], "34")
101
self.assertEquals(r.request.received_headers["user-agent"],
102
"landscape-client/%s" % (VERSION,))
103
self.assertEquals(r.request.received_headers["x-message-api"],
105
self.assertEquals(bpickle.loads(r.content), "HI")
106
result.addCallback(got_result)
110
def test_ssl_verification_negative(self):
112
If the SSL server provides a key which is not verified by the
113
specified public key, then the client should immediately end
114
the connection without uploading any message data.
116
self.log_helper.ignore_errors(PyCurlError)
117
r = DataCollectingResource()
118
context_factory = DefaultOpenSSLContextFactory(BADPRIVKEY,
120
port = reactor.listenSSL(0, server.Site(r), context_factory,
121
interface="127.0.0.1")
122
self.ports.append(port)
123
transport = HTTPTransport("https://localhost:%d/"
124
% (port.getHost().port,),
127
result = deferToThread(transport.exchange, "HI", computer_id="34",
129
def got_result(ignored):
130
self.assertEquals(r.request, None)
131
self.assertEquals(r.content, None)
132
self.assertTrue("server certificate verification failed"
133
in self.logfile.getvalue())
134
result.addCallback(got_result)