Feat: Wallet generation ennabled

Emmanuelonyo · Emmanuelonyo · commit e76050563952 · 2024-10-12T16:35:39.000+01:00
diff --git a/cngn_manager/CryptoWallet.py b/cngn_manager/CryptoWallet.py
@@ -0,0 +1,139 @@
+from ecdsa import SigningKey, SECP256k1
+from mnemonic import Mnemonic
+from bip32utils import BIP32Key, BIP32_HARDEN
+from tronpy import Tron
+from tronpy.keys import PrivateKey, PublicKey
+from hashlib import sha3_256
+from stellar_sdk import Keypair, StrKey
+import coinaddrvalidator
+import nacl.signing
+import nacl.encoding
+from .constants import Network  # Assuming you have a Network class or Enum in constants
+
+class CryptoWallet:
+
+    DERIVATION_PATHS = {
+        Network.ETH: "m/44'/60'/0'/0/0",
+        Network.BSC: "m/44'/60'/0'/0/0",
+        Network.ATC: "m/44'/60'/0'/0/0",
+        Network.MATIC: "m/44'/60'/0'/0/0",
+        Network.TRX: "m/44'/195'/0'/0/0",  # TRON's derivation path
+        Network.XBN: "m/44'/703'/0'/0"     # XBN's derivation path
+    }
+
+    @staticmethod
+    def generate_wallet_with_mnemonic_details(network: str):
+        mnemo = Mnemonic("english")
+        mnemonic = mnemo.generate(strength=128)
+        return CryptoWallet.generate_wallet_from_mnemonic(mnemonic, network)
+
+    @staticmethod
+    def generate_wallet_from_mnemonic(mnemonic: str, network: str):
+        if network == Network.XBN:
+            return CryptoWallet.generate_xbn_wallet(mnemonic)
+        elif network == Network.TRX:
+            return CryptoWallet.generate_trx_wallet(mnemonic)
+
+        private_key = CryptoWallet.get_private_key_from_mnemonic(mnemonic, network)
+        public_key = CryptoWallet.get_public_key(private_key, network)
+        address = CryptoWallet.get_address_from_public_key(public_key, network)
+        
+        return {'mnemonic': mnemonic, 'privateKey': private_key, 'address': address, 'network': network}
+
+    @staticmethod
+    def get_public_key(private_key: str, network: str):
+        if network == Network.XBN:
+            seed = nacl.signing.SigningKey(bytes.fromhex(private_key))
+            return seed.verify_key.encode().hex()
+        sk = SigningKey.from_string(bytes.fromhex(private_key), curve=SECP256k1)
+        return sk.verifying_key.to_string().hex()
+
+    @staticmethod
+    def get_private_key_from_mnemonic(mnemonic: str, network: str):
+        seed = Mnemonic.to_seed(mnemonic)
+        derivation_path = CryptoWallet.DERIVATION_PATHS[network]
+        # Use BIP32Key derivation according to the provided network path
+        key = BIP32Key.fromEntropy(seed)
+        for index in derivation_path.split("/")[1:]:
+            if index.endswith("'"):
+                index = int(index[:-1]) + 0x80000000
+            else:
+                index = int(index)
+            key = key.ChildKey(index)
+        return key.PrivateKey().hex()
+
+    @staticmethod
+    def get_address_from_public_key(public_key: str, network: str):
+        if network in [Network.ETH, Network.BSC, Network.MATIC, Network.ATC]:
+            return CryptoWallet.get_ethereum_style_address(public_key)
+        elif network == Network.TRX:
+            return CryptoWallet.get_tron_address_from_public_key(public_key)
+        raise ValueError(f'Unsupported network: {network}')
+
+    @staticmethod
+    def get_ethereum_style_address(public_key: str):
+        clean_public_key = public_key[2:] if public_key.startswith('04') else public_key
+        hash = sha3_256(bytes.fromhex(clean_public_key)).digest()
+        return '0x' + hash[-20:].hex()
+
+    @staticmethod
+    def generate_trx_wallet(mnemonic: str):
+        private_key = CryptoWallet.get_private_key_from_mnemonic(mnemonic, Network.TRX)
+        tron_private_key = PrivateKey(bytes.fromhex(private_key))
+        address = tron_private_key.public_key.to_base58check_address()
+
+        return {
+            'mnemonic': mnemonic,
+            'privateKey': tron_private_key.hex(),
+            'address': address,
+            'network': Network.TRX
+        }
+
+    @staticmethod
+    def get_tron_address_from_public_key(public_key: str):
+        public_key_bytes = bytes.fromhex(public_key)
+        if len(public_key_bytes) != 64:
+            raise ValueError("Invalid public key length")
+
+        tron_address = PublicKey(public_key_bytes).to_base58check_address()
+        return tron_address
+    
+    @staticmethod
+    def generate_xbn_wallet(mnemonic: str):
+        # Derive BIP32 seed from mnemonic
+        seed = Mnemonic.to_seed(mnemonic)
+        derivation_path = CryptoWallet.DERIVATION_PATHS[Network.XBN]
+        master_key = BIP32Key.fromEntropy(seed)
+        
+        for index in derivation_path.split("/")[1:]:
+            if index.endswith("'"):
+                index = int(index[:-1]) + BIP32_HARDEN # Hardened key
+            else:
+                index = int(index)
+            key = master_key.ChildKey(index)
+
+        raw_secret_key = key.PrivateKey()
+        secret_key = StrKey.encode_ed25519_secret_seed(raw_secret_key)
+        keypair = Keypair.from_secret(secret_key)
+
+        return {
+            'mnemonic': mnemonic,
+            'privateKey': keypair.secret,
+            'address': keypair.public_key,
+            'network': Network.XBN
+        }
+
+    @staticmethod
+    def validate_address(address: str, network: str):
+        if network in [Network.ETH, Network.BSC, Network.MATIC, Network.ATC]:
+            result = coinaddrvalidator.validate(network, address)
+            print(f'Validation result for {network}: {result}')
+            return result
+        elif network == Network.TRX:
+            tron = Tron()
+            return tron.is_address(address)
+        elif network == Network.XBN:
+            result = coinaddrvalidator.validate(network, address)
+            print(f'Validation result for {network}: {result}')
+            return result
+        raise ValueError(f'Unsupported network: {network}')
diff --git a/cngn_manager/__version__.py b/cngn_manager/__version__.py
@@ -10,7 +10,7 @@
 __title__ = "cngn_manager"
 __description__ = "Python HTTP for Humans."
 __url__ = "https://docs.cngn.co"
-__version__ = "0.0.4"
+__version__ = "0.0.5"
 __author__ = "Emmanuel Onyo"
 __author_email__ = "emmanuel.onyo@withconvexity.com"
 __license__ = "MIT License"
diff --git a/cngn_manager/constants.py b/cngn_manager/constants.py
@@ -1,6 +1,11 @@
 class Network:
     BSC = "bsc"
     ETH = "eth"
+    ATC = "atc"
+    MATIC = "matic"
+    TRX = "trx"
+    XBN = "xbn"
+    
     # Add other network options if necessary
 
 class ProviderType:
diff --git a/cngn_manager/main.py b/cngn_manager/main.py
@@ -11,8 +11,9 @@
 from typing import Optional, Dict, Any
 import requests
 from requests.exceptions import RequestException, HTTPError
-from cngn_manager.AESCrypto import AESCrypto
-from cngn_manager.Ed25519Crypto import Ed25519Crypto
+from .AESCrypto import AESCrypto
+from .Ed25519Crypto import Ed25519Crypto
+from .CryptoWallet import CryptoWallet
 
 
 """
@@ -106,3 +107,13 @@ def create_virtual_account(self, data: dict) -> str:
 
     def whitelist_address(self, data: dict) -> str:
         return self.__make_calls("POST", "/api/whiteListAddress", data)
+
+    def generate_wallet_address(self, network: str) -> str:
+        response =  CryptoWallet.generate_wallet_with_mnemonic_details(network)
+        return {
+            "success": True,
+            "data": response
+        }
+
+    def validate_address(self, address, network):
+        return CryptoWallet.validate_address(address, network)
diff --git a/setup.py b/setup.py
@@ -4,15 +4,9 @@
 
 # Dictionary to hold package metadata
 about = {}
-
-# Set the current directory as the base directory
 here = os.path.abspath(os.path.dirname(__file__))
-
-# Read the __version__.py file to get version info
 with open(os.path.join(here, "cngn_manager", "__version__.py"), "r", encoding="utf-8") as f:
     exec(f.read(), about)
-
-# Read the README.md file for long description
 with open(os.path.join(here, "README.md"), "r", encoding="utf-8") as f:
     readme = f.read()
 
@@ -21,6 +15,11 @@
     "requests",
     "cryptography==43.0.1",
     "pynacl==1.5.0",
+    "mnemonic==0.20",
+    "bip32utils==0.3.post4",
+    "hashlib",
+    "tronpy",
+    "stellar-sdk==11.1.0",
 ]
 
 # Define test dependencies
@@ -31,6 +30,7 @@
     "pytest-xdist",
     "PySocks>=1.5.6, !=1.5.7",
     "pytest>=3",
+
 ]
 
 # Setup function to handle packaging
diff --git a/tests/main.py b/tests/main.py
@@ -0,0 +1,131 @@
+import unittest
+from unittest.mock import patch, MagicMock
+from cngn_manager import CNGnManager
+import json
+from cngn_manager.AESCrypto import AESCrypto 
+from cngn_manager.Ed25519Crypto import Ed25519Crypto
+
+from nacl.public import PrivateKey, PublicKey, Box
+from requests.exceptions import RequestException, HTTPError
+
+class TestCNGnManager(unittest.TestCase):
+
+    def setUp(self):
+        # Initialize CNGnManager with test data
+        self.api_key = "test_api_key"
+        self.private_key = "test_private_key"
+        self.encryption_key = "test_encryption_key"
+        self.manager = CNGnManager(self.api_key, self.private_key, self.encryption_key)
+
+    @patch('requests.Session')
+    def test_init(self, mock_session):
+        # Test the initialization of CNGnManager
+        manager = CNGnManager(self.api_key, self.private_key, self.encryption_key)
+        self.assertEqual(manager.api_key, self.api_key)
+        self.assertEqual(manager.api_url, CNGnManager.API_URL)
+        mock_session.assert_called_once()
+
+    @patch.object(AESCrypto, 'encrypt', return_value="encrypted_data")
+    @patch.object(Ed25519Crypto, 'decrypt_with_private_key', return_value='{"decrypted":"data"}')
+    @patch('requests.Session.request')
+    def test_make_calls_success(self, mock_request, mock_decrypt, mock_encrypt):
+        # Mock the response from the API
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"data": "encrypted_response_data"}
+        mock_request.return_value = mock_response
+
+        # Test a successful GET call
+        result = self.manager.get_balance()
+        
+        # Verify that encryption, request, and decryption were called
+        mock_encrypt.assert_called_once_with(json.dumps(None), self.encryption_key)
+        mock_request.assert_called_once_with('GET', f'{self.manager.api_url}/v1/api/balance', json="encrypted_data")
+        mock_decrypt.assert_called_once_with(self.private_key, "encrypted_response_data")
+
+        # Verify the response is properly decrypted
+        self.assertEqual(result, {"data": {"decrypted": "data"}})
+
+    @patch.object(AESCrypto, 'encrypt', return_value="encrypted_data")
+    @patch.object(Ed25519Crypto, 'decrypt_with_private_key')
+    @patch('requests.Session.request', side_effect=RequestException("API request failed"))
+    def test_make_calls_request_error(self, mock_request, mock_decrypt, mock_encrypt):
+        # Test handling of request exceptions
+        result = self.manager.get_balance()
+        
+        # Verify that the request error is handled properly
+        self.assertEqual(result['success'], False)
+        self.assertEqual(result['error'], 'API request failed')
+        self.assertEqual(result['message'], 'API request failed')
+
+    @patch.object(AESCrypto, 'encrypt', return_value="encrypted_data")
+    @patch('requests.Session.request', side_effect=HTTPError(response=MagicMock(status_code=500)))
+    def test_make_calls_http_error(self, mock_request, mock_encrypt):
+        # Test handling of HTTP errors
+        result = self.manager.get_balance()
+        
+        # Verify that HTTPError is handled properly
+        self.assertEqual(result['success'], False)
+        self.assertEqual(result['error'], 'API request failed')
+        self.assertEqual(result['status_code'], 500)
+
+    @patch.object(AESCrypto, 'encrypt', side_effect=Exception("Unexpected error"))
+    def test_make_calls_unexpected_error(self, mock_encrypt):
+        # Test handling of unexpected errors
+        result = self.manager.get_balance()
+        
+        # Verify that unexpected errors are handled properly
+        self.assertEqual(result['success'], False)
+        self.assertEqual(result['error'], 'An unexpected error occurred')
+        self.assertEqual(result['message'], 'Unexpected error')
+        self.assertEqual(result['status_code'], 500)
+
+    @patch.object(AESCrypto, 'encrypt', return_value="encrypted_data")
+    @patch.object(Ed25519Crypto, 'decrypt_with_private_key', return_value='{"decrypted":"data"}')
+    @patch('requests.Session.request')
+    def test_post_calls(self, mock_request, mock_decrypt, mock_encrypt):
+        # Mock the response from the API
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"data": "encrypted_response_data"}
+        mock_request.return_value = mock_response
+
+        # Test a POST call
+        data = {"test": "data"}
+        result = self.manager.swap_between_chains(data)
+
+        # Verify that encryption, request, and decryption were called
+        mock_encrypt.assert_called_once_with(json.dumps(data), self.encryption_key)
+        mock_request.assert_called_once_with('POST', f'{self.manager.api_url}/v1/api/swap', json="encrypted_data")
+        mock_decrypt.assert_called_once_with(self.private_key, "encrypted_response_data")
+
+        # Verify the response is properly decrypted
+        self.assertEqual(result, {"data": {"decrypted": "data"}})
+
+    def test_prepare_request_data(self):
+        aes_crypto = AESCrypto()
+
+        # Test with data
+        data = {"test": "data"}
+        encrypted_data = self.manager._prepare_request_data(data, aes_crypto)
+        self.assertIsNotNone(encrypted_data)
+
+        # Test with no data
+        encrypted_data = self.manager._prepare_request_data(None, aes_crypto)
+        self.assertIsNone(encrypted_data)
+
+    def test_handle_request_error(self):
+        error = RequestException("Request failed")
+        result = self.manager._handle_request_error(error)
+        self.assertEqual(result['success'], False)
+        self.assertEqual(result['error'], 'API request failed')
+        self.assertEqual(result['message'], 'Request failed')
+
+    def test_handle_unexpected_error(self):
+        error = Exception("Something went wrong")
+        result = self.manager._handle_unexpected_error(error)
+        self.assertEqual(result['success'], False)
+        self.assertEqual(result['error'], 'An unexpected error occurred')
+        self.assertEqual(result['message'], 'Something went wrong')
+        self.assertEqual(result['status_code'], 500)
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/main.test.py b/tests/main.test.py
