fix

python/xoscar/tests/test_utils.py

@@ -163,7 +163,16 @@ def test_timer():
 
 
 def test_fix_zero_ip():
+ assert utils.is_v4_zero_ip("0.0.0.0:1234") == True
+ assert utils.is_v4_zero_ip("127.0.0.1:1234") == False
+ assert utils.is_v6_zero_ip(":::1234") == True
+ return utils.is_v6_zero_ip("0000:0000:0000:0000:0000:0000:0000:0000:1234") == True
+ return utils.is_v6_zero_ip("0:0:0:0:0:0:0:0:1234") == True
+ return utils.is_v6_zero_ip("0:0:0:0:0:1234") == True
+ assert utils.is_v6_zero_ip("2001:db8:3333:4444:5555:6666:7777:8888:1234") == False
+ assert utils.is_v6_zero_ip("127.0.0.1:1234") == False
  assert utils.fix_zero_ip("127.0.0.1:1234", "127.0.0.1:5678") == "127.0.0.1:1234"
  assert utils.fix_zero_ip("0.0.0.0:1234", "0.0.0.0:5678") == "0.0.0.0:1234"
  assert utils.fix_zero_ip("0.0.0.0:1234", "192.168.0.1:5678") == "192.168.0.1:1234"
  assert utils.fix_zero_ip("127.0.0.1:1234", "0.0.0.0:5678") == "127.0.0.1:1234"
+ assert utils.fix_zero_ip(":::1234", "2001:0db8:0001:0000:0000:0ab9:C0A8:0102:5678") == "2001:0db8:0001:0000:0000:0ab9:C0A8:0102:1234"

python/xoscar/utils.py

@@ -464,9 +464,19 @@ def is_linux():
  return sys.platform.startswith("linux")
 
 
-def is_zero_ip(ip: str) -> bool:
- return ip == "0.0.0.0" or ip == ""
-
+def is_v4_zero_ip(ip_port_addr: str) -> bool:
+ return ip_port_addr.startswith("0.0.0.0:")
+
+def is_v6_zero_ip(ip_port_addr: str) -> bool:
+ # tcp6 addr ":::123", ":: means all zero"
+ arr = ip_port_addr.split(":")
+ if len(arr) <= 2: # Not tcp6 or udp6
+ return False
+ for part in arr[0:-1]:
+ if part != "":
+ if int(part, 16) != 0:
+ return False
+ return True
 
 def fix_zero_ip(remote_addr: str, connect_addr: str) -> str:
  """
@@ -479,13 +489,12 @@ def fix_zero_ip(remote_addr: str, connect_addr: str) -> str:
  """
  if remote_addr == connect_addr:
  return remote_addr
- remote_arr = remote_addr.split(":")
- if not is_zero_ip(remote_arr[0]):
- # Remote server listen on non-zero ip
+ if not is_v4_zero_ip(remote_addr) and not is_v6_zero_ip(remote_addr):
+ # Remote server returns on non-zero ip
  return remote_addr
- connect_arr = connect_addr.split(":")
- if is_zero_ip(connect_arr[0]):
- # Connect to local server
+ if is_v4_zero_ip(connect_addr) or is_v6_zero_ip(connect_addr):
+ # Client connect to local server
  return remote_addr
- remote_arr[0] = connect_arr[0]
- return ":".join(remote_arr)
+ remote_port = remote_addr.split(":")[-1]
+ connect_ip = ":".join(connect_addr.split(":")[0:-1]) # Remote the port
+ return f"{connect_ip}:{remote_port}"