aboutsummaryrefslogtreecommitdiff
"""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 .exceptions 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(("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(("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