security(supply-chain): access-module clone hardening — path traversal + integrity pin (F-019/021/023)

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/<name>`` 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 '<git-url>#<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(