mirror of
https://github.com/livekit/livekit.git
synced 2026-09-25 17:44:10 +00:00
Paul Wells
1944cb495d
The registry was doing three jobs at once: fencing worker epochs by id, listing a deployment's candidates, and holding its merged route table. The last two are per-deployment state, so it keyed them on (api key, agent name, deployment) and grew a tenancy concept that only an embedder can actually define. Cloud has to lie to it, passing a project id in a field named APIKey. Split them. Scope is one deployment's serving state and stores no identity at all; whoever embeds the package keys a map of scopes however its own tenancy works, and hands the front a resolved one. Registry keeps only the worker-id fence, which is genuinely node-wide: worker ids are server-issued, so an epoch is superseded wherever it was scoped. The front loses its registry, its SingleKeyFallback and FallbackRequest: the resolver now returns the scope and a fallback already curried on the deployment, plus an ok that carries the 401-vs-503 split the empty api key used to encode. routeTable drops its key and takes the scope's logger, so identity is curried in rather than stored. pkg/service takes ownership of the "api key is the tenant" rule, which is true there and nowhere else, and of releasing a scope once nothing holds it. Behavior is unchanged, including serving public routes to an unauthenticated caller when one configured key or one attached tenant makes the key unambiguous.
387 lines
15 KiB
Go
387 lines
15 KiB
Go
// Copyright 2026 LiveKit, Inc.
|
|
|
|
package endpoint
|
|
|
|
import (
|
|
"context"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"net/url"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/require"
|
|
|
|
"github.com/livekit/protocol/livekit"
|
|
"github.com/livekit/protocol/logger"
|
|
)
|
|
|
|
// grantedTo resolves every request to scope with full access.
|
|
func grantedTo(scope *Scope) AccessResolver {
|
|
return func(*http.Request, string, string) (Access, bool) {
|
|
return Access{Scope: scope, Level: AccessGranted}, true
|
|
}
|
|
}
|
|
|
|
// resolveTo resolves every request to a fixed access.
|
|
func resolveTo(a Access) AccessResolver {
|
|
return func(*http.Request, string, string) (Access, bool) { return a, true }
|
|
}
|
|
|
|
func fallbackFront(t *testing.T, fb Fallback, withWorker bool) *Front {
|
|
g, s := NewRegistry(), NewScope(logger.GetLogger())
|
|
if withWorker {
|
|
m, err := ParseManifest([]*livekit.AgentHttp_AgentEndpoint{
|
|
{Path: "/known", Methods: []string{"GET"}, Public: true},
|
|
})
|
|
require.NoError(t, err)
|
|
g.Register(s, NewRegistration(RegistrationParams{WorkerID: "w1", Manifest: m, Session: &fakeSession{}}))
|
|
}
|
|
return NewFront(FrontParams{
|
|
ResolveAccess: resolveTo(Access{Scope: s, Fallback: fb, Level: AccessGranted}),
|
|
Logger: logger.GetLogger(),
|
|
})
|
|
}
|
|
|
|
func serveFront(f *Front, path string) *httptest.ResponseRecorder {
|
|
w := httptest.NewRecorder()
|
|
f.ServeHTTP(w, httptest.NewRequest(http.MethodGet, PathPrefix+"a/d"+path, nil))
|
|
return w
|
|
}
|
|
|
|
// a path no local worker matches hands off to the fallback the resolver curried,
|
|
// which is given the caller's access level; when the fallback serves, the front
|
|
// writes nothing itself.
|
|
func TestFrontFallbackFires(t *testing.T) {
|
|
got := AccessLevel(-1)
|
|
f := fallbackFront(t, func(w http.ResponseWriter, _ *http.Request, level AccessLevel) bool {
|
|
got = level
|
|
w.WriteHeader(http.StatusTeapot) // stands in for a relayed response
|
|
return true
|
|
}, true)
|
|
|
|
w := serveFront(f, "/unknown")
|
|
require.Equal(t, http.StatusTeapot, w.Code)
|
|
require.Equal(t, AccessGranted, got)
|
|
}
|
|
|
|
// a declined fallback with a local worker present falls through to the front's
|
|
// own 404 for the unmatched path.
|
|
func TestFrontFallbackDeclinedMapsStatus(t *testing.T) {
|
|
f := fallbackFront(t, func(http.ResponseWriter, *http.Request, AccessLevel) bool { return false }, true)
|
|
require.Equal(t, http.StatusNotFound, serveFront(f, "/unknown").Code)
|
|
}
|
|
|
|
// a declined fallback with no local worker for the deployment falls through to
|
|
// 503.
|
|
func TestFrontFallbackDeclinedNoCandidates(t *testing.T) {
|
|
f := fallbackFront(t, func(http.ResponseWriter, *http.Request, AccessLevel) bool { return false }, false)
|
|
require.Equal(t, http.StatusServiceUnavailable, serveFront(f, "/unknown").Code)
|
|
}
|
|
|
|
// a resolver that cannot place the request at all challenges, and never reaches
|
|
// a scope. This is the path that used to ride an empty api key.
|
|
func TestFrontUnresolvedAccessIsChallenged(t *testing.T) {
|
|
f := NewFront(FrontParams{
|
|
ResolveAccess: func(*http.Request, string, string) (Access, bool) { return Access{}, false },
|
|
Logger: logger.GetLogger(),
|
|
})
|
|
w := serveFront(f, "/anything")
|
|
require.Equal(t, http.StatusUnauthorized, w.Code)
|
|
require.Equal(t, "Bearer", w.Header().Get("WWW-Authenticate"))
|
|
}
|
|
|
|
// a resolved caller whose deployment no worker holds here, with no fallback, is
|
|
// 503 - distinct from the 401 above, and no longer distinguished by a string.
|
|
func TestFrontResolvedButNoScopeIsUnavailable(t *testing.T) {
|
|
f := NewFront(FrontParams{
|
|
ResolveAccess: resolveTo(Access{Level: AccessGranted}),
|
|
Logger: logger.GetLogger(),
|
|
})
|
|
require.Equal(t, http.StatusServiceUnavailable, serveFront(f, "/anything").Code)
|
|
}
|
|
|
|
func TestRequestIDAcceptsOrRefuses(t *testing.T) {
|
|
atBound := strings.Repeat("x", maxRequestIDLen)
|
|
cases := []struct {
|
|
name string
|
|
values []string
|
|
want string
|
|
ok bool
|
|
}{
|
|
{name: "absent mints one", ok: true},
|
|
{name: "uuid", values: []string{"f81d4fae-7dec-11d0-a765-00a0c91e6bf6"}, want: "f81d4fae-7dec-11d0-a765-00a0c91e6bf6", ok: true},
|
|
{name: "base64url", values: []string{"a-B_c9=="}, want: "a-B_c9==", ok: true},
|
|
{name: "at the length bound", values: []string{atBound}, want: atBound, ok: true},
|
|
{name: "past the length bound", values: []string{atBound + "x"}},
|
|
{name: "empty value", values: []string{""}},
|
|
{name: "newline", values: []string{"ab\ncd"}},
|
|
{name: "space", values: []string{"ab cd"}},
|
|
{name: "non ascii", values: []string{"abc\u00e9"}},
|
|
{name: "two values", values: []string{"a", "b"}},
|
|
}
|
|
for _, tc := range cases {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
r := httptest.NewRequest(http.MethodGet, "/x", nil)
|
|
for _, v := range tc.values {
|
|
r.Header.Add("X-Request-Id", v)
|
|
}
|
|
got, ok := requestID(r)
|
|
require.Equal(t, tc.ok, ok)
|
|
switch {
|
|
case !tc.ok:
|
|
require.Empty(t, got, "a refused token must not reach the worker or the logs")
|
|
case tc.want != "":
|
|
require.Equal(t, tc.want, got, "a client token is never rewritten")
|
|
default:
|
|
require.True(t, strings.HasPrefix(got, "AER_"))
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// a token the front cannot honor is refused before any routing or dispatch, so
|
|
// the caller learns to retry under one it will honor instead of silently losing
|
|
// idempotence to a substitute.
|
|
func TestFrontRefusesInvalidRequestID(t *testing.T) {
|
|
f := fallbackFront(t, nil, true)
|
|
|
|
serve := func(token string) *httptest.ResponseRecorder {
|
|
r := httptest.NewRequest(http.MethodGet, PathPrefix+"a/d/known", nil)
|
|
r.Header.Set("X-Request-Id", token)
|
|
w := httptest.NewRecorder()
|
|
f.ServeHTTP(w, r)
|
|
return w
|
|
}
|
|
|
|
require.Equal(t, http.StatusBadRequest, serve(strings.Repeat("x", maxRequestIDLen+1)).Code)
|
|
// same route, acceptable token: the request reaches dispatch (and fails
|
|
// there for want of a stream), so the 400 above came from the token alone
|
|
require.Equal(t, http.StatusServiceUnavailable, serve("f81d4fae-7dec").Code)
|
|
}
|
|
|
|
// the preamble is built once, so a retry must not hand the worker a budget
|
|
// already spent.
|
|
func TestRefreshTimeoutTracksRemainingBudget(t *testing.T) {
|
|
newAttempt := func(ctx context.Context) *attempt {
|
|
a := &attempt{req: httptest.NewRequest(http.MethodGet, "/x", nil).WithContext(ctx)}
|
|
a.preamble = a.newPreamble()
|
|
return a
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
|
|
defer cancel()
|
|
a := newAttempt(ctx)
|
|
a.refreshTimeout()
|
|
first := a.preamble.GetTimeoutMs()
|
|
require.NotZero(t, first)
|
|
|
|
time.Sleep(25 * time.Millisecond)
|
|
a.refreshTimeout()
|
|
require.Less(t, a.preamble.GetTimeoutMs(), first)
|
|
|
|
// no deadline: 0 is the proto's "no deadline"
|
|
none := newAttempt(context.Background())
|
|
none.refreshTimeout()
|
|
require.Zero(t, none.preamble.GetTimeoutMs())
|
|
|
|
// an expired budget is not "no deadline"
|
|
expired, cancelExpired := context.WithDeadline(context.Background(), time.Now().Add(-time.Second))
|
|
defer cancelExpired()
|
|
spent := newAttempt(expired)
|
|
spent.refreshTimeout()
|
|
require.EqualValues(t, 1, spent.preamble.GetTimeoutMs())
|
|
}
|
|
|
|
// accessFront registers one worker serving a public and a non-public route, and
|
|
// resolves every request to the given access.
|
|
func accessFront(t *testing.T, level AccessLevel, fb Fallback) *Front {
|
|
g, s := NewRegistry(), NewScope(logger.GetLogger())
|
|
m, err := ParseManifest([]*livekit.AgentHttp_AgentEndpoint{
|
|
{Path: "/pub", Methods: []string{"GET"}, Public: true},
|
|
{Path: "/private", Methods: []string{"GET"}, Public: false},
|
|
})
|
|
require.NoError(t, err)
|
|
g.Register(s, NewRegistration(RegistrationParams{WorkerID: "w1", Manifest: m, Session: &fakeSession{}}))
|
|
|
|
return NewFront(FrontParams{
|
|
ResolveAccess: resolveTo(Access{Scope: s, Fallback: fb, Level: level}),
|
|
Logger: logger.GetLogger(),
|
|
})
|
|
}
|
|
|
|
// fakeSession opens no stream, so a request that clears authorization reaches 503.
|
|
func TestFrontPrivateRouteAccessMapping(t *testing.T) {
|
|
anonymous, credentialed, granted := AccessNone, AccessCredentialed, AccessGranted
|
|
|
|
t.Run("anonymous is challenged", func(t *testing.T) {
|
|
w := serveFront(accessFront(t, anonymous, nil), "/private")
|
|
require.Equal(t, http.StatusUnauthorized, w.Code)
|
|
require.Equal(t, "Bearer", w.Header().Get("WWW-Authenticate"))
|
|
})
|
|
t.Run("credential without the grant is refused, not challenged", func(t *testing.T) {
|
|
w := serveFront(accessFront(t, credentialed, nil), "/private")
|
|
require.Equal(t, http.StatusForbidden, w.Code)
|
|
require.Empty(t, w.Header().Get("WWW-Authenticate"))
|
|
})
|
|
t.Run("granted passes authorization", func(t *testing.T) {
|
|
require.Equal(t, http.StatusServiceUnavailable, serveFront(accessFront(t, granted, nil), "/private").Code)
|
|
})
|
|
t.Run("a credential without the grant keeps public access", func(t *testing.T) {
|
|
require.Equal(t, http.StatusServiceUnavailable, serveFront(accessFront(t, credentialed, nil), "/pub").Code)
|
|
})
|
|
t.Run("anonymous keeps public access", func(t *testing.T) {
|
|
require.Equal(t, http.StatusServiceUnavailable, serveFront(accessFront(t, anonymous, nil), "/pub").Code)
|
|
})
|
|
}
|
|
|
|
// the slash-normalized form of a private route is still private.
|
|
func TestFrontDeniedAppliesToNormalizedPath(t *testing.T) {
|
|
f := accessFront(t, AccessCredentialed, nil)
|
|
require.Equal(t, http.StatusForbidden, serveFront(f, "/private/").Code)
|
|
}
|
|
|
|
// another node's worker may declare the same path public.
|
|
func TestFrontDeniedStillRelays(t *testing.T) {
|
|
got := AccessLevel(-1)
|
|
f := accessFront(t, AccessCredentialed, func(w http.ResponseWriter, _ *http.Request, level AccessLevel) bool {
|
|
got = level
|
|
w.WriteHeader(http.StatusTeapot)
|
|
return true
|
|
})
|
|
|
|
require.Equal(t, http.StatusTeapot, serveFront(f, "/private").Code)
|
|
require.Equal(t, AccessCredentialed, got)
|
|
}
|
|
|
|
// the split runs before decoding, so a name or route param may carry any byte
|
|
// a percent-encoded segment can hold.
|
|
func TestSplitEndpointPath(t *testing.T) {
|
|
cases := []struct {
|
|
name string
|
|
target string
|
|
agentName string
|
|
deployment string
|
|
path string
|
|
escPath string
|
|
err error
|
|
}{
|
|
{name: "plain", target: "/agents/a/d/x", agentName: "a", deployment: "d", path: "/x", escPath: "/x"},
|
|
{name: "no tail", target: "/agents/a/d", agentName: "a", deployment: "d", path: "/", escPath: "/"},
|
|
{name: "spaces", target: "/agents/LODHA%20Vayam%20Agent/d/x", agentName: "LODHA Vayam Agent", deployment: "d", path: "/x", escPath: "/x"},
|
|
{name: "colon and star", target: "/agents/prod%3A%2A/d/x", agentName: "prod:*", deployment: "d", path: "/x", escPath: "/x"},
|
|
{name: "at sign", target: "/agents/charlie%40v1.42.0/d/x", agentName: "charlie@v1.42.0", deployment: "d", path: "/x", escPath: "/x"},
|
|
{name: "brackets", target: "/agents/Nathan%20%5BElara%5D/d/x", agentName: "Nathan [Elara]", deployment: "d", path: "/x", escPath: "/x"},
|
|
{name: "slash in name", target: "/agents/a%2Fb/d/x", agentName: "a/b", deployment: "d", path: "/x", escPath: "/x"},
|
|
{name: "non ascii raw", target: "/agents/agent-ü/d/x", agentName: "agent-ü", deployment: "d", path: "/x", escPath: "/x"},
|
|
{name: "non ascii encoded", target: "/agents/agent-%C3%BC/d/x", agentName: "agent-ü", deployment: "d", path: "/x", escPath: "/x"},
|
|
{name: "past the old 64 byte cap", target: "/agents/" + strings.Repeat("n", 77) + "/d/x", agentName: strings.Repeat("n", 77), deployment: "d", path: "/x", escPath: "/x"},
|
|
{name: "deployment encoded", target: "/agents/a/prod%20us/x", agentName: "a", deployment: "prod us", path: "/x", escPath: "/x"},
|
|
|
|
// escPath keeps the client's encoding; path is what the manifest matches
|
|
{name: "encoded tail", target: "/agents/a/d/files/a%2Fb", agentName: "a", deployment: "d", path: "/files/a/b", escPath: "/files/a%2Fb"},
|
|
{name: "escaped percent in tail", target: "/agents/a/d/%2541", agentName: "a", deployment: "d", path: "/%41", escPath: "/%2541"},
|
|
|
|
// "_" addresses the unnamed agent in either form
|
|
{name: "bare underscore is unnamed", target: "/agents/_/d/x", agentName: "", deployment: "d", path: "/x", escPath: "/x"},
|
|
{name: "encoded underscore is unnamed", target: "/agents/%5F/d/x", agentName: "", deployment: "d", path: "/x", escPath: "/x"},
|
|
{name: "double encoded underscore is a name", target: "/agents/%255F/d/x", agentName: "%5F", deployment: "d", path: "/x", escPath: "/x"},
|
|
|
|
{name: "not an endpoint path", target: "/other/x", err: errNotEndpointPath},
|
|
{name: "no deployment segment", target: "/agents/a", err: errNotEndpointPath},
|
|
{name: "empty name", target: "/agents//d/x", err: errNotEndpointPath},
|
|
{name: "empty deployment", target: "/agents/a//x", err: errNotEndpointPath},
|
|
}
|
|
|
|
for _, c := range cases {
|
|
t.Run(c.name, func(t *testing.T) {
|
|
u, err := url.ParseRequestURI(c.target)
|
|
require.NoError(t, err)
|
|
|
|
ep, err := splitEndpointPath(u)
|
|
if c.err != nil {
|
|
require.ErrorIs(t, err, c.err)
|
|
return
|
|
}
|
|
require.NoError(t, err)
|
|
require.Equal(t, c.agentName, ep.agentName)
|
|
require.Equal(t, c.deployment, ep.deployment)
|
|
require.Equal(t, c.path, ep.path)
|
|
require.Equal(t, c.escPath, ep.escPath)
|
|
})
|
|
}
|
|
}
|
|
|
|
// a trailing %2F decodes to a slash without being a separator, so the
|
|
// trailing-slash alternate does not apply to it.
|
|
func TestFrontSlashAlternateIgnoresEncodedSlash(t *testing.T) {
|
|
f := fallbackFront(t, nil, true)
|
|
|
|
// 503 is dispatch reached: the fake session opens no stream
|
|
require.Equal(t, http.StatusServiceUnavailable, serveFront(f, "/known/").Code)
|
|
require.Equal(t, http.StatusNotFound, serveFront(f, "/known%2F").Code)
|
|
}
|
|
|
|
func TestIsReservedAgentName(t *testing.T) {
|
|
for _, n := range []string{"_", ".", ".."} {
|
|
require.True(t, IsReservedAgentName(n), n)
|
|
}
|
|
for _, n := range []string{"", "a", "_x", "x_", "...", "LODHA Vayam Agent", "%5F"} {
|
|
require.False(t, IsReservedAgentName(n), n)
|
|
}
|
|
}
|
|
|
|
// matching runs before anything sizes the request head, so the split is where
|
|
// an over-long path is refused.
|
|
func TestSplitEndpointPathLength(t *testing.T) {
|
|
long := PathPrefix + "a/d/" + strings.Repeat("x", MaxPathLength)
|
|
_, err := splitEndpointPath(&url.URL{Path: long, RawPath: long})
|
|
require.ErrorIs(t, err, errPathTooLong)
|
|
|
|
f := fallbackFront(t, nil, true)
|
|
require.Equal(t, http.StatusRequestURITooLong, serveFront(f, "/"+strings.Repeat("x", MaxPathLength)).Code)
|
|
|
|
ok := PathPrefix + "a/d/" + strings.Repeat("x", MaxPathLength-4)
|
|
_, err = splitEndpointPath(&url.URL{Path: ok, RawPath: ok})
|
|
require.NoError(t, err)
|
|
}
|
|
|
|
// budgetFront registers a worker whose routes are ambiguous enough that the
|
|
// matcher gives up deciding.
|
|
func budgetFront(t *testing.T, level AccessLevel) *Front {
|
|
g, s := NewRegistry(), NewScope(logger.GetLogger())
|
|
m, err := ParseManifest([]*livekit.AgentHttp_AgentEndpoint{
|
|
{Path: "/{a}{b}{c}{d}x", Methods: []string{"GET"}, Public: true},
|
|
})
|
|
require.NoError(t, err)
|
|
require.Len(t, m.Ambiguous(), 1)
|
|
g.Register(s, NewRegistration(RegistrationParams{
|
|
WorkerID: "w1", Manifest: m, Session: &fakeSession{},
|
|
}))
|
|
return NewFront(FrontParams{
|
|
ResolveAccess: resolveTo(Access{Scope: s, Level: level}),
|
|
Logger: logger.GetLogger(),
|
|
})
|
|
}
|
|
|
|
// A table too ambiguous to decide forwards to the worker. No route is decided,
|
|
// so its Public flag is unknown and only a grant clears the request.
|
|
func TestFrontOverBudgetForwardsWithAGrant(t *testing.T) {
|
|
long := "/" + strings.Repeat("a", 512)
|
|
|
|
f := budgetFront(t, AccessGranted)
|
|
// 503 is dispatch reached: the fake session opens no stream
|
|
require.Equal(t, http.StatusServiceUnavailable, serveFront(f, long).Code)
|
|
|
|
f = budgetFront(t, AccessCredentialed)
|
|
require.Equal(t, http.StatusForbidden, serveFront(f, long).Code)
|
|
|
|
f = budgetFront(t, AccessNone)
|
|
require.Equal(t, http.StatusUnauthorized, serveFront(f, long).Code)
|
|
|
|
// a path the same table decides normally is unaffected
|
|
f = budgetFront(t, AccessNone)
|
|
require.Equal(t, http.StatusNotFound, serveFront(f, "/ab").Code)
|
|
}
|