Compare commits

...
36 changed files with 1370 additions and 0 deletions
+5
View File
@@ -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/**"
+7
View File
@@ -185,6 +185,13 @@ modules:
- tests/providers/vercel/**
e2e: []
- name: sdk-supabase
match:
- prowler/providers/supabase/**
tests:
- tests/providers/supabase/**
e2e: []
# ============================================
# SDK - Lib modules
# ============================================
+7
View File
@@ -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": [
+1
View File
@@ -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
+1
View File
@@ -83,6 +83,7 @@ class Provider(str, Enum):
LINODE = "linode"
HUAWEICLOUD = "huaweicloud"
E2ENETWORKS = "e2enetworks"
SUPABASE = "supabase"
# Compliance
+5
View File
@@ -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
+2
View File
@@ -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,
}
+16
View File
@@ -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.",
)
+22
View File
@@ -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),
)
+42
View File
@@ -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())
@@ -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": ""
}
@@ -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>"""
@@ -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"),
]
@@ -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"