90 lines
2.7 KiB
Python
90 lines
2.7 KiB
Python
|
from flask import Blueprint, current_app, jsonify, render_template, request
|
||
|
|
|
||
|
|
bp = Blueprint("main", __name__)
|
||
|
|
|
||
|
|
|
||
|
|
def _check_hook_auth():
|
||
|
|
"""MediaMTX hooks authenticate with a simple shared-secret token so
|
||
|
|
random requests to /hooks/* can't spoof stream start/stop events.
|
||
|
|
Not meant to be a strong secret — just keeps it out of casual reach."""
|
||
|
|
token = current_app.hook_token
|
||
|
|
if not token:
|
||
|
|
return True
|
||
|
|
provided = request.args.get("token") or request.headers.get("X-Hook-Token")
|
||
|
|
return provided == token
|
||
|
|
|
||
|
|
|
||
|
|
@bp.route("/")
|
||
|
|
def index():
|
||
|
|
return render_template(
|
||
|
|
"index.html",
|
||
|
|
domain=current_app.domain,
|
||
|
|
rtmp_port=current_app.rtmp_port,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
@bp.route("/api/active")
|
||
|
|
def api_active():
|
||
|
|
return jsonify(current_app.state.active())
|
||
|
|
|
||
|
|
|
||
|
|
@bp.route("/api/history")
|
||
|
|
def api_history():
|
||
|
|
return jsonify(current_app.state.history())
|
||
|
|
|
||
|
|
|
||
|
|
@bp.route("/health")
|
||
|
|
def health():
|
||
|
|
return jsonify({"status": "ok"})
|
||
|
|
|
||
|
|
|
||
|
|
def _check_creds_auth():
|
||
|
|
"""Separate token from the hook auth so the IRC-bot-facing credential
|
||
|
|
endpoint can be rotated independently of the MediaMTX hook token."""
|
||
|
|
token = current_app.creds_token
|
||
|
|
if not token:
|
||
|
|
return True
|
||
|
|
provided = request.args.get("token") or request.headers.get("X-Creds-Token")
|
||
|
|
return provided == token
|
||
|
|
|
||
|
|
|
||
|
|
@bp.route("/api/creds")
|
||
|
|
def api_creds():
|
||
|
|
"""Returns the current publish URL for the IRC bot to hand out.
|
||
|
|
Not meant to be a strong secret — the IRC channel is a high-trust
|
||
|
|
environment; this just keeps it off the public landing page."""
|
||
|
|
if not _check_creds_auth():
|
||
|
|
return jsonify({"error": "unauthorized"}), 401
|
||
|
|
url = (
|
||
|
|
f"rtmp://{current_app.domain}:{current_app.rtmp_port}/live"
|
||
|
|
f"?user={current_app.publish_user}&pass={current_app.publish_password}"
|
||
|
|
)
|
||
|
|
return jsonify({
|
||
|
|
"url": url,
|
||
|
|
"user": current_app.publish_user,
|
||
|
|
"password": current_app.publish_password,
|
||
|
|
})
|
||
|
|
|
||
|
|
|
||
|
|
@bp.route("/hooks/ready", methods=["POST"])
|
||
|
|
def hook_ready():
|
||
|
|
if not _check_hook_auth():
|
||
|
|
return jsonify({"error": "unauthorized"}), 401
|
||
|
|
path = request.form.get("path") or request.args.get("path")
|
||
|
|
source_type = request.form.get("source_type") or request.args.get("source_type")
|
||
|
|
if not path:
|
||
|
|
return jsonify({"error": "missing path"}), 400
|
||
|
|
current_app.state.on_ready(path, source_type)
|
||
|
|
return jsonify({"ok": True})
|
||
|
|
|
||
|
|
|
||
|
|
@bp.route("/hooks/not-ready", methods=["POST"])
|
||
|
|
def hook_not_ready():
|
||
|
|
if not _check_hook_auth():
|
||
|
|
return jsonify({"error": "unauthorized"}), 401
|
||
|
|
path = request.form.get("path") or request.args.get("path")
|
||
|
|
if not path:
|
||
|
|
return jsonify({"error": "missing path"}), 400
|
||
|
|
current_app.state.on_not_ready(path)
|
||
|
|
return jsonify({"ok": True})
|