Machine users

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"
 
 

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)

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}"
+        }

epdb/logic.py

@@ -220,11 +220,14 @@ class UserManager(object):
             u.save()
 
         if add_to_group:
+            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
 

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.

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)