catarrh/simple
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

net: network-state monitor and permanent-rejection FPLMN (v2.7.0)

Browse Source 
server:
- netsim: LOCI/PSLOCI dummy builders take a status; the EPSLOCI dummy is
  wiped to 0B F6 + FF per UICC_NAA.md C3; roaming_denied now emulates the
  permanent 'PLMN not allowed' rejection (C3a): status 010, EPSNSC dropped,
  EF.FPLMN append with TS 31.102 4.2.16 shift semantics and a home-PLMN
  guard (HPLMNwAcT/EHPLMN, IMSI fallback), optional 'Rejection: write FPLMN'
  toggle; write steps carry the logical key; SCENARIO_SERVICE map;
  insert_fplmn/fplmn_entries/parse_imsi helpers
- netstate.py: cached per-session monitor state for the 12 network EFs,
  step-based patch, simulated service state, derived location with country/
  operator (optional MCC/MNC list) and roaming class
- server: monitor read at equip right after a readable ICCID (skipped
  otherwise), cleared on card removal; GET /api/net-state and POST
  /api/net-state-refresh; net-sim patches the cache from the written bytes
  and re-reads EF.IMSI; Location-status events set the service state and
  re-read EF.IMSI (multi-IMSI applets)

frontend:
- Phone tab: 'Network state' panel next to Network simulation with the
  simulated service badge (Undefined until simulated; normal/limited/no
  service + rejection marker), location/roaming line, compact per-file
  summaries with full-decode tooltips and a Refresh button; no card polling
- EF decoders: EF.FPLMN (FFFFFF gaps are not terminators) and EF.EHPLMN
- i18n EN/RU, help updated; SW cache simple-v203

tests: 295 Python / 453 frontend
This commit is contained in:
Антон Трошин
2026-09-20 12:26:17 +03:00
parent 81b50199cc
commit b1539d7cd5
15 changed files with 1190 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pysim_simple_server/netsim.py
+145 -18
View File
@@ -24,10 +24,15 @@ import time
# Candidate paths per logical file. The first existing one is used, so a
# USIM card is served from ADF.USIM and a GSM SIM from DF.GSM/DF.TELECOM.
FILE_PATHS = {
'imsi': ['ADF.USIM/6F07', 'DF.GSM/6F07'],
'ehplmn': ['ADF.USIM/6FD9'],
'spdi': ['ADF.USIM/6FCD'],
'hplmnwact': ['ADF.USIM/6F62', 'DF.GSM/6F62'],
'epsnsc': ['ADF.USIM/6FE4'],
'loci': ['ADF.USIM/6F7E', 'DF.GSM/6F7E'],
'psloci': ['ADF.USIM/6F73', 'DF.GSM/6F73'],
'epsloci': ['ADF.USIM/6FE3'],
'fplmn': ['ADF.USIM/6F7B', 'DF.GSM/6F7B'],
'kc': ['ADF.USIM/4F20', 'DF.GSM/6F08'],
'kcgprs': ['ADF.USIM/4F52', 'DF.GSM/6F09'],
'smsstatus': ['ADF.USIM/6F43', 'DF.TELECOM/6F43', 'DF.GSM/6F43'],
@@ -35,6 +40,19 @@ FILE_PATHS = {
'cbmir': ['ADF.USIM/6F50', 'DF.GSM/6F50'],
}
# Which simulated service state each scenario establishes (Network state
# panel). Scenarios not listed (churn, cb_reconfig, authenticate) do not
# touch the network registration and leave the previous state.
SCENARIO_SERVICE = {
'cold_boot': 'none',
'attach_eps': 'normal',
'attach_2g': 'normal',
'service_lost': 'none',
'limited_service': 'limited',
'roaming_denied': 'limited',
'sms_received': 'normal',
}
# File status codes (LOCI/PSLOCI/EPSLOCI update status).
ST_UPDATED = 0x00
ST_NOT_UPDATED = 0x01
@@ -91,6 +109,19 @@ def _plmn_bytes(plmn_hex):
return bytes.fromhex(_norm_hex(plmn_hex, 3))
def parse_imsi(data_hex):
"""EF.IMSI content -> IMSI digits, or None. Byte 1 is the length; the
high nibble of byte 2 is the parity/identity nibble (TS 31.102 4.2.2)."""
h = _norm_hex(data_hex)
if len(h) < 4:
return None
body = h[2:]
swapped = ''.join(body[i + 1] + body[i]
for i in range(0, len(body) - 1, 2))
digits = re.sub(r'F+$', '', swapped)
return digits[1:] or None
# ---- EPS NAS Security Context (EF.EPSNSC, TS 31.102 4.2.92) ----
@@ -150,9 +181,10 @@ def build_loci(tmsi_hex, plmn_hex, lac_hex, status=ST_UPDATED, rfu=0xFF):
+ bytes([rfu & 0xFF, status & 0xFF])).hex().upper()
def build_loci_dummy(plmn_hex):
"""Service lost: TMSI FF, PLMN kept, LAC FFFE, status 01."""
return build_loci('FFFFFFFF', plmn_hex, 'FFFE', ST_NOT_UPDATED)
def build_loci_dummy(plmn_hex, status=ST_NOT_UPDATED):
"""Service lost: TMSI FF, PLMN kept, LAC FFFE, status 01 (010 = PLMN not
allowed on a permanent rejection; UICC_NAA.md C3/C3a)."""
return build_loci('FFFFFFFF', plmn_hex, 'FFFE', status)
def build_psloci(ptmsi_hex, sig_hex, plmn_hex, lac_hex, rac_hex,
@@ -165,9 +197,8 @@ def build_psloci(ptmsi_hex, sig_hex, plmn_hex, lac_hex, rac_hex,
+ bytes([status & 0xFF])).hex().upper()
def build_psloci_dummy(plmn_hex):
return build_psloci('FFFFFFFF', 'FFFFFF', plmn_hex, 'FFFE', 'FF',
ST_NOT_UPDATED)
def build_psloci_dummy(plmn_hex, status=ST_NOT_UPDATED):
return build_psloci('FFFFFFFF', 'FFFFFF', plmn_hex, 'FFFE', 'FF', status)
def build_epsloci(guti_hex, plmn_hex, tac_hex, status=ST_UPDATED):
@@ -177,10 +208,39 @@ def build_epsloci(guti_hex, plmn_hex, tac_hex, status=ST_UPDATED):
+ bytes([status & 0xFF])).hex().upper()
def build_epsloci_dummy(plmn_hex):
"""GUTI header 0B F6 + PLMN, identity wiped; TAI PLMN + FFFF; status 01."""
return build_epsloci('0BF6' + plmn_hex + 'FF' * 7, plmn_hex, 'FFFF',
ST_NOT_UPDATED)
def build_epsloci_dummy(status=None):
"""EPSLOCI dummy: the EPS-mobile-identity pair `0B F6` (content length +
GUTI type octet) is kept, the GUTI/TAI/status bytes are wiped
(UICC_NAA.md C3). A rejection status (010 = roaming not allowed) is the
only byte written after the wipe (C3a)."""
out = '0BF6' + 'FF' * (15 if status is not None else 16)
if status is not None:
out += '%02X' % (status & 0xFF)
return out
def fplmn_entries(data_hex):
"""EF.FPLMN content as 3-byte PLMN entries; 'FFFFFF' marks an empty slot
(TS 31.102 4.2.16: valid in any position, never a terminator)."""
h = _norm_hex(data_hex)
return [h[i:i + 6] for i in range(0, len(h) - 5, 6)]
def insert_fplmn(data_hex, plmn_hex):
"""Store a denied PLMN per TS 31.102 4.2.16: fill the first empty slot,
otherwise shift the list left and append (the longest-held entry is
lost). Returns the full updated EF content."""
plmn = _norm_hex(plmn_hex, 3)
entries = fplmn_entries(data_hex)
if not entries:
return plmn
try:
idx = entries.index('FFFFFF')
except ValueError:
entries = entries[1:] + [plmn]
else:
entries[idx] = plmn
return ''.join(entries)
# ---- Ciphering keys and CB/SMS files ----
@@ -419,7 +479,7 @@ class NetSimRunner:
data = _pad_ff(bytes.fromhex(data_hex), size).hex().upper()
_out, sw = self.lchan.update_binary(data)
return self._check(data, sw, action='update_binary', file=label or key,
path=path)
key=key, path=path)
def write_record(self, key, data_hex, record=1, pad=True, label=None, optional=False):
try:
@@ -435,12 +495,68 @@ class NetSimRunner:
data = _pad_ff(bytes.fromhex(data_hex), size).hex().upper()
_out, sw = self.lchan.update_record(record, data)
return self._check(data, sw, action='update_record', file=label or key,
path=path, record=record)
key=key, path=path, record=record)
def read_binary_current(self):
data, sw = self.lchan.read_binary()
return (data or ''), sw
# -- forbidden PLMNs (permanent #11 rejection, UICC_NAA.md C3a)
def home_plmns(self):
"""3-byte HPLMN/EHPLMN entries from the cached Network-state monitor;
TS 23.122: the home network is never stored in EF.FPLMN."""
out = set()
files = ((getattr(self.srv, 'net_state', None) or {}).get('files') or {})
def flat(key):
f = files.get(key) or {}
if f.get('present') and f.get('kind') == 'transparent':
return _norm_hex(f.get('data'))
return None
hp = flat('hplmnwact')
if hp:
# HPLMNwAcT records are 5 bytes (PLMN + access technology); the
# first record is the HPLMN (TS 31.102 4.2.5).
out.add(hp[0:6])
ehp = flat('ehplmn')
if ehp:
out.update(fplmn_entries(ehp))
if not out:
# Fallback: the HPLMN is the IMSI's MCC/MNC. The IMSI does not
# encode the MNC length, so both interpretations are guarded.
imsi = parse_imsi(flat('imsi'))
if imsi and len(imsi) >= 5 and imsi[:3].isdigit():
try:
out.add(plmn_bcd(imsi[0:3], imsi[3:5]))
if len(imsi) >= 6 and imsi[5].isdigit():
out.add(plmn_bcd(imsi[0:3], imsi[3:6]))
except ValueError:
pass
return {h for h in out if h and h != 'FFFFFF'}
def write_fplmn(self, plmn_hex, optional=True):
path = None
try:
path = self._open('fplmn')
except StepError as e:
if optional:
self._add('skip', file='fplmn', note=str(e))
return None
raise
plmn = _norm_hex(plmn_hex, 3)
if plmn in self.home_plmns():
self._add('skip', file='fplmn',
note='home PLMN is never stored (TS 23.122)')
return None
size = self.lchan.selected_file_size()
data, sw = self.read_binary_current()
if sw != '9000' or not data:
data = 'FF' * (size or 12)
new_data = insert_fplmn(data, plmn)
return self.write_binary('fplmn', new_data, pad=False, label='fplmn')
def read_record_current(self, record=1):
data, sw = self.lchan.read_record(record)
return (data or ''), sw
@@ -499,11 +615,16 @@ class NetSimRunner:
label='epsloci', optional=True)
def write_dummy_locations(self, status=ST_NOT_UPDATED):
self.write_binary('loci', build_loci_dummy(self.plmn), label='loci',
optional=True)
self.write_binary('psloci', build_psloci_dummy(self.plmn), label='psloci',
optional=True)
self.write_binary('epsloci', build_epsloci_dummy(self.plmn),
"""Service loss: LOCI/PSLOCI keep the PLMN with the dummy status 01;
EPSLOCI is wiped to `0B F6` + FF (UICC_NAA.md C3). A rejection status
(010 = PLMN not allowed) is written to all three; EPSLOCI then carries
that status byte as the only byte after the wipe (C3a)."""
eps_status = None if status == ST_NOT_UPDATED else status
self.write_binary('loci', build_loci_dummy(self.plmn, status),
label='loci', optional=True)
self.write_binary('psloci', build_psloci_dummy(self.plmn, status),
label='psloci', optional=True)
self.write_binary('epsloci', build_epsloci_dummy(eps_status),
label='epsloci', optional=True)
def invalidate_kc(self):
@@ -571,12 +692,18 @@ class NetSimRunner:
self.write_dummy_locations()
def sc_roaming_denied(self):
# Permanent rejection (NAS cause #11): the Location status event is
# indistinguishable from limited service (TS 102 223 8.27) - the
# difference is the 010 status bytes and the EF.FPLMN entry
# (UICC_NAA.md C3a).
if self.p('send_event', True):
self.send_location_status(self.status(LOC_STATUS_LIMITED))
if self.p('invalidate_epsnsc', True):
self.invalidate_epsnsc(keep_key=bool(self.p('keep_kasme', True)))
self.invalidate_epsnsc(keep_key=False)
if self.p('dummy_locations', True):
self.write_dummy_locations(status=self.status(ST_PLMN_NOT_ALLOWED))
if self.p('write_fplmn', True):
self.write_fplmn(self.plmn)
def sc_churn(self):
count = int(self.p('churn_count', 3))
pysim_simple_server/netstate.py
+289
View File
@@ -0,0 +1,289 @@
# coding=utf-8
"""Network-state monitor for the SIMple lab.
Keeps a cached, per-card-session view of the network-related EFs: the files
the network simulator writes plus IMSI / EHPLMN / SPDI / HPLMNwAcT / FPLMN
(trace study: ``projects/UICC_NAA.md``). It also tracks the *simulated*
service state (normal / limited / no service, ``Undefined`` until something
is simulated) and derives the current location + roaming view for the Phone
tab's "Network state" panel.
This module is pure: the server owns all card I/O (``_netstate_read`` in
``server.py``) and passes freshly read file entries in. ``mcc_mnc_data`` is
the optional MCC/MNC operator list the server loads (``--mcc-mnc-list``).
"""
import re
import time
from pysim_simple_server import netsim
SERVICE_NORMAL = 'normal'
SERVICE_LIMITED = 'limited'
SERVICE_NONE = 'none'
def _def(key, name, fid):
return {'key': key, 'name': name, 'fid': fid,
'paths': list(netsim.FILE_PATHS.get(key) or [])}
# Display order of the monitor panel.
FILE_DEFS = [
_def('imsi', 'EF.IMSI', '6F07'),
_def('ehplmn', 'EF.EHPLMN', '6FD9'),
_def('spdi', 'EF.SPDI', '6FCD'),
_def('hplmnwact', 'EF.HPLMNwAcT', '6F62'),
_def('loci', 'EF.LOCI', '6F7E'),
_def('psloci', 'EF.PSLOCI', '6F73'),
_def('epsloci', 'EF.EPSLOCI', '6FE3'),
_def('epsnsc', 'EF.EPSNSC', '6FE4'),
_def('cbmi', 'EF.CBMI', '6F45'),
_def('cbmir', 'EF.CBMIR', '6F50'),
_def('smsstatus', 'EF.SMSS', '6F43'),
_def('fplmn', 'EF.FPLMN', '6F7B'),
]
MONITORED_KEYS = [d['key'] for d in FILE_DEFS]
FILE_BY_KEY = {d['key']: d for d in FILE_DEFS}
def _norm(value):
return re.sub(r'[^0-9a-fA-F]', '', value or '').upper()
def new_state():
"""Empty monitor state (a new card session)."""
return {'files': {},
'service': {'state': None, 'source': None, 'time': None},
'network': None, 'read_at': None}
def set_service(state, service, source=None):
"""Record the simulated service state ('normal'/'limited'/'none')."""
if not state:
return
state['service'] = {'state': service, 'source': source, 'time': time.time()}
def set_read_time(state):
if state:
state['read_at'] = time.time()
def patch_file(state, key, path, data, source='write', record=None):
"""Patch one cached file from bytes we just wrote (no re-read needed)."""
if not state or key not in FILE_BY_KEY:
return
d = FILE_BY_KEY[key]
cur = (state.get('files') or {}).get(key) or {}
if record is not None or cur.get('kind') == 'record':
records = {r.get('num'): r.get('data', '')
for r in (cur.get('records') or [])}
records[int(record or 1)] = _norm(data)
body = {'kind': 'record',
'records': [{'num': n, 'data': records[n]}
for n in sorted(records)]}
else:
body = {'kind': 'transparent', 'data': _norm(data)}
entry = {'name': d['name'], 'fid': d['fid'],
'path': path or cur.get('path') or (d['paths'][0] if d['paths'] else None),
'present': True, 'source': source, 'updated': time.time()}
entry.update(body)
state.setdefault('files', {})[key] = entry
def apply_steps(state, steps):
"""Patch the cached files from a net-sim step log (writes we performed)."""
if not state:
return
for s in steps or []:
act = s.get('action')
key = s.get('key')
if act in ('update_binary', 'update_record') and key in FILE_BY_KEY:
patch_file(state, key, s.get('path'), s.get('data') or '',
source='write',
record=s.get('record') if act == 'update_record' else None)
def merge_read(state, entries, source='read'):
"""Merge freshly read file entries (server I/O) into the state."""
if not state or not entries:
return
for key, entry in entries.items():
if not entry:
continue
entry = dict(entry)
entry['source'] = source
state.setdefault('files', {})[key] = entry
# ---- decoding helpers (mirror the client-side EF decoders) ----
def parse_imsi(data_hex):
"""EF.IMSI digits or None (TS 31.102 4.2.2)."""
return netsim.parse_imsi(data_hex)
def plmn_from_hex(hex3):
"""3-byte PLMN (TS 24.008 BCD) -> {'mcc','mnc','plmn'} or None."""
h = _norm(hex3)
if len(h) != 6 or h == 'FFFFFF':
return None
b = [int(h[i:i + 2], 16) for i in (0, 2, 4)]
d = lambda v: v & 0x0F
e = lambda v: (v >> 4) & 0x0F
if d(b[0]) > 9 or e(b[0]) > 9 or d(b[1]) > 9:
return None
if d(b[2]) > 9 or e(b[2]) > 9:
return None
mcc = '%d%d%d' % (d(b[0]), e(b[0]), d(b[1]))
mnc = '%d%d' % (d(b[2]), e(b[2]))
mnc3 = e(b[1])
if mnc3 != 0x0F:
if mnc3 > 9:
return None
mnc += '%d' % mnc3
return {'mcc': mcc, 'mnc': mnc, 'plmn': mcc + mnc}
def _plmn_hex_list(data_hex, rec=3):
"""Valid 3-byte PLMN entries of a transparent list file ('FFFFFF' gaps
are skipped, never treated as terminators)."""
h = _norm(data_hex)
step = rec * 2
return [h[i:i + 6] for i in range(0, len(h) - 5, step)
if plmn_from_hex(h[i:i + 6])]
def plmn_list(data_hex, rec=3):
return [plmn_from_hex(h) for h in _plmn_hex_list(data_hex, rec)]
def _transparent(files, key):
f = (files or {}).get(key) or {}
if f.get('present') and f.get('kind') == 'transparent':
return f
return None
def _current_location(files):
"""Current PLMN + area, EPSLOCI (TAI) -> PSLOCI (RAI) -> LOCI (LAI).
Dummies keep the PLMN in LOCI/PSLOCI while EPSLOCI is wiped, so the
fallback order keeps the location known after service loss."""
for key, area, off in (('epsloci', 'TAI', 12), ('psloci', 'RAI', 7),
('loci', 'LAI', 4)):
f = _transparent(files, key)
if not f:
continue
h = _norm(f.get('data'))
if len(h) < (off + 5) * 2:
continue
p = plmn_from_hex(h[off * 2:off * 2 + 6])
if not p:
continue
return {'plmn': p['plmn'], 'mcc': p['mcc'], 'mnc': p['mnc'],
'hex': h[off * 2:off * 2 + 6], 'area': area,
'lac': h[off * 2 + 6:off * 2 + 10], 'source': f.get('source')}
return None
def _home_sets(files):
"""(home PLMN hex, equivalent-home PLMN hex) sets from HPLMNwAcT, EHPLMN
and the IMSI (fallback), as 3-byte uppercase hex."""
home, eq = set(), set()
f = _transparent(files, 'hplmnwact')
if f:
# First 5-byte record is the HPLMN (TS 31.102 4.2.5).
head = _norm(f.get('data'))[0:6]
if plmn_from_hex(head):
home.add(head)
f = _transparent(files, 'ehplmn')
if f:
eq.update(_plmn_hex_list(f.get('data')))
if not home:
imsi = parse_imsi((_transparent(files, 'imsi') or {}).get('data'))
if imsi and len(imsi) >= 5 and imsi[:3].isdigit():
try:
home.add(netsim.plmn_bcd(imsi[0:3], imsi[3:5]))
if len(imsi) >= 6 and imsi[5].isdigit():
home.add(netsim.plmn_bcd(imsi[0:3], imsi[3:6]))
except ValueError:
pass
return home, eq
def _rejected(files):
"""A permanent 'PLMN not allowed' rejection fingerprint: status 010 in a
location file (UICC_NAA.md C3a) or a non-empty EF.FPLMN."""
off = {'loci': 10, 'psloci': 13, 'epsloci': 17}
for key, idx in off.items():
f = _transparent(files, key)
if not f:
continue
h = _norm(f.get('data'))
if len(h) >= (idx + 1) * 2 and h[idx * 2:idx * 2 + 2] == '02':
return True
f = _transparent(files, 'fplmn')
if f and _plmn_hex_list(f.get('data')):
return True
return False
def _operator(mcc_mnc_data, mcc, mnc):
"""Exact MCC/MNC lookup in the optional operator list; a country-only
fallback (first entry of that MCC) keeps the location line useful."""
if not mcc_mnc_data:
return None
try:
mcc_i, mnc_i = int(mcc), int(mnc)
except (TypeError, ValueError):
return None
fallback = None
for e in mcc_mnc_data:
try:
if int(str(e.get('mcc') or '').strip() or -1) != mcc_i:
continue
except ValueError:
continue
if fallback is None:
fallback = e
try:
if int(str(e.get('mnc') or '').strip() or -1) == mnc_i:
return {'country': e.get('countryName'),
'operator': e.get('brand') or e.get('operator')}
except ValueError:
continue
if fallback:
return {'country': fallback.get('countryName'), 'operator': None}
return None
def compute_network(state, mcc_mnc_data=None):
"""Derive the monitor header data from the cached state."""
state = state or {}
files = state.get('files') or {}
loc = _current_location(files)
home, eq = _home_sets(files)
roaming = None
if loc and home:
if loc['hex'] in home:
roaming = 'home'
elif loc['hex'] in eq:
roaming = 'equivalent'
else:
roaming = 'guest'
location = None
if loc:
op = _operator(mcc_mnc_data, loc['mcc'], loc['mnc'])
location = {'plmn': loc['plmn'], 'mcc': loc['mcc'], 'mnc': loc['mnc'],
'area': loc['area'], 'lac': loc['lac'],
'country': (op or {}).get('country'),
'operator': (op or {}).get('operator'),
'roaming': roaming, 'rejected': _rejected(files),
'source': loc.get('source')}
network = {'service': state.get('service') or
{'state': None, 'source': None, 'time': None},
'location': location}
state['network'] = network
return network
pysim_simple_server/server.py
+160 -2
View File
@@ -15,6 +15,7 @@ from pySim.transport import ApduTracer, ProactiveHandler
from pySim.cards import UiccCardBase
from pysim_simple_server import httpota
from pysim_simple_server import netsim
from pysim_simple_server import netstate
from pysim_simple_server import scp81
from smartcard.CardMonitoring import CardMonitor, CardObserver
@@ -24,7 +25,7 @@ from osmocom.construct import GsmOrUcs2Adapter
from osmocom.tlv import BER_TLV_IE
VERSION = '2.6.2'
VERSION = '2.7.0'
MAX_ENVELOPE_SEGMENTS = 5 # max SMS segments for outgoing C-APDU in ENVELOPE
@@ -491,6 +492,105 @@ def _mcc_mnc_random(data, exclude=None):
('countryName', 'countryCode', 'mcc', 'mnc', 'brand', 'operator', 'status')}
def _netstate_read(server, keys=None):
"""Read the monitored network-state EFs (best effort per file).
The Network state panel caches them: the full set is read once at equip
(after a readable ICCID), on demand via /api/net-state-refresh, and only
the card-side files (EF.IMSI) after net-sim / Location-status operations.
"""
app = getattr(server, 'app', None)
rs = getattr(app, 'rs', None) if app else None
if not app or not rs:
return {}
lchan = rs.lchan[0]
out = {}
for d in netstate.FILE_DEFS:
key = d['key']
if keys and key not in keys:
continue
entry = {'name': d['name'], 'fid': d['fid'], 'present': False,
'kind': None, 'source': 'refresh', 'updated': time.time()}
for path in d['paths']:
cleanup = None
try:
_, cleanup = _select_path(lchan, path, app)
ft = _get_file_type(lchan, lchan.selected_file)
entry['path'] = path
if ft in ('linear_fixed', 'cyclic'):
records = []
n = lchan.selected_file_num_of_rec() or 1
for i in range(1, n + 1):
rv = lchan.read_record(i)
data = rv[0] if isinstance(rv, tuple) else rv
records.append({'num': i,
'data': (data or '').upper()})
entry.update({'present': True, 'kind': 'record',
'records': records})
else:
rv = lchan.read_binary()
data = rv[0] if isinstance(rv, tuple) else rv
entry.update({'present': True, 'kind': 'transparent',
'data': (data or '').upper()})
break
except Exception:
continue
finally:
if cleanup:
try:
cleanup()
except Exception:
pass
out[key] = entry
return out
def _netstate_compute(server):
state = getattr(server, 'net_state', None)
if state is None:
return None
data = _mcc_mnc_load(getattr(server, 'mcc_mnc_path', None))
return netstate.compute_network(state, data)
def _netstate_after_net_sim(server, scenario, result):
"""Update the cached Network state after a scenario run: the bytes we
wrote are known without re-reading; the card may update EF.IMSI itself."""
state = getattr(server, 'net_state', None)
if state is None:
return
netstate.apply_steps(state, (result or {}).get('steps') or [])
service = netsim.SCENARIO_SERVICE.get(scenario)
if service and (result or {}).get('success'):
netstate.set_service(state, service, 'net-sim:' + scenario)
netstate.merge_read(state, _netstate_read(server, ['imsi']), source='read')
_netstate_compute(server)
def _netstate_after_event(server, event_type, event_data):
"""Location status changes the simulated service state; the card may also
switch EF.IMSI (multi-IMSI applets) after such an event."""
state = getattr(server, 'net_state', None)
if state is None:
return
try:
ev = int(event_type)
except (TypeError, ValueError):
ev = None
if ev == netsim.EVENT_LOCATION_STATUS and event_data:
status = None
if (len(event_data) >= 3 and event_data[0] == 0x9B
and event_data[1] == 0x01):
status = event_data[2]
service = {0x00: netstate.SERVICE_NORMAL,
0x01: netstate.SERVICE_LIMITED,
0x02: netstate.SERVICE_NONE}.get(status)
if service:
netstate.set_service(state, service, 'event:location-status')
netstate.merge_read(state, _netstate_read(server, ['imsi']), source='read')
_netstate_compute(server)
def _parse_tree_output(output):
lines = (output or '').split('\n')
children = []
@@ -2350,6 +2450,7 @@ def _handle_card_disconnect():
_server_ref.event_list = None
_server_ref.sim_menu = None
_server_ref.iccid = None
_server_ref.net_state = None
_server_ref.equipping = False
_server_ref.card_session = getattr(_server_ref, 'card_session', 0) + 1
_reset_proactive_log()
@@ -2377,8 +2478,21 @@ def _apply_equipped_card(server):
server.iccid = _read_iccid(server.app)
if server.iccid:
_tlog('equip: ICCID %s' % server.iccid)
# Network state monitor: read the network-related EFs right after the
# ICCID (still before the TERMINAL PROFILE opens a CAT session). A
# card without a readable ICCID is considered unusable - give up.
try:
server.net_state = netstate.new_state()
netstate.merge_read(server.net_state, _netstate_read(server),
source='init')
netstate.set_read_time(server.net_state)
_netstate_compute(server)
except Exception as e:
server.net_state = None
_tlog('equip: network state read failed: %s' % e)
else:
_tlog('equip: ICCID not readable')
server.net_state = None
_tlog('equip: ICCID not readable - network state skipped')
_poll_enable()
sm, el = _send_terminal_profile(server.scc, server.terminal_profile)
server.sim_menu = sm
@@ -3219,6 +3333,16 @@ class PysimHandler(BaseHTTPRequestHandler):
self._send_json(resp)
self._log_resp({'available': resp['available'],
'results': len(resp.get('results', []))})
elif self.path == '/api/net-state':
self._log_req()
state = getattr(self.server, 'net_state', None)
if state is None:
self._send_json({'available': False, 'state': None})
self._log_resp({'available': False})
return
_netstate_compute(self.server)
self._send_json({'available': True, 'state': state})
self._log_resp({'available': True})
elif self.path == '/api/status':
self._log_req()
app = self.server.app
@@ -3474,6 +3598,8 @@ class PysimHandler(BaseHTTPRequestHandler):
result = netsim.run_scenario(
sys.modules[__name__], app, scenario, body,
event_list=getattr(self.server, 'event_list', None) or [])
_netstate_after_net_sim(self.server, scenario, result)
result['net_state'] = getattr(self.server, 'net_state', None)
self._send_json(result)
self._log_resp({'scenario': scenario,
'success': result.get('success'),
@@ -3486,6 +3612,30 @@ class PysimHandler(BaseHTTPRequestHandler):
_handle_card_disconnect()
self._send_json({'error': 'simulation failed: %s' % e}, 500)
self._log_resp({'error': str(e)})
elif self.path == '/api/net-state-refresh':
if not self.server.app or not getattr(self.server, 'scc', None):
self._send_json({'error': _err('no_card_state', lang)}, 503)
self._log_resp({'error': _err('no_card_state', lang)})
return
body = self._read_body()
self._log_req(body)
keys = body.get('files') if isinstance(body, dict) else None
state = getattr(self.server, 'net_state', None)
if state is None:
state = netstate.new_state()
self.server.net_state = state
try:
files = _netstate_read(self.server, keys)
except Exception as e:
self._send_json({'error': str(e)}, 500)
self._log_resp({'error': str(e)})
return
netstate.merge_read(state, files, source='refresh')
netstate.set_read_time(state)
_netstate_compute(self.server)
resp = {'available': True, 'state': state}
self._send_json(resp)
self._log_resp({'available': True, 'files': len(files or {})})
elif self.path == '/api/rescue':
scc = self.server.scc
if not scc:
@@ -3878,9 +4028,17 @@ class PysimHandler(BaseHTTPRequestHandler):
event_data = bytes.fromhex(event_data_hex) if event_data_hex else None
try:
data, sw = _send_event_download(scc, event_type, event_data)
try:
is_location = int(event_type) == netsim.EVENT_LOCATION_STATUS
except (TypeError, ValueError):
is_location = False
if is_location:
_netstate_after_event(self.server, event_type, event_data)
resp = {'sw': sw}
if data:
resp['data'] = data
if getattr(self.server, 'net_state', None) is not None:
resp['net_state'] = self.server.net_state
self._send_json(resp)
self._log_resp(resp)
except Exception as e: