Sign in

pulls/pr_01m47d15m3e54sn21z27rpy5n9

public

pr_01m47d15m3e54sn21z27rpy5n9/services/identity/src/lib.rs

578 lines21,569 bytesCodeBlame
1//! The identity service: accounts, sessions, SSH keys and access tokens.
2//!
3//! Reached only through service bindings; see `g1t_contracts::identity` for
4//! the methods and their arguments.
5
6mod crypto;
7mod device;
8mod email;
9mod oauth;
10mod tokens;
11mod workspaces;
12
13use g1t_contracts::identity::*;
14use g1t_contracts::time::{SQL_NOW, rfc3339, sql_after};
15use g1t_contracts::{FailureCode, Outcome, User, Viewer, is_valid_namespace, new_id};
16use g1t_kit::{args, now_ms, reply, rpc_method};
17use serde::Deserialize;
18use tokens::TOKEN_PREFIX;
19use worker::wasm_bindgen::JsValue;
20use worker::{Context, D1Database, Env, Request, Response, Result, event};
21
22const SESSION_TTL_SECONDS: u64 = 30 * 24 * 60 * 60;
23const VERIFY_TTL_SECONDS: u64 = 24 * 60 * 60;
24const RESET_TTL_SECONDS: u64 = 60 * 60;
25const MIN_PASSWORD_LENGTH: usize = 10;
26const PASSWORD_TOO_SHORT: &str = "Use a password of at least 10 characters.";
27
28/// A user as selected from the database; `verified` arrives as 0 or 1.
29#[derive(Deserialize)]
30struct Account {
31 id: String,
32 username: String,
33 verified: u8,
34}
35
36impl From<Account> for User {
37 fn from(row: Account) -> Self {
38 User {
39 id: row.id,
40 username: row.username,
41 verified: row.verified != 0,
42 ..User::default()
43 }
44 }
45}
46
47#[derive(Deserialize)]
48struct UserRow {
49 id: String,
50 username: String,
51 password_hash: String,
52 verified: u8,
53}
54
55/// The owner of an emailed token.
56#[derive(Deserialize)]
57struct TokenOwner {
58 id: String,
59 username: String,
60 email: Option<String>,
61}
62
63#[derive(Deserialize)]
64struct KeyRow {
65 id: String,
66 title: String,
67 fingerprint: String,
68 created_at: String,
69}
70
71impl From<KeyRow> for SshKey {
72 fn from(row: KeyRow) -> Self {
73 SshKey {
74 id: row.id,
75 title: row.title,
76 fingerprint: row.fingerprint,
77 created_at: row.created_at,
78 }
79 }
80}
81
82struct Identity {
83 db: D1Database,
84 env: Env,
85}
86
87impl Identity {
88 /// Runs a query that returns at most one user, for showing to others:
89 /// without their workspaces.
90 async fn find_public_user(&self, sql: &str, param: &str) -> Result<Viewer> {
91 Ok(self
92 .db
93 .prepare(sql)
94 .bind(&[JsValue::from(param)])?
95 .first::<Account>(None)
96 .await?
97 .map(User::from))
98 }
99
100 /// Attaches the workspaces a user belongs to, so that any service can
101 /// authorize them without asking again.
102 async fn with_workspaces(&self, user: Viewer) -> Result<Viewer> {
103 let Some(mut user) = user else {
104 return Ok(None);
105 };
106 user.workspaces = self.memberships(&user.id).await?;
107 Ok(Some(user))
108 }
109
110 /// Runs a query that resolves credentials to at most one user.
111 async fn find_user(&self, sql: &str, param: &str) -> Result<Viewer> {
112 let user = self.find_public_user(sql, param).await?;
113 self.with_workspaces(user).await
114 }
115
116 /// Stores a one-time token of `kind` for the user and returns it.
117 async fn issue_email_token(&self, user_id: &str, kind: &str, ttl: u64) -> Result<String> {
118 let token = crypto::random_hex(32);
119 self.db
120 .prepare(format!(
121 "INSERT INTO email_tokens (id, user_id, kind, expires_at)
122 VALUES (?, ?, ?, {})",
123 sql_after(ttl)
124 ))
125 .bind(&[
126 crypto::sha256_hex(&token).into(),
127 user_id.into(),
128 kind.into(),
129 ])?
130 .run()
131 .await?;
132 Ok(token)
133 }
134
135 /// Consumes a token of `kind`, returning its owner if it was valid.
136 async fn redeem_email_token(&self, token: &str, kind: &str) -> Result<Option<TokenOwner>> {
137 let id = crypto::sha256_hex(token);
138 let owner = self
139 .db
140 .prepare(format!(
141 "SELECT users.id, users.username, users.email FROM email_tokens
142 JOIN users ON users.id = email_tokens.user_id
143 WHERE email_tokens.id = ? AND email_tokens.kind = ?
144 AND email_tokens.expires_at > {SQL_NOW}"
145 ))
146 .bind(&[id.as_str().into(), kind.into()])?
147 .first::<TokenOwner>(None)
148 .await?;
149 if let Some(owner) = &owner {
150 // Every outstanding token of this kind dies with the one used.
151 self.db
152 .prepare("DELETE FROM email_tokens WHERE user_id = ? AND kind = ?")
153 .bind(&[owner.id.as_str().into(), kind.into()])?
154 .run()
155 .await?;
156 }
157 Ok(owner)
158 }
159
160 async fn send_verification(&self, user: &User, email: &str) -> Result<()> {
161 let token = self
162 .issue_email_token(&user.id, "verify", VERIFY_TTL_SECONDS)
163 .await?;
164 email::send_verification(&self.env, email, &user.username, &token).await
165 }
166
167 async fn resend_verification(&self, a: UserArgs) -> Result<Outcome<bool>> {
168 let row = self
169 .db
170 .prepare(
171 "SELECT id, username, email FROM users WHERE id = ? AND email_verified_at IS NULL",
172 )
173 .bind(&[a.user.id.as_str().into()])?
174 .first::<TokenOwner>(None)
175 .await?;
176 let Some(TokenOwner {
177 email: Some(email), ..
178 }) = row
179 else {
180 return Ok(Outcome::fail(
181 FailureCode::Conflict,
182 "This account's email is already confirmed.",
183 ));
184 };
185 self.send_verification(&a.user, &email).await?;
186 Ok(Outcome::Ok(true))
187 }
188
189 async fn verify_email(&self, a: EmailTokenArgs) -> Result<Outcome<User>> {
190 let Some(owner) = self.redeem_email_token(&a.token, "verify").await? else {
191 return Ok(Outcome::fail(
192 FailureCode::Invalid,
193 "This confirmation link is not valid or has expired.",
194 ));
195 };
196 self.db
197 .prepare(format!(
198 "UPDATE users SET email_verified_at = {SQL_NOW} WHERE id = ?"
199 ))
200 .bind(&[owner.id.as_str().into()])?
201 .run()
202 .await?;
203 Ok(Outcome::Ok(User {
204 id: owner.id,
205 username: owner.username,
206 verified: true,
207 ..User::default()
208 }))
209 }
210
211 async fn request_password_reset(&self, a: EmailArgs) -> Result<bool> {
212 let row = self
213 .db
214 .prepare("SELECT id, username, email FROM users WHERE email = ?")
215 .bind(&[a.email.trim().to_lowercase().into()])?
216 .first::<TokenOwner>(None)
217 .await?;
218 if let Some(TokenOwner {
219 id,
220 username,
221 email: Some(email),
222 }) = row
223 {
224 let token = self
225 .issue_email_token(&id, "reset", RESET_TTL_SECONDS)
226 .await?;
227 email::send_password_reset(&self.env, &email, &username, &token).await?;
228 }
229 // The same answer either way, so addresses cannot be probed.
230 Ok(true)
231 }
232
233 async fn reset_password(&self, a: ResetPasswordArgs) -> Result<Outcome<User>> {
234 if a.password.chars().count() < MIN_PASSWORD_LENGTH {
235 return Ok(Outcome::fail(FailureCode::Invalid, PASSWORD_TOO_SHORT));
236 }
237 let Some(owner) = self.redeem_email_token(&a.token, "reset").await? else {
238 return Ok(Outcome::fail(
239 FailureCode::Invalid,
240 "This reset link is not valid or has expired.",
241 ));
242 };
243 // Following an emailed link also proves the address.
244 self.db
245 .prepare(format!(
246 "UPDATE users SET password_hash = ?,
247 email_verified_at = COALESCE(email_verified_at, {SQL_NOW})
248 WHERE id = ?"
249 ))
250 .bind(&[
251 crypto::hash_password(&a.password).into(),
252 owner.id.as_str().into(),
253 ])?
254 .run()
255 .await?;
256 // Anyone signed in with the old password is signed out.
257 self.db
258 .prepare("DELETE FROM sessions WHERE user_id = ?")
259 .bind(&[owner.id.as_str().into()])?
260 .run()
261 .await?;
262 Ok(Outcome::Ok(User {
263 id: owner.id,
264 username: owner.username,
265 verified: true,
266 ..User::default()
267 }))
268 }
269
270 async fn user_for_password(&self, username: &str, password: &str) -> Result<Viewer> {
271 let row = self
272 .db
273 .prepare("SELECT id, username, password_hash, email_verified_at IS NOT NULL AS verified FROM users WHERE username = ?")
274 .bind(&[JsValue::from(username.to_lowercase())])?
275 .first::<UserRow>(None)
276 .await?;
277 let user = row
278 .filter(|row| crypto::verify_password(password, &row.password_hash))
279 .map(|row| User {
280 id: row.id,
281 username: row.username,
282 verified: row.verified != 0,
283 ..User::default()
284 });
285 self.with_workspaces(user).await
286 }
287
288 async fn register(&self, a: RegisterArgs) -> Result<Outcome<SignedIn>> {
289 let username = a.username.trim().to_lowercase();
290 let email = a.email.trim().to_lowercase();
291 let invalid = |message: &str| Ok(Outcome::fail(FailureCode::Invalid, message));
292 if !is_valid_namespace(&username) {
293 return invalid(
294 "Usernames use lowercase letters, digits and single hyphens, up to 39 characters.",
295 );
296 }
297 let well_formed_email = email
298 .split_once('@')
299 .is_some_and(|(local, domain)| !local.is_empty() && domain.contains('.'))
300 && !email.contains(char::is_whitespace);
301 if !well_formed_email {
302 return invalid("Enter a valid email address.");
303 }
304 if a.password.chars().count() < MIN_PASSWORD_LENGTH {
305 return invalid(PASSWORD_TOO_SHORT);
306 }
307 let taken = self
308 .db
309 // Usernames and workspaces share one namespace, so that a name
310 // means the same thing wherever it appears.
311 .prepare(
312 "SELECT username FROM users WHERE username = ? OR email = ?
313 UNION ALL SELECT slug FROM workspaces WHERE slug = ?",
314 )
315 .bind(&[
316 username.as_str().into(),
317 email.as_str().into(),
318 username.as_str().into(),
319 ])?
320 .first::<serde_json::Value>(None)
321 .await?;
322 if taken.is_some() {
323 return Ok(Outcome::fail(
324 FailureCode::Conflict,
325 "That username or email is already registered.",
326 ));
327 }
328 let user = User {
329 id: new_id("usr", now_ms()),
330 username,
331 ..User::default()
332 };
333 self.db
334 .prepare("INSERT INTO users (id, username, email, password_hash) VALUES (?, ?, ?, ?)")
335 .bind(&[
336 user.id.as_str().into(),
337 user.username.as_str().into(),
338 email.as_str().into(),
339 crypto::hash_password(&a.password).into(),
340 ])?
341 .run()
342 .await?;
343 // The account exists either way; the email can be sent again later.
344 if let Err(error) = self.send_verification(&user, &email).await {
345 worker::console_error!("verification email failed: {error}");
346 }
347 self.start_session(user).await
348 }
349
350 async fn sign_in(&self, a: SignInArgs) -> Result<Outcome<SignedIn>> {
351 let Some(user) = self.user_for_password(&a.username, &a.password).await? else {
352 return Ok(Outcome::fail(
353 FailureCode::Unauthenticated,
354 "Incorrect username or password.",
355 ));
356 };
357 self.start_session(user).await
358 }
359
360 async fn start_session(&self, user: User) -> Result<Outcome<SignedIn>> {
361 let session_token = crypto::random_hex(32);
362 self.db
363 .prepare(format!(
364 "INSERT INTO sessions (id, user_id, expires_at) VALUES (?, ?, {})",
365 sql_after(SESSION_TTL_SECONDS)
366 ))
367 .bind(&[
368 crypto::sha256_hex(&session_token).into(),
369 user.id.as_str().into(),
370 ])?
371 .run()
372 .await?;
373 Ok(Outcome::Ok(SignedIn {
374 user,
375 session_token,
376 }))
377 }
378
379 async fn sign_out(&self, a: SessionArgs) -> Result<()> {
380 self.db
381 .prepare("DELETE FROM sessions WHERE id = ?")
382 .bind(&[crypto::sha256_hex(&a.session_token).into()])?
383 .run()
384 .await?;
385 Ok(())
386 }
387
388 async fn user_for_session(&self, a: SessionArgs) -> Result<Viewer> {
389 self.find_user(
390 &format!(
391 "SELECT users.id, users.username, users.email_verified_at IS NOT NULL AS verified
392 FROM sessions JOIN users ON users.id = sessions.user_id
393 WHERE sessions.id = ? AND sessions.expires_at > {SQL_NOW}"
394 ),
395 &crypto::sha256_hex(&a.session_token),
396 )
397 .await
398 }
399
400 async fn user_for_git_credentials(&self, a: GitCredentialsArgs) -> Result<Viewer> {
401 // Like GitHub, a token alone identifies its user.
402 if a.secret.starts_with(TOKEN_PREFIX) {
403 self.user_for_access_token(&a.secret).await
404 } else {
405 self.user_for_password(&a.username, &a.secret).await
406 }
407 }
408
409 async fn user_for_ssh_key(&self, a: FingerprintArgs) -> Result<Viewer> {
410 self.find_user(
411 "SELECT users.id, users.username, users.email_verified_at IS NOT NULL AS verified FROM ssh_keys
412 JOIN users ON users.id = ssh_keys.user_id
413 WHERE fingerprint = ?",
414 &a.fingerprint,
415 )
416 .await
417 }
418
419 async fn user_by_username(&self, a: UsernameArgs) -> Result<Viewer> {
420 self.find_public_user(
421 "SELECT id, username, email_verified_at IS NOT NULL AS verified FROM users WHERE username = ?",
422 &a.username.to_lowercase(),
423 )
424 .await
425 }
426
427 async fn usernames(&self, a: UsernamesArgs) -> Result<std::collections::HashMap<String, String>> {
428 #[derive(serde::Deserialize)]
429 struct Named {
430 id: String,
431 name: String,
432 }
433 let ids: Vec<String> = a.ids.into_iter().take(200).collect();
434 let mut names = std::collections::HashMap::new();
435 if ids.is_empty() {
436 return Ok(names);
437 }
438 let marks = vec!["?"; ids.len()].join(", ");
439 let bind: Vec<worker::wasm_bindgen::JsValue> = ids.iter().map(|id| id.as_str().into()).collect();
440 for sql in [
441 format!("SELECT id, username AS name FROM users WHERE id IN ({marks})"),
442 format!("SELECT id, slug AS name FROM workspaces WHERE id IN ({marks})"),
443 ] {
444 for row in self.db.prepare(sql).bind(&bind)?.all().await?.results::<Named>()? {
445 names.insert(row.id, row.name);
446 }
447 }
448 Ok(names)
449 }
450
451 async fn list_ssh_keys(&self, a: UserArgs) -> Result<Vec<SshKey>> {
452 let rows = self
453 .db
454 .prepare("SELECT id, title, fingerprint, created_at FROM ssh_keys WHERE user_id = ? ORDER BY id")
455 .bind(&[a.user.id.into()])?
456 .all()
457 .await?
458 .results::<KeyRow>()?;
459 Ok(rows.into_iter().map(SshKey::from).collect())
460 }
461
462 async fn add_ssh_key(&self, a: AddSshKeyArgs) -> Result<Outcome<SshKey>> {
463 let Some(key) = crypto::parse_ssh_key(&a.public_key) else {
464 return Ok(Outcome::fail(
465 FailureCode::Invalid,
466 "That is not a valid OpenSSH public key.",
467 ));
468 };
469 let taken = self
470 .db
471 .prepare("SELECT id FROM ssh_keys WHERE fingerprint = ?")
472 .bind(&[key.fingerprint.as_str().into()])?
473 .first::<serde_json::Value>(None)
474 .await?;
475 if taken.is_some() {
476 return Ok(Outcome::fail(
477 FailureCode::Conflict,
478 "That key is already registered.",
479 ));
480 }
481 let now = now_ms();
482 let title = [a.title.trim(), key.comment.as_str(), "SSH key"]
483 .into_iter()
484 .find(|candidate| !candidate.is_empty())
485 .unwrap_or_default()
486 .to_owned();
487 let row = KeyRow {
488 id: new_id("key", now),
489 title,
490 fingerprint: key.fingerprint,
491 created_at: rfc3339(now),
492 };
493 self.db
494 .prepare(
495 "INSERT INTO ssh_keys (id, user_id, title, public_key, fingerprint, created_at)
496 VALUES (?, ?, ?, ?, ?, ?)",
497 )
498 .bind(&[
499 row.id.as_str().into(),
500 a.user.id.into(),
501 row.title.as_str().into(),
502 key.public_key.into(),
503 row.fingerprint.as_str().into(),
504 row.created_at.as_str().into(),
505 ])?
506 .run()
507 .await?;
508 Ok(Outcome::Ok(row.into()))
509 }
510
511 /// Deletes a row the user owns from `table`.
512 async fn remove(&self, table: &str, a: RemoveArgs) -> Result<()> {
513 self.db
514 .prepare(format!("DELETE FROM {table} WHERE id = ? AND user_id = ?"))
515 .bind(&[a.id.into(), a.user.id.into()])?
516 .run()
517 .await?;
518 Ok(())
519 }
520}
521
522#[event(fetch)]
523async fn fetch(mut request: Request, env: Env, _ctx: Context) -> Result<Response> {
524 let Some(method) = rpc_method(&request) else {
525 return Response::error("Not found", 404);
526 };
527 let body: serde_json::Value = request.json().await?;
528 let identity = Identity {
529 db: env.d1("DB")?,
530 env,
531 };
532
533 match method.as_str() {
534 "register" => reply(&identity.register(args(body)?).await?),
535 "sign_in" => reply(&identity.sign_in(args(body)?).await?),
536 "create_workspace" => reply(&identity.create_workspace(args(body)?).await?),
537 "get_workspace" => reply(&identity.get_workspace(args(body)?).await?),
538 "list_members" => reply(&identity.list_members(args(body)?).await?),
539 "add_member" => reply(&identity.add_member(args(body)?).await?),
540 "remove_member" => reply(&identity.remove_member(args(body)?).await?),
541 "update_workspace" => reply(&identity.update_workspace(args(body)?).await?),
542 "list_workspace_tokens" => reply(&identity.list_workspace_tokens(args(body)?).await?),
543 "create_workspace_token" => reply(&identity.create_workspace_token(args(body)?).await?),
544 "remove_workspace_token" => reply(&identity.remove_workspace_token(args(body)?).await?),
545 "oauth_authorize" => reply(&identity.oauth_authorize(args(body)?).await?),
546 "oauth_exchange" => reply(&identity.oauth_exchange(args(body)?).await?),
547 "oauth_refresh" => reply(&identity.oauth_refresh(args(body)?).await?),
548 "list_oauth_grants" => reply(&identity.list_oauth_grants(args(body)?).await?),
549 "revoke_oauth_grant" => reply(&identity.revoke_oauth_grant(args(body)?).await?),
550 "device_start" => reply(&identity.device_start(args(body)?).await?),
551 "device_lookup" => reply(&identity.device_lookup(args(body)?).await?),
552 "device_resolve" => reply(&identity.device_resolve(args(body)?).await?),
553 "device_claim" => reply(&identity.device_claim(args(body)?).await?),
554 "resend_verification" => reply(&identity.resend_verification(args(body)?).await?),
555 "verify_email" => reply(&identity.verify_email(args(body)?).await?),
556 "request_password_reset" => reply(&identity.request_password_reset(args(body)?).await?),
557 "reset_password" => reply(&identity.reset_password(args(body)?).await?),
558 "sign_out" => reply(&identity.sign_out(args(body)?).await?),
559 "user_for_session" => reply(&identity.user_for_session(args(body)?).await?),
560 "user_for_git_credentials" => reply(&identity.user_for_git_credentials(args(body)?).await?),
561 "user_for_access_token" => {
562 let a: TokenArgs = args(body)?;
563 reply(&identity.user_for_access_token(&a.token).await?)
564 }
565 "user_for_ssh_key" => reply(&identity.user_for_ssh_key(args(body)?).await?),
566 "user_by_username" => reply(&identity.user_by_username(args(body)?).await?),
567 "usernames" => reply(&identity.usernames(args(body)?).await?),
568 "list_ssh_keys" => reply(&identity.list_ssh_keys(args(body)?).await?),
569 "add_ssh_key" => reply(&identity.add_ssh_key(args(body)?).await?),
570 "remove_ssh_key" => reply(&identity.remove("ssh_keys", args(body)?).await?),
571 "list_access_tokens" => reply(&identity.list_access_tokens(args(body)?).await?),
572 "create_access_token" => reply(&identity.create_access_token(args(body)?).await?),
573 "create_agent_token" => reply(&identity.create_agent_token(args(body)?).await?),
574 "agent_scope" => reply(&identity.agent_scope(args(body)?).await?),
575 "remove_access_token" => reply(&identity.remove("access_tokens", args(body)?).await?),
576 _ => Response::error("Unknown method", 404),
577 }
578}