Allow document id variant API connect, and improve user cred check wh…

…en accessing variants (#4455) * Allow direct navigation to unique document_id specified variants * changelog * properly check permissions on variant * update changelog * fix changelog * changelog * Merge with safety fix. * leftover typos * if unset, use institute and case of variant * add test * lint... * lint * not that kind of object * Update CHANGELOG.md Co-authored-by: Chiara Rasi <[email protected]> * Refactor a common variant institute and case check * lint.. * TypeHints * would have been nice if that was linted. * one more missing variant test * fix that new route --------- Co-authored-by: Chiara Rasi <[email protected]> Co-authored-by: tereseboderus <[email protected]>
Clinical-Genomics · Mar 4, 2024 · 7b82e9e · 7b82e9e
1 parent 205a7a1
commit 7b82e9e
Show file tree

Hide file tree

Showing 10 changed files with 210 additions and 83 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -9,9 +9,11 @@ About changelog [here](https://keepachangelog.com/en/1.0.0/)
 - Added tags for Sniffles and CNVpytor, two LRS SV callers
 ### Changed
 - In the diagnoses page genes associated with a disease are displayed using hgnc symbol instead of hgnc id
+- Refactor view route to allow navigation directly to unique variant document id, improve permissions check
 ### Fixed
 - Refactored code in cases blueprints and variant_events adapter (set diseases for partial causative variants) to use "disease" instead of "omim" to encompass also ORPHA terms
 - Refactored code in `scout/parse/omim.py` and `scout/parse/disease_terms.py` to use "disease" instead of "phenotype" to differentiate from HPO terms
+- Be more careful about checking access to variant on API access
 
 ## [4.78]
 ### Added

diff --git a/scout/server/blueprints/api/views.py b/scout/server/blueprints/api/views.py
@@ -1,9 +1,11 @@
+from typing import Optional, Tuple
+
 from bson import json_util
 from flask import Blueprint, Response, abort, url_for
 from flask_login import current_user
 
 from scout.server.extensions import store
-from scout.server.utils import institute_and_case
+from scout.server.utils import institute_and_case, variant_institute_and_case
 
 api_bp = Blueprint("api", __name__, url_prefix="/api/v1")
 
@@ -17,19 +19,53 @@ def case(institute_id, case_name):
     return Response(json_util.dumps(case_obj), mimetype="application/json")
 
 
+def _lookup_variant(
+    variant_id: str,
+    institute_id: Optional[str] = None,
+    case_name: Optional[str] = None,
+) -> 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.
+
+    The check without institute and case is strict, as the user cannot choose what institute to
+    match against the ones the user has access to.
+    """
+
+    variant_obj = store.variant(variant_id)
+
+    if not variant_obj:
+        return abort(404)
+
+    institute_obj, case_obj = variant_institute_and_case(
+        store, variant_obj, institute_id, case_name
+    )
+
+    return (institute_obj, case_obj, variant_obj)
+
+
 @api_bp.route("/<institute_id>/<case_name>/<variant_id>")
-def variant(institute_id, case_name, variant_id):
+@api_bp.route("/variant/<variant_id>")
+def variant(
+    variant_id: str, institute_id: Optional[str] = None, case_name: Optional[str] = None
+) -> Optional[Response]:
     """Display a specific SNV variant."""
-    variant_obj = store.variant(variant_id)
+
+    (_, _, variant_obj) = _lookup_variant(variant_id, institute_id, case_name)
+
     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] = None, case_name: Optional[str] = None
+):
     """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(variant_id, institute_id, case_name)
 
     user_obj = store.user(current_user.email)
     link = url_for(
@@ -44,10 +80,13 @@ 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):
+@api_bp.route("/<variant_id>/unpin")
+def unpin_variant(
+    variant_id: str, institute_id: Optional[str] = None, case_name: Optional[str] = None
+):
     """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(variant_id, institute_id, case_name)
 
     user_obj = store.user(current_user.email)
     link = url_for(

diff --git a/scout/server/blueprints/cases/controllers.py b/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,

diff --git a/scout/server/blueprints/variant/controllers.py b/scout/server/blueprints/variant/controllers.py
@@ -1,9 +1,9 @@
 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
+from flask import Markup, abort, current_app, flash, url_for
 from flask_login import current_user
 
 from scout.adapter import MongoAdapter
@@ -33,8 +33,8 @@
     case_has_alignments,
     case_has_mt_alignments,
     case_has_rna_tracks,
-    institute_and_case,
     user_institutes,
+    variant_institute_and_case,
 )
 
 from .utils import (
@@ -54,7 +54,7 @@
 LOG = logging.getLogger(__name__)
 
 
-def tx_overview(variant_obj):
+def tx_overview(variant_obj: dict):
     """Prepares the content of the transcript overview to be shown on variant and general report pages.
        Basically show transcripts that contain RefSeq or are canonical.
 
@@ -128,7 +128,7 @@ def tx_overview(variant_obj):
     )
 
 
-def get_igv_tracks(build="37"):
+def get_igv_tracks(build: str = "37") -> set:
     """Return all available IGV tracks for the given genome build, as a set
 
     Args:
@@ -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,15 +194,19 @@ 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 variant_obj is None or variant_obj.get("case_id") != case_obj["_id"]:
-        return None
+    if not variant_obj:
+        return abort(404)
+
+    if not (institute_obj and case_obj):
+        (institute_obj, case_obj) = variant_institute_and_case(
+            store, variant_obj, institute_id, case_name
+        )
 
     variant_type = variant_type or variant_obj.get("variant_type", "clinical")
 
@@ -375,7 +379,7 @@ def variant(
     }
 
 
-def variant_rank_scores(store, case_obj, variant_obj):
+def variant_rank_scores(store: MongoAdapter, case_obj: dict, variant_obj: dict) -> list:
     """Retrive rank score values and ranges for the variant
 
     Args:
@@ -393,9 +397,6 @@ def variant_rank_scores(store, case_obj, variant_obj):
     ):  # Retrieve rank score results saved in variant document
         rank_score_results = variant_obj.get("rank_score_results")
 
-    rm_link_prefix = None
-    rm_file_extension = None
-
     if variant_obj.get("category") == "sv":
         rank_model_version = case_obj.get("sv_rank_model_version")
         rm_link_prefix = current_app.config.get("SV_RANK_MODEL_LINK_PREFIX")
@@ -514,14 +515,11 @@ def observations(store: MongoAdapter, loqusdb: LoqusDB, variant_obj: dict) -> Di
 
 
 def str_variant_reviewer(
-    store,
-    case_obj,
-    variant_id,
-):
+    store: MongoAdapter,
+    case_obj: dict,
+    variant_id: str,
+) -> dict:
     """Controller populating data and calling REViewer Service to fetch svg.
-    Args:
-        case_obj(dict)
-        variant_obj(dict)
     Returns:
         data(dict): {"individuals": list(dict())}}
             individual dicts being dict with keys:
@@ -577,7 +575,7 @@ def str_variant_reviewer(
     }
 
 
-def variant_acmg(store, institute_id, case_name, variant_id):
+def variant_acmg(store: MongoAdapter, institute_id: str, case_name: str, variant_id: str):
     """Collect data relevant for rendering ACMG classification form.
 
     Args:
@@ -589,8 +587,15 @@ def variant_acmg(store, institute_id, case_name, variant_id):
     Returns:
         data(dict): Things for the template
     """
-    institute_obj, case_obj = institute_and_case(store, institute_id, case_name)
     variant_obj = store.variant(variant_id)
+
+    if not variant_obj:
+        return abort(404)
+
+    institute_obj, case_obj = variant_institute_and_case(
+        store, variant_obj, institute_id, case_name
+    )
+
     return dict(
         institute=institute_obj,
         case=case_obj,
@@ -600,7 +605,9 @@ def variant_acmg(store, institute_id, case_name, variant_id):
     )
 
 
-def check_reset_variant_classification(store, evaluation_obj, link):
+def check_reset_variant_classification(
+    store: MongoAdapter, evaluation_obj: dict, link: str
+) -> bool:
     """Check if this was the last ACMG evaluation left on the variant.
     If there is still a classification we want to remove the classification.
 
@@ -613,32 +620,47 @@ def check_reset_variant_classification(store, evaluation_obj, link):
 
     """
 
-    if len(list(store.get_evaluations_case_specific(evaluation_obj["variant_specific"]))) == 0:
-        variant_obj = store.variant(document_id=evaluation_obj["variant_specific"])
-        acmg_classification = variant_obj.get("acmg_classification")
-        if isinstance(acmg_classification, int):
-            institute_obj, case_obj = institute_and_case(
-                store,
-                evaluation_obj["institute"]["_id"],
-                evaluation_obj["case"]["display_name"],
-            )
-            user_obj = store.user(current_user.email)
-
-            new_acmg = None
-            store.submit_evaluation(
-                variant_obj=variant_obj,
-                user_obj=user_obj,
-                institute_obj=institute_obj,
-                case_obj=case_obj,
-                link=link,
-                classification=new_acmg,
-            )
-            return True
+    if list(store.get_evaluations_case_specific(evaluation_obj["variant_specific"])):
+        return False
 
-    return False
+    variant_obj = store.variant(document_id=evaluation_obj["variant_specific"])
 
+    if not variant_obj:
+        return abort(404)
 
-def variant_acmg_post(store, institute_id, case_name, variant_id, user_email, criteria):
+    acmg_classification = variant_obj.get("acmg_classification")
+
+    if not isinstance(acmg_classification, int):
+        return False
+
+    institute_obj, case_obj = variant_institute_and_case(
+        store,
+        variant_obj,
+        evaluation_obj["institute"]["_id"],
+        evaluation_obj["case"]["display_name"],
+    )
+    user_obj = store.user(current_user.email)
+
+    new_acmg = None
+    store.submit_evaluation(
+        variant_obj=variant_obj,
+        user_obj=user_obj,
+        institute_obj=institute_obj,
+        case_obj=case_obj,
+        link=link,
+        classification=new_acmg,
+    )
+    return True
+
+
+def variant_acmg_post(
+    store: MongoAdapter,
+    institute_id: str,
+    case_name: str,
+    variant_id: str,
+    user_email: str,
+    criteria: list,
+) -> dict:
     """Calculate an ACMG classification based on a list of criteria.
 
     Args:
@@ -653,8 +675,15 @@ def variant_acmg_post(store, institute_id, case_name, variant_id, user_email, cr
         data(dict): Things for the template
 
     """
-    institute_obj, case_obj = institute_and_case(store, institute_id, case_name)
     variant_obj = store.variant(variant_id)
+
+    if not variant_obj:
+        return abort(404)
+
+    institute_obj, case_obj = variant_institute_and_case(
+        store, variant_obj, institute_id, case_name
+    )
+
     user_obj = store.user(user_email)
     variant_link = url_for(
         "variant.variant",

diff --git a/scout/server/blueprints/variant/verification_controllers.py b/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,
     )