diff --git a/scripts/clone_access_modules.py b/scripts/clone_access_modules.py index 86c4db99..feb82a29 100644 --- a/scripts/clone_access_modules.py +++ b/scripts/clone_access_modules.py @@ -2,6 +2,7 @@ import logging import os +import re import shutil import sys import time @@ -67,21 +68,81 @@ def get_repo_url_and_branch(formatted_git_arg): return url, target_branch +ACCESS_MODULES_DIR = "./Access/access_modules" + + +def safe_access_module_path(name): + """Resolve ``./Access/access_modules/`` while rejecting path traversal. + + ``name`` is untrusted: it is derived from a git URL (folder name) or read + from a cloned repository's directory listing. Strip any path components with + ``os.path.basename`` and confirm the resolved path stays inside + ``access_modules`` before it is used as a clone target or rename destination. + """ + safe_name = os.path.basename(str(name).strip().rstrip("/")) + if safe_name in ("", ".", ".."): + raise Exception(f"Unsafe access module name derived from {name!r}") + + target_path = os.path.join(ACCESS_MODULES_DIR, safe_name) + base_dir = os.path.realpath(ACCESS_MODULES_DIR) + resolved = os.path.realpath(target_path) + if resolved != base_dir and not resolved.startswith(base_dir + os.sep): + raise Exception(f"Refusing path outside access_modules: {name!r}") + return target_path + + +def _pinned_commit_sha(target_ref): + """Return target_ref when it is a full 40-char commit SHA (an integrity pin).""" + if target_ref and re.fullmatch(r"[0-9a-fA-F]{40}", target_ref): + return target_ref + return None + + +def verify_module_integrity(repo, url, pinned_sha): + """F-019: verify a cloned module against its pinned commit SHA. + + External modules are cloned at runtime and imported directly, so a + compromised upstream = RCE. If a commit SHA was pinned, check it out and + assert HEAD matches, aborting on mismatch; otherwise warn that integrity + could not be verified. + """ + if not pinned_sha: + logger.warning( + "Access module %s is not pinned to a commit SHA; runtime integrity " + "cannot be verified (F-019). Pin via '#<40-char-commit-sha>'.", + url, + ) + return + repo.git.checkout(pinned_sha) + actual_sha = repo.head.commit.hexsha + if actual_sha.lower() != pinned_sha.lower(): + raise Exception( + f"Integrity check failed for {url}: expected commit " + f"{pinned_sha}, got {actual_sha}" + ) + logger.info("Verified access module %s at pinned commit %s", url, pinned_sha) + + def clone_repo(formatted_git_arg, retry_limit): """ Clone a single repo """ - url, target_branch = get_repo_url_and_branch(formatted_git_arg) + url, target_ref = get_repo_url_and_branch(formatted_git_arg) folder_name = url.split("/").pop()[:-4] - target_folder_path = "./Access/access_modules/" + folder_name + # F-023: folder_name comes from the (untrusted) git URL — sanitize before use + target_folder_path = safe_access_module_path(folder_name) + + # F-019: a full 40-char SHA after '#' is an integrity pin, not a branch. + pinned_sha = _pinned_commit_sha(target_ref) retry_exception = None + repo = None for clone_attempt in range(1, retry_limit + 1): try: logger.info("Cloning Repo") - if target_branch: - Repo.clone_from(url, target_folder_path, branch=target_branch) + if target_ref and not pinned_sha: + repo = Repo.clone_from(url, target_folder_path, branch=target_ref) else: - Repo.clone_from(url, target_folder_path) + repo = Repo.clone_from(url, target_folder_path) except (GitCommandError, Exception) as exception: sleep_time = 10 * clone_attempt logger.error( @@ -104,6 +165,8 @@ def clone_repo(formatted_git_arg, retry_limit): logger.exception("Max retry count reached while cloning repo") raise retry_exception + verify_module_integrity(repo, url, pinned_sha) + return target_folder_path @@ -123,9 +186,11 @@ def move_modules_from_cloned_repo(cloned_path): and each_path not in blacklist_paths ): try: + # F-021: each_path comes from the cloned repo — sanitize the + # rename destination so a crafted dir name can't escape. os.rename( cloned_path + "/" + each_path, - "./Access/access_modules/" + each_path + safe_access_module_path(each_path) ) except Exception as exception: logger.exception(