johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package pds

import (
	"cmp"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"maps"
	"net/http"
	"net/url"
	"strings"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"

	"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/oauth"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/repo"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/serviceauth"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/httpjson"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/opt"
)

const (
	DefaultAccessTokenTTL  = 2 * time.Hour
	DefaultRefreshTokenTTL = 30 * 24 * time.Hour
	DefaultMaxBlobSize     = 5 << 20
	DefaultMaxRequestBytes = 4 << 20
)

type Server struct {
	cfg             Config
	mode            Mode
	repos           repo.RepoStore
	store           *Store
	keys            RepoKeyStore
	resolver        DIDResolver
	plc             PLCSubmitter
	blobs           BlobStore
	seq             *Sequencer
	oauthServer     *oauth.Server
	verifier        *serviceauth.Verifier
	mux             *http.ServeMux
	handler         http.Handler
	sessionSecret   []byte
	serviceDID      string
	audience        string
	accessTTL       time.Duration
	refreshTTL      time.Duration
	maxRequestBytes int64
	maxBlobSize     int64
}

func New(cfg Config) (*Server, error) {
	mode := cfg.Mode
	mode = cmp.Or(mode, ModeFull)
	if mode != ModeFull && mode != ModeRepo {
		return nil, fmt.Errorf("pds: unknown mode %q", mode)
	}
	if cfg.Repos == nil {
		return nil, fmt.Errorf("pds: repo store is required")
	}
	if mode == ModeFull && cfg.Store == nil {
		return nil, fmt.Errorf("pds: account store is required in full mode")
	}
	if mode == ModeFull && cfg.KeyStore == nil {
		return nil, fmt.Errorf("pds: repo key store is required in full mode")
	}
	if mode == ModeRepo && cfg.OAuth != nil {
		return nil, fmt.Errorf("pds: oauth is not available in read-only repo mode")
	}
	secret := append([]byte(nil), cfg.SessionSecret...)
	if mode == ModeFull && len(secret) == 0 {
		return nil, fmt.Errorf("pds: session secret is required in full mode")
	}
	serviceDID := cfg.ServiceDID
	if serviceDID != "" {
		if _, err := syntax.ParseDID(serviceDID); err != nil {
			return nil, fmt.Errorf("pds: invalid service DID: %w", err)
		}
	} else if cfg.PublicBaseURL != "" {
		u, err := url.Parse(cfg.PublicBaseURL)
		if err != nil || u.Host == "" {
			return nil, fmt.Errorf("pds: invalid public base URL")
		}
		serviceDID = "did:web:" + url.QueryEscape(u.Host)
		if _, err := syntax.ParseDID(serviceDID); err != nil {
			return nil, fmt.Errorf("pds: invalid derived service DID: %w", err)
		}
	} else {
		serviceDID = "did:web:localhost"
	}
	audience := cfg.PublicBaseURL
	if serviceDID != "" {
		audience = serviceDID
	}
	audience = cmp.Or(audience, "did:web:localhost")
	accessTTL := cfg.AccessTokenTTL
	accessTTL = opt.Positive(accessTTL, DefaultAccessTokenTTL)
	refreshTTL := cfg.RefreshTokenTTL
	refreshTTL = opt.Positive(refreshTTL, DefaultRefreshTokenTTL)
	maxRequest := cfg.MaxRequestBytes
	maxRequest = opt.Positive(maxRequest, DefaultMaxRequestBytes)
	maxBlob := cfg.MaxBlobSize
	maxBlob = opt.Positive(maxBlob, DefaultMaxBlobSize)
	s := &Server{
		cfg:             cfg,
		mode:            mode,
		repos:           cfg.Repos,
		store:           cfg.Store,
		keys:            cfg.KeyStore,
		plc:             cfg.PLC,
		blobs:           cfg.Blobs,
		sessionSecret:   secret,
		serviceDID:      serviceDID,
		audience:        audience,
		accessTTL:       accessTTL,
		refreshTTL:      refreshTTL,
		maxRequestBytes: maxRequest,
		maxBlobSize:     maxBlob,
		mux:             http.NewServeMux(),
	}
	if mode == ModeRepo {
		s.store = nil
	}
	if cfg.Resolver != nil {
		s.resolver = cfg.Resolver
	} else if cfg.Directory != nil {
		s.resolver = DirectoryResolver{Directory: cfg.Directory}
	}
	if cfg.Directory != nil {
		audiences := append([]string(nil), cfg.ServiceAuthAudiences...)
		if len(audiences) == 0 {
			audiences = []string{audience}
			if cfg.PublicBaseURL != "" && cfg.PublicBaseURL != audience {
				audiences = append(audiences, strings.TrimRight(cfg.PublicBaseURL, "/"))
			}
		}
		s.verifier = &serviceauth.Verifier{Audiences: audiences, Directory: cfg.Directory}
	}
	s.seq = NewSequencer()
	cfg.Repos.SetEventSink(s.seq)
	s.registerRoutes()
	if cfg.OAuth != nil {
		if err := s.initOAuth(cfg.OAuth); err != nil {
			return nil, err
		}
	}
	var handler http.Handler = s.mux
	if s.oauthServer != nil {
		resource := s.oauthServer.ResourceHandler(s.mux)
		handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			if strings.HasPrefix(r.URL.Path, "/oauth/") || strings.HasPrefix(r.URL.Path, "/.well-known/") {
				s.mux.ServeHTTP(w, r)
				return
			}
			resource.ServeHTTP(w, r)
		})
	}
	if mode == ModeRepo {
		handler = s.readOnlyGuard(handler)
	}
	s.handler = handler
	return s, nil
}

func (s *Server) initOAuth(cfg *OAuthConfig) error {
	if s.store == nil {
		return fmt.Errorf("pds: oauth requires an account store")
	}
	issuer := cfg.Issuer
	issuer = cmp.Or(issuer, s.cfg.PublicBaseURL)
	if issuer == "" {
		return fmt.Errorf("pds: oauth issuer is required")
	}
	signer := cfg.Signer
	if signer.DID == "" {
		did, err := syntax.ParseDID(s.serviceDID)
		if err != nil {
			return fmt.Errorf("pds: oauth signer DID is required")
		}
		signer.DID = did
	}
	if signer.Key == nil {
		return fmt.Errorf("pds: oauth signer key is required")
	}
	server, err := oauth.New(oauth.Config{
		Issuer:                  issuer,
		Signer:                  signer,
		Accounts:                s.store,
		Sessions:                s.store,
		Codes:                   s.store,
		Tokens:                  s.store,
		PARs:                    s.store,
		Nonces:                  oauth.NewMemoryNonceStore(),
		HTTPClient:              cfg.HTTPClient,
		AccessTokenTTL:          cfg.AccessTokenTTL,
		RefreshTokenTTL:         cfg.RefreshTokenTTL,
		AuthorizationCodeTTL:    cfg.AuthorizationCodeTTL,
		PARTTL:                  cfg.PARTTL,
		AllowMissingDPoPNonce:   cfg.AllowMissingDPoPNonce,
		DisableLocalhostClients: cfg.DisableLocalhostClients,
		Scopes:                  append([]string(nil), cfg.Scopes...),
	})
	if err != nil {
		return fmt.Errorf("pds: initializing oauth: %w", err)
	}
	server.Register(s.mux)
	s.oauthServer = server
	return nil
}

func (s *Server) registerRoutes() {
	s.mux.HandleFunc("GET /_health", s.handleHealth)
	s.mux.HandleFunc("GET /health", s.handleHealth)
	s.mux.HandleFunc("GET /xrpc/_health", s.handleHealth)
	s.mux.HandleFunc("GET /.well-known/atproto-did", s.handleWellKnownDID)
	s.mux.HandleFunc("GET /.well-known/did.json", s.handleWellKnownDIDDocument)

	s.mux.HandleFunc("GET /xrpc/com.atproto.server.describeServer", s.handleDescribeServer)
	s.mux.HandleFunc("POST /xrpc/com.atproto.server.createAccount", s.handleCreateAccount)
	s.mux.HandleFunc("POST /xrpc/com.atproto.server.createSession", s.handleCreateSession)
	s.mux.HandleFunc("POST /xrpc/com.atproto.server.refreshSession", s.handleRefreshSession)
	s.mux.HandleFunc("POST /xrpc/com.atproto.server.deleteSession", s.handleDeleteSession)
	s.mux.HandleFunc("GET /xrpc/com.atproto.server.getSession", s.requireSession(s.handleGetSession))
	s.mux.HandleFunc("POST /xrpc/com.atproto.server.createAppPassword", s.requireSession(s.handleCreateAppPassword))
	s.mux.HandleFunc("GET /xrpc/com.atproto.server.listAppPasswords", s.requireSession(s.handleListAppPasswords))
	s.mux.HandleFunc("POST /xrpc/com.atproto.server.revokeAppPassword", s.requireSession(s.handleRevokeAppPassword))
	s.mux.HandleFunc("GET /xrpc/com.atproto.server.getServiceAuth", s.requireSession(s.handleGetServiceAuth))
	s.mux.HandleFunc("POST /xrpc/com.atproto.server.getServiceAuth", s.requireSession(s.handleGetServiceAuth))

	s.mux.HandleFunc("GET /xrpc/com.atproto.repo.getRecord", s.handleGetRecord)
	s.mux.HandleFunc("GET /xrpc/com.atproto.repo.listRecords", s.handleListRecords)
	s.mux.HandleFunc("GET /xrpc/com.atproto.repo.describeRepo", s.handleDescribeRepo)
	s.mux.HandleFunc("POST /xrpc/com.atproto.repo.createRecord", s.requireSession(s.handleCreateRecord))
	s.mux.HandleFunc("POST /xrpc/com.atproto.repo.putRecord", s.requireSession(s.handlePutRecord))
	s.mux.HandleFunc("POST /xrpc/com.atproto.repo.deleteRecord", s.requireSession(s.handleDeleteRecord))
	s.mux.HandleFunc("POST /xrpc/com.atproto.repo.applyWrites", s.requireSession(s.handleApplyWrites))
	s.mux.HandleFunc("POST /xrpc/com.atproto.repo.uploadBlob", s.requireSession(s.handleUploadBlob))

	s.mux.HandleFunc("GET /xrpc/com.atproto.sync.getRepo", s.handleGetRepo)
	s.mux.HandleFunc("GET /xrpc/com.atproto.sync.getLatestCommit", s.handleGetLatestCommit)
	s.mux.HandleFunc("GET /xrpc/com.atproto.sync.getRepoStatus", s.handleGetRepoStatus)
	s.mux.HandleFunc("GET /xrpc/com.atproto.repo.getRepoStatus", s.handleGetRepoStatus)
	s.mux.HandleFunc("GET /xrpc/com.atproto.sync.getRecord", s.handleSyncGetRecord)
	s.mux.HandleFunc("GET /xrpc/com.atproto.sync.getBlob", s.handleGetBlob)
	s.mux.HandleFunc("GET /xrpc/com.atproto.sync.subscribeRepos", s.handleSubscribeRepos)

	s.mux.HandleFunc("/xrpc/", s.handleMethodNotFound)
}

var readOnlyBlocked = map[string]bool{
	"/xrpc/com.atproto.server.createAccount":     true,
	"/xrpc/com.atproto.server.createSession":     true,
	"/xrpc/com.atproto.server.refreshSession":    true,
	"/xrpc/com.atproto.server.deleteSession":     true,
	"/xrpc/com.atproto.server.getSession":        true,
	"/xrpc/com.atproto.server.createAppPassword": true,
	"/xrpc/com.atproto.server.listAppPasswords":  true,
	"/xrpc/com.atproto.server.revokeAppPassword": true,
	"/xrpc/com.atproto.server.getServiceAuth":    true,
	"/xrpc/com.atproto.repo.createRecord":        true,
	"/xrpc/com.atproto.repo.putRecord":           true,
	"/xrpc/com.atproto.repo.deleteRecord":        true,
	"/xrpc/com.atproto.repo.applyWrites":         true,
	"/xrpc/com.atproto.repo.uploadBlob":          true,
}

func (s *Server) readOnlyGuard(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if readOnlyBlocked[r.URL.Path] {
			writeXRPCError(w, http.StatusForbidden, "Forbidden", "PDS is running in read-only repo mode")
			return
		}
		next.ServeHTTP(w, r)
	})
}

func (s *Server) Handler() http.Handler {
	return s.handler
}

func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	s.handler.ServeHTTP(w, r)
}

func (s *Server) Mode() Mode {
	return s.mode
}

func (s *Server) Sequencer() *Sequencer {
	return s.seq
}

func (s *Server) Repos() repo.RepoStore {
	return s.repos
}

func (s *Server) Store() *Store {
	return s.store
}

func (s *Server) KeyStore() RepoKeyStore {
	return s.keys
}

func (s *Server) OAuthServer() *oauth.Server {
	return s.oauthServer
}

func (s *Server) PublicBaseURL() string {
	return s.cfg.PublicBaseURL
}

func (s *Server) Version() string {
	return s.cfg.Version
}

func (s *Server) AvailableUserDomains() []string {
	return append([]string(nil), s.cfg.AvailableUserDomains...)
}

func (s *Server) ServiceDID() string {
	return s.serviceDID
}

func (s *Server) UpdateService(ctx context.Context, did syntax.DID, service string) error {
	if s.plc == nil {
		return ErrUnsupported
	}
	return s.plc.UpdateService(ctx, did, service)
}

func (s *Server) handleHealth(w http.ResponseWriter, _ *http.Request) {
	version := s.cfg.Version
	version = cmp.Or(version, "socialweb-pds")
	httpjson.Write(w, http.StatusOK, map[string]string{"version": version})
}

func (s *Server) handleMethodNotFound(w http.ResponseWriter, _ *http.Request) {
	writeXRPCError(w, http.StatusNotFound, "MethodNotFound", "unknown xrpc method")
}

type xrpcErrorBody struct {
	Error   string `json:"error"`
	Message string `json:"message,omitempty"`
}

func writeXRPCError(w http.ResponseWriter, status int, code, message string) {
	httpjson.Write(w, status, xrpcErrorBody{Error: code, Message: message})
}

func (s *Server) decodeJSON(w http.ResponseWriter, r *http.Request, dst any) error {
	r.Body = http.MaxBytesReader(w, r.Body, s.maxRequestBytes)
	dec := json.NewDecoder(r.Body)
	if err := dec.Decode(dst); err != nil {
		if errors.Is(err, io.EOF) {
			return nil
		}
		return err
	}
	return nil
}

func (s *Server) repoError(w http.ResponseWriter, err error) {
	switch {
	case errors.Is(err, repo.ErrInvalidRequest):
		writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", err.Error())
	case errors.Is(err, repo.ErrInvalidSwap):
		writeXRPCError(w, http.StatusBadRequest, "InvalidSwap", err.Error())
	case errors.Is(err, repo.ErrRecordNotFound):
		writeXRPCError(w, http.StatusBadRequest, "RecordNotFound", err.Error())
	case errors.Is(err, repo.ErrRecordExists):
		writeXRPCError(w, http.StatusBadRequest, "RecordAlreadyExists", err.Error())
	case errors.Is(err, repo.ErrRepoNotFound), errors.Is(err, ErrAccountNotFound):
		writeXRPCError(w, http.StatusBadRequest, "RepoNotFound", "repo not found")
	case errors.Is(err, repo.ErrReadOnly):
		writeXRPCError(w, http.StatusForbidden, "Forbidden", "repo is read-only")
	default:
		writeXRPCError(w, http.StatusInternalServerError, "InternalServerError", "internal server error")
	}
}

func (s *Server) openRepo(ctx context.Context, did syntax.DID) (repo.Repo, error) {
	if s.store == nil {
		return nil, repo.ErrReadOnly
	}
	if s.keys == nil {
		return nil, repo.ErrReadOnly
	}
	key, err := s.keys.GetRepoKey(ctx, did)
	if err != nil {
		return nil, fmt.Errorf("pds: loading repo signing key: %w", err)
	}
	return s.repos.OpenRepo(ctx, did, key)
}

func recordType(record map[string]any, collection string) map[string]any {
	if record == nil {
		return nil
	}
	if _, ok := record["$type"]; ok {
		return record
	}
	out := make(map[string]any, len(record)+1)
	out["$type"] = collection
	maps.Copy(out, record)
	return out
}