From 7c891f18cf885f11febda8664eb2532363b0f2c5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nikola=20Forr=C3=B3?= Date: Fri, 16 Aug 2019 11:52:50 +0200 Subject: [PATCH] Add twitch-webhooks service --- docker-compose.yaml.example | 18 ++++++ nginx/nginx.conf | 8 +++ twitch-webhooks/.gitignore | 96 ++++++++++++++++++++++++++++++++ twitch-webhooks/Dockerfile | 14 +++++ twitch-webhooks/app.py | 81 +++++++++++++++++++++++++++ twitch-webhooks/requirements.txt | 5 ++ twitch-webhooks/sender.py | 20 +++++++ twitch-webhooks/twitch.py | 77 +++++++++++++++++++++++++ 8 files changed, 319 insertions(+) create mode 100644 twitch-webhooks/.gitignore create mode 100644 twitch-webhooks/Dockerfile create mode 100644 twitch-webhooks/app.py create mode 100644 twitch-webhooks/requirements.txt create mode 100644 twitch-webhooks/sender.py create mode 100644 twitch-webhooks/twitch.py diff --git a/docker-compose.yaml.example b/docker-compose.yaml.example index 3815b8a..1f2dd3d 100644 --- a/docker-compose.yaml.example +++ b/docker-compose.yaml.example @@ -16,6 +16,7 @@ services: - teespring-api - twitch-cache-api - twitch-subs-api + - twitch-webhooks - cms # Cheese Horde members API service with /data/horde-members mounted as database storage @@ -100,6 +101,23 @@ services: expose: - 5000 + # Twitch webhooks service with /data/twitch-webhooks mounted as data storage + # TWITCH_CLIENT_ID, TWITCH_CHANNEL_ID and CALLBACK_URL + # are needed for Twitch API access and synchronization + twitch-webhooks: + build: + context: ./twitch-webhooks + volumes: + - /data/twitch-webhooks:/twitch-webhooks + environment: + - LEASE_FILE=/twitch-webhooks/subscriptions.json + - CONSUMER_ADDRESS=twitch-events:2828 + - TWITCH_CLIENT_ID=__TWITCH_CLIENT_ID__ + - TWITCH_CHANNEL_ID=__TWITCH_CHANNEL_ID__ + - CALLBACK_URL=__CALLBACK_URL__ + expose: + - 5000 + # CMS service with /data/grav mounted as user data storage cms: build: diff --git a/nginx/nginx.conf b/nginx/nginx.conf index 893ca80..cd738d7 100644 --- a/nginx/nginx.conf +++ b/nginx/nginx.conf @@ -128,6 +128,14 @@ http { proxy_pass http://twitch-subs-api:5000; } + location ^~ /twitch-webhooks/ { + rewrite ^/twitch-webhooks(/.*)$ $1 break; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_http_version 1.1; + proxy_pass http://twitch-webhooks:5000; + } + location ^~ /schedule { return 301 /#livestreams; } diff --git a/twitch-webhooks/.gitignore b/twitch-webhooks/.gitignore new file mode 100644 index 0000000..6a18ad4 --- /dev/null +++ b/twitch-webhooks/.gitignore @@ -0,0 +1,96 @@ +# ---> Python +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +env/ +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*,cover +.hypothesis/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# pyenv +.python-version + +# celery beat schedule file +celerybeat-schedule + +# SageMath parsed files +*.sage.py + +# dotenv +.env + +# virtualenv +.venv +venv/ +ENV/ + +# Spyder project settings +.spyderproject + +# Rope project settings +.ropeproject + diff --git a/twitch-webhooks/Dockerfile b/twitch-webhooks/Dockerfile new file mode 100644 index 0000000..3e9833f --- /dev/null +++ b/twitch-webhooks/Dockerfile @@ -0,0 +1,14 @@ +FROM python:alpine + +WORKDIR /app +COPY . . + +RUN pip install --no-cache-dir --requirement requirements.txt + +RUN addgroup -g 9999 lilia + +EXPOSE 5000 + +USER nobody:lilia + +ENTRYPOINT ["python", "app.py"] diff --git a/twitch-webhooks/app.py b/twitch-webhooks/app.py new file mode 100644 index 0000000..8864c02 --- /dev/null +++ b/twitch-webhooks/app.py @@ -0,0 +1,81 @@ +import datetime +import hashlib +import hmac +import json +import logging +import os + +import flask +import flask_apscheduler +import flask_restful +import flask_restful.reqparse + +from sender import Sender + + +subscriptions = {} + +app = flask.Flask(__name__) +app.logger.setLevel(logging.INFO) +app.config.update( + ERROR_404_HELP=False, + SCHEDULER_TIMEZONE='UTC', + SCHEDULER_JOBS=[ + dict(id='subscribe', + func='twitch:Twitch.subscribe', + args=(app, subscriptions, 86400), + max_instances=1, + trigger='interval', + seconds=3600, + next_run_time=datetime.datetime.utcnow() + datetime.timedelta(seconds=10))]) + +scheduler = flask_apscheduler.APScheduler() +scheduler.init_app(app) + +api = flask_restful.Api(app) + +sender = Sender(app) + + +verification_parser = flask_restful.reqparse.RequestParser() +verification_parser.add_argument('hub.challenge', type=str, required=True) +verification_parser.add_argument('hub.lease_seconds', type=int, required=True) +verification_parser.add_argument('hub.mode', type=str, required=True) +verification_parser.add_argument('hub.topic', type=str, required=True) + + +class WebhooksResource(flask_restful.Resource): + def get(self): + args = verification_parser.parse_args() + verified = False + for sub in subscriptions.values(): + h = '{0[hub.mode]}|{0[hub.topic]}|{0[hub.lease_seconds]}'.format(args) + if hashlib.sha256(h.encode()).hexdigest() == sub['request_hash']: + verified = True + break + if not verified: + flask_restful.abort(400, message='Verification failed') + r = app.make_response(args['hub.challenge']) + r.mimetype = 'text/plain' + r.status_code = 200 + return r + + def post(self): + sha, signature = flask.request.headers.get('X-Hub-Signature').split('=') + data = flask.request.data + for topic, sub in subscriptions.items(): + mac = hmac.new(sub['secret'].encode(), data, getattr(hashlib, sha)) + if not hmac.compare_digest(mac.hexdigest(), signature): + continue + message = flask.request.get_json(force=True) + message['topic'] = topic + sender.send(json.dumps(message).encode()) + return None, 204 + + +api.add_resource(WebhooksResource, '/webhooks') + + +if __name__ == '__main__': + scheduler.start() + app.run(host='0.0.0.0', threaded=True, debug=False) diff --git a/twitch-webhooks/requirements.txt b/twitch-webhooks/requirements.txt new file mode 100644 index 0000000..4b0d18e --- /dev/null +++ b/twitch-webhooks/requirements.txt @@ -0,0 +1,5 @@ +Flask +Flask-APScheduler +Flask-RESTful +python-dateutil +requests-futures diff --git a/twitch-webhooks/sender.py b/twitch-webhooks/sender.py new file mode 100644 index 0000000..0ea3f7a --- /dev/null +++ b/twitch-webhooks/sender.py @@ -0,0 +1,20 @@ +import os +import socket + + +class Sender(object): + def __init__(self, app): + self.app = app + addr, port = os.getenv('CONSUMER_ADDRESS').split(':') + self.address = (addr, int(port)) + self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) + + def __del__(self): + self.socket.close() + + def send(self, message): + try: + self.socket.sendto(message, self.address) + except Exception as e: + with self.app.app_context(): + self.app.logger.error('Failed to send message: %s', str(e)) diff --git a/twitch-webhooks/twitch.py b/twitch-webhooks/twitch.py new file mode 100644 index 0000000..7d4d73f --- /dev/null +++ b/twitch-webhooks/twitch.py @@ -0,0 +1,77 @@ +import datetime +import hashlib +import json +import os +import random +import string + +import dateutil.parser + +from requests_futures.sessions import FuturesSession + + +TOPIC_URLS = { + 'streams': 'https://api.twitch.tv/helix/streams?user_id={0}', + 'followers': 'https://api.twitch.tv/helix/users/follows?first=1&to_id={0}', +} + + +class Twitch(object): + @classmethod + def subscribe(cls, app, subscriptions, lease_time): + def read_leases(lease_file): + result = dict( + streams=dict(secret=None, valid_until=None, request_hash=None), + followers=dict(secret=None, valid_until=None, request_hash=None)) + try: + with open(lease_file) as f: + leases = json.load(f) + except IOError: + return result + else: + result.update(leases) + return result + def write_leases(lease_file, leases): + with open(lease_file, 'w') as f: + json.dump(leases, f, indent=4, sort_keys=True) + def generate_secret(length): + pool = string.ascii_letters + string.digits + return ''.join(random.SystemRandom().choice(pool) for _ in range(length)) + def sub(topic, secret): + client_id = os.getenv('TWITCH_CLIENT_ID') + channel_id = os.getenv('TWITCH_CHANNEL_ID') + callback_url = os.getenv('CALLBACK_URL') + session = FuturesSession() + url = 'https://api.twitch.tv/helix/webhooks/hub' + headers = {'Client-ID': client_id} + data = { + 'hub.mode': 'subscribe', + 'hub.topic': TOPIC_URLS[topic].format(channel_id), + 'hub.callback': callback_url, + 'hub.lease_seconds': lease_time, + 'hub.secret': secret, + } + h = '{0[hub.mode]}|{0[hub.topic]}|{0[hub.lease_seconds]}'.format(data) + request_hash = hashlib.sha256(h.encode()).hexdigest() + return session.post(url, headers=headers, data=data), request_hash + with app.app_context(): + app.logger.info('Refreshing Twitch subscriptions') + lease_file = os.getenv('LEASE_FILE') + leases = read_leases(lease_file) + for topic, lease in leases.items(): + if not lease['secret'] or not lease['valid_until']: + remaining = 0 + if lease['valid_until']: + valid_until = dateutil.parser.parse(lease['valid_until']) + remaining = (valid_until - datetime.datetime.utcnow()).total_seconds() + if remaining < 3600: + app.logger.info('Subscription "%s" expired, re-requesting', topic) + secret = generate_secret(32) + valid_until = datetime.datetime.utcnow() + datetime.timedelta(seconds=lease_time) + request, request_hash = sub(topic, secret) + if request.result().ok: + lease['secret'] = secret + lease['valid_until'] = valid_until.isoformat() + lease['request_hash'] = request_hash + write_leases(lease_file, leases) + subscriptions.update(leases)