Add request message to sign endpoint

.github/workflows/ci.yml

@@ -31,6 +31,16 @@ jobs:
     strategy:
       matrix:
         python-version: ["3.12"]
+    services:
+      redis:
+        image: redis
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+        ports:
+          - 6379:6379
     steps:
     - uses: actions/checkout@v4
     - name: Set up Python ${{ matrix.python-version }}

app/config.py

@@ -14,7 +14,7 @@ class Settings(BaseSettings):
         extra="allow",
         case_sensitive=True,
     )
-    REDIS_URL: str
+    REDIS_URL: str = "redis://"
     NONCE_TTL_SECONDS: int = 60 * 10
 
 

app/models.py

@@ -1,9 +1,31 @@
-from pydantic import BaseModel, constr
+from typing import Optional
+
+from pydantic import AnyUrl, BaseModel, Field, field_validator
+
+from gnosis.eth.utils import fast_is_checksum_address
 
 
 class About(BaseModel):
     version: str
 
 
 class Nonce(BaseModel):
-    nonce: constr(min_length=8, pattern=r"^[A-Za-z0-9]{8,}$")  # noqa: F722
+    nonce: str = Field(min_length=8, pattern=r"^[A-Za-z0-9]{8,}$")
+
+
+class SiweMessageRequest(BaseModel):
+    domain: str = Field(pattern="^[^/?#]+$", examples=["domain.com"])
+    address: str
+    chain_id: int
+    uri: AnyUrl
+    statement: Optional[str] = Field(default=None)
+
+    @field_validator("address")
+    def validate_address(cls, value):
+        if not fast_is_checksum_address(value):
+            raise ValueError("Invalid Ethereum address")
+        return value
+
+
+class SiweMessage(BaseModel):
+    message: str

app/routers/auth.py

@@ -1,6 +1,7 @@
 from fastapi import APIRouter
 
-from ..models import Nonce
+from ..models import Nonce, SiweMessage, SiweMessageRequest
+from ..siwe.message_service import create_siwe_message
 from ..siwe.nonce_repository import get_nonce_repository
 
 router = APIRouter(
@@ -10,5 +11,19 @@
 
 
 @router.get("/nonce", response_model=Nonce)
-async def nonce() -> "Nonce":
+async def get_nonce() -> "Nonce":
     return Nonce(nonce=get_nonce_repository().generate_nonce())
+
+
+@router.post("/messages", response_model=SiweMessage)
+async def request_siwe_message(
+    siwe_message_request: SiweMessageRequest,
+) -> "SiweMessage":
+    siwe_message = create_siwe_message(
+        domain=siwe_message_request.domain,
+        address=siwe_message_request.address,
+        chain_id=siwe_message_request.chain_id,
+        uri=str(siwe_message_request.uri),
+        statement=siwe_message_request.statement,
+    )
+    return SiweMessage(message=siwe_message)

app/siwe/message_service.py

@@ -0,0 +1,30 @@
+from datetime import UTC, datetime, timedelta
+
+from siwe.siwe import ISO8601Datetime, SiweMessage, VersionEnum
+
+from gnosis.eth.utils import fast_to_checksum_address
+
+from ..config import settings
+from .nonce_repository import get_nonce_repository
+
+
+def create_siwe_message(
+    domain: str, address: str, chain_id: int, uri: str, statement=None
+) -> str:
+    nonce = get_nonce_repository().generate_nonce()
+
+    message = SiweMessage(
+        domain=domain,
+        address=fast_to_checksum_address(address),
+        statement=statement or "Welcome to Safe!",
+        uri=uri,
+        version=VersionEnum.one,
+        chain_id=chain_id,
+        nonce=nonce,
+        issued_at=ISO8601Datetime.from_datetime(datetime.now(UTC)),
+        valid_until=ISO8601Datetime.from_datetime(
+            datetime.now(UTC) + timedelta(seconds=settings.NONCE_TTL_SECONDS)
+        ),
+    )
+
+    return message.prepare_message()

app/tests/test_models.py

@@ -0,0 +1,78 @@
+import unittest
+
+from pydantic import ValidationError
+
+from pydantic_core._pydantic_core import Url
+
+from ..models import Nonce, SiweMessageRequest
+
+
+class TestNonce(unittest.TestCase):
+    def test_valid_nonce(self):
+        valid_nonce_data = {"nonce": "abcd1234"}
+        nonce_instance = Nonce(nonce=valid_nonce_data["nonce"])
+        self.assertEqual(nonce_instance.nonce, valid_nonce_data["nonce"])
+
+    def test_invalid_nonce_length(self):
+        with self.assertRaises(ValidationError):
+            Nonce(nonce="abc")
+
+    def test_invalid_nonce_pattern(self):
+        with self.assertRaises(ValidationError):
+            Nonce(nonce="abcd-1234")
+
+
+class TestSiweMessageRequest(unittest.TestCase):
+    def test_valid_siwe_message_request(self):
+        siwe_message_request = SiweMessageRequest(
+            domain="example.com",
+            address="0x32Be343B94f860124dC4fEe278FDCBD38C102D88",
+            chain_id=1,
+            uri=Url("https://example.com/"),
+            statement="Test statement",
+        )
+        self.assertEqual(siwe_message_request.domain, "example.com")
+        self.assertEqual(
+            siwe_message_request.address, "0x32Be343B94f860124dC4fEe278FDCBD38C102D88"
+        )
+        self.assertEqual(siwe_message_request.chain_id, 1)
+        self.assertEqual(str(siwe_message_request.uri), "https://example.com/")
+        self.assertEqual(siwe_message_request.statement, "Test statement")
+
+    def test_invalid_domain(self):
+        with self.assertRaises(ValidationError):
+            SiweMessageRequest(
+                domain="example.com/invalid",
+                address="0x32Be343B94f860124dC4fEe278FDCBD38C102D88",
+                chain_id=1,
+                uri=Url("https://example.com"),
+            )
+
+    def test_invalid_address(self):
+        with self.assertRaises(ValidationError):
+            SiweMessageRequest(
+                domain="example.com",
+                address="0xInvalidEthereumAddress",
+                chain_id=1,
+                uri=Url("https://example.com"),
+                statement="Test statement",
+            )
+
+    def test_invalid_uri(self):
+        with self.assertRaises(ValidationError):
+            SiweMessageRequest(
+                domain="example.com",
+                address="0x32Be343B94f860124dC4fEe278FDCBD38C102D88",
+                chain_id=1,
+                uri=Url("invalid-url"),
+                statement="Test statement",
+            )
+
+    def test_optional_statement(self):
+        siwe_message_request = SiweMessageRequest(
+            domain="example.com",
+            address="0x32Be343B94f860124dC4fEe278FDCBD38C102D88",
+            chain_id=1,
+            uri=Url("https://example.com"),
+        )
+        self.assertIsNone(siwe_message_request.statement)

app/tests/test_router_about.py

@@ -2,12 +2,12 @@
 
 from fastapi.testclient import TestClient
 
-from app import VERSION
-from app.main import app
+from .. import VERSION
+from ..main import app
 
 
 class TestRouterAbout(unittest.TestCase):
-    client = None
+    client: TestClient
 
     @classmethod
     def setUpClass(cls):

app/tests/test_router_auth.py

@@ -1,21 +1,56 @@
 import re
 import unittest
+from unittest import mock
 
 from fastapi.testclient import TestClient
 
-from app.main import app
+import siwe
+
+from ..main import app
 
 
 class TestRouterAuth(unittest.TestCase):
-    client = None
+    client: TestClient
 
     @classmethod
     def setUpClass(cls):
         cls.client = TestClient(app)
 
-    def test_view_about(self):
+    def test_view_get_nonce(self):
         response = self.client.get("/api/v1/auth/nonce")
         self.assertEqual(response.status_code, 200)
         self.assertTrue(
             re.fullmatch(r"^[A-Za-z0-9]{8,}$", response.json().get("nonce"))
         )
+
+    @mock.patch("app.siwe.nonce_repository.NonceRepository.generate_nonce")
+    def test_view_request_siwe_message(self, mock_generate_nonce):
+        test_domain = "example.com"
+        test_address = "0x32Be343B94f860124dC4fEe278FDCBD38C102D88"
+        test_chain_id = 1
+        test_uri = "https://example.com/"
+        test_statement = "Test statement"
+        test_nonce = "testnonce1234"
+
+        mock_generate_nonce.return_value = test_nonce
+
+        payload = {
+            "domain": test_domain,
+            "address": test_address,
+            "chain_id": test_chain_id,
+            "uri": test_uri,
+            "statement": test_statement,
+        }
+
+        response = self.client.post("/api/v1/auth/messages", json=payload)
+
+        self.assertEqual(response.status_code, 200)
+        obtained_siwe_message = siwe.SiweMessage.from_message(
+            response.json().get("message")
+        )
+        self.assertEqual(obtained_siwe_message.domain, test_domain)
+        self.assertEqual(obtained_siwe_message.address, test_address)
+        self.assertEqual(obtained_siwe_message.chain_id, test_chain_id)
+        self.assertEqual(obtained_siwe_message.uri, test_uri)
+        self.assertEqual(obtained_siwe_message.statement, test_statement)
+        self.assertEqual(obtained_siwe_message.nonce, test_nonce)

app/tests/test_router_default.py

@@ -2,11 +2,11 @@
 
 from fastapi.testclient import TestClient
 
-from app.main import app
+from ..main import app
 
 
 class TestRouterDefault(unittest.TestCase):
-    client = None
+    client: TestClient
 
     @classmethod
     def setUpClass(cls):

app/tests/test_siwe_message_service.py

@@ -0,0 +1,84 @@
+import unittest
+from unittest import mock
+
+from siwe.siwe import SiweMessage, VersionEnum
+
+from ..siwe.message_service import create_siwe_message
+
+
+class TestSiweMessageService(unittest.TestCase):
+    @mock.patch("app.siwe.nonce_repository.NonceRepository.generate_nonce")
+    def test_create_siwe_message(self, mock_generate_nonce):
+        test_domain = "example.com"
+        test_address = "0x32Be343B94f860124dC4fEe278FDCBD38C102D88"
+        test_chain_id = 1
+        test_uri = "https://example.com"
+        test_statement = "Test statement"
+        test_nonce = "testnonce1234"
+
+        mock_generate_nonce.return_value = test_nonce
+        with mock.patch("app.config.settings.NONCE_TTL_SECONDS", 100):
+            message_str = create_siwe_message(
+                domain=test_domain,
+                address=test_address,
+                chain_id=test_chain_id,
+                uri=test_uri,
+                statement=test_statement,
+            )
+
+        mock_generate_nonce.assert_called_once()
+
+        siwe_message = SiweMessage.from_message(message_str)
+        issued_at = siwe_message.issued_at
+        valid_until = siwe_message.expiration_time
+
+        expected_message = SiweMessage(
+            domain=test_domain,
+            address=test_address,
+            statement=test_statement,
+            uri=test_uri,
+            version=VersionEnum.one,
+            chain_id=test_chain_id,
+            nonce=test_nonce,
+            issued_at=issued_at,
+            valid_until=valid_until,
+        )
+
+        self.assertEqual(message_str, expected_message.prepare_message())
+
+    @mock.patch("app.siwe.nonce_repository.NonceRepository.generate_nonce")
+    def test_create_siwe_message_without_statement(self, mock_generate_nonce):
+        test_domain = "example.com"
+        test_address = "0x32Be343B94f860124dC4fEe278FDCBD38C102D88"
+        test_chain_id = 1
+        test_uri = "https://example.com"
+        test_nonce = "testnonce1234"
+
+        mock_generate_nonce.return_value = test_nonce
+        with mock.patch("app.config.settings.NONCE_TTL_SECONDS", 100):
+            message_str = create_siwe_message(
+                domain=test_domain,
+                address=test_address,
+                chain_id=test_chain_id,
+                uri=test_uri,
+            )
+
+        mock_generate_nonce.assert_called_once()
+
+        siwe_message = SiweMessage.from_message(message_str)
+        issued_at = siwe_message.issued_at
+        valid_until = siwe_message.expiration_time
+
+        expected_message = SiweMessage(
+            domain=test_domain,
+            address=test_address,
+            statement="Welcome to Safe!",
+            uri=test_uri,
+            version=VersionEnum.one,
+            chain_id=test_chain_id,
+            nonce=test_nonce,
+            issued_at=issued_at,
+            valid_until=valid_until,
+        )
+
+        self.assertEqual(message_str, expected_message.prepare_message())

requirements/prod.txt

@@ -2,4 +2,5 @@ fastapi[all]==0.111.1
 hiredis==3.0.0
 pydantic-settings==2.3.4
 redis[hiredis]==5.0.7
+safe-eth-py==6.0.0b35
 siwe==4.2.0

setup.cfg

@@ -1,7 +1,7 @@
 [flake8]
 max-line-length = 88
 select = C,E,F,W,B,B950
-extend-ignore = E203,E501,F841,W503
+extend-ignore = E203,E501,F841,W503, F722
 exclude = .tox,.git,*/migrations/*,*/static/CACHE/*,docs,node_modules,venv
 
 [pycodestyle]