diff options
| -rw-r--r-- | gn_auth/auth/authorisation/resources/models.py | 47 | ||||
| -rw-r--r-- | gn_auth/auth/authorisation/resources/views.py | 48 |
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 |