Attack Simulation API

Pure core - attack_core

attack_core.pick_target

pick_target(attack_type: str, idx: dict) -> str

Pick a plausible target node for the attack type.

Pure function: relies on a pre-built index dict produced by telecom_attack._build_target_index(). No DB access here.

Parameters:

Name	Type	Description	Default
attack_type	str	key from ATTACK_CATALOG.	required
idx	dict	dict with keys {"cores", "all_cores", "cells_5g", "cells_6g"}.	required

Returns:

Type	Description
str	A node_id or cell_id, or "UNKNOWN" if the pool is empty.

Source code in attack_core.py

def pick_target(attack_type: str, idx: dict) -> str:
    """Pick a plausible target node for the attack type.

    Pure function: relies on a pre-built index dict produced by
    telecom_attack._build_target_index(). No DB access here.

    Args:
        attack_type: key from ATTACK_CATALOG.
        idx: dict with keys {"cores", "all_cores", "cells_5g", "cells_6g"}.

    Returns:
        A node_id or cell_id, or "UNKNOWN" if the pool is empty.
    """
    roles = []
    pool = None

    if attack_type in ("DIAMETER_ULR_FLOOD", "DIAMETER_AIR_FLOOD"):
        roles = ["HSS"]
    elif attack_type == "GTP_C_FLOOD":
        roles = ["SGW", "PGW"]
    elif attack_type == "PFCP_SESSION_FLOOD":
        roles = ["UPF"]
    elif attack_type in ("SIP_REGISTER_FLOOD", "SIP_INVITE_FLOOD"):
        roles = ["IMS"]
    elif attack_type in ("HTTP2_RAPID_RESET", "NFS_NRF_ENUM", "NRF_SBI_SCAN"):
        roles = ["AMF", "SMF", "UPF"]
    elif attack_type == "SCTP_MME_FLOOD":
        roles = ["MME"]
    elif attack_type in ("SS7_MAP_ATI_ABUSE", "SS7_SRI_FLOOD"):
        roles = ["MSC"]
    elif attack_type in ("RIS_PHASE_POISON", "ISAC_SPOOFING", "AI_RAN_POISON"):
        roles = [
            {
                "RIS_PHASE_POISON": "RIS-C",
                "ISAC_SPOOFING": "ISAC",
                "AI_RAN_POISON": "AI-RAN",
            }[attack_type]
        ]
    elif attack_type == "NWDAF_DATA_INJECT":
        roles = ["NWDAF"]
    elif attack_type == "THZ_JAMMING":
        pool = idx.get("cells_6g", [])
    elif attack_type == "QUIC_FLOOD_6G":
        roles = ["AMF"]
    elif attack_type.startswith("ORAN_"):
        pool = idx.get("cells_5g", [])
    else:
        pool = idx.get("all_cores", [])

    if pool is None:
        pool = []
        for r in roles:
            pool.extend(idx.get("cores", {}).get(r, []))

    return random.choice(pool) if pool else "UNKNOWN"

attack_core.generate_scenarios

generate_scenarios(n: int, target_idx: dict, now: datetime | None = None) -> list[dict]

Generate n random attack scenarios with plausible timings.

Pure: does not touch the DB. Caller supplies a pre-built target index (see telecom_attack._build_target_index()).

Guarantees a share of 6G scenarios (RIS_PHASE_POISON, AI_RAN_POISON, THZ_JAMMING, QUIC_FLOOD_6G).

Parameters:

Name	Type	Description	Default
n	int	number of scenarios to generate.	required
target_idx	dict	dict with keys {"cores", "all_cores", "cells_5g", "cells_6g"}.	required
now	datetime | None	reference time; defaults to datetime.now().	None

Returns:

Type	Description
list[dict]	List of dicts matching the attack_scenarios table schema.

Source code in attack_core.py

def generate_scenarios(n: int, target_idx: dict, now: datetime | None = None) -> list[dict]:
    """Generate n random attack scenarios with plausible timings.

    Pure: does not touch the DB. Caller supplies a pre-built target
    index (see telecom_attack._build_target_index()).

    Guarantees a share of 6G scenarios (RIS_PHASE_POISON,
    AI_RAN_POISON, THZ_JAMMING, QUIC_FLOOD_6G).

    Args:
        n: number of scenarios to generate.
        target_idx: dict with keys {"cores", "all_cores",
                    "cells_5g", "cells_6g"}.
        now: reference time; defaults to datetime.now().

    Returns:
        List of dicts matching the attack_scenarios table schema.
    """
    if now is None:
        now = datetime.now()

    scenarios = []
    all_types = list(ATTACK_CATALOG.keys())
    _forced_6g = [
        a
        for a in ("RIS_PHASE_POISON", "AI_RAN_POISON", "THZ_JAMMING", "QUIC_FLOOD_6G")
        if a in ATTACK_CATALOG
    ]
    n_forced = min(len(_forced_6g), n)
    chosen_types = random.sample(_forced_6g, k=n_forced)
    other_types = [t for t in all_types if t not in _forced_6g]
    if n > n_forced and other_types:
        chosen_types += random.choices(other_types, k=n - n_forced)
    random.shuffle(chosen_types)

    for i, atype in enumerate(chosen_types):
        meta = ATTACK_CATALOG[atype]

        start = now - timedelta(hours=random.uniform(0, 24), minutes=random.uniform(0, 60))
        dur = random.choice([10, 30, 60, 120, 300, 600, 1800])
        end = start + timedelta(seconds=dur)

        base_rate = {
            "signaling": (500, 50_000),
            "ddos": (10_000, 2_000_000),
            "ran": (100, 20_000),
            "recon": (5, 500),
            "exfil": (10, 2_000),
        }.get(meta["family"], (100, 10_000))
        rate = random.randint(*base_rate)
        total_events = int(rate * dur * random.uniform(0.7, 1.0))

        scenarios.append(
            {
                "scenario_id": f"ATK-{now.strftime('%Y%m%d')}-{i+1:04d}",
                "name": f"{atype} on {meta['target_iface']}",
                "attack_type": atype,
                "target_layer": meta["layer"],
                "target_node": pick_target(atype, target_idx),
                "target_iface": meta["target_iface"],
                "protocol": meta["protocol"],
                "severity": meta["severity"],
                "start_ts": start.strftime("%Y-%m-%d %H:%M:%S"),
                "end_ts": end.strftime("%Y-%m-%d %H:%M:%S"),
                "duration_sec": dur,
                "rate_pps": rate,
                "total_events": total_events,
                "source_ip": f"{random.randint(1, 223)}.{random.randint(0, 255)}."
                f"{random.randint(0, 255)}.{random.randint(1, 254)}",
                "mitre": meta["mitre"],
                "status": random.choices(
                    ["BLOCKED", "DETECTED", "SUCCESS", "ONGOING"], weights=[0.5, 0.35, 0.1, 0.05]
                )[0],
                "notes": meta["desc"],
            }
        )
    return scenarios

attack_core.expand_events

expand_events(scenarios: list[dict], samples_per_scenario: int = 20) -> list[dict]

Expand each scenario into a time-series of event samples.

Pure: no DB, no I/O.

For each scenario, generate samples_per_scenario equally-spaced samples between start_ts and end_ts. Each sample carries MTTD, MTTR, and KPI impact (latency, drop) derived from pps.

Source code in attack_core.py

def expand_events(scenarios: list[dict], samples_per_scenario: int = 20) -> list[dict]:
    """Expand each scenario into a time-series of event samples.

    Pure: no DB, no I/O.

    For each scenario, generate `samples_per_scenario` equally-spaced
    samples between start_ts and end_ts. Each sample carries MTTD,
    MTTR, and KPI impact (latency, drop) derived from pps.
    """
    events = []
    for s in scenarios:
        start = datetime.strptime(s["start_ts"], "%Y-%m-%d %H:%M:%S")
        dur = s["duration_sec"]
        for k in range(samples_per_scenario):
            offset = dur * k / samples_per_scenario
            ts = start + timedelta(seconds=offset)

            mttd = random.uniform(3, 90) if s["status"] != "SUCCESS" else None
            detected = 1 if mttd is not None and k * dur / samples_per_scenario >= mttd else 0

            mttr = None
            blocked = 0
            if detected:
                mttr = random.uniform(15, 600)
                if k * dur / samples_per_scenario >= mttd + mttr:
                    blocked = 1

            latency = random.gauss(50, 20) + (s["rate_pps"] / 10_000) * random.uniform(5, 30)
            latency = max(1.0, latency)
            drop = min(95.0, (s["rate_pps"] / 100_000) * random.uniform(2, 15))

            events.append(
                {
                    "scenario_id": s["scenario_id"],
                    "ts": ts.strftime("%Y-%m-%d %H:%M:%S"),
                    "attack_type": s["attack_type"],
                    "target_node": s["target_node"],
                    "source_ip": s["source_ip"],
                    "pps": int(s["rate_pps"] * random.uniform(0.6, 1.4)),
                    "latency_ms": round(latency, 2),
                    "drop_pct": round(drop, 2),
                    "detected": detected,
                    "detected_by": random.choice(["SIEM", "SOC-N1", "Anomaly-Engine", "NOC"])
                    if detected
                    else None,
                    "blocked": blocked,
                    "mttd_sec": round(mttd, 1) if detected and mttd is not None else None,
                    "mttr_sec": round(mttr, 1) if blocked and mttr is not None else None,
                    "severity": s["severity"],
                    "notes": "",
                }
            )
    return events

attack_core.attacks_to_alerts

attacks_to_alerts(scenarios: list[dict], events: list[dict]) -> list[dict]

Convert attack scenarios into alert rows for the alerts table.

Pure: no DB, no I/O. Only DETECTED/SUCCESS/ONGOING or CRITICAL scenarios become alerts. Each alert routes to one SOC recipient.

Source code in attack_core.py

def attacks_to_alerts(scenarios: list[dict], events: list[dict]) -> list[dict]:
    """Convert attack scenarios into alert rows for the alerts table.

    Pure: no DB, no I/O. Only DETECTED/SUCCESS/ONGOING or CRITICAL
    scenarios become alerts. Each alert routes to one SOC recipient.
    """
    alerts = []
    ts_now = datetime.now()
    ts_suffix = ts_now.strftime("%H%M%S")
    interesting = [
        s
        for s in scenarios
        if s["status"] in ("DETECTED", "SUCCESS", "ONGOING") or s["severity"] == "CRITICAL"
    ]
    for idx, s in enumerate(interesting[:30]):
        sev_map = {"CRITICAL": "CRITICAL", "HIGH": "HIGH", "MEDIUM": "MEDIUM", "LOW": "LOW"}
        sev = sev_map.get(s["severity"], "MEDIUM")
        soc_to = random.choice(SOC_RECIPIENTS)
        alert_id = f"ALT-{ts_now.strftime('%Y%m%d')}-{ts_suffix}-ATK{idx+1:04d}"
        alerts.append(
            {
                "alert_id": alert_id,
                "timestamp": s["start_ts"],
                "severity": sev,
                "alert_type": f"ATTACK_{s['attack_type']}",
                "msisdn": "-",
                "description": f"{s['attack_type']} on {s['target_node']} "
                f"({s['target_iface']}) @ {s['rate_pps']:,} pps",
                "extra": json_dumps(
                    {
                        "scenario_id": s["scenario_id"],
                        "layer": s["target_layer"],
                        "protocol": s["protocol"],
                        "mitre": s["mitre"],
                        "status": s["status"],
                    }
                ),
                "sms_sent": 1,
                "sms_to": soc_to,
                "sms_body": f"[{sev}] ATTACK {s['attack_type']} → "
                f"{s['target_node']} | {s['status']}",
                "ack": 0,
            }
        )
    return alerts

DB layer - telecom_attack

telecom_attack.init_attack_tables

init_attack_tables()

Source code in telecom_attack.py

def init_attack_tables():
    con = db_connect()
    cur = con.cursor()
    cur.executescript("""
    CREATE TABLE IF NOT EXISTS attack_scenarios(
        scenario_id TEXT PRIMARY KEY,
        name TEXT,
        attack_type TEXT,
        target_layer TEXT,
        target_node TEXT,
        target_iface TEXT,
        protocol TEXT,
        severity TEXT,
        start_ts TEXT,
        end_ts TEXT,
        duration_sec INT,
        rate_pps INT,
        total_events INT,
        source_ip TEXT,
        mitre TEXT,
        status TEXT,
        notes TEXT
    );
    CREATE TABLE IF NOT EXISTS attack_events(
        event_id INTEGER PRIMARY KEY AUTOINCREMENT,
        scenario_id TEXT,
        ts TEXT,
        attack_type TEXT,
        target_node TEXT,
        source_ip TEXT,
        pps INT,
        latency_ms REAL,
        drop_pct REAL,
        detected INT,
        detected_by TEXT,
        blocked INT,
        mttd_sec REAL,
        mttr_sec REAL,
        severity TEXT,
        notes TEXT
    );
    CREATE INDEX IF NOT EXISTS idx_atk_scen ON attack_events(scenario_id);
    CREATE INDEX IF NOT EXISTS idx_atk_ts   ON attack_events(ts);
    CREATE INDEX IF NOT EXISTS idx_atk_type ON attack_events(attack_type);
    """)
    con.commit()
    con.close()

telecom_attack.generate_attack_scenarios

generate_attack_scenarios(n: int = 25) -> list[dict]

Generate n random attack scenarios with plausible timings.

Picks attack types from ATTACK_CATALOG, resolves a target node from the DB (via _build_target_index), and assigns: - realistic start/end timestamps within the last 24h, - a rate (pps) drawn from the attack family's typical range, - a status in {BLOCKED, DETECTED, SUCCESS, ONGOING}.

Guarantees a share of 6G scenarios (RIS_PHASE_POISON, AI_RAN_POISON, THZ_JAMMING, QUIC_FLOOD_6G) so 6G is always represented.

Parameters:

Name	Type	Description	Default
n	int	Number of scenarios to generate. Default 25.	25

Returns:

Type	Description
list[dict]	List of dicts, each with the schema of the attack_scenarios
list[dict]	table (see init_attack_tables()).

Source code in telecom_attack.py

def generate_attack_scenarios(n: int = 25) -> list[dict]:
    """Generate n random attack scenarios with plausible timings.

    Picks attack types from ATTACK_CATALOG, resolves a target node
    from the DB (via _build_target_index), and assigns:
      - realistic start/end timestamps within the last 24h,
      - a rate (pps) drawn from the attack family's typical range,
      - a status in {BLOCKED, DETECTED, SUCCESS, ONGOING}.

    Guarantees a share of 6G scenarios (RIS_PHASE_POISON, AI_RAN_POISON,
    THZ_JAMMING, QUIC_FLOOD_6G) so 6G is always represented.

    Args:
        n: Number of scenarios to generate. Default 25.

    Returns:
        List of dicts, each with the schema of the `attack_scenarios`
        table (see init_attack_tables()).
    """
    init_attack_tables()
    _target_idx = _build_target_index()  # one DB read
    scenarios = []
    now = datetime.now()
    all_types = list(ATTACK_CATALOG.keys())
    # Actually guarantee a share of 6G scenarios,
    # instead of only reordering the list (which random.choice ignored).
    _forced_6g = [
        a
        for a in ("RIS_PHASE_POISON", "AI_RAN_POISON", "THZ_JAMMING", "QUIC_FLOOD_6G")
        if a in ATTACK_CATALOG
    ]
    n_forced = min(len(_forced_6g), n)
    chosen_types = random.sample(_forced_6g, k=n_forced)
    other_types = [t for t in all_types if t not in _forced_6g]
    if n > n_forced and other_types:
        chosen_types += random.choices(other_types, k=n - n_forced)
    random.shuffle(chosen_types)

    for i, atype in enumerate(chosen_types):
        meta = ATTACK_CATALOG[atype]

        # Timing: attacks typically bursty, short-lived
        start = now - timedelta(hours=random.uniform(0, 24), minutes=random.uniform(0, 60))
        dur = random.choice([10, 30, 60, 120, 300, 600, 1800])
        end = start + timedelta(seconds=dur)

        # Rate depends on family
        base_rate = {
            "signaling": (500, 50_000),
            "ddos": (10_000, 2_000_000),
            "ran": (100, 20_000),
            "recon": (5, 500),
            "exfil": (10, 2_000),
        }.get(meta["family"], (100, 10_000))
        rate = random.randint(*base_rate)

        total_events = int(rate * dur * random.uniform(0.7, 1.0))

        scenario = {
            "scenario_id": f"ATK-{now.strftime('%Y%m%d')}-{i+1:04d}",
            "name": f"{atype} on {meta['target_iface']}",
            "attack_type": atype,
            "target_layer": meta["layer"],
            "target_node": _pick_target(atype, _target_idx),
            "target_iface": meta["target_iface"],
            "protocol": meta["protocol"],
            "severity": meta["severity"],
            "start_ts": start.strftime("%Y-%m-%d %H:%M:%S"),
            "end_ts": end.strftime("%Y-%m-%d %H:%M:%S"),
            "duration_sec": dur,
            "rate_pps": rate,
            "total_events": total_events,
            "source_ip": f"{random.randint(1,223)}.{random.randint(0,255)}."
            f"{random.randint(0,255)}.{random.randint(1,254)}",
            "mitre": meta["mitre"],
            "status": random.choices(
                ["BLOCKED", "DETECTED", "SUCCESS", "ONGOING"], weights=[0.5, 0.35, 0.1, 0.05]
            )[0],
            "notes": meta["desc"],
        }
        scenarios.append(scenario)
    return scenarios

telecom_attack.persist_attacks

persist_attacks(scenarios: list[dict], events: list[dict]) -> None

Source code in telecom_attack.py

def persist_attacks(scenarios: list[dict], events: list[dict]) -> None:
    init_attack_tables()
    con = db_connect()
    cur = con.cursor()

    # Wipe previous run
    cur.execute("DELETE FROM attack_scenarios")
    cur.execute("DELETE FROM attack_events")

    cur.executemany(
        """INSERT INTO attack_scenarios VALUES(
        :scenario_id,:name,:attack_type,:target_layer,:target_node,
        :target_iface,:protocol,:severity,:start_ts,:end_ts,
        :duration_sec,:rate_pps,:total_events,:source_ip,:mitre,
        :status,:notes)""",
        scenarios,
    )

    cur.executemany(
        """INSERT INTO attack_events(
        scenario_id,ts,attack_type,target_node,source_ip,pps,
        latency_ms,drop_pct,detected,detected_by,blocked,
        mttd_sec,mttr_sec,severity,notes) VALUES(
        :scenario_id,:ts,:attack_type,:target_node,:source_ip,:pps,
        :latency_ms,:drop_pct,:detected,:detected_by,:blocked,
        :mttd_sec,:mttr_sec,:severity,:notes)""",
        events,
    )

    con.commit()
    con.close()

telecom_attack.inject_attack_alerts

inject_attack_alerts(alerts: list[dict]) -> None

Merge attack-derived alerts into the alerts table.

Source code in telecom_attack.py

def inject_attack_alerts(alerts: list[dict]) -> None:
    """Merge attack-derived alerts into the alerts table."""
    if not alerts:
        return
    con = db_connect()
    cur = con.cursor()
    # wipe previous ATTACK_* alerts and their SMS
    try:
        cur.execute(
            "DELETE FROM sms_alerts WHERE alert_id IN "
            "(SELECT alert_id FROM alerts "
            "WHERE alert_type LIKE 'ATTACK_%')"
        )
        cur.execute("DELETE FROM alerts " "WHERE alert_type LIKE 'ATTACK_%'")
        con.commit()
    except Exception:
        pass
    for a in alerts:
        try:
            cur.execute(
                """INSERT OR IGNORE INTO alerts(
                alert_id,timestamp,severity,alert_type,msisdn,description,
                extra,sms_sent,sms_to,sms_body,ack) VALUES(?,?,?,?,?,?,?,?,?,?,?)""",
                (
                    a["alert_id"],
                    a["timestamp"],
                    a["severity"],
                    a["alert_type"],
                    a["msisdn"],
                    a["description"],
                    a["extra"],
                    a["sms_sent"],
                    a["sms_to"],
                    a["sms_body"],
                    a["ack"],
                ),
            )
            if a["sms_sent"]:
                cur.execute(
                    """INSERT INTO sms_alerts(
                    alert_id,timestamp,recipient,body,status)
                    VALUES(?,?,?,?,?)""",
                    (a["alert_id"], a["timestamp"], a["sms_to"], a["sms_body"], "DELIVERED"),
                )
        except Exception:
            pass
    con.commit()
    con.close()

telecom_attack.run_attack_simulation

run_attack_simulation(n_scenarios: int = 25, samples_per_scenario: int = 20, inject_alerts: bool = True) -> dict

End-to-end attack simulation: generate → persist → alert.

Pipeline

1. generate_attack_scenarios(n_scenarios)
2. expand_events(...) → time-series
3. persist_attacks(...) → attack_scenarios + attack_events tables
4. attacks_to_alerts(...) → alert rows
5. inject_attack_alerts(...) → merge into alerts + sms_alerts

Parameters:

Name	Type	Description	Default
n_scenarios	int	Number of attack scenarios to generate.	25
samples_per_scenario	int	Event samples per scenario.	20
inject_alerts	bool	If True, merge into alerts/sms_alerts tables.	True

Returns:

Type	Description
dict	Summary dict: { "scenarios": int, "events": int, "alerts": int, "by_status": {status: count, ...}, "by_severity": {severity: count, ...}, }

Source code in telecom_attack.py

def run_attack_simulation(
    n_scenarios: int = 25, samples_per_scenario: int = 20, inject_alerts: bool = True
) -> dict:
    """End-to-end attack simulation: generate → persist → alert.

    Pipeline:
        1. generate_attack_scenarios(n_scenarios)
        2. expand_events(...) → time-series
        3. persist_attacks(...) → attack_scenarios + attack_events tables
        4. attacks_to_alerts(...) → alert rows
        5. inject_attack_alerts(...) → merge into alerts + sms_alerts

    Args:
        n_scenarios:         Number of attack scenarios to generate.
        samples_per_scenario: Event samples per scenario.
        inject_alerts:       If True, merge into alerts/sms_alerts tables.

    Returns:
        Summary dict:
            {
              "scenarios": int,
              "events":    int,
              "alerts":    int,
              "by_status":   {status: count, ...},
              "by_severity": {severity: count, ...},
            }
    """
    print(f"[ATTACK] Generating {n_scenarios} scenarios...")
    scenarios = generate_attack_scenarios(n_scenarios)
    print("[ATTACK] Expanding to event time-series...")
    events = expand_events(scenarios, samples_per_scenario)
    print(f"[ATTACK] Persisting {len(scenarios)} scenarios, {len(events)} events...")
    persist_attacks(scenarios, events)
    alerts = attacks_to_alerts(scenarios, events)
    if inject_alerts and alerts:
        print(f"[ATTACK] Injecting {len(alerts)} alerts into alerts table...")
        inject_attack_alerts(alerts)
    summary = {
        "scenarios": len(scenarios),
        "events": len(events),
        "alerts": len(alerts),
        "by_status": {},
        "by_severity": {},
    }
    for s in scenarios:
        summary["by_status"][s["status"]] = summary["by_status"].get(s["status"], 0) + 1
        summary["by_severity"][s["severity"]] = summary["by_severity"].get(s["severity"], 0) + 1
    return summary