package server import ( "errors" "fmt" "io" "net/http" "strconv" "strings" "time" "cc.tiredsysadmin.keeporg/internal/blob" "cc.tiredsysadmin.keeporg/internal/store" ) // dbView is what a client learns about a database it has access to. type dbView struct { ID string `json:"id"` Name string `json:"name"` OwnerGroupID string `json:"owner_group_id"` Perm store.Perm `json:"perm"` MergeStrategy store.Strategy `json:"merge_strategy"` Version int64 `json:"version"` SHA256 string `json:"sha256"` Size int64 `json:"size"` UpdatedAt time.Time `json:"updated_at"` UpdatedBy string `json:"updated_by,omitempty"` } func toView(db *store.Database, p store.Perm) dbView { return dbView{ID: db.ID, Name: db.Name, OwnerGroupID: db.OwnerGroupID, Perm: p, MergeStrategy: db.MergeStrategy, Version: db.Version, SHA256: db.SHA256, Size: db.Size, UpdatedAt: db.UpdatedAt, UpdatedBy: db.UpdatedBy} } func (s *Server) handleListDatabases(w http.ResponseWriter, r *http.Request, u *store.User) { out := []dbView{} s.st.Read(func(st *store.State) { for _, a := range st.Assignments(u.ID) { out = append(out, toView(a.DB, a.Perm)) } }) writeJSON(w, http.StatusOK, map[string]any{"databases": out}) } // handleGetBlob serves the current version of a database, or a historical one // when ?version= is given. func (s *Server) handleGetBlob(w http.ResponseWriter, r *http.Request, u *store.User) { id := r.PathValue("id") var ( db store.Database perm store.Perm ) s.st.Read(func(st *store.State) { perm = st.DatabasePerm(u.ID, id) if d := st.Databases[id]; d != nil { db = *d } }) if !perm.AtLeast(store.PermRead) { fail(w, http.StatusNotFound, "not_found", "no such database") return } version := db.Version if q := r.URL.Query().Get("version"); q != "" { v, err := strconv.ParseInt(q, 10, 64) if err != nil { fail(w, http.StatusBadRequest, "bad_request", "version must be an integer") return } version = v } if version == 0 { // A database that has never been written has no bytes yet. w.Header().Set("ETag", `"0"`) w.Header().Set("Content-Type", "application/octet-stream") w.WriteHeader(http.StatusNoContent) return } data, err := s.blobs.Get(id, version) if err != nil { fail(w, http.StatusNotFound, "not_found", "that version is no longer stored") return } w.Header().Set("ETag", etag(version)) w.Header().Set("X-KeepOrg-SHA256", blob.SHA256(data)) w.Header().Set("X-KeepOrg-Version", strconv.FormatInt(version, 10)) w.Header().Set("Content-Type", "application/octet-stream") w.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=%q", db.Name+".kdbx")) w.Header().Set("Content-Length", strconv.Itoa(len(data))) _, _ = w.Write(data) } // handlePutBlob stores a new version. The client states the version it edited // in If-Match; the stored merge strategy decides what happens if that is no // longer current. func (s *Server) handlePutBlob(w http.ResponseWriter, r *http.Request, u *store.User) { id := r.PathValue("id") var ( perm store.Perm current int64 strategy store.Strategy exists bool ) s.st.Read(func(st *store.State) { perm = st.DatabasePerm(u.ID, id) if d := st.Databases[id]; d != nil { exists, current, strategy = true, d.Version, d.MergeStrategy } }) if !exists || !perm.AtLeast(store.PermRead) { fail(w, http.StatusNotFound, "not_found", "no such database") return } if !perm.AtLeast(store.PermWrite) { fail(w, http.StatusForbidden, "forbidden", "you have read-only access to this database") return } v := decidePush(current, r.Header.Get("If-Match"), strategy) if !v.Accept { w.Header().Set("ETag", etag(current)) w.Header().Set("X-KeepOrg-Strategy", string(strategy)) writeJSON(w, v.Status, map[string]any{ "code": v.Code, "error": v.Message, "current_version": current, "merge_strategy": strategy, }) return } data, err := io.ReadAll(http.MaxBytesReader(w, r.Body, s.cfg.MaxDBSize)) if err != nil { fail(w, http.StatusRequestEntityTooLarge, "too_large", fmt.Sprintf("database exceeds the %d byte limit", s.cfg.MaxDBSize)) return } if len(data) == 0 { fail(w, http.StatusBadRequest, "bad_request", "empty body") return } sum := blob.SHA256(data) if want := r.Header.Get("X-KeepOrg-SHA256"); want != "" && !strings.EqualFold(want, sum) { fail(w, http.StatusBadRequest, "checksum_mismatch", "body does not match the declared digest") return } var ( next int64 unchanged bool watchers []string name string ) err = s.st.Update(func(st *store.State) error { d := st.Databases[id] if d == nil { return errors.New("no such database") } // Re-check under the write lock: a concurrent push may have landed. if vv := decidePush(d.Version, r.Header.Get("If-Match"), d.MergeStrategy); !vv.Accept { return conflictError{current: d.Version, strategy: d.MergeStrategy, verdict: vv} } if d.SHA256 == sum { next, unchanged, name = d.Version, true, d.Name return nil } next = d.Version + 1 if err := s.blobs.Put(id, next, data); err != nil { return err } d.Version, d.SHA256, d.Size = next, sum, int64(len(data)) d.UpdatedAt, d.UpdatedBy = time.Now(), u.Username name = d.Name watchers = st.Watchers(id) return nil }) var ce conflictError switch { case errors.As(err, &ce): w.Header().Set("ETag", etag(ce.current)) w.Header().Set("X-KeepOrg-Strategy", string(ce.strategy)) writeJSON(w, ce.verdict.Status, map[string]any{ "code": ce.verdict.Code, "error": ce.verdict.Message, "current_version": ce.current, "merge_strategy": ce.strategy, }) return case err != nil: fail(w, http.StatusInternalServerError, "internal", err.Error()) return } if !unchanged { _ = s.blobs.Prune(id) s.hub.Publish(watchers, Event{Type: EventUpdated, DatabaseID: id, Name: name, Version: next, SHA256: sum, By: u.Username}) } w.Header().Set("ETag", etag(next)) writeJSON(w, http.StatusOK, map[string]any{ "version": next, "sha256": sum, "unchanged": unchanged, }) } // handleEvents streams database updates to one client over SSE. SSE is enough // here because the channel only ever carries server-to-client notifications. func (s *Server) handleEvents(w http.ResponseWriter, r *http.Request, u *store.User) { flusher, ok := w.(http.Flusher) if !ok { fail(w, http.StatusInternalServerError, "internal", "streaming unsupported") return } id, ch := s.hub.Subscribe(u.ID) defer s.hub.Unsubscribe(id) h := w.Header() h.Set("Content-Type", "text/event-stream") h.Set("Cache-Control", "no-store") h.Set("Connection", "keep-alive") h.Set("X-Accel-Buffering", "no") w.WriteHeader(http.StatusOK) fmt.Fprintf(w, "retry: 3000\n\n") flusher.Flush() keepalive := time.NewTicker(25 * time.Second) defer keepalive.Stop() for { select { case <-r.Context().Done(): return case ev, open := <-ch: if !open { return } fmt.Fprintf(w, "event: %s\ndata: %s\n\n", ev.Type, ev.encode()) flusher.Flush() case <-keepalive.C: fmt.Fprint(w, ": keepalive\n\n") flusher.Flush() } } } // ---- the push decision, kept pure ---- // verdict is the outcome of comparing a push against the stored version. type verdict struct { Accept bool Status int Code string Message string } type conflictError struct { current int64 strategy store.Strategy verdict verdict } func (e conflictError) Error() string { return e.verdict.Message } // decidePush is the whole merge policy. The client sends the version it based // its edit on in If-Match; If-Match: * means "I know I am stale and I intend // to win", which only a client-wins database allows. func decidePush(current int64, ifMatch string, strategy store.Strategy) verdict { ifMatch = strings.TrimSpace(ifMatch) switch { case ifMatch == "": return verdict{Status: http.StatusPreconditionRequired, Code: "precondition_required", Message: "If-Match must carry the version this edit was based on"} case ifMatch == "*": if strategy == store.StrategyClientWins { return verdict{Accept: true} } return verdict{Status: http.StatusConflict, Code: "force_not_allowed", Message: "this database does not allow a client to overwrite the server copy"} } v, err := strconv.ParseInt(strings.Trim(ifMatch, `"`), 10, 64) if err != nil { return verdict{Status: http.StatusBadRequest, Code: "bad_request", Message: "If-Match must be a version number or *"} } if v == current { return verdict{Accept: true} } return verdict{Status: http.StatusConflict, Code: "conflict", Message: fmt.Sprintf("the server is at version %d, this edit was based on %d", current, v)} } func etag(v int64) string { return `"` + strconv.FormatInt(v, 10) + `"` }