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
|
from unittest import TestCase
from hvac import exceptions
from parameterized import parameterized
from tests import utils
from tests.utils.hvac_integration_test_case import HvacIntegrationTestCase
class TestCert(HvacIntegrationTestCase, TestCase):
TEST_MOUNT_POINT = "cert-test"
TEST_ROLE_NAME = "testrole"
TEST_CLIENT_CERTIFICATE_FILE = utils.get_config_file_path("client-cert.pem")
cert = utils.create_client(url="fake")._adapter._kwargs.get("cert")
with open(TEST_CLIENT_CERTIFICATE_FILE, "r") as fp:
TEST_CERTIFICATE = fp.read()
def setUp(self):
super().setUp()
if "%s/" % self.TEST_MOUNT_POINT not in self.client.sys.list_auth_methods():
self.client.sys.enable_auth_method(
method_type="cert",
path=self.TEST_MOUNT_POINT,
)
_ = self.client.auth.cert.create_ca_certificate_role(
name=self.TEST_ROLE_NAME,
certificate=self.TEST_CERTIFICATE,
mount_point=self.TEST_MOUNT_POINT,
)
def tearDown(self):
super().tearDown()
def test_create_ca_certificate_role(self):
response = self.client.auth.cert.create_ca_certificate_role(
name="testrole2",
certificate=self.TEST_CERTIFICATE,
mount_point=self.TEST_MOUNT_POINT,
)
self.assertEqual(first=204, second=response.status_code)
def test_create_ca_certificate_with_filename(self):
response = self.client.auth.cert.create_ca_certificate_role(
name="testrole2",
certificate_file=self.TEST_CLIENT_CERTIFICATE_FILE,
mount_point=self.TEST_MOUNT_POINT,
)
self.assertEqual(first=204, second=response.status_code)
def test_create_ca_certificate_with_filename_deprecated(self):
"""This tests the deprecated feature of passing a certificate file via the certificate argument"""
response = self.client.auth.cert.create_ca_certificate_role(
name="testrole2",
certificate=self.TEST_CLIENT_CERTIFICATE_FILE,
mount_point=self.TEST_MOUNT_POINT,
)
self.assertEqual(first=204, second=response.status_code)
def test_read_ca_certificate_role(self):
response = self.client.auth.cert.read_ca_certificate_role(
name=self.TEST_ROLE_NAME,
mount_point=self.TEST_MOUNT_POINT,
)
self.assertEqual(
first=self.TEST_ROLE_NAME,
second=response["data"]["display_name"],
)
def test_list_certificate_roles(self):
response = self.client.auth.cert.list_certificate_roles(
mount_point=self.TEST_MOUNT_POINT,
)
self.assertEqual(first=response["data"]["keys"], second=[self.TEST_ROLE_NAME])
def test_delete_certificate_role(self):
self.client.auth.cert.create_ca_certificate_role(
name="testrole2",
certificate=self.TEST_CERTIFICATE,
mount_point=self.TEST_MOUNT_POINT,
)
response = self.client.auth.cert.delete_certificate_role(
name="testrole2",
mount_point=self.TEST_MOUNT_POINT,
)
self.assertEqual(first=204, second=response.status_code)
def test_configure_tls_certificate(self):
response = self.client.auth.cert.configure_tls_certificate(
mount_point=self.TEST_MOUNT_POINT
)
self.assertEqual(first=204, second=response.status_code)
@parameterized.expand(
[
(TEST_ROLE_NAME, "", cert[0], cert[1], TEST_MOUNT_POINT),
("", "", cert[0], cert[1], TEST_MOUNT_POINT),
("testrole2", "", cert[0], cert[1], TEST_MOUNT_POINT),
("", "", "bad cert", cert[1], TEST_MOUNT_POINT),
("", "bad ca", cert[0], cert[1], TEST_MOUNT_POINT),
("", True, cert[0], cert[1], TEST_MOUNT_POINT),
("", False, " ", " ", TEST_MOUNT_POINT),
]
)
def test_login(self, name, cacert, cert_pem, key_pem, mount_point):
if cacert or "bad" in [cacert, cert_pem, key_pem]:
with self.assertRaises(exceptions.ParamValidationError):
self.client.auth.cert.login(
name=name,
cacert=cacert,
cert_pem=cert_pem,
mount_point=mount_point,
)
elif (
name != ""
and name
not in self.client.auth.cert.list_certificate_roles(
mount_point=self.TEST_MOUNT_POINT,
)["data"]["keys"]
):
with self.assertRaises(exceptions.InvalidRequest):
with self.assertRaises(OSError):
self.client.auth.cert.login(
name=name,
cacert=cacert,
cert_pem=cert_pem,
mount_point=mount_point,
)
elif "/" not in cert_pem:
with self.assertRaises(OSError):
self.client.auth.cert.login(
name=name,
cacert=cacert,
cert_pem=cert_pem,
mount_point=mount_point,
)
else:
response = self.client.auth.cert.login(
name=name,
cacert=cacert,
cert_pem=cert_pem,
mount_point=mount_point,
)
if name in [self.TEST_ROLE_NAME, ""] and (cacert, cert_pem, key_pem) == (
"",
self.cert[0],
self.cert[1],
):
self.assertIsInstance(response, dict)
class TestCertEnv(TestCert):
use_env = True
|
