tests/test_cms.py

   1 # coding: utf-8
   2 # PyDERASN -- Python ASN.1 DER codec with abstract structures
   3 # Copyright (C) 2017-2020 Sergey Matveev <stargrave@stargrave.org>
   4 #
   5 # This program is free software: you can redistribute it and/or modify
   6 # it under the terms of the GNU Lesser General Public License as
   7 # published by the Free Software Foundation, version 3 of the License.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU Lesser General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU Lesser General Public
  15 # License along with this program.  If not, see
  16 # <http://www.gnu.org/licenses/>.
  17
  18 from datetime import datetime
  19 from hashlib import sha256
  20 from os import urandom
  21 from random import randint
  22 from subprocess import call
  23 from unittest import TestCase
  24
  25 from pyderasn import Any
  26 from pyderasn import Choice
  27 from pyderasn import encode_cer
  28 from pyderasn import Integer
  29 from pyderasn import ObjectIdentifier
  30 from pyderasn import OctetString
  31 from pyderasn import Sequence
  32 from pyderasn import SetOf
  33 from pyderasn import tag_ctxc
  34 from pyderasn import tag_ctxp
  35 from pyderasn import UTCTime
  36 from tests.test_crts import AlgorithmIdentifier
  37 from tests.test_crts import Certificate
  38 from tests.test_crts import SubjectKeyIdentifier
  39 from tests.test_crts import Time
  40
  41
  42 class CMSVersion(Integer):
  43     schema = (
  44         ("v0", 0),
  45         ("v1", 1),
  46         ("v2", 2),
  47         ("v3", 3),
  48         ("v4", 4),
  49         ("v5", 5),
  50     )
  51
  52
  53 class AttributeValue(Any):
  54     pass
  55
  56
  57 class AttributeValues(SetOf):
  58     schema = AttributeValue()
  59
  60
  61 class Attribute(Sequence):
  62     schema = (
  63         ("attrType", ObjectIdentifier()),
  64         ("attrValues", AttributeValues()),
  65     )
  66
  67
  68 class SignatureAlgorithmIdentifier(AlgorithmIdentifier):
  69     pass
  70
  71
  72 class SignedAttributes(SetOf):
  73     schema = Attribute()
  74     bounds = (1, 32)
  75     der_forced = True
  76
  77
  78 class SignerIdentifier(Choice):
  79     schema = (
  80         # ("issuerAndSerialNumber", IssuerAndSerialNumber()),
  81         ("subjectKeyIdentifier", SubjectKeyIdentifier(impl=tag_ctxp(0))),
  82     )
  83
  84
  85 class DigestAlgorithmIdentifiers(SetOf):
  86     schema = AlgorithmIdentifier()
  87
  88
  89 class DigestAlgorithmIdentifier(AlgorithmIdentifier):
  90     pass
  91
  92
  93 class SignatureValue(OctetString):
  94     pass
  95
  96
  97 class SignerInfo(Sequence):
  98     schema = (
  99         ("version", CMSVersion()),
 100         ("sid", SignerIdentifier()),
 101         ("digestAlgorithm", DigestAlgorithmIdentifier()),
 102         ("signedAttrs", SignedAttributes(impl=tag_ctxc(0), optional=True)),
 103         ("signatureAlgorithm", SignatureAlgorithmIdentifier()),
 104         ("signature", SignatureValue()),
 105         # ("unsignedAttrs", UnsignedAttributes(impl=tag_ctxc(1), optional=True)),
 106     )
 107
 108
 109 class SignerInfos(SetOf):
 110     schema = SignerInfo()
 111
 112
 113 class ContentType(ObjectIdentifier):
 114     pass
 115
 116
 117 class EncapsulatedContentInfo(Sequence):
 118     schema = (
 119         ("eContentType", ContentType()),
 120         ("eContent", OctetString(expl=tag_ctxc(0), optional=True)),
 121     )
 122
 123
 124 class CertificateChoices(Choice):
 125     schema = (
 126         ('certificate', Certificate()),
 127         # ...
 128     )
 129
 130
 131 class CertificateSet(SetOf):
 132     schema = CertificateChoices()
 133
 134
 135 class SignedData(Sequence):
 136     schema = (
 137         ("version", CMSVersion()),
 138         ("digestAlgorithms", DigestAlgorithmIdentifiers()),
 139         ("encapContentInfo", EncapsulatedContentInfo()),
 140         ("certificates", CertificateSet(impl=tag_ctxc(0), optional=True)),
 141         # ("crls", RevocationInfoChoices(impl=tag_ctxc(1), optional=True)),
 142         ("signerInfos", SignerInfos()),
 143     )
 144
 145
 146 class ContentInfo(Sequence):
 147     schema = (
 148         ("contentType", ContentType()),
 149         ("content", Any(expl=tag_ctxc(0))),
 150     )
 151
 152
 153 id_signedData = ObjectIdentifier("1.2.840.113549.1.7.2")
 154 id_sha256 = ObjectIdentifier("2.16.840.1.101.3.4.2.1")
 155 id_data = ObjectIdentifier("1.2.840.113549.1.7.1")
 156 id_ecdsa_with_SHA256 = ObjectIdentifier("1.2.840.10045.4.3.2")
 157 id_pkcs9_at_contentType = ObjectIdentifier("1.2.840.113549.1.9.3")
 158 id_pkcs9_at_messageDigest = ObjectIdentifier("1.2.840.113549.1.9.4")
 159 id_ce_subjectKeyIdentifier = ObjectIdentifier("2.5.29.14")
 160
 161
 162 class TestSignedDataCER(TestCase):
 163     def runTest(self):
 164         # openssl ecparam -name prime256v1 -genkey -out key.pem
 165         # openssl req -x509 -new -key key.pem -outform PEM -out cert.pem
 166         #    -days 365 -nodes -subj "/CN=doesnotmatter"
 167         with open("cert.cer", "rb") as fd:
 168             cert = Certificate().decod(fd.read())
 169         for ext in cert["tbsCertificate"]["extensions"]:
 170             if ext["extnID"] == id_ce_subjectKeyIdentifier:
 171                 skid = SubjectKeyIdentifier().decod(bytes(ext["extnValue"]))
 172         ai_sha256 = AlgorithmIdentifier((
 173             ("algorithm", id_sha256),
 174         ))
 175         data = urandom(randint(1000, 3000))
 176         eci = EncapsulatedContentInfo((
 177             ("eContentType", ContentType(id_data)),
 178             ("eContent", OctetString(data)),
 179         ))
 180         signed_attrs = SignedAttributes([
 181             Attribute((
 182                 ("attrType", id_pkcs9_at_contentType),
 183                 ("attrValues", AttributeValues([
 184                     AttributeValue(id_data.encode())
 185                 ])),
 186             )),
 187             Attribute((
 188                 ("attrType", id_pkcs9_at_messageDigest),
 189                 ("attrValues", AttributeValues([
 190                     AttributeValue(OctetString(
 191                         sha256(bytes(eci["eContent"])).digest()
 192                     ).encode()),
 193                 ])),
 194             )),
 195         ])
 196         with open("/tmp/in", "wb") as fd:
 197             fd.write(encode_cer(signed_attrs))
 198         self.assertEqual(0, call(" ".join((
 199             "openssl dgst -sha256",
 200             "-sign key.pem",
 201             "-binary",
 202             "/tmp/in",
 203             "> /tmp/signature",
 204         )), shell=True))
 205         ci = ContentInfo((
 206             ("contentType", ContentType(id_signedData)),
 207             ("content", Any((SignedData((
 208                 ("version", CMSVersion("v3")),
 209                 ("digestAlgorithms", DigestAlgorithmIdentifiers([ai_sha256])),
 210                 ("encapContentInfo", eci),
 211                 ("certificates", CertificateSet([
 212                     CertificateChoices(("certificate", cert)),
 213                 ])),
 214                 ("signerInfos", SignerInfos([SignerInfo((
 215                     ("version", CMSVersion("v3")),
 216                     ("sid", SignerIdentifier(
 217                         ("subjectKeyIdentifier", skid)
 218                     )),
 219                     ("digestAlgorithm", DigestAlgorithmIdentifier(ai_sha256)),
 220                     ("signedAttrs", signed_attrs),
 221                     ("signatureAlgorithm", SignatureAlgorithmIdentifier((
 222                         ("algorithm", id_ecdsa_with_SHA256),
 223                     ))),
 224                     ("signature", SignatureValue(open("/tmp/signature", "rb").read())),
 225                 ))])),
 226             ))))),
 227         ))
 228         with open("/tmp/out.p7m", "wb") as fd:
 229             fd.write(encode_cer(ci))
 230         self.assertEqual(0, call(" ".join((
 231             "openssl cms -verify",
 232             "-inform DER -in /tmp/out.p7m",
 233             "-signer cert.pem -CAfile cert.pem",
 234             "-out /dev/null",
 235         )), shell=True))