3 files changed, 116 insertions, 0 deletions
diff --git a/gn_auth/auth/authorisation/users/masquerade/__init__.py b/gn_auth/auth/authorisation/users/masquerade/__init__.py
new file mode 100644
index 0000000..69d64f0
--- /dev/null
+++ b/gn_auth/auth/authorisation/users/masquerade/__init__.py
@@ -0,0 +1 @@
+"""Package to deal with masquerading."""
diff --git a/gn_auth/auth/authorisation/users/masquerade/models.py b/gn_auth/auth/authorisation/users/masquerade/models.py
new file mode 100644
index 0000000..9f24b6b
--- /dev/null
+++ b/gn_auth/auth/authorisation/users/masquerade/models.py
@@ -0,0 +1,67 @@
+"""Functions for handling masquerade."""
+from uuid import uuid4
+from functools import wraps
+from datetime import datetime
+
+from flask import current_app as app
+
+from gn3.auth import db
+
+from gn3.auth.authorisation.errors import ForbiddenAccess
+from gn3.auth.authorisation.roles.models import user_roles
+
+from gn3.auth.authentication.users import User
+from gn3.auth.authentication.oauth2.models.oauth2token import (
+    OAuth2Token, save_token)
+
+__FIVE_HOURS__ = (60 * 60 * 5)
+
+def can_masquerade(func):
+    """Security decorator."""
+    @wraps(func)
+    def __checker__(*args, **kwargs):
+        if len(args) == 3:
+            conn, token, _masq_user = args
+        elif len(args) == 2:
+            conn, token = args
+        elif len(args) == 1:
+            conn = args[0]
+            token = kwargs["original_token"]
+        else:
+            conn = kwargs["conn"]
+            token = kwargs["original_token"]
+
+        masq_privs = [priv for role in user_roles(conn, token.user)
+                      for priv in role.privileges
+                      if priv.privilege_id == "system:user:masquerade"]
+        if len(masq_privs) == 0:
+            raise ForbiddenAccess(
+                "You do not have the ability to masquerade as another user.")
+        return func(*args, **kwargs)
+    return __checker__
+
+@can_masquerade
+def masquerade_as(
+        conn: db.DbConnection,
+        original_token: OAuth2Token,
+        masqueradee: User) -> OAuth2Token:
+    """Get a token that enables `masquerader` to act as `masqueradee`."""
+    token_details = app.config["OAUTH2_SERVER"].generate_token(
+        client=original_token.client,
+        grant_type="authorization_code",
+        user=masqueradee,
+        expires_in=__FIVE_HOURS__,
+        include_refresh_token=True)
+    new_token = OAuth2Token(
+        token_id=uuid4(),
+        client=original_token.client,
+        token_type=token_details["token_type"],
+        access_token=token_details["access_token"],
+        refresh_token=token_details.get("refresh_token"),
+        scope=original_token.scope,
+        revoked=False,
+        issued_at=datetime.now(),
+        expires_in=token_details["expires_in"],
+        user=masqueradee)
+    save_token(conn, new_token)
+    return new_token
diff --git a/gn_auth/auth/authorisation/users/masquerade/views.py b/gn_auth/auth/authorisation/users/masquerade/views.py
new file mode 100644
index 0000000..43286a1
--- /dev/null
+++ b/gn_auth/auth/authorisation/users/masquerade/views.py
@@ -0,0 +1,48 @@
+"""Endpoints for user masquerade"""
+from uuid import UUID
+from functools import partial
+
+from flask import request, jsonify, Response, Blueprint
+
+from gn3.auth.db_utils import with_db_connection
+from gn3.auth.authorisation.errors import InvalidData
+from gn3.auth.authorisation.checks import require_json
+
+from gn3.auth.authentication.users import user_by_id
+from gn3.auth.authentication.oauth2.resource_server import require_oauth
+
+from .models import masquerade_as
+
+masq = Blueprint("masquerade", __name__)
+
+@masq.route("/", methods=["POST"])
+@require_oauth("profile user masquerade")
+@require_json
+def masquerade() -> Response:
+    """Masquerade as a particular user."""
+    with require_oauth.acquire("profile user masquerade") as token:
+        masqueradee_id = UUID(request.json["masquerade_as"])#type: ignore[index]
+        if masqueradee_id == token.user.user_id:
+            raise InvalidData("You are not allowed to masquerade as yourself.")
+
+        masq_user = with_db_connection(partial(
+            user_by_id, user_id=masqueradee_id))
+        def __masq__(conn):
+            new_token = masquerade_as(conn, original_token=token, masqueradee=masq_user)
+            return new_token
+        def __dump_token__(tok):
+            return {
+                key: value for key, value in (tok._asdict().items())
+                if key in ("access_token", "refresh_token", "expires_in",
+                           "token_type")
+            }
+        return jsonify({
+            "original": {
+                "user": token.user._asdict(),
+                "token": __dump_token__(token)
+            },
+            "masquerade_as": {
+                "user": masq_user._asdict(),
+                "token": __dump_token__(with_db_connection(__masq__))
+            }
+        })