Fix factory-reset.
[gnuk/gnuk.git] / tests / card_reader.py
1 """
2 card_reader.py - a library for smartcard reader
3
4 Copyright (C) 2016  Free Software Initiative of Japan
5 Author: NIIBE Yutaka <gniibe@fsij.org>
6
7 This file is a part of Gnuk, a GnuPG USB Token implementation.
8
9 Gnuk is free software: you can redistribute it and/or modify it
10 under the terms of the GNU General Public License as published by
11 the Free Software Foundation, either version 3 of the License, or
12 (at your option) any later version.
13
14 Gnuk is distributed in the hope that it will be useful, but WITHOUT
15 ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
16 or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public
17 License for more details.
18
19 You should have received a copy of the GNU General Public License
20 along with this program.  If not, see <http://www.gnu.org/licenses/>.
21 """
22
23 import usb.core
24 from struct import pack
25 from usb.util import find_descriptor, claim_interface, get_string, \
26     endpoint_type, endpoint_direction, \
27     ENDPOINT_TYPE_BULK, ENDPOINT_OUT, ENDPOINT_IN
28
29 # USB class, subclass, protocol
30 CCID_CLASS = 0x0B
31 CCID_SUBCLASS = 0x00
32 CCID_PROTOCOL_0 = 0x00
33
34 def ccid_compose(msg_type, seq, slot=0, rsv=0, param=0, data=b""):
35     return pack('<BiBBBH', msg_type, len(data), slot, seq, rsv, param) + data
36
37 IFSC=254
38
39 def compute_edc(pcb, info):
40     edc = pcb
41     edc ^= len(info)
42     for i in range(len(info)):
43         edc ^= info[i]
44     return edc
45
46 def compose_i_block(ns, info, more):
47     pcb = 0x00
48     if ns:
49         pcb |= 0x40
50     if more:
51         pcb |= 0x20
52     edc = compute_edc(pcb, info)
53     return bytes([0, pcb, len(info)]) + info + bytes([edc])
54
55 def compose_r_block(nr, edc_error=0):
56     pcb = 0x80
57     if nr:
58         pcb |= 0x10
59     if edc_error:
60         pcb |= 0x01
61     return bytes([0, pcb, 0, pcb])
62
63 def is_r_block_no_error_or_other(blk):
64     return (((blk[1] & 0xC0) == 0x80 and (blk[1] & 0x2f) == 0x00)) or \
65         ((blk[1] & 0xC0) != 0x80)
66
67 def is_s_block_time_ext(blk):
68     return (blk[1] == 0xC3)
69
70 def is_i_block_last(blk):
71     return ((blk[1] & 0x80) == 0 and (blk[1] & 0x20) == 0)
72
73 def is_i_block_more(blk):
74     return ((blk[1] & 0x80) == 0 and (blk[1] & 0x20) == 0x20)
75
76 def is_edc_error(blk):
77     # to be implemented
78     return 0
79
80 def i_block_content(blk):
81     return blk[3:-1]
82
83 class CardReader(object):
84     def __init__(self, dev):
85         """
86         __init__(dev) -> None
87         Initialize the DEV of CCID.
88         device: usb.core.Device object.
89         """
90
91         cfg = dev.get_active_configuration()
92         intf = find_descriptor(cfg, bInterfaceClass=CCID_CLASS,
93                                bInterfaceSubClass=CCID_SUBCLASS,
94                                bInterfaceProtocol=CCID_PROTOCOL_0)
95         if intf is None:
96             raise ValueError("Not a CCID device")
97
98         claim_interface(dev, intf)
99
100         for ep in intf:
101             if endpoint_type(ep.bmAttributes) == ENDPOINT_TYPE_BULK and \
102                endpoint_direction(ep.bEndpointAddress) == ENDPOINT_OUT:
103                self.__bulkout = ep.bEndpointAddress
104             if endpoint_type(ep.bmAttributes) == ENDPOINT_TYPE_BULK and \
105                endpoint_direction(ep.bEndpointAddress) == ENDPOINT_IN:
106                self.__bulkin = ep.bEndpointAddress
107
108         assert len(intf.extra_descriptors) == 54
109         assert intf.extra_descriptors[1] == 33
110
111         if (intf.extra_descriptors[42] & 0x02):
112             # Short APDU level exchange
113             self.__use_APDU = True
114         elif (intf.extra_descriptors[42] & 0x04):
115             # Short and extended APDU level exchange
116             self.__use_APDU = True
117         elif (intf.extra_descriptors[42] & 0x01):
118             # TPDU level exchange
119             self.__use_APDU = False
120         else:
121             raise ValueError("Unknown exchange level")
122
123         # Check other bits???
124         #       intf.extra_descriptors[40]
125         #       intf.extra_descriptors[41]
126
127         self.__dev = dev
128         self.__timeout = 10000
129         self.__seq = 0
130
131     def get_string(self, num):
132         return get_string(self.__dev, num)
133
134     def increment_seq(self):
135         self.__seq = (self.__seq + 1) & 0xff
136
137     def reset_device(self):
138         try:
139             self.__dev.reset()
140         except:
141             pass
142
143     def is_tpdu_reader(self):
144         return not self.__use_APDU
145
146     def ccid_get_result(self):
147         msg = self.__dev.read(self.__bulkin, 1024, self.__timeout)
148         if len(msg) < 10:
149             print(msg)
150             raise ValueError("ccid_get_result")
151         msg_type = msg[0]
152         data_len = msg[1] + (msg[2]<<8) + (msg[3]<<16) + (msg[4]<<24)
153         slot = msg[5]
154         seq = msg[6]
155         status = msg[7]
156         error = msg[8]
157         chain = msg[9]
158         data = msg[10:]
159         # XXX: check msg_type, data_len, slot, seq, error
160         return (status, chain, data.tobytes())
161
162     def ccid_get_status(self):
163         msg = ccid_compose(0x65, self.__seq)
164         self.__dev.write(self.__bulkout, msg, self.__timeout)
165         self.increment_seq()
166         status, chain, data = self.ccid_get_result()
167         # XXX: check chain, data
168         return status
169
170     def ccid_power_on(self):
171         msg = ccid_compose(0x62, self.__seq, rsv=1) # Vcc=5V
172         self.__dev.write(self.__bulkout, msg, self.__timeout)
173         self.increment_seq()
174         status, chain, data = self.ccid_get_result()
175         # XXX: check status, chain
176         self.atr = data
177         #
178         if self.__use_APDU == False:
179             # TPDU reader configuration
180             self.ns = 0
181             self.nr = 0
182             # Set PPS
183             pps = b"\xFF\x11\x18\xF6"
184             status, chain, ret_pps = self.ccid_send_data_block(pps)
185             # Set parameters
186             param = b"\x18\x10\xFF\x75\x00\xFE\x00"
187             # ^--- This shoud be adapted by ATR string, see update_param_by_atr
188             msg = ccid_compose(0x61, self.__seq, rsv=0x1, data=param)
189             self.__dev.write(self.__bulkout, msg, self.__timeout)
190             self.increment_seq()
191             status, chain, ret_param = self.ccid_get_result()
192             # Send an S-block of changing IFSD=254
193             sblk = b"\x00\xC1\x01\xFE\x3E"
194             status, chain, ret_sblk = self.ccid_send_data_block(sblk)
195         return self.atr
196
197     def ccid_power_off(self):
198         msg = ccid_compose(0x63, self.__seq)
199         self.__dev.write(self.__bulkout, msg, self.__timeout)
200         self.increment_seq()
201         status, chain, data = self.ccid_get_result()
202         # XXX: check chain, data
203         return status
204
205     def ccid_send_data_block(self, data):
206         msg = ccid_compose(0x6f, self.__seq, data=data)
207         self.__dev.write(self.__bulkout, msg, self.__timeout)
208         self.increment_seq()
209         return self.ccid_get_result()
210
211     def ccid_send_cmd(self, data):
212         status, chain, data_rcv = self.ccid_send_data_block(data)
213         if chain == 0:
214             while status == 0x80:
215                 status, chain, data_rcv = self.ccid_get_result()
216             return data_rcv
217         elif chain == 1:
218             d = data_rcv
219             while True:
220                 msg = ccid_compose(0x6f, self.__seq, param=0x10)
221                 self.__dev.write(self.__bulkout, msg, self.__timeout)
222                 self.increment_seq()
223                 status, chain, data_rcv = self.ccid_get_result()
224                 # XXX: check status
225                 d += data_rcv
226                 if chain == 2:
227                     break
228                 elif chain == 3:
229                     continue
230                 else:
231                     raise ValueError("ccid_send_cmd chain")
232             return d
233         else:
234             raise ValueError("ccid_send_cmd")
235
236     def send_tpdu(self, info=None, more=0, response_time_ext=0,
237                   edc_error=0, no_error=0):
238         if info:
239             data = compose_i_block(self.ns, info, more)
240         elif response_time_ext:
241             # compose S-block
242             data = b"\x00\xE3\x00\xE3"
243         elif edc_error:
244             data = compose_r_block(self.nr, edc_error=1)
245         elif no_error:
246             data = compose_r_block(self.nr)
247         msg = ccid_compose(0x6f, self.__seq, data=data)
248         self.__dev.write(self.__bulkout, msg, self.__timeout)
249         self.increment_seq()
250
251     def recv_tpdu(self):
252         status, chain, data = self.ccid_get_result()
253         return data
254
255     def send_cmd(self, cmd):
256         # Simple APDU case
257         if self.__use_APDU:
258             return self.ccid_send_cmd(cmd)
259         # TPDU case
260         while len(cmd) > 254:
261             blk = cmd[0:254]
262             cmd = cmd[254:]
263             while True:
264                 self.send_tpdu(info=blk,more=1)
265                 rblk = self.recv_tpdu()
266                 if is_r_block_no_error_or_other(rblk):
267                     break
268             self.ns = self.ns ^ 1
269         while True:
270             self.send_tpdu(info=cmd)
271             blk = self.recv_tpdu()
272             if is_r_block_no_error_or_other(blk):
273                 break
274         self.ns = self.ns ^ 1
275         res = b""
276         while True:
277             if is_s_block_time_ext(blk):
278                 self.send_tpdu(response_time_ext=1)
279             elif is_i_block_last(blk):
280                 self.nr = self.nr ^ 1
281                 if is_edc_error(blk):
282                     self.send_tpdu(edc_error=1)
283                 else:
284                     res += i_block_content(blk)
285                     break
286             elif is_i_block_more(blk):
287                 self.nr = self.nr ^ 1
288                 if is_edc_error(blk):
289                     self.send_tpdu(edc_error=1)
290                 else:
291                     res += i_block_content(blk)
292                     self.send_tpdu(no_error=1)
293             blk = self.recv_tpdu()
294         return res
295
296
297 class find_class(object):
298     def __init__(self, usb_class):
299         self.__class = usb_class
300     def __call__(self, device):
301         if device.bDeviceClass == self.__class:
302             return True
303         for cfg in device:
304             intf = find_descriptor(cfg, bInterfaceClass=self.__class)
305             if intf is not None:
306                 return True
307         return False
308
309 def get_ccid_device():
310     ccid = None
311     dev_list = usb.core.find(find_all=True, custom_match=find_class(CCID_CLASS))
312     for dev in dev_list:
313         try:
314             ccid = CardReader(dev)
315             print("CCID device: Bus %03d Device %03d" % (dev.bus, dev.address))
316             break
317         except:
318             pass
319     if not ccid:
320         raise ValueError("No CCID device present")
321     status = ccid.ccid_get_status()
322     if status == 0:
323         # It's ON already
324         atr = ccid.ccid_power_on()
325     elif status == 1:
326         atr = ccid.ccid_power_on()
327     else:
328         raise ValueError("Unknown CCID status", status)
329     return ccid