diff --git a/internal/auth/policy.csv b/internal/auth/policy.csv index 892aa7b..73e766f 100644 --- a/internal/auth/policy.csv +++ b/internal/auth/policy.csv @@ -25,6 +25,11 @@ p, viewer, room, watch p, viewer, srs, streams p, guest, config, read +p, guest, srs, streams # 默认 g 关系由代码动态维护，亦可在此预设示例用户 +p, admin, room, chat +p, publisher, room, chat +p, viewer, room, chat +p, guest, room, chat # g, alice, admin diff --git a/internal/server/auth_handler.go b/internal/server/auth_handler.go index b73e072..b224403 100644 --- a/internal/server/auth_handler.go +++ b/internal/server/auth_handler.go @@ -214,9 +214,22 @@ func (s *Server) handleAuthCheck(w http.ResponseWriter, r *http.Request) { return } perms := map[string]bool{} - for _, act := range []string{"list", "publish", "subscribe", "watch", "stop"} { - ok, _ := s.auth.Check(claims.Username, claims.Role, "room", act) - perms["room:"+act] = ok + for _, c := range []struct { + key, obj, act string + }{ + {"room:list", "room", "list"}, + {"room:publish", "room", "publish"}, + {"room:subscribe", "room", "subscribe"}, + {"room:watch", "room", "watch"}, + {"room:stop", "room", "stop"}, + {"room:chat", "room", "chat"}, + {"user:list", "user", "list"}, + {"user:manage", "user", "manage"}, + {"config:read", "config", "read"}, + {"srs:streams", "srs", "streams"}, + } { + ok, _ := s.auth.Check(claims.Username, claims.Role, c.obj, c.act) + perms[c.key] = ok } w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(map[string]interface{}{ diff --git a/internal/server/chat.go b/internal/server/chat.go new file mode 100644 index 0000000..f2385e2 --- /dev/null +++ b/internal/server/chat.go @@ -0,0 +1,294 @@ +package server + +import ( + "encoding/json" + "fmt" + "log" + "net" + "net/http" + "strings" + "sync" + "time" +) + +// chatMessage 是房间弹幕/聊天的一条广播消息。 +type chatMessage struct { + ID string `json:"id"` + Room string `json:"room"` + User string `json:"user"` + Role string `json:"role"` + Message string `json:"message"` + Color string `json:"color,omitempty"` + Host bool `json:"host,omitempty"` + TS int64 `json:"ts"` +} + +// chatHub 维护房间 -> 弹幕订阅者的内存广播（与 roomHub 同构，重启不恢复历史）。 +type chatHub struct { + mu sync.RWMutex + rooms map[string]map[chan []byte]struct{} + // recentMsgs 保存每个房间最近若干条弹幕，用于新订阅者进场回放（重启不恢复，符合内存态设计）。 + recentMsgs map[string][]*chatMessage +} + +// recentLimit 是单房间保留的弹幕历史条数上限。 +const recentLimit = 50 + +func newChatHub() *chatHub { + return &chatHub{rooms: map[string]map[chan []byte]struct{}{}, recentMsgs: map[string][]*chatMessage{}} +} + +func (c *chatHub) subscribe(room string) (chan []byte, func()) { + c.mu.Lock() + defer c.mu.Unlock() + ch := make(chan []byte, 16) + if c.rooms[room] == nil { + c.rooms[room] = map[chan []byte]struct{}{} + } + if c.recentMsgs[room] == nil { + c.recentMsgs[room] = nil + } + c.rooms[room][ch] = struct{}{} + return ch, func() { + c.mu.Lock() + defer c.mu.Unlock() + if m, ok := c.rooms[room]; ok { + delete(m, ch) + if len(m) == 0 { + delete(c.rooms, room) + // 房间再无订阅者时清空历史，避免内存随房间数无界增长。 + delete(c.recentMsgs, room) + } + } + } +} + +// recent 返回房间最近若干条弹幕的快照（副本，调用方无需持有锁）。 +func (c *chatHub) recent(room string) []*chatMessage { + c.mu.RLock() + defer c.mu.RUnlock() + out := make([]*chatMessage, len(c.recentMsgs[room])) + copy(out, c.recentMsgs[room]) + return out +} + +func (c *chatHub) broadcast(room string, msg *chatMessage) { + data, err := json.Marshal(msg) + if err != nil { + return + } + c.mu.Lock() + c.recentMsgs[room] = append(c.recentMsgs[room], msg) + if n := len(c.recentMsgs[room]); n > recentLimit { + c.recentMsgs[room] = c.recentMsgs[room][n-recentLimit:] + } + for ch := range c.rooms[room] { + select { + case ch <- data: + default: + } + } + c.mu.Unlock() +} + +// chatRateLimiter 是弹幕发送的滑动窗口限流（按 房间 + 发送者 + IP 维度），用于防刷屏。 +// 纯内存实现，重启即清空，与弹幕本身的内存态一致；后台定期清理过期 key 防止 map 无界增长。 +type chatRateLimiter struct { + mu sync.Mutex + win map[string][]time.Time + limit int + window time.Duration +} + +func newChatRateLimiter(limit int, window time.Duration) *chatRateLimiter { + r := &chatRateLimiter{win: map[string][]time.Time{}, limit: limit, window: window} + go r.cleanup() + return r +} + +// allow 在窗口内计数未超上限时记入并返回 true，否则返回 false（应拒绝本次发送）。 +func (r *chatRateLimiter) allow(key string, now time.Time) bool { + r.mu.Lock() + defer r.mu.Unlock() + cut := now.Add(-r.window) + ts := r.win[key] + kept := ts[:0] + for _, t := range ts { + if t.After(cut) { + kept = append(kept, t) + } + } + if len(kept) >= r.limit { + r.win[key] = kept + return false + } + r.win[key] = append(kept, now) + return true +} + +// cleanup 周期性删除已无活跃计数的 key。 +func (r *chatRateLimiter) cleanup() { + ticker := time.NewTicker(time.Minute) + defer ticker.Stop() + for range ticker.C { + r.mu.Lock() + cut := time.Now().Add(-r.window) + for k, ts := range r.win { + if len(ts) == 0 || ts[len(ts)-1].Before(cut) { + delete(r.win, k) + } + } + r.mu.Unlock() + } +} + +// clientIP 从代理头或 RemoteAddr 取客户端 IP（取 X-Forwarded-For 第一段）。 +func clientIP(r *http.Request) string { + if x := r.Header.Get("X-Forwarded-For"); x != "" { + if i := strings.IndexByte(x, ','); i >= 0 { + return strings.TrimSpace(x[:i]) + } + return strings.TrimSpace(x) + } + if x := r.Header.Get("X-Real-IP"); x != "" { + return strings.TrimSpace(x) + } + host, _, err := net.SplitHostPort(r.RemoteAddr) + if err != nil { + return r.RemoteAddr + } + return host +} + +// handleRoomChat 以 SSE 推送房间弹幕（前端据此渲染叠加层 / 聊天列表）。 +func (s *Server) handleRoomChat(w http.ResponseWriter, r *http.Request) { + room := r.PathValue("room") + flusher, ok := w.(http.Flusher) + if !ok { + http.Error(w, "streaming unsupported", http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + ch, unsub := s.chat.subscribe(room) + defer unsub() + // 进场回放最近若干条弹幕，避免新订阅者面对空房间（非阻塞，填满通道即止）。 + for _, m := range s.chat.recent(room) { + b, err := json.Marshal(m) + if err != nil { + continue + } + select { + case ch <- b: + default: + break + } + } + ticker := time.NewTicker(25 * time.Second) + defer ticker.Stop() + for { + select { + case <-r.Context().Done(): + return + case msg := <-ch: + fmt.Fprintf(w, "event: chat\ndata: %s\n\n", msg) + flusher.Flush() + case <-ticker.C: + fmt.Fprintf(w, ": ping\n\n") + flusher.Flush() + } + } +} + +// handleRoomChatPost 接收一条弹幕并广播给房间内所有订阅者。 +func (s *Server) handleRoomChatPost(w http.ResponseWriter, r *http.Request) { + room := r.PathValue("room") + var body struct { + Message string `json:"message"` + Color string `json:"color"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil || body.Message == "" { + http.Error(w, "bad request", http.StatusBadRequest) + return + } + if len(body.Message) > 500 { + body.Message = body.Message[:500] + } + // 限流：按 房间 + 发送者 + IP 维度做滑动窗口，防刷屏。 + user, role, id := "guest", "guest", "guest" + if s.auth != nil { + if _, u, err := s.auth.VerifyToken(extractAuthToken(r)); err == nil { + user, role, id = u.Username, u.Role, u.Username + } + } + if s.chatRL != nil { + key := room + "/" + id + "/" + clientIP(r) + if !s.chatRL.allow(key, time.Now()) { + http.Error(w, "too many messages, please slow down", http.StatusTooManyRequests) + return + } + } + msg := &chatMessage{ + ID: randomID(), + Room: room, + User: user, + Role: role, + Message: body.Message, + Color: body.Color, + TS: time.Now().Unix(), + } + s.chat.broadcast(room, msg) + log.Printf("[chat] room=%s user=%s role=%s: %s", room, user, role, body.Message) + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{"ok": true, "id": msg.ID}) +} + +// handleRoomBroadcast 是主播的独立外部推送接口：仅 publisher/admin 可调用（authWrap 已校验 room:publish）， +// 推送的弹幕标记 host=true，经同一 chatHub 广播，主播端与观众端实时可见。供 OBS 脚本 / 机器人 / 管理工具使用。 +func (s *Server) handleRoomBroadcast(w http.ResponseWriter, r *http.Request) { + room := r.PathValue("room") + var body struct { + Message string `json:"message"` + Color string `json:"color"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil || body.Message == "" { + http.Error(w, "bad request", http.StatusBadRequest) + return + } + if len(body.Message) > 500 { + body.Message = body.Message[:500] + } + if s.chat == nil { + http.Error(w, "chat not available", http.StatusServiceUnavailable) + return + } + // 身份来自 token（外部工具用 Authorization: Bearer 调用） + user, role, id := "guest", "guest", "guest" + if s.auth != nil { + if _, u, err := s.auth.VerifyToken(extractAuthToken(r)); err == nil { + user, role, id = u.Username, u.Role, u.Username + } + } + if s.chatRL != nil { + key := room + "/broadcast/" + id + "/" + clientIP(r) + if !s.chatRL.allow(key, time.Now()) { + http.Error(w, "too many messages, please slow down", http.StatusTooManyRequests) + return + } + } + msg := &chatMessage{ + ID: randomID(), + Room: room, + User: user, + Role: role, + Message: body.Message, + Color: body.Color, + Host: true, + TS: time.Now().Unix(), + } + s.chat.broadcast(room, msg) + log.Printf("[chat:broadcast] room=%s user=%s role=%s: %s", room, user, role, body.Message) + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{"ok": true, "id": msg.ID}) +} diff --git a/internal/server/rooms.go b/internal/server/rooms.go index d50b807..b86b7c4 100644 --- a/internal/server/rooms.go +++ b/internal/server/rooms.go @@ -67,6 +67,14 @@ func (h *roomHub) get(name string) *roomEntry { return e } +// createRoom 确保房间占位存在（发布前即出现在列表）。 +func (h *roomHub) createRoom(name string) { + h.mu.Lock() + if _, ok := h.rooms[name]; !ok { + h.rooms[name] = &roomEntry{targets: map[string]*gen.StreamTarget{}, subs: map[chan []byte]struct{}{}} + } + h.mu.Unlock() +} func (h *roomHub) setTarget(room, backend string, t *gen.StreamTarget) { h.mu.Lock() e := h.get(room) diff --git a/internal/server/server.go b/internal/server/server.go index ce7a03f..5052334 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -3,10 +3,12 @@ package server import ( "bytes" "embed" + "encoding/json" "io/fs" "log" "net" "net/http" + "strings" "time" "google.golang.org/grpc" @@ -23,12 +25,15 @@ var staticFS embed.FS // Server 聚合控制面（gRPC + JSON 网关）、媒体面反向代理与静态 UI，并集成登录与 Casbin 鉴权。 type Server struct { - cfg *config.Config - svc *Service - hub *roomHub - srsProxy http.Handler - cfProxy http.Handler - auth *auth.Manager + cfg *config.Config + svc *Service + hub *roomHub + srsProxy http.Handler + cfProxy http.Handler + srsHlsProxy http.Handler + auth *auth.Manager + chat *chatHub + chatRL *chatRateLimiter } func New(cfg *config.Config) *Server { @@ -50,17 +55,16 @@ func New(cfg *config.Config) *Server { s := &Server{cfg: cfg, svc: svc, hub: hub} s.srsProxy = s.srsProxyHandler() s.cfProxy = s.cfProxyHandler() + s.srsHlsProxy = s.srsHlsProxyHandler() mgr, err := auth.NewManager(cfg.JWTSecret, cfg.JWTTTL, cfg.AuthModel, cfg.AuthPolicy, cfg.AuthUserFile) if err != nil { - log.Printf("[warn] auth manager init failed (%v), falling back to memory-only", err) - store, _ := auth.NewUserStore("") - mgr2 := auth.NewJWTManager(cfg.JWTSecret, cfg.JWTTTL) - _ = mgr2 - _ = store + log.Printf("[warn] auth manager init failed (%v), running without auth", err) } else { s.auth = mgr log.Printf("[auth] casbin enabled: model=%s policy=%s users=%s", cfg.AuthModel, cfg.AuthPolicy, cfg.AuthUserFile) svc.auth = mgr + s.chat = newChatHub() + s.chatRL = newChatRateLimiter(5, 3*time.Second) } return s } @@ -72,10 +76,13 @@ func NewWithHub(cfg *config.Config, hub *roomHub) *Server { s := &Server{cfg: cfg, svc: svc, hub: hub} s.srsProxy = s.srsProxyHandler() s.cfProxy = s.cfProxyHandler() + s.srsHlsProxy = s.srsHlsProxyHandler() mgr, err := auth.NewManager(cfg.JWTSecret, cfg.JWTTTL, cfg.AuthModel, cfg.AuthPolicy, cfg.AuthUserFile) if err == nil { s.auth = mgr svc.auth = mgr + s.chat = newChatHub() + s.chatRL = newChatRateLimiter(5, 3*time.Second) } return s } @@ -100,13 +107,30 @@ func (s *Server) StartGRPC() error { return nil } +// handleCreateRoom 创建一个房间占位（发布前即可出现在房间列表）。 +func (s *Server) handleCreateRoom(w http.ResponseWriter, r *http.Request) { + var body struct { + Name string `json:"name"` + } + _ = json.NewDecoder(r.Body).Decode(&body) + name := strings.TrimSpace(body.Name) + if name == "" { + http.Error(w, "room name required", http.StatusBadRequest) + return + } + s.hub.createRoom(name) + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{"ok": true, "room": name}) +} func (s *Server) Handler() http.Handler { mux := http.NewServeMux() sub, _ := fs.Sub(staticFS, "static") - mux.Handle("GET /", s.fileServer("static/index.html")) - mux.Handle("GET /publish", s.fileServer("static/publish.html")) - mux.Handle("GET /watch", s.fileServer("static/watch.html")) - mux.Handle("GET /login", s.fileServer("static/login.html")) + index := s.fileServer("static/index.html") + // SPA 入口：发布/观看/登录均为前端路由，统一回退 index.html + mux.Handle("GET /", index) + mux.Handle("GET /publish", index) + mux.Handle("GET /watch", index) + mux.Handle("GET /login", index) mux.Handle("GET /static/", http.StripPrefix("/static/", http.FileServer(http.FS(sub)))) mux.HandleFunc("POST /api/auth/login", s.handleLogin) @@ -119,11 +143,17 @@ func (s *Server) Handler() http.Handler { mux.Handle("GET /api/config", s.authWrap(http.HandlerFunc(s.handleConfig), "config", "read", false)) mux.Handle("GET /api/rooms", s.authWrap(http.HandlerFunc(s.handleRooms), "room", "list", true)) + mux.Handle("POST /api/rooms", s.authWrap(http.HandlerFunc(s.handleCreateRoom), "room", "list", true)) mux.Handle("POST /api/publish", s.authWrap(http.HandlerFunc(s.handlePublish), "room", "publish", true)) mux.Handle("POST /api/subscribe", s.authWrap(http.HandlerFunc(s.handleSubscribe), "room", "subscribe", true)) mux.Handle("POST /api/stop", s.authWrap(http.HandlerFunc(s.handleStop), "room", "stop", true)) mux.Handle("GET /api/srs/streams", s.authWrap(http.HandlerFunc(s.handleSRSStreams), "srs", "streams", true)) mux.Handle("GET /api/room/{room}/events", s.authWrap(http.HandlerFunc(s.handleRoomEvents), "room", "watch", true)) + mux.Handle("GET /api/room/{room}/chat", s.authWrap(http.HandlerFunc(s.handleRoomChat), "room", "chat", false)) + // 发送与 SSE 订阅一致：guest 亦可发弹幕（RBAC 已授权 room:chat），needAuth=false + mux.Handle("POST /api/room/{room}/chat", s.authWrap(http.HandlerFunc(s.handleRoomChatPost), "room", "chat", false)) + // 主播独立外部推送接口：仅 publisher/admin（room:publish），供 OBS/机器人/管理工具以 Bearer token 调用，消息标记 host + mux.Handle("POST /api/room/{room}/broadcast", s.authWrap(http.HandlerFunc(s.handleRoomBroadcast), "room", "publish", true)) mux.Handle("GET /rtc/v1/", s.authWrap(s.srsProxy, "room", "publish", false)) mux.Handle("POST /rtc/v1/", s.authWrap(s.srsProxy, "room", "publish", false)) @@ -134,6 +164,9 @@ func (s *Server) Handler() http.Handler { mux.Handle("POST /api/cf/", s.authWrap(s.cfProxy, "room", "publish", false)) mux.Handle("PUT /api/cf/", s.authWrap(s.cfProxy, "room", "publish", false)) mux.Handle("DELETE /api/cf/", s.authWrap(s.cfProxy, "room", "publish", false)) + // SRS HLS/FLV 静态切片反代（:8080），前端经 /live/* 拉流 + mux.Handle("GET /live/", s.authWrap(s.srsHlsProxy, "srs", "streams", false)) + mux.Handle("HEAD /live/", s.authWrap(s.srsHlsProxy, "srs", "streams", false)) return mux } @@ -160,4 +193,4 @@ func (s *Server) fileServer(name string) http.HandlerFunc { } http.ServeContent(w, r, name, time.Time{}, bytes.NewReader(data)) } -} \ No newline at end of file +}