FreeCAD

1
# SPDX-License-Identifier: LGPL-2.1-or-later
2
# ***************************************************************************
3
# *                                                                         *
4
# *   Copyright (c) 2022 FreeCAD Project Association                        *
5
# *                                                                         *
6
# *   This file is part of FreeCAD.                                         *
7
# *                                                                         *
8
# *   FreeCAD is free software: you can redistribute it and/or modify it    *
9
# *   under the terms of the GNU Lesser General Public License as           *
10
# *   published by the Free Software Foundation, either version 2.1 of the  *
11
# *   License, or (at your option) any later version.                       *
12
# *                                                                         *
13
# *   FreeCAD is distributed in the hope that it will be useful, but        *
14
# *   WITHOUT ANY WARRANTY; without even the implied warranty of            *
15
# *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU      *
16
# *   Lesser General Public License for more details.                       *
17
# *                                                                         *
18
# *   You should have received a copy of the GNU Lesser General Public      *
19
# *   License along with FreeCAD. If not, see                               *
20
# *   <https://www.gnu.org/licenses/>.                                      *
21
# *                                                                         *
22
# ***************************************************************************
23

24
""" Contains the classes to manage Addon installation: intended as a stable API, safe for external
25
code to call and to rely upon existing. See classes AddonInstaller and MacroInstaller for details.
26
"""
27
import json
28
from datetime import datetime, timezone
29
from enum import IntEnum, auto
30
import os
31
import shutil
32
from typing import List, Optional
33
import tempfile
34
from urllib.parse import urlparse
35
import zipfile
36

37
import FreeCAD
38

39
from PySide import QtCore
40

41
from Addon import Addon
42
import addonmanager_utilities as utils
43
from addonmanager_metadata import get_branch_from_metadata
44
from addonmanager_git import initialize_git, GitFailed
45

46
if FreeCAD.GuiUp:
47
    import NetworkManager  # Requires an event loop
48

49
translate = FreeCAD.Qt.translate
50

51
# pylint: disable=too-few-public-methods
52

53

54
class InstallationMethod(IntEnum):
55
    """For packages installed from a git repository, in most cases it is possible to either use git
56
    or to download a zip archive of the addon. For a local repository, a direct copy may be used
57
    instead. If "ANY" is given, the internal code decides which to use."""
58

59
    GIT = auto()
60
    COPY = auto()
61
    ZIP = auto()
62
    ANY = auto()
63

64

65
class AddonInstaller(QtCore.QObject):
66
    """The core, non-GUI installer class. Usually instantiated and moved to its own thread,
67
    otherwise it will block the GUI (if the GUI is running). In all cases in this class, the
68
    generic Python 'object' is intended to be an Addon-like object that provides, at a minimum,
69
    a 'name', 'url', and 'branch' attribute. The Addon manager uses the Addon class for this
70
    purpose, but external code may use any other class that meets those criteria.
71

72
    Recommended Usage (when running with the GUI up, so you don't block the GUI thread):
73

74
        import functools # With the rest of your imports, for functools.partial
75

76
        ...
77

78
        addon_to_install = MyAddon() # Some class with name, url, and branch attributes
79

80
        self.worker_thread = QtCore.QThread()
81
        self.installer = AddonInstaller(addon_to_install)
82
        self.installer.moveToThread(self.worker_thread)
83
        self.installer.success.connect(self.installation_succeeded)
84
        self.installer.failure.connect(self.installation_failed)
85
        self.installer.finished.connect(self.worker_thread.quit)
86
        self.worker_thread.started.connect(self.installer.run)
87
        self.worker_thread.start() # Returns immediately
88

89
        # On success, the connections above result in self.installation_succeeded being called, and
90
        # on failure, self.installation_failed is called.
91

92

93
    Recommended non-GUI usage (blocks until complete):
94

95
        addon_to_install = MyAddon() # Some class with name, url, and branch attributes
96
        installer = AddonInstaller(addon_to_install)
97
        installer.run()
98

99
    """
100

101
    # Signal: progress_update
102
    # In GUI mode this signal is emitted periodically during long downloads. The two integers are
103
    # the number of bytes downloaded, and the number of bytes expected, respectively. Note that the
104
    # number of bytes expected might be set to 0 to indicate an unknown download size.
105
    progress_update = QtCore.Signal(int, int)
106

107
    # Signals: success and failure
108
    # Emitted when the installation process is complete. The object emitted is the object that the
109
    # installation was requested for (usually of class Addon, but any class that provides a name,
110
    # url, and branch attribute can be used).
111
    success = QtCore.Signal(object)
112
    failure = QtCore.Signal(object, str)
113

114
    # Finished: regardless of the outcome, this is emitted when all work that is going to be done
115
    # is done (i.e. whatever thread this is running in can quit).
116
    finished = QtCore.Signal()
117

118
    allowed_packages = set()
119

120
    def __init__(self, addon: Addon, allow_list: List[str] = None):
121
        """Initialize the installer with an optional list of addons. If provided, then installation
122
        by name is supported, as long as the objects in the list contain a "name" and "url"
123
        property. In most use cases it is expected that addons is a List of Addon objects, but that
124
        is not a requirement. An optional allow_list lets calling code override the allowed Python
125
        packages list with a custom list. It is mostly for unit testing purposes."""
126
        super().__init__()
127
        self.addon_to_install = addon
128

129
        self.git_manager = initialize_git()
130

131
        if allow_list is not None:
132
            AddonInstaller.allowed_packages = set(allow_list if allow_list is not None else [])
133
        elif not AddonInstaller.allowed_packages:
134
            AddonInstaller._load_local_allowed_packages_list()
135
            AddonInstaller._update_allowed_packages_list()
136

137
        basedir = FreeCAD.getUserAppDataDir()
138
        self.installation_path = os.path.join(basedir, "Mod")
139
        self.macro_installation_path = FreeCAD.getUserMacroDir(True)
140
        self.zip_download_index = None
141

142
    def run(self, install_method: InstallationMethod = InstallationMethod.ANY) -> bool:
143
        """Install an addon. Returns True if the addon was installed, or False if not. Emits
144
        either success or failure prior to returning."""
145
        try:
146
            addon_url = self.addon_to_install.url.replace(os.path.sep, "/")
147
            method_to_use = self._determine_install_method(addon_url, install_method)
148
            success = False
149
            if method_to_use == InstallationMethod.ZIP:
150
                success = self._install_by_zip()
151
            elif method_to_use == InstallationMethod.GIT:
152
                success = self._install_by_git()
153
            elif method_to_use == InstallationMethod.COPY:
154
                success = self._install_by_copy()
155
            if (
156
                hasattr(self.addon_to_install, "contains_workbench")
157
                and self.addon_to_install.contains_workbench()
158
            ):
159
                self.addon_to_install.enable_workbench()
160
        except utils.ProcessInterrupted:
161
            pass
162
        except Exception as e:
163
            FreeCAD.Console.PrintLog(e + "\n")
164
            success = False
165
        if success:
166
            if (
167
                hasattr(self.addon_to_install, "contains_workbench")
168
                and self.addon_to_install.contains_workbench()
169
            ):
170
                self.addon_to_install.set_status(Addon.Status.PENDING_RESTART)
171
            else:
172
                self.addon_to_install.set_status(Addon.Status.NO_UPDATE_AVAILABLE)
173
        self.finished.emit()
174
        return success
175

176
    @classmethod
177
    def _load_local_allowed_packages_list(cls) -> None:
178
        """Read in the local allow-list, in case the remote one is unavailable."""
179
        cls.allowed_packages.clear()
180
        allow_file = os.path.join(os.path.dirname(__file__), "ALLOWED_PYTHON_PACKAGES.txt")
181
        if os.path.exists(allow_file):
182
            with open(allow_file, encoding="utf8") as f:
183
                lines = f.readlines()
184
                for line in lines:
185
                    if line and len(line) > 0 and line[0] != "#":
186
                        cls.allowed_packages.add(line.strip().lower())
187

188
    @classmethod
189
    def _update_allowed_packages_list(cls) -> None:
190
        """Get a new remote copy of the allowed packages list from GitHub."""
191
        FreeCAD.Console.PrintLog(
192
            "Attempting to fetch remote copy of ALLOWED_PYTHON_PACKAGES.txt...\n"
193
        )
194
        p = utils.blocking_get(
195
            "https://raw.githubusercontent.com/"
196
            "FreeCAD/FreeCAD-addons/master/ALLOWED_PYTHON_PACKAGES.txt"
197
        )
198
        if p:
199
            FreeCAD.Console.PrintLog(
200
                "Overriding local ALLOWED_PYTHON_PACKAGES.txt with newer remote version\n"
201
            )
202
            p = p.decode("utf8")
203
            lines = p.split("\n")
204
            cls.allowed_packages.clear()  # Unset the locally-defined list
205
            for line in lines:
206
                if line and len(line) > 0 and line[0] != "#":
207
                    cls.allowed_packages.add(line.strip().lower())
208
        else:
209
            FreeCAD.Console.PrintLog(
210
                "Could not fetch remote ALLOWED_PYTHON_PACKAGES.txt, using local copy\n"
211
            )
212

213
    def _determine_install_method(
214
        self, addon_url: str, install_method: InstallationMethod
215
    ) -> Optional[InstallationMethod]:
216
        """Given a URL and preferred installation method, determine the actual installation method
217
        to use. Will return either None, if installation is not possible for the given url and
218
        method, or a specific concrete method (GIT, ZIP, or COPY) based on the inputs."""
219

220
        # If we don't have access to git, and that is the method selected, return early
221
        if not self.git_manager and install_method == InstallationMethod.GIT:
222
            return None
223

224
        parse_result = urlparse(addon_url)
225
        is_git_only = parse_result.scheme in ["git", "ssh", "rsync"]
226
        is_remote = parse_result.scheme in ["http", "https", "git", "ssh", "rsync"]
227
        is_zipfile = parse_result.path.lower().endswith(".zip")
228

229
        # Can't use "copy" for a remote URL
230
        if is_remote and install_method == InstallationMethod.COPY:
231
            return None
232

233
        if is_git_only:
234
            if (
235
                install_method in (InstallationMethod.GIT, InstallationMethod.ANY)
236
            ) and self.git_manager:
237
                # If it's a git-only URL, only git can be used for the installation
238
                return InstallationMethod.GIT
239
            # So if it's not a git installation, return None
240
            return None
241

242
        if is_zipfile:
243
            if install_method == InstallationMethod.GIT:
244
                # Can't use git on zip files
245
                return None
246
            return InstallationMethod.ZIP  # Copy just becomes zip
247
        if not is_remote and install_method == InstallationMethod.ZIP:
248
            return None  # Can't use zip on local paths that aren't zip files
249

250
        # Whatever scheme was passed in appears to be reasonable, return it
251
        if install_method != InstallationMethod.ANY:
252
            return install_method
253

254
        # Prefer to copy, if it's local:
255
        if not is_remote:
256
            return InstallationMethod.COPY
257

258
        # Prefer git if we have git
259
        if self.git_manager:
260
            return InstallationMethod.GIT
261

262
        # Fall back to ZIP in other cases, though this relies on remote hosts falling
263
        # into one of a few particular patterns
264
        return InstallationMethod.ZIP
265

266
    def _install_by_copy(self) -> bool:
267
        """Installs the specified url by copying directly from it into the installation
268
        location. addon_url must be copyable using filesystem operations. Any existing files at
269
        that location are overwritten."""
270
        addon_url = self.addon_to_install.url
271
        if addon_url.startswith("file://"):
272
            addon_url = addon_url[len("file://") :]  # Strip off the file:// part
273
        name = self.addon_to_install.name
274
        shutil.copytree(addon_url, os.path.join(self.installation_path, name), dirs_exist_ok=True)
275
        self._finalize_successful_installation()
276
        return True
277

278
    def _install_by_git(self) -> bool:
279
        """Installs the specified url by using git to clone from it. The URL can be local or remote,
280
        but must represent a git repository, and the url must be in a format that git can handle
281
        (git, ssh, rsync, file, or a bare filesystem path)."""
282
        install_path = os.path.join(self.installation_path, self.addon_to_install.name)
283
        try:
284
            if os.path.isdir(install_path):
285
                old_branch = get_branch_from_metadata(self.addon_to_install.installed_metadata)
286
                new_branch = get_branch_from_metadata(self.addon_to_install.metadata)
287
                if old_branch != new_branch:
288
                    utils.rmdir(install_path)
289
                    self.git_manager.clone(self.addon_to_install.url, install_path)
290
                else:
291
                    self.git_manager.update(install_path)
292
            else:
293
                self.git_manager.clone(self.addon_to_install.url, install_path)
294
            self.git_manager.checkout(install_path, self.addon_to_install.branch)
295
        except GitFailed as e:
296
            self.failure.emit(self.addon_to_install, str(e))
297
            return False
298
        self._finalize_successful_installation()
299
        return True
300

301
    def _install_by_zip(self) -> bool:
302
        """Installs the specified url by downloading the file (if it is remote) and unzipping it
303
        into the appropriate installation location. If the GUI is running the download is
304
        asynchronous, and issues periodic updates about how much data has been downloaded."""
305
        if self.addon_to_install.url.endswith(".zip"):
306
            zip_url = self.addon_to_install.url
307
        else:
308
            zip_url = utils.get_zip_url(self.addon_to_install)
309

310
        FreeCAD.Console.PrintLog(f"Downloading ZIP file from {zip_url}...\n")
311
        parse_result = urlparse(zip_url)
312
        is_remote = parse_result.scheme in ["http", "https"]
313

314
        if is_remote:
315
            if FreeCAD.GuiUp:
316
                self._run_zip_downloader_in_event_loop(zip_url)
317
            else:
318
                zip_data = utils.blocking_get(zip_url)
319
                with tempfile.NamedTemporaryFile(delete=False) as f:
320
                    tempfile_name = f.name
321
                    f.write(zip_data)
322
                self._finalize_zip_installation(tempfile_name)
323
        else:
324
            self._finalize_zip_installation(zip_url)
325
        return True
326

327
    def _run_zip_downloader_in_event_loop(self, zip_url: str):
328
        """Runs the zip downloader in a private event loop. This function does not exit until the
329
        ZIP download is complete. It requires the GUI to be up, and should not be run on the main
330
        GUI thread."""
331
        NetworkManager.AM_NETWORK_MANAGER.progress_made.connect(self._update_zip_status)
332
        NetworkManager.AM_NETWORK_MANAGER.progress_complete.connect(self._finish_zip)
333
        self.zip_download_index = NetworkManager.AM_NETWORK_MANAGER.submit_monitored_get(zip_url)
334
        while self.zip_download_index is not None:
335
            if QtCore.QThread.currentThread().isInterruptionRequested():
336
                break
337
            QtCore.QCoreApplication.processEvents(QtCore.QEventLoop.AllEvents, 50)
338

339
    def _update_zip_status(self, index: int, bytes_read: int, data_size: int):
340
        """Called periodically when downloading a zip file, emits a signal to display the
341
        download progress."""
342
        if index == self.zip_download_index:
343
            self.progress_update.emit(bytes_read, data_size)
344

345
    def _finish_zip(self, index: int, response_code: int, filename: os.PathLike):
346
        """Once the zip download is finished, unzip it into the correct location. Only called if
347
        the GUI is up, and the NetworkManager was responsible for the download. Do not call
348
        directly."""
349
        if index != self.zip_download_index:
350
            return
351
        self.zip_download_index = None
352
        if response_code != 200:
353
            self.failure.emit(
354
                self.addon_to_install,
355
                translate("AddonsInstaller", "Received {} response code from server").format(
356
                    response_code
357
                ),
358
            )
359
            return
360
        QtCore.QCoreApplication.processEvents(QtCore.QEventLoop.AllEvents)
361

362
        FreeCAD.Console.PrintLog("ZIP download complete. Installing...\n")
363
        self._finalize_zip_installation(filename)
364

365
    def _finalize_zip_installation(self, filename: os.PathLike):
366
        """Given a path to a zipfile, extract that file and put its contents in the correct
367
        location. Has special handling for GitHub's zip structure, which places the data in a
368
        subdirectory of the main directory."""
369

370
        destination = os.path.join(self.installation_path, self.addon_to_install.name)
371
        with zipfile.ZipFile(filename, "r") as zfile:
372
            zfile.extractall(destination)
373

374
        # GitHub (and possibly other hosts) put all files in the zip into a subdirectory named
375
        # after the branch. If that is the setup that we just extracted, move all files out of
376
        # that subdirectory.
377
        if self._code_in_branch_subdirectory(destination):
378
            actual_path = os.path.join(
379
                destination, f"{self.addon_to_install.name}-{self.addon_to_install.branch}"
380
            )
381
            FreeCAD.Console.PrintLog(
382
                f"ZIP installation moving code from {actual_path} to {destination}"
383
            )
384
            self._move_code_out_of_subdirectory(destination)
385

386
        FreeCAD.Console.PrintLog("ZIP installation complete.\n")
387
        self._finalize_successful_installation()
388

389
    def _code_in_branch_subdirectory(self, destination: str) -> bool:
390
        test_path = os.path.join(destination, self._expected_subdirectory_name())
391
        FreeCAD.Console.PrintLog(f"Checking for possible zip sub-path {test_path}...")
392
        if os.path.isdir(test_path):
393
            FreeCAD.Console.PrintLog(f"path exists.\n")
394
            return True
395
        FreeCAD.Console.PrintLog(f"path does not exist.\n")
396
        return False
397

398
    def _expected_subdirectory_name(self) -> str:
399
        url = self.addon_to_install.url
400
        if url.endswith(".git"):
401
            url = url[:-4]
402
        _, _, name = url.rpartition("/")
403
        branch = self.addon_to_install.branch
404
        return f"{name}-{branch}"
405

406
    def _move_code_out_of_subdirectory(self, destination):
407
        subdirectory = os.path.join(destination, self._expected_subdirectory_name())
408
        for extracted_filename in os.listdir(os.path.join(destination, subdirectory)):
409
            shutil.move(
410
                os.path.join(destination, subdirectory, extracted_filename),
411
                os.path.join(destination, extracted_filename),
412
            )
413
        os.rmdir(os.path.join(destination, subdirectory))
414

415
    def _finalize_successful_installation(self):
416
        """Perform any necessary additional steps after installing the addon."""
417
        self._update_metadata()
418
        self._install_macros()
419
        self.success.emit(self.addon_to_install)
420

421
    def _update_metadata(self):
422
        """Loads the package metadata from the Addon's downloaded package.xml file."""
423
        package_xml = os.path.join(
424
            self.installation_path, self.addon_to_install.name, "package.xml"
425
        )
426

427
        if hasattr(self.addon_to_install, "metadata") and os.path.isfile(package_xml):
428
            self.addon_to_install.load_metadata_file(package_xml)
429
            self.addon_to_install.installed_version = self.addon_to_install.metadata.version
430
            self.addon_to_install.updated_timestamp = os.path.getmtime(package_xml)
431

432
    def _install_macros(self):
433
        """For any workbenches, copy FCMacro files into the macro directory. Exclude packages that
434
        have preference packs, otherwise we will litter the macro directory with the pre and post
435
        scripts."""
436
        if (
437
            isinstance(self.addon_to_install, Addon)
438
            and self.addon_to_install.contains_preference_pack()
439
        ):
440
            return
441

442
        if not os.path.exists(self.macro_installation_path):
443
            os.makedirs(self.macro_installation_path)
444

445
        installed_macro_files = []
446
        for root, _, files in os.walk(
447
            os.path.join(self.installation_path, self.addon_to_install.name)
448
        ):
449
            for f in files:
450
                if f.lower().endswith(".fcmacro"):
451
                    src = os.path.join(root, f)
452
                    dst = os.path.join(self.macro_installation_path, f)
453
                    shutil.copy2(src, dst)
454
                    installed_macro_files.append(dst)
455
        if installed_macro_files:
456
            with open(
457
                os.path.join(
458
                    self.installation_path,
459
                    self.addon_to_install.name,
460
                    "AM_INSTALLATION_DIGEST.txt",
461
                ),
462
                "a",
463
                encoding="utf-8",
464
            ) as f:
465
                now = datetime.now(timezone.utc)
466
                f.write(
467
                    "# The following files were created outside this installation "
468
                    f"path during the installation of this Addon on {now}:\n"
469
                )
470
                for fcmacro_file in installed_macro_files:
471
                    f.write(fcmacro_file + "\n")
472

473
    @classmethod
474
    def _validate_object(cls, addon: object):
475
        """Make sure the object has the necessary attributes (name, url, and branch) to be
476
        installed."""
477

478
        if not hasattr(addon, "name") or not hasattr(addon, "url") or not hasattr(addon, "branch"):
479
            raise RuntimeError(
480
                "Provided object does not provide a name, url, and/or branch attribute"
481
            )
482

483

484
class MacroInstaller(QtCore.QObject):
485
    """Install a macro."""
486

487
    # Signals: success and failure
488
    # Emitted when the installation process is complete. The object emitted is the object that the
489
    # installation was requested for (usually of class Addon, but any class that provides a macro
490
    # can be used).
491
    success = QtCore.Signal(object)
492
    failure = QtCore.Signal(object)
493

494
    # Finished: regardless of the outcome, this is emitted when all work that is going to be done
495
    # is done (i.e. whatever thread this is running in can quit).
496
    finished = QtCore.Signal()
497

498
    def __init__(self, addon: object):
499
        """The provided addon object must have an attribute called "macro", and that attribute must
500
        itself provide a callable "install" method that takes a single string, the path to the
501
        installation location."""
502
        super().__init__()
503
        self._validate_object(addon)
504
        self.addon_to_install = addon
505
        self.installation_path = FreeCAD.getUserMacroDir(True)
506

507
    def run(self) -> bool:
508
        """Install a macro. Returns True if the macro was installed, or False if not. Emits
509
        either success or failure prior to returning."""
510

511
        # To try to ensure atomicity, perform the installation into a temp directory
512
        macro = self.addon_to_install.macro
513
        with tempfile.TemporaryDirectory() as temp_dir:
514
            temp_install_succeeded, error_list = macro.install(temp_dir)
515
            if not temp_install_succeeded:
516
                FreeCAD.Console.PrintError(
517
                    translate("AddonsInstaller", "Failed to install macro {}").format(macro.name)
518
                    + "\n"
519
                )
520
                for e in error_list:
521
                    FreeCAD.Console.PrintError(e + "\n")
522
                self.failure.emit(self.addon_to_install, "\n".join(error_list))
523
                self.finished.emit()
524
                return False
525

526
            # If it succeeded, move all the files to the macro install location,
527
            # keeping a list of all the files we installed, so they can be removed later
528
            # if this macro is uninstalled.
529
            manifest = []
530
            for item in os.listdir(temp_dir):
531
                src = os.path.join(temp_dir, item)
532
                dst = os.path.join(self.installation_path, item)
533
                shutil.move(src, dst)
534
                manifest.append(dst)
535
            self._write_installation_manifest(manifest)
536
        self.success.emit(self.addon_to_install)
537
        self.addon_to_install.set_status(Addon.Status.NO_UPDATE_AVAILABLE)
538
        self.finished.emit()
539
        return True
540

541
    def _write_installation_manifest(self, manifest):
542
        manifest_file = os.path.join(
543
            self.installation_path, self.addon_to_install.macro.filename + ".manifest"
544
        )
545
        try:
546
            with open(manifest_file, "w", encoding="utf-8") as f:
547
                f.write(json.dumps(manifest, indent="  "))
548
        except OSError as e:
549
            FreeCAD.Console.PrintWarning(
550
                translate("AddonsInstaller", "Failed to create installation manifest " "file:\n")
551
            )
552
            FreeCAD.Console.PrintWarning(manifest_file)
553

554
    @classmethod
555
    def _validate_object(cls, addon: object):
556
        """Make sure this object provides an attribute called "macro" with a method called
557
        "install" """
558
        if (
559
            not hasattr(addon, "macro")
560
            or addon.macro is None
561
            or not hasattr(addon.macro, "install")
562
            or not callable(addon.macro.install)
563
        ):
564
            raise RuntimeError("Provided object does not provide a macro with an install method")
565