Spotter-VM/basic/srv/vm/mgr/wsgiapp.py

283 lines
13 KiB
Python

# -*- coding: utf-8 -*-
import json
import os
from werkzeug.exceptions import BadRequest, HTTPException, NotFound
from werkzeug.routing import Map, Rule
from werkzeug.utils import redirect
from werkzeug.wrappers import Request, Response
from werkzeug.wsgi import ClosingIterator
from jinja2 import Environment, FileSystemLoader
from . import VMMgr, CERT_PUB_FILE
from . import tools
from .validator import InvalidValueException
from .wsgilang import WSGILang
from .wsgisession import WSGISession
SESSION_KEY = os.urandom(26)
class WSGIApp(object):
def __init__(self):
self.jinja_env = Environment(loader=FileSystemLoader('/srv/vm/templates'), autoescape=True, lstrip_blocks=True, trim_blocks=True)
self.jinja_env.globals.update(is_service_autostarted=tools.is_service_autostarted)
self.jinja_env.globals.update(is_service_started=tools.is_service_started)
def __call__(self, environ, start_response):
return self.wsgi_app(environ, start_response)
def wsgi_app(self, environ, start_response):
request = Request(environ)
# Enhance request
request.mgr = VMMgr()
request.session = WSGISession(request.cookies, SESSION_KEY)
request.session.lang = WSGILang()
# Dispatch request
response = self.dispatch_request(request)
# Save session if changed
request.session.save(response)
return response(environ, start_response)
def dispatch_request(self, request):
adapter = self.get_url_map(request.session).bind_to_environ(request.environ)
try:
endpoint, values = adapter.match()
return getattr(self, endpoint)(request, **values)
except NotFound as e:
# Return custom 404 page
response = self.render_template('404.html', request)
response.status_code = 404
return response
except HTTPException as e:
return e
def get_url_map(self, session):
rules = [
Rule('/', endpoint='portal_view'),
Rule('/login', methods=['GET'], endpoint='login_view'),
Rule('/login', methods=['POST'], endpoint='login_action'),
Rule('/logout', endpoint='logout_action')
]
if session['admin']:
rules += [
Rule('/setup-host', endpoint='setup_host_view'),
Rule('/setup-apps', endpoint='setup_apps_view'),
Rule('/update-host', endpoint='update_host_action'),
Rule('/verify-dns', endpoint='verify_dns_action'),
Rule('/verify-https', endpoint='verify_http_action', defaults={'proto': 'https'}),
Rule('/verify-http', endpoint='verify_http_action', defaults={'proto': 'http'}),
Rule('/update-cert', endpoint='update_cert_action'),
Rule('/update-common', endpoint='update_common_action'),
Rule('/update-app-visibility', endpoint='update_app_visibility_action'),
Rule('/update-app-autostart', endpoint='update_app_autostart_action'),
Rule('/start-app', endpoint='start_app_action'),
Rule('/stop-app', endpoint='stop_app_action'),
Rule('/update-password', endpoint='update_password_action'),
Rule('/shutdown-vm', endpoint='shutdown_vm_action'),
Rule('/reboot-vm', endpoint='reboot_vm_action'),
]
return Map(rules)
def render_template(self, template_name, request, **context):
# Enhance context
context['conf'] = request.mgr.conf
context['session'] = request.session
# Render template
t = self.jinja_env.get_template(template_name)
return Response(t.render(context), mimetype='text/html')
def render_json(self, data):
return Response(json.dumps(data), mimetype='application/json')
def login_view(self, request):
return self.render_template('login.html', request)
def login_action(self, request):
password = request.form['password']
if tools.adminpwd_verify(password, request.mgr.conf['host']['adminpwd']):
request.session['admin'] = True
return redirect('/')
else:
return self.render_template('login.html', request, message=request.session.lang.bad_password())
def logout_action(self, request):
request.session.reset()
return redirect('/')
def portal_view(self, request):
# Default view. If domain is set to the default dummy domain, redirects to first-run setup instead.
if request.session['admin']:
return self.render_template('portal-admin.html', request)
return self.render_template('portal-user.html', request)
def setup_host_view(self, request):
# First-run setup view.
ex_ipv4 = tools.get_external_ipv4()
ex_ipv6 = tools.get_external_ipv6()
in_ipv4 = tools.get_local_ipv4()
in_ipv6 = tools.get_local_ipv6()
is_letsencrypt = os.path.exists('/etc/periodic/daily/acme-sh')
cert_info = tools.get_cert_info(CERT_PUB_FILE)
return self.render_template('setup-host.html', request, ex_ipv4=ex_ipv4, ex_ipv6=ex_ipv6, in_ipv4=in_ipv4, in_ipv6=in_ipv6, is_letsencrypt=is_letsencrypt, cert_info=cert_info)
def setup_apps_view(self, request):
# Application manager view.
return self.render_template('setup-apps.html', request)
def update_host_action(self, request):
# Update domain and port, then restart nginx
try:
domain = request.form['domain']
port = request.form['port']
request.mgr.update_host(domain, port, False)
server_name = request.environ['HTTP_X_FORWARDED_SERVER_NAME']
url = '{}/setup-host'.format(tools.compile_url(server_name, port))
response = self.render_json({'ok': request.session.lang.host_updated(url, url)})
response.call_on_close(tools.restart_nginx)
return response
except BadRequest:
return self.render_json({'error': request.session.lang.malformed_request()})
except InvalidValueException as e:
if e.args[0] == 'domain':
return self.render_json({'error': request.session.lang.invalid_domain(domain)})
if e.args[0] == 'port':
return self.render_json({'error': request.session.lang.invalid_port(port)})
def verify_dns_action(self, request):
# Check if all FQDNs for all applications are resolvable and point to current external IP
mgr = request.mgr
domains = [mgr.domain]+['{}.{}'.format(mgr.conf['apps'][app]['host'], mgr.domain) for app in mgr.conf['apps']]
ipv4 = tools.get_external_ipv4()
ipv6 = tools.get_external_ipv6()
for domain in domains:
try:
a = tools.resolve_ip(domain, 'A')
aaaa = tools.resolve_ip(domain, 'AAAA')
if not a and not aaaa:
return self.render_json({'error': request.session.lang.dns_record_does_not_exist(domain)})
if a and a != ipv4:
return self.render_json({'error': request.session.lang.dns_record_mismatch(domain, a, ipv4)})
if aaaa and aaaa != ipv6:
return self.render_json({'error': request.session.lang.dns_record_mismatch(domain, aaaa, ipv6)})
except:
return self.render_json({'error': request.session.lang.dns_timeout()})
return self.render_json({'ok': request.session.lang.dns_records_ok()})
def verify_http_action(self, request, **kwargs):
# Check if all applications are accessible from the internet using 3rd party ping service
proto = kwargs['proto']
mgr = request.mgr
port = mgr.port if proto == 'https' else '80'
domains = [mgr.domain]+['{}.{}'.format(mgr.conf['apps'][app]['host'], mgr.domain) for app in mgr.conf['apps']]
for domain in domains:
url = tools.compile_url(domain, port, proto)
try:
if not tools.ping_url(url):
return self.render_json({'error': request.session.lang.http_host_not_reachable(url)})
except:
return self.render_json({'error': request.session.lang.http_timeout()})
return self.render_json({'ok': request.session.lang.http_hosts_ok(port)})
def update_cert_action(self, request):
# Update certificate - either request via Let's Encrypt or manually upload files
try:
if request.form['method'] not in ['auto', 'manual']:
raise BadRequest()
if request.form['method'] == 'manual':
if not request.files['public']:
return self.render_json({'error': request.session.lang.cert_file_missing()})
if not request.files['private']:
return self.render_json({'error': request.session.lang.key_file_missing()})
request.files['public'].save('/tmp/public.pem')
request.files['private'].save('/tmp/private.pem')
request.mgr.install_cert('/tmp/public.pem', '/tmp/private.pem')
os.unlink('/tmp/public.pem')
os.unlink('/tmp/private.pem')
else:
request.mgr.request_cert()
except BadRequest:
return self.render_json({'error': request.session.lang.malformed_request()})
except:
return self.render_json({'error': request.session.lang.cert_request_error()})
url = tools.compile_url(request.mgr.domain, request.mgr.port)
return self.render_json({'ok': request.session.lang.cert_installed(url, url)})
def update_common_action(self, request):
# Update common settings shared between apps - admin e-mail address, Google Maps API key
try:
request.mgr.update_common(request.form['email'], request.form['gmaps-api-key'])
except BadRequest:
return self.render_json({'error': request.session.lang.malformed_request()})
return self.render_json({'ok': request.session.lang.common_updated()})
def update_app_visibility_action(self, request):
# Update application visibility on portal page
try:
if request.form['value'] == 'true':
request.mgr.show_tiles(request.form['app'])
else:
request.mgr.hide_tiles(request.form['app'])
except (BadRequest, InvalidValueException):
return self.render_json({'error': request.session.lang.malformed_request()})
return self.render_json({'ok': 'ok'})
def update_app_autostart_action(self, request):
# Update value determining if the app should be automatically started after VM boot
try:
if request.form['value'] == 'true':
request.mgr.enable_autostart(request.form['app'])
else:
request.mgr.disable_autostart(request.form['app'])
except (BadRequest, InvalidValueException):
return self.render_json({'error': request.session.lang.malformed_request()})
return self.render_json({'ok': 'ok'})
def start_app_action(self, request):
# Starts application along with its dependencies
try:
request.mgr.start_app(request.form['app'])
except (BadRequest, InvalidValueException):
return self.render_json({'error': request.session.lang.malformed_request()})
except:
return self.render_json({'error': request.session.lang.stop_start_error()})
return self.render_json({'ok': request.session.lang.app_started()})
def stop_app_action(self, request):
# Stops application along with its dependencies
try:
request.mgr.stop_app(request.form['app'])
except (BadRequest, InvalidValueException):
return self.render_json({'error': request.session.lang.malformed_request()})
except:
return self.render_json({'error': request.session.lang.stop_start_error()})
return self.render_json({'ok': request.session.lang.app_stopped()})
def update_password_action(self, request):
# Updates password for both HDD encryption (LUKS-on-LVM) and admin account to vmmgr
try:
if request.form['newpassword'] != request.form['newpassword2']:
return self.render_json({'error': request.session.lang.password_mismatch()})
if request.form['newpassword'] == '':
return self.render_json({'error': request.session.lang.password_empty()})
# No need to explicitly validate old password, update_luks_password will raise exception if it's wrong
request.mgr.update_password(request.form['oldpassword'], request.form['newpassword'])
except:
return self.render_json({'error': request.session.lang.bad_password()})
return self.render_json({'ok': request.session.lang.password_changed()})
def reboot_vm_action(self, request):
# Reboots VM
response = self.render_json({'ok': request.session.lang.reboot_initiated()})
response.call_on_close(tools.reboot_vm)
return response
def shutdown_vm_action(self, request):
# Shuts down VM
response = self.render_json({'ok': request.session.lang.shutdown_initiated()})
response.call_on_close(tools.shutdown_vm)
return response
class InvalidRecordException(Exception):
pass