Skip to main content

max / makenotwork

12.5 KB · 397 lines History Blame Raw
1 //! SyncKit groups: the shared-changelog membership and sealed-GCK-grant layer.
2 //!
3 //! A group is owned by one admin who mints a Group Content Key (GCK) and seals it
4 //! to each member's identity public key. The server stores membership and the
5 //! opaque sealed grants; it never sees the GCK or any plaintext. The group-scoped
6 //! push/pull that reads `sync_log.group_id` lives in `log.rs` (p2-changelog); this
7 //! module owns the group, membership, and grant records. Design: wiki
8 //! synckit-groups-design.
9
10 use sqlx::PgPool;
11
12 use crate::db::models::{DbSyncGroup, DbSyncGroupMember};
13 use crate::db::{SyncAppId, SyncGroupId, UserId};
14 use crate::error::Result;
15
16 /// Create a group and enroll its admin as the first member.
17 ///
18 /// Runs in one transaction: the `sync_groups` row (GCK generation 1) and the
19 /// admin's own `sync_group_members` row (`role = 'admin'`, carrying the admin's
20 /// self-sealed GCK grant) are written together, so a group never exists without
21 /// its admin able to read it.
22 #[tracing::instrument(skip_all)]
23 pub async fn create_group(
24 pool: &PgPool,
25 id: SyncGroupId,
26 app_id: SyncAppId,
27 admin_user_id: UserId,
28 name: &str,
29 admin_sealed_gck: &str,
30 admin_pubkey: &str,
31 ) -> Result<DbSyncGroup> {
32 let mut tx = pool.begin().await?;
33
34 // The id is client-generated (the admin's GCK grant is sealed bound to it
35 // before the group exists). A PK collision surfaces as a DB error.
36 let group = sqlx::query_as::<_, DbSyncGroup>(
37 r"
38 INSERT INTO sync_groups (id, app_id, admin_user_id, name)
39 VALUES ($1, $2, $3, $4)
40 RETURNING id, app_id, admin_user_id, name, gck_version, created_at
41 ",
42 )
43 .bind(id)
44 .bind(app_id)
45 .bind(admin_user_id)
46 .bind(name)
47 .fetch_one(&mut *tx)
48 .await?;
49
50 sqlx::query(
51 r"
52 INSERT INTO sync_group_members (group_id, user_id, role, member_pubkey)
53 VALUES ($1, $2, 'admin', $3)
54 ",
55 )
56 .bind(group.id)
57 .bind(admin_user_id)
58 .bind(admin_pubkey)
59 .execute(&mut *tx)
60 .await?;
61
62 sqlx::query(
63 r"
64 INSERT INTO sync_group_grants (group_id, user_id, gck_version, sealed_gck)
65 VALUES ($1, $2, $3, $4)
66 ",
67 )
68 .bind(group.id)
69 .bind(admin_user_id)
70 .bind(group.gck_version)
71 .bind(admin_sealed_gck)
72 .execute(&mut *tx)
73 .await?;
74
75 tx.commit().await?;
76 Ok(group)
77 }
78
79 /// Fetch a group scoped to its app (the app scope guards against cross-app id
80 /// guessing).
81 #[tracing::instrument(skip_all)]
82 pub async fn get_group(
83 pool: &PgPool,
84 app_id: SyncAppId,
85 group_id: SyncGroupId,
86 ) -> Result<Option<DbSyncGroup>> {
87 let group = sqlx::query_as::<_, DbSyncGroup>(
88 r"
89 SELECT id, app_id, admin_user_id, name, gck_version, created_at
90 FROM sync_groups
91 WHERE app_id = $1 AND id = $2
92 ",
93 )
94 .bind(app_id)
95 .bind(group_id)
96 .fetch_optional(pool)
97 .await?;
98 Ok(group)
99 }
100
101 /// Fetch a group by id alone, without an app scope.
102 ///
103 /// The app scope on [`get_group`] guards against a caller guessing ids. This
104 /// variant exists for the invitation preview, where the unguessable token has
105 /// already named exactly one group and the caller is by definition not a member
106 /// of it yet. Do not reach for it on a path where the caller supplies the id.
107 #[tracing::instrument(skip_all)]
108 pub async fn get_group_by_id(pool: &PgPool, group_id: SyncGroupId) -> Result<Option<DbSyncGroup>> {
109 let group = sqlx::query_as::<_, DbSyncGroup>(
110 r"
111 SELECT id, app_id, admin_user_id, name, gck_version, created_at
112 FROM sync_groups
113 WHERE id = $1
114 ",
115 )
116 .bind(group_id)
117 .fetch_optional(pool)
118 .await?;
119 Ok(group)
120 }
121
122 /// List the groups a user belongs to within an app, most recent first.
123 #[tracing::instrument(skip_all)]
124 pub async fn list_groups_for_user(
125 pool: &PgPool,
126 app_id: SyncAppId,
127 user_id: UserId,
128 ) -> Result<Vec<DbSyncGroup>> {
129 let groups = sqlx::query_as::<_, DbSyncGroup>(
130 r"
131 SELECT g.id, g.app_id, g.admin_user_id, g.name, g.gck_version, g.created_at
132 FROM sync_groups g
133 JOIN sync_group_members m ON m.group_id = g.id
134 WHERE g.app_id = $1 AND m.user_id = $2
135 ORDER BY g.created_at DESC
136 ",
137 )
138 .bind(app_id)
139 .bind(user_id)
140 .fetch_all(pool)
141 .await?;
142 Ok(groups)
143 }
144
145 /// All members of a group.
146 #[tracing::instrument(skip_all)]
147 pub async fn list_members(pool: &PgPool, group_id: SyncGroupId) -> Result<Vec<DbSyncGroupMember>> {
148 let members = sqlx::query_as::<_, DbSyncGroupMember>(
149 r"
150 SELECT m.group_id, m.user_id, u.email, m.role, m.added_at
151 FROM sync_group_members m
152 JOIN users u ON u.id = m.user_id
153 WHERE m.group_id = $1
154 ORDER BY m.added_at ASC
155 ",
156 )
157 .bind(group_id)
158 .fetch_all(pool)
159 .await?;
160 Ok(members)
161 }
162
163 /// The `(user_id, member_pubkey)` of every member with a stored public key. The
164 /// admin fetches this to re-seal a rotated GCK to the remaining members without
165 /// re-collecting their keys out of band. Members without a stored key (legacy
166 /// rows) are omitted.
167 #[tracing::instrument(skip_all)]
168 pub async fn list_member_pubkeys(
169 pool: &PgPool,
170 group_id: SyncGroupId,
171 ) -> Result<Vec<(UserId, String)>> {
172 let rows: Vec<(UserId, String)> = sqlx::query_as(
173 "SELECT user_id, member_pubkey FROM sync_group_members \
174 WHERE group_id = $1 AND member_pubkey IS NOT NULL \
175 ORDER BY added_at ASC",
176 )
177 .bind(group_id)
178 .fetch_all(pool)
179 .await?;
180 Ok(rows)
181 }
182
183 /// Whether `user_id` is a member of `group_id`. The membership gate for
184 /// group-scoped push/pull.
185 #[tracing::instrument(skip_all)]
186 pub async fn is_group_member(
187 pool: &PgPool,
188 group_id: SyncGroupId,
189 user_id: UserId,
190 ) -> Result<bool> {
191 let exists: bool = sqlx::query_scalar(
192 "SELECT EXISTS (SELECT 1 FROM sync_group_members WHERE group_id = $1 AND user_id = $2)",
193 )
194 .bind(group_id)
195 .bind(user_id)
196 .fetch_one(pool)
197 .await?;
198 Ok(exists)
199 }
200
201 /// Whether `user_id` is the admin of `group_id`. Admin-only actions (add/remove
202 /// member, rotate) gate on this.
203 #[tracing::instrument(skip_all)]
204 pub async fn is_group_admin(pool: &PgPool, group_id: SyncGroupId, user_id: UserId) -> Result<bool> {
205 let exists: bool = sqlx::query_scalar(
206 "SELECT EXISTS (SELECT 1 FROM sync_groups WHERE id = $1 AND admin_user_id = $2)",
207 )
208 .bind(group_id)
209 .bind(user_id)
210 .fetch_one(pool)
211 .await?;
212 Ok(exists)
213 }
214
215 /// Fetch a member's newest sealed GCK grant `(sealed_gck, gck_version)`, so the
216 /// member's device can open the current GCK. `None` if the user holds no grant.
217 ///
218 /// Newest, not "the group's current generation": a member added before a rotation
219 /// they were not part of would have no row at the current version, and returning
220 /// nothing would be indistinguishable from not being a member.
221 #[tracing::instrument(skip_all)]
222 pub async fn get_member_grant(
223 pool: &PgPool,
224 group_id: SyncGroupId,
225 user_id: UserId,
226 ) -> Result<Option<(String, i32)>> {
227 let grant: Option<(String, i32)> = sqlx::query_as(
228 r"
229 SELECT sealed_gck, gck_version FROM sync_group_grants
230 WHERE group_id = $1 AND user_id = $2
231 ORDER BY gck_version DESC
232 LIMIT 1
233 ",
234 )
235 .bind(group_id)
236 .bind(user_id)
237 .fetch_optional(pool)
238 .await?;
239 Ok(grant)
240 }
241
242 /// Fetch a member's grant for one specific generation, so a device can decrypt
243 /// entries written before a rotation it lived through. `None` if the member never
244 /// held that generation.
245 #[tracing::instrument(skip_all)]
246 pub async fn get_member_grant_at(
247 pool: &PgPool,
248 group_id: SyncGroupId,
249 user_id: UserId,
250 gck_version: i32,
251 ) -> Result<Option<String>> {
252 let sealed: Option<String> = sqlx::query_scalar(
253 "SELECT sealed_gck FROM sync_group_grants WHERE group_id = $1 AND user_id = $2 AND gck_version = $3",
254 )
255 .bind(group_id)
256 .bind(user_id)
257 .bind(gck_version)
258 .fetch_optional(pool)
259 .await?;
260 Ok(sealed)
261 }
262
263 /// Add a member, or replace an existing member's grant (idempotent upsert).
264 ///
265 /// The admin calls this with a grant it sealed to the member's public key at the
266 /// group's current `gck_version`. Re-adding an existing member updates their role
267 /// and public key, and records their grant for that generation.
268 ///
269 /// A new member is granted the current generation only. Entries written under
270 /// earlier generations stay unreadable to them, which is the intended shape: a
271 /// member sees the group from when they joined, not before.
272 #[tracing::instrument(skip_all)]
273 pub async fn add_or_update_member(
274 pool: &PgPool,
275 group_id: SyncGroupId,
276 user_id: UserId,
277 role: &str,
278 sealed_gck: &str,
279 gck_version: i32,
280 member_pubkey: &str,
281 ) -> Result<()> {
282 let mut tx = pool.begin().await?;
283
284 sqlx::query(
285 r"
286 INSERT INTO sync_group_members (group_id, user_id, role, member_pubkey)
287 VALUES ($1, $2, $3, $4)
288 ON CONFLICT (group_id, user_id)
289 DO UPDATE SET role = EXCLUDED.role,
290 member_pubkey = EXCLUDED.member_pubkey
291 ",
292 )
293 .bind(group_id)
294 .bind(user_id)
295 .bind(role)
296 .bind(member_pubkey)
297 .execute(&mut *tx)
298 .await?;
299
300 sqlx::query(
301 r"
302 INSERT INTO sync_group_grants (group_id, user_id, gck_version, sealed_gck)
303 VALUES ($1, $2, $3, $4)
304 ON CONFLICT (group_id, user_id, gck_version)
305 DO UPDATE SET sealed_gck = EXCLUDED.sealed_gck
306 ",
307 )
308 .bind(group_id)
309 .bind(user_id)
310 .bind(gck_version)
311 .bind(sealed_gck)
312 .execute(&mut *tx)
313 .await?;
314
315 tx.commit().await?;
316 Ok(())
317 }
318
319 /// Remove a member. Returns `true` if a membership row was deleted.
320 ///
321 /// This drops the member's access to future group writes; forward secrecy for
322 /// writes after removal comes from the caller then rotating the GCK
323 /// ([`rotate_group_gck`]). Data the removed member already pulled is,
324 /// unavoidably, already in their hands.
325 #[tracing::instrument(skip_all)]
326 pub async fn remove_member(pool: &PgPool, group_id: SyncGroupId, user_id: UserId) -> Result<bool> {
327 let result = sqlx::query("DELETE FROM sync_group_members WHERE group_id = $1 AND user_id = $2")
328 .bind(group_id)
329 .bind(user_id)
330 .execute(pool)
331 .await?;
332 Ok(result.rows_affected() > 0)
333 }
334
335 /// Rotate the group's GCK to `new_version`, granting the new key to everyone in
336 /// `grants` and removing everyone else.
337 ///
338 /// The admin mints a fresh GCK, seals it to the remaining members' stored public
339 /// keys, and passes `(user_id, sealed_gck)` for each. In one transaction this
340 /// bumps `sync_groups.gck_version`, deletes any member not in `grants` (the
341 /// removed set) along with every grant they held, and records each provided grant
342 /// at `new_version`. Callers must include the admin's own re-sealed grant.
343 ///
344 /// Grants **accumulate**: a remaining member keeps their rows for earlier
345 /// generations, which is what lets them still read entries written before this
346 /// rotation. Only the removed set loses history, and only server-side; whatever
347 /// they already pulled is in their hands.
348 #[tracing::instrument(skip_all)]
349 pub async fn rotate_group_gck(
350 pool: &PgPool,
351 group_id: SyncGroupId,
352 new_version: i32,
353 grants: &[(UserId, String)],
354 ) -> Result<()> {
355 let mut tx = pool.begin().await?;
356
357 sqlx::query("UPDATE sync_groups SET gck_version = $1 WHERE id = $2")
358 .bind(new_version)
359 .bind(group_id)
360 .execute(&mut *tx)
361 .await?;
362
363 // Anyone not in the new grant set is removed by the rotation, and loses every
364 // generation they held rather than just the new one.
365 let keep: Vec<UserId> = grants.iter().map(|(u, _)| *u).collect();
366 sqlx::query("DELETE FROM sync_group_members WHERE group_id = $1 AND user_id <> ALL($2)")
367 .bind(group_id)
368 .bind(&keep)
369 .execute(&mut *tx)
370 .await?;
371 sqlx::query("DELETE FROM sync_group_grants WHERE group_id = $1 AND user_id <> ALL($2)")
372 .bind(group_id)
373 .bind(&keep)
374 .execute(&mut *tx)
375 .await?;
376
377 for (user_id, sealed_gck) in grants {
378 sqlx::query(
379 r"
380 INSERT INTO sync_group_grants (group_id, user_id, gck_version, sealed_gck)
381 VALUES ($1, $2, $3, $4)
382 ON CONFLICT (group_id, user_id, gck_version)
383 DO UPDATE SET sealed_gck = EXCLUDED.sealed_gck
384 ",
385 )
386 .bind(group_id)
387 .bind(user_id)
388 .bind(new_version)
389 .bind(sealed_gck)
390 .execute(&mut *tx)
391 .await?;
392 }
393
394 tx.commit().await?;
395 Ok(())
396 }
397