feat: generic BIP terminal control (Simulator BIP pill) (v3.6.0)

BIP was only reachable through the SCP81 listener; cards that use TCP for
other purposes (an applet's OPEN CHANNEL, a push trigger without HTTP OTA)
now get a generic control surface, independent of SCP81.

Server:
- /api/bip/control starts a plain BIP session in three modes: sink
  (default; a local TcpDumpServer that accepts the card's channels and only
  logs conn/sink-rx/conn-close, never answering; port 0 = ephemeral, the
  bound port is reported), passthru (dial the OPEN CHANNEL destination, TCP
  client only) and redirect (fixed host:port).
- /api/bip/status returns {owner, bip, listener}; /api/bip/log and
  /api/bip/log-clear share the BIP event log with /api/scp81/log.
- The session state is shared with the SCP81 listener and only one session
  runs at a time: _bip_session_stop() stops whichever control started it
  and the control responses report it as `replaced` (both directions,
  SCP81 <-> BIP).  State renamed _SCP81_MODE/_TARGET/_LISTENER/_LINK_EVENTS
  -> _BIP_* plus _BIP_OWNER.
- TcpDumpServer logs conn-close and counts accepted connections; stop()
  joins the accept thread so the port is really free on restart.
- The new control endpoints are blocked during test-script runs.

PWA:
- New Simulator pill BIP (after TR Config): mode select with notes,
  Host/Port rules, Start/Stop with a "replaced" notice, status line (mode,
  bound address, owner, channels), a Channels box and the shared BIP log
  (reuses the SCP81 log renderer, now showing `peer`).
- The SCP81 status line marks a session owned by the BIP pill.

Help EN/RU 8.10, docs/api.md, AGENTS; CAT_TP/UDP recorded as not
implemented.

Tests: tests/test_bip.py (9) and frontend/tests/bip.test.js (6).
600 frontend / 482 Python green; version 3.6.0; sw simple-v273.
This commit is contained in:
2026-09-27 15:28:23 +03:00
parent 73ef3a67ca
commit 18a36a8f27
14 changed files with 924 additions and 97 deletions
+49 -1
View File
@@ -65,9 +65,13 @@ a 3.x PWA).
| `/api/pli-dict` | GET | Current dictionary (hex values per qualifier) |
| `/api/pli-dict` | POST | Update dictionary entries |
| `/api/scp81/bip` | POST | Start/stop the HTTP OTA listener (dump capture or PSK TLS server) |
| `/api/scp81/status` | GET | BIP terminal + listener state (channels, PSK identities, handshake identity) |
| `/api/scp81/status` | GET | BIP terminal + listener state (channels, PSK identities, handshake identity, `owner`) |
| `/api/scp81/log` | GET | HTTP OTA event log (`?after=<seq>`) |
| `/api/scp81/log-clear` | POST | Clear the HTTP OTA event log |
| `/api/bip/control` | POST | Start/stop the generic BIP session (`sink` / `passthru` / `redirect`) |
| `/api/bip/status` | GET | BIP session state + owner (same shape as `/api/scp81/status`) |
| `/api/bip/log` | GET | BIP event log (`?after=<seq>`) |
| `/api/bip/log-clear` | POST | Clear the BIP event log |
| `/api/scp81/queue` | POST | Replace the SCP81 command script (optionally force-restart) |
| `/api/scp81/script` | GET | Active command script + execution state and R-APDUs |
| `/api/scp81/psk-map` | POST | Replace the PSK table of a running TLS listener |
@@ -985,6 +989,47 @@ its run progress.
Stop either mode with `{"action": "stop"}` (also disables the BIP terminal).
### `POST /api/bip/control`
Generic BIP terminal control for the Simulator's **BIP** pill - the terminal
side of the Bearer Independent Protocol for cards that use TCP without HTTP
OTA. The session is shared with the SCP81 listener: only one BIP session runs
at a time, starting either control stops the other, and the response reports
what was stopped as `replaced: {"owner": "scp81"|"bip", "mode": ...}` (the
PWA shows a notice).
```json
{"action": "start", "mode": "sink", "host": "127.0.0.1", "port": 0}
```
Modes:
- `sink` (default) - starts a local TCP listener that accepts the card's BIP
channels and only logs what arrives (`conn`, `sink-rx`, `conn-close`); it
never sends anything back. `port` may be `0`/omitted for an ephemeral port;
the bound address is reported in the status. Use it for cards that open a
TCP connection just to upload data.
- `passthru` - no listener and no target: each channel dials the destination
the card requests in OPEN CHANNEL (TCP client only, no default port; an
incomplete or non-TCP request fails with result `3A`).
- `redirect` - every channel connects to the required `host`/`port`; the
address the card requests is only logged.
`link_events` (default `true`) controls the terminal-side Channel status
events (TS 102 223 7.5.11). An invalid request never disturbs a running
session. Stop with `{"action": "stop"}`; the response carries the same body
as the status endpoints.
### `GET /api/bip/status`
Same shape as `GET /api/scp81/status`: `{"owner": "bip"|"scp81"|null,
"bip": {...}, "listener": {...}}`. A running sink reports
`{"mode": "sink", "host": ..., "port": ..., "connections": N}`.
### `GET /api/bip/log` / `POST /api/bip/log-clear`
The shared BIP event log (`?after=<seq>`) - identical to `/api/scp81/log`.
### `GET /api/scp81/status`
```json
@@ -995,6 +1040,9 @@ Stop either mode with `{"action": "stop"}` (also disables the BIP terminal).
"version_seen": "TLSv1.2", "cipher_seen": "PSK-AES128-CBC-SHA256"}}
```
`owner` is `scp81` or `bip` (whichever control started the session), `null`
when idle; it is returned by both this endpoint and `/api/bip/status`.
Listener modes: `tls` (local PSK TLS server), `dump` (capture-only TCP
listener), `redirect` (no local listener; the BIP channels go straight to the
configured `host:port`, e.g. an external HTTP OTA platform — reported as
+9
View File
@@ -553,6 +553,15 @@
<li><strong>Шаблоны</strong> &mdash; Empty, <em>Обзор меню STK</em>, <em>Апплет через SCP80</em>; редактор шагов — формный, отдельные скрипты экспортируются/импортируются как JSON.</li>
</ul>
<h3 id="bip" class="text-lg font-medium mb-2">8.10 BIP-терминал (обычный TCP)</h3>
<p class="text-sm mb-2">Пилюля <strong>BIP</strong> запускает терминальную сторону Bearer Independent Protocol для карт, использующих TCP без HTTP OTA (SCP81): апплет или пуш-триггер может открыть канал и обменяться сырыми данными. Одновременно работает только одна BIP-сессия &mdash; запуск здесь останавливает слушатель SCP81, а запуск слушателя SCP81 останавливает эту сессию (в обоих случаях сообщается, что было заменено).</p>
<ul class="list-disc list-inside text-sm space-y-1 mb-3">
<li><strong>Локальный приёмник (только журнал)</strong> (по умолчанию) &mdash; принимает каналы карты на локальном слушателе и только записывает принятое в журнал; ничего не отправляется обратно. Оставьте порт пустым для эфемерного порта (занятый адрес появится в статусе). Подходит для карт, которые открывают TCP-соединение только чтобы выгрузить данные.</li>
<li><strong>Адрес карты (OPEN CHANNEL)</strong> &mdash; каждый канал подключается к адресу, который карта запросила в OPEN CHANNEL (Other address + transport port, только TCP-клиент); Host и Port не используются, используется сеть сервера (только для лаборатории).</li>
<li><strong>Заданный адрес</strong> &mdash; каждый канал подключается к указанному Host:Port; запрошенный картой адрес только записывается в журнал.</li>
</ul>
<p class="text-sm mb-2">Строка статуса показывает активный режим, занятый адрес, владельца сессии и открытые каналы; блок <strong>Каналы</strong> перечисляет счётчики байт по каналам, а <strong>Журнал BIP</strong> записывает каждое событие open / send / receive / close с hex-превью (те же события, что показывает вкладка SCP81). CAT_TP через UDP не реализован.</p>
<h2 id="server" class="text-xl font-semibold mb-3 border-b border-gray-300 dark:border-slate-700 pb-1">9. Установка сервера</h2>
<p class="mb-3">Для работы с картой (вкладка &laquo;Картридер&raquo;, &laquo;Симулятор&raquo;, доставка OTA) нужен локальный <a href="https://github.com/anttro/simple" class="text-blue-600 dark:text-blue-400 hover:underline">pysim-simple-server</a> — встроенный в SIMple HTTP-сервер, оборачивающий pySim, работающий с ридером через PC/SC или serial и раздающий сам PWA (откройте <code class="font-mono text-sm">http://127.0.0.1:8080</code>).</p>
+9
View File
@@ -552,6 +552,15 @@
<li><strong>Templates</strong> &mdash; Empty, <em>STK menu browsing</em>, <em>Applet via SCP80</em>; the step editor is form-based and individual scripts export/import as JSON.</li>
</ul>
<h3 id="bip" class="text-lg font-medium mb-2">8.10 BIP terminal (generic TCP)</h3>
<p class="text-sm mb-2">The <strong>BIP</strong> pill runs the terminal side of the Bearer Independent Protocol for cards that use TCP without HTTP OTA (SCP81): an applet or a push trigger can open a channel and exchange raw data. Only one BIP session runs at a time &mdash; starting here stops a running SCP81 listener, and starting the SCP81 listener stops this session (both report what was replaced).</p>
<ul class="list-disc list-inside text-sm space-y-1 mb-3">
<li><strong>Local sink (log only)</strong> (default) &mdash; accepts the card's channels on a local listener and only logs what arrives; nothing is ever sent back. Leave the port empty for an ephemeral port (the bound address appears in the status). Use it for cards that open a TCP connection just to upload data.</li>
<li><strong>Card's destination (OPEN CHANNEL)</strong> &mdash; each channel is connected to the destination the card requests in OPEN CHANNEL (Other address + transport port, TCP client only); Host and Port are not used and the server's network is used (lab only).</li>
<li><strong>Fixed target</strong> &mdash; every channel is connected to the configured Host:Port; the address the card requests is only logged.</li>
</ul>
<p class="text-sm mb-2">The status line shows the active mode, the bound address, the session owner and the open channels; the <strong>Channels</strong> box lists the per-channel byte counts and the <strong>BIP log</strong> records every open / send / receive / close event with a hex preview (the same events the SCP81 tab shows). CAT_TP over UDP is not implemented.</p>
<h2 id="server" class="text-xl font-semibold mb-3 border-b border-gray-300 dark:border-slate-700 pb-1">9. Server installation</h2>
<p class="mb-3">Live card operations (Card reader tab, Simulator, OTA delivery) require the local <a href="https://github.com/anttro/simple" class="text-blue-600 dark:text-blue-400 hover:underline">pysim-simple-server</a> — a small HTTP server bundled with SIMple that wraps pySim, talks to the reader over PC/SC or serial, and also serves the PWA itself (open <code class="font-mono text-sm">http://127.0.0.1:8080</code>).</p>
+251 -5
View File
@@ -1032,6 +1032,7 @@
<div class="flex flex-wrap justify-center gap-2 mb-3">
<button class="phone-subtab px-4 py-1.5 text-sm rounded-full bg-blue-600 text-white" data-phone-sub="phone" onclick="phoneSwitchSubtab('phone')" data-l10n="Phone">Phone</button>
<button class="phone-subtab px-4 py-1.5 text-sm rounded-full bg-gray-200 dark:bg-slate-700 text-gray-700 dark:text-slate-300 hover:bg-gray-300 dark:hover:bg-slate-600" data-phone-sub="tr" onclick="phoneSwitchSubtab('tr')" data-l10n="TR Config">TR Config</button>
<button class="phone-subtab px-4 py-1.5 text-sm rounded-full bg-gray-200 dark:bg-slate-700 text-gray-700 dark:text-slate-300 hover:bg-gray-300 dark:hover:bg-slate-600" data-phone-sub="bip" onclick="phoneSwitchSubtab('bip')" data-l10n="BIP">BIP</button>
<button class="phone-subtab px-4 py-1.5 text-sm rounded-full bg-gray-200 dark:bg-slate-700 text-gray-700 dark:text-slate-300 hover:bg-gray-300 dark:hover:bg-slate-600" data-phone-sub="esim" onclick="phoneSwitchSubtab('esim')" data-l10n="eSIM LPA">eSIM LPA</button>
<button class="phone-subtab px-4 py-1.5 text-sm rounded-full bg-gray-200 dark:bg-slate-700 text-gray-700 dark:text-slate-300 hover:bg-gray-300 dark:hover:bg-slate-600" data-phone-sub="test" onclick="phoneSwitchSubtab('test')" data-l10n="Test script">Test script</button>
</div>
@@ -1183,6 +1184,47 @@
<span class="text-gray-400" data-l10n="Loading...">Loading...</span>
</div>
</div>
<div id="phone-sub-bip" class="hidden">
<fieldset class="border border-gray-200 dark:border-slate-700 rounded p-3 mb-3">
<legend class="px-1 text-xs font-medium text-gray-500 dark:text-slate-400"><span data-l10n="BIP terminal">BIP terminal</span> <span id="bip-state" class="text-xs text-gray-400"></span></legend>
<div class="flex flex-wrap items-end gap-3">
<div>
<label class="block mb-1 text-xs font-medium text-gray-600 dark:text-slate-400" data-l10n="Mode">Mode</label>
<select id="bip-mode" onchange="bipModeChanged()" class="border border-gray-300 dark:border-slate-600 text-sm rounded px-2 py-1.5 dark:bg-slate-800">
<option value="sink" data-l10n="Local sink (log only)" selected>Local sink (log only)</option>
<option value="passthru" data-l10n="Card's destination (OPEN CHANNEL)">Card's destination (OPEN CHANNEL)</option>
<option value="redirect" data-l10n="Fixed target">Fixed target</option>
</select>
</div>
<div>
<label class="block mb-1 text-xs font-medium text-gray-600 dark:text-slate-400" data-l10n="Host">Host</label>
<input id="bip-host" value="127.0.0.1" class="w-36 font-mono border border-gray-300 dark:border-slate-600 text-sm rounded px-2 py-1.5 dark:bg-slate-800">
</div>
<div>
<label class="block mb-1 text-xs font-medium text-gray-600 dark:text-slate-400" data-l10n="Port">Port</label>
<input id="bip-port" placeholder="ephemeral" class="w-24 font-mono border border-gray-300 dark:border-slate-600 text-sm rounded px-2 py-1.5 dark:bg-slate-800">
</div>
<button id="bip-start-btn" data-needs="server" onclick="bipStart()" class="px-3 py-1.5 text-sm rounded bg-blue-600 text-white hover:bg-blue-700 disabled:opacity-40 disabled:cursor-not-allowed" data-l10n="Start">Start</button>
<button id="bip-stop-btn" data-needs="server" onclick="bipStop()" class="px-3 py-1.5 text-sm rounded bg-gray-600 text-white hover:bg-gray-700 disabled:opacity-40 disabled:cursor-not-allowed" data-l10n="Stop">Stop</button>
</div>
<div id="bip-msg" class="text-xs mt-2 hidden"></div>
<div id="bip-sink-note" class="mt-2 text-xs text-gray-500 dark:text-slate-400" data-l10n="Local sink: accepts the card's BIP channels on this Host and Port, logs everything received and never sends anything back. Leave the port empty for an ephemeral port (the bound one appears in the status). Use it for cards that open a TCP connection just to upload data.">Local sink: accepts the card's BIP channels on this Host and Port, logs everything received and never sends anything back. Leave the port empty for an ephemeral port (the bound one appears in the status). Use it for cards that open a TCP connection just to upload data.</div>
<div id="bip-passthru-note" class="hidden mt-2 text-xs text-gray-500 dark:text-slate-400" data-l10n="Card's destination: each BIP channel is connected to the destination the card requests in OPEN CHANNEL (Other address + transport port, TCP client only); Host and Port are not used and the server's network is used (lab only).">Card's destination: each BIP channel is connected to the destination the card requests in OPEN CHANNEL (Other address + transport port, TCP client only); Host and Port are not used and the server's network is used (lab only).</div>
<div id="bip-redirect-note" class="hidden mt-2 text-xs text-gray-500 dark:text-slate-400" data-l10n="Fixed target: every BIP channel is connected to this Host:Port; the address the card requests is only logged.">Fixed target: every BIP channel is connected to this Host:Port; the address the card requests is only logged.</div>
<div class="mt-2 text-xs text-gray-500 dark:text-slate-400" data-l10n="One BIP session at a time: starting here replaces a running SCP81 listener, and starting the SCP81 listener replaces this session.">One BIP session at a time: starting here replaces a running SCP81 listener, and starting the SCP81 listener replaces this session.</div>
</fieldset>
<fieldset class="border border-gray-200 dark:border-slate-700 rounded p-3 mb-3">
<legend class="px-1 text-xs font-medium text-gray-500 dark:text-slate-400" data-l10n="Channels">Channels</legend>
<div id="bip-channels" class="text-xs font-mono text-gray-600 dark:text-slate-400"></div>
</fieldset>
<fieldset class="border border-gray-200 dark:border-slate-700 rounded p-3">
<legend class="px-1 text-xs font-medium text-gray-500 dark:text-slate-400" data-l10n="BIP log">BIP log</legend>
<div class="flex justify-end mb-1">
<button data-needs="server" onclick="bipLogClear()" class="px-2.5 py-1 text-xs rounded bg-gray-200 dark:bg-slate-700 text-gray-700 dark:text-slate-300 hover:bg-gray-300 dark:hover:bg-slate-600 disabled:opacity-40 disabled:cursor-not-allowed" data-l10n="Clear">Clear</button>
</div>
<div id="bip-log" class="text-sm font-mono text-gray-600 dark:text-slate-400 max-h-[55vh] overflow-auto"></div>
</fieldset>
</div>
<div id="phone-sub-esim" class="hidden">
<div id="esim-not-euicc" class="text-sm text-gray-500 dark:text-slate-400 hidden" data-l10n="The equipped card is not an eUICC">The equipped card is not an eUICC</div>
<div id="esim-body" class="space-y-4">
@@ -1622,12 +1664,13 @@
// ===== Version =====
// Single source of truth for the PWA version: shown in the header and used
// by the server version check in pysimConnect().
const SIMPLE_VERSION = '3.5.19';
const SIMPLE_VERSION = '3.6.0';
document.getElementById('app-version').textContent = 'v' + SIMPLE_VERSION;
// ===== Tab switching =====
function switchTab(name) {
if (_scp81Timer && name !== 'scp81') { clearInterval(_scp81Timer); _scp81Timer = null; }
if (_bipTimer && name !== 'phone') { clearInterval(_bipTimer); _bipTimer = null; }
document.querySelectorAll('.tab-content').forEach(el => el.classList.add('hidden'));
document.getElementById('tab-' + name).classList.remove('hidden');
document.querySelectorAll('.tab-btn').forEach(btn => {
@@ -1719,11 +1762,13 @@ function phoneSwitchSubtab(name) {
btn.classList.toggle('text-gray-700', !active);
btn.classList.toggle('dark:text-slate-300', !active);
});
['phone', 'tr', 'esim', 'test'].forEach(s => {
['phone', 'tr', 'bip', 'esim', 'test'].forEach(s => {
document.getElementById('phone-sub-' + s).classList.toggle('hidden', s !== name);
});
if (name === 'tr') {
pysimPliRender();
} else if (name === 'bip') {
bipEnter();
} else if (name === 'esim') {
esimFetchAll();
} else if (name === 'test') {
@@ -1736,7 +1781,7 @@ function phoneSwitchSubtab(name) {
pysimPollStatusInit();
tpRefresh();
}
setHelpAnchor({ phone: 'stk-menu', tr: 'pli-dict', esim: 'esim', test: 'test-script' }[name] || 'stk-menu');
setHelpAnchor({ phone: 'stk-menu', tr: 'pli-dict', bip: 'bip', esim: 'esim', test: 'test-script' }[name] || 'stk-menu');
}
// ===== eSIM / LPA (local ES10a/b/c operations) =====
@@ -11454,6 +11499,7 @@ function scp81LogLine(e) {
if (e.free_volatile !== undefined) parts.push('free vol=' + e.free_volatile);
if (e.channel) parts.push('ch' + e.channel);
if (e.requested) parts.push(e.requested);
if (e.peer) parts.push(e.peer);
if (e.target) parts.push('-> ' + e.target);
if (e.bytes) parts.push(e.bytes + 'B');
if (e.identity) parts.push('id=' + e.identity);
@@ -11752,6 +11798,7 @@ async function scp81StatusRefresh() {
return st;
}
let s = l.mode + (l.host && l.port ? ' ' + l.host + ':' + l.port : '');
if (st.owner === 'bip') s = t('BIP session:') + ' ' + s;
if (l.mode === 'tls') {
const ids = l.psk_identities || [];
s += ' | PSK: ' + ids.length;
@@ -11841,8 +11888,11 @@ async function scp81Start() {
}
try {
const resp = await pysimFetch('/api/scp81/bip', body);
if (resp.ok) scp81Msg(t('Listening.'), 'text-emerald-600 dark:text-emerald-400');
else scp81Msg(resp.error || t('Error'), 'text-red-500');
if (resp.ok) {
const notice = bipReplacedNotice(resp.replaced);
scp81Msg(t('Listening.') + (notice ? ' ' + notice : ''),
'text-emerald-600 dark:text-emerald-400');
} else scp81Msg(resp.error || t('Error'), 'text-red-500');
scp81StatusRefresh();
scp81LogRefresh();
scp81ResultsRefresh();
@@ -12133,6 +12183,175 @@ function scp81Enter() {
}, 2000);
}
// ===== Generic BIP terminal (Simulator -> BIP) =====
let _bipTimer = null;
function bipModeChanged() {
const mode = document.getElementById('bip-mode').value;
const notes = { sink: 'bip-sink-note', passthru: 'bip-passthru-note',
redirect: 'bip-redirect-note' };
for (const key in notes) {
const el = document.getElementById(notes[key]);
if (el) el.classList.toggle('hidden', key !== mode);
}
// Host/Port apply to the fixed target and to the sink bind address;
// passthru dials the destination from the card's OPEN CHANNEL.
for (const id of ['bip-host', 'bip-port']) {
const el = document.getElementById(id);
if (!el) continue;
el.disabled = (mode === 'passthru');
el.classList.toggle('opacity-40', mode === 'passthru');
}
const port = document.getElementById('bip-port');
if (port) port.placeholder = (mode === 'sink') ? t('ephemeral') : '8443';
}
// Build the /api/bip/control start body; {error} for invalid input.
function bipStartBody(mode, host, port) {
const body = { action: 'start', mode: mode };
if (mode === 'passthru') return { body: body };
const h = String(host || '').trim();
const p = String(port || '').trim();
if (mode === 'redirect') {
const n = parseInt(p, 10);
if (!h) return { error: 'Fixed target requires the host' };
if (!Number.isInteger(n) || n < 1 || n > 65535) return { error: 'Port must be 1-65535' };
body.host = h;
body.port = n;
} else {
if (h) body.host = h;
if (p) {
const n = parseInt(p, 10);
if (!Number.isInteger(n) || n < 0 || n > 65535) return { error: 'Port must be 0-65535' };
body.port = n;
}
}
return { body: body };
}
function bipReplacedNotice(replaced) {
if (!replaced || !replaced.owner) return '';
return (replaced.owner === 'scp81' ? t('Stopped the SCP81 listener')
: t('Stopped the BIP session')) + ' (' + replaced.mode + ')';
}
function bipChannelLine(c) {
const parts = ['ch' + c.id, (c.requested || '?') + ' -> ' + (c.target || '?')];
parts.push('in:' + (c.bytes_in || 0) + ' out:' + (c.bytes_out || 0) +
' buffer:' + (c.pending || 0));
parts.push(c.peer_closed ? t('peer closed') : t('open'));
return parts.join(' ');
}
function bipStatusLine(st) {
const bip = (st && st.bip) || {};
const l = (st && st.listener) || null;
const parts = [];
if (l) {
let line = l.mode + (l.host !== undefined && l.port !== undefined
? ' ' + l.host + ':' + l.port : '');
if (l.connections !== undefined) line += ' | ' + l.connections + ' ' + t('connections');
parts.push(line);
}
if (st && st.owner) parts.push(t('owner') + ': ' + st.owner);
const ch = (bip.channels || []).map(bipChannelLine).join(' | ');
if (ch) parts.push(ch);
return parts.join(' \u00b7 ');
}
function bipMsg(text, cls) {
const el = document.getElementById('bip-msg');
el.textContent = text;
el.classList.remove('hidden');
el.className = 'text-xs mt-2 ' + (cls || 'text-gray-500 dark:text-slate-400');
}
async function bipStatusRefresh() {
const el = document.getElementById('bip-state');
const chEl = document.getElementById('bip-channels');
try {
const st = await pysimFetch('/api/bip/status');
el.textContent = bipStatusLine(st);
const chans = (st.bip && st.bip.channels) || [];
chEl.innerHTML = chans.length
? chans.map(c => '<div class="py-0.5 border-b border-gray-100 dark:border-slate-700/50 break-all">' + esc(bipChannelLine(c)) + '</div>').join('')
: '<span class="text-gray-400">' + t('No channels.') + '</span>';
return st;
} catch (err) {
el.textContent = String(err.message || err);
return null;
}
}
async function bipLogRefresh() {
const el = document.getElementById('bip-log');
try {
const log = await pysimFetch('/api/bip/log');
const entries = (log.entries || []).slice(-200);
if (!entries.length) {
el.innerHTML = '<span class="text-gray-400">' + t('No entries yet.') + '</span>';
return;
}
el.innerHTML = entries.map(e =>
'<div class="py-0.5 border-b border-gray-100 dark:border-slate-700/50 break-all">' + scp81LogEntryHtml(e) + '</div>'
).join('');
} catch (err) {
el.innerHTML = '<span class="text-red-500">Error: ' + esc(String(err.message || err)) + '</span>';
}
}
async function bipLogClear() {
try {
await pysimFetch('/api/bip/log-clear', {});
} catch (err) { /* keep the log as it is */ }
bipLogRefresh();
}
async function bipStart() {
const mode = document.getElementById('bip-mode').value;
const built = bipStartBody(mode,
document.getElementById('bip-host').value,
document.getElementById('bip-port').value);
if (built.error) { bipMsg(t(built.error), 'text-red-500'); return; }
try {
const resp = await pysimFetch('/api/bip/control', built.body);
if (!resp.ok) { bipMsg(t('Start failed:') + ' ' + (resp.error || ''), 'text-red-500'); return; }
const notice = bipReplacedNotice(resp.replaced);
bipMsg(t('BIP session started') + (notice ? ' \u00b7 ' + notice : ''),
'text-emerald-600 dark:text-emerald-400');
bipStatusRefresh();
bipLogRefresh();
} catch (err) {
bipMsg(String(err.message || err), 'text-red-500');
}
}
async function bipStop() {
try {
const resp = await pysimFetch('/api/bip/control', { action: 'stop' });
const notice = bipReplacedNotice(resp.replaced);
bipMsg(t('BIP session stopped') + (notice ? ' \u00b7 ' + notice : ''),
'text-gray-500 dark:text-slate-400');
bipStatusRefresh();
bipLogRefresh();
} catch (err) {
bipMsg(String(err.message || err), 'text-red-500');
}
}
function bipEnter() {
bipModeChanged();
bipStatusRefresh();
bipLogRefresh();
setHelpAnchor('bip');
if (_bipTimer) clearInterval(_bipTimer);
_bipTimer = setInterval(() => {
if (document.getElementById('phone-sub-bip').classList.contains('hidden')) return;
bipStatusRefresh();
bipLogRefresh();
}, 2000);
}
// ===== PLI data dictionary =====
function decPlmn(hex3) {
const h = hex3.replace(/\s/g, '').slice(0, 6);
@@ -16290,6 +16509,32 @@ const LANG_RU = {
'The key is only sent to the local server and is never stored or logged.': 'Ключ передаётся только локальному серверу, не сохраняется и не записывается в журнал.',
'Listening.': 'Слушает.',
'Stopped.': 'Остановлено.',
'BIP': 'BIP',
'BIP terminal': 'BIP-терминал',
'Local sink (log only)': 'Локальный приёмник (только журнал)',
'Card\'s destination (OPEN CHANNEL)': 'Адрес карты (OPEN CHANNEL)',
'Fixed target': 'Заданный адрес',
'Channels': 'Каналы',
'BIP log': 'Журнал BIP',
'Local sink: accepts the card\'s BIP channels on this Host and Port, logs everything received and never sends anything back. Leave the port empty for an ephemeral port (the bound one appears in the status). Use it for cards that open a TCP connection just to upload data.': 'Локальный приёмник: принимает BIP-каналы карты на этом Host и Port, записывает всё принятое в журнал и ничего не отправляет обратно. Оставьте порт пустым для эфемерного порта (занятый порт появится в статусе). Подходит для карт, которые открывают TCP-соединение только чтобы выгрузить данные.',
'Card\'s destination: each BIP channel is connected to the destination the card requests in OPEN CHANNEL (Other address + transport port, TCP client only); Host and Port are not used and the server\'s network is used (lab only).': 'Адрес карты: каждый BIP-канал подключается к адресу, который карта запросила в OPEN CHANNEL (Other address + transport port, только TCP-клиент); Host и Port не используются, используется сеть сервера (только для лаборатории).',
'Fixed target: every BIP channel is connected to this Host:Port; the address the card requests is only logged.': 'Заданный адрес: каждый BIP-канал подключается к этому Host:Port; запрошенный картой адрес только записывается в журнал.',
'One BIP session at a time: starting here replaces a running SCP81 listener, and starting the SCP81 listener replaces this session.': 'Одновременно работает только одна BIP-сессия: запуск здесь останавливает слушатель SCP81, а запуск слушателя SCP81 останавливает эту сессию.',
'ephemeral': 'эфемерный',
'Fixed target requires the host': 'Для заданного адреса нужен хост',
'Port must be 1-65535': 'Порт должен быть в диапазоне 1-65535',
'Port must be 0-65535': 'Порт должен быть в диапазоне 0-65535',
'peer closed': 'пир закрыт',
'open': 'открыт',
'connections': 'соединений',
'owner': 'владелец',
'No channels.': 'Каналов нет.',
'Start failed:': 'Запуск не удался:',
'BIP session started': 'BIP-сессия запущена',
'BIP session stopped': 'BIP-сессия остановлена',
'Stopped the SCP81 listener': 'Остановлен слушатель SCP81',
'Stopped the BIP session': 'Остановлена BIP-сессия',
'BIP session:': 'BIP-сессия:',
'not running': 'не запущен',
'configured': 'настроен',
'required': 'обязательно',
@@ -16941,6 +17186,7 @@ function refreshDynamicI18n() {
capRenderAnalysis('ram');
capRenderAnalysis('scripts');
if (isViewVisible('tab-phone')) {
if (isViewVisible('phone-sub-bip')) { bipStatusRefresh(); bipLogRefresh(); }
if (isViewVisible('phone-sub-phone')) {
pysimEventsRender();
pysimProactiveLogRender();
+1 -1
View File
@@ -1,4 +1,4 @@
const CACHE = 'simple-v272';
const CACHE = 'simple-v273';
const URLS = [
'index.html',
'help.html',
+145
View File
@@ -0,0 +1,145 @@
const { test } = require('node:test');
const assert = require('node:assert');
const fs = require('node:fs');
const path = require('node:path');
const html = fs.readFileSync(path.join(__dirname, '..', 'index.html'), 'utf8');
function extractFunc(src, name) {
const re = new RegExp('function\\s+' + name + '\\s*\\([^)]*\\)\\s*\\{');
const m = re.exec(src);
if (!m) throw new Error('function ' + name + ' not found');
let i = m.index + m[0].length - 1;
let depth = 0;
for (; i < src.length; i++) {
if (src[i] === '{') depth++;
else if (src[i] === '}') {
depth--;
if (depth === 0) break;
}
}
return src.slice(m.index, i + 1);
}
globalThis.esc = s => s;
globalThis.t = s => s;
for (const fn of ['bipReplacedNotice', 'bipChannelLine', 'bipStatusLine', 'bipStartBody', 'bipModeChanged']) {
eval(extractFunc(html, fn));
}
function fakeDom(mode) {
const els = {};
for (const id of ['bip-mode', 'bip-host', 'bip-port', 'bip-sink-note',
'bip-passthru-note', 'bip-redirect-note']) {
const el = {
disabled: false, placeholder: '', toggled: {},
classList: { toggle(c, on) { el.toggled[c] = on; } },
};
els[id] = el;
}
els['bip-mode'].value = mode;
globalThis.document = { getElementById: id => els[id] || null };
return els;
}
test('bipStartBody builds the control body per mode', () => {
// passthru: no host/port at all - the card chooses the destination
assert.deepStrictEqual(bipStartBody('passthru', '127.0.0.1', '9'),
{ body: { action: 'start', mode: 'passthru' } });
// sink: host optional, empty port = server-side ephemeral
assert.deepStrictEqual(bipStartBody('sink', '127.0.0.1', ''),
{ body: { action: 'start', mode: 'sink', host: '127.0.0.1' } });
assert.deepStrictEqual(bipStartBody('sink', '', '9101'),
{ body: { action: 'start', mode: 'sink', port: 9101 } });
// fixed target: host and a real port required
assert.deepStrictEqual(bipStartBody('redirect', '10.0.0.5', '8080'),
{ body: { action: 'start', mode: 'redirect', host: '10.0.0.5', port: 8080 } });
assert.strictEqual(bipStartBody('redirect', '', '8080').error, 'Fixed target requires the host');
assert.strictEqual(bipStartBody('redirect', '10.0.0.5', '').error, 'Port must be 1-65535');
assert.strictEqual(bipStartBody('redirect', '10.0.0.5', '0').error, 'Port must be 1-65535');
assert.strictEqual(bipStartBody('redirect', '10.0.0.5', 'x').error, 'Port must be 1-65535');
assert.strictEqual(bipStartBody('sink', '127.0.0.1', 'x').error, 'Port must be 0-65535');
assert.strictEqual(bipStartBody('sink', '127.0.0.1', '70000').error, 'Port must be 0-65535');
});
test('bipReplacedNotice names what the start replaced', () => {
assert.strictEqual(bipReplacedNotice(null), '');
assert.strictEqual(bipReplacedNotice({}), '');
assert.strictEqual(bipReplacedNotice({ owner: 'scp81', mode: 'tls' }),
'Stopped the SCP81 listener (tls)');
assert.strictEqual(bipReplacedNotice({ owner: 'bip', mode: 'sink' }),
'Stopped the BIP session (sink)');
});
test('bipChannelLine summarizes one channel', () => {
assert.strictEqual(bipChannelLine({ id: 1, requested: '10.0.0.5:80',
target: '10.0.0.5:80', bytes_in: 412, bytes_out: 0, pending: 3,
peer_closed: false }),
'ch1 10.0.0.5:80 -> 10.0.0.5:80 in:412 out:0 buffer:3 open');
assert.strictEqual(bipChannelLine({ id: 2, peer_closed: true }),
'ch2 ? -> ? in:0 out:0 buffer:0 peer closed');
});
test('bipStatusLine shows the mode, bound address, connections, owner and channels', () => {
assert.strictEqual(bipStatusLine({ owner: 'bip', bip: { channels: [] },
listener: { mode: 'sink', host: '127.0.0.1', port: 53121, connections: 1 } }),
'sink 127.0.0.1:53121 | 1 connections \u00b7 owner: bip');
assert.strictEqual(bipStatusLine({ owner: 'bip', bip: { channels: [] },
listener: { mode: 'passthru' } }), 'passthru \u00b7 owner: bip');
assert.strictEqual(bipStatusLine(null), '');
const withChannel = bipStatusLine({
owner: 'bip',
bip: { channels: [
{ id: 1, requested: 'a:1', target: 'b:2', bytes_in: 5, bytes_out: 7, pending: 0 }] },
listener: { mode: 'sink', host: 'h', port: 1 },
});
assert.ok(withChannel.includes('ch1 a:1 -> b:2 in:5 out:7 buffer:0 open'), withChannel);
});
test('bipModeChanged toggles the fields and the mode notes', () => {
let els = fakeDom('sink');
bipModeChanged();
assert.strictEqual(els['bip-host'].disabled, false);
assert.strictEqual(els['bip-port'].disabled, false);
assert.strictEqual(els['bip-port'].placeholder, 'ephemeral');
assert.strictEqual(els['bip-sink-note'].toggled.hidden, false);
assert.strictEqual(els['bip-passthru-note'].toggled.hidden, true);
assert.strictEqual(els['bip-redirect-note'].toggled.hidden, true);
els = fakeDom('passthru');
bipModeChanged();
assert.strictEqual(els['bip-host'].disabled, true);
assert.strictEqual(els['bip-port'].disabled, true);
assert.strictEqual(els['bip-host'].toggled['opacity-40'], true);
assert.strictEqual(els['bip-passthru-note'].toggled.hidden, false);
assert.strictEqual(els['bip-sink-note'].toggled.hidden, true);
els = fakeDom('redirect');
bipModeChanged();
assert.strictEqual(els['bip-host'].disabled, false);
assert.strictEqual(els['bip-port'].placeholder, '8443');
assert.strictEqual(els['bip-redirect-note'].toggled.hidden, false);
});
test('the BIP pill, view and API endpoints are wired', () => {
assert.match(html, /data-phone-sub="bip"[^>]*data-l10n="BIP"/);
assert.match(html, /id="phone-sub-bip"/);
assert.match(html, /\['phone', 'tr', 'bip', 'esim', 'test'\]/);
assert.match(html, /else if \(name === 'bip'\) \{\s*bipEnter\(\);/);
assert.match(html, /bip: 'bip'/);
// the poll stops when the Simulator tab is left
assert.match(html, /if \(_bipTimer && name !== 'phone'\)/);
for (const id of ['bip-mode', 'bip-host', 'bip-port', 'bip-start-btn',
'bip-stop-btn', 'bip-state', 'bip-channels', 'bip-log']) {
assert.ok(html.includes('id="' + id + '"'), id);
}
assert.match(html, /\/api\/bip\/control/);
assert.match(html, /\/api\/bip\/status/);
assert.match(html, /\/api\/bip\/log-clear/);
assert.match(html, /\/api\/bip\/log'/);
// the shared one-session rule and the owner marker are both shown
assert.match(html, /One BIP session at a time: starting here replaces a running SCP81 listener/);
assert.match(html, /if \(st\.owner === 'bip'\) s = t\('BIP session:'\)/);
assert.match(html, /if \(e\.peer\) parts\.push\(e\.peer\);/);
assert.match(html, /if \(isViewVisible\('phone-sub-bip'\)\) \{ bipStatusRefresh\(\); bipLogRefresh\(\); \}/);
});
+1 -1
View File
@@ -236,7 +236,7 @@ test('the chip view groups the sections into a grid', () => {
test('the eSIM LPA pill is wired into the Simulator tab', () => {
assert.ok(html.includes('data-phone-sub="esim"'));
assert.ok(html.includes('id="phone-sub-esim"'));
assert.ok(html.includes("'phone', 'tr', 'esim'"));
assert.ok(html.includes("'phone', 'tr', 'bip', 'esim'"));
assert.ok(html.includes('id="esim-refresh-btn"'));
assert.ok(html.includes('id="esim-profiles"'));
assert.ok(html.includes('id="esim-notifications"'));
+14
View File
@@ -31,6 +31,7 @@ const code = extractFunc(html, 'phoneSwitchSubtab') + '\n' +
'globalThis.tpRefresh = () => { globalThis._tp = (globalThis._tp || 0) + 1; };\n' +
'globalThis.esimFetchAll = () => { globalThis._esim = (globalThis._esim || 0) + 1; };\n' +
'globalThis.testInit = () => { globalThis._test = (globalThis._test || 0) + 1; };\n' +
'globalThis.bipEnter = () => { globalThis._bip = (globalThis._bip || 0) + 1; };\n' +
'globalThis.netStateFetch = () => { globalThis._netstate = (globalThis._netstate || 0) + 1; };\n';
eval(code);
@@ -46,12 +47,14 @@ function setup() {
const buttons = [
{ dataset: { phoneSub: 'phone' }, classList: makeClassList() },
{ dataset: { phoneSub: 'tr' }, classList: makeClassList() },
{ dataset: { phoneSub: 'bip' }, classList: makeClassList() },
{ dataset: { phoneSub: 'esim' }, classList: makeClassList() },
{ dataset: { phoneSub: 'test' }, classList: makeClassList() },
];
const panels = {
'phone-sub-phone': { classList: makeClassList() },
'phone-sub-tr': { classList: makeClassList() },
'phone-sub-bip': { classList: makeClassList() },
'phone-sub-esim': { classList: makeClassList() },
'phone-sub-test': { classList: makeClassList() },
};
@@ -63,6 +66,7 @@ function setup() {
globalThis._stk = globalThis._events = globalThis._log = globalThis._poll = globalThis._pli = globalThis._tp = 0;
globalThis._esim = globalThis._netstate = 0;
globalThis._test = 0;
globalThis._bip = 0;
return { buttons, panels };
}
@@ -78,6 +82,16 @@ test('TR Config pill shows the TR panel and renders PLI data', () => {
assert.strictEqual(globalThis._stk, 0);
});
test('BIP pill shows the BIP panel', () => {
const { buttons, panels } = setup();
phoneSwitchSubtab('bip');
assert.ok(!panels['phone-sub-bip'].classList.has('hidden'));
assert.ok(panels['phone-sub-phone'].classList.has('hidden'));
assert.strictEqual(globalThis._anchor, 'bip');
assert.strictEqual(globalThis._bip, 1);
assert.strictEqual(globalThis._stk, 0);
});
test('Phone pill shows the phone panel and renders CAT views', () => {
const { buttons, panels } = setup();
phoneSwitchSubtab('phone');
+1 -1
View File
@@ -258,7 +258,7 @@ test('the item text check offers contains/exact only', () => {
test('the Simulator hosts the Test script pill and its wiring', () => {
assert.match(html, /data-phone-sub="test"[^>]*data-l10n="Test script"/);
assert.match(html, /id="phone-sub-test"/);
assert.match(html, /\['phone', 'tr', 'esim', 'test'\]/);
assert.match(html, /\['phone', 'tr', 'bip', 'esim', 'test'\]/);
assert.match(html, /else if \(name === 'test'\) \{\s*testInit\(\);/);
assert.match(html, /test: 'test-script'/);
assert.match(html, /testRunActive\(\) && el\.closest && !el\.closest\('#phone-sub-test'\)/);
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "pysim-simple-server"
version = "3.5.19"
version = "3.6.0"
description = "HTTP REST server wrapping pysim for the SIMple PWA"
requires-python = ">=3.8"
# pysim is a git-only dependency installed explicitly by setup.bat/setup.sh.
+14
View File
@@ -423,6 +423,7 @@ class TcpDumpServer:
self.on_log = on_log
self.stopped = False
self.conns = []
self.accepted = 0
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.sock.bind((host, int(port)))
@@ -443,11 +444,13 @@ class TcpDumpServer:
except OSError:
break
self.conns.append(conn)
self.accepted += 1
if self.on_log:
self.on_log('conn', peer='%s:%d' % addr[:2])
threading.Thread(target=self._conn_loop, args=(conn, addr), daemon=True).start()
def _conn_loop(self, conn, addr):
total = 0
try:
while not self.stopped:
conn.settimeout(0.2)
@@ -459,9 +462,12 @@ class TcpDumpServer:
break
if not data:
break
total += len(data)
if self.on_rx:
self.on_rx('%s:%d' % addr[:2], data)
finally:
if self.on_log:
self.on_log('conn-close', peer='%s:%d' % addr[:2], bytes=total)
try:
conn.close()
except OSError:
@@ -481,3 +487,11 @@ class TcpDumpServer:
except OSError:
pass
self.conns = []
# Wait for the accept loop to leave accept(): a thread blocked in
# accept() keeps the listening socket alive for up to its poll
# timeout, so the port must be released before stop() returns
# (restarting a listener on the same port must not race).
try:
self.thread.join(timeout=2)
except RuntimeError:
pass
+180 -76
View File
@@ -31,7 +31,7 @@ from osmocom.tlv import BER_TLV_IE
from osmocom.utils import rpad
VERSION = '3.5.19'
VERSION = '3.6.0'
MAX_ENVELOPE_SEGMENTS = 5 # max SMS segments for outgoing C-APDU in ENVELOPE
@@ -1312,15 +1312,20 @@ PLI_QUALIFIER_NAMES = {
_PLI_DATA = {q: '' for q in PLI_QUALIFIER_NAMES}
_BIP = httpota.BipTerminal()
_SCP81_LISTENER = None
# Active listener mode: 'dump' | 'tls' | 'redirect' | 'passthru'.
# 'redirect' pins one target and has no listener object - the BIP channels
# connect straight to the configured external platform (TLS terminated
# there). 'passthru' has neither listener nor target: each channel dials the
# destination the card requests in OPEN CHANNEL. Mode/target are tracked here
# for the status API.
_SCP81_MODE = None
_SCP81_TARGET = None
# Active BIP session, shared by the SCP81 listener and the Simulator's
# generic BIP pill - only one can run at a time:
# 'tls' | 'dump' SCP81 PSK TLS / capture listener
# 'sink' generic BIP local sink (accept + log, never answers)
# 'redirect' pinned target, no listener object: every channel connects
# to the configured platform (TLS terminated there)
# 'passthru' no listener and no target: each channel dials the
# destination the card requests in OPEN CHANNEL
# _BIP_OWNER records which control started it ('scp81' | 'bip') so both
# status views can show who owns the terminal.
_BIP_LISTENER = None
_BIP_MODE = None
_BIP_TARGET = None
_BIP_OWNER = 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
@@ -2057,32 +2062,34 @@ def _handle_bip_command(scc, cmd_num, cmd_type, cmd_qual, raw, dev_src, dev_dst)
return None
def _scp81_listener_status():
if not _SCP81_LISTENER:
if _SCP81_MODE == 'redirect' and _SCP81_TARGET:
return {'mode': 'redirect', 'host': _SCP81_TARGET[0],
'port': _SCP81_TARGET[1],
'target': '%s:%d' % _SCP81_TARGET}
if _SCP81_MODE == 'passthru':
def _bip_listener_status():
if not _BIP_LISTENER:
if _BIP_MODE == 'redirect' and _BIP_TARGET:
return {'mode': 'redirect', 'host': _BIP_TARGET[0],
'port': _BIP_TARGET[1],
'target': '%s:%d' % _BIP_TARGET}
if _BIP_MODE == 'passthru':
# No listener and no pinned target: every channel dials the
# destination the card requests (per-channel targets in the
# BIP status).
return {'mode': 'passthru'}
return None
if isinstance(_SCP81_LISTENER, scp81.PskTlsServer):
return {'mode': 'tls', 'host': _SCP81_LISTENER.host, 'port': _SCP81_LISTENER.port,
'psk_identities': _SCP81_LISTENER.psk_identities,
'psk_wildcard': _SCP81_LISTENER.wildcard_psk is not None,
'identity_seen': _SCP81_LISTENER.identity_seen,
'identity_matched': _SCP81_LISTENER.identity_matched,
'version_seen': _SCP81_LISTENER.version_seen,
'cipher_seen': _SCP81_LISTENER.cipher_seen,
'chunked': _SCP81_LISTENER.chunked,
'chunk_size': _SCP81_LISTENER.chunk_size,
'compact_headers': _SCP81_LISTENER.compact_headers,
'tls_version': _SCP81_LISTENER.tls_version,
'cipher': _SCP81_LISTENER.cipher}
return {'mode': 'dump', 'host': _SCP81_LISTENER.host, 'port': _SCP81_LISTENER.port}
if isinstance(_BIP_LISTENER, scp81.PskTlsServer):
return {'mode': 'tls', 'host': _BIP_LISTENER.host, 'port': _BIP_LISTENER.port,
'psk_identities': _BIP_LISTENER.psk_identities,
'psk_wildcard': _BIP_LISTENER.wildcard_psk is not None,
'identity_seen': _BIP_LISTENER.identity_seen,
'identity_matched': _BIP_LISTENER.identity_matched,
'version_seen': _BIP_LISTENER.version_seen,
'cipher_seen': _BIP_LISTENER.cipher_seen,
'chunked': _BIP_LISTENER.chunked,
'chunk_size': _BIP_LISTENER.chunk_size,
'compact_headers': _BIP_LISTENER.compact_headers,
'tls_version': _BIP_LISTENER.tls_version,
'cipher': _BIP_LISTENER.cipher}
return {'mode': _BIP_MODE or 'dump', 'host': _BIP_LISTENER.host,
'port': _BIP_LISTENER.port,
'connections': _BIP_LISTENER.accepted}
def _bip_data_available(ch):
@@ -2206,7 +2213,7 @@ _SCP81_CHUNKED = False
# Send automatic Channel status (link dropped) events to the card. Suppress
# while testing flows where the terminal closes the connection on purpose:
# the card must drain the buffered response and resume on a new connection.
_SCP81_LINK_EVENTS = True
_BIP_LINK_EVENTS = True
def _ber_len_bytes(n):
@@ -2511,14 +2518,14 @@ def _scp81_update_psk_map(body):
return {'ok': False, 'error': err}
if not table:
return {'ok': False, 'error': 'no usable PSK entries (identity + key required)'}
if not isinstance(_SCP81_LISTENER, scp81.PskTlsServer):
if not isinstance(_BIP_LISTENER, scp81.PskTlsServer):
return {'ok': False, 'error': 'PSK TLS listener is not running'}
_SCP81_LISTENER.set_psk_map(table)
_BIP_LISTENER.set_psk_map(table)
_SCP81_PSKS = dict(table)
_SCP81_PSK_LEGACY = None
_BIP.log('tls-psk-map', identities=sorted(table))
return {'ok': True, 'identities': sorted(table),
'listener': _scp81_listener_status()}
'listener': _bip_listener_status()}
def _scp81_gen_install(body):
@@ -2573,33 +2580,113 @@ def _cap_info_body(body):
'load_file_bytes': len(loadfile_data) // 2, 'memory': info}
def _bip_session_stop():
"""Stop the active BIP session, whichever control started it."""
global _BIP_LISTENER, _BIP_MODE, _BIP_TARGET, _BIP_OWNER
replaced = None
if _BIP_OWNER:
replaced = {'owner': _BIP_OWNER, 'mode': _BIP_MODE}
if _BIP_LISTENER:
_BIP_LISTENER.stop()
_BIP_LISTENER = None
_BIP_MODE = None
_BIP_TARGET = None
_BIP_OWNER = None
_BIP.disable()
return replaced
def _bip_status_body():
return {'owner': _BIP_OWNER, 'bip': _BIP.status(),
'listener': _bip_listener_status()}
def _bip_control(body):
"""Generic BIP terminal control (Simulator -> BIP pill).
'sink' (default) accepts the card's channels on a local listener and only
logs what arrives; 'redirect' forwards every channel to a fixed target;
'passthru' dials the destination from the card's OPEN CHANNEL. Starting
replaces any running BIP session (an SCP81 listener included) - one BIP
session at a time; the response reports what was replaced."""
global _BIP_LISTENER, _BIP_MODE, _BIP_TARGET, _BIP_OWNER, _BIP_LINK_EVENTS
body = body or {}
action = body.get('action', 'start')
if action == 'stop':
replaced = _bip_session_stop()
resp = _bip_status_body()
resp.update({'ok': True, 'replaced': replaced})
return resp
mode = body.get('mode', 'sink')
if mode not in ('passthru', 'redirect', 'sink'):
return {'ok': False, 'error': 'unsupported mode: %s' % mode}
host = (body.get('host') or '').strip() or '127.0.0.1'
raw_port = body.get('port')
if mode == 'redirect':
if not body.get('host') or raw_port in (None, ''):
return {'ok': False,
'error': 'redirect mode requires the target host and port'}
try:
port = int(raw_port)
except (TypeError, ValueError):
return {'ok': False, 'error': 'port is not a number'}
if not 0 < port <= 0xFFFF:
return {'ok': False, 'error': 'port must be 1-65535'}
elif mode == 'sink':
try:
port = int(raw_port) if raw_port not in (None, '') else 0
except (TypeError, ValueError):
return {'ok': False, 'error': 'port is not a number'}
if not 0 <= port <= 0xFFFF:
return {'ok': False, 'error': 'port must be 0-65535 (0 = ephemeral)'}
replaced = _bip_session_stop()
_BIP_LINK_EVENTS = bool(body.get('link_events', True))
_BIP.on_data = _bip_data_available
if mode == 'passthru':
_BIP_MODE = 'passthru'
_BIP_TARGET = None
_BIP.enable(mode='passthru')
elif mode == 'redirect':
_BIP_MODE = 'redirect'
_BIP_TARGET = (host, port)
_BIP.enable(host, port, mode='redirect')
else:
_BIP_LISTENER = httpota.TcpDumpServer(
host, port,
on_rx=lambda peer, data: _BIP.log('sink-rx', peer=peer,
bytes=len(data),
hex=data.hex().upper()[:2000]),
on_log=lambda kind, **fields: _BIP.log(kind, **fields))
_BIP_MODE = 'sink'
_BIP_TARGET = (_BIP_LISTENER.host, _BIP_LISTENER.port)
_BIP.enable(_BIP_LISTENER.host, _BIP_LISTENER.port, mode='redirect')
_BIP_OWNER = 'bip'
resp = _bip_status_body()
resp.update({'ok': True, 'replaced': replaced})
return resp
def _scp81_bip_control(body):
global _SCP81_LISTENER, _SCP81_PSKS, _SCP81_PSK_LEGACY
global _SCP81_MODE, _SCP81_TARGET
global _BIP_LISTENER, _SCP81_PSKS, _SCP81_PSK_LEGACY
global _BIP_MODE, _BIP_TARGET, _BIP_OWNER
global _SCP81_SCRIPT_TEMPLATE, _SCP81_SCRIPT_CR_TAG, _SCP81_NEXT_URI
global _SCP81_LINK_EVENTS, _SCP81_TARGETED_APP
global _BIP_LINK_EVENTS, _SCP81_TARGETED_APP
global _SCP81_CHUNKED
body = body or {}
action = body.get('action', 'start')
if action == 'stop':
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}
replaced = _bip_session_stop()
resp = _bip_status_body()
resp.update({'ok': True, 'replaced': replaced})
return resp
host = body.get('host') or '127.0.0.1'
port = body.get('port')
port = int(port) if port not in (None, '') else 8443
mode = body.get('mode', 'dump')
if _SCP81_LISTENER:
_SCP81_LISTENER.stop()
_SCP81_LISTENER = None
_BIP.disable()
replaced = _bip_session_stop()
# Channel status events (TS 102 223 7.5.11) apply to every mode: the
# terminal reports BIP link changes it detects outside proactive commands.
_SCP81_LINK_EVENTS = bool(body.get('link_events', True))
_BIP_LINK_EVENTS = bool(body.get('link_events', True))
if mode == 'redirect':
# No local listener: the card's BIP channels are redirected straight
# to the configured target (e.g. a production HTTP OTA server), which
@@ -2608,22 +2695,24 @@ def _scp81_bip_control(body):
if not body.get('host') or body.get('port') in (None, ''):
return {'ok': False,
'error': 'redirect mode requires the target host and port'}
_SCP81_MODE = 'redirect'
_SCP81_TARGET = (host, port)
_BIP_MODE = 'redirect'
_BIP_TARGET = (host, port)
_BIP_OWNER = 'scp81'
_BIP.on_data = _bip_data_available
_BIP.enable(host, port, mode='redirect')
return {'ok': True, 'bip': _BIP.status(),
'listener': _scp81_listener_status()}
return {'ok': True, 'replaced': replaced, 'bip': _BIP.status(),
'listener': _bip_listener_status()}
if mode == 'passthru':
# No local listener and no pinned target: every BIP channel dials the
# destination the card requests in OPEN CHANNEL (Other address +
# Transport level port, TCP client only). Host and port are unused.
_SCP81_MODE = 'passthru'
_SCP81_TARGET = None
_BIP_MODE = 'passthru'
_BIP_TARGET = None
_BIP_OWNER = 'scp81'
_BIP.on_data = _bip_data_available
_BIP.enable(mode='passthru')
return {'ok': True, 'bip': _BIP.status(),
'listener': _scp81_listener_status()}
return {'ok': True, 'replaced': replaced, 'bip': _BIP.status(),
'listener': _bip_listener_status()}
if mode == 'tls':
raw_map = body.get('psk_map')
table, err = _parse_psk_map(raw_map)
@@ -2683,7 +2772,7 @@ def _scp81_bip_control(body):
_SCP81_CHUNKED = bool(body.get('chunked', True))
cs = body.get('chunk_size')
chunk_size = int(cs) if cs not in (None, '') else 0
_SCP81_LISTENER = scp81.PskTlsServer(
_BIP_LISTENER = scp81.PskTlsServer(
host, port, psk, identity=identity, psk_map=(table or None),
responder=_scp81_script_responder,
chunked=bool(body.get('chunked', True)),
@@ -2695,30 +2784,34 @@ def _scp81_bip_control(body):
conn_header=(body.get('conn_header') or 'none'),
answer_delay=(body.get('answer_delay') or 0),
on_log=lambda kind, **fields: _BIP.log(kind, **fields))
_SCP81_PSKS = dict(_SCP81_LISTENER.psk_map)
_SCP81_PSKS = dict(_BIP_LISTENER.psk_map)
_SCP81_PSK_LEGACY = None if table else (psk, identity)
_SCP81_MODE = 'tls'
_SCP81_TARGET = (_SCP81_LISTENER.host, _SCP81_LISTENER.port)
_BIP_MODE = 'tls'
_BIP_TARGET = (_BIP_LISTENER.host, _BIP_LISTENER.port)
_BIP_OWNER = 'scp81'
_BIP.on_data = _bip_data_available
_BIP.enable(host, _SCP81_LISTENER.port, mode='redirect')
return {'ok': True, 'bip': _BIP.status(), 'listener': _scp81_listener_status(),
_BIP.enable(host, _BIP_LISTENER.port, mode='redirect')
return {'ok': True, 'replaced': replaced, 'bip': _BIP.status(),
'listener': _bip_listener_status(),
'script': list(_SCP81_SCRIPT_BASE),
'script_kind': _SCP81_SCRIPT_KIND,
'script_template': _SCP81_SCRIPT_TEMPLATE,
'cr_tag': _SCP81_SCRIPT_CR_TAG, 'link_events': _SCP81_LINK_EVENTS,
'cr_tag': _SCP81_SCRIPT_CR_TAG, 'link_events': _BIP_LINK_EVENTS,
'targeted_app': _SCP81_TARGETED_APP,
'chunked': _SCP81_CHUNKED}
if mode != 'dump':
return {'ok': False, 'error': 'unsupported mode: %s' % mode}
_BIP.on_data = _bip_data_available
_SCP81_LISTENER = httpota.TcpDumpServer(
_BIP_LISTENER = httpota.TcpDumpServer(
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, mode='redirect')
return {'ok': True, 'bip': _BIP.status(), 'listener': _scp81_listener_status()}
_BIP_MODE = 'dump'
_BIP_TARGET = (_BIP_LISTENER.host, _BIP_LISTENER.port)
_BIP_OWNER = 'scp81'
_BIP.enable(host, _BIP_LISTENER.port, mode='redirect')
return {'ok': True, 'replaced': replaced, 'bip': _BIP.status(),
'listener': _bip_listener_status()}
def _build_tr(scc, cmd_num, cmd_type, dev_src, dev_dst, cmd_qual):
@@ -3499,7 +3592,7 @@ def _bip_flush_channel_events(scc):
global _FLUSHING_CHANNEL_EVENTS
if _FLUSHING_CHANNEL_EVENTS:
return
if not _SCP81_LINK_EVENTS:
if not _BIP_LINK_EVENTS:
_BIP.take_pending_events()
return
ev_list = getattr(_server_ref, 'event_list', None) or []
@@ -4150,6 +4243,7 @@ _TEST_BLOCKED_PATHS = frozenset([
'/api/esim/profiles', '/api/esim/notifications', '/api/esim/profile',
'/api/scp81/bip', '/api/scp81/queue', '/api/scp81/psk-map',
'/api/scp81/gen-install', '/api/scp81/log-clear',
'/api/bip/control', '/api/bip/log-clear',
])
_TEST_COMMAND_TYPES = {name.upper(): code for code, name in PROACTIVE_TYPE_NAMES.items()}
@@ -4996,12 +5090,13 @@ class PysimHandler(BaseHTTPRequestHandler):
'pending_type': self.server.stk_pending['type'] if self.server.stk_pending else None}
self._send_json(resp)
self._log_resp(resp)
elif self.path == '/api/scp81/status':
elif self.path in ('/api/scp81/status', '/api/bip/status'):
self._log_req()
resp = {'bip': _BIP.status(), 'listener': _scp81_listener_status()}
resp = _bip_status_body()
self._send_json(resp)
self._log_resp(resp)
elif self.path == '/api/scp81/log' or self.path.startswith('/api/scp81/log?'):
elif (self.path == '/api/scp81/log' or self.path.startswith('/api/scp81/log?')
or self.path == '/api/bip/log' or self.path.startswith('/api/bip/log?')):
self._log_req()
after = 0
if '?' in self.path:
@@ -6003,6 +6098,15 @@ class PysimHandler(BaseHTTPRequestHandler):
resp = {'ok': False, 'error': str(e)}
self._send_json(resp)
self._log_resp(resp)
elif self.path == '/api/bip/control':
body = self._read_body()
self._log_req(body)
try:
resp = _bip_control(body)
except Exception as e:
resp = {'ok': False, 'error': str(e)}
self._send_json(resp)
self._log_resp(resp)
elif self.path == '/api/scp81/psk-map':
body = self._read_body()
self._log_req(_redact_psk_fields(body))
@@ -6019,7 +6123,7 @@ class PysimHandler(BaseHTTPRequestHandler):
apdus = [a for a in apdus if a]
if not apdus:
resp = {'ok': False, 'error': 'no apdus given'}
elif _SCP81_LISTENER is None:
elif _BIP_LISTENER is None:
resp = {'ok': False, 'error': 'SCP81 listener is not running'}
else:
queued = _scp81_queue_script(
@@ -6049,7 +6153,7 @@ class PysimHandler(BaseHTTPRequestHandler):
resp = {'ok': False, 'error': str(e)}
self._send_json(resp)
self._log_resp(resp)
elif self.path == '/api/scp81/log-clear':
elif self.path in ('/api/scp81/log-clear', '/api/bip/log-clear'):
body = self._read_body()
self._log_req(body)
_BIP.clear_log()
+238
View File
@@ -0,0 +1,238 @@
#!/usr/bin/env python3
"""Unit tests for the generic BIP terminal control (Simulator -> BIP pill).
The three modes share the SCP81 BIP terminal but are independent of HTTP OTA:
'sink' accepts and only logs (never answers), 'redirect' forwards to a fixed
target and 'passthru' dials the destination from the card's OPEN CHANNEL.
Only one BIP session runs at a time - starting either control replaces the
other.
"""
import socket
import sys
import threading
import time
import unittest
from pathlib import Path
PROJECTS = Path(__file__).resolve().parents[2]
PY_SIM = PROJECTS / 'pysim'
if str(PY_SIM) not in sys.path:
sys.path.insert(0, str(PY_SIM))
import pysim_simple_server.server as server
class PeerServer(threading.Thread):
"""Tiny TCP peer: accepts one connection, greets, records what it gets."""
def __init__(self, greeting=b''):
super().__init__(daemon=True)
self.sock = socket.socket()
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.sock.bind(('127.0.0.1', 0))
self.sock.listen(1)
self.port = self.sock.getsockname()[1]
self.greeting = greeting
self.received = b''
self.conn = None
self.ready = threading.Event()
self.done = threading.Event()
def run(self):
self.sock.settimeout(3)
try:
self.conn, _ = self.sock.accept()
except OSError:
return
self.ready.set()
if self.greeting:
self.conn.sendall(self.greeting)
self.conn.settimeout(3)
try:
while True:
data = self.conn.recv(4096)
if not data:
break
self.received += data
except OSError:
pass
self.done.set()
def stop(self):
for s in (self.conn, self.sock):
try:
if s:
s.close()
except OSError:
pass
def drain(channel_id, tries=20):
"""Receive from a BIP channel with a short retry loop."""
for _ in range(tries):
data = server._BIP.receive(channel_id, 100)
if data:
return data
time.sleep(0.05)
return b''
def kinds():
return [e['kind'] for e in server._BIP.entries_after(0)]
class BipControlTest(unittest.TestCase):
def tearDown(self):
server._bip_control({'action': 'stop'})
def test_sink_is_the_default_and_accepts_ephemeral_ports(self):
resp = server._bip_control({})
self.assertTrue(resp['ok'], resp)
self.assertEqual(resp['owner'], 'bip')
self.assertEqual(resp['bip']['mode'], 'redirect') # BIP-level mode
self.assertEqual(resp['listener']['mode'], 'sink')
self.assertEqual(resp['listener']['host'], '127.0.0.1')
self.assertGreater(resp['listener']['port'], 0) # 0 -> ephemeral
self.assertIsNone(resp['replaced'])
def test_sink_accepts_and_logs_but_never_answers(self):
resp = server._bip_control({'mode': 'sink', 'port': 0})
port = resp['listener']['port']
cid, err = server._BIP.open('10.9.9.9', 1234, 512)
self.assertIsNone(err)
self.assertEqual(server._BIP.channels[cid].target, ('127.0.0.1', port))
self.assertTrue(server._BIP.send(cid, b'CARDHELLO'))
# the sink receives and logs the bytes ...
for _ in range(40):
if 'sink-rx' in kinds():
break
time.sleep(0.05)
rx = [e for e in server._BIP.entries_after(0) if e['kind'] == 'sink-rx']
self.assertTrue(rx, kinds())
self.assertEqual(rx[0]['bytes'], len(b'CARDHELLO'))
self.assertEqual(rx[0]['hex'], b'CARDHELLO'.hex().upper())
# ... and sends nothing back
self.assertEqual(server._BIP.available(cid), 0)
self.assertEqual(drain(cid), b'')
# the listener reports the accepted connection
st = server._bip_status_body()
self.assertEqual(st['listener']['mode'], 'sink')
self.assertEqual(st['listener']['connections'], 1)
def test_redirect_forwards_to_the_fixed_target(self):
peer = PeerServer(greeting=b'PLATFORM')
peer.start()
try:
resp = server._bip_control({'mode': 'redirect',
'host': '127.0.0.1', 'port': peer.port})
self.assertTrue(resp['ok'], resp)
self.assertEqual(resp['listener']['mode'], 'redirect')
self.assertEqual(resp['bip']['target'], '127.0.0.1:%d' % peer.port)
cid, err = server._BIP.open('10.9.9.9', 10174, 512)
self.assertIsNone(err)
self.assertTrue(server._BIP.send(cid, b'CARDHELLO'))
self.assertEqual(drain(cid), b'PLATFORM')
server._BIP.close(cid)
peer.done.wait(3)
self.assertEqual(peer.received, b'CARDHELLO')
finally:
peer.stop()
def test_passthru_dials_the_destination_from_open_channel(self):
peer = PeerServer(greeting=b'PLATFORM')
peer.start()
try:
resp = server._bip_control({'mode': 'passthru'})
self.assertTrue(resp['ok'], resp)
self.assertEqual(resp['listener'], {'mode': 'passthru'})
cid, err = server._BIP.open('127.0.0.1', peer.port, 512, proto=0x02)
self.assertIsNone(err)
self.assertEqual(server._BIP.channels[cid].target,
('127.0.0.1', peer.port))
self.assertTrue(server._BIP.send(cid, b'CARDHELLO'))
self.assertEqual(drain(cid), b'PLATFORM')
finally:
peer.stop()
def test_starting_bip_replaces_the_scp81_listener(self):
peer = PeerServer()
peer.start()
try:
resp = server._scp81_bip_control({'action': 'start', 'mode': 'redirect',
'host': '127.0.0.1', 'port': peer.port})
self.assertTrue(resp['ok'], resp)
self.assertEqual(server._BIP_OWNER, 'scp81')
# the BIP pill takes the terminal over: the SCP81 session is
# reported as replaced and no longer owns the terminal
resp = server._bip_control({'mode': 'sink', 'port': 0})
self.assertTrue(resp['ok'], resp)
self.assertEqual(resp['replaced'], {'owner': 'scp81', 'mode': 'redirect'})
self.assertEqual(server._BIP_OWNER, 'bip')
self.assertEqual(server._BIP_MODE, 'sink')
finally:
peer.stop()
def test_scp81_start_replaces_the_bip_session(self):
resp = server._bip_control({'mode': 'sink', 'port': 0})
self.assertTrue(resp['ok'], resp)
sink_port = resp['listener']['port']
peer = PeerServer()
peer.start()
try:
resp = server._scp81_bip_control({'action': 'start', 'mode': 'redirect',
'host': '127.0.0.1', 'port': peer.port})
self.assertTrue(resp['ok'], resp)
self.assertEqual(resp['replaced'], {'owner': 'bip', 'mode': 'sink'})
self.assertEqual(server._BIP_OWNER, 'scp81')
self.assertEqual(server._BIP_MODE, 'redirect')
finally:
peer.stop()
# the old sink listener is gone: its port accepts nothing anymore
probe = socket.socket()
probe.settimeout(0.3)
self.addCleanup(probe.close)
with self.assertRaises(OSError):
probe.connect(('127.0.0.1', sink_port))
def test_status_reports_the_owner(self):
st = server._bip_status_body()
self.assertIsNone(st['owner'])
self.assertFalse(st['bip']['enabled'])
server._bip_control({'mode': 'passthru'})
st = server._bip_status_body()
self.assertEqual(st['owner'], 'bip')
self.assertTrue(st['bip']['enabled'])
self.assertEqual(st['listener'], {'mode': 'passthru'})
def test_stop_reports_what_was_running(self):
server._bip_control({'mode': 'passthru'})
resp = server._bip_control({'action': 'stop'})
self.assertTrue(resp['ok'], resp)
self.assertEqual(resp['replaced'], {'owner': 'bip', 'mode': 'passthru'})
self.assertIsNone(resp['owner'])
self.assertFalse(resp['bip']['enabled'])
# stopping an idle terminal is fine too
resp = server._bip_control({'action': 'stop'})
self.assertTrue(resp['ok'], resp)
self.assertIsNone(resp['replaced'])
def test_invalid_requests_do_not_start_a_session(self):
for body in ({'mode': 'udp'},
{'mode': 'redirect', 'port': 1234}, # no host
{'mode': 'redirect', 'host': '127.0.0.1'}, # no port
{'mode': 'redirect', 'host': '127.0.0.1', 'port': 'x'},
{'mode': 'redirect', 'host': '127.0.0.1', 'port': 70000},
{'mode': 'sink', 'port': 'x'},
{'mode': 'sink', 'port': 70000}):
resp = server._bip_control(body)
self.assertFalse(resp['ok'], resp)
self.assertTrue(resp['error'], resp)
self.assertFalse(server._BIP.enabled, body)
self.assertIsNone(server._BIP_OWNER, body)
# a valid start still works afterwards
self.assertTrue(server._bip_control({'mode': 'sink', 'port': 0})['ok'])
if __name__ == '__main__':
unittest.main()
+11 -11
View File
@@ -666,7 +666,7 @@ class BipControlTest(unittest.TestCase):
self.assertTrue(server._SCP81_SCRIPT_CR_TAG)
self.assertEqual(server._SCP81_TARGETED_APP, '//aid/A000000151000000')
self.assertEqual(server._SCP81_NEXT_URI, '')
self.assertFalse(server._SCP81_LINK_EVENTS)
self.assertFalse(server._BIP_LINK_EVENTS)
self.assertEqual(resp['script_template'], 'definite')
self.assertTrue(resp['cr_tag'])
self.assertFalse(resp['link_events'])
@@ -676,7 +676,7 @@ class BipControlTest(unittest.TestCase):
server._SCP81_SCRIPT_CR_TAG = False
server._SCP81_TARGETED_APP = None
server._SCP81_NEXT_URI = None
server._SCP81_LINK_EVENTS = True
server._BIP_LINK_EVENTS = True
def test_link_events_apply_to_every_mode(self):
# TS 102 223 7.5.11 events are a BIP-layer feature, not a TLS option.
@@ -684,11 +684,11 @@ class BipControlTest(unittest.TestCase):
'host': '10.11.12.13', 'port': 10174,
'link_events': False})
self.assertTrue(resp['ok'], resp)
self.assertFalse(server._SCP81_LINK_EVENTS)
self.assertFalse(server._BIP_LINK_EVENTS)
resp = server._scp81_bip_control({'action': 'start', 'mode': 'passthru',
'link_events': True})
self.assertTrue(resp['ok'], resp)
self.assertTrue(server._SCP81_LINK_EVENTS)
self.assertTrue(server._BIP_LINK_EVENTS)
def test_tls_handshake_failure_is_logged(self):
resp = server._scp81_bip_control({'action': 'start', 'mode': 'tls',
@@ -770,7 +770,7 @@ class BipControlTest(unittest.TestCase):
resp = server._scp81_bip_control({'action': 'start', 'mode': 'redirect',
'host': '10.11.12.13', 'port': 10174})
self.assertTrue(resp['ok'], resp)
self.assertIsNone(server._SCP81_LISTENER) # no local listener
self.assertIsNone(server._BIP_LISTENER) # no local listener
self.assertEqual(resp['listener']['mode'], 'redirect')
self.assertEqual(resp['listener']['host'], '10.11.12.13')
self.assertEqual(resp['listener']['port'], 10174)
@@ -778,10 +778,10 @@ class BipControlTest(unittest.TestCase):
self.assertTrue(resp['bip']['enabled'])
self.assertEqual(server._BIP.target, ('10.11.12.13', 10174))
# the status endpoint sees the redirect mode while it runs ...
self.assertEqual(server._scp81_listener_status()['mode'], 'redirect')
self.assertEqual(server._bip_listener_status()['mode'], 'redirect')
# ... and stopping clears it (no stale listener in the status)
server._scp81_bip_control({'action': 'stop'})
self.assertIsNone(server._scp81_listener_status())
self.assertIsNone(server._bip_listener_status())
self.assertFalse(server._BIP.enabled)
def test_redirect_mode_requires_an_explicit_target(self):
@@ -795,7 +795,7 @@ class BipControlTest(unittest.TestCase):
resp = server._scp81_bip_control({'action': 'start', 'mode': 'redirect',
'port': 1234})
self.assertFalse(resp['ok'])
self.assertIsNone(server._SCP81_LISTENER)
self.assertIsNone(server._BIP_LISTENER)
self.assertFalse(server._BIP.enabled)
def test_passthru_mode_needs_no_target(self):
@@ -803,15 +803,15 @@ class BipControlTest(unittest.TestCase):
# the destination the card requests in OPEN CHANNEL.
resp = server._scp81_bip_control({'action': 'start', 'mode': 'passthru'})
self.assertTrue(resp['ok'], resp)
self.assertIsNone(server._SCP81_LISTENER)
self.assertIsNone(server._BIP_LISTENER)
self.assertEqual(resp['listener'], {'mode': 'passthru'})
self.assertEqual(server._BIP.mode, 'passthru')
self.assertIsNone(server._BIP.target)
self.assertTrue(resp['bip']['enabled'])
self.assertEqual(resp['bip']['mode'], 'passthru')
self.assertEqual(server._scp81_listener_status(), {'mode': 'passthru'})
self.assertEqual(server._bip_listener_status(), {'mode': 'passthru'})
server._scp81_bip_control({'action': 'stop'})
self.assertIsNone(server._scp81_listener_status())
self.assertIsNone(server._bip_listener_status())
self.assertFalse(server._BIP.enabled)
def test_start_accepts_explicit_script_list(self):