aws_lambda_builders/workflows/dotnet_clipackage/utils.py

"""
Commonly used utilities
"""

import logging
import os
import platform
import subprocess
import zipfile

from aws_lambda_builders.utils import decode, which

LOG = logging.getLogger(__name__)


class OSUtils(object):
    """
    Convenience wrapper around common system functions
    """

    def popen(self, command, stdout=None, stderr=None, env=None, cwd=None):
        p = subprocess.Popen(command, stdout=stdout, stderr=stderr, env=env, cwd=cwd)
        return p

    def is_windows(self):
        return platform.system().lower() == "windows"

    def which(self, executable, executable_search_paths=None):
        return which(executable, executable_search_paths=executable_search_paths)

    def unzip(self, zip_file_path, output_dir, permission=None):
        """
        This method and dependent methods were copied from SAM CLI, but with the addition of deleting the zip file
        https://github.com/aws/aws-sam-cli/blob/458076265651237a662a372f54d5b3df49fd6797/samcli/local/lambdafn/zip.py#L81

        Unzip the given file into the given directory while preserving file permissions in the process.
        Parameters
        ----------
        zip_file_path : str
            Path to the zip file
        output_dir : str
            Path to the directory where the it should be unzipped to
        permission : int
            Permission to set in an octal int form
        """

        with zipfile.ZipFile(zip_file_path, "r") as zip_ref:
            # For each item in the zip file, extract the file and set permissions if available
            for file_info in zip_ref.infolist():
                extracted_path = self._extract(file_info, output_dir, zip_ref)

                # If the extracted_path is a symlink, do not set the permissions. If the target of the symlink does not
                # exist, then os.chmod will fail with FileNotFoundError
                if not os.path.islink(extracted_path):
                    self._set_permissions(file_info, extracted_path)
                    self._override_permissions(extracted_path, permission)

        if not os.path.islink(extracted_path):
            self._override_permissions(output_dir, permission)

        os.remove(zip_file_path)

    def _is_symlink(self, file_info):
        """
        Check the upper 4 bits of the external attribute for a symlink.
        See: https://unix.stackexchange.com/questions/14705/the-zip-formats-external-file-attribute
        Parameters
        ----------
        file_info : zipfile.ZipInfo
            The ZipInfo for a ZipFile
        Returns
        -------
        bool
            A response regarding whether the ZipInfo defines a symlink or not.
        """
        symlink = 0xA
        return (file_info.external_attr >> 28) == symlink

    def _extract(self, file_info, output_dir, zip_ref):
        """
        Unzip the given file into the given directory while preserving file permissions in the process.
        Parameters
        ----------
        file_info : zipfile.ZipInfo
            The ZipInfo for a ZipFile
        output_dir : str
            Path to the directory where the it should be unzipped to
        zip_ref : zipfile.ZipFile
            The ZipFile we are working with.
        Returns
        -------
        string
            Returns the target path the Zip Entry was extracted to.
        """

        # Handle any regular file/directory entries
        if not self._is_symlink(file_info):
            return zip_ref.extract(file_info, output_dir)

        source = decode(zip_ref.read(file_info.filename))
        link_name = os.path.normpath(os.path.join(output_dir, file_info.filename))

        # make leading dirs if needed
        leading_dirs = os.path.dirname(link_name)
        if not os.path.exists(leading_dirs):
            os.makedirs(leading_dirs)

        # If the link already exists, delete it or symlink() fails
        if os.path.lexists(link_name):
            os.remove(link_name)

        # Create a symbolic link pointing to source named link_name.
        os.symlink(source, link_name)

        return link_name

    def _override_permissions(self, path, permission):
        """
        Forcefully override the permissions on the path
        Parameters
        ----------
        path str
            Path where the file or directory
        permission octal int
            Permission to set
        """
        if permission:
            os.chmod(path, permission)

    def _set_permissions(self, zip_file_info, extracted_path):
        """
        Sets permissions on the extracted file by reading the ``external_attr`` property of given file info.
        Parameters
        ----------
        zip_file_info : zipfile.ZipInfo
            Object containing information about a file within a zip archive
        extracted_path : str
            Path where the file has been extracted to
        """

        # Permission information is stored in first two bytes.
        permission = zip_file_info.external_attr >> 16
        if not permission:
            # Zips created on certain Windows machines, however, might not have any permission information on them.
            # Skip setting a permission on these files.
            LOG.debug("File %s in zipfile does not have permission information", zip_file_info.filename)
            return

        os.chmod(extracted_path, permission)

    @property
    def pipe(self):
        return subprocess.PIPE