diff options
Diffstat (limited to 'gn3/auth/authorisation/groups')
| -rw-r--r-- | gn3/auth/authorisation/groups/models.py | 52 | ||||
| -rw-r--r-- | gn3/auth/authorisation/groups/views.py | 59 |
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") |