# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""User authentication and representation.

Provides the User class for representing authenticated users and functions
to determine the current logged-in user from the session.
"""

import os

from flask import session

import settings
from framework import xsrf


class UserNotFoundError(Exception):
    """Raised when a user is not found."""

    pass


class User(object):
    """Provides the email address, nickname, and ID for a user.

    A nickname is a human-readable string that uniquely identifies a
    Google user, akin to a username. For some users, this nickname is
    an email address, but for other users, a different nickname is
    used.

    A user is a Google Accounts user.

    `federated_identity` and `federated_provider` are decommissioned
    and should not be used.

    This class is based on google.appengine.api.users.User class

    """

    __user_id = None
    __federated_identity = None
    __federated_provider = None

    def __init__(
        self,
        email=None,
        _auth_domain=None,
        _user_id=None,
        federated_identity=None,
        federated_provider=None,
        _strict_mode=True,
    ):
        """Constructor.

        Args:
        email: An optional string of the user's email address. It defaults to
            the current user's email address.
        federated_identity: Decommissioned, don't use.
        federated_provider: Decommissioned, don't use.

        Raises:
        UserNotFoundError: If the user is not logged in and both `email` and
            `federated_identity` are empty.
        """
        self.__email = email
        self.__federated_identity = federated_identity
        self.__federated_provider = federated_provider
        self.__auth_domain = _auth_domain
        self.__user_id = _user_id or None

    def nickname(self):
        """Returns the user's nickname.

        The nickname will be a unique, human readable identifier for
        this user with respect to this application. It will be an
        email address for some users, and part of the email address
        for some users.

        Returns:
        The nickname of the user as a string.
        """
        if (
            self.__email
            and self.__auth_domain
            and self.__email.endswith('@' + self.__auth_domain)
        ):
            suffix_len = len(self.__auth_domain) + 1
            return self.__email[:-suffix_len]
        elif self.__federated_identity:
            return self.__federated_identity
        else:
            return self.__email

    def email(self):
        """Returns the user's email address."""
        return self.__email

    def user_id(self):
        """Obtains the user ID of the user.

        Returns:
        A permanent unique identifying string or `None`. If the email
        address was set explicity, this will return `None`.
        """
        return self.__user_id

    def auth_domain(self):
        """Obtains the user's authentication domain.

        Returns:
        A string containing the authentication domain. This method is
        internal and should not be used by client applications.
        """
        return self.__auth_domain

    def federated_identity(self):
        """Decommissioned, don't use.

        Returns:
        A string containing the federated identity of the user. If the
        user is not a federated user, `None` is returned.
        """
        return self.__federated_identity

    def federated_provider(self):
        """Decommissioned, don't use.

        Returns:
        A string containing the federated provider. If the user is not
        a federated user, `None` is returned.
        """
        return self.__federated_provider

    def __unicode__(self):
        """Return a string representation of the user."""
        return six_subset.text_type(self.nickname())  # noqa: F821

    def __str__(self):
        """Return a string representation of the user."""
        return str(self.nickname())

    def __repr__(self):
        """Return a string representation of the user object."""
        values = []
        if self.__email:
            values.append("email='%s'" % self.__email)
        if self.__federated_identity:
            values.append("federated_identity='%s'" % self.__federated_identity)
        if self.__user_id:
            values.append("_user_id='%s'" % self.__user_id)
        return 'users.User(%s)' % ','.join(values)

    def __hash__(self):
        """Return a hash for the user."""
        if self.__federated_identity:
            return hash((self.__federated_identity, self.__auth_domain))
        else:
            return hash((self.__email, self.__auth_domain))

    def __eq__(self, other):
        """Check if this user is equal to another."""
        if not isinstance(other, User):
            return NotImplemented
        if self.__federated_identity:
            return (self.__federated_identity, self.__auth_domain) == (
                other.__federated_identity,
                other.__auth_domain,
            )
        else:
            return (self.__email, self.__auth_domain) == (
                other.__email,
                other.__auth_domain,
            )

    def __lt__(self, other):
        """Check if this user is less than another."""
        if not isinstance(other, User):
            return NotImplemented
        if self.__federated_identity:
            return (self.__federated_identity, self.__auth_domain) < (
                other.__federated_identity,
                other.__auth_domain,
            )
        else:
            return (self.__email, self.__auth_domain) < (
                other.__email,
                other.__auth_domain,
            )

    def get_current_user(self):
        """Retrieves information associated with the requesting user.

        Returns:
        The current user object.
        """
        try:
            return User()
        except UserNotFoundError:
            return None

    def is_current_user_admin(self):
        """Specifies whether the user making a request is an application admin.

        Because administrator status is not persisted in the
        datastore, `is_current_user_admin()` is a separate function
        rather than a member function of the `User` class. The status
        only exists for the user making the current request.

        Returns:
        `True` if the user is an administrator; all other user types
        return `False`.
        """
        # This env variable was set by GAE based on a GAE session cookie.
        # Using Sign-In With Google, it will probably never be present.
        # Hence, currently is always False.
        # We don't use this.  We check a boolean in the AppUser model.
        return (os.environ.get('USER_IS_ADMIN', '0')) == '1'


def get_current_user():
    """Gets the current user based on the environment or test mode."""
    if settings.UNIT_TEST_MODE:
        user_via_env = None
        if os.environ.get('USER_EMAIL', '') != '':
            user_via_env = User(email=os.environ['USER_EMAIL'])
        return user_via_env

    user_info, signature = session.get('signed_user_info', (None, None))
    if user_info:
        try:
            xsrf.validate_token(
                signature,
                str(user_info),
                timeout=xsrf.REFRESH_TOKEN_TIMEOUT_SEC,
            )
            user_via_signed_user_info = User(email=user_info['email'])
            return user_via_signed_user_info

        except xsrf.TokenIncorrect:
            # If anything is not right, give the user a fresh session.
            session.clear()
            pass

    return None  # User is not signed in.


def is_current_user_admin():
    """Return whether the current user has administrative privileges."""
    return False


def add_signed_user_info_to_session(email):
    """Create and sign the user info in the Flask session."""
    user_info = {
        'email': email,
    }
    signature = xsrf.generate_token(str(user_info))
    session['signed_user_info'] = user_info, signature


def refresh_user_session():
    """If the user is signed in, update the signed user info with a new date."""
    user = get_current_user()
    if user:
        add_signed_user_info_to_session(user.email())
