These tests cause router crash when run with AddressSanitizer:
=================================================================
==77196==ERROR: AddressSanitizer: heap-use-after-free on address 0x60c000079340 at pc 0x55d56b132d4b bp 0x7f8cc7f346b0 sp 0x7f8cc7f346a0
READ of size 1 at 0x60c000079340 thread T1
#0 0x55d56b132d4a in nxt_openssl_conn_io_shutdown src/nxt_openssl.c:1466
#1 0x55d56b0f6a25 in nxt_h1p_closing src/nxt_h1proto.c:2069
#2 0x55d56b1009a6 in nxt_h1p_shutdown src/nxt_h1proto.c:2038
#3 0x55d56b1014c3 in nxt_h1p_request_close src/nxt_h1proto.c:1718
#4 0x55d56b1045c0 in nxt_http_request_close_handler src/nxt_http_request.c:864
#5 0x55d56b104988 in nxt_http_request_done src/nxt_http_request.c:795
#6 0x55d56b0ba0c3 in nxt_event_engine_start src/nxt_event_engine.c:542
#7 0x55d56b0dcac2 in nxt_router_thread_start src/nxt_router.c:3645
#8 0x55d56b0b421b in nxt_thread_trampoline src/nxt_thread.c:126
#9 0x7f8ccab95ac2 (/lib/x86_64-linux-gnu/libc.so.6+0x94ac2)
#10 0x7f8ccac2784f (/lib/x86_64-linux-gnu/libc.so.6+0x12684f)
123 lines
2.8 KiB
Python
123 lines
2.8 KiB
Python
import socket
|
|
import ssl
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from unit.applications.tls import ApplicationTLS
|
|
from unit.option import option
|
|
|
|
prerequisites = {'modules': {'openssl': 'any'}}
|
|
|
|
client = ApplicationTLS()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def setup_method_fixture():
|
|
if 'HAS_TLSv1_2' not in dir(ssl) or not ssl.HAS_TLSv1_2:
|
|
pytest.skip('OpenSSL too old')
|
|
|
|
client.certificate()
|
|
|
|
assert 'success' in client.conf(
|
|
{
|
|
"listeners": {
|
|
"*:8080": {
|
|
"pass": "routes",
|
|
"tls": {"certificate": "default"},
|
|
}
|
|
},
|
|
"routes": [{"action": {"return": 200}}],
|
|
"applications": {},
|
|
}
|
|
), 'load application configuration'
|
|
|
|
|
|
def create_socket():
|
|
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
|
|
ctx.check_hostname = False
|
|
ctx.verify_mode = ssl.CERT_NONE
|
|
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
ssl_sock = ctx.wrap_socket(
|
|
s, server_hostname='localhost', do_handshake_on_connect=False
|
|
)
|
|
ssl_sock.connect(('127.0.0.1', 8080))
|
|
|
|
return ssl_sock
|
|
|
|
|
|
def clear_conf():
|
|
assert 'success' in client.conf({"listeners": {}, "applications": {}})
|
|
|
|
|
|
@pytest.mark.skip('not yet')
|
|
def test_reconfigure_tls_switch():
|
|
assert 'success' in client.conf_delete('listeners/*:8080/tls')
|
|
|
|
(_, sock) = client.get(
|
|
headers={'Host': 'localhost', 'Connection': 'keep-alive'},
|
|
start=True,
|
|
read_timeout=1,
|
|
)
|
|
|
|
assert 'success' in client.conf(
|
|
{"pass": "routes", "tls": {"certificate": "default"}},
|
|
'listeners/*:8080',
|
|
)
|
|
|
|
assert client.get(sock=sock)['status'] == 200, 'reconfigure'
|
|
assert client.get_ssl()['status'] == 200, 'reconfigure tls'
|
|
|
|
|
|
def test_reconfigure_tls():
|
|
if option.configure_flag['asan']:
|
|
pytest.skip('not yet, router crash')
|
|
|
|
ssl_sock = create_socket()
|
|
|
|
ssl_sock.sendall("""GET / HTTP/1.1\r\n""".encode())
|
|
|
|
clear_conf()
|
|
|
|
ssl_sock.sendall(
|
|
"""Host: localhost\r\nConnection: close\r\n\r\n""".encode()
|
|
)
|
|
|
|
assert (
|
|
client.recvall(ssl_sock).decode().startswith('HTTP/1.1 200 OK')
|
|
), 'finish request'
|
|
|
|
|
|
def test_reconfigure_tls_2():
|
|
ssl_sock = create_socket()
|
|
|
|
# Waiting for connection completion.
|
|
# Delay should be more than TCP_DEFER_ACCEPT.
|
|
time.sleep(1.5)
|
|
|
|
clear_conf()
|
|
|
|
success = False
|
|
|
|
try:
|
|
ssl_sock.do_handshake()
|
|
except ssl.SSLError:
|
|
ssl_sock.close()
|
|
success = True
|
|
|
|
if not success:
|
|
pytest.fail('Connection is not closed.')
|
|
|
|
|
|
def test_reconfigure_tls_3():
|
|
if option.configure_flag['asan']:
|
|
pytest.skip('not yet, router crash')
|
|
|
|
ssl_sock = create_socket()
|
|
ssl_sock.do_handshake()
|
|
|
|
clear_conf()
|
|
|
|
assert client.get(sock=ssl_sock)['status'] == 408, 'request timeout'
|