import vona.globals as globals
from datetime import datetime
import vona.config as config
import threading
import logging
import os
from flask import (
Flask,
jsonify,
request,
redirect,
abort,
)
from vona.federation import server
from vona.custom import custom
from vona.identity import identity
from vona.appservice import apps
from vona.policy import policy
from vona.client import client
logging.getLogger("werkzeug").disabled = True
logging.getLogger("flask").disabled = True
app = Flask("vona")
app.register_blueprint(identity)
app.register_blueprint(policy)
app.register_blueprint(client)
app.register_blueprint(custom)
app.register_blueprint(server)
app.register_blueprint(apps)
@app.before_request
async def validate_json():
if request.method == "OPTIONS":
return "", 200
elif request.method in ["PUT", "POST", "PATCH"]:
if "media" in request.path:
# Don't check media uploads
return
try:
request.get_json(force=True)
except Exception:
return jsonify({"error": "Content not JSON.", "errcode": "M_NOT_JSON"}), 400
@app.after_request
async def handle_logging(response):
response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
response.headers["Access-Control-Allow-Origin"] = "*"
response.headers["Access-Control-Allow-Headers"] = "*"
response.headers["Access-Control-Allow-Methods"] = "GET, HEAD, POST, PUT, DELETE, OPTIONS"
if request.method == "OPTIONS":
return response
origin = "unknown"
try:
if "Authorization" in request.headers:
if request.headers["Authorization"].split()[0] == "X-Matrix":
origin = (
request.headers["Authorization"]
.split("origin=")[1]
.split(",")[0]
)
while '"' in origin:
origin = origin.replace('"', "")
if origin == config.server_name:
return response
else:
origin = "client"
except Exception:
pass
if request.path.startswith("/.well-known/matrix/"):
return response
print(
f"[{origin}] " +
f'[{datetime.now().strftime("%d/%b/%Y:%H:%M:%S")}] ' +
request.method + " " +
request.full_path.rstrip("?") + " " +
str(response.status_code)
)
return response
# Landing page
@app.route("/")
async def root():
return redirect("/_matrix/static/", 308)
@app.route("/_matrix/static/")
async def matrix_static():
return f'Vona {globals.version} is running
It works! Vona {globals.version} is running
Your Vona server is listening on this port and is ready for messages.
To use this server you\"ll need a Matrix client.
Welcome to the Matrix universe :)
matrix.org
'
# Error handlers
@app.errorhandler(404)
async def not_found(error):
return jsonify({"errcode": "M_UNRECOGNIZED", "error": "Unrecognized request"}), 404
@app.errorhandler(405)
async def invalid_request_method(error):
return jsonify({"errcode": "M_UNRECOGNIZED", "error": "Unrecognized request"}), 405
@app.errorhandler(500)
async def internal_error(error):
return jsonify({"errcode": "M_UNKNOWN", "error": "Internal server error"}), 500
# Well-known endpoints for federation,
# clients, and support information.
@app.route("/.well-known/matrix/server")
async def server():
return jsonify({"m.server": f"{config.server_name}:443"})
@app.route("/.well-known/matrix/support")
async def support():
if config.support:
return jsonify(config.support)
else:
abort(404)
@app.route("/.well-known/matrix/client")
async def client():
return jsonify({
"m.homeserver": {"base_url": f"https://{config.server_name}"},
"m.identity_server": {"base_url": f"https://{config.server_name}"},
})
def federation_self_test():
try:
resp = globals.http_client().get(
path="/",
destination=config.server_name,
)
if str(resp.status_code).startswith("5") or str(resp.status_code).startswith("4"):
print(f"[FATL] Federation self-test failed: status code is not acceptable ({str(resp.status_code)})")
os._exit(1)
print("[INFO] Federation self-test OK")
except Exception as e:
print(f"[FATL] Federation self-test failed: {e}")
os._exit(1)
threading.Thread(target=federation_self_test).start()
app.run(host=config.addr, port=config.port)