Harald Welte | 21caf32 | 2022-07-16 14:06:46 +0200 | [diff] [blame] | 1 | # coding=utf-8 |
| 2 | """APDU (and TPDU) parser for UICC/USIM/ISIM cards. |
| 3 | |
| 4 | The File (and its classes) represent the structure / hierarchy |
| 5 | of the APDUs as seen in SIM/UICC/SIM/ISIM cards. The primary use case |
| 6 | is to perform a meaningful decode of protocol traces taken between card and UE. |
| 7 | |
| 8 | The ancient wirshark dissector developed for GSMTAP generated by SIMtrace |
| 9 | is far too simplistic, while this decoder can utilize all of the information |
| 10 | we already know in pySim about the filesystem structure, file encoding, etc. |
| 11 | """ |
| 12 | |
| 13 | # (C) 2022 by Harald Welte <laforge@osmocom.org> |
| 14 | # |
| 15 | # This program is free software: you can redistribute it and/or modify |
| 16 | # it under the terms of the GNU General Public License as published by |
| 17 | # the Free Software Foundation, either version 2 of the License, or |
| 18 | # (at your option) any later version. |
| 19 | # |
| 20 | # This program is distributed in the hope that it will be useful, |
| 21 | # but WITHOUT ANY WARRANTY; without even the implied warranty of |
| 22 | # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
| 23 | # GNU General Public License for more details. |
| 24 | # |
| 25 | # You should have received a copy of the GNU General Public License |
| 26 | # along with this program. If not, see <http://www.gnu.org/licenses/>. |
| 27 | |
| 28 | |
| 29 | import abc |
| 30 | from termcolor import colored |
| 31 | import typing |
| 32 | from typing import List, Dict, Optional |
| 33 | |
| 34 | from construct import * |
| 35 | from construct import Optional as COptional |
| 36 | from pySim.construct import * |
| 37 | from pySim.utils import * |
Harald Welte | 531894d | 2023-07-11 19:11:11 +0200 | [diff] [blame^] | 38 | from pySim.runtime import RuntimeLchan, RuntimeState, lchan_nr_from_cla |
Harald Welte | 21caf32 | 2022-07-16 14:06:46 +0200 | [diff] [blame] | 39 | from pySim.filesystem import CardADF, CardFile, TransparentEF, LinFixedEF |
| 40 | |
| 41 | """There are multiple levels of decode: |
| 42 | |
| 43 | 1) pure TPDU / APDU level (no filesystem state required to decode) |
| 44 | 1a) the raw C-TPDU + R-TPDU |
| 45 | 1b) the raw C-APDU + R-APDU |
| 46 | 1c) the C-APDU + R-APDU split in its portions (p1/p2/lc/le/cmd/rsp) |
| 47 | 1d) the abstract C-APDU + R-APDU (mostly p1/p2 parsing; SELECT response) |
| 48 | 2) the decoded DATA of command/response APDU |
| 49 | * READ/UPDATE: requires state/context: which file is selected? how to decode it? |
| 50 | """ |
| 51 | |
| 52 | class ApduCommandMeta(abc.ABCMeta): |
| 53 | """A meta-class that we can use to set some class variables when declaring |
| 54 | a derived class of ApduCommand.""" |
| 55 | def __new__(metacls, name, bases, namespace, **kwargs): |
| 56 | x = super().__new__(metacls, name, bases, namespace) |
| 57 | x._name = namespace.get('name', kwargs.get('n', None)) |
| 58 | x._ins = namespace.get('ins', kwargs.get('ins', None)) |
| 59 | x._cla = namespace.get('cla', kwargs.get('cla', None)) |
| 60 | return x |
| 61 | |
| 62 | BytesOrHex = typing.Union[bytes, Hexstr] |
| 63 | |
| 64 | class Tpdu: |
| 65 | def __init__(self, cmd: BytesOrHex, rsp: Optional[BytesOrHex] = None): |
| 66 | if isinstance(cmd, str): |
| 67 | self.cmd = h2b(cmd) |
| 68 | else: |
| 69 | self.cmd = cmd |
| 70 | if isinstance(rsp, str): |
| 71 | self.rsp = h2b(rsp) |
| 72 | else: |
| 73 | self.rsp = rsp |
| 74 | |
| 75 | def __str__(self): |
| 76 | return '%s(%02X %02X %02X %02X %02X %s %s %s)' % (type(self).__name__, self.cla, self.ins, self.p1, |
| 77 | self.p2, self.p3, b2h(self.cmd_data), b2h(self.rsp_data), b2h(self.sw)) |
| 78 | |
| 79 | @property |
| 80 | def cla(self) -> int: |
| 81 | """Return CLA of the C-APDU Header.""" |
| 82 | return self.cmd[0] |
| 83 | |
| 84 | @property |
| 85 | def ins(self) -> int: |
| 86 | """Return INS of the C-APDU Header.""" |
| 87 | return self.cmd[1] |
| 88 | |
| 89 | @property |
| 90 | def p1(self) -> int: |
| 91 | """Return P1 of the C-APDU Header.""" |
| 92 | return self.cmd[2] |
| 93 | |
| 94 | @property |
| 95 | def p2(self) -> int: |
| 96 | """Return P2 of the C-APDU Header.""" |
| 97 | return self.cmd[3] |
| 98 | |
| 99 | @property |
| 100 | def p3(self) -> int: |
| 101 | """Return P3 of the C-APDU Header.""" |
| 102 | return self.cmd[4] |
| 103 | |
| 104 | @property |
| 105 | def cmd_data(self) -> int: |
| 106 | """Return the DATA portion of the C-APDU""" |
| 107 | return self.cmd[5:] |
| 108 | |
| 109 | @property |
| 110 | def sw(self) -> Optional[bytes]: |
| 111 | """Return Status Word (SW) of the R-APDU""" |
| 112 | return self.rsp[-2:] if self.rsp else None |
| 113 | |
| 114 | @property |
| 115 | def rsp_data(self) -> Optional[bytes]: |
| 116 | """Return the DATA portion of the R-APDU""" |
| 117 | return self.rsp[:-2] if self.rsp else None |
| 118 | |
| 119 | |
| 120 | class Apdu(Tpdu): |
| 121 | @property |
| 122 | def lc(self) -> int: |
| 123 | """Return Lc; Length of C-APDU body.""" |
| 124 | return len(self.cmd_data) |
| 125 | |
| 126 | @property |
| 127 | def lr(self) -> int: |
| 128 | """Return Lr; Length of R-APDU body.""" |
| 129 | return len(self.rsp_data) |
| 130 | |
| 131 | @property |
| 132 | def successful(self) -> bool: |
| 133 | """Was the execution of this APDU successful?""" |
| 134 | method = getattr(self, '_is_success', None) |
| 135 | if callable(method): |
| 136 | return method() |
| 137 | # default case: only 9000 is success |
Harald Welte | bc7437d | 2022-12-02 23:21:18 +0100 | [diff] [blame] | 138 | if self.sw == b'\x90\x00': |
| 139 | return True |
| 140 | # This is not really a generic positive APDU SW but specific to UICC/SIM |
| 141 | if self.sw[0] == 0x91: |
| 142 | return True |
| 143 | return False |
Harald Welte | 21caf32 | 2022-07-16 14:06:46 +0200 | [diff] [blame] | 144 | |
| 145 | |
| 146 | class ApduCommand(Apdu, metaclass=ApduCommandMeta): |
| 147 | """Base class from which you would derive individual commands/instructions like SELECT. |
| 148 | A derived class represents a decoder for a specific instruction. |
| 149 | An instance of such a derived class is one concrete APDU.""" |
| 150 | # fall-back constructs if the derived class provides no override |
| 151 | _construct_p1 = Byte |
| 152 | _construct_p2 = Byte |
| 153 | _construct = HexAdapter(GreedyBytes) |
| 154 | _construct_rsp = HexAdapter(GreedyBytes) |
| 155 | |
| 156 | def __init__(self, cmd: BytesOrHex, rsp: Optional[BytesOrHex] = None): |
| 157 | """Instantiate a new ApduCommand from give cmd + resp.""" |
| 158 | # store raw data |
| 159 | super().__init__(cmd, rsp) |
| 160 | # default to 'empty' ID column. To be set to useful values (like record number) |
| 161 | # by derived class {cmd_rsp}_to_dict() or process() methods |
| 162 | self.col_id = '-' |
| 163 | # fields only set by process_* methods |
| 164 | self.file = None |
| 165 | self.lchan = None |
| 166 | self.processed = None |
| 167 | # the methods below could raise exceptions and those handlers might assume cmd_{dict,resp} |
| 168 | self.cmd_dict = None |
| 169 | self.rsp_dict = None |
| 170 | # interpret the data |
| 171 | self.cmd_dict = self.cmd_to_dict() |
| 172 | self.rsp_dict = self.rsp_to_dict() if self.rsp else {} |
| 173 | |
| 174 | |
| 175 | @classmethod |
| 176 | def from_apdu(cls, apdu:Apdu, **kwargs) -> 'ApduCommand': |
| 177 | """Instantiate an ApduCommand from an existing APDU.""" |
| 178 | return cls(cmd=apdu.cmd, rsp=apdu.rsp, **kwargs) |
| 179 | |
| 180 | @classmethod |
| 181 | def from_bytes(cls, buffer:bytes) -> 'ApduCommand': |
| 182 | """Instantiate an ApduCommand from a linear byte buffer containing hdr,cmd,rsp,sw. |
| 183 | This is for example used when parsing GSMTAP traces that traditionally contain the |
| 184 | full command and response portion in one packet: "CLA INS P1 P2 P3 DATA SW" and we |
| 185 | now need to figure out whether the DATA part is part of the CMD or the RSP""" |
| 186 | apdu_case = cls.get_apdu_case(buffer) |
| 187 | if apdu_case in [1, 2]: |
| 188 | # data is part of response |
| 189 | return cls(buffer[:5], buffer[5:]) |
| 190 | elif apdu_case in [3, 4]: |
| 191 | # data is part of command |
| 192 | lc = buffer[4] |
| 193 | return cls(buffer[:5+lc], buffer[5+lc:]) |
| 194 | else: |
| 195 | raise ValueError('%s: Invalid APDU Case %u' % (cls.__name__, apdu_case)) |
| 196 | |
| 197 | @property |
| 198 | def path(self) -> List[str]: |
| 199 | """Return (if known) the path as list of files to the file on which this command operates.""" |
| 200 | if self.file: |
| 201 | return self.file.fully_qualified_path() |
| 202 | else: |
| 203 | return [] |
| 204 | |
| 205 | @property |
| 206 | def path_str(self) -> str: |
| 207 | """Return (if known) the path as string to the file on which this command operates.""" |
| 208 | if self.file: |
| 209 | return self.file.fully_qualified_path_str() |
| 210 | else: |
| 211 | return '' |
| 212 | |
| 213 | @property |
| 214 | def col_sw(self) -> str: |
| 215 | """Return the ansi-colorized status word. Green==OK, Red==Error""" |
| 216 | if self.successful: |
| 217 | return colored(b2h(self.sw), 'green') |
| 218 | else: |
| 219 | return colored(b2h(self.sw), 'red') |
| 220 | |
| 221 | @property |
| 222 | def lchan_nr(self) -> int: |
| 223 | """Logical channel number over which this ApduCommand was transmitted.""" |
| 224 | if self.lchan: |
| 225 | return self.lchan.lchan_nr |
| 226 | else: |
| 227 | return lchan_nr_from_cla(self.cla) |
| 228 | |
| 229 | def __str__(self) -> str: |
| 230 | return '%02u %s(%s): %s' % (self.lchan_nr, type(self).__name__, self.path_str, self.to_dict()) |
| 231 | |
| 232 | def __repr__(self) -> str: |
| 233 | return '%s(INS=%02x,CLA=%s)' % (self.__class__, self.ins, self.cla) |
| 234 | |
| 235 | def _process_fallback(self, rs: RuntimeState): |
| 236 | """Fall-back function to be called if there is no derived-class-specific |
| 237 | process_global or process_on_lchan method. Uses information from APDU decode.""" |
| 238 | self.processed = {} |
| 239 | if not 'p1' in self.cmd_dict: |
| 240 | self.processed = self.to_dict() |
| 241 | else: |
| 242 | self.processed['p1'] = self.cmd_dict['p1'] |
| 243 | self.processed['p2'] = self.cmd_dict['p2'] |
| 244 | if 'body' in self.cmd_dict and self.cmd_dict['body']: |
| 245 | self.processed['cmd'] = self.cmd_dict['body'] |
| 246 | if 'body' in self.rsp_dict and self.rsp_dict['body']: |
| 247 | self.processed['rsp'] = self.rsp_dict['body'] |
| 248 | return self.processed |
| 249 | |
| 250 | def process(self, rs: RuntimeState): |
| 251 | # if there is a global method, use that; else use process_on_lchan |
| 252 | method = getattr(self, 'process_global', None) |
| 253 | if callable(method): |
| 254 | self.processed = method(rs) |
| 255 | return self.processed |
| 256 | method = getattr(self, 'process_on_lchan', None) |
| 257 | if callable(method): |
| 258 | self.lchan = rs.get_lchan_by_cla(self.cla) |
| 259 | self.processed = method(self.lchan) |
| 260 | return self.processed |
| 261 | # if none of the two methods exist: |
| 262 | return self._process_fallback(rs) |
| 263 | |
| 264 | @classmethod |
| 265 | def get_apdu_case(cls, hdr:bytes) -> int: |
| 266 | if hasattr(cls, '_apdu_case'): |
| 267 | return cls._apdu_case |
| 268 | method = getattr(cls, '_get_apdu_case', None) |
| 269 | if callable(method): |
| 270 | return method(hdr) |
| 271 | raise ValueError('%s: Class definition missing _apdu_case attribute or _get_apdu_case method' % cls.__name__) |
| 272 | |
| 273 | @classmethod |
| 274 | def match_cla(cls, cla) -> bool: |
| 275 | """Does the given CLA match the CLA list of the command?.""" |
| 276 | if not isinstance(cla, str): |
| 277 | cla = '%02X' % cla |
| 278 | cla = cla.lower() |
| 279 | # see https://github.com/PyCQA/pylint/issues/7219 |
| 280 | # pylint: disable=no-member |
| 281 | for cla_match in cls._cla: |
| 282 | cla_masked = "" |
| 283 | for i in range(0, 2): |
| 284 | if cla_match[i] == 'X': |
| 285 | cla_masked += 'X' |
| 286 | else: |
| 287 | cla_masked += cla[i] |
| 288 | if cla_masked == cla_match: |
| 289 | return True |
| 290 | return False |
| 291 | |
| 292 | def cmd_to_dict(self) -> Dict: |
| 293 | """Convert the Command part of the APDU to a dict.""" |
| 294 | method = getattr(self, '_decode_cmd', None) |
| 295 | if callable(method): |
| 296 | return method() |
| 297 | else: |
| 298 | r = {} |
| 299 | method = getattr(self, '_decode_p1p2', None) |
| 300 | if callable(method): |
| 301 | r = self._decode_p1p2() |
| 302 | else: |
| 303 | r['p1'] = parse_construct(self._construct_p1, self.p1.to_bytes(1, 'big')) |
| 304 | r['p2'] = parse_construct(self._construct_p2, self.p2.to_bytes(1, 'big')) |
| 305 | r['p3'] = self.p3 |
| 306 | if self.cmd_data: |
| 307 | r['body'] = parse_construct(self._construct, self.cmd_data) |
| 308 | return r |
| 309 | |
| 310 | def rsp_to_dict(self) -> Dict: |
| 311 | """Convert the Response part of the APDU to a dict.""" |
| 312 | method = getattr(self, '_decode_rsp', None) |
| 313 | if callable(method): |
| 314 | return method() |
| 315 | else: |
| 316 | r = {} |
| 317 | if self.rsp_data: |
| 318 | r['body'] = parse_construct(self._construct_rsp, self.rsp_data) |
| 319 | r['sw'] = b2h(self.sw) |
| 320 | return r |
| 321 | |
| 322 | def to_dict(self) -> Dict: |
| 323 | """Convert the entire APDU to a dict.""" |
| 324 | return {'cmd': self.cmd_dict, 'rsp': self.rsp_dict} |
| 325 | |
| 326 | def to_json(self) -> str: |
| 327 | """Convert the entire APDU to JSON.""" |
| 328 | d = self.to_dict() |
| 329 | return json.dumps(d) |
| 330 | |
| 331 | def _determine_file(self, lchan) -> CardFile: |
| 332 | """Helper function for read/update commands that might use SFI instead of selected file. |
| 333 | Expects that the self.cmd_dict has already been populated with the 'file' member.""" |
| 334 | if self.cmd_dict['file'] == 'currently_selected_ef': |
| 335 | self.file = lchan.selected_file |
| 336 | elif self.cmd_dict['file'] == 'sfi': |
| 337 | cwd = lchan.get_cwd() |
| 338 | self.file = cwd.lookup_file_by_sfid(self.cmd_dict['sfi']) |
| 339 | |
| 340 | |
| 341 | class ApduCommandSet: |
| 342 | """A set of card instructions, typically specified within one spec.""" |
| 343 | |
| 344 | def __init__(self, name: str, cmds: List[ApduCommand] =[]): |
| 345 | self.name = name |
| 346 | self.cmds = {c._ins: c for c in cmds} |
| 347 | |
| 348 | def __str__(self) -> str: |
| 349 | return self.name |
| 350 | |
| 351 | def __getitem__(self, idx) -> ApduCommand: |
| 352 | return self.cmds[idx] |
| 353 | |
| 354 | def __add__(self, other) -> 'ApduCommandSet': |
| 355 | if isinstance(other, ApduCommand): |
| 356 | if other.ins in self.cmds: |
| 357 | raise ValueError('%s: INS 0x%02x already defined: %s' % |
| 358 | (self, other.ins, self.cmds[other.ins])) |
| 359 | self.cmds[other.ins] = other |
| 360 | elif isinstance(other, ApduCommandSet): |
| 361 | for c in other.cmds.keys(): |
| 362 | self.cmds[c] = other.cmds[c] |
| 363 | else: |
| 364 | raise ValueError( |
| 365 | '%s: Unsupported type to add operator: %s' % (self, other)) |
| 366 | return self |
| 367 | |
| 368 | def lookup(self, ins, cla=None) -> Optional[ApduCommand]: |
| 369 | """look-up the command within the CommandSet.""" |
| 370 | ins = int(ins) |
| 371 | if not ins in self.cmds: |
| 372 | return None |
| 373 | cmd = self.cmds[ins] |
| 374 | if cla and not cmd.match_cla(cla): |
| 375 | return None |
| 376 | return cmd |
| 377 | |
| 378 | def parse_cmd_apdu(self, apdu: Apdu) -> ApduCommand: |
| 379 | """Parse a Command-APDU. Returns an instance of an ApduCommand derived class.""" |
| 380 | # first look-up which of our member classes match CLA + INS |
| 381 | a_cls = self.lookup(apdu.ins, apdu.cla) |
| 382 | if not a_cls: |
| 383 | raise ValueError('Unknown CLA=%02X INS=%02X' % (apdu.cla, apdu.ins)) |
| 384 | # then create an instance of that class and return it |
| 385 | return a_cls.from_apdu(apdu) |
| 386 | |
| 387 | def parse_cmd_bytes(self, buf:bytes) -> ApduCommand: |
| 388 | """Parse from a buffer (simtrace style). Returns an instance of an ApduCommand derived class.""" |
| 389 | # first look-up which of our member classes match CLA + INS |
| 390 | cla = buf[0] |
| 391 | ins = buf[1] |
| 392 | a_cls = self.lookup(ins, cla) |
| 393 | if not a_cls: |
| 394 | raise ValueError('Unknown CLA=%02X INS=%02X' % (cla, ins)) |
| 395 | # then create an instance of that class and return it |
| 396 | return a_cls.from_bytes(buf) |
| 397 | |
| 398 | |
| 399 | |
| 400 | class ApduHandler(abc.ABC): |
| 401 | @abc.abstractmethod |
| 402 | def input(self, cmd: bytes, rsp: bytes): |
| 403 | pass |
| 404 | |
| 405 | |
| 406 | class TpduFilter(ApduHandler): |
| 407 | """The TpduFilter removes the T=0 specific GET_RESPONSE from the TPDU stream and |
| 408 | calls the ApduHandler only with the actual APDU command and response parts.""" |
| 409 | def __init__(self, apdu_handler: ApduHandler): |
| 410 | self.apdu_handler = apdu_handler |
| 411 | self.state = 'INIT' |
| 412 | self.last_cmd = None |
| 413 | |
| 414 | def input_tpdu(self, tpdu:Tpdu): |
| 415 | # handle SW=61xx / 6Cxx |
| 416 | if tpdu.sw[0] == 0x61 or tpdu.sw[0] == 0x6C: |
| 417 | self.state = 'WAIT_GET_RESPONSE' |
| 418 | # handle successive 61/6c responses by stupid phone/modem OS |
| 419 | if tpdu.ins != 0xC0: |
| 420 | self.last_cmd = tpdu.cmd |
| 421 | return None |
| 422 | else: |
| 423 | if self.last_cmd: |
| 424 | icmd = self.last_cmd |
| 425 | self.last_cmd = None |
| 426 | else: |
| 427 | icmd = tpdu.cmd |
| 428 | apdu = Apdu(icmd, tpdu.rsp) |
| 429 | if self.apdu_handler: |
| 430 | return self.apdu_handler.input(apdu) |
| 431 | else: |
| 432 | return Apdu(icmd, tpdu.rsp) |
| 433 | |
| 434 | def input(self, cmd: bytes, rsp: bytes): |
| 435 | if isinstance(cmd, str): |
| 436 | cmd = bytes.fromhex(cmd) |
| 437 | if isinstance(rsp, str): |
| 438 | rsp = bytes.fromhex(rsp) |
| 439 | tpdu = Tpdu(cmd, rsp) |
| 440 | return self.input_tpdu(tpdu) |
| 441 | |
| 442 | class ApduDecoder(ApduHandler): |
| 443 | def __init__(self, cmd_set: ApduCommandSet): |
| 444 | self.cmd_set = cmd_set |
| 445 | |
| 446 | def input(self, apdu: Apdu): |
| 447 | return self.cmd_set.parse_cmd_apdu(apdu) |
| 448 | |
| 449 | |
| 450 | class CardReset: |
| 451 | pass |