CVE-2022-1592: Server-Side Request Forgery (SSRF) (CWE-918)
value: Server-Side Request Forgery in scout in GitHub repository clinical-genomics/scout prior to v4.42. An attacker could make the application perform arbitrary requests to fishing steal cookie, request to private area, or lead to xss...
CWE description: The web server receives a URL or similar request from an upstream component and retrieves the contents of this URL, but it does not sufficiently ensure that the request is being sent to the expected destination.
CVSS v3 base score: 8.2
Published: 2022-05-05T11:15Z
Repository: https://github.com/clinical-genomics/scout
Language: Python
Files changed: CHANGELOG.md, scout/server/blueprints/alignviewers/controllers.py, scout/server/blueprints/alignviewers/views.py, tests/server/blueprints/alignviewers/test_alignviewers_views.py
Commit date: 2022-05-05 09:11:31 +0200
Commit message: Protect remote_cors from Server-Side Request Forgery (#3326)

* Protect remote_cors from Server-Side Request Forgery

* Improved code and added test

* Fix a couple of comments in a test
Diff: diff --git a/CHANGELOG.md b/CHANGELOG.md
index b99b08274..e27501ff2 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -27,10 +27,11 @@ About changelog [here](https://keepachangelog.com/en/1.0.0/)
 - Handle `attachment_filename` parameter renamed to `download_name` when Flask 2.2 will be released
 - Removed cursor timeout param in cases find adapter function to avoid many code warnings
 - Removed stream argument deprecation warning in tests
 - Handle `no intervals found` warning in load_region test
 - Beacon remove variants
+- Protect remote_cors function in alignviewers view from Server-Side Request Forgery (SSRF)
 
 ## [4.51]
 ### Added
 - Config file containing codecov settings for pull requests
 - Add an IGV.js direct link button from case page
diff --git a/scout/server/blueprints/alignviewers/controllers.py b/scout/server/blueprints/alignviewers/controllers.py
index e123a47ef..59c124aa3 100644
--- a/scout/server/blueprints/alignviewers/controllers.py
+++ b/scout/server/blueprints/alignviewers/controllers.py
@@ -12,19 +12,38 @@ from scout.utils.ensembl_rest_clients import EnsemblRestApiClient
 
 LOG = logging.getLogger(__name__)
 CUSTOM_TRACK_NAMES = ["Genes", "ClinVar", "ClinVar CNVs"]
 
 
+def check_session_tracks(resource):
+    """Make sure that a user requesting a resource is authenticated and resource is in session IGV tracks
+
+    Args:
+        resource(str): a resource on the server or on a remote URL
+
+    Returns
+        True is user has access to resource else False
+    """
+    # Check that user is logged in or that file extension is valid
+    if current_user.is_authenticated is False:
+        LOG.warning("Unauthenticated user requesting resource via remote_static")
+        return False
+    if resource not in session.get("igv_tracks", []):
+        LOG.warning(f"{resource} not in {session.get('igv_tracks', [])}")
+        return False
+    return True
+
+
 def set_session_tracks(display_obj):
     """Save igv tracks as a session object. This way it's easy to verify that a user is requesting one of these files from remote_static view endpoint
 
     Args:
         display_obj(dict): A display object containing case name, list of genes, lucus and tracks
     """
     session_tracks = list(display_obj.get("reference_track", {}).values())
     for key, track_items in display_obj.items():
-        if key not in ["tracks", "custom_tracks", "sample_tracks"]:
+        if key not in ["tracks", "custom_tracks", "sample_tracks", "cloud_public_tracks"]:
             continue
         for track_item in track_items:
             session_tracks += list(track_item.values())
 
     session["igv_tracks"] = session_tracks
diff --git a/scout/server/blueprints/alignviewers/views.py b/scout/server/blueprints/alignviewers/views.py
index 30bcd77db..de2e6d6b8 100644
--- a/scout/server/blueprints/alignviewers/views.py
+++ b/scout/server/blueprints/alignviewers/views.py
@@ -37,10 +37,14 @@ def remote_cors(remote_url):
         communicate CORS headers well, as in cloud tracks on figshare for IGV.js.
 
     Based on code from answers to this thread:
         https://stackoverflow.com/questions/6656363/proxying-to-another-web-service-with-flask/
     """
+    # Check that user is logged in or that file extension is valid
+    if controllers.check_session_tracks(remote_url) is False:
+        return abort(403)
+
     resp = requests.request(
         method=request.method,
         url=remote_url,
         headers={key: value for (key, value) in request.headers if key != "Host"},
         data=request.get_data(),
@@ -68,12 +72,11 @@ def remote_cors(remote_url):
 def remote_static():
     """Stream *large* static files with special requirements."""
     file_path = request.args.get("file") or "."
 
     # Check that user is logged in or that file extension is valid
-    if current_user.is_authenticated is False or file_path not in session.get("igv_tracks", []):
-        LOG.warning(f"{file_path} not in {session.get('igv_tracks', [])}")
+    if controllers.check_session_tracks(file_path) is False:
         return abort(403)
 
     range_header = request.headers.get("Range", None)
     if not range_header and (file_path.endswith(".bam") or file_path.endswith(".cram")):
         return abort(500)
diff --git a/tests/server/blueprints/alignviewers/test_alignviewers_views.py b/tests/server/blueprints/alignviewers/test_alignviewers_views.py
index db7e02a77..6b7b341b5 100644
--- a/tests/server/blueprints/alignviewers/test_alignviewers_views.py
+++ b/tests/server/blueprints/alignviewers/test_alignviewers_views.py
@@ -45,12 +45,13 @@ def test_remote_static(app):
 
     # GIVEN a running demo app
     with app.test_client() as client:
         # GIVEN that user is logged in
         client.get(url_for("auto_login"))
+
+        # GIVEN that resource file exists in user session
         with client.session_transaction() as session:
-            # GIVEN that resource file exists in user session
             session["igv_tracks"] = [file]
 
         # THEN the resource should be available to the user
         resp = client.get(
             url_for(
@@ -59,24 +60,45 @@ def test_remote_static(app):
             )
         )
         assert resp.status_code == 200
 
 
-def test_remote_cors(app):
+def test_remote_cors_wrong_resource(app):
     """Test endpoint that serves as a proxy to the actual remote track on the cloud"""
-    cloud_track_url = "http://google.com"
+    # GIVEN a resource not present in session["igv_tracks"]
+    an_url = "http://google.com"
 
-    # GIVEN an initialized app
-    # GIVEN a valid user and institute
+    # GIVEN a running demo app
     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 the remote cors endpoint is invoked with an url
+        resp = client.get(url_for("alignviewers.remote_cors", remote_url=an_url))
+
+        # THEN it should return forbidden (403)
+        assert resp.status_code == 403
+
+
+def test_remote_cors(app):
+    """Test endpoint that serves as a proxy to the actual remote track on the cloud"""
+    # GIVEN an igv track on the cloud
+    cloud_track_url = "https://s3-eu-west-1.amazonaws.com/pfigshare-u-files/25777460/GRCh37.variant_call.clinical.pathogenic_or_likely_pathogenic.vcf.gz.tbi"
+
+    # GIVEN a running demo app
+    with app.test_client() as client:
+        # GIVEN that the user could be logged in
+        resp = client.get(url_for("auto_login"))
+
+        # GIVEN that resource url exists in user session
+        with client.session_transaction() as session:
+            session["igv_tracks"] = [cloud_track_url]
+
+        # WHEN the remote cors endpoint is invoked with cloud_track_url
         resp = client.get(url_for("alignviewers.remote_cors", remote_url=cloud_track_url))
-        # THEN it should return success response
+
+        # THEN response should be successful
         assert resp.status_code == 200
 
 
 def test_igv_not_authorized(app, user_obj, case_obj, variant_obj):
     """Test view requests and produces igv alignments, when the user dosn't have access to the case"""

Vulnerable code:         if key not in ["tracks", "custom_tracks", "sample_tracks"]:
    if current_user.is_authenticated is False or file_path not in session.get("igv_tracks", []):
        LOG.warning(f"{file_path} not in {session.get('igv_tracks', [])}")
            # GIVEN that resource file exists in user session
def test_remote_cors(app):
    cloud_track_url = "http://google.com"
    # GIVEN an initialized app
    # GIVEN a valid user and institute
        assert resp.status_code == 200
        # THEN it should return success response
Fixed code: - Protect remote_cors function in alignviewers view from Server-Side Request Forgery (SSRF)
def check_session_tracks(resource):
    """Make sure that a user requesting a resource is authenticated and resource is in session IGV tracks

    Args:
        resource(str): a resource on the server or on a remote URL

    Returns
        True is user has access to resource else False
    """
    # Check that user is logged in or that file extension is valid
    if current_user.is_authenticated is False:
        LOG.warning("Unauthenticated user requesting resource via remote_static")
        return False
    if resource not in session.get("igv_tracks", []):
        LOG.warning(f"{resource} not in {session.get('igv_tracks', [])}")
        return False
    return True


        if key not in ["tracks", "custom_tracks", "sample_tracks", "cloud_public_tracks"]:
    # Check that user is logged in or that file extension is valid
    if controllers.check_session_tracks(remote_url) is False:
        return abort(403)

    if controllers.check_session_tracks(file_path) is False:

        # GIVEN that resource file exists in user session
def test_remote_cors_wrong_resource(app):
    # GIVEN a resource not present in session["igv_tracks"]
    an_url = "http://google.com"
    # GIVEN a running demo app
        resp = client.get(url_for("alignviewers.remote_cors", remote_url=an_url))

        # THEN it should return forbidden (403)
        assert resp.status_code == 403


def test_remote_cors(app):
    """Test endpoint that serves as a proxy to the actual remote track on the cloud"""
    # GIVEN an igv track on the cloud
    cloud_track_url = "https://s3-eu-west-1.amazonaws.com/pfigshare-u-files/25777460/GRCh37.variant_call.clinical.pathogenic_or_likely_pathogenic.vcf.gz.tbi"

    # GIVEN a running demo app
    with app.test_client() as client:
        # GIVEN that the user could be logged in
        resp = client.get(url_for("auto_login"))

        # GIVEN that resource url exists in user session
        with client.session_transaction() as session:
            session["igv_tracks"] = [cloud_track_url]

        # WHEN the remote cors endpoint is invoked with cloud_track_url

        # THEN response should be successful