gatehouse-api/gatehouse_app/services/external_auth/linking.py

"""Account linking, authentication, and unlinking flows."""
import logging
import secrets
from datetime import datetime
from typing import Optional, Tuple

from gatehouse_app.models import User, AuthenticationMethod
from gatehouse_app.models.auth.authentication_method import OAuthState
from gatehouse_app.utils.constants import AuthMethodType
from gatehouse_app.services.audit_service import AuditService
from gatehouse_app.services.external_auth.models import ExternalAuthError

logger = logging.getLogger(__name__)


def initiate_link_flow(
    get_provider_config,
    user_id: str,
    provider_type: AuthMethodType,
    organization_id: str,
    redirect_uri: str = None,
) -> Tuple[str, str]:
    from gatehouse_app.services.external_auth._helpers import (
        _compute_s256_challenge,
        _build_authorization_url,
    )

    provider_type_str = provider_type.value if isinstance(provider_type, AuthMethodType) else provider_type
    config = get_provider_config(provider_type, organization_id)

    if redirect_uri and not config.is_redirect_uri_allowed(redirect_uri):
        raise ExternalAuthError("Invalid redirect URI", "INVALID_REDIRECT_URI", 400)

    code_verifier = None
    code_challenge = None
    if provider_type_str not in ('google', 'microsoft'):
        code_verifier = secrets.token_urlsafe(32)
        code_challenge = _compute_s256_challenge(code_verifier)

    state = OAuthState.create_state(
        flow_type="link",
        provider_type=provider_type,
        user_id=user_id,
        organization_id=organization_id,
        redirect_uri=redirect_uri or (config.redirect_uris[0] if config.redirect_uris else None),
        code_verifier=code_verifier,
        code_challenge=code_challenge,
        lifetime_seconds=600,
    )

    auth_url = _build_authorization_url(config=config, state=state)

    AuditService.log_external_auth_link_initiated(
        user_id=user_id,
        organization_id=organization_id,
        provider_type=provider_type_str,
        state_id=state.id,
    )

    return auth_url, state.state


def complete_link_flow(
    get_provider_config,
    provider_type: AuthMethodType,
    authorization_code: str,
    state: str,
    redirect_uri: str,
) -> AuthenticationMethod:
    from gatehouse_app.services.external_auth._helpers import (
        _exchange_code,
        _get_user_info,
        _encrypt_provider_data,
    )

    provider_type_str = provider_type.value if isinstance(provider_type, AuthMethodType) else provider_type

    state_record = OAuthState.query.filter_by(state=state).first()
    if not state_record or not state_record.is_valid():
        AuditService.log_external_auth_link_failed(
            user_id=None,
            organization_id=None,
            provider_type=provider_type_str,
            error_message="Invalid or expired OAuth state",
            failure_reason="invalid_state",
        )
        raise ExternalAuthError("Invalid or expired OAuth state", "INVALID_STATE", 400)

    if state_record.flow_type != "link":
        AuditService.log_external_auth_link_failed(
            user_id=state_record.user_id,
            organization_id=state_record.organization_id,
            provider_type=provider_type_str,
            error_message="Invalid flow type for this operation",
            failure_reason="invalid_flow_type",
        )
        raise ExternalAuthError("Invalid flow type for this operation", "INVALID_FLOW_TYPE", 400)

    if state_record.provider_type != provider_type_str:
        AuditService.log_external_auth_link_failed(
            user_id=state_record.user_id,
            organization_id=state_record.organization_id,
            provider_type=provider_type_str,
            error_message="Provider mismatch",
            failure_reason="provider_mismatch",
        )
        raise ExternalAuthError("Provider mismatch", "PROVIDER_MISMATCH", 400)

    config = get_provider_config(provider_type, state_record.organization_id)

    tokens = _exchange_code(
        config=config,
        code=authorization_code,
        redirect_uri=redirect_uri,
        code_verifier=state_record.code_verifier,
    )

    user_info = _get_user_info(config=config, access_token=tokens["access_token"])

    user = User.query.get(state_record.user_id)
    if not user:
        AuditService.log_external_auth_link_failed(
            user_id=None,
            organization_id=state_record.organization_id,
            provider_type=provider_type_str,
            error_message="User not found",
            failure_reason="user_not_found",
        )
        raise ExternalAuthError("User not found", "USER_NOT_FOUND", 400)

    conflicting = AuthenticationMethod.query.filter(
        AuthenticationMethod.method_type == provider_type,
        AuthenticationMethod.provider_user_id == user_info["provider_user_id"],
        AuthenticationMethod.user_id != user.id,
        AuthenticationMethod.deleted_at == None,
    ).first()
    if conflicting:
        raise ExternalAuthError(
            f"This {provider_type_str} account is already linked to a different Secuird user.",
            "PROVIDER_ALREADY_LINKED",
            409,
        )

    auth_method = AuthenticationMethod.query.filter_by(
        user_id=user.id,
        method_type=provider_type,
        provider_user_id=user_info["provider_user_id"],
    ).first()

    if auth_method:
        # Restore the row if it was previously soft-deleted (re-linking after admin unlink)
        auth_method.deleted_at = None
        auth_method.provider_data = _encrypt_provider_data(tokens, user_info)
        auth_method.verified = user_info.get("email_verified", False)
        auth_method.last_used_at = datetime.utcnow()
        auth_method.save()
    else:
        auth_method = AuthenticationMethod(
            user_id=user.id,
            method_type=provider_type,
            provider_user_id=user_info["provider_user_id"],
            provider_data=_encrypt_provider_data(tokens, user_info),
            verified=user_info.get("email_verified", False),
            is_primary=False,
            last_used_at=datetime.utcnow(),
        )
        auth_method.save()

    state_record.mark_used()

    AuditService.log_external_auth_link_completed(
        user_id=user.id,
        organization_id=state_record.organization_id,
        provider_type=provider_type_str,
        provider_user_id=user_info["provider_user_id"],
        auth_method_id=auth_method.id,
    )

    return auth_method


def authenticate_with_provider(
    get_provider_config,
    provider_type: AuthMethodType,
    organization_id: str,
    authorization_code: str,
    state: str,
    redirect_uri: str,
) -> Tuple[User, dict]:
    from gatehouse_app.services.external_auth._helpers import (
        _exchange_code,
        _get_user_info,
        _encrypt_provider_data,
    )

    provider_type_str = provider_type.value if isinstance(provider_type, AuthMethodType) else provider_type

    state_record = OAuthState.query.filter_by(state=state).first()
    if not state_record or not state_record.is_valid():
        AuditService.log_external_auth_login_failed(
            organization_id=organization_id,
            provider_type=provider_type_str,
            failure_reason="invalid_state",
            error_message="Invalid or expired OAuth state",
        )
        raise ExternalAuthError("Invalid or expired OAuth state", "INVALID_STATE", 400)

    config = get_provider_config(provider_type, organization_id)

    tokens = _exchange_code(
        config=config,
        code=authorization_code,
        redirect_uri=redirect_uri,
        code_verifier=state_record.code_verifier,
    )

    user_info = _get_user_info(config=config, access_token=tokens["access_token"])

    auth_method = AuthenticationMethod.query.filter_by(
        method_type=provider_type,
        provider_user_id=user_info["provider_user_id"],
        deleted_at=None,
    ).first()

    if not auth_method:
        existing_user = User.query.filter_by(email=user_info["email"], deleted_at=None).first()

        if existing_user:
            AuditService.log_external_auth_login_failed(
                organization_id=organization_id,
                provider_type=provider_type_str,
                provider_user_id=user_info["provider_user_id"],
                email=user_info["email"],
                failure_reason="email_exists",
                error_message=f"An account with email {user_info['email']} already exists",
            )
            raise ExternalAuthError(
                f"An account with email {user_info['email']} already exists. "
                "Please log in with your password and link your Google account from settings.",
                "EMAIL_EXISTS",
                400,
            )

        AuditService.log_external_auth_login_failed(
            organization_id=organization_id,
            provider_type=provider_type_str,
            provider_user_id=user_info["provider_user_id"],
            email=user_info["email"],
            failure_reason="account_not_found",
            error_message="No Secuird account matches this external account",
        )
        raise ExternalAuthError(
            "No Secuird account matches this external account. Please register first.",
            "ACCOUNT_NOT_FOUND",
            400,
        )

    user = auth_method.user
    auth_method.provider_data = _encrypt_provider_data(tokens, user_info)
    auth_method.last_used_at = datetime.utcnow()
    auth_method.save()

    state_record.mark_used()

    from gatehouse_app.services.auth_service import AuthService
    session = AuthService.create_session(user=user)

    AuditService.log_external_auth_login(
        user_id=user.id,
        organization_id=organization_id,
        provider_type=provider_type_str,
        provider_user_id=user_info["provider_user_id"],
        auth_method_id=auth_method.id,
        session_id=session.id,
    )

    return user, session.to_dict()


def unlink_provider(
    user_id: str,
    provider_type: AuthMethodType,
    organization_id: str = None,
) -> bool:
    provider_type_str = provider_type.value if isinstance(provider_type, AuthMethodType) else provider_type

    auth_method = AuthenticationMethod.query.filter_by(
        user_id=user_id,
        method_type=provider_type,
        deleted_at=None,
    ).first()

    if not auth_method:
        raise ExternalAuthError("Provider not linked", "PROVIDER_NOT_LINKED", 400)

    other_methods = AuthenticationMethod.query.filter_by(user_id=user_id, deleted_at=None).count()
    if other_methods <= 1:
        raise ExternalAuthError(
            "Cannot unlink the last authentication method",
            "CANNOT_UNLINK_LAST",
            400,
        )

    provider_user_id = auth_method.provider_user_id
    auth_method_id = auth_method.id
    auth_method.delete()

    AuditService.log_external_auth_unlink(
        user_id=user_id,
        organization_id=organization_id,
        provider_type=provider_type_str,
        provider_user_id=provider_user_id,
        auth_method_id=auth_method_id,
    )

    return True


def get_linked_accounts(user_id: str) -> list:
    from gatehouse_app.utils.constants import AuthMethodType as AMT

    methods = AuthenticationMethod.query.filter_by(user_id=user_id, deleted_at=None).all()

    external_providers = [AMT.GOOGLE, AMT.GITHUB, AMT.MICROSOFT]

    return [
        {
            "id": m.id,
            "provider_type": m.method_type.value if hasattr(m.method_type, 'value') else str(m.method_type),
            "provider_user_id": m.provider_user_id,
            "email": m.provider_data.get("email") if m.provider_data else None,
            "name": m.provider_data.get("name") if m.provider_data else None,
            "picture": m.provider_data.get("picture") if m.provider_data else None,
            "verified": m.verified,
            "linked_at": m.created_at.isoformat() if m.created_at else None,
            "last_used_at": m.last_used_at.isoformat() if m.last_used_at else None,
        }
        for m in methods
        if m.method_type in external_providers
        or str(m.method_type) in [p.value for p in external_providers]
    ]