bootnode/api/bootnode/core/iam.py

"""Hanzo IAM integration for multi-tenant authentication.

Supports login via:
- hanzo.id (Hanzo org)
- zoo.id (Zoo Labs org)
- lux.id (Lux Network org)
- pars.id (Pars org)

Uses standard OAuth2/OIDC flow (RFC 6749, OIDC Core).
"""

import os
from datetime import UTC, datetime
from functools import lru_cache
from typing import Any

import httpx
import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jwt import PyJWKClient
from pydantic import BaseModel

from bootnode.config import get_settings


class IAMUser(BaseModel):
    """User from Hanzo IAM."""

    id: str
    name: str
    email: str
    org: str  # hanzo, zoo, lux, pars
    avatar: str | None = None
    roles: list[str] = []
    permissions: list[str] = []
    created_at: datetime | None = None

    @property
    def is_admin(self) -> bool:
        """Check if user has admin role."""
        return "admin" in self.roles

    @property
    def can_create_projects(self) -> bool:
        """Check if user can create projects."""
        return "projects:write" in self.permissions or self.is_admin


class IAMClient:
    """Client for Hanzo IAM service."""

    def __init__(self):
        self.settings = get_settings()
        self._jwks_client: PyJWKClient | None = None

    @property
    def jwks_client(self) -> PyJWKClient:
        """Get JWKS client (lazy initialization)."""
        if self._jwks_client is None:
            jwks_url = f"{self.settings.iam_url}/v1/iam/.well-known/jwks"
            self._jwks_client = PyJWKClient(jwks_url)
        return self._jwks_client

    # All known IAM app client_ids — tokens from any of these are accepted.
    KNOWN_CLIENT_IDS = {"lux-web3", "pars-cloud", "zoo-cloud", "hanzo-cloud"}

    async def verify_token(self, token: str) -> IAMUser:
        """Verify JWT token from hanzo.id and return user."""
        try:
            # Get signing key from JWKS
            signing_key = self.jwks_client.get_signing_key_from_jwt(token)

            # Accept tokens issued for any of our known IAM apps
            allowed_audiences = self.KNOWN_CLIENT_IDS | {self.settings.iam_client_id}

            # Decode and verify token
            payload = jwt.decode(
                token,
                signing_key.key,
                algorithms=["RS256", "ES256"],
                audience=list(allowed_audiences),
                # Lux IAM currently issues tokens with canonical issuer
                # (for example, https://hanzo.id) even when reached via branded domains.
                options={"verify_iss": False},
            )

            # Extract user info
            return IAMUser(
                id=payload.get("sub"),
                name=payload.get("name", ""),
                email=payload.get("email", ""),
                org=payload.get("org", "hanzo"),
                avatar=payload.get("avatar"),
                roles=payload.get("roles", []),
                permissions=payload.get("permissions", []),
                created_at=datetime.fromtimestamp(
                    payload.get("iat", 0), tz=UTC
                ),
            )

        except jwt.ExpiredSignatureError:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Token has expired",
            )
        except jwt.InvalidTokenError as e:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail=f"Invalid token: {str(e)}",
            )

    async def exchange_code(
        self,
        code: str,
        redirect_uri: str,
        client_id: str | None = None,
    ) -> dict[str, Any]:
        """Exchange authorization code for tokens.

        Args:
            code: The authorization code from the OAuth callback.
            redirect_uri: Must match the redirect_uri used in the authorize request.
            client_id: IAM app client_id. Defaults to settings.iam_client_id.
        """
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.settings.iam_url}/v1/iam/oauth/token",
                data={
                    "grant_type": "authorization_code",
                    "client_id": client_id or self.settings.iam_client_id,
                    "client_secret": self.settings.iam_client_secret,
                    "code": code,
                    "redirect_uri": redirect_uri,
                },
            )

            if response.status_code != 200:
                raise HTTPException(
                    status_code=status.HTTP_401_UNAUTHORIZED,
                    detail="Failed to exchange code for token",
                )

            return response.json()


# Dependency injection
@lru_cache
def get_iam_client() -> IAMClient:
    """Get IAM client singleton."""
    return IAMClient()


# Security scheme
bearer_scheme = HTTPBearer(auto_error=False)


async def get_current_user(
    credentials: HTTPAuthorizationCredentials | None = Depends(bearer_scheme),
    iam: IAMClient = Depends(get_iam_client),
) -> IAMUser:
    """Get current authenticated user from hanzo.id token."""
    if credentials is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Not authenticated",
            headers={"WWW-Authenticate": "Bearer"},
        )

    return await iam.verify_token(credentials.credentials)


async def get_optional_user(
    credentials: HTTPAuthorizationCredentials | None = Depends(bearer_scheme),
    iam: IAMClient = Depends(get_iam_client),
) -> IAMUser | None:
    """Get current user if authenticated, None otherwise."""
    if credentials is None:
        return None

    try:
        return await iam.verify_token(credentials.credentials)
    except HTTPException:
        return None


def require_org(allowed_orgs: list[str]):
    """Dependency that requires user to be from specific org(s)."""
    async def _require_org(user: IAMUser = Depends(get_current_user)) -> IAMUser:
        if user.org not in allowed_orgs:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Access restricted to orgs: {allowed_orgs}",
            )
        return user
    return _require_org


def require_permission(permission: str):
    """Dependency that requires user to have specific permission."""
    async def _require_permission(user: IAMUser = Depends(get_current_user)) -> IAMUser:
        if permission not in user.permissions and not user.is_admin:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Missing permission: {permission}",
            )
        return user
    return _require_permission


def fleet_admin_allowlist() -> set[str]:
    """Bootstrap admin allowlist — comma-separated emails from ``FLEET_ADMIN_EMAILS``.

    Lets the sole initial superuser (e.g. ``[email protected]``) reach admin-only
    infra views before the IAM ``admin`` role is provisioned on their token.
    Empty by default → pure role-based gating. Read live from the environment so
    it can be rotated without a process restart.
    """
    raw = os.environ.get("FLEET_ADMIN_EMAILS", "")
    return {e.strip().lower() for e in raw.split(",") if e.strip()}


async def require_admin(user: IAMUser = Depends(get_current_user)) -> IAMUser:
    """Require an authenticated IAM admin (deny-by-default).

    The server-side gate for internal / cross-tenant infra surfaces (e.g. the
    luxd fleet overview). Enforced in the API — never rely on UI hiding.

      * missing / invalid credentials → 401 (raised by ``get_current_user``)
      * authenticated non-admin       → 403
      * admin                         → passes through

    Admin = the IAM ``admin`` role, or an email in the ``FLEET_ADMIN_EMAILS``
    bootstrap allowlist (the sole initial superuser).
    """
    if user.is_admin or user.email.lower() in fleet_admin_allowlist():
        return user
    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN,
        detail="Admin role required for internal fleet/infra views",
    )