# Copyright (C) 2010-2024 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/

"""
RhodeCode authentication plugin for LDAP
"""

import logging

import colander

from rhodecode.authentication.plugins.services.ldap_dao import LdapDao
from rhodecode.translation import _
from rhodecode.authentication.base import RhodeCodeExternalAuthPlugin, hybrid_property
from rhodecode.authentication.schema import AuthnPluginSettingsSchemaBase, TwoFactorAuthnPluginSettingsSchemaMixin
from rhodecode.authentication.routes import AuthnPluginResourceBase
from rhodecode.lib.colander_utils import strip_whitespace
from rhodecode.lib.exceptions import LdapUsernameError, LdapPasswordError, LdapImportError
from rhodecode.lib.str_utils import safe_str
from rhodecode.model.db import User


log = logging.getLogger(__name__)


class LdapError(Exception):
    pass


def plugin_factory(plugin_id, *args, **kwargs):
    """
    Factory function that is called during plugin discovery.
    It returns the plugin instance.
    """
    plugin = RhodeCodeAuthPlugin(plugin_id)
    return plugin


class LdapAuthnResource(AuthnPluginResourceBase):
    pass


class LdapSettingsSchema(TwoFactorAuthnPluginSettingsSchemaMixin, AuthnPluginSettingsSchemaBase):
    tls_kind_choices = ["PLAIN", "LDAPS", "START_TLS"]
    tls_reqcert_choices = ["NEVER", "ALLOW", "TRY", "DEMAND", "HARD"]
    search_scope_choices = ["BASE", "ONELEVEL", "SUBTREE"]

    host = colander.SchemaNode(
        colander.String(),
        default="",
        description=_(
            "Host[s] of the LDAP Server \n"
            "(e.g., 192.168.2.154, or ldap-server.domain.com.\n "
            "Multiple servers can be specified using commas"
        ),
        preparer=strip_whitespace,
        title=_("LDAP Host"),
        widget="string",
    )
    port = colander.SchemaNode(
        colander.Int(),
        default=389,
        description=_(
            "Custom port that the LDAP server is listening on. Default value is: 389, use 636 for LDAPS (SSL)"
        ),
        preparer=strip_whitespace,
        title=_("Port"),
        validator=colander.Range(min=0, max=65536),
        widget="int",
    )

    timeout = colander.SchemaNode(
        colander.Int(),
        default=60 * 5,
        description=_("Timeout for LDAP connection"),
        preparer=strip_whitespace,
        title=_("Connection timeout"),
        validator=colander.Range(min=1),
        widget="int",
    )

    dn_user = colander.SchemaNode(
        colander.String(),
        default="",
        description=_(
            "Optional user DN/account to connect to LDAP if authentication is required. \n"
            "e.g., cn=admin,dc=mydomain,dc=com, or "
            "uid=root,cn=users,dc=mydomain,dc=com, or admin@mydomain.com"
        ),
        missing="",
        preparer=strip_whitespace,
        title=_("Bind account"),
        widget="string",
    )
    dn_pass = colander.SchemaNode(
        colander.String(),
        default="",
        description=_("Password to authenticate for given user DN."),
        missing="",
        preparer=strip_whitespace,
        title=_("Bind account password"),
        widget="password",
    )
    tls_kind = colander.SchemaNode(
        colander.String(),
        default=tls_kind_choices[0],
        description=_("TLS Type"),
        title=_("Connection Security"),
        validator=colander.OneOf(tls_kind_choices),
        widget="select",
    )
    tls_reqcert = colander.SchemaNode(
        colander.String(),
        default=tls_reqcert_choices[0],
        description=_(
            "Require Cert over TLS?. Self-signed and custom "
            "certificates can be used when\n `RhodeCode Certificate` "
            "found in admin > settings > system info page is extended."
        ),
        title=_("Certificate Checks"),
        validator=colander.OneOf(tls_reqcert_choices),
        widget="select",
    )
    tls_cert_file = colander.SchemaNode(
        colander.String(),
        default="",
        description=_(
            "This specifies the PEM-format file path containing "
            "certificates for use in TLS connection.\n"
            "If not specified `TLS Cert dir` will be used"
        ),
        title=_("TLS Cert file"),
        missing="",
        widget="string",
    )
    tls_cert_dir = colander.SchemaNode(
        colander.String(),
        default=LdapDao.default_tls_cert_dir,
        description=_(
            "This specifies the path of a directory that contains individual CA certificates in separate files."
        ),
        title=_("TLS Cert dir"),
        widget="string",
    )
    base_dn = colander.SchemaNode(
        colander.String(),
        default="",
        description=_(
            "Base DN to search. Dynamic bind is supported. Add `$login` marker "
            "in it to be replaced with current user username \n"
            "(e.g., dc=mydomain,dc=com, or ou=Users,dc=mydomain,dc=com)"
        ),
        missing="",
        preparer=strip_whitespace,
        title=_("Base DN"),
        widget="string",
    )
    filter = colander.SchemaNode(
        colander.String(),
        default="",
        description=_(
            "Filter to narrow results \n"
            "(e.g., (&(objectCategory=Person)(objectClass=user)), or \n"
            "(memberof=cn=rc-login,ou=groups,ou=company,dc=mydomain,dc=com)))"
        ),
        missing="",
        preparer=strip_whitespace,
        title=_("LDAP Search Filter"),
        widget="string",
    )

    search_scope = colander.SchemaNode(
        colander.String(),
        default=search_scope_choices[2],
        description=_("How deep to search LDAP. If unsure set to SUBTREE"),
        title=_("LDAP Search Scope"),
        validator=colander.OneOf(search_scope_choices),
        widget="select",
    )
    attr_login = colander.SchemaNode(
        colander.String(),
        default="uid",
        description=_("LDAP Attribute to map to user name (e.g., uid, or sAMAccountName)"),
        preparer=strip_whitespace,
        title=_("Login Attribute"),
        missing_msg=_("The LDAP Login attribute of the CN must be specified"),
        widget="string",
    )
    attr_email = colander.SchemaNode(
        colander.String(),
        default="",
        description=_(
            "LDAP Attribute to map to email address (e.g., mail).\n"
            "Emails are a crucial part of RhodeCode. \n"
            "If possible add a valid email attribute to ldap users."
        ),
        missing="",
        preparer=strip_whitespace,
        title=_("Email Attribute"),
        widget="string",
    )
    attr_firstname = colander.SchemaNode(
        colander.String(),
        default="",
        description=_("LDAP Attribute to map to first name (e.g., givenName)"),
        missing="",
        preparer=strip_whitespace,
        title=_("First Name Attribute"),
        widget="string",
    )
    attr_lastname = colander.SchemaNode(
        colander.String(),
        default="",
        description=_("LDAP Attribute to map to last name (e.g., sn)"),
        missing="",
        preparer=strip_whitespace,
        title=_("Last Name Attribute"),
        widget="string",
    )
    sync_active_directory_users = colander.SchemaNode(
        colander.Bool(),
        default=False,
        description=_(
            "A cron job that periodically retrieves all users from an LDAP-based Active Directory server and syncs "
            "them with the RhodeCode database.\n"
            "Note: This feature is specific to Active Directory. Enabling it for other types of LDAP servers will have no effect."
        ),
        missing=False,
        preparer=strip_whitespace,
        title=_("Sync Active Directory Users"),
        widget="bool",
    )


def get_ldap_args(settings: dict):
    return {
        "server": settings.get("host", ""),
        "base_dn": settings.get("base_dn", ""),
        "port": settings.get("port"),
        "bind_dn": settings.get("dn_user"),
        "bind_pass": settings.get("dn_pass"),
        "tls_kind": settings.get("tls_kind"),
        "tls_reqcert": settings.get("tls_reqcert"),
        "tls_cert_file": settings.get("tls_cert_file"),
        "tls_cert_dir": settings.get("tls_cert_dir"),
        "search_scope": settings.get("search_scope"),
        "attr_login": settings.get("attr_login"),
        "ldap_version": 3,
        "ldap_filter": settings.get("filter"),
        "timeout": settings.get("timeout"),
    }


class RhodeCodeAuthPlugin(RhodeCodeExternalAuthPlugin):
    uid = "ldap"
    # used to define dynamic binding in the
    DYNAMIC_BIND_VAR = "$login"
    _settings_unsafe_keys = ["dn_pass"]

    def includeme(self, config):
        config.add_authn_plugin(self)
        config.add_authn_resource(self.get_id(), LdapAuthnResource(self))
        config.add_view(
            "rhodecode.authentication.views.AuthnPluginViewBase",
            attr="settings_get",
            renderer="rhodecode:templates/admin/auth/plugin_settings.mako",
            request_method="GET",
            route_name="auth_home",
            context=LdapAuthnResource,
        )
        config.add_view(
            "rhodecode.authentication.views.AuthnPluginViewBase",
            attr="settings_post",
            renderer="rhodecode:templates/admin/auth/plugin_settings.mako",
            request_method="POST",
            route_name="auth_home",
            context=LdapAuthnResource,
        )

    def get_settings_schema(self):
        return LdapSettingsSchema()

    def get_display_name(self, load_from_settings=False):
        return _("LDAP")

    @classmethod
    def docs(cls):
        return "https://docs.rhodecode.com/RhodeCode-Enterprise/auth/auth-ldap.html"

    @hybrid_property
    def name(self):
        return "ldap"

    def use_fake_password(self):
        return True

    def user_activation_state(self):
        def_user_perms = User.get_default_user().AuthUser().permissions["global"]
        return "hg.extern_activate.auto" in def_user_perms

    def try_dynamic_binding(self, username, password, current_args):
        """
        Detects marker inside our original bind, and uses dynamic auth if
        present
        """

        org_bind = current_args["bind_dn"]
        passwd = current_args["bind_pass"]

        def has_bind_marker(_username):
            if self.DYNAMIC_BIND_VAR in _username:
                return True

        # we only passed in user with "special" variable
        if org_bind and has_bind_marker(org_bind) and not passwd:
            log.debug(
                "Using dynamic user/password binding for ldap authentication. Replacing `%s` with username",
                self.DYNAMIC_BIND_VAR,
            )
            current_args["bind_dn"] = org_bind.replace(self.DYNAMIC_BIND_VAR, username)
            current_args["bind_pass"] = password

        return current_args

    def auth(self, userobj, username, password, settings, **kwargs):
        """
        Given a user object (which may be null), username, a plaintext password,
        and a settings object (containing all the keys needed as listed in
        settings()), authenticate this user's login attempt.

        Return None on failure. On success, return a dictionary of the form:

            see: RhodeCodeAuthPluginBase.auth_func_attrs
        This is later validated for correctness
        """

        if not username or not password:
            log.debug("Empty username or password skipping...")
            return None

        ldap_args = get_ldap_args(settings)

        ldap_attrs = self.try_dynamic_binding(username, password, ldap_args)

        log.debug("Checking for ldap authentication.")

        try:
            ldap_dao = LdapDao(**ldap_args)
            (user_dn, ldap_attrs) = ldap_dao.authenticate_ldap(username, password)
            log.debug("Got ldap DN response %s", user_dn)

            def get_ldap_attr(k) -> str:
                return safe_str(ldap_attrs.get(settings.get(k), [b""])[0])

            # old attrs fetched from RhodeCode database
            admin = getattr(userobj, "admin", False)
            active = getattr(userobj, "active", True)
            email = getattr(userobj, "email", "")
            username = getattr(userobj, "username", username)
            firstname = getattr(userobj, "firstname", "")
            lastname = getattr(userobj, "lastname", "")
            extern_type = getattr(userobj, "extern_type", "")

            groups = []

            user_attrs = {
                "username": username,
                "firstname": get_ldap_attr("attr_firstname") or firstname,
                "lastname": get_ldap_attr("attr_lastname") or lastname,
                "groups": groups,
                "user_group_sync": False,
                "email": get_ldap_attr("attr_email") or email,
                "admin": admin,
                "active": active,
                "active_from_extern": None,
                "extern_name": user_dn,
                "extern_type": extern_type,
            }

            log.debug("ldap user: %s", user_attrs)
            log.info(
                "user `%s` authenticated correctly",
                user_attrs["username"],
                extra={"action": "user_auth_ok", "auth_module": "auth_ldap", "username": user_attrs["username"]},
            )

            return user_attrs

        except (LdapUsernameError, LdapPasswordError, LdapImportError):
            log.exception("LDAP related exception")
            return None
        except (Exception,):
            log.exception("Other exception")
            return None


def includeme(config):
    plugin_id = f"egg:rhodecode-enterprise-ce#{RhodeCodeAuthPlugin.uid}"
    plugin_factory(plugin_id).includeme(config)
