Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/jupyter_client/localinterfaces.py: 24%

177 statements  

« prev     ^ index     » next       coverage.py v7.2.7, created at 2023-07-01 06:54 +0000

1"""Utilities for identifying local IP addresses.""" 

2# Copyright (c) Jupyter Development Team. 

3# Distributed under the terms of the Modified BSD License. 

4import os 

5import re 

6import socket 

7import subprocess 

8from subprocess import PIPE, Popen 

9from typing import Iterable, List 

10from warnings import warn 

11 

12LOCAL_IPS: List = [] 

13PUBLIC_IPS: List = [] 

14 

15LOCALHOST = "" 

16 

17 

18def _uniq_stable(elems: Iterable) -> List: 

19 """uniq_stable(elems) -> list 

20 

21 Return from an iterable, a list of all the unique elements in the input, 

22 maintaining the order in which they first appear. 

23 """ 

24 seen = set() 

25 value = [] 

26 for x in elems: 

27 if x not in seen: 

28 value.append(x) 

29 seen.add(x) 

30 return value 

31 

32 

33def _get_output(cmd): 

34 """Get output of a command, raising IOError if it fails""" 

35 startupinfo = None 

36 if os.name == "nt": 

37 startupinfo = subprocess.STARTUPINFO() # type:ignore[attr-defined] 

38 startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW # type:ignore[attr-defined] 

39 p = Popen(cmd, stdout=PIPE, stderr=PIPE, startupinfo=startupinfo) # noqa 

40 stdout, stderr = p.communicate() 

41 if p.returncode: 

42 msg = "Failed to run {}: {}".format(cmd, stderr.decode("utf8", "replace")) 

43 raise OSError(msg) 

44 return stdout.decode("utf8", "replace") 

45 

46 

47def _only_once(f): 

48 """decorator to only run a function once""" 

49 f.called = False 

50 

51 def wrapped(**kwargs): 

52 if f.called: 

53 return 

54 ret = f(**kwargs) 

55 f.called = True 

56 return ret 

57 

58 return wrapped 

59 

60 

61def _requires_ips(f): 

62 """decorator to ensure load_ips has been run before f""" 

63 

64 def ips_loaded(*args, **kwargs): 

65 _load_ips() 

66 return f(*args, **kwargs) 

67 

68 return ips_loaded 

69 

70 

71# subprocess-parsing ip finders 

72class NoIPAddresses(Exception): # noqa 

73 pass 

74 

75 

76def _populate_from_list(addrs): 

77 """populate local and public IPs from flat list of all IPs""" 

78 if not addrs: 

79 raise NoIPAddresses 

80 

81 global LOCALHOST 

82 public_ips = [] 

83 local_ips = [] 

84 

85 for ip in addrs: 

86 local_ips.append(ip) 

87 if not ip.startswith("127."): 

88 public_ips.append(ip) 

89 elif not LOCALHOST: 

90 LOCALHOST = ip 

91 

92 if not LOCALHOST or LOCALHOST == "127.0.0.1": 

93 LOCALHOST = "127.0.0.1" 

94 local_ips.insert(0, LOCALHOST) 

95 

96 local_ips.extend(["0.0.0.0", ""]) # noqa 

97 

98 LOCAL_IPS[:] = _uniq_stable(local_ips) 

99 PUBLIC_IPS[:] = _uniq_stable(public_ips) 

100 

101 

102_ifconfig_ipv4_pat = re.compile(r"inet\b.*?(\d+\.\d+\.\d+\.\d+)", re.IGNORECASE) 

103 

104 

105def _load_ips_ifconfig(): 

106 """load ip addresses from `ifconfig` output (posix)""" 

107 

108 try: 

109 out = _get_output("ifconfig") 

110 except OSError: 

111 # no ifconfig, it's usually in /sbin and /sbin is not on everyone's PATH 

112 out = _get_output("/sbin/ifconfig") 

113 

114 lines = out.splitlines() 

115 addrs = [] 

116 for line in lines: 

117 m = _ifconfig_ipv4_pat.match(line.strip()) 

118 if m: 

119 addrs.append(m.group(1)) 

120 _populate_from_list(addrs) 

121 

122 

123def _load_ips_ip(): 

124 """load ip addresses from `ip addr` output (Linux)""" 

125 out = _get_output(["ip", "-f", "inet", "addr"]) 

126 

127 lines = out.splitlines() 

128 addrs = [] 

129 for line in lines: 

130 blocks = line.lower().split() 

131 if (len(blocks) >= 2) and (blocks[0] == "inet"): 

132 addrs.append(blocks[1].split("/")[0]) 

133 _populate_from_list(addrs) 

134 

135 

136_ipconfig_ipv4_pat = re.compile(r"ipv4.*?(\d+\.\d+\.\d+\.\d+)$", re.IGNORECASE) 

137 

138 

139def _load_ips_ipconfig(): 

140 """load ip addresses from `ipconfig` output (Windows)""" 

141 out = _get_output("ipconfig") 

142 

143 lines = out.splitlines() 

144 addrs = [] 

145 for line in lines: 

146 m = _ipconfig_ipv4_pat.match(line.strip()) 

147 if m: 

148 addrs.append(m.group(1)) 

149 _populate_from_list(addrs) 

150 

151 

152def _load_ips_netifaces(): 

153 """load ip addresses with netifaces""" 

154 import netifaces # type: ignore 

155 

156 global LOCALHOST 

157 local_ips = [] 

158 public_ips = [] 

159 

160 # list of iface names, 'lo0', 'eth0', etc. 

161 for iface in netifaces.interfaces(): 

162 # list of ipv4 addrinfo dicts 

163 ipv4s = netifaces.ifaddresses(iface).get(netifaces.AF_INET, []) 

164 for entry in ipv4s: 

165 addr = entry.get("addr") 

166 if not addr: 

167 continue 

168 if not (iface.startswith("lo") or addr.startswith("127.")): 

169 public_ips.append(addr) 

170 elif not LOCALHOST: 

171 LOCALHOST = addr 

172 local_ips.append(addr) 

173 if not LOCALHOST: 

174 # we never found a loopback interface (can this ever happen?), assume common default 

175 LOCALHOST = "127.0.0.1" 

176 local_ips.insert(0, LOCALHOST) 

177 local_ips.extend(["0.0.0.0", ""]) # noqa 

178 LOCAL_IPS[:] = _uniq_stable(local_ips) 

179 PUBLIC_IPS[:] = _uniq_stable(public_ips) 

180 

181 

182def _load_ips_gethostbyname(): 

183 """load ip addresses with socket.gethostbyname_ex 

184 

185 This can be slow. 

186 """ 

187 global LOCALHOST 

188 try: 

189 LOCAL_IPS[:] = socket.gethostbyname_ex("localhost")[2] 

190 except OSError: 

191 # assume common default 

192 LOCAL_IPS[:] = ["127.0.0.1"] 

193 

194 try: 

195 hostname = socket.gethostname() 

196 PUBLIC_IPS[:] = socket.gethostbyname_ex(hostname)[2] 

197 # try hostname.local, in case hostname has been short-circuited to loopback 

198 if not hostname.endswith(".local") and all(ip.startswith("127") for ip in PUBLIC_IPS): 

199 PUBLIC_IPS[:] = socket.gethostbyname_ex(socket.gethostname() + ".local")[2] 

200 except OSError: 

201 pass 

202 finally: 

203 PUBLIC_IPS[:] = _uniq_stable(PUBLIC_IPS) 

204 LOCAL_IPS.extend(PUBLIC_IPS) 

205 

206 # include all-interface aliases: 0.0.0.0 and '' 

207 LOCAL_IPS.extend(["0.0.0.0", ""]) # noqa 

208 

209 LOCAL_IPS[:] = _uniq_stable(LOCAL_IPS) 

210 

211 LOCALHOST = LOCAL_IPS[0] 

212 

213 

214def _load_ips_dumb(): 

215 """Fallback in case of unexpected failure""" 

216 global LOCALHOST 

217 LOCALHOST = "127.0.0.1" 

218 LOCAL_IPS[:] = [LOCALHOST, "0.0.0.0", ""] # noqa 

219 PUBLIC_IPS[:] = [] 

220 

221 

222@_only_once 

223def _load_ips(suppress_exceptions=True): 

224 """load the IPs that point to this machine 

225 

226 This function will only ever be called once. 

227 

228 It will use netifaces to do it quickly if available. 

229 Then it will fallback on parsing the output of ifconfig / ip addr / ipconfig, as appropriate. 

230 Finally, it will fallback on socket.gethostbyname_ex, which can be slow. 

231 """ 

232 

233 try: 

234 # first priority, use netifaces 

235 try: 

236 return _load_ips_netifaces() 

237 except ImportError: 

238 pass 

239 

240 # second priority, parse subprocess output (how reliable is this?) 

241 

242 if os.name == "nt": 

243 try: 

244 return _load_ips_ipconfig() 

245 except (OSError, NoIPAddresses): 

246 pass 

247 else: 

248 try: 

249 return _load_ips_ip() 

250 except (OSError, NoIPAddresses): 

251 pass 

252 try: 

253 return _load_ips_ifconfig() 

254 except (OSError, NoIPAddresses): 

255 pass 

256 

257 # lowest priority, use gethostbyname 

258 

259 return _load_ips_gethostbyname() 

260 except Exception as e: 

261 if not suppress_exceptions: 

262 raise 

263 # unexpected error shouldn't crash, load dumb default values instead. 

264 warn("Unexpected error discovering local network interfaces: %s" % e, stacklevel=2) 

265 _load_ips_dumb() 

266 

267 

268@_requires_ips 

269def local_ips(): 

270 """return the IP addresses that point to this machine""" 

271 return LOCAL_IPS 

272 

273 

274@_requires_ips 

275def public_ips(): 

276 """return the IP addresses for this machine that are visible to other machines""" 

277 return PUBLIC_IPS 

278 

279 

280@_requires_ips 

281def localhost(): 

282 """return ip for localhost (almost always 127.0.0.1)""" 

283 return LOCALHOST 

284 

285 

286@_requires_ips 

287def is_local_ip(ip): 

288 """does `ip` point to this machine?""" 

289 return ip in LOCAL_IPS 

290 

291 

292@_requires_ips 

293def is_public_ip(ip): 

294 """is `ip` a publicly visible address?""" 

295 return ip in PUBLIC_IPS