i.niyazov/frostfs-sdk-python
1
0
Fork 0
forked from TrueCloudLab/frostfs-sdk-python
Code Pull requests Activity

[#3] Added generate proto script create container method

Browse source 
Signed-off-by: Ilyas Niyazov <i.niyazov@yadro.com>
This commit is contained in:
Ilyas Niyazov 2025-03-25 11:10:30 +03:00
parent fba6eaaa9c
commit d2e0c80f7c
16 changed files with 348 additions and 51 deletions
Download patch file Download diff file Expand all files Collapse all files

50
frostfs_sdk/cryptography/key_extension.py
View file

@ -1,5 +1,18 @@
import base58
import ecdsa
import hashlib
from struct import pack, unpack
from Crypto.Hash import RIPEMD160
COMPRESSED_PUBLIC_KEY_LENGTH = 33
NEO_ADDRESS_VERSION = 0x35
UNCOMPRESSED_PUBLIC_KEY_LENGTH = 65
DECODE_ADDRESS_LENGTH = 21
PS_IN_HASH160 = 0x0C
CHECK_SIG_DESCRIPTOR = int.from_bytes(
hashlib.sha256("System.Crypto.CheckSig".encode('ascii')).digest()[:4], byteorder='little'
)
class KeyExtension:
@ -37,12 +50,47 @@ class KeyExtension:
compressed_public_key = public_key.to_string("compressed")
return compressed_public_key
def get_owner_id_by_public_key(self, public_key: bytes) -> str:
if len(public_key) != COMPRESSED_PUBLIC_KEY_LENGTH:
raise ValueError(f"Encoded compressed public key has wrong length. Expected {COMPRESSED_PUBLIC_KEY_LENGTH}, got {len(public_key)}")
script_hash = self.get_script_hash(public_key)
data = bytearray(DECODE_ADDRESS_LENGTH)
data[0] = NEO_ADDRESS_VERSION
data[1:] = script_hash
return base58.b58encode_check(data).decode('utf-8')
def get_script_hash(self, public_key: bytes):
script = self.create_signature_redeem_script(public_key)
sha256_hash = hashlib.sha256(script).digest()
return self.get_ripemd160(sha256_hash)
@staticmethod
def create_signature_redeem_script(public_key: bytes):
if len(public_key) != COMPRESSED_PUBLIC_KEY_LENGTH:
raise ValueError(f"Encoded compressed public key has wrong length. Expected {COMPRESSED_PUBLIC_KEY_LENGTH}, got {len(public_key)}")
script = bytearray([PS_IN_HASH160, COMPRESSED_PUBLIC_KEY_LENGTH])
script.extend(public_key)
script.append(UNCOMPRESSED_PUBLIC_KEY_LENGTH)
script.extend(pack("<I", CHECK_SIG_DESCRIPTOR))
return bytes(script)
@staticmethod
def get_ripemd160(value):
if value is None:
raise ValueError("Input parameter is missing")
digest = RIPEMD160.new()
digest.update(value)
return digest.digest()
@staticmethod
def is_empty(sequence_symbols: bytes | str):
if len(sequence_symbols) == 0 or sequence_symbols is None:
raise ValueError(f"Empty sequence symbols of key: {sequence_symbols}")
return False
@staticmethod
def get_hex_string(value):
if value is None or len(value) == 0: