Enhancement: Deprecating use of langspec

algosdk/future/transaction.py

@@ -1,5 +1,6 @@
 from typing import List, Union
 import base64
+import binascii
 from enum import IntEnum
 import msgpack
 from collections import OrderedDict
@@ -2510,13 +2511,54 @@ class LogicSig:
     """
 
     def __init__(self, program, args=None):
-        if not program or not logic.check_program(program, args):
-            raise error.InvalidProgram()
+        self._sanity_check_program(program)
         self.logic = program
         self.args = args
         self.sig = None
         self.msig = None
 
+    @staticmethod
+    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 dictify(self):
         od = OrderedDict()
         if self.args:
@@ -2553,7 +2595,7 @@ def verify(self, public_key):
             return False
 
         try:
-            logic.check_program(self.logic, self.args)
+            self._sanity_check_program(self.logic)
         except error.InvalidProgram:
             return False
 

algosdk/logic.py

@@ -1,6 +1,7 @@
 import base64
 import json
 import os
+import warnings
 
 from . import constants
 from . import error
@@ -14,6 +15,10 @@
 
 def check_program(program, args=None):
     """
+    NOTE: This method 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:
@@ -26,11 +31,29 @@ def check_program(program, args=None):
     Raises:
         InvalidProgram: on error
     """
+    warnings.warn(
+        "`check_program` relies on metadata (`langspec.json`) that "
+        "does not accurately represent opcode behavior across program versions. "
+        "This method is being deprecated.",
+        DeprecationWarning,
+    )
     ok, _, _ = read_program(program, args)
     return ok
 
 
 def read_program(program, args=None):
+    """
+    NOTE: This method 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.
+    """
+    warnings.warn(
+        "`read_program` relies on metadata (`langspec.json`) that "
+        "does not accurately represent opcode behavior across program versions. "
+        "This method is being deprecated.",
+        DeprecationWarning,
+    )
+
     global spec, opcodes
     intcblock_opcode = 32
     bytecblock_opcode = 38
@@ -107,11 +130,25 @@ def read_program(program, args=None):
 
 
 def check_int_const_block(program, pc):
+    """
+    NOTE: This method is deprecated
+    """
+    warnings.warn(
+        "This method is being deprecated.",
+        DeprecationWarning,
+    )
     size, _ = read_int_const_block(program, pc)
     return size
 
 
 def read_int_const_block(program, pc):
+    """
+    NOTE: This method is deprecated
+    """
+    warnings.warn(
+        "This method is being deprecated.",
+        DeprecationWarning,
+    )
     size = 1
     ints = []
     num_ints, bytes_used = parse_uvarint(program[pc + size :])
@@ -134,11 +171,25 @@ def read_int_const_block(program, pc):
 
 
 def check_byte_const_block(program, pc):
+    """
+    NOTE: This method is deprecated
+    """
+    warnings.warn(
+        "This method is being deprecated.",
+        DeprecationWarning,
+    )
     size, _ = read_byte_const_block(program, pc)
     return size
 
 
 def read_byte_const_block(program, pc):
+    """
+    NOTE: This method is deprecated
+    """
+    warnings.warn(
+        "This method is being deprecated.",
+        DeprecationWarning,
+    )
     size = 1
     bytearrays = []
     num_ints, bytes_used = parse_uvarint(program[pc + size :])
@@ -164,11 +215,25 @@ def read_byte_const_block(program, pc):
 
 
 def check_push_int_block(program, pc):
+    """
+    NOTE: This method is deprecated
+    """
+    warnings.warn(
+        "This method is being deprecated.",
+        DeprecationWarning,
+    )
     size, _ = read_push_int_block(program, pc)
     return size
 
 
 def read_push_int_block(program, pc):
+    """
+    NOTE: This method is deprecated
+    """
+    warnings.warn(
+        "This method is being deprecated.",
+        DeprecationWarning,
+    )
     size = 1
     single_int, bytes_used = parse_uvarint(program[pc + size :])
     if bytes_used <= 0:
@@ -180,11 +245,25 @@ def read_push_int_block(program, pc):
 
 
 def check_push_byte_block(program, pc):
+    """
+    NOTE: This method is deprecated
+    """
+    warnings.warn(
+        "This method is being deprecated.",
+        DeprecationWarning,
+    )
     size, _ = read_push_byte_block(program, pc)
     return size
 
 
 def read_push_byte_block(program, pc):
+    """
+    NOTE: This method is deprecated
+    """
+    warnings.warn(
+        "This method is being deprecated.",
+        DeprecationWarning,
+    )
     size = 1
     item_len, bytes_used = parse_uvarint(program[pc + size :])
     if bytes_used <= 0:
@@ -200,6 +279,13 @@ def read_push_byte_block(program, pc):
 
 
 def parse_uvarint(buf):
+    """
+    NOTE: This method is deprecated
+    """
+    warnings.warn(
+        "This method is being deprecated.",
+        DeprecationWarning,
+    )
     x = 0
     s = 0
     for i, b in enumerate(buf):

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

@@ -1,5 +1,6 @@
 import base64
 import msgpack
+import binascii
 from collections import OrderedDict
 from . import account
 from . import constants
@@ -1318,13 +1319,54 @@ class LogicSig:
     """
 
     def __init__(self, program, args=None):
-        if not program or not logic.check_program(program, args):
-            raise error.InvalidProgram()
+        self._sanity_check_program(program)
         self.logic = program
         self.args = args
         self.sig = None
         self.msig = None
 
+    @staticmethod
+    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 dictify(self):
         od = OrderedDict()
         if self.args:
@@ -1361,7 +1403,7 @@ def verify(self, public_key):
             return False
 
         try:
-            logic.check_program(self.logic, self.args)
+            self._sanity_check_program(self.logic)
         except error.InvalidProgram:
             return False
 

tests/steps/steps.py

@@ -1,6 +1,9 @@
+import os
 import base64
 import random
 import time
+import parse
+from datetime import datetime
 
 from algosdk import (
     account,
@@ -22,12 +25,19 @@
 from algosdk import auction
 from algosdk import util
 from algosdk import logic
-import os
-from datetime import datetime
 
-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
@@ -995,3 +1005,30 @@ 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 the heuristic sanity check throws 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