basic check on certificate validity

Author: rpcme

src/bulk_importer/main.py

@@ -33,14 +33,6 @@
 from aws_lambda_powertools.utilities.data_classes import S3Event
 from aws_lambda_powertools.utilities.typing import LambdaContext
 from aws_lambda_powertools.utilities.validation import validator
-from schemas import INPUT_SCHEMA, OUTPUT_SCHEMA
-
-_LAMBDA_S3_RESOURCE = { "resource" : boto3resource('s3'),
-                        "bucket_name" : os.environ.get("S3_BUCKET_NAME","NONE") }
-_LAMBDA_SQS_RESOURCE = { "resource" : boto3resource('sqs'),
-                         "bucket_name" : os.environ.get("S3_BUCKET_NAME","NONE") }
-_LAMBDA_IOT_RESOURCE = { "resource" : boto3resource('iot'),
-                         "bucket_name" : os.environ.get("S3_BUCKET_NAME","NONE") }
 
 logger = logging.getLogger()
 logger.setLevel("INFO")
@@ -51,10 +43,10 @@
     100: "d",
 }
 
-
 verbose = True
 config = None
 
+# Verify that the certificate is in IoT Core
 def get_certificate(certificateId):
     try:
         response = iot_client.describe_certificate(certificateId=certificateId)
@@ -63,6 +55,7 @@ def get_certificate(certificateId):
         print("Certificate [" + certificateId + "] not found in IoT Core.")
         return None
 
+# Retrieve the certificate Arn. 
 def get_certificate_arn(certificateId):
     try:
         response = iot_client.describe_certificate(certificateId=certificateId)
@@ -109,11 +102,11 @@ def get_thing_group(thingGroupName):
 
 def get_thing_type(typeName):
     try:
-        response = iot_client.describeThingType(thingTypeName=thingTypeName)
+        response = iot_client.describeThingType(thingTypeName=typeName)
         return response.get('thingTypeArn')
     except botocore.exceptions.ClientError as error:
         if error.response['Error']['Code'] == 'ResourceNotFoundException':
-            print("ERROR: You need to configure the Thing Type [" + thingTypeName + "] in your target region first.")
+            print("ERROR: You need to configure the Thing Type [" + typeName + "] in your target region first.")
         if error.response['Error']['Code'] == 'UnauthorizedException':
             print("ERROR: There is a deployment problem with the attached Role. Unable to reach IoT Core object.")
         return None
@@ -157,22 +150,24 @@ def process_thing(thingName, certificateId, thingTypeName):
         return None
 
 def requeue():
-    sqs_client = boto3.client('sqs')
+    sqs_client = boto3client('sqs')
     queueUrl = os.environ.get('QUEUE_TARGET')
     sqs_client.send_message( QueueUrl=queueUrl,
-                         MessageBody=json.dumps(config))
+                             MessageBody=json.dumps(config))
+
+def get_certificate_fingerprint(certificate: x509.Certificate):
+    return binascii.hexlify(certificate.fingerprint(hashes.SHA256())).decode('UTF-8')
 
 def process_certificate(payload):
-    client = boto3.client('iot')
+    client = boto3client('iot')
 
     certificateText = base64.b64decode(eval(payload))
 
     # See if the certificate has already been registered.  If so, bail.
     certificateObj = x509.load_pem_x509_certificate(data=certificateText,
                                                     backend=default_backend())
 
-    fingerprint = binascii.hexlify(certificateObj.fingerprint(hashes.SHA256())).decode('UTF-8')
-    print("Fingerprint: " + fingerprint)
+    fingerprint = get_certificate_fingerprint(certificateObj)
 
     if (get_certificate(fingerprint)):
         try:
@@ -183,7 +178,6 @@ def process_certificate(payload):
         except:
             print("Certificate [" + fingerprint + "] not found in IoT Core. Importing.")
 
-        
     try:
         response = iot_client.register_certificate_without_ca(certificatePem=certificateText.decode('ascii'),
                                                               status='ACTIVE')

src/bulk_importer/schemas.py

@@ -0,0 +1,45 @@
+INPUT_SCHEMA = {
+    "$schema": "http://json-schema.org/draft-07/schema",
+    "$id": "http://example.com/example.json",
+    "type": "object",
+    "title": "Sample input schema",
+    "description": "The Document generation path parameters, Document Type and Customer ID.",
+    "required": ["pathParameters"],
+    "properties": {
+        "pathParameters" : { 
+            "$id": "#/properties/pathParameters",
+            "type": "object",
+            "required": ["docType", "customerId"],
+            "properties": {
+                "docType": {
+                    "$id": "#/properties/pathParameters/docType",
+                    "type": "string",
+                    "title": "The Document Type to Generate",
+                    "examples": ["TestDoc","WELCOME"],
+                    "maxLength": 30,
+                    },
+                "customerId": {
+                    "$id": "#/properties/pathParameters/customerId",
+                    "type": "string",
+                    "title": "The Customer ID to send the document",
+                    "examples": ["TestCustomer","TestCustomer01"],
+                    "maxLength": 30,
+                }
+            }
+        }
+    },
+}
+
+OUTPUT_SCHEMA = {
+    "$schema": "http://json-schema.org/draft-07/schema",
+    "$id": "http://example.com/example.json",
+    "type": "object",
+    "title": "Sample outgoing schema",
+    "description": "The root schema comprises the entire JSON document.",
+    "examples": [{"statusCode": 200, "body": "OK"}],
+    "required": ["statusCode", "body"],
+    "properties": {
+        "statusCode": {"$id": "#/properties/statusCode", "type": "integer", "title": "The statusCode"},
+        "body": {"$id": "#/properties/body", "type": "string", "title": "The response"}
+    },
+}

src/bulk_importer/testable.py

@@ -0,0 +1,19 @@
+from os import environ
+import warnings
+with warnings.catch_warnings():
+    from boto3 import resource
+
+_LAMBDA_SQS_RESOURCE = { "resource" : resource('sqs'),
+                         "queue_name" : environ.get("SQS_QUEUE_NAME","NONE") }
+
+class LambdaSQSClass:
+    """
+    AWS SQS Resource Class
+    """
+    def __init__(self, lambda_sqs_resource):  
+        """
+        Initialize an SQS Resource
+        """
+        self.resource = lambda_sqs_resource["resource"]
+        self.queue_name = lambda_sqs_resource["queue_name"]
+        self.queue = self.resource.Queue(self.queue_name)

src/provider_espressif/main.py

@@ -12,9 +12,6 @@
 from cryptography.hazmat.primitives import serialization
 import base64
 
-from schemas import INPUT_SCHEMA, OUTPUT_SCHEMA
-from testable import LambdaSQSClass, LambdaS3Class
-
 # Given a bucket and object, verify its existence and return the resource.
 def s3_object_stream(bucket_name: str, object_name: str):
     s3 = resource('s3')

test/artifacts/single.pem

@@ -0,0 +1,21 @@
+-----BEGIN CERTIFICATE-----
+MIIDfDCCAmSgAwIBAgIUR671iZbm5q0jCpoLgAGwH6oZKFwwDQYJKoZIhvcNAQEL
+BQAwUzELMAkGA1UEBhMCSU4xCzAJBgNVBAgMAk1IMQ0wCwYDVQQHDARQdW5lMRIw
+EAYDVQQKDAlFc3ByZXNzaWYxFDASBgNVBAMMC2VzcC10ZXN0LWNuMB4XDTI1MDQw
+MzE0MzEwOVoXDTM1MDQwMTE0MzEwOVowezELMAkGA1UEBhMCSU4xCzAJBgNVBAgM
+Ak1IMRowGAYDVQQKDBFFc3ByZXNzaWYgU3lzdGVtczEUMBIGA1UECwwLRXhwcmVz
+c0xpbmsxLTArBgNVBAMMJDZkMWQ1NGJhLWVkODEtNGMzMi04ZTA4LWY5ZjhjODg1
+YTZlZDCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAM+SHRhdFKzPogsu
+F7yKdyzSOPAGcuAMmaGMEb1WRLPMkI9yW2SERyTglOBoyXxUJUj49n5y7AKm5zp4
+nfljsF1DcV4pwgq7FUVjkko3YjGbanxyfwhHUb0wlXWp86S76QcdxbL9rA4pHPam
+M+qc0iy/2YmG18AuWpQ7gwfn3WpFxIeNgTpMP/AqCE/F5pLRaHWvrXXAbVrUa4Tm
+svpCvFwwalR+VjF74TpbRYfLbtRDS/vzgdRLRJzDmnBalc3+A5OcOQbIXOvOySwx
+5E2ermNhqCd0GOJ7L395Kh+2XV4XTCp7Cmsfhnoxo5/UuGuAYHZGio16GqpqZcY7
+5eYsi9kCAwEAAaMgMB4wDAYDVR0TAQH/BAIwADAOBgNVHQ8BAf8EBAMCBaAwDQYJ
+KoZIhvcNAQELBQADggEBAIxi68b/5AfznHOoSddeBDfkgtJV1a2a0n8w6NfGwmtm
+8i7PTAa8tvnUCyAYXvFeaaVsCDMYfSU4dw8vMS3lRPRiUJ09wizxyiOG5Af4q2iY
+msy561rxsdydrvf0TTVPNNwu5MBOyrIUWVZVtAg8SwBWZ7Q+4oTrYvpTkVXOKCXs
++uwxfk7ht4X9w8P6EjM1g1CKpCT9+iRpnD6Gi51ZVGFqIu1a0eWBEAi/zKsWfoc8
+lzeyB/DJbVaz7+MVnI+WIH5/PASI3tbFUq/vF32O1zTm1+PLvOEppPG17p981oEp
+Hz8WLhhGzqsnhW96DyIY4qUU2MxXfg5QzFFQz9jEoH4=
+-----END CERTIFICATE-----

test/unit/src/test_bulk_importer.py

@@ -0,0 +1,51 @@
+import sys
+import os
+import io
+import pytest
+
+import botocore
+
+from boto3 import resource, client
+from moto import mock_aws, settings
+from moto.settings import iot_use_valid_cert
+
+from aws_lambda_powertools.utilities.validation import validate
+
+from unittest import TestCase
+from unittest.mock import MagicMock, patch
+sys.path.append('./src/bulk_importer')
+os.environ['AWS_DEFAULT_REGION'] = "us-east-1"
+from src.bulk_importer.testable import LambdaSQSClass   # pylint: disable=wrong-import-position
+from src.bulk_importer.main import lambda_handler, get_certificate, get_certificate_fingerprint, get_certificate_arn, get_thing, get_policy, get_thing_group, get_thing_type, process_policy, process_thing, requeue, process_certificate, process_thing_group, get_name_from_certificate, process_sqs # pylint: disable=wrong-import-position
+from cryptography import x509
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import serialization
+import base64
+
+@mock_aws(config={'iot': {'use_valid_cert': True}})
+class TestBulkImporter(TestCase):
+    def setUp(self):
+        self.test_sqs_queue_name = "provider"
+        sqs_client = client('sqs', region_name="us-east-1")
+        sqs_client.create_queue(QueueName=self.test_sqs_queue_name)
+        mocked_sqs_resource = resource("sqs")
+        mocked_sqs_resource = { "resource" : resource('sqs'),
+                                "queue_name" : self.test_sqs_queue_name }
+        self.mocked_sqs_class = LambdaSQSClass(mocked_sqs_resource)
+
+    def test_pos_process_certificate(self):
+        with open('./test/artifacts/single.pem', 'rb') as data:
+            pem_obj = x509.load_pem_x509_certificate(data.read(),
+                                                     backend=default_backend())
+            block = pem_obj.public_bytes(encoding=serialization.Encoding.PEM).decode('ascii')
+            cert = str(base64.b64encode(block.encode('ascii')))
+            r = process_certificate(cert)
+            assert (r == get_certificate_fingerprint(pem_obj))
+
+    def tearDown(self):
+        sqs_resource = resource("sqs", region_name="us-east-1")
+        sqs_client = client("sqs", "us-east-1")
+        sqs_queue_url_r = sqs_client.get_queue_url(QueueName=self.test_sqs_queue_name)
+        sqs_queue_url = sqs_queue_url_r['QueueUrl']
+        sqs_resource = sqs_resource.Queue(url=sqs_queue_url)
+        sqs_resource.delete()

test/unit/src/test_provider_espressif.py

@@ -16,7 +16,7 @@
 from src.provider_espressif.testable import LambdaS3Class, LambdaSQSClass   # pylint: disable=wrong-import-position
 from src.provider_espressif.main import lambda_handler, s3_filebuf_bytes, invoke_export  # pylint: disable=wrong-import-position
 from src.provider_espressif.main import s3_object_stream
-from src.provider_espressif.main import INPUT_SCHEMA                     # pylint: disable=wrong-import-position
+from src.provider_espressif.schemas import INPUT_SCHEMA                     # pylint: disable=wrong-import-position
 
 @mock_aws
 class TestProviderEspressif(TestCase):