josh/electric-cherry
1
0
Fork 0
Code Issues 18 Pull Requests Actions Packages Projects Releases Wiki Activity

Batch remote drift checks into agent file manifests

Browse Source
This commit is contained in:
josh
2026-10-05 12:52:33 -04:00
parent 10ab071bce
commit f16879036e
11 changed files with 779 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
static/agent/ec-agent.py
+155 -4
View File
@@ -32,9 +32,13 @@ import urllib.parse
import urllib.request
import uuid
VERSION = "1.4.7"
VERSION = "1.4.8"
LABEL = "io.electric-cherry"
MAX_FILE = 32 * 1024 * 1024
MANIFEST_MAX_ENTRIES = 10000
MANIFEST_MAX_BYTES = 16 * 1024 * 1024
MANIFEST_MAX_SECONDS = 15
MANIFEST_MAX_DEPTH = 64
LIBRARY_DOWNLOAD_WORKERS = 4
ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$")
AGENT_CONFIG = Path("/etc/electric-cherry/agent.json")
@@ -296,6 +300,151 @@ def managed_read(root, value):
return None
def managed_manifest(root, payload, stopping):
"""Hash a bounded managed tree without following runtime-controlled symlinks."""
deadline = time.monotonic() + MANIFEST_MAX_SECONDS
def check_budget():
if stopping.is_set() or time.monotonic() >= deadline:
raise AgentError("Manifest scan exceeded its time limit or was interrupted")
roots = payload.get("roots")
ignored = payload.get("ignoreNames")
paths = payload.get("paths", [])
if not isinstance(roots, list) or not roots or len(roots) > 64:
raise AgentError("Manifest roots must contain between 1 and 64 managed directories")
if not isinstance(ignored, list) or len(ignored) > 256 or any(
not isinstance(name, str) or not name or name in (".", "..") or any(char in name for char in ("/", "\\", "\x00"))
for name in ignored
):
raise AgentError("Manifest ignored names must be plain file or directory names")
if not isinstance(paths, list) or len(paths) > 10000:
raise AgentError("Manifest managed paths must contain at most 10000 files")
for value in roots:
check_budget()
scoped(root, value)
roots = sorted(set(roots))
roots = [value for index, value in enumerate(roots) if not any(value.startswith(parent + "/") for parent in roots[:index])]
ignored = set(ignored)
flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW
entries = 0
result_bytes = len(json.dumps({"files": []}).encode())
files = []
def validate_paths():
for value in paths:
check_budget()
try:
with managed_open(root, value):
pass
except FileNotFoundError:
pass
def identity(info):
return info.st_dev, info.st_ino, info.st_mode, info.st_size, info.st_mtime_ns, info.st_ctime_ns
def open_directory(app, relative):
directory = os.dup(app)
try:
for part in PurePosixPath(relative).parts:
nested = os.open(part, flags, dir_fd=directory)
os.close(directory)
directory = nested
return directory
except BaseException:
os.close(directory)
raise
def scan(directory, relative, depth):
nonlocal entries, result_bytes
check_budget()
if depth > MANIFEST_MAX_DEPTH:
raise AgentError("Manifest scan exceeded its directory depth limit")
before = os.fstat(directory)
with os.scandir(directory) as children:
for child in children:
check_budget()
entries += 1
if entries > MANIFEST_MAX_ENTRIES:
raise AgentError("Manifest scan exceeded its entry limit")
if child.name in ignored:
continue
info = os.stat(child.name, dir_fd=directory, follow_symlinks=False)
if stat_module.S_ISLNK(info.st_mode):
continue
path = relative + "/" + child.name
if stat_module.S_ISDIR(info.st_mode):
nested = os.open(child.name, flags, dir_fd=directory)
try:
if identity(info) != identity(os.fstat(nested)):
raise AgentError("Manifest directory changed during the scan")
scan(nested, path, depth + 1)
if identity(os.fstat(nested)) != identity(os.stat(child.name, dir_fd=directory, follow_symlinks=False)):
raise AgentError("Manifest directory changed during the scan")
finally:
os.close(nested)
elif stat_module.S_ISREG(info.st_mode):
descriptor = os.open(child.name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory)
try:
if identity(info) != identity(os.fstat(descriptor)):
raise AgentError("Manifest file changed during the scan")
sha = hashlib.sha256()
size = 0
with os.fdopen(descriptor, "rb", closefd=False) as stream:
while True:
check_budget()
block = stream.read(1024 * 1024)
if not block:
break
sha.update(block)
size += len(block)
if size != info.st_size or identity(info) != identity(os.fstat(descriptor)) or identity(info) != identity(os.stat(child.name, dir_fd=directory, follow_symlinks=False)):
raise AgentError("Manifest file changed during the scan")
finally:
os.close(descriptor)
item = {"path": path, "size": size, "sha256": sha.hexdigest()}
result_bytes += len(json.dumps(item).encode()) + (2 if files else 0)
if result_bytes > MANIFEST_MAX_BYTES:
raise AgentError("Manifest scan exceeded its result size limit")
files.append(item)
else:
raise AgentError("Manifest scan encountered a non-regular managed file")
if identity(before) != identity(os.fstat(directory)):
raise AgentError("Manifest directory changed during the scan")
try:
app = os.open(root, flags)
try:
app_identity = os.fstat(app)
validate_paths()
for relative in roots:
check_budget()
try:
directory = open_directory(app, relative)
except FileNotFoundError:
continue # A destination directory need not exist yet.
try:
scan(directory, relative, len(PurePosixPath(relative).parts))
current_directory = open_directory(app, relative)
try:
if identity(os.fstat(directory)) != identity(os.fstat(current_directory)):
raise AgentError("Manifest directory changed during the scan")
finally:
os.close(current_directory)
finally:
os.close(directory)
validate_paths()
current = os.stat(root, follow_symlinks=False)
if (current.st_dev, current.st_ino, current.st_mode) != (app_identity.st_dev, app_identity.st_ino, app_identity.st_mode):
raise AgentError("Manifest managed root changed during the scan")
check_budget()
finally:
os.close(app)
except OSError:
raise AgentError("Manifest scan could not read the complete managed tree") from None
return {"files": sorted(files, key=lambda item: item["path"])}
def managed_write(root, value, content, uid=1000, gid=1000):
if content is None:
try:
@@ -1094,7 +1243,7 @@ class Agent:
if line.startswith("PRETTY_NAME="):
os_name = line.split("=", 1)[1].strip('"')
update = self.update_inventory()
capabilities = ["library.remove", "instance.cancel-create", "operations.read-lane", "operations.health-lane"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else [])
capabilities = ["library.remove", "instance.cancel-create", "operations.read-lane", "operations.health-lane", "files.manifest"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else [])
return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, **resources, **({"cpuUsagePercent": cpu_usage} if cpu_usage is not None else {}), "memoryTotal": memory.get("MemTotal", 0), "memoryAvailable": memory.get("MemAvailable", 0), "storageTotal": usage.total, "storageAvailable": usage.free, "dockerVersion": docker_version, "dockerRunning": docker_running, "directoriesWritable": os.access(self.root, os.W_OK), "runtimeImages": sorted(set(images)), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})}
def safe_to_disrupt(self, instance_id, generation):
@@ -1446,6 +1595,8 @@ class Agent:
return result
if kind == "logs.read":
return self.read_log(instance_id, generation, payload)
if kind == "files.manifest":
return managed_manifest(app, payload, self.stopping)
path = scoped(app, payload.get("path", ""), allow_root=kind in ("files.list", "files.stat"))
if kind == "files.read":
content = managed_read(app, payload["path"])
@@ -1967,7 +2118,7 @@ class Agent:
return self.journal.update(operation["id"], "failed", error="Interrupted library download could not be resumed")
if kind == "files.stage":
return self.journal.update(operation["id"], "completed", self.stage(instance_id, generation, payload, operation["id"]))
if kind in ("host.inspect", "instance.inspect", "files.read", "files.list", "files.stat", "logs.read") or kind == "rcon.execute" and payload.get("command") in ("status", "getstatus", "version"):
if kind in ("host.inspect", "instance.inspect", "files.read", "files.list", "files.stat", "files.manifest", "logs.read") or kind == "rcon.execute" and payload.get("command") in ("status", "getstatus", "version"):
return self.journal.update(operation["id"], "completed", self.dispatch(operation))
if kind in ("files.activate", "files.rollback") and checkpoint:
stage = self.instance_dir(instance_id, generation) / "stages" / identifier(payload["stageId"]) / "manifest.json"
@@ -2054,7 +2205,7 @@ class Agent:
@staticmethod
def read_operation(operation):
return operation.get("kind") in ("host.inspect", "instance.inspect", "files.read", "files.list", "files.stat", "logs.read") or (operation.get("kind") == "rcon.execute" and operation.get("payload", {}).get("command") in ("status", "getstatus", "version"))
return operation.get("kind") in ("host.inspect", "instance.inspect", "files.read", "files.list", "files.stat", "files.manifest", "logs.read") or (operation.get("kind") == "rcon.execute" and operation.get("payload", {}).get("command") in ("status", "getstatus", "version"))
def instance_lock(self, operation):
key = operation.get("instanceId")
static/agent/manifest.json
+2 -2
View File
@@ -1,8 +1,8 @@
{
"version": "1.4.7",
"version": "1.4.8",
"protocolVersion": 1,
"platform": "debian-amd64",
"sha256": "e9e1b3869d66eca351dadb045d99451b97c91d15c5b7199a3de3241b550de998",
"sha256": "45eb0ef619fb0ceef2ac843657fd7fdaf8cfa610951f0ddfabdbf5942abc6e8c",
"updaterProtocolVersion": 1,
"updaterSha256": "066bb896726600ece9ce72697ce523c9a707cc9a621505931ba0892dec59d3c8"
}