headplane/remote-access/auth.py

"""
Authentication and authorization for Heady remote access
Integrates with Headplane's OIDC system
"""
import httpx
import os
from typing import Optional, List
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
import jwt
from jwt.exceptions import InvalidTokenError

# Configuration
HEADPLANE_API_URL = os.getenv("HEADPLANE_API_URL", "http://headplane:3000")
JWT_SECRET = os.getenv("JWT_SECRET", "your-secret-key")  # Should match Headplane
JWT_ALGORITHM = os.getenv("JWT_ALGORITHM", "HS256")

security = HTTPBearer()


class User(BaseModel):
    """User model from Headplane session"""
    id: str
    email: str
    preferred_username: str
    groups: List[str]
    role: str  # From OIDC role mapping


class TerminalPermissions(BaseModel):
    """Terminal access permissions based on OIDC role"""
    ssh: bool = False
    rdp: bool = False
    vnc: bool = False
    file_transfer: bool = False
    session_recording: bool = True
    connection_sharing: bool = False
    admin_nodes: bool = False


def get_terminal_permissions(user_groups: List[str]) -> TerminalPermissions:
    """
    Get terminal access permissions based on OIDC groups
    Uses the same role mapping logic as Headplane
    """
    from app.server.web.roles import mapOidcGroupsToRole
    
    role = mapOidcGroupsToRole(user_groups)
    
    permissions_map = {
        'owner': TerminalPermissions(
            ssh=True,
            rdp=True, 
            vnc=True,
            file_transfer=True,
            session_recording=False,  # Owners don't need to be recorded
            connection_sharing=True,
            admin_nodes=True
        ),
        'admin': TerminalPermissions(
            ssh=True,
            rdp=True,
            vnc=True, 
            file_transfer=True,
            session_recording=True,   # Record admin sessions
            connection_sharing=True,
            admin_nodes=True
        ),
        'network_admin': TerminalPermissions(
            ssh=True,
            rdp=False,
            vnc=False,
            file_transfer=True,
            session_recording=True,
            connection_sharing=False,
            admin_nodes=False  # Only access to regular nodes
        ),
        'it_admin': TerminalPermissions(
            ssh=True,
            rdp=True,   # IT needs RDP for Windows support
            vnc=False,
            file_transfer=True,
            session_recording=True,
            connection_sharing=False,
            admin_nodes=False
        ),
        'auditor': TerminalPermissions(
            ssh=False,  # Read-only access
            rdp=False,
            vnc=False,
            file_transfer=False,
            session_recording=False,
            connection_sharing=False,
            admin_nodes=False
        ),
        'member': TerminalPermissions()  # No access by default
    }
    
    return permissions_map.get(role, permissions_map['member'])


async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)) -> User:
    """
    Verify JWT token from Headplane
    """
    try:
        # Decode JWT token
        payload = jwt.decode(
            credentials.credentials,
            JWT_SECRET,
            algorithms=[JWT_ALGORITHM]
        )
        
        user_id = payload.get("sub")
        if user_id is None:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid token"
            )
            
        # Extract user information
        user = User(
            id=user_id,
            email=payload.get("email", ""),
            preferred_username=payload.get("preferred_username", ""),
            groups=payload.get("groups", []),
            role=payload.get("role", "member")
        )
        
        return user
        
    except InvalidTokenError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token"
        )


async def get_current_user(user: User = Depends(verify_token)) -> User:
    """
    Get current authenticated user
    """
    return user


async def check_terminal_access(
    user: User,
    node_name: str,
    protocol: str = "ssh"
) -> bool:
    """
    Check if user has access to terminal on specific node
    """
    permissions = get_terminal_permissions(user.groups)
    
    # Check protocol permissions
    protocol_allowed = getattr(permissions, protocol, False)
    if not protocol_allowed:
        return False
    
    # TODO: Add node-specific access control
    # This would integrate with Headplane's node management
    
    # For now, allow access if protocol is permitted
    return True


async def validate_node_access(user: User, node_name: str) -> bool:
    """
    Validate user has access to specific Tailscale node
    Queries Headplane API for node permissions
    """
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{HEADPLANE_API_URL}/api/nodes/{node_name}/access",
                headers={"Authorization": f"Bearer {user.id}"}  # Use appropriate auth
            )
            
            if response.status_code == 200:
                data = response.json()
                return data.get("has_access", False)
            else:
                return False
                
    except Exception:
        # Fail closed - deny access on API errors
        return False