aioacme

aioacme

Simple async client for ACME protocol.

Installation

pip install aioacme

Usage

Issue certificate with DNS-01 challenge:

import asyncio

import aioacme
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec, rsa

async def main():
    account_key = serialization.load_pem_private_key(...)
    
    async with aioacme.Client(
        account_key=account_key, 
        directory_url=aioacme.LETS_ENCRYPT_STAGING_DIRECTORY
    ) as client:
        order = await client.new_order([aioacme.Identifier("example.com")])
        
        for authorization_uri in order.authorizations:
            authorization = await client.get_authorization(authorization_uri)
            
            if authorization.status is aioacme.AuthorizationStatus.valid:
                # authorization reused
                continue
            
            challenge = next(
                c for c in authorization.challenges if c.type is aioacme.ChallengeType.dns01
            )
            
            domain = client.get_dns_challenge_domain(authorization.identifier.value)
            validation = client.get_dns_challenge_validation(challenge.token)
            
            print(f"Please add TXT record {domain} with the following content: {validation}")
            input("Press Enter when TXT record is created")
            
            await client.answer_challenge(challenge.url)
            
            while authorization.status not in (
                aioacme.AuthorizationStatus.valid,
                aioacme.AuthorizationStatus.invalid
            ):
                await asyncio.sleep(1)
                authorization = await client.get_authorization(authorization_uri)
            
            if authorization.status is aioacme.AuthorizationStatus.invalid:
                raise Exception(f"Authorization status is invalid: {authorization}")
            
        key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
        csr = (
            x509.CertificateSigningRequestBuilder()
            .subject_name(x509.Name([
                x509.NameAttribute(x509.NameOID.COMMON_NAME, "example.com")
            ]))
            .add_extension(x509.SubjectAlternativeName([
                x509.DNSName("example.com")
            ]), critical=False)
            .sign(key, hashes.SHA256())
        )
        order = await client.finalize_order(order.finalize, csr)
        
        while order.status not in {aioacme.OrderStatus.valid, aioacme.OrderStatus.invalid}:
            await asyncio.sleep(1)
            order = await client.get_order(order.uri)
            
        if order.status is aioacme.OrderStatus.invalid:
            raise Exception(f"Order status is invalid: {order}")
        
        cert = await client.get_certificate(order.certificate)
        print(cert.decode("ascii"))
        
        
asyncio.run(main())

License

This project is licensed under the MIT License - see the LICENSE file for details.