diff options
| -rw-r--r-- | gn_auth/auth/authorisation/data/views.py | 99 |
1 files changed, 62 insertions, 37 deletions
diff --git a/gn_auth/auth/authorisation/data/views.py b/gn_auth/auth/authorisation/data/views.py index 584b239..228d95f 100644 --- a/gn_auth/auth/authorisation/data/views.py +++ b/gn_auth/auth/authorisation/data/views.py @@ -104,10 +104,22 @@ def authorisation() -> Response: authconn, _dset_traits["ProbeSet"])) for _rrow in _rtypes } - if len(_all_resources.keys()) == 0: + if (len(_all_resources.keys()) == 0 and + len(_dset_traits.get("Temp", tuple())) == 0): raise NotFoundError( "No resource(s) found for specified trait(s). Do(es) the " "trait(s) actually exist?") + + # Handle Temp traits specially - they should be public/anonymous resources + if len(_dset_traits.get("Temp", tuple())) > 0: + # Create a synthetic public resource for Temp traits + # Use a predictable ID to identify synthetic temp resources + temp_resource_id = "gn-auth-temp-traits" + _all_resources[temp_resource_id] = { + "resource_id": temp_resource_id, + "resource_data": tuple(f"{dset}::{trait}" for dset, trait in _dset_traits["Temp"]) + } + _resource_ids = tuple(_all_resources.keys()) @@ -125,42 +137,55 @@ def authorisation() -> Response: } _paramstr = ", ".join(["?"] * len(_resource_ids)) - try: - with require_oauth.acquire("profile group resource") as _token: - user = _token.user - cursor.execute( - "SELECT ur.resource_id, r.role_id, rp.privilege_id " - "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 " - "WHERE ur.user_id = ? " - f"AND ur.resource_id IN ({_paramstr})", - (str(user.user_id),) + _resource_ids - ) - _privileges_by_resource: dict[str, tuple[str, ...]] = reduce( - lambda acc, curr: { - **acc, - curr["resource_id"]: ( - acc.get(curr["resource_id"], tuple()) - + (curr["privilege_id"],)) - }, - cursor.fetchall(), - {}) - except _HTTPException as exc: - err_msg = json.loads(exc.body) - if err_msg["error"] == "missing_authorization": - cursor.execute( - "SELECT rsc.resource_id " - "FROM resources AS rsc " - "WHERE rsc.public = '1' " - f"AND rsc.resource_id IN ({_paramstr}) ", - _resource_ids) - _privileges_by_resource = { - row["resource_id"]: ('group:resource:view-resource',) - for row in cursor.fetchall() - } - else: - raise exc from None + _privileges_by_resource: dict[str, tuple[str, ...]] = {} + + # Separate synthetic temp resources from real resources + temp_resource_id = "gn-auth-temp-traits" + real_resource_ids = tuple(rid for rid in _resource_ids if rid != temp_resource_id) + + # Query privileges only for real resources + if len(real_resource_ids) > 0: + real_paramstr = ", ".join(["?"] * len(real_resource_ids)) + try: + with require_oauth.acquire("profile group resource") as _token: + user = _token.user + cursor.execute( + "SELECT ur.resource_id, r.role_id, rp.privilege_id " + "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 " + "WHERE ur.user_id = ? " + f"AND ur.resource_id IN ({real_paramstr})", + (str(user.user_id),) + real_resource_ids + ) + _privileges_by_resource = reduce( + lambda acc, curr: { + **acc, + curr["resource_id"]: ( + acc.get(curr["resource_id"], tuple()) + + (curr["privilege_id"],)) + }, + cursor.fetchall(), + {}) + except _HTTPException as exc: + err_msg = json.loads(exc.body) + if err_msg["error"] == "missing_authorization": + cursor.execute( + "SELECT rsc.resource_id " + "FROM resources AS rsc " + "WHERE rsc.public = '1' " + f"AND rsc.resource_id IN ({real_paramstr}) ", + real_resource_ids) + _privileges_by_resource = { + row["resource_id"]: ('group:resource:view-resource',) + for row in cursor.fetchall() + } + else: + raise exc from None + + # Temp resources are always publicly viewable + if temp_resource_id in _resource_ids: + _privileges_by_resource[temp_resource_id] = ('group:resource:view-resource',) return jsonify({ "authorisation": [{ |