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
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import errno
import os
import subprocess
import mock
import testresources
from testtools import matchers
from keystoneclient.common import cms
from keystoneclient import exceptions
from keystoneclient.tests.unit import client_fixtures
from keystoneclient.tests.unit import utils
class CMSTest(utils.TestCase, testresources.ResourcedTestCase):
"""Unit tests for the keystoneclient.common.cms module."""
resources = [('examples', client_fixtures.EXAMPLES_RESOURCE)]
def __init__(self, *args, **kwargs):
super(CMSTest, self).__init__(*args, **kwargs)
process = subprocess.Popen(['openssl', 'version'],
stdout=subprocess.PIPE)
out, err = process.communicate()
# Example output: 'OpenSSL 0.9.8za 5 Jun 2014'
openssl_version = out.split()[1]
if err or openssl_version.startswith(b'0'):
raise Exception('Your version of OpenSSL is not supported. '
'You will need to update it to 1.0 or later.')
def _raise_OSError(*args):
e = OSError()
e.errno = errno.EPIPE
raise e
def test_cms_verify(self):
self.assertRaises(exceptions.CertificateConfigError,
cms.cms_verify,
'data',
'no_exist_cert_file',
'no_exist_ca_file')
def test_token_tocms_to_token(self):
with open(os.path.join(client_fixtures.CMSDIR,
'auth_token_scoped.pem')) as f:
AUTH_TOKEN_SCOPED_CMS = f.read()
self.assertEqual(cms.token_to_cms(self.examples.SIGNED_TOKEN_SCOPED),
AUTH_TOKEN_SCOPED_CMS)
tok = cms.cms_to_token(cms.token_to_cms(
self.examples.SIGNED_TOKEN_SCOPED))
self.assertEqual(tok, self.examples.SIGNED_TOKEN_SCOPED)
def test_asn1_token(self):
self.assertTrue(cms.is_asn1_token(self.examples.SIGNED_TOKEN_SCOPED))
self.assertFalse(cms.is_asn1_token('FOOBAR'))
def test_cms_sign_token_no_files(self):
self.assertRaises(subprocess.CalledProcessError,
cms.cms_sign_token,
self.examples.TOKEN_SCOPED_DATA,
'/no/such/file', '/no/such/key')
def test_cms_sign_token_no_files_pkiz(self):
self.assertRaises(subprocess.CalledProcessError,
cms.pkiz_sign,
self.examples.TOKEN_SCOPED_DATA,
'/no/such/file', '/no/such/key')
def test_cms_sign_token_success(self):
self.assertTrue(
cms.pkiz_sign(self.examples.TOKEN_SCOPED_DATA,
self.examples.SIGNING_CERT_FILE,
self.examples.SIGNING_KEY_FILE))
def test_cms_verify_token_no_files(self):
self.assertRaises(exceptions.CertificateConfigError,
cms.cms_verify,
self.examples.SIGNED_TOKEN_SCOPED,
'/no/such/file', '/no/such/key')
def test_cms_verify_token_no_oserror(self):
with mock.patch('subprocess.Popen.communicate',
new=self._raise_OSError):
try:
cms.cms_verify("x", '/no/such/file', '/no/such/key')
except exceptions.CertificateConfigError as e:
self.assertIn('/no/such/file', e.output)
self.assertIn('Hit OSError ', e.output)
else:
self.fail('Expected exceptions.CertificateConfigError')
def test_cms_verify_token_scoped(self):
cms_content = cms.token_to_cms(self.examples.SIGNED_TOKEN_SCOPED)
self.assertTrue(cms.cms_verify(cms_content,
self.examples.SIGNING_CERT_FILE,
self.examples.SIGNING_CA_FILE))
def test_cms_verify_token_scoped_expired(self):
cms_content = cms.token_to_cms(
self.examples.SIGNED_TOKEN_SCOPED_EXPIRED)
self.assertTrue(cms.cms_verify(cms_content,
self.examples.SIGNING_CERT_FILE,
self.examples.SIGNING_CA_FILE))
def test_cms_verify_token_unscoped(self):
cms_content = cms.token_to_cms(self.examples.SIGNED_TOKEN_UNSCOPED)
self.assertTrue(cms.cms_verify(cms_content,
self.examples.SIGNING_CERT_FILE,
self.examples.SIGNING_CA_FILE))
def test_cms_verify_token_v3_scoped(self):
cms_content = cms.token_to_cms(self.examples.SIGNED_v3_TOKEN_SCOPED)
self.assertTrue(cms.cms_verify(cms_content,
self.examples.SIGNING_CERT_FILE,
self.examples.SIGNING_CA_FILE))
def test_cms_hash_token_no_token_id(self):
token_id = None
self.assertThat(cms.cms_hash_token(token_id), matchers.Is(None))
def test_cms_hash_token_not_pki(self):
"""If the token_id is not a PKI token then it returns the token_id."""
token = 'something'
self.assertFalse(cms.is_asn1_token(token))
self.assertThat(cms.cms_hash_token(token), matchers.Is(token))
def test_cms_hash_token_default_md5(self):
"""The default hash method is md5."""
token = self.examples.SIGNED_TOKEN_SCOPED
token_id_default = cms.cms_hash_token(token)
token_id_md5 = cms.cms_hash_token(token, mode='md5')
self.assertThat(token_id_default, matchers.Equals(token_id_md5))
# md5 hash is 32 chars.
self.assertThat(token_id_default, matchers.HasLength(32))
def test_cms_hash_token_sha256(self):
"""Can also hash with sha256."""
token = self.examples.SIGNED_TOKEN_SCOPED
token_id = cms.cms_hash_token(token, mode='sha256')
# sha256 hash is 64 chars.
self.assertThat(token_id, matchers.HasLength(64))
@mock.patch('keystoneclient.common.cms._check_files_accessible')
def test_process_communicate_handle_oserror_epipe(self, files_acc_mock):
process_mock = mock.Mock()
process_mock.communicate = self._raise_OSError
process_mock.stderr = mock.Mock()
process_mock.stderr.read = mock.Mock(return_value='proc stderr')
files_acc_mock.return_value = 1, ('file_path', 'fileerror')
output, err, retcode = cms._process_communicate_handle_oserror(
process_mock, '', [])
self.assertEqual((output, retcode), ('', 1))
self.assertIn('file_path', err)
self.assertIn('fileerror', err)
self.assertIn('proc stderr', err)
@mock.patch('keystoneclient.common.cms._check_files_accessible')
def test_process_communicate_handle_oserror_epipe_files_ok(
self, files_acc_mock):
process_mock = mock.Mock()
process_mock.communicate = self._raise_OSError
process_mock.stderr = mock.Mock()
process_mock.stderr.read = mock.Mock(return_value='proc stderr')
files_acc_mock.return_value = -1, None
output, err, retcode = cms._process_communicate_handle_oserror(
process_mock, '', [])
self.assertEqual((output, retcode), ('', -1))
self.assertIn('proc stderr', err)
def test_process_communicate_handle_oserror_no_exception(self):
process_mock = mock.Mock()
process_mock.communicate.return_value = 'out', 'err'
process_mock.poll.return_value = 0
output, err, retcode = cms._process_communicate_handle_oserror(
process_mock, '', [])
self.assertEqual(output, 'out')
self.assertEqual(err, 'err')
self.assertEqual(retcode, 0)
def load_tests(loader, tests, pattern):
return testresources.OptimisingTestSuite(tests)
|
