You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
482 lines
16 KiB
482 lines
16 KiB
import sys
|
|
import socket
|
|
import warnings
|
|
|
|
from waitress.compat import (
|
|
PY2,
|
|
WIN,
|
|
)
|
|
|
|
if sys.version_info[:2] == (2, 6): # pragma: no cover
|
|
import unittest2 as unittest
|
|
else: # pragma: no cover
|
|
import unittest
|
|
|
|
|
|
class Test_asbool(unittest.TestCase):
|
|
def _callFUT(self, s):
|
|
from waitress.adjustments import asbool
|
|
|
|
return asbool(s)
|
|
|
|
def test_s_is_None(self):
|
|
result = self._callFUT(None)
|
|
self.assertEqual(result, False)
|
|
|
|
def test_s_is_True(self):
|
|
result = self._callFUT(True)
|
|
self.assertEqual(result, True)
|
|
|
|
def test_s_is_False(self):
|
|
result = self._callFUT(False)
|
|
self.assertEqual(result, False)
|
|
|
|
def test_s_is_true(self):
|
|
result = self._callFUT("True")
|
|
self.assertEqual(result, True)
|
|
|
|
def test_s_is_false(self):
|
|
result = self._callFUT("False")
|
|
self.assertEqual(result, False)
|
|
|
|
def test_s_is_yes(self):
|
|
result = self._callFUT("yes")
|
|
self.assertEqual(result, True)
|
|
|
|
def test_s_is_on(self):
|
|
result = self._callFUT("on")
|
|
self.assertEqual(result, True)
|
|
|
|
def test_s_is_1(self):
|
|
result = self._callFUT(1)
|
|
self.assertEqual(result, True)
|
|
|
|
|
|
class Test_as_socket_list(unittest.TestCase):
|
|
def test_only_sockets_in_list(self):
|
|
from waitress.adjustments import as_socket_list
|
|
|
|
sockets = [
|
|
socket.socket(socket.AF_INET, socket.SOCK_STREAM),
|
|
socket.socket(socket.AF_INET6, socket.SOCK_STREAM),
|
|
]
|
|
if hasattr(socket, "AF_UNIX"):
|
|
sockets.append(socket.socket(socket.AF_UNIX, socket.SOCK_STREAM))
|
|
new_sockets = as_socket_list(sockets)
|
|
self.assertEqual(sockets, new_sockets)
|
|
for sock in sockets:
|
|
sock.close()
|
|
|
|
def test_not_only_sockets_in_list(self):
|
|
from waitress.adjustments import as_socket_list
|
|
|
|
sockets = [
|
|
socket.socket(socket.AF_INET, socket.SOCK_STREAM),
|
|
socket.socket(socket.AF_INET6, socket.SOCK_STREAM),
|
|
{"something": "else"},
|
|
]
|
|
new_sockets = as_socket_list(sockets)
|
|
self.assertEqual(new_sockets, [sockets[0], sockets[1]])
|
|
for sock in [sock for sock in sockets if isinstance(sock, socket.socket)]:
|
|
sock.close()
|
|
|
|
|
|
class TestAdjustments(unittest.TestCase):
|
|
def _hasIPv6(self): # pragma: nocover
|
|
if not socket.has_ipv6:
|
|
return False
|
|
|
|
try:
|
|
socket.getaddrinfo(
|
|
"::1",
|
|
0,
|
|
socket.AF_UNSPEC,
|
|
socket.SOCK_STREAM,
|
|
socket.IPPROTO_TCP,
|
|
socket.AI_PASSIVE | socket.AI_ADDRCONFIG,
|
|
)
|
|
|
|
return True
|
|
except socket.gaierror as e:
|
|
# Check to see what the error is
|
|
if e.errno == socket.EAI_ADDRFAMILY:
|
|
return False
|
|
else:
|
|
raise e
|
|
|
|
def _makeOne(self, **kw):
|
|
from waitress.adjustments import Adjustments
|
|
|
|
return Adjustments(**kw)
|
|
|
|
def test_goodvars(self):
|
|
inst = self._makeOne(
|
|
host="localhost",
|
|
port="8080",
|
|
threads="5",
|
|
trusted_proxy="192.168.1.1",
|
|
trusted_proxy_headers={"forwarded"},
|
|
trusted_proxy_count=2,
|
|
log_untrusted_proxy_headers=True,
|
|
url_scheme="https",
|
|
backlog="20",
|
|
recv_bytes="200",
|
|
send_bytes="300",
|
|
outbuf_overflow="400",
|
|
inbuf_overflow="500",
|
|
connection_limit="1000",
|
|
cleanup_interval="1100",
|
|
channel_timeout="1200",
|
|
log_socket_errors="true",
|
|
max_request_header_size="1300",
|
|
max_request_body_size="1400",
|
|
expose_tracebacks="true",
|
|
ident="abc",
|
|
asyncore_loop_timeout="5",
|
|
asyncore_use_poll=True,
|
|
unix_socket_perms="777",
|
|
url_prefix="///foo/",
|
|
ipv4=True,
|
|
ipv6=False,
|
|
)
|
|
|
|
self.assertEqual(inst.host, "localhost")
|
|
self.assertEqual(inst.port, 8080)
|
|
self.assertEqual(inst.threads, 5)
|
|
self.assertEqual(inst.trusted_proxy, "192.168.1.1")
|
|
self.assertEqual(inst.trusted_proxy_headers, {"forwarded"})
|
|
self.assertEqual(inst.trusted_proxy_count, 2)
|
|
self.assertEqual(inst.log_untrusted_proxy_headers, True)
|
|
self.assertEqual(inst.url_scheme, "https")
|
|
self.assertEqual(inst.backlog, 20)
|
|
self.assertEqual(inst.recv_bytes, 200)
|
|
self.assertEqual(inst.send_bytes, 300)
|
|
self.assertEqual(inst.outbuf_overflow, 400)
|
|
self.assertEqual(inst.inbuf_overflow, 500)
|
|
self.assertEqual(inst.connection_limit, 1000)
|
|
self.assertEqual(inst.cleanup_interval, 1100)
|
|
self.assertEqual(inst.channel_timeout, 1200)
|
|
self.assertEqual(inst.log_socket_errors, True)
|
|
self.assertEqual(inst.max_request_header_size, 1300)
|
|
self.assertEqual(inst.max_request_body_size, 1400)
|
|
self.assertEqual(inst.expose_tracebacks, True)
|
|
self.assertEqual(inst.asyncore_loop_timeout, 5)
|
|
self.assertEqual(inst.asyncore_use_poll, True)
|
|
self.assertEqual(inst.ident, "abc")
|
|
self.assertEqual(inst.unix_socket_perms, 0o777)
|
|
self.assertEqual(inst.url_prefix, "/foo")
|
|
self.assertEqual(inst.ipv4, True)
|
|
self.assertEqual(inst.ipv6, False)
|
|
|
|
bind_pairs = [
|
|
sockaddr[:2]
|
|
for (family, _, _, sockaddr) in inst.listen
|
|
if family == socket.AF_INET
|
|
]
|
|
|
|
# On Travis, somehow we start listening to two sockets when resolving
|
|
# localhost...
|
|
self.assertEqual(("127.0.0.1", 8080), bind_pairs[0])
|
|
|
|
def test_goodvar_listen(self):
|
|
inst = self._makeOne(listen="127.0.0.1")
|
|
|
|
bind_pairs = [(host, port) for (_, _, _, (host, port)) in inst.listen]
|
|
|
|
self.assertEqual(bind_pairs, [("127.0.0.1", 8080)])
|
|
|
|
def test_default_listen(self):
|
|
inst = self._makeOne()
|
|
|
|
bind_pairs = [(host, port) for (_, _, _, (host, port)) in inst.listen]
|
|
|
|
self.assertEqual(bind_pairs, [("0.0.0.0", 8080)])
|
|
|
|
def test_multiple_listen(self):
|
|
inst = self._makeOne(listen="127.0.0.1:9090 127.0.0.1:8080")
|
|
|
|
bind_pairs = [sockaddr[:2] for (_, _, _, sockaddr) in inst.listen]
|
|
|
|
self.assertEqual(bind_pairs, [("127.0.0.1", 9090), ("127.0.0.1", 8080)])
|
|
|
|
def test_wildcard_listen(self):
|
|
inst = self._makeOne(listen="*:8080")
|
|
|
|
bind_pairs = [sockaddr[:2] for (_, _, _, sockaddr) in inst.listen]
|
|
|
|
self.assertTrue(len(bind_pairs) >= 1)
|
|
|
|
def test_ipv6_no_port(self): # pragma: nocover
|
|
if not self._hasIPv6():
|
|
return
|
|
|
|
inst = self._makeOne(listen="[::1]")
|
|
|
|
bind_pairs = [sockaddr[:2] for (_, _, _, sockaddr) in inst.listen]
|
|
|
|
self.assertEqual(bind_pairs, [("::1", 8080)])
|
|
|
|
def test_bad_port(self):
|
|
self.assertRaises(ValueError, self._makeOne, listen="127.0.0.1:test")
|
|
|
|
def test_service_port(self):
|
|
if WIN and PY2: # pragma: no cover
|
|
# On Windows and Python 2 this is broken, so we raise a ValueError
|
|
self.assertRaises(
|
|
ValueError, self._makeOne, listen="127.0.0.1:http",
|
|
)
|
|
return
|
|
|
|
inst = self._makeOne(listen="127.0.0.1:http 0.0.0.0:https")
|
|
|
|
bind_pairs = [sockaddr[:2] for (_, _, _, sockaddr) in inst.listen]
|
|
|
|
self.assertEqual(bind_pairs, [("127.0.0.1", 80), ("0.0.0.0", 443)])
|
|
|
|
def test_dont_mix_host_port_listen(self):
|
|
self.assertRaises(
|
|
ValueError,
|
|
self._makeOne,
|
|
host="localhost",
|
|
port="8080",
|
|
listen="127.0.0.1:8080",
|
|
)
|
|
|
|
def test_good_sockets(self):
|
|
sockets = [
|
|
socket.socket(socket.AF_INET6, socket.SOCK_STREAM),
|
|
socket.socket(socket.AF_INET, socket.SOCK_STREAM),
|
|
]
|
|
inst = self._makeOne(sockets=sockets)
|
|
self.assertEqual(inst.sockets, sockets)
|
|
sockets[0].close()
|
|
sockets[1].close()
|
|
|
|
def test_dont_mix_sockets_and_listen(self):
|
|
sockets = [socket.socket(socket.AF_INET, socket.SOCK_STREAM)]
|
|
self.assertRaises(
|
|
ValueError, self._makeOne, listen="127.0.0.1:8080", sockets=sockets
|
|
)
|
|
sockets[0].close()
|
|
|
|
def test_dont_mix_sockets_and_host_port(self):
|
|
sockets = [socket.socket(socket.AF_INET, socket.SOCK_STREAM)]
|
|
self.assertRaises(
|
|
ValueError, self._makeOne, host="localhost", port="8080", sockets=sockets
|
|
)
|
|
sockets[0].close()
|
|
|
|
def test_dont_mix_sockets_and_unix_socket(self):
|
|
sockets = [socket.socket(socket.AF_INET, socket.SOCK_STREAM)]
|
|
self.assertRaises(
|
|
ValueError, self._makeOne, unix_socket="./tmp/test", sockets=sockets
|
|
)
|
|
sockets[0].close()
|
|
|
|
def test_dont_mix_unix_socket_and_host_port(self):
|
|
self.assertRaises(
|
|
ValueError,
|
|
self._makeOne,
|
|
unix_socket="./tmp/test",
|
|
host="localhost",
|
|
port="8080",
|
|
)
|
|
|
|
def test_dont_mix_unix_socket_and_listen(self):
|
|
self.assertRaises(
|
|
ValueError, self._makeOne, unix_socket="./tmp/test", listen="127.0.0.1:8080"
|
|
)
|
|
|
|
def test_dont_use_unsupported_socket_types(self):
|
|
sockets = [socket.socket(socket.AF_INET, socket.SOCK_DGRAM)]
|
|
self.assertRaises(ValueError, self._makeOne, sockets=sockets)
|
|
sockets[0].close()
|
|
|
|
def test_dont_mix_forwarded_with_x_forwarded(self):
|
|
with self.assertRaises(ValueError) as cm:
|
|
self._makeOne(
|
|
trusted_proxy="localhost",
|
|
trusted_proxy_headers={"forwarded", "x-forwarded-for"},
|
|
)
|
|
|
|
self.assertIn("The Forwarded proxy header", str(cm.exception))
|
|
|
|
def test_unknown_trusted_proxy_header(self):
|
|
with self.assertRaises(ValueError) as cm:
|
|
self._makeOne(
|
|
trusted_proxy="localhost",
|
|
trusted_proxy_headers={"forwarded", "x-forwarded-unknown"},
|
|
)
|
|
|
|
self.assertIn(
|
|
"unknown trusted_proxy_headers value (x-forwarded-unknown)",
|
|
str(cm.exception),
|
|
)
|
|
|
|
def test_trusted_proxy_count_no_trusted_proxy(self):
|
|
with self.assertRaises(ValueError) as cm:
|
|
self._makeOne(trusted_proxy_count=1)
|
|
|
|
self.assertIn("trusted_proxy_count has no meaning", str(cm.exception))
|
|
|
|
def test_trusted_proxy_headers_no_trusted_proxy(self):
|
|
with self.assertRaises(ValueError) as cm:
|
|
self._makeOne(trusted_proxy_headers={"forwarded"})
|
|
|
|
self.assertIn("trusted_proxy_headers has no meaning", str(cm.exception))
|
|
|
|
def test_trusted_proxy_headers_string_list(self):
|
|
inst = self._makeOne(
|
|
trusted_proxy="localhost",
|
|
trusted_proxy_headers="x-forwarded-for x-forwarded-by",
|
|
)
|
|
self.assertEqual(
|
|
inst.trusted_proxy_headers, {"x-forwarded-for", "x-forwarded-by"}
|
|
)
|
|
|
|
def test_trusted_proxy_headers_string_list_newlines(self):
|
|
inst = self._makeOne(
|
|
trusted_proxy="localhost",
|
|
trusted_proxy_headers="x-forwarded-for\nx-forwarded-by\nx-forwarded-host",
|
|
)
|
|
self.assertEqual(
|
|
inst.trusted_proxy_headers,
|
|
{"x-forwarded-for", "x-forwarded-by", "x-forwarded-host"},
|
|
)
|
|
|
|
def test_no_trusted_proxy_headers_trusted_proxy(self):
|
|
with warnings.catch_warnings(record=True) as w:
|
|
warnings.resetwarnings()
|
|
warnings.simplefilter("always")
|
|
self._makeOne(trusted_proxy="localhost")
|
|
|
|
self.assertGreaterEqual(len(w), 1)
|
|
self.assertTrue(issubclass(w[0].category, DeprecationWarning))
|
|
self.assertIn("Implicitly trusting X-Forwarded-Proto", str(w[0]))
|
|
|
|
def test_clear_untrusted_proxy_headers(self):
|
|
with warnings.catch_warnings(record=True) as w:
|
|
warnings.resetwarnings()
|
|
warnings.simplefilter("always")
|
|
self._makeOne(
|
|
trusted_proxy="localhost", trusted_proxy_headers={"x-forwarded-for"}
|
|
)
|
|
|
|
self.assertGreaterEqual(len(w), 1)
|
|
self.assertTrue(issubclass(w[0].category, DeprecationWarning))
|
|
self.assertIn(
|
|
"clear_untrusted_proxy_headers will be set to True", str(w[0])
|
|
)
|
|
|
|
def test_deprecated_send_bytes(self):
|
|
with warnings.catch_warnings(record=True) as w:
|
|
warnings.resetwarnings()
|
|
warnings.simplefilter("always")
|
|
self._makeOne(send_bytes=1)
|
|
|
|
self.assertGreaterEqual(len(w), 1)
|
|
self.assertTrue(issubclass(w[0].category, DeprecationWarning))
|
|
self.assertIn("send_bytes", str(w[0]))
|
|
|
|
def test_badvar(self):
|
|
self.assertRaises(ValueError, self._makeOne, nope=True)
|
|
|
|
def test_ipv4_disabled(self):
|
|
self.assertRaises(
|
|
ValueError, self._makeOne, ipv4=False, listen="127.0.0.1:8080"
|
|
)
|
|
|
|
def test_ipv6_disabled(self):
|
|
self.assertRaises(ValueError, self._makeOne, ipv6=False, listen="[::]:8080")
|
|
|
|
def test_server_header_removable(self):
|
|
inst = self._makeOne(ident=None)
|
|
self.assertEqual(inst.ident, None)
|
|
|
|
inst = self._makeOne(ident="")
|
|
self.assertEqual(inst.ident, None)
|
|
|
|
inst = self._makeOne(ident="specific_header")
|
|
self.assertEqual(inst.ident, "specific_header")
|
|
|
|
|
|
class TestCLI(unittest.TestCase):
|
|
def parse(self, argv):
|
|
from waitress.adjustments import Adjustments
|
|
|
|
return Adjustments.parse_args(argv)
|
|
|
|
def test_noargs(self):
|
|
opts, args = self.parse([])
|
|
self.assertDictEqual(opts, {"call": False, "help": False})
|
|
self.assertSequenceEqual(args, [])
|
|
|
|
def test_help(self):
|
|
opts, args = self.parse(["--help"])
|
|
self.assertDictEqual(opts, {"call": False, "help": True})
|
|
self.assertSequenceEqual(args, [])
|
|
|
|
def test_call(self):
|
|
opts, args = self.parse(["--call"])
|
|
self.assertDictEqual(opts, {"call": True, "help": False})
|
|
self.assertSequenceEqual(args, [])
|
|
|
|
def test_both(self):
|
|
opts, args = self.parse(["--call", "--help"])
|
|
self.assertDictEqual(opts, {"call": True, "help": True})
|
|
self.assertSequenceEqual(args, [])
|
|
|
|
def test_positive_boolean(self):
|
|
opts, args = self.parse(["--expose-tracebacks"])
|
|
self.assertDictContainsSubset({"expose_tracebacks": "true"}, opts)
|
|
self.assertSequenceEqual(args, [])
|
|
|
|
def test_negative_boolean(self):
|
|
opts, args = self.parse(["--no-expose-tracebacks"])
|
|
self.assertDictContainsSubset({"expose_tracebacks": "false"}, opts)
|
|
self.assertSequenceEqual(args, [])
|
|
|
|
def test_cast_params(self):
|
|
opts, args = self.parse(
|
|
["--host=localhost", "--port=80", "--unix-socket-perms=777"]
|
|
)
|
|
self.assertDictContainsSubset(
|
|
{"host": "localhost", "port": "80", "unix_socket_perms": "777",}, opts
|
|
)
|
|
self.assertSequenceEqual(args, [])
|
|
|
|
def test_listen_params(self):
|
|
opts, args = self.parse(["--listen=test:80",])
|
|
|
|
self.assertDictContainsSubset({"listen": " test:80"}, opts)
|
|
self.assertSequenceEqual(args, [])
|
|
|
|
def test_multiple_listen_params(self):
|
|
opts, args = self.parse(["--listen=test:80", "--listen=test:8080",])
|
|
|
|
self.assertDictContainsSubset({"listen": " test:80 test:8080"}, opts)
|
|
self.assertSequenceEqual(args, [])
|
|
|
|
def test_bad_param(self):
|
|
import getopt
|
|
|
|
self.assertRaises(getopt.GetoptError, self.parse, ["--no-host"])
|
|
|
|
|
|
if hasattr(socket, "AF_UNIX"):
|
|
|
|
class TestUnixSocket(unittest.TestCase):
|
|
def _makeOne(self, **kw):
|
|
from waitress.adjustments import Adjustments
|
|
|
|
return Adjustments(**kw)
|
|
|
|
def test_dont_mix_internet_and_unix_sockets(self):
|
|
sockets = [
|
|
socket.socket(socket.AF_INET, socket.SOCK_STREAM),
|
|
socket.socket(socket.AF_UNIX, socket.SOCK_STREAM),
|
|
]
|
|
self.assertRaises(ValueError, self._makeOne, sockets=sockets)
|
|
sockets[0].close()
|
|
sockets[1].close()
|