scripts: Introduce kernel_source.py

`kernel_source.py` contains functions used to check version of desired
kernel, download, extract kernel source to specific directory and
install the headers of specific architecture.

Signed-off-by: Ruoqing He <[email protected]>

Author: RuoqingHe

scripts/lib/kernel_source.py

@@ -0,0 +1,164 @@
+# Copyright 2025 © Institute of Software, CAS. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+import os
+import re
+import tarfile
+import requests
+import subprocess
+import tempfile
+from lib import SUPPORT_ARCHS
+
+KERNEL_ORG_CDN = "https://cdn.kernel.org/pub/linux/kernel"
+
+
+def prepare_source(args):
+    check_kernel_version(args.version)
+
+    # Create `temp_dir` under `/tmp`
+    temp_dir = create_temp_dir(args.version)
+
+    # Download kernel tarball from https://cdn.kernel.org/
+    tarball = download_kernel(args.version, temp_dir)
+
+    # Extract kernel source
+    src_dir = extract_kernel(tarball, temp_dir)
+
+    # If arch is not provided, install headers for all supported archs
+    if args.arch is None:
+        for arch in SUPPORT_ARCHS:
+            installed_header_path = install_headers(
+                src_dir=src_dir,
+                arch=arch,
+                install_path=args.install_path,
+            )
+    else:
+        installed_header_path = install_headers(
+            src_dir=src_dir,
+            arch=args.arch,
+            install_path=args.install_path,
+        )
+
+    print(f"\nSuccessfully installed kernel headers to {installed_header_path}")
+    return installed_header_path
+
+
+def check_kernel_version(version):
+    """
+    Validate if the input kernel version exists in remote. Supports both X.Y
+    (namely X.Y.0 and .0 should be omitted) and X.Y.Z formats
+    """
+    # Validate version format
+    if not re.match(r"^\d+\.\d+(\.\d+)?$", version):
+        raise ValueError("Invalid version format. Use X.Y or X.Y.Z")
+
+    main_ver = version.split(".")[0]
+    base_url = f"{KERNEL_ORG_CDN}/v{main_ver}.x/"
+    tarball = f"linux-{version}.tar.xz"
+
+    try:
+        # Fetch content of `base_url`
+        response = requests.get(base_url, timeout=15)
+        response.raise_for_status()
+
+        # Check for exact filename match
+        if tarball in response.text:
+            print(f"Kernel version {version} found in remote")
+            return
+
+        raise RuntimeError(f"Kernel version {version} not found in remote")
+
+    except requests.exceptions.HTTPError as e:
+        if e.response.status_code == 404:
+            raise RuntimeError(f"Kernel series v{main_ver}.x does not exist")
+
+        raise RuntimeError(f"HTTP error ({e.response.status_code}): {str(e)}")
+    except requests.exceptions.Timeout:
+        raise RuntimeError("Connection timeout while checking version")
+    except requests.exceptions.RequestException as e:
+        raise RuntimeError(f"Network error: {str(e)}")
+
+
+def create_temp_dir(version):
+    prefix = f"linux-{version}-source-"
+    try:
+        temp_dir = tempfile.TemporaryDirectory(prefix=prefix, dir="/tmp", delete=False)
+        return temp_dir.name
+    except OSError as e:
+        raise RuntimeError(f"Failed to create temp directory: {e}") from e
+
+
+def download_kernel(version, temp_dir):
+    version_major = re.match(r"^(\d+)\.\d+(\.\d+)?$", version).group(1)
+    url = f"{KERNEL_ORG_CDN}/v{version_major}.x/linux-{version}.tar.xz"
+    tarball_path = os.path.join(temp_dir, f"linux-{version}.tar.xz")
+    print(f"Downloading {url} to {tarball_path}")
+
+    try:
+        with requests.get(url, stream=True) as response:
+            response.raise_for_status()
+            total_size = int(response.headers.get("content-length", 0))
+            downloaded = 0
+
+            with open(tarball_path, "wb") as f:
+                for chunk in response.iter_content(chunk_size=8192):
+                    f.write(chunk)
+                    downloaded += len(chunk)
+                    if total_size > 0:
+                        progress = downloaded / total_size * 100
+                        print(f"\rDownloading: {progress:.1f}%", end="")
+                print()
+        return tarball_path
+    except Exception as e:
+        raise RuntimeError(f"Download failed: {e}") from e
+
+
+def extract_kernel(tarball_path, temp_dir):
+    print("Extracting...")
+    try:
+        with tarfile.open(tarball_path, "r:xz") as tar:
+            tar.extractall(path=temp_dir)
+            extract_path = os.path.join(
+                temp_dir, f"{os.path.basename(tarball_path).split('.tar')[0]}"
+            )
+            print(f"Extracted to {extract_path}")
+            return extract_path
+    except (tarfile.TarError, IOError) as e:
+        raise RuntimeError(f"Extraction failed: {e}") from e
+
+
+def install_headers(src_dir, arch, install_path):
+    # If install_path is not provided, install to parent directory of src_dir to
+    # prevent messing up with extracted kernel source code
+    if install_path is None:
+        install_path = os.path.dirname(src_dir)
+
+    try:
+        os.makedirs(install_path, exist_ok=True)
+
+        abs_install_path = os.path.abspath(
+            os.path.join(install_path, f"{arch}_headers")
+        )
+        print(f"Installing to {abs_install_path}")
+        result = subprocess.run(
+            [
+                "make",
+                "-C",
+                f"{src_dir}",
+                f"ARCH={arch}",
+                f"INSTALL_HDR_PATH={abs_install_path}",
+                "headers_install",
+            ],
+            check=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+        )
+        print(result.stdout)
+        return install_path
+
+    except subprocess.CalledProcessError as e:
+        raise RuntimeError(
+            f"Header installation failed:\n{e.output}"
+            f"Temporary files kept at: {os.path.dirname(src_dir)}"
+        )