test_url.py 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471
  1. # -*- coding: utf-8 -*-
  2. #
  3. import os
  4. import unittest
  5. from websocket._url import (
  6. _is_address_in_network,
  7. _is_no_proxy_host,
  8. get_proxy_info,
  9. parse_url,
  10. )
  11. from websocket._exceptions import WebSocketProxyException
  12. """
  13. test_url.py
  14. websocket - WebSocket client library for Python
  15. Copyright 2025 engn33r
  16. Licensed under the Apache License, Version 2.0 (the "License");
  17. you may not use this file except in compliance with the License.
  18. You may obtain a copy of the License at
  19. http://www.apache.org/licenses/LICENSE-2.0
  20. Unless required by applicable law or agreed to in writing, software
  21. distributed under the License is distributed on an "AS IS" BASIS,
  22. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  23. See the License for the specific language governing permissions and
  24. limitations under the License.
  25. """
  26. class UrlTest(unittest.TestCase):
  27. def test_address_in_network(self):
  28. self.assertTrue(_is_address_in_network("127.0.0.1", "127.0.0.0/8"))
  29. self.assertTrue(_is_address_in_network("127.1.0.1", "127.0.0.0/8"))
  30. self.assertFalse(_is_address_in_network("127.1.0.1", "127.0.0.0/24"))
  31. self.assertTrue(_is_address_in_network("2001:db8::1", "2001:db8::/64"))
  32. self.assertFalse(_is_address_in_network("2001:db8:1::1", "2001:db8::/64"))
  33. def test_parse_url(self):
  34. p = parse_url("ws://www.example.com/r")
  35. self.assertEqual(p[0], "www.example.com")
  36. self.assertEqual(p[1], 80)
  37. self.assertEqual(p[2], "/r")
  38. self.assertEqual(p[3], False)
  39. p = parse_url("ws://www.example.com/r/")
  40. self.assertEqual(p[0], "www.example.com")
  41. self.assertEqual(p[1], 80)
  42. self.assertEqual(p[2], "/r/")
  43. self.assertEqual(p[3], False)
  44. p = parse_url("ws://www.example.com/")
  45. self.assertEqual(p[0], "www.example.com")
  46. self.assertEqual(p[1], 80)
  47. self.assertEqual(p[2], "/")
  48. self.assertEqual(p[3], False)
  49. p = parse_url("ws://www.example.com")
  50. self.assertEqual(p[0], "www.example.com")
  51. self.assertEqual(p[1], 80)
  52. self.assertEqual(p[2], "/")
  53. self.assertEqual(p[3], False)
  54. p = parse_url("ws://www.example.com:8080/r")
  55. self.assertEqual(p[0], "www.example.com")
  56. self.assertEqual(p[1], 8080)
  57. self.assertEqual(p[2], "/r")
  58. self.assertEqual(p[3], False)
  59. p = parse_url("ws://www.example.com:8080/")
  60. self.assertEqual(p[0], "www.example.com")
  61. self.assertEqual(p[1], 8080)
  62. self.assertEqual(p[2], "/")
  63. self.assertEqual(p[3], False)
  64. p = parse_url("ws://www.example.com:8080")
  65. self.assertEqual(p[0], "www.example.com")
  66. self.assertEqual(p[1], 8080)
  67. self.assertEqual(p[2], "/")
  68. self.assertEqual(p[3], False)
  69. p = parse_url("wss://www.example.com:8080/r")
  70. self.assertEqual(p[0], "www.example.com")
  71. self.assertEqual(p[1], 8080)
  72. self.assertEqual(p[2], "/r")
  73. self.assertEqual(p[3], True)
  74. p = parse_url("wss://www.example.com:8080/r?key=value")
  75. self.assertEqual(p[0], "www.example.com")
  76. self.assertEqual(p[1], 8080)
  77. self.assertEqual(p[2], "/r?key=value")
  78. self.assertEqual(p[3], True)
  79. self.assertRaises(ValueError, parse_url, "http://www.example.com/r")
  80. p = parse_url("ws://[2a03:4000:123:83::3]/r")
  81. self.assertEqual(p[0], "2a03:4000:123:83::3")
  82. self.assertEqual(p[1], 80)
  83. self.assertEqual(p[2], "/r")
  84. self.assertEqual(p[3], False)
  85. p = parse_url("ws://[2a03:4000:123:83::3]:8080/r")
  86. self.assertEqual(p[0], "2a03:4000:123:83::3")
  87. self.assertEqual(p[1], 8080)
  88. self.assertEqual(p[2], "/r")
  89. self.assertEqual(p[3], False)
  90. p = parse_url("wss://[2a03:4000:123:83::3]/r")
  91. self.assertEqual(p[0], "2a03:4000:123:83::3")
  92. self.assertEqual(p[1], 443)
  93. self.assertEqual(p[2], "/r")
  94. self.assertEqual(p[3], True)
  95. p = parse_url("wss://[2a03:4000:123:83::3]:8080/r")
  96. self.assertEqual(p[0], "2a03:4000:123:83::3")
  97. self.assertEqual(p[1], 8080)
  98. self.assertEqual(p[2], "/r")
  99. self.assertEqual(p[3], True)
  100. class IsNoProxyHostTest(unittest.TestCase):
  101. def setUp(self):
  102. self.no_proxy = os.environ.get("no_proxy", None)
  103. if "no_proxy" in os.environ:
  104. del os.environ["no_proxy"]
  105. def tearDown(self):
  106. if self.no_proxy:
  107. os.environ["no_proxy"] = self.no_proxy
  108. elif "no_proxy" in os.environ:
  109. del os.environ["no_proxy"]
  110. def test_match_all(self):
  111. self.assertTrue(_is_no_proxy_host("any.websocket.org", ["*"]))
  112. self.assertTrue(_is_no_proxy_host("192.168.0.1", ["*"]))
  113. self.assertFalse(_is_no_proxy_host("192.168.0.1", ["192.168.1.1"]))
  114. self.assertFalse(
  115. _is_no_proxy_host("any.websocket.org", ["other.websocket.org"])
  116. )
  117. self.assertTrue(
  118. _is_no_proxy_host("any.websocket.org", ["other.websocket.org", "*"])
  119. )
  120. os.environ["no_proxy"] = "*"
  121. self.assertTrue(_is_no_proxy_host("any.websocket.org", None))
  122. self.assertTrue(_is_no_proxy_host("192.168.0.1", None))
  123. os.environ["no_proxy"] = "other.websocket.org, *"
  124. self.assertTrue(_is_no_proxy_host("any.websocket.org", None))
  125. def test_ip_address(self):
  126. self.assertTrue(_is_no_proxy_host("127.0.0.1", ["127.0.0.1"]))
  127. self.assertFalse(_is_no_proxy_host("127.0.0.2", ["127.0.0.1"]))
  128. self.assertTrue(
  129. _is_no_proxy_host("127.0.0.1", ["other.websocket.org", "127.0.0.1"])
  130. )
  131. self.assertFalse(
  132. _is_no_proxy_host("127.0.0.2", ["other.websocket.org", "127.0.0.1"])
  133. )
  134. os.environ["no_proxy"] = "127.0.0.1"
  135. self.assertTrue(_is_no_proxy_host("127.0.0.1", None))
  136. self.assertFalse(_is_no_proxy_host("127.0.0.2", None))
  137. os.environ["no_proxy"] = "other.websocket.org, 127.0.0.1"
  138. self.assertTrue(_is_no_proxy_host("127.0.0.1", None))
  139. self.assertFalse(_is_no_proxy_host("127.0.0.2", None))
  140. def test_ip_address_in_range(self):
  141. self.assertTrue(_is_no_proxy_host("127.0.0.1", ["127.0.0.0/8"]))
  142. self.assertTrue(_is_no_proxy_host("127.0.0.2", ["127.0.0.0/8"]))
  143. self.assertFalse(_is_no_proxy_host("127.1.0.1", ["127.0.0.0/24"]))
  144. self.assertTrue(_is_no_proxy_host("2001:db8::1", ["2001:db8::/64"]))
  145. self.assertFalse(_is_no_proxy_host("2001:db8:1::1", ["2001:db8::/64"]))
  146. os.environ["no_proxy"] = "127.0.0.0/8,2001:db8::/64"
  147. self.assertTrue(_is_no_proxy_host("127.0.0.1", None))
  148. self.assertTrue(_is_no_proxy_host("127.0.0.2", None))
  149. self.assertTrue(_is_no_proxy_host("2001:db8::1", None))
  150. self.assertFalse(_is_no_proxy_host("2001:db8:1::1", None))
  151. os.environ["no_proxy"] = "127.0.0.0/24,2001:db8::/64"
  152. self.assertFalse(_is_no_proxy_host("127.1.0.1", None))
  153. self.assertFalse(_is_no_proxy_host("2001:db8:1::1", None))
  154. def test_hostname_match(self):
  155. self.assertTrue(_is_no_proxy_host("my.websocket.org", ["my.websocket.org"]))
  156. self.assertTrue(
  157. _is_no_proxy_host(
  158. "my.websocket.org", ["other.websocket.org", "my.websocket.org"]
  159. )
  160. )
  161. self.assertFalse(_is_no_proxy_host("my.websocket.org", ["other.websocket.org"]))
  162. os.environ["no_proxy"] = "my.websocket.org"
  163. self.assertTrue(_is_no_proxy_host("my.websocket.org", None))
  164. self.assertFalse(_is_no_proxy_host("other.websocket.org", None))
  165. os.environ["no_proxy"] = "other.websocket.org, my.websocket.org"
  166. self.assertTrue(_is_no_proxy_host("my.websocket.org", None))
  167. def test_hostname_match_domain(self):
  168. self.assertTrue(_is_no_proxy_host("any.websocket.org", [".websocket.org"]))
  169. self.assertTrue(_is_no_proxy_host("my.other.websocket.org", [".websocket.org"]))
  170. self.assertTrue(
  171. _is_no_proxy_host(
  172. "any.websocket.org", ["my.websocket.org", ".websocket.org"]
  173. )
  174. )
  175. self.assertFalse(_is_no_proxy_host("any.websocket.com", [".websocket.org"]))
  176. os.environ["no_proxy"] = ".websocket.org"
  177. self.assertTrue(_is_no_proxy_host("any.websocket.org", None))
  178. self.assertTrue(_is_no_proxy_host("my.other.websocket.org", None))
  179. self.assertFalse(_is_no_proxy_host("any.websocket.com", None))
  180. os.environ["no_proxy"] = "my.websocket.org, .websocket.org"
  181. self.assertTrue(_is_no_proxy_host("any.websocket.org", None))
  182. class ProxyInfoTest(unittest.TestCase):
  183. def setUp(self):
  184. self.http_proxy = os.environ.get("http_proxy", None)
  185. self.https_proxy = os.environ.get("https_proxy", None)
  186. self.no_proxy = os.environ.get("no_proxy", None)
  187. if "http_proxy" in os.environ:
  188. del os.environ["http_proxy"]
  189. if "https_proxy" in os.environ:
  190. del os.environ["https_proxy"]
  191. if "no_proxy" in os.environ:
  192. del os.environ["no_proxy"]
  193. def tearDown(self):
  194. if self.http_proxy:
  195. os.environ["http_proxy"] = self.http_proxy
  196. elif "http_proxy" in os.environ:
  197. del os.environ["http_proxy"]
  198. if self.https_proxy:
  199. os.environ["https_proxy"] = self.https_proxy
  200. elif "https_proxy" in os.environ:
  201. del os.environ["https_proxy"]
  202. if self.no_proxy:
  203. os.environ["no_proxy"] = self.no_proxy
  204. elif "no_proxy" in os.environ:
  205. del os.environ["no_proxy"]
  206. def test_proxy_from_args(self):
  207. self.assertRaises(
  208. WebSocketProxyException,
  209. get_proxy_info,
  210. "echo.websocket.events",
  211. False,
  212. proxy_host="localhost",
  213. )
  214. self.assertEqual(
  215. get_proxy_info(
  216. "echo.websocket.events", False, proxy_host="localhost", proxy_port=3128
  217. ),
  218. ("localhost", 3128, None),
  219. )
  220. self.assertEqual(
  221. get_proxy_info(
  222. "echo.websocket.events", True, proxy_host="localhost", proxy_port=3128
  223. ),
  224. ("localhost", 3128, None),
  225. )
  226. self.assertEqual(
  227. get_proxy_info(
  228. "echo.websocket.events",
  229. False,
  230. proxy_host="localhost",
  231. proxy_port=9001,
  232. proxy_auth=("a", "b"),
  233. ),
  234. ("localhost", 9001, ("a", "b")),
  235. )
  236. self.assertEqual(
  237. get_proxy_info(
  238. "echo.websocket.events",
  239. False,
  240. proxy_host="localhost",
  241. proxy_port=3128,
  242. proxy_auth=("a", "b"),
  243. ),
  244. ("localhost", 3128, ("a", "b")),
  245. )
  246. self.assertEqual(
  247. get_proxy_info(
  248. "echo.websocket.events",
  249. True,
  250. proxy_host="localhost",
  251. proxy_port=8765,
  252. proxy_auth=("a", "b"),
  253. ),
  254. ("localhost", 8765, ("a", "b")),
  255. )
  256. self.assertEqual(
  257. get_proxy_info(
  258. "echo.websocket.events",
  259. True,
  260. proxy_host="localhost",
  261. proxy_port=3128,
  262. proxy_auth=("a", "b"),
  263. ),
  264. ("localhost", 3128, ("a", "b")),
  265. )
  266. self.assertEqual(
  267. get_proxy_info(
  268. "echo.websocket.events",
  269. True,
  270. proxy_host="localhost",
  271. proxy_port=3128,
  272. no_proxy=["example.com"],
  273. proxy_auth=("a", "b"),
  274. ),
  275. ("localhost", 3128, ("a", "b")),
  276. )
  277. self.assertEqual(
  278. get_proxy_info(
  279. "echo.websocket.events",
  280. True,
  281. proxy_host="localhost",
  282. proxy_port=3128,
  283. no_proxy=["echo.websocket.events"],
  284. proxy_auth=("a", "b"),
  285. ),
  286. (None, 0, None),
  287. )
  288. self.assertEqual(
  289. get_proxy_info(
  290. "echo.websocket.events",
  291. True,
  292. proxy_host="localhost",
  293. proxy_port=3128,
  294. no_proxy=[".websocket.events"],
  295. ),
  296. (None, 0, None),
  297. )
  298. def test_proxy_from_env(self):
  299. os.environ["http_proxy"] = "http://localhost/"
  300. self.assertEqual(
  301. get_proxy_info("echo.websocket.events", False), ("localhost", None, None)
  302. )
  303. os.environ["http_proxy"] = "http://localhost:3128/"
  304. self.assertEqual(
  305. get_proxy_info("echo.websocket.events", False), ("localhost", 3128, None)
  306. )
  307. os.environ["http_proxy"] = "http://localhost/"
  308. os.environ["https_proxy"] = "http://localhost2/"
  309. self.assertEqual(
  310. get_proxy_info("echo.websocket.events", False), ("localhost", None, None)
  311. )
  312. os.environ["http_proxy"] = "http://localhost:3128/"
  313. os.environ["https_proxy"] = "http://localhost2:3128/"
  314. self.assertEqual(
  315. get_proxy_info("echo.websocket.events", False), ("localhost", 3128, None)
  316. )
  317. os.environ["http_proxy"] = "http://localhost/"
  318. os.environ["https_proxy"] = "http://localhost2/"
  319. self.assertEqual(
  320. get_proxy_info("echo.websocket.events", True), ("localhost2", None, None)
  321. )
  322. os.environ["http_proxy"] = "http://localhost:3128/"
  323. os.environ["https_proxy"] = "http://localhost2:3128/"
  324. self.assertEqual(
  325. get_proxy_info("echo.websocket.events", True), ("localhost2", 3128, None)
  326. )
  327. os.environ["http_proxy"] = ""
  328. os.environ["https_proxy"] = "http://localhost2/"
  329. self.assertEqual(
  330. get_proxy_info("echo.websocket.events", True), ("localhost2", None, None)
  331. )
  332. self.assertEqual(
  333. get_proxy_info("echo.websocket.events", False), (None, 0, None)
  334. )
  335. os.environ["http_proxy"] = ""
  336. os.environ["https_proxy"] = "http://localhost2:3128/"
  337. self.assertEqual(
  338. get_proxy_info("echo.websocket.events", True), ("localhost2", 3128, None)
  339. )
  340. self.assertEqual(
  341. get_proxy_info("echo.websocket.events", False), (None, 0, None)
  342. )
  343. os.environ["http_proxy"] = "http://localhost/"
  344. os.environ["https_proxy"] = ""
  345. self.assertEqual(get_proxy_info("echo.websocket.events", True), (None, 0, None))
  346. self.assertEqual(
  347. get_proxy_info("echo.websocket.events", False), ("localhost", None, None)
  348. )
  349. os.environ["http_proxy"] = "http://localhost:3128/"
  350. os.environ["https_proxy"] = ""
  351. self.assertEqual(get_proxy_info("echo.websocket.events", True), (None, 0, None))
  352. self.assertEqual(
  353. get_proxy_info("echo.websocket.events", False), ("localhost", 3128, None)
  354. )
  355. os.environ["http_proxy"] = "http://a:b@localhost/"
  356. self.assertEqual(
  357. get_proxy_info("echo.websocket.events", False),
  358. ("localhost", None, ("a", "b")),
  359. )
  360. os.environ["http_proxy"] = "http://a:b@localhost:3128/"
  361. self.assertEqual(
  362. get_proxy_info("echo.websocket.events", False),
  363. ("localhost", 3128, ("a", "b")),
  364. )
  365. os.environ["http_proxy"] = "http://a:b@localhost/"
  366. os.environ["https_proxy"] = "http://a:b@localhost2/"
  367. self.assertEqual(
  368. get_proxy_info("echo.websocket.events", False),
  369. ("localhost", None, ("a", "b")),
  370. )
  371. os.environ["http_proxy"] = "http://a:b@localhost:3128/"
  372. os.environ["https_proxy"] = "http://a:b@localhost2:3128/"
  373. self.assertEqual(
  374. get_proxy_info("echo.websocket.events", False),
  375. ("localhost", 3128, ("a", "b")),
  376. )
  377. os.environ["http_proxy"] = "http://a:b@localhost/"
  378. os.environ["https_proxy"] = "http://a:b@localhost2/"
  379. self.assertEqual(
  380. get_proxy_info("echo.websocket.events", True),
  381. ("localhost2", None, ("a", "b")),
  382. )
  383. os.environ["http_proxy"] = "http://a:b@localhost:3128/"
  384. os.environ["https_proxy"] = "http://a:b@localhost2:3128/"
  385. self.assertEqual(
  386. get_proxy_info("echo.websocket.events", True),
  387. ("localhost2", 3128, ("a", "b")),
  388. )
  389. os.environ["http_proxy"] = (
  390. "http://john%40example.com:P%40SSWORD@localhost:3128/"
  391. )
  392. os.environ["https_proxy"] = (
  393. "http://john%40example.com:P%40SSWORD@localhost2:3128/"
  394. )
  395. self.assertEqual(
  396. get_proxy_info("echo.websocket.events", True),
  397. ("localhost2", 3128, ("john@example.com", "P@SSWORD")),
  398. )
  399. os.environ["http_proxy"] = "http://a:b@localhost/"
  400. os.environ["https_proxy"] = "http://a:b@localhost2/"
  401. os.environ["no_proxy"] = "example1.com,example2.com"
  402. self.assertEqual(
  403. get_proxy_info("example.1.com", True), ("localhost2", None, ("a", "b"))
  404. )
  405. os.environ["http_proxy"] = "http://a:b@localhost:3128/"
  406. os.environ["https_proxy"] = "http://a:b@localhost2:3128/"
  407. os.environ["no_proxy"] = "example1.com,example2.com, echo.websocket.events"
  408. self.assertEqual(get_proxy_info("echo.websocket.events", True), (None, 0, None))
  409. os.environ["http_proxy"] = "http://a:b@localhost:3128/"
  410. os.environ["https_proxy"] = "http://a:b@localhost2:3128/"
  411. os.environ["no_proxy"] = "example1.com,example2.com, .websocket.events"
  412. self.assertEqual(get_proxy_info("echo.websocket.events", True), (None, 0, None))
  413. os.environ["http_proxy"] = "http://a:b@localhost:3128/"
  414. os.environ["https_proxy"] = "http://a:b@localhost2:3128/"
  415. os.environ["no_proxy"] = "127.0.0.0/8, 192.168.0.0/16"
  416. self.assertEqual(get_proxy_info("127.0.0.1", False), (None, 0, None))
  417. self.assertEqual(get_proxy_info("192.168.1.1", False), (None, 0, None))
  418. if __name__ == "__main__":
  419. unittest.main()