diff --git a/pkg/cmd/streamplace.go b/pkg/cmd/streamplace.go index 4a399290..6e1cf4ba 100644 --- a/pkg/cmd/streamplace.go +++ b/pkg/cmd/streamplace.go @@ -215,6 +215,12 @@ func runMain(ctx context.Context, build *config.BuildFlags, platformJobs []jobFu } cli.AccessJWK = accessJWK + serviceAuthKey, err := state.EnsureServiceAuthKey(ctx) + if err != nil { + return err + } + cli.ServiceAuthKey = serviceAuthKey + b := bus.NewBus() atsync := &atproto.ATProtoSynchronizer{ CLI: cli, diff --git a/pkg/config/config.go b/pkg/config/config.go index 66e9fbe2..6a37d957 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -116,6 +116,7 @@ type CLI struct { RateLimitWebsocket int JWK jwk.Key AccessJWK jwk.Key + ServiceAuthKey jwk.Key dataDirFlags []*string DiscordWebhooks []*discordtypes.Webhook NewWebRTCPlayback bool diff --git a/pkg/spxrpc/place_stream_playback.go b/pkg/spxrpc/place_stream_playback.go index 9cca974c..1ad5223b 100644 --- a/pkg/spxrpc/place_stream_playback.go +++ b/pkg/spxrpc/place_stream_playback.go @@ -9,6 +9,8 @@ import ( "github.com/labstack/echo/v4" "github.com/pion/webrtc/v4" "github.com/streamplace/oatproxy/pkg/oatproxy" + "stream.place/streamplace/pkg/aqhttp" + "stream.place/streamplace/pkg/log" ) func (s *Server) handlePlaceStreamPlaybackWhep(ctx context.Context, rendition string, streamer string, r io.Reader, _contentType string) (io.Reader, error) { @@ -20,19 +22,68 @@ func (s *Server) handlePlaceStreamPlaybackWhep(ctx context.Context, rendition st return nil, echo.NewHTTPError(http.StatusBadRequest, "rendition is required") } viewer := "" - session, _ := oatproxy.GetOAuthSession(ctx) - if session != nil { - viewer = session.DID - } repo, err := s.ATSync.SyncBlueskyRepoCached(ctx, streamer) if err != nil { return nil, err } + streamer = repo.DID + session, _ := oatproxy.GetOAuthSession(ctx) + if session != nil { + viewer = session.DID + } else { + svc := GetServiceAuth(ctx) + if svc != nil { + log.Warn(ctx, "service auth present", "service_did", svc.DID) + // this is a signed request from a peer node, allow them to see unpublished streams + viewer = streamer + } + } body, err := io.ReadAll(r) if err != nil { return nil, echo.NewHTTPError(http.StatusBadRequest, "error reading body", err) } offer := webrtc.SessionDescription{Type: webrtc.SDPTypeOffer, SDP: string(body)} + if streamer == viewer { + // this user gets sent right to the origin in case we're unpublished + origin, err := s.statefulDB.GetLatestBroadcastOriginForStreamer(streamer) + if err != nil { + return nil, echo.NewHTTPError(http.StatusInternalServerError, "error getting broadcast origin", err) + } + log.Warn(ctx, "origin", "origin", origin) + myDID := s.cli.ServerDID() + if origin != nil && origin.ServerDID != myDID { + log.Warn(ctx, "proxying to origin", "origin", origin.ServerDID, "myDID", myDID) + token, err := CreateServiceToken(s.cli.ServiceAuthKey, s.cli.ServerDID()) + if err != nil { + return nil, echo.NewHTTPError(http.StatusInternalServerError, "error creating service token", err) + } + parsedOrigin := origin.ServerDID + if len(parsedOrigin) > len("did:web:") && parsedOrigin[:len("did:web:")] == "did:web:" { + parsedOrigin = parsedOrigin[len("did:web:"):] + } + url := "https://" + parsedOrigin + "/xrpc/place.stream.playback.whep?rendition=" + rendition + "&streamer=" + streamer + req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader([]byte(offer.SDP))) + if err != nil { + return nil, echo.NewHTTPError(http.StatusInternalServerError, "failed to construct origin request", err) + } + req.Header.Set("Content-Type", _contentType) + req.Header.Set("X-Streamplace-Service-Auth", token) + resp, err := aqhttp.Client.Do(req) + if err != nil { + return nil, echo.NewHTTPError(http.StatusBadGateway, "error proxying to origin", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return nil, echo.NewHTTPError(resp.StatusCode, "origin error: "+string(body)) + } + data, err := io.ReadAll(resp.Body) + if err != nil { + return nil, echo.NewHTTPError(http.StatusInternalServerError, "error reading origin response", err) + } + return bytes.NewReader(data), nil + } + } answer, err := s.mm.WebRTCPlayback2(ctx, repo.DID, rendition, &offer, viewer) if err != nil { return nil, echo.NewHTTPError(http.StatusInternalServerError, "error playing back", err) diff --git a/pkg/spxrpc/service_auth.go b/pkg/spxrpc/service_auth.go new file mode 100644 index 00000000..5fbb33fc --- /dev/null +++ b/pkg/spxrpc/service_auth.go @@ -0,0 +1,100 @@ +package spxrpc + +import ( + "context" + "fmt" + "time" + + "github.com/labstack/echo/v4" + "github.com/lestrrat-go/jwx/v2/jwa" + "github.com/lestrrat-go/jwx/v2/jwk" + "github.com/lestrrat-go/jwx/v2/jwt" + "stream.place/streamplace/pkg/log" +) + +const serviceAuthHeader = "X-Streamplace-Service-Auth" +const serviceTokenLifetime = 5 * time.Minute + +// ServiceIdentity represents an authenticated peer node in the same station. +type ServiceIdentity struct { + DID string +} + +type serviceAuthContextKeyType struct{} + +var serviceAuthContextKey = serviceAuthContextKeyType{} + +// GetServiceAuth returns the authenticated service identity from the context, +// or nil if the request did not come from an authenticated peer node. +func GetServiceAuth(ctx context.Context) *ServiceIdentity { + v := ctx.Value(serviceAuthContextKey) + if v == nil { + return nil + } + identity, ok := v.(*ServiceIdentity) + if !ok { + return nil + } + return identity +} + +// ServiceAuthMiddleware checks for a service-to-service JWT in the +// X-Streamplace-Service-Auth header. If present and valid, it populates +// the context with the caller's ServiceIdentity. If absent or invalid, +// the request passes through unchanged for the normal OAuth flow. +func (s *Server) ServiceAuthMiddleware() echo.MiddlewareFunc { + return func(next echo.HandlerFunc) echo.HandlerFunc { + return func(c echo.Context) error { + tokenStr := c.Request().Header.Get(serviceAuthHeader) + if tokenStr == "" { + return next(c) + } + + ctx := c.Request().Context() + key := s.cli.ServiceAuthKey + if key == nil { + log.Warn(ctx, "service auth token present but no service auth key configured") + return next(c) + } + + token, err := jwt.Parse([]byte(tokenStr), jwt.WithKey(jwa.HS256, key), jwt.WithValidate(true)) + if err != nil { + log.Warn(ctx, "invalid service auth token", "error", err) + return next(c) + } + + issuer := token.Issuer() + if issuer == "" { + log.Warn(ctx, "service auth token missing issuer claim") + return next(c) + } + + identity := &ServiceIdentity{DID: issuer} + ctx = context.WithValue(ctx, serviceAuthContextKey, identity) + c.SetRequest(c.Request().WithContext(ctx)) + log.Warn(ctx, "authenticated service request", "service_did", issuer) + return next(c) + } + } +} + +// CreateServiceToken generates a signed JWT for authenticating this node +// to another node in the same station. +func CreateServiceToken(key jwk.Key, serverDID string) (string, error) { + now := time.Now() + token, err := jwt.NewBuilder(). + Issuer(serverDID). + IssuedAt(now). + Expiration(now.Add(serviceTokenLifetime)). + Build() + if err != nil { + return "", fmt.Errorf("failed to build service token: %w", err) + } + + signed, err := jwt.Sign(token, jwt.WithKey(jwa.HS256, key)) + if err != nil { + return "", fmt.Errorf("failed to sign service token: %w", err) + } + + return string(signed), nil +} diff --git a/pkg/spxrpc/spxrpc.go b/pkg/spxrpc/spxrpc.go index 8265630d..b2791075 100644 --- a/pkg/spxrpc/spxrpc.go +++ b/pkg/spxrpc/spxrpc.go @@ -57,6 +57,7 @@ func NewServer(ctx context.Context, cli *config.CLI, model model.Model, stateful e.Use(s.ErrorHandlingMiddleware()) e.Use(s.ContextPreservingMiddleware()) e.Use(echomiddleware.Handler("", mdlw)) + e.Use(s.ServiceAuthMiddleware()) e.Use(op.OAuthMiddleware) err := s.RegisterHandlersPlaceStream(e) if err != nil { diff --git a/pkg/statedb/service_auth.go b/pkg/statedb/service_auth.go new file mode 100644 index 00000000..c3d30ab3 --- /dev/null +++ b/pkg/statedb/service_auth.go @@ -0,0 +1,52 @@ +package statedb + +import ( + "context" + "crypto/rand" + "encoding/json" + "fmt" + + "github.com/lestrrat-go/jwx/v2/jwk" + "stream.place/streamplace/pkg/log" +) + +// EnsureServiceAuthKey ensures a shared symmetric key exists in the config table +// for intra-service JWT authentication. All nodes sharing the same database will +// use the same key, enabling mutual authentication within a station. +func (state *StatefulDB) EnsureServiceAuthKey(ctx context.Context) (jwk.Key, error) { + conf, err := state.GetConfig("service-auth-key") + if err != nil { + return nil, fmt.Errorf("failed to get service auth key: %w", err) + } + + if conf != nil { + key, err := jwk.ParseKey(conf.Value) + if err != nil { + return nil, fmt.Errorf("failed to parse service auth key: %w", err) + } + return key, nil + } + + log.Warn(ctx, "no service auth key found, generating new one") + + secret := make([]byte, 32) + if _, err := rand.Read(secret); err != nil { + return nil, fmt.Errorf("failed to generate random bytes: %w", err) + } + + key, err := jwk.FromRaw(secret) + if err != nil { + return nil, fmt.Errorf("failed to create symmetric key: %w", err) + } + + b, err := json.Marshal(key) + if err != nil { + return nil, fmt.Errorf("failed to marshal service auth key: %w", err) + } + + if err := state.PutConfig("service-auth-key", b); err != nil { + return nil, fmt.Errorf("failed to save service auth key: %w", err) + } + + return key, nil +}