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

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

178 statements  

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

2# Copyright (c) Jupyter Development Team. 

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

4from __future__ import annotations 

5 

6import os 

7import re 

8import socket 

9import subprocess 

10from subprocess import PIPE, Popen 

11from typing import Any, Callable, Iterable, Sequence 

12from warnings import warn 

13 

14LOCAL_IPS: list = [] 

15PUBLIC_IPS: list = [] 

16 

17LOCALHOST: str = "" 

18 

19 

20def _uniq_stable(elems: Iterable) -> list: 

21 """uniq_stable(elems) -> list 

22 

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

24 maintaining the order in which they first appear. 

25 """ 

26 seen = set() 

27 value = [] 

28 for x in elems: 

29 if x not in seen: 

30 value.append(x) 

31 seen.add(x) 

32 return value 

33 

34 

35def _get_output(cmd: str | Sequence[str]) -> str: 

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

37 startupinfo = None 

38 if os.name == "nt": 

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

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

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

42 stdout, stderr = p.communicate() 

43 if p.returncode: 

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

45 raise OSError(msg) 

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

47 

48 

49def _only_once(f: Callable) -> Callable: 

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

51 f.called = False # type:ignore[attr-defined] 

52 

53 def wrapped(**kwargs: Any) -> Any: 

54 if f.called: # type:ignore[attr-defined] 

55 return 

56 ret = f(**kwargs) 

57 f.called = True # type:ignore[attr-defined] 

58 return ret 

59 

60 return wrapped 

61 

62 

63def _requires_ips(f: Callable) -> Callable: 

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

65 

66 def ips_loaded(*args: Any, **kwargs: Any) -> Any: 

67 _load_ips() 

68 return f(*args, **kwargs) 

69 

70 return ips_loaded 

71 

72 

73# subprocess-parsing ip finders 

74class NoIPAddresses(Exception): # noqa 

75 pass 

76 

77 

78def _populate_from_list(addrs: Sequence[str] | None) -> None: 

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

80 if not addrs: 

81 raise NoIPAddresses 

82 

83 global LOCALHOST 

84 public_ips = [] 

85 local_ips = [] 

86 

87 for ip in addrs: 

88 local_ips.append(ip) 

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

90 public_ips.append(ip) 

91 elif not LOCALHOST: 

92 LOCALHOST = ip 

93 

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

95 LOCALHOST = "127.0.0.1" 

96 local_ips.insert(0, LOCALHOST) 

97 

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

99 

100 LOCAL_IPS[:] = _uniq_stable(local_ips) 

101 PUBLIC_IPS[:] = _uniq_stable(public_ips) 

102 

103 

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

105 

106 

107def _load_ips_ifconfig() -> None: 

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

109 

110 try: 

111 out = _get_output("ifconfig") 

112 except OSError: 

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

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

115 

116 lines = out.splitlines() 

117 addrs = [] 

118 for line in lines: 

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

120 if m: 

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

122 _populate_from_list(addrs) 

123 

124 

125def _load_ips_ip() -> None: 

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

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

128 

129 lines = out.splitlines() 

130 addrs = [] 

131 for line in lines: 

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

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

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

135 _populate_from_list(addrs) 

136 

137 

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

139 

140 

141def _load_ips_ipconfig() -> None: 

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

143 out = _get_output("ipconfig") 

144 

145 lines = out.splitlines() 

146 addrs = [] 

147 for line in lines: 

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

149 if m: 

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

151 _populate_from_list(addrs) 

152 

153 

154def _load_ips_netifaces() -> None: 

155 """load ip addresses with netifaces""" 

156 import netifaces # type: ignore[import-not-found] 

157 

158 global LOCALHOST 

159 local_ips = [] 

160 public_ips = [] 

161 

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

163 for iface in netifaces.interfaces(): 

164 # list of ipv4 addrinfo dicts 

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

166 for entry in ipv4s: 

167 addr = entry.get("addr") 

168 if not addr: 

169 continue 

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

171 public_ips.append(addr) 

172 elif not LOCALHOST: 

173 LOCALHOST = addr 

174 local_ips.append(addr) 

175 if not LOCALHOST: 

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

177 LOCALHOST = "127.0.0.1" 

178 local_ips.insert(0, LOCALHOST) 

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

180 LOCAL_IPS[:] = _uniq_stable(local_ips) 

181 PUBLIC_IPS[:] = _uniq_stable(public_ips) 

182 

183 

184def _load_ips_gethostbyname() -> None: 

185 """load ip addresses with socket.gethostbyname_ex 

186 

187 This can be slow. 

188 """ 

189 global LOCALHOST 

190 try: 

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

192 except OSError: 

193 # assume common default 

194 LOCAL_IPS[:] = ["127.0.0.1"] 

195 

196 try: 

197 hostname = socket.gethostname() 

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

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

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

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

202 except OSError: 

203 pass 

204 finally: 

205 PUBLIC_IPS[:] = _uniq_stable(PUBLIC_IPS) 

206 LOCAL_IPS.extend(PUBLIC_IPS) 

207 

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

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

210 

211 LOCAL_IPS[:] = _uniq_stable(LOCAL_IPS) 

212 

213 LOCALHOST = LOCAL_IPS[0] 

214 

215 

216def _load_ips_dumb() -> None: 

217 """Fallback in case of unexpected failure""" 

218 global LOCALHOST 

219 LOCALHOST = "127.0.0.1" 

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

221 PUBLIC_IPS[:] = [] 

222 

223 

224@_only_once 

225def _load_ips(suppress_exceptions: bool = True) -> None: 

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

227 

228 This function will only ever be called once. 

229 

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

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

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

233 """ 

234 

235 try: 

236 # first priority, use netifaces 

237 try: 

238 return _load_ips_netifaces() 

239 except ImportError: 

240 pass 

241 

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

243 

244 if os.name == "nt": 

245 try: 

246 return _load_ips_ipconfig() 

247 except (OSError, NoIPAddresses): 

248 pass 

249 else: 

250 try: 

251 return _load_ips_ip() 

252 except (OSError, NoIPAddresses): 

253 pass 

254 try: 

255 return _load_ips_ifconfig() 

256 except (OSError, NoIPAddresses): 

257 pass 

258 

259 # lowest priority, use gethostbyname 

260 

261 return _load_ips_gethostbyname() 

262 except Exception as e: 

263 if not suppress_exceptions: 

264 raise 

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

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

267 _load_ips_dumb() 

268 

269 

270@_requires_ips 

271def local_ips() -> list[str]: 

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

273 return LOCAL_IPS 

274 

275 

276@_requires_ips 

277def public_ips() -> list[str]: 

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

279 return PUBLIC_IPS 

280 

281 

282@_requires_ips 

283def localhost() -> str: 

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

285 return LOCALHOST 

286 

287 

288@_requires_ips 

289def is_local_ip(ip: str) -> bool: 

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

291 return ip in LOCAL_IPS 

292 

293 

294@_requires_ips 

295def is_public_ip(ip: str) -> bool: 

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

297 return ip in PUBLIC_IPS