corenet: restore good working conditions for the 4G part of corenet

Author: mitshell

pycrate_corenet/AuC.db

@@ -5,7 +5,7 @@
 # file used by ServerAuC.py
 # 
 # csv'style file, with the following fields:
-#  - IMSI (digits);
+#  - IMSI (15 or 16 digits);
 #  - K (hex), subscriber authentication key,
 #      16-bytes when using Milenage or comp128
 #      16 or 32-bytes when using TUAK;

pycrate_corenet/HdlrUES1.py

@@ -90,7 +90,7 @@ class UEEMMd(SigStack):
     # Unit: 0: 2s, 1: 1mn, 2: 6mn, 7: deactivated
     _T3412             = {'Unit': 2, 'Value': 2} # 12mn
     #_T3412             = {'Unit': 7, 'Value': 0} # deactivated
-    # 
+    #
     # Reattach attempt after a failure timer: dict {'Unit': uint3, 'Value': uint5}
     # Unit: 0: 2s, 1: 1mn, 2: 6mn, 7: deactivated
     _T3402              = {'Unit': 1, 'Value': 2} # 2mn
@@ -379,7 +379,7 @@ def clear(self):
     def require_auth(self, Proc, ksi=None):
         # ksi is a 2-tuple (TSC 0..1, Value 0..7)
         # check if an EMMAuthentication procedure is required
-        if self.S1.SECNAS_DISABLED or self.AUTH_DISABLED:
+        if self.AUTH_DISABLED:
             return False
         elif ksi is None or ksi[1] == 7:
             self.S1.SEC['KSI'] = None
@@ -426,7 +426,7 @@ def require_auth(self, Proc, ksi=None):
 
     def require_smc(self, Proc):
         # check if an EMMSecurityModeControl procedure is required
-        if self.S1.SECNAS_DISABLED or self.SMC_DISABLED:
+        if self.SMC_DISABLED:
             return False
         #
         elif ProcAbbrLUT[Proc.Name] in self.SMC_DISABLED_PROC:
@@ -602,7 +602,7 @@ def _get_sec_eia(self):
                 if UESecCap._content[8+eia].get_val():
                     return eia
             self._log('INF', 'no matching EIA identifier, using EIA%i' % self.SMC_EIA_DEF)
-            return self.SMC_EEA_DEF
+            return self.SMC_EIA_DEF
 
     #--------------------------------------------------------------------------#
     # network-initiated method (fg task, to be used from the interpreter)
@@ -668,14 +668,14 @@ def req_ident(self, idtype=NAS.IDTYPE_IMSI):
         """
         return self.run_proc(EMMIdentification, IDType=idtype)
 
-    def detach(self, type=1, cause=None):
+    def detach(self, typ=1, cause=None):
         """send an EMM Detach with type and cause (optional) and wait for the
         response or timeout
         """
         if cause is not None:
-            return self.run_proc(EMMDetachCN, EPSDetachTypeMT={'Type': type}, EMMCause=cause)
+            return self.run_proc(EMMDetachCN, EPSDetachType={'Type': typ}, EMMCause=cause)
         else:
-            return self.run_proc(EMMDetachCN, EPSDetachTypeMT={'Type': type})
+            return self.run_proc(EMMDetachCN, EPSDetachType={'Type': typ})
 
     def inform(self, **info):
         """send an EMM information with given info
@@ -993,8 +993,8 @@ def process_trans(self, trans_id):
             # 2.2) check the protocol config options
             if trans['ProtConfig']:
                 IEs['ProtConfig'], pdnaddrreq = self.process_protconfig(pdncfg, trans['ProtConfig'])
-                if not pdnaddrreq:
-                    IEs['PDNAddr'] = b''
+                #if not pdnaddrreq:
+                #    IEs['PDNAddr'] = b''
             #
             if 'NBIFOMContainer' in trans:
                 self._log('WNG', 'NBIFOMContainer IE unsupported')
@@ -1124,9 +1124,9 @@ class UES1d(SigStack):
     #--------------------------------------------------------------------------#
     # global security policy
     #--------------------------------------------------------------------------#
-    # this will systematically bypass all auth and smc procedures,
-    # NAS MAC and UL count verification in the uplink
-    # and setting of the EMM security header (and encryption) in the downlink
+    # this will systematically bypass the NAS security layer:
+    # - NAS MAC and UL count verification in the uplink
+    # - setting of the EMM security header (and encryption) in the downlink
     SECNAS_DISABLED = False
     #
     # finer grained NAS security checks:
@@ -1141,6 +1141,11 @@ class UES1d(SigStack):
     # in the downlink for given NAS message (by name)
     SECNAS_PDU_NOSEC = set()
     #
+    # to force EEA0 despite the selected encryption algorithm for DL NAS messages
+    SECNAS_FORCE_EEA0 = False
+    # to force EIA0 despite the selected IP algorithm for DL NAS messages
+    SECNAS_FORCE_EIA0 = False
+    #
     # format of the security context dict self.SEC:
     # self.SEC is a dict of available 3G / 4G security contexts indexed by KSI,
     # and current KSI in use
@@ -1539,8 +1544,10 @@ def process_nas_sec_servreq(self, ServReq):
         else:
             self.SEC['KSI'] = ue_ksi
         secctx = self.SEC[ue_ksi]
-        #
+        if 'Knasint' not in secctx:
+            self.EMM.set_sec_ctx_smc(ue_ksi)
         sqnmsb, sqnlsb = secctx['UL'] & 0xffffffe0, secctx['UL'] & 0x1f
+        #
         verif_mac = ServReq.mac_verify(secctx['Knasint'], 0, secctx['EIA'], sqnmsb)
         verif_sqn = True if ue_sqn == sqnlsb else False
         #
@@ -1725,8 +1732,7 @@ def output_nas_sec(self, NasTx):
         """
         if self.UE.TRACE_NAS_EPS:
             self._log('TRACE_NAS_EPS_DL', '\n' + NasTx.show())
-        if self.SECNAS_DISABLED or NasTx._name in self.SECNAS_PDU_NOSEC or \
-        NasTx._sec == False:
+        if self.SECNAS_DISABLED or NasTx._name in self.SECNAS_PDU_NOSEC or NasTx._sec == False:
             sec = False
         else:
             ksi = self.SEC['KSI']
@@ -1742,6 +1748,8 @@ def output_nas_sec(self, NasTx):
                 return None
             else:
                 secctx = self.SEC[self.SEC['KSI']]
+                if 'Knasint' not in secctx:
+                    self.EMM.set_sec_ctx_smc(ksi)
                 sqnmsb, sqnlsb = secctx['DL'] & 0xffffff00, secctx['DL'] & 0xff
                 if NasTx._name == 'EMMSecurityModeCommand':
                     # integrity protextion only + new security context
@@ -1753,11 +1761,12 @@ def output_nas_sec(self, NasTx):
                     NasTxSec = NAS.EMMSecProtNASMessage(val={'EMMHeaderSec': {'SecHdr': sh},
                                                              'Seqn': sqnlsb,
                                                              'NASMessage': NasTx.to_bytes()})
-                    if sh == 2:
+                    if sh == 2 and not self.SECNAS_FORCE_EEA0:
                         NasTxSec.encrypt(secctx['Knasenc'], 1, secctx['EEA'], sqnmsb)
-                    NasTxSec.mac_compute(secctx['Knasint'], 1, secctx['EIA'], sqnmsb)
-                except Exception:
-                    self._log('ERR', 'NAS SEC DL: unable to protect the NAS message %s' % NasTx._name)
+                    if not self.SECNAS_FORCE_EIA0:
+                        NasTxSec.mac_compute(secctx['Knasint'], 1, secctx['EIA'], sqnmsb)
+                except Exception as err:
+                    self._log('ERR', 'NAS SEC DL: unable to protect the NAS message %s: %r' % (NasTx._name, err))
                     #self.reset_sec_ctx()
                     return None
                 else:
@@ -1789,7 +1798,7 @@ def ret_s1ap_dnt(self, NasTx, **IEs):
             if buf is None:
                 return self._s1ap_nas_sec_err()
             IEs['NAS_PDU'] = buf
-            NgapProc = self.init_ngap_proc(NGAPDownlinkNASTransport, **IEs)
+            S1apProc = self.init_s1ap_proc(S1APDownlinkNASTransport, **IEs)
             if S1apProc:
                 return [S1apProc]
             else:

pycrate_corenet/ProcCNEMM.py

@@ -646,6 +646,7 @@ def output(self):
         if self.UE.IMEISV is None and self.EMM.SMC_IMEISV_REQ:
             IEs['IMEISVReq'] = 1
         # TODO: check support of NonceUE / NonceMME for mobility procedures
+        # TODO: check support for UE Additional Security Capabilities
         #
         self.set_msg(7, 93, **IEs)
         self.encode_msg(7, 93)
@@ -1013,7 +1014,7 @@ def output(self):
             # prepare the TAIList for the UE:
             # it only contains a single PartialTAIList of type 0 with the TAI of the eNB
             # to which the UE is connected
-            tailist = [{'Type':0, 'PLMN':self.UE.PLMN, 'TACValues':[self.UE.TAC]}]
+            tailist = [{'Type':0, 'PTAI':{'PLMN':self.UE.PLMN, 'TACs':[self.UE.TAC]}}]
             #
             # prepare AttachAccept IEs
             IEs = {'EPSAttachResult': self.att_type,
@@ -1235,7 +1236,7 @@ def output(self):
             self._pdu.append( (time(), 'DL', self._nas_tx) )
         self._log('INF', self._nas_tx['EPSDetachType'].repr())
         # in case of IMSI-detach, a TAU with IMSI attach is expected in response
-        if self._nas_tx['EPSDetachType']['Type'].get_val() != 3:
+        if self._nas_tx['EPSDetachType']['EPSDetachTypeMT']['Type'].get_val() != 3:
             self.init_timer()
         else:
             self.rm_from_emm_stack()

pycrate_corenet/ProcCNESM.py

@@ -131,7 +131,7 @@ def set_msg(self, pd, typ, **kw):
 
     def output(self):
         self._log('ERR', 'output() not implemented')
-        return None
+        return []
 
     def process(self, pdu):
         if self.TRACK_PDU:
@@ -140,12 +140,12 @@ def process(self, pdu):
         self.decode_msg(pdu, self.UEInfo)
         #
         self._log('ERR', 'process() not implemented')
-        return None
+        return []
 
     def postprocess(self, Proc=None):
         self._log('ERR', 'postprocess() not implemented')
         self.rm_from_esm_stack()
-        return None
+        return []
 
     def abort(self):
         # abort this procedure, and all procedures started within this one

pycrate_corenet/ServerAuC.py

@@ -31,8 +31,8 @@
 HOWTO:
 
 1) in order to use AuC, the following parameters and files need to be configured:
--> files AuC.db need to be edited with IMSI and authentication parameters from your (U)SIM cards
--> AuC.AUC_DB_PATH can be change if the AuC.db file is put elsewhere
+-> file AuC.db needs to be edited with IMSI and authentication parameters from your (U)SIM cards
+-> AuC.AUC_DB_PATH can be changed if the AuC.db file is put elsewhere, it refers to the directory in which AuC.db lies
 -> AuC.OP needs to be changed according to your Milenage customization
 
 2) To use the AuC (in case your IMSI is '001010000000001'):
@@ -160,7 +160,7 @@ def __init__(self):
             db_fd = open('%sAuC.db' % self.AUC_DB_PATH, 'r')
             # parse it into a dict object with IMSI as key
             for line in db_fd.readlines():
-                if line[0] != '#' and line.count(';') >= 3:
+                if line and line[0] != '#' and line.count(';') >= 3:
                     fields = line[:-1].split(';')
                     IMSI   = str( fields[0] )
                     K      = unhexlify( fields[1].encode('ascii') )
@@ -320,14 +320,10 @@ def make_3g_vector(self, IMSI, AMF=b'\0\0', RAND=None):
             K, ALG, SQN = K_ALG_SQN_OP
             OP = None
         #
-        if SQN == -1:
-            # Milenage / TUAK not supported
-            self._log('WNG', '[make_3g_vector] IMSI %s does not support Milenage / TUAK' % IMSI)
-            return None
-        #
         # increment SQN counter in the db
-        K_ALG_SQN_OP[2] += 1
-        self._save_required = True
+        if SQN >= 0:
+            K_ALG_SQN_OP[2] += 1
+            self._save_required = True
         #
         # pack SQN from integer to a 48-bit buffer
         SQNb = pack('>Q', SQN)[2:]