from __future__ import with_statement import logging import os import subprocess import sys import tempfile import traceback from datetime import datetime from functools import partial from hashlib import md5 from json import dumps, loads from operator import itemgetter from abl.installer.installer import Updater from abl.vpath.base import URI from .auto_updates import check_and_process_auto_updates from .ipc_channel import IPCChannel from .reporting import SWAPPER_EXT from .reporting import EventTypes as et from .reporting import time_delta_ms from .util import assert_, check_exe_running logger = logging.getLogger(__name__) def create_copy_app_base_path(base_path): # replace the last step with the copy Live name split_path = base_path.last().rsplit(".app", 1) original_live_name, extension = ( split_path if len(split_path) > 1 else (split_path[0], "") ) temp_live_name = ".%s_updated" % original_live_name left_side, _, right_side = str(base_path).rpartition(original_live_name) return URI(left_side + temp_live_name + right_side) def create_args(args_dict): args = [] for key, value in args_dict.iteritems(): args.append("%s=%s" % (key, value)) return args # ---------------------------------------------------------------------------------------- class InstallerInvoker(object): def __init__(self, updater_class=Updater): self.updater_class = updater_class self.ready_to_swap = False def prepare_available_auto_update( self, connector, current_version, to_version, usage_reporter ): update_info = connector.load_update_info() to_versions = update_info["versions"][current_version] to_version_info = to_versions[to_version] # the version must be fully available assert_( lambda: to_version_info["status"] == "available", "Update %s -> %s not available" % (current_version, to_version_info), ) try: usage_reporter.log(et.COPY_INSTALL) logger.info("Copy installation") start = datetime.now() updater = self.updater_class( connector.app_base_path, connector.copy_app_base_path, connector.autoupdates_dir_base_path, ) time_elapsed = time_delta_ms(datetime.now() - start) usage_reporter.log(et.COPIED_INSTALL, elapsed_ms=time_elapsed) logger.info("Copy installation completed (%d ms)", time_elapsed) usage_reporter.log(et.UPDATE_COPY) logger.info("Update cloned installation") start = datetime.now() updater.prepare_auto_update(current_version, to_version, usage_reporter) time_elapsed = time_delta_ms(datetime.now() - start) usage_reporter.log(et.UPDATED_COPY, elapsed_ms=time_elapsed) logger.info("Update cloned installation completed (%d ms)", time_elapsed) except: exc_str = traceback.format_exception( sys.exc_info()[0], sys.exc_info()[1], sys.exc_info()[2] ) usage_reporter.log(et.EXCEPTION, stacktrace="\n".join(exc_str)) logger.error("Preparing automatic update failed:\n stacktrace: %s", exc_str) return finally: connector.cleanup_auto_updates_deltas() self.ready_to_swap = True def apply_available_auto_update(self, connector, usage_reporter): logger.info("Invoking swapper to apply update") usage_reporter.log(et.INVOKING_SWAPPER) if connector.swapper_exe: args = { "--live": str(connector.app_base_path), "--copy": str(connector.copy_app_base_path), } if connector.swapper_sentinel_file: args["--sentinel"] = connector.swapper_sentinel_file if connector.swapper_usage_report_file: args["--usagedata"] = connector.swapper_usage_report_file if connector.parent_pid: args["--parentpid"] = connector.parent_pid swapper_proc = connector.invoke_swapper(create_args(args)) if swapper_proc.poll() is not None: logger.error("Swapper terminated before WebConnector") else: logger.info( "Swapper invoked with pid: %d", getattr(swapper_proc, "pid", 0) ) # Now we have to terminate, because the swapper doesn't do its thing until # the Web Connector is gone. Errors and such will be written to the swapper's # own log file. else: raise Exception("No Ableton Swapper executable specified in configuration") def pending_update_available(self): return self.ready_to_swap def copy_exe_temp(exe_path, temp_dir): exe_name = exe_path.basename() temp_dir_path = temp_dir target_path = URI(temp_dir_path) / exe_name try: exe_path.copy(target_path) except: logger.error(sys.exc_info()[1]) raise Exception("Could not relocate Swapper executable") return target_path # ---------------------------------------------------------------------------------------- class CleanupResidueFailed(Exception): pass class ConnectorBase(object): def __init__(self, config, popen_factory=None, temp_dir_accessor=None): self.app_base_path = URI(config.app_base_path) if self.app_base_path.path.endswith("/") or self.app_base_path.path.endswith( "\\" ): self.app_base_path = self.app_base_path.split()[0] self.autoupdates_dir_base_path = URI(config.autoupdates_dir_base_path) self.auto_updater_exe = config.get("auto_updater_exe", None) self.auto_updater_main = config.get("auto_updater_main", None) self.app_exe_path = config.get("app_exe_path", None) self.swapper_sentinel_file = config.get("swapper_sentinel_file", None) swapper_exe_temp = config.get("swapper_exe", None) if swapper_exe_temp: self.swapper_exe = URI(swapper_exe_temp) else: self.swapper_exe = None self.swapper_usage_report_file = None self.parent_pid = config.get("pid", None) self.copy_app_base_path = create_copy_app_base_path(self.app_base_path) self.invoker = self.create_invoker() if "usage_report_prefix" in config and "aip_usage_data_dir" in config: self.swapper_usage_report_file = os.path.join( config["aip_usage_data_dir"], "%s%s" % (config["usage_report_prefix"], SWAPPER_EXT), ).encode("utf-8") logging_config = config.get("logging", {}) self.log_level = logging_config.get("log_level") self.log_file_path = logging_config.get("file_path") if popen_factory is None: popen_factory = partial( subprocess.Popen, stderr=subprocess.PIPE, stdout=subprocess.PIPE ) self.popen_factory = popen_factory if temp_dir_accessor is None: temp_dir_accessor = tempfile.mkdtemp self.temp_dir_accessor = temp_dir_accessor self.live_ipc_channel = ( IPCChannel(config.get("ipc_channel_id"), events=["continue", "cancel"]) if "ipc_channel_id" in config else None ) self.wait_for_confirmation = config.get("wait_for_confirmation", False) def notify_error(self, immediate=True): """Notify Live that an error occurred when updating""" if self.live_ipc_channel: self.live_ipc_channel.send(u"update_failed", u"", immediate) def create_invoker(self): return InstallerInvoker() def auto_updates_base(self): dir_ = self.autoupdates_dir_base_path / "_autoupdates" if not dir_.exists(): dir_.mkdir() return dir_ def cleanup_dir(self, dir_): logger.info("Cleaning up %s", str(dir_)) try: dir_.remove(recursive=True) except Exception: # failing to remove "_autoupdates" shouldn't stop execution logger.exception("Cleaning up %s failed", str(dir_)) def cleanup_auto_updates_base(self): self.cleanup_dir(self.autoupdates_dir_base_path / "_autoupdates") def cleanup_auto_updates_deltas(self): self.cleanup_dir(self.autoupdates_dir_base_path / "_autoupdates" / "deltas") @property def info_file_name(self): return self.auto_updates_base() / "info.json" def load_update_info(self): info_file = self.info_file_name if info_file.exists(): with info_file.open("r") as inf: return loads(inf.read()) return {} def save_update_info(self, data): temp = self.info_file_name + ".temp" if temp.exists(): temp.remove() with temp.open("wb") as outf: outf.write(dumps(data)) temp.move(self.info_file_name) def invoke_swapper(self, args): assert_(lambda: self.swapper_exe is not None, "No Swapper exe given") tmpdir = self.temp_dir_accessor() copied_swapper_exe = copy_exe_temp(self.swapper_exe, tmpdir) unsafeactivity_lib = ( URI(self.swapper_exe.directory()) / "libRealtimeUnsafeActivityObserverLib.dylib" ) if unsafeactivity_lib.exists(): copy_exe_temp(unsafeactivity_lib, tmpdir) cmd = [str(copied_swapper_exe)] cmd.extend(args) logger.debug("Swapper: %s", " ".join(cmd)) # Make sure to set the working directory to something else; otherwise, on # Windows the Swapper gets Live's directory as its working directory which # prevents the directory from being renamed/deleted. return self.popen_factory(cmd, cwd=str(copied_swapper_exe.directory())) def prepare_available_auto_update(self, current_version, to_version, usage_reporter): self.invoker.prepare_available_auto_update( self, current_version, to_version, usage_reporter ) def apply_available_auto_update(self, usage_reporter): self.invoker.apply_available_auto_update(self, usage_reporter) def check_live_is_running(self): assert self.app_exe_path return check_exe_running(self.app_exe_path) def extraction_dir(self): autoupdate_base = self.auto_updates_base() return autoupdate_base / "pending_update" def cleanup_residue(self): extraction_dir = self.extraction_dir() if extraction_dir.exists(): logger.info("Cleaning up residue") try: extraction_dir.remove(recursive=True) except: logger.exception("Cleaning up residue failed, aborting Web Connector") raise CleanupResidueFailed def pending_update_available(self): return self.invoker.pending_update_available() def skip_version(self, current_version, to_version): info = self.load_update_info() info["versions"][current_version][to_version]["status"] = "skipped" self.save_update_info(info) def check_and_process_auto_updates(self, tags, current_version, usage_reporter): return check_and_process_auto_updates(self, tags, current_version, usage_reporter) def check_for_auto_updates( self, tags, current_version, update_to_version ): # pragma: no cover raise NotImplementedError def retrieve_delta_into(self, delta_name, dest): # pragma: no cover raise NotImplementedError def upload_event_log(self, event_log_content): # pragma: no cover raise NotImplementedError def upload_log(self, run_id, crash_log_content): # pragma: no cover raise NotImplementedError class DirectoryConnector(ConnectorBase): PATTERNS = { "mac-intel": "mac_intel", "mac-universal": "mac_universal", "mac-arm64": "mac_arm", "win64": "win_64", "lite": "lite", "standard": "standard", "intro": "intro", "suite": "suite", "trial": "trial", "beta": "beta", } def __init__(self, config, base, popen_factory=None, temp_dir_accessor=None): super(DirectoryConnector, self).__init__( config, popen_factory=popen_factory, temp_dir_accessor=temp_dir_accessor ) self.base = base self.delta_lut = {} for root, _, filenames in self.base.walk(): for name in filenames: self.delta_lut[name] = root / name self.event_log_base = base / "eventlogs" if not self.event_log_base.exists(): self.event_log_base.mkdir() def delta_info(self, name): return dict(name=name, checksum=self.delta_lut[name].md5(), version=1) def deltas_for_tags(self, vdir, tags): patterns = [self.PATTERNS[tag] for tag in tags] deltas = vdir.glob("*.asu") for pattern in patterns: deltas = [delta for delta in deltas if pattern in str(delta)] deltas = sorted( (self.delta_info(delta.basename()) for delta in deltas), key=itemgetter("name"), ) if deltas: return {vdir.basename(): deltas} return {} def check_for_auto_updates(self, tags, current_version, update_to_version): d = self.base / current_version if not d.exists(): return {} versions = sorted(d.listdir()) if update_to_version is None and versions: update_to_version = versions[-1] if update_to_version not in versions: return {} vdir = d / update_to_version return self.deltas_for_tags(vdir, tags) def retrieve_delta_into(self, delta_name, dest): h = md5() with dest.open("wb") as outf: with self.delta_lut[delta_name].open("rb") as inf: block = inf.read(4096) while block: outf.write(block) h.update(block) block = inf.read(4096) return h.hexdigest() def upload_event_log(self, event_log_content): try: run_id = sorted(int(name) for name in self.event_log_base.listdir())[-1] + 1 except IndexError: run_id = 1 d = self.event_log_base / str(run_id) d.mkdir() with (d / "event_log").open("wb") as outf: outf.write(event_log_content) return run_id def upload_log(self, run_id, kind, log_content): d = self.event_log_base / str(run_id) with (d / ("%s_log" % kind)).open("wb") as outf: outf.write(log_content) return True