add nonce check

This causes attestation to fail if the nonce passed in doesn't match the
nonce returned in the attestation document.

pycape/_attestation.py

@@ -33,13 +33,18 @@ def download_root_cert():
     return root_cert
 
 
-def parse_attestation(attestation, root_cert, checkDate=None):
+def parse_attestation(attestation, root_cert, nonce=None, checkDate=None):
     logger.debug("* Parsing attestation document...")
 
     doc = load_attestation_document(attestation)
 
     doc_cert = doc["certificate"]
     cabundle = doc["cabundle"]
+    doc_nonce = doc["nonce"]
+
+    if nonce is not None and nonce != doc_nonce:
+        raise RuntimeError("error validating nonce")
+
     logger.debug("* Attestation document parsed.")
 
     verify_attestation_signature(attestation, doc_cert)

pycape/_attestation_test.py

@@ -64,11 +64,16 @@ def test_parse_attestation(self):
         cert = create_child_cert(
             intermediate_cert, root_private_key, private_key, cert_subject, ca=False
         )
-        doc_bytes = create_attestation_doc(intermediate_cert, cert)
+
+        nonce = b"abcd1234"
+
+        doc_bytes = create_attestation_doc(intermediate_cert, cert, nonce)
         attestation = create_cose_1_sign_msg(doc_bytes, private_key)
 
         attestation_doc = attest.parse_attestation(
-            attestation, root_cert.public_bytes(Encoding.PEM)
+            attestation,
+            root_cert.public_bytes(Encoding.PEM),
+            nonce=nonce,
         )
         public_key = attestation_doc["public_key"]
         user_data = attestation_doc.get("user_data")
@@ -90,7 +95,9 @@ def test_verify_attestation_signature(self):
         cert = create_child_cert(
             intermediate_cert, root_private_key, private_key, cert_subject, ca=False
         )
-        doc_bytes = create_attestation_doc(intermediate_cert, cert)
+
+        nonce = b"abcd1234"
+        doc_bytes = create_attestation_doc(intermediate_cert, cert, nonce=nonce)
         attestation = create_cose_1_sign_msg(doc_bytes, private_key)
 
         payload = cbor2.loads(attestation)
@@ -113,7 +120,8 @@ def test_verify_cert_chain(self):
             intermediate_cert, root_private_key, private_key, cert_subject
         )
 
-        doc_bytes = create_attestation_doc(intermediate_cert, cert)
+        nonce = b"abcd1234"
+        doc_bytes = create_attestation_doc(intermediate_cert, cert, nonce)
         attestation = create_cose_1_sign_msg(doc_bytes, private_key)
 
         payload = cbor2.loads(attestation)
@@ -142,7 +150,8 @@ def test_verify_cert_chain_fails_if_bad_intermediate(self):
             intermediate_cert, root_private_key, private_key, cert_subject, ca=False
         )
 
-        doc_bytes = create_attestation_doc(intermediate_cert, cert)
+        nonce = b"abcd1234"
+        doc_bytes = create_attestation_doc(intermediate_cert, cert, nonce)
         attestation = create_cose_1_sign_msg(doc_bytes, private_key)
 
         payload = cbor2.loads(attestation)
@@ -167,11 +176,15 @@ def test_verify_pcrs(self):
         cert = create_child_cert(
             intermediate_cert, root_private_key, private_key, cert_subject, ca=False
         )
-        doc_bytes = create_attestation_doc(intermediate_cert, cert)
+
+        nonce = b"abcd1234"
+        doc_bytes = create_attestation_doc(intermediate_cert, cert, nonce)
         attestation = create_cose_1_sign_msg(doc_bytes, private_key)
 
         attestation_doc = attest.parse_attestation(
-            attestation, root_cert.public_bytes(Encoding.PEM)
+            attestation,
+            root_cert.public_bytes(Encoding.PEM),
+            nonce=nonce,
         )
 
         attest.verify_pcrs({"0": [b"pcrpcrpcr".hex()]}, attestation_doc)
@@ -190,16 +203,46 @@ def test_verify_pcrs_fail(self):
         cert = create_child_cert(
             intermediate_cert, root_private_key, private_key, cert_subject, ca=False
         )
-        doc_bytes = create_attestation_doc(intermediate_cert, cert)
+
+        nonce = b"abcd1234"
+        doc_bytes = create_attestation_doc(intermediate_cert, cert, nonce)
         attestation = create_cose_1_sign_msg(doc_bytes, private_key)
 
         attestation_doc = attest.parse_attestation(
-            attestation, root_cert.public_bytes(Encoding.PEM)
+            attestation,
+            root_cert.public_bytes(Encoding.PEM),
+            nonce=nonce,
         )
 
         with pytest.raises(Exception):
             attest.verify_pcrs({"0": [b"pcrpcr".hex()]}, attestation_doc)
 
+    def test_noncefail(self):
+        crv = P384
+        root_private_key = ec.generate_private_key(
+            crv.curve_obj, backend=default_backend()
+        )
+        private_key = ec.generate_private_key(crv.curve_obj, backend=default_backend())
+
+        root_cert = create_root_cert(root_private_key, root_subject)
+        intermediate_cert = create_child_cert(
+            root_cert, root_private_key, root_private_key, intermediate_subject, ca=True
+        )
+        cert = create_child_cert(
+            intermediate_cert, root_private_key, private_key, cert_subject, ca=False
+        )
+
+        nonce = b"abcd1234"
+        doc_bytes = create_attestation_doc(intermediate_cert, cert, nonce)
+        attestation = create_cose_1_sign_msg(doc_bytes, private_key)
+
+        with pytest.raises(RuntimeError):
+            attest.parse_attestation(
+                attestation,
+                root_cert.public_bytes(Encoding.PEM),
+                nonce="fake",
+            )
+
 
 def create_cose_1_sign_msg(payload, private_key):
     crv = P384
@@ -223,7 +266,7 @@ def create_cose_1_sign_msg(payload, private_key):
     return msg.encode(tag=False)
 
 
-def create_attestation_doc(intermediate_cert, cert):
+def create_attestation_doc(intermediate_cert, cert, nonce):
     cert = cert.public_bytes(Encoding.DER)
     intermediate_cert = intermediate_cert.public_bytes(Encoding.DER)
     user_data = json.dumps({"func_hash": "stuff"})
@@ -237,6 +280,7 @@ def create_attestation_doc(intermediate_cert, cert):
         "cabundle": [intermediate_cert],
         "public_key": public_key,
         "user_data": user_data,
+        "nonce": nonce,
     }
 
     return cbor2.encoder.dumps(obj)

pycape/cape.py

@@ -631,8 +631,7 @@ def __init__(self, endpoint, auth_protocol, auth_token, root_cert):
         self._websocket = None
         self._public_key = None
 
-    async def authenticate(self):
-        nonce = _generate_nonce()
+    async def authenticate(self, nonce):
         request = _create_connection_request(nonce)
         _logger.debug("\n> Sending authentication request...")
         await self._websocket.send(request)
@@ -651,8 +650,11 @@ async def bootstrap(self, pcrs: Optional[Dict[str, List[str]]] = None):
         )
         _logger.debug("* Websocket connection established")
 
-        auth_response = await self.authenticate()
-        attestation_doc = attest.parse_attestation(auth_response, self._root_cert)
+        nonce = _generate_nonce()
+        auth_response = await self.authenticate(nonce)
+        attestation_doc = attest.parse_attestation(
+            auth_response, self._root_cert, nonce=nonce
+        )
         self._public_key = attestation_doc["public_key"]
 
         if pcrs is not None:
@@ -686,21 +688,20 @@ async def invoke(self, inputs: bytes) -> bytes:
         return _parse_wss_response(invoke_response)
 
 
-# TODO What should be the length?
-def _generate_nonce(length=8):
+def _generate_nonce(length=16):
     """
     Generates a string of digits between 0 and 9 of a given length
     """
     nonce = "".join([str(random.randint(0, 9)) for i in range(length)])
     _logger.debug(f"* Generated nonce: {nonce}")
-    return nonce
+    return nonce.encode()
 
 
 def _create_connection_request(nonce):
     """
     Returns a json string with nonce
     """
-    request = {"message": {"nonce": nonce}}
+    request = {"message": {"nonce": base64.b64encode(nonce).decode()}}
     return json.dumps(request)
 
 

pycape/cape_test.py

@@ -12,13 +12,16 @@ class TestCape(unittest.TestCase):
     def test_generate_nonce(self):
         length = 8
         nonce = _generate_nonce(length=length)
-        self.assertTrue(isinstance(nonce, str))
+        self.assertTrue(isinstance(nonce, bytes))
         self.assertTrue(len(nonce), length)
 
     def test_create_connection_request(self):
-        nonce = "90444145"
+        nonce = b"90444145"
         conn_req = _create_connection_request(nonce)
-        self.assertEqual(conn_req, json.dumps({"message": {"nonce": "90444145"}}))
+        self.assertEqual(
+            conn_req,
+            json.dumps({"message": {"nonce": base64.b64encode(b"90444145").decode()}}),
+        )
 
     def test_handle_expected_field(self):
         response = '{"message": "connected"}'