Merge pull request #20 from RenovoSolutions/19-add-dry-run-support

Add support for dry runs so testing is easier

function/index.py

@@ -5,24 +5,26 @@
 import datetime
 import os
 import subprocess
-import logging
 
 def upload_to_s3(local_path, keyname):
-  logging.info('Uploading files to S3')
+  print('INFO: Uploading files to S3')
   s3 = boto3.resource('s3')
   data = open(local_path, 'rb')
   s3.Bucket(os.environ['CERTIFICATE_BUCKET']).put_object(Key=f"{os.environ['OBJECT_PREFIX']}{keyname}", Body=data)
 
 def read_and_delete_file(path, filename):
-  upload_to_s3(path, filename)
+  if not os.getenv("DRY_RUN", 'False').lower() in ["true", "1"]:
+    upload_to_s3(path, filename)
 
-  with open(path, 'r') as file:
-    contents = file.read()
-  os.remove(path)
-  return contents
+    with open(path, 'r') as file:
+      contents = file.read()
+    os.remove(path)
+    return contents
+  else:
+    print(f'WARN: Dry run was used so {filename} was not generated.')
 
 def provision_cert(email, domains):
-  certbot.main.main([
+  cerbot_args = [
     'certonly',                             # Obtain a cert but don't install it
     '-n',                                   # Run in non-interactive mode
     '--agree-tos',                          # Agree to the terms of service,
@@ -34,7 +36,11 @@ def provision_cert(email, domains):
     '--work-dir', '/tmp/work-dir/',
     '--logs-dir', '/tmp/logs-dir/',
     '--preferred-chain', os.environ['PREFERRED_CHAIN'],
-  ])
+  ]
+  if os.getenv("DRY_RUN", 'False').lower() in ["true", "1"]:
+    cerbot_args.append("--dry-run")
+
+  certbot.main.main(cerbot_args)
 
   first_domain = domains.split(',')[0]
   path = '/tmp/config-dir/live/' + first_domain + '/'
@@ -47,18 +53,18 @@ def provision_cert(email, domains):
 def should_provision(domains):
   existing_cert = find_existing_cert(domains)
   if existing_cert:
-    logging.info('Cert already exists. Checking date for reissue.')
+    print('INFO: Cert already exists. Checking date for reissue.')
     now = datetime.datetime.now(datetime.timezone.utc)
     not_after = existing_cert['Certificate']['NotAfter']
     reissue = (not_after - now).days <= int(os.environ['REISSUE_DAYS'])
     if reissue:
-      logging.info(f'Cert will expire sometime in the next {os.environ["REISSUE_DAYS"]} days so will be reissued.')
+      print(f'INFO: Cert will expire sometime in the next {os.environ["REISSUE_DAYS"]} days so will be reissued.')
       return reissue
     else:
-      logging.info(f'Cert wont expire in next {os.environ["REISSUE_DAYS"]} days so will NOT be reissued.')
+      print(f'INFO: Cert wont expire in next {os.environ["REISSUE_DAYS"]} days so will NOT be reissued.')
       return reissue
   else:
-    logging.info('Cert not found in ACM. Will issue new cert.')
+    print('INFO: Cert not found in ACM. Will issue new cert.')
     return True
 
 def find_existing_cert(domains):
@@ -78,7 +84,7 @@ def find_existing_cert(domains):
   return None
 
 def notify_via_sns(topic_arn, domains, certificate):
-  logging.info('Sending SNS notification')
+  print('INFO: Sending SNS notification')
   process = subprocess.Popen(['openssl', 'x509', '-noout', '-text'],
     stdin=subprocess.PIPE, stdout=subprocess.PIPE, encoding='utf8')
   stdout, stderr = process.communicate(certificate)
@@ -90,7 +96,7 @@ def notify_via_sns(topic_arn, domains, certificate):
   )
 
 def upload_cert_to_acm(cert, domains):
-  logging.info('Importing cert to ACM')
+  print('INFO: Importing cert to ACM')
   existing_cert = find_existing_cert(domains)
   certificate_arn = existing_cert['Certificate']['CertificateArn'] if existing_cert else None
 
@@ -115,5 +121,10 @@ def handler(event, context):
   domains = os.environ['LETSENCRYPT_DOMAINS']
   if should_provision(domains):
     cert = provision_cert(os.environ['LETSENCRYPT_EMAIL'], domains)
-    upload_cert_to_acm(cert, domains)
-    notify_via_sns(os.environ['NOTIFICATION_SNS_ARN'], domains, cert['certificate'])
+    if not os.getenv("DRY_RUN", 'False').lower() in ["true", "1"]:
+      upload_cert_to_acm(cert, domains)
+      notify_via_sns(os.environ['NOTIFICATION_SNS_ARN'], domains, cert['certificate'])
+    else:
+      print('WARN: Dry run was used so ACM import and S3 upload arent tested.')
+
+handler('', '')

function/requirements.txt

@@ -1,2 +1,3 @@
+boto3
 certbot
 certbot-dns-route53