Ability to generate public QR codes in the UI and admin panel to manage access tokens

Author: BenjaminCharmes

pydatalab/src/pydatalab/permissions.py

@@ -10,19 +10,32 @@
 from pydatalab.logger import LOGGER
 from pydatalab.login import UserRole
 from pydatalab.models.people import AccountStatus
-from pydatalab.mongo import get_database
+from pydatalab.mongo import flask_mongo, get_database
 
 PUBLIC_USER_ID = ObjectId(24 * "0")
 
 
 def active_users_or_get_only(func):
     """Decorator to ensure that only active user accounts can access the route,
     unless it is a GET-route, in which case deactivated accounts can also access it.
-
+    Now also allows access with valid access tokens.
     """
 
     @wraps(func)
     def wrapped_route(*args, **kwargs):
+        access_token = request.args.get("at")
+        refcode = kwargs.get("refcode")
+
+        if not refcode and access_token:
+            path_parts = request.path.strip("/").split("/")
+            if len(path_parts) >= 2 and path_parts[0] == "items":
+                refcode = path_parts[1]
+
+        if access_token and refcode:
+            token_valid = check_access_token(refcode, access_token)
+            if token_valid:
+                return func(*args, **kwargs)
+
         if (
             (
                 current_user.is_authenticated
@@ -70,21 +83,23 @@ def check_access_token(refcode: str, token: str | None = None) -> bool:
 
     """
 
-    if not token:
+    if not token or not refcode:
         return False
 
-    db = get_database()
+    try:
+        if len(refcode.split(":")) != 2:
+            refcode = f"{CONFIG.IDENTIFIER_PREFIX}:{refcode}"
 
-    hashed_token = sha512(token.encode("utf-8")).hexdigest()
+        token_hash = sha512(token.encode("utf-8")).hexdigest()
 
-    access_document = db.api_keys.find_one(
-        {"token": hashed_token}, projection={"refcode": 1, "valid": 1}
-    )
-    if refcode == access_document["refcode"] and access_document["active"]:
-        LOGGER.info("Access to refcode %s granted with token", refcode, token)
-        return True
+        access_token_doc = flask_mongo.db.api_keys.find_one(
+            {"token": token_hash, "refcode": refcode, "active": True, "type": "access_token"}
+        )
+
+        return bool(access_token_doc)
 
-    return False
+    except Exception:
+        return False
 
 
 def get_default_permissions(

pydatalab/src/pydatalab/routes/v0_1/admin.py

@@ -1,4 +1,4 @@
-from hashlib import sha512
+import datetime
 
 from bson import ObjectId
 from flask import Blueprint, jsonify, request
@@ -97,17 +97,86 @@ def save_role(user_id):
     return (jsonify({"status": "success"}), 200)
 
 
-@ADMIN.route("/items/<refcode>/invalidate-access-token")
-def invalidate_access_token(refcode: str, METHODS=["POST"]):
-    request_json = request.get_json()  # noqa: F821 pylint: disable=undefined-variable
-    token = request_json.get("token")
-    if not token:
-        return jsonify({"status": "error", "detail": "No token provided."}), 400
+@ADMIN.route("/items/<refcode>/invalidate-access-token", methods=["POST"])
+def invalidate_access_token(refcode: str):
+    request_json = request.get_json(silent=True) or {}
+
+    if len(refcode.split(":")) != 2:
+        refcode = f"{CONFIG.IDENTIFIER_PREFIX}:{refcode}"
+
+    if request_json.get("token") == "admin-invalidation":
+        query = {"refcode": refcode, "active": True, "type": "access_token"}
+    else:
+        query = {"refcode": refcode, "active": True, "type": "access_token"}
 
     response = flask_mongo.db.api_keys.update_one(
-        {"token": sha512(token.encode("utf-8")).hexdigest()}, {"$set": {"active": False}}
+        query,
+        {
+            "$set": {
+                "active": False,
+                "invalidated_at": datetime.datetime.now(tz=datetime.timezone.utc),
+                "invalidated_by": ObjectId(current_user.id),
+            }
+        },
     )
+
     if response.modified_count == 1:
         return jsonify({"status": "success"}), 200
+    else:
+        return jsonify({"status": "error", "detail": "Token not found or already invalidated"}), 404
+
+
+@ADMIN.route("/access-tokens", methods=["GET"])
+def list_access_tokens():
+    """List all access tokens with their status and metadata."""
+
+    pipeline = [
+        {"$match": {"type": "access_token"}},
+        {
+            "$lookup": {
+                "from": "items",
+                "localField": "refcode",
+                "foreignField": "refcode",
+                "as": "item_info",
+            }
+        },
+        {
+            "$lookup": {
+                "from": "users",
+                "localField": "user",
+                "foreignField": "_id",
+                "as": "user_info",
+            }
+        },
+        {
+            "$project": {
+                "_id": 1,
+                "refcode": 1,
+                "active": 1,
+                "created_at": 1,
+                "invalidated_at": 1,
+                "token": {"$substr": ["$token", 0, 16]},
+                "item_name": {
+                    "$cond": {
+                        "if": {"$gt": [{"$size": "$item_info"}, 0]},
+                        "then": {"$arrayElemAt": ["$item_info.name", 0]},
+                        "else": None,
+                    }
+                },
+                "item_id": {"$arrayElemAt": ["$item_info.item_id", 0]},
+                "item_type": {
+                    "$cond": {
+                        "if": {"$gt": [{"$size": "$item_info"}, 0]},
+                        "then": {"$arrayElemAt": ["$item_info.type", 0]},
+                        "else": "deleted",
+                    }
+                },
+                "created_by": {"$arrayElemAt": ["$user_info.display_name", 0]},
+            }
+        },
+        {"$sort": {"created_at": -1}},
+    ]
+
+    tokens = list(flask_mongo.db.api_keys.aggregate(pipeline))
 
-    return jsonify({"status": "error", "detail": "Unable to invalidate token"}), 400
+    return jsonify({"status": "success", "tokens": tokens}), 200