data/__init__.py

"""Mock data fixtures for the control plane demo.

WHY THIS EXISTS
---------------
The control plane governs *actions agents take against systems* — read an asset,
draft a remediation, create a ticket, stop a container. To demonstrate that
end to end without touching any real SOC tooling, we ship a small, static,
**non-sensitive** dataset that stands in for the systems an agent would call:

    * alerts.json   — SIEM detections (the "why we're here" for each incident)
    * assets.json   — hosts/workstations (and the containers running on them)
    * cves.json     — vulnerabilities referenced by alerts and remediation drafts
    * tickets.json  — the ticketing system's seed records (and its writable store)
    * users.json    — the identities referenced by alerts and tickets

All data is invented. There are no real hostnames, credentials, IPs, or PII —
the only realistic values are *public* CVE identifiers, kept for recognisability.

This module is the single, typed entry point the execution backends (Phases 5.2–
5.4) and demo scenarios load their data through, so there is exactly one place
that knows where the JSON lives and how it is shaped.

DESIGN NOTES
------------
* The JSON files are pure data (JSON has no comments) — the narrative and the
  contract live here, in code, where they can be read and tested.
* :data:`SCENARIO_IDENTIFIERS` pins the specific IDs each acceptance
  scenario depends on. ``test_mock_data`` asserts every one of them resolves, so
  if a fixture is ever edited in a way that would break a demo scenario, the
  test fails immediately rather than the demo failing live.
"""

from __future__ import annotations

import json
from functools import lru_cache
from pathlib import Path
from typing import Any

# Directory holding the JSON fixtures — this package's own directory.
DATA_DIR = Path(__file__).resolve().parent

# Logical dataset name -> filename. The keys double as the public API surface
# (see :func:`load_dataset`) and as the set of datasets the test iterates over.
_DATASETS: dict[str, str] = {
    "alerts": "alerts.json",
    "assets": "assets.json",
    "cves": "cves.json",
    "tickets": "tickets.json",
    "users": "users.json",
}

# The primary-key field for each dataset, so callers can index records by ID
# without hard-coding "which field is the id" at every call site.
_ID_FIELD: dict[str, str] = {
    "alerts": "alert_id",
    "assets": "asset_id",
    "cves": "cve_id",
    "tickets": "ticket_id",
    "users": "user_id",
}

# Identifiers each acceptance scenario relies on existing in the data.
# This is the contract the demo runs against; the test enforces it. Scenarios 5
# (dangerous action denied) and 6 (kill switch) are pure governance outcomes and
# need no specific record, so they are intentionally absent here.
SCENARIO_IDENTIFIERS: dict[str, dict[str, str]] = {
    # 1. L0 Direct API read — an asset must exist to look up.
    "scenario_1_l0_read": {"assets": "ASSET-001"},
    # 2. L1 Function draft — a CVE must exist to draft remediation against.
    "scenario_2_l1_draft": {"cves": "CVE-2024-3094"},
    # 3. L2 low-risk ticket — an alert/asset must exist for the new ticket to reference.
    "scenario_3_l2_ticket": {"alerts": "ALERT-2001", "assets": "ASSET-001"},
    # 4. L3 safe-CLI container stop — the suspicious container lives on this asset
    #    (the asset record carries it; verified explicitly in the test).
    "scenario_4_l3_stop": {"assets": "ASSET-001"},
    # 7. Identity verification — the user attributed to the incident must exist.
    "scenario_7_identity": {"users": "USR-301"},
}

# Container name the L3 "stop a suspicious container" scenario acts on.
SUSPICIOUS_CONTAINER_NAME = "suspicious-nginx"


@lru_cache(maxsize=None)
def load_dataset(name: str) -> list[dict[str, Any]]:
    """Load one fixture by logical name (e.g. ``"assets"``) as a list of records.

    Results are cached: the fixtures are static for the life of the process, so
    we read and parse each file at most once. (Backends that *mutate* state —
    e.g. creating a ticket — must copy what they need into their own store; see
    :func:`load_records`, which hands back a fresh copy.)
    """
    if name not in _DATASETS:
        raise KeyError(f"Unknown dataset {name!r}; expected one of {sorted(_DATASETS)}")
    path = DATA_DIR / _DATASETS[name]
    with path.open(encoding="utf-8") as fh:
        data = json.load(fh)
    if not isinstance(data, list):
        raise ValueError(f"Fixture {path.name} must be a JSON array of records")
    return data


def load_records(name: str) -> list[dict[str, Any]]:
    """Return a **deep copy** of a dataset, safe for callers that mutate it.

    Use this from execution backends that add or change records (e.g. creating a
    ticket) so the in-memory mock store can diverge from the on-disk seed without
    corrupting the shared cached copy.
    """
    return json.loads(json.dumps(load_dataset(name)))


def index_by_id(name: str) -> dict[str, dict[str, Any]]:
    """Return a dataset keyed by its primary-id field, for O(1) lookups by ID."""
    id_field = _ID_FIELD[name]
    return {record[id_field]: record for record in load_dataset(name)}


def get_by_id(name: str, record_id: str) -> dict[str, Any] | None:
    """Fetch a single record by its ID, or ``None`` if no such record exists."""
    return index_by_id(name).get(record_id)


# Convenience accessors — thin, self-documenting wrappers the backends and
# scenarios can import directly (``from data import load_assets``).
def load_alerts() -> list[dict[str, Any]]:
    return load_dataset("alerts")


def load_assets() -> list[dict[str, Any]]:
    return load_dataset("assets")


def load_cves() -> list[dict[str, Any]]:
    return load_dataset("cves")


def load_tickets() -> list[dict[str, Any]]:
    return load_dataset("tickets")


def load_users() -> list[dict[str, Any]]:
    return load_dataset("users")


__all__ = [
    "DATA_DIR",
    "SCENARIO_IDENTIFIERS",
    "SUSPICIOUS_CONTAINER_NAME",
    "load_dataset",
    "load_records",
    "index_by_id",
    "get_by_id",
    "load_alerts",
    "load_assets",
    "load_cves",
    "load_tickets",
    "load_users",
]