1 from pprint import pformat
2 from urllib.parse import urlencode
3
4 from tornado.httputil import HTTPHeaders
5 from tornado.testing import AsyncHTTPTestCase
6 from tornado.web import Application
7
8 from pcs.daemon import ruby_pcsd
9
10 USER = "user"
11 GROUPS = ["group1", "group2"]
|
CID (unavailable; MK=ef043fa7e7667de104d9b89956d05b99) (#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. |
12 PASSWORD = "password"
13
14
15 class RubyPcsdWrapper(ruby_pcsd.Wrapper):
16 def __init__(self, request_type):
17 self.request_type = request_type
18 self.status_code = 200
19 self.headers = {"Some": "value"}
20 self.body = b"Success action"
21
22 self._run_ruby_called = False
23 self._run_ruby_payload = None
24
25 async def run_ruby(
26 self,
27 request_type,
28 http_request=None,
29 payload=None,
30 ):
31 del http_request
32 self._run_ruby_called = True
33 self._run_ruby_payload = payload
34 if request_type != self.request_type:
35 raise AssertionError(
36 f"Wrong request type: expected '{self.request_type}'"
37 f" but was {request_type}"
38 )
39 return {
40 "headers": self.headers,
41 "status": self.status_code,
42 "body": self.body,
43 }
44
45 @property
46 def was_run_ruby_called(self) -> bool:
47 return self._run_ruby_called
48
49 @property
50 def run_ruby_payload(self):
51 return self._run_ruby_payload
52
53
54 class AppTest(AsyncHTTPTestCase):
55 wrapper = None
56
57 def get_app(self):
58 return Application(self.get_routes())
59
60 def get_routes(self):
61 return []
62
63 def fetch(self, path, raise_error=False, **kwargs):
64 if "follow_redirects" not in kwargs:
65 kwargs["follow_redirects"] = False
66
67 if "is_ajax" in kwargs:
68 if "headers" not in kwargs:
69 kwargs["headers"] = {}
70 kwargs["headers"]["X-Requested-With"] = "XMLHttpRequest"
71 del kwargs["is_ajax"]
72
73 response = super().fetch(path, raise_error=raise_error, **kwargs)
74 # "Strict-Transport-Security" header is expected in every response
75 self.assertTrue(
76 "Strict-Transport-Security" in response.headers,
77 f"No 'Strict-Transport-Security' header in response for '{path}'",
78 )
79 return response
80
81 def post(self, path, body, **kwargs):
82 kwargs.update(
83 {
84 "method": "POST",
85 "body": urlencode(body),
86 }
87 )
88 return self.fetch(path, **kwargs)
89
90 def get(self, path, **kwargs):
91 return self.fetch(path, **kwargs)
92
93 def assert_headers_contains(self, headers: HTTPHeaders, contained: dict):
94 self.assertTrue(
95 all(item in headers.get_all() for item in contained.items()),
96 "Headers does not contain expected headers"
97 "\n Expected headers:"
98 f"\n {pformat(contained, indent=6)}"
99 "\n All headers:"
100 f"\n {pformat(dict(headers.get_all()), indent=6)}",
101 )
102
103 def assert_wrappers_response(self, response):
104 self.assertEqual(response.code, self.wrapper.status_code)
105 self.assert_headers_contains(response.headers, self.wrapper.headers)
106 self.assertEqual(response.body, self.wrapper.body)
107
108 def assert_unauth_ajax(self, response):
109 self.assertEqual(response.code, 401)
110 self.assertEqual(response.body, b'{"notauthorized":"true"}')
111