mailinabox/management/daemon.py

272 lines
8.5 KiB
Python
Raw Normal View History

#!/usr/bin/python3
import os, os.path, re, json
2014-08-17 22:43:57 +00:00
from functools import wraps
from flask import Flask, request, render_template, abort, Response
import auth, utils
2014-08-17 22:43:57 +00:00
from mailconfig import get_mail_users, add_mail_user, set_mail_password, remove_mail_user, get_archived_mail_users
from mailconfig import get_mail_user_privileges, add_remove_mail_user_privilege
from mailconfig import get_mail_aliases, get_mail_domains, add_mail_alias, remove_mail_alias
2014-06-03 20:21:17 +00:00
env = utils.load_environment()
2014-06-22 12:55:19 +00:00
auth_service = auth.KeyAuthService()
2014-08-17 22:43:57 +00:00
# We may deploy via a symbolic link, which confuses flask's template finding.
me = __file__
try:
me = os.readlink(__file__)
except OSError:
pass
app = Flask(__name__, template_folder=os.path.abspath(os.path.join(os.path.dirname(me), "templates")))
# Decorator to protect views that require authentication.
def authorized_personnel_only(viewfunc):
@wraps(viewfunc)
def newview(*args, **kwargs):
# Check if the user is authorized.
authorized_status = auth_service.is_authenticated(request, env)
if authorized_status == "OK":
# Authorized. Call view func.
return viewfunc(*args, **kwargs)
# Not authorized. Return a 401 (send auth) and a prompt to authorize by default.
status = 401
headers = { 'WWW-Authenticate': 'Basic realm="{0}"'.format(auth_service.auth_realm) }
if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
# Don't issue a 401 to an AJAX request because the user will
# be prompted for credentials, which is not helpful.
status = 403
headers = None
if request.headers.get('Accept') in (None, "", "*/*"):
# Return plain text output.
return Response(authorized_status+"\n", status=status, mimetype='text/plain', headers=headers)
else:
# Return JSON output.
return Response(json.dumps({
"status": "error",
"reason": authorized_status
}+"\n"), status=status, mimetype='application/json', headers=headers)
return newview
@app.errorhandler(401)
def unauthorized(error):
return auth_service.make_unauthorized_response()
2014-08-17 22:43:57 +00:00
def json_response(data):
return Response(json.dumps(data), status=200, mimetype='application/json')
###################################
# Control Panel (unauthenticated views)
@app.route('/')
def index():
2014-08-17 22:43:57 +00:00
# Render the control panel. This route does not require user authentication
# so it must be safe!
no_admins_exist = (len([user for user in get_mail_users(env, as_json=True) if "admin" in user['privileges']]) == 0)
2014-08-17 22:43:57 +00:00
return render_template('index.html',
hostname=env['PRIMARY_HOSTNAME'],
no_admins_exist=no_admins_exist,
2014-08-17 22:43:57 +00:00
)
@app.route('/me')
def me():
# Is the caller authorized?
authorized_status = auth_service.is_authenticated(request, env)
if authorized_status != "OK":
return json_response({
"status": "not-authorized",
"reason": authorized_status,
})
return json_response({
"status": "authorized",
"api_key": auth_service.key,
})
# MAIL
@app.route('/mail/users')
2014-08-17 22:43:57 +00:00
@authorized_personnel_only
def mail_users():
if request.args.get("format", "") == "json":
2014-08-17 22:43:57 +00:00
return json_response(get_mail_users(env, as_json=True) + get_archived_mail_users(env))
else:
return "".join(x+"\n" for x in get_mail_users(env))
@app.route('/mail/users/add', methods=['POST'])
2014-08-17 22:43:57 +00:00
@authorized_personnel_only
def mail_users_add():
2014-08-17 22:43:57 +00:00
return add_mail_user(request.form.get('email', ''), request.form.get('password', ''), request.form.get('privileges', ''), env)
@app.route('/mail/users/password', methods=['POST'])
2014-08-17 22:43:57 +00:00
@authorized_personnel_only
def mail_users_password():
return set_mail_password(request.form.get('email', ''), request.form.get('password', ''), env)
@app.route('/mail/users/remove', methods=['POST'])
2014-08-17 22:43:57 +00:00
@authorized_personnel_only
def mail_users_remove():
return remove_mail_user(request.form.get('email', ''), env)
@app.route('/mail/users/privileges')
2014-08-17 22:43:57 +00:00
@authorized_personnel_only
def mail_user_privs():
privs = get_mail_user_privileges(request.args.get('email', ''), env)
if isinstance(privs, tuple): return privs # error
return "\n".join(privs)
@app.route('/mail/users/privileges/add', methods=['POST'])
2014-08-17 22:43:57 +00:00
@authorized_personnel_only
def mail_user_privs_add():
return add_remove_mail_user_privilege(request.form.get('email', ''), request.form.get('privilege', ''), "add", env)
@app.route('/mail/users/privileges/remove', methods=['POST'])
2014-08-17 22:43:57 +00:00
@authorized_personnel_only
def mail_user_privs_remove():
return add_remove_mail_user_privilege(request.form.get('email', ''), request.form.get('privilege', ''), "remove", env)
@app.route('/mail/aliases')
2014-08-17 22:43:57 +00:00
@authorized_personnel_only
def mail_aliases():
2014-08-17 22:43:57 +00:00
if request.args.get("format", "") == "json":
return json_response(get_mail_aliases(env, as_json=True))
else:
return "".join(x+"\t"+y+"\n" for x, y in get_mail_aliases(env))
@app.route('/mail/aliases/add', methods=['POST'])
2014-08-17 22:43:57 +00:00
@authorized_personnel_only
def mail_aliases_add():
2014-08-17 22:43:57 +00:00
return add_mail_alias(
request.form.get('source', ''),
request.form.get('destination', ''),
env,
update_if_exists=(request.form.get('update_if_exists', '') == '1')
)
@app.route('/mail/aliases/remove', methods=['POST'])
2014-08-17 22:43:57 +00:00
@authorized_personnel_only
def mail_aliases_remove():
return remove_mail_alias(request.form.get('source', ''), env)
@app.route('/mail/domains')
2014-08-17 22:43:57 +00:00
@authorized_personnel_only
def mail_domains():
return "".join(x+"\n" for x in get_mail_domains(env))
# DNS
@app.route('/dns/update', methods=['POST'])
2014-08-17 22:43:57 +00:00
@authorized_personnel_only
def dns_update():
from dns_update import do_dns_update
try:
2014-08-01 12:05:34 +00:00
return do_dns_update(env, force=request.form.get('force', '') == '1')
except Exception as e:
return (str(e), 500)
@app.route('/dns/set/<qname>', methods=['POST'])
@app.route('/dns/set/<qname>/<rtype>', methods=['POST'])
@app.route('/dns/set/<qname>/<rtype>/<value>', methods=['POST'])
@authorized_personnel_only
def dns_set_record(qname, rtype="A", value=None):
from dns_update import do_dns_update, set_custom_dns_record
try:
# Get the value from the URL, then the POST parameters, or if it is not set then
# use the remote IP address of the request --- makes dynamic DNS easy. To clear a
# value, '' must be explicitly passed.
print(request.environ)
if value is None:
value = request.form.get("value")
if value is None:
value = request.environ.get("HTTP_X_FORWARDED_FOR") # normally REMOTE_ADDR but we're behind nginx as a reverse proxy
if value == '':
# request deletion
value = None
if set_custom_dns_record(qname, rtype, value, env):
return do_dns_update(env)
return "OK"
except ValueError as e:
return (str(e), 400)
2014-08-17 22:43:57 +00:00
@app.route('/dns/dump')
@authorized_personnel_only
def dns_get_dump():
from dns_update import build_recommended_dns
return json_response(build_recommended_dns(env))
# WEB
@app.route('/web/update', methods=['POST'])
2014-08-17 22:43:57 +00:00
@authorized_personnel_only
def web_update():
from web_update import do_web_update
return do_web_update(env)
# System
2014-08-17 22:43:57 +00:00
@app.route('/system/status', methods=["POST"])
@authorized_personnel_only
def system_status():
from status_checks import run_checks
2014-08-17 22:43:57 +00:00
class WebOutput:
def __init__(self):
self.items = []
def add_heading(self, heading):
self.items.append({ "type": "heading", "text": heading, "extra": [] })
def print_ok(self, message):
self.items.append({ "type": "ok", "text": message, "extra": [] })
def print_error(self, message):
self.items.append({ "type": "error", "text": message, "extra": [] })
def print_line(self, message, monospace=False):
self.items[-1]["extra"].append({ "text": message, "monospace": monospace })
output = WebOutput()
run_checks(env, output)
return json_response(output.items)
@app.route('/system/updates')
2014-08-17 22:43:57 +00:00
@authorized_personnel_only
def show_updates():
from status_checks import list_apt_updates
return "".join(
"%s (%s)\n"
% (p["package"], p["version"])
for p in list_apt_updates())
@app.route('/system/update-packages', methods=["POST"])
2014-08-17 22:43:57 +00:00
@authorized_personnel_only
def do_updates():
utils.shell("check_call", ["/usr/bin/apt-get", "-qq", "update"])
return utils.shell("check_output", ["/usr/bin/apt-get", "-y", "upgrade"], env={
"DEBIAN_FRONTEND": "noninteractive"
})
# APP
if __name__ == '__main__':
if "DEBUG" in os.environ: app.debug = True
2014-08-17 22:43:57 +00:00
if "APIKEY" in os.environ: auth_service.key = os.environ["APIKEY"]
if not app.debug:
app.logger.addHandler(utils.create_syslog_handler())
# For testing on the command line, you can use `curl` like so:
# curl --user $(</var/lib/mailinabox/api.key): http://localhost:10222/mail/users
auth_service.write_key()
# For testing in the browser, you can copy the API key that's output to the
# debug console and enter that as the username
app.logger.info('API key: ' + auth_service.key)
app.run(port=10222)