aboutsummaryrefslogtreecommitdiff
path: root/gn3/auth/authorisation/resources.py
blob: 0a919304bfbd587007ce1210681d1b9c8338d7dd (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
"""Handle the management of resources."""
from uuid import UUID, uuid4
from typing import Dict, Sequence, NamedTuple

from gn3.auth import db
from gn3.auth.authentication.users import User

from .checks import authorised_p
from .exceptions import AuthorisationError
from .groups import Group, user_group, is_group_leader, authenticated_user_group

class MissingGroupError(AuthorisationError):
    """Raised for any resource operation without a group."""

class ResourceCategory(NamedTuple):
    """Class representing a resource category."""
    resource_category_id: UUID
    resource_category_key: str
    resource_category_description: str

class Resource(NamedTuple):
    """Class representing a resource."""
    group: Group
    resource_id: UUID
    resource_name: str
    resource_category: ResourceCategory
    public: bool

@authorised_p(("create-resource",),  error_message="Could not create resource")
def create_resource(
        conn: db.DbConnection, resource_name: str,
        resource_category: ResourceCategory) -> Resource:
    """Create a resource item."""
    with db.cursor(conn) as cursor:
        group = authenticated_user_group(conn).maybe(False, lambda val: val)
        if not group:
            raise MissingGroupError(
                "User with no group cannot create a resource.")
        resource = Resource(group, uuid4(), resource_name, resource_category, False)
        cursor.execute(
            "INSERT INTO resources VALUES (?, ?, ?, ?, ?)",
            (str(resource.group.group_id), str(resource.resource_id),
             resource_name,
             str(resource.resource_category.resource_category_id),
             1 if resource.public else 0))

    return resource

def resource_categories(conn: db.DbConnection) -> Sequence[ResourceCategory]:
    """Retrieve all available resource categories"""
    with db.cursor(conn) as cursor:
        cursor.execute("SELECT * FROM resource_categories")
        return tuple(
            ResourceCategory(UUID(row[0]), row[1], row[2])
            for row in cursor.fetchall())
    return tuple()

def public_resources(conn: db.DbConnection) -> Sequence[Resource]:
    """List all resources marked as public"""
    categories = {
        str(cat.resource_category_id): cat for cat in resource_categories(conn)
    }
    with db.cursor(conn) as cursor:
        cursor.execute("SELECT * FROM resources WHERE public=1")
        results = cursor.fetchall()
        group_uuids = tuple(row[0] for row in results)
        query = ("SELECT * FROM groups WHERE group_id IN "
                 f"({', '.join(['?'] * len(group_uuids))})")
        cursor.execute(query, group_uuids)
        groups = {row[0]: Group(UUID(row[0]), row[1]) for row in cursor.fetchall()}
        return tuple(
            Resource(groups[row[0]], UUID(row[1]), row[2], categories[row[3]],
                     bool(row[4]))
            for row in results)

def group_leader_resources(
        cursor: db.DbCursor, user: User, group: Group,
        res_categories: Dict[UUID, ResourceCategory]) -> Sequence[Resource]:
    """Return all the resources available to the group leader"""
    if is_group_leader(cursor, user, group):
        cursor.execute("SELECT * FROM resources WHERE group_id=?",
                       (str(group.group_id),))
        return tuple(
            Resource(group, UUID(row[1]), row[2], res_categories[UUID(row[3])],
                     bool(row[4]))
            for row in cursor.fetchall())
    return tuple()

def user_resources(conn: db.DbConnection, user: User) -> Sequence[Resource]:
    """List the resources available to the user"""
    categories = { # Repeated in `public_resources` function
        cat.resource_category_id: cat for cat in resource_categories(conn)
    }
    with db.cursor(conn) as cursor:
        group = user_group(cursor, user).maybe(False, lambda val: val) # type: ignore[misc]
        if not group:
            return public_resources(conn)

        gl_resources = group_leader_resources(cursor, user, group, categories)

        cursor.execute(
            ("SELECT resources.* FROM group_user_roles_on_resources "
             "LEFT JOIN resources "
             "ON group_user_roles_on_resources.resource_id=resources.resource_id "
             "WHERE group_user_roles_on_resources.group_id = ? "
             "AND group_user_roles_on_resources.user_id = ?"),
            (str(group.group_id), str(user.user_id)))
        private_res = tuple(
                Resource(group, UUID(row[1]), row[2], categories[UUID(row[3])],
                         bool(row[4]))
                for row in cursor.fetchall())

    return tuple(set(private_res).union(gl_resources).union(public_resources(conn)))