Coverage for admin/misc.py: 62%

1# The MIT License (MIT) 

2# 

3# Copyright (c) 2021 RSK Labs Ltd 

4# 

5# Permission is hereby granted, free of charge, to any person obtaining a copy of 

6# this software and associated documentation files (the "Software"), to deal in 

7# the Software without restriction, including without limitation the rights to 

8# use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies 

9# of the Software, and to permit persons to whom the Software is furnished to do 

10# so, subject to the following conditions: 

11# 

12# The above copyright notice and this permission notice shall be included in all 

13# copies or substantial portions of the Software. 

14# 

15# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 

16# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 

17# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE 

18# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER 

19# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, 

20# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE 

21# SOFTWARE. 

22 

23import sys 

24import time 

25from getpass import getpass 

26from ledger.hsm2dongle import HSM2Dongle 

27from sgx.hsm2dongle import HSM2DongleSGX 

28from ledger.pin import BasePin 

29from .dongle_admin import DongleAdmin 

30from .dongle_eth import DongleEth 

31from comm.platform import Platform 

32from .utils import is_hex_string_of_length, normalize_hex_string 

33from .rsk_client import RskClient, RskClientError 

34 

35 

36PIN_ERROR_MESSAGE = ("Invalid pin given. It must be exactly 8 alphanumeric " 

37 "characters with at least one alphabetic character.") 

38PIN_ERROR_MESSAGE_ANYCHARS = ( 

39 "Invalid pin given. It must be composed only of alphanumeric characters.") 

40 

41SIGNER_WAIT_TIME = 3 # seconds 

42 

43ATTESTATION_UD_VALUE_LENGTH = 32 # bytes 

44DEFAULT_ATT_UD_SOURCE = "https://public-node.rsk.co" 

45 

46 

47class AdminError(RuntimeError): 

48 pass 

49 

50 

51def info(s, nl=True): 

52 newline = "\n" if nl else "" 

53 sys.stdout.write(f"{s}{newline}") 

54 sys.stdout.flush() 

55 

56 

57def head(ss, fill="*", nl=True): 

58 if type(ss) == str: 

59 ss = [ss] 

60 

61 maxl = max(map(len, ss)) 

62 info(fill*maxl) 

63 for s in ss: 

64 info(s) 

65 info(fill*maxl, nl=nl) 

66 

67 

68def bls(b): 

69 return "Yes" if b else "No" 

70 

71 

72def not_implemented(options): 

73 info(f"Operation {options.operation} not yet implemented") 

74 return 1 

75 

76 

77def get_hsm(debug): 

78 info("Connecting to HSM... ", False) 

79 if Platform.is_ledger(): 

80 hsm = HSM2Dongle(debug) 

81 elif Platform.is_sgx(): 

82 hsm = get_sgx_hsm(Platform.options("sgx_host"), 

83 Platform.options("sgx_port"), debug) 

84 else: 

85 raise AdminError("Platform not set or unknown platform") 

86 hsm.connect() 

87 info("OK") 

88 return hsm 

89 

90 

91def get_sgx_hsm(host, port, debug): 

92 hsm = HSM2DongleSGX(host, port, debug) 

93 hsm.connect() 

94 info("OK") 

95 return hsm 

96 

97 

98def get_admin_hsm(debug): 

99 info("Connecting to HSM... ", False) 

100 hsm = DongleAdmin(debug) 

101 hsm.connect() 

102 info("OK") 

103 return hsm 

104 

105 

106def dispose_hsm(hsm): 

107 if hsm is None: 

108 return 

109 

110 info("Disconnecting from HSM... ", False) 

111 hsm.disconnect() 

112 info("OK") 

113 

114 

115def get_eth_dongle(debug): 

116 info("Connecting to Ethereum App... ", False) 

117 eth = DongleEth(debug) 

118 eth.connect() 

119 info("OK") 

120 return eth 

121 

122 

123def dispose_eth_dongle(eth): 

124 if eth is None: 

125 return 

126 

127 info("Disconnecting from Ethereum App... ", False) 

128 eth.disconnect() 

129 info("OK") 

130 

131 

132def ask_for_pin(any_pin): 

133 pin = None 

134 while pin is None or not BasePin.is_valid(pin, any_pin): 

135 pin = getpass("> ").encode() 

136 if not BasePin.is_valid(pin, any_pin): 

137 info(PIN_ERROR_MESSAGE if not any_pin else PIN_ERROR_MESSAGE_ANYCHARS) 

138 return pin 

139 

140 

141def wait_for_reconnection(): 

142 # This is only ever needed for Ledger 

143 if Platform.is_ledger(): 

144 time.sleep(SIGNER_WAIT_TIME) 

145 

146 

147def get_ud_value_for_attestation(user_provided_ud_source): 

148 if is_hex_string_of_length(user_provided_ud_source, 

149 ATTESTATION_UD_VALUE_LENGTH, 

150 allow_prefix=True): 

151 # Final value provided by user 

152 ud_value = normalize_hex_string(user_provided_ud_source) 

153 else: 

154 # Final value taken from a specific Rootstock node 

155 try: 

156 rsk_client = RskClient(user_provided_ud_source) 

157 best_block = rsk_client.get_block_by_number( 

158 rsk_client.get_best_block_number()) 

159 ud_value = best_block["hash"][2:] 

160 if not is_hex_string_of_length(ud_value, ATTESTATION_UD_VALUE_LENGTH): 

161 raise ValueError("Got invalid best block from " 

162 f"Rootstock server: {ud_value}") 

163 except RskClientError as e: 

164 raise AdminError(f"While fetching the best Rootstock block hash: {str(e)}") 

165 

166 return ud_value