autopkg/Code/autopkglib/Installer.py

#!/usr/local/autopkg/python
#
# Copyright 2014 Greg Neagle
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""See docstring for Installer class"""

import os.path
import plistlib
import socket
from glob import glob

from autopkglib import ProcessorError
from autopkglib.DmgMounter import DmgMounter

AUTOPKGINSTALLD_SOCKET = "/var/run/autopkginstalld"


__all__ = ["Installer"]


class Installer(DmgMounter):
    """Calls autopkginstalld to install a package."""

    description = __doc__
    input_variables = {
        "pkg_path": {
            "required": True,
            "description": (
                "Path to the package to be installed. Can be inside a disk image."
            ),
        },
        "new_package_request": {
            "required": False,
            "description": (
                "new_package_request is set by the PkgCreator processor to "
                "indicate that a new package was built. If this key is set in "
                "the environment and is False or empty the installation will "
                "be skipped."
            ),
        },
        "download_changed": {
            "required": False,
            "description": (
                "download_changed is set by the URLDownloader processor to "
                "indicate that a new file was downloaded. If this key is set "
                "in the environment and is False or empty the installation "
                "will be skipped, unless new_package_request is non-False."
            ),
        },
    }
    output_variables = {
        "install_result": {"description": "Result of install request."},
        "installer_summary_result": {
            "description": "Description of interesting results."
        },
    }

    def install(self):
        """Build an installation request, send it to autopkginstalld"""
        # clear any pre-existing summary result
        if "installer_summary_result" in self.env:
            del self.env["installer_summary_result"]

        if "new_package_request" in self.env:
            if not self.env["new_package_request"]:
                # PkgCreator did not build a new package, so skip the install
                self.output("Skipping installation: no new package.")
                self.env["install_result"] = "SKIPPED"
                return
        elif "download_changed" in self.env:
            if not self.env["download_changed"]:
                # URLDownloader did not download something new,
                # so skip the install
                self.output("Skipping installation: no new download.")
                self.env["install_result"] = "SKIPPED"
                return

        pkg_path = self.env["pkg_path"]
        # Check if we're trying to copy something inside a dmg.
        (dmg_path, dmg, dmg_pkg_path) = self.parsePathForDMG(pkg_path)
        try:
            if dmg:
                # Mount dmg and copy path inside.
                mount_point = self.mount(dmg_path)
                pkg_path = os.path.join(mount_point, dmg_pkg_path)
            # process path with glob.glob
            matches = glob(pkg_path)
            if len(matches) == 0:
                raise ProcessorError(f"Error processing path '{pkg_path}' with glob. ")
            matched_pkg_path = matches[0]
            if len(matches) > 1:
                self.output(
                    f"WARNING: Multiple paths match 'pkg_path' glob '{pkg_path}':"
                )
                for match in matches:
                    self.output(f"  - {match}")

            if [c for c in "*?[]!" if c in pkg_path]:
                self.output(
                    f"Using path '{matched_pkg_path}' matched from globbed "
                    f"'{pkg_path}'."
                )

            request = {"package": matched_pkg_path}
            result = None
            # Send install request.
            try:
                self.output("Connecting")
                self.connect()
                self.output("Sending installation request")
                result = self.send_request(request)
            except Exception as err:
                result = f"ERROR: {repr(err)}"
            finally:
                self.output("Disconnecting")
                try:
                    self.disconnect()
                except OSError:
                    pass
            # Return result.
            self.output(f"Result: {result}")
            self.env["install_result"] = result
            if result == "DONE":
                self.env["installer_summary_result"] = {
                    "summary_text": ("The following pkgs were successfully installed:"),
                    "data": {"pkg_path": matched_pkg_path},
                }

        finally:
            if dmg:
                self.unmount(dmg_path)

    def connect(self):
        """Connect to autopkginstalld"""
        try:
            self.socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
            self.socket.connect(AUTOPKGINSTALLD_SOCKET)
        except OSError as err:
            raise ProcessorError(
                f"Couldn't connect to autopkginstalld: {err.strerror}"
            ) from err

    def send_request(self, request):
        """Send an install request to autopkginstalld"""
        self.socket.send(plistlib.dumps(request))
        with os.fdopen(self.socket.fileno()) as fileref:
            while True:
                data = fileref.readline()
                if data:
                    if data.startswith("OK:"):
                        return data.replace("OK:", "").rstrip()
                    elif data.startswith("ERROR:"):
                        break
                    else:
                        self.output(data.rstrip())
                else:
                    break

        errors = data.rstrip().split("\n")
        if not errors:
            errors = ["ERROR:No reply from autopkginstalld (crash?), check system logs"]
        raise ProcessorError(", ".join([s.replace("ERROR:", "") for s in errors]))

    def disconnect(self):
        """Disconnect from autopkginstalld"""
        self.socket.close()

    def main(self):
        """Install something!"""
        self.install()


if __name__ == "__main__":
    PROCESSOR = Installer()
    PROCESSOR.execute_shell()