Make ty happy

vona/client/admin.py

@@ -116,7 +116,8 @@ async def evacuate_room(room):
 	req = request.json
 
 	if (
-		"background" in req
+		isinstance(req, dict)
+		and "background" in req
 		and isinstance(req["background"], bool)
 	):
 		background = req["background"]

vona/config/__init__.py

@@ -36,6 +36,10 @@ def _load_toml(path: Path) -> dict:
 	except tomllib.TOMLDecodeError as e:
 		_fatal(f"Invalid TOML configuration: {e}")
 
+	# This will never be reached. It is here 
+	# specifically just to please ty
+	return {}
+
 
 def _read_signing_key_from_path(path_value) -> str | None:
 	p = Path(path_value)
@@ -57,7 +61,7 @@ def _validate_cat_path(cat_path: str) -> Path:
 
 
 def _apply_config(cfg: dict) -> None:
-	global addr, port, server_name, signing_key, cat, support, users_can_register, db
+	global addr, port, server_name, signing_key, cat, support, users_can_register
 
 	if "address" in cfg:
 		addr = str(cfg["address"])

vona/custom/citadel.py

@@ -30,7 +30,16 @@ async def news_stats(event):
 
 @citadel.route("/_matrix/client/r0/citadel/rooms/<room>/closeRoom", methods=["POST"])
 async def close_room(room):
-	store_response = request.json.get("store_response", True)
+	req = request.json
+
+	if (
+		isinstance(req, dict)
+		and "store_response" in req
+		and isinstance(req["store_response"], bool)
+	):
+		store_response = req["store_response"]
+	else:
+		store_response = True
 
 	operation_id = base64.b64encode(
 		bytes(

vona/federation/__init__.py

@@ -248,8 +248,19 @@ async def user_devices(user):
 
 @server.route("/_matrix/federation/v1/user/keys/query", methods=["POST"])
 async def user_keys():
+	req = request.json
+
+	if (
+		isinstance(req, dict)
+		and "device_keys" in req
+		and isinstance(req["device_keys"], dict)
+	):
+		device_keys = req["device_keys"]
+	else:
+		device_keys = {}
+
 	return jsonify({
-		"device_keys": request.json.get("device_keys", {})
+		"device_keys": device_keys
 	})
 
 
@@ -257,7 +268,10 @@ async def user_keys():
 async def invite_user_v2(room, txnId):
 	invite_data = request.json
 
-	if "event" in invite_data:
+	if (
+		isinstance(invite_data, dict)
+		and "event" in invite_data
+	):
 		if "room_version" in invite_data:
 			if invite_data["room_version"] not in [str(i) for i in range(1, 10)]:
 				return jsonify({
@@ -292,10 +306,19 @@ async def invite_user_v2(room, txnId):
 @server.route("/_matrix/federation/v1/invite/<room>/<txnId>", methods=["PUT"])
 async def invite_user_v1(room, txnId):
 	event = request.json
-	content = event.get("content", {})
+	if (
+		isinstance(event, dict)
+		and "content" in event
+		and isinstance(event["content"], dict)
+	):
+		content = event["content"]
+	else:
+		content = {}
 
 	if (
-		"content" in event
+		isinstance(event, dict)
+		and isinstance(content, dict)
+		and "content" in event
 		and "membership" in content
 		and "state_key" in event
 		and "room_id" in event
@@ -325,7 +348,12 @@ async def space_hierachy(room):
 @server.route("/_matrix/federation/v1/org.matrix.msc4358/discover_common_rooms", methods=["POST"])
 @server.route("/_matrix/federation/v1/discover_common_rooms", methods=["POST"])
 async def discover_common_rooms():
-	tags = request.json.get("room_participation_tags", [])
+	req = request.json
+	if isinstance(req, dict):
+		tags = req.get("room_participation_tags", [])
+	else:
+		tags = []
+
 	return jsonify({"recognised_tags": tags})
 
 

vona/globals/__init__.py

@@ -1,7 +1,9 @@
 from resolvematrix import ServerResolver
 from types import SimpleNamespace
 from collections import Counter
+
 import vona.config as config
+import nacl.encoding
 import nacl.signing
 import time as ti
 import hashlib
@@ -15,7 +17,7 @@ import re
 version = "1.5.0"
 
 
-def canonical_json(value):
+def canonical_json(value: dict | list) -> bytes:
 	return json.dumps(
 		value,
 		ensure_ascii=False,
@@ -24,7 +26,7 @@ def canonical_json(value):
 	).encode("UTF-8")
 
 
-def sign_json(data):
+def sign_json(data: dict) -> dict:
 	parts = config.signing_key.split()
 	base64_key = parts[2]
 
@@ -53,7 +55,7 @@ def sign_json(data):
 	return signed_json
 
 
-def sign_json_without_discard(data):
+def sign_json_without_discard(data: dict) -> dict:
 	parts = config.signing_key.split()
 	base64_key = parts[2]
 
@@ -86,7 +88,7 @@ def sign_json_without_discard(data):
 	return data
 
 
-def make_event_id(seed=None):
+def make_event_id(seed: str | int | None = None) -> str:
 	if seed is not None:
 		random.seed(seed)
 
@@ -104,7 +106,7 @@ def make_event_id(seed=None):
 	return event_id
 
 
-def event_hash(event_object):
+def event_hash(event_object: dict) -> str:
 	event_object = dict(event_object)
 
 	event_object.pop("unsigned", None)
@@ -174,7 +176,7 @@ def redact_event(
 	event: dict,
 	for_event_id: bool = False,
 	room_ver: int = 1,
-):
+) -> dict:
 	# Returns a redacted event as per
 	# the algorithm for v1 to v11 rooms.
 
@@ -272,7 +274,7 @@ def redact_event(
 def hash_and_sign_event(
 	event_object: dict,
 	room_ver: int = 1,
-):
+) -> dict:
 	content_hash = event_hash(event_object)
 	event_object["hashes"] = {"sha256": content_hash}
 	stripped_object = redact_event(
@@ -288,7 +290,7 @@ def hash_and_sign_event(
 def make_ref_hash(
 	event: dict,
 	room_ver: int = 3,
-):
+) -> str:
 	stripped = redact_event(
 		event=event,
 		for_event_id=True,
@@ -309,7 +311,7 @@ def make_ref_hash(
 	return "$" + evt_hash
 
 
-def room_version_from_id(room):
+def room_version_from_id(room) -> str:
 	room_id_no_sigil = (
 		room
 		.replace("!", "")

vona/identity/__init__.py

@@ -1,5 +1,7 @@
-from vona.config import server_name, the_funny_number
+import vona.config as config
+import vona.globals as globals
 import time
+import os
 
 from flask import (
 	Blueprint,
@@ -13,10 +15,8 @@ identity = Blueprint("identity", __name__)
 # I'm pretty sure only Element uses this,
 # but oh well.
 
-# https://spec.matrix.org/latest/identity-service-api/#api-version-check
 @identity.route("/_matrix/identity/versions")
 async def versions():
-	# Stolen from the vector.im identity server
 	return jsonify({
 		"versions": [
 			"r0.1.0",
@@ -32,10 +32,11 @@ async def versions():
 	})
 
 
-# https://spec.matrix.org/latest/identity-service-api/#authentication
 @identity.route("/_matrix/identity/v2/account")
 async def account_info():
-	return jsonify({"user_id": f"@vona:{server_name}"})
+	return jsonify({
+		"user_id": f"@vona:{config.server_name}"
+	})
 
 
 @identity.route("/_matrix/identity/v2/account/logout", methods=["POST"])
@@ -45,98 +46,129 @@ async def logout():
 
 @identity.route("/_matrix/identity/v2/account/register", methods=["POST"])
 async def register():
-	return jsonify({"token":"vona"})
+	return jsonify({"token": "vona"})
 
 
-# https://spec.matrix.org/latest/identity-service-api/#terms-of-service
 @identity.route("/_matrix/identity/v2/terms", methods=["GET", "POST"])
 async def policies():
 	if request.method == "GET":
-		return jsonify({"policies":{}})
+		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({
-		"errcode": "M_NOT_FOUND",
+		"public_key": globals.pubkey()
-		"error": "The public key was not found"
+	})
-	}), 404
+
 
 @identity.route("/_matrix/identity/v2/hash_details")
 async def hash_details():
-	return jsonify({"algorithms":["none","sha256"],"lookup_pepper": "vona"})
+	return jsonify({
+		"algorithms": [
+			"none",
+			"sha256",
+		],
+		"lookup_pepper": "vona"
+	})
+
 
 @identity.route("/_matrix/identity/v2/lookup", methods=["POST"])
 async def lookup():
 	req = request.json
 
-	if "addresses" in req:
+	if (
-		return jsonify({"mappings": {req["addresses"][0]: f"@vona:{server_name}"}})
+		isinstance(req, dict)
-	else:
+		and "addresses" in req
-		return jsonify({"errcode": "M_INVALID_PEPPER","error": "Invalid pepper"})
+		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": str(the_funny_number)})
+	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:{server_name}"
+		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)
+		})
+	)
 
-	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": the_funny_number
+		"validated_at": config.the_funny_number
 	})
 
 
-# https://spec.matrix.org/latest/identity-service-api/#invitation-storage
 @identity.route("/_matrix/identity/v2/store-invite", methods=["POST"])
 async def invite():
 	return jsonify({
 		"display_name": "Vona",
 		"public_keys": [
 			{
-				"key_validity_url": f"https://{server_name}/_matrix/identity/v2/pubkey/isvalid",
+				"key_validity_url": f"https://{config.server_name}/_matrix/identity/v2/pubkey/isvalid",
 				"public_key": "ohyeah"
 			},
 			{
-				"key_validity_url": f"https://{server_name}/_matrix/identity/v2/pubkey/ephemeral/isvalid",
+				"key_validity_url": f"https://{config.server_name}/_matrix/identity/v2/pubkey/ephemeral/isvalid",
 				"public_key": "burgerkingfootlettuce"
 			}
 		],
@@ -144,7 +176,6 @@ async def invite():
 	})
 
 
-# https://spec.matrix.org/latest/identity-service-api/#ephemeral-invitation-signing
 @identity.route("/_matrix/identity/v2/sign-ed25519", methods=["POST"])
 async def invite_signing():
 	required_keys = {"mxid", "private_key", "token"}