🚀 Auto-deploy from GitHub

app/api/v1/endpoints/auth.py

@@ -0,0 +1,98 @@
+"""
+Authentication endpoints for token management
+"""
+from fastapi import APIRouter, HTTPException, Depends
+from datetime import datetime, timedelta
+from typing import Dict, Any
+import jwt
+import os
+from ..schemas.auth_schemas import TokenRequest, TokenResponse
+from ....core.config import settings
+
+router = APIRouter()
+
+@router.post("/auth/token", response_model=TokenResponse)
+async def get_access_token(credentials: TokenRequest):
+    """
+    Exchange user credentials for a temporary access token
+    This endpoint allows users to authenticate and receive a JWT token
+    """
+    # Example: validate user credentials against your user database
+    # For now, this is just a placeholder implementation
+    
+    if not validate_user_credentials(credentials.dict()):
+        raise HTTPException(status_code=401, detail="Invalid credentials")
+    
+    # Generate a temporary JWT token
+    payload = {
+        "sub": credentials.username,
+        "exp": datetime.utcnow() + timedelta(hours=24),
+        "iat": datetime.utcnow(),
+        "type": "access_token"
+    }
+    
+    secret_key = os.getenv("SECRET_KEY", "your-secret-key-change-this")
+    token = jwt.encode(payload, secret_key, algorithm="HS256")
+    
+    return TokenResponse(
+        access_token=token,
+        token_type="bearer",
+        expires_in=86400  # 24 hours
+    )
+
+@router.post("/auth/refresh")
+async def refresh_access_token(refresh_token: str):
+    """
+    Refresh an expired access token
+    """
+    try:
+        secret_key = os.getenv("SECRET_KEY", "your-secret-key-change-this")
+        payload = jwt.decode(refresh_token, secret_key, algorithms=["HS256"])
+        
+        if payload.get("type") != "refresh_token":
+            raise HTTPException(status_code=401, detail="Invalid token type")
+        
+        # Generate new access token
+        new_payload = {
+            "sub": payload["sub"],
+            "exp": datetime.utcnow() + timedelta(hours=24),
+            "iat": datetime.utcnow(),
+            "type": "access_token"
+        }
+        
+        new_token = jwt.encode(new_payload, secret_key, algorithm="HS256")
+        
+        return TokenResponse(
+            access_token=new_token,
+            token_type="bearer",
+            expires_in=86400
+        )
+        
+    except jwt.InvalidTokenError:
+        raise HTTPException(status_code=401, detail="Invalid refresh token")
+
+def validate_user_credentials(credentials: Dict[str, Any]) -> bool:
+    """
+    Validate user credentials against your user database
+    
+    TODO: Implement your actual user validation logic here
+    This could check against:
+    - Database users table
+    - LDAP/Active Directory
+    - OAuth provider
+    - etc.
+    """
+    # Placeholder implementation - replace with your actual auth logic
+    username = credentials.get("username")
+    password = credentials.get("password")
+    
+    # Example: simple hardcoded validation (NOT for production!)
+    if username == "demo" and password == "demo123":
+        return True
+    
+    # In a real implementation, you would:
+    # 1. Hash the password and compare with stored hash
+    # 2. Check against your user database
+    # 3. Validate any additional requirements
+    
+    return False

app/api/v1/schemas/auth_schemas.py

@@ -0,0 +1,20 @@
+"""
+Authentication schemas for request/response models
+"""
+from pydantic import BaseModel
+from typing import Optional
+
+class TokenRequest(BaseModel):
+    """Request model for token authentication"""
+    username: str
+    password: str
+
+class TokenResponse(BaseModel):
+    """Response model for token endpoints"""
+    access_token: str
+    token_type: str = "bearer"
+    expires_in: int  # seconds until expiration
+
+class RefreshTokenRequest(BaseModel):
+    """Request model for token refresh"""
+    refresh_token: str

app/core/auth.py

@@ -5,6 +5,8 @@ from fastapi import HTTPException, status, Depends
 from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
 from typing import Optional
 import os
+import jwt
+from datetime import datetime
 from dotenv import load_dotenv
 
 load_dotenv()
@@ -17,7 +19,7 @@ def get_hf_api_key() -> str:
 
 def verify_hf_token(credentials: Optional[HTTPAuthorizationCredentials] = Depends(security)) -> bool:
     """
-    Verify HuggingFace API token
+    Verify HuggingFace API token or JWT token
     Returns True if no authentication is required (public space) or if token is valid
     """
     expected_token = get_hf_api_key()
@@ -34,15 +36,29 @@ def verify_hf_token(credentials: Optional[HTTPAuthorizationCredentials] = Depend
             headers={"WWW-Authenticate": "Bearer"},
         )
     
-    # Verify the token matches
-    if credentials.credentials != expected_token:
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Invalid authentication token",
-            headers={"WWW-Authenticate": "Bearer"},
-        )
+    token = credentials.credentials
+    
+    # Try to verify as HuggingFace API key first
+    if token == expected_token:
+        return True
+    
+    # Try to verify as JWT token (if you implement JWT auth)
+    try:
+        secret_key = os.getenv("SECRET_KEY", "your-secret-key")
+        payload = jwt.decode(token, secret_key, algorithms=["HS256"])
+        # Verify token hasn't expired
+        if payload.get("exp") and datetime.fromtimestamp(payload["exp"]) < datetime.utcnow():
+            raise jwt.ExpiredSignatureError("Token has expired")
+        return True
+    except jwt.InvalidTokenError:
+        pass
     
-    return True
+    # If neither verification method worked, deny access
+    raise HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Invalid authentication token",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
 
 def optional_auth(credentials: Optional[HTTPAuthorizationCredentials] = Depends(security)) -> bool:
     """

app/main.py

@@ -1,7 +1,7 @@
 from fastapi import FastAPI
 from fastapi.staticfiles import StaticFiles
 from fastapi.middleware.cors import CORSMiddleware
-from .api.v1.endpoints import generate, download, health
+from .api.v1.endpoints import generate, download, health, auth
 from .core.config import settings
 from .core.model_loader import get_generator  # Import get_generator
 from contextlib import asynccontextmanager
@@ -95,6 +95,7 @@ app.add_middleware(
 )
 
 app.include_router(health.router, prefix=settings.API_PREFIX, tags=["Health"])
+app.include_router(auth.router, prefix=settings.API_PREFIX, tags=["Authentication"])
 app.include_router(generate.router, prefix=settings.API_PREFIX, tags=["Generate Card"])
 app.include_router(download.router, prefix=settings.API_PREFIX, tags=["Download Card"])