1
0
mirror of https://github.com/trezor/trezor-firmware.git synced 2024-11-13 19:18:56 +00:00

cleanup transports

This commit is contained in:
Pavol Rusnak 2016-06-30 16:23:07 +02:00
parent c972d396ae
commit 3b4f226d48
No known key found for this signature in database
GPG Key ID: 91F3B339B9A02A3D
5 changed files with 4 additions and 192 deletions

View File

@ -35,7 +35,7 @@ ether_units = {
def parse_args(commands):
parser = argparse.ArgumentParser(description='Commandline tool for TREZOR devices.')
parser.add_argument('-v', '--verbose', dest='verbose', action='store_true', help='Prints communication to device')
parser.add_argument('-t', '--transport', dest='transport', choices=['usb', 'udp', 'serial', 'pipe', 'socket', 'bridge'], default='usb', help="Transport used for talking with the device")
parser.add_argument('-t', '--transport', dest='transport', choices=['usb', 'udp', 'pipe', 'bridge'], default='usb', help="Transport used for talking with the device")
parser.add_argument('-p', '--path', dest='path', default='', help="Path used by the transport (usually serial port)")
# parser.add_argument('-dt', '--debuglink-transport', dest='debuglink_transport', choices=['usb', 'serial', 'pipe', 'socket'], default='usb', help="Debuglink transport")
# parser.add_argument('-dp', '--debuglink-path', dest='debuglink_path', default='', help="Path used by the transport (usually serial port)")
@ -82,25 +82,13 @@ def get_transport(transport_string, path, **kwargs):
from trezorlib.transport_udp import UdpTransport
return UdpTransport(path, **kwargs)
if transport_string == 'serial':
from trezorlib.transport_serial import SerialTransport
return SerialTransport(path, **kwargs)
if transport_string == 'pipe':
from trezorlib.transport_pipe import PipeTransport
return PipeTransport(path, is_device=False, **kwargs)
if transport_string == 'socket':
from trezorlib.transport_socket import SocketTransportClient
return SocketTransportClient(path, **kwargs)
if transport_string == 'bridge':
from trezorlib.transport_bridge import BridgeTransport
return BridgeTransport(path, **kwargs)
if transport_string == 'fake':
from trezorlib.transport_fake import FakeTransport
return FakeTransport(path, **kwargs)
return BridgeTransport({'path': path}, **kwargs)
raise NotImplementedError("Unknown transport")

View File

@ -2,6 +2,7 @@
import json
import requests
import binascii
from . import protobuf_json
from . import messages_pb2 as proto
from .transport import TransportV1
@ -16,7 +17,7 @@ class BridgeTransport(TransportV1):
def __init__(self, device, *args, **kwargs):
self.configure()
self.path = device['path']
self.path = binascii.hexlify(device['path'])
self.session = None
self.response = None

View File

@ -1,24 +0,0 @@
'''FakeTransport implements dummy interface for Transport.'''
# Local serial port loopback: socat PTY,link=COM8 PTY,link=COM9
from .transport import Transport, NotImplementedException
class FakeTransport(Transport):
def __init__(self, device, *args, **kwargs):
super(FakeTransport, self).__init__(device, *args, **kwargs)
def _open(self):
pass
def _close(self):
pass
def ready_to_read(self):
return False
def _write(self, msg, protobuf_msg):
pass
def _read(self):
raise NotImplementedException("Not implemented")

View File

@ -1,41 +0,0 @@
from __future__ import print_function
'''SerialTransport implements wire transport over serial port.'''
# Local serial port loopback: socat PTY,link=COM8 PTY,link=COM9
from select import select
import serial
from .transport import Transport
class SerialTransport(Transport):
def __init__(self, device, *args, **kwargs):
self.serial = None
super(SerialTransport, self).__init__(device, *args, **kwargs)
def _open(self):
self.serial = serial.Serial(self.device, 115200, timeout=10, writeTimeout=10)
def _close(self):
self.serial.close()
self.serial = None
def ready_to_read(self):
rlist, _, _ = select([self.serial], [], [], 0)
return len(rlist) > 0
def _write(self, msg, protobuf_msg):
try:
self.serial.write(msg)
self.serial.flush()
except serial.SerialException:
print("Error while writing to socket")
raise
def _read(self):
try:
(msg_type, datalen) = self._read_headers(self.serial)
return (msg_type, self.serial.read(datalen))
except serial.SerialException:
print("Failed to read from device")
raise

View File

@ -1,112 +0,0 @@
from __future__ import print_function
'''SocketTransport implements TCP socket interface for Transport.'''
import socket
from select import select
from .transport import Transport
class SocketTransportClient(Transport):
def __init__(self, device, *args, **kwargs):
device = device.split(':')
if len(device) < 2:
device = ('0.0.0.0', int(device[0]))
else:
device = (device[0], int(device[1]))
self.socket = None
super(SocketTransportClient, self).__init__(device, *args, **kwargs)
def _open(self):
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.socket.connect(self.device)
self.filelike = self.socket.makefile()
def _close(self):
self.socket.close()
self.socket = None
self.filelike = None
def ready_to_read(self):
rlist, _, _ = select([self.socket], [], [], 0)
return len(rlist) > 0
def _write(self, msg, protobuf_msg):
self.socket.sendall(msg)
def _read(self):
try:
(msg_type, datalen) = self._read_headers(self.filelike)
return (msg_type, self.filelike.read(datalen))
except socket.error:
print("Failed to read from device")
return None
class SocketTransport(Transport):
def __init__(self, device, *args, **kwargs):
device = device.split(':')
if len(device) < 2:
device = ('0.0.0.0', int(device[0]))
else:
device = (device[0], int(device[1]))
self.socket = None
self.client = None
self.filelike = None
super(SocketTransport, self).__init__(device, *args, **kwargs)
def _open(self):
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
#self.socket.setblocking(0)
self.socket.bind(self.device)
self.socket.listen(5)
def _disconnect_client(self):
print("Disconnecting client")
if self.client != None:
self.client.close()
self.client = None
self.filelike = None
def _close(self):
self._disconnect_client()
self.socket.close()
self.socket = None
def ready_to_read(self):
if self.filelike:
# Connected
rlist, _, _ = select([self.client], [], [], 0)
return len(rlist) > 0
else:
# Waiting for connection
rlist, _, _ = select([self.socket], [], [], 0)
if len(rlist) > 0:
(self.client, ipaddr) = self.socket.accept()
print("Connected", ipaddr[0])
self.filelike = self.client.makefile()
return self.ready_to_read()
return False
def _write(self, msg, protobuf_msg):
if self.filelike:
# None on disconnected client
try:
self.filelike.write(msg)
self.filelike.flush()
except socket.error:
print("Socket error")
self._disconnect_client()
def _read(self):
try:
(msg_type, datalen) = self._read_headers(self.filelike)
return (msg_type, self.filelike.read(datalen))
except Exception:
print("Failed to read from device")
self._disconnect_client()
return None