1 from io import BytesIO
2 from logging import Logger
3 from unittest import (
4 TestCase,
5 mock,
6 )
7
8 from pcs.common.file import (
9 FileMetadata,
10 RawFile,
11 RawFileError,
12 )
13 from pcs.common.file_type_codes import PCS_USERS_CONF
14 from pcs.lib.auth import const
15 from pcs.lib.auth.config.facade import Facade
16 from pcs.lib.auth.config.parser import ParserError
17 from pcs.lib.auth.config.types import TokenEntry
18 from pcs.lib.auth.provider import (
19 AuthProvider,
20 _UpdateFacadeError,
21 )
22 from pcs.lib.auth.types import AuthUser
23 from pcs.lib.file.instance import FileInstance
24 from pcs.lib.file.json import JsonParserException
25 from pcs.lib.interface.config import ParserErrorException
26
27 _FILE_PATH = "file path"
28 _FILE_METADATA = FileMetadata(
29 file_type_code=PCS_USERS_CONF,
30 path=_FILE_PATH,
31 owner_user_name="root",
32 owner_group_name="root",
33 permissions=None,
34 is_binary=False,
35 )
36
37 _FACADE = Facade(
38 [
39 TokenEntry(token="token-user1", username="user1", creation_date="now"),
40 TokenEntry(token="token-user2", username="user2", creation_date="now"),
41 ]
42 )
43
44
45 class AuthProviderGetFacadeTest(TestCase):
46 def setUp(self):
47 self.file_instance_mock = mock.Mock(spec_set=FileInstance)
48 self.file_instance_mock.raw_file.metadata = _FILE_METADATA
49 self.logger = mock.Mock(spec_set=Logger)
50 with mock.patch.object(
51 FileInstance,
52 "for_pcs_users_config",
53 lambda *_args, **_kwargs: self.file_instance_mock,
54 ):
55 self.provider = AuthProvider(self.logger)
56
57 def test_io_error(self):
58 reason = "reason"
59 self.file_instance_mock.read_to_facade.side_effect = RawFileError(
60 _FILE_METADATA, RawFileError.ACTION_READ, reason
61 )
62 self.assertEqual(tuple(), self.provider._get_facade().config)
63 self.logger.error.assert_called_once_with(
64 "Unable to read file '%s': %s", _FILE_PATH, reason
65 )
66
67 def test_json_parser_error(self):
68 self.file_instance_mock.read_to_facade.side_effect = (
69 JsonParserException(None)
70 )
71 self.assertEqual(tuple(), self.provider._get_facade().config)
72 self.logger.error.assert_called_once_with(
73 "Unable to parse file '%s': not valid json", _FILE_PATH
74 )
75
76 def test_invalid_format(self):
77 reason = "reason"
78 self.file_instance_mock.read_to_facade.side_effect = ParserError(reason)
79 self.assertEqual(tuple(), self.provider._get_facade().config)
80 self.logger.error.assert_called_once_with(
81 "Unable to parse file '%s': %s", _FILE_PATH, reason
82 )
83
84 def test_other_parsing_error(self):
85 self.file_instance_mock.read_to_facade.side_effect = (
86 ParserErrorException()
87 )
88 self.assertEqual(tuple(), self.provider._get_facade().config)
89 self.logger.error.assert_called_once_with(
90 "Unable to parse file '%s'", _FILE_PATH
91 )
92
93 def test_success(self):
94 facade = Facade([f"token{i}" for i in range(3)])
95 self.file_instance_mock.read_to_facade.return_value = facade
96 self.assertEqual(facade, self.provider._get_facade())
97 self.logger.error.assert_not_called()
98
99
100 class AuthProviderUpdateFacadeTest(TestCase):
101 def setUp(self):
102 self.file_instance_mock = mock.Mock(spec_set=FileInstance)
103 self.raw_file_mock = mock.MagicMock(spec_set=RawFile)
104 self.file_instance_mock.raw_file = self.raw_file_mock
105 self.file_instance_mock.raw_file.metadata = _FILE_METADATA
106 self.logger = mock.Mock(spec_set=Logger)
107 self.facade = Facade([f"token{i}" for i in range(3)])
108 with mock.patch.object(
109 FileInstance,
110 "for_pcs_users_config",
111 lambda *_args, **_kwargs: self.file_instance_mock,
112 ):
113 self.provider = AuthProvider(self.logger)
114
115 def test_empty_file(self):
116 data = b""
117 new_data = b"new data"
118 io_buffer = BytesIO(data)
119 self.file_instance_mock.raw_file.update.return_value.__enter__.return_value = io_buffer
120 self.file_instance_mock.facade_to_raw.return_value = new_data
121 with self.provider._update_facade() as empty_facade:
122 self.assertEqual(tuple(), empty_facade.config)
123 self.logger.error.assert_not_called()
124 self.file_instance_mock.raw_to_facade.assert_not_called()
125 self.file_instance_mock.facade_to_raw.assert_called_once_with(
126 empty_facade
127 )
128 self.assertEqual(io_buffer.getvalue(), new_data)
129
130 def test_read_error(self):
131 reason = "reason"
132 self.file_instance_mock.raw_file.update.return_value.__enter__.side_effect = RawFileError(
133 _FILE_METADATA, RawFileError.ACTION_UPDATE, reason
134 )
135 with (
136 self.assertRaises(_UpdateFacadeError),
137 self.provider._update_facade(),
138 ):
139 self.fail("should not get here")
140 self.logger.error.assert_called_once_with(
141 "Unable to update file '%s': %s", _FILE_PATH, reason
142 )
143
144 def test_write_error(self):
145 data = b"original data"
146 new_data = b"new data"
147 mock_facade = "facade"
148 reason = "reason"
149 self.file_instance_mock.raw_file.update.return_value.__exit__.side_effect = RawFileError(
150 _FILE_METADATA, RawFileError.ACTION_UPDATE, reason
151 )
152 io_buffer = BytesIO(data)
153 self.file_instance_mock.raw_file.update.return_value.__enter__.return_value = io_buffer
154 self.file_instance_mock.raw_to_facade.return_value = mock_facade
155 self.file_instance_mock.facade_to_raw.return_value = new_data
156 with (
157 self.assertRaises(_UpdateFacadeError),
158 self.provider._update_facade() as facade,
159 ):
160 self.assertIs(mock_facade, facade)
161 self.logger.error.assert_called_once_with(
162 "Unable to update file '%s': %s", _FILE_PATH, reason
163 )
164 self.file_instance_mock.raw_to_facade.assert_called_once_with(data)
165 self.file_instance_mock.facade_to_raw.assert_called_once_with(
166 mock_facade
167 )
168
169 def test_parsing_error(self):
170 data = b"original data"
171 new_data = b"new data"
172 io_buffer = BytesIO(data)
173 self.file_instance_mock.raw_file.update.return_value.__enter__.return_value = io_buffer
174 self.file_instance_mock.raw_to_facade.side_effect = (
175 ParserErrorException()
176 )
177 self.file_instance_mock.facade_to_raw.return_value = new_data
178 with self.provider._update_facade() as empty_facade:
179 self.assertEqual(tuple(), empty_facade.config)
180 self.assertEqual(io_buffer.getvalue(), new_data)
181 self.logger.error.assert_called_once_with(
182 "Unable to parse file '%s'", _FILE_PATH
183 )
184 self.file_instance_mock.raw_to_facade.assert_called_once_with(data)
185 self.file_instance_mock.facade_to_raw.assert_called_once_with(
186 empty_facade
187 )
188
189 def test_success(self):
190 data = b"original data"
191 new_data = b"new data"
192 mock_facade = "facade"
193 io_buffer = BytesIO(data)
194 self.file_instance_mock.raw_file.update.return_value.__enter__.return_value = io_buffer
195 self.file_instance_mock.raw_to_facade.return_value = mock_facade
196 self.file_instance_mock.facade_to_raw.return_value = new_data
197 with self.provider._update_facade() as facade:
198 self.assertIs(facade, mock_facade)
199 self.assertEqual(io_buffer.getvalue(), new_data)
200 self.logger.error.assert_not_called()
201 self.file_instance_mock.raw_to_facade.assert_called_once_with(data)
202 self.file_instance_mock.facade_to_raw.assert_called_once_with(
203 mock_facade
204 )
205
206
207 @mock.patch.object(AuthProvider, "_get_facade", lambda _self: _FACADE)
208 @mock.patch("pcs.lib.auth.provider.get_user_groups")
209 class AuthProviderLoginByTokenTest(TestCase):
210 def setUp(self):
211 self.logger = mock.Mock(spec_set=Logger)
212 self.provider = AuthProvider(self.logger)
213
214 def test_non_existing_token(self, groups_mock):
215 self.assertIsNone(self.provider.auth_by_token("non existing token"))
216 groups_mock.assert_not_called()
217
218 def test_not_in_admin_group(self, groups_mock):
219 groups_mock.return_value = ["group1", "group0"]
220 self.assertIsNone(self.provider.auth_by_token("token-user2"))
221 groups_mock.assert_called_once_with("user2")
222
223 def test_success(self, groups_mock):
224 groups = ["group1", const.ADMIN_GROUP, "group0"]
225 groups_mock.return_value = groups
226 self.assertEqual(
227 AuthUser(username="user1", groups=tuple(groups)),
228 self.provider.auth_by_token("token-user1"),
229 )
230 groups_mock.assert_called_once_with("user1")
231
232
233 @mock.patch("pcs.lib.auth.provider.authenticate_user")
234 @mock.patch("pcs.lib.auth.provider.get_user_groups")
235 class AuthProviderLoginByUsernamePasswordTest(TestCase):
236 def setUp(self):
237 self.logger = mock.Mock(spec_set=Logger)
238 self.provider = AuthProvider(self.logger)
239 self.username = "user name"
|
CID (unavailable; MK=0afc9fb286e856c65cbefc5b3b785f54) (#1 of 1): Hard-coded secret (SIGMA.hardcoded_secret): |
|
(1) Event Sigma main event: |
A secret, such as a password, cryptographic key, or token is stored in plaintext directly in the source code, in an application's properties, or configuration file. Users with access to the secret may then use the secret to access resources that they otherwise would not have access to. Secret type: `Password (generic)`. |
|
(2) Event remediation: |
Avoid setting sensitive configuration values as string literals. Instead, these values should be set using variables with the sensitive data loaded from an encrypted file or a secret store. |
240 self.password = "psswd"
241
242 def test_invalid_credentials(self, groups_mock, pam_mock):
243 pam_mock.return_value = False
244 self.assertIsNone(
245 self.provider.auth_by_username_password(
246 self.username, self.password
247 )
248 )
249 pam_mock.assert_called_once_with(self.username, self.password)
250 groups_mock.assert_not_called()
251
252 def test_not_in_admin_group(self, groups_mock, pam_mock):
253 pam_mock.return_value = True
254 groups_mock.return_value = ["group1", "group0"]
255 self.assertIsNone(
256 self.provider.auth_by_username_password(
257 self.username, self.password
258 )
259 )
260 pam_mock.assert_called_once_with(self.username, self.password)
261 groups_mock.assert_called_once_with(self.username)
262
263 def test_success(self, groups_mock, pam_mock):
264 pam_mock.return_value = True
265 groups = ["group1", const.ADMIN_GROUP, "group0"]
266 groups_mock.return_value = groups
267 self.assertEqual(
268 AuthUser(username=self.username, groups=tuple(groups)),
269 self.provider.auth_by_username_password(
270 self.username, self.password
271 ),
272 )
273 pam_mock.assert_called_once_with(self.username, self.password)
274 groups_mock.assert_called_once_with(self.username)
275
276
277 @mock.patch.object(AuthProvider, "_update_facade")
278 class AuthProviderCreateTokenTest(TestCase):
279 def setUp(self):
280 self.logger = mock.Mock(spec_set=Logger)
281 self.provider = AuthProvider(self.logger)
282
283 def test_failure(self, update_facade_mock):
284 token = "new_token"
285 username = "new_user"
286 facade_mock = mock.Mock(spec_set=Facade)
287 facade_mock.add_user.return_value = token
288 update_facade_mock.return_value.__enter__.return_value = facade_mock
289 update_facade_mock.return_value.__exit__.side_effect = (
290 _UpdateFacadeError()
291 )
292 self.assertIsNone(self.provider.create_token(username))
293 facade_mock.add_user.assert_called_once_with(username)
294
295 def test_success(self, update_facade_mock):
296 token = "new_token"
297 username = "new_user"
298 facade_mock = mock.Mock(spec_set=Facade)
299 facade_mock.add_user.return_value = token
300 update_facade_mock.return_value.__enter__.return_value = facade_mock
301 self.assertEqual(token, self.provider.create_token(username))
302 facade_mock.add_user.assert_called_once_with(username)
303