from flask import Flask, jsonify, request, redirect
import vona.globals as globals
from datetime import datetime
import vona.config as config
import threading
import logging
import os
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 as e:
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:
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)