Funktionierender Prototyp des Serious Games zur Vermittlung von Wissen zu Software-Engineering-Arbeitsmodellen.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

_util.py 5.8KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175
  1. ###############################################################################
  2. #
  3. # The MIT License (MIT)
  4. #
  5. # Copyright (c) typedef int GmbH
  6. #
  7. # Permission is hereby granted, free of charge, to any person obtaining a copy
  8. # of this software and associated documentation files (the "Software"), to deal
  9. # in the Software without restriction, including without limitation the rights
  10. # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  11. # copies of the Software, and to permit persons to whom the Software is
  12. # furnished to do so, subject to the following conditions:
  13. #
  14. # The above copyright notice and this permission notice shall be included in
  15. # all copies or substantial portions of the Software.
  16. #
  17. # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  18. # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  19. # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  20. # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  21. # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  22. # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  23. # THE SOFTWARE.
  24. #
  25. ###############################################################################
  26. import struct
  27. from binascii import a2b_hex, b2a_hex
  28. from typing import Union, Dict, List
  29. import web3
  30. def make_w3(gateway_config=None):
  31. """
  32. Create a Web3 instance configured and ready-to-use gateway to the blockchain.
  33. :param gateway_config: Blockchain gateway configuration.
  34. :type gateway_config: dict
  35. :return: Configured Web3 instance.
  36. :rtype: :class:`web3.Web3`
  37. """
  38. if gateway_config is None or gateway_config['type'] == 'auto':
  39. w3 = web3.Web3()
  40. elif gateway_config['type'] == 'user':
  41. request_kwargs = gateway_config.get('http_options', {})
  42. w3 = web3.Web3(web3.Web3.HTTPProvider(gateway_config['http'], request_kwargs=request_kwargs))
  43. elif gateway_config['type'] == 'infura':
  44. request_kwargs = gateway_config.get('http_options', {})
  45. project_id = gateway_config['key']
  46. # project_secret = gateway_config['secret']
  47. http_url = 'https://{}.infura.io/v3/{}'.format(gateway_config['network'], project_id)
  48. w3 = web3.Web3(web3.Web3.HTTPProvider(http_url, request_kwargs=request_kwargs))
  49. # https://web3py.readthedocs.io/en/stable/middleware.html#geth-style-proof-of-authority
  50. if gateway_config.get('network', None) == 'rinkeby':
  51. # This middleware is required to connect to geth --dev or the Rinkeby public network.
  52. from web3.middleware import geth_poa_middleware
  53. # inject the poa compatibility middleware to the innermost layer
  54. w3.middleware_onion.inject(geth_poa_middleware, layer=0)
  55. # FIXME
  56. elif gateway_config['type'] == 'cloudflare':
  57. # https://developers.cloudflare.com/web3/ethereum-gateway/reference/supported-networks/
  58. raise NotImplementedError()
  59. # FIXME
  60. elif gateway_config['type'] == 'zksync':
  61. # https://v2-docs.zksync.io/dev/testnet/important-links.html
  62. raise NotImplementedError()
  63. else:
  64. raise RuntimeError('invalid blockchain gateway type "{}"'.format(gateway_config['type']))
  65. return w3
  66. def unpack_uint128(data):
  67. assert data is None or type(data) == bytes, 'data must by bytes, was {}'.format(type(data))
  68. if data and type(data) == bytes:
  69. assert len(data) == 16, 'data must be bytes[16], but was bytes[{}]'.format(len(data))
  70. if data:
  71. return web3.Web3.toInt(data)
  72. else:
  73. return 0
  74. def pack_uint128(value):
  75. assert value is None or (type(value) == int and value >= 0 and value < 2**128)
  76. if value:
  77. data = web3.Web3.toBytes(value)
  78. return b'\x00' * (16 - len(data)) + data
  79. else:
  80. return b'\x00' * 16
  81. def unpack_uint256(data):
  82. assert data is None or type(data) == bytes, 'data must by bytes, was {}'.format(type(data))
  83. if data and type(data) == bytes:
  84. assert len(data) == 32, 'data must be bytes[32], but was bytes[{}]'.format(len(data))
  85. if data:
  86. return int(web3.Web3.toInt(data))
  87. else:
  88. return 0
  89. def pack_uint256(value):
  90. assert value is None or (type(value) == int and value >= 0 and value < 2**256), 'value must be uint256, but was {}'.format(value)
  91. if value:
  92. data = web3.Web3.toBytes(value)
  93. return b'\x00' * (32 - len(data)) + data
  94. else:
  95. return b'\x00' * 32
  96. def pack_ethadr(value: Union[bytes, str], return_dict: bool = False) -> Union[List[int], Dict[str, int]]:
  97. """
  98. :param value:
  99. :param return_dict:
  100. :return:
  101. """
  102. if type(value) == str:
  103. if value.startswith('0x'):
  104. value_bytes = a2b_hex(value[2:])
  105. else:
  106. value_bytes = a2b_hex(value)
  107. elif type(value) == bytes:
  108. value_bytes = value
  109. else:
  110. assert False, 'invalid type {} for value'.format(type(value))
  111. assert len(value_bytes) == 20
  112. w = []
  113. for i in range(5):
  114. w.append(struct.unpack('<I', value_bytes[0 + i * 4:4 + i * 4])[0])
  115. if return_dict:
  116. packed_value = {'w0': w[0], 'w1': w[1], 'w2': w[2], 'w3': w[3], 'w4': w[4]}
  117. else:
  118. packed_value = w
  119. return packed_value
  120. def unpack_ethadr(packed_value: Union[List[int], Dict[str, int]], return_str=False) -> Union[bytes, str]:
  121. """
  122. :param packed_value:
  123. :param return_str:
  124. :return:
  125. """
  126. w = []
  127. if type(packed_value) == dict:
  128. for i in range(5):
  129. w.append(struct.pack('<I', packed_value['w{}'.format(i)]))
  130. elif type(packed_value) == list:
  131. for i in range(5):
  132. w.append(struct.pack('<I', packed_value[i]))
  133. else:
  134. assert False, 'should not arrive here'
  135. if return_str:
  136. return web3.Web3.toChecksumAddress('0x' + b2a_hex(b''.join(w)).decode())
  137. else:
  138. return b''.join(w)