aboutsummaryrefslogtreecommitdiff
path: root/gn3/auth/authorisation/groups
diff options
context:
space:
mode:
Diffstat (limited to 'gn3/auth/authorisation/groups')
-rw-r--r--gn3/auth/authorisation/groups/models.py52
-rw-r--r--gn3/auth/authorisation/groups/views.py59
2 files changed, 55 insertions, 56 deletions
diff --git a/gn3/auth/authorisation/groups/models.py b/gn3/auth/authorisation/groups/models.py
index 5a58322..bbe4ad6 100644
--- a/gn3/auth/authorisation/groups/models.py
+++ b/gn3/auth/authorisation/groups/models.py
@@ -142,30 +142,31 @@ def authenticated_user_group(conn) -> Maybe:
return Nothing
-def user_group(cursor: db.DbCursor, user: User) -> Maybe[Group]:
+def user_group(conn: db.DbConnection, user: User) -> Maybe[Group]:
"""Returns the given user's group"""
- cursor.execute(
- ("SELECT groups.group_id, groups.group_name, groups.group_metadata "
- "FROM group_users "
- "INNER JOIN groups ON group_users.group_id=groups.group_id "
- "WHERE group_users.user_id = ?"),
- (str(user.user_id),))
- groups = tuple(
- Group(UUID(row[0]), row[1], json.loads(row[2] or "{}"))
- for row in cursor.fetchall())
+ with db.cursor(conn) as cursor:
+ cursor.execute(
+ ("SELECT groups.group_id, groups.group_name, groups.group_metadata "
+ "FROM group_users "
+ "INNER JOIN groups ON group_users.group_id=groups.group_id "
+ "WHERE group_users.user_id = ?"),
+ (str(user.user_id),))
+ groups = tuple(
+ Group(UUID(row[0]), row[1], json.loads(row[2] or "{}"))
+ for row in cursor.fetchall())
- if len(groups) > 1:
- raise MembershipError(user, groups)
+ if len(groups) > 1:
+ raise MembershipError(user, groups)
- if len(groups) == 1:
- return Just(groups[0])
+ if len(groups) == 1:
+ return Just(groups[0])
return Nothing
-def is_group_leader(cursor: db.DbCursor, user: User, group: Group):
+def is_group_leader(conn: db.DbConnection, user: User, group: Group) -> bool:
"""Check whether the given `user` is the leader of `group`."""
- ugroup = user_group(cursor, user).maybe(
+ ugroup = user_group(conn, user).maybe(
False, lambda val: val) # type: ignore[arg-type, misc]
if not group:
# User cannot be a group leader if not a member of ANY group
@@ -175,13 +176,14 @@ def is_group_leader(cursor: db.DbCursor, user: User, group: Group):
# User cannot be a group leader if not a member of THIS group
return False
- cursor.execute(
- ("SELECT roles.role_name FROM user_roles LEFT JOIN roles "
- "ON user_roles.role_id = roles.role_id WHERE user_id = ?"),
- (str(user.user_id),))
- role_names = tuple(row[0] for row in cursor.fetchall())
+ with db.cursor(conn) as cursor:
+ cursor.execute(
+ ("SELECT roles.role_name FROM user_roles LEFT JOIN roles "
+ "ON user_roles.role_id = roles.role_id WHERE user_id = ?"),
+ (str(user.user_id),))
+ role_names = tuple(row[0] for row in cursor.fetchall())
- return "group-leader" in role_names
+ return "group-leader" in role_names
def all_groups(conn: db.DbConnection) -> Maybe[Sequence[Group]]:
"""Retrieve all existing groups"""
@@ -258,8 +260,8 @@ def group_by_id(conn: db.DbConnection, group_id: UUID) -> Group:
def join_requests(conn: db.DbConnection, user: User):
"""List all the join requests for the user's group."""
with db.cursor(conn) as cursor:
- group = user_group(cursor, user).maybe(DUMMY_GROUP, lambda grp: grp)# type: ignore[misc]
- if group != DUMMY_GROUP and is_group_leader(cursor, user, group):
+ group = user_group(conn, user).maybe(DUMMY_GROUP, lambda grp: grp)# type: ignore[misc]
+ if group != DUMMY_GROUP and is_group_leader(conn, user, group):
cursor.execute(
"SELECT gjr.*, u.email, u.name FROM group_join_requests AS gjr "
"INNER JOIN users AS u ON gjr.requester_id=u.user_id "
@@ -280,7 +282,7 @@ def accept_reject_join_request(
"""Accept/Reject a join request."""
assert status in ("ACCEPTED", "REJECTED"), f"Invalid status '{status}'."
with db.cursor(conn) as cursor:
- group = user_group(cursor, user).maybe(DUMMY_GROUP, lambda grp: grp) # type: ignore[misc]
+ group = user_group(conn, user).maybe(DUMMY_GROUP, lambda grp: grp) # type: ignore[misc]
cursor.execute("SELECT * FROM group_join_requests WHERE request_id=?",
(str(request_id),))
row = cursor.fetchone()
diff --git a/gn3/auth/authorisation/groups/views.py b/gn3/auth/authorisation/groups/views.py
index cf99975..7b967d7 100644
--- a/gn3/auth/authorisation/groups/views.py
+++ b/gn3/auth/authorisation/groups/views.py
@@ -76,7 +76,7 @@ def request_to_join(group_id: uuid.UUID) -> Response:
def __request__(conn: db.DbConnection, user: User, group_id: uuid.UUID,
message: str):
with db.cursor(conn) as cursor:
- group = user_group(cursor, user).maybe(# type: ignore[misc]
+ group = user_group(conn, user).maybe(# type: ignore[misc]
False, lambda grp: grp)# type: ignore[arg-type]
if group:
error = AuthorisationError(
@@ -148,7 +148,7 @@ def unlinked_data(resource_type: str) -> Response:
with require_oauth.acquire("profile group resource") as the_token:
db_uri = current_app.config["AUTH_DB"]
with db.connection(db_uri) as conn, db.cursor(conn) as cursor:
- ugroup = user_group(cursor, the_token.user).maybe(# type: ignore[misc]
+ ugroup = user_group(conn, the_token.user).maybe(# type: ignore[misc]
DUMMY_GROUP, lambda grp: grp)
if ugroup == DUMMY_GROUP:
return jsonify(tuple())
@@ -233,7 +233,7 @@ def group_roles():
def __list_roles__(conn: db.DbConnection):
## TODO: Check that user has appropriate privileges
with db.cursor(conn) as cursor:
- group = user_group(cursor, the_token.user).maybe(# type: ignore[misc]
+ group = user_group(conn, the_token.user).maybe(# type: ignore[misc]
DUMMY_GROUP, lambda grp: grp)
if group == DUMMY_GROUP:
return tuple()
@@ -291,8 +291,7 @@ def create_group_role():
raise InvalidData(
"At least one privilege needs to be provided.")
- with db.cursor(conn) as cursor:
- group = user_group(cursor, the_token.user).maybe(# type: ignore[misc]
+ group = user_group(conn, the_token.user).maybe(# type: ignore[misc]
DUMMY_GROUP, lambda grp: grp)
if group == DUMMY_GROUP:
@@ -314,9 +313,8 @@ def view_group_role(group_role_id: uuid.UUID):
"""Return the details of the given role."""
with require_oauth.acquire("profile group role") as the_token:
def __group_role__(conn: db.DbConnection) -> GroupRole:
- with db.cursor(conn) as cursor:
- group = user_group(cursor, the_token.user).maybe(#type: ignore[misc]
- DUMMY_GROUP, lambda grp: grp)
+ group = user_group(conn, the_token.user).maybe(#type: ignore[misc]
+ DUMMY_GROUP, lambda grp: grp)
if group == DUMMY_GROUP:
raise AuthorisationError(
@@ -329,29 +327,28 @@ def __add_remove_priv_to_from_role__(conn: db.DbConnection,
direction: str,
user: User) -> GroupRole:
assert direction in ("ADD", "DELETE")
- with db.cursor(conn) as cursor:
- group = user_group(cursor, user).maybe(# type: ignore[misc]
- DUMMY_GROUP, lambda grp: grp)
-
- if group == DUMMY_GROUP:
- raise AuthorisationError(
- "You need to be a member of a group to edit roles.")
- try:
- privilege_id = request.form.get("privilege_id", "")
- assert bool(privilege_id), "Privilege to add must be provided."
- privileges = privileges_by_ids(conn, (privilege_id,))
- if len(privileges) == 0:
- raise NotFoundError("Privilege not found.")
- dir_fns = {
- "ADD": add_privilege_to_group_role,
- "DELETE": delete_privilege_to_group_role
- }
- return dir_fns[direction](
- conn,
- group_role_by_id(conn, group, group_role_id),
- privileges[0])
- except AssertionError as aerr:
- raise InvalidData(aerr.args[0]) from aerr
+ group = user_group(conn, user).maybe(# type: ignore[misc]
+ DUMMY_GROUP, lambda grp: grp)
+
+ if group == DUMMY_GROUP:
+ raise AuthorisationError(
+ "You need to be a member of a group to edit roles.")
+ try:
+ privilege_id = request.form.get("privilege_id", "")
+ assert bool(privilege_id), "Privilege to add must be provided."
+ privileges = privileges_by_ids(conn, (privilege_id,))
+ if len(privileges) == 0:
+ raise NotFoundError("Privilege not found.")
+ dir_fns = {
+ "ADD": add_privilege_to_group_role,
+ "DELETE": delete_privilege_to_group_role
+ }
+ return dir_fns[direction](
+ conn,
+ group_role_by_id(conn, group, group_role_id),
+ privileges[0])
+ except AssertionError as aerr:
+ raise InvalidData(aerr.args[0]) from aerr
@groups.route("/role/<uuid:group_role_id>/privilege/add", methods=["POST"])
@require_oauth("profile group")