Verified Commit 41cbe6d2 authored by Štěpán Henek's avatar Štěpán Henek 🌩

config pages splitted to separate modules

parent d84d8f69
......@@ -14,1133 +14,38 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import typing
from datetime import datetime
import base64
import logging
import time
import uuid
from bottle import Bottle, request, template, response
import bottle
from foris import fapi
from foris.common import login
from foris.guide import Workflow
from foris.utils.translators import gettext_dummy as gettext, _
from foris.config_handlers import (
backups, dns, misc, notifications, wan, lan, updater, wifi, networks,
guest, profile, remote, subordinates
)
from foris.utils.translators import _
from foris.utils import login_required, messages, is_safe_redirect
from foris.middleware.bottle_csrf import CSRFPlugin
from foris.utils.routing import reverse
from foris.state import current_state
from .pages.base import ConfigPageMixin, JoinedPages # TODO refactor plugins and remove this import
from .pages.notifications import NotificationsConfigPage
from .pages.password import PasswordConfigPage
from .pages.remote import RemoteConfigPage
from .pages.subordinates import SubordinatesJoinedPage
from .pages.guide import ProfileConfigPage, GuideFinishedPage
from .pages.networks import NetworksConfigPage
from .pages.wan import WanConfigPage
from .pages.time import TimeConfigPage
from .pages.dns import DNSConfigPage
from .pages.lan import LanConfigPage
from .pages.guest import GuestConfigPage
from .pages.wifi import WifiConfigPage
from .pages.maintenance import MaintenanceConfigPage
from .pages.updater import UpdaterConfigPage
from .pages.about import AboutConfigPage
logger = logging.getLogger(__name__)
class BaseConfigPage(object):
menu_order = 50
slug: typing.Optional[str] = None
page_name: typing.Optional[str] = None
userfriendly_title: typing.Optional[str]
menu_title: typing.Optional[str] = None
subpages: typing.Iterable[typing.Type['ConfigPageMixin']] = []
@staticmethod
def get_menu_tag_static(cls):
if current_state.guide.enabled and current_state.guide.current == cls.slug:
return {
"show": True,
"hint": "",
"text": "<i class='fas fa-reply'></i>",
}
else:
return {
"show": False,
"hint": "",
"text": "",
}
@classmethod
def get_menu_tag(cls):
return ConfigPageMixin.get_menu_tag_static(cls)
@staticmethod
def is_visible_static(cls):
if current_state.guide.enabled:
return cls.slug in current_state.guide.steps
return True
@classmethod
def is_visible(cls):
return ConfigPageMixin.is_visible_static(cls)
@staticmethod
def is_enabled_static(cls):
if current_state.guide.enabled:
return cls.slug in current_state.guide.available_tabs
return True
@classmethod
def is_enabled(cls):
return ConfigPageMixin.is_enabled_static(cls)
@classmethod
def subpage_slugs(cls):
return [e.slug for e in cls.subpages]
class ConfigPageMixin(BaseConfigPage):
# page url part /config/<slug>
template = "config/main"
template_type = "simple"
def call_action(self, action):
"""Call config page action.
:param action:
:return: object that can be passed as HTTP response to Bottle
"""
raise bottle.HTTPError(404, "No actions specified for this page.")
def call_ajax_action(self, action):
"""Call AJAX action.
:param action:
:return: dict of picklable AJAX results
"""
raise bottle.HTTPError(404, "No AJAX actions specified for this page.")
def get_page_form(self, form_name: str, data: dict, controller_id: str) -> typing.Tuple[
fapi.ForisAjaxForm, typing.Callable[[dict], typing.Tuple['str', 'str']]
]:
"""Returns appropriate foris form and handler to generate response
"""
raise bottle.HTTPError(404, "No forms specified for this page.")
def call_insecure(self, identifier):
"""Handels insecure request (no login required)
:param namespace: namespace of the storage (e.g. tokens)
:return: object that can be passed as HTTP response to Bottle
"""
raise bottle.HTTPError(404, "No storage specified for this page.")
def default_template(self, **kwargs):
if self.template_type == "jinja2":
page_template = "%s.html.j2" % self.template
kwargs['template_adapter'] = bottle.Jinja2Template
else:
page_template = self.template
return template(
page_template, title=_(kwargs.pop('title', self.userfriendly_title)), **kwargs)
def render(self, **kwargs):
try:
form = getattr(self, "form")
first_section = form.sections[0]
title = first_section.title
description = first_section.description
except (NotImplementedError, AttributeError):
form = None
title = self.userfriendly_title
description = None
return self.default_template(form=form, title=title, description=description, **kwargs)
def save(self, *args, **kwargs):
no_messages = kwargs.pop("no_messages", False)
result = super(ConfigPageMixin, self).save(*args, **kwargs)
if no_messages:
return result
if result:
messages.success(_("Configuration was successfully saved."))
else:
messages.warning(_("There were some errors in your input."))
return result
class JoinedPages(BaseConfigPage):
userfriendly_title = None
class NotificationsConfigPage(ConfigPageMixin):
slug = "notifications"
menu_order = 9
template = "config/notifications"
userfriendly_title = gettext("Notifications")
template_type = "jinja2"
def render(self, **kwargs):
notifications = current_state.backend.perform(
"router_notifications", "list", {"lang": current_state.language}
)["notifications"]
# show only non displayed notifications
kwargs["notifications"] = [e for e in notifications if not e["displayed"]]
return super(NotificationsConfigPage, self).render(**kwargs)
def _action_dismiss_notifications(self):
notification_ids = request.POST.getall("notification_ids[]")
response = current_state.backend.perform(
"router_notifications", "mark_as_displayed", {"ids": notification_ids})
return response["result"], notification_ids
def call_ajax_action(self, action):
if action == "dismiss-notifications":
bottle.response.set_header("Content-Type", "application/json")
res = self._action_dismiss_notifications()
if res[0]:
return {"success": True, "displayedIDs": res[1]}
else:
return {"success": False}
elif action == "list":
notifications = current_state.backend.perform(
"router_notifications", "list", {"lang": current_state.language}
)["notifications"]
return template(
"_notifications.html.j2",
notifications=[e for e in notifications if not e["displayed"]],
template_adapter=bottle.Jinja2Template,
)
raise ValueError("Unknown AJAX action.")
@classmethod
def get_menu_tag(cls):
return {
"show": True if current_state.notification_count else False,
"hint": _("Number of notifications"),
"text": "%d" % current_state.notification_count,
}
class PasswordConfigPage(ConfigPageMixin, misc.PasswordHandler):
slug = "password"
menu_order = 10
template = "config/password"
template_type = "jinja2"
def __init__(self, *args, **kwargs):
super(PasswordConfigPage, self).__init__(change=current_state.password_set, *args, **kwargs)
def save(self, *args, **kwargs):
result = super(PasswordConfigPage, self).save(no_messages=True, *args, **kwargs)
wrong_old_password = self.form.callback_results.get('wrong_old_password', False)
system_password_no_error = self.form.callback_results.get('system_password_no_error', None)
foris_password_no_error = self.form.callback_results.get('foris_password_no_error', None)
compromised = self.form.callback_results.get("compromised")
if compromised:
messages.error(
_(
"The password you've entered has been compromised. "
"It appears %(count)d times in '%(list)s' list."
) % dict(count=compromised['count'], list=compromised['list'])
)
return result
if wrong_old_password:
messages.error(_("Old password you entered was not valid."))
return result
if system_password_no_error is not None:
if system_password_no_error:
messages.success(_("System password was successfully saved."))
else:
messages.error(_("Failed to save system password."))
if foris_password_no_error is not None:
if foris_password_no_error:
messages.success(_("Foris password was successfully saved."))
else:
messages.error(_("Failed to save Foris password."))
return result
class RemoteConfigPage(ConfigPageMixin, remote.RemoteHandler):
slug = "remote"
menu_order = 11
TOKEN_LINK_EXPIRATION = 30
token_links = {}
template = "config/remote"
userfriendly_title = gettext("Remote Access")
template_type = "jinja2"
@classmethod
def token_cleanup(cls):
now = time.time()
cls.token_links = {k: v for k, v in cls.token_links.items() if now <= v["expiration"]}
def render(self, **kwargs):
data = current_state.backend.perform("remote", "get_status")
kwargs["status"] = data["status"]
kwargs["tokens"] = data["tokens"]
kwargs["backend_data"] = self.backend_data
kwargs["revoke_token_form"] = self.get_token_id_form()
kwargs["generate_token_form"] = self.get_generate_token_form()
kwargs["download_token_form"] = self.get_token_id_form()
return super().render(**kwargs)
def save(self, *args, **kwargs):
kwargs["no_messages"] = True
result = super().save(*args, **kwargs)
if self.form.callback_results["enabled"]:
if self.form.callback_results["result"]:
messages.success(_("Remote access was sucessfully enabled."))
else:
messages.error(
_(
"Failed to enable the remote access. You are probably using "
"a message bus which doesn't support the remote access or "
"the CA for remote access hasn't been generated yet."
)
)
else:
if self.form.callback_results["result"]:
messages.success(_("Remote access was sucessfully disabled."))
else:
messages.error(_("Failed to disable remote access."))
return result
def _check_post(self):
if bottle.request.method != 'POST':
messages.error(_("Wrong HTTP method."))
bottle.redirect(reverse("config_page", page_name="remote"))
def _ajax_list_tokens(self):
data = current_state.backend.perform("remote", "get_status")
return template(
"config/_remote_tokens.html.j2",
tokens=data["tokens"],
template_adapter=bottle.Jinja2Template,
)
def _ajax_revoke_token(self):
self._check_post()
form = self.get_token_id_form(bottle.request.POST.decode())
if not form.data["token_id"]:
raise bottle.HTTPError(404, "token_id not found")
bottle.response.set_header("Content-Type", "application/json")
return current_state.backend.perform(
"remote", "revoke", {"id": form.data["token_id"]})
def _ajax_generate_token(self):
self._check_post()
form = self.get_generate_token_form(bottle.request.POST.decode())
if not form.data["name"]:
raise bottle.HTTPError(404, "name not found")
bottle.response.set_header("Content-Type", "application/json")
return current_state.backend.perform(
"remote", "generate_token", {"name": form.data["name"]})
def _ajax_prepare_token(self):
self._check_post()
RemoteConfigPage.token_cleanup()
form = self.get_token_id_form(bottle.request.POST.decode())
token_id = form.data.get("token_id")
if not token_id:
raise bottle.HTTPError(404, "id not found")
name = form.data.get("name", token_id)
res = current_state.backend.perform("remote", "get_token", {"id": form.data["token_id"]})
if res["status"] != "valid":
raise bottle.HTTPError(404, "token not found")
bottle.response.set_header("Content-Type", "application/json")
new_uuid = uuid.uuid4()
RemoteConfigPage.token_links[str(new_uuid)] = {
"expiration": time.time() + RemoteConfigPage.TOKEN_LINK_EXPIRATION,
"name": name,
"token": base64.b64decode(res["token"]),
}
return {
"url": reverse("config_insecure", page_name="remote", identifier=str(new_uuid)),
"expires_in": RemoteConfigPage.TOKEN_LINK_EXPIRATION,
}
def call_ajax_action(self, action):
if action == "generate-token":
return self._ajax_generate_token()
elif action == "revoke-token":
return self._ajax_revoke_token()
elif action == "list-tokens":
return self._ajax_list_tokens()
elif action == "prepare-token":
return self._ajax_prepare_token()
raise ValueError("Unknown AJAX action.")
def _action_generate_ca(self):
self._check_post()
messages.info(_("Starting to generate CA for remote access."))
current_state.backend.perform("remote", "generate_ca")
# don't need to handle async_id (should influence all clients)
bottle.redirect(reverse("config_page", page_name="remote"))
def _action_delete_ca(self):
self._check_post()
data = current_state.backend.perform("remote", "delete_ca")
if data["result"]:
messages.success(_("CA for remote access was sucessfully deleted."))
else:
messages.error(_("Failed to delete CA for remote access."))
bottle.redirect(reverse("config_page", page_name="remote"))
def call_insecure(self, identifier):
RemoteConfigPage.token_cleanup()
try:
record = RemoteConfigPage.token_links[identifier]
except KeyError:
raise bottle.HTTPError(404, "token url doesn't exists")
bottle.response.set_header("Content-Type", "application/x-gzip")
bottle.response.set_header(
"Content-Disposition", 'attachment; filename="token-%s.tar.gz"' % record["name"]
)
bottle.response.set_header("Content-Length", len(record["token"]))
return record["token"]
def call_action(self, action):
if action == "generate-ca":
self._action_generate_ca()
elif action == "delete-ca":
self._action_delete_ca()
elif action == "download-token":
self._action_download_token()
raise ValueError("Unknown action.")
@classmethod
def is_visible(cls):
if current_state.backend.name != "mqtt":
return False
return ConfigPageMixin.is_visible_static(cls)
@classmethod
def is_enabled(cls):
if current_state.backend.name != "mqtt":
return False
return ConfigPageMixin.is_enabled_static(cls)
class SubordinatesSetupPage(ConfigPageMixin, subordinates.SubordinatesConfigHandler):
slug = "subordinates"
menu_order = 1 # submenu
template = "config/subordinates_setup"
menu_title = gettext("Set up")
userfriendly_title = gettext("Managed devices: Set up")
template_type = "jinja2"
def render(self, **kwargs):
data = current_state.backend.perform("subordinates", "list")
kwargs["subordinates"] = data["subordinates"]
return super().render(**kwargs)
def save(self, *args, **kwargs):
super(SubordinatesSetupPage, self).save(no_messages=True, *args, **kwargs)
data = self.form.callback_results
if data["result"]:
messages.success(_(
"Token was successfully added and client '%(controller_id)s' "
"should be visible in a moment."
) % dict(controller_id=data["controller_id"]))
else:
messages.error(_("Failed to add token."))
return data["result"]
def _check_and_get_controller_id(self):
if bottle.request.method != 'POST':
messages.error(_("Wrong HTTP method."))
bottle.redirect(reverse("config_page", page_name="remote"))
form = self.get_controller_id_form(bottle.request.POST.decode())
if not form.data["controller_id"]:
raise bottle.HTTPError(404, "controller_id not found")
return form.data["controller_id"]
def _ajax_list_subordinates(self):
data = current_state.backend.perform("subordinates", "list")
return template(
"config/_subordinates_list_setup.html.j2",
subordinates=data["subordinates"],
template_adapter=bottle.Jinja2Template,
)
def _ajax_delete(self):
controller_id = self._check_and_get_controller_id()
res = current_state.backend.perform(
"subordinates", "del", {"controller_id": controller_id})
if res["result"]:
return template(
"config/_subordinates_message.html.j2",
message={
"classes": ['success'],
"text": _("Subordinate '%(controller_id)s' was successfully deleted.")
% dict(controller_id=controller_id)
},
template_adapter=bottle.Jinja2Template,
)
else:
return template(
"config/_subordinates_message.html.j2",
message={
"classes": ['error'],
"text": _("Failed to delete subordinate '%(controller_id)s'.")
% dict(controller_id=controller_id)
},
template_adapter=bottle.Jinja2Template,
)
def _ajax_set_enabled(self, enabled):
controller_id = self._check_and_get_controller_id()
res = current_state.backend.perform("subordinates", "set_enabled", {
"controller_id": controller_id,
"enabled": enabled,
})
if res["result"]:
if enabled:
message = {
"classes": ['success'],
"text": _("Subordinate '%(controller_id)s' was sucessfuly enabled.")
% dict(controller_id=controller_id)
}
else:
message = {
"classes": ['success'],
"text": _("Subordinate '%(controller_id)s' was sucessfuly disabled.")
% dict(controller_id=controller_id)
}
else:
if enabled:
message = {
"classes": ['error'],
"text": _("Failed to enable subordinate '%(controller_id)s'.")
% dict(controller_id=controller_id)
}
else:
message = {
"classes": ['error'],
"text": _("Failed to disable subordinate '%(controller_id)s'.")
% dict(controller_id=controller_id)
}
return template(
"config/_subordinates_message.html.j2",
message=message,
template_adapter=bottle.Jinja2Template,
)
def call_ajax_action(self, action):
if action == "list":
return self._ajax_list_subordinates()
elif action == "disable":
return self._ajax_set_enabled(False)
elif action == "enable":
return self._ajax_set_enabled(True)
elif action == "delete":
return self._ajax_delete()
raise ValueError("Unknown AJAX action.")
@classmethod
def is_visible(cls):
if current_state.backend.name != "mqtt":
return False
return ConfigPageMixin.is_visible_static(cls)
@classmethod
def is_enabled(cls):
if current_state.backend.name != "mqtt":
return False
return ConfigPageMixin.is_enabled_static(cls)
def get_page_form(self, form_name: str, data: dict, controller_id: str) -> typing.Tuple[
fapi.ForisAjaxForm, typing.Callable[[dict], typing.Tuple['str', 'str']]
]:
"""Returns appropriate foris form and handler to generate response
"""
form: fapi.ForisAjaxForm
if form_name == "sub-form":
form = subordinates.SubordinatesEditForm(data)
def prepare_message(results: dict) -> dict:
if results["result"]:
message = {
"classes": ['success'],
"text": _("Device '%(controller_id)s' was sucessfully updated.")
% dict(controller_id=data["controller_id"])
}
else:
message = {
"classes": ['error'],
"text": _("Failed to update subordinate '%(controller_id)s'.")
% dict(controller_id=data["controller_id"])
}
return message
form.url = reverse("config_ajax_form", page_name="subordinates", form_name="sub-form")
return form, prepare_message
elif form_name == "subsub-form":
form = subordinates.SubsubordinatesEditForm(data)
def prepare_message(results: dict) -> dict:
if results["result"]:
message = {
"classes": ['success'],
"text": _("Subsubordinate '%(controller_id)s' was sucessfully updated.")
% dict(controller_id=data["controller_id"])
}
else:
message = {
"classes": ['error'],
"text": _("Failed to update subsubordinate '%(controller_id)s'.")
% dict(controller_id=data["controller_id"])
}
return message
form.url = reverse(
"config_ajax_form", page_name="subordinates", form_name="subsub-form")
return form, prepare_message
raise bottle.HTTPError(404, "No form '%s' not found." % form_name)
class SubordinatesWifiPage(ConfigPageMixin):
slug = "subordinates-wifi"
menu_order = 2 # submenu
template = "config/subordinates_wifi"
menu_title = gettext("Wi-Fi")
userfriendly_title = gettext("Managed devices: Wi-Fi")
template_type = "jinja2"
def render(self, **kwargs):
data = current_state.backend.perform("subordinates", "list")
kwargs["subordinates"] = data["subordinates"]
return super().render(**kwargs)
@classmethod
def is_visible(cls):
if current_state.backend.name != "mqtt":
return False
return ConfigPageMixin.is_visible_static(cls)
@classmethod
def is_enabled(cls):
if current_state.backend.name != "mqtt":
return False
return ConfigPageMixin.is_enabled_static(cls)
def _ajax_list_subordinates(self):
data = current_state.backend.perform("subordinates", "list")
return template(
"config/_subordinates_list_wifi.html.j2",
subordinates=data["subordinates"],
template_adapter=bottle.Jinja2Template,
)
def call_ajax_action(self, action):
if action == "list":
return self._ajax_list_subordinates()
raise ValueError("Unknown AJAX action.")
def get_page_form(self, form_name: str, data: dict, controller_id: str) -> typing.Tuple[
fapi.ForisAjaxForm, typing.Callable[[dict], typing.Tuple['str', 'str']]
]:
"""Returns appropriate foris form and handler to generate response
"""
if form_name == "wifi-form":
form = wifi.WifiEditForm(data, controller_id=controller_id)
def prepare_message(results: dict) -> dict:
if results["result"]:
message = {
"classes": ['success'],
"text": _("Wifi settings was sucessfully updated.")
}
else:
message = {
"classes": ['error'],
"text": _("Failed to update Wifi settings.")
}
return message
form.url = reverse("config_ajax_form", page_name="subordinates-wifi", form_name="wifi-form")
return form, prepare_message
raise bottle.HTTPError(404, "No form '%s' not found." % form_name)
class SubordinatesJoinedPage(JoinedPages):
userfriendly_title = gettext("Managed devices")
name = "subordinates-main"
subpages: typing.Iterable[typing.Type['ConfigPageMixin']] = [
SubordinatesSetupPage,
SubordinatesWifiPage,
]