ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-24 13:01:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(migrations): remove saml migrations (#8167)

Browse Source 
Co-authored-by: Víctor Fernández Poyatos <victor@prowler.com>
This commit is contained in:
Adrián Jesús Peña Rodríguez
2025-07-02 17:23:08 +02:00
committed by GitHub
parent ece74e15fd
commit d6f26df2e8
11 changed files with 720 additions and 919 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/src/backend/api/adapters.py
+90 -99
View File
@@ -3,14 +3,7 @@ from django.db import transaction
from api.db_router import MainRouter
from api.db_utils import rls_transaction
from api.models import (
Membership,
Role,
SAMLConfiguration,
Tenant,
User,
UserRoleRelationship,
)
from api.models import Membership, Role, Tenant, User, UserRoleRelationship
class ProwlerSocialAccountAdapter(DefaultSocialAccountAdapter):
@@ -24,8 +17,8 @@ class ProwlerSocialAccountAdapter(DefaultSocialAccountAdapter):
def pre_social_login(self, request, sociallogin):
# Link existing accounts with the same email address
email = sociallogin.account.extra_data.get("email")
if sociallogin.account.provider == "saml":
email = sociallogin.user.email
# if sociallogin.account.provider == "saml":
# email = sociallogin.user.email
if email:
existing_user = self.get_user_by_email(email)
if existing_user:
@@ -38,100 +31,98 @@ class ProwlerSocialAccountAdapter(DefaultSocialAccountAdapter):
"""
with transaction.atomic(using=MainRouter.admin_db):
user = super().save_user(request, sociallogin, form)
provider = sociallogin.account.provider
extra = sociallogin.account.extra_data
if provider == "saml":
# Handle SAML-specific logic
user.first_name = (
extra.get("firstName", [""])[0] if extra.get("firstName") else ""
)
user.last_name = (
extra.get("lastName", [""])[0] if extra.get("lastName") else ""
)
user.company_name = (
extra.get("organization", [""])[0]
if extra.get("organization")
else ""
)
user.name = f"{user.first_name} {user.last_name}".strip()
if user.name == "":
user.name = "N/A"
# if provider == "saml":
# # Handle SAML-specific logic
# user.first_name = (
# extra.get("firstName", [""])[0] if extra.get("firstName") else ""
# )
# user.last_name = (
# extra.get("lastName", [""])[0] if extra.get("lastName") else ""
# )
# user.company_name = (
# extra.get("organization", [""])[0]
# if extra.get("organization")
# else ""
# )
# user.name = f"{user.first_name} {user.last_name}".strip()
# if user.name == "":
# user.name = "N/A"
# user.save(using=MainRouter.admin_db)
# email_domain = user.email.split("@")[-1]
# tenant = (
# SAMLConfiguration.objects.using(MainRouter.admin_db)
# .get(email_domain=email_domain)
# .tenant
# )
# with rls_transaction(str(tenant.id)):
# role_name = (
# extra.get("userType", ["saml_default_role"])[0].strip()
# if extra.get("userType")
# else "saml_default_role"
# )
# try:
# role = Role.objects.using(MainRouter.admin_db).get(
# name=role_name, tenant_id=tenant.id
# )
# except Role.DoesNotExist:
# role = Role.objects.using(MainRouter.admin_db).create(
# name=role_name,
# tenant_id=tenant.id,
# manage_users=False,
# manage_account=False,
# manage_billing=False,
# manage_providers=False,
# manage_integrations=False,
# manage_scans=False,
# unlimited_visibility=False,
# )
# Membership.objects.using(MainRouter.admin_db).create(
# user=user,
# tenant=tenant,
# role=Membership.RoleChoices.MEMBER,
# )
# UserRoleRelationship.objects.using(MainRouter.admin_db).create(
# user=user,
# role=role,
# tenant_id=tenant.id,
# )
# Handle other providers (e.g., GitHub, Google)
user.save(using=MainRouter.admin_db)
social_account_name = extra.get("name")
if social_account_name:
user.name = social_account_name
user.save(using=MainRouter.admin_db)
email_domain = user.email.split("@")[-1]
tenant = (
SAMLConfiguration.objects.using(MainRouter.admin_db)
.get(email_domain=email_domain)
.tenant
tenant = Tenant.objects.using(MainRouter.admin_db).create(
name=f"{user.email.split('@')[0]} default tenant"
)
with rls_transaction(str(tenant.id)):
Membership.objects.using(MainRouter.admin_db).create(
user=user, tenant=tenant, role=Membership.RoleChoices.OWNER
)
with rls_transaction(str(tenant.id)):
role_name = (
extra.get("userType", ["saml_default_role"])[0].strip()
if extra.get("userType")
else "saml_default_role"
)
try:
role = Role.objects.using(MainRouter.admin_db).get(
name=role_name, tenant_id=tenant.id
)
except Role.DoesNotExist:
role = Role.objects.using(MainRouter.admin_db).create(
name=role_name,
tenant_id=tenant.id,
manage_users=False,
manage_account=False,
manage_billing=False,
manage_providers=False,
manage_integrations=False,
manage_scans=False,
unlimited_visibility=False,
)
Membership.objects.using(MainRouter.admin_db).create(
user=user,
tenant=tenant,
role=Membership.RoleChoices.MEMBER,
)
UserRoleRelationship.objects.using(MainRouter.admin_db).create(
user=user,
role=role,
tenant_id=tenant.id,
)
else:
# Handle other providers (e.g., GitHub, Google)
user.save(using=MainRouter.admin_db)
social_account_name = extra.get("name")
if social_account_name:
user.name = social_account_name
user.save(using=MainRouter.admin_db)
tenant = Tenant.objects.using(MainRouter.admin_db).create(
name=f"{user.email.split('@')[0]} default tenant"
role = Role.objects.using(MainRouter.admin_db).create(
name="admin",
tenant_id=tenant.id,
manage_users=True,
manage_account=True,
manage_billing=True,
manage_providers=True,
manage_integrations=True,
manage_scans=True,
unlimited_visibility=True,
)
UserRoleRelationship.objects.using(MainRouter.admin_db).create(
user=user,
role=role,
tenant_id=tenant.id,
)
with rls_transaction(str(tenant.id)):
Membership.objects.using(MainRouter.admin_db).create(
user=user, tenant=tenant, role=Membership.RoleChoices.OWNER
)
role = Role.objects.using(MainRouter.admin_db).create(
name="admin",
tenant_id=tenant.id,
manage_users=True,
manage_account=True,
manage_billing=True,
manage_providers=True,
manage_integrations=True,
manage_scans=True,
unlimited_visibility=True,
)
UserRoleRelationship.objects.using(MainRouter.admin_db).create(
user=user,
role=role,
tenant_id=tenant.id,
)
return user
api/src/backend/api/migrations/0031_lighthouseconfiguration.py → api/src/backend/api/migrations/0030_lighthouseconfiguration.py
+1 -1
View File
@@ -11,7 +11,7 @@ import api.rls
class Migration(migrations.Migration):
dependencies = [
("api", "0030_samlconfigurations"),
("api", "0029_findings_check_index_parent"),
]
operations = [
api/src/backend/api/migrations/0030_samlconfigurations.py
-120
View File
@@ -1,120 +0,0 @@
# Generated by Django 5.1.8 on 2025-05-15 09:54
import uuid
import django.db.models.deletion
from django.db import migrations, models
import api.rls
class Migration(migrations.Migration):
dependencies = [
("api", "0029_findings_check_index_parent"),
]
operations = [
migrations.CreateModel(
name="SAMLDomainIndex",
fields=[
(
"id",
models.BigAutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name="ID",
),
),
("email_domain", models.CharField(max_length=254, unique=True)),
(
"tenant",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE, to="api.tenant"
),
),
],
options={
"db_table": "saml_domain_index",
},
),
migrations.AddConstraint(
model_name="samldomainindex",
constraint=models.UniqueConstraint(
fields=("email_domain", "tenant"),
name="unique_resources_by_email_domain",
),
),
migrations.AddConstraint(
model_name="samldomainindex",
constraint=api.rls.BaseSecurityConstraint(
name="statements_on_samldomainindex",
statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
),
),
migrations.CreateModel(
name="SAMLConfiguration",
fields=[
(
"id",
models.UUIDField(
default=uuid.uuid4,
editable=False,
primary_key=True,
serialize=False,
),
),
(
"email_domain",
models.CharField(
help_text="Email domain used to identify the tenant, e.g. prowlerdemo.com",
max_length=254,
unique=True,
),
),
(
"metadata_xml",
models.TextField(
help_text="Raw IdP metadata XML to configure SingleSignOnService, certificates, etc."
),
),
("created_at", models.DateTimeField(auto_now_add=True)),
("updated_at", models.DateTimeField(auto_now=True)),
(
"tenant",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE, to="api.tenant"
),
),
],
options={
"db_table": "saml_configurations",
},
),
migrations.AddConstraint(
model_name="samlconfiguration",
constraint=api.rls.RowLevelSecurityConstraint(
"tenant_id",
name="rls_on_samlconfiguration",
statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
),
),
migrations.AddConstraint(
model_name="samlconfiguration",
constraint=models.UniqueConstraint(
fields=("tenant",), name="unique_samlconfig_per_tenant"
),
),
migrations.AlterField(
model_name="integration",
name="integration_type",
field=api.db_utils.IntegrationTypeEnumField(
choices=[
("amazon_s3", "Amazon S3"),
("aws_security_hub", "AWS Security Hub"),
("jira", "JIRA"),
("slack", "Slack"),
]
),
),
]
api/src/backend/api/migrations/0032_scan_disable_on_cascade_periodic_tasks.py → api/src/backend/api/migrations/0031_scan_disable_on_cascade_periodic_tasks.py
+1 -1
View File
@@ -6,7 +6,7 @@ from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("api", "0031_lighthouseconfiguration"),
("api", "0030_lighthouseconfiguration"),
("django_celery_beat", "0019_alter_periodictasks_options"),
]
api/src/backend/api/migrations/0033_samltoken.py
-44
View File
@@ -1,44 +0,0 @@
# Generated by Django 5.1.10 on 2025-06-25 10:06
import uuid
import django.db.models.deletion
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("api", "0032_scan_disable_on_cascade_periodic_tasks"),
]
operations = [
migrations.CreateModel(
name="SAMLToken",
fields=[
(
"id",
models.UUIDField(
default=uuid.uuid4,
editable=False,
primary_key=True,
serialize=False,
),
),
("inserted_at", models.DateTimeField(auto_now_add=True)),
("updated_at", models.DateTimeField(auto_now=True)),
("expires_at", models.DateTimeField(editable=False)),
("token", models.JSONField(unique=True)),
(
"user",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
to=settings.AUTH_USER_MODEL,
),
),
],
options={
"db_table": "saml_tokens",
},
),
]
api/src/backend/api/models.py
+198 -205
View File
@@ -1,21 +1,15 @@
import json
import logging
import re
import xml.etree.ElementTree as ET
from datetime import datetime, timedelta, timezone
from uuid import UUID, uuid4
from allauth.socialaccount.models import SocialApp
from config.custom_logging import BackendLogger
from config.settings.social_login import SOCIALACCOUNT_PROVIDERS
from cryptography.fernet import Fernet, InvalidToken
from django.conf import settings
from django.contrib.auth.models import AbstractBaseUser
from django.contrib.postgres.fields import ArrayField
from django.contrib.postgres.indexes import GinIndex
from django.contrib.postgres.search import SearchVector, SearchVectorField
from django.contrib.sites.models import Site
from django.core.exceptions import ValidationError
from django.core.validators import MinLengthValidator
from django.db import models
from django.db.models import Q
@@ -27,7 +21,6 @@ from psqlextra.models import PostgresPartitionedModel
from psqlextra.types import PostgresPartitioningMethod
from uuid6 import uuid7
from api.db_router import MainRouter
from api.db_utils import (
CustomUserManager,
FindingDeltaEnumField,
@@ -1371,242 +1364,242 @@ class IntegrationProviderRelationship(RowLevelSecurityProtectedModel):
]
class SAMLToken(models.Model):
id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
inserted_at = models.DateTimeField(auto_now_add=True, editable=False)
updated_at = models.DateTimeField(auto_now=True, editable=False)
expires_at = models.DateTimeField(editable=False)
token = models.JSONField(unique=True)
user = models.ForeignKey(User, on_delete=models.CASCADE)
# class SAMLToken(models.Model):
# id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
# inserted_at = models.DateTimeField(auto_now_add=True, editable=False)
# updated_at = models.DateTimeField(auto_now=True, editable=False)
# expires_at = models.DateTimeField(editable=False)
# token = models.JSONField(unique=True)
# user = models.ForeignKey(User, on_delete=models.CASCADE)
class Meta:
db_table = "saml_tokens"
# class Meta:
# db_table = "saml_tokens"
def save(self, *args, **kwargs):
if not self.expires_at:
self.expires_at = datetime.now(timezone.utc) + timedelta(seconds=15)
super().save(*args, **kwargs)
# def save(self, *args, **kwargs):
# if not self.expires_at:
# self.expires_at = datetime.now(timezone.utc) + timedelta(seconds=15)
# super().save(*args, **kwargs)
def is_expired(self) -> bool:
return datetime.now(timezone.utc) >= self.expires_at
# def is_expired(self) -> bool:
# return datetime.now(timezone.utc) >= self.expires_at
class SAMLDomainIndex(models.Model):
"""
Public index of SAML domains. No RLS. Used for fast lookup in SAML login flow.
"""
# class SAMLDomainIndex(models.Model):
# """
# Public index of SAML domains. No RLS. Used for fast lookup in SAML login flow.
# """
email_domain = models.CharField(max_length=254, unique=True)
tenant = models.ForeignKey("Tenant", on_delete=models.CASCADE)
# email_domain = models.CharField(max_length=254, unique=True)
# tenant = models.ForeignKey("Tenant", on_delete=models.CASCADE)
class Meta:
db_table = "saml_domain_index"
# class Meta:
# db_table = "saml_domain_index"
constraints = [
models.UniqueConstraint(
fields=("email_domain", "tenant"),
name="unique_resources_by_email_domain",
),
BaseSecurityConstraint(
name="statements_on_%(class)s",
statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
),
]
# constraints = [
# models.UniqueConstraint(
# fields=("email_domain", "tenant"),
# name="unique_resources_by_email_domain",
# ),
# BaseSecurityConstraint(
# name="statements_on_%(class)s",
# statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
# ),
# ]
class SAMLConfiguration(RowLevelSecurityProtectedModel):
"""
Stores per-tenant SAML settings, including email domain and IdP metadata.
Automatically syncs to a SocialApp instance on save.
# class SAMLConfiguration(RowLevelSecurityProtectedModel):
# """
# Stores per-tenant SAML settings, including email domain and IdP metadata.
# Automatically syncs to a SocialApp instance on save.
Note:
This model exists to provide a tenant-aware abstraction over SAML configuration.
It supports row-level security, custom validation, and metadata parsing, enabling
Prowler to expose a clean API and admin interface for managing SAML integrations.
# Note:
# This model exists to provide a tenant-aware abstraction over SAML configuration.
# It supports row-level security, custom validation, and metadata parsing, enabling
# Prowler to expose a clean API and admin interface for managing SAML integrations.
Although Django Allauth uses the SocialApp model to store provider configuration,
it is not designed for multi-tenant use. SocialApp lacks support for tenant scoping,
email domain mapping, and structured metadata handling.
# Although Django Allauth uses the SocialApp model to store provider configuration,
# it is not designed for multi-tenant use. SocialApp lacks support for tenant scoping,
# email domain mapping, and structured metadata handling.
By managing SAMLConfiguration separately, we ensure:
- Strong isolation between tenants via RLS.
- Ownership of raw IdP metadata and its validation.
- An explicit link between SAML config and business-level identifiers (e.g. email domain).
- Programmatic transformation into the SocialApp format used by Allauth.
# By managing SAMLConfiguration separately, we ensure:
# - Strong isolation between tenants via RLS.
# - Ownership of raw IdP metadata and its validation.
# - An explicit link between SAML config and business-level identifiers (e.g. email domain).
# - Programmatic transformation into the SocialApp format used by Allauth.
In short, this model acts as a secure and user-friendly layer over Allauth's lower-level primitives.
"""
# In short, this model acts as a secure and user-friendly layer over Allauth's lower-level primitives.
# """
id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
email_domain = models.CharField(
max_length=254,
unique=True,
help_text="Email domain used to identify the tenant, e.g. prowlerdemo.com",
)
metadata_xml = models.TextField(
help_text="Raw IdP metadata XML to configure SingleSignOnService, certificates, etc."
)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
# id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
# email_domain = models.CharField(
# max_length=254,
# unique=True,
# help_text="Email domain used to identify the tenant, e.g. prowlerdemo.com",
# )
# metadata_xml = models.TextField(
# help_text="Raw IdP metadata XML to configure SingleSignOnService, certificates, etc."
# )
# created_at = models.DateTimeField(auto_now_add=True)
# updated_at = models.DateTimeField(auto_now=True)
class JSONAPIMeta:
resource_name = "saml-configurations"
# class JSONAPIMeta:
# resource_name = "saml-configurations"
class Meta:
db_table = "saml_configurations"
# class Meta:
# db_table = "saml_configurations"
constraints = [
RowLevelSecurityConstraint(
field="tenant_id",
name="rls_on_%(class)s",
statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
),
# 1 config per tenant
models.UniqueConstraint(
fields=["tenant"],
name="unique_samlconfig_per_tenant",
),
]
# constraints = [
# RowLevelSecurityConstraint(
# field="tenant_id",
# name="rls_on_%(class)s",
# statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
# ),
# # 1 config per tenant
# models.UniqueConstraint(
# fields=["tenant"],
# name="unique_samlconfig_per_tenant",
# ),
# ]
def clean(self, old_email_domain=None):
# Domain must not contain @
if "@" in self.email_domain:
raise ValidationError({"email_domain": "Domain must not contain @"})
# def clean(self, old_email_domain=None):
# # Domain must not contain @
# if "@" in self.email_domain:
# raise ValidationError({"email_domain": "Domain must not contain @"})
# Enforce at most one config per tenant
qs = SAMLConfiguration.objects.filter(tenant=self.tenant)
# Exclude ourselves in case of update
if self.pk:
qs = qs.exclude(pk=self.pk)
if qs.exists():
raise ValidationError(
{"tenant": "A SAML configuration already exists for this tenant."}
)
# # Enforce at most one config per tenant
# qs = SAMLConfiguration.objects.filter(tenant=self.tenant)
# # Exclude ourselves in case of update
# if self.pk:
# qs = qs.exclude(pk=self.pk)
# if qs.exists():
# raise ValidationError(
# {"tenant": "A SAML configuration already exists for this tenant."}
# )
# The email domain must be unique in the entire system
qs = SAMLConfiguration.objects.using(MainRouter.admin_db).filter(
email_domain__iexact=self.email_domain
)
if qs.exists() and old_email_domain != self.email_domain:
raise ValidationError(
{"tenant": "There is a problem with your email domain."}
)
# # The email domain must be unique in the entire system
# qs = SAMLConfiguration.objects.using(MainRouter.admin_db).filter(
# email_domain__iexact=self.email_domain
# )
# if qs.exists() and old_email_domain != self.email_domain:
# raise ValidationError(
# {"tenant": "There is a problem with your email domain."}
# )
def save(self, *args, **kwargs):
self.email_domain = self.email_domain.strip().lower()
is_create = not SAMLConfiguration.objects.filter(pk=self.pk).exists()
# def save(self, *args, **kwargs):
# self.email_domain = self.email_domain.strip().lower()
# is_create = not SAMLConfiguration.objects.filter(pk=self.pk).exists()
if not is_create:
old = SAMLConfiguration.objects.get(pk=self.pk)
old_email_domain = old.email_domain
old_metadata_xml = old.metadata_xml
else:
old_email_domain = None
old_metadata_xml = None
# if not is_create:
# old = SAMLConfiguration.objects.get(pk=self.pk)
# old_email_domain = old.email_domain
# old_metadata_xml = old.metadata_xml
# else:
# old_email_domain = None
# old_metadata_xml = None
self.clean(old_email_domain)
super().save(*args, **kwargs)
# self.clean(old_email_domain)
# super().save(*args, **kwargs)
if is_create or (
old_email_domain != self.email_domain
or old_metadata_xml != self.metadata_xml
):
self._sync_social_app(old_email_domain)
# if is_create or (
# old_email_domain != self.email_domain
# or old_metadata_xml != self.metadata_xml
# ):
# self._sync_social_app(old_email_domain)
# Sync the public index
if not is_create and old_email_domain and old_email_domain != self.email_domain:
SAMLDomainIndex.objects.filter(email_domain=old_email_domain).delete()
# # Sync the public index
# if not is_create and old_email_domain and old_email_domain != self.email_domain:
# SAMLDomainIndex.objects.filter(email_domain=old_email_domain).delete()
# Create/update the new domain index
SAMLDomainIndex.objects.update_or_create(
email_domain=self.email_domain, defaults={"tenant": self.tenant}
)
# # Create/update the new domain index
# SAMLDomainIndex.objects.update_or_create(
# email_domain=self.email_domain, defaults={"tenant": self.tenant}
# )
def _parse_metadata(self):
"""
Parse the raw IdP metadata XML and extract:
- entity_id
- sso_url
- slo_url (may be None)
- x509cert (required)
"""
ns = {
"md": "urn:oasis:names:tc:SAML:2.0:metadata",
"ds": "http://www.w3.org/2000/09/xmldsig#",
}
try:
root = ET.fromstring(self.metadata_xml)
except ET.ParseError as e:
raise ValidationError({"metadata_xml": f"Invalid XML: {e}"})
# def _parse_metadata(self):
# """
# Parse the raw IdP metadata XML and extract:
# - entity_id
# - sso_url
# - slo_url (may be None)
# - x509cert (required)
# """
# ns = {
# "md": "urn:oasis:names:tc:SAML:2.0:metadata",
# "ds": "http://www.w3.org/2000/09/xmldsig#",
# }
# try:
# root = ET.fromstring(self.metadata_xml)
# except ET.ParseError as e:
# raise ValidationError({"metadata_xml": f"Invalid XML: {e}"})
# Entity ID
entity_id = root.attrib.get("entityID")
# # Entity ID
# entity_id = root.attrib.get("entityID")
# SSO endpoint (must exist)
sso = root.find(".//md:IDPSSODescriptor/md:SingleSignOnService", ns)
if sso is None or "Location" not in sso.attrib:
raise ValidationError(
{"metadata_xml": "Missing SingleSignOnService in metadata."}
)
sso_url = sso.attrib["Location"]
# # SSO endpoint (must exist)
# sso = root.find(".//md:IDPSSODescriptor/md:SingleSignOnService", ns)
# if sso is None or "Location" not in sso.attrib:
# raise ValidationError(
# {"metadata_xml": "Missing SingleSignOnService in metadata."}
# )
# sso_url = sso.attrib["Location"]
# SLO endpoint (optional)
slo = root.find(".//md:IDPSSODescriptor/md:SingleLogoutService", ns)
slo_url = slo.attrib.get("Location") if slo is not None else None
# # SLO endpoint (optional)
# slo = root.find(".//md:IDPSSODescriptor/md:SingleLogoutService", ns)
# slo_url = slo.attrib.get("Location") if slo is not None else None
# X.509 certificate (required)
cert = root.find(
'.//md:KeyDescriptor[@use="signing"]/ds:KeyInfo/ds:X509Data/ds:X509Certificate',
ns,
)
if cert is None or not cert.text or not cert.text.strip():
raise ValidationError(
{
"metadata_xml": 'Metadata must include a <ds:X509Certificate> under <KeyDescriptor use="signing">.'
}
)
x509cert = cert.text.strip()
# # X.509 certificate (required)
# cert = root.find(
# './/md:KeyDescriptor[@use="signing"]/ds:KeyInfo/ds:X509Data/ds:X509Certificate',
# ns,
# )
# if cert is None or not cert.text or not cert.text.strip():
# raise ValidationError(
# {
# "metadata_xml": 'Metadata must include a <ds:X509Certificate> under <KeyDescriptor use="signing">.'
# }
# )
# x509cert = cert.text.strip()
return {
"entity_id": entity_id,
"sso_url": sso_url,
"slo_url": slo_url,
"x509cert": x509cert,
}
# return {
# "entity_id": entity_id,
# "sso_url": sso_url,
# "slo_url": slo_url,
# "x509cert": x509cert,
# }
def _sync_social_app(self, previous_email_domain=None):
"""
Create or update the corresponding SocialApp based on email_domain.
If the domain changed, update the matching SocialApp.
"""
idp_settings = self._parse_metadata()
settings_dict = SOCIALACCOUNT_PROVIDERS["saml"].copy()
settings_dict["idp"] = idp_settings
# def _sync_social_app(self, previous_email_domain=None):
# """
# Create or update the corresponding SocialApp based on email_domain.
# If the domain changed, update the matching SocialApp.
# """
# idp_settings = self._parse_metadata()
# settings_dict = SOCIALACCOUNT_PROVIDERS["saml"].copy()
# settings_dict["idp"] = idp_settings
current_site = Site.objects.get(id=settings.SITE_ID)
# current_site = Site.objects.get(id=settings.SITE_ID)
social_app_qs = SocialApp.objects.filter(
provider="saml", client_id=previous_email_domain or self.email_domain
)
# social_app_qs = SocialApp.objects.filter(
# provider="saml", client_id=previous_email_domain or self.email_domain
# )
client_id = self.email_domain[:191]
name = f"SAML-{self.email_domain}"[:40]
# client_id = self.email_domain[:191]
# name = f"SAML-{self.email_domain}"[:40]
if social_app_qs.exists():
social_app = social_app_qs.first()
social_app.client_id = client_id
social_app.name = name
social_app.settings = settings_dict
social_app.save()
social_app.sites.set([current_site])
else:
social_app = SocialApp.objects.create(
provider="saml",
client_id=client_id,
name=name,
settings=settings_dict,
)
social_app.sites.set([current_site])
# if social_app_qs.exists():
# social_app = social_app_qs.first()
# social_app.client_id = client_id
# social_app.name = name
# social_app.settings = settings_dict
# social_app.save()
# social_app.sites.set([current_site])
# else:
# social_app = SocialApp.objects.create(
# provider="saml",
# client_id=client_id,
# name=name,
# settings=settings_dict,
# )
# social_app.sites.set([current_site])
class ResourceScanSummary(RowLevelSecurityProtectedModel):
api/src/backend/api/tests/test_adapters.py
+45 -47
View File
@@ -5,8 +5,6 @@ from allauth.socialaccount.models import SocialLogin
from django.contrib.auth import get_user_model
from api.adapters import ProwlerSocialAccountAdapter
from api.db_router import MainRouter
from api.models import Membership, SAMLConfiguration, Tenant
User = get_user_model()
@@ -22,24 +20,24 @@ class TestProwlerSocialAccountAdapter:
adapter = ProwlerSocialAccountAdapter()
assert adapter.get_user_by_email("notfound@example.com") is None
def test_pre_social_login_links_existing_user(self, create_test_user, rf):
adapter = ProwlerSocialAccountAdapter()
# def test_pre_social_login_links_existing_user(self, create_test_user, rf):
# adapter = ProwlerSocialAccountAdapter()
sociallogin = MagicMock(spec=SocialLogin)
sociallogin.account = MagicMock()
sociallogin.account.provider = "saml"
sociallogin.account.extra_data = {}
sociallogin.user = create_test_user
sociallogin.connect = MagicMock()
# sociallogin = MagicMock(spec=SocialLogin)
# sociallogin.account = MagicMock()
# sociallogin.account.provider = "saml"
# sociallogin.account.extra_data = {}
# sociallogin.user = create_test_user
# sociallogin.connect = MagicMock()
adapter.pre_social_login(rf.get("/"), sociallogin)
# adapter.pre_social_login(rf.get("/"), sociallogin)
call_args = sociallogin.connect.call_args
assert call_args is not None
# call_args = sociallogin.connect.call_args
# assert call_args is not None
called_request, called_user = call_args[0]
assert called_request.path == "/"
assert called_user.email == create_test_user.email
# called_request, called_user = call_args[0]
# assert called_request.path == "/"
# assert called_user.email == create_test_user.email
def test_pre_social_login_no_link_if_email_missing(self, rf):
adapter = ProwlerSocialAccountAdapter()
@@ -54,37 +52,37 @@ class TestProwlerSocialAccountAdapter:
sociallogin.connect.assert_not_called()
def test_save_user_saml_flow(
self,
rf,
saml_setup,
saml_sociallogin,
):
adapter = ProwlerSocialAccountAdapter()
request = rf.get("/")
saml_sociallogin.user.email = saml_setup["email"]
saml_sociallogin.account.extra_data = {
"firstName": [],
"lastName": [],
"organization": [],
"userType": [],
}
# def test_save_user_saml_flow(
# self,
# rf,
# saml_setup,
# saml_sociallogin,
# ):
# adapter = ProwlerSocialAccountAdapter()
# request = rf.get("/")
# saml_sociallogin.user.email = saml_setup["email"]
# saml_sociallogin.account.extra_data = {
# "firstName": [],
# "lastName": [],
# "organization": [],
# "userType": [],
# }
tenant = Tenant.objects.using(MainRouter.admin_db).get(
id=saml_setup["tenant_id"]
)
saml_config = SAMLConfiguration.objects.using(MainRouter.admin_db).get(
tenant=tenant
)
assert saml_config.email_domain == saml_setup["domain"]
# tenant = Tenant.objects.using(MainRouter.admin_db).get(
# id=saml_setup["tenant_id"]
# )
# saml_config = SAMLConfiguration.objects.using(MainRouter.admin_db).get(
# tenant=tenant
# )
# assert saml_config.email_domain == saml_setup["domain"]
user = adapter.save_user(request, saml_sociallogin)
# user = adapter.save_user(request, saml_sociallogin)
assert user.name == "N/A"
assert user.company_name == ""
assert user.email == saml_setup["email"]
assert (
Membership.objects.using(MainRouter.admin_db)
.filter(user=user, tenant=tenant)
.exists()
)
# assert user.name == "N/A"
# assert user.company_name == ""
# assert user.email == saml_setup["email"]
# assert (
# Membership.objects.using(MainRouter.admin_db)
# .filter(user=user, tenant=tenant)
# .exists()
# )
api/src/backend/api/tests/test_models.py
+121 -124
View File
@@ -1,9 +1,6 @@
import pytest
from allauth.socialaccount.models import SocialApp
from django.core.exceptions import ValidationError
from api.db_router import MainRouter
from api.models import Resource, ResourceTag, SAMLConfiguration, Tenant
from api.models import Resource, ResourceTag
@pytest.mark.django_db
@@ -125,147 +122,147 @@ class TestResourceModel:
# assert Finding.objects.filter(uid=long_uid).exists()
@pytest.mark.django_db
class TestSAMLConfigurationModel:
VALID_METADATA = """<?xml version='1.0' encoding='UTF-8'?>
<md:EntityDescriptor entityID='TEST' xmlns:md='urn:oasis:names:tc:SAML:2.0:metadata'>
<md:IDPSSODescriptor WantAuthnRequestsSigned='false' protocolSupportEnumeration='urn:oasis:names:tc:SAML:2.0:protocol'>
<md:KeyDescriptor use='signing'>
<ds:KeyInfo xmlns:ds='http://www.w3.org/2000/09/xmldsig#'>
<ds:X509Data>
<ds:X509Certificate>FAKECERTDATA</ds:X509Certificate>
</ds:X509Data>
</ds:KeyInfo>
</md:KeyDescriptor>
<md:SingleSignOnService Binding='urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST' Location='https://idp.test/sso'/>
</md:IDPSSODescriptor>
</md:EntityDescriptor>
"""
# @pytest.mark.django_db
# class TestSAMLConfigurationModel:
# VALID_METADATA = """<?xml version='1.0' encoding='UTF-8'?>
# <md:EntityDescriptor entityID='TEST' xmlns:md='urn:oasis:names:tc:SAML:2.0:metadata'>
# <md:IDPSSODescriptor WantAuthnRequestsSigned='false' protocolSupportEnumeration='urn:oasis:names:tc:SAML:2.0:protocol'>
# <md:KeyDescriptor use='signing'>
# <ds:KeyInfo xmlns:ds='http://www.w3.org/2000/09/xmldsig#'>
# <ds:X509Data>
# <ds:X509Certificate>FAKECERTDATA</ds:X509Certificate>
# </ds:X509Data>
# </ds:KeyInfo>
# </md:KeyDescriptor>
# <md:SingleSignOnService Binding='urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST' Location='https://idp.test/sso'/>
# </md:IDPSSODescriptor>
# </md:EntityDescriptor>
# """
def test_creates_valid_configuration(self):
tenant = Tenant.objects.using(MainRouter.admin_db).create(name="Tenant A")
config = SAMLConfiguration.objects.using(MainRouter.admin_db).create(
email_domain="ssoexample.com",
metadata_xml=TestSAMLConfigurationModel.VALID_METADATA,
tenant=tenant,
)
# def test_creates_valid_configuration(self):
# tenant = Tenant.objects.using(MainRouter.admin_db).create(name="Tenant A")
# config = SAMLConfiguration.objects.using(MainRouter.admin_db).create(
# email_domain="ssoexample.com",
# metadata_xml=TestSAMLConfigurationModel.VALID_METADATA,
# tenant=tenant,
# )
assert config.email_domain == "ssoexample.com"
assert SocialApp.objects.filter(client_id="ssoexample.com").exists()
# assert config.email_domain == "ssoexample.com"
# assert SocialApp.objects.filter(client_id="ssoexample.com").exists()
def test_email_domain_with_at_symbol_fails(self):
tenant = Tenant.objects.using(MainRouter.admin_db).create(name="Tenant B")
config = SAMLConfiguration(
email_domain="invalid@domain.com",
metadata_xml=TestSAMLConfigurationModel.VALID_METADATA,
tenant=tenant,
)
# def test_email_domain_with_at_symbol_fails(self):
# tenant = Tenant.objects.using(MainRouter.admin_db).create(name="Tenant B")
# config = SAMLConfiguration(
# email_domain="invalid@domain.com",
# metadata_xml=TestSAMLConfigurationModel.VALID_METADATA,
# tenant=tenant,
# )
with pytest.raises(ValidationError) as exc_info:
config.clean()
# with pytest.raises(ValidationError) as exc_info:
# config.clean()
errors = exc_info.value.message_dict
assert "email_domain" in errors
assert "Domain must not contain @" in errors["email_domain"][0]
# errors = exc_info.value.message_dict
# assert "email_domain" in errors
# assert "Domain must not contain @" in errors["email_domain"][0]
def test_duplicate_email_domain_fails(self):
tenant1 = Tenant.objects.using(MainRouter.admin_db).create(name="Tenant C1")
tenant2 = Tenant.objects.using(MainRouter.admin_db).create(name="Tenant C2")
# def test_duplicate_email_domain_fails(self):
# tenant1 = Tenant.objects.using(MainRouter.admin_db).create(name="Tenant C1")
# tenant2 = Tenant.objects.using(MainRouter.admin_db).create(name="Tenant C2")
SAMLConfiguration.objects.using(MainRouter.admin_db).create(
email_domain="duplicate.com",
metadata_xml=TestSAMLConfigurationModel.VALID_METADATA,
tenant=tenant1,
)
# SAMLConfiguration.objects.using(MainRouter.admin_db).create(
# email_domain="duplicate.com",
# metadata_xml=TestSAMLConfigurationModel.VALID_METADATA,
# tenant=tenant1,
# )
config = SAMLConfiguration(
email_domain="duplicate.com",
metadata_xml=TestSAMLConfigurationModel.VALID_METADATA,
tenant=tenant2,
)
# config = SAMLConfiguration(
# email_domain="duplicate.com",
# metadata_xml=TestSAMLConfigurationModel.VALID_METADATA,
# tenant=tenant2,
# )
with pytest.raises(ValidationError) as exc_info:
config.clean()
# with pytest.raises(ValidationError) as exc_info:
# config.clean()
errors = exc_info.value.message_dict
assert "tenant" in errors
assert "There is a problem with your email domain." in errors["tenant"][0]
# errors = exc_info.value.message_dict
# assert "tenant" in errors
# assert "There is a problem with your email domain." in errors["tenant"][0]
def test_duplicate_tenant_config_fails(self):
tenant = Tenant.objects.using(MainRouter.admin_db).create(name="Tenant D")
# def test_duplicate_tenant_config_fails(self):
# tenant = Tenant.objects.using(MainRouter.admin_db).create(name="Tenant D")
SAMLConfiguration.objects.using(MainRouter.admin_db).create(
email_domain="unique1.com",
metadata_xml=TestSAMLConfigurationModel.VALID_METADATA,
tenant=tenant,
)
# SAMLConfiguration.objects.using(MainRouter.admin_db).create(
# email_domain="unique1.com",
# metadata_xml=TestSAMLConfigurationModel.VALID_METADATA,
# tenant=tenant,
# )
config = SAMLConfiguration(
email_domain="unique2.com",
metadata_xml=TestSAMLConfigurationModel.VALID_METADATA,
tenant=tenant,
)
# config = SAMLConfiguration(
# email_domain="unique2.com",
# metadata_xml=TestSAMLConfigurationModel.VALID_METADATA,
# tenant=tenant,
# )
with pytest.raises(ValidationError) as exc_info:
config.clean()
# with pytest.raises(ValidationError) as exc_info:
# config.clean()
errors = exc_info.value.message_dict
assert "tenant" in errors
assert (
"A SAML configuration already exists for this tenant."
in errors["tenant"][0]
)
# errors = exc_info.value.message_dict
# assert "tenant" in errors
# assert (
# "A SAML configuration already exists for this tenant."
# in errors["tenant"][0]
# )
def test_invalid_metadata_xml_fails(self):
tenant = Tenant.objects.using(MainRouter.admin_db).create(name="Tenant E")
config = SAMLConfiguration(
email_domain="brokenxml.com",
metadata_xml="<bad<xml>",
tenant=tenant,
)
# def test_invalid_metadata_xml_fails(self):
# tenant = Tenant.objects.using(MainRouter.admin_db).create(name="Tenant E")
# config = SAMLConfiguration(
# email_domain="brokenxml.com",
# metadata_xml="<bad<xml>",
# tenant=tenant,
# )
with pytest.raises(ValidationError) as exc_info:
config._parse_metadata()
# with pytest.raises(ValidationError) as exc_info:
# config._parse_metadata()
errors = exc_info.value.message_dict
assert "metadata_xml" in errors
assert "Invalid XML" in errors["metadata_xml"][0]
assert "not well-formed" in errors["metadata_xml"][0]
# errors = exc_info.value.message_dict
# assert "metadata_xml" in errors
# assert "Invalid XML" in errors["metadata_xml"][0]
# assert "not well-formed" in errors["metadata_xml"][0]
def test_metadata_missing_sso_fails(self):
tenant = Tenant.objects.using(MainRouter.admin_db).create(name="Tenant F")
xml = """<md:EntityDescriptor entityID="x" xmlns:md="urn:oasis:names:tc:SAML:2.0:metadata">
<md:IDPSSODescriptor></md:IDPSSODescriptor>
</md:EntityDescriptor>"""
config = SAMLConfiguration(
email_domain="nosso.com",
metadata_xml=xml,
tenant=tenant,
)
# def test_metadata_missing_sso_fails(self):
# tenant = Tenant.objects.using(MainRouter.admin_db).create(name="Tenant F")
# xml = """<md:EntityDescriptor entityID="x" xmlns:md="urn:oasis:names:tc:SAML:2.0:metadata">
# <md:IDPSSODescriptor></md:IDPSSODescriptor>
# </md:EntityDescriptor>"""
# config = SAMLConfiguration(
# email_domain="nosso.com",
# metadata_xml=xml,
# tenant=tenant,
# )
with pytest.raises(ValidationError) as exc_info:
config._parse_metadata()
# with pytest.raises(ValidationError) as exc_info:
# config._parse_metadata()
errors = exc_info.value.message_dict
assert "metadata_xml" in errors
assert "Missing SingleSignOnService" in errors["metadata_xml"][0]
# errors = exc_info.value.message_dict
# assert "metadata_xml" in errors
# assert "Missing SingleSignOnService" in errors["metadata_xml"][0]
def test_metadata_missing_certificate_fails(self):
tenant = Tenant.objects.using(MainRouter.admin_db).create(name="Tenant G")
xml = """<md:EntityDescriptor entityID="x" xmlns:md="urn:oasis:names:tc:SAML:2.0:metadata">
<md:IDPSSODescriptor>
<md:SingleSignOnService Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect" Location="https://example.com/sso"/>
</md:IDPSSODescriptor>
</md:EntityDescriptor>"""
config = SAMLConfiguration(
email_domain="nocert.com",
metadata_xml=xml,
tenant=tenant,
)
# def test_metadata_missing_certificate_fails(self):
# tenant = Tenant.objects.using(MainRouter.admin_db).create(name="Tenant G")
# xml = """<md:EntityDescriptor entityID="x" xmlns:md="urn:oasis:names:tc:SAML:2.0:metadata">
# <md:IDPSSODescriptor>
# <md:SingleSignOnService Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect" Location="https://example.com/sso"/>
# </md:IDPSSODescriptor>
# </md:EntityDescriptor>"""
# config = SAMLConfiguration(
# email_domain="nocert.com",
# metadata_xml=xml,
# tenant=tenant,
# )
with pytest.raises(ValidationError) as exc_info:
config._parse_metadata()
# with pytest.raises(ValidationError) as exc_info:
# config._parse_metadata()
errors = exc_info.value.message_dict
assert "metadata_xml" in errors
assert "X509Certificate" in errors["metadata_xml"][0]
# errors = exc_info.value.message_dict
# assert "metadata_xml" in errors
# assert "X509Certificate" in errors["metadata_xml"][0]
api/src/backend/api/v1/serializers.py
+12 -13
View File
@@ -29,7 +29,6 @@ from api.models import (
ResourceTag,
Role,
RoleProviderGroupRelationship,
SAMLConfiguration,
Scan,
StateChoices,
StatusChoices,
@@ -2068,23 +2067,23 @@ class IntegrationUpdateSerializer(BaseWriteIntegrationSerializer):
# SSO
class SamlInitiateSerializer(serializers.Serializer):
email_domain = serializers.CharField()
# class SamlInitiateSerializer(serializers.Serializer):
# email_domain = serializers.CharField()
class JSONAPIMeta:
resource_name = "saml-initiate"
# class JSONAPIMeta:
# resource_name = "saml-initiate"
class SamlMetadataSerializer(serializers.Serializer):
class JSONAPIMeta:
resource_name = "saml-meta"
# class SamlMetadataSerializer(serializers.Serializer):
# class JSONAPIMeta:
# resource_name = "saml-meta"
class SAMLConfigurationSerializer(RLSSerializer):
class Meta:
model = SAMLConfiguration
fields = ["id", "email_domain", "metadata_xml", "created_at", "updated_at"]
read_only_fields = ["id", "created_at", "updated_at"]
# class SAMLConfigurationSerializer(RLSSerializer):
# class Meta:
# model = SAMLConfiguration
# fields = ["id", "email_domain", "metadata_xml", "created_at", "updated_at"]
# read_only_fields = ["id", "created_at", "updated_at"]
class LighthouseConfigSerializer(RLSSerializer):
api/src/backend/api/v1/views.py
+203 -213
View File
@@ -1,13 +1,10 @@
import glob
import os
from datetime import datetime, timedelta, timezone
from urllib.parse import urljoin
import sentry_sdk
from allauth.socialaccount.models import SocialAccount, SocialApp
from allauth.socialaccount.providers.github.views import GitHubOAuth2Adapter
from allauth.socialaccount.providers.google.views import GoogleOAuth2Adapter
from allauth.socialaccount.providers.saml.views import FinishACSView, LoginView
from botocore.exceptions import ClientError, NoCredentialsError, ParamValidationError
from celery.result import AsyncResult
from config.env import env
@@ -23,7 +20,6 @@ from django.db import transaction
from django.db.models import Count, Exists, F, OuterRef, Prefetch, Q, Sum
from django.db.models.functions import Coalesce
from django.http import HttpResponse
from django.shortcuts import redirect
from django.urls import reverse
from django.utils.dateparse import parse_date
from django.utils.decorators import method_decorator
@@ -68,7 +64,6 @@ from api.compliance import (
get_compliance_frameworks,
)
from api.db_router import MainRouter
from api.db_utils import rls_transaction
from api.exceptions import TaskFailedException
from api.filters import (
ComplianceOverviewFilter,
@@ -106,9 +101,6 @@ from api.models import (
ResourceScanSummary,
Role,
RoleProviderGroupRelationship,
SAMLConfiguration,
SAMLDomainIndex,
SAMLToken,
Scan,
ScanSummary,
SeverityChoices,
@@ -165,8 +157,6 @@ from api.v1.serializers import (
RoleProviderGroupRelationshipSerializer,
RoleSerializer,
RoleUpdateSerializer,
SAMLConfigurationSerializer,
SamlInitiateSerializer,
ScanComplianceReportSerializer,
ScanCreateSerializer,
ScanReportSerializer,
@@ -403,240 +393,240 @@ class GithubSocialLoginView(SocialLoginView):
return original_response
@extend_schema(exclude=True)
class SAMLTokenValidateView(GenericAPIView):
resource_name = "tokens"
http_method_names = ["post"]
# @extend_schema(exclude=True)
# class SAMLTokenValidateView(GenericAPIView):
# resource_name = "tokens"
# http_method_names = ["post"]
def post(self, request):
token_id = request.query_params.get("id", "invalid")
try:
saml_token = SAMLToken.objects.using(MainRouter.admin_db).get(id=token_id)
except SAMLToken.DoesNotExist:
return Response({"detail": "Invalid token ID."}, status=404)
# def post(self, request):
# token_id = request.query_params.get("id", "invalid")
# try:
# saml_token = SAMLToken.objects.using(MainRouter.admin_db).get(id=token_id)
# except SAMLToken.DoesNotExist:
# return Response({"detail": "Invalid token ID."}, status=404)
if saml_token.is_expired():
return Response({"detail": "Token expired."}, status=400)
# if saml_token.is_expired():
# return Response({"detail": "Token expired."}, status=400)
token_data = saml_token.token
# Currently we don't store the tokens in the database, so we delete the token after use
saml_token.delete()
# token_data = saml_token.token
# # Currently we don't store the tokens in the database, so we delete the token after use
# saml_token.delete()
return Response(token_data, status=200)
# return Response(token_data, status=200)
@extend_schema(exclude=True)
class CustomSAMLLoginView(LoginView):
def dispatch(self, request, *args, **kwargs):
"""
Convert GET requests to POST to bypass allauth's confirmation screen.
# @extend_schema(exclude=True)
# class CustomSAMLLoginView(LoginView):
# def dispatch(self, request, *args, **kwargs):
# """
# Convert GET requests to POST to bypass allauth's confirmation screen.
Why this is necessary:
- django-allauth requires POST for social logins to prevent open redirect attacks
- SAML login links typically use GET requests (e.g., <a href="...">)
- This conversion allows seamless login without user-facing confirmation
# Why this is necessary:
# - django-allauth requires POST for social logins to prevent open redirect attacks
# - SAML login links typically use GET requests (e.g., <a href="...">)
# - This conversion allows seamless login without user-facing confirmation
Security considerations:
1. Preserves CSRF protection: Original POST handling remains intact
2. Avoids global SOCIALACCOUNT_LOGIN_ON_GET=True which would:
- Enable GET logins for ALL providers (security risk)
- Potentially expose open redirect vulnerabilities
3. SAML payloads remain signed/encrypted regardless of HTTP method
4. No sensitive parameters are exposed in URLs (copied to POST body)
# Security considerations:
# 1. Preserves CSRF protection: Original POST handling remains intact
# 2. Avoids global SOCIALACCOUNT_LOGIN_ON_GET=True which would:
# - Enable GET logins for ALL providers (security risk)
# - Potentially expose open redirect vulnerabilities
# 3. SAML payloads remain signed/encrypted regardless of HTTP method
# 4. No sensitive parameters are exposed in URLs (copied to POST body)
This approach maintains security while providing better UX.
"""
if request.method == "GET":
# Convert GET to POST while preserving parameters
request.method = "POST"
# Safe because SAML validates signatures
request.POST = request.GET.copy()
return super().dispatch(request, *args, **kwargs)
# This approach maintains security while providing better UX.
# """
# if request.method == "GET":
# # Convert GET to POST while preserving parameters
# request.method = "POST"
# # Safe because SAML validates signatures
# request.POST = request.GET.copy()
# return super().dispatch(request, *args, **kwargs)
@extend_schema(exclude=True)
class SAMLInitiateAPIView(GenericAPIView):
serializer_class = SamlInitiateSerializer
permission_classes = []
# @extend_schema(exclude=True)
# class SAMLInitiateAPIView(GenericAPIView):
# serializer_class = SamlInitiateSerializer
# permission_classes = []
def post(self, request, *args, **kwargs):
# Validate the input payload and extract the domain
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
email = serializer.validated_data["email_domain"]
domain = email.split("@", 1)[-1].lower()
# def post(self, request, *args, **kwargs):
# # Validate the input payload and extract the domain
# serializer = self.get_serializer(data=request.data)
# serializer.is_valid(raise_exception=True)
# email = serializer.validated_data["email_domain"]
# domain = email.split("@", 1)[-1].lower()
# Retrieve the SAML configuration for the given email domain
try:
check = SAMLDomainIndex.objects.get(email_domain=domain)
with rls_transaction(str(check.tenant_id)):
config = SAMLConfiguration.objects.get(tenant_id=str(check.tenant_id))
except (SAMLDomainIndex.DoesNotExist, SAMLConfiguration.DoesNotExist):
return Response(
{"detail": "Unauthorized domain."}, status=status.HTTP_403_FORBIDDEN
)
# # Retrieve the SAML configuration for the given email domain
# try:
# check = SAMLDomainIndex.objects.get(email_domain=domain)
# with rls_transaction(str(check.tenant_id)):
# config = SAMLConfiguration.objects.get(tenant_id=str(check.tenant_id))
# except (SAMLDomainIndex.DoesNotExist, SAMLConfiguration.DoesNotExist):
# return Response(
# {"detail": "Unauthorized domain."}, status=status.HTTP_403_FORBIDDEN
# )
# Check certificates are not empty (TODO: Validate certificates)
# saml_public_cert = os.getenv("SAML_PUBLIC_CERT", "").strip()
# saml_private_key = os.getenv("SAML_PRIVATE_KEY", "").strip()
# # Check certificates are not empty (TODO: Validate certificates)
# # saml_public_cert = os.getenv("SAML_PUBLIC_CERT", "").strip()
# # saml_private_key = os.getenv("SAML_PRIVATE_KEY", "").strip()
# if not saml_public_cert or not saml_private_key:
# return Response(
# {"detail": "SAML configuration is invalid: missing certificates."},
# status=status.HTTP_403_FORBIDDEN,
# )
# # if not saml_public_cert or not saml_private_key:
# # return Response(
# # {"detail": "SAML configuration is invalid: missing certificates."},
# # status=status.HTTP_403_FORBIDDEN,
# # )
# Build the SAML login URL using the configured API host
api_host = os.getenv("API_BASE_URL")
login_path = reverse(
"saml_login", kwargs={"organization_slug": config.email_domain}
)
login_url = urljoin(api_host, login_path)
# # Build the SAML login URL using the configured API host
# api_host = os.getenv("API_BASE_URL")
# login_path = reverse(
# "saml_login", kwargs={"organization_slug": config.email_domain}
# )
# login_url = urljoin(api_host, login_path)
return redirect(login_url)
# return redirect(login_url)
@extend_schema_view(
list=extend_schema(
tags=["SAML"],
summary="List all SSO configurations",
description="Returns all the SAML-based SSO configurations associated with the current tenant.",
),
retrieve=extend_schema(
tags=["SAML"],
summary="Retrieve SSO configuration details",
description="Returns the details of a specific SAML configuration belonging to the current tenant.",
),
create=extend_schema(
tags=["SAML"],
summary="Create the SSO configuration",
description="Creates a new SAML SSO configuration for the current tenant, including email domain and metadata XML.",
),
partial_update=extend_schema(
tags=["SAML"],
summary="Update the SSO configuration",
description="Partially updates an existing SAML SSO configuration. Supports changes to email domain and metadata XML.",
),
destroy=extend_schema(
tags=["SAML"],
summary="Delete the SSO configuration",
description="Deletes an existing SAML SSO configuration associated with the current tenant.",
),
)
@method_decorator(CACHE_DECORATOR, name="retrieve")
@method_decorator(CACHE_DECORATOR, name="list")
class SAMLConfigurationViewSet(BaseRLSViewSet):
"""
ViewSet for managing SAML SSO configurations per tenant.
# @extend_schema_view(
# list=extend_schema(
# tags=["SAML"],
# summary="List all SSO configurations",
# description="Returns all the SAML-based SSO configurations associated with the current tenant.",
# ),
# retrieve=extend_schema(
# tags=["SAML"],
# summary="Retrieve SSO configuration details",
# description="Returns the details of a specific SAML configuration belonging to the current tenant.",
# ),
# create=extend_schema(
# tags=["SAML"],
# summary="Create the SSO configuration",
# description="Creates a new SAML SSO configuration for the current tenant, including email domain and metadata XML.",
# ),
# partial_update=extend_schema(
# tags=["SAML"],
# summary="Update the SSO configuration",
# description="Partially updates an existing SAML SSO configuration. Supports changes to email domain and metadata XML.",
# ),
# destroy=extend_schema(
# tags=["SAML"],
# summary="Delete the SSO configuration",
# description="Deletes an existing SAML SSO configuration associated with the current tenant.",
# ),
# )
# @method_decorator(CACHE_DECORATOR, name="retrieve")
# @method_decorator(CACHE_DECORATOR, name="list")
# class SAMLConfigurationViewSet(BaseRLSViewSet):
# """
# ViewSet for managing SAML SSO configurations per tenant.
This endpoint allows authorized users to perform CRUD operations on SAMLConfiguration,
which define how a tenant integrates with an external SAML Identity Provider (IdP).
# This endpoint allows authorized users to perform CRUD operations on SAMLConfiguration,
# which define how a tenant integrates with an external SAML Identity Provider (IdP).
Typical use cases include:
- Listing all existing configurations for auditing or UI display.
- Retrieving a single configuration to show setup details.
- Creating or updating a configuration to onboard or modify SAML integration.
- Deleting a configuration when deactivating SAML for a tenant.
"""
# Typical use cases include:
# - Listing all existing configurations for auditing or UI display.
# - Retrieving a single configuration to show setup details.
# - Creating or updating a configuration to onboard or modify SAML integration.
# - Deleting a configuration when deactivating SAML for a tenant.
# """
serializer_class = SAMLConfigurationSerializer
required_permissions = [Permissions.MANAGE_INTEGRATIONS]
queryset = SAMLConfiguration.objects.all()
# serializer_class = SAMLConfigurationSerializer
# required_permissions = [Permissions.MANAGE_INTEGRATIONS]
# queryset = SAMLConfiguration.objects.all()
def get_queryset(self):
# If called during schema generation, return an empty queryset
if getattr(self, "swagger_fake_view", False):
return SAMLConfiguration.objects.none()
return SAMLConfiguration.objects.filter(tenant=self.request.tenant_id)
# def get_queryset(self):
# # If called during schema generation, return an empty queryset
# if getattr(self, "swagger_fake_view", False):
# return SAMLConfiguration.objects.none()
# return SAMLConfiguration.objects.filter(tenant=self.request.tenant_id)
class TenantFinishACSView(FinishACSView):
def dispatch(self, request, organization_slug):
response = super().dispatch(request, organization_slug)
user = getattr(request, "user", None)
if not user or not user.is_authenticated:
return response
# class TenantFinishACSView(FinishACSView):
# def dispatch(self, request, organization_slug):
# response = super().dispatch(request, organization_slug)
# user = getattr(request, "user", None)
# if not user or not user.is_authenticated:
# return response
try:
social_app = SocialApp.objects.get(
provider="saml", client_id=organization_slug
)
social_account = SocialAccount.objects.get(
user=user, provider=social_app.provider
)
except (SocialApp.DoesNotExist, SocialAccount.DoesNotExist):
return response
# try:
# social_app = SocialApp.objects.get(
# provider="saml", client_id=organization_slug
# )
# social_account = SocialAccount.objects.get(
# user=user, provider=social_app.provider
# )
# except (SocialApp.DoesNotExist, SocialAccount.DoesNotExist):
# return response
extra = social_account.extra_data
user.first_name = (
extra.get("firstName", [""])[0] if extra.get("firstName") else ""
)
user.last_name = extra.get("lastName", [""])[0] if extra.get("lastName") else ""
user.company_name = (
extra.get("organization", [""])[0] if extra.get("organization") else ""
)
user.name = f"{user.first_name} {user.last_name}".strip()
if user.name == "":
user.name = "N/A"
user.save()
# extra = social_account.extra_data
# user.first_name = (
# extra.get("firstName", [""])[0] if extra.get("firstName") else ""
# )
# user.last_name = extra.get("lastName", [""])[0] if extra.get("lastName") else ""
# user.company_name = (
# extra.get("organization", [""])[0] if extra.get("organization") else ""
# )
# user.name = f"{user.first_name} {user.last_name}".strip()
# if user.name == "":
# user.name = "N/A"
# user.save()
email_domain = user.email.split("@")[-1]
tenant = (
SAMLConfiguration.objects.using(MainRouter.admin_db)
.get(email_domain=email_domain)
.tenant
)
role_name = (
extra.get("userType", ["saml_default_role"])[0].strip()
if extra.get("userType")
else "saml_default_role"
)
try:
role = Role.objects.using(MainRouter.admin_db).get(
name=role_name, tenant=tenant
)
except Role.DoesNotExist:
role = Role.objects.using(MainRouter.admin_db).create(
name=role_name,
tenant=tenant,
manage_users=False,
manage_account=False,
manage_billing=False,
manage_providers=False,
manage_integrations=False,
manage_scans=False,
unlimited_visibility=False,
)
UserRoleRelationship.objects.using(MainRouter.admin_db).filter(
user=user,
tenant_id=tenant.id,
).delete()
UserRoleRelationship.objects.using(MainRouter.admin_db).create(
user=user,
role=role,
tenant_id=tenant.id,
)
membership, _ = Membership.objects.using(MainRouter.admin_db).get_or_create(
user=user,
tenant=tenant,
defaults={
"user": user,
"tenant": tenant,
"role": Membership.RoleChoices.MEMBER,
},
)
# email_domain = user.email.split("@")[-1]
# tenant = (
# SAMLConfiguration.objects.using(MainRouter.admin_db)
# .get(email_domain=email_domain)
# .tenant
# )
# role_name = (
# extra.get("userType", ["saml_default_role"])[0].strip()
# if extra.get("userType")
# else "saml_default_role"
# )
# try:
# role = Role.objects.using(MainRouter.admin_db).get(
# name=role_name, tenant=tenant
# )
# except Role.DoesNotExist:
# role = Role.objects.using(MainRouter.admin_db).create(
# name=role_name,
# tenant=tenant,
# manage_users=False,
# manage_account=False,
# manage_billing=False,
# manage_providers=False,
# manage_integrations=False,
# manage_scans=False,
# unlimited_visibility=False,
# )
# UserRoleRelationship.objects.using(MainRouter.admin_db).filter(
# user=user,
# tenant_id=tenant.id,
# ).delete()
# UserRoleRelationship.objects.using(MainRouter.admin_db).create(
# user=user,
# role=role,
# tenant_id=tenant.id,
# )
# membership, _ = Membership.objects.using(MainRouter.admin_db).get_or_create(
# user=user,
# tenant=tenant,
# defaults={
# "user": user,
# "tenant": tenant,
# "role": Membership.RoleChoices.MEMBER,
# },
# )
serializer = TokenSocialLoginSerializer(data={"email": user.email})
serializer.is_valid(raise_exception=True)
# serializer = TokenSocialLoginSerializer(data={"email": user.email})
# serializer.is_valid(raise_exception=True)
token_data = serializer.validated_data
saml_token = SAMLToken.objects.using(MainRouter.admin_db).create(
token=token_data, user=user
)
callback_url = env.str("SAML_SSO_CALLBACK_URL")
redirect_url = f"{callback_url}?id={saml_token.id}"
# token_data = serializer.validated_data
# saml_token = SAMLToken.objects.using(MainRouter.admin_db).create(
# token=token_data, user=user
# )
# callback_url = env.str("SAML_SSO_CALLBACK_URL")
# redirect_url = f"{callback_url}?id={saml_token.id}"
return redirect(redirect_url)
# return redirect(redirect_url)
@extend_schema_view(
api/src/backend/conftest.py
+49 -52
View File
@@ -1,9 +1,8 @@
import logging
from datetime import datetime, timedelta, timezone
from unittest.mock import MagicMock, patch
from unittest.mock import patch
import pytest
from allauth.socialaccount.models import SocialLogin
from django.conf import settings
from django.db import connection as django_connection
from django.db import connections as django_connections
@@ -29,8 +28,6 @@ from api.models import (
Resource,
ResourceTag,
Role,
SAMLConfiguration,
SAMLDomainIndex,
Scan,
ScanSummary,
StateChoices,
@@ -1121,62 +1118,62 @@ def latest_scan_finding(authenticated_client, providers_fixture, resources_fixtu
return finding
@pytest.fixture
def saml_setup(tenants_fixture):
tenant_id = tenants_fixture[0].id
domain = "example.com"
# @pytest.fixture
# def saml_setup(tenants_fixture):
# tenant_id = tenants_fixture[0].id
# domain = "example.com"
SAMLDomainIndex.objects.create(email_domain=domain, tenant_id=tenant_id)
# SAMLDomainIndex.objects.create(email_domain=domain, tenant_id=tenant_id)
metadata_xml = """<?xml version='1.0' encoding='UTF-8'?>
<md:EntityDescriptor entityID='TEST' xmlns:md='urn:oasis:names:tc:SAML:2.0:metadata'>
<md:IDPSSODescriptor WantAuthnRequestsSigned='false' protocolSupportEnumeration='urn:oasis:names:tc:SAML:2.0:protocol'>
<md:KeyDescriptor use='signing'>
<ds:KeyInfo xmlns:ds='http://www.w3.org/2000/09/xmldsig#'>
<ds:X509Data>
<ds:X509Certificate>TEST</ds:X509Certificate>
</ds:X509Data>
</ds:KeyInfo>
</md:KeyDescriptor>
<md:NameIDFormat>urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress</md:NameIDFormat>
<md:SingleSignOnService Binding='urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST' Location='https://TEST/sso/saml'/>
<md:SingleSignOnService Binding='urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect' Location='https://TEST/sso/saml'/>
</md:IDPSSODescriptor>
</md:EntityDescriptor>
"""
SAMLConfiguration.objects.create(
tenant_id=str(tenant_id),
email_domain=domain,
metadata_xml=metadata_xml,
)
# metadata_xml = """<?xml version='1.0' encoding='UTF-8'?>
# <md:EntityDescriptor entityID='TEST' xmlns:md='urn:oasis:names:tc:SAML:2.0:metadata'>
# <md:IDPSSODescriptor WantAuthnRequestsSigned='false' protocolSupportEnumeration='urn:oasis:names:tc:SAML:2.0:protocol'>
# <md:KeyDescriptor use='signing'>
# <ds:KeyInfo xmlns:ds='http://www.w3.org/2000/09/xmldsig#'>
# <ds:X509Data>
# <ds:X509Certificate>TEST</ds:X509Certificate>
# </ds:X509Data>
# </ds:KeyInfo>
# </md:KeyDescriptor>
# <md:NameIDFormat>urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress</md:NameIDFormat>
# <md:SingleSignOnService Binding='urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST' Location='https://TEST/sso/saml'/>
# <md:SingleSignOnService Binding='urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect' Location='https://TEST/sso/saml'/>
# </md:IDPSSODescriptor>
# </md:EntityDescriptor>
# """
# SAMLConfiguration.objects.create(
# tenant_id=str(tenant_id),
# email_domain=domain,
# metadata_xml=metadata_xml,
# )
return {
"email": f"user@{domain}",
"domain": domain,
"tenant_id": tenant_id,
}
# return {
# "email": f"user@{domain}",
# "domain": domain,
# "tenant_id": tenant_id,
# }
@pytest.fixture
def saml_sociallogin(users_fixture):
user = users_fixture[0]
user.email = "samlsso@acme.com"
extra_data = {
"firstName": ["Test"],
"lastName": ["User"],
"organization": ["Prowler"],
"userType": ["member"],
}
# @pytest.fixture
# def saml_sociallogin(users_fixture):
# user = users_fixture[0]
# user.email = "samlsso@acme.com"
# extra_data = {
# "firstName": ["Test"],
# "lastName": ["User"],
# "organization": ["Prowler"],
# "userType": ["member"],
# }
account = MagicMock()
account.provider = "saml"
account.extra_data = extra_data
# account = MagicMock()
# account.provider = "saml"
# account.extra_data = extra_data
sociallogin = MagicMock(spec=SocialLogin)
sociallogin.account = account
sociallogin.user = user
# sociallogin = MagicMock(spec=SocialLogin)
# sociallogin.account = account
# sociallogin.user = user
return sociallogin
# return sociallogin
def get_authorization_header(access_token: str) -> dict: