Unable to generate CMS PKCS#7 file as per OpenSSL CMS expectation
I am trying to implement CMS detached signature for artifacts in Python. The sign operation is being done by private key stored inside AWS KMS service. I have code signing certificate issued by internal PKI. Sign operation happens successfully but when I try to verify the signature using OpenSSL (using 3.2.2 version) then it fails.
The error I am getting is below one, previously I was getting ASN.1 formatting error but I think that is fixed, and now signature verification is not working. I am using RSA 2048 key with SHA256. At last, I have provided a sample signature file generated by this code in ASN.1 format.
Admin:~/environment/sign-pkcs-p7s $ openssl cms -verify -inform DER -in sign.p7s -content sign.py -CAfile pca-bundle.pem -purpose any
CMS Verification failure
80D5EDC2327F0000:error:02000068:rsa routines:ossl_rsa_verify:bad signature:crypto/rsa/rsa_sign.c:426:
80D5EDC2327F0000:error:1C880004:Provider routines:rsa_verify:RSA lib:providers/implementations/signature/rsa_sig.c:814:
80D5EDC2327F0000:error:1700009E:CMS routines:CMS_SignerInfo_verify:verification failure:crypto/cms/cms_sd.c:958:
Here is my Python 3 code:
#!/usr/bin/env python3
"""
AWS Key Management Service CMS Signing Tool - Creates detached CMS signatures using AWS KMS
Creates detached CMS (PKCS#7) signatures compatible with the Java BouncyCastle implementation. Referring to working Java implementation CMSCodeSigningObject.java and Signing.java from this Github
https://github.com/aws-samples/diy-code-signing-kms-private-ca
Usage:
python sign_kms_cms.py -k -a -c -i -o
"""
import argparse
import boto3
import sys
import os
from cryptography import x509
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa, ec
from cryptography.x509.oid import ExtensionOID, ExtendedKeyUsageOID
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes
from cryptography import x509
import datetime
try:
from asn1crypto import cms, core, algos
except ImportError:
print("Error: asn1crypto library required. Install with: pip install asn1crypto")
sys.exit(1)
class KMSContentSigner:
"""KMS-based ContentSigner equivalent to Java's KMSCMKContentSignerBuilder"""
def __init__(self, key_id, signing_algorithm):
self.key_id = key_id
self.signing_algorithm = signing_algorithm
self.kms_client = boto3.client('kms')
def sign(self, data_to_sign):
"""Sign data using AWS KMS - equivalent to ContentSigner.getSignature()"""
print(f"Generating signature with key={self.key_id} using {self.signing_algorithm}")
# Calculate digest (KMS requires digest for large data)
hash_alg = self._get_hash_algorithm()
digest = self._calculate_digest(data_to_sign, hash_alg)
print(f"Calculated {hash_alg} digest: {len(digest)} bytes")
response = self.kms_client.sign(
KeyId=self.key_id,
Message=digest,
MessageType='DIGEST',
SigningAlgorithm=self.signing_algorithm
)
signature = response['Signature']
print(f"Signature with key={self.key_id} generated successfully: {len(signature)} bytes")
print(f"Signature hex: {signature[:32].hex()}...")
return signature
def _get_hash_algorithm(self):
"""Get hash algorithm from signing algorithm"""
if 'SHA_256' in self.signing_algorithm:
return 'sha256'
elif 'SHA_384' in self.signing_algorithm:
return 'sha384'
elif 'SHA_512' in self.signing_algorithm:
return 'sha512'
else:
return 'sha256'
def _calculate_digest(self, data, algorithm):
"""Calculate digest of data"""
import hashlib
algorithms = {
'sha256': hashlib.sha256,
'sha384': hashlib.sha384,
'sha512': hashlib.sha512
}
hash_func = algorithms.get(algorithm)
if hash_func is None:
raise ValueError(f"Unsupported algorithm: {algorithm}")
return hash_func(data).digest()
class CMSCodeSigningObject:
"""Python equivalent of Java CMSCodeSigningObject class"""
@staticmethod
def create_detached_signature(key_id, signing_algorithm, data_to_sign, signer_cert, cert_chain=None):
"""Create detached CMS signature following reference implementation"""
print(f"Creating detached CMS signature using {signing_algorithm} algorithm")
# Create KMS content signer
content_signer = KMSContentSigner(key_id, signing_algorithm)
# Create certificate chain (signer + intermediates)
all_certs = [signer_cert]
if cert_chain:
all_certs.extend(cert_chain)
# Calculate message digest for signed attributes
import hashlib
hash_alg = CMSCodeSigningObject._get_hash_algorithm(signing_algorithm)
hash_func = getattr(hashlib, hash_alg)
message_digest = hash_func(data_to_sign).digest()
# Create signed attributes (required for detached signatures)
signed_attrs = cms.CMSAttributes([
cms.CMSAttribute({
'type': 'content_type',
'values': ['data']
}),
cms.CMSAttribute({
'type': 'message_digest',
'values': [message_digest]
})
])
# Sign the signed attributes (DER with SET->SEQUENCE)
signed_attrs_der = signed_attrs.dump()
signed_attrs_for_signing = b'\x30' + signed_attrs_der[1:]
signature_bytes = content_signer.sign(signed_attrs_for_signing)
print(f"Final signature: {signature_bytes[:32].hex()}...")
# Create CMS structure
cms_signature = CMSCodeSigningObject._create_cms_with_signed_attrs(
data_to_sign, signature_bytes, all_certs, signing_algorithm, signed_attrs
)
print("Successfully created detached CMS signature")
return cms_signature
@staticmethod
def _get_hash_algorithm(signing_algorithm):
"""Extract hash algorithm from KMS signing algorithm"""
if 'SHA_256' in signing_algorithm:
return 'sha256'
elif 'SHA_384' in signing_algorithm:
return 'sha384'
elif 'SHA_512' in signing_algorithm:
return 'sha512'
else:
return 'sha256'
@staticmethod
def _get_signature_algorithm(signing_algorithm):
"""Map KMS signing algorithm to CMS signature algorithm"""
if signing_algorithm == 'RSASSA_PKCS1_V1_5_SHA_256':
return 'sha256_rsa'
elif signing_algorithm == 'RSASSA_PKCS1_V1_5_SHA_384':
return 'sha384_rsa'
elif signing_algorithm == 'RSASSA_PKCS1_V1_5_SHA_512':
return 'sha512_rsa'
elif signing_algorithm == 'ECDSA_SHA_256':
return 'sha256_ecdsa'
elif signing_algorithm == 'ECDSA_SHA_384':
return 'sha384_ecdsa'
elif signing_algorithm == 'ECDSA_SHA_512':
return 'sha512_ecdsa'
else:
return 'sha256_rsa'
@staticmethod
def _create_cms_with_signed_attrs(data, signature, certificates, signing_algorithm, signed_attrs):
"""Create CMS structure with signed attributes for detached signatures"""
from asn1crypto import x509 as asn1_x509, util
# Get algorithms
hash_alg = CMSCodeSigningObject._get_hash_algorithm(signing_algorithm)
sig_alg_name = CMSCodeSigningObject._get_signature_algorithm(signing_algorithm)
signer_cert = certificates[0]
# Create SignedData
sd = cms.SignedData()
sd['version'] = 'v1'
sd['encap_content_info'] = util.OrderedDict([
('content_type', 'data')
])
sd['digest_algorithms'] = [util.OrderedDict([
('algorithm', hash_alg),
('parameters', None)
])]
# Add certificate
sd['certificates'] = [asn1_x509.Certificate.load(cert.public_bytes(serialization.Encoding.DER)) for cert in certificates]
# Create signer info with signed attributes
signer_info = cms.SignerInfo()
signer_info['version'] = 1
signer_info['digest_algorithm'] = util.OrderedDict([
('algorithm', hash_alg),
('parameters', None)
])
signer_info['signature_algorithm'] = util.OrderedDict([
('algorithm', sig_alg_name),
('parameters', None)
])
signer_info['signed_attrs'] = signed_attrs
signer_info['signature'] = signature
signer_info['sid'] = cms.SignerIdentifier({
'issuer_and_serial_number': cms.IssuerAndSerialNumber({
'issuer': asn1_x509.Name.load(signer_cert.issuer.public_bytes(default_backend)),
'serial_number': signer_cert.serial_number
})
})
sd['signer_infos'] = [signer_info]
# Create ContentInfo
asn1obj = cms.ContentInfo()
asn1obj['content_type'] = 'signed_data'
asn1obj['content'] = sd
result = asn1obj.dump()
print(f"Created CMS structure: {len(result)} bytes")
return result
def load_certificate(cert_path):
"""Load certificate from file"""
try:
with open(cert_path, 'rb') as f:
cert_data = f.read()
try:
cert = x509.load_pem_x509_certificate(cert_data)
except:
cert = x509.load_der_x509_certificate(cert_data)
# Validate EKU for code signing
try:
eku = cert.extensions.get_extension_for_oid(ExtensionOID.EXTENDED_KEY_USAGE).value
if ExtendedKeyUsageOID.CODE_SIGNING not in eku:
print("Warning: Certificate missing Code Signing EKU")
except x509.ExtensionNotFound:
print("Warning: Certificate missing Extended Key Usage extension")
return cert
except Exception as e:
print(f"Error loading certificate: {e}")
sys.exit(1)
def load_certificates_from_pem_bundle(bundle_path):
"""Loads a list of x509.Certificate objects from a PEM bundle file."""
certs = []
with open(bundle_path, 'rb') as f:
pem_data = f.read()
# Split the bundle file by certificate boundaries
for pem_cert in pem_data.split(b'-----END CERTIFICATE-----'):
if not pem_cert.strip():
continue
pem_cert += b'-----END CERTIFICATE-----'
try:
# Load each certificate using cryptography
cert = x509.load_pem_x509_certificate(pem_cert)
certs.append(cert)
except ValueError:
# Handle potential parsing errors if the file has extra data
continue
return certs
def get_key_type(cert):
"""Determine key type from certificate"""
public_key = cert.public_key()
if isinstance(public_key, rsa.RSAPublicKey):
return 'rsa'
elif isinstance(public_key, ec.EllipticCurvePublicKey):
return 'ec'
else:
raise ValueError("Unsupported key type")
def verify_key_correspondence(key_id, cert, signing_algorithm):
"""Verify KMS key corresponds to certificate"""
try:
kms_client = boto3.client('kms')
response = kms_client.get_public_key(KeyId=key_id)
if signing_algorithm not in response['SigningAlgorithms']:
print(f"Error: Algorithm {signing_algorithm} not supported by KMS key")
sys.exit(1)
kms_public_key = serialization.load_der_public_key(response['PublicKey'])
cert_public_key = cert.public_key()
kms_der = kms_public_key.public_bytes(
encoding=serialization.Encoding.DER,
format=serialization.PublicFormat.SubjectPublicKeyInfo
)
cert_der = cert_public_key.public_bytes(
encoding=serialization.Encoding.DER,
format=serialization.PublicFormat.SubjectPublicKeyInfo
)
if kms_der != cert_der:
print("Error: KMS key does not correspond to certificate public key")
sys.exit(1)
print("✓ KMS key corresponds to certificate public key")
except Exception as e:
print(f"Error verifying key correspondence: {e}")
sys.exit(1)
def main():
parser = argparse.ArgumentParser(description='Create CMS/P7S signatures using AWS KMS')
parser.add_argument('-k', '--key-id', required=True, help='KMS Key ID or ARN')
parser.add_argument('-a', '--algorithm', required=True,
choices=['RSASSA_PKCS1_V1_5_SHA_256', 'RSASSA_PKCS1_V1_5_SHA_384',
'RSASSA_PKCS1_V1_5_SHA_512', 'RSASSA_PSS_SHA_256', 'RSASSA_PSS_SHA_384',
'RSASSA_PSS_SHA_512', 'ECDSA_SHA_256', 'ECDSA_SHA_384', 'ECDSA_SHA_512'],
help='Signing algorithm')
parser.add_argument('-c', '--certificate', required=True, help='Certificate file path')
parser.add_argument('-i', '--input', required=True, help='Input file to sign')
parser.add_argument('-o', '--output', required=True, help='Output P7S signature file')
parser.add_argument('--ca-bundle', help='Certificate chain file (optional)')
args = parser.parse_args()
if not os.path.exists(args.certificate):
print(f"Error: Certificate file not found: {args.certificate}")
sys.exit(1)
# Read input data
try:
with open(args.input, 'rb') as f:
data = f.read()
print(f"Read {len(data)} bytes from {args.input}")
except Exception as e:
print(f"Error reading input file: {e}")
sys.exit(1)
# Load certificate
cert = load_certificate(args.certificate)
# Load certificate chain if provided
cert_chain = []
if args.ca_bundle:
try:
cert_chain = load_certificates_from_pem_bundle(args.ca_bundle)
print(f"Loaded {len(cert_chain)} certificates from CA bundle")
except Exception as e:
print(f"Warning: Could not load certificate chain: {e}")
cert_chain = []
# Verify KMS key corresponds to certificate
verify_key_correspondence(args.key_id, cert, args.algorithm)
# Create detached CMS signature (equivalent to Java CMSCodeSigningObject.createDetachedSignature)
cms_signature = CMSCodeSigningObject.create_detached_signature(
args.key_id, args.algorithm, data, cert, cert_chain
)
# Save P7S file
try:
with open(args.output, 'wb') as f:
f.write(cms_signature)
print(f"✓ CMS signature saved to {args.output}")
except Exception as e:
print(f"Error saving signature file: {e}")
sys.exit(1)
if __name__ == '__main__':
main()
Here is the output of signature file in ASN.1 format:
$ openssl asn1parse -inform DER -in sign.p7s
0:d=0 hl=4 l=1338 cons: SEQUENCE
4:d=1 hl=2 l= 9 prim: OBJECT :pkcs7-signedData
15:d=1 hl=4 l=1323 cons: cont [ 0 ]
19:d=2 hl=4 l=1319 cons: SEQUENCE
23:d=3 hl=2 l= 1 prim: INTEGER :01
26:d=3 hl=2 l= 13 cons: SET
28:d=4 hl=2 l= 11 cons: SEQUENCE
30:d=5 hl=2 l= 9 prim: OBJECT :sha256
41:d=3 hl=2 l= 11 cons: SEQUENCE
43:d=4 hl=2 l= 9 prim: OBJECT :pkcs7-data
54:d=3 hl=4 l= 846 cons: cont [ 0 ]
58:d=4 hl=4 l= 842 cons: SEQUENCE
62:d=5 hl=4 l= 562 cons: SEQUENCE
66:d=6 hl=2 l= 3 cons: cont [ 0 ]
68:d=7 hl=2 l= 1 prim: INTEGER :02
71:d=6 hl=2 l= 17 prim: INTEGER :C7F0A1C02C02D31716154EC59B6D5C9E
90:d=6 hl=2 l= 13 cons: SEQUENCE
92:d=7 hl=2 l= 9 prim: OBJECT :sha256WithRSAEncryption
103:d=7 hl=2 l= 0 prim: NULL
105:d=6 hl=2 l= 41 cons: SEQUENCE
107:d=7 hl=2 l= 39 cons: SET
109:d=8 hl=2 l= 37 cons: SEQUENCE
111:d=9 hl=2 l= 3 prim: OBJECT :commonName
116:d=9 hl=2 l= 30 prim: UTF8STRING :CodeSigningSubordinate-RSA2048
148:d=6 hl=2 l= 30 cons: SEQUENCE
150:d=7 hl=2 l= 13 prim: UTCTIME :251123031015Z
165:d=7 hl=2 l= 13 prim: UTCTIME :261123041015Z
180:d=6 hl=2 l= 33 cons: SEQUENCE
182:d=7 hl=2 l= 31 cons: SET
184:d=8 hl=2 l= 29 cons: SEQUENCE
186:d=9 hl=2 l= 3 prim: OBJECT :commonName
191:d=9 hl=2 l= 22 prim: UTF8STRING :CodeSigningCertificate
215:d=6 hl=4 l= 290 cons: SEQUENCE
219:d=7 hl=2 l= 13 cons: SEQUENCE
221:d=8 hl=2 l= 9 prim: OBJECT :rsaEncryption
232:d=8 hl=2 l= 0 prim: NULL
234:d=7 hl=4 l= 271 prim: BIT STRING
509:d=6 hl=2 l= 117 cons: cont [ 3 ]
511:d=7 hl=2 l= 115 cons: SEQUENCE
513:d=8 hl=2 l= 9 cons: SEQUENCE
515:d=9 hl=2 l= 3 prim: OBJECT :X509v3 Basic Constraints
520:d=9 hl=2 l= 2 prim: OCTET STRING [HEX DUMP]:3000
524:d=8 hl=2 l= 31 cons: SEQUENCE
526:d=9 hl=2 l= 3 prim: OBJECT :X509v3 Authority Key Identifier
531:d=9 hl=2 l= 24 prim: OCTET STRING [HEX DUMP]:30168014E581A06D05DDC81F7EE1781B777DEDF74E74203E
557:d=8 hl=2 l= 29 cons: SEQUENCE
559:d=9 hl=2 l= 3 prim: OBJECT :X509v3 Subject Key Identifier
564:d=9 hl=2 l= 22 prim: OCTET STRING [HEX DUMP]:041429AC6CD88D6CF9EB15A64C89D2EA68F22AF8D1A5
588:d=8 hl=2 l= 14 cons: SEQUENCE
590:d=9 hl=2 l= 3 prim: OBJECT :X509v3 Key Usage
595:d=9 hl=2 l= 1 prim: BOOLEAN :255
598:d=9 hl=2 l= 4 prim: OCTET STRING [HEX DUMP]:03020780
604:d=8 hl=2 l= 22 cons: SEQUENCE
606:d=9 hl=2 l= 3 prim: OBJECT :X509v3 Extended Key Usage
611:d=9 hl=2 l= 1 prim: BOOLEAN :255
614:d=9 hl=2 l= 12 prim: OCTET STRING [HEX DUMP]:300A06082B06010505070303
628:d=5 hl=2 l= 13 cons: SEQUENCE
630:d=6 hl=2 l= 9 prim: OBJECT :sha256WithRSAEncryption
641:d=6 hl=2 l= 0 prim: NULL
643:d=5 hl=4 l= 257 prim: BIT STRING
904:d=3 hl=4 l= 434 cons: SET
908:d=4 hl=4 l= 430 cons: SEQUENCE
912:d=5 hl=2 l= 1 prim: INTEGER :01
915:d=5 hl=2 l= 62 cons: SEQUENCE
917:d=6 hl=2 l= 41 cons: SEQUENCE
919:d=7 hl=2 l= 39 cons: SET
921:d=8 hl=2 l= 37 cons: SEQUENCE
923:d=9 hl=2 l= 3 prim: OBJECT :commonName
928:d=9 hl=2 l= 30 prim: UTF8STRING :CodeSigningSubordinate-RSA2048
960:d=6 hl=2 l= 17 prim: INTEGER :C7F0A1C02C02D31716154EC59B6D5C9E
979:d=5 hl=2 l= 11 cons: SEQUENCE
981:d=6 hl=2 l= 9 prim: OBJECT :sha256
992:d=5 hl=2 l= 75 cons: cont [ 0 ]
994:d=6 hl=2 l= 24 cons: SEQUENCE
996:d=7 hl=2 l= 9 prim: OBJECT :contentType
1007:d=7 hl=2 l= 11 cons: SET
1009:d=8 hl=2 l= 9 prim: OBJECT :pkcs7-data
1020:d=6 hl=2 l= 47 cons: SEQUENCE
1022:d=7 hl=2 l= 9 prim: OBJECT :messageDigest
1033:d=7 hl=2 l= 34 cons: SET
1035:d=8 hl=2 l= 32 prim: OCTET STRING [HEX DUMP]:0BA67F53B5A460B3F1AF60F19B3C55B8B8BA6C396C5514D3279A25ABD72C48D5
1069:d=5 hl=2 l= 11 cons: SEQUENCE
1071:d=6 hl=2 l= 9 prim: OBJECT :sha256WithRSAEncryption
1082:d=5 hl=4 l= 256 prim: OCTET STRING [HEX DUMP]:90758FDF4C3FA525AF078707CECA3F9E501206242BB06F4A55834C7359C1540527B3072107960DDE0191EA2BB39BA93D0D046DB94AF2CFADCAACDDCA6E9CE20F95A964BF062202BE55754623F10F614222E10DFE26FADFC9DF17F42D29D68672A219C8930E73CEA7937AAAA51CB1B271EF27C527391EBBD7206F0F30049D928487C8BBE371A44A3FB195CB766308EE534DE00926A06FD619916230111E0E63DD20A3348874E790CE6A6EB1A08DBD6B4FD6D78019E12F307B0EFB66488D8FAA0409F42D51F80698E52145041B6467CF5E6FA58C2BE2567108DDED76C365A6AEA4BC82B6D634FF318F3D1FAA094331855ED804FA5E2D76E663C125FD1D40AC1F52