Something went wrong. Try again.
An ATProto Powered Reputation Network
Something went wrong. Try again.
11 kB · 301 lines
Go
at master
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302package appview
import ( "crypto/subtle" "database/sql" "errors" "log/slog" "net/http" "net/url" "strconv" "strings"
"provenance/internal/atproto" "provenance/internal/domain" "provenance/internal/httpx" "provenance/internal/logging" "provenance/internal/trust")
const IdentityHeader = "X-Provenance-DID"const metricsMediaType = "text/plain"
type HTTP struct { service *Service key string parameters map[string][]parameter}
func Handler(service *Service, key string) http.Handler { h := &HTTP{service: service, key: key, parameters: map[string][]parameter{}} mux := http.NewServeMux() health := func(w http.ResponseWriter, r *http.Request) { httpx.JSON(w, http.StatusOK, map[string]string{"status": "ok"}) } ready := func(w http.ResponseWriter, r *http.Request) { status := http.StatusOK if !service.index.snapshot(service.now()).Ready { status = http.StatusServiceUnavailable } httpx.JSON(w, status, map[string]bool{"ready": status == http.StatusOK}) } indexing := func(w http.ResponseWriter, r *http.Request) { httpx.JSON(w, http.StatusOK, service.index.snapshot(service.now())) } descriptor := service.algorithm.Describe() threshold := objectSchema(map[string]any{"threshold": map[string]any{"type": "number", "minimum": domain.MinScore, "maximum": domain.MaxScore}}, "threshold") author := queryParameter("author", "string") author.Required = true offset := queryParameter("offset", "integer") offset.Schema["minimum"] = 0 prefix := queryParameter("prefix", "string") prefix.Schema["maxLength"] = domain.MaxTagLength endpoints := []endpoint{ {method: http.MethodGet, path: "/healthz", handler: health, response: objectSchema(map[string]any{"status": scalarSchema("string")}, "status")}, {method: http.MethodGet, path: "/readyz", handler: ready, response: objectSchema(map[string]any{"ready": scalarSchema("boolean")}, "ready")}, {method: http.MethodGet, path: "/metrics", handler: service.index.handler().ServeHTTP, response: scalarSchema("string"), responseMedia: metricsMediaType}, {method: http.MethodGet, path: "/v1/indexing", handler: indexing, private: true, response: scalarSchema("object")}, {method: http.MethodGet, path: "/v1/trust", handler: h.reputation, parameters: reputationParams(descriptor, "/v1/trust"), response: schemaRef("Result")}, {method: http.MethodGet, path: "/v1/reputation", handler: h.aggregate, parameters: aggregateParams(descriptor, "/v1/reputation"), response: schemaRef("Reputation")}, {method: http.MethodGet, path: "/v1/tags", handler: h.tags, parameters: []parameter{prefix}, response: arraySchema(scalarSchema("string"))}, {method: http.MethodGet, path: "/v1/vouches", handler: h.vouches, parameters: []parameter{author, offset}, response: arraySchema(schemaRef("Vouch"))}, {method: http.MethodPost, path: "/v1/participants", handler: h.join, private: true, response: objectSchema(map[string]any{"joined": scalarSchema("boolean")}, "joined")}, {method: http.MethodPost, path: "/v1/sync", handler: h.sync, private: true, response: objectSchema(map[string]any{"synced": scalarSchema("boolean")}, "synced")}, {method: http.MethodGet, path: "/v1/me", handler: h.config, private: true, response: schemaRef("Participant")}, {method: http.MethodPut, path: "/v1/me/threshold", handler: h.threshold, private: true, body: threshold, response: threshold}, {method: http.MethodGet, path: "/v1/me/trust", handler: h.relative, private: true, parameters: reputationParams(descriptor, "/v1/me/trust"), response: schemaRef("Evaluation")}, {method: http.MethodGet, path: "/v1/me/reputation", handler: h.aggregate, private: true, parameters: aggregateParams(descriptor, "/v1/me/reputation"), response: schemaRef("Reputation")}, {method: http.MethodGet, path: "/v1/trusted", handler: h.trusted, private: true, parameters: reputationParams(descriptor, "/v1/trusted"), response: objectSchema(map[string]any{"trusted": scalarSchema("boolean")}, "trusted")}, } for _, e := range endpoints { h.parameters[e.path] = e.parameters handler := e.handler if e.private { handler = h.protect(handler) } mux.HandleFunc(e.method+" "+e.path, handler) } endpoints = append(endpoints, endpoint{method: http.MethodGet, path: "/v1/openapi.json", response: scalarSchema("object")}) document := apiDocument(endpoints, descriptor) document["x-provenance-reputation-algorithm"] = service.reputation.Describe().Name document["x-provenance-soft-vouch-enabled"] = service.softMode != domain.SoftDisabled document["x-provenance-soft-vouch-all-authors"] = domain.SoftAll(service.softMode) document["x-provenance-soft-vouch-lists"] = domain.SoftLists(service.softMode) document["x-provenance-max-depth"] = trust.MaxDepth mux.HandleFunc("GET /v1/openapi.json", func(w http.ResponseWriter, r *http.Request) { httpx.JSON(w, http.StatusOK, document) }) return service.index.instrument(mux)}
func (h *HTTP) protect(next http.HandlerFunc) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { authorization := r.Header.Get("Authorization") provided := strings.TrimPrefix(authorization, "Bearer ") if h.key == "" || !strings.HasPrefix(authorization, "Bearer ") || subtle.ConstantTimeCompare([]byte(provided), []byte(h.key)) != 1 || atproto.ValidDID(r.Header.Get(IdentityHeader)) != nil { httpx.Error(w, http.StatusUnauthorized, "authentication is required") return } next(w, r) }}
func query(r *http.Request) (string, domain.Mode) { tag := r.URL.Query().Get("tag") mode := domain.Mode(r.URL.Query().Get("mode")) if mode == "" { mode = domain.Default if tag != "" { mode = domain.Tagged } } return tag, mode}
func (h *HTTP) reputation(w http.ResponseWriter, r *http.Request) { values, err := url.ParseQuery(r.URL.RawQuery) if err != nil { httpx.Error(w, http.StatusBadRequest, "invalid query encoding") return } options, err := validateQuery(values, h.parameters[r.URL.Path]) if err != nil { httpx.Error(w, http.StatusBadRequest, err.Error()) return } tag, mode := query(r) result, err := h.service.Reputation(r.Context(), r.URL.Query().Get("source"), r.URL.Query().Get("target"), tag, mode, options) if err != nil { h.fail(w, r, err) return } if r.URL.Query().Get("include_path") != "true" { result.Path = nil result.PathScores = nil } httpx.JSON(w, http.StatusOK, result)}
func (h *HTTP) evaluate(r *http.Request) (Evaluation, error) { values, err := url.ParseQuery(r.URL.RawQuery) if err != nil { return Evaluation{}, err } options, err := validateQuery(values, h.parameters[r.URL.Path]) if err != nil { return Evaluation{}, err } tag, mode := query(r) return h.service.Evaluate(r.Context(), r.Header.Get(IdentityHeader), r.URL.Query().Get("target"), tag, mode, options)}
func (h *HTTP) relative(w http.ResponseWriter, r *http.Request) { result, err := h.evaluate(r) if err != nil { h.fail(w, r, err) return } if !result.Trusted && r.URL.Query().Get("include_path") != "true" { result.Path = nil result.PathScores = nil } httpx.JSON(w, http.StatusOK, result)}
func (h *HTTP) trusted(w http.ResponseWriter, r *http.Request) { result, err := h.evaluate(r) if err != nil { h.fail(w, r, err) return } httpx.JSON(w, http.StatusOK, map[string]bool{"trusted": result.Trusted})}
func (h *HTTP) aggregate(w http.ResponseWriter, r *http.Request) { values, err := url.ParseQuery(r.URL.RawQuery) if err != nil { httpx.Error(w, http.StatusBadRequest, "invalid query encoding") return } options, err := validateQuery(values, h.parameters[r.URL.Path]) if err != nil { httpx.Error(w, http.StatusBadRequest, err.Error()) return } source := values.Get("source") threshold := domain.DefaultThreshold if r.URL.Path == "/v1/me/reputation" { source = r.Header.Get(IdentityHeader) p, err := h.service.Config(r.Context(), source) if err != nil { h.fail(w, r, err) return } threshold = p.Threshold } if raw, exists := values["threshold"]; exists { threshold, err = strconv.ParseFloat(raw[0], 64) if err == nil { err = domain.ValidateThreshold(threshold) } if err != nil { httpx.Error(w, http.StatusBadRequest, "invalid threshold") return } } tag, mode := query(r) result, err := h.service.RelativeReputation(r.Context(), source, values.Get("target"), tag, mode, threshold, options) if err != nil { h.fail(w, r, err) return } httpx.JSON(w, http.StatusOK, result)}
func (h *HTTP) tags(w http.ResponseWriter, r *http.Request) { tags, err := h.service.Tags(r.Context(), r.URL.Query().Get("prefix")) if err != nil { h.fail(w, r, err) return } httpx.JSON(w, http.StatusOK, tags)}
func (h *HTTP) vouches(w http.ResponseWriter, r *http.Request) { offset := 0 if raw := r.URL.Query().Get("offset"); raw != "" { var err error offset, err = strconv.Atoi(raw) if err != nil { httpx.Error(w, http.StatusBadRequest, "invalid offset") return } } vouches, err := h.service.Vouches(r.Context(), r.URL.Query().Get("author"), offset) if err != nil { h.fail(w, r, err) return } httpx.JSON(w, http.StatusOK, vouches)}
func (h *HTTP) join(w http.ResponseWriter, r *http.Request) { if err := h.service.Join(r.Context(), r.Header.Get(IdentityHeader)); err != nil { h.fail(w, r, err) return } httpx.JSON(w, http.StatusOK, map[string]bool{"joined": true})}
func (h *HTTP) sync(w http.ResponseWriter, r *http.Request) { if err := h.service.Sync(r.Context(), r.Header.Get(IdentityHeader)); err != nil { h.fail(w, r, err) return } httpx.JSON(w, http.StatusOK, map[string]bool{"synced": true})}
func (h *HTTP) config(w http.ResponseWriter, r *http.Request) { p, err := h.service.Config(r.Context(), r.Header.Get(IdentityHeader)) if err != nil { h.fail(w, r, err) return } httpx.JSON(w, http.StatusOK, p)}
func (h *HTTP) threshold(w http.ResponseWriter, r *http.Request) { var body struct { Threshold *float64 `json:"threshold"` } if err := httpx.Decode(w, r, &body); err != nil || body.Threshold == nil { httpx.Error(w, http.StatusBadRequest, "threshold is required") return } if err := domain.ValidateThreshold(*body.Threshold); err != nil { httpx.Error(w, http.StatusBadRequest, err.Error()) return } if err := h.service.SetThreshold(r.Context(), r.Header.Get(IdentityHeader), *body.Threshold); err != nil { h.fail(w, r, err) return } httpx.JSON(w, http.StatusOK, map[string]float64{"threshold": *body.Threshold})}
func (h *HTTP) fail(w http.ResponseWriter, r *http.Request, err error) { status := http.StatusBadRequest if errors.Is(err, trust.ErrLimit) { status = http.StatusServiceUnavailable } else if errors.Is(err, sql.ErrNoRows) { status = http.StatusNotFound } slog.Warn("appview request failed", "component", "http", "method", r.Method, "path", r.URL.Path, "status", status, "error", logging.Error(err)) if errors.Is(err, trust.ErrLimit) { httpx.Error(w, http.StatusServiceUnavailable, err.Error()) return } if errors.Is(err, sql.ErrNoRows) { httpx.Error(w, http.StatusNotFound, "participant does not exist") return } httpx.Error(w, http.StatusBadRequest, "request failed. Verify the identifiers and tag mode, or retry after the index sync.")}