Fix/port normalization overrides host port (#793)

Author: ruvolof

CHANGELOG.md

@@ -14,6 +14,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+-   Fixed port normalization to allow host port and bind port to be different.
+
 ## [6.1.3] - 2025-07-11
 
 -   Updated Starkiller to v3.0.1

empire/server/core/listener_service.py

@@ -1,6 +1,7 @@
 import copy
 import hashlib
 import logging
+import re
 import typing
 from typing import Any
 
@@ -236,58 +237,43 @@ def _validate_listener_options(
         return template_instance, None
 
     @staticmethod
-    def _normalize_listener_options(instance) -> None:  # noqa: PLR0912 PLR0915
+    def _normalize_listener_options(instance) -> None:  # noqa: PLR0912
         """
         This is adapted from the old set_listener_option which does some coercions on the http fields.
         """
         for option_name, option_meta in instance.options.items():
             value = option_meta["Value"]
             # parse and auto-set some host parameters
             if option_name == "Host":
-                if not value.startswith("http"):
-                    parts = value.split(":")
-                    # if there's a current ssl cert path set, assume this is https
-                    if ("CertPath" in instance.options) and (
-                        instance.options["CertPath"]["Value"] != ""
+                host_rexp = r"^(https?)?:?/?/?([^:]+):?(\d+)?$"
+                matches = re.match(host_rexp, value)
+                try:
+                    protocol, host, port = matches.groups()
+                except AttributeError:
+                    log.error(f"Unable to parse Host value: {value}")
+                    protocol = None
+                    host = value
+                    port = None
+                if not protocol:
+                    if (
+                        "CertPath" in instance.options
+                        and instance.options["CertPath"]["Value"] != ""
                     ):
                         protocol = "https"
-                        default_port = 443
                     else:
                         protocol = "http"
-                        default_port = 80
-
-                elif value.startswith("https"):
-                    value = value.split("//")[1]
-                    parts = value.split(":")
-                    protocol = "https"
-                    default_port = 443
-
-                elif value.startswith("http"):
-                    value = value.split("//")[1]
-                    parts = value.split(":")
-                    protocol = "http"
-                    default_port = 80
-
-                ##################################################################################################################################
-                # Added functionality to Port
-                # Unsure if this section is needed
-                if len(parts) != 1 and parts[-1].isdigit():
-                    # if a port is specified with http://host:port
-                    instance.options["Host"]["Value"] = f"{protocol}://{value}"
-                    if instance.options["Port"]["Value"] == "":
-                        instance.options["Port"]["Value"] = parts[-1]
-                elif instance.options["Port"]["Value"] != "":
-                    # otherwise, check if the port value was manually set
-                    instance.options["Host"]["Value"] = "{}://{}:{}".format(
-                        protocol,
-                        value,
-                        instance.options["Port"]["Value"],
-                    )
-                else:
-                    # otherwise use default port
-                    instance.options["Host"]["Value"] = f"{protocol}://{value}"
-                    if instance.options["Port"]["Value"] == "":
+                default_port = 443 if protocol == "https" else 80
+                try:
+                    int(instance.options["Port"]["Value"])
+                except ValueError:
+                    if port:
+                        instance.options["Port"]["Value"] = port
+                    else:
                         instance.options["Port"]["Value"] = default_port
+                if port:
+                    instance.options["Host"]["Value"] = f"{protocol}://{host}:{port}"
+                else:
+                    instance.options["Host"]["Value"] = f"{protocol}://{host}"
 
             elif option_name == "CertPath" and value != "":
                 instance.options[option_name]["Value"] = value
@@ -301,23 +287,11 @@ def _normalize_listener_options(instance) -> None:  # noqa: PLR0912 PLR0915
             elif option_name == "Port":
                 instance.options[option_name]["Value"] = value
                 # Check if Port is set and add it to host
-                parts = instance.options["Host"]["Value"]
-                if parts.startswith("https"):
-                    address = parts[8:]
-                    address = "".join(address.split(":")[0])
-                    protocol = "https"
-                    instance.options["Host"]["Value"] = "{}://{}:{}".format(
-                        protocol,
-                        address,
-                        instance.options["Port"]["Value"],
-                    )
-                elif parts.startswith("http"):
-                    address = parts[7:]
-                    address = "".join(address.split(":")[0])
-                    protocol = "http"
-                    instance.options["Host"]["Value"] = "{}://{}:{}".format(
-                        protocol,
-                        address,
+                try:
+                    protocol, host, port = instance.options["Host"]["Value"].split(":")
+                except ValueError:
+                    instance.options["Host"]["Value"] = "{}:{}".format(
+                        instance.options["Host"]["Value"],
                         instance.options["Port"]["Value"],
                     )
 

empire/test/test_listener_api.py

@@ -122,6 +122,84 @@ def test_create_listener_template_not_found(client, admin_auth_header):
     assert response.json()["detail"] == "Listener Template qwerty not found"
 
 
+def test_create_listener_normalization_adds_protocol_and_default_port(
+    client, admin_auth_header
+):
+    base_listener = get_base_listener()
+    base_listener["name"] = "temp123"
+    base_listener["options"]["Host"] = "localhost"
+    base_listener["options"]["Port"] = None
+
+    response = client.post(
+        "/api/v2/listeners/", headers=admin_auth_header, json=base_listener
+    )
+    assert response.status_code == status.HTTP_201_CREATED
+    assert response.json()["options"]["Host"] == "http://localhost:80"
+    assert response.json()["options"]["Port"] == "80"
+
+    client.delete(
+        f"/api/v2/listeners/{response.json()['id']}", headers=admin_auth_header
+    )
+
+
+def test_create_listener_normalization_adds_port_to_host(client, admin_auth_header):
+    base_listener = get_base_listener()
+    base_listener["name"] = "temp123"
+    base_listener["options"]["Host"] = "http://localhost"
+    base_listener["options"]["Port"] = "1234"
+
+    response = client.post(
+        "/api/v2/listeners/", headers=admin_auth_header, json=base_listener
+    )
+    assert response.status_code == status.HTTP_201_CREATED
+    assert response.json()["options"]["Host"] == "http://localhost:1234"
+    assert response.json()["options"]["Port"] == "1234"
+
+    client.delete(
+        f"/api/v2/listeners/{response.json()['id']}", headers=admin_auth_header
+    )
+
+
+def test_create_listener_normalization_preserves_user_defined_ports(
+    client, admin_auth_header
+):
+    base_listener = get_base_listener()
+    base_listener["name"] = "temp123"
+    base_listener["options"]["Host"] = "http://localhost:443"
+    base_listener["options"]["Port"] = "1234"
+
+    response = client.post(
+        "/api/v2/listeners/", headers=admin_auth_header, json=base_listener
+    )
+    assert response.status_code == status.HTTP_201_CREATED
+    assert response.json()["options"]["Host"] == "http://localhost:443"
+    assert response.json()["options"]["Port"] == "1234"
+
+    client.delete(
+        f"/api/v2/listeners/{response.json()['id']}", headers=admin_auth_header
+    )
+
+
+def test_create_listener_normalization_sets_host_port_as_bind_port(
+    client, admin_auth_header
+):
+    base_listener = get_base_listener()
+    base_listener["name"] = "temp123"
+    base_listener["options"]["Host"] = "http://localhost:443"
+    base_listener["options"]["Port"] = None
+
+    response = client.post(
+        "/api/v2/listeners/", headers=admin_auth_header, json=base_listener
+    )
+    assert response.status_code == status.HTTP_201_CREATED
+    assert response.json()["options"]["Host"] == "http://localhost:443"
+    assert response.json()["options"]["Port"] == "443"
+
+    client.delete(
+        f"/api/v2/listeners/{response.json()['id']}", headers=admin_auth_header
+    )
+
+
 def test_create_listener(client, admin_auth_header):
     base_listener = get_base_listener()
     base_listener["name"] = "temp123"