Support POST signed_operations, but without the contract's address.

src/routes.py

@@ -248,7 +248,7 @@ async def get_entrypoint(
 # Operations
 @router.post("/operation")
 async def post_operation(
-    call_data: schemas.CallData, db: Session = Depends(database.get_db)
+    call_data: schemas.UnsignedCall, db: Session = Depends(database.get_db)
 ):
     if len(call_data.operations) == 0:
         raise HTTPException(
@@ -295,8 +295,10 @@ async def post_operation(
                 status_code=status.HTTP_400_BAD_REQUEST,
                 detail="Not enough funds."
             )
-        result = await tezos.tezos_manager.queue_operation(call_data.sender,
-                                                           op)
+        result = await tezos.tezos_manager.queue_operation(
+            call_data.sender_address,
+            op
+        )
     except MichelsonError as e:
         print("Received failing operation, discarding")
         print(e)
@@ -312,3 +314,26 @@ async def post_operation(
             detail=f"Unknown exception raised.",
         )
     return result
+
+
+@router.post("/signed_operation")
+async def signed_operation(
+    call_data: schemas.SignedCall, db: Session = Depends(database.get_db)
+):
+    # In order for the user to sign Micheline, we need to
+    # FIXME: this is a serious issue, we should sign the contract address too.
+    signed_data = [x["parameters"]["value"] for x in call_data.operations]
+    if not tezos.check_signature(
+        signed_data,
+        call_data.signature,
+        call_data.sender_key,
+        call_data.micheline_type
+    ):
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Invalid signature."
+        )
+    address = tezos.public_key_hash(call_data.sender_key)
+    call_data = schemas.UnsignedCall(sender_address=address,
+                                     operations=call_data.operations)
+    return await post_operation(call_data, db)

src/schemas.py

@@ -88,8 +88,15 @@ class ContractCreation(ContractBase):
   credit_id: UUID4
 
 # Operations
-# TODO: right now the sender isn't checked, as we use permits anyway
-class CallData(BaseModel):
-    sender: str
+class UnsignedCall(BaseModel):
+    """Data sent when posting an operation. The sender is mandatory."""
+    sender_address: str
     operations: list[dict[str, Any]]
 
+
+class SignedCall(BaseModel):
+    """Data sent when posting an operation. The signature"""
+    sender_key: str
+    operations: list[dict[str, Any]]
+    signature: str
+    micheline_type: Any

src/tezos.py

@@ -123,16 +123,17 @@ def get_public_key(address):
     return key
 
 
-def check_signature(pair_data, signature, public_key):
-    # Type of a withdraw operation
-    pair_type = {
-        "prim": 'pair',
-        "args": [
-            {"prim": 'string'},
-            {"prim": "int"},
-            {"prim": 'mutez'}
-        ]
-    }
+def check_signature(pair_data, signature, public_key, pair_type=None):
+    if pair_type is None:
+        # Type of a withdraw operation
+        pair_type = {
+            "prim": 'pair',
+            "args": [
+                {"prim": 'string'},
+                {"prim": "int"},
+                {"prim": 'mutez'}
+            ]
+        }
     public_key = pytezos.Key.from_encoded_key(public_key)
     matcher = MichelsonType.match(pair_type)
     packed_pair = matcher.from_micheline_value(pair_data).pack()
@@ -143,6 +144,11 @@ def check_signature(pair_data, signature, public_key):
         return False
 
 
+def public_key_hash(public_key: str):
+    key = pytezos.Key.from_encoded_key(public_key)
+    return key.public_key_hash()
+
+
 async def withdraw(tezos_manager, to, amount):
     op = ptz.transaction(source=ptz.key.public_key_hash(),
                          destination=to,
@@ -160,6 +166,7 @@ def __init__(self, ptz):
     # Receive an operation from sender and add it to the waiting queue;
     # blocks until there is a result in self.results
     async def queue_operation(self, sender, operation):
+        print(operation)
         self.results[sender] = "waiting"
         self.ops_queue[sender] = operation
         while self.results[sender] == "waiting":