Simulator API

telecom_net_sim

telecom_net_sim.CellSite dataclass

CellSite(cell_id: str, name: str, tech: str, city: str, lat: float, lon: float, band: str, azimuth: int, tilt: int, tx_dbm: float, backhaul_gbps: float)

A radio cell site (BTS/NodeB/eNodeB/gNodeB).

Attributes:

Name	Type	Description
cell_id	str	Unique cell identifier, e.g. "4G-TEH-0021".
name	str	Human-readable name, e.g. "Tehran-4G-5".
tech	str	Radio access technology: "2G", "3G", "4G", "5G", or "6G".
city	str	City name where the site is deployed.
lat	float	Latitude in decimal degrees.
lon	float	Longitude in decimal degrees.
band	str	Frequency band, e.g. "LTE-B3", "NR-n78", "THz-140".
azimuth	int	Antenna azimuth in degrees (0-359).
tilt	int	Antenna electrical tilt in degrees (0-15).
tx_dbm	float	Transmit power in dBm (20-46).
backhaul_gbps	float	Backhaul capacity in Gbps.

telecom_net_sim.CoreNode dataclass

CoreNode(node_id: str, name: str, role: str, tech: str, city: str, capacity_tps: int)

A core network node (MSC, MME, AMF, UPF, NWDAF, ...).

Attributes:

Name	Type	Description
node_id	str	Unique node identifier, e.g. "AMF-01".
name	str	Human-readable name, e.g. "TELECOM-AMF-TEH-01".
role	str	Functional role: "MSC", "BSC", "RNC", "MME", "SGW", "PGW", "HSS", "PCRF", "AMF", "SMF", "UPF", "IMS", "MMSC", "RCS-AS", "NWDAF", "RIS-C", "ISAC", or "AI-RAN".
tech	str	Technology scope, e.g. "4G/5G" or "6G".
city	str	City where the node is deployed.
capacity_tps	int	Peak throughput capacity in transactions per second.

telecom_net_sim.Subscriber dataclass

Subscriber(msisdn: str, imsi: str, imei: str, city: str, plan: str, kyc_age_days: int, roaming_enabled: bool, risk_score: float, line_class: str, international_access: bool, filter_bypass: bool, clir_enabled: bool, clir_override: bool, priority_qos: int, lawful_intercept: bool, direct_routing: bool, whitelisted_asns: list[str] = list(), encryption_required: bool = False, cipher_suite: str = 'None', key_id: str = '', key_rotation_days: int = 0, e2e_enabled: bool = False)

A mobile subscriber with line-class, QoS, and encryption attributes.

Line classes ("Normal", "VIP", "Government", "Corporate", "Emergency", "Test") determine default privileges and QoS range.

Attributes:

Name	Type	Description
msisdn	str	Mobile number in E.164-ish form ("98" + 10 digits).
imsi	str	International Mobile Subscriber Identity (15 digits).
imei	str	15-digit device identity (Luhn-valid).
city	str	Home city.
plan	str	Tariff plan, e.g. "Prepaid-Basic".
kyc_age_days	int	Days since KYC was last refreshed.
roaming_enabled	bool	True if roaming is allowed.
risk_score	float	Fraud risk score in [0, 1].
line_class	str	One of LINE_CLASSES.
international_access	bool	True if international calls/data allowed.
filter_bypass	bool	True if content filtering is bypassed.
clir_enabled	bool	True if caller-ID suppression is authorized.
clir_override	bool	True if caller-ID can be spoofed.
priority_qos	int	QoS class in [0, 9] (higher = better).
lawful_intercept	bool	True if the line is under LI.
direct_routing	bool	True if the line uses direct routing.
whitelisted_asns	list[str]	ASNs whitelisted for unfiltered access.
encryption_required	bool	True if E2E encryption is mandatory.
cipher_suite	str	Cipher suite name, or "None".
key_id	str	Key identifier for E2E crypto.
key_rotation_days	int	Key rotation interval in days.
e2e_enabled	bool	True if E2E encryption is active.

telecom_net_sim.parse_args

parse_args(argv=None)

Parse CLI arguments.

Returns Namespace with: seed, random_seed, subs, cdrs.

Source code in telecom_net_sim.py

def parse_args(argv=None):
    """Parse CLI arguments.

    Returns Namespace with: seed, random_seed, subs, cdrs.
    """
    p = argparse.ArgumentParser(
        description="TELECOM-NET-SIM v3.0 — Local network simulator",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    p.add_argument("--seed", type=int, default=1403, help="Random seed for reproducibility")
    p.add_argument(
        "--random-seed", action="store_true", help="Ignore --seed and use a random seed instead"
    )
    p.add_argument("--subs", type=int, default=5000, help="Number of subscribers")
    p.add_argument("--cdrs", type=int, default=60_000, help="Number of CDR records")
    return p.parse_args(argv)

telecom_net_sim.build_topology

build_topology()

Source code in telecom_net_sim.py

def build_topology():
    cells: list[CellSite] = []
    counters = defaultdict(int)
    for city, lat, lon, pop, _w in CITIES:
        n4 = max(8, int(pop / 250_000))
        n5 = max(3, int(n4 * 0.35))
        n6 = max(1, int(n5 * 0.40))
        n3 = max(3, int(n4 * 0.4))
        n2 = max(3, int(n4 * 0.5))
        for tech, n in [("2G", n2), ("3G", n3), ("4G", n4), ("5G", n5), ("6G", n6)]:
            for i in range(n):
                counters[tech] += 1
                cid = f"{tech}-{city[:3].upper()}-{counters[tech]:04d}"
                cells.append(
                    CellSite(
                        cell_id=cid,
                        name=f"{city}-{tech}-{i+1}",
                        tech=tech,
                        city=city,
                        lat=lat + random.uniform(-0.20, 0.20),
                        lon=lon + random.uniform(-0.25, 0.25),
                        band=random.choice(BANDS[tech]),
                        azimuth=random.choice(
                            [0, 30, 60, 90, 120, 150, 180, 210, 240, 270, 300, 330]
                        ),
                        tilt=random.choice([2, 4, 6, 8, 10]),
                        tx_dbm=round(random.uniform(20, 46), 1),
                        backhaul_gbps=round(
                            random.uniform(1, 10)
                            if tech in ("2G", "3G")
                            else random.uniform(80, 200)
                            if tech == "6G"
                            else random.uniform(10, 40),
                            1,
                        ),
                    )
                )
    cores: list[CoreNode] = []
    core_plan = [
        ("MSC", "2G/3G", 12),
        ("BSC", "2G", 20),
        ("RNC", "3G", 10),
        ("MME", "4G", 8),
        ("SGW", "4G", 8),
        ("PGW", "4G", 8),
        ("HSS", "4G/5G", 4),
        ("PCRF", "4G", 4),
        ("AMF", "5G", 6),
        ("SMF", "5G", 6),
        ("UPF", "5G", 10),
        ("IMS", "VoLTE/VoWiFi", 6),
        ("MMSC", "MMS", 3),
        ("RCS-AS", "RCS", 3),
        ("NWDAF", "6G", 4),
        ("RIS-C", "6G", 3),
        ("ISAC", "6G", 2),
        ("AI-RAN", "6G", 2),
    ]
    idx = 0
    for role, tech, count in core_plan:
        for i in range(count):
            idx += 1
            city = random.choice(CITIES)[0]
            cores.append(
                CoreNode(
                    node_id=f"{role}-{i+1:02d}",
                    name=f"{OPERATOR}-{role}-{city[:3].upper()}-{i+1:02d}",
                    role=role,
                    tech=tech,
                    city=city,
                    capacity_tps=random.choice([50_000, 100_000, 200_000, 400_000]),
                )
            )
    return cells, cores

telecom_net_sim.build_subscribers

build_subscribers(n: int = 5000) -> list[Subscriber]

Source code in telecom_net_sim.py

def build_subscribers(n: int = 5000) -> list[Subscriber]:
    subs = []
    used_msisdn = set()
    used_imsi = set()
    city_choices = random.choices([c[0] for c in CITIES], weights=[c[4] for c in CITIES], k=n)
    for i in range(n):
        # --- Guarantee unique MSISDN ---
        while True:
            msisdn = gen_msisdn()
            if msisdn not in used_msisdn:
                used_msisdn.add(msisdn)
                break

        # --- Guarantee unique IMSI ---
        while True:
            imsi = gen_imsi()
            if imsi not in used_imsi:
                used_imsi.add(imsi)
                break

        imei = gen_imei()

        line_class = random.choices(LINE_CLASSES, weights=LINE_WEIGHTS)[0]
        qos_lo, qos_hi = QOS_BY_CLASS[line_class]

        intl_access = (
            line_class in {"Government", "Corporate", "VIP", "Emergency", "Test"}
            or random.random() < 0.08
        )
        filter_bypass = line_class in FILTER_BYPASS_CLASSES or (
            line_class == "VIP" and random.random() < 0.3
        )
        clir_enabled = line_class in CLIR_AUTHORIZED
        clir_override = line_class in CLIR_OVERRIDE_CLASSES
        lawful = line_class in {"Government", "Emergency"} or (
            line_class == "VIP" and random.random() < 0.05
        )
        direct_route = line_class in {"Government", "Emergency", "Corporate"}

        whitelisted = []
        if filter_bypass:
            whitelisted = random.sample(
                ["AS15169", "AS13335", "AS16509", "AS32934", "AS8075", "AS20940"],
                k=random.randint(1, 4),
            )

        enc_required = line_class in MANDATORY_ENCRYPTION
        if enc_required:
            e2e = True
        elif line_class in OPTIONAL_ENCRYPTION:
            e2e = random.random() < 0.6
        else:
            e2e = False

        if e2e:
            if line_class in MANDATORY_ENCRYPTION:
                cipher = random.choices(
                    ["Kyber-1024+AES-256", "AES-256-GCM", "ChaCha20-Poly1305"],
                    weights=[0.5, 0.3, 0.2],
                )[0]
            else:
                cipher = random.choice(list(CIPHER_SUITES.keys()))
            key_id = (
                "KEY-" + hashlib.sha1(f"{imsi}{random.random()}".encode()).hexdigest()[:16].upper()
            )
            rot_days = random.choice([30, 60, 90, 180])
        else:
            cipher = "None"
            key_id = ""
            rot_days = 0

        subs.append(
            Subscriber(
                msisdn=msisdn,
                imsi=imsi,
                imei=imei,
                city=city_choices[i],
                plan=random.choices(PLANS, weights=[40, 25, 15, 10, 5, 5])[0],
                kyc_age_days=random.randint(1, 3000),
                roaming_enabled=random.random() < 0.15 or intl_access,
                risk_score=round(random.betavariate(2, 8), 3),
                line_class=line_class,
                international_access=intl_access,
                filter_bypass=filter_bypass,
                clir_enabled=clir_enabled,
                clir_override=clir_override,
                priority_qos=random.randint(qos_lo, qos_hi),
                lawful_intercept=lawful,
                direct_routing=direct_route,
                whitelisted_asns=whitelisted,
                encryption_required=enc_required,
                cipher_suite=cipher,
                key_id=key_id,
                key_rotation_days=rot_days,
                e2e_enabled=e2e,
            )
        )
    return subs

telecom_net_sim.gen_cdrs

gen_cdrs(subs, cells, n: int = 60000) -> list[dict]

Source code in telecom_net_sim.py

def gen_cdrs(subs, cells, n: int = 60_000) -> list[dict]:
    by_city_cells = defaultdict(list)
    for c in cells:
        by_city_cells[c.city].append(c)

    start = datetime.now() - timedelta(days=7)
    cdrs = []
    suspect_sims = set(random.sample([s.imsi for s in subs], k=max(5, len(subs) // 500)))
    suspect_imei = [gen_imei() for _ in range(30)]

    for i in range(n):
        s = random.choice(subs)
        cell = random.choice(by_city_cells[s.city])
        ts = start + timedelta(seconds=random.randint(0, 7 * 24 * 3600))

        # NEW: expanded call type weights
        ctype = random.choices(CALL_TYPES, weights=[48, 16, 5, 22, 3, 6])[0]
        # voice=48%, sms=16%, mms=5%, data=22%, ussd=3%, rcs=6%

        imei_used = s.imei
        if s.imsi in suspect_sims and random.random() < 0.35:
            imei_used = random.choice(suspect_imei)

        dur = 0
        bytes_ = 0

        # NEW: Voice bearer & messaging metadata
        voice_bearer = None
        voice_codec = None
        voice_qci = None
        voice_sig_proto = None
        mms_content = None
        mms_size_bytes = 0
        mms_delivery = None
        rcs_type = None

        if ctype == "voice":
            dur = random.choice([random.randint(5, 120), random.randint(120, 1800)])

            if cell.tech == "6G":
                voice_bearer = "Vo6G"
            elif cell.tech == "5G" and s.priority_qos >= 6:
                voice_bearer = "VoNR"
            elif cell.tech == "4G":
                if s.line_class in ("Government", "Emergency", "VIP"):
                    voice_bearer = "VoLTE"
                else:
                    voice_bearer = random.choices(
                        ["VoLTE", "VoWiFi", "CSFB"], weights=[0.70, 0.15, 0.15]
                    )[0]
            elif cell.tech == "3G":
                voice_bearer = random.choices(["CSFB", "VoWiFi"], weights=[0.85, 0.15])[0]
            else:  # 2G
                voice_bearer = "CSFB"

            bearer_cfg = VOICE_BEARERS[voice_bearer]
            voice_codec = random.choice(bearer_cfg["codec"])
            voice_qci = bearer_cfg["qci"]
            voice_sig_proto = bearer_cfg["sig"]

        elif ctype == "sms":
            dur = 0
            bytes_ = random.randint(60, 480)

        elif ctype == "mms":
            mms_content = random.choice(MMS_CONTENT_TYPES)
            lo, hi = MMS_SIZE_BY_TYPE[mms_content]
            mms_size_bytes = random.randint(lo, hi)
            bytes_ = mms_size_bytes
            dur = random.randint(2, 30)
            mms_delivery = random.choices(
                ["DELIVERED", "EXPIRED", "REJECTED", "PENDING"], weights=[0.88, 0.05, 0.05, 0.02]
            )[0]

        elif ctype == "rcs":
            rcs_type = random.choice(RCS_TYPES)
            dur = random.randint(1, 10)
            bytes_ = random.randint(2_000, 500_000)

        elif ctype == "data":
            dur = random.randint(10, 600)
            bytes_ = random.randint(50_000, 50_000_000)

        else:  # ussd
            dur = random.randint(3, 15)

        result = (
            random.choices(["ANSWERED", "NO_ANSWER", "BUSY", "FAILED"], weights=[80, 10, 5, 5])[0]
            if ctype == "voice"
            else "OK"
        )
        if ctype == "mms":
            result = mms_delivery

        sig = signal_metrics(cell.tech)

        # Special-line behaviour
        is_international = s.international_access and ctype == "data" and random.random() < 0.25
        clir_used = False
        if (
            s.clir_enabled
            and ctype in ("voice", "sms", "mms")
            and random.random() < 0.35
            or (not s.clir_enabled)
            and ctype in ("voice", "sms", "mms")
            and random.random() < 0.01
        ):
            clir_used = True

        if s.direct_routing:
            routing_class = "Direct"
        elif s.priority_qos >= 6:
            routing_class = "Priority"
        elif is_international:
            routing_class = "International"
        else:
            routing_class = "Normal"

        filter_applied = not s.filter_bypass

        # Encryption status
        encrypted = s.e2e_enabled and ctype in ("voice", "sms", "mms", "data", "rcs")
        cipher_used = s.cipher_suite if encrypted else "None"
        enc_failure = encrypted and random.random() < 0.01
        if enc_failure:
            encrypted = False
            cipher_used = "FAILED"

        cdrs.append(
            {
                "record_id": f"CDR-{i+1:08d}",
                "timestamp": ts.strftime("%Y-%m-%d %H:%M:%S"),
                "msisdn": s.msisdn,
                "imsi": s.imsi,
                "imei": imei_used,
                "called": gen_msisdn() if ctype in ("voice", "sms", "mms", "rcs") else None,
                "call_type": ctype,
                "duration_sec": dur,
                "bytes": bytes_,
                "result": result,
                "cell_id": cell.cell_id,
                "tech": cell.tech,
                "city": cell.city,
                "lat": cell.lat,
                "lon": cell.lon,
                "rssi": sig.get("rssi"),
                "rsrp": sig.get("rsrp"),
                "rsrq": sig.get("rsrq"),
                "sinr": sig.get("sinr"),
                "clir_used": clir_used,
                "routing_class": routing_class,
                "filter_applied": filter_applied,
                "international": is_international,
                "line_class": s.line_class,
                "priority_qos": s.priority_qos,
                "encrypted": encrypted,
                "cipher_suite": cipher_used,
                "encryption_failure": enc_failure,
                "key_id": s.key_id if s.e2e_enabled else "",
                "voice_bearer": voice_bearer,
                "voice_codec": voice_codec,
                "voice_qci": voice_qci,
                "voice_sig_proto": voice_sig_proto,
                "mms_content_type": mms_content,
                "mms_size_bytes": mms_size_bytes,
                "mms_delivery": mms_delivery,
                "rcs_type": rcs_type,
            }
        )
    return cdrs

telecom_net_sim.main

main(argv=None)

Source code in telecom_net_sim.py

def main(argv=None):
    global SEED
    args = parse_args(argv)
    if args.random_seed:
        SEED = random.randint(1, 10**9)
    else:
        SEED = args.seed
    random.seed(SEED)

    print("=" * 78)
    print(f" {OPERATOR}-NET-SIM v3.0 | LOCAL-ONLY | Seed={SEED}")
    print("=" * 78)

    print("[1/7] Building network topology...")
    cells, cores = build_topology()
    print(f"      Cells: {len(cells)} | Core nodes: {len(cores)}")

    print("[2/7] Generating subscribers...")
    subs = build_subscribers(args.subs)

    print("[3/7] Generating CDR records...")
    cdrs = gen_cdrs(subs, cells, n=args.cdrs)
    print(f"      {len(cdrs)} records generated.")

    print("[4/7] Running OSINT analysis...")
    osint_data = osint_collect(cells, subs)
    osint_r = osint_report(osint_data)

    print("[5/7] Running SIGINT analysis...")
    sigint_r = sigint_analyze(cdrs, cells)

    print("[5.5/7] Running alert engine + SMS dispatch...")
    alerts = run_alert_engine(subs, cdrs, sigint_r, osint_r)
    print(f"        {len(alerts)} alerts dispatched via SMS.")

    print("[6/7] Persisting to SQLite...")
    persist(cells, cores, subs, cdrs, osint_data, osint_r, sigint_r, alerts)

    print("[7/7] Rendering charts and report...")
    visualize(cells, cdrs, sigint_r, osint_r)
    build_report(cells, cores, subs, cdrs, osint_r, sigint_r, alerts)

    print("[8/8] Running attack simulation...")
    try:
        import telecom_attack

        atk = telecom_attack.run_attack_simulation(
            n_scenarios=25, samples_per_scenario=20, inject_alerts=True
        )
        print(
            f"        {atk['scenarios']} scenarios, {atk['events']} events, "
            f"{atk['alerts']} alerts."
        )
    except Exception as e:
        print(f"        [i] Attack simulation skipped: {e}")

    print("\n[OK] Simulation completed successfully.")
    print(f"[DIR] Outputs: {os.path.abspath(OUT)}")