ben.carrasco/prowler
1
0 Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-10-30 23:34:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(image): add image provider API support with registry scan mode

Browse Source 
- Add IMAGE provider type to API models, serializers, views, and utils
- Add registry credential handling (username/password, token, filters)
- Add UID validation for registry URLs with port validation
- Add connection testing via OCI registry adapter
- Add registry scan mode with OCI, Docker Hub, and ECR adapter layer
- Add per-image progress tracking and Trivy native auth
- Skip compliance/reports for IMAGE provider scans
- Add migration, OpenAPI spec updates, and comprehensive tests
This commit is contained in:
Andoni A. committed 2026-02-18 18:21:20 +01:00
1 parent f7adbc2dcf
commit ad3b57d405
35 files changed
+4128 -968
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

api/CHANGELOG.md
+8
View File
@@ -2,6 +2,14 @@
All notable changes to the **Prowler API** are documented in this file.
## [Unreleased]
### 🚀 Added
- Image provider support with registry credential handling, UID validation, and connection testing
---
## [1.19.0] (Prowler v5.18.0)
### 🚀 Added
api/poetry.lock
Generated
+513 -193
View File
File diff suppressed because it is too large. Load diff
api/pyproject.toml
+2 -2
View File
@@ -24,7 +24,7 @@ dependencies = [
"drf-spectacular-jsonapi==0.5.1",
"gunicorn==23.0.0",
"lxml==5.3.2",
"prowler @ git+https://github.com/prowler-cloud/prowler.git@master",
"prowler @ git+https://github.com/prowler-cloud/prowler.git@feat/PROWLER-940-stage-2-a-image-provider-api-support",
"psycopg2-binary==2.9.9",
"pytest-celery[redis] (>=1.0.1,<2.0.0)",
"sentry-sdk[django] (>=2.20.0,<3.0.0)",
@@ -49,7 +49,7 @@ name = "prowler-api"
package-mode = false
# Needed for the SDK compatibility
requires-python = ">=3.11,<3.13"
version = "1.19.0"
version = "1.20.0"
[project.scripts]
celery = "src.backend.config.settings.celery"
api/src/backend/api/migrations/0076_image_provider.py
+39
View File
@@ -0,0 +1,39 @@
# Generated by Django migration for Image provider support
from django.db import migrations
import api.db_utils
class Migration(migrations.Migration):
dependencies = [
("api", "0075_cloudflare_provider"),
]
operations = [
migrations.AlterField(
model_name="provider",
name="provider",
field=api.db_utils.ProviderEnumField(
choices=[
("aws", "AWS"),
("azure", "Azure"),
("gcp", "GCP"),
("kubernetes", "Kubernetes"),
("m365", "M365"),
("github", "GitHub"),
("mongodbatlas", "MongoDB Atlas"),
("iac", "IaC"),
("oraclecloud", "Oracle Cloud Infrastructure"),
("alibabacloud", "Alibaba Cloud"),
("cloudflare", "Cloudflare"),
("image", "Image"),
],
default="aws",
),
),
migrations.RunSQL(
"ALTER TYPE provider ADD VALUE IF NOT EXISTS 'image';",
reverse_sql=migrations.RunSQL.noop,
),
]
api/src/backend/api/models.py
+23 -17
View File
@@ -288,6 +288,7 @@ class Provider(RowLevelSecurityProtectedModel):
ORACLECLOUD = "oraclecloud", _("Oracle Cloud Infrastructure")
ALIBABACLOUD = "alibabacloud", _("Alibaba Cloud")
CLOUDFLARE = "cloudflare", _("Cloudflare")
IMAGE = "image", _("Image")
@staticmethod
def validate_aws_uid(value):
@@ -410,6 +411,26 @@ class Provider(RowLevelSecurityProtectedModel):
pointer="/data/attributes/uid",
)
@staticmethod
def validate_image_uid(value):
pattern = r"^[a-zA-Z0-9]([a-zA-Z0-9._-]*[a-zA-Z0-9])?(:\d{1,5})?(/[a-zA-Z0-9._-]+)*/?$"
if not re.match(pattern, value):
raise ModelValidationError(
detail="Image provider ID must be a valid registry URL "
"(e.g., docker.io, ghcr.io, registry.example.com:5000).",
code="image-uid",
pointer="/data/attributes/uid",
)
port_match = re.search(r":(\d{1,5})(?=/|$)", value)
if port_match:
port = int(port_match.group(1))
if not 1 <= port <= 65535:
raise ModelValidationError(
detail="Port number must be between 1 and 65535.",
code="image-uid",
pointer="/data/attributes/uid",
)
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)
@@ -433,6 +454,8 @@ class Provider(RowLevelSecurityProtectedModel):
def clean(self):
super().clean()
if self.provider == self.ProviderChoices.IMAGE.value and self.uid:
self.uid = re.sub(r"^https?://", "", self.uid)
getattr(self, f"validate_{self.provider}_uid")(self.uid)
def save(self, *args, **kwargs):
@@ -681,8 +704,6 @@ class AttackPathsScan(RowLevelSecurityProtectedModel):
update_tag = models.BigIntegerField(
null=True, blank=True, help_text="Cartography update tag (epoch)"
)
graph_database = models.CharField(max_length=63, null=True, blank=True)
is_graph_database_deleted = models.BooleanField(default=False)
ingestion_exceptions = models.JSONField(default=dict, null=True, blank=True)
class Meta(RowLevelSecurityProtectedModel.Meta):
@@ -709,21 +730,6 @@ class AttackPathsScan(RowLevelSecurityProtectedModel):
fields=["tenant_id", "scan_id"],
name="aps_scan_lookup_idx",
),
models.Index(
fields=["tenant_id", "provider_id"],
name="aps_active_graph_idx",
include=["graph_database", "id"],
condition=Q(is_graph_database_deleted=False),
),
models.Index(
fields=["tenant_id", "provider_id", "-completed_at"],
name="aps_completed_graph_idx",
include=["graph_database", "id"],
condition=Q(
state=StateChoices.COMPLETED,
is_graph_database_deleted=False,
),
),
]
class JSONAPIMeta:
api/src/backend/api/specs/v1.yaml
+43 -7
View File
@@ -1,7 +1,7 @@
openapi: 3.0.3
info:
title: Prowler API
version: 1.19.0
version: 1.20.0
description: |-
Prowler API specification.
@@ -616,7 +616,7 @@ paths:
operationId: attack_paths_scans_queries_retrieve
description: Retrieve the catalog of Attack Paths queries available for this
Attack Paths scan.
summary: List attack paths queries
summary: List Attack Paths queries
parameters:
- in: query
name: fields[attack-paths-scans]
@@ -714,7 +714,7 @@ paths:
description: Bad request (e.g., Unknown Attack Paths query for the selected
provider)
'404':
description: No attack paths found for the given query and parameters
description: No Attack Paths found for the given query and parameters
'500':
description: Attack Paths query execution failed due to a database error
/api/v1/compliance-overviews:
@@ -12438,6 +12438,8 @@ components:
type: string
name:
type: string
short_description:
type: string
description:
type: string
provider:
@@ -12446,12 +12448,42 @@ components:
type: array
items:
$ref: '#/components/schemas/AttackPathsQueryParameter'
attribution:
allOf:
- $ref: '#/components/schemas/AttackPathsQueryAttribution'
nullable: true
required:
- id
- name
- short_description
- description
- provider
- parameters
AttackPathsQueryAttribution:
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:
- attack-paths-query-attributions
id: {}
attributes:
type: object
properties:
text:
type: string
link:
type: string
required:
- text
- link
AttackPathsQueryParameter:
type: object
required:
@@ -17316,7 +17348,8 @@ components:
required:
- api_key
- api_email
writeOnly: true
- type: object
- type: object
required:
- secret
relationships:
@@ -19346,7 +19379,8 @@ components:
required:
- api_key
- api_email
writeOnly: true
- type: object
- type: object
required:
- secret_type
- secret
@@ -19732,7 +19766,8 @@ components:
required:
- api_key
- api_email
writeOnly: true
- type: object
- type: object
required:
- secret_type
- secret
@@ -20130,7 +20165,8 @@ components:
required:
- api_key
- api_email
writeOnly: true
- type: object
- type: object
required:
- secret
relationships:
api/src/backend/api/tests/test_utils.py
+147
View File
@@ -24,6 +24,7 @@ from prowler.providers.cloudflare.cloudflare_provider import CloudflareProvider
from prowler.providers.gcp.gcp_provider import GcpProvider
from prowler.providers.github.github_provider import GithubProvider
from prowler.providers.iac.iac_provider import IacProvider
from prowler.providers.image.image_provider import ImageProvider
from prowler.providers.kubernetes.kubernetes_provider import KubernetesProvider
from prowler.providers.m365.m365_provider import M365Provider
from prowler.providers.mongodbatlas.mongodbatlas_provider import MongodbatlasProvider
@@ -120,6 +121,7 @@ class TestReturnProwlerProvider:
(Provider.ProviderChoices.IAC.value, IacProvider),
(Provider.ProviderChoices.ALIBABACLOUD.value, AlibabacloudProvider),
(Provider.ProviderChoices.CLOUDFLARE.value, CloudflareProvider),
(Provider.ProviderChoices.IMAGE.value, ImageProvider),
],
)
def test_return_prowler_provider(self, provider_type, expected_provider):
@@ -186,6 +188,54 @@ class TestProwlerProviderConnectionTest:
assert isinstance(connection.error, Provider.secret.RelatedObjectDoesNotExist)
assert str(connection.error) == "Provider has no secret."
@patch(
"prowler.providers.image.lib.registry.factory.create_registry_adapter",
)
@patch("api.utils.return_prowler_provider")
def test_prowler_provider_connection_test_image_success(
self, mock_return_prowler_provider, mock_create_adapter
):
provider = MagicMock()
provider.provider = Provider.ProviderChoices.IMAGE.value
provider.uid = "ghcr.io"
provider.secret.secret = {"registry_token": "tok"}
mock_adapter = MagicMock()
mock_create_adapter.return_value = mock_adapter
connection = prowler_provider_connection_test(provider)
assert connection.is_connected is True
assert connection.error is None
mock_create_adapter.assert_called_once_with(
registry_url="ghcr.io",
username=None,
password=None,
token="tok",
)
mock_adapter.list_repositories.assert_called_once()
@patch(
"prowler.providers.image.lib.registry.factory.create_registry_adapter",
)
@patch("api.utils.return_prowler_provider")
def test_prowler_provider_connection_test_image_failure(
self, mock_return_prowler_provider, mock_create_adapter
):
provider = MagicMock()
provider.provider = Provider.ProviderChoices.IMAGE.value
provider.uid = "ghcr.io"
provider.secret.secret = {"registry_token": "bad-token"}
mock_adapter = MagicMock()
mock_adapter.list_repositories.side_effect = Exception("401 Unauthorized")
mock_create_adapter.return_value = mock_adapter
connection = prowler_provider_connection_test(provider)
assert connection.is_connected is False
assert connection.error == "401 Unauthorized"
class TestGetProwlerProviderKwargs:
@pytest.mark.parametrize(
@@ -330,6 +380,103 @@ class TestGetProwlerProviderKwargs:
}
assert result == expected_result
def test_get_prowler_provider_kwargs_image_provider(self):
"""Test that Image provider gets correct kwargs with registry URL and auth."""
provider_uid = "ghcr.io"
secret_dict = {
"registry_username": "user",
"registry_password": "pass",
}
secret_mock = MagicMock()
secret_mock.secret = secret_dict
provider = MagicMock()
provider.provider = Provider.ProviderChoices.IMAGE.value
provider.secret = secret_mock
provider.uid = provider_uid
result = get_prowler_provider_kwargs(provider)
expected_result = {
"registry": provider_uid,
"registry_username": "user",
"registry_password": "pass",
}
assert result == expected_result
def test_get_prowler_provider_kwargs_image_provider_with_filters(self):
"""Test that Image provider includes scan filters."""
provider_uid = "docker.io"
secret_dict = {
"registry_token": "ghp_abc123",
"image_filter": "my-app.*",
"tag_filter": "v[0-9]+",
"max_images": 50,
}
secret_mock = MagicMock()
secret_mock.secret = secret_dict
provider = MagicMock()
provider.provider = Provider.ProviderChoices.IMAGE.value
provider.secret = secret_mock
provider.uid = provider_uid
result = get_prowler_provider_kwargs(provider)
expected_result = {
"registry": provider_uid,
"registry_token": "ghp_abc123",
"image_filter": "my-app.*",
"tag_filter": "v[0-9]+",
"max_images": 50,
}
assert result == expected_result
def test_get_prowler_provider_kwargs_image_provider_no_auth(self):
"""Test that Image provider works with empty secret for public registries."""
provider_uid = "docker.io"
secret_dict = {}
secret_mock = MagicMock()
secret_mock.secret = secret_dict
provider = MagicMock()
provider.provider = Provider.ProviderChoices.IMAGE.value
provider.secret = secret_mock
provider.uid = provider_uid
result = get_prowler_provider_kwargs(provider)
expected_result = {"registry": provider_uid}
assert result == expected_result
def test_get_prowler_provider_kwargs_image_provider_ignores_mutelist(self):
"""Test that Image provider does NOT receive mutelist_content.
Image provider uses Trivy's built-in logic, so it should not
receive mutelist_content even when a mutelist processor is configured.
"""
provider_uid = "ghcr.io"
secret_dict = {"registry_token": "test_token"}
secret_mock = MagicMock()
secret_mock.secret = secret_dict
mutelist_processor = MagicMock()
mutelist_processor.configuration = {"Mutelist": {"key": "value"}}
provider = MagicMock()
provider.provider = Provider.ProviderChoices.IMAGE.value
provider.secret = secret_mock
provider.uid = provider_uid
result = get_prowler_provider_kwargs(provider, mutelist_processor)
assert "mutelist_content" not in result
expected_result = {
"registry": provider_uid,
"registry_token": "test_token",
}
assert result == expected_result
def test_get_prowler_provider_kwargs_unsupported_provider(self):
# Setup
provider_uid = "provider_uid"
api/src/backend/api/tests/test_views.py
+305 -69
View File
@@ -1179,6 +1179,26 @@ class TestProviderViewSet:
"uid": "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4",
"alias": "Cloudflare Account",
},
{
"provider": "image",
"uid": "ghcr.io",
"alias": "GitHub Registry",
},
{
"provider": "image",
"uid": "docker.io",
"alias": "Docker Hub",
},
{
"provider": "image",
"uid": "123456789012.dkr.ecr.us-east-1.amazonaws.com",
"alias": "ECR",
},
{
"provider": "image",
"uid": "https://registry.example.com:5000",
"alias": "Private",
},
]
),
)
@@ -1598,6 +1618,26 @@ class TestProviderViewSet:
"cloudflare-uid",
"uid",
),
# Image UID validation - too short (below min_length)
(
{
"provider": "image",
"uid": "ab",
"alias": "test",
},
"min_length",
"uid",
),
# Image UID validation - invalid characters (space)
(
{
"provider": "image",
"uid": "not valid!",
"alias": "test",
},
"image-uid",
"uid",
),
]
),
)
@@ -1753,6 +1793,38 @@ class TestProviderViewSet:
assert "Content-Location" in response.headers
assert response.headers["Content-Location"] == f"/api/v1/tasks/{task_mock.id}"
@patch("api.v1.views.Task.objects.get")
@patch("api.v1.views.check_provider_connection_task.delay")
def test_providers_connection_image(
self,
mock_provider_connection,
mock_task_get,
authenticated_client,
providers_fixture,
tasks_fixture,
):
prowler_task = tasks_fixture[0]
task_mock = Mock()
task_mock.id = prowler_task.id
task_mock.status = "PENDING"
mock_provider_connection.return_value = task_mock
mock_task_get.return_value = prowler_task
image_provider = providers_fixture[10]
assert image_provider.provider == "image"
assert image_provider.connected is None
assert image_provider.connection_last_checked_at is None
response = authenticated_client.post(
reverse("provider-connection", kwargs={"pk": image_provider.id})
)
assert response.status_code == status.HTTP_202_ACCEPTED
mock_provider_connection.assert_called_once_with(
provider_id=str(image_provider.id), tenant_id=ANY
)
assert "Content-Location" in response.headers
assert response.headers["Content-Location"] == f"/api/v1/tasks/{task_mock.id}"
def test_providers_connection_invalid_provider(
self, authenticated_client, providers_fixture
):
@@ -1775,17 +1847,17 @@ class TestProviderViewSet:
),
("alias", "aws_testing_1", 1),
("alias.icontains", "aws", 2),
("inserted_at", TODAY, 10),
("inserted_at", TODAY, 11),
(
"inserted_at.gte",
"2024-01-01",
10,
11,
),
("inserted_at.lte", "2024-01-01", 0),
(
"updated_at.gte",
"2024-01-01",
10,
11,
),
("updated_at.lte", "2024-01-01", 0),
]
@@ -2392,6 +2464,39 @@ class TestProviderSecretViewSet:
"api_email": "user@example.com",
},
),
# Image with Docker login
(
Provider.ProviderChoices.IMAGE.value,
ProviderSecret.TypeChoices.STATIC,
{
"registry_username": "user",
"registry_password": "pass",
},
),
# Image with token
(
Provider.ProviderChoices.IMAGE.value,
ProviderSecret.TypeChoices.STATIC,
{
"registry_token": "ghp_abc123",
},
),
# Image with no auth + filters
(
Provider.ProviderChoices.IMAGE.value,
ProviderSecret.TypeChoices.STATIC,
{
"image_filter": "my-app.*",
"tag_filter": "v[0-9]+",
"max_images": 50,
},
),
# Image with empty secret (public registry)
(
Provider.ProviderChoices.IMAGE.value,
ProviderSecret.TypeChoices.STATIC,
{},
),
],
)
def test_provider_secrets_create_valid(
@@ -3830,6 +3935,7 @@ class TestAttackPathsScanViewSet:
AttackPathsQueryDefinition(
id="aws-rds",
name="RDS inventory",
short_description="List account RDS assets.",
description="List account RDS assets",
provider=provider.provider,
cypher="MATCH (n) RETURN n",
@@ -3887,11 +3993,11 @@ class TestAttackPathsScanViewSet:
attack_paths_scan = create_attack_paths_scan(
provider,
scan=scans_fixture[0],
graph_database="tenant-db",
)
query_definition = AttackPathsQueryDefinition(
id="aws-rds",
name="RDS inventory",
short_description="List account RDS assets.",
description="List account RDS assets",
provider=provider.provider,
cypher="MATCH (n) RETURN n",
@@ -3917,10 +4023,16 @@ class TestAttackPathsScanViewSet:
],
}
expected_db_name = f"db-tenant-{attack_paths_scan.provider.tenant_id}"
with (
patch(
"api.v1.views.get_query_by_id", return_value=query_definition
) as mock_get_query,
patch(
"api.v1.views.graph_database.get_database_name",
return_value=expected_db_name,
) as mock_get_db_name,
patch(
"api.v1.views.attack_paths_views_helpers.prepare_query_parameters",
return_value=prepared_parameters,
@@ -3942,17 +4054,18 @@ class TestAttackPathsScanViewSet:
assert response.status_code == status.HTTP_200_OK
mock_get_query.assert_called_once_with("aws-rds")
mock_get_db_name.assert_called_once_with(attack_paths_scan.provider.tenant_id)
mock_prepare.assert_called_once_with(
query_definition,
{},
attack_paths_scan.provider.uid,
)
mock_execute.assert_called_once_with(
attack_paths_scan,
expected_db_name,
query_definition,
prepared_parameters,
)
mock_clear_cache.assert_called_once_with(attack_paths_scan.graph_database)
mock_clear_cache.assert_called_once_with(expected_db_name)
result = response.json()["data"]
attributes = result["attributes"]
assert attributes["nodes"] == graph_payload["nodes"]
@@ -3983,31 +4096,6 @@ class TestAttackPathsScanViewSet:
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert "must be completed" in response.json()["errors"][0]["detail"]
def test_run_attack_paths_query_requires_graph_database(
self,
authenticated_client,
providers_fixture,
scans_fixture,
create_attack_paths_scan,
):
provider = providers_fixture[0]
attack_paths_scan = create_attack_paths_scan(
provider,
scan=scans_fixture[0],
graph_database=None,
)
response = authenticated_client.post(
reverse(
"attack-paths-scans-queries-run", kwargs={"pk": attack_paths_scan.id}
),
data=self._run_payload(),
content_type=API_JSON_CONTENT_TYPE,
)
assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
assert "does not reference a graph database" in str(response.json())
def test_run_attack_paths_query_unknown_query(
self,
authenticated_client,
@@ -4049,6 +4137,7 @@ class TestAttackPathsScanViewSet:
query_definition = AttackPathsQueryDefinition(
id="aws-empty",
name="empty",
short_description="",
description="",
provider=provider.provider,
cypher="MATCH (n) RETURN n",
@@ -10841,25 +10930,20 @@ class TestTenantFinishACSView:
assert "sso_saml_failed=true" in response.url
def test_dispatch_skips_role_mapping_when_single_manage_account_user(
self, create_test_user, tenants_fixture, saml_setup, settings, monkeypatch
self,
create_test_user,
tenants_fixture,
admin_role_fixture,
saml_setup,
settings,
monkeypatch,
):
"""Test that role mapping is skipped when tenant has only one user with MANAGE_ACCOUNT role"""
monkeypatch.setenv("SAML_SSO_CALLBACK_URL", "http://localhost/sso-complete")
user = create_test_user
tenant = tenants_fixture[0]
# Create a single role with manage_account=True for the user
admin_role = Role.objects.using(MainRouter.admin_db).create(
name="admin",
tenant=tenant,
manage_account=True,
manage_users=True,
manage_billing=True,
manage_providers=True,
manage_integrations=True,
manage_scans=True,
unlimited_visibility=True,
)
admin_role = admin_role_fixture
UserRoleRelationship.objects.using(MainRouter.admin_db).create(
user=user, role=admin_role, tenant_id=tenant.id
)
@@ -10930,35 +11014,26 @@ class TestTenantFinishACSView:
.exists()
)
def test_dispatch_applies_role_mapping_when_multiple_manage_account_users(
self, create_test_user, tenants_fixture, saml_setup, settings, monkeypatch
def test_dispatch_skips_role_mapping_when_last_manage_account_user_maps_to_existing_role(
self,
create_test_user,
tenants_fixture,
admin_role_fixture,
roles_fixture,
saml_setup,
settings,
monkeypatch,
):
"""Test that role mapping is applied when tenant has multiple users with MANAGE_ACCOUNT role"""
"""Test that role mapping is skipped when it would remove the last MANAGE_ACCOUNT user"""
monkeypatch.setenv("SAML_SSO_CALLBACK_URL", "http://localhost/sso-complete")
user = create_test_user
tenant = tenants_fixture[0]
# Create a second user with manage_account=True
second_admin = User.objects.using(MainRouter.admin_db).create(
email="admin2@prowler.com", name="Second Admin"
)
admin_role = Role.objects.using(MainRouter.admin_db).create(
name="admin",
tenant=tenant,
manage_account=True,
manage_users=True,
manage_billing=True,
manage_providers=True,
manage_integrations=True,
manage_scans=True,
unlimited_visibility=True,
)
admin_role = admin_role_fixture
viewer_role = roles_fixture[3]
UserRoleRelationship.objects.using(MainRouter.admin_db).create(
user=user, role=admin_role, tenant_id=tenant.id
)
UserRoleRelationship.objects.using(MainRouter.admin_db).create(
user=second_admin, role=admin_role, tenant_id=tenant.id
)
social_account = SocialAccount(
user=user,
@@ -10967,7 +11042,7 @@ class TestTenantFinishACSView:
"firstName": ["John"],
"lastName": ["Doe"],
"organization": ["testing_company"],
"userType": ["viewer"], # This SHOULD be applied
"userType": [viewer_role.name],
},
)
@@ -11005,10 +11080,91 @@ class TestTenantFinishACSView:
assert response.status_code == 302
# Verify the viewer role was created and assigned (role mapping was applied)
viewer_role = Role.objects.using(MainRouter.admin_db).get(
name="viewer", tenant=tenant
assert (
UserRoleRelationship.objects.using(MainRouter.admin_db)
.filter(user=user, role=admin_role, tenant_id=tenant.id)
.exists()
)
assert not (
UserRoleRelationship.objects.using(MainRouter.admin_db)
.filter(user=user, role=viewer_role, tenant_id=tenant.id)
.exists()
)
def test_dispatch_applies_role_mapping_when_multiple_manage_account_users(
self,
create_test_user,
tenants_fixture,
admin_role_fixture,
roles_fixture,
saml_setup,
settings,
monkeypatch,
):
"""Test that role mapping is applied when tenant has multiple users with MANAGE_ACCOUNT role"""
monkeypatch.setenv("SAML_SSO_CALLBACK_URL", "http://localhost/sso-complete")
user = create_test_user
tenant = tenants_fixture[0]
# Create a second user with manage_account=True
second_admin = User.objects.using(MainRouter.admin_db).create(
email="admin2@prowler.com", name="Second Admin"
)
admin_role = admin_role_fixture
viewer_role = roles_fixture[3]
UserRoleRelationship.objects.using(MainRouter.admin_db).create(
user=user, role=admin_role, tenant_id=tenant.id
)
UserRoleRelationship.objects.using(MainRouter.admin_db).create(
user=second_admin, role=admin_role, tenant_id=tenant.id
)
social_account = SocialAccount(
user=user,
provider="saml",
extra_data={
"firstName": ["John"],
"lastName": ["Doe"],
"organization": ["testing_company"],
"userType": [viewer_role.name], # This SHOULD be applied
},
)
request = RequestFactory().get(
reverse("saml_finish_acs", kwargs={"organization_slug": "testtenant"})
)
request.user = user
request.session = {}
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=tenant.id)
mock_saml_config_get.return_value = MagicMock()
mock_user_get.return_value = user
view = TenantFinishACSView.as_view()
response = view(request, organization_slug="testtenant")
assert response.status_code == 302
# Verify the viewer role was assigned (role mapping was applied)
assert (
UserRoleRelationship.objects.using(MainRouter.admin_db)
.filter(user=user, role=viewer_role, tenant_id=tenant.id)
@@ -11022,6 +11178,86 @@ class TestTenantFinishACSView:
.exists()
)
def test_dispatch_applies_role_mapping_for_non_admin_user_with_single_admin(
self,
create_test_user,
tenants_fixture,
admin_role_fixture,
roles_fixture,
saml_setup,
settings,
monkeypatch,
):
"""Test that role mapping is applied for a non-admin user when a single admin exists"""
monkeypatch.setenv("SAML_SSO_CALLBACK_URL", "http://localhost/sso-complete")
admin_user = create_test_user
tenant = tenants_fixture[0]
non_admin_user = User.objects.using(MainRouter.admin_db).create(
email="viewer@prowler.com", name="Viewer"
)
admin_role = admin_role_fixture
viewer_role = roles_fixture[3]
UserRoleRelationship.objects.using(MainRouter.admin_db).create(
user=admin_user, role=admin_role, tenant_id=tenant.id
)
social_account = SocialAccount(
user=non_admin_user,
provider="saml",
extra_data={
"firstName": ["Jane"],
"lastName": ["Doe"],
"organization": ["testing_company"],
"userType": [viewer_role.name],
},
)
request = RequestFactory().get(
reverse("saml_finish_acs", kwargs={"organization_slug": "testtenant"})
)
request.user = non_admin_user
request.session = {}
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=tenant.id)
mock_saml_config_get.return_value = MagicMock()
mock_user_get.return_value = non_admin_user
view = TenantFinishACSView.as_view()
response = view(request, organization_slug="testtenant")
assert response.status_code == 302
assert (
UserRoleRelationship.objects.using(MainRouter.admin_db)
.filter(user=non_admin_user, role=viewer_role, tenant_id=tenant.id)
.exists()
)
assert (
UserRoleRelationship.objects.using(MainRouter.admin_db)
.filter(user=admin_user, role=admin_role, tenant_id=tenant.id)
.exists()
)
@pytest.mark.django_db
class TestLighthouseConfigViewSet:
api/src/backend/api/utils.py
+45 -4
View File
@@ -28,6 +28,7 @@ if TYPE_CHECKING:
from prowler.providers.gcp.gcp_provider import GcpProvider
from prowler.providers.github.github_provider import GithubProvider
from prowler.providers.iac.iac_provider import IacProvider
from prowler.providers.image.image_provider import ImageProvider
from prowler.providers.kubernetes.kubernetes_provider import KubernetesProvider
from prowler.providers.m365.m365_provider import M365Provider
from prowler.providers.mongodbatlas.mongodbatlas_provider import (
@@ -78,9 +79,11 @@ def return_prowler_provider(
AlibabacloudProvider
| AwsProvider
| AzureProvider
| CloudflareProvider
| GcpProvider
| GithubProvider
| IacProvider
| ImageProvider
| KubernetesProvider
| M365Provider
| MongodbatlasProvider
@@ -92,7 +95,7 @@ def return_prowler_provider(
provider (Provider): The provider object containing the provider type and associated secrets.
Returns:
AlibabacloudProvider | AwsProvider | AzureProvider | CloudflareProvider | GcpProvider | GithubProvider | IacProvider | KubernetesProvider | M365Provider | MongodbatlasProvider | OraclecloudProvider: The corresponding provider class.
AlibabacloudProvider | AwsProvider | AzureProvider | CloudflareProvider | GcpProvider | GithubProvider | IacProvider | ImageProvider | KubernetesProvider | M365Provider | MongodbatlasProvider | OraclecloudProvider: The corresponding provider class.
Raises:
ValueError: If the provider type specified in `provider.provider` is not supported.
@@ -152,6 +155,10 @@ def return_prowler_provider(
)
prowler_provider = CloudflareProvider
case Provider.ProviderChoices.IMAGE.value:
from prowler.providers.image.image_provider import ImageProvider
prowler_provider = ImageProvider
case _:
raise ValueError(f"Provider type {provider.provider} not supported")
return prowler_provider
@@ -208,11 +215,30 @@ def get_prowler_provider_kwargs(
**prowler_provider_kwargs,
"filter_accounts": [provider.uid],
}
elif provider.provider == Provider.ProviderChoices.IMAGE.value:
prowler_provider_kwargs = {
"registry": provider.uid,
}
secret = provider.secret.secret
for key in (
"registry_username",
"registry_password",
"registry_token",
"image_filter",
"tag_filter",
):
if key in secret:
prowler_provider_kwargs[key] = secret[key]
if "max_images" in secret:
prowler_provider_kwargs["max_images"] = int(secret["max_images"])
if mutelist_processor:
mutelist_content = mutelist_processor.configuration.get("Mutelist", {})
# IaC provider doesn't support mutelist (uses Trivy's built-in logic)
if mutelist_content and provider.provider != Provider.ProviderChoices.IAC.value:
# IaC and Image providers don't support mutelist (Trivy handles its own logic)
if mutelist_content and provider.provider not in (
Provider.ProviderChoices.IAC.value,
Provider.ProviderChoices.IMAGE.value,
):
prowler_provider_kwargs["mutelist_content"] = mutelist_content
return prowler_provider_kwargs
@@ -229,6 +255,7 @@ def initialize_prowler_provider(
| GcpProvider
| GithubProvider
| IacProvider
| ImageProvider
| KubernetesProvider
| M365Provider
| MongodbatlasProvider
@@ -241,7 +268,7 @@ def initialize_prowler_provider(
mutelist_processor (Processor): The mutelist processor object containing the mutelist configuration.
Returns:
AlibabacloudProvider | AwsProvider | AzureProvider | CloudflareProvider | GcpProvider | GithubProvider | IacProvider | KubernetesProvider | M365Provider | MongodbatlasProvider | OraclecloudProvider: An instance of the corresponding provider class
AlibabacloudProvider | AwsProvider | AzureProvider | CloudflareProvider | GcpProvider | GithubProvider | IacProvider | ImageProvider | KubernetesProvider | M365Provider | MongodbatlasProvider | OraclecloudProvider: An instance of the corresponding provider class
initialized with the provider's secrets.
"""
prowler_provider = return_prowler_provider(provider)
@@ -276,6 +303,20 @@ def prowler_provider_connection_test(provider: Provider) -> Connection:
if "access_token" in prowler_provider_kwargs:
iac_test_kwargs["access_token"] = prowler_provider_kwargs["access_token"]
return prowler_provider.test_connection(**iac_test_kwargs)
elif provider.provider == Provider.ProviderChoices.IMAGE.value:
from prowler.providers.image.lib.registry.factory import create_registry_adapter
try:
adapter = create_registry_adapter(
registry_url=provider.uid,
username=prowler_provider_kwargs.get("registry_username"),
password=prowler_provider_kwargs.get("registry_password"),
token=prowler_provider_kwargs.get("registry_token"),
)
adapter.list_repositories()
return Connection(is_connected=True, error=None)
except Exception as e:
return Connection(is_connected=False, error=str(e))
else:
return prowler_provider.test_connection(
**prowler_provider_kwargs,
api/src/backend/api/v1/serializer_utils/providers.py
+31
View File
@@ -373,6 +373,37 @@ from rest_framework_json_api import serializers
},
"required": ["api_key", "api_email"],
},
{
"type": "object",
"title": "Image Registry Credentials",
"properties": {
"registry_username": {
"type": "string",
"description": "Username for Docker login authentication.",
},
"registry_password": {
"type": "string",
"description": "Password for Docker login authentication.",
},
"registry_token": {
"type": "string",
"description": "Bearer token for registry authentication.",
},
"image_filter": {
"type": "string",
"description": "Regex pattern to filter repository names during registry enumeration.",
},
"tag_filter": {
"type": "string",
"description": "Regex pattern to filter image tags during registry enumeration.",
},
"max_images": {
"type": "integer",
"minimum": 0,
"description": "Maximum number of images to scan (0 = unlimited).",
},
},
},
]
}
)
api/src/backend/api/v1/serializers.py
+43
View File
@@ -1176,6 +1176,14 @@ class AttackPathsScanSerializer(RLSSerializer):
return provider.uid if provider else None
class AttackPathsQueryAttributionSerializer(BaseSerializerV1):
text = serializers.CharField()
link = serializers.CharField()
class JSONAPIMeta:
resource_name = "attack-paths-query-attributions"
class AttackPathsQueryParameterSerializer(BaseSerializerV1):
name = serializers.CharField()
label = serializers.CharField()
@@ -1190,7 +1198,9 @@ class AttackPathsQueryParameterSerializer(BaseSerializerV1):
class AttackPathsQuerySerializer(BaseSerializerV1):
id = serializers.CharField()
name = serializers.CharField()
short_description = serializers.CharField()
description = serializers.CharField()
attribution = AttackPathsQueryAttributionSerializer(allow_null=True, required=False)
provider = serializers.CharField()
parameters = AttackPathsQueryParameterSerializer(many=True)
@@ -1515,6 +1525,8 @@ class BaseWriteProviderSecretSerializer(BaseWriteSerializer):
"or both 'api_key' and 'api_email'."
}
)
elif provider_type == Provider.ProviderChoices.IMAGE.value:
serializer = ImageProviderSecret(data=secret)
else:
raise serializers.ValidationError(
{"provider": f"Provider type not supported {provider_type}"}
@@ -1653,6 +1665,37 @@ class IacProviderSecret(serializers.Serializer):
resource_name = "provider-secrets"
class ImageProviderSecret(serializers.Serializer):
registry_username = serializers.CharField(required=False)
registry_password = serializers.CharField(required=False)
registry_token = serializers.CharField(required=False)
image_filter = serializers.CharField(required=False)
tag_filter = serializers.CharField(required=False)
max_images = serializers.IntegerField(required=False, min_value=0)
def validate(self, attrs):
has_username = attrs.get("registry_username")
has_password = attrs.get("registry_password")
has_token = attrs.get("registry_token")
if (has_username or has_password) and has_token:
raise serializers.ValidationError(
"You cannot provide both registry_username/registry_password and registry_token."
)
if has_username and not has_password:
raise serializers.ValidationError(
"registry_password is required when registry_username is provided."
)
if has_password and not has_username:
raise serializers.ValidationError(
"registry_username is required when registry_password is provided."
)
return super().validate(attrs)
class Meta:
resource_name = "provider-secrets"
class OracleCloudProviderSecret(serializers.Serializer):
user = serializers.CharField()
fingerprint = serializers.CharField()
api/src/backend/api/v1/views.py
+33 -26
View File
@@ -392,7 +392,7 @@ class SchemaView(SpectacularAPIView):
def get(self, request, *args, **kwargs):
spectacular_settings.TITLE = "Prowler API"
spectacular_settings.VERSION = "1.19.0"
spectacular_settings.VERSION = "1.20.0"
spectacular_settings.DESCRIPTION = (
"Prowler API specification.\n\nThis file is auto-generated."
)
@@ -763,27 +763,40 @@ class TenantFinishACSView(FinishACSView):
.tenant
)
# Check if tenant has only one user with MANAGE_ACCOUNT role
users_with_manage_account = (
role_name = (
extra.get("userType", ["no_permissions"])[0].strip()
if extra.get("userType")
else "no_permissions"
)
role = (
Role.objects.using(MainRouter.admin_db)
.filter(name=role_name, tenant=tenant)
.first()
)
# Only skip mapping if it would remove the last MANAGE_ACCOUNT user
remaining_manage_account_users = (
UserRoleRelationship.objects.using(MainRouter.admin_db)
.filter(role__manage_account=True, tenant_id=tenant.id)
.exclude(user_id=user_id)
.values("user")
.distinct()
.count()
)
user_has_manage_account = (
UserRoleRelationship.objects.using(MainRouter.admin_db)
.filter(role__manage_account=True, tenant_id=tenant.id, user_id=user_id)
.exists()
)
role_manage_account = role.manage_account if role else False
would_remove_last_manage_account = (
user_has_manage_account
and remaining_manage_account_users == 0
and not role_manage_account
)
# Only apply role mapping from userType if tenant does NOT have exactly one user with MANAGE_ACCOUNT
if users_with_manage_account != 1:
role_name = (
extra.get("userType", ["no_permissions"])[0].strip()
if extra.get("userType")
else "no_permissions"
)
try:
role = Role.objects.using(MainRouter.admin_db).get(
name=role_name, tenant=tenant
)
except Role.DoesNotExist:
if not would_remove_last_manage_account:
if role is None:
role = Role.objects.using(MainRouter.admin_db).create(
name=role_name,
tenant=tenant,
@@ -2415,15 +2428,6 @@ class AttackPathsScanViewSet(BaseRLSViewSet):
}
)
if not attack_paths_scan.graph_database:
logger.error(
f"The Attack Paths Scan {attack_paths_scan.id} does not reference a graph database"
)
return Response(
{"detail": "The Attack Paths scan does not reference a graph database"},
status=status.HTTP_500_INTERNAL_SERVER_ERROR,
)
payload = attack_paths_views_helpers.normalize_run_payload(request.data)
serializer = AttackPathsQueryRunRequestSerializer(data=payload)
serializer.is_valid(raise_exception=True)
@@ -2437,6 +2441,9 @@ class AttackPathsScanViewSet(BaseRLSViewSet):
{"id": "Unknown Attack Paths query for the selected provider"}
)
database_name = graph_database.get_database_name(
attack_paths_scan.provider.tenant_id
)
parameters = attack_paths_views_helpers.prepare_query_parameters(
query_definition,
serializer.validated_data.get("parameters", {}),
@@ -2444,9 +2451,9 @@ class AttackPathsScanViewSet(BaseRLSViewSet):
)
graph = attack_paths_views_helpers.execute_attack_paths_query(
attack_paths_scan, query_definition, parameters
database_name, query_definition, parameters
)
graph_database.clear_cache(attack_paths_scan.graph_database)
graph_database.clear_cache(database_name)
status_code = status.HTTP_200_OK
if not graph.get("nodes"):
api/src/backend/conftest.py
+8 -2
View File
@@ -537,6 +537,12 @@ def providers_fixture(tenants_fixture):
alias="cloudflare_testing",
tenant_id=tenant.id,
)
provider11 = Provider.objects.create(
provider="image",
uid="ghcr.io",
alias="image_testing",
tenant_id=tenant.id,
)
return (
provider1,
@@ -549,6 +555,7 @@ def providers_fixture(tenants_fixture):
provider8,
provider9,
provider10,
provider11,
)
@@ -1618,7 +1625,6 @@ def create_attack_paths_scan():
scan=None,
state=StateChoices.COMPLETED,
progress=0,
graph_database="tenant-db",
**extra_fields,
):
scan_instance = scan or Scan.objects.create(
@@ -1635,7 +1641,6 @@ def create_attack_paths_scan():
"scan": scan_instance,
"state": state,
"progress": progress,
"graph_database": graph_database,
}
payload.update(extra_fields)
@@ -1663,6 +1668,7 @@ def attack_paths_query_definition_factory():
definition_payload = {
"id": "aws-test",
"name": "Attack Paths Test Query",
"short_description": "Synthetic short description for tests.",
"description": "Synthetic Attack Paths definition for tests.",
"provider": "aws",
"cypher": "RETURN 1",
api/src/backend/tasks/jobs/export.py
+14
View File
@@ -35,6 +35,11 @@ from prowler.lib.outputs.compliance.cis.cis_github import GithubCIS
from prowler.lib.outputs.compliance.cis.cis_kubernetes import KubernetesCIS
from prowler.lib.outputs.compliance.cis.cis_m365 import M365CIS
from prowler.lib.outputs.compliance.cis.cis_oraclecloud import OracleCloudCIS
from prowler.lib.outputs.compliance.csa.csa_alibabacloud import AlibabaCloudCSA
from prowler.lib.outputs.compliance.csa.csa_aws import AWSCSA
from prowler.lib.outputs.compliance.csa.csa_azure import AzureCSA
from prowler.lib.outputs.compliance.csa.csa_gcp import GCPCSA
from prowler.lib.outputs.compliance.csa.csa_oraclecloud import OracleCloudCSA
from prowler.lib.outputs.compliance.ens.ens_aws import AWSENS
from prowler.lib.outputs.compliance.ens.ens_azure import AzureENS
from prowler.lib.outputs.compliance.ens.ens_gcp import GCPENS
@@ -90,6 +95,7 @@ COMPLIANCE_CLASS_MAP = {
(lambda name: name == "prowler_threatscore_aws", ProwlerThreatScoreAWS),
(lambda name: name == "ccc_aws", CCC_AWS),
(lambda name: name.startswith("c5_"), AWSC5),
(lambda name: name.startswith("csa_"), AWSCSA),
],
"azure": [
(lambda name: name.startswith("cis_"), AzureCIS),
@@ -99,6 +105,7 @@ COMPLIANCE_CLASS_MAP = {
(lambda name: name == "ccc_azure", CCC_Azure),
(lambda name: name == "prowler_threatscore_azure", ProwlerThreatScoreAzure),
(lambda name: name == "c5_azure", AzureC5),
(lambda name: name.startswith("csa_"), AzureCSA),
],
"gcp": [
(lambda name: name.startswith("cis_"), GCPCIS),
@@ -108,6 +115,7 @@ COMPLIANCE_CLASS_MAP = {
(lambda name: name == "prowler_threatscore_gcp", ProwlerThreatScoreGCP),
(lambda name: name == "ccc_gcp", CCC_GCP),
(lambda name: name == "c5_gcp", GCPC5),
(lambda name: name.startswith("csa_"), GCPCSA),
],
"kubernetes": [
(lambda name: name.startswith("cis_"), KubernetesCIS),
@@ -129,11 +137,17 @@ COMPLIANCE_CLASS_MAP = {
# IaC provider doesn't have specific compliance frameworks yet
# Trivy handles its own compliance checks
],
"image": [
# Image provider doesn't have specific compliance frameworks yet
# Trivy handles its own compliance checks
],
"oraclecloud": [
(lambda name: name.startswith("cis_"), OracleCloudCIS),
(lambda name: name.startswith("csa_"), OracleCloudCSA),
],
"alibabacloud": [
(lambda name: name.startswith("cis_"), AlibabaCloudCIS),
(lambda name: name.startswith("csa_"), AlibabaCloudCSA),
(
lambda name: name == "prowler_threatscore_alibabacloud",
ProwlerThreatScoreAlibaba,
api/src/backend/tasks/tasks.py
+35 -17
View File
@@ -133,13 +133,41 @@ def _perform_scan_complete_tasks(tenant_id: str, scan_id: str, provider_id: str)
scan_id (str): The ID of the scan that was performed.
provider_id (str): The primary key of the Provider instance that was scanned.
"""
chain(
create_compliance_requirements_task.si(tenant_id=tenant_id, scan_id=scan_id),
update_provider_compliance_scores_task.si(tenant_id=tenant_id, scan_id=scan_id),
).apply_async()
aggregate_attack_surface_task.apply_async(
kwargs={"tenant_id": tenant_id, "scan_id": scan_id}
with rls_transaction(tenant_id):
provider_type = Provider.objects.get(id=provider_id).provider
has_compliance = provider_type not in (
Provider.ProviderChoices.IAC.value,
Provider.ProviderChoices.IMAGE.value,
)
if has_compliance:
chain(
create_compliance_requirements_task.si(
tenant_id=tenant_id, scan_id=scan_id
),
update_provider_compliance_scores_task.si(
tenant_id=tenant_id, scan_id=scan_id
),
).apply_async()
aggregate_attack_surface_task.apply_async(
kwargs={"tenant_id": tenant_id, "scan_id": scan_id}
)
final_group_tasks = [
check_integrations_task.si(
tenant_id=tenant_id,
provider_id=provider_id,
scan_id=scan_id,
),
]
if has_compliance:
final_group_tasks.append(
generate_compliance_reports_task.si(
tenant_id=tenant_id, scan_id=scan_id, provider_id=provider_id
),
)
chain(
perform_scan_summary_task.si(tenant_id=tenant_id, scan_id=scan_id),
group(
@@ -148,17 +176,7 @@ def _perform_scan_complete_tasks(tenant_id: str, scan_id: str, provider_id: str)
scan_id=scan_id, provider_id=provider_id, tenant_id=tenant_id
),
),
group(
# Use optimized task that generates both reports with shared queries
generate_compliance_reports_task.si(
tenant_id=tenant_id, scan_id=scan_id, provider_id=provider_id
),
check_integrations_task.si(
tenant_id=tenant_id,
provider_id=provider_id,
scan_id=scan_id,
),
),
group(*final_group_tasks),
).apply_async()
if can_provider_run_attack_paths_scan(tenant_id, provider_id):