OpenMiInk/third_party/ci-notary-service/sftpnotary/appleutils.py

# SPDX-License-Identifier: BSD-2-Clause
# SPDX-FileCopyrightText: 2023 KDE e.V.
# SPDX-FileContributor: Ingo Klöcker <dev@ingo-kloecker.de>
# SPDX-FileCopyrightText: 2023 Julius Künzel <jk.kdedev@smartlab.uber.space>

import logging
import plistlib
import tarfile
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import Type

from sftpnotary import config, util
from sftpnotary.core import SignFileResponse, SignSingleFileJob, Task, TaskProcessor
from sftpnotary.exceptions import Error
from sftpnotary.sftp import SFTPClient, SFTPError

log = logging.getLogger(__name__)


def getPlistApplicationId(plistFilePath: Path):
    with open(plistFilePath, "rb") as plfile:
        pl = plistlib.load(plfile)

    applicationId = pl["CFBundleIdentifier"]
    return applicationId


def getAppApplicationId(appFilePath: Path):
    plistFile = appFilePath / "Contents" / "Info.plist"

    return getPlistApplicationId(plistFile)


def _extractAppFoldersWithPlistFile(dmgFilePath: Path, extractTo: Path, appGlob: str):
    workingDirectory = dmgFilePath.parent
    SevenZPath = config.settings.getPath("AppleTools", "SevenZipPath") or "7zz"

    command = [
        SevenZPath,
        "x",
        f"-i!{appGlob}/*/*.plist",
        f"-o{extractTo}",
        dmgFilePath,
    ]
    util.runCommand(command, cwd=workingDirectory)

    return sorted(extractTo.glob(appGlob))


def getDmgApplicationId(dmgFilePath: Path):
    with TemporaryDirectory() as tempPathName:
        appLocations = []
        appLocations.extend(
            _extractAppFoldersWithPlistFile(dmgFilePath, Path(tempPathName) / "extracteddmgapp", appGlob="*/*.app")
        )
        appLocations.extend(
            _extractAppFoldersWithPlistFile(dmgFilePath, Path(tempPathName) / "extractedapp", appGlob="*.app")
        )

        if len(appLocations) > 1:
            raise Error("More than one *.app was found.")
        elif not appLocations:
            raise Error("No *.app with a *.plist was found to verify the application id of the DMG.")

        return getAppApplicationId(appLocations[0])


def getApplicationId(filePath: Path):
    if filePath.suffix == ".app":
        return getAppApplicationId(filePath)
    elif filePath.suffix == ".dmg":
        return getDmgApplicationId(filePath)
    else:
        raise Error(
            f"File {filePath} has unsupported file extension {filePath.suffix}. Can not identify application id"
        )


def verifyApplicationId(filePath: Path, allowedApplicationId: str):
    applicationId = getApplicationId(filePath)

    if not applicationId:
        raise Error(
            "The application ID is empty, which may indicate that required "
            "MACOSX_BUNDLE_* properties are not set in the associated cmake target."
        )
    elif applicationId != allowedApplicationId:
        raise Error(
            f"The application ID of {filePath.name} ({applicationId}) did not match the allowed application ID "
            f"({allowedApplicationId}).\nYou should compare the MACOSX_BUNDLE_GUI_IDENTIFIER "
            "cmake target property set in the associated project with those at "
            "https://invent.kde.org/sysadmin/ci-utilities/-/tree/master/signing"
        )


class AppleJob(SignSingleFileJob):
    def __init__(self, sftp: SFTPClient, macAppFilePath: Path, logFolder: Path):
        super().__init__(sftp, macAppFilePath)
        self.logFolder = logFolder

    def start(self):
        self.uploadUnsignedFile()
        self.task.putRequest(self.request)
        self.taskQueue.commitTask(self.task)

    def waitForCompletion(self):
        self.taskQueue.waitForCompletion(self.task)
        log.debug(f"Fetching response of task '{self.task.id}' ...")
        response = None
        try:
            response = self.task.fetchResponse(self.responseClass)
        except SFTPError as e:
            log.error("Error: %s", e)
        if response is None:
            log.error("Error: Got no response. Task failed.")
        elif not response.signedFileName:
            log.error("Error: No signed file to download.")
        elif "/" in response.signedFileName:
            log.error(f"Error: Suspicious file name of signed file: {response.signedFileName!r}.")
        else:
            try:
                remotePath = self.task.path() / response.signedFileName
                localPath = self.localFilePath  # overwrite original (unsigned) file
                self.sftp.download(remotePath, localPath)
                self.success = True
            except SFTPError as e:
                log.error("Error: %s", e)
        self.downloadLogFiles(self.logFolder)
        self.taskQueue.removeTask(self.task)


class AppleTaskProcessor(TaskProcessor):
    resultNameSuffix = ""
    responseClass: Type[SignFileResponse]

    def extractAppFromTar(self, workPath: Path, archiveFilePath: Path):
        # unpack archive
        util.extractTarArchive(archiveFilePath, workPath)
        appLocations = sorted(workPath.glob("*.app"))

        if len(appLocations) > 1:
            raise Error(f"More than one ({len(appLocations)}) *.app was found inside archive.")
        elif not appLocations:
            raise Error("No *.app was found inside the archive.")

        return appLocations[0]

    def downloadApp(self, workPath: Path, remoteFilePath: Path):
        fileExtension = util.getFileSuffix(remoteFilePath)
        if fileExtension == ".dmg":
            # use simple file names locally to avoid issues with user provided file names
            localFilePath = workPath / f"original{fileExtension}"
            self.sftp.download(remoteFilePath, localFilePath)
        else:
            archiveFilePath = workPath / f"archive{fileExtension}"
            self.sftp.download(remoteFilePath, archiveFilePath)
            if not tarfile.is_tarfile(archiveFilePath):
                raise Error(
                    f"The file proposed for signing or notarization has the extension {fileExtension} "
                    "and is no archive. Only .dmg and .app (inside tar archive) are supported."
                )

            localFilePath = self.extractAppFromTar(workPath, archiveFilePath)
        return localFilePath

    def uploadApp(self, workPath: Path, localProcessedFilePath: Path, originalRemotePath: Path, localOrignalPath: Path):
        self.task: Task
        fileExtension = util.getFileSuffix(originalRemotePath)
        if localProcessedFilePath.suffix == ".app":
            # repack archive
            signedArchivePath = workPath / f"archive-{self.resultNameSuffix}{fileExtension}"
            with tarfile.open(name=signedArchivePath, mode="w") as tar:
                # Add our signed/notarized *.app to an archive. We use the name of the orignal app
                # (eg. kdenlive.app) instead of the local work name "processed.app".
                # For other file types we use a suffix like "kdenlive-signed.app",
                # for *.app files we do the same, but to the archive and not the archives content
                tar.add(localProcessedFilePath, arcname=localOrignalPath.name, recursive=True)
            localProcessedFilePath = signedArchivePath
        try:
            remoteSignedFileName = (
                f"{util.removeFileSuffix(originalRemotePath.name)}-{self.resultNameSuffix}{fileExtension}"
            )
            self.sftp.upload(localProcessedFilePath, self.task.path() / remoteSignedFileName)
            return self.responseClass(signedFileName=remoteSignedFileName)
        except Error as e:
            log.error("Error: %s", e)
        except SFTPError as e:
            log.error("Error: %s", e)