PMFlow/backend/app/services/auth/service.py

"""
Authentication & Authorization Service

Provides password hashing, JWT token management, and permission checking.
"""

from datetime import datetime, timedelta
from typing import Optional
import hashlib
import hmac
import os

# Try to import passlib, fallback to simple hashing
from ...utils.timezone import now_cn_aware, now_cn_naive
try:
    from passlib.context import CryptContext
    from jose import JWTError, jwt
    HAS_PASSLIB = True
except ImportError:
    HAS_PASSLIB = False
    # Fallback JWT implementation
    try:
        import jwt as jwt_lib
        jwt = jwt_lib
        JWTError = Exception
    except ImportError:
        jwt = None
        JWTError = Exception

# JWT Configuration
SECRET_KEY = os.getenv("SECRET_KEY", "neural-sync-secret-key-change-in-production")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_HOURS = 24
MAX_BCRYPT_PASSWORD_BYTES = 72

# Password hashing context (if passlib available)
if HAS_PASSLIB:
    pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


class AuthService:
    """Service for authentication operations"""

    @staticmethod
    def password_utf8_bytes(password: str) -> int:
        return len(str(password or "").encode("utf-8"))

    @staticmethod
    def validate_password(password: str) -> None:
        if len(str(password or "")) < 6:
            raise ValueError("密码至少需要 6 个字符")
        if AuthService.password_utf8_bytes(password) > MAX_BCRYPT_PASSWORD_BYTES:
            raise ValueError("密码最长支持 72 个 UTF-8 字节，请缩短后重试")

    @staticmethod
    def verify_password(plain_password: str, hashed_password: str) -> bool:
        """Verify a plain password against a hashed password"""
        if HAS_PASSLIB:
            try:
                # 优先走 bcrypt（新数据）
                if pwd_context.verify(plain_password, hashed_password):
                    return True
            except Exception:
                # 旧库可能是历史 SHA256 哈希，回退兼容
                pass

            return hmac.compare_digest(
                AuthService._simple_hash(plain_password),
                hashed_password,
            )

        # Simple SHA256 fallback for development
        return hmac.compare_digest(
            AuthService._simple_hash(plain_password),
            hashed_password
        )

    @staticmethod
    def hash_password(password: str) -> str:
        """Hash a password for storage"""
        AuthService.validate_password(password)
        if HAS_PASSLIB:
            try:
                return pwd_context.hash(password)
            except ValueError as exc:
                if "72 bytes" in str(exc).lower() and AuthService.password_utf8_bytes(password) > MAX_BCRYPT_PASSWORD_BYTES:
                    raise ValueError("密码最长支持 72 个 UTF-8 字节，请缩短后重试") from exc
                return AuthService._simple_hash(password)
            except Exception:
                return AuthService._simple_hash(password)
        else:
            # Simple SHA256 fallback for development
            return AuthService._simple_hash(password)

    @staticmethod
    def _simple_hash(password: str) -> str:
        """Simple SHA256 hash with salt for development use"""
        salt = SECRET_KEY[:16].encode() if SECRET_KEY else b'neural-sync-salt'
        return hashlib.sha256(password.encode() + salt).hexdigest()

    @staticmethod
    def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
        """Create a JWT access token"""
        if jwt is None:
            # Fallback: return a simple token (not recommended for production)
            import base64
            import json
            token_data = {
                **data,
                "exp": (now_cn_naive() + (expires_delta or timedelta(hours=ACCESS_TOKEN_EXPIRE_HOURS))).timestamp()
            }
            return base64.b64encode(json.dumps(token_data).encode()).decode()

        to_encode = data.copy()
        if expires_delta:
            expire = now_cn_naive() + expires_delta
        else:
            expire = now_cn_naive() + timedelta(hours=ACCESS_TOKEN_EXPIRE_HOURS)
        to_encode.update({"exp": expire})
        return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

    @staticmethod
    def verify_token(token: str) -> Optional[str]:
        """
        Verify a JWT token and return the username

        Returns None if token is invalid or expired
        """
        if jwt is None:
            # Fallback: decode simple token
            import base64
            import json
            try:
                token_data = json.loads(base64.b64decode(token).decode())
                if token_data.get("exp", 0) > now_cn_aware().timestamp():
                    return token_data.get("sub")
                return None
            except:
                return None

        try:
            payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
            username: str = payload.get("sub")
            return username
        except JWTError:
            return None


# Permission matrix for role-based access control
PERMISSION_MATRIX = {
    "BOSS": [
        "view_all_signals",
        "assign_tasks",
        "manage_users",
        "delete_users",
        "manage_domains",
        "view_projects",
        "view_context",
        "view_settings",
        "manage_staff",
    ],
    "ADMIN": [
        "view_all_signals",
        "assign_tasks",
        "manage_domains",
        "view_projects",
        "view_context",
        "view_settings",
        "manage_staff",
    ],
    "AI": [
        "view_assigned_signals",
        "create_experiments",
        "view_context",
    ],
    "FE": [
        "view_assigned_signals",
        "view_assigned_projects",
    ],
    "BE": [
        "view_assigned_signals",
        "view_assigned_projects",
    ],
}


def has_permission(user_role: str, permission: str) -> bool:
    """Check if a role has a specific permission"""
    roles = parse_role_values(user_role)
    return any(permission in PERMISSION_MATRIX.get(role, []) for role in roles)


def has_any_role(user_role: str, allowed_roles: list) -> bool:
    """Check if user_role is in the list of allowed roles"""
    allowed = {str(role).upper() for role in allowed_roles}
    return any(role in allowed for role in parse_role_values(user_role))


def parse_role_values(user_role: str | None) -> list[str]:
    """Parse single/multi role string into uppercase role tokens."""
    raw = str(user_role or "")
    if not raw:
        return []
    values: list[str] = []
    for token in raw.split(","):
        role = token.strip().upper()
        if role and role not in values:
            values.append(role)
    return values


def can_manage_users(user_role: str) -> bool:
    """Check if user can manage other users (BOSS only)"""
    return has_any_role(user_role, ["BOSS"])


def can_assign_tasks(user_role: str) -> bool:
    """Check if user can assign tasks to others"""
    return has_any_role(user_role, ["BOSS", "ADMIN"])