fix(push): enforce per-stream push authentication (S-8); remove webhook egress guard; triage remaining findings

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": {

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": {

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:

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.