session token test

services.py

@@ -2,10 +2,17 @@ from fastapi import FastAPI, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel
 from typing import List
-import json
+import jwt
+import datetime
+from dotenv import load_dotenv
+import os
 
+load_dotenv()
+
+# FastAPI app setup
 app = FastAPI()
 
+# CORS setup to allow cross-origin requests
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],  # Allow all origins; use specific domains for security
@@ -14,16 +21,46 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
+# Secret key for JWT encoding/decoding
+SECRET_KEY = os.getenv("SECRET_KEY", "secret")  # Use a more secure key in production
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 30  # Token expiration time in minutes
+
 # Temporary user dictionary for storing credentials
 USERS = {
     "user1": "password1",
     "user2": "password2",
 }
 
+# Pydantic model to define the structure of user data
 class User(BaseModel):
     username: str
     password: str
 
+# JWT token generation function
+def create_access_token(data: dict, expires_delta: datetime.timedelta = None):
+    """Create a new JWT token."""
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.datetime.utcnow() + datetime.timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
+# JWT token verification function
+def verify_access_token(token: str):
+    """Verify and decode the JWT token."""
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        return payload
+    except jwt.ExpiredSignatureError:
+        raise HTTPException(status_code=401, detail="Token has expired")
+    except jwt.JWTError:
+        raise HTTPException(status_code=401, detail="Invalid token")
+
+# Register a new user
 @app.post("/register")
 async def register_user(user: User):
     """Register a new user."""
@@ -32,24 +69,33 @@ async def register_user(user: User):
     USERS[user.username] = user.password
     return {"message": "User registered successfully"}
 
+# Login a user and generate a JWT session token
 @app.post("/login")
 async def login_user(user: User):
-    """Authenticate a user."""
+    """Authenticate a user and return JWT token."""
     if user.username not in USERS or USERS[user.username] != user.password:
         raise HTTPException(status_code=401, detail="Invalid credentials")
-    return {"message": "Login successful"}
+    # Create JWT token
+    access_token = create_access_token(data={"sub": user.username})
+    return {"access_token": access_token}
+
+# API to validate the session token
+@app.get("/api/validate-token")
+async def validate_token(token: str):
+    """Validate the JWT token."""
+    payload = verify_access_token(token)
+    return {"message": "Token is valid", "username": payload.get("sub")}
 
+# Search for users by username
 @app.get("/search")
 async def search_users(query: str) -> List[str]:
     """Search for users by username."""
     matching_users = [username for username in USERS if query.lower() in username.lower()]
-    
     if not matching_users:
         raise HTTPException(status_code=404, detail="No users found matching the query")
-    
     return matching_users
 
-# New /api/ routes
+# New /api/ routes for CRUD operations
 @app.get("/api/users")
 async def get_all_users() -> List[str]:
     """Get a list of all users."""