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
|
import os
from urllib.parse import urlparse
import pytest
import redis
from redis.exceptions import ConnectionError
@pytest.mark.ssl
class TestSSL:
"""Tests for SSL connections
This relies on the --redis-ssl-url purely for rebuilding the client
and connecting to the appropriate port.
"""
ROOT = os.path.join(os.path.dirname(__file__), "..")
CERT_DIR = os.path.abspath(os.path.join(ROOT, "docker", "stunnel", "keys"))
if not os.path.isdir(CERT_DIR): # github actions package validation case
CERT_DIR = os.path.abspath(
os.path.join(ROOT, "..", "docker", "stunnel", "keys")
)
if not os.path.isdir(CERT_DIR):
raise IOError(f"No SSL certificates found. They should be in {CERT_DIR}")
def test_ssl_with_invalid_cert(self, request):
ssl_url = request.config.option.redis_ssl_url
sslclient = redis.from_url(ssl_url)
with pytest.raises(ConnectionError) as e:
sslclient.ping()
assert "SSL: CERTIFICATE_VERIFY_FAILED" in str(e)
def test_ssl_connection(self, request):
ssl_url = request.config.option.redis_ssl_url
p = urlparse(ssl_url)[1].split(":")
r = redis.Redis(host=p[0], port=p[1], ssl=True, ssl_cert_reqs="none")
assert r.ping()
def test_ssl_connection_without_ssl(self, request):
ssl_url = request.config.option.redis_ssl_url
p = urlparse(ssl_url)[1].split(":")
r = redis.Redis(host=p[0], port=p[1], ssl=False)
with pytest.raises(ConnectionError) as e:
r.ping()
assert "Connection closed by server" in str(e)
def test_validating_self_signed_certificate(self, request):
ssl_url = request.config.option.redis_ssl_url
p = urlparse(ssl_url)[1].split(":")
r = redis.Redis(
host=p[0],
port=p[1],
ssl=True,
ssl_certfile=os.path.join(self.CERT_DIR, "server-cert.pem"),
ssl_keyfile=os.path.join(self.CERT_DIR, "server-key.pem"),
ssl_cert_reqs="required",
ssl_ca_certs=os.path.join(self.CERT_DIR, "server-cert.pem"),
)
assert r.ping()
|
