Allow document id variant API connect, and improve user cred check when accessing variants

CHANGELOG.md

@@ -8,8 +8,10 @@ About changelog [here](https://keepachangelog.com/en/1.0.0/)
 ### Added
 - Added tags for Sniffles and CNVpytor, two LRS SV callers
 ### Changed
+- Refactor view route to allow navigation directly to unique variant document id, improve permissions check
 - In the diagnoses page genes associated with a disease are displayed using hgnc symbol instead of hgnc id
 ### Fixed
+- Be more careful about checking access to variant on api access
 
 ## [4.78]
 ### Added

scout/server/blueprints/api/views.py

@@ -1,3 +1,5 @@
+from typing import Optional, Tuple
+
 from bson import json_util
 from flask import Blueprint, Response, abort, url_for
 from flask_login import current_user
@@ -17,19 +19,54 @@ def case(institute_id, case_name):
     return Response(json_util.dumps(case_obj), mimetype="application/json")
 
 
+def _lookup_variant(
+    institute_id: str, case_name: str, variant_id: str
+) -> Optional[Tuple[dict, dict, dict]]:
+    """Lookup variant using variant document id. Return institute, case, and variant obj dicts.
+
+    The institute and case lookup adds a bit of security, checking if user is admin or
+    has access to institute and case, but since the variant is looked up using document_id,
+    we also run an additional security check that the given case_id matches the variant case_id."""
+
+    variant_obj = store.variant(variant_id)
+    if not variant_obj:
+        return abort(404)
+
+    if not case_name:
+        variant_case_obj = store.case(
+            case_id=variant_obj["case_id"], projection={"display_name": 1}
+        )
+        case_name = variant_case_obj["display_name"]
+
+    if not institute_id:
+        institute_id = variant_obj["institute"]
+
+    (institute_obj, case_obj) = institute_and_case(store, institute_id, case_name)
+
+    if case_obj is None:
+        return abort(404)
+
+    if variant_obj.get("case_id") != case_obj.get("_id"):
+        return abort(403)
+
+    return (institute_obj, case_obj, variant_obj)
+
+
 @api_bp.route("/<institute_id>/<case_name>/<variant_id>")
-def variant(institute_id, case_name, variant_id):
+def variant(institute_id: str, case_name: str, variant_id: str) -> Optional[Response]:
     """Display a specific SNV variant."""
-    variant_obj = store.variant(variant_id)
+
+    (_, _, variant_obj) = _lookup_variant(institute_id, case_name, variant_id)
+
     return Response(json_util.dumps(variant_obj), mimetype="application/json")
 
 
 @api_bp.route("/<institute_id>/<case_name>/<variant_id>/pin")
-def pin_variant(institute_id, case_name, variant_id):
+@api_bp.route("/<variant_id>/pin")
+def pin_variant(variant_id: str, institute_id: Optional[str], case_name: Optional[str]):
     """Pin an existing variant"""
 
-    institute_obj, case_obj = institute_and_case(store, institute_id, case_name)
-    variant_obj = store.variant(variant_id)
+    (institute_obj, case_obj, variant_obj) = _lookup_variant(institute_id, case_name, variant_id)
 
     user_obj = store.user(current_user.email)
     link = url_for(
@@ -46,8 +83,8 @@ def pin_variant(institute_id, case_name, variant_id):
 @api_bp.route("/<institute_id>/<case_name>/<variant_id>/unpin")
 def unpin_variant(institute_id, case_name, variant_id):
     """Un-pin an existing, pinned variant"""
-    institute_obj, case_obj = institute_and_case(store, institute_id, case_name)
-    variant_obj = store.variant(variant_id)
+
+    (institute_obj, case_obj, variant_obj) = _lookup_variant(institute_id, case_name, variant_id)
 
     user_obj = store.user(current_user.email)
     link = url_for(

scout/server/blueprints/cases/controllers.py

@@ -566,9 +566,9 @@ def case_report_variants(store: MongoAdapter, case_obj: dict, institute_obj: dic
     def _get_decorated_var(var_obj: dict) -> dict:
         return variant_decorator(
             store=store,
+            variant_id=None,
             institute_id=institute_obj["_id"],
             case_name=case_obj["display_name"],
-            variant_id=None,
             variant_obj=var_obj,
             add_other=False,
             get_overlapping=False,

scout/server/blueprints/variant/controllers.py

@@ -1,6 +1,6 @@
 import logging
 import os
-from typing import Dict, List
+from typing import Dict, List, Optional
 
 import requests
 from flask import Markup, current_app, flash, url_for
@@ -149,17 +149,17 @@ def get_igv_tracks(build="37"):
 
 
 def variant(
-    store,
-    institute_id,
-    case_name,
-    variant_id=None,
-    variant_obj=None,
-    add_other=True,
-    get_overlapping=True,
-    variant_type=None,
-    case_obj=None,
-    institute_obj=None,
-):
+    store: MongoAdapter,
+    variant_id: Optional[str],
+    institute_id: str = None,
+    case_name: str = None,
+    variant_obj: dict = None,
+    add_other: bool = True,
+    get_overlapping: bool = True,
+    variant_type: str = None,
+    case_obj: dict = None,
+    institute_obj: dict = None,
+) -> Optional[dict]:
     """Pre-process a single variant for the detailed variant view.
 
     Adds information from case and institute that is not present on the variant
@@ -194,13 +194,24 @@ def variant(
         }
 
     """
-    if not (institute_obj and case_obj):
-        institute_obj, case_obj = institute_and_case(store, institute_id, case_name)
+
     # If the variant is already collected we skip this part
     if not variant_obj:
         # NOTE this will query with variant_id == document_id, not the variant_id.
         variant_obj = store.variant(variant_id)
 
+    if not (institute_obj and case_obj):
+        if not case_name:
+            variant_case_obj = store.case(
+                case_id=variant_obj["case_id"], projection={"display_name": 1}
+            )
+            case_name = variant_case_obj["display_name"]
+
+        if not institute_id:
+            institute_id = variant_obj["institute"]
+
+        institute_obj, case_obj = institute_and_case(store, institute_id, case_name)
+
     if variant_obj is None or variant_obj.get("case_id") != case_obj["_id"]:
         return None
 

scout/server/blueprints/variant/verification_controllers.py

@@ -51,9 +51,9 @@ def variant_verification(
 
     data = variant_controller(
         store=store,
+        variant_id=variant_id,
         institute_id=institute_id,
         case_name=case_name,
-        variant_id=variant_id,
         add_other=False,
         get_overlapping=False,
     )

scout/server/blueprints/variant/views.py

@@ -41,14 +41,18 @@ def update_tracks_settings():
     return redirect(request.referrer)
 
 
+@variant_bp.route("/document_id/<variant_id>")
 @variant_bp.route("/<institute_id>/<case_name>/<variant_id>")
 @templated("variant/variant.html")
-def variant(institute_id, case_name, variant_id):
+def variant(variant_id, institute_id=None, case_name=None):
     """Display a specific SNV variant."""
     LOG.debug("Variants view requesting data for variant %s", variant_id)
 
     data = variant_controller(
-        store=store, institute_id=institute_id, case_name=case_name, variant_id=variant_id
+        store=store,
+        variant_id=variant_id,
+        institute_id=institute_id,
+        case_name=case_name,
     )
     if data is None:
         flash("An error occurred while retrieving variant object", "danger")
@@ -70,7 +74,7 @@ def cancer_variant(institute_id, case_name, variant_id):
     LOG.debug("Variants view requesting data for variant %s", variant_id)
 
     data = variant_controller(
-        store=store, institute_id=institute_id, case_name=case_name, variant_id=variant_id
+        store=store, variant_id=variant_id, institute_id=institute_id, case_name=case_name
     )
     if data is None:
         flash("An error occurred while retrieving variant object", "danger")
@@ -94,7 +98,10 @@ def cancer_variant(institute_id, case_name, variant_id):
 def sv_variant(institute_id, case_name, variant_id):
     """Display a specific structural variant."""
     data = variant_controller(
-        store=store, institute_id=institute_id, case_name=case_name, variant_id=variant_id
+        store=store,
+        variant_id=variant_id,
+        institute_id=institute_id,
+        case_name=case_name,
     )
 
     if data is None:

tests/server/blueprints/variant/test_variant_controllers.py

@@ -424,8 +424,8 @@ def test_variant_controller_with_compounds(app, institute_obj, case_obj):
         ## WHEN fetching the variant from the controller
         data = variant(
             store,
-            institute_id,
-            case_name,
+            institute_id=institute_id,
+            case_name=case_name,
             variant_id=var_id,
             add_other=True,
             get_overlapping=True,
@@ -461,9 +461,9 @@ def test_variant_controller_with_clnsig(app, institute_obj, case_obj):
         ## WHEN fetching the variant from the controller
         data = variant(
             store,
-            institute_id,
-            case_name,
             variant_id=var_id,
+            institute_id=institute_id,
+            case_name=case_name,
             add_other=True,
             get_overlapping=True,
             variant_type=category,
@@ -489,9 +489,9 @@ def test_variant_controller(app, institute_obj, case_obj, variant_obj):
         ## WHEN fetching the variant from the controller
         data = variant(
             store,
-            institute_id,
-            case_name,
             variant_id=var_id,
+            institute_id=institute_id,
+            case_name=case_name,
             add_other=True,
             get_overlapping=True,
             variant_type=category,

tests/server/blueprints/variant/test_variant_views.py

@@ -98,6 +98,26 @@ def test_variant(app, institute_obj, case_obj, variant_obj):
         assert resp.status_code == 200
 
 
+def test_variant_by_document_id(app, variant_obj):
+    # GIVEN an initialized app
+    # GIVEN a valid user
+
+    with app.test_client() as client:
+        # GIVEN that the user could be logged in
+        resp = client.get(url_for("auto_login"))
+        assert resp.status_code == 200
+
+        # WHEN sending a request (GET) to the variant page with only a variant ID
+        resp = client.get(
+            url_for(
+                "variant.variant",
+                variant_id=variant_obj["_id"],
+            )
+        )
+        # THEN it should return a page
+        assert resp.status_code == 200
+
+
 def test_cancer_variant(app, cancer_case_obj, cancer_variant_obj):
     # GIVEN a cancer case object with a variant saved in database
     assert store.case_collection.insert_one(cancer_case_obj)