forked from keepkey/python-keepkey
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtransport.py
More file actions
156 lines (125 loc) · 4.78 KB
/
Copy pathtransport.py
File metadata and controls
156 lines (125 loc) · 4.78 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
import struct
from . import mapping
class NotImplementedException(Exception):
pass
class ConnectionError(Exception):
pass
class Transport(object):
def __init__(self, device, *args, **kwargs):
self.device = device
self.session_depth = 0
self._open()
def _open(self):
raise NotImplementedException("Not implemented")
def _close(self):
raise NotImplementedException("Not implemented")
def _write(self, msg, protobuf_msg):
raise NotImplementedException("Not implemented")
def _bridgeWrite(self, msg, protobuf_msg):
raise NotImplementedException("Not implemented")
def _read(self):
raise NotImplementedException("Not implemented")
def _bridgeRead(self):
raise NotImplementedException("Not implemented")
def _session_begin(self):
pass
def _session_end(self):
pass
def ready_to_read(self):
"""
Returns True if there is data to be read from the transport. Otherwise, False.
"""
raise NotImplementedException("Not implemented")
def session_begin(self):
"""
Apply a lock to the device in order to preform synchronous multistep "conversations" with the device. For example, before entering the transaction signing workflow, one begins a session. After the transaction is complete, the session may be ended.
"""
if self.session_depth == 0:
self._session_begin()
self.session_depth += 1
def session_end(self):
"""
End a session. Se session_begin for an in depth description of TREZOR sessions.
"""
self.session_depth -= 1
self.session_depth = max(0, self.session_depth)
if self.session_depth == 0:
self._session_end()
def close(self):
"""
Close the connection to the physical device or file descriptor represented by the Transport.
"""
self._close()
def write(self, msg):
"""
Write mesage to tansport. msg should be a member of a valid `protobuf class <https://developers.google.com/protocol-buffers/docs/pythontutorial>`_ with a SerializeToString() method.
"""
ser = msg.SerializeToString()
header = struct.pack(">HL", mapping.get_type(msg), len(ser))
self._write(b"##" + header + ser, msg)
def bridgeWrite(self, msg):
"""
Write message to transport. msg should be a member of a valid `protobuf class <https://developers.google.com/protocol-buffers/docs/pythontutorial>`_ with a SerializeToString() method.
"""
self._bridgeWrite(msg)
def read(self):
"""
If there is data available to be read from the transport, reads the data and tries to parse it as a protobuf message. If the parsing succeeds, return a protobuf object.
Otherwise, returns None.
"""
if not self.ready_to_read():
return None
data = self._read()
if data is None:
return None
return self._parse_message(data)
def read_blocking(self):
"""
Same as read, except blocks untill data is available to be read.
"""
while True:
data = self._read()
if data != None:
break
return self._parse_message(data)
def bridge_read_blocking(self):
"""
blocks until data is available to be read.
"""
while True:
data = self._bridgeRead()
if data != None:
break
return data
def _parse_message(self, data):
(msg_type, data) = data
if msg_type == 'protobuf':
return data
else:
inst = mapping.get_class(msg_type)()
inst.ParseFromString(data)
return inst
def _read_headers(self, read_f):
# Try to read headers until some sane value are detected
is_ok = False
while not is_ok:
# Align cursor to the beginning of the header ("##")
c = read_f.read(1)
i = 0
while c != b"#":
i += 1
if i >= 64:
# timeout
raise Exception("Timed out while waiting for the magic character")
c = read_f.read(1)
if read_f.read(1) != b"#":
# Second character must be # to be valid header
raise Exception("Second magic character is broken")
# Now we're most likely on the beginning of the header
try:
headerlen = struct.calcsize(">HL")
(msg_type, datalen) = struct.unpack(">HL", read_f.read(headerlen))
break
except:
raise Exception("Cannot parse header length")
return (msg_type, datalen)