diff --git a/api/docs/docs.go b/api/docs/docs.go index 125b8da..91df343 100644 --- a/api/docs/docs.go +++ b/api/docs/docs.go @@ -1941,6 +1941,10 @@ const docTemplate = `{ "name": { "type": "string" }, + "playback_auth": { + "description": "PlaybackAuth overrides the global media-auth default policy for this\nstream: \"public\" (no token) or \"token\" (signed token required). Empty\ninherits auth.media.default_policy. Static rules (IP/country/UA/domains)\nremain global. See internal/mediaauth.", + "type": "string" + }, "protocols": { "description": "Protocols defines which delivery protocols are opened for this stream.\nnil means the field is unset and ResolveStream inherits the template's\nProtocols (or leaves the resolved value nil when no template applies —\npublisher treats nil as \"no protocols enabled\"). An explicit non-nil\npointer — including the zero value \u0026OutputProtocols{} — is an\noperator-asserted override and beats template inheritance.", "allOf": [ @@ -2104,6 +2108,9 @@ const docTemplate = `{ "properties": { "api": { "$ref": "#/definitions/config.APIAuthConfig" + }, + "media": { + "$ref": "#/definitions/config.MediaAuthConfig" } } }, @@ -2227,6 +2234,65 @@ const docTemplate = `{ } } }, + "config.MediaAuthConfig": { + "type": "object", + "properties": { + "allow_countries": { + "type": "array", + "items": { + "type": "string" + } + }, + "allow_ips": { + "description": "Static allow/deny chain. IPs accept exact addresses or CIDR ranges;\nCountries are ISO 3166-1 alpha-2 codes (need a GeoIP DB — see\nSessionsConfig.GeoIPDBPath); UserAgents match case-insensitive substring;\nAllowedDomains match the Referer host (exact or parent domain).", + "type": "array", + "items": { + "type": "string" + } + }, + "allow_user_agents": { + "type": "array", + "items": { + "type": "string" + } + }, + "allowed_domains": { + "type": "array", + "items": { + "type": "string" + } + }, + "default_policy": { + "description": "DefaultPolicy is \"public\" (no token needed) or \"token\" (signed token\nrequired) for streams that don't set their own. Empty = public.", + "type": "string" + }, + "deny_countries": { + "type": "array", + "items": { + "type": "string" + } + }, + "deny_ips": { + "type": "array", + "items": { + "type": "string" + } + }, + "deny_user_agents": { + "type": "array", + "items": { + "type": "string" + } + }, + "enabled": { + "type": "boolean" + }, + "token_secret": { + "description": "TokenSecret is the HMAC-SHA256 key the server uses to VERIFY playback\ntokens. Clients (your app) mint tokens with the same secret — the server\nnever issues them. Required when any stream's effective policy is \"token\".\nSee internal/mediaauth.SignToken for the canonical token format.", + "type": "string" + } + } + }, "config.PublisherConfig": { "type": "object", "properties": { @@ -3242,6 +3308,10 @@ const docTemplate = `{ "description": "Name and Description are template-level metadata. Name surfaces in\nthe API for human-readable lists; Description carries the rationale\nbehind the template's settings. Streams inheriting this template\nkeep their own Name / Description fields — the template metadata is\nfor operator-facing tooling, not for downstream consumers.", "type": "string" }, + "playback_auth": { + "description": "PlaybackAuth is the media-auth policy (\"public\"/\"token\") inherited by\nstreams referencing this template. Empty = inherit global default.", + "type": "string" + }, "prefixes": { "description": "Prefixes is the list of URL-path prefixes that trigger auto-publish.\nWhen an encoder pushes to a path whose first segment(s) match any\nprefix here AND this template has at least one publish:// input, a\nruntime stream is created on the fly. Prefixes must not overlap any\nother template's prefix (validated at save time).", "type": "array", diff --git a/api/docs/swagger.json b/api/docs/swagger.json index bc3160e..b6d6fa5 100644 --- a/api/docs/swagger.json +++ b/api/docs/swagger.json @@ -1934,6 +1934,10 @@ "name": { "type": "string" }, + "playback_auth": { + "description": "PlaybackAuth overrides the global media-auth default policy for this\nstream: \"public\" (no token) or \"token\" (signed token required). Empty\ninherits auth.media.default_policy. Static rules (IP/country/UA/domains)\nremain global. See internal/mediaauth.", + "type": "string" + }, "protocols": { "description": "Protocols defines which delivery protocols are opened for this stream.\nnil means the field is unset and ResolveStream inherits the template's\nProtocols (or leaves the resolved value nil when no template applies —\npublisher treats nil as \"no protocols enabled\"). An explicit non-nil\npointer — including the zero value \u0026OutputProtocols{} — is an\noperator-asserted override and beats template inheritance.", "allOf": [ @@ -2097,6 +2101,9 @@ "properties": { "api": { "$ref": "#/definitions/config.APIAuthConfig" + }, + "media": { + "$ref": "#/definitions/config.MediaAuthConfig" } } }, @@ -2220,6 +2227,65 @@ } } }, + "config.MediaAuthConfig": { + "type": "object", + "properties": { + "allow_countries": { + "type": "array", + "items": { + "type": "string" + } + }, + "allow_ips": { + "description": "Static allow/deny chain. IPs accept exact addresses or CIDR ranges;\nCountries are ISO 3166-1 alpha-2 codes (need a GeoIP DB — see\nSessionsConfig.GeoIPDBPath); UserAgents match case-insensitive substring;\nAllowedDomains match the Referer host (exact or parent domain).", + "type": "array", + "items": { + "type": "string" + } + }, + "allow_user_agents": { + "type": "array", + "items": { + "type": "string" + } + }, + "allowed_domains": { + "type": "array", + "items": { + "type": "string" + } + }, + "default_policy": { + "description": "DefaultPolicy is \"public\" (no token needed) or \"token\" (signed token\nrequired) for streams that don't set their own. Empty = public.", + "type": "string" + }, + "deny_countries": { + "type": "array", + "items": { + "type": "string" + } + }, + "deny_ips": { + "type": "array", + "items": { + "type": "string" + } + }, + "deny_user_agents": { + "type": "array", + "items": { + "type": "string" + } + }, + "enabled": { + "type": "boolean" + }, + "token_secret": { + "description": "TokenSecret is the HMAC-SHA256 key the server uses to VERIFY playback\ntokens. Clients (your app) mint tokens with the same secret — the server\nnever issues them. Required when any stream's effective policy is \"token\".\nSee internal/mediaauth.SignToken for the canonical token format.", + "type": "string" + } + } + }, "config.PublisherConfig": { "type": "object", "properties": { @@ -3235,6 +3301,10 @@ "description": "Name and Description are template-level metadata. Name surfaces in\nthe API for human-readable lists; Description carries the rationale\nbehind the template's settings. Streams inheriting this template\nkeep their own Name / Description fields — the template metadata is\nfor operator-facing tooling, not for downstream consumers.", "type": "string" }, + "playback_auth": { + "description": "PlaybackAuth is the media-auth policy (\"public\"/\"token\") inherited by\nstreams referencing this template. Empty = inherit global default.", + "type": "string" + }, "prefixes": { "description": "Prefixes is the list of URL-path prefixes that trigger auto-publish.\nWhen an encoder pushes to a path whose first segment(s) match any\nprefix here AND this template has at least one publish:// input, a\nruntime stream is created on the fly. Prefixes must not overlap any\nother template's prefix (validated at save time).", "type": "array", diff --git a/api/docs/swagger.yaml b/api/docs/swagger.yaml index 2c67c47..bb2662c 100644 --- a/api/docs/swagger.yaml +++ b/api/docs/swagger.yaml @@ -213,6 +213,13 @@ definitions: type: array name: type: string + playback_auth: + description: |- + PlaybackAuth overrides the global media-auth default policy for this + stream: "public" (no token) or "token" (signed token required). Empty + inherits auth.media.default_policy. Static rules (IP/country/UA/domains) + remain global. See internal/mediaauth. + type: string protocols: allOf: - $ref: '#/definitions/domain.OutputProtocols' @@ -331,6 +338,8 @@ definitions: properties: api: $ref: '#/definitions/config.APIAuthConfig' + media: + $ref: '#/definitions/config.MediaAuthConfig' type: object config.BufferConfig: properties: @@ -442,6 +451,56 @@ definitions: healthy primary will be falsely failed over to a lower priority. type: integer type: object + config.MediaAuthConfig: + properties: + allow_countries: + items: + type: string + type: array + allow_ips: + description: |- + Static allow/deny chain. IPs accept exact addresses or CIDR ranges; + Countries are ISO 3166-1 alpha-2 codes (need a GeoIP DB — see + SessionsConfig.GeoIPDBPath); UserAgents match case-insensitive substring; + AllowedDomains match the Referer host (exact or parent domain). + items: + type: string + type: array + allow_user_agents: + items: + type: string + type: array + allowed_domains: + items: + type: string + type: array + default_policy: + description: |- + DefaultPolicy is "public" (no token needed) or "token" (signed token + required) for streams that don't set their own. Empty = public. + type: string + deny_countries: + items: + type: string + type: array + deny_ips: + items: + type: string + type: array + deny_user_agents: + items: + type: string + type: array + enabled: + type: boolean + token_secret: + description: |- + TokenSecret is the HMAC-SHA256 key the server uses to VERIFY playback + tokens. Clients (your app) mint tokens with the same secret — the server + never issues them. Required when any stream's effective policy is "token". + See internal/mediaauth.SignToken for the canonical token format. + type: string + type: object config.PublisherConfig: properties: dash: @@ -1405,6 +1464,11 @@ definitions: keep their own Name / Description fields — the template metadata is for operator-facing tooling, not for downstream consumers. type: string + playback_auth: + description: |- + PlaybackAuth is the media-auth policy ("public"/"token") inherited by + streams referencing this template. Empty = inherit global default. + type: string prefixes: description: |- Prefixes is the list of URL-path prefixes that trigger auto-publish. diff --git a/cmd/server/main.go b/cmd/server/main.go index 8df5c65..46148f9 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -30,6 +30,7 @@ import ( "github.com/ntt0601zcoder/open-streamer/internal/hooks" "github.com/ntt0601zcoder/open-streamer/internal/ingestor" "github.com/ntt0601zcoder/open-streamer/internal/manager" + "github.com/ntt0601zcoder/open-streamer/internal/mediaauth" "github.com/ntt0601zcoder/open-streamer/internal/metrics" "github.com/ntt0601zcoder/open-streamer/internal/publisher" "github.com/ntt0601zcoder/open-streamer/internal/runtime" @@ -133,6 +134,11 @@ func run() error { // 5. Wire all services. wireServices(injector) + // 5b. Wire the media-plane playback authorizer (token / IP / country / UA / + // referer) into the publisher + API server. Returns the shared authorizer + // so the runtime config-reload path can hot-swap its rules. + mediaAuthz := wireMediaAuth(injector, gcfg) + // 6. Assemble RuntimeManager deps from DI. rtm := runtime.New(ctx, runtime.Deps{ Ingestor: do.MustInvoke[*ingestor.Service](injector), @@ -144,6 +150,7 @@ func run() error { SessionsSvc: do.MustInvoke[*sessions.Service](injector), AutoPublish: do.MustInvoke[*autopublish.Service](injector), APISrv: do.MustInvoke[*api.Server](injector), + MediaAuth: mediaAuthz, Bus: do.MustInvoke[events.Bus](injector), StreamRepo: do.MustInvoke[store.StreamRepository](injector), GlobalConfigRepo: gcRepo, @@ -330,6 +337,48 @@ func wireServices(i *do.RootScope) { // happens at packet-read time, long after the original request that created // the worker has been served. The repo's FindByCode is fast (in-memory or // indexed), so blocking briefly here is acceptable. +// wireMediaAuth builds the shared media-plane playback authorizer and injects +// it into the publisher (RTMP/SRT/RTSP play) and API server (HLS/DASH/MPEGTS). +// The country backend reuses the sessions GeoIP resolver; the per-stream policy +// is resolved from the publisher's in-memory stream table (O(1), no store hit). +func wireMediaAuth(i do.Injector, gcfg *domain.GlobalConfig) *mediaauth.Authorizer { + pub := do.MustInvoke[*publisher.Service](i) + apiSrv := do.MustInvoke[*api.Server](i) + geoIP := do.MustInvoke[*sessions.SwappableGeoIP](i) + streamRepo := do.MustInvoke[store.StreamRepository](i) + templateRepo := do.MustInvoke[store.TemplateRepository](i) + + // policyFor resolves a stream's effective playback policy. Live streams hit + // the publisher's in-memory table (O(1), no store read on the hot path). A + // STOPPED stream — whose DVR archive is still served — falls back to the + // store + template so its per-stream `token` policy isn't silently + // downgraded to the global default. + policyFor := func(code domain.StreamCode) string { + if policy, running := pub.PlaybackPolicy(code); running { + return policy + } + s, err := streamRepo.FindByCode(context.Background(), code) + if err != nil { + return "" + } + if s.Template != nil { + if tpl, terr := templateRepo.FindByCode(context.Background(), *s.Template); terr == nil { + s = domain.ResolveStream(s, tpl) + } + } + return s.PlaybackAuth + } + + cfg := config.AuthConfig{} + if gcfg.Auth != nil { + cfg = *gcfg.Auth + } + authz := mediaauth.New(cfg.Media, geoIP.Country, policyFor) + pub.SetMediaAuthorizer(authz) + apiSrv.SetMediaAuthorizer(authz) + return authz +} + func wireCopyLookup(i do.Injector) { ing := do.MustInvoke[*ingestor.Service](i) repo := do.MustInvoke[store.StreamRepository](i) diff --git a/config/config.go b/config/config.go index 4920276..82b33fd 100644 --- a/config/config.go +++ b/config/config.go @@ -246,10 +246,47 @@ type HooksConfig struct { BatchMaxQueueItems int `mapstructure:"batch_max_queue_items" json:"batch_max_queue_items,omitempty" yaml:"batch_max_queue_items,omitempty"` } -// AuthConfig holds authentication settings. Today it covers only the -// control-plane HTTP API (admin); media/playback auth is a separate plane. +// AuthConfig holds authentication settings: the control-plane HTTP API (admin) +// and the media/playback plane (who may watch a stream). type AuthConfig struct { - API APIAuthConfig `mapstructure:"api" json:"api" yaml:"api"` + API APIAuthConfig `mapstructure:"api" json:"api" yaml:"api"` + Media MediaAuthConfig `mapstructure:"media" json:"media" yaml:"media"` +} + +// MediaAuthConfig configures playback (media-plane) authorization: who may +// watch a stream over HLS/DASH/HTTP-MPEGTS/RTMP/SRT/RTSP. Disabled by default +// (Enabled=false → every request is allowed, preserving prior behaviour). +// +// Evaluation per request (a chain; deny wins, allow-lists restrict, then the +// policy gate): a value on any Deny* list → reject; for any non-empty Allow* +// list the request's value MUST appear on it or it is rejected; finally if the +// effective policy (per-stream override, else DefaultPolicy) is "token" a valid +// signed token is required. AllowedDomains gates the HTTP Referer for embeds. +// Static rules are global; per-stream config overrides only the policy today. +type MediaAuthConfig struct { + Enabled bool `mapstructure:"enabled" json:"enabled" yaml:"enabled"` + + // DefaultPolicy is "public" (no token needed) or "token" (signed token + // required) for streams that don't set their own. Empty = public. + DefaultPolicy string `mapstructure:"default_policy" json:"default_policy,omitempty" yaml:"default_policy,omitempty"` + + // TokenSecret is the HMAC-SHA256 key the server uses to VERIFY playback + // tokens. Clients (your app) mint tokens with the same secret — the server + // never issues them. Required when any stream's effective policy is "token". + // See internal/mediaauth.SignToken for the canonical token format. + TokenSecret string `mapstructure:"token_secret" json:"token_secret,omitempty" yaml:"token_secret,omitempty"` + + // Static allow/deny chain. IPs accept exact addresses or CIDR ranges; + // Countries are ISO 3166-1 alpha-2 codes (need a GeoIP DB — see + // SessionsConfig.GeoIPDBPath); UserAgents match case-insensitive substring; + // AllowedDomains match the Referer host (exact or parent domain). + AllowIPs []string `mapstructure:"allow_ips" json:"allow_ips,omitempty" yaml:"allow_ips,omitempty"` + DenyIPs []string `mapstructure:"deny_ips" json:"deny_ips,omitempty" yaml:"deny_ips,omitempty"` + AllowCountries []string `mapstructure:"allow_countries" json:"allow_countries,omitempty" yaml:"allow_countries,omitempty"` + DenyCountries []string `mapstructure:"deny_countries" json:"deny_countries,omitempty" yaml:"deny_countries,omitempty"` + AllowUserAgents []string `mapstructure:"allow_user_agents" json:"allow_user_agents,omitempty" yaml:"allow_user_agents,omitempty"` + DenyUserAgents []string `mapstructure:"deny_user_agents" json:"deny_user_agents,omitempty" yaml:"deny_user_agents,omitempty"` + AllowedDomains []string `mapstructure:"allowed_domains" json:"allowed_domains,omitempty" yaml:"allowed_domains,omitempty"` } // APIAuthConfig configures HTTP Basic auth on the management API. When diff --git a/docs/audit/security-quality-audit.md b/docs/audit/security-quality-audit.md index 0fcf05a..7d0eff0 100644 --- a/docs/audit/security-quality-audit.md +++ b/docs/audit/security-quality-audit.md @@ -193,6 +193,8 @@ Covered as the HTTP-sink half of **S-2**. `batcher.go:414-446` POSTs to any oper **Fix:** In `srtStreamCode`, strip the query before validation (`if i := strings.IndexByte(streamid, '?'); i >= 0 { streamid = streamid[:i] }`); add table tests. Soften the misleading "auth token" comments. Real playback authz remains the tracked backlog item. +> ✅ **FIXED** in `fix/media-playback-auth` (B / media-plane) — playback authorization now exists and is enforced across all delivery protocols. New `internal/mediaauth` package: a Flussonic-style chain (deny IP/country/UA → allow-list gates → per-stream token-policy gate) with **client-signed, server-verified** HMAC-SHA256 tokens (`token = "."`; constant-time verify, expiry- and stream-code-bound; clients mint via the documented `SignToken` format — there is no server mint endpoint). Wired before any bytes are served: HTTP (HLS/DASH/MPEGTS/DVR) in `dispatchMedia`, plus RTMP/SRT/RTSP play sites. Effective policy = per-stream `Stream.PlaybackAuth` (`public`/`token`, template-inherited) else global `auth.media.default_policy`; static rules (IP/country via the sessions GeoIP, UA, referer/allowed-domains) are global. Disabled by default (no behaviour change); fail-closed when token policy has no secret. Config `auth.media.*`, hot-reloaded. The SRT `?token=` parsing bug is also fixed (`srtStreamCode` strips the query) so tokened streamids resolve. ABR renditions key on the parent code (a token for `` covers `//track_N/…`); stopped-stream DVR archives resolve their policy from the store so a `token` stream isn't downgraded. **Adversarially reviewed** (bypass / IP-spoof / per-protocol gaps) and the findings fixed. Tests: `internal/mediaauth` chain+token suite, `TestStripABRTrackSlug`. **Caveats (separate findings):** IP/country rules trust `r.RemoteAddr` from `RealIP`/XFF, so they're only sound behind a header-overwriting trusted proxy (S-15/S-17); RTMP play carries no token (IP/country still apply); the HTTP-callback (dynamic `on_play`) backend and per-stream rule overrides are a planned follow-up. + --- #### S-14 (LOW) — Subprocess gRPC unix socket is unauthenticated diff --git a/internal/api/dispatch.go b/internal/api/dispatch.go index 89f6c01..1cce158 100644 --- a/internal/api/dispatch.go +++ b/internal/api/dispatch.go @@ -1,6 +1,8 @@ package api import ( + "log/slog" + "net" "net/http" "strings" @@ -8,9 +10,62 @@ import ( "github.com/ntt0601zcoder/open-streamer/internal/domain" "github.com/ntt0601zcoder/open-streamer/internal/dvr/blob" + "github.com/ntt0601zcoder/open-streamer/internal/mediaauth" "github.com/ntt0601zcoder/open-streamer/internal/mediaserve" ) +// mediaAllowed runs the media-plane authorization chain for an HTTP playback +// request (HLS/DASH/MPEGTS). Returns true (allow) when no authorizer is wired +// or media auth is disabled. The client IP comes from r.RemoteAddr, which the +// RealIP middleware has already set from the trusted proxy's forwarded header. +func (s *Server) mediaAllowed(r *http.Request, code domain.StreamCode) bool { + if s.mediaAuth == nil { + return true + } + host, _, err := net.SplitHostPort(r.RemoteAddr) + if err != nil { + host = r.RemoteAddr + } + d := s.mediaAuth.Authorize(mediaauth.AuthRequest{ + // Key auth on the parent stream: ABR renditions live under + // //track_/… so a token minted for must cover them. + Code: stripABRTrackSlug(code), + ClientIP: net.ParseIP(host), + Token: r.URL.Query().Get("token"), + UserAgent: r.UserAgent(), + Referer: r.Referer(), + }) + if !d.Allow { + slog.Info("api: playback denied", "stream_code", code, "reason", d.Reason, "remote", r.RemoteAddr) + } + return d.Allow +} + +// stripABRTrackSlug removes a trailing "/track_" rendition segment so media +// auth keys on the parent stream code — a token minted for covers all +// of its ABR renditions (served at //track_/…). Mirrors the sessions +// tracker's path handling; the slug is a closed internal namespace. +func stripABRTrackSlug(code domain.StreamCode) domain.StreamCode { + s := string(code) + if i := strings.LastIndexByte(s, '/'); i > 0 && isABRTrackSlug(s[i+1:]) { + return domain.StreamCode(s[:i]) + } + return code +} + +func isABRTrackSlug(s string) bool { + const prefix = "track_" + if !strings.HasPrefix(s, prefix) || len(s) == len(prefix) { + return false + } + for _, c := range s[len(prefix):] { + if c < '0' || c > '9' { + return false + } + } + return true +} + // dispatchStreamsSubpath handles every URL under /streams/<...>. The chi // catch-all captures the suffix after /streams/ in the "*" URL param. // @@ -103,6 +158,13 @@ func (s *Server) dispatchMedia() http.HandlerFunc { http.NotFound(w, r) return } + // Media-plane authorization (token / IP / country / UA / referer) for + // all HTTP playback (HLS / DASH / MPEGTS) — B / S-13. No-op when media + // auth is disabled. + if !s.mediaAllowed(r, domain.StreamCode(code)) { + http.Error(w, "forbidden", http.StatusForbidden) + return + } r = setURLParam(r, "code", code) switch file { diff --git a/internal/api/media_authz_test.go b/internal/api/media_authz_test.go new file mode 100644 index 0000000..0ecb9ea --- /dev/null +++ b/internal/api/media_authz_test.go @@ -0,0 +1,31 @@ +package api + +import ( + "testing" + + "github.com/ntt0601zcoder/open-streamer/internal/domain" +) + +// TestStripABRTrackSlug — media auth must key on the parent stream code, not on +// a per-rendition track_ subdir, so a token minted for covers every +// ABR rendition served at //track_/…. Regression for the adversarial +// finding where ABR segments were authed under "/track_N" (token verify +// + per-stream policy both missed). +func TestStripABRTrackSlug(t *testing.T) { + t.Parallel() + cases := []struct{ in, want string }{ + {"ch1/track_1", "ch1"}, + {"ch1/track_12", "ch1"}, + {"region/north/live/track_2", "region/north/live"}, // multi-segment code + {"ch1", "ch1"}, // no slug + {"ch1/index.m3u8", "ch1/index.m3u8"}, // not a track slug — left intact (file split happens earlier) + {"ch1/track_x", "ch1/track_x"}, // non-numeric → not a slug + {"ch1/track_", "ch1/track_"}, // empty number → not a slug + {"track_1", "track_1"}, // no parent code → unchanged + } + for _, c := range cases { + if got := string(stripABRTrackSlug(domain.StreamCode(c.in))); got != c.want { + t.Errorf("stripABRTrackSlug(%q) = %q, want %q", c.in, got, c.want) + } + } +} diff --git a/internal/api/server.go b/internal/api/server.go index 63a2441..6c4bbfe 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -19,6 +19,7 @@ import ( _ "github.com/ntt0601zcoder/open-streamer/api/docs" // swag Register(SwaggerInfo) "github.com/ntt0601zcoder/open-streamer/config" "github.com/ntt0601zcoder/open-streamer/internal/api/handler" + "github.com/ntt0601zcoder/open-streamer/internal/mediaauth" "github.com/ntt0601zcoder/open-streamer/internal/metrics" "github.com/ntt0601zcoder/open-streamer/internal/publisher" "github.com/ntt0601zcoder/open-streamer/internal/sessions" @@ -60,10 +61,18 @@ type Server struct { // Never nil (a disabled authenticator is a pass-through). apiAuth *Auth + // mediaAuth authorizes HTTP playback (HLS/DASH/MPEGTS) — token / IP / + // country / UA / referer. nil = allow all. Set via SetMediaAuthorizer. + mediaAuth *mediaauth.Authorizer + router *chi.Mux http *http.Server } +// SetMediaAuthorizer wires the playback authorizer for HTTP media routes (B / +// S-13). Called from the runtime wiring after the publisher is constructed. +func (s *Server) SetMediaAuthorizer(a *mediaauth.Authorizer) { s.mediaAuth = a } + // New creates a Server and registers it with the DI injector. // The server is constructed but not started — call StartWithConfig to begin accepting connections. func New(i do.Injector) (*Server, error) { diff --git a/internal/domain/stream.go b/internal/domain/stream.go index 041ab57..d706f18 100644 --- a/internal/domain/stream.go +++ b/internal/domain/stream.go @@ -70,6 +70,12 @@ type Stream struct { // StreamKey is used to authenticate RTMP/SRT push ingest. StreamKey string `json:"stream_key" yaml:"stream_key"` + // PlaybackAuth overrides the global media-auth default policy for this + // stream: "public" (no token) or "token" (signed token required). Empty + // inherits auth.media.default_policy. Static rules (IP/country/UA/domains) + // remain global. See internal/mediaauth. + PlaybackAuth string `json:"playback_auth,omitempty" yaml:"playback_auth,omitempty"` + // Status is the runtime lifecycle state. // It is never persisted — always computed from the coordinator's in-memory // state and overlaid by the API layer before returning responses to clients. diff --git a/internal/domain/template.go b/internal/domain/template.go index 23692f3..b852102 100644 --- a/internal/domain/template.go +++ b/internal/domain/template.go @@ -69,6 +69,10 @@ type Template struct { // stream may set its own (or none). StreamKey string `json:"stream_key,omitempty" yaml:"stream_key,omitempty"` + // PlaybackAuth is the media-auth policy ("public"/"token") inherited by + // streams referencing this template. Empty = inherit global default. + PlaybackAuth string `json:"playback_auth,omitempty" yaml:"playback_auth,omitempty"` + // Prefixes is the list of URL-path prefixes that trigger auto-publish. // When an encoder pushes to a path whose first segment(s) match any // prefix here AND this template has at least one publish:// input, a @@ -236,6 +240,9 @@ func ResolveStream(s *Stream, tpl *Template) *Stream { if out.StreamKey == "" && tpl.StreamKey != "" { out.StreamKey = tpl.StreamKey } + if out.PlaybackAuth == "" && tpl.PlaybackAuth != "" { + out.PlaybackAuth = tpl.PlaybackAuth + } if len(out.Inputs) == 0 && len(tpl.Inputs) > 0 { out.Inputs = tpl.Inputs } diff --git a/internal/mediaauth/mediaauth.go b/internal/mediaauth/mediaauth.go new file mode 100644 index 0000000..642fb3d --- /dev/null +++ b/internal/mediaauth/mediaauth.go @@ -0,0 +1,331 @@ +// Package mediaauth authorizes playback (media-plane) requests across every +// delivery protocol (HLS, DASH, HTTP-MPEGTS, RTMP, SRT, RTSP). It is the +// counterpart to the control-plane (admin) auth in internal/api: that gates +// who can CONFIGURE the server; this gates who can WATCH a stream. +// +// One Authorizer is shared by all protocol handlers. Each handler builds an +// AuthRequest from its connection and calls Authorize before allocating any +// streaming state. Evaluation is a Flussonic-style chain — deny wins, allow +// lists restrict, then a per-stream token-policy gate: +// +// 1. ClientIP / Country / User-Agent on a Deny* list → DENY +// 2. any non-empty Allow* list the request value misses → DENY +// 3. AllowedDomains set and the Referer host isn't covered → DENY +// 4. effective policy == "token" and the signed token is missing/invalid → DENY +// 5. otherwise → ALLOW +// +// Disabled (Enabled=false, the default) short-circuits to ALLOW so existing +// deployments are unaffected until an operator turns it on. Config is swapped +// atomically for hot-reload. +package mediaauth + +import ( + "crypto/hmac" + "crypto/sha256" + "crypto/subtle" + "encoding/base64" + "net" + "net/url" + "strconv" + "strings" + "sync/atomic" + "time" + + "github.com/ntt0601zcoder/open-streamer/config" + "github.com/ntt0601zcoder/open-streamer/internal/domain" +) + +// Policy values for a stream's effective playback policy. +const ( + PolicyPublic = "public" // no token required + PolicyToken = "token" // a valid signed token is required + + // clockSkew tolerates small clock differences when checking token expiry. + clockSkew = 60 * time.Second +) + +// AuthRequest is the per-request context a protocol handler hands to Authorize. +type AuthRequest struct { + Code domain.StreamCode + ClientIP net.IP // may be nil if unparseable + Token string // playback token (?token= / streamid / Authorization) + UserAgent string + Referer string +} + +// Decision is the outcome. Allow=false carries a short, non-leaky Reason for +// logs (never returned to the client verbatim). +type Decision struct { + Allow bool + Reason string +} + +func allow() Decision { return Decision{Allow: true} } +func deny(reason string) Decision { return Decision{Allow: false, Reason: reason} } + +// GeoResolver maps a client IP to an ISO 3166-1 alpha-2 country code, or "" if +// unknown / unavailable. Backed by the sessions GeoIP database. +type GeoResolver func(net.IP) string + +// PolicyResolver returns a stream's own playback policy ("public"/"token") or +// "" to inherit the global default. Backed by the publisher's in-memory stream +// table so it costs an O(1) map lookup, not a store read per segment. +type PolicyResolver func(domain.StreamCode) string + +// Authorizer evaluates the chain. Safe for concurrent use. +type Authorizer struct { + state atomic.Pointer[state] + geo GeoResolver + policy PolicyResolver +} + +// state is an immutable, pre-parsed snapshot of MediaAuthConfig. +type state struct { + enabled bool + defaultPolicy string + secret []byte + + allowNets, denyNets []*net.IPNet + allowIPs, denyIPs map[string]struct{} // exact IPs (canonical string) + allowCountries map[string]struct{} + denyCountries map[string]struct{} + allowUAs, denyUAs []string // lower-cased substrings + allowedDomains []string // lower-cased hosts + hasIPAllow bool + hasCountryAllow bool + hasUAAllow bool + hasDomainAllow bool +} + +// New builds an Authorizer. geo and policy may be nil (country rules then never +// match an allow-list, and every stream uses the global default policy). +func New(cfg config.MediaAuthConfig, geo GeoResolver, policy PolicyResolver) *Authorizer { + a := &Authorizer{geo: geo, policy: policy} + a.SetConfig(cfg) + return a +} + +// SetConfig swaps in a freshly-parsed snapshot (config hot-reload). +func (a *Authorizer) SetConfig(cfg config.MediaAuthConfig) { + a.state.Store(buildState(cfg)) +} + +func buildState(cfg config.MediaAuthConfig) *state { + st := &state{ + enabled: cfg.Enabled, + defaultPolicy: strings.ToLower(strings.TrimSpace(cfg.DefaultPolicy)), + secret: []byte(cfg.TokenSecret), + allowIPs: map[string]struct{}{}, + denyIPs: map[string]struct{}{}, + allowCountries: upperSet(cfg.AllowCountries), + denyCountries: upperSet(cfg.DenyCountries), + allowUAs: lowerList(cfg.AllowUserAgents), + denyUAs: lowerList(cfg.DenyUserAgents), + allowedDomains: lowerList(cfg.AllowedDomains), + } + st.allowNets, st.allowIPs = parseIPRules(cfg.AllowIPs) + st.denyNets, st.denyIPs = parseIPRules(cfg.DenyIPs) + st.hasIPAllow = len(st.allowNets) > 0 || len(st.allowIPs) > 0 + st.hasCountryAllow = len(st.allowCountries) > 0 + st.hasUAAllow = len(st.allowUAs) > 0 + st.hasDomainAllow = len(st.allowedDomains) > 0 + return st +} + +// Authorize runs the chain and returns the decision. +func (a *Authorizer) Authorize(req AuthRequest) Decision { + st := a.state.Load() + if st == nil || !st.enabled { + return allow() + } + + // 1. Deny lists (hard block, evaluated first). + if ipInRules(req.ClientIP, st.denyNets, st.denyIPs) { + return deny("ip on deny list") + } + // Resolve country only when a rule needs it. + country := "" + if (len(st.denyCountries) > 0 || st.hasCountryAllow) && a.geo != nil { + country = strings.ToUpper(a.geo(req.ClientIP)) + } + if country != "" { + if _, bad := st.denyCountries[country]; bad { + return deny("country on deny list") + } + } + if uaMatches(req.UserAgent, st.denyUAs) { + return deny("user-agent on deny list") + } + + // 2. Allow gates: each configured list must be satisfied. + if st.hasIPAllow && !ipInRules(req.ClientIP, st.allowNets, st.allowIPs) { + return deny("ip not on allow list") + } + if st.hasCountryAllow { + if _, ok := st.allowCountries[country]; !ok || country == "" { + return deny("country not on allow list") + } + } + if st.hasUAAllow && !uaMatches(req.UserAgent, st.allowUAs) { + return deny("user-agent not on allow list") + } + if st.hasDomainAllow && !domainAllowed(req.Referer, st.allowedDomains) { + return deny("referer domain not allowed") + } + + // 3. Policy gate: per-stream override, else global default. + policy := st.defaultPolicy + if a.policy != nil { + if p := strings.ToLower(strings.TrimSpace(a.policy(req.Code))); p != "" { + policy = p + } + } + if policy == PolicyToken { + if len(st.secret) == 0 { + return deny("token policy but no secret configured") + } + if !st.verify(req.Code, req.Token) { + return deny("missing or invalid token") + } + } + return allow() +} + +// Enabled reports whether media auth is on (for handlers / status). +func (a *Authorizer) Enabled() bool { + st := a.state.Load() + return st != nil && st.enabled +} + +// ── playback token ── + +// SignToken is the Go reference implementation of the playback-token format. +// The server only VERIFIES tokens; clients mint them with the shared secret — +// reproduce this in any language: +// +// exp = future expiry, unix seconds (decimal string) +// msg = "|" +// sig = HMAC-SHA256(secret, msg) // raw 32 bytes +// token = "." + base64url-nopad(sig) +// URL = ...//index.m3u8?token= // (or SRT streamid / RTSP query) +func SignToken(secret []byte, code domain.StreamCode, exp int64) string { + return strconv.FormatInt(exp, 10) + "." + base64.RawURLEncoding.EncodeToString(tokenMAC(secret, code, exp)) +} + +func (st *state) verify(code domain.StreamCode, token string) bool { + dot := strings.IndexByte(token, '.') + if dot <= 0 { + return false + } + exp, err := strconv.ParseInt(token[:dot], 10, 64) + if err != nil { + return false + } + if time.Now().Add(-clockSkew).Unix() > exp { + return false // expired + } + got, err := base64.RawURLEncoding.DecodeString(token[dot+1:]) + if err != nil { + return false + } + want := tokenMAC(st.secret, code, exp) + return subtle.ConstantTimeCompare(got, want) == 1 +} + +func tokenMAC(secret []byte, code domain.StreamCode, exp int64) []byte { + m := hmac.New(sha256.New, secret) + m.Write([]byte(string(code))) + m.Write([]byte{'|'}) + m.Write([]byte(strconv.FormatInt(exp, 10))) + return m.Sum(nil) +} + +// ── helpers ── + +func parseIPRules(rules []string) (nets []*net.IPNet, exact map[string]struct{}) { + exact = map[string]struct{}{} + for _, raw := range rules { + s := strings.TrimSpace(raw) + if s == "" { + continue + } + if _, n, err := net.ParseCIDR(s); err == nil { + nets = append(nets, n) + continue + } + if ip := net.ParseIP(s); ip != nil { + exact[ip.String()] = struct{}{} + } + } + return nets, exact +} + +func ipInRules(ip net.IP, nets []*net.IPNet, exact map[string]struct{}) bool { + if ip == nil { + return false + } + if _, ok := exact[ip.String()]; ok { + return true + } + for _, n := range nets { + if n.Contains(ip) { + return true + } + } + return false +} + +func uaMatches(ua string, patterns []string) bool { + if ua == "" || len(patterns) == 0 { + return false + } + lc := strings.ToLower(ua) + for _, p := range patterns { + if p != "" && strings.Contains(lc, p) { + return true + } + } + return false +} + +// domainAllowed parses the Referer and checks its host against the allow-list +// (exact match or a parent domain, e.g. "example.com" covers "play.example.com"). +func domainAllowed(referer string, domains []string) bool { + if referer == "" { + return false + } + u, err := url.Parse(referer) + if err != nil { + return false + } + host := strings.ToLower(u.Hostname()) + if host == "" { + return false + } + for _, d := range domains { + if host == d || strings.HasSuffix(host, "."+d) { + return true + } + } + return false +} + +func upperSet(in []string) map[string]struct{} { + out := make(map[string]struct{}, len(in)) + for _, s := range in { + if s = strings.ToUpper(strings.TrimSpace(s)); s != "" { + out[s] = struct{}{} + } + } + return out +} + +func lowerList(in []string) []string { + out := make([]string, 0, len(in)) + for _, s := range in { + if s = strings.ToLower(strings.TrimSpace(s)); s != "" { + out = append(out, s) + } + } + return out +} diff --git a/internal/mediaauth/mediaauth_test.go b/internal/mediaauth/mediaauth_test.go new file mode 100644 index 0000000..e3c734d --- /dev/null +++ b/internal/mediaauth/mediaauth_test.go @@ -0,0 +1,254 @@ +package mediaauth + +import ( + "net" + "testing" + "time" + + "github.com/ntt0601zcoder/open-streamer/config" + "github.com/ntt0601zcoder/open-streamer/internal/domain" +) + +const testCode = domain.StreamCode("ch1") + +func ip(s string) net.IP { return net.ParseIP(s) } + +// ── token sign/verify ── + +func TestToken_SignVerifyRoundTrip(t *testing.T) { + t.Parallel() + st := buildState(config.MediaAuthConfig{Enabled: true, TokenSecret: "sekret"}) + exp := time.Now().Add(time.Hour).Unix() + tok := SignToken(st.secret, testCode, exp) + if !st.verify(testCode, tok) { + t.Fatal("valid token failed to verify") + } +} + +func TestToken_Rejects(t *testing.T) { + t.Parallel() + st := buildState(config.MediaAuthConfig{Enabled: true, TokenSecret: "sekret"}) + valid := SignToken(st.secret, testCode, time.Now().Add(time.Hour).Unix()) + + t.Run("expired", func(t *testing.T) { + old := SignToken(st.secret, testCode, time.Now().Add(-2*time.Hour).Unix()) + if st.verify(testCode, old) { + t.Error("expired token verified") + } + }) + t.Run("wrong_stream_code", func(t *testing.T) { + if st.verify("other", valid) { + t.Error("token for ch1 verified against 'other'") + } + }) + t.Run("tampered_sig", func(t *testing.T) { + if st.verify(testCode, valid+"x") { + t.Error("tampered token verified") + } + }) + t.Run("different_secret", func(t *testing.T) { + other := buildState(config.MediaAuthConfig{Enabled: true, TokenSecret: "other"}) + if other.verify(testCode, valid) { + t.Error("token verified under a different secret") + } + }) + t.Run("malformed", func(t *testing.T) { + for _, bad := range []string{"", "noexp", "abc.def", ".sig", "999"} { + if st.verify(testCode, bad) { + t.Errorf("malformed token %q verified", bad) + } + } + }) +} + +// ── chain ── + +func newAuthz(t *testing.T, cfg config.MediaAuthConfig, geo GeoResolver, pol PolicyResolver) *Authorizer { + t.Helper() + return New(cfg, geo, pol) +} + +func TestAuthorize_DisabledAllowsAll(t *testing.T) { + t.Parallel() + a := newAuthz(t, config.MediaAuthConfig{Enabled: false, DefaultPolicy: PolicyToken}, nil, nil) + if d := a.Authorize(AuthRequest{Code: testCode}); !d.Allow { + t.Fatalf("disabled must allow, got deny: %s", d.Reason) + } +} + +func TestAuthorize_IPRules(t *testing.T) { + t.Parallel() + a := newAuthz(t, config.MediaAuthConfig{ + Enabled: true, + DenyIPs: []string{"8.8.8.8", "10.0.0.0/8"}, + AllowIPs: []string{"1.2.3.4", "192.168.0.0/16"}, + }, nil, nil) + + cases := []struct { + ip string + want bool + }{ + {"8.8.8.8", false}, // deny exact + {"10.5.5.5", false}, // deny CIDR + {"1.2.3.4", true}, // allow exact + {"192.168.1.9", true}, // allow CIDR + {"203.0.113.7", false}, // not on allow list → deny + } + for _, c := range cases { + if d := a.Authorize(AuthRequest{Code: testCode, ClientIP: ip(c.ip)}); d.Allow != c.want { + t.Errorf("ip %s: allow=%v want %v (%s)", c.ip, d.Allow, c.want, d.Reason) + } + } +} + +func TestAuthorize_CountryRules(t *testing.T) { + t.Parallel() + geo := func(p net.IP) string { + switch p.String() { + case "1.1.1.1": + return "VN" + case "2.2.2.2": + return "RU" + } + return "" // unknown + } + a := newAuthz(t, config.MediaAuthConfig{ + Enabled: true, + AllowCountries: []string{"VN", "US"}, + DenyCountries: []string{"RU"}, + }, geo, nil) + + if d := a.Authorize(AuthRequest{Code: testCode, ClientIP: ip("1.1.1.1")}); !d.Allow { + t.Errorf("VN must be allowed: %s", d.Reason) + } + if d := a.Authorize(AuthRequest{Code: testCode, ClientIP: ip("2.2.2.2")}); d.Allow { + t.Error("RU must be denied") + } + if d := a.Authorize(AuthRequest{Code: testCode, ClientIP: ip("9.9.9.9")}); d.Allow { + t.Error("unknown country must fail the allow-list gate") + } +} + +func TestAuthorize_UserAgentRules(t *testing.T) { + t.Parallel() + a := newAuthz(t, config.MediaAuthConfig{ + Enabled: true, + DenyUserAgents: []string{"badbot"}, + }, nil, nil) + if d := a.Authorize(AuthRequest{Code: testCode, UserAgent: "Mozilla BadBot/1.0"}); d.Allow { + t.Error("denied UA substring must block") + } + if d := a.Authorize(AuthRequest{Code: testCode, UserAgent: "VLC/3.0"}); !d.Allow { + t.Errorf("allowed UA must pass: %s", d.Reason) + } + + b := newAuthz(t, config.MediaAuthConfig{Enabled: true, AllowUserAgents: []string{"exoplayer"}}, nil, nil) + if d := b.Authorize(AuthRequest{Code: testCode, UserAgent: "ExoPlayerLib/2.1"}); !d.Allow { + t.Errorf("UA on allow list must pass: %s", d.Reason) + } + if d := b.Authorize(AuthRequest{Code: testCode, UserAgent: "curl/8"}); d.Allow { + t.Error("UA off allow list must be denied") + } +} + +func TestAuthorize_AllowedDomains(t *testing.T) { + t.Parallel() + a := newAuthz(t, config.MediaAuthConfig{Enabled: true, AllowedDomains: []string{"example.com"}}, nil, nil) + cases := []struct { + ref string + want bool + }{ + {"https://example.com/player", true}, + {"https://play.example.com/embed", true}, // subdomain + {"https://evil.com/", false}, + {"", false}, // no referer → fail the gate + } + for _, c := range cases { + if d := a.Authorize(AuthRequest{Code: testCode, Referer: c.ref}); d.Allow != c.want { + t.Errorf("referer %q: allow=%v want %v", c.ref, d.Allow, c.want) + } + } +} + +func TestAuthorize_TokenPolicy(t *testing.T) { + t.Parallel() + a := newAuthz(t, config.MediaAuthConfig{Enabled: true, DefaultPolicy: PolicyToken, TokenSecret: "sk"}, nil, nil) + good := SignToken([]byte("sk"), testCode, time.Now().Add(time.Hour).Unix()) + if d := a.Authorize(AuthRequest{Code: testCode, Token: good}); !d.Allow { + t.Errorf("valid token must pass: %s", d.Reason) + } + if d := a.Authorize(AuthRequest{Code: testCode, Token: ""}); d.Allow { + t.Error("missing token under token policy must deny") + } + if d := a.Authorize(AuthRequest{Code: testCode, Token: "bogus.sig"}); d.Allow { + t.Error("bogus token must deny") + } + // A token minted for ch1 must not work for another stream. + if d := a.Authorize(AuthRequest{Code: "ch2", Token: good}); d.Allow { + t.Error("token bound to ch1 must not authorize ch2") + } +} + +func TestAuthorize_TokenPolicyNoSecretFailsClosed(t *testing.T) { + t.Parallel() + a := newAuthz(t, config.MediaAuthConfig{Enabled: true, DefaultPolicy: PolicyToken}, nil, nil) // no secret + if d := a.Authorize(AuthRequest{Code: testCode, Token: "x.y"}); d.Allow { + t.Error("token policy without a secret must fail closed") + } +} + +func TestAuthorize_PerStreamPolicyOverride(t *testing.T) { + t.Parallel() + // Global default = token, but ch1 is explicitly public → allowed w/o token. + pol := func(c domain.StreamCode) string { + if c == "public1" { + return PolicyPublic + } + if c == "token1" { + return PolicyToken + } + return "" + } + a := newAuthz(t, config.MediaAuthConfig{Enabled: true, DefaultPolicy: PolicyToken, TokenSecret: "sk"}, nil, pol) + + if d := a.Authorize(AuthRequest{Code: "public1"}); !d.Allow { + t.Errorf("per-stream public override must allow without token: %s", d.Reason) + } + if d := a.Authorize(AuthRequest{Code: "inherits"}); d.Allow { + t.Error("stream inheriting global token policy must require a token") + } + + // Global default = public, but token1 explicitly requires a token. + b := newAuthz(t, config.MediaAuthConfig{Enabled: true, DefaultPolicy: PolicyPublic, TokenSecret: "sk"}, nil, pol) + if d := b.Authorize(AuthRequest{Code: "token1"}); d.Allow { + t.Error("per-stream token override must require a token even when global is public") + } + if d := b.Authorize(AuthRequest{Code: "other"}); !d.Allow { + t.Errorf("public-default stream must allow: %s", d.Reason) + } +} + +func TestAuthorize_DenyBeatsAllow(t *testing.T) { + t.Parallel() + // IP is on both lists — deny must win. + a := newAuthz(t, config.MediaAuthConfig{ + Enabled: true, + AllowIPs: []string{"1.2.3.4"}, + DenyIPs: []string{"1.2.3.4"}, + }, nil, nil) + if d := a.Authorize(AuthRequest{Code: testCode, ClientIP: ip("1.2.3.4")}); d.Allow { + t.Error("deny must take precedence over allow") + } +} + +func TestAuthorize_HotReload(t *testing.T) { + t.Parallel() + a := newAuthz(t, config.MediaAuthConfig{Enabled: false}, nil, nil) + if d := a.Authorize(AuthRequest{Code: testCode, ClientIP: ip("8.8.8.8")}); !d.Allow { + t.Fatal("disabled should allow") + } + a.SetConfig(config.MediaAuthConfig{Enabled: true, DenyIPs: []string{"8.8.8.8"}}) + if d := a.Authorize(AuthRequest{Code: testCode, ClientIP: ip("8.8.8.8")}); d.Allow { + t.Error("after reload, denied IP must be blocked") + } +} diff --git a/internal/publisher/playauth.go b/internal/publisher/playauth.go new file mode 100644 index 0000000..75716e0 --- /dev/null +++ b/internal/publisher/playauth.go @@ -0,0 +1,64 @@ +package publisher + +// playauth.go — media-plane (playback) authorization glue for the publisher's +// non-HTTP protocols (RTMP / SRT / RTSP). The HTTP protocols (HLS / DASH / +// MPEGTS) are gated in the api dispatcher. See internal/mediaauth (B / S-13). + +import ( + "log/slog" + "net" + + "github.com/ntt0601zcoder/open-streamer/internal/domain" + "github.com/ntt0601zcoder/open-streamer/internal/mediaauth" +) + +// SetMediaAuthorizer wires the playback authorizer. nil disables media auth +// (every playback request is allowed) — the default. +func (s *Service) SetMediaAuthorizer(a *mediaauth.Authorizer) { s.mediaAuth = a } + +// PlaybackPolicy returns a RUNNING stream's per-stream media-auth policy +// override ("public" / "token" / "" = inherit global) and whether the stream is +// currently running. O(1) in-memory lookup, used as the first tier of the +// authorizer's PolicyResolver so the live hot path never reads the store. The +// `running` flag lets the caller fall back to the store for a STOPPED stream +// (whose DVR archive is still served) instead of mistaking its empty in-memory +// policy for "inherit global". +func (s *Service) PlaybackPolicy(code domain.StreamCode) (policy string, running bool) { + s.mu.Lock() + defer s.mu.Unlock() + if ss, ok := s.streams[code]; ok { + return ss.playbackAuth, true + } + return "", false +} + +// playAllowed runs the media-auth chain for one playback request. Returns true +// (allow) when no authorizer is wired or media auth is disabled. Denials are +// logged with the (non-leaky) reason. +func (s *Service) playAllowed(code domain.StreamCode, proto, addr, token, ua, referer string) bool { + if s.mediaAuth == nil { + return true + } + d := s.mediaAuth.Authorize(mediaauth.AuthRequest{ + Code: code, + ClientIP: ipFromAddr(addr), + Token: token, + UserAgent: ua, + Referer: referer, + }) + if !d.Allow { + slog.Info("publisher: playback denied", "stream_code", code, "proto", proto, "reason", d.Reason, "remote", addr) + } + return d.Allow +} + +// ipFromAddr extracts the IP from a "host:port" (or bare host) address string. +func ipFromAddr(addr string) net.IP { + if addr == "" { + return nil + } + if host, _, err := net.SplitHostPort(addr); err == nil { + return net.ParseIP(host) + } + return net.ParseIP(addr) +} diff --git a/internal/publisher/serve_rtmp.go b/internal/publisher/serve_rtmp.go index 7860330..bbc587f 100644 --- a/internal/publisher/serve_rtmp.go +++ b/internal/publisher/serve_rtmp.go @@ -54,6 +54,13 @@ func (s *Service) HandleRTMPPlay( return fmt.Errorf("stream %q not active", key) } + // Media-plane authorization (token / IP / country) before any state. RTMP + // play carries no query/UA in the handshake, so token/UA rules don't apply + // here — IP/country/policy still do (B / S-13). + if !s.playAllowed(code, "rtmp", info.RemoteAddr, "", "", "") { + return fmt.Errorf("stream %q: playback not authorized", key) + } + // Cap concurrent playback connections before allocating the per-session // demux pipeline + tsBuffer (A-1). Released on every return path via defer. if !s.limiter.acquire(code) { diff --git a/internal/publisher/serve_rtsp.go b/internal/publisher/serve_rtsp.go index e9d6e7c..869e569 100644 --- a/internal/publisher/serve_rtsp.go +++ b/internal/publisher/serve_rtsp.go @@ -58,6 +58,7 @@ import ( "github.com/ntt0601zcoder/open-streamer/internal/buffer" "github.com/ntt0601zcoder/open-streamer/internal/domain" + "github.com/ntt0601zcoder/open-streamer/internal/sessions" "github.com/ntt0601zcoder/open-streamer/internal/tsdemux" "github.com/ntt0601zcoder/open-streamer/internal/tsmux" ) @@ -236,6 +237,23 @@ func (h *rtspHandler) OnPlay(ctx *gortsplib.ServerHandlerOnPlayCtx) (*base.Respo return &base.Response{StatusCode: base.StatusOK}, nil } + ua := "" + if ctx.Request != nil { + if v, ok := ctx.Request.Header["User-Agent"]; ok && len(v) > 0 { + ua = v[0] + } + } + + // Media-plane authorization (token / IP / country / UA) before allocating. + // RTSP carries the token in the URL query (rtsp://h/live/code?token=…). + remote := "" + if c := ctx.Conn.NetConn().RemoteAddr(); c != nil { + remote = c.String() + } + if !h.svc.playAllowed(code, "rtsp", remote, sessions.TokenFromQuery(ctx.Query), ua, "") { + return &base.Response{StatusCode: base.StatusUnauthorized}, nil + } + // Cap concurrent playback connections before tracking the session (A-1). // The rtspSessions entry (created below on success) is the release marker // OnSessionClose uses, so it is stored even when the tracker is absent. @@ -244,12 +262,6 @@ func (h *rtspHandler) OnPlay(ctx *gortsplib.ServerHandlerOnPlayCtx) (*base.Respo return &base.Response{StatusCode: base.StatusServiceUnavailable}, nil } - ua := "" - if ctx.Request != nil { - if v, ok := ctx.Request.Header["User-Agent"]; ok && len(v) > 0 { - ua = v[0] - } - } sess := openRTSPSession(context.Background(), h.svc.tracker, code, ctx.Conn.NetConn().RemoteAddr(), ua) h.svc.rtspSessionsMu.Lock() h.svc.rtspSessions[ctx.Session] = &rtspClient{ps: sess, code: code} diff --git a/internal/publisher/serve_srt.go b/internal/publisher/serve_srt.go index efa1ed6..fd89e69 100644 --- a/internal/publisher/serve_srt.go +++ b/internal/publisher/serve_srt.go @@ -31,6 +31,7 @@ import ( srt "github.com/datarhei/gosrt" "github.com/ntt0601zcoder/open-streamer/internal/domain" + "github.com/ntt0601zcoder/open-streamer/internal/sessions" "github.com/ntt0601zcoder/open-streamer/internal/tsmux" ) @@ -117,6 +118,18 @@ func (s *Service) srtHandleSubscribe(ctx context.Context, conn srt.Conn) { return } + // Media-plane authorization (token / IP / country) before any state. SRT + // carries the token in the streamid query (?streamid=live/code?token=…). + srtTok := "" + if sid := conn.StreamId(); sid != "" { + if i := strings.IndexByte(sid, '?'); i >= 0 { + srtTok = sessions.TokenFromQuery(sid[i+1:]) + } + } + if !s.playAllowed(streamCode, "srt", conn.RemoteAddr().String(), srtTok, "", "") { + return + } + // Cap concurrent playback connections before allocating the subscriber + // muxer state (A-1). The conn is closed by the deferred conn.Close above. if !s.limiter.acquire(streamCode) { @@ -239,6 +252,11 @@ func (s *Service) srtHandleSubscribe(ctx context.Context, conn srt.Conn) { // by accident from a half-typed URL. func srtStreamCode(streamid string) string { streamid = strings.TrimSpace(streamid) + // Drop a playback-token query (streamid=live/code?token=…) before deriving + // the code, so the token doesn't bleed into the stream-code lookup. + if i := strings.IndexByte(streamid, '?'); i >= 0 { + streamid = streamid[:i] + } hadLivePrefix := strings.HasPrefix(streamid, "live/") code := strings.TrimPrefix(streamid, "live/") if code == "" { diff --git a/internal/publisher/service.go b/internal/publisher/service.go index 364988e..9f9687f 100644 --- a/internal/publisher/service.go +++ b/internal/publisher/service.go @@ -25,6 +25,7 @@ import ( "github.com/ntt0601zcoder/open-streamer/internal/buffer" "github.com/ntt0601zcoder/open-streamer/internal/domain" "github.com/ntt0601zcoder/open-streamer/internal/events" + "github.com/ntt0601zcoder/open-streamer/internal/mediaauth" "github.com/ntt0601zcoder/open-streamer/internal/metrics" "github.com/ntt0601zcoder/open-streamer/internal/sessions" ) @@ -80,6 +81,11 @@ type streamState struct { // re-querying the store. Updated atomically with mediaBuf when the stream // config changes (Update path). mpegtsEnabled bool + + // playbackAuth is the stream's media-auth policy override ("public"/"token"/ + // "" inherit). Mirrored here so PlaybackPolicy answers the authorizer with + // an O(1) in-memory lookup instead of a store read per request. + playbackAuth string } // Service manages all output workers for active streams. @@ -100,6 +106,11 @@ type Service struct { // the long-lived play/push protocols — see conn_limiter.go (A-1). limiter *connLimiter + // mediaAuth authorizes playback (token / IP / country / UA / referer). + // nil = no media-auth (allow all); set via SetMediaAuthorizer. See + // internal/mediaauth (B / S-13). + mediaAuth *mediaauth.Authorizer + mu sync.Mutex streams map[domain.StreamCode]*streamState @@ -273,6 +284,7 @@ func (s *Service) Start(ctx context.Context, stream *domain.Stream) error { mediaBuf: buffer.PlaybackBufferID(stream.Code, stream.Transcoder), protocols: make(map[string]context.CancelFunc), mpegtsEnabled: p.MPEGTS, + playbackAuth: stream.PlaybackAuth, } s.streams[stream.Code] = ss s.mediaBuffer[stream.Code] = ss.mediaBuf @@ -440,6 +452,7 @@ func (s *Service) UpdateProtocols(ctx context.Context, old, new *domain.Stream) s.mediaBuffer[new.Code] = newBuf } ss.mpegtsEnabled = np.MPEGTS + ss.playbackAuth = new.PlaybackAuth } s.mu.Unlock() if !ok { diff --git a/internal/runtime/manager.go b/internal/runtime/manager.go index 11d2fd2..3242db9 100644 --- a/internal/runtime/manager.go +++ b/internal/runtime/manager.go @@ -20,6 +20,7 @@ import ( "github.com/ntt0601zcoder/open-streamer/internal/hooks" "github.com/ntt0601zcoder/open-streamer/internal/ingestor" "github.com/ntt0601zcoder/open-streamer/internal/manager" + "github.com/ntt0601zcoder/open-streamer/internal/mediaauth" "github.com/ntt0601zcoder/open-streamer/internal/publisher" "github.com/ntt0601zcoder/open-streamer/internal/sessions" "github.com/ntt0601zcoder/open-streamer/internal/store" @@ -45,6 +46,7 @@ type Deps struct { SessionsSvc *sessions.Service AutoPublish *autopublish.Service APISrv *api.Server + MediaAuth *mediaauth.Authorizer Bus events.Bus StreamRepo store.StreamRepository GlobalConfigRepo store.GlobalConfigRepository @@ -242,6 +244,16 @@ func (m *Manager) diff(old, new *domain.GlobalConfig) { m.deps.APISrv.SetAuthConfig(new.Auth) } + // Media-plane auth: hot-swap playback rules (token/IP/country/UA/referer) + // on the shared authorizer used by both the publisher and the API server. + if m.deps.MediaAuth != nil && configChanged(old.Auth, new.Auth) { + media := config.MediaAuthConfig{} + if new.Auth != nil { + media = new.Auth.Media + } + m.deps.MediaAuth.SetConfig(media) + } + // Push the new listeners snapshot to ingestor + publisher BEFORE // diffService can decide to restart them — each Run() reads the cached // listeners value at startup, so the swap must happen before the new