使用cryptography.fernet实现KDC服务器时随机出现InvalidSignature异常求助

问题描述 #

基于cryptography.fernet实现KDC服务器示例时，程序随机出现运行正常或抛出cryptography.exceptions.InvalidSignature: Signature did not match digest异常的情况。密钥仅在main函数启动时创建一次，已排除密钥不一致问题，需要排查异常原因。

代码实现 #

from cryptography.fernet import Fernet
import uuid
import pickle

def generate_challenge():
  return uuid.uuid4().bytes

def serialize(o):
  return pickle.dumps(o)

def deserialize(o):
  return pickle.loads(o)

class InitiateRequest:
  def __init__(self, initiator, responder, challenge):
    self.initiator = initiator    # IDa
    self.responder = responder    # IDb
    self.challenge =  challenge   # N1

class InitiateResponse:
  def __init__(self, session_key, initiator, responder, challenge):
    self.session_key = session_key  # Ks
    self.initiator = initiator      # IDa
    self.responder = responder      # IDb
    self.challenge = challenge      # N1

class InvitationForward:
  def __init__(self, session_key, initiator):
    self.session_key = session_key  # Ks
    self.initiator = initiator      # IDa

class KDCServer:
  def __init__(self, generator: Fernet):
    self.generator = generator
    self.map_keys = {} # user+key pairs

  def subscribe(self, id, key):
    self.map_keys[id] = key
    print("KDCServer: I'm registering key " + str(key) + " for " + id)

  def issue_session_key(self, r: InitiateRequest):
    session_key = self.generator.generate_key()
    response = InitiateResponse(session_key, r.initiator, r.responder, r.challenge) 
    invitation = InvitationForward(session_key, r.initiator)
    print(self.map_keys)
    print("KDCServer: I'm using " + str(self.map_keys[r.initiator]) + " as keyA_KDC")
    keyA_KDC = Fernet(self.map_keys[r.initiator]) 
    print("KDCServer: I'm using " + str(self.map_keys[r.responder]) + " as keyB_KDC")
    keyB_KDC = Fernet(self.map_keys[r.responder])
    print("KDCServer: I've just issued a session key for " + r.initiator + " and " + r.responder)
    return {
        keyA_KDC.encrypt(serialize(response)),   # E(Ka,[Ks|IDa|IDb|N1])
        keyB_KDC.encrypt(serialize(invitation))  # E(Kb,[Ks|IDa])
    }

class User:

  def __init__(self, id:str, key:bytes):
    self.id = id
    self.key = key
    self.session_keys = {}

  def initiate(self, responder):
    challenge = generate_challenge()
    print(self.id + ": Let's retrieve a session key to communicate with " + responder)
    # store request for matching
    self.request = InitiateRequest(self.id, responder, challenge)
    return self.request

  def match_request(self, check:InitiateResponse)->bool:
      return (self.request.challenge == check.challenge) and (self.request.initiator == check.initiator) and (self.request.responder == check.responder)
  
  def accept_response(self, response):
    print(self.id + ": I'm decrypting using my key " + str(self.key))
    check = deserialize(Fernet(self.key).decrypt(response))
    if self.match_request(check):
      self.session_keys[check.responder] = check.session_key # save session key Ks
      print(self.id + ": I've got the session key to communicate with " + check.responder)

  def accept_invitation(self, invitation):
    check = deserialize(Fernet(self.key).decrypt(invitation))
    print(self.id + ": I've accepted the invitation from " + check.initiator)
    self.session_keys[check.initiator] = check.session_key # save session key Ks

  def send_message(self, message, receiver):
    print(self.id + ": I'm sending this message " + str(message) + " using the session key " + str(self.session_keys[receiver]))
    return Fernet(self.session_keys[receiver]).encrypt(message)

  def receive_message(self, cyphered, sender):
    print(self.id + ": I'm decrypting a message using the session key " + str(self.session_keys[sender]))
    message = Fernet(self.session_keys[sender]).decrypt(cyphered)
    print(self.id + ": I've received the message " + str(message) + " from " + sender)

def main():
  alice_key = Fernet.generate_key()
  bob_key = Fernet.generate_key()

  alice = User('Alice', alice_key)
  bob = User('Bob', bob_key)

  server = KDCServer(Fernet)
  server.subscribe(alice.id, alice_key)
  server.subscribe(bob.id, bob_key)

  # Alice sends request to KDC to get a session key
  request = alice.initiate(bob.id)
  response, invitation = server.issue_session_key(request)

  # Alice accepts response from KDC and forwards invitation to Bob
  alice.accept_response(response)
  # Bob accepts invitation from Alice
  bob.accept_invitation(invitation)

  cyphered = bob.send_message(b"My secret message", alice.id)
  alice.receive_message(cyphered, bob.id)

if __name__ == "__main__":
    main()

问题分析 #

异常的核心原因在于KDCServer.issue_session_key方法的返回值类型错误：

• 代码中用大括号{}包裹两个加密结果，创建的是无序集合（set），而非有序的序列结构。
• 集合的元素顺序是随机的，每次运行时，返回的两个加密数据的顺序可能颠倒：
  • 正常情况：response是给Alice的加密响应，invitation是给Bob的加密邀请。
  • 异常情况：response拿到的是给Bob的加密邀请，invitation拿到的是给Alice的加密响应。
• 当顺序颠倒时，Alice用自己的密钥解密Bob的加密数据，Bob用自己的密钥解密Alice的加密数据，自然会触发InvalidSignature异常，且异常出现具有随机性。

解决方案 #

将返回的集合改为有序的元组（tuple），保证两个加密数据的顺序固定：

修改KDCServer.issue_session_key方法的返回语句：

return (
    keyA_KDC.encrypt(serialize(response)),   # E(Ka,[Ks|IDa|IDb|N1])
    keyB_KDC.encrypt(serialize(invitation))  # E(Kb,[Ks|IDa])
)

元组是有序序列，解包时response和invitation会始终对应正确的加密数据，避免因顺序颠倒导致的解密失败。

修改后验证 #

修改后运行程序，每次都会按固定顺序返回加密数据，Alice和Bob能正确解密各自的内容，不会再随机出现签名不匹配的异常。

内容的提问来源于stack exchange，提问作者sam