luna/enodia-sentinal
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add host event rules for bind and capability activity

Browse source
This commit is contained in:
Luna 2026-07-10 04:11:33 -07:00
parent b40ac4252c
commit 9784033f86
No known key found for this signature in database
12 changed files with 316 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

4
enodia_sentinel/events/host_event.py
View file

@ -22,6 +22,8 @@ class HostEvent:
target_uid: int = -1
target_gid: int = -1
mode: int = 0
capabilities: tuple[str, ...] = field(default_factory=tuple)
module_name: str = ""
@property
def argv_str(self) -> str:
@ -44,4 +46,6 @@ class HostEvent:
"target_uid": self.target_uid,
"target_gid": self.target_gid,
"mode": self.mode,
"capabilities": list(self.capabilities),
"module_name": self.module_name,
}

63
enodia_sentinel/events/host_rules.py
View file

@ -21,6 +21,15 @@ _PERSISTENCE_PREFIXES = (
"/etc/ld.so.preload", "/root/.ssh/authorized_keys",
"/etc/sudoers", "/etc/sudoers.d/",
)
_WRITABLE_RUNTIME_PREFIXES = ("/tmp/", "/var/tmp/", "/dev/shm/", "/run/user/")
_SENSITIVE_CAPABILITIES = frozenset({
"CAP_SYS_ADMIN",
"CAP_SYS_MODULE",
"CAP_SYS_PTRACE",
"CAP_DAC_READ_SEARCH",
"CAP_NET_ADMIN",
"CAP_NET_RAW",
})
@dataclass(frozen=True)
@ -40,6 +49,8 @@ class HostRule:
path_prefixes: tuple[str, ...] = ()
target_uids: frozenset[int] = frozenset()
target_gids: frozenset[int] = frozenset()
capabilities_any: frozenset[str] = frozenset()
module_names: frozenset[str] = frozenset()
argv_regex: str | None = None
def __post_init__(self) -> None:
@ -49,7 +60,8 @@ class HostRule:
self.comm, self.parent_comm, self.peer_public is not None,
self.peer_ports, self.peer_port_exclude, self.local_ports,
self.local_port_exclude, self.path_prefixes, self.target_uids,
self.target_gids, self.argv_regex,
self.target_gids, self.capabilities_any, self.module_names,
self.argv_regex,
)):
raise ValueError(f"rule sid={self.sid} has no match conditions")
if self.argv_regex is not None:
@ -84,6 +96,12 @@ class HostRule:
return False
if self.target_gids and ev.target_gid not in self.target_gids:
return False
if self.capabilities_any and not (
{cap.upper() for cap in ev.capabilities} & self.capabilities_any
):
return False
if self.module_names and ev.module_name not in self.module_names:
return False
if self._argv_re is not None and not self._argv_re.search(ev.argv_str):
return False
return True
@ -95,14 +113,17 @@ class HostRule:
key=(
f"host:{self.sid}:{ev.event}:{ev.pid}:"
f"{ev.peer_ip}:{ev.peer_port}:{ev.local_ip}:{ev.local_port}:"
f"{ev.path}:{ev.target_uid}:{ev.target_gid}:{ev.mode}"
f"{ev.path}:{ev.target_uid}:{ev.target_gid}:{ev.mode}:"
f"{','.join(ev.capabilities)}:{ev.module_name}"
),
detail=(
f"sid={self.sid} {self.msg} - pid={ev.pid} ppid={ev.ppid} "
f"comm={ev.comm} event={ev.event} peer={ev.peer_ip}:{ev.peer_port} "
f"local={ev.local_ip}:{ev.local_port} path={ev.path} "
f"target_uid={ev.target_uid} target_gid={ev.target_gid} "
f"mode={oct(ev.mode) if ev.mode else '-'}"
f"mode={oct(ev.mode) if ev.mode else '-'} "
f"capabilities={','.join(ev.capabilities) or '-'} "
f"module={ev.module_name or '-'}"
),
pids=(ev.pid,),
sid=self.sid,
@ -130,6 +151,15 @@ DEFAULT_HOST_RULES: tuple[HostRule, ...] = (
comm=_INTERPRETERS,
local_port_exclude=_COMMON_PUBLIC_PORTS,
),
HostRule(
sid=100073,
msg="Interpreter bound an unusual local port",
severity=Severity.HIGH,
classtype="suspicious-bind",
events=frozenset({"bind"}),
comm=_INTERPRETERS,
local_port_exclude=_COMMON_PUBLIC_PORTS,
),
HostRule(
sid=100069,
msg="Interpreter wrote to a persistence path",
@ -165,6 +195,33 @@ DEFAULT_HOST_RULES: tuple[HostRule, ...] = (
comm=_INTERPRETERS,
target_gids=frozenset({0}),
),
HostRule(
sid=100076,
msg="Interpreter accepted inbound traffic on an unusual local port",
severity=Severity.HIGH,
classtype="suspicious-accept",
events=frozenset({"accept"}),
comm=_INTERPRETERS,
peer_public=True,
local_port_exclude=_COMMON_PUBLIC_PORTS,
),
HostRule(
sid=100077,
msg="Interpreter requested sensitive Linux capabilities",
severity=Severity.HIGH,
classtype="capability-escalation",
events=frozenset({"capset"}),
comm=_INTERPRETERS,
capabilities_any=_SENSITIVE_CAPABILITIES,
),
HostRule(
sid=100078,
msg="Kernel module load requested from a writable runtime path",
severity=Severity.CRITICAL,
classtype="suspicious-module-load",
events=frozenset({"module_load"}),
path_prefixes=_WRITABLE_RUNTIME_PREFIXES,
),
)

45
enodia_sentinel/ruleops.py
View file

@ -174,6 +174,10 @@ def _host_rule_record(rule: HostRule, origin: str) -> dict[str, Any]:
conditions["target_uids"] = sorted(rule.target_uids)
if rule.target_gids:
conditions["target_gids"] = sorted(rule.target_gids)
if rule.capabilities_any:
conditions["capabilities_any"] = sorted(rule.capabilities_any)
if rule.module_names:
conditions["module_names"] = sorted(rule.module_names)
if rule.argv_regex:
conditions["argv_regex"] = rule.argv_regex
return {
@ -280,6 +284,34 @@ _RULE_DOCS: dict[int, dict[str, Any]] = {
],
"drill": "Use `rules test` with a `listen` event whose `comm` is an interpreter and whose `local_port` is not a common service port.",
},
100073: {
"false_positives": [
"Developer or diagnostic scripts binding high local ports intentionally.",
"Short-lived local service wrappers that bind before handing sockets to a supervised process.",
],
"drill": "Use `rules test` with a `bind` event whose `comm` is an interpreter and whose `local_port` is not a common service port.",
},
100076: {
"false_positives": [
"Developer servers or test harnesses accepting inbound public traffic from an interpreter.",
"Administrative troubleshooting with temporary netcat/socat listeners.",
],
"drill": "Use `rules test` with an `accept` event whose `comm` is an interpreter, public `peer_ip`, and unusual `local_port`.",
},
100077: {
"false_positives": [
"Privileged maintenance scripts that intentionally adjust capabilities during controlled administration.",
"Container or network lab setup scripts using interpreters to configure namespaces or packet capture.",
],
"drill": "Use `rules test` with a `capset` event containing a sensitive capability such as `CAP_SYS_ADMIN`.",
},
100078: {
"false_positives": [
"Kernel development labs loading locally built modules from temporary build directories.",
"Driver troubleshooting sessions that intentionally test an unsigned module from a writable path.",
],
"drill": "Use `rules test` with a `module_load` event whose `path` is under `/tmp`, `/var/tmp`, `/dev/shm`, or `/run/user`.",
},
}
@ -368,6 +400,11 @@ def _host_event(event: dict[str, Any]) -> HostEvent:
target_uid=_to_int(event.get("target_uid", event.get("uid_target", -1))),
target_gid=_to_int(event.get("target_gid", event.get("gid_target", -1))),
mode=_to_int(event.get("mode", 0)),
capabilities=_to_str_tuple(event.get(
"capabilities",
event.get("caps", event.get("cap_effective", event.get("capability", ()))),
)),
module_name=str(event.get("module_name", event.get("name", ""))),
)
@ -375,3 +412,11 @@ def _to_int(value: Any) -> int:
if isinstance(value, str):
return int(value, 0)
return int(value)
def _to_str_tuple(value: Any) -> tuple[str, ...]:
if value is None:
return ()
if isinstance(value, str):
return (value.upper(),) if value else ()
return tuple(str(item).upper() for item in value)