TagSet object not in asn1Spec about pyasn1 HOT 8 OPEN

from pyasn1 import debug
debug.setLogger(debug.Debug('all'))

    for extension in core['extensions']:
        if extension['extnID'] == rfc2459.id_ce_subjectAltName:
            octet_string = decoder.decode(extension.getComponentByName('extnValue'), asn1Spec=OctetString())[0]
            (san_list, r) = decoder.decode(octet_string, rfc2459.SubjectAltName())
            for san_struct in san_list:
                if san_struct.getName() == 'dNSName':
                    fqdns.add(str(san_struct.getComponent()))

Traceback (most recent call last):
  File "/root/bin/renew_certificates.py", line 138, in <module>
    config['admin_email'], staging=args.staging, verbose=args.verbose)
  File "/root/bin/renew_certificates.py", line 113, in handle_certificates
    (fqdns, expiration_date) = parse_certificate(cert_path)
  File "/root/bin/renew_certificates.py", line 39, in parse_certificate
    octet_string = decoder.decode(extension.getComponentByName('extnValue'), asn1Spec=OctetString())[0]
  File "/usr/lib/python3.6/site-packages/pyasn1/codec/ber/decoder.py", line 1318, in __call__
    '%s not in asn1Spec: %r' % (tagSet, asn1Spec)
pyasn1.error.PyAsn1Error: <TagSet object at 0x7fe9b1da5668 tags 0:32:16> not in asn1Spec: <OctetString schema object at 0x7fe9b18f5128 tagSet <TagSet object at 0x7fe9b1f97da0 tags 0:0:4> encoding iso-8859-1>

octet_string = decoder.decode(extension.getComponentByName('extnValue'), asn1Spec=OctetString())[0]

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-

import os, sys, pwd, subprocess, re, argparse
from datetime import datetime

from pyasn1_modules import pem, rfc2459
from pyasn1.codec.der import decoder
from pyasn1.type.univ import OctetString

import yaml

CONF_FILE = '/etc/letsencrypt/renew.yaml'
RE_CERTIFICATE_FILENAME = re.compile(r'^(\d+)_cert.crt$')

def parse_certificate(certificate_path):
    fqdns = set()

    substrate = pem.readPemFromFile(open(certificate_path))
    cert = decoder.decode(substrate, asn1Spec=rfc2459.Certificate())[0]
    core = cert['tbsCertificate']

    # Extract CommonName
    for rdnss in core['subject']:
        for rdns in rdnss:
            for name in rdns:
                if name.getComponentByName('type') == rfc2459.id_at_commonName:
                    value = decoder.decode(name.getComponentByName('value'), asn1Spec=rfc2459.DirectoryString())[0]
                    fqdns.add(str(value.getComponent()))

    # extract notAfter datetime
    notAfter = str(core['validity'].getComponentByName('notAfter').getComponent()).strip('Z')
    (year, month, day, hour, minute, seconds) = [int(notAfter[i:i+2]) for i in range(0, len(notAfter), 2)]
    expiration_date = datetime(2000 + year, month, day, hour, minute, seconds)

    # Extract SubjectAltName
    for extension in core['extensions']:
        if extension['extnID'] == rfc2459.id_ce_subjectAltName:
            octet_string = decoder.decode(extension.getComponentByName('extnValue'), asn1Spec=OctetString())[0]
            (san_list, r) = decoder.decode(octet_string, rfc2459.SubjectAltName())
            for san_struct in san_list:
                if san_struct.getName() == 'dNSName':
                    fqdns.add(str(san_struct.getComponent()))
    return (fqdns, expiration_date)

def renew_certificate(cn, webroot, fqdns, working_dir, admin_email, staging = False, verbose = False):
    cert_symlink = 'latest_cert.crt'
    chain_symlink = 'latest_chain.pem'
    fullchain_symlink = 'latest_fullchain.pem'

    os.chdir(working_dir)
    latest = os.readlink(cert_symlink)
    serial = int(RE_CERTIFICATE_FILENAME.match(latest).group(1))

    new_cert = '{:04d}_cert.crt'.format(serial + 1)
    new_fullchain = '{:04d}_fullchain.pem'.format(serial + 1)
    new_chain = '{:04d}_chain.pem'.format(serial + 1)

    command = ['certbot', 'certonly', '-n', '-q',
            '--webroot', '-w', webroot,
            ]
    for fqdn in fqdns:
        command.extend(['-d', fqdn])
    command.extend(['--email', admin_email,
            '--csr', os.path.join(working_dir, cn + '.csr'),
            '--cert-path', os.path.join(working_dir, new_cert),
            '--fullchain-path', os.path.join(working_dir, new_fullchain),
            '--chain-path', os.path.join(working_dir, new_chain),
            ])
    if staging:
        command.extend(['--staging', '--break-my-certs'])
    if verbose:
        subprocess.call(['echo'] + command)
    ret_code = subprocess.call(command)
    if verbose:
        print(ret_code)

    if ret_code == 0:
        if os.path.exists(new_cert):
            if os.path.exists(cert_symlink):
                os.remove(cert_symlink)
            os.symlink(new_cert, cert_symlink)
        if os.path.exists(new_chain):
            if os.path.exists(chain_symlink):
                os.remove(chain_symlink)
            os.symlink(new_chain, chain_symlink)
        if os.path.exists(new_fullchain):
            if os.path.exists(fullchain_symlink):
                os.remove(fullchain_symlink)
            os.symlink(new_fullchain, fullchain_symlink)

def restart_daemons(daemons, verbose = False):
    for daemon in daemons:
        command = ['systemctl', daemon['action'], daemon['name']]
        if verbose:
            subprocess.call(['echo'] + command)
        ret_code = subprocess.call(command)
        if verbose:
            print(ret_code)

def handle_certificates(cert_root, www_root, threshold, daemons, admin_email, staging = False, verbose = False):
    will_restart_daemons = False
    for site in os.listdir(cert_root):
        if verbose:
            print('Evaluating', site)

        site_path = os.path.join(cert_root, site)
        owner = pwd.getpwuid(os.stat(site_path).st_uid).pw_name
        webroot = os.path.join(www_root, owner, site)
        cert_path = os.path.join(site_path, 'latest_cert.crt')

        if os.path.exists(cert_path):
            (fqdns, expiration_date) = parse_certificate(cert_path)
            if verbose:
                print(fqdns)

            now = datetime.now()
            delta = expiration_date - now
            if now >= expiration_date or delta.days <= threshold:
                if verbose:
                    print('Renewing, expired or expires in', delta.days, 'days, less than', threshold)
                renew_certificate(site, webroot, fqdns, site_path, admin_email, staging, verbose)
                will_restart_daemons = True
    if will_restart_daemons:
        restart_daemons(daemons, verbose)

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('-v', '--verbose', action='store_true', help='talk more')
    parser.add_argument('-s', '--staging', action='store_true',
            help='issue staging certificates (useful for testing purposes)')
    parser.add_argument('-c', '--config', default=CONF_FILE,
            help='path to a config file (default: {})'.format(CONF_FILE))
    args = parser.parse_args()
    config = yaml.load(open(args.config))

    handle_certificates(config['certs_root'], config['www_root'], config['threshold'], config['daemons'],
            config['admin_email'], staging=args.staging, verbose=args.verbose)

certs_root:
    /etc/letsencrypt/live
www_root:
    /var/www
admin_email:
    [email protected]
threshold:
    30
daemons:
    - name:
        nginx
      action:
        reload

-----BEGIN CERTIFICATE-----
MIIEVzCCAz+gAwIBAgISA3/ikQnQ0zcdEYpxMrbISAe1MA0GCSqGSIb3DQEBCwUA
MEoxCzAJBgNVBAYTAlVTMRYwFAYDVQQKEw1MZXQncyBFbmNyeXB0MSMwIQYDVQQD
ExpMZXQncyBFbmNyeXB0IEF1dGhvcml0eSBYMzAeFw0xNzA5MzAyMTAwMTVaFw0x
NzEyMjkyMTAwMTVaMBwxGjAYBgNVBAMTEXFjaGF0LmxvcmRyYW4ubmV0MHYwEAYH
KoZIzj0CAQYFK4EEACIDYgAEGoTG6dTiWGMEzrGxZkCCEc5oTxTpndrwhdaXu5VQ
ZBqTFE+biqEtW45Ip5ghsrmI2kKKWjw4IIdpI0SNnIlcvOjWIsKnAC9/HJeS6o3c
O/5YBM0p8fTNoqtnNAxwjxXKo4ICETCCAg0wDgYDVR0PAQH/BAQDAgeAMB0GA1Ud
JQQWMBQGCCsGAQUFBwMBBggrBgEFBQcDAjAMBgNVHRMBAf8EAjAAMB0GA1UdDgQW
BBTqzEN+pvfQifP+lleAPoxI1ltfaTAfBgNVHSMEGDAWgBSoSmpjBH3duubRObem
RWXv86jsoTBvBggrBgEFBQcBAQRjMGEwLgYIKwYBBQUHMAGGImh0dHA6Ly9vY3Nw
LmludC14My5sZXRzZW5jcnlwdC5vcmcwLwYIKwYBBQUHMAKGI2h0dHA6Ly9jZXJ0
LmludC14My5sZXRzZW5jcnlwdC5vcmcvMBwGA1UdEQQVMBOCEXFjaGF0LmxvcmRy
YW4ubmV0MIH+BgNVHSAEgfYwgfMwCAYGZ4EMAQIBMIHmBgsrBgEEAYLfEwEBATCB
1jAmBggrBgEFBQcCARYaaHR0cDovL2Nwcy5sZXRzZW5jcnlwdC5vcmcwgasGCCsG
AQUFBwICMIGeDIGbVGhpcyBDZXJ0aWZpY2F0ZSBtYXkgb25seSBiZSByZWxpZWQg
dXBvbiBieSBSZWx5aW5nIFBhcnRpZXMgYW5kIG9ubHkgaW4gYWNjb3JkYW5jZSB3
aXRoIHRoZSBDZXJ0aWZpY2F0ZSBQb2xpY3kgZm91bmQgYXQgaHR0cHM6Ly9sZXRz
ZW5jcnlwdC5vcmcvcmVwb3NpdG9yeS8wDQYJKoZIhvcNAQELBQADggEBABiP+dRn
Ivm/k/1PYCakaObZlK69I6gGtOdxPAYBZ13QK0DXTRnKXw3NIYfoojV4Q1ld2GzA
5fbUCB9wL0eDb1YguumgoJtJTC+4SrPKfSivGLn6xnfyuu6zd0NsEmXu6c8pvCd/
nbwVHUodY2d/WzD3Uloa1bRwbgnJsWRu7cKAB+tENw1Y5r+kMdMMgcqkHgKP0aoV
y4WKw3bCwG/OY1GRWrVvM0hiE0xl+GbIZEDPGKFbOKNetPdbLPnMXCeAjtD+Jfpb
cnGb05HRpa7yVtKiL4zLfO9odCOTyaL5/kbYSzSV4rspNVta4p1i/Is43BRBLOd3
n5HkV2S7N4xsuCA=
-----END CERTIFICATE-----
Certificate:
    Data:
        Version: 3 (0x2)
        Serial Number:
            03:7f:e2:91:09:d0:d3:37:1d:11:8a:71:32:b6:c8:48:07:b5
    Signature Algorithm: sha256WithRSAEncryption
        Issuer: C = US, O = Let's Encrypt, CN = Let's Encrypt Authority X3
        Validity
            Not Before: Sep 30 21:00:15 2017 GMT
            Not After : Dec 29 21:00:15 2017 GMT
        Subject: CN = qchat.lordran.net
        Subject Public Key Info:
            Public Key Algorithm: id-ecPublicKey
                Public-Key: (384 bit)
                pub:
                    04:1a:84:c6:e9:d4:e2:58:63:04:ce:b1:b1:66:40:
                    82:11:ce:68:4f:14:e9:9d:da:f0:85:d6:97:bb:95:
                    50:64:1a:93:14:4f:9b:8a:a1:2d:5b:8e:48:a7:98:
                    21:b2:b9:88:da:42:8a:5a:3c:38:20:87:69:23:44:
                    8d:9c:89:5c:bc:e8:d6:22:c2:a7:00:2f:7f:1c:97:
                    92:ea:8d:dc:3b:fe:58:04:cd:29:f1:f4:cd:a2:ab:
                    67:34:0c:70:8f:15:ca
                ASN1 OID: secp384r1
                NIST CURVE: P-384
        X509v3 extensions:
            X509v3 Key Usage: critical
                Digital Signature
            X509v3 Extended Key Usage:
                TLS Web Server Authentication, TLS Web Client Authentication
            X509v3 Basic Constraints: critical
                CA:FALSE
            X509v3 Subject Key Identifier:
                EA:CC:43:7E:A6:F7:D0:89:F3:FE:96:57:80:3E:8C:48:D6:5B:5F:69
            X509v3 Authority Key Identifier:
                keyid:A8:4A:6A:63:04:7D:DD:BA:E6:D1:39:B7:A6:45:65:EF:F3:A8:EC:A1

            Authority Information Access:
                OCSP - URI:http://ocsp.int-x3.letsencrypt.org
                CA Issuers - URI:http://cert.int-x3.letsencrypt.org/

            X509v3 Subject Alternative Name:
                DNS:qchat.lordran.net
            X509v3 Certificate Policies:
                Policy: 2.23.140.1.2.1
                Policy: 1.3.6.1.4.1.44947.1.1.1
                  CPS: http://cps.letsencrypt.org
                  User Notice:
                    Explicit Text: This Certificate may only be relied upon by Relying Parties and only in accordance with the Certificate Policy found at https://letsencrypt.org/repository/

    Signature Algorithm: sha256WithRSAEncryption
         18:8f:f9:d4:67:22:f9:bf:93:fd:4f:60:26:a4:68:e6:d9:94:
         ae:bd:23:a8:06:b4:e7:71:3c:06:01:67:5d:d0:2b:40:d7:4d:
         19:ca:5f:0d:cd:21:87:e8:a2:35:78:43:59:5d:d8:6c:c0:e5:
         f6:d4:08:1f:70:2f:47:83:6f:56:20:ba:e9:a0:a0:9b:49:4c:
         2f:b8:4a:b3:ca:7d:28:af:18:b9:fa:c6:77:f2:ba:ee:b3:77:
         43:6c:12:65:ee:e9:cf:29:bc:27:7f:9d:bc:15:1d:4a:1d:63:
         67:7f:5b:30:f7:52:5a:1a:d5:b4:70:6e:09:c9:b1:64:6e:ed:
         c2:80:07:eb:44:37:0d:58:e6:bf:a4:31:d3:0c:81:ca:a4:1e:
         02:8f:d1:aa:15:cb:85:8a:c3:76:c2:c0:6f:ce:63:51:91:5a:
         b5:6f:33:48:62:13:4c:65:f8:66:c8:64:40:cf:18:a1:5b:38:
         a3:5e:b4:f7:5b:2c:f9:cc:5c:27:80:8e:d0:fe:25:fa:5b:72:
         71:9b:d3:91:d1:a5:ae:f2:56:d2:a2:2f:8c:cb:7c:ef:68:74:
         23:93:c9:a2:f9:fe:46:d8:4b:34:95:e2:bb:29:35:5b:5a:e2:
         9d:62:fc:8b:38:dc:14:41:2c:e7:77:9f:91:e4:57:64:bb:37:
         8c:6c:b8:20

    for extension in core['extensions']:
        if extension['extnID'] == rfc2459.id_ce_subjectAltName:
            (san_list, r) = decoder.decode(extension.getComponentByName('extnValue'), rfc2459.SubjectAltName())
            for san_struct in san_list:
                if san_struct.getName() == 'dNSName':
                    fqdns.add(str(san_struct.getComponent()))

Extension  ::=  SEQUENCE  {
     extnID      OBJECT IDENTIFIER,
     critical    BOOLEAN OPTIONAL, -- DEFAULT FALSE XXX
     extnValue   OCTET STRING
}

class Extension(univ.Sequence):
    componentType = namedtype.NamedTypes(
        namedtype.NamedType('extnID', univ.ObjectIdentifier()),
        namedtype.DefaultedNamedType('critical', univ.Boolean('False')),
        namedtype.NamedType('extnValue', univ.Any())
    )

octet_string = decoder.decode(extension.getComponentByName('extnValue'), asn1Spec=OctetString())[0]