Генераторы токенов на Python

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Генератор токенов для SirinVideo, бэкенд `auth_backend { type hs256; }`.
Зависимостей нет — только стандартная библиотека Python 3.8+.

Правьте блок ПАРАМЕТРЫ и запускайте: скрипт печатает готовую ссылку и команду проверки.
"""

# ─────────────────────────── ПАРАМЕТРЫ ───────────────────────────
SECRET      = "секрет-подписи-токенов-зрителей"  # ровно то же значение, что в auth_backend { secret … }
BASE_URL    = "http://media.example.com:8080"    # адрес сервера
STREAM      = "cam"                              # путь потока БЕЗ ведущего слэша
MINUTES     = 15                                 # срок жизни токена
PRESET      = "live"                             # live | ip | label | archive | publish
CLIENT_IP   = "203.0.113.7"                      # для PRESET="ip": адрес зрителя
CLIENT_NET  = ["203.0.113.0/24"]                 # для PRESET="ip": допустимые сети
LABELS      = ["site-north"]                     # для PRESET="label": метки, ТОЧНОЕ совпадение
ISSUER      = None                               # заполнять, только если задан auth_backend { issuer … }
AUDIENCE    = None                               # заполнять, только если задан auth_backend { audience … }
PERMS_CLAIM = "grants"                           # auth_backend { perms_claim … }, по умолчанию "grants"
# ─────────────────────────────────────────────────────────────────

import base64
import hashlib
import hmac
import json
import time


def b64(raw: bytes) -> str:
    """base64url без выравнивания — кодировка частей компактного JWS."""
    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")


def mint(secret: str, payload: dict) -> str:
    """Собрать и подписать JWT. Алгоритм жёстко HS256: сервер сверяет строку `HS256`
    регистрозависимо и не принимает ни `none`, ни подмену типа ключа."""
    header  = b64(json.dumps({"alg": "HS256", "typ": "JWT"}, separators=(",", ":")).encode())
    body    = b64(json.dumps(payload, separators=(",", ":")).encode())
    signing = f"{header}.{body}".encode()
    return f"{header}.{body}.{b64(hmac.new(secret.encode(), signing, hashlib.sha256).digest())}"


def label_from_uuid(text: str) -> str:
    """uuid с дефисами → каноническая 22-символьная метка, как её хранит сервер.
    Метки внутри JWT НЕ приводятся к канонической форме автоматически — писать нужно результат
    этой функции, иначе ответ будет 403."""
    return base64.urlsafe_b64encode(bytes.fromhex(text.replace("-", ""))).rstrip(b"=").decode()


now = int(time.time())

# `sub` — субъект: попадает в учёт сессии и сворачивается в билет. Необязателен.
# `exp` — АБСОЛЮТНЫЙ момент истечения в unix-секундах. Сервер отвергает при exp <= now,
#         БЕЗ допуска на расхождение часов. Если поле опустить — токен станет бессрочным.
payload = {"sub": "viewer-42", "exp": now + MINUTES * 60}

if ISSUER:
    payload["iss"] = ISSUER      # проверяется, только если issuer настроен на сервере
if AUDIENCE:
    payload["aud"] = AUDIENCE    # строка либо список строк

clip_path = None

if PRESET == "live":
    # Минимум: смотреть ОДИН живой поток по HLS.
    #   action=read — живой просмотр (playback = архив, publish = публикация)
    #   path        — точный путь потока; `*` = все потоки; допустим глоб вида `cam*`
    #   protocols   — сузить токен до перечисленных протоколов; поля нет = все
    payload[PERMS_CLAIM] = [{"action": "read", "path": STREAM, "protocols": ["hls"]}]
    check_path = f"/{STREAM}/index.m3u8"

elif PRESET == "ip":
    # То же плюс привязка к адресу. Два независимых механизма, можно вместе:
    #   ip  — ровно один адрес; запрос с другого → 403
    #   net — список CIDR (не более 32); достаточно попасть в один
    # За обратным прокси адрес берётся из X-Forwarded-For ТОЛЬКО если прокси перечислен
    # в `trusted_proxies`, иначе привяжется адрес самого прокси.
    payload["ip"] = CLIENT_IP
    payload["net"] = CLIENT_NET
    payload[PERMS_CLAIM] = [{"action": "read", "path": STREAM, "protocols": ["hls"]}]
    check_path = f"/{STREAM}/index.m3u8"

elif PRESET == "label":
    # Доступ к ГРУППЕ потоков по меткам. Сравнение — точная строка: ни глоба, ни префикса,
    # ни регистронезависимости. Поле `path` можно опустить вовсе.
    payload[PERMS_CLAIM] = [{"action": "read", "labels": LABELS, "protocols": ["hls"]}]
    check_path = f"/{STREAM}/index.m3u8"

elif PRESET == "archive":
    # Архив и выгрузка клипа. Одного `playback` с совпавшим путём хватает на оба, потому что
    # путь совпал. Чтобы выдать ТОЛЬКО экспорт без права смотреть — нужен компактный грант
    # по метке с маской 10 (см. страницу «Метки потоков»).
    payload[PERMS_CLAIM] = [{"action": "playback", "path": STREAM, "protocols": ["hls"]}]
    check_path = f"/{STREAM}/timeshift-{now - 30}.m3u8"
    clip_path = f"/{STREAM}/dvr/clip?from={(now - 30) * 1000}&to={(now - 5) * 1000}"

elif PRESET == "publish":
    # Публикация. Действие publish — отдельное: токен на просмотр публиковать не даёт.
    # protocols: rtmp для RTMP/RTMPS, whip для WebRTC-публикации.
    payload[PERMS_CLAIM] = [{"action": "publish", "path": STREAM, "protocols": ["rtmp", "whip"]}]
    check_path = None

else:
    raise SystemExit(f"неизвестный PRESET: {PRESET}")

token = mint(SECRET, payload)

print("payload:", json.dumps(payload, ensure_ascii=False))
print("token  :", token)
print()

if PRESET == "publish":
    rtmp_host = BASE_URL.split("//", 1)[-1].split(":", 1)[0]
    print("Адрес для кодировщика (порт — из директивы `rtmp`):")
    print(f"  rtmp://{rtmp_host}:1935/{STREAM}?token={token}")
else:
    print("Готовая ссылка на просмотр:")
    print(f"{BASE_URL}{check_path}?token={token}")
    print()
    print("Проверка (ожидается 200):")
    print(f"  curl -s -o /dev/null -w '%{{http_code}}\\n' \"{BASE_URL}{check_path}?token={token}\"")
    print("То же через заголовок:")
    print(f"  curl -s -o /dev/null -w '%{{http_code}}\\n' -H 'Authorization: Bearer {token}' {BASE_URL}{check_path}")
    if clip_path:
        print()
        print("Выгрузка клипа тем же токеном:")
        print(f"  curl -o clip.mp4 \"{BASE_URL}{clip_path}&token={token}\"")
    print()
    print("Дальше плеер обязан идти по ссылкам ИЗ ПЛЕЙЛИСТА — в них уже подставлен ?ssid=.")
    print("Прямой запрос сегмента с ?token= вернёт 401: сегменты проверяются только билетом.")

python3 hs_token.py                                     # печатает готовую ссылку
curl -s -o /dev/null -w '%{http_code}\n' "<напечатанная ссылка>"
curl -s -o /dev/null -w '%{http_code}\n' \
  http://media.example.com:8080/cam/index.m3u8          # контроль: тот же путь без токена

200
401

pip install "PyJWT>=2" "cryptography>=41"

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Генератор токенов для SirinVideo, бэкенд `auth_backend { type jwks; }`.
Зависимости:  pip install "PyJWT>=2" "cryptography>=41"

Скрипт делает две вещи:
  1) при первом запуске создаёт ключевую пару и файл jwks.json — его надо отдавать по HTTP
     по адресу из auth_backend { url … };
  2) подписывает токен зрителя приватным ключом.
"""

# ─────────────────────────── ПАРАМЕТРЫ ───────────────────────────
KEY_FILE    = "idp.pem"                  # приватный ключ, создаётся автоматически
JWKS_FILE   = "jwks.json"                # публичный набор ключей для auth_backend { url … }
KID         = "idp-1"                    # идентификатор ключа, попадает в заголовок токена
ALG         = "RS256"                    # RS256 (RSA-2048) или ES256 (P-256) — иное сервер не примет

BASE_URL    = "http://media.example.com:8080"
STREAM      = "cam"
MINUTES     = 15
PRESET      = "live"                     # live | ip | label | archive | publish
CLIENT_IP   = "203.0.113.7"
CLIENT_NET  = ["203.0.113.0/24"]
LABELS      = ["site-north"]
ISSUER      = "https://idp.example"      # должен совпасть с auth_backend { issuer … }, если он задан
AUDIENCE    = "sirinvideo"               # должен совпасть с auth_backend { audience … }, если он задан
PERMS_CLAIM = "grants"
# ─────────────────────────────────────────────────────────────────

import base64
import json
import os
import time

import jwt as pyjwt
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import ec, rsa


def b64u_int(value: int) -> str:
    raw = value.to_bytes((value.bit_length() + 7) // 8, "big")
    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")


def b64u_bytes(raw: bytes) -> str:
    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")


def label_from_uuid(text: str) -> str:
    """uuid с дефисами → каноническая 22-символьная метка, как её хранит сервер."""
    return base64.urlsafe_b64encode(bytes.fromhex(text.replace("-", ""))).rstrip(b"=").decode()


# ── 1. ключ и набор ключей ───────────────────────────────────────────────────
if not os.path.exists(KEY_FILE):
    key = (rsa.generate_private_key(public_exponent=65537, key_size=2048)
           if ALG == "RS256" else ec.generate_private_key(ec.SECP256R1()))
    with open(KEY_FILE, "wb") as f:
        f.write(key.private_bytes(serialization.Encoding.PEM,
                                  serialization.PrivateFormat.PKCS8,
                                  serialization.NoEncryption()))
    print(f"создан приватный ключ {KEY_FILE}")

private_pem = open(KEY_FILE).read()
key = serialization.load_pem_private_key(private_pem.encode(), password=None)

if ALG == "RS256":
    numbers = key.public_key().public_numbers()
    jwk = {"kty": "RSA", "kid": KID, "use": "sig", "alg": "RS256",
           "n": b64u_int(numbers.n), "e": b64u_int(numbers.e)}
else:
    numbers = key.public_key().public_numbers()
    jwk = {"kty": "EC", "kid": KID, "use": "sig", "alg": "ES256", "crv": "P-256",
           "x": b64u_bytes(numbers.x.to_bytes(32, "big")),
           "y": b64u_bytes(numbers.y.to_bytes(32, "big"))}

with open(JWKS_FILE, "w") as f:
    json.dump({"keys": [jwk]}, f, indent=2)
print(f"записан {JWKS_FILE} — отдайте его по адресу из auth_backend {{ url … }}")

# ── 2. токен ─────────────────────────────────────────────────────────────────
now = int(time.time())
payload = {"sub": "viewer-42", "iat": now, "exp": now + MINUTES * 60}
if ISSUER:
    payload["iss"] = ISSUER
if AUDIENCE:
    payload["aud"] = AUDIENCE

if PRESET == "live":
    payload[PERMS_CLAIM] = [{"action": "read", "path": STREAM, "protocols": ["hls"]}]
    check_path = f"/{STREAM}/index.m3u8"
elif PRESET == "ip":
    payload["ip"] = CLIENT_IP            # ровно один адрес
    payload["net"] = CLIENT_NET          # либо список CIDR, не более 32
    payload[PERMS_CLAIM] = [{"action": "read", "path": STREAM, "protocols": ["hls"]}]
    check_path = f"/{STREAM}/index.m3u8"
elif PRESET == "label":
    # Метка сравнивается ТОЧНОЙ строкой. uuid из конфигурации сервер привёл к 22-символьной
    # форме — используйте label_from_uuid(), а не запись с дефисами.
    payload[PERMS_CLAIM] = [{"action": "read", "labels": LABELS, "protocols": ["hls"]}]
    check_path = f"/{STREAM}/index.m3u8"
elif PRESET == "archive":
    payload[PERMS_CLAIM] = [{"action": "playback", "path": STREAM, "protocols": ["hls"]}]
    check_path = f"/{STREAM}/timeshift-{now - 30}.m3u8"
elif PRESET == "publish":
    payload[PERMS_CLAIM] = [{"action": "publish", "path": STREAM, "protocols": ["rtmp", "whip"]}]
    check_path = None
else:
    raise SystemExit(f"неизвестный PRESET: {PRESET}")

# `kid` в заголовке на практике обязателен: без него сервер переберёт все ключи подходящего
# алгоритма, а неизвестный kid дополнительно вызывает досрочное обновление набора.
token = pyjwt.encode(payload, private_pem, algorithm=ALG, headers={"kid": KID})

print("payload:", json.dumps(payload, ensure_ascii=False))
print("token  :", token)
print()
if check_path:
    print("Готовая ссылка на просмотр:")
    print(f"{BASE_URL}{check_path}?token={token}")
    print()
    print("Проверка (ожидается 200):")
    print(f"  curl -s -o /dev/null -w '%{{http_code}}\\n' \"{BASE_URL}{check_path}?token={token}\"")
else:
    rtmp_host = BASE_URL.split("//", 1)[-1].split(":", 1)[0]
    print("Адрес для кодировщика (порт — из директивы `rtmp`):")
    print(f"  rtmp://{rtmp_host}:1935/{STREAM}?token={token}")

python3 -m http.server 18999 --bind 127.0.0.1 &   # раздаём jwks.json
python3 jwks_token.py
curl -s -o /dev/null -w '%{http_code}\n' "<напечатанная ссылка>"

200

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Минимальная служба авторизации для `auth_backend { type http; url … }`.
Только стандартная библиотека. Запуск:  python3 authd.py
"""

import http.server
import json

LISTEN = ("127.0.0.1", 9099)


class Handler(http.server.BaseHTTPRequestHandler):
    def do_POST(self):
        size = int(self.headers.get("content-length", "0"))
        req = json.loads(self.rfile.read(size) or b"{}")
        # req содержит: ip, user, password, token, action, privilege, path,
        #               labels, protocol, query, user_agent
        allowed = req.get("token") == "secret-string" and req.get("path") == "cam"

        if not allowed:
            # 403 — «авторизация отвергнута». 401 означало бы «учётные данные не приняты»
            # и кормило бы счётчик неудачных попыток по адресу клиента.
            self.send_response(403)
            self.send_header("content-length", "0")
            self.end_headers()
            return

        # Пустое тело с кодом 2xx = «разрешить ровно запрошенную пару действие + путь».
        # JSON уточняет грант; ttl_secs становится сроком жизни билета сессии.
        body = json.dumps({"sub": "viewer-42", "ttl_secs": 120}).encode()
        self.send_response(200)
        self.send_header("content-type", "application/json")
        self.send_header("content-length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, fmt, *args):
        pass  # запросов много — не засоряем консоль


http.server.HTTPServer(LISTEN, Handler).serve_forever()

auth_backend { type http; url http://127.0.0.1:9099/authorize; }

{"ip":"203.0.113.7","user":"","password":"","token":"secret-string","action":"read",
 "privilege":"live","path":"cam","labels":["site-north"],"protocol":"hls",
 "query":"token=secret-string","user_agent":"curl/8.18.0"}

{"ip":"203.0.113.7","user":"","password":"","token":"secret-string","action":"publish",
 "privilege":"publish","path":"cam","labels":[],"protocol":"rtmp",
 "query":"token=secret-string","user_agent":""}

curl -s -o /dev/null -w '%{http_code}\n' \
  "http://media.example.com:8080/cam/index.m3u8?token=secret-string"
curl -s -o /dev/null -w '%{http_code}\n' \
  "http://media.example.com:8080/cam/index.m3u8?token=nope"
# теперь остановите службу и повторите первый запрос
curl -s -o /dev/null -w '%{http_code}\n' \
  "http://media.example.com:8080/cam/index.m3u8?token=secret-string"

200
403
503

auth_backend { type internal; }

principal svc   { token s3cr3t-service-token-value; permission read cam hls; }
principal alice { pass  hunter2;                    permission read cam;     }

curl -s -o /dev/null -w '%{http_code}\n' \
  "http://media.example.com:8080/cam/index.m3u8?token=s3cr3t-service-token-value"
curl -s -o /dev/null -w '%{http_code}\n' -u alice:hunter2 \
  http://media.example.com:8080/cam/index.m3u8

200
200