138 lines
5.5 KiB
Python
138 lines
5.5 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__)
|
|
|
|
SignApkRequest = SingleFileRequest
|
|
SignApkResponse = SignFileResponse
|
|
|
|
|
|
class SignApkJob(SignSingleFileJob):
|
|
def __init__(
|
|
self,
|
|
sftp: SFTPClient,
|
|
apkFilePath: Path,
|
|
token: Optional[str] = None,
|
|
projectPath: Optional[str] = None,
|
|
branch: Optional[str] = None,
|
|
):
|
|
super().__init__(sftp, apkFilePath)
|
|
|
|
self.request = SignApkRequest(fileName=apkFilePath.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(SignApkResponse)
|
|
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 SignApkProcessor(TaskProcessor):
|
|
requestClass = SignApkRequest
|
|
|
|
def doProcess(self):
|
|
remoteApkFilePath = 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
|
|
localApkFilePath = workPath / "unsigned.apk"
|
|
localSignedApkFileName = "signed.apk"
|
|
self.sftp.download(remoteApkFilePath, localApkFilePath)
|
|
self.verifyApplicationId(localApkFilePath)
|
|
self.signApk(localApkFilePath, localSignedApkFileName)
|
|
try:
|
|
# we assume a single suffix
|
|
remoteSignedApkFileName = f"{remoteApkFilePath.stem}-signed{remoteApkFilePath.suffix}"
|
|
self.sftp.upload(workPath / localSignedApkFileName, self.task.path() / remoteSignedApkFileName)
|
|
return SignApkResponse(signedFileName=remoteSignedApkFileName)
|
|
except Error as e:
|
|
log.error("Error: %s", e)
|
|
except SFTPError as e:
|
|
log.error("Error: %s", e)
|
|
|
|
def verifyApplicationId(self, apkFilePath):
|
|
applicationId = apkutils.getApplicationId(apkFilePath)
|
|
allowedApplicationId = projects.settings.get(self.request.projectPath, self.request.branch, "applicationid")
|
|
if applicationId != allowedApplicationId:
|
|
raise Error(
|
|
f"The application ID of the APK ({applicationId}) did not match the allowed application ID "
|
|
f"({allowedApplicationId})."
|
|
)
|
|
|
|
def signApk(self, apkFilePath, signedApkFileName):
|
|
workingDirectory = apkFilePath.parent
|
|
apkFileName = apkFilePath.name
|
|
alignedApkFileName = "aligned.apk"
|
|
sdkToolsPath = config.settings.getPath("ApkSigning", "SDKToolsPath")
|
|
|
|
zipalignPath = Path(sdkToolsPath) / "zipalign" if sdkToolsPath else "zipalign"
|
|
command = [
|
|
zipalignPath,
|
|
"-v", # verbose
|
|
"-p", # page-align uncompressed .so files
|
|
"4", # align other files to 4 bytes
|
|
apkFileName,
|
|
alignedApkFileName,
|
|
]
|
|
util.runCommand(command, cwd=workingDirectory)
|
|
|
|
keystore = projects.settings.getPath(
|
|
self.request.projectPath, self.request.branch, "keystore", config.settings.getPath("ApkSigning", "KeyStore")
|
|
)
|
|
keystorepass = projects.settings.get(
|
|
self.request.projectPath,
|
|
self.request.branch,
|
|
"keystorepass",
|
|
config.settings.get("ApkSigning", "KeyStorePass"),
|
|
)
|
|
if keystorepass.startswith("file:~"):
|
|
keystorepass = f"file:{Path(keystorepass[5:]).expanduser()}"
|
|
apksignerPath = Path(sdkToolsPath) / "apksigner" if sdkToolsPath else "apksigner"
|
|
command = [apksignerPath, "sign", "-v"]
|
|
if keystore:
|
|
command += ["--ks", Path(keystore)]
|
|
if keystorepass:
|
|
command += ["--ks-pass", keystorepass]
|
|
command += ["--out", signedApkFileName, alignedApkFileName]
|
|
commandToLog = sftpnotary.log.maskFollowingItem(command, "--ks-pass")
|
|
util.runCommand(command, commandToLog=commandToLog, cwd=workingDirectory)
|
|
|
|
|
|
class SignApkWorker(Worker):
|
|
processorClass = SignApkProcessor
|