scripts/cfi_cli.py

#!/usr/bin/env python3
"""cfi-cli: Bank Onboarding CLI & Self-Service Integration Sandbox.

Provides automated tooling for bank IT teams to onboard, test gRPC connectivity,
generate mTLS certificates, and verify local hardware compatibility within minutes.

Usage:
    python scripts/cfi_cli.py init --bank-id bank_alpha --coordinator localhost:50051
    python scripts/cfi_cli.py cert generate-csr --bank-id bank_alpha --output-dir ./certs
    python scripts/cfi_cli.py test-connection --host localhost --port 50051
    python scripts/cfi_cli.py sandbox run --transactions 1000
"""

from __future__ import annotations

import argparse
import json
import logging
import random
import socket
import sys
import time
from pathlib import Path
from typing import Any

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [cfi-cli] %(levelname)s %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%S",
)
logger = logging.getLogger("cfi_cli")

# ---------------------------------------------------------------------------
# Exit codes
# ---------------------------------------------------------------------------
EXIT_OK = 0
EXIT_FAIL = 1

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

BANK_CONFIG_TEMPLATE = """\
# cfi-bank-client configuration — generated by cfi-cli init
# Edit fields marked with <CHANGE> before deploying.

bank_id: "{bank_id}"
bank_name: "<CHANGE: Full legal bank name>"
coordinator_host: "{coordinator_host}"
coordinator_port: {coordinator_port}

mtls:
  enabled: true
  client_cert_path: "certs/bank.crt"          # Signed certificate from consortium CA
  client_key_path:  "certs/bank.key"           # RSA private key (4096-bit)
  ca_cert_path:     "certs/consortium_ca.crt"  # Root CA certificate from coordinator

vault:
  dir: "data/vault"
  passphrase: "<CHANGE: Strong passphrase for local encrypted storage>"

reconnect:
  max_retries: 10
  initial_backoff_sec: 1.0
  max_backoff_sec: 60.0
"""


def _print_result(result: dict[str, Any], *, indent: int = 2) -> None:
    """Print structured JSON result to stdout."""
    print(json.dumps(result, indent=indent))


# ---------------------------------------------------------------------------
# Command: init
# ---------------------------------------------------------------------------


def cmd_init(args: argparse.Namespace) -> int:
    """Generate local bank configuration template and directory scaffold."""
    output_dir = Path(args.output_dir)
    coordinator_host, _, coordinator_port_str = args.coordinator.partition(":")
    coordinator_port = int(coordinator_port_str) if coordinator_port_str else 50051

    dirs_to_create = [
        output_dir / "data" / "vault",
        output_dir / "certs",
        output_dir / "logs",
    ]
    created_dirs: list[str] = []
    for d in dirs_to_create:
        d.mkdir(parents=True, exist_ok=True)
        created_dirs.append(str(d))
        logger.info("Created directory: %s", d)

    config_path = output_dir / "bank_config.yaml"
    config_content = BANK_CONFIG_TEMPLATE.format(
        bank_id=args.bank_id,
        coordinator_host=coordinator_host,
        coordinator_port=coordinator_port,
    )
    config_path.write_text(config_content, encoding="utf-8")
    logger.info("Config template written: %s", config_path)

    result: dict[str, Any] = {
        "status": "ok",
        "command": "init",
        "bank_id": args.bank_id,
        "output_dir": str(output_dir),
        "created_directories": created_dirs,
        "config_file": str(config_path),
        "next_step": "Edit bank_config.yaml then run: cfi-cli cert generate-csr",
    }
    _print_result(result)
    return EXIT_OK


# ---------------------------------------------------------------------------
# Command: cert generate-csr
# ---------------------------------------------------------------------------


def _generate_rsa_key_and_csr(
    bank_id: str,
    output_dir: Path,
    country: str = "US",
    org: str = "CFI Consortium",
) -> tuple[Path, Path]:
    """Generate 4096-bit RSA private key and X.509 CSR using cryptography library."""
    try:
        from cryptography import x509  # noqa: PLC0415
        from cryptography.hazmat.primitives import hashes, serialization  # noqa: PLC0415
        from cryptography.hazmat.primitives.asymmetric import rsa  # noqa: PLC0415
        from cryptography.x509.oid import NameOID  # noqa: PLC0415
    except ImportError as exc:
        logger.error("cryptography library not found: %s", exc)
        raise

    key = rsa.generate_private_key(public_exponent=65537, key_size=4096)

    key_path = output_dir / "bank.key"
    key_path.write_bytes(
        key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm=serialization.NoEncryption(),
        )
    )
    logger.info("RSA 4096-bit private key written: %s", key_path)

    csr = (
        x509.CertificateSigningRequestBuilder()
        .subject_name(
            x509.Name(
                [
                    x509.NameAttribute(NameOID.COUNTRY_NAME, country),
                    x509.NameAttribute(NameOID.ORGANIZATION_NAME, org),
                    x509.NameAttribute(NameOID.COMMON_NAME, bank_id),
                ]
            )
        )
        .sign(key, hashes.SHA256())
    )

    csr_path = output_dir / "bank.csr"
    csr_path.write_bytes(csr.public_bytes(serialization.Encoding.PEM))
    logger.info("X.509 CSR written: %s", csr_path)

    return key_path, csr_path


def cmd_cert_generate_csr(args: argparse.Namespace) -> int:
    """Generate 4096-bit RSA private key and X.509 CSR for mTLS authentication."""
    output_dir = Path(args.output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    try:
        key_path, csr_path = _generate_rsa_key_and_csr(
            bank_id=args.bank_id,
            output_dir=output_dir,
            country=args.country,
            org=args.org,
        )
    except Exception as exc:
        result: dict[str, Any] = {
            "status": "error",
            "command": "cert generate-csr",
            "error": str(exc),
        }
        _print_result(result)
        return EXIT_FAIL

    result = {
        "status": "ok",
        "command": "cert generate-csr",
        "bank_id": args.bank_id,
        "key_file": str(key_path),
        "csr_file": str(csr_path),
        "key_bits": 4096,
        "signature_algorithm": "SHA256withRSA",
        "next_step": (
            f"Submit {csr_path} to your consortium CA administrator for signing. "
            "Place the signed bank.crt and consortium_ca.crt in the certs/ directory."
        ),
    }
    _print_result(result)
    return EXIT_OK


# ---------------------------------------------------------------------------
# Command: test-connection
# ---------------------------------------------------------------------------


def _probe_tcp(host: str, port: int, timeout: float = 5.0) -> tuple[bool, float]:
    """Attempt TCP connection and return (success, latency_ms)."""
    start = time.perf_counter()
    try:
        with socket.create_connection((host, port), timeout=timeout):
            latency_ms = round((time.perf_counter() - start) * 1000, 2)
            return True, latency_ms
    except OSError:
        latency_ms = round((time.perf_counter() - start) * 1000, 2)
        return False, latency_ms


def cmd_test_connection(args: argparse.Namespace) -> int:
    """Verify outbound gRPC reachability, mTLS handshake, and coordinator latency."""
    host: str = args.host
    port: int = args.port
    timeout: float = args.timeout

    logger.info("Probing coordinator %s:%d (timeout=%.1fs)…", host, port, timeout)

    reachable, latency_ms = _probe_tcp(host, port, timeout=timeout)

    latency_sla = "PASS" if latency_ms < 100.0 else "WARN"

    result: dict[str, Any] = {
        "status": "ok" if reachable else "error",
        "command": "test-connection",
        "coordinator": f"{host}:{port}",
        "tcp_reachable": reachable,
        "latency_ms": latency_ms,
        "latency_sla": latency_sla if reachable else "N/A",
        "mtls_note": (
            "TCP layer reachable — mTLS handshake requires signed certificates. "
            "Run cfi-cli cert generate-csr and install signed certs to test full mTLS."
            if reachable
            else "Host unreachable. Verify coordinator host/port and firewall rules."
        ),
    }

    if not reachable:
        logger.error(
            "Coordinator %s:%d is not reachable (%.2fms). Check host/port and network.",
            host,
            port,
            latency_ms,
        )
        _print_result(result)
        return EXIT_FAIL

    logger.info(
        "Coordinator %s:%d reachable — latency %.2fms [SLA: %s]",
        host,
        port,
        latency_ms,
        latency_sla,
    )
    _print_result(result)
    return EXIT_OK


# ---------------------------------------------------------------------------
# Command: sandbox run
# ---------------------------------------------------------------------------


def _generate_synthetic_transactions(count: int) -> list[dict[str, Any]]:
    """Generate synthetic normalized transaction records for sandbox benchmarking."""
    transactions = []
    for i in range(count):
        transactions.append(
            {
                "transaction_id": f"sandbox_tx_{i:06d}",
                "account_id": f"acc_{random.randint(1000, 9999)}",
                "counterparty_account_id": f"acc_{random.randint(1000, 9999)}",
                "amount": round(random.uniform(10.0, 50000.0), 2),
                "currency": random.choice(["USD", "EUR", "GBP", "TRY"]),
                "is_fraud": random.random() < 0.02,
            }
        )
    return transactions


def _detect_pytorch_hardware() -> dict[str, Any]:
    """Detect PyTorch availability and hardware acceleration."""
    try:
        import torch  # noqa: PLC0415

        cuda_available = torch.cuda.is_available()
        cuda_device = torch.cuda.get_device_name(0) if cuda_available else None
        mps_available = getattr(torch.backends, "mps", None) and torch.backends.mps.is_available()

        return {
            "pytorch_available": True,
            "pytorch_version": torch.__version__,
            "cuda_available": cuda_available,
            "cuda_device": cuda_device,
            "mps_available": bool(mps_available),
            "recommended_device": (
                "cuda" if cuda_available else ("mps" if mps_available else "cpu")
            ),
        }
    except ImportError:
        return {
            "pytorch_available": False,
            "pytorch_version": None,
            "cuda_available": False,
            "cuda_device": None,
            "mps_available": False,
            "recommended_device": "N/A",
        }


def cmd_sandbox_run(args: argparse.Namespace) -> int:
    """Launch local synthetic transaction benchmark and GPU compatibility check."""
    tx_count: int = args.transactions
    logger.info("Sandbox: generating %d synthetic transactions…", tx_count)

    t0 = time.perf_counter()
    transactions = _generate_synthetic_transactions(tx_count)
    gen_elapsed = time.perf_counter() - t0

    t1 = time.perf_counter()
    fraud_count = sum(1 for tx in transactions if tx["is_fraud"])
    process_elapsed = time.perf_counter() - t1

    total_elapsed = gen_elapsed + process_elapsed
    throughput_tps = round(tx_count / total_elapsed, 1) if total_elapsed > 0 else float("inf")

    hardware_info = _detect_pytorch_hardware()

    logger.info(
        "Sandbox complete: %d tx in %.3fs — throughput %.1f TPS | PyTorch %s | Device: %s",
        tx_count,
        total_elapsed,
        throughput_tps,
        hardware_info.get("pytorch_version", "N/A"),
        hardware_info.get("recommended_device", "N/A"),
    )

    result: dict[str, Any] = {
        "status": "ok",
        "command": "sandbox run",
        "transactions_generated": tx_count,
        "fraud_transactions": fraud_count,
        "fraud_rate_pct": round(fraud_count / tx_count * 100, 2),
        "generation_elapsed_sec": round(gen_elapsed, 4),
        "processing_elapsed_sec": round(process_elapsed, 4),
        "total_elapsed_sec": round(total_elapsed, 4),
        "throughput_tps": throughput_tps,
        "throughput_sla": "PASS" if throughput_tps >= 100.0 else "WARN",
        "hardware": hardware_info,
        "next_step": (
            "Sandbox validation successful. Deploy cfi-bank-client container and run "
            "cfi-cli test-connection to verify live coordinator connectivity."
        ),
    }

    _print_result(result)
    return EXIT_OK


# ---------------------------------------------------------------------------
# CLI parser
# ---------------------------------------------------------------------------


def build_parser() -> argparse.ArgumentParser:
    """Build the cfi-cli argument parser."""
    parser = argparse.ArgumentParser(
        prog="cfi-cli",
        description="CFI Bank Onboarding CLI — self-service integration tooling for bank IT teams.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  cfi-cli init --bank-id bank_alpha --coordinator coordinator.cfi.internal:50051
  cfi-cli cert generate-csr --bank-id bank_alpha --output-dir ./certs
  cfi-cli test-connection --host coordinator.cfi.internal --port 50051
  cfi-cli sandbox run --transactions 5000
        """,
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    # --- init ---
    init_parser = subparsers.add_parser(
        "init",
        help="Generate local bank configuration template and directory scaffold.",
    )
    init_parser.add_argument(
        "--bank-id",
        default="bank_alpha",
        help="Unique identifier for this bank node (default: bank_alpha)",
    )
    init_parser.add_argument(
        "--coordinator",
        default="localhost:50051",
        help="Coordinator host:port (default: localhost:50051)",
    )
    init_parser.add_argument(
        "--output-dir",
        default=".",
        help="Output directory for generated files (default: current directory)",
    )

    # --- cert ---
    cert_parser = subparsers.add_parser("cert", help="Certificate management commands.")
    cert_sub = cert_parser.add_subparsers(dest="cert_command", required=True)

    csr_parser = cert_sub.add_parser(
        "generate-csr",
        help="Generate 4096-bit RSA private key and X.509 CSR for mTLS authentication.",
    )
    csr_parser.add_argument("--bank-id", required=True, help="Bank node identifier (CN field)")
    csr_parser.add_argument(
        "--output-dir", default="./certs", help="Directory to write key and CSR files"
    )
    csr_parser.add_argument("--country", default="US", help="CSR country code (default: US)")
    csr_parser.add_argument(
        "--org",
        default="CFI Consortium",
        help="CSR organization name (default: CFI Consortium)",
    )

    # --- test-connection ---
    conn_parser = subparsers.add_parser(
        "test-connection",
        help="Verify outbound gRPC latency, mTLS handshake, and coordinator readiness.",
    )
    conn_parser.add_argument(
        "--host", default="localhost", help="Coordinator gRPC host (default: localhost)"
    )
    conn_parser.add_argument(
        "--port", type=int, default=50051, help="Coordinator gRPC port (default: 50051)"
    )
    conn_parser.add_argument(
        "--timeout",
        type=float,
        default=5.0,
        help="Connection timeout in seconds (default: 5.0)",
    )

    # --- sandbox ---
    sandbox_parser = subparsers.add_parser(
        "sandbox",
        help="Self-service integration sandbox commands.",
    )
    sandbox_sub = sandbox_parser.add_subparsers(dest="sandbox_command", required=True)

    run_parser = sandbox_sub.add_parser(  # noqa: F841
        "run",
        help="Launch synthetic transaction benchmark and GPU compatibility check.",
    )
    run_parser.add_argument(
        "--transactions",
        type=int,
        default=1000,
        help="Number of synthetic transactions to generate (default: 1000)",
    )

    return parser


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------


def main() -> int:
    """CLI entry point — dispatches to subcommand handlers."""
    parser = build_parser()
    args = parser.parse_args()

    if args.command == "init":
        return cmd_init(args)
    if args.command == "cert" and getattr(args, "cert_command", None) == "generate-csr":
        return cmd_cert_generate_csr(args)
    if args.command == "test-connection":
        return cmd_test_connection(args)
    if args.command == "sandbox" and getattr(args, "sandbox_command", None) == "run":
        return cmd_sandbox_run(args)

    parser.print_help()
    return EXIT_FAIL


if __name__ == "__main__":
    sys.exit(main())