diff --git a/cloudexit/cli/app.py b/cloudexit/cli/app.py index 46c2bd7..7d6f829 100644 --- a/cloudexit/cli/app.py +++ b/cloudexit/cli/app.py @@ -10,6 +10,7 @@ import traceback from dataclasses import replace from typing import Any +from urllib.parse import urlsplit from rich.console import Console from rich.logging import RichHandler from datetime import datetime @@ -98,6 +99,7 @@ prompt_required_inputs, print_help_message, ) +from cloudexit.log_redaction import SecretRedactingFilter from cloudexit.private_files import restrict_tree from cloudexit.terminal import ask, print_step from cloudexit.runconfig.loader import ( @@ -130,9 +132,14 @@ # Loggers with no diagnostic value ever — pinned to WARNING regardless of verbosity, # so they never flood the -vv console or run.log. MSAL (behind the Azure -# credentials) writes the client ID in its debug output, and run.log is a file -# users share. -_ALWAYS_QUIET = ("PIL", "msal") +# credentials) writes the client ID in its debug output, botocore's signing and +# credential-resolution internals carry the credentials themselves, and run.log +# is a file users share. +_ALWAYS_QUIET = ("PIL", "msal", "botocore.auth", "botocore.credentials") + +# A last line of defence for the records that do stay on: both handlers drop +# credentials out of whatever reaches them. +_redacting_filter = SecretRedactingFilter() def configure_logging(verbose: int = 0) -> None: @@ -156,6 +163,7 @@ def configure_logging(verbose: int = 0) -> None: console=console, show_path=False, rich_tracebacks=True ) console_handler.setLevel(console_level) + console_handler.addFilter(_redacting_filter) root.addHandler(console_handler) third_party_level = logging.DEBUG if verbose >= 2 else logging.WARNING @@ -191,6 +199,7 @@ def add_run_log_handler(report_path: str) -> None: file_handler.setFormatter( logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s") ) + file_handler.addFilter(_redacting_filter) logging.getLogger().addHandler(file_handler) _run_log_handler = file_handler except OSError as exc: @@ -352,7 +361,7 @@ def handle_aws(args): if args.non_interactive: exit_strategy = require_env_int( - "ESC_EXIT_STRATEGY", "exit strategy (1 or 3)", {1, 3} + "ESC_EXIT_STRATEGY", "exit strategy (1, 2 or 3)", {1, 2, 3} ) assessment_type = require_env_int( "ESC_ASSESSMENT_TYPE", "assessment type (1 or 2)", {1, 2} @@ -364,13 +373,19 @@ def handle_aws(args): else: provider_details = _aws_provider_from_env() elif tfstate_path: - exit_strategy, assessment_type = prompt_required_inputs() + exit_strategy, assessment_type = prompt_required_inputs( + _asks_assessment_type(args) + ) provider_details = {"tfstatePath": tfstate_path} elif args.profile: provider_details = _aws_provider_from_profile(args.profile) - exit_strategy, assessment_type = prompt_required_inputs() + exit_strategy, assessment_type = prompt_required_inputs( + _asks_assessment_type(args) + ) else: - exit_strategy, assessment_type = prompt_required_inputs() + exit_strategy, assessment_type = prompt_required_inputs( + _asks_assessment_type(args) + ) provider_details = _aws_provider_from_prompt() config = build_config( @@ -596,7 +611,7 @@ def handle_azure(args): if args.non_interactive: exit_strategy = require_env_int( - "ESC_EXIT_STRATEGY", "exit strategy (1 or 3)", {1, 3} + "ESC_EXIT_STRATEGY", "exit strategy (1, 2 or 3)", {1, 2, 3} ) assessment_type = require_env_int( "ESC_ASSESSMENT_TYPE", "assessment type (1 or 2)", {1, 2} @@ -606,13 +621,19 @@ def handle_azure(args): else: provider_details = _azure_provider_noninteractive(args) elif tfstate_path: - exit_strategy, assessment_type = prompt_required_inputs() + exit_strategy, assessment_type = prompt_required_inputs( + _asks_assessment_type(args) + ) provider_details = {"tfstatePath": tfstate_path} elif args.cli: provider_details = _azure_provider_from_cli() - exit_strategy, assessment_type = prompt_required_inputs() + exit_strategy, assessment_type = prompt_required_inputs( + _asks_assessment_type(args) + ) else: - exit_strategy, assessment_type = prompt_required_inputs() + exit_strategy, assessment_type = prompt_required_inputs( + _asks_assessment_type(args) + ) provider_details = _azure_provider_from_prompt() config = build_config( @@ -658,6 +679,26 @@ def _missing_platform(host, key): return "No host and no API key are configured." +def _host_display(host): + # The origin only: a path, a query or credentials have no place in a status + # line the user reads on every run + parts = urlsplit(host or "") + if not parts.hostname: + return host + origin = parts.hostname if not parts.port else f"{parts.hostname}:{parts.port}" + return f"{parts.scheme}://{origin}" if parts.scheme else origin + + +def _asks_assessment_type(args): + # Standard needs an upload, so the question is only real when one will happen + settings = getattr(args, "settings", None) + if settings is None or getattr(args, "dry_run", False): + return False + if settings.upload.value == "off": + return False + return bool(settings.host.value and settings.api_key.value) + + def _upload_mode(settings, dry_run): mode = settings.upload.value if settings else "auto" if not dry_run: @@ -782,7 +823,7 @@ def run_assessment( ) sys.exit(codes.CONFIG) - # Detect ExitCloud Integration + # Detect a configured platform mode = _upload_mode(settings, dry_run) host = settings.host.value if settings else None key = settings.api_key.value if settings else None @@ -813,9 +854,7 @@ def run_assessment( print_step("Upload is off – nothing is sent to a platform.", status="ok") # Without an upload only the Basic assessment is possible if config["assessmentType"] != 1: - print_step( - "Forcing Basic Assessment: uploads are off.", status="warning" - ) + print_step("Basic assessment: Standard needs an upload.", status="ok") config["assessmentType"] = 1 elif configured: # A key that cannot be right is a failed upload, found before the @@ -823,24 +862,24 @@ def run_assessment( # the local report. try: check_key(settings.api_key.value) - print_step("ExitCloud integration configured.", status="ok") + print_step(f"Platform configured: {_host_display(host)}.", status="ok") if flush_spool: _flush_earlier_syncs(settings) except ApiError as error: signin_failure = error_failure(error) print_step( - "ExitCloud integration configured, but the API key is not usable.", + "Platform configured, but the API key is not usable.", status="warning", logs=f"{error.message}. The report is still written locally; the upload counts as failed.", ) else: - print_step("ExitCloud integration not configured.", status="warning") + print_step( + "Offline run: no platform configured; the report is written locally.", + status="ok", + ) # Without a platform only the Basic assessment is possible if config["assessmentType"] != 1: - print_step( - "Forcing Basic Assessment: no platform is configured.", - status="warning", - ) + print_step("Basic assessment: Standard needs an upload.", status="ok") config["assessmentType"] = 1 RUN["skipped_reason"] = ( "dry_run" if dry_run else "off" if mode == "off" else "offline" @@ -1196,18 +1235,6 @@ def run_assessment( egress_failed = True # Output the report path after the separator - if not (dry_run or non_interactive): - console.print("-------------------------------------------") - console.print( - "Need to see how a regulatory-aligned report looks (DORA / FINMA / UK PRA evidence support)?" - ) - console.print( - "We can generate one from your --dry-run output and share a sample with you -- or forward this to your risk or compliance team." - ) - console.print( - "Point-in-time evidence only; not a compliance certification or a complete exit strategy." - ) - console.print("Contact: request_report@escapecloud.io") console.print("-------------------------------------------") console.print("Outputs:", style="bold") if upload_result and upload_result.success: diff --git a/cloudexit/cli/parser.py b/cloudexit/cli/parser.py index 03a5459..6a31c7c 100644 --- a/cloudexit/cli/parser.py +++ b/cloudexit/cli/parser.py @@ -58,7 +58,7 @@ def parse_arguments(): help="Increase log verbosity (-v for INFO, -vv for DEBUG + third-party).", ) - # Platform options: flags > CLOUDEXIT_* environment > config file > legacy + # Platform options: flags > CLOUDEXIT_* environment > config file platform = argparse.ArgumentParser(add_help=False) platform.add_argument( "--host", type=str, help="Platform host (https://, or http:// for localhost)." @@ -186,7 +186,7 @@ def parse_arguments(): action="store_true", help=( "Run without prompts; read all inputs from environment variables " - "(for CI use), for example ESC_EXIT_STRATEGY (1 or 3) and " + "(for CI use), for example ESC_EXIT_STRATEGY (1, 2 or 3) and " "ESC_ASSESSMENT_TYPE (1 or 2)." ), ) @@ -246,7 +246,7 @@ def parse_arguments(): action="store_true", help=( "Run without prompts; read all inputs from environment variables " - "(for CI use), for example ESC_EXIT_STRATEGY (1 or 3) and " + "(for CI use), for example ESC_EXIT_STRATEGY (1, 2 or 3) and " "ESC_ASSESSMENT_TYPE (1 or 2)." ), ) diff --git a/cloudexit/cli/prompts.py b/cloudexit/cli/prompts.py index 0aa04c1..8807ff5 100644 --- a/cloudexit/cli/prompts.py +++ b/cloudexit/cli/prompts.py @@ -6,24 +6,32 @@ logger = logging.getLogger(__name__) -def prompt_required_inputs() -> tuple[int, int]: +def prompt_required_inputs(ask_assessment_type: bool = True) -> tuple[int, int]: while True: try: exit_strategy = int( ask( - "Enter Exit Strategy (1 for 'Repatriation to On-Premises', 3 for 'Migration to Alternate Cloud'): " + "Enter Exit Strategy (1 for 'Repatriation to On-Premises', 2 for 'Hybrid Cloud Adoption', 3 for 'Migration to Alternate Cloud'): " ).strip() ) - if exit_strategy not in [1, 3]: + if exit_strategy not in [1, 2, 3]: raise ValueError("Invalid exit strategy.") # logger.info(f"Exit Strategy selected: {exit_strategy}") break except ValueError as e: logger.debug(f"Invalid exit strategy input: {e}") console.print( - f"{e} Please enter 1 or 3.", style="red", markup=False, emoji=False + f"{e} Please enter 1, 2 or 3.", + style="red", + markup=False, + emoji=False, ) + # Standard only exists on a platform, so without an upload there is + # nothing to choose + if not ask_assessment_type: + return exit_strategy, 1 + while True: try: assessment_type = int( diff --git a/cloudexit/log_redaction.py b/cloudexit/log_redaction.py new file mode 100644 index 0000000..8ea184f --- /dev/null +++ b/cloudexit/log_redaction.py @@ -0,0 +1,51 @@ +import logging +import re + +# botocore builds its request signature from the caller's credentials and logs +# the material at DEBUG, so -vv would otherwise copy the access key ID, the +# session token and the signature into the console and into run.log, which is a +# file users attach to bug reports. Redacting the values keeps the surrounding +# request detail readable. +REDACTED = "REDACTED" + +_SECRET_KEYS = ( + "authorization", + "x-amz-security-token", + "aws_secret_access_key", + "aws_session_token", + "client_secret", +) + +_RULES = [ + ( + re.compile( + rf"(?i)({re.escape(key)}['\"]?\s*[:=]\s*)(b?['\"][^'\"]*['\"]|[^,\r\n]*)" + ), + rf"\1{REDACTED}", + ) + for key in _SECRET_KEYS +] + [ + (re.compile(r"(?i)(credential=)[^/\s,]+(/)"), rf"\1{REDACTED}\2"), + (re.compile(r"(?i)(signature\s*[=:]\s*)[^\s,&'\"]+"), rf"\1{REDACTED}"), +] + + +def redact(text: str) -> str: + for pattern, replacement in _RULES: + text = pattern.sub(replacement, text) + return text + + +class SecretRedactingFilter(logging.Filter): + def filter(self, record: logging.LogRecord) -> bool: + # Redact the formatted text, because the secret often sits in an object + # passed as a %s argument rather than in the format string itself. + try: + message = record.getMessage() + except Exception: + return True + cleaned = redact(message) + if cleaned != message: + record.msg = cleaned + record.args = () + return True diff --git a/cloudexit/report/egress.py b/cloudexit/report/egress.py index 099eacd..0c2bf12 100644 --- a/cloudexit/report/egress.py +++ b/cloudexit/report/egress.py @@ -16,8 +16,6 @@ from cloudexit.providers.aws.egress import ARCHIVE_TIERS as AWS_ARCHIVE_TIERS from cloudexit.providers.azure.egress import ARCHIVE_TIERS as AZURE_ARCHIVE_TIERS from cloudexit.report.context import PROVIDER_NAMES -from cloudexit.report.notices import text_parts -from cloudexit.report.sections import BOOK_DEMO_URL, PRODUCT_SITES from cloudexit.report.wording import BLOCKS logger = logging.getLogger(__name__) @@ -199,9 +197,11 @@ def free_tier_headroom( def _summarize_totals( totals: dict[str, Any], + rows: list[dict[str, Any]], total_fee: float, prices: list[dict[str, Any]], cloud_service_provider: int, + b: dict[str, Any] = BLOCKS, ) -> dict[str, Any]: free_tier_limit = free_tier_limit_display(prices, cloud_service_provider) free_tier = ( @@ -210,9 +210,16 @@ def _summarize_totals( and free_tier_limit is not None ) unknown_count = totals["resources_with_unknown_size"] + # Nothing was measured, so there is no lower bound to show: "0.0 GiB+" would + # claim one of zero. A resource counts as sized when it has a size, the same + # way the "n of m sized" strip counts it. + nothing_sized = bool(rows) and all(row["size_bytes"] is None for row in rows) total_fee_display = format_fee(total_fee) total_data_display = format_gib(totals["known_size_bytes"]) - if unknown_count > 0: + if nothing_sized: + total_fee_display = b["ev_dash"] + total_data_display = b["ev_dash"] + elif unknown_count > 0: total_data_display = f"{total_data_display}+" if not free_tier: total_fee_display = f"{total_fee_display}+" @@ -220,6 +227,7 @@ def _summarize_totals( "free_tier": free_tier, "free_tier_limit": free_tier_limit, "unknown_count": unknown_count, + "nothing_sized": nothing_sized, "total_fee_display": total_fee_display, "total_data_display": total_data_display, } @@ -330,7 +338,7 @@ def load_egress_estimate(report_path: str, json_path: str) -> dict[str, Any]: "total_fee": total_fee, "type_groups": type_groups, "summary": _summarize_totals( - totals, total_fee, pricing, meta["cloud_service_provider"] + totals, rows, total_fee, pricing, meta["cloud_service_provider"] ), } @@ -339,9 +347,12 @@ def _share(part, whole): return part / whole * 100 if whole else 0 -def _plus(text, estimate): +def _plus(text, estimate, b=BLOCKS): # Resources without a size are left out, so the figure is a lower bound - return f"{text}+" if estimate["summary"]["unknown_count"] > 0 else text + summary = estimate["summary"] + if summary["nothing_sized"]: + return b["ev_dash"] + return f"{text}+" if summary["unknown_count"] > 0 else text def introduction_facts(scope, b=BLOCKS): @@ -540,6 +551,8 @@ def cost_facts(estimate, b=BLOCKS): if summary["free_tier"]: volume_stat = (b["eg_st_allowance"], summary["free_tier_limit"], False) + elif summary["nothing_sized"]: + volume_stat = (b["eg_st_billable"], b["ev_dash"], False) else: volume_stat = (b["eg_st_billable"], format_gib(billable), False) return { @@ -685,27 +698,3 @@ def pricing_facts(estimate, b=BLOCKS): ) ], } - - -def next_steps_facts( - executive_id, uploaded=False, platform_url=None, assessment_id=None, b=BLOCKS -): - # Paragraphs as text and (text, url) links, like the notices - sites = dict(PRODUCT_SITES) - paragraphs = [text_parts(b["eg_next_lead"], {}, {"executive_id": executive_id})] - if uploaded: - paragraphs.append([b["eg_next_uploaded"]]) - if platform_url: - paragraphs.append(text_parts(b["fa_uploaded_link"], {"link": platform_url})) - elif assessment_id: - paragraphs.append( - text_parts(b["fa_uploaded_id"], {}, {"assessment_id": assessment_id}) - ) - else: - paragraphs.append( - text_parts( - b["eg_next_platforms"], - {"exitcloud_url": sites["exitcloud"], "book_demo_url": BOOK_DEMO_URL}, - ) - ) - return {"heading": b["eg_heading_next_steps"], "paragraphs": paragraphs} diff --git a/cloudexit/report/egress_report.py b/cloudexit/report/egress_report.py index 8e108f6..a72abf0 100644 --- a/cloudexit/report/egress_report.py +++ b/cloudexit/report/egress_report.py @@ -15,7 +15,6 @@ data_landscape, estimated_costs, introduction, - next_steps, ) from .evidence import build_document_information_page from .flow import chapter_break, section_break @@ -41,8 +40,6 @@ def build_egress_content(context, estimate, styles, b): content += estimated_costs(context, estimate, styles, b) content.append(chapter_break()) content += appendix(context, estimate, styles, b) - content.append(Spacer(1, 12)) - content += next_steps(context, styles, b) content.append(chapter_break()) content += legal_notices(context, styles, b, egress=True) return content diff --git a/cloudexit/report/egress_sections.py b/cloudexit/report/egress_sections.py index 553a46a..9a01519 100644 --- a/cloudexit/report/egress_sections.py +++ b/cloudexit/report/egress_sections.py @@ -12,13 +12,11 @@ cost_facts, introduction_facts, landscape_facts, - next_steps_facts, pricing_facts, ) from .evidence import link_markup from .flow import section -from .notices import parts_markup -from .palette import HAIRLINE, HOTSPOT_COUNT, INK, MUTED, PROVENANCE_BACKGROUND, RULE +from .palette import HAIRLINE, HOTSPOT_COUNT, INK, MUTED from .placeholder import LabelWithPlaceholder from .sections import GLOSSARY_TERM_WIDTH from .table_style import data_table, header_cell, share_bar, summary_strip, text_style @@ -285,40 +283,3 @@ def appendix(context, estimate, styles, b): _glossary_table(facts["glossary"], context.appearance), ] return content - - -def next_steps(context, styles, b): - provenance = context.provenance - facts = next_steps_facts( - context.document_id, - provenance.uploaded, - provenance.platform_url, - provenance.assessment_id, - b, - ) - color = context.appearance.heading2_color - box = Table( - [ - [ - [ - Paragraph(parts_markup(parts, color), styles["ContentStyle"]) - for parts in facts["paragraphs"] - ] - ] - ], - colWidths=[16 * cm], - hAlign="LEFT", - ) - box.setStyle( - TableStyle( - [ - ("BACKGROUND", (0, 0), (-1, -1), HexColor(PROVENANCE_BACKGROUND)), - ("BOX", (0, 0), (-1, -1), 0.6, HexColor(RULE)), - ("LEFTPADDING", (0, 0), (-1, -1), 10), - ("RIGHTPADDING", (0, 0), (-1, -1), 10), - ("TOPPADDING", (0, 0), (-1, -1), 4), - ("BOTTOMPADDING", (0, 0), (-1, -1), 4), - ] - ) - ) - return [_heading(facts, styles), Spacer(1, 4), box] diff --git a/cloudexit/report/html_egress.py b/cloudexit/report/html_egress.py index fa19b09..5048995 100644 --- a/cloudexit/report/html_egress.py +++ b/cloudexit/report/html_egress.py @@ -40,8 +40,13 @@ def _kpis(estimate, sized, unsized, groups, b): summary = estimate["summary"] csp = estimate["meta"]["cloud_service_provider"] total = estimate["totals"]["known_size_bytes"] - headroom = free_tier_headroom(estimate["pricing"], csp, total) + nothing_sized = summary["nothing_sized"] + # Nothing was measured: no volume to show, and no headroom to claim against + headroom = ( + None if nothing_sized else free_tier_headroom(estimate["pricing"], csp, total) + ) plus = "+" if summary["unknown_count"] > 0 else "" + volume = b["ev_dash"] if nothing_sized else gib_number(total) kpis = [ { @@ -55,8 +60,8 @@ def _kpis(estimate, sized, unsized, groups, b): }, { "label": b["eg_st_volume"], - "figure": gib_number(total), - "unit": f"GiB{plus}", + "figure": volume, + "unit": "" if nothing_sized else f"GiB{plus}", "caption": ( b["html_eg_types_sized"].format(sized=len(sized), types=len(groups)) if groups @@ -83,9 +88,11 @@ def _kpis(estimate, sized, unsized, groups, b): kpis.append( { "label": b["eg_st_billable"], - "figure": gib_number(total), - "unit": "GiB", - "caption": b["html_eg_billable_caption"], + "figure": volume, + "unit": "" if nothing_sized else "GiB", + # With nothing sized the prices were never applied, so the + # caption would claim something the run did not establish + "caption": "" if nothing_sized else b["html_eg_billable_caption"], } ) kpis.append( @@ -142,7 +149,8 @@ def _volume(sized, unsized, total, b): def _table(context, estimate, sized, unsized, total, b): - plus = "+" if estimate["summary"]["unknown_count"] > 0 else "" + summary = estimate["summary"] + plus = "+" if summary["unknown_count"] > 0 else "" def row(group): icon = group["icon"].lstrip("/") if group["icon"] else "" @@ -177,8 +185,16 @@ def row(group): "not_sized": b["html_eg_not_sized"], "total": ( b["eg_total"], - f"{gib_number(total, 2)}{plus}", - fee_number(estimate["total_fee"]), + ( + b["ev_dash"] + if summary["nothing_sized"] + else f"{gib_number(total, 2)}{plus}" + ), + ( + b["ev_dash"] + if summary["nothing_sized"] + else fee_number(estimate["total_fee"]) + ), ), } diff --git a/cloudexit/report/sections.py b/cloudexit/report/sections.py index 5df8c54..56c155d 100644 --- a/cloudexit/report/sections.py +++ b/cloudexit/report/sections.py @@ -61,7 +61,6 @@ ("escapecloud", "https://escapecloud.io"), ("exitcloud", "https://exitcloud.io"), ) -BOOK_DEMO_URL = "https://escapecloud.io/book-demo/" AWS_ICONS_URL = "https://aws.amazon.com/architecture/icons/" AWS_TRADEMARK_URL = "https://aws.amazon.com/trademark-guidelines/" AZURE_ICONS_URL = "https://learn.microsoft.com/en-us/azure/architecture/icons/" diff --git a/cloudexit/report/wording.py b/cloudexit/report/wording.py index d54befd..d9db971 100644 --- a/cloudexit/report/wording.py +++ b/cloudexit/report/wording.py @@ -357,7 +357,6 @@ "definition": "One completed assessment of the defined scope, identified by the run ID on page 2.", }, ], - "fa_uploaded_link": "Open the reports: {link}", "fa_uploaded_id": "Assessment ID: {assessment_id}", "eg_heading_introduction": "Introduction", "eg_heading_purpose": "Purpose", @@ -576,19 +575,6 @@ "definition": "Fees a provider charges for moving data or workloads to another provider, including data egress.", }, ], - "eg_heading_next_steps": "Next Steps", - "eg_next_lead": ( - "Use this estimate together with the Executive Summary of the same " - "run ({executive_id}) when planning an exit." - ), - "eg_next_platforms": ( - "Connected ({exitcloud_url}) and Enterprise ({book_demo_url}) combine " - "this estimate with the Exit Readiness Report and version history." - ), - "eg_next_uploaded": ( - "This run was uploaded. The platform combines this estimate with the " - "Exit Readiness Report and version history." - ), "notices_egress": ( "Egress figures are estimates based on the provider's public list " "prices at the time of the run. Actual invoices can differ: " @@ -1046,7 +1032,6 @@ "eg_heading_costs", "eg_heading_pricing", "eg_heading_glossary", - "eg_heading_next_steps", "eg_th_resource_type", "eg_th_measured", "eg_th_resources", diff --git a/cloudexit/settings.py b/cloudexit/settings.py index 7af05e7..d1a0d99 100644 --- a/cloudexit/settings.py +++ b/cloudexit/settings.py @@ -12,11 +12,6 @@ logger = logging.getLogger(__name__) -try: - import config as legacy_config -except ModuleNotFoundError: - legacy_config = None - DEFAULT_TIMEOUT = 10.0 DEFAULT_UPLOAD_BUDGET = 300.0 LOCAL_HTTP_HOSTS = ("localhost", "127.0.0.1") @@ -51,6 +46,16 @@ FILE_KEYS = ("host", "api_key", "api_key_file", "assessment_id", "ca_bundle", "upload") PATH_KEYS = ("api_key_file", "ca_bundle", "spool_dir") +LEGACY_ENV_WARNING = ( + "HOST and KEY are no longer read (removed in 2.0). Use CLOUDEXIT_HOST and " + "CLOUDEXIT_API_KEY_FILE (or CLOUDEXIT_API_KEY)." +) +LEGACY_CONFIG_WARNING = ( + "config.py is no longer read (removed in 2.0). Move HOST and KEY to the " + "config file (see `cloudexit config path`) or to CLOUDEXIT_* environment " + "variables." +) + TRUE_WORDS = ("1", "true", "yes", "on") FALSE_WORDS = ("0", "false", "no", "off") @@ -196,28 +201,6 @@ def _env_layer(environ: Mapping[str, str]) -> dict[str, Setting]: return layer -def _legacy_env_layer(environ: Mapping[str, str]) -> dict[str, Setting]: - host = environ.get("HOST", "").strip() - key = environ.get("KEY", "").strip() - if host and key: - return { - "host": Setting(host, "legacy env HOST"), - "api_key": Setting(key, "legacy env KEY"), - } - if host or key: - logger.debug("Ignoring legacy HOST/KEY: they are only used when both are set.") - return {} - - -def _legacy_config_layer() -> dict[str, Setting]: - layer = {} - for name, attr in (("host", "HOST"), ("api_key", "KEY")): - value = getattr(legacy_config, attr, "") - if isinstance(value, str) and value.strip(): - layer[name] = Setting(value.strip(), "legacy config.py") - return layer - - def _read_config_file(path: Path) -> dict[str, Any] | None: if not path.is_file(): return None @@ -364,22 +347,20 @@ def _check_ca_bundle(setting: Setting) -> Setting: return setting -def _warn_deprecated(settings: Settings) -> None: - sources = { - setting.source - for setting in (settings.host, settings.api_key) - if setting.source.startswith("legacy") - } - if "legacy env HOST" in sources or "legacy env KEY" in sources: - logger.warning( - "The HOST and KEY environment variables are deprecated; use " - "CLOUDEXIT_HOST and CLOUDEXIT_API_KEY_FILE (or CLOUDEXIT_API_KEY)." - ) - if "legacy config.py" in sources: - logger.warning( - "config.py is deprecated; move HOST and KEY to the config file " - "(see `cloudexit config path`) or to CLOUDEXIT_* environment variables." - ) +def _warn_legacy_sources(settings: Settings, environ: Mapping[str, str]) -> None: + # Only for users still relying on the sources 2.0 dropped, so it stays quiet + # once anything else supplies a host or a key. Presence is all that is + # checked: config.py is never imported, opened or parsed, and the values of + # HOST and KEY are never logged or used. + if settings.host.value or settings.api_key.value: + return + notes = [] + if environ.get("HOST", "").strip() and environ.get("KEY", "").strip(): + notes.append(LEGACY_ENV_WARNING) + if Path("config.py").is_file(): + notes.append(LEGACY_CONFIG_WARNING) + if notes: + logger.warning(" ".join(notes)) def load_settings( @@ -404,12 +385,7 @@ def load_settings( else: file_layers = _file_layers(data, path, profile.value) - layers = [ - *top, - *file_layers, - _legacy_env_layer(environ), - _legacy_config_layer(), - ] + layers = [*top, *file_layers] api_key, api_key_file = _resolve_key(layers) host = _first(layers, "host") @@ -441,7 +417,7 @@ def load_settings( config_path=path, config_found=data is not None, ) - _warn_deprecated(settings) + _warn_legacy_sources(settings, environ) return settings diff --git a/cloudexit/terminal.py b/cloudexit/terminal.py index 60cf2f3..101b84e 100644 --- a/cloudexit/terminal.py +++ b/cloudexit/terminal.py @@ -2,9 +2,7 @@ import getpass import sys from rich.console import Console -from rich.markup import escape from rich.style import Style -from time import sleep from cloudexit import codes @@ -45,14 +43,8 @@ def print_step( "pending": "[ ... ]", } - # Handle the pending status with a spinner if status == "pending": - with console.status( - f"{escape(description):<50} [yellow]{status_map['pending']}[/yellow]", - spinner="dots", - ): - sleep(2) - print_step(description, status="ok") + print_step(description, status="ok") elif status == "ok": console.print( f"{description:<50} {status_map['ok']}", diff --git a/config.example.toml b/config.example.toml index 6956c69..d9c1aef 100644 --- a/config.example.toml +++ b/config.example.toml @@ -9,8 +9,6 @@ # 1. command-line flags (--host, --api-key-file, --config-profile, ...) # 2. CLOUDEXIT_* environment variables # 3. this file ([profiles.] first, then [default]) -# 4. deprecated: the HOST and KEY environment variables (only when both are -# set) and config.py # Run `cloudexit config show` to see each value and where it came from. [default] diff --git a/tests/cli/test_app.py b/tests/cli/test_app.py index 0af07c8..5b12a55 100644 --- a/tests/cli/test_app.py +++ b/tests/cli/test_app.py @@ -510,14 +510,28 @@ def test_missing_exit_strategy_exits_config(self): self.assertEqual(ctx.exception.code, codes.CONFIG) def test_invalid_exit_strategy_exits_config(self): - env = {**self._BASE_ENV, "ESC_EXIT_STRATEGY": "9"} + for value in ("4", "9"): + with self.subTest(value=value): + env = {**self._BASE_ENV, "ESC_EXIT_STRATEGY": value} + with ( + patch.dict(os.environ, env, clear=False), + patch("cloudexit.cli.app.console.print"), + ): + with self.assertRaises(SystemExit) as ctx: + app.handle_aws(_ni_aws_args()) + self.assertEqual(ctx.exception.code, codes.CONFIG) + + def test_the_hybrid_exit_strategy_reaches_the_assessment(self): + env = {**self._BASE_ENV, "ESC_EXIT_STRATEGY": "2"} with ( patch.dict(os.environ, env, clear=False), + patch("cloudexit.cli.app.validate_region"), + patch("cloudexit.cli.app.run_assessment") as mock_run, patch("cloudexit.cli.app.console.print"), ): - with self.assertRaises(SystemExit) as ctx: - app.handle_aws(_ni_aws_args()) - self.assertEqual(ctx.exception.code, codes.CONFIG) + app.handle_aws(_ni_aws_args()) + + self.assertEqual(mock_run.call_args[0][0]["exitStrategy"], 2) def test_missing_aws_region_exits_config(self): env = {k: v for k, v in self._BASE_ENV.items() if k != "AWS_DEFAULT_REGION"} @@ -560,6 +574,18 @@ def test_builds_config_from_env_and_calls_run_assessment(self): self.assertEqual(config_arg["providerDetails"]["subscriptionId"], "sub-id-789") self.assertEqual(config_arg["providerDetails"]["resourceGroupName"], "my-rg") + def test_the_hybrid_exit_strategy_reaches_the_assessment(self): + env = {**self._BASE_ENV, "ESC_EXIT_STRATEGY": "2"} + with ( + patch.dict(os.environ, env, clear=False), + patch("azure.identity.ClientSecretCredential", return_value=MagicMock()), + patch("cloudexit.cli.app.run_assessment") as mock_run, + patch("cloudexit.cli.app.console.print"), + ): + app.handle_azure(_ni_azure_args()) + + self.assertEqual(mock_run.call_args[0][0]["exitStrategy"], 2) + def test_uses_default_credential_when_client_secret_missing(self): env = {k: v for k, v in self._BASE_ENV.items() if k != "AZURE_CLIENT_SECRET"} mock_credential = MagicMock() diff --git a/tests/cli/test_prompts.py b/tests/cli/test_prompts.py index 62c293d..bfde679 100644 --- a/tests/cli/test_prompts.py +++ b/tests/cli/test_prompts.py @@ -212,3 +212,46 @@ def test_the_stdin_helper_works_with_an_in_memory_stream(self): if __name__ == "__main__": unittest.main() + + +class AssessmentTypePromptTests(unittest.TestCase): + def test_the_type_is_asked_when_an_upload_will_happen(self): + with patch("cloudexit.cli.prompts.ask", side_effect=["1", "2"]) as asked: + self.assertEqual(prompt_required_inputs(True), (1, 2)) + + self.assertEqual(len(asked.call_args_list), 2) + self.assertIn("Assessment Type", str(asked.call_args_list)) + + def test_without_an_upload_the_type_is_basic_and_never_asked(self): + with patch("cloudexit.cli.prompts.ask", side_effect=["3"]) as asked: + self.assertEqual(prompt_required_inputs(False), (3, 1)) + + self.assertEqual(len(asked.call_args_list), 1) + self.assertNotIn("Assessment Type", str(asked.call_args_list)) + + +class ExitStrategyPromptTests(unittest.TestCase): + def test_the_hybrid_strategy_can_be_chosen(self): + with patch("cloudexit.cli.prompts.ask", side_effect=["2", "2"]): + self.assertEqual(prompt_required_inputs(True), (2, 2)) + + def test_the_prompt_names_all_three_strategies(self): + with patch("cloudexit.cli.prompts.ask", side_effect=["1"]) as asked: + prompt_required_inputs(False) + + question = asked.call_args_list[0][0][0] + for label in ( + "1 for 'Repatriation to On-Premises'", + "2 for 'Hybrid Cloud Adoption'", + "3 for 'Migration to Alternate Cloud'", + ): + self.assertIn(label, question) + + def test_an_out_of_range_strategy_is_refused_and_asked_again(self): + with ( + patch("cloudexit.cli.prompts.ask", side_effect=["4", "2"]), + patch("cloudexit.cli.prompts.console.print") as printed, + ): + self.assertEqual(prompt_required_inputs(False), (2, 1)) + + self.assertIn("Please enter 1, 2 or 3.", str(printed.call_args_list)) diff --git a/tests/cli/test_stub_platform.py b/tests/cli/test_stub_platform.py index 5751941..8bf096a 100644 --- a/tests/cli/test_stub_platform.py +++ b/tests/cli/test_stub_platform.py @@ -193,7 +193,7 @@ def test_off_sends_nothing_even_with_a_key(self): self.assertEqual(platform.requests, []) self.assertEqual(summary["upload"]["status"], "skipped") self.assertEqual(summary["upload"]["reason"], "off") - self.assertIn("Forcing Basic Assessment: uploads are off.", result.stderr) + self.assertIn("Basic assessment: Standard needs an upload.", result.stderr) self.assertEqual(len(self.reports()), 1) def test_the_mode_can_come_from_the_environment(self): @@ -222,7 +222,7 @@ def test_auto_without_a_platform_runs_offline(self): self.assertEqual(summary["upload"]["status"], "skipped") self.assertEqual(summary["upload"]["reason"], "offline") self.assertIn( - "Forcing Basic Assessment: no platform is configured.", + "Basic assessment: Standard needs an upload.", " ".join(result.stderr.split()), ) diff --git a/tests/cli/test_upload_modes.py b/tests/cli/test_upload_modes.py index aa3d4f7..dacf7dd 100644 --- a/tests/cli/test_upload_modes.py +++ b/tests/cli/test_upload_modes.py @@ -2,6 +2,7 @@ import json import sys import unittest +from argparse import Namespace from contextlib import redirect_stderr from pathlib import Path from unittest.mock import patch @@ -24,7 +25,8 @@ ) from tests.report import test_report_flow as flow from cloudexit import codes -from cloudexit.settings import Setting, Settings +from cloudexit.settings import Setting, Settings, load_settings +from tests.support.isolation import isolated_environ OUTDATED = { "code": "collector_catalogue_outdated", @@ -51,7 +53,7 @@ def messages(self): return [args[0] for args, _ in self.steps] def forced(self): - return [m for m in self.messages() if m.startswith("Forcing Basic")] + return [m for m in self.messages() if m.startswith("Basic assessment")] def summary(self, exit_code=0): out = io.StringIO() @@ -67,9 +69,28 @@ def test_auto_without_a_platform_runs_offline_and_succeeds(self): mock_upload.assert_not_called() self.assert_local_report(report) - self.assertEqual( - self.forced(), ["Forcing Basic Assessment: no platform is configured."] - ) + self.assertEqual(self.forced(), ["Basic assessment: Standard needs an upload."]) + + def test_the_legacy_host_and_key_environment_variables_upload_nothing(self): + report = self.report() + environ = { + **isolated_environ(self.tmp), + "HOST": "platform.test", + "KEY": "fixture-key-0123456789", + } + with self.assertLogs("cloudexit.settings", level="WARNING") as logs: + settings = load_settings(None, environ=environ) + + self.assertIsNone(settings.host.value) + self.assertIsNone(settings.api_key.value) + self.assertNotIn("fixture-key-0123456789", logs.output[0]) + + with patch("cloudexit.cli.app.upload") as mock_upload: + self.run_assessment(report, settings=settings, assessment_type=2) + + mock_upload.assert_not_called() + self.assert_local_report(report) + self.assertEqual(self.sync(report)["assessment_type"], 1) def test_auto_with_a_platform_uploads(self): report = self.report() @@ -129,7 +150,7 @@ def test_off_with_a_platform_makes_no_request_and_forces_basic(self): mock_upload.assert_not_called() self.assert_local_report(report) self.assertEqual(self.sync(report)["assessment_type"], 1) - self.assertEqual(self.forced(), ["Forcing Basic Assessment: uploads are off."]) + self.assertEqual(self.forced(), ["Basic assessment: Standard needs an upload."]) self.assertFalse(self.raw(report, "upload.json").exists()) def test_off_does_not_even_check_a_key_that_cannot_be_right(self): @@ -586,3 +607,91 @@ def dispatch(args): if __name__ == "__main__": unittest.main() + + +class TerminalOutputTests(ModeTestCase): + # Nothing in the terminal sells an edition, a contact or a framework + MARKETING = ( + "regulatory-aligned", + "DORA", + "FINMA", + "PRA", + "request_report", + "ExitCloud", + "Light", + "Enterprise", + ) + + def steps_starting(self, prefix): + return [(a, k) for a, k in self.steps if a and a[0].startswith(prefix)] + + def test_an_offline_run_says_what_it_did_and_sells_nothing(self): + report = self.report() + self.run_assessment(report, settings=Settings(), real_terminal=True) + text = self.output.getvalue() + + self.assertIn("Offline run: no platform configured", text) + self.assertIn("Outputs:", text) + for banned in self.MARKETING: + with self.subTest(banned): + self.assertNotIn(banned, text) + + def test_an_offline_run_is_not_a_warning(self): + report = self.report() + self.run_assessment(report, settings=Settings(), assessment_type=1) + + offline = self.steps_starting("Offline run:") + self.assertEqual(len(offline), 1) + self.assertEqual(offline[0][1]["status"], "ok") + self.assertEqual( + [k for _a, k in self.steps if k.get("status") == "warning"], [] + ) + + def test_an_explicit_standard_type_offline_is_reduced_without_a_warning(self): + report = self.report() + self.run_assessment(report, settings=Settings(), assessment_type=2) + + reduced = self.steps_starting("Basic assessment") + self.assertEqual(len(reduced), 1) + self.assertEqual( + reduced[0][0][0], "Basic assessment: Standard needs an upload." + ) + self.assertEqual(reduced[0][1]["status"], "ok") + + def test_a_configured_run_names_the_platform_by_host(self): + report = self.report() + self.run_assessment(report, settings=flow.CONFIGURED, replies=success_replies()) + + self.assertEqual( + [a[0] for a, _k in self.steps_starting("Platform configured")], + ["Platform configured: https://platform.test."], + ) + + def test_the_host_line_carries_no_path_query_or_credentials(self): + for raw, shown in ( + ("https://platform.test", "https://platform.test"), + ("https://platform.test/api/v2?token=abc", "https://platform.test"), + ("https://user:pw@platform.test:8443/x", "https://platform.test:8443"), + ("http://127.0.0.1:8000/path", "http://127.0.0.1:8000"), + ): + with self.subTest(raw): + self.assertEqual(app._host_display(raw), shown) + + +class AssessmentTypeQuestionTests(unittest.TestCase): + def args(self, settings, dry_run=False): + return Namespace(settings=settings, dry_run=dry_run) + + def test_a_configured_platform_that_will_upload_gets_the_question(self): + self.assertTrue(app._asks_assessment_type(self.args(flow.CONFIGURED))) + + def test_an_offline_run_does_not_get_the_question(self): + for settings in (None, Settings(), Settings(host=flow.CONFIGURED.host)): + with self.subTest(settings=settings): + self.assertFalse(app._asks_assessment_type(self.args(settings))) + + def test_upload_off_or_a_dry_run_does_not_get_the_question(self): + self.assertFalse(app._asks_assessment_type(self.args(with_mode("off")))) + self.assertFalse( + app._asks_assessment_type(self.args(flow.CONFIGURED, dry_run=True)) + ) diff --git a/tests/report/test_basic_report.py b/tests/report/test_basic_report.py index 64d3e4d..b56f3c9 100644 --- a/tests/report/test_basic_report.py +++ b/tests/report/test_basic_report.py @@ -998,7 +998,6 @@ def test_the_egress_labels_are_in_the_title_case_list(self): "eg_heading_landscape", "eg_heading_costs", "eg_heading_pricing", - "eg_heading_next_steps", "eg_th_size_gib", "eg_st_largest", ): @@ -1048,7 +1047,7 @@ def test_no_wording_carries_regulatory_words_or_scores(self): for key, text in self.strings(): # The pages that point to the platforms name what they add, and the # Scores row of the HTML report names the results they compute - if key.startswith(("fa_", "eg_next_", "html_scores_", "html_tour_")): + if key.startswith(("fa_", "html_scores_", "html_tour_")): continue with self.subTest(key): self.assertEqual(find_terms(text), []) diff --git a/tests/report/test_egress.py b/tests/report/test_egress.py index 3c9b3d6..ecd11ae 100644 --- a/tests/report/test_egress.py +++ b/tests/report/test_egress.py @@ -6,11 +6,13 @@ from unittest.mock import patch -from cloudexit.egress.model import GIB, write_egress_inventory +from cloudexit.egress.model import GIB, compute_totals, write_egress_inventory from cloudexit.report.helpers import format_gib, gib_number, percent from cloudexit.catalogue.store import active_catalogue_path +from cloudexit.report.wording import BLOCKS from cloudexit.report.egress import ( _build_type_groups, + _summarize_totals, _resolve_icon, build_fee_estimate, calculate_tiered_cost, @@ -457,3 +459,55 @@ def test_no_free_tier_means_no_headroom(self): if __name__ == "__main__": unittest.main() + + +class SummaryDisplayTests(unittest.TestCase): + DASH = BLOCKS["ev_dash"] + + def rows(self, *sizes, flagged=True): + return [ + { + "size_bytes": size, + "size_unknown": size is None and flagged, + "tier_bytes": None, + } + for size in sizes + ] + + def summary(self, rows, total_fee=0.0): + return _summarize_totals(compute_totals(rows, set()), rows, total_fee, [], 1) + + def test_nothing_sized_shows_a_dash_rather_than_a_zero_with_a_plus(self): + summary = self.summary(self.rows(None, None, None)) + + self.assertTrue(summary["nothing_sized"]) + self.assertEqual(summary["total_data_display"], self.DASH) + self.assertEqual(summary["total_fee_display"], self.DASH) + + def test_a_row_with_no_size_and_no_unknown_flag_is_still_not_sized(self): + # One resource type of the egress_unsized sample is shaped this way + rows = self.rows(None, None) + self.rows(None, flagged=False) + summary = self.summary(rows) + + self.assertTrue(summary["nothing_sized"]) + self.assertEqual(summary["total_data_display"], self.DASH) + + def test_partial_sizing_keeps_the_lower_bound_plus(self): + summary = self.summary(self.rows(10 * GIB, None), total_fee=1.5) + + self.assertFalse(summary["nothing_sized"]) + self.assertEqual(summary["total_data_display"], "10.0 GiB+") + self.assertEqual(summary["total_fee_display"], "$1.50+") + + def test_sized_at_exactly_zero_bytes_is_a_figure_not_a_dash(self): + summary = self.summary(self.rows(0)) + + self.assertFalse(summary["nothing_sized"]) + self.assertEqual(summary["total_data_display"], "0.0 GiB") + self.assertEqual(summary["total_fee_display"], "$0.00") + + def test_without_any_data_bearing_resource_nothing_is_claimed_either_way(self): + summary = self.summary([]) + + self.assertFalse(summary["nothing_sized"]) + self.assertEqual(summary["total_data_display"], "0.0 GiB") diff --git a/tests/report/test_egress_report.py b/tests/report/test_egress_report.py index 85e1087..884edf4 100644 --- a/tests/report/test_egress_report.py +++ b/tests/report/test_egress_report.py @@ -53,7 +53,6 @@ (0, "Estimated Costs"), (0, "Appendix – Egress Prices"), (1, "Glossary"), - (0, "Next Steps"), (0, "Legal Notices"), ] MULTI_TIB = [ @@ -326,15 +325,28 @@ def test_the_appendix_names_the_prices_the_sources_and_the_units(self): ): self.assertIn(expected, text) - def test_next_steps_name_the_companion_and_the_platforms(self): - text = self.text(MULTI_TIB, prov=provenance()) - self.assertIn(f"Executive Summary of the same run ({ES_ID})", text) - self.assertIn("Connected (https://exitcloud.io)", text) - self.assertIn("Enterprise (https://escapecloud.io/book-demo/)", text) - - text = self.text(MULTI_TIB, prov=uploaded()) - self.assertIn(f"Open the reports: {URL}", text) - self.assertNotIn("Connected (https://exitcloud.io)", text) + def test_nothing_promotes_a_platform_or_an_edition(self): + for name, prov in (("offline", provenance()), ("uploaded", uploaded())): + with self.subTest(name): + text = self.text(MULTI_TIB, prov=prov) + for banned in ( + "Next Steps", + "exitcloud.io", + "book-demo", + "book a demo", + "Connected", + "Enterprise", + ): + self.assertNotIn(banned, text) + + def test_page_two_still_carries_the_companion_and_the_platform_link(self): + _path, _context, reader = self.render(MULTI_TIB, prov=provenance()) + self.assertIn(f"Companion Document {ES_ID}", page_texts(reader)[1]) + + _path, _context, reader = self.render(MULTI_TIB, prov=uploaded()) + page_two = page_texts(reader)[1] + self.assertIn(f"Companion Document {ES_ID}", page_two) + self.assertIn(URL, page_two) def test_no_date_is_written_month_first(self): for name, provider, prov in self.scenarios(): diff --git a/tests/report/test_report_flow.py b/tests/report/test_report_flow.py index 3e799a6..32ea635 100644 --- a/tests/report/test_report_flow.py +++ b/tests/report/test_report_flow.py @@ -526,7 +526,7 @@ def test_an_offline_run_forces_basic_and_never_uploads(self): LOCAL_RISKS, ) messages = [args[0] for args, _ in self.steps] - self.assertIn("Forcing Basic Assessment: no platform is configured.", messages) + self.assertIn("Basic assessment: Standard needs an upload.", messages) def test_the_local_results_are_computed_when_connected_too(self): report = self.report() @@ -573,7 +573,7 @@ def test_connected_runs_keep_the_requested_depth_in_the_document(self): # The local report is always the Basic report self.assertEqual(self.json_report(report)["meta"]["assessment_type"], 1) messages = [args[0] for args, _ in self.steps] - self.assertEqual([m for m in messages if m.startswith("Forcing Basic")], []) + self.assertEqual([m for m in messages if m.startswith("Basic assessment")], []) def test_platform_text_with_brackets_is_shown_literally_and_the_run_goes_on(self): message = "rejected [/bold] [red]x[/red] :warning: [1, 2]" @@ -770,7 +770,7 @@ def test_a_key_that_cannot_be_right_is_a_failed_upload_not_an_offline_run(self): # A configured platform keeps the requested depth: nothing forces Basic self.assertEqual(self.sync(report)["assessment_type"], 2) messages = [args[0] for args, _ in self.steps] - self.assertEqual([m for m in messages if m.startswith("Forcing Basic")], []) + self.assertEqual([m for m in messages if m.startswith("Basic assessment")], []) def test_basic_is_forced_only_without_platform_configuration(self): for settings in (None, Settings()): diff --git a/tests/samples/test_samples.py b/tests/samples/test_samples.py index 26524b6..a8f459f 100644 --- a/tests/samples/test_samples.py +++ b/tests/samples/test_samples.py @@ -342,17 +342,41 @@ def test_a_state_file_has_no_costs_and_an_unknown_region(self): def test_unsized_egress_is_not_sized_on_every_type(self): text = " ".join(self.pdf_pages("egress_unsized", "egress.pdf")) dump = self.text("egress_unsized", "texts/index-html.txt") + dash = BLOCKS["ev_dash"] self.assertIn("None of the 5 data-bearing resources could be sized.", text) self.assertIn("RESOURCES SIZED 0 of 5", text) self.assertIn("Not sized: Storage Account (2 resources)", text) self.assertIn("SQL Database (1 resource)", text) - # A fee of zero with a +: a lower bound, never a figure - self.assertIn("$0.00+", text) - self.assertIn("0.0 GiB+", text) + # Nothing was measured, so no figure may claim a lower bound of zero + for strip in ("DATA VOLUME", "ESTIMATED FEE (USD)", "BILLABLE VOLUME"): + self.assertIn(f"{strip} {dash}", text) + for kpi in ("Estimated Transfer Fee", "Data Volume", "Billable Volume"): + self.assertIn(f"{kpi} {dash}", dump) + self.assertIn(f"Total {dash} {dash}", dump) + for banned in ("0.0 GiB+", "$0.00+", "0.00+"): + self.assertNotIn(banned, text) + self.assertNotIn(banned, dump) self.assertNotIn("monthly free tier", text) + self.assertNotIn("Free-Tier Headroom", dump) + self.assertNotIn("no free tier in the prices used", dump) self.assertIn("not sized", dump.lower()) + def test_no_egress_pdf_promotes_a_platform_or_an_edition(self): + for case, scenario in self.cases(): + if scenario.egress is None: + continue + with self.subTest(case): + text = " ".join(self.pdf_pages(case, "egress.pdf")) + for banned in ( + "Next Steps", + "exitcloud.io", + "book-demo", + "book a demo", + "Connected", + ): + self.assertNotIn(banned, text) + def test_an_empty_scope_with_costs_has_empty_states_next_to_real_figures(self): text = " ".join(self.pdf_pages("empty_with_costs", "report.pdf")) dump = self.text("empty_with_costs", "texts/index-html.txt") diff --git a/tests/support/egress.py b/tests/support/egress.py index bb2e327..cf334eb 100644 --- a/tests/support/egress.py +++ b/tests/support/egress.py @@ -43,5 +43,5 @@ def empty_estimate(csp=2): "pricing": pricing, "total_fee": 0.0, "type_groups": [], - "summary": _summarize_totals(totals, 0.0, pricing, csp), + "summary": _summarize_totals(totals, [], 0.0, pricing, csp), } diff --git a/tests/test_layering.py b/tests/test_layering.py index 4ec827c..6199e14 100644 --- a/tests/test_layering.py +++ b/tests/test_layering.py @@ -13,6 +13,7 @@ "version": set(), "resources": set(), "private_files": set(), + "log_redaction": set(), "terminal": {"codes"}, "settings": set(), "catalogue": {"settings"}, diff --git a/tests/test_logging.py b/tests/test_logging.py index cb6dc9f..ea8f6cc 100644 --- a/tests/test_logging.py +++ b/tests/test_logging.py @@ -3,13 +3,66 @@ import logging import tempfile import unittest +from io import StringIO from pathlib import Path +from unittest.mock import patch + +from rich.console import Console from cloudexit.cli import app +from cloudexit.log_redaction import redact from tests.support.paths import ROOT PACKAGE = ROOT / "cloudexit" # Loggers of other libraries that the package sets a level on +THIRD_PARTY = ( + "botocore", + "botocore.auth", + "botocore.credentials", + "botocore.endpoint", + "boto3", + "azure", + "urllib3", + "msal", + "PIL", +) + +ACCESS_KEY = "AKIAIOSFODNN7EXAMPLE" +SIGNATURE = "deadbeefcafe0123456789abcdef0123456789abcdef0123456789abcdef0123" +TOKEN = "FwoGZXIvYXdzEExAMPLESESSIONTOKEN" +SECRET = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY" + + +class PreparedRequest: + # botocore reaches the logger with the request object as a %s argument, so + # the secret is not in the format string + def __repr__(self): + return ( + "" + ) + + +class LoggingStateMixin: + def setUp(self): + root = logging.getLogger() + saved = (root.level, list(root.handlers)) + levels = {name: logging.getLogger(name).level for name in THIRD_PARTY} + + def restore(): + root.setLevel(saved[0]) + root.handlers[:] = saved[1] + for name, level in levels.items(): + logging.getLogger(name).setLevel(level) + + self.addCleanup(restore) + + def level(self, name): + return logging.getLogger(name).getEffectiveLevel() def module_loggers(path): @@ -61,25 +114,7 @@ def test_the_run_log_names_the_module_that_wrote_each_line(self): self.assertNotIn("core.engine", text) -class ThirdPartyLevelTests(unittest.TestCase): - NAMES = ("botocore", "boto3", "azure", "urllib3", "msal", "PIL") - - def setUp(self): - root = logging.getLogger() - saved = (root.level, list(root.handlers)) - levels = {name: logging.getLogger(name).level for name in self.NAMES} - - def restore(): - root.setLevel(saved[0]) - root.handlers[:] = saved[1] - for name, level in levels.items(): - logging.getLogger(name).setLevel(level) - - self.addCleanup(restore) - - def level(self, name): - return logging.getLogger(name).getEffectiveLevel() - +class ThirdPartyLevelTests(LoggingStateMixin, unittest.TestCase): def test_msal_is_never_below_warning_because_it_logs_the_client_id(self): for verbose in (0, 1, 2, 3): with self.subTest(verbose=verbose): @@ -106,6 +141,77 @@ def test_the_other_libraries_keep_their_levels(self): self.assertEqual(self.level(name), logging.DEBUG, name) self.assertEqual(self.level("PIL"), logging.WARNING) + def test_botocore_signing_and_credentials_are_never_below_warning(self): + for verbose in (0, 1, 2): + with self.subTest(verbose=verbose): + app.configure_logging(verbose) + self.assertEqual(self.level("botocore.auth"), logging.WARNING) + self.assertEqual(self.level("botocore.credentials"), logging.WARNING) + + def test_the_endpoint_logger_still_shows_requests_at_the_deepest_verbosity(self): + app.configure_logging(2) + self.assertEqual(self.level("botocore.endpoint"), logging.DEBUG) + + +class CredentialRedactionTests(LoggingStateMixin, unittest.TestCase): + def emit(self): + buffer = StringIO() + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + with patch.object(app, "console", Console(file=buffer, width=2000)): + app.configure_logging(2) + app.add_run_log_handler(tmp.name) + try: + endpoint = logging.getLogger("botocore.endpoint") + endpoint.debug("Sending http request: %s", PreparedRequest()) + endpoint.debug( + "Making request with headers Authorization: AWS4-HMAC-SHA256 " + f"Credential={ACCESS_KEY}/20261025/eu-west-1/ec2/aws4_request, " + f"SignedHeaders=host;x-amz-date, Signature={SIGNATURE}" + ) + logging.getLogger("botocore.auth").warning( + "CanonicalRequest:\nPOST\n/\n\nx-amz-security-token:%s\n", TOKEN + ) + logging.getLogger("botocore.credentials").warning( + "Found credentials: aws_secret_access_key=%s, " + "aws_session_token=%s", + SECRET, + TOKEN, + ) + finally: + app.close_run_log_handler() + run_log = (Path(tmp.name) / "run.log").read_text(encoding="utf-8") + # The console wraps long lines, so compare without whitespace + return "".join(buffer.getvalue().split()), run_log + + def test_no_credential_reaches_the_console_or_the_run_log(self): + console_text, run_log = self.emit() + for name, text in (("console", console_text), ("run.log", run_log)): + for secret in (ACCESS_KEY, SIGNATURE, TOKEN, SECRET): + with self.subTest(target=name, secret=secret[:8]): + self.assertNotIn(secret, text) + with self.subTest(target=name): + self.assertIn("REDACTED", text) + + def test_the_request_detail_around_the_secret_is_kept(self): + _, run_log = self.emit() + self.assertIn("ec2.eu-west-1.amazonaws.com", run_log) + self.assertIn("botocore.endpoint - DEBUG", run_log) + self.assertIn("'Authorization': REDACTED", run_log) + + def test_a_credential_scope_outside_a_header_keeps_its_shape(self): + # A presigned URL carries the key ID in the query string, where the + # scope that follows it is worth reading + cleaned = redact( + "https://s3.eu-west-1.amazonaws.com/bucket/key?" + f"X-Amz-Credential={ACCESS_KEY}/20261025/eu-west-1/s3/aws4_request" + f"&X-Amz-Signature={SIGNATURE}" + ) + self.assertIn("Credential=REDACTED/20261025/eu-west-1/s3/aws4_request", cleaned) + self.assertIn("X-Amz-Signature=REDACTED", cleaned) + self.assertNotIn(ACCESS_KEY, cleaned) + self.assertNotIn(SIGNATURE, cleaned) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_settings.py b/tests/test_settings.py index e43a40b..4e4c41d 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -1,8 +1,10 @@ +import ast import contextlib import io import os +import subprocess +import sys import tempfile -import types import unittest from argparse import Namespace from pathlib import Path @@ -25,6 +27,8 @@ LOGGER = "cloudexit.settings" KEY = "sk_live_0123456789abcdef" +LEGACY_HOST = "platform.test" +LEGACY_KEY = "fixture-key-0123456789" class SettingsTestCase(unittest.TestCase): @@ -32,9 +36,6 @@ def setUp(self): tmp = tempfile.TemporaryDirectory() self.addCleanup(tmp.cleanup) self.tmp = Path(tmp.name) - patcher = patch("cloudexit.settings.legacy_config", None) - patcher.start() - self.addCleanup(patcher.stop) @property def config_path(self): @@ -56,10 +57,16 @@ def load(self, flags=None, env=None): environ = {**isolated_environ(self.tmp), **(env or {})} return load_settings(flags, environ=environ) - def legacy(self, **values): - return patch( - "cloudexit.settings.legacy_config", types.SimpleNamespace(**values) + @contextlib.contextmanager + def directory_with_config_py(self): + # A real config.py next to the run, to prove it is never read + legacy = self.tmp / "legacy" + legacy.mkdir(exist_ok=True) + (legacy / "config.py").write_text( + f"HOST = 'old.test'\nKEY = '{LEGACY_KEY}'\n", encoding="utf-8" ) + with contextlib.chdir(legacy): + yield legacy class ConfigFilePathTests(unittest.TestCase): @@ -119,36 +126,18 @@ def test_nothing_configured_is_all_unset(self): self.assertEqual(settings.timeout.source, "default") self.assertFalse(settings.config_found) - def test_host_flag_over_env_over_file_over_legacy(self): + def test_host_flag_over_env_over_file(self): self.write_config('[default]\nhost = "file.test"\n') - env = { - "CLOUDEXIT_HOST": "env.test", - "HOST": "legacy-env.test", - "KEY": "legacy-key", - } - with self.legacy(HOST="legacy-config.test", KEY="legacy-key"): - steps = [ - ( - self.load({"host": "flag.test"}, env), - "https://flag.test", - "flag --host", - ), - (self.load(None, env), "https://env.test", "env CLOUDEXIT_HOST"), - ( - self.load(None, {"HOST": "legacy-env.test", "KEY": "k"}), - "https://file.test", - "config file [default]", - ), - ] - self.config_path.unlink() - steps += [ - ( - self.load(None, {"HOST": "legacy-env.test", "KEY": "k"}), - "https://legacy-env.test", - "legacy env HOST", - ), - (self.load(), "https://legacy-config.test", "legacy config.py"), - ] + env = {"CLOUDEXIT_HOST": "env.test"} + steps = [ + ( + self.load({"host": "flag.test"}, env), + "https://flag.test", + "flag --host", + ), + (self.load(None, env), "https://env.test", "env CLOUDEXIT_HOST"), + (self.load(), "https://file.test", "config file [default]"), + ] for settings, value, source in steps: self.assertEqual(settings.host.value, value) @@ -401,60 +390,123 @@ def test_private_key_file_does_not_warn(self): class LegacySourceTests(SettingsTestCase): - def test_host_and_key_env_vars_are_used_together_with_a_warning(self): - env = {"HOST": "legacy.test", "KEY": "legacy-key-0123456"} + def test_host_and_key_env_vars_are_no_longer_read(self): + env = {"HOST": LEGACY_HOST, "KEY": LEGACY_KEY} with self.assertLogs(LOGGER, level="WARNING") as logs: settings = self.load(None, env) - self.assertEqual(settings.host.value, "https://legacy.test") - self.assertEqual(settings.host.source, "legacy env HOST") - self.assertEqual(settings.api_key.value, "legacy-key-0123456") - self.assertEqual(settings.api_key.source, "legacy env KEY") + self.assertIsNone(settings.host.value) + self.assertIsNone(settings.api_key.value) self.assertEqual(len(logs.output), 1) - self.assertIn("deprecated", logs.output[0]) - self.assertNotIn("legacy-key-0123456", logs.output[0]) + self.assertIn("HOST and KEY are no longer read", logs.output[0]) + self.assertNotIn(LEGACY_HOST, logs.output[0]) + self.assertNotIn(LEGACY_KEY, logs.output[0]) - def test_host_alone_or_key_alone_is_ignored(self): - for env in ({"HOST": "legacy.test"}, {"KEY": "legacy-key-0123456"}): + def test_host_alone_or_key_alone_does_not_warn(self): + for env in ({"HOST": LEGACY_HOST}, {"KEY": LEGACY_KEY}): with self.assertNoLogs(LOGGER, level="WARNING"): settings = self.load(None, env) self.assertIsNone(settings.host.value) self.assertIsNone(settings.api_key.value) - def test_config_py_is_used_with_a_warning(self): - with ( - self.legacy(HOST="old.test", KEY="old-key-0123456789"), - self.assertLogs(LOGGER, level="WARNING") as logs, - ): - settings = self.load() + def test_cloudexit_env_vars_are_used_and_do_not_warn(self): + env = { + "HOST": LEGACY_HOST, + "KEY": LEGACY_KEY, + "CLOUDEXIT_HOST": "env.test", + "CLOUDEXIT_API_KEY": "inline-key-123456", + } + with self.assertNoLogs(LOGGER, level="WARNING"): + settings = self.load(None, env) - self.assertEqual(settings.host.value, "https://old.test") - self.assertEqual(settings.host.source, "legacy config.py") - self.assertEqual(settings.api_key.source, "legacy config.py") - self.assertEqual(len(logs.output), 1) - self.assertIn("config.py is deprecated", logs.output[0]) - self.assertNotIn("old-key-0123456789", logs.output[0]) + self.assertEqual(settings.host.value, "https://env.test") + self.assertEqual(settings.host.source, "env CLOUDEXIT_HOST") + self.assertEqual(settings.api_key.source, "env CLOUDEXIT_API_KEY") - def test_env_pair_beats_config_py(self): - env = {"HOST": "legacy.test", "KEY": "legacy-key-0123456"} - with self.legacy(HOST="old.test", KEY="old-key"): - with self.assertLogs(LOGGER, level="WARNING"): - settings = self.load(None, env) - self.assertEqual(settings.host.source, "legacy env HOST") + def test_a_config_py_in_the_working_directory_is_no_longer_read(self): + with self.directory_with_config_py(): + with self.assertLogs(LOGGER, level="WARNING") as logs: + settings = self.load() - def test_no_warning_when_a_higher_source_covers_everything(self): - with self.legacy(HOST="old.test", KEY="old-key-0123456789"): + self.assertIsNone(settings.host.value) + self.assertIsNone(settings.api_key.value) + self.assertEqual(len(logs.output), 1) + self.assertIn("config.py is no longer read", logs.output[0]) + self.assertNotIn(LEGACY_KEY, logs.output[0]) + + def test_a_host_from_another_source_silences_the_config_py_warning(self): + with self.directory_with_config_py(): with self.assertNoLogs(LOGGER, level="WARNING"): - settings = self.load( - {"host": "flag.test"}, {"CLOUDEXIT_API_KEY": "inline-key-123456"} - ) - self.assertEqual(settings.host.source, "flag --host") + settings = self.load(None, {"CLOUDEXIT_HOST": "env.test"}) + self.assertEqual(settings.host.source, "env CLOUDEXIT_HOST") - def test_an_empty_config_py_or_a_namespace_package_is_ignored(self): - for legacy in (types.SimpleNamespace(), types.SimpleNamespace(HOST="", KEY="")): - with self.legacy(**vars(legacy)): - settings = self.load() - self.assertIsNone(settings.host.value) + def test_both_legacy_sources_give_a_single_warning(self): + env = {"HOST": LEGACY_HOST, "KEY": LEGACY_KEY} + with self.directory_with_config_py(): + with self.assertLogs(LOGGER, level="WARNING") as logs: + self.load(None, env) + + self.assertEqual(len(logs.output), 1) + self.assertIn("HOST and KEY are no longer read", logs.output[0]) + self.assertIn("config.py is no longer read", logs.output[0]) + + def test_no_module_in_the_package_imports_config(self): + offenders = [] + for path in Path("cloudexit").rglob("*.py"): + tree = ast.parse(path.read_text(encoding="utf-8")) + for node in ast.walk(tree): + names = [] + if isinstance(node, ast.Import): + names = [alias.name for alias in node.names] + elif isinstance(node, ast.ImportFrom) and node.level == 0: + names = [node.module or ""] + if any(name.split(".")[0] == "config" for name in names): + offenders.append(str(path)) + self.assertEqual(offenders, []) + + +class LegacyConfigPySubprocessTests(unittest.TestCase): + # `python -m cloudexit` puts the current directory first on sys.path, so a + # config.py there used to be executed at start-up + def setUp(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + self.tmp = Path(tmp.name) + self.marker = self.tmp / "config-py-was-executed" + (self.tmp / "config.py").write_text( + "from pathlib import Path\n" + f"Path({str(self.marker)!r}).write_text('executed')\n" + "HOST = 'old.test'\n" + f"KEY = {LEGACY_KEY!r}\n", + encoding="utf-8", + ) + + def run_cloudexit(self, *args): + root = Path(__file__).resolve().parent.parent + env = {**os.environ, "PYTHONPATH": str(root)} + for name in ("HOST", "KEY", "CLOUDEXIT_HOST", "CLOUDEXIT_API_KEY"): + env.pop(name, None) + return subprocess.run( + [sys.executable, "-m", "cloudexit", *args], + cwd=self.tmp, + env=env, + capture_output=True, + text=True, + ) + + def test_version_does_not_execute_config_py(self): + result = self.run_cloudexit("--version") + self.assertEqual(result.returncode, 0, result.stderr) + self.assertFalse(self.marker.exists()) + + def test_config_show_lists_no_legacy_source(self): + result = self.run_cloudexit("config", "show") + self.assertEqual(result.returncode, 0, result.stderr) + self.assertFalse(self.marker.exists()) + out = result.stdout + result.stderr + self.assertNotIn("legacy", out) + self.assertNotIn("old.test", out) + self.assertNotIn(LEGACY_KEY, out) class DescribeTests(SettingsTestCase):