Add EVM wallet signature verification endpoint

Add POST /api/verify/ethereum for EIP-191 personal-sign verification,
mirroring the existing Solana flow. Works with any EVM chain.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: balogh.adam@icloud.com

api/api_types.py

@@ -140,6 +140,12 @@ class SolanaVerifyRequest(BaseModel):
     signature: str
 
 
+class EvmVerifyRequest(BaseModel):
+    address: str
+    message: str
+    signature: str
+
+
 class ProcessSwapRequest(BaseModel):
     txid: str
     chain: str = "solana"

server/fastapi_server.py

@@ -29,6 +29,7 @@
     Message,
     TokenMetadata,
     SolanaVerifyRequest,
+    EvmVerifyRequest,
     Context,
     UserMessage,
     ProcessSwapRequest,
@@ -262,6 +263,25 @@ async def verify_solana_signature(request: Request):
             logging.error(f"Error verifying SIWX signature: {e}")
             raise HTTPException(status_code=500, detail="Internal server error")
 
+    @app.post("/api/verify/ethereum")
+    async def verify_evm_signature(request: Request):
+        try:
+            request_data = await request.json()
+            verify_request = EvmVerifyRequest(**request_data)
+
+            token = await asyncio.to_thread(
+                service.verify_evm_signature, verify_request
+            )
+            return {"token": token}
+
+        except ValidationError as e:
+            raise HTTPException(status_code=400, detail=str(e))
+        except ValueError as e:
+            raise HTTPException(status_code=401, detail=str(e))
+        except Exception as e:
+            logging.error(f"Error verifying EVM signature: {e}")
+            raise HTTPException(status_code=500, detail="Internal server error")
+
     @app.get("/api/healthcheck")
     async def healthcheck():
         return {"status": "ok"}

server/service.py

@@ -1,9 +1,24 @@
 from nacl.signing import VerifyKey
-from api.api_types import SolanaVerifyRequest
+from eth_account.messages import encode_defunct
+from eth_account import Account
+from api.api_types import SolanaVerifyRequest, EvmVerifyRequest
 from base58 import b58decode
 from server.firebase import auth
 
 
+def _firebase_custom_token(uid: str) -> str:
+    custom_token = auth.create_custom_token(uid)
+
+    if isinstance(custom_token, bytes):
+        token_bytes = custom_token
+    elif isinstance(custom_token, str):
+        token_bytes = custom_token.encode("utf-8")
+    else:
+        token_bytes = str(custom_token).encode("utf-8")
+
+    return token_bytes.decode("utf-8")
+
+
 def verify_solana_signature(verify_request: SolanaVerifyRequest) -> str:
     try:
         public_key = b58decode(verify_request.address)
@@ -14,15 +29,19 @@ def verify_solana_signature(verify_request: SolanaVerifyRequest) -> str:
         verify_key.verify(message, signature)
 
         uid = f"wallet_{verify_request.address}"
-        custom_token = auth.create_custom_token(uid)
-
-        if isinstance(custom_token, bytes):
-            token_bytes = custom_token
-        elif isinstance(custom_token, str):
-            token_bytes = custom_token.encode("utf-8")
-        else:
-            token_bytes = str(custom_token).encode("utf-8")
-
-        return token_bytes.decode("utf-8")
+        return _firebase_custom_token(uid)
     except Exception:
         raise
+
+
+def verify_evm_signature(verify_request: EvmVerifyRequest) -> str:
+    claimed_address = verify_request.address.lower()
+
+    signable = encode_defunct(text=verify_request.message)
+    recovered = Account.recover_message(signable, signature=verify_request.signature)
+
+    if recovered.lower() != claimed_address:
+        raise ValueError("Recovered address does not match claimed address")
+
+    uid = f"wallet_{claimed_address}"
+    return _firebase_custom_token(uid)