1  # By Nicko van Someren, 2021. This code is released into the public domain.
  2  # Small modifications for use in Reticulum, and constant time key exchange
  3  # added by Mark Qvist in 2022.
  4  
  5  # WARNING! Only the X25519PrivateKey.exchange() method attempts to hide execution time.
  6  # In the context of Reticulum, this is sufficient, but it may not be in other systems. If
  7  # this code is to be used to provide cryptographic security in an environment where the
  8  # start and end times of the execution can be guessed, inferred or measured then it is
  9  # critical that steps are taken to hide the execution time, for instance by adding a
 10  # delay so that encrypted packets are not sent until a fixed time after the _start_ of
 11  # execution.
 12  
 13  
 14  import os
 15  import time
 16  
 17  P = 2 ** 255 - 19
 18  _A = 486662
 19  
 20  
 21  def _point_add(point_n, point_m, point_diff):
 22      """Given the projection of two points and their difference, return their sum"""
 23      (xn, zn) = point_n
 24      (xm, zm) = point_m
 25      (x_diff, z_diff) = point_diff
 26      x = (z_diff << 2) * (xm * xn - zm * zn) ** 2
 27      z = (x_diff << 2) * (xm * zn - zm * xn) ** 2
 28      return x % P, z % P
 29  
 30  
 31  def _point_double(point_n):
 32      """Double a point provided in projective coordinates"""
 33      (xn, zn) = point_n
 34      xn2 = xn ** 2
 35      zn2 = zn ** 2
 36      x = (xn2 - zn2) ** 2
 37      xzn = xn * zn
 38      z = 4 * xzn * (xn2 + _A * xzn + zn2)
 39      return x % P, z % P
 40  
 41  
 42  def _const_time_swap(a, b, swap):
 43      """Swap two values in constant time"""
 44      index = int(swap) * 2
 45      temp = (a, b, b, a)
 46      return temp[index:index+2]
 47  
 48  
 49  def _raw_curve25519(base, n):
 50      """Raise the point base to the power n"""
 51      zero = (1, 0)
 52      one = (base, 1)
 53      mP, m1P = zero, one
 54  
 55      for i in reversed(range(256)):
 56          bit = bool(n & (1 << i))
 57          mP, m1P = _const_time_swap(mP, m1P, bit)
 58          mP, m1P = _point_double(mP), _point_add(mP, m1P, one)
 59          mP, m1P = _const_time_swap(mP, m1P, bit)
 60  
 61      x, z = mP
 62      inv_z = pow(z, P - 2, P)
 63      return (x * inv_z) % P
 64  
 65  
 66  def _unpack_number(s):
 67      """Unpack 32 bytes to a 256 bit value"""
 68      if len(s) != 32:
 69          raise ValueError('Curve25519 values must be 32 bytes')
 70      return int.from_bytes(s, "little")
 71  
 72  
 73  def _pack_number(n):
 74      """Pack a value into 32 bytes"""
 75      return n.to_bytes(32, "little")
 76  
 77  
 78  def _fix_secret(n):
 79      """Mask a value to be an acceptable exponent"""
 80      n &= ~7
 81      n &= ~(128 << 8 * 31)
 82      n |= 64 << 8 * 31
 83      return n
 84  
 85  def _fix_base_point(n):
 86      n &= ~(2**255)
 87      return n
 88  
 89  def curve25519(base_point_raw, secret_raw):
 90      """Raise the base point to a given power"""
 91      base_point = _fix_base_point(_unpack_number(base_point_raw))
 92      secret = _fix_secret(_unpack_number(secret_raw))
 93      return _pack_number(_raw_curve25519(base_point, secret))
 94  
 95  
 96  def curve25519_base(secret_raw):
 97      """Raise the generator point to a given power"""
 98      secret = _fix_secret(_unpack_number(secret_raw))
 99      return _pack_number(_raw_curve25519(9, secret))
100  
101  
102  class X25519PublicKey:
103      def __init__(self, x):
104          self.x = x
105  
106      @classmethod
107      def from_public_bytes(cls, data):
108          return cls(_unpack_number(data))
109  
110      def public_bytes(self):
111          return _pack_number(self.x)
112  
113  
114  class X25519PrivateKey:
115      MIN_EXEC_TIME = 0.002
116      MAX_EXEC_TIME = 0.5
117      DELAY_WINDOW = 10
118  
119      T_CLEAR = None
120      T_MAX = 0
121  
122      def __init__(self, a):
123          self.a = a
124  
125      @classmethod
126      def generate(cls):
127          return cls.from_private_bytes(os.urandom(32))
128  
129      @classmethod
130      def from_private_bytes(cls, data):
131          return cls(_fix_secret(_unpack_number(data)))
132  
133      def private_bytes(self):
134          return _pack_number(self.a)
135  
136      def public_key(self):
137          return X25519PublicKey.from_public_bytes(_pack_number(_raw_curve25519(9, self.a)))
138  
139      def exchange(self, peer_public_key):
140          if isinstance(peer_public_key, bytes):
141              peer_public_key = X25519PublicKey.from_public_bytes(peer_public_key)
142  
143          start = time.time()
144          
145          shared = _pack_number(_raw_curve25519(peer_public_key.x, self.a))
146          
147          end = time.time()
148          duration = end-start
149  
150          if X25519PrivateKey.T_CLEAR == None:
151              X25519PrivateKey.T_CLEAR = end + X25519PrivateKey.DELAY_WINDOW
152  
153          if end > X25519PrivateKey.T_CLEAR:
154              X25519PrivateKey.T_CLEAR = end + X25519PrivateKey.DELAY_WINDOW
155              X25519PrivateKey.T_MAX = 0
156          
157          if duration < X25519PrivateKey.T_MAX or duration < X25519PrivateKey.MIN_EXEC_TIME:
158              target = start+X25519PrivateKey.T_MAX
159  
160              if target > start+X25519PrivateKey.MAX_EXEC_TIME:
161                  target = start+X25519PrivateKey.MAX_EXEC_TIME
162  
163              if target < start+X25519PrivateKey.MIN_EXEC_TIME:
164                  target = start+X25519PrivateKey.MIN_EXEC_TIME
165  
166              try:
167                  time.sleep(target-time.time())
168              except Exception as e:
169                  pass
170  
171          elif duration > X25519PrivateKey.T_MAX:
172              X25519PrivateKey.T_MAX = duration
173  
174          return shared