From 67aa3731cbc92db4032fd6046cdaf0b04f834676 Mon Sep 17 00:00:00 2001 From: Tim Lorsbach Date: Thu, 20 Aug 2026 08:59:24 +0200 Subject: [PATCH] PES Adjustments, Check is User is in enviPath Entra Group --- Dockerfile | 1 + bayer/views.py | 120 +++++++++++++++++++++++++++++++-------------- epauth/views.py | 11 ++++- epdb/legacy_api.py | 4 +- 4 files changed, 96 insertions(+), 40 deletions(-) diff --git a/Dockerfile b/Dockerfile index f1e5eee2..f481fd60 100644 --- a/Dockerfile +++ b/Dockerfile @@ -85,6 +85,7 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ libxext6 \ libfontconfig1 \ nano \ + openjdk-21-jre-headless \ && rm -rf /var/lib/apt/lists/* RUN useradd -ms /bin/bash django diff --git a/bayer/views.py b/bayer/views.py index 689bea2e..76c65b08 100644 --- a/bayer/views.py +++ b/bayer/views.py @@ -8,7 +8,7 @@ from django.shortcuts import redirect from bayer.models import PESCompound from epdb.logic import PackageManager -from epdb.models import Pathway, Node +from epdb.models import Pathway, Node, Group from epdb.views import _anonymous_or_real, error from utilities.decorators import package_permission_required @@ -18,6 +18,23 @@ Package = s.GET_PACKAGE_MODEL() logger = logging.getLogger(__name__) +def has_secret_group(user): + """ + Determines if the specified user belongs to any secret group. + + This function checks whether the given user is a member of any group + that is marked as secret. + + Args: + user: The user for whom the check is performed. + + Returns: + bool: True if the user belongs to at least one secret group, + False otherwise. + """ + return Group.objects.filter(secret=True, user_member=user).exists() + + @package_permission_required() def create_pes(request, package_uuid): current_user = _anonymous_or_real(request) @@ -38,7 +55,7 @@ def create_pes(request, package_uuid): if pes_link: try: - pes_data = fetch_pes(request, pes_link) + pes_data = fetch_pes(request, pes_link, current_user) except ValueError as e: return error( request, @@ -98,7 +115,7 @@ def create_pes_node(request, package_uuid, pathway_uuid): if pes_link: try: - pes_data = fetch_pes(request, pes_link) + pes_data = fetch_pes(request, pes_link, current_user) except ValueError as e: return error( request, @@ -157,53 +174,82 @@ def create_pes_node(request, package_uuid, pathway_uuid): return HttpResponseNotAllowed(["POST"]) -def fetch_pes(request, pes_url) -> dict: - from epauth.views import get_access_token_from_request - token = get_access_token_from_request(request) +def get_application_token(prod: bool) -> str: + scope = f"{s.PROD_PES_SCOPE if prod else s.NON_PROD_PES_SCOPE}/.default" - if token is None: - token = pes_url.split('/')[-1] == 'dummy' + url = f"https://login.microsoftonline.com/{s.MS_TENANT_ID}/oauth2/v2.0/token" + data = { + "grant_type": "client_credentials", + "client_id": s.MS_ENTRA_CLIENT_ID, + "client_secret": s.MS_ENTRA_CLIENT_SECRET, + "scope": scope, + } - if token: - for k, v in s.PES_API_MAPPING.items(): - if pes_url.startswith(k): - pes_id = pes_url.split('/')[-1] + try: + response = requests.post(url, data=data) + response.raise_for_status() + return response.json()["access_token"] + except requests.exceptions.HTTPError as e: + logger.error(f"Could not fetch application token: {e}") + raise ValueError(f"Could not fetch application token!") - if pes_id == 'dummy': - import json - res_data = json.load(open(s.BASE_DIR / "fixtures/pes.json")) + +def fetch_pes(request, pes_url, user) -> dict: + + for k, v in s.PES_API_MAPPING.items(): + if pes_url.startswith(k): + + prod = "cropkey-np" not in pes_url + + pes_id = pes_url.split('/')[-1] + + if pes_id == 'dummy': + import json + res_data = json.load(open(s.BASE_DIR / "fixtures/pes.json")) + res_data["pes_url"] = pes_url + return res_data + else: + headers = { + "accept": "*/*", + "authorization": "Bearer " + get_application_token(prod), + } + + # Restrict request if user is not part of any secret group + if not has_secret_group(user): + headers["app-classification-level-restriction"] = "restrict-pes-secret-structure-access" + + + params = {"pes_reg_entity_corporate_id": pes_id} + + res = requests.get(v, headers=headers, params=params, proxies=s.PROXIES or None) + + try: + res.raise_for_status() + pes_data = res.json() + + # Handle missing response + if "detail" in pes_data and "The following PES Reg Entities Corporate Ids could not be found" in pes_data["detail"]: + raise ValueError(f"PES with id {pes_id} not found") + + # Ensure we have a entity + if len(pes_data) == 0: + raise ValueError(f"PES with id {pes_id} not found") + + res_data = pes_data[0] res_data["pes_url"] = pes_url return res_data - else: - headers = {"Authorization": f"Bearer {token['access_token']}"} - params = {"pes_reg_entity_corporate_id": pes_id} - res = requests.get(v, headers=headers, params=params, proxies=s.PROXIES or None) + except requests.exceptions.HTTPError as e: + raise ValueError(f"Error fetching PES with id {pes_id}: {e}") - try: - res.raise_for_status() - pes_data = res.json() - - if len(pes_data) == 0: - raise ValueError(f"PES with id {pes_id} not found") - - res_data = pes_data[0] - res_data["pes_url"] = pes_url - return res_data - - except requests.exceptions.HTTPError as e: - raise ValueError(f"Error fetching PES with id {pes_id}: {e}") - else: - raise ValueError(f"Unknown URL {pes_url}") - else: - raise ValueError("Could not fetch access token from request.") + raise ValueError(f"Unknown URL {pes_url}") def visualize_pes(request): pes_link = request.GET.get('pesLink') if pes_link: - pes_data = fetch_pes(request, pes_link) + pes_data = fetch_pes(request, pes_link, request.user) representations = pes_data.get('representations') diff --git a/epauth/views.py b/epauth/views.py index b7d2e5ee..feede8ee 100644 --- a/epauth/views.py +++ b/epauth/views.py @@ -9,7 +9,7 @@ from django.shortcuts import redirect from epdb.logic import UserManager, GroupManager from epdb.models import Group -from epdb.views import get_remote_address +from epdb.views import get_remote_address, error auth_log = logging.getLogger("auth") @@ -72,6 +72,15 @@ def entra_callback(request): claims = result["id_token_claims"] + if claims.get("roles") is None or claims.get("roles") == [] or "envipath_registered_user" not in claims.get("roles"): + auth_log.error(f"Login attempt by {get_remote_address(request)} failed due to missing role") + return error( + request, + "Login Failed", + "The user is not authenticated. A reason for this might be a missing assignment to the respective enviPath group.", + 403, + ) + user_name = claims.get("name") # preferred_username is a fallback for 2nd CWID user_email = claims.get("emailaddress", claims.get("email", claims.get("preferred_username"))) diff --git a/epdb/legacy_api.py b/epdb/legacy_api.py index 90acea9c..a2b891fb 100644 --- a/epdb/legacy_api.py +++ b/epdb/legacy_api.py @@ -889,7 +889,7 @@ def create_package_compound( from bayer.models import PESCompound try: - pes_data = fetch_pes(request, c.pesLink) + pes_data = fetch_pes(request, c.pesLink, request.user) except ValueError as e: return 400, {"message": f"Could not fetch PES data for {c.pesLink}"} @@ -2014,7 +2014,7 @@ def add_pathway_node(request, package_uuid, pathway_uuid, n: Form[CreateNode]): from bayer.models import PESCompound try: - pes_data = fetch_pes(request, n.pesLink) + pes_data = fetch_pes(request, n.pesLink, request.user) except ValueError as e: return 400, {"message": f"Could not fetch PES data for {n.pesLink}"}