Docs: Merge examples into develop branch

examples/atc.py

@@ -15,6 +15,7 @@
 
 accts = get_accounts()
 acct = accts.pop()
+acct2 = accts.pop()
 
 algod_client = get_algod_client()
 
@@ -85,3 +86,18 @@
     boxes=[[app_id, b"key"]],
 )
 # example: ATC_BOX_REF
+
+# example: ATC_FOREIGN_REFS
+atc = AtomicTransactionComposer()
+atc.add_method_call(
+    app_id,
+    my_method,
+    addr,
+    sp,
+    signer,
+    accounts=[acct2.address],
+    foreign_apps=[1337],
+    foreign_assets=[42],
+    boxes=[[app_id, b"key"]],
+)
+# example: ATC_FOREIGN_REFS

examples/block_watcher/block_watcher.py

@@ -0,0 +1,229 @@
+from algosdk.v2client import algod
+from algosdk import encoding
+from algosdk import transaction
+from typing import List, Dict, Union
+import msgpack
+import base64
+
+# Mainnet Example
+# token = ""
+# host = "https://mainnet-api.algonode.cloud" # Mainnet
+
+# Sandbox Example
+token = "a" * 64
+host = "http://localhost:4001"
+client = algod.AlgodClient(token, host)
+
+
+class SignedTxnWithAD:
+    txn: Union[transaction.SignedTransaction, transaction.Transaction]
+    ad: "ApplyData"
+
+    @staticmethod
+    def from_msgp(stxn, gh, gen) -> "SignedTxnWithAD":
+        s = SignedTxnWithAD()
+        s.txn = parse_signed_transaction_msgp(stxn, gh, gen)
+        s.ad = ApplyData.from_msgp(stxn)
+        return s
+
+
+class ApplyData:
+    closing_amount: int
+    asset_closing_amount: int
+    sender_rewards: int
+    receiver_rewards: int
+    close_rewards: int
+    eval_delta: "EvalDelta"
+    config_asset: int
+    application_id: int
+
+    def __init__(
+        self,
+        closing_amount=0,
+        asset_closing_amount=0,
+        sender_rewards=0,
+        receiver_rewards=0,
+        close_rewards=0,
+        eval_delta=None,
+        config_asset=0,
+        application_id=0,
+    ):
+        self.closing_amount = closing_amount
+        self.asset_closing_amount = asset_closing_amount
+        self.sender_rewards = sender_rewards
+        self.receiver_rewards = receiver_rewards
+        self.close_rewards = close_rewards
+        self.eval_delta = eval_delta
+        self.config_asset = config_asset
+        self.application_id = application_id
+
+    @staticmethod
+    def from_msgp(apply_data: dict) -> "ApplyData":
+        ad = ApplyData()
+        if b"ca" in apply_data:
+            ad.closing_amount = apply_data[b"ca"]
+        if b"aca" in apply_data:
+            ad.asset_closing_amount = apply_data[b"aca"]
+        if b"rs" in apply_data:
+            ad.sender_rewards = apply_data[b"rs"]
+        if b"rr" in apply_data:
+            ad.receiver_rewards = apply_data[b"rr"]
+        if b"rc" in apply_data:
+            ad.close_rewards = apply_data[b"rc"]
+        if b"caid" in apply_data:
+            ad.config_asset = apply_data[b"caid"]
+        if b"apid" in apply_data:
+            ad.application_id = apply_data[b"apid"]
+        if b"dt" in apply_data:
+            ad.eval_delta = EvalDelta.from_msgp(apply_data[b"dt"])
+        return ad
+
+
+class EvalDelta:
+    global_delta: List["StateDelta"] | None
+    local_deltas: Dict[int, "StateDelta"] | None
+    logs: List[str]
+    inner_txns: List["SignedTxnWithAD"]
+
+    def __init__(
+        self,
+        global_delta: List["StateDelta"] | None = None,
+        local_deltas: Dict[int, "StateDelta"] | None = None,
+        logs: List[str] = [],
+        inner_txns: List["SignedTxnWithAD"] = [],
+    ):
+        self.global_delta = global_delta
+        self.local_deltas = local_deltas
+        self.logs = logs
+        self.inner_txns = inner_txns
+
+    @staticmethod
+    def from_msgp(delta: dict) -> "EvalDelta":
+        ed = EvalDelta()
+        if b"gd" in delta:
+            ed.global_delta = [
+                StateDelta.from_msgp(delta[b"gd"][idx]) for idx in delta[b"gd"]
+            ]
+        if b"ld" in delta:
+            ed.local_deltas = {
+                k: StateDelta.from_msgp(v) for k, v in delta[b"ld"].items()
+            }
+        if b"lg" in delta:
+            ed.logs = delta[b"lg"]
+        if b"itx" in delta:
+            ed.inner_txns = [
+                SignedTxnWithAD.from_msgp(itxn, b"", "")
+                for itxn in delta[b"itx"]
+            ]
+        return ed
+
+
+class StateDelta:
+    action: int
+    bytes: bytes
+    uint: int
+
+    @staticmethod
+    def from_msgp(state_delta: dict) -> "StateDelta":
+        sd = StateDelta()
+        if b"at" in state_delta:
+            sd.action = state_delta[b"at"]
+        if b"bs" in state_delta:
+            sd.bytes = base64.b64encode(state_delta[b"bs"])
+        if b"ui" in state_delta:
+            sd.uint = state_delta[b"ui"]
+        return sd
+
+
+def parse_signed_transaction_msgp(
+    txn: dict, gh: bytes, gen: str
+) -> transaction.Transaction:
+    stxn = {
+        "txn": {
+            "gh": gh,
+            "gen": gen,
+            **_stringify_keys(txn[b"txn"]),
+        }
+    }
+    if b"sig" in txn:
+        stxn["sig"] = txn[b"sig"]
+    if b"msig" in txn:
+        stxn["msig"] = _stringify_keys(txn[b"msig"])
+        stxn["msig"]["subsig"] = [
+            _stringify_keys(ss) for ss in stxn["msig"]["subsig"]
+        ]
+    if b"lsig" in txn:
+        stxn["lsig"] = _stringify_keys(txn[b"lsig"])
+    if b"sgnr" in txn:
+        stxn["sgnr"] = txn[b"sgnr"]
+    return encoding.msgpack_decode(stxn)
+
+
+def _stringify_keys(d: dict) -> dict:
+    return {k.decode("utf-8"): v for (k, v) in d.items()}
+
+
+def _txid_to_bytes(txid):
+    return base64.b32decode(encoding._correct_padding(txid))
+
+
+def _bytes_to_txid(b):
+    return base64.b32encode(b).strip(b"=").decode("utf-8")
+
+
+def get_itxn_id(
+    itxn: transaction.Transaction, caller: transaction.Transaction, idx: int
+) -> str:
+    input = b"TX" + _txid_to_bytes(caller.get_txid())
+    input += idx.to_bytes(8, "big")
+    input += base64.b64decode(encoding.msgpack_encode(itxn))
+    return _bytes_to_txid(encoding.checksum(input))
+
+
+def print_ids_recursive(swad: SignedTxnWithAD, level: int):
+    if swad.ad.eval_delta is None:
+        return
+
+    for idx in range(len(swad.ad.eval_delta.inner_txns)):
+        itxn = swad.ad.eval_delta.inner_txns[idx]
+        # These are Transactions not SignedTransactions
+        print(
+            "{} {}: {}".format(
+                "\t" * (level + 1),
+                itxn.txn.type,
+                get_itxn_id(itxn.txn, swad.txn, idx),
+            )
+        )
+        if (
+            itxn.ad.eval_delta is not None
+            and len(itxn.ad.eval_delta.inner_txns) > 0
+        ):
+            print_ids_recursive(itxn, level + 1)
+
+
+if __name__ == "__main__":
+    # Get current round
+    round = client.status()["last-round"]
+
+    while True:
+        block = client.block_info(round, response_format="msgp")
+        dblock = msgpack.unpackb(block, raw=True, strict_map_key=False)
+
+        raw_block = dblock[b"block"]
+        if b"txns" not in raw_block:
+            round = client.status_after_block(round)["last-round"]
+            continue
+
+        gh = raw_block[b"gh"]
+        gen = raw_block[b"gen"].decode("utf-8")
+
+        tx_count = len(raw_block[b"txns"])
+        print(f"Round {round} had {tx_count} txns")
+
+        # Construct SignedTransaction object and print TxIDs
+        for stxn in raw_block[b"txns"]:
+            swad = SignedTxnWithAD.from_msgp(stxn, gh, gen)
+            print(swad.txn.get_txid())
+
+        # Wait for next round
+        round = client.status_after_block(round)["last-round"]

examples/codec.py

@@ -3,6 +3,7 @@
 from utils import get_algod_client, get_accounts
 from algosdk import transaction
 from algosdk import encoding
+from algosdk import logic
 
 # example: CODEC_ADDRESS
 address = "4H5UNRBJ2Q6JENAXQ6HNTGKLKINP4J4VTQBEPK5F3I6RDICMZBPGNH6KD4"
@@ -12,6 +13,14 @@
 assert addr == address
 # example: CODEC_ADDRESS
 
+# example: CODEC_APPLICATION_ACCOUNT
+app_id = 123
+app_addr = logic.get_application_address(app_id)
+
+print(f"Application ID:   {app_id}")
+print(f"Application Addr: {app_addr}")
+# example: CODEC_APPLICATION_ACCOUNT
+
 # example: CODEC_BASE64
 encoded_str = "SGksIEknbSBkZWNvZGVkIGZyb20gYmFzZTY0"
 decoded_str = base64.b64decode(encoded_str).decode("utf-8")

examples/verify_signature.py

@@ -0,0 +1,49 @@
+from algosdk.account import generate_account
+from algosdk import encoding, transaction
+from nacl.signing import VerifyKey
+from nacl.exceptions import BadSignatureError
+from typing import List
+import base64
+
+
+def verify_signed_transaction(raw_stxn: List[bytes]):
+    # example: OFFLINE_VERIFY_SIG
+    # decode the signed transaction
+    stxn = encoding.msgpack_decode(raw_stxn)
+    if stxn.signature is None or len(stxn.signature) == 0:
+        return False
+
+    public_key = stxn.transaction.sender
+    if stxn.authorizing_address is not None:
+        public_key = stxn.authorizing_address
+
+    # Create a VerifyKey from nacl using the 32 byte public key
+    verify_key = VerifyKey(encoding.decode_address(public_key))
+
+    # Generate the message that was signed with TX prefix
+    prefixed_message = b"TX" + base64.b64decode(
+        encoding.msgpack_encode(stxn.transaction)
+    )
+
+    try:
+        # Verify the signature
+        verify_key.verify(prefixed_message, base64.b64decode(stxn.signature))
+        return True
+    except BadSignatureError:
+        return False
+    # example: OFFLINE_VERIFY_SIG
+
+
+def main():
+    sk, addr = generate_account()
+    sp = transaction.SuggestedParams(
+        0, 1, 2, "nDtgfcrOMvfAaxYZSL+gCqA2ot5uBknFJuWE4pVFloo="
+    )
+    txn = transaction.PaymentTxn(addr, sp, addr, 0)
+    stxn = txn.sign(sk)
+    stxn_blob = encoding.msgpack_encode(stxn)
+    print("Valid? ", verify_signed_transaction(stxn_blob))
+
+
+if __name__ == "__main__":
+    main()