add Data Landscape & Egress Estimation stage

README.md

@@ -105,6 +105,17 @@ See the [configuration reference](https://cloudexit.escapecloud.io/config/config
 
 Want to see how a regulatory-aligned report looks (DORA / FINMA / UK PRA)? Run with `--dry-run` and send the output `payload.json` to request_report@escapecloud.io — we'll generate a sample you can share with your risk or compliance team.
 
+## Data Landscape & Egress Estimation (alpha)
+
+Add `--egress` to any assessment to measure how much data lives in the assessed scope and estimate the one-time internet egress fee for moving it out — based on the provider's tiered list prices, with no additional permissions required.
+
+```bash
+python main.py azure --cli --egress
+python main.py aws --profile PROFILE --egress
+```
+
+See the [egress reference](https://cloudexit.escapecloud.io/egress/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:

core/utils_db.py

@@ -19,6 +19,7 @@
     "alternativetechnology",
     "alternativetechnologyorganization",
     "risk",
+    "egresspricing",
 }
 
 

core/utils_egress.py

@@ -0,0 +1,112 @@
+# core/utils_egress.py
+import json
+import logging
+import os
+from typing import Any
+from datetime import datetime, timezone
+
+logger = logging.getLogger("core.engine.egress")
+
+GIB = 1024**3
+
+
+def new_row(
+    resource_id: str, name: str, resource_type: str, label: str, category: str
+) -> dict[str, Any]:
+    return {
+        "id": resource_id,
+        "name": name,
+        "type": resource_type,
+        "label": label,
+        "category": category,
+        "size_bytes": None,
+        "size_unknown": False,
+        "tier_bytes": None,
+        "flags": [],
+        "notes": [],
+    }
+
+
+def format_bytes(size_bytes: int | float | None) -> str:
+    if size_bytes is None:
+        return "n/a"
+    value = float(size_bytes)
+    for unit in ("B", "KiB", "MiB", "GiB", "TiB"):
+        if value < 1024:
+            return f"{value:.1f} {unit}"
+        value /= 1024
+    return f"{value:.1f} PiB"
+
+
+def compute_totals(
+    rows: list[dict[str, Any]], archive_tiers: set[str]
+) -> dict[str, Any]:
+    known_size_bytes = sum(
+        row["size_bytes"] for row in rows if row["size_bytes"] is not None
+    )
+    archive_tier_bytes = sum(
+        size
+        for row in rows
+        for tier, size in (row["tier_bytes"] or {}).items()
+        if tier in archive_tiers
+    )
+    unknown_count = sum(1 for row in rows if row["size_unknown"])
+    return {
+        "known_size_bytes": known_size_bytes,
+        "archive_tier_bytes": archive_tier_bytes,
+        "resources_discovered": len(rows),
+        "resources_with_unknown_size": unknown_count,
+    }
+
+
+def estimate_egress(
+    cloud_service_provider: int,
+    provider_details: dict[str, Any],
+    raw_data_path: str,
+    *,
+    name: str,
+    exit_strategy: int,
+    assessment_type: int,
+) -> dict[str, Any]:
+    try:
+        if cloud_service_provider == 1:  # Azure
+            from .utils_egress_azure import collect_azure_egress
+
+            rows, archive_tiers = collect_azure_egress(provider_details)
+        elif cloud_service_provider == 2:  # AWS
+            from .utils_egress_aws import collect_aws_egress
+
+            rows, archive_tiers = collect_aws_egress(provider_details)
+        else:
+            raise ValueError(
+                f"Unsupported cloud service provider: {cloud_service_provider}"
+            )
+
+        json_payload = {
+            "meta": {
+                "name": name,
+                "cloud_service_provider": cloud_service_provider,
+                "exit_strategy": exit_strategy,
+                "assessment_type": assessment_type,
+                "timestamp": datetime.now(timezone.utc).strftime(
+                    "%Y-%m-%d %H:%M:%S UTC"
+                ),
+            },
+            "data": {
+                "resources": rows,
+                "totals": compute_totals(rows, archive_tiers),
+            },
+        }
+        json_path = os.path.join(raw_data_path, "egress_estimate.json")
+        with open(json_path, "w", encoding="utf-8") as json_file:
+            json.dump(json_payload, json_file, indent=4)
+
+        return {
+            "success": True,
+            "logs": "Egress estimation completed successfully.",
+            "json_path": json_path,
+        }
+
+    except Exception as e:
+        logger.error(f"Error estimating egress: {str(e)}", exc_info=True)
+        return {"success": False, "logs": str(e)}