claude-os/mcp_server/auth.py

"""
Simple authentication for Claude OS.
Uses environment variables for credentials - no database needed.
"""

import os
import secrets
from datetime import datetime, timedelta
from typing import Optional
from fastapi import HTTPException, Security, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt
from passlib.context import CryptContext

# Password hashing
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# JWT settings
SECRET_KEY = os.getenv("CLAUDE_OS_SECRET_KEY", secrets.token_urlsafe(32))
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 * 7  # 7 days

# Security
# auto_error=False: let get_current_user itself decide whether a missing
# token is OK (auth not configured -> guest) or a 401 (auth configured).
# The HTTPBearer default (auto_error=True) rejects any request with no
# Authorization header before get_current_user's body ever runs, which
# breaks the "auth disabled by default" behavior entirely.
security = HTTPBearer(auto_error=False)


def verify_password(plain_password: str, hashed_password: str) -> bool:
    """Verify a password against its hash."""
    return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password: str) -> str:
    """Hash a password."""
    return pwd_context.hash(password)


def get_user_credentials() -> dict:
    """
    Get user credentials from environment variables.

    Set in .env or environment:
        CLAUDE_OS_EMAIL=admin@example.com
        CLAUDE_OS_PASSWORD_HASH=<bcrypt hash>

    To generate a password hash:
        python3 -c "from passlib.context import CryptContext; print(CryptContext(schemes=['bcrypt']).hash('your_password'))"
    """
    email = os.getenv("CLAUDE_OS_EMAIL")
    password_hash = os.getenv("CLAUDE_OS_PASSWORD_HASH")

    # For development, allow plain password (will be hashed on the fly)
    password_plain = os.getenv("CLAUDE_OS_PASSWORD")

    if not email:
        # Default to disabled auth if not configured
        return None

    if password_hash:
        return {"email": email, "password_hash": password_hash}
    elif password_plain:
        # Hash the plain password (for development only)
        return {"email": email, "password_hash": get_password_hash(password_plain)}
    else:
        return None


def authenticate_user(email: str, password: str) -> bool:
    """Authenticate a user by email and password."""
    creds = get_user_credentials()

    if not creds:
        # Auth disabled
        return False

    if email != creds["email"]:
        return False

    return verify_password(password, creds["password_hash"])


def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
    """Create a JWT access token."""
    to_encode = data.copy()

    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + 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 decode_access_token(token: str) -> Optional[dict]:
    """Decode and validate a JWT access token."""
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except jwt.PyJWTError:
        return None


def get_current_user(credentials: Optional[HTTPAuthorizationCredentials] = Security(security)) -> dict:
    """
    Dependency to get the current authenticated user.
    Use in FastAPI routes: current_user = Depends(get_current_user)
    """
    creds = get_user_credentials()

    # If auth not configured, allow all requests
    if not creds:
        return {"email": "guest", "auth_disabled": True}

    if credentials is None:
        raise HTTPException(
            status_code=401,
            detail="Not authenticated",
            headers={"WWW-Authenticate": "Bearer"},
        )

    token = credentials.credentials

    # Service token: lets trusted local integrations (e.g. the code-forge
    # stdio MCP server) call the API without a user login. Only relevant
    # once auth is actually enabled - with no CLAUDE_OS_EMAIL set, everyone
    # already gets the guest bypass above regardless of this.
    service_token = os.getenv("CLAUDE_OS_SERVICE_TOKEN")
    if service_token and secrets.compare_digest(token, service_token):
        return {"email": "service", "service_account": True}

    payload = decode_access_token(token)

    if payload is None:
        raise HTTPException(
            status_code=401,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )

    email = payload.get("sub")
    if email is None:
        raise HTTPException(
            status_code=401,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )

    return {"email": email}


def is_auth_enabled() -> bool:
    """Check if authentication is enabled."""
    return get_user_credentials() is not None