From d63c40e04113bb41d65b363c06b705398442e3e9 Mon Sep 17 00:00:00 2001 From: Tim Lorsbach Date: Wed, 7 Oct 2026 14:59:57 +0200 Subject: [PATCH] Machine users --- bayer/views.py | 17 ++++++++- epdb/admin.py | 6 +++ epdb/legacy_api.py | 91 ++++++++++++++++++++++++++++++++++++++++++++-- epdb/logic.py | 13 ++++--- epdb/models.py | 30 +++++++++++++++ epdb/views.py | 6 +++ 6 files changed, 153 insertions(+), 10 deletions(-) diff --git a/bayer/views.py b/bayer/views.py index bcb9790c..651731d7 100644 --- a/bayer/views.py +++ b/bayer/views.py @@ -18,7 +18,13 @@ Package = s.GET_PACKAGE_MODEL() logger = logging.getLogger(__name__) -def has_secret_group(user): +def has_pes_allowance(user): + if user.principal_type == user.PrincipalType.APP: + return user.app_details.pes_allowance != user.app_details.PESAllowance.NONE + + return True + +def has_secret_pes_allowance(user): """ Determines if the specified user belongs to any secret group. @@ -32,6 +38,9 @@ def has_secret_group(user): bool: True if the user belongs to at least one secret group, False otherwise. """ + if user.principal_type == user.PrincipalType.APP: + return user.app_details.pes_allowance == user.app_details.PESAllowance.SECRET + return Group.objects.filter(secret=True, user_member=user).exists() @@ -196,6 +205,10 @@ def get_application_token(prod: bool) -> str: def fetch_pes(request, pes_url, user) -> dict: + if not has_pes_allowance(user): + logger.info(f"User {user.username} does not have permission to fetch PESs") + raise ValueError(f"User {user.username} does not have permission to fetch PESs") + for k, v in s.PES_API_MAPPING.items(): if pes_url.startswith(k): @@ -215,7 +228,7 @@ def fetch_pes(request, pes_url, user) -> dict: } # Restrict request if user is not part of any secret group - if not has_secret_group(user): + if not has_secret_pes_allowance(user): headers["app-classification-level-restriction"] = "restrict-pes-secret-structure-access" diff --git a/epdb/admin.py b/epdb/admin.py index 64601b1a..dd8ca45d 100644 --- a/epdb/admin.py +++ b/epdb/admin.py @@ -6,6 +6,7 @@ from django.contrib import messages from .models import ( AdditionalInformation, + AppDetails, ClassifierPluginModel, Compound, CompoundStructure, @@ -40,6 +41,10 @@ class AdditionalInformationAdmin(admin.ModelAdmin): pass +class AppDetailsAdmin(admin.ModelAdmin): + pass + + class UserAdmin(admin.ModelAdmin): list_display = [ "username", @@ -282,3 +287,4 @@ admin.site.register(Setting, SettingAdmin) admin.site.register(Scenario, ScenarioAdmin) admin.site.register(ExternalDatabase, ExternalDatabaseAdmin) admin.site.register(ExternalIdentifier, ExternalIdentifierAdmin) +admin.site.register(AppDetails, AppDetailsAdmin) diff --git a/epdb/legacy_api.py b/epdb/legacy_api.py index a2b891fb..09aee278 100644 --- a/epdb/legacy_api.py +++ b/epdb/legacy_api.py @@ -43,7 +43,7 @@ from .models import ( Scenario, SimpleAmbitRule, User, - UserPackagePermission, + UserPackagePermission, AppDetails, ) from .views import delete_with_log, get_remote_address @@ -122,7 +122,6 @@ def validate_token(token: str) -> dict: class MSBearerTokenAuth(HttpBearer): def authenticate(self, request, token): - auth_log.info(f"Authentication request by {get_remote_address(request)}") if token is None: @@ -130,6 +129,11 @@ class MSBearerTokenAuth(HttpBearer): claims = validate_token(token) + if claims.get("roles") is None or claims.get("roles") == [] or "envipath_registered_user" not in claims.get( + "roles"): + auth_log.error(f"API authentication attempt by {get_remote_address(request)} failed due to missing role") + return None + if not User.objects.filter(uuid=claims['oid']).exists(): auth_log.info(f"Authentication request by {get_remote_address(request)} failed!") return None @@ -564,7 +568,7 @@ class UpdatePackage(Schema): write: str | None = None -@router.post("/package/{uuid:package_uuid}", response={200: PackageSchema | Any, 400: Error}) +@router.post("/package/{uuid:package_uuid}", response={200: PackageSchema | Any, 400: Error, 403: Error}) def update_package(request, package_uuid, pack: Form[UpdatePackage]): try: p = get_package_for_write(request.user, package_uuid) @@ -590,6 +594,9 @@ def update_package(request, package_uuid, pack: Form[UpdatePackage]): grantee = GroupManager.get_group_lp(pack.ppsURI) else: grantee = UserManager.get_user_lp(pack.ppsURI) + if grantee.principal_type == request.user.PrincipalType.APP: + if not request.user.is_superuser: + return 403, {"message": "You are not allowed to grant permissions to an app!"} PackageManager.grant_read(request.user, p, grantee) return HttpResponse(status=200) @@ -598,6 +605,9 @@ def update_package(request, package_uuid, pack: Form[UpdatePackage]): grantee = GroupManager.get_group_lp(pack.ppsURI) else: grantee = UserManager.get_user_lp(pack.ppsURI) + if grantee.principal_type == request.user.PrincipalType.APP: + if not request.user.is_superuser: + return 403, {"message": "You are not allowed to grant permissions to an app!"} PackageManager.grant_write(request.user, p, grantee) return HttpResponse(status=200) @@ -2473,3 +2483,78 @@ def export(request, q: Query[PackageExportInSchema]): return 403, { "message": f"Exporting Package with id {q.package_uuid} failed due to insufficient rights!" } + + +class RegisterApplicationInSchema(Schema): + oid: str + client_id: str + name: str + secret: bool = False + data_pool_uuid: str | None = None + pes: str | None + + +@router.post("/integration", response={200: Any, 403: Error}) +def register_app(request, app: Form[RegisterApplicationInSchema]): + try: + user = request.user + if user.is_superuser: + user_name = f"APP_{app.oid}" + user_email = f"{app.oid}@bayer.com" + u = UserManager.create_user(user_name, user_email, None, uuid=app.oid, is_active=True, add_to_group=False) + + # Adjust objects + u.principal_type = User.PrincipalType.APP + + if app.pes: + pes_choice = next( + choice for choice in AppDetails.PESAllowance + if choice.label.lower() == app.pes.lower() + ) + + if pes_choice is None: + pes_choice = AppDetails.PESAllowance.NONE + else: + pes_choice = AppDetails.PESAllowance.NONE + + if app.data_pool_uuid is not None: + if not Group.objects.filter(uuid=app.data_pool_uuid, secret=True).exists(): + return 400, {"message": f"Group with id {app.data_pool_uuid} does not exist!"} + + data_pool = Group.objects.get(uuid=app.data_pool_uuid, secret=True) + else: + data_pool = None + + app_details = AppDetails.objects.create( + owner=user, + client_id=app.client_id, + display_name=app.name, + secret_allowed=app.secret, + data_pool=data_pool, + pes_allowance=pes_choice, + ) + u.app_details = app_details + + u.save() + + # Add the integration to the "All enviPath Users" Group + all_envipath_users = Group.objects.get(name="All enviPath Users") + if not all_envipath_users.user_member.contains(u): + all_envipath_users.user_member.add(u) + + if app.secret: + if app.data_pool_uuid is None or app.data_pool_uuid.strip() == "": + return 400, {"message": "Secret is only allowed if data pool is specified!"} + + default_package = u.default_package + default_package.data_pool = Group.objects.get(uuid=app.data_pool_uuid, secret=True) + default_package.classification_level = default_package.Classification.SECRET + default_package.save() + + return 200, {"message": "Application registered successfully"} + else: + raise ValueError("Only superusers can register apps") + except ValueError as e: + return 403, { + "message": f"Registering application failed due to {e}" + } \ No newline at end of file diff --git a/epdb/logic.py b/epdb/logic.py index f49cffc9..6b322d1e 100644 --- a/epdb/logic.py +++ b/epdb/logic.py @@ -220,11 +220,14 @@ class UserManager(object): u.save() if add_to_group: - g = Group.objects.get(public=True, name="enviPath Users") - g.user_member.add(u) - g.save() - u.default_group = g - u.save() + if Group.objects.filter(public=True, name="enviPath Users").exists(): + g = Group.objects.get(public=True, name="enviPath Users") + g.user_member.add(u) + g.save() + u.default_group = g + u.save() + else: + logger.info(f"add_to_group was set to True but no public group named 'enviPath Users' exists.") return u diff --git a/epdb/models.py b/epdb/models.py index 98318700..16785613 100644 --- a/epdb/models.py +++ b/epdb/models.py @@ -84,6 +84,17 @@ class User(AbstractUser): USERNAME_FIELD = "email" REQUIRED_FIELDS = ["username"] + class PrincipalType(models.IntegerChoices): + USER = 0, "User" + APP = 10 , "App" + + principal_type = models.IntegerField( + choices=PrincipalType, + default=PrincipalType.USER, + ) + + app_details = models.OneToOneField("epdb.AppDetails", on_delete=models.CASCADE, null=True, blank=True) + def get_name(self): return self.username @@ -103,6 +114,25 @@ class User(AbstractUser): return self.default_setting +class AppDetails(models.Model): + owner = models.ForeignKey(User, on_delete=models.CASCADE) + client_id = models.TextField(blank=False, null=False, verbose_name="Client ID") + display_name = models.TextField(blank=False, null=False, verbose_name="Display Name") + + secret_allowed = models.BooleanField(default=False) + data_pool = models.ForeignKey("epdb.Group", null=True, blank=True, on_delete=models.SET_NULL) + + class PESAllowance(models.IntegerChoices): + NONE = 0, "None" + NON_SECRET = 10 , "Non Secret" + SECRET = 20, "Secret" + + pes_allowance = models.IntegerField( + choices=PESAllowance, + default=PESAllowance.NONE, + ) + + class APIToken(TimeStampedModel): """ API authentication token for users. diff --git a/epdb/views.py b/epdb/views.py index fa334be6..b5307d1c 100644 --- a/epdb/views.py +++ b/epdb/views.py @@ -1294,6 +1294,9 @@ def package(request, package_uuid): ) users = users.filter(is_active=True).exclude(username="anonymous") + if not current_user.is_superuser: + users = users.exclude(principal_type=User.PrincipalType.APP) + group_perms = GroupPackagePermission.objects.filter(package=current_package) if current_package.classification_level == current_package.Classification.SECRET: @@ -1386,6 +1389,9 @@ def package(request, package_uuid): elif any([grantee_url, read, write, owner]): if "user" in grantee_url: grantee = UserManager.get_user_lp(grantee_url) + if grantee.principal_type == current_user.PrincipalType.APP: + if not request.user.is_superuser: + return error(request, "You are not allowed to grant permissions to an app!") else: grantee = GroupManager.get_group_lp(grantee_url)