localStorage и сохраняется между перезагрузками.Подключение к локальному pysim-otaman-server для работы с картой: введите URL сервера (по умолчанию http://127.0.0.1:8080) и нажмите Подключиться. Область статуса показывает состояние ридера/карты, а Подключить карту (пере)инициализирует карту после вставки. Подвкладки: Файловый менеджер, Командная строка pySim и Отправка APDU. «Профайлер» и «Симулятор телефона» — отдельные вкладки верхнего уровня.
4.1 Файловый менеджер
-Дерево файловой системы отображается слева; выбор файла открывает панель деталей справа. Элементы сгруппированы: DF выше EF, сортировка по FID или символьному имени (пиллы над деревом; выбор сохраняется в localStorage). При выборе файла над содержимым также показываются FID, тип файла, размер / структура записей и декодированный FCI.
Дерево файловой системы отображается слева; выбор файла открывает панель деталей справа. Элементы сгруппированы: DF выше EF, сортировка по FID или символьному имени (пиллы и кнопка «Проверить все файлы» закреплены над прокручиваемым деревом; выбор сохраняется в localStorage). При выборе файла над содержимым также показываются FID, тип файла, размер / структура записей и декодированный FCI.
- Прочитать — чтение файла (автоопределение transparent/record)
- Редактировать — изменение hex-данных, Сохранить для записи (или Отмена) @@ -429,7 +429,7 @@
- Header — the app version, an INSTALL PWA button (shown when the browser offers installation, enabling offline use), links to the project on GitHub and to this help, an EN/RU language toggle, and a dark/light theme toggle. +
- Header — the app version, an INSTALL PWA button (shown when the browser offers installation, enabling offline use), links to the project on GitHub and to this help, an EN/RU language toggle, and a dark/light theme toggle. Next to the title a small status indicator shows the server/card state (gray dot = connecting, red dot = no server connection, card icons = no card / equipping / equipped) and a compact ADM ✓ (green, administrator PIN verified) or ADM ✗ (red, not verified) badge for the current card session.
- Language and theme choices are stored in
localStorageand persist across reloads. - Top-level tabs: Remote APDU (SIM RFM, USIM RFM, Expanded Script, RAM/GP, HTTP OTA, C-APDU Parser, Response parser), SCP80 (Secured Packet, RAM), SCP81 (Listener, Scripts), Cards, Profiler (list tabs Profiles, Card snapshots, Custom files), Card reader (File manager, pySim command line, Raw APDU), and Phone simulator.
- The help link opens this documentation at the section matching the current view (e.g. the Profiler tab opens §5). @@ -319,7 +319,7 @@
- Read — reads the selected file (auto-detects transparent vs record files)
- Edit — modify hex data, Save to write back (or Cancel) @@ -429,7 +429,7 @@
Отправка события использует ENVELOPE(Event Download) по TS 102 223 / TS 131 111.
6.3 Журнал проактивных команд
-Хронологический список извлечённых проактивных команд. Каждая строка показывает время, код типа, имя и декодированный квалификатор (для команд, у которых он есть). Для команд с данными ответа показывается строка Ответ: с байтами TERMINAL RESPONSE (без служебных TLV); ответы PROVIDE LOCAL INFORMATION декодируются через словарь данных PLI.
Хронологический список извлечённых проактивных команд. Каждая строка показывает время, код типа, имя и декодированный квалификатор (для команд, у которых он есть). Раскрытие строки показывает декодированную команду: текст и пункты DISPLAY TEXT / SELECT ITEM / SET UP MENU, параметры BIP-каналов, действия TIMER MANAGEMENT, названия квалификаторов PROVIDE LOCAL INFORMATION (все стандартные, включая ESN 07, MEID 0B и Supported RATs 1A) и SEND SHORT MESSAGE — разбирается SMS TPDU (тип, TP-MR, TP-DA, TP-PID, TP-DCS, срок действия, TP-UDL), а пользовательские данные показываются как текст для текстовых кодировок (GSM 7-бит, UCS2, 8-бит) или как secured-пакет (TS 31.115) при TP-PID = 7F (SIM data download); заголовки конкатенации UDH декодируются. Для команд с данными ответа показывается строка Ответ: с байтами TERMINAL RESPONSE (без служебных TLV); ответы PROVIDE LOCAL INFORMATION декодируются через словарь данных PLI.
6.4 Опрос STATUS
Кнопка Отправить STATUS отправляет STATUS (F2) вручную. Переключатель Опрос включает фоновый опрос: после настраиваемого интервала бездействия (аргумент сервера --poll-interval, 1–255 с, по умолчанию 30 с, 0 отключает опрос) сервер отправляет STATUS и обрабатывает любую ожидающую проактивную команду. При извлечении карты опрос останавливается, а состояние карты сбрасывается.
1.1 Interface
-
-
Connects to a local pysim-otaman-server for live card operations: enter the server URL (default http://127.0.0.1:8080) and press Connect. The status area shows the reader/card state, and Equip card (re)initializes the card after insertion. Sub-tabs: File manager, pySim command line, and Raw APDU. The Profiler and Phone simulator are separate top-level tabs.
4.1 File manager
-The file system tree is displayed on the left; selecting a file opens its detail pane on the right. Entries are grouped with DFs above EFs and sorted by FID or symbolic Name (pills above the tree; the choice is remembered in localStorage). Selecting a file also shows its FID, file type, size / record layout and the decoded FCI above the content pane.
The file system tree is displayed on the left; selecting a file opens its detail pane on the right. Entries are grouped with DFs above EFs and sorted by FID or symbolic Name (pills pinned above the scrolling tree together with Probe all files; the choice is remembered in localStorage). Selecting a file also shows its FID, file type, size / record layout and the decoded FCI above the content pane.
Sending an event uses ENVELOPE(Event Download) per TS 102 223 / TS 131 111.
6.3 Proactive command log
-Chronological list of fetched proactive commands. Each row shows the elapsed time, type code, name, and a decoded qualifier (for commands that have one). Commands with response data show a Response: line with the TERMINAL RESPONSE bytes (boilerplate TLVs stripped); PROVIDE LOCAL INFORMATION responses are decoded using the PLI data dictionary decoders.
Chronological list of fetched proactive commands. Each row shows the elapsed time, type code, name, and a decoded qualifier (for commands that have one). Expanding a row shows the decoded command: DISPLAY TEXT / SELECT ITEM / SET UP MENU text and items, BIP channel parameters, TIMER MANAGEMENT actions, PROVIDE LOCAL INFORMATION qualifier names (all standard qualifiers, including ESN 07, MEID 0B and Supported RATs 1A), and SEND SHORT MESSAGE — the SMS TPDU is parsed (type, TP-MR, TP-DA, TP-PID, TP-DCS, validity period, TP-UDL) with the user data shown as text for text codings (GSM 7-bit, UCS2, 8-bit) or as a secured packet (TS 31.115) when TP-PID = 7F (SIM data download); UDH concatenation headers are decoded. Commands with response data show a Response: line with the TERMINAL RESPONSE bytes (boilerplate TLVs stripped); PROVIDE LOCAL INFORMATION responses are decoded using the PLI data dictionary decoders.
6.4 STATUS polling
A Send STATUS button issues a manual STATUS (F2). A Polling toggle enables background polling: after a configurable idle interval (server CLI --poll-interval, 1–255 s, default 30 s, 0 disables polling) the server sends STATUS and handles any pending proactive command. Polling stops and card state resets if the card is removed.
-
OTAMan SIM OTA with a Human Face v2.2.0
+OTAMan SIM OTA with a Human Face v2.2.1
●
+
github
@@ -824,15 +825,19 @@
+
github
@@ -824,15 +825,19 @@
-
-
- Sort:
- FID
- Name
+
-
+
+
+
+ Sort:
+ FID
+ Name
+
+
+
+
+
-
-
-
@@ -939,6 +944,7 @@
@@ -959,7 +965,8 @@
PSK keys come from the card presets (Cards tab): the listener picks the key by the identity the card sends in the TLS handshake.
+ PSK keys come from the card presets (Cards tab): the listener picks the key by the identity the card sends in the TLS handshake.
+ Pass-through: every BIP channel the card opens is connected to this Host:Port (the external HTTP OTA platform); TLS is terminated there, the card's requested address is only logged.
@@ -4585,6 +4592,7 @@ async function pysimFetch(path, body) {
let _pysimServerAvailable = null; // null until probed
let _pysimCardEquipped = false;
let _pysimEquipping = false;
+let _pysimAdmVerified = null; // null = no card session
let _pysimAvailabilityTimer = null;
function pysimAvailabilityState() {
@@ -4638,9 +4646,33 @@ function pysimApplyAvailability() {
pysimUpdateStateIndicator();
}
+// Compact ADM state next to the header's card indicator: "ADM ✓" when the
+// administrator PIN was verified (pySim rs.adm_verified), "ADM ✗" otherwise.
+// Hidden without a card session; only rewrites the DOM when the state changes
+// (the 2s /api/status poll calls this on every update).
+function pysimUpdateAdmIndicator(status) {
+ const el = document.getElementById('state-indicator-adm');
+ if (!el) return;
+ const verified = (status && status.connected) ? !!status.adm_verified : null;
+ if (verified === _pysimAdmVerified) return;
+ _pysimAdmVerified = verified;
+ el.classList.remove('text-emerald-600', 'dark:text-emerald-400', 'text-red-500');
+ if (verified === null) {
+ el.classList.add('hidden');
+ el.removeAttribute('title');
+ return;
+ }
+ el.textContent = verified ? 'ADM ✓' : 'ADM ✗';
+ el.classList.remove('hidden');
+ el.classList.add(verified ? 'text-emerald-600' : 'text-red-500');
+ if (verified) el.classList.add('dark:text-emerald-400');
+ el.setAttribute('title', t(verified ? 'ADM verified' : 'ADM not verified'));
+}
+
function pysimSetServerAvailable(available) {
if (_pysimServerAvailable === available) return;
_pysimServerAvailable = available;
+ if (available !== true) pysimUpdateAdmIndicator(null);
pysimApplyAvailability();
}
@@ -6851,15 +6883,18 @@ const EVENT_FORMS = {
const PLI_QUALIFIERS = [
{code:'00',name:'Location Info (MCC, MNC, LAC/TAC, Cell ID)'},{code:'01',name:'IMEI'},
{code:'02',name:'Network Measurement results'},{code:'03',name:'Date, time and time zone'},
- {code:'04',name:'Language setting'},{code:'05',name:'Timing Advance'},
- {code:'06',name:'Access Technology (single)'},{code:'08',name:'IMEISV'},
+ {code:'04',name:'Language setting'},{code:'05',name:'Reserved for GSM (Timing Advance)'},
+ {code:'06',name:'Access Technology (single)'},{code:'07',name:'ESN of the terminal'},
+ {code:'08',name:'IMEISV'},
{code:'09',name:'Search Mode'},{code:'0A',name:'Battery charge state'},
+ {code:'0B',name:'MEID of the terminal'},
{code:'0C',name:'Current WSID'},{code:'0D',name:'Broadcast Network info'},
{code:'0E',name:'Multiple Access Technologies'},{code:'0F',name:'Location Info (multi-RAT)'},
{code:'10',name:'NMR (multi-RAT)'},{code:'11',name:'CSG ID list + HNB name'},
{code:'12',name:'H(e)NB IP address'},{code:'13',name:'H(e)NB surrounding macrocells'},
{code:'14',name:'Current WLAN identifier'},{code:'15',name:'Slices information'},
{code:'16',name:'CAG information list'},{code:'17',name:'Rejected slices information'},
+ {code:'1A',name:'Supported Radio Access Technologies'},
];
async function pysimEventsRender() {
@@ -7085,6 +7120,7 @@ function pysimCardStateUpdate(status) {
_pysimServerAvailable = true;
_pysimCardEquipped = !!status.connected;
_pysimEquipping = !!status.equipping;
+ pysimUpdateAdmIndicator(status);
pysimApplyAvailability();
if (pysimProactiveSeqChanged(status.proactive_seq)
&& isViewVisible('tab-phone') && isViewVisible('phone-sub-phone')) {
@@ -7180,11 +7216,12 @@ const CMD_QUALIFIER_SHORT = {
'23': {0x00:'Digits+Echo', 0x01:'Alpha+Echo', 0x04:'NoEcho'},
'24': {0x00:'Normal', 0x02:'Help', 0x04:'DefaultItem', 0x80:'Nav'},
'25': {0x00:'Normal', 0x04:'Help', 0x80:'Nav'},
- '26': {0x00:'Loc', 0x01:'IMEI', 0x02:'NMR', 0x03:'Time', 0x04:'Lang', 0x05:'TA',
- 0x06:'AccTech', 0x08:'IMEISV', 0x09:'Search', 0x0A:'Batt', 0x0C:'WSID',
+ '26': {0x00:'Loc', 0x01:'IMEI', 0x02:'NMR', 0x03:'Time', 0x04:'Lang', 0x05:'TA (GSM)',
+ 0x06:'AccTech', 0x07:'ESN', 0x08:'IMEISV', 0x09:'Search', 0x0A:'Batt',
+ 0x0B:'MEID', 0x0C:'WSID',
0x0D:'BCInfo', 0x0E:'MultiAT', 0x0F:'MultiLoc', 0x10:'MultiNMR',
0x11:'CSG', 0x12:'HNB-IP', 0x13:'HNB-Macro', 0x14:'WLAN', 0x15:'Slices',
- 0x16:'CAG', 0x17:'RejSlice'},
+ 0x16:'CAG', 0x17:'RejSlice', 0x1A:'SupRAT'},
'27': {0x00:'Start', 0x01:'Deactivate', 0x02:'Get'},
};
@@ -7293,6 +7330,10 @@ function scp81ModeChanged() {
const mode = document.getElementById('scp81-mode').value;
const row = document.getElementById('scp81-script-row');
if (row) row.style.display = (mode === 'tls') ? '' : 'none';
+ const pskNote = document.getElementById('scp81-psk-note');
+ if (pskNote) pskNote.classList.toggle('hidden', mode !== 'tls');
+ const directNote = document.getElementById('scp81-passthru-note');
+ if (directNote) directNote.classList.toggle('hidden', mode !== 'passthru');
}
function scp81SwitchSubtab(name) {
@@ -7666,10 +7707,16 @@ async function scp81LogRefresh() {
async function scp81Start() {
const mode = document.getElementById('scp81-mode').value;
+ const hostVal = document.getElementById('scp81-host').value.trim();
+ const portVal = document.getElementById('scp81-port').value.trim();
+ if (mode === 'passthru' && (!hostVal || !portVal)) {
+ scp81Msg(t('Pass-through requires the target host and port'), 'text-red-500');
+ return;
+ }
const body = {
action: 'start', mode: mode,
- host: document.getElementById('scp81-host').value.trim() || '127.0.0.1',
- port: parseInt(document.getElementById('scp81-port').value.trim() || '8443', 10),
+ host: hostVal || '127.0.0.1',
+ port: parseInt(portVal || '8443', 10),
};
if (mode === 'tls') {
const map = cardsPskMap();
@@ -10734,6 +10781,11 @@ const LANG_RU = {
'LOAD blocks': 'Блоки LOAD',
'clamped from': 'ограничено с',
'auto-fit': 'авто',
+ 'Pass-through (external server)': 'Проброс (внешний сервер)',
+ 'Pass-through: every BIP channel the card opens is connected to this Host:Port (the external HTTP OTA platform); TLS is terminated there, the card\'s requested address is only logged.': 'Проброс: каждый открываемый картой BIP-канал подключается к этому Host:Port (внешняя платформа HTTP OTA); TLS завершается там, запрошенный картой адрес только журналируется.',
+ 'Pass-through requires the target host and port': 'Для режима проброса нужны host и port внешнего сервера',
+ 'ADM verified': 'ADM подтверждён',
+ 'ADM not verified': 'ADM не подтверждён',
};
let currentLang = 'en';
diff --git a/frontend/sw.js b/frontend/sw.js
index 8dec65a..615c3e7 100644
--- a/frontend/sw.js
+++ b/frontend/sw.js
@@ -1,4 +1,4 @@
-const CACHE = 'otaman-v161';
+const CACHE = 'otaman-v165';
const URLS = [
'index.html',
'help.html',
diff --git a/frontend/tests/card_state.test.js b/frontend/tests/card_state.test.js
index db6bda8..31c4968 100644
--- a/frontend/tests/card_state.test.js
+++ b/frontend/tests/card_state.test.js
@@ -23,29 +23,52 @@ function extractFunc(src, name) {
let code = 'var _pysimCardStateKey = null;\nvar _pysimCardSession = null;\n'
+ 'var _pysimServerAvailable = null;\nvar _pysimCardEquipped = false;\n'
- + 'var _pysimProactiveSeq = null;\nvar _pysimStkSig = null;\n';
+ + 'var _pysimProactiveSeq = null;\nvar _pysimStkSig = null;\nvar _pysimAdmVerified = null;\n';
code += extractFunc(html, 'pysimCardStateUpdate') + '\n';
code += extractFunc(html, 'pysimAvailabilityState') + '\n';
code += extractFunc(html, 'pysimControlDisabled') + '\n';
code += extractFunc(html, 'pysimProactiveSeqChanged') + '\n';
code += extractFunc(html, 'pysimStkStatusChanged') + '\n';
+code += extractFunc(html, 'pysimUpdateAdmIndicator') + '\n';
+code += extractFunc(html, 'pysimSetServerAvailable') + '\n';
code += '\nglobalThis.esc = s => s;\n';
code += 'globalThis.t = s => s;\n';
eval(code);
+function fakeIndicator() {
+ const classes = new Set();
+ const el = {
+ classes, textContent: '', title: null,
+ classList: {
+ add: (...c) => c.forEach(x => classes.add(x)),
+ remove: (...c) => c.forEach(x => classes.delete(x)),
+ contains: c => classes.has(c),
+ },
+ setAttribute: (k, v) => { if (k === 'title') el.title = v; },
+ removeAttribute: (k) => { if (k === 'title') el.title = null; },
+ };
+ return el;
+}
+
function setup() {
const el = { textContent: 'status line', innerHTML: '' };
+ const adm = fakeIndicator();
const calls = { connected: [], resets: [], refreshStatus: [], proactive: 0 };
_pysimCardStateKey = null;
_pysimCardSession = null;
_pysimProactiveSeq = null;
- globalThis.document = { getElementById: () => el, querySelectorAll: () => [] };
+ _pysimAdmVerified = null;
+ _pysimServerAvailable = null;
+ globalThis.document = {
+ getElementById: id => id === 'state-indicator-adm' ? adm : el,
+ querySelectorAll: () => [],
+ };
globalThis.pysimSetConnected = v => calls.connected.push(v);
globalThis.pysimResetCardData = refresh => calls.resets.push(refresh);
globalThis.pysimApplyAvailability = () => {};
globalThis.isViewVisible = () => true;
globalThis.pysimProactiveLogRender = () => { calls.proactive++; };
- return { el, calls };
+ return { el, adm, calls };
}
function status(extra) {
@@ -168,3 +191,37 @@ test('pysimStkStatusChanged detects menu state transitions', () => {
assert.ok(pysimStkStatusChanged({ active: true, pending: false }));
assert.ok(!pysimStkStatusChanged(null));
});
+
+test('the header ADM badge shows verified / not verified / hidden', () => {
+ const { adm, calls } = setup();
+ pysimCardStateUpdate(status({ connected: true, adm_verified: true }));
+ assert.ok(!adm.classes.has('hidden'));
+ assert.strictEqual(adm.textContent, 'ADM ✓');
+ assert.ok(adm.classes.has('text-emerald-600'));
+ assert.ok(adm.classes.has('dark:text-emerald-400'));
+ assert.strictEqual(adm.title, 'ADM verified');
+ // an unchanged state must not rewrite the badge
+ adm.textContent = '';
+ pysimCardStateUpdate(status({ connected: true, adm_verified: true }));
+ assert.strictEqual(adm.textContent, '', 'unchanged ADM state rewrote the badge');
+ // verification lost (e.g. card reset)
+ pysimCardStateUpdate(status({ connected: true, adm_verified: false }));
+ assert.strictEqual(adm.textContent, 'ADM ✗');
+ assert.ok(adm.classes.has('text-red-500'));
+ assert.ok(!adm.classes.has('text-emerald-600'));
+ assert.strictEqual(adm.title, 'ADM not verified');
+ // no card session hides it
+ pysimCardStateUpdate(status({ connected: false }));
+ assert.ok(adm.classes.has('hidden'));
+ assert.strictEqual(adm.title, null);
+ // the ADM update must not disturb the connect/reset flow
+ assert.deepStrictEqual(calls.connected, [true, false]);
+});
+
+test('losing the server hides the ADM badge', () => {
+ const { adm } = setup();
+ pysimCardStateUpdate(status({ connected: true, adm_verified: true }));
+ assert.ok(!adm.classes.has('hidden'));
+ pysimSetServerAvailable(false);
+ assert.ok(adm.classes.has('hidden'));
+});
diff --git a/frontend/tests/html.test.js b/frontend/tests/html.test.js
index 9dacc2e..4cb8be7 100644
--- a/frontend/tests/html.test.js
+++ b/frontend/tests/html.test.js
@@ -25,6 +25,23 @@ test('cards list shows the SCP81 PSK column with blue/red row buttons', () => {
assert.match(fn[0], /cardsRemove\(' \+ i \+ '\)" class="[^"]*bg-red-600 text-white/);
});
+test('PLI qualifier tables cover all standard qualifiers', () => {
+ // ESN (07), MEID (0B) and Supported RATs (1A) must at least be named, in
+ // both the TR Config dictionary and the proactive-log short labels.
+ const pli = /const PLI_QUALIFIERS = \[([\s\S]*?)\];/.exec(html);
+ assert.ok(pli, 'PLI_QUALIFIERS not found');
+ for (const code of ['07', '0B', '1A']) {
+ assert.ok(pli[1].includes("{code:'" + code + "'"), 'PLI_QUALIFIERS missing ' + code);
+ }
+ const block = /const CMD_QUALIFIER_SHORT = \{([\s\S]*?)\n\};/.exec(html);
+ assert.ok(block, 'CMD_QUALIFIER_SHORT not found');
+ const short = /'26': \{([^}]*)\}/.exec(block[1]);
+ assert.ok(short, "CMD_QUALIFIER_SHORT['26'] not found");
+ for (const key of ['0x07', '0x0B', '0x1A']) {
+ assert.ok(short[1].includes(key + ':'), 'CMD_QUALIFIER_SHORT 26 missing ' + key);
+ }
+});
+
test('profile rows have a Clone action', () => {
assert.match(html, /onclick="profilerClone\(' \+ i \+ '\)"/);
assert.match(html, /t\('Clone'\)/);
@@ -62,12 +79,24 @@ test('header state indicator and profiler custom-files tab', () => {
assert.ok(!html.includes('data-pysim-sub="custom"'));
});
+test('header status indicator has a compact ADM badge', () => {
+ assert.ok(html.includes('id="state-indicator-adm"'));
+});
+
test('file manager has FID / Name sort pills', () => {
assert.match(html, /data-fs-sort="fid" onclick="pysimFsSetSort\('fid'\)"/);
assert.match(html, /data-fs-sort="name" onclick="pysimFsSetSort\('name'\)"/);
assert.ok(html.includes('pysim-fs-sort-pill'));
});
+test('file manager keeps sort/probe controls above the scrolling tree', () => {
+ // The sort pills and the Probe all files button/status must sit outside
+ // the scrolling tree container so they stay visible while it scrolls.
+ assert.ok(html.indexOf('id="pysim-fs-probe-btn"') < html.indexOf('id="pysim-fs-tree"'));
+ assert.ok(html.indexOf('pysim-fs-sort-pill') < html.indexOf('id="pysim-fs-tree"'));
+ assert.match(html, /style="max-height:65vh"[^>]*>\s*
/);
+});
+
test('custom files form has add/save and cancel controls', () => {
assert.match(html, /id="pysim-cf-add-btn"[^>]*data-l10n="Add"/);
assert.match(html, /id="pysim-cf-cancel-btn"[^>]*class="hidden[^"]*"[^>]*data-l10n="Cancel"/);
diff --git a/pyproject.toml b/pyproject.toml
index 55c0df1..8af706a 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "pysim-otaman-server"
-version = "2.2.0"
+version = "2.2.1"
description = "HTTP REST server wrapping pysim for the OTAMan PWA"
requires-python = ">=3.8"
# pysim is a git-only dependency installed explicitly by setup.bat/setup.sh.
diff --git a/pysim_otaman_server/server.py b/pysim_otaman_server/server.py
index e31737b..0924879 100644
--- a/pysim_otaman_server/server.py
+++ b/pysim_otaman_server/server.py
@@ -21,7 +21,7 @@ from osmocom.construct import GsmOrUcs2Adapter
from osmocom.tlv import BER_TLV_IE
-VERSION = '2.2.0'
+VERSION = '2.2.1'
MAX_ENVELOPE_SEGMENTS = 5 # max SMS segments for outgoing C-APDU in ENVELOPE
@@ -825,11 +825,13 @@ PLI_QUALIFIER_NAMES = {
0x02: 'Network Measurement results',
0x03: 'Date, time and time zone',
0x04: 'Language setting',
- 0x05: 'Timing Advance',
+ 0x05: 'Reserved for GSM (Timing Advance, TS 51 111)',
0x06: 'Access Technology (single)',
+ 0x07: 'ESN of the terminal',
0x08: 'IMEISV',
0x09: 'Search Mode',
0x0A: 'Battery charge state',
+ 0x0B: 'MEID of the terminal',
0x0C: 'Current WSID',
0x0D: 'Broadcast Network information',
0x0E: 'Multiple Access Technologies',
@@ -842,12 +844,18 @@ PLI_QUALIFIER_NAMES = {
0x15: 'Slices information',
0x16: 'CAG information list',
0x17: 'Rejected slices information',
+ 0x1A: 'Supported Radio Access Technologies',
}
_PLI_DATA = {q: '' for q in PLI_QUALIFIER_NAMES}
_BIP = httpota.BipTerminal()
_SCP81_LISTENER = None
+# Active listener mode and target: ('dump'|'tls'|'passthru', host, port).
+# passthru has no listener object - the BIP channels connect straight to the
+# external platform - so the mode/target are tracked here for the status API.
+_SCP81_MODE = None
+_SCP81_TARGET = None
# PSK table of the TLS listener: identity -> key (memory only, never logged or
# persisted; the PWA sends it from the card presets at listener start).
# _SCP81_PSK_LEGACY keeps a single-key start (psk_hex [+ psk_identity]) so an
@@ -1060,6 +1068,140 @@ def _hms_bcd(seconds):
_TIMER_ACTIONS = {0x00: 'Start', 0x01: 'Deactivate', 0x02: 'Get current value'}
+# SMS TPDU type (TS 23.040 TP-MTI, bits 1-2 of the first octet).
+_SMS_MTI = {0: 'SMS-DELIVER', 1: 'SMS-SUBMIT', 2: 'SMS-COMMAND', 3: 'Reserved'}
+
+
+def _unpack_septets(data, count):
+ """Unpack `count` 7-bit septets from the packed GSM default alphabet
+ (TS 23.038): septet n lives in bits 7n..7n+6 of the octet string."""
+ out = bytearray()
+ acc, bits = 0, 0
+ for b in data:
+ acc |= b << bits
+ bits += 8
+ while bits >= 7 and len(out) < count:
+ out.append(acc & 0x7F)
+ acc >>= 7
+ bits -= 7
+ return bytes(out)
+
+
+def _parse_udh(ud):
+ """Parse an SMS user-data header (TS 23.040 9.2.3.24).
+
+ Returns (fields, octets, septets): the decoded IEs, the UDH length in
+ octets and the number of septets it occupies in a 7-bit packed UD."""
+ if len(ud) < 2:
+ return [], 0, 0
+ udhl = ud[0]
+ if udhl < 1 or 1 + udhl > len(ud):
+ return [], 0, 0
+ out = []
+ off = 1
+ while off + 2 <= 1 + udhl:
+ iei, ielen = ud[off], ud[off + 1]
+ val = ud[off + 2: off + 2 + ielen]
+ if iei == 0x00 and ielen == 3:
+ out.append({'label': 'Concat (8-bit ref)',
+ 'value': '%d, part %d/%d' % (val[0], val[1], val[2])})
+ elif iei == 0x08 and ielen == 4:
+ out.append({'label': 'Concat (16-bit ref)',
+ 'value': '%d, part %d/%d' % (int.from_bytes(val[:2], 'big'),
+ val[2], val[3])})
+ else:
+ out.append({'label': 'UDH IE 0x%02X' % iei, 'value': val.hex().upper()})
+ off += 2 + ielen
+ octets = 1 + udhl
+ return out, octets, (octets * 8 + 6) // 7
+
+
+def _decode_sms_ud(pid, dcs, ud, udhi, udl):
+ """Decode the TP-UD of an SMS TPDU: the UDH, a secured packet (PID 0x7F,
+ TS 31.115) or the text message body for a text DCS (TS 23.038)."""
+ out = []
+ header_octets, header_septets = 0, 0
+ if udhi:
+ fields, header_octets, header_septets = _parse_udh(ud)
+ out.extend(fields)
+ data = ud[header_octets:]
+ if pid == 0x7F:
+ # SIM data download: the user data is a secured packet (TS 31.115).
+ out.append({'label': 'Secured packet (TS 31.115)',
+ 'value': '%d bytes: %s' % (len(data), data.hex().upper())})
+ return out
+ cls = dcs & 0x0C
+ try:
+ if cls == 0x00:
+ # GSM 7-bit default alphabet, packed septets; the UDH (if any)
+ # occupies whole septets at the start of the packed data.
+ n = udl if udl else (len(ud) * 8) // 7
+ septets = _unpack_septets(ud[: (n * 7 + 7) // 8], n)[header_septets:]
+ text = codecs.decode(septets, 'gsm03.38')
+ elif cls == 0x08:
+ text = codecs.decode(data, 'utf_16_be')
+ elif cls == 0x04:
+ text = data.decode('latin-1', errors='replace')
+ else:
+ out.append({'label': 'User data',
+ 'value': '%d bytes: %s' % (len(data), data.hex().upper())})
+ return out
+ out.append({'label': 'Text', 'value': text})
+ except Exception:
+ out.append({'label': 'User data',
+ 'value': '%d bytes: %s' % (len(data), data.hex().upper())})
+ return out
+
+
+def _decode_send_sm(raw):
+ """Decode a SEND SHORT MESSAGE command (TS 102 223 6.4.10): alpha
+ identifier, address and the 3GPP-SMS TPDU with its user data."""
+ out = []
+ tlvs = httpota.proactive_tlvs(raw)
+ alpha = _cmd_tlv(tlvs, 0x05)
+ if alpha:
+ try:
+ out.append({'label': 'Alpha', 'value': _STK_DECODE._decode(alpha, {}, 'stk')})
+ except Exception:
+ pass
+ addr = _cmd_tlv(tlvs, 0x06)
+ if addr:
+ try:
+ from pySim.cat import Address
+ a = Address().from_bytes(addr)
+ num = str(a.get('call_number') or '').rstrip('fF')
+ ton = (a.get('ton_npi') or {}).get('type_of_number')
+ out.append({'label': 'Address',
+ 'value': num + (' (%s)' % ton if ton else '')})
+ except Exception:
+ out.append({'label': 'Address', 'value': addr.hex().upper()})
+ tpdu = _cmd_tlv(tlvs, 0x0B)
+ if not tpdu:
+ return out
+ out.append({'label': 'SMS TPDU', 'value': tpdu.hex().upper()})
+ try:
+ mti = tpdu[0] & 0x03
+ out.append({'label': 'Type', 'value': _SMS_MTI.get(mti, 'Reserved')})
+ if mti == 1:
+ from pySim.sms import SMS_SUBMIT
+ s = SMS_SUBMIT.from_bytes(tpdu)
+ out.append({'label': 'TP-MR', 'value': str(s.tp_mr)})
+ if s.tp_da is not None:
+ num = str(getattr(s.tp_da, 'digits', '')).rstrip('fF')
+ out.append({'label': 'TP-DA', 'value': num})
+ out.append({'label': 'TP-PID', 'value': '0x%02X%s' % (
+ s.tp_pid, ' (SIM data download)' if s.tp_pid == 0x7F else '')})
+ out.append({'label': 'TP-DCS', 'value': '0x%02X' % s.tp_dcs})
+ if s.tp_vp is not None:
+ out.append({'label': 'TP-VP', 'value': bytes(s.tp_vp).hex().upper()})
+ out.append({'label': 'TP-UDL', 'value': str(s.tp_udl)})
+ out.extend(_decode_sms_ud(s.tp_pid, s.tp_dcs, bytes(s.tp_ud),
+ bool(s.tp_udhi), s.tp_udl))
+ except Exception:
+ # Malformed TPDU: keep the raw hex line above, never break the log.
+ pass
+ return out
+
def _decode_cmd(cmd_type, raw, qualifier):
"""Decode a fetched proactive command into [{label, value}] pairs."""
@@ -1079,11 +1221,7 @@ def _decode_cmd(cmd_type, raw, qualifier):
return [{'label': 'Events', 'value': ', '.join(names)}]
return []
if cmd_type == 0x13:
- idx = raw.find(b'\x8b')
- if idx >= 0 and idx + 1 < len(raw):
- tlen = raw[idx + 1]
- return [{'label': 'SMS TPDU', 'value': raw[idx + 2: idx + 2 + tlen].hex()}]
- return []
+ return _decode_send_sm(raw)
if cmd_type == 0x21:
text = _parse_display_text(raw)
return [{'label': 'Text', 'value': text}] if text else []
@@ -1323,6 +1461,10 @@ def _handle_bip_command(scc, cmd_num, cmd_type, cmd_qual, raw, dev_src, dev_dst)
def _scp81_listener_status():
if not _SCP81_LISTENER:
+ if _SCP81_MODE == 'passthru' and _SCP81_TARGET:
+ return {'mode': 'passthru', 'host': _SCP81_TARGET[0],
+ 'port': _SCP81_TARGET[1],
+ 'target': '%s:%d' % _SCP81_TARGET}
return None
if isinstance(_SCP81_LISTENER, scp81.PskTlsServer):
return {'mode': 'tls', 'host': _SCP81_LISTENER.host, 'port': _SCP81_LISTENER.port,
@@ -1846,6 +1988,7 @@ def _scp81_gen_install(body):
def _scp81_bip_control(body):
global _SCP81_LISTENER, _SCP81_PSKS, _SCP81_PSK_LEGACY
+ global _SCP81_MODE, _SCP81_TARGET
global _SCP81_SCRIPT_TEMPLATE, _SCP81_SCRIPT_CR_TAG, _SCP81_NEXT_URI
global _SCP81_LINK_EVENTS, _SCP81_TARGETED_APP, _SCP81_APACHE_HEADERS
global _SCP81_CHUNKED
@@ -1855,6 +1998,8 @@ def _scp81_bip_control(body):
if _SCP81_LISTENER:
_SCP81_LISTENER.stop()
_SCP81_LISTENER = None
+ _SCP81_MODE = None
+ _SCP81_TARGET = None
_BIP.disable()
return {'ok': True, 'bip': _BIP.status(), 'listener': None}
host = body.get('host') or '127.0.0.1'
@@ -1865,6 +2010,20 @@ def _scp81_bip_control(body):
_SCP81_LISTENER.stop()
_SCP81_LISTENER = None
_BIP.disable()
+ if mode == 'passthru':
+ # No local listener: the card's BIP channels connect straight to the
+ # external platform (e.g. a production HTTP OTA server), which
+ # terminates TLS and runs the administration dialog. The target is a
+ # configured address, never the address the card requests.
+ if not body.get('host') or body.get('port') in (None, ''):
+ return {'ok': False,
+ 'error': 'passthru mode requires the target host and port'}
+ _SCP81_MODE = 'passthru'
+ _SCP81_TARGET = (host, port)
+ _BIP.on_data = _bip_data_available
+ _BIP.enable(host, port)
+ return {'ok': True, 'bip': _BIP.status(),
+ 'listener': _scp81_listener_status()}
if mode == 'tls':
raw_map = body.get('psk_map')
table, err = _parse_psk_map(raw_map)
@@ -1941,6 +2100,8 @@ def _scp81_bip_control(body):
on_log=lambda kind, **fields: _BIP.log(kind, **fields))
_SCP81_PSKS = dict(_SCP81_LISTENER.psk_map)
_SCP81_PSK_LEGACY = None if table else (psk, identity)
+ _SCP81_MODE = 'tls'
+ _SCP81_TARGET = (_SCP81_LISTENER.host, _SCP81_LISTENER.port)
_BIP.on_data = _bip_data_available
_BIP.enable(host, _SCP81_LISTENER.port)
return {'ok': True, 'bip': _BIP.status(), 'listener': _scp81_listener_status(),
@@ -1958,6 +2119,8 @@ def _scp81_bip_control(body):
host, port,
on_rx=lambda peer, data: _BIP.log('dump-rx', peer=peer, bytes=len(data), hex=data.hex().upper()[:2000]),
on_log=lambda kind, **fields: _BIP.log(kind, **fields))
+ _SCP81_MODE = 'dump'
+ _SCP81_TARGET = (_SCP81_LISTENER.host, _SCP81_LISTENER.port)
_BIP.enable(host, _SCP81_LISTENER.port)
return {'ok': True, 'bip': _BIP.status(), 'listener': _scp81_listener_status()}
diff --git a/tests/test_httpota.py b/tests/test_httpota.py
index 76da1b4..07d723f 100644
--- a/tests/test_httpota.py
+++ b/tests/test_httpota.py
@@ -152,6 +152,32 @@ class BipTerminalTest(unittest.TestCase):
self.assertIn('close', kinds)
peer.stop()
+ def test_passthru_mode_roundtrip_via_bip_control(self):
+ # SCP81 passthru: the control API enables BIP with the external
+ # platform as the target and starts no local listener; the card's
+ # channel talks straight to that platform.
+ peer = PeerServer(greeting=b'PLATFORM')
+ peer.start()
+ try:
+ resp = server._scp81_bip_control({'action': 'start', 'mode': 'passthru',
+ 'host': '127.0.0.1', 'port': peer.port})
+ self.assertTrue(resp['ok'], resp)
+ self.assertEqual(resp['listener']['mode'], 'passthru')
+ self.assertEqual(server._BIP.target, ('127.0.0.1', peer.port))
+ cid, err = server._BIP.open('10.9.9.9', 10174, 512)
+ self.assertIsNone(err)
+ self.assertTrue(server._BIP.send(cid, b'CARDHELLO'))
+ data = b''
+ for _ in range(20):
+ data = server._BIP.receive(cid, 100)
+ if data:
+ break
+ time.sleep(0.05)
+ self.assertEqual(data, b'PLATFORM')
+ finally:
+ server._scp81_bip_control({'action': 'stop'})
+ peer.stop()
+
def test_disabled_terminal_refuses_open(self):
bip = httpota.BipTerminal()
cid, err = bip.open('127.0.0.1', 1, 512)
diff --git a/tests/test_ota_helpers.py b/tests/test_ota_helpers.py
index db880f1..b41e0aa 100644
--- a/tests/test_ota_helpers.py
+++ b/tests/test_ota_helpers.py
@@ -277,6 +277,17 @@ class TestProactiveDecode(unittest.TestCase):
srv._PROACTIVE_SESSION_START = 1234.0
srv._PLI_DATA[0x00] = '93055210011000'
+ @staticmethod
+ def _cmd_raw(cmd_type, qualifier, extras=b''):
+ """A D0-wrapped proactive command (header TLVs + extras)."""
+ body = (bytes([0x81, 0x03, 0x01, cmd_type, qualifier])
+ + bytes([0x82, 0x02, 0x83, 0x81]) + extras)
+ return bytes([0xD0, len(body)]) + body
+
+ @staticmethod
+ def _decoded(cmd_type, raw, qualifier=None):
+ return {d['label']: d['value'] for d in _decode_cmd(cmd_type, raw, qualifier)}
+
def test_decode_cmd_poll_interval(self):
r = _decode_cmd(0x03, bytes.fromhex('d00d8103010300820283818402011e'), None)
self.assertEqual(r, [{'label': 'Interval', 'value': '30 s'}])
@@ -286,13 +297,68 @@ class TestProactiveDecode(unittest.TestCase):
self.assertEqual(r, [{'label': 'Events', 'value': 'Call connected'}])
def test_decode_cmd_send_short_message(self):
- r = _decode_cmd(0x13, bytes.fromhex('d0158103011300820283818b0b916106152670f900a35f020101'), None)
- self.assertEqual(r, [{'label': 'SMS TPDU', 'value': '916106152670f900a35f02'}])
+ # SEND SHORT MESSAGE with an SMS-SUBMIT TPDU carrying GSM-7 text.
+ tpdu = bytes.fromhex('010006912143F5000005E8329BFD06')
+ raw = self._cmd_raw(0x13, 0, bytes([0x8B, len(tpdu)]) + tpdu)
+ r = self._decoded(0x13, raw)
+ self.assertEqual(r['Type'], 'SMS-SUBMIT')
+ self.assertEqual(r['TP-MR'], '0')
+ self.assertEqual(r['TP-DA'], '12345')
+ self.assertEqual(r['TP-PID'], '0x00')
+ self.assertEqual(r['TP-DCS'], '0x00')
+ self.assertEqual(r['TP-UDL'], '5')
+ self.assertEqual(r['Text'], 'hello')
+ self.assertEqual(r['SMS TPDU'], tpdu.hex().upper())
+
+ def test_decode_cmd_send_short_message_udh_8bit(self):
+ # UDHI + concatenation IE (16-bit ref) + 8-bit text data.
+ udh = bytes.fromhex('0608040001020341 42'.replace(' ', ''))
+ tpdu = (bytes.fromhex('4100' '06912143F5' '00' '04' '09') + udh)
+ raw = self._cmd_raw(0x13, 0, bytes([0x8B, len(tpdu)]) + tpdu)
+ r = self._decoded(0x13, raw)
+ self.assertEqual(r['Concat (16-bit ref)'], '1, part 2/3')
+ self.assertEqual(r['Text'], 'AB')
+
+ def test_decode_cmd_send_short_message_ucs2(self):
+ text = 'Тест'.encode('utf-16-be')
+ tpdu = (bytes.fromhex('0100' '06912143F5' '00' '08' '%02X' % len(text))
+ + text)
+ raw = self._cmd_raw(0x13, 0, bytes([0x8B, len(tpdu)]) + tpdu)
+ r = self._decoded(0x13, raw)
+ self.assertEqual(r['TP-DCS'], '0x08')
+ self.assertEqual(r['Text'], 'Тест')
+
+ def test_decode_cmd_send_short_message_secured_packet(self):
+ # PID 0x7F = SIM data download: the UD is a secured packet (TS 31.115).
+ tpdu = bytes.fromhex('0100' '06912143F5' '7F' 'F6' '03' 'AABBCC')
+ raw = self._cmd_raw(0x13, 0, bytes([0x8B, len(tpdu)]) + tpdu)
+ r = self._decoded(0x13, raw)
+ self.assertEqual(r['TP-PID'], '0x7F (SIM data download)')
+ self.assertEqual(r['Secured packet (TS 31.115)'], '3 bytes: AABBCC')
+
+ def test_decode_cmd_send_short_message_malformed_falls_back(self):
+ # A malformed/garbage TPDU must not raise: the raw hex line remains.
+ raw = bytes.fromhex('d0158103011300820283818b0b916106152670f900a35f020101')
+ r = self._decoded(0x13, raw)
+ self.assertEqual(r['SMS TPDU'], '916106152670F900A35F02')
def test_decode_cmd_pli_qualifier_name(self):
r = _decode_cmd(0x26, b'\xd0', 0x00)
self.assertTrue(r[0]['value'].startswith('Location Information (MCC, MNC, LAC/TAC, Cell ID)'))
+ def test_decode_cmd_pli_all_standard_qualifiers_named(self):
+ # TS 102 223 V18.3.0 (PLI qualifier coding): names must exist even
+ # without a special data decoder, e.g. ESN (07) and MEID (0B).
+ cases = {
+ 0x07: 'ESN',
+ 0x0B: 'MEID',
+ 0x1A: 'Supported Radio Access Technologies',
+ 0x05: 'Reserved for GSM',
+ }
+ for qualifier, name in cases.items():
+ r = _decode_cmd(0x26, b'\xd0', qualifier)
+ self.assertIn(name, r[0]['value'], 'qualifier 0x%02X' % qualifier)
+
def test_decode_cmd_timer_management_start(self):
# TS 102 223 6.6.21/8.37/8.38: start timer 3 for 14:07:32
raw = bytes.fromhex('d011810301270082028182a40103a503417023')
diff --git a/tests/test_scp81.py b/tests/test_scp81.py
index d9d96ce..046b838 100644
--- a/tests/test_scp81.py
+++ b/tests/test_scp81.py
@@ -682,6 +682,38 @@ class BipControlTest(unittest.TestCase):
self.assertFalse(resp['ok'])
self.assertIn('unsupported mode', resp['error'])
+ def test_passthru_mode_targets_the_external_server(self):
+ resp = server._scp81_bip_control({'action': 'start', 'mode': 'passthru',
+ 'host': '10.11.12.13', 'port': 10174})
+ self.assertTrue(resp['ok'], resp)
+ self.assertIsNone(server._SCP81_LISTENER) # no local listener
+ self.assertEqual(resp['listener']['mode'], 'passthru')
+ self.assertEqual(resp['listener']['host'], '10.11.12.13')
+ self.assertEqual(resp['listener']['port'], 10174)
+ self.assertEqual(resp['listener']['target'], '10.11.12.13:10174')
+ self.assertTrue(resp['bip']['enabled'])
+ self.assertEqual(server._BIP.target, ('10.11.12.13', 10174))
+ # the status endpoint sees the passthru mode while it runs ...
+ self.assertEqual(server._scp81_listener_status()['mode'], 'passthru')
+ # ... and stopping clears it (no stale listener in the status)
+ server._scp81_bip_control({'action': 'stop'})
+ self.assertIsNone(server._scp81_listener_status())
+ self.assertFalse(server._BIP.enabled)
+
+ def test_passthru_mode_requires_an_explicit_target(self):
+ # No defaults for a remote platform: the target must be configured.
+ resp = server._scp81_bip_control({'action': 'start', 'mode': 'passthru'})
+ self.assertFalse(resp['ok'])
+ self.assertIn('host and port', resp['error'])
+ resp = server._scp81_bip_control({'action': 'start', 'mode': 'passthru',
+ 'host': '10.0.0.1'})
+ self.assertFalse(resp['ok'])
+ resp = server._scp81_bip_control({'action': 'start', 'mode': 'passthru',
+ 'port': 1234})
+ self.assertFalse(resp['ok'])
+ self.assertIsNone(server._SCP81_LISTENER)
+ self.assertFalse(server._BIP.enabled)
+
def test_start_accepts_explicit_script_list(self):
server._SCP81_PSKS = {}
server._SCP81_PSK_LEGACY = None