feat: report what listens, and which Windows service it is

One script: Get-NetTCPConnection -State Listen, Get-NetUDPEndpoint, the
processes holding them and Win32_Service's PIDs. A process running one
service names it; a shared svchost runs several, and which of them holds
the socket Windows does not say, so it names none. http.sys sockets
(WinRM, IIS, SMB) belong to the System process.

The fixture is synthetic: no Windows host was at hand. tools/harvest.py
records the real output. For netOrk#673.
This commit is contained in:
Christian Manivong committed 2026-10-07 07:20:27 +02:00
1 parent 1ce42ef099
commit a6814ad92e
7 files changed
+185 -2

No files matched your search

+73
View File
@@ -362,6 +362,79 @@ class TestGetServices:
]
class TestGetListeningSockets:
"""netOrk #673: what listens, and which Windows service it is."""
def _sockets(self, answer: str | None = None) -> dict:
answer = answer or _fixture("listening.json")
reading = _driver(_PS_LISTENING=answer).get_listening_sockets()
return {(s["proto"], s["port"], s["address"]): s for s in reading["sockets"]}
def test_read_as_an_administrator(self):
reading = _driver(_PS_LISTENING=_fixture("listening.json")).get_listening_sockets()
assert reading["attributed"] is True
assert len(reading["sockets"]) == 11
def test_a_socket_comes_with_its_process_and_service(self):
assert self._sockets()[("tcp", 1433, "0.0.0.0")] == {
"proto": "tcp",
"address": "0.0.0.0",
"port": 1433,
"interface": None,
"process": "sqlservr",
"pid": 4321,
"unit": "MSSQL$SQLEXPRESS",
"container_id": None,
}
def test_a_svchost_of_its_own_names_its_service(self):
assert self._sockets()[("tcp", 3389, "0.0.0.0")]["unit"] == "TermService"
def test_a_shared_svchost_names_none(self):
"""RpcSs and RpcEptMapper share a process: the socket is one of theirs,
and which one Windows does not say."""
rpc = self._sockets()[("tcp", 135, "0.0.0.0")]
assert (rpc["process"], rpc["unit"]) == ("svchost", None)
def test_http_sys_listens_as_the_system_process(self):
winrm = self._sockets()[("tcp", 5985, "0.0.0.0")]
assert (winrm["process"], winrm["pid"], winrm["unit"]) == ("System", 4, None)
def test_the_scope_of_a_link_local_address_is_its_interface(self):
assert self._sockets()[("tcp", 1433, "fe80::1c2a:3b4c:5d6e:7f80")]["interface"] == "12"
def test_a_socket_without_a_process_is_kept(self):
mdns = self._sockets()[("udp", 5353, "0.0.0.0")]
assert (mdns["process"], mdns["pid"], mdns["unit"]) == (None, None, None)
def test_one_socket_arrives_as_a_bare_object(self):
"""ConvertTo-Json unwraps a one-element array."""
answer = json.dumps(
{
"tcp": {"address": "0.0.0.0", "port": 3389, "pid": 1180},
"udp": [],
"processes": {"pid": 1180, "name": "svchost"},
"services": {"pid": 1180, "name": "TermService"},
}
)
assert self._sockets(answer)[("tcp", 3389, "0.0.0.0")]["unit"] == "TermService"
def test_sorted_by_protocol_port_and_address(self):
reading = _driver(_PS_LISTENING=_fixture("listening.json")).get_listening_sockets()
keys = [(s["proto"], s["port"], s["address"]) for s in reading["sockets"]]
assert keys == sorted(keys)
def test_the_script_reads_and_changes_nothing(self):
for verb in ("Set-", "Stop-", "Start-", "Remove-", "New-"):
assert verb not in mod._PS_LISTENING
class TestManageService:
@pytest.mark.parametrize(
("action", "command"),