Blame for env/dynamic_input.py

"""
DynamicInputPipeline — Three-Source Dynamic Scenario Input
===========================================================
Provides a unified interface for three sources of dynamic input that drive
scenario variation in the SOC Simulator:

Source A — Live Threat Intel Feed
    Fetches malicious IPs from URLhaus and Feodo Tracker (both free, no auth).
    Results are cached for 1 hour.  Falls back to hardcoded IPs on failure.

Source B — Performance-Based Difficulty Adaptation
    Tracks blue agent's rolling average score over the last 5 episodes and
    returns difficulty adjustment parameters accordingly.

Source C — Runtime Custom Parameters
    Accepts per-request overrides (attacker_ip, target_user, attack_intensity,
    enable_red_agent, use_live_threat_intel) that merge on top of scenario
    defaults.

Usage
-----
pipeline = DynamicInputPipeline()

# Source A
threat_ips = pipeline.get_live_threat_ips()

# Source B
params = pipeline.adapt_difficulty([0.3, 0.5, 0.6, 0.45, 0.55])

# Source C — typically called by the server layer
merged = pipeline.merge_custom_params(base_scenario_params, custom_params_dict)
"""

from __future__ import annotations

import json
import logging
import time
from typing import Any, Dict, List, Optional

import requests

logger = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

URLHAUS_API_URL = "https://urlhaus-api.abuse.ch/v1/urls/recent/"
FEODO_BLOCKLIST_URL = (
    "https://feodotracker.abuse.ch/downloads/ipblocklist.json"
)

CACHE_TTL_SECONDS = 3600  # 1 hour

# Hardcoded fallback — realistic-looking attacker IPs from documentation ranges
FALLBACK_THREAT_IPS: List[str] = [
    "198.51.100.10",
    "198.51.100.35",
    "198.51.100.77",
    "203.0.113.5",
    "203.0.113.42",
    "203.0.113.99",
    "198.51.100.120",
    "203.0.113.150",
]

# Difficulty adaptation thresholds
LOW_SCORE_THRESHOLD = 0.40
HIGH_SCORE_THRESHOLD = 0.75


# ---------------------------------------------------------------------------
# DynamicInputPipeline
# ---------------------------------------------------------------------------


class DynamicInputPipeline:
    """
    Unified pipeline for three sources of dynamic scenario input.

    Thread Safety
    -------------
    The class keeps a simple in-memory cache.  For single-threaded or
    asyncio-based usage (FastAPI) this is safe.  For multi-process deployment
    consider an external cache (Redis).
    """

    def __init__(
        self,
        cache_ttl: int = CACHE_TTL_SECONDS,
        http_timeout: int = 10,
    ) -> None:
        """
        Parameters
        ----------
        cache_ttl : int
            Number of seconds to cache threat-intel results (default 3600).
        http_timeout : int
            HTTP request timeout in seconds (default 10).
        """
        self._cache_ttl = cache_ttl
        self._http_timeout = http_timeout

        # Source A — threat intel cache
        self._threat_ip_cache: List[str] = []
        self._cache_fetched_at: float = 0.0  # epoch seconds

        # Source B — episode score history (stored internally)
        self._score_history: List[float] = []

    # =========================================================================
    # SOURCE A — Live Threat Intel
    # =========================================================================

    def get_live_threat_ips(self) -> List[str]:
        """
        Return a deduplicated list of malicious IPs from live threat feeds.

        Sources:
        - URLhaus API (recent malware URLs → extract host IPs)
        - Feodo Tracker IP blocklist (JSON)

        Results are cached for ``cache_ttl`` seconds.  On any network /
        parsing failure the method falls back to ``FALLBACK_THREAT_IPS``.

        Returns
        -------
        List[str]
            Deduplicated list of malicious IP strings.
        """
        now = time.monotonic()
        if (
            self._threat_ip_cache
            and (now - self._cache_fetched_at) < self._cache_ttl
        ):
            logger.debug("Returning cached threat IPs (%d entries).", len(self._threat_ip_cache))
            return list(self._threat_ip_cache)

        ips: List[str] = []

        # ── URLhaus ──────────────────────────────────────────────────────────
        try:
            resp = requests.post(
                URLHAUS_API_URL,
                data={"limit": 100},
                timeout=self._http_timeout,
            )
            resp.raise_for_status()
            payload = resp.json()
            for url_entry in payload.get("urls", []):
                host = url_entry.get("host", "")
                # Only keep plain IP addresses (not domain names)
                if host and self._is_ipv4(host):
                    ips.append(host)
            logger.info("URLhaus: fetched %d malicious IPs.", len(ips))
        except Exception as exc:  # noqa: BLE001
            logger.debug("URLhaus fetch failed (typically requires API key now) — %s", exc)

        # ── Feodo Tracker ────────────────────────────────────────────────────
        try:
            resp = requests.get(
                FEODO_BLOCKLIST_URL,
                timeout=self._http_timeout,
            )
            resp.raise_for_status()
            entries = resp.json()
            feodo_ips = [
                e.get("ip_address", "")
                for e in entries
                if isinstance(e, dict) and e.get("ip_address")
            ]
            ips.extend(feodo_ips)
            logger.info("FeodoTracker: fetched %d malicious IPs.", len(feodo_ips))
        except Exception as exc:  # noqa: BLE001
            logger.warning("FeodoTracker fetch failed — %s", exc)

        # Deduplicate + validate
        unique_ips = list(dict.fromkeys(ip for ip in ips if self._is_ipv4(ip)))

        if unique_ips:
            self._threat_ip_cache = unique_ips
            self._cache_fetched_at = now
            return list(unique_ips)

        # All fetches failed — return fallback
        logger.warning(
            "All threat intel sources failed.  Using fallback IP list (%d entries).",
            len(FALLBACK_THREAT_IPS),
        )
        return list(FALLBACK_THREAT_IPS)

    def get_threat_intel_summary(self) -> Dict[str, Any]:
        """
        Return metadata about the current threat intel cache state.

        Returns
        -------
        dict with keys: cached_ip_count, cache_age_seconds, cache_valid,
                        sample_ips (first 5)
        """
        now = time.monotonic()
        age = now - self._cache_fetched_at if self._cache_fetched_at > 0 else None
        return {
            "cached_ip_count": len(self._threat_ip_cache),
            "cache_age_seconds": round(age, 1) if age is not None else None,
            "cache_valid": bool(
                self._threat_ip_cache and age is not None and age < self._cache_ttl
            ),
            "sample_ips": self._threat_ip_cache[:5],
        }

    # =========================================================================
    # SOURCE B — Performance-Based Difficulty Adaptation
    # =========================================================================

    def adapt_difficulty(
        self, score_history: Optional[List[float]] = None
    ) -> Dict[str, Any]:
        """
        Compute difficulty adjustment parameters based on rolling avg score.

        Parameters
        ----------
        score_history : List[float], optional
            Episode scores to use.  If None, uses the internally tracked
            history (updated via ``record_episode_score``).  Uses the last 5
            scores from the provided list.

        Returns
        -------
        dict
            Keys:
            - ``difficulty_level`` : "easy" | "medium" | "hard"
            - ``rolling_avg_score`` : float
            - ``num_attack_events`` : int  (suggested event count)
            - ``inject_decoy_alert`` : bool
            - ``add_lateral_movement`` : bool
            - ``concurrent_attack_streams`` : int
            - ``use_obvious_malicious_ips`` : bool
            - ``rationale`` : str
        """
        history = score_history if score_history is not None else self._score_history
        recent = history[-5:] if len(history) >= 5 else history

        avg = sum(recent) / len(recent) if recent else 0.0

        params: Dict[str, Any] = {
            "rolling_avg_score": round(avg, 4),
            "episodes_considered": len(recent),
        }

        if avg < LOW_SCORE_THRESHOLD:
            # Blue agent struggling — make it easier
            params.update(
                {
                    "difficulty_level": "easy",
                    "num_attack_events": 3,
                    "inject_decoy_alert": False,
                    "add_lateral_movement": False,
                    "concurrent_attack_streams": 1,
                    "use_obvious_malicious_ips": True,
                    "rationale": (
                        f"Rolling avg score {avg:.2f} < {LOW_SCORE_THRESHOLD}. "
                        "Reduced event count and obvious attacker IPs to lower difficulty."
                    ),
                }
            )

        elif LOW_SCORE_THRESHOLD <= avg <= HIGH_SCORE_THRESHOLD:
            # Blue agent performing adequately — maintain current difficulty
            params.update(
                {
                    "difficulty_level": "medium",
                    "num_attack_events": 6,
                    "inject_decoy_alert": False,
                    "add_lateral_movement": False,
                    "concurrent_attack_streams": 1,
                    "use_obvious_malicious_ips": False,
                    "rationale": (
                        f"Rolling avg score {avg:.2f} in [{LOW_SCORE_THRESHOLD}, "
                        f"{HIGH_SCORE_THRESHOLD}]. Maintaining current difficulty."
                    ),
                }
            )

        else:
            # Blue agent excelling — escalate challenge
            params.update(
                {
                    "difficulty_level": "hard",
                    "num_attack_events": 10,
                    "inject_decoy_alert": True,       # A benign IP that looks suspicious
                    "add_lateral_movement": True,     # Add a new lateral movement hop
                    "concurrent_attack_streams": 3,   # Multiple simultaneous attacker IPs
                    "use_obvious_malicious_ips": False,
                    "rationale": (
                        f"Rolling avg score {avg:.2f} > {HIGH_SCORE_THRESHOLD}. "
                        "Injecting decoy alert, adding lateral movement step, "
                        "and increasing concurrent attack streams."
                    ),
                }
            )

        return params

    def record_episode_score(self, score: float) -> None:
        """
        Append an episode's final score to the internal history for
        subsequent ``adapt_difficulty()`` calls.

        Parameters
        ----------
        score : float
            Final grader score in [0, 1].
        """
        self._score_history.append(score)
        # Keep only last 20 to avoid unbounded growth
        if len(self._score_history) > 20:
            self._score_history = self._score_history[-20:]

    @property
    def score_history(self) -> List[float]:
        """Read-only copy of the internal score history."""
        return list(self._score_history)

    # =========================================================================
    # SOURCE C — Runtime Custom Parameter Merging
    # =========================================================================

    def merge_custom_params(
        self,
        base_params: Dict[str, Any],
        custom_params: Optional[Dict[str, Any]],
    ) -> Dict[str, Any]:
        """
        Merge runtime custom parameters on top of base scenario defaults.

        Only keys present in ``custom_params`` with non-None values will
        override the base.  All other base keys are preserved.

        Parameters
        ----------
        base_params : dict
            The scenario's default parameters.
        custom_params : dict or None
            Overrides provided at runtime (from the POST /reset body).

        Returns
        -------
        dict
            Merged parameter dict with custom overrides applied.
        """
        merged = dict(base_params)
        if not custom_params:
            return merged

        for key, value in custom_params.items():
            if value is not None:
                merged[key] = value

        return merged

    # =========================================================================
    # Helpers
    # =========================================================================

    @staticmethod
    def _is_ipv4(s: str) -> bool:
        """Return True if ``s`` looks like a valid IPv4 address."""
        parts = s.split(".")
        if len(parts) != 4:
            return False
        try:
            return all(0 <= int(p) <= 255 for p in parts)
        except ValueError:
            return False