Skip to content

fix: rectify STIssue codec (MPTCurrency case only) #870

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
+101 −18
Open

fix: rectify STIssue codec (MPTCurrency case only) #870

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 22 additions & 0 deletions tests/unit/core/binarycodec/types/test_account_id.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,3 +22,25 @@ def test_from_value_base58(self):
def test_raises_invalid_value_type(self):
invalid_value = 30
self.assertRaises(XRPLBinaryCodecException, AccountID.from_value, invalid_value)

def test_special_account_ACCOUNT_ONE(self):
self.assertEqual(
AccountID.from_value("0000000000000000000000000000000000000001").to_json(),
"rrrrrrrrrrrrrrrrrrrrBZbvji",
)

self.assertEqual(
AccountID.from_value("rrrrrrrrrrrrrrrrrrrrBZbvji").to_hex(),
AccountID.from_value("0000000000000000000000000000000000000001").to_hex(),
)

def test_special_account_ACCOUNT_ZERO(self):
self.assertEqual(
AccountID.from_value("0000000000000000000000000000000000000000").to_json(),
"rrrrrrrrrrrrrrrrrrrrrhoLvTp",
)

self.assertEqual(
AccountID.from_value("rrrrrrrrrrrrrrrrrrrrrhoLvTp").to_hex(),
AccountID.from_value("0000000000000000000000000000000000000000").to_hex(),
)
29 changes: 26 additions & 3 deletions tests/unit/core/binarycodec/types/test_issue.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,30 @@ def test_from_value_mpt(self):
}
self.assertEqual(issue_obj.to_json(), expected)

def test_short_mpt_issuance_id(self):
# A valid mpt_issuance_id is 192 bits long (or 48 characters in hex).
test_input = {
"mpt_issuance_id": "A" * 47,
}
self.assertRaises(XRPLBinaryCodecException, Issue.from_value, test_input)

def test_binary_representation_of_mpt_issuance_id(self):
# The issuer_account is represented by `A` and Sequence number
# (of the MPTokenIssuanceCreate transaction) is represented by `B`.
mpt_issuance_id_in_hex = "A" * 40 + "B" * 8
test_input = {
"mpt_issuance_id": mpt_issuance_id_in_hex,
}
issue_obj = Issue.from_value(test_input)
self.assertEqual(
issue_obj.to_hex(),
mpt_issuance_id_in_hex[:40]
# the below line is the hex representation of the
# black-hole-account-id (ACCOUNT_ONE)
+ "0000000000000000000000000000000000000001" + mpt_issuance_id_in_hex[40:],
)
self.assertEqual(issue_obj.to_json(), test_input)

def test_from_parser_xrp(self):
# Test round-trip: serialize an XRP Issue and then parse it back.
test_input = {"currency": "XRP"}
Expand Down Expand Up @@ -79,10 +103,9 @@ def test_from_parser_mpt(self):
"mpt_issuance_id": "BAADF00DBAADF00DBAADF00DBAADF00DBAADF00DBAADF00D",
}
issue_obj = Issue.from_value(test_input)
# Use the hex representation and pass the fixed length_hint (24 bytes for
# Hash192)
# Use the hex representation
parser = BinaryParser(issue_obj.to_hex())
issue_from_parser = Issue.from_parser(parser, length_hint=24)
issue_from_parser = Issue.from_parser(parser)
expected = {
"mpt_issuance_id": "BAADF00DBAADF00DBAADF00DBAADF00DBAADF00DBAADF00D"
}
Expand Down
68 changes: 53 additions & 15 deletions xrpl/core/binarycodec/types/issue.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,9 @@
from xrpl.core.binarycodec.exceptions import XRPLBinaryCodecException
from xrpl.core.binarycodec.types.account_id import AccountID
from xrpl.core.binarycodec.types.currency import Currency
from xrpl.core.binarycodec.types.hash192 import HASH192_BYTES, Hash192
from xrpl.core.binarycodec.types.hash192 import Hash192
from xrpl.core.binarycodec.types.serialized_type import SerializedType
from xrpl.core.binarycodec.types.uint32 import UInt32
from xrpl.models.currencies import XRP as XRPModel
from xrpl.models.currencies import IssuedCurrency as IssuedCurrencyModel
from xrpl.models.currencies import MPTCurrency as MPTCurrencyModel
Expand All @@ -20,6 +21,10 @@
class Issue(SerializedType):
"""Codec for serializing and deserializing issued currency fields."""

BLACK_HOLED_ACCOUNT_ID = AccountID.from_value(
"0000000000000000000000000000000000000001"
)

def __init__(self: Self, buffer: bytes) -> None:
"""
Construct an Issue from given bytes.
Expand Down Expand Up @@ -54,9 +59,26 @@ def from_value(cls: Type[Self], value: Dict[str, str]) -> Self:
issuer_bytes = bytes(AccountID.from_value(value["issuer"]))
return cls(currency_bytes + issuer_bytes)

# MPT is serialized as:
# - 160 bits MPT issuer account (20 bytes)
# - 160 bits black hole account (20 bytes)
# - 32 bits sequence (4 bytes)
# Please look at STIssue.cpp inside rippled implementation for more details.

if MPTCurrencyModel.is_dict_of_model(value):
if len(value["mpt_issuance_id"]) != 48:
raise XRPLBinaryCodecException(
"Invalid mpt_issuance_id length: expected 48 characters, "
f"received {len(value['mpt_issuance_id'])} characters."
)
mpt_issuance_id_bytes = bytes(Hash192.from_value(value["mpt_issuance_id"]))
return cls(bytes(mpt_issuance_id_bytes))
return cls(
bytes(
mpt_issuance_id_bytes[:20]
+ bytes(cls.BLACK_HOLED_ACCOUNT_ID)
+ bytes(mpt_issuance_id_bytes[20:])
)
)

raise XRPLBinaryCodecException(
"Invalid type to construct an Issue: expected XRP, IssuedCurrency or "
Expand All @@ -80,17 +102,26 @@ def from_parser(
Returns:
The Issue object constructed from a parser.
"""
# Check if it's an MPTIssue by checking mpt_issuance_id byte size
if length_hint == HASH192_BYTES:
mpt_bytes = parser.read(HASH192_BYTES)
return cls(mpt_bytes)
currency_or_account = Currency.from_parser(parser)
if currency_or_account.to_json() == "XRP":
return cls(bytes(currency_or_account))

# check if this is an instance of MPTIssuanceID
issuer_account_id = AccountID.from_parser(parser)
if issuer_account_id.to_json() == cls.BLACK_HOLED_ACCOUNT_ID.to_json():
sequence = UInt32.from_parser(parser)
return cls(
bytes(currency_or_account)
+ bytes(cls.BLACK_HOLED_ACCOUNT_ID)
+ bytes(sequence)
)

return cls(bytes(currency_or_account) + bytes(issuer_account_id))

currency = Currency.from_parser(parser)
if currency.to_json() == "XRP":
return cls(bytes(currency))

issuer = parser.read(20) # the length in bytes of an account ID
return cls(bytes(currency) + issuer)
@classmethod
def _print_buffer(self: Self, buffer: bytes) -> None:
print("DEBUG: Inside Issue._print_buffer(), buffer: ", buffer.hex().upper())
print("DEBUG: Inside Issue._print_buffer(), buffer length: ", len(buffer))
Comment on lines +121 to +124
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue

Fix the incorrect method signature for _print_buffer.

The method has incorrect type annotation - it should use cls for class methods or no first parameter annotation for instance methods, not self with a class method decorator.

Apply this diff to fix the method signature:

-    @classmethod
-    def _print_buffer(self: Self, buffer: bytes) -> None:
+    def _print_buffer(self: Self, buffer: bytes) -> None:

Alternatively, if this should be a class method:

     @classmethod
-    def _print_buffer(self: Self, buffer: bytes) -> None:
+    def _print_buffer(cls: Type[Self], buffer: bytes) -> None:

Also consider whether this debug method should remain in production code, or if it should be removed before merging.

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
@classmethod
def _print_buffer(self: Self, buffer: bytes) -> None:
print("DEBUG: Inside Issue._print_buffer(), buffer: ", buffer.hex().upper())
print("DEBUG: Inside Issue._print_buffer(), buffer length: ", len(buffer))
def _print_buffer(self: Self, buffer: bytes) -> None:
print("DEBUG: Inside Issue._print_buffer(), buffer: ", buffer.hex().upper())
print("DEBUG: Inside Issue._print_buffer(), buffer length: ", len(buffer))
🧰 Tools
🪛 GitHub Actions: Unit test

[error] 122-122: mypy error: Method cannot have explicit self annotation and Self type.

[error] 122-122: mypy error: The erased type of self "xrpl.core.binarycodec.types.issue.Issue" is not a supertype of its class "type[xrpl.core.binarycodec.types.issue.Issue]".

🤖 Prompt for AI Agents 
In xrpl/core/binarycodec/types/issue.py around lines 121 to 124, the
@classmethod uses an incorrect first parameter named self; change the signature
to use cls (def _print_buffer(cls, buffer: bytes) -> None:) if you intend a
class method, or remove @classmethod and use def _print_buffer(self, buffer:
bytes) -> None: for an instance method; also consider removing or gating the
debug prints (print(...)) before merging so this debug helper is not left in
production code.


def to_json(self: Self) -> Union[str, Dict[Any, Any]]:
"""
Expand All @@ -99,9 +130,16 @@ def to_json(self: Self) -> Union[str, Dict[Any, Any]]:
Returns:
The JSON representation of an Issue.
"""
# If the buffer is exactly 24 bytes, treat it as an MPT amount.
if len(self.buffer) == HASH192_BYTES:
return {"mpt_issuance_id": self.to_hex().upper()}
# If the buffer's length is 44 bytes (issuer-account + black-hole-account-id +
# sequence), treat it as a MPTCurrency.
# Note: hexadecimal representation of the buffer's length is doubled because 1
# byte is represented by 2 characters in hex.
if len(self.buffer) == 20 + 20 + 4:
serialized_mpt_in_hex = self.to_hex().upper()
return {
"mpt_issuance_id": serialized_mpt_in_hex[:40]
+ serialized_mpt_in_hex[80:]
}

parser = BinaryParser(self.to_hex())
currency: Union[str, Dict[Any, Any]] = Currency.from_parser(parser).to_json()
Expand Down
Loading