~sgeisenh/streamnotify

fc76a6af0ac896bcd0a031cd05c7000376ae214f — Samuel Eisenhandler 1 year, 5 months ago 8f231eb
Add linter config and type checker
2 files changed, 31 insertions(+), 21 deletions(-)

M server.py
A tox.ini
M server.py => server.py +29 -21
@@ 1,10 1,11 @@
import hmac
import os
from logging.config import dictConfig
from typing import Final
from typing import cast, Final

import requests
from flask import Flask, request, abort
from flask import abort, Flask, Request, request
from flask.typing import ResponseReturnValue

dictConfig(
    {


@@ 33,20 34,16 @@ app = Flask(__name__)
TWITCH_SECRET: Final[bytes] = os.environ["TWITCH_SECRET"].encode()
ZULIP_KEY: Final[str] = os.environ["ZULIP_KEY"]
ZULIP_EMAIL: Final[str] = os.environ["ZULIP_EMAIL"]
MESSAGES_URL = "https://recurse.zulipchat.com/api/v1/messages"
MEMBERS_URL = "https://recurse.zulipchat.com/api/v1/users"

MESSAGES_URL: Final[str] = "https://recurse.zulipchat.com/api/v1/messages"
TWITCH_MESSAGE_ID: Final[str] = "twitch-eventsub-message-id"
TWITCH_MESSAGE_TIMESTAMP: Final[str] = "twitch-eventsub-message-timestamp"
TWITCH_MESSAGE_SIGNATURE: Final[str] = "twitch-eventsub-message-signature"
TWITCH_MESSAGE_TYPE: Final[str] = "twitch-eventsub-message-type"

HMAC_PREFIX: Final[str] = "sha256="

STREAM: Final[str] = "livestream notifications"


def get_hmac_message(request):
def get_hmac_message(request: Request) -> bytes:
    return (
        request.headers[TWITCH_MESSAGE_ID].encode()
        + request.headers[TWITCH_MESSAGE_TIMESTAMP].encode()


@@ 54,46 51,57 @@ def get_hmac_message(request):
    )


def get_hmac(message):
def get_hmac(message: bytes) -> str:
    return hmac.new(TWITCH_SECRET, message, "sha256").hexdigest()


@app.route("/")
def index():
def index() -> ResponseReturnValue:
    return "Hey there, how's it going?"


NOTIFICATION_FORMAT: Final[
    str
] = "Channel [{user_login}](https://twitch.tv/{user_login}) has started a stream!"
SUBJECT_FORMAT: Final[str] = "{user_login} stream online notifications"


@app.route("/callback", methods=["POST"])
def callback():
def callback() -> ResponseReturnValue:
    body = request.json
    assert body is not None
    message = get_hmac_message(request)
    truth = HMAC_PREFIX + get_hmac(message)
    hmac_message = get_hmac_message(request)
    truth = HMAC_PREFIX + get_hmac(hmac_message)

    if not hmac.compare_digest(truth, request.headers[TWITCH_MESSAGE_SIGNATURE]):
        app.logger.warning("Invalid digest")
        abort(403)

    app.logger.info("Valid digest!")

    match request.headers[TWITCH_MESSAGE_TYPE]:
        case "webhook_callback_verification":
            app.logger.info("Replying to challenge!")
            return body["challenge"]
            return cast(str, body["challenge"])
        case "notification":
            # TODO: We should handle duplicate events gracefully
            # https://dev.twitch.tv/docs/eventsub/#handling-duplicate-events
            event = body["event"]
            app.logger.info(f"Received event: {event}")
            user_login = event["broadcaster_user_login"]
            subject = f"{user_login} stream online notifications"
            message = f"Channel [{user_login}](https://twitch.tv/{user_login}) has started a stream!"
            subject = SUBJECT_FORMAT.format(user_login=user_login)
            message = NOTIFICATION_FORMAT.format(user_login=user_login)
            send_message_zulip(subject, message)
    return "", 204


def send_message_zulip(subject, content):
    data = {"type": "stream", "to": STREAM, "subject": subject, "content": content}
ZULIP_STREAM: Final[str] = "livestream notifications"


def send_message_zulip(subject: str, content: str) -> bool:
    data = {
        "type": "stream",
        "to": ZULIP_STREAM,
        "subject": subject,
        "content": content,
    }
    try:
        app.logger.info("Sending message '%s' with data '%s'", content, data)
        response = requests.post(MESSAGES_URL, data=data, auth=(ZULIP_EMAIL, ZULIP_KEY))

A tox.ini => tox.ini +2 -0
@@ 0,0 1,2 @@
[flake8]
max-line-length = 100