feat: read what listens on which address, and which service it is, once for every driver
Whether a service is reachable from outside its host is decided by the address it listens on: 0.0.0.0:5432 is, 127.0.0.1:5432 is not. Reading that is the same on every Linux host, so the command and its parse live here once and a driver only carries the command across (ListeningSocketsMixin, _run_listening_sockets_command). One framed round trip: ss -lntup for every listening TCP and bound UDP socket, then /proc/<pid>/cgroup for each process holding one, which names the systemd service (v2, nested slices, v1's name=systemd hierarchy) or the container (docker-<id>.scope, /docker/<id>) it runs in. - Root: only root sees every process. The script goes as one sh -c argument, so a sudo -n prefix covers all of it; when that brings no report back the reading runs again unprivileged and says it is not attributed. - No -H: iproute2 before 4.10 fails on it, which would read as nothing listening. The header is skipped instead. - A host without ss raises ListeningSocketsUnavailable; a report cut short or a failing ss raises ValueError. - The reading is raw: docker-proxy shows up as docker.service, loopback as loopback. What counts as reachable is the consumer's call. 2.4.0. For netOrk#658.
This commit is contained in:
@@ -0,0 +1,274 @@
|
||||
"""get_listening_sockets: what listens on which address, and which service it is.
|
||||
|
||||
Whether a service is reachable from outside its host is decided by what it
|
||||
listens on -- ``0.0.0.0:5432`` is, ``127.0.0.1:5432`` is not. Reading that is
|
||||
the same on every Linux host: ``ss`` for the sockets, ``/proc/<pid>/cgroup``
|
||||
for the systemd unit or container a process belongs to. So both live here once,
|
||||
and a driver only carries the command across (#658 in netOrk).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import shutil
|
||||
import subprocess
|
||||
|
||||
import pytest
|
||||
|
||||
from napalm_device_types import OSDriver
|
||||
from napalm_device_types.listening import (
|
||||
LISTENING_SOCKETS_COMMAND,
|
||||
ListeningSocketsMixin,
|
||||
ListeningSocketsUnavailable,
|
||||
parse_listening_sockets,
|
||||
)
|
||||
|
||||
CID = "4f1c" + "0" * 60
|
||||
SS = """Netid State Recv-Q Send-Q Local Address:Port Peer Address:PortProcess
|
||||
udp UNCONN 0 0 127.0.0.53%lo:53 0.0.0.0:* users:(("systemd-resolve",pid=612,fd=13))
|
||||
udp UNCONN 0 0 0.0.0.0%ens18:68 0.0.0.0:* users:(("systemd-network",pid=590,fd=22))
|
||||
tcp LISTEN 0 4096 0.0.0.0:5432 0.0.0.0:* users:(("postgres",pid=812,fd=6))
|
||||
tcp LISTEN 0 128 [::]:22 [::]:* users:(("sshd",pid=700,fd=4))
|
||||
tcp LISTEN 0 511 *:80 *:* users:(("nginx",pid=901,fd=6),("nginx",pid=900,fd=6))
|
||||
tcp LISTEN 0 4096 [::ffff:127.0.0.1]:8125 *:* users:(("statsd",pid=950,fd=3))
|
||||
tcp LISTEN 0 4096 0.0.0.0:8080 0.0.0.0:* users:(("docker-proxy",pid=1200,fd=4))
|
||||
tcp LISTEN 0 4096 0.0.0.0:9100 0.0.0.0:* users:(("node_exporter",pid=1300,fd=3))
|
||||
tcp LISTEN 0 64 0.0.0.0:2049 0.0.0.0:*
|
||||
"""
|
||||
CGROUPS = f"""612 0::/system.slice/systemd-resolved.service
|
||||
590 0::/system.slice/systemd-networkd.service
|
||||
812 0::/system.slice/system-postgresql.slice/postgresql@16-main.service
|
||||
700 0::/system.slice/ssh.service
|
||||
900 0::/system.slice/nginx.service
|
||||
901 0::/system.slice/nginx.service
|
||||
950 0::/user.slice/user-1000.slice/session-3.scope
|
||||
1200 0::/system.slice/docker.service
|
||||
1300 0::/system.slice/docker-{CID}.scope
|
||||
"""
|
||||
|
||||
|
||||
def _wire(ss: str = SS, cgroups: str = CGROUPS, *, rc: int = 0, noise: str = "") -> str:
|
||||
"""The report as the command prints it, framed."""
|
||||
return f"{noise}SOCK_BEGIN\n[ss]\n{ss}__SS_RC={rc}\n[cgroups]\n{cgroups}SOCK_END\n"
|
||||
|
||||
|
||||
def _by_port(sockets: list) -> dict:
|
||||
return {(s["proto"], s["port"], s["address"]): s for s in sockets}
|
||||
|
||||
|
||||
class TestParsing:
|
||||
def test_a_socket_comes_with_its_process_and_unit(self):
|
||||
sockets = _by_port(parse_listening_sockets(_wire()))
|
||||
|
||||
assert sockets[("tcp", 5432, "0.0.0.0")] == {
|
||||
"proto": "tcp",
|
||||
"address": "0.0.0.0",
|
||||
"port": 5432,
|
||||
"interface": None,
|
||||
"process": "postgres",
|
||||
"pid": 812,
|
||||
"unit": "postgresql@16-main",
|
||||
"container_id": None,
|
||||
}
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"local, address, interface",
|
||||
[
|
||||
("[::]:22", "::", None),
|
||||
("*:80", "*", None),
|
||||
("127.0.0.53%lo:53", "127.0.0.53", "lo"),
|
||||
("0.0.0.0%ens18:68", "0.0.0.0", "ens18"),
|
||||
("[::ffff:127.0.0.1]:8125", "::ffff:127.0.0.1", None),
|
||||
("[fe80::1%eth0]:546", "fe80::1", "eth0"),
|
||||
(":::22", "::", None), # iproute2 4.9 prints IPv6 without brackets
|
||||
],
|
||||
)
|
||||
def test_every_address_form(self, local: str, address: str, interface):
|
||||
line = f"tcp LISTEN 0 128 {local} *:* users:((\"x\",pid=1,fd=3))\n"
|
||||
|
||||
[socket] = parse_listening_sockets(_wire(line, "1 0::/system.slice/x.service\n"))
|
||||
|
||||
assert (socket["address"], socket["interface"]) == (address, interface)
|
||||
|
||||
def test_the_first_of_several_processes_names_the_socket(self):
|
||||
nginx = _by_port(parse_listening_sockets(_wire()))[("tcp", 80, "*")]
|
||||
|
||||
assert (nginx["process"], nginx["pid"], nginx["unit"]) == ("nginx", 901, "nginx")
|
||||
|
||||
def test_a_socket_without_a_process_is_kept(self):
|
||||
"""The kernel's nfsd, or every socket of another user without root."""
|
||||
nfs = _by_port(parse_listening_sockets(_wire()))[("tcp", 2049, "0.0.0.0")]
|
||||
|
||||
assert (nfs["process"], nfs["pid"], nfs["unit"], nfs["container_id"]) == (
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
)
|
||||
|
||||
def test_the_header_is_no_socket(self):
|
||||
sockets = parse_listening_sockets(_wire())
|
||||
|
||||
assert len(sockets) == 9
|
||||
assert all(s["proto"] in ("tcp", "udp") for s in sockets)
|
||||
|
||||
def test_sorted_by_protocol_port_and_address(self):
|
||||
keys = [(s["proto"], s["port"], s["address"]) for s in parse_listening_sockets(_wire())]
|
||||
|
||||
assert keys == sorted(keys)
|
||||
|
||||
def test_nothing_listening_is_an_empty_list(self):
|
||||
assert parse_listening_sockets(_wire(SS.splitlines(True)[0], "")) == []
|
||||
|
||||
|
||||
class TestCgroups:
|
||||
def _unit_and_container(self, cgroup_lines: str) -> tuple:
|
||||
line = 'tcp LISTEN 0 128 0.0.0.0:1 0.0.0.0:* users:(("p",pid=5,fd=3))\n'
|
||||
[socket] = parse_listening_sockets(_wire(line, cgroup_lines))
|
||||
return socket["unit"], socket["container_id"]
|
||||
|
||||
def test_a_container_on_the_host_network(self):
|
||||
assert self._unit_and_container(f"5 0::/system.slice/docker-{CID}.scope\n") == (
|
||||
None,
|
||||
CID,
|
||||
)
|
||||
|
||||
def test_a_container_under_the_cgroupfs_driver(self):
|
||||
assert self._unit_and_container(f"5 0::/docker/{CID}\n") == (None, CID)
|
||||
|
||||
def test_cgroup_v1_reads_the_systemd_hierarchy(self):
|
||||
lines = "5 12:pids:/user.slice\n5 1:name=systemd:/system.slice/ssh.service\n"
|
||||
|
||||
assert self._unit_and_container(lines) == ("ssh", None)
|
||||
|
||||
def test_an_escaped_template_instance(self):
|
||||
lines = "5 0::/system.slice/system-wg\\x2dquick.slice/wg-quick@wg0.service\n"
|
||||
|
||||
assert self._unit_and_container(lines) == ("wg-quick@wg0", None)
|
||||
|
||||
def test_a_login_session_is_no_unit(self):
|
||||
assert self._unit_and_container("5 0::/user.slice/user-1000.slice/session-3.scope\n") == (
|
||||
None,
|
||||
None,
|
||||
)
|
||||
|
||||
def test_a_process_gone_before_its_cgroup_was_read(self):
|
||||
assert self._unit_and_container("") == (None, None)
|
||||
|
||||
def test_the_docker_daemons_own_proxy_stays_raw(self):
|
||||
"""docker-proxy runs in docker.service. Telling it apart from a real
|
||||
listener is the consumer's business; the reading reports what is there."""
|
||||
proxy = _by_port(parse_listening_sockets(_wire()))[("tcp", 8080, "0.0.0.0")]
|
||||
|
||||
assert (proxy["process"], proxy["unit"]) == ("docker-proxy", "docker")
|
||||
|
||||
|
||||
class TestFailures:
|
||||
def test_whatever_surrounds_the_frame_is_ignored(self):
|
||||
noisy = _wire(noise="$ sh -c '...'\nWelcome to Ubuntu\n")
|
||||
|
||||
assert len(parse_listening_sockets(noisy)) == 9
|
||||
|
||||
def test_output_without_the_frame_raises(self):
|
||||
with pytest.raises(ValueError):
|
||||
parse_listening_sockets("sudo: a password is required\n")
|
||||
|
||||
def test_a_report_cut_short_raises(self):
|
||||
with pytest.raises(ValueError):
|
||||
parse_listening_sockets(_wire()[:-len("SOCK_END\n")])
|
||||
|
||||
def test_a_host_without_ss_is_unavailable(self):
|
||||
with pytest.raises(ListeningSocketsUnavailable):
|
||||
parse_listening_sockets("SOCK_BEGIN\n[no-ss]\nSOCK_END\n")
|
||||
|
||||
def test_ss_failing_raises(self):
|
||||
with pytest.raises(ValueError):
|
||||
parse_listening_sockets(_wire("ss: invalid option -- 'p'\n", "", rc=1))
|
||||
|
||||
def test_unavailable_is_not_a_value_error(self):
|
||||
"""A caller retries a broken report without privilege; a host without
|
||||
ss has nothing to retry."""
|
||||
assert not issubclass(ListeningSocketsUnavailable, ValueError)
|
||||
|
||||
|
||||
class TestTheCommand:
|
||||
def test_the_frame_is_not_in_the_command_itself(self):
|
||||
"""An echoing transport prints the command back; the markers must only
|
||||
appear once the command has run."""
|
||||
assert "SOCK_BEGIN" not in LISTENING_SOCKETS_COMMAND
|
||||
assert "SOCK_END" not in LISTENING_SOCKETS_COMMAND
|
||||
|
||||
def test_it_writes_and_changes_nothing(self):
|
||||
for verb in ("sudo", " > ", ">>", "kill", "rm ", "systemctl "):
|
||||
assert verb not in LISTENING_SOCKETS_COMMAND
|
||||
|
||||
def test_it_is_posix_sh(self):
|
||||
result = subprocess.run(
|
||||
["sh", "-n", "-c", LISTENING_SOCKETS_COMMAND], capture_output=True, text=True
|
||||
)
|
||||
assert result.returncode == 0, result.stderr
|
||||
|
||||
@pytest.mark.skipif(shutil.which("ss") is None, reason="needs ss on this host")
|
||||
def test_it_runs_and_parses_on_this_host(self):
|
||||
out = subprocess.run(
|
||||
["sh", "-c", LISTENING_SOCKETS_COMMAND], capture_output=True, text=True, timeout=60
|
||||
).stdout
|
||||
|
||||
sockets = parse_listening_sockets(out)
|
||||
|
||||
assert all(s["proto"] in ("tcp", "udp") and 0 < s["port"] < 65536 for s in sockets)
|
||||
|
||||
|
||||
class _Driver(ListeningSocketsMixin):
|
||||
def __init__(self, privileged: str, unprivileged: str = "") -> None:
|
||||
self.answers = {True: privileged, False: unprivileged}
|
||||
self.calls: list = []
|
||||
|
||||
def _run_listening_sockets_command(self, command: str, *, privileged: bool) -> str:
|
||||
self.calls.append((command, privileged))
|
||||
return self.answers[privileged]
|
||||
|
||||
|
||||
class TestTheTemplate:
|
||||
def test_a_driver_supplies_only_the_transport(self):
|
||||
driver = _Driver(_wire())
|
||||
|
||||
reading = driver.get_listening_sockets()
|
||||
|
||||
assert reading["attributed"] is True
|
||||
assert len(reading["sockets"]) == 9
|
||||
[(command, privileged)] = driver.calls
|
||||
assert privileged is True
|
||||
|
||||
def test_the_command_reaches_root_as_one_shell(self):
|
||||
"""``sudo -n a; b`` runs only ``a`` as root: the whole script goes as
|
||||
one ``sh -c`` argument."""
|
||||
driver = _Driver(_wire())
|
||||
driver.get_listening_sockets()
|
||||
|
||||
[(command, _privileged)] = driver.calls
|
||||
assert command.startswith("sh -c '")
|
||||
assert subprocess.run(["sh", "-n", "-c", command]).returncode == 0
|
||||
|
||||
def test_without_root_it_reads_what_it_can(self):
|
||||
"""Without sudo, ss names only the user's own processes: the sockets
|
||||
are still worth having, marked as not attributed."""
|
||||
driver = _Driver("sudo: a password is required\n", _wire(cgroups=""))
|
||||
|
||||
reading = driver.get_listening_sockets()
|
||||
|
||||
assert reading["attributed"] is False
|
||||
assert len(reading["sockets"]) == 9
|
||||
assert [p for _c, p in driver.calls] == [True, False]
|
||||
|
||||
def test_a_host_without_ss_is_not_asked_twice(self):
|
||||
driver = _Driver("SOCK_BEGIN\n[no-ss]\nSOCK_END\n")
|
||||
|
||||
with pytest.raises(ListeningSocketsUnavailable):
|
||||
driver.get_listening_sockets()
|
||||
assert len(driver.calls) == 1
|
||||
|
||||
def test_not_every_os_driver_has_it(self):
|
||||
"""A Windows host is an OSDriver too and has no ss: ``hasattr`` has to
|
||||
stay a truthful answer, so the drivers that can mix this in themselves."""
|
||||
assert not issubclass(OSDriver, ListeningSocketsMixin)
|
||||
assert not hasattr(OSDriver, "get_listening_sockets")
|
||||
Reference in New Issue
Block a user