stackpilot/backend/services/port_service.py

"""Port conflict detection before deploying a stack."""
from __future__ import annotations

import os
from typing import Optional

import yaml

from config import settings
from docker_client import DockerError, get_client, safe_call


# --------------------------------------------------------------------------- #
# Parse compose `ports:` entries
# --------------------------------------------------------------------------- #


def parse_compose_ports(yaml_str: str) -> list[dict]:
    """Return [{host_port:int, protocol:'tcp'|'udp', service:str}]."""
    try:
        data = yaml.safe_load(yaml_str) or {}
    except yaml.YAMLError:
        return []
    services = data.get("services") or {}
    out: list[dict] = []
    for svc_name, svc in services.items():
        if not isinstance(svc, dict):
            continue
        for entry in svc.get("ports", []) or []:
            parsed = _parse_port_entry(entry)
            if parsed:
                parsed["service"] = svc_name
                out.append(parsed)
    return out


def _parse_port_entry(entry) -> Optional[dict]:
    # Long form: {target, published, protocol}
    if isinstance(entry, dict):
        published = entry.get("published")
        if published is None:
            return None
        try:
            host_port = int(str(published).split("-")[0])
        except ValueError:
            return None
        return {"host_port": host_port, "protocol": entry.get("protocol", "tcp")}

    # Short form string: "[ip:]host:container[/proto]" or "container"
    s = str(entry)
    proto = "tcp"
    if "/" in s:
        s, proto = s.rsplit("/", 1)
    parts = s.split(":")
    # No host mapping (only container port) -> random host port, no conflict.
    if len(parts) == 1:
        return None
    # host:container  or  ip:host:container
    host = parts[-2]
    try:
        host_port = int(host.split("-")[0])
    except ValueError:
        return None
    return {"host_port": host_port, "protocol": proto}


# --------------------------------------------------------------------------- #
# Host bound ports
# --------------------------------------------------------------------------- #


def _read_proc_net(name: str) -> set[int]:
    ports: set[int] = set()
    path = os.path.join(settings.HOST_PROC_PATH, "net", name)
    if not os.path.isfile(path):
        path = os.path.join("/proc/net", name)
    try:
        with open(path, "r", encoding="utf-8") as fh:
            lines = fh.readlines()[1:]
    except OSError:
        return ports
    for line in lines:
        cols = line.split()
        if len(cols) < 4:
            continue
        local = cols[1]  # hexip:hexport
        state = cols[3]
        # TCP listen state is 0A; for UDP accept all.
        if name.startswith("tcp") and state != "0A":
            continue
        try:
            port = int(local.split(":")[1], 16)
            ports.add(port)
        except (IndexError, ValueError):
            continue
    return ports


def host_listening_ports() -> dict[str, set[int]]:
    return {
        "tcp": _read_proc_net("tcp") | _read_proc_net("tcp6"),
        "udp": _read_proc_net("udp") | _read_proc_net("udp6"),
    }


def docker_bound_ports() -> dict[tuple[int, str], str]:
    """Return {(host_port, proto): container_name}."""
    out: dict[tuple[int, str], str] = {}
    try:
        client = get_client()
        for c in safe_call(client.containers.list):
            bindings = (c.attrs.get("NetworkSettings") or {}).get("Ports") or {}
            for container_port, hosts in bindings.items():
                if not hosts:
                    continue
                proto = container_port.split("/")[-1] if "/" in container_port else "tcp"
                for h in hosts:
                    hp = h.get("HostPort")
                    if hp:
                        out[(int(hp), proto)] = c.name
    except (DockerError, ValueError):
        pass
    return out


# --------------------------------------------------------------------------- #
# Detect conflicts
# --------------------------------------------------------------------------- #


def detect_conflicts(yaml_str: str, ignore_stack: Optional[str] = None) -> list[dict]:
    wanted = parse_compose_ports(yaml_str)
    host_ports = host_listening_ports()
    docker_ports = docker_bound_ports()

    conflicts: list[dict] = []
    for w in wanted:
        port = w["host_port"]
        proto = w.get("protocol", "tcp")
        used_by = None
        owner = docker_ports.get((port, proto))
        if owner:
            # A container from the same stack (re-deploy) is not a conflict.
            if ignore_stack and owner.startswith(f"{ignore_stack}-"):
                continue
            used_by = f"container {owner}"
        elif port in host_ports.get(proto, set()):
            used_by = "host process"
        if used_by:
            conflicts.append(
                {
                    "port": port,
                    "protocol": proto,
                    "service": w.get("service"),
                    "used_by": used_by,
                }
            )
    return conflicts