100 lines
3.2 KiB
Python
100 lines
3.2 KiB
Python
"""Matrix Webhook main request handler."""
|
|
|
|
import json
|
|
import logging
|
|
from http import HTTPStatus
|
|
from hmac import HMAC
|
|
import importlib
|
|
|
|
from markdown import markdown
|
|
|
|
from . import conf, utils
|
|
|
|
LOGGER = logging.getLogger("matrix_webhook.handler")
|
|
|
|
|
|
async def matrix_webhook(request):
|
|
"""
|
|
Coroutine given to the server, st. it knows what to do with an HTTP request.
|
|
|
|
This one handles a POST, checks its content, and forwards it to the matrix room.
|
|
"""
|
|
LOGGER.debug(f"Handling {request=}")
|
|
data_b = await request.read()
|
|
|
|
try:
|
|
data = json.loads(data_b.decode())
|
|
except json.decoder.JSONDecodeError:
|
|
return utils.create_json_response(HTTPStatus.BAD_REQUEST, "Invalid JSON")
|
|
|
|
# legacy naming
|
|
if "text" in data and "body" not in data:
|
|
data["body"] = data["text"]
|
|
|
|
# allow key to be passed as a parameter
|
|
if "key" in request.rel_url.query and "key" not in data:
|
|
data["key"] = request.rel_url.query["key"]
|
|
|
|
if "formatter" in request.rel_url.query:
|
|
try:
|
|
format_type = request.rel_url.query["formatter"]
|
|
plugin = importlib.import_module(
|
|
f"matrix_webhook.formatters.{format_type}", "formatter"
|
|
)
|
|
data = plugin.formatter(data, request.headers)
|
|
except ModuleNotFoundError:
|
|
return utils.create_json_response(
|
|
HTTPStatus.BAD_REQUEST, "Unknown formatter"
|
|
)
|
|
|
|
if (
|
|
"room_id" not in request.rel_url.query
|
|
and "room_id" not in data
|
|
and conf.ROOM_KEYS[f'{data["key"]}']
|
|
):
|
|
data["room_id"] = conf.ROOM_KEYS[f'{data["key"]}']
|
|
if "room_id" in request.rel_url.query and "room_id" not in data:
|
|
data["room_id"] = request.rel_url.query["room_id"]
|
|
if "room_id" not in data:
|
|
data["room_id"] = request.path.lstrip("/")
|
|
|
|
# If we get a good SHA-256 HMAC digest,
|
|
# we can consider that the sender has the right API key
|
|
if "digest" in data:
|
|
if data["digest"] == HMAC(conf.API_KEY.encode(), data_b, "sha256").hexdigest():
|
|
data["key"] = conf.API_KEY
|
|
else: # but if there is a wrong digest, an informative error should be provided
|
|
return utils.create_json_response(
|
|
HTTPStatus.UNAUTHORIZED, "Invalid SHA-256 HMAC digest"
|
|
)
|
|
|
|
missing = []
|
|
for key in ["body", "key", "room_id"]:
|
|
if key not in data or not data[key]:
|
|
missing.append(key)
|
|
if missing:
|
|
return utils.create_json_response(
|
|
HTTPStatus.BAD_REQUEST, f"Missing {', '.join(missing)}"
|
|
)
|
|
|
|
if data["key"] not in conf.API_KEYS:
|
|
return utils.create_json_response(HTTPStatus.UNAUTHORIZED, "Invalid API key")
|
|
|
|
if "formatted_body" in data:
|
|
formatted_body = data["formatted_body"]
|
|
else:
|
|
formatted_body = markdown(str(data["body"]), extensions=["extra"])
|
|
|
|
# try to join room first -> non none response means error
|
|
resp = await utils.join_room(data["room_id"])
|
|
if resp is not None:
|
|
return resp
|
|
|
|
content = {
|
|
"msgtype": "m.text",
|
|
"body": data["body"],
|
|
"format": "org.matrix.custom.html",
|
|
"formatted_body": formatted_body,
|
|
}
|
|
print(conf.ROOM_KEYS)
|
|
return await utils.send_room_message(data["room_id"], content)
|