aboutsummaryrefslogtreecommitdiff
path: root/gn3/auth
diff options
context:
space:
mode:
Diffstat (limited to 'gn3/auth')
-rw-r--r--gn3/auth/authorisation/checks.py19
-rw-r--r--gn3/auth/authorisation/groups.py97
-rw-r--r--gn3/auth/authorisation/resources.py39
-rw-r--r--gn3/auth/authorisation/roles.py25
-rw-r--r--gn3/auth/authorisation/views.py30
5 files changed, 156 insertions, 54 deletions
diff --git a/gn3/auth/authorisation/checks.py b/gn3/auth/authorisation/checks.py
index d847c1e..8fef209 100644
--- a/gn3/auth/authorisation/checks.py
+++ b/gn3/auth/authorisation/checks.py
@@ -1,35 +1,38 @@
"""Functions to check for authorisation."""
from functools import wraps
-from typing import Callable
+from typing import Callable, Optional
from flask import g, current_app as app
from gn3.auth import db
+
from . import privileges as auth_privs
+from .errors import AuthorisationError
+
+from ..authentication.users import User
def authorised_p(
privileges: tuple[str],
error_message: str = (
- "You lack authorisation to perform requested action")):
+ "You lack authorisation to perform requested action"),
+ user: Optional[User] = None):
"""Authorisation decorator."""
assert len(privileges) > 0, "You must provide at least one privilege"
def __build_authoriser__(func: Callable):
@wraps(func)
def __authoriser__(*args, **kwargs):
- if hasattr(g, "user") and g.user:
+ the_user = user or (hasattr(g, "user") and g.user)
+ if the_user:
with db.connection(app.config["AUTH_DB"]) as conn:
user_privileges = tuple(
priv.privilege_id for priv in
- auth_privs.user_privileges(conn, g.user))
+ auth_privs.user_privileges(conn, the_user))
not_assigned = [
priv for priv in privileges if priv not in user_privileges]
if len(not_assigned) == 0:
return func(*args, **kwargs)
- return {
- "status": "error",
- "message": f"Unauthorised: {error_message}"
- }
+ raise AuthorisationError(error_message)
return __authoriser__
return __build_authoriser__
diff --git a/gn3/auth/authorisation/groups.py b/gn3/auth/authorisation/groups.py
index 3367b57..6d1b1a3 100644
--- a/gn3/auth/authorisation/groups.py
+++ b/gn3/auth/authorisation/groups.py
@@ -1,23 +1,31 @@
"""Handle the management of resource/user groups."""
+import json
from uuid import UUID, uuid4
-from typing import Sequence, Iterable, NamedTuple
+from typing import Any, Sequence, Iterable, Optional, NamedTuple
from flask import g
from pymonad.maybe import Just, Maybe, Nothing
from gn3.auth import db
from gn3.auth.authentication.users import User
+from gn3.auth.dictify import register_dictifier
from gn3.auth.authentication.checks import authenticated_p
from .checks import authorised_p
from .privileges import Privilege
-from .roles import Role, create_role
from .errors import AuthorisationError
+from .roles import (
+ Role, create_role, revoke_user_role_by_name, assign_user_role_by_name)
class Group(NamedTuple):
"""Class representing a group."""
group_id: UUID
group_name: str
+ group_metadata: dict[str, Any]
+
+register_dictifier(Group, lambda grp: {
+ "group_id": grp.group_id, "group_name": grp.group_name,
+ "group_metadata": grp.group_metadata})
class GroupRole(NamedTuple):
"""Class representing a role tied/belonging to a group."""
@@ -25,6 +33,9 @@ class GroupRole(NamedTuple):
group: Group
role: Role
+class GroupCreationError(AuthorisationError):
+ """Raised whenever a group creation fails"""
+
class MembershipError(AuthorisationError):
"""Raised when there is an error with a user's membership to a group."""
@@ -39,35 +50,43 @@ class MembershipError(AuthorisationError):
def user_membership(conn: db.DbConnection, user: User) -> Sequence[Group]:
"""Returns all the groups that a member belongs to"""
query = (
- "SELECT groups.group_id, group_name FROM group_users INNER JOIN groups "
+ "SELECT groups.group_id, group_name, groups.group_metadata "
+ "FROM group_users INNER JOIN groups "
"ON group_users.group_id=groups.group_id "
"WHERE group_users.user_id=?")
with db.cursor(conn) as cursor:
cursor.execute(query, (str(user.user_id),))
- groups = tuple(Group(row[0], row[1]) for row in cursor.fetchall())
+ groups = tuple(Group(row[0], row[1], json.loads(row[2]))
+ for row in cursor.fetchall())
return groups
@authenticated_p
-@authorised_p(("system:group:create-group",),
- error_message="Failed to create group.")
-def create_group(conn: db.DbConnection, group_name: str,
- group_leader: User) -> Group:
- """Create a group"""
- group = Group(uuid4(), group_name)
+def create_group(
+ conn: db.DbConnection, group_name: str, group_leader: User,
+ group_description: Optional[str] = None) -> Group:
+ """Create a new group."""
user_groups = user_membership(conn, group_leader)
if len(user_groups) > 0:
raise MembershipError(group_leader, user_groups)
- with db.cursor(conn) as cursor:
- cursor.execute(
- "INSERT INTO groups(group_id, group_name) VALUES (?, ?)",
- (str(group.group_id), group_name))
- cursor.execute(
- "INSERT INTO group_users VALUES (?, ?)",
- (str(group.group_id), str(group_leader.user_id)))
-
- return group
+ @authorised_p(
+ ("system:group:create-group",), (
+ "You do not have the appropriate privileges to enable you to "
+ "create a new group."),
+ group_leader)
+ def __create_group__():
+ with db.cursor(conn) as cursor:
+ new_group = __save_group__(
+ cursor, group_name,(
+ {"group_description": group_description}
+ if group_description else {}))
+ add_user_to_group(cursor, new_group, group_leader)
+ revoke_user_role_by_name(cursor, group_leader, "group-creator")
+ assign_user_role_by_name(cursor, group_leader, "group-leader")
+ return new_group
+
+ return __create_group__()
@authenticated_p
@authorised_p(("group:role:create-role",),
@@ -96,11 +115,12 @@ def authenticated_user_group(conn) -> Maybe:
user = g.user
with db.cursor(conn) as cursor:
cursor.execute(
- ("SELECT groups.group_id, groups.group_name FROM group_users "
+ ("SELECT groups.* FROM group_users "
"INNER JOIN groups ON group_users.group_id=groups.group_id "
"WHERE group_users.user_id = ?"),
(str(user.user_id),))
- groups = tuple(Group(UUID(row[0]), row[1]) for row in cursor.fetchall())
+ groups = tuple(Group(UUID(row[0]), row[1], json.loads(row[2] or "{}"))
+ for row in cursor.fetchall())
if len(groups) > 1:
raise MembershipError(user, groups)
@@ -110,14 +130,17 @@ def authenticated_user_group(conn) -> Maybe:
return Nothing
-def user_group(cursor: db.DbCursor, user: User) -> Maybe:
+def user_group(cursor: db.DbCursor, user: User) -> Maybe[Group]:
"""Returns the given user's group"""
cursor.execute(
- ("SELECT groups.group_id, groups.group_name FROM group_users "
+ ("SELECT groups.group_id, groups.group_name, groups.group_metadata "
+ "FROM group_users "
"INNER JOIN groups ON group_users.group_id=groups.group_id "
"WHERE group_users.user_id = ?"),
(str(user.user_id),))
- groups = tuple(Group(UUID(row[0]), row[1]) for row in cursor.fetchall())
+ groups = tuple(
+ Group(UUID(row[0]), row[1], json.loads(row[2] or "{}"))
+ for row in cursor.fetchall())
if len(groups) > 1:
raise MembershipError(user, groups)
@@ -129,7 +152,7 @@ def user_group(cursor: db.DbCursor, user: User) -> Maybe:
def is_group_leader(cursor: db.DbCursor, user: User, group: Group):
"""Check whether the given `user` is the leader of `group`."""
- ugroup = user_group(cursor, user).maybe(False, lambda val: val) # type: ignore[misc]
+ ugroup = user_group(cursor, user).maybe(False, lambda val: val) # type: ignore[arg-type, misc]
if not group:
# User cannot be a group leader if not a member of ANY group
return False
@@ -153,6 +176,28 @@ def all_groups(conn: db.DbConnection) -> Maybe[Sequence[Group]]:
res = cursor.fetchall()
if res:
return Just(tuple(
- Group(row["group_id"], row["group_name"]) for row in res))
+ Group(row["group_id"], row["group_name"],
+ json.loads(row["group_metadata"])) for row in res))
return Nothing
+
+def __save_group__(
+ cursor: db.DbCursor, group_name: str,
+ group_metadata: dict[str, Any]) -> Group:
+ """Save a group to db"""
+ the_group = Group(uuid4(), group_name, group_metadata)
+ cursor.execute(
+ ("INSERT INTO groups "
+ "VALUES(:group_id, :group_name, :group_metadata) "
+ "ON CONFLICT (group_id) DO UPDATE SET "
+ "group_name=:group_name, group_metadata=:group_metadata"),
+ {"group_id": str(the_group.group_id), "group_name": the_group.group_name,
+ "group_metadata": json.dumps(the_group.group_metadata)})
+ return the_group
+
+def add_user_to_group(cursor: db.DbCursor, the_group: Group, user: User):
+ """Add `user` to `the_group` as a member."""
+ cursor.execute(
+ ("INSERT INTO group_users VALUES (:group_id, :user_id) "
+ "ON CONFLICT (group_id, user_id) DO NOTHING"),
+ {"group_id": str(the_group.group_id), "user_id": str(user.user_id)})
diff --git a/gn3/auth/authorisation/resources.py b/gn3/auth/authorisation/resources.py
index f27d61a..29e50bf 100644
--- a/gn3/auth/authorisation/resources.py
+++ b/gn3/auth/authorisation/resources.py
@@ -1,4 +1,5 @@
"""Handle the management of resources."""
+import json
from uuid import UUID, uuid4
from typing import Dict, Sequence, NamedTuple
@@ -68,7 +69,11 @@ def public_resources(conn: db.DbConnection) -> Sequence[Resource]:
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()}
+ groups = {
+ row[0]: Group(
+ UUID(row[0]), row[1], json.loads(row[2] or "{}"))
+ for row in cursor.fetchall()
+ }
return tuple(
Resource(groups[row[0]], UUID(row[1]), row[2], categories[row[3]],
bool(row[4]))
@@ -93,22 +98,26 @@ def user_resources(conn: db.DbConnection, user: User) -> Sequence[Resource]:
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)
+ def __all_resources__(group) -> Sequence[Resource]:
+ 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(
+ 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({
+ res.resource_id: res
+ for res in
+ (private_res + gl_resources + public_resources(conn))# type: ignore[operator]
+ }.values())
- return tuple(set(private_res).union(gl_resources).union(public_resources(conn)))
+ # Fix the typing here
+ return user_group(cursor, user).map(__all_resources__).maybe(# type: ignore[arg-type,misc]
+ public_resources(conn), lambda res: res)# type: ignore[arg-type,return-value]
diff --git a/gn3/auth/authorisation/roles.py b/gn3/auth/authorisation/roles.py
index e84eb71..cd59a36 100644
--- a/gn3/auth/authorisation/roles.py
+++ b/gn3/auth/authorisation/roles.py
@@ -98,3 +98,28 @@ def assign_default_roles(cursor: db.DbCursor, user: User):
cursor.executemany(
("INSERT INTO user_roles VALUES (:user_id, :role_id)"),
params)
+
+def revoke_user_role_by_name(cursor: db.DbCursor, user: User, role_name: str):
+ """Revoke a role from `user` by the role's name"""
+ cursor.execute(
+ "SELECT role_id FROM roles WHERE role_name=:role_name",
+ {"role_name": role_name})
+ role = cursor.fetchone()
+ if role:
+ cursor.execute(
+ ("DELETE FROM user_roles "
+ "WHERE user_id=:user_id AND role_id=:role_id"),
+ {"user_id": str(user.user_id), "role_id": role["role_id"]})
+
+def assign_user_role_by_name(cursor: db.DbCursor, user: User, role_name: str):
+ """Revoke a role from `user` by the role's name"""
+ cursor.execute(
+ "SELECT role_id FROM roles WHERE role_name=:role_name",
+ {"role_name": role_name})
+ role = cursor.fetchone()
+
+ if role:
+ cursor.execute(
+ ("INSERT INTO user_roles VALUES(:user_id, :role_id) "
+ "ON CONFLICT DO NOTHING"),
+ {"user_id": str(user.user_id), "role_id": role["role_id"]})
diff --git a/gn3/auth/authorisation/views.py b/gn3/auth/authorisation/views.py
index d2f7d47..11e43eb 100644
--- a/gn3/auth/authorisation/views.py
+++ b/gn3/auth/authorisation/views.py
@@ -6,11 +6,13 @@ import sqlite3
from flask import request, jsonify, current_app
from gn3.auth import db
+from gn3.auth.dictify import dictify
from gn3.auth.blueprint import oauth2
from .errors import UserRegistrationError
-from .groups import user_group, all_groups
from .roles import assign_default_roles, user_roles as _user_roles
+from .groups import (
+ user_group, all_groups, GroupCreationError, create_group as _create_group)
from ..authentication.oauth2.resource_server import require_oauth
from ..authentication.users import save_user, set_user_password
@@ -29,7 +31,7 @@ def user_details():
"user_id": user.user_id,
"email": user.email,
"name": user.name,
- "group": group.maybe(False, lambda grp: grp)
+ "group": group.maybe(False, dictify)
})
@oauth2.route("/user-roles", methods=["GET"])
@@ -117,11 +119,29 @@ def register_user():
"unknown_error", "The system experienced an unexpected error.")
@oauth2.route("/groups", methods=["GET"])
-@require_oauth("profile")
+@require_oauth("profile group")
def groups():
"""Return the list of groups that exist."""
with db.connection(current_app.config["AUTH_DB"]) as conn:
the_groups = all_groups(conn)
- print(f"The groups: {the_groups}")
- return jsonify([])
+ return jsonify(the_groups.maybe(
+ [], lambda grps: [dictify(grp) for grp in grps]))
+
+@oauth2.route("/create-group", methods=["POST"])
+@require_oauth("profile group")
+def create_group():
+ """Create a new group."""
+ with require_oauth.acquire("profile group") as the_token:
+ group_name=request.form.get("group_name", "").strip()
+ if not bool(group_name):
+ raise GroupCreationError("Could not create the group.")
+
+ db_uri = current_app.config["AUTH_DB"]
+ with db.connection(db_uri) as conn:
+ user = the_token.user
+ new_group = _create_group(
+ conn, group_name, user, request.form.get("group_description"))
+ return jsonify({
+ **dictify(new_group), "group_leader": dictify(user)
+ })
generated by cgit v1.2.3 (git 2.44.0) at 2024-11-25 21:11:55 +0000