"""Run the real AcTrail fanotify permission-enforcement E2E."""
import argparse
import json
import os
import select
import signal
import subprocess
import sys
import tempfile
import time
from pathlib import Path
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser()
parser.add_argument("--bin-dir", default=os.environ.get("ACTRAIL_BIN_DIR", "target/release"))
parser.add_argument("--daemon-ready-timeout-sec", type=float, default=10.0)
parser.add_argument("--agent-timeout-sec", type=float, default=10.0)
parser.add_argument("--drain-attempts", type=int, default=20)
parser.add_argument("--drain-sleep-sec", type=float, default=0.2)
parser.add_argument("--otel-output", help="export and validate OTEL JSON before cleanup")
return parser.parse_args()
def require_root() -> None:
if os.geteuid() != 0:
raise RuntimeError("fanotify permission events require root/CAP_SYS_ADMIN")
def require_binary(bin_dir: Path, name: str) -> Path:
path = bin_dir / name
if not path.exists():
raise RuntimeError(f"missing binary {path}; build with cargo build --release")
return path
def write_text(path: Path, text: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(text, encoding="utf-8")
def operator_config(tmp: Path, rules_path: Path) -> str:
return f"""# Generated by docs/examples/04.fanotify-enforcement-e2e/run_e2e.py
[control]
socket_path = "{tmp / 'actrail.sock'}"
socket_mode_octal = "660"
pending_connection_max = 256
pid_file = "{tmp / 'actraild.pid'}"
log_path = "{tmp / 'actraild.log'}"
diagnostic_log_level = "info"
[storage]
backend = "sqlite"
[storage.sqlite]
path = "{tmp / 'actrail.sqlite'}"
busy_timeout_ms = 5000
[web]
listen_addr = "127.0.0.1:18080"
request_read_timeout_ms = "1000"
[export.snapshot]
graph_schema_version = "manual-v1"
allow_active_trace_snapshot = true
directory = "{tmp / 'export'}"
payload_bytes_enabled = false
payload_text_enabled = false
[capture]
profile_name = "fanotify-enforcement-e2e"
capabilities = ["proc-lifecycle", "enforcement-file-permission-fanotify"]
[ebpf]
enabled = true
memlock_rlimit = "inherit"
tracked_process_max_entries = 4096
pending_operation_max_entries = 4096
suppressed_fd_max_entries = 4096
suppressed_fd_index_slots_per_process = 64
event_ring_buffer_max_bytes = 1048576
file_path_capture_enabled = false
file_path_max_bytes = 255
[payload.tls]
enabled = false
capture_backend = "tls-sync"
source = "auto"
resolver = "auto"
library = "auto"
library_path = "auto"
binary_path = "disabled"
pattern_path = "disabled"
max_segment_bytes = 4095
max_operation_bytes = 4194304
ring_buffer_bytes = 1048576
pending_operation_max_entries = 4096
retention_max_bytes_per_trace = 10485760
redaction_policy = "authorization-header"
sync_runtime_library_path = "auto"
sync_event_socket_path = "/tmp/actrail-fanotify-e2e-tls-sync.sock"
sync_socket_mode_octal = "660"
sync_match_limit = 8
seccomp_syscalls = ["write", "writev", "sendto", "sendmsg"]
[payload.stdio]
enabled = false
capture_stdin = false
capture_stdout = true
capture_stderr = true
stdin_storage_mode = "full"
stdout_storage_mode = "drop"
stderr_storage_mode = "metadata-only"
max_segment_bytes = 4095
ring_buffer_bytes = 1048576
pending_operation_max_entries = 4096
stream_state_max_entries = 4096
retention_max_bytes_per_trace = 10485760
redaction_policy = "authorization-header"
[payload.socket]
enabled = false
capture_backend = "bpf-copy-seccomp-fallback"
max_segment_bytes = 4095
max_operation_bytes = 4194304
ring_buffer_bytes = 2097152
pending_operation_max_entries = 4096
stream_state_max_entries = 4096
retention_max_bytes_per_trace = 10485760
redaction_policy = "authorization-header"
http_sniff_max_bytes = 8192
seccomp_syscalls = ["write", "sendto"]
[seccomp_notify]
enabled = false
reserved_listener_fd = 253
[process_seccomp]
enabled = false
syscalls = ["execve", "execveat", "fork", "vfork", "clone", "clone3"]
max_args = 64
max_arg_bytes = 4096
pending_max_entries = 4096
[agent_invocation]
enabled = false
commands = ["opencode", ".opencode", "claude"]
[application]
enabled = false
http1_enabled = false
http2_enabled = false
[application.http]
capture_host = false
sse_enabled = false
sse_data_policy = "disabled"
sse_max_buffer_bytes = 4194304
sse_max_data_bytes = 4096
[application.http2]
max_frame_bytes = 16384
max_connection_buffer_bytes = 1048576
emit_data_preview = false
max_data_preview_bytes = 4096
[resource_metrics]
enabled = false
interval_ms = 1000
include_children = true
include_system = true
cpu_alert_percent_millis = "disabled"
memory_alert_rss_kb = "disabled"
[provider]
rules_enabled = false
rules_path = "/etc/actrail/provider-rules.conf"
unknown_provider_label = "unknown"
[enforcement]
enabled = true
backend = "fanotify"
scope = "trace"
rules_path = "{rules_path}"
default_decision = "allow"
mark_strategy = "parent-directories"
audit_enabled = true
event_buffer_bytes = 65536
[supervision]
startup_wait_ms = 5000
shutdown_wait_ms = 5000
poll_interval_ms = 100
"""
def wait_for_daemon(process: subprocess.Popen[str], timeout_sec: float) -> None:
deadline = time.monotonic() + timeout_sec
while time.monotonic() < deadline:
line = read_line_until(process, process.stdout, deadline)
if line:
print(line, end="")
if "daemon listening" in line:
return
if process.poll() is not None:
stderr = process.stderr.read()
raise RuntimeError(f"actraild exited early: {stderr}")
raise RuntimeError("actraild did not report readiness")
def read_agent_pid(process: subprocess.Popen[str], timeout_sec: float) -> int:
deadline = time.monotonic() + timeout_sec
while time.monotonic() < deadline:
line = read_line_until(process, process.stdout, deadline)
if line.startswith("agent_pid="):
print(line, end="")
return int(line.split("=", 1)[1])
if line:
print(line, end="")
if process.poll() is not None:
raise RuntimeError("agent exited before reporting pid")
raise RuntimeError("agent did not report pid")
def run_checked(command: list[str]) -> str:
result = subprocess.run(command, text=True, capture_output=True, check=False)
if result.returncode != 0:
raise RuntimeError(
f"command failed: {' '.join(command)}\nstdout={result.stdout}\nstderr={result.stderr}"
)
if result.stdout:
print(result.stdout, end="")
if result.stderr:
print(result.stderr, end="", file=sys.stderr)
return result.stdout
def wait_for_agent_output(process: subprocess.Popen[str], timeout_sec: float) -> str:
deadline = time.monotonic() + timeout_sec
lines: list[str] = []
while time.monotonic() < deadline:
line = read_line_until(process, process.stdout, deadline)
if line:
print(line, end="")
lines.append(line)
if "pre_exec_redirection=" in line:
break
if process.poll() is not None:
break
exit_code = process.wait(timeout=max(deadline - time.monotonic(), 0.1))
output = "".join(lines)
if exit_code != 0:
raise RuntimeError(f"agent failed with exit={exit_code}: {output}")
return output
def read_line_until(
process: subprocess.Popen[str],
stream,
deadline: float,
) -> str:
if stream is None:
raise RuntimeError("process stream is not captured")
remaining = deadline - time.monotonic()
if remaining <= 0:
return ""
readable, _, _ = select.select([stream], [], [], remaining)
if readable:
return stream.readline()
if process.poll() is not None:
return stream.readline()
return ""
def wait_for_enforcement_events(
actrailctl: Path,
actrailviewer: Path,
config_path: Path,
attempts: int,
sleep_sec: float,
) -> str:
for _ in range(attempts):
run_checked([str(actrailctl), "--config", str(config_path), "list-traces"])
output = run_checked(
[str(actrailviewer), "events", "--config", str(config_path), "--trace-id", "1"]
)
if all(
value in output
for value in [
"Enforcement",
"allow-file",
"deny-file",
"deny-pre-exec-redirection",
]
):
return output
time.sleep(sleep_sec)
raise RuntimeError("actrailviewer did not show expected enforcement events")
def export_and_validate_otel(actrailviewer: Path, config_path: Path, output_path: Path) -> None:
output_path.parent.mkdir(parents=True, exist_ok=True)
if output_path.exists():
output_path.unlink()
run_checked(
[
str(actrailviewer),
"export-otel",
"--config",
str(config_path),
"--trace-id",
"1",
"--output",
str(output_path),
]
)
document = json.loads(output_path.read_text(encoding="utf-8"))
found = {"allow": False, "deny": False, "pre_exec_deny": False}
for span in otel_spans(document):
attributes = otel_attributes(span)
if attributes.get("actrail.action.kind") != "enforcement.decision":
continue
decision = attributes.get("enforcement.decision")
status = attributes.get("actrail.action.status")
result = attributes.get("enforcement.result")
rule_id = attributes.get("enforcement.rule_id")
code = span.get("status", {}).get("code")
if decision == "allow" and status == "success" and result == "allowed":
found["allow"] = rule_id == "allow-file" and code == "STATUS_CODE_OK"
if decision == "deny" and status == "error" and result == "denied":
if rule_id == "deny-file":
found["deny"] = code == "STATUS_CODE_ERROR"
if rule_id == "deny-pre-exec-redirection":
found["pre_exec_deny"] = code == "STATUS_CODE_ERROR"
if not all(found.values()):
raise RuntimeError(f"OTEL export missed enforcement decision spans: {found}")
print(f"otel_output={output_path}")
print("otel_enforcement_spans=allow,deny")
print("otel_pre_exec_redirection_span=deny")
def otel_spans(document: dict) -> list[dict]:
spans: list[dict] = []
for resource_span in document.get("resourceSpans", []):
for scope_span in resource_span.get("scopeSpans", []):
spans.extend(scope_span.get("spans", []))
return spans
def otel_attributes(span: dict) -> dict[str, str]:
output: dict[str, str] = {}
for attribute in span.get("attributes", []):
value = attribute.get("value", {})
if "stringValue" in value:
output[attribute.get("key", "")] = value["stringValue"]
elif "intValue" in value:
output[attribute.get("key", "")] = str(value["intValue"])
return output
def stop_process(process: subprocess.Popen[str]) -> None:
if process.poll() is not None:
return
process.send_signal(signal.SIGTERM)
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
process.kill()
process.wait()
def main() -> int:
args = parse_args()
require_root()
repo = Path.cwd()
bin_dir = repo / args.bin_dir
actraild = require_binary(bin_dir, "actraild")
actrailctl = require_binary(bin_dir, "actrailctl")
actrailviewer = require_binary(bin_dir, "actrailviewer")
agent_script = Path(__file__).resolve().parent / "agent.py"
with tempfile.TemporaryDirectory(prefix="actrail-fanotify-e2e-") as raw_tmp:
tmp = Path(raw_tmp)
allowed = tmp / "targets" / "allowed.txt"
denied = tmp / "targets" / "denied.txt"
pre_exec_denied = tmp / "targets" / "pre-exec-denied.txt"
rules = tmp / "rules.conf"
config = tmp / "operator.conf"
write_text(allowed, "allowed\n")
write_text(denied, "denied\n")
write_text(pre_exec_denied, "pre-exec denied\n")
write_text(
rules,
f"allow-file allow open {allowed}\n"
f"deny-file deny open {denied}\n"
f"deny-pre-exec-redirection deny open {pre_exec_denied}\n",
)
write_text(config, operator_config(tmp, rules))
daemon = subprocess.Popen(
[str(actraild), "--config", str(config), "run"],
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
try:
wait_for_daemon(daemon, args.daemon_ready_timeout_sec)
agent = subprocess.Popen(
[
sys.executable,
str(agent_script),
"--allowed-path",
str(allowed),
"--denied-path",
str(denied),
"--pre-exec-denied-path",
str(pre_exec_denied),
],
text=True,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
try:
pid = read_agent_pid(agent, args.agent_timeout_sec)
run_checked(
[
str(actrailctl),
"--config",
str(config),
"track-add",
"--pid",
str(pid),
"--name",
"fanotify-enforcement-e2e",
]
)
agent.stdin.write("go\n")
agent.stdin.flush()
agent_output = wait_for_agent_output(agent, args.agent_timeout_sec)
if "allowed=ok" not in agent_output:
raise RuntimeError("agent did not confirm allowed file access")
if "denied=permission_denied" not in agent_output:
raise RuntimeError("agent did not confirm denied file access")
if "pre_exec_redirection=permission_denied" not in agent_output:
raise RuntimeError(
"agent did not confirm fork-before-exec redirection was denied"
)
viewer_output = wait_for_enforcement_events(
actrailctl,
actrailviewer,
config,
args.drain_attempts,
args.drain_sleep_sec,
)
if "decision=allow" not in viewer_output or "decision=deny" not in viewer_output:
raise RuntimeError("viewer output missed allow/deny decisions")
if args.otel_output:
export_and_validate_otel(actrailviewer, config, Path(args.otel_output))
finally:
stop_process(agent)
finally:
stop_process(daemon)
return 0
if __name__ == "__main__":
try:
raise SystemExit(main())
except Exception as error:
print(f"fanotify e2e failed: {error}", file=sys.stderr)
raise SystemExit(1)
