aboutsummaryrefslogtreecommitdiff
path: root/gn3/auth/authorisation/roles.py
diff options
context:
space:
mode:
Diffstat (limited to 'gn3/auth/authorisation/roles.py')
-rw-r--r--gn3/auth/authorisation/roles.py25
1 files changed, 25 insertions, 0 deletions
diff --git a/gn3/auth/authorisation/roles.py b/gn3/auth/authorisation/roles.py
index 86759b1..e75163d 100644
--- a/gn3/auth/authorisation/roles.py
+++ b/gn3/auth/authorisation/roles.py
@@ -4,6 +4,7 @@ from functools import reduce
from typing import Any, Sequence, Iterable, NamedTuple
from pymonad.maybe import Just, Maybe, Nothing
+from pymonad.either import Left, Right, Either
from gn3.auth import db
from gn3.auth.dictify import dictify
@@ -12,6 +13,7 @@ from gn3.auth.authentication.checks import authenticated_p
from .checks import authorised_p
from .privileges import Privilege
+from .errors import AuthorisationError
class Role(NamedTuple):
"""Class representing a role: creates immutable objects."""
@@ -25,6 +27,11 @@ class Role(NamedTuple):
"role_id": self.role_id, "role_name": self.role_name,
"privileges": tuple(dictify(priv) for priv in self.privileges)
}
+
+class RoleNotFoundError(AuthorisationError):
+ """Raised whenever we try fetching (a) role(s) that do(es) not exist."""
+ error_code: int = 404
+
@authenticated_p
@authorised_p(("group:role:create-role",), error_message="Could not create role")
def create_role(
@@ -93,6 +100,24 @@ def user_roles(conn: db.DbConnection, user: User) -> Maybe[Sequence[Role]]:
reduce(__organise_privileges__, results, {}).values()))
return Nothing
+def user_role(conn: db.DbConnection, user: User, role_id: UUID) -> Either:
+ """Retrieve a specific non-resource role assigned to the user."""
+ with db.cursor(conn) as cursor:
+ cursor.execute(
+ "SELECT r.*, p.* FROM user_roles AS ur INNER JOIN roles AS r "
+ "ON ur.role_id=r.role_id INNER JOIN role_privileges AS rp "
+ "ON r.role_id=rp.role_id INNER JOIN privileges AS p "
+ "ON rp.privilege_id=p.privilege_id "
+ "WHERE ur.user_id=? AND ur.role_id=?",
+ (str(user.user_id), str(role_id)))
+
+ results = cursor.fetchall()
+ if results:
+ return Right(tuple(
+ reduce(__organise_privileges__, results, {}).values())[0])
+ return Left(RoleNotFoundError(
+ f"Could not find role with id '{role_id}'",))
+
def assign_default_roles(cursor: db.DbCursor, user: User):
"""Assign `user` some default roles."""
cursor.execute(