forked from enviPath/enviPy
Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
5bbe316c24 | ||
|
|
d63c40e041 |
+15
-2
@@ -18,7 +18,13 @@ Package = s.GET_PACKAGE_MODEL()
|
|||||||
logger = logging.getLogger(__name__)
|
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.
|
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,
|
bool: True if the user belongs to at least one secret group,
|
||||||
False otherwise.
|
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()
|
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:
|
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():
|
for k, v in s.PES_API_MAPPING.items():
|
||||||
if pes_url.startswith(k):
|
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
|
# 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"
|
headers["app-classification-level-restriction"] = "restrict-pes-secret-structure-access"
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -6,6 +6,7 @@ from django.contrib import messages
|
|||||||
|
|
||||||
from .models import (
|
from .models import (
|
||||||
AdditionalInformation,
|
AdditionalInformation,
|
||||||
|
AppDetails,
|
||||||
ClassifierPluginModel,
|
ClassifierPluginModel,
|
||||||
Compound,
|
Compound,
|
||||||
CompoundStructure,
|
CompoundStructure,
|
||||||
@@ -40,6 +41,10 @@ class AdditionalInformationAdmin(admin.ModelAdmin):
|
|||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class AppDetailsAdmin(admin.ModelAdmin):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
class UserAdmin(admin.ModelAdmin):
|
class UserAdmin(admin.ModelAdmin):
|
||||||
list_display = [
|
list_display = [
|
||||||
"username",
|
"username",
|
||||||
@@ -282,3 +287,4 @@ admin.site.register(Setting, SettingAdmin)
|
|||||||
admin.site.register(Scenario, ScenarioAdmin)
|
admin.site.register(Scenario, ScenarioAdmin)
|
||||||
admin.site.register(ExternalDatabase, ExternalDatabaseAdmin)
|
admin.site.register(ExternalDatabase, ExternalDatabaseAdmin)
|
||||||
admin.site.register(ExternalIdentifier, ExternalIdentifierAdmin)
|
admin.site.register(ExternalIdentifier, ExternalIdentifierAdmin)
|
||||||
|
admin.site.register(AppDetails, AppDetailsAdmin)
|
||||||
|
|||||||
+88
-3
@@ -43,7 +43,7 @@ from .models import (
|
|||||||
Scenario,
|
Scenario,
|
||||||
SimpleAmbitRule,
|
SimpleAmbitRule,
|
||||||
User,
|
User,
|
||||||
UserPackagePermission,
|
UserPackagePermission, AppDetails,
|
||||||
)
|
)
|
||||||
from .views import delete_with_log, get_remote_address
|
from .views import delete_with_log, get_remote_address
|
||||||
|
|
||||||
@@ -122,7 +122,6 @@ def validate_token(token: str) -> dict:
|
|||||||
class MSBearerTokenAuth(HttpBearer):
|
class MSBearerTokenAuth(HttpBearer):
|
||||||
|
|
||||||
def authenticate(self, request, token):
|
def authenticate(self, request, token):
|
||||||
|
|
||||||
auth_log.info(f"Authentication request by {get_remote_address(request)}")
|
auth_log.info(f"Authentication request by {get_remote_address(request)}")
|
||||||
|
|
||||||
if token is None:
|
if token is None:
|
||||||
@@ -130,6 +129,11 @@ class MSBearerTokenAuth(HttpBearer):
|
|||||||
|
|
||||||
claims = validate_token(token)
|
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():
|
if not User.objects.filter(uuid=claims['oid']).exists():
|
||||||
auth_log.info(f"Authentication request by {get_remote_address(request)} failed!")
|
auth_log.info(f"Authentication request by {get_remote_address(request)} failed!")
|
||||||
return None
|
return None
|
||||||
@@ -564,7 +568,7 @@ class UpdatePackage(Schema):
|
|||||||
write: str | None = None
|
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]):
|
def update_package(request, package_uuid, pack: Form[UpdatePackage]):
|
||||||
try:
|
try:
|
||||||
p = get_package_for_write(request.user, package_uuid)
|
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)
|
grantee = GroupManager.get_group_lp(pack.ppsURI)
|
||||||
else:
|
else:
|
||||||
grantee = UserManager.get_user_lp(pack.ppsURI)
|
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)
|
PackageManager.grant_read(request.user, p, grantee)
|
||||||
return HttpResponse(status=200)
|
return HttpResponse(status=200)
|
||||||
@@ -598,6 +605,9 @@ def update_package(request, package_uuid, pack: Form[UpdatePackage]):
|
|||||||
grantee = GroupManager.get_group_lp(pack.ppsURI)
|
grantee = GroupManager.get_group_lp(pack.ppsURI)
|
||||||
else:
|
else:
|
||||||
grantee = UserManager.get_user_lp(pack.ppsURI)
|
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)
|
PackageManager.grant_write(request.user, p, grantee)
|
||||||
return HttpResponse(status=200)
|
return HttpResponse(status=200)
|
||||||
@@ -2473,3 +2483,78 @@ def export(request, q: Query[PackageExportInSchema]):
|
|||||||
return 403, {
|
return 403, {
|
||||||
"message": f"Exporting Package with id {q.package_uuid} failed due to insufficient rights!"
|
"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}"
|
||||||
|
}
|
||||||
+8
-5
@@ -220,11 +220,14 @@ class UserManager(object):
|
|||||||
u.save()
|
u.save()
|
||||||
|
|
||||||
if add_to_group:
|
if add_to_group:
|
||||||
g = Group.objects.get(public=True, name="enviPath Users")
|
if Group.objects.filter(public=True, name="enviPath Users").exists():
|
||||||
g.user_member.add(u)
|
g = Group.objects.get(public=True, name="enviPath Users")
|
||||||
g.save()
|
g.user_member.add(u)
|
||||||
u.default_group = g
|
g.save()
|
||||||
u.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
|
return u
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,34 @@
|
|||||||
|
# Generated by Django 6.0.3 on 2026-10-07 08:08
|
||||||
|
|
||||||
|
import django.db.models.deletion
|
||||||
|
from django.conf import settings
|
||||||
|
from django.db import migrations, models
|
||||||
|
|
||||||
|
|
||||||
|
class Migration(migrations.Migration):
|
||||||
|
|
||||||
|
dependencies = [
|
||||||
|
('epdb', '0030_auto_20260814_0741'),
|
||||||
|
]
|
||||||
|
|
||||||
|
operations = [
|
||||||
|
migrations.AddField(
|
||||||
|
model_name='user',
|
||||||
|
name='principal_type',
|
||||||
|
field=models.IntegerField(choices=[(0, 'Internal'), (10, 'Restricted')], default=0),
|
||||||
|
),
|
||||||
|
migrations.CreateModel(
|
||||||
|
name='AppDetails',
|
||||||
|
fields=[
|
||||||
|
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||||
|
('client_id', models.TextField(verbose_name='Client ID')),
|
||||||
|
('display_name', models.TextField(verbose_name='Display Name')),
|
||||||
|
('owner', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
|
||||||
|
],
|
||||||
|
),
|
||||||
|
migrations.AddField(
|
||||||
|
model_name='user',
|
||||||
|
name='app_details',
|
||||||
|
field=models.OneToOneField(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, to='epdb.appdetails'),
|
||||||
|
),
|
||||||
|
]
|
||||||
@@ -0,0 +1,18 @@
|
|||||||
|
# Generated by Django 6.0.3 on 2026-10-07 08:56
|
||||||
|
|
||||||
|
from django.db import migrations, models
|
||||||
|
|
||||||
|
|
||||||
|
class Migration(migrations.Migration):
|
||||||
|
|
||||||
|
dependencies = [
|
||||||
|
('epdb', '0031_user_principal_type_appdetails_user_app_details'),
|
||||||
|
]
|
||||||
|
|
||||||
|
operations = [
|
||||||
|
migrations.AlterField(
|
||||||
|
model_name='user',
|
||||||
|
name='principal_type',
|
||||||
|
field=models.IntegerField(choices=[(0, 'User'), (10, 'App')], default=0),
|
||||||
|
),
|
||||||
|
]
|
||||||
@@ -0,0 +1,17 @@
|
|||||||
|
# Generated by Django 6.0.3 on 2026-10-07 08:57
|
||||||
|
|
||||||
|
from django.db import migrations
|
||||||
|
|
||||||
|
|
||||||
|
class Migration(migrations.Migration):
|
||||||
|
|
||||||
|
dependencies = [
|
||||||
|
('epdb', '0032_alter_user_principal_type'),
|
||||||
|
]
|
||||||
|
|
||||||
|
operations = [
|
||||||
|
migrations.RemoveField(
|
||||||
|
model_name='user',
|
||||||
|
name='app_details',
|
||||||
|
),
|
||||||
|
]
|
||||||
@@ -0,0 +1,19 @@
|
|||||||
|
# Generated by Django 6.0.3 on 2026-10-07 10:14
|
||||||
|
|
||||||
|
import django.db.models.deletion
|
||||||
|
from django.db import migrations, models
|
||||||
|
|
||||||
|
|
||||||
|
class Migration(migrations.Migration):
|
||||||
|
|
||||||
|
dependencies = [
|
||||||
|
('epdb', '0033_remove_user_app_details'),
|
||||||
|
]
|
||||||
|
|
||||||
|
operations = [
|
||||||
|
migrations.AddField(
|
||||||
|
model_name='user',
|
||||||
|
name='app_details',
|
||||||
|
field=models.OneToOneField(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, to='epdb.appdetails'),
|
||||||
|
),
|
||||||
|
]
|
||||||
@@ -0,0 +1,24 @@
|
|||||||
|
# Generated by Django 6.0.3 on 2026-10-07 10:26
|
||||||
|
|
||||||
|
import django.db.models.deletion
|
||||||
|
from django.db import migrations, models
|
||||||
|
|
||||||
|
|
||||||
|
class Migration(migrations.Migration):
|
||||||
|
|
||||||
|
dependencies = [
|
||||||
|
('epdb', '0034_user_app_details'),
|
||||||
|
]
|
||||||
|
|
||||||
|
operations = [
|
||||||
|
migrations.AddField(
|
||||||
|
model_name='appdetails',
|
||||||
|
name='data_pool',
|
||||||
|
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, to='epdb.group'),
|
||||||
|
),
|
||||||
|
migrations.AddField(
|
||||||
|
model_name='appdetails',
|
||||||
|
name='secret_allowed',
|
||||||
|
field=models.BooleanField(default=False),
|
||||||
|
),
|
||||||
|
]
|
||||||
@@ -0,0 +1,18 @@
|
|||||||
|
# Generated by Django 6.0.3 on 2026-10-07 10:27
|
||||||
|
|
||||||
|
from django.db import migrations, models
|
||||||
|
|
||||||
|
|
||||||
|
class Migration(migrations.Migration):
|
||||||
|
|
||||||
|
dependencies = [
|
||||||
|
('epdb', '0035_appdetails_data_pool_appdetails_secret_allowed'),
|
||||||
|
]
|
||||||
|
|
||||||
|
operations = [
|
||||||
|
migrations.AddField(
|
||||||
|
model_name='appdetails',
|
||||||
|
name='pes_allowance',
|
||||||
|
field=models.IntegerField(choices=[(0, 'None'), (10, 'Non Secret'), (20, 'Secret')], default=0),
|
||||||
|
),
|
||||||
|
]
|
||||||
@@ -84,6 +84,17 @@ class User(AbstractUser):
|
|||||||
USERNAME_FIELD = "email"
|
USERNAME_FIELD = "email"
|
||||||
REQUIRED_FIELDS = ["username"]
|
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):
|
def get_name(self):
|
||||||
return self.username
|
return self.username
|
||||||
|
|
||||||
@@ -103,6 +114,25 @@ class User(AbstractUser):
|
|||||||
return self.default_setting
|
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):
|
class APIToken(TimeStampedModel):
|
||||||
"""
|
"""
|
||||||
API authentication token for users.
|
API authentication token for users.
|
||||||
|
|||||||
@@ -1294,6 +1294,9 @@ def package(request, package_uuid):
|
|||||||
)
|
)
|
||||||
users = users.filter(is_active=True).exclude(username="anonymous")
|
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)
|
group_perms = GroupPackagePermission.objects.filter(package=current_package)
|
||||||
|
|
||||||
if current_package.classification_level == current_package.Classification.SECRET:
|
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]):
|
elif any([grantee_url, read, write, owner]):
|
||||||
if "user" in grantee_url:
|
if "user" in grantee_url:
|
||||||
grantee = UserManager.get_user_lp(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:
|
else:
|
||||||
grantee = GroupManager.get_group_lp(grantee_url)
|
grantee = GroupManager.get_group_lp(grantee_url)
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user