63 lines
1.9 KiB
Python
63 lines
1.9 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import os
|
|
import time
|
|
from typing import Any
|
|
|
|
|
|
def ingest_token() -> str:
|
|
return (
|
|
os.environ.get("YOUTUBE_INGEST_TOKEN")
|
|
or os.environ.get("LIVE_YOUTUBE_INGEST_TOKEN")
|
|
or ""
|
|
).strip()
|
|
|
|
|
|
def ingest_auth_required() -> bool:
|
|
raw = (
|
|
os.environ.get("YOUTUBE_INGEST_REQUIRE_AUTH")
|
|
or os.environ.get("LIVE_YOUTUBE_INGEST_REQUIRE_AUTH")
|
|
or "1"
|
|
).strip().lower()
|
|
if raw in {"0", "false", "no", "off"}:
|
|
return False
|
|
return bool(ingest_token())
|
|
|
|
|
|
def verify_ingest_payload(payload: dict[str, Any]) -> tuple[bool, str]:
|
|
if not ingest_auth_required():
|
|
return True, ""
|
|
token = ingest_token()
|
|
if not token:
|
|
return False, "ingest auth enabled but YOUTUBE_INGEST_TOKEN is empty"
|
|
|
|
provided = str(payload.get("ingest_token") or payload.get("token") or "").strip()
|
|
if provided and hmac.compare_digest(provided, token):
|
|
return True, ""
|
|
|
|
ts_raw = payload.get("ingest_ts") or payload.get("ts")
|
|
sig = str(payload.get("ingest_sig") or payload.get("sig") or "").strip()
|
|
if sig and ts_raw is not None:
|
|
try:
|
|
ts = int(float(ts_raw))
|
|
except (TypeError, ValueError):
|
|
return False, "invalid ingest timestamp"
|
|
if abs(time.time() - ts) > 900:
|
|
return False, "ingest signature expired"
|
|
body = f"{ts}|{payload.get('process') or payload.get('lane') or 'chrome_extension'}"
|
|
expected = hmac.new(token.encode("utf-8"), body.encode("utf-8"), hashlib.sha256).hexdigest()
|
|
if hmac.compare_digest(sig, expected):
|
|
return True, ""
|
|
|
|
return False, "invalid or missing ingest credentials"
|
|
|
|
|
|
def ingest_config_payload() -> dict[str, Any]:
|
|
return {
|
|
"require_auth": ingest_auth_required(),
|
|
"has_token": bool(ingest_token()),
|
|
"endpoint": "/youtube/live/ingest",
|
|
}
|