import datetime import json import os import socket import subprocess import typer from celilo.ATAKArtifactUtils import version_configs, ATAKArtifactUtils from celilo.Wss import Wss from celilo.common import hazels_home, saru_home app = typer.Typer() print(f"working directory: {saru_home}") cli_version = "0.1.7" @app.command(name="downloadArtifacts", help="Downloads most recent artifact for all ATAK versions") def download_plugin_artifacts(): ATAKArtifactUtils().download_artifacts() @app.command(name="cherry-pick", help="Cherry pick a commit to all ATAK RC branch versions") def cherry_pick(excluded_version: str, commit_hash: str): if commit_hash is None: print("commit hash is required") return for version_config in version_configs: if version_config["version"] != excluded_version: branch = f"{version_config['branch_ref']}" os.system(f"cd {saru_home} && git checkout {branch}-rc") os.system(f"cd {saru_home} && git cherry-pick {commit_hash}") else: print(f"skipping {excluded_version}") def get_digest(stdout): # Signer #1 certificate SHA-256 digest: ccb0994def1e512877a02bf84ef1099ebae3d156dd78e638b77f5c31c9ac7234 lines = stdout.splitlines() for line in lines: if "SHA-256 digest" in line: parts = line.split(':') if len(parts) > 1: digest = parts[1].strip() return digest valid_digests = { "f24a38057275fcecf67be975ab803d12f75dc23581bef69cba9eb03a15bb8c17": "Playstore Signature", "33cdcb132a0ef15c93cdc2f7db9751d88772942070741a62e4d9afabcea49316": "ATAK Production (TPC) Signature", "ccb0994def1e512877a02bf84ef1099ebae3d156dd78e638b77f5c31c9ac7234": "ATAK ODK Signature", } def validate_apk(path, output_format): result = subprocess.run(["apksigner", "verify", "--verbose", "--print-certs", path], capture_output=True, text=True) if result.returncode == 0: obj = {} # print output digest_key = get_digest(result.stdout) valid_digest = valid_digests[digest_key] if output_format == "json": obj["path"] = path obj["digest"] = digest_key obj["desk"] = valid_digest else: print(f"APK Signature for {path}") print(f"\033[90m \t\t certificate SHA-256: {digest_key}\033[0m") print(f"\t\t Digest: \033[92m{valid_digest}\033[0m") print("\033[90m----------------------------------\033[0m") return obj else: print(f"APK signature is invalid for {path}") print(result.stderr) return None @app.command(name="validate-atak-signature", help="Identify the signatures of all ATAK Plugin APK files or directory;" "pass 'json' to output in JSON format otherwise it will print to console") def validate_atak_signature(path: str, format: str = "std"): result = subprocess.run(["which", "apksigner"], capture_output=True, text=True) if result.returncode != 0: print( "apksigner not found, please install Android SDK Build Tools; or add it to your PATH from the Android SDK; " "location-> Android/sdk/build-tools//apksigner") return if result is None or not os.path.exists(path): print(f"APK file not found at {path}") return # if path is a directory, find the APK file if os.path.isdir(path): apk_files = [f for f in os.listdir(path) if f.endswith('.apk')] if not apk_files: print(f"No APK files found in directory {path}") return ary = [] for apk_file in apk_files: full_path = os.path.join(path, apk_file) obj = validate_apk(full_path, format) ary.append(obj) if format == "json": print(json.dumps(ary, indent=4)) @app.command(name="cherry-pick-m", help="Cherry pick a commit to all ATAK RC branch versions from a merged commit") def cherry_pick_m(excluded_version: str, commit_hash: str): if commit_hash is None: print("commit hash is required") return for version_config in version_configs: if version_config["version"] != excluded_version: branch = f"{version_config['branch_ref']}" os.system(f"cd {saru_home} && git checkout {branch}-rc") os.system(f"cd {saru_home} && git cherry-pick -m 1 {commit_hash}") else: print(f"skipping {excluded_version}") @app.command(help="git status for all Saru maintenance branches") def status(branch_suffix: str = ''): for version_config in version_configs: branch = f"{version_config['branch_ref']}" os.system(f"cd {saru_home} && git checkout {branch}-rc") os.system(f"cd {saru_home} && git status") @app.command(help="git push for all Saru maintenance branches") def push(): for version_config in version_configs: branch = f"{version_config['branch_ref']}" os.system(f"cd {saru_home} && git checkout {branch}-rc") os.system(f"cd {saru_home} && git push") @app.command(help="Display the current version of the celilo package") def version(): print(f"ATAK version: {cli_version}") @app.command(help="git pull for all Saru maintenance branches") def pull(): for version_config in version_configs: branch = f"{version_config['branch_ref']}" os.system(f"cd {saru_home} && git checkout {branch}-rc") os.system(f"cd {saru_home} && git pull") def read_properties(file): with open(file, 'rb') as config_file: configs = {} for line in config_file: line = line.decode('utf-8').strip() if line and not line.startswith('#'): key_value = line.split('=') key = key_value[0].strip() value = key_value[1].strip() if len(key_value) > 1 else None configs[key] = value return configs @app.command(help="sign production ATAK APK") def sign_production_apk(path): assert path keystore_properties = read_properties(f"{hazels_home}/keystore.properties") keystore = keystore_properties['storeFile'] storePassword = keystore_properties['storePassword'] keyAlias = keystore_properties['keyAlias'] keyPassword = keystore_properties['keyPassword'] inputs = f"-keystore {keystore} -storepass '{storePassword}' -keypass '{keyPassword}' {path} {keyAlias}" os.system(f"cd {hazels_home}/app && jarsigner -verbose -sigalg SHA256withRSA -digestalg SHA-256 {inputs}") @app.command(help="WSS server") def ws(ip: str = None): if not ip: ip = socket.gethostbyname(socket.gethostname()) print(f"ip not provided using: {ip}") Wss(ip).start() @app.command(help="FW update") def fw_update(ip: str = None): if not ip: ip = socket.gethostbyname(socket.gethostname()) print(f"ip not provided using: {ip}") Wss(ip).start() if __name__ == '__main__': app()