diff --git a/.env b/.env index 00d48ed6a2..3be5e93724 100644 --- a/.env +++ b/.env @@ -41,6 +41,25 @@ POSTGRES_DB=prowler_db # POSTGRES_REPLICA_MAX_ATTEMPTS=3 # POSTGRES_REPLICA_RETRY_BASE_DELAY=0.5 +# Neo4j auth +NEO4J_HOST=neo4j +NEO4J_PORT=7687 +NEO4J_USER=neo4j +NEO4J_PASSWORD=neo4j_password +# Neo4j settings +NEO4J_SERVER_MEMORY_PAGECACHE_SIZE=1G +NEO4J_SERVER_MEMORY_HEAP_INITIAL__SIZE=1G +NEO4J_SERVER_MEMORY_HEAP_MAX__SIZE=1G +NEO4J_POC_EXPORT_FILE_ENABLED=true +NEO4J_APOC_IMPORT_FILE_ENABLED=true +NEO4J_APOC_IMPORT_FILE_USE_NEO4J_CONFIG=true +NEO4J_PLUGINS=["apoc"] +NEO4J_DBMS_SECURITY_PROCEDURES_ALLOWLIST=apoc.* +NEO4J_DBMS_SECURITY_PROCEDURES_UNRESTRICTED=apoc.* +NEO4J_DBMS_CONNECTOR_BOLT_LISTEN_ADDRESS=0.0.0.0:7687 +# Neo4j Prowler settings +NEO4J_INSERT_BATCH_SIZE=500 + # Celery-Prowler task settings TASK_RETRY_DELAY_SECONDS=0.1 TASK_RETRY_ATTEMPTS=5 diff --git a/README.md b/README.md index 3519305248..dc3e74a57b 100644 --- a/README.md +++ b/README.md @@ -75,6 +75,23 @@ prowler dashboard ``` ![Prowler Dashboard](docs/images/products/dashboard.png) + +## Attack Paths + +Attack Paths automatically extends every completed AWS scan with a Neo4j graph that combines Cartography's cloud inventory with Prowler findings. The feature runs in the API worker after each scan and therefore requires: + +- An accessible Neo4j instance (the Docker Compose files already ships a `neo4j` service). +- The following environment variables so Django and Celery can connect: + + | Variable | Description | Default | + | --- | --- | --- | + | `NEO4J_HOST` | Hostname used by the API containers. | `neo4j` | + | `NEO4J_PORT` | Bolt port exposed by Neo4j. | `7687` | + | `NEO4J_USER` / `NEO4J_PASSWORD` | Credentials with rights to create per-tenant databases. | `neo4j` / `neo4j_password` | + +Every AWS provider scan will enqueue an Attack Paths ingestion job automatically. Other cloud providers will be added in future iterations. + + # Prowler at a Glance > [!Tip] > For the most accurate and up-to-date information about checks, services, frameworks, and categories, visit [**Prowler Hub**](https://hub.prowler.com). diff --git a/api/CHANGELOG.md b/api/CHANGELOG.md index 0a54260f21..e8ae6f8a1c 100644 --- a/api/CHANGELOG.md +++ b/api/CHANGELOG.md @@ -4,6 +4,9 @@ All notable changes to the **Prowler API** are documented in this file. ## [1.16.0] (Unreleased) +### Added +- Attack Paths backend support [(#9344)](https://github.com/prowler-cloud/prowler/pull/9344) + ### Changed - Restore the compliance overview endpoint's mandatory filters [(#9330)](https://github.com/prowler-cloud/prowler/pull/9330) diff --git a/api/Dockerfile b/api/Dockerfile index 2d7883a957..de7c84b077 100644 --- a/api/Dockerfile +++ b/api/Dockerfile @@ -2,6 +2,9 @@ FROM python:3.12.10-slim-bookworm AS build LABEL maintainer="https://github.com/prowler-cloud/api" +ARG CARTOGRAPHY_VERSION=0.117.0 +ENV CARTOGRAPHY_VERSION=${CARTOGRAPHY_VERSION} + ARG POWERSHELL_VERSION=7.5.0 ENV POWERSHELL_VERSION=${POWERSHELL_VERSION} @@ -79,6 +82,8 @@ ENV PATH="/home/prowler/.local/bin:$PATH" RUN poetry install --no-root && \ rm -rf ~/.cache/pip +RUN poetry run python -m pip install cartography==${CARTOGRAPHY_VERSION} + RUN poetry run python "$(poetry env info --path)/src/prowler/prowler/providers/m365/lib/powershell/m365_powershell.py" COPY src/backend/ ./backend/ diff --git a/api/poetry.lock b/api/poetry.lock index 3008e78ba7..a8af36f966 100644 --- a/api/poetry.lock +++ b/api/poetry.lock @@ -1,4 +1,4 @@ -# This file is automatically @generated by Poetry 2.1.3 and should not be changed by hand. +# This file is automatically @generated by Poetry 2.1.4 and should not be changed by hand. [[package]] name = "about-time" @@ -3992,6 +3992,26 @@ pyspark = ["pyspark (>=3.5.0)"] pyspark-connect = ["pyspark[connect] (>=3.5.0)"] sqlframe = ["sqlframe (>=3.22.0)"] +[[package]] +name = "neo4j" +version = "5.28.2" +description = "Neo4j Bolt driver for Python" +optional = false +python-versions = ">=3.7" +groups = ["main"] +files = [ + {file = "neo4j-5.28.2-py3-none-any.whl", hash = "sha256:5c53b5c3eee6dee7e920c9724391aa38d7135a651e71b766da00533b92a91a94"}, + {file = "neo4j-5.28.2.tar.gz", hash = "sha256:7d38e27e4f987a45cc9052500c6ee27325cb23dae6509037fe31dd7ddaed70c7"}, +] + +[package.dependencies] +pytz = "*" + +[package.extras] +numpy = ["numpy (>=1.7.0,<3.0.0)"] +pandas = ["numpy (>=1.7.0,<3.0.0)", "pandas (>=1.1.0,<3.0.0)"] +pyarrow = ["pyarrow (>=1.0.0)"] + [[package]] name = "nest-asyncio" version = "1.6.0" @@ -6860,4 +6880,4 @@ type = ["pytest-mypy"] [metadata] lock-version = "2.1" python-versions = ">=3.11,<3.13" -content-hash = "943e2cd6b87229704550d4e140b36509fb9f58896ebb5834b9fbabe28a9ee92f" +content-hash = "18f0a242102d6583c645399c742040e4a324c08c0d16a76bf843700f8ce77f84" diff --git a/api/pyproject.toml b/api/pyproject.toml index ffc8413489..e112745377 100644 --- a/api/pyproject.toml +++ b/api/pyproject.toml @@ -35,7 +35,8 @@ dependencies = [ "markdown (>=3.9,<4.0)", "drf-simple-apikey (==2.2.1)", "matplotlib (>=3.10.6,<4.0.0)", - "reportlab (>=4.4.4,<5.0.0)" + "reportlab (>=4.4.4,<5.0.0)", + "neo4j (<6.0.0)", ] description = "Prowler's API (Django/DRF)" license = "Apache-2.0" diff --git a/api/src/backend/api/apps.py b/api/src/backend/api/apps.py index add97cf376..6c7eadb631 100644 --- a/api/src/backend/api/apps.py +++ b/api/src/backend/api/apps.py @@ -1,4 +1,5 @@ import logging +import atexit import os import sys from pathlib import Path @@ -30,6 +31,7 @@ class ApiConfig(AppConfig): def ready(self): from api import schema_extensions # noqa: F401 from api import signals # noqa: F401 + from api.attack_paths import database as graph_database from api.compliance import load_prowler_compliance # Generate required cryptographic keys if not present, but only if: @@ -39,6 +41,10 @@ class ApiConfig(AppConfig): if "manage.py" not in sys.argv or os.environ.get("RUN_MAIN"): self._ensure_crypto_keys() + if not getattr(settings, "TESTING", False): + graph_database.init_driver() + atexit.register(graph_database.close_driver) + load_prowler_compliance() def _ensure_crypto_keys(self): @@ -54,7 +60,7 @@ class ApiConfig(AppConfig): global _keys_initialized # Skip key generation if running tests - if hasattr(settings, "TESTING") and settings.TESTING: + if getattr(settings, "TESTING", False): return # Skip if already initialized in this process diff --git a/api/src/backend/api/attack_paths/__init__.py b/api/src/backend/api/attack_paths/__init__.py new file mode 100644 index 0000000000..2c3ea4c5d8 --- /dev/null +++ b/api/src/backend/api/attack_paths/__init__.py @@ -0,0 +1,13 @@ +from api.attack_paths.query_definitions import ( + AttackPathsQueryDefinition, + AttackPathsQueryParameterDefinition, + get_queries_for_provider, + get_query_by_id, +) + +__all__ = [ + "AttackPathsQueryDefinition", + "AttackPathsQueryParameterDefinition", + "get_queries_for_provider", + "get_query_by_id", +] diff --git a/api/src/backend/api/attack_paths/database.py b/api/src/backend/api/attack_paths/database.py new file mode 100644 index 0000000000..da8217fa55 --- /dev/null +++ b/api/src/backend/api/attack_paths/database.py @@ -0,0 +1,131 @@ +import logging +import threading + +from contextlib import contextmanager +from typing import Iterator +from uuid import UUID + +import neo4j + +from django.conf import settings + +import neo4j.exceptions + +# Without this Celery goes crazy with Neo4j logging +logging.getLogger("neo4j").setLevel(logging.ERROR) +logging.getLogger("neo4j").propagate = False + +# Module-level process-wide driver singleton +_driver: neo4j.Driver | None = None +_lock = threading.Lock() + +# Base Neo4j functions + + +def get_uri() -> str: + host = settings.DATABASES["neo4j"]["HOST"] + port = settings.DATABASES["neo4j"]["PORT"] + return f"bolt://{host}:{port}" + + +def init_driver() -> neo4j.Driver: + global _driver + if _driver is not None: + return _driver + + with _lock: + if _driver is None: + uri = get_uri() + config = settings.DATABASES["neo4j"] + + _driver = neo4j.GraphDatabase.driver( + uri, auth=(config["USER"], config["PASSWORD"]) + ) + _driver.verify_connectivity() + + return _driver + + +def get_driver() -> neo4j.Driver: + return init_driver() + + +def close_driver() -> None: # TODO: Use it + global _driver + with _lock: + if _driver is not None: + try: + _driver.close() + + finally: + _driver = None + + +@contextmanager +def get_session(database: str | None = None) -> Iterator[neo4j.Session]: + try: + with get_driver().session(database=database) as session: + yield session + + except neo4j.exceptions.Neo4jError as exc: + raise GraphDatabaseQueryException(message=exc.message, code=exc.code) + + +def create_database(database: str) -> None: + query = "CREATE DATABASE $database IF NOT EXISTS" + parameters = {"database": database} + + with get_session() as session: + session.run(query, parameters) + + +def drop_database(database: str) -> None: + query = f"DROP DATABASE `{database}` IF EXISTS DESTROY DATA" + + with get_session() as session: + session.run(query) + + +def drop_subgraph(database: str, root_node_label: str, root_node_id: str) -> int: + query = """ + MATCH (a:__ROOT_NODE_LABEL__ {id: $root_node_id}) + CALL apoc.path.subgraphNodes(a, {}) + YIELD node + DETACH DELETE node + RETURN COUNT(node) AS deleted_nodes_count + """.replace("__ROOT_NODE_LABEL__", root_node_label) + parameters = {"root_node_id": root_node_id} + + with get_session(database) as session: + result = session.run(query, parameters) + + try: + return result.single()["deleted_nodes_count"] + + except neo4j.exceptions.ResultConsumedError: + return 0 # As there are no nodes to delete, the result is empty + + +# Neo4j functions related to Prowler + Cartography +DATABASE_NAME_TEMPLATE = "db-{attack_paths_scan_id}" + + +def get_database_name(attack_paths_scan_id: UUID) -> str: + attack_paths_scan_id_str = str(attack_paths_scan_id).lower() + return DATABASE_NAME_TEMPLATE.format(attack_paths_scan_id=attack_paths_scan_id_str) + + +# Exceptions + + +class GraphDatabaseQueryException(Exception): + def __init__(self, message: str, code: str | None = None) -> None: + super().__init__(message) + self.message = message + self.code = code + + def __str__(self) -> str: + if self.code: + return f"{self.code}: {self.message}" + + return self.message diff --git a/api/src/backend/api/attack_paths/query_definitions.py b/api/src/backend/api/attack_paths/query_definitions.py new file mode 100644 index 0000000000..8a65440d6c --- /dev/null +++ b/api/src/backend/api/attack_paths/query_definitions.py @@ -0,0 +1,346 @@ +from dataclasses import dataclass, field + + +# Dataclases for handling API's Attack Path query definitions and their parameters +@dataclass +class AttackPathsQueryParameterDefinition: + """ + Metadata describing a parameter that must be provided to an Attack Paths query. + """ + + name: str + label: str + data_type: str = "string" + cast: type = str + description: str | None = None + placeholder: str | None = None + + +@dataclass +class AttackPathsQueryDefinition: + """ + Immutable representation of an Attack Path query. + """ + + id: str + name: str + description: str + provider: str + cypher: str + parameters: list[AttackPathsQueryParameterDefinition] = field(default_factory=list) + + +# Accessor functions for API's Attack Paths query definitions +def get_queries_for_provider(provider: str) -> list[AttackPathsQueryDefinition]: + return _QUERY_DEFINITIONS.get(provider, []) + + +def get_query_by_id(query_id: str) -> AttackPathsQueryDefinition | None: + return _QUERIES_BY_ID.get(query_id) + + +# API's Attack Paths query definitions +_QUERY_DEFINITIONS: dict[str, list[AttackPathsQueryDefinition]] = { + "aws": [ + AttackPathsQueryDefinition( + id="aws-internet-exposed-ec2-sensitive-s3-access", + name="Identify internet-exposed EC2 instances with sensitive S3 access", + description="Detect EC2 instances with SSH exposed to the internet that can assume higher-privileged roles to read tagged sensitive S3 buckets despite bucket-level public access blocks.", + provider="aws", + cypher=""" + CALL apoc.create.vNode(['Internet'], {id: 'Internet', name: 'Internet'}) + YIELD node AS internet + + MATCH path_s3 = (aws:AWSAccount {id: $provider_uid})--(s3:S3Bucket)--(t:AWSTag) + WHERE toLower(t.key) = toLower($tag_key) AND toLower(t.value) = toLower($tag_value) + + MATCH path_ec2 = (aws)--(ec2:EC2Instance)--(sg:EC2SecurityGroup)--(ipi:IpPermissionInbound) + WHERE ec2.exposed_internet = true + AND ipi.toport = 22 + + MATCH path_role = (r:AWSRole)--(pol:AWSPolicy)--(stmt:AWSPolicyStatement) + WHERE ANY(x IN stmt.resource WHERE x CONTAINS s3.name) + AND ANY(x IN stmt.action WHERE toLower(x) =~ 's3:(listbucket|getobject).*') + + MATCH path_assume_role = (ec2)-[p:STS_ASSUMEROLE_ALLOW*1..9]-(r:AWSRole) + + CALL apoc.create.vRelationship(internet, 'CAN_ACCESS', {}, ec2) + YIELD rel AS can_access + + UNWIND nodes(path_s3) + nodes(path_ec2) + nodes(path_role) + nodes(path_assume_role) as n + OPTIONAL MATCH (n)-[pfr]-(pf:ProwlerFinding) + WHERE pf.status = 'FAIL' + + RETURN path_s3, path_ec2, path_role, path_assume_role, collect(DISTINCT pf) as dpf, collect(DISTINCT pfr) as dpfr, internet, can_access + """, + parameters=[ + AttackPathsQueryParameterDefinition( + name="tag_key", + label="Tag key", + description="Tag key to filter the S3 bucket, e.g. DataClassification.", + placeholder="DataClassification", + ), + AttackPathsQueryParameterDefinition( + name="tag_value", + label="Tag value", + description="Tag value to filter the S3 bucket, e.g. Sensitive.", + placeholder="Sensitive", + ), + ], + ), + AttackPathsQueryDefinition( + id="aws-rds-instances", + name="Identify provisioned RDS instances", + description="List the selected AWS account alongside the RDS instances it owns.", + provider="aws", + cypher=""" + MATCH path = (aws:AWSAccount {id: $provider_uid})--(rds:RDSInstance) + + UNWIND nodes(path) as n + OPTIONAL MATCH (n)-[pfr]-(pf:ProwlerFinding) + WHERE pf.status = 'FAIL' + + RETURN path, collect(DISTINCT pf) as dpf, collect(DISTINCT pfr) as dpfr + """, + parameters=[], + ), + AttackPathsQueryDefinition( + id="aws-rds-unencrypted-storage", + name="Identify RDS instances without storage encryption", + description="Find RDS instances with storage encryption disabled within the selected account.", + provider="aws", + cypher=""" + MATCH path = (aws:AWSAccount {id: $provider_uid})--(rds:RDSInstance) + WHERE rds.storage_encrypted = false + + UNWIND nodes(path) as n + OPTIONAL MATCH (n)-[pfr]-(pf:ProwlerFinding) + WHERE pf.status = 'FAIL' + + RETURN path, collect(DISTINCT pf) as dpf, collect(DISTINCT pfr) as dpfr + """, + parameters=[], + ), + AttackPathsQueryDefinition( + id="aws-s3-anonymous-access-buckets", + name="Identify S3 buckets with anonymous access", + description="Find S3 buckets that allow anonymous access within the selected account.", + provider="aws", + cypher=""" + MATCH path = (aws:AWSAccount {id: $provider_uid})--(s3:S3Bucket) + WHERE s3.anonymous_access = true + + UNWIND nodes(path) as n + OPTIONAL MATCH (n)-[pfr]-(pf:ProwlerFinding) + WHERE pf.status = 'FAIL' + + RETURN path, collect(DISTINCT pf) as dpf, collect(DISTINCT pfr) as dpfr + """, + parameters=[], + ), + AttackPathsQueryDefinition( + id="aws-iam-statements-allow-all-actions", + name="Identify IAM statements that allow all actions", + description="Find IAM policy statements that allow all actions via '*' within the selected account.", + provider="aws", + cypher=""" + MATCH path = (aws:AWSAccount {id: $provider_uid})--(principal:AWSPrincipal)--(pol:AWSPolicy)--(stmt:AWSPolicyStatement) + WHERE stmt.effect = 'Allow' + AND any(x IN stmt.action WHERE x = '*') + + UNWIND nodes(path) as n + OPTIONAL MATCH (n)-[pfr]-(pf:ProwlerFinding) + WHERE pf.status = 'FAIL' + + RETURN path, collect(DISTINCT pf) as dpf, collect(DISTINCT pfr) as dpfr + """, + parameters=[], + ), + AttackPathsQueryDefinition( + id="aws-iam-statements-allow-delete-policy", + name="Identify IAM statements that allow iam:DeletePolicy", + description="Find IAM policy statements that allow the iam:DeletePolicy action within the selected account.", + provider="aws", + cypher=""" + MATCH path = (aws:AWSAccount {id: $provider_uid})--(principal:AWSPrincipal)--(pol:AWSPolicy)--(stmt:AWSPolicyStatement) + WHERE stmt.effect = 'Allow' + AND any(x IN stmt.action WHERE x = "iam:DeletePolicy") + + UNWIND nodes(path) as n + OPTIONAL MATCH (n)-[pfr]-(pf:ProwlerFinding) + WHERE pf.status = 'FAIL' + + RETURN path, collect(DISTINCT pf) as dpf, collect(DISTINCT pfr) as dpfr + """, + parameters=[], + ), + AttackPathsQueryDefinition( + id="aws-iam-statements-allow-create-actions", + name="Identify IAM statements that allow create actions", + description="Find IAM policy statements that allow actions containing 'create' within the selected account.", + provider="aws", + cypher=""" + MATCH path = (aws:AWSAccount {id: $provider_uid})--(principal:AWSPrincipal)--(pol:AWSPolicy)--(stmt:AWSPolicyStatement) + WHERE stmt.effect = "Allow" + AND any(x IN stmt.action WHERE toLower(x) CONTAINS "create") + + UNWIND nodes(path) as n + OPTIONAL MATCH (n)-[pfr]-(pf:ProwlerFinding) + WHERE pf.status = 'FAIL' + + RETURN path, collect(DISTINCT pf) as dpf, collect(DISTINCT pfr) as dpfr + """, + parameters=[], + ), + AttackPathsQueryDefinition( + id="aws-ec2-instances-internet-exposed", + name="Identify internet-exposed EC2 instances", + description="Find EC2 instances flagged as exposed to the internet within the selected account.", + provider="aws", + cypher=""" + CALL apoc.create.vNode(['Internet'], {id: 'Internet', name: 'Internet'}) + YIELD node AS internet + + MATCH path = (aws:AWSAccount {id: $provider_uid})--(ec2:EC2Instance) + WHERE ec2.exposed_internet = true + + CALL apoc.create.vRelationship(internet, 'CAN_ACCESS', {}, ec2) + YIELD rel AS can_access + + UNWIND nodes(path) as n + OPTIONAL MATCH (n)-[pfr]-(pf:ProwlerFinding) + WHERE pf.status = 'FAIL' + + RETURN path, collect(DISTINCT pf) as dpf, collect(DISTINCT pfr) as dpfr, internet, can_access + """, + parameters=[], + ), + AttackPathsQueryDefinition( + id="aws-security-groups-open-internet-facing", + name="Identify internet-facing resources with open security groups", + description="Find internet-facing resources associated with security groups that allow inbound access from '0.0.0.0/0'.", + provider="aws", + cypher=""" + CALL apoc.create.vNode(['Internet'], {id: 'Internet', name: 'Internet'}) + YIELD node AS internet + + MATCH path_open = (aws:AWSAccount {id: $provider_uid})-[r0]-(open) + MATCH path_sg = (open)-[r1:MEMBER_OF_EC2_SECURITY_GROUP]-(sg:EC2SecurityGroup) + MATCH path_ip = (sg)-[r2:MEMBER_OF_EC2_SECURITY_GROUP]-(ipi:IpPermissionInbound) + MATCH path_ipi = (ipi)-[r3]-(ir:IpRange) + WHERE ir.range = "0.0.0.0/0" + OPTIONAL MATCH path_dns = (dns:AWSDNSRecord)-[:DNS_POINTS_TO]->(lb) + WHERE open.scheme = 'internet-facing' + + CALL apoc.create.vRelationship(internet, 'CAN_ACCESS', {}, open) + YIELD rel AS can_access + + UNWIND nodes(path_open) + nodes(path_sg) + nodes(path_ip) + nodes(path_ipi) + nodes(path_dns) as n + OPTIONAL MATCH (n)-[pfr]-(pf:ProwlerFinding) + WHERE pf.status = 'FAIL' + + RETURN path_open, path_sg, path_ip, path_ipi, path_dns, collect(DISTINCT pf) as dpf, collect(DISTINCT pfr) as dpfr, internet, can_access + """, + parameters=[], + ), + AttackPathsQueryDefinition( + id="aws-classic-elb-internet-exposed", + name="Identify internet-exposed Classic Load Balancers", + description="Find Classic Load Balancers exposed to the internet along with their listeners.", + provider="aws", + cypher=""" + CALL apoc.create.vNode(['Internet'], {id: 'Internet', name: 'Internet'}) + YIELD node AS internet + + MATCH path = (aws:AWSAccount {id: $provider_uid})--(elb:LoadBalancer)--(listener:ELBListener) + WHERE elb.exposed_internet = true + + CALL apoc.create.vRelationship(internet, 'CAN_ACCESS', {}, elb) + YIELD rel AS can_access + + UNWIND nodes(path) as n + OPTIONAL MATCH (n)-[pfr]-(pf:ProwlerFinding) + WHERE pf.status = 'FAIL' + + RETURN path, collect(DISTINCT pf) as dpf, collect(DISTINCT pfr) as dpfr, internet, can_access + """, + parameters=[], + ), + AttackPathsQueryDefinition( + id="aws-elbv2-internet-exposed", + name="Identify internet-exposed ELBv2 load balancers", + description="Find ELBv2 load balancers exposed to the internet along with their listeners.", + provider="aws", + cypher=""" + CALL apoc.create.vNode(['Internet'], {id: 'Internet', name: 'Internet'}) + YIELD node AS internet + + MATCH path = (aws:AWSAccount {id: $provider_uid})--(elbv2:LoadBalancerV2)--(listener:ELBV2Listener) + WHERE elbv2.exposed_internet = true + + CALL apoc.create.vRelationship(internet, 'CAN_ACCESS', {}, elbv2) + YIELD rel AS can_access + + UNWIND nodes(path) as n + OPTIONAL MATCH (n)-[pfr]-(pf:ProwlerFinding) + WHERE pf.status = 'FAIL' + + RETURN path, collect(DISTINCT pf) as dpf, collect(DISTINCT pfr) as dpfr, internet, can_access + """, + parameters=[], + ), + AttackPathsQueryDefinition( + id="aws-public-ip-resource-lookup", + name="Identify resources by public IP address", + description="Given a public IP address, find the related AWS resource and its adjacent node within the selected account.", + provider="aws", + cypher=""" + CALL apoc.create.vNode(['Internet'], {id: 'Internet', name: 'Internet'}) + YIELD node AS internet + + CALL () { + MATCH path = (aws:AWSAccount {id: $provider_uid})-[r]-(x:EC2PrivateIp)-[q]-(y) + WHERE x.public_ip = $ip + RETURN path, x + + UNION MATCH path = (aws:AWSAccount {id: $provider_uid})-[r]-(x:EC2Instance)-[q]-(y) + WHERE x.publicipaddress = $ip + RETURN path, x + + UNION MATCH path = (aws:AWSAccount {id: $provider_uid})-[r]-(x:NetworkInterface)-[q]-(y) + WHERE x.public_ip = $ip + RETURN path, x + + UNION MATCH path = (aws:AWSAccount {id: $provider_uid})-[r]-(x:ElasticIPAddress)-[q]-(y) + WHERE x.public_ip = $ip + RETURN path, x + } + + WITH path, x, internet + + CALL apoc.create.vRelationship(internet, 'CAN_ACCESS', {}, x) + YIELD rel AS can_access + + UNWIND nodes(path) as n + OPTIONAL MATCH (n)-[pfr]-(pf:ProwlerFinding) + WHERE pf.status = 'FAIL' + + RETURN path, collect(DISTINCT pf) as dpf, collect(DISTINCT pfr) as dpfr, internet, can_access + """, + parameters=[ + AttackPathsQueryParameterDefinition( + name="ip", + label="IP address", + description="Public IP address, e.g. 192.0.2.0.", + placeholder="192.0.2.0", + ), + ], + ), + ], +} + +_QUERIES_BY_ID: dict[str, AttackPathsQueryDefinition] = { + definition.id: definition + for definitions in _QUERY_DEFINITIONS.values() + for definition in definitions +} diff --git a/api/src/backend/api/attack_paths/views_helpers.py b/api/src/backend/api/attack_paths/views_helpers.py new file mode 100644 index 0000000000..7418a0302e --- /dev/null +++ b/api/src/backend/api/attack_paths/views_helpers.py @@ -0,0 +1,143 @@ +import logging + +from typing import Any + +from rest_framework.exceptions import APIException, ValidationError + +from api.attack_paths import database as graph_database, AttackPathsQueryDefinition +from api.models import AttackPathsScan +from config.custom_logging import BackendLogger + +logger = logging.getLogger(BackendLogger.API) + + +def normalize_run_payload(raw_data): + if not isinstance(raw_data, dict): # Let the serializer handle this + return raw_data + + if "data" in raw_data and isinstance(raw_data.get("data"), dict): + data_section = raw_data.get("data") or {} + attributes = data_section.get("attributes") or {} + payload = { + "id": attributes.get("id", data_section.get("id")), + "parameters": attributes.get("parameters"), + } + + # Remove `None` parameters to allow defaults downstream + if payload.get("parameters") is None: + payload.pop("parameters") + return payload + + return raw_data + + +def prepare_query_parameters( + definition: AttackPathsQueryDefinition, + provided_parameters: dict[str, Any], + provider_uid: str, +) -> dict[str, Any]: + parameters = dict(provided_parameters or {}) + expected_names = {parameter.name for parameter in definition.parameters} + provided_names = set(parameters.keys()) + + unexpected = provided_names - expected_names + if unexpected: + raise ValidationError( + {"parameters": f"Unknown parameter(s): {', '.join(sorted(unexpected))}"} + ) + + missing = expected_names - provided_names + if missing: + raise ValidationError( + { + "parameters": f"Missing required parameter(s): {', '.join(sorted(missing))}" + } + ) + + clean_parameters = { + "provider_uid": str(provider_uid), + } + + for definition_parameter in definition.parameters: + raw_value = provided_parameters[definition_parameter.name] + + try: + casted_value = definition_parameter.cast(raw_value) + + except (ValueError, TypeError) as exc: + raise ValidationError( + { + "parameters": ( + f"Invalid value for parameter `{definition_parameter.name}`: {str(exc)}" + ) + } + ) + + clean_parameters[definition_parameter.name] = casted_value + + return clean_parameters + + +def execute_attack_paths_query( + attack_paths_scan: AttackPathsScan, + definition: AttackPathsQueryDefinition, + parameters: dict[str, Any], +) -> dict[str, Any]: + try: + with graph_database.get_session(attack_paths_scan.graph_database) as session: + result = session.run(definition.cypher, parameters) + return _serialize_graph(result.graph()) + + except graph_database.GraphDatabaseQueryException as exc: + logger.error(f"Query failed for Attack Paths query `{definition.id}`: {exc}") + raise APIException( + "Attack Paths query execution failed due to a database error" + ) + + +def _serialize_graph(graph): + nodes = [] + for node in graph.nodes: + nodes.append( + { + "id": node.element_id, + "labels": list(node.labels), + "properties": _serialize_properties(node._properties), + }, + ) + + relationships = [] + for relationship in graph.relationships: + relationships.append( + { + "id": relationship.element_id, + "label": relationship.type, + "source": relationship.start_node.element_id, + "target": relationship.end_node.element_id, + "properties": _serialize_properties(relationship._properties), + }, + ) + + return { + "nodes": nodes, + "relationships": relationships, + } + + +def _serialize_properties(properties: dict[str, Any]) -> dict[str, Any]: + """Convert Neo4j property values into JSON-serializable primitives.""" + + def _serialize_value(value: Any) -> Any: + # Neo4j temporal and spatial values expose `to_native` returning Python primitives + if hasattr(value, "to_native") and callable(value.to_native): + return _serialize_value(value.to_native()) + + if isinstance(value, (list, tuple)): + return [_serialize_value(item) for item in value] + + if isinstance(value, dict): + return {key: _serialize_value(val) for key, val in value.items()} + + return value + + return {key: _serialize_value(val) for key, val in properties.items()} diff --git a/api/src/backend/api/filters.py b/api/src/backend/api/filters.py index 1f388a52a9..c11ca193b8 100644 --- a/api/src/backend/api/filters.py +++ b/api/src/backend/api/filters.py @@ -27,6 +27,7 @@ from api.models import ( Finding, Integration, Invitation, + AttackPathsScan, LighthouseProviderConfiguration, LighthouseProviderModels, Membership, @@ -330,6 +331,23 @@ class ScanFilter(ProviderRelationshipFilterSet): } +class AttackPathsScanFilter(ProviderRelationshipFilterSet): + inserted_at = DateFilter(field_name="inserted_at", lookup_expr="date") + completed_at = DateFilter(field_name="completed_at", lookup_expr="date") + started_at = DateFilter(field_name="started_at", lookup_expr="date") + state = ChoiceFilter(choices=StateChoices.choices) + state__in = ChoiceInFilter( + field_name="state", choices=StateChoices.choices, lookup_expr="in" + ) + + class Meta: + model = AttackPathsScan + fields = { + "provider": ["exact", "in"], + "scan": ["exact", "in"], + } + + class TaskFilter(FilterSet): name = CharFilter(field_name="task_runner_task__task_name", lookup_expr="exact") name__icontains = CharFilter( diff --git a/api/src/backend/api/fixtures/dev/8_dev_attack_paths_scans.json b/api/src/backend/api/fixtures/dev/8_dev_attack_paths_scans.json new file mode 100644 index 0000000000..fdf310458a --- /dev/null +++ b/api/src/backend/api/fixtures/dev/8_dev_attack_paths_scans.json @@ -0,0 +1,41 @@ +[ + { + "model": "api.attackpathsscan", + "pk": "a7f0f6de-6f8e-4b3a-8cbe-3f6dd9012345", + "fields": { + "tenant": "12646005-9067-4d2a-a098-8bb378604362", + "provider": "b85601a8-4b45-4194-8135-03fb980ef428", + "scan": "01920573-aa9c-73c9-bcda-f2e35c9b19d2", + "state": "completed", + "progress": 100, + "update_tag": 1693586667, + "graph_database": "db-a7f0f6de-6f8e-4b3a-8cbe-3f6dd9012345", + "is_graph_database_deleted": false, + "task": null, + "inserted_at": "2024-09-01T17:24:37Z", + "updated_at": "2024-09-01T17:44:37Z", + "started_at": "2024-09-01T17:34:37Z", + "completed_at": "2024-09-01T17:44:37Z", + "duration": 269, + "ingestion_exceptions": {} + } + }, + { + "model": "api.attackpathsscan", + "pk": "4a2fb2af-8a60-4d7d-9cae-4ca65e098765", + "fields": { + "tenant": "12646005-9067-4d2a-a098-8bb378604362", + "provider": "15fce1fa-ecaa-433f-a9dc-62553f3a2555", + "scan": "01929f3b-ed2e-7623-ad63-7c37cd37828f", + "state": "executing", + "progress": 48, + "update_tag": 1697625000, + "graph_database": "db-4a2fb2af-8a60-4d7d-9cae-4ca65e098765", + "is_graph_database_deleted": false, + "task": null, + "inserted_at": "2024-10-18T10:55:57Z", + "updated_at": "2024-10-18T10:56:15Z", + "started_at": "2024-10-18T10:56:05Z" + } + } +] diff --git a/api/src/backend/api/migrations/0060_attack_paths_scan.py b/api/src/backend/api/migrations/0060_attack_paths_scan.py new file mode 100644 index 0000000000..8f57039723 --- /dev/null +++ b/api/src/backend/api/migrations/0060_attack_paths_scan.py @@ -0,0 +1,154 @@ +# Generated by Django 5.1.13 on 2025-11-06 16:20 + +import django.db.models.deletion + +from django.db import migrations, models +from uuid6 import uuid7 + +import api.rls + + +class Migration(migrations.Migration): + dependencies = [ + ("api", "0059_compliance_overview_summary"), + ] + + operations = [ + migrations.CreateModel( + name="AttackPathsScan", + fields=[ + ( + "id", + models.UUIDField( + default=uuid7, + editable=False, + primary_key=True, + serialize=False, + ), + ), + ("inserted_at", models.DateTimeField(auto_now_add=True)), + ("updated_at", models.DateTimeField(auto_now=True)), + ( + "state", + api.db_utils.StateEnumField( + choices=[ + ("available", "Available"), + ("scheduled", "Scheduled"), + ("executing", "Executing"), + ("completed", "Completed"), + ("failed", "Failed"), + ("cancelled", "Cancelled"), + ], + default="available", + ), + ), + ("progress", models.IntegerField(default=0)), + ("started_at", models.DateTimeField(blank=True, null=True)), + ("completed_at", models.DateTimeField(blank=True, null=True)), + ( + "duration", + models.IntegerField( + blank=True, help_text="Duration in seconds", null=True + ), + ), + ( + "update_tag", + models.BigIntegerField( + blank=True, + help_text="Cartography update tag (epoch)", + null=True, + ), + ), + ( + "graph_database", + models.CharField(blank=True, max_length=63, null=True), + ), + ( + "is_graph_database_deleted", + models.BooleanField(default=False), + ), + ( + "ingestion_exceptions", + models.JSONField(blank=True, default=dict, null=True), + ), + ( + "provider", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="attack_paths_scans", + related_query_name="attack_paths_scan", + to="api.provider", + ), + ), + ( + "scan", + models.ForeignKey( + blank=True, + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="attack_paths_scans", + related_query_name="attack_paths_scan", + to="api.scan", + ), + ), + ( + "task", + models.ForeignKey( + blank=True, + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="attack_paths_scans", + related_query_name="attack_paths_scan", + to="api.task", + ), + ), + ( + "tenant", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, to="api.tenant" + ), + ), + ], + options={ + "db_table": "attack_paths_scans", + "abstract": False, + "indexes": [ + models.Index( + fields=["tenant_id", "provider_id", "-inserted_at"], + name="aps_prov_ins_desc_idx", + ), + models.Index( + fields=["tenant_id", "state", "-inserted_at"], + name="aps_state_ins_desc_idx", + ), + models.Index( + 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=models.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=models.Q( + ("state", "completed"), + ("is_graph_database_deleted", False), + ), + ), + ], + }, + ), + migrations.AddConstraint( + model_name="attackpathsscan", + constraint=api.rls.RowLevelSecurityConstraint( + "tenant_id", + name="rls_on_attackpathsscan", + statements=["SELECT", "INSERT", "UPDATE", "DELETE"], + ), + ), + ] diff --git a/api/src/backend/api/models.py b/api/src/backend/api/models.py index 1191350ac6..5f00c578e2 100644 --- a/api/src/backend/api/models.py +++ b/api/src/backend/api/models.py @@ -616,6 +616,101 @@ class Scan(RowLevelSecurityProtectedModel): resource_name = "scans" +class AttackPathsScan(RowLevelSecurityProtectedModel): + objects = ActiveProviderManager() + all_objects = models.Manager() + + id = models.UUIDField(primary_key=True, default=uuid7, editable=False) + inserted_at = models.DateTimeField(auto_now_add=True, editable=False) + updated_at = models.DateTimeField(auto_now=True, editable=False) + + state = StateEnumField(choices=StateChoices.choices, default=StateChoices.AVAILABLE) + progress = models.IntegerField(default=0) + + # Timing + started_at = models.DateTimeField(null=True, blank=True) + completed_at = models.DateTimeField(null=True, blank=True) + duration = models.IntegerField( + null=True, blank=True, help_text="Duration in seconds" + ) + + # Relationship to the provider and optional prowler Scan and celery Task + provider = models.ForeignKey( + "Provider", + on_delete=models.CASCADE, + related_name="attack_paths_scans", + related_query_name="attack_paths_scan", + ) + scan = models.ForeignKey( + "Scan", + on_delete=models.SET_NULL, + null=True, + blank=True, + related_name="attack_paths_scans", + related_query_name="attack_paths_scan", + ) + task = models.ForeignKey( + "Task", + on_delete=models.SET_NULL, + null=True, + blank=True, + related_name="attack_paths_scans", + related_query_name="attack_paths_scan", + ) + + # Cartography specific metadata + 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): + db_table = "attack_paths_scans" + + constraints = [ + RowLevelSecurityConstraint( + field="tenant_id", + name="rls_on_%(class)s", + statements=["SELECT", "INSERT", "UPDATE", "DELETE"], + ), + ] + + indexes = [ + models.Index( + fields=["tenant_id", "provider_id", "-inserted_at"], + name="aps_prov_ins_desc_idx", + ), + models.Index( + fields=["tenant_id", "state", "-inserted_at"], + name="aps_state_ins_desc_idx", + ), + models.Index( + 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: + resource_name = "attack-paths-scans" + + class ResourceTag(RowLevelSecurityProtectedModel): id = models.UUIDField(primary_key=True, default=uuid4, editable=False) inserted_at = models.DateTimeField(auto_now_add=True, editable=False) diff --git a/api/src/backend/api/specs/v1.yaml b/api/src/backend/api/specs/v1.yaml index 9e9e33ae2a..4a1791e65b 100644 --- a/api/src/backend/api/specs/v1.yaml +++ b/api/src/backend/api/specs/v1.yaml @@ -280,6 +280,435 @@ paths: schema: $ref: '#/components/schemas/OpenApiResponseResponse' description: API key was successfully revoked + /api/v1/attack-paths-scans: + get: + operationId: attack_paths_scans_list + description: Retrieve Attack Paths scans for the tenant with support for filtering, + ordering, and pagination. + summary: List Attack Paths scans + parameters: + - in: query + name: fields[attack-paths-scans] + schema: + type: array + items: + type: string + enum: + - state + - progress + - provider + - provider_alias + - provider_type + - provider_uid + - scan + - task + - inserted_at + - started_at + - completed_at + - duration + description: endpoint return only specific fields in the response on a per-type + basis by including a fields[TYPE] query parameter. + explode: false + - in: query + name: filter[completed_at] + schema: + type: string + format: date + - in: query + name: filter[inserted_at] + schema: + type: string + format: date + - in: query + name: filter[provider] + schema: + type: string + format: uuid + - in: query + name: filter[provider__in] + schema: + type: array + items: + type: string + format: uuid + description: Multiple values may be separated by commas. + explode: false + style: form + - in: query + name: filter[provider_alias] + schema: + type: string + - in: query + name: filter[provider_alias__icontains] + schema: + type: string + - in: query + name: filter[provider_alias__in] + schema: + type: array + items: + type: string + description: Multiple values may be separated by commas. + explode: false + style: form + - in: query + name: filter[provider_type] + schema: + type: string + x-spec-enum-id: eca8c51e6bd28935 + enum: + - aws + - azure + - gcp + - github + - iac + - kubernetes + - m365 + - mongodbatlas + - oraclecloud + description: |- + * `aws` - AWS + * `azure` - Azure + * `gcp` - GCP + * `kubernetes` - Kubernetes + * `m365` - M365 + * `github` - GitHub + * `mongodbatlas` - MongoDB Atlas + * `iac` - IaC + * `oraclecloud` - Oracle Cloud Infrastructure + - in: query + name: filter[provider_type__in] + schema: + type: array + items: + type: string + x-spec-enum-id: eca8c51e6bd28935 + enum: + - aws + - azure + - gcp + - github + - iac + - kubernetes + - m365 + - mongodbatlas + - oraclecloud + description: |- + Multiple values may be separated by commas. + + * `aws` - AWS + * `azure` - Azure + * `gcp` - GCP + * `kubernetes` - Kubernetes + * `m365` - M365 + * `github` - GitHub + * `mongodbatlas` - MongoDB Atlas + * `iac` - IaC + * `oraclecloud` - Oracle Cloud Infrastructure + explode: false + style: form + - in: query + name: filter[provider_uid] + schema: + type: string + - in: query + name: filter[provider_uid__icontains] + schema: + type: string + - in: query + name: filter[provider_uid__in] + schema: + type: array + items: + type: string + description: Multiple values may be separated by commas. + explode: false + style: form + - in: query + name: filter[scan] + schema: + type: string + format: uuid + - in: query + name: filter[scan__in] + schema: + type: array + items: + type: string + format: uuid + description: Multiple values may be separated by commas. + explode: false + style: form + - name: filter[search] + required: false + in: query + description: A search term. + schema: + type: string + - in: query + name: filter[started_at] + schema: + type: string + format: date + - in: query + name: filter[state] + schema: + type: string + x-spec-enum-id: d38ba07264e1ed34 + enum: + - available + - cancelled + - completed + - executing + - failed + - scheduled + description: |- + * `available` - Available + * `scheduled` - Scheduled + * `executing` - Executing + * `completed` - Completed + * `failed` - Failed + * `cancelled` - Cancelled + - in: query + name: filter[state__in] + schema: + type: array + items: + type: string + x-spec-enum-id: d38ba07264e1ed34 + enum: + - available + - cancelled + - completed + - executing + - failed + - scheduled + description: |- + Multiple values may be separated by commas. + + * `available` - Available + * `scheduled` - Scheduled + * `executing` - Executing + * `completed` - Completed + * `failed` - Failed + * `cancelled` - Cancelled + explode: false + style: form + - in: query + name: include + schema: + type: array + items: + type: string + enum: + - provider + - scan + - task + description: include query parameter to allow the client to customize which + related resources should be returned. + explode: false + - 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: + - inserted_at + - -inserted_at + - started_at + - -started_at + explode: false + tags: + - Attack Paths + security: + - JWT or API Key: [] + responses: + '200': + content: + application/vnd.api+json: + schema: + $ref: '#/components/schemas/PaginatedAttackPathsScanList' + description: '' + /api/v1/attack-paths-scans/{id}: + get: + operationId: attack_paths_scans_retrieve + description: Fetch full details for a specific Attack Paths scan. + summary: Retrieve Attack Paths scan details + parameters: + - in: query + name: fields[attack-paths-scans] + schema: + type: array + items: + type: string + enum: + - state + - progress + - provider + - provider_alias + - provider_type + - provider_uid + - scan + - task + - inserted_at + - started_at + - completed_at + - duration + 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 attack paths scan. + required: true + - in: query + name: include + schema: + type: array + items: + type: string + enum: + - provider + - scan + - task + description: include query parameter to allow the client to customize which + related resources should be returned. + explode: false + tags: + - Attack Paths + security: + - JWT or API Key: [] + responses: + '200': + content: + application/vnd.api+json: + schema: + $ref: '#/components/schemas/AttackPathsScanResponse' + description: '' + /api/v1/attack-paths-scans/{id}/queries: + get: + 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 + parameters: + - in: query + name: fields[attack-paths-scans] + schema: + type: array + items: + type: string + enum: + - state + - progress + - provider + - provider_alias + - provider_type + - provider_uid + - scan + - task + - inserted_at + - started_at + - completed_at + - duration + 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 attack paths scan. + required: true + - in: query + name: include + schema: + type: array + items: + type: string + enum: + - provider + - scan + - task + description: include query parameter to allow the client to customize which + related resources should be returned. + explode: false + tags: + - Attack Paths + security: + - JWT or API Key: [] + responses: + '200': + content: + application/vnd.api+json: + schema: + $ref: '#/components/schemas/PaginatedAttackPathsQueryList' + description: '' + '404': + description: No queries found for the selected provider + /api/v1/attack-paths-scans/{id}/queries/run: + post: + operationId: attack_paths_scans_queries_run_create + description: Execute the selected Attack Paths query against the Attack Paths + graph and return the resulting subgraph. + summary: Execute an Attack Paths query + parameters: + - in: path + name: id + schema: + type: string + format: uuid + description: A UUID string identifying this attack paths scan. + required: true + tags: + - Attack Paths + requestBody: + content: + application/vnd.api+json: + schema: + $ref: '#/components/schemas/AttackPathsQueryRunRequestRequest' + application/x-www-form-urlencoded: + schema: + $ref: '#/components/schemas/AttackPathsQueryRunRequestRequest' + multipart/form-data: + schema: + $ref: '#/components/schemas/AttackPathsQueryRunRequestRequest' + required: true + security: + - JWT or API Key: [] + responses: + '200': + content: + application/vnd.api+json: + schema: + $ref: '#/components/schemas/OpenApiResponseResponse' + description: '' + '400': + 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 + '500': + description: Attack Paths query execution failed due to a database error /api/v1/compliance-overviews: get: operationId: compliance_overviews_list @@ -10618,6 +11047,349 @@ paths: description: '' components: schemas: + AttackPathsNode: + 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: + - attack-paths-query-result-node + attributes: + type: object + properties: + id: + type: string + labels: + type: array + items: + type: string + properties: + type: object + additionalProperties: {} + required: + - id + - labels + - properties + AttackPathsQuery: + 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 + id: {} + attributes: + type: object + properties: + id: + type: string + name: + type: string + description: + type: string + provider: + type: string + parameters: + type: array + items: + $ref: '#/components/schemas/AttackPathsQueryParameter' + required: + - id + - name + - description + - provider + - parameters + AttackPathsQueryParameter: + 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-parameter + id: {} + attributes: + type: object + properties: + name: + type: string + label: + type: string + data_type: + type: string + default: string + description: + type: string + nullable: true + placeholder: + type: string + nullable: true + required: + - name + - label + AttackPathsQueryResult: + 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: + - attack-paths-query-result + attributes: + type: object + properties: + nodes: + type: array + items: + $ref: '#/components/schemas/AttackPathsNode' + relationships: + type: array + items: + $ref: '#/components/schemas/AttackPathsRelationship' + required: + - nodes + - relationships + AttackPathsQueryRunRequestRequest: + 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: + - attack-paths-query-run-request + attributes: + type: object + properties: + id: + type: string + minLength: 1 + parameters: + type: object + additionalProperties: {} + required: + - id + required: + - data + AttackPathsRelationship: + 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: + - attack-paths-query-result-relationship + attributes: + type: object + properties: + id: + type: string + label: + type: string + source: + type: string + target: + type: string + properties: + type: object + additionalProperties: {} + required: + - id + - label + - source + - target + - properties + AttackPathsScan: + 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-scans + id: + type: string + format: uuid + attributes: + type: object + properties: + state: + enum: + - available + - scheduled + - executing + - completed + - failed + - cancelled + type: string + description: |- + * `available` - Available + * `scheduled` - Scheduled + * `executing` - Executing + * `completed` - Completed + * `failed` - Failed + * `cancelled` - Cancelled + x-spec-enum-id: d38ba07264e1ed34 + readOnly: true + progress: + type: integer + maximum: 2147483647 + minimum: -2147483648 + provider_alias: + type: string + readOnly: true + provider_type: + type: string + readOnly: true + provider_uid: + type: string + readOnly: true + inserted_at: + type: string + format: date-time + readOnly: true + started_at: + type: string + format: date-time + nullable: true + completed_at: + type: string + format: date-time + nullable: true + duration: + type: integer + maximum: 2147483647 + minimum: -2147483648 + nullable: true + description: Duration in seconds + relationships: + type: object + properties: + provider: + type: object + properties: + data: + type: object + properties: + id: + type: string + format: uuid + type: + type: string + enum: + - providers + title: Resource Type Name + description: The [type](https://jsonapi.org/format/#document-resource-object-identification) + member is used to describe resource objects that share common + attributes and relationships. + required: + - id + - type + required: + - data + description: The identifier of the related object. + title: Resource Identifier + scan: + type: object + properties: + data: + type: object + properties: + id: + type: string + format: uuid + type: + type: string + enum: + - scans + title: Resource Type Name + description: The [type](https://jsonapi.org/format/#document-resource-object-identification) + member is used to describe resource objects that share common + attributes and relationships. + required: + - id + - type + required: + - data + description: The identifier of the related object. + title: Resource Identifier + nullable: true + task: + type: object + properties: + data: + type: object + properties: + id: + type: string + format: uuid + type: + type: string + enum: + - tasks + title: Resource Type Name + description: The [type](https://jsonapi.org/format/#document-resource-object-identification) + member is used to describe resource objects that share common + attributes and relationships. + required: + - id + - type + required: + - data + description: The identifier of the related object. + title: Resource Identifier + nullable: true + required: + - provider + AttackPathsScanResponse: + type: object + properties: + data: + $ref: '#/components/schemas/AttackPathsScan' + required: + - data ComplianceOverview: type: object required: @@ -13576,6 +14348,24 @@ components: $ref: '#/components/schemas/OverviewSeverity' required: - data + PaginatedAttackPathsQueryList: + type: object + properties: + data: + type: array + items: + $ref: '#/components/schemas/AttackPathsQuery' + required: + - data + PaginatedAttackPathsScanList: + type: object + properties: + data: + type: array + items: + $ref: '#/components/schemas/AttackPathsScan' + required: + - data PaginatedComplianceOverviewAttributesList: type: object properties: @@ -19660,6 +20450,8 @@ tags: revoking tasks that have not started. - name: Scan description: Endpoints for triggering manual scans and viewing scan results. +- name: Attack Paths + description: Endpoints for Attack Paths scan status and executing Attack Paths queries. - name: Schedule description: Endpoints for managing scan schedules, allowing configuration of automated scans with different scheduling options. diff --git a/api/src/backend/api/tests/test_attack_paths.py b/api/src/backend/api/tests/test_attack_paths.py new file mode 100644 index 0000000000..2c4e1484f8 --- /dev/null +++ b/api/src/backend/api/tests/test_attack_paths.py @@ -0,0 +1,172 @@ +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +from rest_framework.exceptions import APIException, ValidationError + +from api.attack_paths import database as graph_database +from api.attack_paths import views_helpers + + +def test_normalize_run_payload_extracts_attributes_section(): + payload = { + "data": { + "id": "ignored", + "attributes": { + "id": "aws-rds", + "parameters": {"ip": "192.0.2.0"}, + }, + } + } + + result = views_helpers.normalize_run_payload(payload) + + assert result == {"id": "aws-rds", "parameters": {"ip": "192.0.2.0"}} + + +def test_normalize_run_payload_passthrough_for_non_dict(): + sentinel = "not-a-dict" + assert views_helpers.normalize_run_payload(sentinel) is sentinel + + +def test_prepare_query_parameters_includes_provider_and_casts( + attack_paths_query_definition_factory, +): + definition = attack_paths_query_definition_factory(cast_type=int) + result = views_helpers.prepare_query_parameters( + definition, + {"limit": "5"}, + provider_uid="123456789012", + ) + + assert result["provider_uid"] == "123456789012" + assert result["limit"] == 5 + + +@pytest.mark.parametrize( + "provided,expected_message", + [ + ({}, "Missing required parameter"), + ({"limit": 10, "extra": True}, "Unknown parameter"), + ], +) +def test_prepare_query_parameters_validates_names( + attack_paths_query_definition_factory, provided, expected_message +): + definition = attack_paths_query_definition_factory() + + with pytest.raises(ValidationError) as exc: + views_helpers.prepare_query_parameters(definition, provided, provider_uid="1") + + assert expected_message in str(exc.value) + + +def test_prepare_query_parameters_validates_cast( + attack_paths_query_definition_factory, +): + definition = attack_paths_query_definition_factory(cast_type=int) + + with pytest.raises(ValidationError) as exc: + views_helpers.prepare_query_parameters( + definition, + {"limit": "not-an-int"}, + provider_uid="1", + ) + + assert "Invalid value" in str(exc.value) + + +def test_execute_attack_paths_query_serializes_graph( + attack_paths_query_definition_factory, attack_paths_graph_stub_classes +): + definition = attack_paths_query_definition_factory( + id="aws-rds", + name="RDS", + description="", + cypher="MATCH (n) RETURN n", + parameters=[], + ) + parameters = {"provider_uid": "123"} + attack_paths_scan = SimpleNamespace(graph_database="tenant-db") + + node = attack_paths_graph_stub_classes.Node( + element_id="node-1", + labels=["AWSAccount"], + properties={ + "name": "account", + "complex": { + "items": [ + attack_paths_graph_stub_classes.NativeValue("value"), + {"nested": 1}, + ] + }, + }, + ) + relationship = attack_paths_graph_stub_classes.Relationship( + element_id="rel-1", + rel_type="OWNS", + start_node=node, + end_node=attack_paths_graph_stub_classes.Node("node-2", ["RDSInstance"], {}), + properties={"weight": 1}, + ) + graph = SimpleNamespace(nodes=[node], relationships=[relationship]) + + run_result = MagicMock() + run_result.graph.return_value = graph + + session = MagicMock() + session.run.return_value = run_result + + session_ctx = MagicMock() + session_ctx.__enter__.return_value = session + session_ctx.__exit__.return_value = False + + with patch( + "api.attack_paths.views_helpers.graph_database.get_session", + return_value=session_ctx, + ) as mock_get_session: + result = views_helpers.execute_attack_paths_query( + attack_paths_scan, definition, parameters + ) + + mock_get_session.assert_called_once_with("tenant-db") + session.run.assert_called_once_with(definition.cypher, parameters) + assert result["nodes"][0]["id"] == "node-1" + assert result["nodes"][0]["properties"]["complex"]["items"][0] == "value" + assert result["relationships"][0]["label"] == "OWNS" + + +def test_execute_attack_paths_query_wraps_graph_errors( + attack_paths_query_definition_factory, +): + definition = attack_paths_query_definition_factory( + id="aws-rds", + name="RDS", + description="", + cypher="MATCH (n) RETURN n", + parameters=[], + ) + attack_paths_scan = SimpleNamespace(graph_database="tenant-db") + parameters = {"provider_uid": "123"} + + class ExplodingContext: + def __enter__(self): + raise graph_database.GraphDatabaseQueryException("boom") + + def __exit__(self, exc_type, exc, tb): + return False + + with ( + patch( + "api.attack_paths.views_helpers.graph_database.get_session", + return_value=ExplodingContext(), + ), + patch("api.attack_paths.views_helpers.logger") as mock_logger, + ): + with pytest.raises(APIException): + views_helpers.execute_attack_paths_query( + attack_paths_scan, definition, parameters + ) + + mock_logger.error.assert_called_once() diff --git a/api/src/backend/api/tests/test_views.py b/api/src/backend/api/tests/test_views.py index 0717528209..7ba0d7094b 100644 --- a/api/src/backend/api/tests/test_views.py +++ b/api/src/backend/api/tests/test_views.py @@ -32,6 +32,10 @@ from django_celery_results.models import TaskResult from rest_framework import status from rest_framework.response import Response +from api.attack_paths import ( + AttackPathsQueryDefinition, + AttackPathsQueryParameterDefinition, +) from api.compliance import get_compliance_frameworks from api.db_router import MainRouter from api.models import ( @@ -3522,6 +3526,420 @@ class TestTaskViewSet: assert response.status_code == status.HTTP_400_BAD_REQUEST +@pytest.mark.django_db +class TestAttackPathsScanViewSet: + @staticmethod + def _run_payload(query_id="aws-rds", parameters=None): + return { + "data": { + "type": "attack-paths-query-run-request", + "attributes": { + "id": query_id, + "parameters": parameters or {}, + }, + } + } + + def test_attack_paths_scans_list_returns_latest_entry_per_provider( + self, + authenticated_client, + providers_fixture, + scans_fixture, + create_attack_paths_scan, + ): + provider = providers_fixture[0] + other_provider = providers_fixture[1] + + older_scan = create_attack_paths_scan( + provider, + scan=scans_fixture[0], + state=StateChoices.AVAILABLE, + progress=10, + ) + latest_scan = create_attack_paths_scan( + provider, + scan=scans_fixture[0], + state=StateChoices.COMPLETED, + progress=95, + ) + other_provider_scan = create_attack_paths_scan( + other_provider, + scan=scans_fixture[2], + state=StateChoices.FAILED, + progress=50, + ) + + response = authenticated_client.get(reverse("attack-paths-scans-list")) + + assert response.status_code == status.HTTP_200_OK + data = response.json()["data"] + ids = {item["id"] for item in data} + assert ids == {str(latest_scan.id), str(other_provider_scan.id)} + assert str(older_scan.id) not in ids + + provider_entry = next( + item + for item in data + if item["relationships"]["provider"]["data"]["id"] == str(provider.id) + ) + + first_attributes = provider_entry["attributes"] + assert first_attributes["provider_alias"] == provider.alias + assert first_attributes["provider_type"] == provider.provider + assert first_attributes["provider_uid"] == provider.uid + + def test_attack_paths_scans_list_respects_provider_group_visibility( + self, + authenticated_client_no_permissions_rbac, + providers_fixture, + create_attack_paths_scan, + ): + client = authenticated_client_no_permissions_rbac + limited_user = client.user + membership = Membership.objects.filter(user=limited_user).first() + tenant = membership.tenant + + allowed_provider = providers_fixture[0] + denied_provider = providers_fixture[1] + + allowed_scan = create_attack_paths_scan(allowed_provider) + create_attack_paths_scan(denied_provider) + + provider_group = ProviderGroup.objects.create( + name="limited-group", + tenant_id=tenant.id, + ) + ProviderGroupMembership.objects.create( + tenant_id=tenant.id, + provider_group=provider_group, + provider=allowed_provider, + ) + limited_role = limited_user.roles.first() + RoleProviderGroupRelationship.objects.create( + tenant_id=tenant.id, + role=limited_role, + provider_group=provider_group, + ) + + response = client.get(reverse("attack-paths-scans-list")) + + assert response.status_code == status.HTTP_200_OK + data = response.json()["data"] + assert len(data) == 1 + assert data[0]["id"] == str(allowed_scan.id) + + def test_attack_paths_scan_retrieve( + 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], + state=StateChoices.COMPLETED, + progress=80, + ) + + response = authenticated_client.get( + reverse("attack-paths-scans-detail", kwargs={"pk": attack_paths_scan.id}) + ) + + assert response.status_code == status.HTTP_200_OK + data = response.json()["data"] + assert data["id"] == str(attack_paths_scan.id) + assert data["relationships"]["provider"]["data"]["id"] == str(provider.id) + assert data["attributes"]["state"] == StateChoices.COMPLETED + + def test_attack_paths_scan_retrieve_not_found_for_foreign_tenant( + self, authenticated_client, create_attack_paths_scan + ): + other_tenant = Tenant.objects.create(name="Foreign AttackPaths Tenant") + foreign_provider = Provider.objects.create( + provider="aws", + uid="333333333333", + alias="foreign", + tenant_id=other_tenant.id, + ) + foreign_scan = create_attack_paths_scan(foreign_provider) + + response = authenticated_client.get( + reverse("attack-paths-scans-detail", kwargs={"pk": foreign_scan.id}) + ) + + assert response.status_code == status.HTTP_404_NOT_FOUND + + def test_attack_paths_queries_returns_catalog( + 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], + ) + + definitions = [ + AttackPathsQueryDefinition( + id="aws-rds", + name="RDS inventory", + description="List account RDS assets", + provider=provider.provider, + cypher="MATCH (n) RETURN n", + parameters=[ + AttackPathsQueryParameterDefinition(name="ip", label="IP address") + ], + ) + ] + + with patch( + "api.v1.views.get_queries_for_provider", return_value=definitions + ) as mock_get_queries: + response = authenticated_client.get( + reverse( + "attack-paths-scans-queries", kwargs={"pk": attack_paths_scan.id} + ) + ) + + assert response.status_code == status.HTTP_200_OK + mock_get_queries.assert_called_once_with(provider.provider) + payload = response.json()["data"] + assert len(payload) == 1 + assert payload[0]["id"] == "aws-rds" + assert payload[0]["attributes"]["name"] == "RDS inventory" + assert payload[0]["attributes"]["parameters"][0]["name"] == "ip" + + def test_attack_paths_queries_returns_404_when_catalog_missing( + 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]) + + with patch("api.v1.views.get_queries_for_provider", return_value=[]): + response = authenticated_client.get( + reverse( + "attack-paths-scans-queries", kwargs={"pk": attack_paths_scan.id} + ) + ) + + assert response.status_code == status.HTTP_404_NOT_FOUND + assert "No queries found" in str(response.json()) + + def test_run_attack_paths_query_returns_graph( + 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="tenant-db", + ) + query_definition = AttackPathsQueryDefinition( + id="aws-rds", + name="RDS inventory", + description="List account RDS assets", + provider=provider.provider, + cypher="MATCH (n) RETURN n", + parameters=[], + ) + prepared_parameters = {"provider_uid": provider.uid} + graph_payload = { + "nodes": [ + { + "id": "node-1", + "labels": ["AWSAccount"], + "properties": {"name": "root"}, + } + ], + "relationships": [ + { + "id": "rel-1", + "label": "OWNS", + "source": "node-1", + "target": "node-2", + "properties": {}, + } + ], + } + + with ( + patch( + "api.v1.views.get_query_by_id", return_value=query_definition + ) as mock_get_query, + patch( + "api.v1.views.attack_paths_views_helpers.prepare_query_parameters", + return_value=prepared_parameters, + ) as mock_prepare, + patch( + "api.v1.views.attack_paths_views_helpers.execute_attack_paths_query", + return_value=graph_payload, + ) as mock_execute, + ): + response = authenticated_client.post( + reverse( + "attack-paths-scans-queries-run", + kwargs={"pk": attack_paths_scan.id}, + ), + data=self._run_payload("aws-rds"), + content_type=API_JSON_CONTENT_TYPE, + ) + + assert response.status_code == status.HTTP_200_OK + mock_get_query.assert_called_once_with("aws-rds") + mock_prepare.assert_called_once_with( + query_definition, + {}, + attack_paths_scan.provider.uid, + ) + mock_execute.assert_called_once_with( + attack_paths_scan, + query_definition, + prepared_parameters, + ) + result = response.json()["data"] + attributes = result["attributes"] + assert attributes["nodes"] == graph_payload["nodes"] + assert attributes["relationships"] == graph_payload["relationships"] + + def test_run_attack_paths_query_requires_completed_scan( + 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], + state=StateChoices.EXECUTING, + ) + + 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_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, + providers_fixture, + scans_fixture, + create_attack_paths_scan, + ): + provider = providers_fixture[0] + attack_paths_scan = create_attack_paths_scan( + provider, + scan=scans_fixture[0], + ) + + with patch("api.v1.views.get_query_by_id", return_value=None): + response = authenticated_client.post( + reverse( + "attack-paths-scans-queries-run", + kwargs={"pk": attack_paths_scan.id}, + ), + data=self._run_payload("unknown-query"), + content_type=API_JSON_CONTENT_TYPE, + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "Unknown Attack Paths query" in response.json()["errors"][0]["detail"] + + def test_run_attack_paths_query_returns_404_when_no_nodes_found( + 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], + ) + query_definition = AttackPathsQueryDefinition( + id="aws-empty", + name="empty", + description="", + provider=provider.provider, + cypher="MATCH (n) RETURN n", + ) + + with ( + patch("api.v1.views.get_query_by_id", return_value=query_definition), + patch( + "api.v1.views.attack_paths_views_helpers.prepare_query_parameters", + return_value={"provider_uid": provider.uid}, + ), + patch( + "api.v1.views.attack_paths_views_helpers.execute_attack_paths_query", + return_value={"nodes": [], "relationships": []}, + ), + ): + response = authenticated_client.post( + reverse( + "attack-paths-scans-queries-run", + kwargs={"pk": attack_paths_scan.id}, + ), + data=self._run_payload("aws-empty"), + content_type=API_JSON_CONTENT_TYPE, + ) + + assert response.status_code == status.HTTP_404_NOT_FOUND + payload = response.json() + if "data" in payload: + attributes = payload["data"].get("attributes", {}) + assert attributes.get("nodes") == [] + assert attributes.get("relationships") == [] + else: + assert "errors" in payload + + @pytest.mark.django_db class TestResourceViewSet: def test_resources_list_none(self, authenticated_client): diff --git a/api/src/backend/api/v1/serializers.py b/api/src/backend/api/v1/serializers.py index d730f9b4fa..a23aa91519 100644 --- a/api/src/backend/api/v1/serializers.py +++ b/api/src/backend/api/v1/serializers.py @@ -21,6 +21,7 @@ from rest_framework_simplejwt.tokens import RefreshToken from api.db_router import MainRouter from api.exceptions import ConflictException from api.models import ( + AttackPathsScan, Finding, Integration, IntegrationProviderRelationship, @@ -1127,6 +1128,109 @@ class ScanComplianceReportSerializer(serializers.Serializer): fields = ["id", "name"] +class AttackPathsScanSerializer(RLSSerializer): + state = StateEnumSerializerField(read_only=True) + provider_alias = serializers.SerializerMethodField(read_only=True) + provider_type = serializers.SerializerMethodField(read_only=True) + provider_uid = serializers.SerializerMethodField(read_only=True) + + class Meta: + model = AttackPathsScan + fields = [ + "id", + "state", + "progress", + "provider", + "provider_alias", + "provider_type", + "provider_uid", + "scan", + "task", + "inserted_at", + "started_at", + "completed_at", + "duration", + ] + + included_serializers = { + "provider": "api.v1.serializers.ProviderIncludeSerializer", + "scan": "api.v1.serializers.ScanIncludeSerializer", + "task": "api.v1.serializers.TaskSerializer", + } + + def get_provider_alias(self, obj): + provider = getattr(obj, "provider", None) + return provider.alias if provider else None + + def get_provider_type(self, obj): + provider = getattr(obj, "provider", None) + return provider.provider if provider else None + + def get_provider_uid(self, obj): + provider = getattr(obj, "provider", None) + return provider.uid if provider else None + + +class AttackPathsQueryParameterSerializer(serializers.Serializer): + name = serializers.CharField() + label = serializers.CharField() + data_type = serializers.CharField(default="string") + description = serializers.CharField(allow_null=True, required=False) + placeholder = serializers.CharField(allow_null=True, required=False) + + class JSONAPIMeta: + resource_name = "attack-paths-query-parameter" + + +class AttackPathsQuerySerializer(serializers.Serializer): + id = serializers.CharField() + name = serializers.CharField() + description = serializers.CharField() + provider = serializers.CharField() + parameters = AttackPathsQueryParameterSerializer(many=True) + + class JSONAPIMeta: + resource_name = "attack-paths-query" + + +class AttackPathsQueryRunRequestSerializer(serializers.Serializer): + id = serializers.CharField() + parameters = serializers.DictField( + child=serializers.JSONField(), allow_empty=True, required=False + ) + + class JSONAPIMeta: + resource_name = "attack-paths-query-run-request" + + +class AttackPathsNodeSerializer(serializers.Serializer): + id = serializers.CharField() + labels = serializers.ListField(child=serializers.CharField()) + properties = serializers.DictField(child=serializers.JSONField()) + + class JSONAPIMeta: + resource_name = "attack-paths-query-result-node" + + +class AttackPathsRelationshipSerializer(serializers.Serializer): + id = serializers.CharField() + label = serializers.CharField() + source = serializers.CharField() + target = serializers.CharField() + properties = serializers.DictField(child=serializers.JSONField()) + + class JSONAPIMeta: + resource_name = "attack-paths-query-result-relationship" + + +class AttackPathsQueryResultSerializer(serializers.Serializer): + nodes = AttackPathsNodeSerializer(many=True) + relationships = AttackPathsRelationshipSerializer(many=True) + + class JSONAPIMeta: + resource_name = "attack-paths-query-result" + + class ResourceTagSerializer(RLSSerializer): """ Serializer for the ResourceTag model diff --git a/api/src/backend/api/v1/urls.py b/api/src/backend/api/v1/urls.py index d879d1476b..840f027b42 100644 --- a/api/src/backend/api/v1/urls.py +++ b/api/src/backend/api/v1/urls.py @@ -4,6 +4,7 @@ from drf_spectacular.views import SpectacularRedocView from rest_framework_nested import routers from api.v1.views import ( + AttackPathsScanViewSet, ComplianceOverviewViewSet, CustomSAMLLoginView, CustomTokenObtainView, @@ -53,6 +54,9 @@ router.register(r"tenants", TenantViewSet, basename="tenant") router.register(r"providers", ProviderViewSet, basename="provider") router.register(r"provider-groups", ProviderGroupViewSet, basename="providergroup") router.register(r"scans", ScanViewSet, basename="scan") +router.register( + r"attack-paths-scans", AttackPathsScanViewSet, basename="attack-paths-scans" +) router.register(r"tasks", TaskViewSet, basename="task") router.register(r"resources", ResourceViewSet, basename="resource") router.register(r"findings", FindingViewSet, basename="finding") diff --git a/api/src/backend/api/v1/views.py b/api/src/backend/api/v1/views.py index d619f32e09..5b7e484350 100644 --- a/api/src/backend/api/v1/views.py +++ b/api/src/backend/api/v1/views.py @@ -3,6 +3,7 @@ import glob import json import logging import os + from collections import defaultdict from copy import deepcopy from datetime import datetime, timedelta, timezone @@ -10,6 +11,7 @@ from decimal import ROUND_HALF_UP, Decimal, InvalidOperation 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 @@ -41,8 +43,9 @@ from django.db.models import ( Sum, Value, When, + Window, ) -from django.db.models.functions import Coalesce +from django.db.models.functions import Coalesce, RowNumber from django.http import HttpResponse, QueryDict from django.shortcuts import redirect from django.urls import reverse @@ -72,22 +75,12 @@ from rest_framework.generics import GenericAPIView, get_object_or_404 from rest_framework.permissions import SAFE_METHODS from rest_framework_json_api.views import RelationshipView, Response from rest_framework_simplejwt.exceptions import InvalidToken, TokenError -from tasks.beat import schedule_provider_scan -from tasks.jobs.export import get_s3_client -from tasks.tasks import ( - backfill_scan_resource_summaries_task, - check_integration_connection_task, - check_lighthouse_connection_task, - check_lighthouse_provider_connection_task, - check_provider_connection_task, - delete_provider_task, - delete_tenant_task, - jira_integration_task, - mute_historical_findings_task, - perform_scan_task, - refresh_lighthouse_provider_models_task, -) +from api.attack_paths import ( + get_queries_for_provider, + get_query_by_id, + views_helpers as attack_paths_views_helpers, +) from api.base_views import BaseRLSViewSet, BaseTenantViewset, BaseUserViewset from api.compliance import ( PROWLER_COMPLIANCE_OVERVIEW_TEMPLATE, @@ -105,6 +98,7 @@ from api.filters import ( InvitationFilter, LatestFindingFilter, LatestResourceFilter, + AttackPathsScanFilter, LighthouseProviderConfigFilter, LighthouseProviderModelsFilter, MembershipFilter, @@ -129,6 +123,7 @@ from api.models import ( Finding, Integration, Invitation, + AttackPathsScan, LighthouseConfiguration, LighthouseProviderConfiguration, LighthouseProviderModels, @@ -170,6 +165,10 @@ from api.utils import ( from api.uuid_utils import datetime_to_uuid7, uuid7_start from api.v1.mixins import DisablePaginationMixin, PaginateByPkMixin, TaskManagementMixin from api.v1.serializers import ( + AttackPathsQueryRunRequestSerializer, + AttackPathsQuerySerializer, + AttackPathsQueryResultSerializer, + AttackPathsScanSerializer, ComplianceOverviewAttributesSerializer, ComplianceOverviewDetailSerializer, ComplianceOverviewDetailThreatscoreSerializer, @@ -247,6 +246,22 @@ from api.v1.serializers import ( UserSerializer, UserUpdateSerializer, ) +from tasks.beat import schedule_provider_scan +from tasks.jobs.attack_paths import db_utils as attack_paths_db_utils +from tasks.jobs.export import get_s3_client +from tasks.tasks import ( + backfill_scan_resource_summaries_task, + check_integration_connection_task, + check_lighthouse_connection_task, + check_lighthouse_provider_connection_task, + check_provider_connection_task, + delete_provider_task, + delete_tenant_task, + jira_integration_task, + mute_historical_findings_task, + perform_scan_task, + refresh_lighthouse_provider_models_task, +) logger = logging.getLogger(BackendLogger.API) @@ -390,6 +405,10 @@ class SchemaView(SpectacularAPIView): "name": "Scan", "description": "Endpoints for triggering manual scans and viewing scan results.", }, + { + "name": "Attack Paths", + "description": "Endpoints for Attack Paths scan status and executing Attack Paths queries.", + }, { "name": "Schedule", "description": "Endpoints for managing scan schedules, allowing configuration of automated " @@ -2140,6 +2159,12 @@ class ScanViewSet(BaseRLSViewSet): }, ) + attack_paths_db_utils.create_attack_paths_scan( + tenant_id=self.request.tenant_id, + scan_id=str(scan.id), + provider_id=str(scan.provider_id), + ) + prowler_task = Task.objects.get(id=task.id) scan.task_id = task.id scan.save(update_fields=["task_id"]) @@ -2220,6 +2245,187 @@ class TaskViewSet(BaseRLSViewSet): ) +@extend_schema_view( + list=extend_schema( + tags=["Attack Paths"], + summary="List Attack Paths scans", + description="Retrieve Attack Paths scans for the tenant with support for filtering, ordering, and pagination.", + ), + retrieve=extend_schema( + tags=["Attack Paths"], + summary="Retrieve Attack Paths scan details", + description="Fetch full details for a specific Attack Paths scan.", + ), + attack_paths_queries=extend_schema( + tags=["Attack Paths"], + summary="List attack paths queries", + description="Retrieve the catalog of Attack Paths queries available for this Attack Paths scan.", + responses={ + 200: OpenApiResponse(AttackPathsQuerySerializer(many=True)), + 404: OpenApiResponse( + description="No queries found for the selected provider" + ), + }, + ), + run_attack_paths_query=extend_schema( + tags=["Attack Paths"], + summary="Execute an Attack Paths query", + description="Execute the selected Attack Paths query against the Attack Paths graph and return the resulting subgraph.", + request=AttackPathsQueryRunRequestSerializer, + responses={ + 200: OpenApiResponse(AttackPathsQueryResultSerializer), + 400: OpenApiResponse( + description="Bad request (e.g., Unknown Attack Paths query for the selected provider)" + ), + 404: OpenApiResponse( + description="No attack paths found for the given query and parameters" + ), + 500: OpenApiResponse( + description="Attack Paths query execution failed due to a database error" + ), + }, + ), +) +class AttackPathsScanViewSet(BaseRLSViewSet): + queryset = AttackPathsScan.objects.all() + serializer_class = AttackPathsScanSerializer + http_method_names = ["get", "post"] + filterset_class = AttackPathsScanFilter + ordering = ["-inserted_at"] + ordering_fields = [ + "inserted_at", + "started_at", + ] + # RBAC required permissions + required_permissions = [Permissions.MANAGE_SCANS] + + def set_required_permissions(self): + if self.request.method in SAFE_METHODS: + self.required_permissions = [] + + else: + self.required_permissions = [Permissions.MANAGE_SCANS] + + def get_serializer_class(self): + if self.action == "run_attack_paths_query": + return AttackPathsQueryRunRequestSerializer + + return super().get_serializer_class() + + def get_queryset(self): + user_roles = get_role(self.request.user) + base_queryset = AttackPathsScan.objects.filter(tenant_id=self.request.tenant_id) + + if user_roles.unlimited_visibility: + queryset = base_queryset + + else: + queryset = base_queryset.filter(provider__in=get_providers(user_roles)) + + return queryset.select_related("provider", "scan", "task") + + def list(self, request, *args, **kwargs): + queryset = self.filter_queryset(self.get_queryset()) + + latest_per_provider = queryset.annotate( + latest_scan_rank=Window( + expression=RowNumber(), + partition_by=[F("provider_id")], + order_by=[F("inserted_at").desc()], + ) + ).filter(latest_scan_rank=1) + + page = self.paginate_queryset(latest_per_provider) + if page is not None: + serializer = self.get_serializer(page, many=True) + return self.get_paginated_response(serializer.data) + + serializer = self.get_serializer(latest_per_provider, many=True) + return Response(serializer.data) + + @extend_schema(exclude=True) + def create(self, request, *args, **kwargs): + raise MethodNotAllowed(method="POST") + + @extend_schema(exclude=True) + def destroy(self, request, *args, **kwargs): + raise MethodNotAllowed(method="DELETE") + + @action( + detail=True, + methods=["get"], + url_path="queries", + url_name="queries", + ) + def attack_paths_queries(self, request, pk=None): + attack_paths_scan = self.get_object() + queries = get_queries_for_provider(attack_paths_scan.provider.provider) + + if not queries: + return Response( + {"detail": "No queries found for the selected provider"}, + status=status.HTTP_404_NOT_FOUND, + ) + + serializer = AttackPathsQuerySerializer(queries, many=True) + return Response(serializer.data, status=status.HTTP_200_OK) + + @action( + detail=True, + methods=["post"], + url_path="queries/run", + url_name="queries-run", + ) + def run_attack_paths_query(self, request, pk=None): + attack_paths_scan = self.get_object() + + if attack_paths_scan.state != StateChoices.COMPLETED: + raise ValidationError( + { + "detail": "The Attack Paths scan must be completed before running Attack Paths queries" + } + ) + + 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) + + query_definition = get_query_by_id(serializer.validated_data["id"]) + if ( + query_definition is None + or query_definition.provider != attack_paths_scan.provider.provider + ): + raise ValidationError( + {"id": "Unknown Attack Paths query for the selected provider"} + ) + + parameters = attack_paths_views_helpers.prepare_query_parameters( + query_definition, + serializer.validated_data.get("parameters", {}), + attack_paths_scan.provider.uid, + ) + + graph = attack_paths_views_helpers.execute_attack_paths_query( + attack_paths_scan, query_definition, parameters + ) + + status_code = status.HTTP_200_OK + if not graph.get("nodes"): + status_code = status.HTTP_404_NOT_FOUND + + response_serializer = AttackPathsQueryResultSerializer(graph) + return Response(response_serializer.data, status=status_code) + + @extend_schema_view( list=extend_schema( tags=["Resource"], @@ -5143,7 +5349,7 @@ class TenantApiKeyViewSet(BaseRLSViewSet): @extend_schema(exclude=True) def destroy(self, request, *args, **kwargs): - raise MethodNotAllowed(method="DESTROY") + raise MethodNotAllowed(method="DELETE") @action(detail=True, methods=["delete"]) def revoke(self, request, *args, **kwargs): diff --git a/api/src/backend/config/celery.py b/api/src/backend/config/celery.py index b3a0ab4b68..aaa1b1c386 100644 --- a/api/src/backend/config/celery.py +++ b/api/src/backend/config/celery.py @@ -1,6 +1,7 @@ import warnings from celery import Celery, Task + from config.env import env # Suppress specific warnings from django-rest-auth: https://github.com/iMerica/dj-rest-auth/issues/684 diff --git a/api/src/backend/config/django/devel.py b/api/src/backend/config/django/devel.py index 12c3c384c7..73662980f6 100644 --- a/api/src/backend/config/django/devel.py +++ b/api/src/backend/config/django/devel.py @@ -36,6 +36,12 @@ DATABASES = { "HOST": env("POSTGRES_REPLICA_HOST", default=default_db_host), "PORT": env("POSTGRES_REPLICA_PORT", default=default_db_port), }, + "neo4j": { + "HOST": env.str("NEO4J_HOST", "neo4j"), + "PORT": env.str("NEO4J_PORT", "7687"), + "USER": env.str("NEO4J_USER", "neo4j"), + "PASSWORD": env.str("NEO4J_PASSWORD", "neo4j_password"), + }, } DATABASES["default"] = DATABASES["prowler_user"] diff --git a/api/src/backend/config/django/production.py b/api/src/backend/config/django/production.py index 5a4cc73044..1b0fc90962 100644 --- a/api/src/backend/config/django/production.py +++ b/api/src/backend/config/django/production.py @@ -37,6 +37,12 @@ DATABASES = { "HOST": env("POSTGRES_REPLICA_HOST", default=default_db_host), "PORT": env("POSTGRES_REPLICA_PORT", default=default_db_port), }, + "neo4j": { + "HOST": env.str("NEO4J_HOST"), + "PORT": env.str("NEO4J_PORT"), + "USER": env.str("NEO4J_USER"), + "PASSWORD": env.str("NEO4J_PASSWORD"), + }, } DATABASES["default"] = DATABASES["prowler_user"] diff --git a/api/src/backend/conftest.py b/api/src/backend/conftest.py index 2988822da2..20221aca2a 100644 --- a/api/src/backend/conftest.py +++ b/api/src/backend/conftest.py @@ -1,8 +1,11 @@ import logging +from types import SimpleNamespace + from datetime import datetime, timedelta, timezone 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 @@ -11,10 +14,14 @@ from django.urls import reverse from django_celery_results.models import TaskResult from rest_framework import status from rest_framework.test import APIClient -from tasks.jobs.backfill import backfill_resource_scan_summaries +from api.attack_paths import ( + AttackPathsQueryDefinition, + AttackPathsQueryParameterDefinition, +) from api.db_utils import rls_transaction from api.models import ( + AttackPathsScan, ComplianceOverview, ComplianceRequirementOverview, Finding, @@ -47,6 +54,7 @@ from api.rls import Tenant from api.v1.serializers import TokenSerializer from prowler.lib.check.models import Severity from prowler.lib.outputs.finding import Status +from tasks.jobs.backfill import backfill_resource_scan_summaries TODAY = str(datetime.today().date()) API_JSON_CONTENT_TYPE = "application/vnd.api+json" @@ -159,22 +167,20 @@ def create_test_user_rbac_no_roles(django_db_setup, django_db_blocker, tenants_f @pytest.fixture(scope="function") -def create_test_user_rbac_limited(django_db_setup, django_db_blocker): +def create_test_user_rbac_limited(django_db_setup, django_db_blocker, tenants_fixture): with django_db_blocker.unblock(): user = User.objects.create_user( name="testing_limited", email="rbac_limited@rbac.com", password=TEST_PASSWORD, ) - tenant = Tenant.objects.create( - name="Tenant Test", - ) + tenant = tenants_fixture[0] Membership.objects.create( user=user, tenant=tenant, role=Membership.RoleChoices.OWNER, ) - Role.objects.create( + role = Role.objects.create( name="limited", tenant_id=tenant.id, manage_users=False, @@ -187,7 +193,7 @@ def create_test_user_rbac_limited(django_db_setup, django_db_blocker): ) UserRoleRelationship.objects.create( user=user, - role=Role.objects.get(name="limited"), + role=role, tenant_id=tenant.id, ) return user @@ -1469,6 +1475,104 @@ def mute_rules_fixture(tenants_fixture, create_test_user, findings_fixture): return mute_rule1, mute_rule2 +@pytest.fixture +def create_attack_paths_scan(): + """Factory fixture to create Attack Paths scans for tests.""" + + def _create( + provider, + *, + scan=None, + state=StateChoices.COMPLETED, + progress=0, + graph_database="tenant-db", + **extra_fields, + ): + scan_instance = scan or Scan.objects.create( + name=extra_fields.pop("scan_name", "Attack Paths Supporting Scan"), + provider=provider, + trigger=Scan.TriggerChoices.MANUAL, + state=extra_fields.pop("scan_state", StateChoices.COMPLETED), + tenant_id=provider.tenant_id, + ) + + payload = { + "tenant_id": provider.tenant_id, + "provider": provider, + "scan": scan_instance, + "state": state, + "progress": progress, + "graph_database": graph_database, + } + payload.update(extra_fields) + + return AttackPathsScan.objects.create(**payload) + + return _create + + +@pytest.fixture +def attack_paths_query_definition_factory(): + """Factory fixture for building Attack Paths query definitions.""" + + def _create(**overrides): + cast_type = overrides.pop("cast_type", str) + parameters = overrides.pop( + "parameters", + [ + AttackPathsQueryParameterDefinition( + name="limit", + label="Limit", + cast=cast_type, + ) + ], + ) + definition_payload = { + "id": "aws-test", + "name": "Attack Paths Test Query", + "description": "Synthetic Attack Paths definition for tests.", + "provider": "aws", + "cypher": "RETURN 1", + "parameters": parameters, + } + definition_payload.update(overrides) + return AttackPathsQueryDefinition(**definition_payload) + + return _create + + +@pytest.fixture +def attack_paths_graph_stub_classes(): + """Provide lightweight graph element stubs for Attack Paths serialization tests.""" + + class AttackPathsNativeValue: + def __init__(self, value): + self._value = value + + def to_native(self): + return self._value + + class AttackPathsNode: + def __init__(self, element_id, labels, properties): + self.element_id = element_id + self.labels = labels + self._properties = properties + + class AttackPathsRelationship: + def __init__(self, element_id, rel_type, start_node, end_node, properties): + self.element_id = element_id + self.type = rel_type + self.start_node = start_node + self.end_node = end_node + self._properties = properties + + return SimpleNamespace( + NativeValue=AttackPathsNativeValue, + Node=AttackPathsNode, + Relationship=AttackPathsRelationship, + ) + + def get_authorization_header(access_token: str) -> dict: return {"Authorization": f"Bearer {access_token}"} diff --git a/api/src/backend/tasks/beat.py b/api/src/backend/tasks/beat.py index a7795e6909..e42f9918ab 100644 --- a/api/src/backend/tasks/beat.py +++ b/api/src/backend/tasks/beat.py @@ -7,6 +7,7 @@ from tasks.tasks import perform_scheduled_scan_task from api.db_utils import rls_transaction from api.exceptions import ConflictException from api.models import Provider, Scan, StateChoices +from tasks.jobs.attack_paths import db_utils as attack_paths_db_utils def schedule_provider_scan(provider_instance: Provider): @@ -39,6 +40,12 @@ def schedule_provider_scan(provider_instance: Provider): scheduled_at=datetime.now(timezone.utc), ) + attack_paths_db_utils.create_attack_paths_scan( + tenant_id=tenant_id, + scan_id=str(scheduled_scan.id), + provider_id=provider_id, + ) + # Schedule the task periodic_task_instance = PeriodicTask.objects.create( interval=schedule, diff --git a/api/src/backend/tasks/jobs/attack_paths/__init__.py b/api/src/backend/tasks/jobs/attack_paths/__init__.py new file mode 100644 index 0000000000..0f586cdcf6 --- /dev/null +++ b/api/src/backend/tasks/jobs/attack_paths/__init__.py @@ -0,0 +1,5 @@ +from tasks.jobs.attack_paths.scan import run as attack_paths_scan + +__all__ = [ + "attack_paths_scan", +] diff --git a/api/src/backend/tasks/jobs/attack_paths/aws.py b/api/src/backend/tasks/jobs/attack_paths/aws.py new file mode 100644 index 0000000000..66b132f305 --- /dev/null +++ b/api/src/backend/tasks/jobs/attack_paths/aws.py @@ -0,0 +1,216 @@ +# Portions of this file are based on code from the Cartography project +# (https://github.com/cartography-cncf/cartography), which is licensed under the Apache 2.0 License. + +from typing import Any + +import boto3 +import neo4j + +from cartography.config import Config as CartographyConfig +from cartography.intel import aws as cartography_aws +from celery.utils.log import get_task_logger + +from api.models import ( + AttackPathsScan as ProwlerAPIAttackPathsScan, + Provider as ProwlerAPIProvider, +) +from prowler.providers.common.provider import Provider as ProwlerSDKProvider +from tasks.jobs.attack_paths import db_utils, utils + +logger = get_task_logger(__name__) + + +def start_aws_ingestion( + neo4j_session: neo4j.Session, + cartography_config: CartographyConfig, + prowler_api_provider: ProwlerAPIProvider, + prowler_sdk_provider: ProwlerSDKProvider, + attack_paths_scan: ProwlerAPIAttackPathsScan, +) -> dict[str, dict[str, str]]: + """ + Code based on Cartography version 0.117.0, specifically on `cartography.intel.aws.__init__.py`. + + For the scan progress updates: + - The caller of this function (`tasks.jobs.attack_paths.scan.run`) has set it to 2. + - When the control returns to the caller, it will be set to 95. + """ + + # Initialize variables common to all jobs + common_job_parameters = { + "UPDATE_TAG": cartography_config.update_tag, + "permission_relationships_file": cartography_config.permission_relationships_file, + "aws_guardduty_severity_threshold": cartography_config.aws_guardduty_severity_threshold, + "aws_cloudtrail_management_events_lookback_hours": cartography_config.aws_cloudtrail_management_events_lookback_hours, + } + + boto3_session = get_boto3_session(prowler_api_provider, prowler_sdk_provider) + regions: list[str] = list(prowler_sdk_provider._enabled_regions) + requested_syncs = list(cartography_aws.RESOURCE_FUNCTIONS.keys()) + + sync_args = cartography_aws._build_aws_sync_kwargs( + neo4j_session, + boto3_session, + regions, + prowler_api_provider.uid, + cartography_config.update_tag, + common_job_parameters, + ) + + # Starting with sync functions + cartography_aws.organizations.sync( + neo4j_session, + {prowler_api_provider.alias: prowler_api_provider.uid}, + cartography_config.update_tag, + common_job_parameters, + ) + db_utils.update_attack_paths_scan_progress(attack_paths_scan, 3) + + # Adding an extra field + common_job_parameters["AWS_ID"] = prowler_api_provider.uid + + cartography_aws._autodiscover_accounts( + neo4j_session, + boto3_session, + prowler_api_provider.uid, + cartography_config.update_tag, + common_job_parameters, + ) + db_utils.update_attack_paths_scan_progress(attack_paths_scan, 4) + + failed_syncs = sync_aws_account( + prowler_api_provider, requested_syncs, sync_args, attack_paths_scan + ) + + if "permission_relationships" in requested_syncs: + cartography_aws.RESOURCE_FUNCTIONS["permission_relationships"](**sync_args) + db_utils.update_attack_paths_scan_progress(attack_paths_scan, 88) + + if "resourcegroupstaggingapi" in requested_syncs: + cartography_aws.RESOURCE_FUNCTIONS["resourcegroupstaggingapi"](**sync_args) + db_utils.update_attack_paths_scan_progress(attack_paths_scan, 89) + + cartography_aws.run_scoped_analysis_job( + "aws_ec2_iaminstanceprofile.json", + neo4j_session, + common_job_parameters, + ) + db_utils.update_attack_paths_scan_progress(attack_paths_scan, 90) + + cartography_aws.run_analysis_job( + "aws_lambda_ecr.json", + neo4j_session, + common_job_parameters, + ) + db_utils.update_attack_paths_scan_progress(attack_paths_scan, 91) + + cartography_aws.merge_module_sync_metadata( + neo4j_session, + group_type="AWSAccount", + group_id=prowler_api_provider.uid, + synced_type="AWSAccount", + update_tag=cartography_config.update_tag, + stat_handler=cartography_aws.stat_handler, + ) + db_utils.update_attack_paths_scan_progress(attack_paths_scan, 92) + + # Removing the added extra field + del common_job_parameters["AWS_ID"] + + cartography_aws.run_cleanup_job( + "aws_post_ingestion_principals_cleanup.json", + neo4j_session, + common_job_parameters, + ) + db_utils.update_attack_paths_scan_progress(attack_paths_scan, 93) + + cartography_aws._perform_aws_analysis( + requested_syncs, neo4j_session, common_job_parameters + ) + db_utils.update_attack_paths_scan_progress(attack_paths_scan, 94) + + return failed_syncs + + +def get_boto3_session( + prowler_api_provider: ProwlerAPIProvider, prowler_sdk_provider: ProwlerSDKProvider +) -> boto3.Session: + boto3_session = prowler_sdk_provider.session.current_session + + aws_accounts_from_session = cartography_aws.organizations.get_aws_account_default( + boto3_session + ) + if not aws_accounts_from_session: + raise Exception( + "No valid AWS credentials could be found. No AWS accounts can be synced." + ) + + aws_account_id_from_session = list(aws_accounts_from_session.values())[0] + if prowler_api_provider.uid != aws_account_id_from_session: + raise Exception( + f"Provider {prowler_api_provider.uid} doesn't match AWS account {aws_account_id_from_session}." + ) + + if boto3_session.region_name is None: + global_region = prowler_sdk_provider.get_global_region() + boto3_session._session.set_config_variable("region", global_region) + + return boto3_session + + +def sync_aws_account( + prowler_api_provider: ProwlerAPIProvider, + requested_syncs: list[str], + sync_args: dict[str, Any], + attack_paths_scan: ProwlerAPIAttackPathsScan, +) -> dict[str, str]: + current_progress = 4 # `cartography_aws._autodiscover_accounts` + max_progress = ( + 87 # `cartography_aws.RESOURCE_FUNCTIONS["permission_relationships"]` - 1 + ) + n_steps = len(requested_syncs) + progress_step = (max_progress - current_progress) / n_steps + + failed_syncs = {} + + for func_name in requested_syncs: + if func_name in cartography_aws.RESOURCE_FUNCTIONS: + logger.info( + f"Syncing function {func_name} for AWS account {prowler_api_provider.uid}" + ) + + # Updating progress, not really the right place but good enough + current_progress += progress_step + db_utils.update_attack_paths_scan_progress( + attack_paths_scan, int(current_progress) + ) + + try: + # Skip permission relationships and tags for now because they rely on data already being in the graph + if func_name not in [ + "permission_relationships", + "resourcegroupstaggingapi", + ]: + cartography_aws.RESOURCE_FUNCTIONS[func_name](**sync_args) + + else: + continue + + except Exception as e: + exception_message = utils.stringify_exception( + e, f"Exception for AWS sync function: {func_name}" + ) + failed_syncs[func_name] = exception_message + + logger.warning( + f"Caught exception syncing function {func_name} from AWS account {prowler_api_provider.uid}. We " + "are continuing on to the next AWS sync function.", + ) + + continue + + else: + raise ValueError( + f'AWS sync function "{func_name}" was specified but does not exist. Did you misspell it?' + ) + + return failed_syncs diff --git a/api/src/backend/tasks/jobs/attack_paths/db_utils.py b/api/src/backend/tasks/jobs/attack_paths/db_utils.py new file mode 100644 index 0000000000..2565dfa872 --- /dev/null +++ b/api/src/backend/tasks/jobs/attack_paths/db_utils.py @@ -0,0 +1,158 @@ +from datetime import datetime, timezone +from typing import Any +from uuid import UUID + +from cartography.config import Config as CartographyConfig + +from api.db_utils import rls_transaction +from api.models import ( + AttackPathsScan as ProwlerAPIAttackPathsScan, + Provider as ProwlerAPIProvider, + StateChoices, +) +from tasks.jobs.attack_paths.providers import is_provider_available + + +def create_attack_paths_scan( + tenant_id: str, + scan_id: str, + provider_id: int, +) -> ProwlerAPIAttackPathsScan | None: + with rls_transaction(tenant_id): + prowler_api_provider = ProwlerAPIProvider.objects.get(id=provider_id) + + if not is_provider_available(prowler_api_provider.provider): + return None + + with rls_transaction(tenant_id): + attack_paths_scan = ProwlerAPIAttackPathsScan.objects.create( + tenant_id=tenant_id, + provider_id=provider_id, + scan_id=scan_id, + state=StateChoices.SCHEDULED, + started_at=datetime.now(tz=timezone.utc), + ) + attack_paths_scan.save() + + return attack_paths_scan + + +def retrieve_attack_paths_scan( + tenant_id: str, + scan_id: str, +) -> ProwlerAPIAttackPathsScan | None: + try: + with rls_transaction(tenant_id): + attack_paths_scan = ProwlerAPIAttackPathsScan.objects.get( + scan_id=scan_id, + ) + + return attack_paths_scan + + except ProwlerAPIAttackPathsScan.DoesNotExist: + return None + + +def starting_attack_paths_scan( + attack_paths_scan: ProwlerAPIAttackPathsScan, + task_id: str, + cartography_config: CartographyConfig, +) -> None: + with rls_transaction(attack_paths_scan.tenant_id): + attack_paths_scan.task_id = task_id + attack_paths_scan.state = StateChoices.EXECUTING + attack_paths_scan.started_at = datetime.now(tz=timezone.utc) + attack_paths_scan.update_tag = cartography_config.update_tag + attack_paths_scan.graph_database = cartography_config.neo4j_database + + attack_paths_scan.save( + update_fields=[ + "task_id", + "state", + "started_at", + "update_tag", + "graph_database", + ] + ) + + +def finish_attack_paths_scan( + attack_paths_scan: ProwlerAPIAttackPathsScan, + state: StateChoices, + ingestion_exceptions: dict[str, Any], +) -> None: + with rls_transaction(attack_paths_scan.tenant_id): + now = datetime.now(tz=timezone.utc) + duration = int((now - attack_paths_scan.started_at).total_seconds()) + + attack_paths_scan.state = state + attack_paths_scan.progress = 100 + attack_paths_scan.completed_at = now + attack_paths_scan.duration = duration + attack_paths_scan.ingestion_exceptions = ingestion_exceptions + + attack_paths_scan.save( + update_fields=[ + "state", + "progress", + "completed_at", + "duration", + "ingestion_exceptions", + ] + ) + + +def update_attack_paths_scan_progress( + attack_paths_scan: ProwlerAPIAttackPathsScan, + progress: int, +) -> None: + with rls_transaction(attack_paths_scan.tenant_id): + attack_paths_scan.progress = progress + attack_paths_scan.save(update_fields=["progress"]) + + +def get_old_attack_paths_scans( + tenant_id: str, + provider_id: str, + attack_paths_scan_id: str, +) -> list[ProwlerAPIAttackPathsScan]: + """ + An `old_attack_paths_scan` is any `completed` Attack Paths scan for the same provider, + with its graph database not deleted, excluding the current Attack Paths scan. + """ + + with rls_transaction(tenant_id): + completed_scans_qs = ( + ProwlerAPIAttackPathsScan.objects.filter( + provider_id=provider_id, + state=StateChoices.COMPLETED, + is_graph_database_deleted=False, + ) + .exclude(id=attack_paths_scan_id) + .all() + ) + + return list(completed_scans_qs) + + +def update_old_attack_paths_scan( + old_attack_paths_scan: ProwlerAPIAttackPathsScan, +) -> None: + with rls_transaction(old_attack_paths_scan.tenant_id): + old_attack_paths_scan.is_graph_database_deleted = True + old_attack_paths_scan.save(update_fields=["is_graph_database_deleted"]) + + +def get_provider_graph_database_names(tenant_id: str, provider_id: str) -> list[str]: + """ + Return existing graph database names for a tenant/provider. + + Note: For accesing the `AttackPathsScan` we need to use `all_objects` manager because the provider is soft-deleted. + """ + with rls_transaction(tenant_id): + graph_databases_names_qs = ProwlerAPIAttackPathsScan.all_objects.filter( + provider_id=provider_id, + is_graph_database_deleted=False, + ).values_list("graph_database", flat=True) + + return list(graph_databases_names_qs) diff --git a/api/src/backend/tasks/jobs/attack_paths/providers.py b/api/src/backend/tasks/jobs/attack_paths/providers.py new file mode 100644 index 0000000000..a0d4c44551 --- /dev/null +++ b/api/src/backend/tasks/jobs/attack_paths/providers.py @@ -0,0 +1,23 @@ +AVAILABLE_PROVIDERS: list[str] = [ + "aws", +] + +ROOT_NODE_LABELS: dict[str, str] = { + "aws": "AWSAccount", +} + +NODE_UID_FIELDS: dict[str, str] = { + "aws": "arn", +} + + +def is_provider_available(provider_type: str) -> bool: + return provider_type in AVAILABLE_PROVIDERS + + +def get_root_node_label(provider_type: str) -> str: + return ROOT_NODE_LABELS.get(provider_type, "UnknownProviderAccount") + + +def get_node_uid_field(provider_type: str) -> str: + return NODE_UID_FIELDS.get(provider_type, "UnknownProviderUID") diff --git a/api/src/backend/tasks/jobs/attack_paths/prowler.py b/api/src/backend/tasks/jobs/attack_paths/prowler.py new file mode 100644 index 0000000000..366fe2da27 --- /dev/null +++ b/api/src/backend/tasks/jobs/attack_paths/prowler.py @@ -0,0 +1,205 @@ +import neo4j + +from cartography.client.core.tx import run_write_query +from cartography.config import Config as CartographyConfig +from celery.utils.log import get_task_logger + +from api.db_utils import rls_transaction +from api.models import Provider, ResourceFindingMapping +from config.env import env +from prowler.config import config as ProwlerConfig +from tasks.jobs.attack_paths.providers import get_node_uid_field, get_root_node_label + +logger = get_task_logger(__name__) + +BATCH_SIZE = env.int("NEO4J_INSERT_BATCH_SIZE", 500) + +INDEX_STATEMENTS = [ + "CREATE INDEX prowler_finding_id IF NOT EXISTS FOR (n:ProwlerFinding) ON (n.id);", + "CREATE INDEX prowler_finding_provider_uid IF NOT EXISTS FOR (n:ProwlerFinding) ON (n.provider_uid);", + "CREATE INDEX prowler_finding_lastupdated IF NOT EXISTS FOR (n:ProwlerFinding) ON (n.lastupdated);", + "CREATE INDEX prowler_finding_check_id IF NOT EXISTS FOR (n:ProwlerFinding) ON (n.status);", +] + +INSERT_STATEMENT_TEMPLATE = """ + UNWIND $findings_data AS finding_data + + MATCH (account:__ROOT_NODE_LABEL__ {id: $provider_uid}) + MATCH (account)-->(resource) + WHERE resource.__NODE_UID_FIELD__ = finding_data.resource_uid + OR resource.id = finding_data.resource_uid + + MERGE (finding:ProwlerFinding {id: finding_data.id}) + ON CREATE SET + finding.id = finding_data.id, + finding.uid = finding_data.uid, + finding.inserted_at = finding_data.inserted_at, + finding.updated_at = finding_data.updated_at, + finding.first_seen_at = finding_data.first_seen_at, + finding.scan_id = finding_data.scan_id, + finding.delta = finding_data.delta, + finding.status = finding_data.status, + finding.status_extended = finding_data.status_extended, + finding.severity = finding_data.severity, + finding.check_id = finding_data.check_id, + finding.check_title = finding_data.check_title, + finding.muted = finding_data.muted, + finding.muted_reason = finding_data.muted_reason, + finding.provider_uid = $provider_uid, + finding.firstseen = timestamp(), + finding.lastupdated = $last_updated, + finding._module_name = 'cartography:prowler', + finding._module_version = $prowler_version + ON MATCH SET + finding.status = finding_data.status, + finding.status_extended = finding_data.status_extended, + finding.lastupdated = $last_updated + + MERGE (resource)-[rel:HAS_FINDING]->(finding) + ON CREATE SET + rel.provider_uid = $provider_uid, + rel.firstseen = timestamp(), + rel.lastupdated = $last_updated, + rel._module_name = 'cartography:prowler', + rel._module_version = $prowler_version + ON MATCH SET + rel.lastupdated = $last_updated +""" + +CLEANUP_STATEMENT = """ + MATCH (finding:ProwlerFinding {provider_uid: $provider_uid}) + WHERE finding.lastupdated < $last_updated + + WITH finding LIMIT $batch_size + + DETACH DELETE finding + + RETURN COUNT(finding) AS deleted_findings_count +""" + + +def create_indexes(neo4j_session: neo4j.Session) -> None: + """ + Code based on Cartography version 0.117.0, specifically on `cartography.intel.create_indexes.run`. + """ + + logger.info("Creating indexes for Prowler node types.") + for statement in INDEX_STATEMENTS: + logger.debug("Executing statement: %s", statement) + run_write_query(neo4j_session, statement) + + +def analysis( + neo4j_session: neo4j.Session, + prowler_api_provider: Provider, + scan_id: str, + config: CartographyConfig, +) -> None: + findings_data = get_provider_last_scan_findings(prowler_api_provider, scan_id) + load_findings(neo4j_session, findings_data, prowler_api_provider, config) + cleanup_findings(neo4j_session, prowler_api_provider, config) + + +def get_provider_last_scan_findings( + prowler_api_provider: Provider, + scan_id: str, +) -> list[dict[str, str]]: + with rls_transaction(prowler_api_provider.tenant_id): + resource_finding_qs = ResourceFindingMapping.objects.filter( + finding__scan_id=scan_id, + ).values( + "resource__uid", + "finding__id", + "finding__uid", + "finding__inserted_at", + "finding__updated_at", + "finding__first_seen_at", + "finding__scan_id", + "finding__delta", + "finding__status", + "finding__status_extended", + "finding__severity", + "finding__check_id", + "finding__check_metadata__checktitle", + "finding__muted", + "finding__muted_reason", + ) + + findings = [] + for resource_finding in resource_finding_qs: + findings.append( + { + "resource_uid": str(resource_finding["resource__uid"]), + "id": str(resource_finding["finding__id"]), + "uid": resource_finding["finding__uid"], + "inserted_at": resource_finding["finding__inserted_at"], + "updated_at": resource_finding["finding__updated_at"], + "first_seen_at": resource_finding["finding__first_seen_at"], + "scan_id": str(resource_finding["finding__scan_id"]), + "delta": resource_finding["finding__delta"], + "status": resource_finding["finding__status"], + "status_extended": resource_finding["finding__status_extended"], + "severity": resource_finding["finding__severity"], + "check_id": str(resource_finding["finding__check_id"]), + "check_title": resource_finding[ + "finding__check_metadata__checktitle" + ], + "muted": resource_finding["finding__muted"], + "muted_reason": resource_finding["finding__muted_reason"], + } + ) + + return findings + + +def load_findings( + neo4j_session: neo4j.Session, + findings_data: list[dict[str, str]], + prowler_api_provider: Provider, + config: CartographyConfig, +) -> None: + replacements = { + "__ROOT_NODE_LABEL__": get_root_node_label(prowler_api_provider.provider), + "__NODE_UID_FIELD__": get_node_uid_field(prowler_api_provider.provider), + } + query = INSERT_STATEMENT_TEMPLATE + for replace_key, replace_value in replacements.items(): + query = query.replace(replace_key, replace_value) + + parameters = { + "provider_uid": str(prowler_api_provider.uid), + "last_updated": config.update_tag, + "prowler_version": ProwlerConfig.prowler_version, + } + + total_length = len(findings_data) + for i in range(0, total_length, BATCH_SIZE): + parameters["findings_data"] = findings_data[i : i + BATCH_SIZE] + + logger.info( + f"Loading findings batch {i // BATCH_SIZE + 1} / {(total_length + BATCH_SIZE - 1) // BATCH_SIZE}" + ) + + neo4j_session.run(query, parameters) + + +def cleanup_findings( + neo4j_session: neo4j.Session, + prowler_api_provider: Provider, + config: CartographyConfig, +) -> None: + parameters = { + "provider_uid": str(prowler_api_provider.uid), + "last_updated": config.update_tag, + "batch_size": BATCH_SIZE, + } + + batch = 1 + deleted_count = 1 + while deleted_count > 0: + logger.info(f"Cleaning findings batch {batch}") + + result = neo4j_session.run(CLEANUP_STATEMENT, parameters) + + deleted_count = result.single().get("deleted_findings_count", 0) + batch += 1 diff --git a/api/src/backend/tasks/jobs/attack_paths/scan.py b/api/src/backend/tasks/jobs/attack_paths/scan.py new file mode 100644 index 0000000000..0538c3f149 --- /dev/null +++ b/api/src/backend/tasks/jobs/attack_paths/scan.py @@ -0,0 +1,163 @@ +import logging +import time +import asyncio + +from typing import Any, Callable + +from cartography.config import Config as CartographyConfig +from cartography.intel import analysis as cartography_analysis +from cartography.intel import create_indexes as cartography_create_indexes +from celery.utils.log import get_task_logger + +from api.attack_paths import database as graph_database +from api.db_utils import rls_transaction +from api.models import ( + Provider as ProwlerAPIProvider, + StateChoices, +) +from api.utils import initialize_prowler_provider +from tasks.jobs.attack_paths import aws, db_utils, prowler, utils + +# Without this Celery goes crazy with Cartography logging +logging.getLogger("cartography").setLevel(logging.ERROR) +logging.getLogger("neo4j").propagate = False + +logger = get_task_logger(__name__) + +CARTOGRAPHY_INGESTION_FUNCTIONS: dict[str, Callable] = { + "aws": aws.start_aws_ingestion, +} + + +def get_cartography_ingestion_function(provider_type: str) -> Callable | None: + return CARTOGRAPHY_INGESTION_FUNCTIONS.get(provider_type) + + +def run(tenant_id: str, scan_id: str, task_id: str) -> dict[str, Any]: + """ + Code based on Cartography version 0.117.0, specifically on `cartography.cli.main`, `cartography.cli.CLI.main`, + `cartography.sync.run_with_config` and `cartography.sync.Sync.run`. + """ + ingestion_exceptions = {} # This will hold any exceptions raised during ingestion + + # Prowler necessary objects + with rls_transaction(tenant_id): + prowler_api_provider = ProwlerAPIProvider.objects.get(scan__pk=scan_id) + prowler_sdk_provider = initialize_prowler_provider(prowler_api_provider) + + # If the provider is still not supported, just return the current `ingestion_exceptions`, that is empty + if not get_cartography_ingestion_function(prowler_api_provider.provider): + return ingestion_exceptions + + # Getting the Attack Paths Scan object and starting it + attack_paths_scan = db_utils.retrieve_attack_paths_scan(tenant_id, scan_id) + if not attack_paths_scan: + logger.warning( + f"No Attack Paths Scan found for scan {scan_id} and tenant {tenant_id}, let's create it then" + ) + attack_paths_scan = db_utils.create_attack_paths_scan( + tenant_id, scan_id, prowler_api_provider.id + ) + + # While creating the Cartography configuration, attributes `neo4j_user` and `neo4j_password` are not really needed in this config object + cartography_config = CartographyConfig( + neo4j_uri=graph_database.get_uri(), + neo4j_database=graph_database.get_database_name(attack_paths_scan.id), + update_tag=int(time.time()), + ) + + # Starting the Attack Paths scan + db_utils.starting_attack_paths_scan(attack_paths_scan, task_id, cartography_config) + + try: + logger.info( + f"Creating Neo4j database {cartography_config.neo4j_database} for tenant {prowler_api_provider.tenant_id}" + ) + + graph_database.create_database(cartography_config.neo4j_database) + db_utils.update_attack_paths_scan_progress(attack_paths_scan, 1) + + logger.info( + f"Starting Cartography ({attack_paths_scan.id}) for " + f"{prowler_api_provider.provider.upper()} provider {prowler_api_provider.id}" + ) + with graph_database.get_session( + cartography_config.neo4j_database + ) as neo4j_session: + # Indexes creation + cartography_create_indexes.run(neo4j_session, cartography_config) + prowler.create_indexes(neo4j_session) + db_utils.update_attack_paths_scan_progress(attack_paths_scan, 2) + + # The real scan, where iterates over cloud services + ingestion_exceptions = _call_within_event_loop( + get_cartography_ingestion_function(prowler_api_provider.provider), + neo4j_session, + cartography_config, + prowler_api_provider, + prowler_sdk_provider, + attack_paths_scan, + ) + + # Post-processing: Just keeping it to be more Cartography compliant + cartography_analysis.run(neo4j_session, cartography_config) + db_utils.update_attack_paths_scan_progress(attack_paths_scan, 95) + + # Adding Prowler nodes and relationships + prowler.analysis( + neo4j_session, prowler_api_provider, scan_id, cartography_config + ) + + logger.info( + f"Completed Cartography ({attack_paths_scan.id}) for " + f"{prowler_api_provider.provider.upper()} provider {prowler_api_provider.id}" + ) + + # Handling databases changes + old_attack_paths_scans = db_utils.get_old_attack_paths_scans( + prowler_api_provider.tenant_id, + prowler_api_provider.id, + attack_paths_scan.id, + ) + for old_attack_paths_scan in old_attack_paths_scans: + graph_database.drop_database(old_attack_paths_scan.graph_database) + db_utils.update_old_attack_paths_scan(old_attack_paths_scan) + + db_utils.finish_attack_paths_scan( + attack_paths_scan, StateChoices.COMPLETED, ingestion_exceptions + ) + return ingestion_exceptions + + except Exception as e: + exception_message = utils.stringify_exception(e, "Cartography failed") + logger.error(exception_message) + ingestion_exceptions["global_cartography_error"] = exception_message + + # Handling databases changes + graph_database.drop_database(cartography_config.neo4j_database) + db_utils.finish_attack_paths_scan( + attack_paths_scan, StateChoices.FAILED, ingestion_exceptions + ) + raise + + +def _call_within_event_loop(fn, *args, **kwargs): + """ + Cartography needs a running event loop, so assuming there is none (Celery task or even regular DRF endpoint), + let's create a new one and set it as the current event loop for this thread. + """ + + loop = asyncio.new_event_loop() + try: + asyncio.set_event_loop(loop) + return fn(*args, **kwargs) + + finally: + try: + loop.run_until_complete(loop.shutdown_asyncgens()) + + except Exception: + pass + + loop.close() + asyncio.set_event_loop(None) diff --git a/api/src/backend/tasks/jobs/attack_paths/utils.py b/api/src/backend/tasks/jobs/attack_paths/utils.py new file mode 100644 index 0000000000..0c737d4158 --- /dev/null +++ b/api/src/backend/tasks/jobs/attack_paths/utils.py @@ -0,0 +1,10 @@ +import traceback + +from datetime import datetime, timezone + + +def stringify_exception(exception: Exception, context: str) -> str: + timestamp = datetime.now(tz=timezone.utc) + exception_traceback = traceback.TracebackException.from_exception(exception) + traceback_string = "".join(exception_traceback.format()) + return f"{timestamp} - {context}\n{traceback_string}" diff --git a/api/src/backend/tasks/jobs/deletion.py b/api/src/backend/tasks/jobs/deletion.py index d72b8de40e..6eee63de6a 100644 --- a/api/src/backend/tasks/jobs/deletion.py +++ b/api/src/backend/tasks/jobs/deletion.py @@ -1,9 +1,19 @@ from celery.utils.log import get_task_logger from django.db import DatabaseError +from api.attack_paths import database as graph_database from api.db_router import MainRouter from api.db_utils import batch_delete, rls_transaction -from api.models import Finding, Provider, Resource, Scan, ScanSummary, Tenant +from api.models import ( + AttackPathsScan, + Finding, + Provider, + Resource, + Scan, + ScanSummary, + Tenant, +) +from tasks.jobs.attack_paths.db_utils import get_provider_graph_database_names logger = get_task_logger(__name__) @@ -23,16 +33,27 @@ def delete_provider(tenant_id: str, pk: str): Raises: Provider.DoesNotExist: If no instance with the provided primary key exists. """ + # Delete the Attack Paths' graph databases related to the provider + graph_database_names = get_provider_graph_database_names(tenant_id, pk) + try: + for graph_database_name in graph_database_names: + graph_database.drop_database(graph_database_name) + except graph_database.GraphDatabaseQueryException as gdb_error: + logger.error(f"Error deleting Provider databases: {gdb_error}") + raise + + # Get all provider related data and delete them in batches with rls_transaction(tenant_id): instance = Provider.all_objects.get(pk=pk) - deletion_summary = {} deletion_steps = [ ("Scan Summaries", ScanSummary.all_objects.filter(scan__provider=instance)), ("Findings", Finding.all_objects.filter(scan__provider=instance)), ("Resources", Resource.all_objects.filter(provider=instance)), ("Scans", Scan.all_objects.filter(provider=instance)), + ("AttackPathsScans", AttackPathsScan.all_objects.filter(provider=instance)), ] + deletion_summary = {} for step_name, queryset in deletion_steps: try: _, step_summary = batch_delete(tenant_id, queryset) @@ -48,6 +69,7 @@ def delete_provider(tenant_id: str, pk: str): except DatabaseError as db_error: logger.error(f"Error deleting Provider: {db_error}") raise + return deletion_summary diff --git a/api/src/backend/tasks/tasks.py b/api/src/backend/tasks/tasks.py index 77b44ff34f..0221996e1b 100644 --- a/api/src/backend/tasks/tasks.py +++ b/api/src/backend/tasks/tasks.py @@ -1,13 +1,26 @@ import os + from datetime import datetime, timedelta, timezone from pathlib import Path from shutil import rmtree from celery import chain, group, shared_task from celery.utils.log import get_task_logger +from django_celery_beat.models import PeriodicTask + +from api.compliance import get_compliance_frameworks +from api.db_router import READ_REPLICA_ALIAS +from api.db_utils import rls_transaction +from api.decorators import set_tenant +from api.models import Finding, Integration, Provider, Scan, ScanSummary, StateChoices +from api.utils import initialize_prowler_provider +from api.v1.serializers import ScanTaskSerializer from config.celery import RLSTask from config.django.base import DJANGO_FINDINGS_BATCH_SIZE, DJANGO_TMP_OUTPUT_DIRECTORY -from django_celery_beat.models import PeriodicTask +from prowler.lib.check.compliance_models import Compliance +from prowler.lib.outputs.compliance.generic.generic import GenericCompliance +from prowler.lib.outputs.finding import Finding as FindingOutput +from tasks.jobs.attack_paths import attack_paths_scan from tasks.jobs.backfill import ( backfill_compliance_summaries, backfill_resource_scan_summaries, @@ -43,17 +56,6 @@ from tasks.jobs.scan import ( ) from tasks.utils import batched, get_next_execution_datetime -from api.compliance import get_compliance_frameworks -from api.db_router import READ_REPLICA_ALIAS -from api.db_utils import rls_transaction -from api.decorators import set_tenant -from api.models import Finding, Integration, Provider, Scan, ScanSummary, StateChoices -from api.utils import initialize_prowler_provider -from api.v1.serializers import ScanTaskSerializer -from prowler.lib.check.compliance_models import Compliance -from prowler.lib.outputs.compliance.generic.generic import GenericCompliance -from prowler.lib.outputs.finding import Finding as FindingOutput - logger = get_task_logger(__name__) @@ -86,6 +88,9 @@ def _perform_scan_complete_tasks(tenant_id: str, scan_id: str, provider_id: str) ), ), ).apply_async() + perform_attack_paths_scan_task.apply_async( + kwargs={"tenant_id": tenant_id, "scan_id": scan_id} + ) @shared_task(base=RLSTask, name="provider-connection-check") @@ -281,6 +286,25 @@ def perform_scan_summary_task(tenant_id: str, scan_id: str): return aggregate_findings(tenant_id=tenant_id, scan_id=scan_id) +# TODO: This task must be queued at the `attack-paths` queue, don't forget to add it to the `docker-entrypoint.sh` file +@shared_task(base=RLSTask, bind=True, name="attack-paths-scan-perform", queue="scans") +def perform_attack_paths_scan_task(self, tenant_id: str, scan_id: str): + """ + Execute an Attack Paths scan for the given provider within the current tenant RLS context. + + Args: + self: The task instance (automatically passed when bind=True). + tenant_id (str): The tenant identifier for RLS context. + scan_id (str): The Prowler scan identifier for obtaining the tenant and provider context. + + Returns: + Any: The result from `attack_paths_scan`, including any per-scan failure details. + """ + return attack_paths_scan( + tenant_id=tenant_id, scan_id=scan_id, task_id=self.request.id + ) + + @shared_task(name="tenant-deletion", queue="deletion", autoretry_for=(Exception,)) def delete_tenant_task(tenant_id: str): return delete_tenant(pk=tenant_id) diff --git a/api/src/backend/tasks/tests/test_attack_paths_scan.py b/api/src/backend/tasks/tests/test_attack_paths_scan.py new file mode 100644 index 0000000000..334c02df11 --- /dev/null +++ b/api/src/backend/tasks/tests/test_attack_paths_scan.py @@ -0,0 +1,412 @@ +from contextlib import nullcontext +from types import SimpleNamespace +from unittest.mock import MagicMock, call, patch + +import pytest + +from api.models import ( + AttackPathsScan, + Finding, + Provider, + Resource, + ResourceFindingMapping, + Scan, + StateChoices, + StatusChoices, +) +from prowler.lib.check.models import Severity +from tasks.jobs.attack_paths import prowler as prowler_module +from tasks.jobs.attack_paths.scan import run as attack_paths_run + + +@pytest.mark.django_db +class TestAttackPathsRun: + def test_run_success_flow(self, tenants_fixture, providers_fixture, scans_fixture): + tenant = tenants_fixture[0] + provider = providers_fixture[0] + provider.provider = Provider.ProviderChoices.AWS + provider.save() + scan = scans_fixture[0] + scan.provider = provider + scan.save() + + attack_paths_scan = AttackPathsScan.objects.create( + tenant_id=tenant.id, + provider=provider, + scan=scan, + state=StateChoices.SCHEDULED, + ) + + mock_session = MagicMock() + session_ctx = MagicMock() + session_ctx.__enter__.return_value = mock_session + session_ctx.__exit__.return_value = False + ingestion_result = {"organizations": "warning"} + ingestion_fn = MagicMock(return_value=ingestion_result) + + with ( + patch( + "tasks.jobs.attack_paths.scan.rls_transaction", + new=lambda *args, **kwargs: nullcontext(), + ), + patch( + "tasks.jobs.attack_paths.scan.initialize_prowler_provider", + return_value=MagicMock(_enabled_regions=["us-east-1"]), + ), + patch( + "tasks.jobs.attack_paths.scan.graph_database.get_uri", + return_value="bolt://neo4j", + ), + patch( + "tasks.jobs.attack_paths.scan.graph_database.get_database_name", + return_value="db-scan-id", + ) as mock_get_db_name, + patch( + "tasks.jobs.attack_paths.scan.graph_database.create_database" + ) as mock_create_db, + patch( + "tasks.jobs.attack_paths.scan.graph_database.get_session", + return_value=session_ctx, + ) as mock_get_session, + patch( + "tasks.jobs.attack_paths.scan.cartography_create_indexes.run" + ) as mock_cartography_indexes, + patch( + "tasks.jobs.attack_paths.scan.cartography_analysis.run" + ) as mock_cartography_analysis, + patch( + "tasks.jobs.attack_paths.scan.prowler.create_indexes" + ) as mock_prowler_indexes, + patch( + "tasks.jobs.attack_paths.scan.prowler.analysis" + ) as mock_prowler_analysis, + patch( + "tasks.jobs.attack_paths.scan.db_utils.retrieve_attack_paths_scan", + return_value=attack_paths_scan, + ) as mock_retrieve_scan, + patch( + "tasks.jobs.attack_paths.scan.db_utils.starting_attack_paths_scan" + ) as mock_starting, + patch( + "tasks.jobs.attack_paths.scan.db_utils.update_attack_paths_scan_progress" + ) as mock_update_progress, + patch( + "tasks.jobs.attack_paths.scan.db_utils.finish_attack_paths_scan" + ) as mock_finish, + patch( + "tasks.jobs.attack_paths.scan.get_cartography_ingestion_function", + return_value=ingestion_fn, + ) as mock_get_ingestion, + patch( + "tasks.jobs.attack_paths.scan._call_within_event_loop", + side_effect=lambda fn, *a, **kw: fn(*a, **kw), + ) as mock_event_loop, + ): + result = attack_paths_run(str(tenant.id), str(scan.id), "task-123") + + assert result == ingestion_result + mock_retrieve_scan.assert_called_once_with(str(tenant.id), str(scan.id)) + mock_starting.assert_called_once() + config = mock_starting.call_args[0][2] + assert config.neo4j_database == "db-scan-id" + + mock_create_db.assert_called_once_with("db-scan-id") + mock_get_session.assert_called_once_with("db-scan-id") + mock_cartography_indexes.assert_called_once_with(mock_session, config) + mock_prowler_indexes.assert_called_once_with(mock_session) + mock_cartography_analysis.assert_called_once_with(mock_session, config) + mock_prowler_analysis.assert_called_once_with( + mock_session, + provider, + str(scan.id), + config, + ) + assert mock_get_ingestion.call_args_list == [ + call(provider.provider), + call(provider.provider), + ] + mock_event_loop.assert_called_once() + mock_update_progress.assert_any_call(attack_paths_scan, 1) + mock_update_progress.assert_any_call(attack_paths_scan, 2) + mock_update_progress.assert_any_call(attack_paths_scan, 95) + mock_finish.assert_called_once_with( + attack_paths_scan, StateChoices.COMPLETED, ingestion_result + ) + mock_get_db_name.assert_called_once_with(attack_paths_scan.id) + + def test_run_failure_marks_scan_failed( + self, tenants_fixture, providers_fixture, scans_fixture + ): + tenant = tenants_fixture[0] + provider = providers_fixture[0] + provider.provider = Provider.ProviderChoices.AWS + provider.save() + scan = scans_fixture[0] + scan.provider = provider + scan.save() + + attack_paths_scan = AttackPathsScan.objects.create( + tenant_id=tenant.id, + provider=provider, + scan=scan, + state=StateChoices.SCHEDULED, + ) + + mock_session = MagicMock() + session_ctx = MagicMock() + session_ctx.__enter__.return_value = mock_session + session_ctx.__exit__.return_value = False + ingestion_fn = MagicMock(side_effect=RuntimeError("ingestion boom")) + + with ( + patch( + "tasks.jobs.attack_paths.scan.rls_transaction", + new=lambda *args, **kwargs: nullcontext(), + ), + patch( + "tasks.jobs.attack_paths.scan.initialize_prowler_provider", + return_value=MagicMock(_enabled_regions=["us-east-1"]), + ), + patch("tasks.jobs.attack_paths.scan.graph_database.get_uri"), + patch( + "tasks.jobs.attack_paths.scan.graph_database.get_database_name", + return_value="db-scan-id", + ), + patch("tasks.jobs.attack_paths.scan.graph_database.create_database"), + patch( + "tasks.jobs.attack_paths.scan.graph_database.get_session", + return_value=session_ctx, + ), + patch("tasks.jobs.attack_paths.scan.cartography_create_indexes.run"), + patch("tasks.jobs.attack_paths.scan.cartography_analysis.run"), + patch("tasks.jobs.attack_paths.scan.prowler.create_indexes"), + patch("tasks.jobs.attack_paths.scan.prowler.analysis"), + patch( + "tasks.jobs.attack_paths.scan.db_utils.retrieve_attack_paths_scan", + return_value=attack_paths_scan, + ), + patch("tasks.jobs.attack_paths.scan.db_utils.starting_attack_paths_scan"), + patch( + "tasks.jobs.attack_paths.scan.db_utils.update_attack_paths_scan_progress" + ), + patch( + "tasks.jobs.attack_paths.scan.db_utils.finish_attack_paths_scan" + ) as mock_finish, + patch( + "tasks.jobs.attack_paths.scan.get_cartography_ingestion_function", + return_value=ingestion_fn, + ), + patch( + "tasks.jobs.attack_paths.scan._call_within_event_loop", + side_effect=lambda fn, *a, **kw: fn(*a, **kw), + ), + patch( + "tasks.jobs.attack_paths.scan.utils.stringify_exception", + return_value="Cartography failed: ingestion boom", + ), + ): + with pytest.raises(RuntimeError, match="ingestion boom"): + attack_paths_run(str(tenant.id), str(scan.id), "task-456") + + failure_args = mock_finish.call_args[0] + assert failure_args[0] is attack_paths_scan + assert failure_args[1] == StateChoices.FAILED + assert failure_args[2] == { + "global_cartography_error": "Cartography failed: ingestion boom" + } + + def test_run_returns_early_for_unsupported_provider(self, tenants_fixture): + tenant = tenants_fixture[0] + provider = Provider.objects.create( + provider=Provider.ProviderChoices.GCP, + uid="gcp-account", + alias="gcp", + tenant_id=tenant.id, + ) + scan = Scan.objects.create( + name="GCP Scan", + provider=provider, + trigger=Scan.TriggerChoices.MANUAL, + state=StateChoices.AVAILABLE, + tenant_id=tenant.id, + ) + + with ( + patch( + "tasks.jobs.attack_paths.scan.rls_transaction", + new=lambda *args, **kwargs: nullcontext(), + ), + patch( + "tasks.jobs.attack_paths.scan.initialize_prowler_provider", + return_value=MagicMock(), + ), + patch( + "tasks.jobs.attack_paths.scan.get_cartography_ingestion_function", + return_value=None, + ) as mock_get_ingestion, + patch( + "tasks.jobs.attack_paths.scan.db_utils.retrieve_attack_paths_scan" + ) as mock_retrieve, + ): + result = attack_paths_run(str(tenant.id), str(scan.id), "task-789") + + assert result == {} + mock_get_ingestion.assert_called_once_with(provider.provider) + mock_retrieve.assert_not_called() + + +@pytest.mark.django_db +class TestAttackPathsProwlerHelpers: + def test_create_indexes_executes_all_statements(self): + mock_session = MagicMock() + with patch("tasks.jobs.attack_paths.prowler.run_write_query") as mock_run_write: + prowler_module.create_indexes(mock_session) + + assert mock_run_write.call_count == len(prowler_module.INDEX_STATEMENTS) + mock_run_write.assert_has_calls( + [call(mock_session, stmt) for stmt in prowler_module.INDEX_STATEMENTS] + ) + + def test_load_findings_batches_requests(self, providers_fixture): + provider = providers_fixture[0] + provider.provider = Provider.ProviderChoices.AWS + provider.save() + + findings = [ + {"id": "1", "resource_uid": "r-1"}, + {"id": "2", "resource_uid": "r-2"}, + ] + config = SimpleNamespace(update_tag=12345) + mock_session = MagicMock() + + with ( + patch.object(prowler_module, "BATCH_SIZE", 1), + patch( + "tasks.jobs.attack_paths.prowler.get_root_node_label", + return_value="AWSAccount", + ), + patch( + "tasks.jobs.attack_paths.prowler.get_node_uid_field", + return_value="arn", + ), + ): + prowler_module.load_findings(mock_session, findings, provider, config) + + assert mock_session.run.call_count == 2 + for call_args in mock_session.run.call_args_list: + params = call_args.args[1] + assert params["provider_uid"] == str(provider.uid) + assert params["last_updated"] == config.update_tag + assert "findings_data" in params + + def test_cleanup_findings_runs_batches(self, providers_fixture): + provider = providers_fixture[0] + config = SimpleNamespace(update_tag=1024) + mock_session = MagicMock() + + first_batch = MagicMock() + first_batch.single.return_value = {"deleted_findings_count": 3} + second_batch = MagicMock() + second_batch.single.return_value = {"deleted_findings_count": 0} + mock_session.run.side_effect = [first_batch, second_batch] + + prowler_module.cleanup_findings(mock_session, provider, config) + + assert mock_session.run.call_count == 2 + params = mock_session.run.call_args.args[1] + assert params["provider_uid"] == str(provider.uid) + assert params["last_updated"] == config.update_tag + + def test_get_provider_last_scan_findings_returns_latest_scan_data( + self, + tenants_fixture, + providers_fixture, + ): + tenant = tenants_fixture[0] + provider = providers_fixture[0] + provider.provider = Provider.ProviderChoices.AWS + provider.save() + + resource = Resource.objects.create( + tenant_id=tenant.id, + provider=provider, + uid="resource-uid", + name="Resource", + region="us-east-1", + service="ec2", + type="instance", + ) + + older_scan = Scan.objects.create( + name="Older", + provider=provider, + trigger=Scan.TriggerChoices.MANUAL, + state=StateChoices.COMPLETED, + tenant_id=tenant.id, + ) + old_finding = Finding.objects.create( + tenant_id=tenant.id, + uid="older-finding", + scan=older_scan, + delta=Finding.DeltaChoices.NEW, + status=StatusChoices.PASS, + status_extended="ok", + severity=Severity.low, + impact=Severity.low, + impact_extended="", + raw_result={}, + check_id="check-old", + check_metadata={"checktitle": "Old"}, + first_seen_at=older_scan.inserted_at, + ) + ResourceFindingMapping.objects.create( + tenant_id=tenant.id, + resource=resource, + finding=old_finding, + ) + + latest_scan = Scan.objects.create( + name="Latest", + provider=provider, + trigger=Scan.TriggerChoices.MANUAL, + state=StateChoices.COMPLETED, + tenant_id=tenant.id, + ) + finding = Finding.objects.create( + tenant_id=tenant.id, + uid="finding-uid", + scan=latest_scan, + delta=Finding.DeltaChoices.NEW, + status=StatusChoices.FAIL, + status_extended="failed", + severity=Severity.high, + impact=Severity.high, + impact_extended="", + raw_result={}, + check_id="check-1", + check_metadata={"checktitle": "Check title"}, + first_seen_at=latest_scan.inserted_at, + ) + ResourceFindingMapping.objects.create( + tenant_id=tenant.id, + resource=resource, + finding=finding, + ) + + latest_scan.refresh_from_db() + + with patch( + "tasks.jobs.attack_paths.prowler.rls_transaction", + new=lambda *args, **kwargs: nullcontext(), + ): + findings_data = prowler_module.get_provider_last_scan_findings( + provider, + str(latest_scan.id), + ) + + assert len(findings_data) == 1 + finding_dict = findings_data[0] + assert finding_dict["id"] == str(finding.id) + assert finding_dict["resource_uid"] == resource.uid + assert finding_dict["check_title"] == "Check title" + assert finding_dict["scan_id"] == str(latest_scan.id) diff --git a/api/src/backend/tasks/tests/test_deletion.py b/api/src/backend/tasks/tests/test_deletion.py index 81cdb44daa..fc90bee0e3 100644 --- a/api/src/backend/tasks/tests/test_deletion.py +++ b/api/src/backend/tasks/tests/test_deletion.py @@ -1,27 +1,60 @@ +from unittest.mock import call, patch + import pytest + from django.core.exceptions import ObjectDoesNotExist -from tasks.jobs.deletion import delete_provider, delete_tenant from api.models import Provider, Tenant +from tasks.jobs.deletion import delete_provider, delete_tenant @pytest.mark.django_db class TestDeleteProvider: def test_delete_provider_success(self, providers_fixture): - instance = providers_fixture[0] - tenant_id = str(instance.tenant_id) - result = delete_provider(tenant_id, instance.id) + with patch( + "tasks.jobs.deletion.get_provider_graph_database_names" + ) as mock_get_provider_graph_database_names, patch( + "tasks.jobs.deletion.graph_database.drop_database" + ) as mock_drop_database: + graph_db_names = ["graph-db-1", "graph-db-2"] + mock_get_provider_graph_database_names.return_value = graph_db_names - assert result - with pytest.raises(ObjectDoesNotExist): - Provider.objects.get(pk=instance.id) + instance = providers_fixture[0] + tenant_id = str(instance.tenant_id) + result = delete_provider(tenant_id, instance.id) + + assert result + with pytest.raises(ObjectDoesNotExist): + Provider.objects.get(pk=instance.id) + + mock_get_provider_graph_database_names.assert_called_once_with( + tenant_id, instance.id + ) + mock_drop_database.assert_has_calls( + [call(graph_db_name) for graph_db_name in graph_db_names] + ) def test_delete_provider_does_not_exist(self, tenants_fixture): - tenant_id = str(tenants_fixture[0].id) - non_existent_pk = "babf6796-cfcc-4fd3-9dcf-88d012247645" + with patch( + "tasks.jobs.deletion.get_provider_graph_database_names" + ) as mock_get_provider_graph_database_names, patch( + "tasks.jobs.deletion.graph_database.drop_database" + ) as mock_drop_database: + graph_db_names = ["graph-db-1"] + mock_get_provider_graph_database_names.return_value = graph_db_names - with pytest.raises(ObjectDoesNotExist): - delete_provider(tenant_id, non_existent_pk) + tenant_id = str(tenants_fixture[0].id) + non_existent_pk = "babf6796-cfcc-4fd3-9dcf-88d012247645" + + with pytest.raises(ObjectDoesNotExist): + delete_provider(tenant_id, non_existent_pk) + + mock_get_provider_graph_database_names.assert_called_once_with( + tenant_id, non_existent_pk + ) + mock_drop_database.assert_has_calls( + [call(graph_db_name) for graph_db_name in graph_db_names] + ) @pytest.mark.django_db @@ -30,33 +63,68 @@ class TestDeleteTenant: """ Test successful deletion of a tenant and its related data. """ - tenant = tenants_fixture[0] - providers = Provider.objects.filter(tenant_id=tenant.id) + with patch( + "tasks.jobs.deletion.get_provider_graph_database_names" + ) as mock_get_provider_graph_database_names, patch( + "tasks.jobs.deletion.graph_database.drop_database" + ) as mock_drop_database: + tenant = tenants_fixture[0] + providers = list(Provider.objects.filter(tenant_id=tenant.id)) - # Ensure the tenant and related providers exist before deletion - assert Tenant.objects.filter(id=tenant.id).exists() - assert providers.exists() + graph_db_names_per_provider = [ + [f"graph-db-{provider.id}"] for provider in providers + ] + mock_get_provider_graph_database_names.side_effect = ( + graph_db_names_per_provider + ) - # Call the function and validate the result - deletion_summary = delete_tenant(tenant.id) + # Ensure the tenant and related providers exist before deletion + assert Tenant.objects.filter(id=tenant.id).exists() + assert providers - assert deletion_summary is not None - assert not Tenant.objects.filter(id=tenant.id).exists() - assert not Provider.objects.filter(tenant_id=tenant.id).exists() + # Call the function and validate the result + deletion_summary = delete_tenant(tenant.id) + + assert deletion_summary is not None + assert not Tenant.objects.filter(id=tenant.id).exists() + assert not Provider.objects.filter(tenant_id=tenant.id).exists() + + expected_calls = [ + call(provider.tenant_id, provider.id) for provider in providers + ] + mock_get_provider_graph_database_names.assert_has_calls( + expected_calls, any_order=True + ) + assert mock_get_provider_graph_database_names.call_count == len( + expected_calls + ) + expected_drop_calls = [ + call(graph_db_name[0]) for graph_db_name in graph_db_names_per_provider + ] + mock_drop_database.assert_has_calls(expected_drop_calls, any_order=True) + assert mock_drop_database.call_count == len(expected_drop_calls) def test_delete_tenant_with_no_providers(self, tenants_fixture): """ Test deletion of a tenant with no related providers. """ - tenant = tenants_fixture[1] # Assume this tenant has no providers - providers = Provider.objects.filter(tenant_id=tenant.id) + with patch( + "tasks.jobs.deletion.get_provider_graph_database_names" + ) as mock_get_provider_graph_database_names, patch( + "tasks.jobs.deletion.graph_database.drop_database" + ) as mock_drop_database: + tenant = tenants_fixture[1] # Assume this tenant has no providers + providers = Provider.objects.filter(tenant_id=tenant.id) - # Ensure the tenant exists but has no related providers - assert Tenant.objects.filter(id=tenant.id).exists() - assert not providers.exists() + # Ensure the tenant exists but has no related providers + assert Tenant.objects.filter(id=tenant.id).exists() + assert not providers.exists() - # Call the function and validate the result - deletion_summary = delete_tenant(tenant.id) + # Call the function and validate the result + deletion_summary = delete_tenant(tenant.id) - assert deletion_summary == {} # No providers, so empty summary - assert not Tenant.objects.filter(id=tenant.id).exists() + assert deletion_summary == {} # No providers, so empty summary + assert not Tenant.objects.filter(id=tenant.id).exists() + + mock_get_provider_graph_database_names.assert_not_called() + mock_drop_database.assert_not_called() diff --git a/api/src/backend/tasks/tests/test_tasks.py b/api/src/backend/tasks/tests/test_tasks.py index 938290d947..98926937fe 100644 --- a/api/src/backend/tasks/tests/test_tasks.py +++ b/api/src/backend/tasks/tests/test_tasks.py @@ -1,24 +1,28 @@ import uuid + +from contextlib import contextmanager from unittest.mock import MagicMock, patch import openai import pytest + from botocore.exceptions import ClientError -from tasks.tasks import ( - _perform_scan_complete_tasks, - check_integrations_task, - check_lighthouse_provider_connection_task, - generate_outputs_task, - refresh_lighthouse_provider_models_task, - s3_integration_task, - security_hub_integration_task, -) from api.models import ( Integration, LighthouseProviderConfiguration, LighthouseProviderModels, ) +from tasks.tasks import ( + _perform_scan_complete_tasks, + check_integrations_task, + check_lighthouse_provider_connection_task, + generate_outputs_task, + perform_attack_paths_scan_task, + refresh_lighthouse_provider_models_task, + s3_integration_task, + security_hub_integration_task, +) # TODO Move this to outputs/reports jobs @@ -529,6 +533,7 @@ class TestGenerateOutputs: class TestScanCompleteTasks: + @patch("tasks.tasks.perform_attack_paths_scan_task.apply_async") @patch("tasks.tasks.create_compliance_requirements_task.apply_async") @patch("tasks.tasks.perform_scan_summary_task.si") @patch("tasks.tasks.generate_outputs_task.si") @@ -541,6 +546,7 @@ class TestScanCompleteTasks: mock_outputs_task, mock_scan_summary_task, mock_compliance_requirements_task, + mock_attack_paths_task, ): """Test that scan complete tasks are properly orchestrated with optimized reports.""" _perform_scan_complete_tasks("tenant-id", "scan-id", "provider-id") @@ -577,6 +583,68 @@ class TestScanCompleteTasks: scan_id="scan-id", ) + mock_attack_paths_task.assert_called_once_with( + kwargs={"tenant_id": "tenant-id", "scan_id": "scan-id"} + ) + + +class TestAttackPathsTasks: + @staticmethod + @contextmanager + def _override_task_request(task, **attrs): + request = task.request + sentinel = object() + previous = {key: getattr(request, key, sentinel) for key in attrs} + for key, value in attrs.items(): + setattr(request, key, value) + + try: + yield + finally: + for key, prev in previous.items(): + if prev is sentinel: + if hasattr(request, key): + delattr(request, key) + else: + setattr(request, key, prev) + + def test_perform_attack_paths_scan_task_calls_runner(self): + with ( + patch("tasks.tasks.attack_paths_scan") as mock_attack_paths_scan, + self._override_task_request( + perform_attack_paths_scan_task, id="celery-task-id" + ), + ): + mock_attack_paths_scan.return_value = {"status": "ok"} + + result = perform_attack_paths_scan_task.run( + tenant_id="tenant-id", scan_id="scan-id" + ) + + mock_attack_paths_scan.assert_called_once_with( + tenant_id="tenant-id", scan_id="scan-id", task_id="celery-task-id" + ) + assert result == {"status": "ok"} + + def test_perform_attack_paths_scan_task_propagates_exception(self): + with ( + patch( + "tasks.tasks.attack_paths_scan", + side_effect=RuntimeError("Exception to propagate"), + ) as mock_attack_paths_scan, + self._override_task_request( + perform_attack_paths_scan_task, id="celery-task-error" + ), + ): + with pytest.raises(RuntimeError, match="Exception to propagate"): + perform_attack_paths_scan_task.run( + tenant_id="tenant-id", scan_id="scan-id" + ) + + mock_attack_paths_scan.assert_called_once_with( + tenant_id="tenant-id", scan_id="scan-id", task_id="celery-task-error" + ) + @pytest.mark.django_db class TestCheckIntegrationsTask: diff --git a/docker-compose-dev.yml b/docker-compose-dev.yml index 05ed89c397..bc16d52eb1 100644 --- a/docker-compose-dev.yml +++ b/docker-compose-dev.yml @@ -1,6 +1,7 @@ services: api-dev: hostname: "prowler-api" + image: prowler-api-dev build: context: ./api dockerfile: Dockerfile @@ -24,6 +25,8 @@ services: condition: service_healthy valkey: condition: service_healthy + neo4j: + condition: service_healthy entrypoint: - "/home/prowler/docker-entrypoint.sh" - "dev" @@ -78,7 +81,40 @@ services: timeout: 5s retries: 3 + neo4j: + image: graphstack/dozerdb:5.26.3.0 + hostname: "neo4j" + volumes: + - ./_data/neo4j:/data + environment: + # We can't add our .env file because some of our current variables are not compatible with Neo4j env vars + # Auth + - NEO4J_AUTH=${NEO4J_USER}/${NEO4J_PASSWORD} + # Memory limits + - NEO4J_server_memory_pagecache_size=${NEO4J_SERVER_MEMORY_PAGECACHE_SIZE} + - NEO4J_server_memory_heap_initial__size=${NEO4J_SERVER_MEMORY_HEAP_INITIAL__SIZE} + - NEO4J_server_memory_heap_max__size=${NEO4J_SERVER_MEMORY_HEAP_MAX__SIZE} + # APOC + - apoc.export.file.enabled=${NEO4J_POC_EXPORT_FILE_ENABLED} + - apoc.import.file.enabled=${NEO4J_APOC_IMPORT_FILE_ENABLED} + - apoc.import.file.use_neo4j_config=${NEO4J_APOC_IMPORT_FILE_USE_NEO4J_CONFIG} + - NEO4J_PLUGINS=${NEO4J_PLUGINS} + - NEO4J_dbms_security_procedures_allowlist=${NEO4J_DBMS_SECURITY_PROCEDURES_ALLOWLIST} + - NEO4J_dbms_security_procedures_unrestricted=${NEO4J_DBMS_SECURITY_PROCEDURES_UNRESTRICTED} + # Networking + - dbms.connector.bolt.listen_address=${NEO4J_DBMS_CONNECTOR_BOLT_LISTEN_ADDRESS} + # 7474 is the UI port + ports: + - 7474:7474 + - ${NEO4J_PORT:-7687}:7687 + healthcheck: + test: ["CMD", "wget", "--no-verbose", "http://localhost:7474"] + interval: 10s + timeout: 10s + retries: 10 + worker-dev: + image: prowler-api-dev build: context: ./api dockerfile: Dockerfile @@ -89,17 +125,23 @@ services: - path: .env required: false volumes: - - "outputs:/tmp/prowler_api_output" + - ./api/src/backend:/home/prowler/backend + - ./api/pyproject.toml:/home/prowler/pyproject.toml + - ./api/docker-entrypoint.sh:/home/prowler/docker-entrypoint.sh + - outputs:/tmp/prowler_api_output depends_on: valkey: condition: service_healthy postgres: condition: service_healthy + neo4j: + condition: service_healthy entrypoint: - "/home/prowler/docker-entrypoint.sh" - "worker" worker-beat: + image: prowler-api-dev build: context: ./api dockerfile: Dockerfile @@ -114,6 +156,8 @@ services: condition: service_healthy postgres: condition: service_healthy + neo4j: + condition: service_healthy entrypoint: - "../docker-entrypoint.sh" - "beat" diff --git a/docker-compose.yml b/docker-compose.yml index 2e2469e0c8..5a94d5faa0 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -63,6 +63,36 @@ services: timeout: 5s retries: 3 + neo4j: + image: graphstack/dozerdb:5.26.3.0 + hostname: "neo4j" + volumes: + - ./_data/neo4j:/data + environment: + # We can't add our .env file because some of our current variables are not compatible with Neo4j env vars + # Auth + - NEO4J_AUTH=${NEO4J_USER}/${NEO4J_PASSWORD} + # Memory limits + - NEO4J_server_memory_pagecache_size=${NEO4J_SERVER_MEMORY_PAGECACHE_SIZE} + - NEO4J_server_memory_heap_initial__size=${NEO4J_SERVER_MEMORY_HEAP_INITIAL__SIZE} + - NEO4J_server_memory_heap_max__size=${NEO4J_SERVER_MEMORY_HEAP_MAX__SIZE} + # APOC + - apoc.export.file.enabled=${NEO4J_POC_EXPORT_FILE_ENABLED} + - apoc.import.file.enabled=${NEO4J_APOC_IMPORT_FILE_ENABLED} + - apoc.import.file.use_neo4j_config=${NEO4J_APOC_IMPORT_FILE_USE_NEO4J_CONFIG} + - NEO4J_PLUGINS=${NEO4J_PLUGINS} + - NEO4J_dbms_security_procedures_allowlist=${NEO4J_DBMS_SECURITY_PROCEDURES_ALLOWLIST} + - NEO4J_dbms_security_procedures_unrestricted=${NEO4J_DBMS_SECURITY_PROCEDURES_UNRESTRICTED} + # Networking + - dbms.connector.bolt.listen_address=${NEO4J_DBMS_CONNECTOR_BOLT_LISTEN_ADDRESS} + ports: + - ${NEO4J_PORT:-7687}:7687 + healthcheck: + test: ["CMD", "wget", "--no-verbose", "http://localhost:7474"] + interval: 10s + timeout: 10s + retries: 10 + worker: image: prowlercloud/prowler-api:${PROWLER_API_VERSION:-stable} env_file: