adding base project

2026-10-25 23:11:19 +05:30
commit 5993c15b70
23 changed files with 1297 additions and 0 deletions
@@ -0,0 +1,41 @@
+import os
+from typing import List, Union
+from pydantic_settings import BaseSettings
+
+class Settings(BaseSettings):
+    PROJECT_NAME: str = "AuthSec FastAPI"
+    API_V1_STR: str = "/api"
+    PORT: int = 9119
+    DEBUG: bool = True
+    
+    # Security
+    SECRET_KEY: str = "cns-secret-key-change-this-in-production-authsec"
+    ALGORITHM: str = "HS256"
+    ACCESS_TOKEN_EXPIRE_MINUTES: int = 1440  # 24 hours
+    
+    # Database
+    DATABASE_URL: str = "sqlite:///./authsec.db"
+    
+    # CORS
+    CORS_ORIGINS: str = "http://localhost:4200,http://localhost:3000,http://127.0.0.1:4200,http://localhost:8080,*"
+
+    # Email / SMTP Settings (mirrors Spring Boot JavaMailSender)
+    SMTP_HOST: str = ""
+    SMTP_PORT: int = 587
+    SMTP_USER: str = ""
+    SMTP_PASSWORD: str = ""
+    SMTP_TLS: bool = True
+    EMAILS_FROM_EMAIL: str = "noreply@authsec.com"
+
+    @property
+    def cors_origins_list(self) -> List[str]:
+        if not self.CORS_ORIGINS:
+            return ["*"]
+        return [origin.strip() for origin in self.CORS_ORIGINS.split(",") if origin.strip()]
+
+    class Config:
+        env_file = ".env"
+        env_file_encoding = "utf-8"
+        case_sensitive = True
+
+settings = Settings()
@@ -0,0 +1,56 @@
+import re
+from urllib.parse import urlparse
+from sqlalchemy import create_engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+from app.core.config import settings
+
+def ensure_database_exists(db_url: str):
+    """
+    If using MySQL, automatically creates the target database if it does not exist
+    (mirrors spring.datasource.url createDatabaseIfNotExist=true).
+    """
+    if db_url.startswith("mysql"):
+        try:
+            import pymysql
+            # Parse connection URL safely
+            clean_url = db_url.replace("mysql+pymysql://", "http://").replace("mysql://", "http://")
+            parsed = urlparse(clean_url)
+            db_name = parsed.path.lstrip("/").split("?")[0]
+            if db_name:
+                conn = pymysql.connect(
+                    host=parsed.hostname,
+                    port=parsed.port or 3306,
+                    user=parsed.username,
+                    password=parsed.password,
+                    connect_timeout=5
+                )
+                with conn.cursor() as cur:
+                    cur.execute(f"CREATE DATABASE IF NOT EXISTS `{db_name}`;")
+                conn.close()
+        except Exception as e:
+            print(f"Note on DB creation: {e}")
+
+ensure_database_exists(settings.DATABASE_URL)
+
+connect_args = {}
+if settings.DATABASE_URL.startswith("sqlite"):
+    connect_args = {"check_same_thread": False}
+
+engine = create_engine(
+    settings.DATABASE_URL,
+    connect_args=connect_args,
+    pool_pre_ping=True
+)
+
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+Base = declarative_base()
+
+def get_db():
+    """Dependency for obtaining database session per request."""
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
@@ -0,0 +1,34 @@
+from typing import Generator, Optional
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from sqlalchemy.orm import Session
+from app.core.database import SessionLocal
+from app.core.security import decode_access_token
+
+security = HTTPBearer(auto_error=False)
+
+def get_db() -> Generator:
+    """Dependency that yields a database session and ensures closure."""
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+def get_current_user_token(credentials: Optional[HTTPAuthorizationCredentials] = Depends(security)) -> Optional[dict]:
+    """Extracts and verifies JWT token payload from Authorization header."""
+    if not credentials:
+        return None
+    token = credentials.credentials
+    payload = decode_access_token(token)
+    return payload
+
+def require_current_user(payload: Optional[dict] = Depends(get_current_user_token)) -> dict:
+    """Ensures request is authenticated with a valid token."""
+    if not payload:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Could not validate credentials or token expired",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    return payload
@@ -0,0 +1,98 @@
+import smtplib
+import logging
+from email.mime.text import MIMEText
+from email.mime.multipart import MIMEMultipart
+from typing import Optional, List
+from app.core.config import settings
+
+logger = logging.getLogger(__name__)
+
+class EmailService:
+    """
+    Email Service mirroring Spring Boot's JavaMailSender / EmailService.
+    Supports SMTP (Gmail, SendGrid, AWS SES, etc.).
+    If SMTP is not configured, logs the message to console (Dev Mode) so flows never crash.
+    """
+
+    def __init__(self):
+        self.host = getattr(settings, "SMTP_HOST", "")
+        self.port = getattr(settings, "SMTP_PORT", 587)
+        self.user = getattr(settings, "SMTP_USER", "")
+        self.password = getattr(settings, "SMTP_PASSWORD", "")
+        self.tls = getattr(settings, "SMTP_TLS", True)
+        self.from_email = getattr(settings, "EMAILS_FROM_EMAIL", "noreply@authsec.com")
+
+    def is_configured(self) -> bool:
+        """Returns True if valid SMTP credentials are provided."""
+        return bool(self.host and self.user and self.password)
+
+    def send_simple_message(self, to: str, subject: str, text: str, from_email: Optional[str] = None) -> bool:
+        """Sends a plain text email (mirrors Spring Boot EmailService.sendSimpleMessage)."""
+        sender = from_email or self.from_email
+
+        if not self.is_configured():
+            print(f"\n==================== [DEV EMAIL SERVICE] ====================")
+            print(f"To:      {to}")
+            print(f"From:    {sender}")
+            print(f"Subject: {subject}")
+            print(f"Body:\n{text}")
+            print(f"=============================================================\n")
+            logger.info(f"[DEV EMAIL] Sent simulated email to {to}: {subject}")
+            return True
+
+        try:
+            msg = MIMEMultipart()
+            msg["From"] = sender
+            msg["To"] = to
+            msg["Subject"] = subject
+            msg.attach(MIMEText(text, "plain"))
+
+            with smtplib.SMTP(self.host, self.port) as server:
+                if self.tls:
+                    server.starttls()
+                server.login(self.user, self.password)
+                server.send_message(msg)
+
+            logger.info(f"Email successfully sent to {to}")
+            return True
+        except Exception as e:
+            logger.error(f"Failed to send email to {to}: {e}")
+            print(f"❌ Email sending error: {e}")
+            return False
+
+    def send_html_message(self, to: str, subject: str, html_content: str, from_email: Optional[str] = None) -> bool:
+        """Sends an HTML formatted email."""
+        sender = from_email or self.from_email
+
+        if not self.is_configured():
+            print(f"\n==================== [DEV EMAIL SERVICE (HTML)] ====================")
+            print(f"To:      {to}")
+            print(f"Subject: {subject}")
+            print(f"HTML Body:\n{html_content}")
+            print(f"===================================================================\n")
+            return True
+
+        try:
+            msg = MIMEMultipart("alternative")
+            msg["From"] = sender
+            msg["To"] = to
+            msg["Subject"] = subject
+            msg.attach(MIMEText(html_content, "html"))
+
+            with smtplib.SMTP(self.host, self.port) as server:
+                if self.tls:
+                    server.starttls()
+                server.login(self.user, self.password)
+                server.send_message(msg)
+
+            return True
+        except Exception as e:
+            logger.error(f"Failed to send HTML email to {to}: {e}")
+            return False
+
+    def adduserviaadmin(self, to: str, subject: str, otp_or_url: str):
+        """Mirrors Spring Boot EmailService.adduserviaadmin."""
+        body = f"Your verification code is: {otp_or_url}"
+        return self.send_simple_message(to, subject, body)
+
+email_service = EmailService()
@@ -0,0 +1,58 @@
+import hashlib
+import secrets
+from datetime import datetime, timedelta
+from typing import Optional, Any, Union
+import jwt
+from app.core.config import settings
+
+def get_password_hash(password: str) -> str:
+    """
+    Hashes a password with PBKDF2-HMAC-SHA256 using Python's standard library.
+    Requires NO C compilers or external dependencies.
+    """
+    salt = secrets.token_hex(16)
+    key = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt.encode("utf-8"), 100000)
+    return f"pbkdf2:sha256:100000${salt}${key.hex()}"
+
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    """Verifies a plain password against the stored hash."""
+    if not hashed_password:
+        return False
+    if "$" not in hashed_password:
+        # Fallback comparison for plain text
+        return plain_password == hashed_password
+    try:
+        parts = hashed_password.split("$")
+        if len(parts) == 3 and parts[0].startswith("pbkdf2:"):
+            _, salt, key_hex = parts
+            check_key = hashlib.pbkdf2_hmac("sha256", plain_password.encode("utf-8"), salt.encode("utf-8"), 100000)
+            return secrets.compare_digest(check_key.hex(), key_hex)
+        return plain_password == hashed_password
+    except Exception:
+        return False
+
+def create_access_token(subject: Union[str, Any], expires_delta: Optional[timedelta] = None, extra_claims: Optional[dict] = None) -> str:
+    """Creates a signed JWT token."""
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
+    
+    to_encode = {
+        "exp": expire,
+        "sub": str(subject),
+        "user_id": subject
+    }
+    if extra_claims:
+        to_encode.update(extra_claims)
+        
+    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
+    return encoded_jwt
+
+def decode_access_token(token: str) -> Optional[dict]:
+    """Decodes a JWT token."""
+    try:
+        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
+        return payload
+    except Exception:
+        return None