# Dependency Installation
!pip install --upgrade pandas requests matplotlib seaborn

# Library Imports
import logging
import os
import random
import time
from datetime import datetime, timedelta, UTC
from collections import deque

import pandas as pd
import requests
import matplotlib.pyplot as plt
import seaborn as sns

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# Suppress specific warnings from libraries if they become too noisy
# For example, to suppress FutureWarnings from seaborn:
import warnings
warnings.filterwarnings("ignore", category=FutureWarning)

def create_audit_state_crypto(exchange_name: str = "Simulated Exchange") -> dict:
    """
    Initializes the state dictionary for the cryptocurrency exchange API audit.

    Parameters
    ----------
    exchange_name : str, optional
        The name of the cryptocurrency exchange being audited, defaults to "Simulated Exchange".

    Returns
    -------
    dict
        An initialized state dictionary.
    """
    logger.info(f"Initializing audit state for exchange: {exchange_name}")
    state = {
        "exchange_name": exchange_name,
        "api_keys": [], # List of dictionaries, each representing an API key
        "audit_findings": [],
        "log_entries": deque(maxlen=1000) # For capturing audit-specific logs
    }
    logger.debug("Crypto audit state initialized successfully.")
    return state

def get_simulated_api_keys(state: dict) -> dict:
    """
    Simulates fetching API key information for demonstration purposes.

    Parameters
    ----------
    state : dict
        The current audit state dictionary.

    Returns
    -------
    dict
        The updated state dictionary with simulated API key details.
    """
    logger.info("Simulating fetching API key information.")
    exchange = state["exchange_name"]

    simulated_keys = [
        {
            "key_id": "key_abc123",
            "label": "Main Trading Bot Key",
            "permissions": ["SPOT_TRADE", "MARGIN_TRADE", "READ_ACCOUNT_INFO"],
            "ip_restricted": True,
            "whitelisted_ips": ["192.168.1.1", "10.0.0.5"],
            "can_withdraw": False,
            "created_at": (datetime.now(UTC) - timedelta(days=random.randint(5, 30))).isoformat() # Recent key
        },
        {
            "key_id": "key_def456",
            "label": "Portfolio Tracker Read-Only",
            "permissions": ["READ_ACCOUNT_INFO"],
            "ip_restricted": False, # Less secure
            "whitelisted_ips": [],
            "can_withdraw": False,
            "created_at": (datetime.now(UTC) - timedelta(days=random.randint(90, 180))).isoformat() # Older key
        },
        {
            "key_id": "key_ghi789",
            "label": "Emergency Withdrawal Key",
            "permissions": ["READ_ACCOUNT_INFO", "WITHDRAW"],
            "ip_restricted": True,
            "whitelisted_ips": ["203.0.113.10"],
            "can_withdraw": True,
            "created_at": (datetime.now(UTC) - timedelta(days=random.randint(300, 400))).isoformat() # Very old key
        },
        {
            "key_id": "key_jkl012",
            "label": "Test Key - Overly Permissive",
            "permissions": ["SPOT_TRADE", "MARGIN_TRADE", "READ_ACCOUNT_INFO", "WITHDRAW", "TRANSFER"],
            "ip_restricted": False,
            "whitelisted_ips": [],
            "can_withdraw": True,
            "created_at": (datetime.now(UTC) - timedelta(days=random.randint(1, 10))).isoformat() # New, but bad
        }
    ]
    state["api_keys"] = simulated_keys
    logger.info(f"Simulated {len(simulated_keys)} API keys for {exchange}.")
    return state

def audit_api_key_lifespan(state: dict, max_key_lifespan_days: int = 90) -> dict:
    """
    Audits the lifespan of API keys against a recommended maximum age.

    Parameters
    ----------
    state : dict
        The current audit state dictionary.
    max_key_lifespan_days : int, optional
        The maximum allowed lifespan for an API key in days, defaults to 90.

    Returns
    -------
    dict
        The updated state dictionary with audit findings related to key age.
    """
    logger.info(f"Auditing API key lifespans (max age: {max_key_lifespan_days} days).")
    current_time = datetime.now(UTC)
    findings = []

    for key in state.get("api_keys", []):
        key_id = key.get("key_id")
        created_at_str = key.get("created_at")

        if created_at_str:
            try:
                created_at = datetime.fromisoformat(created_at_str.replace('Z', '+00:00')).replace(tzinfo=UTC)
                key_age = (current_time - created_at).days

                if key_age > max_key_lifespan_days:
                    findings.append({
                        "type": "alert",
                        "category": "Key Lifespan",
                        "key_id": key_id,
                        "label": key.get("label"),
                        "message": f"API key is {key_age} days old, exceeding recommended {max_key_lifespan_days} days. Consider rotation."
                    })
                    logger.warning(f"Key {key_id} ({key.get('label')}) is {key_age} days old.")
                else:
                    logger.debug(f"Key {key_id} ({key.get('label')}) is {key_age} days old (within limits).")
            except ValueError as e:
                findings.append({
                    "type": "error",
                    "category": "Key Lifespan",
                    "key_id": key_id,
                    "label": key.get("label"),
                    "message": f"Could not parse created_at for key: {e}"
                })
                logger.error(f"Error parsing key creation time for {key_id}: {e}")
        else:
            findings.append({
                "type": "warning",
                "category": "Key Lifespan",
                "key_id": key_id,
                "label": key.get("label"),
                "message": "API key lacks creation timestamp. Cannot audit lifespan."
            })
            logger.warning(f"Key {key_id} lacks creation timestamp.")

    state["audit_findings"].extend(findings)
    logger.info(f"Completed key lifespan audit. Found {len([f for f in findings if f['type'] == 'alert'])} old keys.")
    return state

def audit_least_privilege_crypto(state: dict,
                                 sensitive_permissions: list = ["WITHDRAW", "TRANSFER"],
                                 max_permissions_per_key: int = 3) -> dict:
    """
    Audits API key permissions for least privilege violations.

    Parameters
    ----------
    state : dict
        The current audit state dictionary.
    sensitive_permissions : list, optional
        A list of permission strings considered highly sensitive, defaults to ["WITHDRAW", "TRANSFER"].
    max_permissions_per_key : int, optional
        The maximum number of permissions an API key should ideally have, defaults to 3.

    Returns
    -------
    dict
        The updated state dictionary with audit findings related to least privilege.
    """
    logger.info("Auditing API key permissions for least privilege violations.")
    findings = []

    for key in state.get("api_keys", []):
        key_id = key.get("key_id")
        label = key.get("label")
        permissions = key.get("permissions", [])
        ip_restricted = key.get("ip_restricted", False)

        # Check for sensitive permissions without IP restriction
        for perm in permissions:
            if perm in sensitive_permissions and not ip_restricted:
                findings.append({
                    "type": "alert",
                    "category": "Least Privilege",
                    "key_id": key_id,
                    "label": label,
                    "message": f"API key has sensitive permission '{perm}' but is not IP-restricted. High risk."
                })
                logger.warning(f"Key {key_id} has {perm} without IP restriction.")

        # Check for excessive number of permissions
        if len(permissions) > max_permissions_per_key:
            findings.append({
                "type": "warning",
                "category": "Least Privilege",
                "key_id": key_id,
                "label": label,
                "message": f"API key has {len(permissions)} permissions, exceeding recommended max of {max_permissions_per_key}. Review for over-privilege."
            })
            logger.warning(f"Key {key_id} has {len(permissions)} permissions, potentially over-privileged.")

        logger.debug(f"Audited permissions for key {key_id} ({label}). Permissions: {', '.join(permissions)}.")

    state["audit_findings"].extend(findings)
    logger.info(f"Completed least privilege audit. Found {len([f for f in findings if f['type'] == 'alert'])} high-risk privilege issues.")
    return state

def simulate_exchange_api_call(state: dict, key_id: str, action: str) -> dict:
    """
    Simulates an API call to check if an API key has specific permissions.

    Parameters
    ----------
    state : dict
        The current audit state dictionary.
    key_id : str
        The ID of the API key to simulate with.
    action : str
        The action to simulate (e.g., "SPOT_TRADE", "WITHDRAW", "READ_ACCOUNT_INFO").

    Returns
    -------
    dict
        The updated state with findings on permission checks.
    """
    logger.info(f"Simulating API call for key {key_id} to perform action: {action}.")

    key_details = next((k for k in state["api_keys"] if k["key_id"] == key_id), None)
    if not key_details:
        findings_message = f"Simulated API call failed: API key {key_id} not found in state."
        findings_type = "error"
        logger.error(findings_message)
    else:
        permissions = key_details.get("permissions", [])

        if action in permissions:
            findings_message = f"Simulated API call for key {key_id} to '{action}' succeeded. Permissions appear effective."
            findings_type = "info"
            logger.info(findings_message)
        else:
            findings_message = f"Simulated API call for key {key_id} to '{action}' failed. Permissions are insufficient."
            findings_type = "alert"
            logger.warning(findings_message)

    state["audit_findings"].append({
        "type": findings_type,
        "category": "Permission Check",
        "key_id": key_id,
        "label": key_details.get("label") if key_details else "N/A",
        "action": action,
        "message": findings_message
    })
    return state

def summarize_audit_findings_crypto(state: dict) -> dict:
    """
    Summarizes all audit findings into a pandas DataFrame.

    Parameters
    ----------
    state : dict
        The current audit state dictionary containing all audit findings.

    Returns
    -------
    dict
        The updated state dictionary, with a summary DataFrame added.
    """
    logger.info("Summarizing all audit findings.")
    findings = state.get("audit_findings", [])

    if not findings:
        logger.info("No audit findings to summarize.")
        state["findings_summary_df"] = pd.DataFrame(columns=["Type", "Category", "Key ID", "Label", "Action", "Message"])
        state["findings_count_by_type"] = {}
        return state

    findings_df = pd.DataFrame(findings)
    state["findings_summary_df"] = findings_df
    logger.info(f"Summarized {len(findings)} findings.")

    # Add a count of findings by type for quick overview
    state["findings_count_by_type"] = findings_df.groupby('type').size().to_dict()
    logger.info(f"Findings count by type: {state['findings_count_by_type']}")

    return state

import logging # Added for robustness
from datetime import datetime, timedelta, UTC # Added for robustness

def run_audit_demonstration_crypto(exchange_name: str = "Simulated Exchange") -> dict:
    """
    Runs a complete cryptocurrency exchange API audit demonstration using simulated data.

    Parameters
    ----------
    exchange_name : str, optional
        The name of the cryptocurrency exchange to simulate, defaults to "Simulated Exchange".

    Returns
    -------
    dict
        The final state dictionary with all audit findings and summaries.
    """
    local_logger = logging.getLogger(__name__)
    local_logger.info(f"Starting Crypto Exchange API Permissions Audit Demonstration for {exchange_name}.")

    state = create_audit_state_crypto(exchange_name=exchange_name)

    try:
        # Step 1: Simulate fetching API keys
        state = get_simulated_api_keys(state)

        # Step 2: Audit API Key Lifespan
        state = audit_api_key_lifespan(state, max_key_lifespan_days=90)

        # Step 3: Audit Least Privilege
        state = audit_least_privilege_crypto(state, sensitive_permissions=["WITHDRAW", "TRANSFER"], max_permissions_per_key=3)

        # Step 4: Simulate API Call Permission Checks
        # Example 1: Check if 'Main Trading Bot Key' can SPOT_TRADE (should succeed)
        state = simulate_exchange_api_call(state, "key_abc123", "SPOT_TRADE")
        # Example 2: Check if 'Portfolio Tracker Read-Only' can WITHDRAW (should fail)
        state = simulate_exchange_api_call(state, "key_def456", "WITHDRAW")
        # Example 3: Check if 'Emergency Withdrawal Key' can WITHDRAW (should succeed)
        state = simulate_exchange_api_call(state, "key_ghi789", "WITHDRAW")
        # Example 4: Check if 'Emergency Withdrawal Key' can SPOT_TRADE (should fail if not granted)
        state = simulate_exchange_api_call(state, "key_ghi789", "SPOT_TRADE")

        # Step 5: Summarize Findings
        state = summarize_audit_findings_crypto(state)

    except Exception as e:
        local_logger.critical(f"An unexpected critical error occurred during audit: {e}")
        state["audit_findings"].append({"type": "critical", "message": f"Audit interrupted by critical error: {e}"})

    local_logger.info("Crypto Exchange API Permissions Audit Demonstration Finished.")
    return state


# Execute the demonstration
# Ensure all preceding function definition cells (from 'Library Imports' onwards)
# have been executed to define all necessary functions and imports.
final_crypto_audit_state = run_audit_demonstration_crypto()

if "findings_summary_df" in final_crypto_audit_state:
    print("\n--- Crypto Audit Findings Summary ---")
    display(final_crypto_audit_state["findings_summary_df"])
else:
    print("No audit findings DataFrame available.")

if "findings_count_by_type" in final_crypto_audit_state and final_crypto_audit_state["findings_count_by_type"]:
    findings_counts = pd.Series(final_crypto_audit_state["findings_count_by_type"])

    plt.figure(figsize=(10, 6))
    sns.barplot(x=findings_counts.index, y=findings_counts.values, palette='viridis')
    plt.title('Distribution of Crypto Audit Findings by Type')
    plt.xlabel('Finding Type')
    plt.ylabel('Number of Findings')
    plt.grid(axis='y', linestyle='--', alpha=0.7)
    plt.show()
else:
    print("No findings to visualize by type.")

from datetime import datetime, timedelta, UTC # Added for robustness
import logging # Added for robustness
import pandas as pd # Added for robustness
import matplotlib.pyplot as plt # Added for robustness
import seaborn as sns # Added for robustness

key_lifespans = []
current_time = datetime.now(UTC)

# Need to ensure final_crypto_audit_state is defined from the previous cell's execution
# If it's not defined due to previous errors, this block will still show 'No API keys found...'
if 'final_crypto_audit_state' in locals() or 'final_crypto_audit_state' in globals():
    for key in final_crypto_audit_state.get("api_keys", []):
        created_at_str = key.get("created_at")
        if created_at_str:
            try:
                created_at = datetime.fromisoformat(created_at_str.replace('Z', '+00:00')).replace(tzinfo=UTC)
                key_age = (current_time - created_at).days
                key_lifespans.append(key_age)
            except ValueError:
                # Ensure logger is available for this warning
                if 'logger' not in locals() and 'logger' not in globals():
                    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
                    logger = logging.getLogger(__name__)
                logger.warning(f"Could not parse date for key {key.get('key_id')}")

if key_lifespans:
    plt.figure(figsize=(10, 6))
    sns.histplot(key_lifespans, bins=15, kde=True, color='skyblue')
    plt.title('Distribution of Crypto API Key Lifespans')
    plt.xlabel('Key Age (Days)')
    plt.ylabel('Number of Keys')
    plt.axvline(x=90, color='r', linestyle='--', label='Recommended Max Lifespan (90 Days)')
    plt.legend()
    plt.grid(axis='y', linestyle='--', alpha=0.7)
    plt.show()
else:
    print("No API keys found or processed for lifespan visualization.")