DXR is a code search and navigation tool aimed at making sense of large projects. It supports full-text and regex searches as well as structural queries.

Mercurial (b6d82b1a6b02)

VCS Links

Line Code
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207
#!/usr/bin/env python
#
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.

"""
Reads a specification from stdin and outputs a PKCS7 (CMS) message with
the desired properties.

The specification format is as follows:

sha1:<hex string>
sha256:<hex string>
signer:
<pycert specification>

Eith or both of sha1 and sha256 may be specified. The value of
each hash directive is what will be put in the messageDigest
attribute of the SignerInfo that corresponds to the signature
algorithm defined by the hash algorithm and key type of the
default key. Together, these comprise the signerInfos field of
the SignedData. If neither hash is specified, the signerInfos
will be an empty SET (i.e. there will be no actual signature
information).
The certificate specification must come last.
"""

from pyasn1.codec.der import decoder
from pyasn1.codec.der import encoder
from pyasn1.type import tag, univ
from pyasn1_modules import rfc2315, rfc2459
import StringIO
import base64
import pycert
import pykey
import sys


class Error(Exception):
    """Base class for exceptions in this module."""
    pass


class UnknownDirectiveError(Error):
    """Helper exception type to handle unknown specification
    directives."""

    def __init__(self, directive):
        super(UnknownDirectiveError, self).__init__()
        self.directive = directive

    def __str__(self):
        return 'Unknown directive %s' % repr(self.directive)


class CMS(object):
    """Utility class for reading a CMS specification and
    generating a CMS message"""

    def __init__(self, paramStream):
        self.sha1 = ''
        self.sha256 = ''
        signerSpecification = StringIO.StringIO()
        readingSignerSpecification = False
        for line in paramStream.readlines():
            if readingSignerSpecification:
                print >>signerSpecification, line.strip()
            elif line.strip() == 'signer:':
                readingSignerSpecification = True
            elif line.startswith('sha1:'):
                self.sha1 = line.strip()[len('sha1:'):]
            elif line.startswith('sha256:'):
                self.sha256 = line.strip()[len('sha256:'):]
            else:
                raise UnknownDirectiveError(line.strip())
        signerSpecification.seek(0)
        self.signer = pycert.Certificate(signerSpecification)
        self.signingKey = pykey.keyFromSpecification('default')

    def buildAuthenticatedAttributes(self, value, implicitTag=None):
        """Utility function to build a pyasn1 AuthenticatedAttributes
        object. Useful because when building a SignerInfo, the
        authenticatedAttributes needs to be tagged implicitly, but when
        signing an AuthenticatedAttributes, it needs the explicit SET
        tag."""
        if implicitTag:
            authenticatedAttributes = rfc2315.Attributes().subtype(implicitTag=implicitTag)
        else:
            authenticatedAttributes = rfc2315.Attributes()
        contentTypeAttribute = rfc2315.Attribute()
        # PKCS#9 contentType
        contentTypeAttribute['type'] = univ.ObjectIdentifier('1.2.840.113549.1.9.3')
        contentTypeAttribute['values'] = univ.SetOf(rfc2459.AttributeValue())
        # PKCS#7 data
        contentTypeAttribute['values'][0] = univ.ObjectIdentifier('1.2.840.113549.1.7.1')
        authenticatedAttributes[0] = contentTypeAttribute
        hashAttribute = rfc2315.Attribute()
        # PKCS#9 messageDigest
        hashAttribute['type'] = univ.ObjectIdentifier('1.2.840.113549.1.9.4')
        hashAttribute['values'] = univ.SetOf(rfc2459.AttributeValue())
        hashAttribute['values'][0] = univ.OctetString(hexValue=value)
        authenticatedAttributes[1] = hashAttribute
        return authenticatedAttributes

    def pykeyHashToDigestAlgorithm(self, pykeyHash):
        """Given a pykey hash algorithm identifier, builds an
        AlgorithmIdentifier for use with pyasn1."""
        if pykeyHash == pykey.HASH_SHA1:
            oidString = '1.3.14.3.2.26'
        elif pykeyHash == pykey.HASH_SHA256:
            oidString = '2.16.840.1.101.3.4.2.1'
        else:
            raise pykey.UnknownHashAlgorithmError(pykeyHash)
        algorithmIdentifier = rfc2459.AlgorithmIdentifier()
        algorithmIdentifier['algorithm'] = univ.ObjectIdentifier(oidString)
        # Directly setting parameters to univ.Null doesn't currently work.
        nullEncapsulated = encoder.encode(univ.Null())
        algorithmIdentifier['parameters'] = univ.Any(nullEncapsulated)
        return algorithmIdentifier

    def buildSignerInfo(self, certificate, pykeyHash, digestValue):
        """Given a pyasn1 certificate, a pykey hash identifier
        and a hash value, creates a SignerInfo with the
        appropriate values."""
        signerInfo = rfc2315.SignerInfo()
        signerInfo['version'] = 1
        issuerAndSerialNumber = rfc2315.IssuerAndSerialNumber()
        issuerAndSerialNumber['issuer'] = self.signer.getIssuer()
        issuerAndSerialNumber['serialNumber'] = certificate['tbsCertificate']['serialNumber']
        signerInfo['issuerAndSerialNumber'] = issuerAndSerialNumber
        signerInfo['digestAlgorithm'] = self.pykeyHashToDigestAlgorithm(pykeyHash)
        rsa = rfc2459.AlgorithmIdentifier()
        rsa['algorithm'] = rfc2459.rsaEncryption
        rsa['parameters'] = univ.Null()
        authenticatedAttributes = (
            self.buildAuthenticatedAttributes(digestValue,
                                              implicitTag=tag.Tag(tag.tagClassContext,
                                                                  tag.tagFormatConstructed, 0)))
        authenticatedAttributesTBS = self.buildAuthenticatedAttributes(digestValue)
        signerInfo['authenticatedAttributes'] = authenticatedAttributes
        signerInfo['digestEncryptionAlgorithm'] = rsa
        authenticatedAttributesEncoded = encoder.encode(authenticatedAttributesTBS)
        signature = self.signingKey.sign(authenticatedAttributesEncoded, pykeyHash)
        # signature will be a hexified bit string of the form
        # "'<hex bytes>'H". For some reason that's what BitString wants,
        # but since this is an OCTET STRING, we have to strip off the
        # quotation marks and trailing "H".
        signerInfo['encryptedDigest'] = univ.OctetString(hexValue=signature[1:-2])
        return signerInfo

    def toDER(self):
        contentInfo = rfc2315.ContentInfo()
        contentInfo['contentType'] = rfc2315.signedData

        signedData = rfc2315.SignedData()
        signedData['version'] = rfc2315.Version(1)

        digestAlgorithms = rfc2315.DigestAlgorithmIdentifiers()
        digestAlgorithms[0] = self.pykeyHashToDigestAlgorithm(pykey.HASH_SHA1)
        signedData['digestAlgorithms'] = digestAlgorithms

        dataContentInfo = rfc2315.ContentInfo()
        dataContentInfo['contentType'] = rfc2315.data
        signedData['contentInfo'] = dataContentInfo

        certificates = rfc2315.ExtendedCertificatesAndCertificates().subtype(
            implicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatConstructed, 0))
        extendedCertificateOrCertificate = rfc2315.ExtendedCertificateOrCertificate()
        certificate = decoder.decode(self.signer.toDER(),
                                     asn1Spec=rfc2459.Certificate())[0]
        extendedCertificateOrCertificate['certificate'] = certificate
        certificates[0] = extendedCertificateOrCertificate
        signedData['certificates'] = certificates

        signerInfos = rfc2315.SignerInfos()

        if len(self.sha1) > 0:
            signerInfos[len(signerInfos)] = self.buildSignerInfo(certificate,
                                                                 pykey.HASH_SHA1, self.sha1)
        if len(self.sha256) > 0:
            signerInfos[len(signerInfos)] = self.buildSignerInfo(certificate,
                                                                 pykey.HASH_SHA256, self.sha256)
        signedData['signerInfos'] = signerInfos

        encoded = encoder.encode(signedData)
        anyTag = univ.Any(encoded).subtype(
            explicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatConstructed, 0))

        contentInfo['content'] = anyTag
        return encoder.encode(contentInfo)

    def toPEM(self):
        output = '-----BEGIN PKCS7-----'
        der = self.toDER()
        b64 = base64.b64encode(der)
        while b64:
            output += '\n' + b64[:64]
            b64 = b64[64:]
        output += '\n-----END PKCS7-----\n'
        return output


# When run as a standalone program, this will read a specification from
# stdin and output the certificate as PEM to stdout.
if __name__ == '__main__':
    print(CMS(sys.stdin).toPEM())