aboutsummaryrefslogtreecommitdiff
path: root/gn_auth
diff options
context:
space:
mode:
Diffstat (limited to 'gn_auth')
-rw-r--r--gn_auth/auth/authorisation/resources/models.py47
-rw-r--r--gn_auth/auth/authorisation/resources/views.py48
2 files changed, 91 insertions, 4 deletions
diff --git a/gn_auth/auth/authorisation/resources/models.py b/gn_auth/auth/authorisation/resources/models.py
index 451b165..25ae0fd 100644
--- a/gn_auth/auth/authorisation/resources/models.py
+++ b/gn_auth/auth/authorisation/resources/models.py
@@ -8,6 +8,8 @@ from gn_auth.auth.dictify import dictify
from gn_auth.auth.authentication.users import User
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.authorisation.checks import authorised_p
from gn_auth.auth.authorisation.errors import NotFoundError, AuthorisationError
@@ -371,3 +373,48 @@ def save_resource(
raise AuthorisationError(
"You do not have the appropriate privileges to edit this resource.")
+
+def user_roles_on_resources(conn: db.DbConnection,
+ user: User,
+ resource_ids: tuple[UUID] = tuple()) -> dict:
+ """Get roles on resources for a particular user."""
+ def __setup_roles__(old_roles, row):
+ roles = {role.role_id: role for role in old_roles}
+ role_id = UUID(row["role_id"])
+ role = roles.get(role_id, Role(
+ role_id, row["role_name"], bool(int(row["user_editable"])),
+ tuple()))
+ return tuple({
+ **roles, role_id: Role(
+ role.role_id, role.role_name, role.user_editable,
+ role.privileges + (Privilege(
+ row["privilege_id"], row["privilege_description"]),))
+ }.values())
+
+ def __organise__(acc, row):
+ resid = UUID(row["resource_id"])
+ roles = acc.get(resid, {}).get("roles", tuple())
+ return {
+ **acc,
+ resid: {
+ "roles": __setup_roles__(roles, row)
+ }
+ }
+
+ query = (
+ "SELECT ur.user_id, ur.resource_id, r.*, p.* "
+ "FROM user_roles AS ur "
+ "INNER JOIN roles AS r ON ur.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 ur.user_id=?")
+ params = (str(user.user_id),)
+
+ if len(resource_ids) > 0:
+ pholders = ", ".join(["?"] * len(resource_ids))
+ query = f"{query} AND ur.resource_id IN ({pholders})"
+ params = params + tuple(str(resid) for resid in resource_ids)
+
+ with db.cursor(conn) as cursor:
+ cursor.execute(query, params)
+ return reduce(__organise__, cursor.fetchall(), {})
diff --git a/gn_auth/auth/authorisation/resources/views.py b/gn_auth/auth/authorisation/resources/views.py
index 25f621d..60c751c 100644
--- a/gn_auth/auth/authorisation/resources/views.py
+++ b/gn_auth/auth/authorisation/resources/views.py
@@ -4,6 +4,7 @@ import json
import sqlite3
from functools import reduce
+from authlib.integrations.flask_oauth2.errors import _HTTPException
from flask import request, jsonify, Response, Blueprint, current_app as app
from gn_auth.auth.db import sqlite3 as db
@@ -18,10 +19,10 @@ from gn_auth.auth.authentication.users import User, user_by_id, user_by_email
from .checks import authorised_for
from .models import (
- Resource, resource_data, resource_by_id, resource_categories,
- assign_resource_user, link_data_to_resource, unassign_resource_user,
- resource_category_by_id, unlink_data_from_resource,
- create_resource as _create_resource)
+ Resource, resource_data, resource_by_id, public_resources,
+ resource_categories, assign_resource_user, link_data_to_resource,
+ unassign_resource_user, resource_category_by_id, user_roles_on_resources,
+ unlink_data_from_resource, create_resource as _create_resource)
from .groups.models import Group, resource_owner, group_role_by_id
resources = Blueprint("resources", __name__)
@@ -328,3 +329,42 @@ def toggle_public(resource_id: uuid.UUID) -> Response:
"description": (
"Made resource public" if resource.public
else "Made resource private")})
+
+@resources.route("/authorisation", methods=["POST"])
+def resources_authorisation():
+ """Get user authorisations for given resource(s):"""
+ try:
+ data = request.json
+ assert (data and "resource-ids" in data)
+ resource_ids = tuple(uuid.UUID(resid) for resid in data["resource-ids"])
+ pubres = tuple(
+ res.resource_id for res in with_db_connection(public_resources)
+ if res.resource_id in resource_ids)
+ with require_oauth.acquire("profile resource") as the_token:
+ resources = with_db_connection(lambda conn: user_roles_on_resources(
+ conn, the_token.user, resource_ids))
+ resp = jsonify({
+ str(resid): {
+ "public-read": resid in pubres,
+ "roles": tuple(
+ dictify(rol) for rol in
+ resources.get(resid, {}).get("roles", tuple()))
+ } for resid in resource_ids
+ })
+ except _HTTPException as _httpe:
+ err_msg = json.loads(_httpe.body)
+ if err_msg["error"] == "missing_authorization":
+ resp = jsonify({
+ str(resid): {
+ "public-read": resid in pubres
+ } for resid in resource_ids
+ })
+ except AssertionError as _aerr:
+ resp = jsonify({
+ "status": "bad-request",
+ "error_description": (
+ "Expected a JSON object with a 'resource-ids' key.")
+ })
+ resp.status_code = 400
+
+ return resp