matrix-vona/vona/identity/__init__.py

import vona.config as config
import vona.globals as globals
import time
import os

from flask import (
	Blueprint,
	jsonify,
	request,
)

identity = Blueprint("identity", __name__)

# This implements being an identity server.
# I'm pretty sure only Element uses this,
# but oh well.

@identity.route("/_matrix/identity/versions")
async def versions():
	return jsonify({
		"versions": [
			"r0.1.0",
			"r0.2.0",
			"r0.2.1",
			"r0.3.0",
			"v1.1",
			"v1.2",
			"v1.3",
			"v1.4",
			"v1.5"
		]
	})


@identity.route("/_matrix/identity/v2/account")
async def account_info():
	return jsonify({
		"user_id": f"@vona:{config.server_name}"
	})


@identity.route("/_matrix/identity/v2/account/logout", methods=["POST"])
async def logout():
	return jsonify({})


@identity.route("/_matrix/identity/v2/account/register", methods=["POST"])
async def register():
	return jsonify({"token": "vona"})


@identity.route("/_matrix/identity/v2/terms", methods=["GET", "POST"])
async def policies():
	if request.method == "GET":
		return jsonify({
			"policies": {}
		})

	return jsonify({})


@identity.route("/_matrix/identity/v2")
async def status():
	return jsonify({})


@identity.route("/_matrix/identity/v2/pubkey/ephemeral/isvalid")
@identity.route("/_matrix/identity/v2/pubkey/isvalid")
async def pubkey_validity():
	return jsonify({"valid": True})


@identity.route("/_matrix/identity/v2/pubkey/<key>")
async def get_key(key):
	return jsonify({
		"public_key": globals.pubkey()
	})


@identity.route("/_matrix/identity/v2/hash_details")
async def hash_details():
	return jsonify({
		"algorithms": [
			"none",
			"sha256",
		],
		"lookup_pepper": "vona"
	})


@identity.route("/_matrix/identity/v2/lookup", methods=["POST"])
async def lookup():
	req = request.json

	if (
		isinstance(req, dict)
		and "addresses" in req
		and isinstance(req["addresses"], list)
		and len(req["addresses"]) > 0
	):
		return jsonify({
			"mappings": {
				req["addresses"][0]: f"@vona:{config.server_name}"
			}
		})

	return jsonify({
		"errcode": "M_INVALID_PEPPER",
		"error": "Invalid pepper"
	})


@identity.route("/_matrix/identity/v2/validate/email/requestToken", methods=["POST"])
@identity.route("/_matrix/identity/v2/validate/msisdn/requestToken", methods=["POST"])
async def request_validation_token():
	return jsonify({
		"sid": os.urandom(16).hex()
	})


@identity.route("/_matrix/identity/v2/validate/email/submitToken", methods=["GET", "POST"])
@identity.route("/_matrix/identity/v2/validate/msisdn/submitToken", methods=["GET", "POST"])
async def submit_validation_token():
	return jsonify({"success": True})


@identity.route("/_matrix/identity/v2/3pid/bind", methods=["POST"])
async def threepid_bind():
	if "mxid" in request.get_json():
		mxid = request.get_json()["mxid"]
	else:
		mxid = f"@vona:{config.server_name}"

	return jsonify(
		globals.sign_json({
			"address": "abuse@matrix.org",
			"medium": "email",
			"mxid": mxid,
			"not_after": int(time.time() * 1000 + 604800000),
			"not_before": int(time.time() * 1000 - 604800000),
			"ts": int(time.time() * 1000)
		})
	)


@identity.route("/_matrix/identity/v2/3pid/unbind", methods=["POST"])
async def threepid_unbind():
	return jsonify({})


@identity.route("/_matrix/identity/v2/3pid/getValidated3pid")
async def threepid_validated():
	# Please email abuse@matrix.org
	return jsonify({
		"address": "abuse@matrix.org",
		"medium": "email",
		"validated_at": config.the_funny_number
	})


@identity.route("/_matrix/identity/v2/store-invite", methods=["POST"])
async def invite():
	return jsonify({
		"display_name": "Vona",
		"public_keys": [
			{
				"key_validity_url": f"https://{config.server_name}/_matrix/identity/v2/pubkey/isvalid",
				"public_key": "ohyeah"
			},
			{
				"key_validity_url": f"https://{config.server_name}/_matrix/identity/v2/pubkey/ephemeral/isvalid",
				"public_key": "burgerkingfootlettuce"
			}
		],
		"token": "vona"
	})


@identity.route("/_matrix/identity/v2/sign-ed25519", methods=["POST"])
async def invite_signing():
	required_keys = {"mxid", "private_key", "token"}
	d = request.data.get_json()

	if set(d.keys()) == required_keys:
		return jsonify(globals.sign_json(d))
	else:
		return jsonify({
			"errcode": "M_UNRECOGNIZED",
			"error": "Didn't recognize token"
		}), 404