1 files changed, 446 insertions, 0 deletions
diff --git a/pySim/apdu/__init__.py b/pySim/apdu/__init__.py
new file mode 100644
index 0000000..3961fe9
--- /dev/null
+++ b/pySim/apdu/__init__.py
@@ -0,0 +1,446 @@
+# coding=utf-8
+"""APDU (and TPDU) parser for UICC/USIM/ISIM cards.
+
+The File (and its classes) represent the structure / hierarchy
+of the APDUs as seen in SIM/UICC/SIM/ISIM cards.  The primary use case
+is to perform a meaningful decode of protocol traces taken between card and UE.
+
+The ancient wirshark dissector developed for GSMTAP generated by SIMtrace
+is far too simplistic, while this decoder can utilize all of the information
+we already know in pySim about the filesystem structure, file encoding, etc.
+"""
+
+# (C) 2022 by Harald Welte <laforge@osmocom.org>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+
+import abc
+from termcolor import colored
+import typing
+from typing import List, Dict, Optional
+
+from construct import *
+from construct import Optional as COptional
+from pySim.construct import *
+from pySim.utils import *
+from pySim.filesystem import RuntimeLchan, RuntimeState, lchan_nr_from_cla
+from pySim.filesystem import CardADF, CardFile, TransparentEF, LinFixedEF
+
+"""There are multiple levels of decode:
+
+    1) pure TPDU / APDU level (no filesystem state required to decode)
+        1a) the raw C-TPDU + R-TPDU
+        1b) the raw C-APDU + R-APDU
+        1c) the C-APDU + R-APDU split in its portions (p1/p2/lc/le/cmd/rsp)
+        1d) the abstract C-APDU + R-APDU (mostly p1/p2 parsing; SELECT response)
+    2) the decoded DATA of command/response APDU
+        * READ/UPDATE: requires state/context: which file is selected? how to decode it?
+"""
+
+class ApduCommandMeta(abc.ABCMeta):
+    """A meta-class that we can use to set some class variables when declaring
+       a derived class of ApduCommand."""
+    def __new__(metacls, name, bases, namespace, **kwargs):
+        x = super().__new__(metacls, name, bases, namespace)
+        x._name = namespace.get('name', kwargs.get('n', None))
+        x._ins = namespace.get('ins', kwargs.get('ins', None))
+        x._cla = namespace.get('cla', kwargs.get('cla', None))
+        return x
+
+BytesOrHex = typing.Union[bytes, Hexstr]
+
+class Tpdu:
+    def __init__(self, cmd: BytesOrHex, rsp: Optional[BytesOrHex] = None):
+        if isinstance(cmd, str):
+            self.cmd = h2b(cmd)
+        else:
+            self.cmd = cmd
+        if  isinstance(rsp, str):
+            self.rsp = h2b(rsp)
+        else:
+            self.rsp = rsp
+
+    def __str__(self):
+        return '%s(%02X %02X %02X %02X %02X %s %s %s)' % (type(self).__name__, self.cla, self.ins, self.p1,
+                self.p2, self.p3, b2h(self.cmd_data), b2h(self.rsp_data), b2h(self.sw))
+
+    @property
+    def cla(self) -> int:
+        """Return CLA of the C-APDU Header."""
+        return self.cmd[0]
+
+    @property
+    def ins(self) -> int:
+        """Return INS of the C-APDU Header."""
+        return self.cmd[1]
+
+    @property
+    def p1(self) -> int:
+        """Return P1 of the C-APDU Header."""
+        return self.cmd[2]
+
+    @property
+    def p2(self) -> int:
+        """Return P2 of the C-APDU Header."""
+        return self.cmd[3]
+
+    @property
+    def p3(self) -> int:
+        """Return P3 of the C-APDU Header."""
+        return self.cmd[4]
+
+    @property
+    def cmd_data(self) -> int:
+        """Return the DATA portion of the C-APDU"""
+        return self.cmd[5:]
+
+    @property
+    def sw(self) -> Optional[bytes]:
+        """Return Status Word (SW) of the R-APDU"""
+        return self.rsp[-2:] if self.rsp else None
+
+    @property
+    def rsp_data(self) -> Optional[bytes]:
+        """Return the DATA portion of the R-APDU"""
+        return self.rsp[:-2] if self.rsp else None
+
+
+class Apdu(Tpdu):
+    @property
+    def lc(self) -> int:
+        """Return Lc; Length of C-APDU body."""
+        return len(self.cmd_data)
+
+    @property
+    def lr(self) -> int:
+        """Return Lr; Length of R-APDU body."""
+        return len(self.rsp_data)
+
+    @property
+    def successful(self) -> bool:
+        """Was the execution of this APDU successful?"""
+        method = getattr(self, '_is_success', None)
+        if callable(method):
+            return method()
+        # default case: only 9000 is success
+        return self.sw == b'\x90\x00'
+
+
+class ApduCommand(Apdu, metaclass=ApduCommandMeta):
+    """Base class from which you would derive individual commands/instructions like SELECT.
+       A derived class represents a decoder for a specific instruction.
+       An instance of such a derived class is one concrete APDU."""
+    # fall-back constructs if the derived class provides no override
+    _construct_p1 = Byte
+    _construct_p2 = Byte
+    _construct = HexAdapter(GreedyBytes)
+    _construct_rsp = HexAdapter(GreedyBytes)
+
+    def __init__(self, cmd: BytesOrHex, rsp: Optional[BytesOrHex] = None):
+        """Instantiate a new ApduCommand from give cmd + resp."""
+        # store raw data
+        super().__init__(cmd, rsp)
+        # default to 'empty' ID column. To be set to useful values (like record number)
+        # by derived class {cmd_rsp}_to_dict() or process() methods
+        self.col_id = '-'
+        # fields only set by process_* methods
+        self.file = None
+        self.lchan = None
+        self.processed = None
+        # the methods below could raise exceptions and those handlers might assume cmd_{dict,resp}
+        self.cmd_dict = None
+        self.rsp_dict = None
+        # interpret the data
+        self.cmd_dict = self.cmd_to_dict()
+        self.rsp_dict = self.rsp_to_dict() if self.rsp else {}
+
+
+    @classmethod
+    def from_apdu(cls, apdu:Apdu, **kwargs) -> 'ApduCommand':
+        """Instantiate an ApduCommand from an existing APDU."""
+        return cls(cmd=apdu.cmd, rsp=apdu.rsp, **kwargs)
+
+    @classmethod
+    def from_bytes(cls, buffer:bytes) -> 'ApduCommand':
+        """Instantiate an ApduCommand from a linear byte buffer containing hdr,cmd,rsp,sw.
+        This is for example used when parsing GSMTAP traces that traditionally contain the
+        full command and response portion in one packet: "CLA INS P1 P2 P3 DATA SW" and we
+        now need to figure out whether the DATA part is part of the CMD or the RSP"""
+        apdu_case = cls.get_apdu_case(buffer)
+        if apdu_case in [1, 2]:
+            # data is part of response
+            return cls(buffer[:5], buffer[5:])
+        elif apdu_case in [3, 4]:
+            # data is part of command
+            lc = buffer[4]
+            return cls(buffer[:5+lc], buffer[5+lc:])
+        else:
+            raise ValueError('%s: Invalid APDU Case %u' % (cls.__name__, apdu_case))
+
+    @property
+    def path(self) -> List[str]:
+        """Return (if known) the path as list of files to the file on which this command operates."""
+        if self.file:
+            return self.file.fully_qualified_path()
+        else:
+            return []
+
+    @property
+    def path_str(self) -> str:
+        """Return (if known) the path as string to the file on which this command operates."""
+        if self.file:
+            return self.file.fully_qualified_path_str()
+        else:
+            return ''
+
+    @property
+    def col_sw(self) -> str:
+        """Return the ansi-colorized status word. Green==OK, Red==Error"""
+        if self.successful:
+            return colored(b2h(self.sw), 'green')
+        else:
+            return colored(b2h(self.sw), 'red')
+
+    @property
+    def lchan_nr(self) -> int:
+        """Logical channel number over which this ApduCommand was transmitted."""
+        if self.lchan:
+            return self.lchan.lchan_nr
+        else:
+            return lchan_nr_from_cla(self.cla)
+
+    def __str__(self) -> str:
+        return '%02u %s(%s): %s' % (self.lchan_nr, type(self).__name__, self.path_str, self.to_dict())
+
+    def __repr__(self) -> str:
+        return '%s(INS=%02x,CLA=%s)' % (self.__class__, self.ins, self.cla)
+
+    def _process_fallback(self, rs: RuntimeState):
+        """Fall-back function to be called if there is no derived-class-specific
+        process_global or process_on_lchan method. Uses information from APDU decode."""
+        self.processed = {}
+        if not 'p1' in self.cmd_dict:
+            self.processed = self.to_dict()
+        else:
+            self.processed['p1'] = self.cmd_dict['p1']
+            self.processed['p2'] = self.cmd_dict['p2']
+            if 'body' in self.cmd_dict and self.cmd_dict['body']:
+                self.processed['cmd'] = self.cmd_dict['body']
+            if 'body' in self.rsp_dict and self.rsp_dict['body']:
+                self.processed['rsp'] = self.rsp_dict['body']
+        return self.processed
+
+    def process(self, rs: RuntimeState):
+        # if there is a global method, use that; else use process_on_lchan
+        method = getattr(self, 'process_global', None)
+        if callable(method):
+            self.processed = method(rs)
+            return self.processed
+        method = getattr(self, 'process_on_lchan', None)
+        if callable(method):
+            self.lchan = rs.get_lchan_by_cla(self.cla)
+            self.processed = method(self.lchan)
+            return self.processed
+        # if none of the two methods exist:
+        return self._process_fallback(rs)
+
+    @classmethod
+    def get_apdu_case(cls, hdr:bytes) -> int:
+        if hasattr(cls, '_apdu_case'):
+            return cls._apdu_case
+        method = getattr(cls, '_get_apdu_case', None)
+        if callable(method):
+            return method(hdr)
+        raise ValueError('%s: Class definition missing _apdu_case attribute or _get_apdu_case method' % cls.__name__)
+
+    @classmethod
+    def match_cla(cls, cla) -> bool:
+        """Does the given CLA match the CLA list of the command?."""
+        if not isinstance(cla, str):
+            cla = '%02X' % cla
+        cla = cla.lower()
+        # see https://github.com/PyCQA/pylint/issues/7219
+        # pylint: disable=no-member
+        for cla_match in cls._cla:
+            cla_masked = ""
+            for i in range(0, 2):
+                if cla_match[i] == 'X':
+                    cla_masked += 'X'
+                else:
+                    cla_masked += cla[i]
+            if cla_masked == cla_match:
+                return True
+        return False
+
+    def cmd_to_dict(self) -> Dict:
+        """Convert the Command part of the APDU to a dict."""
+        method = getattr(self, '_decode_cmd', None)
+        if callable(method):
+            return method()
+        else:
+            r = {}
+            method = getattr(self, '_decode_p1p2', None)
+            if callable(method):
+                r = self._decode_p1p2()
+            else:
+                r['p1'] = parse_construct(self._construct_p1, self.p1.to_bytes(1, 'big'))
+                r['p2'] = parse_construct(self._construct_p2, self.p2.to_bytes(1, 'big'))
+            r['p3'] = self.p3
+            if self.cmd_data:
+                r['body'] = parse_construct(self._construct, self.cmd_data)
+            return r
+
+    def rsp_to_dict(self) -> Dict:
+        """Convert the Response part of the APDU to a dict."""
+        method = getattr(self, '_decode_rsp', None)
+        if callable(method):
+            return method()
+        else:
+            r = {}
+            if self.rsp_data:
+                r['body'] = parse_construct(self._construct_rsp, self.rsp_data)
+            r['sw'] = b2h(self.sw)
+            return r
+
+    def to_dict(self) -> Dict:
+        """Convert the entire APDU to a dict."""
+        return {'cmd': self.cmd_dict, 'rsp': self.rsp_dict}
+
+    def to_json(self) -> str:
+        """Convert the entire APDU to JSON."""
+        d = self.to_dict()
+        return json.dumps(d)
+
+    def _determine_file(self, lchan) -> CardFile:
+        """Helper function for read/update commands that might use SFI instead of selected file.
+        Expects that the self.cmd_dict has already been populated with the 'file' member."""
+        if self.cmd_dict['file'] == 'currently_selected_ef':
+            self.file = lchan.selected_file
+        elif self.cmd_dict['file'] == 'sfi':
+            cwd = lchan.get_cwd()
+            self.file = cwd.lookup_file_by_sfid(self.cmd_dict['sfi'])
+
+
+class ApduCommandSet:
+    """A set of card instructions, typically specified within one spec."""
+
+    def __init__(self, name: str, cmds: List[ApduCommand] =[]):
+        self.name = name
+        self.cmds = {c._ins: c for c in cmds}
+
+    def __str__(self) -> str:
+        return self.name
+
+    def __getitem__(self, idx) -> ApduCommand:
+        return self.cmds[idx]
+
+    def __add__(self, other) -> 'ApduCommandSet':
+        if isinstance(other, ApduCommand):
+            if other.ins in self.cmds:
+                raise ValueError('%s: INS 0x%02x already defined: %s' %
+                                 (self, other.ins, self.cmds[other.ins]))
+            self.cmds[other.ins] = other
+        elif isinstance(other, ApduCommandSet):
+            for c in other.cmds.keys():
+                self.cmds[c] = other.cmds[c]
+        else:
+            raise ValueError(
+                '%s: Unsupported type to add operator: %s' % (self, other))
+        return self
+
+    def lookup(self, ins, cla=None) -> Optional[ApduCommand]:
+        """look-up the command within the CommandSet."""
+        ins = int(ins)
+        if not ins in self.cmds:
+            return None
+        cmd = self.cmds[ins]
+        if cla and not cmd.match_cla(cla):
+            return None
+        return cmd
+
+    def parse_cmd_apdu(self, apdu: Apdu) -> ApduCommand:
+        """Parse a Command-APDU. Returns an instance of an ApduCommand derived class."""
+        # first look-up which of our member classes match CLA + INS
+        a_cls = self.lookup(apdu.ins, apdu.cla)
+        if not a_cls:
+            raise ValueError('Unknown CLA=%02X INS=%02X' % (apdu.cla, apdu.ins))
+        # then create an instance of that class and return it
+        return a_cls.from_apdu(apdu)
+
+    def parse_cmd_bytes(self, buf:bytes) -> ApduCommand:
+        """Parse from a buffer (simtrace style). Returns an instance of an ApduCommand derived class."""
+        # first look-up which of our member classes match CLA + INS
+        cla = buf[0]
+        ins = buf[1]
+        a_cls = self.lookup(ins, cla)
+        if not a_cls:
+            raise ValueError('Unknown CLA=%02X INS=%02X' % (cla, ins))
+        # then create an instance of that class and return it
+        return a_cls.from_bytes(buf)
+
+
+
+class ApduHandler(abc.ABC):
+    @abc.abstractmethod
+    def input(self, cmd: bytes, rsp: bytes):
+        pass
+
+
+class TpduFilter(ApduHandler):
+    """The TpduFilter removes the T=0 specific GET_RESPONSE from the TPDU stream and
+       calls the ApduHandler only with the actual APDU command and response parts."""
+    def __init__(self, apdu_handler: ApduHandler):
+        self.apdu_handler = apdu_handler
+        self.state = 'INIT'
+        self.last_cmd = None
+
+    def input_tpdu(self, tpdu:Tpdu):
+        # handle SW=61xx / 6Cxx
+        if tpdu.sw[0] == 0x61 or tpdu.sw[0] == 0x6C:
+            self.state = 'WAIT_GET_RESPONSE'
+            # handle successive 61/6c responses by stupid phone/modem OS
+            if tpdu.ins != 0xC0:
+                self.last_cmd = tpdu.cmd
+            return None
+        else:
+            if self.last_cmd:
+                icmd = self.last_cmd
+                self.last_cmd = None
+            else:
+                icmd = tpdu.cmd
+            apdu = Apdu(icmd, tpdu.rsp)
+            if self.apdu_handler:
+                return self.apdu_handler.input(apdu)
+            else:
+                return Apdu(icmd, tpdu.rsp)
+
+    def input(self, cmd: bytes, rsp: bytes):
+        if isinstance(cmd, str):
+            cmd = bytes.fromhex(cmd)
+        if isinstance(rsp, str):
+            rsp = bytes.fromhex(rsp)
+        tpdu = Tpdu(cmd, rsp)
+        return self.input_tpdu(tpdu)
+
+class ApduDecoder(ApduHandler):
+    def __init__(self, cmd_set: ApduCommandSet):
+        self.cmd_set = cmd_set
+
+    def input(self, apdu: Apdu):
+        return self.cmd_set.parse_cmd_apdu(apdu)
+
+
+class CardReset:
+    pass