chore(release): Bump versions to v5.47.1 (#13204 )

Co-authored-by: prowler-bot <179230569+prowler-bot@users.noreply.github.com>
chore(api): Update prowler dependency to v5.47 for release 5.47.0 (#13202 )
2026-11-02 09:24:27 +00:00 · 2026-10-29 16:37:50 +01:00 · 2026-10-29 14:11:31 +01:00 · 2026-10-29 14:01:05 +01:00 · 2026-10-29 13:54:06 +01:00 · 2026-10-29 12:56:46 +01:00
2071 changed files with 151476 additions and 25926 deletions
@@ -1,6 +1,6 @@
 {
  "name": "prowler-plugins",
-  "description": "Prowler Cloud Security for Claude Code",
+  "description": "Prowler Cloud security and compliance for Claude",
  "owner": {
    "name": "Prowler",
    "email": "support@prowler.com"
@@ -9,7 +9,7 @@
    {
      "name": "prowler",
      "source": "./claude_plugins/prowler",
-      "description": "Prowler for Claude Code — cloud security and compliance skills powered by the Prowler MCP server. Bundles compliance triage and remediation; more skills coming.",
+      "description": "Connect Claude to Prowler Cloud through the Prowler MCP server. Reads providers, scans, findings, compliance results and resources for AWS, Azure, GCP, Kubernetes, Microsoft 365, GitHub and the other providers listed at https://registry.prowler.com, and returns remediation guidance for failed checks. Writes are limited to Prowler Cloud itself: triggering scans, finding triage, mute rules and the mutelist, providers and integrations. Your cloud resources are not changed by the connector.",
      "category": "security",
      "homepage": "https://prowler.com"
    }
@@ -110,11 +110,27 @@ DJANGO_OUTPUT_S3_AWS_SECRET_ACCESS_KEY=""
 DJANGO_OUTPUT_S3_AWS_SESSION_TOKEN=""

 # The AWS region where your S3 bucket is located (e.g., "us-east-1")
+# Required if the bucket uses SSE-KMS: download URLs are then signed with SigV4, which
+# is scoped to this region, so it must match the bucket's
 DJANGO_OUTPUT_S3_AWS_DEFAULT_REGION=""

 # The name of the S3 bucket where scan output should be stored
 DJANGO_OUTPUT_S3_AWS_OUTPUT_BUCKET=""

+# The storage endpoint the API and Celery workers use to upload and list scan output
+# (e.g. "http://minio:9000"). Leave empty on AWS S3. Set it when scan output is stored on
+# S3-compatible object storage such as MinIO instead of real S3.
+# If set without DJANGO_OUTPUT_S3_AWS_PUBLIC_ENDPOINT_URL below, report download URLs are
+# signed against this internal host, and a browser outside the container network cannot open them.
+DJANGO_OUTPUT_S3_AWS_ENDPOINT_URL=""
+
+# The storage address the browser can reach, used only to sign report download URLs
+# (e.g. "https://storage.example.com"). Leave empty on AWS S3. Set it when storage is
+# only reachable inside the container network, such as MinIO on "http://minio:9000".
+# The reverse proxy in front of it must forward the Host header unchanged: SigV4 signs
+# Host, so rewriting it to the internal name invalidates the signature.
+DJANGO_OUTPUT_S3_AWS_PUBLIC_ENDPOINT_URL=""
+
 # Django settings
 DJANGO_ALLOWED_HOSTS=localhost,127.0.0.1,prowler-api
 DJANGO_BIND_ADDRESS=0.0.0.0
@@ -158,7 +174,7 @@ SENTRY_RELEASE=local
 # REO_DEV_CLIENT_ID=

 #### Prowler release version ####
-NEXT_PUBLIC_PROWLER_RELEASE_VERSION=v5.39.0
+NEXT_PUBLIC_PROWLER_RELEASE_VERSION=v5.47.1

 # Social login credentials
 SOCIAL_GOOGLE_OAUTH_CALLBACK_URL="${AUTH_URL}/api/auth/callback/google"
@@ -1,28 +1,28 @@
 # SDK
-/* @prowler-cloud/detection-remediation
-/prowler/ @prowler-cloud/detection-remediation
-/tests/ @prowler-cloud/detection-remediation
-/dashboard/ @prowler-cloud/detection-remediation
-/docs/ @prowler-cloud/detection-remediation
-/examples/ @prowler-cloud/detection-remediation
-/util/ @prowler-cloud/detection-remediation
-/contrib/ @prowler-cloud/detection-remediation
-/permissions/ @prowler-cloud/detection-remediation
-/codecov.yml @prowler-cloud/detection-remediation @prowler-cloud/api
+/* @prowler-cloud/engineering
+/prowler/ @prowler-cloud/engineering
+/tests/ @prowler-cloud/engineering
+/dashboard/ @prowler-cloud/engineering
+/docs/ @prowler-cloud/engineering
+/examples/ @prowler-cloud/engineering
+/util/ @prowler-cloud/engineering
+/contrib/ @prowler-cloud/engineering
+/permissions/ @prowler-cloud/engineering
+/codecov.yml @prowler-cloud/engineering

 # API
-/api/ @prowler-cloud/api
+/api/ @prowler-cloud/engineering

 # UI
-/ui/ @prowler-cloud/ui
+/ui/ @prowler-cloud/engineering

 # AI
-/mcp_server/ @prowler-cloud/detection-remediation
+/mcp_server/ @prowler-cloud/engineering

-# Platform
-/.github/ @prowler-cloud/platform
-/Makefile @prowler-cloud/platform
-/kubernetes/ @prowler-cloud/platform
-**/Dockerfile* @prowler-cloud/platform
-**/docker-compose*.yml @prowler-cloud/platform
-**/docker-compose*.yaml @prowler-cloud/platform
+# CI and infrastructure
+/.github/ @prowler-cloud/engineering
+/Makefile @prowler-cloud/engineering
+/kubernetes/ @prowler-cloud/engineering
+**/Dockerfile* @prowler-cloud/engineering
+**/docker-compose*.yml @prowler-cloud/engineering
+**/docker-compose*.yaml @prowler-cloud/engineering
@@ -46,6 +46,17 @@ runs:
      env:
        GITHUB_TOKEN: ${{ github.token }}
      run: |
+        if grep -q "prowler-cloud/prowler" uv.lock; then
+          :
+        else
+          status=$?
+          if [ "$status" -ne 1 ]; then
+            echo "::error::grep failed reading uv.lock (exit code $status)."
+            exit "$status"
+          fi
+          echo "No prowler-cloud/prowler entry in uv.lock, nothing to update."
+          exit 0
+        fi
        LATEST_COMMIT=$(curl -sf --retry 3 --retry-all-errors --retry-delay 2 --retry-max-time 60 \
          -H "Authorization: Bearer ${GITHUB_TOKEN}" \
          -H "Accept: application/vnd.github+json" \
@@ -66,6 +77,17 @@ runs:
      env:
        GITHUB_TOKEN: ${{ github.token }}
      run: |
+        if grep -q "prowler-cloud/prowler" uv.lock; then
+          :
+        else
+          status=$?
+          if [ "$status" -ne 1 ]; then
+            echo "::error::grep failed reading uv.lock (exit code $status)."
+            exit "$status"
+          fi
+          echo "No prowler-cloud/prowler entry in uv.lock, nothing to update."
+          exit 0
+        fi
        LATEST_COMMIT=$(curl -sf --retry 3 --retry-all-errors --retry-delay 2 --retry-max-time 60 \
          -H "Authorization: Bearer ${GITHUB_TOKEN}" \
          -H "Accept: application/vnd.github+json" \
@@ -64,7 +64,7 @@ runs:
        scanners: 'vuln'
        ignore-unfixed: 'true' # A finding with no available fix is not actionable, so it must not gate
        timeout: '5m'
-        version: 'v0.72.0'
+        version: 'v0.74.0'
      # Not trivyignores: that input drops the .yaml extension Trivy parses by.
      env:
        TRIVY_IGNOREFILE: '.trivyignore.yaml'
@@ -81,7 +81,7 @@ runs:
        scanners: 'vuln'
        ignore-unfixed: 'true' # A finding with no available fix is not actionable, so it must not gate
        timeout: '5m'
-        version: 'v0.72.0'
+        version: 'v0.74.0'
      # Not trivyignores: that input drops the .yaml extension Trivy parses by.
      env:
        TRIVY_IGNOREFILE: '.trivyignore.yaml'
@@ -199,7 +199,7 @@ You MUST structure your response using this EXACT format. Do NOT include anythin

 ### For Check Logic Bug

-```
+```markdown
 ### AI Assessment [Experimental]: Check Logic Bug

 **Component**: {component from issue template}
@@ -297,7 +297,7 @@ Write tests FIRST (TDD). The skills contain all testing conventions and patterns

 ### For Bug (non-check)

-```
+```markdown
 ### AI Assessment [Experimental]: Bug

 **Component**: {CLI/SDK | API | UI | Dashboard | MCP Server | Other}
@@ -378,7 +378,7 @@ Write tests FIRST (TDD). The skills contain all testing conventions and patterns

 ### For Already Fixed

-```
+```markdown
 ### AI Assessment [Experimental]: Already Fixed

 **Component**: {component}
@@ -401,7 +401,7 @@ Upgrade to the latest version. Close the issue as resolved.

 ### For Feature Request

-```
+```markdown
 ### AI Assessment [Experimental]: Feature Request

 **Component**: {component}
@@ -419,7 +419,7 @@ Upgrade to the latest version. Close the issue as resolved.

 ### For Not a Bug

-```
+```markdown
 ### AI Assessment [Experimental]: Not a Bug

 **Component**: {component}
@@ -440,7 +440,7 @@ Upgrade to the latest version. Close the issue as resolved.

 ### For Needs More Information

-```
+```markdown
 ### AI Assessment [Experimental]: Needs More Information

 **Component**: {component or "Unknown"}
@@ -52,6 +52,16 @@ provider/alibabacloud:
      - any-glob-to-any-file: "prowler/providers/alibabacloud/**"
      - any-glob-to-any-file: "tests/providers/alibabacloud/**"

+provider/huaweicloud:
+  - changed-files:
+      - any-glob-to-any-file: "prowler/providers/huaweicloud/**"
+      - any-glob-to-any-file: "tests/providers/huaweicloud/**"
+
+provider/image:
+  - changed-files:
+      - any-glob-to-any-file: "prowler/providers/image/**"
+      - any-glob-to-any-file: "tests/providers/image/**"
+
 provider/cloudflare:
  - changed-files:
      - any-glob-to-any-file: "prowler/providers/cloudflare/**"
@@ -82,6 +92,11 @@ provider/linode:
      - any-glob-to-any-file: "prowler/providers/linode/**"
      - any-glob-to-any-file: "tests/providers/linode/**"

+provider/stackit:
+  - changed-files:
+      - any-glob-to-any-file: "prowler/providers/stackit/**"
+      - any-glob-to-any-file: "tests/providers/stackit/**"
+
 github_actions:
  - changed-files:
      - any-glob-to-any-file: ".github/workflows/*"
@@ -8,11 +8,11 @@ These JSON templates are used with the `slackapi/slack-github-action` using the

 ### Available Templates

-**Container Releases**
+#### Container Releases
 - `container-release-started.json`: Simple one-line notification when container push starts
 - `container-release-completed.json`: Simple one-line notification when container release completes

-**Deployments**
+#### Deployments
 - `deployment-started.json`: Deployment start notification with Block Kit formatting
 - `deployment-completed.json`: Deployment completion notification (updates the start message)

@@ -416,17 +416,17 @@ For deployments that start with one message and update it with the final status:
 ### Container Release (Simple One-Line)

 **Start message:**
-```
+```text
 API container release 4.5.0 push started... View run
 ```

 **Completion message (success):**
-```
+```text
 [✓] API container release 4.5.0 push completed successfully! View run
 ```

 **Completion message (failure):**
-```
+```text
 [✗] API container release 4.5.0 push failed View run
 ```

@@ -451,6 +451,17 @@ modules:
    e2e:
      - ui/tests/home/**

+  - name: ui-registry
+    match:
+      - ui/actions/registry/**
+      - ui/app/**/registry/**
+      - ui/components/registry/**
+      - ui/lib/registry/**
+      - ui/tests/registry/**
+    tests: []
+    e2e:
+      - ui/tests/registry/**
+
  - name: ui-shadcn
    match:
      - ui/components/shadcn/**
@@ -42,6 +42,7 @@ jobs:
            github.com:443
            pypi.org:443
            files.pythonhosted.org:443
+            release-assets.githubusercontent.com:443
            api.github.com:443
            raw.githubusercontent.com:443

@@ -81,6 +81,7 @@ jobs:
            auth.docker.io:443
            production.cloudflare.docker.com:443
            production.cloudfront.docker.com:443
+            tuf-repo-cdn.sigstore.dev:443
            raw.githubusercontent.com:443
            objects.githubusercontent.com:443
            grype.anchore.io:443
@@ -85,6 +85,7 @@ jobs:
            github.com:443
            pypi.org:443
            files.pythonhosted.org:443
+            release-assets.githubusercontent.com:443
            cli.codecov.io:443
            keybase.io:443
            raw.githubusercontent.com:443
@@ -39,7 +39,7 @@ jobs:

      - name: Check labels
        id: label_check
-        uses: agilepathway/label-checker@c3d16ad512e7cea5961df85ff2486bb774caf3c5 # v1.6.65
+        uses: agilepathway/label-checker@c324842522fbd012e4f590afe3b4e591301322ed # v1.6.66
        with:
          allow_failure: true
          prefix_mode: true
@@ -61,6 +61,7 @@ jobs:
            api.github.com:443
            github.com:443
            objects.githubusercontent.com:443
+            release-assets.githubusercontent.com:443
            pypi.org:443
            files.pythonhosted.org:443

@@ -38,13 +38,16 @@ jobs:
      - name: Checkout repository
        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
        with:
-          # PRs only need the diff range; push to master/release walks the new range from event.before.
-          # 50 is enough headroom for the longest realistic PR/push chain without paying for a full clone.
-          fetch-depth: 50
+          # Full history: a shallow clone cuts the PR's ancestry, git can no longer exclude commits
+          # already on the base branch, and TruffleHog scans the cut-off commit as a root (the whole
+          # tree as added). PRs that merge master in hit this and fail on pre-existing content.
+          fetch-depth: 0
          persist-credentials: false

      - name: Scan diff for secrets with TruffleHog
        # Action auto-injects --since-commit/--branch from event payload; passing them in extra_args produces duplicate flags.
-        uses: trufflesecurity/trufflehog@37b77001d0174ebec2fcca2bd83ff83a6d45a3ab # v3.95.3
+        uses: trufflesecurity/trufflehog@e76dff8a7e57d6df0c60ad8cbf5d5cba505a8fc9 # v3.99.2
        with:
+          # The action runs the `latest` image unless told otherwise: pin the scanner to the action's release.
+          version: 3.99.2@sha256:47a84bc18a0d04a165498bbbd3bacfd84176661cbc91f8e0f85467f6a771e99a
          extra_args: --results=verified,unknown
@@ -55,6 +55,10 @@ jobs:
        # Pin must match .pre-commit-config.yaml so prek and CI behave identically.
        # pnpm dlx doesn't accept --ignore-scripts as a flag; the env var
        # disables postinstall scripts on transitives the same way.
+        #
+        # Files come from `git ls-files` because markdownlint doesn't traverse
+        # dot-directories, so `.github/**/*.md` went unlinted.
+        # `.markdownlintignore` still applies to the listed paths.
        env:
          pnpm_config_ignore_scripts: 'true'
-        run: pnpm dlx markdownlint-cli@0.45.0 '**/*.md'
+        run: git ls-files -z '*.md' | xargs -0 -r pnpm dlx markdownlint-cli@0.45.0 --
@@ -114,6 +114,8 @@ jobs:
          egress-policy: block
          allowed-endpoints: >
            auth.docker.io:443
+            dl-cdn.alpinelinux.org:443
+            dualstack.j.sni.global.fastly.net:443
            files.pythonhosted.org:443
            ghcr.io:443
            github.com:443
@@ -77,10 +77,13 @@ jobs:
            auth.docker.io:443
            production.cloudflare.docker.com:443
            production.cloudfront.docker.com:443
+            tuf-repo-cdn.sigstore.dev:443
            ghcr.io:443
            pkg-containers.githubusercontent.com:443
            files.pythonhosted.org:443
            pypi.org:443
+            dl-cdn.alpinelinux.org:443
+            dualstack.j.sni.global.fastly.net:443
            api.github.com:443
            mirror.gcr.io:443
            check.trivy.dev:443
@@ -113,7 +113,7 @@ jobs:

      - name: Publish prowler-mcp package to PyPI
        if: steps.pypi-check.outputs.skip != 'true'
-        uses: pypa/gh-action-pypi-publish@cef221092ed1bacb1cc03d23a2d87d1d172e277b # v1.14.0
+        uses: pypa/gh-action-pypi-publish@dc37677b2e1c63e2034f94d8a5b11f265b73ba33 # v1.14.2
        with:
          packages-dir: ${{ env.WORKING_DIRECTORY }}/dist/
          print-hash: true
@@ -49,6 +49,7 @@ jobs:
            github.com:443
            pypi.org:443
            files.pythonhosted.org:443
+            release-assets.githubusercontent.com:443
            cli.codecov.io:443
            keybase.io:443
            ingest.codecov.io:443
@@ -35,6 +35,7 @@ jobs:
            api.github.com:443
            github.com:443
            objects.githubusercontent.com:443
+            release-assets.githubusercontent.com:443
            pypi.org:443
            files.pythonhosted.org:443

@@ -84,6 +84,8 @@ jobs:
        fi

    - name: Checkout release branch
+      env:
+        GH_TOKEN: ${{ secrets.PROWLER_BOT_ACCESS_TOKEN }}
      run: |
        echo "Checking out branch $BRANCH_NAME for release $PROWLER_VERSION..."
        if git show-ref --verify --quiet "refs/heads/$BRANCH_NAME"; then
@@ -92,8 +94,33 @@ jobs:
        elif git show-ref --verify --quiet "refs/remotes/origin/$BRANCH_NAME"; then
          echo "Branch $BRANCH_NAME exists remotely, checking out..."
          git checkout -b "$BRANCH_NAME" "origin/$BRANCH_NAME"
+        elif [ "$PATCH_VERSION" -eq 0 ]; then
+          # Only cut the branch once the compiled changelogs are on master
+          MASTER_SHA=$(git rev-parse origin/master)
+          VERSION_REGEX="${PROWLER_VERSION//./\\.}"
+          FOUND_CHANGELOG=false
+          for changelog in prowler/CHANGELOG.md api/CHANGELOG.md ui/CHANGELOG.md mcp_server/CHANGELOG.md; do
+            if git show "origin/master:$changelog" 2>/dev/null | grep -qE "^## \[.*\(Prowler v?${VERSION_REGEX}\)"; then
+              FOUND_CHANGELOG=true
+              break
+            fi
+          done
+          if [ "$FOUND_CHANGELOG" != "true" ]; then
+            echo "ERROR: No changelog entry for Prowler $PROWLER_VERSION on master. Merge the compiled changelogs PR first."
+            exit 1
+          fi
+          # Check the SDK version on master before creating the remote branch
+          if ! git show "origin/master:pyproject.toml" | grep -qE "^version = \"${VERSION_REGEX}\"$" \
+            || ! git show "origin/master:prowler/config/config.py" | grep -qE "^prowler_version = \"${VERSION_REGEX}\"$"; then
+            echo "ERROR: pyproject.toml or prowler/config/config.py on master is not at version $PROWLER_VERSION. Fix master before creating $BRANCH_NAME."
+            exit 1
+          fi
+          echo "Branch $BRANCH_NAME does not exist, creating it from master ($MASTER_SHA)..."
+          gh api --method POST "repos/${GITHUB_REPOSITORY}/git/refs" -f ref="refs/heads/${BRANCH_NAME}" -f sha="${MASTER_SHA}"
+          git update-ref "refs/remotes/origin/${BRANCH_NAME}" "$MASTER_SHA"
+          git checkout -b "$BRANCH_NAME" "origin/$BRANCH_NAME"
        else
-          echo "ERROR: Branch $BRANCH_NAME does not exist. For minor releases (X.Y.0), create it manually first. For patch releases (X.Y.Z), the branch should already exist."
+          echo "ERROR: Branch $BRANCH_NAME does not exist. For patch releases (X.Y.Z), the branch should already exist."
          exit 1
        fi

@@ -390,4 +417,3 @@ jobs:
      if: always()
      run: |
        rm -f prowler_changelog.md api_changelog.md ui_changelog.md mcp_changelog.md combined_changelog.md
-
@@ -0,0 +1,27 @@
+name: 'Tools: Release Freeze Lift'
+
+run-name: 'Lift release freeze after Prowler ${{ github.event.release.tag_name }}'
+
+on:
+  release:
+    types:
+      - 'published'
+
+permissions: {}
+
+jobs:
+  release-freeze-lift:
+    if: github.repository == 'prowler-cloud/prowler' && !github.event.release.prerelease
+    runs-on: ubuntu-latest
+    timeout-minutes: 5
+    steps:
+      - name: Harden the runner (Audit all outbound calls)
+        uses: step-security/harden-runner@bf7454d06d71f1098171f2acdf0cd4708d7b5920 # v2.20.0
+        with:
+          egress-policy: audit
+
+      - name: Disable release freeze
+        env:
+          GH_TOKEN: ${{ secrets.PROWLER_BOT_ACCESS_TOKEN }}
+        run: |
+          gh variable set RELEASE_FREEZE --body false --repo "${GITHUB_REPOSITORY}"
@@ -40,6 +40,7 @@ jobs:
            github.com:443
            pypi.org:443
            files.pythonhosted.org:443
+            release-assets.githubusercontent.com:443

      - name: Checkout repository
        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
@@ -81,6 +81,7 @@ jobs:
            auth.docker.io:443
            production.cloudflare.docker.com:443
            production.cloudfront.docker.com:443
+            tuf-repo-cdn.sigstore.dev:443
            api.github.com:443
            mirror.gcr.io:443
            check.trivy.dev:443
@@ -0,0 +1,196 @@
+name: 'SDK: Package Checks'
+
+# Rehearses the PyPI release on every packaging change and once a week, from the
+# consumer's side. Two incidents this guards against:
+#
+# - 5.38.0 shipped an unsatisfiable Requires-Dist (cryptography==50.0.0 while
+#   alibabacloud-tea-openapi and pyopenssl cap it below 49). A [tool.uv] override hid
+#   the conflict inside the repo; pip could not install the wheel and silently
+#   resolved `pip install prowler` to 5.37.1 for a week.
+# - 5.39.0 never published: an unpinned build backend started emitting core metadata
+#   2.5 and the twine bundled in the publish action rejected it.
+#
+# Both were only detectable at release time because nothing built and installed the
+# artifact earlier. The weekly run also catches releases yanked from PyPI after we
+# pinned them (zstd 1.5.7.3, "buggy - not thread safe", sat in uv.lock for months).
+
+on:
+  push:
+    branches:
+      - 'master'
+      - 'v5.*'
+  pull_request:
+    branches:
+      - 'master'
+      - 'v5.*'
+  schedule:
+    # Monday 06:00 UTC. Yanks and upstream releases happen without a commit here.
+    - cron: '0 6 * * 1'
+  workflow_dispatch:
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
+permissions: {}
+
+env:
+  # Must equal the twine bundled in the pypa/gh-action-pypi-publish pin used by
+  # sdk-pypi-release.yml (requirements/runtime.txt in that repo at the pinned tag).
+  # A metadata check that passes here must pass there.
+  TWINE_VERSION: '7.0.0'
+
+jobs:
+  changes:
+    if: github.repository == 'prowler-cloud/prowler'
+    runs-on: ubuntu-latest
+    timeout-minutes: 5
+    permissions:
+      contents: read
+    outputs:
+      # Scheduled and manual runs always execute; pushes and PRs only when a packaging
+      # input changed. Jobs skipped this way still report success to branch protection.
+      run: ${{ github.event_name == 'schedule' || github.event_name == 'workflow_dispatch' || steps.filter.outputs.any_changed == 'true' }}
+
+    steps:
+      - name: Harden the runner (Audit all outbound calls)
+        uses: step-security/harden-runner@bf7454d06d71f1098171f2acdf0cd4708d7b5920 # v2.20.0
+        with:
+          egress-policy: block
+          allowed-endpoints: >
+            github.com:443
+            api.github.com:443
+
+      - name: Checkout repository
+        if: github.event_name == 'push' || github.event_name == 'pull_request'
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+        with:
+          # zizmor: ignore[artipacked]
+          persist-credentials: true # Required by tj-actions/changed-files to fetch PR branch
+
+      - name: Detect packaging changes
+        if: github.event_name == 'push' || github.event_name == 'pull_request'
+        id: filter
+        uses: tj-actions/changed-files@9426d40962ed5378910ee2e21d5f8c6fcbf2dd96 # v47.0.6
+        with:
+          files: |
+            pyproject.toml
+            uv.lock
+            README.md
+            util/replicate_pypi_package.py
+            util/check_yanked_pins.py
+            api/pyproject.toml
+            api/uv.lock
+            mcp_server/pyproject.toml
+            mcp_server/uv.lock
+            .github/workflows/sdk-package-checks.yml
+            .github/workflows/sdk-pypi-release.yml
+            .github/actions/setup-python-uv/**
+
+  install-from-wheel:
+    needs: changes
+    if: needs.changes.outputs.run == 'true'
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    permissions:
+      contents: read
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version:
+          - '3.10'
+          - '3.11'
+          - '3.12'
+          - '3.13'
+        package:
+          - 'prowler'
+        include:
+          # prowler-cloud is the same tree renamed by util/replicate_pypi_package.py;
+          # one Python is enough to prove the rename and its build still work.
+          - python-version: '3.12'
+            package: 'prowler-cloud'
+
+    steps:
+      - name: Harden the runner (Audit all outbound calls)
+        uses: step-security/harden-runner@bf7454d06d71f1098171f2acdf0cd4708d7b5920 # v2.20.0
+        with:
+          egress-policy: block
+          allowed-endpoints: >
+            github.com:443
+            api.github.com:443
+            release-assets.githubusercontent.com:443
+            pypi.org:443
+            files.pythonhosted.org:443
+
+      - name: Checkout repository
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+        with:
+          persist-credentials: false
+
+      - name: Setup Python with uv
+        uses: ./.github/actions/setup-python-uv
+        with:
+          python-version: ${{ matrix.python-version }}
+          install-dependencies: 'false'
+
+      - name: Rename package to prowler-cloud
+        if: matrix.package == 'prowler-cloud'
+        run: |
+          pip install --no-cache-dir toml
+          python util/replicate_pypi_package.py
+
+      - name: Build sdist and wheel
+        run: uv build
+
+      - name: Check metadata with the release workflow's twine
+        run: uvx --from "twine==${TWINE_VERSION}" twine check --strict dist/*
+
+      - name: Install the wheel with pip into a clean virtualenv
+        # Plain pip, --isolated, from outside the repo: consumers never see [tool.uv]
+        # override-dependencies or constraint-dependencies, so neither does this step.
+        run: |
+          python -m venv "${RUNNER_TEMP}/consumer"
+          "${RUNNER_TEMP}/consumer/bin/python" -m pip install --quiet --upgrade pip
+          cd "${RUNNER_TEMP}"
+          "${RUNNER_TEMP}/consumer/bin/python" -m pip install --isolated --no-cache-dir "${GITHUB_WORKSPACE}"/dist/*.whl
+
+      - name: Smoke test the installed CLI
+        run: |
+          cd "${RUNNER_TEMP}"
+          "${RUNNER_TEMP}/consumer/bin/prowler" --version
+          # Loads every AWS check module from the installed wheel: catches files missing
+          # from the package. grep fails the step if the summary line never appears.
+          "${RUNNER_TEMP}/consumer/bin/prowler" aws --list-checks | grep 'available checks'
+
+  pinned-releases-not-yanked:
+    needs: changes
+    if: needs.changes.outputs.run == 'true'
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    permissions:
+      contents: read
+
+    steps:
+      - name: Harden the runner (Audit all outbound calls)
+        uses: step-security/harden-runner@bf7454d06d71f1098171f2acdf0cd4708d7b5920 # v2.20.0
+        with:
+          egress-policy: block
+          allowed-endpoints: >
+            github.com:443
+            api.github.com:443
+            release-assets.githubusercontent.com:443
+            pypi.org:443
+            files.pythonhosted.org:443
+
+      - name: Checkout repository
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+        with:
+          persist-credentials: false
+
+      - name: Set up Python
+        uses: actions/setup-python@e797f83bcb11b83ae66e0230d6156d7c80228e7c # v6.0.0
+        with:
+          python-version: '3.12'
+
+      - name: Check every pinned and locked release against PyPI
+        run: python util/check_yanked_pins.py . api mcp_server
@@ -84,8 +84,18 @@ jobs:
      - name: Build Prowler package
        run: uv build

+      - name: Verify the wheel installs with pip
+        # Same check as "SDK: Package Checks", repeated on the exact artifact about to be
+        # published. Plain pip, --isolated, from outside the repo: an unsatisfiable
+        # Requires-Dist fails here instead of on users' machines (5.38.0 shipped one).
+        run: |
+          python -m venv "${RUNNER_TEMP}/consumer"
+          "${RUNNER_TEMP}/consumer/bin/python" -m pip install --quiet --upgrade pip
+          cd "${RUNNER_TEMP}"
+          "${RUNNER_TEMP}/consumer/bin/python" -m pip install --isolated --no-cache-dir --dry-run "${GITHUB_WORKSPACE}"/dist/*.whl
+
      - name: Publish Prowler package to PyPI
-        uses: pypa/gh-action-pypi-publish@cef221092ed1bacb1cc03d23a2d87d1d172e277b # v1.14.0
+        uses: pypa/gh-action-pypi-publish@dc37677b2e1c63e2034f94d8a5b11f265b73ba33 # v1.14.2
        with:
          print-hash: true

@@ -128,7 +138,17 @@ jobs:
      - name: Build prowler-cloud package
        run: uv build

+      - name: Verify the wheel installs with pip
+        # Same check as "SDK: Package Checks", repeated on the exact artifact about to be
+        # published. Plain pip, --isolated, from outside the repo: an unsatisfiable
+        # Requires-Dist fails here instead of on users' machines (5.38.0 shipped one).
+        run: |
+          python -m venv "${RUNNER_TEMP}/consumer"
+          "${RUNNER_TEMP}/consumer/bin/python" -m pip install --quiet --upgrade pip
+          cd "${RUNNER_TEMP}"
+          "${RUNNER_TEMP}/consumer/bin/python" -m pip install --isolated --no-cache-dir --dry-run "${GITHUB_WORKSPACE}"/dist/*.whl
+
      - name: Publish prowler-cloud package to PyPI
-        uses: pypa/gh-action-pypi-publish@cef221092ed1bacb1cc03d23a2d87d1d172e277b # v1.14.0
+        uses: pypa/gh-action-pypi-publish@dc37677b2e1c63e2034f94d8a5b11f265b73ba33 # v1.14.2
        with:
          print-hash: true
@@ -44,7 +44,10 @@ jobs:
          cache: 'pip'

      - name: Install dependencies
-        run: pip install boto3
+        # Pinned to the versions in pyproject.toml: the ISO partitions region
+        # data comes from the endpoints.json bundled with botocore, so the
+        # botocore version is itself a data source and must be deterministic
+        run: pip install boto3==1.40.61 botocore==1.40.61

      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@d979d5b3a71173a29b74b5b88418bfda9437d885 # v6.1.1
@@ -76,7 +76,7 @@ jobs:

            ### Changes

-            This PR updates the `OCI_COMMERCIAL_REGIONS` dictionary in `prowler/providers/oraclecloud/config.py` with the latest regions fetched from the OCI Identity API (`list_regions()`).
+            This PR updates the `OCI_COMMERCIAL_REGIONS` dictionary in `prowler/providers/oraclecloud/config.py` and the matching list in `ui/lib/provider-credentials/oci-regions.ts` with the latest regions fetched from the OCI Identity API (`list_regions()`).

            - Government regions (`OCI_GOVERNMENT_REGIONS`) are preserved unchanged
            - DOD regions (`OCI_US_DOD_REGIONS`) are preserved unchanged
@@ -640,6 +640,30 @@ jobs:
          flags: prowler-py${{ matrix.python-version }}-e2enetworks
          files: ./e2enetworks_coverage.xml

+      # Huawei Cloud Provider
+      - name: Check if Huawei Cloud files changed
+        if: steps.check-changes.outputs.any_changed == 'true'
+        id: changed-huaweicloud
+        uses: tj-actions/changed-files@9426d40962ed5378910ee2e21d5f8c6fcbf2dd96 # v47.0.6
+        with:
+          files: |
+            ./prowler/**/huaweicloud/**
+            ./tests/**/huaweicloud/**
+            ./uv.lock
+
+      - name: Run Huawei Cloud tests
+        if: steps.changed-huaweicloud.outputs.any_changed == 'true'
+        run: uv run pytest -n auto --cov=./prowler/providers/huaweicloud --cov-report=xml:huaweicloud_coverage.xml tests/providers/huaweicloud
+
+      - name: Upload Huawei Cloud coverage to Codecov
+        if: steps.changed-huaweicloud.outputs.any_changed == 'true'
+        uses: codecov/codecov-action@671740ac38dd9b0130fbe1cec585b89eea48d3de # v5.5.2
+        env:
+          CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
+        with:
+          flags: prowler-py${{ matrix.python-version }}-huaweicloud
+          files: ./huaweicloud_coverage.xml
+
      # External Provider (dynamic loading)
      - name: Check if External Provider files changed
        if: steps.check-changes.outputs.any_changed == 'true'
@@ -57,6 +57,7 @@ jobs:
          egress-policy: block
          allowed-endpoints: >
            github.com:443
+            release-assets.githubusercontent.com:443
            pypi.org:443
            files.pythonhosted.org:443

@@ -78,6 +78,7 @@ jobs:
            auth.docker.io:443
            production.cloudflare.docker.com:443
            production.cloudfront.docker.com:443
+            tuf-repo-cdn.sigstore.dev:443
            registry.npmjs.org:443
            dl-cdn.alpinelinux.org:443
            fonts.googleapis.com:443
@@ -10,12 +10,13 @@ on:
      - master
      - "v5.*"
    paths:
-      - '.github/workflows/ui-e2e-tests-v2.yml'
-      - '.github/test-impact.yml'
-      - 'ui/**'
-      - 'api/**'  # API changes can affect UI E2E
-      - '!ui/CHANGELOG.md'
-      - '!api/CHANGELOG.md'
+      - ".github/workflows/ui-e2e-tests-v2.yml"
+      - ".github/workflows/test-impact-analysis.yml"
+      - ".github/test-impact.yml"
+      - "ui/**"
+      - "api/**" # API changes can affect UI E2E
+      - "!ui/CHANGELOG.md"
+      - "!api/CHANGELOG.md"

 concurrency:
  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
@@ -38,13 +39,16 @@ jobs:
      github.repository == 'prowler-cloud/prowler' &&
      (github.event_name != 'pull_request' || github.event.pull_request.head.repo.fork == false) &&
      (needs.impact-analysis.outputs.has-ui-e2e == 'true' || needs.impact-analysis.outputs.run-all == 'true')
-    runs-on: ubuntu-latest
+    # ubuntu-latest is moving to 26.04 (rollout began October 19, 2026; completion
+    # targeted for November 19, 2026), where `playwright install` refuses to install
+    # Chromium for the pinned Playwright. Stay on 24.04 until the suite upgrades.
+    runs-on: ubuntu-24.04
    env:
-      AUTH_SECRET: 'fallback-ci-secret-for-testing'
+      AUTH_SECRET: "fallback-ci-secret-for-testing"
      AUTH_TRUST_HOST: true
-      NEXTAUTH_URL: 'http://localhost:3000'
-      AUTH_URL: 'http://localhost:3000'
-      UI_API_BASE_URL: 'http://localhost:8080/api/v1'
+      NEXTAUTH_URL: "http://localhost:3000"
+      AUTH_URL: "http://localhost:3000"
+      UI_API_BASE_URL: "http://localhost:8080/api/v1"
      E2E_ADMIN_USER: ${{ secrets.E2E_ADMIN_USER }}
      E2E_ADMIN_PASSWORD: ${{ secrets.E2E_ADMIN_PASSWORD }}
      E2E_AWS_PROVIDER_ACCOUNT_ID: ${{ secrets.E2E_AWS_PROVIDER_ACCOUNT_ID }}
@@ -60,7 +64,7 @@ jobs:
      E2E_M365_SECRET_ID: ${{ secrets.E2E_M365_SECRET_ID }}
      E2E_M365_TENANT_ID: ${{ secrets.E2E_M365_TENANT_ID }}
      E2E_M365_CERTIFICATE_CONTENT: ${{ secrets.E2E_M365_CERTIFICATE_CONTENT }}
-      E2E_KUBERNETES_CONTEXT: 'kind-kind'
+      E2E_KUBERNETES_CONTEXT: "kind-kind"
      E2E_KUBERNETES_KUBECONFIG_PATH: /home/runner/.kube/config
      E2E_GCP_BASE64_SERVICE_ACCOUNT_KEY: ${{ secrets.E2E_GCP_BASE64_SERVICE_ACCOUNT_KEY }}
      E2E_GCP_PROJECT_ID: ${{ secrets.E2E_GCP_PROJECT_ID }}
@@ -237,8 +241,8 @@ jobs:

      - name: Add AWS credentials for testing
        run: |
-          echo "AWS_ACCESS_KEY_ID=${{ secrets.E2E_AWS_PROVIDER_ACCESS_KEY }}" >> .env
-          echo "AWS_SECRET_ACCESS_KEY=${{ secrets.E2E_AWS_PROVIDER_SECRET_KEY }}" >> .env
+          echo "AWS_ACCESS_KEY_ID=${E2E_AWS_PROVIDER_ACCESS_KEY}" >> .env
+          echo "AWS_SECRET_ACCESS_KEY=${E2E_AWS_PROVIDER_SECRET_KEY}" >> .env

      - name: Build API image from current code
        # docker-compose.yml references prowlercloud/prowler-api:latest from the registry,
@@ -292,7 +296,7 @@ jobs:
      - name: Setup Node.js
        uses: actions/setup-node@53b83947a5a98c8d113130e565377fae1a50d02f # v6.3.0
        with:
-          node-version-file: 'ui/.nvmrc'
+          node-version-file: "ui/.nvmrc"

      - name: Setup pnpm
        uses: pnpm/action-setup@fc06bc1257f339d1d5d8b3a19a8cae5388b55320 # v5.0.0
@@ -337,60 +341,65 @@ jobs:
        if: steps.playwright-cache.outputs.cache-hit != 'true'
        run: pnpm run test:e2e:install

-      - name: Run E2E tests
+      - name: Run standard E2E tests
+        id: standard-e2e
        working-directory: ./ui
        run: |
          if [[ "${RUN_ALL_TESTS}" == "true" ]]; then
-            echo "Running ALL E2E tests..."
+            echo "Running all standard E2E tests..."
            pnpm run test:e2e
          else
-            echo "Running targeted E2E tests: ${E2E_TEST_PATHS}"
-            # Convert glob patterns to playwright test paths
-            # e.g., "ui/tests/providers/**" -> "tests/providers"
+            echo "Running targeted standard E2E tests: ${E2E_TEST_PATHS}"
            TEST_PATHS="${E2E_TEST_PATHS}"
-            # Remove ui/ prefix and convert ** to empty (playwright handles recursion)
            TEST_PATHS=$(echo "$TEST_PATHS" | sed 's|ui/||g' | sed 's|\*\*||g' | tr ' ' '\n' | sort -u)
-            # Drop auth setup helpers (not runnable test suites)
-            TEST_PATHS=$(echo "$TEST_PATHS" | grep -v '^tests/setups/')
-            # Safety net: if bare "tests/" appears (from broad patterns like ui/tests/**),
-            # expand to specific subdirs to avoid Playwright discovering setup files
+            TEST_PATHS=$(echo "$TEST_PATHS" | grep -vE '^tests/(setups|registry)/' || true)
+
            if echo "$TEST_PATHS" | grep -qx 'tests/'; then
-              echo "Expanding bare 'tests/' to specific subdirs (excluding setups)..."
              SPECIFIC_DIRS=""
              for dir in tests/*/; do
-                [[ "$dir" == "tests/setups/" ]] && continue
+                [[ "$dir" == "tests/setups/" || "$dir" == "tests/registry/" ]] && continue
                SPECIFIC_DIRS="${SPECIFIC_DIRS}${dir}"$'\n'
              done
-              # Replace "tests/" with specific dirs, keep other paths
-              TEST_PATHS=$(echo "$TEST_PATHS" | grep -vx 'tests/')
+              TEST_PATHS=$(echo "$TEST_PATHS" | grep -vx 'tests/' || true)
              TEST_PATHS="${TEST_PATHS}"$'\n'"${SPECIFIC_DIRS}"
              TEST_PATHS=$(echo "$TEST_PATHS" | grep -v '^$' | sort -u)
            fi
-            if [[ -z "$TEST_PATHS" ]]; then
-              echo "No runnable E2E test paths after filtering setups"
-              exit 0
-            fi
-            # Filter out directories that don't contain any test files
+
            VALID_PATHS=""
-            while IFS= read -r p; do
-              [[ -z "$p" ]] && continue
-              if find "$p" -name '*.spec.ts' -o -name '*.test.ts' 2>/dev/null | head -1 | grep -q .; then
-                VALID_PATHS="${VALID_PATHS}${p}"$'\n'
+            while IFS= read -r path; do
+              [[ -z "$path" ]] && continue
+              if find "$path" -name '*.spec.ts' -o -name '*.test.ts' 2>/dev/null | head -1 | grep -q .; then
+                VALID_PATHS="${VALID_PATHS}${path}"$'\n'
              else
-                echo "Skipping empty test directory: $p"
+                echo "Skipping empty test directory: $path"
              fi
            done <<< "$TEST_PATHS"
            VALID_PATHS=$(echo "$VALID_PATHS" | grep -v '^$' || true)
-            if [[ -z "$VALID_PATHS" ]]; then
-              echo "No test files found in any resolved paths — skipping E2E"
-              exit 0
+
+            if [[ -n "$VALID_PATHS" ]]; then
+              TEST_PATHS=$(echo "$VALID_PATHS" | tr '\n' ' ')
+              echo "Resolved standard test paths: $TEST_PATHS"
+              read -ra test_paths <<< "$TEST_PATHS"
+              pnpm exec playwright test "${test_paths[@]}"
+            else
+              echo "No standard E2E test paths selected."
            fi
-            TEST_PATHS=$(echo "$VALID_PATHS" | tr '\n' ' ')
-            echo "Resolved test paths: $TEST_PATHS"
-            read -ra test_paths <<< "$TEST_PATHS"
-            pnpm exec playwright test "${test_paths[@]}"
          fi

+      - name: Run Registry fixture E2E tests
+        if: |
+          !cancelled() &&
+          (steps.standard-e2e.outcome == 'success' || steps.standard-e2e.outcome == 'failure') &&
+          (env.RUN_ALL_TESTS == 'true' || contains(format(' {0} ', env.E2E_TEST_PATHS), ' ui/tests/registry/'))
+        working-directory: ./ui
+        run: pnpm run test:e2e:registry
+
+      # A worker that crashes on import or a connection test that never resolves
+      # only shows up here; Playwright just reports a missing footer button.
+      - name: Dump API service logs
+        if: failure()
+        run: docker compose logs --no-color --tail=300 api worker worker-beat || true
+
      - name: Upload test reports
        uses: actions/upload-artifact@bbbca2ddaa5d8feaa63e36b76fdaad77386f024f # v7.0.0
        if: failure()
@@ -21,7 +21,10 @@ permissions: {}

 jobs:
  ui-tests:
-    runs-on: ubuntu-latest
+    # ubuntu-latest is moving to 26.04 (rollout began October 19, 2026; completion
+    # targeted for November 19, 2026), where `playwright install` refuses to install
+    # Chromium for the pinned Playwright. Stay on 24.04 until the suite upgrades.
+    runs-on: ubuntu-24.04
    timeout-minutes: 20
    permissions:
      contents: read
@@ -136,10 +139,6 @@ jobs:
        if: steps.check-changes.outputs.any_changed == 'true'
        run: pnpm run healthcheck

-      - name: Check product-tour alignment
-        if: steps.check-changes.outputs.any_changed == 'true'
-        run: pnpm run tour:check
-
      - name: Run pnpm audit
        if: steps.check-changes.outputs.any_changed == 'true'
        run: pnpm run audit
@@ -7,32 +7,19 @@

 ignore:

-  # Modules compiled into the Trivy binary we ship.
-  # Only a Trivy rebuild by its vendor can change these; the version is pinned in our Dockerfile.
-  # CVE-2026-71556 is the same temporary exception documented in .trivyignore.yaml:
-  # Trivy 0.73.0 still embeds go-git 5.19.1, while the 5.19.2 fix is merged only on
-  # Trivy main. Remove this entry with the Trivy exception by 2026-09-15.
-  # https://github.com/aquasecurity/trivy/blob/v0.73.0/go.mod#L46
-  # https://github.com/aquasecurity/trivy/commit/a2edba9a03987ba0d2ebc8212c1a9a1e6979497b
-  - vulnerability: CVE-2026-71556
-    package:
-      name: github.com/go-git/go-git/v5
-  - vulnerability: CVE-2026-56852
-    package:
-      name: golang.org/x/text
-  - vulnerability: GHSA-hrxh-6v49-42gf
-    package:
-      name: google.golang.org/grpc
-  - vulnerability: CVE-2026-50151
-    package:
-      name: oras.land/oras-go/v2
-
  # Shipped inside the PowerShell tarball, in its bundled MicrosoftTeams module.
  # Not a dependency we declare, and not one we can upgrade independently.
  - vulnerability: CVE-2026-26127
    package:
      name: Microsoft.Bcl.Memory

+  # Documented in .trivyignore.yaml: ships inside ExchangeOnlineManagement
+  # 3.10.1, the latest gallery release. Re-check on the next EXO release.
+  - vulnerability: CVE-2026-84008
+    package:
+      name: System.Security.Cryptography.Pkcs
+      version: 10.0.0
+
  # The .NET runtime bundled inside the PowerShell tarball the Dockerfile pins.
  # CVE-2026-62901 is the same temporary exception documented in .trivyignore.yaml:
  # fixed in .NET 9.0.19 / 10.0.11 (2026-08-11), but no published PowerShell release
@@ -81,3 +68,37 @@ ignore:
  - vulnerability: CVE-2026-9669
    package:
      name: python
+  # CVE-2026-82049 (tarfile data/tar filter bypass via a hard link to a symlink) has no
+  # fixed CPython release on any branch: the fix is merged on main and 3.13 only, and the
+  # 3.12 backport is still open. Grype records 3.14.0b1 as the fix, so only-fixed does not
+  # drop it, yet python:3.12.14-slim-trixie reports it too. Prowler never extracts tar
+  # archives to disk: the ECR image inspection reads members in memory with extractfile().
+  # Remove once the base image ships a 3.12 release that includes the backport.
+  # https://github.com/python/cpython/issues/157190
+  # https://github.com/python/cpython/pull/157454
+  - vulnerability: CVE-2026-82049
+    package:
+      name: python
+
+  # zlib in the Alpine base image of the MCP (python:3.13.16-alpine3.23) container.
+  # CVE-2026-85091 is a heap overflow in gz_vacate() reached only through non-blocking
+  # gzwrite() followed by gzprintf() after a write stall; the MCP server never writes
+  # gzip files, so the image never calls that path. The UI image no longer needs this:
+  # node:24.22.1-alpine is built on Alpine 3.24 and ships 1.3.2-r1. TEMPORARY: the fix
+  # landed in the 3.23 index on 2026-10-06, two days after the pinned python image was
+  # built, and could be taken with `apk add --upgrade`, but we wait for the base image
+  # to ship it instead (python:3.13.17-alpine3.23, built 2026-10-22, already does).
+  # Pinned to that version so the rule stops matching on its own once the base image
+  # moves forward. Remove when the MCP digest is bumped, by 2026-11-07.
+  # https://security.alpinelinux.org/vuln/CVE-2026-85091
+  - vulnerability: CVE-2026-85091
+    package:
+      name: zlib
+      version: 1.3.2-r0
+
+  # Documented in .trivyignore.yaml: go-jose 4.1.5 compiled into the Trivy binary (0.76.1,
+  # the latest release). Pinned to 4.1.5 so it stops matching with the next Trivy bump.
+  - vulnerability: CVE-2026-116587
+    package:
+      name: github.com/go-jose/go-jose/v4
+      version: v4.1.5
@@ -68,11 +68,18 @@ vulnerabilities:
    expired_at: 2026-11-30

  # Declared in the SPDX manifest that ships inside PowerShell's MicrosoftTeams module
-  # (Modules/MicrosoftTeams/7.9.0/_manifest/spdx_2.2/manifest.spdx.json). Trivy reads that
+  # (Modules/MicrosoftTeams/8.0.0/_manifest/spdx_2.2/manifest.spdx.json). Trivy reads that
  # SBOM and reports what it declares, which is not the same as what the image contains:
  # there is no Node runtime and no node_modules anywhere in the image, and the .NET
  # assemblies target net472, a Windows-only framework. Nothing here is reachable, and none
  # of it is a dependency we declare -- only Microsoft can change the module's contents.
+# proxy-addr 2.0.7 (CVE-2026-90711) is one more such declaration: no file named proxy-addr
+# exists in the image, only its purl inside that manifest. The UI lock lists proxy-addr 2.0.7
+# through express, but the Next standalone output does not ship it, so the shared entry
+# hides nothing real in the UI image either (checked on prowlercloud/prowler-ui:latest).
+# @modelcontextprotocol/sdk 1.27.1 (CVE-2026-104850) is another: only its purl is in that
+# manifest. The entry is pinned to 1.27.1 because the UI really depends on the SDK (1.26.0
+# via @langchain/mcp-adapters), and a version-less purl would hide that finding too.
  - id: CVE-2020-0606
    purls:
      - "pkg:nuget/Microsoft.WindowsDesktop.App.Ref"
@@ -101,6 +108,23 @@ vulnerabilities:
    purls:
      - "pkg:nuget/System.Security.Cryptography.Xml"
    expired_at: 2027-01-31
+  # Same manifest, pinned to 8.0.4: PowerShell's own copy is also flagged by these CVEs.
+  - id: CVE-2026-96222
+    purls:
+      - "pkg:nuget/System.Security.Cryptography.Xml@8.0.4"
+    expired_at: 2027-01-31
+  - id: CVE-2026-96958
+    purls:
+      - "pkg:nuget/System.Security.Cryptography.Xml@8.0.4"
+    expired_at: 2027-01-31
+  - id: CVE-2026-96959
+    purls:
+      - "pkg:nuget/System.Security.Cryptography.Xml@8.0.4"
+    expired_at: 2027-01-31
+  - id: CVE-2026-96960
+    purls:
+      - "pkg:nuget/System.Security.Cryptography.Xml@8.0.4"
+    expired_at: 2027-01-31
  - id: CVE-2026-13676
    purls:
      - "pkg:npm/fast-uri"
@@ -113,58 +137,100 @@ vulnerabilities:
    purls:
      - "pkg:npm/fast-uri"
    expired_at: 2027-01-31
+  - id: CVE-2026-75899
+    purls:
+      - "pkg:npm/fast-uri"
+    expired_at: 2027-01-31
+  - id: CVE-2026-75975
+    purls:
+      - "pkg:npm/fast-uri"
+    expired_at: 2027-01-31
+  - id: CVE-2026-76172
+    purls:
+      - "pkg:npm/fast-uri"
+    expired_at: 2027-01-31
+  - id: CVE-2026-75931
+    purls:
+      - "pkg:npm/fast-uri"
+    expired_at: 2027-01-31
+  - id: CVE-2026-84292
+    purls:
+      - "pkg:npm/fast-uri"
+    expired_at: 2027-01-31
  - id: CVE-2026-69192
    purls:
      - "pkg:npm/ip-address"
    expired_at: 2027-01-31
+  - id: CVE-2026-101898
+    purls:
+      - "pkg:npm/axios"
+    expired_at: 2027-01-31
+  - id: CVE-2026-101901
+    purls:
+      - "pkg:npm/axios"
+    expired_at: 2027-01-31
+  - id: CVE-2026-101903
+    purls:
+      - "pkg:npm/axios"
+    expired_at: 2027-01-31
+  - id: CVE-2026-101905
+    purls:
+      - "pkg:npm/axios"
+    expired_at: 2027-01-31
+  - id: CVE-2026-101906
+    purls:
+      - "pkg:npm/axios"
+    expired_at: 2027-01-31
+  - id: CVE-2026-101907
+    purls:
+      - "pkg:npm/axios"
+    expired_at: 2027-01-31
+  - id: CVE-2026-101909
+    purls:
+      - "pkg:npm/axios"
+    expired_at: 2027-01-31
+  - id: CVE-2026-90711
+    purls:
+      - "pkg:npm/proxy-addr"
+    expired_at: 2027-01-31
+  - id: CVE-2026-104850
+    purls:
+      - "pkg:npm/%40modelcontextprotocol%2Fsdk@1.27.1"
+    expired_at: 2027-01-31
+  # Two more SDK declarations and follow-redirects 1.16.0 from the same manifest. The SDK
+  # entries stay pinned to 1.27.1 for the reason above. No lock in this repo lists
+  # follow-redirects, so its pin only keeps a future real dependency from being hidden.
+  - id: CVE-2026-111767
+    purls:
+      - "pkg:npm/%40modelcontextprotocol%2Fsdk@1.27.1"
+    expired_at: 2027-01-31
+  - id: CVE-2026-111768
+    purls:
+      - "pkg:npm/%40modelcontextprotocol%2Fsdk@1.27.1"
+    expired_at: 2027-01-31
+  - id: CVE-2026-111755
+    purls:
+      - "pkg:npm/follow-redirects@1.16.0"
+    expired_at: 2027-01-31
+  - id: CVE-2026-111756
+    purls:
+      - "pkg:npm/follow-redirects@1.16.0"
+    expired_at: 2027-01-31

-  # CVE-2026-62901 is a DoS in System.Net.WebSockets (unchecked input for loop condition,
-  # CWE-606), fixed in .NET 9.0.19 / 10.0.11 (published 2026-08-11). The vulnerable runtime
-  # ships inside the PowerShell tarball the Dockerfile pins: 7.5.9 is the latest 7.5.x and
-  # bundles .NET 9.0.18; 7.6.4 bundles .NET 10.0.x < 10.0.11, so no published PowerShell
-  # release contains the fix yet. Prowler only invokes pwsh locally to run M365 module
-  # cmdlets; the image does not accept inbound WebSocket connections, so the DoS path is
-  # not reachable from the network. Remove this temporary suppression as soon as a
-  # PowerShell release shipping .NET 9.0.19+ is available.
-  - id: CVE-2026-62901
+  # Unlike the manifest entries above, this DLL ships in the image:
+  # ExchangeOnlineManagement/3.10.1/netCore/System.Security.Cryptography.Pkcs.dll. It is a
+  # DoS in CMS/PKCS parsing; Prowler only feeds it the tenant's own certificate. 3.10.1 is
+  # the latest gallery release, so only Microsoft can update it. Re-check on the next EXO
+  # release.
+  - id: CVE-2026-84008
    purls:
-      - "pkg:nuget/Microsoft.NETCore.App.Runtime.linux-x64"
-      - "pkg:nuget/Microsoft.NETCore.App.Runtime.linux-arm64"
-    expired_at: 2026-09-15
+      - "pkg:nuget/System.Security.Cryptography.Pkcs@10.0.0"
+    expired_at: 2026-12-15

-  # Modules compiled into the Trivy binary the images ship. The binary is pinned by version
-  # and verified by checksum in the Dockerfile; only a rebuild by its vendor moves these.
-  # CVE-2026-71556 affects go-git worktree operations that can follow symlinks outside a
-  # cloned repository. Trivy 0.72.0 contains go-git 5.19.1, and even the latest published
-  # Trivy release, 0.73.0, still pins that vulnerable version:
-  # https://github.com/aquasecurity/trivy/blob/v0.73.0/go.mod#L46
-  # Trivy main already contains the 5.19.2 fix, but no published release includes it yet:
-  # https://github.com/aquasecurity/trivy/commit/a2edba9a03987ba0d2ebc8212c1a9a1e6979497b
-  # Prowler invokes Trivy only with `fs` on an existing local path or with `image`; it does
-  # not ask Trivy to clone or mutate a Git worktree, so the affected path is not reachable.
-  # Remove this temporary suppression as soon as a fixed Trivy release is available.
-  - id: CVE-2026-71556
+  # go-jose 4.1.5 is compiled into the Trivy binary (usr/local/bin/trivy). Trivy merged the
+  # 4.1.6 bump upstream (aquasecurity/trivy#11488) but 0.76.1 is still the latest release.
+  # Bump TRIVY_VERSION and drop this entry once a release ships it.
+  - id: CVE-2026-116587
    purls:
-      - "pkg:golang/github.com/go-git/go-git/v5"
-    expired_at: 2026-09-15
-
-  - id: CVE-2026-56852
-    purls:
-      - "pkg:golang/golang.org/x/text"
-    expired_at: 2026-12-31
-  - id: GHSA-hrxh-6v49-42gf
-    purls:
-      - "pkg:golang/google.golang.org/grpc"
-    expired_at: 2026-12-31
-  - id: CVE-2026-50151
-    purls:
-      - "pkg:golang/oras.land/oras-go/v2"
-    expired_at: 2026-12-31
-  - id: CVE-2026-50163
-    purls:
-      - "pkg:golang/oras.land/oras-go/v2"
-    expired_at: 2026-12-31
-  - id: CVE-2026-39822
-    purls:
-      - "pkg:golang/stdlib"
-    expired_at: 2026-12-31
+      - "pkg:golang/github.com/go-jose/go-jose/v4@v4.1.5"
+    expired_at: 2026-11-20
@@ -51,7 +51,6 @@ Use these skills for detailed patterns on-demand:
 | `django-migration-psql` | Django migration best practices for PostgreSQL | [SKILL.md](skills/django-migration-psql/SKILL.md) |
 | `postgresql-indexing` | PostgreSQL indexing, EXPLAIN, monitoring, maintenance | [SKILL.md](skills/postgresql-indexing/SKILL.md) |
 | `prowler-attack-paths-query` | Create Attack Paths openCypher queries | [SKILL.md](skills/prowler-attack-paths-query/SKILL.md) |
-| `prowler-tour` | Keep product-tour definitions aligned with the UI | [SKILL.md](skills/prowler-tour/SKILL.md) |
 | `gh-aw` | GitHub Agentic Workflows (gh-aw) | [SKILL.md](skills/gh-aw/SKILL.md) |
 | `skill-creator` | Create new AI agent skills | [SKILL.md](skills/skill-creator/SKILL.md) |

@@ -69,12 +68,10 @@ When performing these actions, ALWAYS invoke the corresponding skill FIRST:
 | Adding new providers | `prowler-provider` |
 | Adding privilege escalation detection queries | `prowler-attack-paths-query` |
 | Adding services to existing providers | `prowler-provider` |
-| Adding, updating, or removing a tour definition (*.tour.ts) | `prowler-tour` |
 | After creating/modifying a skill | `skill-sync` |
 | App Router / Server Actions | `nextjs-16` |
 | Auditing check-to-requirement mappings as a cloud auditor | `prowler-compliance` |
 | Building AI chat features | `ai-sdk-5` |
-| Changing button labels or section headings on a tour-covered page | `prowler-tour` |
 | Committing changes | `prowler-commit` |
 | Configuring MCP servers in agentic workflows | `gh-aw` |
 | Create PR that requires changelog entry | `prowler-changelog` |
@@ -94,7 +91,6 @@ When performing these actions, ALWAYS invoke the corresponding skill FIRST:
 | Creating/updating compliance frameworks | `prowler-compliance` |
 | Debug why a GitHub Actions job is failing | `prowler-ci` |
 | Debugging gh-aw compilation errors | `gh-aw` |
-| Editing a UI file containing data-tour-id attributes | `prowler-tour` |
 | Fill .github/pull_request_template.md (Context/Description/Steps to review/Checklist) | `prowler-pr` |
 | Fixing bug | `tdd` |
 | Fixing compliance JSON bugs (duplicate IDs, empty Section, stale refs) | `prowler-compliance` |
@@ -111,8 +107,6 @@ When performing these actions, ALWAYS invoke the corresponding skill FIRST:
 | Modifying gh-aw workflow frontmatter or safe-outputs | `gh-aw` |
 | Refactoring code | `tdd` |
 | Regenerate AGENTS.md Auto-invoke tables (sync.sh) | `skill-sync` |
-| Renaming or removing a data-tour-id attribute value | `prowler-tour` |
-| Restructuring routes or layouts covered by a tour | `prowler-tour` |
 | Review PR requirements: template, title conventions, changelog gate | `prowler-pr` |
 | Review changelog format and conventions | `prowler-changelog` |
 | Reviewing JSON:API compliance | `jsonapi` |
@@ -1,31 +1,47 @@
-FROM python:3.12.13-slim-trixie@sha256:57cd7c3a7a273101a6485ba99423ee568157882804b1124b4dd04266317710de AS build
+FROM python:3.12.15-slim-trixie@sha256:05cda9777409a9c3ffddd94a4c476b79f0769a0b4857f0c7ed9226b6800b0d6f AS build

 LABEL maintainer="https://github.com/prowler-cloud/prowler"
 LABEL org.opencontainers.image.source="https://github.com/prowler-cloud/prowler"

-ARG POWERSHELL_VERSION=7.5.9
+ARG POWERSHELL_VERSION=7.5.12
 ENV POWERSHELL_VERSION=${POWERSHELL_VERSION}
 # Opt out of PowerShell telemetry (Application Insights -> dc.services.visualstudio.com)
 ENV POWERSHELL_TELEMETRY_OPTOUT=1

-ARG TRIVY_VERSION=0.72.0
+ARG TRIVY_VERSION=0.76.1
 ENV TRIVY_VERSION=${TRIVY_VERSION}

 ARG ZIZMOR_VERSION=1.24.1
 ENV ZIZMOR_VERSION=${ZIZMOR_VERSION}

 # Pinned here, not fetched with the artefact: a compromised release ships its own checksum.
-ARG TRIVY_SHA256_AMD64=bbb64b9695866ce4a7a8f5c9592002c5961cab378577fa3f8a040df362b9b2ea
-ARG TRIVY_SHA256_ARM64=2ca2c023109c2db6b2b77366b6717291452d4531167377d95c79547f0c8e3467
-ARG POWERSHELL_SHA256_AMD64=492ff26bb958336bf61e597ce19e07648b4003bd2a08659e02f0e3e0446ebfe0
-ARG POWERSHELL_SHA256_ARM64=2503b71da3e83635592b092df59a0aca4c3606b4d9b068217bb00be989cb0d56
+ARG TRIVY_SHA256_AMD64=c7cbe3137fb8a48a17574bede3dd6beb16742d1e9772929940fc90082c39d9c0
+ARG TRIVY_SHA256_ARM64=e174dff2a6dba5e3ae41f8d0537a6324405bd08132497340304c6b20e0f16a1f
+ARG POWERSHELL_SHA256_AMD64=db85ae1f45005ffd20e456890585167257151fa99406ccd7ab344f7cf3a766fd
+ARG POWERSHELL_SHA256_ARM64=29eb382cd510e4bd246902650ff6cf9f3b56f959d1bd0be75e0a1e425177c831
 ARG ZIZMOR_SHA256_AMD64=a8000f3c683319a523d3b20df0e75457ba591f049cfcbfa98966631b56733c03
 ARG ZIZMOR_SHA256_ARM64=d66e37ef8a375fb07939c630ebf9709a6e0f20242bdc3faf672a7ed97e0b768d

+# High CVEs fixed in Debian trixie but not yet in the pinned base image:
+#   openssl/libssl3t64/openssl-provider-legacy 3.5.7-1~deb13u2  CVE-2026-14456,
+#   -14457, -18798, -54874, -63072, -63073, -63074, -63075, -63076, -75803
+#   libsqlite3-0 3.46.1-7+deb13u2  CVE-2026-11822, -11824
+#   gzip 1.13-1+deb13u1  CVE-2026-41992
+#   perl-base 5.40.1-6+deb13u1  CVE-2026-42497, -48962, -57432
+#   libssh2-1t64 1.11.1-1+deb13u2  CVE-2026-58050
+#   libpcre2-8-0 10.46-1~deb13u2  CVE-2026-86145, -89161
+#   libbrotli1 1.1.0-2+deb13u1  CVE-2026-111497
+# Taken as a targeted --only-upgrade rather than by moving the digest: the newest
+# published python:3.12-slim-trixie carries the same vulnerable versions. The three
+# openssl packages are flagged separately, so all are named.
+# Drop each one once the base image ships its fixed version.
 # hadolint ignore=DL3008
 RUN apt-get update && apt-get install -y --no-install-recommends \
    wget libicu76 libunwind8 libssl3 libcurl4 ca-certificates apt-transport-https gnupg \
    build-essential pkg-config libzstd-dev zlib1g-dev \
+    && apt-get install -y --no-install-recommends --only-upgrade \
+       util-linux libssl3t64 openssl openssl-provider-legacy \
+       libsqlite3-0 gzip perl-base libssh2-1t64 libpcre2-8-0 libbrotli1 \
    && rm -rf /var/lib/apt/lists/*

 # Install PowerShell
@@ -106,13 +122,15 @@ ENV HOME='/home/prowler'
 ENV PATH="${HOME}/.local/bin:${PATH}"
 #hadolint ignore=DL3013
 RUN pip install --no-cache-dir --upgrade pip && \
-    pip install --no-cache-dir uv==0.12.0
+    pip install --no-cache-dir uv==0.13.4

 RUN uv sync --locked --compile-bytecode && \
    rm -rf ~/.cache/uv

-# Install PowerShell modules
-RUN .venv/bin/python prowler/providers/m365/lib/powershell/m365_powershell.py
+# Install PowerShell modules. Drop the MicrosoftTeams build SBOM: Trivy reports the npm
+# tooling it declares as installed, though none ships (see .trivyignore.yaml).
+RUN .venv/bin/python prowler/providers/m365/lib/powershell/m365_powershell.py && \
+    rm -rf "${HOME}"/.local/share/powershell/Modules/MicrosoftTeams/*/_manifest

 USER root

@@ -126,12 +126,12 @@ Every AWS provider scan will enqueue an Attack Paths ingestion job automatically

 | Provider | Checks | Services | [Compliance Frameworks](https://docs.prowler.com/user-guide/compliance/tutorials/compliance) | [Categories](https://docs.prowler.com/user-guide/cli/tutorials/misc#categories) | Support | Interface |
 |---|---|---|---|---|---|---|
-| AWS | 639 | 86 | 47 | 19 | Official | UI, API, CLI |
-| Azure | 191 | 22 | 21 | 16 | Official | UI, API, CLI |
-| GCP | 109 | 20 | 19 | 12 | Official | UI, API, CLI |
-| Kubernetes | 92 | 7 | 8 | 11 | Official | UI, API, CLI |
+| AWS | 662 | 86 | 50 | 19 | Official | UI, API, CLI |
+| Azure | 191 | 22 | 25 | 16 | Official | UI, API, CLI |
+| GCP | 110 | 20 | 22 | 12 | Official | UI, API, CLI |
+| Kubernetes | 92 | 7 | 11 | 11 | Official | UI, API, CLI |
 | GitHub | 24 | 3 | 2 | 5 | Official | UI, API, CLI |
-| M365 | 143 | 10 | 6 | 10 | Official | UI, API, CLI |
+| M365 | 144 | 10 | 9 | 10 | Official | UI, API, CLI |
 | OCI | 52 | 14 | 5 | 10 | Official | UI, API, CLI |
 | Alibaba Cloud | 63 | 9 | 6 | 9 | Official | UI, API, CLI |
 | Cloudflare | 29 | 3 | 2 | 5 | Official | UI, API, CLI |
@@ -139,7 +139,7 @@ Every AWS provider scan will enqueue an Attack Paths ingestion job automatically
 | MongoDB Atlas | 10 | 3 | 1 | 8 | Official | UI, API, CLI |
 | LLM | [See `promptfoo` docs.](https://www.promptfoo.dev/docs/red-team/plugins/) | N/A | N/A | N/A | Official | CLI |
 | Image | N/A | N/A | N/A | N/A | Official | UI, API, CLI |
-| Google Workspace | 65 | 11 | 3 | 6 | Official | UI, API, CLI |
+| Google Workspace | 65 | 11 | 4 | 6 | Official | UI, API, CLI |
 | OpenStack | 34 | 5 | 1 | 9 | Official | UI, API, CLI |
 | Vercel | 26 | 6 | 1 | 8 | Official | UI, API, CLI |
 | Okta | 29 | 8 | 2 | 2 | Official | UI, API, CLI |
@@ -59,5 +59,9 @@ DJANGO_GITHUB_OAUTH_CLIENT_ID=""
 DJANGO_GITHUB_OAUTH_CLIENT_SECRET=""
 DJANGO_GITHUB_OAUTH_CALLBACK_URL=""

+# Public base URL of the Prowler UI, used to link Jira issues back to findings.
+# Leave empty to omit the link.
+DJANGO_UI_BASE_URL=""
+
 # Deletion Task Batch Size
 DJANGO_DELETION_BATCH_SIZE=5000
@@ -4,6 +4,162 @@ All notable changes to the **Prowler API** are documented in this file.

 <!-- changelog: release notes start -->

+## [1.48.0] (Prowler v5.47.0)
+
+### 🔐 Security
+
+- pymongo to 4.18.3, dnspython to 2.9.0 and oauthlib to 4.0.0 in the API container image, with django-allauth from 65.15.0 to 65.19.7 because earlier releases cap oauthlib below 4, patching the BSON decoding and DNS compression pointer denial of service advisories and the OAuth device flow token disclosure [(#13185)](https://github.com/prowler-cloud/prowler/pull/13185)
+
+---
+
+## [1.47.0] (Prowler v5.46.0)
+
+### 🐞 Fixed
+
+- Compliance report PDF cover table wraps every value, so long account IDs such as OCI tenancy OCIDs no longer overflow their cell [(#13063)](https://github.com/prowler-cloud/prowler/pull/13063)
+- Scan ingestion compares the serialized resource metadata with the stored value and serializes sets in a stable order, so unchanged resources no longer get their metadata, details and updated_at rewritten on every scan [(#13148)](https://github.com/prowler-cloud/prowler/pull/13148)
+- Report generation shares the parsed resource metadata between the findings of a batch and releases each batch before building the next one [(#13149)](https://github.com/prowler-cloud/prowler/pull/13149)
+
+### 🔐 Security
+
+- `GET /resources/metadata`, `GET /resources/metadata/latest`, `GET /findings/metadata` and `GET /findings/metadata/latest` now honour the provider groups of the caller role instead of aggregating every provider in the tenant [(#12964)](https://github.com/prowler-cloud/prowler/pull/12964)
+- Trivy from 0.75.0 to 0.76.0 in the API container image, built with Go 1.27.2 and x/net 0.60.0, patching the HTTP/2 and TLS CVEs bundled in Trivy 0.75.0 [(#13064)](https://github.com/prowler-cloud/prowler/pull/13064)
+- Pillow from 12.3.0 to 13.0.0, certifi to 2026.10.19, aiohttp to 3.14.5 with yarl 1.25.1, uv to 0.13.4 and Trivy to 0.76.1 in the API container image, patching the Pillow JPEG decoder heap overflow, the compromised emSign root CAs bundled in certifi and uv, the aiohttp resource exhaustion advisories and the Go 1.27.2 stdlib CVE in the Trivy binary [(#13107)](https://github.com/prowler-cloud/prowler/pull/13107)
+- Attack Paths custom queries strip block comments before the blocked-operation check and drop APOC virtual nodes and relationships from results, so `LOAD CSV` can no longer be split with a comment nor its response returned under a forged provider label [(#13120)](https://github.com/prowler-cloud/prowler/pull/13120)
+- SAML SSO configuration endpoints require the `manage_account` permission instead of `manage_integrations`, since the configured IdP controls who signs in and which role they receive [(#13121)](https://github.com/prowler-cloud/prowler/pull/13121)
+- djangorestframework from 3.15.2 to 3.18.3 in the API, patching the JSON parser charset and Accept header denial of service and the TokenProxy revocation bypass, and the API container image no longer ships the MicrosoftTeams build SBOM that Trivy read as installed zod and body-parser [(#13157)](https://github.com/prowler-cloud/prowler/pull/13157)
+
+---
+
+## [1.46.0] (Prowler v5.45.0)
+
+### 🐞 Fixed
+
+- OCI provider secrets keep the region as home region for credential validation and scans, instead of always using us-ashburn-1 [(#12865)](https://github.com/prowler-cloud/prowler/pull/12865)
+- Gunicorn workers are now recycled after 10000 requests (plus up to 1000 of jitter) to cap memory growth, configurable with `GUNICORN_MAX_REQUESTS` and `GUNICORN_MAX_REQUESTS_JITTER` [(#13047)](https://github.com/prowler-cloud/prowler/pull/13047)
+
+### 🔐 Security
+
+- `pyjwt` from 2.13.0 to 2.14.0, patching GHSA-ffc3-869f-jxw9 [(#12911)](https://github.com/prowler-cloud/prowler/pull/12911)
+- Sentry error events no longer include user identity, IP addresses, cookies, headers or request bodies, which could contain personal data or provider credentials [(#12912)](https://github.com/prowler-cloud/prowler/pull/12912)
+- Kubernetes kubeconfig validation accepts only inline credentials and known cluster fields, rejecting file-path fields (`tokenFile`, `client-certificate`, `client-key`, `certificate-authority`), command-based authentication (`exec`, `cmd-path`), `auth-provider` directives, proxy URLs and unknown user or cluster keys; stored kubeconfigs are re-validated before each scan and connection test [(#12952)](https://github.com/prowler-cloud/prowler/pull/12952)
+- `django` from 5.1.15 to 5.2.17, patching CVE-2026-15307 [(#12953)](https://github.com/prowler-cloud/prowler/pull/12953)
+- `dulwich` from 1.2.5 to 1.2.9, patching GHSA-5fqc-mrg8-w798, GHSA-8mcx-5rqc-vhmf, GHSA-8w8g-wq8h-fq33 and GHSA-cm62-gvxx-vmxx [(#12953)](https://github.com/prowler-cloud/prowler/pull/12953)
+- `urllib3` from 2.7.0 to 2.8.0, patching CVE-2026-97687 and CVE-2026-97689 [(#12953)](https://github.com/prowler-cloud/prowler/pull/12953)
+- `pymongo` from 4.15.1 to 4.18.2, patching CVE-2026-96748 and CVE-2026-96749 [(#12955)](https://github.com/prowler-cloud/prowler/pull/12955)
+- `GET /api/v1/users/{id}` no longer exposes another user's memberships, roles or tenant names from tenants other than the one in the request token [(#12965)](https://github.com/prowler-cloud/prowler/pull/12965)
+- Python base image from 3.12.13 to 3.12.15 and Trivy from 0.74.0 to 0.75.0 in the API container image, patching CVE-2026-19445, CVE-2026-19553 and the grpc-go and x/crypto CVEs bundled in Trivy [(#13009)](https://github.com/prowler-cloud/prowler/pull/13009)
+- PowerShell from 7.5.11 to 7.5.12 in the API container image, bundling .NET runtime 9.0.21 [(#13037)](https://github.com/prowler-cloud/prowler/pull/13037)
+
+---
+
+## [1.45.0] (Prowler v5.44.0)
+
+### 🚀 Added
+
+- Scan output uploads and downloads can now target S3-compatible object storage such as MinIO directly via `DJANGO_OUTPUT_S3_AWS_ENDPOINT_URL`, instead of relying on process-wide AWS environment variables that also hijacked unrelated AWS API calls [(#12871)](https://github.com/prowler-cloud/prowler/pull/12871)
+
+### 🔄 Changed
+
+- Unify how every endpoint resolves a provider latest completed scan, so overlapping scans no longer make findings, compliance and mute rules read from different scans [(#12858)](https://github.com/prowler-cloud/prowler/pull/12858)
+
+### 🐞 Fixed
+
+- Celery loggers are now declared explicitly in `custom_logging.py` so fatal worker errors are no longer silenced by `disable_existing_loggers=True`. All long-running services in `docker-compose.yml` now have `restart: unless-stopped` so containers recover automatically after unexpected crashes. [(#12465)](https://github.com/prowler-cloud/prowler/pull/12465)
+- Scan report downloads from an S3 bucket with default SSE-KMS encryption no longer fail with an `InvalidArgument` error: when `DJANGO_OUTPUT_S3_AWS_DEFAULT_REGION` is set, presigned download URLs are signed with AWS Signature Version 4 for that region [(#12746)](https://github.com/prowler-cloud/prowler/pull/12746)
+- Adds a periodic sweep that drops orphaned Attack Paths temp Neo4j scan databases left behind when a worker or Neo4j crashes mid-scan, before they accumulate unbounded [(#12832)](https://github.com/prowler-cloud/prowler/pull/12832)
+- Providers whose most recent completed scan has no `completed_at` timestamp are no longer missing from every endpoint that reports a provider's latest scan, which now falls back to scan creation order instead of skipping the provider [(#12858)](https://github.com/prowler-cloud/prowler/pull/12858)
+- Resources no longer keep a stale failed findings count forever when a scoped or imported scan for the same provider completes after a full scan, which used to make the full scan skip its own cleanup [(#12858)](https://github.com/prowler-cloud/prowler/pull/12858)
+- `POST /api/v1/scans` again returns the new scan id in the response `task_args`, which had been empty since the scan broker publish moved to transaction commit [(#12878)](https://github.com/prowler-cloud/prowler/pull/12878)
+- API key authentication no longer locks the key row on every request and now throttles `last_used_at` updates to once per 60 seconds, preventing a hot key from serializing all its requests onto a single locked row [(#12882)](https://github.com/prowler-cloud/prowler/pull/12882)
+- Provider deletion no longer fails when the provider has Attack Paths scans recorded on a sink that is no longer configured, such as Neptune after moving back to Neo4j [(#12894)](https://github.com/prowler-cloud/prowler/pull/12894)
+
+### 🔐 Security
+
+- `DELETE /api/v1/tasks/{id}` requires the permission of the operation that queued the task and rejects provider deletions, and `GET /api/v1/tasks` hides tasks of providers outside the visibility of the role [(#12893)](https://github.com/prowler-cloud/prowler/pull/12893)
+
+---
+
+## [1.44.0] (Prowler v5.43.0)
+
+### 🐞 Fixed
+
+- Report download URLs can be signed against a browser-reachable storage host via `DJANGO_OUTPUT_S3_AWS_PUBLIC_ENDPOINT_URL`, so downloads complete on deployments where storage is only reachable inside the container network [(#12552)](https://github.com/prowler-cloud/prowler/pull/12552)
+- A scan report download no longer fails with a server error when `DJANGO_OUTPUT_S3_AWS_DEFAULT_REGION` is unset, which is common on storage with no meaningful region [(#12552)](https://github.com/prowler-cloud/prowler/pull/12552)
+- Lapsed pending invitations are reported as expired and no longer block a new invitation for the same email [(#12831)](https://github.com/prowler-cloud/prowler/pull/12831)
+
+### 🔐 Security
+
+- `libsqlite3-0`, `gzip`, `perl-base` and `libpcre2-8-0` upgraded in the API container image, patching high Debian CVEs [(#12804)](https://github.com/prowler-cloud/prowler/pull/12804)
+- PowerShell from 7.5.9 to 7.5.11 in the API container image, bundling .NET runtime 9.0.20 and patching CVE-2026-62901 [(#12811)](https://github.com/prowler-cloud/prowler/pull/12811)
+- Bumped `anyio` to 4.14.2 to resolve CVE-2026-63374 [(#12848)](https://github.com/prowler-cloud/prowler/pull/12848)
+
+---
+
+## [1.43.0] (Prowler v5.42.0)
+
+### 🔄 Changed
+
+- Speed up compliance overview ingestion by reading ThreatScore mappings from the compliance template instead of each finding, generating time-ordered `uuid7` row ids and grouping inserted rows by framework and requirement [(#12738)](https://github.com/prowler-cloud/prowler/pull/12738)
+
+---
+
+## [1.42.0] (Prowler v5.41.0)
+
+### 🚀 Added
+
+- Jira issues created from Prowler Cloud now carry the `prowler`, `prowler-<provider>`, `prowler-<severity>`, `prowler-<check-id>`, and `prowler-finding-<finding-uid>` labels, a link back to the finding when `DJANGO_UI_BASE_URL` is configured, and the tenant name [(#12540)](https://github.com/prowler-cloud/prowler/pull/12540)
+
+### 🐞 Fixed
+
+- `POST /api/v1/mute-rules` now updates only each affected provider's latest completed scan and future scans, preventing historical reaggregation from flooding Celery queues [(#12681)](https://github.com/prowler-cloud/prowler/pull/12681)
+
+---
+
+## [1.41.0] (Prowler v5.40.0)
+
+### 🐞 Fixed
+
+- `FINDINGS_TABLE_PARTITION_MAX_AGE_MONTHS` is now applied in months instead of days, and negative values are rejected [(#12580)](https://github.com/prowler-cloud/prowler/pull/12580)
+
+### 🔐 Security
+
+- `sqlparse` upgraded to 0.6.0, patching CVE-2026-54284, CVE-2026-59893, and CVE-2026-71491 [(#12509)](https://github.com/prowler-cloud/prowler/pull/12509)
+- `openssl`, `libssl3t64` and `openssl-provider-legacy` upgraded to 3.5.7-1~deb13u2 in the API container image, patching ten high OpenSSL CVEs [(#12549)](https://github.com/prowler-cloud/prowler/pull/12549)
+
+---
+
+## [1.40.1] (Prowler v5.39.1)
+
+### 🔄 Changed
+
+- Bump alibabacloud-tea-openapi to 0.4.6, oci to 2.184.1 and pyopenssl to 26.4.0 to match the SDK; the cryptography override now names its actual blockers (azure-cli-core pins msal below 1.37, workos 8.3.0 requires cryptography 48) [(#12477)](https://github.com/prowler-cloud/prowler/pull/12477)
+
+### 🐞 Fixed
+
+- Pin zstd to 1.5.7.2; 1.5.7.3 was yanked from PyPI as not thread safe [(#12477)](https://github.com/prowler-cloud/prowler/pull/12477)
+
+### 🔐 Security
+
+- Trivy from v0.72.0 to v0.73.0 in the container image, fixing HIGH CVE-2026-46600 in the bundled `golang.org/x/net` [(#12445)](https://github.com/prowler-cloud/prowler/pull/12445)
+- Trivy v0.74.0 and Debian util-linux 2.41.5-0+deb13u1 in the API container image, patching Go standard library vulnerabilities and CVE-2026-53615 [(#12470)](https://github.com/prowler-cloud/prowler/pull/12470)
+
+---
+
+## [1.40.0] (Prowler v5.39.0)
+
+### 🔄 Changed
+
+- `GET /api/v1/users/me` membership relationships identify the active tenant with `meta.active` for JWT and API key authentication [(#12388)](https://github.com/prowler-cloud/prowler/pull/12388)
+
+### 🐞 Fixed
+
+- Tenant deletion no longer leaves memberships partially removed when exclusive-user cleanup fails [(#12379)](https://github.com/prowler-cloud/prowler/pull/12379)
+- `/api/v1/accounts/saml/{organization_slug}/acs/` rejects non-POST requests before SAML response processing [(#12393)](https://github.com/prowler-cloud/prowler/pull/12393)
+- Social login derives a valid user name when identity providers omit the profile name [(#12413)](https://github.com/prowler-cloud/prowler/pull/12413)
+
+---
+
 ## [1.39.0] (Prowler v5.38.0)

 ### 🚀 Added
@@ -1,26 +1,38 @@
-FROM python:3.12.13-slim-trixie@sha256:57cd7c3a7a273101a6485ba99423ee568157882804b1124b4dd04266317710de AS build
+FROM python:3.12.15-slim-trixie@sha256:05cda9777409a9c3ffddd94a4c476b79f0769a0b4857f0c7ed9226b6800b0d6f AS build

 LABEL maintainer="https://github.com/prowler-cloud/api"

-ARG POWERSHELL_VERSION=7.5.9
+ARG POWERSHELL_VERSION=7.5.12
 ENV POWERSHELL_VERSION=${POWERSHELL_VERSION}
 # Opt out of PowerShell telemetry (Application Insights -> dc.services.visualstudio.com)
 ENV POWERSHELL_TELEMETRY_OPTOUT=1

-ARG TRIVY_VERSION=0.72.0
+ARG TRIVY_VERSION=0.76.1
 ENV TRIVY_VERSION=${TRIVY_VERSION}

 ARG ZIZMOR_VERSION=1.24.1
 ENV ZIZMOR_VERSION=${ZIZMOR_VERSION}

 # Pinned here, not fetched with the artefact: a compromised release ships its own checksum.
-ARG TRIVY_SHA256_AMD64=bbb64b9695866ce4a7a8f5c9592002c5961cab378577fa3f8a040df362b9b2ea
-ARG TRIVY_SHA256_ARM64=2ca2c023109c2db6b2b77366b6717291452d4531167377d95c79547f0c8e3467
-ARG POWERSHELL_SHA256_AMD64=492ff26bb958336bf61e597ce19e07648b4003bd2a08659e02f0e3e0446ebfe0
-ARG POWERSHELL_SHA256_ARM64=2503b71da3e83635592b092df59a0aca4c3606b4d9b068217bb00be989cb0d56
+ARG TRIVY_SHA256_AMD64=c7cbe3137fb8a48a17574bede3dd6beb16742d1e9772929940fc90082c39d9c0
+ARG TRIVY_SHA256_ARM64=e174dff2a6dba5e3ae41f8d0537a6324405bd08132497340304c6b20e0f16a1f
+ARG POWERSHELL_SHA256_AMD64=db85ae1f45005ffd20e456890585167257151fa99406ccd7ab344f7cf3a766fd
+ARG POWERSHELL_SHA256_ARM64=29eb382cd510e4bd246902650ff6cf9f3b56f959d1bd0be75e0a1e425177c831
 ARG ZIZMOR_SHA256_AMD64=a8000f3c683319a523d3b20df0e75457ba591f049cfcbfa98966631b56733c03
 ARG ZIZMOR_SHA256_ARM64=d66e37ef8a375fb07939c630ebf9709a6e0f20242bdc3faf672a7ed97e0b768d

+# High CVEs fixed in Debian trixie but not yet in the pinned base image:
+#   openssl/libssl3t64/openssl-provider-legacy 3.5.7-1~deb13u2  CVE-2026-14456,
+#   -14457, -18798, -54874, -63072, -63073, -63074, -63075, -63076, -75803
+#   libsqlite3-0 3.46.1-7+deb13u2  CVE-2026-11822, -11824
+#   gzip 1.13-1+deb13u1  CVE-2026-41992
+#   perl-base 5.40.1-6+deb13u1  CVE-2026-42497, -48962, -57432
+#   libssh2-1t64 1.11.1-1+deb13u2  CVE-2026-58050
+#   libpcre2-8-0 10.46-1~deb13u2  CVE-2026-86145, -89161
+# Taken as a targeted --only-upgrade rather than by moving the digest: the newest
+# published python:3.12-slim-trixie carries the same vulnerable versions. The three
+# openssl packages are flagged separately, so all are named.
+# Drop each one once the base image ships its fixed version.
 # hadolint ignore=DL3008
 RUN apt-get update && apt-get install -y --no-install-recommends \
    wget \
@@ -36,6 +48,9 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
    libtool \
    libxslt1-dev \
    python3-dev \
+    && apt-get install -y --no-install-recommends --only-upgrade \
+       util-linux libssl3t64 openssl openssl-provider-legacy \
+       libsqlite3-0 gzip perl-base libssh2-1t64 libpcre2-8-0 \
    && rm -rf /var/lib/apt/lists/*

 # Install PowerShell
@@ -110,7 +125,7 @@ RUN mkdir -p /tmp/prowler_api_output
 COPY --chown=prowler:prowler pyproject.toml uv.lock ./

 RUN pip install --no-cache-dir --upgrade pip && \
-    pip install --no-cache-dir uv==0.12.0
+    pip install --no-cache-dir uv==0.13.4

 ENV PATH="/home/prowler/.local/bin:$PATH"

@@ -119,8 +134,10 @@ RUN uv sync --locked --no-install-project && \
    rm -rf ~/.cache/uv

 # Invoked as a module so the base image's Python minor version is not baked
-# into a site-packages path.
-RUN .venv/bin/python -m prowler.providers.m365.lib.powershell.m365_powershell
+# into a site-packages path. Drop the MicrosoftTeams build SBOM: Trivy reports the npm
+# tooling it declares as installed, though none ships (see .trivyignore.yaml).
+RUN .venv/bin/python -m prowler.providers.m365.lib.powershell.m365_powershell && \
+    rm -rf /home/prowler/.local/share/powershell/Modules/MicrosoftTeams/*/_manifest

 USER root

@@ -1 +0,0 @@
-`/api/v1/accounts/saml/{organization_slug}/acs/` rejects non-POST requests before SAML response processing
@@ -1 +0,0 @@
-Tenant deletion no longer leaves memberships partially removed when exclusive-user cleanup fails
@@ -1 +0,0 @@
-`GET /api/v1/users/me` membership relationships identify the active tenant with `meta.active` for JWT and API key authentication
@@ -25,8 +25,8 @@ authors = [{name = "Prowler Engineering", email = "engineering@prowler.com"}]
 dependencies = [
  "celery (==5.6.2)",
  "dj-rest-auth[with_social,jwt] (==7.0.1)",
-  "django (==5.1.15)",
-  "django-allauth[saml] (==65.15.0)",
+  "django (==5.2.17)",
+  "django-allauth[saml] (==65.19.7)",
  "django-celery-beat (==2.9.0)",
  "django-celery-results (==2.6.0)",
  "django-cors-headers==4.4.0",
@@ -34,7 +34,7 @@ dependencies = [
  "django-filter==24.3",
  "django-guid==3.5.0",
  "django-postgres-extra (==2.0.9)",
-  "djangorestframework==3.15.2",
+  "djangorestframework==3.18.3",
  "djangorestframework-jsonapi==7.0.2",
  "djangorestframework-simplejwt (==5.5.1)",
  "drf-nested-routers (==0.95.0)",
@@ -45,7 +45,7 @@ dependencies = [
  "gunicorn==26.0.0",
  "uvloop==0.22.1",
  "lxml==6.1.0",
-  "prowler @ git+https://github.com/prowler-cloud/prowler.git@master",
+  "prowler @ git+https://github.com/prowler-cloud/prowler.git@v5.47",
  "psycopg2-binary==2.9.9",
  "pytest-celery[redis] (==1.3.0)",
  "sentry-sdk[django] (==2.56.0)",
@@ -61,7 +61,7 @@ dependencies = [
  "cartography (==0.138.1)",
  "gevent (==25.9.1)",
  "werkzeug (==3.1.7)",
-  "sqlparse (==0.5.5)",
+  "sqlparse (==0.6.0)",
  "fonttools (==4.62.1)",
  "uvicorn-worker (==0.4.0)"
 ]
@@ -71,7 +71,7 @@ name = "prowler-api"
 package-mode = false
 # Needed for the SDK compatibility
 requires-python = ">=3.11,<3.13"
-version = "1.40.0"
+version = "1.48.1"

 # Shared ruff baseline (kept in sync with mcp_server/pyproject.toml).
 # target-version tracks this project's lowest supported Python.
@@ -92,8 +92,7 @@ extend-select = [

 [tool.uv]
 # Transitive pins matching master to avoid silent drift; bump deliberately.
-# workos and pyopenssl run ahead of master: the versions master pins cap cryptography
-# below 48, so both were bumped to versions that allow it (PROWLER-2310).
+# workos is api-only; pyopenssl matches master (PROWLER-2310).
 constraint-dependencies = [
  "about-time==4.2.1",
  "adal==1.2.7",
@@ -101,7 +100,7 @@ constraint-dependencies = [
  "aiobotocore==2.25.1",
  "aiofiles==24.1.0",
  "aiohappyeyeballs==2.6.1",
-  "aiohttp==3.14.3",
+  "aiohttp==3.14.5",
  "aioitertools==0.13.0",
  "aiosignal==1.4.0",
  "alibabacloud-actiontrail20200706==2.4.1",
@@ -130,7 +129,7 @@ constraint-dependencies = [
  "alibabacloud-sls20201230==5.9.0",
  "alibabacloud-sts20150401==1.1.6",
  "alibabacloud-tea==0.4.3",
-  "alibabacloud-tea-openapi==0.4.5",
+  "alibabacloud-tea-openapi==0.4.6",
  "alibabacloud-tea-util==0.3.14",
  "alibabacloud-tea-xml==0.0.3",
  "alibabacloud-vpc20160428==6.13.0",
@@ -138,7 +137,7 @@ constraint-dependencies = [
  "aliyun-log-fastpb==0.2.0",
  "amqp==5.3.1",
  "annotated-types==0.7.0",
-  "anyio==4.12.1",
+  "anyio==4.14.2",
  "applicationinsights==0.11.10",
  "apscheduler==3.11.2",
  "argcomplete==3.5.3",
@@ -197,7 +196,7 @@ constraint-dependencies = [
  "botocore==1.40.61",
  "cartography==0.138.1",
  "celery==5.6.2",
-  "certifi==2026.1.4",
+  "certifi==2026.10.19",
  "cffi==2.0.0",
  "charset-normalizer==3.4.4",
  "circuitbreaker==2.1.3",
@@ -223,8 +222,8 @@ constraint-dependencies = [
  "dill==0.4.1",
  "distro==1.9.0",
  "dj-rest-auth==7.0.1",
-  "django==5.1.15",
-  "django-allauth==65.15.0",
+  "django==5.2.17",
+  "django-allauth==65.19.7",
  "django-celery-beat==2.9.0",
  "django-celery-results==2.6.0",
  "django-cors-headers==4.4.0",
@@ -235,10 +234,10 @@ constraint-dependencies = [
  "django-postgres-extra==2.0.9",
  "django-silk==5.3.2",
  "django-timezone-field==7.2.1",
-  "djangorestframework==3.15.2",
+  "djangorestframework==3.18.3",
  "djangorestframework-jsonapi==7.0.2",
  "djangorestframework-simplejwt==5.5.1",
-  "dnspython==2.8.0",
+  "dnspython==2.9.0",
  "docker==7.1.0",
  "dogpile-cache==1.5.0",
  "dparse==0.6.4",
@@ -247,7 +246,7 @@ constraint-dependencies = [
  "drf-simple-apikey==2.2.1",
  "drf-spectacular==0.27.2",
  "drf-spectacular-jsonapi==0.5.1",
-  "dulwich==1.2.5",
+  "dulwich==1.2.9",
  "duo-client==5.5.0",
  "durationpy==0.10",
  "email-validator==2.2.0",
@@ -338,8 +337,8 @@ constraint-dependencies = [
  "nest-asyncio==1.6.0",
  "nltk==3.9.4",
  "numpy==2.2.6",
-  "oauthlib==3.3.1",
-  "oci==2.183.0",
+  "oauthlib==4.0.0",
+  "oci==2.184.1",
  "openai==1.109.1",
  "openstacksdk==4.2.0",
  "opentelemetry-api==1.39.1",
@@ -351,7 +350,7 @@ constraint-dependencies = [
  "pagerduty==6.1.0",
  "pandas==2.2.3",
  "pbr==7.0.3",
-  "pillow==12.3.0",
+  "pillow==13.0.0",
  "pkginfo==1.12.1.2",
  "platformdirs==4.5.1",
  "plotly==6.5.2",
@@ -376,11 +375,11 @@ constraint-dependencies = [
  "pydantic-core==2.41.5",
  "pygithub==2.8.0",
  "pygments==2.20.0",
-  "pyjwt==2.13.0",
+  "pyjwt==2.14.0",
  "pylint==3.2.5",
  "pymsalruntime==0.18.1",
  "pynacl==1.6.2",
-  "pyopenssl==26.2.0",
+  "pyopenssl==26.4.0",
  "pyparsing==3.3.2",
  "pyreadline3==3.5.4",
  "pysocks==1.7.1",
@@ -424,7 +423,7 @@ constraint-dependencies = [
  "six==1.17.0",
  "slack-sdk==3.39.0",
  "sniffio==1.3.1",
-  "sqlparse==0.5.5",
+  "sqlparse==0.6.0",
  "statsd==4.0.1",
  "std-uritemplate==2.0.8",
  "stevedore==5.6.0",
@@ -440,7 +439,7 @@ constraint-dependencies = [
  "tzdata==2025.3",
  "tzlocal==5.3.1",
  "uritemplate==4.2.0",
-  "urllib3==2.7.0",
+  "urllib3==2.8.0",
  "uuid6==2024.7.10",
  "uvicorn==0.49.0",
  "uvloop==0.22.1",
@@ -454,11 +453,11 @@ constraint-dependencies = [
  "xlsxwriter==3.2.9",
  "xmlsec==1.3.17",
  "xmltodict==1.0.2",
-  "yarl==1.22.0",
+  "yarl==1.25.1",
  "zipp==3.23.0",
  "zope-event==6.1",
  "zope-interface==8.2",
-  "zstd==1.5.7.3"
+  "zstd==1.5.7.2"
 ]
 # prowler@master needs okta==3.4.2, but cartography 0.138.1 requires okta<1.0.0.
 # Attack Paths does not ingest Okta today, so override the Cartography
@@ -476,16 +475,23 @@ constraint-dependencies = [
 # a constraint cannot satisfy against the SDK's hard pins, so override the whole set
 # to 1.9.10 until the SDK bump propagates to the pinned master rev.
 #
-# prowler@master hard-pins dulwich==0.23.0 and pyjwt==2.12.1 in [project.dependencies].
-# dulwich 1.2.5 patches GHSA-897w-fcg9-f6xj (arbitrary file write) and pyjwt 2.13.0
-# patches PYSEC-2026-179 (HMAC/JWK key-confusion); a constraint cannot satisfy these
-# against the SDK's hard pins, so override them to the patched versions until the SDK
-# bump propagates to the pinned master rev. pyjwt keeps the [crypto] extra because an
+# The API installs prowler from a pinned master git rev whose dulwich and pyjwt pins trail
+# the root pyproject.toml. dulwich 1.2.9 patches GHSA-5fqc-mrg8-w798, GHSA-8mcx-5rqc-vhmf,
+# GHSA-8w8g-wq8h-fq33 and GHSA-cm62-gvxx-vmxx, and pyjwt 2.14.0 patches GHSA-ffc3-869f-jxw9,
+# so override both until that rev is refreshed. pyjwt keeps the [crypto] extra because an
 # override replaces the whole requirement; bare pyjwt would drop it from the consumers
 # that request pyjwt[crypto] and leave cryptography (needed for RS256) only transitive.
 override-dependencies = [
  "okta==3.4.2",
-  # alibabacloud-tea-openapi 0.4.5 caps cryptography below 49 and is the latest release.
+  # huaweicloudsdkcore (via prowler) pins pymongo==4.15.1 exactly; CVE-2026-96748 and
+  # CVE-2026-96749 are fixed in 4.18.2. Only bson types are used, so the override is safe.
+  "pymongo==4.18.3",
+  # prowler requires cryptography==50.0.0. Two api-only dependencies still cap it below
+  # 49 and cannot move yet: msal, pinned exactly by azure-cli-core (2.83.0 -> 1.35.0b1,
+  # 2.89.1 -> 1.36.0, both <49; cartography needs azure-cli-core), and workos 8.3.0
+  # (~=48.0; workos 10.1.1+ needs ~=50.0 and is a separate SDK upgrade). This api is
+  # deployed from this lock with `uv sync --locked`, so the override applies to what runs.
+  # Remove when azure-cli-core pins msal>=1.37.0 and workos is on 10.x.
  "cryptography==50.0.0",
  "azure-mgmt-containerservice==34.1.0",
  "microsoft-kiota-abstractions==1.9.10",
@@ -495,6 +501,6 @@ override-dependencies = [
  "microsoft-kiota-serialization-json==1.9.10",
  "microsoft-kiota-serialization-multipart==1.9.10",
  "microsoft-kiota-serialization-text==1.9.10",
-  "dulwich==1.2.5",
-  "pyjwt[crypto]==2.13.0"
+  "dulwich==1.2.9",
+  "pyjwt[crypto]==2.14.0"
 ]
@@ -12,11 +12,37 @@ from api.models import (
    UserRoleRelationship,
 )
 from api.utils import accept_invitation_for_user
+from django.core.exceptions import ValidationError
 from django.db import transaction
 from django.http import HttpResponseForbidden


 class ProwlerSocialAccountAdapter(DefaultSocialAccountAdapter):
+    @staticmethod
+    def _get_social_account_name(extra_data: dict, email: str) -> str:
+        name_field = User._meta.get_field("name")
+        for value in (
+            extra_data.get("name"),
+            extra_data.get("login"),
+            extra_data.get("username"),
+            email,
+        ):
+            if not isinstance(value, str):
+                continue
+
+            candidate = value.strip()[: name_field.max_length].rstrip()
+            if not candidate:
+                continue
+
+            try:
+                name_field.run_validators(candidate)
+            except ValidationError:
+                continue
+
+            return candidate
+
+        raise ValueError("Social account does not provide a valid user identity.")
+
    @staticmethod
    def get_user_by_email(email: str):
        try:
@@ -116,11 +142,8 @@ class ProwlerSocialAccountAdapter(DefaultSocialAccountAdapter):

            if provider != "saml":
                # Handle other providers (e.g., GitHub, Google)
+                user.name = self._get_social_account_name(extra, user.email)
                user.save(using=MainRouter.admin_db)
-                social_account_name = extra.get("name")
-                if social_account_name:
-                    user.name = social_account_name
-                    user.save(using=MainRouter.admin_db)

                invitation_token = self._get_invitation_token(request)
                if invitation_token:
@@ -12,7 +12,7 @@ Two responsibilities:

 Label-injection pipeline:

-1. **Protect** string literals and line comments (placeholder replacement).
+1. **Protect** string literals and comments (placeholder replacement).
 2. **Split** by top-level clause keywords to track clause context.
 3. **Pass A** - inject into *labeled* node patterns in ALL segments.
 4. **Pass B** - inject into *bare* node patterns in MATCH segments only.
@@ -25,10 +25,12 @@ from rest_framework.exceptions import ValidationError
 from tasks.jobs.attack_paths.config import get_provider_label

 # Step 1 - String / comment protection
-# Single combined regex: strings first, then line comments
+# Single combined regex: strings first, then line and block comments
 # The regex engine finds the leftmost match, so a string like 'https://prowler.com'
 # is consumed as a string before the // inside it can match as a comment
-_PROTECTED_RE = re.compile(r"'(?:[^'\\]|\\.)*'|\"(?:[^\"\\]|\\.)*\"|//[^\n]*")
+_PROTECTED_RE = re.compile(
+    r"'(?:[^'\\]|\\.)*'|\"(?:[^\"\\]|\\.)*\"|//[^\n]*|/\*[\s\S]*?\*/"
+)

 # Step 2 - Clause splitting
 # `OPTIONAL MATCH` must come before `MATCH` to avoid partial matching
@@ -143,18 +145,26 @@ def inject_label(cypher: str, label: str) -> str:

 # Patterns that indicate SSRF or dangerous procedure calls
 # Defense-in-depth layer - the primary control is `neo4j.READ_ACCESS`
+# Neo4j accepts whitespace (and therefore comments) around the dots of a
+# qualified procedure name, so `apoc . load` is `apoc.load`.
+_BLOCKED_APOC_NAMESPACES = (
+    "load",
+    "import",
+    "export",
+    "cypher",
+    "systemdb",
+    "config",
+    "periodic",
+    "do",
+    "trigger",
+    "custom",
+)
 _BLOCKED_PATTERNS = [
    re.compile(r"\bLOAD\s+CSV\b", re.IGNORECASE),
-    re.compile(r"\bapoc\.load\b", re.IGNORECASE),
-    re.compile(r"\bapoc\.import\b", re.IGNORECASE),
-    re.compile(r"\bapoc\.export\b", re.IGNORECASE),
-    re.compile(r"\bapoc\.cypher\b", re.IGNORECASE),
-    re.compile(r"\bapoc\.systemdb\b", re.IGNORECASE),
-    re.compile(r"\bapoc\.config\b", re.IGNORECASE),
-    re.compile(r"\bapoc\.periodic\b", re.IGNORECASE),
-    re.compile(r"\bapoc\.do\b", re.IGNORECASE),
-    re.compile(r"\bapoc\.trigger\b", re.IGNORECASE),
-    re.compile(r"\bapoc\.custom\b", re.IGNORECASE),
+    *(
+        re.compile(rf"\bapoc\s*\.\s*{namespace}\b", re.IGNORECASE)
+        for namespace in _BLOCKED_APOC_NAMESPACES
+    ),
 ]


@@ -162,10 +172,11 @@ def validate_custom_query(cypher: str) -> None:
    """Reject queries containing known SSRF or dangerous procedure patterns.

    Raises ValidationError if a blocked pattern is found.
-    String literals and comments are stripped before matching to avoid
-    false positives.
+    String literals and comments are replaced by a space before matching:
+    literals cannot trigger false positives and a comment cannot split a
+    keyword pair such as `LOAD /**/ CSV`, which Neo4j still parses as `LOAD CSV`.
    """
-    stripped = _PROTECTED_RE.sub("", cypher)
+    stripped = _PROTECTED_RE.sub(" ", cypher)
    for pattern in _BLOCKED_PATTERNS:
        if pattern.search(stripped):
            raise ValidationError({"query": "Query contains a blocked operation"})
@@ -207,6 +207,11 @@ def drop_database(database: str) -> None:
    sink_module.get_backend().drop_database(database)


+def list_databases() -> list[str]:
+    """List database names on the ingest cluster. Temp scan DBs always live here."""
+    return ingest.list_databases()
+
+
 def drop_subgraph(database: str, provider_id: str) -> int:
    return sink_module.get_backend().drop_subgraph(database, provider_id)

@@ -13,6 +13,7 @@ from api.attack_paths.ingest.driver import (
    get_session,
    get_uri,
    init_driver,
+    list_databases,
    run_cypher,
 )

@@ -25,5 +26,6 @@ __all__ = [
    "get_session",
    "get_uri",
    "init_driver",
+    "list_databases",
    "run_cypher",
 ]
@@ -165,6 +165,14 @@ def drop_database(database: str) -> None:
        session.run(f"DROP DATABASE `{database}` IF EXISTS DESTROY DATA")


+def list_databases() -> list[str]:
+    """List every database name on the Neo4j temp-database cluster."""
+    # A cluster returns one row per hosting server, so dedupe on name
+    with get_session() as session:
+        result = session.run("SHOW DATABASES YIELD name RETURN DISTINCT name")
+        return [record["name"] for record in result]
+
+
 def clear_cache(database: str) -> None:
    """Best-effort cache clear for a Neo4j database."""
    from api.attack_paths.database import GraphDatabaseQueryException
@@ -45,6 +45,10 @@ class SinkDatabase(Protocol):
        parameters: dict[str, Any] | None = None,
    ) -> neo4j.graph.Graph: ...

+    def is_persisted_element_id(self, element_id: str) -> bool:
+        """Whether an element id belongs to a stored entity rather than one minted at query time."""
+        return True
+
    def create_database(self, database: str) -> None: ...

    def drop_database(self, database: str) -> None: ...
@@ -7,6 +7,7 @@ path while phase-1 drains tenant DBs.

 import atexit
 import logging
+import re
 import threading
 import time
 from collections.abc import Iterator
@@ -186,6 +187,13 @@ class Neo4jSink(SinkDatabase):

            return session.execute_read(_run)

+    # APOC virtual nodes report a bare negative number and virtual
+    # relationships a random UUID, neither in the stored-entity format.
+    _PERSISTED_ELEMENT_ID_RE = re.compile(r"^\d+:[^:]+:\d+$")
+
+    def is_persisted_element_id(self, element_id: str) -> bool:
+        return bool(self._PERSISTED_ELEMENT_ID_RE.match(element_id))
+
    def create_database(self, database: str) -> None:
        with self.get_session() as session:
            session.run(
@@ -15,6 +15,7 @@ from api.attack_paths.queries.schema import (
    RAW_SCHEMA_URL,
    get_cartography_schema_query,
 )
+from api.attack_paths.sink.base import SinkDatabase
 from api.models import AttackPathsScan
 from config.custom_logging import BackendLogger
 from config.env import env
@@ -135,7 +136,7 @@ def execute_query(
            cypher = inject_provider_label(cypher, provider_id)

        graph = backend.execute_read_query(database_name, cypher, parameters)
-        return _serialize_graph(graph, provider_id)
+        return _serialize_graph(graph, provider_id, backend)

    except graph_database.WriteQueryNotAllowedException:
        raise PermissionDenied(
@@ -174,6 +175,8 @@ def execute_custom_query(
    # 1. `neo4j.READ_ACCESS` - prevents mutations at the driver level
    # 2. `inject_provider_label()` - regex-based label injection scopes node patterns
    # 3. `_serialize_graph()` - post-query filter drops nodes without the provider label
+    #    and any node or relationship the backend did not store (APOC virtual
+    #    entities can carry an arbitrary label and arbitrary properties)
    # 4. `USING QUERY:TIMEOUTMILLISECONDS` on Neptune - server-side runaway cutoff
    #
    # Layer 2 is best-effort (regex can't fully parse Cypher);
@@ -193,7 +196,7 @@ def execute_custom_query(

    try:
        graph = backend.execute_read_query(database_name, cypher, None)
-        serialized = _serialize_graph(graph, provider_id)
+        serialized = _serialize_graph(graph, provider_id, backend)
        return _truncate_graph(serialized)

    except graph_database.ClientStatementException as exc:
@@ -265,13 +268,15 @@ def _truncate_graph(graph: dict[str, Any]) -> dict[str, Any]:
    return graph


-def _serialize_graph(graph, provider_id: str) -> dict[str, Any]:
+def _serialize_graph(graph, provider_id: str, backend: SinkDatabase) -> dict[str, Any]:
    provider_label = get_provider_label(provider_id)

    nodes = []
    kept_node_ids = set()
    for node in graph.nodes:
-        if provider_label not in node.labels:
+        if provider_label not in node.labels or not backend.is_persisted_element_id(
+            node.element_id
+        ):
            continue

        kept_node_ids.add(node.element_id)
@@ -286,7 +291,7 @@ def _serialize_graph(graph, provider_id: str) -> dict[str, Any]:
    filtered_count = len(graph.nodes) - len(nodes)
    if filtered_count > 0:
        logger.debug(
-            f"Filtered {filtered_count} nodes without provider label {provider_label}"
+            f"Filtered {filtered_count} nodes without provider label {provider_label} or not stored"
        )

    relationships = []
@@ -294,6 +299,7 @@ def _serialize_graph(graph, provider_id: str) -> dict[str, Any]:
        if (
            relationship.start_node.element_id not in kept_node_ids
            or relationship.end_node.element_id not in kept_node_ids
+            or not backend.is_persisted_element_id(relationship.element_id)
        ):
            continue

@@ -1,4 +1,5 @@
 import logging
+from datetime import timedelta
 from math import isfinite
 from uuid import UUID

@@ -6,7 +7,7 @@ from api.db_router import MainRouter
 from api.models import TenantAPIKey, TenantAPIKeyManager
 from cryptography.fernet import InvalidToken
 from django.core.exceptions import ObjectDoesNotExist
-from django.db import transaction
+from django.db.models import Q
 from django.utils import timezone
 from drf_simple_apikey.backends import APIKeyAuthentication as BaseAPIKeyAuth
 from drf_simple_apikey.crypto import get_crypto
@@ -18,12 +19,15 @@ from rest_framework_simplejwt.authentication import JWTAuthentication

 logger = logging.getLogger(__name__)

+# Writing on every request makes all requests of a busy key contend on one row
+API_KEY_LAST_USED_AT_THROTTLE_SECONDS = 60
+

 class OrphanedAPIKeyError(Exception):
    """Raised when an API key outlived the user that owns it.

-    Handled by `authenticate`, which commits the revocation written while detecting it
-    and then rejects the request with `AuthenticationFailed`.
+    The revocation is written by a plain `update()` before this is raised, so it is
+    already persisted by the time `authenticate` catches it and rejects the request.
    """


@@ -37,8 +41,9 @@ class TenantAPIKeyAuthentication(BaseAPIKeyAuth):
        """
        Override to use admin connection, bypassing RLS during authentication.

-        Returns the validated API key row, locked with `select_for_update`, so callers
-        must run inside `transaction.atomic(using=MainRouter.admin_db)`.
+        Returns the validated API key row from a single read. `authenticate` builds
+        the auth claims from that same row instead of looking it up again, so a key
+        revoked or orphaned right after validation can't still authenticate.
        """
        try:
            payload = self.key_crypto.decrypt(key)
@@ -67,9 +72,11 @@ class TenantAPIKeyAuthentication(BaseAPIKeyAuth):
            raise AuthenticationFailed("API Key has already expired.")

        try:
+            # Loading `entity` in the same query keeps a user deleted after this read
+            # from turning the later `api_key.entity` access into a 500
            api_key = (
                self.model.objects.using(MainRouter.admin_db)
-                .select_for_update()
+                .select_related("entity")
                .get(id=api_key_pk)
            )
        except ObjectDoesNotExist:
@@ -85,8 +92,9 @@ class TenantAPIKeyAuthentication(BaseAPIKeyAuth):
        # Revoke it as well, so it stops showing up as active and later attempts fail
        # the `revoked` check above like any other revoked key.
        if api_key.entity_id is None:
-            api_key.revoked = True
-            api_key.save(update_fields=["revoked"], using=MainRouter.admin_db)
+            self.model.objects.using(MainRouter.admin_db).filter(
+                id=api_key.id, revoked=False
+            ).update(revoked=True)
            logger.warning(
                "Revoked orphaned API key: prefix=%s tenant=%s",
                api_key.prefix,
@@ -112,34 +120,38 @@ class TenantAPIKeyAuthentication(BaseAPIKeyAuth):
        except ValueError:
            raise AuthenticationFailed("Invalid API Key.")

-        # Validation, the `last_used_at` update and the auth claims all read the same
-        # row, locked until the transaction ends. Looking the key up a second time to
-        # build the claims used to leave a window where a key revoked or orphaned right
-        # after passing validation still authenticated.
-        with transaction.atomic(using=MainRouter.admin_db):
-            try:
-                api_key = self._authenticate_credentials(request, key)
-            except OrphanedAPIKeyError:
-                # Rejected below instead of here: leaving the block normally commits
-                # the revocation `_authenticate_credentials` wrote, while raising from
-                # inside would roll it back.
-                pass
-            else:
-                # The prefix used to be checked by the second lookup
-                if api_key.prefix != prefix:
-                    raise AuthenticationFailed("Invalid API Key.")
+        try:
+            api_key = self._authenticate_credentials(request, key)
+        except OrphanedAPIKeyError:
+            raise AuthenticationFailed("No entity matching this api key.")

-                api_key.last_used_at = timezone.now()
-                api_key.save(update_fields=["last_used_at"], using=MainRouter.admin_db)
+        # The prefix used to be checked by the second lookup
+        if api_key.prefix != prefix:
+            raise AuthenticationFailed("Invalid API Key.")

-                entity = api_key.entity
-                return entity, {
-                    "tenant_id": str(api_key.tenant_id),
-                    "sub": str(entity.id),
-                    "api_key_prefix": api_key.prefix,
-                }
+        self._throttled_touch_last_used_at(api_key)

-        raise AuthenticationFailed("No entity matching this api key.")
+        entity = api_key.entity
+        return entity, {
+            "tenant_id": str(api_key.tenant_id),
+            "sub": str(entity.id),
+            "api_key_prefix": api_key.prefix,
+        }
+
+    @staticmethod
+    def _throttled_touch_last_used_at(api_key: TenantAPIKey) -> None:
+        """Write `last_used_at` at most once per throttle interval, without locking the row."""
+        now = timezone.now()
+        stale_before = now - timedelta(seconds=API_KEY_LAST_USED_AT_THROTTLE_SECONDS)
+
+        if api_key.last_used_at is not None and api_key.last_used_at >= stale_before:
+            return
+
+        TenantAPIKey.objects.using(MainRouter.admin_db).filter(
+            id=api_key.id, revoked=False
+        ).filter(
+            Q(last_used_at__isnull=True) | Q(last_used_at__lt=stale_before)
+        ).update(last_used_at=now)


 class CombinedJWTOrAPIKeyAuthentication(BaseAuthentication):
@@ -409,7 +409,7 @@ def batch_delete(tenant_id, queryset, batch_size=settings.DJANGO_DELETION_BATCH_

    Args:
        tenant_id (str): Tenant ID the queryset belongs to.
-        queryset (QuerySet): The queryset of objects to delete.
+        queryset: The queryset of objects to delete.
        batch_size (int): The number of objects to delete in each batch.

    Returns:
@@ -1439,8 +1439,20 @@ class InvitationFilter(FilterSet):
    inserted_at = DateFilter(field_name="inserted_at", lookup_expr="date")
    updated_at = DateFilter(field_name="updated_at", lookup_expr="date")
    expires_at = DateFilter(field_name="expires_at", lookup_expr="date")
-    state = ChoiceFilter(choices=Invitation.State.choices)
-    state__in = ChoiceInFilter(choices=Invitation.State.choices, lookup_expr="in")
+    state = ChoiceFilter(choices=Invitation.State.choices, method="filter_state")
+    state__in = ChoiceInFilter(
+        choices=Invitation.State.choices, lookup_expr="in", method="filter_state_in"
+    )
+
+    def filter_state(self, queryset, name, value):
+        return self.filter_state_in(queryset, name, [value])
+
+    def filter_state_in(self, queryset, name, value):
+        lapsed = Invitation.lapsed_q()
+        query = Q(state__in=value) & ~lapsed
+        if Invitation.State.EXPIRED in value:
+            query |= lapsed
+        return queryset.filter(query)

    class Meta:
        model = Invitation
@@ -0,0 +1,116 @@
+import uuid
+
+import api.rls
+import django.db.models.deletion
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("api", "0097_attack_paths_scan_db_defaults"),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="TenantOnboardingProfile",
+            fields=[
+                (
+                    "id",
+                    models.UUIDField(
+                        default=uuid.uuid4,
+                        editable=False,
+                        primary_key=True,
+                        serialize=False,
+                    ),
+                ),
+                ("inserted_at", models.DateTimeField(auto_now_add=True)),
+                (
+                    "declared_cloud_accounts",
+                    models.CharField(
+                        blank=True,
+                        choices=[
+                            ("1", "1"),
+                            ("2-10", "2-10"),
+                            ("11-50", "11-50"),
+                            ("51-200", "51-200"),
+                            ("200+", "200+"),
+                        ],
+                        max_length=16,
+                        null=True,
+                    ),
+                ),
+                (
+                    "declared_role",
+                    models.CharField(
+                        blank=True,
+                        choices=[
+                            ("security", "Security"),
+                            ("devops_platform", "DevOps / Platform"),
+                            ("developer", "Developer"),
+                            ("compliance_grc", "Compliance / GRC"),
+                            ("other", "Other"),
+                        ],
+                        max_length=32,
+                        null=True,
+                    ),
+                ),
+                (
+                    "declared_seniority",
+                    models.CharField(
+                        blank=True,
+                        choices=[
+                            ("practitioner", "Practitioner / IC"),
+                            ("lead", "Team lead / Manager"),
+                            ("director", "Director / Head of"),
+                            ("executive", "VP / C-level"),
+                            ("founder", "Founder / Owner"),
+                        ],
+                        max_length=32,
+                        null=True,
+                    ),
+                ),
+                ("skipped", models.BooleanField(default=False)),
+                (
+                    "submitted_by",
+                    models.ForeignKey(
+                        blank=True,
+                        null=True,
+                        on_delete=django.db.models.deletion.SET_NULL,
+                        related_name="tenant_onboarding_profiles",
+                        related_query_name="tenant_onboarding_profile",
+                        to=settings.AUTH_USER_MODEL,
+                    ),
+                ),
+                (
+                    "tenant",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE, to="api.tenant"
+                    ),
+                ),
+            ],
+            options={
+                "db_table": "tenant_onboarding_profiles",
+                "abstract": False,
+            },
+        ),
+        migrations.AddConstraint(
+            model_name="tenantonboardingprofile",
+            constraint=models.UniqueConstraint(
+                fields=("tenant_id",), name="unique_tenant_onboarding_profile"
+            ),
+        ),
+        migrations.AddConstraint(
+            model_name="tenantonboardingprofile",
+            # `statements` written out explicitly: RowLevelSecurityConstraint
+            # .deconstruct() does not serialize it, so an autogenerated
+            # migration falls back to ["SELECT"] and leaves the table without
+            # INSERT/UPDATE/DELETE policies.
+            constraint=api.rls.RowLevelSecurityConstraint(
+                "tenant_id",
+                name="rls_on_tenantonboardingprofile",
+                statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
+            ),
+        ),
+    ]
@@ -0,0 +1,15 @@
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+    # The onboarding profile step was reverted after 0098 had been merged, so
+    # the table goes away through a new migration rather than by deleting 0098.
+    dependencies = [
+        ("api", "0098_tenant_onboarding_profile"),
+    ]
+
+    operations = [
+        migrations.DeleteModel(
+            name="TenantOnboardingProfile",
+        ),
+    ]
@@ -0,0 +1,48 @@
+from django.db import migrations
+
+TASK_NAME = "attack-paths-reap-orphaned-tmp-databases"
+INTERVAL_HOURS = 6
+
+
+def create_periodic_task(apps, schema_editor):
+    IntervalSchedule = apps.get_model("django_celery_beat", "IntervalSchedule")
+    PeriodicTask = apps.get_model("django_celery_beat", "PeriodicTask")
+
+    schedule, _ = IntervalSchedule.objects.get_or_create(
+        every=INTERVAL_HOURS,
+        period="hours",
+    )
+
+    PeriodicTask.objects.update_or_create(
+        name=TASK_NAME,
+        defaults={
+            "task": TASK_NAME,
+            "interval": schedule,
+            "enabled": True,
+        },
+    )
+
+
+def delete_periodic_task(apps, schema_editor):
+    IntervalSchedule = apps.get_model("django_celery_beat", "IntervalSchedule")
+    PeriodicTask = apps.get_model("django_celery_beat", "PeriodicTask")
+
+    PeriodicTask.objects.filter(name=TASK_NAME).delete()
+
+    # Clean up the schedule if no other task references it
+    IntervalSchedule.objects.filter(
+        every=INTERVAL_HOURS,
+        period="hours",
+        periodictask__isnull=True,
+    ).delete()
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("api", "0099_delete_tenant_onboarding_profile"),
+        ("django_celery_beat", "0019_alter_periodictasks_options"),
+    ]
+
+    operations = [
+        migrations.RunPython(create_periodic_task, delete_periodic_task),
+    ]
@@ -617,9 +617,66 @@ class Task(RowLevelSecurityProtectedModel):
        resource_name = "tasks"


+class ScanQuerySet(models.QuerySet):
+    """Shared selectors for "the latest scan of a provider".
+
+    The queryset must already be scoped by the caller: manager, tenant, RBAC,
+    providers and database alias.
+    """
+
+    # How "which completed scan is the provider's current one" is ordered.
+    LATEST_ORDER_BY = (
+        models.F("completed_at").desc(nulls_last=True),
+        models.F("inserted_at").desc(),
+        models.F("id").desc(),
+    )
+
+    def _eligible_for_latest(self) -> "ScanQuerySet":
+        """Restrict to the scans that may be a provider's latest.
+
+        Returns:
+            ScanQuerySet: The completed scans.
+        """
+        return self.filter(state=StateChoices.COMPLETED)
+
+    def latest_per_provider(self) -> "ScanQuerySet":
+        """Pick each provider's latest scan with `DISTINCT ON (provider_id)`.
+
+        Returns:
+            ScanQuerySet: One scan per provider, the latest one.
+        """
+        return (
+            self._eligible_for_latest()
+            .order_by("provider_id", *self.LATEST_ORDER_BY)
+            .distinct("provider_id")
+        )
+
+    def latest_ids_per_provider(self) -> list[UUID]:
+        """Evaluate `latest_per_provider` and return the scan ids.
+
+        The ids are materialised so callers can pass them as a literal `IN`
+        list; as a subquery Postgres misestimates the row count and picks a
+        slow nested loop.
+
+        Returns:
+            list[UUID]: The id of each provider's latest scan.
+        """
+        return list(self.latest_per_provider().values_list("id", flat=True))
+
+    def latest_first(self) -> "ScanQuerySet":
+        """Order eligible scans newest first, without deduplicating per provider.
+
+        Expects the queryset to be already filtered to a single provider.
+
+        Returns:
+            ScanQuerySet: The eligible scans, latest first.
+        """
+        return self._eligible_for_latest().order_by(*self.LATEST_ORDER_BY)
+
+
 class Scan(RowLevelSecurityProtectedModel):
-    objects = ActiveProviderManager()
-    all_objects = models.Manager()
+    objects = ActiveProviderManager.from_queryset(ScanQuerySet)()
+    all_objects = ScanQuerySet.as_manager()

    _SCOPING_SCANNER_ARG_KEYS_CACHE: tuple[str, ...] | None = None

@@ -726,6 +783,12 @@ class Scan(RowLevelSecurityProtectedModel):
                name="scans_prov_state_ins_desc_idx",
            ),
            # TODO This might replace `scans_prov_state_ins_desc_idx` completely. Review usage
+            # Since `ScanQuerySet`, no code path reads a provider's
+            # completed scans by `-inserted_at`. The only query left that
+            # matches this index (and `scans_prov_state_ins_desc_idx` above)
+            # is `GET /scans?filter[provider]=…&filter[state]=completed` with
+            # the default sort. Both are candidates to drop in a follow-up
+            # once production `pg_stat_user_indexes.idx_scan` confirms it.
            models.Index(
                fields=["tenant_id", "provider_id", "-inserted_at"],
                condition=Q(state=StateChoices.COMPLETED),
@@ -1380,6 +1443,15 @@ class Invitation(RowLevelSecurityProtectedModel):
            self.email = self.email.strip().lower()
        super().save(*args, **kwargs)

+    @classmethod
+    def lapsed_q(cls):
+        """Pending invitations whose expiry date has already passed."""
+        return Q(state=cls.State.PENDING, expires_at__lte=datetime.now(UTC))
+
+    @property
+    def is_lapsed(self):
+        return self.state == self.State.PENDING and self.expires_at <= datetime.now(UTC)
+
    class Meta(RowLevelSecurityProtectedModel.Meta):
        db_table = "invitations"

@@ -6,6 +6,7 @@ from api.rls import RowLevelSecurityConstraint
 from api.uuid_utils import datetime_to_uuid7
 from dateutil.relativedelta import relativedelta
 from django.conf import settings
+from django.core.exceptions import ImproperlyConfigured
 from psqlextra.partitioning import (
    PostgresPartitioningError,
    PostgresPartitioningManager,
@@ -153,10 +154,17 @@ class PostgresUUIDv7PartitioningStrategy(PostgresRangePartitioningStrategy):
        )


-def relative_days_or_none(value):
-    if value is None:
+def relative_months_or_none(value):
+    # A negative value would set the cutoff in the future and delete every
+    # partition, so it is rejected rather than silently ignored.
+    if value is not None and value < 0:
+        raise ImproperlyConfigured(
+            "FINDINGS_TABLE_PARTITION_MAX_AGE_MONTHS must not be negative; "
+            "leave it unset or use 0 to keep partitions indefinitely"
+        )
+    if not value:
        return None
-    return relativedelta(days=value)
+    return relativedelta(months=value)


 #
@@ -173,7 +181,7 @@ manager = PostgresPartitioningManager(
                    months=settings.FINDINGS_TABLE_PARTITION_MONTHS
                ),
                count=settings.FINDINGS_TABLE_PARTITION_COUNT,
-                max_age=relative_days_or_none(
+                max_age=relative_months_or_none(
                    settings.FINDINGS_TABLE_PARTITION_MAX_AGE_MONTHS
                ),
                name_format="%Y_%b",
@@ -189,7 +197,7 @@ manager = PostgresPartitioningManager(
                    months=settings.FINDINGS_TABLE_PARTITION_MONTHS
                ),
                count=settings.FINDINGS_TABLE_PARTITION_COUNT,
-                max_age=relative_days_or_none(
+                max_age=relative_months_or_none(
                    settings.FINDINGS_TABLE_PARTITION_MAX_AGE_MONTHS
                ),
                name_format="%Y_%b",
@@ -1,7 +1,7 @@
 from enum import Enum

 from api.db_router import MainRouter
-from api.models import Integration, Provider, Role, User
+from api.models import Integration, Provider, Role, Task, User
 from django.db.models import Q, QuerySet
 from rest_framework.exceptions import PermissionDenied
 from rest_framework.permissions import BasePermission
@@ -17,6 +17,50 @@ class Permissions(Enum):
    UNLIMITED_VISIBILITY = "unlimited_visibility"


+# Revoking a task needs the permission of the operation that queued it.
+# None and unmapped names are not revocable; a revoked provider deletion
+# would leave the provider soft-deleted with nothing re-queuing the cleanup.
+TASK_REVOKE_PERMISSIONS: dict[str, list[Permissions] | None] = {
+    "provider-connection-check": [Permissions.MANAGE_PROVIDERS],
+    "provider-deletion": None,
+    "integration-connection-check": [Permissions.MANAGE_INTEGRATIONS],
+    "integration-s3": [Permissions.MANAGE_INTEGRATIONS],
+    "integration-security-hub": [Permissions.MANAGE_INTEGRATIONS],
+    "integration-jira": [Permissions.MANAGE_INTEGRATIONS],
+    "scan-perform": [Permissions.MANAGE_SCANS],
+    "scan-perform-scheduled": [Permissions.MANAGE_SCANS],
+    "scan-compliance-overviews": [Permissions.MANAGE_SCANS],
+    "scan-compliance-reports": [Permissions.MANAGE_SCANS],
+    "scan-finding-group-summaries": [Permissions.MANAGE_SCANS],
+    "scan-report": [Permissions.MANAGE_SCANS],
+    "attack-paths-scan-perform": [Permissions.MANAGE_SCANS],
+    "findings-mute-latest-scans": [Permissions.MANAGE_SCANS],
+    "lighthouse-connection-check": [],
+    "lighthouse-provider-connection-check": [],
+    "lighthouse-provider-models-refresh": [],
+}
+
+
+def get_user_roles(user: User, tenant_id: str) -> list[Role]:
+    """Return every role assigned to the user in the tenant."""
+    return list(
+        User.objects.using(MainRouter.admin_db)
+        .get(id=user.id)
+        .roles.using(MainRouter.admin_db)
+        .filter(tenant_id=tenant_id)
+    )
+
+
+def roles_have_permissions(
+    roles: list[Role], required_permissions: list[Permissions]
+) -> bool:
+    """Return True when every required permission is granted by at least one role."""
+    return all(
+        any(getattr(role, permission.value, False) for role in roles)
+        for permission in required_permissions
+    )
+
+
 class HasPermissions(BasePermission):
    """
    Custom permission to check if the user's role has the required permissions.
@@ -34,19 +78,11 @@ class HasPermissions(BasePermission):
        if not tenant_id:
            return False

-        user_roles = list(
-            User.objects.using(MainRouter.admin_db)
-            .get(id=request.user.id)
-            .roles.using(MainRouter.admin_db)
-            .filter(tenant_id=tenant_id)
-        )
+        user_roles = get_user_roles(request.user, tenant_id)
        if not user_roles:
            return False

-        return all(
-            any(getattr(role, permission.value, False) for role in user_roles)
-            for permission in required_permissions
-        )
+        return roles_have_permissions(user_roles, required_permissions)


 def get_role(user: User, tenant_id: str) -> Role:
@@ -85,6 +121,25 @@ def get_providers(role: Role) -> QuerySet[Provider]:
    ).distinct()


+def get_tasks(role: Role) -> QuerySet[Task]:
+    """Return the tasks visible to the role: tenant-wide ones and those of its providers."""
+    queryset = Task.objects.filter(tenant_id=role.tenant_id)
+    if role.unlimited_visibility:
+        return queryset
+
+    # Task has no provider FK, so match provider ids inside the stored kwargs.
+    # all_objects keeps a soft-deleted provider visible to its own groups, so the
+    # role that queued its deletion can still follow the task.
+    hidden = Q()
+    for provider_id in (
+        Provider.all_objects.filter(tenant_id=role.tenant_id)
+        .exclude(provider_groups__in=role.provider_groups.all())
+        .values_list("id", flat=True)
+    ):
+        hidden |= Q(task_runner_task__task_kwargs__contains=str(provider_id))
+    return queryset.exclude(hidden) if hidden else queryset
+
+
 def get_integrations(
    role: Role, providers: QuerySet[Provider] | None = None
 ) -> QuerySet[Integration]:
@@ -1,7 +1,7 @@
 openapi: 3.0.3
 info:
  title: Prowler API
-  version: 1.40.0
+  version: 1.48.1
  description: |-
    Prowler API specification.

@@ -14823,7 +14823,9 @@ paths:
    get:
      operationId: api_v1_tasks_list
      description: Retrieve a list of all tasks with options for filtering by name,
-        state, and other criteria.
+        state, and other criteria. Tasks that reference a provider are only returned
+        when the role can access it; tasks without a provider reference are returned
+        for every role.
      summary: List all tasks
      parameters:
      - in: query
@@ -14922,7 +14924,8 @@ paths:
  /api/v1/tasks/{id}:
    get:
      operationId: api_v1_tasks_retrieve
-      description: Fetch detailed information about a specific task by its ID.
+      description: Fetch detailed information about a specific task by its ID. Tasks
+        tied to a provider outside the visibility of the role are not found.
      summary: Retrieve data from a specific task
      parameters:
      - in: query
@@ -14963,7 +14966,9 @@ paths:
    delete:
      operationId: api_v1_tasks_destroy
      description: Try to revoke a task using its ID. Only tasks that are not yet
-        in progress can be revoked.
+        in progress can be revoked, and the caller needs the same permission as the
+        operation that queued the task (for example MANAGE_SCANS for a scan). Provider
+        deletions cannot be revoked.
      summary: Revoke a task
      parameters:
      - in: path
@@ -1,9 +1,9 @@
 import json
-import time
 from datetime import UTC, datetime, timedelta
 from uuid import uuid4

 import pytest
+from api.authentication import API_KEY_LAST_USED_AT_THROTTLE_SECONDS
 from api.db_router import MainRouter
 from api.models import Membership, Role, TenantAPIKey, User, UserRoleRelationship
 from api.signals import revoke_membership_api_keys, revoke_user_api_keys
@@ -11,6 +11,7 @@ from conftest import TEST_PASSWORD, get_api_tokens, get_authorization_header
 from django.db.utils import ConnectionDoesNotExist
 from django.urls import reverse
 from drf_simple_apikey.crypto import get_crypto
+from freezegun import freeze_time
 from rest_framework.test import APIClient
 from rest_framework_simplejwt.token_blacklist.models import (
    BlacklistedToken,
@@ -527,7 +528,7 @@ class TestAPIKeyAuthentication:
    def test_last_used_at_tracking(
        self, create_test_user, tenants_fixture, api_keys_fixture
    ):
-        """Verify last_used_at timestamp updates on each authentication."""
+        """Verify last_used_at timestamp is set on first use and throttled after that."""
        client = APIClient()
        api_key = api_keys_fixture[0]

@@ -536,7 +537,11 @@ class TestAPIKeyAuthentication:

        # Use API key to authenticate
        api_key_headers = get_api_key_header(api_key._raw_key)
-        first_response = client.get(reverse("provider-list"), headers=api_key_headers)
+        start = datetime.now(UTC)
+        with freeze_time(start):
+            first_response = client.get(
+                reverse("provider-list"), headers=api_key_headers
+            )
        assert first_response.status_code == 200

        # Reload from database and check last_used_at is set
@@ -544,17 +549,23 @@ class TestAPIKeyAuthentication:
        first_used_at = api_key.last_used_at
        assert first_used_at is not None

-        # Use the same key again after a small delay
-        time.sleep(0.1)
-
+        # Using the same key again within the throttle interval does not rewrite it
        second_response = client.get(reverse("provider-list"), headers=api_key_headers)
        assert second_response.status_code == 200

-        # Reload and verify last_used_at was updated
        api_key.refresh_from_db()
-        second_used_at = api_key.last_used_at
-        assert second_used_at is not None
-        assert second_used_at > first_used_at
+        assert api_key.last_used_at == first_used_at
+
+        # Past the throttle interval, the next use refreshes it
+        later = start + timedelta(seconds=API_KEY_LAST_USED_AT_THROTTLE_SECONDS + 1)
+        with freeze_time(later):
+            third_response = client.get(
+                reverse("provider-list"), headers=api_key_headers
+            )
+        assert third_response.status_code == 200
+
+        api_key.refresh_from_db()
+        assert api_key.last_used_at > first_used_at


@pytest.mark.django_db
@@ -1441,6 +1452,7 @@ class TestAPIKeyRLSBypass:

        The update to last_used_at during authentication must also use the
        admin database since it occurs before RLS context is established.
+        Past the throttle interval, using the key again refreshes the timestamp.
        """
        client = APIClient()
        api_key = api_keys_fixture[0]
@@ -1448,7 +1460,11 @@ class TestAPIKeyRLSBypass:
        assert api_key.last_used_at is None

        api_key_headers = get_api_key_header(api_key._raw_key)
-        first_response = client.get(reverse("provider-list"), headers=api_key_headers)
+        start = datetime.now(UTC)
+        with freeze_time(start):
+            first_response = client.get(
+                reverse("provider-list"), headers=api_key_headers
+            )

        assert first_response.status_code == 200

@@ -1456,9 +1472,11 @@ class TestAPIKeyRLSBypass:
        first_timestamp = api_key.last_used_at
        assert first_timestamp is not None

-        time.sleep(0.1)
-
-        second_response = client.get(reverse("provider-list"), headers=api_key_headers)
+        later = start + timedelta(seconds=API_KEY_LAST_USED_AT_THROTTLE_SECONDS + 1)
+        with freeze_time(later):
+            second_response = client.get(
+                reverse("provider-list"), headers=api_key_headers
+            )
        assert second_response.status_code == 200

        api_key.refresh_from_db()
@@ -111,6 +111,110 @@ def _verify_local_email(user):
    )


+def test_social_account_name_falls_back_to_login_for_blank_name():
+    adapter = ProwlerSocialAccountAdapter()
+
+    name = adapter._get_social_account_name(
+        {"name": "   ", "login": "octocat"},
+        "verified@example.com",
+    )
+
+    assert name == "octocat"
+
+
+@pytest.mark.parametrize("provider_name", [None, "", "   ", 123, ["name"]])
+def test_social_account_name_ignores_unusable_provider_names(provider_name):
+    adapter = ProwlerSocialAccountAdapter()
+
+    name = adapter._get_social_account_name(
+        {"name": provider_name, "login": "octocat"},
+        "verified@example.com",
+    )
+
+    assert name == "octocat"
+
+
+def test_social_account_name_uses_login_when_name_is_missing():
+    adapter = ProwlerSocialAccountAdapter()
+
+    name = adapter._get_social_account_name(
+        {"login": "octocat"},
+        "verified@example.com",
+    )
+
+    assert name == "octocat"
+
+
+def test_social_account_name_falls_back_to_username_then_email():
+    adapter = ProwlerSocialAccountAdapter()
+
+    username_name = adapter._get_social_account_name(
+        {"name": "ab", "login": None, "username": "  monalisa  "},
+        "verified@example.com",
+    )
+    email_name = adapter._get_social_account_name({}, "  verified@example.com  ")
+
+    assert username_name == "monalisa"
+    assert email_name == "verified@example.com"
+
+
+def test_social_account_name_trims_and_limits_provider_name():
+    adapter = ProwlerSocialAccountAdapter()
+    max_length = User._meta.get_field("name").max_length
+
+    trimmed_name = adapter._get_social_account_name(
+        {"name": "  Ada Lovelace  "},
+        "verified@example.com",
+    )
+    limited_name = adapter._get_social_account_name(
+        {"name": "a" * (max_length + 1)},
+        "verified@example.com",
+    )
+
+    assert trimmed_name == "Ada Lovelace"
+    assert limited_name == "a" * max_length
+
+
+def test_social_account_name_rejects_missing_identity():
+    adapter = ProwlerSocialAccountAdapter()
+
+    with pytest.raises(
+        ValueError,
+        match="Social account does not provide a valid user identity",
+    ):
+        adapter._get_social_account_name({}, "")
+
+
+def test_save_user_applies_normalized_social_account_name(rf):
+    adapter = ProwlerSocialAccountAdapter()
+    request = rf.post("/")
+    request.session = {}
+    sociallogin = MagicMock(spec=SocialLogin)
+    sociallogin.provider = MagicMock()
+    sociallogin.provider.id = "github"
+    sociallogin.account = MagicMock()
+    sociallogin.account.extra_data = {"name": None, "login": "  octocat  "}
+    user = User(email="verified@example.com")
+    user.save = MagicMock()
+    invitation = SimpleNamespace(tenant_id="tenant-id")
+
+    with (
+        patch("api.adapters.super") as mock_super,
+        patch("api.adapters.transaction.atomic"),
+        patch("api.adapters.write_db_alias"),
+        patch.object(adapter, "_get_invitation_token", return_value="token"),
+        patch(
+            "api.adapters.accept_invitation_for_user",
+            return_value=(invitation, True),
+        ),
+    ):
+        mock_super.return_value.save_user.return_value = user
+        saved_user = adapter.save_user(request, sociallogin)
+
+    assert saved_user.name == "octocat"
+    assert request.prowler_invitation_token == "token"
+
+
@pytest.mark.django_db
 class TestProwlerSocialAccountAdapter:
    def test_get_user_by_email_returns_user(self, create_test_user):
@@ -321,7 +321,7 @@ def test_serialize_graph_filters_by_provider_label(attack_paths_graph_stub_class
        relationships=[rel_keep, rel_drop_orphaned],
    )

-    result = views_helpers._serialize_graph(graph, provider_id)
+    result = views_helpers._serialize_graph(graph, provider_id, _all_persisted())

    assert len(result["nodes"]) == 1
    assert result["nodes"][0]["id"] == "n1"
@@ -329,6 +329,71 @@ def test_serialize_graph_filters_by_provider_label(attack_paths_graph_stub_class
    assert result["relationships"][0]["id"] == "r1"


+def _all_persisted():
+    return SimpleNamespace(is_persisted_element_id=lambda element_id: True)
+
+
+def test_serialize_graph_drops_virtual_entities(attack_paths_graph_stub_classes):
+    provider_id = "provider-keep"
+    plabel = get_provider_label(provider_id)
+    stored_ids = {"n1", "n2", "r1"}
+    backend = SimpleNamespace(
+        is_persisted_element_id=lambda element_id: element_id in stored_ids
+    )
+
+    node_1 = attack_paths_graph_stub_classes.Node("n1", ["AWSAccount", plabel], {})
+    node_2 = attack_paths_graph_stub_classes.Node("n2", ["AWSRole", plabel], {})
+    # apoc.create.vNode can carry the provider label and any payload
+    node_virtual = attack_paths_graph_stub_classes.Node(
+        "-1", ["AWSAccount", plabel], {"body": "fetched"}
+    )
+
+    rel_stored = attack_paths_graph_stub_classes.Relationship(
+        "r1", "OWNS", node_1, node_2, {}
+    )
+    # apoc.create.vRelationship between two stored nodes passes the endpoint check
+    rel_virtual = attack_paths_graph_stub_classes.Relationship(
+        "45cc10ea-9618-47fb-abaa-8e82a58d6b40", "V", node_1, node_2, {"body": "fetched"}
+    )
+    rel_to_virtual = attack_paths_graph_stub_classes.Relationship(
+        "r2", "OWNS", node_1, node_virtual, {}
+    )
+
+    graph = SimpleNamespace(
+        nodes=[node_1, node_2, node_virtual],
+        relationships=[rel_stored, rel_virtual, rel_to_virtual],
+    )
+
+    result = views_helpers._serialize_graph(graph, provider_id, backend)
+
+    assert [node["id"] for node in result["nodes"]] == ["n1", "n2"]
+    assert [rel["id"] for rel in result["relationships"]] == ["r1"]
+    assert "fetched" not in str(result)
+
+
+@pytest.mark.parametrize(
+    "element_id, persisted",
+    [
+        ("4:9f1d9d76-bb08-478c-b25d-7e577f967732:0", True),
+        ("5:9f1d9d76-bb08-478c-b25d-7e577f967732:12", True),
+        ("-2", False),
+        ("45cc10ea-9618-47fb-abaa-8e82a58d6b40", False),
+        ("", False),
+    ],
+    ids=[
+        "stored_node",
+        "stored_relationship",
+        "virtual_node",
+        "virtual_relationship",
+        "empty",
+    ],
+)
+def test_neo4j_sink_is_persisted_element_id(element_id, persisted):
+    from api.attack_paths.sink.neo4j import Neo4jSink
+
+    assert Neo4jSink().is_persisted_element_id(element_id) is persisted
+
+
 # -- serialize_graph_as_text -------------------------------------------------------


@@ -187,6 +187,27 @@ class TestRoutingByDatabasePrefix:
        sink_backend_stub.drop_database.assert_called_once_with("db-tenant-abc")
        mock_ingest.drop_database.assert_not_called()

+    def test_list_databases_always_routes_to_ingest(self, sink_backend_stub):
+        with patch("api.attack_paths.database.ingest") as mock_ingest:
+            mock_ingest.list_databases.return_value = ["db-tmp-scan-uuid-1"]
+
+            assert db_module.list_databases() == ["db-tmp-scan-uuid-1"]
+
+        mock_ingest.list_databases.assert_called_once_with()
+
+    def test_ingest_list_databases_dedupes_cluster_rows(self):
+        from api.attack_paths.ingest import driver as ingest_driver
+
+        with patch.object(ingest_driver, "get_session") as mock_get_session:
+            session = mock_get_session.return_value.__enter__.return_value
+            session.run.return_value = [{"name": "db-tmp-scan-uuid-1"}]
+
+            assert ingest_driver.list_databases() == ["db-tmp-scan-uuid-1"]
+
+        session.run.assert_called_once_with(
+            "SHOW DATABASES YIELD name RETURN DISTINCT name"
+        )
+
    def test_clear_cache_routes_temp_to_ingest(self, sink_backend_stub):
        with patch("api.attack_paths.database.ingest") as mock_ingest:
            db_module.clear_cache("db-tmp-scan-uuid-1")
@@ -5,16 +5,18 @@ from uuid import uuid4

 import pytest
 from api.authentication import (
+    API_KEY_LAST_USED_AT_THROTTLE_SECONDS,
    OrphanedAPIKeyError,
    SSEAuthentication,
    TenantAPIKeyAuthentication,
 )
 from api.db_router import MainRouter
-from api.models import TenantAPIKey
+from api.models import TenantAPIKey, User
 from django.db import connections
 from django.db.models.query import QuerySet
 from django.test import RequestFactory
 from django.test.utils import CaptureQueriesContext
+from freezegun import freeze_time
 from rest_framework.exceptions import AuthenticationFailed


@@ -286,14 +288,15 @@ class TestTenantAPIKeyAuthentication:

        assert str(exc_info.value.detail) == "This API Key has been revoked."

-    def test_authenticate_reads_the_api_key_once_under_a_row_lock(
+    def test_authenticate_reads_the_api_key_once_without_a_row_lock(
        self, auth_backend, api_keys_fixture, request_factory
    ):
-        """Test the API key is read a single time and the row is locked.
+        """Test the API key is read a single time and no row is locked.

        Validation, the `last_used_at` update and the claims must all come from the
-        same authoritative row: a second, unlocked lookup would reopen the window
-        where a key revoked in between still authenticates.
+        same authoritative row: a second lookup would reopen the window where a key
+        revoked in between still authenticates. `SELECT ... FOR UPDATE` serialized
+        every request for a hot key onto one locked row and is not used any more.
        """
        api_key = api_keys_fixture[0]

@@ -310,33 +313,40 @@ class TestTenantAPIKeyAuthentication:
        ]

        assert len(api_key_selects) == 1
-        assert "FOR UPDATE" in api_key_selects[0]
+        assert "FOR UPDATE" not in api_key_selects[0]

-    def test_authenticate_ignores_revocation_after_the_locked_read(
+    def test_authenticate_ignores_revocation_after_the_single_read(
        self, auth_backend, api_keys_fixture, request_factory
    ):
        """Test the claims describe the row that was validated, not a later state.

        Regression test: the key used to be looked up again to build the auth dict,
        without rechecking `revoked` or `entity`. A key revoked or orphaned between
-        both reads still authenticated, and the claims came from that stale row. With
-        a single locked read the write below cannot land mid-authentication, and the
-        revocation only takes effect on the next request.
+        both reads still authenticated, and the claims came from that stale row.
+        There is now only a single read, so this race is closed by construction and
+        the revocation only takes effect on the next request.
        """
        api_key = api_keys_fixture[0]
        entity_at_validation = api_key.entity
-        original_save = TenantAPIKey.save
+        original_authenticate_credentials = (
+            TenantAPIKeyAuthentication._authenticate_credentials
+        )

-        def revoke_and_orphan_before_saving(instance, *args, **kwargs):
-            # Runs after validation, right before the claims are built: the exact
-            # window a concurrent revocation or user deletion used to slip into
+        def revoke_and_orphan_after_reading(self, request, key):
+            # Runs right after the single read `authenticate` will use to build the
+            # claims: the exact window a concurrent revocation used to slip into
+            result = original_authenticate_credentials(self, request, key)
            TenantAPIKey.objects.filter(id=api_key.id).update(revoked=True, entity=None)
-            return original_save(instance, *args, **kwargs)
+            return result

        request = request_factory.get("/")
        request.META["HTTP_AUTHORIZATION"] = f"Api-Key {api_key._raw_key}"

-        with patch.object(TenantAPIKey, "save", revoke_and_orphan_before_saving):
+        with patch.object(
+            TenantAPIKeyAuthentication,
+            "_authenticate_credentials",
+            revoke_and_orphan_after_reading,
+        ):
            entity, auth_dict = auth_backend.authenticate(request)

        assert entity == entity_at_validation
@@ -350,6 +360,43 @@ class TestTenantAPIKeyAuthentication:

        assert str(exc_info.value.detail) == "This API Key has been revoked."

+    def test_authenticate_survives_owner_deleted_after_the_single_read(
+        self, auth_backend, api_keys_fixture, request_factory
+    ):
+        """Test a user deleted right after the read does not turn into a 500.
+
+        Without the row lock a concurrent user deletion can land between the read
+        and building the claims. `entity` is loaded by the same query, so no later
+        lookup can raise `DoesNotExist`.
+        """
+        api_key = api_keys_fixture[0]
+        owner_id = api_key.entity_id
+        original_authenticate_credentials = (
+            TenantAPIKeyAuthentication._authenticate_credentials
+        )
+
+        def delete_owner_after_reading(self, request, key):
+            result = original_authenticate_credentials(self, request, key)
+            User.objects.using(MainRouter.admin_db).filter(id=owner_id).delete()
+            return result
+
+        request = request_factory.get("/")
+        request.META["HTTP_AUTHORIZATION"] = f"Api-Key {api_key._raw_key}"
+
+        with patch.object(
+            TenantAPIKeyAuthentication,
+            "_authenticate_credentials",
+            delete_owner_after_reading,
+        ):
+            entity, auth_dict = auth_backend.authenticate(request)
+
+        assert auth_dict["sub"] == str(owner_id)
+        assert entity.id == owner_id
+
+        # From the next request on, the orphaned key is rejected with a 401
+        with pytest.raises(AuthenticationFailed):
+            auth_backend.authenticate(request)
+
    def test_authenticate_expired_api_key(
        self, auth_backend, create_test_user, tenants_fixture, request_factory
    ):
@@ -421,24 +468,90 @@ class TestTenantAPIKeyAuthentication:
        if original_last_used:
            assert api_key.last_used_at > original_last_used

-    def test_authenticate_saves_to_admin_database(
+    def test_authenticate_updates_last_used_at_on_admin_database(
        self, auth_backend, api_keys_fixture, request_factory
    ):
-        """Test that the API key save operation uses admin database."""
+        """Test that the `last_used_at` update runs against the admin database."""
        api_key = api_keys_fixture[0]
        raw_key = api_key._raw_key

        request = request_factory.get("/")
        request.META["HTTP_AUTHORIZATION"] = f"Api-Key {raw_key}"

-        # Mock the save method to verify it's called with using='admin'
-        with patch.object(TenantAPIKey, "save") as mock_save:
+        with CaptureQueriesContext(connections[MainRouter.admin_db]) as captured:
            auth_backend.authenticate(request)

-            # Verify save was called with using=admin_db
-            mock_save.assert_called_once_with(
-                update_fields=["last_used_at"], using=MainRouter.admin_db
-            )
+        api_key_updates = [
+            query["sql"]
+            for query in captured.captured_queries
+            if query["sql"].startswith("UPDATE") and '"api_keys"' in query["sql"]
+        ]
+
+        assert len(api_key_updates) == 1
+        assert "last_used_at" in api_key_updates[0]
+
+    def test_authenticate_does_not_rewrite_last_used_at_within_throttle_interval(
+        self, auth_backend, api_keys_fixture, request_factory
+    ):
+        """Test that a second authentication within the throttle interval is a no-op write."""
+        api_key = api_keys_fixture[0]
+        raw_key = api_key._raw_key
+
+        request = request_factory.get("/")
+        request.META["HTTP_AUTHORIZATION"] = f"Api-Key {raw_key}"
+
+        # First call sets last_used_at
+        auth_backend.authenticate(request)
+        api_key.refresh_from_db()
+        first_used_at = api_key.last_used_at
+        assert first_used_at is not None
+
+        # Second call, still within the throttle interval, must issue no UPDATE
+        with CaptureQueriesContext(connections[MainRouter.admin_db]) as captured:
+            auth_backend.authenticate(request)
+
+        api_key_updates = [
+            query["sql"]
+            for query in captured.captured_queries
+            if query["sql"].startswith("UPDATE") and '"api_keys"' in query["sql"]
+        ]
+        assert api_key_updates == []
+
+        api_key.refresh_from_db()
+        assert api_key.last_used_at == first_used_at
+
+    def test_authenticate_rewrites_last_used_at_after_throttle_interval(
+        self, auth_backend, api_keys_fixture, request_factory
+    ):
+        """Test that `last_used_at` is refreshed once it is older than the throttle interval."""
+        api_key = api_keys_fixture[0]
+        raw_key = api_key._raw_key
+
+        request = request_factory.get("/")
+        request.META["HTTP_AUTHORIZATION"] = f"Api-Key {raw_key}"
+
+        start = datetime.now(UTC)
+        with freeze_time(start):
+            auth_backend.authenticate(request)
+
+        api_key.refresh_from_db()
+        first_used_at = api_key.last_used_at
+        assert first_used_at is not None
+
+        later = start + timedelta(seconds=API_KEY_LAST_USED_AT_THROTTLE_SECONDS + 1)
+        with freeze_time(later):
+            with CaptureQueriesContext(connections[MainRouter.admin_db]) as captured:
+                auth_backend.authenticate(request)
+
+        api_key_updates = [
+            query["sql"]
+            for query in captured.captured_queries
+            if query["sql"].startswith("UPDATE") and '"api_keys"' in query["sql"]
+        ]
+        assert len(api_key_updates) == 1
+
+        api_key.refresh_from_db()
+        assert api_key.last_used_at > first_used_at

    def test_authenticate_returns_correct_auth_dict(
        self, auth_backend, api_keys_fixture, request_factory
@@ -222,6 +222,12 @@ class TestProtection:
        result = _inject(cypher)
        assert result.count(LABEL) == 1

+    def test_block_comment_with_node_pattern(self):
+        cypher = "MATCH (n:AWSRole) /* (m:Fake)\n(k) */ RETURN n"
+        result = _inject(cypher)
+        assert "/* (m:Fake)\n(k) */" in result
+        assert result.count(LABEL) == 1
+

 # ---------------------------------------------------------------------------
 # Clause splitting
@@ -439,6 +445,35 @@ class TestValidation:

        assert "blocked operation" in str(exc.value.detail)

+    @pytest.mark.parametrize(
+        "cypher",
+        [
+            "LOAD /* c */ CSV FROM 'http://169.254.169.254/' AS x RETURN x",
+            "LOAD/**/CSV FROM 'http://evil.com' AS x RETURN x",
+            "LOAD /* a */ /* b */ CSV FROM 'http://evil.com' AS x RETURN x",
+            "LOAD /* multi\nline */ CSV FROM 'http://evil.com' AS x RETURN x",
+            "CALL apoc./**/load.json('http://evil.com/') YIELD value RETURN value",
+            "CALL apoc . load.json('http://evil.com/') YIELD value RETURN value",
+            "CALL apoc.load . json('http://evil.com/') YIELD value RETURN value",
+            "CALL apoc\n.cypher.run('CREATE (n)', {}) YIELD value RETURN value",
+        ],
+        ids=[
+            "LOAD_CSV_block_comment",
+            "LOAD_CSV_empty_block_comment",
+            "LOAD_CSV_two_block_comments",
+            "LOAD_CSV_multiline_block_comment",
+            "apoc.load_block_comment",
+            "apoc.load_spaces_around_dot",
+            "apoc.load_spaces_before_procedure",
+            "apoc.cypher_newline_before_dot",
+        ],
+    )
+    def test_rejects_blocked_patterns_split_by_comments_or_whitespace(self, cypher):
+        with pytest.raises(ValidationError) as exc:
+            validate_custom_query(cypher)
+
+        assert "blocked operation" in str(exc.value.detail)
+
    @pytest.mark.parametrize(
        "cypher",
        [
@@ -448,6 +483,8 @@ class TestValidation:
            "CALL apoc.create.vNode(['Label'], {}) YIELD node RETURN node",
            "MATCH (n) WHERE n.name = 'apoc.load.json' RETURN n",
            'MATCH (n) WHERE n.description = "LOAD CSV is cool" RETURN n',
+            "MATCH (n) /* LOAD CSV is documented here */ RETURN n",
+            "MATCH (n) WHERE n.note = '/* LOAD CSV */' RETURN n",
        ],
        ids=[
            "simple_match",
@@ -456,6 +493,8 @@ class TestValidation:
            "apoc_virtual_node",
            "apoc_load_inside_single_quotes",
            "load_csv_inside_double_quotes",
+            "load_csv_inside_block_comment",
+            "block_comment_inside_string",
        ],
    )
    def test_allows_clean_queries(self, cypher):
@@ -1,14 +1,16 @@
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta

 import pytest
 from allauth.socialaccount.models import SocialApp
 from api.db_router import MainRouter
 from api.models import (
+    Provider,
    ProviderComplianceScore,
    Resource,
    ResourceTag,
    SAMLConfiguration,
    SAMLDomainIndex,
+    Scan,
    StateChoices,
    StatusChoices,
    TenantComplianceSummary,
@@ -524,3 +526,226 @@ class TestTenantComplianceSummaryModel:

        assert summary1.id != summary2.id
        assert summary1.requirements_passed != summary2.requirements_passed
+
+
+def _latest_scan_fixture(tenant, provider, *, completed_at, inserted_at=None, **kwargs):
+    scan = Scan.objects.create(
+        tenant_id=tenant.id,
+        provider=provider,
+        trigger=kwargs.pop("trigger", Scan.TriggerChoices.MANUAL),
+        state=kwargs.pop("state", StateChoices.COMPLETED),
+        completed_at=completed_at,
+        **kwargs,
+    )
+    if inserted_at is not None:
+        # `inserted_at` is auto_now_add, so it has to be forced after the fact.
+        Scan.all_objects.filter(pk=scan.pk).update(inserted_at=inserted_at)
+        scan.refresh_from_db()
+    return scan
+
+
+@pytest.mark.django_db
+class TestScanQuerySetOrdering:
+    def test_completed_later_wins_over_inserted_later(
+        self, tenants_fixture, aws_provider
+    ):
+        """The scan that FINISHED last is current, not the one that started last."""
+        tenant, *_ = tenants_fixture
+        now = datetime.now(UTC)
+
+        finished_last = _latest_scan_fixture(
+            tenant,
+            aws_provider,
+            inserted_at=now - timedelta(hours=3),
+            completed_at=now,
+        )
+        _latest_scan_fixture(
+            tenant,
+            aws_provider,
+            inserted_at=now - timedelta(hours=1),
+            completed_at=now - timedelta(hours=1),
+        )
+
+        assert Scan.all_objects.filter(
+            tenant_id=tenant.id
+        ).latest_ids_per_provider() == [finished_last.id]
+
+    def test_null_completed_at_provider_is_still_returned(
+        self, tenants_fixture, aws_provider
+    ):
+        """NULLS LAST, not `completed_at__isnull=False`.
+
+        Excluding NULL `completed_at` would drop the provider from every
+        "latest" endpoint instead of falling back to `inserted_at`.
+        """
+        tenant, *_ = tenants_fixture
+        only_scan = _latest_scan_fixture(tenant, aws_provider, completed_at=None)
+
+        assert Scan.all_objects.filter(
+            tenant_id=tenant.id
+        ).latest_ids_per_provider() == [only_scan.id]
+
+    def test_null_completed_at_never_outranks_a_finished_scan(
+        self, tenants_fixture, aws_provider
+    ):
+        """Postgres sorts NULLs first under DESC; NULLS LAST is what fixes it."""
+        tenant, *_ = tenants_fixture
+        now = datetime.now(UTC)
+
+        finished = _latest_scan_fixture(
+            tenant,
+            aws_provider,
+            inserted_at=now - timedelta(hours=2),
+            completed_at=now - timedelta(hours=2),
+        )
+        _latest_scan_fixture(
+            tenant,
+            aws_provider,
+            inserted_at=now,
+            completed_at=None,
+        )
+
+        assert Scan.all_objects.filter(
+            tenant_id=tenant.id
+        ).latest_ids_per_provider() == [finished.id]
+
+    def test_id_breaks_an_exact_timestamp_tie_deterministically(
+        self, tenants_fixture, aws_provider
+    ):
+        tenant, *_ = tenants_fixture
+        now = datetime.now(UTC)
+
+        scans = [
+            _latest_scan_fixture(
+                tenant, aws_provider, inserted_at=now, completed_at=now
+            )
+            for _ in range(3)
+        ]
+        expected = max(scan.id for scan in scans)
+
+        picks = {
+            Scan.all_objects.filter(tenant_id=tenant.id).latest_ids_per_provider()[0]
+            for _ in range(5)
+        }
+        assert picks == {expected}
+
+
+@pytest.mark.django_db
+class TestScanQuerySetEligibility:
+    def test_unfinished_scans_are_excluded(self, tenants_fixture, aws_provider):
+        tenant, *_ = tenants_fixture
+        _latest_scan_fixture(
+            tenant,
+            aws_provider,
+            completed_at=None,
+            state=StateChoices.EXECUTING,
+        )
+        assert (
+            Scan.all_objects.filter(tenant_id=tenant.id).latest_ids_per_provider() == []
+        )
+
+
+@pytest.mark.django_db
+class TestScanQuerySetManagerChoice:
+    def test_active_manager_hides_soft_deleted_providers(
+        self, tenants_fixture, aws_provider
+    ):
+        """`Scan.objects` drops soft-deleted providers, `all_objects` keeps them.
+
+        The queryset must not decide this for the caller.
+        """
+        tenant, *_ = tenants_fixture
+        scan = _latest_scan_fixture(
+            tenant, aws_provider, completed_at=datetime.now(UTC)
+        )
+
+        Provider.all_objects.filter(pk=aws_provider.pk).update(is_deleted=True)
+
+        assert Scan.all_objects.filter(
+            tenant_id=tenant.id
+        ).latest_ids_per_provider() == [scan.id]
+        assert Scan.objects.filter(tenant_id=tenant.id).latest_ids_per_provider() == []
+
+
+@pytest.mark.django_db
+class TestScanQuerySetPerProviderScoping:
+    def test_one_scan_per_provider(self, tenants_fixture, aws_provider_pair):
+        tenant, *_ = tenants_fixture
+        provider_one, provider_two = aws_provider_pair
+        now = datetime.now(UTC)
+
+        newest_one = _latest_scan_fixture(tenant, provider_one, completed_at=now)
+        _latest_scan_fixture(tenant, provider_one, completed_at=now - timedelta(days=1))
+        newest_two = _latest_scan_fixture(tenant, provider_two, completed_at=now)
+
+        assert set(
+            Scan.all_objects.filter(tenant_id=tenant.id).latest_ids_per_provider()
+        ) == {newest_one.id, newest_two.id}
+
+    def test_caller_filters_are_preserved(self, tenants_fixture, aws_provider_pair):
+        tenant, *_ = tenants_fixture
+        provider_one, provider_two = aws_provider_pair
+        now = datetime.now(UTC)
+
+        scan_one = _latest_scan_fixture(tenant, provider_one, completed_at=now)
+        _latest_scan_fixture(tenant, provider_two, completed_at=now)
+
+        assert Scan.all_objects.filter(
+            tenant_id=tenant.id, provider__in=[provider_one]
+        ).latest_ids_per_provider() == [scan_one.id]
+
+    def test_latest_first_is_ordered_not_deduplicated(
+        self, tenants_fixture, aws_provider
+    ):
+        tenant, *_ = tenants_fixture
+        now = datetime.now(UTC)
+
+        newest = _latest_scan_fixture(tenant, aws_provider, completed_at=now)
+        older = _latest_scan_fixture(
+            tenant, aws_provider, completed_at=now - timedelta(days=1)
+        )
+
+        ordered = list(
+            Scan.all_objects.filter(
+                tenant_id=tenant.id, provider_id=aws_provider.id
+            ).latest_first()
+        )
+        assert [scan.id for scan in ordered] == [newest.id, older.id]
+
+    def test_tenant_isolation(self, tenants_fixture, aws_provider):
+        tenant, other_tenant, *_ = tenants_fixture
+        _latest_scan_fixture(tenant, aws_provider, completed_at=datetime.now(UTC))
+
+        assert (
+            Scan.all_objects.filter(tenant_id=other_tenant.id).latest_ids_per_provider()
+            == []
+        )
+
+    def test_empty_queryset_returns_empty_list(self, tenants_fixture):
+        tenant, *_ = tenants_fixture
+        assert (
+            Scan.all_objects.filter(tenant_id=tenant.id).latest_ids_per_provider() == []
+        )
+
+
+@pytest.mark.django_db
+class TestScanQuerySetPerProviderQuerysetShape:
+    def test_returns_a_queryset_not_a_list(self, tenants_fixture, aws_provider):
+        tenant, *_ = tenants_fixture
+        _latest_scan_fixture(tenant, aws_provider, completed_at=datetime.now(UTC))
+
+        qs = Scan.all_objects.filter(tenant_id=tenant.id).latest_per_provider()
+        # Callers chain .values(...) / .values_list(...) onto this.
+        assert qs.values_list("provider_id", flat=True).count() == 1
+
+
+@pytest.mark.django_db
+class TestScanQuerySetRelatedManager:
+    def test_reverse_relation_exposes_the_methods(self, tenants_fixture, aws_provider):
+        tenant, *_ = tenants_fixture
+        now = datetime.now(UTC)
+
+        newest = _latest_scan_fixture(tenant, aws_provider, completed_at=now)
+        _latest_scan_fixture(tenant, aws_provider, completed_at=now - timedelta(days=1))
+
+        assert aws_provider.scans.latest_first().first().id == newest.id
@@ -0,0 +1,63 @@
+from datetime import UTC, datetime
+from itertools import islice
+
+import pytest
+from api.partitions import (
+    PostgresUUIDv7PartitioningStrategy,
+    relative_months_or_none,
+)
+from dateutil.relativedelta import relativedelta
+from django.core.exceptions import ImproperlyConfigured
+from psqlextra.partitioning import PostgresTimePartitionSize
+
+
+def build_strategy(max_age):
+    return PostgresUUIDv7PartitioningStrategy(
+        size=PostgresTimePartitionSize(months=1),
+        count=1,
+        start_date=datetime.now(UTC),
+        max_age=max_age,
+        name_format="%Y_%b",
+    )
+
+
+class TestRelativeMonthsOrNone:
+    @pytest.mark.parametrize("value", [None, 0])
+    def test_unset_or_zero_keeps_partitions_indefinitely(self, value):
+        assert relative_months_or_none(value) is None
+
+    @pytest.mark.parametrize("months", [1, 3, 12])
+    def test_value_is_interpreted_as_months(self, months):
+        assert relative_months_or_none(months) == relativedelta(months=months)
+
+    def test_value_is_not_interpreted_as_days(self):
+        assert relative_months_or_none(12) != relativedelta(days=12)
+
+    def test_negative_is_rejected(self):
+        with pytest.raises(ImproperlyConfigured):
+            relative_months_or_none(-12)
+
+
+class TestToDelete:
+    @pytest.mark.parametrize("max_age", [None, relative_months_or_none(0)])
+    def test_nothing_is_deleted_without_max_age(self, max_age):
+        strategy = build_strategy(max_age)
+
+        assert list(islice(strategy.to_delete(), 5)) == []
+
+    def test_first_deleted_partition_is_max_age_old(self):
+        months = 3
+        strategy = build_strategy(relative_months_or_none(months))
+
+        first = next(strategy.to_delete())
+
+        expected = strategy.get_start_datetime() - relativedelta(months=months)
+        assert first.name() == expected.strftime("%Y_%b").lower()
+
+    def test_deleted_partitions_go_further_back_in_time(self):
+        strategy = build_strategy(relative_months_or_none(3))
+
+        names = [p.name() for p in islice(strategy.to_delete(), 3)]
+        starts = [datetime.strptime(n, "%Y_%b") for n in names]
+
+        assert starts == sorted(starts, reverse=True)
@@ -12,6 +12,7 @@ from api.models import (
    ProviderSecret,
    Role,
    RoleProviderGroupRelationship,
+    SAMLConfiguration,
    Scan,
    User,
    UserRoleRelationship,
@@ -1896,6 +1897,90 @@ class TestUserRoleLinkPermissions:
        assert response.status_code == status.HTTP_403_FORBIDDEN


+SAML_METADATA_XML = """<?xml version='1.0' encoding='UTF-8'?>
+<md:EntityDescriptor entityID='TEST' xmlns:md='urn:oasis:names:tc:SAML:2.0:metadata'>
+<md:IDPSSODescriptor WantAuthnRequestsSigned='false' protocolSupportEnumeration='urn:oasis:names:tc:SAML:2.0:protocol'>
+    <md:KeyDescriptor use='signing'>
+    <ds:KeyInfo xmlns:ds='http://www.w3.org/2000/09/xmldsig#'>
+        <ds:X509Data>
+        <ds:X509Certificate>TEST</ds:X509Certificate>
+        </ds:X509Data>
+    </ds:KeyInfo>
+    </md:KeyDescriptor>
+    <md:NameIDFormat>urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress</md:NameIDFormat>
+    <md:SingleSignOnService Binding='urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST' Location='https://TEST/sso/saml'/>
+    <md:SingleSignOnService Binding='urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect' Location='https://TEST/sso/saml'/>
+</md:IDPSSODescriptor>
+</md:EntityDescriptor>
+"""
+
+
+@pytest.mark.django_db
+class TestSAMLConfigurationPermissions:
+    def _create(self, client, email_domain):
+        return client.post(
+            reverse("saml-config-list"),
+            data={"email_domain": email_domain, "metadata_xml": SAML_METADATA_XML},
+            format="json",
+        )
+
+    def test_manage_integrations_only_cannot_create(
+        self, authenticated_client_rbac_manage_integrations_only
+    ):
+        response = self._create(
+            authenticated_client_rbac_manage_integrations_only,
+            "integrations-only-create.com",
+        )
+
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+        assert not SAMLConfiguration.objects.filter(
+            email_domain="integrations-only-create.com"
+        ).exists()
+
+    def test_manage_integrations_only_cannot_read_or_replace(
+        self, authenticated_client_rbac_manage_integrations_only
+    ):
+        client = authenticated_client_rbac_manage_integrations_only
+        tenant = Membership.objects.filter(user=client.user).first().tenant
+        config = SAMLConfiguration.objects.create(
+            tenant_id=str(tenant.id),
+            email_domain="integrations-only-replace.com",
+            metadata_xml=SAML_METADATA_XML,
+        )
+
+        list_response = client.get(reverse("saml-config-list"))
+        patch_response = client.patch(
+            reverse("saml-config-detail", kwargs={"pk": config.id}),
+            data={"metadata_xml": SAML_METADATA_XML},
+            format="json",
+        )
+        delete_response = client.delete(
+            reverse("saml-config-detail", kwargs={"pk": config.id})
+        )
+
+        assert list_response.status_code == status.HTTP_403_FORBIDDEN
+        assert patch_response.status_code == status.HTTP_403_FORBIDDEN
+        assert delete_response.status_code == status.HTTP_403_FORBIDDEN
+        assert SAMLConfiguration.objects.filter(id=config.id).exists()
+
+    def test_manage_account_only_can_create_and_delete(
+        self, authenticated_client_rbac_manage_account
+    ):
+        client = authenticated_client_rbac_manage_account
+
+        create_response = self._create(client, "account-only.com")
+
+        assert create_response.status_code == status.HTTP_201_CREATED
+        config = SAMLConfiguration.objects.get(email_domain="account-only.com")
+
+        delete_response = client.delete(
+            reverse("saml-config-detail", kwargs={"pk": config.id})
+        )
+
+        assert delete_response.status_code == status.HTTP_204_NO_CONTENT
+        assert not SAMLConfiguration.objects.filter(id=config.id).exists()
+
+
@pytest.mark.django_db
 class TestCrossTenantRoleLeak:
    """Regression tests for get_role() cross-tenant privilege leak.
@@ -30,6 +30,21 @@ def test_initialize_sentry_uses_configured_dsn():
    assert mock_init.call_args.kwargs["before_send"] is sentry_settings.before_send


+def test_initialize_sentry_sends_no_personal_data():
+    with (
+        patch.object(
+            sentry_settings.env,
+            "str",
+            return_value="https://fake-public-key@sentry.example.invalid/1",
+        ),
+        patch.object(sentry_settings.sentry_sdk, "init") as mock_init,
+    ):
+        sentry_settings.initialize_sentry()
+
+    assert mock_init.call_args.kwargs["send_default_pii"] is False
+    assert mock_init.call_args.kwargs["max_request_body_size"] == "never"
+
+
 def _make_log_record(msg, level=logging.ERROR, name="test", args=None):
    """Build a real LogRecord so getMessage() works like in production."""
    record = logging.LogRecord(
@@ -5,6 +5,8 @@ from api.v1.serializer_utils.integrations import (
 )
 from api.v1.serializer_utils.providers import ProviderSecretField
 from api.v1.serializers import (
+    KUBERNETES_KUBECONFIG_INVALID_ERROR,
+    KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR,
    ImageProviderSecret,
    IntegrationSerializer,
    IntegrationUpdateSerializer,
@@ -215,36 +217,34 @@ class TestOracleCloudProviderSecret:
        assert serializer.is_valid(), serializer.errors
        assert "region" not in serializer.validated_data

-    def test_accepts_and_ignores_region_field(self):
-        secret = self.valid_secret(region="us-phoenix-1")
-        serializer = OracleCloudProviderSecret(data=secret)
-
-        assert serializer.is_valid(), serializer.errors
-
-        assert "region" not in serializer.validated_data
-
-    @pytest.mark.parametrize(
-        "legacy_field, legacy_value",
-        [
-            ("region", None),
-            ("region", ""),
-            ("region", {"name": "us-ashburn-1"}),
-        ],
-    )
-    def test_accepts_and_ignores_any_legacy_region_value(
-        self, legacy_field, legacy_value
-    ):
+    def test_keeps_region_as_home_region(self):
        serializer = OracleCloudProviderSecret(
-            data=self.valid_secret(**{legacy_field: legacy_value})
+            data=self.valid_secret(region=" me-abudhabi-1 ")
        )

        assert serializer.is_valid(), serializer.errors
+        assert serializer.validated_data["region"] == "me-abudhabi-1"

-        assert legacy_field not in serializer.validated_data
+    def test_rejects_unknown_region(self):
+        serializer = OracleCloudProviderSecret(
+            data=self.valid_secret(region="mars-north-1")
+        )
+
+        assert not serializer.is_valid()
+        assert "region" in serializer.errors
+
+    @pytest.mark.parametrize("legacy_value", [None, "", {"name": "us-ashburn-1"}])
+    def test_drops_blank_or_non_string_region(self, legacy_value):
+        serializer = OracleCloudProviderSecret(
+            data=self.valid_secret(region=legacy_value)
+        )
+
+        assert serializer.is_valid(), serializer.errors
+        assert "region" not in serializer.validated_data


 class TestProviderSecretFieldSchema:
-    def test_oraclecloud_schema_includes_legacy_region_field(self):
+    def test_oraclecloud_schema_region_is_not_deprecated(self):
        schema = ProviderSecretField._spectacular_annotation["field"]
        oraclecloud_schema = next(
            credential_schema
@@ -253,7 +253,7 @@ class TestProviderSecretFieldSchema:
            == "Oracle Cloud Infrastructure (OCI) API Key Credentials"
        )

-        assert oraclecloud_schema["properties"]["region"]["deprecated"] is True
+        assert "deprecated" not in oraclecloud_schema["properties"]["region"]


 class TestKubernetesProviderSecret:
@@ -342,6 +342,278 @@ current-context: test-context
        assert not serializer.is_valid()
        assert "kubeconfig_content" in serializer.errors

+    @staticmethod
+    def _kubeconfig(user_extra="", cluster_extra=""):
+        return f"""
+apiVersion: v1
+kind: Config
+clusters:
+  - name: test-cluster
+    cluster:
+      server: https://kubernetes.example.test
+{cluster_extra}
+users:
+  - name: test-user
+    user:
+      token: test-token
+{user_extra}
+contexts:
+  - name: test-context
+    context:
+      cluster: test-cluster
+      user: test-user
+current-context: test-context
+"""
+
+    @staticmethod
+    def _assert_rejected_without_echo(
+        kubeconfig_content, leaked_value, expected_message
+    ):
+        serializer = KubernetesProviderSecret(
+            data={"kubeconfig_content": kubeconfig_content}
+        )
+
+        assert serializer.is_valid() is False
+        assert serializer.errors["kubeconfig_content"] == [expected_message]
+        assert leaked_value not in str(serializer.errors)
+
+    def test_inline_token_with_certificate_authority_data_is_accepted(self):
+        kubeconfig_content = self._kubeconfig(
+            cluster_extra="      certificate-authority-data: dGVzdA=="
+        )
+
+        serializer = KubernetesProviderSecret(
+            data={"kubeconfig_content": kubeconfig_content}
+        )
+
+        assert serializer.is_valid()
+
+    def test_kubeconfig_with_token_file_is_rejected(self):
+        self._assert_rejected_without_echo(
+            self._kubeconfig(user_extra="      tokenFile: /etc/passwd"),
+            "/etc/passwd",
+            KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR,
+        )
+
+    def test_kubeconfig_with_client_certificate_file_is_rejected(self):
+        self._assert_rejected_without_echo(
+            self._kubeconfig(user_extra="      client-certificate: /etc/passwd"),
+            "/etc/passwd",
+            KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR,
+        )
+
+    def test_kubeconfig_with_client_key_file_is_rejected(self):
+        self._assert_rejected_without_echo(
+            self._kubeconfig(user_extra="      client-key: /etc/passwd"),
+            "/etc/passwd",
+            KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR,
+        )
+
+    def test_kubeconfig_with_certificate_authority_file_is_rejected(self):
+        self._assert_rejected_without_echo(
+            self._kubeconfig(cluster_extra="      certificate-authority: /etc/passwd"),
+            "/etc/passwd",
+            KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR,
+        )
+
+    def test_kubeconfig_with_cluster_proxy_url_is_rejected(self):
+        self._assert_rejected_without_echo(
+            self._kubeconfig(cluster_extra="      proxy-url: http://proxy.evil.test"),
+            "proxy.evil.test",
+            KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR,
+        )
+
+    @pytest.mark.parametrize("provider_name", ["gcp", "oidc", "azure"])
+    def test_kubeconfig_with_any_auth_provider_is_rejected(self, provider_name):
+        self._assert_rejected_without_echo(
+            self._kubeconfig(
+                user_extra=(
+                    "      auth-provider:\n"
+                    f"        name: {provider_name}\n"
+                    "        config:\n"
+                    "          idp-issuer-url: https://idp.evil.test"
+                )
+            ),
+            "idp.evil.test",
+            KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR,
+        )
+
+    def test_kubeconfig_with_unknown_user_key_is_rejected(self):
+        self._assert_rejected_without_echo(
+            self._kubeconfig(user_extra="      bogus: /etc/passwd"),
+            "/etc/passwd",
+            KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR,
+        )
+
+    def test_kubeconfig_with_unknown_cluster_key_is_rejected(self):
+        self._assert_rejected_without_echo(
+            self._kubeconfig(cluster_extra="      bogus: /etc/passwd"),
+            "/etc/passwd",
+            KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR,
+        )
+
+    def test_kubeconfig_with_allowlisted_cluster_keys_is_accepted(self):
+        serializer = KubernetesProviderSecret(
+            data={
+                "kubeconfig_content": self._kubeconfig(
+                    cluster_extra=(
+                        "      certificate-authority-data: dGVzdA==\n"
+                        "      insecure-skip-tls-verify: false\n"
+                        "      tls-server-name: kubernetes.example.test\n"
+                        "      disable-compression: true\n"
+                        "      extensions: []"
+                    )
+                )
+            }
+        )
+
+        assert serializer.is_valid(), serializer.errors
+
+    def test_kubeconfig_with_non_list_clusters_is_rejected(self):
+        kubeconfig_content = """
+apiVersion: v1
+kind: Config
+clusters:
+  name: /etc/passwd
+"""
+        self._assert_rejected_without_echo(
+            kubeconfig_content, "/etc/passwd", KUBERNETES_KUBECONFIG_INVALID_ERROR
+        )
+
+    def test_kubeconfig_with_non_dict_cluster_entry_is_rejected(self):
+        kubeconfig_content = """
+apiVersion: v1
+kind: Config
+clusters:
+  - /etc/passwd
+"""
+        self._assert_rejected_without_echo(
+            kubeconfig_content, "/etc/passwd", KUBERNETES_KUBECONFIG_INVALID_ERROR
+        )
+
+    def test_kubeconfig_with_non_dict_cluster_value_is_rejected(self):
+        kubeconfig_content = """
+apiVersion: v1
+kind: Config
+clusters:
+  - name: test-cluster
+    cluster: /etc/passwd
+"""
+        self._assert_rejected_without_echo(
+            kubeconfig_content, "/etc/passwd", KUBERNETES_KUBECONFIG_INVALID_ERROR
+        )
+
+    @pytest.mark.parametrize(
+        "user_extra",
+        [
+            "      token: test-token",
+            "      username: test-user\n      password: test-password",
+            "      client-certificate-data: dGVzdA==\n      client-key-data: dGVzdA==",
+            "      token: test-token\n      as: other-user",
+            "      token: test-token\n      as-uid: 1234",
+            "      token: test-token\n      as-groups:\n        - group-a",
+            "      token: test-token\n      as-user-extra:\n        scopes:\n          - read",
+        ],
+        ids=[
+            "token",
+            "username-password",
+            "client-cert-data",
+            "as",
+            "as-uid",
+            "as-groups",
+            "as-user-extra",
+        ],
+    )
+    def test_kubeconfig_with_allowlisted_user_keys_is_accepted(self, user_extra):
+        kubeconfig_content = f"""
+apiVersion: v1
+kind: Config
+clusters:
+  - name: test-cluster
+    cluster:
+      server: https://kubernetes.example.test
+      certificate-authority-data: dGVzdA==
+users:
+  - name: test-user
+    user:
+{user_extra}
+contexts:
+  - name: test-context
+    context:
+      cluster: test-cluster
+      user: test-user
+current-context: test-context
+"""
+
+        serializer = KubernetesProviderSecret(
+            data={"kubeconfig_content": kubeconfig_content}
+        )
+
+        assert serializer.is_valid()
+
+    @pytest.mark.parametrize(
+        "kubeconfig_content",
+        [
+            """
+apiVersion: v1
+kind: Config
+clusters:
+  - name: test-cluster
+    cluster:
+      server: https://kubernetes.example.test
+users:
+  - name: good-user
+    user:
+      token: test-token
+  - name: bad-user
+    user:
+      tokenFile: /etc/passwd
+""",
+            """
+apiVersion: v1
+kind: Config
+clusters:
+  - name: good-cluster
+    cluster:
+      server: https://kubernetes.example.test
+  - name: bad-cluster
+    cluster:
+      server: https://kubernetes.example.test
+      certificate-authority: /etc/passwd
+users:
+  - name: test-user
+    user:
+      token: test-token
+""",
+        ],
+        ids=["second-user", "second-cluster"],
+    )
+    def test_kubeconfig_with_bad_second_entry_is_rejected(self, kubeconfig_content):
+        self._assert_rejected_without_echo(
+            kubeconfig_content,
+            "/etc/passwd",
+            KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR,
+        )
+
+    def test_kubeconfig_with_indented_document_is_rejected(self):
+        kubeconfig_content = (
+            "  wrapper:\n"
+            "  users:\n"
+            "  - name: u\n"
+            "    user:\n"
+            "      tokenFile: /etc/passwd\n"
+            "  clusters:\n"
+            "  - name: c\n"
+            "    cluster:\n"
+            "      server: https://example.test\n"
+        )
+
+        self._assert_rejected_without_echo(
+            kubeconfig_content,
+            "/etc/passwd",
+            KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR,
+        )
+
    def test_malformed_kubeconfig_is_rejected(self):
        serializer = KubernetesProviderSecret(
            data={"kubeconfig_content": "apiVersion: ["}
@@ -14,6 +14,7 @@ from api.utils import (
    return_prowler_provider,
    validate_invitation,
 )
+from api.v1.serializers import KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR
 from prowler.providers.alibabacloud.alibabacloud_provider import AlibabacloudProvider
 from prowler.providers.aws.aws_provider import AwsProvider
 from prowler.providers.aws.lib.security_hub.security_hub import SecurityHubConnection
@@ -35,6 +36,34 @@ from prowler.providers.oraclecloud.oraclecloud_provider import OraclecloudProvid
 from prowler.providers.vercel.vercel_provider import VercelProvider
 from rest_framework.exceptions import NotFound, ValidationError

+VALID_KUBECONFIG = """\
+apiVersion: v1
+kind: Config
+clusters:
+- name: test-cluster
+  cluster:
+    server: https://example.invalid:6443
+    certificate-authority-data: Y2E=
+users:
+- name: test-user
+  user:
+    token: test-token
+contexts:
+- name: test-context
+  context:
+    cluster: test-cluster
+    user: test-user
+current-context: test-context
+"""
+
+TOKEN_FILE_KUBECONFIG = VALID_KUBECONFIG.replace(
+    "token: test-token", "tokenFile: /etc/passwd"
+)
+
+EXEC_KUBECONFIG = VALID_KUBECONFIG.replace(
+    "token: test-token", "exec:\n      command: test-command"
+)
+

 class TestMergeDicts:
    def test_simple_merge(self):
@@ -172,7 +201,7 @@ class TestInitializeProwlerProvider:
        )

    @patch("api.utils.return_prowler_provider")
-    def test_initialize_oraclecloud_provider_removes_region_string(
+    def test_initialize_oraclecloud_provider_passes_region_as_home_region(
        self, mock_return_prowler_provider
    ):
        provider = MagicMock()
@@ -182,7 +211,7 @@ class TestInitializeProwlerProvider:
            "fingerprint": "00:11:22:33:44:55:66:77",
            "key_content": "fake-base64-key-content",
            "tenancy": "ocid1.tenancy.oc1..fake",
-            "region": "us-ashburn-1",
+            "region": "me-abudhabi-1",
        }
        mock_return_prowler_provider.return_value = MagicMock()

@@ -193,6 +222,7 @@ class TestInitializeProwlerProvider:
            fingerprint="00:11:22:33:44:55:66:77",
            key_content="fake-base64-key-content",
            tenancy="ocid1.tenancy.oc1..fake",
+            home_region="me-abudhabi-1",
        )

    @patch("api.utils.return_prowler_provider")
@@ -254,15 +284,73 @@ class TestProwlerProviderConnectionTest:
            fingerprint="00:11:22:33:44:55:66:77",
            key_content="fake-base64-key-content",
            tenancy="ocid1.tenancy.oc1..aaaaaaaexample",
-            region=getattr(
-                OraclecloudProvider,
-                "_bootstrap_region",
-                OraclecloudProvider._home_region,
-            ),
+            region=OraclecloudProvider._bootstrap_region,
            provider_id="ocid1.tenancy.oc1..aaaaaaaexample",
            raise_on_exception=False,
        )

+    @patch("api.utils.return_prowler_provider")
+    def test_oraclecloud_connection_test_uses_stored_region_for_identity(
+        self, mock_return_prowler_provider
+    ):
+        provider = MagicMock()
+        provider.uid = "ocid1.tenancy.oc1..aaaaaaaexample"
+        provider.provider = Provider.ProviderChoices.ORACLECLOUD.value
+        provider.secret.secret = {
+            "user": "ocid1.user.oc1..aaaaaaaexample",
+            "fingerprint": "00:11:22:33:44:55:66:77",
+            "key_content": "fake-base64-key-content",
+            "tenancy": "ocid1.tenancy.oc1..aaaaaaaexample",
+            "region": "me-abudhabi-1",
+        }
+        mock_return_prowler_provider.return_value = MagicMock()
+
+        prowler_provider_connection_test(provider)
+
+        mock_return_prowler_provider.return_value.test_connection.assert_called_once_with(
+            user="ocid1.user.oc1..aaaaaaaexample",
+            fingerprint="00:11:22:33:44:55:66:77",
+            key_content="fake-base64-key-content",
+            tenancy="ocid1.tenancy.oc1..aaaaaaaexample",
+            region="me-abudhabi-1",
+            provider_id="ocid1.tenancy.oc1..aaaaaaaexample",
+            raise_on_exception=False,
+        )
+
+    @patch("api.utils.return_prowler_provider")
+    def test_kubernetes_connection_test_rejects_invalid_kubeconfig(
+        self, mock_return_prowler_provider
+    ):
+        provider = MagicMock()
+        provider.provider = Provider.ProviderChoices.KUBERNETES.value
+        provider.uid = "provider_uid"
+        provider.secret.secret = {"kubeconfig_content": TOKEN_FILE_KUBECONFIG}
+
+        connection = prowler_provider_connection_test(provider)
+
+        assert connection.is_connected is False
+        assert (
+            str(connection.error) == KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR
+        )
+        mock_return_prowler_provider.return_value.test_connection.assert_not_called()
+
+    @patch("api.utils.return_prowler_provider")
+    def test_kubernetes_connection_test_with_valid_kubeconfig(
+        self, mock_return_prowler_provider
+    ):
+        provider = MagicMock()
+        provider.provider = Provider.ProviderChoices.KUBERNETES.value
+        provider.uid = "provider_uid"
+        provider.secret.secret = {"kubeconfig_content": VALID_KUBECONFIG}
+
+        prowler_provider_connection_test(provider)
+
+        mock_return_prowler_provider.return_value.test_connection.assert_called_once_with(
+            kubeconfig_content=VALID_KUBECONFIG,
+            provider_id="provider_uid",
+            raise_on_exception=False,
+        )
+
    @pytest.mark.django_db
    @patch("api.utils.return_prowler_provider")
    def test_prowler_provider_connection_test_without_secret(
@@ -380,10 +468,6 @@ class TestGetProwlerProviderKwargs:
                Provider.ProviderChoices.GOOGLEWORKSPACE.value,
                {},
            ),
-            (
-                Provider.ProviderChoices.KUBERNETES.value,
-                {"context": "provider_uid"},
-            ),
            (
                Provider.ProviderChoices.M365.value,
                {},
@@ -434,7 +518,40 @@ class TestGetProwlerProviderKwargs:
        expected_result = {**secret_dict, **expected_extra_kwargs}
        assert result == expected_result

-    def test_get_prowler_provider_kwargs_oraclecloud_removes_region(
+    def test_get_prowler_provider_kwargs_kubernetes_valid_kubeconfig(self):
+        provider = MagicMock()
+        provider.provider = Provider.ProviderChoices.KUBERNETES.value
+        provider.uid = "provider_uid"
+        provider.secret.secret = {"kubeconfig_content": VALID_KUBECONFIG}
+
+        assert get_prowler_provider_kwargs(provider) == {
+            "kubeconfig_content": VALID_KUBECONFIG,
+            "context": "provider_uid",
+        }
+
+    @pytest.mark.parametrize(
+        "secret",
+        [
+            {"kubeconfig_content": TOKEN_FILE_KUBECONFIG},
+            {"kubeconfig_content": EXEC_KUBECONFIG},
+            {"kubeconfig_content": "[]"},
+            {"key": "value"},
+        ],
+    )
+    def test_get_prowler_provider_kwargs_kubernetes_rejects_invalid_kubeconfig(
+        self, secret
+    ):
+        provider = MagicMock()
+        provider.provider = Provider.ProviderChoices.KUBERNETES.value
+        provider.uid = "provider_uid"
+        provider.secret.secret = secret
+
+        with pytest.raises(ValidationError) as exc_info:
+            get_prowler_provider_kwargs(provider)
+
+        assert "/etc/passwd" not in str(exc_info.value)
+
+    def test_get_prowler_provider_kwargs_oraclecloud_maps_region_to_home_region(
        self,
    ):
        secret_dict = {
@@ -461,6 +578,7 @@ class TestGetProwlerProviderKwargs:
            "key_content": "-----BEGIN PRIVATE KEY-----\nfake\n-----END PRIVATE KEY-----",
            "tenancy": "ocid1.tenancy.oc1..fake",
            "pass_phrase": "fake-passphrase",
+            "home_region": "us-ashburn-1",
        }

    def test_get_prowler_provider_kwargs_with_mutelist(self):
@@ -17,7 +17,7 @@ from api.models import (
    Role,
    UserRoleRelationship,
 )
-from api.v1.serializers import FindingMetadataSerializer
+from api.v1.serializers import FindingMetadataSerializer, KubernetesProviderSecret
 from django.contrib.postgres.aggregates import ArrayAgg
 from django.db import transaction
 from django.db.models import Subquery
@@ -201,6 +201,13 @@ def return_prowler_provider(
    return prowler_provider


+def _validate_kubernetes_secret(secret: dict) -> None:
+    """Re-validate a stored kubeconfig at use time; errors never echo its content."""
+    KubernetesProviderSecret(
+        data={"kubeconfig_content": secret.get("kubeconfig_content")}
+    ).is_valid(raise_exception=True)
+
+
 def get_prowler_provider_kwargs(
    provider: Provider, mutelist_processor: Processor | None = None
 ) -> dict:
@@ -225,6 +232,7 @@ def get_prowler_provider_kwargs(
            "project_ids": [provider.uid],
        }
    elif provider.provider == Provider.ProviderChoices.KUBERNETES.value:
+        _validate_kubernetes_secret(prowler_provider_kwargs)
        prowler_provider_kwargs = {**prowler_provider_kwargs, "context": provider.uid}
    elif provider.provider == Provider.ProviderChoices.GITHUB.value:
        if provider.uid:
@@ -302,17 +310,26 @@ def get_prowler_provider_kwargs(
 def _normalize_oraclecloud_provider_kwargs(secret: dict) -> dict:
    """Normalize external OCI secret fields into SDK provider kwargs."""
    prowler_provider_kwargs = secret.copy()
-    prowler_provider_kwargs.pop("region", None)
+    home_region = _oraclecloud_home_region(prowler_provider_kwargs.pop("region", None))
+    if home_region:
+        prowler_provider_kwargs["home_region"] = home_region

    return prowler_provider_kwargs


+def _oraclecloud_home_region(region) -> str | None:
+    """Return the stored OCI region as a home region, ignoring blank or non-string legacy values."""
+    if isinstance(region, str) and region.strip():
+        return region.strip()
+    return None
+
+
 def _normalize_oraclecloud_connection_test_kwargs(secret: dict) -> dict:
    """Normalize external OCI secret fields into test_connection kwargs."""
    from prowler.providers.oraclecloud.oraclecloud_provider import OraclecloudProvider

    prowler_provider_kwargs = secret.copy()
-    prowler_provider_kwargs.pop("region", None)
+    home_region = _oraclecloud_home_region(prowler_provider_kwargs.pop("region", None))

    if (
        prowler_provider_kwargs.get("user")
@@ -323,11 +340,9 @@ def _normalize_oraclecloud_connection_test_kwargs(secret: dict) -> dict:
            or prowler_provider_kwargs.get("key_file")
        )
    ):
-        # Connection validation needs one OCI endpoint, but scans remain unfiltered.
-        prowler_provider_kwargs["region"] = getattr(
-            OraclecloudProvider,
-            "_bootstrap_region",
-            OraclecloudProvider._home_region,
+        # Identity calls only succeed in a region the tenancy is subscribed to.
+        prowler_provider_kwargs["region"] = (
+            home_region or OraclecloudProvider._bootstrap_region
        )

    return prowler_provider_kwargs
@@ -385,6 +400,16 @@ def prowler_provider_connection_test(provider: Provider) -> Connection:
    except Provider.secret.RelatedObjectDoesNotExist as secret_error:
        return Connection(is_connected=False, error=secret_error)

+    if provider.provider == Provider.ProviderChoices.KUBERNETES.value:
+        # A rejected stored kubeconfig is a failed connection, not a task error.
+        try:
+            _validate_kubernetes_secret(prowler_provider_kwargs)
+        except ValidationError as validation_error:
+            return Connection(
+                is_connected=False,
+                error=Exception(validation_error.detail["kubeconfig_content"][0]),
+            )
+
    # For IaC provider, construct the kwargs properly for test_connection
    if provider.provider == Provider.ProviderChoices.IAC.value:
        # Don't pass repository_url from secret, use scan_repository_url with the UID
@@ -214,7 +214,8 @@ from rest_framework_json_api import serializers
                    "kubeconfig_content": {
                        "type": "string",
                        "description": "The content of the Kubernetes kubeconfig file, encoded as a string. "
-                        "Kubeconfig command-based authentication is not supported in Prowler Cloud for security reasons.",
+                        "Only inline credentials are supported: token, username/password, or client-certificate-data with client-key-data for users, and certificate-authority-data for clusters. "
+                        "File-path and auth-provider kubeconfig fields are not supported.",
                    }
                },
                "required": ["kubeconfig_content"],
@@ -301,8 +302,7 @@ from rest_framework_json_api import serializers
                    },
                    "region": {
                        "type": "string",
-                        "deprecated": True,
-                        "description": "Legacy OCI region field accepted for backwards compatibility but ignored; OCI scans all regions.",
+                        "description": "Optional OCI home region (or any region the tenancy is subscribed to) used to validate the credentials. It does not filter the scan, which covers all subscribed regions. Defaults to us-ashburn-1.",
                    },
                },
                "required": ["user", "fingerprint", "tenancy"],
@@ -71,6 +71,7 @@ from django.db import IntegrityError, transaction
 from drf_spectacular.utils import extend_schema_field
 from jwt.exceptions import InvalidKeyError
 from prowler.lib.mutelist.mutelist import Mutelist
+from prowler.providers.oraclecloud.config import OCI_REGIONS
 from rest_framework.reverse import reverse
 from rest_framework.validators import UniqueTogetherValidator
 from rest_framework_json_api import serializers
@@ -376,27 +377,48 @@ class UserSerializer(BaseModelSerializerV1):
    def _can_view_relationships(self, instance) -> bool:
        """Allow self to view own relationships. Require manage_account to view others."""
        role = self.context.get("role")
+        return self._is_self(instance) or bool(role and role.manage_account)
+
+    def _is_self(self, instance) -> bool:
        request = self.context.get("request")
-        is_self = bool(
+        return bool(
            request
            and getattr(request, "user", None)
            and getattr(instance, "id", None) == request.user.id
        )
-        return is_self or (role and role.manage_account)
+
+    def _active_tenant_id(self):
+        return getattr(self.context.get("request"), "tenant_id", None)
+
+    @staticmethod
+    def _in_tenant(related_manager, tenant_id):
+        """Filter a prefetched related manager in memory.
+
+        ``.filter()`` clones the queryset without its result cache, so it would
+        discard the viewset's prefetch and re-query once per serialized user.
+        """
+        return [
+            item
+            for item in related_manager.all()
+            if str(item.tenant_id) == str(tenant_id)
+        ]

    def get_roles(self, instance):
-        return (
-            instance.roles.all()
-            if self._can_view_relationships(instance)
-            else Role.objects.none()
-        )
+        tenant_id = self._active_tenant_id()
+        if not tenant_id or not self._can_view_relationships(instance):
+            return Role.objects.none()
+        return self._in_tenant(instance.roles, tenant_id)

    def get_memberships(self, instance):
-        return (
-            instance.memberships.all()
-            if self._can_view_relationships(instance)
-            else Membership.objects.none()
-        )
+        if not self._can_view_relationships(instance):
+            return Membership.objects.none()
+        # Own memberships span tenants on purpose: the tenant switcher reads them.
+        if self._is_self(instance):
+            return instance.memberships.all()
+        tenant_id = self._active_tenant_id()
+        if not tenant_id:
+            return Membership.objects.none()
+        return self._in_tenant(instance.memberships, tenant_id)


 class UserMeSerializer(UserSerializer):
@@ -730,11 +752,32 @@ class MembershipIncludeSerializer(serializers.ModelSerializer):

    included_serializers = {"tenant": "api.v1.serializers.TenantIncludeSerializer"}

+    _REQUESTER_TENANTS_ATTR = "_prowler_requester_tenants"
+
+    def _requester_tenants(self):
+        """Tenants the requester belongs to, resolved once per request.
+
+        Memoised on the request because DRF builds a serializer instance per
+        included object, so a per-instance cache would still query per membership.
+        """
+        request = self.context.get("request")
+        user = getattr(request, "user", None)
+        if user is None:
+            return {}
+        tenants = getattr(request, self._REQUESTER_TENANTS_ATTR, None)
+        if tenants is None:
+            tenants = {
+                tenant.id: tenant
+                for tenant in Tenant.objects.using(MainRouter.admin_db).filter(
+                    membership__user_id=user.id
+                )
+            }
+            setattr(request, self._REQUESTER_TENANTS_ATTR, tenants)
+        return tenants
+
    def get_tenant(self, instance):
-        try:
-            return Tenant.objects.using(MainRouter.admin_db).get(id=instance.tenant_id)
-        except Tenant.DoesNotExist:
-            return None
+        """Resolve the tenant only when the requester is a member of it."""
+        return self._requester_tenants().get(instance.tenant_id)


 # Provider Groups
@@ -1658,14 +1701,43 @@ class FindingMetadataSerializer(BaseSerializerV1):


 # Provider secrets
-KUBERNETES_KUBECONFIG_UNSUPPORTED_COMMAND_AUTH_ERROR = (
-    "Kubernetes kubeconfig command-based authentication is not supported in "
-    "Prowler Cloud for security reasons."
+KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR = (
+    "Only inline Kubernetes credentials are supported. "
+    "For user authentication use token, username/password, or client-certificate-data with client-key-data; "
+    "clusters may use certificate-authority-data. "
+    "File-path fields (tokenFile, client-certificate, client-key, certificate-authority), "
+    "command-based authentication (exec, cmd-path), auth-provider directives, and proxy URLs are not supported."
 )
 KUBERNETES_KUBECONFIG_INVALID_ERROR = "Invalid Kubernetes kubeconfig content."


-def kubeconfig_contains_unsupported_command_auth(kubeconfig: dict) -> bool:
+KUBECONFIG_ALLOWED_USER_KEYS = frozenset(
+    {
+        "token",
+        "username",
+        "password",
+        "client-certificate-data",
+        "client-key-data",
+        "as",
+        "as-uid",
+        "as-groups",
+        "as-user-extra",
+    }
+)
+KUBECONFIG_ALLOWED_CLUSTER_KEYS = frozenset(
+    {
+        "server",
+        "certificate-authority-data",
+        "insecure-skip-tls-verify",
+        "tls-server-name",
+        "disable-compression",
+        "extensions",
+    }
+)
+
+
+def kubeconfig_is_inline_only_credentials(kubeconfig: dict) -> bool:
+    """Accept only inline credentials; reject file-reference and auth-provider fields."""
    users = kubeconfig.get("users", [])
    if not isinstance(users, list):
        raise ValidationError(KUBERNETES_KUBECONFIG_INVALID_ERROR)
@@ -1678,21 +1750,25 @@ def kubeconfig_contains_unsupported_command_auth(kubeconfig: dict) -> bool:
        if not isinstance(user, dict):
            raise ValidationError(KUBERNETES_KUBECONFIG_INVALID_ERROR)

-        if "exec" in user:
-            return True
+        if any(key not in KUBECONFIG_ALLOWED_USER_KEYS for key in user):
+            return False

-        auth_provider = user.get("auth-provider", {})
-        if not isinstance(auth_provider, dict):
-            continue
+    clusters = kubeconfig.get("clusters", [])
+    if not isinstance(clusters, list):
+        raise ValidationError(KUBERNETES_KUBECONFIG_INVALID_ERROR)

-        auth_provider_config = auth_provider.get("config", {})
-        if not isinstance(auth_provider_config, dict):
-            continue
+    for cluster_entry in clusters:
+        if not isinstance(cluster_entry, dict):
+            raise ValidationError(KUBERNETES_KUBECONFIG_INVALID_ERROR)

-        if "cmd-path" in auth_provider_config:
-            return True
+        cluster = cluster_entry.get("cluster", {})
+        if not isinstance(cluster, dict):
+            raise ValidationError(KUBERNETES_KUBECONFIG_INVALID_ERROR)

-    return False
+        if any(key not in KUBECONFIG_ALLOWED_CLUSTER_KEYS for key in cluster):
+            return False
+
+    return True


 class BaseWriteProviderSecretSerializer(BaseWriteSerializer):
@@ -1875,7 +1951,7 @@ class MongoDBAtlasProviderSecret(serializers.Serializer):


 class KubernetesProviderSecret(serializers.Serializer):
-    kubeconfig_content = serializers.CharField()
+    kubeconfig_content = serializers.CharField(trim_whitespace=False)

    def validate_kubeconfig_content(self, kubeconfig_content):
        try:
@@ -1888,9 +1964,9 @@ class KubernetesProviderSecret(serializers.Serializer):
        if not isinstance(kubeconfig, dict):
            raise serializers.ValidationError(KUBERNETES_KUBECONFIG_INVALID_ERROR)

-        if kubeconfig_contains_unsupported_command_auth(kubeconfig):
+        if not kubeconfig_is_inline_only_credentials(kubeconfig):
            raise serializers.ValidationError(
-                KUBERNETES_KUBECONFIG_UNSUPPORTED_COMMAND_AUTH_ERROR
+                KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR
            )

        return kubeconfig_content
@@ -1917,9 +1993,16 @@ class IacProviderSecret(serializers.Serializer):
        resource_name = "provider-secrets"


-class LegacyOCIRegionField(serializers.Field):
+class OCIHomeRegionField(serializers.Field):
+    """Optional OCI home region; blank or non-string legacy values are dropped."""
+
    def to_internal_value(self, data):
-        return data
+        if not isinstance(data, str) or not data.strip():
+            return None
+        region = data.strip()
+        if region not in OCI_REGIONS:
+            raise serializers.ValidationError(f"Invalid OCI region: {region}")
+        return region

    def to_representation(self, value):
        return value
@@ -1932,10 +2015,11 @@ class OracleCloudProviderSecret(serializers.Serializer):
    key_content = serializers.CharField(required=False)
    tenancy = serializers.CharField()
    pass_phrase = serializers.CharField(required=False)
-    region = LegacyOCIRegionField(required=False, allow_null=True)
+    region = OCIHomeRegionField(required=False, allow_null=True)

    def validate(self, attrs):
-        attrs.pop("region", None)
+        if not attrs.get("region"):
+            attrs.pop("region", None)

        if "key_file" not in attrs and "key_content" not in attrs:
            raise serializers.ValidationError(
@@ -2149,6 +2233,12 @@ class InvitationSerializer(RLSSerializer):
        if tenant_id is not None:
            self.fields["roles"].queryset = Role.objects.filter(tenant_id=tenant_id)

+    def to_representation(self, instance):
+        data = super().to_representation(instance)
+        if instance.is_lapsed:
+            data["state"] = Invitation.State.EXPIRED.value
+        return data
+
    class Meta:
        model = Invitation
        fields = [
@@ -2175,6 +2265,7 @@ class InvitationBaseWriteSerializer(BaseWriteSerializer):
            self.fields["roles"].queryset = Role.objects.filter(tenant_id=tenant_id)

    def validate_email(self, value):
+        value = value.strip().lower()
        user = User.objects.filter(email=value).first()
        tenant_id = self.context["tenant_id"]
        if user and Membership.objects.filter(user=user, tenant=tenant_id).exists():
@@ -2182,9 +2273,13 @@ class InvitationBaseWriteSerializer(BaseWriteSerializer):
                "The user may already be a member of the tenant or there was an issue with the "
                "email provided."
            )
-        if Invitation.objects.filter(
-            email=value, state=Invitation.State.PENDING
-        ).exists():
+        pending_invitations = Invitation.objects.filter(
+            tenant_id=tenant_id, email=value, state=Invitation.State.PENDING
+        )
+        pending_invitations.filter(Invitation.lapsed_q()).update(
+            state=Invitation.State.EXPIRED
+        )
+        if pending_invitations.filter(expires_at__gt=datetime.now(UTC)).exists():
            raise ValidationError(
                "Unable to process your request. Please check the information provided and "
                "try again."
@@ -125,10 +125,14 @@ from api.models import (
 )
 from api.pagination import ComplianceOverviewPagination
 from api.rbac.permissions import (
+    TASK_REVOKE_PERMISSIONS,
    Permissions,
    get_integrations,
    get_providers,
    get_role,
+    get_tasks,
+    get_user_roles,
+    roles_have_permissions,
 )
 from api.renderers import APIJSONRenderer, PlainTextRenderer
 from api.rls import Tenant
@@ -244,7 +248,6 @@ from api.v1.serializers import (
    UserUpdateSerializer,
 )
 from botocore.exceptions import ClientError, NoCredentialsError, ParamValidationError
-from celery import chain
 from celery.result import AsyncResult
 from config.custom_logging import BackendLogger
 from config.env import env
@@ -327,7 +330,7 @@ from rest_framework_simplejwt.token_blacklist.models import (
 )
 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.jobs.export import get_s3_client, get_s3_presign_client
 from tasks.tasks import (
    QUEUED_SCAN_TASK_STATE,
    backfill_compliance_summaries_task,
@@ -342,13 +345,21 @@ from tasks.tasks import (
    enqueue_scan_execution_on_commit,
    get_active_provider_scan,
    jira_integration_task,
-    mute_historical_findings_task,
-    reaggregate_all_finding_group_summaries_task,
+    mute_findings_in_latest_scans_task,
    refresh_lighthouse_provider_models_task,
 )

 logger = logging.getLogger(BackendLogger.API)

+
+def _scans_visible_to(role, tenant_id):
+    """Tenant scans, narrowed to the role's provider groups unless it has unlimited visibility."""
+    scans = Scan.all_objects.filter(tenant_id=tenant_id)
+    if role.unlimited_visibility:
+        return scans
+    return scans.filter(provider__in=get_providers(role))
+
+
 CACHE_DECORATOR = cache_control(
    max_age=django_settings.CACHE_MAX_AGE,
    stale_while_revalidate=django_settings.CACHE_STALE_WHILE_REVALIDATE,
@@ -797,7 +808,9 @@ class SAMLConfigurationViewSet(BaseRLSViewSet):
    """

    serializer_class = SAMLConfigurationSerializer
-    required_permissions = [Permissions.MANAGE_INTEGRATIONS]
+    # The IdP behind this configuration decides who signs in and which role
+    # they get, so writing it is account administration, not an integration.
+    required_permissions = [Permissions.MANAGE_ACCOUNT]
    queryset = SAMLConfiguration.objects.all()

    def get_queryset(self):
@@ -1894,7 +1907,7 @@ class ProviderViewSet(DisablePaginationMixin, BaseRLSViewSet):
            self.required_permissions = [Permissions.MANAGE_PROVIDERS]

    def get_queryset(self):
-        user_roles = get_role(self.request.user, self.request.tenant_id)
+        user_roles = self.user_role
        if user_roles.unlimited_visibility:
            # User has unlimited visibility, return all providers
            queryset = Provider.objects.filter(tenant_id=self.request.tenant_id)
@@ -2409,7 +2422,8 @@ class ScanViewSet(ProviderVisibilityMixin, BaseRLSViewSet):
            }
            if content_type:
                params["ResponseContentType"] = content_type
-            url = client.generate_presigned_url(
+            # The browser follows this URL, so it is signed against the public host.
+            url = (get_s3_presign_client() or client).generate_presigned_url(
                "get_object",
                Params=params,
                ExpiresIn=300,
@@ -2823,6 +2837,15 @@ class ScanViewSet(ProviderVisibilityMixin, BaseRLSViewSet):
                tenant_id=self.request.tenant_id,
                task_id=pre_task_id,
                task_status=(QUEUED_SCAN_TASK_STATE if active_scan else None),
+                # This response is serialized before the on_commit publish,
+                # so without these the caller gets a task id and no scan id.
+                # Kept in step with what `enqueue_scan_execution_on_commit`
+                # publishes below.
+                task_kwargs={
+                    "tenant_id": str(self.request.tenant_id),
+                    "scan_id": str(scan.id),
+                    "provider_id": str(scan.provider_id),
+                },
            )

            if not active_scan:
@@ -2850,17 +2873,29 @@ class ScanViewSet(ProviderVisibilityMixin, BaseRLSViewSet):
    list=extend_schema(
        tags=["Task"],
        summary="List all tasks",
-        description="Retrieve a list of all tasks with options for filtering by name, state, and other criteria.",
+        description=(
+            "Retrieve a list of all tasks with options for filtering by name, state, and other "
+            "criteria. Tasks that reference a provider are only returned when the role can "
+            "access it; tasks without a provider reference are returned for every role."
+        ),
    ),
    retrieve=extend_schema(
        tags=["Task"],
        summary="Retrieve data from a specific task",
-        description="Fetch detailed information about a specific task by its ID.",
+        description=(
+            "Fetch detailed information about a specific task by its ID. Tasks tied to a provider "
+            "outside the visibility of the role are not found."
+        ),
    ),
    destroy=extend_schema(
        tags=["Task"],
        summary="Revoke a task",
-        description="Try to revoke a task using its ID. Only tasks that are not yet in progress can be revoked.",
+        description=(
+            "Try to revoke a task using its ID. Only tasks that are not yet in progress can be "
+            "revoked, and the caller needs the same permission as the operation that queued "
+            "the task (for example MANAGE_SCANS for a scan). Provider deletions cannot be "
+            "revoked."
+        ),
        responses={202: OpenApiResponse(response=TaskSerializer)},
    ),
 )
@@ -2876,13 +2911,26 @@ class TaskViewSet(BaseRLSViewSet):
    required_permissions = []

    def get_queryset(self):
-        return Task.objects.annotate(
-            name=F("task_runner_task__task_name"),
-            state=F("task_runner_task__status"),
-        ).select_related("task_runner_task")
+        return (
+            get_tasks(self.user_role)
+            .annotate(
+                name=F("task_runner_task__task_name"),
+                state=F("task_runner_task__status"),
+            )
+            .select_related("task_runner_task")
+        )

    def destroy(self, request, *args, pk=None, **kwargs):
-        task = get_object_or_404(Task, pk=pk)
+        task = self.get_object()
+        required_permissions = TASK_REVOKE_PERMISSIONS.get(
+            task.task_runner_task.task_name
+        )
+        # Same multi-role semantics as HasPermissions.
+        if required_permissions is None or not roles_have_permissions(
+            get_user_roles(request.user, request.tenant_id), required_permissions
+        ):
+            raise PermissionDenied("You do not have permission to revoke this task.")
+
        if task.task_runner_task.status not in ["PENDING", "RECEIVED"]:
            serializer = TaskSerializer(task)
            return Response(
@@ -3021,7 +3069,7 @@ class AttackPathsScanViewSet(BaseRLSViewSet):
        return super().get_serializer_class()

    def get_queryset(self):
-        user_roles = get_role(self.request.user, self.request.tenant_id)
+        user_roles = self.user_role
        base_queryset = AttackPathsScan.objects.filter(tenant_id=self.request.tenant_id)

        if user_roles.unlimited_visibility:
@@ -3375,7 +3423,7 @@ class ResourceViewSet(PaginateByPkMixin, BaseRLSViewSet):
    required_permissions = []

    def get_queryset(self):
-        user_roles = get_role(self.request.user, self.request.tenant_id)
+        user_roles = self.user_role
        if user_roles.unlimited_visibility:
            # User has unlimited visibility, return all scans
            queryset = Resource.all_objects.filter(tenant_id=self.request.tenant_id)
@@ -3479,12 +3527,8 @@ class ResourceViewSet(PaginateByPkMixin, BaseRLSViewSet):
        filtered_queryset = self.filter_queryset(self.get_queryset())

        latest_scans = (
-            Scan.all_objects.filter(
-                tenant_id=tenant_id,
-                state=StateChoices.COMPLETED,
-            )
-            .order_by("provider_id", "-inserted_at")
-            .distinct("provider_id")
+            Scan.all_objects.filter(tenant_id=tenant_id)
+            .latest_per_provider()
            .values("provider_id")
        )

@@ -3511,8 +3555,15 @@ class ResourceViewSet(PaginateByPkMixin, BaseRLSViewSet):

        tenant_id = request.tenant_id
        query_params = request.query_params
+        user_roles = self.user_role

        queryset = ResourceScanSummary.objects.filter(tenant_id=tenant_id)
+        if not user_roles.unlimited_visibility:
+            queryset = queryset.filter(
+                scan_id__in=_scans_visible_to(user_roles, tenant_id).values_list(
+                    "id", flat=True
+                )
+            )

        if scans := query_params.get("filter[scan__in]") or query_params.get(
            "filter[scan]"
@@ -3590,7 +3641,10 @@ class ResourceViewSet(PaginateByPkMixin, BaseRLSViewSet):
        all_groups = Resource.objects.filter(
            tenant_id=tenant_id,
            groups__isnull=False,
-        ).values_list("groups", flat=True)
+        )
+        if not user_roles.unlimited_visibility:
+            all_groups = all_groups.filter(provider__in=get_providers(user_roles))
+        all_groups = all_groups.values_list("groups", flat=True)
        groups = sorted(
            {g for groups_list in all_groups if groups_list for g in groups_list}
        )
@@ -3615,12 +3669,11 @@ class ResourceViewSet(PaginateByPkMixin, BaseRLSViewSet):
    def metadata_latest(self, request):
        tenant_id = request.tenant_id
        query_params = request.query_params
+        user_roles = self.user_role

-        latest_scans_queryset = (
-            Scan.all_objects.filter(tenant_id=tenant_id, state=StateChoices.COMPLETED)
-            .order_by("provider_id", "-inserted_at")
-            .distinct("provider_id")
-        )
+        latest_scans_queryset = _scans_visible_to(
+            user_roles, tenant_id
+        ).latest_per_provider()

        queryset = ResourceScanSummary.objects.filter(
            tenant_id=tenant_id,
@@ -3660,7 +3713,10 @@ class ResourceViewSet(PaginateByPkMixin, BaseRLSViewSet):
        all_groups = Resource.objects.filter(
            tenant_id=tenant_id,
            groups__isnull=False,
-        ).values_list("groups", flat=True)
+        )
+        if not user_roles.unlimited_visibility:
+            all_groups = all_groups.filter(provider__in=get_providers(user_roles))
+        all_groups = all_groups.values_list("groups", flat=True)
        groups = sorted(
            {g for groups_list in all_groups if groups_list for g in groups_list}
        )
@@ -3997,7 +4053,7 @@ class FindingViewSet(PaginateByPkMixin, BaseRLSViewSet):

    def get_queryset(self):
        tenant_id = self.request.tenant_id
-        user_roles = get_role(self.request.user, self.request.tenant_id)
+        user_roles = self.user_role
        if user_roles.unlimited_visibility:
            # User has unlimited visibility, return all findings
            queryset = Finding.all_objects.filter(tenant_id=tenant_id)
@@ -4071,8 +4127,16 @@ class FindingViewSet(PaginateByPkMixin, BaseRLSViewSet):

        tenant_id = request.tenant_id
        query_params = request.query_params
+        user_roles = self.user_role
+        visible_scan_ids = (
+            None
+            if user_roles.unlimited_visibility
+            else _scans_visible_to(user_roles, tenant_id).values_list("id", flat=True)
+        )

        queryset = ResourceScanSummary.objects.filter(tenant_id=tenant_id)
+        if not user_roles.unlimited_visibility:
+            queryset = queryset.filter(scan_id__in=visible_scan_ids)
        scan_based_filters = {}
        category_scan_filters = {}  # Filters for ScanCategorySummary

@@ -4130,7 +4194,10 @@ class FindingViewSet(PaginateByPkMixin, BaseRLSViewSet):
        if not queryset.exists():
            raw_scans_ids = Scan.objects.filter(
                tenant_id=tenant_id, **scan_based_filters
-            ).values_list("id", "unique_resource_count")
+            )
+            if not user_roles.unlimited_visibility:
+                raw_scans_ids = raw_scans_ids.filter(id__in=visible_scan_ids)
+            raw_scans_ids = raw_scans_ids.values_list("id", "unique_resource_count")
            scan_ids = [
                scan_id for scan_id, count in raw_scans_ids if count and count > 0
            ]
@@ -4172,11 +4239,13 @@ class FindingViewSet(PaginateByPkMixin, BaseRLSViewSet):
        )

        # Get categories from ScanCategorySummary using same scan filters
+        category_summaries = ScanCategorySummary.objects.filter(
+            tenant_id=tenant_id, **category_scan_filters
+        )
+        if not user_roles.unlimited_visibility:
+            category_summaries = category_summaries.filter(scan_id__in=visible_scan_ids)
        categories = list(
-            ScanCategorySummary.objects.filter(
-                tenant_id=tenant_id, **category_scan_filters
-            )
-            .values_list("category", flat=True)
+            category_summaries.values_list("category", flat=True)
            .distinct()
            .order_by("category")
        )
@@ -4207,12 +4276,9 @@ class FindingViewSet(PaginateByPkMixin, BaseRLSViewSet):
        tenant_id = request.tenant_id
        filtered_queryset = self.filter_queryset(self.get_queryset())

-        latest_scan_ids = list(
-            Scan.all_objects.filter(tenant_id=tenant_id, state=StateChoices.COMPLETED)
-            .order_by("provider_id", "-inserted_at")
-            .distinct("provider_id")
-            .values_list("id", flat=True)
-        )
+        latest_scan_ids = Scan.all_objects.filter(
+            tenant_id=tenant_id
+        ).latest_ids_per_provider()
        filtered_queryset = filtered_queryset.filter(
            tenant_id=tenant_id, scan_id__in=latest_scan_ids
        )
@@ -4234,12 +4300,11 @@ class FindingViewSet(PaginateByPkMixin, BaseRLSViewSet):
    def metadata_latest(self, request):
        tenant_id = request.tenant_id
        query_params = request.query_params
+        user_roles = self.user_role

-        latest_scans_queryset = (
-            Scan.all_objects.filter(tenant_id=tenant_id, state=StateChoices.COMPLETED)
-            .order_by("provider_id", "-inserted_at")
-            .distinct("provider_id")
-        )
+        latest_scans_queryset = _scans_visible_to(
+            user_roles, tenant_id
+        ).latest_per_provider()
        raw_latest_scans_ids = list(
            latest_scans_queryset.values_list("id", "unique_resource_count")
        )
@@ -4473,7 +4538,7 @@ class InvitationViewSet(BaseRLSViewSet):

    def partial_update(self, request, *args, **kwargs):
        instance = self.get_object()
-        if instance.state != Invitation.State.PENDING:
+        if instance.state != Invitation.State.PENDING or instance.is_lapsed:
            raise ValidationError(detail="This invitation cannot be updated.")
        serializer = self.get_serializer(
            instance,
@@ -4487,7 +4552,7 @@ class InvitationViewSet(BaseRLSViewSet):

    def destroy(self, request, *args, **kwargs):
        instance = self.get_object()
-        if instance.state != Invitation.State.PENDING:
+        if instance.state != Invitation.State.PENDING or instance.is_lapsed:
            raise ValidationError(detail="This invitation cannot be revoked.")
        instance.state = Invitation.State.REVOKED
        instance.save()
@@ -4980,11 +5045,7 @@ class ComplianceOverviewViewSet(
        if provider_filters:
            scans = scans.filter(**provider_filters)

-        return list(
-            scans.order_by("provider_id", "-inserted_at")
-            .distinct("provider_id")
-            .values_list("id", flat=True)
-        )
+        return scans.latest_ids_per_provider()

    def _filtered_queryset_for_latest_provider_scans(self, latest_scan_ids=None):
        if latest_scan_ids is None:
@@ -5726,14 +5787,9 @@ class OverviewViewSet(ProviderFilterParamsMixin, BaseRLSViewSet):
            else {}
        )

-        latest_scan_ids = (
-            Scan.all_objects.filter(
-                tenant_id=tenant_id, state=StateChoices.COMPLETED, **provider_filter
-            )
-            .order_by("provider_id", "-inserted_at")
-            .distinct("provider_id")
-            .values_list("id", flat=True)
-        )
+        latest_scan_ids = Scan.all_objects.filter(
+            tenant_id=tenant_id, **provider_filter
+        ).latest_ids_per_provider()

        return filtered_queryset.filter(
            tenant_id=tenant_id, scan_id__in=latest_scan_ids
@@ -5760,16 +5816,10 @@ class OverviewViewSet(ProviderFilterParamsMixin, BaseRLSViewSet):

    def _latest_scan_ids_for_allowed_providers(self, tenant_id, provider_filters=None):
        provider_filter = self._get_provider_filter()
-        queryset = Scan.all_objects.filter(
-            tenant_id=tenant_id, state=StateChoices.COMPLETED, **provider_filter
-        )
+        queryset = Scan.all_objects.filter(tenant_id=tenant_id, **provider_filter)
        if provider_filters:
            queryset = queryset.filter(**provider_filters)
-        return (
-            queryset.order_by("provider_id", "-inserted_at")
-            .distinct("provider_id")
-            .values_list("id", flat=True)
-        )
+        return queryset.latest_ids_per_provider()

    @action(detail=False, methods=["get"], url_name="providers")
    def providers(self, request):
@@ -5781,14 +5831,9 @@ class OverviewViewSet(ProviderFilterParamsMixin, BaseRLSViewSet):
            else {}
        )

-        latest_scan_ids = (
-            Scan.all_objects.filter(
-                tenant_id=tenant_id, state=StateChoices.COMPLETED, **provider_filter
-            )
-            .order_by("provider_id", "-inserted_at")
-            .distinct("provider_id")
-            .values_list("id", flat=True)
-        )
+        latest_scan_ids = Scan.all_objects.filter(
+            tenant_id=tenant_id, **provider_filter
+        ).latest_ids_per_provider()

        findings_aggregated = (
            queryset.filter(scan_id__in=latest_scan_ids)
@@ -7551,35 +7596,28 @@ class MuteRuleViewSet(BaseRLSViewSet):
        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)

-        # Create the mute rule
+        tenant_id = str(request.tenant_id)
+        finding_ids = serializer.validated_data["finding_ids"]
+        provider_ids = list(
+            dict.fromkeys(
+                Finding.all_objects.filter(
+                    id__in=finding_ids, tenant_id=tenant_id
+                ).values_list("scan__provider_id", flat=True)
+            )
+        )
+
        mute_rule = serializer.save()

-        tenant_id = str(request.tenant_id)
-        finding_ids = request.data.get("finding_ids", [])
-
-        # Immediately mute the selected findings
-        Finding.all_objects.filter(
-            id__in=finding_ids, tenant_id=tenant_id, muted=False
-        ).update(
-            muted=True,
-            muted_at=mute_rule.inserted_at,
-            muted_reason=mute_rule.reason,
-        )
-
-        # Launch background task for historical muting + reaggregation
        transaction.on_commit(
-            lambda: chain(
-                mute_historical_findings_task.si(
-                    tenant_id=tenant_id,
-                    mute_rule_id=str(mute_rule.id),
-                ),
-                reaggregate_all_finding_group_summaries_task.si(
-                    tenant_id=tenant_id,
-                ),
-            ).apply_async()
+            lambda: mute_findings_in_latest_scans_task.apply_async(
+                kwargs={
+                    "tenant_id": tenant_id,
+                    "mute_rule_id": str(mute_rule.id),
+                    "provider_ids": [str(provider_id) for provider_id in provider_ids],
+                }
+            )
        )

-        # Return the created mute rule
        serializer = self.get_serializer(mute_rule)
        return Response(
            data=serializer.data,
@@ -7942,18 +7980,13 @@ class FindingGroupViewSet(JsonApiFilterMixin, BaseRLSViewSet):

    def _get_latest_findings_per_provider(self, filtered_queryset):
        """Keep only findings from each provider's most recent completed scan."""
-        # Materialize to a literal IN list. Left as a subquery, Postgres can't
-        # estimate the match count and picks a serial nested loop on
-        # resource_finding_mappings when one scan dominates findings
-        latest_scan_ids = list(
-            Scan.objects.filter(
-                tenant_id=self.request.tenant_id,
-                state=StateChoices.COMPLETED,
-            )
-            .order_by("provider_id", "-completed_at", "-inserted_at")
-            .distinct("provider_id")
-            .values_list("id", flat=True)
-        )
+        # `latest_ids_per_provider` materializes to a literal IN list.
+        # Left as a subquery, Postgres can't estimate the match count and picks
+        # a serial nested loop on resource_finding_mappings when one scan
+        # dominates findings
+        latest_scan_ids = Scan.objects.filter(
+            tenant_id=self.request.tenant_id
+        ).latest_ids_per_provider()
        return filtered_queryset.filter(scan_id__in=latest_scan_ids)

    def _post_process_aggregation(self, aggregated_data):
@@ -8899,16 +8932,14 @@ class FindingGroupViewSet(JsonApiFilterMixin, BaseRLSViewSet):
        tenant_id = request.tenant_id
        queryset = self._get_finding_queryset()

-        # Order by -completed_at (matching the /latest summary path and the
-        # daily summary upsert keyed on midnight(completed_at)) so that
-        # overlapping scans do not make /resources and /latest read from
-        # different scans and report diverging counts.
-        latest_scan_ids = (
-            Scan.objects.filter(tenant_id=tenant_id, state=StateChoices.COMPLETED)
-            .order_by("provider_id", "-completed_at", "-inserted_at")
-            .distinct("provider_id")
-            .values_list("id", flat=True)
-        )
+        # The shared selector orders by -completed_at (matching the /latest
+        # summary path and the daily summary upsert keyed on
+        # midnight(completed_at)) so that overlapping scans do not make
+        # /resources and /latest read from different scans and report
+        # diverging counts.
+        latest_scan_ids = Scan.objects.filter(
+            tenant_id=tenant_id
+        ).latest_ids_per_provider()

        normalized_params = self._normalize_jsonapi_params(request.query_params)
        # Remove date filters since we're using latest
@@ -231,6 +231,62 @@ LOGGING = {
            "level": LEVEL,
            "propagate": False,
        },
+        # Celery loggers must be declared explicitly because
+        # disable_existing_loggers=True silences any logger that exists at
+        # dictConfig time but is not named here. Without these, fatal worker
+        # errors (e.g. celery.worker CRITICAL) produce no output.
+        # "celery" must keep propagating: get_task_logger() parents task
+        # loggers under celery.task, so blocking here hides them from root.
+        "celery": {
+            "level": LEVEL,
+            "propagate": True,
+        },
+        "celery.worker": {
+            "handlers": ["tasks_console"],
+            "level": LEVEL,
+            "propagate": False,
+        },
+        "celery.worker.consumer": {
+            "handlers": ["tasks_console"],
+            "level": LEVEL,
+            "propagate": False,
+        },
+        "celery.worker.consumer.consumer": {
+            "handlers": ["tasks_console"],
+            "level": LEVEL,
+            "propagate": False,
+        },
+        "kombu": {
+            "handlers": ["tasks_console"],
+            "level": LEVEL,
+            "propagate": False,
+        },
+        "kombu.transport.redis": {
+            "handlers": ["tasks_console"],
+            "level": LEVEL,
+            "propagate": False,
+        },
+        "billiard": {
+            "handlers": ["tasks_console"],
+            "level": LEVEL,
+            "propagate": False,
+        },
+        "amqp": {
+            "handlers": ["tasks_console"],
+            "level": LEVEL,
+            "propagate": False,
+        },
+        # WARNING keeps task failures but skips one "succeeded" line per task.
+        "celery.app.trace": {
+            "handlers": ["tasks_console"],
+            "level": "WARNING",
+            "propagate": False,
+        },
+        "celery.beat": {
+            "handlers": ["tasks_console"],
+            "level": LEVEL,
+            "propagate": False,
+        },
    },
    # Gunicorn required configuration
    "root": {
@@ -7,6 +7,7 @@ from config.settings.eventstream import *  # noqa
 from config.settings.partitions import *  # noqa
 from config.settings.sentry import *  # noqa
 from config.settings.social_login import *  # noqa
+from django.core.exceptions import ImproperlyConfigured

 SECRET_KEY = env("SECRET_KEY", default="secret")
 DEBUG = env.bool("DJANGO_DEBUG", default=False)
@@ -129,6 +130,8 @@ REST_FRAMEWORK = {
 SPECTACULAR_SETTINGS = {
    "SERVE_INCLUDE_SCHEMA": False,
    "COMPONENT_SPLIT_REQUEST": True,
+    # redoc@latest is 3.x, whose standalone bundle fails as a classic script
+    "REDOC_DIST": "https://cdn.jsdelivr.net/npm/redoc@2.5.1",
    "PREPROCESSING_HOOKS": [
        "drf_spectacular_jsonapi.hooks.fix_nested_path_parameters",
    ],
@@ -295,6 +298,14 @@ DJANGO_OUTPUT_S3_AWS_SECRET_ACCESS_KEY = env.str(
 )
 DJANGO_OUTPUT_S3_AWS_SESSION_TOKEN = env.str("DJANGO_OUTPUT_S3_AWS_SESSION_TOKEN", "")
 DJANGO_OUTPUT_S3_AWS_DEFAULT_REGION = env.str("DJANGO_OUTPUT_S3_AWS_DEFAULT_REGION", "")
+# Storage endpoint the API and Celery workers use to talk to S3-compatible object storage
+# such as MinIO. Empty means the real AWS S3 endpoint, which is unaffected.
+DJANGO_OUTPUT_S3_AWS_ENDPOINT_URL = env.str("DJANGO_OUTPUT_S3_AWS_ENDPOINT_URL", "")
+# Browser-reachable storage host used to sign download URLs. Empty means sign against the
+# same endpoint the API talks to, which is what Prowler Cloud on S3 does.
+DJANGO_OUTPUT_S3_AWS_PUBLIC_ENDPOINT_URL = env.str(
+    "DJANGO_OUTPUT_S3_AWS_PUBLIC_ENDPOINT_URL", ""
+)

 # HTTP Security Headers
 SECURE_CONTENT_TYPE_NOSNIFF = True
@@ -303,6 +314,11 @@ SECURE_REFERRER_POLICY = "strict-origin-when-cross-origin"

 DJANGO_DELETION_BATCH_SIZE = env.int("DJANGO_DELETION_BATCH_SIZE", 5000)

+# Public base URL of the Prowler UI (for example https://cloud.prowler.com). Used to
+# build links back to findings in outbound integrations such as Jira. Empty by
+# default, so self-hosted deployments emit no links unless they configure it.
+UI_BASE_URL = env.str("DJANGO_UI_BASE_URL", "").rstrip("/")
+
 # SAML requirement
 CSRF_COOKIE_SECURE = True
 SESSION_COOKIE_SECURE = True
@@ -315,6 +331,17 @@ ATTACK_PATHS_SCAN_STALE_THRESHOLD_MINUTES = env.int(
    "ATTACK_PATHS_SCAN_STALE_THRESHOLD_MINUTES", 960
 )  # 16h

+# Minimum age (of the scan row, or of the scan id itself when the row is gone) before
+# the periodic reaper will drop an orphaned temp Neo4j database. Keeps a scan that is
+# still legitimately in flight from ever losing its staging database mid-run.
+ATTACK_PATHS_TMP_DB_REAP_SAFETY_MARGIN_HOURS = env.int(
+    "ATTACK_PATHS_TMP_DB_REAP_SAFETY_MARGIN_HOURS", 6
+)
+if ATTACK_PATHS_TMP_DB_REAP_SAFETY_MARGIN_HOURS <= 0:
+    raise ImproperlyConfigured(
+        "ATTACK_PATHS_TMP_DB_REAP_SAFETY_MARGIN_HOURS must be a positive number of hours"
+    )
+
 # Selects where the persistent attack-paths graph is stored. The scan
 # temporary database is always Neo4j; only the sink is configurable.
 # Valid values: "neo4j" (default, OSS and local dev), "neptune" (hosted).
@@ -53,6 +53,10 @@ preload_app = not DEBUG
 # that may take longer, such as complex API operations.
 timeout = env.int("GUNICORN_TIMEOUT", default=120)

+# Recycle workers after N requests (+ jitter) to cap memory growth. GUNICORN_MAX_REQUESTS=0 disables.
+max_requests = env.int("GUNICORN_MAX_REQUESTS", default=10000)
+max_requests_jitter = env.int("GUNICORN_MAX_REQUESTS_JITTER", default=1000)
+
 # Logging
 logconfig_dict = DJANGO_LOGGERS
 gunicorn_logger = logging.getLogger(BackendLogger.GUNICORN)
@@ -193,10 +193,10 @@ def initialize_sentry():

    sentry_sdk.init(
        dsn=sentry_dsn,
-        # Add data like request headers and IP for users,
-        # see https://docs.sentry.io/platforms/python/data-management/data-collected/ for more info
        before_send=before_send,
-        send_default_pii=True,
+        # No user identity, IPs, cookies, headers or request bodies: bodies carry emails and provider details.
+        send_default_pii=False,
+        max_request_body_size="never",
        traces_sample_rate=env.float("DJANGO_SENTRY_TRACES_SAMPLE_RATE", default=0.02),
        _experiments={
            # Set continuous_profiling_auto_start to True
@@ -443,6 +443,47 @@ def authenticated_client_rbac_manage_users_only(
    )


+@pytest.fixture(scope="function")
+def create_test_user_rbac_manage_integrations_only(django_db_setup, django_db_blocker):
+    """User with only manage_integrations permission (no manage_account)."""
+    with django_db_blocker.unblock():
+        user = User.objects.create_user(
+            name="testing_manage_integrations_only",
+            email="rbac_manage_integrations_only@rbac.com",
+            password=TEST_PASSWORD,
+        )
+        tenant = Tenant.objects.create(name="Tenant Test Manage Integrations Only")
+        Membership.objects.create(
+            user=user,
+            tenant=tenant,
+            role=Membership.RoleChoices.OWNER,
+        )
+        role = Role.objects.create(
+            name="manage_integrations_only",
+            tenant_id=tenant.id,
+            manage_users=False,
+            manage_account=False,
+            manage_billing=False,
+            manage_providers=False,
+            manage_integrations=True,
+            manage_scans=False,
+            unlimited_visibility=False,
+        )
+        UserRoleRelationship.objects.create(user=user, role=role, tenant_id=tenant.id)
+    return user
+
+
+@pytest.fixture
+def authenticated_client_rbac_manage_integrations_only(
+    create_test_user_rbac_manage_integrations_only, client
+):
+    return authenticate_client_for_tenant(
+        client,
+        create_test_user_rbac_manage_integrations_only,
+        first_membership_tenant(create_test_user_rbac_manage_integrations_only),
+    )
+
+
@pytest.fixture
 def authenticated_client_rbac(create_test_user_rbac, tenants_fixture, client):
    return authenticate_client_for_tenant(
@@ -0,0 +1,107 @@
+"""Periodic reaper for orphaned temp Neo4j scan databases.
+
+`scan.py` creates a throw-away `db-tmp-scan-<attack_paths_scan_id>` database per
+scan and drops it once the scan finishes, success or failure. When the worker
+or Neo4j itself dies mid-scan, that drop never runs and nothing else ever
+revisits the database - it sits there forever. This sweep lists every temp
+database on the ingest cluster and drops the ones whose scan is gone or has
+been finished for longer than the configured safety margin.
+"""
+
+from datetime import UTC, datetime, timedelta
+
+from api.attack_paths import database as graph_database
+from api.db_router import MainRouter
+from api.models import AttackPathsScan, StateChoices
+from api.uuid_utils import datetime_from_uuid7
+from celery.utils.log import get_task_logger
+from config.django.base import ATTACK_PATHS_TMP_DB_REAP_SAFETY_MARGIN_HOURS
+from uuid6 import UUID as UUID7
+
+logger = get_task_logger(__name__)
+
+TERMINAL_STATES = (
+    StateChoices.COMPLETED,
+    StateChoices.FAILED,
+    StateChoices.CANCELLED,
+)
+
+
+def reap_orphaned_tmp_databases() -> dict:
+    """Drop temp Neo4j scan databases whose scan is gone or long finished.
+
+    A failure listing databases aborts the whole sweep (nothing to iterate).
+    A failure reaping one database is logged and skipped so the rest of the
+    sweep still runs.
+    """
+    now = datetime.now(tz=UTC)
+    safety_margin = timedelta(hours=ATTACK_PATHS_TMP_DB_REAP_SAFETY_MARGIN_HOURS)
+
+    try:
+        databases = graph_database.list_databases()
+    except Exception:
+        logger.exception("Failed to list ingest Neo4j databases for temp-db reap")
+        return {"dropped_count": 0, "databases": []}
+
+    tmp_databases = [
+        name for name in databases if name.startswith(graph_database.TEMP_DB_PREFIX)
+    ]
+
+    dropped: list[str] = []
+    for database in tmp_databases:
+        try:
+            if _is_orphaned(database, now, safety_margin):
+                graph_database.drop_database(database)
+                dropped.append(database)
+                logger.info(f"Dropped orphaned temp Neo4j database `{database}`")
+        except Exception:
+            logger.exception(f"Failed to reap temp Neo4j database `{database}`")
+
+    logger.info(f"Temp Neo4j database reap: {len(dropped)} dropped")
+    return {"dropped_count": len(dropped), "databases": dropped}
+
+
+def _is_orphaned(database: str, now: datetime, safety_margin: timedelta) -> bool:
+    """Decide whether a temp database is safe to drop.
+
+    No scan row: the row was hard-deleted (tenant/provider cleanup) or was
+    never created. Falls back to the scan id's own UUIDv7 timestamp so a
+    database created moments ago is never touched even without a row to check.
+
+    Scan row present: only reapable once it reached a terminal state and has
+    been finished for longer than the safety margin, so a scan still
+    legitimately executing is never touched.
+    """
+    scan_id = database[len(graph_database.TEMP_DB_PREFIX) :]
+
+    try:
+        scan_uuid = UUID7(scan_id)
+    except ValueError:
+        logger.warning(
+            f"Temp database `{database}` has an unparseable scan id, skipping"
+        )
+        return False
+
+    # Global sweep with no tenant context: admin_db bypasses RLS on purpose, the same
+    # way cleanup_stale_attack_paths_scans finds stale scans across every tenant.
+    scan = (
+        AttackPathsScan.all_objects.using(MainRouter.admin_db)
+        .filter(id=scan_uuid)
+        .first()
+    )
+
+    if scan is None:
+        if scan_uuid.version != 7:
+            logger.warning(
+                f"Temp database `{database}` has no scan row and a non-UUIDv7 id, "
+                "skipping"
+            )
+            return False
+        return now - datetime_from_uuid7(scan_uuid) >= safety_margin
+
+    if scan.state not in TERMINAL_STATES:
+        return False
+
+    # `mark_scan_finished` does not touch `updated_at`, so prefer `completed_at`
+    finished_at = scan.completed_at or scan.updated_at
+    return now - finished_at >= safety_margin
@@ -499,10 +499,13 @@ def backfill_provider_compliance_scores(tenant_id: str) -> dict:
                provider_id__in=existing_providers
            )

+        # `completed_scans` keeps its own `completed_at__isnull=False`: this
+        # task writes a *dated* ProviderComplianceScore row, so unlike the read
+        # paths it genuinely cannot use a scan without a `completed_at`.
        scan_info = list(
-            completed_scans.order_by("provider_id", "-completed_at")
-            .distinct("provider_id")
-            .values("id", "provider_id", "completed_at")
+            completed_scans.latest_per_provider().values(
+                "id", "provider_id", "completed_at"
+            )
        )

        if not scan_info:
@@ -13,6 +13,7 @@ from api.models import (
    Tenant,
 )
 from celery.utils.log import get_task_logger
+from django.conf import settings
 from django.db import DatabaseError
 from tasks.jobs.queries import (
    COMPLIANCE_DELETE_EMPTY_TENANT_SUMMARY_SQL,
@@ -106,9 +107,19 @@ def delete_provider(tenant_id: str, pk: str):
    try:
        if attack_paths_sink_backends:
            for sink_backend in attack_paths_sink_backends:
-                sink_module.get_backend_for_name(sink_backend).drop_subgraph(
-                    tenant_database_name, str(pk)
-                )
+                try:
+                    backend = sink_module.get_backend_for_name(sink_backend)
+
+                except RuntimeError as sink_error:
+                    # A retired sink has no connection settings left, and no graph left to drop
+                    if sink_backend == settings.ATTACK_PATHS_SINK_DATABASE.lower():
+                        raise
+                    logger.warning(
+                        f"Skipping graph cleanup on unconfigured sink {sink_backend}: {sink_error}"
+                    )
+                    continue
+
+                backend.drop_subgraph(tenant_database_name, str(pk))
        else:
            graph_database.drop_subgraph(tenant_database_name, str(pk))

@@ -6,6 +6,7 @@ import boto3
 import config.django.base as base
 from api.db_utils import rls_transaction
 from api.models import Scan
+from botocore.config import Config
 from botocore.exceptions import ClientError, NoCredentialsError, ParamValidationError
 from celery.utils.log import get_task_logger
 from django.conf import settings
@@ -206,15 +207,21 @@ def get_s3_client():
    This function attempts to initialize an S3 client by reading the AWS access key, secret key,
    session token, and region from environment variables. It then validates the client by listing
    available S3 buckets. If an error occurs during this process (for example, due to missing or
-    invalid credentials), it falls back to creating an S3 client without explicitly provided credentials,
-    which may rely on other configuration sources (e.g., IAM roles).
+    invalid credentials), it falls back to creating an S3 client without explicitly provided
+    credentials, which may rely on other configuration sources (e.g., IAM roles).
+
+    That fallback is only safe when no explicit endpoint is configured: with an endpoint set, the
+    explicit client already targets the intended S3-compatible storage, and the fallback client
+    would go to the AWS default provider chain instead, an unrelated real-AWS account reachable
+    from the host. So when an endpoint is configured, the original error propagates instead.

    Returns:
        boto3.client: A configured S3 client instance.

    Raises:
-        ClientError, NoCredentialsError, or ParamValidationError if both attempts to create a client fail.
+        ClientError, NoCredentialsError, or ParamValidationError if the client cannot be created.
    """
+    endpoint = settings.DJANGO_OUTPUT_S3_AWS_ENDPOINT_URL
    s3_client = None
    try:
        s3_client = boto3.client(
@@ -222,16 +229,68 @@ def get_s3_client():
            aws_access_key_id=settings.DJANGO_OUTPUT_S3_AWS_ACCESS_KEY_ID,
            aws_secret_access_key=settings.DJANGO_OUTPUT_S3_AWS_SECRET_ACCESS_KEY,
            aws_session_token=settings.DJANGO_OUTPUT_S3_AWS_SESSION_TOKEN,
-            region_name=settings.DJANGO_OUTPUT_S3_AWS_DEFAULT_REGION,
+            # Storage that has no meaningful region, MinIO among it, is usually configured
+            # without one, and botocore rejects an empty region before any request is made.
+            region_name=settings.DJANGO_OUTPUT_S3_AWS_DEFAULT_REGION or "us-east-1",
+            endpoint_url=endpoint or None,
        )
        s3_client.list_buckets()
    except (ClientError, NoCredentialsError, ParamValidationError, ValueError):
+        if endpoint:
+            raise
        s3_client = boto3.client("s3")
        s3_client.list_buckets()

    return s3_client


+def get_s3_presign_client():
+    """Return a client that signs download URLs with SigV4.
+
+    It is used when a public or internal storage host is configured, or when the bucket's
+    region is: boto3 otherwise presigns S3 URLs with SigV2, which S3 rejects for SSE-KMS
+    objects. None means none of those is set and the caller should presign with its own
+    client, which leaves those deployments with the URL they get today.
+
+    The public endpoint wins when both are set: the internal endpoint may only be reachable
+    from inside the cluster, and a URL signed against it would not open in a browser.
+    """
+    endpoint = (
+        settings.DJANGO_OUTPUT_S3_AWS_PUBLIC_ENDPOINT_URL
+        or settings.DJANGO_OUTPUT_S3_AWS_ENDPOINT_URL
+    )
+    if not endpoint and not settings.DJANGO_OUTPUT_S3_AWS_DEFAULT_REGION:
+        return None
+
+    # Blank keys are signed as-is (empty credential scope) instead of deferring to the
+    # provider chain, so static credentials are only passed when they are set.
+    credentials = {}
+    if (
+        settings.DJANGO_OUTPUT_S3_AWS_ACCESS_KEY_ID
+        and settings.DJANGO_OUTPUT_S3_AWS_SECRET_ACCESS_KEY
+    ):
+        credentials = {
+            "aws_access_key_id": settings.DJANGO_OUTPUT_S3_AWS_ACCESS_KEY_ID,
+            "aws_secret_access_key": settings.DJANGO_OUTPUT_S3_AWS_SECRET_ACCESS_KEY,
+            # An empty string is a token as far as botocore is concerned: it appends an
+            # empty X-Amz-Security-Token that storage counts when it recomputes the signature.
+            "aws_session_token": settings.DJANGO_OUTPUT_S3_AWS_SESSION_TOKEN or None,
+        }
+
+    return boto3.client(
+        "s3",
+        **credentials,
+        # SigV4 puts the region in the credential scope, and MinIO answers to us-east-1
+        # unless it was told otherwise, so an empty region would sign an unusable URL.
+        region_name=settings.DJANGO_OUTPUT_S3_AWS_DEFAULT_REGION or "us-east-1",
+        endpoint_url=endpoint or None,
+        # The signature covers the host, so the addressing style has to be pinned rather
+        # than guessed from the endpoint: MinIO serves path-style, and on AWS it keeps the
+        # regional host instead of the global one, which redirects for new buckets.
+        config=Config(signature_version="s3v4", s3={"addressing_style": "path"}),
+    )
+
+
 def _upload_to_s3(
    tenant_id: str, scan_id: str, local_path: str, relative_key: str
 ) -> str | None:
@@ -2,13 +2,16 @@ import os
 import time
 from datetime import UTC, datetime
 from glob import glob
+from urllib.parse import quote

 from api.db_router import READ_REPLICA_ALIAS, MainRouter
 from api.db_utils import REPLICA_MAX_ATTEMPTS, REPLICA_RETRY_BASE_DELAY, rls_transaction
 from api.models import Finding, Integration, Provider
+from api.rls import Tenant
 from api.utils import initialize_prowler_integration, initialize_prowler_provider
 from celery.utils.log import get_task_logger
 from config.django.base import DJANGO_FINDINGS_BATCH_SIZE
+from django.conf import settings
 from django.db import OperationalError
 from prowler.lib.outputs.asff.asff import ASFF
 from prowler.lib.outputs.compliance.generic.generic import GenericCompliance
@@ -16,6 +19,7 @@ from prowler.lib.outputs.csv.csv import CSV
 from prowler.lib.outputs.finding import Finding as FindingOutput
 from prowler.lib.outputs.html.html import HTML
 from prowler.lib.outputs.jira.exceptions.exceptions import JiraBaseException
+from prowler.lib.outputs.jira.jira import Jira
 from prowler.lib.outputs.ocsf.ocsf import OCSF
 from prowler.providers.aws.aws_provider import AwsProvider
 from prowler.providers.aws.lib.s3.s3 import S3
@@ -477,6 +481,55 @@ def upload_security_hub_integration(
        return False


+JIRA_LABEL_PREFIX = "prowler"
+
+
+def build_jira_finding_url(finding_uid: str) -> str:
+    """Build the Prowler UI link for a finding, or "" when no UI base URL is set.
+
+    The link filters by the finding ``uid`` rather than the per-scan record id so
+    it keeps resolving after the finding is seen again in later scans.
+    """
+    base_url = getattr(settings, "UI_BASE_URL", "")
+    if not base_url or not finding_uid:
+        return ""
+    return f"{base_url}/findings?filter[uid]={quote(finding_uid, safe='')}"
+
+
+def build_jira_issue_labels(
+    finding_uid: str, provider: str, severity: str, check_id: str
+) -> list[str]:
+    """Build the deterministic label set written to every Jira issue.
+
+    Labels are prefixed to avoid colliding with customer labels and sanitized so
+    Jira never rejects them; the finding-uid label is what lets a ticket be traced
+    back (or JQL-filtered) to its finding.
+    """
+    raw_labels = [
+        JIRA_LABEL_PREFIX,
+        f"{JIRA_LABEL_PREFIX}-{provider}" if provider else "",
+        f"{JIRA_LABEL_PREFIX}-{severity}" if severity else "",
+        f"{JIRA_LABEL_PREFIX}-{check_id}" if check_id else "",
+        Jira.build_finding_label(finding_uid),
+    ]
+    return Jira.sanitize_labels(raw_labels)
+
+
+def get_tenant_name(tenant_id: str) -> str:
+    """Return the tenant name for the Jira issue "Tenant Info" row, or "" if unknown.
+
+    The name is informational only, so a lookup failure must never block the send.
+    """
+    try:
+        return (
+            Tenant.objects.filter(id=tenant_id).values_list("name", flat=True).first()
+            or ""
+        )
+    except Exception:
+        logger.warning("Could not resolve tenant name for %s", tenant_id)
+        return ""
+
+
 def send_findings_to_jira(
    tenant_id: str,
    integration_id: str,
@@ -487,6 +540,7 @@ def send_findings_to_jira(
    with rls_transaction(tenant_id):
        integration = Integration.objects.get(id=integration_id)
        jira_integration = initialize_prowler_integration(integration)
+        tenant_info = get_tenant_name(tenant_id)

    num_tickets_created = 0
    error_messages = []
@@ -519,6 +573,15 @@ def send_findings_to_jira(
            recommendation = remediation.get("recommendation", {})
            remediation_code = remediation.get("code", {})

+            provider_type = finding_instance.scan.provider.provider
+            issue_labels = build_jira_issue_labels(
+                finding_uid=finding_instance.uid,
+                provider=provider_type,
+                severity=finding_instance.severity,
+                check_id=finding_instance.check_id,
+            )
+            finding_url = build_jira_finding_url(finding_instance.uid)
+
            try:
                # Send the individual finding to Jira
                result = jira_integration.send_finding(
@@ -527,7 +590,7 @@ def send_findings_to_jira(
                    severity=finding_instance.severity,
                    status=finding_instance.status,
                    status_extended=finding_instance.status_extended or "",
-                    provider=finding_instance.scan.provider.provider,
+                    provider=provider_type,
                    region=region,
                    resource_uid=resource_uid,
                    resource_name=resource_name,
@@ -542,6 +605,9 @@ def send_findings_to_jira(
                    compliance=finding_instance.compliance or {},
                    project_key=project_key,
                    issue_type=issue_type,
+                    issue_labels=issue_labels,
+                    finding_url=finding_url,
+                    tenant_info=tenant_info,
                )
            except JiraBaseException as error:
                error_message = error.message or JIRA_GENERIC_SEND_ERROR
@@ -557,6 +623,11 @@ def send_findings_to_jira(

            if result:
                num_tickets_created += 1
+                logger.info(
+                    "Finding %s sent to Jira as %s",
+                    finding_id,
+                    result.get("key") if isinstance(result, dict) else result,
+                )
            else:
                error_message = JIRA_GENERIC_SEND_ERROR
                logger.error(error_message)
@@ -1,63 +1,96 @@
+from collections.abc import Iterable
+
 from api.db_utils import rls_transaction
-from api.models import Finding, MuteRule
+from api.models import Finding, MuteRule, Scan
 from celery.utils.log import get_task_logger
-from config.django.base import DJANGO_FINDINGS_BATCH_SIZE
-from tasks.utils import batched

 logger = get_task_logger(__name__)


-def mute_historical_findings(tenant_id: str, mute_rule_id: str):
-    """
-    Mute historical findings that match the given mute rule.
+def _mute_findings_for_rule(
+    *,
+    tenant_id: str,
+    scan_id: str,
+    finding_uids: Iterable[str],
+    muted_at,
+    muted_reason: str,
+) -> int:
+    finding_uids = list(finding_uids)
+    if not finding_uids:
+        return 0

-    This function processes findings in batches, updating their muted status
-    and adding the mute reason.
+    return Finding.all_objects.filter(
+        tenant_id=tenant_id,
+        scan_id=scan_id,
+        uid__in=finding_uids,
+        muted=False,
+    ).update(
+        muted=True,
+        muted_at=muted_at,
+        muted_reason=muted_reason,
+    )

-    Args:
-        tenant_id (str): The tenant ID for RLS context
-        mute_rule_id (str): The ID of the mute rule to apply

-    Returns:
-        dict: Summary of the muting operation with findings_muted count
-    """
-    findings_muted_count = 0
+def mute_findings_in_latest_scans(
+    tenant_id: str, mute_rule_id: str, provider_ids: list[str]
+) -> dict:
+    """Apply a mute rule to the latest completed scan of each provider."""
+    provider_ids = list(dict.fromkeys(provider_ids))

-    # Get the list of UIDs to mute and the reason
    with rls_transaction(tenant_id):
        mute_rule = MuteRule.objects.get(id=mute_rule_id, tenant_id=tenant_id)
-        finding_uids = mute_rule.finding_uids
-        mute_reason = mute_rule.reason
-        muted_at = mute_rule.inserted_at
+        latest_scans = Scan.objects.filter(
+            tenant_id=tenant_id, provider_id__in=provider_ids
+        ).latest_ids_per_provider()

-    # Query findings that match the UIDs and are not already muted
-    with rls_transaction(tenant_id):
-        findings_to_mute = Finding.objects.filter(
-            tenant_id=tenant_id, uid__in=finding_uids, muted=False
-        )
-        total_findings = findings_to_mute.count()
-
-        logger.info(
-            f"Processing {total_findings} findings for mute rule {mute_rule_id}"
-        )
-
-        if total_findings > 0:
-            for batch, is_last in batched(
-                findings_to_mute.iterator(), DJANGO_FINDINGS_BATCH_SIZE
-            ):
-                batch_ids = [f.id for f in batch]
-                updated_count = Finding.all_objects.filter(
-                    id__in=batch_ids, tenant_id=tenant_id
-                ).update(
-                    muted=True,
-                    muted_at=muted_at,
-                    muted_reason=mute_reason,
-                )
-                findings_muted_count += updated_count
-
-    logger.info(f"Muted {findings_muted_count} findings for rule {mute_rule_id}")
+        changed_scan_ids = []
+        findings_muted = 0
+        for scan_id in latest_scans:
+            updated = _mute_findings_for_rule(
+                tenant_id=tenant_id,
+                scan_id=str(scan_id),
+                finding_uids=mute_rule.finding_uids,
+                muted_at=mute_rule.inserted_at,
+                muted_reason=mute_rule.reason,
+            )
+            if updated:
+                findings_muted += updated
+                changed_scan_ids.append(str(scan_id))

+    logger.info(
+        "Muted %d findings in %d latest scans for rule %s",
+        findings_muted,
+        len(changed_scan_ids),
+        mute_rule_id,
+    )
    return {
-        "findings_muted": findings_muted_count,
+        "findings_muted": findings_muted,
        "rule_id": mute_rule_id,
+        "scan_ids": changed_scan_ids,
    }
+
+
+def reconcile_scan_mute_rules(tenant_id: str, scan_id: str) -> dict:
+    """Apply the current enabled mute rules to one completed scan."""
+    findings_muted = 0
+
+    with rls_transaction(tenant_id):
+        mute_rules = MuteRule.objects.filter(tenant_id=tenant_id, enabled=True).values(
+            "finding_uids", "reason", "inserted_at"
+        )
+
+        for mute_rule in mute_rules:
+            findings_muted += _mute_findings_for_rule(
+                tenant_id=tenant_id,
+                scan_id=scan_id,
+                finding_uids=mute_rule["finding_uids"],
+                muted_at=mute_rule["inserted_at"],
+                muted_reason=mute_rule["reason"],
+            )
+
+    logger.info(
+        "Reconciled mute rules for scan %s; muted %d findings",
+        scan_id,
+        findings_muted,
+    )
+    return {"findings_muted": findings_muted, "scan_id": str(scan_id)}
@@ -84,6 +84,7 @@ _SKIP_RECOVERY = {
    "scan-perform-scheduled",
    "attack-paths-scan-perform",
    "attack-paths-cleanup-stale-scans",
+    "attack-paths-reap-orphaned-tmp-databases",
    "reconcile-orphan-tasks",
 }

@@ -172,14 +172,10 @@ class CISReportGenerator(BaseComplianceReportGenerator):

        # Metadata table via base class helper
        info_rows = self._build_info_rows(data, language=self.config.language)
-        metadata_data = []
-        for label, value in info_rows:
-            if label in ("Name:", "Description:") and value:
-                metadata_data.append(
-                    [label, Paragraph(str(value), self.styles["normal_center"])]
-                )
-            else:
-                metadata_data.append([label, value])
+        metadata_data = [
+            [label, Paragraph(escape_html(str(value)), self.styles["normal_center"])]
+            for label, value in info_rows
+        ]

        metadata_table = Table(metadata_data, colWidths=[2 * inch, 4 * inch])
        metadata_table.setStyle(
@@ -13,7 +13,7 @@ from .base import (
    get_requirement_metadata,
 )
 from .charts import create_horizontal_bar_chart, create_radar_chart
-from .components import get_color_for_compliance
+from .components import escape_html, get_color_for_compliance
 from .config import (
    COLOR_BG_BLUE,
    COLOR_BLUE,
@@ -99,15 +99,10 @@ class ENSReportGenerator(BaseComplianceReportGenerator):

        # Compliance info table - use base class helper for consistency
        info_rows = self._build_info_rows(data, language="es")
-        # Convert tuples to lists and wrap long text in Paragraphs
-        info_data = []
-        for label, value in info_rows:
-            if label in ("Nombre:", "Descripción:") and value:
-                info_data.append(
-                    [label, Paragraph(value, self.styles["normal_center"])]
-                )
-            else:
-                info_data.append([label, value])
+        info_data = [
+            [label, Paragraph(escape_html(str(value)), self.styles["normal_center"])]
+            for label, value in info_rows
+        ]

        info_table = Table(info_data, colWidths=[2 * inch, 4 * inch])
        info_table.setStyle(
@@ -11,6 +11,7 @@ from .base import (
    get_requirement_metadata,
 )
 from .charts import create_horizontal_bar_chart, get_chart_color_for_percentage
+from .components import escape_html
 from .config import (
    COLOR_BORDER_GRAY,
    COLOR_DARK_GRAY,
@@ -111,15 +112,10 @@ class NIS2ReportGenerator(BaseComplianceReportGenerator):

        # Compliance metadata table - use base class helper for consistency
        info_rows = self._build_info_rows(data, language="en")
-        # Convert tuples to lists and wrap long text in Paragraphs
-        metadata_data = []
-        for label, value in info_rows:
-            if label in ("Name:", "Description:") and value:
-                metadata_data.append(
-                    [label, Paragraph(value, self.styles["normal_center"])]
-                )
-            else:
-                metadata_data.append([label, value])
+        metadata_data = [
+            [label, Paragraph(escape_html(str(value)), self.styles["normal_center"])]
+            for label, value in info_rows
+        ]

        metadata_table = Table(metadata_data, colWidths=[2 * inch, 4 * inch])
        metadata_table.setStyle(
@@ -5,7 +5,6 @@ import json
 import random
 import re
 import time
-import uuid
 from collections import defaultdict
 from collections.abc import Callable, Iterable
 from datetime import UTC, datetime
@@ -73,6 +72,7 @@ from tasks.jobs.queries import (
    COMPLIANCE_UPSERT_TENANT_SUMMARY_SQL,
 )
 from tasks.utils import CustomEncoder, batched
+from uuid6 import uuid7

 logger = get_task_logger(__name__)

@@ -997,11 +997,12 @@ def _process_finding_micro_batch(
                        snapshot_resource_fields(resource_uid, resource_instance)
                        resource_instance.type = finding.resource_type
                        updated = True
-                    if resource_instance.metadata != finding.resource_metadata:
+                    serialized_metadata = json.dumps(
+                        finding.resource_metadata, cls=CustomEncoder
+                    )
+                    if resource_instance.metadata != serialized_metadata:
                        snapshot_resource_fields(resource_uid, resource_instance)
-                        resource_instance.metadata = json.dumps(
-                            finding.resource_metadata, cls=CustomEncoder
-                        )
+                        resource_instance.metadata = serialized_metadata
                        updated = True
                    if resource_instance.details != finding.resource_details:
                        snapshot_resource_fields(resource_uid, resource_instance)
@@ -1756,32 +1757,27 @@ def aggregate_findings(tenant_id: str, scan_id: str):


 def _aggregate_findings_by_region(
-    tenant_id: str, scan_id: str, modeled_threatscore_compliance_id: str
+    tenant_id: str,
+    scan_id: str,
+    normalized_threatscore_id: str,
+    threatscore_requirements_by_check: dict[str, list[str]],
 ) -> tuple[dict, dict]:
    """
    Aggregate findings by region using streaming, column-scoped ORM reads.

    Reads only the consumed columns as tuples via ``values_list`` and streams
    them with ``.iterator()``, using the denormalized ``resource_regions`` array
-    instead of ``prefetch_related("resources")``. ``resource_regions`` mirrors the
-    regions of a finding's related resources, so it yields the same per-region
-    tally without joining the resource table.
-
-    Args:
-        tenant_id: Tenant UUID
-        scan_id: Scan UUID
-        modeled_threatscore_compliance_id: ID for ThreatScore compliance framework
+    instead of ``prefetch_related("resources")``. ThreatScore requirement ids
+    are resolved per ``check_id`` from ``threatscore_requirements_by_check``.

    Returns:
        tuple: (check_status_by_region, findings_count_by_compliance)
            - check_status_by_region: {region: {check_id: status}}
-            - findings_count_by_compliance: {region: {normalized_id: {requirement_id: {total, pass}}}}
+            - findings_count_by_compliance: {region: {normalized_threatscore_id: {requirement_id: {total, pass}}}}
    """
    check_status_by_region: dict = {}
    findings_count_by_compliance: dict = {}

-    normalized_id = re.sub(r"[^a-z0-9]", "", modeled_threatscore_compliance_id.lower())
-
    with rls_transaction(tenant_id, using=READ_REPLICA_ALIAS):
        findings = (
            Finding.all_objects.filter(
@@ -1790,14 +1786,12 @@ def _aggregate_findings_by_region(
                muted=False,
                status__in=["PASS", "FAIL"],
            )
-            .values_list("check_id", "status", "resource_regions", "compliance")
+            .values_list("check_id", "status", "resource_regions")
            .iterator(chunk_size=DJANGO_FINDINGS_BATCH_SIZE)
        )

-        for check_id, status, resource_regions, compliance in findings:
-            threatscore_requirements = (compliance or {}).get(
-                modeled_threatscore_compliance_id
-            )
+        for check_id, status, resource_regions in findings:
+            threatscore_requirements = threatscore_requirements_by_check.get(check_id)

            for region in resource_regions or ():
                # Priority: FAIL > any other status
@@ -1809,7 +1803,7 @@ def _aggregate_findings_by_region(
                if threatscore_requirements:
                    compliance_key = findings_count_by_compliance.setdefault(
                        region, {}
-                    ).setdefault(normalized_id, {})
+                    ).setdefault(normalized_threatscore_id, {})

                    for requirement_id in threatscore_requirements:
                        requirement_stats = compliance_key.setdefault(
@@ -1848,15 +1842,28 @@ def create_compliance_requirements(tenant_id: str, scan_id: str):
        compliance_template = PROWLER_COMPLIANCE_OVERVIEW_TEMPLATE[
            provider_instance.provider
        ]
-        modeled_threatscore_compliance_id = "ProwlerThreatScore-1.0"
+        normalized_threatscore_id = _normalized_compliance_key(
+            "ProwlerThreatScore", "1.0"
+        )

        requirement_lookup: dict[str, list[tuple[str, str]]] = {}
+        threatscore_requirements_by_check: dict[str, list[str]] = {}
        for compliance_id, compliance in compliance_template.items():
+            is_threatscore = (
+                _normalized_compliance_key(
+                    compliance["framework"], compliance["version"]
+                )
+                == normalized_threatscore_id
+            )
            for requirement_id, requirement in compliance["requirements"].items():
                for check_id in requirement["checks"].keys():
                    requirement_lookup.setdefault(check_id, []).append(
                        (compliance_id, requirement_id)
                    )
+                    if is_threatscore:
+                        threatscore_requirements_by_check.setdefault(
+                            check_id, []
+                        ).append(requirement_id)

        regions = []
        requirements_created = 0
@@ -1869,7 +1876,10 @@ def create_compliance_requirements(tenant_id: str, scan_id: str):
            # Aggregate findings by region using SQL for optimal performance
            check_status_by_region, findings_count_by_compliance = (
                _aggregate_findings_by_region(
-                    tenant_id, scan_id, modeled_threatscore_compliance_id
+                    tenant_id,
+                    scan_id,
+                    normalized_threatscore_id,
+                    threatscore_requirements_by_check,
                )
            )

@@ -1934,23 +1944,35 @@ def create_compliance_requirements(tenant_id: str, scan_id: str):
            # Yield rows lazily (consumed batch-by-batch by COPY) so peak memory
            # stays bounded; tally requirement_statuses in the same pass. The
            # ORM fallback re-iterates from scratch, so the tally resets first.
+            # Region is the innermost loop so consecutive rows share the leading
+            # columns of the table's secondary indexes.
            def _iter_compliance_requirement_rows():
                requirement_statuses.clear()
-                for region in regions:
-                    region_stats = region_requirement_stats.get(region, {})
-                    region_findings = findings_count_by_compliance.get(region, {})
-                    for (
-                        compliance_id,
-                        framework,
-                        version,
-                        modeled_compliance_id,
-                        requirements,
-                    ) in compliance_plan:
-                        compliance_stats = region_stats.get(compliance_id, {})
-                        compliance_findings = region_findings.get(
-                            modeled_compliance_id, {}
+                for (
+                    compliance_id,
+                    framework,
+                    version,
+                    modeled_compliance_id,
+                    requirements,
+                ) in compliance_plan:
+                    stats_by_region = [
+                        (
+                            region,
+                            region_requirement_stats.get(region, {}).get(
+                                compliance_id, {}
+                            ),
+                            findings_count_by_compliance.get(region, {}).get(
+                                modeled_compliance_id, {}
+                            ),
                        )
-                        for requirement_id, description, total_checks in requirements:
+                        for region in regions
+                    ]
+                    for requirement_id, description, total_checks in requirements:
+                        for (
+                            region,
+                            compliance_stats,
+                            compliance_findings,
+                        ) in stats_by_region:
                            stats = compliance_stats.get(requirement_id)
                            if stats:
                                passed_checks = stats["passed_checks"]
@@ -1981,7 +2003,7 @@ def create_compliance_requirements(tenant_id: str, scan_id: str):
                                requirement_statuses[key]["pass_count"] += 1

                            yield {
-                                "id": uuid.uuid4(),
+                                "id": uuid7(),
                                "tenant_id": tenant_id_str,
                                "inserted_at": utc_datetime_now,
                                "compliance_id": compliance_id,
@@ -2092,7 +2114,9 @@ def aggregate_attack_surface(tenant_id: str, scan_id: str):
            .annotate(
                total=Count("id"),
                failed=Count("id", filter=Q(status="FAIL", muted=False)),
-                muted=Count("id", filter=Q(status="FAIL", muted=True)),
+                # Not `muted`: an annotation named after a model field comes
+                # back as `muted_new` from the psqlextra queryset.
+                muted_count=Count("id", filter=Q(status="FAIL", muted=True)),
            )
        )

@@ -2103,7 +2127,7 @@ def aggregate_attack_surface(tenant_id: str, scan_id: str):

            aggregated_counts[attack_surface_type]["total"] += stats["total"] or 0
            aggregated_counts[attack_surface_type]["failed"] += stats["failed"] or 0
-            aggregated_counts[attack_surface_type]["muted"] += stats["muted"] or 0
+            aggregated_counts[attack_surface_type]["muted"] += stats["muted_count"] or 0

    overview_objects = []
    for attack_surface_type, counts in aggregated_counts.items():
@@ -2679,20 +2703,37 @@ def reset_ephemeral_resource_findings_count(tenant_id: str, scan_id: str) -> dic
    # refreshed). Wiping based on the older scan would zero counts the newer
    # scan just set. Skip and let the newer scan's reset task do the work; if
    # this task was delayed in the queue, that's the correct outcome.
-    # `completed_at__isnull=False` is required: Postgres orders NULL first in
-    # DESC, so a sibling COMPLETED scan with a missing completed_at would sort
-    # as "newest" and incorrectly cause us to skip.
+    #
+    # The comparison must be against the newest *full-scope* scan, which is
+    # what this variable has always been named after but did not use to be:
+    # the query filtered nothing about scope, so any newer scan that is not
+    # full-scope (an imported one, for instance) made the full-scope scan
+    # skip its own cleanup and leave ephemeral resources with a stale
+    # failed_findings_count permanently.
+    #
+    # `is_full_scope()` reads `trigger` plus the scoping keys inside
+    # `scanner_args`, which is not expressible as a WHERE clause, so the
+    # candidates are walked newest-first in Python until the first full-scope
+    # one. The walk needs no cap: `scan` is itself a full-scope candidate, so
+    # it stops at `scan` at the latest, after reading only the scans newer than
+    # it. A fixed window would return None once more newer scoped scans had
+    # landed than it inspected, and skip the cleanup exactly like the bug above.
+    #
+    # NULL `completed_at` no longer needs an explicit filter here: the shared
+    # ordering in `ScanQuerySet.LATEST_ORDER_BY` sorts NULLs last
+    # rather than excluding them, which also fixes the case where a provider
+    # whose completed scans all have a NULL `completed_at` resolved to None and
+    # therefore never ran the reset at all.
    with rls_transaction(tenant_id):
-        latest_full_scope_scan_id = (
-            Scan.objects.filter(
-                tenant_id=tenant_id,
-                provider_id=scan.provider_id,
-                state=StateChoices.COMPLETED,
-                completed_at__isnull=False,
-            )
-            .order_by("-completed_at", "-inserted_at")
-            .values_list("id", flat=True)
-            .first()
+        candidates = (
+            Scan.objects.filter(tenant_id=tenant_id, provider_id=scan.provider_id)
+            .latest_first()
+            .only("id", "trigger", "scanner_args")
+            .iterator(chunk_size=100)
+        )
+        latest_full_scope_scan_id = next(
+            (candidate.id for candidate in candidates if candidate.is_full_scope()),
+            None,
        )
    if latest_full_scope_scan_id != scan.id:
        logger.info(
@@ -1,3 +1,4 @@
+import json
 import os
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
@@ -42,6 +43,7 @@ from tasks.jobs.attack_paths import (
 )
 from tasks.jobs.attack_paths import db_utils as attack_paths_db_utils
 from tasks.jobs.attack_paths.cleanup import cleanup_stale_attack_paths_scans
+from tasks.jobs.attack_paths.tmp_db_reaper import reap_orphaned_tmp_databases
 from tasks.jobs.backfill import (
    aggregate_scan_category_summaries,
    aggregate_scan_resource_group_summaries,
@@ -73,7 +75,10 @@ from tasks.jobs.lighthouse_providers import (
    check_lighthouse_provider_connection,
    refresh_lighthouse_provider_models,
 )
-from tasks.jobs.muting import mute_historical_findings
+from tasks.jobs.muting import (
+    mute_findings_in_latest_scans,
+    reconcile_scan_mute_rules,
+)
 from tasks.jobs.orphan_recovery import reconcile_orphans
 from tasks.jobs.report import (
    STALE_TMP_OUTPUT_MAX_AGE_HOURS,
@@ -160,13 +165,28 @@ def create_scan_task_record(
    task_id: str,
    task_name: str = "scan-perform",
    task_status: str | None = states.PENDING,
+    task_kwargs: dict | None = None,
 ) -> Task:
+    """Pre-create the TaskResult + Task rows for a pre-generated task id.
+
+    Pass ``task_kwargs`` when the response built from this record is serialized
+    before the broker publish. ``task_kwargs`` is otherwise only written by the
+    ``before_task_publish`` signal (``api/signals.py``), and the scan publish is
+    deferred to ``on_commit``, so the 202 would carry an empty ``task_args`` and
+    the caller would have no way to learn the scan id it was just handed a task
+    for. The publish later overwrites the field with the same kwargs as a Python
+    repr; both forms decode to the same dict (``decode_celery_field``).
+    """
    if task_status is None:
        task_status = states.PENDING

+    defaults = {"status": task_status, "task_name": task_name}
+    if task_kwargs is not None:
+        defaults["task_kwargs"] = json.dumps(task_kwargs)
+
    task_result, _ = TaskResult.objects.update_or_create(
        task_id=str(task_id),
-        defaults={"status": task_status, "task_name": task_name},
+        defaults=defaults,
    )
    prowler_task, _ = Task.objects.update_or_create(
        id=str(task_id),
@@ -526,6 +546,7 @@ def perform_scan_task(
            provider_id=provider_id,
            checks_to_execute=checks_to_execute,
        )
+        reconcile_scan_mute_rules(tenant_id, scan_id)
        _perform_scan_complete_tasks(tenant_id, scan_id, provider_id)
        return result
    finally:
@@ -635,6 +656,7 @@ def perform_scheduled_scan_task(self, tenant_id: str, provider_id: str):
            scan_id=str(scan_instance.id),
            provider_id=provider_id,
        )
+        reconcile_scan_mute_rules(tenant_id, str(scan_instance.id))
        _perform_scan_complete_tasks(tenant_id, str(scan_instance.id), provider_id)
        return result
    finally:
@@ -706,6 +728,13 @@ def cleanup_stale_attack_paths_scans_task():
    return cleanup_stale_attack_paths_scans()


+@shared_task(
+    name="attack-paths-reap-orphaned-tmp-databases", queue="attack-paths-scans"
+)
+def reap_orphaned_attack_paths_tmp_databases_task():
+    return reap_orphaned_tmp_databases()
+
+
@shared_task(name="reconcile-orphan-tasks", queue="celery")
 def reconcile_orphan_tasks_task():
    """Periodic watchdog: recover tasks whose worker is gone (deploys, crashes)."""
@@ -868,8 +897,12 @@ def generate_outputs_task(scan_id: str, provider_id: str, tenant_id: str):
    )
    with rls_transaction(tenant_id, using=READ_REPLICA_ALIAS):
        for batch, is_last in batched(qs, DJANGO_FINDINGS_BATCH_SIZE):
+            metadata_cache = {}
            fos = [
-                FindingOutput.transform_api_finding(f, prowler_provider) for f in batch
+                FindingOutput.transform_api_finding(
+                    f, prowler_provider, metadata_cache=metadata_cache
+                )
+                for f in batch
            ]

            # Outputs
@@ -950,6 +983,8 @@ def generate_outputs_task(scan_id: str, provider_id: str, tenant_id: str):
                writer.batch_write_data_to_file()
                writer._data.clear()

+            del fos, metadata_cache
+
    compressed = _compress_output_files(out_dir)

    upload_uri = _upload_to_s3(
@@ -1188,85 +1223,48 @@ def aggregate_finding_group_summaries_task(tenant_id: str, scan_id: str):
    return aggregate_finding_group_summaries(tenant_id=tenant_id, scan_id=scan_id)


-@shared_task(
-    base=RLSTask, name="reaggregate-all-finding-group-summaries", queue="overview"
-)
-@set_tenant(keep_tenant=True)
-def reaggregate_all_finding_group_summaries_task(tenant_id: str):
-    """Reaggregate every pre-aggregated summary table for this tenant.
+def _dispatch_scan_summary_reaggregation(tenant_id: str, scan_ids: list[str]) -> None:
+    if not scan_ids:
+        return

-    Mirrors the unbounded scope of `mute_historical_findings_task`: that task
-    rewrites every Finding row whose UID matches a mute rule, with no time
-    limit. To keep the pre-aggregated tables consistent with that update,
-    this task re-runs the same per-scan aggregation pipeline that scan
-    completion runs on the latest completed scan of every (provider, day)
-    pair, rebuilding the tables that power the read endpoints:
-
-      - `ScanSummary` and `DailySeveritySummary` -> `/overviews/findings`,
-        `/overviews/findings-severity`, `/overviews/services`.
-      - `FindingGroupDailySummary` -> `/finding-groups` and
-        `/finding-groups/latest`.
-      - `ScanGroupSummary` -> `/overviews/resource-groups` (resource
-        inventory).
-      - `ScanCategorySummary` -> `/overviews/categories`.
-      - `AttackSurfaceOverview` -> `/overviews/attack-surfaces`.
-
-    Per-scan pipelines are dispatched in parallel via a Celery group so
-    wallclock scales with the worker pool.
-    """
-    completed_scans = list(
-        Scan.objects.filter(
-            tenant_id=tenant_id,
-            state=StateChoices.COMPLETED,
-            completed_at__isnull=False,
-        )
-        .order_by("-completed_at")
-        .values("id", "completed_at", "provider_id")
+    logger.info(
+        "Reaggregating overview/finding summaries for %d latest scans",
+        len(scan_ids),
    )
-
-    # Keep the latest scan per (provider, day) pair so the daily summary row
-    # the aggregator writes is the most recent snapshot of that day for that
-    # provider. Iterating from most recent to oldest means the first scan we
-    # see for a given key wins.
-    latest_scans: dict[tuple, str] = {}
-    for scan in completed_scans:
-        key = (scan["provider_id"], scan["completed_at"].date())
-        if key not in latest_scans:
-            latest_scans[key] = str(scan["id"])
-
-    scan_ids = list(latest_scans.values())
-    if scan_ids:
-        logger.info(
-            "Reaggregating overview/finding summaries for %d scans (provider x day)",
-            len(scan_ids),
-        )
-        # DailySeveritySummary reads from ScanSummary, so ScanSummary must be
-        # recomputed first; the other aggregators read Finding directly and
-        # can run in parallel with the severity step.
-        group(
-            chain(
-                perform_scan_summary_task.si(tenant_id=tenant_id, scan_id=scan_id),
-                group(
-                    aggregate_daily_severity_task.si(
-                        tenant_id=tenant_id, scan_id=scan_id
-                    ),
-                    aggregate_finding_group_summaries_task.si(
-                        tenant_id=tenant_id, scan_id=scan_id
-                    ),
-                    aggregate_scan_resource_group_summaries_task.si(
-                        tenant_id=tenant_id, scan_id=scan_id
-                    ),
-                    aggregate_scan_category_summaries_task.si(
-                        tenant_id=tenant_id, scan_id=scan_id
-                    ),
-                    aggregate_attack_surface_task.si(
-                        tenant_id=tenant_id, scan_id=scan_id
-                    ),
+    group(
+        chain(
+            perform_scan_summary_task.si(tenant_id=tenant_id, scan_id=scan_id),
+            group(
+                aggregate_daily_severity_task.si(tenant_id=tenant_id, scan_id=scan_id),
+                aggregate_finding_group_summaries_task.si(
+                    tenant_id=tenant_id, scan_id=scan_id
                ),
-            )
-            for scan_id in scan_ids
-        ).apply_async()
-    return {"scans_reaggregated": len(scan_ids)}
+                aggregate_scan_resource_group_summaries_task.si(
+                    tenant_id=tenant_id, scan_id=scan_id
+                ),
+                aggregate_scan_category_summaries_task.si(
+                    tenant_id=tenant_id, scan_id=scan_id
+                ),
+                aggregate_attack_surface_task.si(tenant_id=tenant_id, scan_id=scan_id),
+            ),
+        )
+        for scan_id in scan_ids
+    ).apply_async()
+
+
+@shared_task(base=RLSTask, name="findings-mute-latest-scans", queue="overview")
+@set_tenant(keep_tenant=True)
+def mute_findings_in_latest_scans_task(
+    tenant_id: str, mute_rule_id: str, provider_ids: list[str]
+):
+    """Apply a mute rule to current scans and rebuild only changed summaries."""
+    result = mute_findings_in_latest_scans(
+        tenant_id=tenant_id,
+        mute_rule_id=mute_rule_id,
+        provider_ids=provider_ids,
+    )
+    _dispatch_scan_summary_reaggregation(tenant_id, result["scan_ids"])
+    return result


@shared_task(base=RLSTask, name="lighthouse-connection-check")
@@ -1467,25 +1465,3 @@ def generate_compliance_reports_task(tenant_id: str, scan_id: str, provider_id:
        generate_csa=True,
        generate_cis=True,
    )
-
-
-@shared_task(name="findings-mute-historical")
-def mute_historical_findings_task(tenant_id: str, mute_rule_id: str):
-    """
-    Background task to mute all historical findings matching a mute rule.
-
-    This task processes findings in batches to avoid memory issues with large datasets.
-    It updates the Finding.muted, Finding.muted_at, and Finding.muted_reason fields
-    for all findings whose UID is in the mute rule's finding_uids list.
-
-    Args:
-        tenant_id (str): The tenant ID for RLS context.
-        mute_rule_id (str): The primary key of the MuteRule to apply.
-
-    Returns:
-        dict: A dictionary containing:
-            - 'findings_muted' (int): Total number of findings muted.
-            - 'rule_id' (str): The mute rule ID.
-            - 'status' (str): Final status ('completed').
-    """
-    return mute_historical_findings(tenant_id, mute_rule_id)
@@ -0,0 +1,286 @@
+from datetime import UTC, datetime, timedelta
+from unittest.mock import patch
+from uuid import uuid4
+
+import pytest
+from api.attack_paths.database import TEMP_DB_PREFIX
+from api.models import AttackPathsScan, StateChoices
+from api.uuid_utils import datetime_to_uuid7
+from config.django.base import ATTACK_PATHS_TMP_DB_REAP_SAFETY_MARGIN_HOURS
+
+MARGIN = timedelta(hours=ATTACK_PATHS_TMP_DB_REAP_SAFETY_MARGIN_HOURS)
+
+
+def _tmp_db_name(scan_uuid) -> str:
+    return f"{TEMP_DB_PREFIX}{scan_uuid}"
+
+
+@pytest.mark.django_db
+class TestReapOrphanedTmpDatabases:
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.drop_database")
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.list_databases")
+    def test_ignores_databases_without_the_temp_prefix(self, mock_list, mock_drop):
+        from tasks.jobs.attack_paths.tmp_db_reaper import reap_orphaned_tmp_databases
+
+        mock_list.return_value = ["db-tenant-abc123", "system", "neo4j"]
+
+        result = reap_orphaned_tmp_databases()
+
+        assert result == {"dropped_count": 0, "databases": []}
+        mock_drop.assert_not_called()
+
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.drop_database")
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.list_databases")
+    def test_drops_temp_db_with_no_scan_row_past_safety_margin(
+        self, mock_list, mock_drop
+    ):
+        from tasks.jobs.attack_paths.tmp_db_reaper import reap_orphaned_tmp_databases
+
+        old_scan_id = datetime_to_uuid7(
+            datetime.now(tz=UTC) - MARGIN - timedelta(hours=1)
+        )
+        database = _tmp_db_name(old_scan_id)
+        mock_list.return_value = [database]
+
+        result = reap_orphaned_tmp_databases()
+
+        assert result == {"dropped_count": 1, "databases": [database]}
+        mock_drop.assert_called_once_with(database)
+
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.drop_database")
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.list_databases")
+    def test_preserves_temp_db_with_no_scan_row_inside_safety_margin(
+        self, mock_list, mock_drop
+    ):
+        from tasks.jobs.attack_paths.tmp_db_reaper import reap_orphaned_tmp_databases
+
+        recent_scan_id = datetime_to_uuid7(datetime.now(tz=UTC) - timedelta(minutes=5))
+        database = _tmp_db_name(recent_scan_id)
+        mock_list.return_value = [database]
+
+        result = reap_orphaned_tmp_databases()
+
+        assert result == {"dropped_count": 0, "databases": []}
+        mock_drop.assert_not_called()
+
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.drop_database")
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.list_databases")
+    def test_preserves_temp_db_with_unparseable_scan_id(self, mock_list, mock_drop):
+        from tasks.jobs.attack_paths.tmp_db_reaper import reap_orphaned_tmp_databases
+
+        database = f"{TEMP_DB_PREFIX}not-a-uuid"
+        mock_list.return_value = [database]
+
+        result = reap_orphaned_tmp_databases()
+
+        assert result == {"dropped_count": 0, "databases": []}
+        mock_drop.assert_not_called()
+
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.drop_database")
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.list_databases")
+    def test_drops_terminal_scan_past_safety_margin(
+        self, mock_list, mock_drop, tenants_fixture, aws_provider
+    ):
+        from tasks.jobs.attack_paths.tmp_db_reaper import reap_orphaned_tmp_databases
+
+        tenant = tenants_fixture[0]
+        old_updated_at = datetime.now(tz=UTC) - MARGIN - timedelta(hours=1)
+        scan = AttackPathsScan.objects.create(
+            tenant_id=tenant.id,
+            provider=aws_provider,
+            state=StateChoices.COMPLETED,
+        )
+        AttackPathsScan.objects.filter(id=scan.id).update(updated_at=old_updated_at)
+
+        database = _tmp_db_name(scan.id)
+        mock_list.return_value = [database]
+
+        result = reap_orphaned_tmp_databases()
+
+        assert result == {"dropped_count": 1, "databases": [database]}
+        mock_drop.assert_called_once_with(database)
+
+    @pytest.mark.parametrize(
+        "state",
+        [StateChoices.FAILED, StateChoices.CANCELLED],
+    )
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.drop_database")
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.list_databases")
+    def test_drops_other_terminal_states_past_safety_margin(
+        self, mock_list, mock_drop, tenants_fixture, aws_provider, state
+    ):
+        from tasks.jobs.attack_paths.tmp_db_reaper import reap_orphaned_tmp_databases
+
+        tenant = tenants_fixture[0]
+        old_updated_at = datetime.now(tz=UTC) - MARGIN - timedelta(hours=1)
+        scan = AttackPathsScan.objects.create(
+            tenant_id=tenant.id,
+            provider=aws_provider,
+            state=state,
+        )
+        AttackPathsScan.objects.filter(id=scan.id).update(updated_at=old_updated_at)
+
+        database = _tmp_db_name(scan.id)
+        mock_list.return_value = [database]
+
+        result = reap_orphaned_tmp_databases()
+
+        assert result["dropped_count"] == 1
+        mock_drop.assert_called_once_with(database)
+
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.drop_database")
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.list_databases")
+    def test_preserves_terminal_scan_inside_safety_margin(
+        self, mock_list, mock_drop, tenants_fixture, aws_provider
+    ):
+        from tasks.jobs.attack_paths.tmp_db_reaper import reap_orphaned_tmp_databases
+
+        tenant = tenants_fixture[0]
+        scan = AttackPathsScan.objects.create(
+            tenant_id=tenant.id,
+            provider=aws_provider,
+            state=StateChoices.COMPLETED,
+        )
+
+        database = _tmp_db_name(scan.id)
+        mock_list.return_value = [database]
+
+        result = reap_orphaned_tmp_databases()
+
+        assert result == {"dropped_count": 0, "databases": []}
+        mock_drop.assert_not_called()
+
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.drop_database")
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.list_databases")
+    def test_margin_counts_from_completed_at_not_updated_at(
+        self, mock_list, mock_drop, tenants_fixture, aws_provider
+    ):
+        from tasks.jobs.attack_paths.tmp_db_reaper import reap_orphaned_tmp_databases
+
+        tenant = tenants_fixture[0]
+        old = datetime.now(tz=UTC) - MARGIN - timedelta(hours=1)
+        scan = AttackPathsScan.objects.create(
+            tenant_id=tenant.id,
+            provider=aws_provider,
+            state=StateChoices.COMPLETED,
+        )
+        AttackPathsScan.objects.filter(id=scan.id).update(
+            updated_at=old, completed_at=datetime.now(tz=UTC)
+        )
+
+        database = _tmp_db_name(scan.id)
+        mock_list.return_value = [database]
+
+        result = reap_orphaned_tmp_databases()
+
+        assert result == {"dropped_count": 0, "databases": []}
+        mock_drop.assert_not_called()
+
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.drop_database")
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.list_databases")
+    def test_drops_scan_completed_past_safety_margin(
+        self, mock_list, mock_drop, tenants_fixture, aws_provider
+    ):
+        from tasks.jobs.attack_paths.tmp_db_reaper import reap_orphaned_tmp_databases
+
+        tenant = tenants_fixture[0]
+        old = datetime.now(tz=UTC) - MARGIN - timedelta(hours=1)
+        scan = AttackPathsScan.objects.create(
+            tenant_id=tenant.id,
+            provider=aws_provider,
+            state=StateChoices.FAILED,
+        )
+        AttackPathsScan.objects.filter(id=scan.id).update(
+            updated_at=old, completed_at=old
+        )
+
+        database = _tmp_db_name(scan.id)
+        mock_list.return_value = [database]
+
+        result = reap_orphaned_tmp_databases()
+
+        assert result == {"dropped_count": 1, "databases": [database]}
+        mock_drop.assert_called_once_with(database)
+
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.drop_database")
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.list_databases")
+    def test_never_drops_an_executing_scan_regardless_of_age(
+        self, mock_list, mock_drop, tenants_fixture, aws_provider
+    ):
+        from tasks.jobs.attack_paths.tmp_db_reaper import reap_orphaned_tmp_databases
+
+        tenant = tenants_fixture[0]
+        very_old = datetime.now(tz=UTC) - timedelta(days=30)
+        scan = AttackPathsScan.objects.create(
+            tenant_id=tenant.id,
+            provider=aws_provider,
+            state=StateChoices.EXECUTING,
+        )
+        AttackPathsScan.objects.filter(id=scan.id).update(updated_at=very_old)
+
+        database = _tmp_db_name(scan.id)
+        mock_list.return_value = [database]
+
+        result = reap_orphaned_tmp_databases()
+
+        assert result == {"dropped_count": 0, "databases": []}
+        mock_drop.assert_not_called()
+
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.drop_database")
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.list_databases")
+    def test_one_failed_drop_does_not_stop_the_rest_of_the_sweep(
+        self, mock_list, mock_drop
+    ):
+        from tasks.jobs.attack_paths.tmp_db_reaper import reap_orphaned_tmp_databases
+
+        old_time = datetime.now(tz=UTC) - MARGIN - timedelta(hours=1)
+        failing_scan_id = datetime_to_uuid7(old_time)
+        succeeding_scan_id = datetime_to_uuid7(old_time)
+        failing_db = _tmp_db_name(failing_scan_id)
+        succeeding_db = _tmp_db_name(succeeding_scan_id)
+        mock_list.return_value = [failing_db, succeeding_db]
+        mock_drop.side_effect = [Exception("boom"), None]
+
+        result = reap_orphaned_tmp_databases()
+
+        assert result == {"dropped_count": 1, "databases": [succeeding_db]}
+        assert mock_drop.call_count == 2
+
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.list_databases")
+    def test_returns_empty_result_when_listing_databases_fails(self, mock_list):
+        from tasks.jobs.attack_paths.tmp_db_reaper import reap_orphaned_tmp_databases
+
+        mock_list.side_effect = Exception("neo4j unreachable")
+
+        result = reap_orphaned_tmp_databases()
+
+        assert result == {"dropped_count": 0, "databases": []}
+
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.drop_database")
+    @patch("tasks.jobs.attack_paths.tmp_db_reaper.graph_database.list_databases")
+    def test_preserves_temp_db_with_random_uuid_and_no_row(self, mock_list, mock_drop):
+        """A non-UUIDv7 id with no matching row has no reliable timestamp, so it
+        must be left alone rather than guessed at."""
+        from tasks.jobs.attack_paths.tmp_db_reaper import reap_orphaned_tmp_databases
+
+        database = _tmp_db_name(uuid4())
+        mock_list.return_value = [database]
+
+        result = reap_orphaned_tmp_databases()
+
+        assert result == {"dropped_count": 0, "databases": []}
+        mock_drop.assert_not_called()
+
+
+class TestReapOrphanedTmpDatabasesTask:
+    @patch(
+        "tasks.tasks.reap_orphaned_tmp_databases",
+        return_value={"dropped_count": 2, "databases": ["db-tmp-scan-a"]},
+    )
+    def test_task_invokes_the_reaper(self, mock_reap):
+        from tasks.tasks import reap_orphaned_attack_paths_tmp_databases_task
+
+        result = reap_orphaned_attack_paths_tmp_databases_task.run()
+
+        assert result == {"dropped_count": 2, "databases": ["db-tmp-scan-a"]}
+        mock_reap.assert_called_once_with()
@@ -3,7 +3,8 @@ from datetime import UTC, datetime
 from unittest.mock import MagicMock, patch

 import pytest
-from api.models import Integration, LighthouseConfiguration, Provider
+from api.models import Integration, LighthouseConfiguration, Provider, ProviderSecret
+from api.v1.serializers import KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR
 from tasks.jobs.connection import (
    check_integration_connection,
    check_lighthouse_connection,
@@ -65,17 +66,52 @@ def test_check_provider_connection_exception(

    mock_provider_connection_test.return_value = MagicMock()
    mock_provider_connection_test.return_value.is_connected = False
-    mock_provider_connection_test.return_value.error = Exception()
+    mock_provider_connection_test.return_value.error = Exception(
+        "Unable to assume role"
+    )

    result = check_provider_connection(provider_id="provider_id")

    assert result["connected"] is False
-    assert result["error"] is not None
+    assert result["error"] == "Unable to assume role"

    mock_provider_instance.save.assert_called_once()
    assert mock_provider_instance.connected is False


+@pytest.mark.django_db
+def test_check_provider_connection_with_stored_non_inline_kubeconfig(
+    kubernetes_provider,
+):
+    """A kubeconfig stored before the inline-only rule marks the provider as disconnected."""
+    kubernetes_provider.connected = True
+    kubernetes_provider.save()
+    ProviderSecret.objects.create(
+        tenant_id=kubernetes_provider.tenant_id,
+        provider=kubernetes_provider,
+        secret_type=ProviderSecret.TypeChoices.STATIC,
+        secret={
+            "kubeconfig_content": (
+                "apiVersion: v1\n"
+                "kind: Config\n"
+                "users:\n"
+                "- name: test-user\n"
+                "  user:\n"
+                "    tokenFile: /etc/passwd\n"
+            )
+        },
+    )
+
+    result = check_provider_connection(provider_id=str(kubernetes_provider.id))
+
+    kubernetes_provider.refresh_from_db()
+    assert result == {
+        "connected": False,
+        "error": KUBERNETES_KUBECONFIG_NON_INLINE_CREDENTIALS_ERROR,
+    }
+    assert kubernetes_provider.connected is False
+
+
@pytest.mark.parametrize(
    "lighthouse_data",
    [
				`@@ -1 +0,0 @@`
				`/api/v1/accounts/saml/{organization_slug}/acs/` rejects non-POST requests before SAML response processing
				`@@ -1 +0,0 @@`
				`Tenant deletion no longer leaves memberships partially removed when exclusive-user cleanup fails`
				`@@ -1 +0,0 @@`
				`GET /api/v1/users/me` membership relationships identify the active tenant with `meta.active` for JWT and API key authentication