Add admin panel

- Add `/admin/` enpoint
- Add `Contract` support as an example
- Add basic authentication
- Closes #15

app/config.py

@@ -3,6 +3,7 @@
 """
 
 import os
+import secrets
 
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
@@ -19,6 +20,11 @@ class Settings(BaseSettings):
     DATABASE_POOL_CLASS: str = "AsyncAdaptedQueuePool"
     DATABASE_POOL_SIZE: int = 10
     TEST: bool = False
+    SECRET_KEY: str = secrets.token_urlsafe(
+        32
+    )  # In production it must be defined so it doesn't change
+    ADMIN_USERNAME: str = "admin"
+    ADMIN_PASSWORD: str = "admin"
 
 
 settings = Settings()

app/main.py

@@ -1,7 +1,7 @@
 from fastapi import APIRouter, FastAPI
 
 from . import VERSION
-from .routers import about, contracts, default
+from .routers import about, admin, contracts, default
 
 app = FastAPI(
     title="Safe Decoder Service",
@@ -11,6 +11,8 @@
     redoc_url=None,
 )
 
+admin.load_admin(app)
+
 # Router configuration
 api_v1_router = APIRouter(
     prefix="/api/v1",

app/routers/admin.py

@@ -0,0 +1,63 @@
+import secrets
+
+from fastapi import FastAPI
+
+from sqladmin import Admin, ModelView
+from sqladmin.authentication import AuthenticationBackend
+from starlette.requests import Request
+
+from ..cache import get_redis
+from ..config import settings
+from ..datasources.db.database import get_engine
+from ..datasources.db.models import Contract
+
+
+class AdminAuth(AuthenticationBackend):
+    async def login(self, request: Request) -> bool:
+        form = await request.form()
+        username, password = form["username"], form["password"]
+
+        # Validate username/password credentials
+        if username == settings.ADMIN_USERNAME and settings.ADMIN_PASSWORD:
+            # And update session
+            secret = secrets.token_hex(nbytes=16)
+            request.session.update({"token": secret})
+
+            # Use redis to store the token
+            get_redis().set(
+                f"admin:token:{secret}", 1, ex=7 * 24 * 60 * 60
+            )  # Expire in one week
+
+            return True
+        return False
+
+    async def logout(self, request: Request) -> bool:
+        # Usually you'd want to just clear the session
+        request.session.clear()
+        return True
+
+    async def authenticate(self, request: Request) -> bool:
+        secret = request.session.get("token")
+
+        if not secret:
+            return False
+
+        return bool(get_redis().exists(f"admin:token:{secret}"))
+
+
+class ContractAdmin(ModelView, model=Contract):
+    column_list = [Contract.address, Contract.name, Contract.description]  # type: ignore
+    form_include_pk = True
+    icon = "fa-solid fa-file-contract"
+
+    async def on_model_change(
+        self, data: dict, model: Contract, is_created: bool, request: Request
+    ) -> None:
+        data["address"] = bytes.fromhex(data["address"].strip().replace("0x", ""))
+        return await super().on_model_change(data, model, is_created, request)
+
+
+def load_admin(app: FastAPI):
+    authentication_backend = AdminAuth(secret_key=settings.SECRET_KEY)
+    admin = Admin(app, get_engine(), authentication_backend=authentication_backend)
+    admin.add_view(ContractAdmin)

app/tests/routers/test_admin.py

@@ -0,0 +1,18 @@
+import unittest
+
+from fastapi.testclient import TestClient
+
+from ...main import app
+
+
+class TestRouterAdmin(unittest.TestCase):
+    client: TestClient
+
+    @classmethod
+    def setUpClass(cls):
+        cls.client = TestClient(app)
+
+    def test_admin(self):
+        response = self.client.get("/admin")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("DOCTYPE html", response.text)