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
|
# SPDX-FileCopyrightText: Christian Amsüss and the aiocoap contributors
#
# SPDX-License-Identifier: MIT
import asyncio
import unittest
from . import common
from .test_server import (
WithAsyncLoop,
Destructing,
WithClient,
WithTestServer,
CLEANUPTIME,
)
import aiocoap.proxy.client
import aiocoap.cli.proxy
from aiocoap.util import hostportjoin
class WithReverseProxy(WithAsyncLoop, Destructing):
def setUp(self):
super(WithReverseProxy, self).setUp()
self.reverseproxy = aiocoap.cli.proxy.Main(
[
"--reverse",
"--bind",
hostportjoin(self.proxyhost, self.proxyport),
"--namebased",
"%s:%s" % (self.name_for_real_server, self.servernetloc),
"--pathbased",
"%s:%s" % ("/".join(self.path_for_real_server), self.servernetloc),
],
loop=self.loop,
)
self.loop.run_until_complete(self.reverseproxy.initializing)
def tearDown(self):
super(WithReverseProxy, self).tearDown()
self.loop.run_until_complete(self.reverseproxy.shutdown())
self._del_to_be_sure("reverseproxy")
self.loop.run_until_complete(asyncio.sleep(CLEANUPTIME))
proxyport = 56839
proxyhost = common.loopbackname_v6 or common.loopbackname_v46
proxyaddress = "%s:%d" % (proxyhost, proxyport)
name_for_real_server = "aliasedname"
path_for_real_server = ("aliased", "name")
class TestReverseProxy(WithReverseProxy, WithClient, WithTestServer):
@unittest.skipIf(
common.using_simple6,
"Some proxy tests fail with simple6 (https://github.com/chrysn/aiocoap/issues/88)",
)
def test_routing(self):
yieldfrom = lambda f: self.loop.run_until_complete(f)
def req():
request = aiocoap.Message(code=aiocoap.GET)
request.unresolved_remote = self.proxyaddress
request.opt.uri_path = ("big",)
return request
request = req()
response = yieldfrom(self.client.request(request).response)
self.assertEqual(
response.code,
aiocoap.BAD_REQUEST,
"GET without hostname gave resource (something like BAD_REQUEST expected)",
)
request = req()
request.opt.uri_host = self.name_for_real_server
response = yieldfrom(self.client.request(request).response)
self.assertEqual(
response.code,
aiocoap.CONTENT,
"GET with hostname based proxying was not successful)",
)
request = req()
request.opt.uri_path = self.path_for_real_server + request.opt.uri_path
response = yieldfrom(self.client.request(request).response)
self.assertEqual(
response.code,
aiocoap.CONTENT,
"GET with path based proxying was not successful)",
)
@unittest.skipIf(
common.using_simple6,
"Some proxy tests fail with simple6 (https://github.com/chrysn/aiocoap/issues/88)",
)
def test_options(self):
yieldfrom = lambda f: self.loop.run_until_complete(f)
def req():
request = aiocoap.Message(code=aiocoap.GET)
request.unresolved_remote = self.proxyaddress
request.opt.uri_path = ("big",)
request.opt.uri_host = self.name_for_real_server
return request
request = req()
request.opt.proxy_scheme = "coap"
response = yieldfrom(self.client.request(request).response)
self.assertEqual(
response.code,
aiocoap.BAD_OPTION,
"Reverse proxy supports proxying even though it shouldn't.",
)
request = req()
request.opt.add_option(
aiocoap.optiontypes.StringOption(2**10 + 2, "can't proxy this")
)
response = yieldfrom(self.client.request(request).response)
self.assertEqual(
response.code, aiocoap.BAD_OPTION, "Proxy did not react to unsafe option."
)
request = req()
request.opt.add_option(
aiocoap.optiontypes.StringOption(2**10, "nothing to see here")
)
response = yieldfrom(self.client.request(request).response)
self.assertEqual(
response.code,
aiocoap.CONTENT,
"Proxy did not ignore to safe-to-forward option.",
)
|
