Entra JIT Provisioning with Admin Override Implementation Guide¶

# Locate the User class definition (around line 8-18)
# ADD these columns to the existing model:

from datetime import datetime
from sqlalchemy import Column, Integer, String, Boolean, JSON, DateTime

class User(Base):
    __tablename__ = "app_users"

    id = Column(Integer, primary_key=True, autoincrement=True)
    email = Column(String(255), unique=True, nullable=False, index=True)
    name = Column(String(255))
    password = Column(String(255))  # Nullable for Entra users
    attributes = Column(JSON)  # Legacy, will be deprecated
    active = Column(Boolean, default=True, nullable=False)

    # NEW COLUMNS - Add these:
    scopes = Column(JSON, nullable=False, default=list)  # Runtime source of truth
    auth_provider = Column(String(50), default="local")  # "local" or "azure_ad"
    entra_groups = Column(JSON)  # Last known Entra groups (for audit/reset)
    scopes_overridden = Column(Boolean, default=False, nullable=False)  # Track manual changes
    last_login = Column(DateTime)
    created_at = Column(DateTime, default=datetime.utcnow)

def upgrade() -> None:
    # Should contain operations similar to:
    op.add_column('app_users', sa.Column('scopes', sa.JSON(), nullable=False, server_default='[]'))
    op.add_column('app_users', sa.Column('auth_provider', sa.String(50), nullable=True, server_default='local'))
    op.add_column('app_users', sa.Column('entra_groups', sa.JSON(), nullable=True))
    op.add_column('app_users', sa.Column('scopes_overridden', sa.Boolean(), nullable=False, server_default='false'))
    op.add_column('app_users', sa.Column('last_login', sa.DateTime(), nullable=True))
    op.add_column('app_users', sa.Column('created_at', sa.DateTime(), nullable=True))

    # Add indexes for performance
    op.create_index('idx_app_users_scopes', 'app_users', ['scopes'], postgresql_using='gin')
    op.create_index('idx_app_users_auth_provider', 'app_users', ['auth_provider'])
    op.create_index('idx_app_users_active', 'app_users', ['active'], postgresql_where=sa.text('active = true'))

def downgrade() -> None:
    # Should drop everything added in upgrade()
    op.drop_index('idx_app_users_active', 'app_users')
    op.drop_index('idx_app_users_auth_provider', 'app_users')
    op.drop_index('idx_app_users_scopes', 'app_users')
    op.drop_column('app_users', 'created_at')
    op.drop_column('app_users', 'last_login')
    op.drop_column('app_users', 'scopes_overridden')
    op.drop_column('app_users', 'entra_groups')
    op.drop_column('app_users', 'auth_provider')
    op.drop_column('app_users', 'scopes')

"""
Just-in-Time (JIT) User Provisioning for Entra Authentication

Handles automatic user creation on first login with group-based scope assignment.
"""

import logging
from datetime import datetime
from typing import List, Optional
from sqlalchemy.orm import Session

from app.auth.users.models import User
from app.auth.entra_mappings import get_scopes_for_groups

logger = logging.getLogger(__name__)


def resolve_entra_group_names(group_ids: List[str]) -> List[str]:
    """
    Resolve Entra group IDs to display names via Microsoft Graph API.

    TODO: Implement Graph API integration when needed.
    For now, returns group IDs as-is (if your token includes group names directly).

    Args:
        group_ids: List of Entra group IDs or names from token

    Returns:
        List of resolved group names
    """
    # FUTURE IMPLEMENTATION:
    # from app.auth.graph import get_group_names
    # return get_group_names(group_ids)

    # Current: assumes token contains group names directly
    # If your token contains ObjectIDs instead, implement Graph API call here
    return group_ids


def compute_initial_scopes(entra_groups: List[str]) -> List[str]:
    """
    Compute initial scopes based on Entra group membership.

    Args:
        entra_groups: List of Entra group names

    Returns:
        List of scopes (union of all group scopes)
    """
    scopes = get_scopes_for_groups(entra_groups)
    logger.info(f"Computed initial scopes from groups {entra_groups}: {scopes}")
    return scopes


def provision_new_user(
    db: Session,
    email: str,
    name: str,
    entra_groups: List[str]
) -> User:
    """
    Create new user with group-based scopes.

    Args:
        db: Database session
        email: User email address
        name: User display name
        entra_groups: List of Entra group names

    Returns:
        Newly created User object
    """
    initial_scopes = compute_initial_scopes(entra_groups)

    user = User(
        email=email,
        name=name,
        password=None,  # Entra users don't have local passwords
        scopes=initial_scopes,
        auth_provider="azure_ad",
        entra_groups=entra_groups,
        scopes_overridden=False,
        active=True,
        created_at=datetime.utcnow(),
        last_login=datetime.utcnow()
    )

    db.add(user)
    db.commit()
    db.refresh(user)

    logger.info(
        f"JIT provisioned user {email} with {len(initial_scopes)} scopes "
        f"from groups: {entra_groups}"
    )

    return user


def update_existing_user_login(
    db: Session,
    user: User,
    entra_groups: List[str]
) -> User:
    """
    Update existing user's login timestamp and group membership.

    IMPORTANT: Does NOT update scopes (database is source of truth).
    Only updates audit fields.

    Args:
        db: Database session
        user: Existing User object
        entra_groups: Current Entra group names

    Returns:
        Updated User object
    """
    user.last_login = datetime.utcnow()
    user.entra_groups = entra_groups  # Update for audit purposes

    db.commit()
    db.refresh(user)

    logger.debug(f"Updated login timestamp for user {user.email}")

    return user


def get_or_create_user(
    db: Session,
    email: str,
    name: str,
    entra_group_ids: List[str]
) -> User:
    """
    Main entry point for JIT provisioning.

    - If user exists: Update login timestamp, return user
    - If user is new: Create with group-based scopes

    Args:
        db: Database session
        email: User email from token
        name: User name from token
        entra_group_ids: Group IDs or names from token

    Returns:
        User object (existing or newly created)
    """
    # Check if user exists
    user = db.query(User).filter(User.email == email).first()

    # Resolve group names
    entra_groups = resolve_entra_group_names(entra_group_ids)

    if user:
        # Existing user - just update login info
        return update_existing_user_login(db, user, entra_groups)
    else:
        # New user - JIT provision
        return provision_new_user(db, email, name, entra_groups)

from app.auth.provisioning import get_or_create_user
from app.auth.auth import SessionData
from app.util.database import get_db

@router.get("/auth/azure/callback")
async def azure_callback(
    code: str,
    state: str = Query(None),
    db: Session = Depends(get_db)
):
    """
    Azure AD OAuth callback endpoint.
    Handles JIT user provisioning and session creation.
    """
    try:
        # Exchange authorization code for token
        token_data = await exchange_code_for_token(code)

        # Extract user information from token
        email = token_data.get("email") or token_data.get("preferred_username")
        name = token_data.get("name", email)
        entra_group_ids = token_data.get("groups", [])

        if not email:
            raise HTTPException(400, "Email not found in token")

        # JIT provision or load existing user
        user = get_or_create_user(
            db=db,
            email=email,
            name=name,
            entra_group_ids=entra_group_ids
        )

        if not user.active:
            raise HTTPException(403, "User account is deactivated")

        # Create session with scopes from database
        session_data = SessionData(
            id=user.id,
            email=user.email,
            name=user.name,
            scopes=user.scopes,  # Load from database
            attributes=user.attributes or {}  # Legacy support
        )

        # Store session
        session_id = str(uuid.uuid4())
        await backend.create(session_id, session_data)

        # Create response with session cookie
        redirect_url = state if state else "/"
        response = RedirectResponse(url=redirect_url, status_code=302)
        response.set_cookie(
            key="session_id",
            value=session_id,
            httponly=True,
            secure=settings.ENFORCE_HTTPS,
            samesite="lax"
        )

        logger.info(f"User {email} logged in via Azure AD with {len(user.scopes)} scopes")

        return response

    except Exception as e:
        logger.error(f"Azure AD callback error: {str(e)}", exc_info=True)
        raise HTTPException(500, f"Authentication failed: {str(e)}")

from pydantic import BaseModel
from typing import Optional, List

class SessionData(BaseModel):
    id: Optional[int] = None
    email: Optional[str] = None
    name: Optional[str] = None
    scopes: List[str] = []  # NEW: Add this field
    attributes: Optional[dict] = None  # Legacy

    def has_scope(self, scope: str) -> bool:
        """Check if session has a specific scope."""
        return scope in self.scopes or "*" in self.scopes

    def has_workflow_access(self, workflow: str, action: str = "read") -> bool:
        """Check workflow-specific access with fallback to global."""
        # 1. Try workflow-specific scope
        if self.has_scope(f"workflows:{workflow}:{action}"):
            return True
        # 2. Fall back to global workflow scope
        if self.has_scope(f"workflows:{action}"):
            return True
        # 3. Fall back to wildcard
        return self.has_scope("*")

    def has_resource_access(self, resource: str, action: str, workflow: str = None) -> bool:
        """Check resource access with optional workflow scoping."""
        # 1. Try workflow-specific resource scope
        if workflow and self.has_scope(f"{resource}:{workflow}:{action}"):
            return True
        # 2. Fall back to global resource scope
        if self.has_scope(f"{resource}:{action}"):
            return True
        # 3. Fall back to wildcard
        return self.has_scope("*")

"""
User Management CLI Commands

Provides tools for Slidefactory admins to manage users and scopes.
"""

import click
import sys
from datetime import datetime
from sqlalchemy.orm import Session
from tabulate import tabulate

from app.util.database import SessionLocal
from app.auth.users.models import User
from app.auth.entra_mappings import get_scopes_for_groups, list_all_groups


def print_success(message: str):
    """Print success message in green."""
    click.echo(click.style(f"✓ {message}", fg="green"))


def print_error(message: str):
    """Print error message in red."""
    click.echo(click.style(f"✗ {message}", fg="red"), err=True)


def print_warning(message: str):
    """Print warning message in yellow."""
    click.echo(click.style(f"⚠ {message}", fg="yellow"))


@click.group()
def user():
    """User management commands."""
    pass


@user.command("list")
@click.option("--show-groups", is_flag=True, help="Show Entra groups")
@click.option("--show-scopes", is_flag=True, help="Show all scopes")
@click.option("--provider", help="Filter by auth provider (local/azure_ad)")
def list_users(show_groups, show_scopes, provider):
    """List all users with their access information."""
    db = SessionLocal()
    try:
        query = db.query(User).order_by(User.email)

        if provider:
            query = query.filter(User.auth_provider == provider)

        users = query.all()

        if not users:
            print_warning("No users found")
            return

        # Prepare table data
        headers = ["Email", "Name", "Provider", "Active", "Overridden", "Scopes"]
        if show_groups:
            headers.append("Entra Groups")
        if show_scopes:
            headers.append("All Scopes")

        rows = []
        for user in users:
            row = [
                user.email,
                user.name or "",
                user.auth_provider,
                "Yes" if user.active else "No",
                "Yes" if user.scopes_overridden else "No",
                len(user.scopes or [])
            ]

            if show_groups:
                groups = ", ".join(user.entra_groups or []) if user.entra_groups else "N/A"
                row.append(groups)

            if show_scopes:
                scopes = ", ".join(user.scopes or []) if user.scopes else "None"
                row.append(scopes)

            rows.append(row)

        click.echo(tabulate(rows, headers=headers, tablefmt="simple"))
        click.echo(f"\nTotal: {len(users)} users")

    finally:
        db.close()


@user.command("show")
@click.argument("email")
def show_user(email):
    """Show detailed information for a specific user."""
    db = SessionLocal()
    try:
        user = db.query(User).filter(User.email == email).first()

        if not user:
            print_error(f"User {email} not found")
            sys.exit(1)

        click.echo(f"\n{'='*60}")
        click.echo(f"User: {user.email}")
        click.echo(f"{'='*60}")
        click.echo(f"Name:              {user.name or 'N/A'}")
        click.echo(f"Provider:          {user.auth_provider}")
        click.echo(f"Active:            {user.active}")
        click.echo(f"Scopes Overridden: {user.scopes_overridden}")
        click.echo(f"Created:           {user.created_at or 'Unknown'}")
        click.echo(f"Last Login:        {user.last_login or 'Never'}")

        click.echo(f"\nScopes ({len(user.scopes or [])}):")
        for scope in (user.scopes or []):
            click.echo(f"  - {scope}")

        if user.entra_groups:
            click.echo(f"\nEntra Groups ({len(user.entra_groups)}):")
            for group in user.entra_groups:
                click.echo(f"  - {group}")

        if user.attributes:
            click.echo(f"\nLegacy Attributes:")
            click.echo(f"  {user.attributes}")

        click.echo()

    finally:
        db.close()


@user.command("add-scope")
@click.argument("email")
@click.argument("scope")
def add_scope(email, scope):
    """Add a scope to a user."""
    db = SessionLocal()
    try:
        user = db.query(User).filter(User.email == email).first()

        if not user:
            print_error(f"User {email} not found")
            sys.exit(1)

        current_scopes = set(user.scopes or [])

        if scope in current_scopes:
            print_warning(f"User {email} already has scope: {scope}")
            return

        current_scopes.add(scope)
        user.scopes = sorted(list(current_scopes))
        user.scopes_overridden = True

        db.commit()

        print_success(f"Added scope '{scope}' to user {email}")
        click.echo(f"User now has {len(user.scopes)} scopes")

    finally:
        db.close()


@user.command("remove-scope")
@click.argument("email")
@click.argument("scope")
def remove_scope(email, scope):
    """Remove a scope from a user."""
    db = SessionLocal()
    try:
        user = db.query(User).filter(User.email == email).first()

        if not user:
            print_error(f"User {email} not found")
            sys.exit(1)

        current_scopes = set(user.scopes or [])

        if scope not in current_scopes:
            print_warning(f"User {email} does not have scope: {scope}")
            return

        current_scopes.discard(scope)
        user.scopes = sorted(list(current_scopes))
        user.scopes_overridden = True

        db.commit()

        print_success(f"Removed scope '{scope}' from user {email}")
        click.echo(f"User now has {len(user.scopes)} scopes")

    finally:
        db.close()


@user.command("set-scopes")
@click.argument("email")
@click.argument("scopes")  # Comma-separated
def set_scopes(email, scopes):
    """Replace all scopes for a user (comma-separated list)."""
    db = SessionLocal()
    try:
        user = db.query(User).filter(User.email == email).first()

        if not user:
            print_error(f"User {email} not found")
            sys.exit(1)

        new_scopes = [s.strip() for s in scopes.split(",") if s.strip()]

        if not new_scopes:
            print_error("No valid scopes provided")
            sys.exit(1)

        user.scopes = sorted(new_scopes)
        user.scopes_overridden = True

        db.commit()

        print_success(f"Set {len(new_scopes)} scopes for user {email}")
        for scope in new_scopes:
            click.echo(f"  - {scope}")

    finally:
        db.close()


@user.command("reset-to-groups")
@click.argument("email")
def reset_to_groups(email):
    """Reset user scopes to match their current Entra groups."""
    db = SessionLocal()
    try:
        user = db.query(User).filter(User.email == email).first()

        if not user:
            print_error(f"User {email} not found")
            sys.exit(1)

        if user.auth_provider != "azure_ad":
            print_error(f"User {email} is not an Entra user (provider: {user.auth_provider})")
            sys.exit(1)

        if not user.entra_groups:
            print_error(f"User {email} has no Entra group data stored")
            sys.exit(1)

        new_scopes = get_scopes_for_groups(user.entra_groups)

        click.echo(f"Resetting scopes based on groups: {', '.join(user.entra_groups)}")
        click.echo(f"New scopes ({len(new_scopes)}):")
        for scope in new_scopes:
            click.echo(f"  - {scope}")

        if not click.confirm("\nProceed with reset?"):
            print_warning("Reset cancelled")
            return

        user.scopes = new_scopes
        user.scopes_overridden = False

        db.commit()

        print_success(f"Reset scopes for user {email}")

    finally:
        db.close()


@user.command("activate")
@click.argument("email")
def activate_user(email):
    """Activate a user account."""
    db = SessionLocal()
    try:
        user = db.query(User).filter(User.email == email).first()

        if not user:
            print_error(f"User {email} not found")
            sys.exit(1)

        if user.active:
            print_warning(f"User {email} is already active")
            return

        user.active = True
        db.commit()

        print_success(f"Activated user {email}")

    finally:
        db.close()


@user.command("deactivate")
@click.argument("email")
def deactivate_user(email):
    """Deactivate a user account."""
    db = SessionLocal()
    try:
        user = db.query(User).filter(User.email == email).first()

        if not user:
            print_error(f"User {email} not found")
            sys.exit(1)

        if not user.active:
            print_warning(f"User {email} is already inactive")
            return

        user.active = False
        db.commit()

        print_success(f"Deactivated user {email}")

    finally:
        db.close()


@user.command("list-groups")
def list_entra_groups():
    """List all configured Entra group mappings."""
    groups = list_all_groups()

    click.echo(f"\nConfigured Entra Groups ({len(groups)}):")
    click.echo(f"{'='*60}")

    for group in groups:
        from app.auth.entra_mappings import get_group_mapping
        scopes = get_group_mapping(group)
        click.echo(f"\n{group}")
        click.echo(f"  Scopes: {len(scopes)}")
        for scope in scopes[:5]:  # Show first 5
            click.echo(f"    - {scope}")
        if len(scopes) > 5:
            click.echo(f"    ... and {len(scopes) - 5} more")

    click.echo()


if __name__ == "__main__":
    user()

from cli.commands.user import user

# Register command group
cli.add_command(user)

# Scope presets for easy user creation
SCOPE_PRESETS = {
    'admin': ['*'],
    'workflow-admin': lambda workflow: [
        f'workflows:{workflow}:read',
        f'workflows:{workflow}:execute',
        f'templates:{workflow}:write',
        f'presentations:generate',
        f'results:write',
        f'contexts:write'
    ],
    'workflow-user': lambda workflow: [
        f'workflows:{workflow}:read',
        f'workflows:{workflow}:execute',
        f'templates:{workflow}:read',
        f'presentations:generate',
        f'presentations:read',
        f'results:read'
    ],
    'viewer': [
        'presentations:read',
        'results:read'
    ],
    'api-user': [
        'presentations:generate',
        'presentations:read',
        'results:read',
        'workflows:read'
    ]
}


@user.command("create-local")
@click.argument("email")
@click.option("--name", required=True, help="User display name")
@click.option("--password", prompt=True, hide_input=True, confirmation_prompt=True, help="User password")
@click.option("--scopes", help="Comma-separated list of scopes")
@click.option("--preset", type=click.Choice(['admin', 'workflow-admin', 'workflow-user', 'viewer', 'api-user']),
              help="Use scope preset")
@click.option("--workflow", help="Workflow ID (required for workflow-* presets)")
def create_local_user(email, name, password, scopes, preset, workflow):
    """
    Create a new local user with password and scopes.

    Examples:
        # Create admin user
        python -m cli user create-local admin@example.com --name "Admin" --preset admin

        # Create workflow user
        python -m cli user create-local user@example.com --name "User" --preset workflow-user --workflow esg2

        # Create with custom scopes
        python -m cli user create-local user@example.com --name "User" --scopes "presentations:read,results:read"
    """
    import bcrypt
    from app.auth.users.models import User
    from app.util.database import SessionLocal

    db = SessionLocal()
    try:
        # Check if user already exists
        existing = db.query(User).filter(User.email == email).first()
        if existing:
            print_error(f"User {email} already exists")
            sys.exit(1)

        # Determine scopes
        if preset:
            if preset in ['workflow-admin', 'workflow-user']:
                if not workflow:
                    print_error(f"--workflow required for preset '{preset}'")
                    sys.exit(1)
                user_scopes = SCOPE_PRESETS[preset](workflow)
            else:
                user_scopes = SCOPE_PRESETS[preset]
        elif scopes:
            user_scopes = [s.strip() for s in scopes.split(",") if s.strip()]
            if not user_scopes:
                print_error("No valid scopes provided")
                sys.exit(1)
        else:
            print_error("Either --scopes or --preset must be provided")
            sys.exit(1)

        # Hash password
        password_hash = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')

        # Create user
        user = User(
            email=email,
            name=name,
            password=password_hash,
            scopes=user_scopes,
            auth_provider="local",
            scopes_overridden=False,  # Not overridden, this is initial assignment
            active=True,
            created_at=datetime.utcnow()
        )

        db.add(user)
        db.commit()

        print_success(f"Local user {email} created successfully!")
        click.echo(f"Name: {name}")
        click.echo(f"Provider: local")
        click.echo(f"Scopes ({len(user_scopes)}):")
        for scope in user_scopes:
            click.echo(f"  - {scope}")

    except Exception as e:
        print_error(f"Failed to create user: {str(e)}")
        sys.exit(1)
    finally:
        db.close()


@user.command("change-password")
@click.argument("email")
@click.option("--password", prompt=True, hide_input=True, confirmation_prompt=True, help="New password")
def change_password(email, password):
    """Change password for a local user."""
    import bcrypt
    from app.auth.users.models import User
    from app.util.database import SessionLocal

    db = SessionLocal()
    try:
        user = db.query(User).filter(User.email == email).first()

        if not user:
            print_error(f"User {email} not found")
            sys.exit(1)

        if user.auth_provider != "local":
            print_error(f"Cannot change password for {user.auth_provider} user")
            sys.exit(1)

        # Hash new password
        password_hash = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
        user.password = password_hash

        db.commit()

        print_success(f"Password changed for user {email}")

    except Exception as e:
        print_error(f"Failed to change password: {str(e)}")
        sys.exit(1)
    finally:
        db.close()

"""
Unit tests for JIT user provisioning.
"""

import pytest
from datetime import datetime
from unittest.mock import Mock, patch

from app.auth.provisioning import (
    compute_initial_scopes,
    provision_new_user,
    update_existing_user_login,
    get_or_create_user
)
from app.auth.users.models import User


def test_compute_initial_scopes_single_group():
    """Test scope computation with single group."""
    groups = ["slidefactory-viewers"]
    scopes = compute_initial_scopes(groups)

    assert "presentations:read" in scopes
    assert "results:read" in scopes


def test_compute_initial_scopes_multiple_groups():
    """Test scope computation with multiple groups (union)."""
    groups = ["slidefactory-viewers", "slidefactory-esg-team"]
    scopes = compute_initial_scopes(groups)

    # Should have scopes from both groups
    assert "presentations:read" in scopes  # From viewers
    assert "workflows:esg2:execute" in scopes  # From esg-team


def test_compute_initial_scopes_admin():
    """Test that admin group gets wildcard scope."""
    groups = ["slidefactory-admins"]
    scopes = compute_initial_scopes(groups)

    assert "*" in scopes


def test_compute_initial_scopes_no_matching_groups():
    """Test default scopes when no groups match."""
    groups = ["some-unrelated-group"]
    scopes = compute_initial_scopes(groups)

    # Should get default scopes
    assert "presentations:read" in scopes
    assert len(scopes) == 1


def test_provision_new_user(db_session):
    """Test creating a new user with JIT provisioning."""
    email = "newuser@example.com"
    name = "New User"
    groups = ["slidefactory-users"]

    user = provision_new_user(db_session, email, name, groups)

    assert user.email == email
    assert user.name == name
    assert user.auth_provider == "azure_ad"
    assert user.scopes_overridden is False
    assert user.active is True
    assert len(user.scopes) > 0
    assert "presentations:read" in user.scopes


def test_update_existing_user_login(db_session):
    """Test updating login timestamp for existing user."""
    # Create existing user
    user = User(
        email="existing@example.com",
        name="Existing User",
        scopes=["templates:read"],
        auth_provider="azure_ad",
        scopes_overridden=True,
        active=True
    )
    db_session.add(user)
    db_session.commit()

    # Update login with different groups
    new_groups = ["slidefactory-admins"]
    updated_user = update_existing_user_login(db_session, user, new_groups)

    # Scopes should NOT change (overridden)
    assert updated_user.scopes == ["templates:read"]
    assert updated_user.entra_groups == new_groups
    assert updated_user.last_login is not None


def test_get_or_create_user_new(db_session):
    """Test get_or_create with new user."""
    email = "test@example.com"
    name = "Test User"
    groups = ["slidefactory-users"]

    user = get_or_create_user(db_session, email, name, groups)

    assert user.email == email
    assert user.id is not None


def test_get_or_create_user_existing(db_session):
    """Test get_or_create with existing user."""
    email = "existing@example.com"

    # Create user
    existing = User(
        email=email,
        name="Original Name",
        scopes=["custom:scope"],
        auth_provider="azure_ad",
        active=True
    )
    db_session.add(existing)
    db_session.commit()
    existing_id = existing.id

    # Try to provision again with different groups
    user = get_or_create_user(db_session, email, "Different Name", ["slidefactory-admins"])

    # Should return same user, scopes unchanged
    assert user.id == existing_id
    assert user.scopes == ["custom:scope"]
    assert user.last_login is not None


@pytest.fixture
def db_session():
    """Mock database session for testing."""
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    from app.auth.users.models import Base

    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    session = Session()

    yield session

    session.close()

"""
Integration tests for Entra authentication flow.
"""

import pytest
from fastapi.testclient import TestClient
from unittest.mock import patch, Mock

from app.main import app
from app.auth.users.models import User


@pytest.fixture
def client():
    """Test client."""
    return TestClient(app)


@pytest.fixture
def mock_entra_token():
    """Mock Entra token response."""
    return {
        "email": "testuser@example.com",
        "name": "Test User",
        "groups": ["slidefactory-users"]
    }


def test_azure_callback_new_user(client, db_session, mock_entra_token):
    """Test Azure callback creates new user on first login."""
    with patch("app.auth.azure_router.exchange_code_for_token") as mock_exchange:
        mock_exchange.return_value = mock_entra_token

        response = client.get("/auth/azure/callback?code=test_code")

        assert response.status_code == 302  # Redirect

        # Check user was created
        user = db_session.query(User).filter(
            User.email == "testuser@example.com"
        ).first()

        assert user is not None
        assert user.auth_provider == "azure_ad"
        assert len(user.scopes) > 0
        assert user.scopes_overridden is False


def test_azure_callback_existing_user(client, db_session, mock_entra_token):
    """Test Azure callback updates existing user."""
    # Create existing user with custom scopes
    existing = User(
        email="testuser@example.com",
        name="Test User",
        scopes=["custom:scope", "*"],
        auth_provider="azure_ad",
        scopes_overridden=True,
        active=True
    )
    db_session.add(existing)
    db_session.commit()

    with patch("app.auth.azure_router.exchange_code_for_token") as mock_exchange:
        mock_exchange.return_value = mock_entra_token

        response = client.get("/auth/azure/callback?code=test_code")

        assert response.status_code == 302

        # Check scopes were NOT changed
        db_session.refresh(existing)
        assert existing.scopes == ["custom:scope", "*"]
        assert existing.last_login is not None


def test_azure_callback_inactive_user(client, db_session, mock_entra_token):
    """Test that inactive users cannot log in."""
    # Create inactive user
    inactive = User(
        email="testuser@example.com",
        name="Test User",
        scopes=["presentations:read"],
        auth_provider="azure_ad",
        active=False
    )
    db_session.add(inactive)
    db_session.commit()

    with patch("app.auth.azure_router.exchange_code_for_token") as mock_exchange:
        mock_exchange.return_value = mock_entra_token

        response = client.get("/auth/azure/callback?code=test_code")

        assert response.status_code == 403

"""
Migrate users from config.py ALLOWED_USERS to database.

This script imports users from the hardcoded config file into the database,
converting legacy attributes to scope-based permissions.
"""

import sys
from datetime import datetime
from typing import Dict, List
import bcrypt

from app.util.database import SessionLocal
from app.auth.users.models import User


def convert_attributes_to_scopes(attributes: dict) -> List[str]:
    """
    Convert legacy attributes format to scope-based format.

    Args:
        attributes: Legacy attributes dict with workflows, checks, prompts, etc.

    Returns:
        List of scopes
    """
    scopes = []

    # Convert workflow access
    for workflow in attributes.get('workflows', []):
        scopes.append(f"workflows:{workflow}:read")
        scopes.append(f"workflows:{workflow}:execute")

    # Convert resource permissions
    resource_map = {
        'checks': 'checks',
        'prompts': 'prompts',
        'contexts': 'contexts',
        'users': 'users',
        'files': 'templates',
        'templates': 'templates'
    }

    for attr_key, scope_resource in resource_map.items():
        perms = attributes.get(attr_key, [])
        if 'view' in perms:
            scopes.append(f"{scope_resource}:read")
        if 'admin' in perms:
            scopes.append(f"{scope_resource}:write")
            scopes.append(f"{scope_resource}:delete")

    # Add common permissions based on workflows
    if attributes.get('workflows'):
        # If user has workflows, they probably generate presentations
        if 'presentations:generate' not in scopes:
            scopes.append('presentations:generate')
        if 'presentations:read' not in scopes:
            scopes.append('presentations:read')
        if 'results:read' not in scopes:
            scopes.append('results:read')

    return sorted(list(set(scopes)))


def migrate_config_users(dry_run: bool = True) -> Dict[str, str]:
    """
    Migrate users from config.py to database.

    Args:
        dry_run: If True, don't commit changes

    Returns:
        Dict mapping email to status (created, updated, skipped, error)
    """
    # Import config (this will fail if ALLOWED_USERS is already removed)
    try:
        from app.config import ALLOWED_USERS
    except ImportError:
        print("ERROR: Cannot import ALLOWED_USERS from app.config")
        print("Either config has already been cleaned up, or there's an import error")
        sys.exit(1)

    if not ALLOWED_USERS:
        print("No users found in ALLOWED_USERS")
        return {}

    db = SessionLocal()
    results = {}

    try:
        print(f"\n{'='*60}")
        print(f"Migrating {len(ALLOWED_USERS)} users from config.py")
        print(f"Dry Run: {dry_run}")
        print(f"{'='*60}\n")

        for email, user_config in ALLOWED_USERS.items():
            try:
                # Check if user already exists
                existing = db.query(User).filter(User.email == email).first()

                if existing:
                    print(f"⚠  SKIP: {email} (already exists in database)")
                    results[email] = "skipped"
                    continue

                # Convert attributes to scopes
                attributes = user_config.get('attributes', {})
                scopes = convert_attributes_to_scopes(attributes)

                # Generate a random password (will be reset by admin)
                temp_password = f"RESET_ME_{datetime.utcnow().timestamp()}"
                password_hash = bcrypt.hashpw(temp_password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')

                # Create user
                user = User(
                    id=user_config.get('id') if user_config.get('id', 0) > 0 else None,  # Preserve positive IDs
                    email=email,
                    name=user_config.get('name', email.split('@')[0]),
                    password=password_hash,
                    scopes=scopes,
                    auth_provider="local",
                    attributes=attributes,  # Preserve for reference
                    scopes_overridden=False,
                    active=True,
                    created_at=datetime.utcnow()
                )

                if dry_run:
                    print(f"✓  WOULD CREATE: {email}")
                    print(f"   Name: {user.name}")
                    print(f"   Scopes ({len(scopes)}): {', '.join(scopes[:5])}{'...' if len(scopes) > 5 else ''}")
                    results[email] = "would_create"
                else:
                    db.add(user)
                    db.flush()  # Get the ID without committing
                    print(f"✓  CREATED: {email} (ID: {user.id})")
                    print(f"   Name: {user.name}")
                    print(f"   Scopes ({len(scopes)}): {', '.join(scopes[:5])}{'...' if len(scopes) > 5 else ''}")
                    results[email] = "created"

            except Exception as e:
                print(f"✗  ERROR: {email} - {str(e)}")
                results[email] = f"error: {str(e)}"
                if not dry_run:
                    db.rollback()

        if not dry_run:
            db.commit()
            print(f"\n{'='*60}")
            print("✓ Migration completed and committed to database")
            print(f"{'='*60}\n")
        else:
            print(f"\n{'='*60}")
            print("This was a DRY RUN - no changes made")
            print("Run with --execute to apply changes")
            print(f"{'='*60}\n")

    finally:
        db.close()

    return results


def print_summary(results: Dict[str, str]):
    """Print migration summary."""
    from collections import Counter

    counts = Counter(results.values())

    print("\nMigration Summary:")
    print(f"  Created:      {counts.get('created', 0)}")
    print(f"  Would Create: {counts.get('would_create', 0)}")
    print(f"  Skipped:      {counts.get('skipped', 0)}")
    print(f"  Errors:       {sum(1 for v in results.values() if v.startswith('error'))}")
    print()


if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="Migrate users from config.py to database")
    parser.add_argument("--execute", action="store_true", help="Actually perform migration (default is dry-run)")
    args = parser.parse_args()

    results = migrate_config_users(dry_run=not args.execute)
    print_summary(results)

    if not args.execute:
        print("To execute migration, run: python scripts/migrate_config_users_to_database.py --execute")