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
|
# SPDX-FileCopyrightText: Christian Amsüss and the aiocoap contributors
#
# SPDX-License-Identifier: MIT
import asyncio
import unittest
from . import common
from .test_server import (
Destructing,
WithClient,
WithTestServer,
CLEANUPTIME,
)
import aiocoap.proxy.client
import aiocoap.cli.proxy
from aiocoap.util import hostportjoin
class WithReverseProxy(Destructing):
async def asyncSetUp(self):
await super().asyncSetUp()
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),
],
)
await self.reverseproxy.initializing
async def asyncTearDown(self):
await self.reverseproxy.shutdown()
await self._del_to_be_sure("reverseproxy")
await asyncio.sleep(CLEANUPTIME)
await super().asyncTearDown()
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)",
)
async def test_routing(self):
def req():
request = aiocoap.Message(code=aiocoap.GET)
request.unresolved_remote = self.proxyaddress
request.opt.uri_path = ("big",)
return request
request = req()
response = await 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 = await 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 = await 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)",
)
async def test_options(self):
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 = await 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 = await 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 = await self.client.request(request).response
self.assertEqual(
response.code,
aiocoap.CONTENT,
"Proxy did not ignore to safe-to-forward option.",
)
|
