ria-ran--sim-drop/server/saip_profile.py

"""Optional real SGP.22 SAIP profile-package generation.

Encodes a SIMalliance Interoperable Profile (SAIP) ``ProfileElement`` sequence
using the official ASN.1 schema (``PE_Definitions``) via ``asn1tools``. Unlike
``profile_generator.create_esim_profile`` (the always-available JSON
``profile-package-v2``), this produces real DER-encoded SGP.22 profile elements.

This module is optional and self-contained: it needs only ``asn1tools`` plus the
SAIP ASN.1 schema file. The schema ships with osmocom pySim; we locate it there
without importing pySim's Python package (which pulls in a card-reader stack).
Set ``SAIP_ASN_PATH`` to override the schema location. When neither is present,
``saip_available()`` returns False and callers fall back to profile-package-v2.

Scope: the encoded package currently carries the ProfileHeader (with ICCID) and
the Milenage AKA parameters (Ki/OPc) — the authentication-critical material.
Writing IMSI and other EFs requires USIM-template / GenericFileManagement PEs
with full File Control Parameters; that is the next layer and is not yet emitted.
"""
import functools
import importlib.util
import logging
import os
from typing import Optional

logger = logging.getLogger(__name__)

_ASN_RELPATH = os.path.join("esim", "asn1", "saip", "PE_Definitions-3.3.1.asn")

# Standard "created by default" templates (TCA eUICC prefix 2.23.143.1 +
# templates.2 + N). The eUICC creates the file structure from the template; the
# profile only personalizes the EFs it fills (here EF.ICCID at MF, EF.IMSI at
# ADF.USIM).
MF_TEMPLATE_OID = "2.23.143.1.2.1"
USIM_TEMPLATE_OID = "2.23.143.1.2.4"


def find_saip_asn() -> Optional[str]:
    """Locate the SAIP ASN.1 schema: SAIP_ASN_PATH override, else pySim's copy."""
    override = os.environ.get("SAIP_ASN_PATH")
    if override and os.path.exists(override):
        return override
    try:
        spec = importlib.util.find_spec("pySim")
    except (ImportError, ValueError):
        return None
    if spec and spec.submodule_search_locations:
        for loc in spec.submodule_search_locations:
            candidate = os.path.join(loc, _ASN_RELPATH)
            if os.path.exists(candidate):
                return candidate
    return None


@functools.lru_cache(maxsize=1)
def _compiler():
    """Return a compiled asn1tools spec, or None if prerequisites are missing."""
    try:
        import asn1tools
    except ImportError:
        return None
    path = find_saip_asn()
    if not path:
        return None
    try:
        return asn1tools.compile_files([path], codec="der")
    except Exception as exc:  # malformed/incompatible schema
        logger.warning(f"Failed to compile SAIP ASN.1 schema at {path}: {exc}")
        return None


def saip_available() -> bool:
    """True if a real SAIP profile can be generated in this environment."""
    return _compiler() is not None


def create_saip_profile(
    imsi: str,
    ki: bytes,
    opc: bytes,
    iccid: str,
    profile_name: str = "",
    carrier_name: str = "",
) -> bytes:
    """Build a DER-encoded SAIP ProfileElement sequence.

    Emits: ProfileHeader (ICCID), PE-MF (personalizing EF.ICCID against the MF
    template), PE-USIM (personalizing EF.IMSI against the USIM template), Milenage
    PE-AKAParameter (Ki/OPc), and PE-End.
    """
    asn = _compiler()
    if asn is None:
        raise RuntimeError(
            "SAIP profile generation unavailable: requires asn1tools and the SAIP "
            "ASN.1 schema (install osmocom pySim or set SAIP_ASN_PATH)"
        )

    # Imported here to avoid a hard dependency cycle at module import time.
    from server.profile_generator import enc_iccid, enc_imsi

    ki_bytes = ki if isinstance(ki, bytes) else bytes.fromhex(ki)
    opc_bytes = opc if isinstance(opc, bytes) else bytes.fromhex(opc)

    header = ("header", {
        "major-version": 3,
        "minor-version": 3,
        "profileType": (profile_name or "Emergency 5G")[:100],
        "iccid": enc_iccid(iccid),
        "eUICC-Mandatory-services": {"usim": None, "milenage": None},
        # Declare the file-system templates this profile depends on.
        "eUICC-Mandatory-GFSTEList": [MF_TEMPLATE_OID, USIM_TEMPLATE_OID],
    })
    mf = ("mf", {
        "mf-header": {"identification": 1},
        "templateID": MF_TEMPLATE_OID,
        "mf": [],
        "ef-iccid": [("fillFileContent", enc_iccid(iccid))],
        # Other mandatory files are left to the template (empty = no override).
        "ef-dir": [], "ef-arr": [],
    })
    usim = ("usim", {
        "usim-header": {"identification": 2},
        "templateID": USIM_TEMPLATE_OID,
        "adf-usim": [],
        "ef-imsi": [("fillFileContent", enc_imsi(imsi))],
        # Other mandatory files are left to the template (empty = no override).
        "ef-arr": [], "ef-ust": [], "ef-spn": [], "ef-est": [],
        "ef-acc": [], "ef-ecc": [],
    })
    aka = ("akaParameter", {
        "aka-header": {"identification": 3},
        "algoConfiguration": ("algoParameter", {
            "algorithmID": 1,            # Milenage
            "algorithmOptions": b"\x00",
            "key": ki_bytes,
            "opc": opc_bytes,
        }),
    })
    end = ("end", {"end-header": {"identification": 4}})

    logger.debug(f"Encoding SAIP profile for IMSI {imsi} (ICCID {iccid})")
    pes = (header, mf, usim, aka, end)
    return b"".join(asn.encode("ProfileElement", pe) for pe in pes)


def _split_der(data: bytes):
    """Yield byte-range slices of consecutive top-level DER TLVs in ``data``.

    Stops cleanly at the last complete element so a truncated or malformed tail
    yields the valid prefix instead of raising.
    """
    i, n = 0, len(data)
    while i < n:
        start = i
        first = data[i]
        i += 1
        if first & 0x1F == 0x1F:  # high-tag-number form: skip continuation octets
            while i < n and data[i] & 0x80:
                i += 1
            i += 1  # final tag byte
        if i >= n:  # no room for a length octet
            return
        length_byte = data[i]
        i += 1
        if length_byte & 0x80:  # long-form length
            num = length_byte & 0x7F
            if i + num > n:
                return
            length = int.from_bytes(data[i:i + num], "big")
            i += num
        else:
            length = length_byte
        if i + length > n:  # truncated content
            return
        i += length
        yield data[start:i]


def decode_saip_profile(data: bytes) -> list:
    """Decode a SAIP ProfileElement sequence into a list of (choice, value).

    asn1tools decodes a single value, so we split the concatenated PEs by their
    DER TLV boundaries and decode each one.
    """
    asn = _compiler()
    if asn is None:
        raise RuntimeError("SAIP decoding unavailable (asn1tools/schema missing)")
    return [asn.decode("ProfileElement", pe) for pe in _split_der(data)]