Merge pull request #13 from aws-samples/atalla_migration

atalla migration scripts - initial changes

Author: markcline

LICENSE

@@ -11,4 +11,6 @@ INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
 PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
 HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
 OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
-SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+Using this sample code will require access to Atalla payment hsm hardware that is not provided as part of this sample code and that needs to be provisioned separately by the relevant manufacturers or their resellers. This sample code does not imply an endorsement by any third parties mentioned.

key-import-export/hsm/atalla/assets/atalla-apc-tr34-key-exchange-sequence-diagram - Key Exchange.png

@@ -0,0 +1,146 @@
+import boto3
+import base64
+from botocore.exceptions import ClientError
+from cryptography.hazmat.primitives import serialization
+from cryptography import x509
+import logging
+
+apc_client = boto3.client('payment-cryptography')
+publicKeyCertificateAliasName = 'alias/tr34-key-import-kdh-ca'
+#un-comment to see debug logs for AWS SDK
+#boto3.set_stream_logger('', logging.DEBUG)
+#boto3.set_stream_logger('boto3.resources', logging.INFO)
+
+def verify_certificate_state(key_arn):
+    """
+    Verifies the state of the imported certificate
+    
+    Args:
+        key_arn (str): ARN of the imported key
+        
+    Returns:
+        bool: True if certificate is enabled and complete, False otherwise
+    """
+    try:
+        response = apc_client.describe_key(
+            KeyIdentifier=key_arn
+        )
+        
+        key_state = response['Key']['KeyState']
+        is_enabled = response['Key']['Enabled']
+        
+        if key_state == 'CREATE_COMPLETE' and is_enabled:
+            print(f"Certificate status: {key_state}, Enabled: {is_enabled}")
+            return True
+        else:
+            print(f"Certificate not ready. Status: {key_state}, Enabled: {is_enabled}")
+            return False
+            
+    except ClientError as e:
+        print(f"Error verifying certificate state: {e}")
+        return False
+
+def importPublicCACertificate(kdhPublicKeyCertificate):
+    cert_data = kdhPublicKeyCertificate.encode('ascii')
+    """ with open(publicKeyCertificatePath, 'rb') as cert_file:
+        cert_data = cert_file.read() """
+    # Load the certificate
+    cert = x509.load_pem_x509_certificate(cert_data)
+    
+    #keyAlias = apc_client.get_alias(AliasName=publicKeyCertificateAliasName)
+    
+    kdh_ca_key_arn = apc_client.import_key(Enabled=True, KeyMaterial={
+            'RootCertificatePublicKey': {
+                'KeyAttributes': {
+                    'KeyAlgorithm': 'RSA_2048',
+                    'KeyClass': 'PUBLIC_KEY',
+                    'KeyModesOfUse': {
+                        'Verify': True,
+                    },
+                    'KeyUsage': 'TR31_S0_ASYMMETRIC_KEY_FOR_DIGITAL_SIGNATURE',
+                },
+                'PublicKeyCertificate': base64.b64encode(cert.public_bytes(encoding=serialization.Encoding.PEM)).decode('UTF-8')
+            }
+        }, KeyCheckValueAlgorithm='ANSI_X9_24', Tags=[])['Key']['KeyArn']
+    
+    tagResponse = apc_client.tag_resource(
+        ResourceArn= kdh_ca_key_arn,
+        Tags=[
+            {
+                'Key': 'project',
+                'Value': 'sample-atalla-tr34-exchange'
+            },
+        ]
+    )
+    """     aliasResponse = apc_client.create_alias(
+            AliasName=publicKeyCertificateAliasName,
+            KeyArn=kdh_ca_key_arn
+        )
+        print(f"aliasResponse: {aliasResponse}")
+    """
+    print(f"Imported KDH CA certificate into APC with key ARN: {kdh_ca_key_arn}")
+    return kdh_ca_key_arn
+
+def getKeyIfExixts(alias_name):
+    try:
+        keyAlias = apc_client.get_alias(AliasName=alias_name)
+        
+        # Check if response has the expected structure
+        if (keyAlias and 
+            isinstance(keyAlias, dict) and 
+            'Alias' in keyAlias and 
+            isinstance(keyAlias['Alias'], dict) and 
+            'AliasName' in keyAlias['Alias']):
+            
+            return keyAlias['Alias']['AliasName']
+        return None
+        
+    except apc_client.exceptions.NotFoundException:
+        return None
+
+def deleteKeyIfExixts(key_arn):
+    try:
+        payment_crypto = boto3.client('payment-cryptography')
+    
+        # Schedule the key for deletion
+        response = payment_crypto.delete_key(
+            KeyIdentifier=key_arn,
+            DeleteKeyInDays=3
+        )
+    
+        print(f"Key {key_arn} scheduled for deletion in 3 days")
+        return True
+        
+    except payment_crypto.exceptions.NotFoundException:
+        print(f"Key {key_arn} not found")
+        return False
+    except payment_crypto.exceptions.InvalidStateException:
+        print(f"Key {key_arn} is in an invalid state for deletion")
+        return False
+    except Exception as e:
+        print(f"Error deleting key: {str(e)}")
+        return False
+            
+def importTR34Payload(tr34Payload,nonce,kdh_ca_key_arn,kdh_ca_certificate,importToken):
+    print("Importing TR34 payload into APC ...")
+    # Load the certificate
+    cert = x509.load_pem_x509_certificate(kdh_ca_certificate.encode('ascii'))
+
+    trt34_import_res = apc_client.import_key(
+            Enabled=True,
+            KeyMaterial={
+                "Tr34KeyBlock": {
+                    'CertificateAuthorityPublicKeyIdentifier': kdh_ca_key_arn,
+                    'ImportToken': importToken,
+                    'KeyBlockFormat': 'X9_TR34_2012',
+                    'SigningKeyCertificate': base64.b64encode(cert.public_bytes(encoding=serialization.Encoding.PEM)).decode('UTF-8'),
+                    'WrappedKeyBlock': tr34Payload,
+                    'RandomNonce': nonce,
+                }
+            },
+            KeyCheckValueAlgorithm='ANSI_X9_24',
+            Tags= [{"Key": "Type", "Value" : "Atalla-Test"}]
+        )
+    KeyArn=trt34_import_res['Key']['KeyArn']
+    
+    print(f"Imported TR34 payload with key ARN: {KeyArn}")

key-import-export/hsm/atalla/atalla_to_apc_tr34.py

@@ -0,0 +1,43 @@
+import re
+import time
+import socket
+import binascii
+from typing import Tuple
+
+def createRsaKey(atalla_address):
+    print("Generating RSA key pair from Atalla using command 120")
+    #generate rsa key pair
+    data = '<120#w#010001#2048#>'
+    response = send_data(atalla_address, data.encode(), '<220#(.*?)#(.*?)#(.*?)#>', b'>') #<220#Public Key#Private Key#Check Digits#[Key slot#]>
+    pubKey = response[0]
+    privKey = response[1]
+    return pubKey,privKey
+
+def sign139(keyReference, message, atalla_address ):
+    print("Signing with Atalla using command 139")
+    #valjue of 5 means sha-256
+    data = '<139#5#1#%s######%s#>' % (binascii.hexlify(message).decode().upper(), keyReference)
+    response = send_data(atalla_address, data.encode(), '<239#.*?#(.*?)#.*>', b'>')
+    return response[0]
+
+def send_data(target: Tuple[str, int], data: bytes, pattern: str, terminator: bytes = b''):
+
+    s = socket.create_connection(target, timeout=10)
+    s.settimeout(30)
+    s.sendall(data)
+
+    end_time = time.time() + 30
+
+    output = b''
+    while end_time > time.time():
+        try:
+            output += s.recv(2**16)
+            if terminator and terminator in output:
+                break
+        except socket.timeout:
+            break
+
+    matcher = re.match(pattern, output.decode())
+    if not matcher:
+        raise Exception('Failed to parse data ' + output.decode() + ' for command ' + data.decode())
+    return matcher.groups()

key-import-export/hsm/atalla/helpers/apc_helper.py

@@ -0,0 +1,170 @@
+import boto3
+import time
+from botocore.exceptions import ClientError
+
+TAG_KEY = "APC_TR34"
+controlplane_client = boto3.client("payment-cryptography")
+private_ca = boto3.client("acm-pca")
+
+def create_certificate_authority():
+    # create Certificate Authority for short-lived certificates
+    ca_arn = private_ca.create_certificate_authority(
+        CertificateAuthorityConfiguration={
+            'KeyAlgorithm': 'RSA_2048',
+            'SigningAlgorithm': 'SHA256WITHRSA',
+            'Subject': {
+                'Country': 'US',
+                'Organization': 'AWS Samples',
+                'OrganizationalUnit': 'APC',
+                'State': 'CA',
+                'CommonName': 'AWS Samples',
+            }
+        },
+        CertificateAuthorityType='ROOT',
+        UsageMode='SHORT_LIVED_CERTIFICATE'
+    )['CertificateAuthorityArn']
+
+    state = "CREATING"
+    while state == "CREATING":
+        time.sleep(1)
+        state = private_ca.describe_certificate_authority(CertificateAuthorityArn=ca_arn)['CertificateAuthority'][
+            'Status']
+        print(state)
+    return ca_arn
+
+def create_private_ca():
+    print("Creating AWS Private CA")
+    cert_authority_arn = create_certificate_authority()
+    print("Newly created Private CA ARN: %s" % cert_authority_arn)
+    # add tag to CA
+    private_ca.tag_certificate_authority(
+        CertificateAuthorityArn=cert_authority_arn,
+        Tags=[
+            {
+                'Key': TAG_KEY,
+                'Value': 'Sample'
+            },
+        ]
+    )
+    print("Getting root CA CSR")
+    csr = private_ca.get_certificate_authority_csr(CertificateAuthorityArn=cert_authority_arn)['Csr']
+    print("self-signing root CA CSR")
+    certificate, chain = sign_with_private_ca(cert_authority_arn, csr, {
+        'Value': 10,
+        'Type': 'YEARS'
+    }, template='arn:aws:acm-pca:::template/RootCACertificate/V1')
+    print("Importing signed certificate as ROOT")
+    private_ca.import_certificate_authority_certificate(CertificateAuthorityArn=cert_authority_arn,
+                                                        Certificate=certificate)
+    print("CA Setup complete")
+    return cert_authority_arn
+
+def find_or_create_private_ca():
+    # find existing CA
+    for ca in private_ca.list_certificate_authorities()['CertificateAuthorities']:
+        if ca['Status'] == 'ACTIVE':
+            # get ca tags
+            tags = private_ca.list_tags(CertificateAuthorityArn=ca['Arn'])
+            for tag in tags['Tags']:
+                if tag['Key'] == TAG_KEY:
+                    # if tag is present, use this CA
+                    return ca['Arn']
+    return create_private_ca()
+
+def issue_certificate(csr_content):
+    """
+    Issues a certificate using AWS Private CA
+    
+    Args:
+        ca_arn (str): ARN of the private CA
+        csr_file_path (str): Path to the CSR file
+    
+    Returns:
+        str: Certificate ARN if successful, None otherwise
+    """
+    try:
+        # Create ACM PCA client
+        acmpca_client = boto3.client('acm-pca')
+        ca_arn = find_or_create_private_ca()
+        # Request to issue certificate
+        response = acmpca_client.issue_certificate(
+            CertificateAuthorityArn=ca_arn,
+            Csr=csr_content,
+            SigningAlgorithm='SHA256WITHRSA',
+            Validity={
+                'Value': 7,
+                'Type': 'DAYS'
+            },
+            TemplateArn='arn:aws:acm-pca:::template/EndEntityCertificate/V1'
+        )
+        
+        certificate_arn = response['CertificateArn']
+        
+        # Wait for certificate to be issued
+        waiter = acmpca_client.get_waiter('certificate_issued')
+        waiter.wait(
+            CertificateAuthorityArn=ca_arn,
+            CertificateArn=certificate_arn
+        )
+        
+        # Get the issued certificate
+        response = acmpca_client.get_certificate(
+            CertificateAuthorityArn=ca_arn,
+            CertificateArn=certificate_arn
+        )
+        
+        certificate = response['Certificate']
+        certificate_chain = response['CertificateChain']
+        
+        """ # Save the certificate and chain to files
+        with open('certificate.pem', 'w') as f:
+            f.write(certificate)
+        
+        with open('certificate_chain.pem', 'w') as f:
+            f.write(certificate_chain) """
+            
+        return certificate, certificate_chain
+        
+    except ClientError as e:
+        print(f"Error issuing certificate: {e}")
+        return None
+    
+def sign_with_private_ca(ca_arn, csr, validity, template="arn:aws:acm-pca:::template/EndEntityCertificate/V1"):
+        """
+        Signs the client-side Key with AWS Private CA and returns the Certificate and Certificate Chain
+        :param validity:
+        :param ca_arn:
+        :param csr: Certificate Signing Request
+        :param template: Template ARN to use for the certificate
+        :return:
+        """
+        client = boto3.client('acm-pca')
+        response = client.issue_certificate(
+            CertificateAuthorityArn=ca_arn,
+            Csr=csr,
+            TemplateArn=template,
+            SigningAlgorithm='SHA256WITHRSA',
+            Validity=validity
+        )
+        certificate_arn = response['CertificateArn']
+        time.sleep(0.5)
+
+        while 1:
+            try:
+                certificate_response = client.get_certificate(CertificateArn=certificate_arn,
+                                                              CertificateAuthorityArn=ca_arn)
+                if 'CertificateChain' in certificate_response:
+                    chain = certificate_response['CertificateChain']
+                else:
+                    chain = None
+                return certificate_response['Certificate'], chain
+            except client.exceptions.RequestInProgressException:
+                time.sleep(0.1)    
+
+""" def setup():
+    ca_arn = find_or_create_private_ca()
+    #ca_certificate = private_ca.get_certificate_authority_certificate(CertificateAuthorityArn=ca_arn)['Certificate']
+    print("CA Certificate: %s" % ca_arn)
+    return ca_arn, ca_arn
+
+setup() """