Add KeepOrg sync server: accounts, nested groups, versioned KDBX sync
Why: collaborative KeePass use needs one arbiter of "who wrote last" that never sees a master password, plus access control that follows an org chart. What: stdlib-only Go server. Metadata is a single JSON document with one writer and atomic commit; KDBX payloads are versioned blobs on disk. Access comes from a group tree where a grant on a group reaches every member below it. Login is password+TOTP, invite links, or OIDC through a browser-approved device flow. Pushes carry If-Match and are arbitrated by a per-database merge strategy (server-wins by default). Updates are pushed to clients over SSE. Proved: go test ./internal/... covers RFC 6238 vectors, ACL inheritance and isolation, store rollback, blob pruning, every push verdict, and an end-to-end HTTP pass over login, conflict, force, read-only grants, invite redemption, TOTP enrolment, device login and the event stream.
This commit is contained in:
@@ -0,0 +1,284 @@
|
||||
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) + `"` }
|
||||
Reference in New Issue
Block a user