diff --git a/doc/release_notes.md b/doc/release_notes.md index 013a6f592d..0315c86012 100644 --- a/doc/release_notes.md +++ b/doc/release_notes.md @@ -53,6 +53,8 @@ * Prevent `create_zenodo_deposition_cli` from leaking secrets in logs under `--debug` by scoping debug logging to the local logger instead of the root logger ([#781](https://github.com/open-energy-transition/open-tyndp/pull/781)). +* Add sanitization of CLI inputs passed to `launch_explorer` ([#776](https://github.com/open-energy-transition/open-tyndp/pull/776)). + ## Upcoming PyPSA-Eur Release * Security: SBOM security scan included in CI. diff --git a/rules/sb.smk b/rules/sb.smk index c6c7c121ab..a4e4e68b8e 100644 --- a/rules/sb.smk +++ b/rules/sb.smk @@ -1120,13 +1120,10 @@ rule launch_explorer: import platform import subprocess import sys - from pathlib import Path output_log = str(output[0]) input_files = list(input) - Path(output_log).touch() - # Define command line executable cmd = [ sys.executable, @@ -1137,6 +1134,7 @@ rule launch_explorer: print(params.launch_msg) + # Open logfile before Popen so the log exists when the subprocess validates its path popen_kwargs = { "stdout": open(output_log, "w"), "stderr": subprocess.STDOUT, diff --git a/scripts/sb/launch_explorer.py b/scripts/sb/launch_explorer.py index 33282af75b..49f207e5ef 100644 --- a/scripts/sb/launch_explorer.py +++ b/scripts/sb/launch_explorer.py @@ -20,19 +20,117 @@ logger = logging.getLogger(__name__) +# Exclude privileged ports +MIN_PORT = 1024 +# Highest valid TCP port (2**16 - 1) +MAX_PORT = 65535 -def import_network(fn: str): + +def sanitize_path(path_str: str, label: str) -> Path: """ - Import PyPSA network and set default color for 'none' carrier. + Resolve an untrusted path and confirm it stays inside the working directory. + + Returns the absolute, resolved path, guaranteed to exist and to lie within + the working directory. Absolute inputs, ``..`` traversal, and symlinks + pointing outside the directory are all rejected. + + Parameters + ---------- + path_str : str + Untrusted path, expected to be relative to the working directory. + label : str + Human-readable input name, used in error messages. + + Returns + ------- + pathlib.Path + The resolved path, guaranteed to exist and lie within the working directory. + + Raises + ------ + ValueError + If the path cannot be resolved (missing, unreadable, or a symlink loop) or + if it resolves outside the working directory. """ - n = pypsa.Network(fn) - n.carriers.loc["none", "color"] = "#000000" - return n + path = Path(path_str) + base_dir = Path.cwd().resolve() + try: + resolved = (base_dir / path).resolve(strict=True) + except OSError as e: + raise ValueError(f"Invalid {label}: {e}") from e + + if not resolved.is_relative_to(base_dir): + raise ValueError(f"{label} escapes the working directory: {path_str!r}") + + return resolved + + +def sanitize_port(port_str: str) -> int: + """ + Parse an untrusted TCP port and ensure it lies in the valid range. + + Parameters + ---------- + port_str : str + Untrusted port value to validate. + + Returns + ------- + int + The parsed port number, guaranteed to be within ``MIN_PORT``-``MAX_PORT``. + + Raises + ------ + ValueError + If the value is not an integer within the range ``MIN_PORT``-``MAX_PORT``. + """ + if not (port_str.isdecimal() and MIN_PORT <= (port := int(port_str)) <= MAX_PORT): + raise ValueError( + f"Port must be an integer in {MIN_PORT}-{MAX_PORT}, got: {port_str!r}" + ) + return port + + +def sanitize_input_nc_files(path_strs: list[str]) -> list[str]: + """ + Validate untrusted network file paths. + + Each path must resolve inside the working directory (see ``sanitize_path``), + carry a ``.nc`` suffix, and refer to an existing file. + + Parameters + ---------- + path_strs : list[str] + Untrusted network file paths, expected to be relative to the working + directory. + + Returns + ------- + list[str] + The validated paths resolved against the working directory. + + Raises + ------ + ValueError + If any path fails validation, lacks a ``.nc`` suffix, or does not exist. + """ + files = [] + for path_str in path_strs: + path = sanitize_path(path_str, "network file path") + if path.suffix != ".nc" or not path.is_file(): + raise ValueError(f"Not an existing '.nc' file: {path_str!r}") + files.append(str(path)) + return files def open_browser(port): """ Opens browser with chosen port. + + Parameters + ---------- + port : int + Port on which the PyPSA-Explorer app is launched. """ webbrowser.open_new(f"http://127.0.0.1:{port}") @@ -52,7 +150,7 @@ def open_browser(port): # Get files and output path from snakemake or command line arguments if "snakemake" in globals(): - # Running from Snakemake directly for debugging + # Running with Mock Snakemake directly for debugging # Add parent directory to path to find scripts module sys.path.insert(0, str(Path(__file__).parent.parent.parent)) from scripts._helpers import configure_logging @@ -63,11 +161,11 @@ def open_browser(port): port = snakemake.params.port print("Running from Snakemake directly.") else: - # Running from command line + # CLI entry point for Snakemake, allows the app to be launched as a subprocess logging.basicConfig(level=logging.INFO) - output_log = sys.argv[1] - port = int(sys.argv[2]) - files = sys.argv[3:] + output_log = sanitize_path(sys.argv[1], "log path") + port = sanitize_port(sys.argv[2]) + files = sanitize_input_nc_files(sys.argv[3:]) print("Running from command line.") # Add file handler to write to explorer_launched.log @@ -78,7 +176,7 @@ def open_browser(port): logger.addHandler(file_handler) # Load networks into a dictionary for PyPSA-Explorer - networks = {fn.split("_")[-1].split(".")[0]: import_network(fn) for fn in files} + networks = {fn.split("_")[-1].split(".")[0]: pypsa.Network(fn) for fn in files} logger.info(f"Successfully loaded {len(networks)} networks: {networks}") # Create the PyPSA-Explorer dash app