Sinasezza

chatApp/config/auth.py

@@ -0,0 +1,150 @@
+# auth.py
+from collections.abc import Mapping
+from datetime import UTC, datetime, timedelta
+from typing import Any, Optional
+
+from fastapi import Depends
+from fastapi.security import OAuth2PasswordBearer
+from jose import JWTError, jwt
+from motor.motor_asyncio import AsyncIOMotorCollection
+from passlib.context import CryptContext
+
+from chatApp.config.config import get_settings
+from chatApp.config.database import mongo_db
+from chatApp.config.logs import logger
+from chatApp.models.user import User
+from chatApp.utils.exceptions import credentials_exception
+
+settings = get_settings()
+
+# Password hashing context
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+# OAuth2 scheme
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/token")
+
+# JWT settings
+SECRET_KEY = settings.jwt_secret_key.get_secret_value()
+ALGORITHM = settings.jwt_algorithm
+ACCESS_TOKEN_EXPIRE_MINUTES = settings.access_token_expire_minutes
+
+
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    """
+    Verify if the provided password matches the stored hashed password.
+
+    :param plain_password: The plain text password.
+    :param hashed_password: The hashed password stored in the database.
+    :return: True if passwords match, otherwise False.
+    """
+    return pwd_context.verify(plain_password, hashed_password)
+
+
+def get_password_hash(password: str) -> str:
+    """
+    Hash the given password using the password hashing context.
+
+    :param password: The plain text password to hash.
+    :return: The hashed password.
+    """
+    return pwd_context.hash(password)
+
+
+def create_access_token(
+    data: dict[str, Any], expires_delta: Optional[timedelta] = None
+) -> str:
+    """
+    Create a JWT access token with a specified expiration.
+
+    :param data: The data to encode into the token.
+    :param expires_delta: Optional expiration time delta for the token.
+    :return: The encoded JWT token as a string.
+    """
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.now(UTC) + expires_delta
+    else:
+        expire = datetime.now(UTC) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    to_encode.update({"exp": expire})
+
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
+
+def parse_access_token(token: str) -> dict[str, Any]:
+    """
+    Parse and validate the given JWT token, returning its payload.
+
+    :param token: The JWT token to parse.
+    :return: The payload data from the token.
+    :raises credentials_exception: If the token is invalid or cannot be decoded.
+    """
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        return payload
+    except JWTError as e:
+        logger.error(f"JWT error: {e}")  # Log the error for debugging purposes
+        raise credentials_exception
+
+
+def get_users_collection() -> AsyncIOMotorCollection:
+    """
+    Retrieve the users collection from the MongoDB database.
+
+    :return: The users collection instance.
+    :raises RuntimeError: If the users collection is not initialized.
+    """
+    users_collection = mongo_db.users_collection
+    if users_collection is None:
+        raise RuntimeError("Users collection is not initialized.")
+    return users_collection
+
+
+async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
+    """
+    Retrieve the current user from the database using the provided JWT token.
+
+    :param token: The JWT token used for authentication.
+    :return: The User object representing the authenticated user.
+    :raises credentials_exception: If the user cannot be found or the token is invalid.
+    """
+    # Parse the token to get the payload
+    payload = parse_access_token(token)
+    username: Optional[str] = payload.get("sub")
+
+    if username is None:
+        logger.error("Username is missing in the token payload.")
+        raise credentials_exception
+
+    # Fetch the users_collection within the request scope
+    users_collection = get_users_collection()
+
+    # Properly type the result of the find_one query
+    user: Optional[Mapping[str, Any]] = await users_collection.find_one(
+        {"username": username}
+    )
+
+    # Raise an exception if no user was found
+    if user is None:
+        logger.error(f"User with username {username} not found in database.")
+        raise credentials_exception
+
+    # Construct and return a User instance from the found document
+    return User(**user)
+
+
+async def authenticate_user(username: str, password: str) -> Optional[User]:
+    # Fetch the users_collection within the request scope
+    users_collection = get_users_collection()
+
+    # Properly type the result of the find_one query
+    user: Optional[Mapping[str, Any]] = await users_collection.find_one(
+        {"username": username}
+    )
+
+    # Return None if no user was found or if password verification fails
+    if user is None or not verify_password(password, user["hashed_password"]):
+        return None
+
+    # Construct and return a User instance from the found document
+    return User(**user)

chatApp/config/config.py

@@ -2,33 +2,51 @@
 from pathlib import Path
 
 from dotenv import load_dotenv
-from pydantic import Field
+from pydantic import Field, SecretStr
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
-BASE_DIR = Path(__file__).resolve().parent.parent
+BASE_DIR = Path(__file__).resolve().parent.parent.parent
 load_dotenv(BASE_DIR / ".env")
 
 
 class Settings(BaseSettings):
     debug: bool = Field(default=False)
-    database_url: str = "mongodb://localhost:27017"
-    database_name: str = "chat_app"
-    jwt_secret_key: str = "your-secret-key"
+
+    # database settings
+    database_url: str = Field(default="mongodb://localhost:27017")
+    database_name: str = Field(default="chat_app")
+    max_pool_size: int = 10
+    min_pool_size: int = 1
+
+    # jwt settings
+    jwt_secret_key: SecretStr = Field(default="your-secret-key")
     jwt_algorithm: str = Field(default="HS256")
+    access_token_expire_minutes: int = Field(default=1440)
 
     # CORS settings
     cors_allow_origins: list[str] = Field(default=["*"])
     cors_allow_credentials: bool = Field(default=True)
     cors_allow_methods: list[str] = Field(default=["*"])
     cors_allow_headers: list[str] = Field(default=["*"])
 
+    # logs settings
+    log_level: str = Field(default="INFO")
+    log_file_path: Path = Field(default=BASE_DIR / "logs/app.log")
+    log_max_bytes: int = Field(default=1048576)  # 1 MB
+    log_backup_count: int = Field(default=3)
+
+    # upload settings
+    upload_dir: Path = Field(default=BASE_DIR / "uploads")
+    max_upload_size: int = Field(default=(5 * 1024 * 1024))
+
     model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")
 
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.log_file_path.parent.mkdir(parents=True, exist_ok=True)
+        self.upload_dir.mkdir(parents=True, exist_ok=True)
+
 
 @lru_cache
 def get_settings() -> Settings:
     return Settings()
-
-
-UPLOAD_DIR = BASE_DIR / "uploads"
-MAX_FILE_SIZE = 5 * 1024 * 1024  # 5 MB

chatApp/config/database.py

@@ -1,23 +1,55 @@
-from motor.motor_asyncio import AsyncIOMotorClient
+import logging
+from typing import Optional
+
+from motor.motor_asyncio import (
+    AsyncIOMotorClient,
+    AsyncIOMotorCollection,
+    AsyncIOMotorDatabase,
+)
 
 from .config import get_settings
 
+logger = logging.getLogger(__name__)
 settings = get_settings()
 
-# Global variable to hold the database client
-db_client = None
-db = None
 
+class MongoDB:
+    def __init__(self) -> None:
+        self.db_client: Optional[AsyncIOMotorClient] = None
+        self.db: Optional[AsyncIOMotorDatabase] = None
+        self.users_collection: Optional[AsyncIOMotorCollection] = None
+        self.messages_collection: Optional[AsyncIOMotorCollection] = None
+        self.rooms_collection: Optional[AsyncIOMotorCollection] = None
+
+    async def connect_to_mongodb(self) -> None:
+        try:
+            self.db_client = AsyncIOMotorClient(
+                settings.database_url,
+                maxPoolSize=settings.max_pool_size,
+                minPoolSize=settings.min_pool_size,
+            )
+
+            assert self.db_client is not None
+            self.db = self.db_client[settings.database_name]
+            assert self.db is not None
+
+            # Initialize collections
+            self.users_collection = self.db.get_collection("users")
+            self.messages_collection = self.db.get_collection("messages")
+            self.rooms_collection = self.db.get_collection("rooms")
+
+            # Ping the server to validate the connection
+            await self.db_client.admin.command("ismaster")
+            logger.info("Connected to MongoDB")
+        except Exception as e:
+            logger.error(f"Could not connect to MongoDB: {e}")
+            raise
 
-async def connect_to_mongodb():
-    global db_client, db
-    db_client = AsyncIOMotorClient(settings.database_url, maxPoolSize=10, minPoolSize=1)
-    db = db_client[settings.database_name]
-    print("Connected to MongoDB")
+    async def close_mongodb_connection(self) -> None:
+        if self.db_client:
+            self.db_client.close()
+            logger.info("Closed MongoDB connection")
 
 
-async def close_mongodb_connection():
-    global db_client
-    if db_client:
-        db_client.close()
-        print("Closed MongoDB connection")
+# Create a global instance of MongoDB
+mongo_db = MongoDB()

chatApp/config/logs.py

@@ -0,0 +1,47 @@
+import logging
+from logging import Logger
+from logging.handlers import RotatingFileHandler
+from pathlib import Path
+
+from .config import BASE_DIR, get_settings
+
+settings = get_settings()
+
+# Ensure the log directory exists
+log_path = Path(BASE_DIR / settings.log_file_path)
+try:
+    log_path.parent.mkdir(parents=True, exist_ok=True)
+except Exception as e:
+    print(f"Error creating log directory: {e}")
+    raise
+
+# Define logging format
+LOG_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+
+# Configure the root logger
+try:
+    logging.basicConfig(
+        level=settings.log_level.upper(),  # Ensure it's uppercase
+        format=LOG_FORMAT,
+        handlers=[
+            logging.StreamHandler(),  # Log to console
+            RotatingFileHandler(
+                log_path,
+                maxBytes=settings.log_max_bytes,
+                backupCount=settings.log_backup_count,
+            ),  # Log to file with rotation
+        ],
+    )
+except Exception as e:
+    print(f"Error setting up logging: {e}")
+    raise
+
+# Get a logger instance
+logger: Logger = logging.getLogger(__name__)
+
+# Example log message to test configuration
+logger.info("Logging configuration is set up.")
+
+
+def get_logger(name: str) -> Logger:
+    return logging.getLogger(name)