diff --git a/api/docs/docs.go b/api/docs/docs.go index 927f37da..e327a965 100644 --- a/api/docs/docs.go +++ b/api/docs/docs.go @@ -2646,6 +2646,8 @@ const docTemplate = `{ "domain.EventType": { "type": "string", "enum": [ + "session.opened", + "session.closed", "stream.created", "stream.updated", "stream.started", @@ -2682,9 +2684,7 @@ const docTemplate = `{ "policy.updated", "policy.deleted", "stream.runtime_created", - "stream.runtime_expired", - "session.opened", - "session.closed" + "stream.runtime_expired" ], "x-enum-comments": { "EventDVRSegmentPruned": "retention loop deleted an aged-out segment", @@ -2701,6 +2701,8 @@ const docTemplate = `{ "EventStreamUpdated": "PUT /streams/{code} on existing record" }, "x-enum-descriptions": [ + "", + "", "", "PUT /streams/{code} on existing record", "", @@ -2737,11 +2739,11 @@ const docTemplate = `{ "", "", "", - "", - "", "" ], "x-enum-varnames": [ + "EventSessionOpened", + "EventSessionClosed", "EventStreamCreated", "EventStreamUpdated", "EventStreamStarted", @@ -2778,9 +2780,7 @@ const docTemplate = `{ "EventPolicyUpdated", "EventPolicyDeleted", "EventStreamRuntimeCreated", - "EventStreamRuntimeExpired", - "EventSessionOpened", - "EventSessionClosed" + "EventStreamRuntimeExpired" ] }, "domain.GlobalConfig": { diff --git a/api/docs/swagger.json b/api/docs/swagger.json index d3d56d9e..93900d55 100644 --- a/api/docs/swagger.json +++ b/api/docs/swagger.json @@ -2639,6 +2639,8 @@ "domain.EventType": { "type": "string", "enum": [ + "session.opened", + "session.closed", "stream.created", "stream.updated", "stream.started", @@ -2675,9 +2677,7 @@ "policy.updated", "policy.deleted", "stream.runtime_created", - "stream.runtime_expired", - "session.opened", - "session.closed" + "stream.runtime_expired" ], "x-enum-comments": { "EventDVRSegmentPruned": "retention loop deleted an aged-out segment", @@ -2694,6 +2694,8 @@ "EventStreamUpdated": "PUT /streams/{code} on existing record" }, "x-enum-descriptions": [ + "", + "", "", "PUT /streams/{code} on existing record", "", @@ -2730,11 +2732,11 @@ "", "", "", - "", - "", "" ], "x-enum-varnames": [ + "EventSessionOpened", + "EventSessionClosed", "EventStreamCreated", "EventStreamUpdated", "EventStreamStarted", @@ -2771,9 +2773,7 @@ "EventPolicyUpdated", "EventPolicyDeleted", "EventStreamRuntimeCreated", - "EventStreamRuntimeExpired", - "EventSessionOpened", - "EventSessionClosed" + "EventStreamRuntimeExpired" ] }, "domain.GlobalConfig": { diff --git a/api/docs/swagger.yaml b/api/docs/swagger.yaml index 278bf871..740549a0 100644 --- a/api/docs/swagger.yaml +++ b/api/docs/swagger.yaml @@ -678,6 +678,8 @@ definitions: type: object domain.EventType: enum: + - session.opened + - session.closed - stream.created - stream.updated - stream.started @@ -715,8 +717,6 @@ definitions: - policy.deleted - stream.runtime_created - stream.runtime_expired - - session.opened - - session.closed type: string x-enum-comments: EventDVRSegmentPruned: retention loop deleted an aged-out segment @@ -734,6 +734,8 @@ definitions: EventStreamUpdated: PUT /streams/{code} on existing record x-enum-descriptions: - "" + - "" + - "" - PUT /streams/{code} on existing record - "" - "" @@ -770,9 +772,9 @@ definitions: - "" - "" - "" - - "" - - "" x-enum-varnames: + - EventSessionOpened + - EventSessionClosed - EventStreamCreated - EventStreamUpdated - EventStreamStarted @@ -810,8 +812,6 @@ definitions: - EventPolicyDeleted - EventStreamRuntimeCreated - EventStreamRuntimeExpired - - EventSessionOpened - - EventSessionClosed domain.GlobalConfig: properties: auth: diff --git a/cmd/server/main.go b/cmd/server/main.go index 2a103673..26181a18 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -165,6 +165,10 @@ func run() error { // avoid the ingestor package depending on the store layer. wireCopyLookup(injector) + // 7c. Wire the push-ingest auth resolver (S-8) so the RTMP server enforces + // each stream's configured StreamKey before accepting a publisher. + wirePushAuth(injector) + // 8. Hydrate VOD registry from the store so ingestor workers started in // bootstrap can resolve file:// URLs against the user's mount table. if err := hydrateVODRegistry(ctx, injector); err != nil { @@ -415,6 +419,34 @@ func wireCopyLookup(i do.Injector) { }) } +// wirePushAuth hands the ingestor's RTMP push server a resolver for a stream's +// configured push secret (StreamKey), enforced before a publisher is accepted +// (S-8). Live streams hit the publisher's in-memory table (O(1), and covers +// auto-publish runtime streams with no store record); a stopped/configured +// stream falls back to the store + template merge so an inherited StreamKey is +// honoured. "" = the stream opted out of push auth. +func wirePushAuth(i do.Injector) { + ing := do.MustInvoke[*ingestor.Service](i) + pub := do.MustInvoke[*publisher.Service](i) + streamRepo := do.MustInvoke[store.StreamRepository](i) + templateRepo := do.MustInvoke[store.TemplateRepository](i) + ing.SetPushStreamKeyResolver(func(code domain.StreamCode) string { + if sk, running := pub.PushStreamKey(code); running { + return sk + } + 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.StreamKey + }) +} + // hydrateVODRegistry reads the persisted VOD mount table and seeds the // in-memory registry. Without this, file:// URLs would fail to resolve until // the operator first edited the mount table at runtime. diff --git a/docs/audit/security-quality-audit.md b/docs/audit/security-quality-audit.md index 52cf5b72..88e3ccd4 100644 --- a/docs/audit/security-quality-audit.md +++ b/docs/audit/security-quality-audit.md @@ -50,7 +50,7 @@ --- #### S-2 (CRITICAL) — Unauthenticated arbitrary host-file create/append (and blind SSRF) via hooks -> ✅ **FIXED** in `fix/security-critical-high` — REST `Create`/`Update` now run `domain.Hook.Validate(fileRoot)` (shared with the YAML path, replacing `validateHookTypeTarget`), closing the zero-validation gap. File hooks are confined to `hooks.file_root_dir` (`pathInside` after `Clean`, rejecting `..`) — enforced at the REST/YAML boundary AND re-checked in `deliverFile` at delivery time. The hooks `http.Client` now uses `internal/netguard` so loopback + link-local/cloud-metadata (169.254.169.254) are blocked at dial time (a hook can't pivot to the local admin API or IMDS); internal RFC1918 webhooks stay allowed. Tests: `TestHookValidate`, hook handler tests. **Residual (noted):** file containment is opt-in (`file_root_dir` empty = legacy absolute-path-only); a symlink planted *inside* the configured root is still followed by the `O_APPEND|O_CREATE` write — keep the root symlink-free. **Hot-reload added** (`fix/security-critical-high`): `hooks.file_root_dir` is now read live — `hooks.Service` holds it in an `atomic.Pointer` (`SetConfig`/`FileRootDir()`), the REST handler reads `h.hooks.FileRootDir()` per request instead of caching at boot, and `runtime.Manager.diff` calls `HooksSvc.SetConfig` on a hooks-config change, so setting the root takes effect without a restart. +> ✅ **FIXED** in `fix/security-critical-high` — REST `Create`/`Update` now run `domain.Hook.Validate(fileRoot)` (shared with the YAML path, replacing `validateHookTypeTarget`), closing the zero-validation gap. File hooks are confined to `hooks.file_root_dir` (`pathInside` after `Clean`, rejecting `..`) — enforced at the REST/YAML boundary AND re-checked in `deliverFile` at delivery time. Tests: `TestHookValidate`, hook handler tests. (The HTTP-hook half — an egress dial guard on the webhook client — was added then **removed** by operator decision; see S-6.) **Residual (noted):** file containment is opt-in (`file_root_dir` empty = legacy absolute-path-only); a symlink planted *inside* the configured root is still followed by the `O_APPEND|O_CREATE` write — keep the root symlink-free. **Hot-reload added** (`fix/security-critical-high`): `hooks.file_root_dir` is now read live — `hooks.Service` holds it in an `atomic.Pointer` (`SetConfig`/`FileRootDir()`), the REST handler reads `h.hooks.FileRootDir()` per request instead of caching at boot, and `runtime.Manager.diff` calls `HooksSvc.SetConfig` on a hooks-config change, so setting the root takes effect without a restart. **Files:** `internal/api/handler/hook.go:84` (Create) / `:128` (Update); file sink `internal/hooks/service.go:337-365` (`deliverFile`, only guard `filepath.IsAbs` at `:342`, `O_APPEND|O_CREATE|O_WRONLY 0o644` at `:356`); HTTP sink `internal/hooks/batcher.go:414-446` (`postOnce`); store `internal/store/json/store.go:328-333`. **Merges:** "file-type hook arbitrary write", "hook create/update skip validation". @@ -84,7 +84,7 @@ --- #### S-4 (HIGH) — SSRF via unvalidated ingest/pull input URLs (content- and redirect-chosen targets) -> 🚫 **WON'T FIX / RISK ACCEPTED (operator decision)** in `fix/remove-ingest-ssrf-guard` — the ingest SSRF guard was **removed**. Rationale: this server's primary role is pulling from **internal / LAN sources** (on-prem origins, internal CDNs, LAN cameras, multicast), so blocking private/RFC1918 targets is counterproductive and operationally fragile — the default-off guard caused a real production failover freeze (an internal HLS input was blocked at dial, the worker reconnect-looped, no error surfaced). The guard also only ever covered the HTTP-family pulls (HLS / HTTP-TS); RTSP/RTMP/SRT/UDP were never guarded, so it never provided complete coverage. The decisive compensating control is **S-1 (control-plane Basic auth)**: the admin API that made this SSRF remotely reachable is now authenticated, removing the unauthenticated-remote exploitation angle that made it High. Removed: `netguard.ApplyToTransport` + `CheckRedirect` on the HLS/HTTP-TS clients, the save-time `netguard.ValidateInputURL` speed-bump in `decodeStreamBody`, the `ingestor.allow_private_targets` config flag, and the `netguard:` fail-fast clause in the ingest worker. **Kept:** the `internal/netguard` package itself and its use by the **webhook HTTP client** (S-2 — a hook must never reach the local admin API / IMDS); that loopback+metadata guard stays. `IngestPolicy` / `ValidateInputURL` remain as unused helpers. +> 🚫 **WON'T FIX / RISK ACCEPTED (operator decision)** in `fix/remove-ingest-ssrf-guard` — the ingest SSRF guard was **removed**. Rationale: this server's primary role is pulling from **internal / LAN sources** (on-prem origins, internal CDNs, LAN cameras, multicast), so blocking private/RFC1918 targets is counterproductive and operationally fragile — the default-off guard caused a real production failover freeze (an internal HLS input was blocked at dial, the worker reconnect-looped, no error surfaced). The guard also only ever covered the HTTP-family pulls (HLS / HTTP-TS); RTSP/RTMP/SRT/UDP were never guarded, so it never provided complete coverage. The decisive compensating control is **S-1 (control-plane Basic auth)**: the admin API that made this SSRF remotely reachable is now authenticated, removing the unauthenticated-remote exploitation angle that made it High. Removed: `netguard.ApplyToTransport` + `CheckRedirect` on the HLS/HTTP-TS clients, the save-time `netguard.ValidateInputURL` speed-bump in `decodeStreamBody`, the `ingestor.allow_private_targets` config flag, and the `netguard:` fail-fast clause in the ingest worker. The `internal/netguard` package was initially kept for the webhook client, but that guard was later removed too (operator decision — webhooks legitimately target private/local endpoints; see S-6) and the **package was deleted**. No egress IP guard remains in the codebase. > > 🔕 **Follow-up bug IGNORED (won't fix)** — a later investigation found the ingest guard (a) only covered HLS/HTTP-TS, leaving RTSP/RTMP/SRT internal inputs unguarded, and (b) wasn't retroactive to already-running readers when `allow_private_targets` was toggled. Both are moot now that the guard is removed; marked ignore per operator decision. @@ -118,11 +118,15 @@ --- #### S-6 (MEDIUM) — Blind SSRF via HTTP hook target -Covered as the HTTP-sink half of **S-2**. `batcher.go:414-446` POSTs to any operator-supplied URL with no host allowlist; downgraded from High by verification because delivery is blind (`DeliverTestEvent` returns nil immediately, `service.go:128-132`) and POST-only (weakens IMDS GET credential-theft). Fix: the `DialContext` guard described in S-2. +> 🚫 **WON'T FIX / RISK ACCEPTED (operator decision)** — the webhook egress dial guard (`internal/netguard`, added in the S-2 batch) was **removed** and the package deleted; the hooks client is a plain `&http.Client{}` again. Rationale: (1) delivery is **POST + blind** (`DeliverTestEvent` returns nil immediately) with a **body the caller does not control** (the event JSON) — no response/oracle, and IMDS credential theft needs GET/PUT so POST can't reach it; (2) configuring a hook URL now requires admin auth (**S-1**), so this is no longer an unauthenticated primitive — pointing a webhook at an arbitrary URL is the intended feature; (3) operators legitimately need to push webhooks to **private and local** endpoints (sidecar log collectors, internal services on loopback), which the guard blocked. Residual is negligible: a blind POST with a fixed-shape body to an internal host, gated behind admin auth. + +`batcher.go:414-446` POSTs to any operator-supplied URL with no host allowlist; downgraded from High at audit time because delivery is blind and POST-only. --- #### S-7 (MEDIUM) — Secrets in logs: SRT passphrase and credentialed pull URLs logged in cleartext +> 🚫 **ACCEPTED (operator decision)** — single-tenant host with restricted log / journald access; only URL-embedded secrets are affected (passphrases supplied via `input.params` are not logged). The URL in logs is operationally useful for diagnosing ingest; the local-disclosure risk is accepted. Revisit if logs are shipped to a broader audience. + **Files:** `internal/ingestor/pull/srt.go:76-81` (`slog.Info(... "url", r.input.URL)` on every connect) + `:66` (URL in error); `internal/ingestor/worker.go:103-107` (connect) / `:254-260` (`handleOpenFailure`, repeated during backoff); `hls.go:210-211/308/382/387`; `httpts.go:78/87/93` (these are `fmt.Errorf` wraps, not slog — minor locator correction, but the wrapped URL still reaches the log via `handleOpenFailure`'s `"err"` field); also `rtsp.go` at 12 sites (`:120,135,150,204,231,309,333,405,420,449`) and `rtmp.go:164`. **Trigger:** Any SRT input `srt://host?...&passphrase=secret` or credentialed pull URL (`http://user:pass@host`, `?token=...`, RTSP `user:pass@`) is logged at Info on connect/reconnect/open-failure. A passphrase supplied via `input.Params` (not the URL) is *not* logged — only URL-embedded secrets leak. @@ -136,6 +140,8 @@ Covered as the HTTP-sink half of **S-2**. `batcher.go:414-446` POSTs to any oper --- #### S-8 (MEDIUM) — Push ingest has no authentication; auto-publish lets any client materialise streams +> ✅ **FIXED** in `fix/push-ingest-auth` — push ingest now enforces the per-stream `StreamKey` that was previously defined-but-unwired. An RTMP publisher supplies the secret as `?key=` on the publish URL (`rtmp://host/?key=…`); lal strips the query from the stream name so routing (`rtmpRouteKey`) is unaffected. The RTMP server resolves the stream's configured `StreamKey` via a resolver (publisher in-memory table for live + auto-publish runtime streams → store+template fallback for configured streams, mirroring the media-auth `PolicyResolver`) and constant-time-compares it before claiming the registry slot, so a mismatch is rejected before any slot/pipeline state changes. An empty `StreamKey` opts the stream out (unauthenticated, backward-compatible). Auto-publish enforces the matching template's `StreamKey` inside `ResolveOrCreate` **before** materialising the runtime stream, so an unauthorised push never spins a pipeline. Tests: `TestPushKeyOK`, `TestPushSecretFromQuery`, `TestResolveOrCreate_StreamKeyEnforced`. **Residual (noted):** the concurrent-runtime-stream cap + per-source-IP rate-limit from the original recommendation are not implemented (DoS hardening, orthogonal to the auth gap); SRT push has no server in the codebase, so there is nothing to gate there. + **Files:** `internal/ingestor/push/rtmp_server.go:350-402` (`OnNewRtmpPubSession`, target from `rtmpRouteKey` only — no secret read), `:232-257` (`acquireOrAutoPublish`); `internal/ingestor/registry.go:76-89` (`Acquire` checks only registered + not-active); `internal/ingestor/service.go:607-617` (`pushStreamKey` = stream code); `internal/autopublish/service.go:143-204` (`ResolveOrCreate` — prefix match + code-shape validation only). **Trigger:** Any client that can TCP-connect to the RTMP push port and knows/guesses a registered stream code can publish into an **idle** `publish://` slot (the full path is the only credential). With an `AutoPublishResolver` wired, a push to any path matching a template prefix synthesises a runtime stream and starts a pipeline. @@ -147,6 +153,8 @@ Covered as the HTTP-sink half of **S-2**. `batcher.go:414-446` POSTs to any oper --- #### S-9 (MEDIUM) — Hook HMAC secrets disclosed via GET /hooks and GET /config/yaml +> 🚫 **ACCEPTED (operator decision)** — the control plane is now behind Basic-auth (S-1), so only authenticated admins read hook config, and an admin managing hooks legitimately needs the secret (it is the round-trip source for `PUT /config/yaml`). No unauthenticated exposure remains. Risk accepted. + **Files:** `internal/domain/hook.go:55` (`Secret` with `json:"secret" yaml:"secret"`, no redaction); `internal/api/handler/hook.go:65-72` (List), `:106-114` (Get), `:95/:146` (Create/Update echo); `internal/api/handler/config_yaml.go:60-101` (`GetConfigYAML` marshals `Hooks`), `:207/227` (PUT success returns `hooksAfter`). Secret is the HMAC-SHA256 key (`internal/hooks/batcher.go:413-427`, sets `X-OpenStreamer-Signature`). **Trigger:** `GET /config/yaml` or `GET /hooks` returns every hook's signing secret verbatim. Reachable unauthenticated (S-1); transits plaintext (no TLS). @@ -160,6 +168,8 @@ Covered as the HTTP-sink half of **S-2**. `batcher.go:414-446` POSTs to any oper --- #### S-10 (MEDIUM) — Push destination URLs (stream keys/credentials) exposed via logs, /metrics, and RuntimeStatus +> 🚫 **ACCEPTED (operator decision)** — same trust model as S-7: single-tenant host, admin API + `/metrics` access controlled. Redaction is hardening, not a remote vulnerability. Risk accepted. + **Files:** logs `internal/publisher/push_rtmp.go:166,174,217,254,385,407,415,425,433,438,462` (Info/Warn `"url"` on every lifecycle event, fires on every retry); metrics `internal/metrics/metrics.go:290/371/409` (`dest_url` label) ← `internal/publisher/service.go:678/688`, `internal/publisher/runtime.go:112/126`; API `internal/publisher/runtime.go:214-244` (`PushSnapshot.URL` unredacted) → `internal/api/handler/stream.go:132-133`; event payload `runtime.go:154`. **Merges:** "push URL logged + RuntimeStatus" (low) and "push URL on /metrics" (medium, two reports). @@ -172,6 +182,8 @@ Covered as the HTTP-sink half of **S-2**. `batcher.go:414-446` POSTs to any oper --- #### S-11 (MEDIUM) — Secrets persisted in world-readable store files +> 🚫 **ACCEPTED (operator decision)** — strictly local, single-tenant host, mitigated by the deployment umask; a CWE-276 hardening defect with no remote angle. Risk accepted (file mode could be tightened to 0600 cheaply later if desired). + **Files:** `internal/store/json/store.go:52` (`MkdirAll 0o755`), `:118` (`WriteFile 0o644`) + `:121` rename; identical `internal/store/yaml/store.go:53/119`. No `os.Chmod`/`0o600` anywhere in `internal/store/`. Secrets serialized: `Hook.Secret`, `Stream.StreamKey`, `Input.Headers`/`Input.Params` (Authorization headers / SRT passphrases / S3 keys). **Trigger:** Any local user/process that can traverse to the data dir reads `/open_streamer.json` (mode 0644, umask-masked). Default driver `json`, default dir wired in `cmd/server/main.go:153/165`. @@ -183,6 +195,8 @@ Covered as the HTTP-sink half of **S-2**. `batcher.go:414-446` POSTs to any oper --- #### S-12 (LOW) — HLS timeshift master-playlist injection via unvalidated `from/dur/delay/ago` +> 🚫 **ACCEPTED (operator decision)** — LOW; impact is bounded to the requesting client's own generated playlist (no cross-user / server effect) — a crafted-param nuisance, not a privilege or data boundary. Risk accepted. + **Files:** `internal/api/handler/blob_timeshift.go:106-108` (master branch renders **before** the numeric validation at `:110-119`), `:238-246` (`timeshiftParams` rebuilds `k+"="+v` raw); `internal/dvr/blob/hls.go:20-21/43` (`Fprintf` into the manifest). **Merges:** two reported timeshift-injection findings. @@ -208,6 +222,8 @@ Covered as the HTTP-sink half of **S-2**. `batcher.go:414-446` POSTs to any oper --- #### S-14 (LOW) — Subprocess gRPC unix socket is unauthenticated +> 🚫 **ACCEPTED (operator decision)** — the gRPC unix socket is a filesystem object scoped to the local host / service user (a per-stream child process); local-only, no network exposure. Risk accepted. + **Files:** `cmd/open-streamer-transcoder/main.go:64-94` (`lc.Listen("unix", ...)` no chmod, plain `grpc.NewServer()` no peer check); `internal/transcoder/supervisor.go:534-539` (`pickSocketPath` under `os.TempDir()`). **Verdict:** partially-confirmed. @@ -220,6 +236,8 @@ Covered as the HTTP-sink half of **S-2**. `batcher.go:414-446` POSTs to any oper --- #### S-15 (LOW) — Client IP / session fingerprint spoofable via X-Forwarded-For / X-Real-IP +> 🚫 **ACCEPTED (operator decision)** — IP / country playback rules are only relied upon behind a header-overwriting trusted proxy (documented residual); they are not a security boundary in this deployment. Risk accepted. + **Files:** `internal/sessions/tracker.go:160-177` (`clientIP` reads XFF/X-Real-IP directly) → `fingerprintID` `:295`, GeoIP `:346`, event payload `:622/646`; `internal/api/server.go:163` (`middleware.RealIP` mutates `r.RemoteAddr`). **Trigger:** Send `X-Forwarded-For: ` on any HLS/DASH GET. The IP feeds the session fingerprint, GeoIP country, and attribution. @@ -231,6 +249,8 @@ Covered as the HTTP-sink half of **S-2**. `batcher.go:414-446` POSTs to any oper --- #### S-16 (LOW) — pprof listener: unauthenticated heap/goroutine dumps + always-on block/mutex profiling +> 🚫 **ACCEPTED (operator decision)** — operational profiling endpoint; bound to localhost / an ops-only network at deploy time. Risk accepted as an ops tool. + **Files:** `internal/api/server.go:393-436` (`startPprofListener`), `:414-415` (`SetBlockProfileRate(10ms)`/`SetMutexProfileFraction(100)` unconditional, never reset), `:398-407` (mux, no auth), `:418` (addr used verbatim, no loopback enforcement). `config/config.go:39` — no default, opt-in. **Trigger:** Operator sets `pprof_addr`. Serves `/debug/pprof/*` (heap/goroutine/cmdline/profile/trace) unauthenticated and forces process-wide contention profiling on. A config reload clearing `pprof_addr` stops the listener but leaves the profilers enabled (slightly worse than reported). @@ -436,7 +456,7 @@ Covered as the HTTP-sink half of **S-2**. `batcher.go:414-446` POSTs to any oper #### A-5 (HIGH) — `manager.Register` swallows initial `ingestor.Start` failure → permanent zombie stream reported Active > ✅ **FIXED** in `fix/lifecycle-self-heal` — `Register`'s synchronous start-failure branch now routes through `ReportInputError`, so the input degrades, error history is recorded, and failover / exhausted-handling engage (multi-input promotes a backup; single-input flips to Degraded + probe loop). `Register` also refuses a duplicate registration (no monitor-goroutine leak). Tests `TestRegister_StartFailureDrivesExhausted`, `TestRegister_RefusesDuplicate`. (The connects-but-no-packets Idle blind spot is a noted follow-up.) > -> ✅ **FOLLOW-UP FIXED** in `fix/security-critical-high` — the connects-but-no-packets / never-first-packet blind spot is closed. `collectTimeoutIfNeeded` no longer gates on `StatusActive`: it now fires for the **active** input whenever `now − LastPacketAt > timeout` regardless of status (Idle/Degraded excluded only while a failover is already in flight). `Register` stamps `LastPacketAt = now` on each input so the timeout window measures connect-to-first-packet time as a grace period. A switched-to input (manual `/switch` or auto-failover) that connects but never delivers a packet now times out and fails over to the next priority instead of freezing forever. Complement: the ingest worker (`shouldFailoverImmediately`) now classifies `netguard:`-blocked dials as fail-immediately, so an SSRF-rejected input reports the error and triggers failover instantly rather than reconnect-looping silently. Tests `TestCollectTimeoutIfNeeded_NeverActiveTimesOut`, `TestCollectTimeoutIfNeeded_IdleWithinGrace`, and the `shouldFailoverImmediately` SSRF case. +> ✅ **FOLLOW-UP FIXED** in `fix/security-critical-high` — the connects-but-no-packets / never-first-packet blind spot is closed. `collectTimeoutIfNeeded` no longer gates on `StatusActive`: it now fires for the **active** input whenever `now − LastPacketAt > timeout` regardless of status (Idle/Degraded excluded only while a failover is already in flight). `Register` stamps `LastPacketAt = now` on each input so the timeout window measures connect-to-first-packet time as a grace period. A switched-to input (manual `/switch` or auto-failover) that connects but never delivers a packet now times out and fails over to the next priority instead of freezing forever. Tests `TestCollectTimeoutIfNeeded_NeverActiveTimesOut`, `TestCollectTimeoutIfNeeded_IdleWithinGrace`. **Files:** `internal/manager/service.go:415-421` (error only logged, `Register` returns nil), `:816` (`collectTimeoutIfNeeded` needs StatusActive), `:846/849` (`collectProbeIfNeeded` needs StatusDegraded); `coordinator.go:189-199` (StreamStatus = Active when registered + no degradation), `:964-966` (reconciler skips IsRunning). diff --git a/internal/autopublish/service.go b/internal/autopublish/service.go index 7d643b6f..e3a8f8f6 100644 --- a/internal/autopublish/service.go +++ b/internal/autopublish/service.go @@ -2,6 +2,7 @@ package autopublish import ( "context" + "crypto/subtle" "errors" "fmt" "log/slog" @@ -131,6 +132,12 @@ var ErrNoMatch = errors.New("autopublish: no template matches path") // dead pipeline. var ErrTemplateNoPush = errors.New("autopublish: matching template has no publish:// input") +// ErrUnauthorized is returned when the matching template requires a StreamKey +// and the pusher's supplied secret does not match (S-8). Checked before any +// runtime stream is materialised, so an unauthorised push never spins a +// pipeline. +var ErrUnauthorized = errors.New("autopublish: stream key mismatch") + // ResolveOrCreate looks up (and, if necessary, creates) the runtime // stream whose code is path. Returns the resolved stream code (always // equal to the canonical form of path). Concurrent callers racing on @@ -140,7 +147,7 @@ var ErrTemplateNoPush = errors.New("autopublish: matching template has no publis // The returned stream is ready for the push server's registry.Acquire // the moment this function returns nil — coordinator.Start spawns the // ingestor's push registration synchronously. -func (s *Service) ResolveOrCreate(ctx context.Context, path string) (domain.StreamCode, error) { +func (s *Service) ResolveOrCreate(ctx context.Context, path, secret string) (domain.StreamCode, error) { code := domain.StreamCode(canonPath(path)) if code == "" { return "", fmt.Errorf("autopublish: empty path") @@ -168,6 +175,14 @@ func (s *Service) ResolveOrCreate(ctx context.Context, path string) (domain.Stre if !domain.TemplateAcceptsPush(tpl) { return "", ErrTemplateNoPush } + // Push auth (S-8): when the template configures a StreamKey, the pusher + // must supply a matching ?key= secret. Empty key = the template opted out. + // Checked before coordinator.Start so an unauthorised push never + // materialises a runtime stream. + if tpl.StreamKey != "" && + subtle.ConstantTimeCompare([]byte(secret), []byte(tpl.StreamKey)) != 1 { + return "", ErrUnauthorized + } // Acquire write lock and re-check to defend against a parallel // caller that won the race between the RLock release and now. diff --git a/internal/autopublish/service_test.go b/internal/autopublish/service_test.go index 21d4f27a..c2af36d1 100644 --- a/internal/autopublish/service_test.go +++ b/internal/autopublish/service_test.go @@ -135,7 +135,7 @@ func TestResolveOrCreate_HappyPath(t *testing.T) { }) require.NoError(t, s.RefreshTemplates(context.Background())) - code, err := s.ResolveOrCreate(context.Background(), "live/foo") + code, err := s.ResolveOrCreate(context.Background(), "live/foo", "") require.NoError(t, err) assert.Equal(t, domain.StreamCode("live/foo"), code) @@ -168,7 +168,7 @@ func TestResolveOrCreate_DedupsConcurrentCallers(t *testing.T) { wg.Add(1) go func() { defer wg.Done() - if _, err := s.ResolveOrCreate(context.Background(), "live/foo"); err == nil { + if _, err := s.ResolveOrCreate(context.Background(), "live/foo", ""); err == nil { ok.Add(1) } }() @@ -185,7 +185,7 @@ func TestResolveOrCreate_NoMatchReturnsErr(t *testing.T) { s, _, _ := newServiceWithDeps(t) require.NoError(t, s.RefreshTemplates(context.Background())) - _, err := s.ResolveOrCreate(context.Background(), "unknown/foo") + _, err := s.ResolveOrCreate(context.Background(), "unknown/foo", "") require.ErrorIs(t, err, ErrNoMatch) } @@ -200,13 +200,42 @@ func TestResolveOrCreate_TemplateWithoutPushRejected(t *testing.T) { }) require.NoError(t, s.RefreshTemplates(context.Background())) - _, err := s.ResolveOrCreate(context.Background(), "live/foo") + _, err := s.ResolveOrCreate(context.Background(), "live/foo", "") require.ErrorIs(t, err, ErrTemplateNoPush) cd.mu.Lock() defer cd.mu.Unlock() assert.Empty(t, cd.started) } +// A template with a StreamKey requires a matching ?key= secret (S-8): a +// wrong/absent secret is rejected with ErrUnauthorized and no stream is +// started; the correct secret materialises the stream. +func TestResolveOrCreate_StreamKeyEnforced(t *testing.T) { + s, tr, cd := newServiceWithDeps(t) + tr.put(&domain.Template{ + Code: "profile-a", + Prefixes: []string{"live"}, + Inputs: []domain.Input{{URL: "publish://"}}, + StreamKey: "s3cr3t", + }) + require.NoError(t, s.RefreshTemplates(context.Background())) + + _, err := s.ResolveOrCreate(context.Background(), "live/foo", "wrong") + require.ErrorIs(t, err, ErrUnauthorized) + _, err = s.ResolveOrCreate(context.Background(), "live/foo", "") + require.ErrorIs(t, err, ErrUnauthorized) + cd.mu.Lock() + assert.Empty(t, cd.started, "an unauthorised push must not start a runtime stream") + cd.mu.Unlock() + + code, err := s.ResolveOrCreate(context.Background(), "live/foo", "s3cr3t") + require.NoError(t, err) + assert.Equal(t, domain.StreamCode("live/foo"), code) + cd.mu.Lock() + assert.Len(t, cd.started, 1, "the correct secret must materialise the stream") + cd.mu.Unlock() +} + // Invalid stream codes (e.g. contain "..") must be rejected at the // boundary even when a template prefix accepts the path. func TestResolveOrCreate_InvalidStreamCodeRejected(t *testing.T) { @@ -218,7 +247,7 @@ func TestResolveOrCreate_InvalidStreamCodeRejected(t *testing.T) { }) require.NoError(t, s.RefreshTemplates(context.Background())) - _, err := s.ResolveOrCreate(context.Background(), "live/../escape") + _, err := s.ResolveOrCreate(context.Background(), "live/../escape", "") require.Error(t, err) } @@ -234,7 +263,7 @@ func TestResolveOrCreate_StartFailureLeavesNoEntry(t *testing.T) { }) require.NoError(t, s.RefreshTemplates(context.Background())) - _, err := s.ResolveOrCreate(context.Background(), "live/foo") + _, err := s.ResolveOrCreate(context.Background(), "live/foo", "") require.Error(t, err) assert.False(t, s.IsRuntime("live/foo")) } diff --git a/internal/hooks/service.go b/internal/hooks/service.go index 30e18a65..cf4f56ca 100644 --- a/internal/hooks/service.go +++ b/internal/hooks/service.go @@ -33,7 +33,6 @@ import ( "github.com/ntt0601zcoder/open-streamer/internal/domain" "github.com/ntt0601zcoder/open-streamer/internal/events" "github.com/ntt0601zcoder/open-streamer/internal/metrics" - "github.com/ntt0601zcoder/open-streamer/internal/netguard" "github.com/ntt0601zcoder/open-streamer/internal/store" ) @@ -87,12 +86,11 @@ func New(i do.Injector) (*Service, error) { hookRepo: hookRepo, bus: bus, // No client-level timeout — each delivery applies its own per-hook - // timeout via context.WithTimeout in postOnce / deliverFile. The client - // carries the SSRF dial guard (S-2): loopback / link-local / - // cloud-metadata (169.254.169.254) are blocked so a webhook target can't - // pivot to the local admin API or instance metadata. Private RFC1918 is - // permitted because internal webhooks are a legitimate use. - client: netguard.NewClient(netguard.Policy{AllowPrivate: true}, 0), + // timeout via context.WithTimeout in postOnce / deliverFile. Webhook + // targets are operator-configured (admin-authenticated) and frequently + // point at private / local endpoints (sidecar collectors, internal + // services), so there is no egress IP guard here. + client: &http.Client{}, batchers: make(map[domain.HookID]*httpBatcher), fileLocks: make(map[string]*sync.Mutex), } diff --git a/internal/ingestor/push/rtmp_server.go b/internal/ingestor/push/rtmp_server.go index b3adafee..15cc2585 100644 --- a/internal/ingestor/push/rtmp_server.go +++ b/internal/ingestor/push/rtmp_server.go @@ -34,9 +34,11 @@ package push import ( "context" + "crypto/subtle" "errors" "fmt" "log/slog" + "net/url" "strings" "sync" "time" @@ -142,9 +144,25 @@ type StreamCallbacks struct { // publish, in which case any push to an unregistered path is rejected // with the existing "stream not registered" semantics. type AutoPublishResolver interface { - ResolveOrCreate(ctx context.Context, path string) (domain.StreamCode, error) + ResolveOrCreate(ctx context.Context, path, secret string) (domain.StreamCode, error) } +// StreamKeyResolver returns the configured push-ingest secret for a stream +// (its resolved `StreamKey`), or "" when the stream has no key set — in which +// case push is unauthenticated for that stream. Used by OnNewRtmpPubSession to +// gate a direct push before claiming the registry slot (S-8). nil resolver = +// push auth disabled (legacy behaviour). +type StreamKeyResolver func(domain.StreamCode) string + +// pushKeyQueryParam is the publish-URL query key an encoder uses to carry the +// stream secret: rtmp://host/?key=SECRET. lal strips the query from the +// stream name, so it never affects routing (rtmpRouteKey). +const pushKeyQueryParam = "key" + +// errUnauthorizedPush rejects a publish whose supplied ?key= does not match the +// stream's configured StreamKey. +var errUnauthorizedPush = errors.New("rtmp server: unauthorized (missing or wrong stream key)") + // RTMPServer accepts RTMP push connections from encoders, validates them // against the push Registry, and writes decoded AVPackets into the Buffer // Hub. External RTMP play clients are served via the optional PlayFunc. @@ -153,11 +171,12 @@ type RTMPServer struct { registry Registry normaliserCfg timeline.Config - mu sync.Mutex - pubs map[*rtmp.ServerSession]*pubState - subs map[*rtmp.ServerSession]*subState - playFunc PlayFunc - resolver AutoPublishResolver + mu sync.Mutex + pubs map[*rtmp.ServerSession]*pubState + subs map[*rtmp.ServerSession]*subState + playFunc PlayFunc + resolver AutoPublishResolver + streamKeyFn StreamKeyResolver // streamCallbacks: streamID → per-stream callback set. Populated by // the ingestor.Service at startPushRegistration time so the closures @@ -224,34 +243,81 @@ func (s *RTMPServer) SetAutoPublishResolver(r AutoPublishResolver) { s.mu.Unlock() } +// SetStreamKeyResolver installs the push-auth resolver (S-8). Passing nil +// disables push key enforcement. Safe to call concurrently with Run. +func (s *RTMPServer) SetStreamKeyResolver(fn StreamKeyResolver) { + s.mu.Lock() + s.streamKeyFn = fn + s.mu.Unlock() +} + +// pushKeyOK reports whether secret authorises a push to streamID. A nil +// resolver (auth disabled) or an empty configured key (stream opted out) both +// allow; otherwise the supplied secret must match in constant time. +func (s *RTMPServer) pushKeyOK(streamID domain.StreamCode, secret string) bool { + s.mu.Lock() + fn := s.streamKeyFn + s.mu.Unlock() + if fn == nil { + return true + } + want := fn(streamID) + if want == "" { + return true + } + return subtle.ConstantTimeCompare([]byte(secret), []byte(want)) == 1 +} + +// pushSecretFromQuery extracts the push secret an encoder supplies as the +// `key` query parameter on the publish URL (rtmp://host/?key=SECRET). +func pushSecretFromQuery(rawQuery string) string { + if rawQuery == "" { + return "" + } + v, err := url.ParseQuery(rawQuery) + if err != nil { + return "" + } + return v.Get(pushKeyQueryParam) +} + // acquireOrAutoPublish tries to claim the registry slot for key. On miss, // the auto-publish resolver (when wired) is consulted: it walks template // prefixes, materialises a runtime stream if one matches, and registers // the slot synchronously. The acquire is then retried. Any non-nil error // surfaces as a publish-rejection at the caller. func (s *RTMPServer) acquireOrAutoPublish( - key, remoteAddr string, + key, secret, remoteAddr string, ) (domain.StreamCode, domain.StreamCode, *buffer.Service, error) { - bufWriteID, streamID, buf, err := s.registry.Acquire(key) - if err == nil { - return bufWriteID, streamID, buf, nil + // Registered slot → direct push. Enforce the stream's configured StreamKey + // (S-8) BEFORE claiming the slot, so a bad key never mutates slot state: + // a stream with a non-empty key requires a matching ?key= on the publish + // URL; an empty key means the stream opted out of push auth. + if _, streamID, _, lerr := s.registry.Lookup(key); lerr == nil { + if !s.pushKeyOK(streamID, secret) { + return "", "", nil, errUnauthorizedPush + } + return s.registry.Acquire(key) } + + // Not registered → auto-publish. The resolver enforces the matching + // template's StreamKey against secret before materialising the stream. s.mu.Lock() resolver := s.resolver s.mu.Unlock() if resolver == nil { - return "", "", nil, err + return "", "", nil, fmt.Errorf("ingestor: no stream registered for key %q", key) } // Use a fresh background context — the resolver may outlive the // callback (it spawns goroutines for the runtime stream). lal's // OnNewRtmpPubSession blocks until we return, so the timeout is // effectively the publish handshake's own deadline; nothing extra // to enforce here. - code, rerr := resolver.ResolveOrCreate(context.Background(), key) + code, rerr := resolver.ResolveOrCreate(context.Background(), key, secret) if rerr != nil { slog.Debug("rtmp server: auto-publish resolver declined", "key", key, "remote", remoteAddr, "err", rerr) - return "", "", nil, err // surface the original "not registered" error + return "", "", nil, rerr } return s.registry.Acquire(string(code)) } @@ -357,7 +423,8 @@ func (s *RTMPServer) OnNewRtmpPubSession(session *rtmp.ServerSession) error { ) return errors.New("rtmp server: invalid stream path") } - bufWriteID, streamID, buf, err := s.acquireOrAutoPublish(key, session.GetStat().RemoteAddr) + secret := pushSecretFromQuery(session.RawQuery()) + bufWriteID, streamID, buf, err := s.acquireOrAutoPublish(key, secret, session.GetStat().RemoteAddr) if err != nil { slog.Warn("rtmp server: rejected publisher", "key", key, diff --git a/internal/ingestor/push/rtmp_server_test.go b/internal/ingestor/push/rtmp_server_test.go index d676a53c..39bff631 100644 --- a/internal/ingestor/push/rtmp_server_test.go +++ b/internal/ingestor/push/rtmp_server_test.go @@ -4,6 +4,8 @@ import ( "testing" "github.com/stretchr/testify/assert" + + "github.com/ntt0601zcoder/open-streamer/internal/domain" ) // rtmpRouteKey reconstructs the URL path from `app` + `streamName`, strips @@ -34,3 +36,50 @@ func TestRTMPRouteKey(t *testing.T) { }) } } + +// pushSecretFromQuery pulls the `key` param off the publish-URL query. +func TestPushSecretFromQuery(t *testing.T) { + t.Parallel() + cases := []struct { + name, rawQuery, want string + }{ + {"empty", "", ""}, + {"key present", "key=s3cr3t", "s3cr3t"}, + {"key among others", "a=1&key=s3cr3t&b=2", "s3cr3t"}, + {"no key param", "token=x", ""}, + {"malformed", "%zz", ""}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, pushSecretFromQuery(tc.rawQuery)) + }) + } +} + +// pushKeyOK gates a publish against the stream's configured StreamKey (S-8): +// no resolver or an empty configured key allows; otherwise the secret must +// match in constant time. +func TestPushKeyOK(t *testing.T) { + t.Parallel() + + t.Run("nil resolver allows (auth disabled)", func(t *testing.T) { + s := &RTMPServer{} + assert.True(t, s.pushKeyOK("foo", "")) + assert.True(t, s.pushKeyOK("foo", "anything")) + }) + + t.Run("empty configured key allows (opted out)", func(t *testing.T) { + s := &RTMPServer{} + s.SetStreamKeyResolver(func(domain.StreamCode) string { return "" }) + assert.True(t, s.pushKeyOK("foo", "")) + assert.True(t, s.pushKeyOK("foo", "whatever")) + }) + + t.Run("configured key requires exact match", func(t *testing.T) { + s := &RTMPServer{} + s.SetStreamKeyResolver(func(domain.StreamCode) string { return "s3cr3t" }) + assert.True(t, s.pushKeyOK("foo", "s3cr3t")) + assert.False(t, s.pushKeyOK("foo", "wrong")) + assert.False(t, s.pushKeyOK("foo", "")) + }) +} diff --git a/internal/ingestor/service.go b/internal/ingestor/service.go index db40e04c..c80937cc 100644 --- a/internal/ingestor/service.go +++ b/internal/ingestor/service.go @@ -89,6 +89,9 @@ type Service struct { // Run created the RTMP server. Same deferred-wiring pattern as the // PlayFunc / push callbacks above. pendingAutoPublish push.AutoPublishResolver + // pendingStreamKeyFn holds the push-auth (StreamKey) resolver registered + // before Run created the RTMP server. Same deferred-wiring pattern. + pendingStreamKeyFn push.StreamKeyResolver } // New creates a Service and registers it with the DI injector. @@ -199,6 +202,24 @@ func (s *Service) SetAutoPublishResolver(r push.AutoPublishResolver) { s.mu.Unlock() } +// SetPushStreamKeyResolver installs the push-ingest auth resolver (S-8): the +// RTMP server consults it for the configured StreamKey of a stream before +// accepting a publisher. Same deferred-wiring semantics as +// SetAutoPublishResolver — safe to call before or after Run. nil disables +// push key enforcement. +func (s *Service) SetPushStreamKeyResolver(fn push.StreamKeyResolver) { + s.mu.Lock() + srv := s.rtmpSrv + s.mu.Unlock() + if srv != nil { + srv.SetStreamKeyResolver(fn) + return + } + s.mu.Lock() + s.pendingStreamKeyFn = fn + s.mu.Unlock() +} + // Run starts the shared push/play listeners (RTMP) and blocks until ctx is // cancelled. Other shared listeners (SRT, RTSP) are owned by the publisher and // run from runtime.Manager; the same network port serves both ingest and play @@ -227,6 +248,10 @@ func (s *Service) Run(ctx context.Context) error { rtmpSrv.SetAutoPublishResolver(s.pendingAutoPublish) s.pendingAutoPublish = nil } + if s.pendingStreamKeyFn != nil { + rtmpSrv.SetStreamKeyResolver(s.pendingStreamKeyFn) + s.pendingStreamKeyFn = nil + } // Wire pending push callbacks (deferred from startPushRegistration // when the server didn't exist yet — typical on cold start where // streams are activated before Run binds the listener). diff --git a/internal/netguard/netguard.go b/internal/netguard/netguard.go deleted file mode 100644 index c5ad5e67..00000000 --- a/internal/netguard/netguard.go +++ /dev/null @@ -1,175 +0,0 @@ -// Package netguard provides an SSRF egress guard for the webhook HTTP client, -// which makes outbound requests to operator-supplied URLs. -// -// The core is a dial-time IP filter installed via net.Dialer.Control. Control -// runs with the RESOLVED ip:port for each connection attempt, so it sees the -// real destination even when a hostname resolves to a forbidden address — this -// defeats DNS-rebinding and redirect-to-internal chains that a save-time string -// check cannot. -// -// Link-local (which covers cloud-metadata 169.254.169.254) and the unspecified -// address are rejected UNCONDITIONALLY. Loopback and private (RFC1918 / IPv6-ULA -// / RFC6598) are each gated by the caller's Policy: -// -// - Webhooks: AllowPrivate (internal webhooks are legitimate) but NOT loopback -// (a hook must never reach the local admin API). -// -// NOTE: ingest pull clients (HLS / HTTP-TS) historically applied this guard via -// IngestPolicy + ingestor.allow_private_targets, but ingest SSRF guarding was -// removed by operator decision — the server's primary role is pulling from -// internal/LAN sources, so blocking private targets was counterproductive (see -// the security audit, S-4). IngestPolicy / ValidateInputURL remain as helpers. -package netguard - -import ( - "errors" - "fmt" - "net" - "net/http" - "net/url" - "strings" - "syscall" - "time" -) - -// MaxRedirects caps redirect chains (matches Go's default) and is enforced by -// CheckRedirect so a redirect loop can't be used to amplify or evade the guard. -const MaxRedirects = 10 - -// cgNAT is RFC 6598 shared address space (100.64.0.0/10) — used by some cloud -// metadata services (e.g. Alibaba 100.100.100.200) and never a valid public -// origin, so it is treated like RFC1918 under the AllowPrivate gate. -var cgNAT = mustCIDR("100.64.0.0/10") - -func mustCIDR(s string) *net.IPNet { - _, n, err := net.ParseCIDR(s) - if err != nil { - panic(err) - } - return n -} - -// ErrBlocked is the sentinel wrapped by every guard rejection. -var ErrBlocked = errors.New("netguard: destination blocked") - -// Policy selects which otherwise-internal destinations a guard tolerates. -// Link-local / cloud-metadata / unspecified are always blocked regardless. -type Policy struct { - // AllowLoopback permits 127.0.0.0/8 and ::1. - AllowLoopback bool - // AllowPrivate permits RFC1918, IPv6-ULA, and RFC6598 shared space. - AllowPrivate bool -} - -// IngestPolicy is the policy for ingest pull clients: a single -// allow_private_targets operator opt-in unlocks BOTH loopback and private -// (a trusted-network "allow internal sources" switch). Default false blocks -// all internal targets. -func IngestPolicy(allowPrivate bool) Policy { - return Policy{AllowLoopback: allowPrivate, AllowPrivate: allowPrivate} -} - -// blockedIP reports why ip must not be dialed under p, or nil when allowed. -func blockedIP(ip net.IP, p Policy) error { - switch { - case ip == nil: - return nil // unparseable here — the dialer will fail on its own - case ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast(): - return fmt.Errorf("%w: link-local / cloud-metadata (169.254.0.0/16, fe80::/10)", ErrBlocked) - case ip.IsUnspecified(): - return fmt.Errorf("%w: unspecified address", ErrBlocked) - case ip.IsLoopback(): - if p.AllowLoopback { - return nil - } - return fmt.Errorf("%w: loopback", ErrBlocked) - } - if !p.AllowPrivate && (ip.IsPrivate() || cgNAT.Contains(ip)) { - return fmt.Errorf("%w: private/shared address", ErrBlocked) - } - return nil -} - -// dialControl returns a net.Dialer.Control hook that rejects forbidden resolved -// IPs. address is the post-DNS "ip:port". -func dialControl(p Policy) func(network, address string, c syscall.RawConn) error { - return func(_ /*network*/, address string, _ syscall.RawConn) error { - host, _, err := net.SplitHostPort(address) - if err != nil { - host = address - } - if e := blockedIP(net.ParseIP(host), p); e != nil { - return fmt.Errorf("%s: %w", host, e) - } - return nil - } -} - -// ApplyToTransport installs the dial-time guard on t (replacing its DialContext) -// so connections to forbidden IPs fail at connect time. -func ApplyToTransport(t *http.Transport, p Policy) { - if t == nil { - return - } - d := &net.Dialer{Timeout: 30 * time.Second, KeepAlive: 30 * time.Second, Control: dialControl(p)} - t.DialContext = d.DialContext -} - -// CheckRedirect returns an http.Client.CheckRedirect that caps the chain depth -// and strips credential headers when a redirect crosses to a different host (so -// configured upstream auth isn't leaked to an attacker-chosen target). The -// redirect's own dial is still re-validated by the transport's dial guard. -func CheckRedirect() func(req *http.Request, via []*http.Request) error { - return func(req *http.Request, via []*http.Request) error { - if len(via) >= MaxRedirects { - return fmt.Errorf("netguard: stopped after %d redirects", MaxRedirects) - } - if len(via) > 0 && !strings.EqualFold(req.URL.Host, via[len(via)-1].URL.Host) { - for _, h := range []string{"Authorization", "Cookie", "Proxy-Authorization"} { - req.Header.Del(h) - } - } - return nil - } -} - -// NewClient builds an http.Client with the dial guard + redirect policy. A -// timeout of 0 means no client-level timeout (stream-body callers set 0 and -// rely on per-request context deadlines). -func NewClient(p Policy, timeout time.Duration) *http.Client { - t := http.DefaultTransport.(*http.Transport).Clone() - ApplyToTransport(t, p) - return &http.Client{ - Timeout: timeout, - Transport: t, - CheckRedirect: CheckRedirect(), - } -} - -// ValidateInputURL is a fast, config-independent save-time check: it rejects a -// URL whose host is a literal always-blocked IP (link-local / cloud-metadata / -// unspecified). It deliberately does NOT block loopback or private IPs (those -// are gated at dial time by the Policy, so a LAN or local source still saves) -// and does NOT resolve DNS. The authoritative guard is ApplyToTransport — this -// is fast operator feedback. -func ValidateInputURL(raw string) error { - u, err := url.Parse(strings.TrimSpace(raw)) - if err != nil { - return fmt.Errorf("invalid URL: %w", err) - } - host := u.Hostname() - if host == "" { - return nil // file:// / publish:// / schemes without a host — out of scope - } - ip := net.ParseIP(host) - if ip == nil { - return nil // hostname — resolved + guarded at dial time - } - switch { - case ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast(): - return fmt.Errorf("host %q is a link-local / cloud-metadata address", host) - case ip.IsUnspecified(): - return fmt.Errorf("host %q is the unspecified address", host) - } - return nil -} diff --git a/internal/netguard/netguard_test.go b/internal/netguard/netguard_test.go deleted file mode 100644 index c71dc704..00000000 --- a/internal/netguard/netguard_test.go +++ /dev/null @@ -1,117 +0,0 @@ -package netguard - -import ( - "context" - "net" - "net/http" - "testing" -) - -func TestBlockedIP(t *testing.T) { - t.Parallel() - none := Policy{} - priv := Policy{AllowPrivate: true} // webhooks: private OK, loopback NOT - internal := Policy{AllowLoopback: true, AllowPrivate: true} // ingest opt-in - cases := []struct { - ip string - p Policy - blocked bool - }{ - // Link-local / cloud-metadata / unspecified: blocked UNCONDITIONALLY. - {"169.254.169.254", internal, true}, // AWS/GCP/Azure IMDS (link-local) - {"169.254.1.1", internal, true}, - {"fe80::1", internal, true}, - {"0.0.0.0", internal, true}, - {"::", internal, true}, - // Loopback: blocked by default + under AllowPrivate-only (webhook policy), - // allowed only under AllowLoopback. - {"127.0.0.1", none, true}, - {"127.0.0.1", priv, true}, // webhook policy must NOT reach localhost - {"::1", priv, true}, - {"127.0.0.1", internal, false}, - {"::1", internal, false}, - // Private: blocked by default, allowed under AllowPrivate. - {"10.0.0.1", none, true}, - {"192.168.1.10", none, true}, - {"fc00::1", none, true}, // IPv6 ULA - {"100.100.100.200", none, true}, // RFC6598 CGNAT (Alibaba metadata) - {"10.0.0.1", priv, false}, - {"192.168.1.10", priv, false}, - {"fc00::1", priv, false}, - // Public: always allowed. - {"8.8.8.8", none, false}, - {"1.1.1.1", none, false}, - // Multicast (UDP ingest) is not private → allowed. - {"239.0.0.1", none, false}, - } - for _, c := range cases { - err := blockedIP(net.ParseIP(c.ip), c.p) - if (err != nil) != c.blocked { - t.Errorf("blockedIP(%s, %+v) err=%v, want blocked=%v", c.ip, c.p, err, c.blocked) - } - } -} - -func TestValidateInputURL(t *testing.T) { - t.Parallel() - bad := []string{ - "http://169.254.169.254/latest/meta-data/", - "http://0.0.0.0/x", - } - for _, u := range bad { - if err := ValidateInputURL(u); err == nil { - t.Errorf("ValidateInputURL(%q) = nil, want error", u) - } - } - good := []string{ - "http://cdn.example.com/playlist.m3u8", // hostname → checked at dial time - "https://8.8.8.8/x", - "http://127.0.0.1:8080/x", // loopback NOT blocked at save time (dial-time gated) - "http://10.0.0.5/x", // private NOT blocked at save time (dial-time gated) - "rtsp://192.168.1.9/s", - "udp://239.0.0.1:1234", - "file:///srv/media/a.ts", - "publish://ingest", - } - for _, u := range good { - if err := ValidateInputURL(u); err != nil { - t.Errorf("ValidateInputURL(%q) = %v, want nil", u, err) - } - } -} - -func TestCheckRedirect(t *testing.T) { - t.Parallel() - cr := CheckRedirect() - - ctx := context.Background() - // Depth cap. - via := make([]*http.Request, MaxRedirects) - req, _ := http.NewRequestWithContext(ctx, http.MethodGet, "http://a.example/x", nil) - if err := cr(req, via); err == nil { - t.Error("expected redirect-depth error at the cap") - } - - // Cross-host strips credential headers. - prev, _ := http.NewRequestWithContext(ctx, http.MethodGet, "http://a.example/x", nil) - next, _ := http.NewRequestWithContext(ctx, http.MethodGet, "http://b.example/y", nil) - next.Header.Set("Authorization", "Bearer secret") - next.Header.Set("Cookie", "sid=1") - if err := cr(next, []*http.Request{prev}); err != nil { - t.Fatalf("unexpected error: %v", err) - } - if next.Header.Get("Authorization") != "" || next.Header.Get("Cookie") != "" { - t.Error("credential headers must be stripped on cross-host redirect") - } - - // Same-host keeps headers. - p2, _ := http.NewRequestWithContext(ctx, http.MethodGet, "http://a.example/x", nil) - n2, _ := http.NewRequestWithContext(ctx, http.MethodGet, "http://a.example/y", nil) - n2.Header.Set("Authorization", "Bearer keep") - if err := cr(n2, []*http.Request{p2}); err != nil { - t.Fatalf("unexpected error: %v", err) - } - if n2.Header.Get("Authorization") == "" { - t.Error("same-host redirect must keep headers") - } -} diff --git a/internal/publisher/playauth.go b/internal/publisher/playauth.go index 1bf6b924..4dbf932e 100644 --- a/internal/publisher/playauth.go +++ b/internal/publisher/playauth.go @@ -31,6 +31,22 @@ func (s *Service) PlaybackPolicy(code domain.StreamCode) (policy domain.PolicyCo return "", false } +// PushStreamKey returns a RUNNING stream's configured push-ingest secret +// (its resolved StreamKey; "" = no secret → push unauthenticated for that +// stream) and whether the stream is currently running. O(1) in-memory lookup, +// the first tier of the RTMP server's push-auth resolver (S-8) so the connect +// path avoids a store read for live streams — including auto-publish runtime +// streams that have no on-disk record. The `running` flag lets the caller fall +// back to the store + template for a stopped/configured stream. +func (s *Service) PushStreamKey(code domain.StreamCode) (streamKey string, running bool) { + s.mu.Lock() + defer s.mu.Unlock() + if ss, ok := s.streams[code]; ok { + return ss.streamKey, 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. diff --git a/internal/publisher/service.go b/internal/publisher/service.go index 7f4e6f05..c635bb0c 100644 --- a/internal/publisher/service.go +++ b/internal/publisher/service.go @@ -87,6 +87,11 @@ type streamState struct { // authorizer with an O(1) in-memory lookup instead of a store read per // request. playbackPolicy domain.PolicyCode + // streamKey is the resolved push-ingest secret ("" = push unauthenticated). + // Mirrored here so PushStreamKey answers the RTMP server's push-auth + // resolver (S-8) with an O(1) in-memory lookup for live streams, including + // auto-publish runtime streams not present in the store. + streamKey string } // Service manages all output workers for active streams. @@ -286,6 +291,7 @@ func (s *Service) Start(ctx context.Context, stream *domain.Stream) error { protocols: make(map[string]context.CancelFunc), mpegtsEnabled: p.MPEGTS, playbackPolicy: stream.PlaybackPolicy, + streamKey: stream.StreamKey, } s.streams[stream.Code] = ss s.mediaBuffer[stream.Code] = ss.mediaBuf @@ -454,6 +460,7 @@ func (s *Service) UpdateProtocols(ctx context.Context, old, new *domain.Stream) } ss.mpegtsEnabled = np.MPEGTS ss.playbackPolicy = new.PlaybackPolicy + ss.streamKey = new.StreamKey } s.mu.Unlock() if !ok {