Skip to main content

max / makenotwork

9.5 KB · 305 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, sealed_gck, gck_version, member_pubkey)
53 VALUES ($1, $2, 'admin', $3, $4, $5)
54 ",
55 )
56 .bind(group.id)
57 .bind(admin_user_id)
58 .bind(admin_sealed_gck)
59 .bind(group.gck_version)
60 .bind(admin_pubkey)
61 .execute(&mut *tx)
62 .await?;
63
64 tx.commit().await?;
65 Ok(group)
66 }
67
68 /// Fetch a group scoped to its app (the app scope guards against cross-app id
69 /// guessing).
70 #[tracing::instrument(skip_all)]
71 pub async fn get_group(
72 pool: &PgPool,
73 app_id: SyncAppId,
74 group_id: SyncGroupId,
75 ) -> Result<Option<DbSyncGroup>> {
76 let group = sqlx::query_as::<_, DbSyncGroup>(
77 r"
78 SELECT id, app_id, admin_user_id, name, gck_version, created_at
79 FROM sync_groups
80 WHERE app_id = $1 AND id = $2
81 ",
82 )
83 .bind(app_id)
84 .bind(group_id)
85 .fetch_optional(pool)
86 .await?;
87 Ok(group)
88 }
89
90 /// List the groups a user belongs to within an app, most recent first.
91 #[tracing::instrument(skip_all)]
92 pub async fn list_groups_for_user(
93 pool: &PgPool,
94 app_id: SyncAppId,
95 user_id: UserId,
96 ) -> Result<Vec<DbSyncGroup>> {
97 let groups = sqlx::query_as::<_, DbSyncGroup>(
98 r"
99 SELECT g.id, g.app_id, g.admin_user_id, g.name, g.gck_version, g.created_at
100 FROM sync_groups g
101 JOIN sync_group_members m ON m.group_id = g.id
102 WHERE g.app_id = $1 AND m.user_id = $2
103 ORDER BY g.created_at DESC
104 ",
105 )
106 .bind(app_id)
107 .bind(user_id)
108 .fetch_all(pool)
109 .await?;
110 Ok(groups)
111 }
112
113 /// All members of a group.
114 #[tracing::instrument(skip_all)]
115 pub async fn list_members(pool: &PgPool, group_id: SyncGroupId) -> Result<Vec<DbSyncGroupMember>> {
116 let members = sqlx::query_as::<_, DbSyncGroupMember>(
117 r"
118 SELECT m.group_id, m.user_id, u.email, m.role, m.sealed_gck, m.gck_version, m.added_at
119 FROM sync_group_members m
120 JOIN users u ON u.id = m.user_id
121 WHERE m.group_id = $1
122 ORDER BY m.added_at ASC
123 ",
124 )
125 .bind(group_id)
126 .fetch_all(pool)
127 .await?;
128 Ok(members)
129 }
130
131 /// The `(user_id, member_pubkey)` of every member with a stored public key. The
132 /// admin fetches this to re-seal a rotated GCK to the remaining members without
133 /// re-collecting their keys out of band. Members without a stored key (legacy
134 /// rows) are omitted.
135 #[tracing::instrument(skip_all)]
136 pub async fn list_member_pubkeys(
137 pool: &PgPool,
138 group_id: SyncGroupId,
139 ) -> Result<Vec<(UserId, String)>> {
140 let rows: Vec<(UserId, String)> = sqlx::query_as(
141 "SELECT user_id, member_pubkey FROM sync_group_members \
142 WHERE group_id = $1 AND member_pubkey IS NOT NULL \
143 ORDER BY added_at ASC",
144 )
145 .bind(group_id)
146 .fetch_all(pool)
147 .await?;
148 Ok(rows)
149 }
150
151 /// Whether `user_id` is a member of `group_id`. The membership gate for
152 /// group-scoped push/pull.
153 #[tracing::instrument(skip_all)]
154 pub async fn is_group_member(
155 pool: &PgPool,
156 group_id: SyncGroupId,
157 user_id: UserId,
158 ) -> Result<bool> {
159 let exists: bool = sqlx::query_scalar(
160 "SELECT EXISTS (SELECT 1 FROM sync_group_members WHERE group_id = $1 AND user_id = $2)",
161 )
162 .bind(group_id)
163 .bind(user_id)
164 .fetch_one(pool)
165 .await?;
166 Ok(exists)
167 }
168
169 /// Whether `user_id` is the admin of `group_id`. Admin-only actions (add/remove
170 /// member, rotate) gate on this.
171 #[tracing::instrument(skip_all)]
172 pub async fn is_group_admin(pool: &PgPool, group_id: SyncGroupId, user_id: UserId) -> Result<bool> {
173 let exists: bool = sqlx::query_scalar(
174 "SELECT EXISTS (SELECT 1 FROM sync_groups WHERE id = $1 AND admin_user_id = $2)",
175 )
176 .bind(group_id)
177 .bind(user_id)
178 .fetch_one(pool)
179 .await?;
180 Ok(exists)
181 }
182
183 /// Fetch a member's current sealed GCK grant `(sealed_gck, gck_version)`, so the
184 /// member's device can open the GCK and read the group changelog. `None` if the
185 /// user is not a member.
186 #[tracing::instrument(skip_all)]
187 pub async fn get_member_grant(
188 pool: &PgPool,
189 group_id: SyncGroupId,
190 user_id: UserId,
191 ) -> Result<Option<(String, i32)>> {
192 let grant: Option<(String, i32)> = sqlx::query_as(
193 "SELECT sealed_gck, gck_version FROM sync_group_members WHERE group_id = $1 AND user_id = $2",
194 )
195 .bind(group_id)
196 .bind(user_id)
197 .fetch_optional(pool)
198 .await?;
199 Ok(grant)
200 }
201
202 /// Add a member, or replace an existing member's grant (idempotent upsert).
203 ///
204 /// The admin calls this with a grant it sealed to the member's public key at the
205 /// group's current `gck_version`. Re-adding an existing member updates their
206 /// grant and role in place.
207 #[tracing::instrument(skip_all)]
208 pub async fn add_or_update_member(
209 pool: &PgPool,
210 group_id: SyncGroupId,
211 user_id: UserId,
212 role: &str,
213 sealed_gck: &str,
214 gck_version: i32,
215 member_pubkey: &str,
216 ) -> Result<()> {
217 sqlx::query(
218 r"
219 INSERT INTO sync_group_members (group_id, user_id, role, sealed_gck, gck_version, member_pubkey)
220 VALUES ($1, $2, $3, $4, $5, $6)
221 ON CONFLICT (group_id, user_id)
222 DO UPDATE SET role = EXCLUDED.role,
223 sealed_gck = EXCLUDED.sealed_gck,
224 gck_version = EXCLUDED.gck_version,
225 member_pubkey = EXCLUDED.member_pubkey
226 ",
227 )
228 .bind(group_id)
229 .bind(user_id)
230 .bind(role)
231 .bind(sealed_gck)
232 .bind(gck_version)
233 .bind(member_pubkey)
234 .execute(pool)
235 .await?;
236 Ok(())
237 }
238
239 /// Remove a member. Returns `true` if a membership row was deleted.
240 ///
241 /// This drops the member's access to future group writes; forward secrecy for
242 /// writes after removal comes from the caller then rotating the GCK
243 /// ([`rotate_group_gck`]). Data the removed member already pulled is,
244 /// unavoidably, already in their hands.
245 #[tracing::instrument(skip_all)]
246 pub async fn remove_member(pool: &PgPool, group_id: SyncGroupId, user_id: UserId) -> Result<bool> {
247 let result = sqlx::query("DELETE FROM sync_group_members WHERE group_id = $1 AND user_id = $2")
248 .bind(group_id)
249 .bind(user_id)
250 .execute(pool)
251 .await?;
252 Ok(result.rows_affected() > 0)
253 }
254
255 /// Rotate the group's GCK to `new_version`, atomically replacing every member's
256 /// grant with one sealed under the new key.
257 ///
258 /// The admin mints a fresh GCK, seals it to the remaining members' stored public
259 /// keys, and passes `(user_id, sealed_gck)` for each. In one transaction this
260 /// bumps `sync_groups.gck_version`, deletes any member not in `grants` (the
261 /// removed set), and upserts each provided grant at `new_version`. Callers must
262 /// include the admin's own re-sealed grant.
263 #[tracing::instrument(skip_all)]
264 pub async fn rotate_group_gck(
265 pool: &PgPool,
266 group_id: SyncGroupId,
267 new_version: i32,
268 grants: &[(UserId, String)],
269 ) -> Result<()> {
270 let mut tx = pool.begin().await?;
271
272 sqlx::query("UPDATE sync_groups SET gck_version = $1 WHERE id = $2")
273 .bind(new_version)
274 .bind(group_id)
275 .execute(&mut *tx)
276 .await?;
277
278 // Anyone not in the new grant set is removed by the rotation.
279 let keep: Vec<UserId> = grants.iter().map(|(u, _)| *u).collect();
280 sqlx::query("DELETE FROM sync_group_members WHERE group_id = $1 AND user_id <> ALL($2)")
281 .bind(group_id)
282 .bind(&keep)
283 .execute(&mut *tx)
284 .await?;
285
286 for (user_id, sealed_gck) in grants {
287 sqlx::query(
288 r"
289 UPDATE sync_group_members
290 SET sealed_gck = $3, gck_version = $4
291 WHERE group_id = $1 AND user_id = $2
292 ",
293 )
294 .bind(group_id)
295 .bind(user_id)
296 .bind(sealed_gck)
297 .bind(new_version)
298 .execute(&mut *tx)
299 .await?;
300 }
301
302 tx.commit().await?;
303 Ok(())
304 }
305