Compare commits

...

1 Commits

Author SHA1 Message Date
pedrooot 8ef326b276 feat(scaleway): add new provider 2026-05-14 10:12:07 +02:00
39 changed files with 1634 additions and 17 deletions
+1
View File
@@ -120,6 +120,7 @@ Every AWS provider scan will enqueue an Attack Paths ingestion job automatically
| Google Workspace | 25 | 4 | 2 | 4 | Official | CLI |
| OpenStack | 34 | 5 | 0 | 9 | Official | UI, API, CLI |
| Vercel | 26 | 6 | 0 | 5 | Official | CLI |
| Scaleway | 1 | 1 | 0 | 1 | Unofficial | CLI |
| NHN | 6 | 2 | 1 | 0 | Unofficial | CLI |
> [!Note]
+6
View File
@@ -326,6 +326,12 @@
"user-guide/providers/openstack/authentication"
]
},
{
"group": "Scaleway",
"pages": [
"user-guide/providers/scaleway/getting-started-scaleway"
]
},
{
"group": "Vercel",
"pages": [
+1
View File
@@ -35,6 +35,7 @@ Prowler supports a wide range of providers organized by category:
| **NHN** | Unofficial | Tenants | CLI |
| [OpenStack](/user-guide/providers/openstack/getting-started-openstack) | Official | Projects | UI, API, CLI |
| [Oracle Cloud](/user-guide/providers/oci/getting-started-oci) | Official | Tenancies / Compartments | UI, API, CLI |
| [Scaleway](/user-guide/providers/scaleway/getting-started-scaleway) | Official | Organizations | CLI |
### Infrastructure as Code Providers
@@ -0,0 +1,51 @@
---
title: "Getting Started With Scaleway on Prowler"
---
Prowler for Scaleway scans IAM resources in your Scaleway organization for security misconfigurations. The current release ships one check that flags API keys still owned by the account root user.
## Prerequisites
1. A Scaleway organization with IAM access.
2. A Scaleway API key with at least the `IAMReadOnly` policy bound to a dedicated IAM user (do not use the account root user).
3. Your organization ID (visible at the top right of the Scaleway console).
## Authentication
Prowler reads credentials from the standard Scaleway environment variables:
| Variable | Purpose |
|---|---|
| `SCW_ACCESS_KEY` | API key access key |
| `SCW_SECRET_KEY` | API key secret key |
| `SCW_DEFAULT_ORGANIZATION_ID` | Optional, required when the key bearer is an application |
| `SCW_DEFAULT_PROJECT_ID` | Optional, default project for project-scoped resources |
| `SCW_DEFAULT_REGION` | Optional, defaults to `fr-par` |
Alternatively, pass them as CLI flags (`--access-key`, `--secret-key`, `--organization-id`, `--project-id`, `--region`). The CLI emits a warning when secrets are passed via the command line; environment variables are preferred.
## Run a scan
```bash
export SCW_ACCESS_KEY="SCW..."
export SCW_SECRET_KEY="..."
export SCW_DEFAULT_ORGANIZATION_ID="..."
prowler scaleway
```
To run only the IAM root-key check:
```bash
prowler scaleway --check iam_no_root_api_keys
```
## Checks shipped
| Check ID | Severity | Description |
|---|---|---|
| `iam_no_root_api_keys` | Critical | Fails when any Scaleway IAM API key is still owned by the account root user. |
## Required Scaleway permissions
The API key bearer needs read access to the IAM API in order to list users and API keys. The `IAMReadOnly` policy is sufficient. Refer to the [Scaleway IAM policy reference](https://www.scaleway.com/en/docs/identity-and-access-management/iam/reference-content/permission-sets/) for the full list of permissions.
Generated
+34 -2
View File
@@ -1,4 +1,4 @@
# This file is automatically @generated by Poetry 2.3.2 and should not be changed by hand.
# This file is automatically @generated by Poetry 2.3.4 and should not be changed by hand.
[[package]]
name = "about-time"
@@ -5896,6 +5896,38 @@ pydantic = ">=2.6.0"
ruamel-yaml = ">=0.17.21"
typing-extensions = ">=4.7.1"
[[package]]
name = "scaleway"
version = "2.10.3"
description = "Scaleway SDK for Python"
optional = false
python-versions = ">=3.10"
groups = ["main"]
files = [
{file = "scaleway-2.10.3-py3-none-any.whl", hash = "sha256:dbf381440d6caf37c878cf16445a63f4969a4aac2257c9b72c744d10ff223a0c"},
{file = "scaleway-2.10.3.tar.gz", hash = "sha256:b1f9dd1b1450767205234c6f5a345e5e25dc039c780253d698893b5c344ce594"},
]
[package.dependencies]
scaleway-core = "2.10.3"
[[package]]
name = "scaleway-core"
version = "2.10.3"
description = "Scaleway SDK for Python"
optional = false
python-versions = ">=3.10"
groups = ["main"]
files = [
{file = "scaleway_core-2.10.3-py3-none-any.whl", hash = "sha256:fd4112144554d6adae22ff737555eeb0e38cb1063250b3e88c9aebc1b957793b"},
{file = "scaleway_core-2.10.3.tar.gz", hash = "sha256:56432f755d694669429de51d51c1d0b3361b28dc2f939b28e4cb954610ee76be"},
]
[package.dependencies]
python-dateutil = ">=2.8.2,<3.0.0"
PyYAML = ">=6.0,<7.0"
requests = ">=2.28.1,<3.0.0"
[[package]]
name = "schema"
version = "0.7.5"
@@ -6735,4 +6767,4 @@ files = [
[metadata]
lock-version = "2.1"
python-versions = ">=3.10,<3.13"
content-hash = "d7e2ad41783a864bb845f63ccc10c88ae1e4ac36d61993ea106bbb4a5f58a843"
content-hash = "bd296ddc950772a2115177f7e5499a2489d341fed4a4d103ecc150f39158ba80"
+5
View File
@@ -156,6 +156,7 @@ from prowler.providers.mongodbatlas.models import MongoDBAtlasOutputOptions
from prowler.providers.nhn.models import NHNOutputOptions
from prowler.providers.openstack.models import OpenStackOutputOptions
from prowler.providers.oraclecloud.models import OCIOutputOptions
from prowler.providers.scaleway.models import ScalewayOutputOptions
from prowler.providers.vercel.models import VercelOutputOptions
@@ -426,6 +427,10 @@ def prowler():
output_options = VercelOutputOptions(
args, bulk_checks_metadata, global_provider.identity
)
elif provider == "scaleway":
output_options = ScalewayOutputOptions(
args, bulk_checks_metadata, global_provider.identity
)
# Run the quick inventory for the provider if available
if hasattr(args, "quick_inventory") and args.quick_inventory:
+1
View File
@@ -75,6 +75,7 @@ class Provider(str, Enum):
ALIBABACLOUD = "alibabacloud"
OPENSTACK = "openstack"
IMAGE = "image"
SCALEWAY = "scaleway"
VERCEL = "vercel"
+4
View File
@@ -741,6 +741,10 @@ def execute(
is_finding_muted_args["team_id"] = (
team.id if team else global_provider.identity.user_id
)
elif global_provider.type == "scaleway":
is_finding_muted_args["organization_id"] = (
global_provider.identity.organization_id
)
elif global_provider.type == "oraclecloud":
is_finding_muted_args["tenancy_id"] = (
global_provider.identity.tenancy_id
+47
View File
@@ -1283,6 +1283,53 @@ class CheckReportVercel(Check_Report):
return "global"
class CheckReportScaleway(Check_Report):
"""Contains the Scaleway Check's finding information.
Scaleway scans run at the organization level. Most IAM/account-level
resources are global; regional resources expose a ``region`` attribute
on the underlying object, which we surface as the report ``region``.
"""
resource_name: str
resource_id: str
organization_id: str
def __init__(
self,
metadata: Dict,
resource: Any,
resource_name: str = None,
resource_id: str = None,
organization_id: str = None,
) -> None:
"""Initialize the Scaleway Check's finding information.
Args:
metadata: Check metadata dictionary.
resource: The Scaleway resource being checked.
resource_name: Override for resource name.
resource_id: Override for resource ID.
organization_id: Override for the organization ID.
"""
super().__init__(metadata, resource)
self.resource_name = resource_name or getattr(
resource, "name", getattr(resource, "resource_name", "")
)
self.resource_id = resource_id or getattr(
resource, "id", getattr(resource, "resource_id", "")
)
self.organization_id = organization_id or getattr(
resource, "organization_id", ""
)
self._region = getattr(resource, "region", None) or "global"
@property
def region(self) -> str:
"""Scaleway regional resources expose their own region; IAM is global."""
return self._region
# Testing Pending
def load_check_metadata(metadata_file: str) -> CheckMetadata:
"""
+3 -2
View File
@@ -29,10 +29,10 @@ class ProwlerArgumentParser:
self.parser = argparse.ArgumentParser(
prog="prowler",
formatter_class=RawTextHelpFormatter,
usage="prowler [-h] [--version] {aws,azure,gcp,kubernetes,m365,github,googleworkspace,nhn,mongodbatlas,oraclecloud,alibabacloud,cloudflare,openstack,vercel,dashboard,iac,image,llm} ...",
usage="prowler [-h] [--version] {aws,azure,gcp,kubernetes,m365,github,googleworkspace,nhn,mongodbatlas,oraclecloud,alibabacloud,cloudflare,openstack,scaleway,vercel,dashboard,iac,image,llm} ...",
epilog="""
Available Cloud Providers:
{aws,azure,gcp,kubernetes,m365,github,googleworkspace,iac,llm,image,nhn,mongodbatlas,oraclecloud,alibabacloud,cloudflare,openstack,vercel}
{aws,azure,gcp,kubernetes,m365,github,googleworkspace,iac,llm,image,nhn,mongodbatlas,oraclecloud,alibabacloud,cloudflare,openstack,scaleway,vercel}
aws AWS Provider
azure Azure Provider
gcp GCP Provider
@@ -49,6 +49,7 @@ Available Cloud Providers:
image Container Image Provider
nhn NHN Provider (Unofficial)
mongodbatlas MongoDB Atlas Provider
scaleway Scaleway Provider
vercel Vercel Provider
Available components:
+12
View File
@@ -427,6 +427,18 @@ class Finding(BaseModel):
output_data["resource_uid"] = check_output.resource_id
output_data["region"] = "global"
elif provider.type == "scaleway":
output_data["auth_method"] = "api_key"
output_data["account_uid"] = get_nested_attribute(
provider, "identity.organization_id"
)
output_data["account_name"] = get_nested_attribute(
provider, "identity.bearer_email"
) or get_nested_attribute(provider, "identity.organization_id")
output_data["resource_name"] = check_output.resource_name
output_data["resource_uid"] = check_output.resource_id
output_data["region"] = check_output.region
elif provider.type == "alibabacloud":
output_data["auth_method"] = get_nested_attribute(
provider, "identity.identity_arn"
+77 -12
View File
@@ -73,8 +73,7 @@ class HTML(Output):
elif finding.status == "FAIL":
row_class = "table-danger"
self._data.append(
f"""
self._data.append(f"""
<tr class="{row_class}">
<td>{finding_status}</td>
<td>{finding.metadata.Severity.value}</td>
@@ -89,8 +88,7 @@ class HTML(Output):
<td><p class="show-read-more">{HTML.process_markdown(finding.metadata.Remediation.Recommendation.Text)}</p> <a class="read-more" href="{finding.metadata.Remediation.Recommendation.Url}"><i class="fas fa-external-link-alt"></i></a></td>
<td><p class="show-read-more">{parse_html_string(unroll_dict(finding.compliance, separator=": "))}</p></td>
</tr>
"""
)
""")
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -143,8 +141,7 @@ class HTML(Output):
from_cli (bool): whether the request is from the CLI or not
"""
try:
file_descriptor.write(
f"""<!DOCTYPE html>
file_descriptor.write(f"""<!DOCTYPE html>
<html lang="en">
<head>
<meta http-equiv="Content-Type" content="text/html; charset=UTF-8" />
@@ -253,8 +250,7 @@ class HTML(Output):
<th scope="col">Compliance</th>
</tr>
</thead>
<tbody>"""
)
<tbody>""")
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
@@ -269,8 +265,7 @@ class HTML(Output):
file_descriptor (file): the file descriptor to write the footer
"""
try:
file_descriptor.write(
"""
file_descriptor.write("""
</tbody>
</table>
</div>
@@ -409,8 +404,7 @@ class HTML(Output):
</body>
</html>
"""
)
""")
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
@@ -1400,6 +1394,77 @@ class HTML(Output):
)
return ""
@staticmethod
def get_scaleway_assessment_summary(provider: Provider) -> str:
"""
get_scaleway_assessment_summary gets the HTML assessment summary for the Scaleway provider
Args:
provider (Provider): the Scaleway provider object
Returns:
str: HTML assessment summary for the Scaleway provider
"""
try:
assessment_items = f"""
<li class="list-group-item">
<b>Organization ID:</b> {provider.identity.organization_id}
</li>"""
credentials_items = """
<li class="list-group-item">
<b>Authentication:</b> API Key
</li>"""
access_key = getattr(provider.session, "access_key", None)
if access_key:
credentials_items += f"""
<li class="list-group-item">
<b>Access Key:</b> {access_key}
</li>"""
bearer_type = getattr(provider.identity, "bearer_type", None)
bearer_email = getattr(provider.identity, "bearer_email", None)
bearer_id = getattr(provider.identity, "bearer_id", None)
if bearer_type:
bearer_label = bearer_email or bearer_id or "-"
credentials_items += f"""
<li class="list-group-item">
<b>Bearer:</b> {bearer_type} ({bearer_label})
</li>"""
region = getattr(provider.session, "default_region", None)
if region:
credentials_items += f"""
<li class="list-group-item">
<b>Default Region:</b> {region}
</li>"""
return f"""
<div class="col-md-2">
<div class="card">
<div class="card-header">
Scaleway Assessment Summary
</div>
<ul class="list-group list-group-flush">{assessment_items}
</ul>
</div>
</div>
<div class="col-md-4">
<div class="card">
<div class="card-header">
Scaleway Credentials
</div>
<ul class="list-group list-group-flush">{credentials_items}
</ul>
</div>
</div>"""
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
return ""
@staticmethod
def get_assessment_summary(provider: Provider) -> str:
"""
+2
View File
@@ -40,6 +40,8 @@ def stdout_report(finding, color, verbose, status, fix):
details = finding.location
if finding.check_metadata.Provider == "vercel":
details = finding.region
if finding.check_metadata.Provider == "scaleway":
details = finding.region
if (verbose or fix) and (not status or finding.status in status):
if finding.muted:
+3
View File
@@ -108,6 +108,9 @@ def display_summary_table(
)
else:
audited_entities = provider.identity.username or "Personal Account"
elif provider.type == "scaleway":
entity_type = "Organization"
audited_entities = provider.identity.organization_id
# Check if there are findings and that they are not all MANUAL
if findings and not all(finding.status == "MANUAL" for finding in findings):
+11
View File
@@ -403,6 +403,17 @@ class Provider(ABC):
mutelist_path=arguments.mutelist_file,
fixer_config=fixer_config,
)
elif "scaleway" in provider_class_name.lower():
provider_class(
access_key=getattr(arguments, "access_key", None),
secret_key=getattr(arguments, "secret_key", None),
organization_id=getattr(arguments, "organization_id", None),
project_id=getattr(arguments, "project_id", None),
region=getattr(arguments, "region", None),
config_path=arguments.config_file,
mutelist_path=arguments.mutelist_file,
fixer_config=fixer_config,
)
except TypeError as error:
logger.critical(
@@ -0,0 +1,99 @@
# Exceptions codes from 14000 to 14999 are reserved for Scaleway exceptions
from prowler.exceptions.exceptions import ProwlerException
class ScalewayBaseException(ProwlerException):
"""Base exception for Scaleway provider errors."""
SCALEWAY_ERROR_CODES = {
(14000, "ScalewayCredentialsError"): {
"message": "Scaleway credentials not found or invalid.",
"remediation": (
"Set the SCW_ACCESS_KEY and SCW_SECRET_KEY environment variables "
"with a valid Scaleway API key. Generate one at "
"https://console.scaleway.com/iam/api-keys."
),
},
(14001, "ScalewayAuthenticationError"): {
"message": "Authentication to the Scaleway API failed.",
"remediation": (
"Verify your Scaleway API key is valid, has not expired, and that "
"the bearer has IAM read permissions on the target organization."
),
},
(14002, "ScalewaySessionError"): {
"message": "Failed to create a Scaleway API session.",
"remediation": (
"Check network connectivity and ensure the Scaleway API is "
"reachable at https://api.scaleway.com."
),
},
(14003, "ScalewayIdentityError"): {
"message": "Failed to retrieve Scaleway identity information.",
"remediation": (
"Ensure the API key has permissions to read IAM users and the "
"owning organization metadata."
),
},
(14004, "ScalewayAPIError"): {
"message": "An error occurred while calling the Scaleway API.",
"remediation": (
"Check the Scaleway API status at https://status.scaleway.com "
"and retry. Run with --log-level DEBUG for the full traceback."
),
},
}
def __init__(self, code, file=None, original_exception=None, message=None):
provider = "Scaleway"
error_info = self.SCALEWAY_ERROR_CODES.get((code, self.__class__.__name__))
if error_info is None:
error_info = {
"message": message or "Unknown Scaleway error.",
"remediation": "Check the Scaleway API documentation for more details.",
}
elif message:
error_info = error_info.copy()
error_info["message"] = message
super().__init__(
code=code,
source=provider,
file=file,
original_exception=original_exception,
error_info=error_info,
)
class ScalewayCredentialsError(ScalewayBaseException):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
14000, file=file, original_exception=original_exception, message=message
)
class ScalewayAuthenticationError(ScalewayBaseException):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
14001, file=file, original_exception=original_exception, message=message
)
class ScalewaySessionError(ScalewayBaseException):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
14002, file=file, original_exception=original_exception, message=message
)
class ScalewayIdentityError(ScalewayBaseException):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
14003, file=file, original_exception=original_exception, message=message
)
class ScalewayAPIError(ScalewayBaseException):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
14004, file=file, original_exception=original_exception, message=message
)
@@ -0,0 +1,57 @@
SENSITIVE_ARGUMENTS = frozenset({"--access-key", "--secret-key"})
def init_parser(self):
"""Init the Scaleway provider CLI parser."""
scaleway_parser = self.subparsers.add_parser(
"scaleway",
parents=[self.common_providers_parser],
help="Scaleway Provider",
)
# Authentication
auth_subparser = scaleway_parser.add_argument_group("Authentication")
auth_subparser.add_argument(
"--access-key",
nargs="?",
default=None,
metavar="SCW_ACCESS_KEY",
help=(
"Scaleway API access key. Prefer the SCW_ACCESS_KEY env var "
"instead of passing it on the command line."
),
)
auth_subparser.add_argument(
"--secret-key",
nargs="?",
default=None,
metavar="SCW_SECRET_KEY",
help=(
"Scaleway API secret key. Prefer the SCW_SECRET_KEY env var "
"instead of passing it on the command line."
),
)
# Scope
scope_subparser = scaleway_parser.add_argument_group("Scope")
scope_subparser.add_argument(
"--organization-id",
nargs="?",
default=None,
metavar="SCW_DEFAULT_ORGANIZATION_ID",
help="Scaleway organization ID to scope the audit.",
)
scope_subparser.add_argument(
"--project-id",
nargs="?",
default=None,
metavar="SCW_DEFAULT_PROJECT_ID",
help="Default Scaleway project ID for project-scoped resources.",
)
scope_subparser.add_argument(
"--region",
nargs="?",
default=None,
metavar="SCW_DEFAULT_REGION",
help="Default Scaleway region (fr-par, nl-ams, pl-waw).",
)
@@ -0,0 +1,20 @@
from prowler.lib.check.models import CheckReportScaleway
from prowler.lib.mutelist.mutelist import Mutelist
from prowler.lib.outputs.utils import unroll_dict, unroll_tags
class ScalewayMutelist(Mutelist):
"""Scaleway-specific mutelist helper."""
def is_finding_muted(
self,
finding: CheckReportScaleway,
organization_id: str,
) -> bool:
return self.is_muted(
organization_id,
finding.check_metadata.CheckID,
finding.region or "global",
finding.resource_id or finding.resource_name,
unroll_dict(unroll_tags(finding.resource_tags)),
)
@@ -0,0 +1,44 @@
from prowler.lib.logger import logger
from prowler.providers.scaleway.exceptions.exceptions import ScalewayAPIError
class ScalewayService:
"""Base class for Scaleway services.
Centralizes the provider context (audit/fixer configuration, the
scoping organization, the authenticated ``scaleway.Client``) so each
service only worries about which Scaleway API to call.
"""
def __init__(self, service: str, provider):
self.provider = provider
self.audit_config = provider.audit_config
self.fixer_config = provider.fixer_config
self.service = service.lower() if not service.islower() else service
# Shared authenticated client and the organization in scope
self.client = provider.session.client
self.organization_id = provider.identity.organization_id
def _safe_call(self, label: str, fn, *args, **kwargs):
"""Run a Scaleway SDK call and surface failures as ScalewayAPIError.
Args:
label: Human-readable label for the call (used in logs).
fn: SDK function to invoke.
Returns:
The SDK function result, or ``None`` if the call failed.
"""
try:
return fn(*args, **kwargs)
except Exception as error:
logger.error(
f"{self.service} - {label} failed: "
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
raise ScalewayAPIError(
file=__file__,
original_exception=error,
message=f"Scaleway API call '{label}' failed.",
)
+53
View File
@@ -0,0 +1,53 @@
from typing import Any, Literal, Optional
from pydantic.v1 import BaseModel, Field
from prowler.config.config import output_file_timestamp
from prowler.providers.common.models import ProviderOutputOptions
ScalewayBearerType = Literal["user", "application"]
class ScalewaySession(BaseModel):
"""Scaleway API session information.
Stores the credentials and the underlying ``scaleway.Client`` so every
service can reuse the same authenticated client.
"""
access_key: str
secret_key: str
organization_id: Optional[str] = None
default_project_id: Optional[str] = None
default_region: Optional[str] = None
client: Any = Field(default=None, exclude=True)
class Config:
arbitrary_types_allowed = True
class ScalewayIdentityInfo(BaseModel):
"""Scaleway identity and scoping information."""
organization_id: str
bearer_id: Optional[str] = None
bearer_type: Optional[ScalewayBearerType] = None
bearer_email: Optional[str] = None
account_root_user_id: Optional[str] = None
class ScalewayOutputOptions(ProviderOutputOptions):
"""Customize output filenames for Scaleway scans."""
def __init__(self, arguments, bulk_checks_metadata, identity: ScalewayIdentityInfo):
super().__init__(arguments, bulk_checks_metadata)
if (
not hasattr(arguments, "output_filename")
or arguments.output_filename is None
):
account_fragment = identity.organization_id or "scaleway"
self.output_filename = (
f"prowler-output-{account_fragment}-{output_file_timestamp}"
)
else:
self.output_filename = arguments.output_filename
@@ -0,0 +1,372 @@
import os
from colorama import Fore, Style
from scaleway import Client
from scaleway.iam.v1alpha1 import IamV1Alpha1API
from prowler.config.config import (
default_config_file_path,
get_default_mute_file_path,
load_and_validate_config_file,
)
from prowler.lib.logger import logger
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.scaleway.exceptions.exceptions import (
ScalewayAuthenticationError,
ScalewayCredentialsError,
ScalewayIdentityError,
ScalewaySessionError,
)
from prowler.providers.scaleway.lib.mutelist.mutelist import ScalewayMutelist
from prowler.providers.scaleway.models import (
ScalewayIdentityInfo,
ScalewaySession,
)
class ScalewayProvider(Provider):
"""Scaleway provider.
Authenticates against the Scaleway API using an API key (access key +
secret key) and exposes a single global session that every service
reuses. Scaleway scopes everything to an organization, so the
organization ID is the audit identity.
"""
_type: str = "scaleway"
_session: ScalewaySession
_identity: ScalewayIdentityInfo
_audit_config: dict
_fixer_config: dict
_mutelist: ScalewayMutelist
audit_metadata: Audit_Metadata
def __init__(
self,
# Authentication credentials
access_key: str = None,
secret_key: str = None,
organization_id: str = None,
project_id: str = None,
region: str = None,
# Provider configuration
config_path: str = None,
config_content: dict | None = None,
fixer_config: dict = {},
mutelist_path: str = None,
mutelist_content: dict = None,
):
logger.info("Instantiating Scaleway provider...")
if config_content:
self._audit_config = config_content
else:
if not config_path:
config_path = default_config_file_path
self._audit_config = load_and_validate_config_file(self._type, config_path)
self._session = ScalewayProvider.setup_session(
access_key=access_key,
secret_key=secret_key,
organization_id=organization_id,
project_id=project_id,
region=region,
)
self._identity = ScalewayProvider.setup_identity(self._session)
self._fixer_config = fixer_config
if mutelist_content:
self._mutelist = ScalewayMutelist(mutelist_content=mutelist_content)
else:
if not mutelist_path:
mutelist_path = get_default_mute_file_path(self.type)
self._mutelist = ScalewayMutelist(mutelist_path=mutelist_path)
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) -> ScalewayMutelist:
return self._mutelist
@staticmethod
def setup_session(
access_key: str = None,
secret_key: str = None,
organization_id: str = None,
project_id: str = None,
region: str = None,
) -> ScalewaySession:
"""Initialize the Scaleway API session.
Credentials can be provided as arguments (for API/SDK use) or read
from the official Scaleway environment variables:
- ``SCW_ACCESS_KEY``
- ``SCW_SECRET_KEY``
- ``SCW_DEFAULT_ORGANIZATION_ID``
- ``SCW_DEFAULT_PROJECT_ID``
- ``SCW_DEFAULT_REGION``
Args:
access_key: Scaleway API access key.
secret_key: Scaleway API secret key.
organization_id: Default organization ID to scope the audit.
project_id: Default project ID for project-scoped resources.
region: Default region.
Returns:
ScalewaySession: The initialized session, holding the
authenticated ``scaleway.Client``.
Raises:
ScalewayCredentialsError: Access or secret key missing.
ScalewaySessionError: Client instantiation failed.
"""
access = access_key or os.environ.get("SCW_ACCESS_KEY", "")
secret = secret_key or os.environ.get("SCW_SECRET_KEY", "")
org = organization_id or os.environ.get("SCW_DEFAULT_ORGANIZATION_ID") or None
project = project_id or os.environ.get("SCW_DEFAULT_PROJECT_ID") or None
default_region = region or os.environ.get("SCW_DEFAULT_REGION") or "fr-par"
if not access or not secret:
raise ScalewayCredentialsError(
file=os.path.basename(__file__),
message=(
"Scaleway credentials not found. Provide access_key and "
"secret_key or set the SCW_ACCESS_KEY and SCW_SECRET_KEY "
"environment variables."
),
)
try:
client = Client(
access_key=access,
secret_key=secret,
default_organization_id=org,
default_project_id=project,
default_region=default_region,
)
return ScalewaySession(
access_key=access,
secret_key=secret,
organization_id=org,
default_project_id=project,
default_region=default_region,
client=client,
)
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
raise ScalewaySessionError(
file=os.path.basename(__file__),
original_exception=error,
)
@staticmethod
def setup_identity(session: ScalewaySession) -> ScalewayIdentityInfo:
"""Resolve the audit identity by calling Scaleway IAM.
Uses ``iam.get_api_key`` on the current access key to discover the
bearer (user vs application). When the bearer is a user, the
owning organization is read from the user record; otherwise we
require ``SCW_DEFAULT_ORGANIZATION_ID``.
"""
try:
iam = IamV1Alpha1API(session.client)
current_key = iam.get_api_key(access_key=session.access_key)
bearer_id = current_key.user_id or current_key.application_id
bearer_type = (
"user"
if current_key.user_id
else ("application" if current_key.application_id else None)
)
organization_id = session.organization_id
bearer_email = None
account_root_user_id = None
# If the bearer is a user, resolve the org from the user record
# and surface the email + root user id for the credentials banner.
if current_key.user_id:
user = iam.get_user(user_id=current_key.user_id)
organization_id = organization_id or user.organization_id
bearer_email = user.email
account_root_user_id = user.account_root_user_id
elif current_key.application_id and not organization_id:
# Application keys do not expose the org directly without an
# extra call. The default org from env is preferred.
logger.warning(
"Scaleway application-scoped API key without "
"SCW_DEFAULT_ORGANIZATION_ID. Resource discovery may fail."
)
if not organization_id:
raise ScalewayIdentityError(
file=os.path.basename(__file__),
message=(
"Could not determine the Scaleway organization ID. "
"Set SCW_DEFAULT_ORGANIZATION_ID or use a user-scoped "
"API key."
),
)
return ScalewayIdentityInfo(
organization_id=organization_id,
bearer_id=bearer_id,
bearer_type=bearer_type,
bearer_email=bearer_email,
account_root_user_id=account_root_user_id,
)
except ScalewayIdentityError:
raise
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
raise ScalewayIdentityError(
file=os.path.basename(__file__),
original_exception=error,
)
@staticmethod
def validate_credentials(session: ScalewaySession) -> None:
"""Smoke-test credentials by resolving the current API key.
Uses ``iam.get_api_key`` because it does not require any prior
knowledge of the bearer or the owning organization.
Args:
session: The Scaleway session to validate.
Raises:
ScalewayAuthenticationError: Authentication or authorization
failed against the Scaleway IAM API.
"""
try:
iam = IamV1Alpha1API(session.client)
iam.get_api_key(access_key=session.access_key)
except Exception as error:
raise ScalewayAuthenticationError(
file=os.path.basename(__file__),
original_exception=error,
)
def print_credentials(self) -> None:
report_title = (
f"{Style.BRIGHT}Using the Scaleway credentials below:{Style.RESET_ALL}"
)
report_lines = [
f"Authentication: {Fore.YELLOW}API Key{Style.RESET_ALL}",
f"Access Key: {Fore.YELLOW}{self._session.access_key}{Style.RESET_ALL}",
f"Organization ID: {Fore.YELLOW}{self._identity.organization_id}{Style.RESET_ALL}",
]
if self._identity.bearer_type:
report_lines.append(
f"Bearer: {Fore.YELLOW}{self._identity.bearer_type}"
f" ({self._identity.bearer_email or self._identity.bearer_id})"
f"{Style.RESET_ALL}"
)
if self._session.default_region:
report_lines.append(
f"Default Region: {Fore.YELLOW}{self._session.default_region}{Style.RESET_ALL}"
)
print_boxes(report_lines, report_title)
@staticmethod
def test_connection(
access_key: str = None,
secret_key: str = None,
organization_id: str = None,
raise_on_exception: bool = True,
provider_id: str = None,
) -> Connection:
"""Test connection to Scaleway.
Args:
access_key: Scaleway access key (falls back to SCW_ACCESS_KEY).
secret_key: Scaleway secret key (falls back to SCW_SECRET_KEY).
organization_id: Organization ID to scope the audit.
raise_on_exception: Whether to raise or return errors.
provider_id: Expected Scaleway organization ID. When provided,
the resolved identity must match it; otherwise the test
fails with ``ScalewayAuthenticationError``.
Returns:
Connection: Connection object with is_connected status.
"""
try:
session = ScalewayProvider.setup_session(
access_key=access_key,
secret_key=secret_key,
organization_id=organization_id,
)
ScalewayProvider.validate_credentials(session)
# Guard for API callers that already know the expected
# organization: the credentials must point to that exact org.
if provider_id:
identity = ScalewayProvider.setup_identity(session)
if identity.organization_id != provider_id:
raise ScalewayAuthenticationError(
file=os.path.basename(__file__),
message=(
"The provided credentials do not have access to "
f"the Scaleway organization with ID: {provider_id}"
),
)
return Connection(is_connected=True)
except (
ScalewayCredentialsError,
ScalewaySessionError,
ScalewayAuthenticationError,
) as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
if raise_on_exception:
raise error
return Connection(is_connected=False, error=error)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
formatted_error = ScalewayAuthenticationError(
file=os.path.basename(__file__),
original_exception=error,
)
if raise_on_exception:
raise formatted_error
return Connection(is_connected=False, error=formatted_error)
def validate_arguments(self) -> None:
return None
@@ -0,0 +1,4 @@
from prowler.providers.common.provider import Provider
from prowler.providers.scaleway.services.iam.iam_service import IAM
iam_client = IAM(Provider.get_global_provider())
@@ -0,0 +1,38 @@
{
"Provider": "scaleway",
"CheckID": "iam_no_root_api_keys",
"CheckTitle": "Scaleway IAM API keys must not be owned by the account root user",
"CheckType": [],
"ServiceName": "iam",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "critical",
"ResourceType": "ScalewayAPIKey",
"ResourceGroup": "IAM",
"Description": "**Scaleway API keys** are checked to ensure none is bound to the **account root user**. The account root user is the original Scaleway account owner; its credentials bypass IAM policies and grant unrestricted access to the entire organization.",
"Risk": "API keys owned by the **account root user** cannot be scoped down with IAM policies. Leaking one of these keys yields immediate full control over every project, resource and billing setting in the organization, and rotating them disrupts every automation depending on root credentials.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://www.scaleway.com/en/docs/identity-and-access-management/iam/concepts/#root-account",
"https://www.scaleway.com/en/docs/identity-and-access-management/iam/how-to/create-api-keys/",
"https://www.scaleway.com/en/docs/identity-and-access-management/iam/reference-content/users-and-applications/"
],
"Remediation": {
"Code": {
"CLI": "scw iam api-key delete <ACCESS_KEY>",
"NativeIaC": "",
"Other": "1. Sign in to the Scaleway console as a user with IAM admin permissions.\n2. Create a dedicated IAM user or application scoped with the minimum required policy.\n3. Generate a new API key for that bearer and roll it out to the workloads currently using the root key.\n4. Delete the API key owned by the account root user from the IAM > API keys page.",
"Terraform": ""
},
"Recommendation": {
"Text": "Never use API keys owned by the account root user for automation. Create scoped IAM users or applications, attach the least-privilege policies, and rotate any existing root API keys to that new bearer.",
"Url": "https://hub.prowler.com/check/iam_no_root_api_keys"
}
},
"Categories": [
"identity-access"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,87 @@
from typing import List
from prowler.lib.check.models import Check, CheckReportScaleway
from prowler.providers.scaleway.services.iam.iam_client import iam_client
class iam_no_root_api_keys(Check):
"""Ensure no Scaleway IAM API key is owned by the account root user.
The account root user is the original Scaleway account owner. API keys
bound to that bearer bypass IAM policies and grant unrestricted access
to the entire organization; rotating or losing them is a critical
incident. Day-to-day automation should rely on IAM users or
applications scoped through policies instead.
"""
def execute(self) -> List[CheckReportScaleway]:
"""Iterate over the API keys cached by the IAM service.
The check degrades to ``MANUAL`` when the IAM service could not
load the prerequisite data (users or API keys) — emitting ``PASS``
in those cases would silently mask the very condition the check
exists to detect.
Returns:
One ``CheckReportScaleway`` per discovered API key. ``FAIL``
when the bearer is the account root user, ``PASS`` otherwise.
A single ``MANUAL`` report is emitted when underlying IAM data
is unavailable.
"""
findings: List[CheckReportScaleway] = []
# If we could not even load the users we cannot tell who the root
# bearer is, so every API key would falsely PASS. Surface MANUAL
# explicitly so the operator investigates.
if not iam_client.users_loaded or not iam_client.api_keys_loaded:
placeholder = _IAMDataUnavailableResource(
organization_id=iam_client.organization_id
)
report = CheckReportScaleway(metadata=self.metadata(), resource=placeholder)
report.status = "MANUAL"
report.status_extended = (
"Could not retrieve Scaleway IAM users or API keys for "
f"organization {iam_client.organization_id}. Verify the "
"API key has the IAMReadOnly policy and rerun."
)
findings.append(report)
return findings
root_user_id = iam_client.account_root_user_id
for api_key in iam_client.api_keys:
report = CheckReportScaleway(metadata=self.metadata(), resource=api_key)
if root_user_id and api_key.user_id == root_user_id:
report.status = "FAIL"
report.status_extended = (
f"Scaleway API key {api_key.access_key} is owned by the "
f"account root user ({root_user_id}). Replace it with an "
f"API key bound to a dedicated IAM user or application."
)
else:
report.status = "PASS"
report.status_extended = (
f"Scaleway API key {api_key.access_key} is not owned by "
f"the account root user."
)
findings.append(report)
return findings
class _IAMDataUnavailableResource:
"""Minimal stand-in resource used when the IAM service failed to load.
``CheckReportScaleway`` derives ``resource_name``/``resource_id``/
``region``/``organization_id`` from the resource via ``getattr`` with
defaults, so this lightweight object is enough to materialize a
MANUAL finding without polluting the real domain models.
"""
def __init__(self, organization_id: str):
self.name = "iam-data-unavailable"
self.id = "iam-data-unavailable"
self.organization_id = organization_id
self.region = "global"
@@ -0,0 +1,142 @@
from typing import Optional
from pydantic.v1 import BaseModel
from scaleway.iam.v1alpha1 import IamV1Alpha1API
from prowler.lib.logger import logger
from prowler.providers.scaleway.lib.service.service import ScalewayService
class IAM(ScalewayService):
"""Scaleway IAM service.
Loads the users in scope plus every API key tied to the current
organization. Checks consume the materialized lists; nothing in this
class is lazy. Each load operation tracks success/failure separately
so checks can degrade to ``MANUAL`` when data is incomplete instead of
falsely passing.
"""
def __init__(self, provider):
super().__init__("iam", provider)
self._api = IamV1Alpha1API(self.client)
# Cached state — populated eagerly during construction
self.users: list[ScalewayUser] = []
self.api_keys: list[ScalewayAPIKey] = []
self.account_root_user_id: Optional[str] = None
# Load status flags — checks consult these to surface MANUAL when
# the underlying API call failed rather than reporting empty lists
# as a clean PASS.
self.users_loaded: bool = False
self.api_keys_loaded: bool = False
self._load_users()
self._load_api_keys()
def _load_users(self) -> None:
"""List every IAM user in the audited organization."""
try:
users = self._api.list_users_all(organization_id=self.organization_id)
for user in users:
self.users.append(
ScalewayUser(
id=user.id,
email=user.email,
username=user.username,
organization_id=user.organization_id,
account_root_user_id=user.account_root_user_id,
mfa=bool(getattr(user, "mfa", False)),
type_=(
str(user.type_) if getattr(user, "type_", None) else None
),
status=(
str(user.status) if getattr(user, "status", None) else None
),
)
)
# All users in the same org share the same account_root_user_id.
if self.users and self.users[0].account_root_user_id:
self.account_root_user_id = self.users[0].account_root_user_id
self.users_loaded = True
except Exception as error:
logger.error(
f"{self.service} - Error listing users: "
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def _load_api_keys(self) -> None:
"""List every API key in the audited organization."""
try:
api_keys = self._api.list_api_keys_all(organization_id=self.organization_id)
for key in api_keys:
self.api_keys.append(
ScalewayAPIKey(
access_key=key.access_key,
description=key.description,
user_id=key.user_id,
application_id=key.application_id,
default_project_id=key.default_project_id,
editable=bool(key.editable),
managed=bool(getattr(key, "managed", False)),
creation_ip=key.creation_ip,
created_at=str(key.created_at) if key.created_at else None,
updated_at=str(key.updated_at) if key.updated_at else None,
expires_at=str(key.expires_at) if key.expires_at else None,
)
)
self.api_keys_loaded = True
except Exception as error:
logger.error(
f"{self.service} - Error listing API keys: "
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
class ScalewayUser(BaseModel):
"""Subset of a Scaleway IAM user surface that the checks need."""
id: str
email: Optional[str] = None
username: Optional[str] = None
organization_id: Optional[str] = None
account_root_user_id: Optional[str] = None
mfa: bool = False
type_: Optional[str] = None
status: Optional[str] = None
# Provide name/id for CheckReportScaleway
name: str = ""
def __init__(self, **data):
super().__init__(**data)
self.name = self.email or self.username or self.id
class ScalewayAPIKey(BaseModel):
"""Subset of a Scaleway IAM API key surface that the checks need."""
access_key: str
description: Optional[str] = None
user_id: Optional[str] = None
application_id: Optional[str] = None
default_project_id: Optional[str] = None
editable: bool = False
managed: bool = False
creation_ip: Optional[str] = None
created_at: Optional[str] = None
updated_at: Optional[str] = None
expires_at: Optional[str] = None
# Provide name/id for CheckReportScaleway
name: str = ""
id: str = ""
def __init__(self, **data):
super().__init__(**data)
self.id = self.access_key
self.name = self.description or self.access_key
+2 -1
View File
@@ -87,7 +87,8 @@ dependencies = [
"alibabacloud_actiontrail20200706==2.4.1",
"alibabacloud_cs20151215==6.1.0",
"alibabacloud-rds20140815==12.0.0",
"alibabacloud-sls20201230==5.9.0"
"alibabacloud-sls20201230==5.9.0",
"scaleway==2.10.3"
]
description = "Prowler is an Open Source security tool to perform AWS, GCP and Azure security best practices assessments, audits, incident response, continuous monitoring, hardening and forensics readiness. It contains hundreds of controls covering CIS, NIST 800, NIST CSF, CISA, RBI, FedRAMP, PCI-DSS, GDPR, HIPAA, FFIEC, SOC2, GXP, AWS Well-Architected Framework Security Pillar, AWS Foundational Technical Review (FTR), ENS (Spanish National Security Scheme) and your custom security frameworks."
license = "Apache-2.0"
@@ -0,0 +1,96 @@
from unittest.mock import MagicMock
from prowler.providers.scaleway.models import (
ScalewayIdentityInfo,
ScalewaySession,
)
from prowler.providers.scaleway.services.iam.iam_service import (
ScalewayAPIKey,
ScalewayUser,
)
# Scaleway Identity
ORGANIZATION_ID = "b4ce0bfc-38fc-4c53-8757-548be64add26"
ROOT_USER_ID = "00000000-0000-0000-0000-000000000001"
MEMBER_USER_ID = "00000000-0000-0000-0000-000000000002"
APPLICATION_ID = "00000000-0000-0000-0000-000000000003"
BEARER_EMAIL = "pedro@prowler.com"
# Scaleway Credentials
ACCESS_KEY = "SCWAE000000000000000"
SECRET_KEY = "00000000-0000-0000-0000-000000000000"
# API Key Constants
ROOT_API_KEY = "SCWROOT00000000000000"
USER_API_KEY = "SCWUSER00000000000000"
APP_API_KEY = "SCWAPP000000000000000"
def set_mocked_scaleway_provider(
access_key: str = ACCESS_KEY,
secret_key: str = SECRET_KEY,
identity: ScalewayIdentityInfo = None,
audit_config: dict = None,
):
"""Create a mocked ScalewayProvider for testing."""
provider = MagicMock()
provider.type = "scaleway"
provider.session = ScalewaySession(
access_key=access_key,
secret_key=secret_key,
organization_id=ORGANIZATION_ID,
default_project_id=None,
default_region="fr-par",
client=MagicMock(),
)
provider.identity = identity or ScalewayIdentityInfo(
organization_id=ORGANIZATION_ID,
bearer_id=ROOT_USER_ID,
bearer_type="user",
bearer_email=BEARER_EMAIL,
account_root_user_id=ROOT_USER_ID,
)
provider.audit_config = audit_config or {}
provider.fixer_config = {}
return provider
def make_user(
user_id: str = ROOT_USER_ID,
email: str = BEARER_EMAIL,
account_root_user_id: str = ROOT_USER_ID,
mfa: bool = True,
) -> ScalewayUser:
return ScalewayUser(
id=user_id,
email=email,
username=email.split("@")[0] if email else None,
organization_id=ORGANIZATION_ID,
account_root_user_id=account_root_user_id,
mfa=mfa,
type_="owner" if user_id == account_root_user_id else "member",
status="activated",
)
def make_api_key(
access_key: str = USER_API_KEY,
user_id: str = MEMBER_USER_ID,
application_id: str = None,
description: str = "test key",
expires_at: str = None,
) -> ScalewayAPIKey:
return ScalewayAPIKey(
access_key=access_key,
description=description,
user_id=user_id,
application_id=application_id,
default_project_id=None,
editable=True,
managed=False,
creation_ip=None,
created_at="2026-01-01T00:00:00Z",
updated_at="2026-01-01T00:00:00Z",
expires_at=expires_at,
)
@@ -0,0 +1,106 @@
import os
from unittest import mock
import pytest
from prowler.providers.scaleway.exceptions.exceptions import (
ScalewayAuthenticationError,
ScalewayCredentialsError,
ScalewayIdentityError,
)
from prowler.providers.scaleway.models import ScalewaySession
from prowler.providers.scaleway.scaleway_provider import ScalewayProvider
from tests.providers.scaleway.scaleway_fixtures import (
ACCESS_KEY,
BEARER_EMAIL,
ORGANIZATION_ID,
ROOT_USER_ID,
SECRET_KEY,
)
class Test_ScalewayProvider_setup_session:
def test_missing_access_key_raises_credentials_error(self):
with mock.patch.dict(
os.environ, {"SCW_ACCESS_KEY": "", "SCW_SECRET_KEY": ""}, clear=False
):
os.environ.pop("SCW_ACCESS_KEY", None)
os.environ.pop("SCW_SECRET_KEY", None)
with pytest.raises(ScalewayCredentialsError):
ScalewayProvider.setup_session()
def test_returns_session_with_credentials(self):
session = ScalewayProvider.setup_session(
access_key=ACCESS_KEY,
secret_key=SECRET_KEY,
organization_id=ORGANIZATION_ID,
)
assert isinstance(session, ScalewaySession)
assert session.access_key == ACCESS_KEY
assert session.organization_id == ORGANIZATION_ID
assert session.default_region == "fr-par"
class Test_ScalewayProvider_setup_identity:
def _build_session(self):
return ScalewaySession(
access_key=ACCESS_KEY,
secret_key=SECRET_KEY,
organization_id=ORGANIZATION_ID,
default_region="fr-par",
client=mock.MagicMock(),
)
def test_resolves_user_bearer_identity(self):
session = self._build_session()
api_key = mock.MagicMock(user_id=ROOT_USER_ID, application_id=None)
user = mock.MagicMock(
email=BEARER_EMAIL,
organization_id=ORGANIZATION_ID,
account_root_user_id=ROOT_USER_ID,
)
with mock.patch(
"prowler.providers.scaleway.scaleway_provider.IamV1Alpha1API"
) as iam_cls:
iam = iam_cls.return_value
iam.get_api_key.return_value = api_key
iam.get_user.return_value = user
identity = ScalewayProvider.setup_identity(session)
assert identity.organization_id == ORGANIZATION_ID
assert identity.bearer_type == "user"
assert identity.bearer_id == ROOT_USER_ID
assert identity.bearer_email == BEARER_EMAIL
assert identity.account_root_user_id == ROOT_USER_ID
def test_missing_organization_raises_identity_error(self):
session = self._build_session()
session.organization_id = None
api_key = mock.MagicMock(user_id=None, application_id="app-id")
with mock.patch(
"prowler.providers.scaleway.scaleway_provider.IamV1Alpha1API"
) as iam_cls:
iam = iam_cls.return_value
iam.get_api_key.return_value = api_key
with pytest.raises(ScalewayIdentityError):
ScalewayProvider.setup_identity(session)
class Test_ScalewayProvider_validate_credentials:
def test_invalid_credentials_raise_authentication_error(self):
session = ScalewaySession(
access_key=ACCESS_KEY,
secret_key=SECRET_KEY,
organization_id=ORGANIZATION_ID,
client=mock.MagicMock(),
)
with mock.patch(
"prowler.providers.scaleway.scaleway_provider.IamV1Alpha1API"
) as iam_cls:
iam_cls.return_value.get_api_key.side_effect = Exception("expired")
with pytest.raises(ScalewayAuthenticationError):
ScalewayProvider.validate_credentials(session)
@@ -0,0 +1,172 @@
from unittest import mock
from tests.providers.scaleway.scaleway_fixtures import (
APP_API_KEY,
APPLICATION_ID,
MEMBER_USER_ID,
ORGANIZATION_ID,
ROOT_API_KEY,
ROOT_USER_ID,
USER_API_KEY,
make_api_key,
set_mocked_scaleway_provider,
)
def _patch_clients(iam_client_mock):
"""Patch both the provider and the iam_client singleton."""
return [
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_scaleway_provider(),
),
mock.patch(
"prowler.providers.scaleway.services.iam.iam_no_root_api_keys.iam_no_root_api_keys.iam_client",
new=iam_client_mock,
),
]
class Test_iam_no_root_api_keys:
def test_no_api_keys_returns_empty_findings(self):
iam_client = mock.MagicMock()
iam_client.users_loaded = True
iam_client.api_keys_loaded = True
iam_client.account_root_user_id = ROOT_USER_ID
iam_client.api_keys = []
iam_client.organization_id = ORGANIZATION_ID
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_scaleway_provider(),
),
mock.patch(
"prowler.providers.scaleway.services.iam.iam_no_root_api_keys.iam_no_root_api_keys.iam_client",
new=iam_client,
),
):
from prowler.providers.scaleway.services.iam.iam_no_root_api_keys.iam_no_root_api_keys import (
iam_no_root_api_keys,
)
result = iam_no_root_api_keys().execute()
assert result == []
def test_root_api_key_fails(self):
iam_client = mock.MagicMock()
iam_client.users_loaded = True
iam_client.api_keys_loaded = True
iam_client.account_root_user_id = ROOT_USER_ID
iam_client.api_keys = [
make_api_key(access_key=ROOT_API_KEY, user_id=ROOT_USER_ID)
]
iam_client.organization_id = ORGANIZATION_ID
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_scaleway_provider(),
),
mock.patch(
"prowler.providers.scaleway.services.iam.iam_no_root_api_keys.iam_no_root_api_keys.iam_client",
new=iam_client,
),
):
from prowler.providers.scaleway.services.iam.iam_no_root_api_keys.iam_no_root_api_keys import (
iam_no_root_api_keys,
)
result = iam_no_root_api_keys().execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].resource_id == ROOT_API_KEY
assert ROOT_USER_ID in result[0].status_extended
def test_user_api_key_passes(self):
iam_client = mock.MagicMock()
iam_client.users_loaded = True
iam_client.api_keys_loaded = True
iam_client.account_root_user_id = ROOT_USER_ID
iam_client.api_keys = [
make_api_key(access_key=USER_API_KEY, user_id=MEMBER_USER_ID)
]
iam_client.organization_id = ORGANIZATION_ID
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_scaleway_provider(),
),
mock.patch(
"prowler.providers.scaleway.services.iam.iam_no_root_api_keys.iam_no_root_api_keys.iam_client",
new=iam_client,
),
):
from prowler.providers.scaleway.services.iam.iam_no_root_api_keys.iam_no_root_api_keys import (
iam_no_root_api_keys,
)
result = iam_no_root_api_keys().execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].resource_id == USER_API_KEY
def test_application_api_key_passes(self):
iam_client = mock.MagicMock()
iam_client.users_loaded = True
iam_client.api_keys_loaded = True
iam_client.account_root_user_id = ROOT_USER_ID
iam_client.api_keys = [
make_api_key(
access_key=APP_API_KEY, user_id=None, application_id=APPLICATION_ID
)
]
iam_client.organization_id = ORGANIZATION_ID
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_scaleway_provider(),
),
mock.patch(
"prowler.providers.scaleway.services.iam.iam_no_root_api_keys.iam_no_root_api_keys.iam_client",
new=iam_client,
),
):
from prowler.providers.scaleway.services.iam.iam_no_root_api_keys.iam_no_root_api_keys import (
iam_no_root_api_keys,
)
result = iam_no_root_api_keys().execute()
assert len(result) == 1
assert result[0].status == "PASS"
def test_users_load_failure_returns_manual(self):
iam_client = mock.MagicMock()
iam_client.users_loaded = False
iam_client.api_keys_loaded = True
iam_client.account_root_user_id = None
iam_client.api_keys = [
make_api_key(access_key=ROOT_API_KEY, user_id=ROOT_USER_ID)
]
iam_client.organization_id = ORGANIZATION_ID
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_scaleway_provider(),
),
mock.patch(
"prowler.providers.scaleway.services.iam.iam_no_root_api_keys.iam_no_root_api_keys.iam_client",
new=iam_client,
),
):
from prowler.providers.scaleway.services.iam.iam_no_root_api_keys.iam_no_root_api_keys import (
iam_no_root_api_keys,
)
result = iam_no_root_api_keys().execute()
assert len(result) == 1
assert result[0].status == "MANUAL"
assert "Could not retrieve" in result[0].status_extended
@@ -0,0 +1,84 @@
from unittest import mock
from prowler.providers.scaleway.services.iam.iam_service import IAM
from tests.providers.scaleway.scaleway_fixtures import (
APPLICATION_ID,
MEMBER_USER_ID,
ORGANIZATION_ID,
ROOT_USER_ID,
USER_API_KEY,
set_mocked_scaleway_provider,
)
def _mock_user(
user_id: str, account_root_user_id: str = ROOT_USER_ID, email: str = "u@example.com"
):
user = mock.MagicMock()
user.id = user_id
user.email = email
user.username = email.split("@")[0]
user.organization_id = ORGANIZATION_ID
user.account_root_user_id = account_root_user_id
user.mfa = True
user.type_ = "owner" if user_id == account_root_user_id else "member"
user.status = "activated"
return user
def _mock_api_key(access_key: str, user_id: str = None, application_id: str = None):
key = mock.MagicMock()
key.access_key = access_key
key.description = "test"
key.user_id = user_id
key.application_id = application_id
key.default_project_id = None
key.editable = True
key.managed = False
key.creation_ip = None
key.created_at = None
key.updated_at = None
key.expires_at = None
return key
class Test_IAM_service:
def test_loads_users_and_api_keys(self):
provider = set_mocked_scaleway_provider()
with mock.patch(
"prowler.providers.scaleway.services.iam.iam_service.IamV1Alpha1API"
) as iam_cls:
api = iam_cls.return_value
api.list_users_all.return_value = [
_mock_user(ROOT_USER_ID),
_mock_user(MEMBER_USER_ID, email="m@example.com"),
]
api.list_api_keys_all.return_value = [
_mock_api_key(USER_API_KEY, user_id=MEMBER_USER_ID),
_mock_api_key("SCWAPP", application_id=APPLICATION_ID),
]
iam = IAM(provider)
assert iam.users_loaded is True
assert iam.api_keys_loaded is True
assert iam.account_root_user_id == ROOT_USER_ID
assert len(iam.users) == 2
assert len(iam.api_keys) == 2
def test_marks_users_unloaded_on_error(self):
provider = set_mocked_scaleway_provider()
with mock.patch(
"prowler.providers.scaleway.services.iam.iam_service.IamV1Alpha1API"
) as iam_cls:
api = iam_cls.return_value
api.list_users_all.side_effect = Exception("denied")
api.list_api_keys_all.return_value = []
iam = IAM(provider)
assert iam.users_loaded is False
assert iam.api_keys_loaded is True
assert iam.account_root_user_id is None