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

# SPDX-License-Identifier: BSD-2-Clause
# SPDX-FileCopyrightText: 2023 KDE e.V.
# SPDX-FileCopyrightText: 2023 Ben Cooksley <bcooksley@kde.org>
# SPDX-FileCopyrightText: 2024 Julius Künzel <julius.kuenzel@kde.org>
# SPDX-FileContributor: Ingo Klöcker <dev@ingo-kloecker.de>

import dataclasses
import logging
import os
import tarfile
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import Optional

from pkginfo import BDist, SDist, Wheel

from sftpnotary import config, projects, util
from sftpnotary.core import Job, Request, Response, TaskProcessor, Worker
from sftpnotary.exceptions import Error, InvalidParameter
from sftpnotary.sftp import SFTPClient, SFTPError

log = logging.getLogger(__name__)


@dataclasses.dataclass
class PublishPyPiRequest(Request):
    archiveFileName: Optional[str] = None
    repository: Optional[str] = None

    def validate(self):
        util.validateFileName(self.archiveFileName, strict=True)

        # Ensure the repository name is valid
        validRepoNames = [
            "pypi",
            "testpypi",
        ]

        if self.repository not in validRepoNames:
            raise InvalidParameter(name="repository", value=self.repository)


PublishPyPiResponse = Response


class PublishPyPiJob(Job):
    def __init__(
        self,
        sftp: SFTPClient,
        archiveFilePath: Path,
        token: Optional[str] = None,
        projectPath: Optional[str] = None,
        branch: Optional[str] = None,
        repository: Optional[str] = None,
    ):
        super().__init__(sftp)

        self.archiveFilePath = archiveFilePath
        self.request = PublishPyPiRequest(
            archiveFileName=None, token=token, projectPath=projectPath, branch=branch, repository=repository
        )
        self.resultPath = self.archiveFilePath.parent

    def start(self):
        archiveFileName = self.archiveFilePath.name
        self.sftp.upload(self.archiveFilePath, self.task.path() / archiveFileName)
        self.request.archiveFileName = archiveFileName
        self.task.putRequest(self.request)
        self.taskQueue.commitTask(self.task)

    def waitForCompletion(self, logFileDestination: Path):
        self.taskQueue.waitForCompletion(self.task)
        log.debug(f"Fetching response of task '{self.task.id}' ...")
        response = None
        try:
            response = self.task.fetchResponse(PublishPyPiResponse)
        except SFTPError as e:
            log.error("Error: %s", e)
        if response is None:
            log.error("Error: Got no response. Task failed.")
        else:
            self.success = True
        self.downloadLogFiles(logFileDestination)
        self.taskQueue.removeTask(self.task)


def getPythonPackageName(pkgpath):
    if str(pkgpath).endswith(".tar.gz"):
        dist = SDist
    elif str(pkgpath).endswith(".whl"):
        dist = Wheel
    elif str(pkgpath).endswith(".egg"):
        dist = BDist
    else:
        return None
    pkg = dist(pkgpath)

    return pkg.name


class PublishPyPiProcessor(TaskProcessor):
    requestClass = PublishPyPiRequest

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.configFile = config.settings.getPath("PyPiPublishing", "ConfigFilePath")

        if not self.configFile:
            raise Error("No config file has not been set but it is mandatory")

    def getProjectSetting(self, name, required=True):
        value = projects.settings.get(self.request.projectPath, self.request.branch, name)
        if required and not value:
            raise Error(
                f"Project setting '{name}' not set for branch {self.request.branch} "
                f"of project {self.request.projectPath}"
            )
        return value

    def doProcess(self):
        with TemporaryDirectory() as workPathName:
            workPath = Path(workPathName)

            repository = self.request.repository

            localArchiveFilePath = workPath / self.request.archiveFileName
            self.sftp.download(self.task.path() / self.request.archiveFileName, localArchiveFilePath)

            # Setup a path for us to unpack the archive into...
            unpackedFilesPath = workPath / "unpacked-files"
            os.makedirs(unpackedFilesPath)

            # Unpack the archive....
            archive = tarfile.open(name=localArchiveFilePath, mode="r")
            # Set "nosec B202" because of https://github.com/PyCQA/bandit/issues/1038#issuecomment-1635863856
            archive.extractall(filter="tar", path=unpackedFilesPath)  # nosec B202
            archive.close()

            self.verifyPackageNameDir(unpackedFilesPath)

            # Build up the command to run then run it
            command = [
                "python3",
                "-m",
                "twine",
                "upload",
                "--non-interactive",
                "--disable-progress-bar",
                "--verbose",
                "--config-file",
                self.configFile,
                "--repository",
                repository,
                str(unpackedFilesPath) + "/*",
            ]

            util.runCommand(command, cwd=workPathName)

    def verifyPackageNameDir(self, dirPath):
        for root, dirs, files in os.walk(dirPath):
            for fileName in files:
                self.verifyPackageNameFile(Path(root) / fileName)

    def verifyPackageNameFile(self, packageFilePath):
        packageName = getPythonPackageName(packageFilePath)
        allowedPackageNames = self.getProjectSetting("packagename")
        if isinstance(allowedPackageNames, str):
            allowedPackageNames = [allowedPackageNames]
        if packageName not in allowedPackageNames:
            raise Error(
                f"The package name of {packageFilePath.name} ({packageName}) did not match any of "
                f"the allowed package names ({', '.join(allowedPackageNames)})."
            )


class PublishPyPiWorker(Worker):
    processorClass = PublishPyPiProcessor