diff --git a/api/src/backend/api/adapters.py b/api/src/backend/api/adapters.py index 15f690ca54..b8380c9a48 100644 --- a/api/src/backend/api/adapters.py +++ b/api/src/backend/api/adapters.py @@ -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 diff --git a/api/src/backend/api/migrations/0031_lighthouseconfiguration.py b/api/src/backend/api/migrations/0030_lighthouseconfiguration.py similarity index 98% rename from api/src/backend/api/migrations/0031_lighthouseconfiguration.py rename to api/src/backend/api/migrations/0030_lighthouseconfiguration.py index 9393bbf32f..b12b8cac01 100644 --- a/api/src/backend/api/migrations/0031_lighthouseconfiguration.py +++ b/api/src/backend/api/migrations/0030_lighthouseconfiguration.py @@ -11,7 +11,7 @@ import api.rls class Migration(migrations.Migration): dependencies = [ - ("api", "0030_samlconfigurations"), + ("api", "0029_findings_check_index_parent"), ] operations = [ diff --git a/api/src/backend/api/migrations/0030_samlconfigurations.py b/api/src/backend/api/migrations/0030_samlconfigurations.py deleted file mode 100644 index b5ec1fea1a..0000000000 --- a/api/src/backend/api/migrations/0030_samlconfigurations.py +++ /dev/null @@ -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"), - ] - ), - ), - ] diff --git a/api/src/backend/api/migrations/0032_scan_disable_on_cascade_periodic_tasks.py b/api/src/backend/api/migrations/0031_scan_disable_on_cascade_periodic_tasks.py similarity index 92% rename from api/src/backend/api/migrations/0032_scan_disable_on_cascade_periodic_tasks.py rename to api/src/backend/api/migrations/0031_scan_disable_on_cascade_periodic_tasks.py index 07f3523ef2..c7e990ed13 100644 --- a/api/src/backend/api/migrations/0032_scan_disable_on_cascade_periodic_tasks.py +++ b/api/src/backend/api/migrations/0031_scan_disable_on_cascade_periodic_tasks.py @@ -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"), ] diff --git a/api/src/backend/api/migrations/0033_samltoken.py b/api/src/backend/api/migrations/0033_samltoken.py deleted file mode 100644 index f7abefd2e6..0000000000 --- a/api/src/backend/api/migrations/0033_samltoken.py +++ /dev/null @@ -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", - }, - ), - ] diff --git a/api/src/backend/api/models.py b/api/src/backend/api/models.py index 4d56135f2a..b321c5fe96 100644 --- a/api/src/backend/api/models.py +++ b/api/src/backend/api/models.py @@ -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 under .' - } - ) - 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 under .' +# } +# ) +# 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): diff --git a/api/src/backend/api/tests/test_adapters.py b/api/src/backend/api/tests/test_adapters.py index 05b4195121..b554719836 100644 --- a/api/src/backend/api/tests/test_adapters.py +++ b/api/src/backend/api/tests/test_adapters.py @@ -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() + # ) diff --git a/api/src/backend/api/tests/test_models.py b/api/src/backend/api/tests/test_models.py index d834143ed1..f02cf7b9d2 100644 --- a/api/src/backend/api/tests/test_models.py +++ b/api/src/backend/api/tests/test_models.py @@ -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 = """ - - - - - - FAKECERTDATA - - - - - - - """ +# @pytest.mark.django_db +# class TestSAMLConfigurationModel: +# VALID_METADATA = """ +# +# +# +# +# +# FAKECERTDATA +# +# +# +# +# +# +# """ - 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="", - 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="", +# 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 = """ - - """ - 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 = """ +# +# """ +# 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 = """ - - - - """ - 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 = """ +# +# +# +# """ +# 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] diff --git a/api/src/backend/api/v1/serializers.py b/api/src/backend/api/v1/serializers.py index 7fa716b208..858a9c0456 100644 --- a/api/src/backend/api/v1/serializers.py +++ b/api/src/backend/api/v1/serializers.py @@ -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): diff --git a/api/src/backend/api/v1/views.py b/api/src/backend/api/v1/views.py index 06b55ad716..6fe771ebfc 100644 --- a/api/src/backend/api/v1/views.py +++ b/api/src/backend/api/v1/views.py @@ -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., ) - - 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., ) +# - 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( diff --git a/api/src/backend/conftest.py b/api/src/backend/conftest.py index e3d6ae21f7..e4a03b78cb 100644 --- a/api/src/backend/conftest.py +++ b/api/src/backend/conftest.py @@ -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 = """ - - - - - - TEST - - - - urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress - - - - - """ - SAMLConfiguration.objects.create( - tenant_id=str(tenant_id), - email_domain=domain, - metadata_xml=metadata_xml, - ) +# metadata_xml = """ +# +# +# +# +# +# TEST +# +# +# +# urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress +# +# +# +# +# """ +# 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: