import os import io import hashlib import zipfile import platform import json from github.env import exit_on_error_and_write_summary from actions_logging.app_logging import logger import requests def compute_sha256(file_path): sha256_hash = hashlib.sha256() try: with open(file_path, 'rb') as f: for byte_block in iter(lambda: f.read(4096), b""): sha256_hash.update(byte_block) return sha256_hash.hexdigest() except FileNotFoundError: exit_on_error_and_write_summary(f"File not found: {file_path}") def compare_file_hash(expected_hash, file_path): logger.info(f"Comparing SHA256 hashes for file: {file_path}") calculated_hash = compute_sha256(file_path) if calculated_hash.lower() != expected_hash.lower(): exit_on_error_and_write_summary( f"Error: SHA256 checksum for {file_path} is invalid. Expected {expected_hash}, got {calculated_hash}") logger.info(f"SHA256 checksum for {file_path} is valid.") def download_files(username, password, platforms, base_url, default_destination, download_all): logger.info( f"Downloading files: base_url:{base_url}, default_destination:{default_destination}, download_all={download_all}") current_os_name = "" current_arch = "" if not download_all: current_os_name = os.getenv('OS_NAME') current_arch = os.getenv('ARCH') logger.info(f"from ENV: OS: {current_os_name}, ARCH: {current_arch}") if not current_os_name: logger.info("OS not provided") system = platform.system() logger.info(f"System: {system}") if system == "Darwin": current_os_name = "macOS" else: current_os_name = system logger.info(f"OS: {current_os_name}") if not current_arch: logger.info("ARCH not provided") machine = platform.machine() logger.info(f"machine: {machine}") if machine == "AMD64" or machine == "x86_64": current_arch = "X64" elif machine.startswith("arm") or machine.startswith("aarch"): current_arch = "ARM64" else: current_arch = machine logger.info(f"ARCH: {current_arch}") for item in platforms: logger.info(f"item: {item}") if not download_all: if "os" in item and item["os"] != current_os_name: logger.info(f"Skipping file as it does not match {current_os_name}") continue if "arch" in item and item["arch"] != current_arch: logger.info(f"Skipping file as it does not match {current_arch}") continue download_url = f"{base_url}/{item['source']}" destination = default_destination if "destination" in item: destination = f"{item['destination']}" if download_all and "os" in item: destination = f"{destination}/{item['os']}" if "arch" in item: destination = f"{destination}/{item['arch']}" logger.info(f"destination: {destination}") unzip = False if "unzip" in item: unzip = item["unzip"] hash = "" if "hash" in item: hash = item["hash"] download_file_from_nexus(username, password, download_url, destination, unzip, hash) def download_file_from_nexus(username, password, url, destination_directory, unzip, hash): logger.info( f"Downloading file: url:{url}, destination_directory:{destination_directory}, unzip:{unzip}, hash:{hash}") if not os.path.exists(destination_directory): os.makedirs(destination_directory) logger.info(f"Directory {destination_directory} created.") file_name = os.path.basename(url) destination = os.path.join(destination_directory, file_name) response = requests.get(url, auth=(username, password)) logger.info(f"Download status code: {response.status_code}") response.raise_for_status() logger.info(f'File {file_name} downloaded successfully') with open(f'{destination}', 'wb') as f: f.write(response.content) if hash: compare_file_hash(hash, destination) if unzip: with zipfile.ZipFile(io.BytesIO(response.content)) as z: z.extractall(destination_directory) logger.info(f'File {file_name} unzipped successfully to {destination_directory}') os.remove(destination) def get_config_global(config_path): with open(config_path, "r") as config_file: config_data = json.load(config_file) if "Global" not in config_data: exit_on_error_and_write_summary(f"Global section not found in config.json") return config_data["Global"] def add_nexus_arguments(parser): parser.add_argument('--nexus_url', type=str, nargs='?', help='Nexus repository DNS name (Default: nexus.perimeter81.com)', default='nexus.perimeter81.com') parser.add_argument('--nexus_root_dir', type=str, nargs='?', help='Nexus repository root path (Default: firefly)', default='firefly') parser.add_argument('--nexus_repo_name', type=str, nargs='?', help='Nexus repository name (Default: agent-sign)', default='agent-sign') def get_nexus_base_url(args): return f"https://{args.nexus_url}/repository/{args.nexus_repo_name}/{args.nexus_root_dir}" def append_error(errors, dependency, item, message): if dependency not in errors: errors[dependency] = {} if item not in errors[dependency]: errors[dependency][item] = [] errors[dependency][item].append(message) def validate_manifest_dependencies(dependencies): errors = {} for dependency in dependencies: if not isinstance(dependencies[dependency], dict): append_error(errors, dependency, "", "Invalid dependency: not a dict") continue if "version" not in dependencies[dependency]: append_error(errors, dependency, "", "Missing version") if "platforms" not in dependencies[dependency]: append_error(errors, dependency, "", "Missing platforms") continue platforms = dependencies[dependency]["platforms"] if not isinstance(platforms, list): append_error(errors, dependency, "", "Invalid platforms: not a list") continue for index, item in enumerate(platforms): if not isinstance(item, dict): append_error(errors, dependency, index, "Invalid item: not a dict") continue if "os" in item and item["os"] not in ["Windows", "Linux", "macOS"]: append_error(errors, dependency, index, f"Invalid os: {item['os']}") if "arch" in item and item["arch"] not in ["ARM64", "X64"]: append_error(errors, dependency, index, f"Invalid arch: {item['arch']}") if not item["source"]: append_error(errors, dependency, index, "Missing source") if "unzip" in item and not isinstance(item["unzip"], bool): append_error(errors, dependency, index, "Invalid unzip: not a boolean") continue if "unzip" in item and item["unzip"] and not item["source"].endswith(".zip"): append_error(errors, dependency, index, "Invalid unzip: source is not a zip file") if "hash" not in item or not item["hash"]: append_error(errors, dependency, index, "Missing hash") if errors: logger.error(f"Errors found in manifest.json:") for dependency in errors: logger.error(f"Dependency: {dependency}") for item in errors[dependency]: if item == "": for message in errors[dependency][item]: logger.error(f" {message}") else: logger.error(f" Item {item}:") for message in errors[dependency][item]: logger.error(f" {message}") exit_on_error_and_write_summary(f"Errors found in manifest.json") def split_version(version): # Version comes in form of "1.2.3.4" and we would like to break it into major, minor, patch and build as seperate values return version.split(".") def generate_base_nexus_upload_path(version, branch, gha_run_number, release=False): logger.info(f'branch: {branch}, gha_run_number: {gha_run_number}, release: {release}') major, minor, patch, build = split_version(version) if not release: return f"{branch if branch == 'refs/heads/main' else 'v' + major + '.' + minor}/nightly/130{gha_run_number}" else: return f"v{major}.{minor}/stable/{patch}/{build}"