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

fix(saml): restore SAML, deactivate urls, enable idp-initiate (#8175)

Browse Source
This commit is contained in:
Adrián Jesús Peña Rodríguez
2025-07-07 16:42:11 +02:00
committed by GitHub
parent b38207507a
commit cd97e57521
15 changed files with 1534 additions and 1084 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/CHANGELOG.md
+3
View File
@@ -4,6 +4,9 @@ All notable changes to the **Prowler API** are documented in this file.
## [v1.10.0] (Prowler UNRELEASED)
### Added
- SSO with SAML support [(#8175)](https://github.com/prowler-cloud/prowler/pull/8175)
---
## [v1.9.1] (Prowler v5.8.1)
api/src/backend/api/adapters.py
+31 -90
View File
@@ -17,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.provider.id == "saml":
email = sociallogin.user.email
if email:
existing_user = self.get_user_by_email(email)
if existing_user:
@@ -31,98 +31,39 @@ class ProwlerSocialAccountAdapter(DefaultSocialAccountAdapter):
"""
with transaction.atomic(using=MainRouter.admin_db):
user = super().save_user(request, sociallogin, form)
provider = sociallogin.provider.id
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"
# 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
if provider != "saml":
# 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"
)
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,
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
)
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/0032_saml.py
+150
View File
@@ -0,0 +1,150 @@
# Generated by Django 5.1.10 on 2025-07-02 15:47
import uuid
import django.db.models.deletion
from django.conf import settings
from django.db import migrations, models
import api.db_utils
import api.rls
class Migration(migrations.Migration):
dependencies = [
("api", "0031_scan_disable_on_cascade_periodic_tasks"),
]
operations = [
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"),
]
),
),
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",
},
),
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.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"],
),
),
]
api/src/backend/api/models.py
+207 -198
View File
@@ -1,15 +1,21 @@
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
@@ -21,6 +27,7 @@ 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,
@@ -1364,242 +1371,244 @@ 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.provider_id = idp_settings["entity_id"]
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,
provider_id=idp_settings["entity_id"],
)
social_app.sites.set([current_site])
class ResourceScanSummary(RowLevelSecurityProtectedModel):
api/src/backend/api/specs/v1.yaml
+339
View File
@@ -5152,6 +5152,199 @@ paths:
responses:
'204':
description: Relationship deleted successfully
/api/v1/saml-config:
get:
operationId: saml_config_list
description: Returns all the SAML-based SSO configurations associated with the
current tenant.
summary: List all SSO configurations
parameters:
- in: query
name: fields[saml-configurations]
schema:
type: array
items:
type: string
enum:
- email_domain
- metadata_xml
- created_at
- updated_at
description: endpoint return only specific fields in the response on a per-type
basis by including a fields[TYPE] query parameter.
explode: false
- name: filter[search]
required: false
in: query
description: A search term.
schema:
type: string
- name: page[number]
required: false
in: query
description: A page number within the paginated result set.
schema:
type: integer
- name: page[size]
required: false
in: query
description: Number of results to return per page.
schema:
type: integer
- name: sort
required: false
in: query
description: '[list of fields to sort by](https://jsonapi.org/format/#fetching-sorting)'
schema:
type: array
items:
type: string
enum:
- id
- -id
- email_domain
- -email_domain
- metadata_xml
- -metadata_xml
- created_at
- -created_at
- updated_at
- -updated_at
explode: false
tags:
- SAML
security:
- jwtAuth: []
responses:
'200':
content:
application/vnd.api+json:
schema:
$ref: '#/components/schemas/PaginatedSAMLConfigurationList'
description: ''
post:
operationId: saml_config_create
description: Creates a new SAML SSO configuration for the current tenant, including
email domain and metadata XML.
summary: Create the SSO configuration
tags:
- SAML
requestBody:
content:
application/vnd.api+json:
schema:
$ref: '#/components/schemas/SAMLConfigurationRequest'
application/x-www-form-urlencoded:
schema:
$ref: '#/components/schemas/SAMLConfigurationRequest'
multipart/form-data:
schema:
$ref: '#/components/schemas/SAMLConfigurationRequest'
required: true
security:
- jwtAuth: []
responses:
'201':
content:
application/vnd.api+json:
schema:
$ref: '#/components/schemas/SAMLConfigurationResponse'
description: ''
/api/v1/saml-config/{id}:
get:
operationId: saml_config_retrieve
description: Returns the details of a specific SAML configuration belonging
to the current tenant.
summary: Retrieve SSO configuration details
parameters:
- in: query
name: fields[saml-configurations]
schema:
type: array
items:
type: string
enum:
- email_domain
- metadata_xml
- created_at
- updated_at
description: endpoint return only specific fields in the response on a per-type
basis by including a fields[TYPE] query parameter.
explode: false
- in: path
name: id
schema:
type: string
format: uuid
description: A UUID string identifying this saml configuration.
required: true
tags:
- SAML
security:
- jwtAuth: []
responses:
'200':
content:
application/vnd.api+json:
schema:
$ref: '#/components/schemas/SAMLConfigurationResponse'
description: ''
patch:
operationId: saml_config_partial_update
description: Partially updates an existing SAML SSO configuration. Supports
changes to email domain and metadata XML.
summary: Update the SSO configuration
parameters:
- in: path
name: id
schema:
type: string
format: uuid
description: A UUID string identifying this saml configuration.
required: true
tags:
- SAML
requestBody:
content:
application/vnd.api+json:
schema:
$ref: '#/components/schemas/PatchedSAMLConfigurationRequest'
application/x-www-form-urlencoded:
schema:
$ref: '#/components/schemas/PatchedSAMLConfigurationRequest'
multipart/form-data:
schema:
$ref: '#/components/schemas/PatchedSAMLConfigurationRequest'
required: true
security:
- jwtAuth: []
responses:
'200':
content:
application/vnd.api+json:
schema:
$ref: '#/components/schemas/SAMLConfigurationResponse'
description: ''
delete:
operationId: saml_config_destroy
description: Deletes an existing SAML SSO configuration associated with the
current tenant.
summary: Delete the SSO configuration
parameters:
- in: path
name: id
schema:
type: string
format: uuid
description: A UUID string identifying this saml configuration.
required: true
tags:
- SAML
security:
- jwtAuth: []
responses:
'204':
description: No response body
/api/v1/scans:
get:
operationId: scans_list
@@ -9158,6 +9351,15 @@ components:
$ref: '#/components/schemas/Role'
required:
- data
PaginatedSAMLConfigurationList:
type: object
properties:
data:
type: array
items:
$ref: '#/components/schemas/SAMLConfiguration'
required:
- data
PaginatedScanList:
type: object
properties:
@@ -10042,6 +10244,52 @@ components:
readOnly: true
required:
- data
PatchedSAMLConfigurationRequest:
type: object
properties:
data:
type: object
required:
- type
- id
additionalProperties: false
properties:
type:
type: string
description: The [type](https://jsonapi.org/format/#document-resource-object-identification)
member is used to describe resource objects that share common attributes
and relationships.
enum:
- saml-configurations
id:
type: string
format: uuid
attributes:
type: object
properties:
email_domain:
type: string
minLength: 1
description: Email domain used to identify the tenant, e.g. prowlerdemo.com
maxLength: 254
metadata_xml:
type: string
minLength: 1
description: Raw IdP metadata XML to configure SingleSignOnService,
certificates, etc.
created_at:
type: string
format: date-time
readOnly: true
updated_at:
type: string
format: date-time
readOnly: true
required:
- email_domain
- metadata_xml
required:
- data
PatchedScanUpdateRequest:
type: object
properties:
@@ -12155,6 +12403,97 @@ components:
$ref: '#/components/schemas/Role'
required:
- data
SAMLConfiguration:
type: object
required:
- type
- id
additionalProperties: false
properties:
type:
allOf:
- $ref: '#/components/schemas/SAMLConfigurationTypeEnum'
description: The [type](https://jsonapi.org/format/#document-resource-object-identification)
member is used to describe resource objects that share common attributes
and relationships.
id:
type: string
format: uuid
attributes:
type: object
properties:
email_domain:
type: string
description: Email domain used to identify the tenant, e.g. prowlerdemo.com
maxLength: 254
metadata_xml:
type: string
description: Raw IdP metadata XML to configure SingleSignOnService,
certificates, etc.
created_at:
type: string
format: date-time
readOnly: true
updated_at:
type: string
format: date-time
readOnly: true
required:
- email_domain
- metadata_xml
SAMLConfigurationRequest:
type: object
properties:
data:
type: object
required:
- type
additionalProperties: false
properties:
type:
type: string
description: The [type](https://jsonapi.org/format/#document-resource-object-identification)
member is used to describe resource objects that share common attributes
and relationships.
enum:
- saml-configurations
attributes:
type: object
properties:
email_domain:
type: string
minLength: 1
description: Email domain used to identify the tenant, e.g. prowlerdemo.com
maxLength: 254
metadata_xml:
type: string
minLength: 1
description: Raw IdP metadata XML to configure SingleSignOnService,
certificates, etc.
created_at:
type: string
format: date-time
readOnly: true
updated_at:
type: string
format: date-time
readOnly: true
required:
- email_domain
- metadata_xml
required:
- data
SAMLConfigurationResponse:
type: object
properties:
data:
$ref: '#/components/schemas/SAMLConfiguration'
required:
- data
SAMLConfigurationTypeEnum:
type: string
enum:
- saml-configurations
Scan:
type: object
required:
api/src/backend/api/tests/test_adapters.py
+18 -50
View File
@@ -20,69 +20,37 @@ 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.provider = MagicMock()
sociallogin.provider.id = "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()
sociallogin = MagicMock(spec=SocialLogin)
sociallogin.account = MagicMock()
sociallogin.account.provider = "github"
sociallogin.provider = MagicMock()
sociallogin.user = MagicMock()
sociallogin.provider.id = "saml"
sociallogin.account.extra_data = {}
sociallogin.connect = MagicMock()
adapter.pre_social_login(rf.get("/"), sociallogin)
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": [],
# }
# 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)
# 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
+124 -121
View File
@@ -1,6 +1,9 @@
import pytest
from allauth.socialaccount.models import SocialApp
from django.core.exceptions import ValidationError
from api.models import Resource, ResourceTag
from api.db_router import MainRouter
from api.models import Resource, ResourceTag, SAMLConfiguration, Tenant
@pytest.mark.django_db
@@ -122,147 +125,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/tests/test_views.py
+299 -282
View File
@@ -5,19 +5,26 @@ import os
import tempfile
from datetime import datetime, timedelta, timezone
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import ANY, MagicMock, Mock, patch
from urllib.parse import parse_qs, urlparse
from uuid import uuid4
import jwt
import pytest
from allauth.socialaccount.models import SocialAccount, SocialApp
from botocore.exceptions import ClientError, NoCredentialsError
from conftest import API_JSON_CONTENT_TYPE, TEST_PASSWORD, TEST_USER
from django.conf import settings
from django.http import JsonResponse
from django.test import RequestFactory
from django.urls import reverse
from django_celery_results.models import TaskResult
from rest_framework import status
from rest_framework.response import Response
from api.compliance import get_compliance_frameworks
from api.db_router import MainRouter
from api.models import (
Integration,
Invitation,
@@ -28,6 +35,8 @@ from api.models import (
ProviderSecret,
Role,
RoleProviderGroupRelationship,
SAMLConfiguration,
SAMLToken,
Scan,
StateChoices,
Task,
@@ -35,7 +44,7 @@ from api.models import (
UserRoleRelationship,
)
from api.rls import Tenant
from api.v1.views import ComplianceOverviewViewSet
from api.v1.views import ComplianceOverviewViewSet, TenantFinishACSView
TODAY = str(datetime.today().date())
@@ -5658,334 +5667,342 @@ class TestIntegrationViewSet:
assert response.status_code == status.HTTP_400_BAD_REQUEST
# @pytest.mark.django_db
# class TestSAMLTokenValidation:
# def test_valid_token_returns_tokens(self, authenticated_client, create_test_user):
# user = create_test_user
# valid_token_data = {
# "access": "mock_access_token",
# "refresh": "mock_refresh_token",
# }
# saml_token = SAMLToken.objects.create(
# token=valid_token_data,
# user=user,
# expires_at=datetime.now(timezone.utc) + timedelta(seconds=10),
# )
@pytest.mark.django_db
class TestSAMLTokenValidation:
def test_valid_token_returns_tokens(self, authenticated_client, create_test_user):
user = create_test_user
valid_token_data = {
"access": "mock_access_token",
"refresh": "mock_refresh_token",
}
saml_token = SAMLToken.objects.create(
token=valid_token_data,
user=user,
expires_at=datetime.now(timezone.utc) + timedelta(seconds=10),
)
# url = reverse("token-saml")
# response = authenticated_client.post(f"{url}?id={saml_token.id}")
url = reverse("token-saml")
response = authenticated_client.post(f"{url}?id={saml_token.id}")
# assert response.status_code == status.HTTP_200_OK
# assert response.json() == {"data": valid_token_data}
# assert not SAMLToken.objects.filter(id=saml_token.id).exists()
assert response.status_code == status.HTTP_200_OK
assert response.json() == {"data": valid_token_data}
assert not SAMLToken.objects.filter(id=saml_token.id).exists()
# def test_invalid_token_id_returns_404(self, authenticated_client):
# url = reverse("token-saml")
# response = authenticated_client.post(f"{url}?id={str(uuid4())}")
def test_invalid_token_id_returns_404(self, authenticated_client):
url = reverse("token-saml")
response = authenticated_client.post(f"{url}?id={str(uuid4())}")
# assert response.status_code == status.HTTP_404_NOT_FOUND
# assert response.json()["errors"]["detail"] == "Invalid token ID."
assert response.status_code == status.HTTP_404_NOT_FOUND
assert response.json()["errors"]["detail"] == "Invalid token ID."
# def test_expired_token_returns_400(self, authenticated_client, create_test_user):
# user = create_test_user
# expired_token_data = {
# "access": "expired_access_token",
# "refresh": "expired_refresh_token",
# }
# saml_token = SAMLToken.objects.create(
# token=expired_token_data,
# user=user,
# expires_at=datetime.now(timezone.utc) - timedelta(seconds=1),
# )
def test_expired_token_returns_400(self, authenticated_client, create_test_user):
user = create_test_user
expired_token_data = {
"access": "expired_access_token",
"refresh": "expired_refresh_token",
}
saml_token = SAMLToken.objects.create(
token=expired_token_data,
user=user,
expires_at=datetime.now(timezone.utc) - timedelta(seconds=1),
)
# url = reverse("token-saml")
# response = authenticated_client.post(f"{url}?id={saml_token.id}")
url = reverse("token-saml")
response = authenticated_client.post(f"{url}?id={saml_token.id}")
# assert response.status_code == status.HTTP_400_BAD_REQUEST
# assert response.json()["errors"]["detail"] == "Token expired."
# assert SAMLToken.objects.filter(id=saml_token.id).exists()
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert response.json()["errors"]["detail"] == "Token expired."
assert SAMLToken.objects.filter(id=saml_token.id).exists()
# def test_token_can_be_used_only_once(self, authenticated_client, create_test_user):
# user = create_test_user
# token_data = {
# "access": "single_use_token",
# "refresh": "single_use_refresh",
# }
# saml_token = SAMLToken.objects.create(
# token=token_data,
# user=user,
# expires_at=datetime.now(timezone.utc) + timedelta(seconds=10),
# )
def test_token_can_be_used_only_once(self, authenticated_client, create_test_user):
user = create_test_user
token_data = {
"access": "single_use_token",
"refresh": "single_use_refresh",
}
saml_token = SAMLToken.objects.create(
token=token_data,
user=user,
expires_at=datetime.now(timezone.utc) + timedelta(seconds=10),
)
# url = reverse("token-saml")
url = reverse("token-saml")
# # First use: should succeed
# response1 = authenticated_client.post(f"{url}?id={saml_token.id}")
# assert response1.status_code == status.HTTP_200_OK
# First use: should succeed
response1 = authenticated_client.post(f"{url}?id={saml_token.id}")
assert response1.status_code == status.HTTP_200_OK
# # Second use: should fail (already deleted)
# response2 = authenticated_client.post(f"{url}?id={saml_token.id}")
# assert response2.status_code == status.HTTP_404_NOT_FOUND
# Second use: should fail (already deleted)
response2 = authenticated_client.post(f"{url}?id={saml_token.id}")
assert response2.status_code == status.HTTP_404_NOT_FOUND
# @pytest.mark.django_db
# class TestSAMLInitiateAPIView:
# def test_valid_email_domain_and_certificates(
# self, authenticated_client, saml_setup, monkeypatch
# ):
# monkeypatch.setenv("SAML_PUBLIC_CERT", "fake_cert")
# monkeypatch.setenv("SAML_PRIVATE_KEY", "fake_key")
@pytest.mark.django_db
class TestSAMLInitiateAPIView:
def test_valid_email_domain_and_certificates(
self, authenticated_client, saml_setup, monkeypatch
):
monkeypatch.setenv("SAML_PUBLIC_CERT", "fake_cert")
monkeypatch.setenv("SAML_PRIVATE_KEY", "fake_key")
# url = reverse("api_saml_initiate")
# payload = {"email_domain": saml_setup["email"]}
url = reverse("api_saml_initiate")
payload = {"email_domain": saml_setup["email"]}
# response = authenticated_client.post(url, data=payload, format="json")
response = authenticated_client.post(url, data=payload, format="json")
# assert response.status_code == status.HTTP_302_FOUND
# assert (
# reverse("saml_login", kwargs={"organization_slug": saml_setup["domain"]})
# in response.url
# )
# assert "SAMLRequest" not in response.url
assert response.status_code == status.HTTP_302_FOUND
assert (
reverse("saml_login", kwargs={"organization_slug": saml_setup["domain"]})
in response.url
)
assert "SAMLRequest" not in response.url
# def test_invalid_email_domain(self, authenticated_client):
# url = reverse("api_saml_initiate")
# payload = {"email_domain": "user@unauthorized.com"}
def test_invalid_email_domain(self, authenticated_client):
url = reverse("api_saml_initiate")
payload = {"email_domain": "user@unauthorized.com"}
# response = authenticated_client.post(url, data=payload, format="json")
response = authenticated_client.post(url, data=payload, format="json")
# assert response.status_code == status.HTTP_403_FORBIDDEN
# assert response.json()["errors"]["detail"] == "Unauthorized domain."
assert response.status_code == status.HTTP_403_FORBIDDEN
assert response.json()["errors"]["detail"] == "Unauthorized domain."
# @pytest.mark.django_db
# class TestSAMLConfigurationViewSet:
# def test_list_saml_configurations(self, authenticated_client, saml_setup):
# config = SAMLConfiguration.objects.get(
# email_domain=saml_setup["email"].split("@")[-1]
# )
# response = authenticated_client.get(reverse("saml-config-list"))
# assert response.status_code == status.HTTP_200_OK
# assert (
# response.json()["data"][0]["attributes"]["email_domain"]
# == config.email_domain
# )
@pytest.mark.django_db
class TestSAMLConfigurationViewSet:
def test_list_saml_configurations(self, authenticated_client, saml_setup):
config = SAMLConfiguration.objects.get(
email_domain=saml_setup["email"].split("@")[-1]
)
response = authenticated_client.get(reverse("saml-config-list"))
assert response.status_code == status.HTTP_200_OK
assert (
response.json()["data"][0]["attributes"]["email_domain"]
== config.email_domain
)
# def test_retrieve_saml_configuration(self, authenticated_client, saml_setup):
# config = SAMLConfiguration.objects.get(
# email_domain=saml_setup["email"].split("@")[-1]
# )
# response = authenticated_client.get(
# reverse("saml-config-detail", kwargs={"pk": config.id})
# )
# assert response.status_code == status.HTTP_200_OK
# assert (
# response.json()["data"]["attributes"]["metadata_xml"] == config.metadata_xml
# )
def test_retrieve_saml_configuration(self, authenticated_client, saml_setup):
config = SAMLConfiguration.objects.get(
email_domain=saml_setup["email"].split("@")[-1]
)
response = authenticated_client.get(
reverse("saml-config-detail", kwargs={"pk": config.id})
)
assert response.status_code == status.HTTP_200_OK
assert (
response.json()["data"]["attributes"]["metadata_xml"] == config.metadata_xml
)
# def test_create_saml_configuration(self, authenticated_client, tenants_fixture):
# payload = {
# "email_domain": "newdomain.com",
# "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>
# """,
# }
# response = authenticated_client.post(
# reverse("saml-config-list"), data=payload, format="json"
# )
# assert response.status_code == status.HTTP_201_CREATED
# assert SAMLConfiguration.objects.filter(email_domain="newdomain.com").exists()
def test_create_saml_configuration(self, authenticated_client, tenants_fixture):
payload = {
"email_domain": "newdomain.com",
"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>
""",
}
response = authenticated_client.post(
reverse("saml-config-list"), data=payload, format="json"
)
assert response.status_code == status.HTTP_201_CREATED
assert SAMLConfiguration.objects.filter(email_domain="newdomain.com").exists()
# def test_update_saml_configuration(self, authenticated_client, saml_setup):
# config = SAMLConfiguration.objects.get(
# email_domain=saml_setup["email"].split("@")[-1]
# )
# payload = {
# "data": {
# "type": "saml-configurations",
# "id": str(config.id),
# "attributes": {
# "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>TEST2</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>
# """
# },
# }
# }
# response = authenticated_client.patch(
# reverse("saml-config-detail", kwargs={"pk": config.id}),
# data=payload,
# content_type="application/vnd.api+json",
# )
# assert response.status_code == status.HTTP_200_OK
# config.refresh_from_db()
# assert (
# config.metadata_xml.strip()
# == payload["data"]["attributes"]["metadata_xml"].strip()
# )
def test_update_saml_configuration(self, authenticated_client, saml_setup):
config = SAMLConfiguration.objects.get(
email_domain=saml_setup["email"].split("@")[-1]
)
payload = {
"data": {
"type": "saml-configurations",
"id": str(config.id),
"attributes": {
"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>TEST2</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>
"""
},
}
}
response = authenticated_client.patch(
reverse("saml-config-detail", kwargs={"pk": config.id}),
data=payload,
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_200_OK
config.refresh_from_db()
assert (
config.metadata_xml.strip()
== payload["data"]["attributes"]["metadata_xml"].strip()
)
# def test_delete_saml_configuration(self, authenticated_client, saml_setup):
# config = SAMLConfiguration.objects.get(
# email_domain=saml_setup["email"].split("@")[-1]
# )
# response = authenticated_client.delete(
# reverse("saml-config-detail", kwargs={"pk": config.id})
# )
# assert response.status_code == status.HTTP_204_NO_CONTENT
# assert not SAMLConfiguration.objects.filter(id=config.id).exists()
def test_delete_saml_configuration(self, authenticated_client, saml_setup):
config = SAMLConfiguration.objects.get(
email_domain=saml_setup["email"].split("@")[-1]
)
response = authenticated_client.delete(
reverse("saml-config-detail", kwargs={"pk": config.id})
)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert not SAMLConfiguration.objects.filter(id=config.id).exists()
# @pytest.mark.django_db
# class TestTenantFinishACSView:
# def test_dispatch_skips_if_user_not_authenticated(self):
# request = RequestFactory().get(
# reverse("saml_finish_acs", kwargs={"organization_slug": "testtenant"})
# )
# request.user = type("Anonymous", (), {"is_authenticated": False})()
@pytest.mark.django_db
class TestTenantFinishACSView:
def test_dispatch_skips_if_user_not_authenticated(self):
request = RequestFactory().get(
reverse("saml_finish_acs", kwargs={"organization_slug": "testtenant"})
)
request.user = type("Anonymous", (), {"is_authenticated": False})()
# with patch(
# "allauth.socialaccount.providers.saml.views.get_app_or_404"
# ) as mock_get_app:
# mock_get_app.return_value = SocialApp(
# provider="saml",
# client_id="testtenant",
# name="Test App",
# settings={},
# )
with patch(
"allauth.socialaccount.providers.saml.views.get_app_or_404"
) as mock_get_app:
mock_get_app.return_value = SocialApp(
provider="saml",
client_id="testtenant",
name="Test App",
settings={},
)
# view = TenantFinishACSView.as_view()
# response = view(request, organization_slug="testtenant")
view = TenantFinishACSView.as_view()
response = view(request, organization_slug="testtenant")
# assert response.status_code in [200, 302]
assert response.status_code in [200, 302]
# def test_dispatch_skips_if_social_app_not_found(self, users_fixture):
# request = RequestFactory().get(
# reverse("saml_finish_acs", kwargs={"organization_slug": "testtenant"})
# )
# request.user = users_fixture[0]
def test_dispatch_skips_if_social_app_not_found(self, users_fixture):
request = RequestFactory().get(
reverse("saml_finish_acs", kwargs={"organization_slug": "testtenant"})
)
request.user = users_fixture[0]
# with patch(
# "allauth.socialaccount.providers.saml.views.get_app_or_404"
# ) as mock_get_app:
# mock_get_app.return_value = SocialApp(
# provider="saml",
# client_id="testtenant",
# name="Test App",
# settings={},
# )
with patch(
"allauth.socialaccount.providers.saml.views.get_app_or_404"
) as mock_get_app:
mock_get_app.return_value = SocialApp(
provider="saml",
client_id="testtenant",
name="Test App",
settings={},
)
# view = TenantFinishACSView.as_view()
# response = view(request, organization_slug="testtenant")
view = TenantFinishACSView.as_view()
response = view(request, organization_slug="testtenant")
# assert isinstance(response, JsonResponse) or response.status_code in [200, 302]
assert isinstance(response, JsonResponse) or response.status_code in [200, 302]
# def test_dispatch_sets_user_profile_and_assigns_role_and_creates_token(
# self, create_test_user, tenants_fixture, saml_setup, settings, monkeypatch
# ):
# monkeypatch.setenv("SAML_SSO_CALLBACK_URL", "http://localhost/sso-complete")
# user = create_test_user
# original_email = user.email
# original_name = user.name
# original_company = user.company_name
# user.email = f"doe@{saml_setup['email']}"
def test_dispatch_sets_user_profile_and_assigns_role_and_creates_token(
self, create_test_user, tenants_fixture, saml_setup, settings, monkeypatch
):
monkeypatch.setenv("SAML_SSO_CALLBACK_URL", "http://localhost/sso-complete")
user = create_test_user
original_name = user.name
original_company = user.company_name
user.company_name = "testing_company"
user.is_authenticate = True
# social_account = SocialAccount(
# user=user,
# provider="saml",
# extra_data={
# "firstName": ["John"],
# "lastName": ["Doe"],
# "organization": ["TestOrg"],
# "userType": ["saml_default_role"],
# },
# )
social_account = SocialAccount(
user=user,
provider="saml",
extra_data={
"firstName": ["John"],
"lastName": ["Doe"],
"organization": ["testing_company"],
"userType": ["saml_default_role"],
},
)
# request = RequestFactory().get(
# reverse("saml_finish_acs", kwargs={"organization_slug": "testtenant"})
# )
# request.user = user
request = RequestFactory().get(
reverse("saml_finish_acs", kwargs={"organization_slug": "testtenant"})
)
request.user = user
# with (
# patch(
# "allauth.socialaccount.providers.saml.views.get_app_or_404"
# ) as mock_get_app_or_404,
# patch("allauth.socialaccount.models.SocialApp.objects.get"),
# patch(
# "allauth.socialaccount.models.SocialAccount.objects.get"
# ) as mock_sa_get,
# ):
# mock_get_app_or_404.return_value = MagicMock(
# provider="saml", client_id="testtenant", name="Test App", settings={}
# )
# mock_sa_get.return_value = social_account
with (
patch(
"allauth.socialaccount.providers.saml.views.get_app_or_404"
) as mock_get_app_or_404,
patch(
"allauth.socialaccount.models.SocialApp.objects.get"
) as mock_socialapp_get,
patch(
"allauth.socialaccount.models.SocialAccount.objects.get"
) as mock_sa_get,
patch("api.models.SAMLDomainIndex.objects.get") as mock_saml_domain_get,
patch("api.models.SAMLConfiguration.objects.get") as mock_saml_config_get,
patch("api.models.User.objects.get") as mock_user_get,
):
mock_get_app_or_404.return_value = MagicMock(
provider="saml", client_id="testtenant", name="Test App", settings={}
)
mock_sa_get.return_value = social_account
mock_socialapp_get.return_value = MagicMock(provider_id="saml")
mock_saml_domain_get.return_value = SimpleNamespace(
tenant_id=tenants_fixture[0].id
)
mock_saml_config_get.return_value = MagicMock()
mock_user_get.return_value = user
# view = TenantFinishACSView.as_view()
# response = view(request, organization_slug="testtenant")
view = TenantFinishACSView.as_view()
response = view(request, organization_slug="testtenant")
# assert response.status_code == 302
assert response.status_code == 302
# expected_callback_host = "localhost"
# parsed_url = urlparse(response.url)
# assert parsed_url.netloc == expected_callback_host
# query_params = parse_qs(parsed_url.query)
# assert "id" in query_params
expected_callback_host = "localhost"
parsed_url = urlparse(response.url)
assert parsed_url.netloc == expected_callback_host
query_params = parse_qs(parsed_url.query)
assert "id" in query_params
# token_id = query_params["id"][0]
# token_obj = SAMLToken.objects.get(id=token_id)
# assert token_obj.user == user
# assert not token_obj.is_expired()
token_id = query_params["id"][0]
token_obj = SAMLToken.objects.get(id=token_id)
assert token_obj.user == user
assert not token_obj.is_expired()
# user.refresh_from_db()
# assert user.name == "John Doe"
# assert user.company_name == "TestOrg"
user.refresh_from_db()
assert user.name == "John Doe"
assert user.company_name == "testing_company"
# role = Role.objects.using(MainRouter.admin_db).get(name="saml_default_role")
# assert role.tenant == tenants_fixture[0]
role = Role.objects.using(MainRouter.admin_db).get(name="saml_default_role")
assert role.tenant == tenants_fixture[0]
# assert (
# UserRoleRelationship.objects.using(MainRouter.admin_db)
# .filter(user=user, tenant_id=tenants_fixture[0].id)
# .exists()
# )
assert (
UserRoleRelationship.objects.using(MainRouter.admin_db)
.filter(user=user, tenant_id=tenants_fixture[0].id)
.exists()
)
# # Membership should have been created with default role
# membership = Membership.objects.using(MainRouter.admin_db).get(
# user=user, tenant=tenants_fixture[0]
# )
# assert membership.role == Membership.RoleChoices.MEMBER
# assert membership.user == user
# assert membership.tenant == tenants_fixture[0]
membership = Membership.objects.using(MainRouter.admin_db).get(
user=user, tenant=tenants_fixture[0]
)
assert membership.role == Membership.RoleChoices.MEMBER
assert membership.user == user
assert membership.tenant == tenants_fixture[0]
# # Restore original user state
# user.email = original_email
# user.name = original_name
# user.company_name = original_company
# user.save()
user.name = original_name
user.company_name = original_company
user.save()
@pytest.mark.django_db
api/src/backend/api/v1/serializers.py
+19 -12
View File
@@ -29,6 +29,7 @@ from api.models import (
ResourceTag,
Role,
RoleProviderGroupRelationship,
SAMLConfiguration,
Scan,
StateChoices,
StatusChoices,
@@ -129,6 +130,12 @@ class TokenSerializer(BaseTokenSerializer):
class TokenSocialLoginSerializer(BaseTokenSerializer):
email = serializers.EmailField(write_only=True)
tenant_id = serializers.UUIDField(
write_only=True,
required=False,
help_text="If not provided, the tenant ID of the first membership that was added"
" to the user will be used.",
)
# Output tokens
refresh = serializers.CharField(read_only=True)
@@ -2067,23 +2074,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/urls.py
+37 -19
View File
@@ -1,9 +1,11 @@
from allauth.socialaccount.providers.saml.views import ACSView, MetadataView, SLSView
from django.urls import include, path
from drf_spectacular.views import SpectacularRedocView
from rest_framework_nested import routers
from api.v1.views import (
ComplianceOverviewViewSet,
CustomSAMLLoginView,
CustomTokenObtainView,
CustomTokenRefreshView,
CustomTokenSwitchTenantView,
@@ -23,10 +25,14 @@ from api.v1.views import (
ResourceViewSet,
RoleProviderGroupRelationshipView,
RoleViewSet,
SAMLConfigurationViewSet,
SAMLInitiateAPIView,
SAMLTokenValidateView,
ScanViewSet,
ScheduleViewSet,
SchemaView,
TaskViewSet,
TenantFinishACSView,
TenantMembersViewSet,
TenantViewSet,
UserRoleRelationshipView,
@@ -50,7 +56,7 @@ router.register(
router.register(r"overviews", OverviewViewSet, basename="overview")
router.register(r"schedules", ScheduleViewSet, basename="schedule")
router.register(r"integrations", IntegrationViewSet, basename="integration")
# router.register(r"saml-config", SAMLConfigurationViewSet, basename="saml-config")
router.register(r"saml-config", SAMLConfigurationViewSet, basename="saml-config")
router.register(
r"lighthouse-configurations",
LighthouseConfigViewSet,
@@ -119,24 +125,36 @@ urlpatterns = [
),
name="provider_group-providers-relationship",
),
# API endpoint to start SAML SSO flow (WIP)
# path(
# "auth/saml/initiate/", SAMLInitiateAPIView.as_view(), name="api_saml_initiate"
# ),
# # Custom SAML endpoints (must come before allauth.urls) (WIP)
# path(
# "accounts/saml/<organization_slug>/login/",
# CustomSAMLLoginView.as_view(),
# name="saml_login",
# ),
# path(
# "accounts/saml/<organization_slug>/acs/finish/",
# TenantFinishACSView.as_view(),
# name="saml_finish_acs",
# ),
# Allauth SAML endpoints for tenants (WIP)
# path("accounts/", include("allauth.urls")),
# path("tokens/saml", SAMLTokenValidateView.as_view(), name="token-saml"),
# API endpoint to start SAML SSO flow
path(
"auth/saml/initiate/", SAMLInitiateAPIView.as_view(), name="api_saml_initiate"
),
path(
"accounts/saml/<organization_slug>/login/",
CustomSAMLLoginView.as_view(),
name="saml_login",
),
path(
"accounts/saml/<organization_slug>/acs/",
ACSView.as_view(),
name="saml_acs",
),
path(
"accounts/saml/<organization_slug>/acs/finish/",
TenantFinishACSView.as_view(),
name="saml_finish_acs",
),
path(
"accounts/saml/<organization_slug>/sls/",
SLSView.as_view(),
name="saml_sls",
),
path(
"accounts/saml/<organization_slug>/metadata/",
MetadataView.as_view(),
name="saml_metadata",
),
path("tokens/saml", SAMLTokenValidateView.as_view(), name="token-saml"),
path("tokens/google", GoogleSocialLoginView.as_view(), name="token-google"),
path("tokens/github", GithubSocialLoginView.as_view(), name="token-github"),
path("", include(router.urls)),
api/src/backend/api/v1/views.py
+225 -203
View File
@@ -1,10 +1,13 @@
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
@@ -20,6 +23,7 @@ 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
@@ -64,6 +68,7 @@ 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,
@@ -101,6 +106,9 @@ from api.models import (
ResourceScanSummary,
Role,
RoleProviderGroupRelationship,
SAMLConfiguration,
SAMLDomainIndex,
SAMLToken,
Scan,
ScanSummary,
SeverityChoices,
@@ -157,6 +165,8 @@ from api.v1.serializers import (
RoleProviderGroupRelationshipSerializer,
RoleSerializer,
RoleUpdateSerializer,
SAMLConfigurationSerializer,
SamlInitiateSerializer,
ScanComplianceReportSerializer,
ScanCreateSerializer,
ScanReportSerializer,
@@ -393,240 +403,252 @@ 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"
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
# Defensive check to avoid edge case failures due to inconsistent or incomplete data in the database
# This handles scenarios like partially deleted or missing related objects
try:
check = SAMLDomainIndex.objects.get(email_domain=organization_slug)
with rls_transaction(str(check.tenant_id)):
SAMLConfiguration.objects.get(tenant_id=str(check.tenant_id))
social_app = SocialApp.objects.get(
provider="saml", client_id=organization_slug
)
user_id = User.objects.get(email=str(user)).id
social_account = SocialAccount.objects.get(
user=str(user_id), provider=social_app.provider_id
)
except (
SAMLDomainIndex.DoesNotExist,
SAMLConfiguration.DoesNotExist,
SocialApp.DoesNotExist,
SocialAccount.DoesNotExist,
User.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, "tenant_id": str(tenant.id)}
)
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/config/django/base.py
+5
View File
@@ -11,6 +11,7 @@ SECRET_KEY = env("SECRET_KEY", default="secret")
DEBUG = env.bool("DJANGO_DEBUG", default=False)
ALLOWED_HOSTS = ["localhost", "127.0.0.1"]
SECURE_PROXY_SSL_HEADER = ("HTTP_X_FORWARDED_PROTO", "https")
USE_X_FORWARDED_HOST = True
# Application definition
@@ -248,3 +249,7 @@ X_FRAME_OPTIONS = "DENY"
SECURE_REFERRER_POLICY = "strict-origin-when-cross-origin"
DJANGO_DELETION_BATCH_SIZE = env.int("DJANGO_DELETION_BATCH_SIZE", 5000)
# SAML requirement
CSRF_COOKIE_SECURE = True
SESSION_COOKIE_SECURE = True
api/src/backend/config/settings/social_login.py
+15 -9
View File
@@ -25,9 +25,18 @@ SOCIALACCOUNT_EMAIL_AUTHENTICATION = True
SOCIALACCOUNT_EMAIL_AUTHENTICATION_AUTO_CONNECT = True
SOCIALACCOUNT_ADAPTER = "api.adapters.ProwlerSocialAccountAdapter"
# SAML keys (TODO: Validate certificates)
# SAML_PUBLIC_CERT = env("SAML_PUBLIC_CERT", default="")
# SAML_PRIVATE_KEY = env("SAML_PRIVATE_KEY", default="")
# def inline(pem: str) -> str:
# return "".join(
# line.strip()
# for line in pem.splitlines()
# if "CERTIFICATE" not in line and "KEY" not in line
# )
# # SAML keys (TODO: Validate certificates)
# SAML_PUBLIC_CERT = inline(env("SAML_PUBLIC_CERT", default=""))
# SAML_PRIVATE_KEY = inline(env("SAML_PRIVATE_KEY", default=""))
SOCIALACCOUNT_PROVIDERS = {
"google": {
@@ -60,17 +69,14 @@ SOCIALACCOUNT_PROVIDERS = {
"entity_id": "urn:prowler.com:sp",
},
"advanced": {
# TODO: Validate certificates
# "x509cert": SAML_PUBLIC_CERT,
# "private_key": SAML_PRIVATE_KEY,
# "authn_request_signed": True,
# "want_assertion_signed": True,
# "want_message_signed": True,
# "want_assertion_signed": True,
"reject_idp_initiated_sso": False,
"name_id_format": "urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress",
"authn_request_signed": False,
"logout_request_signed": False,
"logout_response_signed": False,
"want_assertion_encrypted": False,
"want_name_id_encrypted": False,
},
},
}
api/src/backend/conftest.py
+52 -49
View File
@@ -1,8 +1,9 @@
import logging
from datetime import datetime, timedelta, timezone
from unittest.mock import patch
from unittest.mock import MagicMock, 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
@@ -28,6 +29,8 @@ from api.models import (
Resource,
ResourceTag,
Role,
SAMLConfiguration,
SAMLDomainIndex,
Scan,
ScanSummary,
StateChoices,
@@ -1118,62 +1121,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 = "prowler.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: