diff --git a/README.md b/README.md index 5f2d88f..0e3a62c 100644 --- a/README.md +++ b/README.md @@ -46,7 +46,9 @@ Plus `scoring.py` (the 0-100 score) and `plexavo/report/ai_narration.py` Illustrative example (a public S3 bucket, an unencrypted volume, a role that's never been assumed) — this is what a scan actually surfaces, including a -finding walked through `--explain`: +finding's free template remediation (shown by default, no `--explain` +needed; `--explain` would replace this panel with a full AI narrative +instead): ``` Your AWS Security Score: 74/100 (Good) @@ -154,25 +156,31 @@ plexavo scan --profile my-aws-profile --report-html report.html ``` No `--profile`? It uses your default profile / environment variables, -same resolution order as the AWS CLI. No AI, no API key, no cost — this -alone is a complete, genuinely useful scan. +same resolution order as the AWS CLI. No AI, no API key, no cost — and +findings still come with free Next Step / Full Fix Detail guidance +wherever a template exists (10 common check types); this alone is a +complete, genuinely useful scan. -Want plain-English explanations for each finding too (needs the -`[ai]` install above): +Want a full AI-written explanation for *every* finding instead (needs +the `[ai]` install above)? ```bash export ANTHROPIC_API_KEY="sk-ant-..." # your own key, your own account plexavo scan --profile my-aws-profile --explain --report-html report.html --report-pdf report.pdf ``` -- Drop `--explain` for a fast, free scan with raw technical findings only. +- Drop `--explain` and findings still get free template remediation where + available (no API key needed) and raw technical detail otherwise. + `--explain` replaces that with a live AI narrative for every finding, + including the templated ones. - Drop `--report-html`/`--report-pdf` to just see the console table. -- `--explain-limit N` (default 25) caps how many findings get AI narration - in one run, as a safety rail against unexpectedly large real scans. +- `--explain-limit N` (default 25) caps how many findings get a live AI + call when `--explain` is passed, as a safety rail against unexpectedly + large real scans — it doesn't limit the free template remediation. - No `ANTHROPIC_API_KEY` set, or a call fails for any reason (invalid key, rate limit, network issue)? The scan and report are completely - unaffected — you get raw finding detail instead of narration for that - finding, not an error. See [Cost](#cost). + unaffected — that finding falls back to template/raw detail instead of + an AI narrative, not an error. See [Cost](#cost). ### Alternative: pipx @@ -260,10 +268,14 @@ docs/ ## Cost Detection is free (pure Python/boto3), always, regardless of anything -else in this section. AI narration only runs with `--explain`, and even -then: 10 of the most common, narratively-generic finding types are -hand-written templates with zero API cost; only genuinely account-specific -findings call Claude, typically $0.01-0.02 per finding depending on answer +else in this section. Every scan also gets free Next Step / Full Fix +Detail remediation wherever one of 10 hand-written templates matches the +finding type — no flag, no API key, zero cost, on by default. + +Live AI only runs with `--explain`, and when it does it's used for +*every* finding, including the 10 templated ones (a deliberate choice — +"AI narration on" always means fully AI-written content, not a mix of +template and AI), typically $0.01-0.02 per finding depending on answer length. A full scan with `--explain` on a real account is usually well under a dollar. This is **your own** `ANTHROPIC_API_KEY`, in **your own** Anthropic account — this project never sees your key, never embeds one of diff --git a/plexavo/__init__.py b/plexavo/__init__.py index 7f0517a..d1393d3 100644 --- a/plexavo/__init__.py +++ b/plexavo/__init__.py @@ -4,4 +4,4 @@ anyone else. See README.md for usage, or `plexavo scan --help`. """ -__version__ = "0.1.2" +__version__ = "0.2.0" diff --git a/plexavo/aws_profile_setup.py b/plexavo/aws_profile_setup.py new file mode 100644 index 0000000..6b11fb8 --- /dev/null +++ b/plexavo/aws_profile_setup.py @@ -0,0 +1,66 @@ +"""plexavo/aws_profile_setup.py — writes a new named AWS profile to disk. + +Used by the interactive "+ Configure new profile" flow. Writes in the +same layout `aws configure --profile NAME` itself produces: + ~/.aws/credentials -> [NAME] aws_access_key_id / aws_secret_access_key + ~/.aws/config -> [profile NAME] region + (bare [default] for the "default" profile) + +Only the target profile's section is added/updated — every other +section in either file is read back untouched via configparser's +read-modify-write round trip. One known limitation: configparser does +not preserve comment lines on rewrite, so hand-written comments in +these files (uncommon, since they're normally CLI/tool-managed) would +be dropped. Actual key/value data for every other profile is preserved +exactly. + +Respects AWS_SHARED_CREDENTIALS_FILE / AWS_CONFIG_FILE if set, same as +the AWS CLI and boto3 itself. +""" + +from __future__ import annotations + +import configparser +import os +from pathlib import Path + + +def credentials_path() -> Path: + return Path(os.environ.get("AWS_SHARED_CREDENTIALS_FILE", "~/.aws/credentials")).expanduser() + + +def config_path() -> Path: + return Path(os.environ.get("AWS_CONFIG_FILE", "~/.aws/config")).expanduser() + + +def _write_ini(path: Path, section: str, values: dict[str, str]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + parser = configparser.ConfigParser() + if path.exists(): + parser.read(path) + if not parser.has_section(section): + parser.add_section(section) + for key, value in values.items(): + parser.set(section, key, value) + with open(path, "w", encoding="utf-8") as f: + parser.write(f) + try: + os.chmod(path, 0o600) + except OSError: + # Windows doesn't support POSIX file-mode bits the same way — + # best-effort only, same as the AWS CLI itself does here. + pass + + +def write_profile(name: str, access_key_id: str, secret_access_key: str, region: str) -> None: + """Write/overwrite one named profile's credentials and region. + + Never logs or returns the secret — callers must not print it either. + """ + _write_ini(credentials_path(), name, { + "aws_access_key_id": access_key_id, + "aws_secret_access_key": secret_access_key, + }) + + config_section = "default" if name == "default" else f"profile {name}" + _write_ini(config_path(), config_section, {"region": region}) diff --git a/plexavo/cli.py b/plexavo/cli.py index 20be7d5..7470d5c 100644 --- a/plexavo/cli.py +++ b/plexavo/cli.py @@ -12,7 +12,11 @@ """ import argparse +import os +import random import sys +import time +from contextlib import nullcontext from rich.console import Console from rich.panel import Panel @@ -29,12 +33,32 @@ from plexavo.checks import logging as logging_checks from plexavo.checks import usage as usage_checks from plexavo.scoring import calculate_score -from plexavo.report.ai_narration import explain_finding +from plexavo.report.ai_narration import explain_finding, COMMON_CHECK_TEMPLATES from plexavo.report.html_report import build_report_data, generate_html from plexavo.report.pdf import generate_pdf console = Console() +# Cloud-security-themed flavor words shown next to the spinner during a +# scan (real terminals only — see _run_scan's is_terminal branch). Each +# scan shuffles a fresh order so back-to-back runs don't feel identical. +FLAVOR_WORDS = [ + "Pondering privilege escalation paths", + "Sweeping exposed security groups", + "Auditing encryption at rest", + "Cross-referencing CloudTrail history", + "Hunting dormant credentials", + "Tracing IAM trust relationships", + "Probing public S3 exposure", + "Checking GuardDuty coverage", + "Weighing blast radius", + "Chasing unused permissions", + "Untangling trust policies", + "Mapping the attack surface", + "Interrogating access keys", + "Scrutinizing bucket policies", +] + def _build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( @@ -44,7 +68,7 @@ def _build_parser() -> argparse.ArgumentParser: ) parser.add_argument("--version", action="version", version=f"plexavo {__version__}") - subparsers = parser.add_subparsers(dest="command", required=True) + subparsers = parser.add_subparsers(dest="command", required=False) scan = subparsers.add_parser("scan", help="Run a scan against one AWS account") scan.add_argument("--profile", default=None, @@ -53,19 +77,22 @@ def _build_parser() -> argparse.ArgumentParser: scan.add_argument("--region", default=None, help="AWS region to scan (default: your configured region)") scan.add_argument("--explain", action="store_true", - help="Generate AI narratives (WHAT'S WRONG / WHAT AN ATTACKER DOES / HOW TO FIX) for each " - "finding. Templated findings are free and need no API key. Non-templated findings " - "make a live Anthropic API call using YOUR OWN ANTHROPIC_API_KEY and your own " + help="Generate a live AI narrative (WHAT'S WRONG / WHAT AN ATTACKER DOES / HOW TO FIX) for " + "every finding via the Anthropic API, using YOUR OWN ANTHROPIC_API_KEY and your own " "account's credits (typically a few cents for a full scan) — never a project-owned " - "key. No key set, or the call fails for any reason? Falls back to raw finding detail " - "automatically, the scan never stops because of it. Off by default.") + "key. No key set, or a call fails for any reason? That finding falls back to raw " + "detail automatically, the scan never stops because of it. Off by default — without " + "it, findings still get free, no-API-key-needed remediation text wherever a template " + "exists (10 common check types); --explain always uses live AI instead, for every " + "finding, not just the non-templated ones.") scan.add_argument("--explain-limit", type=int, default=25, - help="Safety cap on how many findings to explain in one run, in case a scan turns up more " - "than expected (default: 25)") + help="Safety cap on how many findings get a live AI call when --explain is passed, in " + "case a scan turns up more than expected (default: 25). Doesn't affect the free " + "template remediation, which always applies to every eligible finding.") scan.add_argument("--report-html", metavar="PATH", default=None, help="Write a self-contained, offline HTML report to PATH — no external requests, opens " - "anywhere. Independent of --explain — if --explain wasn't passed, the report shows " - "raw technical detail instead of AI narratives for each finding.") + "anywhere. Includes free template remediation where available even without --explain; " + "--explain replaces that with a full AI narrative for every finding instead.") scan.add_argument("--report-pdf", metavar="PATH", default=None, help="Write a PDF report to PATH. Same --explain behavior as --report-html.") return parser @@ -83,31 +110,62 @@ def _run_scan(args) -> None: console.print(f"[bold]Scanning account:[/bold] {account_id}") console.print(f"[bold]Region:[/bold] {session.region_name}\n") - console.print("Enumerating IAM principals...") - principals = list_all_principals(session) - console.print(f"Found {len(principals)} principals ({sum(p.type == 'user' for p in principals)} users, " - f"{sum(p.type == 'role' for p in principals)} roles)\n") - - console.print("Running checks IAM-01 through IAM-06...") - findings = iam_checks.run_all(session, principals) - - console.print("Running checks IAM-07 through IAM-14 (hygiene)...") - findings += iam_hygiene.run_all(session, principals, account_id) - - console.print("Running checks NET-01 through NET-04...") - findings += network_checks.run_all(session) - - console.print("Running checks STOR-19 through STOR-21...") - findings += storage_checks.run_all(session) - - console.print("Running checks ENC-29 through ENC-31...") - findings += encryption_checks.run_all(session) - - console.print("Running checks LOG-22 through LOG-25...") - findings += logging_checks.run_all(session) - - console.print("Running checks USE-26, USE-27 (usage analysis)...") - findings += usage_checks.run_all(session, principals) + # Real terminal: a live spinner + rotating flavor word + running + # stats, restyled each stage. Piped/non-terminal (CI, log capture): + # console.status() produces ZERO output when console.is_terminal is + # False (confirmed — Rich only renders it in real terminals), so + # scripting users get the exact same plain progress lines as before, + # not silence. This is a deliberate branch, not an oversight. + is_tty = console.is_terminal + findings = [] + principals = [] + stage_num = 0 + total_stages = 8 + start_time = time.monotonic() + flavor_pool = FLAVOR_WORDS.copy() + random.shuffle(flavor_pool) + + status = console.status("Starting scan...") if is_tty else None + + def stage(technical_label: str) -> None: + nonlocal stage_num + stage_num += 1 + if is_tty: + elapsed = time.monotonic() - start_time + word = flavor_pool[(stage_num - 1) % len(flavor_pool)] + stats = (f"[grey50]{technical_label} · Stage {stage_num}/{total_stages} · " + f"Principals: {len(principals)} · Findings so far: {len(findings)} · " + f"Elapsed: {elapsed:0.0f}s[/grey50]") + status.update(f"[grey74]{word}...[/grey74]\n{stats}") + else: + console.print(f"{technical_label}...") + + with status if is_tty else nullcontext(): + stage("Enumerating IAM principals") + principals = list_all_principals(session) + console.print(f"Found {len(principals)} principals ({sum(p.type == 'user' for p in principals)} users, " + f"{sum(p.type == 'role' for p in principals)} roles)\n") + + stage("Running checks IAM-01 through IAM-06") + findings = iam_checks.run_all(session, principals) + + stage("Running checks IAM-07 through IAM-14 (hygiene)") + findings += iam_hygiene.run_all(session, principals, account_id) + + stage("Running checks NET-01 through NET-04") + findings += network_checks.run_all(session) + + stage("Running checks STOR-19 through STOR-21") + findings += storage_checks.run_all(session) + + stage("Running checks ENC-29 through ENC-31") + findings += encryption_checks.run_all(session) + + stage("Running checks LOG-22 through LOG-25") + findings += logging_checks.run_all(session) + + stage("Running checks USE-26, USE-27 (usage analysis)") + findings += usage_checks.run_all(session, principals) result = calculate_score(findings) rating_color = {"Excellent": "green", "Good": "green", "Fair": "yellow", "Poor": "orange3", "Critical": "red"} @@ -138,52 +196,67 @@ def _run_scan(args) -> None: ) console.print(table) - # Built once, reused for both the console panels below (if --explain) - # and report generation (if --report-html/--report-pdf) — explaining - # a finding twice would mean two real API calls for the same finding, - # doubling cost for no reason. + # Built once, reused for both the console panels below and report + # generation (if --report-html/--report-pdf) — explaining a finding + # twice would mean two real API calls for the same finding, doubling + # cost for no reason. explanations = [None] * len(findings) if args.explain: + # AI on: every finding (up to --explain-limit) gets a live API + # call, bypassing the free templates entirely — "AI narration on" + # always means fully AI-written content, never a mix. to_explain = findings[:args.explain_limit] skipped = len(findings) - len(to_explain) - console.print(f"\n[bold]Generating explanations for {len(to_explain)} finding(s)...[/bold]") + console.print(f"\n[bold]Generating AI explanations for {len(to_explain)} finding(s)...[/bold]") if skipped: console.print(f"[yellow]{skipped} additional finding(s) skipped — raise --explain-limit to include them.[/yellow]") - console.print("[dim]Templated findings are free and need no API key. Others use your own " - "ANTHROPIC_API_KEY, if set, and cost a few cents in your own account.[/dim]\n") - - source_counts = {} + console.print("[dim]Uses your own ANTHROPIC_API_KEY, if set, and costs a few cents per finding " + "in your own account.[/dim]\n") for i, f in enumerate(to_explain): - explanation = explain_finding(f) - explanations[i] = explanation - source_counts[explanation.source] = source_counts.get(explanation.source, 0) + 1 - - if explanation.source == "fallback": - # Same clean content the report shows — no raw exception - # text at the console either. This is what "no AI" looks - # like whether that's because no key was set, the key - # was invalid, or the call failed for any other reason. - body = (f"[bold]{f.check_id}[/bold] — {f.resource_arn.split('/')[-1]}\n\n" - f"{explanation.impact}\n\n" - f"[dim]No AI narration for this finding — set ANTHROPIC_API_KEY for " - f"plain-English explanations, or see the raw detail above.[/dim]") - else: - confidence_style = "yellow" if explanation.confidence != "Confirmed" else "dim" - body = (f"[bold]{f.check_id}[/bold] — {f.resource_arn.split('/')[-1]}\n\n" - f"[bold]IMPACT:[/bold] {explanation.impact}\n\n" - f"[{confidence_style}]CONFIDENCE: {explanation.confidence}[/{confidence_style}]\n\n") - if explanation.evidence: - body += f"[bold]EVIDENCE:[/bold] {explanation.evidence}\n\n" - body += (f"[bold green]NEXT STEP:[/bold green] {explanation.next_step}\n\n" - f"[bold]FULL FIX DETAIL:[/bold] {explanation.how_to_fix}") - - console.print(Panel( - body, - border_style=severity_color.get(f.severity.value, "white"), - title=f"[dim]source: {explanation.source}[/dim]", - )) - + explanations[i] = explain_finding(f, use_ai=True) + else: + # AI off (default): free templates only, for every finding, no + # network call, no key needed, no limit — this used to require + # --explain just to see, at zero actual AI cost. + console.print(f"\n[dim]Adding free remediation guidance where a template is available " + f"({len(COMMON_CHECK_TEMPLATES)} check types) — pass --explain for full AI-written " + f"explanations on every finding instead.[/dim]") + for i, f in enumerate(findings): + explanations[i] = explain_finding(f, use_ai=False) + + source_counts = {} + for f, explanation in zip(findings, explanations): + if explanation is None: + continue + source_counts[explanation.source] = source_counts.get(explanation.source, 0) + 1 + + if explanation.source == "fallback": + # Same clean content the report shows — no raw exception + # text at the console either. This is what "no AI" looks + # like whether that's because no key was set, the key + # was invalid, or the call failed for any other reason. + body = (f"[bold]{f.check_id}[/bold] — {f.resource_arn.split('/')[-1]}\n\n" + f"{explanation.impact}\n\n" + f"[dim]No AI narration for this finding — set ANTHROPIC_API_KEY for " + f"plain-English explanations, or see the raw detail above.[/dim]") + else: + confidence_style = "yellow" if explanation.confidence != "Confirmed" else "dim" + body = (f"[bold]{f.check_id}[/bold] — {f.resource_arn.split('/')[-1]}\n\n" + f"[bold]IMPACT:[/bold] {explanation.impact}\n\n" + f"[{confidence_style}]CONFIDENCE: {explanation.confidence}[/{confidence_style}]\n\n") + if explanation.evidence: + body += f"[bold]EVIDENCE:[/bold] {explanation.evidence}\n\n" + body += (f"[bold green]NEXT STEP:[/bold green] {explanation.next_step}\n\n" + f"[bold]FULL FIX DETAIL:[/bold] {explanation.how_to_fix}") + + console.print(Panel( + body, + border_style=severity_color.get(f.severity.value, "white"), + title=f"[dim]source: {explanation.source}[/dim]", + )) + + if source_counts: summary = ", ".join(f"{v} {k}" for k, v in source_counts.items()) console.print(f"\n[dim]{summary}[/dim]") if source_counts.get("fallback"): @@ -194,19 +267,23 @@ def _run_scan(args) -> None: if args.report_html or args.report_pdf: if not args.explain: - console.print("\n[yellow]Generating report without AI narratives (raw technical detail only) — " - "pass --explain too for plain-English explanations in the report.[/yellow]") + console.print("\n[yellow]Report includes free template remediation where available, raw technical " + "detail otherwise — pass --explain for full AI-written explanations on every " + "finding instead.[/yellow]") report_data = build_report_data(findings, result, account_id, explanations) if args.report_html: html = generate_html(report_data) with open(args.report_html, "w", encoding="utf-8") as f: f.write(html) - console.print(f"[green]HTML report written to {args.report_html}[/green]") + # Absolute path, not the raw --report-html value — a bare name + # like "report.html" is easy to lose track of otherwise; this + # always says exactly where it landed. + console.print(f"[green]HTML report written to {os.path.abspath(args.report_html)}[/green]") if args.report_pdf: generate_pdf(report_data, args.report_pdf) - console.print(f"[green]PDF report written to {args.report_pdf}[/green]") + console.print(f"[green]PDF report written to {os.path.abspath(args.report_pdf)}[/green]") def main(): @@ -214,6 +291,12 @@ def main(): args = parser.parse_args() if args.command == "scan": _run_scan(args) + elif args.command is None: + if sys.stdout.isatty(): + from plexavo.interactive import run_interactive + run_interactive() + else: + parser.print_help() if __name__ == "__main__": diff --git a/plexavo/interactive.py b/plexavo/interactive.py new file mode 100644 index 0000000..2542e17 --- /dev/null +++ b/plexavo/interactive.py @@ -0,0 +1,300 @@ +"""plexavo/interactive.py — interactive terminal experience. + +Launched when `plexavo` is run with no subcommand in a real terminal. +This is additive: `plexavo scan --profile ... --report-html ...` (the +scripting/CI path documented in the README) never touches this module +and keeps working exactly as before. + +Built in slices — see plexavo-journal.md for what's live so far. +Slice 1: splash screen. Slice 2: profile picker + live status check. +Slice 3: new-profile write flow. +""" + +from __future__ import annotations + +import os + +import boto3 +from rich import box +from rich.align import Align +from rich.console import Console +from rich.panel import Panel +from rich.prompt import Confirm, IntPrompt, Prompt +from rich.text import Text + +from plexavo import __version__ +from plexavo.auth import get_local_session +from plexavo.aws_profile_setup import write_profile + +WEBSITE = "https://plexavo.com" +NEW_PROFILE_CHOICE = "+ Configure new profile" + +REPORT_FORMATS = [ + ("HTML report", "html"), + ("PDF report", "pdf"), + ("Both HTML and PDF", "both"), + ("Console output only (no file)", "none"), +] + +FEATURES = [ + "AI-powered remediation guidance for every finding", + "IAM privilege escalation path mapping", + "100% local scanning — your AWS credentials never leave this machine", +] + +def _splash_panel() -> Panel: + header = Text() + header.append("PLEXAVO", style="bold white") + header.append(" ") + header.append(f"v{__version__}", style="grey62") + + body = Text() + body.append_text(header) + body.append("\n") + body.append("AWS Security Reimagined", style="italic grey74") + body.append("\n") + body.append(WEBSITE, style="underline grey62") + body.append("\n\n") + body.append("FEATURES\n", style="bold grey74") + for i, feature in enumerate(FEATURES): + body.append(f" · {feature}", style="grey66") + if i != len(FEATURES) - 1: + body.append("\n") + + return Panel( + Align.center(body), + border_style="grey50", + box=box.ROUNDED, + padding=(1, 4), + ) + + +def show_splash(console: Console) -> None: + console.print() + console.print(_splash_panel()) + console.print() + + +def _prompt_profile_menu(console: Console) -> str | None: + """Numbered profile menu — plain line-buffered input (type a number, + press Enter), not a live arrow-key redraw. Deliberate choice: the + prompt_toolkit-based arrow-key menu this replaced was unreliable + across Windows terminal hosts (desyncs, missed keys, sometimes never + registering) — this is immune to that whole class of bug.""" + profiles = sorted(boto3.Session().available_profiles) + if not profiles: + console.print("[grey62]No AWS profiles found in ~/.aws/credentials or ~/.aws/config.[/grey62]\n") + + options = list(profiles) + [NEW_PROFILE_CHOICE] + + console.print("[bold]Select an AWS profile to scan:[/bold]\n") + for i, name in enumerate(options, start=1): + console.print(f" [bold green]{i}[/bold green] {name}") + console.print() + + try: + choice = IntPrompt.ask( + "Enter a number", + choices=[str(i) for i in range(1, len(options) + 1)], + show_choices=False, + console=console, + ) + except (KeyboardInterrupt, EOFError): + return None + + return options[choice - 1] + + +def _prompt_new_profile(console: Console) -> str | None: + """Collects Access Key ID / Secret Access Key (masked) / region / + profile name, writes them to ~/.aws/credentials + ~/.aws/config + (plexavo.aws_profile_setup.write_profile — same layout `aws + configure` itself produces), and returns the new profile's name for + the caller to auto-select. Returns None on cancel (Ctrl+C/EOF) or + if either key field is left empty. + """ + console.print("\n[bold]Configure a new AWS profile[/bold]\n") + existing = set(boto3.Session().available_profiles) + + try: + while True: + name = Prompt.ask("Profile name", console=console).strip() + if not name: + console.print("[red]Profile name can't be empty.[/red]") + continue + if name in existing and not Confirm.ask( + f"Profile '{name}' already exists — overwrite it?", + console=console, default=False, + ): + continue + break + + access_key = Prompt.ask("AWS Access Key ID", console=console).strip() + secret_key = Prompt.ask("AWS Secret Access Key", password=True, console=console).strip() + region = Prompt.ask("Default region", default="us-east-1", console=console).strip() + except (KeyboardInterrupt, EOFError): + return None + + if not access_key or not secret_key: + console.print("[red]Access key and secret key are both required — cancelled.[/red]") + return None + + write_profile(name, access_key, secret_key, region) + console.print(f"\n[green]Saved profile '{name}' to ~/.aws/credentials and ~/.aws/config.[/green]") + return name + + +def _check_profile(console: Console, profile_name: str) -> tuple[boto3.Session, dict] | None: + """Live sts get_caller_identity status check. + + Returns (session, identity) on success, None on failure — reuses + auth.get_local_session so this is the exact same validation the real + scan uses, just surfaced here instead of discovered mid-scan. The + identity dict is handed back so callers (the confirm+run summary) + don't need a second sts call for the same information. + """ + with console.status(f"[grey62]Checking '{profile_name}'...[/grey62]"): + try: + session = get_local_session(profile_name=profile_name) + except RuntimeError as e: + error = str(e) + else: + error = None + + if error is not None: + console.print(Panel( + f"[bold red]✗ Failed[/bold red]\n\n{error}", + border_style="red", box=box.ROUNDED, + )) + return None + + identity = session.client("sts").get_caller_identity() + console.print(Panel( + f"[bold green]✓ Active[/bold green]\n\n" + f"Account: {identity['Account']}\n" + f"ARN: {identity['Arn']}\n" + f"Region: {session.region_name}", + border_style="green", box=box.ROUNDED, + )) + return session, identity + + +def _prompt_report_options(console: Console) -> dict | None: + """Report format + AI-narration choices. Returns None on cancel.""" + console.print("\n[bold]Report format:[/bold]\n") + for i, (label, _) in enumerate(REPORT_FORMATS, start=1): + console.print(f" [bold green]{i}[/bold green] {label}") + console.print() + + try: + idx = IntPrompt.ask( + "Enter a number", + choices=[str(i) for i in range(1, len(REPORT_FORMATS) + 1)], + show_choices=False, + console=console, + ) + fmt = REPORT_FORMATS[idx - 1][1] + + report_html = None + report_pdf = None + if fmt != "none": + console.print( + "\n[grey62]Files save in the folder you launched plexavo from, unless " + "you type a full path.[/grey62]" + ) + if fmt in ("html", "both"): + report_html = Prompt.ask("Name the HTML file", default="report.html", console=console).strip() + if fmt in ("pdf", "both"): + report_pdf = Prompt.ask("Name the PDF file", default="report.pdf", console=console).strip() + + console.print() + key_present = bool(os.environ.get("ANTHROPIC_API_KEY")) + if key_present: + console.print("[bold green]✓ ANTHROPIC_API_KEY detected[/bold green] — full AI narration available.") + else: + console.print("[bold red]✗ ANTHROPIC_API_KEY not set[/bold red] — you'll still get free template " + "remediation for common findings; live AI narration needs a key.") + explain = Confirm.ask( + "Enable full AI narration for every finding? (off still includes free template remediation)", + console=console, default=key_present, + ) + except (KeyboardInterrupt, EOFError): + return None + + return {"report_html": report_html, "report_pdf": report_pdf, "explain": explain} + + +def _confirm_and_run(console: Console, profile: str, identity: dict, session: boto3.Session, options: dict) -> bool: + html_line = os.path.abspath(options["report_html"]) if options["report_html"] else "(skipped)" + pdf_line = os.path.abspath(options["report_pdf"]) if options["report_pdf"] else "(skipped)" + summary = ( + f"Profile: {profile}\n" + f"Account: {identity['Account']}\n" + f"Region: {session.region_name}\n" + f"HTML report: {html_line}\n" + f"PDF report: {pdf_line}\n" + f"AI narration: {'on' if options['explain'] else 'off'}" + ) + console.print() + console.print(Panel(summary, title="Ready to scan", border_style="grey50", box=box.ROUNDED)) + + try: + return Confirm.ask("Run scan now?", console=console, default=True) + except (KeyboardInterrupt, EOFError): + return False + + +def run_interactive() -> None: + """Entry point for bare `plexavo` in a real terminal. + + Slice 1: splash screen. Slice 2: profile picker + live status check. + Slice 3: new-profile write flow. Slice 4: report options + confirm, + wired into the existing cli._run_scan (no duplicated scan logic — + interactive mode drives the exact same code path `plexavo scan` + does, just fed answers gathered via prompts instead of flags). + """ + console = Console() + show_splash(console) + + while True: + choice = _prompt_profile_menu(console) + if choice is None: + console.print("[dim]Cancelled.[/dim]") + return + + if choice == NEW_PROFILE_CHOICE: + choice = _prompt_new_profile(console) + if choice is None: + continue + + console.print() + result = _check_profile(console, choice) + if result is None: + continue + + session, identity = result + break + + profile = choice + while True: + options = _prompt_report_options(console) + if options is None: + console.print("[dim]Cancelled.[/dim]") + return + + if _confirm_and_run(console, profile, identity, session, options): + break + # Declined — loop back and let them re-pick report options. + + console.print() + from plexavo.cli import _run_scan # lazy: avoids a top-level circular import with cli.py + import argparse + + _run_scan(argparse.Namespace( + profile=profile, + region=None, + explain=options["explain"], + explain_limit=25, + report_html=options["report_html"], + report_pdf=options["report_pdf"], + )) diff --git a/plexavo/report/ai_narration.py b/plexavo/report/ai_narration.py index b6d4ed3..6f606fe 100644 --- a/plexavo/report/ai_narration.py +++ b/plexavo/report/ai_narration.py @@ -24,6 +24,8 @@ installed SDK before writing this, not assumed. """ +from __future__ import annotations + import re from dataclasses import dataclass @@ -385,17 +387,23 @@ def _template_enc29(f: Finding) -> Explanation: } -def explain_finding(finding: Finding, client=None) -> Explanation: - """Route to a templated explanation (zero API cost) for the 10 most - common, narratively-generic check types, or call Claude for findings - where account-specific chain/blast-radius reasoning genuinely varies - (which role got chained to, which account is external, which - specific root action fired, etc.). +def explain_finding(finding: Finding, client=None, use_ai: bool = False) -> Explanation | None: + """use_ai=False (default): route to the free template (zero API cost, + zero network) for the 10 most common, narratively-generic check + types; every other check returns None (raw finding detail only — no + API call is ever attempted in this mode, regardless of whether a key + is configured). This is the always-on baseline — no flag needed to + get free remediation guidance where a template exists. + + use_ai=True: every finding, including the 10 templated ones, is sent + to Claude instead — deliberately bypasses the template shortcut so + "AI narration on" always means fully AI-written content, never a mix + of template and AI. `client` is accepted for testability (inject a fake Anthropic client - to prove the templated path never touches the network, or to test - the API path without spending real tokens). Production code never - needs to pass this — it's built automatically from ANTHROPIC_API_KEY. + to test the API path without spending real tokens). Production code + never needs to pass this — it's built automatically from + ANTHROPIC_API_KEY. Deliberately broad exception handling: an AI explanation is enrichment on top of an already-valid, already-computed finding — @@ -405,12 +413,14 @@ def explain_finding(finding: Finding, client=None) -> Explanation: actual AWS state (where silently swallowing an unexpected error could hide a real detection bug). """ - template_fn = COMMON_CHECK_TEMPLATES.get(finding.check_id) - if template_fn: - result = template_fn(finding) - result.confidence = finding.confidence - result.evidence = finding.evidence - return result + if not use_ai: + template_fn = COMMON_CHECK_TEMPLATES.get(finding.check_id) + if template_fn: + result = template_fn(finding) + result.confidence = finding.confidence + result.evidence = finding.evidence + return result + return None try: if client is None: diff --git a/pyproject.toml b/pyproject.toml index 3238e64..bdfdabf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "plexavo" -version = "0.1.2" +version = "0.2.0" description = "Open-source AWS misconfiguration scanner — runs with your own local AWS credentials, nothing sent to anyone else." readme = "README.md" license = { text = "AGPL-3.0-or-later" } diff --git a/tests/test_ai_narration_offline.py b/tests/test_ai_narration_offline.py index 21cc5a5..68009fc 100644 --- a/tests/test_ai_narration_offline.py +++ b/tests/test_ai_narration_offline.py @@ -98,6 +98,22 @@ def create(self, **kwargs): result = explain_finding(finding, client=ExplodingClient()) assert_true("lab-admin" in result.impact, "The real resource name appears in the templated text") +print("\n=== use_ai=False (default): non-templated check returns None, zero network ===") +finding = f("IAM-05") # NOT in COMMON_CHECK_TEMPLATES +result = explain_finding(finding, client=ExplodingClient()) # use_ai=False is the default +assert_true(result is None, f"No template + AI off -> None, no API call attempted (got {result!r})") + +print("\n=== use_ai=True bypasses the template entirely, even for a templated check_id ===") +fake_response_for_iam01 = ( + "IMPACT: This policy grants unrestricted access to every resource.\n\n" + "HOW TO FIX: Replace the wildcard with least-privilege actions.\n\n" + "NEXT STEP: Detach the policy immediately." +) +finding = f("IAM-01") # IS in COMMON_CHECK_TEMPLATES +result = explain_finding(finding, client=FakeClient(response_text=fake_response_for_iam01), use_ai=True) +assert_true(result.source == "api", f"use_ai=True routes a templated check to the API, not the template (got source={result.source})") +assert_true("unrestricted access to every resource" in result.impact, "The AI response is used verbatim, not the template's fixed text") + print("\n=== _parse_sections: well-formatted plain output ===") raw = ("IMPACT: The bucket is public. Calls s3:GetObject to read everything.\n\n" "HOW TO FIX: Run aws s3api put-public-access-block ...\n\n" @@ -133,7 +149,7 @@ def create(self, **kwargs): "NEXT STEP: Edit the policy's Resource field now." ) finding = f("IAM-05") # NOT in COMMON_CHECK_TEMPLATES — must go through the API path -result = explain_finding(finding, client=FakeClient(response_text=fake_response_text)) +result = explain_finding(finding, client=FakeClient(response_text=fake_response_text), use_ai=True) assert_true(result.source == "api", f"Non-templated check routes to the API path (got source={result.source})") assert_true(result.impact == "This role can assume an admin role. Calls sts:AssumeRole on the target role.", "API response parsed correctly end-to-end") assert_true(result.next_step == "Edit the policy's Resource field now.", "next_step parsed correctly from the API response") @@ -150,7 +166,7 @@ def create(self, **kwargs): "text": "IMPACT: x\n\nHOW TO FIX: y\n\nNEXT STEP: z", })() finding = f("IAM-05") -result = explain_finding(finding, client=FakeClient(response_blocks=[thinking_block, text_block])) +result = explain_finding(finding, client=FakeClient(response_blocks=[thinking_block, text_block]), use_ai=True) assert_true(result.source == "api", f"Correctly extracts text past a leading ThinkingBlock (got source={result.source})") assert_true(result.impact == "x", f"Text content correctly parsed despite the thinking block (got: {result.impact!r})") @@ -198,17 +214,17 @@ def create(self, **kwargs): print("\n=== REGRESSION: truncated response (stop_reason=max_tokens) is flagged, not silently returned ===") truncated_text = "IMPACT: x\n\nHOW TO FIX: Step 1: do this\naws iam delete-role-policy \\\n --role-name" finding = f("IAM-05") -result = explain_finding(finding, client=FakeClient(response_text=truncated_text, stop_reason="max_tokens")) +result = explain_finding(finding, client=FakeClient(response_text=truncated_text, stop_reason="max_tokens"), use_ai=True) assert_true("TRUNCATED" in result.how_to_fix, f"Truncation is visibly flagged, not silently returned as if complete (got: {result.how_to_fix!r})") print("\n=== FALSE POSITIVE GUARD: normal, complete response does NOT get flagged as truncated ===") finding = f("IAM-05") -result = explain_finding(finding, client=FakeClient(response_text=fake_response_text, stop_reason="end_turn")) +result = explain_finding(finding, client=FakeClient(response_text=fake_response_text, stop_reason="end_turn"), use_ai=True) assert_true("TRUNCATED" not in result.how_to_fix, "A normal, complete response is not falsely flagged") print("\n=== API path: exception during the call -> graceful fallback, not a crash ===") finding = f("IAM-05", raw_detail="original technical detail here", confidence="Likely — see note", evidence="some evidence") -result = explain_finding(finding, client=FakeClient(raise_exc=ConnectionError("simulated network failure"))) +result = explain_finding(finding, client=FakeClient(raise_exc=ConnectionError("simulated network failure")), use_ai=True) assert_true(result.source == "fallback", f"A failed API call falls back gracefully (got source={result.source})") assert_true(result.impact == "original technical detail here", "Fallback preserves the original raw_detail, doesn't lose the finding") # Per the OSS plan's §3.3 contract: the fallback must be silent in the @@ -243,7 +259,7 @@ def _blocked_import(name, *args, **kwargs): _builtins.__import__ = _blocked_import try: finding = f("IAM-05", raw_detail="original technical detail here") - result = explain_finding(finding) + result = explain_finding(finding, use_ai=True) finally: _builtins.__import__ = _real_import assert_true(result.source == "fallback", f"Missing anthropic package falls back, doesn't crash (got source={result.source})")