signal-bot/signal_bot/portcheck.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Refuse to start on a taken port, and name whatever is already holding it.

uvicorn logs "address already in use" and then shuts down *cleanly*: the
process exits 0, so a supervisor records a successful run and a human reads a
tidy "Application shutdown complete" as if nothing were wrong. That is a
fail-open, and it is exactly how this bot silently failed to start.

Attribution reads /proc directly rather than forking `ss` or `lsof`: no
subprocess, no PATH assumptions, and the whole thing is testable against a
fake proc tree passed in as `proc_root`.
"""

from __future__ import annotations

import socket
from pathlib import Path
from typing import Final

PROC_ROOT: Final = Path("/proc")
_TCP_TABLES: Final = ("net/tcp", "net/tcp6")
_LISTEN_STATE: Final = "0A"
_INODE_FIELD: Final = 9


class PortInUseError(RuntimeError):
    """Raised when the bot's bind address is already taken."""


def _listening_inodes(port: int, proc_root: Path) -> set[str]:
    """Return the socket inodes of every LISTEN socket on the given port."""
    inodes: set[str] = set()
    for table in _TCP_TABLES:
        try:
            # Skip the header row; every remaining line is one socket.
            lines = (proc_root / table).read_text().splitlines()[1:]
        except OSError:
            continue
        for line in lines:
            fields = line.split()
            if len(fields) <= _INODE_FIELD or fields[3] != _LISTEN_STATE:
                continue
            _, _, hex_port = fields[1].rpartition(":")
            try:
                listening_on = int(hex_port, 16)
            except ValueError:
                continue
            if listening_on == port:
                inodes.add(fields[_INODE_FIELD])
    return inodes


def _command_of(pid_dir: Path) -> str:
    """Return the full command line of a process, or a placeholder."""
    try:
        raw = (pid_dir / "cmdline").read_bytes()
    except OSError:
        return "(command unavailable)"
    argv = raw.decode(errors="replace").rstrip("\0").split("\0")
    return " ".join(argv).strip() or "(command unavailable)"


def _owner_of(inodes: set[str], proc_root: Path) -> tuple[int, str] | None:
    """Find the pid holding any of these socket inodes, with its command line."""
    targets = {f"socket:[{inode}]" for inode in inodes}
    for pid_dir in sorted(proc_root.iterdir()):
        if not pid_dir.name.isdigit():
            continue
        try:
            descriptors = sorted((pid_dir / "fd").iterdir())
        except OSError:
            # Another user's process, or one that exited mid-scan.
            continue
        for descriptor in descriptors:
            try:
                link = str(descriptor.readlink())
            except OSError:
                continue
            if link in targets:
                return int(pid_dir.name), _command_of(pid_dir)
    return None


def describe_listener(port: int, proc_root: Path = PROC_ROOT) -> str:
    """Describe whoever is listening on a port, or "" if nobody identifiable is."""
    inodes = _listening_inodes(port, proc_root)
    if not inodes:
        return ""
    owner = _owner_of(inodes, proc_root)
    if owner is None:
        return "held by a process belonging to another user"
    pid, command = owner
    return f"held by pid {pid}: {command}"


def assert_port_free(host: str, port: int, proc_root: Path = PROC_ROOT) -> None:
    """Raise PortInUseError unless host:port can actually be bound.

    Args:
        host: Address the server will bind.
        port: Port the server will bind.
        proc_root: Root of the proc filesystem. Replaced in tests.

    Raises:
        PortInUseError: The address is already in use.
    """
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
        probe.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        try:
            probe.bind((host, port))
        except OSError as exc:
            holder = describe_listener(port, proc_root)
            detail = f" -- {holder}" if holder else ""
            msg = (
                f"cannot bind {host}:{port}{detail}. "
                f"Set BIND_PORT in .env to a free port, or stop that process. "
                f"Refusing to start, because uvicorn would log this and exit 0."
            )
            raise PortInUseError(msg) from exc