1 import json
2 import os
3 import ssl
4 import time
5 from collections.abc import Generator, Mapping
6 from ipaddress import AddressValueError, IPv6Address
7 from logging import Logger
8 from typing import Any, TypeVar
9 from urllib.error import HTTPError, URLError
10 from urllib.request import Request, urlopen
11
12 from pcs import settings
13 from pcs.common.async_tasks.dto import TaskIdentDto, TaskResultDto
14 from pcs.common.async_tasks.types import TaskFinishType
15 from pcs.common.interface.dto import DataTransferObject, from_dict, to_dict
16 from pcs.common.reports import report_dto_to_item
17 from pcs.common.reports.item import ReportItem, ReportItemContext
18 from pcs.lib.external import is_proxy_set
19
20 # There can be self signed certificates.
|
CID (unavailable; MK=81b6af60c8081064285298147f7a26bf) (#1 of 1): Certificate verification disabled (SIGMA.certificate_verification_disabled): |
|
(1) Event Sigma main event: |
The Python application disables certificate validation for the server by calling `_create_unverified_context()`, leaving the application open to a manipulator-in-the-middle (MITM) attack. |
|
(2) Event remediation: |
Use `create_default_context()` to create an instance which requires certificate verification by default. |
21 _UNVERIFIED_SSL_CONTEXT = ssl._create_unverified_context() # noqa: SLF001
22
23 _DTO = TypeVar("_DTO", bound=DataTransferObject)
24
25
26 def _format_addr_for_url(addr: str) -> str:
27 try:
28 IPv6Address(addr)
29 return f"[{addr}]"
30 except AddressValueError:
31 return addr
32
33
34 class ApiV2CommunicationError(Exception):
35 pass
36
37
38 class ApiV2RequestError(Exception):
39 """The request payload is invalid; retrying on another node won't help."""
40
41
42 class NodeConnector:
43 def __init__(
44 self,
45 logger: Logger,
46 addr: str,
47 port: int,
48 token: str,
49 node_name: str,
50 effective_username: str | None,
51 effective_groups: list[str] | None,
52 request_timeout: int,
53 ):
54 self._logger = logger
55 self._host = f"{_format_addr_for_url(addr)}:{port}"
56 self._token = token
57 self._node_name = node_name
58 self._effective_username = effective_username
59 self._effective_groups = effective_groups
60 self._request_timeout = request_timeout
61
62 def task_create(self, cmd_payload: Mapping[str, Any]) -> str:
63 payload = {
64 **cmd_payload,
65 "options": {
66 **cmd_payload.get("options", {}),
67 **self._effective_user_options(),
68 },
69 }
70 request = Request(
71 f"https://{self._host}/api/v2/task/create",
72 data=json.dumps(payload).encode("utf-8"),
73 headers={
74 "Content-Type": "application/json",
75 "Cookie": f"token={self._token}",
76 },
77 method="POST",
78 )
79 response = self._send_request(request, request_error_on_400=True)
80 task_ident = self._parse_response(response, TaskIdentDto).task_ident
81 self._logger.info("Created task %s on %s", task_ident, self._host)
82 return task_ident
83
84 def task_result(
85 self,
86 task_ident: str,
87 poll_interval: float = 1.0,
88 ) -> Generator[tuple[list[ReportItem], bool, Any], None, None]:
89 request = Request(
90 f"https://{self._host}/api/v2/task/result?task_ident={task_ident}",
91 headers={"Cookie": f"token={self._token}"},
92 method="GET",
93 )
94 reports_seen = 0
95 context = ReportItemContext(self._node_name)
96
97 # Poll indefinitely until node returns a finished state. No timeout
98 # parameter — APIv2 supports long-running tasks. This hardcoded
99 # behavior is intentional — no client needs anything else.
100 while True:
101 response = self._send_request(request)
102 result_dto = self._parse_response(response, TaskResultDto)
103
104 new_reports_dto = result_dto.reports[reports_seen:]
105 reports_seen = len(result_dto.reports)
106
107 self._log_poll_cycle(task_ident, result_dto, new_reports_dto)
108
109 new_reports = [
110 report_dto_to_item(r, context) for r in new_reports_dto
111 ]
112 if result_dto.task_finish_type == TaskFinishType.UNFINISHED:
113 if new_reports:
114 yield new_reports, False, None
115 time.sleep(poll_interval)
116 continue
117
118 yield (
119 new_reports,
120 result_dto.task_finish_type == TaskFinishType.SUCCESS,
121 result_dto.result,
122 )
123 return
124
125 def _effective_user_options(self) -> dict[str, Any]:
126 options: dict[str, Any] = {}
127 if self._effective_username:
128 options["effective_username"] = self._effective_username
129 if self._effective_groups:
130 options["effective_groups"] = self._effective_groups
131 return options
132
133 def _parse_response(self, response: str, dto_class: type[_DTO]) -> _DTO:
134 try:
135 return from_dict(dto_class, json.loads(response))
136 except Exception as e:
137 self._logger.error(
138 "Response from %s does not conform to %s %s",
139 self._host,
140 dto_class.__name__,
141 e,
142 )
143 raise ApiV2CommunicationError(f"Unexpected response: {e}") from e
144
145 def _send_request(
146 self, request: Request, request_error_on_400: bool = False
147 ) -> str:
148 try:
149 self._logger.info("Sending request to %s", request.full_url)
150 # Timeout applies to each individual HTTP request, not to the
151 # overall operation. Both task/create and task/result are
152 # lightweight server-side operations that should respond quickly.
153 # The long-running part is the polling loop in task_result, which
154 # is intentionally unbounded.
155 with urlopen(
156 request,
157 context=_UNVERIFIED_SSL_CONTEXT,
158 timeout=self._request_timeout,
159 ) as response:
160 return response.read().decode("utf-8")
161 except HTTPError as e:
162 self._logger.info(
163 "HTTP error: %s %s for %s", e.code, e.reason, request.full_url
164 )
165 # HTTP 400 from task/create means the payload is inherently
166 # invalid — no node can process it. All 400 cases (api_v2.py):
167 # - malformed JSON body (prepare)
168 # - missing request body (RequestBodyMissingError)
169 # - missing required key in CommandDto (_from_dict_exc_handled)
170 # - unexpected keys in request body (_from_dict_exc_handled)
171 # - other payload structure violations (_from_dict_exc_handled)
172 if request_error_on_400 and e.code == 400:
173 raise ApiV2RequestError(
174 f"HTTP error {e.code}: {e.reason}"
175 ) from e
176 raise ApiV2CommunicationError(
177 f"HTTP error {e.code}: {e.reason}"
178 ) from e
179 except URLError as e:
180 self._logger.info(
181 "URL error: %s for %s", e.reason, request.full_url
182 )
183 # Proxy env vars (https_proxy, all_proxy) are respected by
184 # urllib and may cause connection failures when set
185 # unintentionally. We intentionally do not disable proxy — it
186 # can be legitimate (e.g. webUI managing a remote cluster).
187 # Instead, we warn to aid debugging, consistent with the rest
188 # of the codebase (NodeCommunicator, pcs/utils.py, pcsd/pcs.rb).
189 # See: https://bugzilla.redhat.com/show_bug.cgi?id=1315627
190 if is_proxy_set(os.environ):
191 self._logger.warning(
192 "Proxy is set in environment variables, try disabling it"
193 )
194 raise ApiV2CommunicationError(f"URL error: {e.reason}") from e
195 except Exception as e:
196 self._logger.info(
197 "Unexpected error %s for %s",
198 e.__class__.__name__,
199 request.full_url,
200 )
201 raise ApiV2CommunicationError(
202 f"Unexpected error: {e.__class__.__name__}"
203 ) from e
204
205 def _log_poll_cycle(
206 self,
207 task_ident: str,
208 result_dto: TaskResultDto,
209 new_reports_dto: list,
210 ) -> None:
211 finish_type = result_dto.task_finish_type
212 kill_reason = result_dto.kill_reason
213 ident = task_ident
214 reports = [to_dict(r) for r in new_reports_dto]
215
216 if reports:
217 self._logger.debug("Task %s new reports: %s", ident, reports)
218
219 if finish_type == TaskFinishType.UNFINISHED:
220 self._logger.info("Task %s status: %s", ident, finish_type)
221 return
222
223 self._logger.info("Task %s finished with status %s", ident, finish_type)
224
225 if finish_type == TaskFinishType.SUCCESS:
226 self._logger.info("Task %s result: %s", ident, result_dto.result)
227
228 if finish_type == TaskFinishType.KILL and kill_reason:
229 self._logger.info("Task %s kill reason: %s", ident, kill_reason)
230
231
232 class ApiV2Client:
233 CommunicationError = ApiV2CommunicationError
234 RequestError = ApiV2RequestError
235
236 def __init__(
237 self,
238 logger: Logger,
239 effective_username: str | None = None,
240 effective_groups: list[str] | None = None,
241 request_timeout: int | None = None,
242 ):
243 self._logger = logger
244 self._effective_username = effective_username
245 self._effective_groups = effective_groups
246 self._request_timeout = (
247 request_timeout
248 if request_timeout is not None
249 else settings.default_request_timeout
250 )
251
252 def node_connector(
253 self, addr: str, port: int, token: str, node_name: str
254 ) -> NodeConnector:
255 return NodeConnector(
256 self._logger,
257 addr,
258 port,
259 token,
260 node_name,
261 self._effective_username,
262 self._effective_groups,
263 self._request_timeout,
264 )
265