Skip to content

Commit e350337

Browse files
ygndotggygndotgg
committed
fix(rbac): resolve RBAC deadlocks via sequential locking
Eliminated nested `RwLock` acquisitions by introducing `AuthSnapShot` for read paths and sequential lock-dropping for user mutations. Server remains deadlock-free under `quest-parallel` concurrent loads.
1 parent 3899a1b commit e350337

2 files changed

Lines changed: 241 additions & 200 deletions

File tree

‎src/rbac/map.rs‎

Lines changed: 136 additions & 149 deletions
Original file line numberDiff line numberDiff line change
@@ -193,6 +193,104 @@ pub enum SessionKey {
193193

194194
pub type UserSessionMap = HashMap<String, Vec<(SessionKey, DateTime<Utc>)>>;
195195

196+
pub struct AuthSnapShot {
197+
pub username: String,
198+
pub tenant_id: String,
199+
pub permissions: Vec<Permission>,
200+
}
201+
// Unlike `check_auth`, this evaluates a pre-cloned snapshot instead of directly querying the SESSIONS map.
202+
// It executes completely lock-free, avoiding holding the SESSIONS read lock during the heavy permission matching.
203+
// This breaks the lock-order inversion deadlock by ensuring cross-map reads happen outside the SESSIONS lock scope.
204+
pub fn check_auth_snapshot(
205+
snapshot: AuthSnapShot,
206+
required_action: Action,
207+
context_resource: Option<&str>,
208+
context_user: Option<&str>,
209+
) -> Response {
210+
let AuthSnapShot {
211+
username,
212+
tenant_id,
213+
permissions,
214+
} = snapshot;
215+
216+
let mut perms: HashSet<Permission> = HashSet::from_iter(permissions);
217+
perms.extend(aggregate_group_permissions(&username, &tenant_id));
218+
219+
if perms.iter().any(|user_perm| match *user_perm {
220+
Permission::Unit(action) => {
221+
action == required_action || action == Action::All || action == Action::SuperAdmin
222+
}
223+
Permission::Resource(action, ref resource_type) => {
224+
if let Some(resource_type) = resource_type.as_ref() {
225+
match resource_type {
226+
ParseableResourceType::Stream(resource_id)
227+
| ParseableResourceType::Llm(resource_id) => {
228+
let ok_resource = if let Some(context_resource_id) = context_resource {
229+
let is_internal = PARSEABLE
230+
.get_stream(context_resource_id, &Some(tenant_id.to_owned()))
231+
.is_ok_and(|stream| {
232+
stream
233+
.get_stream_type()
234+
.eq(&crate::storage::StreamType::Internal)
235+
});
236+
resource_id == context_resource_id || resource_id == "*" || is_internal
237+
} else {
238+
true
239+
};
240+
(action == required_action
241+
|| action == Action::All
242+
|| action == Action::SuperAdmin)
243+
&& ok_resource
244+
}
245+
ParseableResourceType::All => {
246+
action == required_action
247+
|| action == Action::All
248+
|| action == Action::SuperAdmin
249+
}
250+
}
251+
} else if resource_type.is_none()
252+
&& matches!(
253+
action,
254+
Action::Ingest
255+
| Action::Query
256+
| Action::ListStream
257+
| Action::GetSchema
258+
| Action::GetStats
259+
| Action::GetRetention
260+
| Action::PutRetention
261+
| Action::GetLLM
262+
| Action::QueryLLM
263+
| Action::ListLLM
264+
)
265+
{
266+
let ok_resource = if let Some(context_resource_id) = context_resource {
267+
let is_internal = PARSEABLE
268+
.get_stream(context_resource_id, &Some(tenant_id.to_owned()))
269+
.is_ok_and(|stream| {
270+
stream
271+
.get_stream_type()
272+
.eq(&crate::storage::StreamType::Internal)
273+
});
274+
!is_internal
275+
} else {
276+
true
277+
};
278+
action == required_action && ok_resource
279+
} else {
280+
false
281+
}
282+
}
283+
Permission::SelfUser if required_action == Action::GetUserRoles => {
284+
context_user.map(|x| x == username).unwrap_or_default()
285+
}
286+
_ => false,
287+
}) {
288+
Response::Authorized
289+
} else {
290+
Response::UnAuthorized
291+
}
292+
}
293+
196294
#[derive(Debug, Default)]
197295
pub struct Sessions {
198296
// map session key to user, tenant, and their permission
@@ -218,6 +316,16 @@ impl Sessions {
218316
self.user_sessions.remove(tenant_id);
219317
}
220318

319+
pub fn auth_snapshot(&self, key: &SessionKey) -> Option<AuthSnapShot> {
320+
self.active_sessions
321+
.get(key)
322+
.map(|(username, tenant_id, perms)| AuthSnapShot {
323+
username: username.clone(),
324+
tenant_id: tenant_id.clone(),
325+
permissions: perms.clone(),
326+
})
327+
}
328+
221329
// only checks if the session is expired or not
222330
pub fn is_session_expired(&self, key: &SessionKey) -> bool {
223331
// fetch userid from session key
@@ -328,120 +436,6 @@ impl Sessions {
328436
self.active_sessions.get(key).map(|(_, _, perms)| perms)
329437
}
330438

331-
// returns None if user is not in the map
332-
// Otherwise returns Some(Response) where response is authorized/unauthorized
333-
pub fn check_auth(
334-
&self,
335-
key: &SessionKey,
336-
required_action: Action,
337-
context_resource: Option<&str>,
338-
context_user: Option<&str>,
339-
) -> Option<Response> {
340-
self.active_sessions
341-
.get(key)
342-
.map(|(username, tenant_id, perms)| {
343-
let mut perms: HashSet<Permission> = HashSet::from_iter(perms.clone());
344-
perms.extend(aggregate_group_permissions(username, tenant_id));
345-
346-
if perms.iter().any(|user_perm| {
347-
match *user_perm {
348-
// if any action is ALL then we we authorize
349-
Permission::Unit(action) => {
350-
action == required_action
351-
|| action == Action::All
352-
|| action == Action::SuperAdmin
353-
}
354-
Permission::Resource(action, ref resource_type) => {
355-
if let Some(resource_type) = resource_type.as_ref() {
356-
// default flow for all actions other than global-ingestion (ingestion action without any dataset restriction)
357-
match resource_type {
358-
ParseableResourceType::Stream(resource_id)
359-
| ParseableResourceType::Llm(resource_id) => {
360-
let ok_resource =
361-
if let Some(context_resource_id) = context_resource {
362-
let is_internal = PARSEABLE
363-
.get_stream(
364-
context_resource_id,
365-
&Some(tenant_id.to_owned()),
366-
)
367-
.is_ok_and(|stream| {
368-
stream.get_stream_type().eq(
369-
&crate::storage::StreamType::Internal,
370-
)
371-
});
372-
resource_id == context_resource_id
373-
|| resource_id == "*"
374-
|| is_internal
375-
} else {
376-
// if no resource to match then resource check is not needed
377-
// WHEN IS THIS VALID??
378-
true
379-
};
380-
(action == required_action
381-
|| action == Action::All
382-
|| action == Action::SuperAdmin)
383-
&& ok_resource
384-
}
385-
ParseableResourceType::All => {
386-
action == required_action
387-
|| action == Action::All
388-
|| action == Action::SuperAdmin
389-
}
390-
}
391-
} else if resource_type.is_none()
392-
&& matches!(
393-
action,
394-
Action::Ingest
395-
| Action::Query
396-
| Action::ListStream
397-
| Action::GetSchema
398-
| Action::GetStats
399-
| Action::GetRetention
400-
| Action::PutRetention
401-
| Action::GetLLM
402-
| Action::QueryLLM
403-
| Action::ListLLM
404-
)
405-
{
406-
// flow for global-ingestion / global-query / global-reader / global-writer
407-
let ok_resource =
408-
if let Some(context_resource_id) = context_resource {
409-
let is_internal = PARSEABLE
410-
.get_stream(
411-
context_resource_id,
412-
&Some(tenant_id.to_owned()),
413-
)
414-
.is_ok_and(|stream| {
415-
stream
416-
.get_stream_type()
417-
.eq(&crate::storage::StreamType::Internal)
418-
});
419-
!is_internal
420-
} else {
421-
// if no resource to match then resource check is not needed
422-
// WHEN IS THIS VALID??
423-
true
424-
};
425-
action == required_action && ok_resource
426-
} else {
427-
// the default flow (some resource_type and an action) was covered in the first if
428-
// if the resource type is also None and action is not ingest then return with false
429-
false
430-
}
431-
}
432-
Permission::SelfUser if required_action == Action::GetUserRoles => {
433-
context_user.map(|x| x == username).unwrap_or_default()
434-
}
435-
_ => false,
436-
}
437-
}) {
438-
Response::Authorized
439-
} else {
440-
Response::UnAuthorized
441-
}
442-
})
443-
}
444-
445439
pub fn get_user_and_tenant_id(&self, key: &SessionKey) -> Option<(String, String)> {
446440
self.active_sessions
447441
.get(key)
@@ -493,49 +487,42 @@ impl From<Vec<User>> for Users {
493487

494488
fn aggregate_group_permissions(username: &str, tenant_id: &String) -> HashSet<Permission> {
495489
let mut group_perms = HashSet::new();
496-
497-
let user = if let Some(tenant_users) = users().get(tenant_id)
498-
&& let Some(user) = tenant_users.get(username)
499-
&& !user.protected
500-
{
501-
user.to_owned()
502-
} else {
503-
return group_perms;
490+
let user_groups: HashSet<String> = {
491+
if let Some(tenant_users) = users().get(tenant_id)
492+
&& let Some(user) = tenant_users.get(username)
493+
&& !user.protected
494+
{
495+
user.user_groups.clone()
496+
} else {
497+
return group_perms;
498+
}
504499
};
505-
// let Some(user) = users().get(username).cloned() else {
506-
// return group_perms;
507-
// };
508500

509-
if user.user_groups.is_empty() {
510-
return group_perms;
511-
}
501+
for group_name in &user_groups {
502+
let group_roles: HashSet<String> = {
503+
if let Some(groups) = read_user_groups().get(tenant_id)
504+
&& let Some(group) = groups.get(group_name)
505+
{
506+
group.roles.clone()
507+
} else {
508+
continue;
509+
}
510+
};
512511

513-
for group_name in &user.user_groups {
514-
if let Some(groups) = read_user_groups().get(tenant_id)
515-
&& let Some(group) = groups.get(group_name)
516-
{
517-
for role_name in group.roles.iter() {
518-
let roles = if let Some(tenant_roles) = roles().get(tenant_id)
519-
&& let Some(role) = tenant_roles.get(role_name)
512+
for role_name in group_roles {
513+
let role_privileges = {
514+
if let Some(tenant_roles) = roles().get(tenant_id)
515+
&& let Some(role) = tenant_roles.get(&role_name)
520516
{
521-
role.clone()
517+
role.privileges().to_vec()
522518
} else {
523519
continue;
524-
};
525-
// let Some(privileges) = roles().get(role_name).cloned() else {
526-
// continue;
527-
// };
528-
529-
for role in roles.privileges() {
530-
group_perms.extend(RoleBuilder::from(role).build());
531520
}
521+
};
522+
for privs in role_privileges {
523+
group_perms.extend(RoleBuilder::from(&privs).build());
532524
}
533-
} else {
534-
continue;
535-
};
536-
// let Some(group) = read_user_groups().get(group_name).cloned() else {
537-
// continue;
538-
// };
525+
}
539526
}
540527

541528
group_perms

0 commit comments

Comments
 (0)