ria-ran--sim-drop/tests/test_saip.py

import pytest
import yaml

from server.app import create_app
from server.profile_generator import (
    derive_opc, enc_iccid, enc_imsi, generate_iccid, generate_ki,
)
import server.saip_profile as saip

_OP = bytes.fromhex("63bfa50ee6523365ff14c1f45f88737d")

saip_required = pytest.mark.skipif(
    not saip.saip_available(),
    reason="SAIP unavailable (asn1tools + SAIP ASN.1 schema not installed)",
)


def _write_config(tmp_path, profile_format="v2"):
    cfg = {
        "network": {"mcc": "302", "mnc": "720", "plmn": "302720",
                    "op_key": "63bfa50ee6523365ff14c1f45f88737d"},
        "wifi": {"ssid": "x", "channel": 6, "ip": "192.168.4.1",
                 "subnet": "192.168.4.0/24", "dhcp_range": "a,b"},
        "server": {"host": "0.0.0.0", "port": 5000, "debug": False},
        "database": {"path": str(tmp_path / "subscribers.db")},
        "rate_limiting": {"max_profiles_per_mac": 1, "window_hours": 1},
        "core": {"type": "none"},
        "profile": {"carrier_name": "Ad-Hoc", "profile_name": "Emergency 5G",
                    "apn": "internet", "smdp_address": "esim.local",
                    "format": profile_format},
    }
    path = tmp_path / "config.yaml"
    path.write_text(yaml.safe_dump(cfg))
    return str(path)


class TestSaipEncoding:
    def test_unavailable_raises(self, monkeypatch):
        monkeypatch.setattr(saip, "_compiler", lambda: None)
        with pytest.raises(RuntimeError, match="unavailable"):
            saip.create_saip_profile("302720000000001", generate_ki(),
                                     b"\x00" * 16, generate_iccid())

    def test_decode_unavailable_raises(self, monkeypatch):
        monkeypatch.setattr(saip, "_compiler", lambda: None)
        with pytest.raises(RuntimeError, match="unavailable"):
            saip.decode_saip_profile(b"\x00")

    def test_find_asn_honours_env_override(self, tmp_path, monkeypatch):
        fake = tmp_path / "schema.asn"
        fake.write_text("-- placeholder\n")
        monkeypatch.setenv("SAIP_ASN_PATH", str(fake))
        assert saip.find_saip_asn() == str(fake)

    def test_find_asn_ignores_missing_override(self, monkeypatch):
        monkeypatch.setenv("SAIP_ASN_PATH", "/no/such/schema.asn")
        # Falls through to the pySim lookup (or None); never returns the bad path.
        assert saip.find_saip_asn() != "/no/such/schema.asn"

    @saip_required
    def test_produces_der_bytes(self):
        ki = generate_ki()
        opc = derive_opc(ki, _OP)
        iccid = generate_iccid()
        pkg = saip.create_saip_profile("302720000000001", ki, opc, iccid, "Emergency 5G", "C")
        assert isinstance(pkg, bytes) and len(pkg) > 0
        # First byte is the context tag for the 'header' CHOICE alternative.
        assert pkg[0] == 0xA0

    @saip_required
    def test_full_sequence_decodes(self):
        ki = generate_ki()
        opc = derive_opc(ki, _OP)
        iccid = generate_iccid()
        pkg = saip.create_saip_profile("302720000000001", ki, opc, iccid, "P", "C")
        pes = saip.decode_saip_profile(pkg)
        assert [choice for choice, _ in pes] == ["header", "mf", "usim", "akaParameter", "end"]
        by = dict(pes)
        assert by["header"]["iccid"] == enc_iccid(iccid)
        assert by["header"]["major-version"] == 3
        assert by["header"]["eUICC-Mandatory-GFSTEList"] == [
            saip.MF_TEMPLATE_OID, saip.USIM_TEMPLATE_OID]
        assert by["mf"]["ef-iccid"] == [("fillFileContent", enc_iccid(iccid))]

    @saip_required
    def test_truncated_profile_yields_prefix_not_crash(self):
        pkg = saip.create_saip_profile("302720000000001", generate_ki(),
                                       b"\x22" * 16, generate_iccid())
        full = saip.decode_saip_profile(pkg)
        # Cutting mid-element must not raise; it returns the complete prefix.
        partial = saip.decode_saip_profile(pkg[:-3])
        assert len(partial) < len(full)
        assert [c for c, _ in partial] == [c for c, _ in full][:len(partial)]
        assert saip.decode_saip_profile(b"\xa0\x82") == []  # bogus length, no crash

    @saip_required
    def test_imsi_embedded_in_usim(self):
        imsi = "302720123456789"
        pkg = saip.create_saip_profile(imsi, generate_ki(), b"\x11" * 16, generate_iccid())
        usim = dict(saip.decode_saip_profile(pkg))["usim"]
        assert usim["templateID"] == saip.USIM_TEMPLATE_OID
        assert usim["ef-imsi"] == [("fillFileContent", enc_imsi(imsi))]

    @saip_required
    def test_credentials_embedded(self):
        ki = generate_ki()
        opc = derive_opc(ki, _OP)
        pkg = saip.create_saip_profile("302720000000001", ki, opc, generate_iccid(), "P", "C")
        assert ki in pkg   # raw Ki bytes present in the AKA parameter
        assert opc in pkg


class TestSaipAppIntegration:
    @saip_required
    def test_provision_emits_saip_when_configured(self, tmp_path):
        app = create_app(_write_config(tmp_path, profile_format="saip"))
        client = app.test_client()
        r = client.post("/api/provision", json={"mac_address": "aa:bb:cc:00:11:22"})
        assert r.status_code == 200
        profile = client.get(r.get_json()["profile_url"]).data
        assert profile[0] == 0xA0  # DER, not JSON

    def test_falls_back_to_v2_when_saip_unavailable(self, tmp_path, monkeypatch):
        import server.app as appmod
        monkeypatch.setattr(appmod, "saip_available", lambda: False)
        app = create_app(_write_config(tmp_path, profile_format="saip"))
        client = app.test_client()
        r = client.post("/api/provision", json={"mac_address": "aa:bb:cc:00:11:33"})
        assert r.status_code == 200
        profile = client.get(r.get_json()["profile_url"]).data
        assert profile.lstrip()[:1] == b"{"  # JSON profile-package-v2