"""Handle the management of resource/user groups.""" from uuid import UUID, uuid4 from typing import Sequence, Iterable, NamedTuple from flask import g from pymonad.maybe import Just, Maybe, Nothing from gn3.auth import db from gn3.auth.authentication.users import User from gn3.auth.authentication.checks import authenticated_p from .checks import authorised_p from .privileges import Privilege from .roles import Role, create_role from .errors import AuthorisationError class Group(NamedTuple): """Class representing a group.""" group_id: UUID group_name: str class GroupRole(NamedTuple): """Class representing a role tied/belonging to a group.""" group_role_id: UUID group: Group role: Role class MembershipError(AuthorisationError): """Raised when there is an error with a user's membership to a group.""" def __init__(self, user: User, groups: Sequence[Group]): """Initialise the `MembershipError` exception object.""" groups_str = ", ".join(group.group_name for group in groups) error_message = ( f"User '{user.name} ({user.email})' is a member of {len(groups)} " f"groups ({groups_str})") super().__init__(f"{type(self).__name__}: {error_message}.") def user_membership(conn: db.DbConnection, user: User) -> Sequence[Group]: """Returns all the groups that a member belongs to""" query = ( "SELECT groups.group_id, group_name FROM group_users INNER JOIN groups " "ON group_users.group_id=groups.group_id " "WHERE group_users.user_id=?") with db.cursor(conn) as cursor: cursor.execute(query, (str(user.user_id),)) groups = tuple(Group(row[0], row[1]) for row in cursor.fetchall()) return groups @authenticated_p @authorised_p(("system:group:create-group",), error_message="Failed to create group.") def create_group(conn: db.DbConnection, group_name: str, group_leader: User) -> Group: """Create a group""" group = Group(uuid4(), group_name) user_groups = user_membership(conn, group_leader) if len(user_groups) > 0: raise MembershipError(group_leader, user_groups) with db.cursor(conn) as cursor: cursor.execute( "INSERT INTO groups(group_id, group_name) VALUES (?, ?)", (str(group.group_id), group_name)) cursor.execute( "INSERT INTO group_users VALUES (?, ?)", (str(group.group_id), str(group_leader.user_id))) return group @authenticated_p @authorised_p(("group:role:create-role",), error_message="Could not create the group role") def create_group_role( conn: db.DbConnection, group: Group, role_name: str, privileges: Iterable[Privilege]) -> GroupRole: """Create a role attached to a group.""" with db.cursor(conn) as cursor: group_role_id = uuid4() role = create_role(cursor, role_name, privileges) cursor.execute( ("INSERT INTO group_roles(group_role_id, group_id, role_id) " "VALUES(?, ?, ?)"), (str(group_role_id), str(group.group_id), str(role.role_id))) return GroupRole(group_role_id, group, role) @authenticated_p def authenticated_user_group(conn) -> Maybe: """ Returns the currently authenticated user's group. Look into returning a Maybe object. """ user = g.user with db.cursor(conn) as cursor: cursor.execute( ("SELECT groups.group_id, groups.group_name 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]) for row in cursor.fetchall()) if len(groups) > 1: raise MembershipError(user, groups) if len(groups) == 1: return Just(groups[0]) return Nothing def user_group(cursor: db.DbCursor, user: User) -> Maybe: """Returns the given user's group""" cursor.execute( ("SELECT groups.group_id, groups.group_name 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]) for row in cursor.fetchall()) if len(groups) > 1: raise MembershipError(user, groups) if len(groups) == 1: return Just(groups[0]) return Nothing def is_group_leader(cursor: db.DbCursor, user: User, group: Group): """Check whether the given `user` is the leader of `group`.""" ugroup = user_group(cursor, user).maybe(False, lambda val: val) # type: ignore[misc] if not group: # User cannot be a group leader if not a member of ANY group return False if not ugroup == 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()) return "group-leader" in role_names