diff --git a/README.md b/README.md index 816e928..d0ed975 100644 --- a/README.md +++ b/README.md @@ -90,6 +90,7 @@ The CLI supports multiple input modes for both AWS and Azure. | Interactive (manual credentials) | `python main.py aws` | | AWS CLI profile | `python main.py aws --profile PROFILE` | | JSON config file | `python main.py aws --config config.json` | +| Terraform/OpenTofu state file | `python main.py aws --tfstate infra.tfstate` | | Non-interactive (env vars, CI) | `python main.py aws --non-interactive` | **Azure** @@ -99,6 +100,7 @@ The CLI supports multiple input modes for both AWS and Azure. | Interactive (service principal) | `python main.py azure` | | Azure CLI session | `python main.py azure --cli` | | JSON config file | `python main.py azure --config config.json` | +| Terraform/OpenTofu state file | `python main.py azure --tfstate infra.tfstate` | | Non-interactive (env vars, CI) | `python main.py azure --non-interactive` | See the [configuration reference](https://cloudexit.escapecloud.io/config/config-schema.html) for required permissions and config file format. @@ -116,6 +118,23 @@ python main.py aws --profile PROFILE --egress See the [egress reference](https://cloudexit.escapecloud.io/egress/overview.html) for details. +## Infrastructure-as-Code State Scan (alpha) + +Instead of connecting to a cloud account, cloudexit can build the assessment from a local **Terraform / OpenTofu state file** — no credentials, no API calls, nothing leaves your machine. + +```bash +python main.py aws --tfstate infra.tfstate +python main.py azure --tfstate infra.tfstate +``` + +If your state lives in a remote backend (S3, Azure Storage, Terraform Cloud, …), export it first: + +```bash +terraform state pull > infra.tfstate +``` + +See the [infrastructure-as-code reference](https://cloudexit.escapecloud.io/tfstate/overview.html) for details. + ## CI/CD cloudexit runs headlessly in CI pipelines via `--non-interactive` and environment variables. A ready-made GitHub Action is available: diff --git a/core/engine.py b/core/engine.py index 1d0b4d1..deb1aa7 100644 --- a/core/engine.py +++ b/core/engine.py @@ -1,4 +1,5 @@ # core/engine.py +import json import logging import os import boto3 @@ -14,6 +15,7 @@ from .utils_aws import build_aws_resource_inventory, build_aws_cost_inventory from .utils_azure import build_azure_resource_inventory, build_azure_cost_inventory from .utils_db import connect, load_data +from .utils_tfstate import build_tfstate_resource_inventory from .utils_report import ( generate_html_report, generate_pdf_report, @@ -256,7 +258,16 @@ def create_resource_inventory( try: - if cloud_service_provider == 1: # Azure + if provider_details.get("tfstatePath"): # Terraform / OpenTofu state file + coverage = build_tfstate_resource_inventory( + cloud_service_provider, provider_details, report_path, raw_data_path + ) + return { + "success": True, + "logs": "Resource inventory created successfully.", + "coverage": coverage, + } + elif cloud_service_provider == 1: # Azure build_azure_resource_inventory( cloud_service_provider, provider_details, report_path, raw_data_path ) @@ -497,6 +508,17 @@ def perform_risk_assessment( return {"success": False, "logs": str(e)} +def _load_tfstate_scope(raw_data_path: str) -> dict[str, Any] | None: + # A missing or unreadable manifest must not fail report generation. + manifest_path = os.path.join(raw_data_path, "tfstate_manifest.json") + try: + with open(manifest_path, "r", encoding="utf-8") as manifest_file: + return json.load(manifest_file).get("scope") + except (OSError, json.JSONDecodeError) as e: + logger.warning("Could not read the tfstate manifest for report scope: %s", e) + return None + + # Stage 6 def generate_report( cloud_service_provider: int, @@ -568,6 +590,11 @@ def generate_report( ) # Generate PDF report + tfstate_scope = ( + _load_tfstate_scope(raw_data_path) + if provider_details.get("tfstatePath") + else None + ) reports["PDF"] = generate_pdf_report( provider_details, report_path, @@ -581,6 +608,7 @@ def generate_report( alternatives, alternative_technologies, exit_strategy, + tfstate_scope=tfstate_scope, ) # Generate JSON report diff --git a/core/utils_report.py b/core/utils_report.py index 02937b0..0777439 100644 --- a/core/utils_report.py +++ b/core/utils_report.py @@ -279,15 +279,84 @@ def _build_summary_section(metadata, styles, content_style): return content -def _build_scope_section(metadata, provider_details, styles, content_style): +# Cap on how many subscriptions / resource groups / regions a scope row lists +# before it summarises the remainder. The full list stays in the manifest. +MAX_SCOPE_VALUES = 3 + + +def _shorten(value: str, head: int = 4, tail: int = 4) -> str: + if not isinstance(value, str): + return "N/A" + if len(value) <= head + tail + 1: + return value + return f"{value[:head]}…{value[-tail:]}" + + +def _join_scope_values(values) -> str: + if not values: + return "Not determinable from state" + shown = list(values)[:MAX_SCOPE_VALUES] + remaining = len(values) - len(shown) + text = ", ".join(shown) + return f"{text}, +{remaining} more" if remaining > 0 else text + + +def _build_tfstate_scope_rows(metadata, scope): + rows = [] + + file_name = scope.get("file") or "N/A" + sha256 = scope.get("sha256") + if sha256: + file_name = f"{file_name} (SHA-256 {_shorten(sha256)})" + rows.append(["File", file_name]) + + lineage = scope.get("lineage") + serial = scope.get("serial") + if lineage and serial is not None: + rows.append(["State", f"{lineage} (Serial: {serial})"]) + elif lineage: + rows.append(["State", lineage]) + elif serial is not None: + rows.append(["State", f"Serial: {serial}"]) + + if metadata["cloud_service_provider"] == 1: # Azure + rows.append(["Subscription", _join_scope_values(scope.get("subscriptions"))]) + rows.append( + ["Resource Group(s)", _join_scope_values(scope.get("resource_groups"))] + ) + else: # AWS + rows.append(["Region(s)", _join_scope_values(scope.get("locations"))]) + + return rows + + +def _build_scope_section( + metadata, provider_details, styles, content_style, tfstate_scope=None +): """Page 1: Scope of Assessment table.""" content = [] content.append(Paragraph("Scope of Assessment", styles["Heading2"])) - content.append(Paragraph("Defined scope of assessment:", content_style)) + + tfstate_path = provider_details.get("tfstatePath") + if tfstate_path: + content.append( + Paragraph( + "Assessed from Terraform state. Resources not managed by " + "Terraform are not included.", + content_style, + ) + ) + else: + content.append(Paragraph("Defined scope of assessment:", content_style)) scope_data = [["Name", "Value"]] - if metadata["cloud_service_provider"] == 1: # Azure + if tfstate_path: + # Fall back to the little we know from the config when the manifest is + # unavailable, rather than rendering credential rows that are all N/A. + scope = tfstate_scope or {"file": os.path.basename(tfstate_path)} + scope_data.extend(_build_tfstate_scope_rows(metadata, scope)) + elif metadata["cloud_service_provider"] == 1: # Azure scope_data.extend( [ ["Tenant ID", provider_details.get("tenantId", "N/A")], @@ -708,6 +777,7 @@ def generate_pdf_report( alternatives: list[dict[str, Any]], alternative_technologies: list[dict[str, Any]], exit_strategy: int, + tfstate_scope: dict[str, Any] | None = None, ) -> str: pdf_path = os.path.join(report_path, "report.pdf") @@ -728,7 +798,9 @@ def header_footer(canvas, doc): content = [] content += _build_summary_section(metadata, styles, content_style) - content += _build_scope_section(metadata, provider_details, styles, content_style) + content += _build_scope_section( + metadata, provider_details, styles, content_style, tfstate_scope + ) content += _build_cost_section(cost_data, styles, content_style) content += _build_risk_section( risk_data, diff --git a/core/utils_tfstate.py b/core/utils_tfstate.py new file mode 100644 index 0000000..de32486 --- /dev/null +++ b/core/utils_tfstate.py @@ -0,0 +1,392 @@ +# core/utils_tfstate.py +import hashlib +import json +import os +import logging +import re +import sqlite3 +from typing import Any +from collections import defaultdict + +from .utils_db import connect, load_data + +logger = logging.getLogger("core.engine.tfstate") + +# Terraform >= 0.12 and every OpenTofu release write state format version 4. +SUPPORTED_STATE_VERSION = 4 + +# Attribute keys that carry the region/location of an instance, in lookup order. +# AWS providers expose "region", Azure providers expose "location". +LOCATION_KEYS = ("region", "location") + +ARN_KEY = "arn" + +# arn:partition:service:region:account-id:resource +ARN_REGION_INDEX = 3 +ARN_MIN_FIELDS = 6 + +# Placeholder used when an instance carries no usable location attribute. +UNKNOWN_LOCATION = "unknown" + +# Terraform type prefixes per CSP. A state file can hold resources from any +# number of providers, so these decide which ones belong to the assessment. +CSP_TYPE_PREFIXES = {1: "azurerm_", 2: "aws_"} +CSP_NAMES = {1: "Azure", 2: "AWS"} +CSP_SUBCOMMANDS = {1: "azure", 2: "aws"} + +# Azure resource ids carry the subscription and resource group that define the +# assessed boundary: /subscriptions//resourceGroups//providers/... +AZURE_RESOURCE_ID_PATTERN = re.compile( + r"^/subscriptions/([^/]+)/resourceGroups/([^/]+)/", re.IGNORECASE +) + + +def parse_tfstate(path: str) -> dict[str, Any]: + try: + with open(path, "r", encoding="utf-8") as state_file: + state = json.load(state_file) + except OSError as e: + raise ValueError(f"Could not read Terraform state file '{path}': {e}") + except json.JSONDecodeError as e: + raise ValueError( + f"Terraform state file '{path}' is not valid JSON: {e}. " + "For a remote backend, export it first: " + "`terraform state pull > infra.tfstate`." + ) + + if not isinstance(state, dict): + raise ValueError( + f"Terraform state file '{path}' is not a state document " + "(expected a JSON object at the top level)." + ) + + version = state.get("version") + if version != SUPPORTED_STATE_VERSION: + raise ValueError( + f"Unsupported Terraform state version {version!r} in '{path}'. " + f"Only state format version {SUPPORTED_STATE_VERSION} is supported " + "(Terraform >= 0.12 and all OpenTofu versions produce version 4). " + "For a remote backend, export the current state with " + "`terraform state pull > infra.tfstate`." + ) + + return state + + +def _region_from_arn(arn: Any) -> str: + if not isinstance(arn, str) or not arn.startswith("arn:"): + return "" + + fields = arn.split(":") + if len(fields) < ARN_MIN_FIELDS: + return "" + + return fields[ARN_REGION_INDEX].strip().lower() + + +def _instance_location(attributes: Any) -> str: + if not isinstance(attributes, dict): + return UNKNOWN_LOCATION + + for key in LOCATION_KEYS: + value = attributes.get(key) + if isinstance(value, str): + location = value.strip().lower() + if location: + return location + + return _region_from_arn(attributes.get(ARN_KEY)) or UNKNOWN_LOCATION + + +def _instance_address( + module: str | None, resource_type: str, name: str, index_key: Any +) -> str: + address = f"{resource_type}.{name}" + if module: + address = f"{module}.{address}" + if index_key is not None: + address = f"{address}[{index_key}]" + return address + + +def extract_managed_resources(state: dict[str, Any]) -> list[dict[str, str]]: + records: list[dict[str, str]] = [] + + for resource in state.get("resources") or []: + if not isinstance(resource, dict): + continue + if resource.get("mode") != "managed": + continue + + resource_type = resource.get("type") + name = resource.get("name") + if not resource_type or not name: + logger.debug("Skipping state entry without a type/name: %r", resource) + continue + + module = resource.get("module") + + for instance in resource.get("instances") or []: + if not isinstance(instance, dict): + continue + records.append( + { + "address": _instance_address( + module, resource_type, name, instance.get("index_key") + ), + "type": resource_type, + "location": _instance_location(instance.get("attributes")), + } + ) + + return records + + +def file_sha256(path: str) -> str | None: + digest = hashlib.sha256() + try: + with open(path, "rb") as state_file: + for chunk in iter(lambda: state_file.read(65536), b""): + digest.update(chunk) + except OSError as e: + logger.warning("Could not hash the Terraform state file: %s", e) + return None + return digest.hexdigest() + + +def extract_state_scope( + state: dict[str, Any], cloud_service_provider: int +) -> dict[str, list[str]]: + subscriptions: set[str] = set() + resource_groups: set[str] = set() + + if cloud_service_provider == 1: # Azure + for resource in state.get("resources") or []: + if not isinstance(resource, dict) or resource.get("mode") != "managed": + continue + for instance in resource.get("instances") or []: + if not isinstance(instance, dict): + continue + attributes = instance.get("attributes") + if not isinstance(attributes, dict): + continue + resource_id = attributes.get("id") + if not isinstance(resource_id, str): + continue + match = AZURE_RESOURCE_ID_PATTERN.match(resource_id) + if match: + subscriptions.add(match.group(1)) + resource_groups.add(match.group(2)) + + return { + "subscriptions": sorted(subscriptions), + "resource_groups": sorted(resource_groups), + } + + +def _build_tf_code_mapping( + cloud_service_provider: int, db_path: str +) -> dict[str, dict[str, Any]]: + mapping: dict[str, dict[str, Any]] = {} + duplicates: set[str] = set() + + for item in load_data("resourcetype", db_path=db_path): + if item["csp"] != cloud_service_provider or item["status"] != "t": + continue + + tf_code = (item.get("tf_code") or "").strip() + if not tf_code: + continue + + existing = mapping.get(tf_code) + if existing is None: + mapping[tf_code] = {"id": item["id"], "name": item["name"]} + continue + + if tf_code not in duplicates: + duplicates.add(tf_code) + logger.warning( + "Terraform type '%s' maps to multiple resource types; " + "using the lowest id.", + tf_code, + ) + if item["id"] < existing["id"]: + mapping[tf_code] = {"id": item["id"], "name": item["name"]} + + return mapping + + +def _foreign_provider_summary(foreign_types: dict[str, int]) -> str: + for csp, prefix in CSP_TYPE_PREFIXES.items(): + count = sum(n for t, n in foreign_types.items() if t.startswith(prefix)) + if count: + return f"{count} {CSP_NAMES[csp]} resources" + + total = sum(foreign_types.values()) + sample = ", ".join(sorted(foreign_types)[:3]) + return f"{total} resources from other providers ({sample})" + + +def _no_matching_provider_error( + cloud_service_provider: int, tfstate_path: str, foreign_types: dict[str, int] +) -> ValueError: + own_name = CSP_NAMES.get(cloud_service_provider, "matching") + message = ( + f"No {own_name} resources found in '{os.path.basename(tfstate_path)}'. " + f"The state contains {_foreign_provider_summary(foreign_types)}." + ) + + # Point at the other subcommand only when we could actually assess it. + for csp, prefix in CSP_TYPE_PREFIXES.items(): + if csp == cloud_service_provider: + continue + if any(t.startswith(prefix) for t in foreign_types): + message += ( + f" Did you mean: python3 main.py {CSP_SUBCOMMANDS[csp]} " + f"--tfstate {tfstate_path}" + ) + break + + return ValueError(message) + + +def build_tfstate_resource_inventory( + cloud_service_provider: int, + provider_details: dict[str, Any], + report_path: str, + raw_data_path: str, +) -> dict[str, Any]: + tfstate_path = provider_details["tfstatePath"] + state = parse_tfstate(tfstate_path) + instances = extract_managed_resources(state) + + db_path = os.path.join(report_path, "data", "assessment.db") + resource_type_mapping = _build_tf_code_mapping(cloud_service_provider, db_path) + + # A state file may hold resources from any number of providers. Split them + # up front so a foreign resource is never silently lumped in with the + # same-provider glue types that legitimately have no dataset row. + own_prefix = CSP_TYPE_PREFIXES.get(cloud_service_provider, "") + foreign_types: defaultdict[str, int] = defaultdict(int) + own_instances = [] + for instance in instances: + if own_prefix and not instance["type"].startswith(own_prefix): + foreign_types[instance["type"]] += 1 + else: + own_instances.append(instance) + + # Nothing for the selected provider, but resources for another one: the + # wrong subcommand or the wrong file. An empty state is left alone. + if instances and not own_instances: + raise _no_matching_provider_error( + cloud_service_provider, tfstate_path, dict(foreign_types) + ) + + # Aggregate matched instances, and count the types we have no mapping for. + aggregated_resources: defaultdict[tuple[int, str], int] = defaultdict(int) + unmapped_types: defaultdict[str, int] = defaultdict(int) + counted: list[dict[str, Any]] = [] + + for instance in own_instances: + resource_info = resource_type_mapping.get(instance["type"]) + if not resource_info: + unmapped_types[instance["type"]] += 1 + continue + + resource_type_id = resource_info["id"] + aggregated_resources[(resource_type_id, instance["location"])] += 1 + counted.append( + { + "address": instance["address"], + "type": instance["type"], + "resource_type_id": resource_type_id, + "location": instance["location"], + } + ) + + # Insert aggregated data into SQLite + try: + with connect(db_path=db_path) as conn: + cursor = conn.cursor() + for ( + resource_type_id, + resource_location, + ), resource_count in aggregated_resources.items(): + try: + cursor.execute( + """ + INSERT INTO resource_inventory (resource_type, location, count) + VALUES (?, ?, ?) + ON CONFLICT(resource_type, location) DO UPDATE SET count = excluded.count + """, + (resource_type_id, resource_location, resource_count), + ) + except sqlite3.Error as e: + logger.error( + f"SQLite error while processing aggregated resource: {e}", + exc_info=True, + ) + conn.commit() + except sqlite3.Error as e: + logger.error( + f"Error writing the tfstate resource inventory: {e}", exc_info=True + ) + raise + + # Counts are per instance, not per type: "9 types skipped" can hide any + # number of dropped resources, which is exactly what needs surfacing. + excluded_foreign = sum(foreign_types.values()) + excluded_unmapped = sum(unmapped_types.values()) + coverage = { + "instances_total": len(instances), + "instances_counted": len(counted), + "instances_excluded_other_provider": excluded_foreign, + "instances_excluded_unmapped": excluded_unmapped, + } + + # What the report's Scope of Assessment section renders from. The hash ties + # a report back to the exact file it was produced from. + scope = { + "file": os.path.basename(tfstate_path), + "sha256": file_sha256(tfstate_path), + "lineage": state.get("lineage"), + "serial": state.get("serial"), + "locations": sorted({entry["location"] for entry in counted}), + **extract_state_scope(state, cloud_service_provider), + } + + # Manifest of what was counted. Instance attributes carry secrets + # (passwords, connection strings, keys) and are deliberately never written. + manifest = { + "source_file": os.path.basename(tfstate_path), + "terraform_version": state.get("terraform_version"), + "state_serial": state.get("serial"), + "scope": scope, + "coverage": coverage, + "counted": counted, + "unmapped_types": dict(unmapped_types), + "other_provider_types": dict(foreign_types), + } + + manifest_path = os.path.join(raw_data_path, "tfstate_manifest.json") + try: + with open(manifest_path, "w", encoding="utf-8") as manifest_file: + json.dump(manifest, manifest_file, indent=4) + except OSError as e: + logger.error(f"Could not write the tfstate manifest: {e}", exc_info=True) + + if unmapped_types: + logger.warning( + "%d Terraform resource type(s) had no matching resource type and were " + "skipped; see raw_data/tfstate_manifest.json for details.", + len(unmapped_types), + ) + + if excluded_foreign: + logger.warning( + "%d resource(s) in the state belong to another cloud provider and were " + "not assessed.", + excluded_foreign, + ) + + return coverage diff --git a/main.py b/main.py index 9d41111..28a3faa 100644 --- a/main.py +++ b/main.py @@ -118,6 +118,16 @@ class ConfigError(Exception): pass +def _reject_egress_with_tfstate(args) -> None: + # --egress needs live API access, so it cannot run against a state file. + if getattr(args, "tfstate", None) and getattr(args, "egress", False): + console.print( + "[red]--egress cannot be combined with --tfstate. Egress estimation " + "requires live cloud API access.[/red]" + ) + sys.exit(codes.CONFIG) + + def _aws_provider_from_profile(profile: str) -> dict: if not is_aws_cli_installed(): console.print( @@ -204,6 +214,9 @@ def _aws_provider_from_prompt() -> dict: def handle_aws(args): cloud_provider = 2 + _reject_egress_with_tfstate(args) + tfstate_path = getattr(args, "tfstate", None) + if args.config: config = load_config(args.config) if not config: @@ -234,10 +247,15 @@ def handle_aws(args): assessment_type = require_env_int( "ESC_ASSESSMENT_TYPE", "assessment type (1 or 2)", {1, 2} ) - if args.profile: + if tfstate_path: + provider_details = {"tfstatePath": tfstate_path} + elif args.profile: provider_details = _aws_provider_from_profile(args.profile) else: provider_details = _aws_provider_from_env() + elif tfstate_path: + exit_strategy, assessment_type = prompt_required_inputs() + provider_details = {"tfstatePath": tfstate_path} elif args.profile: provider_details = _aws_provider_from_profile(args.profile) exit_strategy, assessment_type = prompt_required_inputs() @@ -409,6 +427,9 @@ def _azure_provider_from_prompt() -> dict: def handle_azure(args): cloud_provider = 1 + _reject_egress_with_tfstate(args) + tfstate_path = getattr(args, "tfstate", None) + if args.config: config = load_config(args.config) if not config: @@ -439,7 +460,13 @@ def handle_azure(args): assessment_type = require_env_int( "ESC_ASSESSMENT_TYPE", "assessment type (1 or 2)", {1, 2} ) - provider_details = _azure_provider_noninteractive(args) + if tfstate_path: + provider_details = {"tfstatePath": tfstate_path} + else: + provider_details = _azure_provider_noninteractive(args) + elif tfstate_path: + exit_strategy, assessment_type = prompt_required_inputs() + provider_details = {"tfstatePath": tfstate_path} elif args.cli: provider_details = _azure_provider_from_cli() exit_strategy, assessment_type = prompt_required_inputs() @@ -480,6 +507,17 @@ def run_assessment( print_step("Configuration validation failed.", status="error", logs=str(e)) sys.exit(codes.CONFIG) + # tfstate mode reads a local state file: no credentials, no permission + # check, no cost data — and no egress estimation, which needs live APIs. + is_tfstate = bool(config["providerDetails"].get("tfstatePath")) + if is_tfstate and egress: + print_step( + "Configuration validation failed.", + status="error", + logs="--egress requires live cloud API access and cannot be used with a Terraform state file.", + ) + sys.exit(codes.CONFIG) + # Detect ExitCloud Integration mode, jwt = resolve_mode() if dry_run: @@ -514,59 +552,70 @@ def run_assessment( # Stage 1: Verify Credentials console.print("-------------------------------------------") console.print("Stage #1 - Validate Credentials", style="bold") - # Test Connection - connection_success, logs = verify_credentials( - config["cloudServiceProvider"], config["providerDetails"] - ) - if connection_success: - print_step(f"Connecting to {provider_name}...", status="ok") + if is_tfstate: + print_step( + "Skipped - tfstate mode (no cloud credentials used).", status="warning" + ) else: - print_step(f"Connecting to {provider_name}...", status="error") - console.print(f" ↳ {logs}", style="dim") - logger.error(f"Credential verification failed: {logs}") - sys.exit(codes.CREDENTIALS) + # Test Connection + connection_success, logs = verify_credentials( + config["cloudServiceProvider"], config["providerDetails"] + ) + if connection_success: + print_step(f"Connecting to {provider_name}...", status="ok") + else: + print_step(f"Connecting to {provider_name}...", status="error") + console.print(f" ↳ {logs}", style="dim") + logger.error(f"Credential verification failed: {logs}") + sys.exit(codes.CREDENTIALS) console.print("-------------------------------------------") # Stage 2: Test Permissions console.print("Stage #2 - Validate Permissions", style="bold") - # Labels for permission types - permission_reader_label = ( - "Reader" if config["cloudServiceProvider"] == 1 else "ViewOnlyAccess" - ) - permission_cost_label = ( - "Cost Management Reader" - if config["cloudServiceProvider"] == 1 - else "AWSBillingReadOnlyAccess" - ) - - # Test permissions with spinners - with console.status("Validating permissions...", spinner="dots"): - permission_valid, permission_reader, permission_cost, logs = ( - test_permissions( - config["cloudServiceProvider"], config["providerDetails"] - ) - ) - - # Output results for permission checks - if permission_reader: - print_step(f"Checking {permission_reader_label}...", status="ok") - else: + if is_tfstate: print_step( - f"Checking {permission_reader_label}...", status="error", logs=logs + "Skipped - tfstate mode (no cloud permissions needed).", + status="warning", ) - - if permission_cost: - print_step(f"Checking {permission_cost_label}...", status="ok") else: - print_step( - f"Checking {permission_cost_label}...", status="error", logs=logs + # Labels for permission types + permission_reader_label = ( + "Reader" if config["cloudServiceProvider"] == 1 else "ViewOnlyAccess" ) + permission_cost_label = ( + "Cost Management Reader" + if config["cloudServiceProvider"] == 1 + else "AWSBillingReadOnlyAccess" + ) + + # Test permissions with spinners + with console.status("Validating permissions...", spinner="dots"): + permission_valid, permission_reader, permission_cost, logs = ( + test_permissions( + config["cloudServiceProvider"], config["providerDetails"] + ) + ) + + # Output results for permission checks + if permission_reader: + print_step(f"Checking {permission_reader_label}...", status="ok") + else: + print_step( + f"Checking {permission_reader_label}...", status="error", logs=logs + ) + + if permission_cost: + print_step(f"Checking {permission_cost_label}...", status="ok") + else: + print_step( + f"Checking {permission_cost_label}...", status="error", logs=logs + ) - # Exit if permissions are invalid - if not permission_valid: - logger.error(f"Permission validation failed: {logs}") - sys.exit(codes.PERMISSIONS) + # Exit if permissions are invalid + if not permission_valid: + logger.error(f"Permission validation failed: {logs}") + sys.exit(codes.PERMISSIONS) console.print("-------------------------------------------") @@ -585,9 +634,25 @@ def run_assessment( ) if result["success"]: - print_step( - f"Building resource inventory for {provider_name}...", status="ok" - ) + # In tfstate mode, resources belonging to another cloud provider are + # dropped silently by design. Report the exclusion at step level so a + # partially-assessed state cannot pass for a complete one. + coverage = result.get("coverage") or {} + excluded = coverage.get("instances_excluded_other_provider", 0) + if excluded: + print_step( + f"Building resource inventory for {provider_name}...", + status="warning", + logs=( + f"Assessed {coverage['instances_counted']} of " + f"{coverage['instances_total']} resources; {excluded} excluded " + f"(other cloud provider). See raw_data/tfstate_manifest.json." + ), + ) + else: + print_step( + f"Building resource inventory for {provider_name}...", status="ok" + ) else: print_step( f"Building resource inventory for {provider_name}...", @@ -601,27 +666,34 @@ def run_assessment( # Stage 4: Build Cost Inventory console.print("Stage #4 - Build Cost Inventory", style="bold") - # Use a spinner to indicate progress - with console.status( - f"Building cost inventory for {provider_name}...", spinner="dots" - ): - cost_result = create_cost_inventory( - config["cloudServiceProvider"], - config["providerDetails"], - report_path, - raw_data_path, - ) - - # Handle the result - if cost_result["success"]: - print_step(f"Building cost inventory for {provider_name}...", status="ok") - else: + if is_tfstate: print_step( - f"Building cost inventory for {provider_name}...", - status="error", - logs=cost_result["logs"], + "Skipped - tfstate mode (no billing data available).", status="warning" ) - sys.exit(codes.COST_INVENTORY) + else: + # Use a spinner to indicate progress + with console.status( + f"Building cost inventory for {provider_name}...", spinner="dots" + ): + cost_result = create_cost_inventory( + config["cloudServiceProvider"], + config["providerDetails"], + report_path, + raw_data_path, + ) + + # Handle the result + if cost_result["success"]: + print_step( + f"Building cost inventory for {provider_name}...", status="ok" + ) + else: + print_step( + f"Building cost inventory for {provider_name}...", + status="error", + logs=cost_result["logs"], + ) + sys.exit(codes.COST_INVENTORY) console.print("-------------------------------------------") @@ -841,6 +913,8 @@ def parse_arguments(): " python3 main.py azure --config config.json --dry-run\n" " python3 main.py aws --config config.json --egress # Estimate egress data volume\n" " python3 main.py azure --config config.json --egress\n" + " python3 main.py aws --tfstate infra.tfstate # Assess a Terraform/OpenTofu state file\n" + " python3 main.py azure --tfstate infra.tfstate --dry-run\n" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) @@ -879,6 +953,14 @@ def parse_arguments(): type=str, help="AWS profile name to use credentials from ~/.aws/credentials.", ) + aws_group.add_argument( + "--tfstate", + type=str, + help=( + "Path to a Terraform/OpenTofu state file. Builds the inventory from " + "the state instead of the AWS APIs; no credentials are used." + ), + ) aws_parser.add_argument( "--name", type=str, help="Assessment Name (Optional / Max. 50 characters)." ) @@ -916,6 +998,14 @@ def parse_arguments(): action="store_true", help="Use Azure CLI credentials for authentication.", ) + azure_group.add_argument( + "--tfstate", + type=str, + help=( + "Path to a Terraform/OpenTofu state file. Builds the inventory from " + "the state instead of the Azure APIs; no credentials are used." + ), + ) azure_parser.add_argument( "--name", type=str, help="Assessment Name (Optional / Max. 50 characters)." ) diff --git a/tests/test_engine.py b/tests/test_engine.py index e4e46e1..c055445 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -1,7 +1,10 @@ +import json +import tempfile import unittest +from pathlib import Path from unittest.mock import MagicMock, patch -from core.engine import sync_assessment, test_permissions +from core.engine import _load_tfstate_scope, sync_assessment, test_permissions class TestPermissionsAwsHybridMode(unittest.TestCase): @@ -159,5 +162,35 @@ def test_local_db_failure_returns_dict_not_raises(self): self.assertIn("store server risks", result["logs"]) +class LoadTfstateScopeTests(unittest.TestCase): + def _write_manifest(self, directory, payload): + path = Path(directory) / "tfstate_manifest.json" + path.write_text(json.dumps(payload), encoding="utf-8") + + def test_returns_scope_block_from_manifest(self): + scope = {"file": "infra.tfstate", "serial": 7, "locations": ["eu-central-1"]} + with tempfile.TemporaryDirectory() as tmp_dir: + self._write_manifest(tmp_dir, {"scope": scope, "counted": []}) + + self.assertEqual(_load_tfstate_scope(tmp_dir), scope) + + def test_missing_manifest_returns_none_without_raising(self): + with tempfile.TemporaryDirectory() as tmp_dir: + self.assertIsNone(_load_tfstate_scope(tmp_dir)) + + def test_corrupt_manifest_returns_none_without_raising(self): + with tempfile.TemporaryDirectory() as tmp_dir: + path = Path(tmp_dir) / "tfstate_manifest.json" + path.write_text("{not json", encoding="utf-8") + + self.assertIsNone(_load_tfstate_scope(tmp_dir)) + + def test_manifest_without_scope_returns_none(self): + with tempfile.TemporaryDirectory() as tmp_dir: + self._write_manifest(tmp_dir, {"counted": []}) + + self.assertIsNone(_load_tfstate_scope(tmp_dir)) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_pdf_sections.py b/tests/test_pdf_sections.py index 342338c..4c8bd2e 100644 --- a/tests/test_pdf_sections.py +++ b/tests/test_pdf_sections.py @@ -145,6 +145,120 @@ def test_unknown_provider_returns_na(self): self.assertEqual(len(tables), 1) +class BuildTfstateScopeSectionTests(unittest.TestCase): + _SHA = "a3f1" + "b" * 56 + "9c2e" + _LINEAGE = "121f3456-7d0f-a7be-ceb0-c91d44c6a689" + + def setUp(self): + self.styles, self.content_style = _make_styles() + self.fixture = build_report_fixture() + + def _rows(self, metadata, provider_details, tfstate_scope=None): + content = _build_scope_section( + metadata, + provider_details, + self.styles, + self.content_style, + tfstate_scope, + ) + tables = [item for item in content if isinstance(item, Table)] + self.assertEqual(len(tables), 1) + return {row[0]: row[1] for row in tables[0]._cellvalues[1:]} + + def _aws_scope(self, **overrides): + scope = { + "file": "infra.tfstate", + "sha256": self._SHA, + "lineage": self._LINEAGE, + "serial": 7, + "locations": ["eu-central-1"], + "subscriptions": [], + "resource_groups": [], + } + scope.update(overrides) + return scope + + def test_aws_renders_file_state_and_regions(self): + rows = self._rows( + self.fixture["metadata"], + {"tfstatePath": "/tmp/infra.tfstate"}, + self._aws_scope(), + ) + + self.assertEqual(list(rows), ["File", "State", "Region(s)"]) + self.assertEqual(rows["File"], "infra.tfstate (SHA-256 a3f1…9c2e)") + self.assertEqual(rows["State"], f"{self._LINEAGE} (Serial: 7)") + self.assertEqual(rows["Region(s)"], "eu-central-1") + + def test_azure_renders_subscription_and_resource_groups(self): + metadata = {**self.fixture["metadata"], "cloud_service_provider": 1} + scope = self._aws_scope( + subscriptions=["0299bf7a-8ca8-479b-8659-c62e62cd7bae"], + resource_groups=["escape-test-rg-1", "escape-test-rg-2"], + ) + + rows = self._rows(metadata, {"tfstatePath": "/tmp/infra.tfstate"}, scope) + + self.assertEqual( + list(rows), ["File", "State", "Subscription", "Resource Group(s)"] + ) + self.assertEqual(rows["Subscription"], "0299bf7a-8ca8-479b-8659-c62e62cd7bae") + self.assertEqual( + rows["Resource Group(s)"], "escape-test-rg-1, escape-test-rg-2" + ) + # Azure deliberately mirrors the live table, which shows no location. + self.assertNotIn("Location(s)", rows) + self.assertNotIn("Region(s)", rows) + + def test_many_values_are_capped_with_a_remainder(self): + metadata = {**self.fixture["metadata"], "cloud_service_provider": 1} + scope = self._aws_scope( + subscriptions=["sub-1"], + resource_groups=[f"rg-{i}" for i in range(6)], + ) + + rows = self._rows(metadata, {"tfstatePath": "/tmp/infra.tfstate"}, scope) + + self.assertEqual(rows["Resource Group(s)"], "rg-0, rg-1, rg-2, +3 more") + + def test_missing_values_report_not_determinable(self): + rows = self._rows( + self.fixture["metadata"], + {"tfstatePath": "/tmp/infra.tfstate"}, + self._aws_scope(locations=[]), + ) + + self.assertEqual(rows["Region(s)"], "Not determinable from state") + + def test_state_row_omitted_when_lineage_and_serial_absent(self): + rows = self._rows( + self.fixture["metadata"], + {"tfstatePath": "/tmp/infra.tfstate"}, + self._aws_scope(lineage=None, serial=None), + ) + + self.assertEqual(list(rows), ["File", "Region(s)"]) + self.assertEqual(rows["File"], "infra.tfstate (SHA-256 a3f1…9c2e)") + + def test_missing_manifest_falls_back_to_file_name_only(self): + rows = self._rows( + self.fixture["metadata"], {"tfstatePath": "/tmp/infra.tfstate"}, None + ) + + self.assertEqual(rows["File"], "infra.tfstate") + self.assertEqual(rows["Region(s)"], "Not determinable from state") + # Never the credential rows, which would all be N/A here. + self.assertNotIn("Access Key", rows) + + def test_live_mode_table_is_unchanged(self): + rows = self._rows( + self.fixture["metadata"], self.fixture["provider_details"], None + ) + + self.assertEqual(list(rows), ["Access Key", "Secret Key", "Region"]) + self.assertEqual(rows["Region"], "eu-central-1") + + class BuildCostSectionTests(unittest.TestCase): def setUp(self): self.styles, self.content_style = _make_styles() diff --git a/tests/test_report_pipeline.py b/tests/test_report_pipeline.py index 536bad6..415b1fa 100644 --- a/tests/test_report_pipeline.py +++ b/tests/test_report_pipeline.py @@ -141,6 +141,90 @@ def test_generate_pdf_report_creates_non_empty_file(self): self.assertGreater(pdf_file.stat().st_size, 0) +class EmptyCostInventoryReportTests(unittest.TestCase): + """tfstate mode skips Stage 4, so reports must render with resources but no costs.""" + + def test_html_report_renders_resources_without_cost_data(self): + fixture = build_report_fixture() + + with tempfile.TemporaryDirectory() as report_dir: + html_path = generate_html_report( + report_dir, + fixture["metadata"], + fixture["resource_type_mapping"], + fixture["resource_inventory"], + [], + None, + fixture["risk_data"], + fixture["risk_definitions"], + fixture["alternatives"], + fixture["alternative_technologies"], + fixture["exit_strategy"], + fixture["alternative_technology_organizations"], + ) + + self.assertTrue(Path(html_path).exists()) + html = Path(html_path).read_text(encoding="utf-8") + + self.assertIn("EC2 Instance", html) + self.assertIn("No cost data available.", html) + self.assertNotIn('id="costsChart"', html) + + def test_json_report_has_empty_cost_inventory(self): + fixture = build_report_fixture() + + with tempfile.TemporaryDirectory() as tmp_dir: + raw_data_path = Path(tmp_dir) / "raw_data" + raw_data_path.mkdir() + + json_path = generate_json_report( + str(raw_data_path), + fixture["metadata"], + fixture["resource_type_mapping"], + fixture["resource_inventory"], + [], + None, + fixture["risk_data"], + fixture["risk_definitions"], + fixture["alternatives"], + fixture["alternative_technologies"], + fixture["exit_strategy"], + ) + + payload = json.loads(Path(json_path).read_text(encoding="utf-8")) + + self.assertEqual(payload["data"]["cost_inventory"], []) + self.assertEqual( + payload["data"]["resource_inventory"][0]["resource_name"], "EC2 Instance" + ) + + def test_pdf_report_generates_without_cost_data(self): + fixture = build_report_fixture() + + with tempfile.TemporaryDirectory() as report_dir: + stage_report_assets(report_dir) + + pdf_path = generate_pdf_report( + fixture["provider_details"], + report_dir, + fixture["metadata"], + fixture["resource_type_mapping"], + fixture["resource_inventory"], + [], + None, + fixture["risk_data"], + fixture["risk_definitions"], + fixture["alternatives"], + fixture["alternative_technologies"], + fixture["exit_strategy"], + ) + + pdf_file = Path(pdf_path) + + self.assertTrue(pdf_file.exists()) + self.assertGreater(pdf_file.stat().st_size, 0) + + class ReportTransformTests(unittest.TestCase): def test_transform_cost_inventory_for_json_sorts_months(self): unsorted_costs = [ diff --git a/tests/test_utils_and_main.py b/tests/test_utils_and_main.py index 0f8371a..4b80a9a 100644 --- a/tests/test_utils_and_main.py +++ b/tests/test_utils_and_main.py @@ -905,5 +905,249 @@ def test_falls_back_to_config_when_env_not_set(self): self.assertIsNone(token) +TFSTATE_CONFIG = { + "name": "Tfstate Assessment", + "cloudServiceProvider": 2, + "exitStrategy": 1, + "assessmentType": 1, + "providerDetails": {"tfstatePath": "config/aws-01.tfstate"}, +} + + +class TfstateCliTests(unittest.TestCase): + _ENV = {"ESC_EXIT_STRATEGY": "3", "ESC_ASSESSMENT_TYPE": "1"} + + def test_egress_with_tfstate_exits_config_for_aws(self): + with patch("main.console.print"): + with self.assertRaises(SystemExit) as ctx: + main.handle_aws(_ni_aws_args(tfstate="infra.tfstate", egress=True)) + self.assertEqual(ctx.exception.code, codes.CONFIG) + + def test_egress_with_tfstate_exits_config_for_azure(self): + with patch("main.console.print"): + with self.assertRaises(SystemExit) as ctx: + main.handle_azure(_ni_azure_args(tfstate="infra.tfstate", egress=True)) + self.assertEqual(ctx.exception.code, codes.CONFIG) + + def test_aws_non_interactive_needs_no_aws_env_vars(self): + env = { + k: v + for k, v in os.environ.items() + if not k.startswith("AWS_") and not k.startswith("ESC_") + } + env.update(self._ENV) + with ( + patch.dict(os.environ, env, clear=True), + patch("main.run_assessment") as mock_run, + patch("main.console.print"), + ): + main.handle_aws(_ni_aws_args(tfstate="infra.tfstate")) + + mock_run.assert_called_once() + config_arg = mock_run.call_args[0][0] + self.assertEqual(config_arg["exitStrategy"], 3) + self.assertEqual(config_arg["assessmentType"], 1) + self.assertEqual( + config_arg["providerDetails"], {"tfstatePath": "infra.tfstate"} + ) + + def test_azure_non_interactive_needs_no_azure_env_vars(self): + env = { + k: v + for k, v in os.environ.items() + if not k.startswith("AZURE_") and not k.startswith("ESC_") + } + env.update(self._ENV) + with ( + patch.dict(os.environ, env, clear=True), + patch("main.run_assessment") as mock_run, + patch("main.console.print"), + ): + main.handle_azure(_ni_azure_args(tfstate="infra.tfstate")) + + mock_run.assert_called_once() + config_arg = mock_run.call_args[0][0] + self.assertEqual(config_arg["cloudServiceProvider"], 1) + self.assertEqual(config_arg["exitStrategy"], 3) + self.assertEqual( + config_arg["providerDetails"], {"tfstatePath": "infra.tfstate"} + ) + + def test_config_file_carries_tfstate_path_through_handle_aws(self): + with tempfile.TemporaryDirectory() as tmp_dir: + state_path = Path(tmp_dir) / "infra.tfstate" + state_path.write_text( + json.dumps({"version": 4, "resources": []}), encoding="utf-8" + ) + config_path = Path(tmp_dir) / "config.json" + config_path.write_text( + json.dumps( + { + "cloudServiceProvider": 2, + "exitStrategy": 1, + "assessmentType": 1, + "providerDetails": {"tfstatePath": str(state_path)}, + } + ), + encoding="utf-8", + ) + + with ( + patch("main.run_assessment") as mock_run, + patch("main.console.print"), + ): + main.handle_aws(_ni_aws_args(config=str(config_path))) + + config_arg = mock_run.call_args[0][0] + self.assertEqual( + config_arg["providerDetails"], {"tfstatePath": str(state_path)} + ) + # The same config must survive validation without credential fields. + self.assertTrue(main.validate_config(config_arg)) + + def test_tfstate_parses_from_both_subcommands(self): + with patch("sys.argv", ["main.py", "aws", "--tfstate", "infra.tfstate"]): + args = main.parse_arguments() + self.assertEqual(args.tfstate, "infra.tfstate") + + with patch( + "sys.argv", ["main.py", "azure", "--tfstate", "infra.tfstate", "--dry-run"] + ): + args = main.parse_arguments() + self.assertEqual(args.tfstate, "infra.tfstate") + self.assertTrue(args.dry_run) + + def test_tfstate_is_mutually_exclusive_with_config(self): + with patch( + "sys.argv", ["main.py", "aws", "--config", "c.json", "--tfstate", "s"] + ): + with self.assertRaises(SystemExit): + main.parse_arguments() + + def test_run_assessment_skips_credential_permission_and_cost_stages(self): + with tempfile.TemporaryDirectory() as tmp_dir: + raw_data_path = os.path.join(tmp_dir, "raw_data") + os.makedirs(raw_data_path, exist_ok=True) + + with ( + patch("main.validate_config"), + patch("main.resolve_mode", return_value=("offline", None)), + patch("main.create_directory", return_value=(tmp_dir, raw_data_path)), + patch("main.verify_credentials") as mock_creds, + patch("main.test_permissions") as mock_perms, + patch("main.create_cost_inventory") as mock_cost, + patch( + "main.create_resource_inventory", + return_value={"success": True, "logs": ""}, + ) as mock_inventory, + patch( + "main.perform_risk_assessment", + return_value={"success": True, "logs": ""}, + ) as mock_risk, + patch( + "main.generate_report", + return_value={"success": True, "reports": {}}, + ) as mock_report, + patch("main.print_step"), + patch("main.console.print"), + ): + main.run_assessment(TFSTATE_CONFIG.copy(), "aws") + + mock_creds.assert_not_called() + mock_perms.assert_not_called() + mock_cost.assert_not_called() + mock_inventory.assert_called_once_with( + 2, {"tfstatePath": "config/aws-01.tfstate"}, tmp_dir, raw_data_path + ) + mock_risk.assert_called_once() + mock_report.assert_called_once() + + def _run_tfstate_stage3(self, inventory_result): + with tempfile.TemporaryDirectory() as tmp_dir: + raw_data_path = os.path.join(tmp_dir, "raw_data") + os.makedirs(raw_data_path, exist_ok=True) + + with ( + patch("main.validate_config"), + patch("main.resolve_mode", return_value=("offline", None)), + patch("main.create_directory", return_value=(tmp_dir, raw_data_path)), + patch("main.create_resource_inventory", return_value=inventory_result), + patch( + "main.perform_risk_assessment", + return_value={"success": True, "logs": ""}, + ), + patch( + "main.generate_report", + return_value={"success": True, "reports": {}}, + ), + patch("main.print_step") as mock_step, + patch("main.console.print"), + ): + main.run_assessment(TFSTATE_CONFIG.copy(), "aws") + + return [ + call + for call in mock_step.call_args_list + if "resource inventory" in call.args[0] + ] + + def test_excluded_foreign_resources_downgrade_stage3_to_warning(self): + calls = self._run_tfstate_stage3( + { + "success": True, + "logs": "", + "coverage": { + "instances_total": 31, + "instances_counted": 3, + "instances_excluded_other_provider": 27, + "instances_excluded_unmapped": 1, + }, + } + ) + + self.assertEqual(len(calls), 1) + self.assertEqual(calls[0].kwargs["status"], "warning") + logs = calls[0].kwargs["logs"] + self.assertIn("Assessed 3 of 31 resources", logs) + self.assertIn("27 excluded", logs) + + def test_clean_tfstate_run_keeps_stage3_ok(self): + calls = self._run_tfstate_stage3( + { + "success": True, + "logs": "", + "coverage": { + "instances_total": 8, + "instances_counted": 5, + "instances_excluded_other_provider": 0, + "instances_excluded_unmapped": 3, + }, + } + ) + + self.assertEqual(len(calls), 1) + self.assertEqual(calls[0].kwargs["status"], "ok") + + def test_live_mode_result_without_coverage_keeps_stage3_ok(self): + calls = self._run_tfstate_stage3({"success": True, "logs": ""}) + + self.assertEqual(len(calls), 1) + self.assertEqual(calls[0].kwargs["status"], "ok") + + def test_run_assessment_rejects_egress_from_tfstate_config_file(self): + with ( + patch("main.validate_config"), + patch("main.resolve_mode", return_value=("offline", None)), + patch("main.create_directory") as mock_dir, + patch("main.print_step"), + patch("main.console.print"), + ): + with self.assertRaises(SystemExit) as ctx: + main.run_assessment(TFSTATE_CONFIG.copy(), "aws", egress=True) + + self.assertEqual(ctx.exception.code, codes.CONFIG) + mock_dir.assert_not_called() + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_utils_tfstate.py b/tests/test_utils_tfstate.py new file mode 100644 index 0000000..f12983c --- /dev/null +++ b/tests/test_utils_tfstate.py @@ -0,0 +1,890 @@ +import json +import os +import sqlite3 +import tempfile +import unittest +from pathlib import Path + +from core.utils_tfstate import ( + build_tfstate_resource_inventory, + extract_managed_resources, + extract_state_scope, + file_sha256, + parse_tfstate, +) + +# Schema subset the tfstate builder touches, mirroring datasets/data.db. +SCHEMA = """ +CREATE TABLE resourcetype ( + id INTEGER PRIMARY KEY, + csp INTEGER NOT NULL, + code TEXT NOT NULL, + name TEXT NOT NULL, + icon TEXT NOT NULL, + status TEXT CHECK(status IN ('t','f')) NOT NULL, + tf_code TEXT +); +CREATE TABLE resource_inventory ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + resource_type INTEGER NOT NULL, + location TEXT NOT NULL, + count INTEGER NOT NULL, + UNIQUE(resource_type, location) +); +""" + + +def build_state(resources, **overrides): + state = { + "version": 4, + "terraform_version": "1.14.6", + "serial": 7, + "lineage": "e1c2f0c0-0000-0000-0000-000000000000", + "resources": resources, + } + state.update(overrides) + return state + + +def managed(resource_type, name, instances, module=None): + entry = { + "mode": "managed", + "type": resource_type, + "name": name, + "provider": f'provider["registry.terraform.io/hashicorp/{resource_type.split("_")[0]}"]', + "instances": instances, + } + if module: + entry["module"] = module + return entry + + +def instance(attributes=None, index_key=None): + entry = {"schema_version": 0, "attributes": attributes or {}} + if index_key is not None: + entry["index_key"] = index_key + return entry + + +def write_state(directory, state, filename="infra.tfstate"): + path = Path(directory) / filename + path.write_text(json.dumps(state), encoding="utf-8") + return str(path) + + +def seed_db(db_path, rows): + """rows: (id, csp, code, name, status, tf_code)""" + conn = sqlite3.connect(db_path) + conn.executescript(SCHEMA) + conn.executemany( + "INSERT INTO resourcetype (id, csp, code, name, icon, status, tf_code) " + "VALUES (?, ?, ?, ?, '/icons/misc/no_image.png', ?, ?)", + rows, + ) + conn.commit() + conn.close() + + +class ParseTfstateTests(unittest.TestCase): + def test_rejects_missing_file(self): + with self.assertRaisesRegex(ValueError, "Could not read Terraform state file"): + parse_tfstate("/tmp/does-not-exist.tfstate") + + def test_rejects_invalid_json(self): + with tempfile.TemporaryDirectory() as tmp_dir: + path = Path(tmp_dir) / "broken.tfstate" + path.write_text("{not json", encoding="utf-8") + + with self.assertRaisesRegex(ValueError, "not valid JSON"): + parse_tfstate(str(path)) + + def test_rejects_legacy_state_version(self): + with tempfile.TemporaryDirectory() as tmp_dir: + path = write_state(tmp_dir, build_state([], version=3)) + + with self.assertRaisesRegex(ValueError, "version 4"): + parse_tfstate(path) + + def test_rejects_non_object_document(self): + with tempfile.TemporaryDirectory() as tmp_dir: + path = Path(tmp_dir) / "list.tfstate" + path.write_text("[]", encoding="utf-8") + + with self.assertRaisesRegex(ValueError, "not a state document"): + parse_tfstate(str(path)) + + def test_accepts_minimal_valid_state(self): + with tempfile.TemporaryDirectory() as tmp_dir: + path = write_state( + tmp_dir, + build_state( + [ + managed( + "aws_s3_bucket", "this", [instance({"region": "eu-west-1"})] + ) + ] + ), + ) + + state = parse_tfstate(path) + + self.assertEqual(state["version"], 4) + self.assertEqual(state["terraform_version"], "1.14.6") + self.assertEqual(len(state["resources"]), 1) + + +class ExtractManagedResourcesTests(unittest.TestCase): + def test_excludes_data_sources(self): + state = build_state( + [ + { + "mode": "data", + "type": "aws_caller_identity", + "name": "current", + "instances": [instance({"region": "eu-west-1"})], + }, + managed("aws_s3_bucket", "this", [instance({"region": "eu-west-1"})]), + ] + ) + + records = extract_managed_resources(state) + + self.assertEqual([r["type"] for r in records], ["aws_s3_bucket"]) + + def test_one_record_per_instance(self): + state = build_state( + [ + managed( + "aws_dynamodb_table", + "this", + [ + instance({"region": "eu-central-1"}, index_key="alpha"), + instance({"region": "eu-central-1"}, index_key="beta"), + instance({"region": "eu-central-1"}, index_key="gamma"), + ], + ) + ] + ) + + records = extract_managed_resources(state) + + self.assertEqual(len(records), 3) + self.assertEqual({r["location"] for r in records}, {"eu-central-1"}) + + def test_extracts_and_lowercases_aws_region(self): + state = build_state( + [managed("aws_s3_bucket", "this", [instance({"region": " EU-West-1 "})])] + ) + + records = extract_managed_resources(state) + + self.assertEqual(records[0]["location"], "eu-west-1") + + def test_extracts_and_lowercases_azure_location(self): + state = build_state( + [ + managed( + "azurerm_managed_disk", + "this", + [instance({"location": "NorthEurope"})], + ) + ] + ) + + records = extract_managed_resources(state) + + self.assertEqual(records[0]["location"], "northeurope") + + def test_falls_back_to_unknown_location(self): + state = build_state( + [ + managed("aws_iam_role", "this", [instance({"name": "role"})]), + managed("aws_iam_policy", "this", [instance({"region": " "})]), + managed("aws_iam_user", "this", [instance()]), + ] + ) + + records = extract_managed_resources(state) + + self.assertEqual([r["location"] for r in records], ["unknown"] * 3) + + def test_falls_back_to_region_from_own_arn(self): + state = build_state( + [ + managed( + "aws_dynamodb_table", + "this", + [ + instance( + { + "arn": "arn:aws:dynamodb:EU-Central-1:266579820564:table/orders" + } + ) + ], + ) + ] + ) + + records = extract_managed_resources(state) + + self.assertEqual(records[0]["location"], "eu-central-1") + + def test_region_attribute_wins_over_arn(self): + state = build_state( + [ + managed( + "aws_efs_file_system", + "this", + [ + instance( + { + "region": "eu-west-1", + "arn": "arn:aws:elasticfilesystem:us-east-1:266579820564:file-system/fs-1", + } + ) + ], + ) + ] + ) + + records = extract_managed_resources(state) + + self.assertEqual(records[0]["location"], "eu-west-1") + + def test_ignores_arns_of_referenced_resources(self): + # stream_arn / kms_key_arn point at other resources, which may live in a + # different region or account; only the resource's own arn may be read. + state = build_state( + [ + managed( + "aws_dynamodb_table", + "this", + [ + instance( + { + "stream_arn": "arn:aws:dynamodb:us-east-1:999999999999:table/x/stream/2024", + "kms_key_arn": "arn:aws:kms:ap-south-1:999999999999:key/abc", + "restore_source_table_arn": "arn:aws:dynamodb:sa-east-1:999999999999:table/y", + } + ) + ], + ) + ] + ) + + records = extract_managed_resources(state) + + self.assertEqual(records[0]["location"], "unknown") + + def test_global_and_malformed_arns_fall_back_to_unknown(self): + state = build_state( + [ + # Global services write an empty region field by design. + managed( + "aws_s3_bucket", + "this", + [instance({"arn": "arn:aws:s3:::my-bucket"})], + ), + managed( + "aws_iam_role", + "this", + [instance({"arn": "arn:aws:iam::266579820564:role/admin"})], + ), + managed( + "aws_vpc", "trunc", [instance({"arn": "arn:aws:ec2:eu-west-1"})] + ), + managed("aws_vpc", "nonsense", [instance({"arn": "not-an-arn"})]), + managed("aws_vpc", "wrongtype", [instance({"arn": 42})]), + ] + ) + + records = extract_managed_resources(state) + + self.assertEqual([r["location"] for r in records], ["unknown"] * 5) + + def test_arn_account_id_is_never_extracted(self): + state = build_state( + [ + managed( + "aws_dynamodb_table", + "this", + [ + instance( + { + "arn": "arn:aws:dynamodb:eu-central-1:266579820564:table/t" + } + ) + ], + ) + ] + ) + + records = extract_managed_resources(state) + + self.assertEqual(records[0]["location"], "eu-central-1") + self.assertNotIn("266579820564", json.dumps(records)) + + def test_address_includes_module_and_index_key(self): + state = build_state( + [ + managed( + "aws_s3_bucket", + "this", + [instance({"region": "eu-west-1"}, index_key=3)], + module="module.storage", + ), + managed("aws_vpc", "main", [instance({"region": "eu-west-1"})]), + ] + ) + + records = extract_managed_resources(state) + + self.assertEqual(records[0]["address"], "module.storage.aws_s3_bucket.this[3]") + self.assertEqual(records[1]["address"], "aws_vpc.main") + + def test_never_copies_attributes(self): + state = build_state( + [ + managed( + "azurerm_storage_account", + "this", + [ + instance( + { + "location": "westeurope", + "primary_connection_string": "SUPER_SECRET_VALUE", + } + ) + ], + ) + ] + ) + + records = extract_managed_resources(state) + + self.assertEqual(set(records[0].keys()), {"address", "type", "location"}) + self.assertNotIn("SUPER_SECRET_VALUE", json.dumps(records)) + + +class StateScopeTests(unittest.TestCase): + def test_sha256_matches_the_file_contents(self): + import hashlib + + with tempfile.TemporaryDirectory() as tmp_dir: + path = write_state(tmp_dir, build_state([])) + expected = hashlib.sha256(Path(path).read_bytes()).hexdigest() + + self.assertEqual(file_sha256(path), expected) + + def test_sha256_returns_none_for_unreadable_file(self): + self.assertIsNone(file_sha256("/tmp/does-not-exist.tfstate")) + + def test_extracts_azure_subscription_and_resource_groups(self): + state = build_state( + [ + managed( + "azurerm_storage_account", + "this", + [ + instance( + { + "id": "/subscriptions/sub-a/resourceGroups/rg-1/providers/Microsoft.Storage/storageAccounts/x" + } + ), + instance( + { + "id": "/subscriptions/sub-a/resourceGroups/rg-2/providers/Microsoft.Storage/storageAccounts/y" + } + ), + ], + ) + ] + ) + + scope = extract_state_scope(state, 1) + + self.assertEqual(scope["subscriptions"], ["sub-a"]) + self.assertEqual(scope["resource_groups"], ["rg-1", "rg-2"]) + + def test_handles_multiple_subscriptions(self): + state = build_state( + [ + managed( + "azurerm_managed_disk", + "this", + [ + instance( + { + "id": "/subscriptions/sub-b/resourceGroups/rg-1/providers/x" + } + ), + instance( + { + "id": "/subscriptions/sub-a/resourceGroups/rg-1/providers/x" + } + ), + ], + ) + ] + ) + + scope = extract_state_scope(state, 1) + + self.assertEqual(scope["subscriptions"], ["sub-a", "sub-b"]) + + def test_ignores_data_sources_and_unparsable_ids(self): + state = build_state( + [ + { + "mode": "data", + "type": "azurerm_resource_group", + "name": "this", + "instances": [ + instance( + { + "id": "/subscriptions/data-sub/resourceGroups/rg-x/providers/y" + } + ) + ], + }, + managed("azurerm_managed_disk", "a", [instance({"id": "not-an-id"})]), + managed("azurerm_managed_disk", "b", [instance({"id": 42})]), + managed("azurerm_managed_disk", "c", [instance()]), + ] + ) + + scope = extract_state_scope(state, 1) + + self.assertEqual(scope["subscriptions"], []) + self.assertEqual(scope["resource_groups"], []) + + def test_aws_state_yields_no_subscription_or_resource_group(self): + state = build_state( + [ + managed( + "aws_s3_bucket", + "this", + [instance({"region": "eu-west-1", "id": "my-bucket"})], + ) + ] + ) + + scope = extract_state_scope(state, 2) + + self.assertEqual(scope, {"subscriptions": [], "resource_groups": []}) + + +class BuildTfstateResourceInventoryTests(unittest.TestCase): + AWS_ROWS = [ + (444, 2, "AWS.s3.list_buckets.Buckets", "S3 Bucket", "t", "aws_s3_bucket"), + ( + 300, + 2, + "AWS.dynamodb.list_tables.TableNames", + "DynamoDB Table", + "t", + "aws_dynamodb_table", + ), + ( + 500, + 2, + "AWS.ec2.describe_instances.Reservations", + "EC2 Instance", + "f", + "aws_instance", + ), + ( + 600, + 1, + "Microsoft.Compute/virtualMachines", + "Virtual Machine", + "t", + "azurerm_linux_virtual_machine", + ), + ] + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp.cleanup) + self.report_path = self._tmp.name + self.raw_data_path = os.path.join(self.report_path, "raw_data") + os.makedirs(os.path.join(self.report_path, "data")) + os.makedirs(self.raw_data_path) + self.db_path = os.path.join(self.report_path, "data", "assessment.db") + + def _inventory(self): + conn = sqlite3.connect(self.db_path) + rows = conn.execute( + "SELECT resource_type, location, count FROM resource_inventory " + "ORDER BY resource_type, location" + ).fetchall() + conn.close() + return rows + + def _manifest(self): + path = Path(self.raw_data_path) / "tfstate_manifest.json" + return json.loads(path.read_text(encoding="utf-8")), path.read_text( + encoding="utf-8" + ) + + def _build(self, state, csp=2, rows=None, filename="infra.tfstate"): + seed_db(self.db_path, self.AWS_ROWS if rows is None else rows) + state_path = write_state(self._tmp.name, state, filename=filename) + build_tfstate_resource_inventory( + csp, + {"tfstatePath": state_path}, + self.report_path, + self.raw_data_path, + ) + return state_path + + def _build_returning(self, state, csp=2, rows=None): + seed_db(self.db_path, self.AWS_ROWS if rows is None else rows) + state_path = write_state(self._tmp.name, state) + return build_tfstate_resource_inventory( + csp, + {"tfstatePath": state_path}, + self.report_path, + self.raw_data_path, + ) + + def test_arn_derived_regions_aggregate_and_stay_out_of_the_manifest(self): + state = build_state( + [ + managed( + "aws_dynamodb_table", + "tables", + [ + instance( + { + "arn": "arn:aws:dynamodb:eu-central-1:266579820564:table/a" + }, + index_key="a", + ), + instance( + { + "arn": "arn:aws:dynamodb:eu-central-1:266579820564:table/b" + }, + index_key="b", + ), + instance( + {"arn": "arn:aws:dynamodb:us-east-1:266579820564:table/c"}, + index_key="c", + ), + ], + ) + ] + ) + + self._build(state) + + self.assertEqual( + self._inventory(), + [(300, "eu-central-1", 2), (300, "us-east-1", 1)], + ) + + _, serialized = self._manifest() + self.assertNotIn("266579820564", serialized) + self.assertNotIn("arn:aws", serialized) + + def test_aggregates_matched_types_per_location(self): + state = build_state( + [ + managed( + "aws_s3_bucket", + "this", + [ + instance({"region": "eu-west-1"}, index_key=0), + instance({"region": "eu-west-1"}, index_key=1), + instance({"region": "us-east-1"}, index_key=2), + ], + ), + managed( + "aws_dynamodb_table", + "tables", + [instance({"region": "eu-west-1"})], + ), + ] + ) + + self._build(state) + + self.assertEqual( + self._inventory(), + [ + (300, "eu-west-1", 1), + (444, "eu-west-1", 2), + (444, "us-east-1", 1), + ], + ) + + def test_unmapped_types_are_skipped_and_recorded_in_manifest(self): + state = build_state( + [ + managed("aws_s3_bucket", "this", [instance({"region": "eu-west-1"})]), + managed( + "aws_s3_bucket_versioning", + "this", + [ + instance({"region": "eu-west-1"}, index_key=0), + instance({"region": "eu-west-1"}, index_key=1), + ], + ), + ] + ) + + self._build(state) + + self.assertEqual(self._inventory(), [(444, "eu-west-1", 1)]) + + manifest, _ = self._manifest() + self.assertEqual(manifest["unmapped_types"], {"aws_s3_bucket_versioning": 2}) + self.assertEqual( + [entry["type"] for entry in manifest["counted"]], ["aws_s3_bucket"] + ) + + def test_duplicate_tf_code_resolves_to_lowest_id(self): + rows = [ + (444, 2, "AWS.s3.list_buckets.Buckets", "S3 Bucket", "t", "aws_s3_bucket"), + ( + 293, + 2, + "AWS.glacier.list_vaults.VaultList", + "Glacier Vault", + "t", + "aws_s3_bucket", + ), + ] + state = build_state( + [managed("aws_s3_bucket", "this", [instance({"region": "eu-west-1"})])] + ) + + self._build(state, rows=rows) + + self.assertEqual(self._inventory(), [(293, "eu-west-1", 1)]) + + def test_ignores_disabled_rows_and_other_csp_rows(self): + state = build_state( + [ + managed("aws_instance", "this", [instance({"region": "eu-west-1"})]), + managed( + "azurerm_linux_virtual_machine", + "this", + [instance({"location": "westeurope"})], + ), + ] + ) + + self._build(state, csp=2) + + self.assertEqual(self._inventory(), []) + manifest, _ = self._manifest() + # Same-provider glue and foreign resources are reported separately. + self.assertEqual(manifest["unmapped_types"], {"aws_instance": 1}) + self.assertEqual( + manifest["other_provider_types"], {"azurerm_linux_virtual_machine": 1} + ) + + def test_rejects_state_with_no_resources_for_the_selected_provider(self): + state = build_state( + [ + managed( + "azurerm_storage_account", + "this", + [ + instance({"location": "westeurope"}, index_key=0), + instance({"location": "westeurope"}, index_key=1), + ], + ) + ] + ) + seed_db(self.db_path, self.AWS_ROWS) + state_path = write_state(self._tmp.name, state) + + with self.assertRaises(ValueError) as ctx: + build_tfstate_resource_inventory( + 2, {"tfstatePath": state_path}, self.report_path, self.raw_data_path + ) + + message = str(ctx.exception) + self.assertIn("No AWS resources found", message) + self.assertIn("2 Azure resources", message) + self.assertIn("main.py azure --tfstate", message) + + def test_rejects_state_holding_only_unsupported_providers(self): + state = build_state( + [ + managed("google_storage_bucket", "this", [instance({})]), + managed("cloudflare_record", "this", [instance({})]), + ] + ) + seed_db(self.db_path, self.AWS_ROWS) + state_path = write_state(self._tmp.name, state) + + with self.assertRaises(ValueError) as ctx: + build_tfstate_resource_inventory( + 2, {"tfstatePath": state_path}, self.report_path, self.raw_data_path + ) + + message = str(ctx.exception) + self.assertIn("No AWS resources found", message) + self.assertIn("other providers", message) + # Nothing to redirect to — cloudexit cannot assess these. + self.assertNotIn("Did you mean", message) + + def test_empty_state_does_not_raise(self): + coverage = self._build_returning(build_state([])) + + self.assertEqual(coverage["instances_total"], 0) + self.assertEqual(self._inventory(), []) + + def test_mixed_state_counts_own_and_reports_excluded(self): + state = build_state( + [ + managed( + "aws_s3_bucket", + "this", + [ + instance({"region": "eu-west-1"}, index_key=0), + instance({"region": "eu-west-1"}, index_key=1), + instance({"region": "eu-west-1"}, index_key=2), + ], + ), + managed( + "azurerm_storage_account", + "this", + [ + instance({"location": "westeurope"}, index_key=i) + for i in range(27) + ], + ), + managed("aws_s3_bucket_versioning", "this", [instance({})]), + ] + ) + + coverage = self._build_returning(state) + + self.assertEqual(self._inventory(), [(444, "eu-west-1", 3)]) + self.assertEqual( + coverage, + { + "instances_total": 31, + "instances_counted": 3, + "instances_excluded_other_provider": 27, + "instances_excluded_unmapped": 1, + }, + ) + + manifest, _ = self._manifest() + self.assertEqual(manifest["coverage"], coverage) + self.assertEqual( + manifest["other_provider_types"], {"azurerm_storage_account": 27} + ) + self.assertEqual(manifest["unmapped_types"], {"aws_s3_bucket_versioning": 1}) + + def test_manifest_carries_no_attributes(self): + state = build_state( + [ + managed( + "aws_s3_bucket", + "this", + [ + instance( + { + "region": "eu-west-1", + "bucket": "my-bucket", + "generated_password": "SUPER_SECRET_VALUE", + } + ) + ], + ) + ] + ) + + state_path = self._build(state) + + manifest, serialized = self._manifest() + self.assertNotIn("attributes", serialized) + self.assertNotIn("SUPER_SECRET_VALUE", serialized) + self.assertNotIn("my-bucket", serialized) + self.assertEqual(manifest["source_file"], os.path.basename(state_path)) + self.assertNotIn(os.path.dirname(state_path), serialized) + self.assertEqual(manifest["terraform_version"], "1.14.6") + self.assertEqual(manifest["state_serial"], 7) + self.assertEqual( + set(manifest["counted"][0].keys()), + {"address", "type", "resource_type_id", "location"}, + ) + + def test_manifest_scope_block_records_file_state_and_locations(self): + import hashlib + + state = build_state( + [ + managed( + "aws_s3_bucket", + "this", + [ + instance({"region": "eu-west-1"}, index_key=0), + instance({"region": "us-east-1"}, index_key=1), + ], + ) + ] + ) + + state_path = self._build(state) + + manifest, _ = self._manifest() + scope = manifest["scope"] + self.assertEqual(scope["file"], "infra.tfstate") + self.assertEqual( + scope["sha256"], hashlib.sha256(Path(state_path).read_bytes()).hexdigest() + ) + self.assertEqual(scope["lineage"], "e1c2f0c0-0000-0000-0000-000000000000") + self.assertEqual(scope["serial"], 7) + self.assertEqual(scope["locations"], ["eu-west-1", "us-east-1"]) + self.assertEqual(scope["subscriptions"], []) + self.assertEqual(scope["resource_groups"], []) + + def test_rerun_is_idempotent(self): + state = build_state( + [ + managed( + "aws_s3_bucket", + "this", + [ + instance({"region": "eu-west-1"}, index_key=0), + instance({"region": "eu-west-1"}, index_key=1), + ], + ) + ] + ) + + state_path = self._build(state) + build_tfstate_resource_inventory( + 2, + {"tfstatePath": state_path}, + self.report_path, + self.raw_data_path, + ) + + self.assertEqual(self._inventory(), [(444, "eu-west-1", 2)]) + + def test_invalid_state_raises_value_error(self): + seed_db(self.db_path, self.AWS_ROWS) + + with self.assertRaisesRegex(ValueError, "version 4"): + build_tfstate_resource_inventory( + 2, + { + "tfstatePath": write_state( + self._tmp.name, build_state([], version=3) + ) + }, + self.report_path, + self.raw_data_path, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_validate.py b/tests/test_validate.py index eb6cde2..73c8456 100644 --- a/tests/test_validate.py +++ b/tests/test_validate.py @@ -1,4 +1,7 @@ +import json +import tempfile import unittest +from pathlib import Path from utils.validate import validate_config, validate_region @@ -108,5 +111,99 @@ def test_rejects_aws_config_with_invalid_region(self): validate_config(config) +class ValidateTfstateConfigTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp.cleanup) + self.state_path = Path(self._tmp.name) / "infra.tfstate" + self.state_path.write_text( + json.dumps({"version": 4, "resources": []}), encoding="utf-8" + ) + + def _config(self, cloud_service_provider, **provider_details): + return { + "name": "Tfstate Assessment", + "assessmentType": 1, + "cloudServiceProvider": cloud_service_provider, + "exitStrategy": 1, + "providerDetails": provider_details, + } + + def test_accepts_aws_tfstate_config_without_credentials(self): + config = self._config(2, tfstatePath=str(self.state_path)) + + self.assertTrue(validate_config(config)) + + def test_accepts_azure_tfstate_config_without_credentials(self): + config = self._config(1, tfstatePath=str(self.state_path)) + + self.assertTrue(validate_config(config)) + + def test_rejects_missing_tfstate_file(self): + config = self._config(2, tfstatePath=str(self.state_path) + ".missing") + + with self.assertRaisesRegex(ValueError, "Terraform state file not found"): + validate_config(config) + + def test_rejects_empty_tfstate_path(self): + config = self._config(2, tfstatePath=" ") + + with self.assertRaisesRegex(ValueError, "Invalid tfstatePath"): + validate_config(config) + + def test_rejects_non_string_tfstate_path(self): + config = self._config(2, tfstatePath=42) + + with self.assertRaisesRegex(ValueError, "Invalid tfstatePath"): + validate_config(config) + + def test_generic_checks_still_apply_in_tfstate_mode(self): + config = self._config(2, tfstatePath=str(self.state_path)) + config["exitStrategy"] = 9 + + with self.assertRaisesRegex(ValueError, "Invalid exitStrategy"): + validate_config(config) + + def test_rejects_tfstate_combined_with_aws_credentials(self): + config = self._config( + 2, + tfstatePath=str(self.state_path), + accessKey="AKIA_TEST", + secretKey="SECRET_TEST", + ) + + with self.assertRaisesRegex(ValueError, "cannot combine tfstatePath") as ctx: + validate_config(config) + + self.assertIn("accessKey", str(ctx.exception)) + self.assertIn("secretKey", str(ctx.exception)) + + def test_rejects_tfstate_combined_with_azure_credentials(self): + config = self._config( + 1, + tfstatePath=str(self.state_path), + tenantId="tenant-id", + clientSecret="client-secret", + subscriptionId="sub-id", + ) + + with self.assertRaisesRegex(ValueError, "cannot combine tfstatePath"): + validate_config(config) + + def test_rejects_tfstate_combined_with_region(self): + config = self._config( + 2, tfstatePath=str(self.state_path), region="eu-central-1" + ) + + with self.assertRaisesRegex(ValueError, "cannot combine tfstatePath"): + validate_config(config) + + def test_rejects_tfstate_combined_with_cli_credential_object(self): + config = self._config(1, tfstatePath=str(self.state_path), credential=object()) + + with self.assertRaisesRegex(ValueError, "cannot combine tfstatePath"): + validate_config(config) + + if __name__ == "__main__": unittest.main() diff --git a/utils/utils.py b/utils/utils.py index 0a87386..b6f264c 100644 --- a/utils/utils.py +++ b/utils/utils.py @@ -179,7 +179,9 @@ def print_help_message(): console.print(" python3 main.py aws --config config/aws.json") console.print(" python3 main.py aws --profile PROFILE") console.print(" python3 main.py aws --name 'DMS System' ") + console.print(" python3 main.py aws --tfstate infra.tfstate") console.print(" python3 main.py azure") console.print(" python3 main.py azure --config config/azure.json") console.print(" python3 main.py azure --cli") console.print(" python3 main.py azure --name 'DMS System'") + console.print(" python3 main.py azure --tfstate infra.tfstate --dry-run") diff --git a/utils/validate.py b/utils/validate.py index f733116..8d8ade4 100644 --- a/utils/validate.py +++ b/utils/validate.py @@ -1,7 +1,26 @@ # utils/validate.py +import os from typing import Any from .constants import REGION_CHOICES, REQUIRED_FIELDS_AZURE, REQUIRED_FIELDS_AWS +# Fields that only mean something when connecting to a live account. tfstate +# mode derives all of them from the state file, so accepting them alongside +# tfstatePath would silently ignore whatever the user supplied. The CLI already +# rejects --profile/--cli/--config next to --tfstate; this is the same rule for +# configuration files. +LIVE_ONLY_FIELDS = ( + "accessKey", + "secretKey", + "sessionToken", + "region", + "credential", + "tenantId", + "clientId", + "clientSecret", + "subscriptionId", + "resourceGroupName", +) + def validate_region(region: str) -> None: valid_regions = [choice[0] for choice in REGION_CHOICES] @@ -45,6 +64,29 @@ def validate_config(config: dict[str, Any]) -> bool: # Validate providerDetails based on cloudServiceProvider provider_details = config.get("providerDetails", {}) + + # tfstate mode reads a local state file instead of the provider APIs, so no + # credentials (and no region) are involved. + if "tfstatePath" in provider_details: + tfstate_path = provider_details.get("tfstatePath") + if not isinstance(tfstate_path, str) or not tfstate_path.strip(): + raise ValueError( + "Invalid tfstatePath in providerDetails. Must be a non-empty path " + "to a Terraform/OpenTofu state file." + ) + if not os.path.isfile(tfstate_path): + raise ValueError(f"Terraform state file not found: {tfstate_path}") + + conflicting = [f for f in LIVE_ONLY_FIELDS if f in provider_details] + if conflicting: + raise ValueError( + "providerDetails cannot combine tfstatePath with live connection " + f"fields: {', '.join(conflicting)}. Remove them to scan the state " + "file, or remove tfstatePath to run a live assessment." + ) + + return True + if cloud_service_provider == 1: # Azure # Skip validation of clientId and clientSecret if using CLI credentials if provider_details.get("credential") is not None: