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