mirror of
https://tangled.org/evan.jarrett.net/at-container-registry
synced 2026-09-10 04:06:06 +00:00
An unauthenticated token request for a repo on a private hold came back with a signed token granting pull, and /v2/ then 401ed that exact token. The authorization server and the resource server disagreed about the same request. The old behaviour was deliberate — "minting a pull-only token is not a grant", with the hold owning the decision via captain.Public — but a token spec expects the server to issue the subset it will authorize, so granting pull and then refusing it is the wrong shape. Adds an optional AnonymousAuthorizer, kept separate from Authorizer because the anonymous path has no DID and no auth method (three of Authorize's four arguments are meaningless) and because it must drop whole entries rather than narrow actions in place, where entries can belong to different owners. Denied entries are dropped; if nothing granting survives, the caller gets the standard 401 challenge rather than a token with an empty access list, so docker prompts for credentials instead of proceeding to a second 401. The scope-less /v2/ ping and the actionless entry NarrowToPullOnly preserves on purpose both bypass the gate entirely — no identity resolution, no hold lookup — since anonymous discovery depends on them. Fails open on any lookup error, matching the /v2/ check, which states the reason: the hold is the enforcing authority and a transient failure must not break anonymous pulls of public images. /v2/ still enforces; this is a correctness and UX fix, not a security fix, and nothing was exposed. Also closes the successor asymmetry documented under finding 3: /v2/ applies a single-hop migration redirect before checking read access, so judging the pre-migration identity here would have reintroduced the disagreement this gate removes. It was one extra local read of hold_captain_records. Tests pin both directions and prove the chain is not followed past one hop. Costs one directory-cached identity resolution plus two local SQL reads per granting entry, and no call to the hold. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PDqoCE1j3njokkZ9b1C5n9
1249 lines
48 KiB
Go
1249 lines
48 KiB
Go
package appview
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"fmt"
|
|
"html/template"
|
|
"log/slog"
|
|
"net"
|
|
"net/http"
|
|
"net/url"
|
|
"os"
|
|
"os/signal"
|
|
"slices"
|
|
"strings"
|
|
"syscall"
|
|
"time"
|
|
|
|
"github.com/distribution/distribution/v3/registry/api/errcode"
|
|
"github.com/distribution/distribution/v3/registry/handlers"
|
|
"github.com/go-chi/chi/v5"
|
|
chimiddleware "github.com/go-chi/chi/v5/middleware"
|
|
"github.com/klauspost/compress/gzhttp"
|
|
|
|
"atcr.io/pkg/appview/authgate"
|
|
"atcr.io/pkg/appview/db"
|
|
"atcr.io/pkg/appview/holdhealth"
|
|
"atcr.io/pkg/appview/jetstream"
|
|
appviewlabeler "atcr.io/pkg/appview/labeler"
|
|
"atcr.io/pkg/appview/leases"
|
|
"atcr.io/pkg/appview/middleware"
|
|
"atcr.io/pkg/appview/readme"
|
|
"atcr.io/pkg/appview/registryauth"
|
|
"atcr.io/pkg/appview/routes"
|
|
"atcr.io/pkg/appview/storage"
|
|
"atcr.io/pkg/appview/webhooks"
|
|
"atcr.io/pkg/atproto"
|
|
"atcr.io/pkg/auth"
|
|
"atcr.io/pkg/auth/oauth"
|
|
"atcr.io/pkg/auth/token"
|
|
"atcr.io/pkg/billing"
|
|
"atcr.io/pkg/logging"
|
|
|
|
"github.com/bluesky-social/indigo/atproto/atcrypto"
|
|
indigooauth "github.com/bluesky-social/indigo/atproto/auth/oauth"
|
|
)
|
|
|
|
// OAuthPostAuthHook is called after the default OAuth post-auth logic
|
|
// (profile creation, avatar fetch, crew registration). Hooks added after
|
|
// NewAppViewServer but before the first request work correctly.
|
|
type OAuthPostAuthHook func(ctx context.Context, did, handle, pdsEndpoint, sessionID string) error
|
|
|
|
// TokenPostAuthHook is called after the default token post-auth logic
|
|
// (profile creation). Hooks added after NewAppViewServer but before the
|
|
// first request work correctly.
|
|
type TokenPostAuthHook func(ctx context.Context, did, handle, pdsEndpoint, accessToken string) error
|
|
|
|
// AppViewServer is the AppView service with an exposed router for extensibility.
|
|
// Consumers can add routes to Router and hooks before calling Serve().
|
|
type AppViewServer struct {
|
|
// Router is the chi router. Add routes before calling Serve().
|
|
Router chi.Router
|
|
|
|
// Config is the AppView configuration.
|
|
Config *Config
|
|
|
|
// Database is the read-write SQLite database.
|
|
Database *sql.DB
|
|
|
|
// ReadOnlyDB is the read-only SQLite database connection.
|
|
ReadOnlyDB *sql.DB
|
|
|
|
// SessionStore manages web UI sessions.
|
|
SessionStore *db.SessionStore
|
|
|
|
// DeviceStore manages device authorization flows.
|
|
DeviceStore *db.DeviceStore
|
|
|
|
// OAuthStore manages OAuth session persistence.
|
|
OAuthStore *db.OAuthStore
|
|
|
|
// OAuthServer handles OAuth authorization and callback endpoints.
|
|
OAuthServer *oauth.Server
|
|
|
|
// OAuthClientApp is the indigo OAuth client application.
|
|
OAuthClientApp *indigooauth.ClientApp
|
|
|
|
// Refresher manages OAuth session refresh and caching.
|
|
Refresher *oauth.Refresher
|
|
|
|
// Templates are the parsed HTML templates.
|
|
Templates *template.Template
|
|
|
|
// HealthChecker checks hold service health.
|
|
HealthChecker *holdhealth.Checker
|
|
|
|
// ReadmeFetcher fetches README content for repository pages.
|
|
ReadmeFetcher *readme.Fetcher
|
|
|
|
// TokenIssuer issues registry JWTs (nil if auth is not configured).
|
|
TokenIssuer *token.Issuer
|
|
|
|
// HoldAuthorizer checks hold access permissions.
|
|
HoldAuthorizer auth.HoldAuthorizer
|
|
|
|
// OAuthKey is the P-256 private key used for OAuth client auth and appview service identity.
|
|
OAuthKey *atcrypto.PrivateKeyP256
|
|
|
|
// BillingManager handles Stripe billing and tier updates (nil if billing disabled).
|
|
BillingManager *billing.Manager
|
|
|
|
// WebhookDispatcher dispatches scan webhooks (stored in appview DB).
|
|
WebhookDispatcher *webhooks.Dispatcher
|
|
|
|
// Leases elects a single instance to run each background worker.
|
|
Leases *leases.Manager
|
|
|
|
// Private fields for lifecycle management
|
|
oauthHooks []OAuthPostAuthHook
|
|
tokenHooks []TokenPostAuthHook
|
|
httpServer *http.Server
|
|
healthWorker *holdhealth.Worker
|
|
workerCancel context.CancelFunc
|
|
branding *BrandingOverrides
|
|
}
|
|
|
|
// AddOAuthPostAuthHook registers a hook that runs after the default OAuth
|
|
// post-auth logic. Multiple hooks run in registration order.
|
|
func (s *AppViewServer) AddOAuthPostAuthHook(hook OAuthPostAuthHook) {
|
|
s.oauthHooks = append(s.oauthHooks, hook)
|
|
}
|
|
|
|
// AddTokenPostAuthHook registers a hook that runs after the default token
|
|
// post-auth logic. Multiple hooks run in registration order.
|
|
func (s *AppViewServer) AddTokenPostAuthHook(hook TokenPostAuthHook) {
|
|
s.tokenHooks = append(s.tokenHooks, hook)
|
|
}
|
|
|
|
// NewAppViewServer creates a fully-initialized AppView server ready for the
|
|
// consumer to add routes and hooks before calling Serve(). Pass nil for
|
|
// branding to use default atcr.io assets and templates.
|
|
func NewAppViewServer(cfg *Config, branding *BrandingOverrides) (*AppViewServer, error) {
|
|
// Initialize structured logging with optional remote shipping
|
|
logging.InitLoggerWithShipper(cfg.LogLevel, logging.ShipperConfig{
|
|
Backend: cfg.LogShipper.Backend,
|
|
URL: cfg.LogShipper.URL,
|
|
BatchSize: cfg.LogShipper.BatchSize,
|
|
FlushInterval: cfg.LogShipper.FlushInterval,
|
|
Service: "appview",
|
|
Username: cfg.LogShipper.Username,
|
|
Password: cfg.LogShipper.Password,
|
|
})
|
|
|
|
slog.Info("Configuration loaded successfully from environment")
|
|
if cfg.AI.APIKey != "" {
|
|
slog.Info("AI Image Advisor enabled")
|
|
}
|
|
|
|
s := &AppViewServer{
|
|
Config: cfg,
|
|
branding: branding,
|
|
}
|
|
|
|
// Initialize UI database (required for all stores)
|
|
slog.Info("Initializing UI database", "path", cfg.UI.DatabasePath)
|
|
libsqlCfg := db.LibsqlConfig{
|
|
SyncURL: cfg.UI.LibsqlSyncURL,
|
|
AuthToken: cfg.UI.LibsqlAuthToken,
|
|
SyncInterval: cfg.UI.LibsqlSyncInterval,
|
|
}
|
|
s.Database, s.ReadOnlyDB, s.SessionStore = db.InitializeDatabase(cfg.UI.DatabasePath, libsqlCfg)
|
|
if s.Database == nil {
|
|
return nil, fmt.Errorf("failed to initialize UI database - required for session storage")
|
|
}
|
|
|
|
// Initialize hold health checker
|
|
slog.Info("Initializing hold health checker", "cache_ttl", cfg.Health.CacheTTL)
|
|
s.HealthChecker = holdhealth.NewChecker(cfg.Health.CacheTTL)
|
|
|
|
// Initialize README fetcher for rendering repo page descriptions
|
|
s.ReadmeFetcher = readme.NewFetcher()
|
|
|
|
// Start background health check worker
|
|
startupDelay := 5 * time.Second
|
|
dbAdapter := holdhealth.NewDBAdapter(s.Database)
|
|
s.healthWorker = holdhealth.NewWorkerWithStartupDelay(s.HealthChecker, dbAdapter, cfg.Health.CheckInterval, startupDelay)
|
|
|
|
workerCtx, workerCancel := context.WithCancel(context.Background())
|
|
s.workerCancel = workerCancel
|
|
s.healthWorker.Start(workerCtx)
|
|
slog.Info("Hold health worker started", "startup_delay", startupDelay, "refresh_interval", cfg.Health.CheckInterval, "cache_ttl", cfg.Health.CacheTTL)
|
|
|
|
// Leader election for the singleton background workers. The health worker
|
|
// above is deliberately not among them: it only refreshes a cache that each
|
|
// instance needs locally, so running it everywhere is correct.
|
|
s.Leases = leases.NewManager(s.Database, leases.Config{
|
|
Enabled: cfg.Leases.Enabled,
|
|
TTL: cfg.Leases.TTL,
|
|
RenewInterval: cfg.Leases.RenewInterval,
|
|
})
|
|
slog.Info("Lease manager initialized",
|
|
"component", "leases", "enabled", cfg.Leases.Enabled, "holder", s.Leases.HolderID())
|
|
|
|
// The cleanup worker starts later, once the hold authorizer exists: it
|
|
// clears the crew denial backoffs as its first act and needs a handle on it.
|
|
|
|
// Initialize OAuth components
|
|
slog.Info("Initializing OAuth components")
|
|
|
|
s.OAuthStore = db.NewOAuthStore(s.Database)
|
|
slog.Info("Using SQLite for OAuth session storage")
|
|
|
|
s.DeviceStore = db.NewDeviceStore(s.Database)
|
|
slog.Info("Using SQLite for device storage")
|
|
|
|
baseURL := cfg.Server.BaseURL
|
|
defaultHoldDID := cfg.Server.PrimaryHoldDID()
|
|
testMode := cfg.Server.TestMode
|
|
|
|
slog.Debug("Base URL for OAuth", "base_url", baseURL)
|
|
if testMode {
|
|
slog.Info("TEST_MODE enabled - will use HTTP for local DID resolution")
|
|
atproto.SetTestMode(true)
|
|
}
|
|
|
|
oauthKey, err := loadOAuthKey(s.Database)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to load OAuth key: %w", err)
|
|
}
|
|
s.OAuthKey = oauthKey
|
|
|
|
// Create OAuth client app
|
|
desiredScopes := oauth.GetDefaultScopes(defaultHoldDID)
|
|
s.OAuthClientApp, err = oauth.NewClientAppWithKey(baseURL, s.OAuthStore, desiredScopes, oauthKey, cfg.Server.ClientName)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to create OAuth client app: %w", err)
|
|
}
|
|
|
|
// Invalidate sessions with mismatched scopes on startup
|
|
invalidatedCount, err := s.OAuthStore.InvalidateSessionsWithMismatchedScopes(context.Background(), desiredScopes)
|
|
if err != nil {
|
|
slog.Warn("Failed to invalidate sessions with mismatched scopes", "error", err)
|
|
} else if invalidatedCount > 0 {
|
|
slog.Info("Invalidated OAuth sessions due to scope changes", "count", invalidatedCount)
|
|
}
|
|
|
|
// Create oauth token refresher
|
|
s.Refresher = oauth.NewRefresher(s.OAuthClientApp)
|
|
|
|
// Wire up UI session store to refresher
|
|
if s.SessionStore != nil {
|
|
s.Refresher.SetUISessionStore(s.SessionStore)
|
|
}
|
|
|
|
// Set global refresher for middleware
|
|
middleware.SetGlobalRefresher(s.Refresher)
|
|
|
|
// Set global database for hold DID lookups and manifest reference checks
|
|
holdDIDDB := db.NewHoldDIDDB(s.Database)
|
|
middleware.SetGlobalDatabase(holdDIDDB)
|
|
middleware.SetGlobalManifestRefChecker(holdDIDDB)
|
|
|
|
// Set label checker for takedown filtering
|
|
middleware.SetGlobalLabelChecker(db.NewLabelChecker(s.Database))
|
|
|
|
// Create RemoteHoldAuthorizer for hold authorization with caching
|
|
s.HoldAuthorizer = auth.NewRemoteHoldAuthorizer(s.Database, testMode)
|
|
middleware.SetGlobalAuthorizer(s.HoldAuthorizer)
|
|
slog.Info("Hold authorizer initialized with database caching")
|
|
|
|
// Session, OAuth and device-flow cleanup, plus the one-shot denial-cache
|
|
// clear. Started here rather than next to the other workers because it needs
|
|
// the hold authorizer above.
|
|
s.startCleanupWorker(workerCtx)
|
|
|
|
// Initialize billing manager
|
|
appviewDID := DIDFromBaseURL(baseURL)
|
|
s.BillingManager = billing.New(
|
|
&cfg.Billing,
|
|
oauthKey,
|
|
appviewDID,
|
|
cfg.Server.ManagedHolds,
|
|
baseURL,
|
|
s.Database,
|
|
)
|
|
// Allow hold captains to bypass billing feature gates
|
|
if len(cfg.Server.ManagedHolds) > 0 {
|
|
managedHolds := cfg.Server.ManagedHolds
|
|
roDB := s.ReadOnlyDB
|
|
s.BillingManager.SetCaptainChecker(func(userDID string) bool {
|
|
isCaptain, _ := db.IsHoldCaptain(roDB, userDID, managedHolds)
|
|
return isCaptain
|
|
})
|
|
// Paid features require the user's active (default) hold to be a managed
|
|
// hold. An unset default hold means they use the operator's primary
|
|
// managed hold, so it counts as managed. Read the primary DB (not roDB):
|
|
// a hold switch writes default_hold_did to the primary, and the replica
|
|
// may lag, so reading roDB could keep paid features alive briefly after a
|
|
// switch to self-hosted.
|
|
primaryDB := s.Database
|
|
s.BillingManager.SetActiveHoldChecker(func(userDID string) bool {
|
|
holdDID := db.GetUserDefaultHoldDID(primaryDB, userDID)
|
|
return holdDID == "" || slices.Contains(managedHolds, holdDID)
|
|
})
|
|
}
|
|
if s.BillingManager.Enabled() {
|
|
// Fail closed: an empty Stripe webhook secret makes webhooks forgeable
|
|
// (Stripe HMACs with the empty key, which an attacker can reproduce).
|
|
if !s.BillingManager.WebhookConfigured() {
|
|
return nil, fmt.Errorf("billing is enabled but STRIPE_WEBHOOK_SECRET is not set; refusing to start with a forgeable webhook endpoint")
|
|
}
|
|
slog.Info("Billing enabled", "appview_did", appviewDID, "managed_holds", len(cfg.Server.ManagedHolds))
|
|
// Deliberately not leased. holdTierCache is per-process memory fed by
|
|
// read-only ListTiers calls, so there is nothing to serialise, and
|
|
// electing one refresher would leave every other instance reporting
|
|
// "no hold data" from aggregateHoldFeatures. Same reasoning as the
|
|
// hold health worker above.
|
|
go s.BillingManager.RefreshHoldTiers(workerCtx)
|
|
}
|
|
|
|
// Create webhook dispatcher
|
|
appviewMeta := atproto.AppviewMetadata{
|
|
ClientName: cfg.Server.ClientName,
|
|
ClientShortName: cfg.Server.ClientShortName,
|
|
BaseURL: cfg.Server.BaseURL,
|
|
FaviconURL: cfg.Server.BaseURL + "/favicon-96x96.png",
|
|
RegistryDomains: cfg.Server.RegistryDomains,
|
|
}
|
|
// Gate dispatch on the same entitlement the creation form uses, so webhooks
|
|
// stop firing paid behavior when a user loses entitlement (switches to a
|
|
// self-hosted hold or downgrades) after creating them.
|
|
s.WebhookDispatcher = webhooks.NewDispatcher(s.Database, appviewMeta, s.BillingManager.GetWebhookLimits)
|
|
middleware.SetGlobalWebhookDispatcher(s.WebhookDispatcher)
|
|
|
|
// Initialize Jetstream workers
|
|
s.initializeJetstream(workerCtx)
|
|
|
|
// Initialize labeler subscriber. Leased per labeler DID, matching the
|
|
// per-src cursor in labeler_cursor: two subscribers on one labeler would
|
|
// each advance the same cursor and double-apply every takedown.
|
|
if cfg.Labeler.DID != "" {
|
|
labelerDID := cfg.Labeler.DID
|
|
s.Leases.Go(workerCtx, db.LeaseLabeler(labelerDID), func(ctx context.Context) error {
|
|
// Built inside the worker because Stop closes a channel and cannot
|
|
// be called twice; a retry after a lost lease needs a fresh one.
|
|
sub := appviewlabeler.SubscriberFromConfig(labelerDID, s.Database)
|
|
if sub == nil {
|
|
return fmt.Errorf("labeler subscriber unavailable for %q", labelerDID)
|
|
}
|
|
sub.Start()
|
|
slog.Info("Labeler subscriber started", "labeler", labelerDID)
|
|
<-ctx.Done()
|
|
sub.Stop()
|
|
return ctx.Err()
|
|
})
|
|
}
|
|
|
|
// Create main chi router
|
|
mainRouter := chi.NewRouter()
|
|
|
|
mainRouter.Use(chimiddleware.RealIP)
|
|
mainRouter.Use(chimiddleware.Logger)
|
|
mainRouter.Use(chimiddleware.Recoverer)
|
|
mainRouter.Use(compressUIResponses)
|
|
mainRouter.Use(chimiddleware.GetHead)
|
|
mainRouter.Use(routes.CORSMiddleware())
|
|
// Vanity import prefix must match the `module` line in go.mod exactly,
|
|
// otherwise `go install` fails the module-path check.
|
|
mainRouter.Use(middleware.GoImport("atcr.io", cfg.UI.SourceURL))
|
|
|
|
// Domain routing middleware. Gated on the raw config (an empty
|
|
// registry_domains means single-domain, where /v2/ must stay on the UI
|
|
// host) but fed cfg.Auth.Services, which is the same list normalized. The
|
|
// middleware matches a port-stripped host, so a domain configured with a
|
|
// port could never match its own requests when compared raw; sharing the
|
|
// normalized list also keeps routing and the access controller agreeing on
|
|
// exactly one set of names.
|
|
if len(cfg.Server.RegistryDomains) > 0 {
|
|
mainRouter.Use(DomainRoutingMiddleware(cfg.Auth.Services, cfg.Server.BaseURL))
|
|
slog.Info("Domain routing middleware enabled",
|
|
"registry_domains", cfg.Auth.Services,
|
|
"ui_base_url", cfg.Server.BaseURL)
|
|
}
|
|
|
|
// Load templates
|
|
ComputeAssetHashes(branding)
|
|
s.Templates, err = Templates(branding)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to load UI templates: %w", err)
|
|
}
|
|
|
|
// Register UI routes
|
|
routes.RegisterUIRoutes(mainRouter, routes.UIDependencies{
|
|
Database: s.Database,
|
|
ReadOnlyDB: s.ReadOnlyDB,
|
|
SessionStore: s.SessionStore,
|
|
OAuthClientApp: s.OAuthClientApp,
|
|
OAuthStore: s.OAuthStore,
|
|
Refresher: s.Refresher,
|
|
BaseURL: baseURL,
|
|
RegistryDomain: primaryRegistryDomain(cfg.Server.RegistryDomains),
|
|
RegistryDomains: cfg.Server.RegistryDomains,
|
|
DeviceStore: s.DeviceStore,
|
|
HealthChecker: s.HealthChecker,
|
|
ReadmeFetcher: s.ReadmeFetcher,
|
|
Templates: s.Templates,
|
|
DefaultHoldDID: defaultHoldDID,
|
|
ManagedHolds: cfg.Server.ManagedHolds,
|
|
ClientName: cfg.Server.ClientName,
|
|
ClientShortName: cfg.Server.ClientShortName,
|
|
BillingManager: s.BillingManager,
|
|
WebhookDispatcher: s.WebhookDispatcher,
|
|
ClaudeAPIKey: cfg.AI.APIKey,
|
|
SourceURL: cfg.UI.SourceURL,
|
|
BlueskyProfile: cfg.UI.BlueskyProfile,
|
|
LegalConfig: routes.LegalConfig{
|
|
CompanyName: cfg.Legal.CompanyName,
|
|
Jurisdiction: cfg.Legal.Jurisdiction,
|
|
},
|
|
})
|
|
|
|
// Register Stripe webhook route (if billing enabled)
|
|
s.BillingManager.RegisterRoutes(mainRouter)
|
|
|
|
// Create OAuth server
|
|
s.OAuthServer = oauth.NewServer(s.OAuthClientApp)
|
|
s.OAuthServer.SetRefresher(s.Refresher)
|
|
if s.SessionStore != nil {
|
|
s.OAuthServer.SetUISessionStore(s.SessionStore)
|
|
}
|
|
|
|
// Register OAuth post-auth callback (closure captures s for hook dispatch)
|
|
s.OAuthServer.SetPostAuthCallback(func(ctx context.Context, did, handle, pdsEndpoint, sessionID string) error {
|
|
slog.Debug("OAuth post-auth callback", "component", "appview/callback", "did", did)
|
|
|
|
// Create ATProto client with session provider
|
|
client := atproto.NewClientWithSessionProvider(pdsEndpoint, did, s.Refresher)
|
|
|
|
// Ensure sailor profile exists
|
|
slog.Debug("Ensuring profile exists", "component", "appview/callback", "did", did, "default_hold_did", defaultHoldDID)
|
|
if err := storage.EnsureProfile(ctx, client, defaultHoldDID); err != nil {
|
|
slog.Warn("Failed to ensure profile", "component", "appview/callback", "did", did, "error", err)
|
|
} else {
|
|
slog.Debug("Profile ensured", "component", "appview/callback", "did", did)
|
|
}
|
|
|
|
// Fetch user's profile record from PDS
|
|
profileRecord, err := client.GetProfileRecord(ctx, did)
|
|
if err != nil {
|
|
slog.Warn("Failed to fetch profile record", "component", "appview/callback", "did", did, "error", err)
|
|
profileRecord = nil
|
|
}
|
|
|
|
// Construct avatar URL from blob CID
|
|
avatarURL := ""
|
|
if profileRecord != nil && profileRecord.Avatar != nil && profileRecord.Avatar.Ref.Link != "" {
|
|
avatarURL = atproto.BlobCDNURL(did, profileRecord.Avatar.Ref.Link)
|
|
slog.Debug("Constructed avatar URL", "component", "appview/callback", "avatar_url", avatarURL)
|
|
}
|
|
|
|
// Store user in database
|
|
if avatarURL != "" {
|
|
err = db.UpsertUser(s.Database, &db.User{
|
|
DID: did,
|
|
Handle: handle,
|
|
PDSEndpoint: pdsEndpoint,
|
|
Avatar: avatarURL,
|
|
LastSeen: time.Now(),
|
|
})
|
|
} else {
|
|
err = db.UpsertUserIgnoreAvatar(s.Database, &db.User{
|
|
DID: did,
|
|
Handle: handle,
|
|
PDSEndpoint: pdsEndpoint,
|
|
Avatar: avatarURL,
|
|
LastSeen: time.Now(),
|
|
})
|
|
}
|
|
if err != nil {
|
|
slog.Warn("Failed to store user in database", "component", "appview/callback", "error", err)
|
|
return nil
|
|
}
|
|
|
|
slog.Debug("Stored user", "component", "appview/callback", "did", did, "has_avatar", avatarURL != "")
|
|
|
|
// Migrate profile URL→DID if needed
|
|
profile, err := storage.GetProfile(ctx, client)
|
|
if err != nil {
|
|
slog.Warn("Failed to get profile", "component", "appview/callback", "did", did, "error", err)
|
|
return nil
|
|
}
|
|
|
|
var holdDID string
|
|
if profile != nil && profile.DefaultHold != "" {
|
|
if strings.HasPrefix(profile.DefaultHold, "http://") || strings.HasPrefix(profile.DefaultHold, "https://") {
|
|
slog.Debug("Migrating hold URL to DID", "component", "appview/callback", "did", did, "hold_url", profile.DefaultHold)
|
|
|
|
if resolvedDID, resolveErr := atproto.ResolveHoldDID(ctx, profile.DefaultHold); resolveErr != nil {
|
|
slog.Warn("Failed to resolve hold DID from URL", "component", "appview/callback", "did", did, "hold_url", profile.DefaultHold, "error", resolveErr)
|
|
} else {
|
|
holdDID = resolvedDID
|
|
profile.DefaultHold = holdDID
|
|
if err := storage.UpdateProfile(ctx, client, profile); err != nil {
|
|
slog.Warn("Failed to update profile with hold DID", "component", "appview/callback", "did", did, "error", err)
|
|
} else {
|
|
slog.Debug("Updated profile with hold DID", "component", "appview/callback", "hold_did", holdDID)
|
|
}
|
|
}
|
|
} else {
|
|
holdDID = profile.DefaultHold
|
|
}
|
|
// Register crew in background
|
|
slog.Debug("Attempting crew registration", "component", "appview/callback", "did", did, "hold_did", holdDID)
|
|
go func(userDID, pdsEndpoint, holdDID string, refresher *oauth.Refresher, authorizer auth.HoldAuthorizer) {
|
|
ctx := context.Background()
|
|
storage.EnsureCrewMembership(ctx, userDID, holdDID, authorizer,
|
|
func(ctx context.Context, holdDID string) (string, error) {
|
|
return auth.GetOrFetchServiceToken(ctx, refresher, userDID, holdDID, pdsEndpoint)
|
|
})
|
|
}(did, pdsEndpoint, holdDID, s.Refresher, s.HoldAuthorizer)
|
|
}
|
|
|
|
// Drain manifests from old hold to successor in background
|
|
go func(client *atproto.Client, did string) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
|
defer cancel()
|
|
storage.MigrateManifestsForSuccessor(ctx, client, s.HoldAuthorizer, db.NewHoldDIDDB(s.Database), did)
|
|
}(client, did)
|
|
|
|
// Run consumer hooks
|
|
for _, hook := range s.oauthHooks {
|
|
if err := hook(ctx, did, handle, pdsEndpoint, sessionID); err != nil {
|
|
slog.Warn("OAuth post-auth hook error", "component", "appview/callback", "error", err)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
})
|
|
|
|
// Create token issuer
|
|
if cfg.Distribution.Auth[registryauth.AuthType] != nil {
|
|
rsaKey, certDER, err := loadJWTKeyAndCert(s.Database, cfg.Auth.CertPath)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to load JWT key material: %w", err)
|
|
}
|
|
s.TokenIssuer = token.NewIssuerFromKey(rsaKey, certDER, cfg.Auth.PrimaryService(), cfg.Auth.PrimaryService(), cfg.Auth.TokenExpiration)
|
|
slog.Info("Auth keys initialized")
|
|
}
|
|
|
|
// Create registry app (distribution library handler)
|
|
ctx := context.Background()
|
|
app := handlers.NewApp(ctx, cfg.Distribution)
|
|
|
|
// Wrap with auth method extraction middleware, then with the Retry-After
|
|
// emitter so it can read the carrier installed before deeper handlers run.
|
|
// Outermost is the Bearer challenge guard, so any 401 from deep in the stack
|
|
// (e.g. anonymous read of a private hold) still carries WWW-Authenticate.
|
|
wrappedApp := middleware.BearerChallenge(cfg.Server.BaseURL+"/auth/token", cfg.Auth.Services)(
|
|
middleware.RetryAfterMiddleware(middleware.ExtractAuthMethod(app)))
|
|
|
|
// Refuse the global catalog before the distribution handler can see it.
|
|
mountCatalogRefusal(mainRouter)
|
|
|
|
// Mount registry at /v2/
|
|
mainRouter.Handle("/v2/*", wrappedApp)
|
|
|
|
// Mount static files
|
|
if s.SessionStore != nil && s.Templates != nil {
|
|
publicHandler := CacheMiddleware(PublicHandler(branding), 31536000)
|
|
rootFiles, err := PublicRootFiles(branding)
|
|
if err != nil {
|
|
slog.Warn("Failed to scan static root files", "error", err)
|
|
} else {
|
|
for _, filename := range rootFiles {
|
|
file := filename
|
|
mainRouter.Get("/"+file, func(w http.ResponseWriter, r *http.Request) {
|
|
r.URL.Path = "/" + file
|
|
publicHandler.ServeHTTP(w, r)
|
|
})
|
|
}
|
|
slog.Info("Registered dynamic root file routes", "count", len(rootFiles), "files", rootFiles)
|
|
}
|
|
|
|
mainRouter.Handle("/css/*", CacheMiddleware(http.StripPrefix("/css/", PublicSubdir("css", branding)), 31536000))
|
|
mainRouter.Handle("/js/*", CacheMiddleware(http.StripPrefix("/js/", PublicSubdir("js", branding)), 31536000))
|
|
mainRouter.Handle("/fonts/*", CacheMiddleware(http.StripPrefix("/fonts/", PublicSubdir("fonts", branding)), 31536000))
|
|
mainRouter.Handle("/static/*", CacheMiddleware(http.StripPrefix("/static/", PublicSubdir("static", branding)), 31536000))
|
|
|
|
slog.Info("UI enabled", "home", "/", "settings", "/settings")
|
|
}
|
|
|
|
// Mount OAuth endpoints
|
|
mainRouter.Get("/auth/oauth/authorize", s.OAuthServer.ServeAuthorize)
|
|
mainRouter.Get("/auth/oauth/callback", s.OAuthServer.ServeCallback)
|
|
|
|
// OAuth client metadata endpoint
|
|
mainRouter.Get("/oauth-client-metadata.json", func(w http.ResponseWriter, r *http.Request) {
|
|
config := s.OAuthClientApp.Config
|
|
logoURI := cfg.Server.BaseURL + "/web-app-manifest-192x192.png"
|
|
policyURI := cfg.Server.BaseURL + "/privacy"
|
|
tosURI := cfg.Server.BaseURL + "/terms"
|
|
|
|
metadata := config.ClientMetadata()
|
|
metadata.ClientName = &cfg.Server.ClientName
|
|
metadata.ClientURI = &cfg.Server.BaseURL
|
|
metadata.LogoURI = &logoURI
|
|
metadata.PolicyURI = &policyURI
|
|
metadata.TosURI = &tosURI
|
|
|
|
if config.IsConfidential() && metadata.JWKS == nil {
|
|
jwks := config.PublicJWKS()
|
|
metadata.JWKS = &jwks
|
|
}
|
|
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.Header().Set("Access-Control-Allow-Origin", "*")
|
|
w.Header().Set("Cache-Control", "public, max-age=300")
|
|
if err := json.NewEncoder(w).Encode(metadata); err != nil {
|
|
http.Error(w, "Failed to encode metadata", http.StatusInternalServerError)
|
|
}
|
|
})
|
|
|
|
// Mount auth endpoints
|
|
if s.TokenIssuer != nil {
|
|
tokenHandler := token.NewHandler(s.TokenIssuer, s.DeviceStore)
|
|
|
|
// Stamp each JWT with the registry domain the client is pushing to, so
|
|
// the audience names the front door actually used.
|
|
tokenHandler.SetServices(cfg.Auth.Services)
|
|
|
|
tokenHandler.SetOAuthSessionValidator(s.Refresher)
|
|
|
|
// Auth-phase gate: crew reconciliation for any token request, plus
|
|
// hold membership + quota for non-wildcard push. Gating here means
|
|
// the JWT carries the authorization, so /v2/* doesn't need to
|
|
// re-check on every blob.
|
|
tokenHandler.SetAuthorizer(authgate.New(s.Database, s.HoldAuthorizer, s.Refresher, defaultHoldDID))
|
|
|
|
// Credential-less pull gate: drop the scopes whose hold denies
|
|
// anonymous reads, so /auth/token stops signing pull tokens that
|
|
// /v2/ then refuses. Local reads only; fails open.
|
|
tokenHandler.SetAnonymousAuthorizer(authgate.NewAnonymousAuthorizer(s.Database, defaultHoldDID))
|
|
|
|
// Bind the registry JWT lifetime to the AppView↔hold service-auth.
|
|
// Pre-minting the service-auth here lets us stamp the JWT's exp
|
|
// from the cached expiry, so both expire concurrently.
|
|
tokenHandler.SetServiceAuthFetcher(authgate.NewServiceAuthFetcher(s.Database, s.Refresher, defaultHoldDID))
|
|
|
|
// Token post-auth callback (closure captures s for hook dispatch)
|
|
tokenHandler.SetPostAuthCallback(func(ctx context.Context, did, handle, pdsEndpoint, accessToken string) error {
|
|
slog.Debug("Token post-auth callback", "component", "appview/callback", "did", did)
|
|
|
|
atprotoClient := atproto.NewClient(pdsEndpoint, did, accessToken)
|
|
|
|
if err := storage.EnsureProfile(ctx, atprotoClient, defaultHoldDID); err != nil {
|
|
slog.Warn("Failed to ensure profile", "component", "appview/callback", "did", did, "error", err)
|
|
} else {
|
|
slog.Debug("Profile ensured with default hold", "component", "appview/callback", "did", did, "default_hold_did", defaultHoldDID)
|
|
}
|
|
|
|
// Crew enrollment is handled synchronously by authgate.Authorize on
|
|
// the same /auth/token request, which also warms the approval cache.
|
|
// No duplicate goroutine needed here.
|
|
|
|
// Run consumer hooks
|
|
for _, hook := range s.tokenHooks {
|
|
if err := hook(ctx, did, handle, pdsEndpoint, accessToken); err != nil {
|
|
slog.Warn("Token post-auth hook error", "component", "appview/callback", "error", err)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
})
|
|
|
|
// Both Docker token specs are served on the same path: GET is the
|
|
// original Basic-auth form, POST is the OAuth2 form that containerd and
|
|
// Docker try first whenever they hold a secret. Serving only GET made
|
|
// every such client eat a 405 and retry.
|
|
mainRouter.Get("/auth/token", tokenHandler.ServeHTTP)
|
|
mainRouter.Post("/auth/token", tokenHandler.ServeHTTP)
|
|
|
|
// Device authorization endpoints (public)
|
|
routes.RegisterDeviceEndpoints(mainRouter, s.DeviceStore, baseURL)
|
|
|
|
slog.Info("Auth endpoints enabled",
|
|
"basic_auth", "/auth/token",
|
|
"device_code", "/auth/device/code",
|
|
"device_token", "/auth/device/token",
|
|
"oauth_authorize", "/auth/oauth/authorize",
|
|
"oauth_callback", "/auth/oauth/callback",
|
|
"oauth_metadata", "/oauth-client-metadata.json")
|
|
}
|
|
|
|
// Health check endpoint (for Docker health checks / load balancers)
|
|
mainRouter.Get("/health", func(w http.ResponseWriter, r *http.Request) {
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.WriteHeader(http.StatusOK)
|
|
if err := json.NewEncoder(w).Encode(map[string]string{"status": "ok"}); err != nil {
|
|
http.Error(w, "encode error", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
})
|
|
|
|
// Appview metadata endpoint (public, used by holds for branding)
|
|
mainRouter.Get(atproto.AppviewGetMetadata, func(w http.ResponseWriter, r *http.Request) {
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.Header().Set("Cache-Control", "public, max-age=3600")
|
|
if err := json.NewEncoder(w).Encode(atproto.AppviewMetadata{
|
|
ClientName: cfg.Server.ClientName,
|
|
ClientShortName: cfg.Server.ClientShortName,
|
|
BaseURL: cfg.Server.BaseURL,
|
|
FaviconURL: cfg.Server.BaseURL + "/favicon-96x96.png",
|
|
RegistryDomains: cfg.Server.RegistryDomains,
|
|
}); err != nil {
|
|
http.Error(w, "encode error", http.StatusInternalServerError)
|
|
}
|
|
})
|
|
|
|
// Appview DID document endpoint (service identity for key discovery)
|
|
mainRouter.Get("/.well-known/did.json", s.handleDIDDocument)
|
|
|
|
s.Router = mainRouter
|
|
|
|
return s, nil
|
|
}
|
|
|
|
// Serve starts the HTTP server on the configured address and blocks until
|
|
// shutdown signal.
|
|
func (s *AppViewServer) Serve() error {
|
|
listener, err := net.Listen("tcp", s.Config.Server.Addr)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to create listener: %w", err)
|
|
}
|
|
return s.ServeWithListener(listener)
|
|
}
|
|
|
|
// ServeWithListener starts the HTTP server on the provided listener and
|
|
// blocks until a shutdown signal arrives or Shutdown() is called. Tests use
|
|
// this to bind a 127.0.0.1:0 listener and learn the assigned port before
|
|
// driving requests.
|
|
func (s *AppViewServer) ServeWithListener(listener net.Listener) error {
|
|
s.httpServer = &http.Server{
|
|
Handler: s.Router,
|
|
}
|
|
|
|
stop := make(chan os.Signal, 1)
|
|
signal.Notify(stop, os.Interrupt, syscall.SIGTERM)
|
|
defer signal.Stop(stop)
|
|
|
|
serveErr := make(chan error, 1)
|
|
go func() {
|
|
slog.Info("Starting registry server", "addr", listener.Addr().String())
|
|
err := s.httpServer.Serve(listener)
|
|
if err != nil && err != http.ErrServerClosed {
|
|
serveErr <- err
|
|
return
|
|
}
|
|
serveErr <- nil
|
|
}()
|
|
|
|
select {
|
|
case <-stop:
|
|
slog.Info("Shutting down registry server")
|
|
|
|
if s.Config.Server.TestMode {
|
|
listener.Close()
|
|
}
|
|
|
|
slog.Info("Stopping hold health worker")
|
|
s.healthWorker.Stop()
|
|
|
|
if s.workerCancel != nil {
|
|
s.workerCancel()
|
|
}
|
|
s.waitForLeasedWorkers()
|
|
|
|
shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
|
|
if err := s.httpServer.Shutdown(shutdownCtx); err != nil && err != http.ErrServerClosed {
|
|
logging.Shutdown()
|
|
return fmt.Errorf("server shutdown error: %w", err)
|
|
}
|
|
case err := <-serveErr:
|
|
s.healthWorker.Stop()
|
|
if s.workerCancel != nil {
|
|
s.workerCancel()
|
|
}
|
|
s.waitForLeasedWorkers()
|
|
logging.Shutdown()
|
|
if err != nil {
|
|
return fmt.Errorf("server error: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
logging.Shutdown()
|
|
return nil
|
|
}
|
|
|
|
// Shutdown gracefully stops the HTTP server. Safe to call from tests that
|
|
// don't want to send SIGTERM. Production code goes through the signal handler
|
|
// inside ServeWithListener.
|
|
func (s *AppViewServer) Shutdown(ctx context.Context) error {
|
|
if s.httpServer == nil {
|
|
return nil
|
|
}
|
|
return s.httpServer.Shutdown(ctx)
|
|
}
|
|
|
|
// DomainRoutingMiddleware enforces three-tier domain routing:
|
|
//
|
|
// 1. UI domain (BaseURL hostname): serves web UI, auth, and static assets.
|
|
// Blocks /v2/* with an OCI UNSUPPORTED error — registry API lives on
|
|
// the dedicated registry domain(s).
|
|
// 2. Registry domains: allows /v2/* for Docker clients. Redirects everything
|
|
// else to the UI domain with 307 Temporary Redirect.
|
|
// 3. Unknown domains (CDN origins, IPs, etc.): redirects all requests to the
|
|
// UI domain with 307, except /health for load balancer probes.
|
|
func DomainRoutingMiddleware(registryDomains []string, uiBaseURL string) func(http.Handler) http.Handler {
|
|
// Request hosts are normalized to a bare hostname before matching, so the
|
|
// configured domains have to be normalized the same way. They frequently
|
|
// carry a port — config-appview.example.yaml ships "127.0.0.1:5000" — and
|
|
// matching those verbatim meant such an entry could never match anything.
|
|
regDomains := make(map[string]bool, len(registryDomains))
|
|
for _, d := range registryDomains {
|
|
regDomains[hostWithoutPort(d)] = true
|
|
}
|
|
|
|
// Extract UI hostname from BaseURL (e.g., "https://seamark.dev" -> "seamark.dev")
|
|
var uiHost string
|
|
if parsed, err := url.Parse(uiBaseURL); err == nil {
|
|
uiHost = parsed.Hostname()
|
|
}
|
|
|
|
primaryReg := primaryRegistryDomain(registryDomains)
|
|
|
|
return func(next http.Handler) http.Handler {
|
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
host := hostWithoutPort(r.Host)
|
|
|
|
path := r.URL.Path
|
|
isV2 := path == "/v2" || path == "/v2/" || strings.HasPrefix(path, "/v2/")
|
|
|
|
switch {
|
|
case regDomains[host] && isV2:
|
|
// A registry domain gets /v2/* even when it is also the UI host.
|
|
// One host serving both is a legitimate deployment (it is what
|
|
// the dev stack is); checking uiHost first meant listing that
|
|
// host in registry_domains was silently ignored.
|
|
next.ServeHTTP(w, r)
|
|
|
|
case host == uiHost:
|
|
// UI domain: block /v2/*, serve everything else
|
|
if isV2 {
|
|
writeEmptyAllow(w)
|
|
if err := errcode.ServeJSON(w, errcode.ErrorCodeUnsupported.WithMessage(
|
|
fmt.Sprintf("registry API is not available on this domain, use %s", primaryReg),
|
|
)); err != nil {
|
|
slog.Error("failed to write OCI error response", "error", err)
|
|
}
|
|
return
|
|
}
|
|
next.ServeHTTP(w, r)
|
|
|
|
case regDomains[host]:
|
|
// 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
|
|
}
|
|
http.Redirect(w, r, uiBaseURL+r.URL.RequestURI(), http.StatusTemporaryRedirect)
|
|
|
|
default:
|
|
// Unknown domain: allow /health, redirect everything else
|
|
if path == "/health" {
|
|
next.ServeHTTP(w, r)
|
|
return
|
|
}
|
|
http.Redirect(w, r, uiBaseURL+r.URL.RequestURI(), http.StatusTemporaryRedirect)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// hostWithoutPort strips a trailing :port from a host, leaving bare hostnames
|
|
// untouched. net.SplitHostPort is used rather than a LastIndex(":") scan so
|
|
// that bracketed IPv6 literals ("[::1]:5000") normalize to "::1" and match the
|
|
// form url.URL.Hostname() produces for the UI host.
|
|
func hostWithoutPort(h string) string {
|
|
if host, _, err := net.SplitHostPort(h); err == nil {
|
|
return host
|
|
}
|
|
return h
|
|
}
|
|
|
|
// primaryRegistryDomain returns the first registry domain, or empty string if none.
|
|
func primaryRegistryDomain(domains []string) string {
|
|
if len(domains) > 0 {
|
|
return domains[0]
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// DID returns the appview's did:web identity derived from its BaseURL.
|
|
func (s *AppViewServer) DID() string {
|
|
return DIDFromBaseURL(s.Config.Server.BaseURL)
|
|
}
|
|
|
|
// DIDFromBaseURL derives a did:web identifier from a base URL.
|
|
// Per the did:web spec, non-standard ports are percent-encoded.
|
|
// Examples:
|
|
//
|
|
// "https://atcr.io" → "did:web:atcr.io"
|
|
// "http://localhost:5000" → "did:web:localhost%3A5000"
|
|
func DIDFromBaseURL(baseURL string) string {
|
|
u, err := url.Parse(baseURL)
|
|
if err != nil {
|
|
return "did:web:localhost"
|
|
}
|
|
|
|
hostname := u.Hostname()
|
|
if hostname == "" {
|
|
hostname = "localhost"
|
|
}
|
|
|
|
port := u.Port()
|
|
isStandardPort := (u.Scheme == "https" && port == "443") ||
|
|
(u.Scheme == "http" && port == "80") ||
|
|
port == ""
|
|
|
|
if isStandardPort {
|
|
return "did:web:" + hostname
|
|
}
|
|
return fmt.Sprintf("did:web:%s%%3A%s", hostname, port)
|
|
}
|
|
|
|
// handleDIDDocument serves the appview's DID document at /.well-known/did.json.
|
|
// This is a service identity for key discovery — no PDS, no repo, no firehose.
|
|
// Holds use this to discover the appview's P-256 public key for JWT verification.
|
|
func (s *AppViewServer) handleDIDDocument(w http.ResponseWriter, r *http.Request) {
|
|
did := s.DID()
|
|
|
|
pubKey, err := s.OAuthKey.PublicKey()
|
|
if err != nil {
|
|
slog.Error("Failed to get public key for DID document", "error", err)
|
|
http.Error(w, "internal error", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
doc := map[string]any{
|
|
"@context": []string{
|
|
"https://www.w3.org/ns/did/v1",
|
|
"https://w3id.org/security/multikey/v1",
|
|
},
|
|
"id": did,
|
|
"verificationMethod": []map[string]any{
|
|
{
|
|
"id": did + "#appview",
|
|
"type": "Multikey",
|
|
"controller": did,
|
|
"publicKeyMultibase": pubKey.Multibase(),
|
|
},
|
|
},
|
|
"authentication": []string{
|
|
did + "#appview",
|
|
},
|
|
"assertionMethod": []string{
|
|
did + "#appview",
|
|
},
|
|
"service": []map[string]any{
|
|
{
|
|
"id": "#atcr_appview",
|
|
"type": "AtcrAppView",
|
|
"serviceEndpoint": s.Config.Server.BaseURL,
|
|
},
|
|
},
|
|
}
|
|
|
|
w.Header().Set("Content-Type", "application/did+ld+json")
|
|
w.Header().Set("Cache-Control", "public, max-age=3600")
|
|
w.Header().Set("Access-Control-Allow-Origin", "*")
|
|
if err := json.NewEncoder(w).Encode(doc); err != nil {
|
|
slog.Error("Failed to encode DID document", "error", err)
|
|
}
|
|
}
|
|
|
|
// waitForLeasedWorkers gives the leased background workers a moment to stop and
|
|
// release their leases before the process exits.
|
|
//
|
|
// Skipping this does not lose data (every leased worker is restartable and the
|
|
// leases expire on their own), but it does mean the replacement instance waits
|
|
// out the full TTL before it can take over, so a rolling deploy pauses indexing
|
|
// for a minute instead of a second.
|
|
func (s *AppViewServer) waitForLeasedWorkers() {
|
|
if s.Leases == nil {
|
|
return
|
|
}
|
|
const leaseDrainTimeout = 5 * time.Second
|
|
slog.Info("Waiting for leased workers to release", "component", "leases", "timeout", leaseDrainTimeout)
|
|
s.Leases.Wait(leaseDrainTimeout)
|
|
}
|
|
|
|
// startCleanupWorker runs the periodic expiry sweep under the cleanup lease,
|
|
// after clearing the crew denial backoffs once on acquiring it.
|
|
//
|
|
// That clear used to run on every boot, unconditionally, as
|
|
// "DELETE FROM hold_crew_denials" with no scoping. The intent is sound: a
|
|
// restart usually means a fix has shipped, and users sitting on a backoff of up
|
|
// to an hour should get to retry immediately rather than wait it out. The
|
|
// implementation stopped being sound the moment there was more than one
|
|
// instance, because then a rolling deploy wipes the shared table once per
|
|
// instance and every scale-out event wipes it again.
|
|
//
|
|
// Moving it under the lease keeps the behavior (a deploy still clears the
|
|
// backoffs) while making it happen once. Note the in-memory half of
|
|
// ClearAllDenials was always a no-op at startup, since a fresh process has an
|
|
// empty map; the database wipe was the only thing it ever really did.
|
|
func (s *AppViewServer) startCleanupWorker(ctx context.Context) {
|
|
const cleanupInterval = time.Hour
|
|
authorizer := s.HoldAuthorizer
|
|
|
|
s.Leases.Go(ctx, db.LeaseCleanup, func(leaseCtx context.Context) error {
|
|
if remote, ok := authorizer.(*auth.RemoteHoldAuthorizer); ok {
|
|
if err := remote.ClearAllDenials(); err != nil {
|
|
// Not fatal: the denials carry their own next_retry_at and will
|
|
// lapse on their own.
|
|
slog.Warn("Failed to clear denial caches", "error", err)
|
|
}
|
|
}
|
|
return db.RunPeriodicCleanup(leaseCtx, s.Database, s.SessionStore, cleanupInterval)
|
|
})
|
|
}
|
|
|
|
// initializeJetstream initializes the Jetstream workers for real-time events and backfill.
|
|
//
|
|
// Both run under leases. The consumer must be a singleton for correctness:
|
|
// StatsCache is per-process in-memory state whose aggregate is written to
|
|
// repository_stats as an absolute value, so two consumers each write a partial
|
|
// sum as though it were the whole truth. The webhook dispatcher hangs off the
|
|
// same processor, so a second consumer also doubles every delivery. Backfill is
|
|
// leased because it is expensive and re-reads every user's PDS.
|
|
func (s *AppViewServer) initializeJetstream(ctx context.Context) {
|
|
jetstreamURLs := s.Config.Jetstream.URLs
|
|
|
|
// Explicitly empty URLs disables Jetstream. The YAML config always
|
|
// populates a default list, so the only way to reach this branch is to
|
|
// blank the slice in code (tests, embedded deployments). Without this
|
|
// check, NewWorker silently falls back to the public Bluesky endpoint.
|
|
if len(jetstreamURLs) == 0 {
|
|
slog.Info("Jetstream disabled (no URLs configured)", "component", "jetstream")
|
|
return
|
|
}
|
|
|
|
s.Leases.Go(ctx, db.LeaseJetstream, func(leaseCtx context.Context) error {
|
|
worker := jetstream.NewWorker(s.Database, jetstreamURLs, 0)
|
|
// Set webhook dispatcher on live worker (backfill skips dispatch)
|
|
if s.WebhookDispatcher != nil {
|
|
worker.Processor().SetWebhookDispatcher(s.WebhookDispatcher)
|
|
}
|
|
slog.Info("Jetstream real-time worker started", "component", "jetstream", "endpoints", len(jetstreamURLs))
|
|
// Returns only when leaseCtx is done: either shutdown, or the lease was
|
|
// lost and this instance must stop consuming before the new holder
|
|
// starts. A fresh worker is built on each acquisition so its in-memory
|
|
// StatsCache is never carried across a gap in which another instance
|
|
// was the one indexing.
|
|
worker.StartWithFailover(leaseCtx)
|
|
return leaseCtx.Err()
|
|
})
|
|
|
|
if s.Config.Jetstream.BackfillEnabled {
|
|
relayEndpoints := s.Config.Jetstream.RelayEndpoints
|
|
defaultHoldDID := s.Config.Server.PrimaryHoldDID()
|
|
testMode := s.Config.Server.TestMode
|
|
|
|
backfillWorker, err := jetstream.NewBackfillWorker(s.Database, relayEndpoints, defaultHoldDID, testMode, s.Refresher)
|
|
if err != nil {
|
|
slog.Warn("Failed to create backfill worker", "component", "jetstream/backfill", "error", err)
|
|
} else {
|
|
interval := s.Config.Jetstream.BackfillInterval
|
|
// The startup run and the periodic schedule were two goroutines on
|
|
// context.Background(). They are one leased worker now, so a single
|
|
// instance backfills and a shutdown actually stops it mid-run
|
|
// instead of letting it finish against a closing database.
|
|
s.Leases.Go(ctx, db.LeaseBackfill, func(leaseCtx context.Context) error {
|
|
const startupDelay = 5 * time.Second
|
|
slog.Info("Waiting for services to be ready", "component", "jetstream/backfill", "startup_delay", startupDelay)
|
|
select {
|
|
case <-leaseCtx.Done():
|
|
return leaseCtx.Err()
|
|
case <-time.After(startupDelay):
|
|
}
|
|
|
|
runBackfill := func(reason string) {
|
|
slog.Info("Starting backfill", "component", "jetstream/backfill", "reason", reason, "relay_endpoints", relayEndpoints)
|
|
if err := backfillWorker.Start(leaseCtx); err != nil {
|
|
slog.Warn("Backfill finished with error", "component", "jetstream/backfill", "reason", reason, "error", err)
|
|
} else {
|
|
slog.Info("Backfill completed successfully", "component", "jetstream/backfill", "reason", reason)
|
|
}
|
|
}
|
|
|
|
runBackfill("startup")
|
|
|
|
if interval <= 0 {
|
|
slog.Info("Periodic backfill disabled (interval=0), only startup backfill will run", "component", "jetstream/backfill")
|
|
// Hold the lease rather than returning: releasing it would
|
|
// let another instance acquire and run its own startup
|
|
// backfill, turning "once" into "once per instance".
|
|
<-leaseCtx.Done()
|
|
return leaseCtx.Err()
|
|
}
|
|
|
|
slog.Info("Periodic backfill scheduler started", "component", "jetstream/backfill", "interval", interval)
|
|
ticker := time.NewTicker(interval)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-leaseCtx.Done():
|
|
return leaseCtx.Err()
|
|
case <-ticker.C:
|
|
runBackfill("periodic")
|
|
}
|
|
}
|
|
})
|
|
}
|
|
}
|
|
}
|
|
|
|
// gzipUI compresses text responses. Built once: the options are static, so a
|
|
// failure here is a programming error, not a runtime condition.
|
|
//
|
|
// klauspost/compress is already a direct dependency, so gzip costs no new one.
|
|
// Brotli would save roughly 11 KB more across the three largest assets and
|
|
// would add a runtime dependency, which is not a trade worth making.
|
|
var gzipUI = func() func(http.Handler) http.HandlerFunc {
|
|
wrapper, err := gzhttp.NewWrapper(
|
|
// Allowlist rather than denylist. Anything not named here, including
|
|
// every OCI media type, is passed through uncompressed.
|
|
gzhttp.ContentTypes([]string{
|
|
"text/html",
|
|
"text/css",
|
|
"text/plain",
|
|
"application/javascript",
|
|
"text/javascript",
|
|
"application/json",
|
|
"image/svg+xml",
|
|
}),
|
|
// Below ~1 KB the gzip header costs more than it saves.
|
|
gzhttp.MinSize(1024),
|
|
)
|
|
if err != nil {
|
|
panic("appview: bad gzhttp config: " + err.Error())
|
|
}
|
|
return wrapper
|
|
}()
|
|
|
|
// mountCatalogRefusal registers the /v2/_catalog refusal on r. It must be
|
|
// called on the same router that mounts "/v2/*": chi prefers a static pattern
|
|
// over a wildcard, so these two registrations win regardless of declaration
|
|
// order. The trailing-slash form is spelled out because the distribution router
|
|
// runs with StrictSlash(true) and would otherwise answer it with a 301 to the
|
|
// bare path, costing a round trip to reach the same refusal.
|
|
//
|
|
// Handle registers every method, so HEAD and anything else land here too.
|
|
func mountCatalogRefusal(r chi.Router) {
|
|
r.Handle("/v2/_catalog", http.HandlerFunc(handleCatalogUnsupported))
|
|
r.Handle("/v2/_catalog/", http.HandlerFunc(handleCatalogUnsupported))
|
|
}
|
|
|
|
// handleCatalogUnsupported answers GET /v2/_catalog (and every other method and
|
|
// query string on that path) with an OCI UNSUPPORTED error, HTTP 405.
|
|
//
|
|
// This registry has no global catalog and will not grow one. Repositories live
|
|
// in per-user ATProto PDS namespaces, not in a single enumerable store, so
|
|
// there is nothing for the distribution library to walk: buildStorageConfig()
|
|
// hands it a placeholder inmemory driver, which makes its enumeration empty by
|
|
// construction. Left to itself the library answers a bare request with
|
|
// 200 {"repositories":[]} and 400s any n (Catalog.MaxEntries is 0), so the same
|
|
// endpoint both "works" and rejects a legal parameter. Worse, the empty 200
|
|
// asserts that this registry contains no repositories, which is false and
|
|
// silently so. UNSUPPORTED says the true thing.
|
|
//
|
|
// There is no conformance cost. _catalog is not in the OCI distribution spec at
|
|
// all: it is a Docker Registry HTTP API V2 extension, and the OCI spec places
|
|
// repository discovery out of scope. Docker Hub and GHCR refuse _catalog
|
|
// outright; Quay returns an unconditional empty list. None of them 400s a legal
|
|
// n. Fixing this by setting Catalog.MaxEntries was considered and rejected: it
|
|
// would make the lie coherent rather than remove it.
|
|
//
|
|
// HEAD gets the same status and headers, with the body suppressed by net/http.
|
|
func handleCatalogUnsupported(w http.ResponseWriter, _ *http.Request) {
|
|
writeEmptyAllow(w)
|
|
if err := errcode.ServeJSON(w, errcode.ErrorCodeUnsupported.WithMessage(
|
|
"this registry does not offer a global catalog, repositories are namespaced per user: list tags for a known repository at /v2/{handle-or-did}/{image}/tags/list",
|
|
)); err != nil {
|
|
slog.Error("failed to write OCI error response", "error", err)
|
|
}
|
|
}
|
|
|
|
// writeEmptyAllow sets the Allow header required on a 405 response.
|
|
//
|
|
// RFC 9110 makes this a MUST, in both 15.5.6 and 10.2.1: "An origin server MUST
|
|
// generate an Allow header field in a 405 (Method Not Allowed) response." The
|
|
// OCI UNSUPPORTED error code maps to 405 (its descriptor in the distribution
|
|
// library sets HTTPStatusCode: http.StatusMethodNotAllowed), so every place we
|
|
// serve UNSUPPORTED owes the header.
|
|
//
|
|
// The value is deliberately empty. 10.2.1: "An empty Allow field value
|
|
// indicates that the resource allows no methods." That is precisely the case at
|
|
// both call sites — the global catalog does not exist and never will, and the
|
|
// registry API is not served on the UI domain under any method — so naming a
|
|
// method here would advertise something that does not work. Go writes the field
|
|
// as "Allow: " and clients read it back as present-and-empty.
|
|
//
|
|
// Must be called before ServeJSON, which commits the status line.
|
|
func writeEmptyAllow(w http.ResponseWriter) {
|
|
w.Header().Set("Allow", "")
|
|
}
|
|
|
|
// compressUIResponses gzips UI, static and JSON responses and leaves the OCI
|
|
// registry API at /v2/* alone.
|
|
//
|
|
// Go's net/http compresses nothing on its own, and the appview is served by a
|
|
// load balancer that does not compress either, so style.css went out at 180 KB
|
|
// uncompressed and blocked first paint. gzip takes the HTML, CSS and JS from
|
|
// 388 KB to 72 KB on the wire.
|
|
//
|
|
// The /v2 skip is not just an optimisation. Container layers are already
|
|
// gzipped tarballs, so recompressing them on the registry's hot path burns CPU
|
|
// for no gain. The content-type allowlist would catch most of that on its own,
|
|
// but /v2/* also serves application/json for tag listings and errors, so the
|
|
// path check keeps the registry entirely out of the compression path.
|
|
func compressUIResponses(next http.Handler) http.Handler {
|
|
compressed := gzipUI(next)
|
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
path := r.URL.Path
|
|
if path == "/v2" || path == "/v2/" || strings.HasPrefix(path, "/v2/") {
|
|
next.ServeHTTP(w, r)
|
|
return
|
|
}
|
|
compressed.ServeHTTP(w, r)
|
|
})
|
|
}
|