From 9326031b307921599630e8b779aa33a60ad11db4 Mon Sep 17 00:00:00 2001 From: gpotter2 <10530980+gpotter2@users.noreply.github.com> Date: Thu, 12 Nov 2026 22:46:30 +0100 Subject: [PATCH] kerberos/ntlm: check sequence numbers AI-Assisted: yes (Codex GPT-6.1-Sol) --- scapy/layers/kerberos.py | 95 ++++++++++++++++++++++++---------- scapy/layers/ntlm.py | 23 +++++++- test/scapy/layers/kerberos.uts | 2 +- 3 files changed, 91 insertions(+), 29 deletions(-) diff --git a/scapy/layers/kerberos.py b/scapy/layers/kerberos.py index 5ee0356a895..3cb4784d577 100644 --- a/scapy/layers/kerberos.py +++ b/scapy/layers/kerberos.py @@ -5150,7 +5150,19 @@ def GSS_VerifyMICEx(self, Context, msgs, signature): - AES: RFC4121 sect 4.2.6.1 """ - Context.RecvSeqNum = signature.root.SND_SEQ + # [RFC4121] sect 4.2.1: per-message sequence numbers increase by one. + # [RFC2743] sect 1.2.3: replay detection and sequencing are optional + if ( + not Context.passive + and Context.flags + & (GSS_C_FLAGS.GSS_C_REPLAY_FLAG | GSS_C_FLAGS.GSS_C_SEQUENCE_FLAG) + and signature.root.SND_SEQ != Context.RecvSeqNum + ): + raise ValueError( + "ERROR: Sequence numbers don't match: received %s, expected %s" + % (signature.root.SND_SEQ, Context.RecvSeqNum) + ) + if Context.KrbSessionKey.etype in [17, 18]: # AES # Concatenate the ToSign ToSign = b"".join(x.data for x in msgs if x.sign) @@ -5161,9 +5173,12 @@ def GSS_VerifyMICEx(self, Context, msgs, signature): ) else: raise NotImplementedError + if sig != signature.root.SGN_CKSUM: raise ValueError("ERROR: Checksums don't match") + Context.RecvSeqNum += 1 + def GSS_WrapEx(self, Context, msgs, qop_req: GSS_QOP_REQ_FLAGS = 0): """ [MS-KILE] sect 3.4.5.4 @@ -5350,6 +5365,15 @@ def GSS_UnwrapEx(self, Context, msgs, signature): - HMAC-RC4: RFC4757 sect 7.3 """ if Context.KrbSessionKey.etype in [17, 18]: # AES + # [RFC4121] sect 4.2.1; [RFC2743] sect 1.2.3 + if ( + not Context.passive + and Context.flags + & (GSS_C_FLAGS.GSS_C_REPLAY_FLAG | GSS_C_FLAGS.GSS_C_SEQUENCE_FLAG) + and signature.root.SND_SEQ != Context.RecvSeqNum + ): + raise ValueError("ERROR: Sequence numbers don't match") + confidentiality = signature.root.Flags.Sealed # Real separation starts now: RFC4121 sect 4.2.4 if confidentiality: @@ -5392,6 +5416,7 @@ def MakeToSign(Confounder, DecText): hdr.root.RRC = 0 if f16header != bytes(hdr)[:16]: raise ValueError("ERROR: Headers don't match") + Context.RecvSeqNum += 1 # 6. Split (and ignore filler) offset = 0 for msg in msgs: @@ -5429,6 +5454,7 @@ def MakeToSign(Confounder, DecText): # 4. Compare if sig != Mic: raise ValueError("ERROR: Checksums don't match") + Context.RecvSeqNum += 1 # Case without msgs if len(msgs) == 1 and not msgs[0].data: msgs[0].data = Data @@ -5453,6 +5479,14 @@ def MakeToSign(Confounder, DecText): # 2. Decrypt 'SND_SEQ' rc4 = Cipher(decrepit_algorithms.ARC4(Kseq), mode=None).encryptor() seq = rc4.update(tok.root.SND_SEQ)[:4] + # [RFC4757] sect 7.3; [RFC2743] sect 1.2.3 + if ( + not Context.passive + and Context.flags + & (GSS_C_FLAGS.GSS_C_REPLAY_FLAG | GSS_C_FLAGS.GSS_C_SEQUENCE_FLAG) + and struct.unpack(">I", seq)[0] != Context.RecvSeqNum + ): + raise ValueError("ERROR: Sequence numbers don't match") # 3. Compute the 'Kcrypt' key Klocal = strxor(Kss, len(Kss) * b"\xf0") if Context.KrbSessionKey.etype == 24: # EXP @@ -5484,6 +5518,7 @@ def MakeToSign(Confounder, DecText): text=bytes(tok)[:8] + Confounder + ToSign, cksum=tok.root.SGN_CKSUM, ) + Context.RecvSeqNum += 1 return msgs else: raise NotImplementedError @@ -5802,6 +5837,8 @@ def GSS_Init_sec_context( # Retrieve SessionKey repPart = ap_rep.encPart.decrypt(Context.STSessionKey) + # [RFC4120] sect 3.2.5: initial sequence number for server messages. + Context.RecvSeqNum = repPart.seqNumber.val if repPart.seqNumber else 0 if repPart.subkey is not None: Context.SessionKey = repPart.subkey.keyvalue.val Context.KrbSessionKey = repPart.subkey.toKey() @@ -5981,6 +6018,34 @@ def GSS_Accept_sec_context( Context.STSessionKey = tkt.key.toKey() authenticator = ap_req.authenticator.decrypt(Context.STSessionKey) + # Eventually process the "checksum" + if authenticator.cksum and authenticator.cksum.cksumtype == 0x8003: + # KRB-Authenticator + authcksum = authenticator.cksum.checksum + Context.flags = authcksum.Flags + # Check channel bindings + if ( + chan_bindings != GSS_C_NO_CHANNEL_BINDINGS + and chan_bindings.digestMD5() != authcksum.Bnd + and not ( + GSS_S_FLAGS.GSS_S_ALLOW_MISSING_BINDINGS in req_flags + and authcksum.Bnd == GSS_C_NO_CHANNEL_BINDINGS + ) + ): + # Channel binding checks failed. + return Context, None, GSS_S_BAD_BINDINGS + elif ( + chan_bindings != GSS_C_NO_CHANNEL_BINDINGS + and GSS_S_FLAGS.GSS_S_ALLOW_MISSING_BINDINGS not in req_flags + ): + # Uhoh, we required channel bindings + return Context, None, GSS_S_BAD_BINDINGS + + # [RFC4120] sect 3.2.3: initial sequence number for client messages. + Context.RecvSeqNum = ( + authenticator.seqNumber.val if authenticator.seqNumber else 0 + ) + # Check the ticket validity period, the authenticator timestamp and # prevent against replays. # [RFC4120] sect 3.2.3 @@ -6001,7 +6066,7 @@ def GSS_Accept_sec_context( elif replay_key in self.ReplayCache: errorCode = "KRB_AP_ERR_REPEAT" gssErrorCode = GSS_S_DUPLICATE_TOKEN - # Don't check times in passive mode: the capture may be old. + if errorCode: warning("KerberosSSP: %s" % errorCode) err = KRB_GSSAPI_Token( @@ -6019,7 +6084,7 @@ def GSS_Accept_sec_context( ) return Context, err, gssErrorCode - # Store the time for future detections + # Cache authenticators to avoid replay self.ReplayCache[replay_key] = ctime + KRB_MaxClockSkew # We take that opportunity to cleanup any expired items @@ -6040,29 +6105,6 @@ def GSS_Accept_sec_context( Context.KrbSessionKey = self.KEY Context.SessionKey = self.KEY.key - # Eventually process the "checksum" - if authenticator.cksum and authenticator.cksum.cksumtype == 0x8003: - # KRB-Authenticator - authcksum = authenticator.cksum.checksum - Context.flags = authcksum.Flags - # Check channel bindings - if ( - chan_bindings != GSS_C_NO_CHANNEL_BINDINGS - and chan_bindings.digestMD5() != authcksum.Bnd - and not ( - GSS_S_FLAGS.GSS_S_ALLOW_MISSING_BINDINGS in req_flags - and authcksum.Bnd == GSS_C_NO_CHANNEL_BINDINGS - ) - ): - # Channel binding checks failed. - return Context, None, GSS_S_BAD_BINDINGS - elif ( - chan_bindings != GSS_C_NO_CHANNEL_BINDINGS - and GSS_S_FLAGS.GSS_S_ALLOW_MISSING_BINDINGS not in req_flags - ): - # Uhoh, we required channel bindings - return Context, None, GSS_S_BAD_BINDINGS - # Build response (RFC4120 sect 3.2.4) ap_rep = KRB_AP_REP(encPart=EncryptedData()) ap_rep.encPart.encrypt( @@ -6158,6 +6200,7 @@ def GSS_Passive_set_Direction(self, Context: CONTEXT, IsAcceptor=False): if Context.IsAcceptor is not IsAcceptor: return # Swap everything + Context.SendSeqNum, Context.RecvSeqNum = Context.RecvSeqNum, Context.SendSeqNum Context.SendSealKeyUsage, Context.RecvSealKeyUsage = ( Context.RecvSealKeyUsage, Context.SendSealKeyUsage, diff --git a/scapy/layers/ntlm.py b/scapy/layers/ntlm.py index 5f9f98b4433..57c50a3ceb8 100644 --- a/scapy/layers/ntlm.py +++ b/scapy/layers/ntlm.py @@ -1475,18 +1475,28 @@ def GSS_VerifyMICEx(self, Context, msgs, signature): """ [MS-NLMP] sect 3.4.9 """ - Context.RecvSeqNum = signature.SeqNum + # [MS-NLMP] sect 3.4.4; [RFC2743] sect 1.2.3 + if ( + not Context.passive + and Context.flags + & (GSS_C_FLAGS.GSS_C_REPLAY_FLAG | GSS_C_FLAGS.GSS_C_SEQUENCE_FLAG) + and signature.SeqNum != Context.RecvSeqNum + ): + raise ValueError("ERROR: Sequence numbers don't match") + # Concatenate the ToSign ToSign = b"".join(x.data for x in msgs if x.sign) sig = MAC( Context.RecvSealHandle, Context.RecvSignKey, - Context.RecvSeqNum, + signature.SeqNum, ToSign, ) if sig.Checksum != signature.Checksum: raise ValueError("ERROR: Checksums don't match") + Context.RecvSeqNum = signature.SeqNum + 1 + def GSS_WrapEx(self, Context, msgs, qop_req=0): """ [MS-NLMP] sect 3.4.6 @@ -1507,6 +1517,15 @@ def GSS_UnwrapEx(self, Context, msgs, signature): """ [MS-NLMP] sect 3.4.7 """ + # [MS-NLMP] sect 3.4.4: reject before consuming the RC4 stream. + if ( + not Context.passive + and Context.flags + & (GSS_C_FLAGS.GSS_C_REPLAY_FLAG | GSS_C_FLAGS.GSS_C_SEQUENCE_FLAG) + and signature.SeqNum != Context.RecvSeqNum + ): + raise ValueError("ERROR: Sequence numbers don't match") + # Decrypt for msg in msgs: if msg.conf_req_flag: diff --git a/test/scapy/layers/kerberos.uts b/test/scapy/layers/kerberos.uts index c0dd5cfbe58..c51b801c12d 100644 --- a/test/scapy/layers/kerberos.uts +++ b/test/scapy/layers/kerberos.uts @@ -1915,7 +1915,7 @@ assert passiveState == GSS_S_COMPLETE assert passiveContext.passive assert passiveContext.ssp_context.passive -= GSS_Accept_sec_context rejects a replayed AP-REQ += GSS_Accept_sec_context rejects a replayed AP-REQ without the replay flag from scapy.layers.gssapi import GSS_S_DUPLICATE_TOKEN