0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@creatio313(城谷 匠)
in

ImageFlux Live StreamingのWebhook/APIスタブ構築

0
Last updated at Posted at 2026-07-08

概要

背景

さくらインターネット株式会社が提供するライブ配信PaaS「ImageFlux Live Streaming」には、会員限定配信・視聴機能が存在する。端的に言えば、認証・認可された利用者だけがライブ配信や視聴を行うことができる仕組みである。

配信者側

image.png
配信者側の場合は、配信試行時にImageFluxが送信するWebhook通知に対し、認証・認可サーバ（利用者側が構築）が許可/拒否応答をすることによって限定配信を実現する。Webhook通知には配信希望者が任意にメタデータを設定可能であるため、多くの場合メタデータ経由でCookieの値を渡し、権限の判定を行うことになるだろう。

視聴者側

image.png
視聴者側の場合には、あらかじめチャンネル作成者が設定した鍵取得APIを、視聴者が配信ファイル取得時に呼び出し（呼び出し先は配信データに記載されている）、APIサーバが権限を判断する。許可できる場合には、配信データ復号用の鍵をサーバがImageFluxから取得し、利用者に返却することで配信が視聴できる仕組みである。この場合も、多くの場合視聴者がAPIを呼び出す際にCookieの値を渡し、権限の判定を行うことになるだろう。

いずれにおいても、実際の実装においては認証・認可サーバが必要である。

今回は、その実装を省いて「いったん許可/拒否された場合の動きを確認したい」という方向けに、簡易的なスタブを構築してみた。

併せてイベントWebhook通知のログ出力もできるようにし、ImageFlux Live Streamingの各種通知・認証・認可機能を一気に試せるように仕上げた。

成果物

構成は以下の通りで、至って単純である。
image.png
ImageFluxおよび視聴者からの通信は、AppRun共用型で受ける。ゼロスケールできるようにすることで、最小限の費用に抑えるのが狙いだ。
AppRun共用型で動かすコンテナイメージは、コンテナレジストリに格納する。
ログ・メトリクスはモニタリングスイートに保存する。
ソースコードはこちら。

構築

コンテナイメージの準備

コンテナレジストリを作成し、そこにサーバロジックを備えたイメージを格納していく。
組んだプログラムは以下。基本的にはログ出力して許可/拒否応答を返すだけである。HLS配信の復号だけ若干処理がある程度だ。
ImageFlux Live StreamingのAPIトークンは、環境変数で受け取る。

package main

import (
	"bytes"
	"context"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"time"
)

const imageFluxEndpoint = "https://live-api.imageflux.jp/"

var imageFluxClient = &http.Client{Timeout: 10 * time.Second}

/*
***
HLSライブ配信の暗号化鍵を取得するための要求と応答の構造体
***
*/
type getEncryptKeyRequest struct {
	Kid string `json:"kid"`
}
type getEncryptKeyResponse struct {
	EncryptKey string `json:"encrypt_key"`
}

func main() {
	port := os.Getenv("PORT")
	if port == "" {
		port = "8080"
	}

	mux := http.NewServeMux()
	mux.HandleFunc("/health", handleHealth())
	mux.HandleFunc("/auth_webhook_url_allow", handleWebhook(`{"allowed":true}`))
	mux.HandleFunc("/auth_webhook_url_deny", handleWebhook(`{"allowed":false,"reason":"認証に失敗しました。"}`))
	mux.HandleFunc("/encrypt_key_uri_allow", handleHLSAllow())
	mux.HandleFunc("/encrypt_key_uri_deny", handleHLSDeny())
	mux.HandleFunc("/event_webhook_url", handleEventWebhook())

	server := &http.Server{
		Addr:    ":" + port,
		Handler: mux,
	}

	log.Printf("%s番ポートでサーバを起動しました。", port)
	if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
		log.Fatal(err)
	}
}

/*
***
ヘルスチェック用ハンドラ
***
*/
func handleHealth() http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
	}
}

/*
***
Webhookの内容を読み取り、ログに出力し、指定された応答を返却する。
***
*/
func handleWebhook(response string) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		if _, err := readAndLogWebhookBody(r); err != nil {
			log.Printf("Webhookの内容の読み取りに失敗しました： %v", err)
			http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)
			return
		}

		w.Header().Set("Content-Type", "application/json; charset=utf-8")
		w.WriteHeader(http.StatusOK)
		if _, err := w.Write([]byte(response)); err != nil {
			log.Printf("Webhookの応答の書き込みに失敗しました： %v", err)
		}
	}
}

/*
***
Webhookの内容を読み取り、ログに出力する。
***
*/
func handleEventWebhook() http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		if _, err := readAndLogWebhookBody(r); err != nil {
			log.Printf("Webhookの内容の読み取りに失敗しました： %v", err)
			http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)
			return
		}

		w.WriteHeader(http.StatusOK)
	}
}

/*
***
ログ出力関数
***
*/
func readAndLogWebhookBody(r *http.Request) ([]byte, error) {
	body, err := io.ReadAll(r.Body)
	if err != nil {
		return nil, err
	}
	defer r.Body.Close()
	log.Printf("Webhookの内容 %s %s：%s", r.Method, r.URL.Path, string(body))
	return body, nil
}

/*
***
HLSライブ配信視聴の要求を許可し、復号用の鍵を応答として返却する。
***
*/
func handleHLSAllow() http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		logHLSRequest(r)

		setHLSCORSHeaders(w)
		if r.Method == http.MethodOptions {
			w.WriteHeader(http.StatusNoContent)
			return
		}

		if r.Method != http.MethodGet {
			http.Error(w, http.StatusText(http.StatusMethodNotAllowed), http.StatusMethodNotAllowed)
			return
		}

		kid := r.URL.Query().Get("kid")
		if kid == "" {
			http.Error(w, "kidは必須項目です。", http.StatusBadRequest)
			return
		}

		accessToken := os.Getenv("IMAGEFLUX_ACCESS_TOKEN")
		if accessToken == "" {
			log.Printf("環境変数 IMAGEFLUX_ACCESS_TOKEN が未設定です。")
			http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)
			return
		}

		keyHex, err := getEncryptKeyHex(r.Context(), kid, accessToken)
		if err != nil {
			log.Printf("GetEncryptKey API 呼び出しに失敗しました: kid=%s err=%v", kid, err)
			http.Error(w, http.StatusText(http.StatusBadGateway), http.StatusBadGateway)
			return
		}

		keyBin, err := hex.DecodeString(keyHex)
		if err != nil || len(keyBin) != 16 {
			log.Printf("GetEncryptKey API の鍵形式が不正です： key=%q err=%v", keyHex, err)
			http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)
			return
		}

		log.Printf("HLSライブ配信視聴を許可します：%s %s kid=%s", r.Method, r.URL.Path, kid)
		w.Header().Set("Content-Type", "application/octet-stream")
		w.WriteHeader(http.StatusOK)
		if _, err := w.Write(keyBin); err != nil {
			log.Printf("応答の書き込みに失敗しました： %v", err)
		}
	}
}

/*
***
HLSライブ配信の暗号化鍵を取得するための関数
***
*/
func getEncryptKeyHex(ctx context.Context, kid string, accessToken string) (string, error) {
	payload := getEncryptKeyRequest{Kid: kid}
	body, err := json.Marshal(payload)
	if err != nil {
		return "", fmt.Errorf("JSONの生成に失敗しました： %w", err)
	}

	req, err := http.NewRequestWithContext(ctx, http.MethodPost, imageFluxEndpoint, bytes.NewReader(body))
	if err != nil {
		return "", fmt.Errorf("API要求の作成に失敗しました： %w", err)
	}

	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("X-Sora-Target", "ImageFlux_20200707.GetEncryptKey")
	req.Header.Set("Authorization", "Bearer "+accessToken)

	resp, err := imageFluxClient.Do(req)
	if err != nil {
		return "", fmt.Errorf("API要求の実行に失敗しました： %w", err)
	}
	defer resp.Body.Close()

	respBody, err := io.ReadAll(resp.Body)
	if err != nil {
		return "", fmt.Errorf("API応答の読み取りに失敗しました： %w", err)
	}

	log.Printf("ImageFlux API 応答: status=%d body=%s", resp.StatusCode, string(respBody))

	if resp.StatusCode != http.StatusOK {
		return "", fmt.Errorf("API応答のステータスが不正です： status=%d body=%s", resp.StatusCode, string(respBody))
	}

	var parsed getEncryptKeyResponse
	if err := json.Unmarshal(respBody, &parsed); err != nil {
		return "", fmt.Errorf("API応答の解析に失敗しました： %w", err)
	}

	if parsed.EncryptKey == "" {
		return "", fmt.Errorf("API応答の鍵が空です。")
	}

	return parsed.EncryptKey, nil
}

/*
***
HLSライブ配信の暗号化鍵の取得を拒否する。
***
*/
func handleHLSDeny() http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		logHLSRequest(r)

		setHLSCORSHeaders(w)
		if r.Method == http.MethodOptions {
			w.WriteHeader(http.StatusNoContent)
			return
		}

		log.Printf("HLSライブ配信視聴を拒否します：%s %s", r.Method, r.URL.Path)
		http.Error(w, http.StatusText(http.StatusForbidden), http.StatusForbidden)
	}
}

/*
***
HLSライブ配信視聴向けの要求内容をログ出力する。
***
*/
func logHLSRequest(r *http.Request) {
	log.Printf(
		"HLSライブ配信視聴リクエスト受信：method=%s path=%s rawQuery=%q kid=%q remoteAddr=%s origin=%q userAgent=%q",
		r.Method,
		r.URL.Path,
		r.URL.RawQuery,
		r.URL.Query().Get("kid"),
		r.RemoteAddr,
		r.Header.Get("Origin"),
		r.UserAgent(),
	)
}

/*
***
HLSライブ配信のCORSヘッダーを設定する。
***
*/
func setHLSCORSHeaders(w http.ResponseWriter) {
	w.Header().Set("Access-Control-Allow-Origin", "*")
	w.Header().Set("Access-Control-Allow-Methods", "GET, OPTIONS")
	w.Header().Set("Access-Control-Allow-Headers", "*")
}

Dockerfileは以下の通り。

FROM golang:1.26 AS build

WORKDIR /go/src/app
COPY api/go.mod api/go.sum* ./
RUN go mod download
COPY api/ ./

RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o /go/bin/app ./cmd/server

FROM gcr.io/distroless/base-debian13:nonroot-amd64
COPY --from=build /go/bin/app /app
EXPOSE 8080
ENV PORT=8080
ENTRYPOINT ["/app"]

これをdocker build、pushすればよい。

基盤構築

Terraform1発で構築できる。
変数にはコンテナレジストリの認証情報と、ImageFlux Live StreamingのAPIキーを反映する。

resource "sakura_apprun_shared" "imageflux_live_streaming_stub" {
  name = "ImageFlux Live StreamingのWebhook/APIスタブ"

  max_scale       = 3
  min_scale       = 0
  port            = 8080
  timeout_seconds = 60

  components = [{
    name       = "ImageFlux Live StreamingのWebhook/APIスタブコンテナ"
    max_cpu    = "0.5"
    max_memory = "1Gi"
    deploy_source = {
      container_registry = {
        image               = var.container_registry_image
        password_wo         = var.container_registry_password_wo
        password_wo_version = 1
        server              = var.container_registry_server
        username            = var.container_registry_username
      }
    }
    env = [{
      key   = "IMAGEFLUX_ACCESS_TOKEN"
      value = var.imageflux_access_token
    }]
    probe = {
      http_get = {
        path = "/health"
        port = 8080
      }
    }
  }]
  traffics = [{
    version_index = 0
    percent       = 100
  }]
}
resource "sakura_monitoring_suite_log_storage" "ils_stub_log_storage" {
  name                  = "ImageFlux Live StreamingのWebhook/API"
  description           = "ImageFlux Live StreamingのWebhook/APIスタブのログを保存するためのログストレージ"
  classification        = "shared"
  is_system             = false
  retention_period_days = 40
}
resource "sakura_monitoring_suite_log_routing" "ils_stub_log_routing" {
  resource_id    = sakura_apprun_shared.imageflux_live_streaming_stub.resource_id
  storage_id     = sakura_monitoring_suite_log_storage.ils_stub_log_storage.id
  publisher_code = "apprun"
  variant        = "applicationlog"
}
resource "sakura_monitoring_suite_metric_storage" "ils_stub_metric_storage" {
  name        = "ImageFlux Live StreamingのWebhook/APIスタブ"
  description = "ImageFlux Live StreamingのWebhook/APIスタブのメトリクスを保存するためのメトリクスストレージ"
  is_system   = false
}
resource "sakura_monitoring_suite_metric_routing" "ils_stub_metric_routing" {
  resource_id    = sakura_apprun_shared.imageflux_live_streaming_stub.resource_id
  storage_id     = sakura_monitoring_suite_metric_storage.ils_stub_metric_storage.id
  publisher_code = "apprun"
  variant        = "applicationmetrics"
}

おなじみのコマンドを2発叩けばよい。

terraform init
terraform apply

image.png
構築が完了すると、AppRun共用型の公開URLが出力される。
このURLの末尾に以下のようなパスをつけることで、配信・視聴の許可・拒否ができる。イベントWebhook通知のログ記録も可能だ。
配信の許可・拒否

  • /auth_webhook_url_allow
  • /auth_webhook_url_deny

視聴の許可・拒否

  • /encrypt_key_uri_allow
  • /encrypt_key_uri_deny

イベントWebhook通知

  • /event_webhook_url

ImageFlux Live Streamingのチャンネル作成APIでそれぞれどう反映するかは、APIリファレンスを参照されたし。

動作検証

上記で解説したURLを設定して、チャンネル作成をしてみる。今回は全許可で試してみた。

image.png
作成したチャンネルで配信試行すると、すぐ配信許可応答が行われ、配信が開始できる。

image.png
HLS側でのライブ配信視聴時も、復号用鍵の取得要求を受け付け、ImageFlux側からその鍵を取得、返却していることがわかる。

image.png
イベントWebhook通知も問題なくログ出力できている。ここからアーカイブのパスも取得可能だ。

問題なく配信・視聴・ログ出力ができた。URLをdenyに変更すれば、拒否された場合の配信・視聴もシミュレーションできる。

まとめ

今回はAppRun共用型を用いて、ImageFlux Live StreamingのWebhook/APIスタブを構築してみた。HLS配信視聴の部分は「スタブ」という語が適切なのか、と少し迷ったが、動作検証用の仮ロジックをまとめて、今回はスタブと呼ぶことにしたい。
会員限定配信・視聴の動作機序を手軽に見てみたい方は、ぜひ使ってみて欲しい。

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?