From 10b35642a528545f7b08ccec4902a453cbaead23 Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Sun, 15 Feb 2026 15:48:40 -0600 Subject: [PATCH] fix scanner bugs and firehose bugs --- pkg/appview/db/queries.go | 12 ++++----- pkg/appview/server.go | 7 +++-- pkg/hold/pds/auth.go | 7 ++++- pkg/hold/pds/auth_test.go | 10 ++++---- pkg/hold/pds/events.go | 25 ++++++++++++++++++ pkg/hold/pds/scan_broadcaster.go | 41 +++++++++++++++++++++++++----- pkg/hold/pds/xrpc.go | 6 ++++- scanner/internal/client/hold.go | 15 ++++++++--- scanner/internal/scan/extractor.go | 18 +++++++++---- scanner/internal/scan/worker.go | 2 +- 10 files changed, 112 insertions(+), 31 deletions(-) diff --git a/pkg/appview/db/queries.go b/pkg/appview/db/queries.go index a2a2ae4..9038621 100644 --- a/pkg/appview/db/queries.go +++ b/pkg/appview/db/queries.go @@ -105,7 +105,7 @@ func SearchRepositories(db DBTX, query string, limit, offset int, currentUserDID COALESCE(m.artifact_type, 'container-image'), COALESCE((SELECT tag FROM tags WHERE did = m.did AND repository = m.repository ORDER BY created_at DESC LIMIT 1), ''), COALESCE(m.digest, ''), - COALESCE(rs.last_push, m.created_at), + MAX(rs.last_push, m.created_at), COALESCE(rp.avatar_cid, '') FROM matching_repos mr JOIN manifests m ON mr.latest_id = m.id @@ -113,7 +113,7 @@ func SearchRepositories(db DBTX, query string, limit, offset int, currentUserDID JOIN repo_stats ON m.did = repo_stats.did AND m.repository = repo_stats.repository LEFT JOIN repository_stats rs ON m.did = rs.did AND m.repository = rs.repository LEFT JOIN repo_pages rp ON m.did = rp.did AND m.repository = rp.repository - ORDER BY COALESCE(rs.last_push, m.created_at) DESC + ORDER BY MAX(rs.last_push, m.created_at) DESC LIMIT ? OFFSET ? ` @@ -1743,7 +1743,7 @@ func GetRepoCards(db DBTX, limit int, currentUserDID string, sortOrder RepoCardS var orderBy string switch sortOrder { case SortByLastUpdate: - orderBy = "COALESCE(rs.last_push, m.created_at) DESC" + orderBy = "MAX(rs.last_push, m.created_at) DESC" default: // SortByScore orderBy = "(COALESCE(rs.pull_count, 0) + COALESCE((SELECT COUNT(*) FROM stars WHERE owner_did = m.did AND repository = m.repository), 0) * 10) DESC, m.created_at DESC" } @@ -1768,7 +1768,7 @@ func GetRepoCards(db DBTX, limit int, currentUserDID string, sortOrder RepoCardS COALESCE(m.artifact_type, 'container-image'), COALESCE((SELECT tag FROM tags WHERE did = m.did AND repository = m.repository ORDER BY created_at DESC LIMIT 1), ''), COALESCE(m.digest, ''), - COALESCE(rs.last_push, m.created_at), + MAX(rs.last_push, m.created_at), COALESCE(rp.avatar_cid, '') FROM latest_manifests lm JOIN manifests m ON lm.latest_id = m.id @@ -1841,14 +1841,14 @@ func GetUserRepoCards(db DBTX, userDID string, currentUserDID string) ([]RepoCar COALESCE(m.artifact_type, 'container-image'), COALESCE((SELECT tag FROM tags WHERE did = m.did AND repository = m.repository ORDER BY created_at DESC LIMIT 1), ''), COALESCE(m.digest, ''), - COALESCE(rs.last_push, m.created_at), + MAX(rs.last_push, m.created_at), COALESCE(rp.avatar_cid, '') FROM latest_manifests lm JOIN manifests m ON lm.latest_id = m.id JOIN users u ON m.did = u.did LEFT JOIN repository_stats rs ON m.did = rs.did AND m.repository = rs.repository LEFT JOIN repo_pages rp ON m.did = rp.did AND m.repository = rp.repository - ORDER BY COALESCE(rs.last_push, m.created_at) DESC + ORDER BY MAX(rs.last_push, m.created_at) DESC ` rows, err := db.Query(query, userDID, currentUserDID) diff --git a/pkg/appview/server.go b/pkg/appview/server.go index 8e7542e..7619363 100644 --- a/pkg/appview/server.go +++ b/pkg/appview/server.go @@ -647,8 +647,11 @@ func DomainRoutingMiddleware(registryDomains []string, uiBaseURL string) func(ht next.ServeHTTP(w, r) case regDomains[host]: - // Registry domain: allow /v2/*, redirect everything else - if isV2 { + // Registry domain: allow /v2/*, /auth/token, /auth/device/*, redirect everything else + // Auth endpoints must be served directly to avoid 307 redirects that strip + // the Authorization header on cross-host redirects (Go http.Client behavior). + isAuth := path == "/auth/token" || strings.HasPrefix(path, "/auth/device/") + if isV2 || isAuth { next.ServeHTTP(w, r) return } diff --git a/pkg/hold/pds/auth.go b/pkg/hold/pds/auth.go index 365574d..93f4530 100644 --- a/pkg/hold/pds/auth.go +++ b/pkg/hold/pds/auth.go @@ -355,7 +355,8 @@ func ValidateBlobWriteAccess(r *http.Request, pds *HoldPDS, httpClient HTTPClien // If captain.public = false: Requires valid DPoP + OAuth and (captain OR crew with blob:read or blob:write permission). // Note: blob:write implicitly grants blob:read access. // The httpClient parameter is optional and defaults to http.DefaultClient if nil. -func ValidateBlobReadAccess(r *http.Request, pds *HoldPDS, httpClient HTTPClient) (*ValidatedUser, error) { +// If scannerSecret is non-empty, a Bearer token matching it grants full read access (for scanner blob fetches). +func ValidateBlobReadAccess(r *http.Request, pds *HoldPDS, httpClient HTTPClient, scannerSecret string) (*ValidatedUser, error) { // Get captain record to check public setting _, captain, err := pds.GetCaptainRecord(r.Context()) if err != nil { @@ -372,6 +373,10 @@ func ValidateBlobReadAccess(r *http.Request, pds *HoldPDS, httpClient HTTPClient var user *ValidatedUser if strings.HasPrefix(authHeader, "Bearer ") { + // Check if this is a scanner using the shared secret + if scannerSecret != "" && strings.TrimPrefix(authHeader, "Bearer ") == scannerSecret { + return &ValidatedUser{DID: "scanner"}, nil + } // Service token authentication (from AppView via getServiceAuth) user, err = ValidateServiceToken(r, pds.did, httpClient) if err != nil { diff --git a/pkg/hold/pds/auth_test.go b/pkg/hold/pds/auth_test.go index 75f75ec..822b093 100644 --- a/pkg/hold/pds/auth_test.go +++ b/pkg/hold/pds/auth_test.go @@ -724,7 +724,7 @@ func TestValidateBlobReadAccess_PublicHold(t *testing.T) { req := httptest.NewRequest(http.MethodGet, "/test", nil) // This should return nil (public access allowed) for public holds - user, err := ValidateBlobReadAccess(req, pds, nil) + user, err := ValidateBlobReadAccess(req, pds, nil, "") if err != nil { t.Errorf("Expected public access for public hold, got error: %v", err) } @@ -768,7 +768,7 @@ func TestValidateBlobReadAccess_PrivateHold(t *testing.T) { req := httptest.NewRequest(http.MethodGet, "/test", nil) // This should return error (auth required) for private holds - user, err := ValidateBlobReadAccess(req, pds, nil) + user, err := ValidateBlobReadAccess(req, pds, nil, "") if err == nil { t.Error("Expected error for private hold without auth") } @@ -816,7 +816,7 @@ func TestValidateBlobReadAccess_BlobWriteImpliesRead(t *testing.T) { } // This should SUCCEED because blob:write implies blob:read - user, err := ValidateBlobReadAccess(req, pds, mockClient) + user, err := ValidateBlobReadAccess(req, pds, mockClient, "") if err != nil { t.Errorf("Expected blob:write to grant read access, got error: %v", err) } @@ -846,7 +846,7 @@ func TestValidateBlobReadAccess_BlobWriteImpliesRead(t *testing.T) { t.Fatalf("Failed to add DPoP to request: %v", err) } - user, err := ValidateBlobReadAccess(req, pds, mockClient) + user, err := ValidateBlobReadAccess(req, pds, mockClient, "") if err != nil { t.Errorf("Expected blob:read to grant read access, got error: %v", err) } @@ -876,7 +876,7 @@ func TestValidateBlobReadAccess_BlobWriteImpliesRead(t *testing.T) { t.Fatalf("Failed to add DPoP to request: %v", err) } - _, err = ValidateBlobReadAccess(req, pds, mockClient) + _, err = ValidateBlobReadAccess(req, pds, mockClient, "") if err == nil { t.Error("Expected error for crew without read or write permission") } diff --git a/pkg/hold/pds/events.go b/pkg/hold/pds/events.go index 143a82d..9d7ebbd 100644 --- a/pkg/hold/pds/events.go +++ b/pkg/hold/pds/events.go @@ -415,6 +415,11 @@ func (b *EventBroadcaster) Subscribe(conn *websocket.Conn, cursor int64, userAge // else cursor == currentSeq: relay is caught up, just stream new events } + // Start read pump to handle pings/pongs and detect disconnects. + // gorilla/websocket requires an active reader to process control frames; + // without one, pings go unanswered and the relay times out the connection. + go b.readPump(sub) + // Start goroutine to handle sending events to this subscriber go b.handleSubscriber(sub) @@ -686,6 +691,26 @@ func (b *EventBroadcaster) backfillFromMemory(sub *Subscriber, cursor int64) { } } +// readPump reads from the WebSocket to process control frames (ping/pong/close). +// gorilla/websocket automatically responds to pings with pongs when there is an +// active reader. Without this, relays time out the connection. +func (b *EventBroadcaster) readPump(sub *Subscriber) { + defer func() { + b.Unsubscribe(sub) + sub.conn.Close() + }() + + for { + _, _, err := sub.conn.ReadMessage() + if err != nil { + if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) { + slog.Warn("Firehose subscriber disconnected", "remote", sub.conn.RemoteAddr(), "error", err) + } + return + } + } +} + // handleSubscriber handles sending events to a subscriber over WebSocket func (b *EventBroadcaster) handleSubscriber(sub *Subscriber) { defer func() { diff --git a/pkg/hold/pds/scan_broadcaster.go b/pkg/hold/pds/scan_broadcaster.go index 17cf219..d5f205a 100644 --- a/pkg/hold/pds/scan_broadcaster.go +++ b/pkg/hold/pds/scan_broadcaster.go @@ -94,6 +94,18 @@ func NewScanBroadcaster(holdDID, holdEndpoint, secret, dbPath string, s3svc *s3. return nil, fmt.Errorf("failed to ping scan jobs database: %w", err) } + // Set WAL mode and busy timeout (libsql PRAGMAs return rows) + var journalMode string + if err := db.QueryRow("PRAGMA journal_mode = WAL").Scan(&journalMode); err != nil { + db.Close() + return nil, fmt.Errorf("failed to set journal mode: %w", err) + } + var busyTimeout int + if err := db.QueryRow("PRAGMA busy_timeout = 5000").Scan(&busyTimeout); err != nil { + db.Close() + return nil, fmt.Errorf("failed to set busy_timeout: %w", err) + } + sb := &ScanBroadcaster{ subscribers: make([]*ScanSubscriber, 0), db: db, @@ -509,7 +521,9 @@ func (sb *ScanBroadcaster) handleError(sub *ScanSubscriber, msg ScannerMessage) "error", msg.Error) } -// drainPendingJobs sends pending/timed-out jobs to a newly connected scanner +// drainPendingJobs sends pending/timed-out jobs to a newly connected scanner. +// Collects all pending rows first, closes cursor, then assigns and dispatches +// to avoid holding a SELECT cursor open during UPDATEs (prevents SQLite BUSY). func (sb *ScanBroadcaster) drainPendingJobs(sub *ScanSubscriber, cursor int64) { rows, err := sb.db.Query(` SELECT seq, manifest_digest, repository, tag, user_did, user_handle, hold_did, hold_endpoint, tier, config_json, layers_json @@ -521,9 +535,8 @@ func (sb *ScanBroadcaster) drainPendingJobs(sub *ScanSubscriber, cursor int64) { slog.Error("Failed to drain pending scan jobs", "error", err) return } - defer rows.Close() - count := 0 + var jobs []*ScanJobEvent for rows.Next() { job := &ScanJobEvent{Type: "job"} var configJSON, layersJSON string @@ -540,8 +553,12 @@ func (sb *ScanBroadcaster) drainPendingJobs(sub *ScanSubscriber, cursor int64) { job.Config = json.RawMessage(configJSON) job.Layers = json.RawMessage(layersJSON) + jobs = append(jobs, job) + } + rows.Close() - // Assign and dispatch + count := 0 + for _, job := range jobs { _, err = sb.db.Exec(` UPDATE scan_jobs SET status = 'assigned', assigned_to = ?, assigned_at = ? WHERE seq = ? AND status = 'pending' @@ -578,7 +595,9 @@ func (sb *ScanBroadcaster) reDispatchLoop() { } } -// reDispatchTimedOut finds jobs that were assigned but not acked/completed within timeout +// reDispatchTimedOut finds jobs that were assigned but not acked/completed within timeout. +// Collects timed-out rows first, closes cursor, then resets and re-dispatches +// to avoid holding a SELECT cursor open during UPDATEs (prevents SQLite BUSY). func (sb *ScanBroadcaster) reDispatchTimedOut() { timeout := time.Now().Add(-sb.ackTimeout) @@ -592,8 +611,8 @@ func (sb *ScanBroadcaster) reDispatchTimedOut() { slog.Error("Failed to query timed-out scan jobs", "error", err) return } - defer rows.Close() + var jobs []*ScanJobEvent for rows.Next() { job := &ScanJobEvent{Type: "job"} var configJSON, layersJSON string @@ -609,8 +628,11 @@ func (sb *ScanBroadcaster) reDispatchTimedOut() { job.Config = json.RawMessage(configJSON) job.Layers = json.RawMessage(layersJSON) + jobs = append(jobs, job) + } + rows.Close() - // Reset to pending and re-dispatch + for _, job := range jobs { _, err = sb.db.Exec(` UPDATE scan_jobs SET status = 'pending', assigned_to = NULL, assigned_at = NULL WHERE seq = ? @@ -635,6 +657,11 @@ func (sb *ScanBroadcaster) Close() error { return nil } +// Secret returns the scanner shared secret for use in blob read authorization +func (sb *ScanBroadcaster) Secret() string { + return sb.secret +} + // ValidateScannerSecret checks if the provided secret matches func (sb *ScanBroadcaster) ValidateScannerSecret(secret string) bool { return sb.secret != "" && secret == sb.secret diff --git a/pkg/hold/pds/xrpc.go b/pkg/hold/pds/xrpc.go index c859fa8..49b11c8 100644 --- a/pkg/hold/pds/xrpc.go +++ b/pkg/hold/pds/xrpc.go @@ -1108,7 +1108,11 @@ func (h *XRPCHandler) handleGetOCIBlob(w http.ResponseWriter, r *http.Request, d // Validate blob read access (hold access control) // If captain.public = true, returns nil (public access allowed) // If captain.public = false, validates auth and checks for blob:read permission - _, err := ValidateBlobReadAccess(r, h.pds, h.httpClient) + scannerSecret := "" + if h.scanBroadcaster != nil { + scannerSecret = h.scanBroadcaster.Secret() + } + _, err := ValidateBlobReadAccess(r, h.pds, h.httpClient, scannerSecret) if err != nil { slog.Warn("OCI blob authorization failed", "error", err, "digest", digest) http.Error(w, fmt.Sprintf("authorization failed: %v", err), http.StatusForbidden) diff --git a/scanner/internal/client/hold.go b/scanner/internal/client/hold.go index fd36e77..a8becda 100644 --- a/scanner/internal/client/hold.go +++ b/scanner/internal/client/hold.go @@ -213,14 +213,23 @@ func (c *HoldClient) Close() { c.mu.Unlock() } -// GetBlobPresignedURL gets a presigned download URL from the hold service -func GetBlobPresignedURL(holdEndpoint, holdDID, digest string) (string, error) { +// GetBlobPresignedURL gets a presigned download URL from the hold service. +// If secret is non-empty, it is sent as a Bearer token for private hold access. +func GetBlobPresignedURL(holdEndpoint, holdDID, digest, secret string) (string, error) { reqURL := fmt.Sprintf("%s/xrpc/com.atproto.sync.getBlob?did=%s&cid=%s&method=GET", holdEndpoint, url.QueryEscape(holdDID), url.QueryEscape(digest)) - resp, err := http.Get(reqURL) + req, err := http.NewRequest("GET", reqURL, nil) + if err != nil { + return "", fmt.Errorf("failed to create request: %w", err) + } + if secret != "" { + req.Header.Set("Authorization", "Bearer "+secret) + } + + resp, err := http.DefaultClient.Do(req) if err != nil { return "", fmt.Errorf("failed to get presigned URL: %w", err) } diff --git a/scanner/internal/scan/extractor.go b/scanner/internal/scan/extractor.go index 1162620..591edbf 100644 --- a/scanner/internal/scan/extractor.go +++ b/scanner/internal/scan/extractor.go @@ -17,7 +17,7 @@ import ( // extractLayers downloads and extracts all image layers via presigned URLs // Returns the rootfs directory path and a cleanup function -func extractLayers(job *scanner.ScanJob, tmpDir string) (string, func(), error) { +func extractLayers(job *scanner.ScanJob, tmpDir, secret string) (string, func(), error) { scanDir, err := os.MkdirTemp(tmpDir, "scan-*") if err != nil { return "", nil, fmt.Errorf("failed to create temp directory: %w", err) @@ -41,9 +41,13 @@ func extractLayers(job *scanner.ScanJob, tmpDir string) (string, func(), error) } // Download and validate config blob + if job.Config.Digest == "" { + cleanup() + return "", nil, fmt.Errorf("config blob has empty digest, cannot download") + } slog.Info("Downloading config blob", "digest", job.Config.Digest) configPath := filepath.Join(imageDir, "config.json") - if err := downloadBlobViaPresignedURL(job.HoldEndpoint, job.HoldDID, job.Config.Digest, configPath); err != nil { + if err := downloadBlobViaPresignedURL(job.HoldEndpoint, job.HoldDID, job.Config.Digest, configPath, secret); err != nil { cleanup() return "", nil, fmt.Errorf("failed to download config blob: %w", err) } @@ -61,10 +65,14 @@ func extractLayers(job *scanner.ScanJob, tmpDir string) (string, func(), error) // Download and extract each layer for i, layer := range job.Layers { + if layer.Digest == "" { + slog.Warn("Skipping layer with empty digest", "index", i) + continue + } slog.Info("Extracting layer", "index", i, "digest", layer.Digest, "size", layer.Size) layerPath := filepath.Join(layersDir, fmt.Sprintf("layer-%d.tar.gz", i)) - if err := downloadBlobViaPresignedURL(job.HoldEndpoint, job.HoldDID, layer.Digest, layerPath); err != nil { + if err := downloadBlobViaPresignedURL(job.HoldEndpoint, job.HoldDID, layer.Digest, layerPath, secret); err != nil { cleanup() return "", nil, fmt.Errorf("failed to download layer %d: %w", i, err) } @@ -91,8 +99,8 @@ func extractLayers(job *scanner.ScanJob, tmpDir string) (string, func(), error) } // downloadBlobViaPresignedURL gets a presigned URL from the hold and downloads the blob -func downloadBlobViaPresignedURL(holdEndpoint, holdDID, digest, destPath string) error { - presignedURL, err := client.GetBlobPresignedURL(holdEndpoint, holdDID, digest) +func downloadBlobViaPresignedURL(holdEndpoint, holdDID, digest, destPath, secret string) error { + presignedURL, err := client.GetBlobPresignedURL(holdEndpoint, holdDID, digest, secret) if err != nil { return fmt.Errorf("failed to get presigned URL for %s: %w", digest, err) } diff --git a/scanner/internal/scan/worker.go b/scanner/internal/scan/worker.go index ec22c85..ea5b105 100644 --- a/scanner/internal/scan/worker.go +++ b/scanner/internal/scan/worker.go @@ -106,7 +106,7 @@ func (wp *WorkerPool) processJob(ctx context.Context, job *scanner.ScanJob) (*sc // Step 1: Extract image layers from hold via presigned URLs slog.Info("Extracting image layers", "repository", job.Repository) - imageDir, cleanup, err := extractLayers(job, wp.cfg.Vuln.TmpDir) + imageDir, cleanup, err := extractLayers(job, wp.cfg.Vuln.TmpDir, wp.cfg.Hold.Secret) if err != nil { return nil, fmt.Errorf("failed to extract layers: %w", err) }