Flesh out attestation certs and largeBlobStore

Author: BryanJacobs

install_attestation_cert.py

@@ -2,11 +2,16 @@
 
 import sys
 import argparse
+import base64
 
 from fido2.ctap2 import Ctap2
 from fido2.ctap2.base import args as ctap_args
 from fido2.pcsc import CtapPcscDevice
 
+import secrets
+
+from cryptography.hazmat.primitives.asymmetric import ec
+
 from python_tests.ctap.ctap_test import BasicAttestationTestCase
 
 if __name__ == '__main__':
@@ -17,17 +22,47 @@
     parser.add_argument('--aaguid',
                         default=None,
                         help='AAGUID to use, expressed as 16 hex bytes (32-character-long string)')
+    parser.add_argument('--ca-cert-bytes',
+                        default=None,
+                        help='CA certificate, expressed as base64-encoded DER')
+    parser.add_argument('--ca-private-key',
+                        default=None,
+                        help='CA private key, expressed as a hex string')
+    parser.add_argument('--org',
+                        default='ACME',
+                        help='Organization name to use for certificates')
+    parser.add_argument('--country',
+                        default='US',
+                        help='ISO country code to use for certificates')
     args = parser.parse_args()
 
+    if (args.ca_private_key is None) != (args.ca_cert_bytes is None):
+        raise IllegalArgumentException("Either both or neither of CA certificate and private key must be set")
+
     aaguid = None
     if args.aaguid is not None:
         if len(args.aaguid) != 32:
             sys.stderr.write("Invalid AAGUID length!\n")
             sys.exit(1)
         aaguid = bytes.fromhex(args.aaguid)
+    else:
+        aaguid = secrets.token_bytes(16)
 
     tc = BasicAttestationTestCase()
-    at_bytes = tc.gen_attestation_cert(name=args.name, aaguid=aaguid)
+    if args.ca_private_key is None:
+        ca_privkey_and_cert = tc.get_ca_cert(org=args.org)
+    else:
+        ca_privkey_and_cert = bytes.fromhex(args.ca_private_key), base64.b64decode(args.ca_cert_bytes)
+    private_key = ec.generate_private_key(ec.SECP256R1())
+
+    cert_bytes = tc.get_x509_certs(private_key, name=args.name, ca_privkey_and_cert=ca_privkey_and_cert,
+                                   org=args.org, country=args.country)
+    print(f"Using AAGUID: {aaguid.hex()}")
+    print(f"Using CA certificate: {base64.b64encode(cert_bytes[-1])}")
+
+    at_bytes = tc.assemble_cbor_from_attestation_certs(private_key=private_key,
+                                                       cert_bytes=cert_bytes,
+                                                       aaguid=aaguid)
 
     devices = list(CtapPcscDevice.list_devices())
     if len(devices) > 1:
@@ -42,4 +77,4 @@
         0x46,
         ctap_args(at_bytes)
     )
-    print(f"Got response: {res}")
+    print(f"Got response: {res} (empty is good)")

mds.json

@@ -31,7 +31,9 @@
   "cryptoStrength": 128,
   "attachmentHint": ["nfc"],
   "tcDisplay": [],
-  "attestationRootCertificates": [],
+  "attestationRootCertificates": [
+
+  ],
   "authenticatorGetInfo": {
       "versions": [ "FIDO_2_0", "FIDO_2_1", "FIDO_2_1_PRE", "U2F_V2" ],
       "extensions": [ "uvm", "credBlob", "credProtect", "hmac-secret", "largeBlobKey" ],

python_tests/ctap/ctap_test.py

@@ -476,35 +476,14 @@ def install_attestation_cert(self, **kwargs):
     def _short_to_bytes(self, b: int) -> list[int]:
         return [(b & 0xFF00) >> 8, b & 0x00FF]
 
-    def get_x509_certs(self, private_key: EllipticCurvePrivateKey, name: Optional[str] = None,
-                       country: Optional[str] = "US", org: Optional[str] = "ACME") -> list[bytes]:
-        self.public_key = private_key.public_key()
-
-        if name is None:
-            name = secrets.token_hex(4)
-
+    def gen_authenticator_cert_from_ca(self, name: str,
+                                       ca_name: x509.Name,
+                                       ca_privkey: EllipticCurvePrivateKey,
+                                       country: Optional[str] = "US",
+                                       org: Optional[str] = "ACME",
+                                       ) -> bytes:
         now = datetime.now()
 
-        ca_privkey = ec.generate_private_key(ec.SECP256R1())
-        ca_pubkey = ca_privkey.public_key()
-        self.ca_public_key = ca_pubkey
-
-        ca_name = x509.Name([
-            x509.NameAttribute(NameOID.ORGANIZATION_NAME, org),
-            x509.NameAttribute(NameOID.COMMON_NAME, "AuthCA"),
-        ])
-        ca_cert_bytes = (
-            x509.CertificateBuilder()
-            .subject_name(ca_name)
-            .issuer_name(ca_name)
-            .serial_number(x509.random_serial_number())
-            .public_key(ca_pubkey)
-            .not_valid_before(now - timedelta(days=1))
-            .not_valid_after(now + timedelta(days=3650))
-            .sign(private_key=ca_privkey, algorithm=hashes.SHA256())
-            .public_bytes(Encoding.DER)
-        )
-
         this_cert_name = x509.Name([
             x509.NameAttribute(NameOID.COUNTRY_NAME, country),
             x509.NameAttribute(NameOID.ORGANIZATION_NAME, org),
@@ -530,19 +509,60 @@ def get_x509_certs(self, private_key: EllipticCurvePrivateKey, name: Optional[st
             .sign(private_key=ca_privkey, algorithm=hashes.SHA256())
             .public_bytes(Encoding.DER)
         )
-        return [authenticator_cert_bytes, ca_cert_bytes]
 
-    def gen_attestation_cert(self, cert_bytes: Optional[list[bytes]] = None, name: Optional[str] = None,
-                             aaguid: Optional[bytes] = None) -> bytes:
-        if aaguid is None:
-            aaguid = secrets.token_bytes(16)
+        return authenticator_cert_bytes
 
-        self.aaguid = aaguid
-        private_key = ec.generate_private_key(ec.SECP256R1())
+    def get_ca_cert(self, org: str) -> tuple[bytes, bytes]:
+        now = datetime.now()
 
-        if cert_bytes is None:
-            cert_bytes = self.get_x509_certs(private_key, name)
+        ca_privkey = ec.generate_private_key(ec.SECP256R1())
+        ca_pubkey = ca_privkey.public_key()
+        self.ca_public_key = ca_pubkey
+
+        ca_name = x509.Name([
+            x509.NameAttribute(NameOID.ORGANIZATION_NAME, org),
+            x509.NameAttribute(NameOID.COMMON_NAME, "AuthCA"),
+        ])
+        return ca_privkey, (
+            x509.CertificateBuilder()
+            .subject_name(ca_name)
+            .issuer_name(ca_name)
+            .serial_number(x509.random_serial_number())
+            .public_key(ca_pubkey)
+            .not_valid_before(now - timedelta(days=1))
+            .not_valid_after(now + timedelta(days=3650))
+            .sign(private_key=ca_privkey, algorithm=hashes.SHA256())
+            .public_bytes(Encoding.DER)
+        )
+
+    def get_x509_certs(self, private_key: EllipticCurvePrivateKey, name: Optional[str] = None,
+                       country: Optional[str] = "US", org: Optional[str] = "ACME",
+                       ca_privkey_and_cert: Optional[tuple[bytes, bytes]] = None) -> list[bytes]:
+        self.public_key = private_key.public_key()
+
+        if name is None:
+            name = secrets.token_hex(4)
 
+        if ca_privkey_and_cert is None:
+            ca_privkey_and_cert = self.get_ca_cert(org)
+
+        ca_privkey, ca_cert_bytes = ca_privkey_and_cert
+
+        loaded_cacert = x509.load_der_x509_certificate(ca_cert_bytes)
+        ca_name = loaded_cacert.subject
+
+        authenticator_cert_bytes = self.gen_authenticator_cert_from_ca(
+            name=name,
+            ca_name=ca_name,
+            ca_privkey=ca_privkey,
+            country=country,
+            org=org
+        )
+
+        return [authenticator_cert_bytes, ca_cert_bytes]
+
+    def assemble_cbor_from_attestation_certs(self, private_key: bytes, cert_bytes: list[bytes],
+                                             aaguid: bytes) -> bytes:
         num_certs = len(cert_bytes)
         self.cert = cert_bytes[0]
         cert_cbor_bytes = [0x80 + num_certs]
@@ -564,9 +584,24 @@ def gen_attestation_cert(self, cert_bytes: Optional[list[bytes]] = None, name: O
         private_bytes = s.to_bytes(length=32)
         self.assertEqual(32, len(private_bytes))
         cbor_len_bytes = bytes(self._short_to_bytes(len(cert_cbor)))
-        res = self.aaguid + private_bytes + cbor_len_bytes + cert_cbor
+        res = aaguid + private_bytes + cbor_len_bytes + cert_cbor
         return res
 
+    def gen_attestation_cert(self, cert_bytes: Optional[list[bytes]] = None, name: Optional[str] = None,
+                             aaguid: Optional[bytes] = None) -> bytes:
+        if aaguid is None:
+            aaguid = secrets.token_bytes(16)
+
+        self.aaguid = aaguid
+        private_key = ec.generate_private_key(ec.SECP256R1())
+
+        if cert_bytes is None:
+            cert_bytes = self.get_x509_certs(private_key, name)
+
+        return self.assemble_cbor_from_attestation_certs(private_key=private_key,
+                                                         cert_bytes=cert_bytes,
+                                                         aaguid=aaguid)
+
 
 class FixedPinUserInteraction(UserInteraction):
     pin: str

src/main/java/us/q3q/fido2/FIDO2Applet.java

@@ -348,7 +348,8 @@ public final class FIDO2Applet extends Applet implements ExtendedLength {
     /**
      * Storage for the largeBlobs extension
      */
-    private final byte[][] largeBlobStores;
+    private final byte[] largeBlobStoreA;
+    private final byte[] largeBlobStoreB;
     /**
      * Which large blob store is in use
      */
@@ -3648,6 +3649,20 @@ public void process(APDU apdu) throws ISOException {
         transientStorage.resetChainIncomingReadOffset();
     }
 
+    private byte[] getCurrentLargeBlobStore() {
+        if (largeBlobStoreIndex == 0) {
+            return largeBlobStoreA;
+        }
+        return largeBlobStoreB;
+    }
+
+    private byte[] getInactiveLargeBlobStore() {
+        if (largeBlobStoreIndex == 1) {
+            return largeBlobStoreA;
+        }
+        return largeBlobStoreB;
+    }
+
     /**
      * Processes CTAP2 largeBlob extension commands
      *
@@ -3773,7 +3788,9 @@ private void handleLargeBlobs(APDU apdu, byte[] reqBuffer, short lc) {
             sendErrorByte(apdu, FIDOConstants.CTAP1_ERR_INVALID_PARAMETER);
         }
 
-        if (offset < 0 || offset > (short) largeBlobStores[largeBlobStoreIndex].length) {
+        byte[] currentLargeBlobStore = getCurrentLargeBlobStore();
+
+        if (offset < 0 || offset > (short) currentLargeBlobStore.length) {
             // Offset is mandatory and must be reasonable
             sendErrorByte(apdu, FIDOConstants.CTAP1_ERR_INVALID_PARAMETER);
         }
@@ -3804,7 +3821,7 @@ private void handleLargeBlobs(APDU apdu, byte[] reqBuffer, short lc) {
                 if (setTotalLength < 17) {
                     sendErrorByte(apdu, FIDOConstants.CTAP1_ERR_INVALID_PARAMETER);
                 }
-                if (setTotalLength > (short) largeBlobStores[largeBlobStoreIndex].length) {
+                if (setTotalLength > (short) currentLargeBlobStore.length) {
                     sendErrorByte(apdu, FIDOConstants.CTAP2_ERR_LARGE_BLOB_STORAGE_FULL);
                 }
             } else {
@@ -3875,9 +3892,9 @@ private void handleLargeBlobs(APDU apdu, byte[] reqBuffer, short lc) {
      */
     private void handleLargeBlobSet(APDU apdu, byte[] buffer, short incomingDataOffset,
                                     short blobWriteOffset, short incomingDataLength, short totalLength) {
-        final byte tempBlobStoreIndex = (byte)(largeBlobStoreIndex == 0 ? 1 : 0);
+        final byte[] inactiveLargeBlobStore = getInactiveLargeBlobStore();
         Util.arrayCopyNonAtomic(buffer, incomingDataOffset,
-                largeBlobStores[tempBlobStoreIndex], blobWriteOffset, incomingDataLength);
+                inactiveLargeBlobStore, blobWriteOffset, incomingDataLength);
         // Done with incoming request now
         bufferManager.informAPDUBufferAvailability(apdu, (short) 0xFF);
 
@@ -3890,22 +3907,23 @@ private void handleLargeBlobSet(APDU apdu, byte[] buffer, short incomingDataOffs
             byte[] scratch = bufferManager.getBufferForHandle(apdu, scratchHandle);
             short scratchOffset = bufferManager.getOffsetForHandle(scratchHandle);
 
-            sha256.doFinal(largeBlobStores[tempBlobStoreIndex], (short) 0, (short)(totalLength - 16),
+            sha256.doFinal(inactiveLargeBlobStore, (short) 0, (short)(totalLength - 16),
                     scratch, scratchOffset);
 
             if (Util.arrayCompare(scratch, scratchOffset,
-                    largeBlobStores[tempBlobStoreIndex], (short)(totalLength - 16), (short) 16) != 0) {
+                    inactiveLargeBlobStore, (short)(totalLength - 16), (short) 16) != 0) {
                 // hash mismatch
                 sendErrorByte(apdu, FIDOConstants.CTAP2_ERR_INTEGRITY_FAILURE);
             }
 
             bufferManager.release(apdu, scratchHandle, (short) 32);
 
             // Swapperoo the buffers
+            final byte newBlobStoreIndex = (byte)(largeBlobStoreIndex == 0 ? 1 : 0);
             JCSystem.beginTransaction();
             boolean ok = false;
             try {
-                largeBlobStoreIndex = tempBlobStoreIndex;
+                largeBlobStoreIndex = newBlobStoreIndex;
                 largeBlobStoreFill = totalLength;
             } finally {
                 if (ok) {
@@ -3952,7 +3970,7 @@ private void handleLargeBlobGet(APDU apdu, short numBytes, short offset) {
         outBuffer[writeIdx++] = (byte) 0xA1; // map - one key
         outBuffer[writeIdx++] = (byte) 0x01; // map key: config
         writeIdx = encodeIntLenTo(outBuffer, writeIdx, bytesToRetrieve, true);
-        writeIdx = Util.arrayCopyNonAtomic(largeBlobStores[largeBlobStoreIndex], offset,
+        writeIdx = Util.arrayCopyNonAtomic(getCurrentLargeBlobStore(), offset,
                 outBuffer, writeIdx, bytesToRetrieve);
 
         if (outBuffer == bufferMem) {
@@ -5374,10 +5392,12 @@ private void authenticatorReset(APDU apdu) {
         final byte realBlobStoreIndex = largeBlobStoreIndex;
 
         // Empty the large blob store OUTside the main transaction, since it's non-precious and double buffered
-        Util.arrayFillNonAtomic(largeBlobStores[tempBlobStoreIndex], (short) 0,
-                (short) largeBlobStores[tempBlobStoreIndex].length, (byte) 0x00);
+        final byte[] inactiveLargeBlobStore = getInactiveLargeBlobStore();
+        final byte[] activeLargeBlobStore = getCurrentLargeBlobStore();
+        Util.arrayFillNonAtomic(inactiveLargeBlobStore, (short) 0,
+                (short) inactiveLargeBlobStore.length, (byte) 0x00);
         Util.arrayCopyNonAtomic(CannedCBOR.INITIAL_LARGE_BLOB_ARRAY, (short) 0,
-                largeBlobStores[tempBlobStoreIndex], (short) 0, (short) CannedCBOR.INITIAL_LARGE_BLOB_ARRAY.length);
+                inactiveLargeBlobStore, (short) 0, (short) CannedCBOR.INITIAL_LARGE_BLOB_ARRAY.length);
         JCSystem.beginTransaction();
         boolean ok = false;
         try {
@@ -5391,10 +5411,10 @@ private void authenticatorReset(APDU apdu) {
                 JCSystem.abortTransaction();
             }
         }
-        Util.arrayFillNonAtomic(largeBlobStores[realBlobStoreIndex], (short) 0,
-                (short) largeBlobStores[realBlobStoreIndex].length, (byte) 0x00);
+        Util.arrayFillNonAtomic(activeLargeBlobStore, (short) 0,
+                (short) activeLargeBlobStore.length, (byte) 0x00);
         Util.arrayCopyNonAtomic(CannedCBOR.INITIAL_LARGE_BLOB_ARRAY, (short) 0,
-                largeBlobStores[realBlobStoreIndex], (short) 0, (short) CannedCBOR.INITIAL_LARGE_BLOB_ARRAY.length);
+                activeLargeBlobStore, (short) 0, (short) CannedCBOR.INITIAL_LARGE_BLOB_ARRAY.length);
 
         JCSystem.beginTransaction();
         ok = false;
@@ -5594,7 +5614,7 @@ private void sendAuthInfo(APDU apdu) {
 
         buffer[offset++] = 0x0B; // map key: maxSerializedLargeBlobArray: 1 byte = 5
         buffer[offset++] = 0x19; // two-byte integer: 1 byte = 6
-        offset = Util.setShort(buffer, offset, (short) largeBlobStores[largeBlobStoreIndex].length); // 2 bytes = 8
+        offset = Util.setShort(buffer, offset, (short) getCurrentLargeBlobStore().length); // 2 bytes = 8
 
         buffer[offset++] = 0x0C; // map key: forcePinChange: 1 byte = 9
         buffer[offset++] = (byte)(forcePinChange ? 0xF5 : 0xF4); // 1 byte = 10
@@ -6818,11 +6838,12 @@ private FIDO2Applet(byte[] array, short offset, byte length) {
         highSecurityWrappingIV = new byte[IV_LEN];
         lowSecurityWrappingIV = new byte[IV_LEN];
         externalCredentialIV = new byte[IV_LEN];
-        largeBlobStores = new byte[2][largeBlobStoreSize];
+        largeBlobStoreA = new byte[largeBlobStoreSize];
         Util.arrayCopyNonAtomic(CannedCBOR.INITIAL_LARGE_BLOB_ARRAY, (short) 0,
-                largeBlobStores[0], (short) 0, (short) CannedCBOR.INITIAL_LARGE_BLOB_ARRAY.length);
+                largeBlobStoreA, (short) 0, (short) CannedCBOR.INITIAL_LARGE_BLOB_ARRAY.length);
+        largeBlobStoreB = new byte[largeBlobStoreSize];
         Util.arrayCopyNonAtomic(CannedCBOR.INITIAL_LARGE_BLOB_ARRAY, (short) 0,
-                largeBlobStores[1], (short) 0, (short) CannedCBOR.INITIAL_LARGE_BLOB_ARRAY.length);
+                largeBlobStoreB, (short) 0, (short) CannedCBOR.INITIAL_LARGE_BLOB_ARRAY.length);
         largeBlobStoreFill = (short) CannedCBOR.INITIAL_LARGE_BLOB_ARRAY.length;
         highSecurityWrappingKey = getTransientAESKey(); // Our most important treasure, from which all other crypto is born...
         lowSecurityWrappingKey = getPersistentAESKey(); // Not really a treasure