feat: add endpoint for verifying Ed25519 signatures and returning user info

Author: IonesioJunior

.pre-commit-config.yaml

@@ -36,4 +36,5 @@ repos:
           sqlalchemy,
           alembic,
           types-sqlalchemy,
+          types-markdown,
         ]

pyproject.toml

@@ -21,6 +21,7 @@ requires-python = ">=3.9"
 dependencies = [
     "alembic>=1.14.0",
     "argon2-cffi>=25.1.0",
+    "cryptography>=43.0.0",
     "email-validator>=2.3.0",
     "fastapi>=0.121.1",
     "httpx>=0.28.1",

src/syfthub/api/endpoints/users.py

@@ -12,7 +12,12 @@
     fake_users_db,
     get_current_active_user,
 )
-from syfthub.schemas.auth import UserRole
+from syfthub.auth.security import verify_ed25519_signature
+from syfthub.schemas.auth import (
+    SignatureVerificationRequest,
+    SignatureVerificationResponse,
+    UserRole,
+)
 from syfthub.schemas.user import User, UserResponse, UserUpdate
 
 router = APIRouter()
@@ -170,3 +175,52 @@ async def delete_user(
 
     # Delete user
     del fake_users_db[user_id]
+
+
+@router.post("/verify-signature", response_model=SignatureVerificationResponse)
+async def verify_signature(
+    request: SignatureVerificationRequest,
+) -> SignatureVerificationResponse:
+    """Verify an Ed25519 signature and return user information if valid.
+
+    This endpoint allows other services to verify that a signature was created
+    by a valid Syfthub user, enabling Syfthub to act as an identity provider.
+    """
+    # Convert message to bytes for verification
+    message_bytes = request.message.encode("utf-8")
+
+    # Find user by public key
+    user_owner = None
+    for user in fake_users_db.values():
+        if user.public_key == request.public_key and user.is_active:
+            user_owner = user
+            break
+
+    if not user_owner:
+        return SignatureVerificationResponse(
+            verified=False,
+            user_info=None,
+            message="Public key not found or user inactive",
+        )
+
+    # Verify the signature
+    is_valid = verify_ed25519_signature(
+        message_bytes, request.signature, request.public_key
+    )
+
+    if not is_valid:
+        return SignatureVerificationResponse(
+            verified=False, user_info=None, message="Invalid signature"
+        )
+
+    # Return user information (minimal data for privacy)
+    user_info: dict[str, str | int | None] = {
+        "id": user_owner.id,
+        "username": user_owner.username,
+        "full_name": user_owner.full_name,
+        "key_created_at": user_owner.key_created_at.isoformat(),
+    }
+
+    return SignatureVerificationResponse(
+        verified=True, user_info=user_info, message="Signature verified successfully"
+    )

src/syfthub/auth/router.py

@@ -23,14 +23,17 @@
     blacklist_token,
     create_access_token,
     create_refresh_token,
+    generate_ed25519_key_pair,
     hash_password,
     verify_password,
     verify_token,
 )
 from syfthub.schemas.auth import (
-    AuthResponse,
+    Ed25519KeyPair,
+    KeyRegenerationResponse,
     PasswordChange,
     RefreshTokenRequest,
+    RegistrationResponse,
     Token,
     UserRegister,
     UserRole,
@@ -44,9 +47,11 @@
 
 
 @router.post(
-    "/register", response_model=AuthResponse, status_code=status.HTTP_201_CREATED
+    "/register",
+    response_model=RegistrationResponse,
+    status_code=status.HTTP_201_CREATED,
 )
-async def register_user(user_data: UserRegister) -> AuthResponse:
+async def register_user(user_data: UserRegister) -> RegistrationResponse:
     """Register a new user."""
     global user_id_counter
 
@@ -67,6 +72,9 @@ async def register_user(user_data: UserRegister) -> AuthResponse:
     # Create new user
     password_hash = hash_password(user_data.password)
 
+    # Generate Ed25519 key pair for the user
+    key_pair = generate_ed25519_key_pair()
+
     user = User(
         id=user_id_counter,
         username=user_data.username,
@@ -75,9 +83,11 @@ async def register_user(user_data: UserRegister) -> AuthResponse:
         age=user_data.age,
         role=UserRole.USER,  # Default role for new users
         password_hash=password_hash,
+        public_key=key_pair.public_key,
         is_active=True,
         created_at=datetime.now(timezone.utc),
         updated_at=datetime.now(timezone.utc),
+        key_created_at=datetime.now(timezone.utc),
     )
 
     # Store user in database
@@ -103,11 +113,18 @@ async def register_user(user_data: UserRegister) -> AuthResponse:
         "is_active": user.is_active,
     }
 
-    return AuthResponse(
+    # Create key pair response
+    keys = Ed25519KeyPair(
+        private_key=key_pair.private_key,
+        public_key=key_pair.public_key,
+    )
+
+    return RegistrationResponse(
         user=user_dict,
         access_token=access_token,
         refresh_token=refresh_token,
         token_type="bearer",
+        keys=keys,
     )
 
 
@@ -225,6 +242,31 @@ async def change_password(
     fake_users_db[current_user.id] = current_user
 
 
+@router.post("/regenerate-keys", response_model=KeyRegenerationResponse)
+async def regenerate_user_keys(
+    current_user: Annotated[User, Depends(get_current_active_user)],
+) -> KeyRegenerationResponse:
+    """Regenerate Ed25519 key pair for the current user."""
+    # Generate new key pair
+    key_pair = generate_ed25519_key_pair()
+
+    # Update user's public key in database
+    current_user.public_key = key_pair.public_key
+    current_user.key_created_at = datetime.now(timezone.utc)
+    current_user.updated_at = datetime.now(timezone.utc)
+
+    # Save to database
+    fake_users_db[current_user.id] = current_user
+
+    # Create key pair response
+    keys = Ed25519KeyPair(
+        private_key=key_pair.private_key,
+        public_key=key_pair.public_key,
+    )
+
+    return KeyRegenerationResponse(keys=keys)
+
+
 def authenticate_user(username: str, password: str) -> User | None:
     """Authenticate a user by username/email and password."""
     # Try to find user by username or email

src/syfthub/auth/security.py

@@ -2,10 +2,21 @@
 
 from __future__ import annotations
 
+import base64
 from datetime import datetime, timedelta, timezone
 from typing import Any
 
 import jwt  # type: ignore[import-not-found]
+from cryptography.hazmat.primitives.asymmetric.ed25519 import (
+    Ed25519PrivateKey,
+    Ed25519PublicKey,
+)
+from cryptography.hazmat.primitives.serialization import (
+    Encoding,
+    NoEncryption,
+    PrivateFormat,
+    PublicFormat,
+)
 from passlib.context import CryptContext  # type: ignore[import-untyped]
 
 from syfthub.core.config import settings
@@ -114,3 +125,93 @@ def get_token_from_header(authorization: str) -> str | None:
     if not authorization or not authorization.startswith("Bearer "):
         return None
     return authorization.replace("Bearer ", "")
+
+
+class Ed25519KeyPair:
+    """Container for Ed25519 key pair."""
+
+    def __init__(self, private_key: str, public_key: str) -> None:
+        """Initialize key pair with base64 encoded keys."""
+        self.private_key = private_key
+        self.public_key = public_key
+
+
+def generate_ed25519_key_pair() -> Ed25519KeyPair:
+    """Generate a new Ed25519 key pair.
+
+    Returns:
+        Ed25519KeyPair with base64 encoded private and public keys.
+    """
+    # Generate private key
+    private_key = Ed25519PrivateKey.generate()
+
+    # Get public key from private key
+    public_key = private_key.public_key()
+
+    # Serialize private key to bytes
+    private_key_bytes = private_key.private_bytes(
+        encoding=Encoding.Raw,
+        format=PrivateFormat.Raw,
+        encryption_algorithm=NoEncryption(),
+    )
+
+    # Serialize public key to bytes
+    public_key_bytes = public_key.public_bytes(
+        encoding=Encoding.Raw, format=PublicFormat.Raw
+    )
+
+    # Encode to base64 for storage/transmission
+    private_key_b64 = base64.b64encode(private_key_bytes).decode("utf-8")
+    public_key_b64 = base64.b64encode(public_key_bytes).decode("utf-8")
+
+    return Ed25519KeyPair(private_key_b64, public_key_b64)
+
+
+def verify_ed25519_signature(message: bytes, signature: str, public_key: str) -> bool:
+    """Verify an Ed25519 signature.
+
+    Args:
+        message: The original message that was signed (as bytes)
+        signature: Base64 encoded signature
+        public_key: Base64 encoded public key
+
+    Returns:
+        True if signature is valid, False otherwise.
+    """
+    try:
+        # Decode base64 inputs
+        signature_bytes = base64.b64decode(signature)
+        public_key_bytes = base64.b64decode(public_key)
+
+        # Reconstruct public key object
+        public_key_obj = Ed25519PublicKey.from_public_bytes(public_key_bytes)
+
+        # Verify signature
+        public_key_obj.verify(signature_bytes, message)
+        return True
+    except Exception:
+        # Any exception means verification failed
+        return False
+
+
+def sign_message_ed25519(message: bytes, private_key: str) -> str:
+    """Sign a message with Ed25519 private key.
+
+    Args:
+        message: The message to sign (as bytes)
+        private_key: Base64 encoded private key
+
+    Returns:
+        Base64 encoded signature.
+    """
+    # Decode private key
+    private_key_bytes = base64.b64decode(private_key)
+
+    # Reconstruct private key object
+    private_key_obj = Ed25519PrivateKey.from_private_bytes(private_key_bytes)
+
+    # Sign message
+    signature_bytes = private_key_obj.sign(message)
+
+    # Return base64 encoded signature
+    return base64.b64encode(signature_bytes).decode("utf-8")

src/syfthub/database/models.py

@@ -34,6 +34,9 @@ class UserModel(Base):
     age: Mapped[int | None] = mapped_column(Integer, nullable=True)
     role: Mapped[str] = mapped_column(String(20), nullable=False, default="user")
     password_hash: Mapped[str] = mapped_column(String(255), nullable=False)
+    public_key: Mapped[str] = mapped_column(
+        String(88), nullable=False
+    )  # Base64 encoded Ed25519 public key
     is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True),
@@ -46,6 +49,11 @@ class UserModel(Base):
         default=lambda: datetime.now(timezone.utc),
         onupdate=lambda: datetime.now(timezone.utc),
     )
+    key_created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        nullable=False,
+        default=lambda: datetime.now(timezone.utc),
+    )
 
     # Relationships
     datasites: Mapped[list[DatasiteModel]] = relationship(
@@ -56,6 +64,7 @@ class UserModel(Base):
     __table_args__ = (
         Index("idx_users_username", "username"),
         Index("idx_users_email", "email"),
+        Index("idx_users_public_key", "public_key", unique=True),
         Index("idx_users_role", "role"),
         Index("idx_users_is_active", "is_active"),
     )

src/syfthub/database/repositories.py

@@ -110,9 +110,11 @@ def _model_to_schema(user_model: UserModel) -> User:
             age=user_model.age,
             role=UserRole(user_model.role),
             password_hash=user_model.password_hash,
+            public_key=user_model.public_key,
             is_active=user_model.is_active,
             created_at=user_model.created_at,
             updated_at=user_model.updated_at,
+            key_created_at=user_model.key_created_at,
         )
 
 

src/syfthub/main.py

@@ -7,7 +7,7 @@
 from pathlib import Path
 from typing import TYPE_CHECKING
 
-import markdown  # type: ignore
+import markdown
 from fastapi import Depends, FastAPI, HTTPException, Query, Request, status
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.templating import Jinja2Templates
@@ -250,7 +250,7 @@ async def list_owner_public_datasites(
     return [DatasitePublicResponse.model_validate(ds) for ds in accessible_datasites]
 
 
-@app.get("/{owner_slug}/{datasite_slug}")
+@app.get("/{owner_slug}/{datasite_slug}", response_model=None)
 async def get_owner_datasite(
     request: Request,
     owner_slug: str,