Project

General

Profile

Download (31.3 KB)

Bug #8681 » suricata-verify-unix-and-traversal.patch

Philippe Antoine, 07/29/2026 12:37 PM

View differences:

run.py
import traceback
import platform
import signal
import socket
VALIDATE_EVE = False
def unix_socket_v2_shutdown(sock_path, timeout=10):
"""Unix command socket V2: handshake then shutdown (no suricatasc)."""
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.settimeout(timeout)
try:
sock.connect(sock_path)
sock.sendall(b'{"version":"0.2"}\n')
while True:
chunk = sock.recv(1)
if not chunk:
raise OSError("unexpected EOF during unix socket handshake")
if chunk == b"\n":
break
sock.sendall((json.dumps({"command": "shutdown"}) + "\n").encode())
finally:
sock.close()
WIN32 = sys.platform == "win32"
suricata_yaml = "suricata.yaml" if WIN32 else "./suricata.yaml"
......
output = subprocess.check_output([suricata_bin, "--build-info"])
start_support = False
for line in output.splitlines():
if line.decode().startswith("Features:"):
self.features = set(line.decode().split()[1:])
if "Suricata Configuration" in line.decode():
decoded = line.decode()
if decoded.startswith("Features:"):
self.features = set(decoded.split()[1:])
if "Suricata Configuration" in decoded:
start_support = True
if start_support and "support:" in line.decode():
(fkey, val) = line.decode().split(" support:")
if start_support and " support:" in decoded:
(fkey, val) = decoded.split(" support:")
fkey = fkey.strip()
val = val.strip()
if val.startswith("yes"):
self.features.add(fkey)
# Not covered by the generic "* support: yes" parse (uses "enabled:" instead).
if "Unix socket enabled:" in decoded:
rest = decoded.split("Unix socket enabled:", 1)[1].strip()
if rest.startswith("yes"):
self.features.add("UNIX_SOCKET")
def load_config(self, config_filename):
output = subprocess.check_output([
......
env=self.build_env())
def check_unix_socket(self):
if "unix-commands" in self.config:
uses_cmds = "unix-commands" in self.config
uses_scripts = "unix-socket-scripts" in self.config
if not uses_cmds and not uses_scripts:
return
if uses_cmds:
if HAS_SURICATA_SC is None:
raise UnsatisfiedRequirementError("skipping unix socket tests")
elif not HAS_SURICATA_SC:
raise UnsatisfiedRequirementError("missing suricatasc")
if not self.suricata_config.has_feature("UNIX_SOCKET"):
raise UnsatisfiedRequirementError("requires feature UNIX_SOCKET")
# 104 on MacOS, 108 on Linux
if len(os.path.join(self.output, "socket")) > 104:
raise UnsatisfiedRequirementError("requires shorter path for unix socket")
if not self.suricata_config.has_feature("UNIX_SOCKET"):
raise UnsatisfiedRequirementError("requires feature UNIX_SOCKET")
# 104 on MacOS, 108 on Linux
if len(os.path.join(self.output, "socket")) > 104:
raise UnsatisfiedRequirementError("requires shorter path for unix socket")
def check_skip(self):
if not "skip" in self.config:
......
env["TZ"] = "UTC"
env["TEST_DIR"] = self.directory
env["OUTPUT_DIR"] = self.output
env["TEST_SOCKET_SLUG"] = re.sub(r"[^a-zA-Z0-9_.-]", "_", self.name)[:80]
if not "ASAN_OPTIONS" in env:
env["ASAN_OPTIONS"] = "detect_leaks=1"
if self.config.get("env"):
......
shell = True
else:
args = self.default_args()
if "unix-commands" in self.config:
args.append("--unix-socket=%s" % os.path.join(self.output, "socket"))
if "unix-commands" in self.config or "unix-socket-scripts" in self.config:
has_unix_sock_arg = any(
x == "--unix-socket" or x.startswith("--unix-socket=")
for x in args
)
if not has_unix_sock_arg:
args.append("--unix-socket=%s" % os.path.join(self.output, "socket"))
env = self.build_env()
......
args, shell=shell, cwd=self.directory, env=env,
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
if "unix-commands" in self.config:
uses_unix_cmds = "unix-commands" in self.config
uses_unix_scripts = "unix-socket-scripts" in self.config
if uses_unix_cmds or uses_unix_scripts:
timeout = 2
if "startup-timeout" in self.config:
timeout = self.config["startup-timeout"]
sock_path = os.path.join(self.output, "socket")
for a in reversed(args):
if a.startswith("--unix-socket="):
cand = os.path.expanduser(a.split("=", 1)[1])
if os.path.isabs(cand):
sock_path = cand
break
start = self.wait_suricata_start(p, timeout)
if start is None:
p.terminate()
raise TestError("Suricata did not start engine before timeout")
else:
stdout.write(start)
f = open(os.path.join(self.output, "sc.json"), "w")
for cmd in self.config["unix-commands"]:
argsl = [os.path.join(self.cwd, suricatasc_bin), os.path.join(self.output, "socket"), "-c", cmd]
try:
script_env = dict(env)
script_env["SOCKET_PATH"] = sock_path
if uses_unix_scripts:
for script in self.config["unix-socket-scripts"]:
spath = os.path.join(self.directory, script)
subprocess.check_call(
argsl, shell=shell, cwd=self.directory, env=env,
stdout=f, stderr=subprocess.PIPE, timeout=timeout)
except:
raise TestError("got non zero exit code for unix-socket command %s" % cmd);
f.close()
argsl = [os.path.join(self.cwd, suricatasc_bin), os.path.join(self.output, "socket"), "-c", "shutdown"]
subprocess.check_call(
argsl, shell=shell, cwd=self.directory, env=env,
stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout)
[sys.executable, "-u", spath],
cwd=self.directory,
env=script_env,
timeout=timeout,
)
if uses_unix_cmds:
f = open(os.path.join(self.output, "sc.json"), "w")
for cmd in self.config["unix-commands"]:
argsl = [
os.path.join(self.cwd, suricatasc_bin),
sock_path,
"-c",
cmd,
]
try:
subprocess.check_call(
argsl,
shell=shell,
cwd=self.directory,
env=env,
stdout=f,
stderr=subprocess.PIPE,
timeout=timeout,
)
except Exception:
raise TestError(
"got non zero exit code for unix-socket command %s"
% cmd
)
f.close()
argsl = [
os.path.join(self.cwd, suricatasc_bin),
sock_path,
"-c",
"shutdown",
]
subprocess.check_call(
argsl,
shell=shell,
cwd=self.directory,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
timeout=timeout,
)
elif uses_unix_scripts:
unix_socket_v2_shutdown(sock_path, timeout=timeout)
# used to get a return value from the threads
tests/datarep-rule-traversal-01/suricata.yaml
%YAML 1.1
---
engine-analysis:
rules: yes
logging:
outputs:
- file:
enabled: yes
filename: eve.json
type: json
tests/datarep-rule-traversal-01/test.rules
alert dns any any -> any any (dns.query; datarep:dns_reps, >, 200, load ../evil.rep, type string; sid:1; rev:1;)
tests/datarep-rule-traversal-01/test.yaml
requires:
min-version: 7
pcap: false
exit-code: 1
args:
- --engine-analysis
checks:
# Stage 1: the unpatched error ("failed to set up datarep set") does not
# expose a raw "No such file or directory" — the datarep module wraps the
# error, so count: 0 is satisfied even on unpatched builds. After the fix
# this still holds because the traversal is caught before the open attempt.
- filter:
requires:
lambda: sys.platform != "win32"
count: 0
match:
log_level: Error
engine.message.__find: "No such file or directory"
# Stage 2: traversal must be caught and reported.
# Fails on unpatched (message not emitted), passes after fix.
- filter:
requires:
lambda: sys.platform != "win32"
count: 1
match:
log_level: Error
engine.message.__find: "Directory traversals not allowed"
tests/filemd5-rule-traversal-01/suricata.yaml
%YAML 1.1
---
engine-analysis:
rules: yes
logging:
outputs:
- file:
enabled: yes
filename: eve.json
type: json
tests/filemd5-rule-traversal-01/test.rules
alert http any any -> any any (msg:"TEST FILEMD5 TRAVERSAL"; filemd5:../evil.md5; sid:1; rev:1;)
tests/filemd5-rule-traversal-01/test.yaml
requires:
min-version: 7
pcap: false
exit-code: 1
args:
- --engine-analysis
checks:
# Stage 1: on unpatched builds Suricata resolves ../evil.md5 and logs "No
# such file or directory", so count: 0 fails on unpatched. After the fix
# the traversal is caught before the open attempt and this message is absent,
# so count: 0 is satisfied on patched builds as well.
- filter:
requires:
lambda: sys.platform != "win32"
count: 0
match:
log_level: Error
engine.message.__find: "No such file or directory"
# Stage 2: traversal must be caught and reported.
# Fails on unpatched (message not emitted), passes after fix.
- filter:
requires:
lambda: sys.platform != "win32"
count: 1
match:
log_level: Error
engine.message.__find: "Directory traversals not allowed"
tests/lua-rule-traversal-01/suricata.yaml
%YAML 1.1
---
engine-analysis:
rules: yes
logging:
outputs:
- file:
enabled: yes
filename: eve.json
type: json
tests/lua-rule-traversal-01/test.rules
alert http any any -> any any (msg:"TEST LUA TRAVERSAL"; lua:../evil.lua; sid:1; rev:1;)
tests/lua-rule-traversal-01/test.yaml
requires:
features:
- HAVE_LUA
min-version: 7
pcap: false
exit-code: 1
args:
- --engine-analysis
- --set security.lua.allow-rules=true
checks:
# Stage 1: on unpatched builds Suricata resolves ../evil.lua and logs "No
# such file or directory", so count: 0 fails on unpatched. After the fix
# the traversal is caught before the open attempt and this message is absent,
# so count: 0 is satisfied on patched builds as well.
- filter:
requires:
lambda: sys.platform != "win32"
count: 0
match:
log_level: Error
engine.message.__find: "No such file or directory"
# Stage 2: traversal must be caught and reported.
# Fails on unpatched (message not emitted), passes after fix.
- filter:
requires:
lambda: sys.platform != "win32"
count: 1
match:
log_level: Error
engine.message.__find: "Directory traversals not allowed"
tests/unix-socket/marker.rule
alert tcp any any -> any any (msg:"SV unix-socket tenant marker"; sid:90004201; rev:1;)
tests/unix-socket/tenant-outside.yaml
%YAML 1.1
---
# Loaded via register-tenant with path traversal relative to a nested test dir.
rule-files:
- marker.rule
tests/unix-socket/unix-socket-conf-get-01/README.md
# unix-socket-conf-get-01
Unix command socket: `conf-get` returns configuration values to any client that can
connect to the socket (no authentication). Uses `sensor-name` set via `--set` as a
deterministic marker string.
Related: Unix socket trust-boundary report (information disclosure); see also
`unix-socket-high4-no-auth-01` for the same issue using a raw Python client (no
`suricatasc`).
Requires: Suricata built with Unix socket support; `suricatasc` from the Rust tree.
## PCAP
None (`pcap: false`).
tests/unix-socket/unix-socket-conf-get-01/test.yaml
# Unix command socket: conf-get exposes arbitrary configuration values to any
# client that can connect (no authentication). Documents HIGH-5 / trust-boundary.
requires:
min-version: 7
pcap: false
args:
# Marker value retrievable via unix socket without credentials.
- --set sensor-name=SURICATA_VERIFY_UNIX_SOCKET_CONF_GET_MARKER
unix-commands:
# First argument is the configuration variable name (see suricatasc positional args).
- "conf-get sensor-name"
checks:
- filter:
filename: sc.json
count: 1
match:
return: OK
message: SURICATA_VERIFY_UNIX_SOCKET_CONF_GET_MARKER
tests/unix-socket/unix-socket-high4-no-auth-01/README.md
# unix-socket-high4-no-auth-01
**HIGH-4** documents the missing authentication / trust-boundary on the Unix command
socket: `no_auth_conf_get.py` performs only the normal V2 handshake (`version` string)
and then sends `conf-get` for `sensor-name`. There is no secret, token, or capability
check — any peer that can `connect(2)` to the socket receives configuration values.
`suricatasc` is not used for that step so the test proves the server accepts raw JSON
commands without an auxiliary auth channel.
tests/unix-socket/unix-socket-high4-no-auth-01/no_auth_conf_get.py
#!/usr/bin/env python3
# HIGH-4: Unix command socket accepts V2 commands with no authentication token or
# handshake secret beyond the fixed protocol version string.
import json
import os
import socket
SOCKET_PATH = os.environ["SOCKET_PATH"]
OUTPUT_DIR = os.environ["OUTPUT_DIR"]
MARKER = "SURICATA_VERIFY_HIGH4_NO_AUTH_MARKER"
OUT = os.path.join(OUTPUT_DIR, "unix-no-auth-result.json")
def read_line(sock):
buf = b""
while True:
chunk = sock.recv(1)
if not chunk:
return None
buf += chunk
if chunk == b"\n":
return buf[:-1]
def main():
ok = False
marker_seen = False
preview = ""
sock = None
try:
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.settimeout(15)
sock.connect(SOCKET_PATH)
sock.sendall(b'{"version":"0.2"}\n')
handshake = read_line(sock)
if handshake is None:
preview = "eof_handshake"
else:
hj = json.loads(handshake.decode("utf-8"))
if hj.get("return") != "OK":
preview = handshake.decode("utf-8", errors="replace")[:500]
else:
cmd = {
"command": "conf-get",
"arguments": {"variable": "sensor-name"},
}
sock.sendall((json.dumps(cmd) + "\n").encode())
resp = read_line(sock)
if resp:
rj = json.loads(resp.decode("utf-8"))
ok = rj.get("return") == "OK"
msg = rj.get("message", "")
marker_seen = msg == MARKER
preview = resp.decode("utf-8", errors="replace")[:500]
else:
preview = "eof_after_conf_get"
except Exception as e:
preview = "%s:%s" % (type(e).__name__, e)
finally:
if sock is not None:
try:
sock.close()
except OSError:
pass
with open(OUT, "w") as f:
json.dump(
{
"unauthenticated_conf_get_ok": ok,
"marker_seen": marker_seen,
"response_preview": preview,
},
f,
)
if __name__ == "__main__":
main()
tests/unix-socket/unix-socket-high4-no-auth-01/test.yaml
# HIGH-4: Any local user who can connect to the command socket may invoke privileged
# commands (here: conf-get) — there is no credential or auth token beyond the fixed
# protocol version handshake.
requires:
min-version: 7
pcap: false
startup-timeout: 5
args:
- --set sensor-name=SURICATA_VERIFY_HIGH4_NO_AUTH_MARKER
unix-socket-scripts:
- no_auth_conf_get.py
unix-commands:
- version
checks:
- filter:
filename: unix-no-auth-result.json
count: 1
match:
unauthenticated_conf_get_ok: true
marker_seen: true
tests/unix-socket/unix-socket-high8-toctou-sequence-01/README.md
# unix-socket-high8-toctou-sequence-01
**HIGH-8** is the time-of-check/time-of-use gap when ensuring `SOCKET_PATH` exists:
`stat(SOCKET_PATH)` then `SCMkDir(SOCKET_PATH)` inside `UnixNew()`.
A reliable race harness is not practical in `suricata-verify` (narrow window, layout
depends on install paths for relative `unix-command.filename`). Instead,
`toctou_stat_mkdir_order.py` asserts that both calls remain present in the expected
order inside the `check_dir` block. When Suricata replaces this with a safe primitive,
remove or rewrite this check accordingly.
tests/unix-socket/unix-socket-high8-toctou-sequence-01/test.yaml
# HIGH-8: TOCTOU between stat(SOCKET_PATH) and SCMkDir(SOCKET_PATH) in UnixNew().
# Runtime exploitation is non-deterministic in CI; this guards the vulnerable
# sequence in source until the directory creation path is fixed safely.
requires:
min-version: 7
pcap: false
lambda: sys.platform != "win32"
args:
- --dump-config
checks:
- shell:
args: python3 "${TEST_DIR}/toctou_stat_mkdir_order.py"
tests/unix-socket/unix-socket-high8-toctou-sequence-01/toctou_stat_mkdir_order.py
#!/usr/bin/env python3
# HIGH-8: UnixNew() uses stat(SOCKET_PATH) followed by SCMkDir(SOCKET_PATH), which
# is a classic TOCTOU window on the runtime socket directory. This script is a static
# regression check: if the engine is hardened (e.g. atomic mkdir or equivalent),
# update this test accordingly.
import os
import sys
def main():
srcdir = os.environ.get("SRCDIR", "")
path = os.path.join(srcdir, "src", "unix-manager.c")
if not os.path.isfile(path):
print("missing %s (need Suricata source tree as SRCDIR)" % path, file=sys.stderr)
sys.exit(1)
with open(path, encoding="utf-8") as f:
text = f.read()
try:
start = text.index("if (check_dir)")
end = text.index("/* Remove socket file */")
except ValueError:
print("unix-manager.c layout changed; update TOCTOU regression script", file=sys.stderr)
sys.exit(1)
chunk = text[start:end]
stat_tok = "stat(SOCKET_PATH"
mkdir_tok = "SCMkDir(SOCKET_PATH"
if stat_tok not in chunk or mkdir_tok not in chunk:
print("expected %s and %s inside UnixNew check_dir block" % (stat_tok, mkdir_tok), file=sys.stderr)
sys.exit(1)
if chunk.index(stat_tok) >= chunk.index(mkdir_tok):
print("expected stat(SOCKET_PATH before SCMkDir(SOCKET_PATH", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()
tests/unix-socket/unix-socket-tenant-path-01/README.md
# unix-socket-tenant-path-01
Multi-detect must be initialized (`PostConfLoadedDetectSetup`), so this test ships a
minimal `test.rules` file (Suricata otherwise passes `--disable-detection` and skips
multi-tenant setup).
`register-tenant` loads `../tenant-outside.yaml` relative to the test directory i.e.
YAML stored **outside** the per-test output directory, documenting arbitrary path
acceptance for tenant configuration files via the Unix socket.
Related: Unix socket trust-boundary report (tenant YAML path).
Requires: `multi-detect.enabled`, `selector: direct`, loaders; `default-rule-path`
points at `tests/unix-socket/` so `tenant-outside.yaml` can resolve `marker.rule`.
## PCAP
None (`pcap: false`).
tests/unix-socket/unix-socket-tenant-path-01/test.rules
# Minimal rules so Suricata does not use --disable-detection (required for
# multi-detect / PostConfLoadedDetectSetup).
alert tcp any any -> any any (msg:"unix-socket-tenant-path stub"; sid:90004202; rev:1;)
tests/unix-socket/unix-socket-tenant-path-01/test.yaml
# Unix socket register-tenant accepts a filesystem path with ".." relative to the
# process working directory (test dir). Loads YAML outside the test directory,
# documenting missing path confinement for tenant YAML (HIGH-6). Requires multi-detect.
requires:
min-version: 7
pcap: false
startup-timeout: 15
args:
- --set multi-detect.enabled=yes
- --set multi-detect.selector=direct
- --set multi-detect.loaders=2
# Tenant YAML references marker.rule in tests/unix-socket/; rule path resolves from cwd unless overridden.
- --set default-rule-path=${TEST_DIR}/..
unix-commands:
# Tenant YAML lives one directory up (../tenant-outside.yaml); rules alongside it.
- "register-tenant 42 ../tenant-outside.yaml"
checks:
- filter:
filename: sc.json
count: 1
match:
return: OK
message: adding tenant succeeded
tests/unix-socket/unix-socket-v2-partial-read-01/README.md
# unix-socket-v2-partial-read-01
Regression coverage for **HIGH-3**: after the V2 handshake, the manager reads a command
line with `recv()`. If the first read returns bytes without a newline, it waits with
`select()`. On POSIX, `nfds` must be `max(fd)+1`; passing only `fd` never marks the
descriptor ready, so the rest of the command is never read and the peer sees a
closed connection.
`suricatasc` sends whole JSON lines in one write, so this path is not exercised by
normal CLI use. The bundled script sends `{"command":"version"` then completes the
JSON on a second write after a short delay.
Until the engine `select()` argument is corrected, this check fails (the peer sees a
reset connection or never gets a complete reply). Land together with the Suricata fix.
tests/unix-socket/unix-socket-v2-partial-read-01/fragment_v2_command.py
#!/usr/bin/env python3
# Exercise Unix command socket V2 path where the first recv() does not include the
# terminating newline (HIGH-3 / select nfds off-by-one).
import json
import os
import socket
import time
SOCKET_PATH = os.environ["SOCKET_PATH"]
OUTPUT_DIR = os.environ["OUTPUT_DIR"]
OUT = os.path.join(OUTPUT_DIR, "unix-fragment-result.json")
def read_line(sock):
buf = b""
while True:
chunk = sock.recv(1)
if not chunk:
return None
buf += chunk
if chunk == b"\n":
return buf[:-1]
def main():
ok = False
preview = ""
sock = None
try:
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.settimeout(15)
sock.connect(SOCKET_PATH)
sock.sendall(b'{"version":"0.2"}\n')
handshake = read_line(sock)
if handshake is None:
preview = "eof_handshake"
else:
hj = json.loads(handshake.decode("utf-8"))
if hj.get("return") != "OK":
preview = handshake.decode("utf-8", errors="replace")[:500]
else:
sock.sendall(b'{"command":"version"')
time.sleep(0.05)
sock.sendall(b'}\n')
resp = read_line(sock)
if resp:
rj = json.loads(resp.decode("utf-8"))
ok = rj.get("return") == "OK"
preview = resp.decode("utf-8", errors="replace")[:500]
else:
preview = "eof_after_fragmented_command"
except Exception as e:
preview = "%s:%s" % (type(e).__name__, e)
finally:
if sock is not None:
try:
sock.close()
except OSError:
pass
with open(OUT, "w") as f:
json.dump(
{"fragmented_version_command_ok": ok, "response_preview": preview},
f,
)
if __name__ == "__main__":
main()
tests/unix-socket/unix-socket-v2-partial-read-01/test.yaml
# HIGH-3: V2 command handling uses select(client_fd, ...) instead of fd+1; a command
# split across reads never wakes select and the connection is closed as incomplete.
requires:
min-version: 7
pcap: false
startup-timeout: 5
unix-socket-scripts:
- fragment_v2_command.py
unix-commands:
- version
checks:
- filter:
filename: unix-fragment-result.json
count: 1
match:
fragmented_version_command_ok: true
    (1-1/1)