feat(openstack): Add 7 New Compute Security Checks (#9944)

This commit is contained in:
Daniel Barranquero
2026-02-16 11:46:48 +01:00
committed by GitHub
parent 90e317d39f
commit be516f1dfc
37 changed files with 3843 additions and 25 deletions
+1
View File
@@ -17,6 +17,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
- CSA CCM 4.0 for the Oracle Cloud provider [(#10057)](https://github.com/prowler-cloud/prowler/pull/10057)
- OCI regions updater script and CI workflow [(#10020)](https://github.com/prowler-cloud/prowler/pull/10020)
- `image` provider for container image scanning with Trivy integration [(#9984)](https://github.com/prowler-cloud/prowler/pull/9984)
- OpenStack compute 7 new checks [(#9944)](https://github.com/prowler-cloud/prowler/pull/9944)
- CSA CCM 4.0 for the Alibaba Cloud provider [(#10061)](https://github.com/prowler-cloud/prowler/pull/10061)
- ECS Exec (ECS-006) privilege escalation detection via `ecs:ExecuteCommand` + `ecs:DescribeTasks` [(#10066)](https://github.com/prowler-cloud/prowler/pull/10066)
@@ -0,0 +1,37 @@
{
"Provider": "openstack",
"CheckID": "compute_instance_config_drive_enabled",
"CheckTitle": "Compute instances have config drive enabled",
"CheckType": [],
"ServiceName": "compute",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "low",
"ResourceType": "OpenStackInstance",
"ResourceGroup": "compute",
"Description": "**OpenStack compute instances** (VMs) are evaluated to verify that **config drive** is enabled. Config drive provides metadata and user data via a virtual CD-ROM device instead of the metadata service (169.254.169.254). This improves security by eliminating network-based metadata access, which can be vulnerable to SSRF attacks and metadata service exploitation.",
"Risk": "Instances without config drive rely on the metadata service (169.254.169.254), vulnerable to SSRF attacks that extract credentials and SSH keys. Metadata service is vulnerable to spoofing in compromised networks and can become unavailable. Config drive eliminates this attack surface by providing metadata via virtual CD-ROM, removing dependency on network-accessible metadata.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://docs.openstack.org/nova/queens/user/config-drive.html",
"https://cloudinit.readthedocs.io/en/latest/reference/datasources/configdrive.html"
],
"Remediation": {
"Code": {
"CLI": "openstack server show <instance_id> -c config_drive\nopenstack server image create <instance_id> --name backup-snapshot\nopenstack server rebuild <instance_id> --image <image_id> --config-drive true",
"NativeIaC": "",
"Other": "1. Navigate to **Compute > Instances > Launch Instance**\n2. In **Configuration** or **Advanced Options** tab, enable **Config Drive**\n3. Complete instance launch\n\nNote: Config drive cannot be added to existing instances without rebuild",
"Terraform": "```hcl\n# Terraform: enable config drive for compute instance\nresource \"openstack_compute_instance_v2\" \"instance\" {\n name = \"secure-instance\"\n image_id = var.image_id\n flavor_id = var.flavor_id\n key_pair = var.key_pair_name\n security_groups = [\"default\"]\n # Enable config drive for secure metadata injection\n config_drive = true\n\n network {\n name = var.network_name\n }\n\n # Optional: cloud-init will automatically use config drive if available\n user_data = <<-EOF\n #cloud-config\n datasource_list: [ConfigDrive, OpenStack]\n EOF\n}\n```"
},
"Recommendation": {
"Text": "Enable config drive on all instances to eliminate metadata service attack surface. Config drive provides metadata via virtual CD-ROM instead of network-accessible service (169.254.169.254), preventing SSRF attacks. Combine with firewall rules blocking metadata service access from applications. Use config drive in air-gapped environments where metadata service is unavailable.",
"Url": "https://hub.prowler.com/check/compute_instance_config_drive_enabled"
}
},
"Categories": [
"trust-boundaries"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": "Config drive must be enabled at instance creation time and cannot be added to existing instances without rebuild. Some OpenStack deployments may not support config drive (e.g., bare metal provisioning). Config drive is read-only from the guest OS perspective."
}
@@ -0,0 +1,24 @@
from typing import List
from prowler.lib.check.models import Check, CheckReportOpenStack
from prowler.providers.openstack.services.compute.compute_client import compute_client
class compute_instance_config_drive_enabled(Check):
"""Ensure compute instances have config drive enabled for secure metadata injection."""
def execute(self) -> List[CheckReportOpenStack]:
findings: List[CheckReportOpenStack] = []
for instance in compute_client.instances:
report = CheckReportOpenStack(metadata=self.metadata(), resource=instance)
if instance.has_config_drive:
report.status = "PASS"
report.status_extended = f"Instance {instance.name} ({instance.id}) has config drive enabled for secure metadata injection."
else:
report.status = "FAIL"
report.status_extended = f"Instance {instance.name} ({instance.id}) does not have config drive enabled (relies on metadata service)."
findings.append(report)
return findings
@@ -0,0 +1,37 @@
{
"Provider": "openstack",
"CheckID": "compute_instance_isolated_private_network",
"CheckTitle": "Compute instances are isolated in private networks",
"CheckType": [],
"ServiceName": "compute",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "OpenStackInstance",
"ResourceGroup": "compute",
"Description": "**OpenStack compute instances** (VMs) are evaluated to verify **network isolation** by ensuring they have private IP addresses without mixed public/private exposure. Proper network segmentation requires instances to be deployed in private networks and accessed via controlled entry points (bastion hosts, VPN, load balancers) rather than direct public exposure.",
"Risk": "Instances with mixed public/private exposure or only public IPs lack network isolation, allowing unauthorized internet access that bypasses segmentation controls. Attackers can pivot from compromised public instances to internal infrastructure for lateral movement. Flat topology exposes internal services to internet attacks including DDoS and exploit attempts.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://docs.openstack.org/neutron/latest/admin/intro-os-networking.html",
"https://docs.openstack.org/security-guide/networking/architecture.html"
],
"Remediation": {
"Code": {
"CLI": "openstack server remove floating ip <instance_id> <floating_ip>",
"NativeIaC": "",
"Other": "1. Navigate to **Compute > Instances**\n2. Select instance with public IP\n3. Click **Actions > Disassociate Floating IP**\n4. Confirm disassociation\n5. Access instance via bastion host or VPN instead",
"Terraform": "```hcl\n# Terraform: deploy instance in isolated private network\n\n# Private network (no external gateway)\nresource \"openstack_networking_network_v2\" \"private\" {\n name = \"private-network\"\n admin_state_up = true\n}\n\nresource \"openstack_networking_subnet_v2\" \"private_subnet\" {\n name = \"private-subnet\"\n network_id = openstack_networking_network_v2.private.id\n cidr = \"10.0.1.0/24\"\n ip_version = 4\n # No gateway_ip means no external routing\n}\n\n# Instance in private network (no public IP)\nresource \"openstack_compute_instance_v2\" \"app_server\" {\n name = \"app-server\"\n image_id = var.image_id\n flavor_id = var.flavor_id\n key_pair = var.key_pair_name\n security_groups = [\"app-tier-sg\"]\n\n # Attach to private network ONLY\n network {\n uuid = openstack_networking_network_v2.private.id\n }\n\n # Do NOT associate floating IP\n}\n\n# Security group for app tier (least privilege)\nresource \"openstack_networking_secgroup_v2\" \"app_tier_sg\" {\n name = \"app-tier-sg\"\n description = \"App tier security group - private network only\"\n}\n\n# Allow SSH from bastion host only (not internet)\nresource \"openstack_networking_secgroup_rule_v2\" \"app_ssh_from_bastion\" {\n direction = \"ingress\"\n ethertype = \"IPv4\"\n protocol = \"tcp\"\n port_range_min = 22\n port_range_max = 22\n remote_ip_prefix = var.bastion_private_ip # e.g., 10.0.0.5/32\n security_group_id = openstack_networking_secgroup_v2.app_tier_sg.id\n}\n\n# Allow app traffic from web tier only\nresource \"openstack_networking_secgroup_rule_v2\" \"app_backend\" {\n direction = \"ingress\"\n ethertype = \"IPv4\"\n protocol = \"tcp\"\n port_range_min = 8080\n port_range_max = 8080\n remote_ip_prefix = var.web_tier_cidr # e.g., 10.0.2.0/24\n security_group_id = openstack_networking_secgroup_v2.app_tier_sg.id\n}\n```"
},
"Recommendation": {
"Text": "Deploy instances in private networks (RFC1918) with tiered architecture: web tier with public IPs, app/database tiers private only. Never mix public and private IPs on same instance. Use bastion hosts or VPN for operator access, security groups for least-privilege policies, and NAT gateways for outbound access. Enable flow logs to detect lateral movement attempts.",
"Url": "https://hub.prowler.com/check/compute_instance_isolated_private_network"
}
},
"Categories": [
"trust-boundaries"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": "This check flags instances with mixed public/private IPs or only public IPs as not properly isolated. Some architectures legitimately require dual-homed instances (e.g., NAT gateways, VPN endpoints). Review findings in context. Instances without any IPs are also flagged as they lack network configuration."
}
@@ -0,0 +1,63 @@
from typing import List
from prowler.lib.check.models import Check, CheckReportOpenStack
from prowler.providers.openstack.services.compute.compute_client import compute_client
from prowler.providers.openstack.services.compute.lib.ip import is_public_ip
class compute_instance_isolated_private_network(Check):
"""Ensure compute instances are isolated in private networks without mixed public/private exposure."""
def execute(self) -> List[CheckReportOpenStack]:
findings: List[CheckReportOpenStack] = []
for instance in compute_client.instances:
report = CheckReportOpenStack(metadata=self.metadata(), resource=instance)
private_ip_list = []
public_ip_list = []
# Classify IPs from networks dict using actual IP validation
for ip_list in instance.networks.values():
for ip in ip_list:
if is_public_ip(ip):
public_ip_list.append(ip)
else:
private_ip_list.append(ip)
# Also check SDK fields for IPs not present in networks
seen_ips = set(private_ip_list + public_ip_list)
for ip in [
instance.public_v4,
instance.public_v6,
instance.access_ipv4,
instance.access_ipv6,
]:
if ip and ip not in seen_ips and is_public_ip(ip):
public_ip_list.append(ip)
for ip in [instance.private_v4, instance.private_v6]:
if ip and ip not in seen_ips and not is_public_ip(ip):
private_ip_list.append(ip)
has_private_ips = bool(private_ip_list)
has_public_ips = bool(public_ip_list)
# Determine status based on IP classification
if has_private_ips and not has_public_ips:
report.status = "PASS"
ip_display = ", ".join(private_ip_list)
report.status_extended = f"Instance {instance.name} ({instance.id}) is properly isolated in private network with private IPs ({ip_display}) and no public exposure."
elif has_public_ips and has_private_ips:
report.status = "FAIL"
report.status_extended = f"Instance {instance.name} ({instance.id}) has mixed public and private network exposure (not properly isolated)."
elif has_public_ips and not has_private_ips:
report.status = "FAIL"
report.status_extended = f"Instance {instance.name} ({instance.id}) has only public IP addresses (no private network isolation)."
else:
# No IPs at all (edge case)
report.status = "FAIL"
report.status_extended = f"Instance {instance.name} ({instance.id}) has no network configuration (no IPs assigned)."
findings.append(report)
return findings
@@ -0,0 +1,38 @@
{
"Provider": "openstack",
"CheckID": "compute_instance_key_based_authentication",
"CheckTitle": "Compute instances use SSH key-based authentication",
"CheckType": [],
"ServiceName": "compute",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "high",
"ResourceType": "OpenStackInstance",
"ResourceGroup": "compute",
"Description": "**OpenStack compute instances** (VMs) are evaluated to verify that **SSH key-based authentication** is configured by checking for an assigned keypair. Password-based authentication is vulnerable to brute-force attacks, credential stuffing, and phishing. SSH keys provide cryptographic authentication resistant to these attacks.",
"Risk": "Instances without SSH key-based authentication are vulnerable to brute-force password attacks, credential stuffing, and password reuse. Attackers can test common passwords, intercept credentials, or exploit leaked passwords from other breaches. Successful SSH access enables malware injection, lateral movement, privilege escalation, and data exfiltration.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://docs.openstack.org/security-guide/compute/hardening-the-virtualization-layers.html",
"https://docs.openstack.org/api-ref/compute/#keypairs-keypairs"
],
"Remediation": {
"Code": {
"CLI": "chmod 700 ~/.ssh && chmod 600 ~/.ssh/authorized_keys\nsudo systemctl restart sshd\nssh -i ~/.ssh/private_key username@instance_ip",
"NativeIaC": "",
"Other": "1. Navigate to **Compute > Instances > Launch Instance**\n2. In **Key Pair** tab, select existing keypair or create new one\n3. Launch instance with keypair attached",
"Terraform": "```hcl\n# Terraform: ensure compute instance uses SSH key-based authentication\n\n# Create or import keypair\nresource \"openstack_compute_keypair_v2\" \"keypair\" {\n name = \"production-keypair\"\n public_key = file(\"~/.ssh/id_rsa.pub\")\n}\n\nresource \"openstack_compute_instance_v2\" \"instance\" {\n name = \"secure-instance\"\n image_id = var.image_id\n flavor_id = var.flavor_id\n # Critical: attach keypair for SSH key-based authentication\n key_pair = openstack_compute_keypair_v2.keypair.name\n security_groups = [\"default\"]\n\n network {\n name = var.network_name\n }\n\n # Optional: use cloud-init to enforce key-only authentication\n user_data = <<-EOF\n #cloud-config\n ssh_pwauth: false\n disable_root: true\n EOF\n}\n```"
},
"Recommendation": {
"Text": "Always use SSH keys instead of passwords for instance authentication. Generate keys with strong algorithms (RSA 4096-bit, Ed25519). Protect private keys with passphrases and store securely. Disable password authentication in sshd_config. Rotate keypairs periodically and revoke old keys. Use bastion hosts or VPN for SSH access instead of direct internet exposure.",
"Url": "https://hub.prowler.com/check/compute_instance_key_based_authentication"
}
},
"Categories": [
"identity-access",
"encryption"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": "This check verifies keypair assignment via OpenStack metadata. It does not validate authorized_keys file contents or SSH daemon configuration inside the instance. Manual verification may be needed for instances launched without keypairs but later configured with keys."
}
@@ -0,0 +1,24 @@
from typing import List
from prowler.lib.check.models import Check, CheckReportOpenStack
from prowler.providers.openstack.services.compute.compute_client import compute_client
class compute_instance_key_based_authentication(Check):
"""Ensure compute instances use SSH key-based authentication instead of passwords."""
def execute(self) -> List[CheckReportOpenStack]:
findings: List[CheckReportOpenStack] = []
for instance in compute_client.instances:
report = CheckReportOpenStack(metadata=self.metadata(), resource=instance)
if instance.key_name:
report.status = "PASS"
report.status_extended = f"Instance {instance.name} ({instance.id}) is configured with SSH key-based authentication (keypair: {instance.key_name})."
else:
report.status = "FAIL"
report.status_extended = f"Instance {instance.name} ({instance.id}) does not have SSH key-based authentication configured (no keypair assigned)."
findings.append(report)
return findings
@@ -0,0 +1,37 @@
{
"Provider": "openstack",
"CheckID": "compute_instance_locked_status_enabled",
"CheckTitle": "Compute instances have locked status enabled",
"CheckType": [],
"ServiceName": "compute",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "OpenStackInstance",
"ResourceGroup": "compute",
"Description": "**OpenStack compute instances** (VMs) are evaluated to verify that **locked status** is enabled. Locking an instance prevents unauthorized administrative operations (delete, resize, rebuild, etc.) without first unlocking it. This provides an additional layer of protection against accidental or malicious modifications.",
"Risk": "Instances without locked status can be subjected to unauthorized operations (deletion, resize, rebuild) by compromised accounts without additional barriers. Attackers can manipulate unlocked instances to destroy forensic evidence or disrupt production workloads. Accidental termination by operators also poses risk due to lack of change control barriers.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://docs.openstack.org/api-ref/compute/#lock-server-lock",
"https://docs.openstack.org/security-guide/compute.html"
],
"Remediation": {
"Code": {
"CLI": "openstack server lock <instance_id> --reason \"Production instance - requires approval\"\nopenstack server show <instance_id>",
"NativeIaC": "",
"Other": "1. Navigate to **Compute > Instances**\n2. Select the instance to protect\n3. Use CLI or API to lock the instance\n4. Verify locked status is enabled",
"Terraform": "```hcl\n# Note: Terraform openstack_compute_instance_v2 does not support locked status natively\n# Use null_resource with local-exec provisioner as workaround\n\nresource \"openstack_compute_instance_v2\" \"instance\" {\n name = \"production-instance\"\n image_id = var.image_id\n flavor_id = var.flavor_id\n key_pair = var.key_pair_name\n security_groups = [\"default\"]\n\n network {\n name = var.network_name\n }\n}\n\nresource \"null_resource\" \"lock_instance\" {\n depends_on = [openstack_compute_instance_v2.instance]\n\n provisioner \"local-exec\" {\n command = \"openstack server lock ${openstack_compute_instance_v2.instance.id}\"\n }\n}\n```"
},
"Recommendation": {
"Text": "Lock production instances to prevent accidental or unauthorized modifications. Use --reason parameter to document lock purpose. Implement approval workflows requiring consent before unlocking. Apply locks to critical infrastructure (databases, authentication, logging). Use RBAC policies to restrict who can lock/unlock. Audit lock/unlock operations via OpenStack logs.",
"Url": "https://hub.prowler.com/check/compute_instance_locked_status_enabled"
}
},
"Categories": [
"resilience"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": "Locked status prevents administrative operations like delete, resize, rebuild, and shelve. It does not prevent start/stop/reboot operations or guest OS-level changes. Lock operations require specific Nova API permissions."
}
@@ -0,0 +1,29 @@
from typing import List
from prowler.lib.check.models import Check, CheckReportOpenStack
from prowler.providers.openstack.services.compute.compute_client import compute_client
class compute_instance_locked_status_enabled(Check):
"""Ensure compute instances have locked status enabled to prevent unauthorized operations."""
def execute(self) -> List[CheckReportOpenStack]:
findings: List[CheckReportOpenStack] = []
for instance in compute_client.instances:
report = CheckReportOpenStack(metadata=self.metadata(), resource=instance)
if instance.is_locked:
report.status = "PASS"
reason = (
f" (reason: {instance.locked_reason})"
if instance.locked_reason
else ""
)
report.status_extended = f"Instance {instance.name} ({instance.id}) has locked status enabled{reason}."
else:
report.status = "FAIL"
report.status_extended = f"Instance {instance.name} ({instance.id}) does not have locked status enabled."
findings.append(report)
return findings
@@ -0,0 +1,38 @@
{
"Provider": "openstack",
"CheckID": "compute_instance_metadata_sensitive_data",
"CheckTitle": "Compute instance metadata does not contain sensitive data",
"CheckType": [],
"ServiceName": "compute",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "critical",
"ResourceType": "OpenStackInstance",
"ResourceGroup": "compute",
"Description": "**OpenStack compute instance metadata** is evaluated to detect **sensitive data** such as passwords, API keys, secrets, and private keys. Instance metadata is accessible via the metadata service (169.254.169.254) to any process inside the instance. Storing secrets in metadata exposes them to SSRF attacks, compromised applications, and unauthorized access.",
"Risk": "Instance metadata containing sensitive data exposes credentials through the metadata service (169.254.169.254), accessible to any process inside the instance. Attackers exploiting SSRF, compromised applications, or insider threats can extract passwords, API keys, and private keys. Stolen credentials enable unauthorized modifications, privilege escalation, resource deletion, and cryptomining.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://docs.openstack.org/nova/latest/user/metadata.html",
"https://cheatsheetseries.owasp.org/cheatsheets/Secrets_Management_Cheat_Sheet.html"
],
"Remediation": {
"Code": {
"CLI": "openstack server unset --property <sensitive_key> <instance_id>",
"NativeIaC": "",
"Other": "1. Navigate to **Compute > Instances**\n2. Select instance with sensitive metadata\n3. Remove sensitive metadata keys using CLI command\n4. Rotate exposed credentials immediately\n5. Store secrets in Barbican or external secrets manager instead",
"Terraform": "```hcl\n# Terraform: use Barbican for secrets instead of metadata\n\n# Store secret in Barbican\nresource \"openstack_keymanager_secret_v1\" \"db_password\" {\n name = \"database-password\"\n payload = random_password.db_password.result\n payload_content_type = \"text/plain\"\n secret_type = \"passphrase\"\n}\n\nresource \"random_password\" \"db_password\" {\n length = 32\n special = true\n}\n\n# Instance WITHOUT sensitive data in metadata\nresource \"openstack_compute_instance_v2\" \"secure_instance\" {\n name = \"app-server\"\n image_id = var.image_id\n flavor_id = var.flavor_id\n key_pair = var.key_pair_name\n security_groups = [\"app-sg\"]\n\n # Safe metadata (non-sensitive labels only)\n metadata = {\n environment = \"production\"\n application = \"web-app\"\n cost_center = \"engineering\"\n barbican_secret_id = openstack_keymanager_secret_v1.db_password.secret_ref\n }\n\n network {\n name = \"private-network\"\n }\n\n # Use cloud-init to retrieve secrets from Barbican\n user_data = <<-EOF\n #!/bin/bash\n # Retrieve database password from Barbican\n SECRET_REF=\"${openstack_keymanager_secret_v1.db_password.secret_ref}\"\n DB_PASSWORD=$(openstack secret get \"$SECRET_REF\" --payload -f value)\n \n # Configure application with retrieved secret\n echo \"DATABASE_URL=<scheme>://user:$DB_PASSWORD@db-host/dbname\" > /etc/app/config.env\n chmod 600 /etc/app/config.env\n EOF\n}\n\n# ANTI-PATTERN: DO NOT DO THIS\n# resource \"openstack_compute_instance_v2\" \"insecure_instance\" {\n# metadata = {\n# db_password = \"hardcoded-secret-123\" # NEVER store secrets in metadata\n# api_key = \"sk-1234567890abcdef\" # Exposed via metadata service\n# }\n# }\n```"
},
"Recommendation": {
"Text": "Never store secrets in metadata; use Barbican (OpenStack Key Manager), Vault, or external secrets management instead. Retrieve secrets at runtime via APIs. Implement least privilege access to secrets with RBAC. Enable secrets audit logging. Use envelope encryption for secrets at rest. Implement automatic rotation every 90 days. Scan metadata for hardcoded secrets using tools like TruffleHog.",
"Url": "https://hub.prowler.com/check/compute_instance_metadata_sensitive_data"
}
},
"Categories": [
"secrets",
"encryption"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": "This check uses the detect-secrets library to scan for credentials. May produce false positives on metadata keys containing secret-like keywords. Findings should be reviewed manually. The audit_config allows configuring secrets_ignore_patterns to exclude specific patterns and detect_secrets_plugins to customize detection. Metadata is world-readable within instance via 169.254.169.254."
}
@@ -0,0 +1,58 @@
import json
from typing import List
from prowler.lib.check.models import Check, CheckReportOpenStack
from prowler.lib.utils.utils import detect_secrets_scan
from prowler.providers.openstack.services.compute.compute_client import compute_client
class compute_instance_metadata_sensitive_data(Check):
"""Ensure compute instance metadata does not contain sensitive data like passwords or API keys."""
def execute(self) -> List[CheckReportOpenStack]:
findings: List[CheckReportOpenStack] = []
secrets_ignore_patterns = compute_client.audit_config.get(
"secrets_ignore_patterns", []
)
for instance in compute_client.instances:
report = CheckReportOpenStack(metadata=self.metadata(), resource=instance)
report.status = "PASS"
report.status_extended = f"Instance {instance.name} ({instance.id}) metadata does not contain sensitive data."
if instance.metadata:
# Build metadata dict and parallel list of keys (similar to AWS ECS pattern)
dump_metadata = {}
original_metadata_keys = []
for key, value in instance.metadata.items():
dump_metadata[key] = value
original_metadata_keys.append(key)
# Convert metadata dict to JSON string for detect-secrets scanning
metadata_json = json.dumps(dump_metadata, indent=2)
detect_secrets_output = detect_secrets_scan(
data=metadata_json,
excluded_secrets=secrets_ignore_patterns,
detect_secrets_plugins=compute_client.audit_config.get(
"detect_secrets_plugins"
),
)
if detect_secrets_output:
# Map line numbers back to metadata keys using the parallel list
# Line numbering: line 1 = "{", line 2 = first key-value, etc.
secrets_string = ", ".join(
[
f"{secret['type']} in metadata key '{original_metadata_keys[secret['line_number'] - 2]}'"
for secret in detect_secrets_output
if secret["line_number"] - 2 < len(original_metadata_keys)
]
)
report.status = "FAIL"
report.status_extended = f"Instance {instance.name} ({instance.id}) metadata contains potential secrets -> {secrets_string}."
else:
report.status_extended = f"Instance {instance.name} ({instance.id}) has no metadata (no sensitive data exposure risk)."
findings.append(report)
return findings
@@ -0,0 +1,39 @@
{
"Provider": "openstack",
"CheckID": "compute_instance_public_ip_exposed",
"CheckTitle": "Compute instances are not exposed to the internet",
"CheckType": [],
"ServiceName": "compute",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "OpenStackInstance",
"ResourceGroup": "compute",
"Description": "**OpenStack compute instances** are evaluated to verify they are **not exposed to the internet** via public IPs (floating IPs or access IPs). Instances with public IPs are directly reachable from the internet, increasing attack surface. Best practices recommend using **bastion hosts**, **VPN gateways**, or **load balancers** instead.",
"Risk": "Instances with public IPs are directly reachable from the internet, enabling reconnaissance, port scanning, and vulnerability exploitation. Attackers can target instances for brute-force attacks, credential stuffing, and malware injection. Public exposure bypasses network segmentation and defense-in-depth. Compromised public instances become pivot points for lateral movement.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://docs.openstack.org/neutron/latest/admin/intro-nat.html",
"https://docs.openstack.org/security-guide/networking/architecture.html",
"https://docs.openstack.org/neutron/latest/admin/intro-os-networking.html"
],
"Remediation": {
"Code": {
"CLI": "openstack server remove floating ip <instance_id> <floating_ip>\nssh -J bastion private-instance",
"NativeIaC": "",
"Other": "1. Navigate to **Compute > Instances**\n2. Select instance with public IP\n3. Click **Actions > Disassociate Floating IP**\n4. Confirm disassociation\n5. Access via bastion host or VPN instead",
"Terraform": "```hcl\n# Terraform: deploy instance without public IP (private network only)\n\nresource \"openstack_compute_instance_v2\" \"private_instance\" {\n name = \"private-instance\"\n image_id = var.image_id\n flavor_id = var.flavor_id\n key_pair = var.key_pair_name\n security_groups = [\"internal-sg\"]\n\n # Attach to private network ONLY (no floating IP)\n network {\n name = \"private-network\"\n }\n\n # Do NOT create floating IP association\n}\n\n# Bastion host for SSH access (only instance with public IP)\nresource \"openstack_compute_instance_v2\" \"bastion\" {\n name = \"bastion-host\"\n image_id = var.bastion_image_id\n flavor_id = \"small\"\n key_pair = var.bastion_keypair\n security_groups = [\"bastion-sg\"] # Restrict SSH to corporate IPs only\n\n network {\n name = \"dmz-network\"\n }\n}\n\nresource \"openstack_networking_floatingip_v2\" \"bastion_fip\" {\n pool = \"public\"\n}\n\nresource \"openstack_compute_floatingip_associate_v2\" \"bastion_fip_assoc\" {\n floating_ip = openstack_networking_floatingip_v2.bastion_fip.address\n instance_id = openstack_compute_instance_v2.bastion.id\n}\n\n# Security group for bastion (least privilege)\nresource \"openstack_networking_secgroup_v2\" \"bastion_sg\" {\n name = \"bastion-sg\"\n description = \"Allow SSH from corporate IPs only\"\n}\n\nresource \"openstack_networking_secgroup_rule_v2\" \"bastion_ssh\" {\n direction = \"ingress\"\n ethertype = \"IPv4\"\n protocol = \"tcp\"\n port_range_min = 22\n port_range_max = 22\n remote_ip_prefix = var.corporate_cidr # e.g., 203.0.113.0/24\n security_group_id = openstack_networking_secgroup_v2.bastion_sg.id\n}\n```"
},
"Recommendation": {
"Text": "Avoid public IPs on application/database instances; use private networks only. Deploy bastion hosts or VPN gateways for operator access. Use load balancers (Octavia, HAProxy) with public IPs for web traffic instead of direct instance exposure. Apply least-privilege security groups. Enable flow logs. Use cloud NAT gateways for outbound access without inbound exposure.",
"Url": "https://hub.prowler.com/check/compute_instance_public_ip_exposed"
}
},
"Categories": [
"internet-exposed",
"trust-boundaries"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": "This check flags instances with any public IP (floating IP, access IP v4/v6). Some workloads legitimately require public IPs (bastion hosts, NAT gateways, public-facing load balancers). Review findings in context of architecture requirements. Private RFC1918 IPs (10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16) are not flagged."
}
@@ -0,0 +1,62 @@
from typing import List
from prowler.lib.check.models import Check, CheckReportOpenStack
from prowler.providers.openstack.services.compute.compute_client import compute_client
from prowler.providers.openstack.services.compute.lib.ip import is_public_ip
class compute_instance_public_ip_exposed(Check):
"""Ensure compute instances are not exposed to the internet via public IP addresses."""
def execute(self) -> List[CheckReportOpenStack]:
findings: List[CheckReportOpenStack] = []
for instance in compute_client.instances:
report = CheckReportOpenStack(metadata=self.metadata(), resource=instance)
# Collect all potential public IP indicators
public_ips = []
# Check SDK computed properties
if instance.public_v4:
public_ips.append(f"Public IPv4: {instance.public_v4}")
if instance.public_v6:
public_ips.append(f"Public IPv6: {instance.public_v6}")
if instance.access_ipv4:
public_ips.append(f"Access IPv4: {instance.access_ipv4}")
if instance.access_ipv6:
public_ips.append(f"Access IPv6: {instance.access_ipv6}")
# Check networks for any additional public IPs (beyond first one captured in SDK attributes)
# This handles cases where instances have multiple public IPs on different networks
sdk_ips = {
instance.public_v4,
instance.public_v6,
instance.access_ipv4,
instance.access_ipv6,
}
for network_name, ip_list in instance.networks.items():
for ip in ip_list:
# Check if IP is public and not already captured in SDK attributes
if is_public_ip(ip) and ip not in sdk_ips:
public_ips.append(
f"{network_name} network: {ip} (public range)"
)
# Remove duplicates while preserving order
seen = set()
unique_public_ips = []
for ip in public_ips:
if ip not in seen:
seen.add(ip)
unique_public_ips.append(ip)
if not unique_public_ips:
report.status = "PASS"
report.status_extended = f"Instance {instance.name} ({instance.id}) is not exposed to the internet (no public IP addresses or external network attachments detected)."
else:
report.status = "FAIL"
ip_list = ", ".join(unique_public_ips)
report.status_extended = f"Instance {instance.name} ({instance.id}) is exposed to the internet with public IP addresses: {ip_list}."
findings.append(report)
return findings
@@ -10,23 +10,21 @@
"ResourceType": "OpenStackInstance",
"ResourceGroup": "compute",
"Description": "**OpenStack compute instances** (VMs) are evaluated to verify that at least one **security group** is attached. Security groups act as virtual firewalls, controlling ingress and egress traffic. Instances without security groups may have **unrestricted network access**, violating defense-in-depth principles.",
"Risk": "Compute instances without security groups are exposed to unrestricted network traffic, enabling:\n- **Confidentiality**: unauthorized access to services and data exfiltration\n- **Integrity**: injection attacks, tampering, and lateral movement across workloads\n- **Availability**: DDoS, port scanning, and resource exhaustion\nAttackers can probe open ports, exploit vulnerable services, and establish persistence without firewall barriers.",
"Risk": "Instances without security groups are exposed to unrestricted network traffic from any source. Attackers can probe open ports, exploit vulnerable services, conduct injection attacks, and tamper with data without firewall barriers. Lack of network access controls enables unauthorized access, data exfiltration, lateral movement, DDoS attacks, and resource exhaustion.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://docs.openstack.org/nova/latest/user/security-groups.html",
"https://docs.openstack.org/api-ref/compute/",
"https://docs.openstack.org/neutron/latest/admin/intro-os-networking.html",
"https://docs.openstack.org/security-guide/networking/architecture.html"
],
"Remediation": {
"Code": {
"CLI": "openstack server add security group <instance_id> <security_group_name>",
"NativeIaC": "",
"Other": "1. Access the Horizon dashboard\n2. Navigate to **Compute > Instances**\n3. Locate instances without security groups (check the Security Groups column)\n4. Select the instance, then **Actions > Edit Security Groups**\n5. Add at least one security group (e.g., `default` or a custom group with least-privilege rules)\n6. Click **Save** and verify the attachment\n7. Test connectivity to ensure proper ingress/egress rules",
"Other": "1. Navigate to **Compute > Instances**\n2. Select instance without security groups\n3. Click **Actions > Edit Security Groups**\n4. Add at least one security group\n5. Click **Save**",
"Terraform": "```hcl\n# Terraform: ensure compute instance has security groups attached\nresource \"openstack_compute_instance_v2\" \"instance\" {\n name = \"example-instance\"\n image_id = var.image_id\n flavor_id = var.flavor_id\n key_pair = var.key_pair_name\n # Critical: attach at least one security group\n security_groups = [\"default\", \"web-sg\", \"app-sg\"]\n\n network {\n name = var.network_name\n }\n}\n```"
},
"Recommendation": {
"Text": "Apply **defense-in-depth** for network security:\n- Attach **security groups** to all instances; never deploy without firewall rules\n- Follow **least privilege**: allow only required ports and sources; deny all by default\n- Use **separate security groups** per tier (web, app, database) with explicit rules\n- Avoid overly permissive rules like `0.0.0.0/0` ingress on sensitive ports\n- Implement **network segmentation** with isolated networks or VLANs\n- Enable **flow logs** for traffic analysis and anomaly detection\n- Regularly **audit** security group rules and remove stale or overly broad permissions",
"Text": "Attach security groups to all instances following least privilege principle: allow only required ports and sources, deny all by default. Use separate security groups per tier (web, app, database) with explicit rules. Avoid overly permissive rules like 0.0.0.0/0 ingress on sensitive ports. Implement network segmentation with isolated networks. Regularly audit and remove stale rules.",
"Url": "https://hub.prowler.com/check/compute_instance_security_groups_attached"
}
},
@@ -12,23 +12,13 @@ class compute_instance_security_groups_attached(Check):
for instance in compute_client.instances:
report = CheckReportOpenStack(metadata=self.metadata(), resource=instance)
report.resource_id = instance.id
report.resource_name = instance.name
report.region = instance.region
if instance.security_groups and len(instance.security_groups) > 0:
if instance.security_groups:
report.status = "PASS"
sg_names = ", ".join(instance.security_groups)
report.status_extended = (
f"Instance {instance.name} ({instance.id}) has security groups "
f"attached: {sg_names}."
)
report.status_extended = f"Instance {instance.name} ({instance.id}) has security groups attached: {sg_names}."
else:
report.status = "FAIL"
report.status_extended = (
f"Instance {instance.name} ({instance.id}) does not have any "
f"security groups attached."
)
report.status_extended = f"Instance {instance.name} ({instance.id}) does not have any security groups attached."
findings.append(report)
@@ -0,0 +1,39 @@
{
"Provider": "openstack",
"CheckID": "compute_instance_trusted_image_certificates",
"CheckTitle": "Compute instances use trusted image certificates",
"CheckType": [],
"ServiceName": "compute",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "high",
"ResourceType": "OpenStackInstance",
"ResourceGroup": "compute",
"Description": "**OpenStack compute instances** (VMs) are evaluated to verify that **trusted image certificates** are configured. Trusted image certificates enable cryptographic validation of image signatures using Glance image signing (OpenStack Image Signature Verification). This ensures instances are launched from verified, untampered images signed by trusted authorities.",
"Risk": "Instances without trusted certificates can be launched from tampered images containing backdoors, rootkits, or malware. Attackers can inject malicious code into unsigned images, and without signature verification, Nova launches compromised images. Malicious images enable persistence, lateral movement, data exfiltration, service disruption, and cryptomining.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://docs.openstack.org/glance/latest/user/signature.html",
"https://docs.openstack.org/nova/latest/admin/secure-boot.html",
"https://docs.openstack.org/api-ref/image/v2/index.html#image-signature-verification"
],
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "1. Enable Glance image signature verification in glance-api.conf\n2. Install Barbican for certificate storage\n3. Sign images using glance-manage or GPG tooling\n4. Upload certificates to Barbican\n5. Launch instances with trusted-image-certificate-id parameter\n\nNote: This requires administrator privileges and Glance/Nova configuration",
"Terraform": "```hcl\n# Terraform: launch instance with trusted image certificates\n\n# Note: Requires Glance image with signature metadata and Barbican certificate\n\nresource \"openstack_compute_instance_v2\" \"instance\" {\n name = \"secure-instance\"\n image_id = var.signed_image_id # Must be signed image\n flavor_id = var.flavor_id\n key_pair = var.key_pair_name\n security_groups = [\"default\"]\n\n # Specify trusted certificate IDs for signature validation\n # Note: This requires Nova microversion 2.63+ support\n trusted_image_certificates = [\n var.image_signing_cert_id,\n var.backup_cert_id\n ]\n\n network {\n name = var.network_name\n }\n}\n\n# Ensure image has signature metadata\ndata \"openstack_images_image_v2\" \"signed_image\" {\n name = \"ubuntu-22.04-signed\"\n most_recent = true\n\n # Verify signature properties exist\n properties = {\n img_signature = \"required\"\n img_signature_hash_method = \"SHA-256\"\n img_signature_key_type = \"RSA-PSS\"\n img_signature_certificate_uuid = \"required\"\n }\n}\n```"
},
"Recommendation": {
"Text": "Sign all images using GPG or X.509 certificates before uploading to Glance. Use Barbican to store signing certificates securely. Enforce trusted_image_certificates for production instances via policy. Enable signature verification in Nova (verify_glance_signatures=True). Restrict image upload permissions to authorized CI/CD pipelines. Implement certificate rotation every 12 months.",
"Url": "https://hub.prowler.com/check/compute_instance_trusted_image_certificates"
}
},
"Categories": [
"trust-boundaries",
"encryption"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": "Trusted image certificates require OpenStack Glance image signature verification (microversion 2.63+), Barbican for certificate storage, and Nova configured to verify signatures. Many OpenStack deployments do not enable this feature by default. This check may produce false positives for clouds not using image signing."
}
@@ -0,0 +1,25 @@
from typing import List
from prowler.lib.check.models import Check, CheckReportOpenStack
from prowler.providers.openstack.services.compute.compute_client import compute_client
class compute_instance_trusted_image_certificates(Check):
"""Ensure compute instances use trusted image certificates for image signature validation."""
def execute(self) -> List[CheckReportOpenStack]:
findings: List[CheckReportOpenStack] = []
for instance in compute_client.instances:
report = CheckReportOpenStack(metadata=self.metadata(), resource=instance)
if instance.trusted_image_certificates:
report.status = "PASS"
cert_ids = ", ".join(instance.trusted_image_certificates)
report.status_extended = f"Instance {instance.name} ({instance.id}) uses trusted image certificates: {cert_ids}."
else:
report.status = "FAIL"
report.status_extended = f"Instance {instance.name} ({instance.id}) does not use trusted image certificates (image signature validation not enforced)."
findings.append(report)
return findings
@@ -1,7 +1,8 @@
from __future__ import annotations
import ipaddress
from dataclasses import dataclass
from typing import List
from typing import Dict, List
from openstack import exceptions as openstack_exceptions
@@ -27,8 +28,72 @@ class Compute(OpenStackService):
sg_list = getattr(server, "security_groups", None) or []
security_groups = [sg.get("name", "") for sg in sg_list]
# Extract network information from addresses
networks_dict = {}
addresses_attr = getattr(server, "addresses", None)
if addresses_attr:
for net_name, addr_list in addresses_attr.items():
# addr_list is a list of dicts like:
# [{'version': 4, 'addr': '57.128.163.151', 'OS-EXT-IPS:type': 'fixed'}]
ip_list = []
if isinstance(addr_list, list):
for addr_dict in addr_list:
if isinstance(addr_dict, dict) and "addr" in addr_dict:
ip_list.append(addr_dict["addr"])
elif isinstance(addr_dict, str):
# Fallback: if it's just a string IP
ip_list.append(addr_dict)
elif isinstance(addr_list, str):
# Fallback: single string IP
ip_list = [addr_list]
networks_dict[net_name] = ip_list
# Extract trusted image certificates
trusted_certs = (
getattr(server, "trusted_image_certificates", None) or []
)
# Get SDK computed properties
public_v4 = getattr(server, "public_v4", "")
public_v6 = getattr(server, "public_v6", "")
private_v4 = getattr(server, "private_v4", "")
private_v6 = getattr(server, "private_v6", "")
# Fallback: If SDK attributes are not populated, classify IPs from networks
# This handles clouds where SDK computed properties are not available
if (
not (public_v4 or public_v6 or private_v4 or private_v6)
and networks_dict
):
for network_name, ip_list in networks_dict.items():
for ip_str in ip_list:
try:
ip_obj = ipaddress.ip_address(ip_str)
# Classify as private or public
if ip_obj.is_private:
# Assign first private IP found to appropriate field
if ip_obj.version == 4 and not private_v4:
private_v4 = ip_str
elif ip_obj.version == 6 and not private_v6:
private_v6 = ip_str
elif not (
ip_obj.is_loopback
or ip_obj.is_link_local
or ip_obj.is_reserved
or ip_obj.is_multicast
):
# Assign first public IP found to appropriate field
if ip_obj.version == 4 and not public_v4:
public_v4 = ip_str
elif ip_obj.version == 6 and not public_v6:
public_v6 = ip_str
except ValueError:
# Invalid IP address, skip
continue
self.instances.append(
ComputeInstance(
# Basic instance information
id=getattr(server, "id", ""),
name=getattr(server, "name", ""),
status=getattr(server, "status", ""),
@@ -36,6 +101,27 @@ class Compute(OpenStackService):
security_groups=security_groups,
region=self.region,
project_id=self.project_id,
# Access Control & Authentication
is_locked=getattr(server, "is_locked", False),
locked_reason=getattr(server, "locked_reason", ""),
key_name=getattr(server, "key_name", ""),
user_id=getattr(server, "user_id", ""),
# Network Exposure
access_ipv4=getattr(server, "access_ipv4", ""),
access_ipv6=getattr(server, "access_ipv6", ""),
public_v4=public_v4,
public_v6=public_v6,
private_v4=private_v4,
private_v6=private_v6,
networks=networks_dict,
# Configuration Security
has_config_drive=getattr(server, "has_config_drive", False),
metadata=getattr(server, "metadata", {}),
user_data=getattr(server, "user_data", ""),
# Image Trust
trusted_image_certificates=(
trusted_certs if isinstance(trusted_certs, list) else []
),
)
)
except openstack_exceptions.SDKException as error:
@@ -54,6 +140,7 @@ class Compute(OpenStackService):
class ComputeInstance:
"""Represents an OpenStack compute instance (VM)."""
# Basic instance information
id: str
name: str
status: str
@@ -61,3 +148,26 @@ class ComputeInstance:
security_groups: List[str]
region: str
project_id: str
# Access Control & Authentication
is_locked: bool
locked_reason: str
key_name: str
user_id: str
# Network Exposure
access_ipv4: str
access_ipv6: str
public_v4: str
public_v6: str
private_v4: str
private_v6: str
networks: Dict[str, List[str]]
# Configuration Security
has_config_drive: bool
metadata: Dict[str, str]
user_data: str
# Image Trust
trusted_image_certificates: List[str]
@@ -0,0 +1,10 @@
import ipaddress
def is_public_ip(ip_str: str) -> bool:
"""Check if an IP address is public (globally routable, non-multicast)."""
try:
ip = ipaddress.ip_address(ip_str)
return ip.is_global and not ip.is_multicast
except ValueError:
return False
@@ -0,0 +1,229 @@
"""Tests for compute_instance_config_drive_enabled check."""
from unittest import mock
from prowler.providers.openstack.services.compute.compute_service import ComputeInstance
from tests.providers.openstack.openstack_fixtures import (
OPENSTACK_PROJECT_ID,
OPENSTACK_REGION,
set_mocked_openstack_provider,
)
class Test_compute_instance_config_drive_enabled:
"""Test suite for compute_instance_config_drive_enabled check."""
def test_no_instances(self):
"""Test when no instances exist."""
compute_client = mock.MagicMock()
compute_client.instances = []
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_config_drive_enabled.compute_instance_config_drive_enabled.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_config_drive_enabled.compute_instance_config_drive_enabled import (
compute_instance_config_drive_enabled,
)
check = compute_instance_config_drive_enabled()
result = check.execute()
assert len(result) == 0
def test_instance_with_config_drive(self):
"""Test instance with config drive enabled (PASS)."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-1",
name="ConfigDrive Instance",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=True,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_config_drive_enabled.compute_instance_config_drive_enabled.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_config_drive_enabled.compute_instance_config_drive_enabled import (
compute_instance_config_drive_enabled,
)
check = compute_instance_config_drive_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "Instance ConfigDrive Instance (instance-1) has config drive enabled for secure metadata injection."
)
assert result[0].resource_id == "instance-1"
assert result[0].resource_name == "ConfigDrive Instance"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_instance_without_config_drive(self):
"""Test instance without config drive (FAIL)."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-2",
name="No ConfigDrive",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_config_drive_enabled.compute_instance_config_drive_enabled.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_config_drive_enabled.compute_instance_config_drive_enabled import (
compute_instance_config_drive_enabled,
)
check = compute_instance_config_drive_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "Instance No ConfigDrive (instance-2) does not have config drive enabled (relies on metadata service)."
)
assert result[0].resource_id == "instance-2"
assert result[0].resource_name == "No ConfigDrive"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_multiple_instances_mixed(self):
"""Test multiple instances with mixed config drive status."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-pass",
name="Pass",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=[],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=True,
metadata={},
user_data="",
trusted_image_certificates=[],
),
ComputeInstance(
id="instance-fail",
name="Fail",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=[],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
),
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_config_drive_enabled.compute_instance_config_drive_enabled.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_config_drive_enabled.compute_instance_config_drive_enabled import (
compute_instance_config_drive_enabled,
)
check = compute_instance_config_drive_enabled()
result = check.execute()
assert len(result) == 2
assert len([r for r in result if r.status == "PASS"]) == 1
assert len([r for r in result if r.status == "FAIL"]) == 1
@@ -0,0 +1,601 @@
"""Tests for compute_instance_isolated_private_network check."""
from unittest import mock
from prowler.providers.openstack.services.compute.compute_service import ComputeInstance
from tests.providers.openstack.openstack_fixtures import (
OPENSTACK_PROJECT_ID,
OPENSTACK_REGION,
set_mocked_openstack_provider,
)
class Test_compute_instance_isolated_private_network:
"""Test suite for compute_instance_isolated_private_network check."""
def test_no_instances(self):
"""Test when no instances exist."""
compute_client = mock.MagicMock()
compute_client.instances = []
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_isolated_private_network.compute_instance_isolated_private_network.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_isolated_private_network.compute_instance_isolated_private_network import (
compute_instance_isolated_private_network,
)
check = compute_instance_isolated_private_network()
result = check.execute()
assert len(result) == 0
def test_instance_private_only(self):
"""Test instance with private IP only (PASS)."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-1",
name="Isolated Instance",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="10.0.0.5",
private_v6="",
networks={"private": ["10.0.0.5"]},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_isolated_private_network.compute_instance_isolated_private_network.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_isolated_private_network.compute_instance_isolated_private_network import (
compute_instance_isolated_private_network,
)
check = compute_instance_isolated_private_network()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "Instance Isolated Instance (instance-1) is properly isolated in private network with private IPs (10.0.0.5) and no public exposure."
)
assert result[0].resource_id == "instance-1"
assert result[0].resource_name == "Isolated Instance"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_instance_mixed_public_private(self):
"""Test instance with both public and private IPs (FAIL)."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-2",
name="Mixed Instance",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="8.8.4.4",
public_v6="",
private_v4="10.0.0.10",
private_v6="",
networks={"public": ["8.8.4.4"], "private": ["10.0.0.10"]},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_isolated_private_network.compute_instance_isolated_private_network.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_isolated_private_network.compute_instance_isolated_private_network import (
compute_instance_isolated_private_network,
)
check = compute_instance_isolated_private_network()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "Instance Mixed Instance (instance-2) has mixed public and private network exposure (not properly isolated)."
)
assert result[0].resource_id == "instance-2"
assert result[0].resource_name == "Mixed Instance"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_instance_public_only(self):
"""Test instance with only public IP (FAIL)."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-3",
name="Public Only",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="1.1.1.1",
public_v6="",
private_v4="",
private_v6="",
networks={"public": ["1.1.1.1"]},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_isolated_private_network.compute_instance_isolated_private_network.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_isolated_private_network.compute_instance_isolated_private_network import (
compute_instance_isolated_private_network,
)
check = compute_instance_isolated_private_network()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "Instance Public Only (instance-3) has only public IP addresses (no private network isolation)."
)
assert result[0].resource_id == "instance-3"
assert result[0].resource_name == "Public Only"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_instance_no_ips(self):
"""Test instance with no IPs (FAIL)."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-4",
name="No IPs",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_isolated_private_network.compute_instance_isolated_private_network.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_isolated_private_network.compute_instance_isolated_private_network import (
compute_instance_isolated_private_network,
)
check = compute_instance_isolated_private_network()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "Instance No IPs (instance-4) has no network configuration (no IPs assigned)."
)
assert result[0].resource_id == "instance-4"
assert result[0].resource_name == "No IPs"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_instance_private_ipv6_only(self):
"""Test instance with private IPv6 only (PASS)."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-5",
name="IPv6 Private",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="fd00::1",
networks={"private": ["fd00::1"]},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_isolated_private_network.compute_instance_isolated_private_network.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_isolated_private_network.compute_instance_isolated_private_network import (
compute_instance_isolated_private_network,
)
check = compute_instance_isolated_private_network()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "Instance IPv6 Private (instance-5) is properly isolated in private network with private IPs (fd00::1) and no public exposure."
)
assert result[0].resource_id == "instance-5"
assert result[0].resource_name == "IPv6 Private"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_instance_fallback_private_only_networks_dict(self):
"""Test fallback logic: instance with private IP populated by service from networks dict."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-fallback-1",
name="Private Fallback",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="", # Empty
access_ipv6="", # Empty
public_v4="", # Empty
public_v6="", # Empty
private_v4="10.99.1.207", # Populated by service fallback
private_v6="", # Empty
networks={"test-private-net": ["10.99.1.207"]},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_isolated_private_network.compute_instance_isolated_private_network.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_isolated_private_network.compute_instance_isolated_private_network import (
compute_instance_isolated_private_network,
)
check = compute_instance_isolated_private_network()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert "10.99.1.207" in result[0].status_extended
assert "properly isolated in private network" in result[0].status_extended
assert result[0].resource_id == "instance-fallback-1"
def test_instance_fallback_public_only_networks_dict(self):
"""Test fallback logic: instance with public IP populated by service from networks dict."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-fallback-2",
name="Public Fallback",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="8.8.8.8", # Populated by service fallback
public_v6="", # Empty
private_v4="",
private_v6="",
networks={"ext-net": ["8.8.8.8"]},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_isolated_private_network.compute_instance_isolated_private_network.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_isolated_private_network.compute_instance_isolated_private_network import (
compute_instance_isolated_private_network,
)
check = compute_instance_isolated_private_network()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
"only public IP addresses" in result[0].status_extended
or "no private network isolation" in result[0].status_extended
)
assert result[0].resource_id == "instance-fallback-2"
def test_instance_fallback_mixed_networks_dict(self):
"""Test fallback logic: instance with mixed IPs populated by service from networks dict."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-fallback-3",
name="Mixed Fallback",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="8.8.8.8", # Populated by service fallback
public_v6="", # Empty
private_v4="10.0.0.100", # Populated by service fallback
private_v6="", # Empty
networks={
"private-net": ["10.0.0.100"],
"ext-net": ["8.8.8.8"],
},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_isolated_private_network.compute_instance_isolated_private_network.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_isolated_private_network.compute_instance_isolated_private_network import (
compute_instance_isolated_private_network,
)
check = compute_instance_isolated_private_network()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
"mixed public and private network exposure" in result[0].status_extended
)
assert result[0].resource_id == "instance-fallback-3"
def test_instance_access_ipv4_private_treated_as_private(self):
"""Test that access_ipv4 set to a private IP is not treated as public exposure."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-access-priv",
name="Access Private",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="10.0.0.50",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="10.0.0.50",
private_v6="",
networks={"private-net": ["10.0.0.50"]},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_isolated_private_network.compute_instance_isolated_private_network.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_isolated_private_network.compute_instance_isolated_private_network import (
compute_instance_isolated_private_network,
)
check = compute_instance_isolated_private_network()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert "properly isolated in private network" in result[0].status_extended
assert result[0].resource_id == "instance-access-priv"
def test_instance_network_ips_validated_as_public(self):
"""Test that IPs from networks dict are validated as truly public."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-net-pub",
name="Network Public",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={
"my-net": ["10.0.0.5", "8.8.8.8"],
},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_isolated_private_network.compute_instance_isolated_private_network.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_isolated_private_network.compute_instance_isolated_private_network import (
compute_instance_isolated_private_network,
)
check = compute_instance_isolated_private_network()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
"mixed public and private network exposure" in result[0].status_extended
)
assert result[0].resource_id == "instance-net-pub"
@@ -0,0 +1,229 @@
"""Tests for compute_instance_key_based_authentication check."""
from unittest import mock
from prowler.providers.openstack.services.compute.compute_service import ComputeInstance
from tests.providers.openstack.openstack_fixtures import (
OPENSTACK_PROJECT_ID,
OPENSTACK_REGION,
set_mocked_openstack_provider,
)
class Test_compute_instance_key_based_authentication:
"""Test suite for compute_instance_key_based_authentication check."""
def test_no_instances(self):
"""Test when no instances exist."""
compute_client = mock.MagicMock()
compute_client.instances = []
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_key_based_authentication.compute_instance_key_based_authentication.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_key_based_authentication.compute_instance_key_based_authentication import (
compute_instance_key_based_authentication,
)
check = compute_instance_key_based_authentication()
result = check.execute()
assert len(result) == 0
def test_instance_with_keypair(self):
"""Test instance with SSH keypair configured (PASS)."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-1",
name="Secure Instance",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="my-production-keypair",
user_id="user-123",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="10.0.0.5",
private_v6="",
networks={"private": ["10.0.0.5"]},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_key_based_authentication.compute_instance_key_based_authentication.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_key_based_authentication.compute_instance_key_based_authentication import (
compute_instance_key_based_authentication,
)
check = compute_instance_key_based_authentication()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "Instance Secure Instance (instance-1) is configured with SSH key-based authentication (keypair: my-production-keypair)."
)
assert result[0].resource_id == "instance-1"
assert result[0].resource_name == "Secure Instance"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_instance_without_keypair(self):
"""Test instance without SSH keypair (FAIL)."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-2",
name="Insecure Instance",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="user-456",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="10.0.0.10",
private_v6="",
networks={"private": ["10.0.0.10"]},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_key_based_authentication.compute_instance_key_based_authentication.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_key_based_authentication.compute_instance_key_based_authentication import (
compute_instance_key_based_authentication,
)
check = compute_instance_key_based_authentication()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "Instance Insecure Instance (instance-2) does not have SSH key-based authentication configured (no keypair assigned)."
)
assert result[0].resource_id == "instance-2"
assert result[0].resource_name == "Insecure Instance"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_multiple_instances_mixed(self):
"""Test multiple instances with mixed keypair configuration."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-secure",
name="With Key",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=[],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="prod-keypair",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
),
ComputeInstance(
id="instance-insecure",
name="Without Key",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=[],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
),
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_key_based_authentication.compute_instance_key_based_authentication.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_key_based_authentication.compute_instance_key_based_authentication import (
compute_instance_key_based_authentication,
)
check = compute_instance_key_based_authentication()
result = check.execute()
assert len(result) == 2
assert len([r for r in result if r.status == "PASS"]) == 1
assert len([r for r in result if r.status == "FAIL"]) == 1
@@ -0,0 +1,287 @@
"""Tests for compute_instance_locked_status_enabled check."""
from unittest import mock
from prowler.providers.openstack.services.compute.compute_service import ComputeInstance
from tests.providers.openstack.openstack_fixtures import (
OPENSTACK_PROJECT_ID,
OPENSTACK_REGION,
set_mocked_openstack_provider,
)
class Test_compute_instance_locked_status_enabled:
"""Test suite for compute_instance_locked_status_enabled check."""
def test_no_instances(self):
"""Test when no instances exist."""
compute_client = mock.MagicMock()
compute_client.instances = []
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_locked_status_enabled.compute_instance_locked_status_enabled.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_locked_status_enabled.compute_instance_locked_status_enabled import (
compute_instance_locked_status_enabled,
)
check = compute_instance_locked_status_enabled()
result = check.execute()
assert len(result) == 0
def test_instance_locked_with_reason(self):
"""Test instance with locked status enabled and reason (PASS)."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-1",
name="Locked Instance",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=True,
locked_reason="Production instance - do not modify",
key_name="my-keypair",
user_id="user-123",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="10.0.0.5",
private_v6="",
networks={"private": ["10.0.0.5"]},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_locked_status_enabled.compute_instance_locked_status_enabled.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_locked_status_enabled.compute_instance_locked_status_enabled import (
compute_instance_locked_status_enabled,
)
check = compute_instance_locked_status_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "Instance Locked Instance (instance-1) has locked status enabled (reason: Production instance - do not modify)."
)
assert result[0].resource_id == "instance-1"
assert result[0].resource_name == "Locked Instance"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_instance_locked_without_reason(self):
"""Test instance with locked status enabled but no reason (PASS)."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-2",
name="Locked No Reason",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=True,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_locked_status_enabled.compute_instance_locked_status_enabled.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_locked_status_enabled.compute_instance_locked_status_enabled import (
compute_instance_locked_status_enabled,
)
check = compute_instance_locked_status_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "Instance Locked No Reason (instance-2) has locked status enabled."
)
assert result[0].resource_id == "instance-2"
assert result[0].resource_name == "Locked No Reason"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_instance_not_locked(self):
"""Test instance without locked status (FAIL)."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-3",
name="Unlocked Instance",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_locked_status_enabled.compute_instance_locked_status_enabled.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_locked_status_enabled.compute_instance_locked_status_enabled import (
compute_instance_locked_status_enabled,
)
check = compute_instance_locked_status_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "Instance Unlocked Instance (instance-3) does not have locked status enabled."
)
assert result[0].resource_id == "instance-3"
assert result[0].resource_name == "Unlocked Instance"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_multiple_instances_mixed(self):
"""Test multiple instances with mixed locked status."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-locked",
name="Locked",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=[],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=True,
locked_reason="Protected",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
),
ComputeInstance(
id="instance-unlocked",
name="Unlocked",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=[],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
),
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_locked_status_enabled.compute_instance_locked_status_enabled.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_locked_status_enabled.compute_instance_locked_status_enabled import (
compute_instance_locked_status_enabled,
)
check = compute_instance_locked_status_enabled()
result = check.execute()
assert len(result) == 2
assert len([r for r in result if r.status == "PASS"]) == 1
assert len([r for r in result if r.status == "FAIL"]) == 1
@@ -0,0 +1,576 @@
"""Tests for compute_instance_metadata_sensitive_data check."""
from unittest import mock
from prowler.providers.openstack.services.compute.compute_service import ComputeInstance
from tests.providers.openstack.openstack_fixtures import (
OPENSTACK_PROJECT_ID,
OPENSTACK_REGION,
set_mocked_openstack_provider,
)
class Test_compute_instance_metadata_sensitive_data:
"""Test suite for compute_instance_metadata_sensitive_data check."""
def test_no_instances(self):
"""Test when no instances exist."""
compute_client = mock.MagicMock()
compute_client.instances = []
compute_client.audit_config = {}
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_metadata_sensitive_data.compute_instance_metadata_sensitive_data.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_metadata_sensitive_data.compute_instance_metadata_sensitive_data import (
compute_instance_metadata_sensitive_data,
)
check = compute_instance_metadata_sensitive_data()
result = check.execute()
assert len(result) == 0
def test_instance_no_metadata(self):
"""Test instance with no metadata (PASS)."""
compute_client = mock.MagicMock()
compute_client.audit_config = {}
compute_client.instances = [
ComputeInstance(
id="instance-1",
name="No Metadata",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_metadata_sensitive_data.compute_instance_metadata_sensitive_data.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_metadata_sensitive_data.compute_instance_metadata_sensitive_data import (
compute_instance_metadata_sensitive_data,
)
check = compute_instance_metadata_sensitive_data()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "Instance No Metadata (instance-1) has no metadata (no sensitive data exposure risk)."
)
assert result[0].resource_id == "instance-1"
assert result[0].resource_name == "No Metadata"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_instance_safe_metadata(self):
"""Test instance with safe metadata (PASS)."""
compute_client = mock.MagicMock()
compute_client.audit_config = {}
compute_client.instances = [
ComputeInstance(
id="instance-2",
name="Safe Metadata",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={"environment": "production", "application": "web-app"},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_metadata_sensitive_data.compute_instance_metadata_sensitive_data.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_metadata_sensitive_data.compute_instance_metadata_sensitive_data import (
compute_instance_metadata_sensitive_data,
)
check = compute_instance_metadata_sensitive_data()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "Instance Safe Metadata (instance-2) metadata does not contain sensitive data."
)
assert result[0].resource_id == "instance-2"
assert result[0].resource_name == "Safe Metadata"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_instance_password_in_metadata(self):
"""Test instance with password in metadata (FAIL)."""
compute_client = mock.MagicMock()
compute_client.audit_config = {}
compute_client.instances = [
ComputeInstance(
id="instance-3",
name="Password Metadata",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={"db_password": "supersecret123"},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_metadata_sensitive_data.compute_instance_metadata_sensitive_data.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_metadata_sensitive_data.compute_instance_metadata_sensitive_data import (
compute_instance_metadata_sensitive_data,
)
check = compute_instance_metadata_sensitive_data()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert "contains potential secrets" in result[0].status_extended
def test_instance_api_key_in_metadata(self):
"""Test instance with API key in metadata (FAIL)."""
compute_client = mock.MagicMock()
compute_client.audit_config = {}
compute_client.instances = [
ComputeInstance(
id="instance-4",
name="API Key Metadata",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={"api_key": "sk-1234567890"},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_metadata_sensitive_data.compute_instance_metadata_sensitive_data.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_metadata_sensitive_data.compute_instance_metadata_sensitive_data import (
compute_instance_metadata_sensitive_data,
)
check = compute_instance_metadata_sensitive_data()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].status_extended.startswith(
"Instance API Key Metadata (instance-4) metadata contains potential secrets ->"
)
assert result[0].resource_id == "instance-4"
assert result[0].resource_name == "API Key Metadata"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_instance_connection_string_in_metadata(self):
"""Test instance with database connection string in metadata (FAIL)."""
compute_client = mock.MagicMock()
compute_client.audit_config = {}
compute_client.instances = [
ComputeInstance(
id="instance-5",
name="Connection String",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={"db_url": "mysql://admin:s3cret@dbhost:3306/appdb"},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_metadata_sensitive_data.compute_instance_metadata_sensitive_data.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_metadata_sensitive_data.compute_instance_metadata_sensitive_data import (
compute_instance_metadata_sensitive_data,
)
check = compute_instance_metadata_sensitive_data()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].status_extended.startswith(
"Instance Connection String (instance-5) metadata contains potential secrets ->"
)
assert result[0].resource_id == "instance-5"
assert result[0].resource_name == "Connection String"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_instance_private_key_in_metadata(self):
"""Test instance with private key in metadata (FAIL)."""
compute_client = mock.MagicMock()
compute_client.audit_config = {}
compute_client.instances = [
ComputeInstance(
id="instance-6",
name="Private Key",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={"ssh_key": "-----BEGIN RSA PRIVATE KEY-----"},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_metadata_sensitive_data.compute_instance_metadata_sensitive_data.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_metadata_sensitive_data.compute_instance_metadata_sensitive_data import (
compute_instance_metadata_sensitive_data,
)
check = compute_instance_metadata_sensitive_data()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].status_extended.startswith(
"Instance Private Key (instance-6) metadata contains potential secrets ->"
)
assert result[0].resource_id == "instance-6"
assert result[0].resource_name == "Private Key"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_multiple_instances_mixed(self):
"""Test multiple instances with mixed metadata."""
compute_client = mock.MagicMock()
compute_client.audit_config = {}
compute_client.instances = [
ComputeInstance(
id="instance-pass",
name="Safe",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=[],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={"tier": "web"},
user_data="",
trusted_image_certificates=[],
),
ComputeInstance(
id="instance-fail",
name="Unsafe",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=[],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={"admin_password": "secret123"},
user_data="",
trusted_image_certificates=[],
),
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_metadata_sensitive_data.compute_instance_metadata_sensitive_data.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_metadata_sensitive_data.compute_instance_metadata_sensitive_data import (
compute_instance_metadata_sensitive_data,
)
check = compute_instance_metadata_sensitive_data()
result = check.execute()
assert len(result) == 2
assert len([r for r in result if r.status == "PASS"]) == 1
assert len([r for r in result if r.status == "FAIL"]) == 1
def test_instance_multiple_metadata_keys_correct_identification(self):
"""Test that secrets are correctly attributed to the right metadata keys."""
compute_client = mock.MagicMock()
compute_client.audit_config = {}
compute_client.instances = [
ComputeInstance(
id="instance-7",
name="Multiple Keys",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={
"environment": "production",
"application": "web-app",
"db_password": "supersecret123",
"region": "us-east",
},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_metadata_sensitive_data.compute_instance_metadata_sensitive_data.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_metadata_sensitive_data.compute_instance_metadata_sensitive_data import (
compute_instance_metadata_sensitive_data,
)
check = compute_instance_metadata_sensitive_data()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
# Verify the secret is correctly attributed to 'db_password' key
assert "in metadata key 'db_password'" in result[0].status_extended
assert result[0].resource_id == "instance-7"
def test_instance_metadata_key_ordering(self):
"""Test that secret detection works with different key orderings."""
compute_client = mock.MagicMock()
compute_client.audit_config = {}
compute_client.instances = [
ComputeInstance(
id="instance-8",
name="Ordered Keys",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={
"first_key": "safe_value",
"api_key": "sk-1234567890abcdef",
"third_key": "also_safe",
},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_metadata_sensitive_data.compute_instance_metadata_sensitive_data.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_metadata_sensitive_data.compute_instance_metadata_sensitive_data import (
compute_instance_metadata_sensitive_data,
)
check = compute_instance_metadata_sensitive_data()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
# Verify the secret is correctly attributed to 'api_key' key (second in order)
assert "in metadata key 'api_key'" in result[0].status_extended
assert result[0].resource_id == "instance-8"
@@ -0,0 +1,708 @@
"""Tests for compute_instance_public_ip_exposed check."""
from unittest import mock
from prowler.providers.openstack.services.compute.compute_service import ComputeInstance
from tests.providers.openstack.openstack_fixtures import (
OPENSTACK_PROJECT_ID,
OPENSTACK_REGION,
set_mocked_openstack_provider,
)
class Test_compute_instance_public_ip_exposed:
"""Test suite for compute_instance_public_ip_exposed check."""
def test_no_instances(self):
"""Test when no instances exist."""
compute_client = mock.MagicMock()
compute_client.instances = []
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_public_ip_exposed.compute_instance_public_ip_exposed.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_public_ip_exposed.compute_instance_public_ip_exposed import (
compute_instance_public_ip_exposed,
)
check = compute_instance_public_ip_exposed()
result = check.execute()
assert len(result) == 0
def test_instance_without_public_ip(self):
"""Test instance without public IP (PASS)."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-1",
name="Private Instance",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="10.0.0.5",
private_v6="",
networks={"private": ["10.0.0.5"]}, # Processed from addresses
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_public_ip_exposed.compute_instance_public_ip_exposed.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_public_ip_exposed.compute_instance_public_ip_exposed import (
compute_instance_public_ip_exposed,
)
check = compute_instance_public_ip_exposed()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "Instance Private Instance (instance-1) is not exposed to the internet (no public IP addresses or external network attachments detected)."
)
assert result[0].resource_id == "instance-1"
assert result[0].resource_name == "Private Instance"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_instance_with_public_ipv4(self):
"""Test instance with public IPv4 (FAIL)."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-2",
name="Public Instance",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="203.0.113.10",
public_v6="",
private_v4="10.0.0.10",
private_v6="",
networks={"public": ["203.0.113.10"], "private": ["10.0.0.10"]},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_public_ip_exposed.compute_instance_public_ip_exposed.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_public_ip_exposed.compute_instance_public_ip_exposed import (
compute_instance_public_ip_exposed,
)
check = compute_instance_public_ip_exposed()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].status_extended.startswith(
"Instance Public Instance (instance-2) is exposed to the internet with public IP addresses:"
)
assert "203.0.113.10" in result[0].status_extended
assert result[0].resource_id == "instance-2"
assert result[0].resource_name == "Public Instance"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_instance_with_access_ipv4(self):
"""Test instance with access IPv4 (FAIL)."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-3",
name="Access IP Instance",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="198.51.100.5",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="10.0.0.15",
private_v6="",
networks={"private": ["10.0.0.15"]},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_public_ip_exposed.compute_instance_public_ip_exposed.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_public_ip_exposed.compute_instance_public_ip_exposed import (
compute_instance_public_ip_exposed,
)
check = compute_instance_public_ip_exposed()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].status_extended.startswith(
"Instance Access IP Instance (instance-3) is exposed to the internet with public IP addresses:"
)
assert "198.51.100.5" in result[0].status_extended
assert result[0].resource_id == "instance-3"
assert result[0].resource_name == "Access IP Instance"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_instance_with_ipv6(self):
"""Test instance with public IPv6 (FAIL)."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-4",
name="IPv6 Instance",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="2001:db8::1",
public_v4="",
public_v6="",
private_v4="",
private_v6="fd00::1",
networks={"private": ["fd00::1"]},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_public_ip_exposed.compute_instance_public_ip_exposed.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_public_ip_exposed.compute_instance_public_ip_exposed import (
compute_instance_public_ip_exposed,
)
check = compute_instance_public_ip_exposed()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].status_extended.startswith(
"Instance IPv6 Instance (instance-4) is exposed to the internet with public IP addresses:"
)
assert "2001:db8::1" in result[0].status_extended
assert result[0].resource_id == "instance-4"
assert result[0].resource_name == "IPv6 Instance"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_multiple_instances_mixed(self):
"""Test multiple instances with mixed public IP configuration."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-pass",
name="Private",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=[],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="10.0.0.20",
private_v6="",
networks={"private": ["10.0.0.20"]},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
),
ComputeInstance(
id="instance-fail",
name="Public",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=[],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="203.0.113.20",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
),
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_public_ip_exposed.compute_instance_public_ip_exposed.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_public_ip_exposed.compute_instance_public_ip_exposed import (
compute_instance_public_ip_exposed,
)
check = compute_instance_public_ip_exposed()
result = check.execute()
assert len(result) == 2
assert len([r for r in result if r.status == "PASS"]) == 1
assert len([r for r in result if r.status == "FAIL"]) == 1
def test_instance_on_external_network(self):
"""Test instance directly attached to external network (OVH-style)."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-extnet",
name="ExtNet Instance",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="", # SDK might not populate this
public_v6="",
private_v4="",
private_v6="",
networks={
"Ext-Net": ["57.128.163.151", "2001:41d0:801:1000::164b"]
}, # OVH external network
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_public_ip_exposed.compute_instance_public_ip_exposed.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_public_ip_exposed.compute_instance_public_ip_exposed import (
compute_instance_public_ip_exposed,
)
check = compute_instance_public_ip_exposed()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert "57.128.163.151" in result[0].status_extended
assert "Ext-Net" in result[0].status_extended
assert result[0].resource_id
assert result[0].resource_name
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_instance_mixed_networks_private_and_external(self):
"""Test instance with both private and external network attachments."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-mixed",
name="Mixed Networks",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="10.0.0.5",
private_v6="",
networks={
"private-net": ["10.0.0.5"],
"public-network": ["8.8.8.8"], # Real public IP (Google DNS)
},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_public_ip_exposed.compute_instance_public_ip_exposed.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_public_ip_exposed.compute_instance_public_ip_exposed import (
compute_instance_public_ip_exposed,
)
check = compute_instance_public_ip_exposed()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert "8.8.8.8" in result[0].status_extended
assert "public-network" in result[0].status_extended
assert result[0].resource_id
assert result[0].resource_name
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_instance_false_positive_network_names(self):
"""Test that network names containing 'ext' as substring don't cause false positives."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-context",
name="Context Network Instance",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="10.0.0.100",
private_v6="",
networks={
"context-internal": [
"10.0.0.100"
], # Contains "ext" but should not match
"next-hop": ["10.0.0.101"], # Contains "ext" but should not match
"text-processing": [
"10.0.0.102"
], # Contains "ext" but should not match
},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_public_ip_exposed.compute_instance_public_ip_exposed.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_public_ip_exposed.compute_instance_public_ip_exposed import (
compute_instance_public_ip_exposed,
)
check = compute_instance_public_ip_exposed()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "Instance Context Network Instance (instance-context) is not exposed to the internet (no public IP addresses or external network attachments detected)."
)
assert result[0].resource_id == "instance-context"
assert result[0].resource_name == "Context Network Instance"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_instance_word_boundary_ext_network(self):
"""Test that 'ext' as a complete word is properly detected."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-ext-word",
name="Ext Word Boundary Instance",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={
"ext": ["8.8.8.8"], # Word boundary: "ext" alone
"ext-network": ["1.1.1.1"], # Word boundary: "ext" at start
"network-ext": ["9.9.9.9"], # Word boundary: "ext" at end
},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_public_ip_exposed.compute_instance_public_ip_exposed.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_public_ip_exposed.compute_instance_public_ip_exposed import (
compute_instance_public_ip_exposed,
)
check = compute_instance_public_ip_exposed()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
# Should detect at least one external network with public IP
assert "ext" in result[0].status_extended.lower()
assert result[0].resource_id == "instance-ext-word"
assert result[0].resource_name == "Ext Word Boundary Instance"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_instance_public_ip_generic_network_name(self):
"""Test that public IPs are detected regardless of network name (e.g., 'hello')."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-hello",
name="Generic Network Instance",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="8.8.8.8", # Service populates this via fallback
public_v6="",
private_v4="",
private_v6="",
networks={"hello": ["8.8.8.8"]}, # Generic name, but public IP
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_public_ip_exposed.compute_instance_public_ip_exposed.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_public_ip_exposed.compute_instance_public_ip_exposed import (
compute_instance_public_ip_exposed,
)
check = compute_instance_public_ip_exposed()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert "8.8.8.8" in result[0].status_extended
assert result[0].resource_id == "instance-hello"
assert result[0].resource_name == "Generic Network Instance"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_instance_multiple_public_ips_on_different_networks(self):
"""Test that multiple public IPs on different networks are all detected."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-multi-ip",
name="Multiple Public IPs",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="8.8.8.8", # First public IP captured by service
public_v6="",
private_v4="",
private_v6="",
networks={
"network1": ["8.8.8.8"], # First public IP
"network2": ["1.1.1.1"], # Second public IP
"network3": ["9.9.9.9"], # Third public IP
},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_public_ip_exposed.compute_instance_public_ip_exposed.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_public_ip_exposed.compute_instance_public_ip_exposed import (
compute_instance_public_ip_exposed,
)
check = compute_instance_public_ip_exposed()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
# Should detect all three public IPs
assert "8.8.8.8" in result[0].status_extended
assert "1.1.1.1" in result[0].status_extended
assert "9.9.9.9" in result[0].status_extended
# Should show network names for additional IPs
assert "network2" in result[0].status_extended
assert "network3" in result[0].status_extended
assert result[0].resource_id == "instance-multi-ip"
assert result[0].resource_name == "Multiple Public IPs"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
@@ -49,6 +49,21 @@ class Test_compute_instance_security_groups_attached:
security_groups=["default", "web"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
@@ -71,11 +86,14 @@ class Test_compute_instance_security_groups_attached:
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "Instance Instance One (instance-1) has security groups attached: default, web."
)
assert result[0].resource_id == "instance-1"
assert result[0].resource_name == "Instance One"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
assert "has security groups attached" in result[0].status_extended
def test_instance_without_security_groups(self):
"""Test instance without security groups attached (FAIL)."""
@@ -89,6 +107,21 @@ class Test_compute_instance_security_groups_attached:
security_groups=[],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
@@ -111,14 +144,14 @@ class Test_compute_instance_security_groups_attached:
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "Instance Instance Two (instance-2) does not have any security groups attached."
)
assert result[0].resource_id == "instance-2"
assert result[0].resource_name == "Instance Two"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
assert (
"does not have any security groups attached"
in result[0].status_extended
)
def test_multiple_instances_mixed(self):
"""Test multiple instances with mixed results."""
@@ -132,6 +165,21 @@ class Test_compute_instance_security_groups_attached:
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
),
ComputeInstance(
id="instance-fail",
@@ -141,6 +189,21 @@ class Test_compute_instance_security_groups_attached:
security_groups=[],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
),
]
@@ -177,6 +240,21 @@ class Test_compute_instance_security_groups_attached:
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
@@ -198,6 +276,12 @@ class Test_compute_instance_security_groups_attached:
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "Instance (instance-3) has security groups attached: default."
)
assert result[0].resource_id == "instance-3"
assert result[0].resource_name == ""
assert "instance-3" in result[0].status_extended
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
@@ -0,0 +1,229 @@
"""Tests for compute_instance_trusted_image_certificates check."""
from unittest import mock
from prowler.providers.openstack.services.compute.compute_service import ComputeInstance
from tests.providers.openstack.openstack_fixtures import (
OPENSTACK_PROJECT_ID,
OPENSTACK_REGION,
set_mocked_openstack_provider,
)
class Test_compute_instance_trusted_image_certificates:
"""Test suite for compute_instance_trusted_image_certificates check."""
def test_no_instances(self):
"""Test when no instances exist."""
compute_client = mock.MagicMock()
compute_client.instances = []
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_trusted_image_certificates.compute_instance_trusted_image_certificates.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_trusted_image_certificates.compute_instance_trusted_image_certificates import (
compute_instance_trusted_image_certificates,
)
check = compute_instance_trusted_image_certificates()
result = check.execute()
assert len(result) == 0
def test_instance_with_trusted_certificates(self):
"""Test instance with trusted image certificates (PASS)."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-1",
name="Trusted Instance",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=["cert-123", "cert-456"],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_trusted_image_certificates.compute_instance_trusted_image_certificates.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_trusted_image_certificates.compute_instance_trusted_image_certificates import (
compute_instance_trusted_image_certificates,
)
check = compute_instance_trusted_image_certificates()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].status_extended.startswith(
"Instance Trusted Instance (instance-1) uses trusted image certificates:"
)
assert "cert-123" in result[0].status_extended
assert result[0].resource_id == "instance-1"
assert result[0].resource_name == "Trusted Instance"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_instance_without_trusted_certificates(self):
"""Test instance without trusted image certificates (FAIL)."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-2",
name="Untrusted Instance",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=["default"],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
)
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_trusted_image_certificates.compute_instance_trusted_image_certificates.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_trusted_image_certificates.compute_instance_trusted_image_certificates import (
compute_instance_trusted_image_certificates,
)
check = compute_instance_trusted_image_certificates()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "Instance Untrusted Instance (instance-2) does not use trusted image certificates (image signature validation not enforced)."
)
assert result[0].resource_id == "instance-2"
assert result[0].resource_name == "Untrusted Instance"
assert result[0].region == OPENSTACK_REGION
assert result[0].project_id == OPENSTACK_PROJECT_ID
def test_multiple_instances_mixed(self):
"""Test multiple instances with mixed certificate configuration."""
compute_client = mock.MagicMock()
compute_client.instances = [
ComputeInstance(
id="instance-pass",
name="Pass",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=[],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=["cert-789"],
),
ComputeInstance(
id="instance-fail",
name="Fail",
status="ACTIVE",
flavor_id="flavor-1",
security_groups=[],
region=OPENSTACK_REGION,
project_id=OPENSTACK_PROJECT_ID,
is_locked=False,
locked_reason="",
key_name="",
user_id="",
access_ipv4="",
access_ipv6="",
public_v4="",
public_v6="",
private_v4="",
private_v6="",
networks={},
has_config_drive=False,
metadata={},
user_data="",
trusted_image_certificates=[],
),
]
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_openstack_provider(),
),
mock.patch(
"prowler.providers.openstack.services.compute.compute_instance_trusted_image_certificates.compute_instance_trusted_image_certificates.compute_client",
new=compute_client,
),
):
from prowler.providers.openstack.services.compute.compute_instance_trusted_image_certificates.compute_instance_trusted_image_certificates import (
compute_instance_trusted_image_certificates,
)
check = compute_instance_trusted_image_certificates()
result = check.execute()
assert len(result) == 2
assert len([r for r in result if r.status == "PASS"]) == 1
assert len([r for r in result if r.status == "FAIL"]) == 1
@@ -0,0 +1,53 @@
"""Tests for the shared is_public_ip utility."""
from prowler.providers.openstack.services.compute.lib.ip import is_public_ip
class Test_is_public_ip:
def test_public_ipv4(self):
assert is_public_ip("8.8.8.8")
def test_public_ipv4_other(self):
assert is_public_ip("1.1.1.1")
def test_private_ipv4_10(self):
assert not is_public_ip("10.0.0.5")
def test_private_ipv4_172(self):
assert not is_public_ip("172.16.0.1")
def test_private_ipv4_192(self):
assert not is_public_ip("192.168.1.1")
def test_loopback_ipv4(self):
assert not is_public_ip("127.0.0.1")
def test_link_local_ipv4(self):
assert not is_public_ip("169.254.0.1")
def test_multicast_ipv4(self):
assert not is_public_ip("224.0.0.1")
def test_documentation_ipv4_not_global(self):
assert not is_public_ip("203.0.113.10")
def test_public_ipv6(self):
assert is_public_ip("2001:41d0:801:1000::164b")
def test_private_ipv6(self):
assert not is_public_ip("fd00::1")
def test_loopback_ipv6(self):
assert not is_public_ip("::1")
def test_link_local_ipv6(self):
assert not is_public_ip("fe80::1")
def test_documentation_ipv6_not_global(self):
assert not is_public_ip("2001:db8::1")
def test_invalid_ip(self):
assert not is_public_ip("not-an-ip")
def test_empty_string(self):
assert not is_public_ip("")
@@ -44,6 +44,24 @@ class TestComputeService:
mock_server1.status = "ACTIVE"
mock_server1.flavor = {"id": "flavor-1"}
mock_server1.security_groups = [{"name": "default"}]
mock_server1.is_locked = True
mock_server1.locked_reason = "maintenance"
mock_server1.key_name = "my-keypair"
mock_server1.user_id = "user-123"
mock_server1.access_ipv4 = "203.0.113.10"
mock_server1.access_ipv6 = "2001:db8::1"
mock_server1.public_v4 = "203.0.113.10"
mock_server1.public_v6 = ""
mock_server1.private_v4 = "10.0.0.5"
mock_server1.private_v6 = ""
mock_server1.addresses = {
"private": [{"version": 4, "addr": "10.0.0.5"}],
"public": [{"version": 4, "addr": "203.0.113.10"}],
}
mock_server1.has_config_drive = True
mock_server1.metadata = {"environment": "production"}
mock_server1.user_data = "#!/bin/bash\necho hello"
mock_server1.trusted_image_certificates = ["cert-123"]
mock_server2 = MagicMock()
mock_server2.id = "instance-2"
@@ -51,6 +69,21 @@ class TestComputeService:
mock_server2.status = "SHUTOFF"
mock_server2.flavor = {"id": "flavor-2"}
mock_server2.security_groups = [{"name": "web"}, {"name": "db"}]
mock_server2.is_locked = False
mock_server2.locked_reason = ""
mock_server2.key_name = ""
mock_server2.user_id = "user-456"
mock_server2.access_ipv4 = ""
mock_server2.access_ipv6 = ""
mock_server2.public_v4 = ""
mock_server2.public_v6 = ""
mock_server2.private_v4 = "10.0.0.10"
mock_server2.private_v6 = ""
mock_server2.addresses = {"private": [{"version": 4, "addr": "10.0.0.10"}]}
mock_server2.has_config_drive = False
mock_server2.metadata = {}
mock_server2.user_data = ""
mock_server2.trusted_image_certificates = []
provider.connection.compute.servers.return_value = [
mock_server1,
@@ -68,8 +101,27 @@ class TestComputeService:
assert compute.instances[0].security_groups == ["default"]
assert compute.instances[0].region == OPENSTACK_REGION
assert compute.instances[0].project_id == OPENSTACK_PROJECT_ID
assert compute.instances[0].is_locked is True
assert compute.instances[0].locked_reason == "maintenance"
assert compute.instances[0].key_name == "my-keypair"
assert compute.instances[0].user_id == "user-123"
assert compute.instances[0].access_ipv4 == "203.0.113.10"
assert compute.instances[0].access_ipv6 == "2001:db8::1"
assert compute.instances[0].public_v4 == "203.0.113.10"
assert compute.instances[0].private_v4 == "10.0.0.5"
assert compute.instances[0].networks == {
"private": ["10.0.0.5"],
"public": ["203.0.113.10"],
}
assert compute.instances[0].has_config_drive is True
assert compute.instances[0].metadata == {"environment": "production"}
assert compute.instances[0].user_data == "#!/bin/bash\necho hello"
assert compute.instances[0].trusted_image_certificates == ["cert-123"]
assert compute.instances[1].security_groups == ["web", "db"]
assert compute.instances[1].is_locked is False
assert compute.instances[1].key_name == ""
assert compute.instances[1].trusted_image_certificates == []
def test_compute_list_instances_empty(self):
"""Test listing instances when none exist."""
@@ -90,6 +142,21 @@ class TestComputeService:
del mock_server.status
del mock_server.flavor
del mock_server.security_groups
del mock_server.is_locked
del mock_server.locked_reason
del mock_server.key_name
del mock_server.user_id
del mock_server.access_ipv4
del mock_server.access_ipv6
del mock_server.public_v4
del mock_server.public_v6
del mock_server.private_v4
del mock_server.private_v6
del mock_server.addresses
del mock_server.has_config_drive
del mock_server.metadata
del mock_server.user_data
del mock_server.trusted_image_certificates
provider.connection.compute.servers.return_value = [mock_server]
@@ -101,6 +168,21 @@ class TestComputeService:
assert compute.instances[0].status == ""
assert compute.instances[0].flavor_id == ""
assert compute.instances[0].security_groups == []
assert compute.instances[0].is_locked is False
assert compute.instances[0].locked_reason == ""
assert compute.instances[0].key_name == ""
assert compute.instances[0].user_id == ""
assert compute.instances[0].access_ipv4 == ""
assert compute.instances[0].access_ipv6 == ""
assert compute.instances[0].public_v4 == ""
assert compute.instances[0].public_v6 == ""
assert compute.instances[0].private_v4 == ""
assert compute.instances[0].private_v6 == ""
assert compute.instances[0].networks == {}
assert compute.instances[0].has_config_drive is False
assert compute.instances[0].metadata == {}
assert compute.instances[0].user_data == ""
assert compute.instances[0].trusted_image_certificates == []
def test_compute_list_instances_sdk_exception(self):
"""Test handling SDKException when listing instances."""
@@ -133,6 +215,21 @@ class TestComputeService:
mock_server.status = "ACTIVE"
mock_server.flavor = {"id": "flavor-1"}
mock_server.security_groups = [{"name": "default"}]
mock_server.is_locked = False
mock_server.locked_reason = ""
mock_server.key_name = ""
mock_server.user_id = ""
mock_server.access_ipv4 = ""
mock_server.access_ipv6 = ""
mock_server.public_v4 = ""
mock_server.public_v6 = ""
mock_server.private_v4 = ""
mock_server.private_v6 = ""
mock_server.addresses = {}
mock_server.has_config_drive = False
mock_server.metadata = {}
mock_server.user_data = ""
mock_server.trusted_image_certificates = []
yield mock_server
raise Exception("Iterator failed")
@@ -154,6 +251,23 @@ class TestComputeService:
security_groups=["default"],
region="RegionOne",
project_id="project-1",
is_locked=True,
locked_reason="maintenance",
key_name="my-keypair",
user_id="user-123",
access_ipv4="203.0.113.10",
access_ipv6="2001:db8::1",
public_v4="203.0.113.10",
public_v6="",
private_v4="10.0.0.5",
private_v6="",
networks={
"private": ["10.0.0.5"]
}, # Note: This is the processed dict, not addresses
has_config_drive=True,
metadata={"environment": "production"},
user_data="#!/bin/bash\necho hello",
trusted_image_certificates=["cert-123"],
)
assert instance.id == "instance-1"
@@ -163,6 +277,21 @@ class TestComputeService:
assert instance.security_groups == ["default"]
assert instance.region == "RegionOne"
assert instance.project_id == "project-1"
assert instance.is_locked is True
assert instance.locked_reason == "maintenance"
assert instance.key_name == "my-keypair"
assert instance.user_id == "user-123"
assert instance.access_ipv4 == "203.0.113.10"
assert instance.access_ipv6 == "2001:db8::1"
assert instance.public_v4 == "203.0.113.10"
assert instance.public_v6 == ""
assert instance.private_v4 == "10.0.0.5"
assert instance.private_v6 == ""
assert instance.networks == {"private": ["10.0.0.5"]}
assert instance.has_config_drive is True
assert instance.metadata == {"environment": "production"}
assert instance.user_data == "#!/bin/bash\necho hello"
assert instance.trusted_image_certificates == ["cert-123"]
def test_compute_service_inherits_from_base(self):
"""Test Compute service inherits from OpenStackService."""
@@ -180,3 +309,37 @@ class TestComputeService:
assert hasattr(compute, "identity")
assert hasattr(compute, "audit_config")
assert hasattr(compute, "fixer_config")
def test_compute_list_instances_with_none_addresses(self):
"""Test listing instances when addresses attribute is None."""
provider = set_mocked_openstack_provider()
mock_server = MagicMock()
mock_server.id = "instance-1"
mock_server.name = "Instance With None Addresses"
mock_server.status = "ACTIVE"
mock_server.flavor = {"id": "flavor-1"}
mock_server.security_groups = [{"name": "default"}]
mock_server.is_locked = False
mock_server.locked_reason = ""
mock_server.key_name = "test-key"
mock_server.user_id = "user-123"
mock_server.access_ipv4 = ""
mock_server.access_ipv6 = ""
mock_server.public_v4 = ""
mock_server.public_v6 = ""
mock_server.private_v4 = ""
mock_server.private_v6 = ""
mock_server.addresses = None # This is the key test case
mock_server.has_config_drive = False
mock_server.metadata = {}
mock_server.user_data = ""
mock_server.trusted_image_certificates = []
provider.connection.compute.servers.return_value = [mock_server]
compute = Compute(provider)
assert len(compute.instances) == 1
assert compute.instances[0].id == "instance-1"
assert compute.instances[0].networks == {} # Should default to empty dict