# SPDX-License-Identifier: GPL-3.0-or-later
"""stealth_network — unusual socket families used for covert channels.

TCP and UDP are covered by reverse-shell, egress, and listener checks. This
detector watches families attackers use to slide around those paths: raw IP,
SCTP, DCCP, packet sockets, XDP, TIPC, vsock, and MPTCP.
"""
from __future__ import annotations

from collections.abc import Iterator

from ..alert import Alert, Severity
from ..config import Config
from ..netutil import ip_in_cidrs, is_public_ip, split_host_port
from ..system import SystemState

SID_STEALTH_NETWORK = 100036

_WATCHED_KINDS = frozenset({
    "raw", "sctp", "dccp", "packet", "mptcp", "tipc", "xdp", "vsock",
})
_ALWAYS_HIGH = frozenset({"raw", "packet", "xdp"})
_ACTIVE_STATES = frozenset({
    "ESTAB", "LISTEN", "UNCONN", "CONNECTED", "SYN-SENT", "SYN-RECV",
})


def _peer_is_public(peer: str, cfg: Config) -> bool:
    host, _port = split_host_port(peer)
    if not is_public_ip(host):
        return False
    return not ip_in_cidrs(host, cfg.egress_allow_cidrs)


def _severity(kind: str, state: str, peer: str, cfg: Config) -> Severity:
    if kind in _ALWAYS_HIGH:
        return Severity.HIGH
    if state == "LISTEN" or _peer_is_public(peer, cfg):
        return Severity.HIGH
    return Severity.MEDIUM


def detect(state: SystemState, cfg: Config) -> Iterator[Alert]:
    allowed_kinds = set(cfg.stealth_network_allow_kinds)
    for sock in state.sockets:
        kind = sock.kind.lower()
        if kind not in _WATCHED_KINDS:
            continue
        if kind in allowed_kinds:
            continue
        if sock.comm and sock.comm in cfg.stealth_network_allow_comms:
            continue
        if sock.state and sock.state not in _ACTIVE_STATES:
            continue
        yield Alert(
            severity=_severity(kind, sock.state, sock.peer, cfg),
            signature="stealth_network",
            key=f"stealthnet:{kind}:{sock.pid}:{sock.local}:{sock.peer}",
            detail=(
                f"{kind} socket state={sock.state} local={sock.local} "
                f"peer={sock.peer} comm={sock.comm or '?'} pid={sock.pid or '?'}"
            ),
            pids=(sock.pid,) if sock.pid else (),
            sid=SID_STEALTH_NETWORK,
            classtype="covert-channel",
        )