allocate cids better and do proper timeouts in hid

ctap_test.py

@@ -12,7 +12,7 @@ from fido2.utils import Timeout
 import sys,os
 from random import randint
 from binascii import hexlify
-import array,struct
+import array,struct,socket
 
 # Set up a FIDO 2 client using the origin https://example.com
 
@@ -24,6 +24,19 @@ def ForceU2F(client,device):
     client._do_get_assertion = client._ctap1_get_assertion
 
 
+class Packet(object):
+    def __init__(self,data):
+        self.data = data
+
+    def ToWireFormat(self,):
+        return self.data
+
+    @staticmethod
+    def FromWireFormat(pkt_size,data):
+        return Packet(data)
+
+
+
 class Tester():
     def __init__(self,):
         self.origin = 'https://examplo.org'
@@ -35,15 +48,49 @@ class Tester():
         self.dev = dev
         self.ctap = CTAP2(dev)
 
+        # consume timeout error
+        cmd,resp = self.recv_raw()
 
     def send_data(self, cmd, data):
         #print('<<', hexlify(data))
         if type(data) != type(b''):
-            data = data.encode()
+            data = struct.pack('%dB' % len(data), *[ord(x) for x in data])
         with Timeout(1.0) as event:
             return self.dev.call(cmd, data,event)
 
+    def send_raw(self, data, cid = None):
+        if cid is None:
+            cid = self.dev._dev.cid
+        if type(data) != type(b''):
+            data = struct.pack('%dB' % len(data), *[ord(x) for x in data])
+        self.dev._dev.InternalSendPacket(Packet(cid + data))
+
+    def cid(self,):
+        return self.dev._dev.cid
+
+    def recv_raw(self,):
+        cmd,payload = self.dev._dev.InternalRecv()
+        return cmd, payload
+
+    def check_error(self,data,err=None):
+        assert(len(data) == 1)
+        if err is None:
+            if data[0] != 0:
+                raise CtapError(data[0])
+        elif data[0] != err:
+            raise ValueError('Unexpected error: %02x' % data[0])
+
+
     def test_hid(self,):
+        print('Test idle')
+        try:
+            cmd,resp = self.recv_raw()
+        except socket.timeout:
+            print('Pass: Idle')
+
+        print('Test init')
+        r = self.send_data(CTAPHID.INIT, '\x11\x11\x11\x11\x11\x11\x11\x11')
+
         pingdata = os.urandom(100)
         try:
             r = self.send_data(CTAPHID.PING, pingdata)
@@ -62,6 +109,8 @@ class Tester():
             print('7609 byte Ping failed:', e)
         print('PASS: 7609 byte ping')
 
+        print('Test non-active cid')
+
         try:
             r = self.send_data(CTAPHID.WINK, '')
         except CtapError as e:
@@ -90,15 +139,59 @@ class Tester():
         print('PASS: no data msg')
 
         try:
-            r = self.send_data(CTAPHID.INIT, '')
+            r = self.send_data(CTAPHID.INIT, '\x11\x22\x33\x44\x55\x66\x77\x88')
         except CtapError as e:
-            print('resync fail: ', e)
-            return
+            raise RuntimeError('resync fail: ', e)
         print('PASS: resync')
 
+        try:
+            r = self.send_data(0x66, '')
+            raise RuntimeError('Invalid command did not return error')
+        except CtapError as e:
+            assert(e.code == CtapError.ERR.INVALID_COMMAND)
+        print('PASS: invalid HID command')
 
 
+        print('Sending packet with too large of a length.')
+        self.send_raw('\x80\x1d\xba\x00')
+        cmd,resp = self.recv_raw()
+        self.check_error(resp, CtapError.ERR.INVALID_LENGTH)
+        print('PASS: invalid length')
 
+        print('Sending packets that skip a sequence number.')
+        self.send_raw('\x81\x10\x00')
+        self.send_raw('\x00')
+        self.send_raw('\x01')
+        self.send_raw('\x02')
+        # skip 3
+        self.send_raw('\x04')
+        cmd,resp = self.recv_raw()
+        self.check_error(resp, CtapError.ERR.INVALID_SEQ)
+        print('PASS: invalid sequence')
+
+        print('Resync and send ping')
+        try:
+            r = self.send_data(CTAPHID.INIT, '\x11\x22\x33\x44\x55\x66\x77\x88')
+            pingdata = os.urandom(100)
+            r = self.send_data(CTAPHID.PING, pingdata)
+            if (r != pingdata):
+                raise ValueError('Ping data not echo\'d')
+        except CtapError as e:
+            raise RuntimeError('resync fail: ', e)
+        print('PASS: resync and ping')
+
+        print('Send ping and abort it')
+        self.send_raw('\x81\x10\x00')
+        self.send_raw('\x00')
+        self.send_raw('\x01')
+        try:
+            r = self.send_data(CTAPHID.INIT, '\x11\x22\x33\x44\x55\x66\x77\x88')
+        except CtapError as e:
+            raise RuntimeError('resync fail: ', e)
+        print('PASS: interrupt ping with resync')
+
+
+        print('Send ping and abort it')
 
 if __name__ == '__main__':
     t = Tester()