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
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
|
require_relative 'test_helper'
class KubeclientRealClusterTest < Minitest::Test
# Tests here actually connect to a cluster!
# For simplicity, these tests use same config/*.kubeconfig files as test_config.rb,
# so are intended to run from config/update_certs_k0s.rb script.
def setup
if ENV['KUBECLIENT_TEST_REAL_CLUSTER'] == 'true'
WebMock.enable_net_connect!
else
skip('Requires real cluster, see test/config/update_certs_k0s.rb.')
end
end
def teardown
WebMock.disable_net_connect! # Don't allow any connections in other tests.
end
# Partially isolated tests that check Client behavior with given `verify_ssl` value:
# localhost and 127.0.0.1 are among names on the certificate
HOSTNAME_COVERED_BY_CERT = 'https://127.0.0.1:6443'.freeze
# 127.0.0.2 also means localhost but is not included in the certificate.
HOSTNAME_NOT_ON_CERT = 'https://127.0.0.2:6443'.freeze
def test_real_cluster_verify_peer
config = Kubeclient::Config.read(config_file('external.kubeconfig'))
context = config.context
client1 = Kubeclient::Client.new(
HOSTNAME_COVERED_BY_CERT, 'v1',
ssl_options: context.ssl_options.merge(verify_ssl: OpenSSL::SSL::VERIFY_PEER),
auth_options: context.auth_options
)
check_cert_accepted(client1)
client2 = Kubeclient::Client.new(
HOSTNAME_NOT_ON_CERT, 'v1',
ssl_options: context.ssl_options.merge(verify_ssl: OpenSSL::SSL::VERIFY_PEER),
auth_options: context.auth_options
)
check_cert_rejected(client2)
end
def test_real_cluster_verify_none
config = Kubeclient::Config.read(config_file('external.kubeconfig'))
context = config.context
client1 = Kubeclient::Client.new(
HOSTNAME_COVERED_BY_CERT, 'v1',
ssl_options: context.ssl_options.merge(verify_ssl: OpenSSL::SSL::VERIFY_NONE),
auth_options: context.auth_options
)
check_cert_accepted(client1)
client2 = Kubeclient::Client.new(
HOSTNAME_NOT_ON_CERT, 'v1',
ssl_options: context.ssl_options.merge(verify_ssl: OpenSSL::SSL::VERIFY_NONE),
auth_options: context.auth_options
)
check_cert_accepted(client2)
end
# Integration tests that check combined Config -> Client behavior wrt. `verify_ssl`.
# Quite redundant, but this was an embarrasing vulnerability so want to confirm...
def test_real_cluster_concatenated_ca
config = Kubeclient::Config.read(config_file('concatenated-ca.kubeconfig'))
context = config.context
client1 = Kubeclient::Client.new(
HOSTNAME_COVERED_BY_CERT, 'v1',
ssl_options: context.ssl_options, auth_options: context.auth_options
)
check_cert_accepted(client1)
client2 = Kubeclient::Client.new(
HOSTNAME_NOT_ON_CERT, 'v1',
ssl_options: context.ssl_options, auth_options: context.auth_options
)
check_cert_rejected(client2)
end
def test_real_cluster_verify_ssl_with_ca
config = Kubeclient::Config.read(config_file('external.kubeconfig'))
context = config.context
client1 = Kubeclient::Client.new(
HOSTNAME_COVERED_BY_CERT, 'v1',
ssl_options: context.ssl_options, auth_options: context.auth_options
)
check_cert_accepted(client1)
client2 = Kubeclient::Client.new(
HOSTNAME_NOT_ON_CERT, 'v1',
ssl_options: context.ssl_options, auth_options: context.auth_options
)
check_cert_rejected(client2)
end
def test_real_cluster_verify_ssl_without_ca
config = Kubeclient::Config.read(config_file('external-without-ca.kubeconfig'))
context = config.context
# Hostname matches cert but the local cluster uses self-signed certs from custom CA,
# and this config omits CA data, so verification can't succeed.
client1 = Kubeclient::Client.new(
HOSTNAME_COVERED_BY_CERT, 'v1',
ssl_options: context.ssl_options, auth_options: context.auth_options
)
check_cert_rejected(client1)
client2 = Kubeclient::Client.new(
HOSTNAME_NOT_ON_CERT, 'v1',
ssl_options: context.ssl_options, auth_options: context.auth_options
)
check_cert_rejected(client2)
end
def test_real_cluster_insecure_without_ca
config = Kubeclient::Config.read(config_file('insecure.kubeconfig'))
context = config.context
# Hostname matches cert but the local cluster uses self-signed certs from custom CA,
# and this config omits CA data, so verification would fail;
# however, this config specifies `insecure-skip-tls-verify: true` so any cert goes.
client1 = Kubeclient::Client.new(
HOSTNAME_COVERED_BY_CERT, 'v1',
ssl_options: context.ssl_options, auth_options: context.auth_options
)
check_cert_accepted(client1)
client2 = Kubeclient::Client.new(
HOSTNAME_NOT_ON_CERT, 'v1',
ssl_options: context.ssl_options, auth_options: context.auth_options
)
check_cert_accepted(client2)
end
private
# Test cert checking on discovery, CRUD, and watch code paths.
def check_cert_accepted(client)
client.discover
client.get_nodes
exercise_watcher_with_timeout(client.watch_nodes)
end
def check_cert_rejected(client)
# TODO: all OpenSSL exceptions should be wrapped with Kubeclient error.
assert_raises(Kubeclient::HttpError, OpenSSL::SSL::SSLError) do
client.discover
end
# Since discovery fails, methods like .get_nodes, .watch_nodes would all fail
# on method_missing -> discover. Call lower-level methods to test actual connection.
assert_raises(Kubeclient::HttpError, OpenSSL::SSL::SSLError) do
client.get_entities('Node', 'nodes', {})
end
assert_raises(Kubeclient::HttpError, OpenSSL::SSL::SSLError) do
exercise_watcher_with_timeout(client.watch_entities('nodes'))
end
end
def exercise_watcher_with_timeout(watcher)
thread = Thread.new do
sleep(1)
watcher.finish
end
watcher.each do |_notice|
break
end
thread.join
end
end
|
