Spaces:

mominah
/

Chatbot-backend

Running

App Files Files Community

mominah commited on Mar 7

Commit

c760cc8

verified ·

1 Parent(s): c359dc4

Update auth.py

Browse files

Files changed (1) hide show

auth.py +155 -13

auth.py CHANGED Viewed

@@ -1,12 +1,16 @@
 import os
 import uuid
 import logging
 from datetime import datetime, timedelta
 from urllib.parse import quote_plus
 from typing import List, Optional, Any
 from dotenv import load_dotenv
-from fastapi import APIRouter, HTTPException, Depends, Request, UploadFile, File, Form
 from fastapi.responses import JSONResponse
 from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
 from jose import JWTError, jwt
@@ -15,6 +19,12 @@ from pydantic import BaseModel, EmailStr, Field, validator
 from pymongo import MongoClient
 import os.path
 load_dotenv()
 # Setup logging
@@ -34,7 +44,7 @@ oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 # Create an APIRouter instance
 router = APIRouter()
-# Pydantic models
 class User(BaseModel):
     name: str = Field(..., min_length=3, max_length=50)
     email: EmailStr
@@ -86,7 +96,12 @@ class LoginResponse(Token):
 class TokenData(BaseModel):
     email: Optional[str] = None
-# Password hashing
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 def verify_password(plain_password: str, hashed_password: str) -> bool:
@@ -95,6 +110,7 @@ def verify_password(plain_password: str, hashed_password: str) -> bool:
 def get_password_hash(password: str) -> str:
     return pwd_context.hash(password)
 def get_user(email: str) -> Optional[dict]:
     return users_collection.find_one({"email": email})
@@ -132,7 +148,7 @@ def get_current_user(token: str = Depends(oauth2_scheme)) -> dict:
     except JWTError:
         raise HTTPException(status_code=401, detail="Invalid token")
-# Setup for avatar file saving
 AVATAR_DIR = "avatars"
 if not os.path.exists(AVATAR_DIR):
     os.makedirs(AVATAR_DIR)
@@ -160,7 +176,67 @@ def save_avatar_file(file: UploadFile) -> str:
         file.file.close()
     return file_path
-# ----- Auth Endpoints -----
 @router.post("/signup", response_model=Token)
 async def signup(
@@ -196,21 +272,87 @@ async def signup(
         "token_type": "bearer"
     }
-@router.post("/login", response_model=LoginResponse)
-async def login(request: Request, form_data: OAuth2PasswordRequestForm = Depends()):
-    user = authenticate_user(form_data.username, form_data.password)
     if not user:
-        logger.warning(f"Failed login attempt for: {form_data.username}")
-        raise HTTPException(status_code=401, detail="Incorrect username or password")
-    logger.info(f"User logged in: {user['email']}")
     return {
-        "access_token": create_access_token(user["email"]),
-        "refresh_token": create_refresh_token(user["email"]),
         "token_type": "bearer",
         "name": user["name"],
         "avatar": user.get("avatar")
     }
 @router.get("/user/data")
 async def get_user_data(request: Request, current_user: dict = Depends(get_current_user)):
     return {

 import os
 import uuid
 import logging
+import secrets
+import smtplib
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
 from datetime import datetime, timedelta
 from urllib.parse import quote_plus
 from typing import List, Optional, Any
 from dotenv import load_dotenv
+from fastapi import APIRouter, HTTPException, Depends, Request, UploadFile, File, Form, Body
 from fastapi.responses import JSONResponse
 from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
 from jose import JWTError, jwt
 from pymongo import MongoClient
 import os.path
+# Optionally import markdown to convert markdown text to HTML
+try:
+    import markdown
+except ImportError:
+    markdown = None
 load_dotenv()
 # Setup logging
 # Create an APIRouter instance
 router = APIRouter()
+# ------------------ Pydantic Models ------------------
 class User(BaseModel):
     name: str = Field(..., min_length=3, max_length=50)
     email: EmailStr
 class TokenData(BaseModel):
     email: Optional[str] = None
+# Model for OTP-based login
+class OTPLogin(BaseModel):
+    email: EmailStr
+    otp: str
+# ------------------ Password Hashing ------------------
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 def verify_password(plain_password: str, hashed_password: str) -> bool:
 def get_password_hash(password: str) -> str:
     return pwd_context.hash(password)
+# ------------------ User & Auth Helpers ------------------
 def get_user(email: str) -> Optional[dict]:
     return users_collection.find_one({"email": email})
     except JWTError:
         raise HTTPException(status_code=401, detail="Invalid token")
+# ------------------ Avatar File Saving ------------------
 AVATAR_DIR = "avatars"
 if not os.path.exists(AVATAR_DIR):
     os.makedirs(AVATAR_DIR)
         file.file.close()
     return file_path
+# ------------------ OTP and Email Functions ------------------
+def generate_otp(length=6):
+    """Generate a numeric OTP of specified length."""
+    digits = "0123456789"
+    otp = ''.join(secrets.choice(digits) for _ in range(length))
+    return otp
+def send_email_func(sender_email, sender_password, receiver_email, subject, body):
+    smtp_server = "smtp.gmail.com"
+    port = 587
+    message = MIMEMultipart("alternative")
+    message["Subject"] = subject
+    message["From"] = sender_email
+    message["To"] = receiver_email
+    part1 = MIMEText(body, "plain")
+    if markdown:
+        html_content = markdown.markdown(body)
+    else:
+        html_content = f"<pre>{body}</pre>"
+    html_template = f"""\
+<html>
+  <head>
+    <style>
+      body {{
+        font-family: Arial, sans-serif;
+        line-height: 1.6;
+        margin: 20px;
+      }}
+      h1, h2, h3 {{
+        color: #333;
+      }}
+      pre {{
+        background: #f4f4f4;
+        padding: 10px;
+        border: 1px solid #ddd;
+      }}
+    </style>
+  </head>
+  <body>
+    {html_content}
+  </body>
+</html>
+"""
+    part2 = MIMEText(html_template, "html")
+    message.attach(part1)
+    message.attach(part2)
+    try:
+        with smtplib.SMTP(smtp_server, port, timeout=10) as server:
+            server.starttls()
+            server.login(sender_email, sender_password)
+            server.sendmail(sender_email, receiver_email, message.as_string())
+        logger.info("Successfully sent email")
+    except Exception as e:
+        logger.error(f"Error sending email: {e}")
+        raise e
+# -------------------------------------------------------------
+# ------------------ Auth Endpoints ------------------
 @router.post("/signup", response_model=Token)
 async def signup(
         "token_type": "bearer"
     }
+@router.post("/login_otp", response_model=LoginResponse)
+async def login_otp(otp_data: OTPLogin):
+    """
+    OTP-based login endpoint.
+    The user must supply their email and the OTP received via email.
+    """
+    user = get_user(otp_data.email)
     if not user:
+        logger.warning(f"Login OTP attempt for non-existent email: {otp_data.email}")
+        raise HTTPException(status_code=401, detail="User not found")
+    # Check if OTP exists and is valid
+    stored_otp = user.get("otp")
+    otp_expires = user.get("otp_expires")
+    if not stored_otp or not otp_expires:
+        raise HTTPException(status_code=400, detail="OTP not generated. Please request a new OTP.")
+    # Check if OTP is expired (using ISO format stored in DB)
+    if datetime.utcnow() > datetime.fromisoformat(otp_expires):
+        raise HTTPException(status_code=400, detail="OTP has expired. Please request a new OTP.")
+    if otp_data.otp != stored_otp:
+        raise HTTPException(status_code=401, detail="Invalid OTP.")
+    # OTP is valid. Remove OTP fields.
+    users_collection.update_one(
+        {"email": otp_data.email},
+        {"$unset": {"otp": "", "otp_expires": ""}}
+    )
+    logger.info(f"User logged in with OTP: {otp_data.email}")
     return {
+        "access_token": create_access_token(otp_data.email),
+        "refresh_token": create_refresh_token(otp_data.email),
         "token_type": "bearer",
         "name": user["name"],
         "avatar": user.get("avatar")
     }
+@router.post("/send_otp")
+async def send_otp(receiver_email: EmailStr = Body(..., embed=True)):
+    """
+    Generates an OTP, stores it in the user's document with an expiration,
+    and sends it via email to the provided receiver_email.
+    (For demonstration purposes, the OTP is returned. Remove in production.)
+    """
+    user = get_user(receiver_email)
+    if not user:
+        raise HTTPException(status_code=404, detail="User not found. Please sign up first.")
+    sender_email = os.getenv("SENDER_EMAIL")
+    sender_password = os.getenv("SENDER_PASSWORD")
+    if not sender_email or not sender_password:
+        raise HTTPException(status_code=500, detail="Email sender credentials are not configured.")
+    otp = generate_otp(6)
+    # Set OTP to expire in 5 minutes
+    otp_expires = (datetime.utcnow() + timedelta(minutes=1)).isoformat()
+    subject = "Your One-Time Password (OTP)"
+    body = f"""\
+# Your OTP
+Here is your one-time password (OTP):
+`{otp}`
+Please use the code above to proceed with your login.
+"""
+    try:
+        send_email_func(sender_email, sender_password, receiver_email, subject, body)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error sending OTP email: {str(e)}")
+    # Update user document with OTP and expiration
+    users_collection.update_one(
+        {"email": receiver_email},
+        {"$set": {"otp": otp, "otp_expires": otp_expires}}
+    )
+    # For demonstration, return the OTP (remove in production)
+    return {"message": "OTP sent successfully", "otp": otp}
 @router.get("/user/data")
 async def get_user_data(request: Request, current_user: dict = Depends(get_current_user)):
     return {