Code-review fixes: bad-format visibility, view-aware sync, counter hygiene
- The RAM step line (and the plain send's PoR line) now name a bad-format
failure ("bad format 02 (wrong length)") instead of reading "PoR ok": the
step record carries por_bad_format_name and ramStepLine renders it.
- ramSyncCounter follows the view the notice came from ('ram'/'sp' - preset
and keyset selectors alike) and refreshes the keyset labels after a sync.
- The ETSI-release line no longer hides behind the memory block (it needs
only the ELF list).
- /api/send-ota reports final_cntr only for a packet the card accepted: the
failed-envelope path used to persist the packet's own counter on a
rejection (the audit trail's small bump on cntr_low).
- _ram_detect_format's docstring corrected (both expanded codings execute on
the live ISD; the earlier bad format was a length-mismatched template).
- Tests: ramStepLine bad format, the view-aware notice, the accepted/rejected
final_cntr cases, and an end-to-end /api/counter-probe endpoint test
(400s, no write on a rejected walk, accepted+1 stored).
- sw cache simple-v356.
This commit is contained in:
@@ -393,6 +393,31 @@ class CntrLowGuardHttpTests(unittest.TestCase):
|
||||
# the verdict is reported without a value (v3.9.x)
|
||||
self.assertNotIn('card_cntr', resp)
|
||||
self.assertNotIn('suggested_cntr', resp)
|
||||
# a rejected packet reports no counter to use next either (v3.9.x):
|
||||
# the store must not move on a packet the card did not consume
|
||||
self.assertNotIn('final_cntr', resp)
|
||||
|
||||
def test_send_ota_reports_the_next_counter_only_when_accepted(self):
|
||||
patches = [
|
||||
mock.patch.object(self.srv, '_send_secured_packet', lambda *a, **k: {
|
||||
'success': True, 'sw': '9000', 'bytes': 34, 'segments': 1,
|
||||
'response_data': '027100000b0a00000000000000bb0000'}),
|
||||
mock.patch.object(self.srv, '_decode_por', lambda *a, **k: {
|
||||
'response_status': 'por_ok', 'tar': '000000', 'cntr': '00000000BB',
|
||||
'pcntr': 0, 'rpl': 11, 'rhl': 10, 'raw': ''}),
|
||||
]
|
||||
self.server.app = object() # the send path only checks truthiness
|
||||
for p in patches:
|
||||
p.start()
|
||||
try:
|
||||
status, resp = self._post('/api/send-ota', {
|
||||
'sp': '00', 'spi1': '16', 'spi2': '01', 'kic': '15', 'kid': '15',
|
||||
'tar': '000000', 'cntr': '00000000BB'})
|
||||
finally:
|
||||
for p in reversed(patches):
|
||||
p.stop()
|
||||
self.assertEqual(status, 200, resp)
|
||||
self.assertEqual(resp['final_cntr'], '00000000BC')
|
||||
|
||||
def test_the_ram_chain_stops_at_a_low_counter(self):
|
||||
calls = []
|
||||
@@ -517,6 +542,48 @@ class PresetStoreHttpTests(unittest.TestCase):
|
||||
self.assertEqual(resp['presets'], [])
|
||||
self.assertEqual(resp['path'], str(self.store.path))
|
||||
|
||||
def test_counter_probe_endpoint(self):
|
||||
status, resp = self._post('/api/presets', self._preset(iccid='8970119000004600098'))
|
||||
self.assertEqual(status, 200, resp)
|
||||
pid = resp['preset']['id']
|
||||
status, resp = self._post('/api/counter-probe', {})
|
||||
self.assertEqual(status, 400, resp)
|
||||
self.assertIn('preset_id', resp['error'])
|
||||
status, resp = self._post('/api/counter-probe', {'preset_id': 'nope'})
|
||||
self.assertEqual(status, 400, resp)
|
||||
self.assertIn('unknown preset', resp['error'])
|
||||
# a walk that never gets accepted leaves the store untouched
|
||||
with mock.patch.object(self.srv, '_build_secured_packet',
|
||||
lambda *a, **k: ('00' * 20, {})), \
|
||||
mock.patch.object(self.srv, '_send_secured_packet', lambda *a, **k: {
|
||||
'success': True, 'sw': '9000', 'bytes': 34, 'segments': 1,
|
||||
'response_data': '027100000b0a00000000000000000000'}), \
|
||||
mock.patch.object(self.srv, '_decode_por', lambda *a, **k: {
|
||||
'response_status': 'cntr_low', 'tar': '000000', 'cntr': '0000000000',
|
||||
'pcntr': 0, 'rpl': 11, 'rhl': 10, 'raw': ''}):
|
||||
status, resp = self._post('/api/counter-probe',
|
||||
{'preset_id': pid, 'max_attempts': 2})
|
||||
self.assertEqual(status, 200, resp)
|
||||
self.assertFalse(resp['success'], resp)
|
||||
self.assertEqual(resp['packets'], 2)
|
||||
self.assertEqual(resp['stopped'], 'attempts')
|
||||
self.assertEqual(self.store.get(pid)['keysets'][0]['cntr'], '0000000001')
|
||||
# an accepted walk stores accepted+1 (the first ladder value)
|
||||
with mock.patch.object(self.srv, '_build_secured_packet',
|
||||
lambda *a, **k: ('00' * 20, {})), \
|
||||
mock.patch.object(self.srv, '_send_secured_packet', lambda *a, **k: {
|
||||
'success': True, 'sw': '9000', 'bytes': 34, 'segments': 1,
|
||||
'response_data': '027100000b0a00000000000000000000'}), \
|
||||
mock.patch.object(self.srv, '_decode_por', lambda *a, **k: {
|
||||
'response_status': 'por_ok', 'tar': '000000', 'cntr': '0000000002',
|
||||
'pcntr': 0, 'rpl': 11, 'rhl': 10, 'raw': ''}):
|
||||
status, resp = self._post('/api/counter-probe', {'preset_id': pid})
|
||||
self.assertEqual(status, 200, resp)
|
||||
self.assertTrue(resp['success'], resp)
|
||||
self.assertEqual(resp['accepted_cntr'], '0000000002')
|
||||
self.assertEqual(resp['stored_cntr'], '0000000003')
|
||||
self.assertEqual(self.store.get(pid)['keysets'][0]['cntr'], '0000000003')
|
||||
|
||||
def test_create_update_delete_round_trip(self):
|
||||
status, resp = self._post('/api/presets', self._preset(iccid='8970119000004600098'))
|
||||
self.assertEqual(status, 200, resp)
|
||||
|
||||
Reference in New Issue
Block a user