saas_backend/app/config/security.py

from datetime import datetime, timedelta, timezone
from typing import Optional, Dict, Any
import bcrypt
import jwt
from fastapi import HTTPException, status
import re
import secrets
import string
from app.config.settings import settings

class SecurityUtils:
    """Security utility class for authentication and authorization."""
    
    @staticmethod
    def hash_password(password: str) -> str:
        """Hash a password using bcrypt."""
        salt = bcrypt.gensalt(rounds=settings.BCRYPT_ROUNDS)
        return bcrypt.hashpw(password.encode('utf-8'), salt).decode('utf-8')
    
    @staticmethod
    def verify_password(plain_password: str, hashed_password: str) -> bool:
        """Verify a password against its hash."""
        return bcrypt.checkpw(plain_password.encode('utf-8'), hashed_password.encode('utf-8'))
    
    @staticmethod
    def generate_access_token(data: Dict[str, Any], tenant_id: Optional[Any] = None) -> str:
        """Generate JWT access token."""
        to_encode = data.copy()
        expire = datetime.now(timezone.utc) + timedelta(seconds=settings.ACCESS_TOKEN_EXPIRES)
        to_encode.update({"exp": expire, "type": "access"})
        
        # Include tenant_id if provided
        if tenant_id:
            to_encode["tenant_id"] = str(tenant_id)
        
        return jwt.encode(
            to_encode,
            settings.ACCESS_TOKEN_SECRET,
            algorithm="HS256"
        )
    
    @staticmethod
    def generate_refresh_token(data: Dict[str, Any], tenant_id: Optional[Any] = None) -> str:
        """Generate JWT refresh token."""
        to_encode = data.copy()
        expire = datetime.now(timezone.utc) + timedelta(seconds=settings.REFRESH_TOKEN_EXPIRES)
        to_encode.update({"exp": expire, "type": "refresh"})
        
        # Include tenant_id if provided
        if tenant_id:
            to_encode["tenant_id"] = str(tenant_id)
        
        return jwt.encode(
            to_encode,
            settings.REFRESH_TOKEN_SECRET,
            algorithm="HS256"
        )
    
    @staticmethod
    def verify_access_token(token: str) -> Dict[str, Any]:
        """Verify and decode JWT access token."""
        try:
            payload = jwt.decode(
                token,
                settings.ACCESS_TOKEN_SECRET,
                algorithms=["HS256"]
            )
            
            if payload.get("type") != "access":
                raise HTTPException(
                    status_code=status.HTTP_401_UNAUTHORIZED,
                    detail="Invalid token type"
                )
            
            return payload
        except jwt.ExpiredSignatureError:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Access token has expired"
            )
        except jwt.InvalidTokenError:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid access token"
            )
    
    @staticmethod
    def verify_refresh_token(token: str) -> Dict[str, Any]:
        """Verify and decode JWT refresh token."""
        try:
            payload = jwt.decode(
                token,
                settings.REFRESH_TOKEN_SECRET,
                algorithms=["HS256"]
            )
            
            if payload.get("type") != "refresh":
                raise HTTPException(
                    status_code=status.HTTP_401_UNAUTHORIZED,
                    detail="Invalid token type"
                )
            
            return payload
        except jwt.ExpiredSignatureError:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Refresh token has expired"
            )
        except jwt.InvalidTokenError:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid refresh token"
            )
    
    @staticmethod
    def generate_otp(length: int = 6) -> str:
        """Generate a random OTP."""
        return ''.join(secrets.choice(string.digits) for _ in range(length))
    
    @staticmethod
    def validate_password_strength(password: str) -> bool:
        """Validate password strength."""
        if len(password) < 8:
            return False
        
        if not re.search(r'[A-Z]', password):
            return False
        
        if not re.search(r'[a-z]', password):
            return False
        
        if not re.search(r'\d', password):
            return False

        if not re.search(r'[!@#$%^&*(),.?":{}|<>]', password):
            return False

        return True

    @staticmethod
    def generate_module_token(data: Dict[str, Any], module_id: str, ttl_seconds: int = 900) -> str:
        """Generate short-lived module-scoped JWT (15 min default)."""
        to_encode = data.copy()
        expire = datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds)
        to_encode.update({
            "exp": expire,
            "type": "module_access",
            "aud": str(module_id),
            "iat": datetime.now(timezone.utc).timestamp()
        })
        
        if not settings.SAAS_PRIVATE_KEY:
             raise ValueError("SAAS_PRIVATE_KEY is not configured. Cannot sign module identity tokens.")
             
        return jwt.encode(
            to_encode, 
            settings.SAAS_PRIVATE_KEY, 
            algorithm="RS256",
            headers={"kid": settings.SAAS_KEY_ID}
        )

    @staticmethod
    def validate_email(email: str) -> bool:
        """Validate email format."""
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        return re.match(pattern, email) is not None
    
    @staticmethod
    def validate_ip_address(ip: str) -> bool:
        """Validate IP address format (IPv4 and IPv6)."""
        ipv4_pattern = r'^(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)(\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)){3}$'
        ipv6_pattern = r'^(([0-9a-fA-F]{1,4}:){7,7}[0-9a-fA-F]{1,4}|([0-9a-fA-F]{1,4}:){1,7}:|([0-9a-fA-F]{1,4}:){1,6}:[0-9a-fA-F]{1,4}|([0-9a-fA-F]{1,4}:){1,5}(:[0-9a-fA-F]{1,4}){1,2}|([0-9a-fA-F]{1,4}:){1,4}(:[0-9a-fA-F]{1,4}){1,3}|([0-9a-fA-F]{1,4}:){1,3}(:[0-9a-fA-F]{1,4}){1,4}|([0-9a-fA-F]{1,4}:){1,2}(:[0-9a-fA-F]{1,4}){1,5}|[0-9a-fA-F]{1,4}:((:[0-9a-fA-F]{1,4}){1,6})|:((:[0-9a-fA-F]{1,4}){1,7}|:))$'
        
        return re.match(ipv4_pattern, ip) is not None or re.match(ipv6_pattern, ip) is not None

security = SecurityUtils()