mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-10-04 02:04:06 +00:00
Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7f8d95c097 | ||
|
|
ad4b82b63f | ||
|
|
4c748e8e88 | ||
|
|
36f74584a7 | ||
|
|
5a1430a49f |
@@ -82,6 +82,11 @@ provider/vercel:
|
||||
- any-glob-to-any-file: "prowler/providers/vercel/**"
|
||||
- any-glob-to-any-file: "tests/providers/vercel/**"
|
||||
|
||||
provider/supabase:
|
||||
- changed-files:
|
||||
- any-glob-to-any-file: "prowler/providers/supabase/**"
|
||||
- any-glob-to-any-file: "tests/providers/supabase/**"
|
||||
|
||||
provider/okta:
|
||||
- changed-files:
|
||||
- any-glob-to-any-file: "prowler/providers/okta/**"
|
||||
|
||||
@@ -185,6 +185,13 @@ modules:
|
||||
- tests/providers/vercel/**
|
||||
e2e: []
|
||||
|
||||
- name: sdk-supabase
|
||||
match:
|
||||
- prowler/providers/supabase/**
|
||||
tests:
|
||||
- tests/providers/supabase/**
|
||||
e2e: []
|
||||
|
||||
# ============================================
|
||||
# SDK - Lib modules
|
||||
# ============================================
|
||||
|
||||
@@ -465,6 +465,13 @@
|
||||
"user-guide/providers/stackit/authentication"
|
||||
]
|
||||
},
|
||||
{
|
||||
"group": "Supabase",
|
||||
"pages": [
|
||||
"user-guide/providers/supabase/getting-started-supabase",
|
||||
"user-guide/providers/supabase/authentication"
|
||||
]
|
||||
},
|
||||
{
|
||||
"group": "Vercel",
|
||||
"pages": [
|
||||
|
||||
@@ -72,6 +72,7 @@ Prowler supports a wide range of providers organized by category:
|
||||
| [M365](/user-guide/providers/microsoft365/getting-started-m365) | Official | Tenants | UI, API, CLI |
|
||||
| [MongoDB Atlas](/user-guide/providers/mongodbatlas/getting-started-mongodbatlas) | Official | Organizations | UI, API, CLI |
|
||||
| [Okta](/user-guide/providers/okta/getting-started-okta) | Official | Organizations | CLI |
|
||||
| [Supabase](/user-guide/providers/supabase/getting-started-supabase) | [Contact us](https://prowler.com/contact) | Organizations / Members | CLI |
|
||||
| [Vercel](/user-guide/providers/vercel/getting-started-vercel) | Official | Teams / Projects | UI, API, CLI |
|
||||
|
||||
### Kubernetes
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
---
|
||||
title: "Supabase Authentication in Prowler"
|
||||
---
|
||||
|
||||
Prowler for Supabase authenticates to the hosted Supabase Management API using a **Personal Access Token** (PAT). Prowler reads the token exclusively from `SUPABASE_ACCESS_TOKEN`; there is no credential CLI argument, and Prowler does not read Supabase CLI state files.
|
||||
|
||||
## Personal Access Token Security
|
||||
|
||||
Supabase PATs inherit the privileges of the user account that created them. Use a dedicated least-privilege or read-only organization account where available, set a suitable token expiry, store the token in a secret manager, and rotate it regularly.
|
||||
|
||||
<Warning>
|
||||
A PAT is not independently scoped to read-only Management API access. Compromise of the token grants the same Management API privileges as its issuing user account.
|
||||
</Warning>
|
||||
|
||||
## Required Access
|
||||
|
||||
The issuing account must be able to call:
|
||||
|
||||
- [`GET /v1/organizations`](https://supabase.com/docs/reference/api/v1-list-all-organizations)
|
||||
- [`GET /v1/organizations/{slug}/members`](https://supabase.com/docs/reference/api/v1-list-organization-members)
|
||||
|
||||
An invalid token returns `401`, insufficient organization permissions return `403`, and rate limiting returns `429`. Prowler treats all three as scan errors rather than compliant results. Supabase applies a standard limit of approximately 120 Management API requests per minute for each user and scope; Prowler uses the response rate-limit reset headers before retrying.
|
||||
|
||||
## Configure Authentication
|
||||
|
||||
Create a PAT from [Supabase Account Tokens](https://supabase.com/dashboard/account/tokens), then export it:
|
||||
|
||||
```bash
|
||||
export SUPABASE_ACCESS_TOKEN="your-personal-access-token"
|
||||
prowler supabase
|
||||
```
|
||||
|
||||
For CI/CD, configure `SUPABASE_ACCESS_TOKEN` as a masked secret environment variable. Do not place it in command arguments, repository files, or job logs.
|
||||
|
||||
## Out of Scope
|
||||
|
||||
This proof of concept does not support:
|
||||
|
||||
- OAuth 2.0 Authorization Code with Proof Key for Code Exchange (PKCE)
|
||||
- Direct PostgreSQL authentication
|
||||
- Self-hosted Supabase deployments
|
||||
- Supabase CLI credential or state files
|
||||
@@ -0,0 +1,40 @@
|
||||
---
|
||||
title: "Getting Started With Supabase on Prowler"
|
||||
---
|
||||
|
||||
Prowler for Supabase scans hosted Supabase Cloud organizations through the Management API. This proof of concept checks whether each organization member has multi-factor authentication (MFA) enabled.
|
||||
|
||||
<Warning>
|
||||
This proof of concept supports Prowler SDK and Prowler CLI only. It does not support onboarding through Prowler Cloud, Prowler Private Cloud, or Prowler Local Server; self-hosted Supabase; direct PostgreSQL access; or OAuth.
|
||||
</Warning>
|
||||
|
||||
## Prerequisites
|
||||
|
||||
1. A hosted Supabase Cloud account with access to at least one organization.
|
||||
2. A Supabase Personal Access Token (PAT) for an account that can list organizations and organization members.
|
||||
3. A dedicated least-privilege or read-only organization account where your Supabase plan and organization roles allow one.
|
||||
|
||||
See [Supabase Authentication in Prowler](./authentication) for PAT setup and security guidance.
|
||||
|
||||
## Run a Scan
|
||||
|
||||
Set the PAT through the environment and run Prowler:
|
||||
|
||||
```bash
|
||||
export SUPABASE_ACCESS_TOKEN="your-personal-access-token"
|
||||
prowler supabase
|
||||
```
|
||||
|
||||
To run only the organization member MFA check:
|
||||
|
||||
```bash
|
||||
prowler supabase --check organizations_member_mfa_enabled
|
||||
```
|
||||
|
||||
## Checks Shipped
|
||||
|
||||
| Check ID | Severity | Description |
|
||||
|---|---|---|
|
||||
| `organizations_member_mfa_enabled` | High | Creates one finding per organization member and fails members whose Supabase account does not have MFA enabled. |
|
||||
|
||||
The finding resource UID is the Management API `user_id`. Prowler does not include member email addresses in findings or normal logs.
|
||||
@@ -0,0 +1 @@
|
||||
Supabase provider with the `organizations_member_mfa_enabled` check
|
||||
@@ -83,6 +83,7 @@ class Provider(str, Enum):
|
||||
LINODE = "linode"
|
||||
HUAWEICLOUD = "huaweicloud"
|
||||
E2ENETWORKS = "e2enetworks"
|
||||
SUPABASE = "supabase"
|
||||
|
||||
|
||||
# Compliance
|
||||
|
||||
@@ -716,6 +716,11 @@ cloudflare:
|
||||
# Set to 0 to disable retries
|
||||
max_retries: 3
|
||||
|
||||
# Supabase Configuration
|
||||
supabase:
|
||||
# Maximum retries for Management API requests (0 disables retries)
|
||||
max_retries: 3
|
||||
|
||||
# Vercel Configuration
|
||||
vercel:
|
||||
# vercel.deployment_production_uses_stable_target
|
||||
|
||||
@@ -17,6 +17,7 @@ from prowler.config.schema.m365 import M365ProviderConfig
|
||||
from prowler.config.schema.mongodbatlas import MongoDBAtlasProviderConfig
|
||||
from prowler.config.schema.okta import OktaProviderConfig
|
||||
from prowler.config.schema.openstack import OpenStackProviderConfig
|
||||
from prowler.config.schema.supabase import SupabaseProviderConfig
|
||||
from prowler.config.schema.vercel import VercelProviderConfig
|
||||
|
||||
SCHEMAS: dict[str, type[ProviderConfigBase]] = {
|
||||
@@ -33,4 +34,5 @@ SCHEMAS: dict[str, type[ProviderConfigBase]] = {
|
||||
"okta": OktaProviderConfig,
|
||||
"alibabacloud": AlibabaCloudProviderConfig,
|
||||
"openstack": OpenStackProviderConfig,
|
||||
"supabase": SupabaseProviderConfig,
|
||||
}
|
||||
|
||||
@@ -0,0 +1,16 @@
|
||||
from typing import Optional
|
||||
|
||||
from pydantic import Field
|
||||
|
||||
from prowler.config.schema.base import ProviderConfigBase
|
||||
|
||||
|
||||
class SupabaseProviderConfig(ProviderConfigBase):
|
||||
"""Supabase provider configuration schema."""
|
||||
|
||||
max_retries: Optional[int] = Field(
|
||||
default=None,
|
||||
ge=0,
|
||||
le=10,
|
||||
description="Maximum retries for Supabase API requests. Range: 0..10.",
|
||||
)
|
||||
@@ -1447,6 +1447,28 @@ class CheckReportVercel(Check_Report):
|
||||
return "global"
|
||||
|
||||
|
||||
@dataclass
|
||||
class CheckReportSupabase(Check_Report):
|
||||
"""Contains a Supabase organization member finding."""
|
||||
|
||||
resource_name: str
|
||||
resource_id: str
|
||||
organization_slug: str
|
||||
organization_name: str
|
||||
|
||||
def __init__(self, metadata: Dict, resource: Any) -> None:
|
||||
super().__init__(metadata, resource)
|
||||
self.resource_name = getattr(resource, "name", "")
|
||||
self.resource_id = getattr(resource, "id", "")
|
||||
self.organization_slug = getattr(resource, "organization_slug", "")
|
||||
self.organization_name = getattr(resource, "organization_name", "")
|
||||
|
||||
@property
|
||||
def region(self) -> str:
|
||||
"""Supabase organization membership is global."""
|
||||
return "global"
|
||||
|
||||
|
||||
@dataclass
|
||||
class CheckReportScaleway(Check_Report):
|
||||
"""Contains the Scaleway Check's finding information.
|
||||
|
||||
@@ -0,0 +1,104 @@
|
||||
from prowler.exceptions.exceptions import ProwlerException
|
||||
|
||||
|
||||
# Exceptions codes from 21000 to 21999 are reserved for Supabase exceptions
|
||||
class SupabaseBaseException(ProwlerException):
|
||||
"""Base exception for Supabase provider errors."""
|
||||
|
||||
SUPABASE_ERROR_CODES = {
|
||||
(21000, "SupabaseCredentialsError"): {
|
||||
"message": "Supabase credentials were not found.",
|
||||
"remediation": "Set SUPABASE_ACCESS_TOKEN to a valid Supabase Personal Access Token.",
|
||||
},
|
||||
(21001, "SupabaseAuthenticationError"): {
|
||||
"message": "Supabase Management API authentication failed.",
|
||||
"remediation": "Verify that SUPABASE_ACCESS_TOKEN is valid and has not expired or been revoked.",
|
||||
},
|
||||
(21002, "SupabaseInsufficientPermissionsError"): {
|
||||
"message": "Supabase Management API permissions are insufficient.",
|
||||
"remediation": "Use a Personal Access Token from an account that can read the target organizations and their members.",
|
||||
},
|
||||
(21003, "SupabaseRateLimitError"): {
|
||||
"message": "The Supabase Management API rate limit was exceeded.",
|
||||
"remediation": "Wait for the rate-limit window to reset before retrying the scan.",
|
||||
},
|
||||
(21004, "SupabaseSessionError"): {
|
||||
"message": "Failed to create a Supabase Management API session.",
|
||||
"remediation": "Check the local HTTP client configuration and retry.",
|
||||
},
|
||||
(21005, "SupabaseIdentityError"): {
|
||||
"message": "Failed to retrieve Supabase organization identity information.",
|
||||
"remediation": "Ensure the token can call GET /v1/organizations.",
|
||||
},
|
||||
(21006, "SupabaseAPIError"): {
|
||||
"message": "A Supabase Management API request failed.",
|
||||
"remediation": "Check Supabase service status, network connectivity, and the API response before retrying.",
|
||||
},
|
||||
}
|
||||
|
||||
def __init__(self, code, file=None, original_exception=None, message=None):
|
||||
error_info = self.SUPABASE_ERROR_CODES.get((code, self.__class__.__name__))
|
||||
if error_info is None:
|
||||
error_info = {
|
||||
"message": message or "Unknown Supabase error.",
|
||||
"remediation": "Review the Supabase Management API documentation.",
|
||||
}
|
||||
elif message:
|
||||
error_info = error_info.copy()
|
||||
error_info["message"] = message
|
||||
super().__init__(
|
||||
code=code,
|
||||
source="Supabase",
|
||||
file=file,
|
||||
original_exception=original_exception,
|
||||
error_info=error_info,
|
||||
)
|
||||
|
||||
|
||||
class SupabaseCredentialsError(SupabaseBaseException):
|
||||
def __init__(self, file=None, original_exception=None, message=None):
|
||||
super().__init__(
|
||||
21000, file=file, original_exception=original_exception, message=message
|
||||
)
|
||||
|
||||
|
||||
class SupabaseAuthenticationError(SupabaseBaseException):
|
||||
def __init__(self, file=None, original_exception=None, message=None):
|
||||
super().__init__(
|
||||
21001, file=file, original_exception=original_exception, message=message
|
||||
)
|
||||
|
||||
|
||||
class SupabaseInsufficientPermissionsError(SupabaseBaseException):
|
||||
def __init__(self, file=None, original_exception=None, message=None):
|
||||
super().__init__(
|
||||
21002, file=file, original_exception=original_exception, message=message
|
||||
)
|
||||
|
||||
|
||||
class SupabaseRateLimitError(SupabaseBaseException):
|
||||
def __init__(self, file=None, original_exception=None, message=None):
|
||||
super().__init__(
|
||||
21003, file=file, original_exception=original_exception, message=message
|
||||
)
|
||||
|
||||
|
||||
class SupabaseSessionError(SupabaseBaseException):
|
||||
def __init__(self, file=None, original_exception=None, message=None):
|
||||
super().__init__(
|
||||
21004, file=file, original_exception=original_exception, message=message
|
||||
)
|
||||
|
||||
|
||||
class SupabaseIdentityError(SupabaseBaseException):
|
||||
def __init__(self, file=None, original_exception=None, message=None):
|
||||
super().__init__(
|
||||
21005, file=file, original_exception=original_exception, message=message
|
||||
)
|
||||
|
||||
|
||||
class SupabaseAPIError(SupabaseBaseException):
|
||||
def __init__(self, file=None, original_exception=None, message=None):
|
||||
super().__init__(
|
||||
21006, file=file, original_exception=original_exception, message=message
|
||||
)
|
||||
@@ -0,0 +1,7 @@
|
||||
def init_parser(self):
|
||||
"""Initialize the Supabase provider CLI parser."""
|
||||
self.subparsers.add_parser(
|
||||
"supabase",
|
||||
parents=[self.common_providers_parser],
|
||||
help="Supabase Provider (PoC)",
|
||||
)
|
||||
@@ -0,0 +1,16 @@
|
||||
from prowler.lib.check.models import CheckReportSupabase
|
||||
from prowler.lib.mutelist.mutelist import Mutelist
|
||||
from prowler.lib.outputs.utils import unroll_dict, unroll_tags
|
||||
|
||||
|
||||
class SupabaseMutelist(Mutelist):
|
||||
"""Supabase-specific mutelist helper."""
|
||||
|
||||
def is_finding_muted(self, finding: CheckReportSupabase) -> bool:
|
||||
return self.is_muted(
|
||||
finding.organization_slug,
|
||||
finding.check_metadata.CheckID,
|
||||
"global",
|
||||
finding.resource_id or finding.resource_name,
|
||||
unroll_dict(unroll_tags(finding.resource_tags)),
|
||||
)
|
||||
@@ -0,0 +1,132 @@
|
||||
import time
|
||||
from email.utils import parsedate_to_datetime
|
||||
from math import isfinite
|
||||
|
||||
import requests
|
||||
|
||||
from prowler.lib.logger import logger
|
||||
from prowler.providers.supabase.exceptions.exceptions import (
|
||||
SupabaseAPIError,
|
||||
SupabaseAuthenticationError,
|
||||
SupabaseInsufficientPermissionsError,
|
||||
SupabaseRateLimitError,
|
||||
)
|
||||
|
||||
MAX_RATE_LIMIT_DELAY = 3600
|
||||
MIN_UNIX_TIMESTAMP = 1_000_000_000
|
||||
RATE_LIMIT_RETRIES_EXHAUSTED_MESSAGE = (
|
||||
"Supabase API rate limit remained active after retries."
|
||||
)
|
||||
API_REQUEST_RETRIES_EXHAUSTED_MESSAGE = (
|
||||
"Supabase Management API request failed after retries."
|
||||
)
|
||||
SUPABASE_API_REQUEST_RETRY_WARNING = (
|
||||
"Supabase API request failed; retrying in {delay} seconds."
|
||||
)
|
||||
|
||||
|
||||
def _bounded_delay(seconds: float) -> int:
|
||||
"""Return a non-negative, bounded delay in whole seconds."""
|
||||
if not isfinite(seconds):
|
||||
raise ValueError("Rate-limit delay must be finite.")
|
||||
return int(max(0, min(seconds, MAX_RATE_LIMIT_DELAY)))
|
||||
|
||||
|
||||
def _rate_limit_delay(headers: dict) -> int:
|
||||
"""Return the server-requested rate-limit delay in seconds."""
|
||||
reset = headers.get("X-RateLimit-Reset")
|
||||
if reset is not None:
|
||||
try:
|
||||
reset_seconds = float(reset)
|
||||
if reset_seconds >= MIN_UNIX_TIMESTAMP:
|
||||
reset_seconds -= time.time()
|
||||
return _bounded_delay(reset_seconds)
|
||||
except (TypeError, ValueError):
|
||||
pass
|
||||
|
||||
retry_after = headers.get("Retry-After")
|
||||
if retry_after is not None:
|
||||
try:
|
||||
return _bounded_delay(float(retry_after))
|
||||
except (TypeError, ValueError):
|
||||
try:
|
||||
retry_at = parsedate_to_datetime(retry_after)
|
||||
return _bounded_delay(retry_at.timestamp() - time.time())
|
||||
except (TypeError, ValueError, OverflowError, OSError):
|
||||
pass
|
||||
|
||||
return 1
|
||||
|
||||
|
||||
def request_json(session, path: str, max_retries: int = 3):
|
||||
"""Make a Management API GET request with explicit authorization errors."""
|
||||
url = f"{session.base_url}{path}"
|
||||
for attempt in range(max_retries + 1):
|
||||
try:
|
||||
response = session.http_session.get(url, timeout=30)
|
||||
if response.status_code == 401:
|
||||
raise SupabaseAuthenticationError(
|
||||
file=__file__,
|
||||
message="Invalid or expired Supabase access token.",
|
||||
)
|
||||
if response.status_code == 403:
|
||||
raise SupabaseInsufficientPermissionsError(
|
||||
file=__file__,
|
||||
message=(
|
||||
"The Supabase access token cannot read the requested "
|
||||
"organization data."
|
||||
),
|
||||
)
|
||||
if response.status_code == 429:
|
||||
delay = _rate_limit_delay(response.headers)
|
||||
if attempt < max_retries:
|
||||
logger.warning(
|
||||
"Supabase API rate limit reached; "
|
||||
f"retrying in {delay} seconds."
|
||||
)
|
||||
time.sleep(delay)
|
||||
continue
|
||||
raise SupabaseRateLimitError(
|
||||
file=__file__,
|
||||
message=RATE_LIMIT_RETRIES_EXHAUSTED_MESSAGE,
|
||||
)
|
||||
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
except (
|
||||
SupabaseAuthenticationError,
|
||||
SupabaseInsufficientPermissionsError,
|
||||
SupabaseRateLimitError,
|
||||
):
|
||||
raise
|
||||
except (requests.exceptions.RequestException, ValueError) as error:
|
||||
if attempt < max_retries:
|
||||
delay = 2**attempt
|
||||
msg = SUPABASE_API_REQUEST_RETRY_WARNING.format(delay=delay)
|
||||
logger.warning(msg)
|
||||
time.sleep(delay)
|
||||
continue
|
||||
raise SupabaseAPIError(
|
||||
file=__file__,
|
||||
original_exception=error,
|
||||
message=API_REQUEST_RETRIES_EXHAUSTED_MESSAGE,
|
||||
)
|
||||
|
||||
|
||||
class SupabaseService:
|
||||
"""Base class for Supabase services."""
|
||||
|
||||
def __init__(self, service: str, provider):
|
||||
self.provider = provider
|
||||
self.session = provider.session
|
||||
self.audit_config = provider.audit_config
|
||||
self.fixer_config = provider.fixer_config
|
||||
self.service = service.lower()
|
||||
|
||||
def _get(self, path: str):
|
||||
"""Return decoded JSON from a Management API endpoint."""
|
||||
return request_json(
|
||||
self.session,
|
||||
path,
|
||||
max_retries=self.audit_config.get("max_retries", 3),
|
||||
)
|
||||
@@ -0,0 +1,42 @@
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from prowler.config.config import output_file_timestamp
|
||||
from prowler.providers.common.models import ProviderOutputOptions
|
||||
|
||||
|
||||
class SupabaseSession(BaseModel):
|
||||
"""Supabase Management API session."""
|
||||
|
||||
access_token: str = Field(exclude=True, repr=False)
|
||||
base_url: str = "https://api.supabase.com"
|
||||
http_session: Any = Field(default=None, exclude=True, repr=False)
|
||||
|
||||
|
||||
class SupabaseOrganization(BaseModel):
|
||||
"""Supabase organization visible to the authenticated account."""
|
||||
|
||||
id: str
|
||||
slug: str
|
||||
name: str
|
||||
|
||||
|
||||
class SupabaseIdentityInfo(BaseModel):
|
||||
"""Supabase identity and organization scope."""
|
||||
|
||||
organizations: list[SupabaseOrganization] = Field(default_factory=list)
|
||||
|
||||
|
||||
class SupabaseOutputOptions(ProviderOutputOptions):
|
||||
"""Customize output filenames for Supabase scans."""
|
||||
|
||||
def __init__(self, arguments, bulk_checks_metadata, identity: SupabaseIdentityInfo):
|
||||
super().__init__(arguments, bulk_checks_metadata)
|
||||
if getattr(arguments, "output_filename", None) is None:
|
||||
fragment = (
|
||||
identity.organizations[0].slug if identity.organizations else "supabase"
|
||||
)
|
||||
self.output_filename = f"prowler-output-{fragment}-{output_file_timestamp}"
|
||||
else:
|
||||
self.output_filename = arguments.output_filename
|
||||
@@ -0,0 +1,6 @@
|
||||
from prowler.providers.common.provider import Provider
|
||||
from prowler.providers.supabase.services.organizations.organizations_service import (
|
||||
Organizations,
|
||||
)
|
||||
|
||||
organizations_client = Organizations(Provider.get_global_provider())
|
||||
+37
@@ -0,0 +1,37 @@
|
||||
{
|
||||
"Provider": "supabase",
|
||||
"CheckID": "organizations_member_mfa_enabled",
|
||||
"CheckTitle": "Supabase organization members have multi-factor authentication enabled",
|
||||
"CheckType": [],
|
||||
"ServiceName": "organizations",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "high",
|
||||
"ResourceType": "NotDefined",
|
||||
"ResourceGroup": "IAM",
|
||||
"Description": "This check verifies whether each Supabase organization member has multi-factor authentication enabled for their Supabase account.",
|
||||
"Risk": "Without multi-factor authentication, compromised credentials can allow unauthorized access to Supabase organizations, projects, and management operations available to the member.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://supabase.com/docs/reference/api/v1-list-organization-members",
|
||||
"https://supabase.com/docs/guides/platform/multi-factor-authentication"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "Ask the organization member to open Supabase Dashboard > Account Settings > Security and enroll a TOTP factor. Organization owners on eligible plans can also enforce MFA for all members.",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Require every organization member to enroll multi-factor authentication and periodically review membership for accounts that do not meet this requirement.",
|
||||
"Url": "https://hub.prowler.com/check/organizations_member_mfa_enabled"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"identity-access"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": ""
|
||||
}
|
||||
+28
@@ -0,0 +1,28 @@
|
||||
from prowler.lib.check.models import Check, CheckReportSupabase
|
||||
from prowler.providers.supabase.services.organizations.organizations_client import (
|
||||
organizations_client,
|
||||
)
|
||||
|
||||
|
||||
class organizations_member_mfa_enabled(Check):
|
||||
"""Check whether each Supabase organization member has MFA enabled."""
|
||||
|
||||
def execute(self) -> list[CheckReportSupabase]:
|
||||
"""Return one MFA finding per organization member."""
|
||||
findings = []
|
||||
for member in organizations_client.members.values():
|
||||
report = CheckReportSupabase(metadata=self.metadata(), resource=member)
|
||||
if member.mfa_enabled:
|
||||
report.status = "PASS"
|
||||
report.status_extended = (
|
||||
f"Supabase organization {member.organization_slug} member "
|
||||
f"{member.id} has MFA enabled."
|
||||
)
|
||||
else:
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
f"Supabase organization {member.organization_slug} member "
|
||||
f"{member.id} does not have MFA enabled."
|
||||
)
|
||||
findings.append(report)
|
||||
return findings
|
||||
@@ -0,0 +1,37 @@
|
||||
from pydantic import BaseModel
|
||||
|
||||
from prowler.providers.supabase.lib.service.service import SupabaseService
|
||||
|
||||
|
||||
class Organizations(SupabaseService):
|
||||
"""Retrieve Supabase organization members."""
|
||||
|
||||
def __init__(self, provider):
|
||||
super().__init__("Organizations", provider)
|
||||
self.members: dict[str, SupabaseOrganizationMember] = {}
|
||||
self._fetch_members()
|
||||
|
||||
def _fetch_members(self) -> None:
|
||||
"""Collect members from every organization in the authenticated scope."""
|
||||
for organization in self.provider.identity.organizations:
|
||||
members = self._get(f"/v1/organizations/{organization.slug}/members")
|
||||
for member in members:
|
||||
user_id = member["user_id"]
|
||||
resource = SupabaseOrganizationMember(
|
||||
id=user_id,
|
||||
name=f"member {user_id}",
|
||||
organization_slug=organization.slug,
|
||||
organization_name=organization.name,
|
||||
mfa_enabled=member["mfa_enabled"],
|
||||
)
|
||||
self.members[f"{organization.slug}:{user_id}"] = resource
|
||||
|
||||
|
||||
class SupabaseOrganizationMember(BaseModel):
|
||||
"""Security-relevant Supabase organization member attributes."""
|
||||
|
||||
id: str
|
||||
name: str
|
||||
organization_slug: str
|
||||
organization_name: str
|
||||
mfa_enabled: bool
|
||||
@@ -0,0 +1,236 @@
|
||||
import os
|
||||
from argparse import Namespace
|
||||
|
||||
import requests
|
||||
from colorama import Fore, Style
|
||||
|
||||
from prowler.config.config import (
|
||||
default_config_file_path,
|
||||
get_default_mute_file_path,
|
||||
load_and_validate_config_file,
|
||||
)
|
||||
from prowler.lib.utils.utils import print_boxes
|
||||
from prowler.providers.common.models import Audit_Metadata, Connection
|
||||
from prowler.providers.common.provider import Provider
|
||||
from prowler.providers.supabase.exceptions.exceptions import (
|
||||
SupabaseAPIError,
|
||||
SupabaseAuthenticationError,
|
||||
SupabaseCredentialsError,
|
||||
SupabaseIdentityError,
|
||||
SupabaseInsufficientPermissionsError,
|
||||
SupabaseRateLimitError,
|
||||
SupabaseSessionError,
|
||||
)
|
||||
from prowler.providers.supabase.lib.mutelist.mutelist import SupabaseMutelist
|
||||
from prowler.providers.supabase.lib.service.service import request_json
|
||||
from prowler.providers.supabase.models import (
|
||||
SupabaseIdentityInfo,
|
||||
SupabaseOrganization,
|
||||
SupabaseOutputOptions,
|
||||
SupabaseSession,
|
||||
)
|
||||
|
||||
|
||||
class SupabaseProvider(Provider):
|
||||
"""Hosted Supabase Cloud Management API provider."""
|
||||
|
||||
_type = "supabase"
|
||||
_cli_help_text = "Supabase Provider (PoC)"
|
||||
sdk_only = True
|
||||
audit_metadata: Audit_Metadata
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
access_token: str = None,
|
||||
config_path: str = None,
|
||||
config_content: dict | None = None,
|
||||
fixer_config: dict = {},
|
||||
mutelist_path: str = None,
|
||||
mutelist_content: dict = None,
|
||||
):
|
||||
self._audit_config = config_content or load_and_validate_config_file(
|
||||
self._type, config_path or default_config_file_path
|
||||
)
|
||||
self._session = self.setup_session(access_token)
|
||||
self._identity = self.setup_identity(
|
||||
self._session, self._audit_config.get("max_retries", 3)
|
||||
)
|
||||
self._fixer_config = fixer_config
|
||||
self._mutelist = (
|
||||
SupabaseMutelist(mutelist_content=mutelist_content)
|
||||
if mutelist_content
|
||||
else SupabaseMutelist(
|
||||
mutelist_path=mutelist_path or get_default_mute_file_path(self.type)
|
||||
)
|
||||
)
|
||||
Provider.set_global_provider(self)
|
||||
|
||||
@property
|
||||
def type(self):
|
||||
return self._type
|
||||
|
||||
@property
|
||||
def session(self):
|
||||
return self._session
|
||||
|
||||
@property
|
||||
def identity(self):
|
||||
return self._identity
|
||||
|
||||
@property
|
||||
def audit_config(self):
|
||||
return self._audit_config
|
||||
|
||||
@property
|
||||
def fixer_config(self):
|
||||
return self._fixer_config
|
||||
|
||||
@property
|
||||
def mutelist(self):
|
||||
return self._mutelist
|
||||
|
||||
@staticmethod
|
||||
def setup_session(access_token: str = None) -> SupabaseSession:
|
||||
"""Create a Bearer-token Management API session."""
|
||||
token = access_token or os.environ.get("SUPABASE_ACCESS_TOKEN", "")
|
||||
if not token:
|
||||
raise SupabaseCredentialsError(file=os.path.basename(__file__))
|
||||
try:
|
||||
http_session = requests.Session()
|
||||
http_session.headers.update(
|
||||
{
|
||||
"Authorization": f"Bearer {token}",
|
||||
"Accept": "application/json",
|
||||
}
|
||||
)
|
||||
return SupabaseSession(
|
||||
access_token=token,
|
||||
http_session=http_session,
|
||||
)
|
||||
except Exception as error:
|
||||
raise SupabaseSessionError(
|
||||
file=os.path.basename(__file__), original_exception=error
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def setup_identity(
|
||||
session: SupabaseSession, max_retries: int = 3
|
||||
) -> SupabaseIdentityInfo:
|
||||
"""List organizations to validate credentials and establish scan scope."""
|
||||
try:
|
||||
organizations = request_json(
|
||||
session, "/v1/organizations", max_retries=max_retries
|
||||
)
|
||||
return SupabaseIdentityInfo(
|
||||
organizations=[
|
||||
SupabaseOrganization(
|
||||
id=organization["id"],
|
||||
slug=organization["slug"],
|
||||
name=organization["name"],
|
||||
)
|
||||
for organization in organizations
|
||||
]
|
||||
)
|
||||
except (
|
||||
SupabaseAuthenticationError,
|
||||
SupabaseInsufficientPermissionsError,
|
||||
SupabaseRateLimitError,
|
||||
SupabaseAPIError,
|
||||
):
|
||||
raise
|
||||
except Exception as error:
|
||||
raise SupabaseIdentityError(
|
||||
file=os.path.basename(__file__), original_exception=error
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def from_cli_args(cls, arguments: Namespace, fixer_config: dict):
|
||||
"""Create the provider from non-secret CLI configuration."""
|
||||
return cls(
|
||||
config_path=arguments.config_file,
|
||||
mutelist_path=arguments.mutelist_file,
|
||||
fixer_config=fixer_config,
|
||||
)
|
||||
|
||||
def print_credentials(self) -> None:
|
||||
organizations = (
|
||||
", ".join(organization.slug for organization in self.identity.organizations)
|
||||
or "none"
|
||||
)
|
||||
print_boxes(
|
||||
[
|
||||
f"Authentication: {Fore.YELLOW}Personal Access Token{Style.RESET_ALL}",
|
||||
f"Organizations: {Fore.YELLOW}{organizations}{Style.RESET_ALL}",
|
||||
],
|
||||
f"{Style.BRIGHT}Using the Supabase credentials below:{Style.RESET_ALL}",
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def test_connection(
|
||||
access_token: str = None,
|
||||
raise_on_exception: bool = True,
|
||||
provider_id: str = None,
|
||||
) -> Connection:
|
||||
"""Test access to the Supabase organizations endpoint."""
|
||||
try:
|
||||
session = SupabaseProvider.setup_session(access_token)
|
||||
identity = SupabaseProvider.setup_identity(session, max_retries=0)
|
||||
if provider_id and provider_id not in {
|
||||
identifier
|
||||
for organization in identity.organizations
|
||||
for identifier in (organization.id, organization.slug)
|
||||
}:
|
||||
raise SupabaseIdentityError(
|
||||
file=os.path.basename(__file__),
|
||||
message=(
|
||||
f"Supabase organization '{provider_id}' is not accessible "
|
||||
"with the supplied token."
|
||||
),
|
||||
)
|
||||
return Connection(is_connected=True)
|
||||
except Exception as error:
|
||||
if raise_on_exception:
|
||||
raise
|
||||
return Connection(is_connected=False, error=error)
|
||||
|
||||
def validate_arguments(self) -> None:
|
||||
return None
|
||||
|
||||
def get_output_options(self, arguments, bulk_checks_metadata):
|
||||
return SupabaseOutputOptions(arguments, bulk_checks_metadata, self.identity)
|
||||
|
||||
def get_stdout_detail(self, _finding) -> str:
|
||||
return "global"
|
||||
|
||||
def get_summary_entity(self) -> tuple[str, str]:
|
||||
organizations = ", ".join(
|
||||
f"{organization.name} ({organization.slug})"
|
||||
for organization in self.identity.organizations
|
||||
)
|
||||
return "Organization", organizations or "No organizations"
|
||||
|
||||
def get_finding_output_data(self, check_output) -> dict:
|
||||
return {
|
||||
"auth_method": "personal_access_token",
|
||||
"account_uid": check_output.organization_slug,
|
||||
"account_name": check_output.organization_name,
|
||||
"resource_name": check_output.resource_name,
|
||||
"resource_uid": check_output.resource_id,
|
||||
"region": "global",
|
||||
}
|
||||
|
||||
def get_html_assessment_summary(self) -> str:
|
||||
organizations = (
|
||||
", ".join(organization.slug for organization in self.identity.organizations)
|
||||
or "none"
|
||||
)
|
||||
return f"""
|
||||
<div class="col-md-4">
|
||||
<div class="card">
|
||||
<div class="card-header">Supabase Assessment Summary</div>
|
||||
<ul class="list-group list-group-flush">
|
||||
<li class="list-group-item"><b>Organizations:</b> {organizations}</li>
|
||||
<li class="list-group-item"><b>Authentication:</b> Personal Access Token</li>
|
||||
</ul>
|
||||
</div>
|
||||
</div>"""
|
||||
+101
@@ -0,0 +1,101 @@
|
||||
from unittest import mock
|
||||
|
||||
from prowler.providers.supabase.services.organizations.organizations_service import (
|
||||
SupabaseOrganizationMember,
|
||||
)
|
||||
from tests.providers.supabase.supabase_fixtures import (
|
||||
ORGANIZATION_NAME,
|
||||
ORGANIZATION_SLUG,
|
||||
USER_ID,
|
||||
set_mocked_supabase_provider,
|
||||
)
|
||||
|
||||
|
||||
class Test_organizations_member_mfa_enabled:
|
||||
def _execute(self, members):
|
||||
organizations_client = mock.MagicMock()
|
||||
organizations_client.members = members
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_supabase_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.supabase.services.organizations.organizations_member_mfa_enabled.organizations_member_mfa_enabled.organizations_client",
|
||||
new=organizations_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.supabase.services.organizations.organizations_member_mfa_enabled.organizations_member_mfa_enabled import (
|
||||
organizations_member_mfa_enabled,
|
||||
)
|
||||
|
||||
return organizations_member_mfa_enabled().execute()
|
||||
|
||||
def test_no_members(self):
|
||||
assert self._execute({}) == []
|
||||
|
||||
def test_member_with_mfa_passes_without_email(self):
|
||||
member = SupabaseOrganizationMember(
|
||||
id=USER_ID,
|
||||
name=f"member {USER_ID}",
|
||||
organization_slug=ORGANIZATION_SLUG,
|
||||
organization_name=ORGANIZATION_NAME,
|
||||
mfa_enabled=True,
|
||||
)
|
||||
|
||||
result = self._execute({f"{ORGANIZATION_SLUG}:{USER_ID}": member})
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert result[0].resource_id == USER_ID
|
||||
assert result[0].organization_slug == ORGANIZATION_SLUG
|
||||
assert result[0].status_extended == (
|
||||
f"Supabase organization {ORGANIZATION_SLUG} member {USER_ID} has MFA enabled."
|
||||
)
|
||||
|
||||
def test_member_without_mfa_fails(self):
|
||||
member = SupabaseOrganizationMember(
|
||||
id=USER_ID,
|
||||
name=f"member {USER_ID}",
|
||||
organization_slug=ORGANIZATION_SLUG,
|
||||
organization_name=ORGANIZATION_NAME,
|
||||
mfa_enabled=False,
|
||||
)
|
||||
|
||||
result = self._execute({f"{ORGANIZATION_SLUG}:{USER_ID}": member})
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert result[0].resource_id == USER_ID
|
||||
assert result[0].status_extended == (
|
||||
f"Supabase organization {ORGANIZATION_SLUG} member {USER_ID} does not have MFA enabled."
|
||||
)
|
||||
|
||||
def test_multiple_members_return_independent_pass_and_fail_findings(self):
|
||||
enabled_member = SupabaseOrganizationMember(
|
||||
id="enabled-user",
|
||||
name="member enabled-user",
|
||||
organization_slug=ORGANIZATION_SLUG,
|
||||
organization_name=ORGANIZATION_NAME,
|
||||
mfa_enabled=True,
|
||||
)
|
||||
disabled_member = SupabaseOrganizationMember(
|
||||
id="disabled-user",
|
||||
name="member disabled-user",
|
||||
organization_slug=ORGANIZATION_SLUG,
|
||||
organization_name=ORGANIZATION_NAME,
|
||||
mfa_enabled=False,
|
||||
)
|
||||
|
||||
result = self._execute(
|
||||
{
|
||||
f"{ORGANIZATION_SLUG}:enabled-user": enabled_member,
|
||||
f"{ORGANIZATION_SLUG}:disabled-user": disabled_member,
|
||||
}
|
||||
)
|
||||
|
||||
assert [(finding.resource_id, finding.status) for finding in result] == [
|
||||
("enabled-user", "PASS"),
|
||||
("disabled-user", "FAIL"),
|
||||
]
|
||||
+147
@@ -0,0 +1,147 @@
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
from prowler.providers.supabase.exceptions.exceptions import (
|
||||
SupabaseAuthenticationError,
|
||||
SupabaseInsufficientPermissionsError,
|
||||
SupabaseRateLimitError,
|
||||
)
|
||||
from prowler.providers.supabase.lib.service.service import _rate_limit_delay
|
||||
from prowler.providers.supabase.services.organizations import (
|
||||
organizations_service,
|
||||
)
|
||||
from tests.providers.supabase.supabase_fixtures import (
|
||||
ORGANIZATION_SLUG,
|
||||
USER_ID,
|
||||
set_mocked_supabase_provider,
|
||||
)
|
||||
|
||||
|
||||
class TestOrganizationsService:
|
||||
def test_fetches_members_without_storing_email(self):
|
||||
provider = set_mocked_supabase_provider()
|
||||
response = mock.MagicMock(status_code=200)
|
||||
response.json.return_value = [
|
||||
{
|
||||
"user_id": USER_ID,
|
||||
"user_name": "Test User",
|
||||
"email": "private@example.com",
|
||||
"role_name": "Owner",
|
||||
"mfa_enabled": False,
|
||||
}
|
||||
]
|
||||
provider.session.http_session.get.return_value = response
|
||||
|
||||
service = organizations_service.Organizations(provider)
|
||||
|
||||
member = service.members[f"{ORGANIZATION_SLUG}:{USER_ID}"]
|
||||
assert member.id == USER_ID
|
||||
assert member.name == f"member {USER_ID}"
|
||||
assert member.organization_slug == ORGANIZATION_SLUG
|
||||
assert member.mfa_enabled is False
|
||||
assert "email" not in member.model_dump()
|
||||
assert "private@example.com" not in repr(member)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("status_code", "exception"),
|
||||
[
|
||||
(401, SupabaseAuthenticationError),
|
||||
(403, SupabaseInsufficientPermissionsError),
|
||||
(429, SupabaseRateLimitError),
|
||||
],
|
||||
)
|
||||
def test_api_errors_abort_collection_instead_of_producing_empty_results(
|
||||
self, status_code, exception
|
||||
):
|
||||
provider = set_mocked_supabase_provider()
|
||||
provider.session.http_session.get.return_value = mock.MagicMock(
|
||||
status_code=status_code,
|
||||
headers={"X-RateLimit-Reset": "0"},
|
||||
)
|
||||
|
||||
with pytest.raises(exception):
|
||||
organizations_service.Organizations(provider)
|
||||
|
||||
def test_rate_limit_retries_using_reset_header(self):
|
||||
provider = set_mocked_supabase_provider()
|
||||
provider.audit_config = {"max_retries": 1}
|
||||
rate_limited = mock.MagicMock(
|
||||
status_code=429,
|
||||
headers={"X-RateLimit-Reset": "2"},
|
||||
)
|
||||
success = mock.MagicMock(status_code=200)
|
||||
success.json.return_value = []
|
||||
provider.session.http_session.get.side_effect = [rate_limited, success]
|
||||
|
||||
with mock.patch(
|
||||
"prowler.providers.supabase.lib.service.service.time.sleep"
|
||||
) as sleep:
|
||||
service = organizations_service.Organizations(provider)
|
||||
|
||||
assert service.members == {}
|
||||
sleep.assert_called_once_with(2)
|
||||
|
||||
|
||||
class TestRateLimitDelay:
|
||||
def test_relative_reset_seconds(self):
|
||||
assert _rate_limit_delay({"X-RateLimit-Reset": "2"}) == 2
|
||||
|
||||
def test_retry_after_delta_seconds(self):
|
||||
assert _rate_limit_delay({"Retry-After": "3"}) == 3
|
||||
|
||||
@mock.patch(
|
||||
"prowler.providers.supabase.lib.service.service.time.time",
|
||||
return_value=1_700_000_000,
|
||||
)
|
||||
def test_absolute_reset_timestamp(self, _):
|
||||
assert _rate_limit_delay({"X-RateLimit-Reset": "1700000005"}) == 5
|
||||
|
||||
@mock.patch(
|
||||
"prowler.providers.supabase.lib.service.service.time.time",
|
||||
return_value=1_445_412_475,
|
||||
)
|
||||
def test_retry_after_http_date(self, _):
|
||||
headers = {"Retry-After": "Wed, 21 Oct 2015 07:28:00 GMT"}
|
||||
|
||||
assert _rate_limit_delay(headers) == 5
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"malformed_value",
|
||||
["not-a-timestamp", "inf", "nan"],
|
||||
)
|
||||
def test_malformed_headers_use_fallback(self, malformed_value):
|
||||
headers = {
|
||||
"X-RateLimit-Reset": malformed_value,
|
||||
"Retry-After": malformed_value,
|
||||
}
|
||||
|
||||
assert _rate_limit_delay(headers) == 1
|
||||
|
||||
def test_malformed_reset_uses_retry_after(self):
|
||||
headers = {
|
||||
"X-RateLimit-Reset": "not-a-timestamp",
|
||||
"Retry-After": "4",
|
||||
}
|
||||
|
||||
assert _rate_limit_delay(headers) == 4
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"headers",
|
||||
[
|
||||
{"X-RateLimit-Reset": "-5"},
|
||||
{"Retry-After": "-5"},
|
||||
],
|
||||
)
|
||||
def test_negative_delay_is_clamped_to_zero(self, headers):
|
||||
assert _rate_limit_delay(headers) == 0
|
||||
|
||||
@mock.patch(
|
||||
"prowler.providers.supabase.lib.service.service.time.time",
|
||||
return_value=1_700_000_000,
|
||||
)
|
||||
def test_past_absolute_reset_is_clamped_to_zero(self, _):
|
||||
assert _rate_limit_delay({"X-RateLimit-Reset": "1699999995"}) == 0
|
||||
|
||||
def test_relative_delay_is_bounded(self):
|
||||
assert _rate_limit_delay({"X-RateLimit-Reset": "7200"}) == 3600
|
||||
@@ -0,0 +1,34 @@
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from prowler.providers.supabase.models import (
|
||||
SupabaseIdentityInfo,
|
||||
SupabaseOrganization,
|
||||
SupabaseSession,
|
||||
)
|
||||
|
||||
ACCESS_TOKEN = "sbp_test_token"
|
||||
ORGANIZATION_ID = "org-id"
|
||||
ORGANIZATION_NAME = "Test Organization"
|
||||
ORGANIZATION_SLUG = "test-organization"
|
||||
USER_ID = "user-id"
|
||||
|
||||
|
||||
def set_mocked_supabase_provider():
|
||||
provider = MagicMock()
|
||||
provider.type = "supabase"
|
||||
provider.session = SupabaseSession(
|
||||
access_token=ACCESS_TOKEN,
|
||||
http_session=MagicMock(),
|
||||
)
|
||||
provider.identity = SupabaseIdentityInfo(
|
||||
organizations=[
|
||||
SupabaseOrganization(
|
||||
id=ORGANIZATION_ID,
|
||||
name=ORGANIZATION_NAME,
|
||||
slug=ORGANIZATION_SLUG,
|
||||
)
|
||||
]
|
||||
)
|
||||
provider.audit_config = {"max_retries": 0}
|
||||
provider.fixer_config = {}
|
||||
return provider
|
||||
@@ -0,0 +1,40 @@
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from prowler.providers.supabase.lib.mutelist.mutelist import SupabaseMutelist
|
||||
from tests.providers.supabase.supabase_fixtures import (
|
||||
ORGANIZATION_SLUG,
|
||||
USER_ID,
|
||||
)
|
||||
|
||||
|
||||
class TestSupabaseMutelist:
|
||||
@pytest.mark.parametrize(
|
||||
("resource_id", "expected"),
|
||||
[(USER_ID, True), ("another-user", False)],
|
||||
)
|
||||
def test_matches_organization_check_and_member(self, resource_id, expected):
|
||||
mutelist = SupabaseMutelist(
|
||||
mutelist_content={
|
||||
"Accounts": {
|
||||
ORGANIZATION_SLUG: {
|
||||
"Checks": {
|
||||
"organizations_member_mfa_enabled": {
|
||||
"Regions": ["global"],
|
||||
"Resources": [USER_ID],
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
)
|
||||
finding = MagicMock(
|
||||
organization_slug=ORGANIZATION_SLUG,
|
||||
resource_id=resource_id,
|
||||
resource_name=f"member {resource_id}",
|
||||
resource_tags=[],
|
||||
)
|
||||
finding.check_metadata.CheckID = "organizations_member_mfa_enabled"
|
||||
|
||||
assert mutelist.is_finding_muted(finding) is expected
|
||||
@@ -0,0 +1,254 @@
|
||||
import os
|
||||
from argparse import Namespace
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
from prowler.config.config import Provider as ProviderName
|
||||
from prowler.lib.check.models import CheckReportSupabase
|
||||
from prowler.lib.cli.parser import ProwlerArgumentParser
|
||||
from prowler.lib.outputs.finding import Finding
|
||||
from prowler.lib.outputs.html.html import HTML
|
||||
from prowler.providers.common.provider import Provider
|
||||
from prowler.providers.supabase.exceptions.exceptions import (
|
||||
SupabaseAuthenticationError,
|
||||
SupabaseCredentialsError,
|
||||
SupabaseInsufficientPermissionsError,
|
||||
SupabaseRateLimitError,
|
||||
)
|
||||
from prowler.providers.supabase.models import SupabaseOrganization, SupabaseSession
|
||||
from prowler.providers.supabase.services.organizations.organizations_service import (
|
||||
SupabaseOrganizationMember,
|
||||
)
|
||||
from prowler.providers.supabase.supabase_provider import SupabaseProvider
|
||||
from tests.providers.supabase.supabase_fixtures import (
|
||||
ACCESS_TOKEN,
|
||||
ORGANIZATION_ID,
|
||||
ORGANIZATION_NAME,
|
||||
ORGANIZATION_SLUG,
|
||||
USER_ID,
|
||||
)
|
||||
|
||||
|
||||
class TestSupabaseProvider:
|
||||
def test_setup_session_uses_environment_token(self):
|
||||
with mock.patch.dict(
|
||||
os.environ, {"SUPABASE_ACCESS_TOKEN": ACCESS_TOKEN}, clear=True
|
||||
):
|
||||
session = SupabaseProvider.setup_session()
|
||||
|
||||
assert session.access_token == ACCESS_TOKEN
|
||||
assert session.http_session.headers["Authorization"] == f"Bearer {ACCESS_TOKEN}"
|
||||
|
||||
def test_setup_session_requires_environment_token(self):
|
||||
with mock.patch.dict(os.environ, {}, clear=True):
|
||||
with pytest.raises(SupabaseCredentialsError):
|
||||
SupabaseProvider.setup_session()
|
||||
|
||||
def test_access_token_is_not_serialized_or_represented(self):
|
||||
session = SupabaseSession(access_token=ACCESS_TOKEN)
|
||||
|
||||
assert ACCESS_TOKEN not in repr(session)
|
||||
assert ACCESS_TOKEN not in str(session)
|
||||
assert ACCESS_TOKEN not in session.model_dump_json()
|
||||
assert "access_token" not in session.model_dump()
|
||||
|
||||
def test_setup_identity_lists_organizations_without_member_pii(self):
|
||||
session = SupabaseSession(
|
||||
access_token=ACCESS_TOKEN,
|
||||
http_session=mock.MagicMock(),
|
||||
)
|
||||
response = mock.MagicMock(status_code=200)
|
||||
response.json.return_value = [
|
||||
{
|
||||
"id": ORGANIZATION_ID,
|
||||
"name": ORGANIZATION_NAME,
|
||||
"slug": ORGANIZATION_SLUG,
|
||||
}
|
||||
]
|
||||
session.http_session.get.return_value = response
|
||||
|
||||
identity = SupabaseProvider.setup_identity(session, max_retries=0)
|
||||
|
||||
assert identity.organizations[0].slug == ORGANIZATION_SLUG
|
||||
session.http_session.get.assert_called_once_with(
|
||||
"https://api.supabase.com/v1/organizations", timeout=30
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("status_code", "exception"),
|
||||
[
|
||||
(401, SupabaseAuthenticationError),
|
||||
(403, SupabaseInsufficientPermissionsError),
|
||||
(429, SupabaseRateLimitError),
|
||||
],
|
||||
)
|
||||
def test_setup_identity_preserves_management_api_errors(
|
||||
self, status_code, exception
|
||||
):
|
||||
session = SupabaseSession(
|
||||
access_token=ACCESS_TOKEN,
|
||||
http_session=mock.MagicMock(),
|
||||
)
|
||||
session.http_session.get.return_value = mock.MagicMock(
|
||||
status_code=status_code,
|
||||
headers={"X-RateLimit-Reset": "0"},
|
||||
)
|
||||
|
||||
with pytest.raises(exception):
|
||||
SupabaseProvider.setup_identity(session, max_retries=0)
|
||||
|
||||
def test_from_cli_args_uses_environment_only(self):
|
||||
arguments = Namespace(
|
||||
config_file=None,
|
||||
mutelist_file=None,
|
||||
)
|
||||
|
||||
with (
|
||||
mock.patch.dict(
|
||||
os.environ, {"SUPABASE_ACCESS_TOKEN": ACCESS_TOKEN}, clear=True
|
||||
),
|
||||
mock.patch.object(
|
||||
SupabaseProvider,
|
||||
"setup_identity",
|
||||
return_value=mock.MagicMock(organizations=[]),
|
||||
),
|
||||
):
|
||||
provider = SupabaseProvider.from_cli_args(arguments, fixer_config={})
|
||||
|
||||
assert provider.type == "supabase"
|
||||
assert not hasattr(arguments, "supabase_access_token")
|
||||
|
||||
def test_parser_discovers_supabase_without_secret_argument(self):
|
||||
arguments = ProwlerArgumentParser().parse(
|
||||
["prowler", "supabase", "--list-checks"]
|
||||
)
|
||||
|
||||
assert arguments.provider == "supabase"
|
||||
assert not hasattr(arguments, "supabase_access_token")
|
||||
|
||||
def test_provider_registry_and_class_resolution(self):
|
||||
assert ProviderName.SUPABASE.value == "supabase"
|
||||
assert Provider.get_class("supabase") is SupabaseProvider
|
||||
assert SupabaseProvider.sdk_only is True
|
||||
|
||||
|
||||
class TestSupabaseProviderOutputHooks:
|
||||
def test_finding_output_uses_organization_and_member_ids(self):
|
||||
provider = SupabaseProvider.__new__(SupabaseProvider)
|
||||
provider._identity = mock.MagicMock(organizations=[])
|
||||
check_output = mock.MagicMock(
|
||||
organization_slug=ORGANIZATION_SLUG,
|
||||
organization_name=ORGANIZATION_NAME,
|
||||
resource_name="member user-id",
|
||||
resource_id="user-id",
|
||||
)
|
||||
|
||||
output = provider.get_finding_output_data(check_output)
|
||||
|
||||
assert output == {
|
||||
"auth_method": "personal_access_token",
|
||||
"account_uid": ORGANIZATION_SLUG,
|
||||
"account_name": ORGANIZATION_NAME,
|
||||
"resource_name": "member user-id",
|
||||
"resource_uid": "user-id",
|
||||
"region": "global",
|
||||
}
|
||||
|
||||
def test_finding_output_pipeline_uses_supabase_fields(self):
|
||||
provider = SupabaseProvider.__new__(SupabaseProvider)
|
||||
provider._identity = mock.MagicMock(organizations=[])
|
||||
member = SupabaseOrganizationMember(
|
||||
id=USER_ID,
|
||||
name=f"member {USER_ID}",
|
||||
organization_slug=ORGANIZATION_SLUG,
|
||||
organization_name=ORGANIZATION_NAME,
|
||||
mfa_enabled=False,
|
||||
)
|
||||
metadata = Path(
|
||||
"prowler/providers/supabase/services/organizations/"
|
||||
"organizations_member_mfa_enabled/"
|
||||
"organizations_member_mfa_enabled.metadata.json"
|
||||
).read_text()
|
||||
check_output = CheckReportSupabase(metadata=metadata, resource=member)
|
||||
check_output.status = "FAIL"
|
||||
check_output.status_extended = "Member does not have MFA enabled."
|
||||
|
||||
finding = Finding.generate_output(
|
||||
provider, check_output, Namespace(unix_timestamp=False)
|
||||
)
|
||||
|
||||
assert finding.provider == "supabase"
|
||||
assert finding.account_uid == ORGANIZATION_SLUG
|
||||
assert finding.account_name == ORGANIZATION_NAME
|
||||
assert finding.resource_name == f"member {USER_ID}"
|
||||
assert finding.resource_uid == USER_ID
|
||||
assert finding.region == "global"
|
||||
assert finding.auth_method == "personal_access_token"
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("output_filename", "expected"),
|
||||
[
|
||||
(None, f"prowler-output-{ORGANIZATION_SLUG}-"),
|
||||
("custom-report", "custom-report"),
|
||||
],
|
||||
)
|
||||
def test_output_options_use_organization_slug_or_explicit_name(
|
||||
self, output_filename, expected
|
||||
):
|
||||
provider = SupabaseProvider.__new__(SupabaseProvider)
|
||||
provider._identity = mock.MagicMock(
|
||||
organizations=[
|
||||
SupabaseOrganization(
|
||||
id=ORGANIZATION_ID,
|
||||
name=ORGANIZATION_NAME,
|
||||
slug=ORGANIZATION_SLUG,
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
output_options = provider.get_output_options(
|
||||
Namespace(output_filename=output_filename), {}
|
||||
)
|
||||
|
||||
if output_filename:
|
||||
assert output_options.output_filename == expected
|
||||
else:
|
||||
assert output_options.output_filename.startswith(expected)
|
||||
|
||||
def test_html_assessment_summary_uses_supabase_hook(self):
|
||||
provider = SupabaseProvider.__new__(SupabaseProvider)
|
||||
provider._identity = mock.MagicMock(
|
||||
organizations=[
|
||||
SupabaseOrganization(
|
||||
id=ORGANIZATION_ID,
|
||||
name=ORGANIZATION_NAME,
|
||||
slug=ORGANIZATION_SLUG,
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
summary = HTML.get_assessment_summary(provider)
|
||||
|
||||
assert "Supabase Assessment Summary" in summary
|
||||
assert f"<b>Organizations:</b> {ORGANIZATION_SLUG}" in summary
|
||||
assert "<b>Authentication:</b> Personal Access Token" in summary
|
||||
|
||||
def test_summary_and_stdout_hooks_are_global(self):
|
||||
provider = SupabaseProvider.__new__(SupabaseProvider)
|
||||
provider._identity = mock.MagicMock(
|
||||
organizations=[
|
||||
SupabaseOrganization(
|
||||
id=ORGANIZATION_ID,
|
||||
name=ORGANIZATION_NAME,
|
||||
slug=ORGANIZATION_SLUG,
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
assert provider.get_summary_entity() == (
|
||||
"Organization",
|
||||
f"{ORGANIZATION_NAME} ({ORGANIZATION_SLUG})",
|
||||
)
|
||||
assert provider.get_stdout_detail(mock.MagicMock()) == "global"
|
||||
Reference in New Issue
Block a user