From 15a7546a59d92bcb86ac72f9369a08517a675c42 Mon Sep 17 00:00:00 2001 From: xuhongyuan Date: Wed, 19 Aug 2026 16:42:17 +0800 Subject: [PATCH] =?UTF-8?q?feat(auth):=20=E5=9F=BA=E4=BA=8E=20casbin=20?= =?UTF-8?q?=E7=9A=84=E7=99=BB=E5=BD=95=E4=B8=8E=20RBAC=20=E6=9D=83?= =?UTF-8?q?=E9=99=90=E7=B3=BB=E7=BB=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 引入 github.com/casbin/casbin/v2 实现 RBAC (model.conf + policy.csv) - 角色: admin/publisher/viewer/guest,资源: config/room/user/srs,动作: read/list/publish/subscribe/stop/watch/manage - JWT 登录 (golang-jwt/jwt/v5,bcrypt 存储,data/users.json 持久化,种子用户 admin/publisher/viewer) - 中间件: HTTP AuthorizeMiddleware + gRPC Unary/Stream 拦截器,支持 Authorization Bearer/Cookie/Query/X-Token - 新增 API: POST /api/auth/login|register|logout, GET /api/auth/me|users|check, POST /api/auth/users/role (Casbin 鉴权) - 保护业务路由: /api/publish/subscribe/stop/rooms/room/*/events/srs/streams 及媒体代理 /rtc/v1/* /api/cf/* - gRPC 契约新增 Login/Register/GetMe/ListUsers/UpdateUserRole (api/live_sfu.proto + gen) - 前端 login.html + app.js 自动附加 Authorization,SSE 通过 ?token 传递,未登录自动跳转 - 配置: JWT_SECRET/JWT_TTL/CASBIN_MODEL/CASBIN_POLICY/AUTH_USER_FILE/ALLOW_REGISTER - 集成 TursoDB 持久化与测试覆盖 (auth_test.go) --- .env.example | 7 + .gitignore | 2 + README.md | 42 +- api/live_sfu.proto | 64 ++- cmd/server/main.go | 6 + data/.gitkeep | 0 deploy/srs.conf | 6 + docs/streaming-pipeline.md | 222 +++++++++ gen/live_sfu.pb.go | 728 ++++++++++++++++++++++++++-- gen/live_sfu_grpc.pb.go | 204 +++++++- go.mod | 20 +- go.sum | 44 +- internal/auth/auth.go | 99 ++++ internal/auth/auth_test.go | 167 +++++++ internal/auth/enforcer.go | 141 ++++++ internal/auth/jwt.go | 70 +++ internal/auth/middleware.go | 210 ++++++++ internal/auth/model.conf | 14 + internal/auth/policy.csv | 30 ++ internal/auth/store.go | 222 +++++++++ internal/config/config.go | 70 ++- internal/db/db.go | 174 +++++++ internal/server/auth_handler.go | 247 ++++++++++ internal/server/grpc.go | 21 +- internal/server/proxy.go | 25 + internal/server/rooms.go | 64 ++- internal/server/server.go | 114 ++++- internal/server/service.go | 140 +++++- internal/server/static/app.js | 155 +++++- internal/server/static/index.html | 8 +- internal/server/static/login.html | 152 ++++++ internal/server/static/publish.html | 7 +- internal/server/static/watch.html | 7 +- 33 files changed, 3351 insertions(+), 131 deletions(-) create mode 100644 data/.gitkeep create mode 100644 docs/streaming-pipeline.md create mode 100644 internal/auth/auth.go create mode 100644 internal/auth/auth_test.go create mode 100644 internal/auth/enforcer.go create mode 100644 internal/auth/jwt.go create mode 100644 internal/auth/middleware.go create mode 100644 internal/auth/model.conf create mode 100644 internal/auth/policy.csv create mode 100644 internal/auth/store.go create mode 100644 internal/db/db.go create mode 100644 internal/server/auth_handler.go create mode 100644 internal/server/static/login.html diff --git a/.env.example b/.env.example index eff815f..f767d88 100644 --- a/.env.example +++ b/.env.example @@ -7,6 +7,7 @@ SFU_PROVIDER=cloudflare,srs # SRS(本地对照后端:docker compose up -d srs) SRS_API_BASE=http://localhost:1985 +SRS_HTTP_BASE=http://localhost:8080 SRS_APP=live SRS_SECRET= SRS_CANDIDATE=127.0.0.1 @@ -20,3 +21,9 @@ CF_STUN_URL=stun:stun.cloudflare.com:3478 # 推流 JWT(SRS 入口鉴权,可选;设为 1 后 WHIP 必须带有效 token) DEMO_TOKEN_SECRET= SFU_TOKEN_REQUIRED=0 + +# Turso 嵌入式(默认 DB,房间分发拓扑持久化) +# 仅支持嵌入 file: 模式,开箱即用,无需远端凭证 +TURSO_DATABASE_URL=file:./data/live-sfu.db?cache=shared&_journal_mode=WAL +# 兼容 DATABASE_URL +# DATABASE_URL=file:./data/live-sfu.db?cache=shared&_journal_mode=WAL \ No newline at end of file diff --git a/.gitignore b/.gitignore index 82194b9..25e198e 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,5 @@ .DS_Store .env *.local.env +data/* +!data/.gitkeep diff --git a/README.md b/README.md index 4cad176..7a1a69f 100644 --- a/README.md +++ b/README.md @@ -6,6 +6,7 @@ - 一路推流 → SFU 扇出 → 多路拉流。 - 同一路发布可同时落到 **Cloudflare Realtime** 与 **SRS** 两条分发链路(分流)。 - 浏览器经服务端反向代理直连 SFU,凭证不出服务端。 +- **默认 DB 使用 Turso 嵌入式 (libSQL file)**:房间分发拓扑持久化到本地文件。 ## 目录结构 @@ -15,6 +16,7 @@ app/live-sfu-demo/ gen/ # buf generate 产出(Go) internal/ config/ # 运行参数(对齐 GOSpeak 的 env 布局) + db/ # Turso 嵌入式 (libSQL file) 持久化 sfu/cloudflare/ # Cloudflare Realtime REST 客户端 + Provider sfu/srs/ # SRS Provider server/ # gRPC 服务 + JSON 网关 + SRS/CF 媒体反代 + 房间扇出状态 @@ -26,12 +28,12 @@ app/live-sfu-demo/ ## 运行(SRS 链路,开箱即跑) ```bash -# 1) 起本地 SRS(WHIP/WHEP 后端) +# 1) 起本地 SRS(WHIP/WHEP/HLS 三协议后端,见 docs/streaming-pipeline.md) cd app/live-sfu-demo/deploy && SRS_CANDIDATE=127.0.0.1 docker compose up -d srs -# 2) 起 Demo 控制面 +# 2) 起 Demo 控制面(默认 Turso 嵌入式,无需额外配置) cd app/live-sfu-demo -cp .env.example .env # 可选;SRS 链路无需任何凭证 +cp .env.example .env go run ./cmd/server # 3) 打开浏览器 @@ -41,6 +43,28 @@ go run ./cmd/server 两个标签页用同一房间名即可配对。发布页勾选的分发后端会在「分发状态」中实时显示。 +## 默认 DB:Turso 嵌入式 + +房间的“分发目标”拓扑(`stream_targets`)默认持久化到 **Turso 嵌入式**(`github.com/tursodatabase/go-libsql` 的 SQLite 兼容 file 模式),而非纯内存。 + +- **DSN**:`TURSO_DATABASE_URL=file:./data/live-sfu.db?cache=shared&_journal_mode=WAL`(默认) + - 兼容 `DATABASE_URL` 覆盖 + - 仅支持嵌入 file: / :memory:,拒绝 libsql:// 远程(嵌入适配专注单机持久化) + - 自动建表 `(stream_targets, rooms)`,`SetMaxOpenConns(1)` 适配 SQLite 单写模型 + - 重启后自动 `LoadAll` 恢复房间 + +```bash +# 默认即嵌入文件 +TURSO_DATABASE_URL=file:./data/live-sfu.db?cache=shared&_journal_mode=WAL + +# 内存(测试) +TURSO_DATABASE_URL=file::memory:?cache=shared + +# 验证持久化:发布后重启,/api/rooms 仍在 +curl http://localhost:8088/api/rooms | jq +sqlite3 data/live-sfu.db "select room, backend, stream from stream_targets;" +``` + ## 运行(Cloudflare Realtime 链路,主 SFU) 在 `.env` 填入 Cloudflare Realtime 凭证后,`go run ./cmd/server` 即启用主 SFU: @@ -67,10 +91,18 @@ buf generate api `WatchRoom`(服务端流式推送房间分发拓扑)。gRPC 监听 `GRPC_PORT`(默认 9090),浏览器走同端口 的 JSON 网关(`protojson`)。 +## 文档 + +* 推流链路详解(PC → WHIP → SRS → HLS/WHEP/FLV):[`docs/streaming-pipeline.md`](docs/streaming-pipeline.md) + ## 分流拓扑 ``` publisher ──WHIP/tracks.new──▶ Cloudflare Realtime SFU ──▶ viewer(s) - └────WHIP────────────▶ SRS SFU (WHEP) ──▶ viewer(s) - 房间分发目标由 WatchRoom 实时广播,观众任选后端拉流 + └────WHIP────────────▶ SRS SFU ─┬─▶ WHEP viewer(s) (低延时 0.2-0.5s) + ├─▶ HLS viewer(s) (PC→WHIP→SRS→HLS 延时 5-10s,全端兼容) + └─▶ FLV viewer(s) + 一路 WHIP 推流,SRS 自动 remux 三协议同出;房间分发目标由 WatchRoom 实时广播,观众任选后端拉流 ``` + +> 详见 [`docs/streaming-pipeline.md`](docs/streaming-pipeline.md) — 默认已开启 HLS,`Go` 网关反代 `/live/*.m3u8`,前端 iOS 原生 / hls.js 双兼容。 \ No newline at end of file diff --git a/api/live_sfu.proto b/api/live_sfu.proto index 6dc37a7..e744043 100644 --- a/api/live_sfu.proto +++ b/api/live_sfu.proto @@ -16,6 +16,12 @@ service LiveSFU { rpc Subscribe(SubscribeRequest) returns (SubscribeResponse); rpc StopStream(StopStreamRequest) returns (StopStreamResponse); rpc WatchRoom(WatchRoomRequest) returns (stream RoomEvent); + // ---- Auth & Casbin ---- + rpc Login(LoginRequest) returns (LoginResponse); + rpc Register(RegisterRequest) returns (RegisterResponse); + rpc GetMe(GetMeRequest) returns (GetMeResponse); + rpc ListUsers(ListUsersRequest) returns (ListUsersResponse); + rpc UpdateUserRole(UpdateUserRoleRequest) returns (UpdateUserRoleResponse); } enum BackendKind { @@ -45,13 +51,12 @@ message GetConfigResponse { bool token_required = 3; } -// StreamTarget 表示一个房间在某后端的一条分发目标(SFU 扇出出口)。 message StreamTarget { BackendKind backend = 1; - string session_id = 2; // Cloudflare: 发布者 sessionId - string stream = 3; // SRS: stream 名 - string publish_token = 4;// SRS: 推流 JWT(可选) - string url = 5; // 拉流播放地址(SRS WHEP,可选) + string session_id = 2; + string stream = 3; + string publish_token = 4; + string url = 5; int64 published_at = 6; } @@ -105,3 +110,52 @@ message RoomEvent { string room = 1; repeated StreamTarget targets = 2; } + +// ===== Auth & RBAC (Casbin) ===== +message UserInfo { + string username = 1; + string role = 2; + int64 created_at = 3; +} + +message LoginRequest { + string username = 1; + string password = 2; +} +message LoginResponse { + string token = 1; + UserInfo user = 2; + int64 expires_at = 3; +} + +message RegisterRequest { + string username = 1; + string password = 2; + string role = 3; +} +message RegisterResponse { + string token = 1; + UserInfo user = 2; + int64 expires_at = 3; +} + +message GetMeRequest {} +message GetMeResponse { + UserInfo user = 1; + int64 expires_at = 2; + int64 issued_at = 3; +} + +message ListUsersRequest {} +message ListUsersResponse { + repeated UserInfo users = 1; +} + +message UpdateUserRoleRequest { + string username = 1; + string role = 2; +} +message UpdateUserRoleResponse { + bool ok = 1; + UserInfo user = 2; +} diff --git a/cmd/server/main.go b/cmd/server/main.go index 8b4844f..2e46034 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -11,6 +11,11 @@ import ( func main() { cfg := config.Load() srv := server.New(cfg) + defer func() { + if err := srv.Close(); err != nil { + log.Printf("[warn] db close: %v", err) + } + }() if err := srv.StartGRPC(); err != nil { log.Printf("[warn] grpc control plane failed to start: %v", err) } @@ -18,6 +23,7 @@ func main() { log.Printf("live-sfu demo ready: http://localhost:%s (grpc :%s)", cfg.HTTPPort, cfg.GRPCPort) log.Printf("backends (primary first): %v", cfg.ProviderList()) log.Printf("cloudflare configured: %v", cfg.CFAppID != "" && cfg.CFAppSecret != "") + log.Printf("turso embedded db: %s", cfg.DatabaseURL) if err := http.ListenAndServe(addr, srv.Handler()); err != nil { log.Fatalf("http server error: %v", err) } diff --git a/data/.gitkeep b/data/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/deploy/srs.conf b/deploy/srs.conf index f5ec0b1..9886457 100644 --- a/deploy/srs.conf +++ b/deploy/srs.conf @@ -35,4 +35,10 @@ vhost __defaultVhost__ { enabled on; mount [vhost]/[app]/[stream].flv; } + hls { + enabled on; + hls_path ./objs/nginx/html; + hls_fragment 10; + hls_window 60; + } } diff --git a/docs/streaming-pipeline.md b/docs/streaming-pipeline.md new file mode 100644 index 0000000..67a4a1b --- /dev/null +++ b/docs/streaming-pipeline.md @@ -0,0 +1,222 @@ +# 推流链路文档 · PC → WHIP → SRS → HLS / WHEP / FLV + +> 默认推流链路已升级为 **一路 WHIP 推流,SRS 三协议同出**。本文档沉淀该链路的拓扑、配置与验证方法,便于后续维护与排查。 + +## 1. 拓扑总览 + +``` + ┌─→ WHEP (WebRTC 0.2-0.5s) →