specify authorization level needed when querying katsu

bento_beacon/endpoints/info.py

@@ -90,13 +90,13 @@ def beacon_overview():
 @info.route("/individual_schema", methods=["GET", "POST"])
 @authz_middleware.deco_public_endpoint
 def get_individual_schema():
-    return katsu_get(current_app.config["KATSU_INDIVIDUAL_SCHEMA_ENDPOINT"])
+    return katsu_get(current_app.config["KATSU_INDIVIDUAL_SCHEMA_ENDPOINT"], requires_auth="none")
 
 
 @info.route("/experiment_schema", methods=["GET", "POST"])
 @authz_middleware.deco_public_endpoint
 def get_experiment_schema():
-    return katsu_get(current_app.config["KATSU_EXPERIMENT_SCHEMA_ENDPOINT"])
+    return katsu_get(current_app.config["KATSU_EXPERIMENT_SCHEMA_ENDPOINT"], requires_auth="none")
 
 
 # -------------------------------------------------------

bento_beacon/utils/katsu_utils.py

@@ -1,10 +1,12 @@
 import requests
 from flask import current_app
+from functools import reduce
 from json import JSONDecodeError
 from urllib.parse import urlsplit, urlunsplit
+from typing import Literal
 from .exceptions import APIException, InvalidQuery
-from functools import reduce
 from ..authz.access import create_access_header_or_fall_back
+from ..authz.headers import auth_header_from_request
 
 
 def katsu_filters_query(beacon_filters, datatype, get_biosample_ids=False):
@@ -73,7 +75,7 @@ def katsu_network_call(payload, endpoint=None):
 
 
 # used for GET calls at particular katsu endpoints, eg /biosamples
-def katsu_get(endpoint, id=None, query=""):
+def katsu_get(endpoint, id=None, query="", requires_auth: Literal["none", "forwarded", "full"] = "none"):
     c = current_app.config
     katsu_base_url = c["KATSU_BASE_URL"]
     timeout = current_app.config["KATSU_TIMEOUT"]
@@ -92,7 +94,12 @@ def katsu_get(endpoint, id=None, query=""):
     )
 
     try:
-        r = requests.get(query_url, headers=create_access_header_or_fall_back(), timeout=timeout)
+        headers = {}
+        if requires_auth == "forwarded":
+            headers = auth_header_from_request()
+        elif requires_auth == "full":
+            headers = create_access_header_or_fall_back()
+        r = requests.get(query_url, headers=headers, timeout=timeout)
         katsu_response = r.json()
 
     except JSONDecodeError:
@@ -114,12 +121,13 @@ def katsu_get(endpoint, id=None, query=""):
 def search_from_config(config_filters):
     # query error checking handled in katsu
     query_string = "&".join(f'{cf["id"]}{cf["operator"]}{cf["value"]}' for cf in config_filters)
-    response = katsu_get(current_app.config["KATSU_BEACON_SEARCH"], query=query_string)
+    response = katsu_get(current_app.config["KATSU_BEACON_SEARCH"], query=query_string, requires_auth="full")
     return response.get("matches", [])
 
 
 def get_katsu_config_search_fields():
-    fields = katsu_get(current_app.config["KATSU_PUBLIC_CONFIG_ENDPOINT"])
+    # Use forwarded auth for getting available search fields, which may be limited based on access level
+    fields = katsu_get(current_app.config["KATSU_PUBLIC_CONFIG_ENDPOINT"], requires_auth="forwarded")
     current_app.config["KATSU_CONFIG_SEARCH_FIELDS"] = fields
     return fields
 
@@ -252,7 +260,7 @@ def get_filtering_terms():
 def katsu_total_individuals_count():
     c = current_app.config
     endpoint = c["KATSU_INDIVIDUALS_ENDPOINT"]
-    count_response = katsu_get(endpoint, query="page_size=1")
+    count_response = katsu_get(endpoint, query="page_size=1", requires_auth="full")
     count = count_response.get("count")
     return count
 
@@ -261,7 +269,8 @@ def katsu_datasets(id=None):
     c = current_app.config
     endpoint = c["KATSU_DATASETS_ENDPOINT"]
     try:
-        response = katsu_get(endpoint, id, query="format=phenopackets")
+        # right now, the datasets endpoint doesn't need any authorization for listing
+        response = katsu_get(endpoint, id, query="format=phenopackets", requires_auth="none")
     except APIException:
         return {}
 
@@ -295,12 +304,12 @@ def search_summary_statistics(ids):
 
 
 def overview_statistics():
-    return katsu_get(current_app.config["KATSU_PRIVATE_OVERVIEW"])
+    return katsu_get(current_app.config["KATSU_PRIVATE_OVERVIEW"], requires_auth="full")
 
 
 def katsu_censorship_settings() -> tuple[int | None, int | None]:
     # TODO: should be project-dataset scoped
-    rules = katsu_get(current_app.config["KATSU_PUBLIC_RULES"])
+    rules = katsu_get(current_app.config["KATSU_PUBLIC_RULES"], requires_auth="forwarded")
     max_filters = rules.get("max_query_parameters")
     count_threshold = rules.get("count_threshold")
     # return even if None