at-container-registry/pkg/appview/authgate/push_authorizer.go

// Package authgate implements the auth-phase gate for ATCR's registry JWT
// issuance. The gate runs once per token request (push or pull): it always
// reconciles crew membership so first-time CLI users on a private hold
// can pass the hold-side read check, and additionally enforces hold
// membership (captain or crew with blob:write) plus storage quota for
// non-wildcard push scopes. Once the JWT is signed, the registry hot path
// trusts it for its short lifetime — no per-/v2/-request re-authorization.
package authgate

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"slices"

	"atcr.io/pkg/appview/storage"
	"atcr.io/pkg/atproto"
	"atcr.io/pkg/auth"
	"atcr.io/pkg/auth/oauth"
	"atcr.io/pkg/auth/token"
)

// Authorizer enforces hold membership and quota at /auth/token.
//
// Captain/crew checks are served from local Jetstream-fed tables
// (hold_captain_records, hold_crew_members) so they're sub-millisecond.
// Quota is a live GET to hold's public io.atcr.hold.getQuota endpoint,
// which fails open on network errors so a brief hold outage doesn't lock
// users out of pushing entirely.
type Authorizer struct {
	holdResolver
	holdAuthorizer auth.HoldAuthorizer
	refresher      *oauth.Refresher
	httpClient     *http.Client
}

// Option configures an Authorizer. Use New(..., WithX(...)) to set them.
type Option func(*Authorizer)

// WithHTTPClient overrides the HTTP client used for the live quota call.
// Production wiring leaves this at http.DefaultClient; tests pass an
// httptest.Server-backed client to drive checkQuota deterministically.
func WithHTTPClient(c *http.Client) Option {
	return func(a *Authorizer) { a.httpClient = c }
}

// New constructs an Authorizer. defaultHoldDID is the AppView's fallback
// hold (used when a user's sailor profile hasn't recorded one yet). The
// refresher is required for OAuth-flow service-token minting during
// EnsureCrewMembership reconciliation; pass nil if OAuth isn't configured.
func New(db *sql.DB, holdAuthorizer auth.HoldAuthorizer, refresher *oauth.Refresher, defaultHoldDID string, opts ...Option) *Authorizer {
	a := &Authorizer{
		holdResolver:   holdResolver{db: db, defaultHoldDID: defaultHoldDID},
		holdAuthorizer: holdAuthorizer,
		refresher:      refresher,
		httpClient:     http.DefaultClient,
	}
	for _, opt := range opts {
		opt(a)
	}
	return a
}

// Authorize satisfies token.Authorizer.
//
// authMethod is one of token.AuthMethodOAuth or token.AuthMethodAppPassword
// and selects which service-token fetcher EnsureCrewMembership uses for
// reconciliation. We can't infer it from `did` alone because either flow
// can produce a valid session for the same identity.
//
// Crew reconciliation runs for both push and pull token requests so a
// first-time CLI user (especially OAuth/credential-helper) doesn't get
// 403'd on their first pull from a private hold. The remaining gates
// (membership requirement, quota) only apply to non-wildcard push.
func (a *Authorizer) Authorize(ctx context.Context, did, authMethod string, access []auth.AccessEntry) error {
	holdDID, err := a.resolveHoldDID(ctx, did)
	if err != nil {
		return err
	}
	if holdDID == "" {
		// No hold configured anywhere. Registry can't function without one,
		// but that's a config problem — not the gate's job to surface.
		return nil
	}

	// Captain check first — they own the hold, so they have all permissions
	// (no reconciliation needed, no membership requirement to check). This
	// is the common case (most users push to their own hold) and lets us
	// skip both the EnsureCrewMembership reconciliation and the
	// hold_crew_members lookup below.
	captain, err := a.isCaptain(ctx, did, holdDID)
	if err != nil {
		return err
	}

	if !captain {
		// Reconcile crew membership for first-time CLI users (push or pull).
		// Idempotent and short-circuits via approval cache when the record is
		// already known. Best-effort: errors are logged inside the helper.
		if fetcher := a.serviceTokenFetcher(ctx, authMethod, did); fetcher != nil {
			storage.EnsureCrewMembership(ctx, did, holdDID, a.holdAuthorizer, fetcher)
		}
	}

	if !hasNonWildcardPushScope(access) {
		// Pull-only or wildcard scope — reconciliation done above is enough.
		// Membership requirement and quota only apply to push.
		return nil
	}

	if !captain {
		if err := a.checkCrewBlobWrite(ctx, did, holdDID); err != nil {
			return err
		}
	}

	if err := a.checkQuota(ctx, did, holdDID); err != nil {
		// Over quota. Denying outright would also deny the delete our own
		// error text tells the user to perform, and OCI clients bundle the
		// two (docker and crane both request pull,push,delete for a delete),
		// so an over-quota user has no way to free space. Grant the non-push
		// subset when delete was asked for. A plain push is still denied, so
		// the quota message still reaches the client that needs to see it.
		if dropPushForDelete(access) {
			slog.Info("push gate: over quota, granting delete without push",
				"did", did, "hold_did", holdDID, "reason", err)
			return nil
		}
		return err
	}
	return nil
}

// dropPushForDelete removes the push action from every repository entry when
// the request also asks for delete, reporting whether it did.
//
// The narrowing is in place because the caller passes this same slice to the
// JWT issuer, so the reduced action set is what the token actually grants.
// See the Authorizer contract in pkg/auth/token.
func dropPushForDelete(access []auth.AccessEntry) bool {
	requestsDelete := false
	for _, entry := range access {
		if entry.Type == "repository" && entry.Name != "*" && slices.Contains(entry.Actions, "delete") {
			requestsDelete = true
			break
		}
	}
	if !requestsDelete {
		return false
	}

	for i := range access {
		if access[i].Type != "repository" {
			continue
		}
		// Clone first: DeleteFunc compacts in place, so filtering the stored
		// slice directly would scramble any other holder of that same array.
		access[i].Actions = slices.DeleteFunc(
			slices.Clone(access[i].Actions),
			func(action string) bool { return action == "push" },
		)
	}
	return true
}

// isCaptain returns true if userDID owns holdDID per the local Jetstream-
// fed hold_captain_records table. Returns (false, nil) if no captain record
// exists yet (hold not yet ingested, or doesn't exist).
func (a *Authorizer) isCaptain(ctx context.Context, userDID, holdDID string) (bool, error) {
	var ownerDID string
	err := a.db.QueryRowContext(ctx,
		"SELECT owner_did FROM hold_captain_records WHERE hold_did = ?", holdDID,
	).Scan(&ownerDID)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, fmt.Errorf("look up hold captain: %w", err)
	}
	return ownerDID == userDID, nil
}

func hasNonWildcardPushScope(access []auth.AccessEntry) bool {
	for _, entry := range access {
		if entry.Type != "repository" || entry.Name == "*" {
			continue
		}
		if slices.Contains(entry.Actions, "push") {
			return true
		}
	}
	return false
}

// checkCrewBlobWrite returns nil if userDID is a crew member of holdDID with
// blob:write permission. Caller must have already determined that userDID is
// not the captain (captains have all permissions implicitly).
func (a *Authorizer) checkCrewBlobWrite(ctx context.Context, userDID, holdDID string) error {
	var permsJSON sql.NullString
	err := a.db.QueryRowContext(ctx,
		"SELECT permissions FROM hold_crew_members WHERE hold_did = ? AND member_did = ?",
		holdDID, userDID,
	).Scan(&permsJSON)
	if errors.Is(err, sql.ErrNoRows) {
		return fmt.Errorf("crew membership required: %s is not a member of hold %s", userDID, holdDID)
	}
	if err != nil {
		return fmt.Errorf("look up crew membership: %w", err)
	}

	if !permissionsAllowBlobWrite(permsJSON.String) {
		return fmt.Errorf("crew membership lacks blob:write on hold %s", holdDID)
	}
	return nil
}

func permissionsAllowBlobWrite(permsJSON string) bool {
	if permsJSON == "" {
		return false
	}
	var perms []string
	if err := json.Unmarshal([]byte(permsJSON), &perms); err != nil {
		return false
	}
	return slices.Contains(perms, "blob:write")
}

// checkQuota fails open on network errors. The hold's quota endpoint is
// public (no auth), but if the hold is unreachable we don't want every
// push everywhere to error out — Docker users would see denied:quota for
// reasons unrelated to their actual usage.
func (a *Authorizer) checkQuota(ctx context.Context, userDID, holdDID string) error {
	stats, err := atproto.FetchQuotaStats(ctx, a.httpClient, holdDID, userDID)
	if err != nil {
		slog.Warn("push gate: quota call failed; allowing push", "did", userDID, "hold_did", holdDID, "error", err)
		return nil
	}

	if stats.Limit != nil && stats.TotalSize >= *stats.Limit {
		return fmt.Errorf("quota exceeded: %s / %s used by %s. Delete images to free space",
			formatGB(stats.TotalSize), formatGB(*stats.Limit), a.identityLabel(ctx, userDID))
	}
	return nil
}

// formatGB renders bytes as gigabytes with two decimals — matches the
// percentage display in the appview UI so users see consistent units.
func formatGB(b int64) string {
	return fmt.Sprintf("%.2f GB", float64(b)/(1024*1024*1024))
}

// identityLabel returns "handle (did)" when the local users table knows the
// handle, falling back to the bare DID. The users table is Jetstream-fed, so
// for first-seen identities (rare on a push, since they've already OAuth'd)
// the handle may be missing — the DID alone still uniquely identifies them.
func (a *Authorizer) identityLabel(ctx context.Context, did string) string {
	var handle string
	err := a.db.QueryRowContext(ctx, "SELECT handle FROM users WHERE did = ?", did).Scan(&handle)
	if err != nil || handle == "" || handle == did {
		return did
	}
	return fmt.Sprintf("%s (%s)", handle, did)
}

// serviceTokenFetcher returns the appropriate fetcher for the auth method,
// or nil if none is available (e.g. OAuth flow with no refresher configured).
// ctx is used for the cold-cache identity lookup inside resolvePDS.
func (a *Authorizer) serviceTokenFetcher(ctx context.Context, authMethod, userDID string) storage.ServiceTokenFetcher {
	pdsEndpoint, err := a.resolvePDS(ctx, userDID)
	if err != nil || pdsEndpoint == "" {
		return nil
	}

	switch authMethod {
	case token.AuthMethodAppPassword:
		return func(ctx context.Context, holdDID string) (string, error) {
			return auth.GetOrFetchServiceTokenWithAppPassword(ctx, userDID, holdDID, pdsEndpoint)
		}
	case token.AuthMethodOAuth:
		if a.refresher == nil {
			return nil
		}
		return func(ctx context.Context, holdDID string) (string, error) {
			return auth.GetOrFetchServiceToken(ctx, a.refresher, userDID, holdDID, pdsEndpoint)
		}
	default:
		return nil
	}
}

func (a *Authorizer) resolvePDS(ctx context.Context, userDID string) (string, error) {
	// Cached identity directory lookup (24h TTL). Per-call cost <1ms after
	// the first resolution per user; on a cold miss this issues live DNS/
	// HTTPS lookups, so we honor ctx so a client disconnect cancels them.
	_, _, pdsEndpoint, err := atproto.ResolveIdentity(ctx, userDID)
	if err != nil {
		return "", err
	}
	return pdsEndpoint, nil
}