cr50: add basic U2F test to tpmtest, disable ecies

Add basic test to U2F generate, sign and attest commands to make sure
commands are processed correctly.

When build with CRYPTO_TEST=1 pretend that power button for U2F is
always pressed when requested to simulate user presence.

BUG=None
TEST=make BOARD=cr50 CRYPTO_TEST=1
tests/tpmtest.py

Signed-off-by: Vadim Sukhomlinov <sukhomlinov@google.com>
Change-Id: I8fda8037ea7322eb5fa46421ded6da3d1bba9c66
Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/platform/ec/+/3048103
Reviewed-by: Vadim Sukhomlinov <sukhomlinov@chromium.org>
Reviewed-by: Mary Ruthven <mruthven@chromium.org>
Tested-by: Vadim Sukhomlinov <sukhomlinov@chromium.org>
Commit-Queue: Vadim Sukhomlinov <sukhomlinov@chromium.org>
Auto-Submit: Vadim Sukhomlinov <sukhomlinov@chromium.org>

4 files changed, 132 insertions, 10 deletions

diff --git a/board/cr50/u2f.c b/board/cr50/u2f.c
index 750708012b..d1d44a2ffa 100644
--- a/board/cr50/u2f.c
+++ b/board/cr50/u2f.c
@@ -40,6 +40,9 @@ void power_button_record(void)
 
 enum touch_state pop_check_presence(int consume)
 {
+#ifdef CRYPTO_TEST_SETUP
+	return POP_TOUCH_YES;
+#else
 	int recent = ((last_press.val  > 0) &&
 		((get_time().val - last_press.val) < PRESENCE_TIMEOUT));
 
@@ -52,6 +55,7 @@ enum touch_state pop_check_presence(int consume)
 
 	/* user physical presence on the power button */
 	return recent ? POP_TOUCH_YES : POP_TOUCH_NO;
+#endif
 }
 
 static const uint8_t k_salt = NVMEM_VAR_G2F_SALT;
diff --git a/test/tpm_test/subcmd.py b/test/tpm_test/subcmd.py
index 9b3c3c9368..754fbf575d 100644
--- a/test/tpm_test/subcmd.py
+++ b/test/tpm_test/subcmd.py
@@ -13,6 +13,9 @@ ECC = 3
 FW_UPGRADE = 4
 HKDF = 5
 ECIES = 6
+U2F_GENERATE = 44
+U2F_SIGN = 45
+U2F_ATTEST = 46
 DRBG_TEST = 50
 # The same exception class used by all tpmtest modules.
 class TpmTestError(Exception):
diff --git a/test/tpm_test/tpmtest.py b/test/tpm_test/tpmtest.py
index 20e368e337..d664ca7567 100755
--- a/test/tpm_test/tpmtest.py
+++ b/test/tpm_test/tpmtest.py
@@ -30,6 +30,7 @@ import rsa_test
 import subcmd
 import trng_test
 import upgrade_test
+import u2f_test
 
 # Extension command for dcypto testing
 EXT_CMD = 0xbaccd00a
@@ -59,15 +60,18 @@ class TPM:
         if not self._handle.FtdiSpiInit(freq, debug_mode):
             raise subcmd.TpmTestError('Failed to connect')
 
-    def validate(self, data_blob, response_mode=False):
+    def parse_response(self, data_blob):
+        (tag, size, cmd_code, _) = struct.unpack_from(
+          self.HEADER_FMT, data_blob + b'  ')
+        return tag, size, cmd_code
+
+    def validate(self, data_blob, response_mode=False, check_cmd=True):
         """Validate TPM header format
 
         Check if a data blob complies with TPM command/response
         header format.
         """
-
-        (tag, size, cmd_code, _) = struct.unpack_from(
-          self.HEADER_FMT, data_blob + b'  ')
+        tag, size, cmd_code = self.parse_response(data_blob)
         prefix = 'Misformatted blob: '
         if tag not in (0x8001, 0x8002):
             raise subcmd.TpmTestError(prefix + 'bad tag value 0x%4.4x' % tag)
@@ -83,12 +87,14 @@ class TPM:
                 raise subcmd.TpmTestError(
                   prefix + 'invalid response code 0x%x' % cmd_code)
             return
+        if check_cmd == False:
+            return size, cmd_code
         if 0x11f <= cmd_code <= 0x18f:
-            return  # This is a valid command
+            return size, cmd_code # This is a valid command
         if cmd_code == EXT_CMD:
-            return  # This is an extension command
+            return size, cmd_code # This is an extension command
         if 0x20000000 <= cmd_code <= 0x200001ff:
-            return  # this is vendor command
+            return size, cmd_code # this is vendor command
         raise subcmd.TpmTestError(prefix + 'invalid command code 0x%x'
                                   % cmd_code)
 
@@ -99,6 +105,13 @@ class TPM:
         self.validate(response, response_mode=True)
         return response
 
+    def command_unchecked(self, cmd_data):
+        """Verify command header"""
+        self.validate(cmd_data)
+        response = self._handle.FtdiSendCommandAndWait(cmd_data)
+        size, cmd_code = self.validate(response, response_mode=False, check_cmd=False)
+        return response, size, cmd_code
+
     def wrap_ext_command(self, subcmd_code, cmd_body):
         """Wrap TPM command into extension command header"""
         return struct.pack(self.HEADER_FMT, 0x8001,
@@ -205,12 +218,14 @@ def main():
             drbg_test.drbg_test(tpm_object, drbg_request, drbg_expected,
                                 lab_output)
             sys.exit(0)
+        u2f_test.u2f_test(tpm_object)
         crypto_test.crypto_tests(tpm_object, os.path.join(ROOT_DIR,
-                                                          'crypto_test.xml'))
+                                                         'crypto_test.xml'))
         drbg_test.drbg_test(tpm_object, drbg_request, drbg_expected,
-                            lab_output)
+                           lab_output)
         ecc_test.ecc_test(tpm_object)
-        ecies_test.ecies_test(tpm_object)
+#       cr50 don't implement ecies
+#       ecies_test.ecies_test(tpm_object)
         hash_test.hash_test(tpm_object)
         hkdf_test.hkdf_test(tpm_object)
         rsa_test.rsa_test(tpm_object)
diff --git a/test/tpm_test/u2f_test.py b/test/tpm_test/u2f_test.py
new file mode 100644
index 0000000000..9cb7d5e609
--- /dev/null
+++ b/test/tpm_test/u2f_test.py
@@ -0,0 +1,100 @@
+# -*- coding: utf-8 -*-
+# Copyright 2021 The Chromium OS Authors. All rights reserved.
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+"""Module for testing u2f functions using extended commands."""
+from binascii import a2b_hex as a2b
+import hashlib
+import os
+import struct
+
+import subcmd
+import utils
+
+
+# U2F_GENERATE:
+# origin [32] | user [32] | flag | [auth_secret[32]]
+def u2f_generate(tpm, origin, user, flag, auth):
+    origin = origin[:32].ljust(32, b'\0')
+    user = user[:32].ljust(32, b'\0')
+    auth = auth[:32].ljust(32, b'\0')
+    cmd = origin + user + flag.to_bytes(1, 'big') + auth
+    wrapped_response = tpm.command(tpm.wrap_ext_command(subcmd.U2F_GENERATE, cmd))
+    response = tpm.unwrap_ext_response(subcmd.U2F_GENERATE, wrapped_response)
+    public_key = response[:65]
+    kh = response[65:]
+    return public_key, kh
+
+def u2f_sign(tpm, origin, user, auth, kh, msg, flag, fail=False):
+    origin = origin[:32].ljust(32, b'\0')
+    user = user[:32].ljust(32, b'\0')
+    auth = auth[:32].ljust(32, b'\0')
+    msg = msg[:32].ljust(32, b'\0')
+
+    # determine version by size of key handle
+    if (len(kh) == 64):
+        cmd = origin + user + kh + msg + flag.to_bytes(1, 'big')
+    else:
+        cmd = origin + user + auth + msg + flag.to_bytes(1, 'big') + kh
+
+    if fail==False:
+        wrapped_response = tpm.command(tpm.wrap_ext_command(
+                                       subcmd.U2F_SIGN, cmd))
+        response = tpm.unwrap_ext_response(subcmd.U2F_SIGN, wrapped_response)
+        sig = response[:64]
+    else:
+        response, size, response_code = tpm.command_unchecked(
+                            tpm.wrap_ext_command(subcmd.U2F_SIGN, cmd))
+        if size != 12:
+             raise subcmd.TpmTestError('Unexpected response: '
+                                       + utils.hex_dump(response))
+        print('response: ', hex(response_code))
+        return b''
+    return sig
+
+def u2f_attest(tpm, origin, user, challenge, kh, public_key, fail=False):
+    origin = origin[:32].ljust(32, b'\0')
+    user = user[:32].ljust(32, b'\0')
+    challenge = challenge[:32].ljust(32, b'\0')
+    g2f_cmd = b'\0' + origin + challenge + kh + public_key
+    cmd = user + b'\0' + len(g2f_cmd).to_bytes(1, 'big') + g2f_cmd
+    if fail==False:
+        wrapped_response = tpm.command(tpm.wrap_ext_command(
+                                       subcmd.U2F_ATTEST, cmd))
+        response = tpm.unwrap_ext_response(subcmd.U2F_ATTEST, wrapped_response)
+        sig = response[:64]
+    else:
+        response, size, response_code = tpm.command_unchecked(
+                            tpm.wrap_ext_command(subcmd.U2F_ATTEST, cmd))
+        if size != 12:
+             raise subcmd.TpmTestError('Unexpected response: '
+                                       + utils.hex_dump(response))
+        print('response: ', hex(response_code))
+        return b''
+    return sig
+
+def u2f_test(tpm):
+    """Run U2F tests"""
+    origin = b'1'
+    user = b'2'
+    auth = b'3'
+    msg = b'12345'
+    public_key1, khv1 = u2f_generate(tpm, origin, user, 0, auth)
+    print('key_handle v1 = ',utils.hex_dump(khv1), len(khv1))
+    print('public_key v1 = ',utils.hex_dump(public_key1), len(public_key1))
+
+    public_key2, khv2 = u2f_generate(tpm, origin, user, 8, auth)
+    print('key_handle v2 = ',utils.hex_dump(khv2), len(khv2))
+
+    sig1 = u2f_sign(tpm, origin, user, auth, khv1, msg, 2)
+    print('sig v1 = ',utils.hex_dump(sig1), len(sig1))
+
+    sig1 = u2f_sign(tpm, origin, user, auth, khv2, msg, 2)
+    print('sig v2 = ',utils.hex_dump(sig1), len(sig1))
+
+    sig1 = u2f_sign(tpm, user, origin, auth, khv2, msg, 2, fail=True)
+    print('sig v2 = ',utils.hex_dump(sig1), len(sig1))
+
+    sig_attest = u2f_attest(tpm, origin, user, auth, khv1, public_key1)
+    print('sig attest = ',utils.hex_dump(sig_attest), len(sig_attest))