Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
24 changes: 17 additions & 7 deletions application/account_facts.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
import hashlib
import json
import re
import unicodedata
from collections.abc import Mapping
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation, localcontext
Expand All @@ -28,7 +29,19 @@
MAX_EARN_POSITIONS = 10_000
_HEX_SHA256 = re.compile(r"[0-9a-f]{64}\Z")
_GIT_SHA = re.compile(r"[0-9a-f]{40}\Z")
_ASSET = re.compile(r"[A-Z0-9]{1,128}\Z")
_HANGUL_FILLERS = frozenset("\u115f\u1160\u3164\uffa0")


def _valid_asset(value: Any) -> bool:
if not isinstance(value, str) or not 1 <= len(value) <= 128:
return False
for character in value:
if character.isascii():
if not ("A" <= character <= "Z" or "0" <= character <= "9"):
return False
elif character in _HANGUL_FILLERS or unicodedata.category(character)[0] not in {"L", "N"}:
return False
return True


class AccountFactsUnavailable(ValueError):
Expand Down Expand Up @@ -189,9 +202,7 @@ def collect_account_facts(
asset = row.get("asset")
if not isinstance(asset, str):
raise _fail("spot_asset_type_invalid")
if not asset.isascii():
raise _fail("spot_asset_non_ascii")
if not _ASSET.fullmatch(asset):
if not _valid_asset(asset):
raise _fail("spot_asset_format_invalid")
if asset in assets:
raise _fail("spot_asset_duplicate")
Expand Down Expand Up @@ -237,8 +248,7 @@ def collect_account_facts(
asset = row.get("asset")
product_id = row.get("productId")
if (
not isinstance(asset, str)
or not _ASSET.fullmatch(asset)
not _valid_asset(asset)
or not isinstance(product_id, str)
or not product_id
or product_id in seen_products
Expand Down Expand Up @@ -383,7 +393,7 @@ def validate_account_facts_payload(payload: Any) -> dict[str, Any]:
}:
raise _fail("payload_invalid")
asset = row.get("asset")
if not isinstance(asset, str) or not _ASSET.fullmatch(asset) or asset in seen_assets:
if not _valid_asset(asset) or asset in seen_assets:
raise _fail("payload_invalid")
seen_assets.add(asset)
amount_texts = [row.get(name) for name in ("spot_free", "spot_locked", "flexible_earn", "quantity")]
Expand Down
86 changes: 85 additions & 1 deletion tests/test_binance_account_facts.py
Original file line number Diff line number Diff line change
Expand Up @@ -446,7 +446,7 @@ def get_account(self):
({"uid": "123456", "balances": ["synthetic-private-row"]}, "spot_balance_row_invalid"),
({"uid": "123456", "balances": [{"asset": 17}]}, "spot_asset_type_invalid"),
({"uid": "123456", "balances": [{"asset": "synthetic-bad-symbol"}]}, "spot_asset_format_invalid"),
({"uid": "123456", "balances": [{"asset": "币"}]}, "spot_asset_non_ascii"),
({"uid": "123456", "balances": [{"asset": "A\u200b"}]}, "spot_asset_format_invalid"),
({"uid": "123456", "balances": [
{"asset": "SYNTHETIC", "free": "0", "locked": "0"},
{"asset": "SYNTHETIC", "free": "0", "locked": "0"},
Expand All @@ -465,6 +465,90 @@ def get_account(self):
assert client.calls == [("spot",)]


@pytest.mark.parametrize("asset", ["币", "12", "𐐀", "A" * 128, "汉" * 128])
def test_unicode_asset_rules_preserve_spot_earn_and_payload_values(asset):
class UnicodeAsset(FakeBinance):
def get_account(self):
self.calls.append(("spot",))
return {
"uid": "123456",
"balances": [{"asset": asset, "free": "0.5", "locked": "0"}],
}

def get_simple_earn_flexible_product_position(self, *, current, size):
self.calls.append(("earn", current, size))
return {
"rows": [{"asset": asset, "productId": "synthetic-product", "totalAmount": "0.25"}],
"total": 1,
}

client = UnicodeAsset()
payload = _collect(client)
assert payload["assets"] == [{
"asset": asset,
"quantity": "0.75",
"spot_free": "0.5",
"spot_locked": "0",
"flexible_earn": "0.25",
}]
encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
round_tripped = json.loads(encoded)
assert round_tripped["assets"][0]["asset"] == asset
validate_account_facts_payload(round_tripped)


@pytest.mark.parametrize("asset", [
"", " ", "btc", "A-B", "A\u0301", "A\u200b", "A\u202e", "A\ud800", "😀", "A" * 129,
"A\u115f", "A\u1160", "A\u3164", "A\uffa0",
])
def test_unicode_asset_rules_reject_unsafe_or_out_of_contract_values(asset):
class InvalidAsset(FakeBinance):
def get_account(self):
self.calls.append(("spot",))
return {
"uid": "123456",
"balances": [{"asset": asset, "free": "0.5", "locked": "0"}],
}

with pytest.raises(AccountFactsUnavailable, match="spot_asset_format_invalid"):
_collect(InvalidAsset())


@pytest.mark.parametrize("asset", ["A\u115f", "A\u1160", "A\u3164", "A\uffa0"])
def test_unicode_earn_asset_rejects_invisible_hangul_fillers(asset):
client = FakeBinance(earn_pages=[{
"rows": [{"asset": asset, "productId": "synthetic-product", "totalAmount": "0.25"}],
"total": 1,
}])
with pytest.raises(AccountFactsUnavailable, match="flexible_earn_page_invalid"):
_collect(client)
assert client.calls == [("spot",), ("earn", 1, 100)]


@pytest.mark.parametrize("asset", ["A\u115f", "A\u1160", "A\u3164", "A\uffa0"])
def test_payload_validator_rejects_invisible_hangul_fillers(asset):
payload = _collect()
payload["assets"][0]["asset"] = asset
with pytest.raises(AccountFactsUnavailable, match="payload_invalid"):
validate_account_facts_payload(payload)


def test_exact_duplicate_unicode_asset_is_rejected_without_normalization():
class DuplicateUnicodeAsset(FakeBinance):
def get_account(self):
self.calls.append(("spot",))
return {
"uid": "123456",
"balances": [
{"asset": "币", "free": "0.5", "locked": "0"},
{"asset": "币", "free": "0.5", "locked": "0"},
],
}

with pytest.raises(AccountFactsUnavailable, match="spot_asset_duplicate"):
_collect(DuplicateUnicodeAsset())


def test_rejects_earn_asset_without_an_explicit_spot_balance_row():
class SpotWithoutEarnAsset(FakeBinance):
def get_account(self):
Expand Down
Loading