Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/Crypto/Random/_UserFriendlyRNG.py: 82%

1# -*- coding: utf-8 -*- 

2# 

3# Random/_UserFriendlyRNG.py : A user-friendly random number generator 

4# 

5# Written in 2008 by Dwayne C. Litzenberger <dlitz@dlitz.net> 

6# 

7# =================================================================== 

8# The contents of this file are dedicated to the public domain. To 

9# the extent that dedication to the public domain is not available, 

10# everyone is granted a worldwide, perpetual, royalty-free, 

11# non-exclusive license to exercise all rights associated with the 

12# contents of this file for any purpose whatsoever. 

13# No rights are reserved. 

14# 

15# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, 

16# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF 

17# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND 

18# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS 

19# BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN 

20# ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN 

21# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE 

22# SOFTWARE. 

23# =================================================================== 

24 

25__revision__ = "$Id$" 

26 

27import sys 

28if sys.version_info[0] == 2 and sys.version_info[1] == 1: 

29 from Crypto.Util.py21compat import * 

30 

31import os 

32import threading 

33import struct 

34import time 

35from math import floor 

36 

37from Crypto.Random import OSRNG 

38from Crypto.Random.Fortuna import FortunaAccumulator 

39 

40class _EntropySource(object): 

41 def __init__(self, accumulator, src_num): 

42 self._fortuna = accumulator 

43 self._src_num = src_num 

44 self._pool_num = 0 

45 

46 def feed(self, data): 

47 self._fortuna.add_random_event(self._src_num, self._pool_num, data) 

48 self._pool_num = (self._pool_num + 1) & 31 

49 

50class _EntropyCollector(object): 

51 

52 def __init__(self, accumulator): 

53 self._osrng = OSRNG.new() 

54 self._osrng_es = _EntropySource(accumulator, 255) 

55 self._time_es = _EntropySource(accumulator, 254) 

56 self._clock_es = _EntropySource(accumulator, 253) 

57 

58 def reinit(self): 

59 # Add 256 bits to each of the 32 pools, twice. (For a total of 16384 

60 # bits collected from the operating system.) 

61 for i in range(2): 

62 block = self._osrng.read(32*32) 

63 for p in range(32): 

64 self._osrng_es.feed(block[p*32:(p+1)*32]) 

65 block = None 

66 self._osrng.flush() 

67 

68 def collect(self): 

69 # Collect 64 bits of entropy from the operating system and feed it to Fortuna. 

70 self._osrng_es.feed(self._osrng.read(8)) 

71 

72 # Add the fractional part of time.time() 

73 t = time.time() 

74 self._time_es.feed(struct.pack("@I", int(2**30 * (t - floor(t))))) 

75 

76 # Add the fractional part of time.perf_counter() 

77 t = time.perf_counter() 

78 self._clock_es.feed(struct.pack("@I", int(2**30 * (t - floor(t))))) 

79 

80 

81class _UserFriendlyRNG(object): 

82 

83 def __init__(self): 

84 self.closed = False 

85 self._fa = FortunaAccumulator.FortunaAccumulator() 

86 self._ec = _EntropyCollector(self._fa) 

87 self.reinit() 

88 

89 def reinit(self): 

90 """Initialize the random number generator and seed it with entropy from 

91 the operating system. 

92 """ 

93 

94 # Save the pid (helps ensure that Crypto.Random.atfork() gets called) 

95 self._pid = os.getpid() 

96 

97 # Collect entropy from the operating system and feed it to 

98 # FortunaAccumulator 

99 self._ec.reinit() 

100 

101 # Override FortunaAccumulator's 100ms minimum re-seed interval. This 

102 # is necessary to avoid a race condition between this function and 

103 # self.read(), which that can otherwise cause forked child processes to 

104 # produce identical output. (e.g. CVE-2013-1445) 

105 # 

106 # Note that if this function can be called frequently by an attacker, 

107 # (and if the bits from OSRNG are insufficiently random) it will weaken 

108 # Fortuna's ability to resist a state compromise extension attack. 

109 self._fa._forget_last_reseed() 

110 

111 def close(self): 

112 self.closed = True 

113 self._osrng = None 

114 self._fa = None 

115 

116 def flush(self): 

117 pass 

118 

119 def read(self, N): 

120 """Return N bytes from the RNG.""" 

121 if self.closed: 

122 raise ValueError("I/O operation on closed file") 

123 if not isinstance(N, int): 

124 raise TypeError("an integer is required") 

125 if N < 0: 

126 raise ValueError("cannot read to end of infinite stream") 

127 

128 # Collect some entropy and feed it to Fortuna 

129 self._ec.collect() 

130 

131 # Ask Fortuna to generate some bytes 

132 retval = self._fa.random_data(N) 

133 

134 # Check that we haven't forked in the meantime. (If we have, we don't 

135 # want to use the data, because it might have been duplicated in the 

136 # parent process. 

137 self._check_pid() 

138 

139 # Return the random data. 

140 return retval 

141 

142 def _check_pid(self): 

143 # Lame fork detection to remind developers to invoke Random.atfork() 

144 # after every call to os.fork(). Note that this check is not reliable, 

145 # since process IDs can be reused on most operating systems. 

146 # 

147 # You need to do Random.atfork() in the child process after every call 

148 # to os.fork() to avoid reusing PRNG state. If you want to avoid 

149 # leaking PRNG state to child processes (for example, if you are using 

150 # os.setuid()) then you should also invoke Random.atfork() in the 

151 # *parent* process. 

152 if os.getpid() != self._pid: 

153 raise AssertionError("PID check failed. RNG must be re-initialized after fork(). Hint: Try Random.atfork()") 

154 

155 

156class _LockingUserFriendlyRNG(_UserFriendlyRNG): 

157 def __init__(self): 

158 self._lock = threading.Lock() 

159 _UserFriendlyRNG.__init__(self) 

160 

161 def close(self): 

162 self._lock.acquire() 

163 try: 

164 return _UserFriendlyRNG.close(self) 

165 finally: 

166 self._lock.release() 

167 

168 def reinit(self): 

169 self._lock.acquire() 

170 try: 

171 return _UserFriendlyRNG.reinit(self) 

172 finally: 

173 self._lock.release() 

174 

175 def read(self, bytes): 

176 self._lock.acquire() 

177 try: 

178 return _UserFriendlyRNG.read(self, bytes) 

179 finally: 

180 self._lock.release() 

181 

182class RNGFile(object): 

183 def __init__(self, singleton): 

184 self.closed = False 

185 self._singleton = singleton 

186 

187 # PEP 343: Support for the "with" statement 

188 def __enter__(self): 

189 """PEP 343 support""" 

190 def __exit__(self): 

191 """PEP 343 support""" 

192 self.close() 

193 

194 def close(self): 

195 # Don't actually close the singleton, just close this RNGFile instance. 

196 self.closed = True 

197 self._singleton = None 

198 

199 def read(self, bytes): 

200 if self.closed: 

201 raise ValueError("I/O operation on closed file") 

202 return self._singleton.read(bytes) 

203 

204 def flush(self): 

205 if self.closed: 

206 raise ValueError("I/O operation on closed file") 

207 

208_singleton_lock = threading.Lock() 

209_singleton = None 

210def _get_singleton(): 

211 global _singleton 

212 _singleton_lock.acquire() 

213 try: 

214 if _singleton is None: 

215 _singleton = _LockingUserFriendlyRNG() 

216 return _singleton 

217 finally: 

218 _singleton_lock.release() 

219 

220def new(): 

221 return RNGFile(_get_singleton()) 

222 

223def reinit(): 

224 _get_singleton().reinit() 

225 

226def get_random_bytes(n): 

227 """Return the specified number of cryptographically-strong random bytes.""" 

228 return _get_singleton().read(n) 

229 

230# vim:set ts=4 sw=4 sts=4 expandtab: