liuyu/qlib
1
0
Fork 0
mirror of https://github.com/microsoft/qlib.git synced 2026-06-06 05:51:17 +08:00
Code Activity

Fixing Security Vulnerabilities (#1941)

Browse Source 
* Fixing Security Vulnerabilities

* Fixing Pylint Error

* Fixing Security Vulnerabilities windows

* format with black

* using returncode to locate problems

* fix pylint error
This commit is contained in:
Linlang
2025-05-28 20:43:58 +08:00
committed by GitHub
parent 3e72593b8c
commit 3525514704
1 changed files with 36 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
qlib/__init__.py
View File

@@ -5,6 +5,7 @@ from pathlib import Path
__version__ = "0.9.6.99" __version__ = "0.9.6.99"
__version__bak = __version__ # This version is backup for QlibConfig.reset_qlib_version __version__bak = __version__ # This version is backup for QlibConfig.reset_qlib_version
import os import os
import re
from typing import Union from typing import Union
from ruamel.yaml import YAML from ruamel.yaml import YAML
import logging import logging
@@ -80,34 +81,41 @@ def _mount_nfs_uri(provider_uri, mount_path, auto_mount: bool = False):
LOG = get_module_logger("mount nfs", level=logging.INFO) LOG = get_module_logger("mount nfs", level=logging.INFO)
if mount_path is None: if mount_path is None:
raise ValueError(f"Invalid mount path: {mount_path}!") raise ValueError(f"Invalid mount path: {mount_path}!")
if not re.match(r"^[a-zA-Z0-9.:/\-_]+$", provider_uri):
raise ValueError(f"Invalid provider_uri format: {provider_uri}")
# FIXME: the C["provider_uri"] is modified in this function # FIXME: the C["provider_uri"] is modified in this function
# If it is not modified, we can pass only provider_uri or mount_path instead of C # If it is not modified, we can pass only provider_uri or mount_path instead of C
mount_command = "sudo mount.nfs %s %s" % (provider_uri, mount_path) mount_command = ["sudo", "mount.nfs", provider_uri, mount_path]
# If the provider uri looks like this 172.23.233.89//data/csdesign' # If the provider uri looks like this 172.23.233.89//data/csdesign'
# It will be a nfs path. The client provider will be used # It will be a nfs path. The client provider will be used
if not auto_mount: # pylint: disable=R1702 if not auto_mount: # pylint: disable=R1702
if not Path(mount_path).exists(): if not Path(mount_path).exists():
raise FileNotFoundError( raise FileNotFoundError(
f"Invalid mount path: {mount_path}! Please mount manually: {mount_command} or Set init parameter `auto_mount=True`" f"Invalid mount path: {mount_path}! Please mount manually: {' '.join(mount_command)} or Set init parameter `auto_mount=True`"
) )
else: else:
# Judging system type # Judging system type
sys_type = platform.system() sys_type = platform.system()
if "windows" in sys_type.lower(): if "windows" in sys_type.lower():
# system: window # system: window
exec_result = os.popen(f"mount -o anon {provider_uri} {mount_path}") try:
result = exec_result.read() subprocess.run(
if "85" in result: ["mount", "-o", "anon", provider_uri, mount_path],
LOG.warning(f"{provider_uri} on Windows:{mount_path} is already mounted") capture_output=True,
elif "53" in result: text=True,
raise OSError("not find network path") check=True,
elif "error" in result or "错误" in result: )
raise OSError("Invalid mount path") LOG.info("Mount finished.")
elif provider_uri in result: except subprocess.CalledProcessError as e:
LOG.info("window success mount..") error_output = (e.stdout or "") + (e.stderr or "")
else: if e.returncode == 85:
raise OSError(f"unknown error: {result}") LOG.warning(f"{provider_uri} already mounted at {mount_path}")
elif e.returncode == 53:
raise OSError("Network path not found") from e
elif "error" in error_output.lower() or "错误" in error_output:
raise OSError("Invalid mount path") from e
else:
raise OSError(f"Unknown mount error: {error_output.strip()}") from e
else: else:
# system: linux/Unix/Mac # system: linux/Unix/Mac
# check mount # check mount
@@ -119,12 +127,13 @@ def _mount_nfs_uri(provider_uri, mount_path, auto_mount: bool = False):
_is_mount = False _is_mount = False
while _check_level_num: while _check_level_num:
with subprocess.Popen( with subprocess.Popen(
'mount | grep "{}"'.format(_remote_uri), ["mount"],
shell=True, text=True,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.STDOUT, stderr=subprocess.STDOUT,
) as shell_r: ) as shell_r:
_command_log = shell_r.stdout.readlines() _command_log = shell_r.stdout.readlines()
_command_log = [line for line in _command_log if _remote_uri in line]
if len(_command_log) > 0: if len(_command_log) > 0:
for _c in _command_log: for _c in _command_log:
_temp_mount = _c.decode("utf-8").split(" ")[2] _temp_mount = _c.decode("utf-8").split(" ")[2]
@@ -152,16 +161,16 @@ def _mount_nfs_uri(provider_uri, mount_path, auto_mount: bool = False):
if not command_res: if not command_res:
raise OSError("nfs-common is not found, please install it by execute: sudo apt install nfs-common") raise OSError("nfs-common is not found, please install it by execute: sudo apt install nfs-common")
# manually mount # manually mount
command_status = os.system(mount_command) try:
if command_status == 256: subprocess.run(mount_command, check=True, capture_output=True, text=True)
raise OSError( LOG.info("Mount finished.")
f"mount {provider_uri} on {mount_path} error! Needs SUDO! Please mount manually: {mount_command}" except subprocess.CalledProcessError as e:
) if e.returncode == 256:
elif command_status == 32512: raise OSError("Mount failed: requires sudo or permission denied") from e
# LOG.error("Command error") elif e.returncode == 32512:
raise OSError(f"mount {provider_uri} on {mount_path} error! Command error") raise OSError(f"mount {provider_uri} on {mount_path} error! Command error") from e
elif command_status == 0: else:
LOG.info("Mount finished") raise OSError(f"Mount failed: {e.stderr}") from e
else: else:
LOG.warning(f"{_remote_uri} on {_mount_path} is already mounted") LOG.warning(f"{_remote_uri} on {_mount_path} is already mounted")