crafty-4/app/classes/web/public_handler.py

import logging
import json
import nh3
from jsonschema import validate
from jsonschema.exceptions import ValidationError

from app.classes.shared.helpers import Helpers
from app.classes.models.users import HelperUsers
from app.classes.web.base_handler import BaseHandler

logger = logging.getLogger(__name__)
auth_log = logging.getLogger("auth")


class PublicHandler(BaseHandler):
    def set_current_user(self, user_id: str = None):
        expire_days = self.helper.get_setting("cookie_expire")

        # if helper comes back with false
        if not expire_days:
            expire_days = "5"

        if user_id is not None:
            self.set_cookie(
                "token",
                self.controller.authentication.generate(user_id),
                expires_days=int(expire_days),
            )
        else:
            self.clear_cookie("token")
            # self.clear_cookie("user")
            # self.clear_cookie("user_data")

    def get(self, page=None):
        # pylint: disable=no-member
        error = nh3.clean(self.get_argument("error", "Invalid Login!"))
        error_msg = nh3.clean(self.get_argument("error_msg", ""))
        # pylint: enable=no-member

        page_data = {
            "version": self.helper.get_version_string(),
            "error": error,
            "lang": self.helper.get_setting("language"),
            "lang_page": self.helper.get_lang_page(self.helper.get_setting("language")),
            "query": "",
            "background": self.controller.cached_login,
            "login_opacity": self.controller.management.get_login_opacity(),
        }

        if self.request.query:
            request_query = self.request.query_arguments.get("next")
            if not request_query:
                self.redirect("/login")
            page_data["query"] = request_query[0].decode()

        # sensible defaults
        template = "public/404.html"

        if page == "login":
            template = "public/login.html"

        elif page == "404":
            template = "public/404.html"

        elif page == "error":
            template = "public/error.html"

        elif page == "offline":
            template = "public/offline.html"

        elif page == "logout":
            exec_user = self.get_current_user()
            self.clear_cookie("token")
            # Delete anti-lockout-user on lockout...it's one time use
            if exec_user[2]["username"] == "anti-lockout-user":
                self.controller.users.stop_anti_lockout()
            # self.clear_cookie("user")
            # self.clear_cookie("user_data")
            self.redirect("/login")
            return

        # if we have no page, let's go to login
        else:
            return self.redirect("/login")

        self.render(
            template,
            data=page_data,
            translate=self.translator.translate,
            error_msg=error_msg,
        )

    def post(self, page=None):
        login_schema = {
            "type": "object",
            "properties": {
                "username": {
                    "type": "string",
                },
                "password": {"type": "string"},
            },
            "required": ["username", "password"],
            "additionalProperties": False,
        }
        try:
            data = json.loads(self.request.body)
        except json.decoder.JSONDecodeError as e:
            logger.error(
                "Invalid JSON schema for API"
                f" login attempt from {self.get_remote_ip()}"
            )
            return self.finish_json(
                400, {"status": "error", "error": "INVALID_JSON", "error_data": str(e)}
            )

        try:
            validate(data, login_schema)
        except ValidationError as e:
            logger.error(
                "Invalid JSON schema for API"
                f" login attempt from {self.get_remote_ip()}"
            )
            return self.finish_json(
                400,
                {
                    "status": "error",
                    "error": "VWggb2ghIFN0aW5reS 🪠",
                    "error_data": str(e),
                },
            )

        page_data = {
            "version": self.helper.get_version_string(),
            "lang": self.helper.get_setting("language"),
            "lang_page": self.helper.get_lang_page(self.helper.get_setting("language")),
            "query": "",
        }
        if self.request.query:
            page_data["query"] = self.request.query_arguments.get("next")[0].decode()

        if page == "login":
            data = json.loads(self.request.body)

            auth_log.info(
                f"User attempting to authenticate from {self.get_remote_ip()}"
            )
            entered_username = nh3.clean(data["username"])  # pylint: disable=no-member
            entered_password = data["password"]

            try:
                user_id = HelperUsers.get_user_id_by_name(entered_username.lower())
                user_data = HelperUsers.get_user_model(user_id)
            except:
                self.controller.log_attempt(self.get_remote_ip(), entered_username)
                auth_log.error(
                    f"User attempted to log into {entered_username}."
                    f" Authentication failed from remote IP {self.get_remote_ip()}"
                    " Users does not exist."
                )
                self.finish_json(
                    403,
                    {
                        "status": "error",
                        "error": self.helper.translation.translate(
                            "login", "incorrect", self.helper.get_setting("language")
                        ),
                    },
                )
                # self.clear_cookie("user")
                # self.clear_cookie("user_data")
                return self.clear_cookie("token")
            # if we don't have a user
            if not user_data:
                auth_log.error(
                    f"User attempted to log into {entered_username}. Authentication"
                    f" failed from remote IP {self.get_remote_ip()}"
                    " User does not exist."
                )
                self.controller.log_attempt(self.get_remote_ip(), entered_username)
                self.finish_json(
                    403,
                    {
                        "status": "error",
                        "error": self.helper.translation.translate(
                            "login", "incorrect", self.helper.get_setting("language")
                        ),
                    },
                )
                # self.clear_cookie("user")
                # self.clear_cookie("user_data")
                return self.clear_cookie("token")

            # if they are disabled
            if not user_data.enabled:
                auth_log.error(
                    f"User attempted to log into {entered_username}. "
                    f"Authentication failed from remote IP {self.get_remote_ip()}."
                    " User account disabled"
                )
                self.controller.log_attempt(self.get_remote_ip(), entered_username)
                self.finish_json(
                    403,
                    {
                        "status": "error",
                        "error": self.helper.translation.translate(
                            "login", "disabled", self.helper.get_setting("language")
                        ),
                    },
                )
                # self.clear_cookie("user")
                # self.clear_cookie("user_data")
                return self.clear_cookie("token")
            login_result = self.helper.verify_pass(entered_password, user_data.password)

            # Valid Login
            if login_result:
                self.set_current_user(user_data.user_id)
                logger.info(
                    f"User: {user_data} Logged in from IP: {self.get_remote_ip()}"
                )
                if not user_data.last_ip and user_data.username == "admin":
                    self.controller.first_login = True
                # record this login
                user_data.last_ip = self.get_remote_ip()
                user_data.last_login = Helpers.get_time_as_string()
                user_data.save()
                auth_log.info(
                    f"{entered_username} successfully"
                    " authenticated and logged"
                    f" into panel from remote IP {self.get_remote_ip()}"
                )
                # log this login
                self.controller.management.add_to_audit_log(
                    user_data.user_id, "Logged in", None, self.get_remote_ip()
                )

                return self.finish_json(
                    200, {"status": "ok", "data": {"message": "login successful!"}}
                )

            # We'll continue on and handle unsuccessful logins
            auth_log.error(
                f"User attempted to log into {entered_username}."
                f" Authentication failed from remote IP {self.get_remote_ip()}"
            )
            self.controller.log_attempt(self.get_remote_ip(), entered_username)
            # self.clear_cookie("user")
            # self.clear_cookie("user_data")
            self.clear_cookie("token")
            error_msg = self.helper.translation.translate(
                "login", "incorrect", self.helper.get_setting("language")
            )
            if entered_password == "app/config/default-creds.txt":
                error_msg += ". "
                error_msg += self.helper.translation.translate(
                    "login", "defaultPath", self.helper.get_setting("language")
                )
            # log this failed login attempt
            self.controller.management.add_to_audit_log(
                user_data.user_id, "Tried to log in", None, self.get_remote_ip()
            )
            return self.finish_json(
                403,
                {"status": "error", "error": error_msg},
            )
        else:
            self.redirect("/login?")