(self, *, url: str, args: Dict[str, Dict[str, any]])
| 353 | f.close() |
| 354 | |
| 355 | def _perform_urllib_http_request(self, *, url: str, args: Dict[str, Dict[str, any]]) -> Dict[str, any]: |
| 356 | headers = args["headers"] |
| 357 | if args["json"]: |
| 358 | body = json.dumps(args["json"]) |
| 359 | headers["Content-Type"] = "application/json;charset=utf-8" |
| 360 | elif args["data"]: |
| 361 | boundary = f"--------------{uuid.uuid4()}" |
| 362 | sep_boundary = b"\r\n--" + boundary.encode("ascii") |
| 363 | end_boundary = sep_boundary + b"--\r\n" |
| 364 | body = io.BytesIO() |
| 365 | data = args["data"] |
| 366 | for key, value in data.items(): |
| 367 | readable = getattr(value, "readable", None) |
| 368 | if readable and value.readable(): |
| 369 | filename = "Uploaded file" |
| 370 | name_attr = getattr(value, "name", None) |
| 371 | if name_attr: |
| 372 | filename = name_attr.decode("utf-8") if isinstance(name_attr, bytes) else name_attr |
| 373 | if "filename" in data: |
| 374 | filename = data["filename"] |
| 375 | mimetype = mimetypes.guess_type(filename)[0] or "application/octet-stream" |
| 376 | title = ( |
| 377 | f'\r\nContent-Disposition: form-data; name="{key}"; filename="{filename}"\r\n' |
| 378 | + f"Content-Type: {mimetype}\r\n" |
| 379 | ) |
| 380 | value = value.read() |
| 381 | else: |
| 382 | title = f'\r\nContent-Disposition: form-data; name="{key}"\r\n' |
| 383 | value = str(value).encode("utf-8") |
| 384 | body.write(sep_boundary) |
| 385 | body.write(title.encode("utf-8")) |
| 386 | body.write(b"\r\n") |
| 387 | body.write(value) |
| 388 | |
| 389 | body.write(end_boundary) |
| 390 | body = body.getvalue() |
| 391 | headers["Content-Type"] = f"multipart/form-data; boundary={boundary}" |
| 392 | headers["Content-Length"] = len(body) |
| 393 | elif args["params"]: |
| 394 | body = urlencode(args["params"]) |
| 395 | headers["Content-Type"] = "application/x-www-form-urlencoded" |
| 396 | else: |
| 397 | body = None |
| 398 | |
| 399 | if isinstance(body, str): |
| 400 | body = body.encode("utf-8") |
| 401 | |
| 402 | # NOTE: Intentionally ignore the `http_verb` here |
| 403 | # Slack APIs accepts any API method requests with POST methods |
| 404 | try: |
| 405 | # urllib not only opens http:// or https:// URLs, but also ftp:// and file://. |
| 406 | # With this it might be possible to open local files on the executing machine |
| 407 | # which might be a security risk if the URL to open can be manipulated by an external user. |
| 408 | # (BAN-B310) |
| 409 | if url.lower().startswith("http"): |
| 410 | req = Request(method="POST", url=url, data=body, headers=headers) |
| 411 | opener: Optional[OpenerDirector] = None |
| 412 | if self.proxy is not None: |
no test coverage detected