fix(jellycompat): harden autoscan auth and batch scan enqueue
- Reject nil API keys and bound last-used update with a 5s timeout - Stop leaking internal queue errors in autoscan responses - Batch scan enqueues via new CreateBatch and reuse folder list across path resolves
This commit is contained in:
@@ -5,6 +5,7 @@ import (
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/Silo-Server/silo-server/internal/models"
|
||||
)
|
||||
@@ -83,7 +84,7 @@ func (a *AdminAPIKeyAuthenticator) authenticate(r *http.Request) adminAPIKeyAuth
|
||||
return adminAPIKeyAuthResult{ctx: r.Context(), status: http.StatusUnauthorized}
|
||||
}
|
||||
apiKey, err := a.keys.GetByKey(r.Context(), token)
|
||||
if err != nil {
|
||||
if err != nil || apiKey == nil {
|
||||
return adminAPIKeyAuthResult{ctx: r.Context(), status: http.StatusUnauthorized}
|
||||
}
|
||||
user, err := a.users.GetByID(r.Context(), apiKey.UserID)
|
||||
@@ -94,7 +95,9 @@ func (a *AdminAPIKeyAuthenticator) authenticate(r *http.Request) adminAPIKeyAuth
|
||||
return adminAPIKeyAuthResult{ctx: context.WithValue(r.Context(), adminAPIKeyKey, false), status: http.StatusForbidden}
|
||||
}
|
||||
go func(id int64) {
|
||||
if err := a.keys.UpdateLastUsed(context.Background(), id); err != nil {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
if err := a.keys.UpdateLastUsed(ctx, id); err != nil {
|
||||
slog.Debug("jellycompat api key last-used update failed", "id", id, "error", err)
|
||||
}
|
||||
}(apiKey.ID)
|
||||
|
||||
@@ -158,18 +158,78 @@ func TestRequireAdminAPIKey_RejectsNonAdminKey(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestRequireAdminAPIKey_RejectsNilAPIKey(t *testing.T) {
|
||||
authn := NewAdminAPIKeyAuthenticator(
|
||||
&fakeAPIKeyValidator{returnNilWithoutError: true},
|
||||
&fakeAPIKeyUserLoader{user: &models.User{ID: 2, Role: "admin", Enabled: true}},
|
||||
)
|
||||
req := httptest.NewRequest("GET", "/Library/VirtualFolders", nil)
|
||||
req.Header.Set("X-Emby-Token", "sa_test")
|
||||
rec := httptest.NewRecorder()
|
||||
|
||||
authn.RequireAdminAPIKey(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
t.Fatal("handler should not run")
|
||||
})).ServeHTTP(rec, req)
|
||||
|
||||
if rec.Code != http.StatusUnauthorized {
|
||||
t.Fatalf("expected 401, got %d: %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestRequireAdminAPIKey_LastUsedUpdateHasDeadline(t *testing.T) {
|
||||
called := make(chan bool, 1)
|
||||
authn := NewAdminAPIKeyAuthenticator(
|
||||
&fakeAPIKeyValidator{
|
||||
key: &models.APIKey{ID: 1, UserID: 2, Key: "sa_test"},
|
||||
update: func(ctx context.Context, _ int64) error {
|
||||
_, ok := ctx.Deadline()
|
||||
called <- ok
|
||||
return nil
|
||||
},
|
||||
},
|
||||
&fakeAPIKeyUserLoader{user: &models.User{ID: 2, Role: "admin", Enabled: true}},
|
||||
)
|
||||
req := httptest.NewRequest("GET", "/Library/VirtualFolders", nil)
|
||||
req.Header.Set("X-Emby-Token", "sa_test")
|
||||
rec := httptest.NewRecorder()
|
||||
|
||||
authn.RequireAdminAPIKey(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
})).ServeHTTP(rec, req)
|
||||
|
||||
if rec.Code != http.StatusNoContent {
|
||||
t.Fatalf("expected 204, got %d: %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
select {
|
||||
case ok := <-called:
|
||||
if !ok {
|
||||
t.Fatal("expected last-used update context to have a deadline")
|
||||
}
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("timed out waiting for last-used update")
|
||||
}
|
||||
}
|
||||
|
||||
type fakeAPIKeyValidator struct {
|
||||
key *models.APIKey
|
||||
key *models.APIKey
|
||||
returnNilWithoutError bool
|
||||
update func(context.Context, int64) error
|
||||
}
|
||||
|
||||
func (f *fakeAPIKeyValidator) GetByKey(_ context.Context, key string) (*models.APIKey, error) {
|
||||
if f.returnNilWithoutError {
|
||||
return nil, nil
|
||||
}
|
||||
if f.key != nil && f.key.Key == key {
|
||||
return f.key, nil
|
||||
}
|
||||
return nil, auth.ErrAPIKeyNotFound
|
||||
}
|
||||
|
||||
func (f *fakeAPIKeyValidator) UpdateLastUsed(context.Context, int64) error {
|
||||
func (f *fakeAPIKeyValidator) UpdateLastUsed(ctx context.Context, id int64) error {
|
||||
if f.update != nil {
|
||||
return f.update(ctx, id)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -4,7 +4,6 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"strings"
|
||||
@@ -142,5 +141,5 @@ func writeScanTriggerError(w http.ResponseWriter, err error) {
|
||||
return
|
||||
}
|
||||
slog.Error("jellycompat autoscan: scan update failed", "error", err)
|
||||
writeError(w, http.StatusInternalServerError, "InternalServerError", fmt.Sprintf("Failed to process scan update: %v", err))
|
||||
writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to process scan update")
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
@@ -13,6 +14,7 @@ import (
|
||||
|
||||
"github.com/Silo-Server/silo-server/internal/catalog"
|
||||
"github.com/Silo-Server/silo-server/internal/models"
|
||||
"github.com/Silo-Server/silo-server/internal/scantrigger"
|
||||
)
|
||||
|
||||
type fakeAutoscanFolders struct {
|
||||
@@ -33,7 +35,9 @@ func (f *fakeAutoscanFolders) List(context.Context) ([]*models.MediaFolder, erro
|
||||
}
|
||||
|
||||
type fakeAutoscanQueue struct {
|
||||
calls []queuedScan
|
||||
calls []queuedScan
|
||||
batches [][]scantrigger.Target
|
||||
batchErr error
|
||||
}
|
||||
|
||||
type queuedScan struct {
|
||||
@@ -48,6 +52,23 @@ func (q *fakeAutoscanQueue) EnqueueScan(_ context.Context, folderID int, mode, p
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func (q *fakeAutoscanQueue) EnqueueScans(_ context.Context, targets []scantrigger.Target) error {
|
||||
copied := append([]scantrigger.Target(nil), targets...)
|
||||
q.batches = append(q.batches, copied)
|
||||
if q.batchErr != nil {
|
||||
return q.batchErr
|
||||
}
|
||||
for _, target := range targets {
|
||||
q.calls = append(q.calls, queuedScan{
|
||||
libraryID: target.LibraryID,
|
||||
mode: target.Mode,
|
||||
path: target.Path,
|
||||
trigger: target.Trigger,
|
||||
})
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestAutoscanVirtualFoldersIncludesEnabledLocationsForAdminKey(t *testing.T) {
|
||||
enabledRoot := t.TempDir()
|
||||
disabledRoot := t.TempDir()
|
||||
@@ -104,6 +125,9 @@ func TestAutoscanMediaUpdatedEnqueuesResolvedPath(t *testing.T) {
|
||||
if len(queue.calls) != 1 {
|
||||
t.Fatalf("expected one queued scan, got %d", len(queue.calls))
|
||||
}
|
||||
if len(queue.batches) != 1 {
|
||||
t.Fatalf("expected one batch enqueue, got %d", len(queue.batches))
|
||||
}
|
||||
if queue.calls[0].libraryID != 3 || queue.calls[0].mode != "file" || queue.calls[0].path != filePath || queue.calls[0].trigger != "jellyfin_autoscan" {
|
||||
t.Fatalf("unexpected queued scan: %#v", queue.calls[0])
|
||||
}
|
||||
@@ -144,3 +168,32 @@ func TestAutoscanMediaUpdatedAllOrFail(t *testing.T) {
|
||||
t.Fatalf("expected no partial enqueue, got %#v", queue.calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAutoscanMediaUpdatedHidesInternalQueueError(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
filePath := filepath.Join(root, "Movie.mkv")
|
||||
if err := os.WriteFile(filePath, []byte("test"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
queue := &fakeAutoscanQueue{batchErr: errors.New("database password leaked")}
|
||||
handler := NewAutoscanHandler(&fakeAutoscanFolders{folders: []*models.MediaFolder{{
|
||||
ID: 5,
|
||||
Name: "Movies",
|
||||
Type: "movie",
|
||||
Enabled: true,
|
||||
Paths: []string{root},
|
||||
}}}, queue, NewResourceIDCodec(), nil)
|
||||
|
||||
body := []byte(`{"Updates":[{"path":` + strconv.Quote(filePath) + `,"updateType":"Modified"}]}`)
|
||||
req := httptest.NewRequest(http.MethodPost, "/Library/Media/Updated", bytes.NewReader(body))
|
||||
rec := httptest.NewRecorder()
|
||||
|
||||
handler.HandleMediaUpdated(rec, req)
|
||||
|
||||
if rec.Code != http.StatusInternalServerError {
|
||||
t.Fatalf("expected 500, got %d: %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
if bytes.Contains(rec.Body.Bytes(), []byte("database password leaked")) {
|
||||
t.Fatalf("response leaked internal error: %s", rec.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -121,6 +121,68 @@ func (r *Repository) Create(ctx context.Context, input CreateInput) (*models.Sca
|
||||
return nil, false, fmt.Errorf("create scan run: %w", err)
|
||||
}
|
||||
|
||||
func (r *Repository) CreateBatch(ctx context.Context, inputs []CreateInput) ([]*models.ScanRun, []bool, error) {
|
||||
if len(inputs) == 0 {
|
||||
return []*models.ScanRun{}, []bool{}, nil
|
||||
}
|
||||
tx, err := r.pool.BeginTx(ctx, pgx.TxOptions{})
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("begin scan run batch: %w", err)
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
|
||||
runs := make([]*models.ScanRun, 0, len(inputs))
|
||||
created := make([]bool, 0, len(inputs))
|
||||
for _, input := range inputs {
|
||||
run, err := scanRunRow(tx.QueryRow(ctx, `
|
||||
INSERT INTO scan_runs (
|
||||
id, media_folder_id, mode, path, trigger, status
|
||||
) VALUES ($1, $2, $3, $4, $5, $6)
|
||||
ON CONFLICT DO NOTHING
|
||||
RETURNING `+scanRunColumns,
|
||||
ulid.Make().String(),
|
||||
input.LibraryID,
|
||||
input.Mode,
|
||||
input.Path,
|
||||
input.Trigger,
|
||||
StatusAccepted,
|
||||
))
|
||||
if err == nil {
|
||||
runs = append(runs, run)
|
||||
created = append(created, true)
|
||||
continue
|
||||
}
|
||||
if !errors.Is(err, ErrScanRunNotFound) {
|
||||
return nil, nil, fmt.Errorf("create scan run: %w", err)
|
||||
}
|
||||
|
||||
existing, lookupErr := scanRunRow(tx.QueryRow(ctx, `
|
||||
SELECT `+scanRunColumns+`
|
||||
FROM scan_runs
|
||||
WHERE media_folder_id = $1
|
||||
AND mode = $2
|
||||
AND path = $3
|
||||
AND status = ANY($4)
|
||||
ORDER BY requested_at ASC
|
||||
LIMIT 1`,
|
||||
input.LibraryID,
|
||||
input.Mode,
|
||||
input.Path,
|
||||
[]string{StatusAccepted, StatusRunning},
|
||||
))
|
||||
if lookupErr != nil {
|
||||
return nil, nil, lookupErr
|
||||
}
|
||||
runs = append(runs, existing)
|
||||
created = append(created, false)
|
||||
}
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return nil, nil, fmt.Errorf("commit scan run batch: %w", err)
|
||||
}
|
||||
return runs, created, nil
|
||||
}
|
||||
|
||||
func (r *Repository) GetActiveByScope(ctx context.Context, libraryID int, mode, path string) (*models.ScanRun, error) {
|
||||
return scanRunRow(r.pool.QueryRow(ctx, `
|
||||
SELECT `+scanRunColumns+`
|
||||
|
||||
@@ -13,6 +13,7 @@ import (
|
||||
evt "github.com/Silo-Server/silo-server/internal/events"
|
||||
"github.com/Silo-Server/silo-server/internal/libraryingest"
|
||||
"github.com/Silo-Server/silo-server/internal/models"
|
||||
"github.com/Silo-Server/silo-server/internal/scantrigger"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -131,6 +132,31 @@ func (s *Service) EnqueueScan(ctx context.Context, folderID int, mode, path, tri
|
||||
return created, nil
|
||||
}
|
||||
|
||||
func (s *Service) EnqueueScans(ctx context.Context, targets []scantrigger.Target) error {
|
||||
if s == nil || s.repo == nil {
|
||||
return fmt.Errorf("scan queue is not configured")
|
||||
}
|
||||
inputs := make([]CreateInput, 0, len(targets))
|
||||
for _, target := range targets {
|
||||
inputs = append(inputs, CreateInput{
|
||||
LibraryID: target.LibraryID,
|
||||
Mode: target.Mode,
|
||||
Path: target.Path,
|
||||
Trigger: target.Trigger,
|
||||
})
|
||||
}
|
||||
runs, created, err := s.repo.CreateBatch(ctx, inputs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for i, run := range runs {
|
||||
if i < len(created) && created[i] {
|
||||
s.publish(ctx, "scan.accepted", run)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Service) CancelAcceptedByLibrary(ctx context.Context, libraryID int) (int, error) {
|
||||
if s == nil || s.repo == nil {
|
||||
return 0, nil
|
||||
|
||||
@@ -27,6 +27,7 @@ type FolderRepository interface {
|
||||
|
||||
type Queuer interface {
|
||||
EnqueueScan(ctx context.Context, folderID int, mode, path, trigger string) (bool, error)
|
||||
EnqueueScans(ctx context.Context, targets []Target) error
|
||||
}
|
||||
|
||||
type Request struct {
|
||||
@@ -63,8 +64,23 @@ func NewResolver(folders FolderRepository) *Resolver {
|
||||
|
||||
func (r *Resolver) ResolveAll(ctx context.Context, requests []Request) ([]Target, error) {
|
||||
targets := make([]Target, 0, len(requests))
|
||||
var pathFolders []*models.MediaFolder
|
||||
pathFoldersLoaded := false
|
||||
for _, req := range requests {
|
||||
target, err := r.Resolve(ctx, req)
|
||||
usePathFolders := req.LibraryID == nil && strings.TrimSpace(req.Path) != ""
|
||||
if usePathFolders && !pathFoldersLoaded {
|
||||
if r == nil || r.folders == nil {
|
||||
return nil, &RequestError{Status: http.StatusServiceUnavailable, Code: "unavailable", Message: "Scanner not available"}
|
||||
}
|
||||
folders, listErr := r.folders.List(ctx)
|
||||
if listErr != nil {
|
||||
return nil, fmt.Errorf("listing libraries for scan: %w", listErr)
|
||||
}
|
||||
pathFolders = folders
|
||||
pathFoldersLoaded = true
|
||||
}
|
||||
|
||||
target, err := r.resolve(ctx, req, pathFolders, usePathFolders)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -74,6 +90,10 @@ func (r *Resolver) ResolveAll(ctx context.Context, requests []Request) ([]Target
|
||||
}
|
||||
|
||||
func (r *Resolver) Resolve(ctx context.Context, req Request) (*Target, error) {
|
||||
return r.resolve(ctx, req, nil, false)
|
||||
}
|
||||
|
||||
func (r *Resolver) resolve(ctx context.Context, req Request, pathFolders []*models.MediaFolder, usePathFolders bool) (*Target, error) {
|
||||
if r == nil || r.folders == nil {
|
||||
return nil, &RequestError{Status: http.StatusServiceUnavailable, Code: "unavailable", Message: "Scanner not available"}
|
||||
}
|
||||
@@ -115,9 +135,13 @@ func (r *Resolver) Resolve(ctx context.Context, req Request) (*Target, error) {
|
||||
return nil, &RequestError{Status: http.StatusBadRequest, Code: "bad_request", Message: "Path does not belong to the specified library"}
|
||||
}
|
||||
} else {
|
||||
folders, listErr := r.folders.List(ctx)
|
||||
if listErr != nil {
|
||||
return nil, fmt.Errorf("listing libraries for scan: %w", listErr)
|
||||
folders := pathFolders
|
||||
if !usePathFolders {
|
||||
var listErr error
|
||||
folders, listErr = r.folders.List(ctx)
|
||||
if listErr != nil {
|
||||
return nil, fmt.Errorf("listing libraries for scan: %w", listErr)
|
||||
}
|
||||
}
|
||||
folder, matchedRoot, err = MatchFolderForPath(cleanPath, folders)
|
||||
if err != nil {
|
||||
@@ -150,10 +174,8 @@ func EnqueueAll(ctx context.Context, queue Queuer, targets []Target) error {
|
||||
if queue == nil {
|
||||
return &RequestError{Status: http.StatusServiceUnavailable, Code: "unavailable", Message: "Scanner not available"}
|
||||
}
|
||||
for _, target := range targets {
|
||||
if _, err := queue.EnqueueScan(ctx, target.LibraryID, target.Mode, target.Path, target.Trigger); err != nil {
|
||||
return fmt.Errorf("queueing library scan: %w", err)
|
||||
}
|
||||
if err := queue.EnqueueScans(ctx, targets); err != nil {
|
||||
return fmt.Errorf("queueing library scans: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -13,7 +13,8 @@ import (
|
||||
)
|
||||
|
||||
type fakeFolderRepo struct {
|
||||
folders []*models.MediaFolder
|
||||
folders []*models.MediaFolder
|
||||
listCalls int
|
||||
}
|
||||
|
||||
func (r *fakeFolderRepo) GetByID(_ context.Context, id int) (*models.MediaFolder, error) {
|
||||
@@ -26,6 +27,7 @@ func (r *fakeFolderRepo) GetByID(_ context.Context, id int) (*models.MediaFolder
|
||||
}
|
||||
|
||||
func (r *fakeFolderRepo) List(context.Context) ([]*models.MediaFolder, error) {
|
||||
r.listCalls++
|
||||
return r.folders, nil
|
||||
}
|
||||
|
||||
@@ -135,3 +137,73 @@ func TestResolveAllIsAllOrFail(t *testing.T) {
|
||||
t.Fatalf("unexpected error message: %q", reqErr.Message)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveAllReusesPathOnlyLibraryList(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
first := filepath.Join(root, "First.mkv")
|
||||
second := filepath.Join(root, "Second.mkv")
|
||||
for _, path := range []string{first, second} {
|
||||
if err := os.WriteFile(path, []byte("test"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
repo := &fakeFolderRepo{folders: []*models.MediaFolder{{
|
||||
ID: 12,
|
||||
Name: "Movies",
|
||||
Enabled: true,
|
||||
Paths: []string{root},
|
||||
}}}
|
||||
|
||||
targets, err := NewResolver(repo).ResolveAll(context.Background(), []Request{
|
||||
{Path: first},
|
||||
{Path: second},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("ResolveAll returned error: %v", err)
|
||||
}
|
||||
if len(targets) != 2 {
|
||||
t.Fatalf("expected two targets, got %d", len(targets))
|
||||
}
|
||||
if repo.listCalls != 1 {
|
||||
t.Fatalf("expected one folder list lookup, got %d", repo.listCalls)
|
||||
}
|
||||
}
|
||||
|
||||
type fakeQueue struct {
|
||||
calls []Target
|
||||
batches [][]Target
|
||||
batchErr error
|
||||
}
|
||||
|
||||
func (q *fakeQueue) EnqueueScan(_ context.Context, folderID int, mode, path, trigger string) (bool, error) {
|
||||
q.calls = append(q.calls, Target{LibraryID: folderID, Mode: mode, Path: path, Trigger: trigger})
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func (q *fakeQueue) EnqueueScans(_ context.Context, targets []Target) error {
|
||||
copied := append([]Target(nil), targets...)
|
||||
q.batches = append(q.batches, copied)
|
||||
if q.batchErr != nil {
|
||||
return q.batchErr
|
||||
}
|
||||
q.calls = append(q.calls, targets...)
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestEnqueueAllUsesBatchQueue(t *testing.T) {
|
||||
queue := &fakeQueue{}
|
||||
targets := []Target{
|
||||
{LibraryID: 1, Mode: ModeFile, Path: "/media/one.mkv", Trigger: "autoscan"},
|
||||
{LibraryID: 1, Mode: ModeFile, Path: "/media/two.mkv", Trigger: "autoscan"},
|
||||
}
|
||||
|
||||
if err := EnqueueAll(context.Background(), queue, targets); err != nil {
|
||||
t.Fatalf("EnqueueAll returned error: %v", err)
|
||||
}
|
||||
if len(queue.batches) != 1 {
|
||||
t.Fatalf("expected one batch enqueue, got %d", len(queue.batches))
|
||||
}
|
||||
if len(queue.calls) != 2 {
|
||||
t.Fatalf("expected two queued calls from batch, got %d", len(queue.calls))
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user