unrhodecode/rhodecode/apps/login/views.py

# Copyright (C) 2016-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/

import hashlib
import re
import secrets
import time
import dataclasses
import datetime
import formencode
import formencode.htmlfill
import logging
import urllib.parse
import requests

from pyramid.httpexceptions import HTTPFound

import rhodecode
from rhodecode.apps._base import BaseAppView
from rhodecode.authentication.base import authenticate, HTTP_TYPE
from rhodecode.events import UserRegistered, trigger
from rhodecode.lib import helpers as h
from rhodecode.lib import audit_logger
from rhodecode.lib.auth import AuthUser, HasPermissionAnyDecorator, CSRFRequired, LoginRequired, NotAnonymous
from rhodecode.lib.base import get_ip_addr
from rhodecode.lib.exceptions import UserCreationError
from rhodecode.lib.utils2 import safe_str
from rhodecode.model.db import User
from rhodecode.model.forms import LoginForm, OTPVerifyForm
from rhodecode.model.meta import Session
from rhodecode.model.settings import SettingsModel
from rhodecode.model.user import UserModel
from rhodecode.translation import _


log = logging.getLogger(__name__)


@dataclasses.dataclass
class CaptchaData:
    active: bool
    private_key: str
    public_key: str


def store_user_in_session(session, user_identifier, remember=False):
    user = User.get_by_username_or_primary_email(user_identifier)
    auth_user = AuthUser(user.user_id)
    auth_user.set_authenticated()
    cs = auth_user.get_cookie_store()
    session["rhodecode_user"] = cs
    user.update_lastlogin()
    Session().commit()

    # If they want to be remembered, update the cookie
    if remember and hasattr(session, '_set_cookie_expires'):
        _year = datetime.datetime.now() + datetime.timedelta(seconds=60 * 60 * 24 * 365)
        session._set_cookie_expires(_year)

    if hasattr(session, 'save'):
        session.save()

    safe_cs = cs.copy()
    safe_cs["password"] = "****"
    log.info("user %s is now authenticated and stored in session, session attrs %s", user_identifier, safe_cs)

    # Cookie is set automatically by Pyramid's session response callback
    return None


def get_came_from(request):
    came_from = safe_str(request.GET.get("came_from", ""))
    parsed = urllib.parse.urlparse(came_from)

    allowed_schemes = ["http", "https"]
    default_came_from = h.route_path("home")
    if parsed.scheme and parsed.scheme not in allowed_schemes:
        log.error("Suspicious URL scheme detected %s for url %s", parsed.scheme, parsed)
        came_from = default_came_from
    elif parsed.netloc and request.host != parsed.netloc:
        log.error("Suspicious NETLOC detected %s for url %s server url is: %s", parsed.netloc, parsed, request.host)
        came_from = default_came_from
    elif any(bad_char in came_from for bad_char in ("\r", "\n")):
        log.error("Header injection detected `%s` for url %s server url ", parsed.path, parsed)
        came_from = default_came_from

    return came_from or default_came_from


def _derive_username_from_email(email):
    local_part = email.split("@")[0]
    # sanitize: keep only alphanumeric, dash, underscore, dot
    username = re.sub(r'[^a-zA-Z0-9._-]', '_', local_part).lower()
    if not username:
        username = 'user'
    # ensure unique
    base = username
    suffix = 0
    while User.get_by_username(username):
        suffix += 1
        username = f'{base}_{suffix}'
    return username


class LoginView(BaseAppView):
    def load_default_context(self):
        c = self._get_local_tmpl_context()
        c.came_from = get_came_from(self.request)
        return c

    def _get_captcha_data(self, *args, **kwargs):
        settings = SettingsModel().get_all_settings(*args, **kwargs)
        private_key = settings.get("rhodecode_captcha_private_key", "")
        public_key = settings.get("rhodecode_captcha_public_key", "")
        active = bool(private_key)
        return CaptchaData(active=active, private_key=private_key, public_key=public_key)

    def validate_captcha(self, private_key):
        captcha_rs = self.request.POST.get("g-recaptcha-response")
        url = "https://www.google.com/recaptcha/api/siteverify"
        params = {"secret": private_key, "response": captcha_rs, "remoteip": get_ip_addr(self.request.environ)}
        verify_rs = requests.get(url, params=params, verify=True, timeout=60)
        verify_rs = verify_rs.json()
        captcha_status = verify_rs.get("success", False)
        captcha_errors = verify_rs.get("error-codes", [])
        if not isinstance(captcha_errors, list):
            captcha_errors = [captcha_errors]
        captcha_errors = ", ".join(captcha_errors)
        captcha_message = ""
        if captcha_status is False:
            captcha_message = "Bad captcha. Errors: {}".format(captcha_errors)

        return captcha_status, captcha_message

    def _send_otp_email(self, email, code, purpose):
        from rhodecode.lib.celerylib import tasks, run_task
        from rhodecode.model.notification import EmailNotificationModel

        import rhodecode
        smtp_server = rhodecode.CONFIG.get('smtp_server')
        log_code = code if not smtp_server else 'X' * len(code)
        log.info('[OTP] code=%s email=%s purpose=%s', log_code, email, purpose)

        email_kwargs = {
            'otp_code': code,
            'purpose': purpose,
        }

        (subject, email_body, email_body_plaintext) = EmailNotificationModel().render_email(
            EmailNotificationModel.TYPE_OTP_CODE, **email_kwargs
        )

        run_task(tasks.send_email, [email], subject, email_body_plaintext, email_body)

    # ---- Login Step 1: Email ----

    def login(self):
        c = self.load_default_context()
        auth_user = self._rhodecode_user

        # redirect if already logged in
        if auth_user.is_authenticated and not auth_user.is_default and auth_user.ip_allowed:
            raise HTTPFound(c.came_from)

        # check if we use headers plugin, and try to login using it.
        try:
            log.debug("Running PRE-AUTH for headers based authentication")
            auth_info = authenticate("", "", self.request.environ, HTTP_TYPE, skip_missing=True)
            if auth_info:
                headers = store_user_in_session(self.session, auth_info.get("username"))
                raise HTTPFound(c.came_from, headers=headers)
        except UserCreationError as e:
            log.error(e)
            h.flash(e, category="error")

        return self._get_template_context(c)

    def login_post(self):
        c = self.load_default_context()

        login_form = LoginForm(self.request.translate)()

        try:
            form_result = login_form.to_python(self.request.POST)
            email = form_result["email"]
            remember = form_result["remember"]

            # Throttle to prevent enumeration
            time.sleep(1)

            user = User.get_by_email(email)
            if user and user.active:
                # Existing active user — generate OTP on user object
                code = user.generate_otp("login")
                Session().commit()
                self._send_otp_email(email, code, "login")
                self.session["otp_is_new_user"] = False
            elif user and not user.active:
                # Inactive user — don't send OTP but still redirect (anti-enumeration)
                log.debug("login OTP request for inactive email %s", email)
                self.session["otp_is_new_user"] = False
            else:
                # Unknown email — auto-register if allowed
                default_user = User.get_default_user()
                perms = default_user.AuthUser().permissions["global"]
                registration_allowed = (
                    "hg.register.auto_activate" in perms
                    or "hg.register.manual_activate" in perms
                )

                if registration_allowed:
                    # Generate session-based OTP for new user registration
                    code = ''.join(str(secrets.randbelow(10)) for _ in range(6))
                    code_hash = hashlib.sha256(code.encode('utf-8')).hexdigest()
                    self.session["otp_hash"] = code_hash
                    self.session["otp_created_at"] = time.time()
                    self.session["otp_attempts"] = 0
                    self.session["otp_is_new_user"] = True
                    self._send_otp_email(email, code, "login")
                else:
                    log.debug("login OTP request for unknown email %s, registration disabled", email)
                    self.session["otp_is_new_user"] = False

            # Always store in session and redirect (prevent email enumeration)
            self.session["otp_email"] = email
            self.session["otp_remember"] = remember
            self.session["otp_came_from"] = c.came_from
            if hasattr(self.session, 'save'):
                self.session.save()

            raise HTTPFound(self.request.route_path("login_otp_verify"))

        except formencode.Invalid as errors:
            defaults = errors.value
            render_ctx = {
                "errors": errors.error_dict,
                "defaults": defaults,
            }
            return self._get_template_context(c, **render_ctx)

    # ---- Login Step 2: OTP Verify ----

    def login_otp_verify(self):
        c = self.load_default_context()
        email = self.session.get("otp_email")
        if not email:
            raise HTTPFound(self.request.route_path("login"))
        return self._get_template_context(c,
            verify_route="login_otp_verify",
            otp_digits=6,
            purpose="login",
            otp_email=email,
        )

    def login_otp_verify_post(self):
        c = self.load_default_context()
        email = self.session.get("otp_email")
        is_new_user = self.session.get("otp_is_new_user", False)
        log.debug("OTP verify: email=%s is_new_user=%s session_keys=%s",
                  email, is_new_user, list(self.session.keys()))
        if not email:
            raise HTTPFound(self.request.route_path("login"))

        otp_form = OTPVerifyForm(self.request.translate, digits=6)()

        try:
            form_result = otp_form.to_python(self.request.POST)
            code = form_result["otp"]

            if is_new_user:
                self._verify_new_user_otp(email, code, form_result)
            else:
                self._verify_existing_user_otp(email, code, form_result)

        except formencode.Invalid as errors:
            defaults = errors.value
            render_ctx = {
                "errors": errors.error_dict,
                "defaults": defaults,
                "verify_route": "login_otp_verify",
                "otp_digits": 6,
                "purpose": "login",
                "otp_email": email,
            }
            return self._get_template_context(c, **render_ctx)

    def _verify_existing_user_otp(self, email, code, form_result):
        user = User.get_by_email(email)
        if not user or not user.verify_otp(code, "login"):
            Session().commit()
            error_msg = _("Invalid or expired code. Please try again.")
            raise formencode.Invalid(error_msg, form_result, None,
                                     error_dict={"otp": error_msg})

        Session().commit()

        remember = self.session.get("otp_remember", False)
        came_from = self.session.get("otp_came_from", h.route_path("home"))

        # Clear OTP session data
        self._clear_otp_session()

        headers = store_user_in_session(self.session, user_identifier=email, remember=remember)

        audit_user = audit_logger.UserWrap(
            username=user.username, ip_addr=self.request.remote_addr
        )
        action_data = {"user_agent": self.request.user_agent}
        audit_logger.store_web("user.login.success", action_data=action_data, user=audit_user, commit=True)

        raise HTTPFound(came_from, headers=headers)

    def _verify_new_user_otp(self, email, code, form_result):
        from rhodecode.authentication.plugins import auth_rhodecode

        stored_hash = self.session.get("otp_hash")
        created_at = self.session.get("otp_created_at", 0)
        attempts = self.session.get("otp_attempts", 0)

        if attempts >= 5:
            self._clear_otp_session()
            h.flash(_("Too many attempts. Please start over."), category="error")
            raise HTTPFound(self.request.route_path("login"))

        if (time.time() - created_at) > 300:
            self._clear_otp_session()
            h.flash(_("Code expired. Please start over."), category="error")
            raise HTTPFound(self.request.route_path("login"))

        self.session["otp_attempts"] = attempts + 1
        if hasattr(self.session, 'save'):
            self.session.save()

        code_hash = hashlib.sha256(code.encode('utf-8')).hexdigest()
        if code_hash != stored_hash:
            error_msg = _("Invalid code. Please try again.")
            raise formencode.Invalid(error_msg, form_result, None,
                                     error_dict={"otp": error_msg})

        # OTP verified — create the user account
        auto_active = "hg.register.auto_activate" in User.get_default_user().AuthUser().permissions["global"]
        username = _derive_username_from_email(email)

        # First real user becomes admin (bootstrap)
        system_users = {User.DEFAULT_USER, User.GHOST_USER, User.AI_USER, User.SECURITY_USER}
        real_user_count = User.query().filter(~User.username.in_(system_users)).count()
        is_first_user = real_user_count == 0

        form_data = {
            "username": username,
            "password": secrets.token_hex(32),
            "email": email,
            "firstname": "",
            "lastname": "",
            "active": True if is_first_user else auto_active,
            "admin": is_first_user,
        }

        extern_name = auth_rhodecode.RhodeCodeAuthPlugin.uid
        extern_type = auth_rhodecode.RhodeCodeAuthPlugin.uid

        new_user = UserModel().create_registration(form_data, extern_name=extern_name, extern_type=extern_type)

        action_data = {"data": new_user.get_api_data(), "user_agent": self.request.user_agent}
        audit_user = audit_logger.UserWrap(
            username=new_user.username, user_id=new_user.user_id, ip_addr=self.request.remote_addr
        )
        audit_logger.store_web("user.register", action_data=action_data, user=audit_user)

        event = UserRegistered(user=new_user, session=self.session)
        trigger(event)
        Session().commit()

        # Clean up OTP session data
        self._clear_otp_session()

        if is_first_user or auto_active:
            headers = store_user_in_session(self.session, user_identifier=email, remember=False)
            h.flash(_("Your account has been created. Welcome!"), category="success")
            raise HTTPFound(h.route_path("home"), headers=headers)
        else:
            h.flash(_("Your account has been created and awaits admin approval."), category="success")
            raise HTTPFound(self.request.route_path("login"))

    def _clear_otp_session(self):
        for key in list(self.session.keys()):
            if key.startswith("otp_"):
                self.session.pop(key, None)
        if hasattr(self.session, 'save'):
            self.session.save()

    # ---- Resend OTP ----

    def resend_otp(self):
        email = self.session.get("otp_email")
        if not email:
            return {"status": "error", "message": "No pending verification."}

        # Rate limit: one per 60s
        last_sent = self.session.get("otp_last_sent", 0)
        if time.time() - last_sent < 60:
            return {"status": "error", "message": "Please wait before requesting a new code."}

        is_new_user = self.session.get("otp_is_new_user", False)

        if is_new_user:
            # New user: regenerate OTP stored in session
            code = ''.join(str(secrets.randbelow(10)) for _ in range(6))
            code_hash = hashlib.sha256(code.encode('utf-8')).hexdigest()
            self.session["otp_hash"] = code_hash
            self.session["otp_created_at"] = time.time()
            self.session["otp_attempts"] = 0
        else:
            # Existing user: regenerate on the user object
            user = User.get_by_email(email)
            if user and user.active:
                code = user.generate_otp("login")
                Session().commit()
            else:
                # Fake success to prevent enumeration
                self.session["otp_last_sent"] = time.time()
                if hasattr(self.session, 'save'):
                    self.session.save()
                return {"status": "ok", "message": "Code sent."}

        self._send_otp_email(email, code, "verification")
        self.session["otp_last_sent"] = time.time()
        if hasattr(self.session, 'save'):
            self.session.save()
        return {"status": "ok", "message": "Code sent."}

    # ---- Logout ----

    @CSRFRequired()
    def logout(self):
        auth_user = self._rhodecode_user
        log.info("Deleting session for user: `%s`", auth_user)

        action_data = {"user_agent": self.request.user_agent}
        audit_logger.store_web("user.logout", action_data=action_data, user=auth_user, commit=True)
        self.session.delete()
        return HTTPFound(h.route_path("home"))

    # ---- Registration (redirects to unified login flow) ----

    def register(self, defaults=None, errors=None):
        raise HTTPFound(self.request.route_path("login"))

    # ---- Sudo OTP ----

    @LoginRequired()
    @NotAnonymous()
    def sudo_otp_verify(self):
        c = self.load_default_context()
        return self._get_template_context(c)

    @LoginRequired()
    @NotAnonymous()
    @CSRFRequired()
    def sudo_otp_verify_post(self):
        c = self.load_default_context()
        user = self._rhodecode_db_user

        otp_form = OTPVerifyForm(self.request.translate, digits=8)()

        try:
            form_result = otp_form.to_python(self.request.POST)
            code = form_result["otp"]

            if not user.verify_otp(code, "sudo"):
                Session().commit()
                error_msg = _("Invalid or expired code. Please try again.")
                raise formencode.Invalid(error_msg, form_result, None,
                                         error_dict={"otp": error_msg})

            Session().commit()

            self.session["sudo_otp_verified_at"] = time.time()
            next_url = self.session.pop("sudo_next_url", h.route_path("home"))
            if hasattr(self.session, 'save'):
                self.session.save()

            raise HTTPFound(next_url)

        except formencode.Invalid as errors:
            defaults = errors.value
            render_ctx = {
                "errors": errors.error_dict,
                "defaults": defaults,
            }
            return self._get_template_context(c, **render_ctx)