clover's git

home-infra

1	use crate::*;
2	use base64::{
3	Engine,
4	engine::general_purpose::{STANDARD, URL_SAFE_NO_PAD},
5	};
6	use rusqlite::{Connection, OpenFlags, OptionalExtension};
7	use sha2::{Digest, Sha256};
8	use std::os::unix::fs::{DirBuilderExt, PermissionsExt};
9
10	const CATALOGS: &[(&str, &str, &[&str])] = &[
11	(
12	"observability",
13	"Logs and traces",
14	&["observability:read", "offline_access"],
15	),
16	(
17	"agents",
18	"Local agents",
19	&["sessions:read", "sessions:write", "offline_access"],
20	),
21	(
22	"shale",
23	"Shale",
24	&["shale:read", "shale:write", "offline_access"],
25	),
26	];
27
28	pub struct Store {
29	pub(crate) db: Mutex<Connection>,
30	pub origin: url::Url,
31	}
32	pub(crate) fn secret() -> String {
33	URL_SAFE_NO_PAD.encode(rand::random::<[u8; 32]>())
34	}
35	pub(crate) fn hash(value: &str) -> String {
36	format!("{:x}", Sha256::digest(value.as_bytes()))
37	}
38	pub(crate) fn get(db: &Connection, key: &str) -> Result<Value> {
39	let row: Option<String> = db
40	.query_row(
41	"SELECT value FROM records WHERE key=? AND (expires=0 OR expires>?)",
42	rusqlite::params![key, now() as i64],
43	|r| r.get(0),
44	)
45	.optional()?;
46	Ok(row
47	.map(|s| serde_json::from_str(&s))
48	.transpose()?
49	.unwrap_or(Value::Null))
50	}
51	pub(crate) fn put(db: &Connection, key: &str, value: &Value, ttl: i64) -> Result<()> {
52	db.execute(
53	"DELETE FROM records WHERE expires>0 AND expires<=?",
54	[now() as i64],
55	)?;
56	let count: i64 = db.query_row("SELECT count(*) FROM records", [], |r| r.get(0))?;
57	if count >= 16384 && get(db, key)?.is_null() {
58	return Err(Error::new(
59	503,
60	"The connection store is full. Remove an unused connection and retry.",
61	));
62	}
63	db.execute(
64	"INSERT OR REPLACE INTO records VALUES (?,?,?)",
65	rusqlite::params![
66	key,
67	value.to_string(),
68	if ttl == 0 { 0 } else { now() as i64 + ttl }
69	],
70	)?;
71	Ok(())
72	}
73	pub(crate) fn delete(db: &Connection, key: &str) -> Result<()> {
74	db.execute("DELETE FROM records WHERE key=?", [key])?;
75	Ok(())
76	}
77	pub(crate) fn revoke(db: &Connection, grant: &str) -> Result<()> {
78	db.execute(
79	"DELETE FROM records WHERE key=? OR json_extract(value,'$.grant')=?",
80	rusqlite::params![format!("grant:{grant}"), grant],
81	)?;
82	Ok(())
83	}
84	pub(crate) fn list(db: &Connection, prefix: &str) -> Result<Vec<Value>> {
85	let mut query = db.prepare(
86	"SELECT value FROM records WHERE substr(key,1,?)=? AND (expires=0 OR expires>?)",
87	)?;
88	let rows = query.query_map(rusqlite::params![prefix.len(), prefix, now() as i64], |r| {
89	r.get::<_, String>(0)
90	})?;
91	rows.map(|r| Ok(serde_json::from_str(&r?)?)).collect()
92	}
93	fn fail(code: &str) -> Error {
94	Error::new(400, code)
95	}
96	fn redirect(value: &str) -> Result<url::Url> {
97	let url = url::Url::parse(value).map_err(|_| fail("invalid_redirect_uri"))?;
98	let loopback = matches!(url.host_str(), Some("localhost" | "127.0.0.1" | "[::1]"));
99	if value.len() > 4096
100	|| !url.username().is_empty()
101	|| url.password().is_some()
102	|| url.fragment().is_some()
103	|| url.host_str().is_none()
104	|| !(url.scheme() == "https" || url.scheme() == "http" && loopback)
105	{
106	return Err(fail("invalid_redirect_uri"));
107	}
108	Ok(url)
109	}
110	impl Store {
111	pub fn new(data: &std::path::Path, origin: &str) -> Result<Self> {
112	let origin = url::Url::parse(origin)?;
113	if origin.path() != "/"
114	|| origin.query().is_some()
115	|| origin.fragment().is_some()
116	|| !origin.username().is_empty()
117	|| origin.password().is_some()
118	|| origin.scheme() != "https"
119	&& !(origin.scheme() == "http"
120	&& matches!(origin.host_str(), Some("localhost" | "127.0.0.1" | "[::1]")))
121	{
122	return Err(fail("Set an HTTPS origin for MCP connections."));
123	}
124	std::fs::DirBuilder::new()
125	.recursive(true)
126	.mode(0o700)
127	.create(data)?;
128	let path = data.join("connections.sqlite");
129	let db = Connection::open_with_flags(
130	&path,
131	OpenFlags::SQLITE_OPEN_READ_WRITE
132	| OpenFlags::SQLITE_OPEN_CREATE
133	| OpenFlags::SQLITE_OPEN_NOFOLLOW,
134	)?;
135	std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600))?;
136	db.execute_batch("PRAGMA journal_mode=WAL; PRAGMA busy_timeout=5000; CREATE TABLE IF NOT EXISTS records (key TEXT PRIMARY KEY, value TEXT NOT NULL, expires INTEGER NOT NULL);")?;
137	Ok(Self {
138	db: Mutex::new(db),
139	origin,
140	})
141	}
142	pub fn resource(&self, catalog: &str) -> String {
143	self.origin
144	.join(&format!("mcp/{catalog}"))
145	.unwrap()
146	.to_string()
147	}
148	fn client(
149	&self,
150	db: &Connection,
151	input: &HashMap<String, String>,
152	headers: &HeaderMap,
153	) -> Result<Value> {
154	let basic = headers.get("authorization").and_then(|v| v.to_str().ok());
155	let (id, credential) = if let Some(basic) = basic {
156	if input.contains_key("client_secret") {
157	return Err(fail("invalid_client"));
158	}
159	let bytes = STANDARD
160	.decode(
161	basic
162	.strip_prefix("Basic ")
163	.ok_or_else(|| fail("invalid_client"))?,
164	)
165	.map_err(|_| fail("invalid_client"))?;
166	let value = String::from_utf8(bytes).map_err(|_| fail("invalid_client"))?;
167	let (id, credential) = value
168	.split_once(':')
169	.ok_or_else(|| fail("invalid_client"))?;
170	(id.to_owned(), Some(credential.to_owned()))
171	} else {
172	(
173	input.get("client_id").cloned().unwrap_or_default(),
174	input.get("client_secret").cloned(),
175	)
176	};
177	if input.get("client_id").is_some_and(|v| v != &id) {
178	return Err(fail("invalid_client"));
179	}
180	let client = get(db, &format!("client:{id}"))?;
181	let method = string(&client["token_endpoint_auth_method"]);
182	let valid = match method {
183	"none" => basic.is_none() && credential.is_none(),
184	"client_secret_basic" => {
185	basic.is_some()
186	&& credential.as_ref().is_some_and(|v| {
187	hash(v)
188	.as_bytes()
189	.ct_eq(string(&client["secret_hash"]).as_bytes())
190	.unwrap_u8()
191	== 1
192	})
193	}
194	"client_secret_post" => {
195	basic.is_none()
196	&& credential.as_ref().is_some_and(|v| {
197	hash(v)
198	.as_bytes()
199	.ct_eq(string(&client["secret_hash"]).as_bytes())
200	.unwrap_u8()
201	== 1
202	})
203	}
204	_ => false,
205	};
206	if valid {
207	Ok(client)
208	} else {
209	Err(fail("invalid_client"))
210	}
211	}
212	fn issue(&self, db: &Connection, grant: &Value) -> Result<Value> {
213	let access = secret();
214	put(
215	db,
216	&format!("access:{}", hash(&access)),
217	&json!({"grant":grant["id"],"resource":grant["resource"]}),
218	3600,
219	)?;
220	let mut response = json!({"access_token":access,"token_type":"Bearer","expires_in":3600,"scope":array(&grant["scopes"]).iter().map(string).collect::<Vec<_>>().join(" ")});
221	if array(&grant["scopes"])
222	.iter()
223	.any(|s| s == "offline_access")
224	{
225	let refresh = secret();
226	put(
227	db,
228	&format!("refresh:{}", hash(&refresh)),
229	&json!({"grant":grant["id"]}),
230	30 * 86400,
231	)?;
232	response["refresh_token"] = json!(refresh);
233	}
234	Ok(response)
235	}
236	pub fn authenticate(&self, headers: &HeaderMap, resource: &str) -> Result<Value> {
237	let token = headers
238	.get("authorization")
239	.and_then(|v| v.to_str().ok())
240	.and_then(|s| s.strip_prefix("Bearer "))
241	.filter(|s| !s.is_empty() && s.len() <= 256)
242	.ok_or_else(|| Error::new(401, "invalid_token"))?;
243	let db = self.db.lock().unwrap();
244	let access = get(&db, &format!("access:{}", hash(token)))?;
245	let grant = get(&db, &format!("grant:{}", string(&access["grant"])))?;
246	if grant.is_null() || access["resource"] != resource || grant["resource"] != resource {
247	return Err(Error::new(401, "invalid_token"));
248	}
249	Ok(grant)
250	}
251	fn exchange(
252	&self,
253	db: &Connection,
254	input: &HashMap<String, String>,
255	headers: &HeaderMap,
256	) -> Result<(String, Value)> {
257	let client = self.client(db, input, headers)?;
258	let (key, grant) = match input.get("grant_type").map(String::as_str) {
259	Some("authorization_code") => {
260	let key = format!(
261	"code:{}",
262	hash(input.get("code").map(String::as_str).unwrap_or_default())
263	);
264	let code = get(db, &key)?;
265	let verifier = input
266	.get("code_verifier")
267	.map(String::as_str)
268	.unwrap_or_default();
269	if code.is_null()
270	|| code["client"] != client["client_id"]
271	|| input.get("redirect_uri").map(String::as_str) != code["redirect"].as_str()
272	|| !(43..=128).contains(&verifier.len())
273	|| !verifier
274	.bytes()
275	.all(|b| b.is_ascii_alphanumeric() || b"-._~".contains(&b))
276	|| URL_SAFE_NO_PAD.encode(Sha256::digest(verifier.as_bytes()))
277	!= string(&code["challenge"])
278	{
279	return Err(fail("invalid_grant"));
280	}
281	if input
282	.get("resource")
283	.is_some_and(|r| code["resource"] != r.as_str())
284	{
285	return Err(fail("invalid_target"));
286	}
287	let grant = get(db, &format!("grant:{}", string(&code["grant"])))?;
288	if grant.is_null() {
289	return Err(fail("invalid_grant"));
290	}
291	(key, grant)
292	}
293	Some("refresh_token") => {
294	let fingerprint = hash(
295	input
296	.get("refresh_token")
297	.map(String::as_str)
298	.unwrap_or_default(),
299	);
300	let key = format!("refresh:{fingerprint}");
301	let token = get(db, &key)?;
302	let used = get(db, &format!("used:{fingerprint}"))?;
303	if !used.is_null() {
304	let grant = get(db, &format!("grant:{}", string(&used["grant"])))?;
305	if grant["client"] == client["client_id"] {
306	revoke(db, string(&used["grant"]))?;
307	}
308	return Err(fail("invalid_grant"));
309	}
310	let grant = get(db, &format!("grant:{}", string(&token["grant"])))?;
311	if grant.is_null() || grant["client"] != client["client_id"] {
312	return Err(fail("invalid_grant"));
313	}
314	if input
315	.get("resource")
316	.is_some_and(|r| grant["resource"] != r.as_str())
317	{
318	return Err(fail("invalid_target"));
319	}
320	if input.get("scope").is_some_and(|s| {
321	s.split_whitespace()
322	.collect::<std::collections::HashSet<_>>()
323	!= array(&grant["scopes"]).iter().map(string).collect()
324	}) {
325	return Err(fail("invalid_scope"));
326	}
327	(key, grant)
328	}
329	_ => return Err(fail("unsupported_grant_type")),
330	};
331	Ok((key, grant))
332	}
333	fn oauth(
334	&self,
335	path: &str,
336	method: &Method,
337	input: &HashMap<String, String>,
338	headers: &HeaderMap,
339	body: Value,
340	) -> Result<Response> {
341	let mut db = self.db.lock().unwrap();
342	let tx = db.transaction()?;
343	let value = match (path, method) {
344	("register", &Method::POST) => {
345	let name = body["client_name"]
346	.as_str()
347	.filter(|s| !s.is_empty() && s.len() <= 128)
348	.ok_or_else(|| fail("invalid_client_metadata"))?;
349	let uris = body["redirect_uris"]
350	.as_array()
351	.filter(|a| !a.is_empty() && a.len() <= 8)
352	.ok_or_else(|| fail("invalid_client_metadata"))?;
353	for uri in uris {
354	redirect(uri.as_str().ok_or_else(|| fail("invalid_redirect_uri"))?)?;
355	}
356	let auth = body["token_endpoint_auth_method"]
357	.as_str()
358	.unwrap_or("none");
359	if !["none", "client_secret_post", "client_secret_basic"].contains(&auth)
360	|| body.get("grant_types").is_some_and(|v| {
361	v.as_array().is_none_or(|types| {
362	types.is_empty()
363	|| types
364	.iter()
365	.any(|s| s != "authorization_code" && s != "refresh_token")
366	})
367	})
368	|| body
369	.get("response_types")
370	.is_some_and(|v| v != &json!(["code"]))
371	{
372	return Err(fail("invalid_client_metadata"));
373	}
374	if list(&tx, "client:")?.len() >= 4096 {
375	return Err(Error::new(429, "too_many_clients"));
376	}
377	let id = uuid::Uuid::new_v4().to_string();
378	let credential = secret();
379	let mut client = json!({"client_id":id,"client_name":name,"redirect_uris":uris,"token_endpoint_auth_method":auth,"grant_types":["authorization_code","refresh_token"],"response_types":["code"],"client_id_issued_at":now() as i64});
380	if auth != "none" {
381	client["secret_hash"] = json!(hash(&credential));
382	}
383	put(&tx, &format!("client:{id}"), &client, 0)?;
384	client.as_object_mut().unwrap().remove("secret_hash");
385	if auth != "none" {
386	client["client_secret"] = json!(credential);
387	client["client_secret_expires_at"] = json!(0);
388	}
389	tx.commit()?;
390	return Ok((StatusCode::CREATED, axum::Json(client)).into_response());
391	}
392	("authorize", &Method::GET) => {
393	let id = input
394	.get("client_id")
395	.map(String::as_str)
396	.unwrap_or_default();
397	let client = get(&tx, &format!("client:{id}"))?;
398	let uri = input
399	.get("redirect_uri")
400	.map(String::as_str)
401	.unwrap_or_default();
402	let challenge = input
403	.get("code_challenge")
404	.map(String::as_str)
405	.unwrap_or_default();
406	let resource = input
407	.get("resource")
408	.map(String::as_str)
409	.unwrap_or_default();
410	let scopes_allowed = CATALOGS
411	.iter()
412	.find(|(id, _, _)| resource == self.resource(id))
413	.map(|(_, _, scopes)| *scopes)
414	.ok_or_else(|| fail("invalid_target"))?;
415	let scopes: Vec<_> = input
416	.get("scope")
417	.map(String::as_str)
418	.unwrap_or(scopes_allowed[0])
419	.split_whitespace()
420	.collect();
421	if client.is_null() || !array(&client["redirect_uris"]).iter().any(|v| v == uri) {
422	return Err(fail("invalid_redirect_uri"));
423	}
424	if input.get("response_type").map(String::as_str) != Some("code")
425	|| input.get("code_challenge_method").map(String::as_str) != Some("S256")
426	|| challenge.len() != 43
427	|| !challenge
428	.bytes()
429	.all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_')
430	{
431	return Err(fail("invalid_request"));
432	}
433	if !scopes.contains(&scopes_allowed[0])
434	|| scopes.iter().any(|scope| !CATALOGS.iter().any(|(_, _, allowed)| allowed.contains(scope)))
435	{
436	return Err(fail("invalid_scope"));
437	}
438	let scopes: Vec<_> = scopes.into_iter().filter(|scope| scopes_allowed.contains(scope)).collect();
439	let pending = secret();
440	put(
441	&tx,
442	&format!("pending:{}", hash(&pending)),
443	&json!({"client":id,"redirect":uri,"challenge":challenge,"resource":resource,"scopes":scopes,"state":input.get("state"),"owner":null}),
444	600,
445	)?;
446	tx.commit()?;
447	return Ok((
448	StatusCode::FOUND,
449	[("location", format!("/connect/{}", encoded(&pending)))],
450	)
451	.into_response());
452	}
453	("token", &Method::POST) => {
454	let (key, grant) = match self.exchange(&tx, input, headers) {
455	Ok(exchange) => exchange,
456	Err(error) => {
457	tx.commit()?;
458	return Err(error);
459	}
460	};
461	delete(&tx, &key)?;
462	if let Some(fingerprint) = key.strip_prefix("refresh:") {
463	put(
464	&tx,
465	&format!("used:{fingerprint}"),
466	&json!({"grant":grant["id"]}),
467	30 * 86400,
468	)?;
469	}
470	self.issue(&tx, &grant)?
471	}
472	("revoke", &Method::POST) => {
473	let client = self.client(&tx, input, headers)?;
474	let fingerprint = hash(input.get("token").map(String::as_str).unwrap_or_default());
475	for kind in ["access", "refresh", "used"] {
476	let token = get(&tx, &format!("{kind}:{fingerprint}"))?;
477	let grant = get(&tx, &format!("grant:{}", string(&token["grant"])))?;
478	if !grant.is_null() && grant["client"] == client["client_id"] {
479	revoke(&tx, string(&grant["id"]))?;
480	}
481	}
482	tx.commit()?;
483	return Ok(StatusCode::OK.into_response());
484	}
485	_ => return Err(Error::new(404, "No endpoint here.")),
486	};
487	tx.commit()?;
488	Ok(axum::Json(value).into_response())
489	}
490	}
491
492	#[derive(Clone)]
493	pub(crate) struct Grant(pub Value);
494	pub(crate) fn active_owner(app: &App, grant: &Value) -> Result<bool> {
495	let profile = match auth::user(&app.auth.db.lock().unwrap(), string(&grant["user"])) {
496	Ok(profile) => profile,
497	Err(error) if error.status == 404 => return Ok(false),
498	Err(error) => return Err(error),
499	};
500	Ok(profile["enabled"] == true
501	&& !guest::is_guest(&profile)
502	&& (grant["resource"] != app.mcp.resource("observability")
503	|| array(&profile["groups"])
504	.iter()
505	.any(|role| role["name"] == "infra-admin")))
506	}
507	pub fn router<H: rmcp::ServerHandler>(
508	app: Arc<App>,
509	catalog: &str,
510	handler: impl Fn() -> std::result::Result<H, std::io::Error> + Send + Sync + 'static,
511	) -> Router {
512	use rmcp::transport::streamable_http_server::{
513	StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager,
514	};
515	let resource = app.mcp.resource(catalog);
516	let metadata = format!(
517	"{}.well-known/oauth-protected-resource/mcp/{catalog}",
518	app.mcp.origin.as_str()
519	);
520	let mut config = StreamableHttpServerConfig::default();
521	config.legacy_session_mode = false;
522	config.json_response = true;
523	config.allowed_hosts =
524	vec![app.mcp.origin[url::Position::BeforeHost..url::Position::AfterPort].to_owned()];
525	config.allowed_origins = vec![format!(
526	"{}://{}:{}",
527	app.mcp.origin.scheme(),
528	app.mcp.origin.host_str().unwrap(),
529	app.mcp.origin.port_or_known_default().unwrap()
530	)];
531	let service =
532	StreamableHttpService::new(handler, Arc::new(LocalSessionManager::default()), config);
533	Router::new()
534	.nest_service(&format!("/mcp/{catalog}"), service)
535	.layer(axum::middleware::from_fn(
536	move |mut request: Request, next: axum::middleware::Next| {
537	let (app, resource, metadata) = (app.clone(), resource.clone(), metadata.clone());
538	async move {
539	if request.headers().get("origin").is_some_and(|origin| {
540	origin.to_str().ok()
541	!= Some(app.mcp.origin.origin().ascii_serialization().as_str())
542	}) {
543	return Error::new(403, "This origin cannot use the connector.")
544	.into_response();
545	}
546	match app
547	.mcp
548	.authenticate(request.headers(), &resource)
549	.and_then(|grant| {
550	if active_owner(&app, &grant)? {
551	Ok(grant)
552	} else {
553	Err(Error::new(401, "invalid_token"))
554	}
555	}) {
556	Ok(grant) => {
557	request.extensions_mut().insert(Grant(grant));
558	if let Some(value) = request.headers_mut().get_mut("authorization") {
559	value.set_sensitive(true);
560	}
561	next.run(request).await
562	}
563	Err(_) => (
564	StatusCode::UNAUTHORIZED,
565	[(
566	"www-authenticate",
567	format!("Bearer resource_metadata=\"{metadata}\""),
568	)],
569	"This connection expired. Connect again.",
570	)
571	.into_response(),
572	}
573	}
574	},
575	))
576	}
577
578	pub fn public(path: &str) -> bool {
579	path == "/.well-known/openid-configuration"
580	|| path.starts_with("/oauth/")
581	|| CATALOGS.iter().any(|(id, _, _)| {
582	path == format!("/mcp/{id}") || path.starts_with(&format!("/mcp/{id}/"))
583	})
584	|| path.starts_with("/.well-known/oauth-")
585	|| path == "/pairing"
586	|| path == "/agent/connect"
587	|| matches!(
588	path,
589	"/agent/install.sh" | "/agent/install.ps1" | "/agent/setup.mjs" | "/agent/relay.mjs"
590	)
591	|| path.starts_with("/api/v1/")
592	}
593	pub async fn oauth(State(app): State<Arc<App>>, request: Request) -> Response {
594	if request.uri().path().starts_with("/oauth/shale/") {
595	return shale::oauth(app, request).await;
596	}
597	let result = async {
598	let path = request.uri().path().to_owned();
599	if path=="/.well-known/oauth-authorization-server" && request.method()==Method::GET {
600	let base = app.mcp.origin.as_str();
601	let scopes: std::collections::BTreeSet<_> = CATALOGS.iter().flat_map(|(_, _, scopes)| scopes.iter()).collect();
602	return Ok(axum::Json(json!({"issuer":base,"authorization_endpoint":format!("{base}oauth/authorize"),"token_endpoint":format!("{base}oauth/token"),"registration_endpoint":format!("{base}oauth/register"),"revocation_endpoint":format!("{base}oauth/revoke"),"response_types_supported":["code"],"grant_types_supported":["authorization_code","refresh_token"],"token_endpoint_auth_methods_supported":["none","client_secret_post","client_secret_basic"],"code_challenge_methods_supported":["S256"],"scopes_supported":scopes})).into_response());
603	}
604	if request.method() == Method::GET {
605	for (catalog, _, scopes) in CATALOGS {
606	if path == format!("/.well-known/oauth-protected-resource/mcp/{catalog}") {
607	return Ok(axum::Json(json!({"resource":app.mcp.resource(catalog),"authorization_servers":[app.mcp.origin.as_str()],"scopes_supported":scopes,"bearer_methods_supported":["header"]})).into_response());
608	}
609	}
610	}
611	let method = request.method().clone();
612	let headers = request.headers().clone();
613	let query = request.uri().query().unwrap_or_default().to_owned();
614	let bytes = axum::body::to_bytes(request.into_body(),65536).await.map_err(|_| fail("invalid_request"))?;
615	let registration = path=="/oauth/register";
616	let encoded = if method==Method::GET {query.as_bytes()} else {&bytes};
617	if method==Method::POST && !headers.get("content-type").and_then(|v|v.to_str().ok()).is_some_and(|s| s.split(';').next()==Some(if registration {"application/json"} else {"application/x-www-form-urlencoded"})) { return Err(fail("invalid_request")); }
618	let mut input = HashMap::new();
619	if !registration { for (key,value) in url::form_urlencoded::parse(encoded) { if input.insert(key.into_owned(),value.into_owned()).is_some() {return Err(fail("invalid_request"));} } }
620	let body = if registration {serde_json::from_slice(&bytes).map_err(|_| fail("invalid_client_metadata"))?} else {Value::Null};
621	if path == "/oauth/token" && method == Method::POST {
622	let (_, grant) = app.mcp.exchange(&app.mcp.db.lock().unwrap(), &input, &headers)?;
623	if !active_owner(&app, &grant)? {
624	revoke(&app.mcp.db.lock().unwrap(), string(&grant["id"]))?;
625	return Err(fail("invalid_grant"));
626	}
627	}
628	app.mcp.oauth(path.trim_start_matches("/oauth/"),&method,&input,&headers,body)
629	}.await;
630	let mut response = match result {
631	Ok(response) => response,
632	Err(error) => (
633	StatusCode::from_u16(if error.message == "invalid_client" {
634	401
635	} else {
636	error.status
637	})
638	.unwrap_or(StatusCode::BAD_REQUEST),
639	axum::Json(
640	json!({"error":if error.status >= 500 {"server_error"} else {&error.message}}),
641	),
642	)
643	.into_response(),
644	};
645	response
646	.headers_mut()
647	.insert("cache-control", "no-store".parse().unwrap());
648	response
649	.headers_mut()
650	.insert("pragma", "no-cache".parse().unwrap());
651	response
652	}
653
654	fn chosen_resources(body: &Value, resources: &[Value], shale: bool) -> Result<Value> {
655	if shale && body["resources"] == "all" {
656	return Ok(json!("all"));
657	}
658	let chosen = body["resources"]
659	.as_array()
660	.filter(|items| !items.is_empty() && items.len() <= resources.len())
661	.ok_or_else(|| Error::new(400, "Choose each available resource once."))?;
662	if chosen.iter().enumerate().any(|(index, item)| {
663	!resources.iter().any(|resource| item == &resource["id"]) || chosen[..index].contains(item)
664	}) {
665	return Err(Error::new(
666	403,
667	"Choose resources available to your account.",
668	));
669	}
670	Ok(json!(chosen))
671	}
672
673	pub async fn manage(
674	app: Arc<App>,
675	method: &Method,
676	parts: &[&str],
677	me: &Value,
678	body: Value,
679	headers: &HeaderMap,
680	) -> Result<Response> {
681	if method != Method::GET
682	&& (me["viewing"] == true
683	|| headers.get("origin").and_then(|h| h.to_str().ok())
684	!= Some(app.mcp.origin.origin().ascii_serialization().as_str()))
685	{
686	return Err(Error::new(
687	403,
688	"Open MCP settings from your own signed-in account.",
689	));
690	}
691	let owner = users::self_user(&app, me).await?;
692	if owner["enabled"] != true {
693	return Err(Error::new(403, "This account is disabled."));
694	}
695	let owner_id = string(&owner["id"]);
696	if parts == ["shale"] {
697	return shale::manage(app, method, &owner, &body).await;
698	}
699	if parts == ["relay", "live"] && method == Method::GET {
700	let owner = owner_id.to_owned();
701	let stream = WatchStream::new(app.relay.changes.subscribe()).map(move |_| {
702	let machines = relay::machines(&app.mcp.db.lock().unwrap(), &owner)
703	.map(|machines| app.relay.view(machines, None));
704	machines
705	.map(|machines| Event::default().json_data(machines).unwrap())
706	.map_err(|error| std::io::Error::other(error.message))
707	});
708	return Ok(Sse::new(stream)
709	.keep_alive(axum::response::sse::KeepAlive::default())
710	.into_response());
711	}
712	let consent_resource = if let ["connections", id] = parts
713	&& method != Method::DELETE
714	{
715	let grant = get(&app.mcp.db.lock().unwrap(), &format!("grant:{id}"))?;
716	if grant["user"] != owner_id {
717	return Err(Error::new(404, "No connection with that ID."));
718	}
719	string(&grant["resource"]).to_owned()
720	} else if let ["consent", id] = parts {
721	let pending = get(
722	&app.mcp.db.lock().unwrap(),
723	&format!("pending:{}", hash(id)),
724	)?;
725	if pending.is_null() {
726	return Err(Error::new(
727	404,
728	"This connection request expired. Start it again.",
729	));
730	}
731	if !pending["owner"].is_null() && pending["owner"] != owner_id {
732	return Err(Error::new(
733	403,
734	"This connection request belongs to another account.",
735	));
736	}
737	string(&pending["resource"]).to_owned()
738	} else {
739	String::new()
740	};
741	let agent_consent = consent_resource == app.mcp.resource("agents");
742	let shale_consent = consent_resource == app.mcp.resource("shale");
743	let catalog = CATALOGS
744	.iter()
745	.find(|(id, _, _)| consent_resource == app.mcp.resource(id))
746	.map(|(id, _, _)| *id);
747	let mut linked = true;
748	let mut resource_error = None;
749	let resources: Vec<Value> = if body["deny"] == true {
750	Vec::new()
751	} else if agent_consent {
752	relay::machines(&app.mcp.db.lock().unwrap(), owner_id)?
753	.into_iter()
754	.map(|m| json!({"id":m["id"],"name":m["name"]}))
755	.collect()
756	} else if shale_consent {
757	let available = if body["resources"] == "all" {
758	shale::verified_session(&app, owner_id)
759	.await
760	.map(|_| Vec::new())
761	} else {
762	shale::repositories(&app, owner_id).await
763	};
764	match available {
765	Ok(repositories) => repositories,
766	Err(error) if error.status == 401 => {
767	linked = false;
768	Vec::new()
769	}
770	Err(error) if method == Method::GET => {
771	resource_error = Some(error.message);
772	Vec::new()
773	}
774	Err(error) => return Err(error),
775	}
776	} else if consent_resource == app.mcp.resource("observability")
777	&& array(&me["sections"]).iter().any(|s| s == "admin")
778	&& array(&owner["groups"])
779	.iter()
780	.any(|g| g["name"] == "infra-admin")
781	{
782	core::scan(app.clone())
783	.await?
784	.value
785	.as_object()
786	.unwrap()
787	.keys()
788	.map(|id| json!({"id":id,"name":id}))
789	.collect()
790	} else {
791	Vec::new()
792	};
793	let mut db = app.mcp.db.lock().unwrap();
794	let tx = db.transaction()?;
795	let value = match parts {
796	[] if method == Method::GET => {
797	let mut grants = list(&tx, "grant:")?;
798	grants.retain(|g| g["user"] == owner_id);
799	let machines = relay::machines(&tx, owner_id)?;
800	let connections = grants.iter().map(|grant| {
801	let name = if grant["client"].is_null() {grant["name"].clone()} else {get(&tx, &format!("client:{}", string(&grant["client"])))?["client_name"].clone()};
802	let resources = if grant["resource"] == app.mcp.resource("shale") && grant["resources"] == "all" {json!("all")} else {json!(array(&grant["resources"]).iter().map(|id| if grant["resource"] == app.mcp.resource("agents") {machines.iter().find(|m| m["id"] == *id).map(|m| m["name"].clone()).unwrap_or_else(|| json!("Unlinked machine"))} else {id.clone()}).collect::<Vec<_>>())};
803	let catalog = CATALOGS.iter().find(|(id, _, _)| grant["resource"] == app.mcp.resource(id)).map(|(id, _, _)| *id);
804	Ok(json!({"id":grant["id"],"name":name,"catalog":catalog,"resources":resources,"scopes":grant["scopes"],"createdAt":grant["createdAt"]}))
805	}).collect::<Result<Vec<_>>>()?;
806	let shale = get(&tx, &format!("shale-session:{owner_id}"))?;
807	let catalogs: Vec<_> = CATALOGS
808	.iter()
809	.map(|(id, name, _)| json!({"id":id,"name":name,"endpoint":app.mcp.resource(id)}))
810	.collect();
811	json!({"catalogs":catalogs,"connections":connections,"machines":app.relay.view(machines,None),"shale":if shale["origin"] != app.shale.origin.as_str() {Value::Null} else {json!({"linkedAt":shale["linkedAt"]})}})
812	}
813	["consent", id] if method == Method::GET || method == Method::POST => {
814	let key = format!("pending:{}", hash(id));
815	let mut pending = get(&tx, &key)?;
816	if pending.is_null() {
817	return Err(Error::new(
818	404,
819	"This connection request expired. Start it again.",
820	));
821	}
822	if !pending["owner"].is_null() && pending["owner"] != owner_id {
823	return Err(Error::new(
824	403,
825	"This connection request belongs to another account.",
826	));
827	}
828	pending["owner"] = json!(owner_id);
829	if method == Method::POST && body["deny"] == true {
830	delete(&tx, &key)?;
831	let mut target = redirect(string(&pending["redirect"]))?;
832	target
833	.query_pairs_mut()
834	.append_pair("error", "access_denied")
835	.append_pair("iss", app.mcp.origin.as_str());
836	if let Some(state) = pending["state"].as_str() {
837	target.query_pairs_mut().append_pair("state", state);
838	}
839	tx.commit()?;
840	return Ok(axum::Json(json!({"redirect":target.as_str()})).into_response());
841	}
842	if method == Method::GET {
843	tx.execute(
844	"UPDATE records SET value=? WHERE key=?",
845	rusqlite::params![pending.to_string(), key],
846	)?;
847	json!({"client":get(&tx,&format!("client:{}",string(&pending["client"])))?["client_name"],"catalog":catalog,"account":owner["username"],"redirectHost":redirect(string(&pending["redirect"]))?.host_str(),"scopes":pending["scopes"],"resources":resources,"linked":linked,"resourceError":resource_error})
848	} else {
849	if shale_consent
850	&& (!linked
851	|| get(&tx, &format!("shale-session:{owner_id}"))?["origin"]
852	!= app.shale.origin.as_str())
853	{
854	return Err(Error::new(
855	401,
856	"Link your Shale account before allowing repository access.",
857	));
858	}
859	let chosen = chosen_resources(&body, &resources, shale_consent)?;
860	if list(&tx, "grant:")?
861	.iter()
862	.filter(|g| g["user"] == owner_id)
863	.count()
864	>= 256
865	{
866	return Err(Error::new(
867	409,
868	"Remove an unused connection before adding another.",
869	));
870	}
871	let grant_id = uuid::Uuid::new_v4().to_string();
872	let mut grant = json!({"id":grant_id,"user":owner_id,"client":pending["client"],"resource":pending["resource"],"scopes":pending["scopes"],"resources":chosen,"createdAt":now()});
873	if agent_consent {
874	grant["targets"] = json!(chosen);
875	}
876	put(&tx, &format!("grant:{grant_id}"), &grant, 0)?;
877	let code = secret();
878	pending["grant"] = json!(grant_id);
879	put(&tx, &format!("code:{}", hash(&code)), &pending, 300)?;
880	delete(&tx, &key)?;
881	let mut target = redirect(string(&pending["redirect"]))?;
882	target
883	.query_pairs_mut()
884	.append_pair("code", &code)
885	.append_pair("iss", app.mcp.origin.as_str());
886	if let Some(state) = pending["state"].as_str() {
887	target.query_pairs_mut().append_pair("state", state);
888	}
889	json!({"redirect":target.as_str()})
890	}
891	}
892	["relay", rest @ ..] => relay::manage(&app, &tx, rest, method, owner_id, &body)?,
893	["connections", id]
894	if method == Method::GET || method == Method::POST || method == Method::DELETE =>
895	{
896	let key = format!("grant:{id}");
897	let mut grant = get(&tx, &key)?;
898	if grant["user"] != owner_id {
899	return Err(Error::new(404, "No connection with that ID."));
900	}
901	if method == Method::DELETE {
902	revoke(&tx, id)?;
903	Value::Null
904	} else if method == Method::GET {
905	json!({"resources": resources, "selected": grant["resources"], "linked": linked,"resourceError":resource_error})
906	} else {
907	if shale_consent && !linked {
908	return Err(Error::new(
909	401,
910	"Link your Shale account before allowing repository access.",
911	));
912	}
913	let chosen = chosen_resources(&body, &resources, shale_consent)?;
914	grant["resources"] = json!(chosen);
915	if agent_consent {
916	grant["targets"] = json!(chosen);
917	}
918	tx.execute(
919	"UPDATE records SET value=? WHERE key=?",
920	rusqlite::params![grant.to_string(), key],
921	)?;
922	Value::Null
923	}
924	}
925	_ => return Err(Error::new(404, "No endpoint here.")),
926	};
927	tx.commit()?;
928	if matches!(parts, ["relay", "pair"] | ["relay", "machines", _]) {
929	app.relay.changes.send_replace(());
930	}
931	Ok(if value.is_null() {
932	StatusCode::NO_CONTENT.into_response()
933	} else {
934	axum::Json(value).into_response()
935	})
936	}
937
938	#[cfg(test)]
939	mod tests {
940	#[test]
941	fn resource_updates_reject_empty_duplicates_and_foreign_choices() {
942	let resources = vec![json!({"id":"alpha"}), json!({"id":"beta"})];
943	assert_eq!(
944	chosen_resources(&json!({"resources":["beta"]}), &resources, false).unwrap(),
945	json!(["beta"])
946	);
947	for selected in [
948	json!([]),
949	json!(["alpha", "alpha"]),
950	json!(["foreign"]),
951	json!([null]),
952	] {
953	assert!(chosen_resources(&json!({"resources":selected}), &resources, false).is_err());
954	}
955	assert_eq!(
956	chosen_resources(&json!({"resources":"all"}), &[], true).unwrap(),
957	json!("all")
958	);
959	assert!(chosen_resources(&json!({"resources":"all"}), &resources, false).is_err());
960	}
961
962	use super::*;
963	struct Fixture {
964	store: Arc<Store>,
965	path: std::path::PathBuf,
966	}
967	impl Fixture {
968	fn new() -> Self {
969	let path = std::env::temp_dir()
970	.canonicalize()
971	.unwrap()
972	.join(format!("studio-mcp-test-{}", uuid::Uuid::new_v4()));
973	let store = Arc::new(Store::new(&path, "https://globe.studio.test").unwrap());
974	Self { store, path }
975	}
976	fn code(&self, client: &str, code: &str) -> HashMap<String, String> {
977	let grant = uuid::Uuid::new_v4().to_string();
978	let db = self.store.db.lock().unwrap();
979	put(&db, &format!("client:{client}"), &json!({"client_id":client,"token_endpoint_auth_method":"none","redirect_uris":["http://127.0.0.1:20001/callback"]}),0).unwrap();
980	put(&db, &format!("grant:{grant}"), &json!({"id":grant,"user":"one","client":client,"resource":self.store.resource("observability"),"scopes":["observability:read","offline_access"],"resources":["allowed"]}),0).unwrap();
981	let verifier = "v".repeat(43);
982	put(&db, &format!("code:{}",hash(code)), &json!({"client":client,"redirect":"http://127.0.0.1:20001/callback","challenge":URL_SAFE_NO_PAD.encode(Sha256::digest(verifier.as_bytes())),"resource":self.store.resource("observability"),"grant":grant}),300).unwrap();
983	fields(
984	json!({"grant_type":"authorization_code","client_id":client,"code":code,"code_verifier":verifier,"redirect_uri":"http://127.0.0.1:20001/callback","resource":self.store.resource("observability")}),
985	)
986	}
987	}
988	impl Drop for Fixture {
989	fn drop(&mut self) {
990	std::fs::remove_dir_all(&self.path).unwrap();
991	}
992	}
993	fn fields(value: Value) -> HashMap<String, String> {
994	value
995	.as_object()
996	.unwrap()
997	.iter()
998	.map(|(k, v)| (k.clone(), string(v).to_owned()))
999	.collect()
1000	}
1001	async fn value(response: Response) -> Value {
1002	serde_json::from_slice(
1003	&axum::body::to_bytes(response.into_body(), 65536)
1004	.await
1005	.unwrap(),
1006	)
1007	.unwrap()
1008	}
1009	fn bearer(token: &str) -> HeaderMap {
1010	let mut headers = HeaderMap::new();
1011	headers.insert("authorization", format!("Bearer {token}").parse().unwrap());
1012	headers
1013	}
1014	#[test]
1015	fn revocation_removes_durable_keys_and_family_records() {
1016	let fixture = Fixture::new();
1017	let db = fixture.store.db.lock().unwrap();
1018	put(&db, "grant:revoked", &json!({"id":"revoked"}), 0).unwrap();
1019	for kind in ["access", "refresh", "used", "code"] {
1020	put(&db, &format!("{kind}:one"), &json!({"grant":"revoked"}), 0).unwrap();
1021	put(
1022	&db,
1023	&format!("{kind}:other"),
1024	&json!({"grant":"retained"}),
1025	0,
1026	)
1027	.unwrap();
1028	}
1029	revoke(&db, "revoked").unwrap();
1030	assert!(get(&db, "grant:revoked").unwrap().is_null());
1031	for kind in ["access", "refresh", "used", "code"] {
1032	assert!(get(&db, &format!("{kind}:one")).unwrap().is_null());
1033	assert!(!get(&db, &format!("{kind}:other")).unwrap().is_null());
1034	}
1035	}
1036	#[tokio::test]
1037	async fn authorization_restricts_issuer_scopes_to_the_requested_resource() {
1038	let fixture = Fixture::new();
1039	let uri = "http://127.0.0.1:57072/callback/H9rAWI-8eQbv";
1040	let client = value(fixture.store.oauth("register", &Method::POST, &HashMap::new(), &HeaderMap::new(),
1041	json!({"client_name":"Codex","redirect_uris":[uri]})).unwrap()).await;
1042	let mut request = fields(json!({"response_type":"code","client_id":client["client_id"],"state":"fixture-state",
1043	"code_challenge":"AdJpfUsRa6XMg54f2khmgo2onMwePN3vmU9DOOoaTzo","code_challenge_method":"S256",
1044	"redirect_uri":uri,"scope":"observability:read offline_access sessions:read sessions:write shale:read shale:write"}));
1045	for (catalog, _, allowed) in CATALOGS {
1046	request.insert("resource".into(), fixture.store.resource(catalog));
1047	let response = fixture.store.oauth("authorize", &Method::GET, &request, &HeaderMap::new(), Value::Null).unwrap();
1048	assert_eq!(response.status(), StatusCode::FOUND);
1049	let pending_id = response.headers()["location"].to_str().unwrap().strip_prefix("/connect/").unwrap();
1050	let db = fixture.store.db.lock().unwrap();
1051	let mut pending = get(&db, &format!("pending:{}", hash(pending_id))).unwrap();
1052	let expected: Vec<_> = request["scope"].split_whitespace().filter(|scope| allowed.contains(scope)).collect();
1053	assert_eq!(pending["scopes"], json!(expected));
1054	assert_eq!(pending["redirect"], uri);
1055	assert_eq!(pending["state"], "fixture-state");
1056	pending["id"] = json!(format!("fixture-{catalog}"));
1057	put(&db, &format!("grant:fixture-{catalog}"), &pending, 0).unwrap();
1058	let tokens = fixture.store.issue(&db, &pending).unwrap();
1059	assert_eq!(tokens["scope"], expected.join(" "));
1060	assert!(tokens["refresh_token"].is_string());
1061	drop(db);
1062	let headers = bearer(string(&tokens["access_token"]));
1063	assert!(fixture.store.authenticate(&headers, &fixture.store.resource(catalog)).is_ok());
1064	for (other, _, _) in CATALOGS.iter().filter(|(other, _, _)| other != catalog) {
1065	assert!(fixture.store.authenticate(&headers, &fixture.store.resource(other)).is_err());
1066	}
1067	}
1068	for scopes in ["shale:read unknown:scope", "observability:read offline_access", "offline_access", ""] {
1069	request.insert("resource".into(), fixture.store.resource("shale"));
1070	request.insert("scope".into(), scopes.into());
1071	assert_eq!(fixture.store.oauth("authorize", &Method::GET, &request, &HeaderMap::new(), Value::Null).unwrap_err().message, "invalid_scope");
1072	}
1073	request.insert("scope".into(), "shale:read".into());
1074	request.insert("redirect_uri".into(), uri.replace("H9rAWI-8eQbv", "different"));
1075	assert_eq!(fixture.store.oauth("authorize", &Method::GET, &request, &HeaderMap::new(), Value::Null).unwrap_err().message, "invalid_redirect_uri");
1076	}
1077	#[tokio::test]
1078	async fn code_binds_pkce_client_redirect_and_audience_without_consuming_on_failure() {
1079	let fixture = Fixture::new();
1080	let input = fixture.code("one", "owned-code");
1081	fixture.code("two", "other-code");
1082	for (key, wrong) in [
1083	("client_id", "two"),
1084	("code_verifier", "wrong"),
1085	("redirect_uri", "http://127.0.0.1:20002/callback"),
1086	("resource", "https://other.invalid/mcp/observability"),
1087	] {
1088	let mut attempt = input.clone();
1089	attempt.insert(key.into(), wrong.into());
1090	assert!(
1091	fixture
1092	.store
1093	.oauth(
1094	"token",
1095	&Method::POST,
1096	&attempt,
1097	&HeaderMap::new(),
1098	Value::Null
1099	)
1100	.is_err()
1101	);
1102	}
1103	let tokens = value(
1104	fixture
1105	.store
1106	.oauth(
1107	"token",
1108	&Method::POST,
1109	&input,
1110	&HeaderMap::new(),
1111	Value::Null,
1112	)
1113	.unwrap(),
1114	)
1115	.await;
1116	assert!(
1117	fixture
1118	.store
1119	.oauth(
1120	"token",
1121	&Method::POST,
1122	&input,
1123	&HeaderMap::new(),
1124	Value::Null
1125	)
1126	.is_err()
1127	);
1128	assert!(
1129	fixture
1130	.store
1131	.authenticate(
1132	&bearer(string(&tokens["access_token"])),
1133	&fixture.store.resource("observability")
1134	)
1135	.is_ok()
1136	);
1137	assert!(
1138	fixture
1139	.store
1140	.authenticate(
1141	&bearer(string(&tokens["access_token"])),
1142	"https://other.invalid/mcp/observability"
1143	)
1144	.is_err()
1145	);
1146	assert_eq!(
1147	std::fs::metadata(fixture.path.join("connections.sqlite"))
1148	.unwrap()
1149	.permissions()
1150	.mode()
1151	& 0o777,
1152	0o600
1153	);
1154	let rows = fixture
1155	.store
1156	.db
1157	.lock()
1158	.unwrap()
1159	.prepare("SELECT key,value FROM records")
1160	.unwrap()
1161	.query_map([], |row| {
1162	Ok(format!(
1163	"{} {}",
1164	row.get::<_, String>(0)?,
1165	row.get::<_, String>(1)?
1166	))
1167	})
1168	.unwrap()
1169	.collect::<std::result::Result<Vec<_>, _>>()
1170	.unwrap()
1171	.join("\n");
1172	for token in [
1173	"owned-code",
1174	string(&tokens["access_token"]),
1175	string(&tokens["refresh_token"]),
1176	] {
1177	assert!(!rows.contains(token));
1178	}
1179	}
1180	#[tokio::test]
1181	async fn refresh_replay_revokes_family_but_another_client_cannot_revoke_it() {
1182	let fixture = Fixture::new();
1183	let input = fixture.code("one", "owned-code");
1184	fixture.code("two", "other-code");
1185	let tokens = value(
1186	fixture
1187	.store
1188	.oauth(
1189	"token",
1190	&Method::POST,
1191	&input,
1192	&HeaderMap::new(),
1193	Value::Null,
1194	)
1195	.unwrap(),
1196	)
1197	.await;
1198	let refresh = fields(
1199	json!({"grant_type":"refresh_token","client_id":"one","refresh_token":tokens["refresh_token"]}),
1200	);
1201	let mut wrong = refresh.clone();
1202	wrong.insert("client_id".into(), "two".into());
1203	assert!(
1204	fixture
1205	.store
1206	.oauth(
1207	"token",
1208	&Method::POST,
1209	&wrong,
1210	&HeaderMap::new(),
1211	Value::Null
1212	)
1213	.is_err()
1214	);
1215	fixture
1216	.store
1217	.oauth(
1218	"revoke",
1219	&Method::POST,
1220	&fields(json!({"client_id":"two","token":tokens["access_token"]})),
1221	&HeaderMap::new(),
1222	Value::Null,
1223	)
1224	.unwrap();
1225	let access = bearer(string(&tokens["access_token"]));
1226	assert!(
1227	fixture
1228	.store
1229	.authenticate(&access, &fixture.store.resource("observability"))
1230	.is_ok()
1231	);
1232	let rotated = value(
1233	fixture
1234	.store
1235	.oauth(
1236	"token",
1237	&Method::POST,
1238	&refresh,
1239	&HeaderMap::new(),
1240	Value::Null,
1241	)
1242	.unwrap(),
1243	)
1244	.await;
1245	assert_ne!(rotated["refresh_token"], tokens["refresh_token"]);
1246	assert!(
1247	fixture
1248	.store
1249	.oauth(
1250	"token",
1251	&Method::POST,
1252	&wrong,
1253	&HeaderMap::new(),
1254	Value::Null
1255	)
1256	.is_err()
1257	);
1258	assert!(
1259	fixture
1260	.store
1261	.authenticate(&access, &fixture.store.resource("observability"))
1262	.is_ok()
1263	);
1264	assert!(
1265	fixture
1266	.store
1267	.oauth(
1268	"token",
1269	&Method::POST,
1270	&refresh,
1271	&HeaderMap::new(),
1272	Value::Null
1273	)
1274	.is_err()
1275	);
1276	assert!(
1277	fixture
1278	.store
1279	.authenticate(&access, &fixture.store.resource("observability"))
1280	.is_err()
1281	);
1282	assert!(
1283	fixture
1284	.store
1285	.authenticate(
1286	&bearer(string(&rotated["access_token"])),
1287	&fixture.store.resource("observability")
1288	)
1289	.is_err()
1290	);
1291	assert!(fixture.store.oauth("token",&Method::POST,&fields(json!({"grant_type":"refresh_token","client_id":"one","refresh_token":rotated["refresh_token"]})),&HeaderMap::new(),Value::Null).is_err());
1292	}
1293	#[test]
1294	fn concurrent_code_exchange_has_one_winner() {
1295	let fixture = Fixture::new();
1296	let input = fixture.code("one", "concurrent-code");
1297	let start = Arc::new(std::sync::Barrier::new(9));
1298	let workers = (0..8)
1299	.map(|_| {
1300	let store = fixture.store.clone();
1301	let input = input.clone();
1302	let start = start.clone();
1303	std::thread::spawn(move || {
1304	start.wait();
1305	store
1306	.oauth(
1307	"token",
1308	&Method::POST,
1309	&input,
1310	&HeaderMap::new(),
1311	Value::Null,
1312	)
1313	.is_ok()
1314	})
1315	})
1316	.collect::<Vec<_>>();
1317	start.wait();
1318	assert_eq!(
1319	workers
1320	.into_iter()
1321	.map(|w| w.join().unwrap() as u32)
1322	.sum::<u32>(),
1323	1
1324	);
1325	}
1326	#[tokio::test]
1327	async fn confidential_client_secret_is_required_and_hashed() {
1328	let fixture = Fixture::new();
1329	let client=value(fixture.store.oauth("register",&Method::POST,&HashMap::new(),&HeaderMap::new(),json!({"client_name":"Confidential","redirect_uris":["https://client.invalid/callback"],"token_endpoint_auth_method":"client_secret_basic"})).unwrap()).await;
1330	let id = string(&client["client_id"]);
1331	let stored = get(&fixture.store.db.lock().unwrap(), &format!("client:{id}")).unwrap();
1332	assert!(stored.get("client_secret").is_none());
1333	assert_ne!(stored["secret_hash"], client["client_secret"]);
1334	let input = fields(json!({"client_id":id,"token":"unknown"}));
1335	assert!(
1336	fixture
1337	.store
1338	.oauth(
1339	"revoke",
1340	&Method::POST,
1341	&input,
1342	&HeaderMap::new(),
1343	Value::Null
1344	)
1345	.is_err()
1346	);
1347	let mut headers = HeaderMap::new();
1348	headers.insert(
1349	"authorization",
1350	format!(
1351	"Basic {}",
1352	STANDARD.encode(format!("{id}:{}", string(&client["client_secret"])))
1353	)
1354	.parse()
1355	.unwrap(),
1356	);
1357	assert!(
1358	fixture
1359	.store
1360	.oauth("revoke", &Method::POST, &input, &headers, Value::Null)
1361	.is_ok()
1362	);
1363	let mut duplicate = input.clone();
1364	duplicate.insert(
1365	"client_secret".into(),
1366	string(&client["client_secret"]).into(),
1367	);
1368	assert!(
1369	fixture
1370	.store
1371	.oauth("revoke", &Method::POST, &duplicate, &headers, Value::Null)
1372	.is_err()
1373	);
1374	}
1375	}