security/manager/tools/pycms.py

   1 #!/usr/bin/env python
   2 #
   3 # This Source Code Form is subject to the terms of the Mozilla Public
   4 # License, v. 2.0. If a copy of the MPL was not distributed with this
   5 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
   6
   7 """
   8 Reads a specification from stdin and outputs a PKCS7 (CMS) message with
   9 the desired properties.
  10
  11 The specification format is as follows:
  12
  13 sha1:<hex string>
  14 sha256:<hex string>
  15 signer:
  16 <pycert specification>
  17
  18 Eith or both of sha1 and sha256 may be specified. The value of
  19 each hash directive is what will be put in the messageDigest
  20 attribute of the SignerInfo that corresponds to the signature
  21 algorithm defined by the hash algorithm and key type of the
  22 default key. Together, these comprise the signerInfos field of
  23 the SignedData. If neither hash is specified, the signerInfos
  24 will be an empty SET (i.e. there will be no actual signature
  25 information).
  26 The certificate specification must come last.
  27 """
  28
  29 import base64
  30 import sys
  31 from io import StringIO
  32
  33 import pycert
  34 import pykey
  35 from pyasn1.codec.der import decoder, encoder
  36 from pyasn1.type import tag, univ
  37 from pyasn1_modules import rfc2315, rfc2459
  38
  39
  40 class Error(Exception):
  41     """Base class for exceptions in this module."""
  42
  43     pass
  44
  45
  46 class UnknownDirectiveError(Error):
  47     """Helper exception type to handle unknown specification
  48     directives."""
  49
  50     def __init__(self, directive):
  51         super(UnknownDirectiveError, self).__init__()
  52         self.directive = directive
  53
  54     def __str__(self):
  55         return "Unknown directive %s" % repr(self.directive)
  56
  57
  58 class CMS(object):
  59     """Utility class for reading a CMS specification and
  60     generating a CMS message"""
  61
  62     def __init__(self, paramStream):
  63         self.sha1 = ""
  64         self.sha256 = ""
  65         signerSpecification = StringIO()
  66         readingSignerSpecification = False
  67         for line in paramStream.readlines():
  68             if readingSignerSpecification:
  69                 print(line.strip(), file=signerSpecification)
  70             elif line.strip() == "signer:":
  71                 readingSignerSpecification = True
  72             elif line.startswith("sha1:"):
  73                 self.sha1 = line.strip()[len("sha1:") :]
  74             elif line.startswith("sha256:"):
  75                 self.sha256 = line.strip()[len("sha256:") :]
  76             else:
  77                 raise UnknownDirectiveError(line.strip())
  78         signerSpecification.seek(0)
  79         self.signer = pycert.Certificate(signerSpecification)
  80         self.signingKey = pykey.keyFromSpecification("default")
  81
  82     def buildAuthenticatedAttributes(self, value, implicitTag=None):
  83         """Utility function to build a pyasn1 AuthenticatedAttributes
  84         object. Useful because when building a SignerInfo, the
  85         authenticatedAttributes needs to be tagged implicitly, but when
  86         signing an AuthenticatedAttributes, it needs the explicit SET
  87         tag."""
  88         if implicitTag:
  89             authenticatedAttributes = rfc2315.Attributes().subtype(
  90                 implicitTag=implicitTag
  91             )
  92         else:
  93             authenticatedAttributes = rfc2315.Attributes()
  94         contentTypeAttribute = rfc2315.Attribute()
  95         # PKCS#9 contentType
  96         contentTypeAttribute["type"] = univ.ObjectIdentifier("1.2.840.113549.1.9.3")
  97         contentTypeAttribute["values"] = univ.SetOf(rfc2459.AttributeValue())
  98         # PKCS#7 data
  99         contentTypeAttribute["values"][0] = univ.ObjectIdentifier(
 100             "1.2.840.113549.1.7.1"
 101         )
 102         authenticatedAttributes[0] = contentTypeAttribute
 103         hashAttribute = rfc2315.Attribute()
 104         # PKCS#9 messageDigest
 105         hashAttribute["type"] = univ.ObjectIdentifier("1.2.840.113549.1.9.4")
 106         hashAttribute["values"] = univ.SetOf(rfc2459.AttributeValue())
 107         hashAttribute["values"][0] = univ.OctetString(hexValue=value)
 108         authenticatedAttributes[1] = hashAttribute
 109         return authenticatedAttributes
 110
 111     def pykeyHashToDigestAlgorithm(self, pykeyHash):
 112         """Given a pykey hash algorithm identifier, builds an
 113         AlgorithmIdentifier for use with pyasn1."""
 114         if pykeyHash == pykey.HASH_SHA1:
 115             oidString = "1.3.14.3.2.26"
 116         elif pykeyHash == pykey.HASH_SHA256:
 117             oidString = "2.16.840.1.101.3.4.2.1"
 118         else:
 119             raise pykey.UnknownHashAlgorithmError(pykeyHash)
 120         algorithmIdentifier = rfc2459.AlgorithmIdentifier()
 121         algorithmIdentifier["algorithm"] = univ.ObjectIdentifier(oidString)
 122         # Directly setting parameters to univ.Null doesn't currently work.
 123         nullEncapsulated = encoder.encode(univ.Null())
 124         algorithmIdentifier["parameters"] = univ.Any(nullEncapsulated)
 125         return algorithmIdentifier
 126
 127     def buildSignerInfo(self, certificate, pykeyHash, digestValue):
 128         """Given a pyasn1 certificate, a pykey hash identifier
 129         and a hash value, creates a SignerInfo with the
 130         appropriate values."""
 131         signerInfo = rfc2315.SignerInfo()
 132         signerInfo["version"] = 1
 133         issuerAndSerialNumber = rfc2315.IssuerAndSerialNumber()
 134         issuerAndSerialNumber["issuer"] = self.signer.getIssuer()
 135         issuerAndSerialNumber["serialNumber"] = certificate["tbsCertificate"][
 136             "serialNumber"
 137         ]
 138         signerInfo["issuerAndSerialNumber"] = issuerAndSerialNumber
 139         signerInfo["digestAlgorithm"] = self.pykeyHashToDigestAlgorithm(pykeyHash)
 140         rsa = rfc2459.AlgorithmIdentifier()
 141         rsa["algorithm"] = rfc2459.rsaEncryption
 142         rsa["parameters"] = univ.Null()
 143         authenticatedAttributes = self.buildAuthenticatedAttributes(
 144             digestValue,
 145             implicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatConstructed, 0),
 146         )
 147         authenticatedAttributesTBS = self.buildAuthenticatedAttributes(digestValue)
 148         signerInfo["authenticatedAttributes"] = authenticatedAttributes
 149         signerInfo["digestEncryptionAlgorithm"] = rsa
 150         authenticatedAttributesEncoded = encoder.encode(authenticatedAttributesTBS)
 151         signature = self.signingKey.sign(authenticatedAttributesEncoded, pykeyHash)
 152         # signature will be a hexified bit string of the form
 153         # "'<hex bytes>'H". For some reason that's what BitString wants,
 154         # but since this is an OCTET STRING, we have to strip off the
 155         # quotation marks and trailing "H".
 156         signerInfo["encryptedDigest"] = univ.OctetString(hexValue=signature[1:-2])
 157         return signerInfo
 158
 159     def toDER(self):
 160         contentInfo = rfc2315.ContentInfo()
 161         contentInfo["contentType"] = rfc2315.signedData
 162
 163         signedData = rfc2315.SignedData()
 164         signedData["version"] = rfc2315.Version(1)
 165
 166         digestAlgorithms = rfc2315.DigestAlgorithmIdentifiers()
 167         digestAlgorithms[0] = self.pykeyHashToDigestAlgorithm(pykey.HASH_SHA1)
 168         signedData["digestAlgorithms"] = digestAlgorithms
 169
 170         dataContentInfo = rfc2315.ContentInfo()
 171         dataContentInfo["contentType"] = rfc2315.data
 172         signedData["contentInfo"] = dataContentInfo
 173
 174         certificates = rfc2315.ExtendedCertificatesAndCertificates().subtype(
 175             implicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatConstructed, 0)
 176         )
 177         extendedCertificateOrCertificate = rfc2315.ExtendedCertificateOrCertificate()
 178         certificate = decoder.decode(
 179             self.signer.toDER(), asn1Spec=rfc2459.Certificate()
 180         )[0]
 181         extendedCertificateOrCertificate["certificate"] = certificate
 182         certificates[0] = extendedCertificateOrCertificate
 183         signedData["certificates"] = certificates
 184
 185         signerInfos = rfc2315.SignerInfos()
 186
 187         if len(self.sha1) > 0:
 188             signerInfos[len(signerInfos)] = self.buildSignerInfo(
 189                 certificate, pykey.HASH_SHA1, self.sha1
 190             )
 191         if len(self.sha256) > 0:
 192             signerInfos[len(signerInfos)] = self.buildSignerInfo(
 193                 certificate, pykey.HASH_SHA256, self.sha256
 194             )
 195         signedData["signerInfos"] = signerInfos
 196
 197         encoded = encoder.encode(signedData)
 198         anyTag = univ.Any(encoded).subtype(
 199             explicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatConstructed, 0)
 200         )
 201
 202         contentInfo["content"] = anyTag
 203         return encoder.encode(contentInfo)
 204
 205     def toPEM(self):
 206         output = "-----BEGIN PKCS7-----"
 207         der = self.toDER()
 208         b64 = base64.b64encode(der)
 209         while b64:
 210             output += "\n" + b64[:64]
 211             b64 = b64[64:]
 212         output += "\n-----END PKCS7-----\n"
 213         return output
 214
 215
 216 # When run as a standalone program, this will read a specification from
 217 # stdin and output the certificate as PEM to stdout.
 218 if __name__ == "__main__":
 219     print(CMS(sys.stdin).toPEM())