Merge pull request #132 from sepalani/cfg-ext-ip

Add ExternalIP config field

Author: sepalani

.github/workflows/python-ci.yml

@@ -69,4 +69,4 @@ jobs:
           pip install ruff
       - name: Lint with ruff
         run: |
-          ruff --output-format=github --exclude externals .
+          ruff check --output-format=github --exclude externals .

config.ini

@@ -4,6 +4,7 @@ DefaultKey = cert/server.key
 
 [OPN]
 IP = 0.0.0.0
+ExternalIP =
 Port = 8200
 Name = OpnServer
 MaxThread = 1
@@ -17,6 +18,7 @@ LogToWindow = ON
 
 [LMP]
 IP = 0.0.0.0
+ExternalIP =
 Port = 8201
 Name = LmpServer
 MaxThread = 1
@@ -30,6 +32,7 @@ LogToWindow = ON
 
 [FMP]
 IP = 0.0.0.0
+ExternalIP =
 Port = 8202
 Name = FmpServer
 MaxThread = 0
@@ -43,6 +46,7 @@ LogToWindow = ON
 
 [RFP]
 IP = 0.0.0.0
+ExternalIP =
 Port = 8203
 Name = RfpServer
 MaxThread = 1

mh/database.py

@@ -337,7 +337,7 @@ def __init__(self, name, server_type, gate_count=40, capacity=2000,
         self.name = name
         self.parent = None
         self.server_type = server_type
-        self.addr = addr
+        self.addr = addr  # public IP address
         self.port = port
         self.gates = [
             Gate("City Gate{}".format(i), self)

mh/pat.py

@@ -8,7 +8,7 @@
 import traceback
 from datetime import timedelta
 
-from other.utils import Logger, get_config, get_ip, hexdump, to_str
+from other.utils import Logger, get_config, get_external_ip, hexdump, to_str
 
 import mh.pat_item as pati
 import mh.server as server
@@ -583,7 +583,7 @@ def recvReqLmpConnect(self, packet_id, data, seq):
         TODO: I don't think it's related to LMP protocol.
         """
         config = get_config("LMP")
-        self.sendAnsLmpConnect(get_ip(config["IP"]), config["Port"], seq)
+        self.sendAnsLmpConnect(get_external_ip(config), config["Port"], seq)
 
     def sendAnsLmpConnect(self, address, port, seq):
         """AnsLmpConnect packet.
@@ -1071,7 +1071,7 @@ def recvReqFmpInfo(self, packet_id, data, seq):
         fields = pati.unpack_bytes(data, 4)
         server = self.session.join_server(index)
         config = get_config("FMP")
-        fmp_addr = get_ip(config["IP"])
+        fmp_addr = get_external_ip(config)
         fmp_port = config["Port"]
         fmp_data = pati.FmpData()
         fmp_data.server_address = pati.String(server.addr or fmp_addr)

mh/pat_item.py

@@ -8,9 +8,11 @@
 
 from collections import OrderedDict
 from mh.constants import pad
-from other.utils import to_bytearray, get_config, get_ip, GenericUnpacker
+from other.utils import to_bytearray, get_config, get_external_ip, \
+    GenericUnpacker
 from mh.database import Server, Gate, City
 
+
 class ItemType:
     Custom = 0
     Byte = 1
@@ -960,7 +962,7 @@ def get_fmp_servers(session, first_index, count):
     assert first_index > 0, "Invalid list index"
 
     config = get_config("FMP")
-    fmp_addr = get_ip(config["IP"])
+    fmp_addr = get_external_ip(config)
     fmp_port = config["Port"]
 
     data = b""

other/utils.py

@@ -23,6 +23,7 @@
     # Python 3
     basestring = str
     import configparser as ConfigParser
+    from typing import Any  # noqa: F401
 
 CONFIG_FILE = "config.ini"
 LOG_FOLDER = "logs"
@@ -221,6 +222,7 @@ def get_config(name, config_file=CONFIG_FILE):
     config.read(config_file)
     return {
         "IP": config.get(name, "IP"),
+        "ExternalIP": config.get(name, "ExternalIP"),
         "Port": config.getint(name, "Port"),
         "Name": config.get(name, "Name"),
         "MaxThread": config.getint(name, "MaxThread"),
@@ -239,19 +241,31 @@ def get_config(name, config_file=CONFIG_FILE):
 
 
 def get_default_ip():
+    # type: () -> str
     """Get the default IP address"""
     s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
     s.connect(("8.8.8.8", 80))
-    ip = s.getsockname()[0]
+    ip = s.getsockname()[0]  # type: str
     s.close()
     return ip
 
 
 def get_ip(ip):
+    # type: (str) -> str
     """Return the IP address that will be used."""
     return get_default_ip() if ip == "0.0.0.0" else ip
 
 
+def get_external_ip(config):
+    # type: (dict[str, Any]) -> str
+    """Return the IP address advertised by the server.
+
+    It's useful when the public IP address can't easily be retrieved.
+    For instance, when behind a NAT or some cloud infrastructures.
+    """
+    return config["ExternalIP"] or get_ip(config["IP"])
+
+
 def argparse_from_config(config):
     """Argument parser from config."""
     import argparse