Monorepo for Tangled
knotserver/xrpc: add/remove knot members directly
Lewis: May this revision serve well! <lewis@tangled.org>
+393
+6
knotserver/db/known_dids.go
+6
knotserver/db/known_dids.go
···
5
5
return err
6
6
}
7
7
8
+
func IsDidKnown(q DBTX, did string) (bool, error) {
9
+
var exists bool
10
+
err := q.QueryRow(`select exists (select 1 from known_dids where did = ?)`, did).Scan(&exists)
11
+
return exists, err
12
+
}
13
+
8
14
func RemoveDid(q DBTX, did string) error {
9
15
_, err := q.Exec(`delete from known_dids where did = ?`, did)
10
16
return err
+195
knotserver/xrpc/acl_saga.go
+195
knotserver/xrpc/acl_saga.go
···
1
+
package xrpc
2
+
3
+
import (
4
+
"context"
5
+
"database/sql"
6
+
"log/slog"
7
+
"net/http"
8
+
9
+
"github.com/bluesky-social/indigo/atproto/syntax"
10
+
"tangled.org/core/knotserver/db"
11
+
"tangled.org/core/knotserver/keys"
12
+
xrpcerr "tangled.org/core/xrpc/errors"
13
+
)
14
+
15
+
type aclGrant struct {
16
+
role string
17
+
subject syntax.DID
18
+
inAcl func() (bool, error)
19
+
inTable func() (bool, error)
20
+
insertRow func(*sql.Tx) error
21
+
deleteRow func() error
22
+
grantAcl func() error
23
+
emit func() error
24
+
}
25
+
26
+
type aclRevoke struct {
27
+
role string
28
+
subject syntax.DID
29
+
inAcl func() (bool, error)
30
+
inTable func() (bool, error)
31
+
removeAcl func() (bool, error)
32
+
restoreAcl func() error
33
+
deleteRow func(*sql.Tx) error
34
+
emit func() error
35
+
}
36
+
37
+
func (h *Xrpc) applyAclGrant(ctx context.Context, l *slog.Logger, g aclGrant) (int, *xrpcerr.XrpcError) {
38
+
fail := func(status int, e xrpcerr.XrpcError) (int, *xrpcerr.XrpcError) {
39
+
return status, &e
40
+
}
41
+
42
+
inAcl, err := g.inAcl()
43
+
if err != nil {
44
+
return fail(http.StatusInternalServerError, xrpcerr.GenericError(err))
45
+
}
46
+
inTable, err := g.inTable()
47
+
if err != nil {
48
+
return fail(http.StatusInternalServerError, xrpcerr.GenericError(err))
49
+
}
50
+
if inAcl && inTable {
51
+
l.Info("subject already granted, no-op", "role", g.role, "subject", g.subject)
52
+
return http.StatusOK, nil
53
+
}
54
+
55
+
didKnown, err := db.IsDidKnown(h.Db, g.subject.String())
56
+
if err != nil {
57
+
return fail(http.StatusInternalServerError, xrpcerr.GenericError(err))
58
+
}
59
+
60
+
tx, err := h.Db.BeginTx(ctx, nil)
61
+
if err != nil {
62
+
return fail(http.StatusInternalServerError, xrpcerr.GenericError(err))
63
+
}
64
+
committed := false
65
+
defer func() {
66
+
if !committed {
67
+
tx.Rollback()
68
+
}
69
+
}()
70
+
71
+
if err := db.AddDid(tx, g.subject.String()); err != nil {
72
+
return fail(http.StatusInternalServerError, xrpcerr.GenericError(err))
73
+
}
74
+
if err := g.insertRow(tx); err != nil {
75
+
return fail(http.StatusInternalServerError, xrpcerr.GenericError(err))
76
+
}
77
+
if err := tx.Commit(); err != nil {
78
+
return fail(http.StatusInternalServerError, xrpcerr.GenericError(err))
79
+
}
80
+
committed = true
81
+
82
+
if err := g.grantAcl(); err != nil {
83
+
if !inTable {
84
+
if rbErr := g.deleteRow(); rbErr != nil {
85
+
l.Error("failed to roll back row after ACL grant failed", "role", g.role, "subject", g.subject, "error", rbErr)
86
+
}
87
+
}
88
+
if !didKnown {
89
+
if rbErr := db.RemoveDid(h.Db, g.subject.String()); rbErr != nil {
90
+
l.Error("failed to roll back known_did after ACL grant failed", "role", g.role, "subject", g.subject, "error", rbErr)
91
+
}
92
+
}
93
+
return fail(http.StatusInternalServerError, xrpcerr.GenericError(err))
94
+
}
95
+
96
+
h.Ingester.AddDid(g.subject.String())
97
+
h.fetchKeysAsync(ctx, l, g.subject)
98
+
99
+
if g.emit != nil {
100
+
if err := g.emit(); err != nil {
101
+
l.Error("failed to emit acl grant event, appview reconcile will catch up", "role", g.role, "subject", g.subject, "error", err)
102
+
}
103
+
}
104
+
105
+
l.Info("granted", "role", g.role, "subject", g.subject)
106
+
return http.StatusOK, nil
107
+
}
108
+
109
+
func (h *Xrpc) applyAclRevoke(ctx context.Context, l *slog.Logger, rv aclRevoke) (int, *xrpcerr.XrpcError) {
110
+
fail := func(status int, e xrpcerr.XrpcError) (int, *xrpcerr.XrpcError) {
111
+
return status, &e
112
+
}
113
+
114
+
inAcl, err := rv.inAcl()
115
+
if err != nil {
116
+
return fail(http.StatusInternalServerError, xrpcerr.GenericError(err))
117
+
}
118
+
inTable, err := rv.inTable()
119
+
if err != nil {
120
+
return fail(http.StatusInternalServerError, xrpcerr.GenericError(err))
121
+
}
122
+
if !inAcl && !inTable {
123
+
l.Info("subject not granted, no-op", "role", rv.role, "subject", rv.subject)
124
+
return http.StatusOK, nil
125
+
}
126
+
127
+
removed := false
128
+
if inAcl {
129
+
removed, err = rv.removeAcl()
130
+
if err != nil {
131
+
return fail(http.StatusInternalServerError, xrpcerr.GenericError(err))
132
+
}
133
+
}
134
+
135
+
failRestoringACL := func(status int, e xrpcerr.XrpcError) (int, *xrpcerr.XrpcError) {
136
+
if removed {
137
+
if rbErr := rv.restoreAcl(); rbErr != nil {
138
+
l.Error("failed to restore ACL after remove rollback", "role", rv.role, "subject", rv.subject, "error", rbErr)
139
+
}
140
+
}
141
+
return fail(status, e)
142
+
}
143
+
144
+
stillKnown, err := h.Enforcer.HasAnyPolicyForUser(rv.subject.String())
145
+
if err != nil {
146
+
return failRestoringACL(http.StatusInternalServerError, xrpcerr.GenericError(err))
147
+
}
148
+
149
+
tx, err := h.Db.BeginTx(ctx, nil)
150
+
if err != nil {
151
+
return failRestoringACL(http.StatusInternalServerError, xrpcerr.GenericError(err))
152
+
}
153
+
committed := false
154
+
defer func() {
155
+
if !committed {
156
+
tx.Rollback()
157
+
}
158
+
}()
159
+
160
+
if err := rv.deleteRow(tx); err != nil {
161
+
return failRestoringACL(http.StatusInternalServerError, xrpcerr.GenericError(err))
162
+
}
163
+
if !stillKnown {
164
+
if err := db.RemoveDid(tx, rv.subject.String()); err != nil {
165
+
return failRestoringACL(http.StatusInternalServerError, xrpcerr.GenericError(err))
166
+
}
167
+
}
168
+
if err := tx.Commit(); err != nil {
169
+
return failRestoringACL(http.StatusInternalServerError, xrpcerr.GenericError(err))
170
+
}
171
+
committed = true
172
+
173
+
if !stillKnown {
174
+
h.Ingester.RemoveDid(rv.subject.String())
175
+
}
176
+
177
+
if rv.emit != nil {
178
+
if err := rv.emit(); err != nil {
179
+
l.Error("failed to emit acl revoke event, appview reconcile will catch up", "role", rv.role, "subject", rv.subject, "error", err)
180
+
}
181
+
}
182
+
183
+
l.Info("revoked", "role", rv.role, "subject", rv.subject, "did_dropped", !stillKnown)
184
+
return http.StatusOK, nil
185
+
}
186
+
187
+
func (h *Xrpc) fetchKeysAsync(ctx context.Context, l *slog.Logger, subject syntax.DID) {
188
+
kctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), keyFetchTimeout)
189
+
go func() {
190
+
defer cancel()
191
+
if err := keys.FetchAndStore(kctx, h.Resolver.Directory(), h.Db, subject.String()); err != nil {
192
+
l.Warn("failed to fetch subject public keys, continuing", "subject", subject, "error", err)
193
+
}
194
+
}()
195
+
}
+98
knotserver/xrpc/add_member.go
+98
knotserver/xrpc/add_member.go
···
1
+
package xrpc
2
+
3
+
import (
4
+
"context"
5
+
"database/sql"
6
+
"encoding/json"
7
+
"log/slog"
8
+
"net/http"
9
+
"time"
10
+
11
+
"github.com/bluesky-social/indigo/atproto/syntax"
12
+
"tangled.org/core/api/tangled"
13
+
"tangled.org/core/knotserver/db"
14
+
"tangled.org/core/rbac"
15
+
xrpcerr "tangled.org/core/xrpc/errors"
16
+
)
17
+
18
+
const keyFetchTimeout = 15 * time.Second
19
+
20
+
func (h *Xrpc) AddMember(w http.ResponseWriter, r *http.Request) {
21
+
l := h.Logger.With("handler", "AddMember")
22
+
fail := func(e xrpcerr.XrpcError, status int) {
23
+
l.Error("failed", "kind", e.Tag, "error", e.Message)
24
+
writeError(w, e, status)
25
+
}
26
+
27
+
actorDid, ok := r.Context().Value(ActorDid).(syntax.DID)
28
+
if !ok {
29
+
fail(xrpcerr.MissingActorDidError, http.StatusForbidden)
30
+
return
31
+
}
32
+
33
+
allowed, err := h.Enforcer.IsKnotInviteAllowed(actorDid.String(), rbac.ThisServer)
34
+
if err != nil {
35
+
fail(xrpcerr.GenericError(err), http.StatusInternalServerError)
36
+
return
37
+
}
38
+
if !allowed {
39
+
fail(xrpcerr.AccessControlError(actorDid.String()), http.StatusForbidden)
40
+
return
41
+
}
42
+
43
+
var data tangled.KnotAddMember_Input
44
+
if err := json.NewDecoder(r.Body).Decode(&data); err != nil {
45
+
fail(xrpcerr.GenericError(err), http.StatusBadRequest)
46
+
return
47
+
}
48
+
49
+
subject, err := syntax.ParseDID(data.Subject)
50
+
if err != nil {
51
+
fail(xrpcerr.GenericError(err), http.StatusBadRequest)
52
+
return
53
+
}
54
+
55
+
status, xerr := h.addMemberToKnot(r.Context(), l, actorDid, subject)
56
+
if xerr != nil {
57
+
fail(*xerr, status)
58
+
return
59
+
}
60
+
w.WriteHeader(status)
61
+
}
62
+
63
+
func (h *Xrpc) addMemberToKnot(ctx context.Context, l *slog.Logger, addedBy, subject syntax.DID) (int, *xrpcerr.XrpcError) {
64
+
isOwner, err := h.Enforcer.IsKnotOwner(subject.String(), rbac.ThisServer)
65
+
if err != nil {
66
+
e := xrpcerr.GenericError(err)
67
+
return http.StatusInternalServerError, &e
68
+
}
69
+
if isOwner {
70
+
l.Info("subject is the knot owner, no-op", "subject", subject)
71
+
return http.StatusOK, nil
72
+
}
73
+
74
+
return h.applyAclGrant(ctx, l, aclGrant{
75
+
role: "member",
76
+
subject: subject,
77
+
inAcl: func() (bool, error) {
78
+
return h.Enforcer.IsKnotMember(subject.String(), rbac.ThisServer)
79
+
},
80
+
inTable: func() (bool, error) {
81
+
n, err := db.CountKnotMembersBySubject(h.Db, subject.String())
82
+
return n > 0, err
83
+
},
84
+
insertRow: func(tx *sql.Tx) error {
85
+
return db.AddKnotMemberDirect(tx, addedBy, subject)
86
+
},
87
+
deleteRow: func() error {
88
+
return db.RemoveKnotMemberDirect(h.Db, subject)
89
+
},
90
+
grantAcl: func() error {
91
+
_, err := h.Enforcer.TryAddKnotMember(rbac.ThisServer, subject.String())
92
+
return err
93
+
},
94
+
emit: func() error {
95
+
return h.Db.EmitKnotMemberUpdate(h.Notifier, db.AclOpAdd, subject)
96
+
},
97
+
})
98
+
}
+92
knotserver/xrpc/remove_member.go
+92
knotserver/xrpc/remove_member.go
···
1
+
package xrpc
2
+
3
+
import (
4
+
"database/sql"
5
+
"encoding/json"
6
+
"net/http"
7
+
8
+
"github.com/bluesky-social/indigo/atproto/syntax"
9
+
"tangled.org/core/api/tangled"
10
+
"tangled.org/core/knotserver/db"
11
+
"tangled.org/core/rbac"
12
+
xrpcerr "tangled.org/core/xrpc/errors"
13
+
)
14
+
15
+
func (h *Xrpc) RemoveMember(w http.ResponseWriter, r *http.Request) {
16
+
l := h.Logger.With("handler", "RemoveMember")
17
+
fail := func(e xrpcerr.XrpcError, status int) {
18
+
l.Error("failed", "kind", e.Tag, "error", e.Message)
19
+
writeError(w, e, status)
20
+
}
21
+
22
+
actorDid, ok := r.Context().Value(ActorDid).(syntax.DID)
23
+
if !ok {
24
+
fail(xrpcerr.MissingActorDidError, http.StatusForbidden)
25
+
return
26
+
}
27
+
28
+
allowed, err := h.Enforcer.IsKnotInviteAllowed(actorDid.String(), rbac.ThisServer)
29
+
if err != nil {
30
+
fail(xrpcerr.GenericError(err), http.StatusInternalServerError)
31
+
return
32
+
}
33
+
if !allowed {
34
+
fail(xrpcerr.AccessControlError(actorDid.String()), http.StatusForbidden)
35
+
return
36
+
}
37
+
38
+
var data tangled.KnotRemoveMember_Input
39
+
if err := json.NewDecoder(r.Body).Decode(&data); err != nil {
40
+
fail(xrpcerr.GenericError(err), http.StatusBadRequest)
41
+
return
42
+
}
43
+
44
+
subject, err := syntax.ParseDID(data.Subject)
45
+
if err != nil {
46
+
fail(xrpcerr.GenericError(err), http.StatusBadRequest)
47
+
return
48
+
}
49
+
50
+
isOwner, err := h.Enforcer.IsKnotOwner(subject.String(), rbac.ThisServer)
51
+
if err != nil {
52
+
fail(xrpcerr.GenericError(err), http.StatusInternalServerError)
53
+
return
54
+
}
55
+
if isOwner {
56
+
fail(xrpcerr.NewXrpcError(
57
+
xrpcerr.WithTag("InvalidRequest"),
58
+
xrpcerr.WithMessage("cannot remove the knot owner"),
59
+
), http.StatusBadRequest)
60
+
return
61
+
}
62
+
63
+
status, xerr := h.applyAclRevoke(r.Context(), l, aclRevoke{
64
+
role: "member",
65
+
subject: subject,
66
+
inAcl: func() (bool, error) {
67
+
return h.Enforcer.IsKnotMember(subject.String(), rbac.ThisServer)
68
+
},
69
+
inTable: func() (bool, error) {
70
+
n, err := db.CountKnotMembersBySubject(h.Db, subject.String())
71
+
return n > 0, err
72
+
},
73
+
removeAcl: func() (bool, error) {
74
+
return h.Enforcer.TryRemoveKnotMember(rbac.ThisServer, subject.String())
75
+
},
76
+
restoreAcl: func() error {
77
+
_, err := h.Enforcer.TryAddKnotMember(rbac.ThisServer, subject.String())
78
+
return err
79
+
},
80
+
deleteRow: func(tx *sql.Tx) error {
81
+
return db.RemoveKnotMemberBySubject(tx, subject)
82
+
},
83
+
emit: func() error {
84
+
return h.Db.EmitKnotMemberUpdate(h.Notifier, db.AclOpRemove, subject)
85
+
},
86
+
})
87
+
if xerr != nil {
88
+
fail(*xerr, status)
89
+
return
90
+
}
91
+
w.WriteHeader(status)
92
+
}
+2
knotserver/xrpc/xrpc.go
+2
knotserver/xrpc/xrpc.go
···
56
56
r.Post("/"+tangled.RepoForkSyncNSID, x.ForkSync)
57
57
r.Post("/"+tangled.RepoHiddenRefNSID, x.HiddenRef)
58
58
r.Post("/"+tangled.RepoMergeNSID, x.Merge)
59
+
r.Post("/"+tangled.KnotAddMemberNSID, x.AddMember)
60
+
r.Post("/"+tangled.KnotRemoveMemberNSID, x.RemoveMember)
59
61
})
60
62
61
63
// merge check is an open endpoint