Enhancement: Deprecating use of langspec

.test-env

@@ -1,6 +1,7 @@
 # Configs for testing repo download:
 SDK_TESTING_URL="https://github.com/algorand/algorand-sdk-testing"
-SDK_TESTING_BRANCH="master"
+# TODO revert back to "master" after sdk-testing is merged
+SDK_TESTING_BRANCH="deprecate-langspec"
 SDK_TESTING_HARNESS="test-harness"
 
 VERBOSE_HARNESS=0

algosdk/future/transaction.py

@@ -2505,8 +2505,7 @@ class LogicSig:
     """
 
     def __init__(self, program, args=None):
-        if not program or not logic.check_program(program, args):
-            raise error.InvalidProgram()
+        logic._sanity_check_program(program)
         self.logic = program
         self.args = args
         self.sig = None
@@ -2548,7 +2547,7 @@ def verify(self, public_key):
             return False
 
         try:
-            logic.check_program(self.logic, self.args)
+            logic._sanity_check_program(self.logic)
         except error.InvalidProgram:
             return False
 

algosdk/logic.py

@@ -1,4 +1,5 @@
 import base64
+import binascii
 import json
 import os
 
@@ -12,8 +13,54 @@
 opcodes = None
 
 
+def _sanity_check_program(program):
+    """
+    Performs heuristic program validation:
+    check if passed in bytes are Algorand address, or they are B64 encoded, rather than Teal bytes
+
+    Args:
+        program (bytes): compiled program
+    """
+
+    def is_ascii_printable(program_bytes):
+        return all(
+            map(
+                lambda x: x == ord("\n") or (ord(" ") <= x <= ord("~")),
+                program_bytes,
+            )
+        )
+
+    if not program:
+        raise error.InvalidProgram("empty program")
+
+    if is_ascii_printable(program):
+        try:
+            encoding.decode_address(program.decode("utf-8"))
+            raise error.InvalidProgram(
+                "requesting program bytes, get Algorand address"
+            )
+        except error.WrongChecksumError:
+            pass
+        except error.WrongKeyLengthError:
+            pass
+
+        try:
+            base64.b64decode(program.decode("utf-8"))
+            raise error.InvalidProgram("program should not be b64 encoded")
+        except binascii.Error:
+            pass
+
+        raise error.InvalidProgram(
+            "program bytes are all ASCII printable characters, not looking like Teal byte code"
+        )
+
+
 def check_program(program, args=None):
     """
+    NOTE: This class is deprecated:
+    Validation relies on metadata (`langspec.json`) that does not accurately represent opcode behavior across program versions.
+    The behavior of `check_program` relies on `langspec.json`. Thus, this method is being deprecated.
+
     Performs program checking for max length and cost
 
     Args:
@@ -31,6 +78,12 @@ def check_program(program, args=None):
 
 
 def read_program(program, args=None):
+    """
+    NOTE: This class is deprecated:
+    Validation relies on metadata (`langspec.json`) that does not accurately represent opcode behavior across program versions.
+    The behavior of `read_program` relies on `langspec.json`. Thus, this method is being deprecated.
+    """
+
     global spec, opcodes
     intcblock_opcode = 32
     bytecblock_opcode = 38
@@ -107,11 +160,17 @@ def read_program(program, args=None):
 
 
 def check_int_const_block(program, pc):
+    """
+    NOTE: This class is deprecated
+    """
     size, _ = read_int_const_block(program, pc)
     return size
 
 
 def read_int_const_block(program, pc):
+    """
+    NOTE: This class is deprecated
+    """
     size = 1
     ints = []
     num_ints, bytes_used = parse_uvarint(program[pc + size :])
@@ -134,11 +193,17 @@ def read_int_const_block(program, pc):
 
 
 def check_byte_const_block(program, pc):
+    """
+    NOTE: This class is deprecated
+    """
     size, _ = read_byte_const_block(program, pc)
     return size
 
 
 def read_byte_const_block(program, pc):
+    """
+    NOTE: This class is deprecated
+    """
     size = 1
     bytearrays = []
     num_ints, bytes_used = parse_uvarint(program[pc + size :])
@@ -164,11 +229,17 @@ def read_byte_const_block(program, pc):
 
 
 def check_push_int_block(program, pc):
+    """
+    NOTE: This class is deprecated
+    """
     size, _ = read_push_int_block(program, pc)
     return size
 
 
 def read_push_int_block(program, pc):
+    """
+    NOTE: This class is deprecated
+    """
     size = 1
     single_int, bytes_used = parse_uvarint(program[pc + size :])
     if bytes_used <= 0:
@@ -180,11 +251,17 @@ def read_push_int_block(program, pc):
 
 
 def check_push_byte_block(program, pc):
+    """
+    NOTE: This class is deprecated
+    """
     size, _ = read_push_byte_block(program, pc)
     return size
 
 
 def read_push_byte_block(program, pc):
+    """
+    NOTE: This class is deprecated
+    """
     size = 1
     item_len, bytes_used = parse_uvarint(program[pc + size :])
     if bytes_used <= 0:
@@ -200,6 +277,9 @@ def read_push_byte_block(program, pc):
 
 
 def parse_uvarint(buf):
+    """
+    NOTE: This class is deprecated
+    """
     x = 0
     s = 0
     for i, b in enumerate(buf):
@@ -215,6 +295,8 @@ def parse_uvarint(buf):
 
 def address(program):
     """
+    NOTE: This class is deprecated
+
     Return the address of the program.
 
     Args:
@@ -230,6 +312,8 @@ def address(program):
 
 def teal_sign(private_key, data, contract_addr):
     """
+    NOTE: This class is deprecated
+
     Return the signature suitable for ed25519verify TEAL opcode
 
     Args:
@@ -254,6 +338,8 @@ def teal_sign(private_key, data, contract_addr):
 
 def teal_sign_from_program(private_key, data, program):
     """
+    NOTE: This class is deprecated
+
     Return the signature suitable for ed25519verify TEAL opcode
 
     Args:
@@ -270,6 +356,8 @@ def teal_sign_from_program(private_key, data, program):
 
 def get_application_address(appID: int) -> str:
     """
+    NOTE: This class is deprecated
+
     Return the escrow address of an application.
 
     Args:

algosdk/testing/dryrun.py

@@ -511,7 +511,7 @@ def build_dryrun_request(
     ):
         """
         Helper function for creation DryrunRequest object from a program.
-        By default it uses logic sig mode
+        By default, it uses logic sig mode
         and if app_idx / on_complete are set then application call is made
 
         Args:

algosdk/transaction.py

@@ -1318,8 +1318,7 @@ class LogicSig:
     """
 
     def __init__(self, program, args=None):
-        if not program or not logic.check_program(program, args):
-            raise error.InvalidProgram()
+        logic._sanity_check_program(program)
         self.logic = program
         self.args = args
         self.sig = None
@@ -1361,7 +1360,7 @@ def verify(self, public_key):
             return False
 
         try:
-            logic.check_program(self.logic, self.args)
+            logic._sanity_check_program(self.logic)
         except error.InvalidProgram:
             return False
 

tests/steps/steps.py

@@ -1,6 +1,7 @@
 import base64
 import random
 import time
+import parse
 
 from algosdk import (
     account,
@@ -14,9 +15,18 @@
     wallet,
 )
 from algosdk.future import transaction
-from behave import given, then, when
+from behave import given, then, when, register_type
 from nacl.signing import SigningKey
 
+
+@parse.with_pattern(r".*")
+def parse_string(text):
+    return text
+
+
+register_type(MaybeString=parse_string)
+
+
 token = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
 algod_port = 60000
 kmd_port = 60001
@@ -984,3 +994,28 @@ def buildTxn(t, sender, params):
     elif "nonparticipation" in t:
         txn = transaction.KeyregNonparticipatingTxn(sender, params)
     return txn
+
+
+@given(
+    'a base64 encoded program bytes for heuristic sanity check "{b64encoded:MaybeString}"'
+)
+def take_b64_encoded_bytes(context, b64encoded):
+    context.seemingly_program = base64.b64decode(b64encoded)
+
+
+@when("I start heuristic sanity check over the bytes")
+def heuristic_check_over_bytes(context):
+    context.sanity_check_err = ""
+
+    try:
+        transaction.LogicSig(context.seemingly_program)
+    except Exception as e:
+        context.sanity_check_err = str(e)
+
+
+@then('if there exists an error, the error contains "{err_msg:MaybeString}"')
+def check_error_if_matching(context, err_msg: str = None):
+    if len(err_msg) > 0:
+        assert err_msg in context.sanity_check_err
+    else:
+        assert len(context.sanity_check_err) == 0

tests/unit.tags

@@ -12,6 +12,7 @@
 @unit.indexer.ledger_refactoring
 @unit.indexer.logs
 @unit.offline
+@unit.program_sanity_check
 @unit.rekey
 @unit.responses
 @unit.responses.231
@@ -20,4 +21,4 @@
 @unit.tealsign
 @unit.transactions
 @unit.transactions.keyreg
-@unit.transactions.payment
+@unit.transactions.payment