156 lines
6.3 KiB
Python
156 lines
6.3 KiB
Python
# SPDX-License-Identifier: BSD-2-Clause
|
|
# SPDX-FileCopyrightText: 2023 KDE e.V.
|
|
# SPDX-FileContributor: Ingo Klöcker <dev@ingo-kloecker.de>
|
|
|
|
import logging
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
from typing import Optional
|
|
|
|
import sftpnotary.log
|
|
from sftpnotary import apkutils, config, projects, util
|
|
from sftpnotary.core import SignFileResponse, SignSingleFileJob, SingleFileRequest, TaskProcessor, Worker
|
|
from sftpnotary.exceptions import Error
|
|
from sftpnotary.sftp import SFTPClient, SFTPError
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
SignAabRequest = SingleFileRequest
|
|
SignAabResponse = SignFileResponse
|
|
|
|
|
|
class SignAabJob(SignSingleFileJob):
|
|
def __init__(
|
|
self,
|
|
sftp: SFTPClient,
|
|
aabFilePath: Path,
|
|
token: Optional[str] = None,
|
|
projectPath: Optional[str] = None,
|
|
branch: Optional[str] = None,
|
|
):
|
|
super().__init__(sftp, aabFilePath)
|
|
|
|
self.request = SignAabRequest(fileName=aabFilePath.name, token=token, projectPath=projectPath, branch=branch)
|
|
|
|
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(SignAabResponse)
|
|
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.localFilePath.parent)
|
|
self.taskQueue.removeTask(self.task)
|
|
|
|
|
|
class SignAabBaseProcessor(TaskProcessor):
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self.jarsignerPath = config.settings.getPath("GooglePlayPublishing", "JarsignerPath")
|
|
|
|
def getProjectSetting(self, name, required=True, default=None):
|
|
value = projects.settings.get(self.request.projectPath, self.request.branch, name, default=default)
|
|
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 verifyApplicationId(self, filePaths, allowedApplicationId):
|
|
for filePath in filePaths:
|
|
applicationId = apkutils.getApplicationId(filePath)
|
|
if applicationId != allowedApplicationId:
|
|
raise Error(
|
|
f"The application ID of {filePath.name} ({applicationId}) did not match the allowed application ID "
|
|
f"({allowedApplicationId})."
|
|
)
|
|
|
|
def signAab(self, aabFilePath):
|
|
workingDirectory = aabFilePath.parent
|
|
aabFileName = aabFilePath.name
|
|
signedAabFileName = "signed.aab"
|
|
|
|
keystore = projects.settings.getPath(
|
|
self.request.projectPath,
|
|
self.request.branch,
|
|
"uploadkeystore",
|
|
config.settings.getPath("GooglePlayPublishing", "UploadKeyStore"),
|
|
)
|
|
if not keystore:
|
|
raise Error(
|
|
f"Project setting 'uploadkeystore' not set for branch {self.request.branch} "
|
|
f"of project {self.request.projectPath}"
|
|
)
|
|
keystorepass = self.getProjectSetting(
|
|
"uploadkeystorepass", default=config.settings.get("GooglePlayPublishing", "UploadKeyStorePass")
|
|
)
|
|
if keystorepass.startswith("file:~"):
|
|
keystorepass = f"file:{Path(keystorepass[5:]).expanduser()}"
|
|
keyalias = self.getProjectSetting(
|
|
"uploadkeyalias", default=config.settings.get("GooglePlayPublishing", "UploadKeyAlias")
|
|
)
|
|
|
|
command = [self.jarsignerPath, "-verbose", "-sigalg", "SHA256withRSA", "-digestalg", "SHA-256"]
|
|
command += ["-keystore", Path(keystore)]
|
|
if keystorepass.startswith("file:"):
|
|
command += ["-storepass:file", keystorepass[5:]]
|
|
elif keystorepass.startswith("env:"):
|
|
command += ["-storepass:env", keystorepass[4:]]
|
|
else:
|
|
command += ["-storepass", keystorepass]
|
|
command += ["-signedjar", signedAabFileName, aabFileName, keyalias]
|
|
commandToLog = sftpnotary.log.maskFollowingItem(command, "-storepass")
|
|
util.runCommand(command, commandToLog=commandToLog, cwd=workingDirectory)
|
|
|
|
return workingDirectory / signedAabFileName
|
|
|
|
|
|
class SignAabProcessor(SignAabBaseProcessor):
|
|
requestClass = SignAabRequest
|
|
|
|
def doProcess(self):
|
|
remoteAabFilePath = self.task.path() / self.request.fileName
|
|
with TemporaryDirectory() as workPathName:
|
|
workPath = Path(workPathName)
|
|
# use simple file names locally to avoid issues with user provided file names
|
|
localAabFilePath = workPath / "unsigned.aab"
|
|
self.sftp.download(remoteAabFilePath, localAabFilePath)
|
|
|
|
allowedApplicationId = projects.settings.get(self.request.projectPath, self.request.branch, "applicationid")
|
|
self.verifyApplicationId([localAabFilePath], allowedApplicationId)
|
|
localSignedAabFilePath = self.signAab(localAabFilePath)
|
|
|
|
try:
|
|
# we assume a single suffix
|
|
remoteSignedAabFileName = f"{remoteAabFilePath.stem}-signed{remoteAabFilePath.suffix}"
|
|
self.sftp.upload(localSignedAabFilePath, self.task.path() / remoteSignedAabFileName)
|
|
return SignAabResponse(signedFileName=remoteSignedAabFileName)
|
|
except Error as e:
|
|
log.error("Error: %s", e)
|
|
except SFTPError as e:
|
|
log.error("Error: %s", e)
|
|
|
|
|
|
class SignAabWorker(Worker):
|
|
processorClass = SignAabProcessor
|