import sys import requests import zipfile import json import os import platform import io nexus_url = 'nexus.perimeter81.com' nexus_root_dir = 'firefly' nexus_repo_name = 'agent-sign' manifest_path = 'manifest.json' config_path = 'deployment/config/config.json' def download_file(url, destination_directory, unzip): print(f"Downloading file: url:{url}, destination_directory:{destination_directory}, unzip:{unzip}") if not os.path.exists(destination_directory): os.makedirs(destination_directory) print(f"Directory {destination_directory} created.") file_name = os.path.basename(url) destination = os.path.join(destination_directory, file_name) if nexus_b64_creds: headers = { "Authorization": f"Basic {nexus_b64_creds}" } response = requests.get(url, headers=headers) else: response = requests.get(url, auth=(username, password)) print(f"Download status code: {response.status_code}") response.raise_for_status() print(f'File {file_name} downloaded successfully') if unzip: with zipfile.ZipFile(io.BytesIO(response.content)) as z: z.extractall(destination_directory) print(f'File {file_name} unzipped successfully to {destination_directory}') else: with open(f'{destination}', 'wb') as f: f.write(response.content) 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(): if "name" not in manifest_data: print("Missing name in manifest.json") sys.exit(1) if "version" not in manifest_data: print("Missing version in manifest.json") sys.exit(1) if "dependencies" not in manifest_data: print("Missing dependencies in manifest.json") sys.exit(1) errors = {} for dependency in manifest_data["dependencies"]: if not isinstance(manifest_data["dependencies"][dependency], dict): append_error(errors, dependency, "", "Invalid dependency: not a dict") continue if "version" not in manifest_data["dependencies"][dependency]: append_error(errors, dependency, "", "Missing version") if "platforms" not in manifest_data["dependencies"][dependency]: append_error(errors, dependency, "", "Missing platforms") continue platforms = manifest_data["dependencies"][dependency]["platforms"] if not isinstance(platforms, list): append_error(errors, dependency, "", "Invalid platforms: not a list") continue for index, item in enumerate(manifest_data["dependencies"][dependency]["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 errors: print(f"Errors found in manifest.json:") for dependency in errors: print(f"Dependency: {dependency}") for item in errors[dependency]: if item == "": for message in errors[dependency][item]: print(f" {message}") else: print(f" Item {item}:") for message in errors[dependency][item]: print(f" {message}") sys.exit(1) def download_files(): for dependency in manifest_data["dependencies"]: print(f"dependency: {dependency}") version = manifest_data["dependencies"][dependency]["version"] print(f"version: {version}") base_url = f"https://{nexus_url}/repository/{nexus_repo_name}/{nexus_root_dir}/{dependency}/{version}" for item in manifest_data["dependencies"][dependency]["platforms"]: print(f"item: {item}") if "os" in item and item["os"] != os_name: print(f"Skipping file as it does not match {os_name}") continue if "arch" in item and item["arch"] != arch: print(f"Skipping file as it does not match {arch}") continue download_url = f"{base_url}/{item['source']}" destination = libsDirectoryPath if "destination" in item: destination = f"{item['destination']}" print(f"destination: {destination}") unzip = False if "unzip" in item: unzip = item["unzip"] download_file(download_url, destination, unzip) if __name__ == "__main__": nexus_b64_creds = os.getenv('NEXUS_B64_CREDS') engines_download = os.getenv('DOWNLOAD_ENGINES') if nexus_b64_creds: print(f"got Nexus B64 creds") else: from dotenv import load_dotenv load_dotenv() username = os.getenv("NEXUS_USERNAME") password = os.getenv("NEXUS_PASSWORD") print(f"NEXUS_USERNAME: {username}") os_name = os.getenv('OS_NAME') arch = os.getenv('ARCH') print(f"from ENV: OS: {os_name}, ARCH: {arch}") if not os_name: print("OS not provided") system = platform.system() print(f"System: {system}") if system == "Darwin": os_name = "macOS" else: os_name = system print(f"OS: {os_name}") if not arch: print("ARCH not provided") machine = platform.machine() print(f"machine: {machine}") if machine == "AMD64" or machine == "x86_64": arch = "X64" elif machine.startswith("arm") or machine.startswith("aarch"): arch = "ARM64" else: arch = machine print(f"ARCH: {arch}") print(f"current working directory: {os.getcwd()}") # New functionality starts here FILE_PATH_TO_DOWNLOAD = os.getenv('FILE_PATH_TO_DOWNLOAD') TARGET_PATH = os.getenv('TARGET_PATH') if FILE_PATH_TO_DOWNLOAD: print("FILE_PATH_TO_DOWNLOAD is set, downloading specified file...") # Build the download URL download_url = f"https://{nexus_url}/repository/{nexus_repo_name}/{FILE_PATH_TO_DOWNLOAD}" # Set destination directory if TARGET_PATH: destination_directory = TARGET_PATH else: destination_directory = os.getcwd() # Determine whether to unzip based on file extension unzip = FILE_PATH_TO_DOWNLOAD.endswith('.zip') # Call download_file download_file(download_url, destination_directory, unzip) sys.exit() # New functionality ends here with open(manifest_path, "r") as manifest_file: manifest_data = json.load(manifest_file) with open(config_path, "r") as config_file: config_data = json.load(config_file) if "Global" not in config_data: print(f"Global section not found in config.json") sys.exit(1) if "libsDirectoryPath" not in config_data["Global"]: print(f"libsDirectoryPath not found in Global section of config.json") sys.exit(1) libsDirectoryPath = config_data["Global"]["libsDirectoryPath"] if engines_download == 'true': print("Start Downloading Engines...") url = "https://nexus.perimeter81.com/repository/agent-sign/firefly/localav/engines/engines.zip" workDirectoryPath = config_data["Global"]["workDirectoryPath"] if workDirectoryPath == "": workDirectoryPath = os.getcwd() dest_path = workDirectoryPath + os.sep + "testing" + os.sep + "blades" + os.sep + "threat-prevention" + os.sep + "engines" else: dest_path = workDirectoryPath + os.sep + "engines" download_file(url, dest_path, True) sys.exit() validate_manifest() download_files()