This commit is contained in:
lebaudantoine
2025-10-02 18:28:35 +02:00
parent b64e5a0084
commit fe967bdb2d
6 changed files with 101 additions and 9 deletions
+3
View File
@@ -62,8 +62,11 @@ SCREEN_RECORDING_BASE_URL=http://localhost:3000/recordings
# Telephony
ROOM_TELEPHONY_ENABLED=True
ROOM_TELEPHONY_DEFAULT_COUNTRY: 'FR'
ROOM_TELEPHONY_PHONE_NUMBER: '+33901020304'
FRONTEND_USE_FRENCH_GOV_FOOTER=False
FRONTEND_USE_PROCONNECT_BUTTON=False
INTEGRATIONS_JWT_SECRET_KEY=devKey
INTEGRATIONS_APP_BASE_URL: http://localhost:3000
+10 -2
View File
@@ -155,6 +155,13 @@ class RecordingAdmin(admin.ModelAdmin):
return str(owners[0].user)
class ServiceAccountDomainInline(admin.TabularInline):
"""Wip."""
model = models.ServiceAccountDomain
extra = 0
class ServiceAccountAdminForm(forms.ModelForm):
"""Wip."""
@@ -179,14 +186,15 @@ class ServiceAccountAdmin(admin.ModelAdmin):
list_display = ("id", "name", "get_scopes_display")
fields = ["name", "id", "created_at", "updated_at", "scopes"]
readonly_fields = ["id", "created_at", "updated_at"]
inlines = [ServiceAccountDomainInline]
def get_scopes_display(self, obj):
"""Display scopes in list view."""
if obj.scopes:
return ", ".join(obj.scopes)
return "No scopes"
return _("No scopes")
get_scopes_display.short_description = "Scopes"
get_scopes_display.short_description = _("Scopes")
@admin.register(models.ServiceAccountAPIKey)
+13 -3
View File
@@ -49,11 +49,21 @@ class IntegrationViewSet(viewsets.GenericViewSet):
api_key = models.ServiceAccountAPIKey.objects.get_from_key(key)
service_account = api_key.service_account
# todo - extract all this logic in a service
# todo - check if email is allowed by the regex
email = serializer.validated_data["email"]
if not service_account.can_impersonate_email(email):
logger.warning(
"Service account %s denied impersonation of %s",
service_account.id,
email,
)
return drf_response.Response(
{"error": "Access denied"},
status=drf_status.HTTP_403_FORBIDDEN,
)
try:
user = models.User.objects.get(email=serializer.validated_data["email"])
user = models.User.objects.get(email=email)
except models.User.DoesNotExist as e:
# todo - create unknown user
raise drf_exceptions.NotFound(
@@ -1,7 +1,10 @@
# Generated by Django 5.2.6 on 2025-09-28 15:41
# Generated by Django 5.2.6 on 2025-09-28 19:40
import django.contrib.postgres.fields
import django.core.validators
import django.db.models.deletion
import uuid
from django.conf import settings
from django.db import migrations, models
@@ -20,13 +23,13 @@ class Migration(migrations.Migration):
('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated on')),
('name', models.CharField(help_text='Descriptive name for this service account.', max_length=255, verbose_name='Service account name')),
('active', models.BooleanField(default=True)),
('scopes', django.contrib.postgres.fields.ArrayField(base_field=models.CharField(choices=[('rooms:create', 'Create rooms'), ('rooms:list', 'List rooms'), ('rooms:retrieve', 'Retrieve room details'), ('rooms:update', 'Update rooms'), ('rooms:delete', 'Delete rooms')], max_length=50), default=list, size=None)),
('scopes', django.contrib.postgres.fields.ArrayField(base_field=models.CharField(choices=[('rooms:create', 'Create rooms'), ('rooms:list', 'List rooms'), ('rooms:retrieve', 'Retrieve room details'), ('rooms:update', 'Update rooms'), ('rooms:delete', 'Delete rooms')], max_length=50), blank=True, default=list, null=True, size=None)),
],
options={
'verbose_name': 'Service account',
'verbose_name_plural': 'Service accounts',
'db_table': 'meet_service_account',
'abstract': False,
'ordering': ('-created_at',),
},
),
migrations.CreateModel(
@@ -48,4 +51,21 @@ class Migration(migrations.Migration):
'abstract': False,
},
),
migrations.CreateModel(
name='ServiceAccountDomain',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created on')),
('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated on')),
('domain', models.CharField(help_text='Email domain that can be impersonated.', max_length=253, validators=[django.core.validators.DomainNameValidator(accept_idna=False, message='Enter a valid domain')], verbose_name='Domain')),
('service_account', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='allowed_domains', to='core.serviceaccount')),
],
options={
'verbose_name': 'Service account domain',
'verbose_name_plural': 'Service account domains',
'db_table': 'meet_service_account_domain',
'ordering': ('domain',),
'unique_together': {('service_account', 'domain')},
},
),
]
+48
View File
@@ -744,6 +744,8 @@ class ServiceAccount(BaseModel):
scopes = ArrayField(
models.CharField(max_length=50, choices=ServiceAccountScope.choices),
default=list,
blank=True,
null=True,
)
class Meta:
@@ -755,6 +757,52 @@ class ServiceAccount(BaseModel):
def __str__(self):
return f"{self.name!s}"
def can_impersonate_email(self, email):
"""Check if this service account can impersonate the given email."""
if not self.allowed_domains.exists():
return True # No domain restrictions
domain = email.split("@")[-1]
return self.allowed_domains.filter(domain__iexact=domain).exists()
class ServiceAccountDomain(BaseModel):
"""Domain allowed for service account impersonation."""
domain = models.CharField(
max_length=253, # Max domain length per RFC
validators=[
validators.DomainNameValidator(
accept_idna=False,
message=_("Enter a valid domain"),
)
],
verbose_name=_("Domain"),
help_text=_("Email domain that can be impersonated."),
)
service_account = models.ForeignKey(
"ServiceAccount",
on_delete=models.CASCADE,
related_name="allowed_domains",
)
class Meta:
db_table = "meet_service_account_domain"
ordering = ("domain",)
verbose_name = _("Service account domain")
verbose_name_plural = _("Service account domains")
unique_together = [("service_account", "domain")]
def __str__(self):
return self.domain
def save(self, *args, **kwargs):
# Normalize domain to lowercase
self.domain = self.domain.lower().strip()
super().save(*args, **kwargs)
class ServiceAccountAPIKey(AbstractAPIKey):
"""Wip."""
+4 -1
View File
@@ -33,7 +33,10 @@ def generate_slug():
"""Wip."""
sizes = [3, 4, 3]
parts = ["".join(secrets.choices(string.ascii_lowercase, k=size)) for size in sizes]
parts = [
"".join(secrets.choice(string.ascii_lowercase) for _ in range(size))
for size in sizes
]
return "-".join(parts)