diff --git a/gn_auth/auth/system/admin/resources.py b/gn_auth/auth/system/admin/resources.py
index d3a9f54..59d7686 100644
--- a/gn_auth/auth/system/admin/resources.py
+++ b/gn_auth/auth/system/admin/resources.py
@@ -1,8 +1,71 @@
"""Administrative endpoints concerning resources."""
+from uuid import UUID
-from flask import Blueprint
+from flask import jsonify, Response, Blueprint, current_app as app
+
+from gn_libs import sqlite3 as db
+
+from gn_auth.auth.errors import ForbiddenAccess
+from gn_auth.auth.requests import request_json
+from gn_auth.auth.authentication.users import user_by_id
+from gn_auth.auth.authentication.oauth2.resource_server import require_oauth
+from gn_auth.auth.authorisation.roles.models import (
+ user_roles_on_resource,
+ assign_user_role_by_name,
+ unassign_user_role_by_name)
+from gn_auth.auth.authorisation.resources.system.models import system_resource
resources = Blueprint("resources", __name__)
-# TODO: assign-owner and revoke-owner go here.
+def _require_assign_owner_privilege(conn, user) -> None:
+ """Raise ForbiddenAccess if user lacks system:resource:assign-owner on the system resource."""
+ _sys = system_resource(conn)
+ sys_roles = user_roles_on_resource(conn, user.user_id, _sys.resource_id)
+ sys_privs = tuple(
+ priv.privilege_id for role in sys_roles for priv in role.privileges)
+ if "system:resource:assign-owner" not in sys_privs:
+ raise ForbiddenAccess(
+ "You need the 'system:resource:assign-owner' privilege.")
+
+
+@resources.route("/<uuid:resource_id>/assign-owner", methods=["POST"])
+def assign_resource_owner(resource_id: UUID) -> Response:
+ """Assign the resource-owner role to a user on the given resource.
+
+ Only users with system:resource:assign-owner (sysadmins) may call this.
+ This is the correct path to bootstrap ownership on a resource that has
+ no owner yet.
+ """
+ with (require_oauth.acquire("profile group resource") as _token,
+ db.connection(app.config["AUTH_DB"]) as conn):
+ _require_assign_owner_privilege(conn, _token.user)
+ form = request_json()
+ target = user_by_id(conn, UUID(form["user_id"]))
+ with db.cursor(conn) as cursor:
+ assign_user_role_by_name(cursor, target, resource_id, "resource-owner")
+ return jsonify({
+ "user_id": form["user_id"],
+ "resource_id": str(resource_id),
+ "role_name": "resource-owner",
+ })
+
+
+@resources.route("/<uuid:resource_id>/revoke-owner", methods=["POST"])
+def revoke_resource_owner(resource_id: UUID) -> Response:
+ """Revoke the resource-owner role from a user on the given resource.
+
+ Requires the same system:resource:assign-owner privilege as assign-owner.
+ """
+ with (require_oauth.acquire("profile group resource") as _token,
+ db.connection(app.config["AUTH_DB"]) as conn):
+ _require_assign_owner_privilege(conn, _token.user)
+ form = request_json()
+ target = user_by_id(conn, UUID(form["user_id"]))
+ with db.cursor(conn) as cursor:
+ unassign_user_role_by_name(cursor, target, resource_id, "resource-owner")
+ return jsonify({
+ "user_id": form["user_id"],
+ "resource_id": str(resource_id),
+ "role_name": "resource-owner",
+ })
|
