66 lines
2.3 KiB
Python
66 lines
2.3 KiB
Python
# 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",
|
|
)
|