From c3b179f46e75e92de942a2d45d49c74cb887efc8 Mon Sep 17 00:00:00 2001
From: Frederick Muriuki Muriithi
Date: Fri, 7 Jun 2024 11:48:29 -0500
Subject: Replace `…/group/roles` endpoint with `…/resource/…/roles` endpoint.

The `…/group/roles` endpoint relied on the now deleted `group_roles`
table that caused the implementation to be prone to privilege
escalation attacks.

This commit provides the `…/resource/…/roles` endpoint that provides
the required functionality without the exposure.
---
 .../auth/authorisation/resources/groups/views.py   | 29 ---------------
 gn_auth/auth/authorisation/resources/views.py      | 41 ++++++++++++++++++++++
 2 files changed, 41 insertions(+), 29 deletions(-)

diff --git a/gn_auth/auth/authorisation/resources/groups/views.py b/gn_auth/auth/authorisation/resources/groups/views.py
index fb1a831..ef6bb0d 100644
--- a/gn_auth/auth/authorisation/resources/groups/views.py
+++ b/gn_auth/auth/authorisation/resources/groups/views.py
@@ -282,35 +282,6 @@ def link_data() -> Response:
 
         return jsonify(with_db_connection(__link__))
 
-@groups.route("/roles", methods=["GET"])
-@require_oauth("profile group")
-def group_roles():
-    """Return a list of all available group roles."""
-    with require_oauth.acquire("profile group role") as the_token:
-        def __list_roles__(conn: db.DbConnection):
-            ## TODO: Check that user has appropriate privileges
-            with db.cursor(conn) as cursor:
-                group = user_group(conn, the_token.user).maybe(# type: ignore[misc]
-                    DUMMY_GROUP, lambda grp: grp)
-                if group == DUMMY_GROUP:
-                    return tuple()
-                cursor.execute(
-                    "SELECT gr.group_role_id, r.* "
-                    "FROM group_roles AS gr INNER JOIN roles AS r "
-                    "ON gr.role_id=r.role_id "
-                    "WHERE group_id=?",
-                    (str(group.group_id),))
-                return tuple(
-                    GroupRole(uuid.UUID(row["group_role_id"]),
-                              group,
-                              Role(uuid.UUID(row["role_id"]),
-                                   row["role_name"],
-                                   bool(int(row["user_editable"])),
-                                   tuple()))
-                    for row in cursor.fetchall())
-        return jsonify(tuple(
-            asdict(role) for role in with_db_connection(__list_roles__)))
-
 @groups.route("/privileges", methods=["GET"])
 @require_oauth("profile group")
 def group_privileges():
diff --git a/gn_auth/auth/authorisation/resources/views.py b/gn_auth/auth/authorisation/resources/views.py
index c481ef9..3a61704 100644
--- a/gn_auth/auth/authorisation/resources/views.py
+++ b/gn_auth/auth/authorisation/resources/views.py
@@ -17,6 +17,7 @@ from gn_auth.auth.db import sqlite3 as db
 from gn_auth.auth.db.sqlite3 import with_db_connection
 
 from gn_auth.auth.authorisation.roles import Role
+from gn_auth.auth.authorisation.privileges import Privilege
 from gn_auth.auth.errors import InvalidData, InconsistencyError, AuthorisationError
 
 from gn_auth.auth.authentication.oauth2.resource_server import require_oauth
@@ -339,6 +340,46 @@ def toggle_public(resource_id: uuid.UUID) -> Response:
                 "Made resource public" if resource.public
                 else "Made resource private")})
 
+
+@resources.route("<uuid:resource_id>/roles", methods=["GET"])
+@require_oauth("profile group resource role")
+def resource_roles(resource_id: uuid.UUID) -> Response:
+    """Return the roles the user has to act on a given resource."""
+    with require_oauth.acquire("profile group resource role") as _token:
+        def __resultset_to_roles__(roles, row):
+            _role = roles.get(row["role_id"])
+            return {
+                **roles,
+                row["role_id"]: Role(
+                    role_id=uuid.UUID(row["role_id"]),
+                    role_name=row["role_name"],
+                    user_editable=bool(row["user_editable"]),
+                    privileges=(
+                        (_role.privileges if bool(_role) else tuple()) +
+                        (Privilege(
+                            privilege_id=row["privilege_id"],
+                            privilege_description=row[
+                                "privilege_description"]),)))
+                }
+
+
+        def __roles__(conn: db.DbConnection) -> tuple[Role, ...]:
+            with db.cursor(conn) as cursor:
+                cursor.execute(
+                    "SELECT r.*, p.* FROM resource_roles AS rr "
+                    "INNER JOIN roles AS r  ON rr.role_id=r.role_id "
+                    "INNER JOIN role_privileges AS rp ON r.role_id=rp.role_id "
+                    "INNER JOIN privileges AS p "
+                    "ON rp.privilege_id=p.privilege_id "
+                    "WHERE rr.resource_id=? AND rr.role_created_by=?",
+                    (str(resource_id), str(_token.user.user_id)))
+                return tuple(reduce(
+                    __resultset_to_roles__, cursor.fetchall(), {}).values())
+
+
+        return jsonify(with_db_connection(__roles__))
+
+
 @resources.route("/authorisation", methods=["POST"])
 def resources_authorisation():
     """Get user authorisations for given resource(s):"""
-- 
cgit v1.2.3