340 lines
15 KiB
Python
340 lines
15 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
|
|
import os
|
|
import re
|
|
import shutil
|
|
from pathlib import Path
|
|
from tempfile import NamedTemporaryFile, TemporaryDirectory
|
|
from typing import Dict, List, Optional
|
|
from zipfile import ZipFile
|
|
|
|
import sftpnotary.log
|
|
from sftpnotary import config, projects, util
|
|
from sftpnotary.core import Job, SignFileResponse, SingleFileRequest, TaskProcessor, Worker
|
|
from sftpnotary.exceptions import Error
|
|
from sftpnotary.sftp import SFTPClient, SFTPError
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
SignWindowsBinariesRequest = SingleFileRequest
|
|
SignWindowsBinariesResponse = SignFileResponse
|
|
|
|
|
|
class SignWindowsBinariesJob(Job):
|
|
archiveName = "binaries.zip"
|
|
|
|
def __init__(
|
|
self,
|
|
sftp: SFTPClient,
|
|
filePaths: List[Path],
|
|
logFolder: Path,
|
|
token: Optional[str] = None,
|
|
projectPath: Optional[str] = None,
|
|
branch: Optional[str] = None,
|
|
):
|
|
super().__init__(sftp)
|
|
|
|
self.filePaths = filePaths
|
|
self.workingDirectory: Optional[TemporaryDirectory] = None
|
|
if len(self.filePaths) == 1:
|
|
self.request = SignWindowsBinariesRequest(
|
|
fileName=self.filePaths[0].name, token=token, projectPath=projectPath, branch=branch
|
|
)
|
|
else:
|
|
self.request = SignWindowsBinariesRequest(
|
|
fileName=self.archiveName, token=token, projectPath=projectPath, branch=branch
|
|
)
|
|
self.logFolder = logFolder
|
|
self.originalFilePaths: Dict[str, Path] = {}
|
|
|
|
def start(self):
|
|
if len(self.filePaths) == 1:
|
|
self.localFilePath = self.filePaths[0]
|
|
else:
|
|
self.workingDirectory = TemporaryDirectory()
|
|
self.localFilePath = Path(self.workingDirectory.name) / self.archiveName
|
|
self.packageBinaries(self.localFilePath)
|
|
|
|
self.sftp.upload(self.localFilePath, self.task.path() / self.request.fileName)
|
|
self.task.putRequest(self.request)
|
|
self.taskQueue.commitTask(self.task)
|
|
|
|
def packageBinaries(self, archivePath):
|
|
# put all files to sign into a zip archive; to avoid problems with paths
|
|
# and weird file names we use consecutive numbers with the original extension
|
|
# as names in the archive
|
|
with ZipFile(archivePath, "w") as zipFile:
|
|
for number, binaryPath in enumerate(self.filePaths, start=1):
|
|
arcName = str(number) + binaryPath.suffix
|
|
self.originalFilePaths[arcName] = binaryPath
|
|
log.debug(f"Adding {binaryPath} as {arcName} to {archivePath.name}")
|
|
zipFile.write(binaryPath, arcname=arcName)
|
|
|
|
def extractSignedBinaries(self, archivePath):
|
|
log.debug(f"Extracting signed binaries from {archivePath.name}")
|
|
# ZipFile cannot extract a file with another filename than the one listed
|
|
# in the archive; therefore, we extract all expected files to our working
|
|
# directory (which only contains the archive) and then we "move" the
|
|
# extracted signed binaries to their original locations
|
|
extractPath = Path(self.workingDirectory.name)
|
|
with ZipFile(archivePath, "r") as zipFile:
|
|
filesToExtract = [arcName for arcName in zipFile.namelist() if arcName in self.originalFilePaths]
|
|
zipFile.extractall(path=extractPath, members=filesToExtract)
|
|
for arcName in filesToExtract:
|
|
originalFilePath = self.originalFilePaths[arcName]
|
|
log.debug(f"Moving {arcName} to {originalFilePath}")
|
|
# use shutil.copyfile() + os.unlink() because there's no sane way to do a 'mv'
|
|
# that overwrites existing files and that works across filesystem boundaries
|
|
shutil.copyfile(extractPath / arcName, originalFilePath)
|
|
os.unlink(extractPath / arcName)
|
|
|
|
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(SignWindowsBinariesResponse)
|
|
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)
|
|
|
|
if len(self.filePaths) > 1:
|
|
self.extractSignedBinaries(self.localFilePath)
|
|
|
|
self.success = True
|
|
except SFTPError as e:
|
|
log.error("Error: %s", e)
|
|
if self.workingDirectory is not None:
|
|
self.workingDirectory.cleanup()
|
|
self.downloadLogFiles(self.logFolder)
|
|
self.taskQueue.removeTask(self.task)
|
|
|
|
|
|
# expected file names are a number optionally followed by a suffix;
|
|
# in particular, they have no path components
|
|
_expectedFileNamesRegExp = re.compile(r"\d+(?:\.\w*)", re.ASCII)
|
|
|
|
|
|
def filterBinaries(fileNames: List[str]) -> List[str]:
|
|
return [fileName for fileName in fileNames if _expectedFileNamesRegExp.fullmatch(fileName)]
|
|
|
|
|
|
def extractUnsignedBinaries(archivePath, extractPath):
|
|
with ZipFile(archivePath, "r") as zipFile:
|
|
filesToExtract = filterBinaries(zipFile.namelist())
|
|
zipFile.extractall(path=extractPath, members=filesToExtract)
|
|
log.debug(f"Extracted files {filesToExtract} from {archivePath.name} to {extractPath}")
|
|
return filesToExtract
|
|
|
|
|
|
def packageSignedBinaries(archivePath, filePaths):
|
|
with ZipFile(archivePath, "w") as zipFile:
|
|
for signedFilePath in filePaths:
|
|
arcName = signedFilePath.name
|
|
log.debug(f"Adding {signedFilePath.name} to {archivePath.name}")
|
|
zipFile.write(signedFilePath, arcname=arcName)
|
|
|
|
|
|
class SignWindowsBinariesProcessor(TaskProcessor):
|
|
requestClass = SignWindowsBinariesRequest
|
|
|
|
def doProcess(self):
|
|
# 1. download the binary/installer or the archive with the binaries to sign
|
|
# 2. extract the binaries to sign (if necessary)
|
|
# 3. sign the binaries
|
|
# 4. package the signed binaries (if necessary)
|
|
# 5. upload the signed binary/installer or the archive with the signed binaries
|
|
remoteUnsignedFilePath = self.task.path() / self.request.fileName
|
|
signSingleFile = remoteUnsignedFilePath.suffix != ".zip"
|
|
with TemporaryDirectory(dir=config.settings.get("General", "TempDirectory")) as workPathName:
|
|
workPath = Path(workPathName)
|
|
|
|
if signSingleFile:
|
|
# use simple file names locally to avoid issues with user provided file names
|
|
localFilePath = workPath / ("file" + remoteUnsignedFilePath.suffix)
|
|
self.sftp.download(remoteUnsignedFilePath, localFilePath)
|
|
|
|
if localFilePath.suffix == ".appx":
|
|
self.verifyApplicationId(localFilePath)
|
|
|
|
self.signFiles([localFilePath.name], path=localFilePath.parent)
|
|
|
|
localSignedFilePath = localFilePath
|
|
# we assume a single suffix
|
|
remoteSignedFileName = f"{remoteUnsignedFilePath.stem}-signed{remoteUnsignedFilePath.suffix}"
|
|
else:
|
|
# use simple file names locally to avoid issues with user provided file names
|
|
localUnsignedFilePath = workPath / "unsigned.zip"
|
|
self.sftp.download(remoteUnsignedFilePath, localUnsignedFilePath)
|
|
|
|
extractPath = workPath / "files"
|
|
os.mkdir(extractPath)
|
|
fileNames = extractUnsignedBinaries(localUnsignedFilePath, extractPath)
|
|
|
|
self.signFiles(fileNames, path=extractPath)
|
|
|
|
filePaths = [extractPath / extractedFile for extractedFile in fileNames]
|
|
localSignedFilePath = workPath / "signed.zip"
|
|
packageSignedBinaries(localSignedFilePath, filePaths)
|
|
|
|
remoteSignedFileName = localSignedFilePath.name
|
|
|
|
try:
|
|
self.sftp.upload(localSignedFilePath, self.task.path() / remoteSignedFileName)
|
|
return SignWindowsBinariesResponse(signedFileName=remoteSignedFileName)
|
|
except Error as e:
|
|
log.error("Error: %s", e)
|
|
except SFTPError as e:
|
|
log.error("Error: %s", e)
|
|
|
|
def verifyApplicationId(self, appxFilePath):
|
|
allowedApplicationId = projects.settings.get(self.request.projectPath, self.request.branch, "applicationid")
|
|
applicationId = self.getApplicationId(appxFilePath)
|
|
if applicationId != allowedApplicationId:
|
|
raise Error(
|
|
f"The application ID of the APPX ({applicationId}) does not match the allowed application ID "
|
|
f"({allowedApplicationId})."
|
|
)
|
|
|
|
def getApplicationId(self, appxFilePath) -> str:
|
|
with ZipFile(appxFilePath) as appx:
|
|
with appx.open("AppxManifest.xml") as manifest:
|
|
# XML processing is dangerous; therefore, simply parse the XML as text
|
|
# https://docs.python.org/3/library/xml.html#xml-vulnerabilities
|
|
for line in manifest.readlines():
|
|
# look for a line like <Identity Name="MyCompany.MySuite.MyApp"
|
|
line = line.strip()
|
|
if line.startswith(b"<Identity "):
|
|
m = re.match(b'<Identity Name="([^"]*)"', line)
|
|
if m is not None:
|
|
return m.group(1).decode()
|
|
# stop looking if Identity element didn't match
|
|
break
|
|
raise Error(f"The application ID could not be extracted from the file {appxFilePath.name}.")
|
|
|
|
def signFiles(self, fileNames: List[str], path: Path):
|
|
if not fileNames:
|
|
log.info("No files to sign")
|
|
return
|
|
|
|
signingUtility = config.settings.get("CodeSigning", "Utility", "codesign")
|
|
|
|
if signingUtility == "jsign":
|
|
self.signFilesWithJSign(fileNames, path)
|
|
elif signingUtility == "codesign":
|
|
self.signFilesWithCodesign(fileNames, path)
|
|
else:
|
|
raise Error(
|
|
"System configuration error - only jsign (ebourg.github.io/jsign/) and "
|
|
"codesign (Microsoft Windows SDK) are supported"
|
|
)
|
|
|
|
def signFilesWithJSign(self, fileNames: List[str], path: Path):
|
|
log.info(f"Signing files: {', '.join(fileNames)}")
|
|
workingDirectory = path
|
|
|
|
jsignPath = Path(config.settings.get("CodeSigning", "JSignPath"))
|
|
certFile = config.settings.get("CodeSigning", "Certificate", "")
|
|
password = config.settings.get("CodeSigning", "Password", "")
|
|
subjectName = config.settings.get("CodeSigning", "CommonName", "")
|
|
command = [
|
|
jsignPath,
|
|
"sign",
|
|
"--verbose",
|
|
"--storetype",
|
|
"PKCS11",
|
|
"--tsaurl",
|
|
"http://timestamp.digicert.com",
|
|
"--alg",
|
|
"SHA-256",
|
|
]
|
|
if certFile:
|
|
command += ["--keystore", certFile]
|
|
if password:
|
|
command += ["--storepass", password]
|
|
if subjectName:
|
|
command += ["--alias", subjectName]
|
|
|
|
# To avoid issues with command line length we pass the list of files to sign to JSign by way of a file
|
|
with NamedTemporaryFile(mode="w", delete=False) as filesToSign:
|
|
# Write that list and ensure it is flushed
|
|
filesToSign.write("\n".join(fileNames))
|
|
filesToSign.close()
|
|
|
|
commandToRun = command + [f"@{filesToSign.name}"]
|
|
commandToLog = sftpnotary.log.maskFollowingItem(commandToRun, "--storepass")
|
|
util.runCommand(commandToRun, commandToLog=commandToLog, cwd=workingDirectory)
|
|
|
|
# Cleanup
|
|
# In an ideal world we would use delete_on_close=False, however the linters we use do not like that
|
|
os.unlink(filesToSign.name)
|
|
|
|
def signFilesWithCodesign(self, fileNames: List[str], path: Path):
|
|
log.info(f"Signing files: {', '.join(fileNames)}")
|
|
workingDirectory = path
|
|
|
|
signtoolPath = Path(config.settings.get("CodeSigning", "SignTool"))
|
|
certFile = config.settings.get("CodeSigning", "Certificate", "")
|
|
password = config.settings.get("CodeSigning", "Password", "")
|
|
subjectName = config.settings.get("CodeSigning", "CommonName", "")
|
|
command = [
|
|
signtoolPath,
|
|
"sign",
|
|
"/v",
|
|
"/tr",
|
|
"http://timestamp.digicert.com",
|
|
"/td",
|
|
"SHA256",
|
|
"/fd",
|
|
"SHA256",
|
|
"/a",
|
|
]
|
|
if certFile:
|
|
command += ["/f", certFile]
|
|
if password:
|
|
command += ["/p", password]
|
|
if subjectName:
|
|
command += ["/n", subjectName]
|
|
|
|
# The maximum length of the string that you can use at the command prompt is 8191 characters.
|
|
# https://docs.microsoft.com/en-US/troubleshoot/windows-client/shell-experience/command-line-string-limitation
|
|
# Ensure that we stay below this limit.
|
|
availableSize = 8191 - sum(len(str(part)) + 1 for part in command) # + 1 to account for spaces
|
|
for arguments in util.collectArgumentsUpToSize(fileNames, availableSize):
|
|
commandToRun = command + arguments
|
|
commandToLog = sftpnotary.log.maskFollowingItem(commandToRun, "/p")
|
|
util.runCommand(commandToRun, commandToLog=commandToLog, cwd=workingDirectory)
|
|
|
|
|
|
class SignWindowsBinariesWorker(Worker):
|
|
processorClass = SignWindowsBinariesProcessor
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
|
|
self.checkConfig()
|
|
|
|
def checkConfig(self):
|
|
log.debug("Checking config ...")
|
|
mandatoryOptions = (("CodeSigning", "Utility"),)
|
|
for option in mandatoryOptions:
|
|
if option not in config.settings:
|
|
raise LookupError(f"Option '{option[0], option[1]}' not found in settings.")
|
|
|
|
signingUtility = config.settings.get("CodeSigning", "Utility", "codesign")
|
|
signtoolPath = Path(config.settings.get("CodeSigning", "SignTool"))
|
|
if signingUtility == "codesign" and not signtoolPath.exists():
|
|
raise FileNotFoundError(f"SignTool not found: {signtoolPath}")
|