amadan

oficina / docs Public

Add guarded TeamMode Home completion and fixed-lease stream seams

Paul Campbell pushed by paul@keymail.dev 07df446962208e10735b37e41c077dda06956e7a parent dd1f126
7 files changed, +268 −16
  • internal/docs/home_exchange.go +17 −11
  • internal/docs/identity.go +6 −0
  • internal/docs/stream.go +39 −1
  • internal/docs/team_config.go +8 −4
  • internal/docs/team_home.go +62 −0
  • internal/docs/team_home_test.go +79 −0
  • internal/docs/team_stream.go +57 −0

CI passed — run details

diff --git a/internal/docs/home_exchange.go b/internal/docs/home_exchange.go
index 19a2f8f..ec0b796 100644
--- a/internal/docs/home_exchange.go
+++ b/internal/docs/home_exchange.go
@@ -220,17 +220,23 @@ func (b *homeBridge) complete(w http.ResponseWriter, r *http.Request, claims ass
setHomeCookie(w, homeProofCookie, secret, 300)
return nil
}
- id, _, err := b.links.Resolve(r.Context(), claims.Issuer, claims.Subject)
- if errors.Is(err, linkstore.ErrRefused) {
- return rejectHome(r, "unknown", http.StatusForbidden)
- }
- if err != nil {
- return errHomeBridge
- }
- if result.session != nil {
- person, err := b.existingPerson(r, *result.session)
- if err != nil || person.ID != id {
- return rejectHome(r, "failed", http.StatusForbidden)
+ if b.store.team != nil {
+ if err := b.completeTeamPrincipal(r, claims, result.session); err != nil {
+ return err
+ }
+ } else {
+ id, _, err := b.links.Resolve(r.Context(), claims.Issuer, claims.Subject)
+ if errors.Is(err, linkstore.ErrRefused) {
+ return rejectHome(r, "unknown", http.StatusForbidden)
+ }
+ if err != nil {
+ return errHomeBridge
+ }
+ if result.session != nil {
+ person, err := b.existingPerson(r, *result.session)
+ if err != nil || person.ID != id {
+ return rejectHome(r, "failed", http.StatusForbidden)
+ }
}
}
if err := b.sessions.SignIn(w, r, sessions.Session{Subject: claims.Subject, Method: homeAssertionMethod, AuthTime: time.Unix(claims.AuthTime, 0)}); err != nil {
diff --git a/internal/docs/identity.go b/internal/docs/identity.go
index 2fa28eb..f070227 100644
--- a/internal/docs/identity.go
+++ b/internal/docs/identity.go
@@ -47,6 +47,9 @@ func (v identityResolver) resolve(r *http.Request) (requestIdentity, error) {
if !ok || session.Subject == "" || v.store == nil {
return requestIdentity{}, errIdentityRefused
}
+ if v.store.team != nil && v.team == nil {
+ return requestIdentity{}, errIdentityRefused
+ }
if session.Method == homeAssertionMethod {
if v.team != nil {
p, err := v.team.principals.Resolve(r.Context(), v.team.bindings.Issuer, session.Subject)
@@ -144,6 +147,9 @@ func requestActor(r *http.Request, store *Store) (Actor, error) {
if id, ok := currentIdentity(r); ok {
return id.Actor, nil
}
+ if store.team != nil {
+ return Actor{}, errIdentityRefused
+ }
id, err := (identityResolver{store: store}).resolve(r)
return id.Actor, err
}
diff --git a/internal/docs/stream.go b/internal/docs/stream.go
index 23639c1..bb94a60 100644
--- a/internal/docs/stream.go
+++ b/internal/docs/stream.go
@@ -6,6 +6,7 @@ import (
"strconv"
"time"
+ "amadan.net/oficina/home/roster"
"github.com/carlosframework/rastrillo"
)
@@ -86,6 +87,31 @@ func (h *handlers) stream(w http.ResponseWriter, r *http.Request) {
return
}
+ var guard *teamStreamGuard
+ var leaseTimer *time.Timer
+ var leaseDeadline <-chan time.Time
+ if h.store.team != nil {
+ a, ok := r.Context().Value(accessContextKey{}).(*requestAccess)
+ if !ok || a == nil || a.runtime != h.store.team {
+ http.Error(w, rastrillo.T(r, "docs.error.generic"), http.StatusServiceUnavailable)
+ return
+ }
+ guard = &teamStreamGuard{store: h.store, access: a, request: r, docID: doc.ID}
+ if _, ok := guard.check(); !ok {
+ http.Error(w, rastrillo.T(r, "docs.error.generic"), http.StatusServiceUnavailable)
+ return
+ }
+ if a.decision.State == roster.Member {
+ remaining := guard.remaining()
+ if remaining <= 0 {
+ http.Error(w, rastrillo.T(r, "docs.error.generic"), http.StatusServiceUnavailable)
+ return
+ }
+ leaseTimer = time.NewTimer(remaining)
+ defer leaseTimer.Stop()
+ leaseDeadline = leaseTimer.C
+ }
+ }
rc := http.NewResponseController(w)
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-store")
@@ -98,7 +124,17 @@ func (h *handlers) stream(w http.ResponseWriter, r *http.Request) {
// unbounded — while a write error ends the handler, because it
// means the client is gone.
write := func(text string) bool {
- _ = rc.SetWriteDeadline(time.Now().Add(defaultStreamWriteDeadline))
+ limit := defaultStreamWriteDeadline
+ if guard != nil {
+ var ok bool
+ limit, ok = guard.check()
+ if !ok {
+ return false
+ }
+ }
+ if err := rc.SetWriteDeadline(time.Now().Add(limit)); err != nil && guard != nil {
+ return false
+ }
if _, err := io.WriteString(w, text); err != nil {
return false
}
@@ -126,6 +162,8 @@ func (h *handlers) stream(w http.ResponseWriter, r *http.Request) {
for {
select {
+ case <-leaseDeadline:
+ return
case <-r.Context().Done():
// The client went away, or the server is draining. Either
// way this handler must return so a graceful shutdown's
diff --git a/internal/docs/team_config.go b/internal/docs/team_config.go
index 38ee412..3e3bda6 100644
--- a/internal/docs/team_config.go
+++ b/internal/docs/team_config.go
@@ -6,6 +6,8 @@ import (
"database/sql"
"encoding/base64"
"errors"
+ "github.com/carlosframework/rastrillo/sessions"
+ "net/http"
"amadan.net/oficina/home/handoff/seal"
"amadan.net/oficina/home/roster"
@@ -68,10 +70,12 @@ func validatedTeamConfig(cfg *TeamConfig, home *homeSettings, origin string) (*t
}
type teamRuntime struct {
- cache *roster.Cache
- principals *principalstore.Store
- bindings roster.Bindings
- db *sql.DB
+ bridge *homeBridge
+ readSession func(context.Context, *http.Request) (sessions.Session, bool)
+ cache *roster.Cache
+ principals *principalstore.Store
+ bindings roster.Bindings
+ db *sql.DB
}
func newTeamRuntime(ctx context.Context, settings *teamSettings, db *sql.DB, home *homeSettings) (*teamRuntime, error) {
diff --git a/internal/docs/team_home.go b/internal/docs/team_home.go
new file mode 100644
index 0000000..e51efcb
--- /dev/null
+++ b/internal/docs/team_home.go
@@ -0,0 +1,62 @@
+package docs
+
+import (
+ "errors"
+ "net/http"
+
+ "amadan.net/rastrillo/rastrillo/assertion"
+ "github.com/carlosframework/docs/internal/docs/principalstore"
+ "github.com/carlosframework/rastrillo/sessions"
+)
+
+// existingPrincipal is read-only; an existing browser session cannot cause an
+// unknown incoming identity to allocate a second account during completion.
+func (b *homeBridge) existingPrincipal(r *http.Request, session sessions.Session) (principalstore.Principal, error) {
+ t := b.store.team
+ if t == nil {
+ return principalstore.Principal{}, errIdentityRefused
+ }
+ if session.Method == homeAssertionMethod {
+ return t.principals.Resolve(r.Context(), t.bindings.Issuer, session.Subject)
+ }
+ person, e := b.existingPerson(r, session)
+ if e != nil {
+ return principalstore.Principal{}, e
+ }
+ p, e := t.principals.Person(r.Context(), person.ID)
+ if e == nil && p.Kind != principalstore.Local {
+ return principalstore.Principal{}, errIdentityRefused
+ }
+ return p, e
+}
+func (b *homeBridge) completeTeamPrincipal(r *http.Request, claims assertion.Claims, current *sessions.Session) error {
+ t := b.store.team
+ p, e := t.principals.Resolve(r.Context(), claims.Issuer, claims.Subject)
+ if e != nil && !errors.Is(e, principalstore.ErrRefused) {
+ return rejectHome(r, "failed", http.StatusServiceUnavailable)
+ }
+ if current != nil {
+ if e != nil {
+ return rejectHome(r, "failed", http.StatusForbidden)
+ }
+ existing, err := b.existingPrincipal(r, *current)
+ if errors.Is(err, principalstore.ErrUnavailable) || errors.Is(err, errIdentityStorage) {
+ return rejectHome(r, "failed", http.StatusServiceUnavailable)
+ }
+ if err != nil || existing.ID != p.ID {
+ return rejectHome(r, "failed", http.StatusForbidden)
+ }
+ return nil
+ }
+ if e == nil {
+ return nil
+ }
+ _, e = t.principals.EnsureHome(r.Context(), principalstore.HomeIdentity{Issuer: claims.Issuer, Subject: claims.Subject})
+ if errors.Is(e, principalstore.ErrRefused) {
+ return rejectHome(r, "unknown", http.StatusForbidden)
+ }
+ if e != nil {
+ return rejectHome(r, "failed", http.StatusServiceUnavailable)
+ }
+ return nil
+}
diff --git a/internal/docs/team_home_test.go b/internal/docs/team_home_test.go
new file mode 100644
index 0000000..2cbf266
--- /dev/null
+++ b/internal/docs/team_home_test.go
@@ -0,0 +1,79 @@
+package docs
+
+import (
+ "context"
+ "errors"
+ "net/http/httptest"
+ "strings"
+ "testing"
+ "time"
+
+ "amadan.net/oficina/home/roster"
+ "amadan.net/rastrillo/rastrillo/assertion"
+ "github.com/carlosframework/docs/internal/docs/principalstore"
+ "github.com/carlosframework/rastrillo/sessions"
+)
+
+func TestTeamCompletionCannotAllocateAroundCurrentAccount(t *testing.T) {
+ app := newSampleDocumentsInternalTestApp(t, "0")
+ s := app.Store()
+ ctx := t.Context()
+ local, e := s.EnsurePerson(ctx, "local@example.com")
+ if e != nil {
+ t.Fatal(e)
+ }
+ if _, e = s.Create(ctx, local, "existing"); e != nil {
+ t.Fatal(e)
+ }
+ rt, clock, subject := testTeamRoster(t, s, app)
+ calls := 0
+ rt.principals, e = principalstore.New(app.db.Writer(), principalstore.Config{Issuer: rt.bindings.Issuer, Team: rt.bindings.Team, RequireMembership: func(ctx context.Context, q principalstore.MembershipRequest) error {
+ calls++
+ d := rt.cache.Check(ctx, q.Issuer, q.Team, q.Subject)
+ if d.State == roster.Member && d.Lease.Valid() {
+ return nil
+ }
+ if d.State == roster.NotMember {
+ return principalstore.ErrRefused
+ }
+ return principalstore.ErrUnavailable
+ }})
+ if e != nil {
+ t.Fatal(e)
+ }
+ b := &homeBridge{store: s, resolver: identityResolver{store: s, team: rt}}
+ result := &homeFinishResult{}
+ r := httptest.NewRequest("GET", "https://docs.example/auth/home/finish", nil).WithContext(context.WithValue(ctx, homeFinishKey{}, result))
+ claims := assertion.Claims{Issuer: rt.bindings.Issuer, Subject: subject}
+ current := &sessions.Session{Subject: "local@example.com", Method: "magiclink", At: time.Now()}
+ if e = b.completeTeamPrincipal(r, claims, current); e == nil || calls != 0 {
+ t.Fatal("current account conflict reached allocation", e)
+ }
+ if _, e = rt.principals.Resolve(ctx, claims.Issuer, subject); !errors.Is(e, principalstore.ErrRefused) {
+ t.Fatal("conflict allocated principal", e)
+ }
+ if e = b.completeTeamPrincipal(r, claims, nil); e != nil || calls != 1 {
+ t.Fatal("verified fresh member could not create stable identity", e)
+ }
+ p, e := rt.principals.Resolve(ctx, claims.Issuer, subject)
+ if e != nil || p.ID != principalstore.HomeIDMin || p.VerifiedAddress != nil {
+ t.Fatal("Home principal changed address or allocation", e)
+ }
+ var people, docs int64
+ s.db.Model(&Person{}).Count(&people)
+ s.db.Model(&Document{}).Count(&docs)
+ if people != 1 || docs != 1 {
+ t.Fatal("Home completion fabricated local person or sample documents")
+ }
+ clock.step(60 * time.Second)
+ if e = b.completeTeamPrincipal(r, claims, &sessions.Session{Subject: subject, Method: homeAssertionMethod}); e != nil || calls != 1 {
+ t.Fatal("known identity requires no new allocation or standing", e)
+ }
+ claims.Subject = "pr_" + strings.Repeat("b", 26)
+ if e = b.completeTeamPrincipal(r, claims, nil); e == nil || result.status != 503 {
+ t.Fatal("unavailable roster did not refuse unknown identity", e)
+ }
+ if _, e = rt.principals.Resolve(ctx, claims.Issuer, claims.Subject); !errors.Is(e, principalstore.ErrRefused) {
+ t.Fatal("unknown roster allocated principal", e)
+ }
+}
diff --git a/internal/docs/team_stream.go b/internal/docs/team_stream.go
new file mode 100644
index 0000000..5098f9e
--- /dev/null
+++ b/internal/docs/team_stream.go
@@ -0,0 +1,57 @@
+package docs
+
+import (
+ "context"
+ "net/http"
+ "time"
+
+ "amadan.net/oficina/home/roster"
+ "github.com/carlosframework/rastrillo/sessions"
+)
+
+type teamStreamGuard struct {
+ store *Store
+ access *requestAccess
+ request *http.Request
+ docID int64
+}
+
+func (g *teamStreamGuard) remaining() time.Duration {
+ if g.access.decision.State == roster.Member {
+ return g.access.decision.Lease.Remaining()
+ }
+ return defaultStreamWriteDeadline
+}
+
+// A stream never renews its lease. Reconnection obtains fresh admission, so a
+// blocked cache publisher cannot hold an expired stream open during refresh.
+func (g *teamStreamGuard) check() (time.Duration, bool) {
+ t := g.store.team
+ remaining := g.remaining()
+ if t == nil || t.bridge == nil || t.readSession == nil || remaining <= 0 {
+ return 0, false
+ }
+ limit := min(remaining, defaultStreamWriteDeadline)
+ ctx, cancel := context.WithTimeout(g.request.Context(), limit)
+ defer cancel()
+ if _, present, ok := homeCookie(g.request, t.bridge.sessionCookie); !ok || !present {
+ return 0, false
+ }
+ session, ok := t.readSession(ctx, g.request)
+ if !ok || ctx.Err() != nil {
+ return 0, false
+ }
+ request := sessions.WithSession(g.request.WithContext(ctx), session)
+ id, e := t.bridge.resolver.resolve(request)
+ if e != nil || !sameActor(id.Actor, g.access.actor) || id.HomeSubject != g.access.subject {
+ return 0, false
+ }
+ if e = g.store.checkPermission(ctx, g.store.db.WithContext(ctx), id.Actor, g.docID, permissionRead); e != nil {
+ return 0, false
+ }
+ remaining = g.remaining()
+ if ctx.Err() != nil || remaining <= 0 {
+ return 0, false
+ }
+ return min(remaining, defaultStreamWriteDeadline), true
+}