Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from socket import AF_INET as IP4_FAMILY
from typing import Optional, List
from neo.Core.TX.Transaction import Transaction as OrigTransaction
from neo.Network.common import msgrouter, wait_for
from neo.Network.common.singleton import Singleton
from neo.Network import utils as networkutils
from neo.Network.mempool import MemPool
from neo.Network.node import NeoNode
from neo.Network.protocol import NeoProtocol
from neo.Network.relaycache import RelayCache
from neo.Network.requestinfo import RequestInfo
from neo.Settings import settings
from neo.logging import log_manager
logger = log_manager.getLogger('network')
class NodeManager(Singleton):
PEER_QUERY_INTERVAL = 15
NODE_POOL_CHECK_INTERVAL = 10 # 2.5 * PEER_QUERY_INTERVAL # this allows for enough time to get new addresses
ONE_MINUTE = 60
MAX_ERROR_COUNT = 5 # maximum number of times adding a block or header may fail before we disconnect it
MAX_TIMEOUT_COUNT = 15 # maximum count the node responds slower than our threshold
MAX_NODE_POOL_ERROR = 2
MAX_NODE_POOL_ERROR_COUNT = 0
# we override init instead of __init__ due to the Singleton (read class documentation)
def init(self):
import binascii
from neo.Core.IO.Mixins import SerializableMixin
from neo.Settings import settings
from neo.Core.Helper import Helper
from neo.Core.Cryptography.Helper import bin_dbl_sha256
from neo.Core.Size import Size as s
from neo.logging import log_manager
logger = log_manager.getLogger()
class ChecksumException(Exception):
pass
class Message(SerializableMixin):
PayloadMaxSize = b'\x02000000'
PayloadMaxSizeInt = int.from_bytes(PayloadMaxSize, 'big')
Magic = None
Command = None
Checksum = None
import json
import binascii
from neo.Core.TX.Transaction import ContractTransaction
from neo.SmartContract.Contract import Contract, ContractType
from neo.SmartContract.ContractParameterType import ContractParameterType, ToName
from neo.VM.ScriptBuilder import ScriptBuilder
from neo.IO.MemoryStream import StreamManager, MemoryStream
from neo.Core.IO.BinaryReader import BinaryReader
from neo.Core.IO.BinaryWriter import BinaryWriter
from neo.VM import OpCode
from neo.Core.Witness import Witness
from neo.logging import log_manager
from neo.Core.Cryptography.ECCurve import ECDSA
from neo.Blockchain import GetBlockchain
logger = log_manager.getLogger('vm')
class ContractParamater:
def __init__(self, type):
if isinstance(type, ContractParameterType):
self.Type = type
elif isinstance(type, int):
self.Type = ContractParameterType(type)
else:
raise Exception("Invalid Contract Parameter Type %s. Must be ContractParameterType or int" % type)
self.Value = None
def ToJson(self):
jsn = {}
jsn['type'] = self.Type.name
return jsn
"""
These are the user-preferences. For the network and system preferences, take a
look at `Settings.py`. Use it like this example:
from neo.UserPreferences import preferences
print(preferences.token_style)
preferences.set_theme("light")
"""
import json
from json.decoder import JSONDecodeError
from neo.Settings import FILENAME_PREFERENCES
from neo.logging import log_manager
logger = log_manager.getLogger()
PREFERENCES_DEFAULT = {
"theme": "dark",
"themes": {
"dark": {
"Command": "#ff0066",
"Default": "#00ee00",
"Neo": "#0000ee",
"Number": "#ffffff"
},
"light": {
"Command": "#ff0066",
"Default": "#008800",
"Neo": "#0000ee",
"Number": "#000000"
}
from neo.Prompt.Commands.LoadSmartContract import LoadContract, GatherContractDetails
from neo.Prompt import Utils as PromptUtils
from neo.Prompt.Commands.BuildNRun import Build, BuildAndRun, LoadAndRun
from neo.Prompt.Commands.Invoke import TestInvokeContract, InvokeContract, test_invoke
from neo.Core.Blockchain import Blockchain
from neocore.UInt160 import UInt160
from neo.SmartContract.ContractParameter import ContractParameter
from neo.Network.neonetwork.common import blocking_prompt as prompt
from neocore.Fixed8 import Fixed8
from neo.Implementations.Blockchains.LevelDB.DebugStorage import DebugStorage
from distutils import util
from neo.Settings import settings
from neo.Prompt.PromptPrinter import prompt_print as print
from neo.logging import log_manager
logger = log_manager.getLogger()
class CommandSC(CommandBase):
def __init__(self):
super().__init__()
self.register_sub_command(CommandSCBuild())
self.register_sub_command(CommandSCBuildRun())
self.register_sub_command(CommandSCLoadRun())
self.register_sub_command(CommandSCTestInvoke())
self.register_sub_command(CommandSCDeploy())
self.register_sub_command(CommandSCDebugStorage())
def command_desc(self):
return CommandDesc('sc', 'develop smart contracts')
from neo.Core.Blockchain import Blockchain
from neo.Wallets.Coin import CoinState
from neo.Core.TX.TransactionAttribute import TransactionAttribute, TransactionAttributeUsage
from neo.SmartContract.ContractParameter import ContractParameterType
from neo.Core.Cryptography.ECCurve import ECDSA
from decimal import Decimal
from neo.logging import log_manager
from neo.Wallets import NEP5Token
from neo.Core.Cryptography.Crypto import Crypto
from typing import TYPE_CHECKING
from neo.Network.common import blocking_prompt as prompt
if TYPE_CHECKING:
from neo.Wallets.Wallet import Wallet
logger = log_manager.getLogger()
def get_asset_attachments(params):
to_remove = []
neo_to_attach = None
gas_to_attach = None
for item in params:
if type(item) is str:
if '--attach-neo=' in item:
to_remove.append(item)
try:
neo_to_attach = Fixed8.TryParse(int(item.replace('--attach-neo=', '')))
except Exception as e:
pass
if '--attach-gas=' in item:
import plyvel
import threading
from contextlib import contextmanager
from neo.Storage.Implementation.AbstractDBImplementation import (
AbstractDBImplementation
)
from neo.Storage.Common.DBPrefix import DBPrefix
from neo.Storage.Interface.DBProperties import DBProperties
import neo.Storage.Implementation.LevelDB.LevelDBSnapshot
from neo.logging import log_manager
logger = log_manager.getLogger()
"""
Description:
Backend implementation for the LevelDB database.
It overrides all methods from the `AbstractDBImplementation` class.
The database factory (`DBFactory`) uses these methods to dynamically
generate a conforming database instance for internal usage.
Usage:
For a new database implementation all methods defined in
AbstractDBImplementation have to be implemented.
"""
class LevelDBImpl(AbstractDBImplementation):
from neo.Core.TX.ClaimTransaction import ClaimTransaction
from neocore.Cryptography.Helper import scripthash_to_address
from neocore.Cryptography.Crypto import Crypto
from neo.Wallets.AddressState import AddressState
from neo.Wallets.Coin import Coin
from neocore.KeyPair import KeyPair
from neo.Wallets import NEP5Token
from neo.Settings import settings
from neocore.Fixed8 import Fixed8
from neocore.UInt160 import UInt160
from neocore.UInt256 import UInt256
from neo.Core.Helper import Helper
from neo.Wallets.utils import to_aes_key
from neo.logging import log_manager
logger = log_manager.getLogger()
class Wallet:
AddressVersion = None
_path = ''
_iv = None
_master_key = None
_keys = {} # holds keypairs
_contracts = {} # holds Contracts
_tokens = {} # holds references to NEP5 tokens
_watch_only = [] # holds set of hashes
_coins = {} # holds Coin References
_current_height = 0
import importlib
from .MemoryStream import StreamManager
from neo.Core.IO.BinaryReader import BinaryReader
from neo.Core.TX.Transaction import Transaction
from neo.logging import log_manager
logger = log_manager.getLogger()
class Helper:
@staticmethod
def AsSerializableWithType(buffer, class_name):
"""
Args:
buffer (BytesIO/bytes): stream to deserialize `class_name` to.
class_name (str): a full path to the class to be deserialized into. e.g. 'neo.Core.Block.Block'
Returns:
object: if deserialization is successful.
None: if deserialization failed.
"""
import binascii
from neo.Core.UInt256 import UInt256
from neo.Core.IO.Mixins import SerializableMixin
from neo.Core.Size import Size as s
from neo.Core.Size import GetVarSize
from neo.logging import log_manager
logger = log_manager.getLogger()
class InvPayload(SerializableMixin):
Type = None
Hashes = []
def __init__(self, type=None, hashes=None):
"""
Create an instance.
Args:
type (neo.Network.InventoryType):
hashes (list): of bytearray items.
"""
self.Type = type
self.Hashes = hashes if hashes else []